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)