test(e2e): cover dependent wake-up after land

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() {