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 +}