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

## Summary

### Why?

A request's trail read `batched → speculating → speculated → speculating → speculated → landing → landed`, and the repeats looked like the pipeline regressing. They were not a reporting glitch: `RequestStatusSpeculated` meant "a build passed on a path still consistent with how its dependencies are resolving", so it was published while the batch was still blocked, and `reportSpeculation` republished `speculating` whenever a dependency later resolved against that path's guess. Each extra pair was one speculative guess that passed and was then invalidated.

That made `speculated` a per-path, provisional fact wearing a status — the exact shape `RequestEvent` exists for. A batch is not done speculating until it can be sent to merge; waiting on dependencies is still speculating.

### What?

Two events join the vocabulary. `waiting` records that a path passed and the batch has nothing of its own left to run; `invalidated` records that a dependency resolved against the guess that path made. Both are occurrence-keyed on the path ID, so a passed path re-observed across runs collapses to one entry.

`waiting` is gated on `outcomeWait` rather than on merely holding a live passed path. A merge is decided on that same predicate — `mergeablePath` implies `livePassedPath` — so an ungated report would claim a wait on every request that merges straight through. `reportSpeculation` moves below `decide` to see the outcome; both it and `decide` only read, so the reorder observes nothing different.

`speculated` stays a status but now means speculation finished, published from `dispatchMerge` once the batch is cleared to merge. It goes ahead of the dispatch because the merge stage publishes `landing` as its first act on receiving one, and both statuses are non-terminal — so a `speculated` sent afterwards could carry the later timestamp and beat `landing` in the summary.

The `hadPassed && !hasPassed` republish of `speculating` is gone. The status never leaves, so there is nothing to republish, and the oscillation goes with it.

One trade-off worth naming: `speculated` is now near-instantaneous, so "is this batch blocked on dependencies?" is answerable from the latest event rather than from the status.

## Test Plan

✅ `bazel test //submitqueue/... //platform/...` — 68 tests pass

The two `reportSpeculation` tests now assert events. New coverage: a merging head reports `speculated` and no wait — the gate's regression test — and `speculated` is published before the merge dispatch.

`test/e2e/submitqueue/suite_test.go` needs no change: `speculating → speculated → landing → landed` still holds as an ordered subsequence, now for a different reason and at a different point in time.

## Issue

Closes https://linear.app/uber/issue/CODEM-443

Author: behinddwalls

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,60 @@ 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"])
 }
+
+// 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)
+}