feat(orchestrator): add queue-wide prioritize stage (#351)

## Summary
### Why?

Selection is per batch and blind to other batches, so it cannot ration a
shared build budget: if every batch promoted generously, their combined
demand could swamp CI. The speculation design closes that gap with a
prioritization step that sees every candidate path across a queue's
in-flight batches and admits only what fits the queue's concurrent-build
budget. That reconcile is queue-scoped, not batch-scoped, so it gets its
own pipeline stage rather than piggybacking on the per-batch speculate
flow.

### What?

New `prioritize` topic (payload: `entity.QueueID`, partitioned by queue
name) and controller. `entity.QueueID` is introduced here with
ToBytes/QueueIDFromBytes, mirroring the BatchID payload pattern for
queue-scoped stages. Each invocation is a full budget round for one
queue: load every Speculating batch's speculation tree (skipping batches
not yet speculated), flatten the queue-wide candidates (Selected /
Prioritized / Building paths), hand them to the queue's
`prioritizer.Prioritizer`, and apply the sparse decisions as captured
intent — Promote flips Selected→Prioritized; Cancel on a Building path
flips it to Cancelling (this controller never talks to the build system:
the build stage owns runner interaction and enacts the persisted intent,
retried on every build message until the build terminates); Cancel on a
not-yet-building Prioritized path drops it straight to Cancelled;
illegal decisions are logged and skipped so a policy bug cannot corrupt
tree state. Path lookup for decision application lives on the entity
(`SpeculationTree.PathIndex`), keyed by the path ID decisions carry; the
controller maps each ID back to its tree via the candidates it loaded
this round (never by parsing the ID), and treats a duplicate decision
for the same path as a policy bug — logged and skipped like any other
illegal decision. Each affected tree is persisted under its own
optimistic lock (version arithmetic in the controller); any conflict
nacks the round, which is safe to replay because decisions are
recomputed from freshly read state — a redelivered round carries no
memory of what a previous attempt picked and needs none, since
already-promoted paths are counted as slot holders rather than
re-promoted. After applying, the controller republishes to build for
every batch whose tree carries work the build stage still has to enact —
a Prioritized path with no build, or a Cancelling intent — healing
dropped build messages idempotently.

Wiring: topic registered with an automatic DLQ pair. The DLQ reconciler
re-arms the queue rather than terminalizing an entity: it logs the
failure and republishes a fresh prioritize round under a distinct,
deterministic message ID, so a queue whose only pending work is waiting
on prioritization is not stranded when a round dead-letters. The requeue
cannot poison-loop — a round carries only the queue name and recomputes
from live state — and a persistently failing round cycles
retry-ladder→DLQ→requeue at full-ladder cadence, visible in metrics,
converging once the fault heals. The sticky prioritizer (never preempts)
backed by a static admit-all parity limit is wired as the default
per-queue profile. Tree-entity docs are cleaned to describe the data
itself (states, invariants, uniqueness) rather than narrating which
stage reads or writes what. Nothing publishes to the topic yet — the
speculate rework turns it on.

## Test Plan
✅ `make gazelle && make fmt && make test` and `bazel build
//service/submitqueue/orchestrator/...`. Controller unit tests cover:
empty queue ack, missing-tree skip, promote transition + versioned
update + build republish, illegal-decision skip, cancel-on-building
capturing intent (Cancelling persisted, build republished, no runner
dependency at all), cancel-on-prioritized dropping straight to
Cancelled, version-mismatch and prioritizer errors nacking, and
republish for pre-existing prioritized paths with zero new decisions.
DLQ reconciler tests cover the requeued round's ID/partition/payload,
publish-failure nack, and malformed/empty payload rejection. Entity
tests cover `SpeculationTree.PathIndex` (order-sensitive base matching,
empty tree).

## Issues


## Stack
1. #350
1. @ #351
1. #353

Author: behinddwalls

service/submitqueue/orchestrator/server/BUILD.bazel

@@ -41,6 +41,9 @@ go_library(
         "//submitqueue/extension/scorer/composite:go_default_library",
         "//submitqueue/extension/scorer/fake:go_default_library",
         "//submitqueue/extension/scorer/heuristic:go_default_library",
+        "//submitqueue/extension/speculation/prioritizationlimit/static:go_default_library",
+        "//submitqueue/extension/speculation/prioritizer:go_default_library",
+        "//submitqueue/extension/speculation/prioritizer/sticky:go_default_library",
         "//submitqueue/extension/storage:go_default_library",
         "//submitqueue/extension/storage/mysql:go_default_library",
         "//submitqueue/extension/validator/fake:go_default_library",
@@ -54,6 +57,7 @@ go_library(
         "//submitqueue/orchestrator/controller/merge:go_default_library",
         "//submitqueue/orchestrator/controller/mergeconflictsignal:go_default_library",
         "//submitqueue/orchestrator/controller/mergesignal:go_default_library",
+        "//submitqueue/orchestrator/controller/prioritize:go_default_library",
         "//submitqueue/orchestrator/controller/score:go_default_library",
         "//submitqueue/orchestrator/controller/speculate:go_default_library",
         "//submitqueue/orchestrator/controller/start:go_default_library",

service/submitqueue/orchestrator/server/main.go

@@ -61,6 +61,9 @@ import (
 	"github.com/uber/submitqueue/submitqueue/extension/scorer/composite"
 	scorerfake "github.com/uber/submitqueue/submitqueue/extension/scorer/fake"
 	"github.com/uber/submitqueue/submitqueue/extension/scorer/heuristic"
+	prioritizationlimitstatic "github.com/uber/submitqueue/submitqueue/extension/speculation/prioritizationlimit/static"
+	"github.com/uber/submitqueue/submitqueue/extension/speculation/prioritizer"
+	"github.com/uber/submitqueue/submitqueue/extension/speculation/prioritizer/sticky"
 	"github.com/uber/submitqueue/submitqueue/extension/storage"
 	mysqlstorage "github.com/uber/submitqueue/submitqueue/extension/storage/mysql"
 	validatorfake "github.com/uber/submitqueue/submitqueue/extension/validator/fake"
@@ -74,6 +77,7 @@ import (
 	"github.com/uber/submitqueue/submitqueue/orchestrator/controller/merge"
 	"github.com/uber/submitqueue/submitqueue/orchestrator/controller/mergeconflictsignal"
 	"github.com/uber/submitqueue/submitqueue/orchestrator/controller/mergesignal"
+	"github.com/uber/submitqueue/submitqueue/orchestrator/controller/prioritize"
 	"github.com/uber/submitqueue/submitqueue/orchestrator/controller/score"
 	"github.com/uber/submitqueue/submitqueue/orchestrator/controller/speculate"
 	"github.com/uber/submitqueue/submitqueue/orchestrator/controller/start"
@@ -244,13 +248,14 @@ func run() error {
 	brf := buildRunnerFactory{queues}
 	scf := scorerFactory{queues}
 	cof := analyzerFactory{queues}
+	prf := prioritizerFactory{queues}
 
 	// Register controllers
-	primaryCount, err := registerPrimaryControllers(primaryConsumer, logger.Sugar(), scope, registry, cpf, brf, scf, cof, cnt, store)
+	primaryCount, err := registerPrimaryControllers(primaryConsumer, logger.Sugar(), scope, registry, cpf, brf, scf, cof, prf, cnt, store)
 	if err != nil {
 		return err
 	}
-	dlqCount, err := registerDLQControllers(dlqConsumer, logger.Sugar(), scope, store)
+	dlqCount, err := registerDLQControllers(dlqConsumer, logger.Sugar(), scope, registry, store)
 	if err != nil {
 		return err
 	}
@@ -380,6 +385,7 @@ func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRe
 		{topickey.TopicKeyBatch, "batch", "orchestrator-batch"},
 		{topickey.TopicKeyScore, "score", "orchestrator-score"},
 		{topickey.TopicKeySpeculate, "speculate", "orchestrator-speculate"},
+		{topickey.TopicKeyPrioritize, "prioritize", "orchestrator-prioritize"},
 		{topickey.TopicKeyBuild, "build", "orchestrator-build"},
 		{topickey.TopicKeyBuildSignal, "buildsignal", "orchestrator-buildsignal"},
 		{topickey.TopicKeyMerge, "merge", "orchestrator-merge"},
@@ -471,6 +477,13 @@ func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRe
 // merge-conflict-check queue (⇢); runway performs the merge attempt and
 // publishes the result to merge-conflict-check-signal, which mergeconflictsignal
 // consumes before fanning the request out to batch.
+//
+// prioritize sits alongside this per-batch flow rather than in its line: it
+// is queue-wide, not batch-scoped. Its message carries only a queue name; on
+// each invocation it loads every Speculating batch's speculation tree for
+// that queue, ranks the queue-wide candidate paths against the queue's build
+// budget, applies the resulting decisions, and republishes to build for any
+// path newly (or still) cleared to run.
 
 // TODO(wiring abstraction): queueExtensions + queueRegistry currently live here
 // as example-local wiring. Evaluate promoting them into a defined abstraction in
@@ -493,6 +506,7 @@ type queueExtensions struct {
 	buildRunner    buildrunner.BuildRunner
 	scorer         scorer.Scorer
 	analyzer       conflict.Analyzer
+	prioritizer    prioritizer.Prioritizer
 }
 
 // queueRegistry maps a queue name to its extensions, falling back to a default
@@ -538,7 +552,13 @@ func (f analyzerFactory) For(cfg conflict.Config) (conflict.Analyzer, error) {
 	return f.reg.get(cfg.QueueName).analyzer, nil
 }
 
-func registerPrimaryControllers(c consumer.Consumer, logger *zap.SugaredLogger, scope tally.Scope, registry consumer.TopicRegistry, cpf changeprovider.Factory, brf buildrunner.Factory, scf scorer.Factory, cof conflict.Factory, cnt counter.Counter, store storage.Storage) (int, error) {
+type prioritizerFactory struct{ reg queueRegistry }
+
+func (f prioritizerFactory) For(cfg prioritizer.Config) (prioritizer.Prioritizer, error) {
+	return f.reg.get(cfg.QueueName).prioritizer, nil
+}
+
+func registerPrimaryControllers(c consumer.Consumer, logger *zap.SugaredLogger, scope tally.Scope, registry consumer.TopicRegistry, cpf changeprovider.Factory, brf buildrunner.Factory, scf scorer.Factory, cof conflict.Factory, prf prioritizer.Factory, cnt counter.Counter, store storage.Storage) (int, error) {
 	var count int
 	requestController := start.NewController(
 		logger,
@@ -637,6 +657,20 @@ func registerPrimaryControllers(c consumer.Consumer, logger *zap.SugaredLogger,
 	}
 	count++
 
+	prioritizeController := prioritize.NewController(
+		logger,
+		scope,
+		store,
+		prf,
+		registry,
+		topickey.TopicKeyPrioritize,
+		"orchestrator-prioritize",
+	)
+	if err := c.Register(prioritizeController); err != nil {
+		return count, fmt.Errorf("failed to register prioritize controller: %w", err)
+	}
+	count++
+
 	buildController := build.NewController(
 		logger,
 		scope,
@@ -712,7 +746,7 @@ func registerPrimaryControllers(c consumer.Consumer, logger *zap.SugaredLogger,
 // registers them with the DLQ consumer. Each reconciler drives the affected
 // request or batch into a terminal Error/Failed state so the gateway stops
 // reporting it as stuck-in-progress.
-func registerDLQControllers(c consumer.Consumer, logger *zap.SugaredLogger, scope tally.Scope, store storage.Storage) (int, error) {
+func registerDLQControllers(c consumer.Consumer, logger *zap.SugaredLogger, scope tally.Scope, registry consumer.TopicRegistry, store storage.Storage) (int, error) {
 	dlqScope := scope.SubScope("dlq")
 	dlqRegs := []struct {
 		name string
@@ -725,6 +759,7 @@ func registerDLQControllers(c consumer.Consumer, logger *zap.SugaredLogger, scop
 		{"batch_dlq", dlq.NewDLQRequestController(logger, dlqScope, store, dlq.DecodeRequestID, dlq.TopicKey(topickey.TopicKeyBatch), "orchestrator-batch-dlq")},
 		{"score_dlq", dlq.NewDLQBatchController(logger, dlqScope, store, dlq.TopicKey(topickey.TopicKeyScore), "orchestrator-score-dlq")},
 		{"speculate_dlq", dlq.NewDLQBatchController(logger, dlqScope, store, dlq.TopicKey(topickey.TopicKeySpeculate), "orchestrator-speculate-dlq")},
+		{"prioritize_dlq", dlq.NewDLQQueueController(logger, dlqScope, registry, dlq.TopicKey(topickey.TopicKeyPrioritize), "orchestrator-prioritize-dlq")},
 		{"build_dlq", dlq.NewDLQBatchController(logger, dlqScope, store, dlq.TopicKey(topickey.TopicKeyBuild), "orchestrator-build-dlq")},
 		{"buildsignal_dlq", dlq.NewDLQBuildSignalController(logger, dlqScope, store, dlq.TopicKey(topickey.TopicKeyBuildSignal), "orchestrator-buildsignal-dlq")},
 		{"merge_dlq", dlq.NewDLQBatchController(logger, dlqScope, store, dlq.TopicKey(topickey.TopicKeyMerge), "orchestrator-merge-dlq")},
@@ -859,6 +894,11 @@ func newPhabChangeProvider(logger *zap.Logger, scope tally.Scope) (changeprovide
 	}), nil
 }
 
+// defaultPrioritizationLimit is the baseline queue-wide concurrent-build
+// budget handed to the sticky prioritizer. It is a parity default —
+// effectively admit-all — until per-queue budgets are configured.
+const defaultPrioritizationLimit = 1000
+
 // newQueueRegistry builds the per-queue extension profiles for the example.
 // Edge integrations (change provider) and the build
 // runner form a shared baseline; each per-queue profile starts from that
@@ -880,16 +920,19 @@ func newQueueRegistry(logger *zap.Logger, scope tally.Scope, resolver changeset.
 
 	// Baseline profile: shared edge integrations + a fake build runner (every
 	// build succeeds unless a head URI carries a failure marker), plus permissive
-	// defaults for scorer and conflict. The build runner instance is shared by
-	// the build and buildsignal controllers (same profile, same instance) so a
-	// build's recorded outcome survives across their separate factory lookups.
+	// defaults for scorer, conflict, and prioritization. The build runner
+	// instance is shared by the build and buildsignal controllers (same
+	// profile, same instance) so a build's recorded outcome survives across
+	// their separate factory lookups.
 	//
 	// The scorer is wrapped by scorerfake so a change URI carrying
 	// "sq-fake=score-error" forces a scoring error end-to-end; it is a pure
 	// passthrough otherwise. The analyzer is wrapped by conflictfake with a nil
 	// predicate (passthrough) — swap the predicate (e.g. conflictfake.FailAlways)
 	// on a queue to exercise the analyzer error path, as e2e-conflict-error-queue
-	// below does.
+	// below does. The prioritizer is sticky over a static budget: it never
+	// preempts a running build and admits Selected candidates by score until
+	// defaultPrioritizationLimit concurrent builds are in flight.
 	base := queueExtensions{
 		changeProvider: cp,
 		buildRunner:    buildfake.New(resolver),
@@ -900,7 +943,8 @@ func newQueueRegistry(logger *zap.Logger, scope tally.Scope, resolver changeset.
 		)),
 		// TODO: replace the delegate with a real analyzer (e.g. Tango target
 		// analysis). "all" serializes the queue conservatively.
-		analyzer: conflictfake.New(all.New(), nil),
+		analyzer:    conflictfake.New(all.New(), nil),
+		prioritizer: sticky.New(prioritizationlimitstatic.New(defaultPrioritizationLimit)),
 	}
 
 	// test-queue: bucketed heuristic scorer; conservative (serialized) conflicts

submitqueue/core/topickey/topickey.go

@@ -33,6 +33,14 @@ const (
 	TopicKeyScore TopicKey = "score"
 	// TopicKeySpeculate is the pipeline stage where scored batches are published for speculation.
 	TopicKeySpeculate TopicKey = "speculate"
+	// TopicKeyPrioritize is the queue-wide reconcile stage that rations the
+	// build budget across every in-flight batch of a queue. Each message
+	// carries a QueueID; the consumer loads every Speculating batch's tree,
+	// runs the queue's Prioritizer over the candidate paths, applies the
+	// resulting decisions — promoting paths into the build budget, or
+	// cancelling in-flight paths a preemptive policy evicts — and republishes
+	// to TopicKeyBuild for any path cleared to run.
+	TopicKeyPrioritize TopicKey = "prioritize"
 	// TopicKeyBuild is the pipeline stage where speculated batches are published for builds.
 	TopicKeyBuild TopicKey = "build"
 	// TopicKeyBuildSignal is the polling stage for triggered builds. Each

submitqueue/entity/BUILD.bazel

@@ -14,6 +14,7 @@ go_library(
         "land_request.go",
         "merge_result.go",
         "push_result.go",
+        "queue.go",
         "queue_config.go",
         "request.go",
         "request_log.go",
@@ -36,6 +37,7 @@ go_test(
         "build_test.go",
         "cancel_request_test.go",
         "land_request_test.go",
+        "queue_test.go",
         "request_log_test.go",
         "request_test.go",
         "speculation_tree_test.go",

submitqueue/entity/queue.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 entity
+
+import "encoding/json"
+
+// QueueID is the queue-message payload for queue-scoped pipeline stages. It
+// carries only the queue name.
+type QueueID struct {
+	// Name is the merge-queue name the message targets.
+	Name string `json:"name"`
+}
+
+// ToBytes serializes the QueueID to JSON bytes for queue message payload.
+func (q QueueID) ToBytes() ([]byte, error) {
+	return json.Marshal(q)
+}
+
+// QueueIDFromBytes deserializes a QueueID from JSON bytes.
+func QueueIDFromBytes(data []byte) (QueueID, error) {
+	var qid QueueID
+	err := json.Unmarshal(data, &qid)
+	return qid, err
+}

submitqueue/entity/queue_test.go

@@ -0,0 +1,72 @@
+// 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 entity
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestQueueID_SerializationRoundTrip(t *testing.T) {
+	tests := []struct {
+		name    string
+		queueID QueueID
+	}{
+		{
+			name:    "simple queue name",
+			queueID: QueueID{Name: "queueA"},
+		},
+		{
+			name:    "another queue name",
+			queueID: QueueID{Name: "queueB"},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			data, err := tt.queueID.ToBytes()
+			require.NoError(t, err)
+
+			deserialized, err := QueueIDFromBytes(data)
+			require.NoError(t, err)
+
+			assert.Equal(t, tt.queueID, deserialized)
+		})
+	}
+}
+
+func TestQueueIDFromBytes_InvalidJSON(t *testing.T) {
+	_, err := QueueIDFromBytes([]byte(`{"invalid": json"}`))
+	assert.Error(t, err)
+}
+
+func TestQueueIDFromBytes_EmptyJSON(t *testing.T) {
+	queueID, err := QueueIDFromBytes([]byte(`{}`))
+	require.NoError(t, err)
+
+	assert.Empty(t, queueID.Name)
+}
+
+func TestQueueIDFromBytes_EmptyBytes(t *testing.T) {
+	_, err := QueueIDFromBytes([]byte{})
+	assert.Error(t, err)
+}
+
+func TestQueueIDFromBytes_NilBytes(t *testing.T) {
+	_, err := QueueIDFromBytes(nil)
+	assert.Error(t, err)
+}