From 180b4303d362da295fc759e6d99441c8cae098ef Mon Sep 17 00:00:00 2001 From: Preetam Dwivedi Date: Wed, 15 Jul 2026 13:05:21 -0700 Subject: [PATCH] feat(speculation): optimistic merge finalization behind a confirmed hand-off MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Waiting for a base dependency to be confirmed landed before finalizing a dependent costs a full runway round-trip per chain link. Per review direction, finalization is now optimistic: a base that is published for merge already counts as landed for a dependent's merge readiness, and a failed merge unwinds through signals instead of being prevented by waiting. The readiness rules live as two pure predicates in a new submitqueue/core/speculation package. PathMergeConfirmed is the strict rule and is not new logic — it is finalize's previously package-private mergeableNow, semantics unchanged: the path's own build Passed, every base dependency Succeeded or Cancelled out of the way, every non-base dependency ruled out. PathMergePossible is identical except a base still in Merging is tolerated: its merge may yet succeed (confirming the assumption) or fail (refuting it). Non-base dependencies get no such tolerance in either predicate — an undecided non-base dependency may yet land and invalidate the assumption set, so it must be ruled out, not bet on. The pair leaves the speculate package because the merge stage now enforces the same rules at the hand-off and sibling controllers do not import each other; per review, the shared home is core — the domain-internal layer for logic shared across stages — keeping entity a plain data layer. Speculate-only predicates (pathDead, viable) stay package-private as before — promotion to core/speculation is reserved for rules a second stage must evaluate identically. speculate finalizes on PathMergePossible: publish the merge trigger, CAS to Merging, and wake dependents immediately — a Merging batch already counts as landed for them, so a chain finalizes link after link without waiting on runway. The old no-op Merging branch becomes superviseMerging: every wake re-publishes the dependent fan-out (self-heal for a fan-out lost right after the CAS), and if no path is possible anymore — the base failed before any runway hand-off — the batch is failed with the shared terminal sequence (failBatch, extracted from finalize's no-viable-path branch), which re-speculates its own dependents in turn. That is the signal-back flow: mergesignal fails the base, the terminal fan-out wakes each dependent, pre-Merging dependents re-plan via pathDead, and Merging dependents are failed by supervision. What makes the optimism safe is the merge stage: the queue does not preserve per-partition order across nack backoff, so "trigger published first" never implies "reaches runway first", and ordering is instead enforced by state at the hand-off. Before building the runway request, the merge stage classifies the batch via the same predicates: confirmed hands off; possible (a base still unsettled) re-checks after WaitDelayMs by re-publishing its own trigger with a freshly minted message ID (the queue dedups publishes on (topic, partition, id) even against consumed rows, so reusing the consumed ID would silently end the cycle); refuted acks, drops, and nudges speculate directly with a minted ID — the terminal fan-out wake publishes under the bare batch ID, which that same dedup can silently swallow, and supervision must reliably run to fail the batch. Runway therefore only ever sees batches whose bases actually landed, in any delivery order. Nor does the delayed re-check surrender a serialization guarantee: per-partition FIFO was never the mechanism (nack backoff reorders delivery), and dependency order is the only order the hand-off must preserve — batches without a dependency edge have no overlapping targets, so their merges commute, while the gate holds back exactly the batches whose bases are unsettled. On runway's queue the gate yields an invariant stronger than ordering: a dependent's request is published only after its base is terminal, so a dependency chain's requests never even coexist in flight — which is also what lets each request carry only its own batch's steps, counting on the base already being on the target. A batch with no dependencies short-circuits as confirmed, so the common case costs no extra reads. Supervision never races a runway verdict: hand-off requires confirmed assumptions, confirmed outcomes are terminal dependency states, and possible ⊇ confirmed, so a handed-off batch always takes supervision's wake branch. Two hardening rules round the design out. A base in Cancelling counts as unsettled, exactly like Merging: it always settles to Cancelled or Succeeded (both confirming the assumption) or Failed (refuting it), so treating it as refuted would strand a waiting dependent in Merging forever when a slow-merging base gets cancelled — the wait cycle would drop with nothing left to re-create it — or fail a dependent whose bet was about to come true. And supervision re-arms the merge trigger with a minted ID on every healthy wake, so the delayed re-check chain — the only live message for a waiting Merging batch — has a self-heal if it is ever lost. The supervision re-arm is also what paces a chain: a settling base's fan-out re-arms its dependents' triggers at event speed, leaving the delayed re-check as a pure backstop whose cost is one cheap re-check per parked batch per cycle — merge.go's package doc carries the full ordering model with a worked chain example. Tests: core/speculation gains an exhaustive Confirmed/Possible table including the Cancelling-base split; speculate's merge gate pins that a Merging base now finalizes and wakes dependents, MergingSupervision covers wake-on-possible (Merging and Cancelling bases, asserting the minted-ID trigger re-arm) and fail-on-refuted, and the pure-function table moves to speculation.PathMergePossible; merge gains gate tests for all three outcomes — the minted-ID delayed re-check with no runway publish for both unsettled base states, and the refuted drop asserting the minted-ID speculate nudge. go build ./... and go test ./submitqueue/... ./service/... ./platform/... pass. bazel test //submitqueue/... //service/... passes. make e2e-test passes. make gazelle && make fmt — no BUILD changes, formatting clean. --- doc/rfc/submitqueue/speculation.md | 4 +- submitqueue/core/core.go | 13 +- submitqueue/core/speculation/BUILD.bazel | 19 ++ .../core/speculation/merge_readiness.go | 84 ++++++ .../core/speculation/merge_readiness_test.go | 144 ++++++++++ .../orchestrator/controller/merge/BUILD.bazel | 2 + .../orchestrator/controller/merge/merge.go | 234 ++++++++++++++++- .../controller/merge/merge_test.go | 163 ++++++++++++ .../controller/speculate/BUILD.bazel | 2 + .../controller/speculate/speculate.go | 245 ++++++++++++------ .../controller/speculate/speculate_test.go | 136 ++++++++-- 11 files changed, 928 insertions(+), 118 deletions(-) create mode 100644 submitqueue/core/speculation/BUILD.bazel create mode 100644 submitqueue/core/speculation/merge_readiness.go create mode 100644 submitqueue/core/speculation/merge_readiness_test.go diff --git a/doc/rfc/submitqueue/speculation.md b/doc/rfc/submitqueue/speculation.md index 97fad456..cd6bc0d9 100644 --- a/doc/rfc/submitqueue/speculation.md +++ b/doc/rfc/submitqueue/speculation.md @@ -121,7 +121,7 @@ The controller persists the **entire** tree — every enumerated path together w | `building → cancelled` | the build was cancelled | | `candidate → cancelled` | the path's base broke before it was ever sent | -Actions the selector can emit: `Build` (send this path to the build controller) or `Cancel` (drop it; cancel any build in flight). The selector leaves a path as-is by simply omitting it from its decisions. Note there is no merge/finalize action: **merging is the controller's job, not the selector's.** A path becomes mergeable when its build `passed` *and* its base matches what actually landed — that is deterministic, not a policy choice, so the controller finalizes it on its own (the existing `tryFinalize` → `merge` reconciliation). The selector only decides where to spend build resources; the prioritizer decides which of those actually run. +Actions the selector can emit: `Build` (send this path to the build controller) or `Cancel` (drop it; cancel any build in flight). The selector leaves a path as-is by simply omitting it from its decisions. Note there is no merge/finalize action: **merging is the controller's job, not the selector's.** A path becomes mergeable when its build `passed` *and* its base is out of the way — landed, cancelled, or itself already published for merge (optimistic finalization; the merge stage re-verifies dependency outcomes before handing anything to the merge executor, so a lost bet unwinds through signals instead of landing wrong) — that is deterministic, not a policy choice, so the controller finalizes it on its own. The selector only decides where to spend build resources; the prioritizer decides which of those actually run. Why `selected` is distinct from `building`: the selector only *sends* a path to the build controller. The build is then subject to prioritization and resources and may not start immediately. So `Build` moves the path to `selected`; speculate does not assert `building` itself — it learns a build actually started only from a build signal, and only then records `building` and the `BuildID`. Between the two, the path is sent but unconfirmed, and the selector treats `selected` as "already sent — don't re-send, but still cancellable." "Base invalid" is not a status — it is one of the *triggers* that sends a path to `cancelled`. @@ -180,7 +180,7 @@ The selector runs an optimistic top-1 policy: it returns `Build` for `[q1]→q2` [] q2 candidate 0.90 [] q2 candidate 0.90 ← selector now returns Build ``` -- **Bet holds** — `q1` succeeds on its own (its `[]→q1` build passes and it merges): the build of `[q1]→q2` ran against exactly the tree `q2` will land on, so `q2` is mergeable and the controller finalizes it (publishes to `merge`) — no selector action involved. `q1` and `q2` were validated in parallel — the latency win. Because `q1` landed, the `[]→q2` fallback is now invalid — `q2` must include `q1` — so it is dropped: the selector returns `Cancel` for it and the controller cancels any build still in flight. +- **Bet holds** — `q1` succeeds on its own (its `[]→q1` build passes and it heads to merge): the build of `[q1]→q2` ran against exactly the tree `q2` will land on, so `q2` is mergeable as soon as `q1` is published for merge and the controller finalizes it right behind — no selector action involved. `q1` and `q2` were validated in parallel — the latency win. Because `q1` landed, the `[]→q2` fallback is now invalid — `q2` must include `q1` — so it is dropped: the selector returns `Cancel` for it and the controller cancels any build still in flight. - **Bet fails** — `q1` fails: the `[q1]→q2` path's base is broken, so the controller stamps it `cancelled`. Re-running the selector over the updated tree returns `Build` for the surviving `[]→q2` candidate; `q2` still lands, just without the head start. Re-speculation needs no special undo path: the controller refreshes statuses, and the selector simply re-runs over the updated tree. diff --git a/submitqueue/core/core.go b/submitqueue/core/core.go index abb2e052..6a471846 100644 --- a/submitqueue/core/core.go +++ b/submitqueue/core/core.go @@ -12,10 +12,11 @@ // See the License for the specific language governing permissions and // limitations under the License. -// Package core groups infrastructure shared across SubmitQueue's own services -// (gateway and orchestrator) — the SubmitQueue-scoped analogue of the repo-level -// core/. Cross-domain infrastructure lives in the top-level core/; this package -// is for plumbing private to SubmitQueue. Subpackages: core/consumer (queue -// consumption framework) and core/request (request lifecycle shared by gateway -// and orchestrator). +// Package core groups infrastructure and domain logic shared across +// SubmitQueue's own services and pipeline stages — the SubmitQueue-scoped +// analogue of the repo-level platform/. Cross-domain code lives under +// platform/; this package is for plumbing and shared rules private to +// SubmitQueue. Subpackages hold queue topic keys (topickey), request +// lifecycle plumbing (request), changeset resolution (changeset), and +// speculation rules evaluated by multiple stages (speculation). package core diff --git a/submitqueue/core/speculation/BUILD.bazel b/submitqueue/core/speculation/BUILD.bazel new file mode 100644 index 00000000..988dabf6 --- /dev/null +++ b/submitqueue/core/speculation/BUILD.bazel @@ -0,0 +1,19 @@ +load("@rules_go//go:def.bzl", "go_library", "go_test") + +go_library( + name = "go_default_library", + srcs = ["merge_readiness.go"], + importpath = "github.com/uber/submitqueue/submitqueue/core/speculation", + visibility = ["//visibility:public"], + deps = ["//submitqueue/entity:go_default_library"], +) + +go_test( + name = "go_default_test", + srcs = ["merge_readiness_test.go"], + embed = [":go_default_library"], + deps = [ + "//submitqueue/entity:go_default_library", + "@com_github_stretchr_testify//assert:go_default_library", + ], +) diff --git a/submitqueue/core/speculation/merge_readiness.go b/submitqueue/core/speculation/merge_readiness.go new file mode 100644 index 00000000..954c46e9 --- /dev/null +++ b/submitqueue/core/speculation/merge_readiness.go @@ -0,0 +1,84 @@ +// 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 speculation holds pure speculation-domain rules that more than one +// orchestrator stage must evaluate identically. It interprets entity state but +// owns none: no storage, no queues, no pipeline step — just functions of +// entities in, verdicts out. Rules consumed by a single stage do not belong +// here; they stay package-private in that stage until a second consumer +// appears. +package speculation + +import "github.com/uber/submitqueue/submitqueue/entity" + +// PathMergeConfirmed reports whether p's merge assumptions are fully +// confirmed by dependency outcomes: p's own build Passed, every base +// dependency either landed (entity.BatchStateSucceeded) or was cancelled out +// of the way (entity.BatchStateCancelled), and every dependency of the head +// outside the base has been ruled out (entity.BatchStateFailed or +// entity.BatchStateCancelled). depByID holds the head's dependency batches +// keyed by batch ID; a base entry absent from the map is treated as out of +// the way. +func PathMergeConfirmed(p entity.SpeculationPathInfo, depByID map[string]entity.Batch) bool { + return pathMergeReady(p, depByID, false) +} + +// PathMergePossible reports whether p's merge assumptions are confirmed or +// may still confirm: identical to PathMergeConfirmed, except a base +// dependency still in entity.BatchStateMerging or entity.BatchStateCancelling +// is tolerated. Both are transient and always settle to a terminal state that +// either confirms the assumption (Succeeded or Cancelled) or refutes it +// (Failed), so neither is a verdict yet. A path that is not possible can +// never merge. A dependency outside the base gets no such tolerance in +// either predicate: a still-undecided non-base dependency may yet land and +// invalidate the assumption set, so it must be ruled out, not bet on. +func PathMergePossible(p entity.SpeculationPathInfo, depByID map[string]entity.Batch) bool { + return pathMergeReady(p, depByID, true) +} + +// pathMergeReady is the shared evaluation behind PathMergeConfirmed and +// PathMergePossible; tolerateUnsettledBase is the only difference between +// the two. +func pathMergeReady(p entity.SpeculationPathInfo, depByID map[string]entity.Batch, tolerateUnsettledBase bool) bool { + if p.Status != entity.SpeculationPathStatusPassed { + return false + } + + 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 + } + switch d.State { + case entity.BatchStateSucceeded, entity.BatchStateCancelled: + case entity.BatchStateMerging, entity.BatchStateCancelling: + if !tolerateUnsettledBase { + return false + } + default: + return false + } + } + for id, d := range depByID { + if inBase[id] { + continue + } + if d.State != entity.BatchStateFailed && d.State != entity.BatchStateCancelled { + return false + } + } + return true +} diff --git a/submitqueue/core/speculation/merge_readiness_test.go b/submitqueue/core/speculation/merge_readiness_test.go new file mode 100644 index 00000000..6e31da0a --- /dev/null +++ b/submitqueue/core/speculation/merge_readiness_test.go @@ -0,0 +1,144 @@ +// 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 speculation + +import ( + "testing" + + "github.com/stretchr/testify/assert" + + "github.com/uber/submitqueue/submitqueue/entity" +) + +// PathMergeConfirmed and PathMergePossible differ in exactly one input: a +// base dependency in BatchStateMerging. Table-test the split and the shared +// rules (Passed required, base landed-or-cancelled, non-base ruled out, +// missing base entries tolerated). +func TestPathMergeConfirmedAndPossible(t *testing.T) { + base := "q/batch/1" + head := "q/batch/2" + basePath := entity.SpeculationPath{Base: []string{base}, Head: head} + alonePath := entity.SpeculationPath{Head: head} + + tests := []struct { + name string + path entity.SpeculationPath + status entity.SpeculationPathStatus + deps map[string]entity.Batch + wantConfirmed bool + wantPossible bool + }{ + { + name: "base_succeeded_confirmed", + path: basePath, + status: entity.SpeculationPathStatusPassed, + deps: map[string]entity.Batch{base: {ID: base, State: entity.BatchStateSucceeded}}, + wantConfirmed: true, + wantPossible: true, + }, + { + name: "base_cancelled_confirmed", + path: basePath, + status: entity.SpeculationPathStatusPassed, + deps: map[string]entity.Batch{base: {ID: base, State: entity.BatchStateCancelled}}, + wantConfirmed: true, + wantPossible: true, + }, + { + name: "base_merging_possible_not_confirmed", + path: basePath, + status: entity.SpeculationPathStatusPassed, + deps: map[string]entity.Batch{base: {ID: base, State: entity.BatchStateMerging}}, + wantConfirmed: false, + wantPossible: true, + }, + { + // Cancelling is transient: it settles to Cancelled or Succeeded + // (both confirming) or Failed (refuting), so it is a wait, not + // a verdict. + name: "base_cancelling_possible_not_confirmed", + path: basePath, + status: entity.SpeculationPathStatusPassed, + deps: map[string]entity.Batch{base: {ID: base, State: entity.BatchStateCancelling}}, + wantConfirmed: false, + wantPossible: true, + }, + { + name: "base_failed_neither", + path: basePath, + status: entity.SpeculationPathStatusPassed, + deps: map[string]entity.Batch{base: {ID: base, State: entity.BatchStateFailed}}, + wantConfirmed: false, + wantPossible: false, + }, + { + name: "base_speculating_neither", + path: basePath, + status: entity.SpeculationPathStatusPassed, + deps: map[string]entity.Batch{base: {ID: base, State: entity.BatchStateSpeculating}}, + wantConfirmed: false, + wantPossible: false, + }, + { + name: "base_absent_from_deps_tolerated", + path: basePath, + status: entity.SpeculationPathStatusPassed, + deps: map[string]entity.Batch{}, + wantConfirmed: true, + wantPossible: true, + }, + { + name: "non_base_failed_ruled_out", + path: alonePath, + status: entity.SpeculationPathStatusPassed, + deps: map[string]entity.Batch{base: {ID: base, State: entity.BatchStateFailed}}, + wantConfirmed: true, + wantPossible: true, + }, + { + name: "non_base_merging_blocks_both", + path: alonePath, + status: entity.SpeculationPathStatusPassed, + // No Merging tolerance outside the base: the dependency may yet + // land and invalidate the assumption set. + deps: map[string]entity.Batch{base: {ID: base, State: entity.BatchStateMerging}}, + wantConfirmed: false, + wantPossible: false, + }, + { + name: "non_base_succeeded_blocks_both", + path: alonePath, + status: entity.SpeculationPathStatusPassed, + deps: map[string]entity.Batch{base: {ID: base, State: entity.BatchStateSucceeded}}, + wantConfirmed: false, + wantPossible: false, + }, + { + name: "unpassed_path_blocks_both", + path: basePath, + status: entity.SpeculationPathStatusBuilding, + deps: map[string]entity.Batch{base: {ID: base, State: entity.BatchStateSucceeded}}, + wantConfirmed: false, + wantPossible: false, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + info := entity.SpeculationPathInfo{Path: tt.path, Status: tt.status} + assert.Equal(t, tt.wantConfirmed, PathMergeConfirmed(info, tt.deps)) + assert.Equal(t, tt.wantPossible, PathMergePossible(info, tt.deps)) + }) + } +} diff --git a/submitqueue/orchestrator/controller/merge/BUILD.bazel b/submitqueue/orchestrator/controller/merge/BUILD.bazel index 2e8710ab..053822b3 100644 --- a/submitqueue/orchestrator/controller/merge/BUILD.bazel +++ b/submitqueue/orchestrator/controller/merge/BUILD.bazel @@ -13,6 +13,8 @@ go_library( "//platform/base/messagequeue:go_default_library", "//platform/consumer:go_default_library", "//platform/metrics:go_default_library", + "//submitqueue/core/speculation:go_default_library", + "//submitqueue/core/topickey:go_default_library", "//submitqueue/entity:go_default_library", "//submitqueue/extension/storage:go_default_library", "@com_github_uber_go_tally//:go_default_library", diff --git a/submitqueue/orchestrator/controller/merge/merge.go b/submitqueue/orchestrator/controller/merge/merge.go index d1d886ce..e00cec7f 100644 --- a/submitqueue/orchestrator/controller/merge/merge.go +++ b/submitqueue/orchestrator/controller/merge/merge.go @@ -19,11 +19,59 @@ // correlation id. Runway performs the merge out of process and publishes the // result to the merge-signal queue, which the mergesignal stage consumes and // correlates back to the batch by that id. +// +// # The hand-off gate +// +// Speculation finalizes optimistically: a batch is published here as soon as +// its base is itself published for merge, not once it has landed. This stage +// is the guard that makes the optimism safe. Every delivery re-derives the +// batch's merge readiness from current dependency state (the shared +// speculation.PathMergeConfirmed / PathMergePossible rules) and takes one of +// three actions: confirmed hands off to runway; possible — a base still +// Merging, or Cancelling on its way to a terminal state — parks the batch, +// acking now and re-checking after WaitDelayMs; refuted — a base failed — +// drops the trigger and nudges speculate, which owns failing the batch. +// Runway therefore only ever sees batches whose bases actually landed, +// whatever order messages arrive in. +// +// # Ordering +// +// Two queues with different contracts are in play here, and only one of them +// is ordered. The trigger topic this stage consumes is an unordered work +// queue of "re-evaluate this batch" wake-ups: every delivery re-reads state, +// so processing order is irrelevant, and a parked batch acks and re-queues +// itself rather than holding the partition. Per-partition FIFO could not be +// trusted for correctness anyway — nack backoff delivers later offsets past +// a backing-off earlier one. Runway's queue is where dependency order binds, +// and there the gate yields an invariant stronger than ordering: for every +// dependency edge, the dependent's MergeRequest is published only after the +// base is terminal, so a chain's requests never even coexist in flight and +// runway needs no cross-request ordering. That is also what makes the +// request shape valid at all — a request carries only its own batch's steps, +// counting on the base's changes already being on the target. Batches with +// no dependency edge have no overlapping targets, so their merges commute +// and need no ordering in the first place. +// +// Concretely, take a chain B1 <- B2 <- B3, all finalized optimistically and +// all in Merging. B1 (no unsettled base) confirms and hands off; B2 and B3 +// are possible-but-not-confirmed, so each parks. Their delayed re-checks now +// interleave on the trigger topic in whatever order — B3 ahead of B2 one +// cycle, B2 ahead of B3 the next — and none of it matters: B3's re-check +// cannot confirm before B2 has landed no matter when it runs, it just parks +// again. When B1's verdict returns, mergesignal fans out, speculate wakes +// each dependent, and Merging supervision re-arms the waiting batch's +// trigger immediately — so B2 confirms at event speed and the chain hands +// off link by link, each link paced by its runway round-trip. The +// WaitDelayMs re-check is a backstop for a lost or dedup-swallowed wake, not +// the pace. The cost of a parked chain of depth D is D futile re-checks per +// cycle — a few point reads and one delayed publish each — which is why +// WaitDelayMs can be raised freely on a hot queue. package merge import ( "context" "fmt" + "time" "github.com/uber-go/tally" "go.uber.org/zap" @@ -35,10 +83,21 @@ import ( entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/submitqueue/core/speculation" + "github.com/uber/submitqueue/submitqueue/core/topickey" "github.com/uber/submitqueue/submitqueue/entity" "github.com/uber/submitqueue/submitqueue/extension/storage" ) +// WaitDelayMs is the delay before re-checking a batch whose merge is +// triggered but whose base dependency has not settled yet (Merging or +// Cancelling). The re-check is the backup wake channel — the prompt one is +// the settling base's fan-out re-arming the trigger through speculate's +// Merging supervision — so raising it trades only backstop latency, never +// chain pace (see the package doc's Ordering section). Var (not const) so +// tests can shorten it; the orchestrator always uses the default. +var WaitDelayMs int64 = 2000 + // Controller handles merge queue messages. Implements consumer.Controller. // // It loads the batch and its member requests, assembles the full merge request @@ -83,13 +142,19 @@ func NewController( } } -// Process publishes the full merge request to runway. Returns nil to ack -// (success), or error to nack/reject. +// Process publishes the full merge request to runway once the batch's +// dependency outcomes are confirmed. Speculation triggers this stage +// optimistically — possibly while a base dependency is still unsettled — so +// the hand-off is gated: confirmed hands off, still-possible re-checks +// after WaitDelayMs, refuted acks and drops (speculate owns failing the +// batch). Returns nil to ack (success), or error to nack/reject. // -// Error classification: deserialize and storage failures are non-retryable -// (reject to DLQ). The publish to runway is retryable — it is the hand-off that -// keeps the merge alive, so a transient enqueue blip should replay rather than -// strand the batch. +// Error classification: deserialize failures are non-retryable and reject +// to DLQ. Storage reads and every publish here — to runway, the delayed +// self re-check, and the refuted-batch speculate nudge — retry for +// transient causes via the shared MySQL classifier and otherwise +// dead-letter; the publishes are what keep the merge alive, so an enqueue +// blip should replay rather than strand the batch. func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) { const opName = "process" @@ -123,7 +188,8 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r // kicked off for a batch that will not proceed. Unlike the old synchronous // merge there is no terminal re-fan-out here — the mergesignal stage owns the // state transition and fan-out once runway's result returns, so a redelivery - // at this stage simply acks. + // at this stage simply acks. This also ends the delayed re-check cycle for a + // batch that speculate failed while it was waiting. if entity.IsBatchStateHalted(batch.State) { metrics.NamedCounter(c.metricsScope, opName, "skipped_halted", 1) c.logger.Infow("skipping merge for halted batch", @@ -133,6 +199,65 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r return nil } + // The optimistic hand-off gate: only publish to runway once some path's + // merge assumptions are confirmed by dependency outcomes. Ordering is + // enforced by state, not by queue position — see the package doc's + // Ordering section for the full model: why this trigger topic is + // unordered, why a dependency chain's runway requests never coexist in + // flight, and what a parked chain costs per re-check cycle. + confirmed, possible, err := c.mergeReadiness(ctx, batch) + if err != nil { + return err + } + if !confirmed { + if possible { + // A base is still unsettled (Merging or Cancelling): it will + // land, cancel out of the way, or fail — all of which settle + // this classification one way or the other. Keep the attempt + // alive by re-checking after a delay; each cycle mints a + // fresh message ID because the queue dedups publishes on + // (topic, partition, id) even against consumed rows — reusing + // the consumed message's ID would silently end the cycle. The + // delay is a backstop, not the chain's pace: a settling base's + // terminal fan-out re-arms this trigger promptly through + // speculate's Merging supervision (see the package doc's + // Ordering section). + // + // TODO(merge-wait-model): the park-and-poll model is a + // deliberate first cut — it rides an unordered queue as a + // delay timer and burns one futile re-check per parked batch + // per cycle, O(chain depth) every WaitDelayMs. Figure out a + // better model: wake purely on dependency events with a + // hardened delivery path, back the delay off by time spent in + // Merging, or park only the chain's frontier batch. + metrics.NamedCounter(c.metricsScope, opName, "waiting_on_base", 1) + c.logger.Debugw("base dependency not settled; delaying runway hand-off", + "batch_id", batch.ID, + ) + if err := c.republishSelfAfter(ctx, batch, WaitDelayMs); err != nil { + metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1) + return fmt.Errorf("failed to re-publish merge trigger for batch %s: %w", batch.ID, err) + } + return nil + } + // No path can confirm anymore — a base failed after the optimistic + // finalize. Never hand off: runway must not see a batch whose base + // did not land. Speculate's Merging supervision fails the batch — + // and it is nudged directly with a minted message ID, because the + // dependent-side terminal fan-out publishes under the batch ID and + // the queue's (topic, partition, id) dedup can silently swallow + // that wake against an un-GC'd consumed row. + metrics.NamedCounter(c.metricsScope, opName, "refuted", 1) + c.logger.Warnw("no path can confirm; dropping merge trigger", + "batch_id", batch.ID, + ) + if err := c.publishSpeculateNudge(ctx, batch); err != nil { + metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1) + return fmt.Errorf("failed to nudge speculate for refuted batch %s: %w", batch.ID, err) + } + return nil + } + // Build the full payload runway needs to perform the merge. The batch id is // the client-owned correlation id, so a redelivery republishes the same id // and runway dedupes on it; the result is matched straight back to the batch. @@ -156,6 +281,101 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r return nil // Success - message will be acked } +// mergeReadiness classifies the batch's merge assumptions against its +// dependencies' current states: confirmed when some path's assumptions are +// fully confirmed (speculation.PathMergeConfirmed), else possible when some +// path may still confirm because a base is still unsettled +// (speculation.PathMergePossible), else neither. A batch with no dependencies is +// trivially confirmed — speculation only triggers a merge once some path's +// own build Passed, and with nothing to wait on that cannot regress — so +// the common no-dependency case costs no extra reads. +func (c *Controller) mergeReadiness(ctx context.Context, batch entity.Batch) (confirmed, possible bool, err error) { + if len(batch.Dependencies) == 0 { + return true, true, nil + } + + depByID := make(map[string]entity.Batch, len(batch.Dependencies)) + for _, depID := range batch.Dependencies { + d, derr := c.store.GetBatchStore().Get(ctx, depID) + if derr != nil { + metrics.NamedCounter(c.metricsScope, "process", "storage_errors", 1) + return false, false, fmt.Errorf("failed to get dependency batch %s of %s: %w", depID, batch.ID, derr) + } + depByID[depID] = d + } + + // A merge trigger only exists because speculate finalized from this + // batch's tree, and trees are never deleted — a miss is corrupted + // state, surfaced as an error so the message dead-letters loudly. + tree, terr := c.store.GetSpeculationTreeStore().Get(ctx, batch.ID) + if terr != nil { + metrics.NamedCounter(c.metricsScope, "process", "storage_errors", 1) + return false, false, fmt.Errorf("failed to get speculation tree for batch %s: %w", batch.ID, terr) + } + + for _, p := range tree.Paths { + if speculation.PathMergeConfirmed(p, depByID) { + return true, true, nil + } + if speculation.PathMergePossible(p, depByID) { + possible = true + } + } + return false, possible, nil +} + +// republishSelfAfter re-publishes the batch's merge trigger to this +// controller's own topic after delayMs, minting a fresh message ID per +// cycle: the queue dedups publishes on (topic, partition, id) against +// un-GC'd rows including consumed ones, so reusing an ID would silently end +// the re-check cycle. +func (c *Controller) republishSelfAfter(ctx context.Context, batch entity.Batch, delayMs int64) error { + payload, err := entity.BatchID{ID: batch.ID}.ToBytes() + if err != nil { + return fmt.Errorf("failed to serialize batch ID: %w", err) + } + + msgID := fmt.Sprintf("%s/mergewait/%d", batch.ID, time.Now().UnixMilli()) + msg := entityqueue.NewMessage(msgID, payload, batch.Queue, nil) + + q, ok := c.registry.Queue(c.topicKey) + if !ok { + return fmt.Errorf("no queue registered for topic key %s", c.topicKey) + } + topicName, ok := c.registry.TopicName(c.topicKey) + if !ok { + return fmt.Errorf("no topic name registered for topic key %s", c.topicKey) + } + return q.Publisher().PublishAfter(ctx, topicName, msg, delayMs) +} + +// publishSpeculateNudge wakes the speculate stage for a batch whose merge +// trigger was refuted, minting a fresh message ID so the queue's +// (topic, partition, id) publish dedup cannot swallow it the way it can a +// terminal fan-out wake published under the bare batch ID. +func (c *Controller) publishSpeculateNudge(ctx context.Context, batch entity.Batch) error { + payload, err := entity.BatchID{ID: batch.ID}.ToBytes() + if err != nil { + return fmt.Errorf("failed to serialize batch ID: %w", err) + } + + msgID := fmt.Sprintf("%s/mergerefuted/%d", batch.ID, time.Now().UnixMilli()) + msg := entityqueue.NewMessage(msgID, payload, batch.Queue, nil) + + q, ok := c.registry.Queue(topickey.TopicKeySpeculate) + if !ok { + return fmt.Errorf("no queue registered for topic key %s", topickey.TopicKeySpeculate) + } + topicName, ok := c.registry.TopicName(topickey.TopicKeySpeculate) + if !ok { + return fmt.Errorf("no topic name registered for topic key %s", topickey.TopicKeySpeculate) + } + if err := q.Publisher().Publish(ctx, topicName, msg); err != nil { + return fmt.Errorf("failed to publish message: %w", err) + } + return nil +} + // buildMergeRequest loads the batch's member requests and assembles the runway // merge request: one MergeStep per request, in Contains order, attributed by // request id and carrying that request's change and land strategy. diff --git a/submitqueue/orchestrator/controller/merge/merge_test.go b/submitqueue/orchestrator/controller/merge/merge_test.go index 8207cb44..a0095397 100644 --- a/submitqueue/orchestrator/controller/merge/merge_test.go +++ b/submitqueue/orchestrator/controller/merge/merge_test.go @@ -17,6 +17,7 @@ package merge import ( "context" "fmt" + "strings" "testing" "github.com/stretchr/testify/assert" @@ -242,3 +243,165 @@ func TestProcess_BatchStoreGetFailureNotRetryable(t *testing.T) { require.Error(t, err) assert.False(t, errs.IsRetryable(err)) } + +// gateStorage wires the storage shape the dependency-outcome gate needs: the +// batch, its one dependency, and its speculation tree carrying a single +// Passed path based on that dependency. +func gateStorage(ctrl *gomock.Controller, batch, dep entity.Batch, tree entity.SpeculationTree) *storagemock.MockStorage { + batchStore := storagemock.NewMockBatchStore(ctrl) + batchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil) + batchStore.EXPECT().Get(gomock.Any(), dep.ID).Return(dep, nil) + treeStore := storagemock.NewMockSpeculationTreeStore(ctrl) + treeStore.EXPECT().Get(gomock.Any(), batch.ID).Return(tree, nil) + + store := storagemock.NewMockStorage(ctrl) + store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() + store.EXPECT().GetSpeculationTreeStore().Return(treeStore).AnyTimes() + return store +} + +// gateFixtures returns a Merging batch with one dependency in depState, that +// dependency, and a tree whose single Passed path is based on it. +func gateFixtures(depState entity.BatchState) (entity.Batch, entity.Batch, entity.SpeculationTree) { + dep := entity.Batch{ID: "test-queue/batch/0", Queue: "test-queue", State: depState, Version: 3} + batch := entity.Batch{ + ID: "test-queue/batch/1", + Queue: "test-queue", + Contains: []string{"test-queue/1"}, + Dependencies: []string{dep.ID}, + State: entity.BatchStateMerging, + Version: 2, + } + tree := entity.SpeculationTree{ + BatchID: batch.ID, + Version: 4, + Paths: []entity.SpeculationPathInfo{{ + ID: "test-queue/batch/1/path/0", + Path: entity.SpeculationPath{Base: []string{dep.ID}, Head: batch.ID}, + Status: entity.SpeculationPathStatusPassed, + }}, + } + return batch, dep, tree +} + +// TestProcess_GateConfirmedPublishesToRunway: a landed base confirms the +// path's assumptions, so the gate opens and the full merge request is handed +// to runway. +func TestProcess_GateConfirmedPublishesToRunway(t *testing.T) { + ctrl := gomock.NewController(t) + + batch, dep, tree := gateFixtures(entity.BatchStateSucceeded) + store := gateStorage(ctrl, batch, dep, tree) + + req := entity.Request{ID: "test-queue/1", Queue: "test-queue", LandStrategy: mergestrategy.MergeStrategyRebase} + reqStore := storagemock.NewMockRequestStore(ctrl) + reqStore.EXPECT().Get(gomock.Any(), req.ID).Return(req, nil) + store.EXPECT().GetRequestStore().Return(reqStore).AnyTimes() + + var gotTopic string + var gotPayload []byte + pub := queuemock.NewMockPublisher(ctrl) + pub.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn( + func(_ context.Context, topic string, msg entityqueue.Message) error { + gotTopic = topic + gotPayload = msg.Payload + return nil + }, + ) + q := queuemock.NewMockQueue(ctrl) + q.EXPECT().Publisher().Return(pub).AnyTimes() + registry, err := consumer.NewTopicRegistry([]consumer.TopicConfig{ + {Key: runwaymq.TopicKeyMerge, Name: "merge", Queue: q}, + {Key: topickey.TopicKeyMerge, Name: "submitqueue-merge", Queue: q}, + }) + require.NoError(t, err) + + c := newController(t, store, registry) + require.NoError(t, c.Process(context.Background(), newDelivery(t, ctrl, batch.ID, batch.Queue))) + + assert.Equal(t, "merge", gotTopic) + got := &runwaymq.MergeRequest{} + require.NoError(t, runwaymq.Unmarshal(gotPayload, got)) + assert.Equal(t, batch.ID, got.Id) + require.Len(t, got.Steps, 1) +} + +// TestProcess_GateUnsettledBaseWaits: a base still in Merging — or in +// transient Cancelling on its way to a terminal state — keeps the gate +// closed but the attempt alive: the trigger is re-published to this stage's +// own topic after WaitDelayMs with a freshly minted message ID (reusing the +// consumed ID would be silently deduped and end the cycle). Nothing reaches +// runway. +func TestProcess_GateUnsettledBaseWaits(t *testing.T) { + for _, depState := range []entity.BatchState{ + entity.BatchStateMerging, + entity.BatchStateCancelling, + } { + t.Run(string(depState), func(t *testing.T) { + ctrl := gomock.NewController(t) + + batch, dep, tree := gateFixtures(depState) + store := gateStorage(ctrl, batch, dep, tree) + // No request-store reads: the merge request is never built while waiting. + + pub := queuemock.NewMockPublisher(ctrl) + pub.EXPECT(). + PublishAfter(gomock.Any(), "submitqueue-merge", gomock.Any(), WaitDelayMs). + DoAndReturn(func(_ context.Context, _ string, msg entityqueue.Message, _ int64) error { + assert.True(t, strings.HasPrefix(msg.ID, batch.ID+"/mergewait/"), "re-check message must mint a fresh ID, got %q", msg.ID) + bid, err := entity.BatchIDFromBytes(msg.Payload) + require.NoError(t, err) + assert.Equal(t, batch.ID, bid.ID) + assert.Equal(t, batch.Queue, msg.PartitionKey) + return nil + }) + // No Publish expectation: a runway hand-off while waiting fails the test. + q := queuemock.NewMockQueue(ctrl) + q.EXPECT().Publisher().Return(pub).AnyTimes() + registry, err := consumer.NewTopicRegistry([]consumer.TopicConfig{ + {Key: runwaymq.TopicKeyMerge, Name: "merge", Queue: q}, + {Key: topickey.TopicKeyMerge, Name: "submitqueue-merge", Queue: q}, + }) + require.NoError(t, err) + + c := newController(t, store, registry) + require.NoError(t, c.Process(context.Background(), newDelivery(t, ctrl, batch.ID, batch.Queue))) + }) + } +} + +// TestProcess_GateRefutedDropsAndNudgesSpeculate: a failed base refutes +// every path, so the trigger is dropped — no runway hand-off, no re-check — +// and speculate is nudged directly with a minted message ID so its Merging +// supervision reliably runs and fails the batch (the terminal fan-out wake +// publishes under the bare batch ID, which publish dedup can swallow). +func TestProcess_GateRefutedDropsAndNudgesSpeculate(t *testing.T) { + ctrl := gomock.NewController(t) + + batch, dep, tree := gateFixtures(entity.BatchStateFailed) + store := gateStorage(ctrl, batch, dep, tree) + + // The only publish is the speculate nudge: no runway hand-off, no + // delayed re-check. + pub := queuemock.NewMockPublisher(ctrl) + pub.EXPECT(). + Publish(gomock.Any(), "speculate", gomock.Any()). + DoAndReturn(func(_ context.Context, _ string, msg entityqueue.Message) error { + assert.True(t, strings.HasPrefix(msg.ID, batch.ID+"/mergerefuted/"), "nudge must mint a fresh ID, got %q", msg.ID) + bid, err := entity.BatchIDFromBytes(msg.Payload) + require.NoError(t, err) + assert.Equal(t, batch.ID, bid.ID) + return nil + }) + q := queuemock.NewMockQueue(ctrl) + q.EXPECT().Publisher().Return(pub).AnyTimes() + registry, err := consumer.NewTopicRegistry([]consumer.TopicConfig{ + {Key: runwaymq.TopicKeyMerge, Name: "merge", Queue: q}, + {Key: topickey.TopicKeyMerge, Name: "submitqueue-merge", Queue: q}, + {Key: topickey.TopicKeySpeculate, Name: "speculate", Queue: q}, + }) + require.NoError(t, err) + + c := newController(t, store, registry) + require.NoError(t, c.Process(context.Background(), newDelivery(t, ctrl, batch.ID, batch.Queue))) +} diff --git a/submitqueue/orchestrator/controller/speculate/BUILD.bazel b/submitqueue/orchestrator/controller/speculate/BUILD.bazel index b6f312da..578549ac 100644 --- a/submitqueue/orchestrator/controller/speculate/BUILD.bazel +++ b/submitqueue/orchestrator/controller/speculate/BUILD.bazel @@ -9,6 +9,7 @@ go_library( "//platform/base/messagequeue:go_default_library", "//platform/consumer:go_default_library", "//platform/metrics:go_default_library", + "//submitqueue/core/speculation:go_default_library", "//submitqueue/core/topickey:go_default_library", "//submitqueue/entity:go_default_library", "//submitqueue/extension/speculation/dependencylimit:go_default_library", @@ -30,6 +31,7 @@ go_test( "//platform/consumer:go_default_library", "//platform/errs:go_default_library", "//platform/extension/messagequeue/mock:go_default_library", + "//submitqueue/core/speculation:go_default_library", "//submitqueue/core/topickey:go_default_library", "//submitqueue/entity:go_default_library", "//submitqueue/extension/speculation/dependencylimit/fake:go_default_library", diff --git a/submitqueue/orchestrator/controller/speculate/speculate.go b/submitqueue/orchestrator/controller/speculate/speculate.go index 99d5db01..2b5e3311 100644 --- a/submitqueue/orchestrator/controller/speculate/speculate.go +++ b/submitqueue/orchestrator/controller/speculate/speculate.go @@ -18,11 +18,13 @@ import ( "context" "errors" "fmt" + "time" "github.com/uber-go/tally" entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/submitqueue/core/speculation" "github.com/uber/submitqueue/submitqueue/core/topickey" "github.com/uber/submitqueue/submitqueue/entity" "github.com/uber/submitqueue/submitqueue/extension/speculation/dependencylimit" @@ -57,7 +59,10 @@ import ( // Cancelled), route the persisted intents to the build stage via // prioritize, and — only once every build has quiesced — respeculate // dependents, CAS to terminal Cancelled, publish to conclude. -// - Merging → no-op (owned by the merge controller). +// - Merging → superviseMerging: wake dependents (a Merging batch already +// counts as landed for their merge readiness — see below) and fail the +// batch if no path can confirm anymore (its base failed before the +// runway hand-off ever happened). // - Terminal → re-publish the dependent fan-out and the conclude event. // Every terminal transition is routed back through this controller, so // this branch is how waiting dependents learn a dependency resolved; @@ -86,22 +91,26 @@ import ( // does not publish straight to build: a path only reaches build once // prioritize admits it under the queue's build budget, so the pipeline is // speculate → prioritize → build. -// 9. Finalize: if some path is mergeable right now, publish to merge and CAS -// to Merging. Otherwise, if no path can ever merge, CAS to Failed and -// publish to conclude. Otherwise wait for the next event. +// 9. Finalize: if some path is mergeable now — optimistically, see below — +// publish to merge, CAS to Merging, and wake dependents. Otherwise, if +// no path can ever merge, CAS to Failed and publish to conclude. +// Otherwise wait for the next event. // // Two things must both hold before a path merges: its own build Passed, and -// its base actually landed (Succeeded). The build result is checked here on -// purpose — a failed build should stop the merge directly, not surface -// later as a runway merge failure. +// its base is out of the way. The build result is checked here on purpose — +// a failed build should stop the merge directly, not surface later as a +// runway merge failure. // -// A base that is only Merging — published for merge but not yet confirmed — -// does NOT count as landed. The queue can deliver messages out of order -// when one is retrying, so our merge request could reach runway before the -// base's, and we would land changes that were only validated on top of a -// base that never landed. Merging optimistically behind a Merging base -// would require the merge stage to verify dependency outcomes before -// handing off to runway; relaxing this check alone is not safe. +// The base check is optimistic: a base that is published for merge +// (Merging) already counts, so a dependent finalizes right behind its base +// instead of waiting a runway round-trip per chain link +// (speculation.PathMergePossible). The safety net lives downstream — the merge +// stage holds the runway hand-off until the dependency outcomes are +// actually confirmed (speculation.PathMergeConfirmed), so out-of-order delivery +// cannot land a dependent whose base never landed; and if a base's merge +// fails, mergesignal drives it terminal, the dependent fan-out re-wakes +// this controller, and superviseMerging fails every dependent whose bet +// died. // // No code in this file talks to a build runner. Every decision that stops a // build — a dead or deselected path during speculateBatch (applySelection @@ -111,7 +120,7 @@ import ( // path settles to Cancelled only once its build is observed terminal. // // The controller is re-triggered on every relevant downstream event -// (buildsignal, a prioritize tree write, merge, a dependency's own +// (buildsignal, a prioritize tree write, mergesignal, a dependency's own // speculate pass), so each call simply re-evaluates the current state and // either advances or waits. type Controller struct { @@ -207,10 +216,11 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r return c.fanout(ctx, batch.ID, batch.Queue) } - // Merging is owned by the merge controller, which has its own self-heal. + // Merging: the merge stage owns the batch's forward motion; speculate + // supervises — waking dependents and failing the batch if its + // optimistic bet died. See superviseMerging. if batch.State == entity.BatchStateMerging { - metrics.NamedCounter(c.metricsScope, opName, "noop_merging", 1) - return nil + return c.superviseMerging(ctx, batch) } switch batch.State { @@ -766,20 +776,21 @@ func (c *Controller) applySelection(batch entity.Batch, tree entity.SpeculationT // finalize settles the batch's forward step from the tree. Exactly one of // three outcomes applies per pass: // -// 1. Merge now: some path is mergeableNow (its build passed and its -// assumption set holds) — publish to merge, CAS the batch to Merging, -// and wake dependents, for whom the Merging batch already counts as -// landed (optimistic merge finalization; see mergeableNow). +// 1. Merge now: some path's merge assumptions hold — optimistically: its +// own build Passed and its base is landed, cancelled out of the way, or +// itself published for merge (speculation.PathMergePossible; see the +// Controller doc) — publish to merge, CAS the batch to Merging, and +// wake dependents, for whom this batch now counts as landed. // 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. +// merge. The !anyViable branch below runs the shared Failed sequence +// (failBatch): CAS to Failed, dependent fan-out, 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) { + if !speculation.PathMergePossible(p, depByID) { continue } @@ -793,8 +804,16 @@ func (c *Controller) finalize(ctx context.Context, batch entity.Batch, tree enti metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) return fmt.Errorf("failed to update batch %s state to merging: %w", batch.ID, err) } + batch.Version = newVersion + batch.State = entity.BatchStateMerging metrics.NamedCounter(c.metricsScope, opName, "merging", 1) - return nil + + // Wake dependents now: a Merging batch already counts as landed for + // their merge readiness. Fan out AFTER the CAS so a woken dependent + // reads this batch as Merging; a crash in between is healed by + // superviseMerging, which re-runs the fan-out on every later + // delivery. + return c.respeculateDependents(ctx, batch) } anyViable := false @@ -818,34 +837,7 @@ func (c *Controller) finalize(ctx context.Context, batch entity.Batch, tree enti "cancelled_paths", cancelledCount, ) metrics.NamedCounter(c.metricsScope, opName, "no_viable_path", 1) - - 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 - } - - 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 + return c.failBatch(ctx, batch) } metrics.NamedCounter(c.metricsScope, opName, "waiting_on_deps", 1) @@ -853,43 +845,98 @@ func (c *Controller) finalize(ctx context.Context, batch entity.Batch, tree enti return nil } -// 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. +// failBatch runs the shared terminal-Failed sequence for a batch that can +// never merge: CAS to Failed, wake dependents, publish to conclude. Callers +// log and count their own reason before calling. // -// A base dependency in Merging deliberately does NOT count as landed: the -// queue can deliver messages out of order when one is retrying, so merging -// behind an unconfirmed base could land changes validated on a base that -// never landed. See the Merging note on the Controller doc. -func mergeableNow(p entity.SpeculationPathInfo, depByID map[string]entity.Batch) bool { - if p.Status != entity.SpeculationPathStatusPassed { - return false +// The fan-out runs AFTER the CAS and BEFORE concluding, mirroring +// cancelBatch's ordering. It 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. +func (c *Controller) failBatch(ctx context.Context, batch entity.Batch) error { + 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 - 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.respeculateDependents(ctx, batch); err != nil { + return err } - for id, d := range depByID { - if inBase[id] { - continue - } - if d.State != entity.BatchStateFailed && 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) + } + return nil +} + +// superviseMerging handles a speculate wake for a batch already in Merging. +// The merge stage owns the batch's forward motion; this branch owns two +// things: +// +// - Wake dependents. A Merging batch already counts as landed for their +// merge readiness (optimistic merge finalization), and re-publishing +// the fan-out on every delivery self-heals one lost to a crash right +// after the Merging CAS, exactly like the terminal branch. +// - Fail the batch if its optimistic bet died. A batch may finalize while +// a base dependency is itself still Merging; the merge stage only hands +// off to runway once some path's assumptions are confirmed. If the base +// failed instead, no hand-off will ever happen and nothing else would +// move this batch again — so when no path is possible anymore, run the +// shared Failed sequence. This never races a runway verdict: hand-off +// requires confirmed assumptions, confirmed outcomes are terminal +// dependency states that cannot un-happen, and possible ⊇ confirmed — +// so a handed-off batch always takes the wake branch here. +func (c *Controller) superviseMerging(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 + } + + // A Merging batch always has a tree — it finalized from one, and trees + // are never deleted — so a miss is corrupted state: the error rejects + // the message and the DLQ reconciler fails the batch loudly. + tree, err := c.store.GetSpeculationTreeStore().Get(ctx, batch.ID) + if err != nil { + metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) + return fmt.Errorf("failed to get speculation tree for batch %s: %w", batch.ID, err) + } + + for _, p := range tree.Paths { + if speculation.PathMergePossible(p, depByID) { + metrics.NamedCounter(c.metricsScope, opName, "merging_fanout", 1) + // Re-arm the merge trigger before waking dependents. The merge + // stage's delayed re-check cycle is the only live message for a + // waiting Merging batch; if that chain is ever lost, nothing + // else would re-create it, so every healthy supervision wake + // re-arms it. Duplicated triggers converge: a confirmed batch + // hands off idempotently (runway dedups on the correlation id) + // and halted batches ack the trigger away. + if err := c.republishMergeTrigger(ctx, batch); err != nil { + metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1) + return fmt.Errorf("failed to re-arm merge trigger for batch %s: %w", batch.ID, err) + } + return c.respeculateDependents(ctx, batch) } } - return true + + metrics.NamedCounter(c.metricsScope, opName, "merge_bet_refuted", 1) + c.logger.Warnw("no path can confirm for merging batch; failing batch", + "batch_id", batch.ID, + ) + return c.failBatch(ctx, batch) } // viable reports whether p could still merge in the future: it has not @@ -1142,6 +1189,34 @@ func (c *Controller) cancelPathStatus(ctx context.Context, batch entity.Batch, p return entity.SpeculationPathStatusCancelling, nil } +// republishMergeTrigger re-arms the batch's merge trigger with a freshly +// minted message ID. The minted ID matters: the queue dedups publishes on +// (topic, partition, id) against un-GC'd rows including consumed ones, so +// re-publishing under the batch ID — the ID finalize's original trigger +// used — could be silently swallowed and heal nothing. +func (c *Controller) republishMergeTrigger(ctx context.Context, batch entity.Batch) error { + payload, err := entity.BatchID{ID: batch.ID}.ToBytes() + if err != nil { + return fmt.Errorf("failed to serialize batch ID: %w", err) + } + + msgID := fmt.Sprintf("%s/mergeheal/%d", batch.ID, time.Now().UnixMilli()) + msg := entityqueue.NewMessage(msgID, payload, batch.Queue, nil) + + q, ok := c.registry.Queue(topickey.TopicKeyMerge) + if !ok { + return fmt.Errorf("no queue registered for topic key %s", topickey.TopicKeyMerge) + } + topicName, ok := c.registry.TopicName(topickey.TopicKeyMerge) + if !ok { + return fmt.Errorf("no topic name registered for topic key %s", topickey.TopicKeyMerge) + } + if err := q.Publisher().Publish(ctx, topicName, msg); err != nil { + return fmt.Errorf("failed to publish message: %w", err) + } + return nil +} + // respeculateDependents publishes a speculate event for every batch that // depends on the given batch. The batch controller creates a BatchDependent // row (with Dependents possibly empty) for every batch it persists, so a diff --git a/submitqueue/orchestrator/controller/speculate/speculate_test.go b/submitqueue/orchestrator/controller/speculate/speculate_test.go index db9bcef6..1bdf6a68 100644 --- a/submitqueue/orchestrator/controller/speculate/speculate_test.go +++ b/submitqueue/orchestrator/controller/speculate/speculate_test.go @@ -17,6 +17,7 @@ package speculate import ( "context" "fmt" + "strings" "testing" "github.com/stretchr/testify/assert" @@ -26,6 +27,7 @@ import ( "github.com/uber/submitqueue/platform/consumer" "github.com/uber/submitqueue/platform/errs" queuemock "github.com/uber/submitqueue/platform/extension/messagequeue/mock" + "github.com/uber/submitqueue/submitqueue/core/speculation" "github.com/uber/submitqueue/submitqueue/core/topickey" "github.com/uber/submitqueue/submitqueue/entity" dependencylimitfake "github.com/uber/submitqueue/submitqueue/extension/speculation/dependencylimit/fake" @@ -249,17 +251,34 @@ func TestController_Process_UnrecognizedState(t *testing.T) { require.Error(t, runProcess(t, ctrl, h.controller, batch.ID)) } -// Merging is owned by the merge controller — speculate is a no-op for it. -func TestController_Process_MergingNoOp(t *testing.T) { +// A Merging batch is supervised, not advanced: state moves only when the +// bet died (covered by TestController_Process_MergingSupervision). Here the +// bet is fine, so the only effect is the dependent wake-up. +func TestController_Process_MergingWakesDependentsOnly(t *testing.T) { ctrl := gomock.NewController(t) h := newTestHarness(t, ctrl) batch := testBatch(entity.BatchStateMerging) + // No dependencies and a Passed path: trivially still possible. + tree := entity.SpeculationTree{ + BatchID: batch.ID, + Version: 2, + Paths: []entity.SpeculationPathInfo{{ID: "test-queue/batch/1/path/0", Path: entity.SpeculationPath{Head: batch.ID}, Status: entity.SpeculationPathStatusPassed, BuildID: "build-1"}}, + } h.batchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil) - // No UpdateState, no tree access, no publish expected. + h.treeStore.EXPECT().Get(gomock.Any(), batch.ID).Return(tree, nil) + // No UpdateState expected — supervision never advances a healthy batch. + h.depStore.EXPECT().Get(gomock.Any(), batch.ID).Return(entity.BatchDependent{ + BatchID: batch.ID, + Dependents: []string{"test-queue/batch/2"}, + Version: 1, + }, nil) require.NoError(t, runProcess(t, ctrl, h.controller, batch.ID)) - assert.Empty(t, *h.records) + require.Len(t, *h.records, 2) + assert.Equal(t, "submitqueue-merge", (*h.records)[0].topic) + assert.True(t, strings.HasPrefix((*h.records)[0].msgID, batch.ID+"/mergeheal/")) + assert.Equal(t, pubRec{topic: "speculate", msgID: "test-queue/batch/2"}, (*h.records)[1]) } // Terminal states wake dependents (re-publish speculate for each) and then @@ -518,10 +537,10 @@ func TestController_Process_MergeGate(t *testing.T) { wantMerge bool }{ {name: "passed_and_base_dep_succeeded_merges", pathStatus: entity.SpeculationPathStatusPassed, depState: entity.BatchStateSucceeded, wantMerge: true}, - // A base dependency merely published for merge (Merging) must NOT - // count as landed: the queue reorders across nack backoff, so an - // optimistic dependent could overtake its base on the way to runway. - {name: "passed_and_base_dep_merging_waits", pathStatus: entity.SpeculationPathStatusPassed, depState: entity.BatchStateMerging, wantMerge: false}, + // Optimistic merge finalization: a base already published for merge + // counts as landed — the merge stage confirms outcomes before the + // runway hand-off, so finalizing early is safe. + {name: "passed_and_base_dep_merging_merges", pathStatus: entity.SpeculationPathStatusPassed, depState: entity.BatchStateMerging, 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}, } @@ -549,13 +568,23 @@ func TestController_Process_MergeGate(t *testing.T) { if tt.wantMerge { h.batchStore.EXPECT().UpdateState(gomock.Any(), batch.ID, int32(1), int32(2), entity.BatchStateMerging).Return(nil) + // Entering Merging wakes dependents right after the CAS — + // this batch now counts as landed for their merge readiness. + h.depStore.EXPECT().Get(gomock.Any(), batch.ID).Return(entity.BatchDependent{ + BatchID: batch.ID, + Dependents: []string{"test-queue/batch/9"}, + Version: 1, + }, 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}) + wantRecords = append(wantRecords, + pubRec{topic: "submitqueue-merge", msgID: batch.ID}, + pubRec{topic: "speculate", msgID: "test-queue/batch/9"}, + ) } assert.ElementsMatch(t, wantRecords, *h.records) }) @@ -583,6 +612,10 @@ func TestController_Process_CancelledBaseDepToleratedMerges(t *testing.T) { // 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) + 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.ElementsMatch(t, []pubRec{ @@ -591,6 +624,72 @@ func TestController_Process_CancelledBaseDepToleratedMerges(t *testing.T) { }, *h.records) } +// Merging: the merge stage owns forward motion; speculate supervises. While +// some path's assumptions are still possible (base Succeeded or itself +// Merging), each wake just re-publishes the dependent fan-out — the +// optimistic "this batch counts as landed" wake-up and its crash self-heal. +// Once no path can confirm anymore (the base failed before any runway +// hand-off), the batch is failed with the shared terminal sequence. +func TestController_Process_MergingSupervision(t *testing.T) { + tests := []struct { + name string + depState entity.BatchState + wantFail bool + }{ + {name: "base_merging_wakes_dependents", depState: entity.BatchStateMerging, wantFail: false}, + // Cancelling is transient — it settles to a state that confirms or + // refutes — so supervision waits rather than failing the batch. + {name: "base_cancelling_wakes_dependents", depState: entity.BatchStateCancelling, wantFail: false}, + {name: "base_succeeded_wakes_dependents", depState: entity.BatchStateSucceeded, wantFail: false}, + {name: "base_failed_fails_batch", depState: entity.BatchStateFailed, wantFail: true}, + } + 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.BatchStateMerging, dep.ID) + path := entity.SpeculationPath{Base: []string{dep.ID}, Head: batch.ID} + tree := entity.SpeculationTree{ + BatchID: batch.ID, + Version: 3, + 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) + h.batchStore.EXPECT().Get(gomock.Any(), dep.ID).Return(dep, nil) + h.treeStore.EXPECT().Get(gomock.Any(), batch.ID).Return(tree, nil) + h.depStore.EXPECT().Get(gomock.Any(), batch.ID).Return(entity.BatchDependent{ + BatchID: batch.ID, + Dependents: []string{"test-queue/batch/2"}, + Version: 1, + }, nil) + + if tt.wantFail { + h.batchStore.EXPECT().UpdateState(gomock.Any(), batch.ID, int32(1), int32(2), entity.BatchStateFailed).Return(nil) + } + + require.NoError(t, runProcess(t, ctrl, h.controller, batch.ID)) + + if tt.wantFail { + assert.Equal(t, []pubRec{ + {topic: "speculate", msgID: "test-queue/batch/2"}, + {topic: "conclude", msgID: batch.ID}, + }, *h.records) + return + } + // A healthy wake re-arms the merge trigger (minted ID — never + // the bare batch ID, which publish dedup could swallow) before + // waking dependents. + require.Len(t, *h.records, 2) + rearm := (*h.records)[0] + assert.Equal(t, "submitqueue-merge", rearm.topic) + assert.True(t, strings.HasPrefix(rearm.msgID, batch.ID+"/mergeheal/"), "re-arm must mint a fresh ID, got %q", rearm.msgID) + assert.Equal(t, pubRec{topic: "speculate", msgID: "test-queue/batch/2"}, (*h.records)[1]) + }) + } +} + // 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 @@ -1155,9 +1254,11 @@ func TestPathDead(t *testing.T) { } } -// 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) { +// speculation.PathMergePossible (finalize's optimistic gate) and viable are +// pure functions over a path and its dependencies; table-test the +// combinations not already covered end-to-end. core/speculation's own tests +// cover the Possible/Confirmed split exhaustively. +func TestPathMergePossibleAndViable(t *testing.T) { base := "q/batch/1" head := "q/batch/2" @@ -1186,13 +1287,12 @@ func TestMergeableNowAndViable(t *testing.T) { pathBaseIsBase: true, }, { - // A base dependency published for merge is NOT landed yet: the - // queue reorders across nack backoff, so merging now could - // overtake the base on the way to runway. - name: "passed_base_merging_waits", + // Optimistic: a base already published for merge counts — the + // merge stage confirms outcomes before the runway hand-off. + name: "passed_base_merging_possible", status: entity.SpeculationPathStatusPassed, deps: map[string]entity.Batch{base: {ID: base, State: entity.BatchStateMerging}}, - wantMergeable: false, + wantMergeable: true, wantViable: true, pathBaseIsBase: true, }, @@ -1233,7 +1333,7 @@ func TestMergeableNowAndViable(t *testing.T) { 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.wantMergeable, speculation.PathMergePossible(info, tt.deps)) assert.Equal(t, tt.wantViable, viable(info, tt.deps)) }) }