From 5774143f50c31c42a114d658d024ee423b1e0bc0 Mon Sep 17 00:00:00 2001 From: mnoah1 Date: Mon, 31 Aug 2026 14:52:27 +0000 Subject: [PATCH 1/5] fix(messagequeue): apply retry backoff to nacked messages MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Summary: Intent: - Ensure explicit Nacks honor the existing retry backoff configuration instead of retrying on every poll cycle. - Prevent transient downstream failures from immediately consuming the retry budget and dead-lettering messages. Changes: - Calculate each retry delay from the delivery attempt and configured initial delay, multiplier, and maximum. - Persist the delayed visibility timestamp so backoff survives restarts and subscriber handoffs. - Cover delay calculation, overflow handling, persistence, and redelivery behavior with unit and integration tests. --- Generated by the 🪄 [pr-create](https://sg.uberinternal.com/code.uber.internal/uber-code/devexp-agent-marketplace/-/blob/claude-code/plugins/dev/uber-dev/skills/pr-create/SKILL.md) skill in devexp-agent-marketplace --- platform/consumer/consumer.go | 3 - .../mysql/delivery_state_store.go | 14 ++-- .../mysql/delivery_state_store_test.go | 29 +++++++- .../messagequeue/mysql/mock_stores.go | 8 +-- .../extension/messagequeue/mysql/stores.go | 5 +- .../messagequeue/mysql/subscriber.go | 31 +++++++- .../messagequeue/mysql/subscriber_test.go | 71 +++++++++++++++---- .../messagequeue/subscription_config.go | 9 ++- .../messagequeue/mysql/queue_test.go | 44 +++++++++--- 9 files changed, 171 insertions(+), 43 deletions(-) diff --git a/platform/consumer/consumer.go b/platform/consumer/consumer.go index 9e1fb03af..687d15663 100644 --- a/platform/consumer/consumer.go +++ b/platform/consumer/consumer.go @@ -489,9 +489,6 @@ func (m *consumer) processDelivery(ctx context.Context, controller Controller, d "elapsed_ms", elapsed.Milliseconds(), ) - // Nack requeues immediately - the visibility timeout spaces retries. - // The failure travels with it so that the attempt which finally spends - // the retry budget can dead-letter saying why. nackOp := metrics.Begin(controllerScope, "nack", metrics.StorageLatencyBuckets, metrics.TagsFromContext(ctx)...) nackErr := delivery.Nack(ctx, controllerFailure) nackOp.Complete(nackErr) diff --git a/platform/extension/messagequeue/mysql/delivery_state_store.go b/platform/extension/messagequeue/mysql/delivery_state_store.go index 712247c2f..c149bf9b8 100644 --- a/platform/extension/messagequeue/mysql/delivery_state_store.go +++ b/platform/extension/messagequeue/mysql/delivery_state_store.go @@ -18,6 +18,7 @@ import ( "context" "database/sql" "fmt" + "math" "time" "github.com/uber-go/tally" @@ -146,16 +147,21 @@ func (s *sqldeliveryStateStore) MarkAcked(ctx context.Context, consumerGroup, to return nil } -// MarkNacked sets invisible_until = now, making the message immediately -// eligible for redelivery on the next poll. +// MarkNacked makes the message eligible for redelivery after delayMs. // retry_count is NOT incremented here — it is incremented by MarkDelivered on redelivery. -func (s *sqldeliveryStateStore) MarkNacked(ctx context.Context, consumerGroup, topic, partitionKey string, offset int64) (retErr error) { +func (s *sqldeliveryStateStore) MarkNacked(ctx context.Context, consumerGroup, topic, partitionKey string, offset int64, delayMs int64) (retErr error) { op := metrics.Begin(s.scope, "mark_nacked", metrics.StorageLatencyBuckets, metrics.NewTag("topic", topic), metrics.NewTag("consumer_group", consumerGroup)) defer func() { op.Complete(retErr) }() - invisibleUntil := time.Now().UnixMilli() + nowMs := time.Now().UnixMilli() + invisibleUntil := nowMs + if delayMs > math.MaxInt64-nowMs { + invisibleUntil = math.MaxInt64 + } else if delayMs > 0 { + invisibleUntil += delayMs + } _, err := s.db.ExecContext(ctx, fmt.Sprintf(` INSERT INTO %s (consumer_group, topic, partition_key, message_offset, acked, invisible_until, retry_count) diff --git a/platform/extension/messagequeue/mysql/delivery_state_store_test.go b/platform/extension/messagequeue/mysql/delivery_state_store_test.go index 89235b8d6..cd8adfe37 100644 --- a/platform/extension/messagequeue/mysql/delivery_state_store_test.go +++ b/platform/extension/messagequeue/mysql/delivery_state_store_test.go @@ -18,7 +18,9 @@ import ( "context" "database/sql" "database/sql/driver" + "math" "testing" + "time" "github.com/DATA-DOG/go-sqlmock" "github.com/stretchr/testify/assert" @@ -27,6 +29,13 @@ import ( "go.uber.org/zap/zaptest" ) +type unixMillisAtLeast int64 + +func (minimum unixMillisAtLeast) Match(value driver.Value) bool { + got, ok := value.(int64) + return ok && got >= int64(minimum) +} + func newTestDeliveryStateStoreWithMock(t *testing.T) (deliveryStateStore, *sql.DB, sqlmock.Sqlmock) { t.Helper() db, mock, err := sqlmock.New() @@ -201,18 +210,20 @@ func TestDeliveryStateStore_MarkNacked(t *testing.T) { t.Run(tt.name, func(t *testing.T) { store, db, mock := newTestDeliveryStateStoreWithMock(t) defer db.Close() + const retryDelayMs = int64(1000) + invisibleUntil := unixMillisAtLeast(time.Now().UnixMilli() + retryDelayMs) if tt.wantErr { mock.ExpectExec("INSERT INTO queue_delivery_state"). - WithArgs("group-1", "orders", "part-1", int64(5), sqlmock.AnyArg()). + WithArgs("group-1", "orders", "part-1", int64(5), invisibleUntil). WillReturnError(assert.AnError) } else { mock.ExpectExec("INSERT INTO queue_delivery_state"). - WithArgs("group-1", "orders", "part-1", int64(5), sqlmock.AnyArg()). + WithArgs("group-1", "orders", "part-1", int64(5), invisibleUntil). WillReturnResult(sqlmock.NewResult(1, 1)) } - err := store.MarkNacked(context.Background(), "group-1", "orders", "part-1", 5) + err := store.MarkNacked(context.Background(), "group-1", "orders", "part-1", 5, retryDelayMs) if tt.wantErr { require.Error(t, err) @@ -224,6 +235,18 @@ func TestDeliveryStateStore_MarkNacked(t *testing.T) { } } +func TestDeliveryStateStore_MarkNackedSaturatesTimestamp(t *testing.T) { + store, db, mock := newTestDeliveryStateStoreWithMock(t) + defer db.Close() + + mock.ExpectExec("INSERT INTO queue_delivery_state"). + WithArgs("group-1", "orders", "part-1", int64(5), int64(math.MaxInt64)). + WillReturnResult(sqlmock.NewResult(1, 1)) + + require.NoError(t, store.MarkNacked(context.Background(), "group-1", "orders", "part-1", 5, math.MaxInt64)) + assert.NoError(t, mock.ExpectationsWereMet()) +} + func TestDeliveryStateStore_MarkPostponed(t *testing.T) { tests := []struct { name string diff --git a/platform/extension/messagequeue/mysql/mock_stores.go b/platform/extension/messagequeue/mysql/mock_stores.go index c787beaca..614c93d11 100644 --- a/platform/extension/messagequeue/mysql/mock_stores.go +++ b/platform/extension/messagequeue/mysql/mock_stores.go @@ -533,17 +533,17 @@ func (mr *MockdeliveryStateStoreMockRecorder) MarkDelivered(ctx, consumerGroup, } // MarkNacked mocks base method. -func (m *MockdeliveryStateStore) MarkNacked(ctx context.Context, consumerGroup, topic, partitionKey string, offset int64) error { +func (m *MockdeliveryStateStore) MarkNacked(ctx context.Context, consumerGroup, topic, partitionKey string, offset, delayMs int64) error { m.ctrl.T.Helper() - ret := m.ctrl.Call(m, "MarkNacked", ctx, consumerGroup, topic, partitionKey, offset) + ret := m.ctrl.Call(m, "MarkNacked", ctx, consumerGroup, topic, partitionKey, offset, delayMs) ret0, _ := ret[0].(error) return ret0 } // MarkNacked indicates an expected call of MarkNacked. -func (mr *MockdeliveryStateStoreMockRecorder) MarkNacked(ctx, consumerGroup, topic, partitionKey, offset any) *gomock.Call { +func (mr *MockdeliveryStateStoreMockRecorder) MarkNacked(ctx, consumerGroup, topic, partitionKey, offset, delayMs any) *gomock.Call { mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "MarkNacked", reflect.TypeOf((*MockdeliveryStateStore)(nil).MarkNacked), ctx, consumerGroup, topic, partitionKey, offset) + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "MarkNacked", reflect.TypeOf((*MockdeliveryStateStore)(nil).MarkNacked), ctx, consumerGroup, topic, partitionKey, offset, delayMs) } // MarkPostponed mocks base method. diff --git a/platform/extension/messagequeue/mysql/stores.go b/platform/extension/messagequeue/mysql/stores.go index c54fc2465..199eeb314 100644 --- a/platform/extension/messagequeue/mysql/stores.go +++ b/platform/extension/messagequeue/mysql/stores.go @@ -218,9 +218,8 @@ type deliveryStateStore interface { // MarkAcked sets acked = TRUE to indicate this group has processed the message. MarkAcked(ctx context.Context, consumerGroup, topic, partitionKey string, offset int64) error - // MarkNacked makes the message immediately eligible for redelivery - // (invisible_until = now). - MarkNacked(ctx context.Context, consumerGroup, topic, partitionKey string, offset int64) error + // MarkNacked makes the message eligible for redelivery after delayMs. + MarkNacked(ctx context.Context, consumerGroup, topic, partitionKey string, offset int64, delayMs int64) error // MarkPostponed sets invisible_until = now + delay, resets retry_count, and // sets the postponed flag. The message becomes a partition barrier until it diff --git a/platform/extension/messagequeue/mysql/subscriber.go b/platform/extension/messagequeue/mysql/subscriber.go index 6e655c07f..a665d49d9 100644 --- a/platform/extension/messagequeue/mysql/subscriber.go +++ b/platform/extension/messagequeue/mysql/subscriber.go @@ -18,6 +18,7 @@ import ( "context" "errors" "fmt" + "math" "sort" "strconv" "sync" @@ -331,9 +332,8 @@ func (d *sqlDelivery) Nack(ctx context.Context, f failure.Failure) error { return d.deadLetter(ctx, f) } - // Mark as nacked in delivery state (per consumer group): immediately - // eligible for redelivery on the next poll. - if err := d.subscriber.deliveryStateStore.MarkNacked(ctx, d.consumerGroup, d.topic, d.partitionKey, d.offset); err != nil { + retryDelayMs := retryBackoffMs(d.retry, d.attempt) + if err := d.subscriber.deliveryStateStore.MarkNacked(ctx, d.consumerGroup, d.topic, d.partitionKey, d.offset, retryDelayMs); err != nil { return err } @@ -341,6 +341,7 @@ func (d *sqlDelivery) Nack(ctx context.Context, f failure.Failure) error { "topic", d.topic, "partition_key", d.partitionKey, "message_id", d.messageID, + "retry_delay_ms", retryDelayMs, ) d.acknowledged = true @@ -1508,3 +1509,27 @@ func (s *subscriber) Close() (retErr error) { s.logger.Infow("subscriber closed") return nil } + +func retryBackoffMs(retry extqueue.RetryConfig, attempt int) int64 { + backoffMs := retry.InitialBackoffMs + if backoffMs <= 0 { + return 0 + } + if retry.MaxBackoffMs > 0 && backoffMs >= retry.MaxBackoffMs { + return retry.MaxBackoffMs + } + + multiplier := retry.BackoffMultiplier + if multiplier <= 1 || math.IsNaN(multiplier) || attempt <= 1 { + return backoffMs + } + + backoff := float64(backoffMs) * math.Pow(multiplier, float64(attempt-1)) + if retry.MaxBackoffMs > 0 && backoff >= float64(retry.MaxBackoffMs) { + return retry.MaxBackoffMs + } + if backoff >= float64(math.MaxInt64) { + return math.MaxInt64 + } + return int64(backoff) +} diff --git a/platform/extension/messagequeue/mysql/subscriber_test.go b/platform/extension/messagequeue/mysql/subscriber_test.go index 91f680f09..307cdc1fa 100644 --- a/platform/extension/messagequeue/mysql/subscriber_test.go +++ b/platform/extension/messagequeue/mysql/subscriber_test.go @@ -18,6 +18,7 @@ import ( "context" "errors" "fmt" + "math" "testing" "time" @@ -64,7 +65,7 @@ func newTestDeliveryStateStore(ctrl *gomock.Controller) *MockdeliveryStateStore mockDS := NewMockdeliveryStateStore(ctrl) mockDS.EXPECT().MarkDelivered(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(0, nil).AnyTimes() mockDS.EXPECT().MarkAcked(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes() - mockDS.EXPECT().MarkNacked(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes() + mockDS.EXPECT().MarkNacked(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes() mockDS.EXPECT().GetDeliveryState(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(DeliveryState{}, false, nil).AnyTimes() mockDS.EXPECT().AdvanceWatermark(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(int64(0), nil).AnyTimes() mockDS.EXPECT().ExtendVisibility(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes() @@ -444,18 +445,64 @@ func TestSQLDelivery_Reject(t *testing.T) { // early would silently cost every message a retry. func TestSQLDelivery_NackDeadLettersWhenBudgetSpent(t *testing.T) { tests := []struct { - name string - attempt int - maxAttempts int - wantDLQ bool + name string + attempt int + retry extqueue.RetryConfig + wantDLQ bool + wantRetryDelayMs int64 }{ - {name: "budget remaining", attempt: 1, maxAttempts: 3}, - {name: "one attempt left", attempt: 2, maxAttempts: 3}, - {name: "final attempt dead-letters", attempt: 3, maxAttempts: 3, wantDLQ: true}, - {name: "single-attempt budget dead-letters at once", attempt: 1, maxAttempts: 1, wantDLQ: true}, + { + name: "first retry uses initial delay", attempt: 1, + retry: extqueue.RetryConfig{MaxAttempts: 3, InitialBackoffMs: 1000, MaxBackoffMs: 30000, BackoffMultiplier: 2}, + wantRetryDelayMs: 1000, + }, + { + name: "second retry multiplies delay", attempt: 2, + retry: extqueue.RetryConfig{MaxAttempts: 3, InitialBackoffMs: 1000, MaxBackoffMs: 30000, BackoffMultiplier: 2}, + wantRetryDelayMs: 2000, + }, + { + name: "delay is capped", attempt: 6, + retry: extqueue.RetryConfig{MaxAttempts: 10, InitialBackoffMs: 1000, MaxBackoffMs: 30000, BackoffMultiplier: 2}, + wantRetryDelayMs: 30000, + }, + { + name: "initial delay is capped", attempt: 1, + retry: extqueue.RetryConfig{MaxAttempts: 3, InitialBackoffMs: 5000, MaxBackoffMs: 2000, BackoffMultiplier: 2}, + wantRetryDelayMs: 2000, + }, + { + name: "unset multiplier uses constant delay", attempt: 2, + retry: extqueue.RetryConfig{MaxAttempts: 3, InitialBackoffMs: 1000, MaxBackoffMs: 30000}, + wantRetryDelayMs: 1000, + }, + { + name: "fractional multiplier compounds from initial delay", attempt: 3, + retry: extqueue.RetryConfig{MaxAttempts: 4, InitialBackoffMs: 1, BackoffMultiplier: 1.5}, + wantRetryDelayMs: 2, + }, + { + name: "uncapped overflow saturates", attempt: 2, + retry: extqueue.RetryConfig{MaxAttempts: 3, InitialBackoffMs: math.MaxInt64, BackoffMultiplier: 2}, + wantRetryDelayMs: math.MaxInt64, + }, + { + name: "unset initial delay retries immediately", attempt: 1, + retry: extqueue.RetryConfig{MaxAttempts: 3}, + }, + { + name: "final attempt dead-letters", attempt: 3, + retry: extqueue.RetryConfig{MaxAttempts: 3, InitialBackoffMs: 1000, MaxBackoffMs: 30000, BackoffMultiplier: 2}, + wantDLQ: true, + }, + { + name: "single-attempt budget dead-letters at once", attempt: 1, + retry: extqueue.RetryConfig{MaxAttempts: 1}, + wantDLQ: true, + }, // A zero budget is not "dead-letter immediately" — it is unconfigured, // and the poll loop still governs. - {name: "unset budget never dead-letters here", attempt: 9, maxAttempts: 0}, + {name: "unset budget never dead-letters here", attempt: 9}, } for _, tt := range tests { @@ -477,7 +524,7 @@ func TestSQLDelivery_NackDeadLettersWhenBudgetSpent(t *testing.T) { ) dlqConfig := extqueue.DLQConfig{Enabled: true, TopicSuffix: "_dlq"} - d := newDeliveryForTest(sub, tt.attempt, dlqConfig, extqueue.RetryConfig{MaxAttempts: tt.maxAttempts}) + d := newDeliveryForTest(sub, tt.attempt, dlqConfig, tt.retry) f := failure.New("boom", failure.Subject{Type: "batch", ID: "q/batch/1"}) @@ -490,7 +537,7 @@ func TestSQLDelivery_NackDeadLettersWhenBudgetSpent(t *testing.T) { ).Return(nil) } else { mockDeliveryState.EXPECT().MarkNacked( - gomock.Any(), "test-group", "test_topic", "part-1", int64(100), + gomock.Any(), "test-group", "test_topic", "part-1", int64(100), tt.wantRetryDelayMs, ).Return(nil) } diff --git a/platform/extension/messagequeue/subscription_config.go b/platform/extension/messagequeue/subscription_config.go index b23c884cb..f6ed66de2 100644 --- a/platform/extension/messagequeue/subscription_config.go +++ b/platform/extension/messagequeue/subscription_config.go @@ -66,13 +66,16 @@ type RetryConfig struct { // After this many attempts, the message is moved to DLQ (if enabled). MaxAttempts int - // InitialBackoffMs is the initial backoff duration for retries (in milliseconds). + // InitialBackoffMs is the delay after the first failed attempt (in milliseconds). + // A non-positive value disables retry delay. InitialBackoffMs int64 - // MaxBackoffMs is the maximum backoff duration (in milliseconds). + // MaxBackoffMs is the maximum retry delay (in milliseconds). + // A non-positive value leaves the delay uncapped. MaxBackoffMs int64 - // BackoffMultiplier is the multiplier for exponential backoff. + // BackoffMultiplier scales the delay after each failed attempt. + // Values less than or equal to one produce a constant delay. BackoffMultiplier float64 } diff --git a/test/integration/extension/messagequeue/mysql/queue_test.go b/test/integration/extension/messagequeue/mysql/queue_test.go index 1fb3f21e5..d238d93bb 100644 --- a/test/integration/extension/messagequeue/mysql/queue_test.go +++ b/test/integration/extension/messagequeue/mysql/queue_test.go @@ -650,6 +650,41 @@ func (s *SQLQueueIntegrationSuite) TestVisibilityTimeoutAndRetry() { t.Logf("Successfully tested ExtendVisibilityTimeout and visibility timeout retry") } +func (s *SQLQueueIntegrationSuite) TestNackBackoff() { + t := s.T() + + signalCh := make(chan queueMySQL.HookSignal, 100) + q, err := queueMySQL.NewQueue(queueMySQL.Params{ + DB: s.db, + Logger: zaptest.NewLogger(t), + MetricsScope: tally.NoopScope, + OnSignal: signalCh, + }) + require.NoError(t, err) + defer q.Close() + + const retryDelayMs = 500 + subConfig := testSubConfig("worker-1", "nack-backoff-consumer") + subConfig.PollIntervalMs = 50 + subConfig.Retry.InitialBackoffMs = retryDelayMs + subConfig.Retry.MaxBackoffMs = 2 * retryDelayMs + subConfig.Retry.BackoffMultiplier = 2 + + deliveryChan, err := q.Subscriber().Subscribe(s.ctx, "nack_backoff_topic", subConfig) + require.NoError(t, err) + require.NoError(t, q.Publisher().Publish(s.ctx, "nack_backoff_topic", + entityqueue.NewMessage("retry-msg", []byte("test"), "retry-partition", nil))) + + firstDelivery := receive(t, deliveryChan) + assert.Equal(t, 1, firstDelivery.Attempt()) + require.NoError(t, firstDelivery.Nack(s.ctx, failure.New("retry later"))) + + assertNoDelivery(t, deliveryChan, signalCh, queueMySQL.SignalDeliveryCheck, 3) + retryDelivery := receive(t, deliveryChan) + assert.Equal(t, 2, retryDelivery.Attempt()) + require.NoError(t, retryDelivery.Ack(s.ctx)) +} + func (s *SQLQueueIntegrationSuite) TestIdempotentPublish() { t := s.T() @@ -1132,9 +1167,6 @@ func (s *SQLQueueIntegrationSuite) TestDeadLetterQueue() { t.Logf("Published poison message, will nack repeatedly") - // Receive and nack the message MaxAttempts times. - // Each iteration: receive the message, nack with 0 delay, then wait for - // the visibility timeout to expire so the message becomes deliverable again. // Each nack carries why it failed; the last one is the reason recorded // against the dead letter. for attempt := 1; attempt <= subConfig.Retry.MaxAttempts; attempt++ { @@ -1143,7 +1175,6 @@ func (s *SQLQueueIntegrationSuite) TestDeadLetterQueue() { assert.Equal(t, attempt, delivery.Attempt()) assert.Equal(t, "poison-msg", delivery.Message().ID) - // Nack without delay to retry immediately nackFailure := failure.New( fmt.Sprintf("processing failed on attempt %d", attempt), failure.Subject{Type: "widget", ID: "widget-7"}, @@ -2793,7 +2824,7 @@ func (s *SQLQueueIntegrationSuite) TestCrashAfterRetryLimitDoesNotLoseMessages() require.NoError(t, deliveries["msg-A"].Ack(s.ctx)) t.Logf("Acked msg-A") - // Nack B — immediately visible again for redelivery + // Nack B — eligible for redelivery after its retry backoff. require.NoError(t, deliveries["msg-B"].Nack(s.ctx, failure.New("msg-B failed"))) t.Logf("Nacked msg-B, waiting for retry-limit to trigger auto-DLQ") @@ -2809,9 +2840,6 @@ func (s *SQLQueueIntegrationSuite) TestCrashAfterRetryLimitDoesNotLoseMessages() // Give the poll loop time to process the nack and auto-DLQ msg-B // We can't use event-driven wait here because auto-DLQ happens inside pollAndDeliver // without delivering to the channel. A brief pause lets the poll loop run. - // The poll interval is 100ms and nack delay is 100ms, so 1s is generous. - // Actually, we CAN just crash and let worker-2 recover everything. - // Simulate crash q1.Close() t.Logf("Worker-1 crashed (queue closed)") From e4afe5674bfc4e7485a4d47dbd44dca82f521881 Mon Sep 17 00:00:00 2001 From: mnoah1 Date: Mon, 31 Aug 2026 16:33:36 +0000 Subject: [PATCH 2/5] test(messagequeue): make auto-DLQ recovery deterministic --- .../messagequeue/mysql/queue_test.go | 31 +++++++++++-------- 1 file changed, 18 insertions(+), 13 deletions(-) diff --git a/test/integration/extension/messagequeue/mysql/queue_test.go b/test/integration/extension/messagequeue/mysql/queue_test.go index d238d93bb..a87503019 100644 --- a/test/integration/extension/messagequeue/mysql/queue_test.go +++ b/test/integration/extension/messagequeue/mysql/queue_test.go @@ -2824,23 +2824,28 @@ func (s *SQLQueueIntegrationSuite) TestCrashAfterRetryLimitDoesNotLoseMessages() require.NoError(t, deliveries["msg-A"].Ack(s.ctx)) t.Logf("Acked msg-A") - // Nack B — eligible for redelivery after its retry backoff. + // Keep C in flight while B exhausts its retry budget. + heldVisibilityMs := subConfig.LeaseDurationMs + subConfig.VisibilityTimeoutMs + require.NoError(t, deliveries["msg-C"].ExtendVisibilityTimeout(s.ctx, heldVisibilityMs)) + require.NoError(t, deliveries["msg-B"].Nack(s.ctx, failure.New("msg-B failed"))) - t.Logf("Nacked msg-B, waiting for retry-limit to trigger auto-DLQ") + retryDelivery := receive(t, deliveryChan1) + require.Equal(t, "msg-B", retryDelivery.Message().ID) + require.Equal(t, 2, retryDelivery.Attempt()) - // Do NOT ack msg-C — simulating in-flight at crash time. + dlqTopic := topic + subConfig.DLQ.TopicSuffix + dlqConfig := testSubConfig("worker-1", "crash-retry-dlq-cg") + dlqDeliveryChan, err := q1.Subscriber().Subscribe(s.ctx, dlqTopic, dlqConfig) + require.NoError(t, err) - // Wait for msg-B to be redelivered and auto-DLQ'd by the poll loop. - // The poll loop picks up the nacked msg-B, sees retry_count >= MaxAttempts, moves it to DLQ. - // We just need to wait long enough for that to happen before crashing. - // A short sleep is acceptable here as we're waiting for the subscriber's - // internal processing, not for a test condition. But let's use receive - // to see if B comes back (it shouldn't, since auto-DLQ handles it internally). + const expireVisibilityMs = int64(1) + require.NoError(t, retryDelivery.ExtendVisibilityTimeout(s.ctx, expireVisibilityMs)) + dlqDelivery := receive(t, dlqDeliveryChan) + require.Equal(t, "msg-B", dlqDelivery.Message().ID) + require.NoError(t, dlqDelivery.Ack(s.ctx)) - // Give the poll loop time to process the nack and auto-DLQ msg-B - // We can't use event-driven wait here because auto-DLQ happens inside pollAndDeliver - // without delivering to the channel. A brief pause lets the poll loop run. - // Simulate crash + // Crash with C still unacked, but make it immediately recoverable once the lease expires. + require.NoError(t, deliveries["msg-C"].ExtendVisibilityTimeout(s.ctx, expireVisibilityMs)) q1.Close() t.Logf("Worker-1 crashed (queue closed)") From 03b831189fb6b758311004c483f2a1be17828d9d Mon Sep 17 00:00:00 2001 From: mnoah1 Date: Mon, 31 Aug 2026 19:36:11 +0000 Subject: [PATCH 3/5] fix(messagequeue): bound retry backoff delay --- .../mysql/delivery_state_store.go | 5 +++-- .../mysql/delivery_state_store_test.go | 9 +++----- .../messagequeue/mysql/subscriber.go | 21 ++++++++++++------- .../messagequeue/mysql/subscriber_test.go | 12 +++++++---- .../messagequeue/subscription_config.go | 3 ++- 5 files changed, 30 insertions(+), 20 deletions(-) diff --git a/platform/extension/messagequeue/mysql/delivery_state_store.go b/platform/extension/messagequeue/mysql/delivery_state_store.go index c149bf9b8..adf22dbac 100644 --- a/platform/extension/messagequeue/mysql/delivery_state_store.go +++ b/platform/extension/messagequeue/mysql/delivery_state_store.go @@ -158,8 +158,9 @@ func (s *sqldeliveryStateStore) MarkNacked(ctx context.Context, consumerGroup, t nowMs := time.Now().UnixMilli() invisibleUntil := nowMs if delayMs > math.MaxInt64-nowMs { - invisibleUntil = math.MaxInt64 - } else if delayMs > 0 { + return fmt.Errorf("mark nacked topic=%s partition=%s offset=%d: retry delay %d overflows visibility timestamp", topic, partitionKey, offset, delayMs) + } + if delayMs > 0 { invisibleUntil += delayMs } diff --git a/platform/extension/messagequeue/mysql/delivery_state_store_test.go b/platform/extension/messagequeue/mysql/delivery_state_store_test.go index cd8adfe37..c94fedd4c 100644 --- a/platform/extension/messagequeue/mysql/delivery_state_store_test.go +++ b/platform/extension/messagequeue/mysql/delivery_state_store_test.go @@ -235,15 +235,12 @@ func TestDeliveryStateStore_MarkNacked(t *testing.T) { } } -func TestDeliveryStateStore_MarkNackedSaturatesTimestamp(t *testing.T) { +func TestDeliveryStateStore_MarkNackedRejectsTimestampOverflow(t *testing.T) { store, db, mock := newTestDeliveryStateStoreWithMock(t) defer db.Close() - mock.ExpectExec("INSERT INTO queue_delivery_state"). - WithArgs("group-1", "orders", "part-1", int64(5), int64(math.MaxInt64)). - WillReturnResult(sqlmock.NewResult(1, 1)) - - require.NoError(t, store.MarkNacked(context.Background(), "group-1", "orders", "part-1", 5, math.MaxInt64)) + err := store.MarkNacked(context.Background(), "group-1", "orders", "part-1", 5, math.MaxInt64) + require.ErrorContains(t, err, "overflows visibility timestamp") assert.NoError(t, mock.ExpectationsWereMet()) } diff --git a/platform/extension/messagequeue/mysql/subscriber.go b/platform/extension/messagequeue/mysql/subscriber.go index a665d49d9..63c1c23c7 100644 --- a/platform/extension/messagequeue/mysql/subscriber.go +++ b/platform/extension/messagequeue/mysql/subscriber.go @@ -75,6 +75,11 @@ const ( // discovered partitions, so nothing would ever steal (and thereby // refresh or remove) a stale lease on a partition with no messages. leasePurgeAfterLeaseDurations = 10 + + // maxRetryBackoffMs bounds how long one failed message can pin its + // partition's contiguous ack watermark. Callers may choose a lower cap; + // this ceiling also applies when MaxBackoffMs is unset. + maxRetryBackoffMs = int64(time.Minute / time.Millisecond) ) // gcTickInterval is the number of poll ticks between garbage collection runs. @@ -1515,8 +1520,13 @@ func retryBackoffMs(retry extqueue.RetryConfig, attempt int) int64 { if backoffMs <= 0 { return 0 } - if retry.MaxBackoffMs > 0 && backoffMs >= retry.MaxBackoffMs { - return retry.MaxBackoffMs + + maxBackoffMs := retry.MaxBackoffMs + if maxBackoffMs <= 0 || maxBackoffMs > maxRetryBackoffMs { + maxBackoffMs = maxRetryBackoffMs + } + if backoffMs >= maxBackoffMs { + return maxBackoffMs } multiplier := retry.BackoffMultiplier @@ -1525,11 +1535,8 @@ func retryBackoffMs(retry extqueue.RetryConfig, attempt int) int64 { } backoff := float64(backoffMs) * math.Pow(multiplier, float64(attempt-1)) - if retry.MaxBackoffMs > 0 && backoff >= float64(retry.MaxBackoffMs) { - return retry.MaxBackoffMs - } - if backoff >= float64(math.MaxInt64) { - return math.MaxInt64 + if backoff >= float64(maxBackoffMs) { + return maxBackoffMs } return int64(backoff) } diff --git a/platform/extension/messagequeue/mysql/subscriber_test.go b/platform/extension/messagequeue/mysql/subscriber_test.go index 307cdc1fa..0f7f83990 100644 --- a/platform/extension/messagequeue/mysql/subscriber_test.go +++ b/platform/extension/messagequeue/mysql/subscriber_test.go @@ -18,7 +18,6 @@ import ( "context" "errors" "fmt" - "math" "testing" "time" @@ -482,9 +481,14 @@ func TestSQLDelivery_NackDeadLettersWhenBudgetSpent(t *testing.T) { wantRetryDelayMs: 2, }, { - name: "uncapped overflow saturates", attempt: 2, - retry: extqueue.RetryConfig{MaxAttempts: 3, InitialBackoffMs: math.MaxInt64, BackoffMultiplier: 2}, - wantRetryDelayMs: math.MaxInt64, + name: "uncapped backoff uses safety ceiling", attempt: 3, + retry: extqueue.RetryConfig{MaxAttempts: 4, InitialBackoffMs: 1000, BackoffMultiplier: 10}, + wantRetryDelayMs: maxRetryBackoffMs, + }, + { + name: "configured cap cannot exceed safety ceiling", attempt: 3, + retry: extqueue.RetryConfig{MaxAttempts: 4, InitialBackoffMs: 1000, MaxBackoffMs: 120000, BackoffMultiplier: 10}, + wantRetryDelayMs: maxRetryBackoffMs, }, { name: "unset initial delay retries immediately", attempt: 1, diff --git a/platform/extension/messagequeue/subscription_config.go b/platform/extension/messagequeue/subscription_config.go index f6ed66de2..766a65e68 100644 --- a/platform/extension/messagequeue/subscription_config.go +++ b/platform/extension/messagequeue/subscription_config.go @@ -71,7 +71,8 @@ type RetryConfig struct { InitialBackoffMs int64 // MaxBackoffMs is the maximum retry delay (in milliseconds). - // A non-positive value leaves the delay uncapped. + // A non-positive value leaves the delay without a caller-provided cap; + // implementations may still impose a safety ceiling. MaxBackoffMs int64 // BackoffMultiplier scales the delay after each failed attempt. From 4931f0508e2153587c213be20407b0928b19e9c4 Mon Sep 17 00:00:00 2001 From: mnoah1 Date: Mon, 31 Aug 2026 19:51:42 +0000 Subject: [PATCH 4/5] fix(messagequeue): validate retry configuration --- .../mysql/delivery_state_store.go | 11 ++----- .../mysql/delivery_state_store_test.go | 25 +++++++++++---- .../messagequeue/mysql/subscriber.go | 28 ++++++++++++++++ .../messagequeue/mysql/subscriber_test.go | 32 +++++++++++++++++++ 4 files changed, 81 insertions(+), 15 deletions(-) diff --git a/platform/extension/messagequeue/mysql/delivery_state_store.go b/platform/extension/messagequeue/mysql/delivery_state_store.go index adf22dbac..2794ce91e 100644 --- a/platform/extension/messagequeue/mysql/delivery_state_store.go +++ b/platform/extension/messagequeue/mysql/delivery_state_store.go @@ -18,7 +18,6 @@ import ( "context" "database/sql" "fmt" - "math" "time" "github.com/uber-go/tally" @@ -155,14 +154,10 @@ func (s *sqldeliveryStateStore) MarkNacked(ctx context.Context, consumerGroup, t metrics.NewTag("consumer_group", consumerGroup)) defer func() { op.Complete(retErr) }() - nowMs := time.Now().UnixMilli() - invisibleUntil := nowMs - if delayMs > math.MaxInt64-nowMs { - return fmt.Errorf("mark nacked topic=%s partition=%s offset=%d: retry delay %d overflows visibility timestamp", topic, partitionKey, offset, delayMs) - } - if delayMs > 0 { - invisibleUntil += delayMs + if delayMs < 0 || delayMs > maxRetryBackoffMs { + return fmt.Errorf("mark nacked topic=%s partition=%s offset=%d: retry delay %d is outside [0, %d]", topic, partitionKey, offset, delayMs, maxRetryBackoffMs) } + invisibleUntil := time.Now().UnixMilli() + delayMs _, err := s.db.ExecContext(ctx, fmt.Sprintf(` INSERT INTO %s (consumer_group, topic, partition_key, message_offset, acked, invisible_until, retry_count) diff --git a/platform/extension/messagequeue/mysql/delivery_state_store_test.go b/platform/extension/messagequeue/mysql/delivery_state_store_test.go index c94fedd4c..919294821 100644 --- a/platform/extension/messagequeue/mysql/delivery_state_store_test.go +++ b/platform/extension/messagequeue/mysql/delivery_state_store_test.go @@ -18,7 +18,6 @@ import ( "context" "database/sql" "database/sql/driver" - "math" "testing" "time" @@ -235,13 +234,25 @@ func TestDeliveryStateStore_MarkNacked(t *testing.T) { } } -func TestDeliveryStateStore_MarkNackedRejectsTimestampOverflow(t *testing.T) { - store, db, mock := newTestDeliveryStateStoreWithMock(t) - defer db.Close() +func TestDeliveryStateStore_MarkNackedRejectsInvalidDelay(t *testing.T) { + tests := []struct { + name string + delayMs int64 + }{ + {name: "negative", delayMs: -1}, + {name: "above safety ceiling", delayMs: maxRetryBackoffMs + 1}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + store, db, mock := newTestDeliveryStateStoreWithMock(t) + defer db.Close() - err := store.MarkNacked(context.Background(), "group-1", "orders", "part-1", 5, math.MaxInt64) - require.ErrorContains(t, err, "overflows visibility timestamp") - assert.NoError(t, mock.ExpectationsWereMet()) + err := store.MarkNacked(context.Background(), "group-1", "orders", "part-1", 5, tt.delayMs) + require.ErrorContains(t, err, "is outside") + assert.NoError(t, mock.ExpectationsWereMet()) + }) + } } func TestDeliveryStateStore_MarkPostponed(t *testing.T) { diff --git a/platform/extension/messagequeue/mysql/subscriber.go b/platform/extension/messagequeue/mysql/subscriber.go index 63c1c23c7..5dc7311b9 100644 --- a/platform/extension/messagequeue/mysql/subscriber.go +++ b/platform/extension/messagequeue/mysql/subscriber.go @@ -502,6 +502,9 @@ func (s *subscriber) Subscribe(ctx context.Context, topic string, config extqueu if closed { return nil, ErrSubscriberClosed } + if err := validateRetryConfig(config.Retry); err != nil { + return nil, fmt.Errorf("subscribe topic %q: invalid retry config: %w", topic, err) + } // Create subscription key (topic + consumer group must be unique) subKey := topic + ":" + config.ConsumerGroup @@ -1540,3 +1543,28 @@ func retryBackoffMs(retry extqueue.RetryConfig, attempt int) int64 { } return int64(backoff) } + +func validateRetryConfig(retry extqueue.RetryConfig) error { + if retry.MaxAttempts < 0 { + return fmt.Errorf("retry MaxAttempts must be non-negative, got %d", retry.MaxAttempts) + } + if retry.InitialBackoffMs < 0 { + return fmt.Errorf("retry InitialBackoffMs must be non-negative, got %d", retry.InitialBackoffMs) + } + if retry.MaxBackoffMs < 0 { + return fmt.Errorf("retry MaxBackoffMs must be non-negative, got %d", retry.MaxBackoffMs) + } + if retry.InitialBackoffMs > maxRetryBackoffMs { + return fmt.Errorf("retry InitialBackoffMs must not exceed %d, got %d", maxRetryBackoffMs, retry.InitialBackoffMs) + } + if retry.MaxBackoffMs > maxRetryBackoffMs { + return fmt.Errorf("retry MaxBackoffMs must not exceed %d, got %d", maxRetryBackoffMs, retry.MaxBackoffMs) + } + if retry.MaxBackoffMs > 0 && retry.InitialBackoffMs > retry.MaxBackoffMs { + return fmt.Errorf("retry InitialBackoffMs (%d) must not exceed MaxBackoffMs (%d)", retry.InitialBackoffMs, retry.MaxBackoffMs) + } + if retry.BackoffMultiplier < 0 || math.IsNaN(retry.BackoffMultiplier) || math.IsInf(retry.BackoffMultiplier, 0) { + return fmt.Errorf("retry BackoffMultiplier must be finite and non-negative, got %v", retry.BackoffMultiplier) + } + return nil +} diff --git a/platform/extension/messagequeue/mysql/subscriber_test.go b/platform/extension/messagequeue/mysql/subscriber_test.go index 0f7f83990..8167e615e 100644 --- a/platform/extension/messagequeue/mysql/subscriber_test.go +++ b/platform/extension/messagequeue/mysql/subscriber_test.go @@ -18,6 +18,7 @@ import ( "context" "errors" "fmt" + "math" "testing" "time" @@ -146,6 +147,37 @@ func TestSubscriber_Subscribe(t *testing.T) { } } +func TestSubscriber_SubscribeRejectsInvalidRetryConfig(t *testing.T) { + tests := []struct { + name string + retry extqueue.RetryConfig + wantErr string + }{ + {name: "negative max attempts", retry: extqueue.RetryConfig{MaxAttempts: -1}, wantErr: "MaxAttempts"}, + {name: "negative initial backoff", retry: extqueue.RetryConfig{InitialBackoffMs: -1}, wantErr: "InitialBackoffMs"}, + {name: "negative max backoff", retry: extqueue.RetryConfig{MaxBackoffMs: -1}, wantErr: "MaxBackoffMs"}, + {name: "initial backoff above safety ceiling", retry: extqueue.RetryConfig{InitialBackoffMs: maxRetryBackoffMs + 1}, wantErr: "InitialBackoffMs"}, + {name: "max backoff above safety ceiling", retry: extqueue.RetryConfig{MaxBackoffMs: maxRetryBackoffMs + 1}, wantErr: "MaxBackoffMs"}, + {name: "initial backoff above configured maximum", retry: extqueue.RetryConfig{InitialBackoffMs: 2000, MaxBackoffMs: 1000}, wantErr: "must not exceed MaxBackoffMs"}, + {name: "negative multiplier", retry: extqueue.RetryConfig{BackoffMultiplier: -1}, wantErr: "BackoffMultiplier"}, + {name: "NaN multiplier", retry: extqueue.RetryConfig{BackoffMultiplier: math.NaN()}, wantErr: "BackoffMultiplier"}, + {name: "infinite multiplier", retry: extqueue.RetryConfig{BackoffMultiplier: math.Inf(1)}, wantErr: "BackoffMultiplier"}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + sub := setupSubscriberTest(t, NewMockmessageStore(ctrl), NewMockoffsetStore(ctrl), NewMockpartitionLeaseStore(ctrl)) + t.Cleanup(func() { require.NoError(t, sub.Close()) }) + + ch, err := sub.Subscribe(context.Background(), "test_topic", extqueue.SubscriptionConfig{Retry: tt.retry}) + require.Nil(t, ch) + require.ErrorContains(t, err, "invalid retry config") + assert.ErrorContains(t, err, tt.wantErr) + }) + } +} + func TestSubscriber_SubscribeContextCancellation(t *testing.T) { ctrl := gomock.NewController(t) defer ctrl.Finish() From 0fe9e48c1a53cedc025b4dddb102a440276cc669 Mon Sep 17 00:00:00 2001 From: mnoah1 Date: Mon, 31 Aug 2026 20:09:36 +0000 Subject: [PATCH 5/5] feat(messagequeue): classify invalid subscription config --- platform/extension/messagequeue/mysql/errors.go | 3 +++ platform/extension/messagequeue/mysql/subscriber.go | 2 +- platform/extension/messagequeue/mysql/subscriber_test.go | 2 +- 3 files changed, 5 insertions(+), 2 deletions(-) diff --git a/platform/extension/messagequeue/mysql/errors.go b/platform/extension/messagequeue/mysql/errors.go index 4dbfba96b..53ff882b2 100644 --- a/platform/extension/messagequeue/mysql/errors.go +++ b/platform/extension/messagequeue/mysql/errors.go @@ -28,6 +28,9 @@ var ErrPublisherClosed = errors.New("publisher is closed") // ErrSubscriberClosed is returned when attempting to subscribe after the subscriber has been closed. var ErrSubscriberClosed = errors.New("subscriber is closed") +// ErrInvalidConfig is returned when subscription configuration is invalid. +var ErrInvalidConfig = errors.New("invalid subscription config") + // ErrAlreadyAcknowledged is returned when attempting to ack/nack a delivery that was already processed type ErrAlreadyAcknowledged struct { DeliveryID string diff --git a/platform/extension/messagequeue/mysql/subscriber.go b/platform/extension/messagequeue/mysql/subscriber.go index 5dc7311b9..f96ce8af4 100644 --- a/platform/extension/messagequeue/mysql/subscriber.go +++ b/platform/extension/messagequeue/mysql/subscriber.go @@ -503,7 +503,7 @@ func (s *subscriber) Subscribe(ctx context.Context, topic string, config extqueu return nil, ErrSubscriberClosed } if err := validateRetryConfig(config.Retry); err != nil { - return nil, fmt.Errorf("subscribe topic %q: invalid retry config: %w", topic, err) + return nil, fmt.Errorf("subscribe topic %q: %w: %v", topic, ErrInvalidConfig, err) } // Create subscription key (topic + consumer group must be unique) diff --git a/platform/extension/messagequeue/mysql/subscriber_test.go b/platform/extension/messagequeue/mysql/subscriber_test.go index 8167e615e..086147981 100644 --- a/platform/extension/messagequeue/mysql/subscriber_test.go +++ b/platform/extension/messagequeue/mysql/subscriber_test.go @@ -172,7 +172,7 @@ func TestSubscriber_SubscribeRejectsInvalidRetryConfig(t *testing.T) { ch, err := sub.Subscribe(context.Background(), "test_topic", extqueue.SubscriptionConfig{Retry: tt.retry}) require.Nil(t, ch) - require.ErrorContains(t, err, "invalid retry config") + require.ErrorIs(t, err, ErrInvalidConfig) assert.ErrorContains(t, err, tt.wantErr) }) }