diff --git a/internal/deliverymq/retry.go b/internal/deliverymq/retry.go index 8196316e1..e43bd38a8 100644 --- a/internal/deliverymq/retry.go +++ b/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. diff --git a/internal/rsmq/rsmq.go b/internal/rsmq/rsmq.go index 51ef47feb..7eb104984 100644 --- a/internal/rsmq/rsmq.go +++ b/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 } diff --git a/internal/rsmq/rsmq_test.go b/internal/rsmq/rsmq_test.go index db9e76d94..d264a0fc2 100644 --- a/internal/rsmq/rsmq_test.go +++ b/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 { diff --git a/internal/scheduler/backoff_test.go b/internal/scheduler/backoff_test.go new file mode 100644 index 000000000..792966b61 --- /dev/null +++ b/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)) + }) + } +} diff --git a/internal/scheduler/scheduler.go b/internal/scheduler/scheduler.go index 9968f7beb..86f6a4b90 100644 --- a/internal/scheduler/scheduler.go +++ b/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) diff --git a/internal/scheduler/scheduler_test.go b/internal/scheduler/scheduler_test.go index 9965d20ee..6920f4b53 100644 --- a/internal/scheduler/scheduler_test.go +++ b/internal/scheduler/scheduler_test.go @@ -39,6 +39,10 @@ func (m *mockRSMQ) SendMessage(qname string, message string, delay uint, opts .. return m.inner.SendMessage(qname, message, delay, opts...) } +func (m *mockRSMQ) ChangeMessageVisibility(qname string, id string, vt uint) error { + return m.inner.ChangeMessageVisibility(qname, id, vt) +} + func (m *mockRSMQ) DeleteMessage(qname string, id string) error { return m.inner.DeleteMessage(qname, id) } @@ -47,6 +51,18 @@ func (m *mockRSMQ) Quit() error { return m.inner.Quit() } +// immediateVisibilityRSMQ records requested visibility timeouts while making +// messages immediately visible so backoff tests do not wait in real time. +type immediateVisibilityRSMQ struct { + rsmq.Client + visibilityTimeouts chan uint +} + +func (m *immediateVisibilityRSMQ) ChangeMessageVisibility(qname string, id string, vt uint) error { + m.visibilityTimeouts <- vt + return m.Client.ChangeMessageVisibility(qname, id, 0) +} + // alwaysFailRSMQ is a test double that always fails ReceiveMessage. type alwaysFailRSMQ struct { err error @@ -59,8 +75,9 @@ func (m *alwaysFailRSMQ) ReceiveMessage(string, uint) (*rsmq.QueueMessage, error func (m *alwaysFailRSMQ) SendMessage(string, string, uint, ...rsmq.SendMessageOption) (string, error) { return "", nil } -func (m *alwaysFailRSMQ) DeleteMessage(string, string) error { return nil } -func (m *alwaysFailRSMQ) Quit() error { return nil } +func (m *alwaysFailRSMQ) ChangeMessageVisibility(string, string, uint) error { return nil } +func (m *alwaysFailRSMQ) DeleteMessage(string, string) error { return nil } +func (m *alwaysFailRSMQ) Quit() error { return nil } // createRSMQClient creates an RSMQ client for testing func createRSMQClient(t *testing.T, redisConfig *iredis.RedisConfig) *rsmq.RedisSMQ { @@ -360,6 +377,58 @@ func TestScheduler_Cancel(t *testing.T) { }) } +func TestScheduler_MaxReceiveCountMovesToDLQ(t *testing.T) { + t.Parallel() + + redisConfig := testutil.CreateTestRedisConfig(t) + rsmqClient := createRSMQClient(t, redisConfig) + immediateClient := &immediateVisibilityRSMQ{ + Client: rsmqClient, + visibilityTimeouts: make(chan uint, 2), + } + logger := testutil.CreateTestLogger(t) + + task := "poison_task" + var execCount atomic.Int64 + exec := func(_ context.Context, _ string) error { + execCount.Add(1) + return errors.New("permanent failure") + } + + ctx, cancel := context.WithCancel(context.Background()) + s := scheduler.New("scheduler", immediateClient, exec, + scheduler.WithVisibilityTimeout(1), + scheduler.WithPollBackoff(10*time.Millisecond), + scheduler.WithMaxReceiveCount(2), + scheduler.WithLogger(logger)) + require.NoError(t, s.Init(ctx)) + defer func() { cancel(); s.Shutdown() }() + + go s.Monitor(ctx) + + require.NoError(t, s.Schedule(ctx, task, 0)) + + var dlqMsg *rsmq.QueueMessage + require.Eventually(t, func() bool { + msg, err := rsmqClient.ReceiveMessage("scheduler-dlq", rsmq.UnsetVt) + if err != nil || msg == nil { + return false + } + dlqMsg = msg + return true + }, 10*time.Second, 100*time.Millisecond, "message should land in the DLQ") + + require.Equal(t, task, dlqMsg.Message, "DLQ message should carry the original task") + require.EqualValues(t, 2, execCount.Load(), "executor should run exactly maxReceiveCount times") + require.EqualValues(t, 1, <-immediateClient.visibilityTimeouts) + require.EqualValues(t, 2, <-immediateClient.visibilityTimeouts) + + // The message must be gone from the main queue. + msg, err := rsmqClient.ReceiveMessage("scheduler", rsmq.UnsetVt) + require.NoError(t, err) + require.Nil(t, msg, "main queue should be empty after DLQ routing") +} + func TestScheduler_MonitorRetriesTransientErrors(t *testing.T) { t.Parallel()