From d6273e51c5b85165a1cd18ffa5120d77c6b1f539 Mon Sep 17 00:00:00 2001 From: Preetam Dwivedi Date: Mon, 3 Aug 2026 10:03:00 -0700 Subject: [PATCH] feat(orchestrator): dispatch builds per speculation path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary ### Why? The build stage triggered one build per batch, on the batch's full dependency list. Speculation needs one build per *path*, on the subset of dependencies the path assumes will succeed — that subset is what lets a batch be verified before the batches ahead of it resolve. ### What? The dispatch controller now reads the head's path set and starts a build for every pending path, on that path's own base. It **only starts builds** — stopping them belongs to the poll loop (next commit) — and it **never writes the path set**, which stays the speculate run's single-writer state; what this stage knows is the build it started, recorded in per-build rows of its own. The write order is Trigger → Build record → write-once link (`entity.PathBuild`) → signal; each write makes the previous one reachable, and the link is the idempotency point for redeliveries. A lost link race publishes both builds' signals and acks — the poll loop keeps the one the link names. On redeliveries only, every live linked path gets its signal re-published, closing the crash window between link and signal without forking duplicate poll chains past the queue's dedup horizon. The build signal carries only the runner's build ID and partitions on it, so one slow build cannot block a head's other paths. Known gap: a crash between `Trigger` and the link orphans that build; the fix is an idempotency key on `BuildRunner.Trigger`, marked TODO. Also adds `submitqueue/core/publish` — the registry-lookup-and-send plumbing and the message-ID deduplication rule in one place. ## Test Plan ✅ `bazel test //submitqueue/orchestrator/controller/build/... //submitqueue/core/publish/...` — write order, the base holding only assumed-success dependencies, redelivery republish without rebuild, lost-race double signal, halted batch starting nothing, redelivery healing for live linked paths, and a pin that no path-set write of any kind happens. ✅ `make fmt`, `make gazelle`, `make mocks` # Conflicts: # submitqueue/orchestrator/controller/build/BUILD.bazel # submitqueue/orchestrator/controller/build/build.go # submitqueue/orchestrator/controller/build/build_test.go # Please enter the commit message for your changes. Lines starting # with '#' will be kept; you may remove them yourself if you want to. # An empty message aborts the commit. # # interactive rebase in progress; onto 0ecb87dd # Last command done (1 command done): # pick d870cb4c # feat(orchestrator): dispatch builds per speculation path # No commands remaining. # You are currently rebasing branch 'preetam/speculation-build-dispatch' on '0ecb87dd'. # # Changes to be committed: # new file: submitqueue/core/publish/BUILD.bazel # new file: submitqueue/core/publish/publish.go # new file: submitqueue/core/publish/publish_test.go # modified: submitqueue/orchestrator/controller/build/BUILD.bazel # modified: submitqueue/orchestrator/controller/build/build.go # modified: submitqueue/orchestrator/controller/build/build_test.go # --- submitqueue/core/publish/BUILD.bazel | 26 + submitqueue/core/publish/publish.go | 67 ++ submitqueue/core/publish/publish_test.go | 80 +++ .../orchestrator/controller/build/BUILD.bazel | 5 +- .../orchestrator/controller/build/build.go | 312 ++++++-- .../controller/build/build_test.go | 666 +++++++++++------- 6 files changed, 800 insertions(+), 356 deletions(-) create mode 100644 submitqueue/core/publish/BUILD.bazel create mode 100644 submitqueue/core/publish/publish.go create mode 100644 submitqueue/core/publish/publish_test.go diff --git a/submitqueue/core/publish/BUILD.bazel b/submitqueue/core/publish/BUILD.bazel new file mode 100644 index 00000000..4e25bee4 --- /dev/null +++ b/submitqueue/core/publish/BUILD.bazel @@ -0,0 +1,26 @@ +load("@rules_go//go:def.bzl", "go_library", "go_test") + +go_library( + name = "go_default_library", + srcs = ["publish.go"], + importpath = "github.com/uber/submitqueue/submitqueue/core/publish", + visibility = ["//visibility:public"], + deps = [ + "//platform/base/messagequeue:go_default_library", + "//platform/consumer:go_default_library", + ], +) + +go_test( + name = "go_default_test", + srcs = ["publish_test.go"], + embed = [":go_default_library"], + deps = [ + "//platform/base/messagequeue:go_default_library", + "//platform/consumer:go_default_library", + "//platform/extension/messagequeue/mock:go_default_library", + "@com_github_stretchr_testify//assert:go_default_library", + "@com_github_stretchr_testify//require:go_default_library", + "@org_uber_go_mock//gomock:go_default_library", + ], +) diff --git a/submitqueue/core/publish/publish.go b/submitqueue/core/publish/publish.go new file mode 100644 index 00000000..dd942a2a --- /dev/null +++ b/submitqueue/core/publish/publish.go @@ -0,0 +1,67 @@ +// 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 publish sends a message to the queue behind a topic key. It owns the +// lookup-and-send plumbing every orchestrator stage otherwise repeats — resolve +// the key to a queue and a topic name, wrap the payload in a message, publish — +// and the message-ID convention that controls deduplication (see UniqueID). +package publish + +import ( + "context" + "fmt" + "sync/atomic" + "time" + + entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" + "github.com/uber/submitqueue/platform/consumer" +) + +// Message publishes payload to the topic registered for key. +// +// msgID selects the dedup behavior, so the caller must choose it deliberately. +// The queue deduplicates on (topic, partition key, message ID) against every +// row it has not garbage-collected yet, consumed ones included: +// +// - A stable msgID (an entity's own ID) makes a repeat publish a silent +// no-op. Right for a hand-off that must happen at most once per entity. +// - UniqueID(id) makes every publish distinct. Right for signals that are +// re-sent by design — wake-ups, polls, re-dispatches — where a swallowed +// repeat would stall the pipeline. +func Message(ctx context.Context, registry consumer.TopicRegistry, key consumer.TopicKey, msgID string, payload []byte, partitionKey string) error { + q, ok := registry.Queue(key) + if !ok { + return fmt.Errorf("no queue registered for topic key %s", key) + } + topicName, ok := registry.TopicName(key) + if !ok { + return fmt.Errorf("no topic name registered for topic key %s", key) + } + + msg := entityqueue.NewMessage(msgID, payload, partitionKey, nil) + return q.Publisher().Publish(ctx, topicName, msg) +} + +// sequence breaks ties between UniqueID calls that land on the same clock +// tick: some platforms quantize time.Now coarsely enough for consecutive calls +// to read the same nanosecond. +var sequence atomic.Uint64 + +// UniqueID returns a message ID no earlier publish for the same entity has +// used, so the queue's (topic, partition key, message ID) dedup never swallows +// the repeat. Use it for every publish that is re-sent by design; reusing the +// bare entity ID instead would make the second publish a silent no-op. +func UniqueID(id string) string { + return fmt.Sprintf("%s@%d-%d", id, time.Now().UnixNano(), sequence.Add(1)) +} diff --git a/submitqueue/core/publish/publish_test.go b/submitqueue/core/publish/publish_test.go new file mode 100644 index 00000000..78bf5cbc --- /dev/null +++ b/submitqueue/core/publish/publish_test.go @@ -0,0 +1,80 @@ +// 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 publish + +import ( + "context" + "strings" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" + "github.com/uber/submitqueue/platform/consumer" + queuemock "github.com/uber/submitqueue/platform/extension/messagequeue/mock" + "go.uber.org/mock/gomock" +) + +const testKey consumer.TopicKey = "test-topic-key" + +func newTestRegistry(t *testing.T, ctrl *gomock.Controller) (consumer.TopicRegistry, *queuemock.MockPublisher) { + t.Helper() + + publisher := queuemock.NewMockPublisher(ctrl) + q := queuemock.NewMockQueue(ctrl) + q.EXPECT().Publisher().Return(publisher).AnyTimes() + + registry, err := consumer.NewTopicRegistry( + []consumer.TopicConfig{{Key: testKey, Name: "test-topic", Queue: q}}, + ) + require.NoError(t, err) + return registry, publisher +} + +func TestMessage(t *testing.T) { + ctrl := gomock.NewController(t) + registry, publisher := newTestRegistry(t, ctrl) + + var published entityqueue.Message + publisher.EXPECT(). + Publish(gomock.Any(), "test-topic", gomock.Any()). + DoAndReturn(func(_ context.Context, _ string, msg entityqueue.Message) error { + published = msg + return nil + }) + + err := Message(context.Background(), registry, testKey, "msg-1", []byte("payload"), "partition-1") + require.NoError(t, err) + assert.Equal(t, "msg-1", published.ID) + assert.Equal(t, []byte("payload"), published.Payload) + assert.Equal(t, "partition-1", published.PartitionKey) +} + +func TestMessage_UnregisteredKey(t *testing.T) { + ctrl := gomock.NewController(t) + registry, _ := newTestRegistry(t, ctrl) + + err := Message(context.Background(), registry, "unregistered-key", "msg-1", []byte("payload"), "partition-1") + require.Error(t, err) +} + +func TestUniqueID(t *testing.T) { + a := UniqueID("batch-1") + b := UniqueID("batch-1") + + assert.True(t, strings.HasPrefix(a, "batch-1@")) + assert.True(t, strings.HasPrefix(b, "batch-1@")) + assert.NotEqual(t, a, b) +} diff --git a/submitqueue/orchestrator/controller/build/BUILD.bazel b/submitqueue/orchestrator/controller/build/BUILD.bazel index 9301f3ce..c9d39f3e 100644 --- a/submitqueue/orchestrator/controller/build/BUILD.bazel +++ b/submitqueue/orchestrator/controller/build/BUILD.bazel @@ -6,9 +6,9 @@ go_library( importpath = "github.com/uber/submitqueue/submitqueue/orchestrator/controller/build", visibility = ["//visibility:public"], deps = [ - "//platform/base/messagequeue:go_default_library", "//platform/consumer:go_default_library", "//platform/metrics:go_default_library", + "//submitqueue/core/publish:go_default_library", "//submitqueue/core/topickey:go_default_library", "//submitqueue/entity:go_default_library", "//submitqueue/extension/buildrunner:go_default_library", @@ -26,13 +26,10 @@ go_test( "//platform/base/messagequeue:go_default_library", "//platform/consumer:go_default_library", "//platform/consumer/mock:go_default_library", - "//platform/errs:go_default_library", "//platform/extension/messagequeue/mock:go_default_library", - "//submitqueue/core/changeset/fake:go_default_library", "//submitqueue/core/topickey:go_default_library", "//submitqueue/entity:go_default_library", "//submitqueue/extension/buildrunner:go_default_library", - "//submitqueue/extension/buildrunner/fake:go_default_library", "//submitqueue/extension/buildrunner/mock:go_default_library", "//submitqueue/extension/storage:go_default_library", "//submitqueue/extension/storage/mock:go_default_library", diff --git a/submitqueue/orchestrator/controller/build/build.go b/submitqueue/orchestrator/controller/build/build.go index 0c4c0d28..fbd1083a 100644 --- a/submitqueue/orchestrator/controller/build/build.go +++ b/submitqueue/orchestrator/controller/build/build.go @@ -12,6 +12,20 @@ // See the License for the specific language governing permissions and // limitations under the License. +// Package build starts the builds a head batch's speculation paths have been +// funded for. A message names only the head; the path set is the instruction, +// and each pending entry in it is a build to start. Nothing about the action +// travels on the wire, so a dispatch cannot go stale between publish and +// delivery: a path refuted after its message was sent is simply no longer +// pending by the time the set is read. +// +// This stage only starts builds. Stopping them is the poll loop's job: every +// build started here is handed to the buildsignal stage, which follows it to a +// terminal state and stops it the moment its path no longer wants it. The +// split is what keeps this stage simple — it decides whether a build should +// exist, never whether one should die — and it holds together because of one +// invariant this stage maintains: every build that gets a link gets a signal +// (see ensureSignal for the crash case). package build import ( @@ -20,9 +34,9 @@ import ( "fmt" "github.com/uber-go/tally" - entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/submitqueue/core/publish" "github.com/uber/submitqueue/submitqueue/core/topickey" "github.com/uber/submitqueue/submitqueue/entity" "github.com/uber/submitqueue/submitqueue/extension/buildrunner" @@ -31,7 +45,6 @@ import ( ) // Controller handles build queue messages. -// It consumes batches, triggers builds, and publishes scheduled builds to the build signal stage (which processes build results). // Implements consumer.Controller interface for integration with the consumer. type Controller struct { logger *zap.SugaredLogger @@ -46,6 +59,9 @@ type Controller struct { // Verify Controller implements consumer.Controller interface at compile time. var _ consumer.Controller = (*Controller)(nil) +// opName is the metric operation name shared by every emit in this file. +const opName = "process" + // NewController creates a new build controller for the orchestrator. func NewController( logger *zap.SugaredLogger, @@ -67,148 +83,288 @@ func NewController( } } -// Process processes a build delivery from the queue. -// Deserializes the batch, triggers a build, and publishes a build entity to the build signal topic. +// Process starts a build for every pending path in the head's set. // Returns nil to ack (success), or error to nack (retry). +// +// This controller never writes the path set. The set is the speculate run's +// state — what it decided to fund and what it wants stopped — and a second +// writer on it would make speculate lose compare-and-swap races across its own +// far longer read-decide-write window. What this stage knows is the build it +// started, and that goes in records of its own. +// +// A failure partway through nacks the whole message and redelivery re-runs all +// of it: starts are made idempotent by the link record each one writes (see +// startPath), and a redelivery additionally re-publishes the signal for any +// live path already linked to a build, in case the crashed attempt died +// between the link and the signal (see ensureSignal). func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) error { - const opName = "process" - msg := delivery.Message() - // Deserialize batch ID from payload bid, err := entity.BatchIDFromBytes(msg.Payload) if err != nil { metrics.NamedCounter(c.metricsScope, opName, "deserialize_errors", 1) return fmt.Errorf("failed to deserialize batch ID: %w", err) } - // Fetch batch from storage batch, err := c.store.GetBatchStore().Get(ctx, bid.ID) if err != nil { metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) return fmt.Errorf("failed to get batch %s: %w", bid.ID, err) } - c.logger.Infow("received build event", + set, err := c.store.GetSpeculationPathSetStore().Get(ctx, batch.ID) + if err != nil { + if errors.Is(err, storage.ErrNotFound) { + // No speculation run has funded this head yet, so there is nothing + // to dispatch. A later run publishes again once it has. + metrics.NamedCounter(c.metricsScope, opName, "no_path_set", 1) + return nil + } + metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) + return fmt.Errorf("failed to get path set for batch %s: %w", batch.ID, err) + } + + c.logger.Debugw("received build event", "batch_id", batch.ID, "queue", batch.Queue, "state", string(batch.State), - "version", batch.Version, + "paths", len(set.Paths), "attempt", delivery.Attempt(), "partition_key", msg.PartitionKey, ) - // If the batch is halted (terminal OR cancelling), skip triggering CI and - // ack. This is a forward-progress controller: per the cancel design, the - // speculate controller owns cancelling any in-flight Build and driving the - // batch to its terminal state, so the build stage simply short-circuits - // while speculate does the work. No external CI is ever kicked off. - if entity.IsBatchStateHalted(batch.State) { - metrics.NamedCounter(c.metricsScope, opName, "skipped_halted", 1) - c.logger.Infow("skipping build for halted batch", - "batch_id", batch.ID, - "state", string(batch.State), - ) - return nil + // halted covers terminal and cancelling batches, and gates only starts: + // a halted batch gets no new builds, and its running ones are stopped by + // the poll loop, which reads the batch state on every poll. + halted := entity.IsBatchStateHalted(batch.State) + + // Attempt here is the queue's redelivery counter for this message — how + // many times it has been handed out without being acked — and has nothing + // to do with a path's build attempt. A value above 1 means an earlier + // processing of this same message died part-way (nacked or crashed), and + // only then can a linked build be missing its signal — a first delivery + // has not published any — so only then is the repair worth the republishes + // it would otherwise scatter (see ensureSignal). + recovering := delivery.Attempt() > 1 + + for _, entry := range set.Paths { + if entry.Status.IsTerminal() { + continue + } + + if entry.Status == entity.SpeculationPathStatusPending && !halted { + if err := c.startPath(ctx, batch, entry); err != nil { + return err + } + continue + } + + if entry.Status == entity.SpeculationPathStatusPending { + metrics.NamedCounter(c.metricsScope, opName, "skipped_halted", 1) + } + + if recovering { + if err := c.ensureSignal(ctx, batch, entry); err != nil { + return err + } + } } - // Load the dependency batches (base) as identity; the build runner resolves - // each batch's changes itself. head is this batch. - base, err := c.loadBatches(ctx, batch.Dependencies) - if err != nil { + return nil +} + +// startPath triggers the build for a pending path and records it. +// +// The base is the path's own — the dependencies it assumes will succeed, in its +// order. This is the behavioral heart of speculation: dependencies the path +// assumes will fail, and ones it ignores, are absent from the base, which is +// what lets the head be verified before they resolve. +// +// The write order is Trigger, then the Build record, then the link, then the +// signal — each write makes the previous one reachable. The Build record gives +// the runner's ID a home, the link makes the build findable from the path the +// caller holds, and the signal puts the poll loop behind it. The link is the +// idempotency point: a redelivery that finds it re-publishes the signal and +// stops, and a concurrent dispatch that loses the link's first-insert race +// hands both builds to the poll loop, which keeps the one the link names (see +// the lost-race branch below). +// +// A crash between the Trigger and the link orphans the build — redelivery +// cannot find it and triggers again — the accepted cost of not being able to +// name a build before the runner mints its ID. +// +// TODO: pass an idempotency key derived from (path ID, attempt) once +// BuildRunner.Trigger accepts one, so a retry re-attaches to the existing build +// instead of orphaning it. Only the runner can close this window, because the +// build exists before anything here can write it down. +func (c *Controller) startPath(ctx context.Context, batch entity.Batch, entry entity.SpeculationPathEntry) error { + existing, err := c.store.GetPathBuildStore().Get(ctx, entry.ID, entry.Attempt) + switch { + case err == nil: + // Already dispatched and named; all that can be missing is the signal, + // and a re-publish is deduped by the queue when it is not. + metrics.NamedCounter(c.metricsScope, opName, "already_dispatched", 1) + return c.publishBuildSignal(ctx, existing.BuildID) + case !errors.Is(err, storage.ErrNotFound): metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) - return fmt.Errorf("failed to load dependency batches for batch %s: %w", batch.ID, err) + return fmt.Errorf("failed to look up build for path %s attempt %d: %w", entry.ID, entry.Attempt, err) + } + + base, err := c.loadBase(ctx, entry.Path) + if err != nil { + return err } - // Trigger the build with the queue's build runner. metadata is nil - // until a caller-supplied source materializes (e.g. requester / ticket - // pulled off the originating LandRequest). buildRunner, err := c.buildRunners.For(buildrunner.Config{QueueName: batch.Queue}) if err != nil { metrics.NamedCounter(c.metricsScope, opName, "trigger_errors", 1) return fmt.Errorf("failed to build runner for batch %s: %w", batch.ID, err) } + + // metadata is nil until a caller-supplied source materializes (e.g. + // requester / ticket pulled off the originating LandRequest). buildID, err := buildRunner.Trigger(ctx, base, batch, nil) if err != nil { metrics.NamedCounter(c.metricsScope, opName, "trigger_errors", 1) - return fmt.Errorf("failed to trigger build for batch %s: %w", batch.ID, err) + return fmt.Errorf("failed to trigger build for path %s attempt %d: %w", entry.ID, entry.Attempt, err) } build := entity.Build{ ID: buildID.ID, BatchID: batch.ID, + PathID: entry.ID, + Attempt: entry.Attempt, Status: entity.BuildStatusAccepted, } - - // Persist the initial Build snapshot so the buildsignal poll loop has a - // row to Update against. ErrAlreadyExists is benign — a redelivery - // of this message after a previous successful Create. if err := c.store.GetBuildStore().Create(ctx, build); err != nil && !errors.Is(err, storage.ErrAlreadyExists) { metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) - return fmt.Errorf("failed to persist build %s: %w", build.ID, err) + return fmt.Errorf("failed to record build %s: %w", buildID.ID, err) } - // Hand off to the buildsignal poll loop; it calls Status, updates the - // persisted Build, publishes to speculate, and holds its delivery - // between polls until terminal. - if err := c.publish(ctx, topickey.TopicKeyBuildSignal, build); err != nil { - metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1) - return fmt.Errorf("failed to publish to buildsignal: %w", err) + link := entity.PathBuild{PathID: entry.ID, Attempt: entry.Attempt, BuildID: buildID.ID} + if err := c.store.GetPathBuildStore().Create(ctx, link); err != nil { + if errors.Is(err, storage.ErrAlreadyExists) { + // Another dispatch named this attempt first. Its build is the one + // the rest of the system will watch; ours is surplus. Hand both to + // the poll loop: it keeps the build the link names and stops the + // other one — and the winner's signal must be sent from here too, + // because the dispatch that won may have died before sending it, + // and acking this message would retire the redelivery that would + // otherwise repair that. + metrics.NamedCounter(c.metricsScope, opName, "lost_dispatch_race", 1) + c.logger.Infow("lost the dispatch race; handing both builds to the poll loop", + "batch_id", batch.ID, + "path_id", entry.ID, + "attempt", entry.Attempt, + "surplus_build_id", buildID.ID, + ) + if err := c.publishBuildSignal(ctx, buildID.ID); err != nil { + return err + } + winner, err := c.store.GetPathBuildStore().Get(ctx, entry.ID, entry.Attempt) + if err != nil { + metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) + return fmt.Errorf("failed to look up the winning build for path %s attempt %d: %w", entry.ID, entry.Attempt, err) + } + return c.publishBuildSignal(ctx, winner.BuildID) + } + metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) + return fmt.Errorf("failed to link path %s attempt %d to build %s: %w", entry.ID, entry.Attempt, buildID.ID, err) } - c.logger.Infow("published build to buildsignal", + metrics.NamedCounter(c.metricsScope, opName, "build_triggered", 1) + c.logger.Infow("triggered build for speculation path", "batch_id", batch.ID, - "build_id", build.ID, - "status", string(build.Status), - "topic_key", topickey.TopicKeyBuildSignal, + "path_id", entry.ID, + "attempt", entry.Attempt, + "build_id", buildID.ID, + "base_size", len(base), ) - return nil // Success - message will be acked + return c.publishBuildSignal(ctx, buildID.ID) +} + +// ensureSignal re-publishes the build signal for a live path whose build is +// already linked. Called only on redeliveries. +// +// It closes the one crack in the "every linked build gets a signal" invariant: +// a dispatch that dies between writing the link and publishing the signal +// leaves a build no poll chain will ever watch. The message it was processing +// was never acked, so it comes back — but by then the entry may have moved on +// to building (an observation found the link) or cancelling (the run called it +// off), or its batch may have halted: states no start would touch. Without +// this, such a build runs unobserved forever and its path never settles. +// +// The republish uses the build ID as the message ID, so it dedups against the +// original signal whenever that signal was actually sent. The dedup horizon is +// the queue's GC of consumed rows, which is why this runs only on redelivery: +// scattered over every ordinary dispatch, late republishes would now and then +// slip past dedup and fork a second, redundant poll chain for a healthy build. +func (c *Controller) ensureSignal(ctx context.Context, batch entity.Batch, entry entity.SpeculationPathEntry) error { + link, err := c.store.GetPathBuildStore().Get(ctx, entry.ID, entry.Attempt) + if errors.Is(err, storage.ErrNotFound) { + // Nothing was dispatched for this attempt; there is no build to watch. + return nil + } + if err != nil { + metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) + return fmt.Errorf("failed to look up build for path %s attempt %d: %w", entry.ID, entry.Attempt, err) + } + + metrics.NamedCounter(c.metricsScope, opName, "signal_ensured", 1) + c.logger.Debugw("re-published build signal on redelivery", + "batch_id", batch.ID, + "path_id", entry.ID, + "attempt", entry.Attempt, + "build_id", link.BuildID, + ) + return c.publishBuildSignal(ctx, link.BuildID) } -// loadBatches loads each batch by ID, preserving order. Used to load the base -// (dependency batches) identity handed to BuildRunner.Trigger; the build runner -// resolves each batch's changes itself. -func (c *Controller) loadBatches(ctx context.Context, batchIDs []string) ([]entity.Batch, error) { - if len(batchIDs) == 0 { +// loadBase loads the batches the path is stacked on top of. +// +// Which dependencies those are is the path's own to say — see +// SpeculationPath.Base — so this only resolves the IDs it is +// given. Nothing about assumptions is interpreted here. +func (c *Controller) loadBase(ctx context.Context, path entity.SpeculationPath) ([]entity.Batch, error) { + deps := path.Base() + if len(deps) == 0 { return nil, nil } - batches := make([]entity.Batch, 0, len(batchIDs)) - for _, bID := range batchIDs { - b, err := c.store.GetBatchStore().Get(ctx, bID) + + base := make([]entity.Batch, 0, len(deps)) + for _, depID := range deps { + b, err := c.store.GetBatchStore().Get(ctx, depID) if err != nil { - return nil, fmt.Errorf("failed to get batch %s: %w", bID, err) + metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) + return nil, fmt.Errorf("failed to get dependency batch %s of path %s: %w", depID, path.ID(), err) } - batches = append(batches, b) + base = append(base, b) } - return batches, nil + return base, nil } -// publish publishes a build's ID to the specified topic key. Only the -// identifier travels on the queue; the consumer loads the full Build from -// storage, keeping the message small and the store the single source of truth. -func (c *Controller) publish(ctx context.Context, key consumer.TopicKey, build entity.Build) error { - payload, err := entity.BuildID{ID: build.ID}.ToBytes() +// publishBuildSignal hands a build off to the poll loop. +// +// Only the runner's build ID travels, and it is also the partition key: the +// poll loop writes nothing but that build's own record, so there is nothing to +// serialize across builds, and successive polls of one build stay ordered +// because they share the key. Partitioning by batch instead would put every +// path of a head behind whichever of its builds polls slowest. +// +// The build ID is the message ID too — a stable ID on purpose, so a repeat +// hand-off for the same build dedups away while the original signal is still +// in the queue's un-GC'd window (see publish.Message). +func (c *Controller) publishBuildSignal(ctx context.Context, buildID string) error { + payload, err := entity.BuildID{ID: buildID}.ToBytes() if err != nil { return fmt.Errorf("failed to serialize build ID: %w", err) } - msg := entityqueue.NewMessage(build.ID, payload, build.BatchID, nil) - - q, ok := c.registry.Queue(key) - if !ok { - return fmt.Errorf("no queue registered for topic key %s", key) - } - - topicName, ok := c.registry.TopicName(key) - if !ok { - return fmt.Errorf("no topic name registered for topic key %s", key) - } - - if err := q.Publisher().Publish(ctx, topicName, msg); err != nil { - return fmt.Errorf("failed to publish message: %w", err) + if err := publish.Message(ctx, c.registry, topickey.TopicKeyBuildSignal, buildID, payload, buildID); err != nil { + metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1) + return fmt.Errorf("failed to publish to buildsignal: %w", err) } - return nil } diff --git a/submitqueue/orchestrator/controller/build/build_test.go b/submitqueue/orchestrator/controller/build/build_test.go index 3790b17a..c78aa5f8 100644 --- a/submitqueue/orchestrator/controller/build/build_test.go +++ b/submitqueue/orchestrator/controller/build/build_test.go @@ -25,13 +25,10 @@ import ( entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" consumermock "github.com/uber/submitqueue/platform/consumer/mock" - "github.com/uber/submitqueue/platform/errs" queuemock "github.com/uber/submitqueue/platform/extension/messagequeue/mock" - changesetfake "github.com/uber/submitqueue/submitqueue/core/changeset/fake" "github.com/uber/submitqueue/submitqueue/core/topickey" "github.com/uber/submitqueue/submitqueue/entity" "github.com/uber/submitqueue/submitqueue/extension/buildrunner" - buildfake "github.com/uber/submitqueue/submitqueue/extension/buildrunner/fake" buildrunnermock "github.com/uber/submitqueue/submitqueue/extension/buildrunner/mock" "github.com/uber/submitqueue/submitqueue/extension/storage" storagemock "github.com/uber/submitqueue/submitqueue/extension/storage/mock" @@ -39,347 +36,468 @@ import ( "go.uber.org/zap/zaptest" ) -// batchIDPayload serializes a BatchID to JSON bytes for test message payloads. +const ( + headID = "test-queue/batch/head" + depA = "test-queue/batch/depA" + depB = "test-queue/batch/depB" + depC = "test-queue/batch/depC" +) + +// staticBuildRunnerFactory is a test factory that returns a fixed BuildRunner. +type staticBuildRunnerFactory struct{ r buildrunner.BuildRunner } + +func (f staticBuildRunnerFactory) For(buildrunner.Config) (buildrunner.BuildRunner, error) { + return f.r, nil +} + func batchIDPayload(t *testing.T, id string) []byte { + t.Helper() payload, err := entity.BatchID{ID: id}.ToBytes() require.NoError(t, err) return payload } -// testBatch returns a standard test batch for build tests. -func testBatch() entity.Batch { +// headBatch returns the head batch in the given state, depending on all three deps. +func headBatch(state entity.BatchState) entity.Batch { return entity.Batch{ - ID: "test-queue/batch/1", - Queue: "test-queue", - State: entity.BatchStateCreated, - Version: 1, + ID: headID, + Queue: "test-queue", + State: state, + Dependencies: []string{depA, depB, depC}, + Version: 1, } } -// newMockStorage creates a MockStorage with a MockBatchStore that returns the -// given batch on Get, a no-op MockRequestStore, and a MockBuildStore that -// accepts any Create call. Tests that care about Create arguments build their -// own MockBuildStore. -func newMockStorage(ctrl *gomock.Controller, batch entity.Batch) *storagemock.MockStorage { - mockBatchStore := storagemock.NewMockBatchStore(ctrl) - mockBatchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil).AnyTimes() - - mockRequestStore := storagemock.NewMockRequestStore(ctrl) - - mockBuildStore := storagemock.NewMockBuildStore(ctrl) - mockBuildStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil).AnyTimes() +// pathEntry builds one path-set entry for the head, assuming depA succeeds, +// depB fails, and depC is ignored — so only depA belongs in the build base. +func pathEntry(status entity.SpeculationPathStatus, attempt int) entity.SpeculationPathEntry { + path := entity.SpeculationPath{ + Head: headID, + Dependencies: []entity.PathDependency{ + {Batch: depA, Assumption: entity.DependencyAssumptionSucceeds}, + {Batch: depB, Assumption: entity.DependencyAssumptionFails}, + {Batch: depC, Assumption: entity.DependencyAssumptionIgnored}, + }, + } + return entity.SpeculationPathEntry{ + ID: path.ID(), + Path: path, + Status: status, + Attempt: attempt, + Version: 1, + } +} - store := storagemock.NewMockStorage(ctrl) - store.EXPECT().GetBatchStore().Return(mockBatchStore).AnyTimes() - store.EXPECT().GetRequestStore().Return(mockRequestStore).AnyTimes() - store.EXPECT().GetBuildStore().Return(mockBuildStore).AnyTimes() - return store +// testDeps holds the mocks a test may want to set expectations on. +type testDeps struct { + store *storagemock.MockStorage + batches *storagemock.MockBatchStore + pathSets *storagemock.MockSpeculationPathSetStore + builds *storagemock.MockBuildStore + pathBuilds *storagemock.MockPathBuildStore + runner *buildrunnermock.MockBuildRunner + publisher *queuemock.MockPublisher } -// newTestController creates a controller with test dependencies. br is the -// build runner to inject; pass buildfake.New(changesetfake.New()) for the pass-through default. -// staticBuildRunnerFactory is a test factory that returns a fixed BuildRunner -// for any entityqueue. -type staticBuildRunnerFactory struct{ r buildrunner.BuildRunner } +// newTestController wires a controller over fresh mocks. The batch store answers +// Get for the head and every dependency; everything else is left to the test. +func newTestController(t *testing.T, ctrl *gomock.Controller, batch entity.Batch) (*Controller, *testDeps) { + t.Helper() -func (f staticBuildRunnerFactory) For(buildrunner.Config) (buildrunner.BuildRunner, error) { - return f.r, nil -} + batches := storagemock.NewMockBatchStore(ctrl) + batches.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil).AnyTimes() + for _, dep := range []string{depA, depB, depC} { + batches.EXPECT().Get(gomock.Any(), dep). + Return(entity.Batch{ID: dep, Queue: batch.Queue, State: entity.BatchStateSucceeded}, nil).AnyTimes() + } -// The wired registry exposes only the buildsignal topic — that is what the -// controller publishes to after the RFC refactor. -func newTestController(t *testing.T, ctrl *gomock.Controller, store *storagemock.MockStorage, br buildrunner.BuildRunner, publishErr error) *Controller { - logger := zaptest.NewLogger(t).Sugar() - scope := tally.NoopScope + pathSets := storagemock.NewMockSpeculationPathSetStore(ctrl) + builds := storagemock.NewMockBuildStore(ctrl) + pathBuilds := storagemock.NewMockPathBuildStore(ctrl) - mockPub := queuemock.NewMockPublisher(ctrl) - mockPub.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn( - func(ctx context.Context, topic string, msg entityqueue.Message) error { - return publishErr - }, - ).AnyTimes() + store := storagemock.NewMockStorage(ctrl) + store.EXPECT().GetBatchStore().Return(batches).AnyTimes() + store.EXPECT().GetSpeculationPathSetStore().Return(pathSets).AnyTimes() + store.EXPECT().GetBuildStore().Return(builds).AnyTimes() + store.EXPECT().GetPathBuildStore().Return(pathBuilds).AnyTimes() - mockQ := queuemock.NewMockQueue(ctrl) - mockQ.EXPECT().Publisher().Return(mockPub).AnyTimes() + publisher := queuemock.NewMockPublisher(ctrl) + q := queuemock.NewMockQueue(ctrl) + q.EXPECT().Publisher().Return(publisher).AnyTimes() registry, err := consumer.NewTopicRegistry( - []consumer.TopicConfig{{Key: topickey.TopicKeyBuildSignal, Name: "buildsignal", Queue: mockQ}}, + []consumer.TopicConfig{{Key: topickey.TopicKeyBuildSignal, Name: "buildsignal", Queue: q}}, ) require.NoError(t, err) - return NewController(logger, scope, store, staticBuildRunnerFactory{r: br}, registry, topickey.TopicKeyBuild, "orchestrator-build") -} - -func TestNewController(t *testing.T) { - ctrl := gomock.NewController(t) - batch := testBatch() - store := newMockStorage(ctrl, batch) - controller := newTestController(t, ctrl, store, buildfake.New(changesetfake.New()), nil) - - require.NotNil(t, controller) - assert.Equal(t, topickey.TopicKeyBuild, controller.TopicKey()) - assert.Equal(t, "orchestrator-build", controller.ConsumerGroup()) - assert.Equal(t, "build", controller.Name()) -} - -func TestController_Process_Success(t *testing.T) { - ctrl := gomock.NewController(t) - - batch := testBatch() - store := newMockStorage(ctrl, batch) - controller := newTestController(t, ctrl, store, buildfake.New(changesetfake.New()), nil) - - msg := entityqueue.NewMessage(batch.ID, batchIDPayload(t, batch.ID), batch.Queue, nil) - delivery := consumermock.NewMockDelivery(ctrl) - delivery.EXPECT().Message().Return(msg).AnyTimes() - delivery.EXPECT().Attempt().Return(1).AnyTimes() - - err := controller.Process(context.Background(), delivery) - require.NoError(t, err) -} + runner := buildrunnermock.NewMockBuildRunner(ctrl) -// TestController_Process_TriggersWithBaseAndHead verifies the controller hands -// BuildRunner.Trigger the base (dependency batches in order) and head (this -// batch) as identity, persists the initial Accepted Build, and publishes it to -// the buildsignal topic. The runner resolves each batch's changes itself. -func TestController_Process_TriggersWithBaseAndHead(t *testing.T) { - ctrl := gomock.NewController(t) + controller := NewController( + zaptest.NewLogger(t).Sugar(), tally.NoopScope, store, + staticBuildRunnerFactory{r: runner}, registry, topickey.TopicKeyBuild, "orchestrator-build", + ) - depBatch := entity.Batch{ - ID: "test-queue/batch/dep", - Queue: "test-queue", - Contains: []string{"test-queue/dep-1"}, - } - headBatch := entity.Batch{ - ID: "test-queue/batch/head", - Queue: "test-queue", - State: entity.BatchStateSpeculating, - Version: 1, - Dependencies: []string{depBatch.ID}, - Contains: []string{"test-queue/head-1", "test-queue/head-2"}, + return controller, &testDeps{ + store: store, batches: batches, pathSets: pathSets, + builds: builds, pathBuilds: pathBuilds, + runner: runner, publisher: publisher, } +} - mockBatchStore := storagemock.NewMockBatchStore(ctrl) - mockBatchStore.EXPECT().Get(gomock.Any(), headBatch.ID).Return(headBatch, nil).AnyTimes() - mockBatchStore.EXPECT().Get(gomock.Any(), depBatch.ID).Return(depBatch, nil).AnyTimes() +// processAttempt delivers the head's batch ID with the given delivery attempt. +func processAttempt(t *testing.T, ctrl *gomock.Controller, c *Controller, attempt int) error { + t.Helper() + msg := entityqueue.NewMessage("msg-1", batchIDPayload(t, headID), "test-queue", nil) + d := consumermock.NewMockDelivery(ctrl) + d.EXPECT().Message().Return(msg).AnyTimes() + d.EXPECT().Attempt().Return(attempt).AnyTimes() + return c.Process(context.Background(), d) +} - var created entity.Build - mockBuildStore := storagemock.NewMockBuildStore(ctrl) - mockBuildStore.EXPECT().Create(gomock.Any(), gomock.Any()).DoAndReturn( - func(_ context.Context, b entity.Build) error { - created = b - return nil - }, - ).Times(1) +// process delivers the head's batch ID as a first delivery. +func process(t *testing.T, ctrl *gomock.Controller, c *Controller) error { + t.Helper() + return processAttempt(t, ctrl, c, 1) +} - store := storagemock.NewMockStorage(ctrl) - store.EXPECT().GetBatchStore().Return(mockBatchStore).AnyTimes() - store.EXPECT().GetBuildStore().Return(mockBuildStore).AnyTimes() - - br := buildrunnermock.NewMockBuildRunner(ctrl) - // base is the dependency batches (identity); head is this batch. - br.EXPECT().Trigger(gomock.Any(), []entity.Batch{depBatch}, headBatch, gomock.Nil()).Return(entity.BuildID{ID: "build-xyz"}, nil) - - var publishedTopic string - var published entity.BuildID - mockPub := queuemock.NewMockPublisher(ctrl) - mockPub.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn( - func(_ context.Context, topic string, msg entityqueue.Message) error { - publishedTopic = topic - bid, err := entity.BuildIDFromBytes(msg.Payload) +// expectSignal expects one buildsignal publish for the given build, asserting +// the payload carries the build ID and the message partitions on it. +func expectSignal(t *testing.T, deps *testDeps, buildID string) { + t.Helper() + deps.publisher.EXPECT().Publish(gomock.Any(), "buildsignal", gomock.Any()). + DoAndReturn(func(_ context.Context, _ string, msg entityqueue.Message) error { + got, err := entity.BuildIDFromBytes(msg.Payload) require.NoError(t, err) - published = bid + assert.Equal(t, buildID, got.ID) + assert.Equal(t, buildID, msg.PartitionKey, + "polls partition per build so one slow build cannot block a head's others") return nil - }, - ) - mockQ := queuemock.NewMockQueue(ctrl) - mockQ.EXPECT().Publisher().Return(mockPub).AnyTimes() - registry, err := consumer.NewTopicRegistry( - []consumer.TopicConfig{{Key: topickey.TopicKeyBuildSignal, Name: "buildsignal", Queue: mockQ}}, - ) - require.NoError(t, err) + }) +} - controller := NewController(zaptest.NewLogger(t).Sugar(), tally.NoopScope, store, staticBuildRunnerFactory{r: br}, registry, topickey.TopicKeyBuild, "orchestrator-build") +// notDispatched makes the reverse lookup miss, i.e. this attempt has no build yet. +func notDispatched(deps *testDeps) { + deps.pathBuilds.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any()). + Return(entity.PathBuild{}, storage.ErrNotFound).AnyTimes() +} - msg := entityqueue.NewMessage(headBatch.ID, batchIDPayload(t, headBatch.ID), headBatch.Queue, nil) - delivery := consumermock.NewMockDelivery(ctrl) - delivery.EXPECT().Message().Return(msg).AnyTimes() - delivery.EXPECT().Attempt().Return(1).AnyTimes() +func TestNewController(t *testing.T) { + ctrl := gomock.NewController(t) + c, _ := newTestController(t, ctrl, headBatch(entity.BatchStateSpeculating)) - require.NoError(t, controller.Process(context.Background(), delivery)) + assert.Equal(t, topickey.TopicKeyBuild, c.TopicKey()) + assert.Equal(t, "orchestrator-build", c.ConsumerGroup()) + assert.Equal(t, "build", c.Name()) +} - // Only the build ID is published to buildsignal. - assert.Equal(t, "buildsignal", publishedTopic) - assert.Equal(t, "build-xyz", published.ID) +// TestProcess_TriggersWithThePathsBase is the behavioral heart of +// speculation: the build base is the dependencies the path assumes succeed, not +// the head's full dependency list. depB is assumed to fail and depC is ignored, +// so neither may appear. +// +// It also pins the write order — Trigger, Build record, link, signal — because +// each write is what makes the previous one reachable: the link must never name +// a build that has no record, and a signal must never be sent for a build the +// link does not yet make findable. +func TestProcess_TriggersWithThePathsBase(t *testing.T) { + ctrl := gomock.NewController(t) + batch := headBatch(entity.BatchStateSpeculating) + c, deps := newTestController(t, ctrl, batch) + + entry := pathEntry(entity.SpeculationPathStatusPending, 1) + deps.pathSets.EXPECT().Get(gomock.Any(), headID).Return(entity.SpeculationPathSet{ + Head: headID, Paths: []entity.SpeculationPathEntry{entry}, Version: 4, + }, nil) + notDispatched(deps) + + wantBase := []entity.Batch{{ID: depA, Queue: "test-queue", State: entity.BatchStateSucceeded}} + + gomock.InOrder( + deps.runner.EXPECT(). + Trigger(gomock.Any(), wantBase, batch, nil). + Return(entity.BuildID{ID: "build-1"}, nil), + deps.builds.EXPECT().Create(gomock.Any(), gomock.Any()). + DoAndReturn(func(_ context.Context, b entity.Build) error { + assert.Equal(t, "build-1", b.ID) + assert.Equal(t, headID, b.BatchID) + assert.Equal(t, entry.ID, b.PathID) + assert.Equal(t, 1, b.Attempt) + return nil + }), + deps.pathBuilds.EXPECT().Create(gomock.Any(), entity.PathBuild{ + PathID: entry.ID, Attempt: 1, BuildID: "build-1", + }).Return(nil), + deps.publisher.EXPECT().Publish(gomock.Any(), "buildsignal", gomock.Any()). + DoAndReturn(func(_ context.Context, _ string, msg entityqueue.Message) error { + got, err := entity.BuildIDFromBytes(msg.Payload) + require.NoError(t, err) + assert.Equal(t, "build-1", got.ID) + assert.Equal(t, "build-1", msg.PartitionKey, + "polls partition per build so one slow build cannot block a head's others") + return nil + }), + ) - // The full Build is persisted to storage (the source of truth the poll - // loop reloads), and its ID matches what was published. - assert.Equal(t, "build-xyz", created.ID) - assert.Equal(t, headBatch.ID, created.BatchID) - assert.Equal(t, entity.BuildStatusAccepted, created.Status) - assert.Equal(t, published.ID, created.ID) + require.NoError(t, process(t, ctrl, c)) } -// TestController_Process_BuildStoreAlreadyExistsIsSwallowed covers the -// redelivery case: Create returns ErrAlreadyExists, the controller proceeds -// to publish to buildsignal anyway. The polling loop will pick up the -// existing row via Update. -func TestController_Process_BuildStoreAlreadyExistsIsSwallowed(t *testing.T) { +// The path set belongs to the speculate run. This stage must never write it, or +// speculate loses compare-and-swap races across its far longer window. And a +// cancelling entry is not this stage's work at all on an ordinary delivery — +// the poll loop stops unwanted builds — so it must not even be looked up. +func TestProcess_NeverWritesThePathSetAndIgnoresCancelling(t *testing.T) { ctrl := gomock.NewController(t) + c, deps := newTestController(t, ctrl, headBatch(entity.BatchStateSpeculating)) + + pending := pathEntry(entity.SpeculationPathStatusPending, 1) + cancelling := pathEntry(entity.SpeculationPathStatusCancelling, 1) + cancelling.ID = "cancelling-path" + + deps.pathSets.EXPECT().Get(gomock.Any(), headID).Return(entity.SpeculationPathSet{ + Head: headID, Paths: []entity.SpeculationPathEntry{pending, cancelling}, Version: 2, + }, nil) + + deps.pathBuilds.EXPECT().Get(gomock.Any(), pending.ID, 1). + Return(entity.PathBuild{}, storage.ErrNotFound) + + deps.runner.EXPECT().Trigger(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()). + Return(entity.BuildID{ID: "build-1"}, nil) + deps.builds.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil) + deps.pathBuilds.EXPECT().Create(gomock.Any(), entity.PathBuild{ + PathID: pending.ID, Attempt: 1, BuildID: "build-1", + }).Return(nil) + expectSignal(t, deps, "build-1") + + // No pathSets.Update, no pathSets.Create, no runner.Cancel, and no lookup of + // the cancelling path: gomock fails the test on any of them. + require.NoError(t, process(t, ctrl, c)) +} - batch := testBatch() - - mockBatchStore := storagemock.NewMockBatchStore(ctrl) - mockBatchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil).AnyTimes() - mockBuildStore := storagemock.NewMockBuildStore(ctrl) - mockBuildStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(storage.ErrAlreadyExists) +// A redelivered dispatch must re-publish the existing build's signal rather +// than start a second build for the same attempt. +func TestProcess_RedeliveryDoesNotRebuild(t *testing.T) { + ctrl := gomock.NewController(t) + c, deps := newTestController(t, ctrl, headBatch(entity.BatchStateSpeculating)) - store := storagemock.NewMockStorage(ctrl) - store.EXPECT().GetBatchStore().Return(mockBatchStore).AnyTimes() - store.EXPECT().GetRequestStore().Return(storagemock.NewMockRequestStore(ctrl)).AnyTimes() - store.EXPECT().GetBuildStore().Return(mockBuildStore).AnyTimes() - - br := buildrunnermock.NewMockBuildRunner(ctrl) - br.EXPECT().Trigger(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(entity.BuildID{ID: "build-dup"}, nil) - - publishCalled := false - mockPub := queuemock.NewMockPublisher(ctrl) - mockPub.EXPECT().Publish(gomock.Any(), "buildsignal", gomock.Any()).DoAndReturn( - func(_ context.Context, _ string, _ entityqueue.Message) error { - publishCalled = true - return nil - }, - ).Times(1) - mockQ := queuemock.NewMockQueue(ctrl) - mockQ.EXPECT().Publisher().Return(mockPub).AnyTimes() - registry, err := consumer.NewTopicRegistry( - []consumer.TopicConfig{{Key: topickey.TopicKeyBuildSignal, Name: "buildsignal", Queue: mockQ}}, - ) - require.NoError(t, err) - controller := NewController(zaptest.NewLogger(t).Sugar(), tally.NoopScope, store, staticBuildRunnerFactory{r: br}, registry, topickey.TopicKeyBuild, "orchestrator-build") + entry := pathEntry(entity.SpeculationPathStatusPending, 1) + deps.pathSets.EXPECT().Get(gomock.Any(), headID).Return(entity.SpeculationPathSet{ + Head: headID, Paths: []entity.SpeculationPathEntry{entry}, Version: 1, + }, nil) + deps.pathBuilds.EXPECT().Get(gomock.Any(), entry.ID, 1). + Return(entity.PathBuild{PathID: entry.ID, Attempt: 1, BuildID: "build-existing"}, nil) - msg := entityqueue.NewMessage(batch.ID, batchIDPayload(t, batch.ID), batch.Queue, nil) - delivery := consumermock.NewMockDelivery(ctrl) - delivery.EXPECT().Message().Return(msg).AnyTimes() - delivery.EXPECT().Attempt().Return(1).AnyTimes() + expectSignal(t, deps, "build-existing") - require.NoError(t, controller.Process(context.Background(), delivery)) - assert.True(t, publishCalled, "publish to buildsignal must run even when Create reports ErrAlreadyExists") + // No Trigger and no Creates. + require.NoError(t, process(t, ctrl, c)) } -// TestController_Process_TriggerFailure verifies a build-runner failure is -// surfaced as an error (nack) and nothing is persisted or published. -func TestController_Process_TriggerFailure(t *testing.T) { +// A crash between the link and the signal is the one window the short-circuit +// exists for: both records are present, nothing was ever published, so the +// re-publish is not suppressed by the queue's publish idempotency. +func TestProcess_RepublishesWhenOnlyTheSignalIsMissing(t *testing.T) { ctrl := gomock.NewController(t) + c, deps := newTestController(t, ctrl, headBatch(entity.BatchStateSpeculating)) - batch := testBatch() - mockBatchStore := storagemock.NewMockBatchStore(ctrl) - mockBatchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil).AnyTimes() - store := storagemock.NewMockStorage(ctrl) - store.EXPECT().GetBatchStore().Return(mockBatchStore).AnyTimes() - store.EXPECT().GetRequestStore().Return(storagemock.NewMockRequestStore(ctrl)).AnyTimes() - // No build store expectation: Trigger failure must short-circuit before Create. - - br := buildrunnermock.NewMockBuildRunner(ctrl) - br.EXPECT().Trigger(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()). - Return(entity.BuildID{}, fmt.Errorf("provider down")) - - registry, err := consumer.NewTopicRegistry( - []consumer.TopicConfig{{Key: topickey.TopicKeyBuildSignal, Name: "buildsignal", Queue: queuemock.NewMockQueue(ctrl)}}, - ) - require.NoError(t, err) - controller := NewController(zaptest.NewLogger(t).Sugar(), tally.NoopScope, store, staticBuildRunnerFactory{r: br}, registry, topickey.TopicKeyBuild, "orchestrator-build") + entry := pathEntry(entity.SpeculationPathStatusPending, 1) + deps.pathSets.EXPECT().Get(gomock.Any(), headID).Return(entity.SpeculationPathSet{ + Head: headID, Paths: []entity.SpeculationPathEntry{entry}, Version: 1, + }, nil) + deps.pathBuilds.EXPECT().Get(gomock.Any(), entry.ID, 1). + Return(entity.PathBuild{PathID: entry.ID, Attempt: 1, BuildID: "build-1"}, nil) - msg := entityqueue.NewMessage(batch.ID, batchIDPayload(t, batch.ID), batch.Queue, nil) - delivery := consumermock.NewMockDelivery(ctrl) - delivery.EXPECT().Message().Return(msg).AnyTimes() - delivery.EXPECT().Attempt().Return(1).AnyTimes() + expectSignal(t, deps, "build-1") - require.Error(t, controller.Process(context.Background(), delivery)) + // No Trigger and no writes: the records are already consistent. + require.NoError(t, process(t, ctrl, c)) } -func TestController_Process_StorageFailure(t *testing.T) { +// Losing the link's first-insert race means another dispatch named this attempt +// first. Both builds are handed to the poll loop — the surplus so it gets +// stopped, the winner because the dispatch that won may have died before +// signalling, and acking this message retires the redelivery that would have +// repaired that. +func TestProcess_LostDispatchRaceHandsBothBuildsToThePollLoop(t *testing.T) { ctrl := gomock.NewController(t) + c, deps := newTestController(t, ctrl, headBatch(entity.BatchStateSpeculating)) + + entry := pathEntry(entity.SpeculationPathStatusPending, 1) + deps.pathSets.EXPECT().Get(gomock.Any(), headID).Return(entity.SpeculationPathSet{ + Head: headID, Paths: []entity.SpeculationPathEntry{entry}, Version: 1, + }, nil) + + gomock.InOrder( + deps.pathBuilds.EXPECT().Get(gomock.Any(), entry.ID, 1). + Return(entity.PathBuild{}, storage.ErrNotFound), + deps.pathBuilds.EXPECT().Create(gomock.Any(), entity.PathBuild{ + PathID: entry.ID, Attempt: 1, BuildID: "build-surplus", + }).Return(storage.ErrAlreadyExists), + deps.pathBuilds.EXPECT().Get(gomock.Any(), entry.ID, 1). + Return(entity.PathBuild{PathID: entry.ID, Attempt: 1, BuildID: "build-winner"}, nil), + ) - mockBatchStore := storagemock.NewMockBatchStore(ctrl) - mockBatchStore.EXPECT().Get(gomock.Any(), "test-queue/batch/1").Return(entity.Batch{}, fmt.Errorf("db connection lost")) - store := storagemock.NewMockStorage(ctrl) - store.EXPECT().GetBatchStore().Return(mockBatchStore).AnyTimes() - store.EXPECT().GetRequestStore().Return(storagemock.NewMockRequestStore(ctrl)).AnyTimes() - store.EXPECT().GetBuildStore().Return(storagemock.NewMockBuildStore(ctrl)).AnyTimes() - - controller := newTestController(t, ctrl, store, buildfake.New(changesetfake.New()), nil) + deps.runner.EXPECT().Trigger(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()). + Return(entity.BuildID{ID: "build-surplus"}, nil) + deps.builds.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil) - msg := entityqueue.NewMessage("test-queue/batch/1", batchIDPayload(t, "test-queue/batch/1"), "test-queue", nil) - delivery := consumermock.NewMockDelivery(ctrl) - delivery.EXPECT().Message().Return(msg).AnyTimes() - delivery.EXPECT().Attempt().Return(1).AnyTimes() + expectSignal(t, deps, "build-surplus") + expectSignal(t, deps, "build-winner") - err := controller.Process(context.Background(), delivery) - require.Error(t, err) - assert.False(t, errs.IsRetryable(err)) + // The race is resolved, not retried: no runner.Cancel here (the poll loop + // stops the surplus build), and the message acks. + require.NoError(t, process(t, ctrl, c)) } -func TestController_Process_PublishFailure(t *testing.T) { +// A halted batch gets no new builds. Stopping its running ones is the poll +// loop's job, so on a first delivery there is nothing else to do here. +func TestProcess_HaltedBatchStartsNothing(t *testing.T) { ctrl := gomock.NewController(t) + c, deps := newTestController(t, ctrl, headBatch(entity.BatchStateCancelling)) - batch := testBatch() - store := newMockStorage(ctrl, batch) - controller := newTestController(t, ctrl, store, buildfake.New(changesetfake.New()), fmt.Errorf("publish failed")) + pending := pathEntry(entity.SpeculationPathStatusPending, 1) + cancelling := pathEntry(entity.SpeculationPathStatusCancelling, 1) + cancelling.ID = "cancelling-path" - msg := entityqueue.NewMessage(batch.ID, batchIDPayload(t, batch.ID), batch.Queue, nil) - delivery := consumermock.NewMockDelivery(ctrl) - delivery.EXPECT().Message().Return(msg).AnyTimes() - delivery.EXPECT().Attempt().Return(1).AnyTimes() + deps.pathSets.EXPECT().Get(gomock.Any(), headID).Return(entity.SpeculationPathSet{ + Head: headID, Paths: []entity.SpeculationPathEntry{pending, cancelling}, Version: 1, + }, nil) - err := controller.Process(context.Background(), delivery) - assert.Error(t, err) + // No lookups, no Trigger, no Cancel, no publishes: gomock fails the test if + // anything is started or stopped from here. + require.NoError(t, process(t, ctrl, c)) } -func TestController_InterfaceImplementation(t *testing.T) { +// A redelivery means an earlier attempt died without acking — possibly between +// writing a link and publishing its signal, after which the entry may have +// moved to a state no start would touch. Every live linked path therefore gets +// its signal re-published, so no build is left without a poll chain. The +// re-publish dedups against the original signal whenever it was actually sent. +func TestProcess_RedeliveryRepublishesSignalsForLivePaths(t *testing.T) { ctrl := gomock.NewController(t) - batch := testBatch() - store := newMockStorage(ctrl, batch) - controller := newTestController(t, ctrl, store, buildfake.New(changesetfake.New()), nil) - - var _ consumer.Controller = controller + c, deps := newTestController(t, ctrl, headBatch(entity.BatchStateCancelling)) + + building := pathEntry(entity.SpeculationPathStatusBuilding, 1) + building.ID = "building-path" + cancelling := pathEntry(entity.SpeculationPathStatusCancelling, 1) + cancelling.ID = "cancelling-path" + neverDispatched := pathEntry(entity.SpeculationPathStatusCancelling, 1) + neverDispatched.ID = "never-dispatched-path" + done := pathEntry(entity.SpeculationPathStatusCancelled, 1) + done.ID = "done-path" + + deps.pathSets.EXPECT().Get(gomock.Any(), headID).Return(entity.SpeculationPathSet{ + Head: headID, + Paths: []entity.SpeculationPathEntry{building, cancelling, neverDispatched, done}, + }, nil) + + deps.pathBuilds.EXPECT().Get(gomock.Any(), "building-path", 1). + Return(entity.PathBuild{PathID: "building-path", Attempt: 1, BuildID: "build-a"}, nil) + deps.pathBuilds.EXPECT().Get(gomock.Any(), "cancelling-path", 1). + Return(entity.PathBuild{PathID: "cancelling-path", Attempt: 1, BuildID: "build-b"}, nil) + deps.pathBuilds.EXPECT().Get(gomock.Any(), "never-dispatched-path", 1). + Return(entity.PathBuild{}, storage.ErrNotFound) + + expectSignal(t, deps, "build-a") + expectSignal(t, deps, "build-b") + + // The terminal path is never looked up, the unlinked one publishes nothing, + // and no build is started or cancelled from here. + require.NoError(t, processAttempt(t, ctrl, c, 2)) } -// A batch in any halted state (terminal OR cancelling) must short-circuit: -// the build controller acks without triggering an external CI run and without -// publishing anything. Per the cancel design the speculate controller owns -// cancelling in-flight builds and driving the batch terminal, so the build -// stage simply does no work. Cancelling is included because the cancel -// controller is mid-flight; both halted branches reach the same observable -// behaviour (no build performed). -func TestController_Process_HaltedShortCircuit(t *testing.T) { - for _, state := range []entity.BatchState{ - entity.BatchStateCancelled, - entity.BatchStateCancelling, - entity.BatchStateSucceeded, - entity.BatchStateFailed, - } { - t.Run(string(state), func(t *testing.T) { - ctrl := gomock.NewController(t) +// TestProcess_NoPathSet covers a head nothing has speculated on yet. +func TestProcess_NoPathSet(t *testing.T) { + ctrl := gomock.NewController(t) + c, deps := newTestController(t, ctrl, headBatch(entity.BatchStateCreated)) - batch := testBatch() - batch.State = state - store := newMockStorage(ctrl, batch) + deps.pathSets.EXPECT().Get(gomock.Any(), headID).Return(entity.SpeculationPathSet{}, storage.ErrNotFound) - // No Trigger expectation: a stray CI trigger on a halted batch - // fails the test. - br := buildrunnermock.NewMockBuildRunner(ctrl) + require.NoError(t, process(t, ctrl, c)) +} - // Sentinel publish error: the halted path must not publish. If it - // does, Process surfaces this error and require.NoError catches it. - controller := newTestController(t, ctrl, store, br, fmt.Errorf("should not publish")) +func TestProcess_Errors(t *testing.T) { + tests := []struct { + name string + setup func(deps *testDeps, entry entity.SpeculationPathEntry) + }{ + { + name: "path set read failure", + setup: func(deps *testDeps, _ entity.SpeculationPathEntry) { + deps.pathSets.EXPECT().Get(gomock.Any(), headID). + Return(entity.SpeculationPathSet{}, fmt.Errorf("connection reset")) + }, + }, + { + name: "reverse lookup failure", + setup: func(deps *testDeps, entry entity.SpeculationPathEntry) { + deps.pathSets.EXPECT().Get(gomock.Any(), headID).Return(entity.SpeculationPathSet{ + Head: headID, Paths: []entity.SpeculationPathEntry{entry}, Version: 1, + }, nil) + deps.pathBuilds.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any()). + Return(entity.PathBuild{}, fmt.Errorf("connection reset")) + }, + }, + { + name: "trigger failure", + setup: func(deps *testDeps, entry entity.SpeculationPathEntry) { + deps.pathSets.EXPECT().Get(gomock.Any(), headID).Return(entity.SpeculationPathSet{ + Head: headID, Paths: []entity.SpeculationPathEntry{entry}, Version: 1, + }, nil) + notDispatched(deps) + deps.runner.EXPECT().Trigger(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()). + Return(entity.BuildID{}, fmt.Errorf("runner unavailable")) + }, + }, + { + // The build exists and is recorded, but the link write failed for an + // infra reason. Nacking is what repairs it: redelivery re-triggers, + // orphaning build-1 — the accepted cost of not being able to name a + // build before the runner mints its ID. + name: "link write failure", + setup: func(deps *testDeps, entry entity.SpeculationPathEntry) { + deps.pathSets.EXPECT().Get(gomock.Any(), headID).Return(entity.SpeculationPathSet{ + Head: headID, Paths: []entity.SpeculationPathEntry{entry}, Version: 1, + }, nil) + notDispatched(deps) + deps.runner.EXPECT().Trigger(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()). + Return(entity.BuildID{ID: "build-1"}, nil) + deps.builds.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil) + deps.pathBuilds.EXPECT().Create(gomock.Any(), gomock.Any()). + Return(fmt.Errorf("connection reset")) + }, + }, + { + // The build was triggered but could not be recorded. The link is + // never written and nothing is published: a link must never name a + // build that has no record, and a signal must never point the poll + // loop at one. + name: "build record failure", + setup: func(deps *testDeps, entry entity.SpeculationPathEntry) { + deps.pathSets.EXPECT().Get(gomock.Any(), headID).Return(entity.SpeculationPathSet{ + Head: headID, Paths: []entity.SpeculationPathEntry{entry}, Version: 1, + }, nil) + notDispatched(deps) + deps.runner.EXPECT().Trigger(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()). + Return(entity.BuildID{ID: "build-1"}, nil) + deps.builds.EXPECT().Create(gomock.Any(), gomock.Any()). + Return(fmt.Errorf("connection reset")) + }, + }, + } - msg := entityqueue.NewMessage(batch.ID, batchIDPayload(t, batch.ID), batch.Queue, nil) - delivery := consumermock.NewMockDelivery(ctrl) - delivery.EXPECT().Message().Return(msg).AnyTimes() - delivery.EXPECT().Attempt().Return(1).AnyTimes() + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + c, deps := newTestController(t, ctrl, headBatch(entity.BatchStateSpeculating)) + tt.setup(deps, pathEntry(entity.SpeculationPathStatusPending, 1)) - require.NoError(t, controller.Process(context.Background(), delivery)) + require.Error(t, process(t, ctrl, c)) }) } } + +func TestController_InterfaceImplementation(t *testing.T) { + ctrl := gomock.NewController(t) + c, _ := newTestController(t, ctrl, headBatch(entity.BatchStateSpeculating)) + var _ consumer.Controller = c +}