feat(speculation): allocator contract and sticky impl

submitqueue/extension/speculation/allocator/BUILD.bazel

@@ -0,0 +1,12 @@
+load("@rules_go//go:def.bzl", "go_library")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["allocator.go"],
+    importpath = "github.com/uber/submitqueue/submitqueue/extension/speculation/allocator",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//submitqueue/entity:go_default_library",
+        "//submitqueue/extension/speculation/generator:go_default_library",
+    ],
+)

submitqueue/extension/speculation/allocator/README.md

@@ -0,0 +1,15 @@
+# allocator
+
+The `allocator` package defines a composition point used by the `standard` `Speculator`: an `Allocator` spends the queue's build budget over a candidate iterator. Like `generator`, it is **not** a controller-facing extension — an alternate `Speculator` need not use or expose this split — so there is no `Config` or `Factory` here; an `Allocator` is chosen when the `standard` `Speculator` is constructed.
+
+`Allocate` pulls candidates from the iterator in best-first order and matches them against the queue's current path sets by path ID, so a pending or building path stays funded rather than starting a new attempt. The generator enumerates the whole path space, so a path whose build already finished comes back round as a candidate; suppressing it is the `Allocator`'s job — a terminal candidate is skipped, spending nothing. Pending, building, and cancelling paths all charge the budget (a cancelling build holds its slot until it reaches a terminal state); terminal paths charge nothing. It returns the build and cancel actions that spend the budget.
+
+Because cancellation is best-effort, an `Allocator` should not spend capacity it merely expects a cancel to release — a build cancelled to make room keeps charging the budget until its cancel reaches a terminal state, so the queue converges over successive runs rather than oversubscribing the hard cap on concurrent builds in a single pass.
+
+## `sticky`
+
+The `sticky` `Allocator` fills only free budget slots and leaves every in-flight build running. Against the budget it counts the paths that still hold a build slot — `pending`, `building`, and `cancelling` (a cancel is a request, and the build keeps its slot until it actually stops). It then proposes a build for each new candidate, in order, until the budget fills — skipping candidates already terminal in the path sets and candidates already funded, neither of which spends a slot; it never proposes a cancel.
+
+A stored path with no status at all is a data defect. It holds no slot, since a status that never turns terminal would pin one forever, and `sticky` logs a warning so it does not pass unnoticed.
+
+The trade-off: because `sticky` never preempts, a better candidate that arrives while the budget is full waits for a running build to finish rather than displacing it — it never discards work already started, but it also cannot react to a newly-attractive path mid-flight. Its policy counterpart is a preempting `Allocator` that cancels a low-value in-flight path to fund a better candidate, spending a cancel now to converge faster over the next run.

submitqueue/extension/speculation/allocator/allocator.go

@@ -0,0 +1,42 @@
+// 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 allocator defines the budget-spending composition point used by the
+// standard Speculator. An Allocator spends the queue's build budget over a
+// candidate iterator. It is not a controller-facing extension — an alternate
+// Speculator need not use or expose this split.
+package allocator
+
+//go:generate mockgen -source=allocator.go -destination=mock/allocator_mock.go -package=mock
+
+import (
+	"context"
+
+	"github.com/uber/submitqueue/submitqueue/entity"
+	"github.com/uber/submitqueue/submitqueue/extension/speculation/generator"
+)
+
+// Allocator decides how to spend the queue's build budget over the generator's
+// candidate stream, returning the build and cancel actions to take this run. How
+// it rations the budget across new candidates and paths already in flight — and
+// whether it preempts running builds — is the implementation's policy.
+type Allocator interface {
+	// Allocate returns the build and cancel actions to take. It reconciles the
+	// candidate stream against the queue's current path sets by path ID: a
+	// candidate matching an in-flight path stays funded rather than starting a
+	// new attempt, and one matching a path whose build already finished is
+	// skipped — the generator enumerates the whole space, so suppressing
+	// finished paths is the Allocator's job.
+	Allocate(ctx context.Context, pathSets []entity.SpeculationPathSet, iter generator.PathIterator) ([]entity.Speculation, error)
+}

submitqueue/extension/speculation/allocator/mock/BUILD.bazel

@@ -0,0 +1,13 @@
+load("@rules_go//go:def.bzl", "go_library")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["allocator_mock.go"],
+    importpath = "github.com/uber/submitqueue/submitqueue/extension/speculation/allocator/mock",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//submitqueue/entity:go_default_library",
+        "//submitqueue/extension/speculation/generator:go_default_library",
+        "@org_uber_go_mock//gomock:go_default_library",
+    ],
+)

submitqueue/extension/speculation/allocator/sticky/BUILD.bazel

@@ -0,0 +1,29 @@
+load("@rules_go//go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["sticky.go"],
+    importpath = "github.com/uber/submitqueue/submitqueue/extension/speculation/allocator/sticky",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//submitqueue/entity:go_default_library",
+        "//submitqueue/extension/speculation/allocator:go_default_library",
+        "//submitqueue/extension/speculation/generator:go_default_library",
+        "@org_uber_go_zap//:go_default_library",
+    ],
+)
+
+go_test(
+    name = "go_default_test",
+    srcs = ["sticky_test.go"],
+    embed = [":go_default_library"],
+    deps = [
+        "//submitqueue/entity:go_default_library",
+        "@com_github_stretchr_testify//assert:go_default_library",
+        "@com_github_stretchr_testify//require:go_default_library",
+        "@org_uber_go_zap//:go_default_library",
+        "@org_uber_go_zap//zapcore:go_default_library",
+        "@org_uber_go_zap//zaptest:go_default_library",
+        "@org_uber_go_zap//zaptest/observer:go_default_library",
+    ],
+)

submitqueue/extension/speculation/allocator/sticky/sticky.go

@@ -0,0 +1,118 @@
+// 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 sticky provides an allocator.Allocator that fills only free budget
+// slots and leaves every in-flight build running — it never preempts. Its policy
+// counterpart is a preempting allocator, which would cancel a low-value in-flight
+// path to fund a better candidate; sticky trades that responsiveness for never
+// discarding a build it has already started.
+package sticky
+
+import (
+	"context"
+
+	"go.uber.org/zap"
+
+	"github.com/uber/submitqueue/submitqueue/entity"
+	"github.com/uber/submitqueue/submitqueue/extension/speculation/allocator"
+	"github.com/uber/submitqueue/submitqueue/extension/speculation/generator"
+)
+
+// alloc is a sticky allocator.Allocator. budget is the queue's cap on concurrent
+// builds; pending, building, and cancelling paths all charge against it.
+type alloc struct {
+	logger *zap.SugaredLogger
+	budget int
+}
+
+// New returns a sticky allocator.Allocator with the given concurrent-build
+// budget.
+func New(logger *zap.SugaredLogger, budget int) allocator.Allocator {
+	return alloc{logger: logger.Named("sticky"), budget: budget}
+}
+
+// Allocate keeps every in-flight path funded and pulls candidates in order into
+// the remaining free budget, proposing a build for each new path until the
+// budget fills. It never proposes a cancel.
+func (a alloc) Allocate(ctx context.Context, pathSets []entity.SpeculationPathSet, iter generator.PathIterator) ([]entity.Speculation, error) {
+	// Paths already holding a build slot. This set does double duty: it is both
+	// the count of budget already spent and the guard against proposing a second
+	// build for something already running, so a status belongs here if either
+	// reason applies.
+	funded := make(map[string]bool)
+	// Paths whose build already finished. They hold no slot, but they must not
+	// be built again either.
+	terminal := make(map[string]bool)
+	for _, set := range pathSets {
+		for _, entry := range set.Paths {
+			if entry.Status == entity.SpeculationPathStatusUnknown {
+				// A stored path should always carry a real status, so this is a
+				// data defect. Skip it rather than counting it: a status that
+				// never turns terminal would pin a build slot forever. That leaves
+				// the path free to be proposed again, which is self-correcting,
+				// but the malformed entry is worth knowing about.
+				a.logger.Warnw("speculation path has no status",
+					"batch_id", set.BatchID, "path_id", entry.ID)
+				continue
+			}
+			if entry.Status.IsTerminal() {
+				terminal[entry.ID] = true
+				continue
+			}
+			// Anything else that has not finished is still holding a slot:
+			// pending and building obviously, and cancelling too, since the build
+			// keeps its slot until it actually stops.
+			funded[entry.ID] = true
+		}
+	}
+
+	free := a.budget - len(funded)
+
+	var out []entity.Speculation
+	proposed := make(map[string]bool)
+	for free > 0 {
+		c, ok, err := iter.Next(ctx)
+		if err != nil {
+			return nil, err
+		}
+		if !ok {
+			// ok=false means the generator has nothing left to offer: every path
+			// the queue could speculate on has been seen. Stopping here with
+			// budget still free is normal, not a failure — there is simply
+			// nothing else worth building right now.
+			break
+		}
+		id := c.Path.ID()
+		if terminal[id] {
+			// This path's build already ran. The generator enumerates the whole
+			// space with no knowledge of the path sets, so finished paths come
+			// back round as candidates; they need no slot and must not be rebuilt.
+			continue
+		}
+		if funded[id] || proposed[id] {
+			// funded: this path is already building. It needs no new action and
+			// is already charging the budget, so skip it without spending
+			// anything.
+			//
+			// proposed: a generator is not supposed to repeat a candidate, so
+			// this should never fire. It is a cheap guard against one that does,
+			// since a duplicate would otherwise be paid for twice.
+			continue
+		}
+		out = append(out, entity.Speculation{Path: c.Path, Action: entity.PathActionBuild})
+		proposed[id] = true
+		free--
+	}
+	return out, nil
+}