From 620212365793202e1a574d6101fe02d819aee02e Mon Sep 17 00:00:00 2001 From: Preetam Dwivedi Date: Mon, 10 Aug 2026 09:25:03 -0700 Subject: [PATCH 1/2] fix(speculation): stop one dependency failing the whole snapshot MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary ### Why? Speculation died for an entire queue whenever any batch depended on one that had reached `merging`: ``` speculator failed for queue demo-queue: score dependency "demo-queue/batch/1": failed to resolve storage for queue "": queue name must not be empty ``` The root cause is a caller contract violation, not a generator bug. `speculator.Speculate` documents `batches` as "every in-flight batch of the queue, plus any finalized batch still referenced as a dependency by an in-flight one". `ask` was passing `snap.speculating` — the speculating heads alone. `read` already assembles the right set in `snap.batches`; it simply was not the slice handed over. A merging dependency was therefore absent from the generator's index, `batchByID[id]` returned a zero `entity.Batch`, and that zero batch reached the scorer with an empty `Queue`. The blast radius was the whole queue rather than one batch, because `Generate` scores every unresolved dependency up front to seed its heap and returned on the first error. `standard.Speculate` then short-circuited before the allocator pulled a single candidate: no path failed, none was ever produced, and the fall-back-to-the-next-path machinery sits downstream of a stage that had already died. A measured run of 20 requests left 9 in `error` and 11 wedged in `speculating` with nothing recorded against them. ### What? **Hand the Speculator the whole queue.** `ask` now flattens `snap.batches` — every batch the run read — and passes that. No ordering is imposed: the generator's heap comparator is a strict total order, so its candidate sequence is identical whatever order the input arrives in (checked over 200 shuffles), and a Speculator that read meaning into input order would be relying on something the contract never offered. `ask`'s doc comment argued for the narrow slice and is rewritten. Widening is safe because of the commit below this one. A head this run has just decided still reads as `Speculating` in the snapshot — `finalize` records only terminal outcomes — and both the Speculator and `check` read head eligibility off that same map, so a stale entry would fool both filters at once. What stops it mattering is that a head now merges only once *every* dependency has settled: the generator pins them all, can therefore construct nothing but the path that already passed, and the allocator skips that as finished. Verified by driving the real `bestfirst`/`sticky` pair with a merged head and settled dependencies — zero actions proposed for it. That ordering is load-bearing, which is why the merge gate is the parent rather than a follow-up. With the old gate a head could merge past a dependency that was still live; the generator would then see that dependency as an open question, offer a path ID the set had never held, and the allocator would fund a fresh build for a batch already handed to Runway. **Make one unpriceable dependency cost only its own estimate.** `score` now substitutes `defaultProbability` when the scorer returns an error, and never calls the scorer at all for a dependency the snapshot did not carry — that batch is zero in every field, so scoring it would price some other batch entirely or fail on its empty queue name. Context cancellation is still fatal, including when it surfaces *as* the scorer's error: the loop checks `ctx` before each call, so a context that dies during the last one would otherwise be absorbed as an unpriceable dependency and hand back an iterator to a caller that has already gone. Scorer failures stay observable through the scorer's own metrics span, which already reports them via `op.Complete(retErr)`. That is the whole containment fix. An earlier revision of this branch also made scoring lazy — heads seeded at an optimistic bound and priced on first pull — and it has been dropped. The only admissible bound for an unpriced head is `log 1`, identical for every head, so the first pull priced the entire queue anyway; the laziness bought one narrow case (a run that pulls nothing because the budget is saturated) in exchange for a placeholder, an admissibility argument, priced and unpriced items sharing a heap, and five reworked tests. Defaulting on failure fixes the bug on its own. **`Merging` is left as an open question in the generator.** Tempting to pin it to *succeeds* — the batch looks committed to landing — but a merge can fail, so nothing is settled, and it would put a state-specific policy inside the search when whether a path betting against a merging batch is worth funding is a question of price that belongs to the scorer. The allocator already draws exactly this line — "no batch state enters this decision — `merging` and the rest are states of a batch, never of a path" — and the generator holds it too. ## Test Plan - ✅ `make test` — 96/96 pass - ✅ `make lint`, `make check-gazelle`, `make check-tidy` New and reworked coverage, per defect: - `TestRun_PassesSnapshotToSpeculator` — the Speculator receives every batch the run read, in a stable order - `TestBestFirst_AbsorbsScorerError` — a scorer error costs that dependency its estimate and nothing else - `TestBestFirst_NeverScoresAnAbsentDependency` — an absent dependency never reaches the scorer, even when the caller hands over a malformed snapshot - `TestBestFirst_MergingDependencyStaysOpen` — a merging dependency is priced like any other and keeps both sides - `TestBestFirst_HonorsCancelledContext/a scorer that fails on a dead context ends the run` The last was added for a defect a review of this branch turned up, and was confirmed to fail against the code as it stood before the fix. Not verified end to end: `make demo-pr` lives on the `sq/demo-pr` branch, so reproducing the original 20-request run needs that target ported across worktrees plus a Docker stack. ## Issue Fixes https://linear.app/uber/issue/CODEM-424 That issue proposed lazy scoring as its primary fix; this lands the containment it was after without the algorithm change, for the reasons above. Follow-up filed as https://linear.app/uber/issue/CODEM-428 — this stops queues wedging this way, but a queue already wedged still has no event that will wake it, because speculation is edge-triggered only. The merge-gate defect found while investigating this one — a head could merge on a *fails* assumption that had not come true — is the parent commit, since the widening here relies on the invariant it restores. --- .../speculation-generator-best-first.md | 3 +- .../extension/speculation/generator/README.md | 2 +- .../speculation/generator/bestfirst/README.md | 4 +- .../generator/bestfirst/bestfirst.go | 36 +++++++-- .../generator/bestfirst/bestfirst_test.go | 74 ++++++++++++++++++- .../orchestrator/controller/speculate/run.go | 25 ++++--- .../controller/speculate/run_test.go | 10 +-- .../controller/speculate/speculate_test.go | 12 +-- 8 files changed, 131 insertions(+), 35 deletions(-) diff --git a/doc/rfc/submitqueue/speculation-generator-best-first.md b/doc/rfc/submitqueue/speculation-generator-best-first.md index 47df525f8..cfe87f37e 100644 --- a/doc/rfc/submitqueue/speculation-generator-best-first.md +++ b/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. diff --git a/submitqueue/extension/speculation/generator/README.md b/submitqueue/extension/speculation/generator/README.md index 61a0613a2..93d6de520 100644 --- a/submitqueue/extension/speculation/generator/README.md +++ b/submitqueue/extension/speculation/generator/README.md @@ -2,7 +2,7 @@ The `generator` package is a piece the `standard` `Speculator` is built from: a `Generator` produces the queue's candidate paths as one ordered stream across all heads. It is **not** controller-facing — the speculate controller only knows the `Speculator` contract, and a different `Speculator` need not split its work this way. So there is no `Config` or `Factory` here; a `Generator` is chosen when the `standard` `Speculator` is constructed. -`Generate` starts the stream over the queue's live batches and returns an `Iterator`. Beyond any ranking work required up front, the generator computes only the candidates the caller pulls. A cancelled or expired context ends the stream with its error. The snapshot must include every batch a head's direct dependencies reference; a snapshot that does not — or that carries empty or duplicate IDs, or a head with an empty, duplicate, or self dependency — is malformed input and errors instead of opening a stream. +`Generate` starts the stream over the queue's live batches and returns an `Iterator`. Beyond any ranking work required up front, the generator computes only the candidates the caller pulls. A cancelled or expired context ends the stream with its error. The snapshot must include every batch a head's direct dependencies reference, carry unique non-empty IDs, and give no head an empty, duplicate, or self dependency. Those are the caller's preconditions: a generator may assume them and is not required to detect a breach, so a malformed snapshot yields undefined candidates rather than an error. Candidates never repeat and never contradict a known fact. Beyond that, the order is the `Generator`'s own: it yields candidates in whatever ranking it implements, and each carries the score it ranked by — higher first, on a scale the generator defines. Consumers take the iterator in the order given and do not interpret the score. Scores mean something only within the run and are never stored. diff --git a/submitqueue/extension/speculation/generator/bestfirst/README.md b/submitqueue/extension/speculation/generator/bestfirst/README.md index 8f924a865..bd34769dd 100644 --- a/submitqueue/extension/speculation/generator/bestfirst/README.md +++ b/submitqueue/extension/speculation/generator/bestfirst/README.md @@ -10,6 +10,8 @@ The [best-first speculation path generation RFC](../../../../../doc/rfc/submitqu - `Next` removes the highest-ranked candidate, advances only that head's stream, constructs that candidate's complete path, and returns it. Pulling long enough returns every path exactly once in non-increasing score order. - Ranking scores are sums of log probabilities, avoiding underflow while preserving probability order. They are meaningful only within the run that produced them. - Exact ties prefer fewer flips; head ID then decides between heads (the cross-head heap holds one candidate per head), and taken flip indexes decide within a head. -- The snapshot must contain every batch a head's direct dependencies reference; a snapshot missing one — or carrying empty or duplicate batch IDs, or a head with an empty, duplicate, or self dependency — is malformed input and errors instead of opening a stream. Any defaulting for a batch that is hard to score belongs to the scorer, not the generator. +- A dependency counts as resolved only once it is terminal. Merging and cancelling are both still in progress and either can end the other way, so both stay open questions here. Whether a path betting against a merging dependency is worth funding is a matter of price, and price is the scorer's to say. +- A dependency that cannot be priced — the scorer call failed, the score was not a probability, or the snapshot never carried the batch — is treated as very likely to succeed rather than ending the run. One unusable number costs its own estimate, never the queue's whole set of candidates. A batch missing from the snapshot is never handed to the scorer at all: it would resolve to a zero batch belonging to no queue. +- The snapshot must contain every batch a head's direct dependencies reference, carry unique non-empty batch IDs, and give no head an empty, duplicate, or self dependency. That is the caller's precondition, not something checked here: a malformed snapshot yields undefined candidates rather than an error. The behavior is covered by `bestfirst_test.go`. diff --git a/submitqueue/extension/speculation/generator/bestfirst/bestfirst.go b/submitqueue/extension/speculation/generator/bestfirst/bestfirst.go index 228da1d36..7d72999c3 100644 --- a/submitqueue/extension/speculation/generator/bestfirst/bestfirst.go +++ b/submitqueue/extension/speculation/generator/bestfirst/bestfirst.go @@ -25,7 +25,6 @@ import ( "cmp" "container/heap" "context" - "fmt" "maps" "math" "slices" @@ -104,26 +103,47 @@ func speculatingHeads(batches []entity.Batch, batchByID map[string]entity.Batch) // score asks the scorer for each unresolved dependency exactly once, however // many heads wait on it. +// +// A dependency that cannot be priced takes defaultProbability rather than +// ending the run — one unusable number must not cost the queue every candidate +// it had. Only cancellation is an error. func (g *bestFirst) score(ctx context.Context, ids []string, batchByID map[string]entity.Batch) (map[string]float64, error) { probabilityByID := make(map[string]float64, len(ids)) for _, id := range ids { if err := ctx.Err(); err != nil { return nil, err } - probability, err := g.scorer.Score(ctx, batchByID[id]) + batch, known := batchByID[id] + if !known { + // A batch the snapshot never carried is zero in every field, not + // just missing — scoring it would price some other batch entirely, + // or fail on its empty queue. It is unpriceable, not cheap. + probabilityByID[id] = defaultProbability + continue + } + probability, err := g.scorer.Score(ctx, batch) if err != nil { - return nil, fmt.Errorf("score dependency %q: %w", id, err) + // A scorer that failed because the caller went away has not found + // an unpriceable dependency — it has found a dead ctx, which ends + // the run. The loop's own check would not catch it on the last + // dependency, and a cancelled Generate must never hand back an + // iterator. + if ctxErr := ctx.Err(); ctxErr != nil { + return nil, ctxErr + } + probability = defaultProbability } probabilityByID[id] = asProbability(probability) } return probabilityByID, nil } -// defaultProbability stands in for a score that is not a probability. It is -// optimistic on purpose: a dependency nobody could estimate is treated as very -// likely to succeed, which keeps its head's preferred path near the front -// rather than burying it or dropping the queue's whole snapshot on one bad -// number. +// defaultProbability stands in for a score that is not a probability, one the +// scorer could not produce at all, and one for a dependency the snapshot never +// carried. It is optimistic on purpose: a dependency nobody could estimate is +// treated as very likely to succeed, which keeps its head's preferred path near +// the front rather than burying it or dropping the queue's whole snapshot on +// one bad number. const defaultProbability = 0.95 // asProbability keeps a usable score and substitutes the default for anything diff --git a/submitqueue/extension/speculation/generator/bestfirst/bestfirst_test.go b/submitqueue/extension/speculation/generator/bestfirst/bestfirst_test.go index 923d03a8a..4e3b7c713 100644 --- a/submitqueue/extension/speculation/generator/bestfirst/bestfirst_test.go +++ b/submitqueue/extension/speculation/generator/bestfirst/bestfirst_test.go @@ -381,15 +381,61 @@ func TestBestFirst_HeadWithNoDependencies(t *testing.T) { assert.InDelta(t, math.Log(1.0), cands[0].RankingScore, 1e-9) } -func TestBestFirst_PropagatesScorerError(t *testing.T) { +// A scorer that cannot price a dependency costs that dependency its estimate, +// nothing more. The queue keeps every candidate it had, ranked as if the +// dependency were very likely to succeed. +func TestBestFirst_AbsorbsScorerError(t *testing.T) { batches := []entity.Batch{ {ID: "q/A", State: entity.BatchStateSpeculating}, {ID: "q/H", State: entity.BatchStateSpeculating, Dependencies: []string{"q/A"}}, } iter, err := New(errScorer{}).Generate(context.Background(), batches) - assert.Error(t, err) - assert.Nil(t, iter) + require.NoError(t, err) + + cands := forHead(drainAll(t, iter), "q/H") + require.Len(t, cands, 2) + assert.Equal(t, entity.DependencyAssumptionSucceeds, assumptionFor(cands[0].Path, "q/A")) + assert.InDelta(t, math.Log(defaultProbability), cands[0].RankingScore, 1e-9) +} + +// A dependency the snapshot never carried is unpriceable, not cheap: the zero +// batch it would resolve to belongs to no queue, so it must never reach the +// scorer. +func TestBestFirst_NeverScoresAnAbsentDependency(t *testing.T) { + batches := []entity.Batch{ + {ID: "q/H", State: entity.BatchStateSpeculating, Dependencies: []string{"q/missing"}}, + } + sc := newCountingScorer(map[string]float64{}) + + iter, err := New(sc).Generate(context.Background(), batches) + require.NoError(t, err) + + cands := drainAll(t, iter) + require.Len(t, cands, 2, "the absent dependency is still an open question with two sides") + assert.Zero(t, sc.total, "the scorer is never handed a batch the snapshot did not carry") + assert.InDelta(t, math.Log(defaultProbability), cands[0].RankingScore, 1e-9) +} + +// A merging dependency is still in progress — the merge can fail — so it stays +// an open question here like any other. Whether a path betting against it is +// worth funding is a matter of price, which is the scorer's to say, not a +// state the search hard-codes. +func TestBestFirst_MergingDependencyStaysOpen(t *testing.T) { + batches := []entity.Batch{ + {ID: "q/landing", State: entity.BatchStateMerging}, + {ID: "q/H", State: entity.BatchStateSpeculating, Dependencies: []string{"q/landing"}}, + } + sc := newCountingScorer(map[string]float64{"q/landing": 0.9}) + + iter, err := New(sc).Generate(context.Background(), batches) + require.NoError(t, err) + cands := drainAll(t, iter) + + assert.Equal(t, 1, sc.calls["q/landing"], "a merging dependency is priced like any other") + require.Len(t, cands, 2, "both sides of a merge that has not landed yet") + assert.Equal(t, entity.DependencyAssumptionSucceeds, assumptionFor(cands[0].Path, "q/landing")) + assert.Equal(t, entity.DependencyAssumptionFails, assumptionFor(cands[1].Path, "q/landing")) } func TestBestFirst_GeneratesOnlyWhatIsPulled(t *testing.T) { @@ -774,6 +820,28 @@ func TestBestFirst_HonorsCancelledContext(t *testing.T) { assert.False(t, ok) assert.Equal(t, entity.CandidatePath{}, c) }) + + t.Run("a scorer that fails on a dead context ends the run", func(t *testing.T) { + // The loop checks ctx before each call, so a context that dies during + // the LAST call is the one it cannot catch — and absorbing that as an + // unpriceable dependency would hand an iterator back to a caller that + // has already gone. + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + iter, err := New(cancellingScorer{cancel: cancel}).Generate(ctx, batches) + require.ErrorIs(t, err, context.Canceled) + assert.Nil(t, iter) + }) +} + +// cancellingScorer kills the context and then fails, the way a scorer whose +// own call was cancelled would. +type cancellingScorer struct{ cancel context.CancelFunc } + +func (s cancellingScorer) Score(context.Context, entity.Batch) (float64, error) { + s.cancel() + return 0, context.Canceled } func TestBestFirst_DefaultsScoreOutsideUnitInterval(t *testing.T) { diff --git a/submitqueue/orchestrator/controller/speculate/run.go b/submitqueue/orchestrator/controller/speculate/run.go index d5cfeb579..044df5455 100644 --- a/submitqueue/orchestrator/controller/speculate/run.go +++ b/submitqueue/orchestrator/controller/speculate/run.go @@ -18,6 +18,8 @@ import ( "context" "errors" "fmt" + "maps" + "slices" "github.com/uber/submitqueue/platform/metrics" corebatch "github.com/uber/submitqueue/submitqueue/core/batch" @@ -240,17 +242,20 @@ func terminalPathStatus(status entity.BuildStatus) entity.SpeculationPathStatus // ask hands the snapshot to the queue's Speculator. Its answer is a proposal, // not an instruction: check decides what is actually enacted. // -// The two arguments are deliberately different slices of the queue. Only -// speculating heads are offered as action targets, because only they are open -// to new work. Every in-flight path set is handed over, though, whatever -// state its head is in: a path holds its CI slot until its build actually -// stops, so a merging head's superseded siblings and a cancelling head's live -// builds spend the budget just like a speculating head's do. Hiding them -// would let the allocator count occupied slots as free and oversubscribe CI. +// Both arguments carry the whole queue, whatever state each batch is in. A +// head's dependencies are the facts its paths are built from, so a dependency +// withheld is one the Speculator has to plan around blind. Every in-flight +// path set goes over for the same reason: a path holds its CI slot until its +// build actually stops, so a merging head's superseded siblings and a +// cancelling head's live builds spend the budget just like a speculating +// head's do. Hiding either would let the allocator count occupied slots as +// free and oversubscribe CI. // -// Passing foreign sets cannot widen what gets proposed: a path ID hashes its +// Passing the full queue cannot widen what gets proposed: a path ID hashes its // head, and check rejects any proposal aimed at a head that is not -// speculating. +// speculating. A head this run has just decided still reads as Speculating +// here, but it can only rebuild the path that already passed — see +// mergeablePath — which the allocator skips as finished. func (c *Controller) ask(ctx context.Context, queue string, snap snapshot) ([]entity.Speculation, error) { spec, err := c.speculators.For(speculator.Config{QueueName: queue}) if err != nil { @@ -265,7 +270,7 @@ func (c *Controller) ask(ctx context.Context, queue string, snap snapshot) ([]en } } - proposals, err := spec.Speculate(ctx, snap.speculating, sets) + proposals, err := spec.Speculate(ctx, slices.Collect(maps.Values(snap.batches)), sets) if err != nil { metrics.NamedCounter(c.metricsScope, opName, "speculator_errors", 1) return nil, fmt.Errorf("speculator failed for queue %s: %w", queue, err) diff --git a/submitqueue/orchestrator/controller/speculate/run_test.go b/submitqueue/orchestrator/controller/speculate/run_test.go index 8e27f4fc2..172f27885 100644 --- a/submitqueue/orchestrator/controller/speculate/run_test.go +++ b/submitqueue/orchestrator/controller/speculate/run_test.go @@ -94,7 +94,7 @@ func (h *runHarness) failPublishTo(topic string) { h.failTopic = topic } -// speculatedOver returns the IDs of the heads the Speculator was offered. +// speculatedOver returns the IDs of the batches the Speculator was offered. func (h *runHarness) speculatedOver() []string { ids := make([]string, 0, len(h.spec.gotBatches)) for _, b := range h.spec.gotBatches { @@ -280,8 +280,8 @@ func TestRun_PassesSnapshotToSpeculator(t *testing.T) { require.NoError(t, h.run(head)) require.Equal(t, 1, spec.calls) - require.Len(t, spec.gotBatches, 1) - assert.Equal(t, head, spec.gotBatches[0].ID, "only speculating heads are action targets") + assert.ElementsMatch(t, []string{dep1, dep2, head, merging.ID}, h.speculatedOver(), + "every batch the run read, in no particular order") require.Len(t, spec.gotSets, 1) assert.Equal(t, int32(3), spec.gotSets[0].Version) } @@ -962,8 +962,8 @@ func TestRun_SpeculatorSeesPathSetsOfNonOpenHeads(t *testing.T) { require.NoError(t, h.run(head)) - assert.Equal(t, []entity.Batch{open}, spec.gotBatches, - "only an open head may be an action target") + assert.ElementsMatch(t, []entity.Batch{open, merging}, spec.gotBatches, + "a closed head is still a fact the open ones are planned against") require.Len(t, spec.gotSets, 2, "every in-flight path set counts against the budget") assert.Equal(t, merging.ID, spec.gotSets[1].Head) } diff --git a/submitqueue/orchestrator/controller/speculate/speculate_test.go b/submitqueue/orchestrator/controller/speculate/speculate_test.go index d5b34a6a7..1d46b0ae8 100644 --- a/submitqueue/orchestrator/controller/speculate/speculate_test.go +++ b/submitqueue/orchestrator/controller/speculate/speculate_test.go @@ -35,14 +35,14 @@ import ( ) // quietSpeculator proposes nothing, which is what tests of the message-level -// branches want: the run happens but changes no paths. It records the heads it -// was offered, so a test can assert that a run reached them at all. +// branches want: the run happens but changes no paths. It records the batches +// it was offered, so a test can assert that a run reached them at all. type quietSpeculator struct { - heads []entity.Batch + saw []entity.Batch } func (s *quietSpeculator) Speculate(_ context.Context, batches []entity.Batch, _ []entity.SpeculationPathSet) ([]entity.Speculation, error) { - s.heads = append(s.heads, batches...) + s.saw = append(s.saw, batches...) return nil, nil } @@ -239,8 +239,8 @@ func TestProcess_TerminalReplansQueue(t *testing.T) { Return(entity.SpeculationPathSet{}, storage.ErrNotFound) require.NoError(t, h.process(t, ctrl, batch.ID)) - assert.Equal(t, []entity.Batch{dependent}, h.spec.heads, - "the dependent must be re-planned against the terminal outcome") + assert.ElementsMatch(t, []entity.Batch{batch, dependent}, h.spec.saw, + "the dependent must be re-planned against the terminal outcome, which it can only be weighed against if the terminal batch comes too") } // A Merging batch is the merge stage's to finish; the run still happens for the From a995972638297b809053462cfe449bb9be58abf5 Mon Sep 17 00:00:00 2001 From: Preetam Dwivedi Date: Mon, 10 Aug 2026 14:52:24 -0700 Subject: [PATCH 2/2] feat(messagequeue)!: carry a structured failure across the dead-letter boundary MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## 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 --- platform/base/failure/BUILD.bazel | 18 +++ platform/base/failure/failure.go | 112 +++++++++++++ platform/base/failure/failure_test.go | 127 +++++++++++++++ platform/consumer/BUILD.bazel | 2 + platform/consumer/consumer.go | 14 +- platform/consumer/consumer_test.go | 92 ++++++++++- platform/consumer/controller.go | 15 ++ platform/consumer/mock/BUILD.bazel | 1 + platform/consumer/mock/controller_mock.go | 16 ++ platform/errs/BUILD.bazel | 4 + platform/errs/failure.go | 120 ++++++++++++++ platform/errs/failure_test.go | 151 ++++++++++++++++++ platform/extension/messagequeue/BUILD.bazel | 5 +- platform/extension/messagequeue/delivery.go | 19 ++- .../extension/messagequeue/mock/BUILD.bazel | 1 + .../messagequeue/mock/delivery_mock.go | 32 +++- .../extension/messagequeue/mysql/BUILD.bazel | 2 + .../messagequeue/mysql/message_store.go | 40 ++++- .../messagequeue/mysql/message_store_test.go | 43 ++++- .../messagequeue/mysql/mock_stores.go | 9 +- .../mysql/schema/queue_messages.sql | 10 ++ .../extension/messagequeue/mysql/stores.go | 9 +- .../messagequeue/mysql/subscriber.go | 114 +++++++++++-- .../messagequeue/mysql/subscriber_test.go | 125 ++++++++++++--- .../extension/messagequeue/mysql/BUILD.bazel | 1 + .../messagequeue/mysql/queue_test.go | 29 +++- 26 files changed, 1030 insertions(+), 81 deletions(-) create mode 100644 platform/base/failure/BUILD.bazel create mode 100644 platform/base/failure/failure.go create mode 100644 platform/base/failure/failure_test.go create mode 100644 platform/errs/failure.go create mode 100644 platform/errs/failure_test.go diff --git a/platform/base/failure/BUILD.bazel b/platform/base/failure/BUILD.bazel new file mode 100644 index 000000000..8407355ca --- /dev/null +++ b/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", + ], +) diff --git a/platform/base/failure/failure.go b/platform/base/failure/failure.go new file mode 100644 index 000000000..e23bc76c6 --- /dev/null +++ b/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 +} diff --git a/platform/base/failure/failure_test.go b/platform/base/failure/failure_test.go new file mode 100644 index 000000000..1fce929e5 --- /dev/null +++ b/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 +} diff --git a/platform/consumer/BUILD.bazel b/platform/consumer/BUILD.bazel index 9214f4713..b8951219e 100644 --- a/platform/consumer/BUILD.bazel +++ b/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", diff --git a/platform/consumer/consumer.go b/platform/consumer/consumer.go index 80921215d..f3dcdbb4d 100644 --- a/platform/consumer/consumer.go +++ b/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", diff --git a/platform/consumer/consumer_test.go b/platform/consumer/consumer_test.go index 0e6e5cc15..6e2eb5d63 100644 --- a/platform/consumer/consumer_test.go +++ b/platform/consumer/consumer_test.go @@ -26,6 +26,7 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "github.com/uber-go/tally" + "github.com/uber/submitqueue/platform/base/failure" entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/errs" "github.com/uber/submitqueue/platform/extension/consumergate" @@ -115,11 +116,11 @@ func setupDelivery(del *queuemock.MockDelivery, msg entityqueue.Message, ackErr, close(done) return ackErr }).MaxTimes(1) - del.EXPECT().Nack(gomock.Any()).DoAndReturn(func(ctx context.Context) error { + del.EXPECT().Nack(gomock.Any(), gomock.Any()).DoAndReturn(func(ctx context.Context, _ failure.Failure) error { close(done) return nackErr }).MaxTimes(1) - del.EXPECT().Reject(gomock.Any(), gomock.Any()).DoAndReturn(func(ctx context.Context, reason string) error { + del.EXPECT().Reject(gomock.Any(), gomock.Any()).DoAndReturn(func(ctx context.Context, _ failure.Failure) error { close(done) return nil }).MaxTimes(1) @@ -458,7 +459,7 @@ func TestConsumer_ProcessDelivery_Hold(t *testing.T) { return tt.postponeErr }) case "nack": - mockDel.EXPECT().Nack(gomock.Any()).DoAndReturn(func(ctx context.Context) error { + mockDel.EXPECT().Nack(gomock.Any(), gomock.Any()).DoAndReturn(func(ctx context.Context, _ failure.Failure) error { close(done) return nil }) @@ -515,7 +516,7 @@ func TestConsumer_ProcessDelivery_NonRetryableError(t *testing.T) { mockDel.EXPECT().ReceivedAt().Return(time.Now().UnixMilli()).AnyTimes() mockDel.EXPECT().Metadata().Return(nil).AnyTimes() mockDel.EXPECT().DeliveryID().Return(msg.ID).AnyTimes() - mockDel.EXPECT().Reject(gomock.Any(), gomock.Any()).DoAndReturn(func(ctx context.Context, reason string) error { + mockDel.EXPECT().Reject(gomock.Any(), gomock.Any()).DoAndReturn(func(ctx context.Context, _ failure.Failure) error { close(done) return nil }).Times(1) @@ -527,6 +528,87 @@ func TestConsumer_ProcessDelivery_NonRetryableError(t *testing.T) { require.NoError(t, err) } +// The failure handed to the queue is built from whatever the controller +// attributed, and a controller that attributes nothing must still produce +// exactly what callers sent before failures carried structure: the error text +// and nothing else. +func TestConsumer_ProcessDelivery_FailureFromControllerError(t *testing.T) { + tests := []struct { + name string + controllerFn func(ctx context.Context, delivery Delivery) error + wantMessage string + wantSubjects []failure.Subject + }{ + { + name: "unattributed error carries only its message", + controllerFn: func(ctx context.Context, delivery Delivery) error { + return fmt.Errorf("bad payload") + }, + wantMessage: "bad payload", + }, + { + name: "attributed error carries its subjects", + controllerFn: func(ctx context.Context, delivery Delivery) error { + return errs.Attribute( + fmt.Errorf("speculator failed"), + failure.Subject{Type: "queue", ID: "test-queue"}, + ) + }, + wantMessage: "speculator failed", + wantSubjects: []failure.Subject{{Type: "queue", ID: "test-queue"}}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + logger := zaptest.NewLogger(t).Sugar() + + deliveryChan := make(chan extqueue.Delivery, 1) + mockSub := queuemock.NewMockSubscriber(ctrl) + mockSub.EXPECT().Subscribe(gomock.Any(), gomock.Any(), gomock.Any()).Return(deliveryChan, nil) + + mockQ := queuemock.NewMockQueue(ctrl) + mockQ.EXPECT().Subscriber().Return(mockSub) + + reg := newRegistry(t, mockQ, testTopicKeyStart, "test-group") + c := New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor(), consumergatenoop.New()) + + handler := &testController{} + setupController(handler, "test-handler", testTopicKeyStart, "test-group", tt.controllerFn) + require.NoError(t, c.Register(handler)) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + require.NoError(t, c.Start(ctx)) + + msg := entityqueue.NewMessage("msg-1", []byte("bad"), "partition1", nil) + done := make(chan struct{}) + var got failure.Failure + + mockDel := queuemock.NewMockDelivery(ctrl) + mockDel.EXPECT().Message().Return(msg).AnyTimes() + mockDel.EXPECT().Attempt().Return(1).AnyTimes() + mockDel.EXPECT().ReceivedAt().Return(time.Now().UnixMilli()).AnyTimes() + mockDel.EXPECT().Metadata().Return(nil).AnyTimes() + mockDel.EXPECT().DeliveryID().Return(msg.ID).AnyTimes() + mockDel.EXPECT().Reject(gomock.Any(), gomock.Any()).DoAndReturn(func(ctx context.Context, f failure.Failure) error { + got = f + close(done) + return nil + }).Times(1) + + deliveryChan <- mockDel + <-done + + assert.Equal(t, tt.wantMessage, got.Message) + assert.Equal(t, tt.wantSubjects, got.Subjects) + + require.NoError(t, c.Stop(30000)) + }) + } +} + func TestConsumer_Stop(t *testing.T) { ctrl := gomock.NewController(t) logger := zaptest.NewLogger(t).Sugar() @@ -922,7 +1004,7 @@ func TestConsumer_PerPartitionProcessing(t *testing.T) { mockDelA.EXPECT().Metadata().Return(nil).AnyTimes() mockDelA.EXPECT().DeliveryID().Return(msgA.ID).AnyTimes() mockDelA.EXPECT().Ack(gomock.Any()).Return(nil).MaxTimes(1) - mockDelA.EXPECT().Nack(gomock.Any()).Return(nil).MaxTimes(1) + mockDelA.EXPECT().Nack(gomock.Any(), gomock.Any()).Return(nil).MaxTimes(1) deliveryChan <- mockDelA diff --git a/platform/consumer/controller.go b/platform/consumer/controller.go index ec335b8a8..fef8c316b 100644 --- a/platform/consumer/controller.go +++ b/platform/consumer/controller.go @@ -19,6 +19,7 @@ package consumer import ( "context" + "github.com/uber/submitqueue/platform/base/failure" entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" extqueue "github.com/uber/submitqueue/platform/extension/messagequeue" ) @@ -64,6 +65,16 @@ type Delivery interface { // Metadata returns backend-specific delivery metadata. Metadata() map[string]string + + // Failure returns why this message was dead-lettered, and whether it was + // dead-lettered at all. Only a controller subscribed to a DLQ topic sees + // true; for anything else there is no failure to report. + // + // Where the message names one entity but the failure was about another — + // a stage whose work spans more of the queue than the message does — the + // subjects here are what says so. A reconciler that acts on the message's + // own entity regardless will terminate the wrong one. + Failure() (failure.Failure, bool) } // deliveryWrapper wraps extension/entityqueue.Delivery and exposes only the safe subset of methods. @@ -109,6 +120,10 @@ func (d *deliveryWrapper) Metadata() map[string]string { return d.delivery.Metadata() } +func (d *deliveryWrapper) Failure() (failure.Failure, bool) { + return d.delivery.Failure() +} + // Controller processes queue deliveries. Controllers contain business logic and are registered with the Consumer. // The Controller interface enables clean separation of concerns: // - Controller focuses on business logic (deserialize, process, return error status) diff --git a/platform/consumer/mock/BUILD.bazel b/platform/consumer/mock/BUILD.bazel index 260fd6baa..da7c1afcd 100644 --- a/platform/consumer/mock/BUILD.bazel +++ b/platform/consumer/mock/BUILD.bazel @@ -6,6 +6,7 @@ go_library( importpath = "github.com/uber/submitqueue/platform/consumer/mock", visibility = ["//visibility:public"], deps = [ + "//platform/base/failure:go_default_library", "//platform/base/messagequeue:go_default_library", "//platform/consumer:go_default_library", "@org_uber_go_mock//gomock:go_default_library", diff --git a/platform/consumer/mock/controller_mock.go b/platform/consumer/mock/controller_mock.go index 29aae4949..dcb6f91d1 100644 --- a/platform/consumer/mock/controller_mock.go +++ b/platform/consumer/mock/controller_mock.go @@ -13,6 +13,7 @@ import ( context "context" reflect "reflect" + failure "github.com/uber/submitqueue/platform/base/failure" messagequeue "github.com/uber/submitqueue/platform/base/messagequeue" consumer "github.com/uber/submitqueue/platform/consumer" gomock "go.uber.org/mock/gomock" @@ -84,6 +85,21 @@ func (mr *MockDeliveryMockRecorder) ExtendVisibilityTimeout(ctx, durationMillis return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ExtendVisibilityTimeout", reflect.TypeOf((*MockDelivery)(nil).ExtendVisibilityTimeout), ctx, durationMillis) } +// Failure mocks base method. +func (m *MockDelivery) Failure() (failure.Failure, bool) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Failure") + ret0, _ := ret[0].(failure.Failure) + ret1, _ := ret[1].(bool) + return ret0, ret1 +} + +// Failure indicates an expected call of Failure. +func (mr *MockDeliveryMockRecorder) Failure() *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Failure", reflect.TypeOf((*MockDelivery)(nil).Failure)) +} + // Hold mocks base method. func (m *MockDelivery) Hold(delayMs int64) { m.ctrl.T.Helper() diff --git a/platform/errs/BUILD.bazel b/platform/errs/BUILD.bazel index 06ab487bf..c7de244e9 100644 --- a/platform/errs/BUILD.bazel +++ b/platform/errs/BUILD.bazel @@ -4,20 +4,24 @@ go_library( name = "go_default_library", srcs = [ "errs.go", + "failure.go", "processor.go", ], importpath = "github.com/uber/submitqueue/platform/errs", visibility = ["//visibility:public"], + deps = ["//platform/base/failure:go_default_library"], ) go_test( name = "go_default_test", srcs = [ "errs_test.go", + "failure_test.go", "processor_test.go", ], embed = [":go_default_library"], deps = [ + "//platform/base/failure:go_default_library", "@com_github_stretchr_testify//assert:go_default_library", "@com_github_stretchr_testify//require:go_default_library", ], diff --git a/platform/errs/failure.go b/platform/errs/failure.go new file mode 100644 index 000000000..bebf60d5f --- /dev/null +++ b/platform/errs/failure.go @@ -0,0 +1,120 @@ +// 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 errs + +import ( + "errors" + + "github.com/uber/submitqueue/platform/base/failure" +) + +// attributedError carries attribution — what a failure is about — alongside an +// error, without saying anything about its severity or retryability. Those +// remain the classifiers' business, and this wrapper is deliberately invisible +// to them: it is neither a userError nor an infraError, so the classifier +// processor walks straight through it to the cause underneath. +type attributedError struct { + // cause is the underlying error. + cause error + // subjects are the entities this failure is about. + subjects []failure.Subject + // detail is free-form structured context. + detail map[string]any +} + +// Attribute returns err carrying the entities it is about, so a consumer +// downstream can act on the right thing instead of guessing from the message. +// +// Use it where the code knows something the error value cannot express — most +// often that a failure inside a job covering many records is about one +// particular record, or about none of them individually. It does not change +// whether the error is retryable; a classifier still decides that from the +// cause. +// +// A nil error is returned unchanged, so it is safe to apply unconditionally. +func Attribute(err error, subjects ...failure.Subject) error { + if err == nil { + return nil + } + return &attributedError{cause: err, subjects: subjects} +} + +// Detail returns err carrying free-form structured context. It composes with +// Attribute in either order; see Attribution for how several layers combine. +// +// A nil error is returned unchanged. +func Detail(err error, detail map[string]any) error { + if err == nil { + return nil + } + return &attributedError{cause: err, detail: detail} +} + +// Error returns the error message. +func (e *attributedError) Error() string { + return e.cause.Error() +} + +// Unwrap returns the underlying cause for errors.Is/As compatibility. +func (e *attributedError) Unwrap() error { + return e.cause +} + +// Attribution reads back everything Attribute and Detail put on err, as the +// failure a consumer should record. +// +// Message is always the error's own text, so the result is usable whether or +// not anything was attributed; an unattributed error simply yields a failure +// with no subjects and no detail. Where several layers of the chain carry +// attribution, subjects accumulate outermost-first and deduplicate, and detail +// keys are resolved outermost-wins — the outer layer is the later writer and +// has the wider view. +// +// A nil error yields the zero Failure. +func Attribution(err error) failure.Failure { + if err == nil { + return failure.Failure{} + } + + result := failure.Failure{Message: err.Error()} + seen := make(map[failure.Subject]bool) + + for node := err; node != nil; node = errors.Unwrap(node) { + attributed, ok := node.(*attributedError) + if !ok { + continue + } + + for _, s := range attributed.subjects { + if seen[s] { + continue + } + seen[s] = true + result.Subjects = append(result.Subjects, s) + } + + for k, v := range attributed.detail { + if _, taken := result.Detail[k]; taken { + continue + } + if result.Detail == nil { + result.Detail = make(map[string]any) + } + result.Detail[k] = v + } + } + + return result +} diff --git a/platform/errs/failure_test.go b/platform/errs/failure_test.go new file mode 100644 index 000000000..f4d39ae1f --- /dev/null +++ b/platform/errs/failure_test.go @@ -0,0 +1,151 @@ +// 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 errs + +import ( + "errors" + "fmt" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/uber/submitqueue/platform/base/failure" +) + +// driverError stands in for a backend's own error type — the thing a real +// classifier (e.g. mysqlerrs) recognises at the bottom of a chain. Defined +// here rather than importing a real classifier, which would be an import +// cycle: every classifier package imports errs. +type driverError struct{ code int } + +func (e driverError) Error() string { return fmt.Sprintf("driver error %d", e.code) } + +// driverClassifier recognises only driverError, so a verdict proves the +// processor actually reached that node rather than stopping earlier. +type driverClassifier struct{} + +func (driverClassifier) Classify(err error) Verdict { + var de driverError + if errors.As(err, &de) { + return InfraRetryable + } + return Unknown +} + +func TestAttributionReadsBackSubjects(t *testing.T) { + err := Attribute(errors.New("boom"), failure.Subject{Type: "batch", ID: "q/batch/1"}) + + got := Attribution(err) + assert.Equal(t, "boom", got.Message) + assert.Equal(t, []failure.Subject{{Type: "batch", ID: "q/batch/1"}}, got.Subjects) +} + +// An unattributed error still yields a usable failure — the message alone. +// That is what lets a caller build one unconditionally. +func TestAttributionOfPlainError(t *testing.T) { + got := Attribution(errors.New("boom")) + + assert.Equal(t, "boom", got.Message) + assert.Empty(t, got.Subjects) + assert.Empty(t, got.Detail) +} + +func TestAttributionOfNil(t *testing.T) { + assert.Equal(t, failure.Failure{}, Attribution(nil)) +} + +func TestAttributeNilIsNil(t *testing.T) { + assert.NoError(t, Attribute(nil, failure.Subject{Type: "batch", ID: "x"})) + assert.NoError(t, Detail(nil, map[string]any{"k": "v"})) +} + +// Attribution survives further wrapping, which is the normal shape: a leaf +// attributes the entity it knows about, outer frames add context with %w. +func TestAttributionThroughOuterWrap(t *testing.T) { + inner := Attribute(errors.New("boom"), failure.Subject{Type: "batch", ID: "q/batch/1"}) + outer := fmt.Errorf("run failed: %w", inner) + + got := Attribution(outer) + assert.Equal(t, "run failed: boom", got.Message) + assert.Equal(t, []failure.Subject{{Type: "batch", ID: "q/batch/1"}}, got.Subjects) +} + +func TestAttributionMergesLayers(t *testing.T) { + err := Attribute(errors.New("boom"), failure.Subject{Type: "batch", ID: "inner"}) + err = Detail(err, map[string]any{"stage": "dispatch", "shared": "inner"}) + err = Attribute(err, failure.Subject{Type: "queue", ID: "q"}, failure.Subject{Type: "batch", ID: "inner"}) + err = Detail(err, map[string]any{"shared": "outer"}) + + got := Attribution(err) + + // Outermost first, and the repeated subject appears once. + assert.Equal(t, []failure.Subject{ + {Type: "queue", ID: "q"}, + {Type: "batch", ID: "inner"}, + }, got.Subjects) + // Outermost wins on a key collision. + assert.Equal(t, "outer", got.Detail["shared"]) + assert.Equal(t, "dispatch", got.Detail["stage"]) +} + +// The regression that matters: attribution must be invisible to classification. +// If the wrapper broke the chain walk, every storage error would silently stop +// being retryable. +func TestAttributionDoesNotHideTheCauseFromClassifiers(t *testing.T) { + cause := driverError{code: 1213} + attributed := Attribute(fmt.Errorf("write failed: %w", cause), failure.Subject{Type: "batch", ID: "q/batch/1"}) + + var de driverError + require.True(t, errors.As(attributed, &de), "errors.As must reach the cause through the wrapper") + assert.Equal(t, 1213, de.code) + + processed := NewClassifierProcessor(driverClassifier{}).Process(attributed) + assert.True(t, IsRetryable(processed), "classification must survive attribution") + + // And the attribution survives classification wrapping it in turn. + assert.Equal(t, []failure.Subject{{Type: "batch", ID: "q/batch/1"}}, Attribution(processed).Subjects) +} + +// Attribution carries no verdict of its own, so an attributed error with +// nothing else on it stays non-retryable by default. +func TestAttributionAloneIsNotRetryable(t *testing.T) { + err := Attribute(errors.New("boom"), failure.Subject{Type: "queue", ID: "q"}) + + assert.False(t, IsRetryable(err)) + assert.False(t, IsUserError(err)) +} + +func TestAttributionComposesWithFrameworkWraps(t *testing.T) { + tests := []struct { + name string + wrap func(error) error + retryable bool + userError bool + }{ + {"retryable", NewRetryableError, true, false}, + {"user", NewUserError, false, true}, + {"dependency retryable", NewRetryableDependencyError, true, false}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + err := tt.wrap(Attribute(errors.New("boom"), failure.Subject{Type: "batch", ID: "b"})) + + assert.Equal(t, tt.retryable, IsRetryable(err)) + assert.Equal(t, tt.userError, IsUserError(err)) + assert.Equal(t, []failure.Subject{{Type: "batch", ID: "b"}}, Attribution(err).Subjects) + }) + } +} diff --git a/platform/extension/messagequeue/BUILD.bazel b/platform/extension/messagequeue/BUILD.bazel index efaa2c51b..c98112f7b 100644 --- a/platform/extension/messagequeue/BUILD.bazel +++ b/platform/extension/messagequeue/BUILD.bazel @@ -11,7 +11,10 @@ go_library( ], importpath = "github.com/uber/submitqueue/platform/extension/messagequeue", visibility = ["//visibility:public"], - deps = ["//platform/base/messagequeue:go_default_library"], + deps = [ + "//platform/base/failure:go_default_library", + "//platform/base/messagequeue:go_default_library", + ], ) go_test( diff --git a/platform/extension/messagequeue/delivery.go b/platform/extension/messagequeue/delivery.go index 9b56a56af..941c6c040 100644 --- a/platform/extension/messagequeue/delivery.go +++ b/platform/extension/messagequeue/delivery.go @@ -19,6 +19,7 @@ package messagequeue import ( "context" + "github.com/uber/submitqueue/platform/base/failure" entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" ) @@ -39,7 +40,12 @@ type Delivery interface { // The message is requeued for redelivery immediately; the visibility // timeout is what spaces retries (a crash or missed ack redelivers on the // same schedule). The redelivery counts toward the failure budget. - Nack(ctx context.Context) error + // + // f describes the failure. It is carried so that the redelivery which + // finally exhausts the budget can dead-letter with the reason that caused + // it, rather than with a generic one — a nack whose reason is dropped + // leaves the eventual dead letter unable to say what went wrong. + Nack(ctx context.Context, f failure.Failure) error // Postpone finishes this delivery as "processed successfully, redeliver // later": the message becomes invisible for delayMs and acts as a barrier — @@ -51,9 +57,10 @@ type Delivery interface { // Reject moves the message to the dead letter entityqueue. // Use for poison pill messages that should never be retried. - // reason is stored as last_error in the DLQ for debugging. + // f is recorded with the dead-lettered message for diagnosis and is what + // Failure returns when it is redelivered from the DLQ. // If DLQ is not configured, the message is acked (removed from queue). - Reject(ctx context.Context, reason string) error + Reject(ctx context.Context, f failure.Failure) error // ExtendVisibilityTimeout extends the time before this message becomes // visible to other consumers. Use when processing takes longer than expected. @@ -71,4 +78,10 @@ type Delivery interface { // Metadata returns backend-specific delivery metadata. Metadata() map[string]string + + // Failure returns why this message was dead-lettered, and whether it was + // dead-lettered at all. It reports false for a message delivered from its + // original topic, so a DLQ consumer can distinguish "no failure recorded" + // from a failure that recorded nothing. + Failure() (failure.Failure, bool) } diff --git a/platform/extension/messagequeue/mock/BUILD.bazel b/platform/extension/messagequeue/mock/BUILD.bazel index 099e1b485..0516a850c 100644 --- a/platform/extension/messagequeue/mock/BUILD.bazel +++ b/platform/extension/messagequeue/mock/BUILD.bazel @@ -11,6 +11,7 @@ go_library( importpath = "github.com/uber/submitqueue/platform/extension/messagequeue/mock", visibility = ["//visibility:public"], deps = [ + "//platform/base/failure:go_default_library", "//platform/base/messagequeue:go_default_library", "//platform/extension/messagequeue:go_default_library", "@org_uber_go_mock//gomock:go_default_library", diff --git a/platform/extension/messagequeue/mock/delivery_mock.go b/platform/extension/messagequeue/mock/delivery_mock.go index e591a747f..41e77a430 100644 --- a/platform/extension/messagequeue/mock/delivery_mock.go +++ b/platform/extension/messagequeue/mock/delivery_mock.go @@ -13,6 +13,7 @@ import ( context "context" reflect "reflect" + failure "github.com/uber/submitqueue/platform/base/failure" messagequeue "github.com/uber/submitqueue/platform/base/messagequeue" gomock "go.uber.org/mock/gomock" ) @@ -97,6 +98,21 @@ func (mr *MockDeliveryMockRecorder) ExtendVisibilityTimeout(ctx, durationMillis return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ExtendVisibilityTimeout", reflect.TypeOf((*MockDelivery)(nil).ExtendVisibilityTimeout), ctx, durationMillis) } +// Failure mocks base method. +func (m *MockDelivery) Failure() (failure.Failure, bool) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Failure") + ret0, _ := ret[0].(failure.Failure) + ret1, _ := ret[1].(bool) + return ret0, ret1 +} + +// Failure indicates an expected call of Failure. +func (mr *MockDeliveryMockRecorder) Failure() *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Failure", reflect.TypeOf((*MockDelivery)(nil).Failure)) +} + // Message mocks base method. func (m *MockDelivery) Message() messagequeue.Message { m.ctrl.T.Helper() @@ -126,17 +142,17 @@ func (mr *MockDeliveryMockRecorder) Metadata() *gomock.Call { } // Nack mocks base method. -func (m *MockDelivery) Nack(ctx context.Context) error { +func (m *MockDelivery) Nack(ctx context.Context, f failure.Failure) error { m.ctrl.T.Helper() - ret := m.ctrl.Call(m, "Nack", ctx) + ret := m.ctrl.Call(m, "Nack", ctx, f) ret0, _ := ret[0].(error) return ret0 } // Nack indicates an expected call of Nack. -func (mr *MockDeliveryMockRecorder) Nack(ctx any) *gomock.Call { +func (mr *MockDeliveryMockRecorder) Nack(ctx, f any) *gomock.Call { mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Nack", reflect.TypeOf((*MockDelivery)(nil).Nack), ctx) + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Nack", reflect.TypeOf((*MockDelivery)(nil).Nack), ctx, f) } // Postpone mocks base method. @@ -168,15 +184,15 @@ func (mr *MockDeliveryMockRecorder) ReceivedAt() *gomock.Call { } // Reject mocks base method. -func (m *MockDelivery) Reject(ctx context.Context, reason string) error { +func (m *MockDelivery) Reject(ctx context.Context, f failure.Failure) error { m.ctrl.T.Helper() - ret := m.ctrl.Call(m, "Reject", ctx, reason) + ret := m.ctrl.Call(m, "Reject", ctx, f) ret0, _ := ret[0].(error) return ret0 } // Reject indicates an expected call of Reject. -func (mr *MockDeliveryMockRecorder) Reject(ctx, reason any) *gomock.Call { +func (mr *MockDeliveryMockRecorder) Reject(ctx, f any) *gomock.Call { mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Reject", reflect.TypeOf((*MockDelivery)(nil).Reject), ctx, reason) + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Reject", reflect.TypeOf((*MockDelivery)(nil).Reject), ctx, f) } diff --git a/platform/extension/messagequeue/mysql/BUILD.bazel b/platform/extension/messagequeue/mysql/BUILD.bazel index 762ac7da3..1a0a7f457 100644 --- a/platform/extension/messagequeue/mysql/BUILD.bazel +++ b/platform/extension/messagequeue/mysql/BUILD.bazel @@ -19,6 +19,7 @@ go_library( importpath = "github.com/uber/submitqueue/platform/extension/messagequeue/mysql", visibility = ["//visibility:public"], deps = [ + "//platform/base/failure:go_default_library", "//platform/base/messagequeue:go_default_library", "//platform/extension/messagequeue:go_default_library", "//platform/metrics:go_default_library", @@ -42,6 +43,7 @@ go_test( ], embed = [":go_default_library"], deps = [ + "//platform/base/failure:go_default_library", "//platform/base/messagequeue:go_default_library", "//platform/extension/messagequeue:go_default_library", "@com_github_data_dog_go_sqlmock//:go_default_library", diff --git a/platform/extension/messagequeue/mysql/message_store.go b/platform/extension/messagequeue/mysql/message_store.go index 3667b16c9..78f9c9bf5 100644 --- a/platform/extension/messagequeue/mysql/message_store.go +++ b/platform/extension/messagequeue/mysql/message_store.go @@ -24,6 +24,7 @@ import ( "github.com/uber-go/tally" "go.uber.org/zap" + "github.com/uber/submitqueue/platform/base/failure" entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/metrics" ) @@ -74,9 +75,13 @@ func (s *sqlmessageStore) Insert(ctx context.Context, topic string, messages []e // ON DUPLICATE KEY UPDATE topic=topic is a no-op write that makes MySQL // swallow the unique-key violation without mutating the existing row. + // + // The DLQ columns take their "normal message" values here. failure_detail + // is NULL rather than an empty sentinel: it is a JSON column, which rejects + // ''. stmt, err := tx.PrepareContext(ctx, fmt.Sprintf(` - INSERT INTO %s (topic, id, payload, metadata, partition_key, created_at, published_at, failed_at, failure_count, last_error, original_topic) - VALUES (?, ?, ?, ?, ?, ?, ?, 0, 0, '', '') + INSERT INTO %s (topic, id, payload, metadata, partition_key, created_at, published_at, failed_at, failure_count, last_error, original_topic, failure_detail) + VALUES (?, ?, ?, ?, ?, ?, ?, 0, 0, '', '', NULL) ON DUPLICATE KEY UPDATE topic = topic `, MessagesTableName)) if err != nil { @@ -143,7 +148,7 @@ func (s *sqlmessageStore) FetchByOffset(ctx context.Context, topic string, parti defer func() { op.Complete(retErr) }() rows, err := s.db.QueryContext(ctx, fmt.Sprintf(` - SELECT offset, id, payload, metadata, partition_key, published_at, failed_at, failure_count, last_error, original_topic + SELECT offset, id, payload, metadata, partition_key, published_at, failed_at, failure_count, last_error, original_topic, failure_detail FROM %s WHERE topic = ? AND partition_key = ? AND offset > ? ORDER BY offset @@ -168,9 +173,10 @@ func (s *sqlmessageStore) FetchByOffset(ctx context.Context, topic string, parti failureCount int lastError string originalTopic string + failureDetail []byte ) - if err := rows.Scan(&offset, &id, &payload, &metadataJSON, &partKey, &publishedAtMilli, &failedAt, &failureCount, &lastError, &originalTopic); err != nil { + if err := rows.Scan(&offset, &id, &payload, &metadataJSON, &partKey, &publishedAtMilli, &failedAt, &failureCount, &lastError, &originalTopic, &failureDetail); err != nil { return nil, fmt.Errorf("scan row topic=%s partition=%s: %w", topic, partitionKey, err) } @@ -195,6 +201,7 @@ func (s *sqlmessageStore) FetchByOffset(ctx context.Context, topic string, parti FailureCount: failureCount, LastError: lastError, OriginalTopic: originalTopic, + FailureDetail: failureDetail, }) } @@ -214,13 +221,30 @@ func (s *sqlmessageStore) FetchByOffset(ctx context.Context, topic string, parti // MoveToDLQ atomically moves a message to the DLQ by reinserting it with the DLQ topic name // The message is inserted back into queue_messages table with the DLQ topic (original + suffix) // This allows DLQ messages to be consumed using the normal subscriber -func (s *sqlmessageStore) MoveToDLQ(ctx context.Context, topic string, partitionKey string, messageID string, failureCount int, lastError string, dlqTopicSuffix string) (retErr error) { +// +// The failure is split across two columns: its message into last_error, so the +// row stays readable without decoding anything, and its subjects and detail +// into failure_detail. A failure with no structure leaves failure_detail NULL, +// which is what an unattributed dead letter looks like. +func (s *sqlmessageStore) MoveToDLQ(ctx context.Context, topic string, partitionKey string, messageID string, failureCount int, f failure.Failure, dlqTopicSuffix string) (retErr error) { op := metrics.Begin(s.scope, "move_to_dlq", metrics.StorageLatencyBuckets, metrics.NewTag("topic", topic)) defer func() { op.Complete(retErr) }() // Construct DLQ topic name dlqTopic := topic + dlqTopicSuffix + failureDetail, err := failure.Encode(f) + if err != nil { + return fmt.Errorf("encode failure detail topic=%s message=%s: %w", topic, messageID, err) + } + // Bind NULL explicitly when there is no structure. A nil []byte would leave + // the column's value up to the driver, and an empty string is not valid + // JSON, so the insert could be rejected outright. + var failureDetailArg any + if len(failureDetail) > 0 { + failureDetailArg = failureDetail + } + tx, err := s.db.BeginTx(ctx, nil) if err != nil { return fmt.Errorf("begin transaction topic=%s message=%s: %w", topic, messageID, err) @@ -257,9 +281,9 @@ func (s *sqlmessageStore) MoveToDLQ(ctx context.Context, topic string, partition // Insert into queue_messages table with DLQ topic name and DLQ-specific fields. now := time.Now().UnixMilli() _, err = tx.ExecContext(ctx, fmt.Sprintf(` - INSERT INTO %s (topic, id, payload, metadata, partition_key, created_at, published_at, failed_at, failure_count, last_error, original_topic) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) - `, MessagesTableName), dlqTopic, messageID, payload, metadataJSON, fetchPartKey, createdAtMilli, publishedAtMilli, now, failureCount, lastError, topic) + INSERT INTO %s (topic, id, payload, metadata, partition_key, created_at, published_at, failed_at, failure_count, last_error, original_topic, failure_detail) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + `, MessagesTableName), dlqTopic, messageID, payload, metadataJSON, fetchPartKey, createdAtMilli, publishedAtMilli, now, failureCount, f.Message, topic, failureDetailArg) if err != nil { return fmt.Errorf("insert into DLQ topic=%s dlq=%s partition=%s message=%s: %w", topic, dlqTopic, partitionKey, messageID, err) diff --git a/platform/extension/messagequeue/mysql/message_store_test.go b/platform/extension/messagequeue/mysql/message_store_test.go index ca0913a1e..acd83d730 100644 --- a/platform/extension/messagequeue/mysql/message_store_test.go +++ b/platform/extension/messagequeue/mysql/message_store_test.go @@ -27,6 +27,7 @@ import ( "github.com/uber-go/tally" "go.uber.org/zap/zaptest" + "github.com/uber/submitqueue/platform/base/failure" entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" ) @@ -144,8 +145,8 @@ func TestMessageStore_FetchByOffset(t *testing.T) { limit := 10 // Mock query results (no transaction, simple SELECT) - rows := sqlmock.NewRows([]string{"offset", "id", "payload", "metadata", "partition_key", "published_at", "failed_at", "failure_count", "last_error", "original_topic"}). - AddRow(int64(1), "msg1", []byte("payload1"), []byte("{}"), "part1", time.Now().UnixMilli(), int64(0), 0, "", "") + rows := sqlmock.NewRows([]string{"offset", "id", "payload", "metadata", "partition_key", "published_at", "failed_at", "failure_count", "last_error", "original_topic", "failure_detail"}). + AddRow(int64(1), "msg1", []byte("payload1"), []byte("{}"), "part1", time.Now().UnixMilli(), int64(0), 0, "", "", nil) mock.ExpectQuery("SELECT (.+) FROM queue_messages"). WithArgs(topic, partitionKey, currentOffset, limit). @@ -182,9 +183,11 @@ func TestMessageStore_MoveToDLQ(t *testing.T) { WithArgs(topic, partitionKey, messageID). WillReturnRows(rows) - // Expect insert into queue_messages with DLQ topic + // Expect insert into queue_messages with DLQ topic. The failure's message + // goes to last_error; failure_detail is NULL because this failure names no + // subjects — see TestMessageStore_MoveToDLQ_WritesFailureDetail. mock.ExpectExec("INSERT INTO queue_messages"). - WithArgs(dlqTopic, messageID, sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), failureCount, lastError, topic). + WithArgs(dlqTopic, messageID, sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), failureCount, lastError, topic, nil). WillReturnResult(sqlmock.NewResult(1, 1)) // Expect delete from main table (now includes partition_key in WHERE) @@ -195,11 +198,41 @@ func TestMessageStore_MoveToDLQ(t *testing.T) { // Expect commit mock.ExpectCommit() - err := store.MoveToDLQ(ctx, topic, partitionKey, messageID, failureCount, lastError, dlqTopicSuffix) + err := store.MoveToDLQ(ctx, topic, partitionKey, messageID, failureCount, failure.New(lastError), dlqTopicSuffix) require.NoError(t, err) require.NoError(t, mock.ExpectationsWereMet()) } +// The structured half of a failure is what lets a DLQ consumer act on the right +// entity, so it has to reach the row rather than being flattened into prose. +func TestMessageStore_MoveToDLQ_WritesFailureDetail(t *testing.T) { + db, mock, err := sqlmock.New() + require.NoError(t, err) + defer db.Close() + + store := newMessageStore(db, zaptest.NewLogger(t).Sugar(), tally.NoopScope) + + f := failure.New("speculator failed", failure.Subject{Type: "queue", ID: "test-queue"}) + encoded, err := failure.Encode(f) + require.NoError(t, err) + + mock.ExpectBegin() + mock.ExpectQuery("SELECT (.+) FROM queue_messages"). + WithArgs("test_topic", "part1", "msg1"). + WillReturnRows(sqlmock.NewRows([]string{"payload", "metadata", "partition_key", "created_at", "published_at"}). + AddRow([]byte("payload1"), nil, "part1", int64(1), int64(2))) + mock.ExpectExec("INSERT INTO queue_messages"). + WithArgs("test_topic_dlq", "msg1", sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), 3, "speculator failed", "test_topic", encoded). + WillReturnResult(sqlmock.NewResult(1, 1)) + mock.ExpectExec("DELETE FROM queue_messages"). + WithArgs("test_topic", "part1", "msg1"). + WillReturnResult(sqlmock.NewResult(0, 1)) + mock.ExpectCommit() + + require.NoError(t, store.MoveToDLQ(context.Background(), "test_topic", "part1", "msg1", 3, f, "_dlq")) + require.NoError(t, mock.ExpectationsWereMet()) +} + func TestMessageStore_GetOffsetsAbove(t *testing.T) { tests := []struct { name string diff --git a/platform/extension/messagequeue/mysql/mock_stores.go b/platform/extension/messagequeue/mysql/mock_stores.go index a411e53c1..c787beaca 100644 --- a/platform/extension/messagequeue/mysql/mock_stores.go +++ b/platform/extension/messagequeue/mysql/mock_stores.go @@ -13,6 +13,7 @@ import ( context "context" reflect "reflect" + failure "github.com/uber/submitqueue/platform/base/failure" messagequeue "github.com/uber/submitqueue/platform/base/messagequeue" gomock "go.uber.org/mock/gomock" ) @@ -115,17 +116,17 @@ func (mr *MockmessageStoreMockRecorder) Insert(ctx, topic, messages any) *gomock } // MoveToDLQ mocks base method. -func (m *MockmessageStore) MoveToDLQ(ctx context.Context, topic, partitionKey, messageID string, failureCount int, lastError, dlqTopicSuffix string) error { +func (m *MockmessageStore) MoveToDLQ(ctx context.Context, topic, partitionKey, messageID string, failureCount int, f failure.Failure, dlqTopicSuffix string) error { m.ctrl.T.Helper() - ret := m.ctrl.Call(m, "MoveToDLQ", ctx, topic, partitionKey, messageID, failureCount, lastError, dlqTopicSuffix) + ret := m.ctrl.Call(m, "MoveToDLQ", ctx, topic, partitionKey, messageID, failureCount, f, dlqTopicSuffix) ret0, _ := ret[0].(error) return ret0 } // MoveToDLQ indicates an expected call of MoveToDLQ. -func (mr *MockmessageStoreMockRecorder) MoveToDLQ(ctx, topic, partitionKey, messageID, failureCount, lastError, dlqTopicSuffix any) *gomock.Call { +func (mr *MockmessageStoreMockRecorder) MoveToDLQ(ctx, topic, partitionKey, messageID, failureCount, f, dlqTopicSuffix any) *gomock.Call { mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "MoveToDLQ", reflect.TypeOf((*MockmessageStore)(nil).MoveToDLQ), ctx, topic, partitionKey, messageID, failureCount, lastError, dlqTopicSuffix) + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "MoveToDLQ", reflect.TypeOf((*MockmessageStore)(nil).MoveToDLQ), ctx, topic, partitionKey, messageID, failureCount, f, dlqTopicSuffix) } // MockoffsetStore is a mock of offsetStore interface. diff --git a/platform/extension/messagequeue/mysql/schema/queue_messages.sql b/platform/extension/messagequeue/mysql/schema/queue_messages.sql index 50e887b25..a3a655365 100644 --- a/platform/extension/messagequeue/mysql/schema/queue_messages.sql +++ b/platform/extension/messagequeue/mysql/schema/queue_messages.sql @@ -31,6 +31,16 @@ CREATE TABLE IF NOT EXISTS queue_messages ( failure_count INT UNSIGNED NOT NULL, last_error TEXT NOT NULL, original_topic VARCHAR(255) NOT NULL, + -- failure_detail holds the structured half of the failure: which entities it + -- was about, plus free-form context. last_error keeps the human-readable + -- message, so this column never has to be decoded to read one, and a plain + -- SELECT last_error stays useful. + -- + -- NULL rather than an empty-string sentinel like its neighbours: a JSON + -- column rejects '' as invalid, and NULL is the honest reading of a failure + -- that recorded no structure — including every row written before this + -- column existed, and the retry-limit backstop, which has none to record. + failure_detail JSON, -- Supports: SELECT ... WHERE topic=? AND partition_key=? AND offset > ? ORDER BY offset -- Used by subscribers to poll for messages within their assigned partition diff --git a/platform/extension/messagequeue/mysql/stores.go b/platform/extension/messagequeue/mysql/stores.go index d2c43b497..c54fc2465 100644 --- a/platform/extension/messagequeue/mysql/stores.go +++ b/platform/extension/messagequeue/mysql/stores.go @@ -19,6 +19,7 @@ package mysql import ( "context" + "github.com/uber/submitqueue/platform/base/failure" entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" ) @@ -53,6 +54,10 @@ type messageRow struct { LastError string // OriginalTopic is the topic where the message originally failed ("" for normal messages) OriginalTopic string + // FailureDetail is the encoded structured half of the failure — its + // subjects and free-form context. Empty for normal messages, and for a DLQ + // message whose failure recorded no structure. + FailureDetail []byte } // messageStore handles message table operations (internal use only) @@ -69,7 +74,9 @@ type messageStore interface { // MoveToDLQ moves a message to the dead letter queue // dlqTopicSuffix is appended to the original topic to form the DLQ topic name - MoveToDLQ(ctx context.Context, topic string, partitionKey string, messageID string, failureCount int, lastError string, dlqTopicSuffix string) error + // f is split across the row: its message into last_error, its structured + // half into failure_detail. + MoveToDLQ(ctx context.Context, topic string, partitionKey string, messageID string, failureCount int, f failure.Failure, dlqTopicSuffix string) error // GarbageCollect deletes messages with offset <= minAckedOffset. // The caller (subscriber) is responsible for computing minAckedOffset from the diff --git a/platform/extension/messagequeue/mysql/subscriber.go b/platform/extension/messagequeue/mysql/subscriber.go index e0f1cef41..3b02d995d 100644 --- a/platform/extension/messagequeue/mysql/subscriber.go +++ b/platform/extension/messagequeue/mysql/subscriber.go @@ -25,6 +25,7 @@ import ( "github.com/uber-go/tally" "go.uber.org/zap" + "github.com/uber/submitqueue/platform/base/failure" entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" extqueue "github.com/uber/submitqueue/platform/extension/messagequeue" "github.com/uber/submitqueue/platform/metrics" @@ -190,6 +191,18 @@ type sqlDelivery struct { // DLQ configuration for Reject dlqConfig extqueue.DLQConfig + // retry is the subscription's retry budget. Nack needs it to recognise the + // attempt that spends the last of it, which is the one delivery still + // holding the reason the message is about to be dead-lettered for. + retry extqueue.RetryConfig + + // failure is why this message was dead-lettered, reassembled from the row. + // Only meaningful when failed is set, i.e. when this is a redelivery from + // a DLQ topic. + failure failure.Failure + // failed records whether this message arrived from a DLQ topic. + failed bool + // Track acknowledgment state mu sync.Mutex acknowledged bool @@ -207,6 +220,9 @@ func newSQLDelivery( messageID string, consumerGroup string, dlqConfig extqueue.DLQConfig, + retry extqueue.RetryConfig, + f failure.Failure, + failed bool, ) *sqlDelivery { return &sqlDelivery{ msg: msg, @@ -221,6 +237,9 @@ func newSQLDelivery( messageID: messageID, consumerGroup: consumerGroup, dlqConfig: dlqConfig, + retry: retry, + failure: f, + failed: failed, acknowledged: false, } } @@ -271,7 +290,19 @@ func (d *sqlDelivery) Ack(ctx context.Context) error { } // Nack implements extqueue.Delivery.Nack -func (d *sqlDelivery) Nack(ctx context.Context) error { +// +// When this attempt has spent the retry budget, the message is dead-lettered +// here rather than nacked, so it carries f — the reason it actually failed. +// The poll loop would otherwise pick it up on the next round and dead-letter +// it with a generic reason, f having been discarded when this delivery ended. +// That path still exists as a backstop for a delivery that never reaches Nack +// at all (a crash, or a missed ack redelivered by the visibility timeout); it +// is just no longer the common one. +// +// Message count is unchanged: at attempt N the next poll would see retry_count +// = N and dead-letter iff N >= MaxAttempts, which is exactly the condition +// below. +func (d *sqlDelivery) Nack(ctx context.Context, f failure.Failure) error { d.mu.Lock() defer d.mu.Unlock() @@ -279,6 +310,18 @@ func (d *sqlDelivery) Nack(ctx context.Context) error { return &ErrAlreadyAcknowledged{DeliveryID: d.deliveryID} } + if d.retry.MaxAttempts > 0 && d.attempt >= d.retry.MaxAttempts { + d.subscriber.logger.Warnw("message exhausted retry budget, dead-lettering", + "topic", d.topic, + "partition_key", d.partitionKey, + "message_id", d.messageID, + "attempt", d.attempt, + "max_attempts", d.retry.MaxAttempts, + "reason", f.Message, + ) + return d.deadLetter(ctx, f) + } + // Mark as nacked in delivery state (per consumer group): immediately // eligible for redelivery on the next poll. if err := d.subscriber.deliveryStateStore.MarkNacked(ctx, d.consumerGroup, d.topic, d.partitionKey, d.offset); err != nil { @@ -322,7 +365,7 @@ func (d *sqlDelivery) Postpone(ctx context.Context, delayMs int64) error { } // Reject implements extqueue.Delivery.Reject -func (d *sqlDelivery) Reject(ctx context.Context, reason string) error { +func (d *sqlDelivery) Reject(ctx context.Context, f failure.Failure) error { d.mu.Lock() defer d.mu.Unlock() @@ -330,33 +373,39 @@ func (d *sqlDelivery) Reject(ctx context.Context, reason string) error { return &ErrAlreadyAcknowledged{DeliveryID: d.deliveryID} } + return d.deadLetter(ctx, f) +} + +// deadLetter ends this delivery by moving the message to the DLQ with f +// recorded against it, or by simply acking when no DLQ is configured — there +// is nowhere else to put it, and leaving it would redeliver forever. +// +// Callers hold d.mu. +func (d *sqlDelivery) deadLetter(ctx context.Context, f failure.Failure) error { if d.dlqConfig.Enabled { // Move to DLQ if err := d.subscriber.messageStore.MoveToDLQ( - ctx, d.topic, d.partitionKey, d.messageID, d.attempt, reason, d.dlqConfig.TopicSuffix, + ctx, d.topic, d.partitionKey, d.messageID, d.attempt, f, d.dlqConfig.TopicSuffix, ); err != nil { return fmt.Errorf("failed to move message to DLQ: %w", err) } + } - // Mark as acked in delivery state. Watermark advancement is deferred - // to the poll loop, same as Ack. - if err := d.subscriber.deliveryStateStore.MarkAcked(ctx, d.consumerGroup, d.topic, d.partitionKey, d.offset); err != nil { - return fmt.Errorf("mark acked after DLQ move: %w", err) - } - - } else { - // DLQ disabled — mark as acked (remove from processing). - // Watermark advancement is deferred to the poll loop, same as Ack. - if err := d.subscriber.deliveryStateStore.MarkAcked(ctx, d.consumerGroup, d.topic, d.partitionKey, d.offset); err != nil { - return err - } - + // Mark as acked in delivery state. Watermark advancement is deferred + // to the poll loop, same as Ack. + if err := d.subscriber.deliveryStateStore.MarkAcked(ctx, d.consumerGroup, d.topic, d.partitionKey, d.offset); err != nil { + return fmt.Errorf("mark acked after DLQ move: %w", err) } d.acknowledged = true return nil } +// Failure implements extqueue.Delivery.Failure +func (d *sqlDelivery) Failure() (failure.Failure, bool) { + return d.failure, d.failed +} + // ExtendVisibilityTimeout implements extqueue.Delivery.ExtendVisibilityTimeout func (d *sqlDelivery) ExtendVisibilityTimeout(ctx context.Context, durationMillis int64) error { d.mu.Lock() @@ -1037,8 +1086,15 @@ func (w *partitionWorker) pollAndDeliver(ctx context.Context) (retErr error) { // Move to DLQ if enabled — must succeed before marking acked, // otherwise the message is lost from both main queue and DLQ. + // + // The reason is generic here because this path has no failing + // delivery to ask: Nack dead-letters the attempt that spends the + // budget, carrying the real reason, so reaching this point means + // the message never got that far — a crash, or a delivery whose + // visibility timeout expired unacked. if cfg.DLQ.Enabled { - if err := s.messageStore.MoveToDLQ(ctx, sub.topic, partitionKey, row.ID, retryCount, "exceeded retry limit", cfg.DLQ.TopicSuffix); err != nil { + retryLimitFailure := failure.New("exceeded retry limit") + if err := s.messageStore.MoveToDLQ(ctx, sub.topic, partitionKey, row.ID, retryCount, retryLimitFailure, cfg.DLQ.TopicSuffix); err != nil { return fmt.Errorf("move to DLQ message=%s: %w", row.ID, err) } } @@ -1086,6 +1142,27 @@ func (w *partitionWorker) pollAndDeliver(ctx context.Context) (retErr error) { deliveryMetadata["dlq.original_topic"] = row.OriginalTopic } + // Reassemble the failure recorded when this message was dead-lettered. + // A malformed detail column degrades to the message alone rather than + // failing the poll: a corrupt diagnostic must not stop delivery. + var ( + rowFailure failure.Failure + failed = row.FailedAt > 0 + ) + if failed { + decoded, err := failure.Decode(row.FailureDetail) + if err != nil { + s.logger.Warnw("ignoring unreadable failure detail on dlq message", + "topic", sub.topic, + "partition_key", partitionKey, + "message_id", row.ID, + "error", err, + ) + } + decoded.Message = row.LastError + rowFailure = decoded + } + // Create SQL delivery implementation delivery := newSQLDelivery( msg, @@ -1099,6 +1176,9 @@ func (w *partitionWorker) pollAndDeliver(ctx context.Context) (retErr error) { row.ID, cfg.ConsumerGroup, cfg.DLQ, + cfg.Retry, + rowFailure, + failed, ) // Deliver message diff --git a/platform/extension/messagequeue/mysql/subscriber_test.go b/platform/extension/messagequeue/mysql/subscriber_test.go index 263d80d9f..a5542df31 100644 --- a/platform/extension/messagequeue/mysql/subscriber_test.go +++ b/platform/extension/messagequeue/mysql/subscriber_test.go @@ -27,10 +27,22 @@ import ( "go.uber.org/mock/gomock" "go.uber.org/zap/zaptest" + "github.com/uber/submitqueue/platform/base/failure" entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" extqueue "github.com/uber/submitqueue/platform/extension/messagequeue" ) +// newDeliveryForTest builds a delivery against the standard fixture message, +// so a test only names the parts it cares about. +func newDeliveryForTest(sub *subscriber, attempt int, dlq extqueue.DLQConfig, retry extqueue.RetryConfig) *sqlDelivery { + msg := entityqueue.NewMessage("msg-1", []byte("payload"), "part-1", nil) + return newSQLDelivery( + msg, "1", attempt, nil, + sub, "test_topic", "part-1", 100, "msg-1", "test-group", + dlq, retry, failure.Failure{}, false, + ) +} + func testSubscriptionConfig() extqueue.SubscriptionConfig { return extqueue.DefaultSubscriptionConfig("test-subscriber", "test-consumer") } @@ -163,12 +175,7 @@ func TestSQLDelivery_Ack(t *testing.T) { mockDeliveryState, ) - msg := entityqueue.NewMessage("msg-1", []byte("payload"), "part-1", nil) - d := newSQLDelivery( - msg, "1", 1, nil, - sub, "test_topic", "part-1", 100, "msg-1", "test-group", - extqueue.DLQConfig{}, - ) + d := newDeliveryForTest(sub, 1, extqueue.DLQConfig{}, extqueue.RetryConfig{}) if tt.alreadyAcked { d.acknowledged = true @@ -235,12 +242,7 @@ func TestSQLDelivery_Postpone(t *testing.T) { mockDeliveryState, ) - msg := entityqueue.NewMessage("msg-1", []byte("payload"), "part-1", nil) - d := newSQLDelivery( - msg, "1", 1, nil, - sub, "test_topic", "part-1", 100, "msg-1", "test-group", - extqueue.DLQConfig{}, - ) + d := newDeliveryForTest(sub, 1, extqueue.DLQConfig{}, extqueue.RetryConfig{}) if tt.alreadyAcked { d.acknowledged = true @@ -318,17 +320,12 @@ func TestSQLDelivery_Reject(t *testing.T) { mockDeliveryState, ) - msg := entityqueue.NewMessage("msg-1", []byte("payload"), "part-1", nil) dlqConfig := extqueue.DLQConfig{ Enabled: tt.dlqEnabled, TopicSuffix: "_dlq", } - d := newSQLDelivery( - msg, "1", 1, nil, - sub, "test_topic", "part-1", 100, "msg-1", "test-group", - dlqConfig, - ) + d := newDeliveryForTest(sub, 1, dlqConfig, extqueue.RetryConfig{}) if tt.alreadyAcked { d.acknowledged = true @@ -336,7 +333,7 @@ func TestSQLDelivery_Reject(t *testing.T) { if tt.expectMoveDLQ { mockMsgStore.EXPECT().MoveToDLQ( - gomock.Any(), "test_topic", "part-1", "msg-1", 1, "bad payload", "_dlq", + gomock.Any(), "test_topic", "part-1", "msg-1", 1, failure.New("bad payload"), "_dlq", ).Return(tt.moveToDLQErr) if tt.moveToDLQErr == nil { @@ -352,7 +349,7 @@ func TestSQLDelivery_Reject(t *testing.T) { ).Return(nil) } - err := d.Reject(context.Background(), "bad payload") + err := d.Reject(context.Background(), failure.New("bad payload")) if tt.expectErr { require.Error(t, err) @@ -365,6 +362,94 @@ func TestSQLDelivery_Reject(t *testing.T) { } } +// A nack that spends the last of the retry budget dead-letters here, with the +// reason it was given. Left to the poll loop, the message would be +// dead-lettered on the next round with a generic reason and this one lost. +// +// The boundary matters as much as the behaviour: dead-lettering one attempt too +// early would silently cost every message a retry. +func TestSQLDelivery_NackDeadLettersWhenBudgetSpent(t *testing.T) { + tests := []struct { + name string + attempt int + maxAttempts int + wantDLQ bool + }{ + {name: "budget remaining", attempt: 1, maxAttempts: 3}, + {name: "one attempt left", attempt: 2, maxAttempts: 3}, + {name: "final attempt dead-letters", attempt: 3, maxAttempts: 3, wantDLQ: true}, + {name: "single-attempt budget dead-letters at once", attempt: 1, maxAttempts: 1, wantDLQ: true}, + // A zero budget is not "dead-letter immediately" — it is unconfigured, + // and the poll loop still governs. + {name: "unset budget never dead-letters here", attempt: 9, maxAttempts: 0}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + defer ctrl.Finish() + + mockMsgStore := NewMockmessageStore(ctrl) + mockDeliveryState := NewMockdeliveryStateStore(ctrl) + + sub := NewSubscriber( + zaptest.NewLogger(t).Sugar(), + tally.NoopScope, + mockMsgStore, + NewMockoffsetStore(ctrl), + NewMockpartitionLeaseStore(ctrl), + newTestHeartbeatStore(ctrl), + mockDeliveryState, + ) + + dlqConfig := extqueue.DLQConfig{Enabled: true, TopicSuffix: "_dlq"} + d := newDeliveryForTest(sub, tt.attempt, dlqConfig, extqueue.RetryConfig{MaxAttempts: tt.maxAttempts}) + + f := failure.New("boom", failure.Subject{Type: "batch", ID: "q/batch/1"}) + + if tt.wantDLQ { + mockMsgStore.EXPECT().MoveToDLQ( + gomock.Any(), "test_topic", "part-1", "msg-1", tt.attempt, f, "_dlq", + ).Return(nil) + mockDeliveryState.EXPECT().MarkAcked( + gomock.Any(), "test-group", "test_topic", "part-1", int64(100), + ).Return(nil) + } else { + mockDeliveryState.EXPECT().MarkNacked( + gomock.Any(), "test-group", "test_topic", "part-1", int64(100), + ).Return(nil) + } + + require.NoError(t, d.Nack(context.Background(), f)) + assert.True(t, d.acknowledged) + }) + } +} + +// A message arriving from its original topic has no failure to report, which is +// how a DLQ consumer tells "nothing recorded" apart from a recorded failure +// that named nothing. +func TestSQLDelivery_FailureAbsentOnNormalDelivery(t *testing.T) { + ctrl := gomock.NewController(t) + defer ctrl.Finish() + + sub := NewSubscriber( + zaptest.NewLogger(t).Sugar(), + tally.NoopScope, + NewMockmessageStore(ctrl), + NewMockoffsetStore(ctrl), + NewMockpartitionLeaseStore(ctrl), + newTestHeartbeatStore(ctrl), + NewMockdeliveryStateStore(ctrl), + ) + + d := newDeliveryForTest(sub, 1, extqueue.DLQConfig{}, extqueue.RetryConfig{}) + + got, failed := d.Failure() + assert.False(t, failed) + assert.Equal(t, failure.Failure{}, got) +} + // TestSubscriber_Close tests subscriber close behavior func TestSubscriber_Close(t *testing.T) { tests := []struct { diff --git a/test/integration/extension/messagequeue/mysql/BUILD.bazel b/test/integration/extension/messagequeue/mysql/BUILD.bazel index 227478415..e6803cf72 100644 --- a/test/integration/extension/messagequeue/mysql/BUILD.bazel +++ b/test/integration/extension/messagequeue/mysql/BUILD.bazel @@ -12,6 +12,7 @@ go_test( "requires-network", ], deps = [ + "//platform/base/failure:go_default_library", "//platform/base/messagequeue:go_default_library", "//platform/extension/messagequeue:go_default_library", "//platform/extension/messagequeue/mysql:go_default_library", diff --git a/test/integration/extension/messagequeue/mysql/queue_test.go b/test/integration/extension/messagequeue/mysql/queue_test.go index 740fdf4eb..65de942c8 100644 --- a/test/integration/extension/messagequeue/mysql/queue_test.go +++ b/test/integration/extension/messagequeue/mysql/queue_test.go @@ -30,6 +30,7 @@ import ( "github.com/uber-go/tally" "go.uber.org/zap/zaptest" + "github.com/uber/submitqueue/platform/base/failure" entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" extqueue "github.com/uber/submitqueue/platform/extension/messagequeue" queueMySQL "github.com/uber/submitqueue/platform/extension/messagequeue/mysql" @@ -1082,6 +1083,8 @@ func (s *SQLQueueIntegrationSuite) TestDeadLetterQueue() { // Receive and nack the message MaxAttempts times. // Each iteration: receive the message, nack with 0 delay, then wait for // the visibility timeout to expire so the message becomes deliverable again. + // Each nack carries why it failed; the last one is the reason recorded + // against the dead letter. for attempt := 1; attempt <= subConfig.Retry.MaxAttempts; attempt++ { delivery := receive(t, deliveryChan) t.Logf("Attempt %d: received message, nacking", delivery.Attempt()) @@ -1089,7 +1092,11 @@ func (s *SQLQueueIntegrationSuite) TestDeadLetterQueue() { assert.Equal(t, "poison-msg", delivery.Message().ID) // Nack without delay to retry immediately - require.NoError(t, delivery.Nack(s.ctx)) + nackFailure := failure.New( + fmt.Sprintf("processing failed on attempt %d", attempt), + failure.Subject{Type: "widget", ID: "widget-7"}, + ) + require.NoError(t, delivery.Nack(s.ctx, nackFailure)) } // After MaxAttempts, message should be moved to DLQ topic @@ -1124,7 +1131,17 @@ func (s *SQLQueueIntegrationSuite) TestDeadLetterQueue() { // Verify values assert.Equal(t, topic, metadata["dlq.original_topic"]) assert.Equal(t, fmt.Sprintf("%d", subConfig.Retry.MaxAttempts), metadata["dlq.failure_count"]) - assert.Equal(t, "exceeded retry limit", metadata["dlq.last_error"]) + + // The reason recorded is the one the final nack gave, not a generic + // "retry limit" placeholder — that is the point of carrying it on Nack. + assert.Equal(t, fmt.Sprintf("processing failed on attempt %d", subConfig.Retry.MaxAttempts), metadata["dlq.last_error"]) + + // And the structured half survives the round trip, so a DLQ consumer can + // act on the entity the failure was about. + dlqFailure, failed := dlqDelivery.Failure() + require.True(t, failed, "a message delivered from a DLQ topic reports a failure") + assert.Equal(t, metadata["dlq.last_error"], dlqFailure.Message) + assert.Equal(t, []string{"widget-7"}, dlqFailure.IDsOfType("widget")) failedAt := metadata["dlq.failed_at"] failedAtInt, err := strconv.ParseInt(failedAt, 10, 64) @@ -2363,7 +2380,7 @@ func (s *SQLQueueIntegrationSuite) TestPostponeResetsRetryBudget() { delivery := receive(t, deliveryChan) assert.Equal(t, attempt, delivery.Attempt()) assert.Equal(t, "wait-then-poison", delivery.Message().ID) - require.NoError(t, delivery.Nack(s.ctx)) + require.NoError(t, delivery.Nack(s.ctx, failure.New("still poison"))) t.Logf("Attempt %d: nacked", delivery.Attempt()) } @@ -2472,7 +2489,7 @@ func (s *SQLQueueIntegrationSuite) TestMultipleConsumerGroupsIndependentState() // CG-alpha: nack msg-1, ack msg-2 d1a := receive(t, ch1) assert.Equal(t, "shared-1", d1a.Message().ID) - require.NoError(t, d1a.Nack(s.ctx)) + require.NoError(t, d1a.Nack(s.ctx, failure.New("cg-alpha retry"))) t.Logf("cg-alpha nacked shared-1") d2a := receive(t, ch1) @@ -2547,7 +2564,7 @@ func (s *SQLQueueIntegrationSuite) TestCrashAfterRejectDoesNotLoseMessages() { require.NoError(t, deliveries["msg-A"].Ack(s.ctx)) t.Logf("Acked msg-A") - require.NoError(t, deliveries["msg-B"].Reject(s.ctx, "bad payload")) + require.NoError(t, deliveries["msg-B"].Reject(s.ctx, failure.New("bad payload"))) t.Logf("Rejected msg-B → DLQ") // Do NOT ack msg-C — simulating in-flight at crash time @@ -2658,7 +2675,7 @@ func (s *SQLQueueIntegrationSuite) TestCrashAfterRetryLimitDoesNotLoseMessages() t.Logf("Acked msg-A") // Nack B — immediately visible again for redelivery - require.NoError(t, deliveries["msg-B"].Nack(s.ctx)) + require.NoError(t, deliveries["msg-B"].Nack(s.ctx, failure.New("msg-B failed"))) t.Logf("Nacked msg-B, waiting for retry-limit to trigger auto-DLQ") // Do NOT ack msg-C — simulating in-flight at crash time.