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

## Summary

### Why?

Today the only thing that crosses the dead-letter boundary is a string, and on the most common path not even that.

`Nack(ctx)` takes no reason at all, so when a retryable error exhausts its budget the poll loop dead-letters the message with the hardcoded literal `"exceeded retry limit"`. The error that actually caused it exists only in a log line. `Reject` does better — it passes `err.Error()` — but that is a flattened `fmt.Errorf` chain, so a consumer wanting to know *what* failed has nothing to read but prose.

That is a problem for any stage whose work is wider than the message that triggers it. Such a stage can fail because of an entity the message does not name, or because of nothing in particular, and a dead-letter reconciler has no way to tell — so it acts on the entity it does have and can terminate the wrong one.

### What?

A failure now travels as data. `platform/base/failure` defines `Failure{Message, Subjects, Detail}`, where a `Subject{Type, ID}` names an entity the failure is about. Types are domain-chosen, so the platform stays domain-agnostic. A failure is always about *something*: when no single record is at fault the subject is the wider thing that is, which leaves an empty subject list to mean "unattributed" rather than "nothing was to blame".

The pieces:

- `errs.Attribute` / `errs.Detail` attach subjects and context to an error; `errs.Attribution` reads them back, merging across layers. The wrapper implements `Unwrap` and is neither a `userError` nor an `infraError`, so classification walks straight through it and retryability is unchanged.
- `Nack` and `Reject` take a `Failure`, and `Delivery.Failure()` returns the one recorded against a dead-lettered message.
- The consumer builds the failure from the error the controller returned, defaulting the message to `err.Error()`. A controller that attributes nothing produces exactly what callers sent before, which is what makes this change behaviour-neutral.
- `Nack` also dead-letters directly once the retry budget is spent, so the attempt still holding the reason is the one that records it. The poll-time check stays as a backstop for a delivery that never reaches `Nack` — a crash, or a visibility timeout — and that path keeps the generic literal. Message count is unchanged: at attempt N the next poll would have dead-lettered iff `N >= MaxAttempts`, which is the condition `Nack` now applies.

Storage splits the failure across two columns of `queue_messages`. `last_error` keeps the human-readable message, unchanged in meaning, so nothing has to decode it and `SELECT last_error` stays useful. A new `failure_detail JSON` column holds subjects and detail. The split is what makes the round trip unambiguous — one column would force a decoder to guess whether the text is an envelope or a message, and any error whose text happened to be valid JSON would decode as a malformed envelope and lose the message. An absent `failure_detail` means unattributed, with no heuristic involved, which is exactly the state of rows written before this column existed and of the retry-limit backstop.

`failure_detail` is nullable rather than taking an empty sentinel like its neighbours: a JSON column rejects `''`. `MoveToDLQ` binds SQL NULL explicitly instead of a nil `[]byte`, so the value does not depend on driver conversion.

No behaviour changes for any existing caller. This is the mechanism a follow-up uses to attribute speculation failures and stop them stranding a queue.

## Test Plan

✅ `bazel test //platform/... //submitqueue/... //runway/... //stovepipe/...`
✅ `bazel test //test/integration/extension/messagequeue/...`
✅ `make lint check-tidy check-gazelle check-mocks`

New coverage:

- `platform/base/failure` — codec round trip including nested detail, and the `float64` number-decoding behaviour pinned so it cannot surprise a caller later.
- `platform/errs` — the regression that matters: `errors.As` still reaches a wrapped driver error *through* the envelope, and a classifier still marks it retryable. Had the wrapper broken the chain walk, every storage error would have silently stopped being retryable.
- `platform/extension/messagequeue/mysql` — the nack-time dead-letter boundary as a table (budget remaining, one attempt left, final attempt, single-attempt budget, unset budget), because dead-lettering one attempt early would silently cost every message a retry.
- `platform/consumer` — an unattributed controller error yields a message-only failure, proving behaviour-neutrality.
- Integration — the DLQ test previously asserted `dlq.last_error == "exceeded retry limit"`; it now asserts the real reason from the final nack, and that the subject survives the round trip.

## Issue

Part of CODEM-428

Author: behinddwalls

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",