From 56f89a1b928d7df9601f114f7b8d031954cd7ee6 Mon Sep 17 00:00:00 2001 From: Preetam Dwivedi Date: Mon, 7 Sep 2026 16:53:52 -0700 Subject: [PATCH 1/8] feat(speculation): wire predictor into speculation pipeline ## Summary ### Why? The predictor implementation is additive until the speculation pipeline supplies each dependency's run-local path evidence and uses the revised probability for ranking. ### What? Thread path sets through the Generator contract and standard Speculator, replace `bestfirst`'s scorer dependency with the predictor, and compose the evidence predictor from per-queue YAML configuration in orchestrator profiles. Neutral default factors preserve scorer-only ranking when no factors are configured. ## Test Plan - `bazel test //submitqueue/extension/speculation/generator/... //submitqueue/extension/speculation/speculator/... //service/submitqueue/orchestrator/server:go_default_test` - `make check-gazelle` # Conflicts: # submitqueue/extension/speculation/generator/bestfirst/bestfirst.go # submitqueue/extension/speculation/generator/bestfirst/bestfirst_test.go # submitqueue/extension/speculation/speculator/standard/standard_test.go # Please enter the commit message for your changes. Lines starting # with '#' will be kept; you may remove them yourself if you want to. # An empty message aborts the commit. # # interactive rebase in progress; onto 0ff6ea72 # Last command done (1 command done): # pick 15b2fa2e # feat(speculation): wire predictor into speculation pipeline # Next commands to do (4 remaining commands): # pick 60d6cc4a # docs(speculation): align predictor configuration terms # pick 7fa81830 # fix(speculation): enforce predictor wiring contract # You are currently rebasing branch 'preetam/outcome-predictor-wiring' on '0ff6ea72'. # # Changes to be committed: # modified: service/submitqueue/orchestrator/server/BUILD.bazel # modified: service/submitqueue/orchestrator/server/config.go # modified: service/submitqueue/orchestrator/server/config_test.go # modified: service/submitqueue/orchestrator/server/profiles.go # modified: service/submitqueue/orchestrator/server/profiles_test.go # modified: submitqueue/extension/speculation/generator/bestfirst/BUILD.bazel # modified: submitqueue/extension/speculation/generator/bestfirst/bestfirst.go # modified: submitqueue/extension/speculation/generator/bestfirst/bestfirst_test.go # modified: submitqueue/extension/speculation/generator/generator.go # modified: submitqueue/extension/speculation/generator/mock/generator_mock.go # modified: submitqueue/extension/speculation/speculator/standard/standard.go # modified: submitqueue/extension/speculation/speculator/standard/standard_test.go # --- .../orchestrator/server/BUILD.bazel | 3 + .../submitqueue/orchestrator/server/config.go | 69 ++++++- .../orchestrator/server/config_test.go | 46 +++++ .../orchestrator/server/profiles.go | 76 +++++++- .../orchestrator/server/profiles_test.go | 59 ++++-- .../generator/bestfirst/BUILD.bazel | 2 + .../generator/bestfirst/bestfirst.go | 41 +++-- .../generator/bestfirst/bestfirst_test.go | 173 ++++++++++++++---- .../speculation/generator/generator.go | 6 +- .../generator/mock/generator_mock.go | 8 +- .../speculator/standard/standard.go | 2 +- .../speculator/standard/standard_test.go | 11 +- 12 files changed, 406 insertions(+), 90 deletions(-) diff --git a/service/submitqueue/orchestrator/server/BUILD.bazel b/service/submitqueue/orchestrator/server/BUILD.bazel index 33d9b694b..fde019fcc 100644 --- a/service/submitqueue/orchestrator/server/BUILD.bazel +++ b/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", diff --git a/service/submitqueue/orchestrator/server/config.go b/service/submitqueue/orchestrator/server/config.go index 945374522..517c7c432 100644 --- a/service/submitqueue/orchestrator/server/config.go +++ b/service/submitqueue/orchestrator/server/config.go @@ -67,6 +67,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: odds multiplied by one. +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 +126,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 +136,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 @@ -239,6 +257,19 @@ 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 multiply the odds of the scorer's price, one per piece of + // evidence, keyed by evidence name. An omitted factor is neutral, so an + // omitted block ranks on the scorer's price alone. Values are hand-set + // placeholders, not measured: they are uncalibrated until the fitting work + // in doc/rfc/submitqueue/outcome-predictor.md lands. + 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 +331,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,6 +394,9 @@ func (c profilesConfig) resolve(q namedQueueProfileConfig) queueProfileConfig { if q.Speculator != nil { profile.Speculator = *q.Speculator } + if q.Predictor != nil { + profile.Predictor = *q.Predictor + } return profile } @@ -374,7 +413,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 +598,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 pin every batch carrying the evidence to a probability of + // zero, and a negative multiplier on odds means nothing at all. + if factor <= 0 { + return fmt.Errorf("%s: predictor factor %q is %v, must be 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, diff --git a/service/submitqueue/orchestrator/server/config_test.go b/service/submitqueue/orchestrator/server/config_test.go index 2de8278c5..ef1f3edbd 100644 --- a/service/submitqueue/orchestrator/server/config_test.go +++ b/service/submitqueue/orchestrator/server/config_test.go @@ -666,3 +666,49 @@ 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: "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) +} diff --git a/service/submitqueue/orchestrator/server/profiles.go b/service/submitqueue/orchestrator/server/profiles.go index 6a66e5110..deccb0c76 100644 --- a/service/submitqueue/orchestrator/server/profiles.go +++ b/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 } diff --git a/service/submitqueue/orchestrator/server/profiles_test.go b/service/submitqueue/orchestrator/server/profiles_test.go index 25a7fb7ec..ccf297018 100644 --- a/service/submitqueue/orchestrator/server/profiles_test.go +++ b/service/submitqueue/orchestrator/server/profiles_test.go @@ -15,15 +15,19 @@ package main import ( + "context" "errors" "testing" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "github.com/uber-go/tally" + "github.com/uber/submitqueue/submitqueue/entity" "github.com/uber/submitqueue/submitqueue/extension/buildrunner" "github.com/uber/submitqueue/submitqueue/extension/changeprovider" "github.com/uber/submitqueue/submitqueue/extension/conflict" + "github.com/uber/submitqueue/submitqueue/extension/speculation/predictor" "github.com/uber/submitqueue/submitqueue/extension/speculation/scorer" "github.com/uber/submitqueue/submitqueue/extension/speculation/speculator" "github.com/uber/submitqueue/submitqueue/extension/storage" @@ -37,8 +41,15 @@ type recorder struct { analyzer string storage string scorer string + predictor string } +// stubScorer stands in wherever a Profile's scorer has to be real rather than +// nil, because something is composed over it. +type stubScorer struct{} + +func (stubScorer) Score(_ context.Context, _ entity.Batch) (float64, error) { return 0.5, nil } + // profileRecording returns a Profile whose every factory records the queue name // it receives into rec and returns a nil implementation. Nil is fine: these // tests are about what reaches the factory, not what it builds. @@ -62,6 +73,10 @@ func profileRecording(rec *recorder) Profile { }), Scorer: scorerFunc(func(c scorer.Config) (scorer.Scorer, error) { rec.scorer = c.QueueName + return stubScorer{}, nil + }), + Predictor: predictorFunc(func(c predictor.Config) (predictor.Predictor, error) { + rec.predictor = c.QueueName return nil, nil }), } @@ -103,20 +118,23 @@ func TestProfilesForwardQueueNameToFactories(t *testing.T) { require.NoError(t, err) _, err = profiles.ScorerFactory().For(scorer.Config{QueueName: tt.queue}) require.NoError(t, err) + _, err = profiles.PredictorFactory().For(predictor.Config{QueueName: tt.queue}) + require.NoError(t, err) assert.Equal(t, tt.queue, rec.changeProvider) assert.Equal(t, tt.queue, rec.buildRunner) assert.Equal(t, tt.queue, rec.analyzer) assert.Equal(t, tt.queue, rec.storage) assert.Equal(t, tt.queue, rec.scorer) + assert.Equal(t, tt.queue, rec.predictor) }) } } -// TestWithSpeculatorResolvesScorerAtSameQueue covers the one seam that resolves -// another seam: the speculator is composed from the profile's scorer, and must -// ask for it at the queue it was itself asked for. -func TestWithSpeculatorResolvesScorerAtSameQueue(t *testing.T) { +// The speculator is composed from the profile's predictor, which is itself +// composed over the profile's scorer. Each has to be asked for at the queue the +// one above it was asked for, or an implementation is built for the wrong one. +func TestWithSpeculatorResolvesPredictorAtSameQueue(t *testing.T) { var rec recorder profile := withSpeculator(profileRecording(&rec), defaultBuildBudget) profiles := Profiles{defaultProfile: profile} @@ -124,20 +142,39 @@ func TestWithSpeculatorResolvesScorerAtSameQueue(t *testing.T) { spec, err := profiles.SpeculatorFactory().For(speculator.Config{QueueName: "unlisted-queue"}) require.NoError(t, err) assert.NotNil(t, spec) + assert.Equal(t, "unlisted-queue", rec.predictor) +} + +func TestWithPredictorResolvesScorerAtSameQueue(t *testing.T) { + var rec recorder + profile := withPredictor(profileRecording(&rec), predictorConfig{}, tally.NoopScope) + + pred, err := profile.Predictor.For(predictor.Config{QueueName: "unlisted-queue"}) + require.NoError(t, err) + assert.NotNil(t, pred) assert.Equal(t, "unlisted-queue", rec.scorer) } -// TestWithSpeculatorPropagatesScorerError covers the error path the factory -// conversion introduced: resolving the scorer can now fail where reading a -// struct field could not, and the failure must surface rather than yielding a -// speculator built over a nil scorer. -func TestWithSpeculatorPropagatesScorerError(t *testing.T) { - sentinel := errors.New("scorer unavailable") +// Resolving either level can fail where reading a struct field could not, and +// the failure must surface rather than yielding something built over a nil. +func TestWithSpeculatorPropagatesPredictorError(t *testing.T) { + sentinel := errors.New("predictor unavailable") profile := withSpeculator(Profile{ - Scorer: scorerFunc(func(scorer.Config) (scorer.Scorer, error) { return nil, sentinel }), + Predictor: predictorFunc(func(predictor.Config) (predictor.Predictor, error) { return nil, sentinel }), }, defaultBuildBudget) spec, err := profile.Speculator.For(speculator.Config{QueueName: "any-queue"}) require.ErrorIs(t, err, sentinel) assert.Nil(t, spec) } + +func TestWithPredictorPropagatesScorerError(t *testing.T) { + sentinel := errors.New("scorer unavailable") + profile := withPredictor(Profile{ + Scorer: scorerFunc(func(scorer.Config) (scorer.Scorer, error) { return nil, sentinel }), + }, predictorConfig{}, tally.NoopScope) + + pred, err := profile.Predictor.For(predictor.Config{QueueName: "any-queue"}) + require.ErrorIs(t, err, sentinel) + assert.Nil(t, pred) +} diff --git a/submitqueue/extension/speculation/generator/bestfirst/BUILD.bazel b/submitqueue/extension/speculation/generator/bestfirst/BUILD.bazel index 3fbed48cb..574e64a60 100644 --- a/submitqueue/extension/speculation/generator/bestfirst/BUILD.bazel +++ b/submitqueue/extension/speculation/generator/bestfirst/BUILD.bazel @@ -20,7 +20,9 @@ go_test( "//submitqueue/entity:go_default_library", "//submitqueue/extension/speculation/generator:go_default_library", "//submitqueue/extension/speculation/scorer:go_default_library", + "//submitqueue/extension/speculation/scorer/evidence:go_default_library", "@com_github_stretchr_testify//assert:go_default_library", "@com_github_stretchr_testify//require:go_default_library", + "@com_github_uber_go_tally//:go_default_library", ], ) diff --git a/submitqueue/extension/speculation/generator/bestfirst/bestfirst.go b/submitqueue/extension/speculation/generator/bestfirst/bestfirst.go index 43965d815..02e028324 100644 --- a/submitqueue/extension/speculation/generator/bestfirst/bestfirst.go +++ b/submitqueue/extension/speculation/generator/bestfirst/bestfirst.go @@ -14,8 +14,8 @@ // Package bestfirst provides a probability-ordered speculation path generator. // -// Throughout, a probability is the [0, 1] value a scorer gives; a score is its -// logarithm. Scores are summed and compared, never exponentiated, so wide +// Throughout, a probability is the [0, 1] value a scorer gives; a score is +// its logarithm. Scores are summed and compared, never exponentiated, so wide // heads cannot underflow into ties. The algorithm — per-head streams // enumerating flip subsets lazily, merged through one global heap — is // documented in doc/rfc/submitqueue/speculation-generator-best-first.md. @@ -49,10 +49,10 @@ func New(s scorer.Scorer) generator.Generator { return &bestFirst{scorer: s} } -// Generate scores the unresolved dependencies of the snapshot's Speculating +// Generate prices the unresolved dependencies of the snapshot's Speculating // heads and opens a lazy global best-first iterator. The snapshot is taken as // given: it is the caller's to keep well formed, and nothing here re-checks it. -func (g *bestFirst) Generate(ctx context.Context, batches []entity.Batch) (generator.Iterator, error) { +func (g *bestFirst) Generate(ctx context.Context, batches []entity.Batch, pathSets []entity.SpeculationPathSet) (generator.Iterator, error) { if err := ctx.Err(); err != nil { return nil, err } @@ -61,9 +61,13 @@ func (g *bestFirst) Generate(ctx context.Context, batches []entity.Batch) (gener for _, batch := range batches { batchByID[batch.ID] = batch } + pathsByHead := make(map[string]entity.SpeculationPathSet, len(pathSets)) + for _, set := range pathSets { + pathsByHead[set.Head] = set + } heads, unresolvedIDs := speculatingHeads(batches, batchByID) - probabilityByID, err := g.score(ctx, unresolvedIDs, batchByID) + probabilityByID, err := g.score(ctx, unresolvedIDs, batchByID, pathsByHead) if err != nil { return nil, err } @@ -101,13 +105,14 @@ func speculatingHeads(batches []entity.Batch, batchByID map[string]entity.Batch) return heads, slices.Sorted(maps.Keys(unresolved)) } -// score asks the scorer for each unresolved dependency exactly once, however -// many heads wait on it. +// score asks the scorer for each unresolved dependency exactly once, +// however many heads wait on it. Each dependency is priced against its own path +// set, zero-valued for one that has never speculated. // // A dependency that cannot be priced takes defaultProbability rather than // ending the run — one unusable number must not cost the queue every candidate // it had. Only cancellation is an error. -func (g *bestFirst) score(ctx context.Context, ids []string, batchByID map[string]entity.Batch) (map[string]float64, error) { +func (g *bestFirst) score(ctx context.Context, ids []string, batchByID map[string]entity.Batch, pathsByHead map[string]entity.SpeculationPathSet) (map[string]float64, error) { probabilityByID := make(map[string]float64, len(ids)) for _, id := range ids { if err := ctx.Err(); err != nil { @@ -116,16 +121,16 @@ func (g *bestFirst) score(ctx context.Context, ids []string, batchByID map[strin batch, known := batchByID[id] if !known { // A batch the snapshot never carried is zero in every field, not - // just missing — scoring it would price some other batch entirely, + // just missing — pricing it would price some other batch entirely, // or fail on its empty queue. It is unpriceable, not cheap. probabilityByID[id] = defaultProbability continue } - probability, err := g.scorer.Score(ctx, batch, entity.SpeculationPathSet{}) + probability, err := g.scorer.Score(ctx, batch, pathsByHead[id]) if err != nil { - // A scorer that failed because the caller went away has not found - // an unpriceable dependency — it has found a dead ctx, which ends - // the run. The loop's own check would not catch it on the last + // A scorer that failed because the caller went away has not + // found an unpriceable dependency — it has found a dead ctx, which + // ends the run. The loop's own check would not catch it on the last // dependency, and a cancelled Generate must never hand back an // iterator. if ctxErr := ctx.Err(); ctxErr != nil { @@ -139,11 +144,11 @@ func (g *bestFirst) score(ctx context.Context, ids []string, batchByID map[strin } // defaultProbability stands in for a score that is not a probability, one the -// scorer could not produce at all, and one for a dependency the snapshot never -// carried. It is optimistic on purpose: a dependency nobody could estimate is -// treated as very likely to succeed, which keeps its head's preferred path near -// the front rather than burying it or dropping the queue's whole snapshot on -// one bad number. +// scorer could not produce at all, and one for a dependency the snapshot +// never carried. It is optimistic on purpose: a dependency nobody could +// estimate is treated as very likely to succeed, which keeps its head's preferred +// path near the front rather than burying it or dropping the queue's whole +// snapshot on one bad number. const defaultProbability = 0.95 // asProbability keeps a usable score and substitutes the default for anything diff --git a/submitqueue/extension/speculation/generator/bestfirst/bestfirst_test.go b/submitqueue/extension/speculation/generator/bestfirst/bestfirst_test.go index 9b92d23c5..567fad0d3 100644 --- a/submitqueue/extension/speculation/generator/bestfirst/bestfirst_test.go +++ b/submitqueue/extension/speculation/generator/bestfirst/bestfirst_test.go @@ -26,14 +26,16 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "github.com/uber-go/tally" "github.com/uber/submitqueue/submitqueue/entity" "github.com/uber/submitqueue/submitqueue/extension/speculation/generator" "github.com/uber/submitqueue/submitqueue/extension/speculation/scorer" + "github.com/uber/submitqueue/submitqueue/extension/speculation/scorer/evidence" ) -// stubScorer scores each batch by ID, defaulting to 0.5 for unknown batches. It -// is a minimal scorer.Scorer for exercising the generator without a resolver. +// stubScorer scores each batch by ID, defaulting to 0.5 for unknown batches. +// It is a minimal scorer.Scorer for exercising the generator without a resolver. type stubScorer struct { scores map[string]float64 } @@ -125,7 +127,7 @@ func iteratorOf(t *testing.T, iter generator.Iterator) *candidateIterator { return it } -// countingScorer records how many times each batch is scored. +// countingScorer records how many times each batch is priced. type countingScorer struct { scores map[string]float64 calls map[string]int @@ -145,17 +147,19 @@ func (c *countingScorer) Score(_ context.Context, b entity.Batch, _ entity.Specu return 0.5, nil } -// errScorer always fails, to exercise error propagation from scoring. +// errScorer always fails, to exercise error propagation from pricing. type errScorer struct{} func (errScorer) Score(context.Context, entity.Batch, entity.SpeculationPathSet) (float64, error) { return 0, assert.AnError } -// constScorer scores every batch identically, regardless of ID. +// constScorer prices every batch identically, regardless of ID. type constScorer struct{ v float64 } -func (c constScorer) Score(context.Context, entity.Batch, entity.SpeculationPathSet) (float64, error) { return c.v, nil } +func (c constScorer) Score(context.Context, entity.Batch, entity.SpeculationPathSet) (float64, error) { + return c.v, nil +} // wideHead builds one Speculating head over n unresolved dependencies, each at a // distinct score so no two combinations tie. @@ -181,7 +185,7 @@ func TestBestFirst_OrderingAndEnumeration(t *testing.T) { } sc := scored(map[string]float64{"q/A": 0.9, "q/B": 0.8}) - iter, err := New(sc).Generate(context.Background(), batches) + iter, err := New(sc).Generate(context.Background(), batches, nil) require.NoError(t, err) cands := forHead(drainAll(t, iter), "q/C") @@ -222,7 +226,7 @@ func TestBestFirst_PinsResolvedDependencies(t *testing.T) { {ID: "q/A", State: tt.state}, {ID: "q/H", State: entity.BatchStateSpeculating, Dependencies: []string{"q/A"}}, } - iter, err := New(scored(nil)).Generate(context.Background(), batches) + iter, err := New(scored(nil)).Generate(context.Background(), batches, nil) require.NoError(t, err) cands := forHead(drainAll(t, iter), "q/H") @@ -245,7 +249,7 @@ func TestBestFirst_ResolvedDependenciesDropOutOfSearch(t *testing.T) { Dependencies: []string{"q/succeeded", "q/failed", "q/open"}}, } iter, err := New(scored(map[string]float64{"q/open": 0.7})). - Generate(context.Background(), batches) + Generate(context.Background(), batches, nil) require.NoError(t, err) cands := forHead(drainAll(t, iter), "q/H") @@ -269,7 +273,7 @@ func TestBestFirst_EmitsExactSequenceAcrossHeads(t *testing.T) { } sc := scored(map[string]float64{"q/A": 0.9, "q/B": 0.8}) - iter, err := New(sc).Generate(context.Background(), batches) + iter, err := New(sc).Generate(context.Background(), batches, nil) require.NoError(t, err) cands := drainAll(t, iter) @@ -304,7 +308,7 @@ func TestBestFirst_PreferredAssumptionFollowsScore(t *testing.T) { } sc := scored(map[string]float64{"q/high": 0.8, "q/low": 0.3}) - iter, err := New(sc).Generate(context.Background(), batches) + iter, err := New(sc).Generate(context.Background(), batches, nil) require.NoError(t, err) cands := drainAll(t, iter) @@ -352,7 +356,7 @@ func TestBestFirst_OnlySpeculatingHeadsProduceCandidates(t *testing.T) { t.Run(name, func(t *testing.T) { batches := []entity.Batch{{ID: "q/H", State: tt.state}} - iter, err := New(scored(nil)).Generate(context.Background(), batches) + iter, err := New(scored(nil)).Generate(context.Background(), batches, nil) require.NoError(t, err) cands := drainAll(t, iter) @@ -371,7 +375,7 @@ func TestBestFirst_HeadWithNoDependencies(t *testing.T) { {ID: "q/H", State: entity.BatchStateSpeculating}, } - iter, err := New(scored(nil)).Generate(context.Background(), batches) + iter, err := New(scored(nil)).Generate(context.Background(), batches, nil) require.NoError(t, err) cands := drainAll(t, iter) @@ -390,7 +394,7 @@ func TestBestFirst_AbsorbsScorerError(t *testing.T) { {ID: "q/H", State: entity.BatchStateSpeculating, Dependencies: []string{"q/A"}}, } - iter, err := New(errScorer{}).Generate(context.Background(), batches) + iter, err := New(errScorer{}).Generate(context.Background(), batches, nil) require.NoError(t, err) cands := forHead(drainAll(t, iter), "q/H") @@ -408,7 +412,7 @@ func TestBestFirst_NeverScoresAnAbsentDependency(t *testing.T) { } sc := newCountingScorer(map[string]float64{}) - iter, err := New(sc).Generate(context.Background(), batches) + iter, err := New(sc).Generate(context.Background(), batches, nil) require.NoError(t, err) cands := drainAll(t, iter) @@ -417,6 +421,95 @@ func TestBestFirst_NeverScoresAnAbsentDependency(t *testing.T) { assert.InDelta(t, math.Log(defaultProbability), cands[0].RankingScore, 1e-9) } +// recordingScorer keeps the path set each batch was priced against. +type recordingScorer struct { + seen map[string]entity.SpeculationPathSet +} + +func (r *recordingScorer) Score(_ context.Context, b entity.Batch, paths entity.SpeculationPathSet) (float64, error) { + r.seen[b.ID] = paths + return 0.5, nil +} + +// Each dependency is priced against its own progress, not the queue's. A +// dependency with no set has simply never speculated, which is silence rather +// than an error. +func TestBestFirst_PricesEachDependencyAgainstItsOwnPathSet(t *testing.T) { + batches := []entity.Batch{ + {ID: "q/built", State: entity.BatchStateSpeculating}, + {ID: "q/fresh", State: entity.BatchStateSpeculating}, + {ID: "q/H", State: entity.BatchStateSpeculating, Dependencies: []string{"q/built", "q/fresh"}}, + } + built := entity.SpeculationPathSet{ + Queue: "q", + Head: "q/built", + Paths: []entity.SpeculationPathEntry{{ID: "p1", Status: entity.SpeculationPathStatusPassed}}, + } + pred := &recordingScorer{seen: map[string]entity.SpeculationPathSet{}} + + _, err := New(pred).Generate(context.Background(), batches, []entity.SpeculationPathSet{built}) + require.NoError(t, err) + + assert.Equal(t, built, pred.seen["q/built"], "a dependency is priced against its own set") + assert.Equal(t, entity.SpeculationPathSet{}, pred.seen["q/fresh"], "a dependency that never speculated has no set") +} + +// flatScorer prices every batch the same, so ranking can only move when the +// evidence scorer sees a path set. +type flatScorer struct{} + +func (flatScorer) Score(context.Context, entity.Batch, entity.SpeculationPathSet) (float64, error) { return 0.5, nil } + +// PathPassed on a green all-succeed build is the join the generator exists to +// consume: same scorer price, different evidence, different rank. +func TestBestFirst_EvidencePathPassedRanksTheGreenDependencyFirst(t *testing.T) { + batches := []entity.Batch{ + {ID: "q/built", State: entity.BatchStateSpeculating}, + {ID: "q/fresh", State: entity.BatchStateSpeculating}, + {ID: "q/H", State: entity.BatchStateSpeculating, Dependencies: []string{"q/built", "q/fresh"}}, + } + built := entity.SpeculationPathSet{ + Queue: "q", + Head: "q/built", + Paths: []entity.SpeculationPathEntry{{ + ID: "p1", + Status: entity.SpeculationPathStatusPassed, + Path: entity.SpeculationPath{ + Head: "q/built", + Dependencies: []entity.PathDependency{{ + Batch: "q/dep0", + Assumption: entity.DependencyAssumptionSucceeds, + }}, + }, + }}, + } + pred, err := evidence.New( + scorer.Config{QueueName: "q"}, + flatScorer{}, + evidence.Factors{PathPassed: 9, PathFailed: 1, Merging: 1, Cancelling: 1}, + tally.NoopScope, + ) + require.NoError(t, err) + + iter, err := New(pred).Generate(context.Background(), batches, []entity.SpeculationPathSet{built}) + require.NoError(t, err) + cands := forHead(drainAll(t, iter), "q/H") + require.NotEmpty(t, cands) + assert.Equal(t, entity.DependencyAssumptionSucceeds, assumptionFor(cands[0].Path, "q/built")) + + var failScore float64 + foundFail := false + for _, c := range cands { + if assumptionFor(c.Path, "q/built") == entity.DependencyAssumptionFails { + failScore = c.RankingScore + foundFail = true + break + } + } + require.True(t, foundFail) + assert.Greater(t, cands[0].RankingScore, failScore) +} + // A merging dependency is still in progress — the merge can fail — so it stays // an open question here like any other. Whether a path betting against it is // worth funding is a matter of price, which is the scorer's to say, not a @@ -428,7 +521,7 @@ func TestBestFirst_MergingDependencyStaysOpen(t *testing.T) { } sc := newCountingScorer(map[string]float64{"q/landing": 0.9}) - iter, err := New(sc).Generate(context.Background(), batches) + iter, err := New(sc).Generate(context.Background(), batches, nil) require.NoError(t, err) cands := drainAll(t, iter) @@ -443,7 +536,7 @@ func TestBestFirst_GeneratesOnlyWhatIsPulled(t *testing.T) { const deps, space = 12, 1 << 12 batches, sc := wideHead(deps) - iter, err := New(sc).Generate(context.Background(), batches) + iter, err := New(sc).Generate(context.Background(), batches, nil) require.NoError(t, err) it := iteratorOf(t, iter) @@ -480,7 +573,7 @@ func TestBestFirst_DrainYieldsEveryCombinationOnce(t *testing.T) { } sc := scored(map[string]float64{"q/A": 0.9, "q/B": 0.7, "q/C": 0.6}) - iter, err := New(sc).Generate(context.Background(), batches) + iter, err := New(sc).Generate(context.Background(), batches, nil) require.NoError(t, err) cands := drainAll(t, iter) @@ -530,7 +623,7 @@ func TestBestFirst_ScoresGloballyNonIncreasing(t *testing.T) { } sc := scored(map[string]float64{"q/A": 0.85, "q/B": 0.3, "q/C": 0.65}) - iter, err := New(sc).Generate(context.Background(), batches) + iter, err := New(sc).Generate(context.Background(), batches, nil) require.NoError(t, err) cands := drainAll(t, iter) @@ -554,7 +647,7 @@ func TestBestFirst_EqualScoresOrderDeterministically(t *testing.T) { {ID: "q/c", State: entity.BatchStateSpeculating}, } - iter, err := New(scored(nil)).Generate(context.Background(), batches) + iter, err := New(scored(nil)).Generate(context.Background(), batches, nil) require.NoError(t, err) cands := drainAll(t, iter) @@ -577,7 +670,7 @@ func TestBestFirst_EqualScoresOrderDeterministically(t *testing.T) { } sc := scored(map[string]float64{"q/coinA": 0.5, "q/coinB": 0.5}) - iter, err := New(sc).Generate(context.Background(), batches) + iter, err := New(sc).Generate(context.Background(), batches, nil) require.NoError(t, err) cands := drainAll(t, iter) @@ -612,9 +705,9 @@ func TestBestFirst_EqualScoresOrderDeterministically(t *testing.T) { } sc := scored(map[string]float64{"q/A": 0.5, "q/B": 0.5, "q/C": 0.5}) - first, err := New(sc).Generate(context.Background(), batches) + first, err := New(sc).Generate(context.Background(), batches, nil) require.NoError(t, err) - second, err := New(sc).Generate(context.Background(), batches) + second, err := New(sc).Generate(context.Background(), batches, nil) require.NoError(t, err) a, b := drainAll(t, first), drainAll(t, second) @@ -627,7 +720,7 @@ func TestBestFirst_NextMakesNoScorerCalls(t *testing.T) { batches, _ := wideHead(6) sc := newCountingScorer(map[string]float64{}) - iter, err := New(sc).Generate(context.Background(), batches) + iter, err := New(sc).Generate(context.Background(), batches, nil) require.NoError(t, err) afterGenerate := sc.total @@ -647,7 +740,7 @@ func TestBestFirst_MemoizesDependencyScoresAcrossHeads(t *testing.T) { } sc := newCountingScorer(map[string]float64{"q/shared": 0.7}) - _, err := New(sc).Generate(context.Background(), batches) + _, err := New(sc).Generate(context.Background(), batches, nil) require.NoError(t, err) assert.Equal(t, 1, sc.calls["q/shared"], "a shared dependency is scored once") @@ -679,7 +772,7 @@ func TestBestFirst_MatchesBruteForceEnumeration(t *testing.T) { ID: "q/H", State: entity.BatchStateSpeculating, Dependencies: deps, }) - iter, err := New(scored(scores)).Generate(context.Background(), batches) + iter, err := New(scored(scores)).Generate(context.Background(), batches, nil) require.NoError(t, err) got := drainAll(t, iter) @@ -750,7 +843,7 @@ func TestBestFirst_WideHeadsRankWithoutUnderflow(t *testing.T) { wide("q/narrow", narrowWidth) wide("q/wide", wideWidth) - iter, err := New(constScorer{depScore}).Generate(context.Background(), batches) + iter, err := New(constScorer{depScore}).Generate(context.Background(), batches, nil) require.NoError(t, err) first, ok, err := iter.Next(context.Background()) @@ -788,7 +881,7 @@ func TestBestFirst_HonorsCancelledContext(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) cancel() - iter, err := New(scored(nil)).Generate(ctx, batches) + iter, err := New(scored(nil)).Generate(ctx, batches, nil) require.ErrorIs(t, err, context.Canceled) assert.Nil(t, iter) }) @@ -797,7 +890,7 @@ func TestBestFirst_HonorsCancelledContext(t *testing.T) { ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(-time.Minute)) defer cancel() - _, err := New(scored(nil)).Generate(ctx, batches) + _, err := New(scored(nil)).Generate(ctx, batches, nil) require.ErrorIs(t, err, context.DeadlineExceeded) }) @@ -805,7 +898,7 @@ func TestBestFirst_HonorsCancelledContext(t *testing.T) { // Generate on a live context so the stream has candidates waiting; the // cancel lands between pulls, which is where a caller that has given up // actually stops. - iter, err := New(scored(nil)).Generate(context.Background(), batches) + iter, err := New(scored(nil)).Generate(context.Background(), batches, nil) require.NoError(t, err) _, ok, err := iter.Next(context.Background()) @@ -829,14 +922,14 @@ func TestBestFirst_HonorsCancelledContext(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() - iter, err := New(cancellingScorer{cancel: cancel}).Generate(ctx, batches) + iter, err := New(cancellingScorer{cancel: cancel}).Generate(ctx, batches, nil) require.ErrorIs(t, err, context.Canceled) assert.Nil(t, iter) }) } -// cancellingScorer kills the context and then fails, the way a scorer whose -// own call was cancelled would. +// cancellingScorer kills the context and then fails, the way a scorer +// whose own call was cancelled would. type cancellingScorer struct{ cancel context.CancelFunc } func (s cancellingScorer) Score(context.Context, entity.Batch, entity.SpeculationPathSet) (float64, error) { @@ -870,7 +963,7 @@ func TestBestFirst_DefaultsScoreOutsideUnitInterval(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - iter, err := New(constScorer{tt.score}).Generate(context.Background(), batches) + iter, err := New(constScorer{tt.score}).Generate(context.Background(), batches, nil) require.NoError(t, err) cands := drainAll(t, iter) @@ -898,7 +991,7 @@ func TestBestFirst_ImpossibleFlipScoresNegativeInfinity(t *testing.T) { } sc := scored(map[string]float64{"q/certain": 1.0, "q/toss": 0.6}) - iter, err := New(sc).Generate(context.Background(), batches) + iter, err := New(sc).Generate(context.Background(), batches, nil) require.NoError(t, err) cands := drainAll(t, iter) @@ -932,7 +1025,7 @@ func TestBestFirst_ResolvedDependenciesAreNeverScored(t *testing.T) { } sc := newCountingScorer(map[string]float64{"q/running": 0.8}) - iter, err := New(sc).Generate(context.Background(), batches) + iter, err := New(sc).Generate(context.Background(), batches, nil) require.NoError(t, err) cands := drainAll(t, iter) @@ -953,7 +1046,7 @@ func TestBestFirst_ReturnedPathsAreIndependent(t *testing.T) { // scribbling on what it was handed must not reach the paths still to come. batches, _ := wideHead(3) iter, err := New(scored(map[string]float64{"q/dep00": 0.9, "q/dep01": 0.8, "q/dep02": 0.7})). - Generate(context.Background(), batches) + Generate(context.Background(), batches, nil) require.NoError(t, err) first, ok, err := iter.Next(context.Background()) @@ -1010,7 +1103,7 @@ func TestBestFirst_ScoresAreSummedFromTheHeadsBestScore(t *testing.T) { want[s.scoreFor(taken)]++ } - iter, err := New(scored(scores)).Generate(context.Background(), batches) + iter, err := New(scored(scores)).Generate(context.Background(), batches, nil) require.NoError(t, err) got := map[float64]int{} for _, c := range drainAll(t, iter) { @@ -1038,7 +1131,7 @@ func TestBestFirst_UntouchedHeadsNeverWorkOutFlips(t *testing.T) { scores[dep] = 0.6 + 0.005*float64(i) } - iter, err := New(scored(scores)).Generate(context.Background(), batches) + iter, err := New(scored(scores)).Generate(context.Background(), batches, nil) require.NoError(t, err) it := iteratorOf(t, iter) @@ -1078,7 +1171,7 @@ func TestBestFirst_AFailedPullConsumesNothing(t *testing.T) { } sc := scored(map[string]float64{"q/dep": 0.8}) - iter, err := New(sc).Generate(context.Background(), batches) + iter, err := New(sc).Generate(context.Background(), batches, nil) require.NoError(t, err) cancelled, cancel := context.WithCancel(context.Background()) diff --git a/submitqueue/extension/speculation/generator/generator.go b/submitqueue/extension/speculation/generator/generator.go index 3f7a6f8cf..3e8da561c 100644 --- a/submitqueue/extension/speculation/generator/generator.go +++ b/submitqueue/extension/speculation/generator/generator.go @@ -45,7 +45,11 @@ type Generator interface { // duplicate, or self dependency. That is a precondition the caller owns: a // generator may assume it and is not required to detect a breach, so a // malformed snapshot yields undefined candidates rather than an error. - Generate(ctx context.Context, batches []entity.Batch) (Iterator, error) + // + // pathSets is what each batch's builds have done so far, at most one set per + // head and none for a batch nothing has speculated on. It is part of the + // same snapshot as batches and carries the same holding rules. + Generate(ctx context.Context, batches []entity.Batch, pathSets []entity.SpeculationPathSet) (Iterator, error) } // Iterator is a pull-based stream of candidate paths. Beyond what ranking diff --git a/submitqueue/extension/speculation/generator/mock/generator_mock.go b/submitqueue/extension/speculation/generator/mock/generator_mock.go index 22740a6bb..14ceb5559 100644 --- a/submitqueue/extension/speculation/generator/mock/generator_mock.go +++ b/submitqueue/extension/speculation/generator/mock/generator_mock.go @@ -43,18 +43,18 @@ func (m *MockGenerator) EXPECT() *MockGeneratorMockRecorder { } // Generate mocks base method. -func (m *MockGenerator) Generate(ctx context.Context, batches []entity.Batch) (generator.Iterator, error) { +func (m *MockGenerator) Generate(ctx context.Context, batches []entity.Batch, pathSets []entity.SpeculationPathSet) (generator.Iterator, error) { m.ctrl.T.Helper() - ret := m.ctrl.Call(m, "Generate", ctx, batches) + ret := m.ctrl.Call(m, "Generate", ctx, batches, pathSets) ret0, _ := ret[0].(generator.Iterator) ret1, _ := ret[1].(error) return ret0, ret1 } // Generate indicates an expected call of Generate. -func (mr *MockGeneratorMockRecorder) Generate(ctx, batches any) *gomock.Call { +func (mr *MockGeneratorMockRecorder) Generate(ctx, batches, pathSets any) *gomock.Call { mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Generate", reflect.TypeOf((*MockGenerator)(nil).Generate), ctx, batches) + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Generate", reflect.TypeOf((*MockGenerator)(nil).Generate), ctx, batches, pathSets) } // MockIterator is a mock of Iterator interface. diff --git a/submitqueue/extension/speculation/speculator/standard/standard.go b/submitqueue/extension/speculation/speculator/standard/standard.go index c9b4ecd34..e61083c4a 100644 --- a/submitqueue/extension/speculation/speculator/standard/standard.go +++ b/submitqueue/extension/speculation/speculator/standard/standard.go @@ -47,7 +47,7 @@ func New(cfg speculator.Config, gen generator.Generator, alloc allocator.Allocat // allocator spend the budget over the resulting candidate iterator, reconciling // it against the path sets. func (s spec) Speculate(ctx context.Context, batches []entity.Batch, pathSets []entity.SpeculationPathSet) ([]entity.Speculation, error) { - iter, err := s.gen.Generate(ctx, batches) + iter, err := s.gen.Generate(ctx, batches, pathSets) if err != nil { return nil, err } diff --git a/submitqueue/extension/speculation/speculator/standard/standard_test.go b/submitqueue/extension/speculation/speculator/standard/standard_test.go index 1b7560137..d3b6370f4 100644 --- a/submitqueue/extension/speculation/speculator/standard/standard_test.go +++ b/submitqueue/extension/speculation/speculator/standard/standard_test.go @@ -44,10 +44,13 @@ func assumptionFor(p entity.SpeculationPath, dep string) entity.DependencyAssump return entity.DependencyAssumptionUnknown } -// constScorer is a minimal scorer.Scorer that scores every batch identically. +// constScorer is a minimal scorer.Scorer that prices every +// batch identically. type constScorer struct{ v float64 } -func (c constScorer) Score(context.Context, entity.Batch, entity.SpeculationPathSet) (float64, error) { return c.v, nil } +func (c constScorer) Score(context.Context, entity.Batch, entity.SpeculationPathSet) (float64, error) { + return c.v, nil +} func TestComposed_EndToEnd_NaivePair(t *testing.T) { batches := []entity.Batch{ @@ -89,7 +92,7 @@ func TestComposed_WiresGeneratorIntoAllocator(t *testing.T) { gen := generatormock.NewMockGenerator(ctrl) alloc := allocatormock.NewMockAllocator(ctrl) - gen.EXPECT().Generate(gomock.Any(), batches).Return(iter, nil) + gen.EXPECT().Generate(gomock.Any(), batches, gomock.Any()).Return(iter, nil) alloc.EXPECT().Allocate(gomock.Any(), pathSets, iter).Return(want, nil) got, err := New(testCfg, gen, alloc).Speculate(context.Background(), batches, pathSets) @@ -104,7 +107,7 @@ func TestComposed_PropagatesGeneratorError(t *testing.T) { gen := generatormock.NewMockGenerator(ctrl) alloc := allocatormock.NewMockAllocator(ctrl) - gen.EXPECT().Generate(gomock.Any(), gomock.Any()).Return(nil, errGenerate) + gen.EXPECT().Generate(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, errGenerate) // Allocate must not be called when Generate fails (no alloc.EXPECT()). _, err := New(testCfg, gen, alloc).Speculate(context.Background(), nil, nil) From 80e088c751cf6eeb7ee77eef45a57d1ace9e923e Mon Sep 17 00:00:00 2001 From: Preetam Dwivedi Date: Mon, 7 Sep 2026 17:13:05 -0700 Subject: [PATCH 2/8] docs(speculation): align predictor configuration terms MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary ### Why? The orchestrator configuration comments still described factors as odds multipliers and referenced fitting work removed from the RFC. ### What? Describe factors directly as revisions to the scorer price and retain the RFC's neutral, positive-factor contract without changing configuration behavior. ## Test Plan - ✅ `./tool/bazel test //submitqueue/extension/speculation/predictor/... //submitqueue/extension/speculation/generator/... //submitqueue/extension/speculation/speculator/... //service/submitqueue/orchestrator/server:go_default_test` - ✅ `make check-gazelle` --- service/submitqueue/orchestrator/server/config.go | 14 ++++++-------- 1 file changed, 6 insertions(+), 8 deletions(-) diff --git a/service/submitqueue/orchestrator/server/config.go b/service/submitqueue/orchestrator/server/config.go index 517c7c432..529319ba7 100644 --- a/service/submitqueue/orchestrator/server/config.go +++ b/service/submitqueue/orchestrator/server/config.go @@ -80,7 +80,7 @@ const ( factorCancelling = "cancelling" ) -// neutralFactor leaves the scorer's price untouched: odds multiplied by one. +// neutralFactor leaves the scorer's price untouched. const neutralFactor = 1.0 // defaultBuildBudget is how many builds a queue may have occupying CI at once @@ -262,11 +262,9 @@ type speculatorConfig struct { // own, so it is not named again here. type predictorConfig struct { Type string `yaml:"type"` - // Factors multiply the odds of the scorer's price, one per piece of - // evidence, keyed by evidence name. An omitted factor is neutral, so an - // omitted block ranks on the scorer's price alone. Values are hand-set - // placeholders, not measured: they are uncalibrated until the fitting work - // in doc/rfc/submitqueue/outcome-predictor.md lands. + // Factors revise the scorer's price, one per piece of evidence and keyed by + // evidence name. An omitted factor is neutral, so an omitted block ranks on + // the scorer's price alone. Factors map[string]float64 `yaml:"factors"` } @@ -614,8 +612,8 @@ func (p *predictorConfig) normalizeAndValidate(where string) error { default: return fmt.Errorf("%s: unknown predictor factor %q", where, name) } - // Zero would pin every batch carrying the evidence to a probability of - // zero, and a negative multiplier on odds means nothing at all. + // Zero would permanently pin matching batches to 0; negatives cannot + // represent either direction in the factor contract. if factor <= 0 { return fmt.Errorf("%s: predictor factor %q is %v, must be positive", where, name, factor) } From 03569eb13e48c2e0176be2ef7382aa0d395c5129 Mon Sep 17 00:00:00 2001 From: Preetam Dwivedi Date: Mon, 7 Sep 2026 17:29:08 -0700 Subject: [PATCH 3/8] fix(speculation): enforce predictor wiring contract MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary ### Why? Configuration accepted infinite evidence factors even though the predictor rejects them, and the speculator composition test did not prove that path-set evidence reaches the Generator. ### What? Reject non-finite factors during profile loading, cover infinite YAML values, and require the exact path-set snapshot in the Generator wiring expectation. ## Test Plan - ✅ `make fmt` - ✅ `./tool/bazel test //submitqueue/extension/speculation/predictor/... //submitqueue/extension/speculation/generator/... //submitqueue/extension/speculation/speculator/... //service/submitqueue/orchestrator/server:go_default_test` --- service/submitqueue/orchestrator/server/config.go | 5 +++-- service/submitqueue/orchestrator/server/config_test.go | 1 + .../speculation/speculator/standard/standard_test.go | 2 +- 3 files changed, 5 insertions(+), 3 deletions(-) diff --git a/service/submitqueue/orchestrator/server/config.go b/service/submitqueue/orchestrator/server/config.go index 529319ba7..73635072d 100644 --- a/service/submitqueue/orchestrator/server/config.go +++ b/service/submitqueue/orchestrator/server/config.go @@ -16,6 +16,7 @@ package main import ( "fmt" + "math" "os" "time" @@ -614,8 +615,8 @@ func (p *predictorConfig) normalizeAndValidate(where string) error { } // Zero would permanently pin matching batches to 0; negatives cannot // represent either direction in the factor contract. - if factor <= 0 { - return fmt.Errorf("%s: predictor factor %q is %v, must be positive", where, name, factor) + 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 diff --git a/service/submitqueue/orchestrator/server/config_test.go b/service/submitqueue/orchestrator/server/config_test.go index ef1f3edbd..aab3653db 100644 --- a/service/submitqueue/orchestrator/server/config_test.go +++ b/service/submitqueue/orchestrator/server/config_test.go @@ -676,6 +676,7 @@ func TestLoadProfilesConfig_RejectsBadPredictors(t *testing.T) { {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 { diff --git a/submitqueue/extension/speculation/speculator/standard/standard_test.go b/submitqueue/extension/speculation/speculator/standard/standard_test.go index d3b6370f4..a25a4e1c2 100644 --- a/submitqueue/extension/speculation/speculator/standard/standard_test.go +++ b/submitqueue/extension/speculation/speculator/standard/standard_test.go @@ -92,7 +92,7 @@ func TestComposed_WiresGeneratorIntoAllocator(t *testing.T) { gen := generatormock.NewMockGenerator(ctrl) alloc := allocatormock.NewMockAllocator(ctrl) - gen.EXPECT().Generate(gomock.Any(), batches, gomock.Any()).Return(iter, nil) + gen.EXPECT().Generate(gomock.Any(), batches, pathSets).Return(iter, nil) alloc.EXPECT().Allocate(gomock.Any(), pathSets, iter).Return(want, nil) got, err := New(testCfg, gen, alloc).Speculate(context.Background(), batches, pathSets) From 9d1c08732dddabf9841af450145a280c007815e3 Mon Sep 17 00:00:00 2001 From: Preetam Dwivedi Date: Mon, 7 Sep 2026 17:41:31 -0700 Subject: [PATCH 4/8] fix(speculation): overlay queue predictor factors A queue predictor block now revises named factors instead of replacing the whole map, so defaults like pathFailed stay in force. Best-first tests rank pathFailed, cancelling, and merging through the evidence predictor rather than a stub. # Conflicts: # submitqueue/extension/speculation/generator/bestfirst/bestfirst_test.go # Please enter the commit message for your changes. Lines starting # with '#' will be kept; you may remove them yourself if you want to. # An empty message aborts the commit. # # interactive rebase in progress; onto 0ff6ea72 # Last commands done (4 commands done): # pick 7fa81830 # fix(speculation): enforce predictor wiring contract # pick f636c332 # fix(speculation): overlay queue predictor factors # Next command to do (1 remaining command): # pick a97225bc # docs(speculation): document predictor wiring # You are currently rebasing branch 'preetam/outcome-predictor-wiring' on '0ff6ea72'. # # Changes to be committed: # modified: service/submitqueue/orchestrator/server/config.go # modified: service/submitqueue/orchestrator/server/config_test.go # modified: submitqueue/extension/speculation/generator/bestfirst/bestfirst_test.go # --- .../submitqueue/orchestrator/server/config.go | 29 ++++++- .../orchestrator/server/config_test.go | 15 ++++ .../generator/bestfirst/bestfirst_test.go | 87 +++++++++++++------ 3 files changed, 101 insertions(+), 30 deletions(-) diff --git a/service/submitqueue/orchestrator/server/config.go b/service/submitqueue/orchestrator/server/config.go index 73635072d..a88a8bbd4 100644 --- a/service/submitqueue/orchestrator/server/config.go +++ b/service/submitqueue/orchestrator/server/config.go @@ -16,6 +16,7 @@ package main import ( "fmt" + "maps" "math" "os" "time" @@ -249,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, @@ -264,8 +265,9 @@ type speculatorConfig struct { 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 factor is neutral, so an omitted block ranks on - // the scorer's price alone. + // 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"` } @@ -394,11 +396,30 @@ func (c profilesConfig) resolve(q namedQueueProfileConfig) queueProfileConfig { profile.Speculator = *q.Speculator } if q.Predictor != nil { - profile.Predictor = *q.Predictor + 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 diff --git a/service/submitqueue/orchestrator/server/config_test.go b/service/submitqueue/orchestrator/server/config_test.go index aab3653db..976d9400f 100644 --- a/service/submitqueue/orchestrator/server/config_test.go +++ b/service/submitqueue/orchestrator/server/config_test.go @@ -713,3 +713,18 @@ func TestLoadProfilesConfig_ReadsPredictorFactors(t *testing.T) { 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) +} diff --git a/submitqueue/extension/speculation/generator/bestfirst/bestfirst_test.go b/submitqueue/extension/speculation/generator/bestfirst/bestfirst_test.go index 567fad0d3..d68aa5852 100644 --- a/submitqueue/extension/speculation/generator/bestfirst/bestfirst_test.go +++ b/submitqueue/extension/speculation/generator/bestfirst/bestfirst_test.go @@ -460,22 +460,22 @@ type flatScorer struct{} func (flatScorer) Score(context.Context, entity.Batch, entity.SpeculationPathSet) (float64, error) { return 0.5, nil } -// PathPassed on a green all-succeed build is the join the generator exists to -// consume: same scorer price, different evidence, different rank. -func TestBestFirst_EvidencePathPassedRanksTheGreenDependencyFirst(t *testing.T) { - batches := []entity.Batch{ - {ID: "q/built", State: entity.BatchStateSpeculating}, - {ID: "q/fresh", State: entity.BatchStateSpeculating}, - {ID: "q/H", State: entity.BatchStateSpeculating, Dependencies: []string{"q/built", "q/fresh"}}, - } - built := entity.SpeculationPathSet{ +func evidenceScorer(t *testing.T, factors evidence.Factors) scorer.Scorer { + t.Helper() + s, err := evidence.New(scorer.Config{QueueName: "q"}, flatScorer{}, factors, tally.NoopScope) + require.NoError(t, err) + return s +} + +func allSucceedSet(head string, status entity.SpeculationPathStatus) entity.SpeculationPathSet { + return entity.SpeculationPathSet{ Queue: "q", - Head: "q/built", + Head: head, Paths: []entity.SpeculationPathEntry{{ ID: "p1", - Status: entity.SpeculationPathStatusPassed, + Status: status, Path: entity.SpeculationPath{ - Head: "q/built", + Head: head, Dependencies: []entity.PathDependency{{ Batch: "q/dep0", Assumption: entity.DependencyAssumptionSucceeds, @@ -483,15 +483,19 @@ func TestBestFirst_EvidencePathPassedRanksTheGreenDependencyFirst(t *testing.T) }, }}, } - pred, err := evidence.New( - scorer.Config{QueueName: "q"}, - flatScorer{}, - evidence.Factors{PathPassed: 9, PathFailed: 1, Merging: 1, Cancelling: 1}, - tally.NoopScope, - ) - require.NoError(t, err) +} - iter, err := New(pred).Generate(context.Background(), batches, []entity.SpeculationPathSet{built}) +// PathPassed on a green all-succeed build is the join the generator exists to +// consume: same scorer price, different evidence, different rank. +func TestBestFirst_EvidencePathPassedRanksTheGreenDependencyFirst(t *testing.T) { + batches := []entity.Batch{ + {ID: "q/built", State: entity.BatchStateSpeculating}, + {ID: "q/fresh", State: entity.BatchStateSpeculating}, + {ID: "q/H", State: entity.BatchStateSpeculating, Dependencies: []string{"q/built", "q/fresh"}}, + } + pred := evidenceScorer(t, evidence.Factors{PathPassed: 9, PathFailed: 1, Merging: 1, Cancelling: 1}) + + iter, err := New(pred).Generate(context.Background(), batches, []entity.SpeculationPathSet{allSucceedSet("q/built", entity.SpeculationPathStatusPassed)}) require.NoError(t, err) cands := forHead(drainAll(t, iter), "q/H") require.NotEmpty(t, cands) @@ -510,25 +514,56 @@ func TestBestFirst_EvidencePathPassedRanksTheGreenDependencyFirst(t *testing.T) assert.Greater(t, cands[0].RankingScore, failScore) } +func TestBestFirst_EvidencePathFailedPrefersTheFailedSide(t *testing.T) { + batches := []entity.Batch{ + {ID: "q/failed", State: entity.BatchStateSpeculating}, + {ID: "q/H", State: entity.BatchStateSpeculating, Dependencies: []string{"q/failed"}}, + } + pred := evidenceScorer(t, evidence.Factors{PathPassed: 1, PathFailed: 0.25, Merging: 1, Cancelling: 1}) + + iter, err := New(pred).Generate(context.Background(), batches, []entity.SpeculationPathSet{allSucceedSet("q/failed", entity.SpeculationPathStatusFailed)}) + require.NoError(t, err) + cands := forHead(drainAll(t, iter), "q/H") + require.Len(t, cands, 2) + assert.Equal(t, entity.DependencyAssumptionFails, assumptionFor(cands[0].Path, "q/failed")) + assert.Equal(t, entity.DependencyAssumptionSucceeds, assumptionFor(cands[1].Path, "q/failed")) + assert.Greater(t, cands[0].RankingScore, cands[1].RankingScore) +} + +func TestBestFirst_EvidenceCancellingPrefersTheFailedSide(t *testing.T) { + batches := []entity.Batch{ + {ID: "q/stopping", State: entity.BatchStateCancelling}, + {ID: "q/H", State: entity.BatchStateSpeculating, Dependencies: []string{"q/stopping"}}, + } + pred := evidenceScorer(t, evidence.Factors{PathPassed: 1, PathFailed: 1, Merging: 1, Cancelling: 0.25}) + + iter, err := New(pred).Generate(context.Background(), batches, nil) + require.NoError(t, err) + cands := drainAll(t, iter) + require.Len(t, cands, 2) + assert.Equal(t, entity.DependencyAssumptionFails, assumptionFor(cands[0].Path, "q/stopping")) + assert.Equal(t, entity.DependencyAssumptionSucceeds, assumptionFor(cands[1].Path, "q/stopping")) + assert.Greater(t, cands[0].RankingScore, cands[1].RankingScore) +} + // A merging dependency is still in progress — the merge can fail — so it stays -// an open question here like any other. Whether a path betting against it is -// worth funding is a matter of price, which is the scorer's to say, not a -// state the search hard-codes. +// an open question here like any other. How much it is worth is a scorer +// price, not a fact the search hard-codes. func TestBestFirst_MergingDependencyStaysOpen(t *testing.T) { batches := []entity.Batch{ {ID: "q/landing", State: entity.BatchStateMerging}, {ID: "q/H", State: entity.BatchStateSpeculating, Dependencies: []string{"q/landing"}}, } - sc := newCountingScorer(map[string]float64{"q/landing": 0.9}) + pred := evidenceScorer(t, evidence.Factors{PathPassed: 1, PathFailed: 1, Merging: 19, Cancelling: 1}) - iter, err := New(sc).Generate(context.Background(), batches, nil) + iter, err := New(pred).Generate(context.Background(), batches, nil) require.NoError(t, err) cands := drainAll(t, iter) - assert.Equal(t, 1, sc.calls["q/landing"], "a merging dependency is priced like any other") require.Len(t, cands, 2, "both sides of a merge that has not landed yet") assert.Equal(t, entity.DependencyAssumptionSucceeds, assumptionFor(cands[0].Path, "q/landing")) assert.Equal(t, entity.DependencyAssumptionFails, assumptionFor(cands[1].Path, "q/landing")) + assert.Greater(t, cands[0].RankingScore, cands[1].RankingScore) } func TestBestFirst_GeneratesOnlyWhatIsPulled(t *testing.T) { From dc9298fa8ab592db003ba895b354a5b3f6ef5c97 Mon Sep 17 00:00:00 2001 From: Preetam Dwivedi Date: Mon, 7 Sep 2026 17:56:41 -0700 Subject: [PATCH 5/8] docs(speculation): document predictor wiring Update generator and standard-speculator guides for the shared path-set snapshot and predictor-backed best-first ranking. --- submitqueue/extension/speculation/generator/README.md | 2 +- .../extension/speculation/generator/bestfirst/README.md | 8 ++++---- .../extension/speculation/speculator/standard/README.md | 2 +- 3 files changed, 6 insertions(+), 6 deletions(-) diff --git a/submitqueue/extension/speculation/generator/README.md b/submitqueue/extension/speculation/generator/README.md index 93d6de520..c1f0166b1 100644 --- a/submitqueue/extension/speculation/generator/README.md +++ b/submitqueue/extension/speculation/generator/README.md @@ -2,7 +2,7 @@ The `generator` package is a piece the `standard` `Speculator` is built from: a `Generator` produces the queue's candidate paths as one ordered stream across all heads. It is **not** controller-facing — the speculate controller only knows the `Speculator` contract, and a different `Speculator` need not split its work this way. So there is no `Config` or `Factory` here; a `Generator` is chosen when the `standard` `Speculator` is constructed. -`Generate` starts the stream over the queue's live batches and returns an `Iterator`. Beyond any ranking work required up front, the generator computes only the candidates the caller pulls. A cancelled or expired context ends the stream with its error. The snapshot must include every batch a head's direct dependencies reference, carry unique non-empty IDs, and give no head an empty, duplicate, or self dependency. Those are the caller's preconditions: a generator may assume them and is not required to detect a breach, so a malformed snapshot yields undefined candidates rather than an error. +`Generate` starts the stream over the queue's live batches and path sets and returns an `Iterator`. The path sets are what each batch's builds have done so far — at most one set per head, none for a batch nothing has speculated on. They are part of the same snapshot as the batches and carry the same holding rules: callers must not mutate either while pulling. Beyond any ranking work required up front, the generator computes only the candidates the caller pulls. A cancelled or expired context ends the stream with its error. The snapshot must include every batch a head's direct dependencies reference, carry unique non-empty IDs, and give no head an empty, duplicate, or self dependency. Those are the caller's preconditions: a generator may assume them and is not required to detect a breach, so a malformed snapshot yields undefined candidates rather than an error. Candidates never repeat and never contradict a known fact. Beyond that, the order is the `Generator`'s own: it yields candidates in whatever ranking it implements, and each carries the score it ranked by — higher first, on a scale the generator defines. Consumers take the iterator in the order given and do not interpret the score. Scores mean something only within the run and are never stored. diff --git a/submitqueue/extension/speculation/generator/bestfirst/README.md b/submitqueue/extension/speculation/generator/bestfirst/README.md index bd34769dd..fc458d118 100644 --- a/submitqueue/extension/speculation/generator/bestfirst/README.md +++ b/submitqueue/extension/speculation/generator/bestfirst/README.md @@ -2,16 +2,16 @@ `bestfirst` implements `generator.Generator` by ranking candidate paths by the probability that all their dependency assumptions hold. It returns one path per pull across all speculating heads without enumerating every combination up front. -The [best-first speculation path generation RFC](../../../../../doc/rfc/submitqueue/speculation-generator-best-first.md) defines the terminology, algorithm, correctness argument, worked example, and alternatives considered. This README records only the package's operational behavior. +The [best-first speculation path generation RFC](../../../../../doc/rfc/submitqueue/speculation-generator-best-first.md) defines the terminology, algorithm, correctness argument, worked example, and alternatives considered. Path-set evidence and factor semantics live in the [outcome predictor RFC](../../../../../doc/rfc/submitqueue/outcome-predictor.md). This README records only the package's operational behavior. ## Behavior -- `Generate` validates the snapshot, scores each unique unresolved direct dependency once, fixes assumptions for resolved dependencies, calculates each head's best score, and seeds the global heap with every eligible head's best-path candidate. Each head's remaining paths wait in that head's own lazy stream, whose flips are worked out only when the head is first handed out. +- `Generate` takes the queue's live batches and path sets as one snapshot, predicts each unique unresolved direct dependency once against that dependency's own path set, fixes assumptions for resolved dependencies, calculates each head's best score, and seeds the global heap with every eligible head's best-path candidate. Each head's remaining paths wait in that head's own lazy stream, whose flips are worked out only when the head is first handed out. - `Next` removes the highest-ranked candidate, advances only that head's stream, constructs that candidate's complete path, and returns it. Pulling long enough returns every path exactly once in non-increasing score order. - Ranking scores are sums of log probabilities, avoiding underflow while preserving probability order. They are meaningful only within the run that produced them. - Exact ties prefer fewer flips; head ID then decides between heads (the cross-head heap holds one candidate per head), and taken flip indexes decide within a head. -- A dependency counts as resolved only once it is terminal. Merging and cancelling are both still in progress and either can end the other way, so both stay open questions here. Whether a path betting against a merging dependency is worth funding is a matter of price, and price is the scorer's to say. -- A dependency that cannot be priced — the scorer call failed, the score was not a probability, or the snapshot never carried the batch — is treated as very likely to succeed rather than ending the run. One unusable number costs its own estimate, never the queue's whole set of candidates. A batch missing from the snapshot is never handed to the scorer at all: it would resolve to a zero batch belonging to no queue. +- A dependency counts as resolved only once it is terminal. Merging and cancelling are both still in progress and either can end the other way, so both stay open questions here. How much a merging or cancelling dependency is worth is a predictor price, not a fact the search hard-codes. +- A dependency that cannot be priced — the predictor call failed, the probability was not in `[0, 1]`, or the snapshot never carried the batch — is treated as very likely to succeed rather than ending the run. One unusable number costs its own estimate, never the queue's whole set of candidates. A batch missing from the snapshot is never handed to the predictor at all: it would resolve to a zero batch belonging to no queue. - The snapshot must contain every batch a head's direct dependencies reference, carry unique non-empty batch IDs, and give no head an empty, duplicate, or self dependency. That is the caller's precondition, not something checked here: a malformed snapshot yields undefined candidates rather than an error. The behavior is covered by `bestfirst_test.go`. diff --git a/submitqueue/extension/speculation/speculator/standard/README.md b/submitqueue/extension/speculation/speculator/standard/README.md index c5d869e65..c797b2ebe 100644 --- a/submitqueue/extension/speculation/speculator/standard/README.md +++ b/submitqueue/extension/speculation/speculator/standard/README.md @@ -6,7 +6,7 @@ Each run it considers candidate paths in descending order of their probability o When the budget runs out, everything below the cut waits for a later run. That is safe because the propose-side cannot invent a batch verdict: the speculate controller still decides merge from the persisted paths, including complete coverage of unsettled dependencies. -Both halves are swappable. The ranking is the `Generator`'s: the default `bestfirst` scores each path by the probability that all its assumptions hold. The budget policy is the `Allocator`'s: the default `sticky` fills only free slots and never preempts, where a preempting allocator would cancel a low-value in-flight path to fund a better one. +Both halves are swappable. The ranking is the `Generator`'s: the default `bestfirst` asks the queue's predictor for each unresolved dependency's probability of reaching Succeeded, then ranks paths by the probability that all their assumptions hold. The budget policy is the `Allocator`'s: the default `sticky` fills only free slots and never preempts, where a preempting allocator would cancel a low-value in-flight path to fund a better one. `standard` itself decides nothing — it connects the `Generator`'s stream to the `Allocator` — so changing prioritization or budget behavior means swapping a part, not writing a new `Speculator`. From 73addbe763799d388dafa7243461f2fa8f15c67f Mon Sep 17 00:00:00 2001 From: Preetam Dwivedi Date: Fri, 11 Sep 2026 09:46:45 -0700 Subject: [PATCH 6/8] feat(speculation): rank bestfirst on evidence Scorer Compose evidence around a nested base from YAML type evidence. Drop the sibling predictor block. Overlay named factors; replace base wholesale. --- .../orchestrator/server/BUILD.bazel | 4 +- .../submitqueue/orchestrator/server/config.go | 154 +++++++++--------- .../orchestrator/server/config_test.go | 62 +++---- .../submitqueue/orchestrator/server/main.go | 26 +-- .../orchestrator/server/profiles.go | 97 ++++------- .../orchestrator/server/profiles_test.go | 51 ++---- .../speculation/generator/bestfirst/README.md | 6 +- .../speculation/speculator/standard/README.md | 2 +- 8 files changed, 177 insertions(+), 225 deletions(-) diff --git a/service/submitqueue/orchestrator/server/BUILD.bazel b/service/submitqueue/orchestrator/server/BUILD.bazel index fde019fcc..6dc577df0 100644 --- a/service/submitqueue/orchestrator/server/BUILD.bazel +++ b/service/submitqueue/orchestrator/server/BUILD.bazel @@ -52,10 +52,9 @@ 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/evidence:go_default_library", "//submitqueue/extension/speculation/scorer/fake:go_default_library", "//submitqueue/extension/speculation/scorer/heuristic:go_default_library", "//submitqueue/extension/speculation/speculator:go_default_library", @@ -123,7 +122,6 @@ 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", diff --git a/service/submitqueue/orchestrator/server/config.go b/service/submitqueue/orchestrator/server/config.go index a88a8bbd4..4ac098208 100644 --- a/service/submitqueue/orchestrator/server/config.go +++ b/service/submitqueue/orchestrator/server/config.go @@ -55,6 +55,7 @@ const ( // Scorer types selectable from configuration. const ( + scorerTypeEvidence = "evidence" scorerTypeHeuristic = "heuristic" scorerTypeComposite = "composite" ) @@ -69,10 +70,7 @@ 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 +// Evidence an evidence scorer prices, as named in configuration. The set is // closed: a factor under any other name would be applied to nothing and never // noticed. const ( @@ -128,7 +126,6 @@ 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. @@ -138,7 +135,6 @@ 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 @@ -228,11 +224,19 @@ type analyzerConfig struct { FailAlways bool `yaml:"failAlways"` } -// scorerConfig selects how a queue ranks candidate speculation paths. There is -// no scoring stage: the scorer feeds the queue's speculator, which is composed -// from it rather than configured separately. +// scorerConfig selects how a queue ranks candidate speculation paths. The +// ranking scorer is evidence wrapping a nested content base. Heuristic and +// composite belong on base (and on composite components), not at the top level. type scorerConfig struct { Type string `yaml:"type"` + // Factors revise the base price, one per piece of evidence (evidence only). + // An omitted key keeps the inherited value, or 1 if neither defaults nor + // the queue named it. + Factors map[string]float64 `yaml:"factors"` + // Base is the content scorer evidence revises (evidence only). An omitted + // base on defaults is the default heuristic; a present base on a queue + // replaces the default base wholesale. + Base *scorerConfig `yaml:"base"` // Buckets map a batch's total lines changed onto a score (heuristic only). Buckets []bucketConfig `yaml:"buckets"` // Components are the scorers a composite combines, keyed by name. @@ -250,7 +254,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 predictor, and what +// `type`: there is one speculator, composed from the queue's scorer, 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, @@ -259,18 +263,6 @@ 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) @@ -323,7 +315,7 @@ func (c *profilesConfig) normalizeAndValidate() error { } } if q.Scorer != nil { - if err := q.Scorer.normalizeAndValidate(where); err != nil { + if err := q.Scorer.normalizeOverlay(where); err != nil { return err } } @@ -332,11 +324,6 @@ func (c *profilesConfig) normalizeAndValidate() error { return err } } - if q.Predictor != nil { - if err := q.Predictor.normalizeAndValidate(where); err != nil { - return err - } - } } return c.validateGitRepoPaths() } @@ -390,33 +377,33 @@ func (c profilesConfig) resolve(q namedQueueProfileConfig) queueProfileConfig { profile.Analyzer = *q.Analyzer } if q.Scorer != nil { - profile.Scorer = *q.Scorer + profile.Scorer = overlayScorer(profile.Scorer, *q.Scorer) } 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 { +// overlayScorer keeps default factors the queue did not name. A present base +// replaces the default base wholesale. Type stays evidence unless the override +// names one, which must still be evidence. +func overlayScorer(base, override scorerConfig) scorerConfig { if override.Type != "" { base.Type = override.Type } - if len(override.Factors) == 0 { - return base + if len(override.Factors) > 0 { + merged := maps.Clone(base.Factors) + if merged == nil { + merged = make(map[string]float64, len(override.Factors)) + } + maps.Copy(merged, override.Factors) + base.Factors = merged } - merged := maps.Clone(base.Factors) - if merged == nil { - merged = make(map[string]float64, len(override.Factors)) + if override.Base != nil { + copied := *override.Base + base.Base = &copied } - maps.Copy(merged, override.Factors) - base.Factors = merged return base } @@ -436,7 +423,7 @@ func (p *queueProfileConfig) normalizeAndValidate(where string) error { if err := p.Speculator.normalizeAndValidate(where); err != nil { return err } - return p.Predictor.normalizeAndValidate(where) + return nil } func (c *changeProviderConfig) normalizeAndValidate(where string) error { @@ -576,10 +563,56 @@ func (a *analyzerConfig) normalizeAndValidate(where string) error { } } -// normalizeAndValidate applies defaults and rejects a scorer that could not be -// built. An empty block is a flat heuristic: every batch scores the same, which -// is the neutral choice for a queue with no opinion about ordering. +// normalizeAndValidate applies defaults and rejects a ranking scorer that +// could not be built. An empty block is evidence wrapping the default +// heuristic, with every factor neutral. func (s *scorerConfig) normalizeAndValidate(where string) error { + return s.normalizeRanking(where, true) +} + +// normalizeOverlay validates a queue's scorer override without inventing a +// base: omitted base means inherit the default base. +func (s *scorerConfig) normalizeOverlay(where string) error { + return s.normalizeRanking(where, false) +} + +func (s *scorerConfig) normalizeRanking(where string, fillBase bool) error { + if s.Type == "" { + s.Type = scorerTypeEvidence + } + if s.Type != scorerTypeEvidence { + return fmt.Errorf("%s: scorer type %q belongs under base, not at the ranking layer", where, s.Type) + } + if err := validateFactors(where, s.Factors); err != nil { + return err + } + if s.Base != nil { + return s.Base.normalizeContent(where + " base") + } + if fillBase { + s.Base = &scorerConfig{} + return s.Base.normalizeContent(where + " base") + } + return nil +} + +func validateFactors(where string, factors map[string]float64) error { + for name, factor := range factors { + switch name { + case factorPathPassed, factorPathFailed, factorMerging, factorCancelling: + default: + return fmt.Errorf("%s: unknown scorer 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: scorer factor %q is %v, must be finite and positive", where, name, factor) + } + } + return nil +} + +func (s *scorerConfig) normalizeContent(where string) error { if s.Type == "" { s.Type = scorerTypeHeuristic } @@ -601,7 +634,7 @@ func (s *scorerConfig) normalizeAndValidate(where string) error { return fmt.Errorf("%s: composite scorer needs at least one component", where) } for name, component := range s.Components { - if err := component.normalizeAndValidate(fmt.Sprintf("%s component %q", where, name)); err != nil { + if err := component.normalizeContent(fmt.Sprintf("%s component %q", where, name)); err != nil { return err } s.Components[name] = component @@ -618,31 +651,6 @@ 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, diff --git a/service/submitqueue/orchestrator/server/config_test.go b/service/submitqueue/orchestrator/server/config_test.go index 976d9400f..d120166c1 100644 --- a/service/submitqueue/orchestrator/server/config_test.go +++ b/service/submitqueue/orchestrator/server/config_test.go @@ -344,20 +344,24 @@ func TestDefaultProfilesConfig_KeepsPerQueueScorers(t *testing.T) { byName[q.Name] = q } - assert.Equal(t, scorerTypeHeuristic, cfg.Defaults.Scorer.Type) - assert.Len(t, cfg.Defaults.Scorer.Buckets, 1, "the baseline scores every batch alike") + assert.Equal(t, scorerTypeEvidence, cfg.Defaults.Scorer.Type) + require.NotNil(t, cfg.Defaults.Scorer.Base) + assert.Equal(t, scorerTypeHeuristic, cfg.Defaults.Scorer.Base.Type) + assert.Len(t, cfg.Defaults.Scorer.Base.Buckets, 1, "the baseline scores every batch alike") bucketed, ok := byName["test-queue"] require.True(t, ok) require.NotNil(t, bucketed.Scorer) - assert.Equal(t, scorerTypeHeuristic, bucketed.Scorer.Type) - assert.Len(t, bucketed.Scorer.Buckets, 4, "smaller batches must rank ahead of larger ones") + require.NotNil(t, bucketed.Scorer.Base) + assert.Equal(t, scorerTypeHeuristic, bucketed.Scorer.Base.Type) + assert.Len(t, bucketed.Scorer.Base.Buckets, 4, "smaller batches must rank ahead of larger ones") comp, ok := byName["e2e-test-queue"] require.True(t, ok) require.NotNil(t, comp.Scorer) - assert.Equal(t, scorerTypeComposite, comp.Scorer.Type) - assert.ElementsMatch(t, []string{"size", "flat"}, keysOf(comp.Scorer.Components)) + require.NotNil(t, comp.Scorer.Base) + assert.Equal(t, scorerTypeComposite, comp.Scorer.Base.Type) + assert.ElementsMatch(t, []string{"size", "flat"}, keysOf(comp.Scorer.Base.Components)) } func keysOf(m map[string]scorerConfig) []string { @@ -654,10 +658,11 @@ func TestLoadProfilesConfig_RejectsBadScorers(t *testing.T) { contents string }{ {name: "unknown scorer type", contents: "defaults:\n scorer: {type: vibes}\n"}, - {name: "composite with no components", contents: "defaults:\n scorer: {type: composite}\n"}, - {name: "unknown combine", contents: "defaults:\n scorer:\n type: composite\n combine: median\n components: {a: {type: heuristic}}\n"}, - {name: "score out of range", contents: "defaults:\n scorer:\n type: heuristic\n buckets: [{min: 0, max: 10, score: 2.0}]\n"}, - {name: "inverted bucket", contents: "defaults:\n scorer:\n type: heuristic\n buckets: [{min: 10, max: 1, score: 0.5}]\n"}, + {name: "top-level heuristic", contents: "defaults:\n scorer: {type: heuristic}\n"}, + {name: "composite with no components", contents: "defaults:\n scorer:\n base: {type: composite}\n"}, + {name: "unknown combine", contents: "defaults:\n scorer:\n base:\n type: composite\n combine: median\n components: {a: {type: heuristic}}\n"}, + {name: "score out of range", contents: "defaults:\n scorer:\n base:\n type: heuristic\n buckets: [{min: 0, max: 10, score: 2.0}]\n"}, + {name: "inverted bucket", contents: "defaults:\n scorer:\n base:\n type: heuristic\n buckets: [{min: 10, max: 1, score: 0.5}]\n"}, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { @@ -667,17 +672,16 @@ func TestLoadProfilesConfig_RejectsBadScorers(t *testing.T) { } } -func TestLoadProfilesConfig_RejectsBadPredictors(t *testing.T) { +func TestLoadProfilesConfig_RejectsBadScorerFactors(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"}, + {name: "unknown factor", contents: "defaults:\n scorer:\n factors: {pathPased: 2}\n"}, + {name: "zero factor", contents: "defaults:\n scorer:\n factors: {merging: 0}\n"}, + {name: "negative factor", contents: "defaults:\n scorer:\n factors: {pathFailed: -1}\n"}, + {name: "infinite factor", contents: "defaults:\n scorer:\n factors: {pathPassed: .inf}\n"}, + {name: "bad factor on a queue override", contents: "defaults: {}\nqueues:\n - name: q\n scorer:\n factors: {merging: 0}\n"}, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { @@ -687,44 +691,46 @@ func TestLoadProfilesConfig_RejectsBadPredictors(t *testing.T) { } } -// An omitted predictor block leaves the queue ranking on its scorer's price +// An omitted factors map leaves the queue ranking on its base price // alone, which is what every queue does until someone states a factor. -func TestLoadProfilesConfig_DefaultsThePredictorToNeutral(t *testing.T) { +func TestLoadProfilesConfig_DefaultsTheScorerToEvidence(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) + assert.Equal(t, scorerTypeEvidence, cfg.Defaults.Scorer.Type) + require.NotNil(t, cfg.Defaults.Scorer.Base) + assert.Equal(t, scorerTypeHeuristic, cfg.Defaults.Scorer.Base.Type) - factors := factorsFrom(cfg.resolve(cfg.Queues[0]).Predictor) + factors := factorsFrom(cfg.resolve(cfg.Queues[0]).Scorer) 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) { +func TestLoadProfilesConfig_ReadsScorerFactors(t *testing.T) { cfg, err := loadProfilesConfig(writeProfiles(t, - "defaults:\n predictor:\n factors: {pathPassed: 10, pathFailed: 0.3, merging: 12, cancelling: 0.1}\n")) + "defaults:\n scorer:\n factors: {pathPassed: 10, pathFailed: 0.3, merging: 12, cancelling: 0.1}\n")) require.NoError(t, err) - factors := factorsFrom(cfg.Defaults.Predictor) + factors := factorsFrom(cfg.Defaults.Scorer) 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) { +func TestLoadProfilesConfig_QueueScorerFactorsOverlayDefaults(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")) + "defaults:\n scorer:\n factors: {pathPassed: 10, pathFailed: 0.3, merging: 12, cancelling: 0.1}\nqueues:\n - name: q\n scorer:\n factors: {pathPassed: 4}\n")) require.NoError(t, err) - factors := factorsFrom(cfg.resolve(cfg.Queues[0]).Predictor) + factors := factorsFrom(cfg.resolve(cfg.Queues[0]).Scorer) 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) + defaults := factorsFrom(cfg.Defaults.Scorer) assert.Equal(t, 10.0, defaults.PathPassed) } diff --git a/service/submitqueue/orchestrator/server/main.go b/service/submitqueue/orchestrator/server/main.go index fe781b28a..9ff02a40e 100644 --- a/service/submitqueue/orchestrator/server/main.go +++ b/service/submitqueue/orchestrator/server/main.go @@ -369,12 +369,14 @@ func defaultProfilesConfig() profilesConfig { // Bucketed scoring: smaller batches are likelier to land, so they // rank ahead of larger ones. Conflicts stay conservative. {Name: "test-queue", Scorer: &scorerConfig{ - Type: scorerTypeHeuristic, - Buckets: []bucketConfig{ - {Min: 0, Max: 1, Score: 0.95}, - {Min: 2, Max: 5, Score: 0.80}, - {Min: 6, Max: 20, Score: 0.60}, - {Min: 21, Max: maxBucket, Score: 0.40}, + Base: &scorerConfig{ + Type: scorerTypeHeuristic, + Buckets: []bucketConfig{ + {Min: 0, Max: 1, Score: 0.95}, + {Min: 2, Max: 5, Score: 0.80}, + {Min: 6, Max: 20, Score: 0.60}, + {Min: 21, Max: maxBucket, Score: 0.40}, + }, }, }}, // Maximum parallelism: nothing ever conflicts. Scored by a @@ -382,11 +384,13 @@ func defaultProfilesConfig() profilesConfig { {Name: "e2e-test-queue", Analyzer: &analyzerConfig{Type: analyzerTypeNone}, Scorer: &scorerConfig{ - Type: scorerTypeComposite, - Combine: combineAvg, - Components: map[string]scorerConfig{ - "size": {Type: scorerTypeHeuristic, Buckets: []bucketConfig{{Min: 0, Max: maxBucket, Score: 0.8}}}, - "flat": {Type: scorerTypeHeuristic, Buckets: []bucketConfig{{Min: 0, Max: maxBucket, Score: 0.6}}}, + Base: &scorerConfig{ + Type: scorerTypeComposite, + Combine: combineAvg, + Components: map[string]scorerConfig{ + "size": {Type: scorerTypeHeuristic, Buckets: []bucketConfig{{Min: 0, Max: maxBucket, Score: 0.8}}}, + "flat": {Type: scorerTypeHeuristic, Buckets: []bucketConfig{{Min: 0, Max: maxBucket, Score: 0.6}}}, + }, }, }, }, diff --git a/service/submitqueue/orchestrator/server/profiles.go b/service/submitqueue/orchestrator/server/profiles.go index deccb0c76..00cb07a99 100644 --- a/service/submitqueue/orchestrator/server/profiles.go +++ b/service/submitqueue/orchestrator/server/profiles.go @@ -47,10 +47,9 @@ 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" + "github.com/uber/submitqueue/submitqueue/extension/speculation/scorer/evidence" scorerfake "github.com/uber/submitqueue/submitqueue/extension/speculation/scorer/fake" "github.com/uber/submitqueue/submitqueue/extension/speculation/scorer/heuristic" "github.com/uber/submitqueue/submitqueue/extension/speculation/speculator" @@ -77,15 +76,11 @@ type Profile struct { // splits queues across storage backends overrides this per queue. Storage storage.Factory - // Scorer holds this queue's scoring profile. There is no scoring stage: the + // Scorer holds this queue's ranking 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.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 @@ -148,14 +143,6 @@ 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 { @@ -190,10 +177,6 @@ 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) } @@ -283,38 +266,42 @@ func (b *profileBuilder) build(cfg queueProfileConfig, where string) (Profile, e if err != nil { return Profile{}, err } - // 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{ + return withSpeculator(Profile{ ChangeProvider: provider, BuildRunner: runner, Analyzer: analyzer, Storage: b.stores, Scorer: sc, - }, cfg.Predictor, b.scope), cfg.Speculator.BuildBudget), nil + }, 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. +// 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 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 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}) +// The scorer is resolved lazily, at the queue the speculator itself was +// asked for, so the queue's identity reaches the ranking scorer. +func withSpeculator(p Profile, buildBudget int) Profile { + p.Speculator = speculatorFunc(func(c speculator.Config) (speculator.Speculator, error) { + s, 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 specstandard.New(c, bestfirst.New(s), sticky.New(buildBudget)), nil }) 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 { +// batchLines buckets a batch by total lines changed across all its changes — +// larger batches are likelier to fail to land. +func batchLines(_ context.Context, changes entity.BatchChanges) (int, error) { + return changes.TotalLinesChanged(), nil +} + +func factorsFrom(cfg scorerConfig) evidence.Factors { factors := evidence.AllOnes() for name, factor := range cfg.Factors { switch name { @@ -331,33 +318,6 @@ func factorsFrom(cfg predictorConfig) evidence.Factors { return factors } -// withSpeculator returns the profile with its speculator composed from its own -// 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 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) { - pred, err := p.Predictor.For(predictor.Config{QueueName: c.QueueName}) - if err != nil { - return nil, fmt.Errorf("failed to resolve predictor for queue %q: %w", c.QueueName, err) - } - return specstandard.New(c, bestfirst.New(pred), sticky.New(buildBudget)), nil - }) - return p -} - -// batchLines buckets a batch by total lines changed across all its changes — -// larger batches are likelier to fail to land. -func batchLines(_ context.Context, changes entity.BatchChanges) (int, error) { - return changes.TotalLinesChanged(), nil -} - // newScorerFactory builds the configured scorer's factory. // // Every scorer is wrapped by scorerfake so a change URI carrying @@ -367,11 +327,18 @@ func batchLines(_ context.Context, changes entity.BatchChanges) (int, error) { // The configuration is walked once up front so an unusable scorer config fails // at wiring time rather than on the first queue that resolves it. func (b *profileBuilder) newScorerFactory(cfg scorerConfig, where string) (scorer.Factory, error) { - if _, err := b.buildScorer(scorer.Config{}, cfg, where, "scorer"); err != nil { + if cfg.Base == nil { + return nil, fmt.Errorf("%s: evidence scorer needs a base", where) + } + if _, err := b.buildScorer(scorer.Config{}, *cfg.Base, where, "scorer.base"); err != nil { return nil, err } return scorerFunc(func(c scorer.Config) (scorer.Scorer, error) { - inner, err := b.buildScorer(c, cfg, where, "scorer") + base, err := b.buildScorer(c, *cfg.Base, where, "scorer.base") + if err != nil { + return nil, err + } + inner, err := evidence.New(c, base, factorsFrom(cfg), b.scope.SubScope("scorer")) if err != nil { return nil, err } diff --git a/service/submitqueue/orchestrator/server/profiles_test.go b/service/submitqueue/orchestrator/server/profiles_test.go index ccf297018..05b41159b 100644 --- a/service/submitqueue/orchestrator/server/profiles_test.go +++ b/service/submitqueue/orchestrator/server/profiles_test.go @@ -21,13 +21,11 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" - "github.com/uber-go/tally" "github.com/uber/submitqueue/submitqueue/entity" "github.com/uber/submitqueue/submitqueue/extension/buildrunner" "github.com/uber/submitqueue/submitqueue/extension/changeprovider" "github.com/uber/submitqueue/submitqueue/extension/conflict" - "github.com/uber/submitqueue/submitqueue/extension/speculation/predictor" "github.com/uber/submitqueue/submitqueue/extension/speculation/scorer" "github.com/uber/submitqueue/submitqueue/extension/speculation/speculator" "github.com/uber/submitqueue/submitqueue/extension/storage" @@ -41,14 +39,15 @@ type recorder struct { analyzer string storage string scorer string - predictor string } // stubScorer stands in wherever a Profile's scorer has to be real rather than // nil, because something is composed over it. type stubScorer struct{} -func (stubScorer) Score(_ context.Context, _ entity.Batch) (float64, error) { return 0.5, nil } +func (stubScorer) Score(_ context.Context, _ entity.Batch, _ entity.SpeculationPathSet) (float64, error) { + return 0.5, nil +} // profileRecording returns a Profile whose every factory records the queue name // it receives into rec and returns a nil implementation. Nil is fine: these @@ -75,10 +74,6 @@ func profileRecording(rec *recorder) Profile { rec.scorer = c.QueueName return stubScorer{}, nil }), - Predictor: predictorFunc(func(c predictor.Config) (predictor.Predictor, error) { - rec.predictor = c.QueueName - return nil, nil - }), } } @@ -118,23 +113,20 @@ func TestProfilesForwardQueueNameToFactories(t *testing.T) { require.NoError(t, err) _, err = profiles.ScorerFactory().For(scorer.Config{QueueName: tt.queue}) require.NoError(t, err) - _, err = profiles.PredictorFactory().For(predictor.Config{QueueName: tt.queue}) - require.NoError(t, err) assert.Equal(t, tt.queue, rec.changeProvider) assert.Equal(t, tt.queue, rec.buildRunner) assert.Equal(t, tt.queue, rec.analyzer) assert.Equal(t, tt.queue, rec.storage) assert.Equal(t, tt.queue, rec.scorer) - assert.Equal(t, tt.queue, rec.predictor) }) } } -// The speculator is composed from the profile's predictor, which is itself -// composed over the profile's scorer. Each has to be asked for at the queue the -// one above it was asked for, or an implementation is built for the wrong one. -func TestWithSpeculatorResolvesPredictorAtSameQueue(t *testing.T) { +// The speculator is composed from the profile's scorer. It has to be asked for +// at the queue the speculator was asked for, or an implementation is built for +// the wrong one. +func TestWithSpeculatorResolvesScorerAtSameQueue(t *testing.T) { var rec recorder profile := withSpeculator(profileRecording(&rec), defaultBuildBudget) profiles := Profiles{defaultProfile: profile} @@ -142,39 +134,16 @@ func TestWithSpeculatorResolvesPredictorAtSameQueue(t *testing.T) { spec, err := profiles.SpeculatorFactory().For(speculator.Config{QueueName: "unlisted-queue"}) require.NoError(t, err) assert.NotNil(t, spec) - assert.Equal(t, "unlisted-queue", rec.predictor) -} - -func TestWithPredictorResolvesScorerAtSameQueue(t *testing.T) { - var rec recorder - profile := withPredictor(profileRecording(&rec), predictorConfig{}, tally.NoopScope) - - pred, err := profile.Predictor.For(predictor.Config{QueueName: "unlisted-queue"}) - require.NoError(t, err) - assert.NotNil(t, pred) assert.Equal(t, "unlisted-queue", rec.scorer) } -// Resolving either level can fail where reading a struct field could not, and -// the failure must surface rather than yielding something built over a nil. -func TestWithSpeculatorPropagatesPredictorError(t *testing.T) { - sentinel := errors.New("predictor unavailable") +func TestWithSpeculatorPropagatesScorerError(t *testing.T) { + sentinel := errors.New("scorer unavailable") profile := withSpeculator(Profile{ - Predictor: predictorFunc(func(predictor.Config) (predictor.Predictor, error) { return nil, sentinel }), + Scorer: scorerFunc(func(scorer.Config) (scorer.Scorer, error) { return nil, sentinel }), }, defaultBuildBudget) spec, err := profile.Speculator.For(speculator.Config{QueueName: "any-queue"}) require.ErrorIs(t, err, sentinel) assert.Nil(t, spec) } - -func TestWithPredictorPropagatesScorerError(t *testing.T) { - sentinel := errors.New("scorer unavailable") - profile := withPredictor(Profile{ - Scorer: scorerFunc(func(scorer.Config) (scorer.Scorer, error) { return nil, sentinel }), - }, predictorConfig{}, tally.NoopScope) - - pred, err := profile.Predictor.For(predictor.Config{QueueName: "any-queue"}) - require.ErrorIs(t, err, sentinel) - assert.Nil(t, pred) -} diff --git a/submitqueue/extension/speculation/generator/bestfirst/README.md b/submitqueue/extension/speculation/generator/bestfirst/README.md index fc458d118..858f0ff87 100644 --- a/submitqueue/extension/speculation/generator/bestfirst/README.md +++ b/submitqueue/extension/speculation/generator/bestfirst/README.md @@ -6,12 +6,12 @@ The [best-first speculation path generation RFC](../../../../../doc/rfc/submitqu ## Behavior -- `Generate` takes the queue's live batches and path sets as one snapshot, predicts each unique unresolved direct dependency once against that dependency's own path set, fixes assumptions for resolved dependencies, calculates each head's best score, and seeds the global heap with every eligible head's best-path candidate. Each head's remaining paths wait in that head's own lazy stream, whose flips are worked out only when the head is first handed out. +- `Generate` takes the queue's live batches and path sets as one snapshot, scores each unique unresolved direct dependency once against that dependency's own path set, fixes assumptions for resolved dependencies, calculates each head's best score, and seeds the global heap with every eligible head's best-path candidate. Each head's remaining paths wait in that head's own lazy stream, whose flips are worked out only when the head is first handed out. - `Next` removes the highest-ranked candidate, advances only that head's stream, constructs that candidate's complete path, and returns it. Pulling long enough returns every path exactly once in non-increasing score order. - Ranking scores are sums of log probabilities, avoiding underflow while preserving probability order. They are meaningful only within the run that produced them. - Exact ties prefer fewer flips; head ID then decides between heads (the cross-head heap holds one candidate per head), and taken flip indexes decide within a head. -- A dependency counts as resolved only once it is terminal. Merging and cancelling are both still in progress and either can end the other way, so both stay open questions here. How much a merging or cancelling dependency is worth is a predictor price, not a fact the search hard-codes. -- A dependency that cannot be priced — the predictor call failed, the probability was not in `[0, 1]`, or the snapshot never carried the batch — is treated as very likely to succeed rather than ending the run. One unusable number costs its own estimate, never the queue's whole set of candidates. A batch missing from the snapshot is never handed to the predictor at all: it would resolve to a zero batch belonging to no queue. +- A dependency counts as resolved only once it is terminal. Merging and cancelling are both still in progress and either can end the other way, so both stay open questions here. How much a merging or cancelling dependency is worth is a scorer price, not a fact the search hard-codes. +- A dependency that cannot be priced — the scorer call failed, the probability was not in `[0, 1]`, or the snapshot never carried the batch — is treated as very likely to succeed rather than ending the run. One unusable number costs its own estimate, never the queue's whole set of candidates. A batch missing from the snapshot is never handed to the scorer at all: it would resolve to a zero batch belonging to no queue. - The snapshot must contain every batch a head's direct dependencies reference, carry unique non-empty batch IDs, and give no head an empty, duplicate, or self dependency. That is the caller's precondition, not something checked here: a malformed snapshot yields undefined candidates rather than an error. The behavior is covered by `bestfirst_test.go`. diff --git a/submitqueue/extension/speculation/speculator/standard/README.md b/submitqueue/extension/speculation/speculator/standard/README.md index c797b2ebe..1325f081b 100644 --- a/submitqueue/extension/speculation/speculator/standard/README.md +++ b/submitqueue/extension/speculation/speculator/standard/README.md @@ -6,7 +6,7 @@ Each run it considers candidate paths in descending order of their probability o When the budget runs out, everything below the cut waits for a later run. That is safe because the propose-side cannot invent a batch verdict: the speculate controller still decides merge from the persisted paths, including complete coverage of unsettled dependencies. -Both halves are swappable. The ranking is the `Generator`'s: the default `bestfirst` asks the queue's predictor for each unresolved dependency's probability of reaching Succeeded, then ranks paths by the probability that all their assumptions hold. The budget policy is the `Allocator`'s: the default `sticky` fills only free slots and never preempts, where a preempting allocator would cancel a low-value in-flight path to fund a better one. +Both halves are swappable. The ranking is the `Generator`'s: the default `bestfirst` asks the queue's scorer for each unresolved dependency's probability of reaching Succeeded, then ranks paths by the probability that all their assumptions hold. The budget policy is the `Allocator`'s: the default `sticky` fills only free slots and never preempts, where a preempting allocator would cancel a low-value in-flight path to fund a better one. `standard` itself decides nothing — it connects the `Generator`'s stream to the `Allocator` — so changing prioritization or budget behavior means swapping a part, not writing a new `Speculator`. From 84ca524695741dce1fb10238a0d730683e68e2fc Mon Sep 17 00:00:00 2001 From: Preetam Dwivedi Date: Fri, 11 Sep 2026 12:57:52 -0700 Subject: [PATCH 7/8] fix(speculation): reject ranking-layer heuristic fields Buckets, components, and combine on the evidence scorer used to be ignored, so a migrated heuristic block ranked every batch at the default flat price. Fail those configs, replace base wholesale in tests, and construct evidence at profile build. --- .../submitqueue/orchestrator/server/config.go | 3 ++ .../orchestrator/server/config_test.go | 29 +++++++++++++++++++ .../orchestrator/server/profiles.go | 6 +++- 3 files changed, 37 insertions(+), 1 deletion(-) diff --git a/service/submitqueue/orchestrator/server/config.go b/service/submitqueue/orchestrator/server/config.go index 4ac098208..f5ef97f8e 100644 --- a/service/submitqueue/orchestrator/server/config.go +++ b/service/submitqueue/orchestrator/server/config.go @@ -583,6 +583,9 @@ func (s *scorerConfig) normalizeRanking(where string, fillBase bool) error { if s.Type != scorerTypeEvidence { return fmt.Errorf("%s: scorer type %q belongs under base, not at the ranking layer", where, s.Type) } + if len(s.Buckets) > 0 || len(s.Components) > 0 || s.Combine != "" { + return fmt.Errorf("%s: buckets, components, and combine belong under base", where) + } if err := validateFactors(where, s.Factors); err != nil { return err } diff --git a/service/submitqueue/orchestrator/server/config_test.go b/service/submitqueue/orchestrator/server/config_test.go index d120166c1..5bc2551ac 100644 --- a/service/submitqueue/orchestrator/server/config_test.go +++ b/service/submitqueue/orchestrator/server/config_test.go @@ -659,6 +659,10 @@ func TestLoadProfilesConfig_RejectsBadScorers(t *testing.T) { }{ {name: "unknown scorer type", contents: "defaults:\n scorer: {type: vibes}\n"}, {name: "top-level heuristic", contents: "defaults:\n scorer: {type: heuristic}\n"}, + {name: "top-level buckets without type", contents: "defaults:\n scorer:\n buckets: [{min: 0, max: 10, score: 0.9}]\n"}, + {name: "buckets next to evidence", contents: "defaults:\n scorer:\n type: evidence\n buckets: [{min: 0, max: 10, score: 0.9}]\n"}, + {name: "top-level combine", contents: "defaults:\n scorer:\n combine: avg\n"}, + {name: "queue overlay buckets without type", contents: "defaults: {}\nqueues:\n - name: q\n scorer:\n buckets: [{min: 0, max: 10, score: 0.9}]\n"}, {name: "composite with no components", contents: "defaults:\n scorer:\n base: {type: composite}\n"}, {name: "unknown combine", contents: "defaults:\n scorer:\n base:\n type: composite\n combine: median\n components: {a: {type: heuristic}}\n"}, {name: "score out of range", contents: "defaults:\n scorer:\n base:\n type: heuristic\n buckets: [{min: 0, max: 10, score: 2.0}]\n"}, @@ -734,3 +738,28 @@ func TestLoadProfilesConfig_QueueScorerFactorsOverlayDefaults(t *testing.T) { defaults := factorsFrom(cfg.Defaults.Scorer) assert.Equal(t, 10.0, defaults.PathPassed) } + +func TestLoadProfilesConfig_QueueScorerBaseReplacesDefaultBase(t *testing.T) { + cfg, err := loadProfilesConfig(writeProfiles(t, ""+ + "defaults:\n"+ + " scorer:\n"+ + " factors: {pathPassed: 10}\n"+ + " base:\n"+ + " type: heuristic\n"+ + " buckets: [{min: 0, max: 1000, score: 0.4}]\n"+ + "queues:\n"+ + " - name: q\n"+ + " scorer:\n"+ + " base:\n"+ + " type: heuristic\n"+ + " buckets: [{min: 0, max: 1000, score: 0.9}]\n")) + require.NoError(t, err) + + resolved := cfg.resolve(cfg.Queues[0]).Scorer + require.NotNil(t, resolved.Base) + assert.Equal(t, 0.9, resolved.Base.Buckets[0].Score) + assert.Equal(t, 10.0, factorsFrom(resolved).PathPassed) + + require.NotNil(t, cfg.Defaults.Scorer.Base) + assert.Equal(t, 0.4, cfg.Defaults.Scorer.Base.Buckets[0].Score) +} diff --git a/service/submitqueue/orchestrator/server/profiles.go b/service/submitqueue/orchestrator/server/profiles.go index 00cb07a99..2030e9d04 100644 --- a/service/submitqueue/orchestrator/server/profiles.go +++ b/service/submitqueue/orchestrator/server/profiles.go @@ -330,9 +330,13 @@ func (b *profileBuilder) newScorerFactory(cfg scorerConfig, where string) (score if cfg.Base == nil { return nil, fmt.Errorf("%s: evidence scorer needs a base", where) } - if _, err := b.buildScorer(scorer.Config{}, *cfg.Base, where, "scorer.base"); err != nil { + base, err := b.buildScorer(scorer.Config{}, *cfg.Base, where, "scorer.base") + if err != nil { return nil, err } + if _, err := evidence.New(scorer.Config{}, base, factorsFrom(cfg), b.scope.SubScope("scorer")); err != nil { + return nil, fmt.Errorf("%s: %w", where, err) + } return scorerFunc(func(c scorer.Config) (scorer.Scorer, error) { base, err := b.buildScorer(c, *cfg.Base, where, "scorer.base") if err != nil { From 7068a1b5687fc1cc407c024f65722a7d83ba9129 Mon Sep 17 00:00:00 2001 From: Preetam Dwivedi Date: Fri, 11 Sep 2026 14:08:43 -0700 Subject: [PATCH 8/8] fix(speculation): reject ranking fields on content scorers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary ### Why? A factors map or nested base under heuristic/composite used to load and then be ignored, so a misplaced pathPassed looked configured while ranking stayed at 1. ### What? Fail those configs in normalizeContent, including composite components and a queue overlay base. ## Test Plan ✅ `go test ./service/submitqueue/orchestrator/server/ -run TestLoadProfilesConfig_RejectsBadScorers` --- service/submitqueue/orchestrator/server/config.go | 3 +++ service/submitqueue/orchestrator/server/config_test.go | 4 ++++ 2 files changed, 7 insertions(+) diff --git a/service/submitqueue/orchestrator/server/config.go b/service/submitqueue/orchestrator/server/config.go index f5ef97f8e..d36fd9d7c 100644 --- a/service/submitqueue/orchestrator/server/config.go +++ b/service/submitqueue/orchestrator/server/config.go @@ -616,6 +616,9 @@ func validateFactors(where string, factors map[string]float64) error { } func (s *scorerConfig) normalizeContent(where string) error { + if len(s.Factors) > 0 || s.Base != nil { + return fmt.Errorf("%s: factors and base belong on the ranking scorer, not under base", where) + } if s.Type == "" { s.Type = scorerTypeHeuristic } diff --git a/service/submitqueue/orchestrator/server/config_test.go b/service/submitqueue/orchestrator/server/config_test.go index 5bc2551ac..82aa8175e 100644 --- a/service/submitqueue/orchestrator/server/config_test.go +++ b/service/submitqueue/orchestrator/server/config_test.go @@ -667,6 +667,10 @@ func TestLoadProfilesConfig_RejectsBadScorers(t *testing.T) { {name: "unknown combine", contents: "defaults:\n scorer:\n base:\n type: composite\n combine: median\n components: {a: {type: heuristic}}\n"}, {name: "score out of range", contents: "defaults:\n scorer:\n base:\n type: heuristic\n buckets: [{min: 0, max: 10, score: 2.0}]\n"}, {name: "inverted bucket", contents: "defaults:\n scorer:\n base:\n type: heuristic\n buckets: [{min: 10, max: 1, score: 0.5}]\n"}, + {name: "factors under heuristic base", contents: "defaults:\n scorer:\n base:\n type: heuristic\n factors: {pathPassed: 10}\n"}, + {name: "nested base under heuristic", contents: "defaults:\n scorer:\n base:\n type: heuristic\n base: {type: heuristic}\n"}, + {name: "factors on a composite component", contents: "defaults:\n scorer:\n base:\n type: composite\n components:\n a:\n type: heuristic\n factors: {pathPassed: 10}\n"}, + {name: "factors under queue overlay base", contents: "defaults: {}\nqueues:\n - name: q\n scorer:\n base:\n type: heuristic\n factors: {pathPassed: 10}\n"}, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) {