Merge branch 'main' into mnoah1/stovepipe-process-gate-defer

Author: mnoah1

CLAUDE.md

@@ -120,6 +120,7 @@ Domain objects live under each domain's `entity/` tree, or under `platform/base/
 4. Every field must have a comment
 5. Reference other entities by ID (string or int), not directly
 6. String enums with sentinel values (`""` for unknown)
+7. Docs describe the data, not the choreography — say what a type or field *is* and its invariants (immutability, uniqueness scope, units, valid range), never which controller/stage/seam reads or writes it. Ownership and write-path rules live with the code that owns them (controller, store, or extension docs). Lifecycle enums may define states in terms of pipeline stages where that *is* the state's meaning (e.g. "admitted under the build budget"), but must not name the components that perform transitions.
 
 ### Extensions
 

api/runway/messagequeue/proto/merge.proto

@@ -48,7 +48,7 @@ message MergeStep {
 // Runway echoes it back unchanged.
 message MergeRequest {
     option (uber.base.messagequeue.topic_keys) = "merge-conflict-check";
-    option (uber.base.messagequeue.topic_keys) = "merge";
+    option (uber.base.messagequeue.topic_keys) = "runway-merge";
 
     // id is the client-owned correlation id for this request (one per request).
     // Runway echoes it back on the result unchanged.

api/runway/messagequeue/protopb/merge.pb.go

@@ -34,7 +34,7 @@ const (
 	// TopicKeyMerge carries committing merge requests. A client publishes a
 	// MergeRequest here; Runway applies the steps, commits the result, and
 	// reports the revisions it produced.
-	TopicKeyMerge TopicKey = "merge"
+	TopicKeyMerge TopicKey = "runway-merge"
 	// TopicKeyMergeSignal carries committing merge results. Runway publishes a
 	// MergeResult here (with the produced revisions populated); the requesting
 	// client consumes it.

api/runway/messagequeue/topics.go

@@ -62,7 +62,10 @@ type topicGroup struct {
 }
 
 // NewTopicRegistry creates a new TopicRegistry from a list of TopicConfigs.
-// Returns an error if any topic name is invalid.
+// Returns an error if any topic name is invalid, or if two configs share a
+// topic key — a duplicate key would silently shadow the earlier entry (last
+// write wins on the key→queue/name maps), routing publishes and subscriptions
+// registered against one topic onto another.
 func NewTopicRegistry(configs []TopicConfig) (TopicRegistry, error) {
 	queues := make(map[TopicKey]extqueue.Queue, len(configs))
 	topicNames := make(map[TopicKey]string, len(configs))
@@ -73,6 +76,12 @@ func NewTopicRegistry(configs []TopicConfig) (TopicRegistry, error) {
 			return TopicRegistry{}, fmt.Errorf("invalid topic name for key %s: %w", cfg.Key, err)
 		}
 
+		if existing, ok := topicNames[cfg.Key]; ok {
+			return TopicRegistry{}, fmt.Errorf(
+				"duplicate topic key %s: already registered with name %q, cannot also register name %q",
+				cfg.Key, existing, cfg.Name)
+		}
+
 		queues[cfg.Key] = cfg.Queue
 		topicNames[cfg.Key] = cfg.Name
 

platform/consumer/registry.go

@@ -275,7 +275,7 @@ func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRe
 		},
 		{
 			Key:   runwaymq.TopicKeyMerge,
-			Name:  "merge",
+			Name:  "runway-merge",
 			Queue: q,
 			Subscription: extqueue.DefaultSubscriptionConfig(
 				subscriberName, "runway-merge",

service/runway/server/main.go

@@ -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/BUILD.bazel

@@ -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,9 +385,10 @@ 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"},
+		{topickey.TopicKeyMerge, "submitqueue-merge", "orchestrator-merge"},
 		{runwaymq.TopicKeyMergeSignal, "merge-signal", "orchestrator-mergesignal"},
 		{topickey.TopicKeyConclude, "conclude", "orchestrator-conclude"},
 	}
@@ -450,7 +456,7 @@ func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRe
 	// consumed primary topic above.
 	configs = append(configs, consumer.TopicConfig{
 		Key:   runwaymq.TopicKeyMerge,
-		Name:  "merge",
+		Name:  "runway-merge",
 		Queue: q,
 	})
 
@@ -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

service/submitqueue/orchestrator/server/main.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
@@ -42,7 +50,7 @@ const (
 	// PublishAfter when the build has not yet reached a terminal state.
 	TopicKeyBuildSignal TopicKey = "buildsignal"
 	// TopicKeyMerge is the pipeline stage where speculated batches are published for merging.
-	TopicKeyMerge TopicKey = "merge"
+	TopicKeyMerge TopicKey = "submitqueue-merge"
 	// TopicKeyConclude is the pipeline stage where merged requests are published for conclusion.
 	TopicKeyConclude TopicKey = "conclude"
 	// TopicKeyLog is the pipeline stage where per-request logs are written.