test(speculate): cover speculation across an unresolved dependency

## Summary

### Why?

Nothing in `test/integration/` touches the speculate pipeline — the orchestrator integration suite is `TestPingAPI` and nothing else — so the only end-to-end coverage was the happy path, which has no dependencies and therefore never speculates across one. The events this stack introduces had unit coverage only.

### What?

`e2e-respeculate-queue` is registered in the gateway's queue list. It 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 e2e 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 then 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.

Two harness helpers come with it: `awaitEvent`, since an event is never a current status and the history is its only witness, and `assertStatusCount`, which is what pins the no-oscillation property the status change is for.

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.

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

## Test Plan

✅ `bazel test //submitqueue/... //platform/... //service/...` — 73 tests pass
✅ `bazel test //test/e2e/...` — 3/3 pass, including the new scenario

# Conflicts:
#	service/submitqueue/gateway/server/queues.yaml
#	test/e2e/submitqueue/harness_test.go

# Please enter the commit message for your changes. Lines starting
# with '#' will be kept; you may remove them yourself if you want to.
# An empty message aborts the commit.
#
# interactive rebase in progress; onto bcea46e
# Last commands done (2 commands done):
#    pick e407612 # feat(speculate): hold speculating until the batch can be sent to merge
#    pick 14839e5 # test(speculate): cover speculation across an unresolved dependency
# No commands remaining.
# You are currently rebasing branch 'preetam/codem-443-speculation-events' on 'bcea46ec'.
#
# Changes to be committed:
#	modified:   service/submitqueue/gateway/server/queues.yaml
#	modified:   submitqueue/orchestrator/controller/speculate/run_test.go
#	modified:   test/e2e/submitqueue/harness_test.go
#	modified:   test/e2e/submitqueue/suite_test.go
#

# Conflicts:
#	service/submitqueue/gateway/server/queues.yaml

# Please enter the commit message for your changes. Lines starting
# with '#' will be kept; you may remove them yourself if you want to.
# An empty message aborts the commit.
#
# interactive rebase in progress; onto 42d1cb7
# Last commands done (2 commands done):
#    pick cfaa178 # feat(speculate): hold speculating until the batch can be sent to merge
#    pick 0f79d0b # test(speculate): cover speculation across an unresolved dependency
# No commands remaining.
# You are currently rebasing branch 'preetam/codem-443-speculation-events' on '42d1cb72'.
#
# Changes to be committed:
#	modified:   service/submitqueue/gateway/server/queues.yaml
#	modified:   submitqueue/orchestrator/controller/speculate/run_test.go
#	modified:   test/e2e/submitqueue/harness_test.go
#	modified:   test/e2e/submitqueue/suite_test.go
#

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/orchestrator/controller/speculate/run_test.go

@@ -1469,6 +1469,54 @@ func TestRun_ReportsInvalidatedWhenPassedPathBreaks(t *testing.T) {
 	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.

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

test/e2e/submitqueue/suite_test.go

@@ -432,6 +432,62 @@ func (s *E2EIntegrationSuite) TestReadAPIs() {
 	assert.Equal(t, secondSummary.Request.LastError, secondEvents[len(secondEvents)-1].LastError)
 }
 
+// TestLand_DependentBatch_StaysSpeculatingAcrossAnUnresolvedDependency covers
+// the oscillation the request log used to report as a regression: a head
+// speculates while a dependency is unresolved, the dependency then fails, and
+// the head re-plans and lands anyway. Throughout, it is speculating exactly
+// once — the trail must never revisit a stage.
+//
+// The wait is forced rather than raced. Batch IDs come from a per-queue counter
+// as "<queue>/batch/<n>", so the leader on a fresh queue is batch/1, and the
+// build topic partitions by batch — closing the gate on that partition before
+// anything is published holds the leader's build and nothing else, so the
+// follower reaches a passed path while its dependency is still outstanding.
+//
+// No invalidated event is asserted. 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 too; it never loses its last live passed path, which is what
+// invalidated reports. The unit tests cover that state directly.
+func (s *E2EIntegrationSuite) TestLand_DependentBatch_StaysSpeculatingAcrossAnUnresolvedDependency() {
+	const queue = "e2e-respeculate-queue"
+	const gateGroup = "orchestrator"
+	leaderBatch := queue + "/batch/1"
+
+	s.closeGate(gateGroup, leaderBatch, "e2e: hold the leader's build so its dependent speculates first")
+	defer s.openGate(gateGroup, leaderBatch)
+
+	leader := s.land(queue, "github://github.example.com/uber/e2e-respeculate/pull/1/1111111111111111111111111111111111111111?sq-fake=build-fail")
+	follower := s.land(queue, "github://github.example.com/uber/e2e-respeculate/pull/2/2222222222222222222222222222222222222222")
+	s.log.Logf("Landed leader=%s (build held) follower=%s", leader.sqid, follower.sqid)
+
+	// The baseline analyzer serializes the queue, so the follower depends on the
+	// leader and speculates on it succeeding. That build passes while the leader
+	// is still held: the follower's own work is done and only the leader is
+	// outstanding, which is the wait.
+	s.awaitEvent(follower, entity.RequestEventWaiting)
+	assert.Equal(s.T(), entity.RequestStatusSpeculating, s.mustStatus(follower),
+		"a head waiting on its dependency has not finished speculating")
+
+	// Release the leader. Its build fails, contradicting the guess the follower
+	// speculated on, and the follower has to reach the trunk another way.
+	s.openGate(gateGroup, leaderBatch)
+	assert.Equal(s.T(), entity.RequestStatusError, s.awaitTerminal(leader),
+		"the leader's build carries a failure marker, so it must not land")
+
+	s.awaitStatus(follower, entity.RequestStatusLanded)
+	s.assertStatusesInOrder(follower,
+		entity.RequestStatusSpeculating,
+		entity.RequestStatusSpeculated,
+		entity.RequestStatusLanding,
+		entity.RequestStatusLanded,
+	)
+
+	// The point of the exercise: one trip through speculation, however many
+	// guesses it took. A second entry renders as the pipeline going backwards.
+	s.assertStatusCount(follower, entity.RequestStatusSpeculating, 1)
+	s.assertStatusCount(follower, entity.RequestStatusSpeculated, 1)
+}
+
 // TestCancelRequest_InvalidSqid verifies the gateway rejects an empty sqid
 // synchronously before publishing anything to the cancel queue.
 func (s *E2EIntegrationSuite) TestCancelRequest_InvalidSqid() {