test(e2e): event-driven harness with per-controller parking and deterministic cancel test

Rework the SubmitQueue e2e suite so tests block on signals the pipeline
pushes to them instead of polling with hard-coded timeouts, add a
per-controller parking primitive so any pipeline stage can be starved
deterministically, and add a cancellation test that deterministically
reaches the terminal "cancelled" state.

- Queue observer (observer_test.go): test-owned consumer groups subscribe
  to the log topic and runway's check/signal topics over the published
  mysql-queue port. Delivery state is keyed per consumer group, so the
  observer receives a copy of every message without stealing from the
  services and without any service changes. Tests await pushed events
  (status transitions, runway's answers on the wire).
- No hard-coded timeouts: every wait is bounded by a suite context whose
  deadline derives from the test binary deadline (Bazel test timeout)
  minus a teardown margin. The old persistTimeout constant is gone; the
  remaining polls (Status RPC and request_log, both eventually consistent
  behind the gateway log consumer) run at a fixed cadence under that
  context.
- StageHold (testutil/stagehold.go): parks any single queue consumer
  controller with zero service changes by planting a phantom partition
  lease (far-future lease_renewed_at) in the MySQL queue's
  partition-leases table for (consumer_group, topic, partition_key). The
  plain INSERT fails loudly if the partition is already leased, so holds
  must be planted before the stage first consumes (pre-hold contract).
  Release() deletes the row (idempotent after success, retryable on
  failure) and consumption resumes on the service's next discovery tick.
  Held messages are never delivered, so parking has no retry-budget or
  DLQ interaction.
- Stage catalog (stage_test.go): enumerates every queue consumer
  controller across the three services — gateway (1), orchestrator
  (12 primary + 12 DLQ), and runway (2) — with a holdStage() suite
  helper, so tests can starve any controller x partition.
- Lease-semantics contract test (integration, messagequeue/mysql):
  pins the guarantees StageHold relies on — a held partition starves the
  group's consumer while other partitions keep flowing, releasing the
  hold resumes delivery, and Release is idempotent.
- Deterministic cancel test: park the request at the merge-conflict-check
  boundary with a StageHold on runway's merge-conflict-check consumer
  (partition = queue name) instead of stopping/starting the runway
  container (confirmed by a pushed event), cancel, await the pushed
  "cancelled" event, then assert the terminal store state, that no batch
  ever enrolled the request, and that runway's late signal after the hold
  is released does not resurrect it. The connectRunway re-dial helper is
  gone: services start once in SetupSuite and stop once in TearDownSuite.
- Runway is now first-class in the suite: gRPC ping test, graceful
  shutdown exit-code check, and happy-path assertions that runway
  answered both the merge-conflict check and the merge with SUCCEEDED
  (the merge signal must carry the request's step IDs — the assertion
  that exposed the merge topic collision fixed in the parent commit).
- testutil: ComposeStack drops StartService — there is no on-demand
  service start/stop mid-suite anymore (StopService is retained for the
  graceful-shutdown verification in teardown); compose command prefers
  docker compose V2 (the harness relies on V2-only "up --wait"); the
  stack uses a background context so teardown outlives the suite wait
  budget; TestLogger is safe for concurrent use.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: sbalabanov-zz

test/e2e/submitqueue/BUILD.bazel

@@ -2,8 +2,13 @@ load("@rules_go//go:def.bzl", "go_test")
 
 go_test(
     name = "go_default_test",
+    # First run builds the service images inside compose up; give the suite a
+    # 900s budget (the suite context derives every wait bound from it).
+    timeout = "long",
     srcs = [
         "harness_test.go",
+        "observer_test.go",
+        "stage_test.go",
         "suite_test.go",
     ],
     data = [
@@ -22,8 +27,16 @@ go_test(
     deps = [
         "//api/base/change/protopb:go_default_library",
         "//api/base/mergestrategy/protopb:go_default_library",
+        "//api/runway/messagequeue:go_default_library",
+        "//api/runway/messagequeue/protopb:go_default_library",
+        "//api/runway/protopb:go_default_library",
         "//api/submitqueue/gateway/protopb:go_default_library",
         "//api/submitqueue/orchestrator/protopb:go_default_library",
+        "//platform/consumer:go_default_library",
+        "//platform/errs:go_default_library",
+        "//platform/extension/messagequeue:go_default_library",
+        "//platform/extension/messagequeue/mysql:go_default_library",
+        "//submitqueue/core/topickey:go_default_library",
         "//submitqueue/entity:go_default_library",
         "//submitqueue/extension/storage:go_default_library",
         "//submitqueue/extension/storage/mysql:go_default_library",
@@ -35,5 +48,6 @@ go_test(
         "@org_golang_google_grpc//:go_default_library",
         "@org_golang_google_grpc//codes:go_default_library",
         "@org_golang_google_grpc//status:go_default_library",
+        "@org_uber_go_zap//:go_default_library",
     ],
 )

test/e2e/submitqueue/harness_test.go

@@ -16,26 +16,49 @@ package e2e_test
 
 // Reusable e2e helpers so tests read as intent, not plumbing. They drive the
 // stack through the real gateway gRPC surface (Land / Cancel / Status) and
-// observe outcomes two ways:
+// observe outcomes on four read-only planes:
 //
-//   - black-box, by polling the Status RPC to a target/terminal status; and
-//   - white-box, by reading the request_log timeline (RequestLogStore.List on
-//     mysql-app) to assert the ordered stage progression.
+//   - event plane: pushed events from the queue observer (observer_test.go) —
+//     status transitions on the log topic, runway's check/merge answers on the
+//     signal topics. await* helpers block on these; no polling.
+//   - control plane: the Status RPC, the customer-facing view. It reads the
+//     request_log the gateway's log consumer persists, so it lags the published
+//     event; awaitStatusRPC bridges the gap with a ctx-bounded poll.
+//   - state plane: the operating stores on mysql-app (request, batch). The
+//     orchestrator CAS-writes state *before* publishing the corresponding log
+//     event, so a state read placed after an observed event is deterministic.
+//   - timeline: the request_log audit trail on mysql-app, persisted
+//     asynchronously by the gateway's log consumer; awaitStatusesInOrder polls
+//     it to convergence.
 //
-// Convergence is bounded by require.Eventually (persistTimeout /
-// persistPollInterval) rather than time.Sleep: the pipeline consumers run inside
-// containers, so there is no in-process signal to await; a timeout here means a
-// stage is genuinely stuck, not a timing race.
+// Every wait is bounded by the suite context deadline (derived from the test
+// binary's own deadline in suite_test.go) — there are no per-wait timeout
+// constants. pollInterval below is a poll cadence, not a timeout.
 
 import (
+	"time"
+
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	changepb "github.com/uber/submitqueue/api/base/change/protopb"
 	mergestrategypb "github.com/uber/submitqueue/api/base/mergestrategy/protopb"
+	runwaymq "github.com/uber/submitqueue/api/runway/messagequeue"
+	runwaymqpb "github.com/uber/submitqueue/api/runway/messagequeue/protopb"
 	gatewaypb "github.com/uber/submitqueue/api/submitqueue/gateway/protopb"
+	"github.com/uber/submitqueue/submitqueue/core/topickey"
 	"github.com/uber/submitqueue/submitqueue/entity"
 )
 
+// pollInterval is the cadence of the ctx-bounded polls against the control
+// plane and the request_log (both are eventually consistent with the event
+// plane). It bounds how often we re-query, never how long we wait.
+const pollInterval = 250 * time.Millisecond
+
+// fallbackWaitBudget bounds eventually() only when the test binary runs with
+// no deadline at all (go test -timeout=0); under Bazel the deadline is always
+// set from the test target's timeout and this value is unused.
+const fallbackWaitBudget = time.Hour
+
 // land submits a request with the default REBASE strategy and returns its sqid.
 // URIs may carry "sq-fake=<token>" markers to steer negative paths (see
 // submitqueue/core/fakemarker); the happy path uses a plain change URI.
@@ -51,6 +74,88 @@ func (s *E2EIntegrationSuite) land(queue string, uris ...string) string {
 	return resp.Sqid
 }
 
+// cancel requests cancellation of a request via the gateway.
+func (s *E2EIntegrationSuite) cancel(sqid, reason string) {
+	_, err := s.gatewayClient.Cancel(s.ctx, &gatewaypb.CancelRequest{Sqid: sqid, Reason: reason})
+	require.NoError(s.T(), err, "Cancel failed for %s", sqid)
+}
+
+// eventually polls cond at pollInterval until it returns true, bounded by the
+// suite context deadline (the single wait budget for the whole suite).
+func (s *E2EIntegrationSuite) eventually(cond func() bool, msgAndArgs ...interface{}) {
+	s.T().Helper()
+	waitFor := fallbackWaitBudget
+	if deadline, ok := s.ctx.Deadline(); ok {
+		waitFor = time.Until(deadline)
+	}
+	require.Eventually(s.T(), cond, waitFor, pollInterval, msgAndArgs...)
+}
+
+// awaitEvent blocks until the observer has recorded an event matching match.
+// desc names the awaited condition in the failure message.
+func (s *E2EIntegrationSuite) awaitEvent(desc string, match func(pipelineEvent) bool) pipelineEvent {
+	s.T().Helper()
+	ev, err := s.observer.recorder.await(s.ctx, match)
+	require.NoErrorf(s.T(), err, "no %s observed before the suite deadline", desc)
+	return ev
+}
+
+// awaitStatusEvent blocks until the pipeline publishes the given status for the
+// request on the log topic. Because every orchestrator controller CAS-writes
+// request state before publishing the matching log entry, state-plane reads
+// placed after this call observe the corresponding state deterministically.
+func (s *E2EIntegrationSuite) awaitStatusEvent(sqid string, want entity.RequestStatus) {
+	s.T().Helper()
+	s.log.Logf("Awaiting %q status event for %s", want, sqid)
+	s.awaitEvent(string(want)+" status event for "+sqid, func(e pipelineEvent) bool {
+		return e.topic == topickey.TopicKeyLog && e.requestID == sqid && e.status == want
+	})
+}
+
+// awaitCheckRequested blocks until the orchestrator hands the request's
+// merge-conflict check to runway (validate published to the runway-owned
+// merge-conflict-check topic). This is the "parked at the runway boundary"
+// marker: with runway paused, a request that reached this point can make no
+// further forward progress until runway resumes.
+func (s *E2EIntegrationSuite) awaitCheckRequested(sqid string) {
+	s.T().Helper()
+	s.log.Logf("Awaiting merge-conflict-check request for %s", sqid)
+	s.awaitEvent("merge-conflict-check request for "+sqid, func(e pipelineEvent) bool {
+		return e.topic == runwaymq.TopicKeyMergeConflictCheck && e.requestID == sqid
+	})
+}
+
+// awaitCheckSignal blocks until runway publishes the merge-conflict-check
+// result for the request and returns it. Proof on the wire that runway consumed
+// and answered the check.
+func (s *E2EIntegrationSuite) awaitCheckSignal(sqid string) pipelineEvent {
+	s.T().Helper()
+	s.log.Logf("Awaiting merge-conflict-check signal for %s", sqid)
+	return s.awaitEvent("merge-conflict-check signal for "+sqid, func(e pipelineEvent) bool {
+		return e.topic == runwaymq.TopicKeyMergeConflictCheckSignal && e.requestID == sqid
+	})
+}
+
+// awaitMergeSignal blocks until runway publishes a committing-merge result
+// covering the request and returns it. Merge results carry the batch ID as the
+// top-level correlation ID; the request is matched through the step IDs, which
+// the orchestrator sets to the request sqid.
+func (s *E2EIntegrationSuite) awaitMergeSignal(sqid string) pipelineEvent {
+	s.T().Helper()
+	s.log.Logf("Awaiting merge signal covering %s", sqid)
+	return s.awaitEvent("merge signal covering "+sqid, func(e pipelineEvent) bool {
+		if e.topic != runwaymq.TopicKeyMergeSignal {
+			return false
+		}
+		for _, id := range e.stepIDs {
+			if id == sqid {
+				return true
+			}
+		}
+		return false
+	})
+}
+
 // currentStatus reads the request's current customer-facing status via the
 // Status RPC. A transport error is returned so callers can keep polling.
 func (s *E2EIntegrationSuite) currentStatus(sqid string) (entity.RequestStatus, error) {
@@ -61,42 +166,23 @@ func (s *E2EIntegrationSuite) currentStatus(sqid string) (entity.RequestStatus,
 	return entity.RequestStatus(resp.Status), nil
 }
 
-// awaitStatus polls Status until the request reaches exactly want.
-func (s *E2EIntegrationSuite) awaitStatus(sqid string, want entity.RequestStatus) {
-	t := s.T()
-	require.Eventually(t, func() bool {
+// awaitStatusRPC polls the Status RPC until it reports want. The Status RPC
+// reads the request_log persisted by the gateway's log consumer, which lags the
+// published event the test already observed — this bridges that persistence
+// gap; it is not the primary wait.
+func (s *E2EIntegrationSuite) awaitStatusRPC(sqid string, want entity.RequestStatus) {
+	s.T().Helper()
+	s.eventually(func() bool {
 		got, err := s.currentStatus(sqid)
 		if err != nil {
 			s.log.Logf("Status(%s) not ready yet: %v", sqid, err)
 			return false
 		}
 		s.log.Logf("Status(%s) = %q (want %q)", sqid, got, want)
 		return got == want
-	}, persistTimeout, persistPollInterval,
-		"request %s should reach status %q", sqid, want)
+	}, "request %s should reach status %q on the Status RPC", sqid, want)
 }
 
-// awaitTerminal polls Status until the request reaches a terminal status
-// (landed, error, or cancelled) and returns it.
-func (s *E2EIntegrationSuite) awaitTerminal(sqid string) entity.RequestStatus {
-	t := s.T()
-	var last entity.RequestStatus
-	require.Eventually(t, func() bool {
-		got, err := s.currentStatus(sqid)
-		if err != nil {
-			s.log.Logf("Status(%s) not ready yet: %v", sqid, err)
-			return false
-		}
-		last = got
-		s.log.Logf("Status(%s) = %q (awaiting terminal)", sqid, got)
-		return isTerminalStatus(got)
-	}, persistTimeout, persistPollInterval,
-		"request %s should reach a terminal status", sqid)
-	return last
-}
-
-// timeline returns the ordered status history from the request_log (the audit
-// trail persisted by the gateway log consumer on mysql-app).
 // timeline returns the ordered status history from the request_log (the audit
 // trail persisted by the gateway log consumer on mysql-app). These are the
 // customer-facing RequestStatus values — the only ordered history in the system
@@ -112,22 +198,43 @@ func (s *E2EIntegrationSuite) timeline(sqid string) []entity.RequestStatus {
 	return statuses
 }
 
-// assertStatusesInOrder asserts that want appears as an ordered subsequence of
-// the request_log status timeline. It tolerates intermediate statuses (so it is
-// not a change-detector), asserting only the relative order of the statuses that
-// matter.
-func (s *E2EIntegrationSuite) assertStatusesInOrder(sqid string, want ...entity.RequestStatus) {
-	t := s.T()
-	got := s.timeline(sqid)
+// containsInOrder reports whether want appears as an ordered subsequence of got.
+func containsInOrder(got []entity.RequestStatus, want []entity.RequestStatus) bool {
 	matched := 0
 	for _, st := range got {
 		if matched < len(want) && st == want[matched] {
 			matched++
 		}
 	}
-	assert.Equalf(t, len(want), matched,
-		"request_log for %s should contain %v as an ordered subsequence; got %v",
-		sqid, want, got)
+	return matched == len(want)
+}
+
+// awaitStatusesInOrder polls the request_log until want appears as an ordered
+// subsequence of the persisted status timeline. It tolerates intermediate
+// statuses (so it is not a change-detector), asserting only the relative order
+// of the statuses that matter. Polling (rather than a pure event await) is
+// needed because persistence into the request_log is the gateway log consumer's
+// job and lags the events the observer sees.
+func (s *E2EIntegrationSuite) awaitStatusesInOrder(sqid string, want ...entity.RequestStatus) {
+	s.T().Helper()
+	s.eventually(func() bool {
+		got := s.timeline(sqid)
+		if containsInOrder(got, want) {
+			return true
+		}
+		s.log.Logf("request_log for %s not yet %v; currently %v", sqid, want, got)
+		return false
+	}, "request_log for %s should contain %v as an ordered subsequence", sqid, want)
+}
+
+// assertStatusNeverLogged asserts that the persisted timeline contains no entry
+// with the given status. Call it after the terminal status has been persisted
+// (awaitStatusRPC/awaitStatusesInOrder), when the timeline is complete.
+func (s *E2EIntegrationSuite) assertStatusNeverLogged(sqid string, status entity.RequestStatus) {
+	s.T().Helper()
+	got := s.timeline(sqid)
+	assert.NotContainsf(s.T(), got, status,
+		"request_log for %s should never contain %q; got %v", sqid, status, got)
 }
 
 // terminalState reads the request's current internal RequestState from the
@@ -141,6 +248,20 @@ func (s *E2EIntegrationSuite) terminalState(sqid string) entity.RequestState {
 	return req.State
 }
 
+// assertNoBatchContains asserts that no batch in the queue — active or terminal
+// — ever enrolled the request.
+func (s *E2EIntegrationSuite) assertNoBatchContains(queue, sqid string) {
+	t := s.T()
+	allStates := append(entity.ActiveBatchStates(),
+		entity.BatchStateSucceeded, entity.BatchStateFailed, entity.BatchStateCancelled)
+	batches, err := s.batchStore.GetByQueueAndStates(s.ctx, queue, allStates)
+	require.NoError(t, err, "failed to list batches for queue %s", queue)
+	for _, b := range batches {
+		assert.NotContainsf(t, b.Contains, sqid,
+			"batch %s should not contain request %s", b.ID, sqid)
+	}
+}
+
 // lastError returns the LastError reported by the Status RPC (populated on the
 // error path).
 func (s *E2EIntegrationSuite) lastError(sqid string) string {
@@ -150,12 +271,8 @@ func (s *E2EIntegrationSuite) lastError(sqid string) string {
 	return resp.LastError
 }
 
-// isTerminalStatus reports whether a customer-facing status is terminal.
-func isTerminalStatus(status entity.RequestStatus) bool {
-	switch status {
-	case entity.RequestStatusLanded, entity.RequestStatusError, entity.RequestStatusCancelled:
-		return true
-	default:
-		return false
-	}
+// assertOutcomeSucceeded asserts a runway signal event reports SUCCEEDED.
+func (s *E2EIntegrationSuite) assertOutcomeSucceeded(ev pipelineEvent, what string) {
+	assert.Equalf(s.T(), runwaymqpb.Outcome_SUCCEEDED, ev.outcome,
+		"%s should report SUCCEEDED, got %s", what, ev.outcome)
 }