From 68fa422c209405d768fa3d6622d2f863d341d4a8 Mon Sep 17 00:00:00 2001 From: sergeyb Date: Fri, 4 Sep 2026 22:22:07 +0000 Subject: [PATCH] fix(queue): ISS-013 keep DLQ reconciliation retrying MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Summary: Intent: - Keep final-DLQ reconciliation durable during dependency outages longer than any finite attempt budget. - Preserve finite retry budgets for primary subscriptions. Changes: - Define MaxAttempts zero as unlimited in both direct Nack and visibility-expiry poll paths. - Configure the shared DLQ subscription for unlimited retries with second-level dead-lettering disabled. - Verify the orchestrator pipeline and Runway wiring inherit the shared behavior and update the operational docs. Reproduction: - A signal or storage dependency remains unavailable for more than 1000 DLQ reconciliation attempts. - Previously the finite cap was exhausted; because the reconciliation subscription had its own DLQ disabled, MySQL acknowledged the row and advanced past it, losing the reconciliation message. - The row now remains retryable until reconciliation succeeds or an operator removes it. --- 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/errs/README.md | 2 +- platform/extension/messagequeue/README.md | 4 +- .../extension/messagequeue/mysql/README.md | 2 +- .../messagequeue/mysql/subscriber.go | 9 +- .../messagequeue/mysql/subscriber_test.go | 95 ++++++++++++++++++- .../messagequeue/subscription_config.go | 17 ++-- .../messagequeue/subscription_config_test.go | 6 +- platform/pipeline/pipeline_test.go | 2 + service/runway/server/BUILD.bazel | 4 + service/runway/server/main.go | 2 +- service/runway/server/main_test.go | 74 +++++++++++++++ stovepipe/controller/dlq/request.go | 7 +- .../orchestrator/controller/dlq/README.md | 2 +- 13 files changed, 196 insertions(+), 30 deletions(-) create mode 100644 service/runway/server/main_test.go diff --git a/platform/errs/README.md b/platform/errs/README.md index b86c3b104..054346723 100644 --- a/platform/errs/README.md +++ b/platform/errs/README.md @@ -85,7 +85,7 @@ One operational consequence worth knowing before relying on any of this: **retry ### Choosing a processor - **Primary pipeline consumer** → `NewClassifierProcessor(...)`. Controllers' explicit `NewUserError` / `NewDependencyError` wraps must survive so user errors don't get retried, and unclassified backend errors must be inspected by the registered classifiers. -- **DLQ reconciliation consumer** → `AlwaysRetryableProcessor`. The DLQ is the last stop; any unprocessable message must come back for another attempt rather than silently drop. The DLQ subscription itself runs with a very high `Retry.MaxAttempts` and with its own DLQ disabled, so "always retryable + bounded-but-effectively-infinite attempts" is the convergence guarantee. +- **DLQ reconciliation consumer** → `AlwaysRetryableProcessor`. The DLQ is the last stop; any unprocessable message must come back for another attempt rather than silently drop. The DLQ subscription itself runs with unlimited attempts (`Retry.MaxAttempts = 0`) and with its own DLQ disabled, so every returned error remains retryable until reconciliation succeeds or an operator removes the message. ## Adding a Backend-Specific Classifier diff --git a/platform/extension/messagequeue/README.md b/platform/extension/messagequeue/README.md index ab3b30837..595f346cf 100644 --- a/platform/extension/messagequeue/README.md +++ b/platform/extension/messagequeue/README.md @@ -53,7 +53,7 @@ type Delivery interface { - **Reject** — poison pill, move to DLQ (or ack if DLQ disabled) - **ExtendVisibilityTimeout** — extend processing window for long-running work -**`Postpone` vs `Nack` vs `ExtendVisibilityTimeout`:** `Nack` is a failure — the message is immediately eligible again, the redelivery counts toward `Retry.MaxAttempts` and eventually trips the DLQ, and later offsets in the partition keep flowing past the nacked message (a failed message must not halt its partition). `Postpone` is a deliberate wait — the redelivery happens after the chosen delay, resets the failure streak (it restarts at attempt 1), and blocks the partition behind it until it redelivers, in order. `ExtendVisibilityTimeout` is neither: the delivery is still being processed and stays in flight. +**`Postpone` vs `Nack` vs `ExtendVisibilityTimeout`:** `Nack` is a failure — the message is immediately eligible again, the redelivery counts toward `Retry.MaxAttempts` and eventually trips the DLQ when the limit is finite, and later offsets in the partition keep flowing past the nacked message (a failed message must not halt its partition). `Postpone` is a deliberate wait — the redelivery happens after the chosen delay, resets the failure streak (it restarts at attempt 1), and blocks the partition behind it until it redelivers, in order. `ExtendVisibilityTimeout` is neither: the delivery is still being processed and stays in flight. ### SubscriptionConfig @@ -70,6 +70,8 @@ cfg.DLQ.Enabled = true See `subscription_config.go` for all fields and defaults. +`Retry.MaxAttempts` uses zero to mean unlimited attempts. `DLQSubscriptionConfig` selects this mode and disables a second-level DLQ so reconciliation messages remain retryable until they converge or an operator removes them. + ## Usage ```go diff --git a/platform/extension/messagequeue/mysql/README.md b/platform/extension/messagequeue/mysql/README.md index c9fa8c69a..6a78133d8 100644 --- a/platform/extension/messagequeue/mysql/README.md +++ b/platform/extension/messagequeue/mysql/README.md @@ -74,7 +74,7 @@ subConfig.DLQ.TopicSuffix = "_dlq" // DLQ topic suffix | `VisibilityTimeoutMs` | How long messages are invisible after fetch. Must exceed max processing time for `BatchSize=1` | | `LeaseRenewalIntervalMs` | How often to renew partition leases | | `LeaseDurationMs` | How long leases remain valid without renewal | -| `Retry.MaxAttempts` | Maximum processing attempts before DLQ | +| `Retry.MaxAttempts` | Maximum processing attempts before DLQ; zero retries indefinitely | | `DLQ.TopicSuffix` | Suffix appended to topic name for DLQ (e.g., `"orders"` → `"orders_dlq"`) | ## Package Layout diff --git a/platform/extension/messagequeue/mysql/subscriber.go b/platform/extension/messagequeue/mysql/subscriber.go index f96ce8af4..c27be43df 100644 --- a/platform/extension/messagequeue/mysql/subscriber.go +++ b/platform/extension/messagequeue/mysql/subscriber.go @@ -325,7 +325,7 @@ func (d *sqlDelivery) Nack(ctx context.Context, f failure.Failure) error { return &ErrAlreadyAcknowledged{DeliveryID: d.deliveryID} } - if d.retry.MaxAttempts > 0 && d.attempt >= d.retry.MaxAttempts { + if retryBudgetExhausted(d.retry.MaxAttempts, d.attempt) { d.subscriber.logger.Warnw("message exhausted retry budget, dead-lettering", "topic", d.topic, "partition_key", d.partitionKey, @@ -1120,8 +1120,7 @@ func (w *partitionWorker) pollAndDeliver(ctx context.Context) (retErr error) { return fmt.Errorf("mark delivered offset=%d: %w", row.Offset, err) } - // Check if message has exceeded retry limit - if retryCount >= cfg.Retry.MaxAttempts { + if retryBudgetExhausted(cfg.Retry.MaxAttempts, retryCount) { s.logger.Warnw("message exceeded retry limit", "topic", sub.topic, "consumer_group", cfg.ConsumerGroup, @@ -1544,6 +1543,10 @@ func retryBackoffMs(retry extqueue.RetryConfig, attempt int) int64 { return int64(backoff) } +func retryBudgetExhausted(maxAttempts, attempts int) bool { + return maxAttempts > 0 && attempts >= maxAttempts +} + func validateRetryConfig(retry extqueue.RetryConfig) error { if retry.MaxAttempts < 0 { return fmt.Errorf("retry MaxAttempts must be non-negative, got %d", retry.MaxAttempts) diff --git a/platform/extension/messagequeue/mysql/subscriber_test.go b/platform/extension/messagequeue/mysql/subscriber_test.go index 086147981..73d010958 100644 --- a/platform/extension/messagequeue/mysql/subscriber_test.go +++ b/platform/extension/messagequeue/mysql/subscriber_test.go @@ -536,9 +536,7 @@ func TestSQLDelivery_NackDeadLettersWhenBudgetSpent(t *testing.T) { 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}, + {name: "unlimited budget keeps retrying beyond former cap", attempt: 1001}, } for _, tt := range tests { @@ -583,6 +581,97 @@ func TestSQLDelivery_NackDeadLettersWhenBudgetSpent(t *testing.T) { } } +func TestPartitionWorker_PollRetryLimit(t *testing.T) { + tests := []struct { + name string + maxAttempts int + retryCount int + expectAck bool + expectDelivery bool + expectedAttempt int + }{ + { + name: "finite subscription acknowledges after visibility expiry exhausts retries", + maxAttempts: 3, + retryCount: 3, + expectAck: true, + }, + { + name: "unlimited subscription redelivers after visibility expiry", + maxAttempts: 0, + retryCount: 1001, + expectDelivery: true, + expectedAttempt: 1002, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + mockMessageStore := NewMockmessageStore(ctrl) + mockOffsetStore := NewMockoffsetStore(ctrl) + mockDeliveryState := NewMockdeliveryStateStore(ctrl) + + s := NewSubscriber( + zaptest.NewLogger(t).Sugar(), + tally.NoopScope, + mockMessageStore, + mockOffsetStore, + NewMockpartitionLeaseStore(ctrl), + newTestHeartbeatStore(ctrl), + mockDeliveryState, + ) + + cfg := testSubscriptionConfig() + cfg.Retry.MaxAttempts = tt.maxAttempts + cfg.DLQ.Enabled = false + deliveryCh := make(chan extqueue.Delivery, 1) + sub := &subscription{ + topic: "test_topic", + config: cfg, + deliveryCh: deliveryCh, + workers: make(map[string]*partitionWorker), + } + worker := &partitionWorker{ + partitionKey: "part-1", + sub: sub, + subscriber: s, + done: make(chan struct{}), + } + row := messageRow{ + ID: "msg-1", + Offset: 1, + PartitionKey: "part-1", + Payload: []byte("payload"), + PublishedAt: time.Now().UnixMilli(), + } + + mockOffsetStore.EXPECT().Initialize(gomock.Any(), "test_topic", "part-1", cfg.ConsumerGroup).Return(nil) + mockOffsetStore.EXPECT().GetAckedOffset(gomock.Any(), "test_topic", "part-1", cfg.ConsumerGroup).Return(int64(0), nil).Times(2) + mockMessageStore.EXPECT().FetchByOffset(gomock.Any(), "test_topic", "part-1", int64(0), cfg.BatchSize).Return([]messageRow{row}, nil) + mockDeliveryState.EXPECT().GetDeliveryState(gomock.Any(), cfg.ConsumerGroup, "test_topic", "part-1", int64(1)). + Return(DeliveryState{InvisibleUntil: time.Now().Add(-time.Second).UnixMilli(), RetryCount: tt.retryCount}, true, nil) + mockDeliveryState.EXPECT().MarkDelivered(gomock.Any(), cfg.ConsumerGroup, "test_topic", "part-1", int64(1), cfg.VisibilityTimeoutMs). + Return(tt.retryCount, nil) + if tt.expectAck { + mockDeliveryState.EXPECT().MarkAcked(gomock.Any(), cfg.ConsumerGroup, "test_topic", "part-1", int64(1)).Return(nil) + } + mockMessageStore.EXPECT().GetOffsetsAbove(gomock.Any(), "test_topic", "part-1", int64(0), watermarkAdvancementLimit).Return([]int64{1}, nil) + mockDeliveryState.EXPECT().AdvanceWatermark(gomock.Any(), cfg.ConsumerGroup, "test_topic", "part-1", int64(0), []int64{1}).Return(int64(0), nil) + + require.NoError(t, worker.pollAndDeliver(context.Background())) + + select { + case delivery := <-deliveryCh: + require.True(t, tt.expectDelivery) + assert.Equal(t, tt.expectedAttempt, delivery.Attempt()) + default: + assert.False(t, tt.expectDelivery) + } + }) + } +} + // A message arriving from its original topic has no failure to report, which is // how a DLQ consumer tells "nothing recorded" apart from a recorded failure // that named nothing. diff --git a/platform/extension/messagequeue/subscription_config.go b/platform/extension/messagequeue/subscription_config.go index 766a65e68..1522e6e71 100644 --- a/platform/extension/messagequeue/subscription_config.go +++ b/platform/extension/messagequeue/subscription_config.go @@ -64,6 +64,7 @@ type SubscriptionConfig struct { type RetryConfig struct { // MaxAttempts is the maximum number of processing attempts. // After this many attempts, the message is moved to DLQ (if enabled). + // Zero means unlimited attempts. MaxAttempts int // InitialBackoffMs is the delay after the first failed attempt (in milliseconds). @@ -90,20 +91,14 @@ type DLQConfig struct { TopicSuffix string } -// DLQSubscriptionConfig returns a SubscriptionConfig for consuming a dead-letter -// topic (DLQ reconciliation). It starts from DefaultSubscriptionConfig and applies -// the two overrides every DLQ consumer needs: -// -// - DLQ.Enabled is false, so a reconciliation failure retries in place instead of -// cascading to a second-level "_dlq_dlq" topic that nobody consumes. -// - Retry.MaxAttempts is a very high backstop so the per-message retry budget -// effectively never runs out. This pairs with errs.AlwaysRetryableProcessor -// wired into the DLQ consumer: reconciliation converges eventually instead of -// being silently dropped after the default retry count. +// DLQSubscriptionConfig returns a final-DLQ reconciliation subscription. +// It disables a second-level DLQ and sets MaxAttempts to zero (unlimited). +// Paired with errs.AlwaysRetryableProcessor, errors redeliver until the +// reconciliation converges or an operator removes the message. func DLQSubscriptionConfig(subscriberName, consumerGroup string) SubscriptionConfig { config := DefaultSubscriptionConfig(subscriberName, consumerGroup) config.DLQ.Enabled = false - config.Retry.MaxAttempts = 1000 + config.Retry.MaxAttempts = 0 return config } diff --git a/platform/extension/messagequeue/subscription_config_test.go b/platform/extension/messagequeue/subscription_config_test.go index efb63d60a..9c431a262 100644 --- a/platform/extension/messagequeue/subscription_config_test.go +++ b/platform/extension/messagequeue/subscription_config_test.go @@ -78,11 +78,9 @@ func TestDLQSubscriptionConfig(t *testing.T) { assert.Equal(t, "worker-1", config.SubscriberName) assert.Equal(t, "consumer-1-dlq", config.ConsumerGroup) - - // The DLQ consumer must not dead-letter its own failures (no "_dlq_dlq" - // cascade) and needs a far larger retry budget than a primary consumer. assert.False(t, config.DLQ.Enabled) - assert.Greater(t, config.Retry.MaxAttempts, DefaultSubscriptionConfig("worker-1", "consumer-1").Retry.MaxAttempts) + assert.Zero(t, config.Retry.MaxAttempts) + assert.Positive(t, DefaultSubscriptionConfig("worker-1", "consumer-1").Retry.MaxAttempts) } func TestSubscriptionConfig_DifferentConsumerGroups(t *testing.T) { diff --git a/platform/pipeline/pipeline_test.go b/platform/pipeline/pipeline_test.go index a0d79af70..f08b47645 100644 --- a/platform/pipeline/pipeline_test.go +++ b/platform/pipeline/pipeline_test.go @@ -415,6 +415,7 @@ func TestBuildTopicConfigs(t *testing.T) { assert.Equal(t, consumer.TopicKey("start"), configs[0].Key) assert.Equal(t, "start", configs[0].Name) assert.Equal(t, "orchestrator", configs[0].Subscription.ConsumerGroup) + assert.Positive(t, configs[0].Subscription.Retry.MaxAttempts) // Verify DLQ config derived from primary. assert.Equal(t, consumer.TopicKey("start_dlq"), configs[1].Key) @@ -425,6 +426,7 @@ func TestBuildTopicConfigs(t *testing.T) { expected := extqueue.DLQSubscriptionConfig("test-sub", "orchestrator-dlq") assert.Equal(t, expected.DLQ.Enabled, configs[1].Subscription.DLQ.Enabled) assert.Equal(t, expected.Retry.MaxAttempts, configs[1].Subscription.Retry.MaxAttempts) + assert.Zero(t, configs[1].Subscription.Retry.MaxAttempts) // Verify validate stage (primary + DLQ). assert.Equal(t, consumer.TopicKey("validate"), configs[2].Key) diff --git a/service/runway/server/BUILD.bazel b/service/runway/server/BUILD.bazel index 113cd6e23..125523804 100644 --- a/service/runway/server/BUILD.bazel +++ b/service/runway/server/BUILD.bazel @@ -79,6 +79,7 @@ go_test( srcs = [ "checkout_test.go", "config_test.go", + "main_test.go", ], # Checkout provisioning runs real git, so the test uses the same pinned # runtime the merger does rather than whatever git the host happens to have. @@ -97,7 +98,10 @@ go_test( }, deps = [ "//api/base/mergestrategy/protopb:go_default_library", + "//api/runway/messagequeue:go_default_library", + "//platform/consumer:go_default_library", "//platform/git/exectest:go_default_library", + "//runway/controller/dlq:go_default_library", "//runway/extension/merger/git:go_default_library", "@com_github_stretchr_testify//assert:go_default_library", "@com_github_stretchr_testify//require:go_default_library", diff --git a/service/runway/server/main.go b/service/runway/server/main.go index dc9985001..28319a3a1 100644 --- a/service/runway/server/main.go +++ b/service/runway/server/main.go @@ -643,7 +643,7 @@ func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRe // DLQ topics: the reconciler consumes these and republishes a FAILED // result to the corresponding signal topic. Names match the primary // topic name plus the "_dlq" suffix the subscriber uses when - // dead-lettering (see dlq.TopicKey / DefaultSubscriptionConfig). + // dead-lettering (see dlq.TopicKey / DLQSubscriptionConfig). { Key: dlq.TopicKey(runwaymq.TopicKeyMergeConflictCheck), Name: "merge-conflict-check_dlq", diff --git a/service/runway/server/main_test.go b/service/runway/server/main_test.go new file mode 100644 index 000000000..e209c32fd --- /dev/null +++ b/service/runway/server/main_test.go @@ -0,0 +1,74 @@ +// Copyright (c) 2026 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package main + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + runwaymq "github.com/uber/submitqueue/api/runway/messagequeue" + "github.com/uber/submitqueue/platform/consumer" + "github.com/uber/submitqueue/runway/controller/dlq" +) + +func TestNewTopicRegistry_RetryBudgets(t *testing.T) { + registry, err := newTopicRegistry(nil, "runway-test") + require.NoError(t, err) + + tests := []struct { + name string + topicKey consumer.TopicKey + consumerGroup string + unlimited bool + }{ + { + name: "merge conflict check primary remains finite", + topicKey: runwaymq.TopicKeyMergeConflictCheck, + consumerGroup: "runway-mergeconflictcheck", + }, + { + name: "merge conflict check dlq is unlimited", + topicKey: dlq.TopicKey(runwaymq.TopicKeyMergeConflictCheck), + consumerGroup: "runway-mergeconflictcheck-dlq", + unlimited: true, + }, + { + name: "merge primary remains finite", + topicKey: runwaymq.TopicKeyMerge, + consumerGroup: "runway-merge", + }, + { + name: "merge dlq is unlimited", + topicKey: dlq.TopicKey(runwaymq.TopicKeyMerge), + consumerGroup: "runway-merge-dlq", + unlimited: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + config, found := registry.SubscriptionConfig(tt.topicKey, tt.consumerGroup) + require.True(t, found) + if tt.unlimited { + assert.Zero(t, config.Retry.MaxAttempts) + assert.False(t, config.DLQ.Enabled) + return + } + assert.Positive(t, config.Retry.MaxAttempts) + assert.True(t, config.DLQ.Enabled) + }) + } +} diff --git a/stovepipe/controller/dlq/request.go b/stovepipe/controller/dlq/request.go index 89f4a63cd..9f9ff8bc5 100644 --- a/stovepipe/controller/dlq/request.go +++ b/stovepipe/controller/dlq/request.go @@ -78,10 +78,9 @@ func (c *requestController) Process(ctx context.Context, delivery consumer.Deliv // classifies every error as retryable. That is deliberate — the recoverable // cause is deployment skew, where a newer producer's payload shape reaches a // not-yet-upgraded consumer and decodes fine once the rollout completes. A - // genuinely malformed payload exhausts the DLQ subscription's MaxAttempts - // backstop and is dropped by the subscriber with a warning log; acking it here - // instead would skip reconciliation silently and leave the referenced request - // non-terminal. + // genuinely malformed payload remains available for operator inspection and + // removal; acking it here would skip reconciliation silently and leave the + // referenced request non-terminal. return fmt.Errorf("failed to decode dlq payload: %w", err) } if pr.Id == "" { diff --git a/submitqueue/orchestrator/controller/dlq/README.md b/submitqueue/orchestrator/controller/dlq/README.md index befa1bb9f..b605e39af 100644 --- a/submitqueue/orchestrator/controller/dlq/README.md +++ b/submitqueue/orchestrator/controller/dlq/README.md @@ -12,7 +12,7 @@ This package contains the controllers that drain each primary pipeline topic's ` ## Convergence guarantee -DLQ consumers are wired with `errs.AlwaysRetryableProcessor` and a very high `Retry.MaxAttempts` (currently 1000). Together with `DLQ.Enabled = false` on the DLQ subscription itself, this means any non-nil error returned from a DLQ controller — including a plain unclassified infra error — is forced retryable and redelivered rather than silently dropped. The combination is "always retryable + bounded-but-effectively-infinite attempts" and is the property the package relies on for convergence. +DLQ consumers are wired with `errs.AlwaysRetryableProcessor`, unlimited attempts (`Retry.MaxAttempts = 0`), and `DLQ.Enabled = false` on the DLQ subscription itself. Any non-nil error returned from a DLQ controller — including a plain unclassified infra error — is therefore forced retryable and redelivered rather than silently dropped. This is the convergence guarantee the package relies on. The recognised error condition is handled explicitly in `dlq.go`: