From d6ce535fa3b7a779c862528eed5c6c598e84fc61 Mon Sep 17 00:00:00 2001 From: sergeyb Date: Tue, 14 Jul 2026 01:09:24 +0000 Subject: [PATCH 1/2] fix(submitqueue): keep internal merge stage topic distinct from runway merge queue MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The orchestrator registers its internal merge pipeline stage and the runway-owned merge queue in one topic registry over one shared queue backend, and both used the topic key "merge" (topickey.TopicKeyMerge == runwaymq.TopicKeyMerge). The publish-only runway entry silently shadowed the internal stage entry (last write wins), collapsing the two topics into one: - runway consumed the internal batch-ID messages as zero-step MergeRequests and reported SUCCEEDED without ever seeing the real steps; - the orchestrator merge controller's real MergeRequest publish was then silently dropped by the queue's (topic, partition_key, id) dedupe, because it reuses the batch ID already published to the same topic. The pipeline still reached "landed" only because the zero-step result correlates by batch ID — with a real merger nothing would ever have been merged. Found by the e2e harness asserting runway's merge signal carries the request's step IDs. Rename the internal stage key and wire name to "mergebatch", and make consumer.NewTopicRegistry reject duplicate topic keys so a future collision fails at startup instead of silently misrouting messages. Co-Authored-By: Claude Fable 5 --- platform/consumer/registry.go | 11 ++++++++++- service/submitqueue/orchestrator/server/main.go | 7 ++++++- submitqueue/core/topickey/topickey.go | 6 +++++- submitqueue/orchestrator/controller/dlq/batch_test.go | 4 ++-- 4 files changed, 23 insertions(+), 5 deletions(-) diff --git a/platform/consumer/registry.go b/platform/consumer/registry.go index fd3f8b30..0cfe52bf 100644 --- a/platform/consumer/registry.go +++ b/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 diff --git a/service/submitqueue/orchestrator/server/main.go b/service/submitqueue/orchestrator/server/main.go index 2038b2e7..a5d84ade 100644 --- a/service/submitqueue/orchestrator/server/main.go +++ b/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"}, } diff --git a/submitqueue/core/topickey/topickey.go b/submitqueue/core/topickey/topickey.go index 426de3d3..1a294e34 100644 --- a/submitqueue/core/topickey/topickey.go +++ b/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. diff --git a/submitqueue/orchestrator/controller/dlq/batch_test.go b/submitqueue/orchestrator/controller/dlq/batch_test.go index 69c8e317..0d3143ca 100644 --- a/submitqueue/orchestrator/controller/dlq/batch_test.go +++ b/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()) } From 09e73e35952924cc6762a4292c0c33965b00a75d Mon Sep 17 00:00:00 2001 From: sergeyb Date: Tue, 14 Jul 2026 01:09:42 +0000 Subject: [PATCH 2/2] test(e2e): event-driven harness with per-controller parking and deterministic cancel test MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- test/e2e/submitqueue/BUILD.bazel | 14 + test/e2e/submitqueue/harness_test.go | 223 ++++++++++---- test/e2e/submitqueue/observer_test.go | 284 +++++++++++++++++ test/e2e/submitqueue/stage_test.go | 124 ++++++++ test/e2e/submitqueue/suite_test.go | 285 +++++++++++++----- .../extension/messagequeue/mysql/BUILD.bazel | 5 +- .../messagequeue/mysql/stagehold_test.go | 182 +++++++++++ test/testutil/BUILD.bazel | 2 + test/testutil/compose.go | 14 +- test/testutil/logger.go | 6 + test/testutil/stagehold.go | 151 ++++++++++ 11 files changed, 1154 insertions(+), 136 deletions(-) create mode 100644 test/e2e/submitqueue/observer_test.go create mode 100644 test/e2e/submitqueue/stage_test.go create mode 100644 test/integration/extension/messagequeue/mysql/stagehold_test.go create mode 100644 test/testutil/stagehold.go diff --git a/test/e2e/submitqueue/BUILD.bazel b/test/e2e/submitqueue/BUILD.bazel index eda42174..f0309ec1 100644 --- a/test/e2e/submitqueue/BUILD.bazel +++ b/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", ], ) diff --git a/test/e2e/submitqueue/harness_test.go b/test/e2e/submitqueue/harness_test.go index b8d18eef..aad7936f 100644 --- a/test/e2e/submitqueue/harness_test.go +++ b/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=" 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,10 +166,13 @@ 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) @@ -72,31 +180,9 @@ func (s *E2EIntegrationSuite) awaitStatus(sqid string, want entity.RequestStatus } 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) } diff --git a/test/e2e/submitqueue/observer_test.go b/test/e2e/submitqueue/observer_test.go new file mode 100644 index 00000000..1fa6d091 --- /dev/null +++ b/test/e2e/submitqueue/observer_test.go @@ -0,0 +1,284 @@ +// Copyright (c) 2025 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package e2e_test + +// The queue observer is the suite's push-based signal plane. The pipeline +// already narrates itself onto the queue — the orchestrator publishes every +// customer-facing status transition to the log topic, and runway publishes its +// merge-conflict-check and merge answers to the signal topics. Queue delivery +// state is keyed by (consumer_group, topic, partition_key, offset), so consumer +// groups are independent cursors: the observer subscribes with its own +// test-owned group over the published mysql-queue port and receives a copy of +// every message, without stealing anything from the services' own groups and +// without any service change. +// +// Tests therefore block on pushed events ("the cancelled status was published", +// "runway answered the check for this request") instead of polling on a guessed +// interval. Bounded-ness comes from the suite context deadline, not per-wait +// timeout constants. + +import ( + "context" + "database/sql" + "fmt" + "sync" + "testing" + + "github.com/uber-go/tally" + runwaymq "github.com/uber/submitqueue/api/runway/messagequeue" + runwaymqpb "github.com/uber/submitqueue/api/runway/messagequeue/protopb" + "github.com/uber/submitqueue/platform/consumer" + "github.com/uber/submitqueue/platform/errs" + extqueue "github.com/uber/submitqueue/platform/extension/messagequeue" + queueMySQL "github.com/uber/submitqueue/platform/extension/messagequeue/mysql" + "github.com/uber/submitqueue/submitqueue/core/topickey" + "github.com/uber/submitqueue/submitqueue/entity" + "github.com/uber/submitqueue/test/testutil" + "go.uber.org/zap" +) + +// observerGroup is the consumer group all observer subscriptions use. The +// compose stack (and its queue database) is created fresh per suite run, so a +// fixed name cannot collide with an earlier run's offsets. +const observerGroup = "e2e-observer" + +// pipelineEvent is one message observed on a pipeline topic, reduced to the +// fields tests correlate and assert on. +type pipelineEvent struct { + // topic is the topic key the message was observed on. + topic consumer.TopicKey + // requestID correlates the event to a request: RequestLog.RequestID on the + // log topic, MergeRequest.Id on merge-conflict-check (the request sqid), and + // MergeResult.Id on the signal topics (the sqid for check signals, the batch + // ID for merge signals). + requestID string + // status is the published request status. Log topic only. + status entity.RequestStatus + // outcome is runway's verdict. Signal topics only. + outcome runwaymqpb.Outcome + // stepIDs are the step correlation IDs carried by check/signal payloads. + // The orchestrator sets each StepId to the request sqid, so merge-signal + // events (whose top-level Id is the batch ID) match requests through here. + stepIDs []string +} + +// eventRecorder accumulates observed events and lets callers block until an +// event matching a predicate exists — scanning history first, so an await +// placed after the fact still succeeds. +type eventRecorder struct { + mu sync.Mutex + events []pipelineEvent + changed chan struct{} // closed and replaced on every append +} + +func newEventRecorder() *eventRecorder { + return &eventRecorder{changed: make(chan struct{})} +} + +func (r *eventRecorder) add(e pipelineEvent) { + r.mu.Lock() + defer r.mu.Unlock() + r.events = append(r.events, e) + close(r.changed) + r.changed = make(chan struct{}) +} + +// await returns the first recorded event matching match, blocking for new +// events until ctx expires. Events recorded before the call are considered. +func (r *eventRecorder) await(ctx context.Context, match func(pipelineEvent) bool) (pipelineEvent, error) { + next := 0 + for { + r.mu.Lock() + for ; next < len(r.events); next++ { + if match(r.events[next]) { + e := r.events[next] + r.mu.Unlock() + return e, nil + } + } + changed := r.changed + r.mu.Unlock() + + select { + case <-ctx.Done(): + return pipelineEvent{}, ctx.Err() + case <-changed: + } + } +} + +// observeController is a consumer.Controller that decodes deliveries from one +// topic into pipelineEvents. It always acks: an undecodable payload is logged +// and recorded as a bare event rather than nacked, because the observer must +// never influence delivery on the topics it watches. +type observeController struct { + name string + topicKey consumer.TopicKey + decode func(payload []byte) (pipelineEvent, error) + recorder *eventRecorder + log *testutil.TestLogger +} + +var _ consumer.Controller = (*observeController)(nil) + +func (c *observeController) Process(_ context.Context, d consumer.Delivery) error { + e, err := c.decode(d.Message().Payload) + if err != nil { + c.log.Logf("observer: undecodable payload on %s: %v", c.topicKey, err) + } + e.topic = c.topicKey + c.recorder.add(e) + return nil +} + +func (c *observeController) Name() string { return c.name } +func (c *observeController) TopicKey() consumer.TopicKey { return c.topicKey } +func (c *observeController) ConsumerGroup() string { return observerGroup } + +// queueObserver owns the observer consumer and the recorder tests await on. +type queueObserver struct { + recorder *eventRecorder + consumer consumer.Consumer + queue extqueue.Queue +} + +// startQueueObserver subscribes test-owned consumer groups to the log topic and +// runway's check/signal topics on the given queue database, and starts pumping +// decoded events into the returned recorder. The *sql.DB stays owned by the +// caller; stop() releases only the observer's own resources. +func startQueueObserver(t *testing.T, log *testutil.TestLogger, ctx context.Context, queueDB *sql.DB) (*queueObserver, error) { + t.Helper() + + logger, err := zap.NewDevelopment() + if err != nil { + return nil, fmt.Errorf("failed to create observer logger: %w", err) + } + + q, err := queueMySQL.NewQueue(queueMySQL.Params{ + DB: queueDB, + Logger: logger.Named("e2e-observer"), + MetricsScope: tally.NoopScope, + }) + if err != nil { + return nil, fmt.Errorf("failed to create observer queue: %w", err) + } + + recorder := newEventRecorder() + + // The observed topic set: every wire name must match what the services + // register (see the orchestrator and runway server wiring). merge-signal + // events carry the batch ID as MergeResult.Id and the request sqids as + // StepIds. The internal "merge" stage topic is deliberately not observed — + // it multiplexes payload kinds — the merge-signal answer is the useful fact. + topics := []struct { + key consumer.TopicKey + name string + decode func([]byte) (pipelineEvent, error) + }{ + {topickey.TopicKeyLog, "log", decodeRequestLog}, + {runwaymq.TopicKeyMergeConflictCheck, "merge-conflict-check", decodeMergeRequest}, + {runwaymq.TopicKeyMergeConflictCheckSignal, "merge-conflict-check-signal", decodeMergeResult}, + {runwaymq.TopicKeyMergeSignal, "merge-signal", decodeMergeResult}, + } + + configs := make([]consumer.TopicConfig, 0, len(topics)) + for _, tc := range topics { + // The observer is a pure tap: no DLQ (its controllers never nack, and a + // dead-letter topic under a test-only group would be noise in the queue + // the services share). + sub := extqueue.DefaultSubscriptionConfig(observerGroup, observerGroup) + sub.DLQ.Enabled = false + configs = append(configs, consumer.TopicConfig{ + Key: tc.key, + Name: tc.name, + Queue: q, + Subscription: sub, + }) + } + + registry, err := consumer.NewTopicRegistry(configs) + if err != nil { + q.Close() + return nil, fmt.Errorf("failed to create observer topic registry: %w", err) + } + + // No classifiers: observer controllers never return an error, so no + // retry/classification policy is ever exercised. + cons := consumer.New(logger.Sugar().Named("e2e-observer"), tally.NoopScope, registry, errs.NewClassifierProcessor()) + for _, tc := range topics { + ctl := &observeController{ + name: fmt.Sprintf("e2e-observer-%s", tc.name), + topicKey: tc.key, + decode: tc.decode, + recorder: recorder, + log: log, + } + if err := cons.Register(ctl); err != nil { + q.Close() + return nil, fmt.Errorf("failed to register observer controller for %s: %w", tc.name, err) + } + } + + if err := cons.Start(ctx); err != nil { + q.Close() + return nil, fmt.Errorf("failed to start observer consumer: %w", err) + } + + log.Logf("Queue observer started (group %s)", observerGroup) + return &queueObserver{recorder: recorder, consumer: cons, queue: q}, nil +} + +// stop shuts down the observer consumer and its queue handles. The underlying +// *sql.DB belongs to the caller and is left open. +func (o *queueObserver) stop(timeoutMs int64) error { + stopErr := o.consumer.Stop(timeoutMs) + closeErr := o.queue.Close() + if stopErr != nil { + return stopErr + } + return closeErr +} + +func decodeRequestLog(payload []byte) (pipelineEvent, error) { + l, err := entity.RequestLogFromBytes(payload) + if err != nil { + return pipelineEvent{}, fmt.Errorf("request log: %w", err) + } + return pipelineEvent{requestID: l.RequestID, status: l.Status}, nil +} + +func decodeMergeRequest(payload []byte) (pipelineEvent, error) { + req := &runwaymq.MergeRequest{} + if err := runwaymq.Unmarshal(payload, req); err != nil { + return pipelineEvent{}, fmt.Errorf("merge request: %w", err) + } + e := pipelineEvent{requestID: req.GetId()} + for _, s := range req.GetSteps() { + e.stepIDs = append(e.stepIDs, s.GetStepId()) + } + return e, nil +} + +func decodeMergeResult(payload []byte) (pipelineEvent, error) { + res := &runwaymq.MergeResult{} + if err := runwaymq.Unmarshal(payload, res); err != nil { + return pipelineEvent{}, fmt.Errorf("merge result: %w", err) + } + e := pipelineEvent{requestID: res.GetId(), outcome: res.GetOutcome()} + for _, s := range res.GetSteps() { + e.stepIDs = append(e.stepIDs, s.GetStepId()) + } + return e, nil +} diff --git a/test/e2e/submitqueue/stage_test.go b/test/e2e/submitqueue/stage_test.go new file mode 100644 index 00000000..7b1e6486 --- /dev/null +++ b/test/e2e/submitqueue/stage_test.go @@ -0,0 +1,124 @@ +// Copyright (c) 2025 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package e2e_test + +// Stage catalog: every queue consumer controller across all three services, +// enumerated from the actual wiring in the server mains. This knowledge is +// duplicated here so the test can plant pre-holds on any stage; see +// NewStageHold in test/testutil for the mechanism. +// +// Source of truth for each service's topic→consumer-group mapping: +// - Gateway: service/submitqueue/gateway/server/main.go +// - Orchestrator: service/submitqueue/orchestrator/server/main.go +// - Runway: service/runway/server/main.go + +import ( + "github.com/stretchr/testify/require" + "github.com/uber/submitqueue/test/testutil" +) + +// stage identifies a queue consumer controller by its topic wire name and +// consumer group, sufficient to plant a StageHold on a specific partition. +type stage struct { + // topic is the wire name registered in the topic registry (the Name field + // of consumer.TopicConfig, not the TopicKey constant). + topic string + // consumerGroup is the group suffix the service subscribes with. + consumerGroup string +} + +// --------------------------------------------------------------------------- +// Gateway stages (source: service/submitqueue/gateway/server/main.go) +// --------------------------------------------------------------------------- + +var ( + // stageGatewayLog is the gateway's log consumer that persists request_log + // entries published by the orchestrator. + stageGatewayLog = stage{topic: "log", consumerGroup: "gateway-log"} +) + +// --------------------------------------------------------------------------- +// Orchestrator primary stages (source: service/submitqueue/orchestrator/server/main.go) +// --------------------------------------------------------------------------- + +var ( + stageOrchestratorStart = stage{topic: "start", consumerGroup: "orchestrator-start"} + stageOrchestratorCancel = stage{topic: "cancel", consumerGroup: "orchestrator-cancel"} + stageOrchestratorValidate = stage{topic: "validate", consumerGroup: "orchestrator-validate"} + stageOrchestratorMergeConflictSignal = stage{topic: "merge-conflict-check-signal", consumerGroup: "orchestrator-mergeconflictsignal"} + stageOrchestratorBatch = stage{topic: "batch", consumerGroup: "orchestrator-batch"} + stageOrchestratorScore = stage{topic: "score", consumerGroup: "orchestrator-score"} + stageOrchestratorSpeculate = stage{topic: "speculate", consumerGroup: "orchestrator-speculate"} + stageOrchestratorBuild = stage{topic: "build", consumerGroup: "orchestrator-build"} + stageOrchestratorBuildSignal = stage{topic: "buildsignal", consumerGroup: "orchestrator-buildsignal"} + stageOrchestratorMerge = stage{topic: "mergebatch", consumerGroup: "orchestrator-merge"} + stageOrchestratorMergeSignal = stage{topic: "merge-signal", consumerGroup: "orchestrator-mergesignal"} + stageOrchestratorConclude = stage{topic: "conclude", consumerGroup: "orchestrator-conclude"} +) + +// --------------------------------------------------------------------------- +// Orchestrator DLQ stages (source: service/submitqueue/orchestrator/server/main.go) +// Each DLQ consumer group is the primary group suffixed with "-dlq", and each +// DLQ topic is the primary topic suffixed with "_dlq". +// --------------------------------------------------------------------------- + +var ( + stageOrchestratorStartDLQ = stage{topic: "start_dlq", consumerGroup: "orchestrator-start-dlq"} + stageOrchestratorCancelDLQ = stage{topic: "cancel_dlq", consumerGroup: "orchestrator-cancel-dlq"} + stageOrchestratorValidateDLQ = stage{topic: "validate_dlq", consumerGroup: "orchestrator-validate-dlq"} + stageOrchestratorMergeConflictSignalDLQ = stage{topic: "merge-conflict-check-signal_dlq", consumerGroup: "orchestrator-mergeconflictsignal-dlq"} + stageOrchestratorBatchDLQ = stage{topic: "batch_dlq", consumerGroup: "orchestrator-batch-dlq"} + stageOrchestratorScoreDLQ = stage{topic: "score_dlq", consumerGroup: "orchestrator-score-dlq"} + stageOrchestratorSpeculateDLQ = stage{topic: "speculate_dlq", consumerGroup: "orchestrator-speculate-dlq"} + stageOrchestratorBuildDLQ = stage{topic: "build_dlq", consumerGroup: "orchestrator-build-dlq"} + stageOrchestratorBuildSignalDLQ = stage{topic: "buildsignal_dlq", consumerGroup: "orchestrator-buildsignal-dlq"} + stageOrchestratorMergeDLQ = stage{topic: "mergebatch_dlq", consumerGroup: "orchestrator-merge-dlq"} + stageOrchestratorMergeSignalDLQ = stage{topic: "merge-signal_dlq", consumerGroup: "orchestrator-mergesignal-dlq"} + stageOrchestratorConcludeDLQ = stage{topic: "conclude_dlq", consumerGroup: "orchestrator-conclude-dlq"} +) + +// --------------------------------------------------------------------------- +// Runway stages (source: service/runway/server/main.go) +// --------------------------------------------------------------------------- + +var ( + // stageRunwayMergeConflictCheck is runway's merge-conflict-check consumer + // (dry-run merge attempts). + stageRunwayMergeConflictCheck = stage{topic: "merge-conflict-check", consumerGroup: "runway-mergeconflictcheck"} + // stageRunwayMerge is runway's committing-merge consumer. + stageRunwayMerge = stage{topic: "merge", consumerGroup: "runway-merge"} +) + +// holdStage plants a phantom partition lease that starves the given stage's +// consumer for the specified partition key. The hold is released automatically +// via t.Cleanup; callers may also call Release() on the returned handle to +// resume consumption earlier. +// +// This must be called BEFORE the partition's first message is published (the +// hold is a pre-hold — see testutil.StageHold for the limitation). +func (s *E2EIntegrationSuite) holdStage(st stage, partitionKey string) *testutil.StageHold { + s.T().Helper() + hold, err := testutil.NewStageHold( + s.log, + s.queueDB, + st.consumerGroup, + st.topic, + partitionKey, + s.T().Cleanup, + ) + require.NoError(s.T(), err, "failed to plant stage hold on %s/%s for partition %s", + st.topic, st.consumerGroup, partitionKey) + return hold +} diff --git a/test/e2e/submitqueue/suite_test.go b/test/e2e/submitqueue/suite_test.go index 19d27bc3..02abda1b 100644 --- a/test/e2e/submitqueue/suite_test.go +++ b/test/e2e/submitqueue/suite_test.go @@ -17,10 +17,23 @@ package e2e_test // E2E Integration Tests // // These tests use docker-compose from service/submitqueue/docker-compose.yml -// which requires pre-built Linux binaries. +// (Gateway + Orchestrator + Runway + 2 MySQL DBs) which requires pre-built +// Linux binaries. All three services start in SetupSuite and stop only in +// TearDownSuite. Tests that need to park the pipeline at a stage boundary use +// StageHold (test/testutil/stagehold.go) to starve individual queue consumer +// controllers without touching the services. // // Run with make target (builds binaries + runs test): // make e2e-test +// +// Waiting discipline: tests block on events the pipeline pushes onto its own +// queue topics (see observer_test.go) — the log topic for status transitions, +// runway's signal topics for check/merge answers. There are no per-wait timeout +// constants; every wait is bounded by the suite context, whose deadline derives +// from the test binary's own deadline (which Bazel sets from the test target +// timeout) minus a teardown margin. Where a plane is eventually consistent with +// the event plane (the Status RPC and the request_log, both fed by the gateway +// log consumer), helpers poll at a fixed cadence under the same context. import ( "context" @@ -33,6 +46,7 @@ import ( "github.com/stretchr/testify/require" "github.com/stretchr/testify/suite" "github.com/uber-go/tally" + runwaypb "github.com/uber/submitqueue/api/runway/protopb" gatewaypb "github.com/uber/submitqueue/api/submitqueue/gateway/protopb" orchestratorpb "github.com/uber/submitqueue/api/submitqueue/orchestrator/protopb" "github.com/uber/submitqueue/submitqueue/entity" @@ -47,34 +61,51 @@ import ( type E2EIntegrationSuite struct { suite.Suite ctx context.Context + ctxCancel context.CancelFunc log *testutil.TestLogger stack *testutil.ComposeStack gatewayClient gatewaypb.SubmitQueueGatewayClient orchestratorClient orchestratorpb.SubmitQueueOrchestratorClient + runwayClient runwaypb.RunwayClient db *sql.DB // App database queueDB *sql.DB // Queue database requestLog storage.RequestLogStore // White-box view of the request_log status timeline (app DB) requestStore storage.RequestStore // White-box view of the internal RequestState (app DB) + batchStore storage.BatchStore // White-box view of batch enrollment (app DB) + observer *queueObserver // Push-based event plane (test-owned consumer groups on mysql-queue) } func TestE2EIntegration(t *testing.T) { suite.Run(t, new(E2EIntegrationSuite)) } -// The gateway log consumer runs inside the gateway-service container, so this -// suite can only observe persistence black-box through the Status RPC — there is -// no in-process channel/HookSignal to wait on across the container boundary. A -// bounded poll is therefore the deterministic-enough analog: persistTimeout is a -// safety net (a failure here means something is genuinely stuck, not a timing -// race), and persistPollInterval bounds how often we re-query. -const ( - persistTimeout = 30 * time.Second - persistPollInterval = 500 * time.Millisecond -) +// maxTeardownMargin caps how much of the test deadline is reserved for +// teardown (graceful service stops, compose down, diagnostics). +const maxTeardownMargin = 90 * time.Second + +// stopTimeoutSec bounds a graceful service stop (SIGTERM → SIGKILL). It must +// exceed the services' 30s consumer drain window. +const stopTimeoutSec = 60 + +// suiteContext derives the context that bounds every wait in the suite from +// the test binary's own deadline (go test -timeout; under Bazel, the test +// target's timeout), reserving a slice for teardown. With no deadline set the +// context is unbounded and eventually() falls back to fallbackWaitBudget. +func suiteContext(t *testing.T) (context.Context, context.CancelFunc) { + deadline, ok := t.Deadline() + if !ok { + return context.WithCancel(context.Background()) + } + margin := time.Until(deadline) / 10 + if margin > maxTeardownMargin { + margin = maxTeardownMargin + } + return context.WithDeadline(context.Background(), deadline.Add(-margin)) +} func (s *E2EIntegrationSuite) SetupSuite() { t := s.T() - s.ctx = context.Background() + s.ctx, s.ctxCancel = suiteContext(t) s.log = testutil.NewTestLogger(t) s.log.Logf("Starting E2E integration test suite using docker-compose") @@ -85,10 +116,15 @@ func (s *E2EIntegrationSuite) SetupSuite() { // Use docker-compose from service/submitqueue (full stack) // NOTE: Assumes Linux binaries are pre-built via make target + // + // The stack gets a background context, not the deadline-bounded suite + // context: compose lifecycle commands (stop, down, diagnostics) must keep + // working during teardown even when the suite's wait budget has expired. + // The Bazel test timeout is the hard bound on the whole process. composeFile := filepath.Join(repoRoot, "service/submitqueue/docker-compose.yml") - s.stack = testutil.NewComposeStack(t, s.log, s.ctx, composeFile, "e2e-submitqueue") + s.stack = testutil.NewComposeStack(t, s.log, context.Background(), composeFile, "e2e-submitqueue") - // Start the compose stack (Gateway + Orchestrator + 2 MySQL DBs) + // Start the compose stack (Gateway + Orchestrator + Runway + 2 MySQL DBs) err := s.stack.Up() require.NoError(t, err, "failed to start compose stack") @@ -112,9 +148,17 @@ func (s *E2EIntegrationSuite) SetupSuite() { s.log.Logf("Schemas applied successfully") // White-box handles on the app DB: the request_log audit trail (ordered - // status history) and the operating store (point-in-time RequestState). + // status history), the operating store (point-in-time RequestState), and + // the batch store (batch enrollment). s.requestLog = storagemysql.NewRequestLogStore(s.db, tally.NoopScope) s.requestStore = storagemysql.NewRequestStore(s.db, tally.NoopScope) + s.batchStore = storagemysql.NewBatchStore(s.db, tally.NoopScope) + + // Event plane: subscribe test-owned consumer groups to the pipeline's own + // topics so the services push their progress to the test (observer_test.go). + // Must start after the queue schema is applied. + s.observer, err = startQueueObserver(t, s.log, s.ctx, s.queueDB) + require.NoError(t, err, "failed to start queue observer") // Connect to Gateway gRPC service var gatewayConn *grpc.ClientConn @@ -128,6 +172,16 @@ func (s *E2EIntegrationSuite) SetupSuite() { require.NoError(t, err, "failed to connect to orchestrator") s.orchestratorClient = orchestratorpb.NewSubmitQueueOrchestratorClient(orchestratorConn) + // Connect to Runway gRPC service — the landing service must be up for any + // request to progress past validation (it answers the merge-conflict-check + // and merge queues). All three services start here and stop only in + // TearDownSuite; individual tests use StageHold to starve controllers + // without touching the services. + var runwayConn *grpc.ClientConn + runwayConn, err = s.stack.ConnectGRPC("runway-service", 8080) + require.NoError(t, err, "failed to connect to runway") + s.runwayClient = runwaypb.NewRunwayClient(runwayConn) + s.log.Logf("E2E integration test suite ready") } @@ -135,29 +189,36 @@ func (s *E2EIntegrationSuite) TearDownSuite() { t := s.T() s.log.Logf("Tearing down E2E integration test suite") + // Stop the observer first: it consumes from the same queue database the + // services drain on shutdown, and it must not outlive the suite context. + if s.observer != nil { + if err := s.observer.stop(5000); err != nil { + s.log.Logf("Warning: failed to stop queue observer: %v", err) + } + } + // Gracefully stop services via SIGTERM and verify exit codes before compose teardown. - // Use a 60s timeout to exceed the orchestrator's 30s consumer drain window. - // Stop both services first so their shutdown runs in parallel, then check exit codes. - const stopTimeoutSec = 60 + // Stop all services first so their shutdown runs in parallel, then check exit codes. const wantExitCode = 143 // 128 + SIGTERM (15) - gatewayStopErr := s.stack.StopService("gateway-service", stopTimeoutSec) - orchestratorStopErr := s.stack.StopService("orchestrator-service", stopTimeoutSec) + services := []string{"gateway-service", "orchestrator-service", "runway-service"} + stopErrs := make(map[string]error, len(services)) + for _, svc := range services { + stopErrs[svc] = s.stack.StopService(svc, stopTimeoutSec) + } - if assert.NoError(t, gatewayStopErr, "failed to stop gateway service") { - exitCode, err := s.stack.ServiceExitCode("gateway-service") - if assert.NoError(t, err, "failed to get gateway exit code") { - assert.Equal(t, wantExitCode, exitCode, - "gateway should exit with 128+SIGTERM (%d) on graceful shutdown", wantExitCode) + for _, svc := range services { + if assert.NoErrorf(t, stopErrs[svc], "failed to stop %s", svc) { + exitCode, err := s.stack.ServiceExitCode(svc) + if assert.NoErrorf(t, err, "failed to get %s exit code", svc) { + assert.Equalf(t, wantExitCode, exitCode, + "%s should exit with 128+SIGTERM (%d) on graceful shutdown", svc, wantExitCode) + } } } - if assert.NoError(t, orchestratorStopErr, "failed to stop orchestrator service") { - exitCode, err := s.stack.ServiceExitCode("orchestrator-service") - if assert.NoError(t, err, "failed to get orchestrator exit code") { - assert.Equal(t, wantExitCode, exitCode, - "orchestrator should exit with 128+SIGTERM (%d) on graceful shutdown", wantExitCode) - } + if s.ctxCancel != nil { + s.ctxCancel() } // Compose stack cleanup handled automatically by t.Cleanup @@ -177,12 +238,23 @@ func (s *E2EIntegrationSuite) TestPingOrchestrator() { s.log.Logf("Orchestrator ping: %s", resp.Message) } +func (s *E2EIntegrationSuite) TestPingRunway() { + resp, err := s.runwayClient.Ping(s.ctx, &runwaypb.PingRequest{Message: "e2e test"}) + require.NoError(s.T(), err, "Runway Ping failed") + assert.Equal(s.T(), "runway", resp.ServiceName) + s.log.Logf("Runway ping: %s", resp.Message) +} + // TestLand_HappyPath_ReachesLanded drives a single request through the whole // pipeline to terminal success on the fully-hermetic e2e-test-queue (no -// conflicts, fake build succeeds, noop runway signals SUCCEEDED for both the -// merge-conflict check and the merge). It asserts three views: the black-box -// terminal Status, the ordered request_log status history, and the internal -// RequestState in the operating store. +// conflicts, fake build succeeds, noop runway merger answers SUCCEEDED for both +// the merge-conflict check and the merge — the merge outcome is faked at the +// same fidelity as the build outcome). +// +// It asserts four views: runway's answers on its signal topics (proof the +// landing service participated), the pushed "landed" status event, the internal +// RequestState in the operating store, and the customer-facing Status RPC plus +// the ordered request_log history. // // This also exercises the request-log ownership invariant end-to-end: the // orchestrator only *publishes* log entries to the log topic (it never writes @@ -192,30 +264,40 @@ func (s *E2EIntegrationSuite) TestPingOrchestrator() { // presence in the timeline proves the path works. func (s *E2EIntegrationSuite) TestLand_HappyPath_ReachesLanded() { sqid := s.land("e2e-test-queue", "github://github.example.com/uber/e2e-service/pull/123/abcdef0123456789abcdef0123456789abcdef01") - s.log.Logf("Land (happy path) succeeded: sqid=%s; waiting for landed", sqid) - - // Black-box: the customer-facing status reaches landed. - s.awaitStatus(sqid, entity.RequestStatusLanded) - - // White-box (status history): the request_log is the only ordered trail. All - // status entries for a request share its request_id partition on the log - // topic (ordered delivery) and the terminal "landed" is published last, so - // once "landed" is observed the earlier statuses are already persisted. This - // is a tolerant ordered-subsequence match — display statuses the pipeline - // does not emit (e.g. validating, speculating, building) are omitted. - s.assertStatusesInOrder(sqid, + s.log.Logf("Land (happy path) succeeded: sqid=%s; awaiting pipeline events", sqid) + + // Runway round trips, observed on the wire: the orchestrator handed the + // check to runway, and runway answered both the dry-run check and the + // committing merge with SUCCEEDED. + checkSignal := s.awaitCheckSignal(sqid) + s.assertOutcomeSucceeded(checkSignal, "runway merge-conflict-check signal") + mergeSignal := s.awaitMergeSignal(sqid) + s.assertOutcomeSucceeded(mergeSignal, "runway merge signal") + + // Event plane: the pipeline published the terminal "landed" status. + s.awaitStatusEvent(sqid, entity.RequestStatusLanded) + + // White-box (internal state): the operating store's authoritative + // RequestState settled on landed. The orchestrator CAS-writes state before + // publishing the log event, so this read is deterministic after the await. + assert.Equal(s.T(), entity.RequestStateLanded, s.terminalState(sqid), + "operating store should show request %s in terminal state landed", sqid) + + // Black-box: the customer-facing status converges to landed once the + // gateway's log consumer persists the entry the observer already saw. + s.awaitStatusRPC(sqid, entity.RequestStatusLanded) + + // White-box (status history): the request_log is the only ordered trail. + // This is a tolerant ordered-subsequence match — display statuses the + // pipeline does not emit (e.g. validating, speculating, building) are + // omitted. + s.awaitStatusesInOrder(sqid, entity.RequestStatusAccepted, entity.RequestStatusStarted, entity.RequestStatusBatched, entity.RequestStatusScored, entity.RequestStatusLanded, ) - - // White-box (internal state): the operating store's authoritative - // RequestState settled on landed. RequestState is point-in-time, so this is a - // terminal check, not a sequence. - assert.Equal(s.T(), entity.RequestStateLanded, s.terminalState(sqid), - "operating store should show request %s in terminal state landed", sqid) } // TestCancelRequest_InvalidSqid verifies the gateway rejects an empty sqid @@ -230,33 +312,84 @@ func (s *E2EIntegrationSuite) TestCancelRequest_InvalidSqid() { "empty sqid should map to InvalidArgument; got %s", st.Code()) } -// TestCancel_RecordsIntent verifies the deterministic half of the cancel flow: -// Cancel returns OK and the gateway synchronously records a "cancelling" intent -// entry in the request_log (written directly to the app DB before the RPC -// returns, right after the Land "accepted" entry). +// TestCancel_BeforeBatch_ReachesCancelled drives a cancellation to its terminal +// "cancelled" state deterministically. Cancellation is best-effort and normally +// races the pipeline (the hermetic happy path lands in ~2s), so the test parks +// the pipeline first: a StageHold on runway's merge-conflict-check consumer +// starves it for the test's queue partition, preventing it from processing the +// check request. A request advances exactly to the merge-conflict-check +// hand-off and can go no further — the batch stage is unreachable until +// runway's signal arrives. The park point is confirmed by a pushed event +// (validate's check request observed on the runway queue), not by sleeping. +// Runway itself keeps running throughout; only the single partition is starved. // -// It deliberately does NOT assert the terminal "cancelled" outcome. Cancellation -// is best-effort and races the pipeline: on the hermetic stack the happy path -// reaches "landed" in ~2s, and a cancel published before the orchestrator's -// start controller has created the request is rejected to the DLQ and reconciled -// to "error". Asserting a terminal "cancelled" deterministically needs a -// pipeline-pause lever (e.g. a runway "park" marker that withholds the -// merge-conflict-check signal so the request is caught pre-batch) — that is the -// next incremental, per-stage addition on top of this harness. -func (s *E2EIntegrationSuite) TestCancel_RecordsIntent() { +// The cancel then wins with certainty: the cancel controller takes its +// request-only fast path (Cancelling → Cancelled) because no batch ever +// enrolled the request. After the terminal event, the hold is released and the +// test verifies the parked merge-conflict check is answered by runway and its +// stale signal does not resurrect the request — cancelled is terminal under the +// CAS state machine, and the mergeconflictsignal controller skips halted +// requests. +func (s *E2EIntegrationSuite) TestCancel_BeforeBatch_ReachesCancelled() { t := s.T() + const queue = "e2e-cancel-queue" + + // Park the pipeline at the runway boundary by holding runway's + // merge-conflict-check consumer for this queue's partition. The hold must + // be planted BEFORE the first message (Land publishes the start message + // which eventually fans out to the merge-conflict-check topic partitioned + // by queue name). t.Cleanup releases the hold automatically if Release() + // is not called explicitly below. + hold := s.holdStage(stageRunwayMergeConflictCheck, queue) + + sqid := s.land(queue, "github://github.example.com/uber/e2e-cancel/pull/9999/abcdef0123456789abcdef0123456789abcdef01") + s.log.Logf("Land (cancel path) succeeded: sqid=%s; awaiting park at runway boundary", sqid) + + // The orchestrator has started the request and handed its merge-conflict + // check to runway. Runway's consumer is starved for this partition, so the + // check request is parked on the wire: no signal can arrive, hence no + // batch can enroll the request. + s.awaitCheckRequested(sqid) + + s.cancel(sqid, "e2e cancel test") + + // The intent entry is written synchronously by the gateway before Cancel + // returns, right after Land's "accepted" — both are readable immediately. + s.awaitStatusesInOrder(sqid, + entity.RequestStatusAccepted, + entity.RequestStatusCancelling, + ) - sqid := s.land("e2e-cancel-queue", "github://github.example.com/uber/e2e-cancel/pull/9999/abcdef0123456789abcdef0123456789abcdef01") - s.log.Logf("Land (cancel path) succeeded: sqid=%s; cancelling", sqid) - - _, err := s.gatewayClient.Cancel(s.ctx, &gatewaypb.CancelRequest{Sqid: sqid, Reason: "e2e cancel test"}) - require.NoError(t, err, "Cancel failed") - - // The gateway writes "accepted" (on Land) and "cancelling" (on Cancel) - // synchronously to the same store, so both are present the moment Cancel - // returns — no polling needed. - s.assertStatusesInOrder(sqid, + // Terminal outcome, pushed: the cancel controller's request-only fast path + // published "cancelled" on the log topic. The state CAS precedes the + // publish, so the store reads below are deterministic. + s.awaitStatusEvent(sqid, entity.RequestStatusCancelled) + assert.Equal(t, entity.RequestStateCancelled, s.terminalState(sqid), + "operating store should show request %s in terminal state cancelled", sqid) + s.assertNoBatchContains(queue, sqid) + + // Release the hold: runway's next discovery tick re-acquires the partition + // and processes the parked check request. Its (now stale) signal is + // observed on the wire; the mergeconflictsignal controller must drop it + // for the halted request. + hold.Release() + s.awaitCheckSignal(sqid) + + // Cancelled is terminal: the late signal must not move the request. The + // state machine's CAS transitions make regression impossible by + // construction; these reads pin the observable outcome. + assert.Equal(t, entity.RequestStateCancelled, s.terminalState(sqid), + "request %s must remain cancelled after the stale runway signal", sqid) + s.assertNoBatchContains(queue, sqid) + + // Black-box view converges to cancelled, and the completed timeline shows + // the full intent→terminal order with no batch enrollment ever logged. + s.awaitStatusRPC(sqid, entity.RequestStatusCancelled) + s.awaitStatusesInOrder(sqid, entity.RequestStatusAccepted, + entity.RequestStatusStarted, entity.RequestStatusCancelling, + entity.RequestStatusCancelled, ) + s.assertStatusNeverLogged(sqid, entity.RequestStatusBatched) } diff --git a/test/integration/extension/messagequeue/mysql/BUILD.bazel b/test/integration/extension/messagequeue/mysql/BUILD.bazel index 92139c2f..25111c51 100644 --- a/test/integration/extension/messagequeue/mysql/BUILD.bazel +++ b/test/integration/extension/messagequeue/mysql/BUILD.bazel @@ -2,7 +2,10 @@ load("@rules_go//go:def.bzl", "go_test") go_test( name = "go_default_test", - srcs = ["queue_test.go"], + srcs = [ + "queue_test.go", + "stagehold_test.go", + ], data = [ "docker-compose.yml", "//platform/extension/messagequeue/mysql/schema", diff --git a/test/integration/extension/messagequeue/mysql/stagehold_test.go b/test/integration/extension/messagequeue/mysql/stagehold_test.go new file mode 100644 index 00000000..dc9b1428 --- /dev/null +++ b/test/integration/extension/messagequeue/mysql/stagehold_test.go @@ -0,0 +1,182 @@ +// Copyright (c) 2025 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package mysql + +// Contract integration tests for testutil.StageHold. These pin the +// partition-lease semantics that StageHold relies on, so a future change to the +// lease algebra fails THIS test loudly instead of hanging e2e tests. + +import ( + "context" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/stretchr/testify/suite" + "github.com/uber-go/tally" + "go.uber.org/zap/zaptest" + + entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" + extqueue "github.com/uber/submitqueue/platform/extension/messagequeue" + queueMySQL "github.com/uber/submitqueue/platform/extension/messagequeue/mysql" + "github.com/uber/submitqueue/test/testutil" +) + +// StageHoldContractSuite pins the partition-lease semantics testutil.StageHold +// relies on: a phantom lease with a far-future renewal timestamp prevents +// delivery for the held partition without affecting other partitions, and +// deleting the phantom row resumes delivery. +type StageHoldContractSuite struct { + suite.Suite + ctx context.Context + stack *testutil.ComposeStack + log *testutil.TestLogger +} + +func TestStageHoldContract(t *testing.T) { + suite.Run(t, new(StageHoldContractSuite)) +} + +func (s *StageHoldContractSuite) SetupSuite() { + t := s.T() + s.ctx = context.Background() + s.log = testutil.NewTestLogger(t) + + s.stack = testutil.NewComposeStack( + t, + s.log, + s.ctx, + "docker-compose.yml", + "ext-messagequeue-stagehold", + ) + + err := s.stack.Up() + require.NoError(t, err, "failed to start compose stack") + + t.Cleanup(func() { + s.log.Logf("Tearing down StageHold contract test suite") + }) +} + +// TestStageHold_BlocksHeldPartition_AllowsOthers verifies the core contract: +// - A pre-planted hold starves the held partition: messages published there +// are never delivered while the hold is active. +// - Messages to other partitions on the same topic and consumer group are +// delivered normally (proves the hold is per-partition, not per-topic). +// - After Release(), the held partition's messages are delivered. +// +// The test is deterministic (no time.Sleep): delivery is signaled via channels, +// and non-delivery is proven by observing that the subscriber completed at least +// one full discovery+poll cycle without delivering the held message. +func (s *StageHoldContractSuite) TestStageHold_BlocksHeldPartition_AllowsOthers() { + t := s.T() + + db, err := s.stack.ConnectMySQLService("mysql") + require.NoError(t, err) + + testutil.ApplySchema(t, s.log, db, testutil.SchemaDir("platform/extension/messagequeue/mysql/schema")) + + topic := "stagehold_contract_topic" + consumerGroup := "stagehold-cg" + heldPartition := "partition-held" + freePartition := "partition-free" + + // Plant the hold BEFORE publishing or subscribing. + hold, err := testutil.NewStageHold(s.log, db, consumerGroup, topic, heldPartition, t.Cleanup) + require.NoError(t, err) + + signalCh := make(chan queueMySQL.HookSignal, 100) + q, err := queueMySQL.NewQueue(queueMySQL.Params{ + DB: db, + Logger: zaptest.NewLogger(t), + MetricsScope: tally.NoopScope, + OnSignal: signalCh, + }) + require.NoError(t, err) + defer q.Close() + + publisher := q.Publisher() + subscriber := q.Subscriber() + + subConfig := extqueue.DefaultSubscriptionConfig("worker-1", consumerGroup) + subConfig.PollIntervalMs = 100 + deliveryChan, err := subscriber.Subscribe(s.ctx, topic, subConfig) + require.NoError(t, err) + + // Publish msg1 to the HELD partition, msg2 to the FREE partition. + msg1 := entityqueue.NewMessage("held-msg-1", []byte("held"), heldPartition, nil) + msg2 := entityqueue.NewMessage("free-msg-2", []byte("free"), freePartition, nil) + require.NoError(t, publisher.Publish(s.ctx, topic, msg1)) + require.NoError(t, publisher.Publish(s.ctx, topic, msg2)) + + // msg2 (free partition) should be delivered. + d2 := receiveWithTimeout(t, deliveryChan) + assert.Equal(t, "free-msg-2", d2.Message().ID, "free partition message should be delivered") + require.NoError(t, d2.Ack(s.ctx)) + t.Logf("Received and acked free partition message") + + // Publish msg3 to the free partition and await it — proves at least one + // more full poll/discovery cycle has elapsed while the held partition + // stayed starved. + msg3 := entityqueue.NewMessage("free-msg-3", []byte("free-2"), freePartition, nil) + require.NoError(t, publisher.Publish(s.ctx, topic, msg3)) + + d3 := receiveWithTimeout(t, deliveryChan) + assert.Equal(t, "free-msg-3", d3.Message().ID, "second free partition message should be delivered") + require.NoError(t, d3.Ack(s.ctx)) + t.Logf("Received and acked second free partition message (proves poll cycles elapsed)") + + // Assert msg1 (held partition) has NOT been delivered. After two successful + // deliveries from the free partition, the subscriber has completed at least + // two full poll cycles — if the held partition were accessible, msg1 would + // have appeared. + assertNoDelivery(t, deliveryChan, signalCh, queueMySQL.SignalDeliveryCheck, 2) + t.Logf("Confirmed: held partition message not delivered while hold is active") + + // Release the hold: the subscriber's next discovery tick re-acquires the + // partition and delivers msg1. + hold.Release() + t.Logf("Hold released; awaiting held partition delivery") + + d1 := receiveWithTimeout(t, deliveryChan) + assert.Equal(t, "held-msg-1", d1.Message().ID, "held partition message should be delivered after release") + require.NoError(t, d1.Ack(s.ctx)) + t.Logf("Received and acked held partition message after release") + + // Prove no more messages remain. + assertNoDelivery(t, deliveryChan, signalCh, queueMySQL.SignalDeliveryCheck, 2) + t.Logf("StageHold contract verified: hold blocks, release resumes, other partitions unaffected") +} + +// TestStageHold_ReleaseIsIdempotent verifies that calling Release() multiple +// times is safe (no error, no side effect beyond the first call). +func (s *StageHoldContractSuite) TestStageHold_ReleaseIsIdempotent() { + t := s.T() + + db, err := s.stack.ConnectMySQLService("mysql") + require.NoError(t, err) + + testutil.ApplySchema(t, s.log, db, testutil.SchemaDir("platform/extension/messagequeue/mysql/schema")) + + hold, err := testutil.NewStageHold(s.log, db, "idempotent-cg", "idempotent-topic", "pk", t.Cleanup) + require.NoError(t, err) + + // Release three times — should not panic or error. + hold.Release() + hold.Release() + hold.Release() + + t.Logf("Verified: Release() is idempotent") +} diff --git a/test/testutil/BUILD.bazel b/test/testutil/BUILD.bazel index 0f548dc6..479d52c4 100644 --- a/test/testutil/BUILD.bazel +++ b/test/testutil/BUILD.bazel @@ -6,10 +6,12 @@ go_library( "compose.go", "logger.go", "schema.go", + "stagehold.go", ], importpath = "github.com/uber/submitqueue/test/testutil", visibility = ["//test:__subpackages__"], deps = [ + "//platform/extension/messagequeue/mysql:go_default_library", "@com_github_go_sql_driver_mysql//:go_default_library", "@com_github_stretchr_testify//require:go_default_library", "@org_golang_google_grpc//:go_default_library", diff --git a/test/testutil/compose.go b/test/testutil/compose.go index 8f0e0663..d2a84207 100644 --- a/test/testutil/compose.go +++ b/test/testutil/compose.go @@ -43,15 +43,16 @@ type ComposeStack struct { } // getDockerComposeCommand returns the docker-compose command to use. -// Tries "docker-compose" first (V1), falls back to "docker compose" (V2). +// Prefers "docker compose" (V2) because the harness relies on V2-only behavior +// ("up --wait"); falls back to standalone "docker-compose" (V1) otherwise. func getDockerComposeCommand() []string { - // Try docker-compose (V1) - if _, err := exec.LookPath("docker-compose"); err == nil { - return []string{"docker-compose"} + // Probe for the V2 plugin: `docker compose version` fails when absent. + if err := exec.Command("docker", "compose", "version").Run(); err == nil { + return []string{"docker", "compose"} } - // Fall back to docker compose (V2) - return []string{"docker", "compose"} + // Fall back to standalone docker-compose (V1). + return []string{"docker-compose"} } // NewComposeStack creates a new compose stack from the given docker-compose file. @@ -285,6 +286,7 @@ func (s *ComposeStack) ConnectGRPC(serviceName string, containerPort int) (*grpc // StopService sends SIGTERM to a service and waits for it to stop. // timeoutSec is the maximum time to wait before Docker sends SIGKILL. +// Used during TearDownSuite to verify graceful shutdown and exit codes. func (s *ComposeStack) StopService(serviceName string, timeoutSec int) error { s.t.Helper() diff --git a/test/testutil/logger.go b/test/testutil/logger.go index 50bf3da8..6093013a 100644 --- a/test/testutil/logger.go +++ b/test/testutil/logger.go @@ -15,13 +15,17 @@ package testutil import ( + "sync" "testing" "time" ) // TestLogger is a simple test-aware logger that records elapsed time between logs. +// Safe for concurrent use — harness goroutines (e.g. queue observers) may log +// while the test goroutine does. type TestLogger struct { t *testing.T // The testing object to report logs to. + mu sync.Mutex // Guards last. last time.Time // Timestamp of the last log, for elapsed calculation. } @@ -35,10 +39,12 @@ func NewTestLogger(t *testing.T) *TestLogger { func (l *TestLogger) Logf(format string, args ...any) { l.t.Helper() now := time.Now() + l.mu.Lock() delta := "" if !l.last.IsZero() { delta = " +" + now.Sub(l.last).Truncate(time.Millisecond).String() } l.last = now + l.mu.Unlock() l.t.Logf("[%s%s] "+format, append([]any{now.Format(time.RFC3339Nano), delta}, args...)...) } diff --git a/test/testutil/stagehold.go b/test/testutil/stagehold.go new file mode 100644 index 00000000..650cd61f --- /dev/null +++ b/test/testutil/stagehold.go @@ -0,0 +1,151 @@ +// Copyright (c) 2025 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package testutil + +import ( + "database/sql" + "fmt" + "sync" + "time" + + queueMySQL "github.com/uber/submitqueue/platform/extension/messagequeue/mysql" +) + +// StageHold is a test helper that starves a single queue consumer controller by +// planting a phantom partition lease in the MySQL queue's partition_leases table. +// +// Mechanism: the MySQL queue implementation (platform/extension/messagequeue/mysql) +// schedules delivery via partition leases keyed by (consumer_group, topic, +// partition_key). TryAcquireLease steals a lease only when lease_renewed_at is +// older than now minus the lease duration. By inserting a row owned by a phantom +// subscriber ("e2e-hold") with lease_renewed_at set far in the future, the +// partition is held indefinitely — the real service subscriber's discovery loop +// finds the lease unexpired, yields acquired=false, and skips the partition. +// Messages published to the held partition accumulate but are never delivered. +// Deleting the phantom row lets the service's next discovery tick re-acquire the +// lease and resume consumption. +// +// IMPORTANT LIMITATION: this is a PRE-hold primitive — it must be planted before +// the partition's first message is published (or at least before the service's +// subscriber discovers and acquires the partition). You cannot deterministically +// steal an actively renewed lease from a running subscriber because TryAcquireLease +// only steals when the existing renewal timestamp is stale. NewStageHold enforces +// this: it fails loudly if a lease row for the partition already exists. +// +// This helper is explicitly MySQL-queue-specific. It relies on the +// queue_partition_leases schema and the lease-expiry semantics documented in +// partition_lease_store.go. It uses the exported PartitionLeasesTableName constant +// from the mysql queue package. +// +// StageHold is shared by pointer (an exception to the repo's value-type +// preference) because release ownership is shared: the t.Cleanup registered by +// NewStageHold and the caller's explicit Release() must observe the same +// released state. +type StageHold struct { + // db is the queue database connection. + db *sql.DB + // consumerGroup identifies the consumer group whose partition is held. + consumerGroup string + // topic is the queue topic name. + topic string + // partitionKey is the partition being starved. + partitionKey string + // log receives structured diagnostics. + log *TestLogger + // mu guards released. + mu sync.Mutex + // released is set only after a successful DELETE, so a failed release + // attempt (e.g. a transient DB error) stays retryable by a later call — + // including the t.Cleanup-registered one. + released bool +} + +// phantomSubscriber is the leased_by value for phantom leases. It must not +// collide with any real subscriber name. +const phantomSubscriber = "e2e-hold" + +// holdDuration is how far into the future the phantom lease's renewal timestamp +// is set. Several hours ensures no real subscriber can steal it during a test. +const holdDuration = 4 * time.Hour + +// NewStageHold plants a phantom partition lease and returns a handle. The lease +// is removed automatically via the provided cleanup registrar (typically +// t.Cleanup) if Release is not called explicitly. +// +// The plant is a plain INSERT: if a lease row for (consumerGroup, topic, +// partitionKey) already exists — whether owned by a real subscriber or another +// hold — planting fails with an error instead of silently stealing the lease. +// An existing row means the hold was planted too late (the service already +// acquired the partition, violating the pre-hold contract) or collides with +// another test's hold on the same partition. +func NewStageHold(log *TestLogger, db *sql.DB, consumerGroup, topic, partitionKey string, cleanup func(func())) (*StageHold, error) { + now := time.Now().UnixMilli() + futureRenewal := now + holdDuration.Milliseconds() + + _, err := db.Exec(fmt.Sprintf(` + INSERT INTO %s (consumer_group, topic, partition_key, leased_by, leased_at, lease_renewed_at) + VALUES (?, ?, ?, ?, ?, ?) + `, queueMySQL.PartitionLeasesTableName), + consumerGroup, topic, partitionKey, phantomSubscriber, now, futureRenewal) + if err != nil { + return nil, fmt.Errorf( + "plant stage hold (group=%s topic=%s partition=%s): partition may already be leased — "+ + "a pre-hold must be planted before the service acquires the partition, and no two holds "+ + "may target the same partition: %w", + consumerGroup, topic, partitionKey, err) + } + + h := &StageHold{ + db: db, + consumerGroup: consumerGroup, + topic: topic, + partitionKey: partitionKey, + log: log, + } + + log.Logf("StageHold planted: group=%s topic=%s partition=%s (phantom renewal %s ahead)", + consumerGroup, topic, partitionKey, holdDuration) + + cleanup(func() { h.Release() }) + + return h, nil +} + +// Release removes the phantom lease, allowing the real subscriber to re-acquire +// the partition on its next discovery tick. Release is idempotent after a +// successful delete; a failed delete leaves the hold releasable so a later call +// (including the cleanup-registered one) can retry. +func (h *StageHold) Release() { + h.mu.Lock() + defer h.mu.Unlock() + if h.released { + return + } + + _, err := h.db.Exec(fmt.Sprintf(` + DELETE FROM %s + WHERE consumer_group = ? AND topic = ? AND partition_key = ? AND leased_by = ? + `, queueMySQL.PartitionLeasesTableName), + h.consumerGroup, h.topic, h.partitionKey, phantomSubscriber) + if err != nil { + h.log.Logf("StageHold release failed (group=%s topic=%s partition=%s), will retry on next Release: %v", + h.consumerGroup, h.topic, h.partitionKey, err) + return + } + + h.released = true + h.log.Logf("StageHold released: group=%s topic=%s partition=%s", + h.consumerGroup, h.topic, h.partitionKey) +}