feat(messagequeue)!: carry a structured failure across the dead-letter boundary

doc/rfc/submitqueue/speculation-generator-best-first.md

@@ -49,7 +49,7 @@ The batch being built is written before its assumptions. For example, `C [A succ
 
 `Generate` receives the queue's live batches as a snapshot and takes it as given. A well-formed snapshot carries unique, non-empty batch IDs, includes every batch a head's direct dependencies reference, and gives no head an empty, duplicate, or self dependency. Those are preconditions the caller owns, established where the snapshot is assembled. The generator does not re-check them: it is on the hot path of every run, the checks it could make are the ones an assembled-correctly snapshot can never fail, and paying for them here only spreads the same contract across two places. A malformed snapshot yields undefined candidates rather than an error.
 
-A score that is not a probability is the one bad input the generator absorbs, because it arrives from the injected scorer rather than from the caller and there is no earlier point that could catch it. A score outside `[0, 1]`, or `NaN`, is replaced with a default of 0.95 — optimistic on purpose, so a dependency nobody could estimate keeps its head's preferred path near the front instead of burying it or failing the whole run on one number. Any deliberate defaulting still belongs to the scorer implementation, which knows what information it does and does not have; this is only the floor under it.
+A dependency the generator cannot price is the one bad input it absorbs, because it arrives from the injected scorer rather than from the caller and there is no earlier point that could catch it. Three cases take the same 0.95 default: a score outside `[0, 1]` or `NaN`, a scorer call that returned an error, and a dependency the snapshot never carried. The default is optimistic on purpose, so a dependency nobody could estimate keeps its head's preferred path near the front instead of burying it or failing the whole run on one number — and failing the whole run is the real hazard, because `Generate` seeds the heap for every head at once, so one unpriceable dependency would otherwise cost the queue every candidate it had. A batch absent from the snapshot is never passed to the scorer at all: it would resolve to a zero-valued batch belonging to no queue, so scoring it would price some other batch entirely or fail on the empty queue name. Any deliberate defaulting still belongs to the scorer implementation, which knows what information it does and does not have; this is only the floor under it.
 
 ## Step 1: `Generate` prepares each head
 
@@ -447,6 +447,7 @@ The ordering stays the same. `CandidatePath.RankingScore` contains this logarith
 - `Succeeded` fixes an assumption to succeeds.
 - `Failed` or `Cancelled` fixes an assumption to fails.
 - `Cancelling` remains undecided because cancellation may lose a race with completion.
+- `Merging` also remains undecided, because a merge can fail. It is tempting to treat it as committed to landing and skip the scorer call, but that puts a state-specific policy inside the search: whether a path betting against a merging batch is worth funding is a question of price, and price belongs to the scorer. The allocator draws the same line — "no batch state enters this decision" — and the generator holds it too. Nothing is lost by staying open, because a head can never merge ahead of a dependency it took a position on (see [speculation.md](speculation.md)); the cost of an unlikely path is budget, which is the allocator's to ration.
 - A fixed assumption stays in the returned path but contributes probability 1 and has no flip.
 - A shared dependency is scored once per run.
 

platform/base/failure/BUILD.bazel

@@ -0,0 +1,18 @@
+load("@rules_go//go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["failure.go"],
+    importpath = "github.com/uber/submitqueue/platform/base/failure",
+    visibility = ["//visibility:public"],
+)
+
+go_test(
+    name = "go_default_test",
+    srcs = ["failure_test.go"],
+    embed = [":go_default_library"],
+    deps = [
+        "@com_github_stretchr_testify//assert:go_default_library",
+        "@com_github_stretchr_testify//require:go_default_library",
+    ],
+)

platform/base/failure/failure.go

@@ -0,0 +1,112 @@
+// 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 failure holds the shared description of why processing failed: a
+// human-readable message, the entities the failure is about, and free-form
+// detail. It is the vocabulary a producer of a failure and a consumer of it
+// share when they are separated by a queue, so the consumer reads fields
+// rather than parsing prose.
+//
+// The package is deliberately domain-agnostic. It says a failure has subjects
+// and what shape a subject is; which subject types exist is a domain's own
+// business.
+package failure
+
+import "encoding/json"
+
+// Subject names one entity a failure is about.
+//
+// Its purpose is attribution: a consumer reconciling a failure has to know
+// what to act on, and the entity named on the message that failed is not
+// always the entity at fault — a job that reads many records can fail because
+// of any of them, or because of none of them individually.
+type Subject struct {
+	// Type labels what kind of entity ID names, e.g. "batch" or "queue".
+	// Values are chosen by the domain that raises the failure; this package
+	// neither defines nor validates them. Empty means the type is unknown.
+	Type string `json:"type"`
+	// ID identifies the entity within its type. Opaque here: no format is
+	// assumed and none is parsed.
+	ID string `json:"id"`
+}
+
+// Failure describes why processing failed.
+//
+// A failure is always about something. When no single record is at fault, the
+// subject is the wider thing that is — the queue, the tenant, the job — rather
+// than an empty list. That keeps absence from carrying meaning: no subjects at
+// all means the failure is *unattributed*, which is a genuine third state
+// (nothing recorded one, or the record predates attribution) and not a claim
+// that nothing was to blame.
+type Failure struct {
+	// Message is the human-readable reason, typically an error's text. It is
+	// the one field always present, and the one a person reads first.
+	Message string `json:"-"`
+	// Subjects are the entities this failure is about, in no significant
+	// order. Empty means unattributed — see the type comment.
+	Subjects []Subject `json:"subjects,omitempty"`
+	// Detail is free-form structured context: whatever the producer knows that
+	// does not fit the message. Values survive a JSON round trip, so numbers
+	// come back as float64 regardless of what went in.
+	Detail map[string]any `json:"detail,omitempty"`
+}
+
+// New builds a Failure with a message and the subjects it is about.
+func New(message string, subjects ...Subject) Failure {
+	return Failure{Message: message, Subjects: subjects}
+}
+
+// IDsOfType returns the IDs of every subject with the given type, in the order
+// they appear. The result is empty when the failure names no such subject,
+// which is how a consumer asks "is this about one of mine?" without inspecting
+// the slice itself.
+func (f Failure) IDsOfType(subjectType string) []string {
+	var ids []string
+	for _, s := range f.Subjects {
+		if s.Type == subjectType {
+			ids = append(ids, s.ID)
+		}
+	}
+	return ids
+}
+
+// Encode returns the JSON encoding of the structured half of f — its subjects
+// and detail — or nil when there is no structure to store.
+//
+// Message is deliberately excluded. It travels as plain text alongside this
+// blob so that it stays legible to anything reading the underlying store
+// directly, and so decoding never has to guess whether a stored string is an
+// encoded failure or a message that merely looks like one.
+func Encode(f Failure) ([]byte, error) {
+	if len(f.Subjects) == 0 && len(f.Detail) == 0 {
+		return nil, nil
+	}
+	return json.Marshal(f)
+}
+
+// Decode parses the structured half produced by Encode. Empty input yields the
+// zero Failure, which is how an unattributed failure reads.
+//
+// The returned Message is always empty: the caller holds it separately and
+// fills it in.
+func Decode(data []byte) (Failure, error) {
+	if len(data) == 0 {
+		return Failure{}, nil
+	}
+	var f Failure
+	if err := json.Unmarshal(data, &f); err != nil {
+		return Failure{}, err
+	}
+	return f, nil
+}

platform/base/failure/failure_test.go

@@ -0,0 +1,127 @@
+// 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 failure
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestRoundTrip(t *testing.T) {
+	tests := []struct {
+		name string
+		in   Failure
+		want Failure
+	}{
+		{
+			name: "subjects and detail",
+			in: New("speculator failed", Subject{Type: "queue", ID: "test-queue"}).
+				withDetail(map[string]any{"stage": "ask"}),
+			want: Failure{
+				Subjects: []Subject{{Type: "queue", ID: "test-queue"}},
+				Detail:   map[string]any{"stage": "ask"},
+			},
+		},
+		{
+			name: "several subjects keep their order",
+			in:   New("two at fault", Subject{Type: "batch", ID: "q/batch/2"}, Subject{Type: "batch", ID: "q/batch/1"}),
+			want: Failure{Subjects: []Subject{{Type: "batch", ID: "q/batch/2"}, {Type: "batch", ID: "q/batch/1"}}},
+		},
+		{
+			name: "nested detail survives",
+			in:   Failure{Detail: map[string]any{"path": map[string]any{"id": "abc"}}},
+			want: Failure{Detail: map[string]any{"path": map[string]any{"id": "abc"}}},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			encoded, err := Encode(tt.in)
+			require.NoError(t, err)
+			require.NotEmpty(t, encoded)
+
+			got, err := Decode(encoded)
+			require.NoError(t, err)
+			assert.Equal(t, tt.want, got)
+		})
+	}
+}
+
+// The message is carried outside the blob so that whatever stores it keeps a
+// legible column, and so decoding never has to tell an encoded failure apart
+// from a message that happens to look like one.
+func TestEncodeOmitsMessage(t *testing.T) {
+	encoded, err := Encode(New("boom", Subject{Type: "batch", ID: "q/batch/1"}))
+	require.NoError(t, err)
+	assert.NotContains(t, string(encoded), "boom")
+
+	got, err := Decode(encoded)
+	require.NoError(t, err)
+	assert.Empty(t, got.Message)
+	assert.Equal(t, []Subject{{Type: "batch", ID: "q/batch/1"}}, got.Subjects)
+}
+
+// Nothing structured means nothing to store, which is what lets a caller treat
+// an absent blob as "unattributed" without a sentinel.
+func TestEncodeNothingStructured(t *testing.T) {
+	encoded, err := Encode(New("just a message"))
+	require.NoError(t, err)
+	assert.Nil(t, encoded)
+}
+
+func TestDecodeEmpty(t *testing.T) {
+	got, err := Decode(nil)
+	require.NoError(t, err)
+	assert.Equal(t, Failure{}, got)
+}
+
+func TestDecodeMalformed(t *testing.T) {
+	_, err := Decode([]byte("not json"))
+	assert.Error(t, err)
+}
+
+// Detail goes through encoding/json, so every number returns as a float64
+// whatever its Go type going in. Pinned because a caller that stores an int64
+// and reads it back expecting one would otherwise find out at runtime.
+func TestDetailNumbersDecodeAsFloat64(t *testing.T) {
+	encoded, err := Encode(Failure{Detail: map[string]any{"attempt": int64(3)}})
+	require.NoError(t, err)
+
+	got, err := Decode(encoded)
+	require.NoError(t, err)
+	assert.Equal(t, float64(3), got.Detail["attempt"])
+}
+
+func TestIDsOfType(t *testing.T) {
+	f := New("mixed",
+		Subject{Type: "batch", ID: "q/batch/1"},
+		Subject{Type: "queue", ID: "q"},
+		Subject{Type: "batch", ID: "q/batch/2"},
+	)
+
+	assert.Equal(t, []string{"q/batch/1", "q/batch/2"}, f.IDsOfType("batch"))
+	assert.Equal(t, []string{"q"}, f.IDsOfType("queue"))
+	assert.Empty(t, f.IDsOfType("request"))
+	assert.Empty(t, Failure{}.IDsOfType("batch"))
+}
+
+// withDetail keeps the table above readable; New covers message and subjects,
+// which is what most callers set.
+func (f Failure) withDetail(detail map[string]any) Failure {
+	f.Detail = detail
+	return f
+}

platform/consumer/BUILD.bazel

@@ -10,6 +10,7 @@ go_library(
     importpath = "github.com/uber/submitqueue/platform/consumer",
     visibility = ["//visibility:public"],
     deps = [
+        "//platform/base/failure:go_default_library",
         "//platform/base/messagequeue:go_default_library",
         "//platform/errs:go_default_library",
         "//platform/extension/consumergate:go_default_library",
@@ -28,6 +29,7 @@ go_test(
     ],
     embed = [":go_default_library"],
     deps = [
+        "//platform/base/failure:go_default_library",
         "//platform/base/messagequeue:go_default_library",
         "//platform/errs:go_default_library",
         "//platform/extension/consumergate:go_default_library",

platform/consumer/consumer.go

@@ -424,6 +424,12 @@ func (m *consumer) processDelivery(ctx context.Context, controller Controller, d
 		// cancelled by the processing context during shutdown.
 		isCanceled := errors.Is(err, context.Canceled)
 
+		// Whatever the controller attributed the failure to, plus the error's
+		// own text as the message. A controller that attributed nothing yields
+		// the message alone, which is what every caller sent before failures
+		// carried structure.
+		controllerFailure := errs.Attribution(err)
+
 		// Check if the error is non-retryable (poison pill message)
 		if !errs.IsRetryable(err) {
 			m.logger.Errorw("non-retryable controller error, rejecting message",
@@ -438,7 +444,7 @@ func (m *consumer) processDelivery(ctx context.Context, controller Controller, d
 
 			// Reject moves to DLQ (or acks if DLQ disabled)
 			rejectOp := metrics.Begin(controllerScope, "reject", metrics.StorageLatencyBuckets)
-			rejectErr := delivery.Reject(ctx, err.Error())
+			rejectErr := delivery.Reject(ctx, controllerFailure)
 			rejectOp.Complete(rejectErr)
 			if rejectErr != nil {
 				m.logger.Errorw("failed to reject non-retryable message",
@@ -468,9 +474,11 @@ func (m *consumer) processDelivery(ctx context.Context, controller Controller, d
 			"elapsed_ms", elapsed.Milliseconds(),
 		)
 
-		// Nack requeues immediately - the visibility timeout spaces retries
+		// Nack requeues immediately - the visibility timeout spaces retries.
+		// The failure travels with it so that the attempt which finally spends
+		// the retry budget can dead-letter saying why.
 		nackOp := metrics.Begin(controllerScope, "nack", metrics.StorageLatencyBuckets)
-		nackErr := delivery.Nack(ctx)
+		nackErr := delivery.Nack(ctx, controllerFailure)
 		nackOp.Complete(nackErr)
 		if nackErr != nil {
 			m.logger.Errorw("failed to nack message",