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

## Summary

### Why?

Two things go wrong when a speculate message dead-letters, and together they leave a queue quietly stuck.

The first is misattribution. The speculate stage takes a message naming one batch but does a job covering the whole queue: a run lists every in-flight batch, hands them all to the Speculator, and commits outcomes for any of them. Only the handful of errors before the run even starts are about the batch on the message. Everything after is about some other batch, or about the queue itself. Yet the stage shared `dlq.NewDLQBatchController` with build, merge, and conclude — which marks *the named* batch `Failed` and errors all its requests. So a Speculator error, or a storage failure writing another head's path set, terminated a batch that was never at fault while the real culprit carried on.

The second is that the reconcile restored nothing. `failBatch` published no message at all. Speculation is driven only by messages, and a batch admitted to `Speculating` produces no build to signal and no merge to conclude — so once the message that would have funded it is gone, nothing is left to look at it again. The dead letter consumed the queue's last edge and left none behind, stranding every other admitted batch with no error recorded against it and its requests still reading `batched`. From the outside the queue looked like it was working.

### What?

Speculate now says what its failures are about, and its dead letters act on that.

Every error return is attributed with `errs.Attribute`: the message's batch for the errors raised before the run, the specific batch for errors raised while looping over the queue, and `{Type: queue}` for listing the queue and for both Speculator calls, where no batch is at fault. Attribution is added through `entity.BatchSubject` / `entity.QueueSubject`, and a counter tagged by subject type makes queue-scoped failures graphable. Retryability is untouched — the classifiers still read the cause underneath.

The stage gets its own reconciler, `dlq/speculate.go`, instead of the shared batch one:

- Batch subjects are taken at their word and those batches are failed, which may not be the batch on the message.
- A queue subject, or no subjects at all, falls back to the message's batch — a guess, but DLQ reconciliation exists so requests cannot sit non-terminal forever, and that guarantee has to hold even when nothing can say which batch was at fault. Which of the three happened is recorded as `dlq.attribution`, so a fallback is never mistaken for a confident answer.
- Afterwards it republishes one speculate message naming a still-live batch, restoring the edge it consumed. Two guards make it terminate: it runs only after a reconcile that actually transitioned a batch, so a redelivery cannot loop, and it names a *live* batch rather than the one just failed, so each pass fails one more. A genuinely broken queue therefore drains to empty with a reason recorded against every batch, bounded by the batch count, while a queue whose failure was transient or queue-wide recovers on the next run having lost one batch instead of stranding all of them.

Separately, the shared `failRequest` passed `nil` where `RequestLog.Metadata` goes, discarding the failure count, originating topic, and timestamp the reconciler already had in hand and logged. It now carries them, along with the failure's subjects and detail. No schema or proto change is needed: `RequestLog.Metadata` is already persisted and already exposed on the gateway's status and history, so this lands in front of users directly. All five reconcilers benefit.

## Test Plan

✅ `bazel test //...`
✅ `bazel test //test/... --sandbox_writable_path=$HOME/.docker --jobs=1` (11 container suites)
✅ `make lint check-tidy check-gazelle check-mocks`

New coverage in `dlq/speculate_test.go`:

- Attribution as a table: a batch subject fails *that* batch rather than the message's; a queue subject and a subject-less failure each fall back, and each records which it was.
- The re-trigger publishes for a queue that still holds live batches, choosing deterministically (`ListByStates` promises no order).
- The loop guard: an already-failed batch publishes nothing even with live batches remaining, which is what stops a permanently failing queue from re-triggering forever.

## Issue

Closes CODEM-428

Author: behinddwalls

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
 }