From 2334a1905caf902335b518b5ece7771204e5d9be Mon Sep 17 00:00:00 2001 From: Preetam Dwivedi Date: Fri, 10 Jul 2026 08:23:31 -0700 Subject: [PATCH] feat(speculation): route builds through the tree via prioritize MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary ### Why? The speculate controller previously drove a batch's own build directly (publish to build, CAS to Speculating) while the speculation tree it maintained was pure shadow-mode bookkeeping nobody read. To make the tree the actual source of truth for what gets built, the pipeline needs to flip: speculate hands off to prioritize, and prioritize's admission decisions (Prioritized paths) are what the build stage acts on. This also completes the ownership rule this stack is building toward: the build stage becomes the sole owner of runner interaction for path-level work. Other stages (prioritize's preemptive eviction, speculate's reconcile for dead paths, both landing in a later commit) only ever record a cancel decision as intent (SpeculationPathStatusCancelling) on the tree; nothing but the build stage calls BuildRunner.Cancel. ### What? speculate.go: speculateBatch no longer publishes to build. It CASes Created/Scored batches to Speculating, then publishes the queue to prioritize on every pass (after the CAS, so a concurrent prioritize round never observes a batch as not-yet-Speculating with no later wake-up). The legacy tryFinalize step is otherwise unchanged and still only runs once a batch has reached Speculating on this or a prior pass, so merge timing is unchanged for now — tightening it to gate on the tree's own path outcome is deferred to a later commit in this stack. build.go is replaced with its tree-driven form: it loads the batch's speculation tree and, per path, either triggers a build for a Prioritized path with no build yet, or enacts a persisted Cancelling intent. Trigger dedup is on the path->build mapping (SpeculationPathBuildStore), checked before Trigger runs since it is readable up front unlike a Build row. The write order is Trigger -> Build.Create -> mapping.Create -> publish to buildsignal; a crash between Trigger and the mapping write is recovered by redelivery re-running the same path, and a lost mapping-create race defers to the winner (republishing buildsignal for its build) rather than erroring. A Prioritized path whose mapping already resolves to a non-terminal build republishes buildsignal to close the crash window between the original Create and its original publish; a terminal build is a no-op, since reconcile is what will fold that outcome into the path's status. The new Cancelling arm resolves the path's build via the same mapping, calls runner.Cancel if the build is not already terminal, and republishes buildsignal so the poll loop reacts promptly. Every lookup miss (no mapping yet, or a mapping pointing at a missing build row) is tolerated as a no-op rather than an error, leaving the intent for a later reconcile pass to settle. A missing speculation tree, by contrast, is an invariant violation — the tree is created before the batch is ever published to prioritize and is never deleted — so it surfaces as a plain error and dead-letters, failing the batch loudly instead of retrying against broken state. buildsignal.go picks up its RFC-final wording (build's ID doubling as the runner handle) plus a corrected error-classification doc paragraph — classification is per error cause via the wired classifier walk, not the per-call-site policy the old text described — with no behavior change; the test file is byte-identical to before. ## Test Plan ✅ `go build ./submitqueue/... ./service/...` ✅ `go vet ./submitqueue/... ./service/...` ✅ `bazel test //submitqueue/... //service/...` — 56/56 targets pass, including new build.go tests for the Cancelling arm (non-terminal enacts + republishes, terminal no-ops, no-mapping/dangling-mapping skip, runner error surfaces with no silent ack). ✅ `make gazelle` / `make fmt` — no diffs beyond the intended source changes. ✅ `make e2e-test` — 2/2 targets pass (stovepipe, submitqueue), confirming parity: prioritize still admits the single chain path and exactly one build runs per batch. --- .../orchestrator/controller/build/build.go | 361 +++++++- .../controller/build/build_test.go | 851 ++++++++++++++---- .../controller/buildsignal/buildsignal.go | 41 +- .../controller/speculate/speculate.go | 127 ++- .../controller/speculate/speculate_test.go | 84 +- 5 files changed, 1127 insertions(+), 337 deletions(-) diff --git a/submitqueue/orchestrator/controller/build/build.go b/submitqueue/orchestrator/controller/build/build.go index 8745f772..5848c173 100644 --- a/submitqueue/orchestrator/controller/build/build.go +++ b/submitqueue/orchestrator/controller/build/build.go @@ -18,6 +18,7 @@ import ( "context" "errors" "fmt" + "time" "github.com/uber-go/tally" entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" @@ -31,7 +32,27 @@ import ( ) // Controller handles build queue messages. -// It consumes batches, triggers builds, and publishes scheduled builds to the build signal stage (which processes build results). +// +// It loads the batch's speculation tree and, for each path, acts on the +// path's status: a Prioritized path that has no build yet gets one triggered +// through the queue's build runner, and a Cancelling path has its in-flight +// build cancelled. Every other status is left untouched, and the tree is +// never written here — path statuses are read-only inputs to this +// controller. It is the sole caller of the build runner for path-level work, +// which is why persisted Cancelling intents are enacted here rather than +// where they were decided. +// +// Halted (terminal or Cancelling) batches are skipped before any path work: +// batch-level cancellation has its own owner, and the skip guarantees this +// controller never starts CI for a batch that is being torn down. +// +// Dedup for triggering is on the path->build mapping (PathBuildStore), not +// on a Build row keyed by a derived key: the mapping is readable before +// Trigger ever runs, so it is checked first. A crash between Trigger +// succeeding and the mapping Create means redelivery re-triggers a fresh +// build for the same path; the new mapping Create then races the +// (never-persisted) old one and simply wins, orphaning the earlier CI +// build. See the per-path loop below for the full crash-safety ordering. // Implements consumer.Controller interface for integration with the consumer. type Controller struct { logger *zap.SugaredLogger @@ -46,6 +67,9 @@ type Controller struct { // Verify Controller implements consumer.Controller interface at compile time. var _ consumer.Controller = (*Controller)(nil) +// opName is the metric operation name shared by every emit in this file. +const opName = "process" + // NewController creates a new build controller for the orchestrator. func NewController( logger *zap.SugaredLogger, @@ -68,11 +92,12 @@ func NewController( } // Process processes a build delivery from the queue. -// Deserializes the batch, triggers a build, and publishes a build entity to the build signal topic. -// Returns nil to ack (success), or error to nack (retry). +// Deserializes the batch, loads its speculation tree, and for every path +// either triggers a build (Prioritized, no build yet) or enacts a persisted +// cancel intent (Cancelling), publishing to the build signal topic as +// appropriate. +// Returns nil to ack (success), or error to nack/reject. func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) { - const opName = "process" - op := metrics.Begin(c.metricsScope, opName) defer func() { op.Complete(retErr) }() @@ -115,58 +140,316 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r return nil } - // Load the dependency batches (base) as identity; the build runner resolves - // each batch's changes itself. head is this batch. - base, err := c.loadBatches(ctx, batch.Dependencies) + // Load this batch's speculation tree. A build message is only ever + // published by the prioritize stage after it has read the tree, the tree + // is created before the batch is ever published to prioritize, and trees + // are never deleted — so by the time this message is readable the tree + // exists. A Get miss here is an invariant violation (corrupted or + // manually mutated state), not a transient condition: the error + // propagates unclassified, the message dead-letters, and the DLQ + // consumer fails the batch loudly instead of retrying against broken + // state. + tree, err := c.store.GetSpeculationTreeStore().Get(ctx, batch.ID) if err != nil { metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) - return fmt.Errorf("failed to load dependency batches for batch %s: %w", batch.ID, err) + return fmt.Errorf("failed to get speculation tree for batch %s: %w", batch.ID, err) + } + + // Resolve the build runner lazily: every path in this message shares the + // same queue, so a pass whose paths need no triggering or cancelling + // never resolves one. + resolveRunner := c.runnerResolver(batch.Queue) + + triggered := 0 + for _, p := range tree.Paths { + if p.Status == entity.SpeculationPathStatusCancelling { + // A Cancelling path is a persisted cancel intent — recorded + // elsewhere, enacted here by the sole caller of the build runner + // for path-level work. + if err := c.enactCancel(ctx, batch, p, resolveRunner); err != nil { + return err + } + continue + } + + if p.Status != entity.SpeculationPathStatusPrioritized { + // Every other status is another stage's concern: Candidate and + // Selected haven't been admitted yet, Building/Passed/Failed + // already have a build in flight or resolved, and Cancelled is + // terminal. This controller moves admitted (Prioritized) paths + // forward and enacts persisted Cancelling intents (handled + // above) — nothing else. + continue + } + + // Per-path ordering matters for crash-safety. Dedup happens BEFORE + // Trigger, keyed on the path->build mapping (readable up front, + // unlike a Build row that only exists after Trigger succeeds). Once + // a path needs triggering, the order is: (1) Trigger with the + // runner, (2) persist the Build row, (3) persist the path->build + // mapping. A crash between (1) and (3) is recovered by redelivery + // re-running this loop: the mapping is still absent, so the path is + // (re-)triggered and the fresh mapping Create either wins outright + // or loses a race to a concurrent delivery that got there first — in + // which case we defer to the winner instead of erroring (see below). + pb, err := c.store.GetSpeculationPathBuildStore().Get(ctx, p.ID) + if err == nil { + // A mapping already exists for this path: a build was already + // triggered, either by a previous pass or a previous delivery of + // this message. Load it and decide what (if anything) to do. + b, err := c.store.GetBuildStore().Get(ctx, pb.BuildID) + if err != nil { + if errors.Is(err, storage.ErrNotFound) { + // Invariant breach: the write order below guarantees a + // mapping is only created once its build row exists. The + // mapping is still authoritative even though its target + // is missing, so we do not trigger a duplicate build for + // this path — we just skip it defensively. + metrics.NamedCounter(c.metricsScope, opName, "mapping_dangling", 1) + c.logger.Warnw("path->build mapping points at a missing build; skipping path", + "batch_id", batch.ID, + "path_id", p.ID, + "build_id", pb.BuildID, + ) + continue + } + metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) + return fmt.Errorf("failed to get build %s for path %s of batch %s: %w", pb.BuildID, p.ID, batch.ID, err) + } + + if b.Status.IsTerminal() { + // The path's build already ran to completion; a later + // reconcile pass folds that outcome into the path's status, + // so triggering here would only race it. A Prioritized path + // pointing at a terminal build is therefore always the + // pre-reconcile window — never + // "build it again": in the current model a path is built at + // most once, Cancel is only ever applied to paths whose + // assumption is dead (or preemptively evicted), and no stage + // re-admits a Cancelled path. If a future preemptive policy + // adds cancel-then-readmit, this branch is the seam it lands + // in: re-trigger and conditionally re-point the mapping to + // the fresh build under SpeculationPathBuild.Version. + continue + } + + // Non-terminal existing build: republish buildsignal to close the + // crash window between the original Create and the original + // buildsignal publish. This creates duplicate poll loops on + // redelivery, but that is harmless — buildsignal's self-republish + // reuses the message ID (build.ID), and the mysql queue publisher + // dedups publishes on (topic, partition, id), so redundant + // re-triggers of the poll loop coalesce rather than double-poll + // forever. + if err := c.publish(ctx, topickey.TopicKeyBuildSignal, b); err != nil { + metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1) + return fmt.Errorf("failed to re-publish to buildsignal: %w", err) + } + continue + } else if !errors.Is(err, storage.ErrNotFound) { + metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) + return fmt.Errorf("failed to get path->build mapping for path %s of batch %s: %w", p.ID, batch.ID, err) + } + + // No mapping yet: load the path's base (assumed-good predecessor + // batches) as identity; the build runner resolves each batch's + // changes itself. head is this batch. + base, err := c.loadBatches(ctx, p.Path.Base) + if err != nil { + metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) + return fmt.Errorf("failed to load base batches for path (head=%s, base=%v) of batch %s: %w", p.Path.Head, p.Path.Base, batch.ID, err) + } + + // Trigger the build with the queue's build runner. metadata is nil + // until a caller-supplied source materializes (e.g. requester / + // ticket pulled off the originating LandRequest). + runner, err := resolveRunner() + if err != nil { + metrics.NamedCounter(c.metricsScope, opName, "trigger_errors", 1) + return fmt.Errorf("failed to get build runner for batch %s: %w", batch.ID, err) + } + runnerID, err := runner.Trigger(ctx, base, batch, nil) + if err != nil { + metrics.NamedCounter(c.metricsScope, opName, "trigger_errors", 1) + return fmt.Errorf("failed to trigger build for batch %s: %w", batch.ID, err) + } + + build := entity.Build{ + ID: runnerID.ID, + BatchID: batch.ID, + SpeculationPathID: p.ID, + Status: entity.BuildStatusAccepted, + } + + // Persist the initial Build snapshot so the buildsignal poll loop has + // a row to UpdateStatus against. ErrAlreadyExists is benign — a + // redelivery of this message after a previous successful Create for + // this specific path. This tolerance is not the dedup mechanism (the + // mapping below is); it stays for the same crash-safety reason as + // before: a build row can pre-exist from a prior partial pass. + if err := c.store.GetBuildStore().Create(ctx, build); err != nil && !errors.Is(err, storage.ErrAlreadyExists) { + metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) + return fmt.Errorf("failed to persist build %s: %w", build.ID, err) + } + + // Persist the path->build mapping. This is the dedup marker for the + // path, checked at the top of this loop before Trigger runs. + mapping := entity.SpeculationPathBuild{ + PathID: p.ID, + BuildID: build.ID, + BatchID: batch.ID, + Version: 1, + CreatedAt: time.Now().UnixMilli(), + } + if err := c.store.GetSpeculationPathBuildStore().Create(ctx, mapping); err != nil { + if errors.Is(err, storage.ErrAlreadyExists) { + // A concurrent delivery already won the mapping race for this + // path. Our just-created build row above is now an accepted + // orphan: lost mapping races cost one orphaned CI build by + // design — duplicates are the retry/redundancy mechanism, and + // the recorded mapping (not the build row) is the source of + // truth for "which build resolves this path." Republish + // buildsignal for the winner so it has an active poll loop + // even if its own publish was lost. + metrics.NamedCounter(c.metricsScope, opName, "trigger_race_lost", 1) + winner, gerr := c.store.GetSpeculationPathBuildStore().Get(ctx, p.ID) + if gerr != nil { + metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) + return fmt.Errorf("failed to re-get path->build mapping for path %s of batch %s: %w", p.ID, batch.ID, gerr) + } + winnerBuild, gerr := c.store.GetBuildStore().Get(ctx, winner.BuildID) + if gerr != nil { + metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) + return fmt.Errorf("failed to get winning build %s for path %s of batch %s: %w", winner.BuildID, p.ID, batch.ID, gerr) + } + if err := c.publish(ctx, topickey.TopicKeyBuildSignal, winnerBuild); err != nil { + metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1) + return fmt.Errorf("failed to re-publish to buildsignal: %w", err) + } + continue + } + metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) + return fmt.Errorf("failed to persist path->build mapping for path %s of batch %s: %w", p.ID, batch.ID, err) + } + + // Hand off to the buildsignal poll loop; it calls Status, updates the + // persisted Build, publishes to speculate, and re-publishes itself + // via PublishAfter until terminal. + if err := c.publish(ctx, topickey.TopicKeyBuildSignal, build); err != nil { + metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1) + return fmt.Errorf("failed to publish to buildsignal: %w", err) + } + + c.logger.Infow("published build to buildsignal", + "batch_id", batch.ID, + "build_id", build.ID, + "status", string(build.Status), + "topic_key", topickey.TopicKeyBuildSignal, + ) + + triggered++ + } + + if triggered == 0 { + // Either no path was Prioritized, or every Prioritized path already + // had a matching Build row. This is a no-op republish, not an error. + metrics.NamedCounter(c.metricsScope, opName, "no_paths_to_trigger", 1) + } + + return nil // Success - message will be acked +} + +// runnerResolver returns a lazily-memoizing accessor for queue's build +// runner: the factory is consulted on the first call only, and every +// subsequent call reuses that result. Callers that never invoke the accessor +// never resolve a runner at all. +func (c *Controller) runnerResolver(queue string) func() (buildrunner.BuildRunner, error) { + var runner buildrunner.BuildRunner + return func() (buildrunner.BuildRunner, error) { + if runner != nil { + return runner, nil + } + r, err := c.buildRunners.For(buildrunner.Config{QueueName: queue}) + if err != nil { + return nil, err + } + runner = r + return r, nil } +} - // Trigger the build with the queue's build runner. metadata is nil - // until a caller-supplied source materializes (e.g. requester / ticket - // pulled off the originating LandRequest). - buildRunner, err := c.buildRunners.For(buildrunner.Config{QueueName: batch.Queue}) +// enactCancel enacts a persisted Cancelling intent for path p: it resolves +// the path's build via the path->build mapping, issues a runner Cancel +// against it if the build is not already terminal, and republishes +// buildsignal so the poll loop observes the cancellation promptly instead of +// waiting out its current delay. Cancel is idempotent from the runner's +// point of view, so redelivery (e.g. after a crash between Cancel and the +// republish) simply re-issues it. +// +// Every lookup miss here is tolerated rather than treated as an error: a +// path can be marked Cancelling before this controller ever triggered a +// build for it (no mapping yet), or the mapping can point at a build row +// that a defensive skip elsewhere left dangling. In both cases there is +// nothing to cancel, so the intent is left for speculate's reconcile to +// settle the path to Cancelled on its next pass. +func (c *Controller) enactCancel(ctx context.Context, batch entity.Batch, p entity.SpeculationPathInfo, resolveRunner func() (buildrunner.BuildRunner, error)) error { + pb, err := c.store.GetSpeculationPathBuildStore().Get(ctx, p.ID) if err != nil { - metrics.NamedCounter(c.metricsScope, opName, "trigger_errors", 1) - return fmt.Errorf("failed to build runner for batch %s: %w", batch.ID, err) + if errors.Is(err, storage.ErrNotFound) { + metrics.NamedCounter(c.metricsScope, opName, "cancel_no_mapping", 1) + return nil + } + metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) + return fmt.Errorf("failed to get path->build mapping for path %s of batch %s: %w", p.ID, batch.ID, err) } - buildID, err := buildRunner.Trigger(ctx, base, batch, nil) + + b, err := c.store.GetBuildStore().Get(ctx, pb.BuildID) if err != nil { - metrics.NamedCounter(c.metricsScope, opName, "trigger_errors", 1) - return fmt.Errorf("failed to trigger build for batch %s: %w", batch.ID, err) + if errors.Is(err, storage.ErrNotFound) { + // Same invariant-breach defense as the trigger path above: the + // mapping is authoritative even though its target is missing, so + // skip rather than guessing. + metrics.NamedCounter(c.metricsScope, opName, "mapping_dangling", 1) + c.logger.Warnw("path->build mapping points at a missing build; skipping cancel", + "batch_id", batch.ID, + "path_id", p.ID, + "build_id", pb.BuildID, + ) + return nil + } + metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) + return fmt.Errorf("failed to get build %s for path %s of batch %s: %w", pb.BuildID, p.ID, batch.ID, err) } - build := entity.Build{ - ID: buildID.ID, - BatchID: batch.ID, - Status: entity.BuildStatusAccepted, + if b.Status.IsTerminal() { + // Nothing in flight to cancel; speculate's reconcile settles the + // path's own status on its next pass. + metrics.NamedCounter(c.metricsScope, opName, "cancel_already_terminal", 1) + return nil } - // Persist the initial Build snapshot so the buildsignal poll loop has a - // row to UpdateStatus against. ErrAlreadyExists is benign — a redelivery - // of this message after a previous successful Create. - if err := c.store.GetBuildStore().Create(ctx, build); err != nil && !errors.Is(err, storage.ErrAlreadyExists) { - metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) - return fmt.Errorf("failed to persist build %s: %w", build.ID, err) + runner, err := resolveRunner() + if err != nil { + metrics.NamedCounter(c.metricsScope, opName, "cancel_errors", 1) + return fmt.Errorf("failed to get build runner for batch %s: %w", batch.ID, err) + } + if err := runner.Cancel(ctx, entity.BuildID{ID: b.ID}); err != nil { + metrics.NamedCounter(c.metricsScope, opName, "cancel_errors", 1) + return fmt.Errorf("failed to cancel build %s for path %s of batch %s: %w", b.ID, p.ID, batch.ID, err) } - // Hand off to the buildsignal poll loop; it calls Status, updates the - // persisted Build, publishes to speculate, and re-publishes itself via - // PublishAfter until terminal. - if err := c.publish(ctx, topickey.TopicKeyBuildSignal, build); err != nil { + if err := c.publish(ctx, topickey.TopicKeyBuildSignal, b); err != nil { metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1) - return fmt.Errorf("failed to publish to buildsignal: %w", err) + return fmt.Errorf("failed to re-publish to buildsignal after cancel: %w", err) } - c.logger.Infow("published build to buildsignal", + metrics.NamedCounter(c.metricsScope, opName, "cancel_enacted", 1) + c.logger.Infow("enacted cancel for speculation path", "batch_id", batch.ID, - "build_id", build.ID, - "status", string(build.Status), - "topic_key", topickey.TopicKeyBuildSignal, + "path_id", p.ID, + "build_id", b.ID, ) - - return nil // Success - message will be acked + return nil } // loadBatches loads each batch by ID, preserving order. Used to load the base diff --git a/submitqueue/orchestrator/controller/build/build_test.go b/submitqueue/orchestrator/controller/build/build_test.go index 7347ae29..224f1425 100644 --- a/submitqueue/orchestrator/controller/build/build_test.go +++ b/submitqueue/orchestrator/controller/build/build_test.go @@ -55,46 +55,55 @@ func testBatch() entity.Batch { } } -// newMockStorage creates a MockStorage with a MockBatchStore that returns the -// given batch on Get, a no-op MockRequestStore, and a MockBuildStore that -// accepts any Create call. Tests that care about Create arguments build their -// own MockBuildStore. -func newMockStorage(ctrl *gomock.Controller, batch entity.Batch) *storagemock.MockStorage { - mockBatchStore := storagemock.NewMockBatchStore(ctrl) - mockBatchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil).AnyTimes() - - mockRequestStore := storagemock.NewMockRequestStore(ctrl) - - mockBuildStore := storagemock.NewMockBuildStore(ctrl) - mockBuildStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil).AnyTimes() +// newMockStorage wires a MockStorage backed by mock batch/speculation-tree/ +// build/path-build sub-stores. Callers set whatever EXPECT() calls each test +// needs on the returned sub-stores. +func newMockStorage(ctrl *gomock.Controller) (*storagemock.MockStorage, *storagemock.MockBatchStore, *storagemock.MockSpeculationTreeStore, *storagemock.MockBuildStore, *storagemock.MockSpeculationPathBuildStore) { + batchStore := storagemock.NewMockBatchStore(ctrl) + treeStore := storagemock.NewMockSpeculationTreeStore(ctrl) + buildStore := storagemock.NewMockBuildStore(ctrl) + pathBuildStore := storagemock.NewMockSpeculationPathBuildStore(ctrl) store := storagemock.NewMockStorage(ctrl) - store.EXPECT().GetBatchStore().Return(mockBatchStore).AnyTimes() - store.EXPECT().GetRequestStore().Return(mockRequestStore).AnyTimes() - store.EXPECT().GetBuildStore().Return(mockBuildStore).AnyTimes() - return store + store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() + store.EXPECT().GetSpeculationTreeStore().Return(treeStore).AnyTimes() + store.EXPECT().GetBuildStore().Return(buildStore).AnyTimes() + store.EXPECT().GetSpeculationPathBuildStore().Return(pathBuildStore).AnyTimes() + return store, batchStore, treeStore, buildStore, pathBuildStore } -// newTestController creates a controller with test dependencies. br is the -// build runner to inject; pass buildfake.New(changesetfake.New()) for the pass-through default. // staticBuildRunnerFactory is a test factory that returns a fixed BuildRunner -// for any entityqueue. +// for any queue config. type staticBuildRunnerFactory struct{ r buildrunner.BuildRunner } func (f staticBuildRunnerFactory) For(buildrunner.Config) (buildrunner.BuildRunner, error) { return f.r, nil } +// newTestController creates a controller with test dependencies. br is the +// build runner to inject; pass buildfake.New(changesetfake.New()) for the +// pass-through default. publishErr, if non-nil, is returned by every publish +// call. published, if non-nil, accumulates the BuildID of every message +// published to buildsignal, in call order. +// // The wired registry exposes only the buildsignal topic — that is what the -// controller publishes to after the RFC refactor. -func newTestController(t *testing.T, ctrl *gomock.Controller, store *storagemock.MockStorage, br buildrunner.BuildRunner, publishErr error) *Controller { +// controller publishes to. +func newTestController(t *testing.T, ctrl *gomock.Controller, store *storagemock.MockStorage, br buildrunner.BuildRunner, publishErr error, published *[]string) *Controller { logger := zaptest.NewLogger(t).Sugar() scope := tally.NoopScope mockPub := queuemock.NewMockPublisher(ctrl) mockPub.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn( - func(ctx context.Context, topic string, msg entityqueue.Message) error { - return publishErr + func(_ context.Context, _ string, msg entityqueue.Message) error { + if publishErr != nil { + return publishErr + } + if published != nil { + bid, err := entity.BuildIDFromBytes(msg.Payload) + require.NoError(t, err) + *published = append(*published, bid.ID) + } + return nil }, ).AnyTimes() @@ -109,11 +118,19 @@ func newTestController(t *testing.T, ctrl *gomock.Controller, store *storagemock return NewController(logger, scope, store, staticBuildRunnerFactory{r: br}, registry, topickey.TopicKeyBuild, "orchestrator-build") } +// runProcess builds a delivery for batch and invokes Process once. +func runProcess(t *testing.T, ctrl *gomock.Controller, controller *Controller, batch entity.Batch) error { + msg := entityqueue.NewMessage(batch.ID, batchIDPayload(t, batch.ID), batch.Queue, nil) + delivery := queuemock.NewMockDelivery(ctrl) + delivery.EXPECT().Message().Return(msg).AnyTimes() + delivery.EXPECT().Attempt().Return(1).AnyTimes() + return controller.Process(context.Background(), delivery) +} + func TestNewController(t *testing.T) { ctrl := gomock.NewController(t) - batch := testBatch() - store := newMockStorage(ctrl, batch) - controller := newTestController(t, ctrl, store, buildfake.New(changesetfake.New()), nil) + store, _, _, _, _ := newMockStorage(ctrl) + controller := newTestController(t, ctrl, store, buildfake.New(changesetfake.New()), nil, nil) require.NotNil(t, controller) assert.Equal(t, topickey.TopicKeyBuild, controller.TopicKey()) @@ -121,236 +138,309 @@ func TestNewController(t *testing.T) { assert.Equal(t, "build", controller.Name()) } -func TestController_Process_Success(t *testing.T) { +func TestController_InterfaceImplementation(t *testing.T) { + ctrl := gomock.NewController(t) + store, _, _, _, _ := newMockStorage(ctrl) + controller := newTestController(t, ctrl, store, buildfake.New(changesetfake.New()), nil, nil) + + var _ consumer.Controller = controller +} + +// TestController_Process_NoPrioritizedPaths covers an empty tree (or a tree +// with no Prioritized paths): the controller does no work and acks. +func TestController_Process_NoPrioritizedPaths(t *testing.T) { ctrl := gomock.NewController(t) batch := testBatch() - store := newMockStorage(ctrl, batch) - controller := newTestController(t, ctrl, store, buildfake.New(changesetfake.New()), nil) + store, batchStore, treeStore, _, _ := newMockStorage(ctrl) + batchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil).AnyTimes() + treeStore.EXPECT().Get(gomock.Any(), batch.ID).Return(entity.SpeculationTree{BatchID: batch.ID, Version: 1}, nil) + // No PathBuildStore.Get expectation: an empty tree has no Prioritized path + // to dedup-check. - msg := entityqueue.NewMessage(batch.ID, batchIDPayload(t, batch.ID), batch.Queue, nil) - delivery := queuemock.NewMockDelivery(ctrl) - delivery.EXPECT().Message().Return(msg).AnyTimes() - delivery.EXPECT().Attempt().Return(1).AnyTimes() + br := buildrunnermock.NewMockBuildRunner(ctrl) + // No Trigger expectation: a stray CI trigger with no Prioritized path + // fails the test. + controller := newTestController(t, ctrl, store, br, nil, nil) - err := controller.Process(context.Background(), delivery) - require.NoError(t, err) + require.NoError(t, runProcess(t, ctrl, controller, batch)) } -// TestController_Process_TriggersWithBaseAndHead verifies the controller hands -// BuildRunner.Trigger the base (dependency batches in order) and head (this -// batch) as identity, persists the initial Accepted Build, and publishes it to -// the buildsignal topic. The runner resolves each batch's changes itself. -func TestController_Process_TriggersWithBaseAndHead(t *testing.T) { +// TestController_Process_MultiplePrioritizedPathsTriggerAll covers two +// Prioritized paths with no existing mapping: both must trigger, each with +// its own path's base (in order), and each must persist a Build row and a +// path->build mapping, then publish. +func TestController_Process_MultiplePrioritizedPathsTriggerAll(t *testing.T) { ctrl := gomock.NewController(t) - depBatch := entity.Batch{ - ID: "test-queue/batch/dep", - Queue: "test-queue", - Contains: []string{"test-queue/dep-1"}, - } - headBatch := entity.Batch{ - ID: "test-queue/batch/head", - Queue: "test-queue", - State: entity.BatchStateSpeculating, - Version: 1, - Dependencies: []string{depBatch.ID}, - Contains: []string{"test-queue/head-1", "test-queue/head-2"}, + batch := testBatch() + depA := entity.Batch{ID: "test-queue/batch/dep-a", Queue: "test-queue"} + depB := entity.Batch{ID: "test-queue/batch/dep-b", Queue: "test-queue"} + + pathDirect := entity.SpeculationPath{Base: nil, Head: batch.ID} + pathOnDeps := entity.SpeculationPath{Base: []string{depB.ID, depA.ID}, Head: batch.ID} + pathDirectID := "test-queue/batch/1/path/0" + pathOnDepsID := "test-queue/batch/1/path/1" + + tree := entity.SpeculationTree{ + BatchID: batch.ID, + Version: 1, + Paths: []entity.SpeculationPathInfo{ + {ID: pathDirectID, Path: pathDirect, Status: entity.SpeculationPathStatusPrioritized}, + {ID: pathOnDepsID, Path: pathOnDeps, Status: entity.SpeculationPathStatusPrioritized}, + }, } - mockBatchStore := storagemock.NewMockBatchStore(ctrl) - mockBatchStore.EXPECT().Get(gomock.Any(), headBatch.ID).Return(headBatch, nil).AnyTimes() - mockBatchStore.EXPECT().Get(gomock.Any(), depBatch.ID).Return(depBatch, nil).AnyTimes() + store, batchStore, treeStore, buildStore, pathBuildStore := newMockStorage(ctrl) + batchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil).AnyTimes() + batchStore.EXPECT().Get(gomock.Any(), depA.ID).Return(depA, nil).AnyTimes() + batchStore.EXPECT().Get(gomock.Any(), depB.ID).Return(depB, nil).AnyTimes() + treeStore.EXPECT().Get(gomock.Any(), batch.ID).Return(tree, nil) + pathBuildStore.EXPECT().Get(gomock.Any(), pathDirectID).Return(entity.SpeculationPathBuild{}, storage.ErrNotFound) + pathBuildStore.EXPECT().Get(gomock.Any(), pathOnDepsID).Return(entity.SpeculationPathBuild{}, storage.ErrNotFound) - var created entity.Build - mockBuildStore := storagemock.NewMockBuildStore(ctrl) - mockBuildStore.EXPECT().Create(gomock.Any(), gomock.Any()).DoAndReturn( + var createdBuilds []entity.Build + buildStore.EXPECT().Create(gomock.Any(), gomock.Any()).DoAndReturn( func(_ context.Context, b entity.Build) error { - created = b + createdBuilds = append(createdBuilds, b) return nil }, - ).Times(1) - - store := storagemock.NewMockStorage(ctrl) - store.EXPECT().GetBatchStore().Return(mockBatchStore).AnyTimes() - store.EXPECT().GetBuildStore().Return(mockBuildStore).AnyTimes() + ).Times(2) - br := buildrunnermock.NewMockBuildRunner(ctrl) - // base is the dependency batches (identity); head is this batch. - br.EXPECT().Trigger(gomock.Any(), []entity.Batch{depBatch}, headBatch, gomock.Nil()).Return(entity.BuildID{ID: "build-xyz"}, nil) - - var publishedTopic string - var published entity.BuildID - mockPub := queuemock.NewMockPublisher(ctrl) - mockPub.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn( - func(_ context.Context, topic string, msg entityqueue.Message) error { - publishedTopic = topic - bid, err := entity.BuildIDFromBytes(msg.Payload) - require.NoError(t, err) - published = bid + var createdMappings []entity.SpeculationPathBuild + pathBuildStore.EXPECT().Create(gomock.Any(), gomock.Any()).DoAndReturn( + func(_ context.Context, pb entity.SpeculationPathBuild) error { + createdMappings = append(createdMappings, pb) return nil }, - ) - mockQ := queuemock.NewMockQueue(ctrl) - mockQ.EXPECT().Publisher().Return(mockPub).AnyTimes() - registry, err := consumer.NewTopicRegistry( - []consumer.TopicConfig{{Key: topickey.TopicKeyBuildSignal, Name: "buildsignal", Queue: mockQ}}, - ) - require.NoError(t, err) + ).Times(2) + + br := buildrunnermock.NewMockBuildRunner(ctrl) + // Each path's base is passed to Trigger in the path's own order. + br.EXPECT().Trigger(gomock.Any(), []entity.Batch(nil), batch, gomock.Nil()).Return(entity.BuildID{ID: "build-direct"}, nil) + br.EXPECT().Trigger(gomock.Any(), []entity.Batch{depB, depA}, batch, gomock.Nil()).Return(entity.BuildID{ID: "build-on-deps"}, nil) - controller := NewController(zaptest.NewLogger(t).Sugar(), tally.NoopScope, store, staticBuildRunnerFactory{r: br}, registry, topickey.TopicKeyBuild, "orchestrator-build") + var published []string + controller := newTestController(t, ctrl, store, br, nil, &published) - msg := entityqueue.NewMessage(headBatch.ID, batchIDPayload(t, headBatch.ID), headBatch.Queue, nil) - delivery := queuemock.NewMockDelivery(ctrl) - delivery.EXPECT().Message().Return(msg).AnyTimes() - delivery.EXPECT().Attempt().Return(1).AnyTimes() + require.NoError(t, runProcess(t, ctrl, controller, batch)) - require.NoError(t, controller.Process(context.Background(), delivery)) + require.Len(t, createdBuilds, 2) + gotPathIDs := map[string]string{} + for _, b := range createdBuilds { + gotPathIDs[b.ID] = b.SpeculationPathID + } + assert.Equal(t, pathDirectID, gotPathIDs["build-direct"]) + assert.Equal(t, pathOnDepsID, gotPathIDs["build-on-deps"]) - // Only the build ID is published to buildsignal. - assert.Equal(t, "buildsignal", publishedTopic) - assert.Equal(t, "build-xyz", published.ID) + require.Len(t, createdMappings, 2) + gotMappingBuildIDs := map[string]string{} + for _, pb := range createdMappings { + gotMappingBuildIDs[pb.PathID] = pb.BuildID + } + assert.Equal(t, "build-direct", gotMappingBuildIDs[pathDirectID]) + assert.Equal(t, "build-on-deps", gotMappingBuildIDs[pathOnDepsID]) - // The full Build is persisted to storage (the source of truth the poll - // loop reloads), and its ID matches what was published. - assert.Equal(t, "build-xyz", created.ID) - assert.Equal(t, headBatch.ID, created.BatchID) - assert.Equal(t, entity.BuildStatusAccepted, created.Status) - assert.Equal(t, published.ID, created.ID) + assert.ElementsMatch(t, []string{"build-direct", "build-on-deps"}, published) } -// TestController_Process_BuildStoreAlreadyExistsIsSwallowed covers the -// redelivery case: Create returns ErrAlreadyExists, the controller proceeds -// to publish to buildsignal anyway. The polling loop will pick up the -// existing row via UpdateStatus. -func TestController_Process_BuildStoreAlreadyExistsIsSwallowed(t *testing.T) { +// TestController_Process_ExistingMappingNonTerminalBuildRepublishes covers +// the case where a Prioritized path already has a path->build mapping whose +// build is still non-terminal (redelivery, or a prior partial pass): it must +// not trigger or persist again, but it must republish buildsignal for the +// existing build so a lost original publish is healed. +func TestController_Process_ExistingMappingNonTerminalBuildRepublishes(t *testing.T) { ctrl := gomock.NewController(t) batch := testBatch() + path := entity.SpeculationPath{Base: nil, Head: batch.ID} + pathID := "test-queue/batch/1/path/0" + tree := entity.SpeculationTree{ + BatchID: batch.ID, + Version: 1, + Paths: []entity.SpeculationPathInfo{ + {ID: pathID, Path: path, Status: entity.SpeculationPathStatusPrioritized, BuildID: "runner-existing"}, + }, + } + existingBuild := entity.Build{ID: "runner-existing", BatchID: batch.ID, SpeculationPathID: pathID, Status: entity.BuildStatusAccepted} - mockBatchStore := storagemock.NewMockBatchStore(ctrl) - mockBatchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil).AnyTimes() - mockBuildStore := storagemock.NewMockBuildStore(ctrl) - mockBuildStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(storage.ErrAlreadyExists) - - store := storagemock.NewMockStorage(ctrl) - store.EXPECT().GetBatchStore().Return(mockBatchStore).AnyTimes() - store.EXPECT().GetRequestStore().Return(storagemock.NewMockRequestStore(ctrl)).AnyTimes() - store.EXPECT().GetBuildStore().Return(mockBuildStore).AnyTimes() + store, batchStore, treeStore, buildStore, pathBuildStore := newMockStorage(ctrl) + batchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil).AnyTimes() + treeStore.EXPECT().Get(gomock.Any(), batch.ID).Return(tree, nil) + pathBuildStore.EXPECT().Get(gomock.Any(), pathID).Return(entity.SpeculationPathBuild{PathID: pathID, BuildID: "runner-existing"}, nil) + buildStore.EXPECT().Get(gomock.Any(), "runner-existing").Return(existingBuild, nil) + // No Create expectation on either store: an already-mapped, non-terminal + // path must not trigger or persist again. br := buildrunnermock.NewMockBuildRunner(ctrl) - br.EXPECT().Trigger(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(entity.BuildID{ID: "build-dup"}, nil) + // No Trigger expectation: triggering an already-mapped path fails the test. - publishCalled := false - mockPub := queuemock.NewMockPublisher(ctrl) - mockPub.EXPECT().Publish(gomock.Any(), "buildsignal", gomock.Any()).DoAndReturn( - func(_ context.Context, _ string, _ entityqueue.Message) error { - publishCalled = true - return nil + var published []string + controller := newTestController(t, ctrl, store, br, nil, &published) + + require.NoError(t, runProcess(t, ctrl, controller, batch)) + assert.Equal(t, []string{"runner-existing"}, published, "must republish buildsignal for the existing non-terminal build") +} + +// TestController_Process_ExistingMappingTerminalBuildNoOp covers the case +// where a Prioritized path's existing mapping resolves to an already-terminal +// build: nothing happens at all, not even a republish. +func TestController_Process_ExistingMappingTerminalBuildNoOp(t *testing.T) { + ctrl := gomock.NewController(t) + + batch := testBatch() + path := entity.SpeculationPath{Base: nil, Head: batch.ID} + pathID := "test-queue/batch/1/path/0" + tree := entity.SpeculationTree{ + BatchID: batch.ID, + Version: 1, + Paths: []entity.SpeculationPathInfo{ + {ID: pathID, Path: path, Status: entity.SpeculationPathStatusPrioritized, BuildID: "runner-existing"}, }, - ).Times(1) - mockQ := queuemock.NewMockQueue(ctrl) - mockQ.EXPECT().Publisher().Return(mockPub).AnyTimes() - registry, err := consumer.NewTopicRegistry( - []consumer.TopicConfig{{Key: topickey.TopicKeyBuildSignal, Name: "buildsignal", Queue: mockQ}}, - ) - require.NoError(t, err) - controller := NewController(zaptest.NewLogger(t).Sugar(), tally.NoopScope, store, staticBuildRunnerFactory{r: br}, registry, topickey.TopicKeyBuild, "orchestrator-build") + } + existingBuild := entity.Build{ID: "runner-existing", BatchID: batch.ID, SpeculationPathID: pathID, Status: entity.BuildStatusSucceeded} - msg := entityqueue.NewMessage(batch.ID, batchIDPayload(t, batch.ID), batch.Queue, nil) - delivery := queuemock.NewMockDelivery(ctrl) - delivery.EXPECT().Message().Return(msg).AnyTimes() - delivery.EXPECT().Attempt().Return(1).AnyTimes() + store, batchStore, treeStore, buildStore, pathBuildStore := newMockStorage(ctrl) + batchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil).AnyTimes() + treeStore.EXPECT().Get(gomock.Any(), batch.ID).Return(tree, nil) + pathBuildStore.EXPECT().Get(gomock.Any(), pathID).Return(entity.SpeculationPathBuild{PathID: pathID, BuildID: "runner-existing"}, nil) + buildStore.EXPECT().Get(gomock.Any(), "runner-existing").Return(existingBuild, nil) + // No Create, no Trigger, and no publish expected: the build already + // reached a terminal state. - require.NoError(t, controller.Process(context.Background(), delivery)) - assert.True(t, publishCalled, "publish to buildsignal must run even when Create reports ErrAlreadyExists") + br := buildrunnermock.NewMockBuildRunner(ctrl) + + var published []string + controller := newTestController(t, ctrl, store, br, nil, &published) + + require.NoError(t, runProcess(t, ctrl, controller, batch)) + assert.Empty(t, published) } -// TestController_Process_TriggerFailure verifies a build-runner failure is -// surfaced as an error (nack) and nothing is persisted or published. -func TestController_Process_TriggerFailure(t *testing.T) { +// TestController_Process_MappingDanglingInvariantBreachTolerated covers a +// path whose mapping resolves to a build row that no longer exists — an +// invariant breach (the write order in the trigger flow guarantees a mapping +// is only created once its build row exists). It must not crash, error, or +// trigger a duplicate build for the path; the mapping stays authoritative. +func TestController_Process_MappingDanglingInvariantBreachTolerated(t *testing.T) { ctrl := gomock.NewController(t) batch := testBatch() - mockBatchStore := storagemock.NewMockBatchStore(ctrl) - mockBatchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil).AnyTimes() - store := storagemock.NewMockStorage(ctrl) - store.EXPECT().GetBatchStore().Return(mockBatchStore).AnyTimes() - store.EXPECT().GetRequestStore().Return(storagemock.NewMockRequestStore(ctrl)).AnyTimes() - // No build store expectation: Trigger failure must short-circuit before Create. + path := entity.SpeculationPath{Base: nil, Head: batch.ID} + pathID := "test-queue/batch/1/path/0" + tree := entity.SpeculationTree{ + BatchID: batch.ID, + Version: 1, + Paths: []entity.SpeculationPathInfo{ + {ID: pathID, Path: path, Status: entity.SpeculationPathStatusPrioritized, BuildID: "missing-build"}, + }, + } - br := buildrunnermock.NewMockBuildRunner(ctrl) - br.EXPECT().Trigger(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()). - Return(entity.BuildID{}, fmt.Errorf("provider down")) + store, batchStore, treeStore, buildStore, pathBuildStore := newMockStorage(ctrl) + batchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil).AnyTimes() + treeStore.EXPECT().Get(gomock.Any(), batch.ID).Return(tree, nil) + pathBuildStore.EXPECT().Get(gomock.Any(), pathID).Return(entity.SpeculationPathBuild{PathID: pathID, BuildID: "missing-build"}, nil) + buildStore.EXPECT().Get(gomock.Any(), "missing-build").Return(entity.Build{}, storage.ErrNotFound) - registry, err := consumer.NewTopicRegistry( - []consumer.TopicConfig{{Key: topickey.TopicKeyBuildSignal, Name: "buildsignal", Queue: queuemock.NewMockQueue(ctrl)}}, - ) - require.NoError(t, err) - controller := NewController(zaptest.NewLogger(t).Sugar(), tally.NoopScope, store, staticBuildRunnerFactory{r: br}, registry, topickey.TopicKeyBuild, "orchestrator-build") + br := buildrunnermock.NewMockBuildRunner(ctrl) + // No Trigger, no Create, no publish expected — the dangling mapping is + // authoritative even though its target build is missing. - msg := entityqueue.NewMessage(batch.ID, batchIDPayload(t, batch.ID), batch.Queue, nil) - delivery := queuemock.NewMockDelivery(ctrl) - delivery.EXPECT().Message().Return(msg).AnyTimes() - delivery.EXPECT().Attempt().Return(1).AnyTimes() + var published []string + controller := newTestController(t, ctrl, store, br, nil, &published) - require.Error(t, controller.Process(context.Background(), delivery)) + require.NoError(t, runProcess(t, ctrl, controller, batch)) + assert.Empty(t, published) } -func TestController_Process_StorageFailure(t *testing.T) { +// TestController_Process_OnlyPrioritizedPathsTrigger covers a tree with +// Selected, Building, and Passed paths alongside one Prioritized path: +// only the Prioritized path triggers. +func TestController_Process_OnlyPrioritizedPathsTrigger(t *testing.T) { ctrl := gomock.NewController(t) - mockBatchStore := storagemock.NewMockBatchStore(ctrl) - mockBatchStore.EXPECT().Get(gomock.Any(), "test-queue/batch/1").Return(entity.Batch{}, fmt.Errorf("db connection lost")) - store := storagemock.NewMockStorage(ctrl) - store.EXPECT().GetBatchStore().Return(mockBatchStore).AnyTimes() - store.EXPECT().GetRequestStore().Return(storagemock.NewMockRequestStore(ctrl)).AnyTimes() - store.EXPECT().GetBuildStore().Return(storagemock.NewMockBuildStore(ctrl)).AnyTimes() + batch := testBatch() + selectedPath := entity.SpeculationPath{Base: []string{"test-queue/batch/other-1"}, Head: batch.ID} + buildingPath := entity.SpeculationPath{Base: []string{"test-queue/batch/other-2"}, Head: batch.ID} + passedPath := entity.SpeculationPath{Base: []string{"test-queue/batch/other-3"}, Head: batch.ID} + prioritizedPath := entity.SpeculationPath{Base: nil, Head: batch.ID} + prioritizedPathID := "test-queue/batch/1/path/3" + + tree := entity.SpeculationTree{ + BatchID: batch.ID, + Version: 1, + Paths: []entity.SpeculationPathInfo{ + {ID: "test-queue/batch/1/path/0", Path: selectedPath, Status: entity.SpeculationPathStatusSelected}, + {ID: "test-queue/batch/1/path/1", Path: buildingPath, Status: entity.SpeculationPathStatusBuilding, BuildID: "build-in-flight"}, + {ID: "test-queue/batch/1/path/2", Path: passedPath, Status: entity.SpeculationPathStatusPassed, BuildID: "build-done"}, + {ID: prioritizedPathID, Path: prioritizedPath, Status: entity.SpeculationPathStatusPrioritized}, + }, + } - controller := newTestController(t, ctrl, store, buildfake.New(changesetfake.New()), nil) + store, batchStore, treeStore, buildStore, pathBuildStore := newMockStorage(ctrl) + batchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil).AnyTimes() + treeStore.EXPECT().Get(gomock.Any(), batch.ID).Return(tree, nil) + // Only the Prioritized path reaches the dedup Get; Selected, Building, + // and Passed paths continue before it. + pathBuildStore.EXPECT().Get(gomock.Any(), prioritizedPathID).Return(entity.SpeculationPathBuild{}, storage.ErrNotFound) + buildStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil).Times(1) + pathBuildStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil).Times(1) - msg := entityqueue.NewMessage("test-queue/batch/1", batchIDPayload(t, "test-queue/batch/1"), "test-queue", nil) - delivery := queuemock.NewMockDelivery(ctrl) - delivery.EXPECT().Message().Return(msg).AnyTimes() - delivery.EXPECT().Attempt().Return(1).AnyTimes() + br := buildrunnermock.NewMockBuildRunner(ctrl) + // Only the Prioritized path (empty base, since its Path.Base is nil) may trigger. + br.EXPECT().Trigger(gomock.Any(), []entity.Batch(nil), batch, gomock.Nil()).Return(entity.BuildID{ID: "build-new"}, nil) - err := controller.Process(context.Background(), delivery) - require.Error(t, err) - assert.False(t, errs.IsRetryable(err)) + var published []string + controller := newTestController(t, ctrl, store, br, nil, &published) + + require.NoError(t, runProcess(t, ctrl, controller, batch)) + assert.Equal(t, []string{"build-new"}, published) } -func TestController_Process_PublishFailure(t *testing.T) { +// TestController_Process_SpeculationTreeNotFound covers a missing speculation +// tree: the tree exists before any build message is published and is never +// deleted, so a Get miss is an invariant violation. Process must surface the +// error (dead-lettering the message so the DLQ consumer fails the batch +// loudly), not ack. +func TestController_Process_SpeculationTreeNotFound(t *testing.T) { ctrl := gomock.NewController(t) batch := testBatch() - store := newMockStorage(ctrl, batch) - controller := newTestController(t, ctrl, store, buildfake.New(changesetfake.New()), fmt.Errorf("publish failed")) + store, batchStore, treeStore, _, _ := newMockStorage(ctrl) + batchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil).AnyTimes() + treeStore.EXPECT().Get(gomock.Any(), batch.ID).Return(entity.SpeculationTree{}, storage.ErrNotFound) - msg := entityqueue.NewMessage(batch.ID, batchIDPayload(t, batch.ID), batch.Queue, nil) - delivery := queuemock.NewMockDelivery(ctrl) - delivery.EXPECT().Message().Return(msg).AnyTimes() - delivery.EXPECT().Attempt().Return(1).AnyTimes() + controller := newTestController(t, ctrl, store, buildfake.New(changesetfake.New()), nil, nil) - err := controller.Process(context.Background(), delivery) - assert.Error(t, err) + err := runProcess(t, ctrl, controller, batch) + require.Error(t, err) } -func TestController_InterfaceImplementation(t *testing.T) { +// TestController_Process_SpeculationTreeStorageFailure covers a +// non-ErrNotFound failure reading the speculation tree: it is wrapped and +// returned like other storage failures in this controller. +func TestController_Process_SpeculationTreeStorageFailure(t *testing.T) { ctrl := gomock.NewController(t) + batch := testBatch() - store := newMockStorage(ctrl, batch) - controller := newTestController(t, ctrl, store, buildfake.New(changesetfake.New()), nil) + store, batchStore, treeStore, _, _ := newMockStorage(ctrl) + batchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil).AnyTimes() + treeStore.EXPECT().Get(gomock.Any(), batch.ID).Return(entity.SpeculationTree{}, fmt.Errorf("db connection lost")) - var _ consumer.Controller = controller + controller := newTestController(t, ctrl, store, buildfake.New(changesetfake.New()), nil, nil) + + err := runProcess(t, ctrl, controller, batch) + require.Error(t, err) } -// A batch in any halted state (terminal OR cancelling) must short-circuit: -// the build controller acks without triggering an external CI run and without -// publishing anything. Per the cancel design the speculate controller owns -// cancelling in-flight builds and driving the batch terminal, so the build -// stage simply does no work. Cancelling is included because the cancel -// controller is mid-flight; both halted branches reach the same observable -// behaviour (no build performed). +// TestController_Process_HaltedShortCircuit: a batch in any halted state +// (terminal OR cancelling) must short-circuit before touching the +// speculation tree or build stores: the build controller acks without +// triggering an external CI run and without publishing anything. Per the +// cancel design the speculate controller owns cancelling in-flight builds +// and driving the batch terminal, so the build stage simply does no work. +// Cancelling is included because the cancel controller is mid-flight; both +// halted branches reach the same observable behaviour (no build performed). func TestController_Process_HaltedShortCircuit(t *testing.T) { for _, state := range []entity.BatchState{ entity.BatchStateCancelled, @@ -363,7 +453,10 @@ func TestController_Process_HaltedShortCircuit(t *testing.T) { batch := testBatch() batch.State = state - store := newMockStorage(ctrl, batch) + store, batchStore, _, _, _ := newMockStorage(ctrl) + batchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil).AnyTimes() + // No tree/build/path-build store expectations: a halted batch must + // never reach the speculation tree or build stores. // No Trigger expectation: a stray CI trigger on a halted batch // fails the test. @@ -371,14 +464,374 @@ func TestController_Process_HaltedShortCircuit(t *testing.T) { // Sentinel publish error: the halted path must not publish. If it // does, Process surfaces this error and require.NoError catches it. - controller := newTestController(t, ctrl, store, br, fmt.Errorf("should not publish")) - - msg := entityqueue.NewMessage(batch.ID, batchIDPayload(t, batch.ID), batch.Queue, nil) - delivery := queuemock.NewMockDelivery(ctrl) - delivery.EXPECT().Message().Return(msg).AnyTimes() - delivery.EXPECT().Attempt().Return(1).AnyTimes() + controller := newTestController(t, ctrl, store, br, fmt.Errorf("should not publish"), nil) - require.NoError(t, controller.Process(context.Background(), delivery)) + require.NoError(t, runProcess(t, ctrl, controller, batch)) }) } } + +// TestController_Process_CreateAlreadyExistsTolerated covers the redelivery +// case for a single path: the Build row Create returns ErrAlreadyExists (a +// build row can pre-exist from a prior partial pass), but the path->build +// mapping Create still succeeds (no concurrent winner), so the controller +// proceeds to publish to buildsignal. The polling loop will pick up the +// existing row via UpdateStatus. +func TestController_Process_CreateAlreadyExistsTolerated(t *testing.T) { + ctrl := gomock.NewController(t) + + batch := testBatch() + path := entity.SpeculationPath{Base: nil, Head: batch.ID} + pathID := "test-queue/batch/1/path/0" + tree := entity.SpeculationTree{ + BatchID: batch.ID, + Version: 1, + Paths: []entity.SpeculationPathInfo{{ID: pathID, Path: path, Status: entity.SpeculationPathStatusPrioritized}}, + } + + store, batchStore, treeStore, buildStore, pathBuildStore := newMockStorage(ctrl) + batchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil).AnyTimes() + treeStore.EXPECT().Get(gomock.Any(), batch.ID).Return(tree, nil) + pathBuildStore.EXPECT().Get(gomock.Any(), pathID).Return(entity.SpeculationPathBuild{}, storage.ErrNotFound) + buildStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(storage.ErrAlreadyExists) + pathBuildStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil) + + br := buildrunnermock.NewMockBuildRunner(ctrl) + br.EXPECT().Trigger(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(entity.BuildID{ID: "build-dup"}, nil) + + var published []string + controller := newTestController(t, ctrl, store, br, nil, &published) + + require.NoError(t, runProcess(t, ctrl, controller, batch)) + assert.Equal(t, []string{"build-dup"}, published, "publish to buildsignal must run even when the Build row Create reports ErrAlreadyExists") +} + +// TestController_Process_PathBuildCreateRaceLostRepublishesWinner covers a +// concurrent delivery winning the path->build mapping race: our own Trigger +// and Build Create succeed, but the mapping Create loses to a winner. The +// controller must not error — it re-reads the mapping and republishes +// buildsignal for the winner's build, not our own. +func TestController_Process_PathBuildCreateRaceLostRepublishesWinner(t *testing.T) { + ctrl := gomock.NewController(t) + + batch := testBatch() + path := entity.SpeculationPath{Base: nil, Head: batch.ID} + pathID := "test-queue/batch/1/path/0" + tree := entity.SpeculationTree{ + BatchID: batch.ID, + Version: 1, + Paths: []entity.SpeculationPathInfo{{ID: pathID, Path: path, Status: entity.SpeculationPathStatusPrioritized}}, + } + + store, batchStore, treeStore, buildStore, pathBuildStore := newMockStorage(ctrl) + batchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil).AnyTimes() + treeStore.EXPECT().Get(gomock.Any(), batch.ID).Return(tree, nil) + pathBuildStore.EXPECT().Get(gomock.Any(), pathID).Return(entity.SpeculationPathBuild{}, storage.ErrNotFound) + buildStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil) + pathBuildStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(storage.ErrAlreadyExists) + pathBuildStore.EXPECT().Get(gomock.Any(), pathID).Return(entity.SpeculationPathBuild{PathID: pathID, BuildID: "winner-build-id"}, nil) + winnerBuild := entity.Build{ID: "winner-build-id", BatchID: batch.ID, SpeculationPathID: pathID, Status: entity.BuildStatusAccepted} + buildStore.EXPECT().Get(gomock.Any(), "winner-build-id").Return(winnerBuild, nil) + + br := buildrunnermock.NewMockBuildRunner(ctrl) + br.EXPECT().Trigger(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(entity.BuildID{ID: "our-losing-build"}, nil) + + var published []string + controller := newTestController(t, ctrl, store, br, nil, &published) + + require.NoError(t, runProcess(t, ctrl, controller, batch)) + assert.Equal(t, []string{"winner-build-id"}, published, "must republish buildsignal for the mapping race winner, not our own build") +} + +// TestController_Process_TriggerFailure verifies a build-runner failure for +// a Prioritized path is surfaced as an error, and neither Create is +// ever called for that path. +func TestController_Process_TriggerFailure(t *testing.T) { + ctrl := gomock.NewController(t) + + batch := testBatch() + path := entity.SpeculationPath{Base: nil, Head: batch.ID} + pathID := "test-queue/batch/1/path/0" + tree := entity.SpeculationTree{ + BatchID: batch.ID, + Version: 1, + Paths: []entity.SpeculationPathInfo{{ID: pathID, Path: path, Status: entity.SpeculationPathStatusPrioritized}}, + } + + store, batchStore, treeStore, _, pathBuildStore := newMockStorage(ctrl) + batchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil).AnyTimes() + treeStore.EXPECT().Get(gomock.Any(), batch.ID).Return(tree, nil) + pathBuildStore.EXPECT().Get(gomock.Any(), pathID).Return(entity.SpeculationPathBuild{}, storage.ErrNotFound) + // No Create expectation on either store: a Trigger failure must + // short-circuit before either Create. + + br := buildrunnermock.NewMockBuildRunner(ctrl) + br.EXPECT().Trigger(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()). + Return(entity.BuildID{}, fmt.Errorf("provider down")) + + controller := newTestController(t, ctrl, store, br, nil, nil) + + require.Error(t, runProcess(t, ctrl, controller, batch)) +} + +// TestController_Process_EmptyBasePath is the trigger-flow happy path for a +// single Prioritized path with no existing mapping and a nil base (it builds +// directly on the target). The ordering Trigger -> Build Create -> mapping +// Create -> publish is load-bearing for crash-safety, so it is pinned down +// with gomock.InOrder. +func TestController_Process_EmptyBasePath(t *testing.T) { + ctrl := gomock.NewController(t) + + batch := testBatch() + path := entity.SpeculationPath{Base: nil, Head: batch.ID} + pathID := "test-queue/batch/1/path/0" + tree := entity.SpeculationTree{ + BatchID: batch.ID, + Version: 1, + Paths: []entity.SpeculationPathInfo{{ID: pathID, Path: path, Status: entity.SpeculationPathStatusPrioritized}}, + } + + store, batchStore, treeStore, buildStore, pathBuildStore := newMockStorage(ctrl) + batchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil).AnyTimes() + treeStore.EXPECT().Get(gomock.Any(), batch.ID).Return(tree, nil) + pathBuildStore.EXPECT().Get(gomock.Any(), pathID).Return(entity.SpeculationPathBuild{}, storage.ErrNotFound) + + br := buildrunnermock.NewMockBuildRunner(ctrl) + trigger := br.EXPECT().Trigger(gomock.Any(), []entity.Batch(nil), batch, gomock.Nil()).Return(entity.BuildID{ID: "build-no-base"}, nil) + + var capturedMapping entity.SpeculationPathBuild + buildCreate := buildStore.EXPECT().Create(gomock.Any(), entity.Build{ + ID: "build-no-base", + BatchID: batch.ID, + SpeculationPathID: pathID, + Status: entity.BuildStatusAccepted, + }).Return(nil) + mappingCreate := pathBuildStore.EXPECT().Create(gomock.Any(), gomock.Any()).DoAndReturn( + func(_ context.Context, pb entity.SpeculationPathBuild) error { + capturedMapping = pb + return nil + }, + ) + + var published []string + controller := newTestController(t, ctrl, store, br, nil, &published) + + gomock.InOrder(trigger, buildCreate, mappingCreate) + + require.NoError(t, runProcess(t, ctrl, controller, batch)) + assert.Equal(t, []string{"build-no-base"}, published) + assert.Equal(t, pathID, capturedMapping.PathID) + assert.Equal(t, "build-no-base", capturedMapping.BuildID) +} + +// TestController_Process_BatchStorageFailure covers a failure loading the +// head batch itself: the error is wrapped, non-retryable by default, and the +// speculation tree is never consulted. +func TestController_Process_BatchStorageFailure(t *testing.T) { + ctrl := gomock.NewController(t) + + store, batchStore, _, _, _ := newMockStorage(ctrl) + batchStore.EXPECT().Get(gomock.Any(), "test-queue/batch/1").Return(entity.Batch{}, fmt.Errorf("db connection lost")) + // No tree/build/path-build store expectations: a batch load failure must + // short-circuit before the speculation tree is consulted. + + controller := newTestController(t, ctrl, store, buildfake.New(changesetfake.New()), nil, nil) + + err := runProcess(t, ctrl, controller, entity.Batch{ID: "test-queue/batch/1", Queue: "test-queue"}) + require.Error(t, err) + assert.False(t, errs.IsRetryable(err)) +} + +// TestController_Process_PublishFailure covers a publish failure for a +// triggered path: the error propagates, even though Trigger and both Creates +// succeeded. +func TestController_Process_PublishFailure(t *testing.T) { + ctrl := gomock.NewController(t) + + batch := testBatch() + path := entity.SpeculationPath{Base: nil, Head: batch.ID} + pathID := "test-queue/batch/1/path/0" + tree := entity.SpeculationTree{ + BatchID: batch.ID, + Version: 1, + Paths: []entity.SpeculationPathInfo{{ID: pathID, Path: path, Status: entity.SpeculationPathStatusPrioritized}}, + } + + store, batchStore, treeStore, buildStore, pathBuildStore := newMockStorage(ctrl) + batchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil).AnyTimes() + treeStore.EXPECT().Get(gomock.Any(), batch.ID).Return(tree, nil) + pathBuildStore.EXPECT().Get(gomock.Any(), pathID).Return(entity.SpeculationPathBuild{}, storage.ErrNotFound) + buildStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil) + pathBuildStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil) + + controller := newTestController(t, ctrl, store, buildfake.New(changesetfake.New()), fmt.Errorf("publish failed"), nil) + + err := runProcess(t, ctrl, controller, batch) + assert.Error(t, err) +} + +// TestController_Process_CancellingNonTerminalBuildEnactsCancel covers the +// intent/enactment split (D1/D2/D4): a Cancelling path with a non-terminal +// build must have its build runner Cancel called and buildsignal republished +// so the poll loop observes the cancellation promptly. Cancel decisions are +// recorded elsewhere (prioritize's preemption, speculate's reconcile) as +// intent only; this controller is the sole owner of the runner call. +func TestController_Process_CancellingNonTerminalBuildEnactsCancel(t *testing.T) { + ctrl := gomock.NewController(t) + + batch := testBatch() + path := entity.SpeculationPath{Base: nil, Head: batch.ID} + pathID := "test-queue/batch/1/path/0" + tree := entity.SpeculationTree{ + BatchID: batch.ID, + Version: 1, + Paths: []entity.SpeculationPathInfo{{ID: pathID, Path: path, Status: entity.SpeculationPathStatusCancelling}}, + } + inFlightBuild := entity.Build{ID: "build-in-flight", BatchID: batch.ID, SpeculationPathID: pathID, Status: entity.BuildStatusRunning} + + store, batchStore, treeStore, buildStore, pathBuildStore := newMockStorage(ctrl) + batchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil).AnyTimes() + treeStore.EXPECT().Get(gomock.Any(), batch.ID).Return(tree, nil) + pathBuildStore.EXPECT().Get(gomock.Any(), pathID).Return(entity.SpeculationPathBuild{PathID: pathID, BuildID: "build-in-flight"}, nil) + buildStore.EXPECT().Get(gomock.Any(), "build-in-flight").Return(inFlightBuild, nil) + // No Trigger, no Create on either store: a Cancelling path never triggers. + + br := buildrunnermock.NewMockBuildRunner(ctrl) + br.EXPECT().Cancel(gomock.Any(), entity.BuildID{ID: "build-in-flight"}).Return(nil) + + var published []string + controller := newTestController(t, ctrl, store, br, nil, &published) + + require.NoError(t, runProcess(t, ctrl, controller, batch)) + assert.Equal(t, []string{"build-in-flight"}, published, "must republish buildsignal after enacting the cancel") +} + +// TestController_Process_CancellingTerminalBuildNoOp covers a Cancelling path +// whose build has already reached a terminal status (e.g. it finished +// naturally between the cancel decision and this pass): nothing to cancel, +// so Cancel must never be called and nothing is published. +func TestController_Process_CancellingTerminalBuildNoOp(t *testing.T) { + ctrl := gomock.NewController(t) + + batch := testBatch() + path := entity.SpeculationPath{Base: nil, Head: batch.ID} + pathID := "test-queue/batch/1/path/0" + tree := entity.SpeculationTree{ + BatchID: batch.ID, + Version: 1, + Paths: []entity.SpeculationPathInfo{{ID: pathID, Path: path, Status: entity.SpeculationPathStatusCancelling}}, + } + doneBuild := entity.Build{ID: "build-done", BatchID: batch.ID, SpeculationPathID: pathID, Status: entity.BuildStatusSucceeded} + + store, batchStore, treeStore, buildStore, pathBuildStore := newMockStorage(ctrl) + batchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil).AnyTimes() + treeStore.EXPECT().Get(gomock.Any(), batch.ID).Return(tree, nil) + pathBuildStore.EXPECT().Get(gomock.Any(), pathID).Return(entity.SpeculationPathBuild{PathID: pathID, BuildID: "build-done"}, nil) + buildStore.EXPECT().Get(gomock.Any(), "build-done").Return(doneBuild, nil) + + br := buildrunnermock.NewMockBuildRunner(ctrl) + // No Cancel expectation: a terminal build must not be cancelled. + + var published []string + controller := newTestController(t, ctrl, store, br, nil, &published) + + require.NoError(t, runProcess(t, ctrl, controller, batch)) + assert.Empty(t, published) +} + +// TestController_Process_CancellingNoMappingSkips covers a Cancelling path +// with no path->build mapping yet (e.g. the intent was recorded before this +// controller ever triggered a build for the path): there is nothing to +// cancel, so the pass is a silent no-op left for speculate's reconcile to +// settle. +func TestController_Process_CancellingNoMappingSkips(t *testing.T) { + ctrl := gomock.NewController(t) + + batch := testBatch() + path := entity.SpeculationPath{Base: nil, Head: batch.ID} + pathID := "test-queue/batch/1/path/0" + tree := entity.SpeculationTree{ + BatchID: batch.ID, + Version: 1, + Paths: []entity.SpeculationPathInfo{{ID: pathID, Path: path, Status: entity.SpeculationPathStatusCancelling}}, + } + + store, batchStore, treeStore, _, pathBuildStore := newMockStorage(ctrl) + batchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil).AnyTimes() + treeStore.EXPECT().Get(gomock.Any(), batch.ID).Return(tree, nil) + pathBuildStore.EXPECT().Get(gomock.Any(), pathID).Return(entity.SpeculationPathBuild{}, storage.ErrNotFound) + // No BuildStore.Get, no Cancel: nothing to look up or cancel. + + br := buildrunnermock.NewMockBuildRunner(ctrl) + + var published []string + controller := newTestController(t, ctrl, store, br, nil, &published) + + require.NoError(t, runProcess(t, ctrl, controller, batch)) + assert.Empty(t, published) +} + +// TestController_Process_CancellingMappingDanglingSkips covers a Cancelling +// path whose mapping resolves to a build row that no longer exists (the same +// invariant-breach defense as the trigger flow): tolerated as a no-op rather +// than an error or a fresh trigger. +func TestController_Process_CancellingMappingDanglingSkips(t *testing.T) { + ctrl := gomock.NewController(t) + + batch := testBatch() + path := entity.SpeculationPath{Base: nil, Head: batch.ID} + pathID := "test-queue/batch/1/path/0" + tree := entity.SpeculationTree{ + BatchID: batch.ID, + Version: 1, + Paths: []entity.SpeculationPathInfo{{ID: pathID, Path: path, Status: entity.SpeculationPathStatusCancelling}}, + } + + store, batchStore, treeStore, buildStore, pathBuildStore := newMockStorage(ctrl) + batchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil).AnyTimes() + treeStore.EXPECT().Get(gomock.Any(), batch.ID).Return(tree, nil) + pathBuildStore.EXPECT().Get(gomock.Any(), pathID).Return(entity.SpeculationPathBuild{PathID: pathID, BuildID: "missing-build"}, nil) + buildStore.EXPECT().Get(gomock.Any(), "missing-build").Return(entity.Build{}, storage.ErrNotFound) + + br := buildrunnermock.NewMockBuildRunner(ctrl) + // No Cancel expectation — the dangling mapping is authoritative even + // though its target build is missing. + + var published []string + controller := newTestController(t, ctrl, store, br, nil, &published) + + require.NoError(t, runProcess(t, ctrl, controller, batch)) + assert.Empty(t, published) +} + +// TestController_Process_CancelErrorSurfaces covers a runner Cancel failure: +// it must surface as an error (never a silent ack), and buildsignal must not +// be republished for a cancel that never completed. +func TestController_Process_CancelErrorSurfaces(t *testing.T) { + ctrl := gomock.NewController(t) + + batch := testBatch() + path := entity.SpeculationPath{Base: nil, Head: batch.ID} + pathID := "test-queue/batch/1/path/0" + tree := entity.SpeculationTree{ + BatchID: batch.ID, + Version: 1, + Paths: []entity.SpeculationPathInfo{{ID: pathID, Path: path, Status: entity.SpeculationPathStatusCancelling}}, + } + inFlightBuild := entity.Build{ID: "build-in-flight", BatchID: batch.ID, SpeculationPathID: pathID, Status: entity.BuildStatusRunning} + + store, batchStore, treeStore, buildStore, pathBuildStore := newMockStorage(ctrl) + batchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil).AnyTimes() + treeStore.EXPECT().Get(gomock.Any(), batch.ID).Return(tree, nil) + pathBuildStore.EXPECT().Get(gomock.Any(), pathID).Return(entity.SpeculationPathBuild{PathID: pathID, BuildID: "build-in-flight"}, nil) + buildStore.EXPECT().Get(gomock.Any(), "build-in-flight").Return(inFlightBuild, nil) + + br := buildrunnermock.NewMockBuildRunner(ctrl) + br.EXPECT().Cancel(gomock.Any(), entity.BuildID{ID: "build-in-flight"}).Return(fmt.Errorf("runner unreachable")) + + var published []string + controller := newTestController(t, ctrl, store, br, nil, &published) + + require.Error(t, runProcess(t, ctrl, controller, batch)) + assert.Empty(t, published, "must not republish buildsignal when the cancel itself failed") +} diff --git a/submitqueue/orchestrator/controller/buildsignal/buildsignal.go b/submitqueue/orchestrator/controller/buildsignal/buildsignal.go index 9be2d484..33c089aa 100644 --- a/submitqueue/orchestrator/controller/buildsignal/buildsignal.go +++ b/submitqueue/orchestrator/controller/buildsignal/buildsignal.go @@ -13,13 +13,16 @@ // 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. +// a Build's ID; the controller loads the row, calls BuildRunner.Status +// against the build's ID — which is both the build store's primary key and +// the runner-minted handle, there is no separate store-key/runner-ID split +// — writes the latest status to the BuildStore (keyed by the same ID), +// 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 build partitions independently by +// its ID, 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 import ( @@ -93,14 +96,14 @@ func NewController( // a delayed message back to this topic when the build is still in flight. // Returns nil to ack (success), or error to nack/reject. // -// Error classification: deserialize, Status, UpdateStatus, 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. +// Error classification: retryability is decided per error cause by the +// wired classifier walk, not per call site. Only transient infra causes the +// classifiers recognize (context cancellation, transient MySQL driver/server +// errors — which covers both storage calls and queue publishes, since both +// run on the same driver) nack and replay up to MaxAttempts. Everything +// else — malformed payloads, domain storage sentinels, runner Status +// failures — is non-retryable by default and rejects to DLQ on the first +// failure, where the buildsignal DLQ consumer fails the batch loudly. func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) { const opName = "process" @@ -116,9 +119,9 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r 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's ID travels on the queue; load the full Build from + // storage, which is the single source of truth for its BatchID, its ID + // (also the runner handle), and the snapshot the poll loop updates. build, err := c.store.GetBuildStore().Get(ctx, buildID.ID) if err != nil { metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) @@ -146,7 +149,7 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r return fmt.Errorf("failed to build runner for batch %s: %w", batch.ID, err) } - status, _, err := buildRunner.Status(ctx, buildID) + status, _, err := buildRunner.Status(ctx, entity.BuildID{ID: build.ID}) if err != nil { metrics.NamedCounter(c.metricsScope, opName, "status_errors", 1) return fmt.Errorf("failed to get status for build %s: %w", buildID.ID, err) diff --git a/submitqueue/orchestrator/controller/speculate/speculate.go b/submitqueue/orchestrator/controller/speculate/speculate.go index 50380d2d..04971601 100644 --- a/submitqueue/orchestrator/controller/speculate/speculate.go +++ b/submitqueue/orchestrator/controller/speculate/speculate.go @@ -35,24 +35,25 @@ import ( // Controller handles speculate queue messages. // -// Each invocation reconciles the batch's entity.SpeculationTree and advances -// the batch one step in the state machine. Tree reconciliation is pure -// mechanics: the controller runs the queue's configured seams — enumerator -// (path structure), path scorer (path scores), selector (path decisions) — -// validates their outputs, applies them to the persisted tree, and writes it -// back under optimistic concurrency. Which paths exist, how they score, and -// which are promoted or cancelled are the seams' decisions alone; the -// controller never originates one. No downstream stage reads the tree yet, -// so the forward step below is driven by the batch's own state, not the -// tree. +// Each invocation reconciles the batch's entity.SpeculationTree, publishes +// the batch's queue to prioritize, and advances the batch one step in the +// state machine. Tree reconciliation is pure mechanics: the controller runs +// the queue's configured seams — enumerator (path structure), path scorer +// (path scores), selector (path decisions) — validates their outputs, +// applies them to the persisted tree, and writes it back under optimistic +// concurrency. Which paths exist, how they score, and which are promoted or +// cancelled are the seams' decisions alone; the controller never originates +// one. The batch's own merge timing is still driven by dependency states +// (tryFinalize), pending a later change that gates it on the tree's path +// outcomes. // // Per invocation, the controller advances the batch one step in the state // machine: // // - Created, Scored, or Speculating → speculateBatch: reconcile the tree -// (created on the first pass the dependency gate admits), then advance -// the batch — publish to build and CAS to Speculating for -// Created/Scored, or tryFinalize for Speculating. +// (created on the first pass the dependency gate admits), CAS to +// Speculating for Created/Scored, publish to prioritize every pass, and +// run tryFinalize once the batch has reached Speculating. // - Cancelling → cancel any in-flight Build entity, respeculate // dependents, CAS to terminal Cancelled, publish to conclude. The // cancel controller hands the batch off in this state and speculate @@ -66,7 +67,7 @@ import ( // // Cancel decisions are recorded as path status only // (SpeculationPathStatusCancelling on an in-flight build) — nothing in this -// file calls a build runner. +// file calls a build runner or triggers builds for admitted paths. // // The controller is re-triggered on every relevant downstream event // (buildsignal, merge), so each call simply re-evaluates the current @@ -178,9 +179,9 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r // speculateBatch is the unified entry point for Created, Scored, and // Speculating batches. It loads the batch's speculation tree (creating it on // the first pass the dependency gate admits), applies the scorer's and -// selector's outputs, persists the tree if anything changed, and then -// advances the batch itself: Created/Scored publish to build and CAS to -// Speculating; Speculating runs tryFinalize. +// selector's outputs, persists the tree if anything changed, CASes the batch +// to Speculating on its first pass, publishes the queue to prioritize every +// pass, and runs tryFinalize once the batch has reached Speculating. func (c *Controller) speculateBatch(ctx context.Context, batch entity.Batch) error { deps, err := c.fetchDependencies(ctx, batch) if err != nil { @@ -249,16 +250,38 @@ func (c *Controller) speculateBatch(ctx context.Context, batch entity.Batch) err tree.Version = newVersion } - // The tree does not influence the forward step below — no downstream - // stage consumes it yet. - switch batch.State { - case entity.BatchStateCreated, entity.BatchStateScored: - return c.startSpeculation(ctx, batch) - case entity.BatchStateSpeculating: + // Optimistic CAS: if the version has already advanced (concurrent + // speculate), the next event will see the new state and behave correctly. + originalState := batch.State + if batch.State == entity.BatchStateCreated || batch.State == entity.BatchStateScored { + newVersion := batch.Version + 1 + if err := c.store.GetBatchStore().UpdateState(ctx, batch.ID, batch.Version, newVersion, entity.BatchStateSpeculating); err != nil { + metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) + return fmt.Errorf("failed to update batch %s state to speculating: %w", batch.ID, err) + } + batch.Version = newVersion + batch.State = entity.BatchStateSpeculating + } + + // Publish to prioritize every pass, after the CAS: publishing before the + // CAS could let a concurrent prioritize round observe the batch as not + // yet Speculating and skip it with no later wake-up. Publishing after is + // healed by redelivery — an error or crash here nacks the message, and + // the next pass republishes, since this runs on every pass regardless of + // whether anything changed. + if err := c.publishQueue(ctx, topickey.TopicKeyPrioritize, batch.Queue); err != nil { + metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1) + return fmt.Errorf("failed to publish queue %s to prioritize: %w", batch.Queue, err) + } + + // Legacy forward path: tryFinalize only runs once a batch has already + // reached Speculating on a prior pass — merge timing is unchanged from + // before prioritize/build existed. This tightens in a later commit to + // gate on the tree's own path outcome instead of dependency states alone. + if originalState == entity.BatchStateSpeculating { return c.tryFinalize(ctx, batch) - default: - return nil } + return nil } // dependencyGateBlocks reports whether batch's count of active dependencies @@ -502,31 +525,6 @@ func (c *Controller) applySelection(batch entity.Batch, tree entity.SpeculationT return tree, changed } -// startSpeculation publishes the batch to the build stage, then transitions -// it to Speculating. -func (c *Controller) startSpeculation(ctx context.Context, batch entity.Batch) error { - c.logger.Infow("starting speculation", - "batch_id", batch.ID, - "speculation_chain", append(append([]string{}, batch.Dependencies...), batch.ID), - ) - - if err := c.publish(ctx, topickey.TopicKeyBuild, batch.ID, batch.Queue); err != nil { - metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1) - return fmt.Errorf("failed to publish to build: %w", err) - } - - // Optimistic CAS: if the version has already advanced (concurrent speculate), - // the next event will see the new state and behave correctly. - newVersion := batch.Version + 1 - if err := c.store.GetBatchStore().UpdateState(ctx, batch.ID, batch.Version, newVersion, entity.BatchStateSpeculating); err != nil { - metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) - return fmt.Errorf("failed to update batch %s state to speculating: %w", batch.ID, err) - } - - metrics.NamedCounter(c.metricsScope, opName, "started_speculation", 1) - return nil -} - // tryFinalize publishes to merge and transitions to Merging iff every // dependency batch has reached Succeeded. Cancelled deps are treated as // out-of-the-way: the cancelled batch will never land, so it can no longer @@ -808,6 +806,35 @@ func (c *Controller) publish(ctx context.Context, key consumer.TopicKey, batchID return nil } +// publishQueue publishes a queue name to the specified topic key. Used for +// queue-scoped stages (prioritize) whose message payload carries no batch +// identity at all — just the queue to re-evaluate. +func (c *Controller) publishQueue(ctx context.Context, key consumer.TopicKey, queue string) error { + qid := entity.QueueID{Name: queue} + payload, err := qid.ToBytes() + if err != nil { + return fmt.Errorf("failed to serialize queue ID: %w", err) + } + + msg := entityqueue.NewMessage(queue, payload, queue, nil) + + q, ok := c.registry.Queue(key) + if !ok { + return fmt.Errorf("no queue registered for topic key %s", key) + } + + topicName, ok := c.registry.TopicName(key) + if !ok { + return fmt.Errorf("no topic name registered for topic key %s", key) + } + + if err := q.Publisher().Publish(ctx, topicName, msg); err != nil { + return fmt.Errorf("failed to publish message: %w", err) + } + + return nil +} + // Name returns the controller name for logging and metrics. func (c *Controller) Name() string { return "speculate" diff --git a/submitqueue/orchestrator/controller/speculate/speculate_test.go b/submitqueue/orchestrator/controller/speculate/speculate_test.go index d9425bec..1d755049 100644 --- a/submitqueue/orchestrator/controller/speculate/speculate_test.go +++ b/submitqueue/orchestrator/controller/speculate/speculate_test.go @@ -91,22 +91,29 @@ type testHarness struct { // dependency limit is effectively ungated (1000). Every publish succeeds and // is appended to records. func newTestHarness(t *testing.T, ctrl *gomock.Controller) *testHarness { - return newHarness(t, ctrl, nil, 1000) + return newHarness(t, ctrl, nil, "", 1000) } // newFailingPublishHarness is identical to newTestHarness except every // publish returns publishErr instead of succeeding. func newFailingPublishHarness(t *testing.T, ctrl *gomock.Controller, publishErr error) *testHarness { - return newHarness(t, ctrl, publishErr, 1000) + return newHarness(t, ctrl, publishErr, "", 1000) +} + +// newTopicFailingPublishHarness is identical to newTestHarness except +// publishes to failTopic return publishErr; publishes to every other topic +// succeed and are recorded. +func newTopicFailingPublishHarness(t *testing.T, ctrl *gomock.Controller, publishErr error, failTopic string) *testHarness { + return newHarness(t, ctrl, publishErr, failTopic, 1000) } // newDependencyLimitHarness is identical to newTestHarness except the // dependency limit is set to limit instead of the default ungated value. func newDependencyLimitHarness(t *testing.T, ctrl *gomock.Controller, limit int) *testHarness { - return newHarness(t, ctrl, nil, limit) + return newHarness(t, ctrl, nil, "", limit) } -func newHarness(t *testing.T, ctrl *gomock.Controller, publishErr error, depLimit int) *testHarness { +func newHarness(t *testing.T, ctrl *gomock.Controller, publishErr error, failTopic string, depLimit int) *testHarness { batchStore := storagemock.NewMockBatchStore(ctrl) treeStore := storagemock.NewMockSpeculationTreeStore(ctrl) buildStore := storagemock.NewMockBuildStore(ctrl) @@ -138,7 +145,7 @@ func newHarness(t *testing.T, ctrl *gomock.Controller, publishErr error, depLimi pub := queuemock.NewMockPublisher(ctrl) pub.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn( func(_ context.Context, topic string, msg entityqueue.Message) error { - if publishErr != nil { + if publishErr != nil && (failTopic == "" || topic == failTopic) { return publishErr } records = append(records, pubRec{topic: topic, msgID: msg.ID}) @@ -150,7 +157,7 @@ func newHarness(t *testing.T, ctrl *gomock.Controller, publishErr error, depLimi registry, err := consumer.NewTopicRegistry([]consumer.TopicConfig{ {Key: topickey.TopicKeySpeculate, Name: "speculate", Queue: mockQ}, - {Key: topickey.TopicKeyBuild, Name: "build", Queue: mockQ}, + {Key: topickey.TopicKeyPrioritize, Name: "prioritize", Queue: mockQ}, {Key: topickey.TopicKeyMerge, Name: "merge", Queue: mockQ}, {Key: topickey.TopicKeyConclude, Name: "conclude", Queue: mockQ}, }) @@ -322,8 +329,9 @@ func TestController_Process_TerminalMissingDependentRowErrors(t *testing.T) { // Created batch: no tree yet -> enumerate, stamp Candidate, mint path IDs, // Version 1, Create. Score echoes. Select promotes the lone path to Selected -// -> tree changed -> Update(1,2,...). The forward step then runs -// unchanged: batch CASes Created -> Speculating and publishes to build. +// -> tree changed -> Update(1,2,...). The forward step then CASes Created -> +// Speculating and publishes the queue to prioritize; tryFinalize does not run +// on this first pass (it only runs once the batch has reached Speculating). func TestController_Process_CreatedEnumeratesScoresSelects(t *testing.T) { ctrl := gomock.NewController(t) h := newTestHarness(t, ctrl) @@ -353,7 +361,7 @@ func TestController_Process_CreatedEnumeratesScoresSelects(t *testing.T) { h.batchStore.EXPECT().UpdateState(gomock.Any(), batch.ID, int32(1), int32(2), entity.BatchStateSpeculating).Return(nil) require.NoError(t, runProcess(t, ctrl, h.controller, batch.ID)) - assert.Equal(t, []pubRec{{topic: "build", msgID: batch.ID}}, *h.records) + assert.Equal(t, []pubRec{{topic: "prioritize", msgID: batch.Queue}}, *h.records) } // TestController_Process_CreateTreeMintsPathIDs pins down the exact minted @@ -381,12 +389,12 @@ func TestController_Process_CreateTreeMintsPathIDs(t *testing.T) { h.batchStore.EXPECT().UpdateState(gomock.Any(), batch.ID, int32(1), int32(2), entity.BatchStateSpeculating).Return(nil) require.NoError(t, runProcess(t, ctrl, h.controller, batch.ID)) - assert.Equal(t, []pubRec{{topic: "build", msgID: batch.ID}}, *h.records) + assert.Equal(t, []pubRec{{topic: "prioritize", msgID: batch.Queue}}, *h.records) } // Create racing with a concurrent creator: ErrAlreadyExists must fall back to -// re-reading the winner's tree rather than erroring. The forward step -// still runs afterward. +// re-reading the winner's tree rather than erroring. The forward step still +// runs afterward. func TestController_Process_CreateTreeAlreadyExistsRereads(t *testing.T) { ctrl := gomock.NewController(t) h := newTestHarness(t, ctrl) @@ -415,7 +423,7 @@ func TestController_Process_CreateTreeAlreadyExistsRereads(t *testing.T) { h.batchStore.EXPECT().UpdateState(gomock.Any(), batch.ID, int32(1), int32(2), entity.BatchStateSpeculating).Return(nil) require.NoError(t, runProcess(t, ctrl, h.controller, batch.ID)) - assert.Equal(t, []pubRec{{topic: "build", msgID: batch.ID}}, *h.records) + assert.Equal(t, []pubRec{{topic: "prioritize", msgID: batch.Queue}}, *h.records) } // Dependency gate: too many active dependencies for a batch with no tree yet @@ -440,9 +448,10 @@ func TestController_Process_DependencyGateBlocksTreeCreation(t *testing.T) { assert.Empty(t, *h.records) } -// A pass over an unchanged tree must not call Update. The forward -// step still runs: here a still-pending dependency makes tryFinalize wait, -// so no publish happens at all. +// A pass over an unchanged tree must not call Update. The batch still +// publishes the queue to prioritize every pass regardless of tree changes; +// tryFinalize waits on the still-pending dependency, so no merge/conclude +// publish happens. func TestController_Process_NoChangeSkipsTreeUpdate(t *testing.T) { ctrl := gomock.NewController(t) h := newTestHarness(t, ctrl) @@ -460,10 +469,10 @@ func TestController_Process_NoChangeSkipsTreeUpdate(t *testing.T) { h.treeStore.EXPECT().Get(gomock.Any(), batch.ID).Return(tree, nil) // Fake scorer echoes, fake selector decides nothing -> no change. // No treeStore.Update expected. tryFinalize waits on the pending dep, so - // no UpdateState/publish either. + // no UpdateState/merge publish either. require.NoError(t, runProcess(t, ctrl, h.controller, batch.ID)) - assert.Empty(t, *h.records) + assert.Equal(t, []pubRec{{topic: "prioritize", msgID: batch.Queue}}, *h.records) } // A version mismatch on the speculation tree Update must surface as an error @@ -494,7 +503,8 @@ func TestController_Process_TreeUpdateVersionMismatchErrors(t *testing.T) { // tryFinalize: Speculating with no deps should publish to merge and CAS to // Merging. The batch already has an (empty) speculation tree from its -// Created/Scored pass. +// Created/Scored pass. Every pass also publishes the queue to prioritize, +// before tryFinalize's own merge publish. func TestController_Process_FinalizeNoDeps(t *testing.T) { ctrl := gomock.NewController(t) h := newTestHarness(t, ctrl) @@ -505,7 +515,10 @@ func TestController_Process_FinalizeNoDeps(t *testing.T) { h.batchStore.EXPECT().UpdateState(gomock.Any(), batch.ID, int32(1), int32(2), entity.BatchStateMerging).Return(nil) require.NoError(t, runProcess(t, ctrl, h.controller, batch.ID)) - assert.Equal(t, []pubRec{{topic: "merge", msgID: batch.ID}}, *h.records) + assert.Equal(t, []pubRec{ + {topic: "prioritize", msgID: batch.Queue}, + {topic: "merge", msgID: batch.ID}, + }, *h.records) } // tryFinalize: Speculating with all deps Succeeded should publish to merge @@ -526,11 +539,15 @@ func TestController_Process_FinalizeAllDepsSucceeded(t *testing.T) { h.batchStore.EXPECT().UpdateState(gomock.Any(), batch.ID, int32(1), int32(2), entity.BatchStateMerging).Return(nil) require.NoError(t, runProcess(t, ctrl, h.controller, batch.ID)) - assert.Equal(t, []pubRec{{topic: "merge", msgID: batch.ID}}, *h.records) + assert.Equal(t, []pubRec{ + {topic: "prioritize", msgID: batch.Queue}, + {topic: "merge", msgID: batch.ID}, + }, *h.records) } -// tryFinalize: Speculating with a dep still in flight is a no-op (no -// publish, no state change). +// tryFinalize: Speculating with a dep still in flight is a no-op (no merge +// publish, no state change). The queue is still published to prioritize +// every pass. func TestController_Process_WaitingOnDep(t *testing.T) { ctrl := gomock.NewController(t) h := newTestHarness(t, ctrl) @@ -545,7 +562,7 @@ func TestController_Process_WaitingOnDep(t *testing.T) { // No UpdateState expected — gomock will fail if it is called. require.NoError(t, runProcess(t, ctrl, h.controller, batch.ID)) - assert.Empty(t, *h.records) + assert.Equal(t, []pubRec{{topic: "prioritize", msgID: batch.Queue}}, *h.records) } // tryFinalize: a failed dep must fail the batch (Speculating → Failed), wake @@ -572,6 +589,7 @@ func TestController_Process_FailedDepFailsBatch(t *testing.T) { require.NoError(t, runProcess(t, ctrl, h.controller, batch.ID)) assert.Equal(t, []pubRec{ + {topic: "prioritize", msgID: batch.Queue}, {topic: "speculate", msgID: "test-queue/batch/2"}, {topic: "conclude", msgID: batch.ID}, }, *h.records) @@ -579,10 +597,11 @@ func TestController_Process_FailedDepFailsBatch(t *testing.T) { // A dependent-wake publish failure after the terminal CAS nacks the message; // redelivery converges through the terminal branch, which re-runs the -// fan-out and the conclude publish. +// fan-out and the conclude publish. Only the speculate topic fails here — +// the prioritize publish earlier in the pass succeeds and is recorded. func TestController_Process_FailedDepDependentPublishFailure(t *testing.T) { ctrl := gomock.NewController(t) - h := newFailingPublishHarness(t, ctrl, fmt.Errorf("publish failed")) + h := newTopicFailingPublishHarness(t, ctrl, fmt.Errorf("publish failed"), "speculate") dep := entity.Batch{ID: "test-queue/batch/0", Queue: "test-queue", State: entity.BatchStateFailed, Version: 1} batch := testBatch(entity.BatchStateSpeculating, dep.ID) @@ -599,7 +618,7 @@ func TestController_Process_FailedDepDependentPublishFailure(t *testing.T) { }, nil) require.Error(t, runProcess(t, ctrl, h.controller, batch.ID)) - assert.Empty(t, *h.records) + assert.Equal(t, []pubRec{{topic: "prioritize", msgID: batch.Queue}}, *h.records) } // tryFinalize: a cancelled dep is treated as out-of-the-way — it will never @@ -621,7 +640,10 @@ func TestController_Process_CancelledDepSkipped(t *testing.T) { h.batchStore.EXPECT().UpdateState(gomock.Any(), batch.ID, int32(1), int32(2), entity.BatchStateMerging).Return(nil) require.NoError(t, runProcess(t, ctrl, h.controller, batch.ID)) - assert.Equal(t, []pubRec{{topic: "merge", msgID: batch.ID}}, *h.records) + assert.Equal(t, []pubRec{ + {topic: "prioritize", msgID: batch.Queue}, + {topic: "merge", msgID: batch.ID}, + }, *h.records) } // Cancelling drives the terminal-cancellation flow: cancel any in-flight @@ -746,7 +768,9 @@ func TestController_Process_CancellingTerminalCASVersionMismatch(t *testing.T) { assert.Empty(t, *h.records) } -// Publish failure must not advance the batch state. +// Publish failure must not advance the batch state further: the CAS to +// Speculating (which precedes the prioritize publish) still lands, but +// tryFinalize must never run since the publish error aborts speculateBatch. func TestController_Process_PublishFailure(t *testing.T) { ctrl := gomock.NewController(t) h := newFailingPublishHarness(t, ctrl, fmt.Errorf("publish failed")) @@ -754,7 +778,7 @@ func TestController_Process_PublishFailure(t *testing.T) { h.batchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil) h.treeStore.EXPECT().Get(gomock.Any(), batch.ID).Return(entity.SpeculationTree{BatchID: batch.ID, Version: 1, Paths: []entity.SpeculationPathInfo{}}, nil) - // No UpdateState expected — publish fails before we get there. + h.batchStore.EXPECT().UpdateState(gomock.Any(), batch.ID, int32(1), int32(2), entity.BatchStateSpeculating).Return(nil) require.Error(t, runProcess(t, ctrl, h.controller, batch.ID)) }