From a2e7f8b3dd68e8bd3edabed80e348e0c18895f66 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. Batch state writes go through `corebatch.Transition` rather than a raw CAS, so the queue's membership record moves with the state. A raw CAS leaves the batch filed under the bucket it just left, and because records are only ever added, every later run of that queue goes on listing and hydrating it. This covers both the outcome write added here and `admit`, which had the same shape. ## 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` # Conflicts: # submitqueue/orchestrator/controller/speculate/BUILD.bazel # submitqueue/orchestrator/controller/speculate/speculate_test.go # Please enter the commit message for your changes. Lines starting # with '#' will be kept; you may remove them yourself if you want to. # An empty message aborts the commit. # # interactive rebase in progress; onto ec05b712 # Last command done (1 command done): # pick 4462c891 # feat(orchestrator)!: finalize batches from their speculation paths # No commands remaining. # You are currently rebasing branch 'preetam/speculation-finalization' on 'ec05b712'. # # Changes to be committed: # modified: submitqueue/orchestrator/controller/cancel/cancel.go # modified: submitqueue/orchestrator/controller/speculate/BUILD.bazel # modified: submitqueue/orchestrator/controller/speculate/check.go # modified: submitqueue/orchestrator/controller/speculate/dispatch.go # modified: submitqueue/orchestrator/controller/speculate/doc.go # new file: submitqueue/orchestrator/controller/speculate/finalize.go # new file: submitqueue/orchestrator/controller/speculate/outcome.go # new file: submitqueue/orchestrator/controller/speculate/outcome_test.go # modified: submitqueue/orchestrator/controller/speculate/run.go # modified: submitqueue/orchestrator/controller/speculate/run_test.go # modified: submitqueue/orchestrator/controller/speculate/snapshot.go # modified: submitqueue/orchestrator/controller/speculate/speculate.go # modified: submitqueue/orchestrator/controller/speculate/speculate_test.go # # Conflicts: # submitqueue/orchestrator/controller/speculate/speculate.go # Please enter the commit message for your changes. Lines starting # with '#' will be kept; you may remove them yourself if you want to. # An empty message aborts the commit. # # interactive rebase in progress; onto 9a490fe4 # Last command done (1 command done): # pick ec12146f # feat(orchestrator)!: finalize batches from their speculation paths # No commands remaining. # You are currently rebasing branch 'preetam/speculation-finalization' on '9a490fe4'. # # Changes to be committed: # modified: submitqueue/orchestrator/controller/speculate/BUILD.bazel # modified: submitqueue/orchestrator/controller/speculate/check.go # modified: submitqueue/orchestrator/controller/speculate/dispatch.go # modified: submitqueue/orchestrator/controller/speculate/doc.go # new file: submitqueue/orchestrator/controller/speculate/finalize.go # new file: submitqueue/orchestrator/controller/speculate/outcome.go # new file: submitqueue/orchestrator/controller/speculate/outcome_test.go # modified: submitqueue/orchestrator/controller/speculate/run.go # modified: submitqueue/orchestrator/controller/speculate/run_test.go # modified: submitqueue/orchestrator/controller/speculate/snapshot.go # modified: submitqueue/orchestrator/controller/speculate/speculate.go # modified: submitqueue/orchestrator/controller/speculate/speculate_test.go # # Conflicts: # submitqueue/orchestrator/controller/speculate/dispatch.go # submitqueue/orchestrator/controller/speculate/run.go # submitqueue/orchestrator/controller/speculate/run_test.go # submitqueue/orchestrator/controller/speculate/speculate.go # submitqueue/orchestrator/controller/speculate/speculate_test.go # Please enter the commit message for your changes. Lines starting # with '#' will be kept; you may remove them yourself if you want to. # An empty message aborts the commit. # # interactive rebase in progress; onto e6ea890a # Last command done (1 command done): # pick 25e03214 # feat(orchestrator)!: finalize batches from their speculation paths # No commands remaining. # You are currently rebasing branch 'preetam/speculation-finalization' on 'e6ea890a'. # # Changes to be committed: # modified: submitqueue/orchestrator/controller/speculate/BUILD.bazel # modified: submitqueue/orchestrator/controller/speculate/check.go # modified: submitqueue/orchestrator/controller/speculate/dispatch.go # modified: submitqueue/orchestrator/controller/speculate/doc.go # new file: submitqueue/orchestrator/controller/speculate/finalize.go # new file: submitqueue/orchestrator/controller/speculate/outcome.go # new file: submitqueue/orchestrator/controller/speculate/outcome_test.go # modified: submitqueue/orchestrator/controller/speculate/run.go # modified: submitqueue/orchestrator/controller/speculate/run_test.go # modified: submitqueue/orchestrator/controller/speculate/snapshot.go # modified: submitqueue/orchestrator/controller/speculate/speculate.go # modified: submitqueue/orchestrator/controller/speculate/speculate_test.go # # Conflicts: # submitqueue/orchestrator/controller/speculate/run_test.go # submitqueue/orchestrator/controller/speculate/speculate.go # Please enter the commit message for your changes. Lines starting # with '#' will be kept; you may remove them yourself if you want to. # An empty message aborts the commit. # # interactive rebase in progress; onto 643005d1 # Last command done (1 command done): # pick 432bd246 # feat(orchestrator)!: finalize batches from their speculation paths # No commands remaining. # You are currently rebasing branch 'preetam/speculation-finalization' on '643005d1'. # # Changes to be committed: # modified: submitqueue/orchestrator/controller/speculate/BUILD.bazel # modified: submitqueue/orchestrator/controller/speculate/check.go # modified: submitqueue/orchestrator/controller/speculate/dispatch.go # modified: submitqueue/orchestrator/controller/speculate/doc.go # new file: submitqueue/orchestrator/controller/speculate/finalize.go # new file: submitqueue/orchestrator/controller/speculate/outcome.go # new file: submitqueue/orchestrator/controller/speculate/outcome_test.go # modified: submitqueue/orchestrator/controller/speculate/run.go # modified: submitqueue/orchestrator/controller/speculate/run_test.go # modified: submitqueue/orchestrator/controller/speculate/snapshot.go # modified: submitqueue/orchestrator/controller/speculate/speculate.go # modified: submitqueue/orchestrator/controller/speculate/speculate_test.go # --- .../controller/speculate/BUILD.bazel | 4 +- .../controller/speculate/check.go | 4 +- .../controller/speculate/dispatch.go | 36 +- .../orchestrator/controller/speculate/doc.go | 90 ++- .../controller/speculate/finalize.go | 390 +++++++++ .../controller/speculate/outcome.go | 134 ++++ .../controller/speculate/outcome_test.go | 237 ++++++ .../orchestrator/controller/speculate/run.go | 108 ++- .../controller/speculate/run_test.go | 758 +++++++++++++++++- .../controller/speculate/snapshot.go | 43 +- .../controller/speculate/speculate.go | 391 ++------- .../controller/speculate/speculate_test.go | 740 ++++------------- 12 files changed, 1890 insertions(+), 1045 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 9f1467d2..f4eafb86 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", @@ -30,6 +32,7 @@ go_test( name = "go_default_test", srcs = [ "check_test.go", + "outcome_test.go", "run_test.go", "snapshot_test.go", "speculate_test.go", @@ -39,7 +42,6 @@ go_test( "//platform/base/messagequeue:go_default_library", "//platform/consumer:go_default_library", "//platform/consumer/mock: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 f802b7ed..24f99105 100644 --- a/submitqueue/orchestrator/controller/speculate/dispatch.go +++ b/submitqueue/orchestrator/controller/speculate/dispatch.go @@ -26,11 +26,19 @@ 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. -func (c *Controller) dispatch(ctx context.Context, store storage.Storage, queue string, snap snapshot, kept []entity.Speculation) error { +// +// 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() // Group the accepted proposals by head so each head is written once. @@ -39,7 +47,7 @@ func (c *Controller) dispatch(ctx context.Context, store storage.Storage, queue 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, store storage.Storage, queue } if changed { - if err := c.persist(ctx, store, set, exists); err != nil { + if _, err := c.persist(ctx, snap.store, 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, store storage.Storage, queue } // 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, store storage.Storage, 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, store storage.Storage, set entity.SpeculationPathSet, exists bool) (entity.SpeculationPathSet, error) { pathSets := store.GetSpeculationPathSetStore() if !exists { @@ -111,23 +120,24 @@ func (c *Controller) persist(ctx context.Context, store storage.Storage, set ent 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 := pathSets.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..668f7f0d --- /dev/null +++ b/submitqueue/orchestrator/controller/speculate/finalize.go @@ -0,0 +1,390 @@ +// 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" + corebatch "github.com/uber/submitqueue/submitqueue/core/batch" + "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, snap.store, 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, snap.store, 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, store storage.Storage, 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, 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, store, 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 + } + + // Through Transition, so the queue's membership record moves with the + // state. A raw CAS would leave the batch filed under the bucket it just + // left, and since records are only ever added, every later run of this + // queue would go on listing and hydrating it. + if _, err := corebatch.Transition(ctx, store, batch, state); 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, 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, 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, store storage.Storage, batch entity.Batch) error { + if err := c.publishBatchID(ctx, topickey.TopicKeySpeculate, batch.ID, batch.Queue, 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 d80c36ab..d5cfeb57 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" corebatch "github.com/uber/submitqueue/submitqueue/core/batch" @@ -28,30 +27,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, store storage.Storage, queue string) error { - snap, err := c.read(ctx, store, 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, store storage.Storage, trigger entity.Batch) error { + snap, err := c.read(ctx, store, 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 } @@ -61,12 +60,12 @@ func (c *Controller) run(ctx context.Context, store storage.Storage, queue strin 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, store, queue, snap, kept) + return c.dispatch(ctx, trigger.Queue, snap, kept) } // read builds the run's snapshot. Batches come first because their dependency @@ -80,7 +79,9 @@ func (c *Controller) read(ctx context.Context, store storage.Storage, queue stri } snap := snapshot{ + store: store, 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)), } @@ -161,8 +162,18 @@ func (c *Controller) updatePathsFromBuilds(ctx context.Context, store storage.St // 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 { @@ -226,49 +237,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 { @@ -277,7 +259,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 ed3f3f8f..150c30f9 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 store *storagemock.MockStorage @@ -60,9 +76,38 @@ type runHarness struct { builds *storagemock.MockBuildStore spec *scriptedSpeculator published []string + // filed and unfiled are the queue's membership-record writes: a state + // change has to move the record, or the batch stays listed under the + // bucket it left and every later run re-reads it. + filed []entity.QueueBatchState + unfiled []entity.QueueBatchState // messages are the published messages themselves, so a test can assert // what a publish carried and how it was partitioned. messages []entityqueue.Message + // 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(), h.store, entity.Batch{ID: triggerID, Queue: "q"}) } // newRunHarness wires a controller whose queue read returns inFlight. @@ -90,6 +135,18 @@ func newRunHarness(t *testing.T, ctrl *gomock.Controller, spec *scriptedSpeculat return records, nil }, ).AnyTimes() + queueStates.EXPECT().Put(gomock.Any(), gomock.Any()).DoAndReturn( + func(_ context.Context, record entity.QueueBatchState) error { + h.filed = append(h.filed, record) + return nil + }, + ).AnyTimes() + queueStates.EXPECT().Delete(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn( + func(_ context.Context, state entity.BatchState, batchID string) error { + h.unfiled = append(h.unfiled, entity.QueueBatchState{State: state, BatchID: batchID}) + return nil + }, + ).AnyTimes() h.pathSets = storagemock.NewMockSpeculationPathSetStore(ctrl) h.pathBuilds = storagemock.NewMockPathBuildStore(ctrl) @@ -97,8 +154,8 @@ func newRunHarness(t *testing.T, ctrl *gomock.Controller, spec *scriptedSpeculat store := storagemock.NewMockStorage(ctrl) h.store = store - store.EXPECT().GetBatchStore().Return(h.batches).AnyTimes() store.EXPECT().GetQueueBatchStateStore().Return(queueStates).AnyTimes() + 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() @@ -106,6 +163,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) h.messages = append(h.messages, msg) return nil @@ -118,6 +178,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) @@ -170,7 +231,7 @@ func TestRun_FundsProposedPath(t *testing.T) { return nil }) - require.NoError(t, h.controller.run(context.Background(), h.store, "q")) + require.NoError(t, h.run(head)) assert.Equal(t, []string{"build"}, h.published) } @@ -191,7 +252,7 @@ func TestRun_DispatchStampsQueueAndPartitionsByHead(t *testing.T) { h.pathSets.EXPECT().Get(gomock.Any(), head).Return(entity.SpeculationPathSet{}, storage.ErrNotFound) h.pathSets.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil) - require.NoError(t, h.controller.run(context.Background(), h.store, "q")) + require.NoError(t, h.run(head)) require.Len(t, h.messages, 1) got, err := entity.BatchIDFromBytes(h.messages[0].Payload) @@ -216,7 +277,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(), h.store, "q")) + require.NoError(t, h.run(head)) require.Equal(t, 1, spec.calls) require.Len(t, spec.gotBatches, 1) @@ -250,7 +311,7 @@ func TestRun_CancelsBrokenPath(t *testing.T) { return nil }) - require.NoError(t, h.controller.run(context.Background(), h.store, "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") } @@ -271,7 +332,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(), h.store, "q")) + require.NoError(t, h.run(head)) assert.Empty(t, h.published) } @@ -292,7 +353,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(), h.store, "q")) + require.NoError(t, h.run(head)) // The head still has no actionable path, so nothing is dispatched. assert.Empty(t, h.published) @@ -316,7 +377,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(), h.store, "q")) + require.NoError(t, h.run(head)) assert.Equal(t, []string{"build"}, h.published) } @@ -339,7 +400,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(), h.store, "q")) + require.NoError(t, h.run(head)) assert.Empty(t, h.published, "a head whose write was lost is not dispatched on stale state") } @@ -355,7 +416,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(), h.store, "q")) + require.Error(t, h.run(head)) } // A queue with nothing speculating never reaches the Speculator. @@ -364,7 +425,7 @@ func TestRun_NoSpeculatingBatches(t *testing.T) { spec := &scriptedSpeculator{} h := newRunHarness(t, ctrl, spec, nil) - require.NoError(t, h.controller.run(context.Background(), h.store, "q")) + require.NoError(t, h.run(head)) assert.Zero(t, spec.calls) } @@ -412,7 +473,7 @@ func TestRun_RecordsFinishedBuildsOnPaths(t *testing.T) { return nil }) - require.NoError(t, h.controller.run(context.Background(), h.store, "q")) + require.NoError(t, h.run(head)) }) } } @@ -441,7 +502,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(), h.store, "q")) + require.NoError(t, h.run(head)) assert.Empty(t, h.published) } @@ -452,7 +513,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, @@ -462,7 +525,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(), h.store, "q")) + require.NoError(t, h.run(head)) } // Broken paths are cancelled before the Speculator is asked, so it reasons over the @@ -486,7 +549,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(), h.store, "q")) + require.NoError(t, h.run(head)) require.Equal(t, 1, spec.calls) require.Len(t, spec.gotSets, 1) @@ -540,3 +603,666 @@ 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) + assert.Contains(t, h.filed, entity.QueueBatchState{Queue: "q", State: entity.BatchStateMerging, BatchID: head}, + "the outcome must file the head under its new state") + assert.Contains(t, h.unfiled, entity.QueueBatchState{State: entity.BatchStateSpeculating, BatchID: head}, + "and drop the record under the state it left") +} + +// 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, staticStorageFactory{store: storagemock.NewMockStorage(ctrl)}, + staticSpeculatorFactory{}, registry, topickey.TopicKeySpeculate, "orchestrator-speculate", + ) + + require.NoError(t, c.publishBatchID(context.Background(), topickey.TopicKeyConclude, head, "q", "q")) + require.NoError(t, c.publishBatchID(context.Background(), topickey.TopicKeyConclude, head, "q", "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..03d73854 100644 --- a/submitqueue/orchestrator/controller/speculate/snapshot.go +++ b/submitqueue/orchestrator/controller/speculate/snapshot.go @@ -14,7 +14,10 @@ package speculate -import "github.com/uber/submitqueue/submitqueue/entity" +import ( + "github.com/uber/submitqueue/submitqueue/entity" + "github.com/uber/submitqueue/submitqueue/extension/storage" +) // snapshot is one run's working state: the queue as it was read, plus the // changes the run has decided on so far. The store is read once — two @@ -22,21 +25,33 @@ import "github.com/uber/submitqueue/submitqueue/entity" // and everything the run concludes is folded back in here before anything // else is derived from it. type snapshot struct { + // store is the queue-scoped store aggregate this run reads and writes + // through, resolved once from the triggering message's queue. + store storage.Storage // batches is every batch the run can reason about, by ID: the queue's // 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 +62,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 634f94e1..a9720d95 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" @@ -31,32 +30,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 @@ -94,8 +70,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() @@ -126,339 +117,44 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er return fmt.Errorf("payload queue %q does not match queue %q of batch %s", bid.Queue, batch.Queue, batch.ID) } - // 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, store, 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) - // Repair the membership record for the same crash window: a prior - // attempt may have CAS'd to terminal without completing the record move. - if err := corebatch.EnsureRecord(ctx, store, batch); err != nil { - metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) + if err := c.fanout(ctx, batch.ID, batch.Queue); err != nil { return err } - if batch.State == entity.BatchStateCancelled { - if err := c.respeculateDependents(ctx, store, batch); err != nil { - return err - } - } - return c.fanout(ctx, batch.ID, batch.Queue) - } - - // 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, store, batch) - if err != nil { - return err - } - batch = admitted - } - - if err := c.run(ctx, store, batch.Queue); err != nil { - return err + return c.run(ctx, store, batch) } - // 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) - // A redelivery can land here after a tryFinalize attempt crashed - // between its CAS and the record move; repair before acking. - if err := corebatch.EnsureRecord(ctx, store, batch); err != nil { - metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) + if batch.State == entity.BatchStateCreated { + if _, err := c.admit(ctx, store, batch); err != nil { return err } - return nil - } - - if batch.State == entity.BatchStateSpeculating { - return c.tryFinalize(ctx, store, 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, store, 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, store storage.Storage, batch entity.Batch) (entity.Batch, error) { - newVersion := batch.Version + 1 - batch.State = entity.BatchStateSpeculating - if err := store.GetBatchStore().Update(ctx, batch, batch.Version, newVersion); err != nil { + // Through Transition for the same reason as applyOutcome: the membership + // record has to leave the created bucket with the batch. + updated, err := corebatch.Transition(ctx, store, batch, entity.BatchStateSpeculating) + if err != nil { metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) - return batch, fmt.Errorf("failed to update batch %s state to speculating: %w", batch.ID, err) + return batch, err } - batch.Version = newVersion - batch.State = entity.BatchStateSpeculating metrics.NamedCounter(c.metricsScope, opName, "admitted", 1) c.logger.Infow("admitted batch to speculation", - "batch_id", batch.ID, - "queue", batch.Queue, - "dependencies", batch.Dependencies, - ) - 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, store storage.Storage, batch entity.Batch) error { - deps, err := c.fetchDependencies(ctx, store, 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, store, 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, batch.Queue); err != nil { - metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1) - return fmt.Errorf("failed to publish to merge: %w", err) - } - - if _, err := corebatch.Transition(ctx, store, batch, entity.BatchStateMerging); err != nil { - metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) - return 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, store storage.Storage, 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), + "batch_id", updated.ID, + "queue", updated.Queue, + "dependencies", updated.Dependencies, ) - - batch, err := corebatch.Transition(ctx, store, batch, entity.BatchStateFailed) - if err != nil { - metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) - return err - } - - if err := c.publishBatchID(ctx, topickey.TopicKeyConclude, batch.ID, batch.Queue, 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, store storage.Storage, 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, store, batch); err != nil { - return err - } - - batch, err := corebatch.Transition(ctx, store, batch, entity.BatchStateCancelled) - if err != nil { - metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) - return err - } - - if err := c.respeculateDependents(ctx, store, batch); err != nil { - return err - } - - if err := c.publishBatchID(ctx, topickey.TopicKeyConclude, batch.ID, batch.Queue, 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, store storage.Storage, batch entity.Batch) error { - build, err := 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 := 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, store storage.Storage, batch entity.Batch) error { - bd, err := 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, 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, store storage.Storage, batch entity.Batch) ([]entity.Batch, error) { - deps := make([]entity.Batch, 0, len(batch.Dependencies)) - for _, depID := range batch.Dependencies { - d, err := 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 + return updated, nil } // fanout re-publishes downstream events for a batch that has already reached @@ -473,9 +169,12 @@ func (c *Controller) fanout(ctx context.Context, batchID, queue string) error { } // publishBatchID publishes a batch ID to the topic behind key, stamped with -// the batch's queue and partitioned by partitionKey. 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. +// the batch's queue and partitioned by partitionKey. +// +// 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. // // queue and partitionKey are separate because they answer different questions. // queue is what the payload asserts the batch belongs to, and the consumer @@ -488,7 +187,7 @@ func (c *Controller) publishBatchID(ctx context.Context, key consumer.TopicKey, if err != nil { return fmt.Errorf("failed to serialize batch ID: %w", err) } - return publish.Message(ctx, c.registry, key, batchID, payload, partitionKey) + return publish.Message(ctx, c.registry, key, publish.UniqueID(batchID), payload, partitionKey) } // 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 c66f203b..d5b34a6a 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" @@ -25,7 +24,6 @@ import ( entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" consumermock "github.com/uber/submitqueue/platform/consumer/mock" - "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" @@ -36,35 +34,15 @@ import ( "go.uber.org/zap/zaptest" ) -// newQueueBatchStateStore returns a QueueBatchStateStore mock that accepts any -// membership-record write; these tests never list record buckets. -// staticStorageFactory resolves every queue to one fixed store aggregate. -type staticStorageFactory struct{ store storage.Storage } - -// For returns the fixed store aggregate for any queue. -func (f staticStorageFactory) For(storage.Config) (storage.Storage, error) { return f.store, nil } - -func newQueueBatchStateStore(ctrl *gomock.Controller) *storagemock.MockQueueBatchStateStore { - s := storagemock.NewMockQueueBatchStateStore(ctrl) - s.EXPECT().Put(gomock.Any(), gomock.Any()).Return(nil).AnyTimes() - s.EXPECT().Delete(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes() - // An empty List keeps the speculation run quiet: the queue lists no - // in-flight batches, so the run returns before reading any path set or - // asking the Speculator. run_test.go covers the run itself. - s.EXPECT().List(gomock.Any(), gomock.Any()).Return(nil, nil).AnyTimes() - return s +// 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 } -func batchWithState(batch entity.Batch, state entity.BatchState) entity.Batch { - batch.State = state - return 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 } @@ -75,612 +53,238 @@ func (f staticSpeculatorFactory) For(speculator.Config) (speculator.Speculator, return f.s, nil } -// batchIDPayload serializes a BatchID to JSON bytes for test message payloads. +// staticStorageFactory resolves every queue to one fixed store aggregate. +type staticStorageFactory struct{ store storage.Storage } + +// For returns the fixed store aggregate for any queue. +func (f staticStorageFactory) For(storage.Config) (storage.Storage, error) { return f.store, nil } + +// listsInFlight makes the queue read return exactly these batches: each is +// filed under its state's membership bucket and hydrated back through the +// batch store. With no arguments the run is quiet — the queue lists no +// in-flight batches, so the run returns before reading any path set or asking +// the Speculator; run_test.go covers the run itself. +func (h *procHarness) listsInFlight(batches ...entity.Batch) { + h.queueStates.EXPECT().List(gomock.Any(), gomock.Any()).DoAndReturn( + func(_ context.Context, state entity.BatchState) ([]entity.QueueBatchState, error) { + var records []entity.QueueBatchState + for _, b := range batches { + if b.State == state { + records = append(records, entity.QueueBatchState{Queue: b.Queue, State: state, BatchID: b.ID}) + } + } + return records, nil + }, + ).AnyTimes() + for _, b := range batches { + h.batches.EXPECT().Get(gomock.Any(), b.ID).Return(b, nil).AnyTimes() + } +} + 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 + queueStates *storagemock.MockQueueBatchStateStore + 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.queueStates = storagemock.NewMockQueueBatchStateStore(ctrl) + h.queueStates.EXPECT().Put(gomock.Any(), gomock.Any()).Return(nil).AnyTimes() + h.queueStates.EXPECT().Delete(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes() + 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().GetQueueBatchStateStore().Return(h.queueStates).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, staticStorageFactory{store: store}, staticSpeculatorFactory{s: quietSpeculator{}}, registry, topickey.TopicKeySpeculate, "orchestrator-speculate") + h.controller = NewController( + zaptest.NewLogger(t).Sugar(), tally.NoopScope, staticStorageFactory{store: 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 := consumermock.NewMockDelivery(ctrl) - delivery.EXPECT().Message().Return(msg).AnyTimes() - delivery.EXPECT().Attempt().Return(1).AnyTimes() - return controller.Process(context.Background(), delivery) + d := consumermock.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) - store.EXPECT().GetQueueBatchStateStore().Return(newQueueBatchStateStore(ctrl)).AnyTimes() - controller := newTestController(t, ctrl, store, nil) + h := newProcHarness(t, ctrl, 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()) + assert.Equal(t, topickey.TopicKeySpeculate, h.controller.TopicKey()) + assert.Equal(t, "orchestrator-speculate", h.controller.ConsumerGroup()) + assert.Equal(t, "speculate", h.controller.Name()) - var _ consumer.Controller = controller + var _ consumer.Controller = h.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().GetQueueBatchStateStore().Return(newQueueBatchStateStore(ctrl)).AnyTimes() - store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() - - 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) { +// 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) - batch := testBatch(entity.BatchStateSpeculating) + h := newProcHarness(t, ctrl, nil) + batch := testBatch(entity.BatchStateCreated) - 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) + 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) + h.listsInFlight() - store := storagemock.NewMockStorage(ctrl) - store.EXPECT().GetQueueBatchStateStore().Return(newQueueBatchStateStore(ctrl)).AnyTimes() - store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() - - 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") } -// 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().GetQueueBatchStateStore().Return(newQueueBatchStateStore(ctrl)).AnyTimes() - store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() - - 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) - - 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. - - store := storagemock.NewMockStorage(ctrl) - store.EXPECT().GetQueueBatchStateStore().Return(newQueueBatchStateStore(ctrl)).AnyTimes() - store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() - - 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().GetQueueBatchStateStore().Return(newQueueBatchStateStore(ctrl)).AnyTimes() - store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() - - controller := newTestController(t, ctrl, store, nil) - require.NoError(t, runProcess(t, ctrl, controller, batch.ID)) -} - -// 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) { - 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) - - store := storagemock.NewMockStorage(ctrl) - store.EXPECT().GetQueueBatchStateStore().Return(newQueueBatchStateStore(ctrl)).AnyTimes() - store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() - - 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().GetQueueBatchStateStore().Return(newQueueBatchStateStore(ctrl)).AnyTimes() - store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() - - controller := newTestController(t, ctrl, store, nil) - require.NoError(t, runProcess(t, ctrl, controller, batch.ID)) -} - -// 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. - - store := storagemock.NewMockStorage(ctrl) - store.EXPECT().GetQueueBatchStateStore().Return(newQueueBatchStateStore(ctrl)).AnyTimes() - store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() - - // 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) + h.batches.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil) + h.listsInFlight() - logger := zaptest.NewLogger(t).Sugar() - controller := NewController(logger, tally.NoopScope, staticStorageFactory{store: 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().GetQueueBatchStateStore().Return(newQueueBatchStateStore(ctrl)).AnyTimes() - store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() - 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, staticStorageFactory{store: 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) + h := newProcHarness(t, ctrl, nil) + batch := testBatch(entity.BatchStateSucceeded) - buildStore := storagemock.NewMockBuildStore(ctrl) - build := entity.Build{ - ID: batch.ID, BatchID: batch.ID, Status: entity.BuildStatusRunning, + dependent := entity.Batch{ + ID: "test-queue/batch/2", Queue: "test-queue", + State: entity.BatchStateSpeculating, Dependencies: []string{batch.ID}, Version: 1, } - 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) - store := storagemock.NewMockStorage(ctrl) - store.EXPECT().GetQueueBatchStateStore().Return(newQueueBatchStateStore(ctrl)).AnyTimes() - store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() - store.EXPECT().GetBuildStore().Return(buildStore).AnyTimes() - store.EXPECT().GetBatchDependentStore().Return(depStore).AnyTimes() - - 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) + h.batches.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil) + h.listsInFlight(dependent) + h.batches.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil) + h.pathSets.EXPECT().Get(gomock.Any(), dependent.ID). + Return(entity.SpeculationPathSet{}, storage.ErrNotFound) - logger := zaptest.NewLogger(t).Sugar() - controller := NewController(logger, tally.NoopScope, staticStorageFactory{store: 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().GetQueueBatchStateStore().Return(newQueueBatchStateStore(ctrl)).AnyTimes() - store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() - 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)) + 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") } -// 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) { +// 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) - 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().GetQueueBatchStateStore().Return(newQueueBatchStateStore(ctrl)).AnyTimes() - store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() - 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().GetQueueBatchStateStore().Return(newQueueBatchStateStore(ctrl)).AnyTimes() - store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() - 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, staticStorageFactory{store: 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().GetQueueBatchStateStore().Return(newQueueBatchStateStore(ctrl)).AnyTimes() - store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() - 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, staticStorageFactory{store: 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) -} - -// 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().GetQueueBatchStateStore().Return(newQueueBatchStateStore(ctrl)).AnyTimes() - store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() - - 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) { - ctrl := gomock.NewController(t) - - batchStore := storagemock.NewMockBatchStore(ctrl) - batchStore.EXPECT().Get(gomock.Any(), "test-queue/batch/1").Return(entity.Batch{}, fmt.Errorf("db connection lost")) + h := newProcHarness(t, ctrl, nil) + batch := testBatch(entity.BatchStateMerging) - store := storagemock.NewMockStorage(ctrl) - store.EXPECT().GetQueueBatchStateStore().Return(newQueueBatchStateStore(ctrl)).AnyTimes() - store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() + h.batches.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil) + h.listsInFlight() - 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 := consumermock.NewMockDelivery(ctrl) + d.EXPECT().Message().Return(msg).AnyTimes() + d.EXPECT().Attempt().Return(1).AnyTimes() - store := storagemock.NewMockStorage(ctrl) - store.EXPECT().GetQueueBatchStateStore().Return(newQueueBatchStateStore(ctrl)).AnyTimes() - store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() + 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) - store.EXPECT().GetQueueBatchStateStore().Return(newQueueBatchStateStore(ctrl)).AnyTimes() - 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 := consumermock.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)) + }) }