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