diff --git a/stovepipe/entity/queue.go b/stovepipe/entity/queue.go index f45eb953..23d31efb 100644 --- a/stovepipe/entity/queue.go +++ b/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"` diff --git a/stovepipe/extension/storage/mysql/queue_store.go b/stovepipe/extension/storage/mysql/queue_store.go index f82a9e1e..aaf3cb16 100644 --- a/stovepipe/extension/storage/mysql/queue_store.go +++ b/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,7 +78,7 @@ 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, @@ -85,6 +86,7 @@ func (q *queueStore) Get(ctx context.Context, name string) (ret entity.Queue, re &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, ) diff --git a/stovepipe/extension/storage/mysql/queue_store_test.go b/stovepipe/extension/storage/mysql/queue_store_test.go index d5507ea8..4bf6be9a 100644 --- a/stovepipe/extension/storage/mysql/queue_store_test.go +++ b/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,7 +60,7 @@ 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)) }, }, @@ -67,7 +68,7 @@ func TestQueueStore_Create(t *testing.T) { 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,7 +201,7 @@ 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)) }, }, @@ -206,7 +209,7 @@ func TestQueueStore_Update(t *testing.T) { 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, diff --git a/stovepipe/extension/storage/mysql/schema/queue.sql b/stovepipe/extension/storage/mysql/schema/queue.sql index 872e5af3..c10463f0 100644 --- a/stovepipe/extension/storage/mysql/schema/queue.sql +++ b/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; diff --git a/test/integration/stovepipe/extension/storage/suite.go b/test/integration/stovepipe/extension/storage/suite.go index 7d636db9..922a0fdf 100644 --- a/test/integration/stovepipe/extension/storage/suite.go +++ b/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,6 +139,7 @@ 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)) @@ -145,6 +147,7 @@ func (s *QueueStoreContractSuite) TestQueueStore_UpdateCAS() { 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)