feat(speculation): cancel batches through the tree via prioritize

submitqueue/entity/batch.go

@@ -38,9 +38,9 @@ const (
 	// batch has not yet been transitioned to BatchStateCancelled. A batch in this state may still reach
 	// BatchStateSucceeded or BatchStateFailed if a concurrent merge wins the race (e.g. the push had
 	// already completed before the cancel CAS observed the batch); those terminal states prevail.
-	// Forward-progress controllers must treat this state as halted (no new work). The speculate
-	// controller owns the transition to the terminal BatchStateCancelled and the downstream fan-out
-	// (cancelling in-flight builds, respeculating dependents, publishing to conclude).
+	// The state holds while the batch's in-flight builds wind down: no new pipeline work may start for
+	// the batch, and the transition to the terminal BatchStateCancelled follows once every build has
+	// been observed in a terminal status.
 	BatchStateCancelling BatchState = "cancelling"
 	// BatchStateCancelled is the terminal state of a batch that was cancelled before completion.
 	BatchStateCancelled BatchState = "cancelled"
@@ -62,9 +62,9 @@ func (s BatchState) IsTerminal() bool {
 }
 
 // IsBatchStateHalted returns true if the batch is either terminal or in the process of being cancelled.
-// Forward-progress controllers (score, build, buildsignal, speculate, merge) use this to short-circuit
-// work for batches that the user has asked to cancel — even though Cancelling is non-terminal, no
-// further pipeline work should start (cancel will write the terminal state and fan out).
+// Use it to gate work that must not start for a batch that will not proceed — even though Cancelling
+// is non-terminal, a halted batch makes no forward progress (cancellation will write the terminal
+// state once in-flight work quiesces).
 func IsBatchStateHalted(s BatchState) bool {
 	return s.IsTerminal() || s == BatchStateCancelling
 }

submitqueue/entity/request_log.go

@@ -58,7 +58,7 @@ const (
 	// RequestStatusSpeculated indicates that the request has been successfully speculated and is ready to be validated via a build system.
 	RequestStatusSpeculated RequestStatus = "speculated"
 
-	// RequestStatusBuilding indicates that the request is currently being built (e.g., CI/CD system is building the change on top of the speculation path).
+	// RequestStatusBuilding indicates that the request is currently being built (e.g., the build system is building the change on top of the speculation path).
 	RequestStatusBuilding RequestStatus = "building"
 
 	// RequestStatusBuilt indicates that the request has finished the build step successfully and can move to the next phase, either wait for other requests to finish or move to the land phase.

submitqueue/orchestrator/controller/build/build.go

@@ -42,17 +42,23 @@ import (
 // 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.
+// Terminal batches are skipped before any path work: the cancel flow only
+// writes its terminal state after every build has quiesced, and other
+// terminal transitions leave stragglers to run out, so there is nothing for
+// this controller to enact. A Cancelling batch is the deliberate exception —
+// it is being torn down batch-wide, so the loop cancels every path's
+// in-flight build regardless of the path's recorded status (the sweep that
+// records per-path intents may not have run yet), which is how batch-level
+// cancellation reaches the runner — but no new build is ever triggered 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.
+// (never-persisted) old one and simply wins, orphaning the earlier build
+// in the build system. 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
@@ -95,7 +101,9 @@ func NewController(
 // 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.
+// appropriate. For a Cancelling batch it instead cancels every path's
+// in-flight build, whatever the path's recorded status, and triggers
+// nothing.
 // Returns nil to ack (success), or error to nack/reject.
 func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) {
 	op := metrics.Begin(c.metricsScope, opName)
@@ -126,14 +134,16 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r
 		"partition_key", msg.PartitionKey,
 	)
 
-	// If the batch is halted (terminal OR cancelling), skip triggering CI and
-	// ack. This is a forward-progress controller: per the cancel design, the
-	// speculate controller owns cancelling any in-flight Build and driving the
-	// batch to its terminal state, so the build stage simply short-circuits
-	// while speculate does the work. No external CI is ever kicked off.
-	if entity.IsBatchStateHalted(batch.State) {
-		metrics.NamedCounter(c.metricsScope, opName, "skipped_halted", 1)
-		c.logger.Infow("skipping build for halted batch",
+	// Terminal batches are settled — the cancel flow only writes Cancelled
+	// once every build has been observed terminal, and Failed/Succeeded
+	// leave any straggler builds to run out — so there is no path work left
+	// here. Cancelling batches proceed: the per-path loop below cancels
+	// every in-flight build of a batch being torn down (that is how a
+	// batch-level cancel reaches the runner) and guarantees no new build is
+	// ever kicked off for it.
+	if batch.State.IsTerminal() {
+		metrics.NamedCounter(c.metricsScope, opName, "skipped_terminal", 1)
+		c.logger.Infow("skipping build for terminal batch",
 			"batch_id", batch.ID,
 			"state", string(batch.State),
 		)
@@ -162,6 +172,31 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r
 
 	triggered := 0
 	for _, p := range tree.Paths {
+		if batch.State == entity.BatchStateCancelling {
+			// Batch-wide teardown: every path of a Cancelling batch is
+			// doomed, so any in-flight build is cancelled right here,
+			// regardless of the path's recorded status. A path can still
+			// read Building when this message races the tree sweep that
+			// records the intents, or already read Cancelled from another
+			// signal — a speculation-level cancel (a selector or
+			// prioritizer decision) that landed before its build was
+			// stamped onto the tree — and waiting for the sweep would only
+			// leave the build running in the build system one round-trip
+			// longer. Passed and Failed are skipped without I/O: those
+			// statuses derive from terminal builds, so there is nothing to
+			// stop. enactCancel tolerates paths with no build. Never
+			// trigger a new build for a batch being torn down; speculate's
+			// sweep owns settling the path statuses and the terminal batch
+			// write.
+			if p.Status == entity.SpeculationPathStatusPassed || p.Status == entity.SpeculationPathStatusFailed {
+				continue
+			}
+			if err := c.enactCancel(ctx, batch, p, resolveRunner); err != nil {
+				return err
+			}
+			continue
+		}
+
 		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
@@ -305,8 +340,8 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r
 			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
+				// orphan: lost mapping races cost one orphaned build in the
+				// build system 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
@@ -378,20 +413,21 @@ func (c *Controller) runnerResolver(queue string) func() (buildrunner.BuildRunne
 	}
 }
 
-// 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.
+// enactCancel stops path p's build, if one is in flight: 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. Called both for a path's persisted Cancelling intent
+// and for every path of a Cancelling batch's batch-wide teardown. 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
+// path can be marked for cancellation 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.
+// nothing to cancel, so settling the path's status is left to speculate's
+// 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 {

submitqueue/orchestrator/controller/build/build_test.go

@@ -159,7 +159,7 @@ func TestController_Process_NoPrioritizedPaths(t *testing.T) {
 	// to dedup-check.
 
 	br := buildrunnermock.NewMockBuildRunner(ctrl)
-	// No Trigger expectation: a stray CI trigger with no Prioritized path
+	// No Trigger expectation: a stray build trigger with no Prioritized path
 	// fails the test.
 	controller := newTestController(t, ctrl, store, br, nil, nil)
 
@@ -433,18 +433,15 @@ func TestController_Process_SpeculationTreeStorageFailure(t *testing.T) {
 	require.Error(t, err)
 }
 
-// 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) {
+// TestController_Process_TerminalShortCircuit: a terminal batch must
+// short-circuit before touching the speculation tree or build stores: the
+// build controller acks without triggering a build in the build system and without
+// publishing anything. The cancel flow only writes its terminal state after
+// every build has quiesced, and other terminal transitions leave stragglers
+// to run out, so there is nothing left for this controller to enact.
+func TestController_Process_TerminalShortCircuit(t *testing.T) {
 	for _, state := range []entity.BatchState{
 		entity.BatchStateCancelled,
-		entity.BatchStateCancelling,
 		entity.BatchStateSucceeded,
 		entity.BatchStateFailed,
 	} {
@@ -455,14 +452,14 @@ func TestController_Process_HaltedShortCircuit(t *testing.T) {
 			batch.State = state
 			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 tree/build/path-build store expectations: a terminal batch
+			// must never reach the speculation tree or build stores.
 
-			// No Trigger expectation: a stray CI trigger on a halted batch
+			// No Trigger expectation: a stray build trigger on a terminal batch
 			// fails the test.
 			br := buildrunnermock.NewMockBuildRunner(ctrl)
 
-			// Sentinel publish error: the halted path must not publish. If it
+			// Sentinel publish error: the terminal 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"), nil)
 
@@ -471,6 +468,62 @@ func TestController_Process_HaltedShortCircuit(t *testing.T) {
 	}
 }
 
+// TestController_Process_CancellingBatchCancelsInFlightNeverTriggers: a
+// batch in BatchStateCancelling is being torn down batch-wide, so the loop
+// cancels every path's in-flight build regardless of the path's recorded
+// status — including a path still recorded as Building, which the tree
+// sweep may not have marked Cancelling yet — while never triggering a build.
+// Passed paths are skipped without any storage read (their builds are
+// terminal by construction), and a path with no build yet is a tolerated
+// no-op (the Prioritized path here gets no Trigger and no Create).
+func TestController_Process_CancellingBatchCancelsInFlightNeverTriggers(t *testing.T) {
+	ctrl := gomock.NewController(t)
+
+	batch := testBatch()
+	batch.State = entity.BatchStateCancelling
+	cancellingID := "test-queue/batch/1/path/0"
+	buildingID := "test-queue/batch/1/path/1"
+	prioritizedID := "test-queue/batch/1/path/2"
+	passedID := "test-queue/batch/1/path/3"
+	tree := entity.SpeculationTree{
+		BatchID: batch.ID,
+		Version: 2,
+		Paths: []entity.SpeculationPathInfo{
+			{ID: cancellingID, Path: entity.SpeculationPath{Head: batch.ID}, Status: entity.SpeculationPathStatusCancelling},
+			{ID: buildingID, Path: entity.SpeculationPath{Base: []string{"test-queue/batch/0"}, Head: batch.ID}, Status: entity.SpeculationPathStatusBuilding, BuildID: "runner-2"},
+			{ID: prioritizedID, Path: entity.SpeculationPath{Head: batch.ID}, Status: entity.SpeculationPathStatusPrioritized},
+			{ID: passedID, Path: entity.SpeculationPath{Head: batch.ID}, Status: entity.SpeculationPathStatusPassed, BuildID: "runner-4"},
+		},
+	}
+
+	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)
+	// The Cancelling and Building paths resolve to live builds and are both
+	// cancelled; the Prioritized path has no mapping (nothing to stop); the
+	// Passed path is never even looked up.
+	pathBuildStore.EXPECT().Get(gomock.Any(), cancellingID).Return(entity.SpeculationPathBuild{PathID: cancellingID, BuildID: "runner-1"}, nil)
+	buildStore.EXPECT().Get(gomock.Any(), "runner-1").Return(entity.Build{
+		ID: "runner-1", BatchID: batch.ID, SpeculationPathID: cancellingID, Status: entity.BuildStatusRunning,
+	}, nil)
+	pathBuildStore.EXPECT().Get(gomock.Any(), buildingID).Return(entity.SpeculationPathBuild{PathID: buildingID, BuildID: "runner-2"}, nil)
+	buildStore.EXPECT().Get(gomock.Any(), "runner-2").Return(entity.Build{
+		ID: "runner-2", BatchID: batch.ID, SpeculationPathID: buildingID, Status: entity.BuildStatusRunning,
+	}, nil)
+	pathBuildStore.EXPECT().Get(gomock.Any(), prioritizedID).Return(entity.SpeculationPathBuild{}, storage.ErrNotFound)
+
+	br := buildrunnermock.NewMockBuildRunner(ctrl)
+	br.EXPECT().Cancel(gomock.Any(), entity.BuildID{ID: "runner-1"}).Return(nil)
+	br.EXPECT().Cancel(gomock.Any(), entity.BuildID{ID: "runner-2"}).Return(nil)
+	// No Trigger expectation: a stray build trigger fails the test.
+
+	var published []string
+	controller := newTestController(t, ctrl, store, br, nil, &published)
+
+	require.NoError(t, runProcess(t, ctrl, controller, batch))
+	assert.Equal(t, []string{"runner-1", "runner-2"}, published, "each enacted cancel must republish buildsignal so the poll loop observes the stop")
+}
+
 // 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

submitqueue/orchestrator/controller/buildsignal/buildsignal.go

@@ -136,7 +136,9 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r
 	)
 
 	// Load the batch first: it gives us the queue (needed to build the right
-	// BuildRunner) and lets us short-circuit halted batches before polling.
+	// BuildRunner) and the state that decides, after the poll, whether the
+	// result is persisted and propagated (non-terminal batches, including
+	// Cancelling) or dropped (terminal batches).
 	batch, err := c.store.GetBatchStore().Get(ctx, build.BatchID)
 	if err != nil {
 		metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1)
@@ -155,14 +157,16 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r
 		return fmt.Errorf("failed to get status for build %s: %w", buildID.ID, err)
 	}
 
-	// Short-circuit if the batch is already halted (terminal OR cancelling).
-	// Speculate is already idempotent on terminal, but skipping the publish
-	// avoids noise. For Cancelling batches the cancel controller owns the
-	// terminal write and the downstream fan-out, so further pipeline work
-	// would race against it; silent ack is the only safe action.
-	if entity.IsBatchStateHalted(batch.State) {
-		metrics.NamedCounter(c.metricsScope, opName, "skipped_halted", 1)
-		c.logger.Infow("skipping buildsignal publish for halted batch",
+	// Short-circuit only once the batch is terminal: its outcome is settled,
+	// so recording further status or waking speculate would just be noise.
+	// A Cancelling batch, by contrast, NEEDS this loop to keep running — the
+	// cancel flow parks the batch in Cancelling until every build has been
+	// observed terminal, and it is precisely these polls that record the
+	// wind-down and re-publish speculate so the cancel sweep re-evaluates
+	// and eventually performs the terminal write.
+	if batch.State.IsTerminal() {
+		metrics.NamedCounter(c.metricsScope, opName, "skipped_terminal", 1)
+		c.logger.Infow("skipping buildsignal publish for terminal batch",
 			"batch_id", batch.ID,
 			"state", string(batch.State),
 		)