feat(stovepipe): extend Request entity to support new process step

stovepipe/entity/BUILD.bazel

@@ -5,14 +5,18 @@ go_library(
     srcs = [
         "queue.go",
         "request.go",
+        "request_id.go",
     ],
     importpath = "github.com/uber/submitqueue/stovepipe/entity",
     visibility = ["//visibility:public"],
 )
 
 go_test(
     name = "go_default_test",
-    srcs = ["request_test.go"],
+    srcs = [
+        "request_id_test.go",
+        "request_test.go",
+    ],
     embed = [":go_default_library"],
     deps = [
         "@com_github_stretchr_testify//assert:go_default_library",

stovepipe/entity/queue.go

@@ -34,9 +34,9 @@ type Queue struct {
 	// InFlightCount is the number of trunk validations admitted by process but not yet terminal.
 	InFlightCount int32 `json:"in_flight_count"`
 
-	// LatestRequestSeq is the highest request sequence accepted for this queue. Used to coalesce
-	// backlog so intermediate heads are skipped without superseding the newest head.
-	LatestRequestSeq int64 `json:"latest_request_seq"`
+	// LatestRequestID is the request id of the newest head ingest accepted for this queue.
+	// Empty until the first request is created. Coalescing compares IDs via CompareRequestID.
+	LatestRequestID string `json:"latest_request_id"`
 
 	// Version is the version of the object. It is used for optimistic locking.
 	// Versioning starts at 1 and is incremented for each change to the object.

stovepipe/entity/request.go

@@ -29,8 +29,25 @@ const (
 	RequestStateUnknown RequestState = ""
 	// RequestStateAccepted is the initial state of a request: a new commit has been observed
 	// for the queue and the request has been admitted into the pipeline, but no validation
-	// strategy has been chosen yet. Later stages (process, build, record) add their own states.
+	// strategy has been chosen yet.
 	RequestStateAccepted RequestState = "accepted"
+	// RequestStateProcessing means process admitted the request, recorded build strategy and
+	// baseline, and published to build.
+	RequestStateProcessing RequestState = "processing"
+	// RequestStateSuperseded means process skipped the request because a newer head exists.
+	RequestStateSuperseded RequestState = "superseded"
+)
+
+// BuildStrategy defines how build validates the request's commit.
+type BuildStrategy string
+
+const (
+	// BuildStrategyUnknown is the zero value before process chooses a strategy.
+	BuildStrategyUnknown BuildStrategy = ""
+	// BuildStrategyIncrementalSinceGreen validates only the delta since the pinned baseline URI.
+	BuildStrategyIncrementalSinceGreen BuildStrategy = "incremental_since_green"
+	// BuildStrategyFull validates the whole repo from scratch.
+	BuildStrategyFull BuildStrategy = "full"
 )
 
 // Request represents a single validation of a queue at a particular commit. The queue reports
@@ -48,9 +65,19 @@ type Request struct {
 	// and is the stable handle the ingest caller supplies.
 	Queue string `json:"queue"`
 	// URI is the opaque, VCS-agnostic locator of the commit under validation, as produced by the
-	// SourceControl extension. It is empty until SourceControl resolution is wired in.
+	// SourceControl extension.
 	URI string `json:"uri"`
 
+	// ****************
+	// Set once at process admit — empty until accepted→processing; never overwritten after
+	// ****************
+
+	// BuildStrategy is the validation scope process chose when admitting this request.
+	BuildStrategy BuildStrategy `json:"build_strategy"`
+	// BaseURI is the base URI to be used for this request when the strategy is incremental.
+	// Empty for full builds and cold start.
+	BaseURI string `json:"base_uri"`
+
 	// ****************
 	// Following fields could be changed throughout the lifecycle of the request
 	// ****************

stovepipe/entity/request_id.go

@@ -0,0 +1,60 @@
+// 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 entity
+
+import (
+	"fmt"
+	"strconv"
+	"strings"
+)
+
+// CompareRequestID compares ingest order of two request IDs in the same queue.
+// Returns -1 if a is older than b, 0 if equal, 1 if a is newer than b.
+// IDs must follow the format request/<queue>/<counter>.
+func CompareRequestID(queue, a, b string) (int, error) {
+	if a == b {
+		return 0, nil
+	}
+	aCounter, err := requestCounter(a, queue)
+	if err != nil {
+		return 0, err
+	}
+	bCounter, err := requestCounter(b, queue)
+	if err != nil {
+		return 0, err
+	}
+	switch {
+	case aCounter < bCounter:
+		return -1, nil
+	case aCounter > bCounter:
+		return 1, nil
+	default:
+		return 0, nil
+	}
+}
+
+// requestIDPrefix returns the expected prefix for request ids in queue: "request/<queue>/".
+func requestIDPrefix(queue string) string {
+	return "request/" + queue + "/"
+}
+
+// requestCounter extracts the per-queue counter suffix from a request id.
+func requestCounter(id, queue string) (int64, error) {
+	prefix := requestIDPrefix(queue)
+	if !strings.HasPrefix(id, prefix) {
+		return 0, fmt.Errorf("request id %q does not match queue %q", id, queue)
+	}
+	return strconv.ParseInt(id[len(prefix):], 10, 64)
+}

stovepipe/entity/request_id_test.go

@@ -0,0 +1,77 @@
+// 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 entity
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestCompareRequestID(t *testing.T) {
+	const queue = "monorepo/main"
+
+	tests := []struct {
+		name    string
+		a       string
+		b       string
+		want    int
+		wantErr bool
+	}{
+		{
+			name: "older vs newer",
+			a:    "request/monorepo/main/7",
+			b:    "request/monorepo/main/10",
+			want: -1,
+		},
+		{
+			name: "newer vs older",
+			a:    "request/monorepo/main/10",
+			b:    "request/monorepo/main/7",
+			want: 1,
+		},
+		{
+			name: "equal",
+			a:    "request/monorepo/main/42",
+			b:    "request/monorepo/main/42",
+			want: 0,
+		},
+		{
+			name: "numeric not lexicographic",
+			a:    "request/monorepo/main/9",
+			b:    "request/monorepo/main/10",
+			want: -1,
+		},
+		{
+			name:    "wrong queue prefix",
+			a:       "request/other/1",
+			b:       "request/monorepo/main/2",
+			wantErr: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := CompareRequestID(queue, tt.a, tt.b)
+			if tt.wantErr {
+				require.Error(t, err)
+				return
+			}
+			require.NoError(t, err)
+			assert.Equal(t, tt.want, got)
+		})
+	}
+}

stovepipe/entity/request_test.go

@@ -37,13 +37,15 @@ func TestRequest_SerializationRoundTrip(t *testing.T) {
 			},
 		},
 		{
-			name: "uri not yet resolved",
+			name: "processing with strategy and baseline",
 			req: Request{
-				ID:      "request/monorepo/main/101",
-				Queue:   "monorepo/main",
-				URI:     "",
-				State:   RequestStateAccepted,
-				Version: 1,
+				ID:            "request/monorepo/main/101",
+				Queue:         "monorepo/main",
+				URI:           "git://remote/monorepo/main/bbbb2222",
+				State:         RequestStateProcessing,
+				BuildStrategy: BuildStrategyIncrementalSinceGreen,
+				BaseURI:       "git://remote/monorepo/main/green-aaaa",
+				Version:       2,
 			},
 		},
 	}

stovepipe/extension/storage/mysql/queue_store.go

@@ -42,12 +42,12 @@ func (q *queueStore) Create(ctx context.Context, queue entity.Queue) (retErr err
 	defer func() { op.Complete(retErr) }()
 
 	_, err := q.db.ExecContext(ctx,
-		`INSERT INTO queue (name, last_green_uri, in_flight_count, latest_request_seq, version)
+		`INSERT INTO queue (name, last_green_uri, in_flight_count, latest_request_id, version)
 		 VALUES (?, ?, ?, ?, ?)`,
 		queue.Name,
 		queue.LastGreenURI,
 		queue.InFlightCount,
-		queue.LatestRequestSeq,
+		queue.LatestRequestID,
 		queue.Version,
 	)
 	if err != nil {
@@ -66,13 +66,13 @@ func (q *queueStore) Get(ctx context.Context, name string) (ret entity.Queue, re
 
 	var queue entity.Queue
 	err := q.db.QueryRowContext(ctx,
-		"SELECT name, last_green_uri, in_flight_count, latest_request_seq, version FROM queue WHERE name = ?",
+		"SELECT name, last_green_uri, in_flight_count, latest_request_id, version FROM queue WHERE name = ?",
 		name,
 	).Scan(
 		&queue.Name,
 		&queue.LastGreenURI,
 		&queue.InFlightCount,
-		&queue.LatestRequestSeq,
+		&queue.LatestRequestID,
 		&queue.Version,
 	)
 
@@ -94,11 +94,11 @@ func (q *queueStore) Update(ctx context.Context, queue entity.Queue, oldVersion,
 
 	result, err := q.db.ExecContext(ctx,
 		`UPDATE queue
-		 SET last_green_uri = ?, in_flight_count = ?, latest_request_seq = ?, version = ?
+		 SET last_green_uri = ?, in_flight_count = ?, latest_request_id = ?, version = ?
 		 WHERE name = ? AND version = ?`,
 		queue.LastGreenURI,
 		queue.InFlightCount,
-		queue.LatestRequestSeq,
+		queue.LatestRequestID,
 		newVersion,
 		queue.Name,
 		oldVersion,

stovepipe/extension/storage/mysql/request_store.go

@@ -48,8 +48,15 @@ func (r *requestStore) Create(ctx context.Context, request entity.Request) (retE
 	defer func() { op.Complete(retErr) }()
 
 	_, err := r.db.ExecContext(ctx,
-		"INSERT INTO request (id, queue, uri, state, version) VALUES (?, ?, ?, ?, ?)",
-		request.ID, request.Queue, request.URI, request.State, request.Version,
+		`INSERT INTO request (id, queue, uri, state, build_strategy, base_uri, version)
+		 VALUES (?, ?, ?, ?, ?, ?, ?)`,
+		request.ID,
+		request.Queue,
+		request.URI,
+		request.State,
+		request.BuildStrategy,
+		request.BaseURI,
+		request.Version,
 	)
 	if err != nil {
 		if isDuplicateEntry(err) {
@@ -68,9 +75,18 @@ func (r *requestStore) Get(ctx context.Context, id string) (ret entity.Request,
 
 	var req entity.Request
 	err := r.db.QueryRowContext(ctx,
-		"SELECT id, queue, uri, state, version FROM request WHERE id = ?",
+		`SELECT id, queue, uri, state, build_strategy, base_uri, version
+		 FROM request WHERE id = ?`,
 		id,
-	).Scan(&req.ID, &req.Queue, &req.URI, &req.State, &req.Version)
+	).Scan(
+		&req.ID,
+		&req.Queue,
+		&req.URI,
+		&req.State,
+		&req.BuildStrategy,
+		&req.BaseURI,
+		&req.Version,
+	)
 
 	if errors.Is(err, sql.ErrNoRows) {
 		return entity.Request{}, storage.WrapNotFound(err)
@@ -82,7 +98,7 @@ func (r *requestStore) Get(ctx context.Context, id string) (ret entity.Request,
 	return req, nil
 }
 
-// Update persists the mutable fields of request (uri, state) if the stored version matches
+// Update persists the mutable fields of request (uri, state, build_strategy, base_uri) if the
 // oldVersion, writing newVersion. Returns ErrVersionMismatch if the stored version does not match
 // (including when the request does not exist). This is a pure conditional write; the caller owns
 // version arithmetic.
@@ -91,8 +107,16 @@ func (r *requestStore) Update(ctx context.Context, request entity.Request, oldVe
 	defer func() { op.Complete(retErr) }()
 
 	result, err := r.db.ExecContext(ctx,
-		"UPDATE request SET uri = ?, state = ?, version = ? WHERE id = ? AND version = ?",
-		request.URI, request.State, newVersion, request.ID, oldVersion,
+		`UPDATE request
+		 SET uri = ?, state = ?, build_strategy = ?, base_uri = ?, version = ?
+		 WHERE id = ? AND version = ?`,
+		request.URI,
+		request.State,
+		request.BuildStrategy,
+		request.BaseURI,
+		newVersion,
+		request.ID,
+		oldVersion,
 	)
 	if err != nil {
 		return fmt.Errorf(

stovepipe/extension/storage/mysql/schema/queue.sql

@@ -1,10 +1,10 @@
 -- queue holds per-queue coordination state for the validation pipeline: the last-green
--- bookmark, in-flight gate count, and latest-request sequence pointer.
+-- bookmark, in-flight gate count, and latest-request id pointer.
 CREATE TABLE IF NOT EXISTS queue (
     name               VARCHAR(255) NOT NULL,
     last_green_uri     VARCHAR(255) NOT NULL DEFAULT '',
     in_flight_count    INT          NOT NULL DEFAULT 0,
-    latest_request_seq BIGINT       NOT NULL DEFAULT 0,
+    latest_request_id  VARCHAR(255) NOT NULL DEFAULT '',
     version            INT          NOT NULL,
     PRIMARY KEY (name)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

stovepipe/extension/storage/mysql/schema/request.sql

@@ -2,10 +2,12 @@
 -- VCS-agnostic commit locator; it may be empty until SourceControl resolution is wired in.
 -- No timestamps: created/updated times are not part of the Request entity.
 CREATE TABLE IF NOT EXISTS request (
-    id      VARCHAR(255) NOT NULL,
-    queue   VARCHAR(255) NOT NULL,
-    uri     VARCHAR(255) NOT NULL,
-    state   VARCHAR(64)  NOT NULL,
-    version INT          NOT NULL,
+    id              VARCHAR(255) NOT NULL,
+    queue           VARCHAR(255) NOT NULL,
+    uri             VARCHAR(255) NOT NULL,
+    state           VARCHAR(64)  NOT NULL,
+    build_strategy  VARCHAR(64)  NOT NULL DEFAULT '',
+    base_uri        VARCHAR(255) NOT NULL DEFAULT '',
+    version         INT          NOT NULL,
     PRIMARY KEY (id)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;