test(e2e): add stovepipe end-to-end suite driving the ingest→process pipeline

test/e2e/stovepipe/BUILD.bazel

@@ -0,0 +1,30 @@
+load("@rules_go//go:def.bzl", "go_test")
+
+go_test(
+    name = "go_default_test",
+    srcs = [
+        "harness_test.go",
+        "suite_test.go",
+    ],
+    data = [
+        "//:MODULE.bazel",
+        "//:go.mod",
+        "//platform/extension/messagequeue/mysql/schema",
+        "//service/stovepipe:docker-compose.yml",
+        "//stovepipe/extension/storage/mysql/schema",
+    ],
+    tags = [
+        "e2e",
+        "external",
+        "integration",
+    ],
+    deps = [
+        "//api/stovepipe/protopb:go_default_library",
+        "//test/testutil:go_default_library",
+        "@com_github_go_sql_driver_mysql//:go_default_library",
+        "@com_github_stretchr_testify//assert:go_default_library",
+        "@com_github_stretchr_testify//require:go_default_library",
+        "@com_github_stretchr_testify//suite:go_default_library",
+        "@org_golang_google_grpc//:go_default_library",
+    ],
+)

test/e2e/stovepipe/harness_test.go

@@ -0,0 +1,123 @@
+// 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 Stovepipe gRPC surface (Ingest) and observe outcomes
+// two ways:
+//
+//   - the synchronous side effects of Ingest via raw SQL on the storage DB
+//     (the request row and its (queue, URI) mapping) and the queue DB (the
+//     published process message); and
+//   - the asynchronous completion of the process stage by polling the queue
+//     backend's per-consumer-group delivery state until the message is acked.
+//
+// Convergence is bounded by require.Eventually rather than time.Sleep: the
+// process consumer runs inside the stovepipe-service container, so there is no
+// in-process signal to await; a timeout here means the stage is genuinely stuck,
+// not a timing race.
+
+import (
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	pb "github.com/uber/submitqueue/api/stovepipe/protopb"
+)
+
+// The process consumer's topic and consumer group as wired in
+// service/stovepipe/server/main.go (topic name "process", consumer group
+// "stovepipe-process"). awaitProcessed reads the queue backend's delivery state
+// keyed by this group.
+const (
+	processTopic         = "process"
+	processConsumerGroup = "stovepipe-process"
+)
+
+// ingest admits a queue's head commit into the pipeline and returns the minted
+// request id.
+func (s *StovepipeE2ESuite) ingest(queue string) string {
+	t := s.T()
+	resp, err := s.client.Ingest(s.ctx, &pb.IngestRequest{Queue: queue})
+	require.NoError(t, err, "Ingest failed for queue %s", queue)
+	require.NotEmpty(t, resp.Id, "Ingest returned an empty id for queue %s", queue)
+	return resp.Id
+}
+
+// requestRowCount returns the number of request rows with the given id (0 or 1).
+func (s *StovepipeE2ESuite) requestRowCount(id string) int {
+	t := s.T()
+	var count int
+	require.NoError(t, s.db.QueryRow("SELECT COUNT(*) FROM request WHERE id = ?", id).Scan(&count),
+		"failed to count request rows for %s", id)
+	return count
+}
+
+// uriMapping returns the request id the (queue, URI) mapping points at.
+func (s *StovepipeE2ESuite) uriMapping(queue string) string {
+	t := s.T()
+	var mappedID string
+	require.NoError(t, s.db.QueryRow("SELECT request_id FROM request_uri WHERE queue = ?", queue).Scan(&mappedID),
+		"failed to read request_uri mapping for queue %s", queue)
+	return mappedID
+}
+
+// publishedMessageCount returns the number of process messages published for the
+// given request id (0 or 1).
+func (s *StovepipeE2ESuite) publishedMessageCount(id string) int {
+	t := s.T()
+	var count int
+	require.NoError(t, s.queueDB.QueryRow("SELECT COUNT(*) FROM queue_messages WHERE id = ?", id).Scan(&count),
+		"failed to count queue messages for %s", id)
+	return count
+}
+
+// awaitProcessed blocks until the process consumer has acked a message on the
+// given queue's partition, proving the ingest→process pipeline ran end-to-end.
+// The durable signal is the consumer group's acked-offset watermark in
+// queue_offsets: it starts at 0 and only advances once a message is acked (see
+// offset_store.go). We poll that rather than the message's own delivery-state
+// row because the queue GCs acked messages (and their delivery state) from
+// queue_messages once the watermark passes them. The queue uses the queue name
+// as the message's partition key (see the ingest controller), so the partition
+// key here is the queue.
+func (s *StovepipeE2ESuite) awaitProcessed(queue string) {
+	t := s.T()
+	const query = `
+		SELECT offset_acked
+		FROM queue_offsets
+		WHERE consumer_group = ? AND topic = ? AND partition_key = ?`
+	require.Eventually(t, func() bool {
+		var ackedOffset int64
+		err := s.queueDB.QueryRow(query, processConsumerGroup, processTopic, queue).Scan(&ackedOffset)
+		if err != nil {
+			// sql.ErrNoRows means the partition offset is not initialized yet.
+			s.log.Logf("acked offset for queue %s not ready yet: %v", queue, err)
+			return false
+		}
+		s.log.Logf("acked offset for queue %s = %d (want > 0)", queue, ackedOffset)
+		return ackedOffset > 0
+	}, processTimeout, processPollInterval,
+		"process consumer group %q on topic %q should advance the acked offset for queue %s",
+		processConsumerGroup, processTopic, queue)
+}
+
+// assertIngestPersisted asserts the synchronous side effects of a successful
+// Ingest: the request row, the (queue, URI) mapping pointing at the minted id,
+// and exactly one published process message.
+func (s *StovepipeE2ESuite) assertIngestPersisted(queue, id string) {
+	t := s.T()
+	assert.Equal(t, 1, s.requestRowCount(id), "request row should be persisted for %s", id)
+	assert.Equal(t, id, s.uriMapping(queue), "URI mapping should point at the minted request id")
+	assert.Equal(t, 1, s.publishedMessageCount(id), "should have published one process message for %s", id)
+}

test/e2e/stovepipe/suite_test.go

@@ -0,0 +1,157 @@
+// 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
+
+// Stovepipe end-to-end tests.
+//
+// These tests use docker-compose from service/stovepipe/docker-compose.yml,
+// which requires a pre-built Linux binary. Run with the make target (builds
+// binaries + runs the test):
+//
+//   make e2e-test
+//
+// or only this package (after building the binary):
+//
+//   bazel test //test/e2e/stovepipe:stovepipe_test
+//
+// The stack runs the Stovepipe gRPC service plus a storage MySQL (request,
+// request_uri) and a queue MySQL (the process stage). Unlike the integration
+// suite (test/integration/stovepipe), which asserts only that Ingest *publishes*
+// a process message, this suite additionally drives the asynchronous process
+// consumer to completion — proving the ingest→process pipeline runs end-to-end.
+
+import (
+	"context"
+	"database/sql"
+	"path/filepath"
+	"testing"
+	"time"
+
+	_ "github.com/go-sql-driver/mysql"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"github.com/stretchr/testify/suite"
+	pb "github.com/uber/submitqueue/api/stovepipe/protopb"
+	"github.com/uber/submitqueue/test/testutil"
+	"google.golang.org/grpc"
+)
+
+// The process consumer runs inside the stovepipe-service container, so this
+// suite can only observe its completion black-box through the queue backend's
+// delivery-state table — there is no in-process signal to await across the
+// container boundary. A bounded poll is the deterministic-enough analog:
+// processTimeout is a safety net (a failure here means the stage is genuinely
+// stuck, not a timing race) and processPollInterval bounds re-query frequency.
+const (
+	processTimeout      = 30 * time.Second
+	processPollInterval = 500 * time.Millisecond
+)
+
+type StovepipeE2ESuite struct {
+	suite.Suite
+	ctx     context.Context
+	log     *testutil.TestLogger
+	stack   *testutil.ComposeStack
+	client  pb.StovepipeClient
+	db      *sql.DB // storage database (request, request_uri)
+	queueDB *sql.DB // queue database (process stage)
+}
+
+func TestStovepipeE2E(t *testing.T) {
+	suite.Run(t, new(StovepipeE2ESuite))
+}
+
+func (s *StovepipeE2ESuite) SetupSuite() {
+	t := s.T()
+	s.ctx = context.Background()
+	s.log = testutil.NewTestLogger(t)
+
+	s.log.Logf("Starting Stovepipe e2e test suite using docker-compose")
+
+	// Set REPO_ROOT for the docker-compose build context.
+	repoRoot := testutil.FindRepoRoot(t)
+	t.Setenv("REPO_ROOT", repoRoot)
+
+	composeFile := filepath.Join(repoRoot, "service/stovepipe/docker-compose.yml")
+	s.stack = testutil.NewComposeStack(t, s.log, s.ctx, composeFile, "e2e-stovepipe")
+
+	err := s.stack.Up()
+	require.NoError(t, err, "failed to start compose stack")
+
+	s.db, err = s.stack.ConnectMySQLService("mysql-app")
+	require.NoError(t, err, "failed to connect to storage MySQL")
+
+	s.queueDB, err = s.stack.ConnectMySQLService("mysql-queue")
+	require.NoError(t, err, "failed to connect to queue MySQL")
+
+	// Apply schemas after the stack is up; the service connects lazily and the
+	// consumer retries, so the boot ordering is tolerated.
+	testutil.ApplySchema(t, s.log, s.db, testutil.SchemaDir("stovepipe/extension/storage/mysql/schema"))
+	testutil.ApplySchema(t, s.log, s.queueDB, testutil.SchemaDir("platform/extension/messagequeue/mysql/schema"))
+
+	var conn *grpc.ClientConn
+	conn, err = s.stack.ConnectGRPC("stovepipe-service", 8080)
+	require.NoError(t, err, "failed to connect to stovepipe service")
+	s.client = pb.NewStovepipeClient(conn)
+
+	s.log.Logf("Stovepipe e2e test suite ready")
+}
+
+func (s *StovepipeE2ESuite) TearDownSuite() {
+	// Compose stack cleanup is handled automatically by t.Cleanup (registered in
+	// NewComposeStack).
+	s.log.Logf("Tearing down Stovepipe e2e test suite")
+}
+
+func (s *StovepipeE2ESuite) TestPing() {
+	t := s.T()
+	resp, err := s.client.Ping(s.ctx, &pb.PingRequest{Message: "e2e test"})
+	require.NoError(t, err, "Stovepipe Ping failed")
+	assert.Equal(t, "stovepipe", resp.ServiceName)
+	assert.NotEmpty(t, resp.Message)
+	assert.NotZero(t, resp.Timestamp)
+}
+
+// TestIngest_HappyPath_Processes drives a queue's head commit through the whole
+// pipeline. Ingest synchronously resolves the head URI via the (fake)
+// SourceControl, persists the Request and its (queue, URI) mapping, and publishes
+// the request id to the process stage; the process consumer then drains that
+// message. This asserts both the synchronous side effects and — the piece the
+// integration suite does not cover — that the async process stage acked the
+// message.
+func (s *StovepipeE2ESuite) TestIngest_HappyPath_Processes() {
+	const queue = "monorepo/main"
+
+	id := s.ingest(queue)
+	s.log.Logf("Ingest succeeded: id=%s; waiting for process stage", id)
+
+	// Synchronous side effects of Ingest.
+	s.assertIngestPersisted(queue, id)
+
+	// Asynchronous completion: the process consumer acked the message.
+	s.awaitProcessed(queue)
+}
+
+// TestIngest_Idempotent verifies that re-ingesting the same queue resolves the
+// same head URI and dedups to the same request id.
+func (s *StovepipeE2ESuite) TestIngest_Idempotent() {
+	const queue = "monorepo/release"
+
+	id := s.ingest(queue)
+	s.log.Logf("First ingest: id=%s", id)
+
+	id2 := s.ingest(queue)
+	assert.Equal(s.T(), id, id2, "re-ingest of the same head should dedup to the same id")
+}