@@ -465,6 +465,102 @@ func (s *E2EIntegrationSuite) TestDependentBatch_NoBypassWhenCoverageIsIncomplet
465465 s .awaitStatus (trigger , entity .RequestStatusLanded )
466466}
467467
468+ // TestDependentBatch_IsWokenByTheLandAhead proves that a batch waiting on
469+ // another is woken when that one lands — an edge the queue was silently
470+ // dropping.
471+ //
472+ // A landed batch fans out to speculate so its dependents can re-plan. That
473+ // message used to reuse the bare batch ID, which the batch controller had
474+ // already published to the same topic and partition when the batch was
475+ // created. The queue deduplicates against rows it has not collected yet,
476+ // consumed ones included, so the wake-up was reported as a success, stored
477+ // nothing, and never arrived.
478+ //
479+ // Ordinarily something else re-plans the queue soon enough to hide that. This
480+ // test removes every other source of a wake-up, as stop → observe → start:
481+ //
482+ // 1. Stop: close the gate for runway-merge on this queue, before landing, so
483+ // the lead batch cannot complete its land.
484+ // 2. Land the lead. It runs to the land hand-off and parks there.
485+ // 3. Land the dependent. The queue's analyzer serializes conservatively, so
486+ // its batch depends on the lead's, which is in-flight (Landing counts).
487+ // 4. Fund only the path that assumes the lead succeeds, and hold every real
488+ // build for the dependent so complete coverage cannot bypass the lead.
489+ // 5. Observe: wait for the dependent to record "waiting". From here the only
490+ // event that can make its funded path landable is the lead landing.
491+ // 6. Start: open the gate. The lead lands and fans out.
492+ //
493+ // The dependent reaching "landed" is therefore attributable to the fan-out
494+ // alone. Against the old code it rests at "speculating" and the suite runs to
495+ // Bazel's timeout, which is how the harness reports a pipeline that stalled.
496+ func (s * E2EIntegrationSuite ) TestDependentBatch_IsWokenByTheLandAhead () {
497+ t := s .T ()
498+
499+ const queue = "e2e-chain-queue"
500+ const runwayGateGroup = "runway-merge"
501+ const orchestratorGateGroup = "orchestrator"
502+ gateTopic := runwaymq .TopicKeyMerge .String ()
503+
504+ s .closeGate (queue , runwayGateGroup , queue , "e2e: hold the lead land while the dependent waits" )
505+ // Reopen even if an assertion below fails, so teardown does not stop the
506+ // stack with a delivery still parked. Opening twice is a no-op.
507+ defer s .openGate (queue , runwayGateGroup , queue )
508+
509+ lead := s .land (queue , "github://github.example.com/uber/e2e-chain/pull/1/abcdef0123456789abcdef0123456789abcdef01" )
510+ s .log .Logf ("Landed lead request %s; awaiting its land to park" , lead .sqid )
511+
512+ // The land request is keyed by batch, so name the batch to prove the
513+ // parked delivery is this request's land and not some other.
514+ leadBatch := s .awaitBatchID (lead )
515+ parked := s .awaitParked (runwayGateGroup , gateTopic , leadBatch )
516+ assert .Equal (t , queue , parked .PartitionKey , "land request should be partitioned by queue" )
517+
518+ // The lead is provably stopped mid-land. A request landed now serializes
519+ // behind it.
520+ dependent := s .land (queue , "github://github.example.com/uber/e2e-chain/pull/2/1234567890abcdef1234567890abcdef12345678" )
521+ dependentBatch := s .awaitBatchID (dependent )
522+ require .NotEqual (t , leadBatch , dependentBatch , "the two requests must be carried by different batches" )
523+ s .closeGate (queue , orchestratorGateGroup , dependentBatch , "e2e: hold dependent builds so only the seeded path exists" )
524+ defer s .openGate (queue , orchestratorGateGroup , dependentBatch )
525+
526+ leadState , err := s .appStorage .For (queue )
527+ require .NoError (t , err )
528+ got , err := leadState .GetBatchStore ().Get (s .ctx , dependentBatch )
529+ require .NoError (t , err , "failed to read the dependent batch" )
530+ require .Contains (t , got .Dependencies , leadBatch ,
531+ "batch %s must depend on the in-flight %s for this test to exercise anything" , dependentBatch , leadBatch )
532+
533+ // Leave the real builds parked and fund only the world where the lead lands.
534+ // A later request wakes the queue so it admits the stranded batch and
535+ // reports the wait without adding the missing failure-assumption path.
536+ s .strandInCreated (queue , dependentBatch )
537+ s .seedPassedPath (queue , entity.SpeculationPath {
538+ Head : dependentBatch ,
539+ Dependencies : []entity.PathDependency {
540+ {Batch : leadBatch , Assumption : entity .DependencyAssumptionSucceeds },
541+ },
542+ })
543+ trigger := s .land (queue , "github://github.example.com/uber/e2e-chain/pull/3/fedcba9876543210fedcba9876543210fedcba98" )
544+ s .awaitBatchID (trigger )
545+
546+ s .awaitEvent (dependent , entity .RequestEventWaiting )
547+ s .log .Logf ("Dependent %s has passed its build and waits only on %s" , dependent .sqid , leadBatch )
548+
549+ // Start: the lead lands, and its fan-out is now the only thing that can
550+ // move the dependent.
551+ s .openGate (queue , runwayGateGroup , queue )
552+ s .awaitUnparked (runwayGateGroup , gateTopic , leadBatch )
553+
554+ s .awaitStatus (lead , entity .RequestStatusLanded )
555+ s .awaitStatus (dependent , entity .RequestStatusLanded )
556+
557+ assert .Equal (t , entity .RequestStateLanded , s .terminalState (dependent ),
558+ "the dependent must land once the batch it waited on landed" )
559+
560+ s .openGate (queue , orchestratorGateGroup , dependentBatch )
561+ s .awaitStatus (trigger , entity .RequestStatusLanded )
562+ }
563+
468564// TestReadAPIs validates all five request read endpoints against receipts
469565// created through the public Land API.
470566func (s * E2EIntegrationSuite ) TestReadAPIs () {
0 commit comments