From e9d90d28724a14d7f88a89c55a42b10bc54beb0e Mon Sep 17 00:00:00 2001 From: Preetam Dwivedi Date: Mon, 7 Sep 2026 16:53:52 -0700 Subject: [PATCH 1/4] feat(speculation): add outcome predictor implementation ## Summary ### Why? A scorer prices change content, but speculation also needs a separate contract for revising that price with evidence observed during a run. Keeping the concerns separate avoids adding path data that every content scorer would discard. ### What? Add the `predictor.Predictor` contract and an evidence implementation that converts the scorer probability to odds, applies factors for passed and failed all-succeed paths plus merging and cancelling states, and converts the result back to a probability. Include generated mocks and unit coverage for neutral factors, compounding evidence, path filtering, bounds, validation, and scorer failures. ## Test Plan - `bazel test //submitqueue/extension/speculation/predictor/...` - `make check-gazelle` --- .../speculation/predictor/BUILD.bazel | 9 + .../predictor/evidence/BUILD.bazel | 29 ++ .../predictor/evidence/evidence.go | 183 +++++++++++++ .../predictor/evidence/evidence_test.go | 257 ++++++++++++++++++ .../speculation/predictor/mock/BUILD.bazel | 13 + .../predictor/mock/predictor_mock.go | 97 +++++++ .../speculation/predictor/predictor.go | 56 ++++ 7 files changed, 644 insertions(+) create mode 100644 submitqueue/extension/speculation/predictor/BUILD.bazel create mode 100644 submitqueue/extension/speculation/predictor/evidence/BUILD.bazel create mode 100644 submitqueue/extension/speculation/predictor/evidence/evidence.go create mode 100644 submitqueue/extension/speculation/predictor/evidence/evidence_test.go create mode 100644 submitqueue/extension/speculation/predictor/mock/BUILD.bazel create mode 100644 submitqueue/extension/speculation/predictor/mock/predictor_mock.go create mode 100644 submitqueue/extension/speculation/predictor/predictor.go diff --git a/submitqueue/extension/speculation/predictor/BUILD.bazel b/submitqueue/extension/speculation/predictor/BUILD.bazel new file mode 100644 index 00000000..fc689f24 --- /dev/null +++ b/submitqueue/extension/speculation/predictor/BUILD.bazel @@ -0,0 +1,9 @@ +load("@rules_go//go:def.bzl", "go_library") + +go_library( + name = "go_default_library", + srcs = ["predictor.go"], + importpath = "github.com/uber/submitqueue/submitqueue/extension/speculation/predictor", + visibility = ["//visibility:public"], + deps = ["//submitqueue/entity:go_default_library"], +) diff --git a/submitqueue/extension/speculation/predictor/evidence/BUILD.bazel b/submitqueue/extension/speculation/predictor/evidence/BUILD.bazel new file mode 100644 index 00000000..f0e2eca9 --- /dev/null +++ b/submitqueue/extension/speculation/predictor/evidence/BUILD.bazel @@ -0,0 +1,29 @@ +load("@rules_go//go:def.bzl", "go_library", "go_test") + +go_library( + name = "go_default_library", + srcs = ["evidence.go"], + importpath = "github.com/uber/submitqueue/submitqueue/extension/speculation/predictor/evidence", + visibility = ["//visibility:public"], + deps = [ + "//platform/metrics:go_default_library", + "//submitqueue/entity:go_default_library", + "//submitqueue/extension/speculation/predictor:go_default_library", + "//submitqueue/extension/speculation/scorer:go_default_library", + "@com_github_uber_go_tally//:go_default_library", + ], +) + +go_test( + name = "go_default_test", + srcs = ["evidence_test.go"], + embed = [":go_default_library"], + deps = [ + "//submitqueue/entity:go_default_library", + "//submitqueue/extension/speculation/predictor:go_default_library", + "//submitqueue/extension/speculation/scorer: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/predictor/evidence/evidence.go b/submitqueue/extension/speculation/predictor/evidence/evidence.go new file mode 100644 index 00000000..75927a65 --- /dev/null +++ b/submitqueue/extension/speculation/predictor/evidence/evidence.go @@ -0,0 +1,183 @@ +// Copyright (c) 2025 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Package evidence revises a Scorer's price by multiplying its odds by one +// factor per piece of evidence about the batch's progress. +// +// Odds rather than the probability itself, because a factor then means the same +// thing wherever it applies and the result cannot leave [0, 1]. Written as logs +// and summed, the same arithmetic is a logistic regression, which is what lets +// hand-written factors later be replaced by fitted ones without changing the +// form. See doc/rfc/submitqueue/outcome-predictor.md. +package evidence + +import ( + "fmt" + "math" + + "context" + + "github.com/uber-go/tally" + "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/submitqueue/entity" + "github.com/uber/submitqueue/submitqueue/extension/speculation/predictor" + "github.com/uber/submitqueue/submitqueue/extension/speculation/scorer" +) + +// Factors are the odds multipliers, one per piece of evidence. A factor of 1 +// leaves the price alone. Named fields rather than a keyed map, so an evidence +// name that does not exist fails to compile instead of being ignored. +type Factors struct { + // PathPassed applies once when a build has passed on the batch's + // all-succeed path. + PathPassed float64 + // PathFailed applies once per failed all-succeed path, compounding. + PathFailed float64 + // Merging applies while the batch is merging. + Merging float64 + // Cancelling applies while the batch is cancelling. + Cancelling float64 +} + +// AllOnes is the neutral set: the prediction is the scorer's price. +func AllOnes() Factors { + return Factors{PathPassed: 1, PathFailed: 1, Merging: 1, Cancelling: 1} +} + +// epsilon bounds the price away from 0 and 1, which have no finite odds. +// Without it a certain scorer could never be revised by any evidence — and +// certainty about an unfinished batch is the scorer overstating what it sees. +const epsilon = 1e-6 + +// evidence is a predictor.Predictor that revises a scorer's price. +type evidence struct { + // cfg is the per-queue identity this predictor was built for. + cfg predictor.Config + // base prices the batch's change; its price is what the factors revise. + base scorer.Scorer + // factors are the odds multipliers applied to that price. + factors Factors + // scope is the tally scope for emitting metrics. + scope tally.Scope +} + +// New creates an evidence predictor bound to the queue named in cfg, revising +// base's price by factors. +// +// It returns an error rather than panic on a nil base or a non-positive factor: +// configuration rejects those today, but the fitted-factor file loader planned +// in doc/rfc/submitqueue/outcome-predictor.md bypasses configuration entirely, +// and on that path this check is the only guard. +func New(cfg predictor.Config, base scorer.Scorer, factors Factors, scope tally.Scope) (predictor.Predictor, error) { + if base == nil { + return nil, fmt.Errorf("evidence.New: base must not be nil") + } + for name, factor := range map[string]float64{ + "PathPassed": factors.PathPassed, + "PathFailed": factors.PathFailed, + "Merging": factors.Merging, + "Cancelling": factors.Cancelling, + } { + // Zero would pin the prediction to 0 and negative has no meaning as a + // multiplier on odds. + if !(factor > 0) { + return nil, fmt.Errorf("evidence.New: factor %s must be positive, got %v", name, factor) + } + } + return &evidence{cfg: cfg, base: base, factors: factors, scope: scope}, nil +} + +// Predict prices the batch's change through the base scorer, then multiplies +// the odds of that price by one factor per piece of evidence. +func (r *evidence) Predict(ctx context.Context, batch entity.Batch, paths entity.SpeculationPathSet) (ret predictor.Probability, retErr error) { + op := metrics.Begin(r.scope, "predict", metrics.FastLatencyBuckets) + defer func() { op.Complete(retErr) }() + + price, err := r.base.Score(ctx, batch) + if err != nil { + return 0, err + } + // A price that is not a probability is a broken scorer, not a low opinion of + // the batch. Saying so leaves the caller to fall back on its own default, + // where clamping would hand back a number that looks deliberate. + if !(price >= 0 && price <= 1) { + return 0, fmt.Errorf("base scorer returned %v, which is not a probability", price) + } + + odds := oddsOf(math.Min(math.Max(price, epsilon), 1-epsilon)) + if hasPassedAllSucceedPath(paths) { + odds *= r.factors.PathPassed + } + odds *= math.Pow(r.factors.PathFailed, float64(countFailed(paths))) + switch batch.State { + case entity.BatchStateMerging: + odds *= r.factors.Merging + case entity.BatchStateCancelling: + odds *= r.factors.Cancelling + } + return probabilityOf(odds), nil +} + +// oddsOf converts a probability to odds. p is bounded away from 1, so this is +// finite. +func oddsOf(p float64) float64 { + return p / (1 - p) +} + +// probabilityOf converts odds back to a probability. Overflowed odds read as +// certainty rather than the NaN the division would produce. +func probabilityOf(odds float64) predictor.Probability { + if math.IsInf(odds, 1) { + return 1 + } + return predictor.Probability(odds / (1 + odds)) +} + +// hasPassedAllSucceedPath reports a passed build on the batch's all-succeed +// path. Only that path counts: one built without a dependency's changes says +// nothing about a candidate that assumes the dependency lands. +func hasPassedAllSucceedPath(paths entity.SpeculationPathSet) bool { + for _, entry := range paths.Paths { + if entry.Status != entity.SpeculationPathStatusPassed { + continue + } + if assumesAllSucceed(entry.Path) { + return true + } + } + return false +} + +// assumesAllSucceed reports whether every dependency is assumed to succeed. +func assumesAllSucceed(path entity.SpeculationPath) bool { + for _, dep := range path.Dependencies { + if dep.Assumption != entity.DependencyAssumptionSucceeds { + return false + } + } + return true +} + +// countFailed counts failed builds on the batch's all-succeed path; each one +// compounds. Flip-subset failures are ignored: they were built under different +// assumptions, the same filter PathPassed uses. +func countFailed(paths entity.SpeculationPathSet) int { + failed := 0 + for _, entry := range paths.Paths { + if entry.Status == entity.SpeculationPathStatusFailed && assumesAllSucceed(entry.Path) { + failed++ + } + } + return failed +} diff --git a/submitqueue/extension/speculation/predictor/evidence/evidence_test.go b/submitqueue/extension/speculation/predictor/evidence/evidence_test.go new file mode 100644 index 00000000..25033cba --- /dev/null +++ b/submitqueue/extension/speculation/predictor/evidence/evidence_test.go @@ -0,0 +1,257 @@ +// Copyright (c) 2025 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package evidence + +import ( + "context" + "fmt" + "math" + "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/speculation/predictor" + "github.com/uber/submitqueue/submitqueue/extension/speculation/scorer" +) + +// testCfg is the per-queue identity used by every case in this file. +var testCfg = predictor.Config{QueueName: "test-queue"} + +// fixedScorer always returns the same price. +type fixedScorer struct{ price float64 } + +func (f fixedScorer) Score(_ context.Context, _ entity.Batch) (float64, error) { + return f.price, nil +} + +// errorScorer always fails. +type errorScorer struct{} + +func (errorScorer) Score(_ context.Context, _ entity.Batch) (float64, error) { + return 0, fmt.Errorf("scorer failed") +} + +// pathSet builds a set whose entries carry the given statuses, every path +// assuming all of its dependencies succeed. +func pathSet(statuses ...entity.SpeculationPathStatus) entity.SpeculationPathSet { + set := entity.SpeculationPathSet{Queue: "q", Head: "q/batch/1"} + for i, status := range statuses { + set.Paths = append(set.Paths, entity.SpeculationPathEntry{ + ID: fmt.Sprintf("path-%d", i), + Status: status, + Path: entity.SpeculationPath{ + Head: "q/batch/1", + Dependencies: []entity.PathDependency{{Batch: "q/batch/0", Assumption: entity.DependencyAssumptionSucceeds}}, + }, + }) + } + return set +} + +// predict runs one prediction with all-neutral factors except those overridden. +func predict(t *testing.T, price float64, factors Factors, batch entity.Batch, paths entity.SpeculationPathSet) float64 { + t.Helper() + p, err := New(testCfg, fixedScorer{price: price}, factors, tally.NoopScope) + require.NoError(t, err) + got, err := p.Predict(context.Background(), batch, paths) + require.NoError(t, err) + return float64(got) +} + +func TestPredict_NeutralFactorsReturnTheScorersPrice(t *testing.T) { + for _, price := range []float64{0.01, 0.25, 0.5, 0.6, 0.9, 0.99} { + t.Run(fmt.Sprintf("price %v", price), func(t *testing.T) { + got := predict(t, price, AllOnes(), entity.Batch{}, pathSet(entity.SpeculationPathStatusPassed, entity.SpeculationPathStatusFailed)) + assert.InDelta(t, price, got, 1e-9) + }) + } +} + +func TestPredict_AppliesOneFactorPerEvidence(t *testing.T) { + // 0.5 has odds of exactly 1, so the resulting odds are the factor itself and + // the expected probability is factor/(1+factor). + tests := []struct { + name string + factors Factors + batch entity.Batch + paths entity.SpeculationPathSet + want float64 + }{ + { + name: "a passed path", + factors: Factors{PathPassed: 9, PathFailed: 1, Merging: 1, Cancelling: 1}, + paths: pathSet(entity.SpeculationPathStatusPassed), + want: 0.9, + }, + { + name: "no passed path leaves the price alone", + factors: Factors{PathPassed: 9, PathFailed: 1, Merging: 1, Cancelling: 1}, + paths: pathSet(entity.SpeculationPathStatusBuilding), + want: 0.5, + }, + { + name: "one failed path", + factors: Factors{PathPassed: 1, PathFailed: 0.25, Merging: 1, Cancelling: 1}, + paths: pathSet(entity.SpeculationPathStatusFailed), + want: 0.2, + }, + { + name: "failed paths compound", + factors: Factors{PathPassed: 1, PathFailed: 0.5, Merging: 1, Cancelling: 1}, + paths: pathSet(entity.SpeculationPathStatusFailed, entity.SpeculationPathStatusFailed), + want: 0.2, + }, + { + name: "merging", + factors: Factors{PathPassed: 1, PathFailed: 1, Merging: 19, Cancelling: 1}, + batch: entity.Batch{State: entity.BatchStateMerging}, + want: 0.95, + }, + { + name: "cancelling", + factors: Factors{PathPassed: 1, PathFailed: 1, Merging: 1, Cancelling: 0.25}, + batch: entity.Batch{State: entity.BatchStateCancelling}, + want: 0.2, + }, + { + name: "a state with no factor leaves the price alone", + factors: Factors{PathPassed: 1, PathFailed: 1, Merging: 19, Cancelling: 0.25}, + batch: entity.Batch{State: entity.BatchStateSpeculating}, + want: 0.5, + }, + { + name: "evidence compounds across kinds", + factors: Factors{PathPassed: 4, PathFailed: 1, Merging: 3, Cancelling: 1}, + batch: entity.Batch{State: entity.BatchStateMerging}, + paths: pathSet(entity.SpeculationPathStatusPassed), + want: 0.923076923, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.InDelta(t, tt.want, predict(t, 0.5, tt.factors, tt.batch, tt.paths), 1e-9) + }) + } +} + +// A path built without one of its dependencies proves nothing about a candidate +// that assumes the dependency lands, which is what stacking on this batch means. +func TestPredict_IgnoresAPassedPathThatAssumesAFailure(t *testing.T) { + paths := pathSet(entity.SpeculationPathStatusPassed) + paths.Paths[0].Path.Dependencies[0].Assumption = entity.DependencyAssumptionFails + + factors := Factors{PathPassed: 9, PathFailed: 1, Merging: 1, Cancelling: 1} + assert.InDelta(t, 0.5, predict(t, 0.5, factors, entity.Batch{}, paths), 1e-9) +} + +// A failed flip-subset must not drag down a green all-succeed build: it was +// built under different assumptions, the same filter PathPassed uses. +func TestPredict_IgnoresAFailedPathThatAssumesAFailure(t *testing.T) { + paths := pathSet(entity.SpeculationPathStatusPassed, entity.SpeculationPathStatusFailed) + paths.Paths[1].Path.Dependencies[0].Assumption = entity.DependencyAssumptionFails + + factors := Factors{PathPassed: 9, PathFailed: 0.25, Merging: 1, Cancelling: 1} + assert.InDelta(t, 0.9, predict(t, 0.5, factors, entity.Batch{}, paths), 1e-9) +} + +func TestPredict_APathWithNoDependenciesCounts(t *testing.T) { + paths := pathSet(entity.SpeculationPathStatusPassed) + paths.Paths[0].Path.Dependencies = nil + + factors := Factors{PathPassed: 9, PathFailed: 1, Merging: 1, Cancelling: 1} + assert.InDelta(t, 0.9, predict(t, 0.5, factors, entity.Batch{}, paths), 1e-9) +} + +func TestPredict_AFailedPathWithNoDependenciesCounts(t *testing.T) { + paths := pathSet(entity.SpeculationPathStatusFailed) + paths.Paths[0].Path.Dependencies = nil + + factors := Factors{PathPassed: 1, PathFailed: 0.25, Merging: 1, Cancelling: 1} + assert.InDelta(t, 0.2, predict(t, 0.5, factors, entity.Batch{}, paths), 1e-9) +} + +func TestPredict_AnEmptyPathSetIsNoEvidence(t *testing.T) { + factors := Factors{PathPassed: 9, PathFailed: 0.1, Merging: 1, Cancelling: 1} + assert.InDelta(t, 0.5, predict(t, 0.5, factors, entity.Batch{}, entity.SpeculationPathSet{}), 1e-9) +} + +// A scorer certain either way still has to be movable, or no evidence could ever +// revise a price the scorer had no business being certain about. +func TestPredict_CertainPricesStayInRangeAndStillMove(t *testing.T) { + tests := []struct { + name string + price float64 + factor float64 + wantAbove float64 + wantBelow float64 + }{ + {name: "certain success, evidence against", price: 1, factor: 0.5, wantAbove: 0.99, wantBelow: 1}, + {name: "certain failure, evidence for", price: 0, factor: 2, wantAbove: 0, wantBelow: 0.01}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + factors := AllOnes() + factors.PathPassed = tt.factor + got := predict(t, tt.price, factors, entity.Batch{}, pathSet(entity.SpeculationPathStatusPassed)) + assert.Greater(t, got, tt.wantAbove) + assert.Less(t, got, tt.wantBelow) + }) + } +} + +func TestPredict_RejectsAPriceThatIsNotAProbability(t *testing.T) { + for _, price := range []float64{-0.1, 1.5, math.NaN()} { + t.Run(fmt.Sprintf("price %v", price), func(t *testing.T) { + p, err := New(testCfg, fixedScorer{price: price}, AllOnes(), tally.NoopScope) + require.NoError(t, err) + _, err = p.Predict(context.Background(), entity.Batch{}, entity.SpeculationPathSet{}) + require.Error(t, err) + }) + } +} + +func TestPredict_PropagatesAScorerError(t *testing.T) { + p, err := New(testCfg, errorScorer{}, AllOnes(), tally.NoopScope) + require.NoError(t, err) + _, err = p.Predict(context.Background(), entity.Batch{}, entity.SpeculationPathSet{}) + require.Error(t, err) +} + +func TestNew_RejectsUnusableConstruction(t *testing.T) { + zeroed := AllOnes() + zeroed.Merging = 0 + negative := AllOnes() + negative.PathFailed = -1 + + tests := []struct { + name string + base scorer.Scorer + factors Factors + }{ + {name: "nil base", base: nil, factors: AllOnes()}, + {name: "zero factor", base: fixedScorer{price: 0.5}, factors: zeroed}, + {name: "negative factor", base: fixedScorer{price: 0.5}, factors: negative}, + {name: "unset factors", base: fixedScorer{price: 0.5}, factors: Factors{}}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + p, err := New(testCfg, tt.base, tt.factors, tally.NoopScope) + require.Error(t, err) + assert.Nil(t, p) + }) + } +} diff --git a/submitqueue/extension/speculation/predictor/mock/BUILD.bazel b/submitqueue/extension/speculation/predictor/mock/BUILD.bazel new file mode 100644 index 00000000..fd84517d --- /dev/null +++ b/submitqueue/extension/speculation/predictor/mock/BUILD.bazel @@ -0,0 +1,13 @@ +load("@rules_go//go:def.bzl", "go_library") + +go_library( + name = "go_default_library", + srcs = ["predictor_mock.go"], + importpath = "github.com/uber/submitqueue/submitqueue/extension/speculation/predictor/mock", + visibility = ["//visibility:public"], + deps = [ + "//submitqueue/entity:go_default_library", + "//submitqueue/extension/speculation/predictor:go_default_library", + "@org_uber_go_mock//gomock:go_default_library", + ], +) diff --git a/submitqueue/extension/speculation/predictor/mock/predictor_mock.go b/submitqueue/extension/speculation/predictor/mock/predictor_mock.go new file mode 100644 index 00000000..01088820 --- /dev/null +++ b/submitqueue/extension/speculation/predictor/mock/predictor_mock.go @@ -0,0 +1,97 @@ +// Code generated by MockGen. DO NOT EDIT. +// Source: predictor.go +// +// Generated by this command: +// +// mockgen -source=predictor.go -destination=mock/predictor_mock.go -package=mock +// + +// Package mock is a generated GoMock package. +package mock + +import ( + context "context" + reflect "reflect" + + entity "github.com/uber/submitqueue/submitqueue/entity" + predictor "github.com/uber/submitqueue/submitqueue/extension/speculation/predictor" + gomock "go.uber.org/mock/gomock" +) + +// MockPredictor is a mock of Predictor interface. +type MockPredictor struct { + ctrl *gomock.Controller + recorder *MockPredictorMockRecorder + isgomock struct{} +} + +// MockPredictorMockRecorder is the mock recorder for MockPredictor. +type MockPredictorMockRecorder struct { + mock *MockPredictor +} + +// NewMockPredictor creates a new mock instance. +func NewMockPredictor(ctrl *gomock.Controller) *MockPredictor { + mock := &MockPredictor{ctrl: ctrl} + mock.recorder = &MockPredictorMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *MockPredictor) EXPECT() *MockPredictorMockRecorder { + return m.recorder +} + +// Predict mocks base method. +func (m *MockPredictor) Predict(ctx context.Context, batch entity.Batch, paths entity.SpeculationPathSet) (predictor.Probability, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Predict", ctx, batch, paths) + ret0, _ := ret[0].(predictor.Probability) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// Predict indicates an expected call of Predict. +func (mr *MockPredictorMockRecorder) Predict(ctx, batch, paths any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Predict", reflect.TypeOf((*MockPredictor)(nil).Predict), ctx, batch, paths) +} + +// MockFactory is a mock of Factory interface. +type MockFactory struct { + ctrl *gomock.Controller + recorder *MockFactoryMockRecorder + isgomock struct{} +} + +// MockFactoryMockRecorder is the mock recorder for MockFactory. +type MockFactoryMockRecorder struct { + mock *MockFactory +} + +// NewMockFactory creates a new mock instance. +func NewMockFactory(ctrl *gomock.Controller) *MockFactory { + mock := &MockFactory{ctrl: ctrl} + mock.recorder = &MockFactoryMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *MockFactory) EXPECT() *MockFactoryMockRecorder { + return m.recorder +} + +// For mocks base method. +func (m *MockFactory) For(cfg predictor.Config) (predictor.Predictor, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "For", cfg) + ret0, _ := ret[0].(predictor.Predictor) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// For indicates an expected call of For. +func (mr *MockFactoryMockRecorder) For(cfg any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "For", reflect.TypeOf((*MockFactory)(nil).For), cfg) +} diff --git a/submitqueue/extension/speculation/predictor/predictor.go b/submitqueue/extension/speculation/predictor/predictor.go new file mode 100644 index 00000000..66b57c01 --- /dev/null +++ b/submitqueue/extension/speculation/predictor/predictor.go @@ -0,0 +1,56 @@ +// Copyright (c) 2025 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Package predictor defines how likely a batch is to succeed, given both what +// it changes and what has happened to it so far. A Scorer prices the change; a +// Predictor is built over one and revises its price with the batch's observed +// progress. +package predictor + +//go:generate mockgen -source=predictor.go -destination=mock/predictor_mock.go -package=mock + +import ( + "context" + + "github.com/uber/submitqueue/submitqueue/entity" +) + +// Probability is how likely an outcome is, from 0.0 to 1.0. +type Probability float64 + +// Predictor estimates a batch's final outcome. +type Predictor interface { + // Predict returns how likely the batch is to reach Succeeded with its + // changes landed. A passing build is necessary but not sufficient. + // + // paths is the batch's own build progress, zero-valued for a batch nothing + // has speculated on. Callers may predict every batch a queue waits on, so + // anything expensive belongs behind the implementation's own cache. + Predict(ctx context.Context, batch entity.Batch, paths entity.SpeculationPathSet) (Probability, error) +} + +// Config carries the per-queue identity handed to a Factory. The system knows +// only the queue name; everything an implementation needs is injected at +// construction by the integrator. +type Config struct { + // QueueName identifies the queue this Predictor serves. + QueueName string +} + +// Factory builds the Predictor for a queue. Implementations inject what they +// need at construction, including the Scorer whose price they revise. +type Factory interface { + // For returns the Predictor for the given queue. + For(cfg Config) (Predictor, error) +} From 32d92336e788edff6a52ff9867365e5f632796c0 Mon Sep 17 00:00:00 2001 From: Preetam Dwivedi Date: Mon, 7 Sep 2026 17:12:09 -0700 Subject: [PATCH 2/4] refactor(speculation): express prediction in factor terms MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary ### Why? The evidence implementation still described the model as logistic regression over odds and referenced fitting work that the RFC no longer proposes. That made the code's vocabulary diverge from the factor contract exposed to operators. ### What? Combine the applicable evidence factors first and revise the scorer price with the equivalent bounded formula. Remove stale fitting rationale and keep implementation and tests in the RFC's scorer-price and factor terminology without changing behavior. ## Test Plan - ✅ `./tool/bazel test //submitqueue/extension/speculation/predictor/...` --- .../predictor/evidence/evidence.go | 60 +++++++------------ .../predictor/evidence/evidence_test.go | 3 +- 2 files changed, 21 insertions(+), 42 deletions(-) diff --git a/submitqueue/extension/speculation/predictor/evidence/evidence.go b/submitqueue/extension/speculation/predictor/evidence/evidence.go index 75927a65..6b4e38c2 100644 --- a/submitqueue/extension/speculation/predictor/evidence/evidence.go +++ b/submitqueue/extension/speculation/predictor/evidence/evidence.go @@ -12,14 +12,8 @@ // See the License for the specific language governing permissions and // limitations under the License. -// Package evidence revises a Scorer's price by multiplying its odds by one -// factor per piece of evidence about the batch's progress. -// -// Odds rather than the probability itself, because a factor then means the same -// thing wherever it applies and the result cannot leave [0, 1]. Written as logs -// and summed, the same arithmetic is a logistic regression, which is what lets -// hand-written factors later be replaced by fitted ones without changing the -// form. See doc/rfc/submitqueue/outcome-predictor.md. +// Package evidence revises a Scorer's price with factors for observed batch +// progress. See doc/rfc/submitqueue/outcome-predictor.md. package evidence import ( @@ -35,9 +29,8 @@ import ( "github.com/uber/submitqueue/submitqueue/extension/speculation/scorer" ) -// Factors are the odds multipliers, one per piece of evidence. A factor of 1 -// leaves the price alone. Named fields rather than a keyed map, so an evidence -// name that does not exist fails to compile instead of being ignored. +// Factors revise the scorer's price, one per piece of evidence. A factor of 1 +// leaves the price alone. Named fields make unknown evidence fail to compile. type Factors struct { // PathPassed applies once when a build has passed on the batch's // all-succeed path. @@ -55,9 +48,7 @@ func AllOnes() Factors { return Factors{PathPassed: 1, PathFailed: 1, Merging: 1, Cancelling: 1} } -// epsilon bounds the price away from 0 and 1, which have no finite odds. -// Without it a certain scorer could never be revised by any evidence — and -// certainty about an unfinished batch is the scorer overstating what it sees. +// epsilon keeps exact certainty revisable while remaining close to the scorer. const epsilon = 1e-6 // evidence is a predictor.Predictor that revises a scorer's price. @@ -66,7 +57,7 @@ type evidence struct { cfg predictor.Config // base prices the batch's change; its price is what the factors revise. base scorer.Scorer - // factors are the odds multipliers applied to that price. + // factors revise the scorer's price with observed evidence. factors Factors // scope is the tally scope for emitting metrics. scope tally.Scope @@ -75,10 +66,7 @@ type evidence struct { // New creates an evidence predictor bound to the queue named in cfg, revising // base's price by factors. // -// It returns an error rather than panic on a nil base or a non-positive factor: -// configuration rejects those today, but the fitted-factor file loader planned -// in doc/rfc/submitqueue/outcome-predictor.md bypasses configuration entirely, -// and on that path this check is the only guard. +// It rejects a nil base and non-positive factors. func New(cfg predictor.Config, base scorer.Scorer, factors Factors, scope tally.Scope) (predictor.Predictor, error) { if base == nil { return nil, fmt.Errorf("evidence.New: base must not be nil") @@ -89,8 +77,8 @@ func New(cfg predictor.Config, base scorer.Scorer, factors Factors, scope tally. "Merging": factors.Merging, "Cancelling": factors.Cancelling, } { - // Zero would pin the prediction to 0 and negative has no meaning as a - // multiplier on odds. + // Zero would permanently pin matching batches to 0; negatives cannot + // represent either direction in the factor contract. if !(factor > 0) { return nil, fmt.Errorf("evidence.New: factor %s must be positive, got %v", name, factor) } @@ -98,8 +86,8 @@ func New(cfg predictor.Config, base scorer.Scorer, factors Factors, scope tally. return &evidence{cfg: cfg, base: base, factors: factors, scope: scope}, nil } -// Predict prices the batch's change through the base scorer, then multiplies -// the odds of that price by one factor per piece of evidence. +// Predict prices the batch's change, combines its evidence factors, and revises +// the scorer's price with the result. func (r *evidence) Predict(ctx context.Context, batch entity.Batch, paths entity.SpeculationPathSet) (ret predictor.Probability, retErr error) { op := metrics.Begin(r.scope, "predict", metrics.FastLatencyBuckets) defer func() { op.Complete(retErr) }() @@ -115,33 +103,25 @@ func (r *evidence) Predict(ctx context.Context, batch entity.Batch, paths entity return 0, fmt.Errorf("base scorer returned %v, which is not a probability", price) } - odds := oddsOf(math.Min(math.Max(price, epsilon), 1-epsilon)) + factor := math.Pow(r.factors.PathFailed, float64(countFailed(paths))) if hasPassedAllSucceedPath(paths) { - odds *= r.factors.PathPassed + factor *= r.factors.PathPassed } - odds *= math.Pow(r.factors.PathFailed, float64(countFailed(paths))) switch batch.State { case entity.BatchStateMerging: - odds *= r.factors.Merging + factor *= r.factors.Merging case entity.BatchStateCancelling: - odds *= r.factors.Cancelling + factor *= r.factors.Cancelling } - return probabilityOf(odds), nil -} - -// oddsOf converts a probability to odds. p is bounded away from 1, so this is -// finite. -func oddsOf(p float64) float64 { - return p / (1 - p) + return revise(math.Min(math.Max(price, epsilon), 1-epsilon), factor), nil } -// probabilityOf converts odds back to a probability. Overflowed odds read as -// certainty rather than the NaN the division would produce. -func probabilityOf(odds float64) predictor.Probability { - if math.IsInf(odds, 1) { +// revise applies the combined factor while keeping the result a probability. +func revise(price, factor float64) predictor.Probability { + if math.IsInf(factor, 1) { return 1 } - return predictor.Probability(odds / (1 + odds)) + return predictor.Probability(price * factor / (1 - price + price*factor)) } // hasPassedAllSucceedPath reports a passed build on the batch's all-succeed diff --git a/submitqueue/extension/speculation/predictor/evidence/evidence_test.go b/submitqueue/extension/speculation/predictor/evidence/evidence_test.go index 25033cba..18c16ae5 100644 --- a/submitqueue/extension/speculation/predictor/evidence/evidence_test.go +++ b/submitqueue/extension/speculation/predictor/evidence/evidence_test.go @@ -82,8 +82,7 @@ func TestPredict_NeutralFactorsReturnTheScorersPrice(t *testing.T) { } func TestPredict_AppliesOneFactorPerEvidence(t *testing.T) { - // 0.5 has odds of exactly 1, so the resulting odds are the factor itself and - // the expected probability is factor/(1+factor). + // At scorer price 0.5, factor f revises the price to f/(1+f). tests := []struct { name string factors Factors From 0fc7ffebb01d4d95dfe293b20d96dd2c056a91e9 Mon Sep 17 00:00:00 2001 From: Preetam Dwivedi Date: Mon, 7 Sep 2026 17:28:02 -0700 Subject: [PATCH 3/4] fix(speculation): preserve predictor factor contract MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary ### Why? Neutral prediction changed exact scorer prices at 0 and 1, non-finite factors could create certainty, and failed-path compounding relied on duplicate logical paths that a valid path set cannot contain. ### What? Return the scorer price unchanged for a neutral combined factor, reject non-finite configured factors, keep revised outputs strictly inside the probability range, and apply failed all-succeeds evidence at most once. Extend tests for exact endpoints, large factors, and infinite-factor rejection. ## Test Plan - ✅ `./tool/bazel test //submitqueue/extension/speculation/predictor/...` --- .../predictor/evidence/evidence.go | 33 +++++++++++-------- .../predictor/evidence/evidence_test.go | 22 ++++++++----- 2 files changed, 33 insertions(+), 22 deletions(-) diff --git a/submitqueue/extension/speculation/predictor/evidence/evidence.go b/submitqueue/extension/speculation/predictor/evidence/evidence.go index 6b4e38c2..52eac912 100644 --- a/submitqueue/extension/speculation/predictor/evidence/evidence.go +++ b/submitqueue/extension/speculation/predictor/evidence/evidence.go @@ -35,7 +35,7 @@ type Factors struct { // PathPassed applies once when a build has passed on the batch's // all-succeed path. PathPassed float64 - // PathFailed applies once per failed all-succeed path, compounding. + // PathFailed applies once when the all-succeed path has failed. PathFailed float64 // Merging applies while the batch is merging. Merging float64 @@ -66,7 +66,7 @@ type evidence struct { // New creates an evidence predictor bound to the queue named in cfg, revising // base's price by factors. // -// It rejects a nil base and non-positive factors. +// It rejects a nil base and factors that are non-finite or not positive. func New(cfg predictor.Config, base scorer.Scorer, factors Factors, scope tally.Scope) (predictor.Predictor, error) { if base == nil { return nil, fmt.Errorf("evidence.New: base must not be nil") @@ -79,8 +79,8 @@ func New(cfg predictor.Config, base scorer.Scorer, factors Factors, scope tally. } { // Zero would permanently pin matching batches to 0; negatives cannot // represent either direction in the factor contract. - if !(factor > 0) { - return nil, fmt.Errorf("evidence.New: factor %s must be positive, got %v", name, factor) + if !(factor > 0) || math.IsInf(factor, 0) { + return nil, fmt.Errorf("evidence.New: factor %s must be finite and positive, got %v", name, factor) } } return &evidence{cfg: cfg, base: base, factors: factors, scope: scope}, nil @@ -103,25 +103,32 @@ func (r *evidence) Predict(ctx context.Context, batch entity.Batch, paths entity return 0, fmt.Errorf("base scorer returned %v, which is not a probability", price) } - factor := math.Pow(r.factors.PathFailed, float64(countFailed(paths))) + factor := 1.0 if hasPassedAllSucceedPath(paths) { factor *= r.factors.PathPassed } + if hasFailedAllSucceedPath(paths) { + factor *= r.factors.PathFailed + } switch batch.State { case entity.BatchStateMerging: factor *= r.factors.Merging case entity.BatchStateCancelling: factor *= r.factors.Cancelling } + if factor == 1 { + return predictor.Probability(price), nil + } return revise(math.Min(math.Max(price, epsilon), 1-epsilon), factor), nil } // revise applies the combined factor while keeping the result a probability. func revise(price, factor float64) predictor.Probability { if math.IsInf(factor, 1) { - return 1 + return 1 - epsilon } - return predictor.Probability(price * factor / (1 - price + price*factor)) + revised := price * factor / (1 - price + price*factor) + return predictor.Probability(math.Min(math.Max(revised, epsilon), 1-epsilon)) } // hasPassedAllSucceedPath reports a passed build on the batch's all-succeed @@ -149,15 +156,13 @@ func assumesAllSucceed(path entity.SpeculationPath) bool { return true } -// countFailed counts failed builds on the batch's all-succeed path; each one -// compounds. Flip-subset failures are ignored: they were built under different -// assumptions, the same filter PathPassed uses. -func countFailed(paths entity.SpeculationPathSet) int { - failed := 0 +// hasFailedAllSucceedPath reports a failed build on the batch's all-succeed +// path. Flip-subset failures were built under different assumptions. +func hasFailedAllSucceedPath(paths entity.SpeculationPathSet) bool { for _, entry := range paths.Paths { if entry.Status == entity.SpeculationPathStatusFailed && assumesAllSucceed(entry.Path) { - failed++ + return true } } - return failed + return false } diff --git a/submitqueue/extension/speculation/predictor/evidence/evidence_test.go b/submitqueue/extension/speculation/predictor/evidence/evidence_test.go index 18c16ae5..a1222fdc 100644 --- a/submitqueue/extension/speculation/predictor/evidence/evidence_test.go +++ b/submitqueue/extension/speculation/predictor/evidence/evidence_test.go @@ -73,10 +73,10 @@ func predict(t *testing.T, price float64, factors Factors, batch entity.Batch, p } func TestPredict_NeutralFactorsReturnTheScorersPrice(t *testing.T) { - for _, price := range []float64{0.01, 0.25, 0.5, 0.6, 0.9, 0.99} { + for _, price := range []float64{0, 0.01, 0.25, 0.5, 0.6, 0.9, 0.99, 1} { t.Run(fmt.Sprintf("price %v", price), func(t *testing.T) { got := predict(t, price, AllOnes(), entity.Batch{}, pathSet(entity.SpeculationPathStatusPassed, entity.SpeculationPathStatusFailed)) - assert.InDelta(t, price, got, 1e-9) + assert.Equal(t, price, got) }) } } @@ -108,12 +108,6 @@ func TestPredict_AppliesOneFactorPerEvidence(t *testing.T) { paths: pathSet(entity.SpeculationPathStatusFailed), want: 0.2, }, - { - name: "failed paths compound", - factors: Factors{PathPassed: 1, PathFailed: 0.5, Merging: 1, Cancelling: 1}, - paths: pathSet(entity.SpeculationPathStatusFailed, entity.SpeculationPathStatusFailed), - want: 0.2, - }, { name: "merging", factors: Factors{PathPassed: 1, PathFailed: 1, Merging: 19, Cancelling: 1}, @@ -212,6 +206,15 @@ func TestPredict_CertainPricesStayInRangeAndStillMove(t *testing.T) { } } +func TestPredict_LargeFactorsDoNotProduceCertainty(t *testing.T) { + factors := AllOnes() + factors.PathPassed = math.MaxFloat64 + + got := predict(t, 0.5, factors, entity.Batch{}, pathSet(entity.SpeculationPathStatusPassed)) + assert.Greater(t, got, 0.99) + assert.Less(t, got, 1.0) +} + func TestPredict_RejectsAPriceThatIsNotAProbability(t *testing.T) { for _, price := range []float64{-0.1, 1.5, math.NaN()} { t.Run(fmt.Sprintf("price %v", price), func(t *testing.T) { @@ -235,6 +238,8 @@ func TestNew_RejectsUnusableConstruction(t *testing.T) { zeroed.Merging = 0 negative := AllOnes() negative.PathFailed = -1 + infinite := AllOnes() + infinite.PathPassed = math.Inf(1) tests := []struct { name string @@ -244,6 +249,7 @@ func TestNew_RejectsUnusableConstruction(t *testing.T) { {name: "nil base", base: nil, factors: AllOnes()}, {name: "zero factor", base: fixedScorer{price: 0.5}, factors: zeroed}, {name: "negative factor", base: fixedScorer{price: 0.5}, factors: negative}, + {name: "infinite factor", base: fixedScorer{price: 0.5}, factors: infinite}, {name: "unset factors", base: fixedScorer{price: 0.5}, factors: Factors{}}, } for _, tt := range tests { From 1068958178ec70904237329965c38d84711d4cc8 Mon Sep 17 00:00:00 2001 From: Preetam Dwivedi Date: Mon, 7 Sep 2026 17:56:27 -0700 Subject: [PATCH 4/4] docs(speculation): document predictor extension Add the predictor package guide and clarify how it composes over the scorer without adding path evidence to the scorer contract. --- .../extension/speculation/predictor/README.md | 17 +++++++++++++++++ .../extension/speculation/scorer/README.md | 2 ++ 2 files changed, 19 insertions(+) create mode 100644 submitqueue/extension/speculation/predictor/README.md diff --git a/submitqueue/extension/speculation/predictor/README.md b/submitqueue/extension/speculation/predictor/README.md new file mode 100644 index 00000000..b32c0851 --- /dev/null +++ b/submitqueue/extension/speculation/predictor/README.md @@ -0,0 +1,17 @@ +# predictor + +A `Predictor` returns how likely a batch is to reach `Succeeded` with its changes landed, given both what it changes and what this speculate run has already observed. It is built over the queue's `Scorer`, which prices the change from content signals; the predictor revises that price with path-set evidence and batch state. + +`Predict` is handed the batch identity and that batch's own `SpeculationPathSet` — zero-valued when nothing has speculated on it yet. Callers may predict every unresolved dependency a queue waits on, so anything expensive belongs behind the implementation's own cache. + +Like the other extensions, a `Predictor` is selected **per queue** by the wiring layer through the `Config` (queue name) and `Factory` interface. The default `standard` `Speculator` composes its `Generator` over the queue's predictor, which in turn composes over the queue's scorer. + +See [doc/rfc/submitqueue/outcome-predictor.md](../../../../doc/rfc/submitqueue/outcome-predictor.md) for the factor contract, evidence rules, and configuration shape. + +## Implementations + +**`evidence`** revises the scorer's price with YAML-configured factors for `pathPassed`, `pathFailed`, `merging`, and `cancelling`. A factor of `1` leaves the scorer's price alone; every factor defaults to `1` until someone sets one. Only paths that assume every dependency succeeds count as path evidence. + +## Adding a backend + +Create a package under `predictor//` whose `New(...)` returns a `predictor.Predictor`, injecting whatever it needs at construction — typically the queue's `Scorer`, factor configuration, and a metrics scope. Do not add a `Config` or `Factory` implementation here; per-queue routing and the factory adapter live in the wiring layer. diff --git a/submitqueue/extension/speculation/scorer/README.md b/submitqueue/extension/speculation/scorer/README.md index d8b63693..4d70ed15 100644 --- a/submitqueue/extension/speculation/scorer/README.md +++ b/submitqueue/extension/speculation/scorer/README.md @@ -4,6 +4,8 @@ A `Scorer` returns the probability that a batch ultimately succeeds — reaches Callers may score every batch a queue is waiting on, so implementations should be cheap. A speculation run scores each batch at most once, but it does not carry results across runs; anything expensive belongs behind the implementation's own cache. +The default speculation pipeline does not rank on the scorer directly. The queue's `Predictor` is built over its `Scorer` and revises the scorer's price with path-set evidence before `bestfirst` ranks paths. The scorer still prices only the change; it does not see path sets. + Like the other extensions, a `Scorer` is selected **per queue** by the wiring layer through the `Config` (queue name) and `Factory` interface. ## Implementations