feat(orchestrator): compose per-queue speculators and turn speculation on

## Summary

### Why?

Everything below this commit landed inert: the orchestrator passed the speculate run a placeholder that proposes nothing, so no path was ever funded and no speculative build started. The machinery is all reviewed; nothing configures it per queue or switches it on.

### What?

This is the activation switch. Per-queue profiles gain a `Scorer` and a `Speculator`: each queue's speculator is composed from its own scorer as `standard.New(bestfirst.New(scorer), sticky.New(budget))`, with a factory adapter beside the existing ones so routing stays in the wiring layer, and `main.go` swaps the placeholder for `profiles.SpeculatorFactory()`.

The scorer profiles are the policy knobs: the baseline scores everything 0.5, test-queue buckets by lines changed, and e2e-test-queue exercises the composite scorer. Every scorer is wrapped by `scorerfake` so a change URI carrying a failure marker forces a scoring error end-to-end. The build budget is a wiring-level constant (4 concurrent builds per queue) with a TODO to move it onto `entity.QueueConfig`.

With this commit, paths are funded, builds run per path, and batches finalize from their paths — the whole stack goes live in one revertable step.

## Test Plan

✅ `bazel build //...`, `make fmt`, `make gazelle`

✅ `make e2e-test` — the full pipeline lands changes end-to-end with per-path speculation on.

Author: behinddwalls

service/submitqueue/orchestrator/server/BUILD.bazel

@@ -38,7 +38,14 @@ go_library(
         "//submitqueue/extension/conflict/fake:go_default_library",
         "//submitqueue/extension/conflict/fileoverlap:go_default_library",
         "//submitqueue/extension/conflict/none:go_default_library",
+        "//submitqueue/extension/scorer:go_default_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/allocator/sticky:go_default_library",
+        "//submitqueue/extension/speculation/generator/bestfirst:go_default_library",
         "//submitqueue/extension/speculation/speculator:go_default_library",
+        "//submitqueue/extension/speculation/speculator/standard:go_default_library",
         "//submitqueue/extension/storage/mysql:go_default_library",
         "//submitqueue/extension/validator/fake:go_default_library",
         "//submitqueue/orchestrator:go_default_library",

service/submitqueue/orchestrator/server/main.go

@@ -42,13 +42,11 @@ import (
 	"github.com/uber/submitqueue/platform/http"
 	"github.com/uber/submitqueue/platform/pipeline"
 	"github.com/uber/submitqueue/submitqueue/core/changeset"
-	"github.com/uber/submitqueue/submitqueue/entity"
 	"github.com/uber/submitqueue/submitqueue/extension/changeprovider"
 	cpfake "github.com/uber/submitqueue/submitqueue/extension/changeprovider/fake"
 	githubprovider "github.com/uber/submitqueue/submitqueue/extension/changeprovider/github"
 	phabprovider "github.com/uber/submitqueue/submitqueue/extension/changeprovider/phabricator"
 	routingprovider "github.com/uber/submitqueue/submitqueue/extension/changeprovider/routing"
-	"github.com/uber/submitqueue/submitqueue/extension/speculation/speculator"
 	mysqlstorage "github.com/uber/submitqueue/submitqueue/extension/storage/mysql"
 	validatorfake "github.com/uber/submitqueue/submitqueue/extension/validator/fake"
 	"github.com/uber/submitqueue/submitqueue/orchestrator"
@@ -198,12 +196,8 @@ func run() error {
 		BuildRunner:    profiles.BuildRunnerFactory(),
 		ChangeProvider: profiles.ChangeProviderFactory(),
 		Analyzer:       profiles.AnalyzerFactory(),
-		// Speculation is wired but inert: the placeholder below proposes
-		// nothing, so no path is ever funded and no speculative build starts.
-		// The wiring change at the top of this stack replaces it with real
-		// per-queue speculators composed from each profile's scorer.
-		Speculator: noopSpeculators{},
-		Validator:  validatorfake.NewFactory(),
+		Speculator:     profiles.SpeculatorFactory(),
+		Validator:      validatorfake.NewFactory(),
 	}
 
 	// Assemble the pipeline: one call builds the topic registry, creates
@@ -438,20 +432,3 @@ func parseTimeout(envVal string, defaultVal time.Duration) time.Duration {
 	}
 	return defaultVal
 }
-
-// noopSpeculators resolves every queue to a speculator that proposes nothing.
-// It keeps the speculate stage inert — no path funded, no build started —
-// until per-queue speculators are composed in the profiles.
-type noopSpeculators struct{}
-
-// For returns the propose-nothing speculator for any queue.
-func (noopSpeculators) For(speculator.Config) (speculator.Speculator, error) {
-	return noopSpeculator{}, nil
-}
-
-type noopSpeculator struct{}
-
-// Speculate proposes no actions, whatever the queue looks like.
-func (noopSpeculator) Speculate(context.Context, []entity.Batch, []entity.SpeculationPathSet) ([]entity.Speculation, error) {
-	return nil, nil
-}

service/submitqueue/orchestrator/server/profiles.go

@@ -15,10 +15,12 @@
 package main
 
 import (
+	"context"
 	"fmt"
 
 	"github.com/uber-go/tally"
 	"github.com/uber/submitqueue/submitqueue/core/changeset"
+	"github.com/uber/submitqueue/submitqueue/entity"
 	"github.com/uber/submitqueue/submitqueue/extension/buildrunner"
 	buildfake "github.com/uber/submitqueue/submitqueue/extension/buildrunner/fake"
 	"github.com/uber/submitqueue/submitqueue/extension/changeprovider"
@@ -27,6 +29,14 @@ import (
 	conflictfake "github.com/uber/submitqueue/submitqueue/extension/conflict/fake"
 	"github.com/uber/submitqueue/submitqueue/extension/conflict/fileoverlap"
 	"github.com/uber/submitqueue/submitqueue/extension/conflict/none"
+	"github.com/uber/submitqueue/submitqueue/extension/scorer"
+	"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"
+	"github.com/uber/submitqueue/submitqueue/extension/speculation/allocator/sticky"
+	"github.com/uber/submitqueue/submitqueue/extension/speculation/generator/bestfirst"
+	"github.com/uber/submitqueue/submitqueue/extension/speculation/speculator"
+	specstandard "github.com/uber/submitqueue/submitqueue/extension/speculation/speculator/standard"
 	"go.uber.org/zap"
 )
 
@@ -43,6 +53,15 @@ type Profile struct {
 
 	// Analyzer detects conflicts between concurrent batches in this queue.
 	Analyzer conflict.Analyzer
+
+	// Scorer holds this queue's scoring profile. There is no scoring stage: the
+	// scorer feeds the queue's speculator, which ranks candidate paths by how
+	// likely their assumptions are to hold.
+	Scorer scorer.Scorer
+
+	// Speculator decides which of this queue's speculation paths to build and
+	// which running ones to preempt, within the build budget.
+	Speculator speculator.Speculator
 }
 
 // Profiles maps a queue name to its extension Profile, falling back to a
@@ -86,6 +105,14 @@ func (p Profiles) AnalyzerFactory() conflict.Factory {
 	})
 }
 
+// SpeculatorFactory returns a speculator.Factory that resolves the Speculator
+// for each queue from the profile registry.
+func (p Profiles) SpeculatorFactory() speculator.Factory {
+	return speculatorFunc(func(c speculator.Config) (speculator.Speculator, error) {
+		return p.For(c.QueueName).Speculator, nil
+	})
+}
+
 // Thin func-type adapters — the http.HandlerFunc trick applied to each
 // extension Factory interface. Each func type satisfies the Factory contract,
 // letting Profiles cross the host/library boundary without dedicated structs.
@@ -104,37 +131,68 @@ type analyzerFunc func(conflict.Config) (conflict.Analyzer, error)
 
 func (f analyzerFunc) For(c conflict.Config) (conflict.Analyzer, error) { return f(c) }
 
+type speculatorFunc func(speculator.Config) (speculator.Speculator, error)
+
+func (f speculatorFunc) For(c speculator.Config) (speculator.Speculator, error) { return f(c) }
+
 // newProfiles 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 baseline and overrides
-// only the extensions that differ — here the conflict analyzer.
+// only the extensions that differ — here the conflict analyzer and the scorer.
 // Queues without an explicit profile fall back to the baseline.
 func newProfiles(logger *zap.Logger, scope tally.Scope, resolver changeset.Resolver) (Profiles, error) {
 	cp, err := newChangeProvider(logger, scope)
 	if err != nil {
 		return Profiles{}, fmt.Errorf("failed to create change provider: %w", err)
 	}
 
+	// batchLines buckets a batch by total lines changed across all its changes —
+	// larger batches are likelier to fail to land.
+	batchLines := func(_ context.Context, changes entity.BatchChanges) (int, error) {
+		return changes.TotalLinesChanged(), nil
+	}
+
 	// Baseline profile: shared edge integrations + a fake build runner (every
-	// build succeeds unless a head URI carries a failure marker). 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.
+	// 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.
 	//
-	// 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
+	// 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.
 	base := Profile{
 		ChangeProvider: cp,
 		BuildRunner:    buildfake.New(resolver),
 		// TODO: replace the delegate with a real analyzer (e.g. Tango target
 		// analysis). "all" serializes the queue conservatively.
 		Analyzer: conflictfake.New(all.New(), nil),
+		Scorer: scorerfake.New(resolver, heuristic.New(
+			resolver,
+			[]heuristic.Bucket{{Min: 0, Max: 1<<31 - 1, Score: 0.5}},
+			batchLines, scope.SubScope("scorer.default"),
+		)),
 	}
 
+	// test-queue: bucketed heuristic scorer; conservative (serialized) conflicts
+	// inherited from the baseline.
+	testQueue := base
+	testQueue.Scorer = scorerfake.New(resolver, heuristic.New(
+		resolver,
+		[]heuristic.Bucket{
+			{Min: 0, Max: 1, Score: 0.95},
+			{Min: 2, Max: 5, Score: 0.80},
+			{Min: 6, Max: 20, Score: 0.60},
+			{Min: 21, Max: 1<<31 - 1, Score: 0.40},
+		},
+		batchLines, scope.SubScope("scorer.test-queue"),
+	))
+
 	// e2e-conflict-error-queue: every conflict analysis fails, exercising the
-	// analyzer error path. Edge integrations inherit the baseline.
+	// analyzer error path. Scorer/edge integrations inherit the baseline.
 	conflictErrQueue := base
 	conflictErrQueue.Analyzer = conflictfake.New(all.New(), conflictfake.FailAlways)
 
@@ -143,16 +201,44 @@ func newProfiles(logger *zap.Logger, scope tally.Scope, resolver changeset.Resol
 	fileOverlapQueue := base
 	fileOverlapQueue.Analyzer = fileoverlap.New(resolver)
 
-	// e2e-test-queue: no conflicts (maximum parallelism).
+	// e2e-test-queue: composite scorer; no conflicts (maximum parallelism).
 	e2eQueue := base
 	e2eQueue.Analyzer = conflictfake.New(none.New(), nil)
+	e2eQueue.Scorer = scorerfake.New(resolver, composite.New(
+		map[string]scorer.Scorer{
+			"size": heuristic.New(resolver, []heuristic.Bucket{{Min: 0, Max: 1<<31 - 1, Score: 0.8}}, batchLines, scope),
+			"flat": heuristic.New(resolver, []heuristic.Bucket{{Min: 0, Max: 1<<31 - 1, Score: 0.6}}, batchLines, scope),
+		},
+		composite.Avg, scope.SubScope("scorer.e2e-test-queue"),
+	))
 
+	// The speculator is composed last, because it is built from whatever scorer
+	// the profile ended up with.
 	return Profiles{
-		defaultProfile: base,
+		defaultProfile: withSpeculator(base),
 		byQueue: map[string]Profile{
-			"e2e-test-queue":           e2eQueue,
-			"e2e-conflict-error-queue": conflictErrQueue,
-			"file-overlap-queue":       fileOverlapQueue,
+			"test-queue":               withSpeculator(testQueue),
+			"e2e-test-queue":           withSpeculator(e2eQueue),
+			"e2e-conflict-error-queue": withSpeculator(conflictErrQueue),
+			"file-overlap-queue":       withSpeculator(fileOverlapQueue),
 		},
 	}, nil
 }
+
+// defaultBuildBudget caps how many builds a queue may have occupying CI at
+// once. It is the only rationing lever the allocator has.
+//
+// TODO: move this onto entity.QueueConfig so operators can tune it per queue
+// without a code change. QueueConfig carries only the queue name today.
+const defaultBuildBudget = 4
+
+// withSpeculator returns the profile with its speculator composed from its own
+// scorer: bestfirst ranks a queue's candidate paths by how likely all their
+// assumptions are to hold, and sticky spends the build budget down that ranking
+// without preempting builds already running. Swapping either part changes the
+// policy without touching the speculate controller, which depends only on the
+// Speculator contract.
+func withSpeculator(p Profile) Profile {
+	p.Speculator = specstandard.New(bestfirst.New(p.Scorer), sticky.New(defaultBuildBudget))
+	return p
+}