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/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/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..52eac912 --- /dev/null +++ b/submitqueue/extension/speculation/predictor/evidence/evidence.go @@ -0,0 +1,168 @@ +// 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 with factors for observed batch +// progress. 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 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. + PathPassed float64 + // PathFailed applies once when the all-succeed path has failed. + 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 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. +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 revise the scorer's price with observed evidence. + 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 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") + } + for name, factor := range map[string]float64{ + "PathPassed": factors.PathPassed, + "PathFailed": factors.PathFailed, + "Merging": factors.Merging, + "Cancelling": factors.Cancelling, + } { + // 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 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 +} + +// 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) }() + + 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) + } + + 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 - epsilon + } + 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 +// 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 +} + +// 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) { + return true + } + } + return false +} 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..a1222fdc --- /dev/null +++ b/submitqueue/extension/speculation/predictor/evidence/evidence_test.go @@ -0,0 +1,262 @@ +// 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, 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.Equal(t, price, got) + }) + } +} + +func TestPredict_AppliesOneFactorPerEvidence(t *testing.T) { + // At scorer price 0.5, factor f revises the price to f/(1+f). + 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: "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_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) { + 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 + infinite := AllOnes() + infinite.PathPassed = math.Inf(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: "infinite factor", base: fixedScorer{price: 0.5}, factors: infinite}, + {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) +} 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