fix(queue): ISS-013 keep DLQ reconciliation retrying

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.

---

<sub>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</sub>

Author: sbalabanov-zz

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
 

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

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

platform/extension/messagequeue/mysql/subscriber.go

@@ -341,7 +341,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,
@@ -1188,8 +1188,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,
@@ -1610,6 +1609,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)

platform/extension/messagequeue/mysql/subscriber_test.go

@@ -594,9 +594,6 @@ 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},
 	}
 
 	for _, tt := range tests {
@@ -642,6 +639,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.

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
 }
 

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

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)

service/runway/server/BUILD.bazel

@@ -100,6 +100,7 @@ go_test(
     },
     deps = [
         "//api/base/mergestrategy/protopb:go_default_library",
+        "//api/runway/messagequeue:go_default_library",
         "//platform/base/failure:go_default_library",
         "//platform/base/messagequeue:go_default_library",
         "//platform/consumer:go_default_library",
@@ -108,6 +109,7 @@ go_test(
         "//platform/extension/messagequeue/mock:go_default_library",
         "//platform/git/exec: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",

service/runway/server/main.go

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