diff --git a/test/e2e/submitqueue/suite_test.go b/test/e2e/submitqueue/suite_test.go index ca4b9b83..623ad965 100644 --- a/test/e2e/submitqueue/suite_test.go +++ b/test/e2e/submitqueue/suite_test.go @@ -465,6 +465,102 @@ func (s *E2EIntegrationSuite) TestDependentBatch_NoBypassWhenCoverageIsIncomplet s.awaitStatus(trigger, entity.RequestStatusLanded) } +// TestDependentBatch_IsWokenByTheLandAhead proves that a batch waiting on +// another is woken when that one lands — an edge the queue was silently +// dropping. +// +// A landed batch fans out to speculate so its dependents can re-plan. That +// message used to reuse the bare batch ID, which the batch controller had +// already published to the same topic and partition when the batch was +// created. The queue deduplicates against rows it has not collected yet, +// consumed ones included, so the wake-up was reported as a success, stored +// nothing, and never arrived. +// +// Ordinarily something else re-plans the queue soon enough to hide that. This +// test removes every other source of a wake-up, as stop → observe → start: +// +// 1. Stop: close the gate for runway-merge on this queue, before landing, so +// the lead batch cannot complete its land. +// 2. Land the lead. It runs to the land 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 (Landing counts). +// 4. Fund only the path that assumes the lead succeeds, and hold every real +// build for the dependent so complete coverage cannot bypass the lead. +// 5. Observe: wait for the dependent to record "waiting". From here the only +// event that can make its funded path landable is the lead landing. +// 6. Start: open the gate. The lead lands and fans out. +// +// The dependent reaching "landed" is therefore attributable to the fan-out +// 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_IsWokenByTheLandAhead() { + t := s.T() + + const queue = "e2e-chain-queue" + const runwayGateGroup = "runway-merge" + const orchestratorGateGroup = "orchestrator" + gateTopic := runwaymq.TopicKeyMerge.String() + + s.closeGate(runwayGateGroup, queue, "e2e: hold the lead land while the dependent waits") + // Reopen even if an assertion below fails, so teardown does not stop the + // stack with a delivery still parked. Opening twice is a no-op. + defer s.openGate(runwayGateGroup, queue) + + lead := s.land(queue, "github://github.example.com/uber/e2e-chain/pull/1/abcdef0123456789abcdef0123456789abcdef01") + s.log.Logf("Landed lead request %s; awaiting its land to park", lead.sqid) + + // The land request is keyed by batch, so name the batch to prove the + // parked delivery is this request's land and not some other. + leadBatch := s.awaitBatchID(lead) + parked := s.awaitParked(runwayGateGroup, gateTopic, leadBatch) + assert.Equal(t, queue, parked.PartitionKey, "land request should be partitioned by queue") + + // The lead is provably stopped mid-land. A request landed now serializes + // behind it. + dependent := s.land(queue, "github://github.example.com/uber/e2e-chain/pull/2/1234567890abcdef1234567890abcdef12345678") + dependentBatch := s.awaitBatchID(dependent) + require.NotEqual(t, leadBatch, dependentBatch, "the two requests must be carried by different batches") + s.closeGate(orchestratorGateGroup, dependentBatch, "e2e: hold dependent builds so only the seeded path exists") + defer s.openGate(orchestratorGateGroup, dependentBatch) + + leadState, err := s.appStorage.For(queue) + require.NoError(t, err) + got, err := leadState.GetBatchStore().Get(s.ctx, dependentBatch) + require.NoError(t, err, "failed to read the dependent batch") + require.Contains(t, got.Dependencies, leadBatch, + "batch %s must depend on the in-flight %s for this test to exercise anything", dependentBatch, leadBatch) + + // Leave the real builds parked and fund only the world where the lead lands. + // A later request wakes the queue so it admits the stranded batch and + // reports the wait without adding the missing failure-assumption path. + s.strandInCreated(queue, dependentBatch) + s.seedPassedPath(queue, entity.SpeculationPath{ + Head: dependentBatch, + Dependencies: []entity.PathDependency{ + {Batch: leadBatch, Assumption: entity.DependencyAssumptionSucceeds}, + }, + }) + trigger := s.land(queue, "github://github.example.com/uber/e2e-chain/pull/3/fedcba9876543210fedcba9876543210fedcba98") + s.awaitBatchID(trigger) + + 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 lands, and its fan-out is now the only thing that can + // move the dependent. + s.openGate(runwayGateGroup, queue) + s.awaitUnparked(runwayGateGroup, gateTopic, leadBatch) + + s.awaitStatus(lead, entity.RequestStatusLanded) + s.awaitStatus(dependent, entity.RequestStatusLanded) + + assert.Equal(t, entity.RequestStateLanded, s.terminalState(dependent), + "the dependent must land once the batch it waited on landed") + + s.openGate(orchestratorGateGroup, dependentBatch) + s.awaitStatus(trigger, entity.RequestStatusLanded) +} + // TestReadAPIs validates all five request read endpoints against receipts // created through the public Land API. func (s *E2EIntegrationSuite) TestReadAPIs() {