From 20e0c3db2a05d12133e496311072e6407a864f22 Mon Sep 17 00:00:00 2001 From: Preetam Dwivedi Date: Tue, 11 Aug 2026 20:19:56 -0700 Subject: [PATCH] test(e2e): prove a dependent batch is woken by the merge ahead of it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## 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. --- .../submitqueue/gateway/server/queues.yaml | 5 ++ test/e2e/submitqueue/harness_test.go | 24 ++++++ test/e2e/submitqueue/suite_test.go | 78 +++++++++++++++++++ 3 files changed, 107 insertions(+) diff --git a/service/submitqueue/gateway/server/queues.yaml b/service/submitqueue/gateway/server/queues.yaml index df7571fe..9ad47050 100644 --- a/service/submitqueue/gateway/server/queues.yaml +++ b/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. diff --git a/test/e2e/submitqueue/harness_test.go b/test/e2e/submitqueue/harness_test.go index d6e3a19b..8475a2e6 100644 --- a/test/e2e/submitqueue/harness_test.go +++ b/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 diff --git a/test/e2e/submitqueue/suite_test.go b/test/e2e/submitqueue/suite_test.go index def81116..dfc27479 100644 --- a/test/e2e/submitqueue/suite_test.go +++ b/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() {