feat(speculate): hold speculating until the batch can be sent to merge

service/submitqueue/gateway/server/queues.yaml

@@ -23,3 +23,7 @@ queues:
   # pipeline runs against a real repository. See
   # service/submitqueue/demo/provider and doc/howto/PROVIDER-E2E.md.
   - name: demo-queue
+  # Inherits the baseline "all" analyzer, which serializes the queue, so a
+  # second request lands as a batch depending on the first. e2e uses that to
+  # exercise speculation across an unresolved dependency.
+  - name: e2e-respeculate-queue

submitqueue/entity/request_log.go

@@ -54,11 +54,12 @@ const (
 	// RequestStatusBatched indicates that the request has been included in a new batch and will be sent to speculation.
 	RequestStatusBatched RequestStatus = "batched"
 
-	// RequestStatusSpeculating indicates that the batch containing the request has been admitted to speculation: candidate paths are being planned and built.
+	// RequestStatusSpeculating indicates that the batch containing the request is in speculation:
+	// planning, building, or waiting for its dependencies to settle. None of those leaves it able to land.
 	RequestStatusSpeculating RequestStatus = "speculating"
 
-	// RequestStatusSpeculated indicates that the batch containing the request has a build that passed on a path still
-	// consistent with how its dependencies are resolving, and is waiting for those dependencies to settle before it can land.
+	// RequestStatusSpeculated indicates that the batch containing the request has finished speculating:
+	// a build passed on a path whose assumptions all held, and the batch has been cleared to merge.
 	RequestStatusSpeculated RequestStatus = "speculated"
 
 	// RequestStatusLanding indicates that the request is actively being landed (e.g., source control operation is in progress to push the change to the target branch).
@@ -84,17 +85,17 @@ const (
 // RequestEvent is something that happened to a request while it sat at a status,
 // rather than a status of its own.
 //
-// Build progress is what the distinction exists for. A batch funds several
-// speculation paths at once and each is built separately, so a build starting or
-// finishing says nothing about where the request as a whole is — it is still
-// speculating. Were these statuses, one build succeeding while its siblings ran
-// would report the request as finished, and go on reporting it that way until the
-// batch resolved, because nothing else publishes in between.
+// Speculation is what the distinction exists for. A batch funds several paths at
+// once and each is built separately, so a build starting or finishing, or one
+// path passing and later being contradicted, says nothing about where the request
+// as a whole is — it is still speculating. Were these statuses, one build
+// succeeding while its siblings ran would report the request as finished, and go
+// on reporting it that way until the batch resolved.
 //
-// Events are not unique per request: each names one build, and a batch may be
-// built many times as speculation re-plans. They belong in a request's history
-// and are never its current status — which is enforced by the type, since a
-// RequestEvent cannot be assigned to RequestSummary.Status.
+// Events are not unique per request: each names one path or build, and a batch
+// may be re-planned many times. They belong in a request's history and are never
+// its current status — which is enforced by the type, since a RequestEvent cannot
+// be assigned to RequestSummary.Status.
 type RequestEvent string
 
 const (
@@ -107,6 +108,14 @@ const (
 	// RequestEventBuilt indicates that one build verifying one speculation path of the batch containing the request finished successfully.
 	// A build that fails or is cancelled records nothing.
 	RequestEventBuilt RequestEvent = "built"
+
+	// RequestEventWaiting indicates that one speculation path of the batch containing the request passed,
+	// leaving the batch nothing of its own to run and waiting on its dependencies to settle.
+	RequestEventWaiting RequestEvent = "waiting"
+
+	// RequestEventInvalidated indicates that a dependency resolved against the guess made by the passed path
+	// the batch containing the request was waiting on, so that path can no longer carry it.
+	RequestEventInvalidated RequestEvent = "invalidated"
 )
 
 // RequestLogType is what a log entry records: the request reaching a status, or

submitqueue/orchestrator/controller/speculate/finalize.go

@@ -83,11 +83,12 @@ func (c *Controller) finalize(ctx context.Context, snap *snapshot) error {
 				snap.markDirty(batch.ID)
 			}
 
-			if err := c.reportSpeculation(ctx, batch, set, *snap, before, hadPassed); err != nil {
+			decision := decide(batch, set, *snap)
+
+			if err := c.reportSpeculation(ctx, batch, set, *snap, before, hadPassed, decision); err != nil {
 				return err
 			}
 
-			decision := decide(batch, set, *snap)
 			if decision == outcomeWait {
 				stillOpen = append(stillOpen, batch)
 				continue
@@ -129,52 +130,41 @@ func (c *Controller) finalize(ctx context.Context, snap *snapshot) error {
 	return nil
 }
 
-// reportSpeculation tells a head's members how far speculation has got.
+// reportSpeculation records what the fold above did to a head's passed path.
+// Both facts are per-path and the head stays BatchStateSpeculating throughout,
+// so neither is a status and the request log is the only place they show up.
 //
-// Two moments are worth reporting and neither is a batch state — a head is
-// BatchStateSpeculating from admission until its outcome, so the request log is
-// the only place either becomes visible:
+// before comes from passedEntry, not livePassedPath: that predicate and the
+// fold both exclude a contradicted path, so two livePassedPath calls could
+// never see the loss. Only the run that does the breaking sees it at all.
 //
-//   - the head has a live passed path. Its own work is done and what remains is
-//     other batches finishing, a wait that can run for minutes and reads very
-//     differently to still building.
-//   - it just lost the one it had, because a dependency resolved against that
-//     path's guess. The head is back to building, and without this its members
-//     would go on reading as speculated through the whole rebuild.
-//
-// The second is why before is taken from passedEntry rather than livePassedPath:
-// both that predicate and the fold above exclude a contradicted path, so a pair
-// of livePassedPath calls could never see the loss happen. What is compared is
-// "held a passed build" before the fold against "still has one worth waiting on"
-// after it, and only the run that does the breaking sees the difference — every
-// later run finds the entry already cancelled.
-//
-// Both facts are derived from the snapshot rather than stored, so this runs on
-// every pass over an open head and relies on the occurrence to collapse the
-// repeats: a path ID hashes its head along with its assumptions, so it names the
-// batch too, and one passed path re-observed by a hundred runs is a single entry
-// while a different path winning after a re-plan is correctly a new one.
+// Nothing is stored, so this runs on every pass and leans on the occurrence to
+// collapse repeats — a path ID hashes its head with its assumptions, so one
+// passed path re-observed stays one entry while a re-plan's winner is a new one.
 func (c *Controller) reportSpeculation(
 	ctx context.Context,
 	batch entity.Batch,
 	set entity.SpeculationPathSet,
 	snap snapshot,
 	before entity.SpeculationPathEntry,
 	hadPassed bool,
+	decision outcome,
 ) error {
 	after, hasPassed := livePassedPath(set, snap)
 
-	status, path := entity.RequestStatusSpeculated, after
+	// A merge is decided on the same live passed path, so an ungated report
+	// would claim a wait on every head that merges straight through.
+	event, path := entity.RequestEventWaiting, after
 	switch {
-	case hasPassed:
-	case hadPassed:
-		status, path = entity.RequestStatusSpeculating, before
+	case hasPassed && decision == outcomeWait:
+	case hadPassed && !hasPassed:
+		event, path = entity.RequestEventInvalidated, before
 	default:
 		return nil
 	}
 
-	if err := corerequest.PublishBatchLogs(ctx, c.registry, batch.Queue, batch.Contains,
-		status, path.ID, map[string]string{
+	if err := corerequest.PublishBatchEvents(ctx, c.registry, batch.Queue, batch.Contains,
+		event, path.ID, map[string]string{
 			"batch_id": batch.ID,
 			"path_id":  path.ID,
 		},
@@ -183,7 +173,7 @@ func (c *Controller) reportSpeculation(
 		// Attributed to this head, not the trigger: the loop walks the whole
 		// queue, so the batch whose members could not be told is usually not the
 		// one the message named.
-		return c.attributed(fmt.Errorf("failed to publish request logs for batch %s: %w", batch.ID, err),
+		return c.attributed(fmt.Errorf("failed to publish request events for batch %s: %w", batch.ID, err),
 			entity.BatchSubject(batch.ID))
 	}
 	return nil
@@ -369,10 +359,18 @@ func (c *Controller) applyOutcome(ctx context.Context, store storage.Storage, ba
 	return true, nil
 }
 
-// dispatchMerge hands a batch to the merge stage under a stable ID, so both a
-// redelivery and the Merging self-heal dedupe against the request already sent
-// rather than asking Runway to merge the batch twice.
+// dispatchMerge reports speculation finished and hands the batch to the merge
+// stage. The stable ID means a redelivery or the Merging self-heal dedupes
+// against the request already sent instead of merging twice; the status goes
+// first so it cannot be timestamped after the landing the dispatch triggers.
 func (c *Controller) dispatchMerge(ctx context.Context, batch entity.Batch) error {
+	if err := corerequest.PublishBatchLogs(ctx, c.registry, batch.Queue, batch.Contains,
+		entity.RequestStatusSpeculated, batch.ID, map[string]string{"batch_id": batch.ID},
+	); err != nil {
+		metrics.NamedCounter(c.metricsScope, opName, "request_log_errors", 1)
+		return fmt.Errorf("failed to publish request logs for batch %s: %w", batch.ID, err)
+	}
+
 	if err := c.publishBatchID(ctx, topickey.TopicKeyMerge, publish.IntentID(batch.ID, "merge-dispatch"), batch.ID, batch.Queue, batch.Queue); err != nil {
 		metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1)
 		return fmt.Errorf("failed to publish batch %s to merge: %w", batch.ID, err)

submitqueue/orchestrator/controller/speculate/run_test.go

@@ -1406,10 +1406,9 @@ func memberHead() entity.Batch {
 	return h
 }
 
-// A head whose build passed but whose dependencies have not all settled is in
-// the one part of speculation worth naming: its own work is done, and what
-// remains is other batches finishing. Without this its members read as still
-// building for the whole of that wait.
+// A head whose build passed but whose dependencies have not all settled has
+// nothing of its own left to run, a wait that reads very differently to still
+// building.
 func TestRun_ReportsPassedPathWhileWaiting(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	passed := pathOver(entity.DependencyAssumptionSucceeds, entity.DependencyAssumptionSucceeds)
@@ -1434,15 +1433,15 @@ func TestRun_ReportsPassedPathWhileWaiting(t *testing.T) {
 
 	require.Len(t, h.logs, 1)
 	assert.Equal(t, "q/1", h.logs[0].RequestID)
-	assert.Equal(t, entity.RequestStatusSpeculated, h.logs[0].Status)
+	assert.Equal(t, entity.RequestLogTypeEvent, h.logs[0].Type)
+	assert.Equal(t, entity.RequestEventWaiting, h.logs[0].Event)
 	assert.Equal(t, head, h.logs[0].Metadata["batch_id"])
 	assert.Equal(t, entry.ID, h.logs[0].Metadata["path_id"])
 }
 
 // The other half: a dependency that resolves against a passed path's guess
-// takes the head's waiting room away and puts it back to building. Reporting
-// that is what stops the members reading as speculated through the rebuild.
-func TestRun_ReportsBackToSpeculatingWhenPassedPathBreaks(t *testing.T) {
+// takes the head's waiting room away.
+func TestRun_ReportsInvalidatedWhenPassedPathBreaks(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	passed := pathOver(entity.DependencyAssumptionSucceeds, entity.DependencyAssumptionFails)
 	spec := &scriptedSpeculator{}
@@ -1465,6 +1464,108 @@ func TestRun_ReportsBackToSpeculatingWhenPassedPathBreaks(t *testing.T) {
 	require.NoError(t, h.run(head))
 
 	require.Len(t, h.logs, 1)
-	assert.Equal(t, entity.RequestStatusSpeculating, h.logs[0].Status)
+	assert.Equal(t, entity.RequestLogTypeEvent, h.logs[0].Type)
+	assert.Equal(t, entity.RequestEventInvalidated, h.logs[0].Event)
 	assert.Equal(t, entry.ID, h.logs[0].Metadata["path_id"])
 }
+
+// The e2e shape: the dependency turns terminal in the same run that walks the
+// head resting on it, so the break is seen by a later generation of the loop
+// rather than by the read.
+func TestRun_ReportsInvalidatedWhenTheDependencyFailsInTheSameRun(t *testing.T) {
+	ctrl := gomock.NewController(t)
+
+	leader := entity.Batch{ID: dep1, Queue: "q", State: entity.BatchStateSpeculating, Version: 1}
+	followerBatch := entity.Batch{
+		ID: head, Queue: "q", Contains: []string{"q/1"},
+		State: entity.BatchStateSpeculating, Dependencies: []string{dep1}, Version: 1,
+	}
+
+	h := newRunHarness(t, ctrl, &scriptedSpeculator{}, []entity.Batch{leader, followerBatch})
+	h.noBuildsDispatched()
+
+	// The leader has nothing left that can pass, so this run fails it.
+	h.pathSets.EXPECT().Get(gomock.Any(), dep1).Return(entity.SpeculationPathSet{
+		Head:    dep1,
+		Paths:   []entity.SpeculationPathEntry{entryFor(entity.SpeculationPath{Head: dep1}, entity.SpeculationPathStatusFailed)},
+		Version: 1,
+	}, nil).AnyTimes()
+	h.batches.EXPECT().
+		Update(gomock.Any(), updateTo{id: dep1, state: entity.BatchStateFailed}, int32(1), int32(2)).Return(nil)
+
+	// The follower passed on the guess that the leader would succeed.
+	passed := entity.SpeculationPath{
+		Head:         head,
+		Dependencies: []entity.PathDependency{{Batch: dep1, Assumption: entity.DependencyAssumptionSucceeds}},
+	}
+	entry := entryFor(passed, entity.SpeculationPathStatusPassed)
+	h.pathSets.EXPECT().Get(gomock.Any(), head).Return(entity.SpeculationPathSet{
+		Head:    head,
+		Paths:   []entity.SpeculationPathEntry{entry},
+		Version: 1,
+	}, nil).AnyTimes()
+	h.pathSets.EXPECT().Update(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
+
+	require.NoError(t, h.run(dep1))
+
+	var got []entity.RequestEvent
+	for _, entry := range h.logs {
+		if entry.Type == entity.RequestLogTypeEvent {
+			got = append(got, entry.Event)
+		}
+	}
+	assert.Contains(t, got, entity.RequestEventInvalidated)
+}
+
+// A merge is decided on the same live passed path a wait would be reported
+// from, so without the gate every landed request would carry a wait it never
+// had.
+func TestRun_MergingHeadReportsSpeculatedAndNoWait(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	passed := pathOver(entity.DependencyAssumptionSucceeds, entity.DependencyAssumptionSucceeds)
+
+	h := newRunHarness(t, ctrl, &scriptedSpeculator{}, []entity.Batch{memberHead()})
+	h.noBuildsDispatched()
+	h.batches.EXPECT().Get(gomock.Any(), dep1).Return(entity.Batch{ID: dep1, State: entity.BatchStateSucceeded}, nil)
+	h.batches.EXPECT().Get(gomock.Any(), dep2).Return(entity.Batch{ID: dep2, State: entity.BatchStateSucceeded}, nil)
+	h.pathSets.EXPECT().Get(gomock.Any(), head).Return(entity.SpeculationPathSet{
+		Head:    head,
+		Paths:   []entity.SpeculationPathEntry{entryFor(passed, entity.SpeculationPathStatusPassed)},
+		Version: 1,
+	}, nil).AnyTimes()
+	h.batches.EXPECT().
+		Update(gomock.Any(), updateTo{id: head, state: entity.BatchStateMerging}, int32(1), int32(2)).Return(nil)
+
+	require.NoError(t, h.run(head))
+
+	require.Len(t, h.logs, 1)
+	assert.Equal(t, entity.RequestLogTypeStatus, h.logs[0].Type)
+	assert.Equal(t, entity.RequestStatusSpeculated, h.logs[0].Status)
+	assert.Equal(t, head, h.logs[0].Metadata["batch_id"])
+}
+
+// The merge stage publishes landing as its first act on the dispatch. Both
+// statuses are non-terminal, so the summary is decided on timestamp alone and
+// a speculated sent afterwards would beat the landing it precedes.
+func TestRun_SpeculatedIsReportedBeforeTheMergeDispatch(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	passed := pathOver(entity.DependencyAssumptionSucceeds, entity.DependencyAssumptionSucceeds)
+
+	h := newRunHarness(t, ctrl, &scriptedSpeculator{}, []entity.Batch{memberHead()})
+	h.failPublishTo("submitqueue-merge")
+	h.noBuildsDispatched()
+	h.batches.EXPECT().Get(gomock.Any(), dep1).Return(entity.Batch{ID: dep1, State: entity.BatchStateSucceeded}, nil)
+	h.batches.EXPECT().Get(gomock.Any(), dep2).Return(entity.Batch{ID: dep2, State: entity.BatchStateSucceeded}, nil)
+	h.pathSets.EXPECT().Get(gomock.Any(), head).Return(entity.SpeculationPathSet{
+		Head:    head,
+		Paths:   []entity.SpeculationPathEntry{entryFor(passed, entity.SpeculationPathStatusPassed)},
+		Version: 1,
+	}, nil).AnyTimes()
+	h.batches.EXPECT().
+		Update(gomock.Any(), updateTo{id: head, state: entity.BatchStateMerging}, int32(1), int32(2)).Return(nil)
+
+	require.Error(t, h.run(head))
+
+	require.Len(t, h.logs, 1)
+	assert.Equal(t, entity.RequestStatusSpeculated, h.logs[0].Status)
+}

test/e2e/submitqueue/harness_test.go

@@ -207,6 +207,46 @@ func (s *E2EIntegrationSuite) awaitBatchID(req request) string {
 	return batchID
 }
 
+// mustStatus reads the current status and fails the test if it is unreadable.
+func (s *E2EIntegrationSuite) mustStatus(req request) entity.RequestStatus {
+	t := s.T()
+	got, err := s.currentStatus(req)
+	require.NoError(t, err, "GetRequestSummaryByID failed for %s", req.sqid)
+	return got
+}
+
+// awaitEvent polls GetRequestHistoryByID until want appears in the request's
+// event timeline. Unlike a status, an event is never the current position, so
+// there is nothing to poll on the summary — the history is the only witness.
+func (s *E2EIntegrationSuite) awaitEvent(req request, want entity.RequestEvent) {
+	pollUntil(persistPollInterval, func() bool {
+		got := s.eventTimeline(req)
+		s.log.Logf("events(%s) = %v (want %q)", req.sqid, got, want)
+		for _, e := range got {
+			if e == want {
+				return true
+			}
+		}
+		return false
+	})
+}
+
+// assertStatusCount asserts how many times a status appears in the timeline.
+// A status that recurs is not merely noisy: the client renders each entry as a
+// fresh step, so a stage revisited reads as the pipeline going backwards.
+func (s *E2EIntegrationSuite) assertStatusCount(req request, status entity.RequestStatus, want int) {
+	t := s.T()
+	got := s.timeline(req)
+	seen := 0
+	for _, st := range got {
+		if st == status {
+			seen++
+		}
+	}
+	assert.Equalf(t, want, seen,
+		"GetRequestHistoryByID for %s should record %q %d time(s); got %v", req.sqid, status, want, got)
+}
+
 // closeGate closes the consumer gate for the consumer group, scoped to one
 // partition (the queue name for pipeline topics). The gate must be closed
 // before the message that must be caught is published — that makes the stop