feat(stovepipe): add DLQ reconciliation for the process stage

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/BUILD.bazel

@@ -15,6 +15,7 @@ go_library(
         "//platform/extension/messagequeue/mysql:go_default_library",
         "//service/stovepipe/server/mapper:go_default_library",
         "//stovepipe/controller:go_default_library",
+        "//stovepipe/controller/dlq:go_default_library",
         "//stovepipe/controller/process:go_default_library",
         "//stovepipe/core/messagequeue:go_default_library",
         "//stovepipe/extension/queueconfig/default:go_default_library",

service/stovepipe/server/main.go

@@ -37,6 +37,7 @@ import (
 	queueMySQL "github.com/uber/submitqueue/platform/extension/messagequeue/mysql"
 	"github.com/uber/submitqueue/service/stovepipe/server/mapper"
 	"github.com/uber/submitqueue/stovepipe/controller"
+	"github.com/uber/submitqueue/stovepipe/controller/dlq"
 	"github.com/uber/submitqueue/stovepipe/controller/process"
 	stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue"
 	queueconfigdefault "github.com/uber/submitqueue/stovepipe/extension/queueconfig/default"
@@ -208,13 +209,21 @@ func run() error {
 		return fmt.Errorf("failed to create topic registry: %w", err)
 	}
 
-	// Consumer running the process stage.
+	// Two consumers share the topic registry but apply different error classification
+	// policies. The primary consumer runs the standard classifier walk. The DLQ consumer
+	// uses AlwaysRetryableProcessor so every non-nil error from a DLQ controller is
+	// forced retryable — reconciliation must redeliver on any failure because the DLQ
+	// subscription is a final destination (DLQ.Enabled is false on it, so there is no
+	// further DLQ to fall back on).
 	primaryConsumer := consumer.New(logger.Sugar(), scope.SubScope("consumer"), registry,
 		errs.NewClassifierProcessor(
 			genericerrs.Classifier,
 			mysqlerrs.Classifier,
 		),
 	)
+	dlqConsumer := consumer.New(logger.Sugar(), scope.SubScope("consumer-dlq"), registry,
+		errs.AlwaysRetryableProcessor,
+	)
 
 	processController := process.NewController(
 		logger.Sugar(),
@@ -230,10 +239,23 @@ func run() error {
 		return fmt.Errorf("failed to register process controller: %w", err)
 	}
 
+	processDLQController := dlq.NewController(logger.Sugar(), scope, store, dlq.TopicKey(stovepipemq.TopicKeyProcess), "stovepipe-process-dlq")
+	if err := dlqConsumer.Register(processDLQController); err != nil {
+		return fmt.Errorf("failed to register process dlq controller: %w", err)
+	}
+
+	// Start consumers. DLQ first because Start begins processing messages
+	// immediately; if the primary consumer then fails to start, the half we
+	// already started is the DLQ side, whose work is idempotent reconciliation
+	// and is safe to interrupt mid-flight for rollback.
+	if err := dlqConsumer.Start(ctx); err != nil {
+		return fmt.Errorf("failed to start dlq consumer: %w", err)
+	}
 	if err := primaryConsumer.Start(ctx); err != nil {
-		return fmt.Errorf("failed to start consumer: %w", err)
+		stopErr := dlqConsumer.Stop(30000)
+		return errors.Join(fmt.Errorf("failed to start consumer: %w", err), stopErr)
 	}
-	logger.Info("consumer started")
+	logger.Info("consumers started")
 
 	// Create gRPC server
 	grpcServer := grpc.NewServer()
@@ -298,9 +320,15 @@ func run() error {
 		serverErr = fmt.Errorf("GRPC server exited with error: %w", serverErr)
 	}
 
-	consumerStopErr := primaryConsumer.Stop(30000)
+	// Stop consumers in reverse start order: primary first, then DLQ. The primary
+	// pipeline writes the state that DLQ reconciliation reads, so draining primary
+	// first means in-flight DLQ reconciliation finishes against a settled primary
+	// rather than racing its shutdown.
+	primaryStopErr := primaryConsumer.Stop(30000)
+	dlqStopErr := dlqConsumer.Stop(30000)
+	consumerStopErr := errors.Join(primaryStopErr, dlqStopErr)
 	if consumerStopErr != nil {
-		consumerStopErr = fmt.Errorf("failed to stop consumer: %w", consumerStopErr)
+		consumerStopErr = fmt.Errorf("failed to stop consumers: %w", consumerStopErr)
 	}
 
 	if consumerStopErr != nil || serverErr != nil {
@@ -312,6 +340,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.
 func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRegistry, error) {
 	return consumer.NewTopicRegistry([]consumer.TopicConfig{
 		{
@@ -327,5 +357,11 @@ func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRe
 			Name:  "build",
 			Queue: q,
 		},
+		{
+			Key:          dlq.TopicKey(stovepipemq.TopicKeyProcess),
+			Name:         "process_dlq",
+			Queue:        q,
+			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/BUILD.bazel

@@ -0,0 +1,40 @@
+load("@rules_go//go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "go_default_library",
+    srcs = [
+        "dlq.go",
+        "request.go",
+    ],
+    importpath = "github.com/uber/submitqueue/stovepipe/controller/dlq",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//platform/consumer:go_default_library",
+        "//platform/metrics:go_default_library",
+        "//stovepipe/core/messagequeue:go_default_library",
+        "//stovepipe/entity:go_default_library",
+        "//stovepipe/extension/storage:go_default_library",
+        "@com_github_uber_go_tally//:go_default_library",
+        "@org_uber_go_zap//:go_default_library",
+    ],
+)
+
+go_test(
+    name = "go_default_test",
+    srcs = ["dlq_test.go"],
+    embed = [":go_default_library"],
+    deps = [
+        "//platform/base/messagequeue:go_default_library",
+        "//platform/consumer:go_default_library",
+        "//platform/extension/messagequeue/mock:go_default_library",
+        "//stovepipe/core/messagequeue:go_default_library",
+        "//stovepipe/entity:go_default_library",
+        "//stovepipe/extension/storage:go_default_library",
+        "//stovepipe/extension/storage/mock:go_default_library",
+        "@com_github_stretchr_testify//assert:go_default_library",
+        "@com_github_stretchr_testify//require:go_default_library",
+        "@com_github_uber_go_tally//:go_default_library",
+        "@org_uber_go_mock//gomock:go_default_library",
+        "@org_uber_go_zap//:go_default_library",
+    ],
+)

stovepipe/controller/dlq/dlq.go

@@ -0,0 +1,151 @@
+// Copyright (c) 2025 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 dlq contains controllers that consume messages from a pipeline stage's
+// dead-letter topic and reconcile the affected request into a terminal state.
+//
+// Background. The consumer framework moves a message to its DLQ after the controller
+// for the original topic returns a non-retryable error or exhausts retries on a
+// retryable error. Without DLQ reconciliation the affected request would remain stuck
+// in a non-terminal state (accepted, processing) forever — a caller gating deployments
+// on greenness would see that indistinguishably from "not yet validated" (see
+// doc/rfc/stovepipe/workflow.md#fail-closed-on-unprocessable-work).
+//
+// Reconciliation strategy. Each DLQ topic carries the same payload as its originating
+// topic (the queue framework preserves the bytes verbatim under a new `{topic}_dlq`
+// 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
+// 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.
+package dlq
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"github.com/uber/submitqueue/platform/consumer"
+	"github.com/uber/submitqueue/stovepipe/entity"
+	"github.com/uber/submitqueue/stovepipe/extension/storage"
+	"go.uber.org/zap"
+)
+
+// topicSuffix is appended to a primary topic key to derive the corresponding DLQ topic
+// key. The queue extension's DefaultSubscriptionConfig also uses "_dlq" as the DLQ
+// topic suffix; keeping both in sync is intentional so a registered DLQ subscription's
+// topic name matches the controller's TopicKey().
+const topicSuffix = "_dlq"
+
+// TopicKey returns the DLQ topic key for the given primary pipeline topic. It is
+// exported so the stovepipe wiring layer can build matching pairs without duplicating
+// the suffix literal.
+func TopicKey(main consumer.TopicKey) consumer.TopicKey {
+	return consumer.TopicKey(string(main) + topicSuffix)
+}
+
+// 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. 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 {
+		if errors.Is(err, storage.ErrNotFound) {
+			logger.Warnw("dlq reconcile: request not found, skipping",
+				"request_id", requestID,
+			)
+			return nil
+		}
+		return fmt.Errorf("failed to get request %s: %w", requestID, err)
+	}
+
+	if request.State.IsTerminal() {
+		logger.Infow("dlq reconcile: request already terminal, skipping",
+			"request_id", requestID,
+			"state", string(request.State),
+		)
+		return nil
+	}
+
+	if request.State == entity.RequestStateProcessing {
+		if err := releaseSlot(ctx, store, logger, request.Queue); err != nil {
+			return fmt.Errorf("failed to release queue slot for request %s: %w", requestID, err)
+		}
+	}
+
+	updated := request
+	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 recorded_not_green: %w", requestID, err)
+	}
+	logger.Infow("dlq reconcile: request forced terminal not-green",
+		"request_id", requestID,
+		"previous_state", string(request.State),
+	)
+	return nil
+}
+
+// releaseSlot CAS-decrements the queue's in_flight_count, retrying on version
+// conflicts, mirroring process.Controller's own CAS-retry loop for queue updates.
+func releaseSlot(ctx context.Context, store storage.Storage, logger *zap.SugaredLogger, queueName string) error {
+	queueStore := store.GetQueueStore()
+
+	for {
+		queueRow, err := queueStore.Get(ctx, queueName)
+		if err != nil {
+			if errors.Is(err, storage.ErrNotFound) {
+				logger.Warnw("dlq reconcile: queue not found, skipping slot release",
+					"queue", queueName,
+				)
+				return nil
+			}
+			return fmt.Errorf("failed to get queue %s: %w", queueName, err)
+		}
+
+		if queueRow.InFlightCount <= 0 {
+			logger.Warnw("dlq reconcile: queue in_flight_count already at zero, skipping slot release",
+				"queue", queueName,
+			)
+			return nil
+		}
+
+		updated := queueRow
+		updated.InFlightCount--
+		newVersion := queueRow.Version + 1
+		if err := queueStore.Update(ctx, updated, queueRow.Version, newVersion); err != nil {
+			if errors.Is(err, storage.ErrVersionMismatch) {
+				continue
+			}
+			return fmt.Errorf("failed to release slot for queue %s: %w", queueName, err)
+		}
+		logger.Infow("dlq reconcile: released queue slot",
+			"queue", queueName,
+			"in_flight_count", updated.InFlightCount,
+		)
+		return nil
+	}
+}