feat(deliverymq): bound retry task receives with backoff and DLQ

internal/deliverymq/retry.go

@@ -21,6 +21,10 @@ type RetryEventGetter interface {
 	ListAttempt(ctx context.Context, request logstore.ListAttemptRequest) (logstore.ListAttemptResponse, error)
 }
 
+// retryMaxReceiveCount bounds receives of a retry task before it is moved to
+// the "deliverymq-retry-dlq" queue; separate from the delivery retry max limit.
+const retryMaxReceiveCount = 5
+
 // RetrySchedulerOption is a functional option for configuring the retry scheduler.
 type RetrySchedulerOption func(*retrySchedulerConfig)
 
@@ -112,13 +116,15 @@ func NewRetryScheduler(deliverymq *DeliveryMQ, redisConfig *redis.RedisConfig, d
 		return nil
 	}
 
+	schedulerOpts := []scheduler.Option{
+		scheduler.WithPollBackoff(pollBackoff),
+		scheduler.WithLogger(logger),
+		scheduler.WithMaxReceiveCount(retryMaxReceiveCount),
+	}
 	if cfg.visibilityTimeout > 0 {
-		return scheduler.New("deliverymq-retry", rsmqClient, exec,
-			scheduler.WithPollBackoff(pollBackoff),
-			scheduler.WithVisibilityTimeout(cfg.visibilityTimeout),
-			scheduler.WithLogger(logger)), nil
+		schedulerOpts = append(schedulerOpts, scheduler.WithVisibilityTimeout(cfg.visibilityTimeout))
 	}
-	return scheduler.New("deliverymq-retry", rsmqClient, exec, scheduler.WithPollBackoff(pollBackoff), scheduler.WithLogger(logger)), nil
+	return scheduler.New("deliverymq-retry", rsmqClient, exec, schedulerOpts...), nil
 }
 
 // RetryTask contains the minimal info needed to retry a delivery.

internal/rsmq/rsmq.go

@@ -40,11 +40,12 @@ const (
 
 const (
 	defaultNs      = "rsmq"
-	defaultVt      = 30
 	defaultDelay   = 0
 	defaultMaxsize = 65536
 )
 
+const DefaultVt uint = 30
+
 // Errors returned on rsmq operation
 var (
 	ErrQueueNotFound   = errors.New("queue not found")
@@ -146,6 +147,7 @@ type Client interface {
 	CreateQueue(qname string, vt uint, delay uint, maxsize int) error
 	ReceiveMessage(qname string, vt uint) (*QueueMessage, error)
 	SendMessage(qname string, message string, delay uint, opts ...SendMessageOption) (string, error)
+	ChangeMessageVisibility(qname string, id string, vt uint) error
 	DeleteMessage(qname string, id string) error
 	Quit() error
 }
@@ -189,7 +191,7 @@ func NewRedisSMQ(client RedisClient, ns string, logger ...*logging.Logger) *Redi
 //	err:=redisRsmq.CreateQueue(qname,rsmq.UnsetVt,rsmq.UnsetDelay,rsmq.UnsetMaxsize)
 func (rsmq *RedisSMQ) CreateQueue(qname string, vt uint, delay uint, maxsize int) error {
 	if vt == UnsetVt {
-		vt = defaultVt
+		vt = DefaultVt
 	}
 	if delay == UnsetDelay {
 		delay = defaultDelay
@@ -510,6 +512,9 @@ func (rsmq *RedisSMQ) SendMessage(qname string, message string, delay uint, opts
 	})
 	tx.HSet(key+q, messageID, message)
 	tx.HIncrBy(key+q, "totalsent", 1)
+	if options.id != "" {
+		tx.HDel(key+q, messageID+":rc", messageID+":fr")
+	}
 	if _, err := tx.Exec(); err != nil {
 		return "", err
 	}

internal/rsmq/rsmq_test.go

@@ -173,7 +173,7 @@ func (s *RSMQSuite) TestGetQueueAttributes() {
 	queAttrib, err := s.rsmq.GetQueueAttributes(qname)
 	assert.Nil(t, err, "error is not nil on getting queue attributes")
 	assert.NotNil(t, queAttrib, "queueAttributes is nil")
-	assert.EqualValues(t, defaultVt, queAttrib.Vt, "queueAttributes vt is not as expected")
+	assert.EqualValues(t, DefaultVt, queAttrib.Vt, "queueAttributes vt is not as expected")
 	assert.EqualValues(t, defaultDelay, queAttrib.Delay, "queueAttributes delay is not as expected")
 	assert.Equal(t, defaultMaxsize, queAttrib.Maxsize, "queueAttributes maxsize is not as expected")
 	assert.Zero(t, queAttrib.TotalRecv, "queueAttributes totalRecv is not zero")
@@ -304,7 +304,7 @@ func (s *RSMQSuite) TestSetQueueAttributes() {
 		queAttrib, err = s.rsmq.GetQueueAttributes(qname)
 		assert.Nil(t, err, "error is not nil on getting queue attributes")
 		assert.NotNil(t, queAttrib, "queueAttributes is nil")
-		assert.EqualValues(t, defaultVt, queAttrib.Vt, "queueAttributes vt is not as expected")
+		assert.EqualValues(t, DefaultVt, queAttrib.Vt, "queueAttributes vt is not as expected")
 	})
 
 	t.Run("error when the queue attribute delay is not valid", func(t *testing.T) {
@@ -734,6 +734,48 @@ func (s *RSMQSuite) TestSendMessageWithCustomID() {
 		}
 	})
 
+	t.Run("override resets receive count", func(t *testing.T) {
+		err := s.rsmq.CreateQueue(qname, UnsetVt, UnsetDelay, UnsetMaxsize)
+		if err != nil {
+			t.Fatal(err)
+		}
+		defer s.rsmq.DeleteQueue(qname)
+
+		_, err = s.rsmq.SendMessage(qname, "first message", UnsetDelay, WithMessageID(customID))
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		for i := uint64(1); i <= 2; i++ {
+			msg, err := s.rsmq.ReceiveMessage(qname, 0)
+			if err != nil {
+				t.Fatal(err)
+			}
+			if msg == nil {
+				t.Fatal("expected to receive a message")
+			}
+			if msg.Rc != i {
+				t.Errorf("expected receive count %d, got %d", i, msg.Rc)
+			}
+		}
+
+		_, err = s.rsmq.SendMessage(qname, "second message", UnsetDelay, WithMessageID(customID))
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		msg, err := s.rsmq.ReceiveMessage(qname, 0)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if msg == nil {
+			t.Fatal("expected to receive the overwritten message")
+		}
+		if msg.Rc != 1 {
+			t.Errorf("expected receive count 1 after override, got %d", msg.Rc)
+		}
+	})
+
 	t.Run("override changes delay timing", func(t *testing.T) {
 		err := s.rsmq.CreateQueue(qname, UnsetVt, UnsetDelay, UnsetMaxsize)
 		if err != nil {

internal/scheduler/backoff_test.go

@@ -0,0 +1,60 @@
+package scheduler
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestCalculateExecBackoff(t *testing.T) {
+	tests := []struct {
+		name         string
+		base         time.Duration
+		receiveCount uint64
+		maxBackoff   time.Duration
+		want         time.Duration
+	}{
+		{
+			name:         "first receive uses base",
+			base:         30 * time.Second,
+			receiveCount: 1,
+			maxBackoff:   15 * time.Minute,
+			want:         30 * time.Second,
+		},
+		{
+			name:         "backoff doubles per receive",
+			base:         30 * time.Second,
+			receiveCount: 5,
+			maxBackoff:   15 * time.Minute,
+			want:         8 * time.Minute,
+		},
+		{
+			name:         "backoff is capped",
+			base:         30 * time.Second,
+			receiveCount: 6,
+			maxBackoff:   15 * time.Minute,
+			want:         15 * time.Minute,
+		},
+		{
+			name:         "base above cap is capped on first receive",
+			base:         30 * time.Minute,
+			receiveCount: 1,
+			maxBackoff:   15 * time.Minute,
+			want:         15 * time.Minute,
+		},
+		{
+			name:         "large receive count does not overflow",
+			base:         30 * time.Second,
+			receiveCount: ^uint64(0),
+			maxBackoff:   15 * time.Minute,
+			want:         15 * time.Minute,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			require.Equal(t, tt.want, calculateExecBackoff(tt.base, tt.receiveCount, tt.maxBackoff))
+		})
+	}
+}

internal/scheduler/scheduler.go

@@ -42,34 +42,50 @@ type config struct {
 	pollBackoff          time.Duration
 	maxConsecutiveErrors int
 	maxErrorBackoff      time.Duration
+	maxReceiveCount      uint64
+	maxExecBackoff       time.Duration
 	logger               *logging.Logger
 }
 
-func WithVisibilityTimeout(vt uint) func(*config) {
+type Option func(*config)
+
+func WithVisibilityTimeout(vt uint) Option {
 	return func(c *config) {
 		c.visibilityTimeout = vt
 	}
 }
 
-func WithPollBackoff(backoff time.Duration) func(*config) {
+func WithPollBackoff(backoff time.Duration) Option {
 	return func(c *config) {
 		c.pollBackoff = backoff
 	}
 }
 
-func WithMaxConsecutiveErrors(n int) func(*config) {
+func WithMaxConsecutiveErrors(n int) Option {
 	return func(c *config) {
 		c.maxConsecutiveErrors = n
 	}
 }
 
-func WithLogger(logger *logging.Logger) func(*config) {
+func WithMaxReceiveCount(n uint64) Option {
+	return func(c *config) {
+		c.maxReceiveCount = n
+	}
+}
+
+func WithMaxExecBackoff(d time.Duration) Option {
+	return func(c *config) {
+		c.maxExecBackoff = d
+	}
+}
+
+func WithLogger(logger *logging.Logger) Option {
 	return func(c *config) {
 		c.logger = logger
 	}
 }
 
-func New(name string, rsmqClient rsmq.Client, exec func(context.Context, string) error, opts ...func(*config)) Scheduler {
+func New(name string, rsmqClient rsmq.Client, exec func(context.Context, string) error, opts ...Option) Scheduler {
 	// Error retry schedule (with 100ms pollBackoff used by retrymq):
 	//
 	//   Error  Backoff    Cumulative
@@ -95,6 +111,7 @@ func New(name string, rsmqClient rsmq.Client, exec func(context.Context, string)
 		pollBackoff:          200 * time.Millisecond,
 		maxConsecutiveErrors: 10,
 		maxErrorBackoff:      15 * time.Second,
+		maxExecBackoff:       15 * time.Minute,
 	}
 	for _, opt := range opts {
 		opt(config)
@@ -112,9 +129,18 @@ func (s *schedulerImpl) Init(ctx context.Context) error {
 	if err := s.rsmqClient.CreateQueue(s.name, s.config.visibilityTimeout, rsmq.UnsetDelay, rsmq.UnsetMaxsize); err != nil && err != rsmq.ErrQueueExists {
 		return err
 	}
+	if s.config.maxReceiveCount > 0 {
+		if err := s.rsmqClient.CreateQueue(s.dlqName(), rsmq.UnsetVt, rsmq.UnsetDelay, rsmq.UnsetMaxsize); err != nil && err != rsmq.ErrQueueExists {
+			return err
+		}
+	}
 	return nil
 }
 
+func (s *schedulerImpl) dlqName() string {
+	return s.name + "-dlq"
+}
+
 func (s *schedulerImpl) Schedule(ctx context.Context, task string, delay time.Duration, opts ...ScheduleOption) error {
 	// Parse options
 	options := &ScheduleOptions{}
@@ -167,9 +193,16 @@ func (s *schedulerImpl) Monitor(ctx context.Context) error {
 				time.Sleep(s.config.pollBackoff)
 				continue
 			}
+			if s.config.maxReceiveCount > 0 && msg.Rc > s.config.maxReceiveCount {
+				s.moveToDLQ(ctx, msg)
+				continue
+			}
 			// TODO: consider using a worker pool to limit the number of concurrent executions
 			go func() {
 				if err := s.exec(ctx, msg.Message); err != nil {
+					if s.config.maxReceiveCount > 0 {
+						s.backoffMessage(ctx, msg)
+					}
 					return
 				}
 				if err := s.rsmqClient.DeleteMessage(s.name, msg.ID); err != nil {
@@ -180,6 +213,66 @@ func (s *schedulerImpl) Monitor(ctx context.Context) error {
 	}
 }
 
+// moveToDLQ dead-letters a message that exceeded the max receive count. The
+// DLQ send happens before the delete so a failure between the two steps can
+// only duplicate the message, not lose it.
+func (s *schedulerImpl) moveToDLQ(ctx context.Context, msg *rsmq.QueueMessage) {
+	logger := s.config.logger.Ctx(ctx)
+	if _, err := s.rsmqClient.SendMessage(s.dlqName(), msg.Message, 0); err != nil {
+		logger.Error("failed to move task to dead-letter queue",
+			zap.Error(err),
+			zap.String("queue", s.name),
+			zap.String("message_id", msg.ID),
+			zap.Uint64("receive_count", msg.Rc))
+		return
+	}
+	if err := s.rsmqClient.DeleteMessage(s.name, msg.ID); err != nil && err != rsmq.ErrMessageNotFound {
+		logger.Error("failed to delete task after moving to dead-letter queue",
+			zap.Error(err),
+			zap.String("queue", s.name),
+			zap.String("message_id", msg.ID))
+		return
+	}
+	logger.Error("task exceeded max receive count, moved to dead-letter queue",
+		zap.String("queue", s.name),
+		zap.String("dlq", s.dlqName()),
+		zap.String("message_id", msg.ID),
+		zap.Uint64("receive_count", msg.Rc),
+		zap.String("task", msg.Message))
+}
+
+// backoffMessage reschedules a failed message with exponential backoff
+// (vt, 2*vt, 4*vt, ... capped at maxExecBackoff).
+func (s *schedulerImpl) backoffMessage(ctx context.Context, msg *rsmq.QueueMessage) {
+	base := s.config.visibilityTimeout
+	if base == rsmq.UnsetVt {
+		base = rsmq.DefaultVt
+	}
+	backoff := calculateExecBackoff(time.Duration(base)*time.Second, msg.Rc, s.config.maxExecBackoff)
+	if err := s.rsmqClient.ChangeMessageVisibility(s.name, msg.ID, uint(backoff/time.Second)); err != nil && err != rsmq.ErrMessageNotFound {
+		s.config.logger.Ctx(ctx).Warn("failed to extend backoff for failed task",
+			zap.Error(err),
+			zap.String("queue", s.name),
+			zap.String("message_id", msg.ID),
+			zap.Uint64("receive_count", msg.Rc))
+	}
+}
+
+func calculateExecBackoff(base time.Duration, receiveCount uint64, maxBackoff time.Duration) time.Duration {
+	if base <= 0 || maxBackoff <= 0 {
+		return 0
+	}
+
+	backoff := min(base, maxBackoff)
+	for i := uint64(1); i < receiveCount && backoff < maxBackoff; i++ {
+		if backoff >= maxBackoff-backoff {
+			return maxBackoff
+		}
+		backoff *= 2
+	}
+	return backoff
+}
+
 func (s *schedulerImpl) Cancel(ctx context.Context, taskID string) error {
 	// Generate the RSMQ ID for this task
 	rsmqID := generateRSMQID(taskID)