diff --git a/doc/rfc/submitqueue/speculation-generator-best-first.md b/doc/rfc/submitqueue/speculation-generator-best-first.md index cfe87f37e..458e5e007 100644 --- a/doc/rfc/submitqueue/speculation-generator-best-first.md +++ b/doc/rfc/submitqueue/speculation-generator-best-first.md @@ -447,7 +447,7 @@ The ordering stays the same. `CandidatePath.RankingScore` contains this logarith - `Succeeded` fixes an assumption to succeeds. - `Failed` or `Cancelled` fixes an assumption to fails. - `Cancelling` remains undecided because cancellation may lose a race with completion. -- `Merging` also remains undecided, because a merge can fail. It is tempting to treat it as committed to landing and skip the scorer call, but that puts a state-specific policy inside the search: whether a path betting against a merging batch is worth funding is a question of price, and price belongs to the scorer. The allocator draws the same line — "no batch state enters this decision" — and the generator holds it too. Nothing is lost by staying open, because a head can never merge ahead of a dependency it took a position on (see [speculation.md](speculation.md)); the cost of an unlikely path is budget, which is the allocator's to ration. +- `Merging` also remains undecided, because a merge can fail. It is tempting to treat it as committed to landing and skip the scorer call, but that puts a state-specific policy inside the search: whether a path betting against a merging batch is worth funding is a question of price, and price belongs to the scorer. The allocator draws the same line — "no batch state enters this decision" — and the generator holds it too. Nothing is lost by staying open: a single passed path still waits for the merge result, while passed paths covering every outcome let the controller bypass the dependency (see [speculation.md](speculation.md)). Funding the unlikely side spends budget, which is the allocator's to ration. - A fixed assumption stays in the returned path but contributes probability 1 and has no flip. - A shared dependency is scored once per run. diff --git a/doc/rfc/submitqueue/speculation.md b/doc/rfc/submitqueue/speculation.md index ebdd49896..1887cff7e 100644 --- a/doc/rfc/submitqueue/speculation.md +++ b/doc/rfc/submitqueue/speculation.md @@ -4,7 +4,7 @@ A merge queue that verifies one change at a time is limited by its slowest build Work enters SubmitQueue as **batches** — changes verified and merged together. Two batches **conflict** when they touch the same code, which makes the earlier one a **dependency** of the later. A **path** is one set of assumptions about how a batch's dependencies resolve, and the batch it builds is the path's **head**. -On every queue update the **speculate controller** reruns from scratch: it reads the current state, applies the incoming signals, asks a pluggable **Speculator** which paths are worth building within the CI budget, and persists only those. Everything else is recomputed next time, never stored. Merging stays strict: a batch merges only after its dependencies resolve and a matching build has passed. +On every queue update the **speculate controller** reruns from scratch: it reads the current state, applies the incoming signals, asks a pluggable **Speculator** which paths are worth building within the CI budget, and persists only those. Everything else is recomputed next time, never stored. A batch normally merges after its dependencies resolve and a matching build has passed; complete passed coverage of every unresolved outcome lets it bypass those dependencies. ## The speculation run @@ -54,7 +54,7 @@ Every write is a compare-and-swap: a writer that loses re-reads on a later run. Verdicts are controller-owned facts: the Speculator can neither compute nor veto them. -- **Merge (strict).** Each path carries an assumption about every dependency — *succeeds* (built on top of) or *fails* (built without). Once a path's build has passed and every dependency has finished the way the path assumed — one assumed *succeeds* has merged, one assumed *fails* has failed or been cancelled — the speculate controller moves the head to Merging and hands it to Runway. A dependency that is merely *merging* has not finished, because a merge can fail, so it is still waited on. If that hand-off is lost, the next run re-sends it. The same run sets the head's remaining in-flight paths *cancelling*: once one path has passed the others cannot help, and they hold CI slots until they stop. The mergesignal controller records Runway's terminal result: success marks the head Succeeded, while failure marks it Failed. The result publishes a single dirty signal — no per-dependent fan-out — and the next run refutes paths whose assumption disagrees with the result: *fails* assumptions after success, *succeeds* assumptions after failure. The hand-off is idempotent, so Runway reports success without another merge when the change is already present. Down a chain, each head waits for its predecessors to settle, so a chain merges one at a time. +- **Merge.** Each path carries an assumption about every dependency — *succeeds* (built on top of) or *fails* (built without). Normally, once a path's build has passed and every dependency has finished the way the path assumed — one assumed *succeeds* has merged, one assumed *fails* has failed or been cancelled — the speculate controller moves the head to Merging and hands it to Runway. A dependency that is merely *merging* has not finished, because a merge can fail, so a single matching path still waits for the answer. Complete passed coverage is the exception described in Bypass large diff: it lets a head merge before those answers arrive. If the hand-off is lost, the next run re-sends it. The same run sets the head's remaining in-flight paths *cancelling*: once the head can merge they cannot help, and they hold CI slots until they stop. The mergesignal controller records Runway's terminal result: success marks the head Succeeded, while failure marks it Failed. The result publishes a single dirty signal — no per-dependent fan-out — and the next run refutes paths whose assumption disagrees with the result: *fails* assumptions after success, *succeeds* assumptions after failure. The hand-off is idempotent, so Runway reports success without another merge when the change is already present. A chain ordinarily merges one at a time, but a fully covered head can bypass its unsettled predecessors. - **Failure (no viable path).** A batch fails when every possible future has a failed build — no path can pass, so it can never merge. - **Cancel.** A cancelled batch is driven terminal: its in-flight paths are set *cancelling*, then the batch is marked Cancelled once they stop (see Cancellation). @@ -74,9 +74,9 @@ Example of the payoff either way: `H` conflicts with `B1` and weak `B2`. Relax ` If a batch's passed builds cover *every* way its dependencies could resolve, the outcome is the same either way — so it can merge now, ahead of them. Classic case: a small change stuck behind a slow one is built both with and without it; both pass, and it merges immediately. -The default Speculator covers the whole space only when doing so is cheap enough, and funds the extra candidates within the build budget. The controller merges early only when a passed path exists for every combination of the dependencies — it reads that straight off the path records. If any combination is missing or unbuilt, the head waits normally. +The controller checks coverage over only the dependencies that have not settled yet. Settled dependencies pin each surviving path to the outcome that actually happened; for every combination of the remaining dependencies, the path set must contain a passed, unbroken path with that combination of assumptions. If any combination is missing, unbuilt, failed, or contradicted by a settled dependency, the head waits normally. The check only observes paths the Speculator already funded — it does not fund the exponential path space itself or alter the queue's build budget. -**Not yet implemented on the controller side.** `decide`/`mergeablePath` gate on a single passed path whose assumptions have all been settled by the dependency's actual state; nothing enumerates the combinations. The distinction matters: a *single* passed path that assumed a dependency would fail is not complete coverage, and merging on it while that dependency is still live would put a combination on the trunk that no build validated. Coverage is what makes early merge sound — one path betting the right way is not. +Coverage makes the bypass sound because whichever way the dependencies later resolve, a passed build already validated the resulting set of changes. The build order and merge order differ: a path assuming dependency `D` succeeds validates `D` then head `H`, while bypass lands `H` before `D`. SubmitQueue treats those orders as content-equivalent. Runway still performs the real merge, so if the reordered changes conflict textually, the older dependency can fail after the newer head has bypassed it; this is an accepted cost of landing the fully covered head early rather than a licence to put unmergeable content on the target. ### Cancellation diff --git a/submitqueue/entity/request_log.go b/submitqueue/entity/request_log.go index eac65c040..72bad9ef6 100644 --- a/submitqueue/entity/request_log.go +++ b/submitqueue/entity/request_log.go @@ -58,11 +58,11 @@ const ( RequestStatusBatched RequestStatus = "batched" // RequestStatusSpeculating indicates that the batch containing the request is in speculation: - // planning, building, or waiting for its dependencies to settle. None of those leaves it able to land. + // planning, building, or waiting until either its dependencies settle or passed paths cover every possible outcome. RequestStatusSpeculating RequestStatus = "speculating" // RequestStatusSpeculated indicates that the batch containing the request has finished speculating: - // a build passed on a path whose assumptions all held, and the batch has been cleared to merge. + // either a passed path's assumptions all held, or passed paths cover every outcome of its unsettled dependencies. RequestStatusSpeculated RequestStatus = "speculated" // RequestStatusLanding indicates that the request is actively being landed (e.g., source control operation is in progress to push the change to the target branch). diff --git a/submitqueue/extension/speculation/speculator/standard/README.md b/submitqueue/extension/speculation/speculator/standard/README.md index 02e9f5f7e..c5d869e65 100644 --- a/submitqueue/extension/speculation/speculator/standard/README.md +++ b/submitqueue/extension/speculation/speculator/standard/README.md @@ -4,7 +4,7 @@ The `standard` `Speculator` funds the queue's most promising speculation paths f Each run it considers candidate paths in descending order of their probability of being the future that actually happens, and proposes builds down that ranking. Paths already pending or building keep the slot they hold rather than restarting; paths whose builds already finished are skipped for as long as their records remain in the supplied path sets, so a finished path can be proposed again — for a retry, say — once retention drops it; new builds fill whatever budget remains. -When the budget runs out, everything below the cut waits for a later run. That is safe because a batch's verdict never depends on what was funded — only on how its dependencies resolve and which builds pass. +When the budget runs out, everything below the cut waits for a later run. That is safe because the propose-side cannot invent a batch verdict: the speculate controller still decides merge from the persisted paths, including complete coverage of unsettled dependencies. Both halves are swappable. The ranking is the `Generator`'s: the default `bestfirst` scores each path by the probability that all its assumptions hold. The budget policy is the `Allocator`'s: the default `sticky` fills only free slots and never preempts, where a preempting allocator would cancel a low-value in-flight path to fund a better one. diff --git a/submitqueue/orchestrator/controller/speculate/BUILD.bazel b/submitqueue/orchestrator/controller/speculate/BUILD.bazel index 01b4113af..c0fd13583 100644 --- a/submitqueue/orchestrator/controller/speculate/BUILD.bazel +++ b/submitqueue/orchestrator/controller/speculate/BUILD.bazel @@ -3,6 +3,7 @@ load("@rules_go//go:def.bzl", "go_library", "go_test") go_library( name = "go_default_library", srcs = [ + "bypass.go", "check.go", "dispatch.go", "doc.go", diff --git a/submitqueue/orchestrator/controller/speculate/bypass.go b/submitqueue/orchestrator/controller/speculate/bypass.go new file mode 100644 index 000000000..9177a2d46 --- /dev/null +++ b/submitqueue/orchestrator/controller/speculate/bypass.go @@ -0,0 +1,79 @@ +// 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" + +// bypassablePath returns a passed path once passed builds cover every possible +// outcome of the head's unsettled dependencies. Settled dependencies stay +// pinned to reality through assumptionBroken. +// +// A path's combination is read positionally — the i-th assumption belongs to +// the head's i-th dependency — which isWellFormed's order check is what +// licenses: two paths with the same assumptions in different orders are +// different stacks, not the same outcome. +func bypassablePath(head entity.Batch, set entity.SpeculationPathSet, snap snapshot) (entity.SpeculationPathEntry, bool) { + unsettled := unsettledDependencyIndices(head, snap) + if len(unsettled) == 0 { + return entity.SpeculationPathEntry{}, false + } + + required := 1 + for range unsettled { + if required > len(set.Paths)/2 { + return entity.SpeculationPathEntry{}, false + } + required *= 2 + } + + seen := make(map[string]struct{}, required) + var winner entity.SpeculationPathEntry + for _, entry := range set.Paths { + if entry.Status != entity.SpeculationPathStatusPassed || + assumptionBroken(entry.Path, snap) || + !isWellFormed(entry.Path, head) { + continue + } + + signature := make([]byte, len(unsettled)) + for i, depIndex := range unsettled { + if entry.Path.Dependencies[depIndex].Assumption == entity.DependencyAssumptionFails { + signature[i] = 'f' + } else { + signature[i] = 's' + } + } + key := string(signature) + if _, exists := seen[key]; exists { + continue + } + seen[key] = struct{}{} + if len(seen) == 1 { + winner = entry + } + } + + return winner, len(seen) == required +} + +func unsettledDependencyIndices(head entity.Batch, snap snapshot) []int { + var indices []int + for i, depID := range head.Dependencies { + if !snap.batchState(depID).IsTerminal() { + indices = append(indices, i) + } + } + return indices +} diff --git a/submitqueue/orchestrator/controller/speculate/check.go b/submitqueue/orchestrator/controller/speculate/check.go index f82cd8a21..14528c913 100644 --- a/submitqueue/orchestrator/controller/speculate/check.go +++ b/submitqueue/orchestrator/controller/speculate/check.go @@ -29,7 +29,7 @@ const ( rejectHeadNotSpeculating rejection = "head_not_speculating" // rejectMalformedPath is a path whose assumptions do not line up with its // head's dependency list: one missing or extra, a duplicate, a wrong head, - // or a made-up assumption value. + // an assumption out of queue order, or a made-up assumption value. rejectMalformedPath rejection = "malformed_path" // rejectBrokenAssumption is a path with an assumption a finished // dependency has already proven wrong. @@ -122,7 +122,13 @@ func rejectionReason(proposal entity.Speculation, snap snapshot) (rejection, boo // isWellFormed reports whether a path is a proper guess about its head: // exactly one assumption for each of the head's dependencies, no more and no -// fewer, and every assumption a real value. +// fewer, in the head's dependency order, and every assumption a real value. +// +// Order is load-bearing: Base() projects the path positionally, so a path +// whose dependencies are permuted describes a stack the build runner applied +// in a different order — a different tree, which no verdict may count as the +// combination its assumptions name. With the length check and position-wise +// equality, a missing, extra, or duplicate dependency is also impossible. // // A malformed path is not merely suboptimal, it is unmergeable — the merge // preconditions are read off the path's assumptions (see mergeablePath), so a @@ -135,16 +141,10 @@ func isWellFormed(path entity.SpeculationPath, head entity.Batch) bool { return false } - required := make(map[string]struct{}, len(head.Dependencies)) - for _, dep := range head.Dependencies { - required[dep] = struct{}{} - } - - for _, dep := range path.Dependencies { - if _, ok := required[dep.Batch]; !ok { + for i, dep := range path.Dependencies { + if dep.Batch != head.Dependencies[i] { return false } - delete(required, dep.Batch) switch dep.Assumption { case entity.DependencyAssumptionSucceeds, @@ -154,7 +154,7 @@ func isWellFormed(path entity.SpeculationPath, head entity.Batch) bool { } } - return len(required) == 0 + return true } // findPath returns the entry for a path ID in the set. diff --git a/submitqueue/orchestrator/controller/speculate/check_test.go b/submitqueue/orchestrator/controller/speculate/check_test.go index b095f9274..ee2e25f5f 100644 --- a/submitqueue/orchestrator/controller/speculate/check_test.go +++ b/submitqueue/orchestrator/controller/speculate/check_test.go @@ -102,6 +102,18 @@ func TestFilterProposals_Rejects(t *testing.T) { snap: checkSnapshot(entity.BatchStateSpeculating), want: rejectMalformedPath, }, + { + name: "path with dependencies out of queue order", + proposal: entity.Speculation{ + Path: entity.SpeculationPath{Head: head, Dependencies: []entity.PathDependency{ + {Batch: dep2, Assumption: entity.DependencyAssumptionFails}, + {Batch: dep1, Assumption: entity.DependencyAssumptionSucceeds}, + }}, + Action: entity.PathActionBuild, + }, + snap: checkSnapshot(entity.BatchStateSpeculating), + want: rejectMalformedPath, + }, { name: "cancel on a path that is not stored", proposal: entity.Speculation{Path: valid, Action: entity.PathActionCancel}, @@ -208,12 +220,12 @@ func TestIsWellFormed(t *testing.T) { want: true, }, { - name: "order does not matter", + name: "dependencies out of queue order", path: entity.SpeculationPath{Head: head, Dependencies: []entity.PathDependency{ {Batch: dep2, Assumption: entity.DependencyAssumptionSucceeds}, {Batch: dep1, Assumption: entity.DependencyAssumptionFails}, }}, - want: true, + want: false, }, { name: "missing a dependency", diff --git a/submitqueue/orchestrator/controller/speculate/doc.go b/submitqueue/orchestrator/controller/speculate/doc.go index 1a5e82263..a8927cdbb 100644 --- a/submitqueue/orchestrator/controller/speculate/doc.go +++ b/submitqueue/orchestrator/controller/speculate/doc.go @@ -24,7 +24,8 @@ // 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 batch merges the moment the guessed-on dependencies -// land — it never waits for a build of its own to start afterwards. +// land. If passed paths cover every possible outcome, the batch can merge +// before those dependencies settle. // // # Paths // @@ -46,6 +47,8 @@ // // Fund both and every future is covered: // +// - While A is still unresolved, both P1 and P2 passing lets B bypass A and +// merge immediately: either possible future has already been validated. // - 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. @@ -75,9 +78,8 @@ // 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. 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. +// its build is cancelled to free the slot. A path is superseded when its head +// becomes mergeable, so any still-running siblings 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 (or proof nothing was ever @@ -89,8 +91,8 @@ // // # The life of a batch, as seen from here // -// Created ──admit──► Speculating ──┬── merge ──► Merging (merge stage takes over) -// └── fail ───► Failed +// Created ──admit──► Speculating ──┬── merge or bypass ──► Merging +// └── fail ─────────────► Failed // user cancel (cancel stage): // ... ──► Cancelling ── every path stopped ──► Cancelled // diff --git a/submitqueue/orchestrator/controller/speculate/finalize.go b/submitqueue/orchestrator/controller/speculate/finalize.go index 8096e0863..fb61e91ac 100644 --- a/submitqueue/orchestrator/controller/speculate/finalize.go +++ b/submitqueue/orchestrator/controller/speculate/finalize.go @@ -33,9 +33,9 @@ import ( // 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 +// out is dead, a head whose passed builds establish a merge verdict 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. // @@ -98,7 +98,10 @@ func (c *Controller) finalize(ctx context.Context, snap *snapshot) error { // 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) + winner, ok := mergeablePath(set, *snap) + if !ok { + winner, _ = bypassablePath(batch, set, *snap) + } if supersede(&set, winner.ID, nowMs) { snap.pathSets[batch.ID] = set snap.markDirty(batch.ID) @@ -449,9 +452,9 @@ func cancelBrokenPathsInSet(set *entity.SpeculationPathSet, snap snapshot, nowMs }) } -// 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. +// supersede stops every path other than the winner once the head can merge. +// Its live siblings cannot help any more but still hold 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 diff --git a/submitqueue/orchestrator/controller/speculate/outcome.go b/submitqueue/orchestrator/controller/speculate/outcome.go index 44f60bbb2..cd6c0c2c7 100644 --- a/submitqueue/orchestrator/controller/speculate/outcome.go +++ b/submitqueue/orchestrator/controller/speculate/outcome.go @@ -25,8 +25,9 @@ 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 means the head can be handed to the merge stage: either a + // passed path's assumptions have all come true, or passed paths cover every + // possible outcome of its unsettled dependencies. outcomeMerge outcome = "merge" // outcomeFail means no future remains in which the head could pass. outcomeFail outcome = "fail" @@ -54,6 +55,9 @@ func decide(head entity.Batch, set entity.SpeculationPathSet, snap snapshot) out if _, ok := mergeablePath(set, snap); ok { return outcomeMerge } + if _, ok := bypassablePath(head, set, snap); ok { + return outcomeMerge + } if hasNoViableFuture(head, set, snap) { return outcomeFail } diff --git a/submitqueue/orchestrator/controller/speculate/outcome_test.go b/submitqueue/orchestrator/controller/speculate/outcome_test.go index c21e6e43d..f1b6fd591 100644 --- a/submitqueue/orchestrator/controller/speculate/outcome_test.go +++ b/submitqueue/orchestrator/controller/speculate/outcome_test.go @@ -154,6 +154,137 @@ func TestMergeablePath_ExcludesBrokenPassedPath(t *testing.T) { assert.False(t, ok) } +func TestBypassablePath(t *testing.T) { + const ( + succeeds = entity.DependencyAssumptionSucceeds + fails = entity.DependencyAssumptionFails + ) + allPaths := []entity.SpeculationPathEntry{ + passedPath(succeeds, succeeds), + passedPath(succeeds, fails), + passedPath(fails, succeeds), + passedPath(fails, fails), + } + headBatch := entity.Batch{ID: head, Dependencies: []string{dep1, dep2}} + + tests := []struct { + name string + head entity.Batch + set entity.SpeculationPathSet + dep1State entity.BatchState + dep2State entity.BatchState + want bool + }{ + { + name: "bypasses when every outcome of two unsettled dependencies passed", + head: headBatch, + set: setOf(allPaths...), + dep1State: entity.BatchStateSpeculating, + dep2State: entity.BatchStateMerging, + want: true, + }, + { + name: "waits when one outcome is missing", + head: headBatch, + set: setOf(allPaths[:3]...), + dep1State: entity.BatchStateSpeculating, + dep2State: entity.BatchStateSpeculating, + }, + { + name: "covers only the unsettled dependency when another has succeeded", + head: headBatch, + set: setOf( + passedPath(succeeds, succeeds), + passedPath(succeeds, fails), + ), + dep1State: entity.BatchStateSucceeded, + dep2State: entity.BatchStateSpeculating, + want: true, + }, + { + name: "does not count a path contradicted by a settled dependency", + head: headBatch, + set: setOf( + passedPath(succeeds, succeeds), + passedPath(fails, fails), + ), + dep1State: entity.BatchStateSucceeded, + dep2State: entity.BatchStateSpeculating, + }, + { + name: "does not count a duplicate outcome twice", + head: headBatch, + set: setOf( + passedPath(succeeds, succeeds), + passedPath(succeeds, succeeds), + passedPath(fails, succeeds), + passedPath(fails, fails), + ), + dep1State: entity.BatchStateSpeculating, + dep2State: entity.BatchStateSpeculating, + }, + { + name: "does not count an unpassed outcome", + head: headBatch, + set: setOf( + allPaths[0], + allPaths[1], + allPaths[2], + entryFor(pathOver(fails, fails), entity.SpeculationPathStatusFailed), + ), + dep1State: entity.BatchStateSpeculating, + dep2State: entity.BatchStateSpeculating, + }, + { + name: "does not count a malformed path", + head: headBatch, + set: setOf( + allPaths[0], + allPaths[1], + allPaths[2], + passedPath(fails), + ), + dep1State: entity.BatchStateSpeculating, + dep2State: entity.BatchStateSpeculating, + }, + { + // A path stacked [dep2, dep1] built a different tree than [dep1, dep2]: + // the base feeds the runner in path order (see build.loadBase). It + // cannot stand in for the canonical combination its assumptions name. + name: "does not count a reordered path", + head: headBatch, + set: setOf( + passedPath(succeeds, succeeds), + passedPath(succeeds, fails), + passedPath(fails, succeeds), + entryFor(entity.SpeculationPath{Head: head, Dependencies: []entity.PathDependency{ + {Batch: dep2, Assumption: fails}, + {Batch: dep1, Assumption: fails}, + }}, entity.SpeculationPathStatusPassed), + ), + dep1State: entity.BatchStateSpeculating, + dep2State: entity.BatchStateSpeculating, + }, + { + name: "leaves fully settled dependencies to strict merge", + head: headBatch, + set: setOf(allPaths...), + dep1State: entity.BatchStateSucceeded, + dep2State: entity.BatchStateFailed, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + winner, ok := bypassablePath(tt.head, tt.set, snapWith(tt.dep1State, tt.dep2State)) + assert.Equal(t, tt.want, ok) + if tt.want { + assert.NotEmpty(t, winner.ID) + } + }) + } +} + // livePassedPath is mergeablePath without the settled requirement, and the gap // between the two is the head's waiting room: its own work is done and all that // remains is other batches finishing. That window is reported to the members, @@ -279,6 +410,15 @@ func TestDecide(t *testing.T) { // A passed path wins over a failed sibling: one way through is enough. assert.Equal(t, outcomeMerge, decide(headBatch, setOf(failed, passed), allResolved)) + + allUnresolved := snapWith(entity.BatchStateSpeculating, entity.BatchStateSpeculating) + fullCoverage := setOf( + passedPath(entity.DependencyAssumptionSucceeds, entity.DependencyAssumptionSucceeds), + passedPath(entity.DependencyAssumptionSucceeds, entity.DependencyAssumptionFails), + passedPath(entity.DependencyAssumptionFails, entity.DependencyAssumptionSucceeds), + passedPath(entity.DependencyAssumptionFails, entity.DependencyAssumptionFails), + ) + assert.Equal(t, outcomeMerge, decide(headBatch, fullCoverage, allUnresolved)) } // Once a path has passed, its siblings cannot help the head but are still diff --git a/submitqueue/orchestrator/controller/speculate/run_test.go b/submitqueue/orchestrator/controller/speculate/run_test.go index 306373640..0ae432fbe 100644 --- a/submitqueue/orchestrator/controller/speculate/run_test.go +++ b/submitqueue/orchestrator/controller/speculate/run_test.go @@ -1549,6 +1549,37 @@ func TestRun_MergingHeadReportsSpeculatedAndNoWait(t *testing.T) { assert.Equal(t, head, h.logs[0].Metadata["batch_id"]) } +func TestRun_BypassesUnsettledDependenciesWithFullCoverage(t *testing.T) { + ctrl := gomock.NewController(t) + const ( + succeeds = entity.DependencyAssumptionSucceeds + fails = entity.DependencyAssumptionFails + ) + + h := newRunHarness(t, ctrl, &scriptedSpeculator{}, []entity.Batch{memberHead()}) + 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.BatchStateMerging}, nil) + h.pathSets.EXPECT().Get(gomock.Any(), head).Return(entity.SpeculationPathSet{ + Head: head, + Paths: []entity.SpeculationPathEntry{ + passedPath(succeeds, succeeds), + passedPath(succeeds, fails), + passedPath(fails, succeeds), + passedPath(fails, fails), + }, + Version: 1, + }, nil).AnyTimes() + 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.Equal(t, []string{"submitqueue-merge"}, h.published) + assert.Zero(t, h.spec.calls) + require.Len(t, h.logs, 1) + assert.Equal(t, entity.RequestStatusSpeculated, h.logs[0].Status) +} + // The merge stage publishes landing as its first act on the dispatch. Both // statuses are non-terminal, so the summary is decided on timestamp alone and // a speculated sent afterwards would beat the landing it precedes. diff --git a/test/e2e/submitqueue/BUILD.bazel b/test/e2e/submitqueue/BUILD.bazel index 2733aa86a..680b0f6d3 100644 --- a/test/e2e/submitqueue/BUILD.bazel +++ b/test/e2e/submitqueue/BUILD.bazel @@ -48,6 +48,7 @@ go_test( "//submitqueue/core/batch:go_default_library", "//submitqueue/core/topickey:go_default_library", "//submitqueue/entity:go_default_library", + "//submitqueue/extension/storage:go_default_library", "//submitqueue/extension/storage/mysql:go_default_library", "//test/testutil:go_default_library", "@com_github_stretchr_testify//assert:go_default_library", diff --git a/test/e2e/submitqueue/harness_test.go b/test/e2e/submitqueue/harness_test.go index cb149ae5a..4ee147f49 100644 --- a/test/e2e/submitqueue/harness_test.go +++ b/test/e2e/submitqueue/harness_test.go @@ -42,6 +42,7 @@ import ( 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" "go.uber.org/zap" ) @@ -321,6 +322,39 @@ func (s *E2EIntegrationSuite) awaitBatchState(queue, batchID string, want entity }) } +// seedPassedPath writes one already-passed speculation path directly, so a +// test can hold the other side of a dependency back deterministically. +func (s *E2EIntegrationSuite) seedPassedPath(queue string, path entity.SpeculationPath) { + t := s.T() + store, err := s.appStorage.For(queue) + require.NoError(t, err, "failed to resolve operating store for queue %s", queue) + pathSets := store.GetSpeculationPathSetStore() + + set, err := pathSets.Get(s.ctx, path.Head) + if storage.IsNotFound(err) { + set = entity.SpeculationPathSet{Queue: queue, Head: path.Head} + } else { + require.NoError(t, err, "failed to read path set for %s", path.Head) + } + + set.Paths = append(set.Paths, entity.SpeculationPathEntry{ + ID: path.ID(), + Path: path, + Status: entity.SpeculationPathStatusPassed, + // Attempt 1 is the first build of a path and matches every entry the + // speculate run itself writes. + Attempt: 1, + Version: 1, + }) + newVersion := set.Version + 1 + if set.Version == 0 { + require.NoError(t, pathSets.Create(s.ctx, set), "failed to create path set for %s", path.Head) + } else { + require.NoError(t, pathSets.Update(s.ctx, set, set.Version, newVersion), + "failed to seed passed path %s for %s", path.ID(), path.Head) + } +} + // strandInCreated puts a batch back into Created, reproducing the state a batch // is left in when it is promoted but its announcement never reaches speculate. // Nothing will name it on the speculate topic again, so only a run that looks diff --git a/test/e2e/submitqueue/suite_test.go b/test/e2e/submitqueue/suite_test.go index 7097647d6..16a32be7f 100644 --- a/test/e2e/submitqueue/suite_test.go +++ b/test/e2e/submitqueue/suite_test.go @@ -276,34 +276,10 @@ func (s *E2EIntegrationSuite) TestLand_HappyPath_ReachesLanded() { "operating store should show request %s in terminal state landed", req.sqid) } -// TestDependentBatch_IsWokenByTheMergeAhead proves that a batch waiting on -// another is woken when that one merges — the edge CODEM-303 was silently -// dropping. -// -// A merged batch fans out to speculate so its dependents can re-plan. That -// message used to reuse the bare batch ID, which the batch controller had -// already published to the same topic and partition when the batch was -// created. The queue deduplicates against rows it has not collected yet, -// consumed ones included, so the wake-up was reported as a success, stored -// nothing, and never arrived. -// -// Ordinarily something else re-plans the queue soon enough to hide that. This -// test removes every other source of a wake-up, as stop → observe → start: -// -// 1. Stop: close the gate for runway-merge on this queue, before landing, so -// the lead batch cannot complete its merge. -// 2. Land the lead. It runs to the merge hand-off and parks there. -// 3. Land the dependent. The queue's analyzer serializes conservatively, so -// its batch depends on the lead's, which is in-flight (Merging counts). -// 4. Observe: wait for the dependent to record "waiting" — its speculative -// build has already passed, so its own build signals are finished. From -// here the only thing that can advance it is the lead merging. -// 5. Start: open the gate. The lead merges and fans out. -// -// The dependent reaching "landed" is therefore attributable to the fan-out -// alone. Against the old code it rests at "speculating" and the suite runs to -// Bazel's timeout, which is how the harness reports a pipeline that stalled. -func (s *E2EIntegrationSuite) TestDependentBatch_IsWokenByTheMergeAhead() { +// TestDependentBatch_BypassesMergingDependency proves that a dependency still +// waiting on Runway is unresolved for strict merge but can be bypassed once +// passed paths cover both of its possible outcomes. +func (s *E2EIntegrationSuite) TestDependentBatch_BypassesMergingDependency() { t := s.T() const queue = "e2e-chain-queue" @@ -337,12 +313,11 @@ func (s *E2EIntegrationSuite) TestDependentBatch_IsWokenByTheMergeAhead() { require.Contains(t, got.Dependencies, leadBatch, "batch %s must depend on the in-flight %s for this test to exercise anything", dependentBatch, leadBatch) - // Its speculative build passes while the lead is still parked, so by the - // time the gate opens the dependent has no build signals left to wake it. - // That rest is an event, not a status: a batch blocked on a dependency has - // not finished speculating, so it stays "speculating" until it can merge. - s.awaitEvent(dependent, entity.RequestEventWaiting) - s.log.Logf("Dependent %s has passed its build and waits only on %s", dependent.sqid, leadBatch) + // Both paths pass while the lead is still parked. A Merging dependency is + // unresolved because its merge can fail, so the durable Merging state proves + // the dependent advanced through complete coverage rather than strict merge. + s.awaitBatchState(queue, dependentBatch, entity.BatchStateMerging) + s.log.Logf("Dependent %s bypassed merging batch %s", dependent.sqid, leadBatch) // Start: the lead merges, and its fan-out is now the only thing that can // move the dependent. @@ -352,8 +327,103 @@ func (s *E2EIntegrationSuite) TestDependentBatch_IsWokenByTheMergeAhead() { s.awaitStatus(lead, entity.RequestStatusLanded) s.awaitStatus(dependent, entity.RequestStatusLanded) - assert.Equal(t, entity.RequestStateLanded, s.terminalState(dependent), - "the dependent must land once the batch it waited on merged") + assert.Equal(t, entity.RequestStateLanded, s.terminalState(dependent)) +} + +// TestDependentBatch_BypassedHeadLandsFirst proves the bypass does not just +// dispatch a merge: the dependent lands while its dependency is still held +// mid-build, and only then does the dependency proceed. +func (s *E2EIntegrationSuite) TestDependentBatch_BypassedHeadLandsFirst() { + t := s.T() + + const queue = "e2e-chain-queue" + const gateGroup = "orchestrator" + + lead := s.land(queue, "github://github.example.com/uber/e2e-bypass/pull/1/aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa") + heldBatch := s.awaitBatchID(lead) + s.closeGate(gateGroup, heldBatch, "e2e: hold the leader's build so the follower can fully cover it") + defer s.openGate(gateGroup, heldBatch) + s.awaitBatchState(queue, heldBatch, entity.BatchStateSpeculating) + + follower := s.land(queue, "github://github.example.com/uber/e2e-bypass/pull/2/bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb") + followerBatch := s.awaitBatchID(follower) + require.NotEqual(t, heldBatch, followerBatch) + + // The follower builds with and without the held leader while the leader's + // own build is parked; complete coverage hands it to the merge stage. + s.awaitBatchState(queue, followerBatch, entity.BatchStateMerging) + s.awaitStatus(follower, entity.RequestStatusLanded) + assert.Equal(t, entity.RequestStatusSpeculating, s.mustStatus(lead), + "the follower must land while its dependency is still held") + + s.openGate(gateGroup, heldBatch) + s.awaitStatus(lead, entity.RequestStatusLanded) + + s.assertStatusesInOrder(follower, + entity.RequestStatusSpeculating, + entity.RequestStatusSpeculated, + entity.RequestStatusLanding, + entity.RequestStatusLanded, + ) +} + +// TestDependentBatch_NoBypassWhenCoverageIsIncomplete proves the complement: +// with only the "dependency succeeds" side passed, a head waits for its +// dependency to resolve and merges strictly, never dispatching ahead of it. +// +// Partial coverage is seeded directly: the follower's batch is stranded in +// Created (build held), its "succeeds" path is written as passed, and a +// trigger request wakes the queue. The follower's real speculative builds are +// parked on its held batch partition, so the funded set stays exactly one +// path. The single passed path is a live passed path, so the run reports the +// wait — the signal this test then uses to prove nothing else advanced it. +func (s *E2EIntegrationSuite) TestDependentBatch_NoBypassWhenCoverageIsIncomplete() { + const queue = "e2e-chain-queue" + const gateGroup = "orchestrator" + + lead := s.land(queue, "github://github.example.com/uber/e2e-nobypass/pull/1/cccccccccccccccccccccccccccccccccccccccc") + leadBatch := s.awaitBatchID(lead) + s.awaitBatchState(queue, leadBatch, entity.BatchStateSpeculating) + + follower := s.land(queue, "github://github.example.com/uber/e2e-nobypass/pull/2/dddddddddddddddddddddddddddddddddddddddd") + heldBatch := s.awaitBatchID(follower) + s.closeGate(gateGroup, heldBatch, "e2e: hold the follower's builds so only the seeded path exists") + defer s.openGate(gateGroup, heldBatch) + + // Strand the follower in Created, then seed exactly one passed path: the + // guess that the lead succeeds. Its builds are parked, so nothing can add + // the "fails" path while the queue is quiet. + s.strandInCreated(queue, heldBatch) + s.seedPassedPath(queue, entity.SpeculationPath{ + Head: heldBatch, + Dependencies: []entity.PathDependency{ + {Batch: leadBatch, Assumption: entity.DependencyAssumptionSucceeds}, + }, + }) + + trigger := s.land(queue, "github://github.example.com/uber/e2e-nobypass/pull/3/eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee") + s.awaitBatchID(trigger) + + // The run that admits the follower also reports the wait: its one passed + // path covers only one of the lead's two outcomes, so it cannot merge + // until the lead resolves. + s.awaitEvent(follower, entity.RequestEventWaiting) + + // "No bypass" must not be read off the follower's batch state: waiting is + // a recorded event, not a moment, and the lead is ungated — it can land + // before this test looks, at which point the seeded path is mergeable + // *strictly* and the batch rightly moves to Merging. The bypass would + // show up differently: the follower merging while the lead is still + // unresolved. Assert the true invariant instead — the follower only lands + // after the lead has landed. + s.awaitStatus(lead, entity.RequestStatusLanded) + s.awaitStatus(follower, entity.RequestStatusLanded) + + // Let the queue finish. The follower's own speculative builds were parked + // on the held partition; releasing it lets any surviving work drain. How + // the queue ultimately converges is exercised by the other tests. + s.openGate(gateGroup, heldBatch) + s.awaitStatus(trigger, entity.RequestStatusLanded) } // TestReadAPIs validates all five request read endpoints against receipts @@ -434,23 +504,16 @@ func (s *E2EIntegrationSuite) TestReadAPIs() { assert.Equal(t, secondSummary.Request.LastError, secondEvents[len(secondEvents)-1].LastError) } -// TestLand_DependentBatch_StaysSpeculatingAcrossAnUnresolvedDependency covers -// the oscillation the request log used to report as a regression: a head -// speculates while a dependency is unresolved, the dependency then fails, and -// the head re-plans and lands anyway. Throughout, it is speculating exactly -// once — the trail must never revisit a stage. +// TestLand_DependentBatch_BypassesAnUnresolvedDependency proves the complete +// payoff: a follower built both with and without a held leader lands before the +// leader resolves. // // The wait is forced rather than raced. Batch IDs come from a per-queue counter // as "/batch/", so the leader on a fresh queue is batch/1, and the // build topic partitions by batch — closing the gate on that partition before // anything is published holds the leader's build and nothing else, so the // follower reaches a passed path while its dependency is still outstanding. -// -// No invalidated event is asserted. A passed path stops occupying build budget, -// so by the time the leader fails the follower has usually funded the other side -// of the guess too; it never loses its last live passed path, which is what -// invalidated reports. The unit tests cover that state directly. -func (s *E2EIntegrationSuite) TestLand_DependentBatch_StaysSpeculatingAcrossAnUnresolvedDependency() { +func (s *E2EIntegrationSuite) TestLand_DependentBatch_BypassesAnUnresolvedDependency() { const queue = "e2e-respeculate-queue" const gateGroup = "orchestrator" leaderBatch := queue + "/batch/1" @@ -462,21 +525,18 @@ func (s *E2EIntegrationSuite) TestLand_DependentBatch_StaysSpeculatingAcrossAnUn follower := s.land(queue, "github://github.example.com/uber/e2e-respeculate/pull/2/2222222222222222222222222222222222222222") s.log.Logf("Landed leader=%s (build held) follower=%s", leader.sqid, follower.sqid) - // The baseline analyzer serializes the queue, so the follower depends on the - // leader and speculates on it succeeding. That build passes while the leader - // is still held: the follower's own work is done and only the leader is - // outstanding, which is the wait. - s.awaitEvent(follower, entity.RequestEventWaiting) - assert.Equal(s.T(), entity.RequestStatusSpeculating, s.mustStatus(follower), - "a head waiting on its dependency has not finished speculating") + // The baseline analyzer serializes the queue, and the build budget lets the + // follower validate both possible outcomes while the leader is held. + s.awaitStatus(follower, entity.RequestStatusLanded) + assert.Equal(s.T(), entity.RequestStatusSpeculating, s.mustStatus(leader), + "the follower must land before the held leader resolves") - // Release the leader. Its build fails, contradicting the guess the follower - // speculated on, and the follower has to reach the trunk another way. + // Release the leader only after the follower has landed. Its later failure is + // one of the outcomes the follower already validated. s.openGate(gateGroup, leaderBatch) assert.Equal(s.T(), entity.RequestStatusError, s.awaitTerminal(leader), "the leader's build carries a failure marker, so it must not land") - s.awaitStatus(follower, entity.RequestStatusLanded) s.assertStatusesInOrder(follower, entity.RequestStatusSpeculating, entity.RequestStatusSpeculated,