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

platform/consumer/registry.go

@@ -62,7 +62,10 @@ type topicGroup struct {
 }
 
 // NewTopicRegistry creates a new TopicRegistry from a list of TopicConfigs.
-// Returns an error if any topic name is invalid.
+// Returns an error if any topic name is invalid, or if two configs share a
+// topic key — a duplicate key would silently shadow the earlier entry (last
+// write wins on the key→queue/name maps), routing publishes and subscriptions
+// registered against one topic onto another.
 func NewTopicRegistry(configs []TopicConfig) (TopicRegistry, error) {
 	queues := make(map[TopicKey]extqueue.Queue, len(configs))
 	topicNames := make(map[TopicKey]string, len(configs))
@@ -73,6 +76,12 @@ func NewTopicRegistry(configs []TopicConfig) (TopicRegistry, error) {
 			return TopicRegistry{}, fmt.Errorf("invalid topic name for key %s: %w", cfg.Key, err)
 		}
 
+		if existing, ok := topicNames[cfg.Key]; ok {
+			return TopicRegistry{}, fmt.Errorf(
+				"duplicate topic key %s: already registered with name %q, cannot also register name %q",
+				cfg.Key, existing, cfg.Name)
+		}
+
 		queues[cfg.Key] = cfg.Queue
 		topicNames[cfg.Key] = cfg.Name
 

service/submitqueue/orchestrator/server/main.go

@@ -382,7 +382,12 @@ func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRe
 		{topickey.TopicKeySpeculate, "speculate", "orchestrator-speculate"},
 		{topickey.TopicKeyBuild, "build", "orchestrator-build"},
 		{topickey.TopicKeyBuildSignal, "buildsignal", "orchestrator-buildsignal"},
-		{topickey.TopicKeyMerge, "merge", "orchestrator-merge"},
+		// The internal merge stage is deliberately distinct from the
+		// runway-owned "merge" queue registered below — both in key (see
+		// topickey.TopicKeyMerge) and in wire name. Both topics live in the
+		// same registry over the same queue backend, so a shared key would
+		// shadow one entry and a shared name would interleave their payloads.
+		{topickey.TopicKeyMerge, "mergebatch", "orchestrator-merge"},
 		{runwaymq.TopicKeyMergeSignal, "merge-signal", "orchestrator-mergesignal"},
 		{topickey.TopicKeyConclude, "conclude", "orchestrator-conclude"},
 	}

submitqueue/core/topickey/topickey.go

@@ -42,7 +42,11 @@ const (
 	// PublishAfter when the build has not yet reached a terminal state.
 	TopicKeyBuildSignal TopicKey = "buildsignal"
 	// TopicKeyMerge is the pipeline stage where speculated batches are published for merging.
-	TopicKeyMerge TopicKey = "merge"
+	// The key value is deliberately NOT "merge": the orchestrator registers this internal
+	// stage alongside the runway-owned merge queue (runwaymq.TopicKeyMerge == "merge") in
+	// one registry over one queue backend, so both the key and the wire name must differ
+	// from runway's or the two topics collapse into each other.
+	TopicKeyMerge TopicKey = "mergebatch"
 	// TopicKeyConclude is the pipeline stage where merged requests are published for conclusion.
 	TopicKeyConclude TopicKey = "conclude"
 	// TopicKeyLog is the pipeline stage where per-request logs are written.

submitqueue/orchestrator/controller/dlq/batch_test.go

@@ -34,8 +34,8 @@ func TestDLQBatchController_InterfaceAndAccessors(t *testing.T) {
 
 	c := NewDLQBatchController(zaptest.NewLogger(t).Sugar(), testScope(), store, TopicKey(topickey.TopicKeyMerge), "orchestrator-merge-dlq")
 
-	assert.Equal(t, "merge_dlq", c.Name())
-	assert.Equal(t, consumer.TopicKey("merge_dlq"), c.TopicKey())
+	assert.Equal(t, "mergebatch_dlq", c.Name())
+	assert.Equal(t, consumer.TopicKey("mergebatch_dlq"), c.TopicKey())
 	assert.Equal(t, "orchestrator-merge-dlq", c.ConsumerGroup())
 }
 

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)
 }