feat(speculation): add prioritizer + prioritization-limit extensions

Makefile

@@ -364,7 +364,7 @@ local-stovepipe-stop: ## Stop the Stovepipe service
 
 mocks: ## Generate mock files using mockgen
 	@echo "Generating mocks..."
-	@$(BAZEL) run @rules_go//go -- generate ./submitqueue/extension/storage/... ./submitqueue/extension/buildrunner/... ./submitqueue/extension/changeprovider/... ./platform/extension/counter/... ./platform/extension/messagequeue/... ./submitqueue/extension/queueconfig/... ./submitqueue/extension/mergechecker/... ./submitqueue/extension/pusher/... ./submitqueue/extension/scorer/... ./submitqueue/extension/conflict/... ./submitqueue/extension/speculation/enumerator/... ./submitqueue/extension/speculation/dependencylimit/... ./submitqueue/extension/speculation/selector/... ./submitqueue/extension/speculation/selectionlimit/... ./submitqueue/extension/validator/... ./submitqueue/extension/speculation/pathscorer/... ./platform/consumer/... ./stovepipe/extension/storage/... ./stovepipe/extension/sourcecontrol/...
+	@$(BAZEL) run @rules_go//go -- generate ./submitqueue/extension/storage/... ./submitqueue/extension/buildrunner/... ./submitqueue/extension/changeprovider/... ./platform/extension/counter/... ./platform/extension/messagequeue/... ./submitqueue/extension/queueconfig/... ./submitqueue/extension/mergechecker/... ./submitqueue/extension/pusher/... ./submitqueue/extension/scorer/... ./submitqueue/extension/conflict/... ./submitqueue/extension/speculation/enumerator/... ./submitqueue/extension/speculation/dependencylimit/... ./submitqueue/extension/speculation/selector/... ./submitqueue/extension/speculation/selectionlimit/... ./submitqueue/extension/speculation/prioritizer/... ./submitqueue/extension/speculation/prioritizationlimit/... ./submitqueue/extension/validator/... ./submitqueue/extension/speculation/pathscorer/... ./platform/consumer/... ./stovepipe/extension/storage/... ./stovepipe/extension/sourcecontrol/...
 	@echo "Mocks generated successfully!"
 
 proto: ## Generate protobuf files from .proto definitions

submitqueue/extension/speculation/prioritizationlimit/BUILD.bazel

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

submitqueue/extension/speculation/prioritizationlimit/README.md

@@ -0,0 +1,17 @@
+# Prioritization Limit
+
+Vendor-agnostic "how much" policy that bounds how many builds a queue may run at once — the queue's concurrent-build budget.
+
+See the [Speculation RFC](/doc/rfc/submitqueue/speculation.md) for the end-to-end design and how limits fit into the two-layer speculation model.
+
+## Prioritization Limit
+
+The prioritization limit is the [prioritizer](../prioritizer)'s companion. The prioritizer decides **which** of the queue's pending builds run — its ranking across all in-flight batches; the prioritization limit decides **how many** fit at once. It is the queue-wide resource knob, the ultimate cap on speculation's demand on CI.
+
+The value is **signal-driven**, not a fixed constant. Its primary input is the build system's available capacity, but a policy may also weigh cost budgets, time of day, or an experiment toggle.
+
+It is **injected into the prioritizer** at construction and called by it, never passed as a method parameter — following the repo's extension-contract pattern, keeping the prioritizer interface limit-free and stable, and letting the limit be swapped independently of prioritizer logic.
+
+## Factory
+
+A per-queue factory returns the limit policy for a queue, following the repo's extension contract. It is handed only the queue identity; the signals a policy weighs — a capacity feed, cost budgets, config — are injected at construction by the integrator in the wiring layer, which is also where the limit is handed to the prioritizer. Computing the limit itself takes no further inputs.

submitqueue/extension/speculation/prioritizationlimit/fake/BUILD.bazel

@@ -0,0 +1,19 @@
+load("@rules_go//go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["fake.go"],
+    importpath = "github.com/uber/submitqueue/submitqueue/extension/speculation/prioritizationlimit/fake",
+    visibility = ["//visibility:public"],
+    deps = ["//submitqueue/extension/speculation/prioritizationlimit:go_default_library"],
+)
+
+go_test(
+    name = "go_default_test",
+    srcs = ["fake_test.go"],
+    embed = [":go_default_library"],
+    deps = [
+        "@com_github_stretchr_testify//assert:go_default_library",
+        "@com_github_stretchr_testify//require:go_default_library",
+    ],
+)

submitqueue/extension/speculation/prioritizationlimit/fake/fake.go

@@ -0,0 +1,53 @@
+// 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 fake provides a programmable prioritizationlimit.PrioritizationLimit
+// for tests and examples. New sets the value returned by Limit; FailWith injects
+// an error on every call. It is intended for examples and tests only, never
+// production.
+package fake
+
+import (
+	"context"
+
+	"github.com/uber/submitqueue/submitqueue/extension/speculation/prioritizationlimit"
+)
+
+// PrioritizationLimit is a programmable prioritizationlimit.PrioritizationLimit.
+type PrioritizationLimit struct {
+	limit int
+	err   error
+}
+
+// New returns a fake PrioritizationLimit whose Limit returns the given value.
+func New(value int) *PrioritizationLimit {
+	return &PrioritizationLimit{limit: value}
+}
+
+// FailWith makes every Limit call return err.
+func (l *PrioritizationLimit) FailWith(err error) *PrioritizationLimit {
+	l.err = err
+	return l
+}
+
+// Limit returns the configured value, or the injected error if FailWith was set.
+func (l *PrioritizationLimit) Limit(_ context.Context) (int, error) {
+	if l.err != nil {
+		return 0, l.err
+	}
+	return l.limit, nil
+}
+
+// ensure the fake satisfies the interface.
+var _ prioritizationlimit.PrioritizationLimit = (*PrioritizationLimit)(nil)

submitqueue/extension/speculation/prioritizationlimit/fake/fake_test.go

@@ -0,0 +1,36 @@
+// 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 fake
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestLimit_ReturnsConfiguredValue(t *testing.T) {
+	got, err := New(8).Limit(context.Background())
+	require.NoError(t, err)
+	assert.Equal(t, 8, got)
+}
+
+func TestLimit_FailWith(t *testing.T) {
+	sentinel := errors.New("boom")
+	_, err := New(8).FailWith(sentinel).Limit(context.Background())
+	require.ErrorIs(t, err, sentinel)
+}

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

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

submitqueue/extension/speculation/prioritizationlimit/prioritizationlimit.go

@@ -0,0 +1,57 @@
+// 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 prioritizationlimit
+
+//go:generate mockgen -source=prioritizationlimit.go -destination=mock/prioritizationlimit_mock.go -package=mock
+
+import "context"
+
+// PrioritizationLimit is the "how much" policy that bounds how many builds a
+// queue may run at once — the queue's concurrent-build budget.
+//
+// It is the prioritizer's companion: the prioritizer decides *which* of the
+// queue's pending builds run (its ranking across all in-flight batches); the
+// prioritization limit decides *how many* fit at once. It is the queue-wide
+// resource knob, the ultimate cap on speculation's demand on CI.
+//
+// The value is dynamic: it may change between calls, so the prioritizer reads it
+// each round rather than caching it.
+//
+// It is injected into the prioritizer at construction and called by it, never
+// passed as a method parameter, keeping the prioritizer interface limit-free and
+// stable.
+type PrioritizationLimit interface {
+	// Limit returns the current maximum number of concurrent builds for the
+	// queue. The prioritizer admits at most this many candidates. It takes no
+	// parameters; anything an implementation needs is injected at construction.
+	Limit(ctx context.Context) (int, error)
+}
+
+// Config carries the per-queue identity handed to a Factory. The system knows
+// only the queue name; everything a policy needs to compute the limit (a
+// capacity feed, cost budgets, config) is injected at construction by the
+// integrator.
+type Config struct {
+	// QueueName identifies the queue this PrioritizationLimit serves.
+	QueueName string
+}
+
+// Factory builds the PrioritizationLimit for a queue. Implementations are
+// provided by integrators (and tests) and inject whatever signals they need at
+// construction.
+type Factory interface {
+	// For returns the PrioritizationLimit for the given queue.
+	For(cfg Config) (PrioritizationLimit, error)
+}

submitqueue/extension/speculation/prioritizer/BUILD.bazel

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

submitqueue/extension/speculation/prioritizer/README.md

@@ -0,0 +1,21 @@
+# Speculation Prioritizer
+
+Vendor-agnostic interface for the queue-wide policy that rations a shared build budget across every in-flight batch in a queue.
+
+See the [Speculation RFC](/doc/rfc/submitqueue/speculation.md) for the end-to-end design and how prioritization fits into the orchestrator pipeline.
+
+## Prioritizer
+
+Selection is per batch and blind to other batches, so it cannot ration a shared budget: if every batch promoted generously, their combined demand could swamp CI. The prioritizer closes that gap. It sees every path across all of the queue's in-flight batches that is running or wants to run, ranks them by each path's score (plus any fairness or tie-break policy), and admits only the subset that fits the queue's concurrent-build budget. Selection expresses *desire* per batch; prioritization reconciles that desire against *supply* — it is the queue-wide enforcer.
+
+The controller hands the prioritizer the queue's candidate paths directly — every path that is `Selected` (wants a slot) or `Prioritized`/`Building` (holds a slot), each carrying its identity and score. It returns **sparse decisions**, each naming a path by its ID: `Promote` to admit a pending path, `Cancel` to preempt a running one. Paths it omits are left as-is, and it returns at most one decision per path. It never writes: the controller maps each decision back to its tree (it remembers which tree each candidate came from), maps it to a status transition (`Promote` → `Prioritized`, `Cancel` → `Cancelling`) applied under that tree's optimistic lock, and enacts it, staying the single writer.
+
+`Prioritized` means **admitted under the queue's build budget and cleared to build, but not yet building** — the path holds a slot and is waiting for the build stage to dispatch it and a build signal to confirm it (at which point it becomes `Building`). It is deliberately not called "admitted": batches are admitted at several points in the queue's lifecycle, while `Prioritized` names exactly the grant this seam makes. Contrast it with `Selected`, which is only the batch-level *desire* for a slot — a selected path has no claim on the budget until the prioritizer promotes it.
+
+**Whether to preempt is the prioritizer's own policy**, swappable per queue. A sticky-slots implementation never emits `Cancel` for a running path — it only fills free slots, and a higher-priority path waits until a slot frees. A preemptive implementation ranks running and pending paths together and may `Cancel` a running path to admit a higher-scored one. Both read the same input through the same interface; only the ranking/eviction logic differs. (Preemption discards in-flight CI work, so "fill free slots only" is a common default.)
+
+Prioritization is queue-wide — a different vantage point than the per-batch seams (enumerator, scorer, selector) — but it acts in the same currency: it ranks on the same path `Score` the scorer produces (a probability in [0, 1] under the [path scorer](../pathscorer)'s contract — the prioritizer needs no other assumption about the values, only that all paths in a queue share the scale) and emits the same `PathDecision` the selector does. Where the queue-wide reconcile runs in the pipeline is an integration detail; the contract here is unaffected by it.
+
+## Factory
+
+A per-queue factory returns the prioritizer for a queue, following the repo's extension contract. It is handed only the queue identity; the prioritization limit, fairness policy, and capacity signals are injected at construction by the integrator in the wiring layer, which resolves per-queue settings through `queueconfig`. Prioritization itself stays config-free.

submitqueue/extension/speculation/prioritizer/fake/BUILD.bazel

@@ -0,0 +1,23 @@
+load("@rules_go//go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["fake.go"],
+    importpath = "github.com/uber/submitqueue/submitqueue/extension/speculation/prioritizer/fake",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//submitqueue/entity:go_default_library",
+        "//submitqueue/extension/speculation/prioritizer:go_default_library",
+    ],
+)
+
+go_test(
+    name = "go_default_test",
+    srcs = ["fake_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",
+    ],
+)