feat(stovepipe): add Queue.LastGreenRequestID in Stovepipe Queue

stovepipe/entity/queue.go

@@ -31,6 +31,11 @@ type Queue struct {
 	// whole-repo validation recorded green (health degree 0). Empty until the first such outcome.
 	LastGreenURI string `json:"last_green_uri"`
 
+	// LastGreenRequestID is the request that established LastGreenURI. The bookmark only
+	// moves forward: a green outcome adopts the pair only when this id is empty or older
+	// than the candidate's, compared by ingest order via CompareRequestID.
+	LastGreenRequestID string `json:"last_green_request_id"`
+
 	// InFlightCount is the number of trunk validations admitted by process but not yet terminal.
 	InFlightCount int32 `json:"in_flight_count"`
 

stovepipe/extension/storage/mysql/queue_store.go

@@ -49,13 +49,14 @@ func (q *queueStore) Create(ctx context.Context, queue entity.Queue) (retErr err
 	}
 
 	_, err := q.db.ExecContext(ctx,
-		`INSERT INTO queue (name, last_green_uri, in_flight_count, latest_request_id, version)
-		 VALUES (?, ?, ?, ?, ?)`,
+		`INSERT INTO queue (name, last_green_uri, in_flight_count, latest_request_id, version, last_green_request_id)
+		 VALUES (?, ?, ?, ?, ?, ?)`,
 		queue.Name,
 		queue.LastGreenURI,
 		queue.InFlightCount,
 		queue.LatestRequestID,
 		queue.Version,
+		queue.LastGreenRequestID,
 	)
 	if err != nil {
 		if isDuplicateEntry(err) {
@@ -77,14 +78,15 @@ 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_id, version FROM queue WHERE name = ?",
+		"SELECT name, last_green_uri, in_flight_count, latest_request_id, version, last_green_request_id FROM queue WHERE name = ?",
 		name,
 	).Scan(
 		&queue.Name,
 		&queue.LastGreenURI,
 		&queue.InFlightCount,
 		&queue.LatestRequestID,
 		&queue.Version,
+		&queue.LastGreenRequestID,
 	)
 
 	if errors.Is(err, sql.ErrNoRows) {
@@ -109,12 +111,13 @@ 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_id = ?, version = ?
+		 SET last_green_uri = ?, in_flight_count = ?, latest_request_id = ?, version = ?, last_green_request_id = ?
 		 WHERE name = ? AND version = ?`,
 		queue.LastGreenURI,
 		queue.InFlightCount,
 		queue.LatestRequestID,
 		newVersion,
+		queue.LastGreenRequestID,
 		queue.Name,
 		oldVersion,
 	)

stovepipe/extension/storage/mysql/queue_store_test.go

@@ -42,11 +42,12 @@ func setupQueueStoreTest(t *testing.T) (*sql.DB, sqlmock.Sqlmock, storage.QueueS
 
 func TestQueueStore_Create(t *testing.T) {
 	queue := entity.Queue{
-		Name:            "monorepo/main",
-		LastGreenURI:    "git://remote/monorepo/main/green",
-		InFlightCount:   0,
-		LatestRequestID: "request/monorepo/main/1",
-		Version:         1,
+		Name:               "monorepo/main",
+		LastGreenURI:       "git://remote/monorepo/main/green",
+		LastGreenRequestID: "request/monorepo/main/1",
+		InFlightCount:      0,
+		LatestRequestID:    "request/monorepo/main/1",
+		Version:            1,
 	}
 
 	tests := []struct {
@@ -59,15 +60,15 @@ func TestQueueStore_Create(t *testing.T) {
 			name: "success",
 			setup: func(mock sqlmock.Sqlmock) {
 				mock.ExpectExec("INSERT INTO queue").
-					WithArgs(queue.Name, queue.LastGreenURI, queue.InFlightCount, queue.LatestRequestID, queue.Version).
+					WithArgs(queue.Name, queue.LastGreenURI, queue.InFlightCount, queue.LatestRequestID, queue.Version, queue.LastGreenRequestID).
 					WillReturnResult(sqlmock.NewResult(0, 1))
 			},
 		},
 		{
 			name: "duplicate name returns ErrAlreadyExists",
 			setup: func(mock sqlmock.Sqlmock) {
 				mock.ExpectExec("INSERT INTO queue").
-					WithArgs(queue.Name, queue.LastGreenURI, queue.InFlightCount, queue.LatestRequestID, queue.Version).
+					WithArgs(queue.Name, queue.LastGreenURI, queue.InFlightCount, queue.LatestRequestID, queue.Version, queue.LastGreenRequestID).
 					WillReturnError(&mysql.MySQLError{Number: mysqlErrDuplicateEntry})
 			},
 			wantErr:   true,
@@ -77,7 +78,7 @@ func TestQueueStore_Create(t *testing.T) {
 			name: "other exec error",
 			setup: func(mock sqlmock.Sqlmock) {
 				mock.ExpectExec("INSERT INTO queue").
-					WithArgs(queue.Name, queue.LastGreenURI, queue.InFlightCount, queue.LatestRequestID, queue.Version).
+					WithArgs(queue.Name, queue.LastGreenURI, queue.InFlightCount, queue.LatestRequestID, queue.Version, queue.LastGreenRequestID).
 					WillReturnError(fmt.Errorf("connection reset"))
 			},
 			wantErr: true,
@@ -107,11 +108,12 @@ func TestQueueStore_Create(t *testing.T) {
 
 func TestQueueStore_Get(t *testing.T) {
 	want := entity.Queue{
-		Name:            "monorepo/main",
-		LastGreenURI:    "git://remote/monorepo/main/green",
-		InFlightCount:   2,
-		LatestRequestID: "request/monorepo/main/3",
-		Version:         3,
+		Name:               "monorepo/main",
+		LastGreenURI:       "git://remote/monorepo/main/green",
+		LastGreenRequestID: "request/monorepo/main/2",
+		InFlightCount:      2,
+		LatestRequestID:    "request/monorepo/main/3",
+		Version:            3,
 	}
 
 	tests := []struct {
@@ -126,9 +128,9 @@ func TestQueueStore_Get(t *testing.T) {
 			name:      "found",
 			queueName: want.Name,
 			setup: func(mock sqlmock.Sqlmock) {
-				rows := sqlmock.NewRows([]string{"name", "last_green_uri", "in_flight_count", "latest_request_id", "version"}).
-					AddRow(want.Name, want.LastGreenURI, want.InFlightCount, want.LatestRequestID, want.Version)
-				mock.ExpectQuery("SELECT name, last_green_uri, in_flight_count, latest_request_id, version FROM queue").
+				rows := sqlmock.NewRows([]string{"name", "last_green_uri", "in_flight_count", "latest_request_id", "version", "last_green_request_id"}).
+					AddRow(want.Name, want.LastGreenURI, want.InFlightCount, want.LatestRequestID, want.Version, want.LastGreenRequestID)
+				mock.ExpectQuery("SELECT name, last_green_uri, in_flight_count, latest_request_id, version, last_green_request_id FROM queue").
 					WithArgs(want.Name).
 					WillReturnRows(rows)
 			},
@@ -138,7 +140,7 @@ func TestQueueStore_Get(t *testing.T) {
 			name:      "not found",
 			queueName: want.Name,
 			setup: func(mock sqlmock.Sqlmock) {
-				mock.ExpectQuery("SELECT name, last_green_uri, in_flight_count, latest_request_id, version FROM queue").
+				mock.ExpectQuery("SELECT name, last_green_uri, in_flight_count, latest_request_id, version, last_green_request_id FROM queue").
 					WithArgs(want.Name).
 					WillReturnError(sql.ErrNoRows)
 			},
@@ -149,7 +151,7 @@ func TestQueueStore_Get(t *testing.T) {
 			name:      "query error",
 			queueName: want.Name,
 			setup: func(mock sqlmock.Sqlmock) {
-				mock.ExpectQuery("SELECT name, last_green_uri, in_flight_count, latest_request_id, version FROM queue").
+				mock.ExpectQuery("SELECT name, last_green_uri, in_flight_count, latest_request_id, version, last_green_request_id FROM queue").
 					WithArgs(want.Name).
 					WillReturnError(fmt.Errorf("connection reset"))
 			},
@@ -181,10 +183,11 @@ func TestQueueStore_Get(t *testing.T) {
 
 func TestQueueStore_Update(t *testing.T) {
 	queue := entity.Queue{
-		Name:            "monorepo/main",
-		LastGreenURI:    "git://remote/monorepo/main/green",
-		InFlightCount:   1,
-		LatestRequestID: "request/monorepo/main/2",
+		Name:               "monorepo/main",
+		LastGreenURI:       "git://remote/monorepo/main/green",
+		LastGreenRequestID: "request/monorepo/main/1",
+		InFlightCount:      1,
+		LatestRequestID:    "request/monorepo/main/2",
 	}
 	const oldVersion, newVersion = int32(1), int32(2)
 
@@ -198,15 +201,15 @@ func TestQueueStore_Update(t *testing.T) {
 			name: "success",
 			setup: func(mock sqlmock.Sqlmock) {
 				mock.ExpectExec("UPDATE queue").
-					WithArgs(queue.LastGreenURI, queue.InFlightCount, queue.LatestRequestID, newVersion, queue.Name, oldVersion).
+					WithArgs(queue.LastGreenURI, queue.InFlightCount, queue.LatestRequestID, newVersion, queue.LastGreenRequestID, queue.Name, oldVersion).
 					WillReturnResult(sqlmock.NewResult(0, 1))
 			},
 		},
 		{
 			name: "version mismatch",
 			setup: func(mock sqlmock.Sqlmock) {
 				mock.ExpectExec("UPDATE queue").
-					WithArgs(queue.LastGreenURI, queue.InFlightCount, queue.LatestRequestID, newVersion, queue.Name, oldVersion).
+					WithArgs(queue.LastGreenURI, queue.InFlightCount, queue.LatestRequestID, newVersion, queue.LastGreenRequestID, queue.Name, oldVersion).
 					WillReturnResult(sqlmock.NewResult(0, 0))
 			},
 			wantErr:   true,
@@ -216,7 +219,7 @@ func TestQueueStore_Update(t *testing.T) {
 			name: "exec error",
 			setup: func(mock sqlmock.Sqlmock) {
 				mock.ExpectExec("UPDATE queue").
-					WithArgs(queue.LastGreenURI, queue.InFlightCount, queue.LatestRequestID, newVersion, queue.Name, oldVersion).
+					WithArgs(queue.LastGreenURI, queue.InFlightCount, queue.LatestRequestID, newVersion, queue.LastGreenRequestID, queue.Name, oldVersion).
 					WillReturnError(fmt.Errorf("connection reset"))
 			},
 			wantErr: true,
@@ -225,7 +228,7 @@ func TestQueueStore_Update(t *testing.T) {
 			name: "rows affected error",
 			setup: func(mock sqlmock.Sqlmock) {
 				mock.ExpectExec("UPDATE queue").
-					WithArgs(queue.LastGreenURI, queue.InFlightCount, queue.LatestRequestID, newVersion, queue.Name, oldVersion).
+					WithArgs(queue.LastGreenURI, queue.InFlightCount, queue.LatestRequestID, newVersion, queue.LastGreenRequestID, queue.Name, oldVersion).
 					WillReturnResult(sqlmock.NewErrorResult(fmt.Errorf("driver error")))
 			},
 			wantErr: true,

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

@@ -1,10 +1,11 @@
 -- queue holds per-queue coordination state for the validation pipeline: the last-green
 -- 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_id  VARCHAR(255) NOT NULL DEFAULT '',
-    version            INT          NOT NULL,
+    name                   VARCHAR(255) NOT NULL,
+    last_green_uri         VARCHAR(255) NOT NULL DEFAULT '',
+    in_flight_count        INT          NOT NULL DEFAULT 0,
+    latest_request_id      VARCHAR(255) NOT NULL DEFAULT '',
+    version                INT          NOT NULL,
+    last_green_request_id  VARCHAR(255) NOT NULL DEFAULT '',
     PRIMARY KEY (name)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

test/integration/stovepipe/extension/storage/suite.go

@@ -87,10 +87,11 @@ func (s *QueueStoreContractSuite) TestQueueStore_CreateWithFields() {
 	const name = "contract/defaults"
 
 	toCreate := entity.Queue{
-		Name:            name,
-		LastGreenURI:    "git://remote/monorepo/main/green-bbbb",
-		LatestRequestID: "request/contract/defaults/99",
-		Version:         1,
+		Name:               name,
+		LastGreenURI:       "git://remote/monorepo/main/green-bbbb",
+		LastGreenRequestID: "request/contract/defaults/98",
+		LatestRequestID:    "request/contract/defaults/99",
+		Version:            1,
 	}
 	require.NoError(t, s.storeFor(name).Create(s.ctx, toCreate))
 
@@ -138,13 +139,15 @@ func (s *QueueStoreContractSuite) TestQueueStore_UpdateCAS() {
 
 	updated := created
 	updated.LastGreenURI = "git://remote/monorepo/main/green-cccc"
+	updated.LastGreenRequestID = "request/contract/update-cas/41"
 	updated.LatestRequestID = "request/contract/update-cas/42"
 	updated.InFlightCount = 1
 	require.NoError(t, s.storeFor(name).Update(s.ctx, updated, 1, 2))
 
 	got, err := s.storeFor(name).Get(s.ctx, name)
 	require.NoError(t, err)
 	assert.Equal(t, updated.LastGreenURI, got.LastGreenURI)
+	assert.Equal(t, "request/contract/update-cas/41", got.LastGreenRequestID)
 	assert.Equal(t, "request/contract/update-cas/42", got.LatestRequestID)
 	assert.Equal(t, int32(1), got.InFlightCount)
 	assert.Equal(t, int32(2), got.Version)