uber · sbalabanov · Sep 4, 2026
@@ -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
 

@@ -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

@@ -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

@@ -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)

@@ -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.

@@ -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
 }
 

@@ -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) {

@@ -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)

@@ -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",

@@ -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",

@@ -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)
+		})
+	}
+}
@@ -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 == "" {

@@ -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`: