diff --git a/test/e2e/stovepipe/BUILD.bazel b/test/e2e/stovepipe/BUILD.bazel index b68903aa..bf502d72 100644 --- a/test/e2e/stovepipe/BUILD.bazel +++ b/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", diff --git a/test/e2e/stovepipe/harness_test.go b/test/e2e/stovepipe/harness_test.go index 6aa812c9..a3882861 100644 --- a/test/e2e/stovepipe/harness_test.go +++ b/test/e2e/stovepipe/harness_test.go @@ -14,20 +14,9 @@ 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" @@ -35,15 +24,6 @@ import ( 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) } diff --git a/test/e2e/stovepipe/suite_test.go b/test/e2e/stovepipe/suite_test.go index 04c75d38..a273d417 100644 --- a/test/e2e/stovepipe/suite_test.go +++ b/test/e2e/stovepipe/suite_test.go @@ -26,18 +26,15 @@ 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" @@ -45,20 +42,10 @@ import ( "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 diff --git a/test/integration/stovepipe/BUILD.bazel b/test/integration/stovepipe/BUILD.bazel index 3ca8251d..e1d6b1da 100644 --- a/test/integration/stovepipe/BUILD.bazel +++ b/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", diff --git a/test/integration/stovepipe/process_test.go b/test/integration/stovepipe/process_test.go new file mode 100644 index 00000000..13905651 --- /dev/null +++ b/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)) +} diff --git a/test/integration/stovepipe/suite_test.go b/test/integration/stovepipe/suite_test.go index 3f7495eb..59f8a8c3 100644 --- a/test/integration/stovepipe/suite_test.go +++ b/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") diff --git a/test/testutil/stovepipe/BUILD.bazel b/test/testutil/stovepipe/BUILD.bazel new file mode 100644 index 00000000..2c807bbf --- /dev/null +++ b/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", + ], +) diff --git a/test/testutil/stovepipe/process.go b/test/testutil/stovepipe/process.go new file mode 100644 index 00000000..9b59a967 --- /dev/null +++ b/test/testutil/stovepipe/process.go @@ -0,0 +1,289 @@ +// 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 provides shared helpers for Stovepipe integration and e2e +// tests that observe process-stage outcomes through the storage and queue DBs. +package stovepipe + +import ( + "context" + "database/sql" + "testing" + "time" + + "github.com/stretchr/testify/require" + stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue" + "github.com/uber/submitqueue/stovepipe/entity" +) + +const ( + // Topic and consumer group names match service/stovepipe/server/main.go. + ProcessTopic = "process" + ProcessConsumerGroup = "stovepipe-process" + BuildTopic = "build" + ProcessDLQTopic = "process_dlq" + ProcessDLQConsumerGroup = "stovepipe-process-dlq" + + defaultProcessTimeout = 30 * time.Second + defaultProcessPollInterval = 500 * time.Millisecond +) + +// RequestRow is a snapshot of the mutable process-relevant request fields. +type RequestRow struct { + ID string + Queue string + URI string + State string + BuildStrategy string + BaseURI string +} + +// QueueRow is a snapshot of per-queue coordination state process reads and writes. +type QueueRow struct { + Name string + LastGreenURI string + InFlightCount int32 + LatestRequestID string +} + +// ReadRequest loads the request row for id from the storage database. +func ReadRequest(ctx context.Context, db *sql.DB, id string) (RequestRow, error) { + var row RequestRow + err := db.QueryRowContext(ctx, ` + SELECT id, queue, uri, state, build_strategy, base_uri + FROM request + WHERE id = ?`, id).Scan( + &row.ID, &row.Queue, &row.URI, &row.State, &row.BuildStrategy, &row.BaseURI, + ) + return row, err +} + +// ReadQueue loads the queue coordination row for name from the storage database. +func ReadQueue(ctx context.Context, db *sql.DB, name string) (QueueRow, error) { + var row QueueRow + err := db.QueryRowContext(ctx, ` + SELECT name, last_green_uri, in_flight_count, latest_request_id + FROM queue + WHERE name = ?`, name).Scan( + &row.Name, &row.LastGreenURI, &row.InFlightCount, &row.LatestRequestID, + ) + return row, err +} + +// AwaitRequestState polls until the request reaches wantState or the timeout elapses. +func AwaitRequestState(t testing.TB, db *sql.DB, id, wantState string) RequestRow { + t.Helper() + ctx := context.Background() + var got RequestRow + require.Eventually(t, func() bool { + row, err := ReadRequest(ctx, db, id) + if err != nil { + t.Logf("Request %s not readable yet: %v", id, err) + return false + } + got = row + t.Logf("Request %s state=%q (want %q)", id, row.State, wantState) + return row.State == wantState + }, defaultProcessTimeout, defaultProcessPollInterval, + "request %s should reach state %q", id, wantState) + return got +} + +// AwaitProcessAcked polls the queue consumer group's acked offset for the queue +// partition until it advances past zero. +func AwaitProcessAcked(t testing.TB, queueDB *sql.DB, queue string) { + t.Helper() + awaitAckedAfter(t, queueDB, ProcessConsumerGroup, ProcessTopic, queue, 0) +} + +// ProcessAckedOffset returns the process consumer's current acknowledged offset. +func ProcessAckedOffset(t testing.TB, queueDB *sql.DB, queue string) int64 { + t.Helper() + var offset int64 + err := queueDB.QueryRow(` + SELECT offset_acked + FROM queue_offsets + WHERE consumer_group = ? AND topic = ? AND partition_key = ?`, + ProcessConsumerGroup, ProcessTopic, queue, + ).Scan(&offset) + if err == sql.ErrNoRows { + return 0 + } + require.NoError(t, err) + return offset +} + +// AwaitProcessAckedAfter waits for a process delivery newer than priorOffset. +func AwaitProcessAckedAfter(t testing.TB, queueDB *sql.DB, queue string, priorOffset int64) { + t.Helper() + awaitAckedAfter(t, queueDB, ProcessConsumerGroup, ProcessTopic, queue, priorOffset) +} + +// AwaitProcessDLQAcked waits for the DLQ reconciler to acknowledge a delivery. +func AwaitProcessDLQAcked(t testing.TB, queueDB *sql.DB, queue string) { + t.Helper() + awaitAckedAfter(t, queueDB, ProcessDLQConsumerGroup, ProcessDLQTopic, queue, 0) +} + +func awaitAckedAfter( + t testing.TB, + queueDB *sql.DB, + consumerGroup, topic, queue string, + priorOffset int64, +) { + t.Helper() + 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 := queueDB.QueryRow(query, consumerGroup, topic, queue).Scan(&ackedOffset) + if err != nil { + t.Logf("Ack offset not ready: consumer_group=%s topic=%s queue=%s: %v", + consumerGroup, topic, queue, err) + return false + } + t.Logf("Ack offset: consumer_group=%s topic=%s queue=%s offset=%d (want > %d)", + consumerGroup, topic, queue, ackedOffset, priorOffset) + return ackedOffset > priorOffset + }, defaultProcessTimeout, defaultProcessPollInterval, + "consumer group %s should ack a newer message on topic %s queue %s", + consumerGroup, topic, queue) +} + +// AwaitBuildRequest waits for and decodes the build-stage message for id. +func AwaitBuildRequest(t testing.TB, queueDB *sql.DB, id string) *stovepipemq.BuildRequest { + t.Helper() + var payload []byte + require.Eventually(t, func() bool { + err := queueDB.QueryRow(` + SELECT payload + FROM queue_messages + WHERE topic = ? AND partition_key = ? AND id = ?`, + BuildTopic, id, id, + ).Scan(&payload) + if err != nil { + t.Logf("Build request %s not published yet: %v", id, err) + return false + } + t.Logf("Build request %s found", id) + return true + }, defaultProcessTimeout, defaultProcessPollInterval, + "build request %s should be published", id) + + msg := &stovepipemq.BuildRequest{} + require.NoError(t, stovepipemq.Unmarshal(payload, msg)) + require.Equal(t, id, msg.Id) + return msg +} + +// BuildMessageCount returns the number of build-stage messages for id. +func BuildMessageCount(t testing.TB, queueDB *sql.DB, id string) int { + t.Helper() + var count int + require.NoError(t, queueDB.QueryRow(` + SELECT COUNT(*) + FROM queue_messages + WHERE topic = ? AND partition_key = ? AND id = ?`, + BuildTopic, id, id, + ).Scan(&count)) + t.Logf("Build message count: id=%s count=%d", id, count) + return count +} + +// AssertColdStartAdmitted waits for process to admit id, then checks the cold-start +// full-build outcome: processing state, full strategy, empty baseline, and one slot claimed. +func AssertColdStartAdmitted(t testing.TB, storageDB, queueDB *sql.DB, queue, id string) RequestRow { + t.Helper() + AwaitProcessAcked(t, queueDB, queue) + row := AwaitRequestState(t, storageDB, id, string(entity.RequestStateProcessing)) + require.Equal(t, string(entity.BuildStrategyFull), row.BuildStrategy, "cold start should use full build") + require.Empty(t, row.BaseURI, "cold start should have no baseline URI") + + ctx := context.Background() + qrow, err := ReadQueue(ctx, storageDB, queue) + require.NoError(t, err) + require.Equal(t, int32(1), qrow.InFlightCount, "admit should claim one build slot") + require.Equal(t, id, qrow.LatestRequestID, "latest pointer should still reference the admitted head") + AwaitBuildRequest(t, queueDB, id) + return row +} + +// PublishProcessMessage inserts a process-stage delivery for id on queue into the +// queue database. Tests use this to drive scenarios ingest cannot produce alone +// (for example coalescing an older head behind a newer latest_request_id pointer). +func PublishProcessMessage(t testing.TB, queueDB *sql.DB, id, queue string) { + t.Helper() + PublishProcessDelivery(t, queueDB, id, id, queue) +} + +// PublishProcessDelivery inserts a process-stage delivery whose queue message +// identity may differ from the referenced request identity. +func PublishProcessDelivery(t testing.TB, queueDB *sql.DB, requestID, messageID, queue string) { + t.Helper() + payload, err := stovepipemq.Marshal(&stovepipemq.ProcessRequest{Id: requestID}) + require.NoError(t, err) + + now := time.Now().UnixMilli() + _, err = queueDB.Exec(` + INSERT INTO queue_messages ( + topic, partition_key, id, payload, metadata, + created_at, published_at, visible_after, + failed_at, failure_count, last_error, original_topic + ) VALUES (?, ?, ?, ?, NULL, ?, ?, 0, 0, 0, '', '')`, + ProcessTopic, queue, messageID, payload, now, now, + ) + require.NoError(t, err) + t.Logf("Published process delivery: request_id=%s message_id=%s queue=%s", + requestID, messageID, queue) +} + +// SeedCoalesceScenario prepares an older and newer accepted head on the same queue +// with latest_request_id pointing at the newer id, then publishes a process message +// for the older head so the running consumer should supersede it on sight. +func SeedCoalesceScenario( + t testing.TB, + storageDB, queueDB *sql.DB, + queue, olderID, newerID, olderURI, newerURI string, +) { + t.Helper() + ctx := context.Background() + + _, err := storageDB.ExecContext(ctx, ` + INSERT INTO queue (name, last_green_uri, in_flight_count, latest_request_id, version) + VALUES (?, '', 0, ?, 1) + ON DUPLICATE KEY UPDATE latest_request_id = VALUES(latest_request_id), version = version + 1`, + queue, newerID, + ) + require.NoError(t, err) + + for _, spec := range []struct { + id, uri string + }{ + {olderID, olderURI}, + {newerID, newerURI}, + } { + _, err := storageDB.ExecContext(ctx, ` + INSERT INTO request (id, queue, uri, state, build_strategy, base_uri, version) + VALUES (?, ?, ?, ?, '', '', 1)`, + spec.id, queue, spec.uri, string(entity.RequestStateAccepted), + ) + require.NoError(t, err) + } + + PublishProcessMessage(t, queueDB, olderID, queue) + t.Logf("Seeded coalescing scenario: older_id=%s newer_id=%s queue=%s", + olderID, newerID, queue) +}