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