diff --git a/stovepipe/entity/BUILD.bazel b/stovepipe/entity/BUILD.bazel index d4798469..80ddde0b 100644 --- a/stovepipe/entity/BUILD.bazel +++ b/stovepipe/entity/BUILD.bazel @@ -11,6 +11,7 @@ go_library( "request_history.go", "request_id.go", "request_log.go", + "request_summary.go", "validation_fact.go", ], importpath = "github.com/uber/submitqueue/stovepipe/entity", diff --git a/stovepipe/entity/request_summary.go b/stovepipe/entity/request_summary.go new file mode 100644 index 00000000..2a217956 --- /dev/null +++ b/stovepipe/entity/request_summary.go @@ -0,0 +1,35 @@ +// Copyright (c) 2026 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 + +// RequestSummary is the materialized current view of a validation request. +type RequestSummary struct { + // RequestID is the globally unique request identifier. + RequestID string + // Queue is the queue containing the request. + Queue string + // URI is the commit validated by the request. + URI string + // BaseURI is the incremental-validation baseline and is empty before selection or for a full build. + BaseURI string + // State is the current durable request lifecycle state. + State RequestState + // RequestVersion is the version of the request state represented by this summary. + RequestVersion int32 + // StateTimestampMs is when the represented state was first retained, in Unix milliseconds. + StateTimestampMs int64 + // Version is the optimistic-lock version of this materialized view. + Version int32 +} diff --git a/stovepipe/extension/storage/BUILD.bazel b/stovepipe/extension/storage/BUILD.bazel index 02056713..4ba45797 100644 --- a/stovepipe/extension/storage/BUILD.bazel +++ b/stovepipe/extension/storage/BUILD.bazel @@ -7,6 +7,7 @@ go_library( "queue_store.go", "request_log_store.go", "request_store.go", + "request_summary_store.go", "request_uri_store.go", "storage.go", "validation_fact_store.go", diff --git a/stovepipe/extension/storage/mock/BUILD.bazel b/stovepipe/extension/storage/mock/BUILD.bazel index f1796eef..d8e636c7 100644 --- a/stovepipe/extension/storage/mock/BUILD.bazel +++ b/stovepipe/extension/storage/mock/BUILD.bazel @@ -7,6 +7,7 @@ go_library( "queue_store_mock.go", "request_log_store_mock.go", "request_store_mock.go", + "request_summary_store_mock.go", "request_uri_store_mock.go", "storage_mock.go", "validation_fact_store_mock.go", diff --git a/stovepipe/extension/storage/mock/request_summary_store_mock.go b/stovepipe/extension/storage/mock/request_summary_store_mock.go new file mode 100644 index 00000000..5f688846 --- /dev/null +++ b/stovepipe/extension/storage/mock/request_summary_store_mock.go @@ -0,0 +1,85 @@ +// Code generated by MockGen. DO NOT EDIT. +// Source: request_summary_store.go +// +// Generated by this command: +// +// mockgen -source=request_summary_store.go -destination=mock/request_summary_store_mock.go -package=mock +// + +// Package mock is a generated GoMock package. +package mock + +import ( + context "context" + reflect "reflect" + + entity "github.com/uber/submitqueue/stovepipe/entity" + gomock "go.uber.org/mock/gomock" +) + +// MockRequestSummaryStore is a mock of RequestSummaryStore interface. +type MockRequestSummaryStore struct { + ctrl *gomock.Controller + recorder *MockRequestSummaryStoreMockRecorder + isgomock struct{} +} + +// MockRequestSummaryStoreMockRecorder is the mock recorder for MockRequestSummaryStore. +type MockRequestSummaryStoreMockRecorder struct { + mock *MockRequestSummaryStore +} + +// NewMockRequestSummaryStore creates a new mock instance. +func NewMockRequestSummaryStore(ctrl *gomock.Controller) *MockRequestSummaryStore { + mock := &MockRequestSummaryStore{ctrl: ctrl} + mock.recorder = &MockRequestSummaryStoreMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *MockRequestSummaryStore) EXPECT() *MockRequestSummaryStoreMockRecorder { + return m.recorder +} + +// Create mocks base method. +func (m *MockRequestSummaryStore) Create(ctx context.Context, summary entity.RequestSummary) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Create", ctx, summary) + ret0, _ := ret[0].(error) + return ret0 +} + +// Create indicates an expected call of Create. +func (mr *MockRequestSummaryStoreMockRecorder) Create(ctx, summary any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Create", reflect.TypeOf((*MockRequestSummaryStore)(nil).Create), ctx, summary) +} + +// Get mocks base method. +func (m *MockRequestSummaryStore) Get(ctx context.Context, requestID string) (entity.RequestSummary, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Get", ctx, requestID) + ret0, _ := ret[0].(entity.RequestSummary) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// Get indicates an expected call of Get. +func (mr *MockRequestSummaryStoreMockRecorder) Get(ctx, requestID any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Get", reflect.TypeOf((*MockRequestSummaryStore)(nil).Get), ctx, requestID) +} + +// Update mocks base method. +func (m *MockRequestSummaryStore) Update(ctx context.Context, summary entity.RequestSummary, oldVersion, newVersion int32) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Update", ctx, summary, oldVersion, newVersion) + ret0, _ := ret[0].(error) + return ret0 +} + +// Update indicates an expected call of Update. +func (mr *MockRequestSummaryStoreMockRecorder) Update(ctx, summary, oldVersion, newVersion any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Update", reflect.TypeOf((*MockRequestSummaryStore)(nil).Update), ctx, summary, oldVersion, newVersion) +} diff --git a/stovepipe/extension/storage/mock/storage_mock.go b/stovepipe/extension/storage/mock/storage_mock.go index eb901e6f..95d0a602 100644 --- a/stovepipe/extension/storage/mock/storage_mock.go +++ b/stovepipe/extension/storage/mock/storage_mock.go @@ -135,6 +135,20 @@ func (mr *MockStorageMockRecorder) GetRequestStore() *gomock.Call { return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetRequestStore", reflect.TypeOf((*MockStorage)(nil).GetRequestStore)) } +// GetRequestSummaryStore mocks base method. +func (m *MockStorage) GetRequestSummaryStore() storage.RequestSummaryStore { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "GetRequestSummaryStore") + ret0, _ := ret[0].(storage.RequestSummaryStore) + return ret0 +} + +// GetRequestSummaryStore indicates an expected call of GetRequestSummaryStore. +func (mr *MockStorageMockRecorder) GetRequestSummaryStore() *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetRequestSummaryStore", reflect.TypeOf((*MockStorage)(nil).GetRequestSummaryStore)) +} + // GetRequestURIStore mocks base method. func (m *MockStorage) GetRequestURIStore() storage.RequestURIStore { m.ctrl.T.Helper() diff --git a/stovepipe/extension/storage/mysql/BUILD.bazel b/stovepipe/extension/storage/mysql/BUILD.bazel index bc875e5d..7f85a564 100644 --- a/stovepipe/extension/storage/mysql/BUILD.bazel +++ b/stovepipe/extension/storage/mysql/BUILD.bazel @@ -7,6 +7,7 @@ go_library( "queue_store.go", "request_log_store.go", "request_store.go", + "request_summary_store.go", "request_uri_store.go", "storage.go", "validation_fact_store.go", @@ -29,6 +30,7 @@ go_test( "queue_store_test.go", "request_log_store_test.go", "request_store_test.go", + "request_summary_store_test.go", "request_uri_store_test.go", "storage_test.go", "validation_fact_store_test.go", diff --git a/stovepipe/extension/storage/mysql/request_summary_store.go b/stovepipe/extension/storage/mysql/request_summary_store.go new file mode 100644 index 00000000..697adc77 --- /dev/null +++ b/stovepipe/extension/storage/mysql/request_summary_store.go @@ -0,0 +1,131 @@ +// Copyright (c) 2026 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 mysql + +import ( + "context" + "database/sql" + "errors" + "fmt" + + "github.com/uber-go/tally" + + "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/storage" +) + +type requestSummaryStore struct { + db *sql.DB + scope tally.Scope + queue string +} + +// NewRequestSummaryStore creates a MySQL-backed RequestSummaryStore. +func NewRequestSummaryStore(db *sql.DB, scope tally.Scope, queue string) storage.RequestSummaryStore { + return &requestSummaryStore{db: db, scope: scope, queue: queue} +} + +func (s *requestSummaryStore) Create(ctx context.Context, summary entity.RequestSummary) (retErr error) { + op := metrics.Begin(s.scope, "create", metrics.StorageLatencyBuckets) + defer func() { op.Complete(retErr) }() + + if summary.Queue != s.queue { + return fmt.Errorf("request summary %q queue %q does not match the store's bound queue %q", summary.RequestID, summary.Queue, s.queue) + } + + _, err := s.db.ExecContext(ctx, ` + INSERT INTO request_summary ( + queue, request_id, uri, base_uri, state, request_version, state_timestamp_ms, version + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`, + summary.Queue, + summary.RequestID, + summary.URI, + summary.BaseURI, + summary.State, + summary.RequestVersion, + summary.StateTimestampMs, + summary.Version, + ) + if err != nil { + if isDuplicateEntry(err) { + return fmt.Errorf("request summary request_id=%q: %w", summary.RequestID, storage.ErrAlreadyExists) + } + return fmt.Errorf("failed to insert request summary request_id=%q: %w", summary.RequestID, err) + } + return nil +} + +func (s *requestSummaryStore) Get(ctx context.Context, requestID string) (ret entity.RequestSummary, retErr error) { + op := metrics.Begin(s.scope, "get", metrics.StorageLatencyBuckets) + defer func() { op.Complete(retErr) }() + + err := s.db.QueryRowContext(ctx, ` + SELECT queue, request_id, uri, base_uri, state, request_version, state_timestamp_ms, version + FROM request_summary + WHERE queue = ? AND request_id = ?`, + s.queue, requestID, + ).Scan( + &ret.Queue, + &ret.RequestID, + &ret.URI, + &ret.BaseURI, + &ret.State, + &ret.RequestVersion, + &ret.StateTimestampMs, + &ret.Version, + ) + if errors.Is(err, sql.ErrNoRows) { + return entity.RequestSummary{}, storage.WrapNotFound(err) + } + if err != nil { + return entity.RequestSummary{}, fmt.Errorf("failed to get request summary request_id=%q: %w", requestID, err) + } + return ret, nil +} + +func (s *requestSummaryStore) Update(ctx context.Context, summary entity.RequestSummary, oldVersion, newVersion int32) (retErr error) { + op := metrics.Begin(s.scope, "update", metrics.StorageLatencyBuckets) + defer func() { op.Complete(retErr) }() + + if summary.Queue != s.queue { + return fmt.Errorf("request summary %q queue %q does not match the store's bound queue %q", summary.RequestID, summary.Queue, s.queue) + } + + result, err := s.db.ExecContext(ctx, ` + UPDATE request_summary + SET base_uri = ?, state = ?, request_version = ?, state_timestamp_ms = ?, version = ? + WHERE queue = ? AND request_id = ? AND version = ?`, + summary.BaseURI, + summary.State, + summary.RequestVersion, + summary.StateTimestampMs, + newVersion, + summary.Queue, + summary.RequestID, + oldVersion, + ) + if err != nil { + return fmt.Errorf("failed to update request summary request_id=%q old_version=%d new_version=%d: %w", summary.RequestID, oldVersion, newVersion, err) + } + rowsAffected, err := result.RowsAffected() + if err != nil { + return fmt.Errorf("failed to get request summary update rows request_id=%q: %w", summary.RequestID, err) + } + if rowsAffected != 1 { + return fmt.Errorf("request summary request_id=%q expected_version=%d: %w", summary.RequestID, oldVersion, storage.ErrVersionMismatch) + } + return nil +} diff --git a/stovepipe/extension/storage/mysql/request_summary_store_test.go b/stovepipe/extension/storage/mysql/request_summary_store_test.go new file mode 100644 index 00000000..dfdc3c53 --- /dev/null +++ b/stovepipe/extension/storage/mysql/request_summary_store_test.go @@ -0,0 +1,200 @@ +// Copyright (c) 2026 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 mysql + +import ( + "context" + "database/sql" + "fmt" + "testing" + + "github.com/DATA-DOG/go-sqlmock" + "github.com/go-sql-driver/mysql" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/storage" +) + +const testRequestSummaryQueue = "monorepo/main" + +func testRequestSummary() entity.RequestSummary { + return entity.RequestSummary{ + RequestID: "request/monorepo/main/1", + Queue: testRequestSummaryQueue, + URI: "git://repo/head", + BaseURI: "git://repo/base", + State: entity.RequestStateProcessing, + RequestVersion: 2, + StateTimestampMs: 1735689600000, + Version: 1, + } +} + +func setupRequestSummaryStoreTest(t *testing.T) (*sql.DB, sqlmock.Sqlmock, storage.RequestSummaryStore) { + t.Helper() + db, mock, err := sqlmock.New() + require.NoError(t, err) + return db, mock, NewRequestSummaryStore(db, testMetrics(), testRequestSummaryQueue) +} + +func TestRequestSummaryStoreCreate(t *testing.T) { + summary := testRequestSummary() + tests := []struct { + name string + summary entity.RequestSummary + setup func(sqlmock.Sqlmock) + wantErrIs error + wantErr bool + }{ + { + name: "success", + summary: summary, + setup: func(mock sqlmock.Sqlmock) { + mock.ExpectExec("INSERT INTO request_summary"). + WithArgs(summary.Queue, summary.RequestID, summary.URI, summary.BaseURI, summary.State, summary.RequestVersion, summary.StateTimestampMs, summary.Version). + WillReturnResult(sqlmock.NewResult(0, 1)) + }, + }, + { + name: "duplicate", + summary: summary, + setup: func(mock sqlmock.Sqlmock) { + mock.ExpectExec("INSERT INTO request_summary"). + WithArgs(summary.Queue, summary.RequestID, summary.URI, summary.BaseURI, summary.State, summary.RequestVersion, summary.StateTimestampMs, summary.Version). + WillReturnError(&mysql.MySQLError{Number: mysqlErrDuplicateEntry}) + }, + wantErr: true, + wantErrIs: storage.ErrAlreadyExists, + }, + { + name: "wrong queue", + summary: func() entity.RequestSummary { + other := summary + other.Queue = "other" + return other + }(), + wantErr: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + db, mock, store := setupRequestSummaryStoreTest(t) + defer db.Close() + if tt.setup != nil { + tt.setup(mock) + } + + err := store.Create(context.Background(), tt.summary) + if tt.wantErr { + require.Error(t, err) + if tt.wantErrIs != nil { + assert.ErrorIs(t, err, tt.wantErrIs) + } + } else { + require.NoError(t, err) + } + require.NoError(t, mock.ExpectationsWereMet()) + }) + } +} + +func TestRequestSummaryStoreGet(t *testing.T) { + want := testRequestSummary() + tests := []struct { + name string + setup func(sqlmock.Sqlmock) + wantErrIs error + wantErr bool + }{ + { + name: "found", + setup: func(mock sqlmock.Sqlmock) { + rows := sqlmock.NewRows([]string{"queue", "request_id", "uri", "base_uri", "state", "request_version", "state_timestamp_ms", "version"}). + AddRow(want.Queue, want.RequestID, want.URI, want.BaseURI, want.State, want.RequestVersion, want.StateTimestampMs, want.Version) + mock.ExpectQuery("SELECT queue, request_id, uri, base_uri, state").WithArgs(testRequestSummaryQueue, want.RequestID).WillReturnRows(rows) + }, + }, + { + name: "not found", + setup: func(mock sqlmock.Sqlmock) { + mock.ExpectQuery("SELECT queue, request_id, uri, base_uri, state").WithArgs(testRequestSummaryQueue, want.RequestID).WillReturnError(sql.ErrNoRows) + }, + wantErr: true, + wantErrIs: storage.ErrNotFound, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + db, mock, store := setupRequestSummaryStoreTest(t) + defer db.Close() + tt.setup(mock) + + got, err := store.Get(context.Background(), want.RequestID) + if tt.wantErr { + require.Error(t, err) + assert.ErrorIs(t, err, tt.wantErrIs) + } else { + require.NoError(t, err) + assert.Equal(t, want, got) + } + require.NoError(t, mock.ExpectationsWereMet()) + }) + } +} + +func TestRequestSummaryStoreUpdate(t *testing.T) { + summary := testRequestSummary() + const oldVersion, newVersion = int32(1), int32(2) + tests := []struct { + name string + result sql.Result + execErr error + wantErrIs error + wantErr bool + }{ + {name: "success", result: sqlmock.NewResult(0, 1)}, + {name: "version mismatch", result: sqlmock.NewResult(0, 0), wantErr: true, wantErrIs: storage.ErrVersionMismatch}, + {name: "database failure", execErr: fmt.Errorf("connection reset"), wantErr: true}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + db, mock, store := setupRequestSummaryStoreTest(t) + defer db.Close() + expectation := mock.ExpectExec("UPDATE request_summary"). + WithArgs(summary.BaseURI, summary.State, summary.RequestVersion, summary.StateTimestampMs, newVersion, summary.Queue, summary.RequestID, oldVersion) + if tt.execErr != nil { + expectation.WillReturnError(tt.execErr) + } else { + expectation.WillReturnResult(tt.result) + } + + err := store.Update(context.Background(), summary, oldVersion, newVersion) + if tt.wantErr { + require.Error(t, err) + if tt.wantErrIs != nil { + assert.ErrorIs(t, err, tt.wantErrIs) + } + } else { + require.NoError(t, err) + } + require.NoError(t, mock.ExpectationsWereMet()) + }) + } +} diff --git a/stovepipe/extension/storage/mysql/schema/request_summary.sql b/stovepipe/extension/storage/mysql/schema/request_summary.sql new file mode 100644 index 00000000..ce096a51 --- /dev/null +++ b/stovepipe/extension/storage/mysql/schema/request_summary.sql @@ -0,0 +1,14 @@ +-- request_summary is the request-ID-keyed materialized view used by public status reads. The +-- request version identifies the winning lifecycle state; version guards concurrent projection +-- writers. queue leads the primary key so the table remains shardable by queue. +CREATE TABLE IF NOT EXISTS request_summary ( + queue VARCHAR(255) NOT NULL, + request_id VARCHAR(255) NOT NULL, + uri VARCHAR(255) NOT NULL, + base_uri VARCHAR(255) NOT NULL, + state VARCHAR(64) NOT NULL, + request_version INT NOT NULL, + state_timestamp_ms BIGINT NOT NULL, + version INT NOT NULL, + PRIMARY KEY (queue, request_id) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; diff --git a/stovepipe/extension/storage/mysql/storage.go b/stovepipe/extension/storage/mysql/storage.go index c25b8a32..b1d3476a 100644 --- a/stovepipe/extension/storage/mysql/storage.go +++ b/stovepipe/extension/storage/mysql/storage.go @@ -48,6 +48,7 @@ func (s *Storage) For(queueName string) (storage.Storage, error) { requestStore: NewRequestStore(s.db, s.scope.SubScope("request_store"), queueName), requestURIStore: NewRequestURIStore(s.db, s.scope.SubScope("request_uri_store"), queueName), requestLogStore: NewRequestLogStore(s.db, s.scope.SubScope("request_log_store"), queueName), + requestSummaryStore: NewRequestSummaryStore(s.db, s.scope.SubScope("request_summary_store"), queueName), queueStore: NewQueueStore(s.db, s.scope.SubScope("queue_store"), queueName), buildStore: NewBuildStore(s.db, s.scope.SubScope("build_store"), queueName), validationFactStore: NewValidationFactStore(s.db, s.scope.SubScope("validation_fact_store"), queueName), @@ -64,6 +65,7 @@ type mysqlStorage struct { requestStore storage.RequestStore requestURIStore storage.RequestURIStore requestLogStore storage.RequestLogStore + requestSummaryStore storage.RequestSummaryStore queueStore storage.QueueStore buildStore storage.BuildStore validationFactStore storage.ValidationFactStore @@ -87,6 +89,11 @@ func (f *mysqlStorage) GetRequestLogStore() storage.RequestLogStore { return f.requestLogStore } +// GetRequestSummaryStore returns the MySQL-backed RequestSummaryStore. +func (f *mysqlStorage) GetRequestSummaryStore() storage.RequestSummaryStore { + return f.requestSummaryStore +} + // GetQueueStore returns the MySQL-backed QueueStore. func (f *mysqlStorage) GetQueueStore() storage.QueueStore { return f.queueStore diff --git a/stovepipe/extension/storage/mysql/storage_test.go b/stovepipe/extension/storage/mysql/storage_test.go index bbd8b6fd..bba83594 100644 --- a/stovepipe/extension/storage/mysql/storage_test.go +++ b/stovepipe/extension/storage/mysql/storage_test.go @@ -41,6 +41,7 @@ func TestNewStorage(t *testing.T) { assert.NotNil(t, bound.GetRequestStore()) assert.NotNil(t, bound.GetRequestURIStore()) assert.NotNil(t, bound.GetRequestLogStore()) + assert.NotNil(t, bound.GetRequestSummaryStore()) assert.NotNil(t, bound.GetQueueStore()) assert.NotNil(t, bound.GetBuildStore()) diff --git a/stovepipe/extension/storage/request_summary_store.go b/stovepipe/extension/storage/request_summary_store.go new file mode 100644 index 00000000..7a800036 --- /dev/null +++ b/stovepipe/extension/storage/request_summary_store.go @@ -0,0 +1,35 @@ +// Copyright (c) 2026 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 storage + +//go:generate mockgen -source=request_summary_store.go -destination=mock/request_summary_store_mock.go -package=mock + +import ( + "context" + + "github.com/uber/submitqueue/stovepipe/entity" +) + +// RequestSummaryStore persists queue-scoped request status projections. +type RequestSummaryStore interface { + // Create persists summary and returns ErrAlreadyExists when its identity exists. + Create(ctx context.Context, summary entity.RequestSummary) error + + // Get returns the summary identified by requestID, or ErrNotFound when absent. + Get(ctx context.Context, requestID string) (entity.RequestSummary, error) + + // Update conditionally replaces summary using the supplied optimistic-lock versions. + Update(ctx context.Context, summary entity.RequestSummary, oldVersion, newVersion int32) error +} diff --git a/stovepipe/extension/storage/storage.go b/stovepipe/extension/storage/storage.go index f9470f88..20a0149b 100644 --- a/stovepipe/extension/storage/storage.go +++ b/stovepipe/extension/storage/storage.go @@ -78,6 +78,9 @@ type Storage interface { // GetRequestLogStore returns the RequestLogStore instance. GetRequestLogStore() RequestLogStore + // GetRequestSummaryStore returns the RequestSummaryStore instance. + GetRequestSummaryStore() RequestSummaryStore + // GetQueueStore returns the QueueStore instance. GetQueueStore() QueueStore diff --git a/test/integration/stovepipe/extension/storage/mysql/storage_test.go b/test/integration/stovepipe/extension/storage/mysql/storage_test.go index 242030d4..ab24ebb3 100644 --- a/test/integration/stovepipe/extension/storage/mysql/storage_test.go +++ b/test/integration/stovepipe/extension/storage/mysql/storage_test.go @@ -102,6 +102,16 @@ func TestMySQLStorage(t *testing.T) { testSuite.SetFactory(factory) suite.Run(t, testSuite) }) + + t.Run("RequestSummaryStore", func(t *testing.T) { + resetStorage(t, db) + bound, err := backend.For("monorepo/main") + require.NoError(t, err) + suite.Run(t, &MySQLRequestSummaryStoreSuite{ + ctx: ctx, + store: bound.GetRequestSummaryStore(), + }) + }) } func resetStorage(t *testing.T, db *sql.DB) { @@ -109,6 +119,7 @@ func resetStorage(t *testing.T, db *sql.DB) { for _, statement := range []string{ "TRUNCATE TABLE request_log", + "TRUNCATE TABLE request_summary", "TRUNCATE TABLE request_uri", "TRUNCATE TABLE request", "TRUNCATE TABLE build", @@ -120,6 +131,43 @@ func resetStorage(t *testing.T, db *sql.DB) { } } +// MySQLRequestSummaryStoreSuite exercises summary projection storage against MySQL. +type MySQLRequestSummaryStoreSuite struct { + suite.Suite + ctx context.Context + store storage.RequestSummaryStore +} + +func (s *MySQLRequestSummaryStoreSuite) TestCreateGetAndUpdate() { + summary := entity.RequestSummary{ + RequestID: "request/monorepo/main/summary", Queue: "monorepo/main", URI: "git://repo/head", + State: entity.RequestStateAccepted, RequestVersion: 1, StateTimestampMs: 1000, Version: 1, + } + require.NoError(s.T(), s.store.Create(s.ctx, summary)) + + got, err := s.store.Get(s.ctx, summary.RequestID) + require.NoError(s.T(), err) + require.Equal(s.T(), summary, got) + + updated := summary + updated.BaseURI = "git://repo/base" + updated.State = entity.RequestStateProcessing + updated.RequestVersion = 2 + updated.StateTimestampMs = 2000 + require.NoError(s.T(), s.store.Update(s.ctx, updated, 1, 2)) + updated.Version = 2 + + got, err = s.store.Get(s.ctx, summary.RequestID) + require.NoError(s.T(), err) + require.Equal(s.T(), updated, got) + require.ErrorIs(s.T(), s.store.Update(s.ctx, updated, 1, 2), storage.ErrVersionMismatch) +} + +func (s *MySQLRequestSummaryStoreSuite) TestGetNotFound() { + _, err := s.store.Get(s.ctx, "request/monorepo/main/missing-summary") + require.True(s.T(), storage.IsNotFound(err)) +} + // MySQLRequestStoreSuite exercises the MySQL-backed RequestStore against a real MySQL instance. type MySQLRequestStoreSuite struct { suite.Suite