fix(speculate): blame the batch that failed, and stop the queue stranding

submitqueue/entity/BUILD.bazel

@@ -24,11 +24,13 @@ go_library(
         "request_log.go",
         "request_summary.go",
         "speculation.go",
+        "subject.go",
     ],
     importpath = "github.com/uber/submitqueue/submitqueue/entity",
     visibility = ["//visibility:public"],
     deps = [
         "//platform/base/change:go_default_library",
+        "//platform/base/failure:go_default_library",
         "//platform/base/mergestrategy:go_default_library",
     ],
 )

submitqueue/entity/subject.go

@@ -0,0 +1,46 @@
+// 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 entity
+
+import "github.com/uber/submitqueue/platform/base/failure"
+
+// Subject types name what a failure is about. The queue layer treats the type
+// as opaque; these are SubmitQueue's own vocabulary for it.
+const (
+	// SubjectTypeBatch identifies a subject by batch ID.
+	SubjectTypeBatch = "batch"
+	// SubjectTypeQueue identifies a subject by queue name. Used where no single
+	// batch is at fault — a failure reading or planning the queue as a whole.
+	SubjectTypeQueue = "queue"
+	// SubjectTypeRequest identifies a subject by request ID.
+	SubjectTypeRequest = "request"
+)
+
+// BatchSubject names a batch as what a failure is about.
+func BatchSubject(batchID string) failure.Subject {
+	return failure.Subject{Type: SubjectTypeBatch, ID: batchID}
+}
+
+// QueueSubject names a queue as what a failure is about. It is the honest
+// subject for work that spans the queue — listing it, planning it — where
+// blaming any one batch would be a guess.
+func QueueSubject(queue string) failure.Subject {
+	return failure.Subject{Type: SubjectTypeQueue, ID: queue}
+}
+
+// RequestSubject names a request as what a failure is about.
+func RequestSubject(requestID string) failure.Subject {
+	return failure.Subject{Type: SubjectTypeRequest, ID: requestID}
+}

submitqueue/orchestrator/controller/dlq/BUILD.bazel

@@ -10,6 +10,7 @@ go_library(
         "mergeconflictsignal.go",
         "mergesignal.go",
         "request.go",
+        "speculate.go",
     ],
     importpath = "github.com/uber/submitqueue/submitqueue/orchestrator/controller/dlq",
     visibility = ["//visibility:public"],
@@ -18,7 +19,9 @@ go_library(
         "//platform/consumer:go_default_library",
         "//platform/metrics:go_default_library",
         "//submitqueue/core/batch:go_default_library",
+        "//submitqueue/core/publish:go_default_library",
         "//submitqueue/core/request:go_default_library",
+        "//submitqueue/core/topickey:go_default_library",
         "//submitqueue/entity:go_default_library",
         "//submitqueue/extension/storage:go_default_library",
         "@com_github_uber_go_tally//:go_default_library",
@@ -37,11 +40,13 @@ go_test(
         "mergesignal_test.go",
         "publisher_test.go",
         "request_test.go",
+        "speculate_test.go",
     ],
     embed = [":go_default_library"],
     deps = [
         "//api/runway/messagequeue:go_default_library",
         "//api/runway/messagequeue/protopb:go_default_library",
+        "//platform/base/failure:go_default_library",
         "//platform/base/messagequeue:go_default_library",
         "//platform/consumer:go_default_library",
         "//platform/consumer/mock:go_default_library",

submitqueue/orchestrator/controller/dlq/batch.go

@@ -27,15 +27,21 @@ import (
 )
 
 // batchController is the DLQ reconciler for batch-scoped pipeline stages
-// (speculate, build, merge, conclude). All four topics carry a
-// BatchID payload, so this controller is registered four times — one per
-// topic, each with the matching DLQ topic key and consumer group.
+// (build, merge, conclude). All three topics carry a BatchID payload, so this
+// controller is registered three times — one per topic, each with the matching
+// DLQ topic key and consumer group.
 //
 // On each delivery the controller decodes the BatchID, transitions the batch
 // to BatchStateFailed (idempotent if already halted), and fans out by
 // transitioning each member request to RequestStateError. The fan-out exists
 // because conclude — which normally drives request state from batch state —
 // will not run for a DLQ'd batch.
+//
+// Blaming the batch on the message is right for these stages because their
+// work is that batch: whatever failed, it failed doing this batch's build,
+// merge, or conclusion. The speculate stage is not like that — it re-plans a
+// whole queue from a message that names one batch — so it has its own
+// reconciler; see speculate.go.
 type batchController struct {
 	logger        *zap.SugaredLogger
 	metricsScope  tally.Scope
@@ -92,6 +98,7 @@ func (c *batchController) Process(ctx context.Context, delivery consumer.Deliver
 		return fmt.Errorf("failed to resolve storage for queue %q: %w", bid.Queue, err)
 	}
 
+	lastError, failureMeta := failureContext(delivery)
 	dmeta := delivery.Metadata()
 	c.logger.Warnw("dlq message received",
 		"batch_id", bid.ID,
@@ -101,7 +108,7 @@ func (c *batchController) Process(ctx context.Context, delivery consumer.Deliver
 		"dlq_last_error", dmeta["dlq.last_error"],
 	)
 
-	if err := failBatch(ctx, store, c.registry, c.logger, bid.ID, dmeta["dlq.last_error"]); err != nil {
+	if _, err := failBatch(ctx, store, c.registry, c.logger, bid.ID, lastError, failureMeta); err != nil {
 		metrics.NamedCounter(c.metricsScope, opName, "reconcile_errors", 1)
 		return err
 	}

submitqueue/orchestrator/controller/dlq/buildsignal.go

@@ -90,6 +90,7 @@ func (c *buildSignalController) Process(ctx context.Context, delivery consumer.D
 		return fmt.Errorf("failed to resolve storage for queue %q: %w", buildID.Queue, err)
 	}
 
+	lastError, failureMeta := failureContext(delivery)
 	dmeta := delivery.Metadata()
 	c.logger.Warnw("dlq message received",
 		"build_id", buildID.ID,
@@ -121,7 +122,7 @@ func (c *buildSignalController) Process(ctx context.Context, delivery consumer.D
 		return nil
 	}
 
-	if err := failBatch(ctx, store, c.registry, c.logger, build.BatchID, dmeta["dlq.last_error"]); err != nil {
+	if _, err := failBatch(ctx, store, c.registry, c.logger, build.BatchID, lastError, failureMeta); err != nil {
 		metrics.NamedCounter(c.metricsScope, opName, "reconcile_errors", 1)
 		return err
 	}

submitqueue/orchestrator/controller/dlq/dlq.go

@@ -37,6 +37,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"strings"
 
 	"github.com/uber/submitqueue/platform/consumer"
 	corebatch "github.com/uber/submitqueue/submitqueue/core/batch"
@@ -62,19 +63,81 @@ func TopicKey(main consumer.TopicKey) consumer.TopicKey {
 	return consumer.TopicKey(string(main) + topicSuffix)
 }
 
+// failureContext reads everything the queue recorded about a dead-lettered
+// message: the human-readable reason, and a metadata map to carry alongside it
+// on the terminal request log.
+//
+// The map is what makes a dead letter diagnosable after the fact. The queue
+// already hands the reconciler the failure count, the topic it failed on, and
+// when — and, when the producer attributed it, which entities it was about.
+// All of that used to be logged and then dropped; the request log is where a
+// user can actually see it, through the gateway's status and history.
+//
+// Values are flattened to strings because RequestLog.Metadata and the gateway's
+// wire contract are both string maps. Nested detail becomes dotted keys, which
+// read well in a display surface where a JSON blob would not.
+func failureContext(delivery consumer.Delivery) (string, map[string]string) {
+	dmeta := delivery.Metadata()
+	metadata := make(map[string]string, len(dmeta))
+	for _, key := range []string{"dlq.original_topic", "dlq.failure_count", "dlq.failed_at"} {
+		if v, ok := dmeta[key]; ok && v != "" {
+			metadata[key] = v
+		}
+	}
+
+	lastError := dmeta["dlq.last_error"]
+
+	f, failed := delivery.Failure()
+	if !failed {
+		return lastError, metadata
+	}
+	if f.Message != "" {
+		lastError = f.Message
+	}
+
+	// One key per subject type, so several batches at fault read as a list
+	// rather than overwriting each other.
+	byType := make(map[string][]string, len(f.Subjects))
+	for _, s := range f.Subjects {
+		byType[s.Type] = append(byType[s.Type], s.ID)
+	}
+	for subjectType, ids := range byType {
+		metadata["dlq.subject."+subjectType] = strings.Join(ids, ",")
+	}
+
+	flattenDetail("dlq.detail", f.Detail, metadata)
+
+	return lastError, metadata
+}
+
+// flattenDetail writes a JSON-shaped document into a string map, joining nested
+// keys with dots. Values are rendered with %v: this is a display surface, not a
+// contract, so a readable rendering beats a faithful one.
+func flattenDetail(prefix string, detail map[string]any, out map[string]string) {
+	for k, v := range detail {
+		key := prefix + "." + k
+		if nested, ok := v.(map[string]any); ok {
+			flattenDetail(key, nested, out)
+			continue
+		}
+		out[key] = fmt.Sprintf("%v", v)
+	}
+}
+
 // failRequest transitions a non-terminal request to RequestStateError and
 // appends the matching RequestStatusError log. Redelivery for an existing Error
 // state repeats materialization to repair a previous partial attempt. A
 // different terminal outcome is left unchanged.
 // lastError is the failure reason preserved by the queue in DLQ delivery
-// metadata and is exposed through Status and History for diagnosis.
+// metadata and is exposed through Status and History for diagnosis, alongside
+// metadata carrying the rest of the failure context.
 //
 // A request in RequestStateCancelling is reconciled to RequestStateError, not
 // left in place: DLQ means the pipeline failed to converge, so we cannot
 // confirm the cancel completed cleanly. Writing Error is the honest signal and
 // keeps the request from being stuck in a non-terminal state forever.
-func failRequest(ctx context.Context, store storage.Storage, registry consumer.TopicRegistry, logger *zap.SugaredLogger, requestID, lastError string) error {
-	res, err := requestcore.TerminateRequest(ctx, store, registry, requestID, entity.RequestStateError, lastError, nil)
+func failRequest(ctx context.Context, store storage.Storage, registry consumer.TopicRegistry, logger *zap.SugaredLogger, requestID, lastError string, metadata map[string]string) error {
+	res, err := requestcore.TerminateRequest(ctx, store, registry, requestID, entity.RequestStateError, lastError, metadata)
 	if err != nil {
 		return fmt.Errorf("dlq reconcile request %s failed: %w", requestID, err)
 	}
@@ -112,19 +175,27 @@ func failRequest(ctx context.Context, store storage.Storage, registry consumer.T
 // Idempotency: an existing Failed batch repeats fan-out because a previous
 // attempt may have crashed after updating the batch. Succeeded and Cancelled
 // are different terminal outcomes and do not fan out errors.
-// lastError is propagated to each member request's terminal Error log.
-func failBatch(ctx context.Context, store storage.Storage, registry consumer.TopicRegistry, logger *zap.SugaredLogger, batchID, lastError string) error {
+// lastError and metadata are propagated to each member request's terminal
+// Error log.
+//
+// It reports whether it transitioned the batch. A caller that only wants to act
+// on real progress — republishing to wake the queue, say — can then tell a
+// first reconcile from a redelivery of one already done, and avoid doing it
+// again forever.
+func failBatch(ctx context.Context, store storage.Storage, registry consumer.TopicRegistry, logger *zap.SugaredLogger, batchID, lastError string, metadata map[string]string) (bool, error) {
 	batch, err := store.GetBatchStore().Get(ctx, batchID)
 	if err != nil {
 		if errors.Is(err, storage.ErrNotFound) {
 			logger.Warnw("dlq reconcile: batch not found, skipping",
 				"batch_id", batchID,
 			)
-			return nil
+			return false, nil
 		}
-		return fmt.Errorf("failed to get batch %s: %w", batchID, err)
+		return false, fmt.Errorf("failed to get batch %s: %w", batchID, err)
 	}
 
+	transitioned := false
+
 	switch batch.State {
 	case entity.BatchStateFailed:
 		logger.Infow("dlq reconcile: batch already failed, repairing request fan-out",
@@ -133,31 +204,32 @@ func failBatch(ctx context.Context, store storage.Storage, registry consumer.Top
 		// A prior attempt may have CAS'd to Failed without completing the
 		// membership record move; repair it alongside the fan-out.
 		if err := corebatch.EnsureRecord(ctx, store, batch); err != nil {
-			return err
+			return false, err
 		}
 	case entity.BatchStateSucceeded, entity.BatchStateCancelled:
 		logger.Infow("dlq reconcile: batch has a different terminal outcome, skipping",
 			"batch_id", batchID,
 			"state", string(batch.State),
 		)
-		return nil
+		return false, nil
 	default:
 		previousState := batch.State
 		updated, err := corebatch.Transition(ctx, store, batch, entity.BatchStateFailed)
 		if err != nil {
-			return err
+			return false, err
 		}
 		batch = updated
+		transitioned = true
 		logger.Infow("dlq reconcile: batch marked failed",
 			"batch_id", batchID,
 			"previous_state", string(previousState),
 		)
 	}
 
 	for _, requestID := range batch.Contains {
-		if err := failRequest(ctx, store, registry, logger, requestID, lastError); err != nil {
-			return fmt.Errorf("fan-out for batch %s: %w", batchID, err)
+		if err := failRequest(ctx, store, registry, logger, requestID, lastError, metadata); err != nil {
+			return transitioned, fmt.Errorf("fan-out for batch %s: %w", batchID, err)
 		}
 	}
-	return nil
+	return transitioned, nil
 }