test(e2e): drive submitqueue request Land→landed with runway in the stack

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

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",

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=<token>" 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
+	}
+}