feat(speculation): wire predictor into speculation pipeline

service/submitqueue/orchestrator/server/BUILD.bazel

@@ -52,6 +52,8 @@ go_library(
         "//submitqueue/extension/conflict/pathoverlap:go_default_library",
         "//submitqueue/extension/speculation/allocator/sticky:go_default_library",
         "//submitqueue/extension/speculation/generator/bestfirst:go_default_library",
+        "//submitqueue/extension/speculation/predictor:go_default_library",
+        "//submitqueue/extension/speculation/predictor/evidence:go_default_library",
         "//submitqueue/extension/speculation/scorer:go_default_library",
         "//submitqueue/extension/speculation/scorer/composite:go_default_library",
         "//submitqueue/extension/speculation/scorer/fake:go_default_library",
@@ -121,6 +123,7 @@ go_test(
         "//submitqueue/extension/buildrunner:go_default_library",
         "//submitqueue/extension/changeprovider:go_default_library",
         "//submitqueue/extension/conflict:go_default_library",
+        "//submitqueue/extension/speculation/predictor:go_default_library",
         "//submitqueue/extension/speculation/scorer:go_default_library",
         "//submitqueue/extension/speculation/speculator:go_default_library",
         "//submitqueue/extension/storage:go_default_library",

service/submitqueue/orchestrator/server/config.go

@@ -16,6 +16,8 @@ package main
 
 import (
 	"fmt"
+	"maps"
+	"math"
 	"os"
 	"time"
 
@@ -67,6 +69,22 @@ const (
 // Ways a composite scorer combines its components.
 const combineAvg = "avg"
 
+// Predictor types selectable from configuration.
+const predictorTypeEvidence = "evidence"
+
+// Evidence an evidence predictor prices, as named in configuration. The set is
+// closed: a factor under any other name would be applied to nothing and never
+// noticed.
+const (
+	factorPathPassed = "pathPassed"
+	factorPathFailed = "pathFailed"
+	factorMerging    = "merging"
+	factorCancelling = "cancelling"
+)
+
+// neutralFactor leaves the scorer's price untouched.
+const neutralFactor = 1.0
+
 // defaultBuildBudget is how many builds a queue may have occupying CI at once
 // when it states no budget of its own. Four is enough for speculation to be
 // visible — a queue that can only build one path never speculates — while
@@ -110,6 +128,7 @@ type namedQueueProfileConfig struct {
 	Analyzer       *analyzerConfig       `yaml:"analyzer"`
 	Scorer         *scorerConfig         `yaml:"scorer"`
 	Speculator     *speculatorConfig     `yaml:"speculator"`
+	Predictor      *predictorConfig      `yaml:"predictor"`
 }
 
 // queueProfileConfig is the full set of extensions a queue resolves to.
@@ -119,6 +138,7 @@ type queueProfileConfig struct {
 	Analyzer       analyzerConfig       `yaml:"analyzer"`
 	Scorer         scorerConfig         `yaml:"scorer"`
 	Speculator     speculatorConfig     `yaml:"speculator"`
+	Predictor      predictorConfig      `yaml:"predictor"`
 }
 
 // changeProviderConfig selects how change metadata is fetched. The github and
@@ -230,7 +250,7 @@ type bucketConfig struct {
 }
 
 // speculatorConfig tunes how much CI a queue's speculation may occupy. It has no
-// `type`: there is one speculator, composed from the queue's scorer, and what
+// `type`: there is one speculator, composed from the queue's predictor, and what
 // varies between queues is what it is allowed to spend.
 type speculatorConfig struct {
 	// BuildBudget caps how many builds this queue may have occupying CI at once,
@@ -239,6 +259,18 @@ type speculatorConfig struct {
 	BuildBudget int `yaml:"buildBudget"`
 }
 
+// predictorConfig tunes how a queue turns its scorer's price into the
+// probability the generator ranks on. The scorer being revised is the queue's
+// own, so it is not named again here.
+type predictorConfig struct {
+	Type string `yaml:"type"`
+	// Factors revise the scorer's price, one per piece of evidence and keyed by
+	// evidence name. An omitted key keeps the inherited value, or 1 if neither
+	// defaults nor the queue named it. An omitted predictor block inherits the
+	// whole default, so every factor stays 1 until someone sets one.
+	Factors map[string]float64 `yaml:"factors"`
+}
+
 // loadProfilesConfig reads and validates the profiles configuration at path.
 func loadProfilesConfig(path string) (profilesConfig, error) {
 	data, err := os.ReadFile(path)
@@ -300,6 +332,11 @@ func (c *profilesConfig) normalizeAndValidate() error {
 				return err
 			}
 		}
+		if q.Predictor != nil {
+			if err := q.Predictor.normalizeAndValidate(where); err != nil {
+				return err
+			}
+		}
 	}
 	return c.validateGitRepoPaths()
 }
@@ -358,9 +395,31 @@ func (c profilesConfig) resolve(q namedQueueProfileConfig) queueProfileConfig {
 	if q.Speculator != nil {
 		profile.Speculator = *q.Speculator
 	}
+	if q.Predictor != nil {
+		profile.Predictor = overlayPredictor(profile.Predictor, *q.Predictor)
+	}
 	return profile
 }
 
+// overlayPredictor keeps default factors the queue did not name. A present
+// predictor block is otherwise a normal extension override: type replaces when
+// set, and named factor keys win.
+func overlayPredictor(base, override predictorConfig) predictorConfig {
+	if override.Type != "" {
+		base.Type = override.Type
+	}
+	if len(override.Factors) == 0 {
+		return base
+	}
+	merged := maps.Clone(base.Factors)
+	if merged == nil {
+		merged = make(map[string]float64, len(override.Factors))
+	}
+	maps.Copy(merged, override.Factors)
+	base.Factors = merged
+	return base
+}
+
 func (p *queueProfileConfig) normalizeAndValidate(where string) error {
 	if err := p.ChangeProvider.normalizeAndValidate(where); err != nil {
 		return err
@@ -374,7 +433,10 @@ func (p *queueProfileConfig) normalizeAndValidate(where string) error {
 	if err := p.Scorer.normalizeAndValidate(where); err != nil {
 		return err
 	}
-	return p.Speculator.normalizeAndValidate(where)
+	if err := p.Speculator.normalizeAndValidate(where); err != nil {
+		return err
+	}
+	return p.Predictor.normalizeAndValidate(where)
 }
 
 func (c *changeProviderConfig) normalizeAndValidate(where string) error {
@@ -556,6 +618,31 @@ func (s *scorerConfig) normalizeAndValidate(where string) error {
 	return nil
 }
 
+// normalizeAndValidate applies defaults and rejects a predictor that could not
+// be built. An empty block is an evidence predictor with every factor neutral,
+// which prices a batch at exactly its scorer's price.
+func (p *predictorConfig) normalizeAndValidate(where string) error {
+	if p.Type == "" {
+		p.Type = predictorTypeEvidence
+	}
+	if p.Type != predictorTypeEvidence {
+		return fmt.Errorf("%s: unknown predictor type %q", where, p.Type)
+	}
+	for name, factor := range p.Factors {
+		switch name {
+		case factorPathPassed, factorPathFailed, factorMerging, factorCancelling:
+		default:
+			return fmt.Errorf("%s: unknown predictor factor %q", where, name)
+		}
+		// Zero would permanently pin matching batches to 0; negatives cannot
+		// represent either direction in the factor contract.
+		if !(factor > 0) || math.IsInf(factor, 0) {
+			return fmt.Errorf("%s: predictor factor %q is %v, must be finite and positive", where, name, factor)
+		}
+	}
+	return nil
+}
+
 func (s *speculatorConfig) normalizeAndValidate(where string) error {
 	// A negative budget is rejected rather than clamped: sticky would compute no
 	// free slots from it, so the queue would batch and then never build anything,

service/submitqueue/orchestrator/server/config_test.go

@@ -666,3 +666,65 @@ func TestLoadProfilesConfig_RejectsBadScorers(t *testing.T) {
 		})
 	}
 }
+
+func TestLoadProfilesConfig_RejectsBadPredictors(t *testing.T) {
+	tests := []struct {
+		name     string
+		contents string
+	}{
+		{name: "unknown predictor type", contents: "defaults:\n  predictor: {type: vibes}\n"},
+		{name: "unknown factor", contents: "defaults:\n  predictor:\n    factors: {pathPased: 2}\n"},
+		{name: "zero factor", contents: "defaults:\n  predictor:\n    factors: {merging: 0}\n"},
+		{name: "negative factor", contents: "defaults:\n  predictor:\n    factors: {pathFailed: -1}\n"},
+		{name: "infinite factor", contents: "defaults:\n  predictor:\n    factors: {pathPassed: .inf}\n"},
+		{name: "bad factor on a queue override", contents: "defaults: {}\nqueues:\n  - name: q\n    predictor:\n      factors: {merging: 0}\n"},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			_, err := loadProfilesConfig(writeProfiles(t, tt.contents))
+			require.Error(t, err)
+		})
+	}
+}
+
+// An omitted predictor block leaves the queue ranking on its scorer's price
+// alone, which is what every queue does until someone states a factor.
+func TestLoadProfilesConfig_DefaultsThePredictorToNeutral(t *testing.T) {
+	cfg, err := loadProfilesConfig(writeProfiles(t, "defaults: {}\nqueues:\n  - name: q\n"))
+	require.NoError(t, err)
+
+	assert.Equal(t, predictorTypeEvidence, cfg.Defaults.Predictor.Type)
+
+	factors := factorsFrom(cfg.resolve(cfg.Queues[0]).Predictor)
+	assert.Equal(t, neutralFactor, factors.PathPassed)
+	assert.Equal(t, neutralFactor, factors.PathFailed)
+	assert.Equal(t, neutralFactor, factors.Merging)
+	assert.Equal(t, neutralFactor, factors.Cancelling)
+}
+
+func TestLoadProfilesConfig_ReadsPredictorFactors(t *testing.T) {
+	cfg, err := loadProfilesConfig(writeProfiles(t,
+		"defaults:\n  predictor:\n    factors: {pathPassed: 10, pathFailed: 0.3, merging: 12, cancelling: 0.1}\n"))
+	require.NoError(t, err)
+
+	factors := factorsFrom(cfg.Defaults.Predictor)
+	assert.Equal(t, 10.0, factors.PathPassed)
+	assert.Equal(t, 0.3, factors.PathFailed)
+	assert.Equal(t, 12.0, factors.Merging)
+	assert.Equal(t, 0.1, factors.Cancelling)
+}
+
+func TestLoadProfilesConfig_QueuePredictorFactorsOverlayDefaults(t *testing.T) {
+	cfg, err := loadProfilesConfig(writeProfiles(t,
+		"defaults:\n  predictor:\n    factors: {pathPassed: 10, pathFailed: 0.3, merging: 12, cancelling: 0.1}\nqueues:\n  - name: q\n    predictor:\n      factors: {pathPassed: 4}\n"))
+	require.NoError(t, err)
+
+	factors := factorsFrom(cfg.resolve(cfg.Queues[0]).Predictor)
+	assert.Equal(t, 4.0, factors.PathPassed)
+	assert.Equal(t, 0.3, factors.PathFailed)
+	assert.Equal(t, 12.0, factors.Merging)
+	assert.Equal(t, 0.1, factors.Cancelling)
+
+	defaults := factorsFrom(cfg.Defaults.Predictor)
+	assert.Equal(t, 10.0, defaults.PathPassed)
+}

service/submitqueue/orchestrator/server/profiles.go

@@ -47,6 +47,8 @@ import (
 	"github.com/uber/submitqueue/submitqueue/extension/conflict/pathoverlap"
 	"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/predictor"
+	"github.com/uber/submitqueue/submitqueue/extension/speculation/predictor/evidence"
 	"github.com/uber/submitqueue/submitqueue/extension/speculation/scorer"
 	"github.com/uber/submitqueue/submitqueue/extension/speculation/scorer/composite"
 	scorerfake "github.com/uber/submitqueue/submitqueue/extension/speculation/scorer/fake"
@@ -80,6 +82,10 @@ type Profile struct {
 	// likely their assumptions are to hold.
 	Scorer scorer.Factory
 
+	// Predictor turns this queue's scorer price into the probability the
+	// generator ranks on, revising it with the batch's observed progress.
+	Predictor predictor.Factory
+
 	// Speculator decides which of this queue's speculation paths to build and
 	// which running ones to preempt, within the build budget.
 	Speculator speculator.Factory
@@ -142,6 +148,14 @@ func (p Profiles) ScorerFactory() scorer.Factory {
 	})
 }
 
+// PredictorFactory returns a predictor.Factory that resolves the
+// Predictor for each queue from the profile registry.
+func (p Profiles) PredictorFactory() predictor.Factory {
+	return predictorFunc(func(c predictor.Config) (predictor.Predictor, error) {
+		return p.For(c.QueueName).Predictor.For(c)
+	})
+}
+
 // StorageFactory returns a storage.Factory that routes each queue to its
 // profile's storage backend before binding the queue-scoped store aggregate.
 func (p Profiles) StorageFactory() storage.Factory {
@@ -176,6 +190,10 @@ type scorerFunc func(scorer.Config) (scorer.Scorer, error)
 
 func (f scorerFunc) For(c scorer.Config) (scorer.Scorer, error) { return f(c) }
 
+type predictorFunc func(predictor.Config) (predictor.Predictor, error)
+
+func (f predictorFunc) For(c predictor.Config) (predictor.Predictor, 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) }
@@ -265,33 +283,71 @@ func (b *profileBuilder) build(cfg queueProfileConfig, where string) (Profile, e
 	if err != nil {
 		return Profile{}, err
 	}
-	// The speculator is composed last, because it is built from whatever scorer
-	// the profile ended up with.
-	return withSpeculator(Profile{
+	// The predictor and the speculator are composed last, because each is built
+	// from what the profile ended up with one level below it.
+	return withSpeculator(withPredictor(Profile{
 		ChangeProvider: provider,
 		BuildRunner:    runner,
 		Analyzer:       analyzer,
 		Storage:        b.stores,
 		Scorer:         sc,
-	}, cfg.Speculator.BuildBudget), nil
+	}, cfg.Predictor, b.scope), cfg.Speculator.BuildBudget), nil
+}
+
+// withPredictor returns the profile with its predictor composed over its own
+// scorer: the scorer prices the batch's change, and the predictor revises that
+// price with what the batch's builds have done.
+//
+// The scorer is resolved lazily, at the queue the predictor itself was asked
+// for, so the queue's identity reaches one level down into the scorer too.
+func withPredictor(p Profile, cfg predictorConfig, scope tally.Scope) Profile {
+	p.Predictor = predictorFunc(func(c predictor.Config) (predictor.Predictor, error) {
+		sc, err := p.Scorer.For(scorer.Config{QueueName: c.QueueName})
+		if err != nil {
+			return nil, fmt.Errorf("failed to resolve scorer for queue %q: %w", c.QueueName, err)
+		}
+		return evidence.New(c, sc, factorsFrom(cfg), scope.SubScope("predictor"))
+	})
+	return p
+}
+
+// factorsFrom reads the configured factors onto the named fields the predictor
+// takes, leaving an unstated one neutral. Names are validated when the config
+// is loaded.
+func factorsFrom(cfg predictorConfig) evidence.Factors {
+	factors := evidence.AllOnes()
+	for name, factor := range cfg.Factors {
+		switch name {
+		case factorPathPassed:
+			factors.PathPassed = factor
+		case factorPathFailed:
+			factors.PathFailed = factor
+		case factorMerging:
+			factors.Merging = factor
+		case factorCancelling:
+			factors.Cancelling = factor
+		}
+	}
+	return factors
 }
 
 // withSpeculator returns the profile with its speculator composed from its own
-// scorer: bestfirst ranks a queue's candidate paths by how likely all their
+// predictor: bestfirst ranks a queue's candidate paths by how likely all their
 // assumptions are to hold, and sticky spends buildBudget 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.
 //
-// The scorer is resolved lazily, at the queue the speculator itself was asked
-// for, so the queue's identity reaches one level down into the scorer too.
+// The predictor is resolved lazily, at the queue the speculator itself was
+// asked for, so the queue's identity reaches down through the predictor to the
+// scorer under it.
 func withSpeculator(p Profile, buildBudget int) Profile {
 	p.Speculator = speculatorFunc(func(c speculator.Config) (speculator.Speculator, error) {
-		sc, err := p.Scorer.For(scorer.Config{QueueName: c.QueueName})
+		pred, err := p.Predictor.For(predictor.Config{QueueName: c.QueueName})
 		if err != nil {
-			return nil, fmt.Errorf("failed to resolve scorer for queue %q: %w", c.QueueName, err)
+			return nil, fmt.Errorf("failed to resolve predictor for queue %q: %w", c.QueueName, err)
 		}
-		return specstandard.New(c, bestfirst.New(sc), sticky.New(buildBudget)), nil
+		return specstandard.New(c, bestfirst.New(pred), sticky.New(buildBudget)), nil
 	})
 	return p
 }