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/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..1863b820 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,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) +} 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..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. @@ -432,6 +434,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() {