From 5732f0fb12928f76bb24cf48effe044d9d09204b Mon Sep 17 00:00:00 2001 From: Preetam Dwivedi Date: Wed, 12 Aug 2026 16:00:30 -0700 Subject: [PATCH 1/3] feat(speculate): hold speculating until the batch can be sent to merge MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## 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 --- submitqueue/entity/request_log.go | 35 +++++---- .../controller/speculate/finalize.go | 68 +++++++++--------- .../controller/speculate/run_test.go | 71 ++++++++++++++++--- 3 files changed, 117 insertions(+), 57 deletions(-) diff --git a/submitqueue/entity/request_log.go b/submitqueue/entity/request_log.go index 3206dfb1..aefb841e 100644 --- a/submitqueue/entity/request_log.go +++ b/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 diff --git a/submitqueue/orchestrator/controller/speculate/finalize.go b/submitqueue/orchestrator/controller/speculate/finalize.go index 2f105317..a74f81f2 100644 --- a/submitqueue/orchestrator/controller/speculate/finalize.go +++ b/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,31 +130,17 @@ 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, @@ -161,20 +148,23 @@ func (c *Controller) reportSpeculation( 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) diff --git a/submitqueue/orchestrator/controller/speculate/run_test.go b/submitqueue/orchestrator/controller/speculate/run_test.go index 914d031a..ca88992a 100644 --- a/submitqueue/orchestrator/controller/speculate/run_test.go +++ b/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) +} From 7a4444323eaed9def0b244d55d21cca8ccb397b8 Mon Sep 17 00:00:00 2001 From: Preetam Dwivedi Date: Wed, 12 Aug 2026 19:35:19 -0700 Subject: [PATCH 2/3] test(speculate): cover speculation across an unresolved dependency MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## 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 `/batch/`, 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 bcea46ec # Last commands done (2 commands done): # pick e4076129 # feat(speculate): hold speculating until the batch can be sent to merge # pick 14839e58 # 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 42d1cb72 # Last commands done (2 commands done): # pick cfaa1786 # feat(speculate): hold speculating until the batch can be sent to merge # pick 0f79d0ba # 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 # --- .../submitqueue/gateway/server/queues.yaml | 4 ++ .../controller/speculate/run_test.go | 48 ++++++++++++++++ test/e2e/submitqueue/harness_test.go | 40 +++++++++++++ test/e2e/submitqueue/suite_test.go | 56 +++++++++++++++++++ 4 files changed, 148 insertions(+) diff --git a/service/submitqueue/gateway/server/queues.yaml b/service/submitqueue/gateway/server/queues.yaml index cc5bbd3e..a331805f 100644 --- a/service/submitqueue/gateway/server/queues.yaml +++ b/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 diff --git a/submitqueue/orchestrator/controller/speculate/run_test.go b/submitqueue/orchestrator/controller/speculate/run_test.go index ca88992a..1863b820 100644 --- a/submitqueue/orchestrator/controller/speculate/run_test.go +++ b/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. diff --git a/test/e2e/submitqueue/harness_test.go b/test/e2e/submitqueue/harness_test.go index 8475a2e6..c3f229e4 100644 --- a/test/e2e/submitqueue/harness_test.go +++ b/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 diff --git a/test/e2e/submitqueue/suite_test.go b/test/e2e/submitqueue/suite_test.go index dfc27479..7523b5a8 100644 --- a/test/e2e/submitqueue/suite_test.go +++ b/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 "/batch/", 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() { From dfc9e1bbe20d5eb5f77815f52676f0f7e51feff1 Mon Sep 17 00:00:00 2001 From: Preetam Dwivedi Date: Thu, 13 Aug 2026 15:00:17 -0700 Subject: [PATCH 3/3] test(e2e): wait for the waiting event, not the speculated status MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary ### Why? `TestDependentBatch_IsWokenByTheMergeAhead` parks the lead batch's merge behind a closed gate, waits for the dependent to reach `speculated`, and only then opens the gate. That ordering encodes the old meaning of `speculated` — a build passed on a path still consistent with how its dependencies are resolving — which a batch could reach while its dependency was still outstanding. Holding `speculating` until the batch can be sent to merge removes that resting point. A dependent blocked on the parked lead now stays `speculating`, and `speculated` arrives only once the lead has merged and the dependent is itself cleared to merge. So the test waits for a status that cannot arrive until it opens the gate, and it does not open the gate until that status arrives. The suite runs to Bazel's timeout. The two changes had not met before: #576 landed on main after this branch was cut, so CI had never run them together. ### What? The observation step waits for the `waiting` event instead of the `speculated` status. It is the same fact the test was reaching for — the dependent has passed its own build and only the lead is outstanding — expressed as the signal that now carries it, and reachable while the lead is still parked. Nothing else moves. The gate still opens next, and the lead and the dependent are still asserted to land, so the dependent's wake-up remains attributable to the fan-out alone. ## Test Plan ✅ `bazel test //test/e2e/submitqueue:go_default_test` — passes in 120s, against a 300s timeout before ✅ `bazel test //submitqueue/... //platform/...` — 69 tests pass --- test/e2e/submitqueue/suite_test.go | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/test/e2e/submitqueue/suite_test.go b/test/e2e/submitqueue/suite_test.go index 7523b5a8..323fadb9 100644 --- a/test/e2e/submitqueue/suite_test.go +++ b/test/e2e/submitqueue/suite_test.go @@ -295,13 +295,13 @@ func (s *E2EIntegrationSuite) TestLand_HappyPath_ReachesLanded() { // 2. Land the lead. It runs to the merge hand-off and parks there. // 3. Land the dependent. The queue's analyzer serializes conservatively, so // its batch depends on the lead's, which is in-flight (Merging counts). -// 4. Observe: wait for the dependent to reach "speculated" — its speculative +// 4. Observe: wait for the dependent to record "waiting" — its speculative // build has already passed, so its own build signals are finished. From // here the only thing that can advance it is the lead merging. // 5. Start: open the gate. The lead merges and fans out. // // The dependent reaching "landed" is therefore attributable to the fan-out -// alone. Against the old code it stays at "speculated" and the suite runs to +// alone. Against the old code it rests at "speculating" and the suite runs to // Bazel's timeout, which is how the harness reports a pipeline that stalled. func (s *E2EIntegrationSuite) TestDependentBatch_IsWokenByTheMergeAhead() { t := s.T() @@ -339,8 +339,10 @@ func (s *E2EIntegrationSuite) TestDependentBatch_IsWokenByTheMergeAhead() { // Its speculative build passes while the lead is still parked, so by the // time the gate opens the dependent has no build signals left to wake it. - s.awaitStatus(dependent, entity.RequestStatusSpeculated) - s.log.Logf("Dependent %s is speculated and waiting only on %s", dependent.sqid, leadBatch) + // That rest is an event, not a status: a batch blocked on a dependency has + // not finished speculating, so it stays "speculating" until it can merge. + s.awaitEvent(dependent, entity.RequestEventWaiting) + s.log.Logf("Dependent %s has passed its build and waits only on %s", dependent.sqid, leadBatch) // Start: the lead merges, and its fan-out is now the only thing that can // move the dependent.