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

## 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.

The second commit adds the end-to-end coverage this had been missing.
Nothing in `test/integration/` touches the speculate pipeline — the
orchestrator integration suite is `TestPingAPI` and nothing else — and
the e2e happy path has no dependencies, so it never speculates across
one. `e2e-respeculate-queue` is registered in the gateway's queue list
and deliberately takes no profile of its own: falling through to the
baseline is what gives it the `all` analyzer, which serializes the queue
so a second request becomes a batch depending on the first.

The new test forces the wait rather than racing it. Batch IDs come from
a per-queue counter as `<queue>/batch/<n>`, so on a fresh queue the
leader is `batch/1`, and the build topic partitions by batch — closing
the consumer gate on that partition before anything is published holds
the leader's build and nothing else. The follower reaches a passed path
while its dependency is still outstanding, reports `waiting`, and is
asserted to still be `speculating`. Releasing the gate fails the leader,
and the follower re-plans and lands with `speculating` and `speculated`
recorded exactly once each.

`invalidated` is deliberately not asserted end to end. A passed path
stops occupying build budget, so by the time the leader fails the
follower has usually funded the other side of the guess as well; it
never loses its last live passed path, which is the state `invalidated`
reports. Forcing that end to end would mean starving the queue's budget,
which cannot be done without also starving the follower's first build. A
unit test covers the case e2e cannot reach deterministically: a
dependency turning terminal in the same run that walks the head resting
on it, so the break is seen by a later generation of the finalize loop
rather than by the read.

A third commit adapts `TestDependentBatch_IsWokenByTheMergeAhead`, which
#576 added to main after this branch was cut. It parks the lead batch's
merge behind a closed gate, waits for the dependent to reach
`speculated`, and only then opens the gate — an ordering that encodes
the old meaning of `speculated`, which a batch could reach while its
dependency was still outstanding. Under this change the dependent rests
at `speculating` instead, so the test waited for a status that could not
arrive until it opened the gate, and did not open the gate until it
arrived. It now waits for the `waiting` event: the same fact it was
reaching for, expressed as the signal that carries it, and reachable
while the lead is still parked. The gate still opens next and both
requests are still asserted to land, so the wake-up remains attributable
to the fan-out alone.

## Test Plan

✅ `bazel test //submitqueue/... //platform/... //service/...` — 73 tests
pass
✅ `bazel test //test/e2e/...` — 3/3 pass, including the new scenario
✅ `make lint`, `make check-gazelle`, `make check-tidy`, `make
check-mocks`

Re-verified after rebasing onto main at `42d1cb72`: `bazel test
//submitqueue/... //platform/...` — 69 tests pass; `bazel test
//test/e2e/submitqueue` — passes in 120s, against the 300s timeout it
hit before the third commit.

The two `reportSpeculation` tests now assert events. New unit coverage:
a merging head reports `speculated` and no wait (the gate's regression
test); `speculated` is published before the `TopicKeyMerge` dispatch;
and the same-run cascade reports `invalidated`.

`TestLand_HappyPath_ReachesLanded` 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

## Stack
1. #585
1. @ #586

Author: behinddwalls

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