diff --git a/CLAUDE.md b/CLAUDE.md index 941840a8..8d174e7a 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -30,6 +30,13 @@ if err := store.UpdateStatus(ctx, request.ID, request.Version, newVersion, newSt request.Version = newVersion ``` +**Speculation paths and trees — O(paths), never 2^N:** + +A batch's speculation tree persists one entry per enumerated path, and every downstream pass (speculate's reconcile, prioritize rounds, the build stage's per-path loop) re-walks the whole tree. Code that touches paths or trees must respect two bounds: + +1. **Path count is the enumerator's contract to bound.** The enumerator implementation decides which paths — and how many — a batch gets; expect several per batch (the single chain path is a naive parity baseline, not the design target). The hard rule: a bounded frontier of assumption sets, never the 2^N power set of a batch's N dependencies. Any algorithm that wants to "consider all subsets" of the dependencies is a design smell — enumerate a bounded candidate set and push ranking/pruning into the scorer/selector/prioritizer seams. +2. **Per-pass work stays O(paths) with point reads.** A pass over a tree may spend at most O(1) point reads per unresolved path (e.g. path->build mapping, then build), must skip settled paths (terminal statuses) without I/O, and must not fan out more than O(paths) messages. The tree is persisted as a single row, so an unbounded path set blows up the write path as well as every read. + ## Architecture ### Project Layout diff --git a/submitqueue/orchestrator/controller/speculate/speculate.go b/submitqueue/orchestrator/controller/speculate/speculate.go index 04971601..239444d8 100644 --- a/submitqueue/orchestrator/controller/speculate/speculate.go +++ b/submitqueue/orchestrator/controller/speculate/speculate.go @@ -35,7 +35,8 @@ import ( // Controller handles speculate queue messages. // -// Each invocation reconciles the batch's entity.SpeculationTree, publishes +// Each invocation reconciles the batch's entity.SpeculationTree — against +// the build stage's outcomes and the current dependency states — publishes // the batch's queue to prioritize, and advances the batch one step in the // state machine. Tree reconciliation is pure mechanics: the controller runs // the queue's configured seams — enumerator (path structure), path scorer @@ -43,17 +44,18 @@ import ( // applies them to the persisted tree, and writes it back under optimistic // concurrency. Which paths exist, how they score, and which are promoted or // cancelled are the seams' decisions alone; the controller never originates -// one. The batch's own merge timing is still driven by dependency states -// (tryFinalize), pending a later change that gates it on the tree's path -// outcomes. +// one. Finalization is driven by the tree itself: publish to merge as soon +// as any path is actually mergeable, or fail the batch once no path can +// ever merge again. // // Per invocation, the controller advances the batch one step in the state // machine: // // - Created, Scored, or Speculating → speculateBatch: reconcile the tree -// (created on the first pass the dependency gate admits), CAS to -// Speculating for Created/Scored, publish to prioritize every pass, and -// run tryFinalize once the batch has reached Speculating. +// (created on the first pass the dependency gate admits) against build +// outcomes and dead dependencies, CAS to Speculating for Created/Scored, +// publish to prioritize every pass, and finalize — merge a mergeable +// path, fail the batch if none remain viable, or wait. // - Cancelling → cancel any in-flight Build entity, respeculate // dependents, CAS to terminal Cancelled, publish to conclude. The // cancel controller hands the batch off in this state and speculate @@ -147,12 +149,13 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r } // Terminal state: wake dependents and re-publish conclude. This branch is - // the dependent wake-up for success and failure — mergesignal routes every - // terminal transition back through speculate under the batch's own ID, and - // re-publishing the dependents here lets each of them re-run its own - // dependency gate / tryFinalize against the new dependency state. On - // redelivery the same branch doubles as the crash self-heal: both the - // dependent fan-out and the conclude publish are idempotent re-sends. + // the dependent wake-up for every terminal transition — mergesignal routes + // merge outcomes back through speculate under the batch's own ID, the + // finalize pass below re-publishes a batch it fails, and re-publishing the + // dependents here lets each of them re-run its own dependency gate / + // finalize pass against the new dependency state. On redelivery the same + // branch doubles as the crash self-heal: both the dependent fan-out and + // the conclude publish are idempotent re-sends. if batch.State.IsTerminal() { metrics.NamedCounter(c.metricsScope, opName, "self_heal_terminal", 1) if err := c.respeculateDependents(ctx, batch); err != nil { @@ -178,15 +181,20 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r // speculateBatch is the unified entry point for Created, Scored, and // Speculating batches. It loads the batch's speculation tree (creating it on -// the first pass the dependency gate admits), applies the scorer's and -// selector's outputs, persists the tree if anything changed, CASes the batch -// to Speculating on its first pass, publishes the queue to prioritize every -// pass, and runs tryFinalize once the batch has reached Speculating. +// the first pass the dependency gate admits), reconciles it against build +// outcomes and dead dependencies, applies the scorer's and selector's +// outputs, persists the tree if anything changed, CASes the batch to +// Speculating on its first pass, publishes the queue to prioritize every +// pass, and then finalizes the batch from the tree. func (c *Controller) speculateBatch(ctx context.Context, batch entity.Batch) error { deps, err := c.fetchDependencies(ctx, batch) if err != nil { return err } + depByID := make(map[string]entity.Batch, len(deps)) + for _, d := range deps { + depByID[d.ID] = d + } tree, err := c.store.GetSpeculationTreeStore().Get(ctx, batch.ID) if err != nil { @@ -196,7 +204,8 @@ func (c *Controller) speculateBatch(ctx context.Context, batch entity.Batch) err } // Dependency gate: only consulted before a tree exists. Once a batch - // has a tree, later passes just re-score/re-select it. + // has a tree, the tree itself reconciles dependency outcomes on every + // pass, so the gate no longer applies. blocked, gerr := c.dependencyGateBlocks(ctx, batch, deps) if gerr != nil { return gerr @@ -216,6 +225,11 @@ func (c *Controller) speculateBatch(ctx context.Context, batch entity.Batch) err } } + tree, reconChanged, err := c.reconcile(ctx, batch, tree, depByID) + if err != nil { + return err + } + scr, err := c.scorers.For(pathscorer.Config{QueueName: batch.Queue}) if err != nil { metrics.NamedCounter(c.metricsScope, opName, "scorer_errors", 1) @@ -241,7 +255,7 @@ func (c *Controller) speculateBatch(ctx context.Context, batch entity.Batch) err tree, selectionChanged := c.applySelection(batch, tree, decisions) - if scoresChanged || selectionChanged { + if reconChanged || scoresChanged || selectionChanged { newVersion := tree.Version + 1 if err := c.store.GetSpeculationTreeStore().Update(ctx, batch.ID, tree.Version, newVersion, tree.Paths); err != nil { metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) @@ -252,7 +266,6 @@ func (c *Controller) speculateBatch(ctx context.Context, batch entity.Batch) err // Optimistic CAS: if the version has already advanced (concurrent // speculate), the next event will see the new state and behave correctly. - originalState := batch.State if batch.State == entity.BatchStateCreated || batch.State == entity.BatchStateScored { newVersion := batch.Version + 1 if err := c.store.GetBatchStore().UpdateState(ctx, batch.ID, batch.Version, newVersion, entity.BatchStateSpeculating); err != nil { @@ -274,14 +287,7 @@ func (c *Controller) speculateBatch(ctx context.Context, batch entity.Batch) err return fmt.Errorf("failed to publish queue %s to prioritize: %w", batch.Queue, err) } - // Legacy forward path: tryFinalize only runs once a batch has already - // reached Speculating on a prior pass — merge timing is unchanged from - // before prioritize/build existed. This tightens in a later commit to - // gate on the tree's own path outcome instead of dependency states alone. - if originalState == entity.BatchStateSpeculating { - return c.tryFinalize(ctx, batch) - } - return nil + return c.finalize(ctx, batch, tree, depByID) } // dependencyGateBlocks reports whether batch's count of active dependencies @@ -442,6 +448,195 @@ func (c *Controller) applyScores(batch entity.Batch, tree entity.SpeculationTree return tree, changed } +// reconcile folds each path's build outcome into its status, then marks any +// non-terminal, non-Cancelling path dead if pathDead reports its assumption +// has become invalid. +// +// Dead-path cancellation is captured as intent only, mirroring +// applySelection: a Building path drops to Cancelling with no runner call — +// nothing in this file talks to a build runner. The build stage enacts the +// intent on a later pass, which it reliably gets: speculateBatch publishes to +// prioritize on every pass regardless of whether anything changed, and +// prioritize republishes to build for every batch whose tree still carries a +// Cancelling path, not just ones it touched this round. A path with no build +// yet drops straight to Cancelled — there is nothing running to enact. +// +// The returned bool reports whether any path actually changed, so the caller +// can skip the conditional write on a no-op pass. +// +// Cost: one pass over the tree with at most two point reads (path->build +// mapping, then build) per path that is still unresolved — paths already at +// a terminal status (Passed/Failed/Cancelled) are settled and skipped +// without touching storage. Statuses are folded into the tree's own path +// slice in place, mirroring applyScores/applySelection — no per-pass copy. +// The tree's path count itself is the enumerator's to decide and bound: +// implementations are expected to return several paths per batch, but +// always a bounded frontier — never the 2^N power set of dependency +// subsets (see extension/speculation/enumerator) — so a reconcile pass +// stays O(paths) in reads and CPU, allocating nothing. +func (c *Controller) reconcile(ctx context.Context, batch entity.Batch, tree entity.SpeculationTree, depByID map[string]entity.Batch) (entity.SpeculationTree, bool, error) { + paths := tree.Paths + changed := false + + for i, p := range paths { + if isTerminalPathStatus(p.Status) { + // Settled outcome: reconcileStatus never downgrades a terminal + // path and the dead-path sweep below skips them too, so there + // is nothing to read or fold for this path. + continue + } + pb, err := c.store.GetSpeculationPathBuildStore().Get(ctx, p.ID) + if err != nil { + if errors.Is(err, storage.ErrNotFound) { + continue + } + metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) + return entity.SpeculationTree{}, false, fmt.Errorf("failed to get path->build mapping for path %s of batch %s: %w", p.ID, batch.ID, err) + } + + b, err := c.store.GetBuildStore().Get(ctx, pb.BuildID) + if err != nil { + if errors.Is(err, storage.ErrNotFound) { + // Invariant breach: the write order in the build controller + // guarantees a mapping is only created once its build row + // exists. Treat this path like "no build yet" rather than + // crashing or nack-looping. + metrics.NamedCounter(c.metricsScope, opName, "mapping_dangling", 1) + c.logger.Warnw("path->build mapping points at a missing build; treating path as unbuilt", + "batch_id", batch.ID, + "path_id", p.ID, + "build_id", pb.BuildID, + ) + continue + } + metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) + return entity.SpeculationTree{}, false, fmt.Errorf("failed to get build %s for path %s of batch %s: %w", pb.BuildID, p.ID, batch.ID, err) + } + newStatus := reconcileStatus(p.Status, b.Status) + if p.BuildID != b.ID || p.Status != newStatus { + p.BuildID = b.ID + p.Status = newStatus + paths[i] = p + changed = true + } + } + + for i, p := range paths { + if isTerminalPathStatus(p.Status) || p.Status == entity.SpeculationPathStatusCancelling { + continue + } + if !pathDead(p.Path, depByID) { + continue + } + + if p.Status == entity.SpeculationPathStatusBuilding { + p.Status = entity.SpeculationPathStatusCancelling + metrics.NamedCounter(c.metricsScope, opName, "path_dead_cancelling", 1) + } else { + p.Status = entity.SpeculationPathStatusCancelled + metrics.NamedCounter(c.metricsScope, opName, "path_dead_cancelled", 1) + } + paths[i] = p + changed = true + } + + tree.Paths = paths + return tree, changed, nil +} + +// reconcileStatus folds a build outcome into a path's status. Pure O(1) +// status arithmetic — no I/O; reconcile calls it once per unresolved path. +// +// Two rules shape it: +// +// - A terminal path status (Passed/Failed/Cancelled) is a settled outcome +// and is never downgraded, whatever the build now reports. +// - Cancelling records a decision about the PATH — it was deselected, or +// its merge assumptions died — not a prediction about the build's +// outcome. Once recorded, the path can never merge; the only open +// question is when the runner side quiesces. So a Cancelling path holds +// until its build reaches ANY terminal state and then settles to +// Cancelled — including a build that raced to Succeeded before the +// cancel intent was enacted: its result is moot for a path that will +// never merge. (The build stage only calls runner.Cancel on builds that +// are still running; an already-terminal build is left untouched, and +// this mapping simply settles the path.) +// +// For a live (non-Cancelling) path, the build outcome maps directly: +// Succeeded -> Passed, Failed -> Failed, Cancelled -> Cancelled, +// Accepted/Running -> Building. +func reconcileStatus(current entity.SpeculationPathStatus, build entity.BuildStatus) entity.SpeculationPathStatus { + if isTerminalPathStatus(current) { + return current + } + if current == entity.SpeculationPathStatusCancelling { + if build.IsTerminal() { + return entity.SpeculationPathStatusCancelled + } + return entity.SpeculationPathStatusCancelling + } + switch build { + case entity.BuildStatusSucceeded: + return entity.SpeculationPathStatusPassed + case entity.BuildStatusFailed: + return entity.SpeculationPathStatusFailed + case entity.BuildStatusCancelled: + return entity.SpeculationPathStatusCancelled + case entity.BuildStatusAccepted, entity.BuildStatusRunning: + return entity.SpeculationPathStatusBuilding + default: + return current + } +} + +// isTerminalPathStatus reports whether s is a settled path outcome that must +// never be overwritten by a later reconcile pass. +func isTerminalPathStatus(s entity.SpeculationPathStatus) bool { + switch s { + case entity.SpeculationPathStatusPassed, entity.SpeculationPathStatusFailed, entity.SpeculationPathStatusCancelled: + return true + default: + return false + } +} + +// pathDead reports whether path can never merge — its assumption set has +// been invalidated by a dependency outcome: +// +// - a base dependency reached Failed: the path built on top of changes +// that will never land, so its build result can never back a merge; +// - a dependency of the head OUTSIDE the base reached Succeeded: the path +// bet that batch would not land first, and it did. +// +// A Cancelled base dependency deliberately does NOT dead the path (Phase-A +// leniency): the cancelled batch will never land, so it can no longer +// conflict, and the path's build being stale against the original +// (uncancelled) assumption set is accepted — this mirrors the pre-tree +// chain semantics. With exactly one path per batch today (chain +// enumerator), deading that only path on a benign cancel would fail batches +// that currently survive. Once multi-path enumeration lands and a sibling +// path can pick up the slack, a Cancelled base dependency should dead the +// path like a Failed one — tracked in +// https://github.com/uber/submitqueue/issues/369. +func pathDead(path entity.SpeculationPath, depByID map[string]entity.Batch) bool { + inBase := make(map[string]bool, len(path.Base)) + for _, id := range path.Base { + inBase[id] = true + if d, ok := depByID[id]; ok && d.State == entity.BatchStateFailed { + return true + } + } + for id, d := range depByID { + if inBase[id] { + continue + } + if d.State == entity.BatchStateSucceeded { + return true + } + } + return false +} + // applySelection applies the selector's decisions to tree.Paths, mutating it // in place. The selector owns the policy (which paths to promote or cancel); // this function owns only the bookkeeping, mapping each decision onto the @@ -525,99 +720,138 @@ func (c *Controller) applySelection(batch entity.Batch, tree entity.SpeculationT return tree, changed } -// 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: each dependency's own terminal pass re-publishes this batch to -// speculate (the terminal branch in Process), so waiting never needs a poll. +// finalize settles the batch's forward step from the tree. Exactly one of +// three outcomes applies per pass: // -// TODO: when a dependency fails we currently fail this batch outright. -// We will need to respeculate the failed paths — drop the failed dep -// from the chain and re-issue speculation for the surviving ordering(s) -// — instead of cascading the failure into requests that could still land. -func (c *Controller) tryFinalize(ctx context.Context, batch entity.Batch) error { - deps, err := c.fetchDependencies(ctx, batch) - if err != nil { - return err +// 1. Merge now: some path is mergeableNow (its build passed and its +// assumption set holds) — publish to merge and CAS the batch to +// Merging. +// 2. Wait: no path is mergeable yet, but at least one is still viable — +// no-op; the next event (a build outcome via buildsignal, a dependency +// going terminal) re-runs this evaluation. +// 3. Fail: no viable path remains — every path has Failed, been Cancelled, +// or gone dead with no chance of ever merging — so the batch can never +// merge. The !anyViable branch below CASes the batch to Failed, fans +// out to dependents, and publishes to conclude. +func (c *Controller) finalize(ctx context.Context, batch entity.Batch, tree entity.SpeculationTree, depByID map[string]entity.Batch) error { + for _, p := range tree.Paths { + if !mergeableNow(p, depByID) { + continue + } + + if err := c.publish(ctx, topickey.TopicKeyMerge, batch.ID, batch.Queue); err != nil { + metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1) + return fmt.Errorf("failed to publish to merge: %w", err) + } + + newVersion := batch.Version + 1 + if err := c.store.GetBatchStore().UpdateState(ctx, batch.ID, batch.Version, newVersion, entity.BatchStateMerging); err != nil { + metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) + return fmt.Errorf("failed to update batch %s state to merging: %w", batch.ID, err) + } + metrics.NamedCounter(c.metricsScope, opName, "merging", 1) + return nil } - 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. - c.metricsScope.Counter("dependency_cancelled_skipped").Inc(1) - c.logger.Infow("dependency cancelled; dropping from speculation chain", - "batch_id", batch.ID, - "dependency_id", d.ID, - ) - case entity.BatchStateFailed: - return c.failOnDependency(ctx, batch, d) - default: - pending = append(pending, d.ID) + anyViable := false + failedCount, cancelledCount := 0, 0 + for _, p := range tree.Paths { + if viable(p, depByID) { + anyViable = true + } + switch p.Status { + case entity.SpeculationPathStatusFailed: + failedCount++ + case entity.SpeculationPathStatusCancelled, entity.SpeculationPathStatusCancelling: + cancelledCount++ } } - if len(pending) > 0 { - metrics.NamedCounter(c.metricsScope, opName, "waiting_on_deps", 1) - c.logger.Debugw("dependencies still in flight; waiting", + if !anyViable { + c.logger.Warnw("no viable speculation path remains; failing batch", "batch_id", batch.ID, - "pending_dependency_ids", pending, + "failed_paths", failedCount, + "cancelled_paths", cancelledCount, ) - return nil - } + metrics.NamedCounter(c.metricsScope, opName, "no_viable_path", 1) - if err := c.publish(ctx, topickey.TopicKeyMerge, batch.ID, batch.Queue); err != nil { - metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1) - return fmt.Errorf("failed to publish to merge: %w", err) - } + newVersion := batch.Version + 1 + if err := c.store.GetBatchStore().UpdateState(ctx, batch.ID, batch.Version, newVersion, entity.BatchStateFailed); err != nil { + metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) + return fmt.Errorf("failed to update batch %s state to failed: %w", batch.ID, err) + } + batch.Version = newVersion + batch.State = entity.BatchStateFailed + + // Fan out to dependents BEFORE concluding, mirroring cancelBatch's + // ordering. The fan-out is publish-only — one speculate wake-up per + // dependent, each processed asynchronously on its own delivery; + // nothing is re-evaluated inline here. A dependent only drops this + // batch from its tree (dead chain path) and lets a surviving path + // take over once it observes the terminal Failed state. Nothing else + // re-notifies them — a batch failed here never reaches mergesignal, + // whose fan-out covers the merge-failure flow. A crash after the CAS + // is recovered by the terminal self-heal branch, which re-runs this + // fan-out for Failed. + if err := c.respeculateDependents(ctx, batch); err != nil { + return err + } - newVersion := batch.Version + 1 - if err := c.store.GetBatchStore().UpdateState(ctx, batch.ID, batch.Version, newVersion, entity.BatchStateMerging); err != nil { - metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) - return fmt.Errorf("failed to update batch %s state to merging: %w", batch.ID, err) + if err := c.publish(ctx, topickey.TopicKeyConclude, batch.ID, batch.Queue); err != nil { + metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1) + return fmt.Errorf("failed to publish to conclude: %w", err) + } + return nil } + metrics.NamedCounter(c.metricsScope, opName, "waiting_on_deps", 1) + c.logger.Debugw("no path mergeable yet; waiting", "batch_id", batch.ID) 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. The batch's own dependents are -// woken right after the terminal CAS so the failure cascades downstream -// instead of stranding them mid-wait; a crash between the CAS and the -// fan-out is recovered by the terminal branch in Process on redelivery. -func (c *Controller) failOnDependency(ctx context.Context, batch entity.Batch, dep entity.Batch) error { - metrics.NamedCounter(c.metricsScope, opName, "dependency_failed", 1) - c.logger.Warnw("dependency in non-succeeding terminal state; failing batch", - "batch_id", batch.ID, - "dependency_id", dep.ID, - "dependency_state", string(dep.State), - ) - - newVersion := batch.Version + 1 - if err := c.store.GetBatchStore().UpdateState(ctx, batch.ID, batch.Version, newVersion, entity.BatchStateFailed); err != nil { - metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) - return fmt.Errorf("failed to update batch %s state to failed: %w", batch.ID, err) +// mergeableNow reports whether p is ready to merge right now: its own build +// Passed, every base dependency has either landed or been cancelled out of +// the way, and every dependency of the head NOT in the base has been ruled +// out (Failed or Cancelled) rather than still possibly landing. The last +// condition is what makes this stricter than a plain "base landed" check: a +// still in-flight non-base dependency might yet land and dead the path (see +// pathDead), so merging must wait for it to resolve one way or the other. +func mergeableNow(p entity.SpeculationPathInfo, depByID map[string]entity.Batch) bool { + if p.Status != entity.SpeculationPathStatusPassed { + return false } - if err := c.respeculateDependents(ctx, batch); err != nil { - return err + inBase := make(map[string]bool, len(p.Path.Base)) + for _, id := range p.Path.Base { + inBase[id] = true + d, ok := depByID[id] + if !ok { + continue + } + if d.State != entity.BatchStateSucceeded && d.State != entity.BatchStateCancelled { + return false + } } - - if err := c.publish(ctx, topickey.TopicKeyConclude, batch.ID, batch.Queue); err != nil { - metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1) - return fmt.Errorf("failed to publish to conclude: %w", err) + for id, d := range depByID { + if inBase[id] { + continue + } + if d.State != entity.BatchStateFailed && d.State != entity.BatchStateCancelled { + return false + } } + return true +} - return nil +// viable reports whether p could still merge in the future: it has not +// reached a non-succeeding terminal or cancelling status, and it is not +// dead. A batch with no viable path left can never merge and is failed. +func viable(p entity.SpeculationPathInfo, depByID map[string]entity.Batch) bool { + switch p.Status { + case entity.SpeculationPathStatusFailed, entity.SpeculationPathStatusCancelled, entity.SpeculationPathStatusCancelling: + return false + } + return !pathDead(p.Path, depByID) } // cancelBatch drives a batch from BatchStateCancelling to BatchStateCancelled. @@ -634,9 +868,9 @@ func (c *Controller) failOnDependency(ctx context.Context, batch entity.Batch, d // 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. +// dependent fan-out: finalize 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 @@ -737,11 +971,12 @@ func (c *Controller) respeculateDependents(ctx context.Context, batch entity.Bat 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. + // equivalent of speculateBatch) 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.publish(ctx, topickey.TopicKeySpeculate, depID, batch.Queue); err != nil { metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1) return fmt.Errorf("failed to publish dependent batch %s to speculate: %w", depID, err) diff --git a/submitqueue/orchestrator/controller/speculate/speculate_test.go b/submitqueue/orchestrator/controller/speculate/speculate_test.go index 9af43920..830108f6 100644 --- a/submitqueue/orchestrator/controller/speculate/speculate_test.go +++ b/submitqueue/orchestrator/controller/speculate/speculate_test.go @@ -71,18 +71,19 @@ func testBatch(state entity.BatchState, deps ...string) entity.Batch { // that tests script per case. Every publish is recorded so tests can assert // both content and order. There is no build-runner seam yet: nothing in this // package's speculate.go calls one — cancel decisions on the speculation tree -// are captured as intent only. +// (both applySelection's and reconcile's) are captured as intent only. type testHarness struct { - controller *Controller - batchStore *storagemock.MockBatchStore - treeStore *storagemock.MockSpeculationTreeStore - buildStore *storagemock.MockBuildStore - depStore *storagemock.MockBatchDependentStore - enum *enumeratorfake.Enumerator - scorer *scorerfake.Scorer - selector *selectorfake.Selector - depLimit *dependencylimitfake.DependencyLimit - records *[]pubRec + controller *Controller + batchStore *storagemock.MockBatchStore + treeStore *storagemock.MockSpeculationTreeStore + buildStore *storagemock.MockBuildStore + pathBuildStore *storagemock.MockSpeculationPathBuildStore + depStore *storagemock.MockBatchDependentStore + enum *enumeratorfake.Enumerator + scorer *scorerfake.Scorer + selector *selectorfake.Selector + depLimit *dependencylimitfake.DependencyLimit + records *[]pubRec } // newTestHarness builds a harness whose seams default to parity behavior: @@ -117,12 +118,14 @@ func newHarness(t *testing.T, ctrl *gomock.Controller, publishErr error, failTop batchStore := storagemock.NewMockBatchStore(ctrl) treeStore := storagemock.NewMockSpeculationTreeStore(ctrl) buildStore := storagemock.NewMockBuildStore(ctrl) + pathBuildStore := storagemock.NewMockSpeculationPathBuildStore(ctrl) depStore := storagemock.NewMockBatchDependentStore(ctrl) store := storagemock.NewMockStorage(ctrl) store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() store.EXPECT().GetSpeculationTreeStore().Return(treeStore).AnyTimes() store.EXPECT().GetBuildStore().Return(buildStore).AnyTimes() + store.EXPECT().GetSpeculationPathBuildStore().Return(pathBuildStore).AnyTimes() store.EXPECT().GetBatchDependentStore().Return(depStore).AnyTimes() enum := enumeratorfake.New() @@ -177,16 +180,17 @@ func newHarness(t *testing.T, ctrl *gomock.Controller, publishErr error, failTop ) return &testHarness{ - controller: c, - batchStore: batchStore, - treeStore: treeStore, - buildStore: buildStore, - depStore: depStore, - enum: enum, - scorer: scr, - selector: sel, - depLimit: dl, - records: &records, + controller: c, + batchStore: batchStore, + treeStore: treeStore, + buildStore: buildStore, + pathBuildStore: pathBuildStore, + depStore: depStore, + enum: enum, + scorer: scr, + selector: sel, + depLimit: dl, + records: &records, } } @@ -258,11 +262,10 @@ func TestController_Process_MergingNoOp(t *testing.T) { } // Terminal states wake dependents (re-publish speculate for each) and then -// re-publish conclude — this is both the normal dependent wake-up (mergesignal -// routes every terminal transition of a batch back through speculate under -// the batch's own ID) and the crash self-heal for a lost publish. State must -// not change (no UpdateState), and neither BuildStore nor -// SpeculationTreeStore is touched. +// re-publish conclude — this is both the normal dependent wake-up (every +// terminal transition is routed back through speculate under the batch's own +// ID) and the crash self-heal for a lost publish. State must not change (no +// UpdateState), and neither BuildStore nor SpeculationTreeStore is touched. func TestController_Process_TerminalSelfHeals(t *testing.T) { for _, state := range []entity.BatchState{ entity.BatchStateSucceeded, @@ -314,7 +317,7 @@ func TestController_Process_TerminalSelfHealNoDependents(t *testing.T) { // A missing BatchDependent row is a storage invariant violation (the batch // controller creates the row before the batch itself), surfaced as an error -// so the message nacks — unlike a stale dependent entry, which is tolerated. +// so the message nacks — the fan-out must not be silently skipped. func TestController_Process_TerminalMissingDependentRowErrors(t *testing.T) { ctrl := gomock.NewController(t) h := newTestHarness(t, ctrl) @@ -328,10 +331,11 @@ func TestController_Process_TerminalMissingDependentRowErrors(t *testing.T) { } // Created batch: no tree yet -> enumerate, stamp Candidate, mint path IDs, -// Version 1, Create. Score echoes. Select promotes the lone path to Selected -// -> tree changed -> Update(1,2,...). The forward step then CASes Created -> -// Speculating and publishes the queue to prioritize; tryFinalize does not run -// on this first pass (it only runs once the batch has reached Speculating). +// Version 1, Create. Reconcile is a no-op (no build mapping yet). Score +// echoes. Select promotes the lone path to Selected -> tree changed -> +// Update(1,2,...). The forward step then CASes Created -> Speculating and +// publishes the queue to prioritize. No path is mergeable yet (Selected, not +// Passed) and the lone path is still viable, so finalize just waits. func TestController_Process_CreatedEnumeratesScoresSelects(t *testing.T) { ctrl := gomock.NewController(t) h := newTestHarness(t, ctrl) @@ -350,6 +354,7 @@ func TestController_Process_CreatedEnumeratesScoresSelects(t *testing.T) { assert.Equal(t, int32(1), tree.Version) return nil }) + h.pathBuildStore.EXPECT().Get(gomock.Any(), fmt.Sprintf("%s/path/0", batch.ID)).Return(entity.SpeculationPathBuild{}, storage.ErrNotFound) h.selector.SetDecisions(entity.PathDecision{PathID: fmt.Sprintf("%s/path/0", batch.ID), Action: entity.SpeculationPathActionPromote}) h.treeStore.EXPECT().Update(gomock.Any(), batch.ID, int32(1), int32(2), gomock.Any()). @@ -386,6 +391,8 @@ func TestController_Process_CreateTreeMintsPathIDs(t *testing.T) { } return nil }) + h.pathBuildStore.EXPECT().Get(gomock.Any(), fmt.Sprintf("%s/path/0", batch.ID)).Return(entity.SpeculationPathBuild{}, storage.ErrNotFound) + h.pathBuildStore.EXPECT().Get(gomock.Any(), fmt.Sprintf("%s/path/1", batch.ID)).Return(entity.SpeculationPathBuild{}, storage.ErrNotFound) h.batchStore.EXPECT().UpdateState(gomock.Any(), batch.ID, int32(1), int32(2), entity.BatchStateSpeculating).Return(nil) require.NoError(t, runProcess(t, ctrl, h.controller, batch.ID)) @@ -419,6 +426,7 @@ func TestController_Process_CreateTreeAlreadyExistsRereads(t *testing.T) { assert.NotEmpty(t, tree.Paths[0].ID, "minted path ID must not be empty") return storage.ErrAlreadyExists }) + h.pathBuildStore.EXPECT().Get(gomock.Any(), "test-queue/batch/1/path/0").Return(entity.SpeculationPathBuild{}, storage.ErrNotFound) // No changes this pass (selector decides nothing) -> no Update call. h.batchStore.EXPECT().UpdateState(gomock.Any(), batch.ID, int32(1), int32(2), entity.BatchStateSpeculating).Return(nil) @@ -427,8 +435,8 @@ func TestController_Process_CreateTreeAlreadyExistsRereads(t *testing.T) { } // Dependency gate: too many active dependencies for a batch with no tree yet -// blocks enumeration entirely (ack without creating a tree, no forward -// step either). A later dependency-terminal event re-triggers speculate. +// blocks enumeration entirely (ack without creating a tree, no forward step +// either). A later dependency-terminal event re-triggers speculate. func TestController_Process_DependencyGateBlocksTreeCreation(t *testing.T) { ctrl := gomock.NewController(t) h := newDependencyLimitHarness(t, ctrl, 1) @@ -448,15 +456,12 @@ func TestController_Process_DependencyGateBlocksTreeCreation(t *testing.T) { assert.Empty(t, *h.records) } -// A pass over an unchanged tree must not call Update. The batch still -// publishes the queue to prioritize every pass regardless of tree changes; -// tryFinalize waits on the still-pending dependency, so no merge/conclude -// publish happens. +// A pass over an unchanged tree must not call Update, but prioritize is still +// published every pass regardless of whether anything changed. func TestController_Process_NoChangeSkipsTreeUpdate(t *testing.T) { ctrl := gomock.NewController(t) h := newTestHarness(t, ctrl) - dep := entity.Batch{ID: "test-queue/batch/0", Queue: "test-queue", State: entity.BatchStateSpeculating, Version: 1} - batch := testBatch(entity.BatchStateSpeculating, dep.ID) + batch := testBatch(entity.BatchStateSpeculating) path := entity.SpeculationPath{Head: batch.ID} tree := entity.SpeculationTree{ BatchID: batch.ID, @@ -465,19 +470,17 @@ func TestController_Process_NoChangeSkipsTreeUpdate(t *testing.T) { } h.batchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil) - h.batchStore.EXPECT().Get(gomock.Any(), dep.ID).Return(dep, nil).AnyTimes() h.treeStore.EXPECT().Get(gomock.Any(), batch.ID).Return(tree, nil) + h.pathBuildStore.EXPECT().Get(gomock.Any(), "test-queue/batch/1/path/0").Return(entity.SpeculationPathBuild{}, storage.ErrNotFound) // Fake scorer echoes, fake selector decides nothing -> no change. - // No treeStore.Update expected. tryFinalize waits on the pending dep, so - // no UpdateState/merge publish either. + // No treeStore.Update, no batchStore.UpdateState expected (already Speculating). require.NoError(t, runProcess(t, ctrl, h.controller, batch.ID)) assert.Equal(t, []pubRec{{topic: "prioritize", msgID: batch.Queue}}, *h.records) } // A version mismatch on the speculation tree Update must surface as an error -// (nack; the round is recomputed on redelivery) and must not reach the -// forward step. +// (nack; the round is recomputed on redelivery) and must not reach finalize. func TestController_Process_TreeUpdateVersionMismatchErrors(t *testing.T) { ctrl := gomock.NewController(t) h := newTestHarness(t, ctrl) @@ -491,9 +494,10 @@ func TestController_Process_TreeUpdateVersionMismatchErrors(t *testing.T) { h.batchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil) h.treeStore.EXPECT().Get(gomock.Any(), batch.ID).Return(tree, nil) + h.pathBuildStore.EXPECT().Get(gomock.Any(), "test-queue/batch/1/path/0").Return(entity.SpeculationPathBuild{}, storage.ErrNotFound) h.selector.SetDecisions(entity.PathDecision{PathID: tree.Paths[0].ID, Action: entity.SpeculationPathActionPromote}) h.treeStore.EXPECT().Update(gomock.Any(), batch.ID, int32(1), int32(2), gomock.Any()).Return(storage.ErrVersionMismatch) - // tryFinalize must never run: no batchStore.UpdateState, no merge publish. + // finalize must never run: no batchStore.UpdateState, no merge publish. err := runProcess(t, ctrl, h.controller, batch.ID) require.Error(t, err) @@ -501,119 +505,210 @@ func TestController_Process_TreeUpdateVersionMismatchErrors(t *testing.T) { assert.Empty(t, *h.records) } -// tryFinalize: Speculating with no deps should publish to merge and CAS to -// Merging. The batch already has an (empty) speculation tree from its -// Created/Scored pass. Every pass also publishes the queue to prioritize, -// before tryFinalize's own merge publish. -func TestController_Process_FinalizeNoDeps(t *testing.T) { - ctrl := gomock.NewController(t) - h := newTestHarness(t, ctrl) - batch := testBatch(entity.BatchStateSpeculating) +// Merge gate: a Passed path with every base dependency landed (or cancelled) +// and every non-base dependency ruled out merges immediately; a pending +// non-terminal base dependency waits; a path whose own build has not yet +// passed waits even if every dependency has resolved favorably. +func TestController_Process_MergeGate(t *testing.T) { + tests := []struct { + name string + pathStatus entity.SpeculationPathStatus + depState entity.BatchState + wantMerge bool + }{ + {name: "passed_and_base_dep_succeeded_merges", pathStatus: entity.SpeculationPathStatusPassed, depState: entity.BatchStateSucceeded, wantMerge: true}, + {name: "passed_and_base_dep_pending_waits", pathStatus: entity.SpeculationPathStatusPassed, depState: entity.BatchStateSpeculating, wantMerge: false}, + {name: "building_and_base_dep_succeeded_waits", pathStatus: entity.SpeculationPathStatusBuilding, depState: entity.BatchStateSucceeded, wantMerge: false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + h := newTestHarness(t, ctrl) + dep := entity.Batch{ID: "test-queue/batch/0", Queue: "test-queue", State: tt.depState, Version: 1} + batch := testBatch(entity.BatchStateSpeculating, dep.ID) + path := entity.SpeculationPath{Base: []string{dep.ID}, Head: batch.ID} + tree := entity.SpeculationTree{ + BatchID: batch.ID, + Version: 2, + Paths: []entity.SpeculationPathInfo{{ID: "test-queue/batch/1/path/0", Path: path, Status: tt.pathStatus, BuildID: "build-1"}}, + } - h.batchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil) - h.treeStore.EXPECT().Get(gomock.Any(), batch.ID).Return(entity.SpeculationTree{BatchID: batch.ID, Version: 1, Paths: []entity.SpeculationPathInfo{}}, nil) - h.batchStore.EXPECT().UpdateState(gomock.Any(), batch.ID, int32(1), int32(2), entity.BatchStateMerging).Return(nil) + h.batchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil) + h.batchStore.EXPECT().Get(gomock.Any(), dep.ID).Return(dep, nil) + h.treeStore.EXPECT().Get(gomock.Any(), batch.ID).Return(tree, nil) + if !isTerminalPathStatus(tt.pathStatus) { + // Terminal paths are settled: reconcile skips their + // path->build read entirely. + h.pathBuildStore.EXPECT().Get(gomock.Any(), "test-queue/batch/1/path/0").Return(entity.SpeculationPathBuild{}, storage.ErrNotFound) + } - require.NoError(t, runProcess(t, ctrl, h.controller, batch.ID)) - assert.Equal(t, []pubRec{ - {topic: "prioritize", msgID: batch.Queue}, - {topic: "submitqueue-merge", msgID: batch.ID}, - }, *h.records) + if tt.wantMerge { + h.batchStore.EXPECT().UpdateState(gomock.Any(), batch.ID, int32(1), int32(2), entity.BatchStateMerging).Return(nil) + } + + require.NoError(t, runProcess(t, ctrl, h.controller, batch.ID)) + + wantRecords := []pubRec{{topic: "prioritize", msgID: "test-queue"}} + if tt.wantMerge { + wantRecords = append(wantRecords, pubRec{topic: "submitqueue-merge", msgID: batch.ID}) + } + assert.ElementsMatch(t, wantRecords, *h.records) + }) + } } -// tryFinalize: Speculating with all deps Succeeded should publish to merge -// and CAS to Merging. -func TestController_Process_FinalizeAllDepsSucceeded(t *testing.T) { +// A Cancelled base dependency is tolerated by mergeableNow exactly like +// Succeeded — parity with today's chain semantics (the cancelled batch never +// lands, so it cannot conflict). +func TestController_Process_CancelledBaseDepToleratedMerges(t *testing.T) { ctrl := gomock.NewController(t) h := newTestHarness(t, ctrl) - 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) + dep := entity.Batch{ID: "test-queue/batch/0", Queue: "test-queue", State: entity.BatchStateCancelled, Version: 1} + batch := testBatch(entity.BatchStateSpeculating, dep.ID) + path := entity.SpeculationPath{Base: []string{dep.ID}, Head: batch.ID} + tree := entity.SpeculationTree{ + BatchID: batch.ID, + Version: 1, + Paths: []entity.SpeculationPathInfo{{ID: "test-queue/batch/1/path/0", Path: path, Status: entity.SpeculationPathStatusPassed, BuildID: "build-1"}}, + } h.batchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil) - // speculateBatch fetches dependencies once for the pre-tree gate check, - // and tryFinalize fetches them again itself. - h.batchStore.EXPECT().Get(gomock.Any(), depA.ID).Return(depA, nil).Times(2) - h.batchStore.EXPECT().Get(gomock.Any(), depB.ID).Return(depB, nil).Times(2) - h.treeStore.EXPECT().Get(gomock.Any(), batch.ID).Return(entity.SpeculationTree{BatchID: batch.ID, Version: 1, Paths: []entity.SpeculationPathInfo{}}, nil) + h.batchStore.EXPECT().Get(gomock.Any(), dep.ID).Return(dep, nil) + h.treeStore.EXPECT().Get(gomock.Any(), batch.ID).Return(tree, nil) + // No path->build read: the path is already terminal (Passed), so + // reconcile skips it without touching storage. h.batchStore.EXPECT().UpdateState(gomock.Any(), batch.ID, int32(1), int32(2), entity.BatchStateMerging).Return(nil) require.NoError(t, runProcess(t, ctrl, h.controller, batch.ID)) - assert.Equal(t, []pubRec{ - {topic: "prioritize", msgID: batch.Queue}, + assert.ElementsMatch(t, []pubRec{ + {topic: "prioritize", msgID: "test-queue"}, {topic: "submitqueue-merge", msgID: batch.ID}, }, *h.records) } -// tryFinalize: Speculating with a dep still in flight is a no-op (no merge -// publish, no state change). The queue is still published to prioritize -// every pass. -func TestController_Process_WaitingOnDep(t *testing.T) { - ctrl := gomock.NewController(t) - h := newTestHarness(t, ctrl) - dep := entity.Batch{ID: "test-queue/batch/0", Queue: "test-queue", State: entity.BatchStateSpeculating, Version: 1} - batch := testBatch(entity.BatchStateSpeculating, dep.ID) +// A failed base dependency deads the path: a Building path is captured as a +// cancel intent (Cancelling) with no runner call (D1/D2) — nothing in this +// file talks to a build runner. Enactment is the build stage's job, reached +// via the prioritize round speculate publishes every pass. A pre-build path +// drops straight to Cancelled. Either way, with no viable path left the +// batch fails and conclude is published. +func TestController_Process_FailedBaseDepDeadsPath(t *testing.T) { + tests := []struct { + name string + pathStatus entity.SpeculationPathStatus + wantStatus entity.SpeculationPathStatus + }{ + {name: "building_path_marks_cancelling", pathStatus: entity.SpeculationPathStatusBuilding, wantStatus: entity.SpeculationPathStatusCancelling}, + {name: "pre_build_path_drops_to_cancelled", pathStatus: entity.SpeculationPathStatusCandidate, wantStatus: entity.SpeculationPathStatusCancelled}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + h := newTestHarness(t, ctrl) + dep := entity.Batch{ID: "test-queue/batch/0", Queue: "test-queue", State: entity.BatchStateFailed, Version: 1} + batch := testBatch(entity.BatchStateSpeculating, dep.ID) + path := entity.SpeculationPath{Base: []string{dep.ID}, Head: batch.ID} + tree := entity.SpeculationTree{ + BatchID: batch.ID, + Version: 2, + Paths: []entity.SpeculationPathInfo{{ID: "test-queue/batch/1/path/0", Path: path, Status: tt.pathStatus}}, + } - h.batchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil) - // speculateBatch fetches dependencies once for the pre-tree gate check, - // and tryFinalize fetches them again itself. - h.batchStore.EXPECT().Get(gomock.Any(), dep.ID).Return(dep, nil).Times(2) - h.treeStore.EXPECT().Get(gomock.Any(), batch.ID).Return(entity.SpeculationTree{BatchID: batch.ID, Version: 1, Paths: []entity.SpeculationPathInfo{}}, nil) - // No UpdateState expected — gomock will fail if it is called. + h.batchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil) + h.batchStore.EXPECT().Get(gomock.Any(), dep.ID).Return(dep, nil) + h.treeStore.EXPECT().Get(gomock.Any(), batch.ID).Return(tree, nil) + h.pathBuildStore.EXPECT().Get(gomock.Any(), "test-queue/batch/1/path/0").Return(entity.SpeculationPathBuild{}, storage.ErrNotFound) + // No runner interaction expected regardless of pathStatus: reconcile + // only ever records intent (D1/D2). + h.treeStore.EXPECT().Update(gomock.Any(), batch.ID, int32(2), int32(3), gomock.Any()). + DoAndReturn(func(_ context.Context, _ string, _, _ int32, paths []entity.SpeculationPathInfo) error { + require.Len(t, paths, 1) + assert.Equal(t, tt.wantStatus, paths[0].Status) + return nil + }) + h.batchStore.EXPECT().UpdateState(gomock.Any(), batch.ID, int32(1), int32(2), entity.BatchStateFailed).Return(nil) + h.depStore.EXPECT().Get(gomock.Any(), batch.ID).Return(entity.BatchDependent{ + BatchID: batch.ID, + Version: 1, + }, nil) - require.NoError(t, runProcess(t, ctrl, h.controller, batch.ID)) - assert.Equal(t, []pubRec{{topic: "prioritize", msgID: batch.Queue}}, *h.records) + require.NoError(t, runProcess(t, ctrl, h.controller, batch.ID)) + assert.ElementsMatch(t, []pubRec{ + {topic: "prioritize", msgID: batch.Queue}, + {topic: "conclude", msgID: batch.ID}, + }, *h.records) + }) + } } -// tryFinalize: a failed dep must fail the batch (Speculating → Failed), wake -// its dependents (so the failure cascades), and publish to conclude. -// Otherwise the batch livelocks. -func TestController_Process_FailedDepFailsBatch(t *testing.T) { +// The path's own build failing (independent of any dependency) fails the +// path and, with no other path viable, fails the batch. +func TestController_Process_OwnBuildFailedFailsBatch(t *testing.T) { ctrl := gomock.NewController(t) h := newTestHarness(t, ctrl) - 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"} + batch := testBatch(entity.BatchStateSpeculating) + path := entity.SpeculationPath{Head: batch.ID} + tree := entity.SpeculationTree{ + BatchID: batch.ID, + Version: 5, + Paths: []entity.SpeculationPathInfo{{ID: "test-queue/batch/1/path/0", Path: path, Status: entity.SpeculationPathStatusBuilding, BuildID: "build-1"}}, + } h.batchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil) - // speculateBatch fetches dependencies once for the pre-tree gate check, - // and tryFinalize fetches them again itself. - h.batchStore.EXPECT().Get(gomock.Any(), dep.ID).Return(dep, nil).Times(2) - h.treeStore.EXPECT().Get(gomock.Any(), batch.ID).Return(entity.SpeculationTree{BatchID: batch.ID, Version: 1, Paths: []entity.SpeculationPathInfo{}}, nil) + h.treeStore.EXPECT().Get(gomock.Any(), batch.ID).Return(tree, nil) + h.pathBuildStore.EXPECT().Get(gomock.Any(), "test-queue/batch/1/path/0").Return(entity.SpeculationPathBuild{PathID: "test-queue/batch/1/path/0", BuildID: "build-1"}, nil) + h.buildStore.EXPECT().Get(gomock.Any(), "build-1").Return(entity.Build{ + ID: "build-1", BatchID: batch.ID, SpeculationPathID: "test-queue/batch/1/path/0", Status: entity.BuildStatusFailed, + }, nil) + h.treeStore.EXPECT().Update(gomock.Any(), batch.ID, int32(5), int32(6), gomock.Any()). + DoAndReturn(func(_ context.Context, _ string, _, _ int32, paths []entity.SpeculationPathInfo) error { + require.Len(t, paths, 1) + assert.Equal(t, entity.SpeculationPathStatusFailed, paths[0].Status) + return nil + }) h.batchStore.EXPECT().UpdateState(gomock.Any(), batch.ID, int32(1), int32(2), entity.BatchStateFailed).Return(nil) + // Failing the batch must fan out to its dependents so their dead chain + // paths get reconciled and a surviving path can take over. h.depStore.EXPECT().Get(gomock.Any(), batch.ID).Return(entity.BatchDependent{ BatchID: batch.ID, - Dependents: []string{"test-queue/batch/2"}, + Dependents: []string{"test-queue/batch/9"}, Version: 1, }, nil) require.NoError(t, runProcess(t, ctrl, h.controller, batch.ID)) - assert.Equal(t, []pubRec{ + assert.ElementsMatch(t, []pubRec{ {topic: "prioritize", msgID: batch.Queue}, - {topic: "speculate", msgID: "test-queue/batch/2"}, + {topic: "speculate", msgID: "test-queue/batch/9"}, {topic: "conclude", msgID: batch.ID}, }, *h.records) } -// A dependent-wake publish failure after the terminal CAS nacks the message; -// redelivery converges through the terminal branch, which re-runs the -// fan-out and the conclude publish. Only the speculate topic fails here — +// A dependent-wake publish failure after finalize's terminal CAS nacks the +// message; redelivery converges through the terminal branch, which re-runs +// the fan-out and the conclude publish. Only the speculate topic fails here — // the prioritize publish earlier in the pass succeeds and is recorded. -func TestController_Process_FailedDepDependentPublishFailure(t *testing.T) { +func TestController_Process_FinalizeDependentPublishFailure(t *testing.T) { ctrl := gomock.NewController(t) h := newTopicFailingPublishHarness(t, ctrl, fmt.Errorf("publish failed"), "speculate") - dep := entity.Batch{ID: "test-queue/batch/0", Queue: "test-queue", State: entity.BatchStateFailed, Version: 1} - batch := testBatch(entity.BatchStateSpeculating, dep.ID) + batch := testBatch(entity.BatchStateSpeculating) + path := entity.SpeculationPath{Head: batch.ID} + tree := entity.SpeculationTree{ + BatchID: batch.ID, + Version: 5, + Paths: []entity.SpeculationPathInfo{{ID: "test-queue/batch/1/path/0", Path: path, Status: entity.SpeculationPathStatusBuilding, BuildID: "build-1"}}, + } h.batchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil) - // speculateBatch fetches dependencies once for the pre-tree gate check, - // and tryFinalize fetches them again itself. - h.batchStore.EXPECT().Get(gomock.Any(), dep.ID).Return(dep, nil).Times(2) - h.treeStore.EXPECT().Get(gomock.Any(), batch.ID).Return(entity.SpeculationTree{BatchID: batch.ID, Version: 1, Paths: []entity.SpeculationPathInfo{}}, nil) + h.treeStore.EXPECT().Get(gomock.Any(), batch.ID).Return(tree, nil) + h.pathBuildStore.EXPECT().Get(gomock.Any(), "test-queue/batch/1/path/0").Return(entity.SpeculationPathBuild{PathID: "test-queue/batch/1/path/0", BuildID: "build-1"}, nil) + h.buildStore.EXPECT().Get(gomock.Any(), "build-1").Return(entity.Build{ + ID: "build-1", BatchID: batch.ID, SpeculationPathID: "test-queue/batch/1/path/0", Status: entity.BuildStatusFailed, + }, nil) + h.treeStore.EXPECT().Update(gomock.Any(), batch.ID, int32(5), int32(6), gomock.Any()).Return(nil) h.batchStore.EXPECT().UpdateState(gomock.Any(), batch.ID, int32(1), int32(2), entity.BatchStateFailed).Return(nil) h.depStore.EXPECT().Get(gomock.Any(), batch.ID).Return(entity.BatchDependent{ BatchID: batch.ID, - Dependents: []string{"test-queue/batch/2"}, + Dependents: []string{"test-queue/batch/9"}, Version: 1, }, nil) @@ -621,34 +716,51 @@ func TestController_Process_FailedDepDependentPublishFailure(t *testing.T) { assert.Equal(t, []pubRec{{topic: "prioritize", msgID: batch.Queue}}, *h.records) } -// 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) { +// Hand-built two-path tree: a dependency of the head landing (Succeeded) +// while NOT in a path's base deads that path, even though the sibling path +// (which included the dependency in its base) survives. +func TestController_Process_NonBaseDependencyLandingDeadsSiblingPath(t *testing.T) { ctrl := gomock.NewController(t) h := newTestHarness(t, ctrl) - 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) + dep := entity.Batch{ID: "test-queue/batch/0", Queue: "test-queue", State: entity.BatchStateSucceeded, Version: 1} + batch := testBatch(entity.BatchStateSpeculating, dep.ID) + + pathAlone := entity.SpeculationPath{Head: batch.ID} // does not assume dep lands + pathWithBase := entity.SpeculationPath{Base: []string{dep.ID}, Head: batch.ID} // assumes dep lands + tree := entity.SpeculationTree{ + BatchID: batch.ID, + Version: 1, + Paths: []entity.SpeculationPathInfo{ + {ID: "test-queue/batch/1/path/0", Path: pathAlone, Status: entity.SpeculationPathStatusSelected}, + {ID: "test-queue/batch/1/path/1", Path: pathWithBase, Status: entity.SpeculationPathStatusSelected}, + }, + } h.batchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil) - // speculateBatch fetches dependencies once for the pre-tree gate check, - // and tryFinalize fetches them again itself. - h.batchStore.EXPECT().Get(gomock.Any(), depCancelled.ID).Return(depCancelled, nil).Times(2) - h.batchStore.EXPECT().Get(gomock.Any(), depSucceeded.ID).Return(depSucceeded, nil).Times(2) - h.treeStore.EXPECT().Get(gomock.Any(), batch.ID).Return(entity.SpeculationTree{BatchID: batch.ID, Version: 1, Paths: []entity.SpeculationPathInfo{}}, nil) - h.batchStore.EXPECT().UpdateState(gomock.Any(), batch.ID, int32(1), int32(2), entity.BatchStateMerging).Return(nil) + h.batchStore.EXPECT().Get(gomock.Any(), dep.ID).Return(dep, nil) + h.treeStore.EXPECT().Get(gomock.Any(), batch.ID).Return(tree, nil) + h.pathBuildStore.EXPECT().Get(gomock.Any(), "test-queue/batch/1/path/0").Return(entity.SpeculationPathBuild{}, storage.ErrNotFound) + h.pathBuildStore.EXPECT().Get(gomock.Any(), "test-queue/batch/1/path/1").Return(entity.SpeculationPathBuild{}, storage.ErrNotFound) + h.treeStore.EXPECT().Update(gomock.Any(), batch.ID, int32(1), int32(2), gomock.Any()). + DoAndReturn(func(_ context.Context, _ string, _, _ int32, paths []entity.SpeculationPathInfo) error { + require.Len(t, paths, 2) + byHeadOnly := func(p entity.SpeculationPathInfo) bool { return len(p.Path.Base) == 0 } + for _, p := range paths { + if byHeadOnly(p) { + assert.Equal(t, entity.SpeculationPathStatusCancelled, p.Status, "path that didn't base on the landed dep must be dead") + } else { + assert.Equal(t, entity.SpeculationPathStatusSelected, p.Status, "path that based on the landed dep must survive") + } + } + return nil + }) require.NoError(t, runProcess(t, ctrl, h.controller, batch.ID)) - assert.Equal(t, []pubRec{ - {topic: "prioritize", msgID: batch.Queue}, - {topic: "submitqueue-merge", msgID: batch.ID}, - }, *h.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 order with a running build and +// 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 @@ -770,7 +882,7 @@ func TestController_Process_CancellingTerminalCASVersionMismatch(t *testing.T) { // Publish failure must not advance the batch state further: the CAS to // Speculating (which precedes the prioritize publish) still lands, but -// tryFinalize must never run since the publish error aborts speculateBatch. +// finalize must never run since the publish error aborts speculateBatch. func TestController_Process_PublishFailure(t *testing.T) { ctrl := gomock.NewController(t) h := newFailingPublishHarness(t, ctrl, fmt.Errorf("publish failed")) @@ -782,3 +894,155 @@ func TestController_Process_PublishFailure(t *testing.T) { require.Error(t, runProcess(t, ctrl, h.controller, batch.ID)) } + +// A publish failure on an already-Speculating batch (no CAS in play) must +// still abort before finalize runs. +func TestController_Process_PublishFailureAlreadySpeculating(t *testing.T) { + ctrl := gomock.NewController(t) + h := newFailingPublishHarness(t, ctrl, fmt.Errorf("publish failed")) + batch := testBatch(entity.BatchStateSpeculating) + path := entity.SpeculationPath{Head: batch.ID} + tree := entity.SpeculationTree{ + BatchID: batch.ID, + Version: 1, + Paths: []entity.SpeculationPathInfo{{ID: "test-queue/batch/1/path/0", Path: path, Status: entity.SpeculationPathStatusSelected}}, + } + + h.batchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil) + h.treeStore.EXPECT().Get(gomock.Any(), batch.ID).Return(tree, nil) + h.pathBuildStore.EXPECT().Get(gomock.Any(), "test-queue/batch/1/path/0").Return(entity.SpeculationPathBuild{}, storage.ErrNotFound) + + require.Error(t, runProcess(t, ctrl, h.controller, batch.ID)) +} + +// reconcileStatus is a pure function: table-test every transition, with +// special attention to "never downgrade a terminal status". +func TestReconcileStatus(t *testing.T) { + tests := []struct { + name string + current entity.SpeculationPathStatus + build entity.BuildStatus + want entity.SpeculationPathStatus + }{ + {"accepted_maps_to_building", entity.SpeculationPathStatusPrioritized, entity.BuildStatusAccepted, entity.SpeculationPathStatusBuilding}, + {"running_maps_to_building", entity.SpeculationPathStatusBuilding, entity.BuildStatusRunning, entity.SpeculationPathStatusBuilding}, + {"succeeded_maps_to_passed", entity.SpeculationPathStatusBuilding, entity.BuildStatusSucceeded, entity.SpeculationPathStatusPassed}, + {"failed_maps_to_failed", entity.SpeculationPathStatusBuilding, entity.BuildStatusFailed, entity.SpeculationPathStatusFailed}, + {"cancelled_maps_to_cancelled", entity.SpeculationPathStatusBuilding, entity.BuildStatusCancelled, entity.SpeculationPathStatusCancelled}, + {"cancelling_stays_until_build_terminal", entity.SpeculationPathStatusCancelling, entity.BuildStatusRunning, entity.SpeculationPathStatusCancelling}, + {"cancelling_resolves_on_terminal_build", entity.SpeculationPathStatusCancelling, entity.BuildStatusCancelled, entity.SpeculationPathStatusCancelled}, + {"passed_never_downgraded", entity.SpeculationPathStatusPassed, entity.BuildStatusFailed, entity.SpeculationPathStatusPassed}, + {"failed_never_downgraded", entity.SpeculationPathStatusFailed, entity.BuildStatusSucceeded, entity.SpeculationPathStatusFailed}, + {"cancelled_never_downgraded", entity.SpeculationPathStatusCancelled, entity.BuildStatusSucceeded, entity.SpeculationPathStatusCancelled}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.Equal(t, tt.want, reconcileStatus(tt.current, tt.build)) + }) + } +} + +// pathDead is a pure function: table-test the base-failed and +// landed-non-base-dependency cases, plus the deliberate Cancelled-base +// leniency. +func TestPathDead(t *testing.T) { + head := "q/batch/2" + tests := []struct { + name string + path entity.SpeculationPath + deps map[string]entity.Batch + want bool + }{ + { + name: "base_dep_failed_is_dead", + path: entity.SpeculationPath{Base: []string{"q/batch/1"}, Head: head}, + deps: map[string]entity.Batch{"q/batch/1": {ID: "q/batch/1", State: entity.BatchStateFailed}}, + want: true, + }, + { + name: "base_dep_cancelled_is_tolerated", + path: entity.SpeculationPath{Base: []string{"q/batch/1"}, Head: head}, + deps: map[string]entity.Batch{"q/batch/1": {ID: "q/batch/1", State: entity.BatchStateCancelled}}, + want: false, + }, + { + name: "non_base_dep_landed_is_dead", + path: entity.SpeculationPath{Head: head}, + deps: map[string]entity.Batch{"q/batch/1": {ID: "q/batch/1", State: entity.BatchStateSucceeded}}, + want: true, + }, + { + name: "non_base_dep_pending_is_not_dead", + path: entity.SpeculationPath{Head: head}, + deps: map[string]entity.Batch{"q/batch/1": {ID: "q/batch/1", State: entity.BatchStateSpeculating}}, + want: false, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.Equal(t, tt.want, pathDead(tt.path, tt.deps)) + }) + } +} + +// mergeableNow and viable are pure functions over a path and its +// dependencies; table-test the combinations not already covered end-to-end. +func TestMergeableNowAndViable(t *testing.T) { + base := "q/batch/1" + head := "q/batch/2" + + tests := []struct { + name string + status entity.SpeculationPathStatus + deps map[string]entity.Batch + wantMergeable bool + wantViable bool + pathBaseIsBase bool + }{ + { + name: "passed_base_succeeded_no_extra_deps", + status: entity.SpeculationPathStatusPassed, + deps: map[string]entity.Batch{base: {ID: base, State: entity.BatchStateSucceeded}}, + wantMergeable: true, + wantViable: true, + pathBaseIsBase: true, + }, + { + name: "passed_base_pending", + status: entity.SpeculationPathStatusPassed, + deps: map[string]entity.Batch{base: {ID: base, State: entity.BatchStateSpeculating}}, + wantMergeable: false, + wantViable: true, + pathBaseIsBase: true, + }, + { + name: "failed_path_not_viable", + status: entity.SpeculationPathStatusFailed, + deps: map[string]entity.Batch{}, + wantMergeable: false, + wantViable: false, + pathBaseIsBase: false, + }, + { + name: "cancelling_path_not_viable", + status: entity.SpeculationPathStatusCancelling, + deps: map[string]entity.Batch{}, + wantMergeable: false, + wantViable: false, + pathBaseIsBase: false, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + var path entity.SpeculationPath + if tt.pathBaseIsBase { + path = entity.SpeculationPath{Base: []string{base}, Head: head} + } else { + path = entity.SpeculationPath{Head: head} + } + info := entity.SpeculationPathInfo{Path: path, Status: tt.status} + assert.Equal(t, tt.wantMergeable, mergeableNow(info, tt.deps)) + assert.Equal(t, tt.wantViable, viable(info, tt.deps)) + }) + } +}