test(e2e): prove a dependent batch is woken by the merge ahead of it (#576)

## Summary

### Why?

The parent change fixes a dropped wake-up: a merged batch fans out to
speculate so its dependents can re-plan, and that message used to reuse
the bare batch ID the batch controller had already published to the same
topic and partition at creation. The queue deduplicates against rows it
has not collected yet, consumed ones included, so the fan-out was
reported as a success, stored nothing, and never arrived.

That fix shipped with unit coverage on the message ID and integration
coverage on the queue semantics, but nothing exercised the path the bug
actually broke. It is also a path that hides easily: any other event
re-plans the queue and moves the dependent along, so a naive two-request
test passes with or without the fix.

### What?

A new e2e case isolates the fan-out as the only possible wake-up,
following the stop → observe → start shape
`TestCancel_CaughtPreBatch_NeverLands` already uses:

1. Close the `runway-merge` gate for the queue before landing, so the
lead batch cannot complete its merge.
2. Land the lead; wait for its merge to park, keyed by the lead's batch
ID.
3. Land the dependent. Its batch serializes behind the lead's, which is
in-flight (`Merging` is a dependency state).
4. Wait for the dependent to reach `speculated` — its speculative build
has already passed, so its own build signals are finished and nothing
else will wake it.
5. Open the gate. The lead merges and fans out.

The dependent reaching `landed` is then attributable to the fan-out
alone.

Supporting changes: `e2e-chain-queue` is registered in `queues.yaml`. It
is deliberately absent from the orchestrator's per-queue profiles so it
falls through to the baseline profile and its `all` conflict analyzer,
which serializes every new batch behind every in-flight one — that is
what builds the chain. A new `awaitBatchID` harness helper resolves a
request's batch ID from the operating store, since merge messages are
keyed by batch rather than by the sqid a test holds.

## Test Plan

- ✅ `bazel test //test/e2e/...` — 3 suites, including the new case
(~33s)
- ✅ `make lint-license`, `make lint-message-id`, `make lint-queue-shard`
- **Confirmed the test fails against the unfixed code.** Reverting the
parent's `mergesignal` message ID to the bare batch ID leaves the
dependent stuck at `speculated` and the suite runs to Bazel's timeout
(`TIMEOUT in 240.3s` with `--test_timeout=240`); with the fix it passes
in 33s.

A stalled pipeline surfaces as a test timeout rather than an assertion
failure, which is how this harness reports non-convergence — `pollUntil`
has no deadline of its own by design, so Bazel's timeout is the only
one.


## Stack
1. #574
1. @ #576

Author: behinddwalls

service/submitqueue/gateway/server/queues.yaml

@@ -7,6 +7,11 @@ queues:
   - name: test-queue
   - name: e2e-test-queue
   - name: e2e-cancel-queue
+  # Not listed in the orchestrator's per-queue profiles, so it falls through to
+  # the baseline profile and its "all" conflict analyzer: every new batch
+  # serializes behind every in-flight one. That is what lets e2e build a
+  # dependency chain and exercise how a dependent is woken.
+  - name: e2e-chain-queue
   # Routes to an analyzer that always errors (conflictfake.FailAlways) so e2e can
   # exercise the conflict-analysis error path. See newQueueRegistry in the
   # orchestrator example server.

test/e2e/submitqueue/harness_test.go

@@ -183,6 +183,30 @@ func (s *E2EIntegrationSuite) assertStatusesNever(req request, banned ...entity.
 	}
 }
 
+// awaitBatchID polls the operating store until the request has been claimed by
+// a batch and returns that batch's ID.
+//
+// Messages about a batch are keyed by the batch ID, not the sqid the test
+// holds, so a test that wants to name one has to resolve it. Polling because
+// the claim happens asynchronously, several stages after Land returns.
+func (s *E2EIntegrationSuite) awaitBatchID(req request) string {
+	t := s.T()
+	store, err := s.appStorage.For(req.queue)
+	require.NoError(t, err, "failed to resolve operating store for queue %s", req.queue)
+
+	var batchID string
+	pollUntil(persistPollInterval, func() bool {
+		associations, err := store.GetRequestBatchStore().GetByRequestID(s.ctx, req.sqid)
+		if err != nil || len(associations) == 0 {
+			return false
+		}
+		batchID = associations[0].BatchID
+		return true
+	})
+	s.log.Logf("Request %s is carried by batch %s", req.sqid, batchID)
+	return batchID
+}
+
 // 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

test/e2e/submitqueue/suite_test.go

@@ -276,6 +276,84 @@ func (s *E2EIntegrationSuite) TestLand_HappyPath_ReachesLanded() {
 		"operating store should show request %s in terminal state landed", req.sqid)
 }
 
+// TestDependentBatch_IsWokenByTheMergeAhead proves that a batch waiting on
+// another is woken when that one merges — the edge CODEM-303 was silently
+// dropping.
+//
+// A merged 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 merge.
+//  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
+//     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
+// Bazel's timeout, which is how the harness reports a pipeline that stalled.
+func (s *E2EIntegrationSuite) TestDependentBatch_IsWokenByTheMergeAhead() {
+	t := s.T()
+
+	const queue = "e2e-chain-queue"
+	const gateGroup = "runway-merge"
+	gateTopic := runwaymq.TopicKeyMerge.String()
+
+	s.closeGate(gateGroup, queue, "e2e: hold the lead merge so the dependent finishes building first")
+	// 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(gateGroup, queue)
+
+	lead := s.land(queue, "github://github.example.com/uber/e2e-chain/pull/1/abcdef0123456789abcdef0123456789abcdef01")
+	s.log.Logf("Landed lead request %s; awaiting its merge to park", lead.sqid)
+
+	// The merge request is keyed by batch, so name the batch to prove the
+	// parked delivery is this request's merge and not some other.
+	leadBatch := s.awaitBatchID(lead)
+	parked := s.awaitParked(gateGroup, gateTopic, leadBatch)
+	assert.Equal(t, queue, parked.PartitionKey, "merge request should be partitioned by queue")
+
+	// The lead is provably stopped mid-merge. 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")
+
+	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)
+
+	// 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)
+
+	// Start: the lead merges, and its fan-out is now the only thing that can
+	// move the dependent.
+	s.openGate(gateGroup, queue)
+	s.awaitUnparked(gateGroup, 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 merged")
+}
+
 // TestReadAPIs validates all five request read endpoints against receipts
 // created through the public Land API.
 func (s *E2EIntegrationSuite) TestReadAPIs() {