feat(speculation): optimistic merge finalization behind a confirmed hand-off

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.

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

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",
+    ],
+)

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

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

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",