Skip to content

feat(speculate): write the speculation tree in shadow mode #353

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
behinddwalls wants to merge 2 commits into
base: main
Choose a base branch
from
+2,446 −526
Open

feat(speculate): write the speculation tree in shadow mode #353

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
61ff6cc
feat(orchestrator): add queue-wide prioritize stage
behinddwalls Jul 9, 2026
e30249e
feat(speculate): write the speculation tree in shadow mode
behinddwalls Jul 10, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 12 additions & 0 deletions service/submitqueue/orchestrator/server/BUILD.bazel
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,17 @@ go_library(
"//submitqueue/extension/scorer/composite:go_default_library",
"//submitqueue/extension/scorer/fake:go_default_library",
"//submitqueue/extension/scorer/heuristic:go_default_library",
"//submitqueue/extension/speculation/dependencylimit:go_default_library",
"//submitqueue/extension/speculation/dependencylimit/static:go_default_library",
"//submitqueue/extension/speculation/enumerator:go_default_library",
"//submitqueue/extension/speculation/enumerator/chain:go_default_library",
"//submitqueue/extension/speculation/pathscorer:go_default_library",
"//submitqueue/extension/speculation/pathscorer/probability:go_default_library",
"//submitqueue/extension/speculation/prioritizationlimit/static:go_default_library",
"//submitqueue/extension/speculation/prioritizer:go_default_library",
"//submitqueue/extension/speculation/prioritizer/sticky:go_default_library",
"//submitqueue/extension/speculation/selector:go_default_library",
"//submitqueue/extension/speculation/selector/all:go_default_library",
"//submitqueue/extension/storage:go_default_library",
"//submitqueue/extension/storage/mysql:go_default_library",
"//submitqueue/extension/validator/fake:go_default_library",
Expand All @@ -54,6 +65,7 @@ go_library(
"//submitqueue/orchestrator/controller/merge:go_default_library",
"//submitqueue/orchestrator/controller/mergeconflictsignal:go_default_library",
"//submitqueue/orchestrator/controller/mergesignal:go_default_library",
"//submitqueue/orchestrator/controller/prioritize:go_default_library",
"//submitqueue/orchestrator/controller/score:go_default_library",
"//submitqueue/orchestrator/controller/speculate:go_default_library",
"//submitqueue/orchestrator/controller/start:go_default_library",
Expand Down
133 changes: 118 additions & 15 deletions service/submitqueue/orchestrator/server/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,6 +61,17 @@ import (
"github.com/uber/submitqueue/submitqueue/extension/scorer/composite"
scorerfake "github.com/uber/submitqueue/submitqueue/extension/scorer/fake"
"github.com/uber/submitqueue/submitqueue/extension/scorer/heuristic"
"github.com/uber/submitqueue/submitqueue/extension/speculation/dependencylimit"
dependencylimitstatic "github.com/uber/submitqueue/submitqueue/extension/speculation/dependencylimit/static"
"github.com/uber/submitqueue/submitqueue/extension/speculation/enumerator"
"github.com/uber/submitqueue/submitqueue/extension/speculation/enumerator/chain"
"github.com/uber/submitqueue/submitqueue/extension/speculation/pathscorer"
pathscorerprobability "github.com/uber/submitqueue/submitqueue/extension/speculation/pathscorer/probability"
prioritizationlimitstatic "github.com/uber/submitqueue/submitqueue/extension/speculation/prioritizationlimit/static"
"github.com/uber/submitqueue/submitqueue/extension/speculation/prioritizer"
"github.com/uber/submitqueue/submitqueue/extension/speculation/prioritizer/sticky"
"github.com/uber/submitqueue/submitqueue/extension/speculation/selector"
selectorall "github.com/uber/submitqueue/submitqueue/extension/speculation/selector/all"
"github.com/uber/submitqueue/submitqueue/extension/storage"
mysqlstorage "github.com/uber/submitqueue/submitqueue/extension/storage/mysql"
validatorfake "github.com/uber/submitqueue/submitqueue/extension/validator/fake"
Expand All @@ -74,6 +85,7 @@ import (
"github.com/uber/submitqueue/submitqueue/orchestrator/controller/merge"
"github.com/uber/submitqueue/submitqueue/orchestrator/controller/mergeconflictsignal"
"github.com/uber/submitqueue/submitqueue/orchestrator/controller/mergesignal"
"github.com/uber/submitqueue/submitqueue/orchestrator/controller/prioritize"
"github.com/uber/submitqueue/submitqueue/orchestrator/controller/score"
"github.com/uber/submitqueue/submitqueue/orchestrator/controller/speculate"
"github.com/uber/submitqueue/submitqueue/orchestrator/controller/start"
Expand Down Expand Up @@ -234,7 +246,7 @@ func run() error {
// back to a baseline profile for queues without an explicit entry. This is
// the single place queue topology is known; the extension packages stay
// queue-agnostic.
queues, err := newQueueRegistry(logger, scope, changeset.New(store.GetRequestStore(), store.GetChangeStore()))
queues, err := newQueueRegistry(logger, scope, changeset.New(store.GetRequestStore(), store.GetChangeStore()), store.GetBatchStore())
if err != nil {
return fmt.Errorf("failed to build queue registry: %w", err)
}
Expand All @@ -244,13 +256,18 @@ func run() error {
brf := buildRunnerFactory{queues}
scf := scorerFactory{queues}
cof := analyzerFactory{queues}
prf := prioritizerFactory{queues}
enf := enumeratorFactory{queues}
ssf := speculationScorerFactory{queues}
slf := speculationSelectorFactory{queues}
dlf := dependencyLimitFactory{queues}

// Register controllers
primaryCount, err := registerPrimaryControllers(primaryConsumer, logger.Sugar(), scope, registry, cpf, brf, scf, cof, cnt, store)
primaryCount, err := registerPrimaryControllers(primaryConsumer, logger.Sugar(), scope, registry, cpf, brf, scf, cof, prf, enf, ssf, slf, dlf, cnt, store)
if err != nil {
return err
}
dlqCount, err := registerDLQControllers(dlqConsumer, logger.Sugar(), scope, store)
dlqCount, err := registerDLQControllers(dlqConsumer, logger.Sugar(), scope, registry, store)
if err != nil {
return err
}
Expand Down Expand Up @@ -380,6 +397,7 @@ func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRe
{topickey.TopicKeyBatch, "batch", "orchestrator-batch"},
{topickey.TopicKeyScore, "score", "orchestrator-score"},
{topickey.TopicKeySpeculate, "speculate", "orchestrator-speculate"},
{topickey.TopicKeyPrioritize, "prioritize", "orchestrator-prioritize"},
{topickey.TopicKeyBuild, "build", "orchestrator-build"},
{topickey.TopicKeyBuildSignal, "buildsignal", "orchestrator-buildsignal"},
{topickey.TopicKeyMerge, "merge", "orchestrator-merge"},
Expand Down Expand Up @@ -471,6 +489,13 @@ func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRe
// merge-conflict-check queue (⇢); runway performs the merge attempt and
// publishes the result to merge-conflict-check-signal, which mergeconflictsignal
// consumes before fanning the request out to batch.
//
// prioritize sits alongside this per-batch flow rather than in its line: it
// is queue-wide, not batch-scoped. Its message carries only a queue name; on
// each invocation it loads every Speculating batch's speculation tree for
// that queue, ranks the queue-wide candidate paths against the queue's build
// budget, applies the resulting decisions, and republishes to build for any
// path newly (or still) cleared to run.

// TODO(wiring abstraction): queueExtensions + queueRegistry currently live here
// as example-local wiring. Evaluate promoting them into a defined abstraction in
Expand All @@ -489,10 +514,15 @@ func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRe
// read as "for this queue, here are its scorer, analyzer, change provider, …", and lets
// a queue profile start from a baseline and override only what differs.
type queueExtensions struct {
changeProvider changeprovider.ChangeProvider
buildRunner buildrunner.BuildRunner
scorer scorer.Scorer
analyzer conflict.Analyzer
changeProvider changeprovider.ChangeProvider
buildRunner buildrunner.BuildRunner
scorer scorer.Scorer
analyzer conflict.Analyzer
prioritizer prioritizer.Prioritizer
enumerator enumerator.Enumerator
speculationScorer pathscorer.Scorer
speculationSelector selector.Selector
dependencyLimit dependencylimit.DependencyLimit
}

// queueRegistry maps a queue name to its extensions, falling back to a default
Expand Down Expand Up @@ -538,7 +568,37 @@ func (f analyzerFactory) For(cfg conflict.Config) (conflict.Analyzer, error) {
return f.reg.get(cfg.QueueName).analyzer, nil
}

func registerPrimaryControllers(c consumer.Consumer, logger *zap.SugaredLogger, scope tally.Scope, registry consumer.TopicRegistry, cpf changeprovider.Factory, brf buildrunner.Factory, scf scorer.Factory, cof conflict.Factory, cnt counter.Counter, store storage.Storage) (int, error) {
type prioritizerFactory struct{ reg queueRegistry }

func (f prioritizerFactory) For(cfg prioritizer.Config) (prioritizer.Prioritizer, error) {
return f.reg.get(cfg.QueueName).prioritizer, nil
}

type enumeratorFactory struct{ reg queueRegistry }

func (f enumeratorFactory) For(cfg enumerator.Config) (enumerator.Enumerator, error) {
return f.reg.get(cfg.QueueName).enumerator, nil
}

type speculationScorerFactory struct{ reg queueRegistry }

func (f speculationScorerFactory) For(cfg pathscorer.Config) (pathscorer.Scorer, error) {
return f.reg.get(cfg.QueueName).speculationScorer, nil
}

type speculationSelectorFactory struct{ reg queueRegistry }

func (f speculationSelectorFactory) For(cfg selector.Config) (selector.Selector, error) {
return f.reg.get(cfg.QueueName).speculationSelector, nil
}

type dependencyLimitFactory struct{ reg queueRegistry }

func (f dependencyLimitFactory) For(cfg dependencylimit.Config) (dependencylimit.DependencyLimit, error) {
return f.reg.get(cfg.QueueName).dependencyLimit, nil
}

func registerPrimaryControllers(c consumer.Consumer, logger *zap.SugaredLogger, scope tally.Scope, registry consumer.TopicRegistry, cpf changeprovider.Factory, brf buildrunner.Factory, scf scorer.Factory, cof conflict.Factory, prf prioritizer.Factory, enf enumerator.Factory, ssf pathscorer.Factory, slf selector.Factory, dlf dependencylimit.Factory, cnt counter.Counter, store storage.Storage) (int, error) {
var count int
requestController := start.NewController(
logger,
Expand Down Expand Up @@ -628,6 +688,10 @@ func registerPrimaryControllers(c consumer.Consumer, logger *zap.SugaredLogger,
logger,
scope,
store,
enf,
ssf,
slf,
dlf,
registry,
topickey.TopicKeySpeculate,
"orchestrator-speculate",
Expand All @@ -637,6 +701,20 @@ func registerPrimaryControllers(c consumer.Consumer, logger *zap.SugaredLogger,
}
count++

prioritizeController := prioritize.NewController(
logger,
scope,
store,
prf,
registry,
topickey.TopicKeyPrioritize,
"orchestrator-prioritize",
)
if err := c.Register(prioritizeController); err != nil {
return count, fmt.Errorf("failed to register prioritize controller: %w", err)
}
count++

buildController := build.NewController(
logger,
scope,
Expand Down Expand Up @@ -712,7 +790,7 @@ func registerPrimaryControllers(c consumer.Consumer, logger *zap.SugaredLogger,
// registers them with the DLQ consumer. Each reconciler drives the affected
// request or batch into a terminal Error/Failed state so the gateway stops
// reporting it as stuck-in-progress.
func registerDLQControllers(c consumer.Consumer, logger *zap.SugaredLogger, scope tally.Scope, store storage.Storage) (int, error) {
func registerDLQControllers(c consumer.Consumer, logger *zap.SugaredLogger, scope tally.Scope, registry consumer.TopicRegistry, store storage.Storage) (int, error) {
dlqScope := scope.SubScope("dlq")
dlqRegs := []struct {
name string
Expand All @@ -725,6 +803,7 @@ func registerDLQControllers(c consumer.Consumer, logger *zap.SugaredLogger, scop
{"batch_dlq", dlq.NewDLQRequestController(logger, dlqScope, store, dlq.DecodeRequestID, dlq.TopicKey(topickey.TopicKeyBatch), "orchestrator-batch-dlq")},
{"score_dlq", dlq.NewDLQBatchController(logger, dlqScope, store, dlq.TopicKey(topickey.TopicKeyScore), "orchestrator-score-dlq")},
{"speculate_dlq", dlq.NewDLQBatchController(logger, dlqScope, store, dlq.TopicKey(topickey.TopicKeySpeculate), "orchestrator-speculate-dlq")},
{"prioritize_dlq", dlq.NewDLQQueueController(logger, dlqScope, registry, dlq.TopicKey(topickey.TopicKeyPrioritize), "orchestrator-prioritize-dlq")},
{"build_dlq", dlq.NewDLQBatchController(logger, dlqScope, store, dlq.TopicKey(topickey.TopicKeyBuild), "orchestrator-build-dlq")},
{"buildsignal_dlq", dlq.NewDLQBuildSignalController(logger, dlqScope, store, dlq.TopicKey(topickey.TopicKeyBuildSignal), "orchestrator-buildsignal-dlq")},
{"merge_dlq", dlq.NewDLQBatchController(logger, dlqScope, store, dlq.TopicKey(topickey.TopicKeyMerge), "orchestrator-merge-dlq")},
Expand Down Expand Up @@ -859,14 +938,24 @@ func newPhabChangeProvider(logger *zap.Logger, scope tally.Scope) (changeprovide
}), nil
}

// defaultPrioritizationLimit is the baseline queue-wide concurrent-build
// budget handed to the sticky prioritizer. It is a parity default —
// effectively admit-all — until per-queue budgets are configured.
const defaultPrioritizationLimit = 1000

// defaultDependencyLimit is the baseline cap on active dependencies a batch
// may speculate over. It is a parity default — effectively ungated — until
// per-queue limits are configured.
const defaultDependencyLimit = 1000

// newQueueRegistry builds the per-queue extension profiles for the example.
// Edge integrations (change provider) and the build
// runner form a shared baseline; each per-queue profile starts from that
// baseline and overrides only the extensions that differ — here the scorer and
// conflict analyzer. Queues without an explicit profile fall back to the
// baseline. This is the one place queue topology lives; extension packages stay
// queue-agnostic.
func newQueueRegistry(logger *zap.Logger, scope tally.Scope, resolver changeset.Resolver) (queueRegistry, error) {
func newQueueRegistry(logger *zap.Logger, scope tally.Scope, resolver changeset.Resolver, batchStore storage.BatchStore) (queueRegistry, error) {
cp, err := newChangeProvider(logger, scope)
if err != nil {
return queueRegistry{}, fmt.Errorf("failed to create change provider: %w", err)
Expand All @@ -880,16 +969,25 @@ func newQueueRegistry(logger *zap.Logger, scope tally.Scope, resolver changeset.

// Baseline profile: shared edge integrations + a fake build runner (every
// build succeeds unless a head URI carries a failure marker), plus permissive
// defaults for scorer and conflict. The build runner instance is shared by
// the build and buildsignal controllers (same profile, same instance) so a
// build's recorded outcome survives across their separate factory lookups.
// defaults for scorer, conflict, and prioritization. The build runner
// instance is shared by the build and buildsignal controllers (same
// profile, same instance) so a build's recorded outcome survives across
// their separate factory lookups.
//
// The scorer is wrapped by scorerfake so a change URI carrying
// "sq-fake=score-error" forces a scoring error end-to-end; it is a pure
// passthrough otherwise. The analyzer is wrapped by conflictfake with a nil
// predicate (passthrough) — swap the predicate (e.g. conflictfake.FailAlways)
// on a queue to exercise the analyzer error path, as e2e-conflict-error-queue
// below does.
// below does. The prioritizer is sticky over a static budget: it never
// preempts a running build and admits Selected candidates by score until
// defaultPrioritizationLimit concurrent builds are in flight.
//
// The speculation seams default to the single-chain parity policies:
// chain enumerates one path per batch (built on the full ordered
// dependency chain), probability scores paths from the batches'
// predicted-success probabilities and resolved outcomes, all promotes
// every candidate, and the dependency limit is effectively ungated.
base := queueExtensions{
changeProvider: cp,
buildRunner: buildfake.New(resolver),
Expand All @@ -900,7 +998,12 @@ func newQueueRegistry(logger *zap.Logger, scope tally.Scope, resolver changeset.
)),
// TODO: replace the delegate with a real analyzer (e.g. Tango target
// analysis). "all" serializes the queue conservatively.
analyzer: conflictfake.New(all.New(), nil),
analyzer: conflictfake.New(all.New(), nil),
prioritizer: sticky.New(prioritizationlimitstatic.New(defaultPrioritizationLimit)),
enumerator: chain.New(),
speculationScorer: pathscorerprobability.New(batchStore),
speculationSelector: selectorall.New(),
dependencyLimit: dependencylimitstatic.New(defaultDependencyLimit),
}

// test-queue: bucketed heuristic scorer; conservative (serialized) conflicts
Expand Down
8 changes: 8 additions & 0 deletions submitqueue/core/topickey/topickey.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,14 @@ const (
TopicKeyScore TopicKey = "score"
// TopicKeySpeculate is the pipeline stage where scored batches are published for speculation.
TopicKeySpeculate TopicKey = "speculate"
// TopicKeyPrioritize is the queue-wide reconcile stage that rations the
// build budget across every in-flight batch of a queue. Each message
// carries a QueueID; the consumer loads every Speculating batch's tree,
// runs the queue's Prioritizer over the candidate paths, applies the
// resulting decisions — promoting paths into the build budget, or
// cancelling in-flight paths a preemptive policy evicts — and republishes
// to TopicKeyBuild for any path cleared to run.
TopicKeyPrioritize TopicKey = "prioritize"
// TopicKeyBuild is the pipeline stage where speculated batches are published for builds.
TopicKeyBuild TopicKey = "build"
// TopicKeyBuildSignal is the polling stage for triggered builds. Each
Expand Down
2 changes: 2 additions & 0 deletions submitqueue/entity/BUILD.bazel
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@ go_library(
"land_request.go",
"merge_result.go",
"push_result.go",
"queue.go",
"queue_config.go",
"request.go",
"request_log.go",
Expand All @@ -36,6 +37,7 @@ go_test(
"build_test.go",
"cancel_request_test.go",
"land_request_test.go",
"queue_test.go",
"request_log_test.go",
"request_test.go",
"speculation_tree_test.go",
Expand Down
36 changes: 36 additions & 0 deletions submitqueue/entity/queue.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,36 @@
// 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 entity

import "encoding/json"

// QueueID is the queue-message payload for queue-scoped pipeline stages. It
// carries only the queue name.
type QueueID struct {
// Name is the merge-queue name the message targets.
Name string `json:"name"`
}

// ToBytes serializes the QueueID to JSON bytes for queue message payload.
func (q QueueID) ToBytes() ([]byte, error) {
return json.Marshal(q)
}

// QueueIDFromBytes deserializes a QueueID from JSON bytes.
func QueueIDFromBytes(data []byte) (QueueID, error) {
var qid QueueID
err := json.Unmarshal(data, &qid)
return qid, err
}
Loading
Loading