refactor(stovepipe): address process-DLQ review feedback

Review feedback on the process DLQ PR: the DLQ subscription overrides
(DLQ.Enabled=false, MaxAttempts=1000) were duplicated across wiring files,
and two doc comments in the DLQ reconciler claimed behavior the code does
not have (release-first ordering prevents double release; decode errors
are non-retryable under AlwaysRetryableProcessor).

- platform/extension/messagequeue/subscription_config.go: add
  DLQSubscriptionConfig helper owning the DLQ-consumer subscription policy
- platform/extension/messagequeue/subscription_config_test.go: test the
  helper's contract
- service/stovepipe/server/main.go: use the helper
- service/submitqueue/orchestrator/server/main.go: use the helper
- stovepipe/controller/dlq/dlq.go: correct failRequest comment to state
  the real crash trade-off (double release vs leaked slot)
- stovepipe/controller/dlq/request.go: correct decode-error comment —
  retried deliberately to heal deployment skew, bounded by MaxAttempts

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: gsturges

platform/extension/messagequeue/subscription_config.go

@@ -78,6 +78,23 @@ 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.
+func DLQSubscriptionConfig(subscriberName, consumerGroup string) SubscriptionConfig {
+	config := DefaultSubscriptionConfig(subscriberName, consumerGroup)
+	config.DLQ.Enabled = false
+	config.Retry.MaxAttempts = 1000
+	return config
+}
+
 // DefaultSubscriptionConfig returns a SubscriptionConfig with sensible defaults.
 func DefaultSubscriptionConfig(subscriberName, consumerGroup string) SubscriptionConfig {
 	return SubscriptionConfig{

platform/extension/messagequeue/subscription_config_test.go

@@ -73,6 +73,18 @@ func TestSubscriptionConfig_CustomValues(t *testing.T) {
 	assert.Equal(t, "_dead", config.DLQ.TopicSuffix)
 }
 
+func TestDLQSubscriptionConfig(t *testing.T) {
+	config := DLQSubscriptionConfig("worker-1", "consumer-1-dlq")
+
+	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)
+}
+
 func TestSubscriptionConfig_DifferentConsumerGroups(t *testing.T) {
 	// Test that different consumer groups get independent configs
 	tests := []struct {

service/stovepipe/server/main.go

@@ -341,14 +341,8 @@ func run() error {
 // newTopicRegistry builds the TopicRegistry for Stovepipe's internal pipeline queues. ingest
 // publishes to process; process publishes admitted requests to the publish-only build topic.
 // The process_dlq topic is the dead-letter destination the queue backend routes to (per
-// DefaultSubscriptionConfig's DLQ.TopicSuffix) when the process controller exhausts retries;
-// DLQ.Enabled is false on its own subscription so a reconciliation failure retries in place
-// rather than cascading to a further DLQ.
+// DefaultSubscriptionConfig's DLQ.TopicSuffix) when the process controller exhausts retries.
 func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRegistry, error) {
-	dlqSub := extqueue.DefaultSubscriptionConfig(subscriberName, "stovepipe-process-dlq")
-	dlqSub.DLQ.Enabled = false
-	dlqSub.Retry.MaxAttempts = 1000
-
 	return consumer.NewTopicRegistry([]consumer.TopicConfig{
 		{
 			Key:   stovepipemq.TopicKeyProcess,
@@ -367,7 +361,7 @@ func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRe
 			Key:          dlq.TopicKey(stovepipemq.TopicKeyProcess),
 			Name:         "process_dlq",
 			Queue:        q,
-			Subscription: dlqSub,
+			Subscription: extqueue.DLQSubscriptionConfig(subscriberName, "stovepipe-process-dlq"),
 		},
 	})
 }

service/submitqueue/orchestrator/server/main.go

@@ -403,27 +403,15 @@ func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRe
 				subscriberName, t.groupSuffix,
 			),
 		})
-		// DLQ subscription for the same primary stage. DLQ is disabled here
-		// to avoid a "_dlq_dlq" cascade: if DLQ reconciliation itself fails,
-		// the consumer retries forever and the failure is surfaced via logs
-		// and metrics rather than being moved to a second-level dead-letter
-		// topic that nobody consumes.
-		//
-		// MaxAttempts is bumped to a very high value so the per-message
-		// retry budget effectively never runs out — this pairs with the
-		// AlwaysRetryableProcessor wired into the DLQ consumer to guarantee
-		// reconciliation eventually converges instead of being silently
-		// dropped after the default retry count.
-		dlqSub := extqueue.DefaultSubscriptionConfig(
-			subscriberName, t.groupSuffix+"-dlq",
-		)
-		dlqSub.DLQ.Enabled = false
-		dlqSub.Retry.MaxAttempts = 1000
+		// DLQ subscription for the same primary stage. DLQSubscriptionConfig
+		// disables the subscription's own DLQ (no "_dlq_dlq" cascade) and sets
+		// an effectively unlimited retry budget to pair with the
+		// AlwaysRetryableProcessor wired into the DLQ consumer.
 		configs = append(configs, consumer.TopicConfig{
 			Key:          dlq.TopicKey(t.key),
 			Name:         t.name + "_dlq",
 			Queue:        q,
-			Subscription: dlqSub,
+			Subscription: extqueue.DLQSubscriptionConfig(subscriberName, t.groupSuffix+"-dlq"),
 		})
 	}
 

stovepipe/controller/dlq/dlq.go

@@ -27,8 +27,7 @@
 // name). The DLQ controller decodes that payload to recover the affected request, then
 // transitions it to RequestStateRecordedNotGreen — the conservative not-green verdict
 // for gating (see entity.RequestState) — with an idempotent optimistic-locking write so
-// concurrent
-// activity (a late successful pipeline transition) wins cleanly. If the request had
+// concurrent activity (a late successful pipeline transition) wins cleanly. If the request had
 // already been admitted (processing) and was holding a concurrency slot, the
 // reconciler also releases it by CAS-decrementing the queue's in_flight_count, per
 // doc/rfc/stovepipe/steps/process.md#in_flight_count-integrity.
@@ -61,8 +60,16 @@ func TopicKey(main consumer.TopicKey) consumer.TopicKey {
 // failRequest transitions request to RequestStateRecordedNotGreen if it is not already
 // in a terminal state. If the request had reached RequestStateProcessing — meaning process's
 // admit step already CAS-incremented the queue's in_flight_count for it — the queue's
-// slot is released first, so a crash between the two writes leaves the count still
-// bound to a non-terminal request rather than double-released.
+// slot is released first. Queue and Request are separate entities with no cross-entity
+// transaction, so the two writes cannot be atomic and the ordering picks which crash
+// failure mode we accept: a crash between the writes leaves the request non-terminal,
+// redelivery re-runs reconciliation, and releaseSlot (which tracks no per-request slot
+// ownership) decrements again — transiently over-admitting by one slot until the
+// under-count re-converges at releaseSlot's zero clamp. The reverse order would leak
+// the slot instead: redelivery skips terminal requests, permanently shrinking the
+// queue's capacity toward a wedge. Over-admission is the failure mode we prefer. See
+// doc/rfc/stovepipe/steps/process.md#in_flight_count-integrity for the broader
+// counter-drift story.
 func failRequest(ctx context.Context, store storage.Storage, logger *zap.SugaredLogger, requestID string) error {
 	request, err := store.GetRequestStore().Get(ctx, requestID)
 	if err != nil {
@@ -93,7 +100,7 @@ func failRequest(ctx context.Context, store storage.Storage, logger *zap.Sugared
 	updated.State = entity.RequestStateRecordedNotGreen
 	newVersion := request.Version + 1
 	if err := store.GetRequestStore().Update(ctx, updated, request.Version, newVersion); err != nil {
-		return fmt.Errorf("failed to update request %s state to failed: %w", requestID, err)
+		return fmt.Errorf("failed to update request %s state to recorded_not_green: %w", requestID, err)
 	}
 	logger.Infow("dlq reconcile: request forced terminal not-green",
 		"request_id", requestID,

stovepipe/controller/dlq/request.go

@@ -75,8 +75,15 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r
 	pr := &stovepipemq.ProcessRequest{}
 	if err := stovepipemq.Unmarshal(msg.Payload, pr); err != nil {
 		metrics.NamedCounter(c.metricsScope, _opName, "deserialize_errors", 1)
-		// Malformed DLQ payload is non-retryable: a re-delivery will decode the same
-		// bytes and fail the same way.
+		// Decoding the same bytes normally fails deterministically, but this error is
+		// still retried: the DLQ consumer's AlwaysRetryableProcessor (see Process doc)
+		// 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.
 		return fmt.Errorf("failed to decode dlq payload: %w", err)
 	}
 	if pr.Id == "" {