From 93573aeec6688024daaad699f4623aa23d9e1ad1 Mon Sep 17 00:00:00 2001 From: Preetam Dwivedi Date: Sat, 1 Aug 2026 10:44:58 -0700 Subject: [PATCH] feat(orchestrator)!: finalize batches from their speculation paths MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary ### Why? Speculation was building the right things and merging on the wrong rule. A head still waited for *every* dependency to succeed before it could merge — the same rule as before any of this — so a batch built without a slow neighbour sat behind that neighbour anyway. The paths were being earned and then ignored. This is the commit that collects them. ### What? A head merges once one of its **passed** paths has had every dependency it *assumed would succeed* actually merge; assumed-failing and ignored dependencies impose no wait. That is the whole speedup: the head waits on what its passing build was actually stacked on, not on its full dependency list. A head fails only when no funded path has a future left, and the winner's siblings are superseded — cancelled to free their slots. Outcomes are reached in `finalize`, before the Speculator is asked, and committed one generation at a time, so a cascade (A fails → B's last path breaks → B fails → maybe C too) resolves in a single run without ever enacting a dependent of an outcome whose own write lost its compare-and-swap. A batch decided by a cascade gets a recovery signal before it turns terminal, since no retry of the triggering message would ever revisit it. Cancellation joins the run: a cancelling batch is just another batch the run walks — its live paths are marked cancelling, the poll loop stops their builds, and whichever later run sees them all stopped drives the batch to Cancelled. A cancelling path with no build link is cancelled immediately; there is deliberately no reservation state and no staleness bound, so a crashed dispatch can never keep a batch out of its terminal state. The legacy per-batch finalizer is deleted; `speculate.go` keeps only admission and message routing, and the package doc gains the batch lifecycle, a worked example, and the finalize step. ## Test Plan ✅ `bazel test //submitqueue/orchestrator/...` — the merge rule is table-driven per assumption; failure, cascade commit ordering, and CAS-loss isolation are covered, and cancellation end to end: never-dispatched, live-build, no-paths, and lost-race cases. ✅ `make fmt`, `make gazelle` --- .../controller/speculate/BUILD.bazel | 4 +- .../controller/speculate/check.go | 4 +- .../controller/speculate/dispatch.go | 34 +- .../orchestrator/controller/speculate/doc.go | 90 ++- .../controller/speculate/finalize.go | 388 +++++++++ .../controller/speculate/outcome.go | 134 ++++ .../controller/speculate/outcome_test.go | 237 ++++++ .../orchestrator/controller/speculate/run.go | 107 ++- .../controller/speculate/run_test.go | 733 +++++++++++++++++- .../controller/speculate/snapshot.go | 35 +- .../controller/speculate/speculate.go | 367 +-------- .../controller/speculate/speculate_test.go | 693 ++++------------- 12 files changed, 1818 insertions(+), 1008 deletions(-) create mode 100644 submitqueue/orchestrator/controller/speculate/finalize.go create mode 100644 submitqueue/orchestrator/controller/speculate/outcome.go create mode 100644 submitqueue/orchestrator/controller/speculate/outcome_test.go diff --git a/submitqueue/orchestrator/controller/speculate/BUILD.bazel b/submitqueue/orchestrator/controller/speculate/BUILD.bazel index c4085b42..23169181 100644 --- a/submitqueue/orchestrator/controller/speculate/BUILD.bazel +++ b/submitqueue/orchestrator/controller/speculate/BUILD.bazel @@ -6,6 +6,8 @@ go_library( "check.go", "dispatch.go", "doc.go", + "finalize.go", + "outcome.go", "run.go", "snapshot.go", "speculate.go", @@ -29,6 +31,7 @@ go_test( name = "go_default_test", srcs = [ "check_test.go", + "outcome_test.go", "run_test.go", "snapshot_test.go", "speculate_test.go", @@ -37,7 +40,6 @@ go_test( deps = [ "//platform/base/messagequeue:go_default_library", "//platform/consumer:go_default_library", - "//platform/errs:go_default_library", "//platform/extension/messagequeue/mock:go_default_library", "//submitqueue/core/topickey:go_default_library", "//submitqueue/entity:go_default_library", diff --git a/submitqueue/orchestrator/controller/speculate/check.go b/submitqueue/orchestrator/controller/speculate/check.go index 175127fc..eb140d37 100644 --- a/submitqueue/orchestrator/controller/speculate/check.go +++ b/submitqueue/orchestrator/controller/speculate/check.go @@ -125,8 +125,8 @@ func rejectionReason(proposal entity.Speculation, snap snapshot) (rejection, boo // fewer, and every assumption a real value. // // A malformed path is not merely suboptimal, it is unmergeable — the merge -// preconditions are read off the path's assumptions, so a path missing a -// dependency would let its head merge without waiting for it. +// preconditions are read off the path's assumptions (see mergeablePath), so a +// path missing a dependency would let its head merge without waiting for it. func isWellFormed(path entity.SpeculationPath, head entity.Batch) bool { if path.Head != head.ID { return false diff --git a/submitqueue/orchestrator/controller/speculate/dispatch.go b/submitqueue/orchestrator/controller/speculate/dispatch.go index f1854690..7c214fc4 100644 --- a/submitqueue/orchestrator/controller/speculate/dispatch.go +++ b/submitqueue/orchestrator/controller/speculate/dispatch.go @@ -26,10 +26,18 @@ import ( "github.com/uber/submitqueue/submitqueue/extension/storage" ) -// dispatch saves each head's decisions and hands the build stage its work. -// Everything decided this run — build results, broken-path cancellations and +// dispatch saves what finalize left over and hands the build stage its work. +// Everything still outstanding — build results, broken-path cancellations and // accepted proposals — is folded together per head, so a head costs one // compare-and-swap however many of its paths changed. +// +// It walks every in-flight batch, not only the speculating ones. Proposals +// apply to speculating heads alone and are simply absent for the rest, but +// observations are not: a merging or cancelling head's paths keep holding CI +// slots until their builds stop, and this is the only writer that can record +// that they have. Batches already finalized arrive here clean — +// commitOutcome persisted their set with their outcome — so only their +// dispatch is left to do. func (c *Controller) dispatch(ctx context.Context, queue string, snap snapshot, kept []entity.Speculation) error { nowMs := time.Now().UnixMilli() @@ -39,7 +47,7 @@ func (c *Controller) dispatch(ctx context.Context, queue string, snap snapshot, byHead[proposal.Path.Head] = append(byHead[proposal.Path.Head], proposal) } - for _, batch := range snap.speculating { + for _, batch := range snap.inFlight { // A head with no stored set is one nothing has been funded for yet. It // gets an empty set to fold this run's proposals into, which persist // then creates; a head that ends the run with no paths writes nothing @@ -57,7 +65,7 @@ func (c *Controller) dispatch(ctx context.Context, queue string, snap snapshot, } if changed { - if err := c.persist(ctx, set, exists); err != nil { + if _, err := c.persist(ctx, set, exists); err != nil { if errors.Is(err, storage.ErrVersionMismatch) { // Skipped rather than failed, and nothing is lost by that. // @@ -101,8 +109,9 @@ func (c *Controller) dispatch(ctx context.Context, queue string, snap snapshot, } // persist writes a head's path set, creating it if this run is the first to -// fund the head. -func (c *Controller) persist(ctx context.Context, set entity.SpeculationPathSet, exists bool) error { +// fund the head. It returns the set as stored, with its version advanced, so +// a caller that keeps the set around goes on holding a current copy. +func (c *Controller) persist(ctx context.Context, set entity.SpeculationPathSet, exists bool) (entity.SpeculationPathSet, error) { store := c.store.GetSpeculationPathSetStore() if !exists { @@ -111,23 +120,24 @@ func (c *Controller) persist(ctx context.Context, set entity.SpeculationPathSet, if errors.Is(err, storage.ErrAlreadyExists) { // Another writer created it between this run's read and now. // Treat it as a lost race: the next run reads the winner. - return storage.ErrVersionMismatch + return set, storage.ErrVersionMismatch } metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) - return fmt.Errorf("failed to create path set for batch %s: %w", set.Head, err) + return set, fmt.Errorf("failed to create path set for batch %s: %w", set.Head, err) } - return nil + return set, nil } newVersion := set.Version + 1 if err := store.Update(ctx, set, set.Version, newVersion); err != nil { if errors.Is(err, storage.ErrVersionMismatch) { - return err + return set, err } metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) - return fmt.Errorf("failed to update path set for batch %s: %w", set.Head, err) + return set, fmt.Errorf("failed to update path set for batch %s: %w", set.Head, err) } - return nil + set.Version = newVersion + return set, nil } // applyProposal folds one accepted proposal into the set and reports whether diff --git a/submitqueue/orchestrator/controller/speculate/doc.go b/submitqueue/orchestrator/controller/speculate/doc.go index 10127d7a..bed664f3 100644 --- a/submitqueue/orchestrator/controller/speculate/doc.go +++ b/submitqueue/orchestrator/controller/speculate/doc.go @@ -12,23 +12,19 @@ // See the License for the specific language governing permissions and // limitations under the License. -// Package speculate plans a queue's speculative builds: which guesses about -// the queue's future are worth building, within the queue's cap on concurrent -// builds (the build budget). -// -// Batch outcomes — merge or fail — are still decided by the legacy per-batch -// finalizer in speculate.go, which waits on every dependency. Deriving them -// from the paths planned here replaces it in the next change; this package -// doc grows with it. +// Package speculate plans a queue's speculative builds and finalizes each +// batch's outcome from their results. It is the orchestrator's decision +// stage: builds are started by the build stage and watched — and stopped — +// by the buildsignal stage, but what to build and what a finished build +// means are decided here. // // # Why speculation // // Batches in a queue depend on the batches ahead of them, so without // speculation everything is serial: C waits for B, B waits for A. Speculation // builds a batch against a guess about how its dependencies turn out. When -// the guess holds, the head's build has already run by the time its -// dependencies resolve — it never waits for a build of its own to start -// afterwards. +// the guess holds, the batch merges the moment the guessed-on dependencies +// land — it never waits for a build of its own to start afterwards. // // # Paths // @@ -38,6 +34,26 @@ // a path *is* its guess; building the same guess again is a new attempt of // the same path, and (path ID, attempt) names the resulting build. // +// # A worked example +// +// A two-batch queue, where B depends on A and A is still building: +// +// queue: A ← B +// +// B's speculation space is two paths: +// P1 = [A succeeds] B built on top of A's result +// P2 = [A fails] B built without A +// +// Fund both and every future is covered: +// +// - A succeeds and P1 passed: B merges the moment A lands. P2's guess +// ("A fails") is broken — it can no longer come true — so its build is +// cancelled to free the slot. +// - A fails and P2 passed: B merges without A, again with no new build. +// P1's guess is broken. +// - A resolved either way, and every unbroken path failed: no future +// remains in which B passes, so B fails. +// // # The life of a path // // A path's status tracks its current attempt: @@ -46,10 +62,11 @@ // (no entry) ─────────► pending ─────────► building ──────┬──► passed // │ │ └──► failed // "stop this": │ │ -// broken or ▼ ▼ -// preempted ─────► cancelling ◄────┘ -// │ -// │ build observed stopped +// broken, ▼ ▼ +// superseded, ─────► cancelling ◄────┘ +// head cancelled, │ +// or preempted │ build observed stopped, or +// │ nothing was ever dispatched // ▼ // cancelled // @@ -58,14 +75,27 @@ // building, and every pending, building, and cancelling path holds its slot // until its build stops. A path is broken once a dependency's actual result // proves one of its assumptions wrong: its guess can no longer come true, so -// its build is cancelled to free the slot. +// its build is cancelled to free the slot. A path is superseded when a +// sibling path of the same head passes — that sibling will carry the head out +// of the queue, so the others are cancelled too. // // Cancelling is intent, not fact: the build keeps its slot until CI actually -// stops it, and only an observation of that stop moves the path to cancelled. -// The intent needs no dispatch of its own — the poll loop reads it off the set -// and asks the runner to stop the build. A terminal path can be resurrected by -// a new build proposal — status returns to pending and Attempt increments, the -// one backwards step in the diagram. +// stops it, and only an observation of that stop (or proof nothing was ever +// dispatched) moves the path to cancelled. The intent needs no dispatch of its +// own — the poll loop reads it off the set and asks the runner to stop the +// build. A terminal path can be resurrected by a new build proposal — status +// returns to pending and Attempt increments, the one backwards step in the +// diagram. +// +// # The life of a batch, as seen from here +// +// Created ──admit──► Speculating ──┬── merge ──► Merging (merge stage takes over) +// └── fail ───► Failed +// user cancel (cancel stage): +// ... ──► Cancelling ── every path stopped ──► Cancelled +// +// Failed and Cancelled fan out to the conclude stage, which reconciles the +// batch's requests. // // # How a run works // @@ -75,16 +105,18 @@ // reordered signals are harmless, and a later run repairs whatever an // earlier one left half-done. // -// signal ──► read ──► cancel ──► ask ──► check ──► dispatch -// one broken the filter save changes, -// read of paths Specu- its hand builds to -// queue + lator proposals the build stage -// paths +// signal ──► read ──► finalize ──► ask ──► check ──► dispatch +// one enact the the filter save changes, +// read of outcomes Specu- its hand builds to +// queue + the facts lator proposals the build stage +// paths already +// decide // // The Speculator is the extension that proposes which paths to fund or -// preempt. It only ever proposes: check.go filters its answer, and broken -// paths are cancelled before it is asked, so it reasons over facts as they -// now stand rather than over a picture the run is about to invalidate. +// preempt. It only ever proposes: check.go filters its answer, and outcomes +// are computed here, never by the extension. Finalize runs before ask so the +// Speculator reasons over facts as they now stand, not over a picture the run +// is about to invalidate. // // # Ownership // diff --git a/submitqueue/orchestrator/controller/speculate/finalize.go b/submitqueue/orchestrator/controller/speculate/finalize.go new file mode 100644 index 00000000..77ca3dae --- /dev/null +++ b/submitqueue/orchestrator/controller/speculate/finalize.go @@ -0,0 +1,388 @@ +// 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 speculate + +import ( + "context" + "errors" + "fmt" + "time" + + "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/submitqueue/core/topickey" + "github.com/uber/submitqueue/submitqueue/entity" + "github.com/uber/submitqueue/submitqueue/extension/storage" +) + +// finalize reaches and enacts every outcome this run can conclude, and leaves +// snap.speculating holding the heads still open to new work. +// +// Everything here is a fact, not a choice: a path a resolved dependency ruled +// out is dead, a head whose passed build's assumptions all came true +// merges, and a batch the user cancelled is finished once its last build +// stops. Finalizing before the Speculator is asked is what keeps its work from +// being wasted — asked first, it would propose builds for a head that is +// already merging. +// +// Outcomes cascade, and the head loop runs to a fixed point to collapse a +// whole cascade into this one run: +// +// A fails ──breaks──► B's last live path [A succeeds] ──► B fails +// ──breaks──► C's [A succeeds, ...] paths ──► maybe C too +// +// One pass would only catch dependents that happen to come after their +// prerequisite in queue order — and that order is not even specified — +// leaving the rest to wait for an unrelated later signal. The loop terminates +// because every iteration but the last finalizes at least one head. +// +// Each generation is committed before the next is derived from it, because an +// outcome must be durable before anything is allowed to depend on it. +// Deciding the whole cascade up front and writing afterwards would enact +// dependents of an outcome whose own write then lost its compare-and-swap — +// and the loser of that race is not always benign: a cancellation loses +// precisely to a merge that got there first, which leaves the batch +// *succeeded*, after its dependents were already failed on the assumption it +// was cancelled. Committing per generation costs no extra reads — the +// snapshot is read once, and the writes are ones this run makes anyway. +func (c *Controller) finalize(ctx context.Context, snap *snapshot) error { + nowMs := time.Now().UnixMilli() + + // Cancellations first: a cancelled batch is a resolved dependency like any + // other, and the heads below have to see it as one. + if err := c.finalizeCancellations(ctx, snap, nowMs); err != nil { + return err + } + + open := snap.speculating + for { + decided := 0 + var stillOpen []entity.Batch + + for _, batch := range open { + // Fold in what the facts already imply about this head's paths: + // a finished dependency breaks every path that bet against it. + set, exists := snap.pathSets[batch.ID] + if exists && cancelBrokenPathsInSet(&set, *snap, nowMs) { + snap.pathSets[batch.ID] = set + snap.markDirty(batch.ID) + } + + decision := decide(batch, set, *snap) + if decision == outcomeWait { + stillOpen = append(stillOpen, batch) + continue + } + + if decision == outcomeMerge { + // The winning path carries the head out of the queue; its + // siblings cannot help it any more and are still holding CI + // slots the rest of the queue could use. + winner, _ := mergeablePath(set, *snap) + if supersede(&set, winner.ID, nowMs) { + snap.pathSets[batch.ID] = set + snap.markDirty(batch.ID) + } + } + + committed, err := c.commitOutcome(ctx, snap, batch, decision) + if err != nil { + return err + } + if !committed { + // Another writer owns this batch now, so our view of it is + // stale. Dropped rather than kept open: nothing may be derived + // from an outcome that did not land, and the Speculator must + // not be offered a head whose set we could not write. + continue + } + c.recordOutcome(snap, batch.ID, decision) + decided++ + } + + open = stillOpen + if decided == 0 { + break + } + } + snap.speculating = open + + return nil +} + +// finalizeCancellations marks every path of a cancelling batch stopped and, +// once they all have, drives the batch to cancelled. +// +// This is the whole of the cancel hand-off: the cancel controller writes the +// user's intent and publishes the batch here, and from then on a cancelling +// batch is just another batch the run walks. A batch is not cancelled the +// moment it is asked to be — its builds hold their CI slots until they +// actually stop — so the run marks the paths, the poll loop asks the runner +// to stop them, and whichever later run sees them stopped finishes the +// job. That is why cancellation is best effort, and why a merge that wins the +// race still prevails. +func (c *Controller) finalizeCancellations(ctx context.Context, snap *snapshot, nowMs int64) error { + // TODO(respeculate-collateral): re-enqueue Land for every request in batch.Contains + // except the user-cancelled request. Today the whole batch dies (per spec) and the + // collateral requests need a fresh request ID and a re-publish to TopicKeyStart so + // they can be re-batched without the cancelled change. + for _, batch := range snap.inFlight { + if batch.State != entity.BatchStateCancelling { + continue + } + metrics.NamedCounter(c.metricsScope, opName, "cancel_batch", 1) + + set, exists := snap.pathSets[batch.ID] + if exists && cancelAllPaths(&set, nowMs) { + snap.pathSets[batch.ID] = set + snap.markDirty(batch.ID) + } + + // An absent set means nothing was ever funded, so there is nothing to + // wait for. Statuses here already reflect what the build stages saw — + // read folded that in — so a path still reading as unstopped really is. + if exists && !allPathsStopped(set) { + metrics.NamedCounter(c.metricsScope, opName, "cancel_awaiting_paths", 1) + c.logger.Infow("cancelling batch; waiting for its builds to stop", + "batch_id", batch.ID, + "queue", batch.Queue, + ) + continue + } + + committed, err := c.commitOutcome(ctx, snap, batch, outcomeCancel) + if err != nil { + return err + } + if committed { + c.recordOutcome(snap, batch.ID, outcomeCancel) + } + } + return nil +} + +// commitOutcome persists a decided batch's path changes and enacts its +// outcome, reporting whether the outcome actually landed. False means another +// writer moved the head on, and nothing else in this run may be derived from +// this outcome. +// +// The two writes belong together because the outcome is only meaningful with +// the paths it was read from. A lost path-set compare-and-swap means the +// outcome decided from our copy may no longer be right, so the state write is +// not attempted. +func (c *Controller) commitOutcome(ctx context.Context, snap *snapshot, batch entity.Batch, decision outcome) (bool, error) { + if snap.isDirty(batch.ID) { + set, exists := snap.pathSets[batch.ID] + stored, err := c.persist(ctx, set, exists) + if err != nil { + if errors.Is(err, storage.ErrVersionMismatch) { + metrics.NamedCounter(c.metricsScope, opName, "path_set_cas_lost", 1) + c.logger.Infow("lost a path set write; the next run re-plans this head", + "batch_id", batch.ID, + "queue", batch.Queue, + ) + // Abandon the stale copy so write does not retry it. + snap.markClean(batch.ID) + return false, nil + } + return false, err + } + snap.pathSets[batch.ID] = stored + snap.markClean(batch.ID) + } + + return c.applyOutcome(ctx, batch, decision, snap.isTrigger(batch.ID)) +} + +// recordOutcome writes an outcome's terminal state back into the snapshot so +// the rest of the run reasons from it, exactly as it would from an outcome +// recorded before the run started. It is called only once that state is +// durable — see commitOutcome — because everything concluded about the +// batches stacked on this one is derived from it. +// +// Only a terminal outcome is recorded. Merging is not terminal — a head +// stacked on this one assumed it would *succeed*, and it has not yet — so a +// merge outcome resolves nothing for anybody else. +func (c *Controller) recordOutcome(snap *snapshot, batchID string, decision outcome) { + state, terminal := decision.terminalState() + if !terminal { + return + } + batch := snap.batches[batchID] + batch.State = state + snap.batches[batchID] = batch +} + +// applyOutcome enacts a decided outcome on a batch, reporting whether the +// state write landed. +// +// The publish order differs per arm, but it is one rule read twice: a publish +// may precede a state write only when the consumer does not read the state +// that write produces. The merge stage correlates on the batch ID alone, so +// telling it before the write is safe — a batch recorded Merging that Runway +// never heard about would merely stall. Conclude does read the state (it +// reconciles requests from it and rejects a non-terminal batch outright), so +// it is published only after the write, or it would race the consumer into +// the dead-letter queue. +// +// Losing the state compare-and-swap is not an error: another writer got +// there, and the next run reads whatever they wrote. It is reported as not +// landed, because the outcome this run reached is not the one that took +// effect. +func (c *Controller) applyOutcome(ctx context.Context, batch entity.Batch, decision outcome, isTriggerBatch bool) (bool, error) { + var state entity.BatchState + terminal := false + + switch decision { + case outcomeMerge: + state = entity.BatchStateMerging + if err := c.publishBatchID(ctx, topickey.TopicKeyMerge, batch.ID, batch.Queue); err != nil { + metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1) + return false, fmt.Errorf("failed to publish batch %s to merge: %w", batch.ID, err) + } + + case outcomeFail, outcomeCancel: + state, terminal = decision.terminalState() + // A batch decided by a cascade is not the one on the message, so no + // retry or dead letter would ever come back to it — give it a recovery + // message of its own before it turns terminal. + if !isTriggerBatch { + if err := c.recoverable(ctx, batch); err != nil { + return false, err + } + } + + default: + // outcomeWait: nothing to enact. Listed explicitly so an unknown or + // zero outcome can never fall into an enacting arm. + return false, nil + } + + newVersion := batch.Version + 1 + updated := batch + updated.State = state + if err := c.store.GetBatchStore().Update(ctx, updated, batch.Version, newVersion); err != nil { + if errors.Is(err, storage.ErrVersionMismatch) { + metrics.NamedCounter(c.metricsScope, opName, "outcome_cas_lost", 1) + return false, nil + } + metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) + return false, fmt.Errorf("failed to update batch %s state to %s: %w", batch.ID, state, err) + } + + metrics.NamedCounter(c.metricsScope, opName, "outcome", 1, metrics.NewTag("outcome", string(decision))) + c.logger.Infow("batch outcome", + "batch_id", batch.ID, + "queue", batch.Queue, + "outcome", string(decision), + "state", string(state), + ) + + if terminal { + if err := c.publishBatchID(ctx, topickey.TopicKeyConclude, batch.ID, batch.Queue); err != nil { + metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1) + return true, fmt.Errorf("failed to publish batch %s to conclude: %w", batch.ID, err) + } + } + return true, nil +} + +// recoverable gives a batch a message of its own before this run makes it +// terminal, so its fan-out cannot be stranded by a failure afterwards. +// +// Every other terminal batch is repaired through the message that names it: a +// redelivery finds it terminal and re-publishes from Process's self-heal +// branch, and a persistent failure lands it in the dead-letter queue by name. +// A batch decided by a cascade has neither — it is not the batch on the +// message, and once terminal it is gone from the queue listing — so without +// this its requests would simply stay unreconciled. +// +// Published before the state write, not after: sent afterwards it is one more +// thing that can fail exactly when everything else is failing, leaving the +// obligation created and the means to discharge it gone. Sent first, a +// failure means nothing was written at all and the retry re-derives the whole +// decision from unchanged state. A duplicate is harmless — Speculate +// tolerates any batch state, and if the write never lands the message is just +// a nudge that re-plans a queue nothing has changed. +func (c *Controller) recoverable(ctx context.Context, batch entity.Batch) error { + if err := c.publishBatchID(ctx, topickey.TopicKeySpeculate, batch.ID, batch.Queue); err != nil { + metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1) + return fmt.Errorf("failed to publish recovery signal for batch %s: %w", batch.ID, err) + } + metrics.NamedCounter(c.metricsScope, opName, "recovery_signal_published", 1) + return nil +} + +// markCancelling asks every live path in the set to stop — except the ones +// keep returns true for — and reports whether anything changed. A live path +// is one neither terminal nor already cancelling. +// +// Cancelling is intent, not fact: the build may still be occupying CI, and +// only an observation of it actually stopping (or of nothing ever having been +// dispatched) moves the path on to cancelled. cancelBrokenPathsInSet, +// supersede, and cancelAllPaths are the three reasons to ask. +func markCancelling(set *entity.SpeculationPathSet, nowMs int64, keep func(entity.SpeculationPathEntry) bool) bool { + changed := false + for i := range set.Paths { + entry := &set.Paths[i] + if entry.Status.IsTerminal() || entry.Status == entity.SpeculationPathStatusCancelling { + continue + } + if keep(*entry) { + continue + } + entry.Status = entity.SpeculationPathStatusCancelling + entry.UpdatedAtMs = nowMs + changed = true + } + return changed +} + +// cancelBrokenPathsInSet stops every in-flight path with a broken assumption. +// Those guesses can no longer come true, so their builds are only spending +// budget. +func cancelBrokenPathsInSet(set *entity.SpeculationPathSet, snap snapshot, nowMs int64) bool { + return markCancelling(set, nowMs, func(entry entity.SpeculationPathEntry) bool { + return !assumptionBroken(entry.Path, snap) + }) +} + +// supersede stops every path other than the winner. Once one path has passed, +// its siblings cannot help the head any more — but they are still holding CI +// slots the rest of the queue could use. +func supersede(set *entity.SpeculationPathSet, winnerID string, nowMs int64) bool { + return markCancelling(set, nowMs, func(entry entity.SpeculationPathEntry) bool { + return entry.ID == winnerID + }) +} + +// cancelAllPaths stops every path that is still running. Used when the head +// itself was cancelled by the user, so no path can help it any more. +func cancelAllPaths(set *entity.SpeculationPathSet, nowMs int64) bool { + return markCancelling(set, nowMs, func(entity.SpeculationPathEntry) bool { + return false + }) +} + +// allPathsStopped reports whether none of the head's paths still holds a +// build. A cancelling path has not stopped: its build occupies CI until an +// observation records it terminal. +func allPathsStopped(set entity.SpeculationPathSet) bool { + for _, entry := range set.Paths { + if !entry.Status.IsTerminal() { + return false + } + } + return true +} diff --git a/submitqueue/orchestrator/controller/speculate/outcome.go b/submitqueue/orchestrator/controller/speculate/outcome.go new file mode 100644 index 00000000..16932ad1 --- /dev/null +++ b/submitqueue/orchestrator/controller/speculate/outcome.go @@ -0,0 +1,134 @@ +// 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 speculate + +import "github.com/uber/submitqueue/submitqueue/entity" + +// outcome is what a run has concluded about one batch. It is computed from +// the snapshot alone — decide is a pure function of the facts — which is what +// guarantees a swapped-in Speculator can change which paths run but never a +// batch's outcome. +type outcome string + +const ( + // outcomeWait means the batch's outcome is not decided yet. + outcomeWait outcome = "wait" + // outcomeMerge means a passed path's assumptions have all come true, so + // the head can be handed to the merge stage. + outcomeMerge outcome = "merge" + // outcomeFail means no future remains in which the head could pass. + outcomeFail outcome = "fail" + // outcomeCancel means a batch the user asked to cancel has had every path + // stop, so nothing of it is still running. + outcomeCancel outcome = "cancel" +) + +// terminalState returns the batch state an outcome writes, and whether the +// outcome leaves the batch terminal. Merge is the odd one out: it hands the +// batch to the merge stage, which owns the terminal write that follows. +func (v outcome) terminalState() (entity.BatchState, bool) { + switch v { + case outcomeFail: + return entity.BatchStateFailed, true + case outcomeCancel: + return entity.BatchStateCancelled, true + default: + return entity.BatchStateUnknown, false + } +} + +// decide returns the run's outcome on one open head, from the snapshot alone. +func decide(head entity.Batch, set entity.SpeculationPathSet, snap snapshot) outcome { + if _, ok := mergeablePath(set, snap); ok { + return outcomeMerge + } + if hasNoViableFuture(head, set, snap) { + return outcomeFail + } + return outcomeWait +} + +// mergeablePath returns a passed path whose merge preconditions are met: +// every dependency it assumed would succeed has actually merged. +// +// This is what makes speculation pay. The head waits only on the dependencies +// the passed build was stacked on — not on its full dependency list — so a +// batch built without a slow neighbour, or with that neighbour relaxed to +// ignored, merges as soon as the ones it actually built on have landed. +// +// A dependency assumed to fail imposes no wait: the path is broken the +// moment that dependency succeeds, so a still-live path has already been +// vindicated on it. An ignored dependency imposes no wait by definition. +func mergeablePath(set entity.SpeculationPathSet, snap snapshot) (entity.SpeculationPathEntry, bool) { + for _, entry := range set.Paths { + if entry.Status != entity.SpeculationPathStatusPassed { + continue + } + if assumptionBroken(entry.Path, snap) { + continue + } + if allAssumedSucceedingMerged(entry.Path, snap) { + return entry, true + } + } + return entity.SpeculationPathEntry{}, false +} + +// allAssumedSucceedingMerged reports whether every dependency the path +// assumed would succeed has reached Succeeded. +func allAssumedSucceedingMerged(path entity.SpeculationPath, snap snapshot) bool { + for _, dep := range path.Dependencies { + if dep.Assumption != entity.DependencyAssumptionSucceeds { + continue + } + if snap.batchState(dep.Batch) != entity.BatchStateSucceeded { + return false + } + } + return true +} + +// hasNoViableFuture reports whether the head can never pass: every dependency +// has resolved, and every path consistent with how they resolved has a failed +// build. +// +// Deliberately conservative — failing a batch that could still have landed is +// far worse than failing it a tick later, so this errs toward waiting: +// +// - While any dependency is unresolved, an untried future may still exist, +// so the answer is no. +// - A head with no unbroken paths at all is not failed either: it simply +// has nothing funded yet, and the next run's Speculator will propose +// something. +func hasNoViableFuture(head entity.Batch, set entity.SpeculationPathSet, snap snapshot) bool { + for _, depID := range head.Dependencies { + if !snap.batchState(depID).IsTerminal() { + return false + } + } + + live := 0 + for _, entry := range set.Paths { + if assumptionBroken(entry.Path, snap) { + continue + } + live++ + if entry.Status != entity.SpeculationPathStatusFailed { + return false + } + } + + return live > 0 +} diff --git a/submitqueue/orchestrator/controller/speculate/outcome_test.go b/submitqueue/orchestrator/controller/speculate/outcome_test.go new file mode 100644 index 00000000..97801ee5 --- /dev/null +++ b/submitqueue/orchestrator/controller/speculate/outcome_test.go @@ -0,0 +1,237 @@ +// 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 speculate + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/uber/submitqueue/submitqueue/entity" +) + +// passedPath returns a passed entry for a path over the given assumptions. +func passedPath(assumptions ...entity.DependencyAssumption) entity.SpeculationPathEntry { + return entryFor(pathOver(assumptions...), entity.SpeculationPathStatusPassed) +} + +func setOf(entries ...entity.SpeculationPathEntry) entity.SpeculationPathSet { + return entity.SpeculationPathSet{Head: head, Paths: entries} +} + +// The payoff case: a head merges as soon as the dependencies its passed build +// was stacked on have landed, without waiting for the ones it was built +// without or told to ignore. +func TestMergeablePath(t *testing.T) { + const ( + succeeds = entity.DependencyAssumptionSucceeds + fails = entity.DependencyAssumptionFails + ignored = entity.DependencyAssumptionIgnored + ) + + tests := []struct { + name string + assumption [2]entity.DependencyAssumption + dep1State entity.BatchState + dep2State entity.BatchState + want bool + }{ + { + name: "waits for an assumed-succeeding dependency to merge", + assumption: [2]entity.DependencyAssumption{succeeds, ignored}, + dep1State: entity.BatchStateSpeculating, + want: false, + }, + { + name: "merges once it has", + assumption: [2]entity.DependencyAssumption{succeeds, ignored}, + dep1State: entity.BatchStateSucceeded, + want: true, + }, + { + name: "an assumed-failing dependency imposes no wait", + assumption: [2]entity.DependencyAssumption{fails, ignored}, + dep1State: entity.BatchStateSpeculating, + want: true, + }, + { + name: "an ignored dependency imposes no wait", + assumption: [2]entity.DependencyAssumption{ignored, ignored}, + dep1State: entity.BatchStateSpeculating, + want: true, + }, + { + name: "one unmerged dependency is enough to wait", + assumption: [2]entity.DependencyAssumption{succeeds, succeeds}, + dep1State: entity.BatchStateSucceeded, + dep2State: entity.BatchStateSpeculating, + want: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + dep2 := tt.dep2State + if dep2 == "" { + dep2 = entity.BatchStateSpeculating + } + set := setOf(passedPath(tt.assumption[0], tt.assumption[1])) + _, ok := mergeablePath(set, snapWith(tt.dep1State, dep2)) + assert.Equal(t, tt.want, ok) + }) + } +} + +// A path that has not passed cannot carry the head out of the queue. +func TestMergeablePath_IgnoresUnpassedPaths(t *testing.T) { + for _, status := range []entity.SpeculationPathStatus{ + entity.SpeculationPathStatusPending, + entity.SpeculationPathStatusBuilding, + entity.SpeculationPathStatusFailed, + entity.SpeculationPathStatusCancelled, + entity.SpeculationPathStatusCancelling, + } { + t.Run(string(status), func(t *testing.T) { + set := setOf(entryFor( + pathOver(entity.DependencyAssumptionIgnored, entity.DependencyAssumptionIgnored), status)) + _, ok := mergeablePath(set, snapWith(entity.BatchStateSucceeded, entity.BatchStateSucceeded)) + assert.False(t, ok) + }) + } +} + +// A passed build whose assumptions reality has since contradicted is not a +// licence to merge — it verified a world that did not happen. +func TestMergeablePath_ExcludesBrokenPassedPath(t *testing.T) { + set := setOf(passedPath(entity.DependencyAssumptionFails, entity.DependencyAssumptionIgnored)) + + // The path was built without dep1, but dep1 landed after all. + _, ok := mergeablePath(set, snapWith(entity.BatchStateSucceeded, entity.BatchStateSpeculating)) + assert.False(t, ok) +} + +func TestHasNoViableFuture(t *testing.T) { + headBatch := entity.Batch{ID: head, Dependencies: []string{dep1, dep2}} + failed := entryFor( + pathOver(entity.DependencyAssumptionIgnored, entity.DependencyAssumptionIgnored), + entity.SpeculationPathStatusFailed) + + t.Run("waits while a dependency is unresolved", func(t *testing.T) { + // An unresolved dependency means futures the queue has not tried yet. + snap := snapWith(entity.BatchStateSucceeded, entity.BatchStateSpeculating) + assert.False(t, hasNoViableFuture(headBatch, setOf(failed), snap)) + }) + + t.Run("fails once everything is resolved and every live path failed", func(t *testing.T) { + snap := snapWith(entity.BatchStateSucceeded, entity.BatchStateSucceeded) + assert.True(t, hasNoViableFuture(headBatch, setOf(failed), snap)) + }) + + t.Run("does not fail while a path is still running", func(t *testing.T) { + running := entryFor( + pathOver(entity.DependencyAssumptionIgnored, entity.DependencyAssumptionIgnored), + entity.SpeculationPathStatusBuilding) + running.ID = "still-running" + snap := snapWith(entity.BatchStateSucceeded, entity.BatchStateSucceeded) + assert.False(t, hasNoViableFuture(headBatch, setOf(failed, running), snap)) + }) + + t.Run("does not fail a head with nothing funded", func(t *testing.T) { + snap := snapWith(entity.BatchStateSucceeded, entity.BatchStateSucceeded) + assert.False(t, hasNoViableFuture(headBatch, setOf(), snap), + "an unfunded head is waiting for the Speculator, not out of options") + }) + + t.Run("ignores broken paths when judging", func(t *testing.T) { + // This path assumed dep1 would fail; it succeeded, so the failed build + // tells us nothing about a future that can still happen. + brokenFail := entryFor( + pathOver(entity.DependencyAssumptionFails, entity.DependencyAssumptionIgnored), + entity.SpeculationPathStatusFailed) + snap := snapWith(entity.BatchStateSucceeded, entity.BatchStateSucceeded) + assert.False(t, hasNoViableFuture(headBatch, setOf(brokenFail), snap)) + }) +} + +func TestDecide(t *testing.T) { + headBatch := entity.Batch{ID: head, Dependencies: []string{dep1, dep2}} + allResolved := snapWith(entity.BatchStateSucceeded, entity.BatchStateSucceeded) + + passed := passedPath(entity.DependencyAssumptionSucceeds, entity.DependencyAssumptionSucceeds) + failed := entryFor( + pathOver(entity.DependencyAssumptionIgnored, entity.DependencyAssumptionIgnored), + entity.SpeculationPathStatusFailed) + building := entryFor( + pathOver(entity.DependencyAssumptionSucceeds, entity.DependencyAssumptionIgnored), + entity.SpeculationPathStatusBuilding) + + assert.Equal(t, outcomeMerge, decide(headBatch, setOf(passed), allResolved)) + assert.Equal(t, outcomeFail, decide(headBatch, setOf(failed), allResolved)) + assert.Equal(t, outcomeWait, decide(headBatch, setOf(building), allResolved)) + + // A passed path wins over a failed sibling: one way through is enough. + assert.Equal(t, outcomeMerge, decide(headBatch, setOf(failed, passed), allResolved)) +} + +// Once a path has passed, its siblings cannot help the head but are still +// holding CI slots the rest of the queue could use. +func TestSupersede(t *testing.T) { + winner := passedPath(entity.DependencyAssumptionSucceeds, entity.DependencyAssumptionIgnored) + sibling := entryFor( + pathOver(entity.DependencyAssumptionFails, entity.DependencyAssumptionIgnored), + entity.SpeculationPathStatusBuilding) + finished := entryFor( + pathOver(entity.DependencyAssumptionIgnored, entity.DependencyAssumptionIgnored), + entity.SpeculationPathStatusFailed) + + set := setOf(winner, sibling, finished) + assert.True(t, supersede(&set, winner.ID, 99)) + + assert.Equal(t, entity.SpeculationPathStatusPassed, set.Paths[0].Status, "the winner is untouched") + assert.Equal(t, entity.SpeculationPathStatusCancelling, set.Paths[1].Status) + assert.Equal(t, int64(99), set.Paths[1].UpdatedAtMs) + assert.Equal(t, entity.SpeculationPathStatusFailed, set.Paths[2].Status, "a finished path is left alone") + + assert.False(t, supersede(&set, winner.ID, 100), "a second pass changes nothing") +} + +func TestAllPathsStopped(t *testing.T) { + running := entryFor(pathOver(entity.DependencyAssumptionIgnored, entity.DependencyAssumptionIgnored), + entity.SpeculationPathStatusBuilding) + cancelling := entryFor(pathOver(entity.DependencyAssumptionSucceeds, entity.DependencyAssumptionIgnored), + entity.SpeculationPathStatusCancelling) + done := entryFor(pathOver(entity.DependencyAssumptionFails, entity.DependencyAssumptionIgnored), + entity.SpeculationPathStatusCancelled) + + assert.True(t, allPathsStopped(setOf(done))) + assert.True(t, allPathsStopped(setOf())) + assert.False(t, allPathsStopped(setOf(running))) + assert.False(t, allPathsStopped(setOf(cancelling)), + "a cancelling build holds its CI slot until it actually stops") +} + +func TestCancelAllPaths(t *testing.T) { + running := entryFor(pathOver(entity.DependencyAssumptionIgnored, entity.DependencyAssumptionIgnored), + entity.SpeculationPathStatusBuilding) + done := entryFor(pathOver(entity.DependencyAssumptionFails, entity.DependencyAssumptionIgnored), + entity.SpeculationPathStatusPassed) + + set := setOf(running, done) + assert.True(t, cancelAllPaths(&set, 7)) + assert.Equal(t, entity.SpeculationPathStatusCancelling, set.Paths[0].Status) + assert.Equal(t, entity.SpeculationPathStatusPassed, set.Paths[1].Status, + "a finished build has no slot to release") + + assert.False(t, cancelAllPaths(&set, 8), "a second pass changes nothing") +} diff --git a/submitqueue/orchestrator/controller/speculate/run.go b/submitqueue/orchestrator/controller/speculate/run.go index bfcb6db7..f138a790 100644 --- a/submitqueue/orchestrator/controller/speculate/run.go +++ b/submitqueue/orchestrator/controller/speculate/run.go @@ -18,7 +18,6 @@ import ( "context" "errors" "fmt" - "time" "github.com/uber/submitqueue/platform/metrics" "github.com/uber/submitqueue/submitqueue/entity" @@ -27,30 +26,30 @@ import ( ) // run re-plans a whole queue from a single read of its state, in the five -// steps the package doc lays out: read, cancel broken paths, ask, check, -// dispatch. +// steps the package doc lays out: read, finalize, ask, check, dispatch. // // The batch on the triggering message only says which queue woke up; nothing -// about the run depends on which batch it was, or on any earlier run. That is -// what makes duplicated, delayed, and reordered signals harmless, and what -// lets a later run repair anything an earlier one left half-done. -// -// Cancelling broken paths before asking is what keeps the Speculator's work -// from being wasted: it reasons over the queue as the facts have already left -// it, rather than over a picture this run is about to invalidate. -func (c *Controller) run(ctx context.Context, queue string) error { - snap, err := c.read(ctx, queue) +// about the plan depends on which batch it was, or on any earlier run. Its +// identity is carried through only for crash recovery — see snapshot.trigger. +func (c *Controller) run(ctx context.Context, trigger entity.Batch) error { + snap, err := c.read(ctx, trigger.Queue) if err != nil { return err } + snap.trigger = trigger.ID + + if err := c.finalize(ctx, &snap); err != nil { + return err + } if len(snap.speculating) == 0 { - // No head is open to new work, so there is nothing to speculate about. - return nil + // No head is open to new work, so there is nothing to ask the + // Speculator. The dispatch step still runs: what the build stages saw about a + // merging or cancelling head's paths has to be persisted so those + // paths stop counting against the budget. + return c.dispatch(ctx, trigger.Queue, snap, nil) } - c.cancelBrokenPaths(&snap) - - proposals, err := c.ask(ctx, queue, snap) + proposals, err := c.ask(ctx, trigger.Queue, snap) if err != nil { return err } @@ -60,12 +59,12 @@ func (c *Controller) run(ctx context.Context, queue string) error { metrics.NamedCounter(c.metricsScope, opName, "speculation_rejected", 1, metrics.NewTag("reason", string(reason))) c.logger.Warnw("dropped a speculator proposal", - "queue", queue, + "queue", trigger.Queue, "reason", string(reason), ) } - return c.dispatch(ctx, queue, snap, kept) + return c.dispatch(ctx, trigger.Queue, snap, kept) } // read builds the run's snapshot. Batches come first because their dependency @@ -83,6 +82,7 @@ func (c *Controller) read(ctx context.Context, queue string) (snapshot, error) { snap := snapshot{ batches: make(map[string]entity.Batch, len(inFlight)), + inFlight: inFlight, pathSets: make(map[string]entity.SpeculationPathSet, len(inFlight)), dirty: make(map[string]bool, len(inFlight)), } @@ -163,8 +163,18 @@ func (c *Controller) updatePathsFromBuilds(ctx context.Context, set *entity.Spec // materializes after this read still gets its signal, so the poll // loop finds it and stops it if its path no longer wants it. // - // A pending path keeps the run's own intent: its dispatch is - // re-sent rather than abandoned. + // A cancelling path therefore has nothing this run must wait for, + // and is marked cancelled here; nothing else would ever finish the + // job, and left cancelling it would hold a budget slot forever. The + // slot may be released a few seconds before a mid-flight build + // actually stops — a transient budget overshoot on a build already + // being killed. A pending path keeps the run's own intent: its + // dispatch is re-sent rather than abandoned. + if entry.Status == entity.SpeculationPathStatusCancelling { + entry.Status = entity.SpeculationPathStatusCancelled + changed = true + metrics.NamedCounter(c.metricsScope, opName, "path_cancelled_undispatched", 1) + } continue } if err != nil { @@ -228,49 +238,20 @@ func terminalPathStatus(status entity.BuildStatus) entity.SpeculationPathStatus } } -// cancelBrokenPaths marks cancelling, across every head in the snapshot, each -// path with a broken assumption. Such a path can never merge its head, so this -// is a fact, not a choice — and folding it in before the Speculator is asked -// keeps it from proposing work on top of paths that are already dead, which -// check would only throw away. -func (c *Controller) cancelBrokenPaths(snap *snapshot) { - nowMs := time.Now().UnixMilli() - - for _, batch := range snap.speculating { - set, exists := snap.pathSets[batch.ID] - if !exists { - continue - } - if cancelBrokenPathsInSet(&set, *snap, nowMs) { - snap.pathSets[batch.ID] = set - snap.markDirty(batch.ID) - } - } -} - -// cancelBrokenPathsInSet marks cancelling every live path in one set with a -// broken assumption, and reports whether anything changed. Cancelling rather -// than cancelled, because the path's build may still be occupying CI — only -// the signal that sees it stop can call it terminal. -func cancelBrokenPathsInSet(set *entity.SpeculationPathSet, snap snapshot, nowMs int64) bool { - changed := false - for i := range set.Paths { - entry := &set.Paths[i] - if entry.Status.IsTerminal() || entry.Status == entity.SpeculationPathStatusCancelling { - continue - } - if !assumptionBroken(entry.Path, snap) { - continue - } - entry.Status = entity.SpeculationPathStatusCancelling - entry.UpdatedAtMs = nowMs - changed = true - } - return changed -} - // 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. +// +// Passing foreign sets cannot widen what gets proposed: a path ID hashes its +// head, and check rejects any proposal aimed at a head that is not +// speculating. 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 { @@ -279,7 +260,7 @@ func (c *Controller) ask(ctx context.Context, queue string, snap snapshot) ([]en } sets := make([]entity.SpeculationPathSet, 0, len(snap.pathSets)) - for _, batch := range snap.speculating { + for _, batch := range snap.inFlight { if set, ok := snap.pathSets[batch.ID]; ok { sets = append(sets, set) } diff --git a/submitqueue/orchestrator/controller/speculate/run_test.go b/submitqueue/orchestrator/controller/speculate/run_test.go index ab42d303..62f3dc29 100644 --- a/submitqueue/orchestrator/controller/speculate/run_test.go +++ b/submitqueue/orchestrator/controller/speculate/run_test.go @@ -51,6 +51,22 @@ func (s *scriptedSpeculator) Speculate(_ context.Context, batches []entity.Batch return s.proposals, s.err } +// updateTo matches a BatchStore.Update argument by its ID and target state — +// the two things an outcome write is asserted on. +type updateTo struct { + id string + state entity.BatchState +} + +func (m updateTo) Matches(x any) bool { + b, ok := x.(entity.Batch) + return ok && b.ID == m.id && b.State == m.state +} + +func (m updateTo) String() string { + return fmt.Sprintf("batch %s updated to state %s", m.id, m.state) +} + type runHarness struct { controller *Controller batches *storagemock.MockBatchStore @@ -59,6 +75,30 @@ type runHarness struct { builds *storagemock.MockBuildStore spec *scriptedSpeculator published []string + // failTopic, when set, makes every publish to that topic fail. + failTopic string +} + +// failPublishTo makes publishes to one topic fail, leaving the others working, +// so a test can isolate the recovery path for a single lost publish. +func (h *runHarness) failPublishTo(topic string) { + h.failTopic = topic +} + +// speculatedOver returns the IDs of the heads the Speculator was offered. +func (h *runHarness) speculatedOver() []string { + ids := make([]string, 0, len(h.spec.gotBatches)) + for _, b := range h.spec.gotBatches { + ids = append(ids, b.ID) + } + return ids +} + +// run drives a run whose triggering message names triggerID. Which batch that +// is only matters for recovery: a retry of that message revisits that batch and +// no other, so it is the one batch that needs no recovery signal of its own. +func (h *runHarness) run(triggerID string) error { + return h.controller.run(context.Background(), entity.Batch{ID: triggerID, Queue: "q"}) } // newRunHarness wires a controller whose queue read returns inFlight. @@ -85,6 +125,9 @@ func newRunHarness(t *testing.T, ctrl *gomock.Controller, spec *scriptedSpeculat pub := queuemock.NewMockPublisher(ctrl) pub.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn( func(_ context.Context, topic string, msg entityqueue.Message) error { + if topic == h.failTopic { + return assert.AnError + } h.published = append(h.published, topic) return nil }, @@ -96,6 +139,7 @@ func newRunHarness(t *testing.T, ctrl *gomock.Controller, spec *scriptedSpeculat {Key: topickey.TopicKeyBuild, Name: "build", Queue: q}, {Key: topickey.TopicKeyMerge, Name: "submitqueue-merge", Queue: q}, {Key: topickey.TopicKeyConclude, Name: "conclude", Queue: q}, + {Key: topickey.TopicKeySpeculate, Name: "speculate", Queue: q}, }) require.NoError(t, err) @@ -144,7 +188,7 @@ func TestRun_FundsProposedPath(t *testing.T) { return nil }) - require.NoError(t, h.controller.run(context.Background(), "q")) + require.NoError(t, h.run(head)) assert.Equal(t, []string{"build"}, h.published) } @@ -163,7 +207,7 @@ func TestRun_PassesSnapshotToSpeculator(t *testing.T) { h.pathSets.EXPECT().Get(gomock.Any(), head).Return(existing, nil) h.pathSets.EXPECT().Get(gomock.Any(), merging.ID).Return(entity.SpeculationPathSet{}, storage.ErrNotFound) - require.NoError(t, h.controller.run(context.Background(), "q")) + require.NoError(t, h.run(head)) require.Equal(t, 1, spec.calls) require.Len(t, spec.gotBatches, 1) @@ -197,7 +241,7 @@ func TestRun_CancelsBrokenPath(t *testing.T) { return nil }) - require.NoError(t, h.controller.run(context.Background(), "q")) + require.NoError(t, h.run(head)) assert.Empty(t, h.published, "a cancelling path needs no dispatch; the poll loop reads the stop off the set") } @@ -218,7 +262,7 @@ func TestRun_DropsRejectedProposal(t *testing.T) { h.pathSets.EXPECT().Get(gomock.Any(), head).Return(entity.SpeculationPathSet{}, storage.ErrNotFound) // No Create and no Update: nothing was accepted. - require.NoError(t, h.controller.run(context.Background(), "q")) + require.NoError(t, h.run(head)) assert.Empty(t, h.published) } @@ -239,7 +283,7 @@ func TestRun_AlreadyFundedPathIsNotRefunded(t *testing.T) { }, nil) // No Update: the path keeps the slot and the attempt it already has. - require.NoError(t, h.controller.run(context.Background(), "q")) + require.NoError(t, h.run(head)) // The head still has no actionable path, so nothing is dispatched. assert.Empty(t, h.published) @@ -263,7 +307,7 @@ func TestRun_RedispatchesPendingPath(t *testing.T) { }, nil) // Nothing changed, so no write — but the dispatch goes out again. - require.NoError(t, h.controller.run(context.Background(), "q")) + require.NoError(t, h.run(head)) assert.Equal(t, []string{"build"}, h.published) } @@ -286,7 +330,7 @@ func TestRun_LostCASIsNotAnError(t *testing.T) { h.pathSets.EXPECT().Update(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()). Return(storage.ErrVersionMismatch) - require.NoError(t, h.controller.run(context.Background(), "q")) + require.NoError(t, h.run(head)) assert.Empty(t, h.published, "a head whose write was lost is not dispatched on stale state") } @@ -302,7 +346,7 @@ func TestRun_SpeculatorFailure(t *testing.T) { h.batches.EXPECT().Get(gomock.Any(), dep2).Return(entity.Batch{ID: dep2, State: entity.BatchStateSpeculating}, nil) h.pathSets.EXPECT().Get(gomock.Any(), head).Return(entity.SpeculationPathSet{}, storage.ErrNotFound) - require.Error(t, h.controller.run(context.Background(), "q")) + require.Error(t, h.run(head)) } // A queue with nothing speculating never reaches the Speculator. @@ -311,7 +355,7 @@ func TestRun_NoSpeculatingBatches(t *testing.T) { spec := &scriptedSpeculator{} h := newRunHarness(t, ctrl, spec, nil) - require.NoError(t, h.controller.run(context.Background(), "q")) + require.NoError(t, h.run(head)) assert.Zero(t, spec.calls) } @@ -359,7 +403,7 @@ func TestRun_RecordsFinishedBuildsOnPaths(t *testing.T) { return nil }) - require.NoError(t, h.controller.run(context.Background(), "q")) + require.NoError(t, h.run(head)) }) } } @@ -388,7 +432,7 @@ func TestRun_BuildUpdateDoesNotOverrideCancellingIntent(t *testing.T) { // Nothing changed, so no write — and no dispatch either: the poll loop is // what keeps asking the runner to stop, not the build stage. - require.NoError(t, h.controller.run(context.Background(), "q")) + require.NoError(t, h.run(head)) assert.Empty(t, h.published) } @@ -399,7 +443,9 @@ func TestRun_DoesNotReReadFinishedPaths(t *testing.T) { spec := &scriptedSpeculator{} h := newRunHarness(t, ctrl, spec, []entity.Batch{speculatingHead()}) - h.batches.EXPECT().Get(gomock.Any(), dep1).Return(entity.Batch{ID: dep1, State: entity.BatchStateSucceeded}, nil) + // dep1 is unresolved, so the passed path cannot merge — this test is about + // observation being skipped, not about outcomes. + h.batches.EXPECT().Get(gomock.Any(), dep1).Return(entity.Batch{ID: dep1, State: entity.BatchStateSpeculating}, nil) h.batches.EXPECT().Get(gomock.Any(), dep2).Return(entity.Batch{ID: dep2, State: entity.BatchStateFailed}, nil) h.pathSets.EXPECT().Get(gomock.Any(), head).Return(entity.SpeculationPathSet{ Head: head, @@ -409,7 +455,7 @@ func TestRun_DoesNotReReadFinishedPaths(t *testing.T) { // The default pathBuilds expectation is AnyTimes, so assert on the build // store: a finished entry must not reach it. - require.NoError(t, h.controller.run(context.Background(), "q")) + require.NoError(t, h.run(head)) } // Broken paths are cancelled before the Speculator is asked, so it reasons over the @@ -433,7 +479,7 @@ func TestRun_BrokenPathsAreVisibleToTheSpeculator(t *testing.T) { }, nil) h.pathSets.EXPECT().Update(gomock.Any(), gomock.Any(), int32(2), int32(3)).Return(nil) - require.NoError(t, h.controller.run(context.Background(), "q")) + require.NoError(t, h.run(head)) require.Equal(t, 1, spec.calls) require.Len(t, spec.gotSets, 1) @@ -487,3 +533,662 @@ func TestCancelBrokenPathsInSet_LeavesFinishedPaths(t *testing.T) { }) } } + +// A head whose outcome is already decided is not offered to the Speculator. +// Asking would invite proposals for a head that is on its way out of the queue, +// and this run would fund those paths and supersede them in the same set. +func TestRun_DecidedHeadIsNotSpeculatedOn(t *testing.T) { + ctrl := gomock.NewController(t) + passed := pathOver(entity.DependencyAssumptionSucceeds, entity.DependencyAssumptionFails) + spec := &scriptedSpeculator{ + // Would be applied if the head were still offered, creating a pending + // path that supersede then cancels in the same write. + proposals: []entity.Speculation{{ + Path: pathOver(entity.DependencyAssumptionFails, entity.DependencyAssumptionFails), + Action: entity.PathActionBuild, + }}, + } + + h := newRunHarness(t, ctrl, spec, []entity.Batch{speculatingHead()}) + h.noBuildsDispatched() + // Both dependencies resolved the way the passed path assumed, so it merges. + h.batches.EXPECT().Get(gomock.Any(), dep1).Return(entity.Batch{ID: dep1, State: entity.BatchStateSucceeded}, nil) + h.batches.EXPECT().Get(gomock.Any(), dep2).Return(entity.Batch{ID: dep2, State: entity.BatchStateFailed}, nil) + h.pathSets.EXPECT().Get(gomock.Any(), head).Return(entity.SpeculationPathSet{ + Head: head, + Paths: []entity.SpeculationPathEntry{entryFor(passed, entity.SpeculationPathStatusPassed)}, + Version: 1, + }, nil) + + // The head moves to merging. Its set is untouched: the only path is the + // winner, and no proposal was applied. + h.batches.EXPECT(). + Update(gomock.Any(), updateTo{id: head, state: entity.BatchStateMerging}, int32(1), int32(2)). + Return(nil) + + require.NoError(t, h.run(head)) + + assert.Zero(t, spec.calls, "a decided head leaves nothing to speculate about") + assert.Equal(t, []string{"submitqueue-merge"}, h.published) +} + +// The churn this ordering removes: a mergeable head must not gain a funded path +// that the same run immediately cancels — the dispatch would have started CI +// for work nothing waits for. +func TestRun_MergeableHeadGainsNoNewPath(t *testing.T) { + ctrl := gomock.NewController(t) + passed := pathOver(entity.DependencyAssumptionSucceeds, entity.DependencyAssumptionFails) + other := pathOver(entity.DependencyAssumptionFails, entity.DependencyAssumptionFails) + spec := &scriptedSpeculator{ + proposals: []entity.Speculation{{Path: other, Action: entity.PathActionBuild}}, + } + + h := newRunHarness(t, ctrl, spec, []entity.Batch{speculatingHead()}) + h.noBuildsDispatched() + h.batches.EXPECT().Get(gomock.Any(), dep1).Return(entity.Batch{ID: dep1, State: entity.BatchStateSucceeded}, nil) + h.batches.EXPECT().Get(gomock.Any(), dep2).Return(entity.Batch{ID: dep2, State: entity.BatchStateFailed}, nil) + + // A sibling is still building, so supersede has something to cancel and the + // set is written — which is exactly where a stray funded path would show up. + sibling := entryFor(other, entity.SpeculationPathStatusBuilding) + sibling.ID = "sibling-path" + h.pathSets.EXPECT().Get(gomock.Any(), head).Return(entity.SpeculationPathSet{ + Head: head, + Paths: []entity.SpeculationPathEntry{ + entryFor(passed, entity.SpeculationPathStatusPassed), + sibling, + }, + Version: 1, + }, nil) + + h.pathSets.EXPECT().Update(gomock.Any(), gomock.Any(), int32(1), int32(2)). + DoAndReturn(func(_ context.Context, s entity.SpeculationPathSet, _, _ int32) error { + require.Len(t, s.Paths, 2, "no path was funded for a head that is merging") + assert.Equal(t, entity.SpeculationPathStatusPassed, s.Paths[0].Status) + assert.Equal(t, entity.SpeculationPathStatusCancelling, s.Paths[1].Status) + return nil + }) + h.batches.EXPECT(). + Update(gomock.Any(), updateTo{id: head, state: entity.BatchStateMerging}, int32(1), int32(2)). + Return(nil) + + require.NoError(t, h.run(head)) + assert.Zero(t, spec.calls) +} + +// A head with no future left fails, and the write order is what keeps conclude +// usable: conclude reconciles requests from the batch's state and rejects a +// non-terminal one, so it is published only once the terminal write has landed. +func TestRun_FailedHeadConcludesAfterTheStateWrite(t *testing.T) { + ctrl := gomock.NewController(t) + spec := &scriptedSpeculator{} + failed := pathOver(entity.DependencyAssumptionSucceeds, entity.DependencyAssumptionSucceeds) + + h := newRunHarness(t, ctrl, spec, []entity.Batch{speculatingHead()}) + h.noBuildsDispatched() + h.batches.EXPECT().Get(gomock.Any(), dep1).Return(entity.Batch{ID: dep1, State: entity.BatchStateSucceeded}, nil) + h.batches.EXPECT().Get(gomock.Any(), dep2).Return(entity.Batch{ID: dep2, State: entity.BatchStateSucceeded}, nil) + h.pathSets.EXPECT().Get(gomock.Any(), head).Return(entity.SpeculationPathSet{ + Head: head, + Paths: []entity.SpeculationPathEntry{entryFor(failed, entity.SpeculationPathStatusFailed)}, + Version: 1, + }, nil) + + var publishedBeforeWrite []string + h.batches.EXPECT(). + Update(gomock.Any(), updateTo{id: head, state: entity.BatchStateFailed}, int32(1), int32(2)). + DoAndReturn(func(context.Context, entity.Batch, int32, int32) error { + publishedBeforeWrite = append([]string(nil), h.published...) + return nil + }) + + require.NoError(t, h.run(head)) + assert.Equal(t, []string{"conclude"}, h.published) + assert.Empty(t, publishedBeforeWrite, + "conclude rejects a non-terminal batch, so it must not be published before the write") +} + +// A failure resolves a dependency, which can fail everything stacked on it. The +// whole cascade finalizes inside one run: a single pass would only catch the +// dependents that happen to come later in queue order, and the rest would wait +// for an unrelated signal to wake the queue again. +func TestRun_FailureCascadesWithinOneRun(t *testing.T) { + ctrl := gomock.NewController(t) + spec := &scriptedSpeculator{} + + // dependent is listed first, so a single forward pass would evaluate it + // before the head it is stacked on has failed. + dependent := entity.Batch{ + ID: "q/batch/dependent", Queue: "q", State: entity.BatchStateSpeculating, + Dependencies: []string{head}, Version: 1, + } + failing := entity.Batch{ + ID: head, Queue: "q", State: entity.BatchStateSpeculating, Version: 1, + } + + h := newRunHarness(t, ctrl, spec, []entity.Batch{dependent, failing}) + h.noBuildsDispatched() + + failingPath := entity.SpeculationPath{Head: head} + h.pathSets.EXPECT().Get(gomock.Any(), head).Return(entity.SpeculationPathSet{ + Head: head, + Paths: []entity.SpeculationPathEntry{entryFor(failingPath, entity.SpeculationPathStatusFailed)}, + Version: 1, + }, nil) + + // The dependent bet that its dependency would fail and built on that, and + // that build failed. The bet is vindicated rather than broken the moment + // the dependency fails, which leaves the dependent with a live path that + // cannot pass — so it can only be decided by the cascade, and only once the + // dependency is terminal. + dependentPath := entity.SpeculationPath{ + Head: dependent.ID, + Dependencies: []entity.PathDependency{{Batch: head, Assumption: entity.DependencyAssumptionFails}}, + } + h.pathSets.EXPECT().Get(gomock.Any(), dependent.ID).Return(entity.SpeculationPathSet{ + Head: dependent.ID, + Paths: []entity.SpeculationPathEntry{entryFor(dependentPath, entity.SpeculationPathStatusFailed)}, + Version: 1, + }, nil) + + h.batches.EXPECT(). + Update(gomock.Any(), updateTo{id: head, state: entity.BatchStateFailed}, int32(1), int32(2)).Return(nil) + h.batches.EXPECT(). + Update(gomock.Any(), updateTo{id: dependent.ID, state: entity.BatchStateFailed}, int32(1), int32(2)).Return(nil) + h.pathSets.EXPECT().Update(gomock.Any(), gomock.Any(), int32(1), int32(2)).Return(nil).AnyTimes() + + require.NoError(t, h.run(head)) + assert.Zero(t, spec.calls, "no head is left open, so there is nothing to speculate about") +} + +// A cancelling batch is driven to terminal by the run like any other batch, +// off the same read: its paths are marked stopped, and once they all are the +// batch is cancelled and its requests reconciled. +func TestRun_CancellingBatchFinalizesInTheRun(t *testing.T) { + ctrl := gomock.NewController(t) + spec := &scriptedSpeculator{} + + cancelling := entity.Batch{ + ID: head, Queue: "q", State: entity.BatchStateCancelling, Version: 1, + } + + h := newRunHarness(t, ctrl, spec, []entity.Batch{cancelling}) + h.pathSets.EXPECT().Get(gomock.Any(), head).Return(entity.SpeculationPathSet{ + Head: head, + Paths: []entity.SpeculationPathEntry{ + {ID: "p1", Status: entity.SpeculationPathStatusCancelling, Attempt: 1}, + }, + Version: 2, + }, nil) + // The build stages record what CI did on the Build row alone; folding that + // in is what lets the run see the path has actually stopped. + h.pathBuilds.EXPECT().Get(gomock.Any(), "p1", 1). + Return(entity.PathBuild{PathID: "p1", Attempt: 1, BuildID: "b1"}, nil) + h.builds.EXPECT().Get(gomock.Any(), "b1"). + Return(entity.Build{ID: "b1", Status: entity.BuildStatusCancelled}, nil) + h.pathSets.EXPECT().Update(gomock.Any(), gomock.Any(), int32(2), int32(3)). + DoAndReturn(func(_ context.Context, s entity.SpeculationPathSet, _, _ int32) error { + assert.Equal(t, entity.SpeculationPathStatusCancelled, s.Paths[0].Status) + return nil + }) + h.batches.EXPECT(). + Update(gomock.Any(), updateTo{id: head, state: entity.BatchStateCancelled}, int32(1), int32(2)).Return(nil) + + require.NoError(t, h.run(head)) + assert.Equal(t, []string{"conclude"}, h.published) +} + +// A cancelling batch whose builds are still running is not terminal yet. Its +// paths are marked cancelling — which is all the poll loop needs to stop them +// — and the batch waits: marking it cancelled here would declare it stopped +// while CI still held its slots. +func TestRun_CancellingBatchWaitsForBuildsToStop(t *testing.T) { + ctrl := gomock.NewController(t) + spec := &scriptedSpeculator{} + + cancelling := entity.Batch{ + ID: head, Queue: "q", State: entity.BatchStateCancelling, Version: 1, + } + + h := newRunHarness(t, ctrl, spec, []entity.Batch{cancelling}) + h.pathSets.EXPECT().Get(gomock.Any(), head).Return(entity.SpeculationPathSet{ + Head: head, + Paths: []entity.SpeculationPathEntry{ + {ID: "p1", Status: entity.SpeculationPathStatusBuilding, Attempt: 1}, + }, + Version: 2, + }, nil) + h.pathBuilds.EXPECT().Get(gomock.Any(), "p1", 1). + Return(entity.PathBuild{PathID: "p1", Attempt: 1, BuildID: "b1"}, nil) + h.builds.EXPECT().Get(gomock.Any(), "b1"). + Return(entity.Build{ID: "b1", Status: entity.BuildStatusRunning}, nil) + h.pathSets.EXPECT().Update(gomock.Any(), gomock.Any(), int32(2), int32(3)). + DoAndReturn(func(_ context.Context, s entity.SpeculationPathSet, _, _ int32) error { + assert.Equal(t, entity.SpeculationPathStatusCancelling, s.Paths[0].Status) + return nil + }) + + // No batch Update: the batch is not terminal yet. And no publish: + // stopping the running build is the poll loop's job, not a dispatch. + require.NoError(t, h.run(head)) + assert.Empty(t, h.published) +} + +// A path cancelled before its build was ever dispatched has nothing to stop: +// no link means no build the poll loop could ever be watching. Left alone the +// path would hold a budget slot and keep its batch out of a terminal state +// forever, so the run marks it cancelled from that absence. +func TestRun_CancellingPathWithNoBuildIsCancelled(t *testing.T) { + ctrl := gomock.NewController(t) + spec := &scriptedSpeculator{} + + cancelling := entity.Batch{ + ID: head, Queue: "q", State: entity.BatchStateCancelling, Version: 1, + } + + h := newRunHarness(t, ctrl, spec, []entity.Batch{cancelling}) + h.noBuildsDispatched() + h.pathSets.EXPECT().Get(gomock.Any(), head).Return(entity.SpeculationPathSet{ + Head: head, + Paths: []entity.SpeculationPathEntry{ + {ID: "p1", Status: entity.SpeculationPathStatusCancelling, Attempt: 1}, + }, + Version: 2, + }, nil) + h.pathSets.EXPECT().Update(gomock.Any(), gomock.Any(), int32(2), int32(3)). + DoAndReturn(func(_ context.Context, s entity.SpeculationPathSet, _, _ int32) error { + assert.Equal(t, entity.SpeculationPathStatusCancelled, s.Paths[0].Status) + return nil + }) + h.batches.EXPECT(). + Update(gomock.Any(), updateTo{id: head, state: entity.BatchStateCancelled}, int32(1), int32(2)).Return(nil) + + require.NoError(t, h.run(head)) + assert.Equal(t, []string{"conclude"}, h.published) +} + +// A batch cancelled before anything was funded has no builds to wait on. +func TestRun_CancellingBatchWithNoPaths(t *testing.T) { + ctrl := gomock.NewController(t) + spec := &scriptedSpeculator{} + + cancelling := entity.Batch{ + ID: head, Queue: "q", State: entity.BatchStateCancelling, Version: 1, + } + + h := newRunHarness(t, ctrl, spec, []entity.Batch{cancelling}) + h.pathSets.EXPECT().Get(gomock.Any(), head).Return(entity.SpeculationPathSet{}, storage.ErrNotFound) + h.batches.EXPECT(). + Update(gomock.Any(), updateTo{id: head, state: entity.BatchStateCancelled}, int32(1), int32(2)).Return(nil) + + require.NoError(t, h.run(head)) + assert.Equal(t, []string{"conclude"}, h.published) +} + +// Losing the path-set race must not let the terminal write proceed off a view +// another writer has already replaced — the batch could still be holding a +// build the winning write knows about. +func TestRun_CancellingLostPathSetRaceSkipsTheTerminalWrite(t *testing.T) { + ctrl := gomock.NewController(t) + spec := &scriptedSpeculator{} + + cancelling := entity.Batch{ + ID: head, Queue: "q", State: entity.BatchStateCancelling, Version: 1, + } + + h := newRunHarness(t, ctrl, spec, []entity.Batch{cancelling}) + h.noBuildsDispatched() + h.pathSets.EXPECT().Get(gomock.Any(), head).Return(entity.SpeculationPathSet{ + Head: head, + Paths: []entity.SpeculationPathEntry{ + {ID: "p1", Status: entity.SpeculationPathStatusCancelling, Attempt: 1}, + }, + Version: 2, + }, nil) + h.pathSets.EXPECT().Update(gomock.Any(), gomock.Any(), int32(2), int32(3)). + Return(storage.ErrVersionMismatch) + + // No batch Update and no conclude: the next run re-reads and re-decides. + require.NoError(t, h.run(head)) + assert.Empty(t, h.published) +} + +// The allocator rations a budget measured in occupied CI slots, and a path +// holds its slot until its build actually stops. A merging head's superseded +// siblings are still running, so hiding their set would let the allocator count +// those slots as free and oversubscribe CI. +func TestRun_SpeculatorSeesPathSetsOfNonOpenHeads(t *testing.T) { + ctrl := gomock.NewController(t) + spec := &scriptedSpeculator{} + + merging := entity.Batch{ + ID: "q/batch/merging", Queue: "q", State: entity.BatchStateMerging, Version: 1, + } + open := entity.Batch{ID: head, Queue: "q", State: entity.BatchStateSpeculating, Version: 1} + + h := newRunHarness(t, ctrl, spec, []entity.Batch{open, merging}) + h.noBuildsDispatched() + h.pathSets.EXPECT().Get(gomock.Any(), head). + Return(entity.SpeculationPathSet{Head: head, Version: 1}, nil) + h.pathSets.EXPECT().Get(gomock.Any(), merging.ID).Return(entity.SpeculationPathSet{ + Head: merging.ID, + Paths: []entity.SpeculationPathEntry{ + {ID: "still-running", Status: entity.SpeculationPathStatusCancelling, Attempt: 1}, + }, + Version: 1, + }, nil) + // The undispatched cancelling path is marked cancelled, so the merging head's set is + // rewritten even though it is closed to new work. + h.pathSets.EXPECT().Update(gomock.Any(), gomock.Any(), int32(1), int32(2)). + DoAndReturn(func(_ context.Context, s entity.SpeculationPathSet, _, _ int32) error { + assert.Equal(t, merging.ID, s.Head) + assert.Equal(t, entity.SpeculationPathStatusCancelled, s.Paths[0].Status) + return nil + }) + + require.NoError(t, h.run(head)) + + assert.Equal(t, []entity.Batch{open}, spec.gotBatches, + "only an open head may be an action target") + require.Len(t, spec.gotSets, 2, "every in-flight path set counts against the budget") + assert.Equal(t, merging.ID, spec.gotSets[1].Head) +} + +// A queue whose only in-flight head is closed to new work still has to be +// written: the build stages record what CI did on the Build row alone, and this +// is the only writer that can finish the path it belongs to. Left unwritten, +// the path would keep charging the budget. +func TestRun_PersistsObservationsWithNoOpenHead(t *testing.T) { + ctrl := gomock.NewController(t) + spec := &scriptedSpeculator{} + + merging := entity.Batch{ + ID: "q/batch/merging", Queue: "q", State: entity.BatchStateMerging, Version: 1, + } + + h := newRunHarness(t, ctrl, spec, []entity.Batch{merging}) + h.pathSets.EXPECT().Get(gomock.Any(), merging.ID).Return(entity.SpeculationPathSet{ + Head: merging.ID, + Paths: []entity.SpeculationPathEntry{ + {ID: "p1", Status: entity.SpeculationPathStatusCancelling, Attempt: 1}, + }, + Version: 3, + }, nil) + h.pathBuilds.EXPECT().Get(gomock.Any(), "p1", 1). + Return(entity.PathBuild{PathID: "p1", Attempt: 1, BuildID: "b1"}, nil) + h.builds.EXPECT().Get(gomock.Any(), "b1"). + Return(entity.Build{ID: "b1", Status: entity.BuildStatusCancelled}, nil) + h.pathSets.EXPECT().Update(gomock.Any(), gomock.Any(), int32(3), int32(4)). + DoAndReturn(func(_ context.Context, s entity.SpeculationPathSet, _, _ int32) error { + assert.Equal(t, entity.SpeculationPathStatusCancelled, s.Paths[0].Status) + return nil + }) + + require.NoError(t, h.run(head)) + assert.Zero(t, spec.calls, "no head is open, so there is nothing to speculate about") + h.noBuildsDispatched() +} + +// Repeat publishes for one batch must reach the queue. It deduplicates on +// (topic, partition key, message ID) against rows it has not collected yet, +// consumed ones included, so a bare batch ID would silently drop the re-sends +// this controller relies on. +func TestPublish_MintsADistinctMessageIDPerPublish(t *testing.T) { + ctrl := gomock.NewController(t) + + var ids []string + pub := queuemock.NewMockPublisher(ctrl) + pub.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn( + func(_ context.Context, _ string, msg entityqueue.Message) error { + ids = append(ids, msg.ID) + return nil + }, + ).Times(2) + q := queuemock.NewMockQueue(ctrl) + q.EXPECT().Publisher().Return(pub).AnyTimes() + + registry, err := consumer.NewTopicRegistry([]consumer.TopicConfig{ + {Key: topickey.TopicKeyConclude, Name: "conclude", Queue: q}, + }) + require.NoError(t, err) + + c := NewController( + zaptest.NewLogger(t).Sugar(), tally.NoopScope, storagemock.NewMockStorage(ctrl), + staticSpeculatorFactory{}, registry, topickey.TopicKeySpeculate, "orchestrator-speculate", + ) + + require.NoError(t, c.publishBatchID(context.Background(), topickey.TopicKeyConclude, head, "q")) + require.NoError(t, c.publishBatchID(context.Background(), topickey.TopicKeyConclude, head, "q")) + + require.Len(t, ids, 2) + assert.NotEqual(t, ids[0], ids[1]) +} + +// cascadePair wires a queue where `prerequisite` reaches a terminal outcome +// and `derived` fails only because of it: derived bet that prerequisite would +// not succeed, built on that, and its build failed. Until prerequisite is terminal the +// bet is unresolved and derived waits; once it is, derived has a live path that +// cannot pass. +func cascadePair(t *testing.T, ctrl *gomock.Controller, prerequisiteState entity.BatchState) (*runHarness, entity.Batch, entity.Batch) { + t.Helper() + + prerequisite := entity.Batch{ID: head, Queue: "q", State: prerequisiteState, Version: 1} + derived := entity.Batch{ + ID: "q/batch/derived", Queue: "q", State: entity.BatchStateSpeculating, + Dependencies: []string{head}, Version: 1, + } + + h := newRunHarness(t, ctrl, &scriptedSpeculator{}, []entity.Batch{prerequisite, derived}) + + derivedPath := entity.SpeculationPath{ + Head: derived.ID, + Dependencies: []entity.PathDependency{{Batch: head, Assumption: entity.DependencyAssumptionFails}}, + } + h.pathSets.EXPECT().Get(gomock.Any(), derived.ID).Return(entity.SpeculationPathSet{ + Head: derived.ID, + Paths: []entity.SpeculationPathEntry{entryFor(derivedPath, entity.SpeculationPathStatusFailed)}, + Version: 1, + }, nil) + + return h, prerequisite, derived +} + +// An outcome is only a fact once it commits. When the prerequisite's state write +// loses its race, nothing derived from that outcome may be enacted — the winner +// may have written something else entirely. A cancellation loses precisely to a +// merge that got there first, which leaves the batch succeeded, and a dependent +// broken by that success must not already have been failed on the assumption +// it was cancelled. +func TestRun_CascadeStopsWhenThePrerequisiteStateCASLoses(t *testing.T) { + ctrl := gomock.NewController(t) + h, prerequisite, derived := cascadePair(t, ctrl, entity.BatchStateCancelling) + h.noBuildsDispatched() + + h.pathSets.EXPECT().Get(gomock.Any(), prerequisite.ID). + Return(entity.SpeculationPathSet{}, storage.ErrNotFound) + h.batches.EXPECT(). + Update(gomock.Any(), updateTo{id: prerequisite.ID, state: entity.BatchStateCancelled}, int32(1), int32(2)). + Return(storage.ErrVersionMismatch) + + // No Update for the dependent: gomock fails the test if one happens. + require.NoError(t, h.run(head)) + assert.Empty(t, h.published, "nothing may be concluded off an outcome that did not commit") + assert.Contains(t, h.speculatedOver(), derived.ID, + "the dependency is unresolved in committed state, so the dependent is still open") +} + +// The path set and the outcome are one decision. Losing the set's race means +// another writer has moved the head on, so the outcome read off our copy is not +// enacted — and neither is anything derived from it. +func TestRun_CascadeStopsWhenThePrerequisitePathSetCASLoses(t *testing.T) { + ctrl := gomock.NewController(t) + h, prerequisite, _ := cascadePair(t, ctrl, entity.BatchStateCancelling) + h.noBuildsDispatched() + + h.pathSets.EXPECT().Get(gomock.Any(), prerequisite.ID).Return(entity.SpeculationPathSet{ + Head: prerequisite.ID, + Paths: []entity.SpeculationPathEntry{ + {ID: "p1", Status: entity.SpeculationPathStatusCancelling, Attempt: 1}, + }, + Version: 2, + }, nil) + h.pathSets.EXPECT().Update(gomock.Any(), gomock.Any(), int32(2), int32(3)). + Return(storage.ErrVersionMismatch) + + // No batch Update at all: not for the prerequisite, not for the dependent. + require.NoError(t, h.run(head)) + assert.Empty(t, h.published) +} + +// The abort is scoped to what actually depended on the lost outcome. A batch +// this run decided on its own evidence still commits. +func TestRun_UnrelatedOutcomeStillCommitsAfterALostCAS(t *testing.T) { + ctrl := gomock.NewController(t) + + losing := entity.Batch{ID: head, Queue: "q", State: entity.BatchStateCancelling, Version: 1} + independent := entity.Batch{ + ID: "q/batch/independent", Queue: "q", State: entity.BatchStateSpeculating, Version: 1, + } + + h := newRunHarness(t, ctrl, &scriptedSpeculator{}, []entity.Batch{losing, independent}) + h.noBuildsDispatched() + + h.pathSets.EXPECT().Get(gomock.Any(), losing.ID). + Return(entity.SpeculationPathSet{}, storage.ErrNotFound) + h.batches.EXPECT(). + Update(gomock.Any(), updateTo{id: losing.ID, state: entity.BatchStateCancelled}, int32(1), int32(2)). + Return(storage.ErrVersionMismatch) + + // The independent head has no dependencies and one failed path, so its + // outcome rests on nothing this run recorded. + ownPath := entity.SpeculationPath{Head: independent.ID} + h.pathSets.EXPECT().Get(gomock.Any(), independent.ID).Return(entity.SpeculationPathSet{ + Head: independent.ID, + Paths: []entity.SpeculationPathEntry{entryFor(ownPath, entity.SpeculationPathStatusFailed)}, + Version: 1, + }, nil) + h.batches.EXPECT(). + Update(gomock.Any(), updateTo{id: independent.ID, state: entity.BatchStateFailed}, int32(1), int32(2)). + Return(nil) + + require.NoError(t, h.run(head)) + assert.Equal(t, []string{"speculate", "conclude"}, h.published, + "the independent batch is not the one on the message, so it is given a "+ + "signal naming it before it is made terminal") +} + +// A batch decided inside the run has its set written once, by the same step +// that enacts its outcome — the dispatch step must not come along afterwards and +// try again against the version it has just superseded. +func TestRun_FinalizedBatchPathSetIsWrittenOnce(t *testing.T) { + ctrl := gomock.NewController(t) + + cancelling := entity.Batch{ID: head, Queue: "q", State: entity.BatchStateCancelling, Version: 1} + h := newRunHarness(t, ctrl, &scriptedSpeculator{}, []entity.Batch{cancelling}) + h.noBuildsDispatched() + + h.pathSets.EXPECT().Get(gomock.Any(), head).Return(entity.SpeculationPathSet{ + Head: head, + Paths: []entity.SpeculationPathEntry{ + {ID: "p1", Status: entity.SpeculationPathStatusCancelling, Attempt: 1}, + }, + Version: 2, + }, nil) + h.pathSets.EXPECT().Update(gomock.Any(), gomock.Any(), int32(2), int32(3)).Return(nil).Times(1) + h.batches.EXPECT(). + Update(gomock.Any(), updateTo{id: head, state: entity.BatchStateCancelled}, int32(1), int32(2)).Return(nil) + + require.NoError(t, h.run(head)) +} + +// A batch this run drove terminal cannot always be repaired by replaying the +// message. The retry names whichever batch woke the run, and once terminal this +// one is gone from the queue listing — so if it was decided by a cascade, +// neither the retry nor the dead-letter fan-out will ever name it. +// +// It is therefore given a message of its own, and given it *before* the state +// write. A signal sent afterwards is one more thing that can fail at the moment +// everything else is failing, which is exactly when it is needed; sent first, a +// failure means nothing was written and the retry re-derives the lot. +func TestRun_CascadeDerivedBatchIsGivenARecoverySignalBeforeItIsTerminal(t *testing.T) { + ctrl := gomock.NewController(t) + + failing := entity.Batch{ID: "q/batch/derived", Queue: "q", State: entity.BatchStateSpeculating, Version: 1} + h := newRunHarness(t, ctrl, &scriptedSpeculator{}, []entity.Batch{failing}) + h.noBuildsDispatched() + + failedPath := entity.SpeculationPath{Head: failing.ID} + h.pathSets.EXPECT().Get(gomock.Any(), failing.ID).Return(entity.SpeculationPathSet{ + Head: failing.ID, + Paths: []entity.SpeculationPathEntry{entryFor(failedPath, entity.SpeculationPathStatusFailed)}, + Version: 1, + }, nil) + + var publishedBeforeWrite []string + h.batches.EXPECT(). + Update(gomock.Any(), updateTo{id: failing.ID, state: entity.BatchStateFailed}, int32(1), int32(2)). + DoAndReturn(func(context.Context, entity.Batch, int32, int32) error { + publishedBeforeWrite = append([]string(nil), h.published...) + return nil + }) + + // The message names some other batch, so a retry would never come back here. + require.NoError(t, h.run(head)) + + assert.Equal(t, []string{"speculate"}, publishedBeforeWrite, + "the signal has to exist before the terminal state it recovers") + assert.Equal(t, []string{"speculate", "conclude"}, h.published) +} + +// The batch on the message needs no signal of its own: a retry re-reads it, +// finds it terminal, and re-publishes from the self-heal branch, and a message +// that dead-letters already names the batch the fan-out there repairs. +func TestRun_TriggerBatchNeedsNoRecoverySignal(t *testing.T) { + ctrl := gomock.NewController(t) + + failing := entity.Batch{ID: head, Queue: "q", State: entity.BatchStateSpeculating, Version: 1} + h := newRunHarness(t, ctrl, &scriptedSpeculator{}, []entity.Batch{failing}) + h.noBuildsDispatched() + + failedPath := entity.SpeculationPath{Head: head} + h.pathSets.EXPECT().Get(gomock.Any(), head).Return(entity.SpeculationPathSet{ + Head: head, + Paths: []entity.SpeculationPathEntry{entryFor(failedPath, entity.SpeculationPathStatusFailed)}, + Version: 1, + }, nil) + h.batches.EXPECT(). + Update(gomock.Any(), updateTo{id: head, state: entity.BatchStateFailed}, int32(1), int32(2)).Return(nil) + + require.NoError(t, h.run(head)) + assert.Equal(t, []string{"conclude"}, h.published) +} + +// A cancelling path whose build is still running finishes only when CI actually +// stops. The run writes nothing (the intent is already recorded), publishes +// nothing (the poll loop is what keeps asking the runner to stop), and the +// batch stays out of terminal until the stop is observed. +func TestRun_CancellingPathWithALiveBuildStaysCancelling(t *testing.T) { + ctrl := gomock.NewController(t) + + cancelling := entity.Batch{ID: head, Queue: "q", State: entity.BatchStateCancelling, Version: 1} + h := newRunHarness(t, ctrl, &scriptedSpeculator{}, []entity.Batch{cancelling}) + + h.pathSets.EXPECT().Get(gomock.Any(), head).Return(entity.SpeculationPathSet{ + Head: head, + Paths: []entity.SpeculationPathEntry{{ + ID: "p1", + Status: entity.SpeculationPathStatusCancelling, + Attempt: 1, + }}, + Version: 2, + }, nil) + h.pathBuilds.EXPECT().Get(gomock.Any(), "p1", 1). + Return(entity.PathBuild{PathID: "p1", Attempt: 1, BuildID: "b1"}, nil) + h.builds.EXPECT().Get(gomock.Any(), "b1"). + Return(entity.Build{ID: "b1", Status: entity.BuildStatusRunning}, nil) + + // No path-set write, no batch Update, and no publish: gomock and the + // published assertion fail the test if any happens. + require.NoError(t, h.run(head)) + assert.Empty(t, h.published) +} diff --git a/submitqueue/orchestrator/controller/speculate/snapshot.go b/submitqueue/orchestrator/controller/speculate/snapshot.go index cd439766..2c2dc706 100644 --- a/submitqueue/orchestrator/controller/speculate/snapshot.go +++ b/submitqueue/orchestrator/controller/speculate/snapshot.go @@ -26,17 +26,26 @@ type snapshot struct { // in-flight batches plus any finalized batch still named as a dependency // of one of them. batches map[string]entity.Batch - // speculating is the queue's Speculating batches, in queue order. These are - // the heads open to new work: what the Speculator is handed, and what the - // dispatch step walks. + // inFlight is the queue's in-flight batches in queue order, whatever their + // state. This is what the dispatch step walks: a merging or cancelling head + // is closed to new work, but its paths still hold CI slots and their + // observations still need persisting. + inFlight []entity.Batch + // trigger is the batch named on the message that woke this run. The plan + // never depends on it; it matters only for crash recovery — see + // applyOutcome. + trigger string + // speculating is the heads still open to new work: the batches that were + // Speculating at read time, minus the ones finalize has since decided. Only + // these are offered to the Speculator as action targets. speculating []entity.Batch // pathSets is each head's in-memory path set, by head batch ID. Statuses // already reflect what each path's build actually did — see // (*Controller).updatePathsFromBuilds. pathSets map[string]entity.SpeculationPathSet // dirty marks heads whose in-memory set differs from what is stored. - // Always touch it through markDirty / isDirty so every step of the - // handshake is greppable; see those methods for the contract. + // Always touch it through markDirty / markClean / isDirty so every step + // of the handshake is greppable; see those methods for the contract. dirty map[string]bool } @@ -47,12 +56,26 @@ func (s *snapshot) markDirty(id string) { s.dirty[id] = true } +// markClean records that nothing is left to flush for this head: either the +// set was persisted, or its write lost a compare-and-swap and the in-memory +// copy was abandoned (the next run re-reads the winner). +func (s *snapshot) markClean(id string) { + s.dirty[id] = false +} + // isDirty reports whether the head's set still needs persisting. A head with -// no entry is not dirty, so the unguarded map read is deliberate. +// no entry is not dirty — read loads it, finalize cleans what it wrote — so +// the unguarded map read is deliberate. func (s snapshot) isDirty(id string) bool { return s.dirty[id] } +// isTrigger reports whether the batch is the one named on the message being +// processed — the one batch a redelivery of that message would revisit. +func (s snapshot) isTrigger(id string) bool { + return s.trigger == id +} + // batchState returns a batch's state, or BatchStateUnknown for a batch the // run never read — which resolves no assumption either way. func (s snapshot) batchState(id string) entity.BatchState { diff --git a/submitqueue/orchestrator/controller/speculate/speculate.go b/submitqueue/orchestrator/controller/speculate/speculate.go index 3f2e8cfb..bd62ba12 100644 --- a/submitqueue/orchestrator/controller/speculate/speculate.go +++ b/submitqueue/orchestrator/controller/speculate/speculate.go @@ -16,7 +16,6 @@ package speculate import ( "context" - "errors" "fmt" "github.com/uber-go/tally" @@ -30,32 +29,9 @@ import ( "go.uber.org/zap" ) -// Controller handles speculate queue messages. -// -// Each message is a dirty signal: it names the batch that changed, but only so -// the queue wakes up. The controller then re-plans that whole queue from a -// single read — see run — asking the Speculator which paths are worth building -// within the budget and cancelling the ones a resolved dependency has ruled -// out. Nothing carries over between runs, so duplicated or reordered signals -// are harmless and a later run repairs whatever an earlier one left half-done. -// -// Batch verdicts are still the naive per-batch state machine below, which -// advances the triggering batch one step: -// -// - Created → admit to Speculating so the Speculator can act on it. -// - Speculating → if all deps are Succeeded, publish to merge and -// transition to Merging; otherwise no-op (or fail-fast if a dep is -// in a non-succeeding terminal state). -// - Cancelling → cancel any in-flight Build entity, respeculate -// dependents, CAS to terminal Cancelled, publish to conclude. -// - Merging → no-op (owned by the merge controller). -// - Terminal → re-fan-out to conclude for self-healing in case a -// prior publish was lost. -// -// Waiting on every dependency is strictly stricter than waiting on the ones a -// passed path assumed, so this is correct while the path-aware finalization -// that replaces it is written — it just does not yet collect the speedup the -// paths are earning. +// Controller handles speculate queue messages: each one is a dirty signal +// naming a batch whose queue should be re-planned. The package doc has the +// full model; Process below is the entry point. type Controller struct { logger *zap.SugaredLogger metricsScope tally.Scope @@ -93,8 +69,23 @@ func NewController( } } -// Process re-plans the triggering batch's queue, then advances that batch one -// step along the legacy per-batch state machine (see the package doc). +// Process re-plans the triggering batch's queue. The run does almost all of +// the work — see run.go — and the message's own batch matters in only two +// places on the way in: +// +// - Created: the batch is admitted first, which makes it visible to the +// Speculator (proposals may only target Speculating heads). Reaching an +// outcome on it in the same run is safe: a merge needs a passed path, and +// a head admitted this instant has no paths at all. +// - Already terminal: its conclude publish is repeated in case a previous +// one was lost — idempotent on the batch ID — and the run that follows is +// how dependents learn of an outcome no run has seen yet (a batch +// finalized by another stage, e.g. the merge signal recording a landed +// push, was never seen breaking the paths that bet against it). +// +// Everything else — funding paths, cancelling broken ones, driving a +// cancellation to terminal, reaching outcomes — happens inside the run, which +// covers this batch along with the rest of its queue. // Returns nil to ack (success), or error to nack (retry). func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) error { msg := delivery.Message() @@ -111,72 +102,28 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er return fmt.Errorf("failed to get batch %s: %w", bid.ID, err) } - // Cancelling intent: the cancel controller has handed this batch off to - // speculate to drive to terminal. Cancel in-flight builds, fan out to - // dependents, CAS to terminal Cancelled, and publish to conclude. - if batch.State == entity.BatchStateCancelling { - return c.cancelBatch(ctx, batch) - } - - // Terminal state: re-fan-out for self-healing in case a previous publish - // was lost. Always re-publish to conclude (idempotent on the batch ID). - // For Cancelled specifically also re-publish to dependents — a crash - // between the terminal CAS and the dependent publish would otherwise - // leave them stuck waiting on a Cancelled dep. if batch.State.IsTerminal() { metrics.NamedCounter(c.metricsScope, opName, "self_heal_terminal", 1) - if batch.State == entity.BatchStateCancelled { - if err := c.respeculateDependents(ctx, batch); err != nil { - return err - } + if err := c.fanout(ctx, batch.ID, batch.Queue); err != nil { + return err } - return c.fanout(ctx, batch.ID, batch.Queue) + return c.run(ctx, batch) } - // A Created batch is admitted before the run so the Speculator can see it: - // proposals may only target Speculating heads. - wasCreated := batch.State == entity.BatchStateCreated - if wasCreated { - admitted, err := c.admit(ctx, batch) - if err != nil { + if batch.State == entity.BatchStateCreated { + if _, err := c.admit(ctx, batch); err != nil { return err } - batch = admitted } - if err := c.run(ctx, batch.Queue); err != nil { - return err - } - - // A freshly admitted head stops here rather than falling through to the - // finalizer. The run above funded its first paths and nothing has been - // built yet — finalizing on the same message would let a batch with no - // dependencies merge before any build had run. - if wasCreated { - return nil - } - - // Merging is owned by the merge controller, which has its own self-heal. - if batch.State == entity.BatchStateMerging { - metrics.NamedCounter(c.metricsScope, opName, "noop_merging", 1) - return nil - } - - if batch.State == entity.BatchStateSpeculating { - return c.tryFinalize(ctx, batch) - } - - metrics.NamedCounter(c.metricsScope, opName, "unexpected_state", 1) - return fmt.Errorf("unexpected batch state %q for batch %s", batch.State, batch.ID) + return c.run(ctx, batch) } // admit moves a batch from Created to Speculating, which is what makes it -// visible to the Speculator as an action target. It returns the batch with the -// new state and version so the caller keeps writing against a current copy. -// -// It no longer dispatches anything itself: which paths to build for this head -// is the run's decision, taken over the whole queue rather than one batch at a -// time. +// visible to the Speculator as an action target. It returns the batch with +// the new state and version so the caller keeps writing against a current +// copy. Which paths to build for the new head is not decided here — that is +// the run's call, taken over the whole queue rather than one batch at a time. func (c *Controller) admit(ctx context.Context, batch entity.Batch) (entity.Batch, error) { newVersion := batch.Version + 1 batch.State = entity.BatchStateSpeculating @@ -196,250 +143,6 @@ func (c *Controller) admit(ctx context.Context, batch entity.Batch) (entity.Batc return batch, nil } -// tryFinalize publishes to merge and transitions to Merging iff every -// dependency batch has reached Succeeded. Cancelled deps are treated as -// out-of-the-way: the cancelled batch will never land, so it can no longer -// conflict — drop it from the chain and proceed. Failed deps still cascade -// via failOnDependency. If some deps are still in flight, the call is a -// no-op and waits for the next event. -// -// TODO: when a dependency fails we currently fail this batch outright. -// We will need to respeculate the failed paths — drop the failed dep -// from the chain and re-issue speculation for the surviving ordering(s) -// — instead of cascading the failure into requests that could still land. -func (c *Controller) tryFinalize(ctx context.Context, batch entity.Batch) error { - deps, err := c.fetchDependencies(ctx, batch) - if err != nil { - return err - } - - pending := make([]string, 0, len(deps)) - for _, d := range deps { - switch d.State { - case entity.BatchStateSucceeded: - // ok - case entity.BatchStateCancelled: - // Out-of-the-way: the cancelled batch will never land, so it can - // no longer conflict. Drop it from the chain and continue. - metrics.NamedCounter(c.metricsScope, opName, "dependency_cancelled_skipped", 1) - c.logger.Infow("dependency cancelled; dropping from speculation chain", - "batch_id", batch.ID, - "dependency_id", d.ID, - ) - case entity.BatchStateFailed: - return c.failOnDependency(ctx, batch, d) - default: - pending = append(pending, d.ID) - } - } - - if len(pending) > 0 { - metrics.NamedCounter(c.metricsScope, opName, "waiting_on_deps", 1) - c.logger.Debugw("dependencies still in flight; waiting", - "batch_id", batch.ID, - "pending_dependency_ids", pending, - ) - return nil - } - - if err := c.publishBatchID(ctx, topickey.TopicKeyMerge, batch.ID, batch.Queue); err != nil { - metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1) - return fmt.Errorf("failed to publish to merge: %w", err) - } - - newVersion := batch.Version + 1 - batch.State = entity.BatchStateMerging - if err := c.store.GetBatchStore().Update(ctx, batch, batch.Version, newVersion); err != nil { - metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) - return fmt.Errorf("failed to update batch %s state to merging: %w", batch.ID, err) - } - - return nil -} - -// failOnDependency transitions a Speculating batch to Failed when one of its -// dependencies has reached a non-succeeding terminal state, then publishes to -// the conclude queue so the request store and request log get reconciled. -// Without this transition the batch would sit in Speculating forever — no -// downstream event ever fires for it again. -func (c *Controller) failOnDependency(ctx context.Context, batch entity.Batch, dep entity.Batch) error { - metrics.NamedCounter(c.metricsScope, opName, "dependency_failed", 1) - c.logger.Warnw("dependency in non-succeeding terminal state; failing batch", - "batch_id", batch.ID, - "dependency_id", dep.ID, - "dependency_state", string(dep.State), - ) - - newVersion := batch.Version + 1 - batch.State = entity.BatchStateFailed - if err := c.store.GetBatchStore().Update(ctx, batch, batch.Version, newVersion); err != nil { - metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) - return fmt.Errorf("failed to update batch %s state to failed: %w", batch.ID, err) - } - batch.Version = newVersion - - if err := c.publishBatchID(ctx, topickey.TopicKeyConclude, batch.ID, batch.Queue); err != nil { - metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1) - return fmt.Errorf("failed to publish to conclude: %w", err) - } - - return nil -} - -// cancelBatch drives a batch from BatchStateCancelling to BatchStateCancelled. -// The cancel controller records the user's intent (Cancelling) and hands the -// batch off; speculate owns the rest because all the work that must precede -// the terminal write — flipping in-flight builds, respeculating dependents — -// already lives in the speculate domain. The terminal transition is the -// single writer of every non-Cancelling batch state across the system. -// -// Order matters for correctness: -// -// 1. Cancel the in-flight Build entity (build.ID == batch.ID; one Get + one -// Update covers all builds for this batch). A future external CI -// integration hooks in here. Idempotent: tolerate ErrNotFound (no build -// was scheduled), skip if already terminal. -// -// 2. CAS the batch to terminal Cancelled. This must happen BEFORE the -// dependent fan-out: tryFinalize only drops a Cancelled dep from the -// chain, so dependents woken with the dep still in Cancelling would -// wait pending and never get pinged again. -// -// 3. Re-publish each downstream dependent to speculate so they can drop -// this cancelled batch from their chain and advance (or finalize, if -// this was their last outstanding dep). -// -// 4. Publish to conclude so contained requests reach RequestStateCancelled. -// -// A crash between steps 2 and 3/4 is recovered on redelivery via the -// terminal self-heal branch, which re-runs the dependent fan-out and the -// conclude publish for already-Cancelled batches. -// -// storage.ErrVersionMismatch on the terminal CAS is returned as-is because it -// is intrinsically retryable; the redelivery will land in the -// self-heal branch and complete the fan-out. -func (c *Controller) cancelBatch(ctx context.Context, batch entity.Batch) error { - metrics.NamedCounter(c.metricsScope, opName, "cancel_batch", 1) - c.logger.Infow("cancelling batch", - "batch_id", batch.ID, - "queue", batch.Queue, - ) - - // TODO(respeculate-collateral): re-enqueue Land for every request in batch.Contains - // except the user-cancelled request. Today the whole batch dies (per spec) and the - // collateral requests need a fresh request ID and a re-publish to TopicKeyStart so - // they can be re-batched without the cancelled change. - - if err := c.cancelBuild(ctx, batch); err != nil { - return err - } - - newVersion := batch.Version + 1 - batch.State = entity.BatchStateCancelled - if err := c.store.GetBatchStore().Update(ctx, batch, batch.Version, newVersion); err != nil { - metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) - return fmt.Errorf("failed to update batch %s state to cancelled: %w", batch.ID, err) - } - batch.Version = newVersion - - if err := c.respeculateDependents(ctx, batch); err != nil { - return err - } - - if err := c.publishBatchID(ctx, topickey.TopicKeyConclude, batch.ID, batch.Queue); err != nil { - metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1) - return fmt.Errorf("failed to publish to conclude: %w", err) - } - - return nil -} - -// cancelBuild flips any in-flight Build entity for the batch to -// BuildStatusCancelled. Builds use build.ID == batch.ID, so a single Get -// covers every build scheduled for the batch. Tolerates ErrNotFound (no -// build was ever scheduled — the batch was cancelled before speculation -// started building) and skips already-terminal builds. -// -// This is the hook point for a future external CI integration: today the -// system has no external runner, so the local state flip is the complete -// cancellation. Once a runner exists, it must be invoked here before the -// local Update. -func (c *Controller) cancelBuild(ctx context.Context, batch entity.Batch) error { - build, err := c.store.GetBuildStore().Get(ctx, batch.ID) - if err != nil { - if errors.Is(err, storage.ErrNotFound) { - metrics.NamedCounter(c.metricsScope, opName, "cancel_build_not_found", 1) - return nil - } - metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) - return fmt.Errorf("failed to get build for batch %s: %w", batch.ID, err) - } - - if build.Status.IsTerminal() { - metrics.NamedCounter(c.metricsScope, opName, "cancel_build_already_terminal", 1) - return nil - } - - updatedBuild := build - updatedBuild.Status = entity.BuildStatusCancelled - if err := c.store.GetBuildStore().Update(ctx, updatedBuild); err != nil { - metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) - return fmt.Errorf("failed to cancel build for batch %s: %w", batch.ID, err) - } - metrics.NamedCounter(c.metricsScope, opName, "cancel_build_done", 1) - return nil -} - -// respeculateDependents publishes a speculate event for every batch that -// depends on the given batch. The batch controller creates a BatchDependent -// row (with Dependents possibly empty) for every batch it persists, so a -// missing row at this point is a storage invariant violation, not a normal -// "no dependents" case — surface ErrNotFound as a regular storage error so -// the message nacks and either an operator or the batch controller's own -// crash-recovery can resolve the inconsistency. -// -// Called both from the cancelBatch terminal flow and from the terminal -// self-heal branch on redelivery of an already-Cancelled batch. -func (c *Controller) respeculateDependents(ctx context.Context, batch entity.Batch) error { - bd, err := c.store.GetBatchDependentStore().Get(ctx, batch.ID) - if err != nil { - metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) - return fmt.Errorf("failed to get batch dependents for batch %s: %w", batch.ID, err) - } - - for _, depID := range bd.Dependents { - // Alternative: process each dependent inline (load batch, run the - // equivalent of tryFinalize) instead of publishing back to ourselves. - // Rejected for now: per-message retry isolation, fresh per-dependent - // reads, consumer-pool parallelism / backpressure, and the existing - // state-machine dispatch in Process all argue for the publish. Revisit - // if the extra message hop ever shows up as latency or cost. - if err := c.publishBatchID(ctx, topickey.TopicKeySpeculate, depID, batch.Queue); err != nil { - metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1) - return fmt.Errorf("failed to publish dependent batch %s to speculate: %w", depID, err) - } - metrics.NamedCounter(c.metricsScope, opName, "dependent_respeculated", 1) - } - return nil -} - -// fetchDependencies loads each batch in batch.Dependencies. Any storage error -// is surfaced as a retryable infra failure; missing dependencies should not -// happen in practice, but if one does it is treated the same as a transient -// fetch failure (i.e. the message is retried). -func (c *Controller) fetchDependencies(ctx context.Context, batch entity.Batch) ([]entity.Batch, error) { - deps := make([]entity.Batch, 0, len(batch.Dependencies)) - for _, depID := range batch.Dependencies { - d, err := c.store.GetBatchStore().Get(ctx, depID) - if err != nil { - metrics.NamedCounter(c.metricsScope, opName, "dependency_fetch_errors", 1) - return nil, fmt.Errorf("failed to get dependency batch %s of %s: %w", depID, batch.ID, err) - } - deps = append(deps, d) - } - return deps, nil -} - // fanout re-publishes downstream events for a batch that has already reached // a terminal state. Used for self-healing when a previous publish was lost: // re-sending to conclude guarantees request-state reconciliation. @@ -451,15 +154,17 @@ func (c *Controller) fanout(ctx context.Context, batchID, partitionKey string) e return nil } -// publishBatchID publishes a batch ID to the topic behind key. The batch ID -// doubles as the message ID, so the queue deduplicates repeat publishes for -// the same batch against rows it has not collected yet. +// publishBatchID publishes a batch ID to the topic behind key. Every publish +// gets a distinct message ID (publish.UniqueID): this controller re-publishes +// by design — a dispatch is re-sent until the build stage records it, and a +// terminal batch repeats its fan-out in case an earlier one was lost — and a +// stable message ID would make the queue swallow those repeats. func (c *Controller) publishBatchID(ctx context.Context, key consumer.TopicKey, batchID string, partitionKey string) error { payload, err := entity.BatchID{ID: batchID}.ToBytes() if err != nil { return fmt.Errorf("failed to serialize batch ID: %w", err) } - return publish.Message(ctx, c.registry, key, batchID, payload, partitionKey, 0) + return publish.Message(ctx, c.registry, key, publish.UniqueID(batchID), payload, partitionKey, 0) } // Name returns the controller name for logging and metrics. diff --git a/submitqueue/orchestrator/controller/speculate/speculate_test.go b/submitqueue/orchestrator/controller/speculate/speculate_test.go index 15ce16ab..33dc61bb 100644 --- a/submitqueue/orchestrator/controller/speculate/speculate_test.go +++ b/submitqueue/orchestrator/controller/speculate/speculate_test.go @@ -16,7 +16,6 @@ package speculate import ( "context" - "fmt" "testing" "github.com/stretchr/testify/assert" @@ -24,7 +23,6 @@ import ( "github.com/uber-go/tally" entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" - "github.com/uber/submitqueue/platform/errs" queuemock "github.com/uber/submitqueue/platform/extension/messagequeue/mock" "github.com/uber/submitqueue/submitqueue/core/topickey" "github.com/uber/submitqueue/submitqueue/entity" @@ -35,16 +33,15 @@ import ( "go.uber.org/zap/zaptest" ) -func batchWithState(batch entity.Batch, state entity.BatchState) entity.Batch { - batch.State = state - return batch +// 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. +type quietSpeculator struct { + heads []entity.Batch } -// quietSpeculator proposes nothing, which is what tests focused on the verdict -// state machine want: the run happens but changes no paths. -type quietSpeculator struct{} - -func (quietSpeculator) Speculate(context.Context, []entity.Batch, []entity.SpeculationPathSet) ([]entity.Speculation, error) { +func (s *quietSpeculator) Speculate(_ context.Context, batches []entity.Batch, _ []entity.SpeculationPathSet) ([]entity.Speculation, error) { + s.heads = append(s.heads, batches...) return nil, nil } @@ -57,618 +54,214 @@ func (f staticSpeculatorFactory) For(speculator.Config) (speculator.Speculator, // stubQuietRun makes the speculation run a no-op: the queue lists no in-flight // batches, so the run returns before reading any path set or asking the -// Speculator. Tests below exercise the verdict state machine, which the run is -// deliberately independent of; run_test.go covers the run itself. +// Speculator. run_test.go covers the run itself. func stubQuietRun(batchStore *storagemock.MockBatchStore) { batchStore.EXPECT(). GetByQueueAndStates(gomock.Any(), gomock.Any(), gomock.Any()). Return(nil, nil).AnyTimes() } -// batchIDPayload serializes a BatchID to JSON bytes for test message payloads. func batchIDPayload(t *testing.T, id string) []byte { + t.Helper() payload, err := entity.BatchID{ID: id}.ToBytes() require.NoError(t, err) return payload } -// testBatch returns a standard test batch with the given state and dependencies. func testBatch(state entity.BatchState, deps ...string) entity.Batch { return entity.Batch{ ID: "test-queue/batch/1", Queue: "test-queue", - Contains: []string{"test-queue/1"}, Dependencies: deps, State: state, Version: 1, } } -// newTestController wires a controller with a registry covering all topics the -// speculate controller may publish to. The publisher returns publishErr (or nil). -func newTestController(t *testing.T, ctrl *gomock.Controller, store *storagemock.MockStorage, publishErr error) *Controller { - logger := zaptest.NewLogger(t).Sugar() - scope := tally.NoopScope +// procHarness wires a controller and records which topics were published to. +type procHarness struct { + controller *Controller + batches *storagemock.MockBatchStore + pathSets *storagemock.MockSpeculationPathSetStore + pathBuilds *storagemock.MockPathBuildStore + builds *storagemock.MockBuildStore + spec *quietSpeculator + published []string +} - mockPub := queuemock.NewMockPublisher(ctrl) - mockPub.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn( - func(ctx context.Context, topic string, msg entityqueue.Message) error { - return publishErr - }, - ).AnyTimes() +func newProcHarness(t *testing.T, ctrl *gomock.Controller, publishErr error) *procHarness { + t.Helper() + h := &procHarness{spec: &quietSpeculator{}} - mockQ := queuemock.NewMockQueue(ctrl) - mockQ.EXPECT().Publisher().Return(mockPub).AnyTimes() + h.batches = storagemock.NewMockBatchStore(ctrl) + h.pathSets = storagemock.NewMockSpeculationPathSetStore(ctrl) + h.pathBuilds = storagemock.NewMockPathBuildStore(ctrl) + h.builds = storagemock.NewMockBuildStore(ctrl) - registry, err := consumer.NewTopicRegistry( - []consumer.TopicConfig{ - {Key: topickey.TopicKeyBuild, Name: "build", Queue: mockQ}, - {Key: topickey.TopicKeyMerge, Name: "submitqueue-merge", Queue: mockQ}, - {Key: topickey.TopicKeyConclude, Name: "conclude", Queue: mockQ}, - {Key: topickey.TopicKeyLog, Name: "log", Queue: mockQ}, + store := storagemock.NewMockStorage(ctrl) + store.EXPECT().GetBatchStore().Return(h.batches).AnyTimes() + store.EXPECT().GetSpeculationPathSetStore().Return(h.pathSets).AnyTimes() + store.EXPECT().GetPathBuildStore().Return(h.pathBuilds).AnyTimes() + store.EXPECT().GetBuildStore().Return(h.builds).AnyTimes() + + pub := queuemock.NewMockPublisher(ctrl) + pub.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn( + func(_ context.Context, topic string, _ entityqueue.Message) error { + if publishErr != nil { + return publishErr + } + h.published = append(h.published, topic) + return nil }, - ) + ).AnyTimes() + q := queuemock.NewMockQueue(ctrl) + q.EXPECT().Publisher().Return(pub).AnyTimes() + + registry, err := consumer.NewTopicRegistry([]consumer.TopicConfig{ + {Key: topickey.TopicKeyBuild, Name: "build", Queue: q}, + {Key: topickey.TopicKeyMerge, Name: "submitqueue-merge", Queue: q}, + {Key: topickey.TopicKeyConclude, Name: "conclude", Queue: q}, + {Key: topickey.TopicKeySpeculate, Name: "speculate", Queue: q}, + }) require.NoError(t, err) - return NewController(logger, scope, store, staticSpeculatorFactory{s: quietSpeculator{}}, registry, topickey.TopicKeySpeculate, "orchestrator-speculate") + h.controller = NewController( + zaptest.NewLogger(t).Sugar(), tally.NoopScope, store, + staticSpeculatorFactory{s: h.spec}, registry, + topickey.TopicKeySpeculate, "orchestrator-speculate", + ) + return h } -// runProcess builds a delivery for batchID and invokes Process once. -func runProcess(t *testing.T, ctrl *gomock.Controller, controller *Controller, batchID string) error { +func (h *procHarness) process(t *testing.T, ctrl *gomock.Controller, batchID string) error { + t.Helper() msg := entityqueue.NewMessage(batchID, batchIDPayload(t, batchID), "test-queue", nil) - delivery := queuemock.NewMockDelivery(ctrl) - delivery.EXPECT().Message().Return(msg).AnyTimes() - delivery.EXPECT().Attempt().Return(1).AnyTimes() - return controller.Process(context.Background(), delivery) + d := queuemock.NewMockDelivery(ctrl) + d.EXPECT().Message().Return(msg).AnyTimes() + d.EXPECT().Attempt().Return(1).AnyTimes() + return h.controller.Process(context.Background(), d) } func TestNewController(t *testing.T) { ctrl := gomock.NewController(t) - store := storagemock.NewMockStorage(ctrl) - controller := newTestController(t, ctrl, store, nil) - - require.NotNil(t, controller) - assert.Equal(t, topickey.TopicKeySpeculate, controller.TopicKey()) - assert.Equal(t, "orchestrator-speculate", controller.ConsumerGroup()) - assert.Equal(t, "speculate", controller.Name()) - - var _ consumer.Controller = controller -} - -// startSpeculation: Created should publish to build and CAS to Speculating with newVersion = oldVersion+1. -func TestController_Process_StartSpeculation(t *testing.T) { - tests := []struct { - name string - state entity.BatchState - }{ - {name: "from_created", state: entity.BatchStateCreated}, - } - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - ctrl := gomock.NewController(t) - batch := testBatch(tt.state) - - batchStore := storagemock.NewMockBatchStore(ctrl) - batchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil) - batchStore.EXPECT().Update(gomock.Any(), batchWithState(batch, entity.BatchStateSpeculating), int32(1), int32(2)).Return(nil) - - store := storagemock.NewMockStorage(ctrl) - store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() - stubQuietRun(batchStore) - - controller := newTestController(t, ctrl, store, nil) - require.NoError(t, runProcess(t, ctrl, controller, batch.ID)) - }) - } -} - -// tryFinalize: Speculating with no deps should publish to merge and CAS to Merging. -func TestController_Process_FinalizeNoDeps(t *testing.T) { - ctrl := gomock.NewController(t) - batch := testBatch(entity.BatchStateSpeculating) - - batchStore := storagemock.NewMockBatchStore(ctrl) - batchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil) - batchStore.EXPECT().Update(gomock.Any(), batchWithState(batch, entity.BatchStateMerging), int32(1), int32(2)).Return(nil) - - store := storagemock.NewMockStorage(ctrl) - store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() - stubQuietRun(batchStore) - - controller := newTestController(t, ctrl, store, nil) - require.NoError(t, runProcess(t, ctrl, controller, batch.ID)) -} - -// tryFinalize: Speculating with all deps Succeeded should publish to merge and CAS to Merging. -func TestController_Process_FinalizeAllDepsSucceeded(t *testing.T) { - ctrl := gomock.NewController(t) - depA := entity.Batch{ID: "test-queue/batch/0a", Queue: "test-queue", State: entity.BatchStateSucceeded, Version: 5} - depB := entity.Batch{ID: "test-queue/batch/0b", Queue: "test-queue", State: entity.BatchStateSucceeded, Version: 3} - batch := testBatch(entity.BatchStateSpeculating, depA.ID, depB.ID) - - batchStore := storagemock.NewMockBatchStore(ctrl) - batchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil) - batchStore.EXPECT().Get(gomock.Any(), depA.ID).Return(depA, nil) - batchStore.EXPECT().Get(gomock.Any(), depB.ID).Return(depB, nil) - batchStore.EXPECT().Update(gomock.Any(), batchWithState(batch, entity.BatchStateMerging), int32(1), int32(2)).Return(nil) - - store := storagemock.NewMockStorage(ctrl) - store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() - stubQuietRun(batchStore) - - controller := newTestController(t, ctrl, store, nil) - require.NoError(t, runProcess(t, ctrl, controller, batch.ID)) -} - -// tryFinalize: Speculating with a dep still in flight is a no-op (no publish, no state change). -func TestController_Process_WaitingOnDep(t *testing.T) { - ctrl := gomock.NewController(t) - dep := entity.Batch{ID: "test-queue/batch/0", Queue: "test-queue", State: entity.BatchStateSpeculating, Version: 1} - batch := testBatch(entity.BatchStateSpeculating, dep.ID) + h := newProcHarness(t, ctrl, nil) - batchStore := storagemock.NewMockBatchStore(ctrl) - batchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil) - batchStore.EXPECT().Get(gomock.Any(), dep.ID).Return(dep, nil) - // No Update expected — gomock will fail if it is called. + assert.Equal(t, topickey.TopicKeySpeculate, h.controller.TopicKey()) + assert.Equal(t, "orchestrator-speculate", h.controller.ConsumerGroup()) + assert.Equal(t, "speculate", h.controller.Name()) - store := storagemock.NewMockStorage(ctrl) - store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() - stubQuietRun(batchStore) - - controller := newTestController(t, ctrl, store, nil) - require.NoError(t, runProcess(t, ctrl, controller, batch.ID)) -} - -// tryFinalize: a failed dep must fail the batch (Speculating → Failed) and -// publish to conclude. Otherwise the batch livelocks. -func TestController_Process_FailedDepFailsBatch(t *testing.T) { - ctrl := gomock.NewController(t) - dep := entity.Batch{ID: "test-queue/batch/0", Queue: "test-queue", State: entity.BatchStateFailed, Version: 1} - batch := testBatch(entity.BatchStateSpeculating, dep.ID) - batch.Contains = []string{"test-queue/req/1", "test-queue/req/2"} - - batchStore := storagemock.NewMockBatchStore(ctrl) - batchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil) - batchStore.EXPECT().Get(gomock.Any(), dep.ID).Return(dep, nil) - batchStore.EXPECT().Update(gomock.Any(), batchWithState(batch, entity.BatchStateFailed), int32(1), int32(2)).Return(nil) - - store := storagemock.NewMockStorage(ctrl) - store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() - stubQuietRun(batchStore) - - controller := newTestController(t, ctrl, store, nil) - require.NoError(t, runProcess(t, ctrl, controller, batch.ID)) + var _ consumer.Controller = h.controller } -// tryFinalize: a cancelled dep is treated as out-of-the-way — it will never -// land and can no longer conflict. The dep is dropped from the chain and the -// batch advances to Merging as if the cancelled dep had succeeded. -func TestController_Process_CancelledDepSkipped(t *testing.T) { +// A Created batch is admitted so the Speculator can act on it, and must not +// reach an outcome on the same message — nothing has been built yet. +func TestProcess_AdmitsCreatedBatch(t *testing.T) { ctrl := gomock.NewController(t) - depCancelled := entity.Batch{ID: "test-queue/batch/0a", Queue: "test-queue", State: entity.BatchStateCancelled, Version: 2} - depSucceeded := entity.Batch{ID: "test-queue/batch/0b", Queue: "test-queue", State: entity.BatchStateSucceeded, Version: 5} - batch := testBatch(entity.BatchStateSpeculating, depCancelled.ID, depSucceeded.ID) - - batchStore := storagemock.NewMockBatchStore(ctrl) - batchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil) - batchStore.EXPECT().Get(gomock.Any(), depCancelled.ID).Return(depCancelled, nil) - batchStore.EXPECT().Get(gomock.Any(), depSucceeded.ID).Return(depSucceeded, nil) - batchStore.EXPECT().Update(gomock.Any(), batchWithState(batch, entity.BatchStateMerging), int32(1), int32(2)).Return(nil) + h := newProcHarness(t, ctrl, nil) + batch := testBatch(entity.BatchStateCreated) - store := storagemock.NewMockStorage(ctrl) - store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() - stubQuietRun(batchStore) - - controller := newTestController(t, ctrl, store, nil) - require.NoError(t, runProcess(t, ctrl, controller, batch.ID)) -} - -// Merging is owned by the merge controller — speculate is a no-op for it. -func TestController_Process_MergingNoOp(t *testing.T) { - ctrl := gomock.NewController(t) - batch := testBatch(entity.BatchStateMerging) - - batchStore := storagemock.NewMockBatchStore(ctrl) - batchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil) - // No Update expected. - - store := storagemock.NewMockStorage(ctrl) - store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() - stubQuietRun(batchStore) + h.batches.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil) + h.batches.EXPECT(). + Update(gomock.Any(), updateTo{id: batch.ID, state: entity.BatchStateSpeculating}, int32(1), int32(2)). + Return(nil) + stubQuietRun(h.batches) - controller := newTestController(t, ctrl, store, nil) - require.NoError(t, runProcess(t, ctrl, controller, batch.ID)) + require.NoError(t, h.process(t, ctrl, batch.ID)) + assert.Empty(t, h.published, "a batch cannot merge on the message that admitted it") } -// Terminal states re-fan-out to conclude for self-healing in case a previous -// publish was lost. State must not change (no Update). The Cancelled -// terminal also re-fans-out dependents and is covered separately in -// TestController_Process_CancelledTerminalSelfHealsDependents. -func TestController_Process_TerminalSelfHeals(t *testing.T) { +// A terminal batch re-publishes to conclude so a lost publish is repaired, and +// re-plans its queue so dependents see an outcome that was recorded after the +// run which produced it had already taken its snapshot. +func TestProcess_TerminalSelfHeals(t *testing.T) { for _, state := range []entity.BatchState{ entity.BatchStateSucceeded, entity.BatchStateFailed, + entity.BatchStateCancelled, } { t.Run(string(state), func(t *testing.T) { ctrl := gomock.NewController(t) + h := newProcHarness(t, ctrl, nil) batch := testBatch(state) - batchStore := storagemock.NewMockBatchStore(ctrl) - batchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil) - // No Update expected. + h.batches.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil) + stubQuietRun(h.batches) - store := storagemock.NewMockStorage(ctrl) - store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() - stubQuietRun(batchStore) - - // Require exactly one publish to the conclude topic for self-healing. - mockPub := queuemock.NewMockPublisher(ctrl) - mockPub.EXPECT().Publish(gomock.Any(), "conclude", gomock.Any()).Return(nil).Times(1) - - mockQ := queuemock.NewMockQueue(ctrl) - mockQ.EXPECT().Publisher().Return(mockPub).AnyTimes() - - registry, err := consumer.NewTopicRegistry( - []consumer.TopicConfig{ - {Key: topickey.TopicKeyConclude, Name: "conclude", Queue: mockQ}, - }, - ) - require.NoError(t, err) - - logger := zaptest.NewLogger(t).Sugar() - controller := NewController(logger, tally.NoopScope, store, staticSpeculatorFactory{s: quietSpeculator{}}, registry, topickey.TopicKeySpeculate, "orchestrator-speculate") - - require.NoError(t, runProcess(t, ctrl, controller, batch.ID)) + require.NoError(t, h.process(t, ctrl, batch.ID)) + assert.Equal(t, []string{"conclude"}, h.published) }) } } -// Cancelled is terminal: redelivery must re-fan-out dependents (so a crash -// between the terminal CAS and the dependent publish does not strand them) -// AND re-publish to conclude. State must not change (no Update; no -// build cancel). The BuildStore must not be touched on this self-heal path. -func TestController_Process_CancelledTerminalSelfHealsDependents(t *testing.T) { - ctrl := gomock.NewController(t) - batch := testBatch(entity.BatchStateCancelled) - - batchStore := storagemock.NewMockBatchStore(ctrl) - batchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil) - // No Update expected. - - depStore := storagemock.NewMockBatchDependentStore(ctrl) - depStore.EXPECT().Get(gomock.Any(), batch.ID).Return(entity.BatchDependent{ - BatchID: batch.ID, - Dependents: []string{"test-queue/batch/2", "test-queue/batch/3"}, - Version: 1, - }, nil) - - store := storagemock.NewMockStorage(ctrl) - store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() - stubQuietRun(batchStore) - store.EXPECT().GetBatchDependentStore().Return(depStore).AnyTimes() - // BuildStore must NOT be touched on the terminal self-heal path. - - type pubRec struct { - topic string - msgID string - } - var records []pubRec - mockPub := queuemock.NewMockPublisher(ctrl) - mockPub.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn( - func(_ context.Context, topic string, msg entityqueue.Message) error { - records = append(records, pubRec{topic: topic, msgID: msg.ID}) - return nil - }).AnyTimes() - - mockQ := queuemock.NewMockQueue(ctrl) - mockQ.EXPECT().Publisher().Return(mockPub).AnyTimes() - - registry, err := consumer.NewTopicRegistry( - []consumer.TopicConfig{ - {Key: topickey.TopicKeyConclude, Name: "conclude", Queue: mockQ}, - {Key: topickey.TopicKeySpeculate, Name: "speculate", Queue: mockQ}, - }, - ) - require.NoError(t, err) - - logger := zaptest.NewLogger(t).Sugar() - controller := NewController(logger, tally.NoopScope, store, staticSpeculatorFactory{s: quietSpeculator{}}, registry, topickey.TopicKeySpeculate, "orchestrator-speculate") - - require.NoError(t, runProcess(t, ctrl, controller, batch.ID)) - - assert.Equal(t, []pubRec{ - {topic: "speculate", msgID: "test-queue/batch/2"}, - {topic: "speculate", msgID: "test-queue/batch/3"}, - {topic: "conclude", msgID: batch.ID}, - }, records) -} - -// Cancelling drives the terminal-cancellation flow: cancel any in-flight -// build, CAS the batch to Cancelled, fan out dependents, publish to -// conclude. Validates the full happy-path order with a running build and -// a couple of dependents. Order matters: dependents must publish AFTER the -// terminal CAS so the woken dependents observe the dep as Cancelled (and -// drop it from their chain) rather than as still-Cancelling (which would -// leave them waiting on a state nobody is going to nudge). -func TestController_Process_CancellingTerminalFlow(t *testing.T) { +// A terminal batch re-plans its queue rather than only reconciling itself. The +// run that finalized it computed every dependent against a snapshot taken +// before the transition, so without this a dependent whose own builds have all +// finished would never learn the outcome. +func TestProcess_TerminalReplansQueue(t *testing.T) { ctrl := gomock.NewController(t) - batch := testBatch(entity.BatchStateCancelling) - - batchStore := storagemock.NewMockBatchStore(ctrl) - batchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil) - batchStore.EXPECT().Update(gomock.Any(), batchWithState(batch, entity.BatchStateCancelled), int32(1), int32(2)).Return(nil) - - buildStore := storagemock.NewMockBuildStore(ctrl) - build := entity.Build{ - ID: batch.ID, BatchID: batch.ID, Status: entity.BuildStatusRunning, - } - buildStore.EXPECT().Get(gomock.Any(), batch.ID).Return(build, nil) - updatedBuild := build - updatedBuild.Status = entity.BuildStatusCancelled - buildStore.EXPECT().Update(gomock.Any(), updatedBuild).Return(nil) - - depStore := storagemock.NewMockBatchDependentStore(ctrl) - depStore.EXPECT().Get(gomock.Any(), batch.ID).Return(entity.BatchDependent{ - BatchID: batch.ID, - Dependents: []string{"test-queue/batch/2", "test-queue/batch/3"}, - Version: 1, - }, nil) + h := newProcHarness(t, ctrl, nil) + batch := testBatch(entity.BatchStateSucceeded) - store := storagemock.NewMockStorage(ctrl) - store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() - stubQuietRun(batchStore) - store.EXPECT().GetBuildStore().Return(buildStore).AnyTimes() - store.EXPECT().GetBatchDependentStore().Return(depStore).AnyTimes() - - type pubRec struct { - topic string - msgID string + dependent := entity.Batch{ + ID: "test-queue/batch/2", Queue: "test-queue", + State: entity.BatchStateSpeculating, Dependencies: []string{batch.ID}, Version: 1, } - var records []pubRec - mockPub := queuemock.NewMockPublisher(ctrl) - mockPub.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn( - func(_ context.Context, topic string, msg entityqueue.Message) error { - records = append(records, pubRec{topic: topic, msgID: msg.ID}) - return nil - }).AnyTimes() - - mockQ := queuemock.NewMockQueue(ctrl) - mockQ.EXPECT().Publisher().Return(mockPub).AnyTimes() - - registry, err := consumer.NewTopicRegistry( - []consumer.TopicConfig{ - {Key: topickey.TopicKeyConclude, Name: "conclude", Queue: mockQ}, - {Key: topickey.TopicKeySpeculate, Name: "speculate", Queue: mockQ}, - }, - ) - require.NoError(t, err) - - logger := zaptest.NewLogger(t).Sugar() - controller := NewController(logger, tally.NoopScope, store, staticSpeculatorFactory{s: quietSpeculator{}}, registry, topickey.TopicKeySpeculate, "orchestrator-speculate") - - require.NoError(t, runProcess(t, ctrl, controller, batch.ID)) - - assert.Equal(t, []pubRec{ - {topic: "speculate", msgID: "test-queue/batch/2"}, - {topic: "speculate", msgID: "test-queue/batch/3"}, - {topic: "conclude", msgID: batch.ID}, - }, records) -} - -// If the build for the batch has already reached a terminal status (e.g. CI -// finished naturally between the cancel intent and the speculate pickup), the -// cancellation must not re-flip it — Update must never fire. The rest -// of the flow (terminal batch CAS, dependent fan-out, conclude) still runs. -func TestController_Process_CancellingBuildAlreadyTerminal(t *testing.T) { - ctrl := gomock.NewController(t) - batch := testBatch(entity.BatchStateCancelling) - - batchStore := storagemock.NewMockBatchStore(ctrl) - batchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil) - batchStore.EXPECT().Update(gomock.Any(), batchWithState(batch, entity.BatchStateCancelled), int32(1), int32(2)).Return(nil) - - buildStore := storagemock.NewMockBuildStore(ctrl) - buildStore.EXPECT().Get(gomock.Any(), batch.ID).Return(entity.Build{ - ID: batch.ID, BatchID: batch.ID, Status: entity.BuildStatusSucceeded, - }, nil) - // No Update expected — the build is already terminal. - - depStore := storagemock.NewMockBatchDependentStore(ctrl) - depStore.EXPECT().Get(gomock.Any(), batch.ID).Return(entity.BatchDependent{ - BatchID: batch.ID, Version: 1, - }, nil) - - store := storagemock.NewMockStorage(ctrl) - store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() - stubQuietRun(batchStore) - store.EXPECT().GetBuildStore().Return(buildStore).AnyTimes() - store.EXPECT().GetBatchDependentStore().Return(depStore).AnyTimes() - - controller := newTestController(t, ctrl, store, nil) - require.NoError(t, runProcess(t, ctrl, controller, batch.ID)) -} - -// If no Build entity exists for the batch (e.g. cancel arrived before -// speculation started building), the BuildStore.Get NotFound must be -// tolerated and the rest of the cancellation flow must continue. -func TestController_Process_CancellingNoBuildYet(t *testing.T) { - ctrl := gomock.NewController(t) - batch := testBatch(entity.BatchStateCancelling) - - batchStore := storagemock.NewMockBatchStore(ctrl) - batchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil) - batchStore.EXPECT().Update(gomock.Any(), batchWithState(batch, entity.BatchStateCancelled), int32(1), int32(2)).Return(nil) - - buildStore := storagemock.NewMockBuildStore(ctrl) - buildStore.EXPECT().Get(gomock.Any(), batch.ID).Return(entity.Build{}, storage.ErrNotFound) - // No Update expected. - - depStore := storagemock.NewMockBatchDependentStore(ctrl) - depStore.EXPECT().Get(gomock.Any(), batch.ID).Return(entity.BatchDependent{ - BatchID: batch.ID, Version: 1, - }, nil) - - store := storagemock.NewMockStorage(ctrl) - store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() - stubQuietRun(batchStore) - store.EXPECT().GetBuildStore().Return(buildStore).AnyTimes() - store.EXPECT().GetBatchDependentStore().Return(depStore).AnyTimes() - - controller := newTestController(t, ctrl, store, nil) - require.NoError(t, runProcess(t, ctrl, controller, batch.ID)) -} - -// A batch whose BatchDependent row exists with an empty Dependents list must -// still drive itself to terminal and publish to conclude. This is the normal -// "no dependents" path: the batch controller creates the row with an empty -// list at batch creation time and it stays empty if no later batch conflicts. -func TestController_Process_CancellingNoDependents(t *testing.T) { - ctrl := gomock.NewController(t) - batch := testBatch(entity.BatchStateCancelling) - - batchStore := storagemock.NewMockBatchStore(ctrl) - batchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil) - batchStore.EXPECT().Update(gomock.Any(), batchWithState(batch, entity.BatchStateCancelled), int32(1), int32(2)).Return(nil) - buildStore := storagemock.NewMockBuildStore(ctrl) - buildStore.EXPECT().Get(gomock.Any(), batch.ID).Return(entity.Build{}, storage.ErrNotFound) - - depStore := storagemock.NewMockBatchDependentStore(ctrl) - depStore.EXPECT().Get(gomock.Any(), batch.ID).Return(entity.BatchDependent{BatchID: batch.ID, Dependents: []string{}, Version: 1}, nil) - - store := storagemock.NewMockStorage(ctrl) - store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() - stubQuietRun(batchStore) - store.EXPECT().GetBuildStore().Return(buildStore).AnyTimes() - store.EXPECT().GetBatchDependentStore().Return(depStore).AnyTimes() - - mockPub := queuemock.NewMockPublisher(ctrl) - mockPub.EXPECT().Publish(gomock.Any(), "conclude", gomock.Any()).Return(nil).Times(1) - - mockQ := queuemock.NewMockQueue(ctrl) - mockQ.EXPECT().Publisher().Return(mockPub).AnyTimes() - - registry, err := consumer.NewTopicRegistry( - []consumer.TopicConfig{ - {Key: topickey.TopicKeyConclude, Name: "conclude", Queue: mockQ}, - }, - ) - require.NoError(t, err) - - logger := zaptest.NewLogger(t).Sugar() - controller := NewController(logger, tally.NoopScope, store, staticSpeculatorFactory{s: quietSpeculator{}}, registry, topickey.TopicKeySpeculate, "orchestrator-speculate") - - require.NoError(t, runProcess(t, ctrl, controller, batch.ID)) -} - -// storage.ErrVersionMismatch on the terminal CAS must surface as an error -// with the underlying sentinel in the chain so its intrinsic retryable -// classification survives. The dependent fan-out and conclude publish must -// NOT run if the terminal CAS failed — on redelivery the self-heal branch -// will pick up the (now-terminal) state and complete the fan-out. -func TestController_Process_CancellingTerminalCASVersionMismatch(t *testing.T) { - ctrl := gomock.NewController(t) - batch := testBatch(entity.BatchStateCancelling) - - batchStore := storagemock.NewMockBatchStore(ctrl) - batchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil) - batchStore.EXPECT().Update(gomock.Any(), batchWithState(batch, entity.BatchStateCancelled), int32(1), int32(2)). - Return(storage.ErrVersionMismatch) - - buildStore := storagemock.NewMockBuildStore(ctrl) - buildStore.EXPECT().Get(gomock.Any(), batch.ID).Return(entity.Build{}, storage.ErrNotFound) - - store := storagemock.NewMockStorage(ctrl) - store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() - stubQuietRun(batchStore) - store.EXPECT().GetBuildStore().Return(buildStore).AnyTimes() - // BatchDependentStore must NOT be touched — terminal CAS failed before fan-out. - - // No publish expected (terminal CAS failed before fan-out). - mockPub := queuemock.NewMockPublisher(ctrl) - mockQ := queuemock.NewMockQueue(ctrl) - mockQ.EXPECT().Publisher().Return(mockPub).AnyTimes() - - registry, err := consumer.NewTopicRegistry( - []consumer.TopicConfig{ - {Key: topickey.TopicKeyConclude, Name: "conclude", Queue: mockQ}, - {Key: topickey.TopicKeySpeculate, Name: "speculate", Queue: mockQ}, - }, - ) - require.NoError(t, err) - - logger := zaptest.NewLogger(t).Sugar() - controller := NewController(logger, tally.NoopScope, store, staticSpeculatorFactory{s: quietSpeculator{}}, registry, topickey.TopicKeySpeculate, "orchestrator-speculate") - - err = runProcess(t, ctrl, controller, batch.ID) - require.Error(t, err) - assert.ErrorIs(t, err, storage.ErrVersionMismatch) + h.batches.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil) + h.batches.EXPECT(). + GetByQueueAndStates(gomock.Any(), "test-queue", entity.ActiveBatchStates()). + Return([]entity.Batch{dependent}, nil) + h.batches.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil) + h.pathSets.EXPECT().Get(gomock.Any(), dependent.ID). + 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") } -// An unrecognized state must surface as an error so the message is nacked -// instead of silently acked — silently acking would drop the event. -func TestController_Process_UnrecognizedState(t *testing.T) { - ctrl := gomock.NewController(t) - batch := testBatch(entity.BatchStateUnknown) - - batchStore := storagemock.NewMockBatchStore(ctrl) - batchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil) - - store := storagemock.NewMockStorage(ctrl) - store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() - stubQuietRun(batchStore) - - controller := newTestController(t, ctrl, store, nil) - require.Error(t, runProcess(t, ctrl, controller, batch.ID)) -} - -// Storage failure on the primary batch fetch surfaces as an error and is not -// retryable per the controller default (plain fmt.Errorf). -func TestController_Process_StorageFailure(t *testing.T) { +// A Merging batch is the merge stage's to finish; the run still happens for the +// rest of the queue, but this batch is not an action target. +func TestProcess_MergingRunsButDoesNotAct(t *testing.T) { ctrl := gomock.NewController(t) + h := newProcHarness(t, ctrl, nil) + batch := testBatch(entity.BatchStateMerging) - batchStore := storagemock.NewMockBatchStore(ctrl) - batchStore.EXPECT().Get(gomock.Any(), "test-queue/batch/1").Return(entity.Batch{}, fmt.Errorf("db connection lost")) + h.batches.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil) + stubQuietRun(h.batches) - store := storagemock.NewMockStorage(ctrl) - store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() - stubQuietRun(batchStore) - - controller := newTestController(t, ctrl, store, nil) - err := runProcess(t, ctrl, controller, "test-queue/batch/1") - require.Error(t, err) - assert.False(t, errs.IsRetryable(err)) + require.NoError(t, h.process(t, ctrl, batch.ID)) + assert.Empty(t, h.published) } -// Publish failure must not advance the batch state. -// A failed merge publish must abort before the batch is moved to Merging: -// a batch recorded as merging that Runway was never told about would stall. -func TestController_Process_PublishFailure(t *testing.T) { - ctrl := gomock.NewController(t) - batch := testBatch(entity.BatchStateSpeculating) +func TestProcess_Errors(t *testing.T) { + t.Run("malformed payload", func(t *testing.T) { + ctrl := gomock.NewController(t) + h := newProcHarness(t, ctrl, nil) - batchStore := storagemock.NewMockBatchStore(ctrl) - batchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil) - // No Update expected — publish fails before we get there. + msg := entityqueue.NewMessage("anything", []byte("not-json"), "test-queue", nil) + d := queuemock.NewMockDelivery(ctrl) + d.EXPECT().Message().Return(msg).AnyTimes() + d.EXPECT().Attempt().Return(1).AnyTimes() - store := storagemock.NewMockStorage(ctrl) - store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() - stubQuietRun(batchStore) + require.Error(t, h.controller.Process(context.Background(), d)) + }) - controller := newTestController(t, ctrl, store, fmt.Errorf("publish failed")) - require.Error(t, runProcess(t, ctrl, controller, batch.ID)) -} + t.Run("batch read failure", func(t *testing.T) { + ctrl := gomock.NewController(t) + h := newProcHarness(t, ctrl, nil) + h.batches.EXPECT().Get(gomock.Any(), "test-queue/batch/1"). + Return(entity.Batch{}, storage.ErrNotFound) -// Malformed payload: deserialize error. -func TestController_Process_BadPayload(t *testing.T) { - ctrl := gomock.NewController(t) - store := storagemock.NewMockStorage(ctrl) - controller := newTestController(t, ctrl, store, nil) + require.Error(t, h.process(t, ctrl, "test-queue/batch/1")) + }) - msg := entityqueue.NewMessage("anything", []byte("not-json"), "test-queue", nil) - delivery := queuemock.NewMockDelivery(ctrl) - delivery.EXPECT().Message().Return(msg).AnyTimes() - delivery.EXPECT().Attempt().Return(1).AnyTimes() + t.Run("conclude publish failure on a terminal batch", func(t *testing.T) { + ctrl := gomock.NewController(t) + h := newProcHarness(t, ctrl, assert.AnError) + batch := testBatch(entity.BatchStateSucceeded) + h.batches.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil) - require.Error(t, controller.Process(context.Background(), delivery)) + require.Error(t, h.process(t, ctrl, batch.ID)) + }) }