From 6cabb60f1e3f9f282681e411f02648967843cb8e Mon Sep 17 00:00:00 2001 From: Preetam Dwivedi Date: Thu, 30 Jul 2026 13:51:08 -0700 Subject: [PATCH] feat(orchestrator): poll builds and stop the ones nothing wants MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary ### Why? The poll loop was writing the path set on every terminal build — the third concurrent writer on a row where the speculate run, which holds a version across its whole Speculator call, was structurally the one to lose. And with the build stage now start-only, something has to enact cancellation. ### What? The poll loop becomes speculation's kill mechanism. On every poll of a non-terminal build it checks whether anything still wants the build running — batch not halted, the path's entry live and on this build's attempt, the attempt's link naming this very build — and asks the runner to cancel when nothing does. That one level-triggered check subsumes path cancels, batch halts, superseded attempts, and lost dispatch races: no cancel message exists to go stale, and a check that misses one poll is remade on the next. It cannot cancel a wanted build: every "unwanted" condition is permanent once true, so a stale read only errs toward keeping, and store anomalies (a set, entry, or link that cannot legitimately be missing) also keep the build — a cancel is irreversible. The Cancel call is best-effort so a failure never kills the poll chain that would retry it. The path set is read as that kill list and never written. Polls now partition on the build ID rather than the batch, and each re-poll mints a distinct message ID so the queue never dedups it away. The halted short-circuit stays removed: a cancelling batch reaches terminal only once its builds stop, and this loop is both what stops them and what watches them stop. ## Test Plan ✅ `bazel test //submitqueue/orchestrator/...` — every unwanted condition cancels; a wanted build never sees a Cancel; anomalies keep the build; a failed Cancel does not fail the poll; statuses recorded per terminal state; and no path-set write happens at all (the set store is wired read-only on the mock). ✅ `make fmt`, `make gazelle` --- .../controller/buildsignal/BUILD.bazel | 4 +- .../controller/buildsignal/buildsignal.go | 289 ++++++--- .../buildsignal/buildsignal_test.go | 574 ++++++++++++------ .../controller/dlq/buildsignal.go | 36 +- .../controller/dlq/buildsignal_test.go | 42 +- 5 files changed, 621 insertions(+), 324 deletions(-) diff --git a/submitqueue/orchestrator/controller/buildsignal/BUILD.bazel b/submitqueue/orchestrator/controller/buildsignal/BUILD.bazel index 96799c87..e79b4ad1 100644 --- a/submitqueue/orchestrator/controller/buildsignal/BUILD.bazel +++ b/submitqueue/orchestrator/controller/buildsignal/BUILD.bazel @@ -6,9 +6,9 @@ go_library( importpath = "github.com/uber/submitqueue/submitqueue/orchestrator/controller/buildsignal", 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", @@ -25,11 +25,11 @@ go_test( deps = [ "//platform/base/messagequeue:go_default_library", "//platform/consumer:go_default_library", - "//platform/errs:go_default_library", "//platform/extension/messagequeue/mock:go_default_library", "//submitqueue/core/topickey:go_default_library", "//submitqueue/entity:go_default_library", "//submitqueue/extension/buildrunner/mock:go_default_library", + "//submitqueue/extension/storage:go_default_library", "//submitqueue/extension/storage/mock:go_default_library", "@com_github_stretchr_testify//assert:go_default_library", "@com_github_stretchr_testify//require:go_default_library", diff --git a/submitqueue/orchestrator/controller/buildsignal/buildsignal.go b/submitqueue/orchestrator/controller/buildsignal/buildsignal.go index d8bd67c1..597cc29f 100644 --- a/submitqueue/orchestrator/controller/buildsignal/buildsignal.go +++ b/submitqueue/orchestrator/controller/buildsignal/buildsignal.go @@ -12,24 +12,42 @@ // See the License for the specific language governing permissions and // limitations under the License. -// Package buildsignal implements the build poll loop. Each message carries -// a Build; the controller calls BuildRunner.Status, writes the latest -// status to the BuildStore, publishes the batch ID to TopicKeySpeculate -// so the state machine re-evaluates, and re-publishes itself via -// PublishAfter when the build has not yet reached a terminal state. Each -// buildID partitions independently, so slow polls on one build do not -// block others. A webhook-capable backend can publish into this same -// topic — the controller cannot tell a poll-driven message from a push. +// Package buildsignal implements the build poll loop. Each message names one +// build by the runner's own ID; the controller calls BuildRunner.Status, writes +// the latest status to that build's record, and wakes the speculate run for its +// batch, re-publishing itself via PublishAfter while the build is still in +// flight. +// +// The poll loop is also where builds are stopped. It follows every build to a +// terminal state anyway, so on each poll it checks whether anything still +// wants the build running — the batch not halted, the path's current attempt, +// with a status that is not a stop, linked to this very build — and asks the +// runner to cancel when nothing does. That makes cancellation level-triggered: +// the speculate run records intent in the path set and nothing more, no cancel +// message exists to go stale, and a check that misses one poll is remade on +// the next. +// +// The path set is read here as that kill list, and never written — the +// speculate run stays its only writer, which is what lets a run hold one +// version of a head's paths across its whole decision without a poll +// invalidating it. The read must come from the primary: a stale replica read +// could report a wanted build unwanted, and a cancel is irreversible. +// +// Each build partitions independently, so slow polls on one build do not block +// another, and successive polls of one build stay ordered. A webhook-capable +// backend can publish into this same topic — the controller cannot tell a +// poll-driven message from a push. package buildsignal import ( "context" + "errors" "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" @@ -52,6 +70,9 @@ var ( PollDelayRunningMs int64 = 2000 ) +// opName is the metric operation name shared by every emit in this file. +const opName = "process" + // Controller consumes build signal messages, polls BuildRunner.Status, // persists the result, and drives the polling loop. type Controller struct { @@ -88,22 +109,30 @@ func NewController( } } -// Process polls the build's current status, persists it, publishes the -// batch ID to speculate so the state machine re-evaluates, and re-publishes -// a delayed message back to this topic when the build is still in flight. +// Process polls one attempt's build status, stops the build if nothing wants +// it running any more, persists the status, wakes the speculate run, and +// re-publishes a delayed message while the build is still in flight. // Returns nil to ack (success), or error to nack/reject. // -// Error classification: deserialize, Status, Update, and the speculate -// publish stay non-retryable — they reject straight to DLQ on the first -// failure, where the operational republish path is the recovery mechanism. -// Only the PublishAfter self-reschedule is retryable: it is the poll loop's -// heartbeat and runs only after status/persist/speculate have all succeeded, -// so a transient enqueue blip nacks and replays (up to MaxAttempts) rather -// than silently stalling the build, then still falls through to DLQ if it -// persists. +// There is deliberately no short-circuit for halted batches. A cancelling batch +// reaches its terminal state only once its paths stop, and this loop is the +// only thing watching them stop — and, now, the thing stopping them: speculate +// marks a path cancelling, the next poll here asks the runner to cancel, and a +// later poll observes CI actually stop and records it. Skipping the work — +// including the reschedule — for a halted batch would leave every cancelled +// batch stranded in Cancelling forever. +// +// Error classification: deserialize, Status, the kill-list reads, the +// persistence writes, and the speculate publish stay non-retryable — they +// reject straight to DLQ on the first failure, where the operational republish +// path is the recovery mechanism. The Cancel call is best-effort instead: +// failing the message for it would kill the poll chain that is the only thing +// that will retry the cancel, so a failure is logged and the next poll remakes +// the whole check. Only the PublishAfter self-reschedule is retryable: it is +// the poll loop's heartbeat and runs only after everything else has succeeded, +// so a transient enqueue blip nacks and replays rather than silently stalling +// the build, then still falls through to DLQ if it persists. func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) error { - const opName = "process" - msg := delivery.Message() buildID, err := entity.BuildIDFromBytes(msg.Payload) @@ -113,30 +142,27 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er return fmt.Errorf("failed to deserialize build ID: %w", err) } - // Only the build ID travels on the queue; load the full Build from - // storage, which is the single source of truth for its BatchID and the - // snapshot the poll loop updates. + // Only the build ID travels on the queue; the record is the source of truth + // for which batch this build belongs to and what it last reported. build, err := c.store.GetBuildStore().Get(ctx, buildID.ID) if err != nil { metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) return fmt.Errorf("failed to get build %s: %w", buildID.ID, err) } - c.logger.Debugw("polling build status", - "build_id", build.ID, - "batch_id", build.BatchID, - "attempt", delivery.Attempt(), - "partition_key", msg.PartitionKey, - ) - - // Load the batch first: it gives us the queue (needed to build the right - // BuildRunner) and lets us short-circuit halted batches before polling. batch, err := c.store.GetBatchStore().Get(ctx, build.BatchID) if err != nil { metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) return fmt.Errorf("failed to get batch %s: %w", build.BatchID, err) } + c.logger.Debugw("polling build status", + "build_id", build.ID, + "batch_id", build.BatchID, + "delivery_attempt", delivery.Attempt(), + "partition_key", msg.PartitionKey, + ) + buildRunner, err := c.buildRunners.For(buildrunner.Config{QueueName: batch.Queue}) if err != nil { metrics.NamedCounter(c.metricsScope, opName, "status_errors", 1) @@ -149,30 +175,47 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er return fmt.Errorf("failed to get status for build %s: %w", buildID.ID, err) } - // Short-circuit if the batch is already halted (terminal OR cancelling). - // Speculate is already idempotent on terminal, but skipping the publish - // avoids noise. For Cancelling batches the cancel controller owns the - // terminal write and the downstream fan-out, so further pipeline work - // would race against it; silent ack is the only safe action. - if entity.IsBatchStateHalted(batch.State) { - metrics.NamedCounter(c.metricsScope, opName, "skipped_halted", 1) - c.logger.Infow("skipping buildsignal publish for halted batch", - "batch_id", batch.ID, - "state", string(batch.State), - ) - return nil + // Reconcile before recording: a build still running that nothing wants any + // more is asked to stop. Best-effort by design — the reschedule below is + // what guarantees the request is remade, so a failed Cancel must not fail + // the message and take that reschedule with it. + if !status.IsTerminal() { + stop, err := c.unwanted(ctx, batch, build) + if err != nil { + return err + } + if stop { + if err := buildRunner.Cancel(ctx, buildID); err != nil { + metrics.NamedCounter(c.metricsScope, opName, "cancel_errors", 1) + c.logger.Warnw("failed to cancel an unwanted build; the next poll retries", + "build_id", build.ID, + "batch_id", build.BatchID, + "error", err, + ) + } else { + metrics.NamedCounter(c.metricsScope, opName, "build_cancelled", 1) + c.logger.Infow("requested cancellation of a build nothing wants running", + "build_id", build.ID, + "batch_id", build.BatchID, + "path_id", build.PathID, + "attempt", build.Attempt, + ) + } + } } - updatedBuild := build - updatedBuild.Status = status - - if err := c.store.GetBuildStore().Update(ctx, updatedBuild); err != nil { - metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) - return fmt.Errorf("failed to update status for build %s: %w", build.ID, err) + if status != build.Status { + build.Status = status + if err := c.store.GetBuildStore().Update(ctx, build); err != nil { + metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) + return fmt.Errorf("failed to update status for build %s: %w", build.ID, err) + } } - // Re-evaluate the batch state machine with the latest build status. - if err := c.publishBatchID(ctx, topickey.TopicKeySpeculate, updatedBuild.BatchID, msg.PartitionKey); err != nil { + // Wake the speculate run so it re-plans the queue with this result. It + // reads the status from the record above rather than being told it, so a + // duplicated or reordered signal costs nothing. + if err := c.publishBatchID(ctx, topickey.TopicKeySpeculate, batch.ID, batch.Queue); err != nil { metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1) return fmt.Errorf("failed to publish to speculate: %w", err) } @@ -180,8 +223,8 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er if status.IsTerminal() { metrics.NamedCounter(c.metricsScope, opName, "terminal", 1, metrics.NewTag("status", string(status))) c.logger.Infow("build reached terminal status", - "build_id", updatedBuild.ID, - "batch_id", updatedBuild.BatchID, + "build_id", build.ID, + "batch_id", build.BatchID, "status", string(status), ) return nil @@ -189,13 +232,13 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er delayMs := pollDelay(status) metrics.NamedCounter(c.metricsScope, opName, "rescheduled", 1, metrics.NewTag("status", string(status))) - if err := c.publishBuild(ctx, c.topicKey, updatedBuild, delayMs); err != nil { + if err := c.publishBuildID(ctx, c.topicKey, buildID, delayMs); err != nil { metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1) return fmt.Errorf("failed to re-publish to buildsignal: %w", err) } c.logger.Debugw("rescheduled build status poll", - "build_id", updatedBuild.ID, + "build_id", build.ID, "status", string(status), "delay_ms", delayMs, ) @@ -213,59 +256,119 @@ func pollDelay(status entity.BuildStatus) int64 { } } -// publishBuild publishes a build's ID to the topic identified by key. delayMs > 0 -// uses PublishAfter; otherwise it uses Publish. Only the identifier travels on -// the queue — the consumer reloads the full Build from storage. -func (c *Controller) publishBuild(ctx context.Context, key consumer.TopicKey, build entity.Build, delayMs int64) error { - payload, err := entity.BuildID{ID: build.ID}.ToBytes() - if err != nil { - return fmt.Errorf("failed to serialize build ID: %w", err) +// unwanted reports whether nothing wants this build running any more: its +// batch has halted, its path was called off or has moved to another attempt, +// or the attempt's link names a different build (this one lost a dispatch +// race). Every one of those conditions is permanent once true, so a stale read +// can only err toward keeping a build — never toward cancelling a wanted one. +// +// The two anomaly cases run the other way on purpose. A missing set or entry +// cannot legitimately happen — the dispatch read the entry out of the set to +// start this build, and entries are not removed — and a missing link cannot +// either, because the signal that led here is published after the link. Both +// therefore indicate store corruption, and since a cancel is irreversible, a +// corrupt kill list keeps the build rather than killing it; a halted batch is +// still caught by the first check, which needs none of those records. +func (c *Controller) unwanted(ctx context.Context, batch entity.Batch, build entity.Build) (bool, error) { + if entity.IsBatchStateHalted(batch.State) { + return true, nil } - msg := entityqueue.NewMessage(build.ID, payload, build.BatchID, nil) + // A build without path coordinates predates per-path dispatch; the batch + // state above is the only kill list it has. + if build.PathID == "" { + return false, nil + } - q, ok := c.registry.Queue(key) - if !ok { - return fmt.Errorf("no queue registered for topic key %s", key) + set, err := c.store.GetSpeculationPathSetStore().Get(ctx, batch.ID) + if err != nil { + if errors.Is(err, storage.ErrNotFound) { + metrics.NamedCounter(c.metricsScope, opName, "kill_list_anomalies", 1) + c.logger.Warnw("build exists but its head has no path set; keeping the build", + "build_id", build.ID, + "batch_id", batch.ID, + ) + return false, nil + } + metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) + return false, fmt.Errorf("failed to get path set for batch %s: %w", batch.ID, err) } - topicName, ok := c.registry.TopicName(key) - if !ok { - return fmt.Errorf("no topic name registered for topic key %s", key) + entry, found := findEntry(set, build.PathID) + if !found { + metrics.NamedCounter(c.metricsScope, opName, "kill_list_anomalies", 1) + c.logger.Warnw("build exists but its path is gone from the set; keeping the build", + "build_id", build.ID, + "batch_id", batch.ID, + "path_id", build.PathID, + ) + return false, nil } - publisher := q.Publisher() - if delayMs > 0 { - return publisher.PublishAfter(ctx, topicName, msg, delayMs) + switch entry.Status { + case entity.SpeculationPathStatusCancelling, entity.SpeculationPathStatusCancelled: + return true, nil } - return publisher.Publish(ctx, topicName, msg) -} -// publishBatchID publishes a batch ID to the topic identified by key. -func (c *Controller) publishBatchID(ctx context.Context, key consumer.TopicKey, batchID string, partitionKey string) error { - bid := entity.BatchID{ID: batchID} - payload, err := bid.ToBytes() + // The path has moved on to a newer attempt; this build belongs to a + // superseded one. + if entry.Attempt != build.Attempt { + return true, nil + } + + link, err := c.store.GetPathBuildStore().Get(ctx, build.PathID, build.Attempt) if err != nil { - return fmt.Errorf("failed to serialize batch ID: %w", err) + if errors.Is(err, storage.ErrNotFound) { + metrics.NamedCounter(c.metricsScope, opName, "kill_list_anomalies", 1) + c.logger.Warnw("build exists but its attempt has no link; keeping the build", + "build_id", build.ID, + "path_id", build.PathID, + "attempt", build.Attempt, + ) + return false, nil + } + metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) + return false, fmt.Errorf("failed to look up build for path %s attempt %d: %w", build.PathID, build.Attempt, err) } - msg := entityqueue.NewMessage(batchID, payload, partitionKey, nil) + // The attempt's build is a different one: this build lost the dispatch + // race, and nothing downstream will ever look at it. + return link.BuildID != build.ID, nil +} - q, ok := c.registry.Queue(key) - if !ok { - return fmt.Errorf("no queue registered for topic key %s", key) +// findEntry returns the set's entry for a path ID. +func findEntry(set entity.SpeculationPathSet, pathID string) (entity.SpeculationPathEntry, bool) { + for _, entry := range set.Paths { + if entry.ID == pathID { + return entry, true + } } + return entity.SpeculationPathEntry{}, false +} - topicName, ok := c.registry.TopicName(key) - if !ok { - return fmt.Errorf("no topic name registered for topic key %s", key) +// publishBuildID re-publishes a build to the topic identified by key. delayMs > 0 +// delays delivery by that long. The build ID is also the partition key, so one +// build's polls stay ordered while builds of the same batch proceed +// independently. Each poll gets a distinct message ID (publish.UniqueID) so the +// queue never deduplicates a re-poll against an earlier message for the same +// build. +func (c *Controller) publishBuildID(ctx context.Context, key consumer.TopicKey, buildID entity.BuildID, delayMs int64) error { + payload, err := buildID.ToBytes() + if err != nil { + return fmt.Errorf("failed to serialize build ID: %w", err) } + return publish.Message(ctx, c.registry, key, publish.UniqueID(buildID.ID), payload, buildID.ID, delayMs) +} - if err := q.Publisher().Publish(ctx, topicName, msg); err != nil { - return fmt.Errorf("failed to publish message: %w", err) +// publishBatchID publishes a batch ID to the topic identified by key, with a +// distinct message ID per publish (publish.UniqueID) so a later wake-up for the +// same batch is never deduplicated away. +func (c *Controller) publishBatchID(ctx context.Context, key consumer.TopicKey, batchID string, partitionKey string) error { + payload, err := entity.BatchID{ID: batchID}.ToBytes() + if err != nil { + return fmt.Errorf("failed to serialize batch ID: %w", err) } - - return nil + return publish.Message(ctx, c.registry, key, publish.UniqueID(batchID), payload, partitionKey, 0) } // Name returns the controller name for logging and metrics. diff --git a/submitqueue/orchestrator/controller/buildsignal/buildsignal_test.go b/submitqueue/orchestrator/controller/buildsignal/buildsignal_test.go index 0953bf9e..5f27e746 100644 --- a/submitqueue/orchestrator/controller/buildsignal/buildsignal_test.go +++ b/submitqueue/orchestrator/controller/buildsignal/buildsignal_test.go @@ -24,29 +24,38 @@ import ( "github.com/uber-go/tally" entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" - "github.com/uber/submitqueue/platform/errs" queuemock "github.com/uber/submitqueue/platform/extension/messagequeue/mock" "github.com/uber/submitqueue/submitqueue/core/topickey" "github.com/uber/submitqueue/submitqueue/entity" 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" "go.uber.org/mock/gomock" "go.uber.org/zap/zaptest" ) +const ( + testPathID = "path-abc" + testAttempt = 2 + testBuildID = "build-7" + testBatchID = "test-queue/batch/head" +) + // testHarness wires a Controller against mock queues for two topic keys // (buildsignal and speculate) so individual tests can assert which // Publish / PublishAfter happens. type testHarness struct { controller *Controller br *buildrunnermock.MockBuildRunner - buildStore *storagemock.MockBuildStore + builds *storagemock.MockBuildStore batchStore *storagemock.MockBatchStore + pathSets *storagemock.MockSpeculationPathSetStore + pathBuilds *storagemock.MockPathBuildStore signalPub *queuemock.MockPublisher speculatePub *queuemock.MockPublisher } -func newTestHarness(t *testing.T, ctrl *gomock.Controller) *testHarness { +func newTestHarness(t *testing.T, ctrl *gomock.Controller, batchState entity.BatchState) *testHarness { br := buildrunnermock.NewMockBuildRunner(ctrl) brFactory := buildrunnermock.NewMockFactory(ctrl) brFactory.EXPECT().For(gomock.Any()).Return(br, nil).AnyTimes() @@ -65,11 +74,23 @@ func newTestHarness(t *testing.T, ctrl *gomock.Controller) *testHarness { }) require.NoError(t, err) - buildStore := storagemock.NewMockBuildStore(ctrl) + builds := storagemock.NewMockBuildStore(ctrl) batchStore := storagemock.NewMockBatchStore(ctrl) + batchStore.EXPECT().Get(gomock.Any(), testBatchID).Return(entity.Batch{ + ID: testBatchID, Queue: "test-queue", State: batchState, Version: 1, + }, nil).AnyTimes() + + // The path set and link stores are wired read-only: their getters answer, + // but no write expectation exists, so any Create/Update from this + // controller fails the test. + pathSets := storagemock.NewMockSpeculationPathSetStore(ctrl) + pathBuilds := storagemock.NewMockPathBuildStore(ctrl) + store := storagemock.NewMockStorage(ctrl) - store.EXPECT().GetBuildStore().Return(buildStore).AnyTimes() + store.EXPECT().GetBuildStore().Return(builds).AnyTimes() store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() + store.EXPECT().GetSpeculationPathSetStore().Return(pathSets).AnyTimes() + store.EXPECT().GetPathBuildStore().Return(pathBuilds).AnyTimes() c := NewController( zaptest.NewLogger(t).Sugar(), @@ -83,30 +104,57 @@ func newTestHarness(t *testing.T, ctrl *gomock.Controller) *testHarness { return &testHarness{ controller: c, br: br, - buildStore: buildStore, + builds: builds, batchStore: batchStore, + pathSets: pathSets, + pathBuilds: pathBuilds, signalPub: signalPub, speculatePub: speculatePub, } } -// buildDelivery builds a delivery whose payload is the build's ID, matching -// the on-queue contract: only the identifier travels, the consumer loads the -// full Build from storage. -func buildDelivery(t *testing.T, ctrl *gomock.Controller, b entity.Build) consumer.Delivery { +// wanted wires the kill-list reads to say the build is still wanted: its entry +// is live on the build's own attempt, and the attempt's link names this build. +func (h *testHarness) wanted() { + h.pathSets.EXPECT().Get(gomock.Any(), testBatchID).Return(entity.SpeculationPathSet{ + Head: testBatchID, + Paths: []entity.SpeculationPathEntry{ + {ID: testPathID, Status: entity.SpeculationPathStatusBuilding, Attempt: testAttempt}, + }, + }, nil).AnyTimes() + h.pathBuilds.EXPECT().Get(gomock.Any(), testPathID, testAttempt).Return(entity.PathBuild{ + PathID: testPathID, Attempt: testAttempt, BuildID: testBuildID, + }, nil).AnyTimes() +} + +// delivery builds a delivery whose payload is the attempt's key, matching the +// on-queue contract: only the identifier travels, and the consumer loads the +// execution record — including the build ID — from storage. +func delivery(t *testing.T, ctrl *gomock.Controller) consumer.Delivery { t.Helper() - payload, err := entity.BuildID{ID: b.ID}.ToBytes() + payload, err := entity.BuildID{ID: testBuildID}.ToBytes() require.NoError(t, err) - msg := entityqueue.NewMessage(b.ID, payload, b.BatchID, nil) + msg := entityqueue.NewMessage(testBuildID, payload, testBuildID, nil) d := queuemock.NewMockDelivery(ctrl) d.EXPECT().Message().Return(msg).AnyTimes() d.EXPECT().Attempt().Return(1).AnyTimes() return d } +// testBuild returns the build under test, in the given status. +func testBuild(status entity.BuildStatus) entity.Build { + return entity.Build{ + ID: testBuildID, + BatchID: testBatchID, + PathID: testPathID, + Attempt: testAttempt, + Status: status, + } +} + func TestController_Identity(t *testing.T) { ctrl := gomock.NewController(t) - h := newTestHarness(t, ctrl) + h := newTestHarness(t, ctrl, entity.BatchStateSpeculating) assert.Equal(t, "buildsignal", h.controller.Name()) assert.Equal(t, topickey.TopicKeyBuildSignal, h.controller.TopicKey()) @@ -115,10 +163,10 @@ func TestController_Identity(t *testing.T) { var _ consumer.Controller = h.controller } -// TestController_Process_Terminal verifies a terminal poll persists the -// status, publishes the batch ID to speculate, and does NOT re-publish to -// buildsignal. -func TestController_Process_Terminal(t *testing.T) { +// The poll loop records status on the build and wakes the run. It must never +// write the path set — that is the speculate run's state, and it is read here +// only as the kill list. +func TestProcess_RecordsStatusAndNeverWritesThePathSet(t *testing.T) { tests := []struct { name string status entity.BuildStatus @@ -131,194 +179,382 @@ func TestController_Process_Terminal(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { ctrl := gomock.NewController(t) - h := newTestHarness(t, ctrl) - - build := entity.Build{ID: "b-1", BatchID: "batch-1", Status: entity.BuildStatusAccepted} - updatedBuild := build - updatedBuild.Status = tt.status - - h.buildStore.EXPECT().Get(gomock.Any(), build.ID).Return(build, nil) - h.br.EXPECT().Status(gomock.Any(), entity.BuildID{ID: build.ID}).Return(tt.status, entity.BuildMetadata{}, nil) - h.batchStore.EXPECT().Get(gomock.Any(), build.BatchID).Return(entity.Batch{ID: build.BatchID, State: entity.BatchStateSpeculating}, nil) - h.buildStore.EXPECT().Update(gomock.Any(), updatedBuild).Return(nil) - h.speculatePub.EXPECT(). - Publish(gomock.Any(), "speculate", gomock.AssignableToTypeOf(entityqueue.Message{})). - DoAndReturn(func(_ context.Context, _ string, msg entityqueue.Message) error { - bid, err := entity.BatchIDFromBytes(msg.Payload) - require.NoError(t, err) - assert.Equal(t, build.BatchID, bid.ID) - return nil - }).Times(1) - // No PublishAfter expected on terminal. - - err := h.controller.Process(context.Background(), buildDelivery(t, ctrl, build)) - require.NoError(t, err) + h := newTestHarness(t, ctrl, entity.BatchStateSpeculating) + + h.builds.EXPECT().Get(gomock.Any(), testBuildID).Return(testBuild(entity.BuildStatusRunning), nil) + h.br.EXPECT().Status(gomock.Any(), entity.BuildID{ID: testBuildID}).Return(tt.status, nil, nil) + h.builds.EXPECT().Update(gomock.Any(), testBuild(tt.status)).Return(nil) + + h.speculatePub.EXPECT().Publish(gomock.Any(), "speculate", gomock.Any()).Return(nil) + // Terminal: no reschedule, no Cancel, and not even a kill-list read + // — a finished build has nothing left to stop. Any path-set write + // fails the test (none is expected on the mock). + + require.NoError(t, h.controller.Process(context.Background(), delivery(t, ctrl))) }) } } -// TestController_Process_NonTerminal verifies a non-terminal poll persists -// the status, publishes to speculate, AND re-publishes to buildsignal via -// PublishAfter with the per-status delay. -func TestController_Process_NonTerminal(t *testing.T) { +// While a build is in flight the loop reschedules itself, keeping the build ID +// as the partition key so one build's polls stay ordered. +func TestProcess_NonTerminalReschedulesOnTheBuildPartition(t *testing.T) { + ctrl := gomock.NewController(t) + h := newTestHarness(t, ctrl, entity.BatchStateSpeculating) + h.wanted() + + h.builds.EXPECT().Get(gomock.Any(), testBuildID).Return(testBuild(entity.BuildStatusAccepted), nil) + h.br.EXPECT().Status(gomock.Any(), gomock.Any()).Return(entity.BuildStatusRunning, nil, nil) + h.builds.EXPECT().Update(gomock.Any(), testBuild(entity.BuildStatusRunning)).Return(nil) + + h.speculatePub.EXPECT().Publish(gomock.Any(), "speculate", gomock.Any()).Return(nil) + h.signalPub.EXPECT().PublishAfter(gomock.Any(), "buildsignal", gomock.Any(), PollDelayRunningMs). + DoAndReturn(func(_ context.Context, _ string, msg entityqueue.Message, _ int64) error { + assert.Equal(t, testBuildID, msg.PartitionKey, + "a build's polls must stay on its own partition") + assert.NotEqual(t, testBuildID, msg.ID, + "a distinct message ID per poll, or the queue dedups the re-poll") + return nil + }) + + require.NoError(t, h.controller.Process(context.Background(), delivery(t, ctrl))) +} + +// An unchanged status is not rewritten on every poll of a long build. +func TestProcess_UnchangedStatusSkipsWrite(t *testing.T) { + ctrl := gomock.NewController(t) + h := newTestHarness(t, ctrl, entity.BatchStateSpeculating) + h.wanted() + + h.builds.EXPECT().Get(gomock.Any(), testBuildID).Return(testBuild(entity.BuildStatusRunning), nil) + h.br.EXPECT().Status(gomock.Any(), gomock.Any()).Return(entity.BuildStatusRunning, nil, nil) + // No Update expected. + h.speculatePub.EXPECT().Publish(gomock.Any(), "speculate", gomock.Any()).Return(nil) + h.signalPub.EXPECT().PublishAfter(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil) + + require.NoError(t, h.controller.Process(context.Background(), delivery(t, ctrl))) +} + +// TestProcess_StopsUnwantedBuilds covers the kill list: a running build is +// cancelled the moment nothing wants it — its path called off, its attempt +// superseded, its link naming a different build, or its whole batch halted — +// and the poll keeps running so a later poll records the stop. +func TestProcess_StopsUnwantedBuilds(t *testing.T) { tests := []struct { - name string - status entity.BuildStatus - wantDelayMs int64 + name string + batchState entity.BatchState + setup func(h *testHarness) }{ - {"accepted uses accepted delay", entity.BuildStatusAccepted, PollDelayAcceptedMs}, - {"running uses running delay", entity.BuildStatusRunning, PollDelayRunningMs}, + { + name: "path cancelling", + batchState: entity.BatchStateSpeculating, + setup: func(h *testHarness) { + h.pathSets.EXPECT().Get(gomock.Any(), testBatchID).Return(entity.SpeculationPathSet{ + Head: testBatchID, + Paths: []entity.SpeculationPathEntry{ + {ID: testPathID, Status: entity.SpeculationPathStatusCancelling, Attempt: testAttempt}, + }, + }, nil) + }, + }, + { + name: "path cancelled", + batchState: entity.BatchStateSpeculating, + setup: func(h *testHarness) { + h.pathSets.EXPECT().Get(gomock.Any(), testBatchID).Return(entity.SpeculationPathSet{ + Head: testBatchID, + Paths: []entity.SpeculationPathEntry{ + {ID: testPathID, Status: entity.SpeculationPathStatusCancelled, Attempt: testAttempt}, + }, + }, nil) + }, + }, + { + name: "attempt superseded", + batchState: entity.BatchStateSpeculating, + setup: func(h *testHarness) { + h.pathSets.EXPECT().Get(gomock.Any(), testBatchID).Return(entity.SpeculationPathSet{ + Head: testBatchID, + Paths: []entity.SpeculationPathEntry{ + {ID: testPathID, Status: entity.SpeculationPathStatusPending, Attempt: testAttempt + 1}, + }, + }, nil) + }, + }, + { + name: "link names another build", + batchState: entity.BatchStateSpeculating, + setup: func(h *testHarness) { + h.pathSets.EXPECT().Get(gomock.Any(), testBatchID).Return(entity.SpeculationPathSet{ + Head: testBatchID, + Paths: []entity.SpeculationPathEntry{ + {ID: testPathID, Status: entity.SpeculationPathStatusBuilding, Attempt: testAttempt}, + }, + }, nil) + h.pathBuilds.EXPECT().Get(gomock.Any(), testPathID, testAttempt).Return(entity.PathBuild{ + PathID: testPathID, Attempt: testAttempt, BuildID: "build-winner", + }, nil) + }, + }, + { + name: "batch halted", + batchState: entity.BatchStateCancelling, + setup: func(h *testHarness) { + // No kill-list reads: the batch state alone decides, so no + // set/link expectations exist and any read fails the test. + }, + }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { ctrl := gomock.NewController(t) - h := newTestHarness(t, ctrl) - - build := entity.Build{ID: "b-2", BatchID: "batch-2", Status: entity.BuildStatusAccepted} - updatedBuild := build - updatedBuild.Status = tt.status - - h.buildStore.EXPECT().Get(gomock.Any(), build.ID).Return(build, nil) - h.br.EXPECT().Status(gomock.Any(), entity.BuildID{ID: build.ID}).Return(tt.status, entity.BuildMetadata{}, nil) - h.batchStore.EXPECT().Get(gomock.Any(), build.BatchID).Return(entity.Batch{ID: build.BatchID, State: entity.BatchStateSpeculating}, nil) - h.buildStore.EXPECT().Update(gomock.Any(), updatedBuild).Return(nil) - h.speculatePub.EXPECT().Publish(gomock.Any(), "speculate", gomock.Any()).Return(nil).Times(1) - h.signalPub.EXPECT(). - PublishAfter(gomock.Any(), "buildsignal", gomock.AssignableToTypeOf(entityqueue.Message{}), tt.wantDelayMs). - DoAndReturn(func(_ context.Context, _ string, msg entityqueue.Message, _ int64) error { - bid, err := entity.BuildIDFromBytes(msg.Payload) - require.NoError(t, err) - // Re-published payload carries only the build ID. - assert.Equal(t, build.ID, bid.ID) - return nil - }).Times(1) - - err := h.controller.Process(context.Background(), buildDelivery(t, ctrl, build)) - require.NoError(t, err) + h := newTestHarness(t, ctrl, tt.batchState) + tt.setup(h) + + h.builds.EXPECT().Get(gomock.Any(), testBuildID).Return(testBuild(entity.BuildStatusRunning), nil) + h.br.EXPECT().Status(gomock.Any(), gomock.Any()).Return(entity.BuildStatusRunning, nil, nil) + h.br.EXPECT().Cancel(gomock.Any(), entity.BuildID{ID: testBuildID}).Return(nil) + + h.speculatePub.EXPECT().Publish(gomock.Any(), "speculate", gomock.Any()).Return(nil) + h.signalPub.EXPECT().PublishAfter(gomock.Any(), "buildsignal", gomock.Any(), PollDelayRunningMs).Return(nil) + + require.NoError(t, h.controller.Process(context.Background(), delivery(t, ctrl))) }) } } -func TestController_Process_StatusError(t *testing.T) { - ctrl := gomock.NewController(t) - h := newTestHarness(t, ctrl) +// TestProcess_KeepsWantedBuilds is the inverse: a live entry on the build's own +// attempt, linked to this very build, must never be cancelled. +func TestProcess_KeepsWantedBuilds(t *testing.T) { + for _, status := range []entity.SpeculationPathStatus{ + entity.SpeculationPathStatusPending, + entity.SpeculationPathStatusBuilding, + entity.SpeculationPathStatusPassed, + } { + t.Run(string(status), func(t *testing.T) { + ctrl := gomock.NewController(t) + h := newTestHarness(t, ctrl, entity.BatchStateSpeculating) + + h.pathSets.EXPECT().Get(gomock.Any(), testBatchID).Return(entity.SpeculationPathSet{ + Head: testBatchID, + Paths: []entity.SpeculationPathEntry{ + {ID: testPathID, Status: status, Attempt: testAttempt}, + }, + }, nil) + h.pathBuilds.EXPECT().Get(gomock.Any(), testPathID, testAttempt).Return(entity.PathBuild{ + PathID: testPathID, Attempt: testAttempt, BuildID: testBuildID, + }, nil) + + h.builds.EXPECT().Get(gomock.Any(), testBuildID).Return(testBuild(entity.BuildStatusRunning), nil) + h.br.EXPECT().Status(gomock.Any(), gomock.Any()).Return(entity.BuildStatusRunning, nil, nil) + h.speculatePub.EXPECT().Publish(gomock.Any(), "speculate", gomock.Any()).Return(nil) + h.signalPub.EXPECT().PublishAfter(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil) + + // No Cancel: gomock fails the test if one happens. + require.NoError(t, h.controller.Process(context.Background(), delivery(t, ctrl))) + }) + } +} - build := entity.Build{ID: "b-3", BatchID: "batch-3", Status: entity.BuildStatusAccepted} +// A kill list that cannot legitimately be missing keeps the build when it is: +// a cancel is irreversible, so store anomalies err toward keeping. The halted +// check still covers every real stop that matters for such a batch. +func TestProcess_AnomalousKillListKeepsTheBuild(t *testing.T) { + tests := []struct { + name string + setup func(h *testHarness) + }{ + { + name: "path set missing", + setup: func(h *testHarness) { + h.pathSets.EXPECT().Get(gomock.Any(), testBatchID). + Return(entity.SpeculationPathSet{}, storage.ErrNotFound) + }, + }, + { + name: "entry missing from the set", + setup: func(h *testHarness) { + h.pathSets.EXPECT().Get(gomock.Any(), testBatchID).Return(entity.SpeculationPathSet{ + Head: testBatchID, + Paths: []entity.SpeculationPathEntry{ + {ID: "some-other-path", Status: entity.SpeculationPathStatusBuilding, Attempt: 1}, + }, + }, nil) + }, + }, + { + name: "link missing", + setup: func(h *testHarness) { + h.pathSets.EXPECT().Get(gomock.Any(), testBatchID).Return(entity.SpeculationPathSet{ + Head: testBatchID, + Paths: []entity.SpeculationPathEntry{ + {ID: testPathID, Status: entity.SpeculationPathStatusBuilding, Attempt: testAttempt}, + }, + }, nil) + h.pathBuilds.EXPECT().Get(gomock.Any(), testPathID, testAttempt). + Return(entity.PathBuild{}, storage.ErrNotFound) + }, + }, + } - h.buildStore.EXPECT().Get(gomock.Any(), build.ID).Return(build, nil) - h.batchStore.EXPECT().Get(gomock.Any(), build.BatchID).Return(entity.Batch{ID: build.BatchID, State: entity.BatchStateSpeculating}, nil) - h.br.EXPECT().Status(gomock.Any(), entity.BuildID{ID: build.ID}).Return(entity.BuildStatusUnknown, nil, errors.New("provider down")) - // No Update, no Publish, no PublishAfter expected. + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + h := newTestHarness(t, ctrl, entity.BatchStateSpeculating) + tt.setup(h) - err := h.controller.Process(context.Background(), buildDelivery(t, ctrl, build)) - require.Error(t, err) - // Non-retryable: rejects to DLQ on first failure; republish is the recovery path. - assert.False(t, errs.IsRetryable(err)) -} + h.builds.EXPECT().Get(gomock.Any(), testBuildID).Return(testBuild(entity.BuildStatusRunning), nil) + h.br.EXPECT().Status(gomock.Any(), gomock.Any()).Return(entity.BuildStatusRunning, nil, nil) + h.speculatePub.EXPECT().Publish(gomock.Any(), "speculate", gomock.Any()).Return(nil) + h.signalPub.EXPECT().PublishAfter(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil) -func TestController_Process_UpdateError(t *testing.T) { - ctrl := gomock.NewController(t) - h := newTestHarness(t, ctrl) - - build := entity.Build{ID: "b-4", BatchID: "batch-4", Status: entity.BuildStatusAccepted} - updatedBuild := build - updatedBuild.Status = entity.BuildStatusRunning - - h.buildStore.EXPECT().Get(gomock.Any(), build.ID).Return(build, nil) - h.br.EXPECT().Status(gomock.Any(), entity.BuildID{ID: build.ID}).Return(entity.BuildStatusRunning, nil, nil) - h.batchStore.EXPECT().Get(gomock.Any(), build.BatchID).Return(entity.Batch{ID: build.BatchID, State: entity.BatchStateSpeculating}, nil) - h.buildStore.EXPECT().Update(gomock.Any(), updatedBuild). - Return(errors.New("db unreachable")) - // No Publish / PublishAfter expected after the store failure. - - err := h.controller.Process(context.Background(), buildDelivery(t, ctrl, build)) - require.Error(t, err) - // Non-retryable: rejects to DLQ on first failure; republish is the recovery path. - assert.False(t, errs.IsRetryable(err)) + // No Cancel: gomock fails the test if one happens. + require.NoError(t, h.controller.Process(context.Background(), delivery(t, ctrl))) + }) + } } -// TestController_Process_RepublishError verifies that a failure to re-publish -// the delayed poll message surfaces an error. The preceding -// status/persist/speculate steps all succeed. -func TestController_Process_RepublishError(t *testing.T) { +// A failed Cancel must not fail the message: the reschedule is the only thing +// that will retry the cancel, so the poll survives and remakes the check. +func TestProcess_CancelFailureDoesNotFailThePoll(t *testing.T) { ctrl := gomock.NewController(t) - h := newTestHarness(t, ctrl) - - build := entity.Build{ID: "b-5", BatchID: "batch-5", Status: entity.BuildStatusAccepted} - updatedBuild := build - updatedBuild.Status = entity.BuildStatusRunning - - h.buildStore.EXPECT().Get(gomock.Any(), build.ID).Return(build, nil) - h.br.EXPECT().Status(gomock.Any(), entity.BuildID{ID: build.ID}).Return(entity.BuildStatusRunning, entity.BuildMetadata{}, nil) - h.batchStore.EXPECT().Get(gomock.Any(), build.BatchID).Return(entity.Batch{ID: build.BatchID, State: entity.BatchStateSpeculating}, nil) - h.buildStore.EXPECT().Update(gomock.Any(), updatedBuild).Return(nil) - h.speculatePub.EXPECT().Publish(gomock.Any(), "speculate", gomock.Any()).Return(nil).Times(1) - h.signalPub.EXPECT(). - PublishAfter(gomock.Any(), "buildsignal", gomock.Any(), PollDelayRunningMs). - Return(errors.New("queue unavailable")).Times(1) - - err := h.controller.Process(context.Background(), buildDelivery(t, ctrl, build)) - require.Error(t, err) + h := newTestHarness(t, ctrl, entity.BatchStateCancelling) + + h.builds.EXPECT().Get(gomock.Any(), testBuildID).Return(testBuild(entity.BuildStatusRunning), nil) + h.br.EXPECT().Status(gomock.Any(), gomock.Any()).Return(entity.BuildStatusRunning, nil, nil) + h.br.EXPECT().Cancel(gomock.Any(), entity.BuildID{ID: testBuildID}). + Return(errors.New("runner unavailable")) + + h.speculatePub.EXPECT().Publish(gomock.Any(), "speculate", gomock.Any()).Return(nil) + h.signalPub.EXPECT().PublishAfter(gomock.Any(), "buildsignal", gomock.Any(), PollDelayRunningMs).Return(nil) + + require.NoError(t, h.controller.Process(context.Background(), delivery(t, ctrl))) } -// TestController_Process_GetError verifies that a failure to load the Build -// from storage (only the ID is on the queue) surfaces an error. Non-retryable: -// it rejects to DLQ on first failure, consistent with other storage reads. -func TestController_Process_GetError(t *testing.T) { +// A build without path coordinates predates per-path dispatch. The batch state +// is the only kill list it has: no set or link is read for it. +func TestProcess_BuildWithoutPathChecksOnlyTheBatch(t *testing.T) { ctrl := gomock.NewController(t) - h := newTestHarness(t, ctrl) + h := newTestHarness(t, ctrl, entity.BatchStateSpeculating) - build := entity.Build{ID: "b-6", BatchID: "batch-6", Status: entity.BuildStatusAccepted} + legacy := testBuild(entity.BuildStatusRunning) + legacy.PathID = "" + legacy.Attempt = 0 - h.buildStore.EXPECT().Get(gomock.Any(), build.ID).Return(entity.Build{}, errors.New("db unreachable")) - // No Status / Update / Publish expected once the load fails. + h.builds.EXPECT().Get(gomock.Any(), testBuildID).Return(legacy, nil) + h.br.EXPECT().Status(gomock.Any(), gomock.Any()).Return(entity.BuildStatusRunning, nil, nil) + h.speculatePub.EXPECT().Publish(gomock.Any(), "speculate", gomock.Any()).Return(nil) + h.signalPub.EXPECT().PublishAfter(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil) - err := h.controller.Process(context.Background(), buildDelivery(t, ctrl, build)) - require.Error(t, err) - assert.False(t, errs.IsRetryable(err)) + // No set/link reads and no Cancel: gomock fails the test on any of them. + require.NoError(t, h.controller.Process(context.Background(), delivery(t, ctrl))) } -func TestController_Process_MalformedPayload(t *testing.T) { - ctrl := gomock.NewController(t) - h := newTestHarness(t, ctrl) +// TestProcess_HaltedBatchStillRuns is the cancellation-correctness case. A +// cancelling batch reaches terminal only once its builds stop, and this loop is +// what stops them and watches them stop, so a halted batch must still be +// polled, cancelled, recorded and rescheduled. Short-circuiting strands it in +// Cancelling forever. +func TestProcess_HaltedBatchStillRuns(t *testing.T) { + t.Run("cancels and reschedules while the build runs", func(t *testing.T) { + ctrl := gomock.NewController(t) + h := newTestHarness(t, ctrl, entity.BatchStateCancelling) + + h.builds.EXPECT().Get(gomock.Any(), testBuildID).Return(testBuild(entity.BuildStatusRunning), nil) + h.br.EXPECT().Status(gomock.Any(), gomock.Any()).Return(entity.BuildStatusRunning, nil, nil) + h.br.EXPECT().Cancel(gomock.Any(), entity.BuildID{ID: testBuildID}).Return(nil) + h.speculatePub.EXPECT().Publish(gomock.Any(), "speculate", gomock.Any()).Return(nil) + h.signalPub.EXPECT().PublishAfter(gomock.Any(), "buildsignal", gomock.Any(), PollDelayRunningMs).Return(nil) + + require.NoError(t, h.controller.Process(context.Background(), delivery(t, ctrl))) + }) - msg := entityqueue.NewMessage("bad", []byte(`{"invalid"`), "batch-bad", nil) - d := queuemock.NewMockDelivery(ctrl) - d.EXPECT().Message().Return(msg).AnyTimes() - d.EXPECT().Attempt().Return(1).AnyTimes() + t.Run("still records the terminal status", func(t *testing.T) { + ctrl := gomock.NewController(t) + h := newTestHarness(t, ctrl, entity.BatchStateCancelling) - err := h.controller.Process(context.Background(), d) - require.Error(t, err) -} + h.builds.EXPECT().Get(gomock.Any(), testBuildID).Return(testBuild(entity.BuildStatusRunning), nil) + h.br.EXPECT().Status(gomock.Any(), gomock.Any()).Return(entity.BuildStatusCancelled, nil, nil) + h.builds.EXPECT().Update(gomock.Any(), testBuild(entity.BuildStatusCancelled)).Return(nil) + h.speculatePub.EXPECT().Publish(gomock.Any(), "speculate", gomock.Any()).Return(nil) -// A halted batch (terminal OR cancelling) must short-circuit: just ack, no -// status persist and no publish to speculate. For terminal: speculate is -// already idempotent on terminal, but skipping the publish keeps the system -// from re-emitting noise. For Cancelling: the cancel controller owns the -// terminal write and downstream fan-out, so any further pipeline work would -// race against it. -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) - h := newTestHarness(t, ctrl) + // Terminal: no Cancel and no reschedule. + require.NoError(t, h.controller.Process(context.Background(), delivery(t, ctrl))) + }) +} - build := entity.Build{ID: "b-halt", BatchID: "batch-halt", Status: entity.BuildStatusAccepted} +func TestProcess_Errors(t *testing.T) { + tests := []struct { + name string + setup func(h *testHarness) + }{ + { + name: "build read failure", + setup: func(h *testHarness) { + h.builds.EXPECT().Get(gomock.Any(), testBuildID). + Return(entity.Build{}, errors.New("connection reset")) + }, + }, + { + name: "status failure", + setup: func(h *testHarness) { + h.builds.EXPECT().Get(gomock.Any(), testBuildID).Return(testBuild(entity.BuildStatusRunning), nil) + h.br.EXPECT().Status(gomock.Any(), gomock.Any()). + Return(entity.BuildStatusUnknown, nil, errors.New("runner unavailable")) + }, + }, + { + name: "kill list set read failure", + setup: func(h *testHarness) { + h.builds.EXPECT().Get(gomock.Any(), testBuildID).Return(testBuild(entity.BuildStatusRunning), nil) + h.br.EXPECT().Status(gomock.Any(), gomock.Any()).Return(entity.BuildStatusRunning, nil, nil) + h.pathSets.EXPECT().Get(gomock.Any(), testBatchID). + Return(entity.SpeculationPathSet{}, errors.New("connection reset")) + }, + }, + { + name: "kill list link read failure", + setup: func(h *testHarness) { + h.builds.EXPECT().Get(gomock.Any(), testBuildID).Return(testBuild(entity.BuildStatusRunning), nil) + h.br.EXPECT().Status(gomock.Any(), gomock.Any()).Return(entity.BuildStatusRunning, nil, nil) + h.pathSets.EXPECT().Get(gomock.Any(), testBatchID).Return(entity.SpeculationPathSet{ + Head: testBatchID, + Paths: []entity.SpeculationPathEntry{ + {ID: testPathID, Status: entity.SpeculationPathStatusBuilding, Attempt: testAttempt}, + }, + }, nil) + h.pathBuilds.EXPECT().Get(gomock.Any(), testPathID, testAttempt). + Return(entity.PathBuild{}, errors.New("connection reset")) + }, + }, + { + name: "status write failure", + setup: func(h *testHarness) { + h.builds.EXPECT().Get(gomock.Any(), testBuildID).Return(testBuild(entity.BuildStatusRunning), nil) + h.br.EXPECT().Status(gomock.Any(), gomock.Any()).Return(entity.BuildStatusSucceeded, nil, nil) + h.builds.EXPECT().Update(gomock.Any(), gomock.Any()). + Return(errors.New("connection reset")) + }, + }, + { + name: "speculate publish failure", + setup: func(h *testHarness) { + h.wanted() + h.builds.EXPECT().Get(gomock.Any(), testBuildID).Return(testBuild(entity.BuildStatusRunning), nil) + h.br.EXPECT().Status(gomock.Any(), gomock.Any()).Return(entity.BuildStatusRunning, nil, nil) + h.speculatePub.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()). + Return(errors.New("queue down")) + }, + }, + } - h.buildStore.EXPECT().Get(gomock.Any(), build.ID).Return(build, nil) - h.br.EXPECT().Status(gomock.Any(), entity.BuildID{ID: build.ID}).Return(entity.BuildStatusRunning, entity.BuildMetadata{}, nil) - h.batchStore.EXPECT().Get(gomock.Any(), build.BatchID).Return(entity.Batch{ID: build.BatchID, State: state}, nil) - // Halted: no Update, no speculate Publish, no buildsignal - // PublishAfter. The harness publishers have no expectations, so any - // publish fails the test. + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + h := newTestHarness(t, ctrl, entity.BatchStateSpeculating) + tt.setup(h) - require.NoError(t, h.controller.Process(context.Background(), buildDelivery(t, ctrl, build))) + require.Error(t, h.controller.Process(context.Background(), delivery(t, ctrl))) }) } } diff --git a/submitqueue/orchestrator/controller/dlq/buildsignal.go b/submitqueue/orchestrator/controller/dlq/buildsignal.go index b45c2255..51e4a1d5 100644 --- a/submitqueue/orchestrator/controller/dlq/buildsignal.go +++ b/submitqueue/orchestrator/controller/dlq/buildsignal.go @@ -28,13 +28,13 @@ import ( ) // buildSignalController is the DLQ reconciler for the buildsignal topic. Its -// payload carries a BuildID, so reconciliation needs an extra hop: look up -// the Build to recover its BatchID, then fan out via failBatch. +// payload names a build, so reconciliation needs one hop: read the build to +// recover its batch, then fan out via failBatch. // -// The build itself is left in whatever non-terminal state the build runner -// last reported. Fixing the build entity is not useful here — the -// pipeline's source of truth for "did this batch finish" is the batch state, -// and that is what gates the gateway response and conclude. +// The build is left in whatever non-terminal state the runner last reported. +// Fixing it is not useful here — the pipeline's source of truth for "did this +// batch finish" is the batch state, and that is what gates the gateway response +// and conclude. type buildSignalController struct { logger *zap.SugaredLogger metricsScope tally.Scope @@ -73,47 +73,43 @@ func (c *buildSignalController) Process(ctx context.Context, delivery consumer.D msg := delivery.Message() - bid, err := entity.BuildIDFromBytes(msg.Payload) + buildID, err := entity.BuildIDFromBytes(msg.Payload) if err != nil { metrics.NamedCounter(c.metricsScope, opName, "deserialize_errors", 1) return fmt.Errorf("failed to decode build id from dlq payload: %w", err) } - if bid.ID == "" { + if buildID.ID == "" { metrics.NamedCounter(c.metricsScope, opName, "empty_id_errors", 1) return fmt.Errorf("dlq payload decoded to empty build id") } dmeta := delivery.Metadata() c.logger.Warnw("dlq message received", - "build_id", bid.ID, + "build_id", buildID.ID, "attempt", delivery.Attempt(), "dlq_original_topic", dmeta["dlq.original_topic"], "dlq_failure_count", dmeta["dlq.failure_count"], "dlq_last_error", dmeta["dlq.last_error"], ) - build, err := c.store.GetBuildStore().Get(ctx, bid.ID) + build, err := c.store.GetBuildStore().Get(ctx, buildID.ID) if err != nil { if errors.Is(err, storage.ErrNotFound) { - // The build was never persisted (e.g. the build controller crashed - // before Create). There is no batch to reconcile from this signal — - // any associated batch should be reconciled from its own DLQ. - c.logger.Warnw("dlq reconcile: build not found, skipping", - "build_id", bid.ID, - ) + // The build was never recorded (e.g. a crash between triggering and + // writing it down). There is no batch to reconcile from this + // signal; any associated batch reconciles from its own DLQ. + c.logger.Warnw("dlq reconcile: build not found, skipping", "build_id", buildID.ID) metrics.NamedCounter(c.metricsScope, opName, "build_not_found", 1) return nil } metrics.NamedCounter(c.metricsScope, opName, "build_store_errors", 1) - return fmt.Errorf("failed to get build %s: %w", bid.ID, err) + return fmt.Errorf("failed to get build %s: %w", buildID.ID, err) } if build.BatchID == "" { // Defensive: a build without a batch is malformed and there is nothing // to fan out to. Log and ack so the DLQ does not grow unbounded. - c.logger.Errorw("dlq reconcile: build has empty batch id, skipping", - "build_id", bid.ID, - ) + c.logger.Errorw("dlq reconcile: build has empty batch id, skipping", "build_id", buildID.ID) metrics.NamedCounter(c.metricsScope, opName, "build_missing_batch", 1) return nil } diff --git a/submitqueue/orchestrator/controller/dlq/buildsignal_test.go b/submitqueue/orchestrator/controller/dlq/buildsignal_test.go index 10a3eae4..46d7d75c 100644 --- a/submitqueue/orchestrator/controller/dlq/buildsignal_test.go +++ b/submitqueue/orchestrator/controller/dlq/buildsignal_test.go @@ -23,7 +23,6 @@ import ( "github.com/uber/submitqueue/platform/consumer" "github.com/uber/submitqueue/submitqueue/core/topickey" "github.com/uber/submitqueue/submitqueue/entity" - "github.com/uber/submitqueue/submitqueue/extension/storage" storagemock "github.com/uber/submitqueue/submitqueue/extension/storage/mock" "go.uber.org/mock/gomock" "go.uber.org/zap/zaptest" @@ -45,7 +44,8 @@ func TestDLQBuildSignalController_Process_FansOutToBatch(t *testing.T) { buildStore := storagemock.NewMockBuildStore(ctrl) buildStore.EXPECT().Get(gomock.Any(), "build-1").Return(entity.Build{ - ID: "build-1", BatchID: "q/batch/2", Status: entity.BuildStatusRunning, + ID: "build-1", BatchID: "q/batch/2", PathID: "path-1", Attempt: 1, + Status: entity.BuildStatusRunning, }, nil) batchStore := storagemock.NewMockBatchStore(ctrl) @@ -81,44 +81,6 @@ func TestDLQBuildSignalController_Process_FansOutToBatch(t *testing.T) { require.NoError(t, c.Process(context.Background(), delivery)) } -func TestDLQBuildSignalController_Process_BuildNotFoundIsNoOp(t *testing.T) { - ctrl := gomock.NewController(t) - - buildStore := storagemock.NewMockBuildStore(ctrl) - buildStore.EXPECT().Get(gomock.Any(), "build-1").Return(entity.Build{}, storage.ErrNotFound) - - store := storagemock.NewMockStorage(ctrl) - store.EXPECT().GetBuildStore().Return(buildStore).AnyTimes() - - c := NewDLQBuildSignalController(zaptest.NewLogger(t).Sugar(), testScope(), store, consumer.TopicRegistry{}, TopicKey(topickey.TopicKeyBuildSignal), "orchestrator-buildsignal-dlq") - - payload, err := entity.BuildID{ID: "build-1"}.ToBytes() - require.NoError(t, err) - - delivery := newMockDelivery(ctrl, payload) - require.NoError(t, c.Process(context.Background(), delivery)) -} - -func TestDLQBuildSignalController_Process_BuildMissingBatchIsNoOp(t *testing.T) { - ctrl := gomock.NewController(t) - - buildStore := storagemock.NewMockBuildStore(ctrl) - buildStore.EXPECT().Get(gomock.Any(), "build-1").Return(entity.Build{ - ID: "build-1", BatchID: "", Status: entity.BuildStatusRunning, - }, nil) - - store := storagemock.NewMockStorage(ctrl) - store.EXPECT().GetBuildStore().Return(buildStore).AnyTimes() - - c := NewDLQBuildSignalController(zaptest.NewLogger(t).Sugar(), testScope(), store, consumer.TopicRegistry{}, TopicKey(topickey.TopicKeyBuildSignal), "orchestrator-buildsignal-dlq") - - payload, err := entity.BuildID{ID: "build-1"}.ToBytes() - require.NoError(t, err) - - delivery := newMockDelivery(ctrl, payload) - require.NoError(t, c.Process(context.Background(), delivery)) -} - func TestDLQBuildSignalController_Process_MalformedPayloadFails(t *testing.T) { ctrl := gomock.NewController(t)