[draft/wip] test(stovepipe): cover process pipeline integration

test/e2e/stovepipe/BUILD.bazel

@@ -21,6 +21,7 @@ go_test(
     deps = [
         "//api/stovepipe/protopb:go_default_library",
         "//test/testutil:go_default_library",
+        "//test/testutil/stovepipe: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",

test/e2e/stovepipe/harness_test.go

@@ -14,36 +14,16 @@
 
 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.
+// Reusable e2e helpers drive the stack through the real Stovepipe gRPC surface
+// and assert durable synchronous storage effects. Shared process helpers assert
+// asynchronous downstream outcomes.
 
 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 {
@@ -72,52 +52,11 @@ func (s *StovepipeE2ESuite) uriMapping(queue string) string {
 	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.
+// Ingest: the request row and the (queue, URI) mapping pointing at the minted
+// id. Process publication is asserted through its durable downstream outcome.
 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

@@ -26,39 +26,26 @@ package e2e_test
 //
 //   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.
+// The stack runs the Stovepipe gRPC service plus storage and queue MySQL
+// backends. Scenarios start through the public gRPC API and assert durable
+// downstream outcomes, while integration tests may seed backend state directly.
 
 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"
+	stovepipetest "github.com/uber/submitqueue/test/testutil/stovepipe"
 	"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
@@ -125,12 +112,8 @@ func (s *StovepipeE2ESuite) TestPing() {
 }
 
 // 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.
+// pipeline. It verifies the asynchronous process stage admits the request,
+// claims one slot, chooses the cold-start strategy, and publishes to build.
 func (s *StovepipeE2ESuite) TestIngest_HappyPath_Processes() {
 	const queue = "monorepo/main"
 
@@ -140,8 +123,8 @@ func (s *StovepipeE2ESuite) TestIngest_HappyPath_Processes() {
 	// Synchronous side effects of Ingest.
 	s.assertIngestPersisted(queue, id)
 
-	// Asynchronous completion: the process consumer acked the message.
-	s.awaitProcessed(queue)
+	// Asynchronous outcome, including the build-stage handoff.
+	stovepipetest.AssertColdStartAdmitted(s.T(), s.db, s.queueDB, queue, id)
 }
 
 // TestIngest_Idempotent verifies that re-ingesting the same queue resolves the

test/integration/stovepipe/BUILD.bazel

@@ -2,7 +2,10 @@ load("@rules_go//go:def.bzl", "go_test")
 
 go_test(
     name = "go_default_test",
-    srcs = ["suite_test.go"],
+    srcs = [
+        "process_test.go",
+        "suite_test.go",
+    ],
     data = [
         "//:MODULE.bazel",
         "//:go.mod",
@@ -16,7 +19,9 @@ go_test(
     ],
     deps = [
         "//api/stovepipe/protopb:go_default_library",
+        "//stovepipe/entity:go_default_library",
         "//test/testutil:go_default_library",
+        "//test/testutil/stovepipe: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",

test/integration/stovepipe/process_test.go

@@ -0,0 +1,133 @@
+// 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 stovepipe
+
+import (
+	"github.com/stretchr/testify/require"
+	pb "github.com/uber/submitqueue/api/stovepipe/protopb"
+	"github.com/uber/submitqueue/stovepipe/entity"
+	stovepipetest "github.com/uber/submitqueue/test/testutil/stovepipe"
+)
+
+// TestProcessAdmitsColdStartHead ingests a queue head and waits for the running
+// process consumer to admit it with the cold-start full-build outcome.
+func (s *StovepipeIntegrationSuite) TestProcessAdmitsColdStartHead() {
+	t := s.T()
+	const queue = "monorepo/process-admit"
+
+	s.log.Logf("Ingesting cold-start head for queue=%s", queue)
+	resp, err := s.client.Ingest(s.ctx, &pb.IngestRequest{Queue: queue})
+	require.NoError(t, err)
+	require.NotEmpty(t, resp.Id)
+
+	s.log.Logf("Ingest succeeded: id=%s; waiting for process admission and build handoff", resp.Id)
+	stovepipetest.AssertColdStartAdmitted(t, s.db, s.queueDB, queue, resp.Id)
+}
+
+// TestProcessSupersedesOlderHead seeds a stale head behind a newer
+// latest_request_id pointer and verifies process supersedes it without spending
+// a build slot.
+func (s *StovepipeIntegrationSuite) TestProcessSupersedesOlderHead() {
+	t := s.T()
+	const (
+		queue    = "monorepo/process-coalesce"
+		olderID  = "request/monorepo/process-coalesce/1"
+		newerID  = "request/monorepo/process-coalesce/2"
+		olderURI = "git://monorepo/process-coalesce/old"
+		newerURI = "git://monorepo/process-coalesce/HEAD"
+	)
+
+	s.log.Logf("Seeding stale process head: older_id=%s newer_id=%s queue=%s", olderID, newerID, queue)
+	stovepipetest.SeedCoalesceScenario(t, s.db, s.queueDB, queue, olderID, newerID, olderURI, newerURI)
+	s.log.Logf("Waiting for process consumer to supersede older_id=%s", olderID)
+	stovepipetest.AwaitProcessAcked(t, s.queueDB, queue)
+
+	older := stovepipetest.AwaitRequestState(t, s.db, olderID, string(entity.RequestStateSuperseded))
+	require.Equal(t, queue, older.Queue)
+
+	newer, err := stovepipetest.ReadRequest(s.ctx, s.db, newerID)
+	require.NoError(t, err)
+	require.Equal(t, string(entity.RequestStateAccepted), newer.State)
+
+	qrow, err := stovepipetest.ReadQueue(s.ctx, s.db, queue)
+	require.NoError(t, err)
+	require.Equal(t, int32(0), qrow.InFlightCount)
+	require.Equal(t, newerID, qrow.LatestRequestID)
+}
+
+// TestProcessRedeliveryIsIdempotent verifies a redelivery of an admitted request
+// does not claim another slot or create another build message.
+func (s *StovepipeIntegrationSuite) TestProcessRedeliveryIsIdempotent() {
+	t := s.T()
+	const queue = "monorepo/process-redelivery"
+
+	s.log.Logf("Ingesting request for redelivery test: queue=%s", queue)
+	resp, err := s.client.Ingest(s.ctx, &pb.IngestRequest{Queue: queue})
+	require.NoError(t, err)
+	require.NotEmpty(t, resp.Id)
+
+	stovepipetest.AssertColdStartAdmitted(t, s.db, s.queueDB, queue, resp.Id)
+	priorOffset := stovepipetest.ProcessAckedOffset(t, s.queueDB, queue)
+
+	s.log.Logf("Publishing redelivery: id=%s prior_offset=%d", resp.Id, priorOffset)
+	stovepipetest.PublishProcessDelivery(t, s.queueDB, resp.Id, resp.Id+"/redelivery", queue)
+	stovepipetest.AwaitProcessAckedAfter(t, s.queueDB, queue, priorOffset)
+
+	request, err := stovepipetest.ReadRequest(s.ctx, s.db, resp.Id)
+	require.NoError(t, err)
+	require.Equal(t, string(entity.RequestStateProcessing), request.State)
+
+	qrow, err := stovepipetest.ReadQueue(s.ctx, s.db, queue)
+	require.NoError(t, err)
+	require.Equal(t, int32(1), qrow.InFlightCount)
+	require.Equal(t, 1, stovepipetest.BuildMessageCount(t, s.queueDB, resp.Id))
+}
+
+// TestProcessNonRetryableFailureIsReconciledThroughDLQ verifies the real
+// classifier, queue DLQ routing, and reconciliation consumer work together.
+func (s *StovepipeIntegrationSuite) TestProcessNonRetryableFailureIsReconciledThroughDLQ() {
+	t := s.T()
+	const (
+		queue = "monorepo/process-dlq"
+		id    = "request/monorepo/process-dlq/1"
+	)
+
+	s.log.Logf("Seeding non-retryable process failure: id=%s queue=%s", id, queue)
+	_, err := s.db.ExecContext(s.ctx, `
+		INSERT INTO queue (name, last_green_uri, in_flight_count, latest_request_id, version)
+		VALUES (?, '', 0, ?, 1)`,
+		queue, "request/another-queue/2",
+	)
+	require.NoError(t, err)
+
+	_, err = s.db.ExecContext(s.ctx, `
+		INSERT INTO request (id, queue, uri, state, build_strategy, base_uri, version)
+		VALUES (?, ?, ?, ?, '', '', 1)`,
+		id, queue, "git://monorepo/process-dlq/HEAD", string(entity.RequestStateAccepted),
+	)
+	require.NoError(t, err)
+
+	stovepipetest.PublishProcessMessage(t, s.queueDB, id, queue)
+
+	s.log.Logf("Waiting for DLQ reconciliation: id=%s", id)
+	request := stovepipetest.AwaitRequestState(t, s.db, id, string(entity.RequestStateRecordedNotGreen))
+	require.Equal(t, queue, request.Queue)
+	stovepipetest.AwaitProcessDLQAcked(t, s.queueDB, queue)
+
+	qrow, err := stovepipetest.ReadQueue(s.ctx, s.db, queue)
+	require.NoError(t, err)
+	require.Zero(t, qrow.InFlightCount)
+	require.Zero(t, stovepipetest.BuildMessageCount(t, s.queueDB, id))
+}

test/integration/stovepipe/suite_test.go

@@ -103,10 +103,9 @@ func (s *StovepipeIntegrationSuite) TestPingAPI() {
 	assert.NotZero(t, resp.Timestamp)
 }
 
-// TestIngestAPI exercises the full ingest path: the controller 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. A second ingest of
-// the same queue resolves the same head and dedups to the same id.
+// TestIngestAPI exercises the durable ingest effects: the controller resolves
+// the head URI via the fake SourceControl, persists the Request and its
+// (queue, URI) mapping, and deduplicates a repeated ingest.
 func (s *StovepipeIntegrationSuite) TestIngestAPI() {
 	t := s.T()
 
@@ -127,11 +126,6 @@ func (s *StovepipeIntegrationSuite) TestIngestAPI() {
 	require.NoError(t, s.db.QueryRow("SELECT request_id FROM request_uri WHERE queue = ?", queue).Scan(&mappedID))
 	assert.Equal(t, id, mappedID, "URI mapping should point at the minted request id")
 
-	// Message published to the process topic.
-	var msgCount int
-	require.NoError(t, s.queueDB.QueryRow("SELECT COUNT(*) FROM queue_messages WHERE id = ?", id).Scan(&msgCount))
-	assert.Equal(t, 1, msgCount, "should have published one process message")
-
 	// Re-ingesting the same queue resolves the same head URI and dedups.
 	resp2, err := s.client.Ingest(s.ctx, &pb.IngestRequest{Queue: queue})
 	require.NoError(t, err, "second Ingest failed")

test/testutil/stovepipe/BUILD.bazel

@@ -0,0 +1,13 @@
+load("@rules_go//go:def.bzl", "go_library")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["process.go"],
+    importpath = "github.com/uber/submitqueue/test/testutil/stovepipe",
+    visibility = ["//test:__subpackages__"],
+    deps = [
+        "//stovepipe/core/messagequeue:go_default_library",
+        "//stovepipe/entity:go_default_library",
+        "@com_github_stretchr_testify//require:go_default_library",
+    ],
+)