Merge branch 'main' into mnoah1/stovepipe-periodic-metrics

Author: mnoah1

doc/rfc/stovepipe/steps/buildsignal.md

@@ -122,13 +122,24 @@ Per `platform/errs`'s non-retryable-by-default rule (see [platform/errs/README.m
 
 | Failure | Disposition | Why |
 |---|---|---|
-| `Status` call | raw error; classifier decides | Deliberately left open rather than fixed either way — runner timeout/connection is transient, "runner not deployed for this queue" is not, and only a backend classifier can tell them apart. |
+| `Status` call | raw error; classifier decides | Deliberately left open rather than fixed either way — runner timeout/connection is transient, "runner not deployed for this queue" is not, and only a backend classifier can tell them apart. **This means the `BuildRunner` backend has to classify**: an unclassified transport or HTTP error gets the non-retryable default, so one proxy blip ends the poll chain (see below). |
 | `Update` CAS conflict (`ErrVersionMismatch`) | declaration-level retryable | A concurrent (redelivered) writer moved the row; reload and re-check converges. |
 
 `Build`/`Request` not found (`storage.ErrNotFound`) are **not** in this table: storage is required to be read-after-write consistent (see [storage README](stovepipe/extension/storage/README.md)), so a miss here is already the correct default (non-retryable, straight to DLQ) rather than a departure worth overriding.
 
 Everything else — factory lookup, an `Update` store error other than a CAS conflict, and the `record` publish — is returned raw with no override, because the default is already correct: a queue with no registered runner is a config error, and storage/queue failures dead-letter and let DLQ reconciliation recover. The poll loop itself no longer has a publish to fail: holding is a local outcome, and a failed postpone write in the framework lapses into a normal visibility-timeout redelivery, so the loop's liveness never rides on an enqueue succeeding.
 
+### What it costs when a backend does not classify `Status` errors
+
+Leaving `Status` to the classifier only works if the backend classifies. A `BuildRunner` whose transport returns plain `fmt.Errorf` values gets the non-retryable default, and here that default is expensive: dead-lettering ends the *only* poll chain for a build that is still running, and the request keeps holding one of the queue's `in_flight_count` build slots until reconciliation gives it back. A single `502` from a proxy in front of the build API then looks exactly like "this build can never be polled".
+
+Two things keep a blip from stalling a queue, and a backend needs both:
+
+- **The backend classifies its own failures.** Transport errors and 5xx/429/408 responses are `errs.NewRetryableDependencyError`. A 4xx about the request itself — unknown build, forbidden — is `errs.NewDependencyError`. Only the layer that sees the status code can tell these apart, which is why the table above leaves the call to it.
+- **The retry budget is worth something.** Retryable means nack, and a nacked message comes back on the next poll, so `Retry.MaxAttempts` counts attempts rather than time — the default three are spent in a few hundred milliseconds. Raising `MaxAttempts` on this subscription buys a little more, but each attempt is another request at a dependency that is already failing, so it does not stretch to cover a proxy restart. Until nacks are spaced by the configured retry backoff, it is the reconciler below rather than the retry budget that keeps a longer outage from costing the queue a slot.
+
+When the budget does run out the message dead-letters, and the buildsignal DLQ reconciler (`stovepipe/controller/dlq/buildsignal.go`) is what makes that recoverable: it maps the build back to its request, releases the slot, and marks the request `failed`. A deployment that registers the primary consumers but not that reconciler has no fail-closed path for this stage, and loses a slot for good every time this happens.
+
 ## Idempotency
 
 Every branch is safe under at-least-once redelivery:

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