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)) }) }