From ea3af8ad36072a007d031a01c37a561213a6b421 Mon Sep 17 00:00:00 2001 From: Preetam Dwivedi Date: Mon, 6 Jul 2026 10:40:30 -0700 Subject: [PATCH] =?UTF-8?q?test(e2e):=20drive=20submitqueue=20request=20La?= =?UTF-8?q?nd=E2=86=92landed=20with=20runway=20in=20the=20stack?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary ### Why? The e2e suite barely validated anything: pings, a Land that only checked a non-empty sqid, one test that stopped at status `started`, and a thin cancel smoke test with a standing TODO. It also could not exercise the full pipeline, because `service/submitqueue/docker-compose.yml` did not include runway — the orchestrator blocks on runway's merge-conflict-check and merge signals, so every request stalled at `started`. ### What? Add `runway-service` to `service/submitqueue/docker-compose.yml`, sharing the existing `mysql-queue` (no app DB). Runway already publishes SUCCEEDED signals via its Merger extension (the noop merger wired in `service/runway/server`), so with it in the stack a request now flows all the way to `landed`. Add a reusable e2e harness (`test/e2e/submitqueue/harness_test.go`): `land`, `awaitStatus`, `awaitTerminal`, `timeline` + `assertStatusesInOrder` (white-box ordered `request_log` status history, tolerant subsequence), `terminalState` (white-box internal `RequestState` from the operating store), and `lastError` — so new per-stage and error-path tests drop in as a few lines. Rewrite the suite: `TestLand_HappyPath_ReachesLanded` drives a request to terminal success and asserts three views — the black-box `landed` Status, the ordered status history `accepted → started → batched → scored → landed`, and the operating store's internal `RequestStateLanded` (this also covers the request-log ownership invariant: every status but the synchronous `accepted` reaches storage only via the orchestrator-publishes / gateway-persists log path). The pings and graceful-shutdown checks are kept. The cancel test asserts the deterministic half (Cancel returns OK and the gateway synchronously records the `cancelling` intent). Asserting terminal `cancelled` is racy — a cancel processed before the orchestrator's start controller creates the request is rejected to the DLQ and reconciled to `error` — and needs a pipeline-pause lever, deferred to the next per-stage increment. ## Test Plan ✅ `make e2e-test` — full Docker stack (gateway + orchestrator + runway + 2 MySQL) passes; a request reaches `landed` end to end. ✅ `make fmt`; `make check-gazelle` / `make check-tidy` verified clean (only intended edits). --- service/submitqueue/docker-compose.yml | 20 +++ test/e2e/submitqueue/BUILD.bazel | 8 +- test/e2e/submitqueue/harness_test.go | 161 +++++++++++++++++++++++++ test/e2e/submitqueue/suite_test.go | 150 +++++++++++------------ 4 files changed, 263 insertions(+), 76 deletions(-) create mode 100644 test/e2e/submitqueue/harness_test.go diff --git a/service/submitqueue/docker-compose.yml b/service/submitqueue/docker-compose.yml index c8c49d53..e8d2c4bd 100644 --- a/service/submitqueue/docker-compose.yml +++ b/service/submitqueue/docker-compose.yml @@ -81,3 +81,23 @@ services: condition: service_healthy mysql-queue: condition: service_healthy + + # Runway performs the merge-conflict check and the committing merge. The + # orchestrator hands work to it on runway-owned topics and blocks on the + # signal topics, so without runway a request stalls at "started". It only + # needs the queue DB (shared with the orchestrator over mysql-queue); it has + # no application database of its own. + runway-service: + build: + context: ${REPO_ROOT} + dockerfile: service/runway/server/Dockerfile + ports: + - "8080" # Random ephemeral port to avoid conflicts + environment: + - PORT=:8080 + # Queue infrastructure connection (shared with the orchestrator) + - QUEUE_MYSQL_DSN=root:root@tcp(mysql-queue:3306)/submitqueue?parseTime=true + - HOSTNAME=runway-dev + depends_on: + mysql-queue: + condition: service_healthy diff --git a/test/e2e/submitqueue/BUILD.bazel b/test/e2e/submitqueue/BUILD.bazel index 903166ce..cdf282de 100644 --- a/test/e2e/submitqueue/BUILD.bazel +++ b/test/e2e/submitqueue/BUILD.bazel @@ -2,7 +2,10 @@ load("@rules_go//go:def.bzl", "go_test") go_test( name = "submitqueue_test", - srcs = ["suite_test.go"], + srcs = [ + "harness_test.go", + "suite_test.go", + ], data = [ "//:MODULE.bazel", "//:go.mod", @@ -22,10 +25,13 @@ go_test( "//api/submitqueue/gateway/protopb", "//api/submitqueue/orchestrator/protopb", "//submitqueue/entity", + "//submitqueue/extension/storage", + "//submitqueue/extension/storage/mysql", "//test/testutil", "@com_github_stretchr_testify//assert", "@com_github_stretchr_testify//require", "@com_github_stretchr_testify//suite", + "@com_github_uber_go_tally//:tally", "@org_golang_google_grpc//:grpc", "@org_golang_google_grpc//codes", "@org_golang_google_grpc//status", diff --git a/test/e2e/submitqueue/harness_test.go b/test/e2e/submitqueue/harness_test.go new file mode 100644 index 00000000..b8d18eef --- /dev/null +++ b/test/e2e/submitqueue/harness_test.go @@ -0,0 +1,161 @@ +// 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 + +// 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: +// +// - 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. +// +// 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. + +import ( + "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" + gatewaypb "github.com/uber/submitqueue/api/submitqueue/gateway/protopb" + "github.com/uber/submitqueue/submitqueue/entity" +) + +// 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. +func (s *E2EIntegrationSuite) land(queue string, uris ...string) string { + t := s.T() + resp, err := s.gatewayClient.Land(s.ctx, &gatewaypb.LandRequest{ + Queue: queue, + Change: &changepb.Change{Uris: uris}, + Strategy: mergestrategypb.Strategy_REBASE, + }) + require.NoError(t, err, "Land failed for queue %s", queue) + require.NotEmpty(t, resp.Sqid, "Land returned an empty sqid for queue %s", queue) + return resp.Sqid +} + +// 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) { + resp, err := s.gatewayClient.Status(s.ctx, &gatewaypb.StatusRequest{Sqid: sqid}) + if err != nil { + return entity.RequestStatusUnknown, err + } + 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 { + got, err := s.currentStatus(sqid) + if err != nil { + s.log.Logf("Status(%s) not ready yet: %v", sqid, err) + return false + } + s.log.Logf("Status(%s) = %q (want %q)", sqid, got, want) + return got == want + }, persistTimeout, persistPollInterval, + "request %s should reach status %q", sqid, want) +} + +// 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 +// (the internal RequestState is point-in-time, see terminalState). +func (s *E2EIntegrationSuite) timeline(sqid string) []entity.RequestStatus { + t := s.T() + logs, err := s.requestLog.List(s.ctx, sqid) + require.NoError(t, err, "failed to list request_log for %s", sqid) + statuses := make([]entity.RequestStatus, len(logs)) + for i, l := range logs { + statuses[i] = l.Status + } + 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) + 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) +} + +// terminalState reads the request's current internal RequestState from the +// operating store (mysql-app). Unlike the status timeline, RequestState is +// point-in-time — the Request entity is updated in place under optimistic +// locking, so only the current (terminal, once settled) value is observable. +func (s *E2EIntegrationSuite) terminalState(sqid string) entity.RequestState { + t := s.T() + req, err := s.requestStore.Get(s.ctx, sqid) + require.NoError(t, err, "failed to get request %s from operating store", sqid) + return req.State +} + +// lastError returns the LastError reported by the Status RPC (populated on the +// error path). +func (s *E2EIntegrationSuite) lastError(sqid string) string { + t := s.T() + resp, err := s.gatewayClient.Status(s.ctx, &gatewaypb.StatusRequest{Sqid: sqid}) + require.NoError(t, err, "Status failed for %s", sqid) + 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 + } +} diff --git a/test/e2e/submitqueue/suite_test.go b/test/e2e/submitqueue/suite_test.go index 69a254ff..eeb87826 100644 --- a/test/e2e/submitqueue/suite_test.go +++ b/test/e2e/submitqueue/suite_test.go @@ -32,11 +32,12 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "github.com/stretchr/testify/suite" - changepb "github.com/uber/submitqueue/api/base/change/protopb" - mergestrategypb "github.com/uber/submitqueue/api/base/mergestrategy/protopb" + "github.com/uber-go/tally" gatewaypb "github.com/uber/submitqueue/api/submitqueue/gateway/protopb" orchestratorpb "github.com/uber/submitqueue/api/submitqueue/orchestrator/protopb" "github.com/uber/submitqueue/submitqueue/entity" + "github.com/uber/submitqueue/submitqueue/extension/storage" + storagemysql "github.com/uber/submitqueue/submitqueue/extension/storage/mysql" "github.com/uber/submitqueue/test/testutil" "google.golang.org/grpc" "google.golang.org/grpc/codes" @@ -50,8 +51,10 @@ type E2EIntegrationSuite struct { stack *testutil.ComposeStack gatewayClient gatewaypb.SubmitQueueGatewayClient orchestratorClient orchestratorpb.SubmitQueueOrchestratorClient - db *sql.DB // App database - queueDB *sql.DB // Queue database + 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) } func TestE2EIntegration(t *testing.T) { @@ -108,6 +111,11 @@ 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). + s.requestLog = storagemysql.NewRequestLogStore(s.db, tally.NoopScope) + s.requestStore = storagemysql.NewRequestStore(s.db, tally.NoopScope) + // Connect to Gateway gRPC service var gatewayConn *grpc.ClientConn gatewayConn, err = s.stack.ConnectGRPC("gateway-service", 8080) @@ -169,56 +177,45 @@ func (s *E2EIntegrationSuite) TestPingOrchestrator() { s.log.Logf("Orchestrator ping: %s", resp.Message) } -func (s *E2EIntegrationSuite) TestLandRequest_SinglePR() { - req := &gatewaypb.LandRequest{ - Queue: "e2e-test-queue", - Change: &changepb.Change{Uris: []string{"github://uber/e2e-service/pull/123/abcdef0123456789abcdef0123456789abcdef01"}}, - Strategy: mergestrategypb.Strategy_REBASE, - } - - s.log.Logf("Sending Land request (single PR) for queue=%s", req.Queue) - resp, err := s.gatewayClient.Land(s.ctx, req) - require.NoError(s.T(), err, "Land request failed") - require.NotEmpty(s.T(), resp.Sqid, "SQID should not be empty") - s.log.Logf("Land request (single PR) succeeded: sqid=%s", resp.Sqid) -} - -// TestLandRequest_PersistsStartedLogViaGatewayConsumer verifies the request-log -// ownership invariant end-to-end: the orchestrator only *publishes* request log -// entries to the log topic (it never writes the request log itself), and the -// gateway's log consumer drains that topic and persists them to storage. +// 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. // -// We observe this through the gateway Status RPC: immediately after Land the -// status is "accepted" (the gateway's synchronous direct write), and once the -// orchestrator's start controller publishes "started" to the log topic, the -// gateway consumer persists it and Status advances to "started". Seeing -// "started" therefore proves the publish→consume→persist path works across both -// services. -func (s *E2EIntegrationSuite) TestLandRequest_PersistsStartedLogViaGatewayConsumer() { - t := s.T() - - landResp, err := s.gatewayClient.Land(s.ctx, &gatewaypb.LandRequest{ - Queue: "e2e-test-queue", - Change: &changepb.Change{Uris: []string{"github://uber/e2e-startlog/pull/4242/abcdef0123456789abcdef0123456789abcdef01"}}, - Strategy: mergestrategypb.Strategy_REBASE, - }) - require.NoError(t, err, "Land request failed") - require.NotEmpty(t, landResp.Sqid, "SQID should not be empty") - sqid := landResp.Sqid - s.log.Logf("Land succeeded: sqid=%s; waiting for gateway consumer to persist 'started'", sqid) - - require.Eventually(t, func() bool { - resp, statusErr := s.gatewayClient.Status(s.ctx, &gatewaypb.StatusRequest{Sqid: sqid}) - if statusErr != nil { - s.log.Logf("Status(%s) not ready yet: %v", sqid, statusErr) - return false - } - s.log.Logf("Status(%s) = %q", sqid, resp.Status) - return resp.Status == string(entity.RequestStatusStarted) - }, persistTimeout, persistPollInterval, - "request %s should reach status %q via the gateway log consumer", sqid, entity.RequestStatusStarted) - - s.log.Logf("Gateway consumer persisted orchestrator-published 'started' log for sqid=%s", sqid) +// This also exercises the request-log ownership invariant end-to-end: the +// orchestrator only *publishes* log entries to the log topic (it never writes +// the request log itself), and the gateway's log consumer drains that topic and +// persists them. Every status below except the synchronous "accepted" reaches +// storage only via that cross-service publish→consume→persist path, so its +// presence in the timeline proves the path works. +func (s *E2EIntegrationSuite) TestLand_HappyPath_ReachesLanded() { + sqid := s.land("e2e-test-queue", "github://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, + 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 @@ -233,30 +230,33 @@ func (s *E2EIntegrationSuite) TestCancelRequest_InvalidSqid() { "empty sqid should map to InvalidArgument; got %s", st.Code()) } -// TestCancelRequest_BeforeBatch is intentionally a thin smoke test of the -// Land + Cancel RPC envelope: Land to mint a sqid, then Cancel that sqid, and -// assert both calls return OK. It does not poll for terminal state, log -// entries, or any orchestrator-side progression. +// 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). // -// TODO(e2e): harden this test once the e2e fixture story is in better shape. -// Add async assertions that the request reaches RequestStateCancelled, that -// request_log contains both `cancelling` and `cancelled` entries, and that a -// second Cancel is idempotent. -func (s *E2EIntegrationSuite) TestCancelRequest_BeforeBatch() { +// 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() { t := s.T() - landReq := &gatewaypb.LandRequest{ - Queue: "e2e-cancel-queue", - Change: &changepb.Change{Uris: []string{"github://uber/e2e-nonexistent/pull/9999/deadbeef"}}, - Strategy: mergestrategypb.Strategy_REBASE, - } - landResp, err := s.gatewayClient.Land(s.ctx, landReq) - require.NoError(t, err, "Land failed") - require.NotEmpty(t, landResp.Sqid) - - _, err = s.gatewayClient.Cancel(s.ctx, &gatewaypb.CancelRequest{ - Sqid: landResp.Sqid, - Reason: "e2e cancel smoke test", - }) + sqid := s.land("e2e-cancel-queue", "github://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, + entity.RequestStatusAccepted, + entity.RequestStatusCancelling, + ) }