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

service/stovepipe/server/BUILD.bazel

@@ -14,6 +14,7 @@ go_library(
         "//platform/extension/messagequeue:go_default_library",
         "//platform/extension/messagequeue/mysql: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

@@ -36,6 +36,7 @@ import (
 	extqueue "github.com/uber/submitqueue/platform/extension/messagequeue"
 	queueMySQL "github.com/uber/submitqueue/platform/extension/messagequeue/mysql"
 	"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"
@@ -202,23 +203,44 @@ 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(), scope, store, queueconfigdefault.NewStore(), stovepipemq.TopicKeyProcess, "stovepipe-process")
 	if err := primaryConsumer.Register(processController); err != nil {
 		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()
@@ -283,9 +305,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 {
@@ -296,8 +324,15 @@ func run() error {
 }
 
 // newTopicRegistry builds the TopicRegistry for Stovepipe's internal pipeline queues. ingest
-// publishes to the process topic and the process consumer subscribes to it.
+// publishes to the process topic and the process consumer subscribes to it. 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.
 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,
@@ -307,5 +342,11 @@ func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRe
 				subscriberName, "stovepipe-process",
 			),
 		},
+		{
+			Key:          dlq.TopicKey(stovepipemq.TopicKeyProcess),
+			Name:         "process_dlq",
+			Queue:        q,
+			Subscription: dlqSub,
+		},
 	})
 }

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,143 @@
+// 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 RequestStateFailed — a conservative terminal state treated as
+// not-green for gating — 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 RequestStateFailed 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.
+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 entity.IsRequestStateTerminal(request.State) {
+		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.RequestStateFailed
+	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)
+	}
+	logger.Infow("dlq reconcile: request marked terminal failed",
+		"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
+	}
+}