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,66 @@ func (c *IngestController) ensureRequest(ctx context.Context, id, queue, uri str
 	return request, nil
 }
 
+// ensureQueue returns the queue row for name, creating it if it does not yet exist.
+// A concurrent creator (ErrAlreadyExists) is resolved by re-reading the canonical row.
+func (c *IngestController) ensureQueue(ctx context.Context, name string) (entity.Queue, error) {
+	queueStore := c.store.GetQueueStore()
+
+	got, err := queueStore.Get(ctx, name)
+	if err == nil {
+		return got, nil
+	}
+	if !errors.Is(err, storage.ErrNotFound) {
+		return entity.Queue{}, fmt.Errorf("IngestController failed to load queue %s: %w", name, err)
+	}
+
+	queue := entity.Queue{
+		Name:    name,
+		Version: 1,
+	}
+	if err := queueStore.Create(ctx, queue); err != nil {
+		if !errors.Is(err, storage.ErrAlreadyExists) {
+			return entity.Queue{}, fmt.Errorf("IngestController failed to persist queue %s: %w", name, err)
+		}
+		// Raced with a concurrent creator; read the canonical row.
+		return queueStore.Get(ctx, name)
+	}
+	return queue, 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 := c.ensureQueue(ctx, queue)
+		if err != nil {
+			return 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,23 @@ func expectResolve(m ingestMocks) {
 	m.sc.EXPECT().Latest(gomock.Any()).Return(testURI, nil)
 }
 
+// expectAdvanceLatestRequestID wires Get + Create + Update for queue.latest_request_id.
+func expectAdvanceLatestRequestID(m ingestMocks, queue, id string) {
+	m.queueStore.EXPECT().Get(gomock.Any(), queue).Return(entity.Queue{}, storage.ErrNotFound)
+	m.queueStore.EXPECT().Create(gomock.Any(), entity.Queue{Name: queue, Version: 1}).Return(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 Get when latest_request_id is already at id.
+func expectAdvanceLatestRequestIDNoOp(m ingestMocks, queue, id string) {
+	m.queueStore.EXPECT().Get(gomock.Any(), queue).Return(entity.Queue{
+		Name:            queue,
+		LatestRequestID: id,
+		Version:         1,
+	}, nil)
+}
+
 func TestIngestController_Ingest(t *testing.T) {
 	tests := []struct {
 		name        string
@@ -104,6 +124,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 +136,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 +150,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 +165,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 +229,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,

test/integration/stovepipe/suite_test.go

@@ -136,6 +136,11 @@ func (s *StovepipeIntegrationSuite) TestIngestAPI() {
 	resp2, err := s.client.Ingest(s.ctx, &pb.IngestRequest{Queue: queue})
 	require.NoError(t, err, "second Ingest failed")
 	assert.Equal(t, id, resp2.Id, "re-ingest of the same head should dedup to the same id")
+
+	// latest_request_id on the queue row.
+	var latestRequestID string
+	require.NoError(t, s.db.QueryRow("SELECT latest_request_id FROM queue WHERE name = ?", queue).Scan(&latestRequestID))
+	assert.Equal(t, id, latestRequestID, "queue latest_request_id should point at the minted request")
 }
 
 // TestIngestEmptyQueue verifies the request-validation error surfaces over gRPC.