feat(stovepipe): advance queue latest_request_id on ingest

stovepipe/controller/ingest.go

@@ -126,6 +126,10 @@ func (c *IngestController) Ingest(ctx context.Context, req *pb.IngestRequest) (r
 		return nil, err
 	}
 
+	if err := c.advanceQueueLatestRequestID(ctx, queue, id); err != nil {
+		return nil, err
+	}
+
 	// Publish while the request is still pre-pipeline (Accepted). The process consumer is
 	// idempotent (keyed on the request id, at-least-once), so re-publishing on a retry or a
 	// duplicate report is safe and closes the "request created but publish failed" gap. Once
@@ -211,6 +215,39 @@ func (c *IngestController) ensureRequest(ctx context.Context, id, queue, uri str
 	return request, nil
 }
 
+// advanceQueueLatestRequestID CAS-updates queue.latest_request_id to id when id is newer.
+// Retries on optimistic-lock conflicts so concurrent ingests converge.
+func (c *IngestController) advanceQueueLatestRequestID(ctx context.Context, queue, id string) error {
+	queueStore := c.store.GetQueueStore()
+
+	for {
+		queueRow, err := queueStore.GetOrCreate(ctx, queue, entity.Queue{Version: 1})
+		if err != nil {
+			return fmt.Errorf("IngestController failed to load queue %s: %w", queue, err)
+		}
+		if queueRow.LatestRequestID != "" {
+			cmp, err := entity.CompareRequestID(queue, id, queueRow.LatestRequestID)
+			if err != nil {
+				return fmt.Errorf("IngestController failed to compare request ids for queue %s: %w", queue, err)
+			}
+			if cmp <= 0 {
+				return nil
+			}
+		}
+
+		updated := queueRow
+		updated.LatestRequestID = id
+		newVersion := queueRow.Version + 1
+		if err := queueStore.Update(ctx, updated, queueRow.Version, newVersion); err != nil {
+			if errors.Is(err, storage.ErrVersionMismatch) {
+				continue
+			}
+			return fmt.Errorf("IngestController failed to update queue %s latest_request_id: %w", queue, err)
+		}
+		return nil
+	}
+}
+
 // publishProcess publishes the request ID to the process stage, partitioned by queue so a
 // queue's requests stay ordered.
 func (c *IngestController) publishProcess(ctx context.Context, id, queue string) error {

stovepipe/controller/ingest_test.go

@@ -43,29 +43,32 @@ const (
 
 // ingestMocks bundles the mocks an Ingest test case wires expectations on.
 type ingestMocks struct {
-	counter   *countermock.MockCounter
-	factory   *scmock.MockFactory
-	sc        *scmock.MockSourceControl
-	reqStore  *storagemock.MockRequestStore
-	uriStore  *storagemock.MockRequestURIStore
-	publisher *mqmock.MockPublisher
+	counter    *countermock.MockCounter
+	factory    *scmock.MockFactory
+	sc         *scmock.MockSourceControl
+	reqStore   *storagemock.MockRequestStore
+	uriStore   *storagemock.MockRequestURIStore
+	queueStore *storagemock.MockQueueStore
+	publisher  *mqmock.MockPublisher
 }
 
 func newIngestController(t *testing.T, ctrl *gomock.Controller) (*IngestController, ingestMocks) {
 	t.Helper()
 
 	m := ingestMocks{
-		counter:   countermock.NewMockCounter(ctrl),
-		factory:   scmock.NewMockFactory(ctrl),
-		sc:        scmock.NewMockSourceControl(ctrl),
-		reqStore:  storagemock.NewMockRequestStore(ctrl),
-		uriStore:  storagemock.NewMockRequestURIStore(ctrl),
-		publisher: mqmock.NewMockPublisher(ctrl),
+		counter:    countermock.NewMockCounter(ctrl),
+		factory:    scmock.NewMockFactory(ctrl),
+		sc:         scmock.NewMockSourceControl(ctrl),
+		reqStore:   storagemock.NewMockRequestStore(ctrl),
+		uriStore:   storagemock.NewMockRequestURIStore(ctrl),
+		queueStore: storagemock.NewMockQueueStore(ctrl),
+		publisher:  mqmock.NewMockPublisher(ctrl),
 	}
 
 	store := storagemock.NewMockStorage(ctrl)
 	store.EXPECT().GetRequestStore().Return(m.reqStore).AnyTimes()
 	store.EXPECT().GetRequestURIStore().Return(m.uriStore).AnyTimes()
+	store.EXPECT().GetQueueStore().Return(m.queueStore).AnyTimes()
 
 	queue := mqmock.NewMockQueue(ctrl)
 	queue.EXPECT().Publisher().Return(m.publisher).AnyTimes()
@@ -85,6 +88,25 @@ func expectResolve(m ingestMocks) {
 	m.sc.EXPECT().Latest(gomock.Any()).Return(testURI, nil)
 }
 
+// expectAdvanceLatestRequestID wires GetOrCreate + Update for queue.latest_request_id.
+func expectAdvanceLatestRequestID(m ingestMocks, queue, id string) {
+	m.queueStore.EXPECT().GetOrCreate(gomock.Any(), queue, entity.Queue{Version: 1}).Return(entity.Queue{
+		Name:    queue,
+		Version: 1,
+	}, nil)
+	updated := entity.Queue{Name: queue, LatestRequestID: id, Version: 1}
+	m.queueStore.EXPECT().Update(gomock.Any(), updated, int32(1), int32(2)).Return(nil)
+}
+
+// expectAdvanceLatestRequestIDNoOp wires GetOrCreate when latest_request_id is already at id.
+func expectAdvanceLatestRequestIDNoOp(m ingestMocks, queue, id string) {
+	m.queueStore.EXPECT().GetOrCreate(gomock.Any(), queue, entity.Queue{Version: 1}).Return(entity.Queue{
+		Name:            queue,
+		LatestRequestID: id,
+		Version:         1,
+	}, nil)
+}
+
 func TestIngestController_Ingest(t *testing.T) {
 	tests := []struct {
 		name        string
@@ -104,6 +126,7 @@ func TestIngestController_Ingest(t *testing.T) {
 				m.uriStore.EXPECT().Create(gomock.Any(), testQueue, testURI, "request/monorepo/main/7").Return(nil)
 				m.reqStore.EXPECT().Get(gomock.Any(), "request/monorepo/main/7").Return(entity.Request{}, storage.ErrNotFound)
 				m.reqStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil)
+				expectAdvanceLatestRequestID(m, testQueue, "request/monorepo/main/7")
 				m.publisher.EXPECT().Publish(gomock.Any(), "process", gomock.Any()).Return(nil)
 			},
 			wantID: "request/monorepo/main/7",
@@ -115,6 +138,7 @@ func TestIngestController_Ingest(t *testing.T) {
 				expectResolve(m)
 				m.uriStore.EXPECT().GetIDByURI(gomock.Any(), testQueue, testURI).Return("request/monorepo/main/3", nil)
 				m.reqStore.EXPECT().Get(gomock.Any(), "request/monorepo/main/3").Return(entity.Request{ID: "request/monorepo/main/3", State: entity.RequestStateAccepted}, nil)
+				expectAdvanceLatestRequestIDNoOp(m, testQueue, "request/monorepo/main/3")
 				m.publisher.EXPECT().Publish(gomock.Any(), "process", gomock.Any()).Return(nil)
 			},
 			wantID: "request/monorepo/main/3",
@@ -128,6 +152,7 @@ func TestIngestController_Ingest(t *testing.T) {
 				m.uriStore.EXPECT().GetIDByURI(gomock.Any(), testQueue, testURI).Return("request/monorepo/main/3", nil)
 				m.reqStore.EXPECT().Get(gomock.Any(), "request/monorepo/main/3").Return(entity.Request{}, storage.ErrNotFound)
 				m.reqStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil)
+				expectAdvanceLatestRequestID(m, testQueue, "request/monorepo/main/3")
 				m.publisher.EXPECT().Publish(gomock.Any(), "process", gomock.Any()).Return(nil)
 			},
 			wantID: "request/monorepo/main/3",
@@ -142,6 +167,7 @@ func TestIngestController_Ingest(t *testing.T) {
 				m.uriStore.EXPECT().Create(gomock.Any(), testQueue, testURI, "request/monorepo/main/7").Return(storage.ErrAlreadyExists)
 				m.uriStore.EXPECT().GetIDByURI(gomock.Any(), testQueue, testURI).Return("request/monorepo/main/3", nil)
 				m.reqStore.EXPECT().Get(gomock.Any(), "request/monorepo/main/3").Return(entity.Request{ID: "request/monorepo/main/3", State: entity.RequestStateAccepted}, nil)
+				expectAdvanceLatestRequestIDNoOp(m, testQueue, "request/monorepo/main/3")
 				m.publisher.EXPECT().Publish(gomock.Any(), "process", gomock.Any()).Return(nil)
 			},
 			wantID: "request/monorepo/main/3",
@@ -205,6 +231,7 @@ func TestIngestController_Ingest(t *testing.T) {
 				m.uriStore.EXPECT().Create(gomock.Any(), testQueue, testURI, gomock.Any()).Return(nil)
 				m.reqStore.EXPECT().Get(gomock.Any(), gomock.Any()).Return(entity.Request{}, storage.ErrNotFound)
 				m.reqStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil)
+				expectAdvanceLatestRequestID(m, testQueue, "request/monorepo/main/7")
 				m.publisher.EXPECT().Publish(gomock.Any(), "process", gomock.Any()).Return(errors.New("queue down"))
 			},
 			wantErr: true,

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"
+	// BuildStrategyFullMonorepo validates the whole monorepo from scratch.
+	BuildStrategyFullMonorepo BuildStrategy = "full_monorepo"
 )
 
 // 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-monorepo 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)
+		})
+	}
+}