From 1309e7c05cecff50d734eacf612ab7e06b0b373b Mon Sep 17 00:00:00 2001 From: Lendrit Ibrahimi Date: Sat, 18 Jul 2026 20:05:20 +0200 Subject: [PATCH 1/2] feat(deliverymq): bound retry task receives with backoff and DLQ Failed retry tasks now back off exponentially using rsmq's receive count, instead of reappearing every 30s forever. Tasks received more than RETRY_MAX_RECEIVE_COUNT times are moved to the deliverymq-retry-dlq queue. --- docs/content/self-hosting/configuration.mdoc | 1 + internal/config/config.go | 19 ++-- internal/config/config_test.go | 3 + internal/config/logging.go | 1 + internal/config/validation.go | 4 + internal/config/validation_test.go | 36 +++++++ internal/deliverymq/retry.go | 24 ++++- internal/rsmq/rsmq.go | 6 +- internal/rsmq/rsmq_test.go | 4 +- internal/scheduler/backoff_test.go | 60 +++++++++++ internal/scheduler/scheduler.go | 103 ++++++++++++++++++- internal/scheduler/scheduler_test.go | 69 +++++++++++++ internal/services/builder.go | 3 + 13 files changed, 311 insertions(+), 22 deletions(-) create mode 100644 internal/scheduler/backoff_test.go diff --git a/docs/content/self-hosting/configuration.mdoc b/docs/content/self-hosting/configuration.mdoc index b0d1317ee..0797072d6 100644 --- a/docs/content/self-hosting/configuration.mdoc +++ b/docs/content/self-hosting/configuration.mdoc @@ -71,6 +71,7 @@ Choose one for event log persistence: | `MAX_RETRY_LIMIT` | `10` | Max retry attempts before giving up | | `RETRY_INTERVAL_SECONDS` | `30` | Base interval for exponential backoff retries | | `RETRY_SCHEDULE` | — | Comma-separated retry delays in seconds (overrides interval/limit) | +| `RETRY_MAX_RECEIVE_COUNT` | `5` | Max times a retry task is received before moving to the `deliverymq-retry-dlq` dead-letter queue | ## Topics diff --git a/internal/config/config.go b/internal/config/config.go index 25c4357c1..0c9f0ce74 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -82,6 +82,7 @@ type Config struct { RetryMaxLimit int `yaml:"retry_max_limit" env:"MAX_RETRY_LIMIT" desc:"Maximum number of retry attempts for a single event delivery before giving up. Ignored if retry_schedule is provided." required:"N"` RetryPollBackoffMs int `yaml:"retry_poll_backoff_ms" env:"RETRY_POLL_BACKOFF_MS" desc:"Backoff time in milliseconds when the retry monitor finds no messages to process. When a retry message is found, the monitor immediately polls for the next message without delay. Lower values provide faster retry processing but increase Redis load. For serverless Redis providers (Upstash, ElastiCache Serverless), consider increasing to 5000-10000ms to reduce costs. Default: 100" required:"N"` RetryVisibilityTimeoutSeconds int `yaml:"retry_visibility_timeout_seconds" env:"RETRY_VISIBILITY_TIMEOUT_SECONDS" desc:"Time in seconds a retry message is hidden after being received before becoming visible again for reprocessing. This applies when event data is temporarily unavailable (e.g., race condition with log persistence). Default: 30" required:"N"` + RetryMaxReceiveCount int `yaml:"retry_max_receive_count" env:"RETRY_MAX_RECEIVE_COUNT" desc:"Maximum number of times a retry task can be received before it is moved to the deliverymq-retry-dlq queue. This is separate from retry_max_limit, which bounds delivery attempts to the destination. Set to 0 to disable. Default: 5" required:"N"` // Event Delivery MaxDestinationsPerTenant int `yaml:"max_destinations_per_tenant" env:"MAX_DESTINATIONS_PER_TENANT" desc:"Maximum number of destinations allowed per tenant/organization." required:"N"` @@ -117,14 +118,15 @@ type Config struct { } var ( - ErrMismatchedServiceType = errors.New("config validation error: service type mismatch") - ErrInvalidServiceType = errors.New("config validation error: invalid service type") - ErrMissingRedis = errors.New("config validation error: redis configuration is required") - ErrMissingLogStorage = errors.New("config validation error: log storage must be provided") - ErrMissingMQs = errors.New("config validation error: message queue configuration is required") - ErrMissingAESSecret = errors.New("config validation error: AES encryption secret is required") - ErrInvalidPortalProxyURL = errors.New("config validation error: invalid portal proxy url") - ErrInvalidDeploymentID = errors.New("config validation error: deployment_id must contain only alphanumeric characters, hyphens, and underscores (max 64 characters)") + ErrMismatchedServiceType = errors.New("config validation error: service type mismatch") + ErrInvalidServiceType = errors.New("config validation error: invalid service type") + ErrMissingRedis = errors.New("config validation error: redis configuration is required") + ErrMissingLogStorage = errors.New("config validation error: log storage must be provided") + ErrMissingMQs = errors.New("config validation error: message queue configuration is required") + ErrMissingAESSecret = errors.New("config validation error: AES encryption secret is required") + ErrInvalidPortalProxyURL = errors.New("config validation error: invalid portal proxy url") + ErrInvalidDeploymentID = errors.New("config validation error: deployment_id must contain only alphanumeric characters, hyphens, and underscores (max 64 characters)") + ErrInvalidRetryMaxReceiveCount = errors.New("config validation error: retry_max_receive_count must be greater than or equal to 0") ) func (c *Config) InitDefaults() { @@ -170,6 +172,7 @@ func (c *Config) InitDefaults() { c.RetryMaxLimit = 10 c.RetryPollBackoffMs = 100 c.RetryVisibilityTimeoutSeconds = 30 + c.RetryMaxReceiveCount = 5 c.MaxDestinationsPerTenant = 20 c.DeliveryTimeoutSeconds = 5 c.PublishIdempotencyKeyTTL = 3600 // 1 hour diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 35332ba28..c6a31b826 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -73,6 +73,7 @@ func TestDefaultValues(t *testing.T) { assert.Equal(t, []int{}, cfg.RetrySchedule) assert.Equal(t, 30, cfg.RetryIntervalSeconds) assert.Equal(t, 10, cfg.RetryMaxLimit) + assert.Equal(t, 5, cfg.RetryMaxReceiveCount) assert.Equal(t, 20, cfg.MaxDestinationsPerTenant) assert.Equal(t, 5, cfg.DeliveryTimeoutSeconds) assert.Equal(t, "config/outpost/destinations", cfg.Destinations.MetadataPath) @@ -107,6 +108,7 @@ publish_max_concurrency: 5 delivery_max_concurrency: 5 log_max_concurrency: 5 retry_interval_seconds: 60 +retry_max_receive_count: 7 max_destinations_per_tenant: 50 delivery_timeout_seconds: 10 aes_encryption_secret: test-secret @@ -132,6 +134,7 @@ aes_encryption_secret: test-secret assert.Equal(t, 5, cfg.DeliveryMaxConcurrency) assert.Equal(t, 5, cfg.LogMaxConcurrency) assert.Equal(t, 60, cfg.RetryIntervalSeconds) + assert.Equal(t, 7, cfg.RetryMaxReceiveCount) assert.Equal(t, 50, cfg.MaxDestinationsPerTenant) assert.Equal(t, 10, cfg.DeliveryTimeoutSeconds) assert.Equal(t, "test-secret", cfg.AESEncryptionSecret) diff --git a/internal/config/logging.go b/internal/config/logging.go index 94c625630..d973f6ac6 100644 --- a/internal/config/logging.go +++ b/internal/config/logging.go @@ -98,6 +98,7 @@ func (c *Config) LogConfigurationSummary() []zap.Field { zap.Ints("retry_schedule", c.RetrySchedule), zap.Int("retry_interval_seconds", c.RetryIntervalSeconds), zap.Int("retry_max_limit", c.RetryMaxLimit), + zap.Int("retry_max_receive_count", c.RetryMaxReceiveCount), // Event Delivery zap.Int("max_destinations_per_tenant", c.MaxDestinationsPerTenant), diff --git a/internal/config/validation.go b/internal/config/validation.go index 377c65d80..26d40bb41 100644 --- a/internal/config/validation.go +++ b/internal/config/validation.go @@ -180,6 +180,10 @@ func (c *Config) validateDeploymentID() error { // validateRetryConfiguration validates and adjusts the retry configuration func (c *Config) validateRetryConfiguration() error { + if c.RetryMaxReceiveCount < 0 { + return ErrInvalidRetryMaxReceiveCount + } + // If retry_schedule is provided, override retry_max_limit to match schedule length if len(c.RetrySchedule) > 0 { c.RetryMaxLimit = len(c.RetrySchedule) diff --git a/internal/config/validation_test.go b/internal/config/validation_test.go index 423e1fe51..f122a1e2e 100644 --- a/internal/config/validation_test.go +++ b/internal/config/validation_test.go @@ -338,6 +338,42 @@ func TestMisc(t *testing.T) { } } +func TestRetryMaxReceiveCount(t *testing.T) { + tests := []struct { + name string + maxReceiveCount int + wantErr error + }{ + { + name: "positive value is valid", + maxReceiveCount: 5, + }, + { + name: "zero disables the limit", + maxReceiveCount: 0, + }, + { + name: "negative value is invalid", + maxReceiveCount: -1, + wantErr: config.ErrInvalidRetryMaxReceiveCount, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + cfg := validConfig() + cfg.RetryMaxReceiveCount = tt.maxReceiveCount + + err := cfg.Validate(config.Flags{}) + if tt.wantErr != nil { + assert.ErrorIs(t, err, tt.wantErr) + } else { + assert.NoError(t, err) + } + }) + } +} + func TestOpenTelemetry(t *testing.T) { tests := []struct { name string diff --git a/internal/deliverymq/retry.go b/internal/deliverymq/retry.go index 8196316e1..a67ae51e2 100644 --- a/internal/deliverymq/retry.go +++ b/internal/deliverymq/retry.go @@ -26,6 +26,7 @@ type RetrySchedulerOption func(*retrySchedulerConfig) type retrySchedulerConfig struct { visibilityTimeout uint + maxReceiveCount uint64 } // WithRetryVisibilityTimeout sets the visibility timeout for the retry scheduler queue. @@ -37,6 +38,15 @@ func WithRetryVisibilityTimeout(vt uint) RetrySchedulerOption { } } +// WithRetryMaxReceiveCount limits how many times a retry task can be received +// before it is moved to the "deliverymq-retry-dlq" dead-letter queue. This is +// separate from the delivery retry max limit; 0 disables the limit. +func WithRetryMaxReceiveCount(n uint64) RetrySchedulerOption { + return func(c *retrySchedulerConfig) { + c.maxReceiveCount = n + } +} + func NewRetryScheduler(deliverymq *DeliveryMQ, redisConfig *redis.RedisConfig, deploymentID string, pollBackoff time.Duration, logger *logging.Logger, eventGetter RetryEventGetter, opts ...RetrySchedulerOption) (scheduler.Scheduler, error) { cfg := &retrySchedulerConfig{} for _, opt := range opts { @@ -112,13 +122,17 @@ func NewRetryScheduler(deliverymq *DeliveryMQ, redisConfig *redis.RedisConfig, d return nil } + schedulerOpts := []scheduler.Option{ + scheduler.WithPollBackoff(pollBackoff), + scheduler.WithLogger(logger), + } 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)) + } + if cfg.maxReceiveCount > 0 { + schedulerOpts = append(schedulerOpts, scheduler.WithMaxReceiveCount(cfg.maxReceiveCount)) } - 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..6bca6580f 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 diff --git a/internal/rsmq/rsmq_test.go b/internal/rsmq/rsmq_test.go index db9e76d94..0941a7447 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) { 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..b90cd2001 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,6 +75,7 @@ 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) ChangeMessageVisibility(string, string, uint) error { return nil } func (m *alwaysFailRSMQ) DeleteMessage(string, string) error { return nil } func (m *alwaysFailRSMQ) Quit() error { return nil } @@ -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() diff --git a/internal/services/builder.go b/internal/services/builder.go index 950f4650c..be0c08573 100644 --- a/internal/services/builder.go +++ b/internal/services/builder.go @@ -611,6 +611,9 @@ func (s *serviceInstance) initRetryScheduler(ctx context.Context, cfg *config.Co if cfg.RetryVisibilityTimeoutSeconds > 0 { retrySchedulerOpts = append(retrySchedulerOpts, deliverymq.WithRetryVisibilityTimeout(uint(cfg.RetryVisibilityTimeoutSeconds))) } + if cfg.RetryMaxReceiveCount > 0 { + retrySchedulerOpts = append(retrySchedulerOpts, deliverymq.WithRetryMaxReceiveCount(uint64(cfg.RetryMaxReceiveCount))) + } retryScheduler, err := deliverymq.NewRetryScheduler(s.deliveryMQ, cfg.Redis.ToConfig(), cfg.DeploymentID, pollBackoff, logger, s.logStore, retrySchedulerOpts...) if err != nil { logger.Error("failed to create delivery MQ retry scheduler", zap.String("service", s.name), zap.Error(err)) From f887cec15c19b871f6abad30fa021e669f3cd132 Mon Sep 17 00:00:00 2001 From: Lendrit Ibrahimi Date: Wed, 22 Jul 2026 14:58:49 +0200 Subject: [PATCH 2/2] fix(deliverymq): hardcode retry max receive count, reset rc on schedule overwrite --- docs/content/self-hosting/configuration.mdoc | 1 - internal/config/config.go | 19 ++++----- internal/config/config_test.go | 3 -- internal/config/logging.go | 1 - internal/config/validation.go | 4 -- internal/config/validation_test.go | 36 ----------------- internal/deliverymq/retry.go | 18 +++------ internal/rsmq/rsmq.go | 3 ++ internal/rsmq/rsmq_test.go | 42 ++++++++++++++++++++ internal/scheduler/scheduler_test.go | 4 +- internal/services/builder.go | 3 -- 11 files changed, 60 insertions(+), 74 deletions(-) diff --git a/docs/content/self-hosting/configuration.mdoc b/docs/content/self-hosting/configuration.mdoc index 0797072d6..b0d1317ee 100644 --- a/docs/content/self-hosting/configuration.mdoc +++ b/docs/content/self-hosting/configuration.mdoc @@ -71,7 +71,6 @@ Choose one for event log persistence: | `MAX_RETRY_LIMIT` | `10` | Max retry attempts before giving up | | `RETRY_INTERVAL_SECONDS` | `30` | Base interval for exponential backoff retries | | `RETRY_SCHEDULE` | — | Comma-separated retry delays in seconds (overrides interval/limit) | -| `RETRY_MAX_RECEIVE_COUNT` | `5` | Max times a retry task is received before moving to the `deliverymq-retry-dlq` dead-letter queue | ## Topics diff --git a/internal/config/config.go b/internal/config/config.go index 0c9f0ce74..25c4357c1 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -82,7 +82,6 @@ type Config struct { RetryMaxLimit int `yaml:"retry_max_limit" env:"MAX_RETRY_LIMIT" desc:"Maximum number of retry attempts for a single event delivery before giving up. Ignored if retry_schedule is provided." required:"N"` RetryPollBackoffMs int `yaml:"retry_poll_backoff_ms" env:"RETRY_POLL_BACKOFF_MS" desc:"Backoff time in milliseconds when the retry monitor finds no messages to process. When a retry message is found, the monitor immediately polls for the next message without delay. Lower values provide faster retry processing but increase Redis load. For serverless Redis providers (Upstash, ElastiCache Serverless), consider increasing to 5000-10000ms to reduce costs. Default: 100" required:"N"` RetryVisibilityTimeoutSeconds int `yaml:"retry_visibility_timeout_seconds" env:"RETRY_VISIBILITY_TIMEOUT_SECONDS" desc:"Time in seconds a retry message is hidden after being received before becoming visible again for reprocessing. This applies when event data is temporarily unavailable (e.g., race condition with log persistence). Default: 30" required:"N"` - RetryMaxReceiveCount int `yaml:"retry_max_receive_count" env:"RETRY_MAX_RECEIVE_COUNT" desc:"Maximum number of times a retry task can be received before it is moved to the deliverymq-retry-dlq queue. This is separate from retry_max_limit, which bounds delivery attempts to the destination. Set to 0 to disable. Default: 5" required:"N"` // Event Delivery MaxDestinationsPerTenant int `yaml:"max_destinations_per_tenant" env:"MAX_DESTINATIONS_PER_TENANT" desc:"Maximum number of destinations allowed per tenant/organization." required:"N"` @@ -118,15 +117,14 @@ type Config struct { } var ( - ErrMismatchedServiceType = errors.New("config validation error: service type mismatch") - ErrInvalidServiceType = errors.New("config validation error: invalid service type") - ErrMissingRedis = errors.New("config validation error: redis configuration is required") - ErrMissingLogStorage = errors.New("config validation error: log storage must be provided") - ErrMissingMQs = errors.New("config validation error: message queue configuration is required") - ErrMissingAESSecret = errors.New("config validation error: AES encryption secret is required") - ErrInvalidPortalProxyURL = errors.New("config validation error: invalid portal proxy url") - ErrInvalidDeploymentID = errors.New("config validation error: deployment_id must contain only alphanumeric characters, hyphens, and underscores (max 64 characters)") - ErrInvalidRetryMaxReceiveCount = errors.New("config validation error: retry_max_receive_count must be greater than or equal to 0") + ErrMismatchedServiceType = errors.New("config validation error: service type mismatch") + ErrInvalidServiceType = errors.New("config validation error: invalid service type") + ErrMissingRedis = errors.New("config validation error: redis configuration is required") + ErrMissingLogStorage = errors.New("config validation error: log storage must be provided") + ErrMissingMQs = errors.New("config validation error: message queue configuration is required") + ErrMissingAESSecret = errors.New("config validation error: AES encryption secret is required") + ErrInvalidPortalProxyURL = errors.New("config validation error: invalid portal proxy url") + ErrInvalidDeploymentID = errors.New("config validation error: deployment_id must contain only alphanumeric characters, hyphens, and underscores (max 64 characters)") ) func (c *Config) InitDefaults() { @@ -172,7 +170,6 @@ func (c *Config) InitDefaults() { c.RetryMaxLimit = 10 c.RetryPollBackoffMs = 100 c.RetryVisibilityTimeoutSeconds = 30 - c.RetryMaxReceiveCount = 5 c.MaxDestinationsPerTenant = 20 c.DeliveryTimeoutSeconds = 5 c.PublishIdempotencyKeyTTL = 3600 // 1 hour diff --git a/internal/config/config_test.go b/internal/config/config_test.go index c6a31b826..35332ba28 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -73,7 +73,6 @@ func TestDefaultValues(t *testing.T) { assert.Equal(t, []int{}, cfg.RetrySchedule) assert.Equal(t, 30, cfg.RetryIntervalSeconds) assert.Equal(t, 10, cfg.RetryMaxLimit) - assert.Equal(t, 5, cfg.RetryMaxReceiveCount) assert.Equal(t, 20, cfg.MaxDestinationsPerTenant) assert.Equal(t, 5, cfg.DeliveryTimeoutSeconds) assert.Equal(t, "config/outpost/destinations", cfg.Destinations.MetadataPath) @@ -108,7 +107,6 @@ publish_max_concurrency: 5 delivery_max_concurrency: 5 log_max_concurrency: 5 retry_interval_seconds: 60 -retry_max_receive_count: 7 max_destinations_per_tenant: 50 delivery_timeout_seconds: 10 aes_encryption_secret: test-secret @@ -134,7 +132,6 @@ aes_encryption_secret: test-secret assert.Equal(t, 5, cfg.DeliveryMaxConcurrency) assert.Equal(t, 5, cfg.LogMaxConcurrency) assert.Equal(t, 60, cfg.RetryIntervalSeconds) - assert.Equal(t, 7, cfg.RetryMaxReceiveCount) assert.Equal(t, 50, cfg.MaxDestinationsPerTenant) assert.Equal(t, 10, cfg.DeliveryTimeoutSeconds) assert.Equal(t, "test-secret", cfg.AESEncryptionSecret) diff --git a/internal/config/logging.go b/internal/config/logging.go index d973f6ac6..94c625630 100644 --- a/internal/config/logging.go +++ b/internal/config/logging.go @@ -98,7 +98,6 @@ func (c *Config) LogConfigurationSummary() []zap.Field { zap.Ints("retry_schedule", c.RetrySchedule), zap.Int("retry_interval_seconds", c.RetryIntervalSeconds), zap.Int("retry_max_limit", c.RetryMaxLimit), - zap.Int("retry_max_receive_count", c.RetryMaxReceiveCount), // Event Delivery zap.Int("max_destinations_per_tenant", c.MaxDestinationsPerTenant), diff --git a/internal/config/validation.go b/internal/config/validation.go index 26d40bb41..377c65d80 100644 --- a/internal/config/validation.go +++ b/internal/config/validation.go @@ -180,10 +180,6 @@ func (c *Config) validateDeploymentID() error { // validateRetryConfiguration validates and adjusts the retry configuration func (c *Config) validateRetryConfiguration() error { - if c.RetryMaxReceiveCount < 0 { - return ErrInvalidRetryMaxReceiveCount - } - // If retry_schedule is provided, override retry_max_limit to match schedule length if len(c.RetrySchedule) > 0 { c.RetryMaxLimit = len(c.RetrySchedule) diff --git a/internal/config/validation_test.go b/internal/config/validation_test.go index f122a1e2e..423e1fe51 100644 --- a/internal/config/validation_test.go +++ b/internal/config/validation_test.go @@ -338,42 +338,6 @@ func TestMisc(t *testing.T) { } } -func TestRetryMaxReceiveCount(t *testing.T) { - tests := []struct { - name string - maxReceiveCount int - wantErr error - }{ - { - name: "positive value is valid", - maxReceiveCount: 5, - }, - { - name: "zero disables the limit", - maxReceiveCount: 0, - }, - { - name: "negative value is invalid", - maxReceiveCount: -1, - wantErr: config.ErrInvalidRetryMaxReceiveCount, - }, - } - - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - cfg := validConfig() - cfg.RetryMaxReceiveCount = tt.maxReceiveCount - - err := cfg.Validate(config.Flags{}) - if tt.wantErr != nil { - assert.ErrorIs(t, err, tt.wantErr) - } else { - assert.NoError(t, err) - } - }) - } -} - func TestOpenTelemetry(t *testing.T) { tests := []struct { name string diff --git a/internal/deliverymq/retry.go b/internal/deliverymq/retry.go index a67ae51e2..e43bd38a8 100644 --- a/internal/deliverymq/retry.go +++ b/internal/deliverymq/retry.go @@ -21,12 +21,15 @@ 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) type retrySchedulerConfig struct { visibilityTimeout uint - maxReceiveCount uint64 } // WithRetryVisibilityTimeout sets the visibility timeout for the retry scheduler queue. @@ -38,15 +41,6 @@ func WithRetryVisibilityTimeout(vt uint) RetrySchedulerOption { } } -// WithRetryMaxReceiveCount limits how many times a retry task can be received -// before it is moved to the "deliverymq-retry-dlq" dead-letter queue. This is -// separate from the delivery retry max limit; 0 disables the limit. -func WithRetryMaxReceiveCount(n uint64) RetrySchedulerOption { - return func(c *retrySchedulerConfig) { - c.maxReceiveCount = n - } -} - func NewRetryScheduler(deliverymq *DeliveryMQ, redisConfig *redis.RedisConfig, deploymentID string, pollBackoff time.Duration, logger *logging.Logger, eventGetter RetryEventGetter, opts ...RetrySchedulerOption) (scheduler.Scheduler, error) { cfg := &retrySchedulerConfig{} for _, opt := range opts { @@ -125,13 +119,11 @@ func NewRetryScheduler(deliverymq *DeliveryMQ, redisConfig *redis.RedisConfig, d schedulerOpts := []scheduler.Option{ scheduler.WithPollBackoff(pollBackoff), scheduler.WithLogger(logger), + scheduler.WithMaxReceiveCount(retryMaxReceiveCount), } if cfg.visibilityTimeout > 0 { schedulerOpts = append(schedulerOpts, scheduler.WithVisibilityTimeout(cfg.visibilityTimeout)) } - if cfg.maxReceiveCount > 0 { - schedulerOpts = append(schedulerOpts, scheduler.WithMaxReceiveCount(cfg.maxReceiveCount)) - } return scheduler.New("deliverymq-retry", rsmqClient, exec, schedulerOpts...), nil } diff --git a/internal/rsmq/rsmq.go b/internal/rsmq/rsmq.go index 6bca6580f..7eb104984 100644 --- a/internal/rsmq/rsmq.go +++ b/internal/rsmq/rsmq.go @@ -512,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 0941a7447..d264a0fc2 100644 --- a/internal/rsmq/rsmq_test.go +++ b/internal/rsmq/rsmq_test.go @@ -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/scheduler_test.go b/internal/scheduler/scheduler_test.go index b90cd2001..6920f4b53 100644 --- a/internal/scheduler/scheduler_test.go +++ b/internal/scheduler/scheduler_test.go @@ -76,8 +76,8 @@ func (m *alwaysFailRSMQ) SendMessage(string, string, uint, ...rsmq.SendMessageOp 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 } +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 { diff --git a/internal/services/builder.go b/internal/services/builder.go index be0c08573..950f4650c 100644 --- a/internal/services/builder.go +++ b/internal/services/builder.go @@ -611,9 +611,6 @@ func (s *serviceInstance) initRetryScheduler(ctx context.Context, cfg *config.Co if cfg.RetryVisibilityTimeoutSeconds > 0 { retrySchedulerOpts = append(retrySchedulerOpts, deliverymq.WithRetryVisibilityTimeout(uint(cfg.RetryVisibilityTimeoutSeconds))) } - if cfg.RetryMaxReceiveCount > 0 { - retrySchedulerOpts = append(retrySchedulerOpts, deliverymq.WithRetryMaxReceiveCount(uint64(cfg.RetryMaxReceiveCount))) - } retryScheduler, err := deliverymq.NewRetryScheduler(s.deliveryMQ, cfg.Redis.ToConfig(), cfg.DeploymentID, pollBackoff, logger, s.logStore, retrySchedulerOpts...) if err != nil { logger.Error("failed to create delivery MQ retry scheduler", zap.String("service", s.name), zap.Error(err))