diff --git a/submitqueue/extension/storage/BUILD.bazel b/submitqueue/extension/storage/BUILD.bazel index aa02fefb..428550d8 100644 --- a/submitqueue/extension/storage/BUILD.bazel +++ b/submitqueue/extension/storage/BUILD.bazel @@ -8,7 +8,10 @@ go_library( "build_store.go", "change_store.go", "request_log_store.go", + "request_queue_summary_store.go", "request_store.go", + "request_summary_store.go", + "request_uri_store.go", "speculation_tree_store.go", "storage.go", ], diff --git a/submitqueue/extension/storage/mock/BUILD.bazel b/submitqueue/extension/storage/mock/BUILD.bazel index 9224f547..77b0bf61 100644 --- a/submitqueue/extension/storage/mock/BUILD.bazel +++ b/submitqueue/extension/storage/mock/BUILD.bazel @@ -8,7 +8,10 @@ go_library( "build_store_mock.go", "change_store_mock.go", "request_log_store_mock.go", + "request_queue_summary_store_mock.go", "request_store_mock.go", + "request_summary_store_mock.go", + "request_uri_store_mock.go", "speculation_tree_store_mock.go", "storage_mock.go", ], diff --git a/submitqueue/extension/storage/mock/request_queue_summary_store_mock.go b/submitqueue/extension/storage/mock/request_queue_summary_store_mock.go new file mode 100644 index 00000000..5d7698af --- /dev/null +++ b/submitqueue/extension/storage/mock/request_queue_summary_store_mock.go @@ -0,0 +1,101 @@ +// Code generated by MockGen. DO NOT EDIT. +// Source: request_queue_summary_store.go +// +// Generated by this command: +// +// mockgen -source=request_queue_summary_store.go -destination=mock/request_queue_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/submitqueue/entity" + storage "github.com/uber/submitqueue/submitqueue/extension/storage" + gomock "go.uber.org/mock/gomock" +) + +// MockRequestQueueSummaryStore is a mock of RequestQueueSummaryStore interface. +type MockRequestQueueSummaryStore struct { + ctrl *gomock.Controller + recorder *MockRequestQueueSummaryStoreMockRecorder + isgomock struct{} +} + +// MockRequestQueueSummaryStoreMockRecorder is the mock recorder for MockRequestQueueSummaryStore. +type MockRequestQueueSummaryStoreMockRecorder struct { + mock *MockRequestQueueSummaryStore +} + +// NewMockRequestQueueSummaryStore creates a new mock instance. +func NewMockRequestQueueSummaryStore(ctrl *gomock.Controller) *MockRequestQueueSummaryStore { + mock := &MockRequestQueueSummaryStore{ctrl: ctrl} + mock.recorder = &MockRequestQueueSummaryStoreMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *MockRequestQueueSummaryStore) EXPECT() *MockRequestQueueSummaryStoreMockRecorder { + return m.recorder +} + +// Create mocks base method. +func (m *MockRequestQueueSummaryStore) Create(ctx context.Context, summary entity.RequestQueueSummary) 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 *MockRequestQueueSummaryStoreMockRecorder) Create(ctx, summary any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Create", reflect.TypeOf((*MockRequestQueueSummaryStore)(nil).Create), ctx, summary) +} + +// Get mocks base method. +func (m *MockRequestQueueSummaryStore) Get(ctx context.Context, queue string, receivedAtMs int64, requestID string) (entity.RequestQueueSummary, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Get", ctx, queue, receivedAtMs, requestID) + ret0, _ := ret[0].(entity.RequestQueueSummary) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// Get indicates an expected call of Get. +func (mr *MockRequestQueueSummaryStoreMockRecorder) Get(ctx, queue, receivedAtMs, requestID any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Get", reflect.TypeOf((*MockRequestQueueSummaryStore)(nil).Get), ctx, queue, receivedAtMs, requestID) +} + +// List mocks base method. +func (m *MockRequestQueueSummaryStore) List(ctx context.Context, query storage.RequestQueueSummaryQuery) ([]entity.RequestQueueSummary, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "List", ctx, query) + ret0, _ := ret[0].([]entity.RequestQueueSummary) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// List indicates an expected call of List. +func (mr *MockRequestQueueSummaryStoreMockRecorder) List(ctx, query any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "List", reflect.TypeOf((*MockRequestQueueSummaryStore)(nil).List), ctx, query) +} + +// Update mocks base method. +func (m *MockRequestQueueSummaryStore) Update(ctx context.Context, summary entity.RequestQueueSummary, 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 *MockRequestQueueSummaryStoreMockRecorder) Update(ctx, summary, oldVersion, newVersion any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Update", reflect.TypeOf((*MockRequestQueueSummaryStore)(nil).Update), ctx, summary, oldVersion, newVersion) +} diff --git a/submitqueue/extension/storage/mock/request_summary_store_mock.go b/submitqueue/extension/storage/mock/request_summary_store_mock.go new file mode 100644 index 00000000..aeb4adb3 --- /dev/null +++ b/submitqueue/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/submitqueue/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/submitqueue/extension/storage/mock/request_uri_store_mock.go b/submitqueue/extension/storage/mock/request_uri_store_mock.go new file mode 100644 index 00000000..1904a3f0 --- /dev/null +++ b/submitqueue/extension/storage/mock/request_uri_store_mock.go @@ -0,0 +1,71 @@ +// Code generated by MockGen. DO NOT EDIT. +// Source: request_uri_store.go +// +// Generated by this command: +// +// mockgen -source=request_uri_store.go -destination=mock/request_uri_store_mock.go -package=mock +// + +// Package mock is a generated GoMock package. +package mock + +import ( + context "context" + reflect "reflect" + + entity "github.com/uber/submitqueue/submitqueue/entity" + gomock "go.uber.org/mock/gomock" +) + +// MockRequestURIStore is a mock of RequestURIStore interface. +type MockRequestURIStore struct { + ctrl *gomock.Controller + recorder *MockRequestURIStoreMockRecorder + isgomock struct{} +} + +// MockRequestURIStoreMockRecorder is the mock recorder for MockRequestURIStore. +type MockRequestURIStoreMockRecorder struct { + mock *MockRequestURIStore +} + +// NewMockRequestURIStore creates a new mock instance. +func NewMockRequestURIStore(ctrl *gomock.Controller) *MockRequestURIStore { + mock := &MockRequestURIStore{ctrl: ctrl} + mock.recorder = &MockRequestURIStoreMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *MockRequestURIStore) EXPECT() *MockRequestURIStoreMockRecorder { + return m.recorder +} + +// Create mocks base method. +func (m *MockRequestURIStore) Create(ctx context.Context, mapping entity.RequestURI) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Create", ctx, mapping) + ret0, _ := ret[0].(error) + return ret0 +} + +// Create indicates an expected call of Create. +func (mr *MockRequestURIStoreMockRecorder) Create(ctx, mapping any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Create", reflect.TypeOf((*MockRequestURIStore)(nil).Create), ctx, mapping) +} + +// ListByURI mocks base method. +func (m *MockRequestURIStore) ListByURI(ctx context.Context, changeURI string, limit int) ([]entity.RequestURI, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "ListByURI", ctx, changeURI, limit) + ret0, _ := ret[0].([]entity.RequestURI) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// ListByURI indicates an expected call of ListByURI. +func (mr *MockRequestURIStoreMockRecorder) ListByURI(ctx, changeURI, limit any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ListByURI", reflect.TypeOf((*MockRequestURIStore)(nil).ListByURI), ctx, changeURI, limit) +} diff --git a/submitqueue/extension/storage/mock/storage_mock.go b/submitqueue/extension/storage/mock/storage_mock.go index 4133bc2a..5bd575fa 100644 --- a/submitqueue/extension/storage/mock/storage_mock.go +++ b/submitqueue/extension/storage/mock/storage_mock.go @@ -124,6 +124,20 @@ func (mr *MockStorageMockRecorder) GetRequestLogStore() *gomock.Call { return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetRequestLogStore", reflect.TypeOf((*MockStorage)(nil).GetRequestLogStore)) } +// GetRequestQueueSummaryStore mocks base method. +func (m *MockStorage) GetRequestQueueSummaryStore() storage.RequestQueueSummaryStore { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "GetRequestQueueSummaryStore") + ret0, _ := ret[0].(storage.RequestQueueSummaryStore) + return ret0 +} + +// GetRequestQueueSummaryStore indicates an expected call of GetRequestQueueSummaryStore. +func (mr *MockStorageMockRecorder) GetRequestQueueSummaryStore() *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetRequestQueueSummaryStore", reflect.TypeOf((*MockStorage)(nil).GetRequestQueueSummaryStore)) +} + // GetRequestStore mocks base method. func (m *MockStorage) GetRequestStore() storage.RequestStore { m.ctrl.T.Helper() @@ -138,6 +152,34 @@ 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() + ret := m.ctrl.Call(m, "GetRequestURIStore") + ret0, _ := ret[0].(storage.RequestURIStore) + return ret0 +} + +// GetRequestURIStore indicates an expected call of GetRequestURIStore. +func (mr *MockStorageMockRecorder) GetRequestURIStore() *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetRequestURIStore", reflect.TypeOf((*MockStorage)(nil).GetRequestURIStore)) +} + // GetSpeculationTreeStore mocks base method. func (m *MockStorage) GetSpeculationTreeStore() storage.SpeculationTreeStore { m.ctrl.T.Helper() diff --git a/submitqueue/extension/storage/mysql/BUILD.bazel b/submitqueue/extension/storage/mysql/BUILD.bazel index 5cb431e4..a49e4127 100644 --- a/submitqueue/extension/storage/mysql/BUILD.bazel +++ b/submitqueue/extension/storage/mysql/BUILD.bazel @@ -8,7 +8,10 @@ go_library( "build_store.go", "change_store.go", "request_log_store.go", + "request_queue_summary_store.go", "request_store.go", + "request_summary_store.go", + "request_uri_store.go", "speculation_tree_store.go", "storage.go", ], diff --git a/submitqueue/extension/storage/mysql/batch_dependent_store.go b/submitqueue/extension/storage/mysql/batch_dependent_store.go index 7bf8047d..64de6431 100644 --- a/submitqueue/extension/storage/mysql/batch_dependent_store.go +++ b/submitqueue/extension/storage/mysql/batch_dependent_store.go @@ -82,7 +82,7 @@ func (s *batchDependentStore) Create(ctx context.Context, batchDependent entity. ) if err != nil { var mysqlErr *mysql.MySQLError - if errors.As(err, &mysqlErr) && mysqlErr.Number == 1062 { + if errors.As(err, &mysqlErr) && mysqlErr.Number == mysqlErrDuplicateEntry { return fmt.Errorf("batch dependent entity batchID=%s: %w", batchDependent.BatchID, storage.ErrAlreadyExists) } return fmt.Errorf("failed to insert batch dependent entity batchID=%s: %w", batchDependent.BatchID, err) diff --git a/submitqueue/extension/storage/mysql/batch_store.go b/submitqueue/extension/storage/mysql/batch_store.go index 4f0168ee..07020dac 100644 --- a/submitqueue/extension/storage/mysql/batch_store.go +++ b/submitqueue/extension/storage/mysql/batch_store.go @@ -93,7 +93,7 @@ func (s *batchStore) Create(ctx context.Context, batch entity.Batch) (retErr err ) if err != nil { var mysqlErr *mysql.MySQLError - if errors.As(err, &mysqlErr) && mysqlErr.Number == 1062 { + if errors.As(err, &mysqlErr) && mysqlErr.Number == mysqlErrDuplicateEntry { return fmt.Errorf("batch entity id=%s: %w", batch.ID, storage.ErrAlreadyExists) } return fmt.Errorf("failed to insert batch entity id=%s: %w", batch.ID, err) diff --git a/submitqueue/extension/storage/mysql/build_store.go b/submitqueue/extension/storage/mysql/build_store.go index d04eae53..383e5df2 100644 --- a/submitqueue/extension/storage/mysql/build_store.go +++ b/submitqueue/extension/storage/mysql/build_store.go @@ -82,7 +82,7 @@ func (s *buildStore) Create(ctx context.Context, build entity.Build) (retErr err ) if err != nil { var mysqlErr *mysql.MySQLError - if errors.As(err, &mysqlErr) && mysqlErr.Number == 1062 { + if errors.As(err, &mysqlErr) && mysqlErr.Number == mysqlErrDuplicateEntry { return fmt.Errorf("build entity id=%s: %w", build.ID, storage.ErrAlreadyExists) } return fmt.Errorf("failed to insert build entity id=%s: %w", build.ID, err) diff --git a/submitqueue/extension/storage/mysql/request_queue_summary_store.go b/submitqueue/extension/storage/mysql/request_queue_summary_store.go new file mode 100644 index 00000000..6f89e11f --- /dev/null +++ b/submitqueue/extension/storage/mysql/request_queue_summary_store.go @@ -0,0 +1,160 @@ +// 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 mysql + +import ( + "context" + "database/sql" + "encoding/json" + "errors" + "fmt" + + "github.com/go-sql-driver/mysql" + "github.com/uber-go/tally" + + "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/submitqueue/entity" + "github.com/uber/submitqueue/submitqueue/extension/storage" +) + +type requestQueueSummaryStore struct { + db *sql.DB + scope tally.Scope +} + +// NewRequestQueueSummaryStore creates a MySQL-backed RequestQueueSummaryStore. +func NewRequestQueueSummaryStore(db *sql.DB, scope tally.Scope) storage.RequestQueueSummaryStore { + return &requestQueueSummaryStore{db: db, scope: scope} +} + +func (s *requestQueueSummaryStore) Create(ctx context.Context, summary entity.RequestQueueSummary) (retErr error) { + op := metrics.Begin(s.scope, "create") + defer func() { op.Complete(retErr) }() + + changeURIsJSON, metadataJSON, err := marshalSummaryJSON(summary.ChangeURIs, summary.Metadata) + if err != nil { + return fmt.Errorf("failed to marshal queue summary request_id=%s: %w", summary.RequestID, err) + } + _, err = s.db.ExecContext(ctx, ` + INSERT INTO request_summary_by_queue ( + queue, received_at_ms, request_id, change_uris, status, + projection_version, last_error, metadata + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`, + summary.Queue, summary.ReceivedAtMs, summary.RequestID, changeURIsJSON, + summary.Status, summary.ProjectionVersion, summary.LastError, metadataJSON, + ) + if err != nil { + var mysqlErr *mysql.MySQLError + if errors.As(err, &mysqlErr) && mysqlErr.Number == mysqlErrDuplicateEntry { + return fmt.Errorf("queue summary queue=%s received_at_ms=%d request_id=%s: %w", summary.Queue, summary.ReceivedAtMs, summary.RequestID, storage.ErrAlreadyExists) + } + return fmt.Errorf("failed to insert queue summary request_id=%s: %w", summary.RequestID, err) + } + return nil +} + +func (s *requestQueueSummaryStore) Get(ctx context.Context, queue string, receivedAtMs int64, requestID string) (ret entity.RequestQueueSummary, retErr error) { + op := metrics.Begin(s.scope, "get") + defer func() { op.Complete(retErr) }() + + var changeURIsJSON []byte + var metadataJSON []byte + err := s.db.QueryRowContext(ctx, ` + SELECT queue, received_at_ms, request_id, change_uris, status, + projection_version, last_error, metadata + FROM request_summary_by_queue + WHERE queue = ? AND received_at_ms = ? AND request_id = ?`, queue, receivedAtMs, requestID, + ).Scan(&ret.Queue, &ret.ReceivedAtMs, &ret.RequestID, &changeURIsJSON, &ret.Status, &ret.ProjectionVersion, &ret.LastError, &metadataJSON) + if errors.Is(err, sql.ErrNoRows) { + return entity.RequestQueueSummary{}, storage.WrapNotFound(err) + } + if err != nil { + return entity.RequestQueueSummary{}, fmt.Errorf("failed to get queue summary queue=%s received_at_ms=%d request_id=%s: %w", queue, receivedAtMs, requestID, err) + } + if err := unmarshalSummaryJSON(changeURIsJSON, metadataJSON, &ret.ChangeURIs, &ret.Metadata); err != nil { + return entity.RequestQueueSummary{}, fmt.Errorf("failed to decode queue summary request_id=%s: %w", requestID, err) + } + return ret, nil +} + +func (s *requestQueueSummaryStore) Update(ctx context.Context, summary entity.RequestQueueSummary, oldVersion, newVersion int32) (retErr error) { + op := metrics.Begin(s.scope, "update") + defer func() { op.Complete(retErr) }() + + metadataJSON, err := json.Marshal(normalizeMetadata(summary.Metadata)) + if err != nil { + return fmt.Errorf("failed to marshal queue summary metadata request_id=%s: %w", summary.RequestID, err) + } + result, err := s.db.ExecContext(ctx, ` + UPDATE request_summary_by_queue + SET status = ?, projection_version = ?, last_error = ?, metadata = ? + WHERE queue = ? AND received_at_ms = ? AND request_id = ? AND projection_version = ?`, + summary.Status, newVersion, summary.LastError, metadataJSON, + summary.Queue, summary.ReceivedAtMs, summary.RequestID, oldVersion, + ) + if err != nil { + return fmt.Errorf("failed to update queue summary request_id=%s 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 queue summary update rows request_id=%s: %w", summary.RequestID, err) + } + if rowsAffected != 1 { + return fmt.Errorf("queue summary request_id=%s expected_version=%d: %w", summary.RequestID, oldVersion, storage.ErrVersionMismatch) + } + return nil +} + +func (s *requestQueueSummaryStore) List(ctx context.Context, query storage.RequestQueueSummaryQuery) (ret []entity.RequestQueueSummary, retErr error) { + op := metrics.Begin(s.scope, "list") + defer func() { op.Complete(retErr) }() + + statement := ` + SELECT queue, received_at_ms, request_id, change_uris, status, + projection_version, last_error, metadata + FROM request_summary_by_queue + WHERE queue = ? AND received_at_ms >= ? AND received_at_ms < ?` + args := []any{query.Queue, query.ReceivedAtOrAfterMs, query.ReceivedBeforeMs} + if query.HasCursor { + statement += " AND (received_at_ms < ? OR (received_at_ms = ? AND request_id < ?))" + args = append(args, query.Cursor.ReceivedAtMs, query.Cursor.ReceivedAtMs, query.Cursor.RequestID) + } + statement += " ORDER BY received_at_ms DESC, request_id DESC LIMIT ?" + args = append(args, query.Limit) + + rows, err := s.db.QueryContext(ctx, statement, args...) + if err != nil { + return nil, fmt.Errorf("failed to list queue summaries queue=%s: %w", query.Queue, err) + } + defer rows.Close() + + results := make([]entity.RequestQueueSummary, 0) + for rows.Next() { + var summary entity.RequestQueueSummary + var changeURIsJSON []byte + var metadataJSON []byte + if err := rows.Scan(&summary.Queue, &summary.ReceivedAtMs, &summary.RequestID, &changeURIsJSON, &summary.Status, &summary.ProjectionVersion, &summary.LastError, &metadataJSON); err != nil { + return nil, fmt.Errorf("failed to scan queue summary queue=%s: %w", query.Queue, err) + } + if err := unmarshalSummaryJSON(changeURIsJSON, metadataJSON, &summary.ChangeURIs, &summary.Metadata); err != nil { + return nil, fmt.Errorf("failed to decode queue summary request_id=%s: %w", summary.RequestID, err) + } + results = append(results, summary) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("failed to iterate queue summaries queue=%s: %w", query.Queue, err) + } + return results, nil +} diff --git a/submitqueue/extension/storage/mysql/request_store.go b/submitqueue/extension/storage/mysql/request_store.go index d332498d..fd542c9d 100644 --- a/submitqueue/extension/storage/mysql/request_store.go +++ b/submitqueue/extension/storage/mysql/request_store.go @@ -84,9 +84,7 @@ func (r *requestStore) Create(ctx context.Context, request entity.Request) (retE ) if err != nil { var mysqlErr *mysql.MySQLError - if errors.As(err, &mysqlErr) && mysqlErr.Number == 1062 { - // MySQL error code 1062 is "Duplicate entry". Hopefully it will never change with new versions of MySQL. - // Also it requires to have a single unique index on the table. + if errors.As(err, &mysqlErr) && mysqlErr.Number == mysqlErrDuplicateEntry { return fmt.Errorf("request entity id=%s: %w", request.ID, storage.ErrAlreadyExists) } return fmt.Errorf("failed to insert request entity id=%s: %w", request.ID, err) diff --git a/submitqueue/extension/storage/mysql/request_summary_store.go b/submitqueue/extension/storage/mysql/request_summary_store.go new file mode 100644 index 00000000..a35dfdda --- /dev/null +++ b/submitqueue/extension/storage/mysql/request_summary_store.go @@ -0,0 +1,171 @@ +// 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 mysql + +import ( + "context" + "database/sql" + "encoding/json" + "errors" + "fmt" + + "github.com/go-sql-driver/mysql" + "github.com/uber-go/tally" + + "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/submitqueue/entity" + "github.com/uber/submitqueue/submitqueue/extension/storage" +) + +type requestSummaryStore struct { + db *sql.DB + scope tally.Scope +} + +// NewRequestSummaryStore creates a MySQL-backed RequestSummaryStore. +func NewRequestSummaryStore(db *sql.DB, scope tally.Scope) storage.RequestSummaryStore { + return &requestSummaryStore{db: db, scope: scope} +} + +func (s *requestSummaryStore) Create(ctx context.Context, summary entity.RequestSummary) (retErr error) { + op := metrics.Begin(s.scope, "create") + defer func() { op.Complete(retErr) }() + + changeURIsJSON, metadataJSON, err := marshalSummaryJSON(summary.ChangeURIs, summary.Metadata) + if err != nil { + return fmt.Errorf("failed to marshal request summary request_id=%s: %w", summary.RequestID, err) + } + + _, err = s.db.ExecContext(ctx, ` + INSERT INTO request_summary ( + request_id, queue, change_uris, received_at_ms, status, request_version, + status_timestamp_ms, projection_version, + last_error, metadata + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + summary.RequestID, summary.Queue, changeURIsJSON, summary.ReceivedAtMs, summary.Status, + summary.RequestVersion, summary.StatusTimestampMs, summary.ProjectionVersion, + summary.LastError, metadataJSON, + ) + if err != nil { + var mysqlErr *mysql.MySQLError + if errors.As(err, &mysqlErr) && mysqlErr.Number == mysqlErrDuplicateEntry { + return fmt.Errorf("request summary request_id=%s: %w", summary.RequestID, storage.ErrAlreadyExists) + } + return fmt.Errorf("failed to insert request summary request_id=%s: %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") + defer func() { op.Complete(retErr) }() + + var changeURIsJSON []byte + var metadataJSON []byte + err := s.db.QueryRowContext(ctx, ` + SELECT request_id, queue, change_uris, received_at_ms, status, request_version, + status_timestamp_ms, projection_version, + last_error, metadata + FROM request_summary + WHERE request_id = ?`, requestID, + ).Scan( + &ret.RequestID, &ret.Queue, &changeURIsJSON, &ret.ReceivedAtMs, &ret.Status, + &ret.RequestVersion, &ret.StatusTimestampMs, &ret.ProjectionVersion, + &ret.LastError, &metadataJSON, + ) + 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=%s: %w", requestID, err) + } + if err := unmarshalSummaryJSON(changeURIsJSON, metadataJSON, &ret.ChangeURIs, &ret.Metadata); err != nil { + return entity.RequestSummary{}, fmt.Errorf("failed to decode request summary request_id=%s: %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") + defer func() { op.Complete(retErr) }() + + metadata := normalizeMetadata(summary.Metadata) + metadataJSON, err := json.Marshal(metadata) + if err != nil { + return fmt.Errorf("failed to marshal request summary metadata request_id=%s: %w", summary.RequestID, err) + } + + result, err := s.db.ExecContext(ctx, ` + UPDATE request_summary + SET status = ?, request_version = ?, status_timestamp_ms = ?, + projection_version = ?, last_error = ?, metadata = ? + WHERE request_id = ? AND projection_version = ?`, + summary.Status, summary.RequestVersion, summary.StatusTimestampMs, + newVersion, summary.LastError, metadataJSON, + summary.RequestID, oldVersion, + ) + if err != nil { + return fmt.Errorf("failed to update request summary request_id=%s 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=%s: %w", summary.RequestID, err) + } + if rowsAffected != 1 { + return fmt.Errorf("request summary request_id=%s expected_version=%d: %w", summary.RequestID, oldVersion, storage.ErrVersionMismatch) + } + + return nil +} + +func marshalSummaryJSON(changeURIs []string, metadata map[string]string) ([]byte, []byte, error) { + changeURIsJSON, err := json.Marshal(normalizeChangeURIs(changeURIs)) + if err != nil { + return nil, nil, err + } + metadataJSON, err := json.Marshal(normalizeMetadata(metadata)) + if err != nil { + return nil, nil, err + } + return changeURIsJSON, metadataJSON, nil +} + +func unmarshalSummaryJSON(changeURIsJSON, metadataJSON []byte, changeURIs *[]string, metadata *map[string]string) error { + if err := json.Unmarshal(changeURIsJSON, changeURIs); err != nil { + return err + } + if err := json.Unmarshal(metadataJSON, metadata); err != nil { + return err + } + *changeURIs = normalizeChangeURIs(*changeURIs) + *metadata = normalizeMetadata(*metadata) + return nil +} + +func normalizeChangeURIs(changeURIs []string) []string { + if changeURIs == nil { + return []string{} + } + return changeURIs +} + +func normalizeMetadata(metadata map[string]string) map[string]string { + if metadata == nil { + return map[string]string{} + } + return metadata +} diff --git a/submitqueue/extension/storage/mysql/request_uri_store.go b/submitqueue/extension/storage/mysql/request_uri_store.go new file mode 100644 index 00000000..b8fb0273 --- /dev/null +++ b/submitqueue/extension/storage/mysql/request_uri_store.go @@ -0,0 +1,86 @@ +// 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 mysql + +import ( + "context" + "database/sql" + "errors" + "fmt" + + "github.com/go-sql-driver/mysql" + "github.com/uber-go/tally" + + "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/submitqueue/entity" + "github.com/uber/submitqueue/submitqueue/extension/storage" +) + +type requestURIStore struct { + db *sql.DB + scope tally.Scope +} + +// NewRequestURIStore creates a MySQL-backed RequestURIStore. +func NewRequestURIStore(db *sql.DB, scope tally.Scope) storage.RequestURIStore { + return &requestURIStore{db: db, scope: scope} +} + +func (s *requestURIStore) Create(ctx context.Context, mapping entity.RequestURI) (retErr error) { + op := metrics.Begin(s.scope, "create") + defer func() { op.Complete(retErr) }() + + _, err := s.db.ExecContext(ctx, + "INSERT INTO request_by_change_uri (change_uri, received_at_ms, request_id) VALUES (?, ?, ?)", + mapping.ChangeURI, mapping.ReceivedAtMs, mapping.RequestID, + ) + if err != nil { + var mysqlErr *mysql.MySQLError + if errors.As(err, &mysqlErr) && mysqlErr.Number == mysqlErrDuplicateEntry { + return fmt.Errorf("request URI change_uri=%s received_at_ms=%d request_id=%s: %w", mapping.ChangeURI, mapping.ReceivedAtMs, mapping.RequestID, storage.ErrAlreadyExists) + } + return fmt.Errorf("failed to insert request URI request_id=%s change_uri=%s: %w", mapping.RequestID, mapping.ChangeURI, err) + } + return nil +} + +func (s *requestURIStore) ListByURI(ctx context.Context, changeURI string, limit int) (ret []entity.RequestURI, retErr error) { + op := metrics.Begin(s.scope, "list_by_uri") + defer func() { op.Complete(retErr) }() + + rows, err := s.db.QueryContext(ctx, ` + SELECT change_uri, received_at_ms, request_id + FROM request_by_change_uri + WHERE change_uri = ? + ORDER BY received_at_ms DESC, request_id DESC + LIMIT ?`, changeURI, limit) + if err != nil { + return nil, fmt.Errorf("failed to list request URIs change_uri=%s: %w", changeURI, err) + } + defer rows.Close() + + results := make([]entity.RequestURI, 0) + for rows.Next() { + var mapping entity.RequestURI + if err := rows.Scan(&mapping.ChangeURI, &mapping.ReceivedAtMs, &mapping.RequestID); err != nil { + return nil, fmt.Errorf("failed to scan request URI change_uri=%s: %w", changeURI, err) + } + results = append(results, mapping) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("failed to iterate request URIs change_uri=%s: %w", changeURI, err) + } + return results, nil +} diff --git a/submitqueue/extension/storage/mysql/speculation_tree_store.go b/submitqueue/extension/storage/mysql/speculation_tree_store.go index 564cd778..e57775ab 100644 --- a/submitqueue/extension/storage/mysql/speculation_tree_store.go +++ b/submitqueue/extension/storage/mysql/speculation_tree_store.go @@ -82,7 +82,7 @@ func (s *speculationTreeStore) Create(ctx context.Context, speculationTree entit ) if err != nil { var mysqlErr *mysql.MySQLError - if errors.As(err, &mysqlErr) && mysqlErr.Number == 1062 { + if errors.As(err, &mysqlErr) && mysqlErr.Number == mysqlErrDuplicateEntry { return fmt.Errorf("speculation tree entity batchID=%s: %w", speculationTree.BatchID, storage.ErrAlreadyExists) } return fmt.Errorf("failed to insert speculation tree entity batchID=%s: %w", speculationTree.BatchID, err) diff --git a/submitqueue/extension/storage/mysql/storage.go b/submitqueue/extension/storage/mysql/storage.go index 4ba0cf41..1cb9161b 100644 --- a/submitqueue/extension/storage/mysql/storage.go +++ b/submitqueue/extension/storage/mysql/storage.go @@ -23,6 +23,10 @@ import ( "github.com/uber/submitqueue/submitqueue/extension/storage" ) +// mysqlErrDuplicateEntry is MySQL error code 1062 ("Duplicate entry"), returned on a unique or primary key violation. +// It requires a unique index on the table to be raised. +const mysqlErrDuplicateEntry = 1062 + type mysqlStorage struct { db *sql.DB requestStore storage.RequestStore @@ -32,6 +36,9 @@ type mysqlStorage struct { buildStore storage.BuildStore speculationTreeStore storage.SpeculationTreeStore requestLogStore storage.RequestLogStore + requestSummaryStore storage.RequestSummaryStore + requestQueueStore storage.RequestQueueSummaryStore + requestURIStore storage.RequestURIStore } // NewStorage creates a new MySQL storage. @@ -45,6 +52,9 @@ func NewStorage(db *sql.DB, scope tally.Scope) (storage.Storage, error) { buildStore: NewBuildStore(db, scope.SubScope("build_store")), speculationTreeStore: NewSpeculationTreeStore(db, scope.SubScope("speculation_tree_store")), requestLogStore: NewRequestLogStore(db, scope.SubScope("request_log_store")), + requestSummaryStore: NewRequestSummaryStore(db, scope.SubScope("request_summary_store")), + requestQueueStore: NewRequestQueueSummaryStore(db, scope.SubScope("request_queue_summary_store")), + requestURIStore: NewRequestURIStore(db, scope.SubScope("request_uri_store")), }, nil } @@ -83,6 +93,21 @@ func (f *mysqlStorage) GetRequestLogStore() storage.RequestLogStore { return f.requestLogStore } +// GetRequestSummaryStore returns the MySQL-backed RequestSummaryStore. +func (f *mysqlStorage) GetRequestSummaryStore() storage.RequestSummaryStore { + return f.requestSummaryStore +} + +// GetRequestQueueSummaryStore returns the MySQL-backed RequestQueueSummaryStore. +func (f *mysqlStorage) GetRequestQueueSummaryStore() storage.RequestQueueSummaryStore { + return f.requestQueueStore +} + +// GetRequestURIStore returns the MySQL-backed RequestURIStore. +func (f *mysqlStorage) GetRequestURIStore() storage.RequestURIStore { + return f.requestURIStore +} + // Close closes the underlying database connection. func (f *mysqlStorage) Close() error { return f.db.Close() diff --git a/submitqueue/extension/storage/request_queue_summary_store.go b/submitqueue/extension/storage/request_queue_summary_store.go new file mode 100644 index 00000000..68acf211 --- /dev/null +++ b/submitqueue/extension/storage/request_queue_summary_store.go @@ -0,0 +1,63 @@ +// 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 storage + +//go:generate mockgen -source=request_queue_summary_store.go -destination=mock/request_queue_summary_store_mock.go -package=mock + +import ( + "context" + + "github.com/uber/submitqueue/submitqueue/entity" +) + +// RequestQueueSummaryCursor is the exclusive keyset boundary for a descending queue-summary query. +type RequestQueueSummaryCursor struct { + // ReceivedAtMs is the receipt timestamp of the last item from the previous page. + ReceivedAtMs int64 + // RequestID is the request ID of the last item from the previous page. + RequestID string +} + +// RequestQueueSummaryQuery specifies one bounded queue-summary page query. +type RequestQueueSummaryQuery struct { + // Queue is the exact queue partition to scan. + Queue string + // ReceivedAtOrAfterMs is the inclusive lower receipt-time bound. + ReceivedAtOrAfterMs int64 + // ReceivedBeforeMs is the exclusive upper receipt-time bound. + ReceivedBeforeMs int64 + // Cursor is an exclusive continuation boundary when HasCursor is true. + Cursor RequestQueueSummaryCursor + // HasCursor selects whether Cursor participates in the query. + HasCursor bool + // Limit is the maximum number of rows returned and must be positive. + Limit int +} + +// RequestQueueSummaryStore persists the queue-ordered request projection. +type RequestQueueSummaryStore interface { + // Create inserts summary and returns ErrAlreadyExists when its full primary key already exists. + Create(ctx context.Context, summary entity.RequestQueueSummary) error + + // Get returns the row identified by its full primary key, or ErrNotFound when absent. + Get(ctx context.Context, queue string, receivedAtMs int64, requestID string) (entity.RequestQueueSummary, error) + + // Update conditionally replaces mutable fields when the persisted projection version equals oldVersion. + // The store writes newVersion exactly as supplied and returns ErrVersionMismatch when the guard does not match. + Update(ctx context.Context, summary entity.RequestQueueSummary, oldVersion, newVersion int32) error + + // List returns at most query.Limit rows ordered by received_at_ms descending, then request_id descending. + List(ctx context.Context, query RequestQueueSummaryQuery) ([]entity.RequestQueueSummary, error) +} diff --git a/submitqueue/extension/storage/request_summary_store.go b/submitqueue/extension/storage/request_summary_store.go new file mode 100644 index 00000000..5e2a657a --- /dev/null +++ b/submitqueue/extension/storage/request_summary_store.go @@ -0,0 +1,37 @@ +// 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 storage + +//go:generate mockgen -source=request_summary_store.go -destination=mock/request_summary_store_mock.go -package=mock + +import ( + "context" + + "github.com/uber/submitqueue/submitqueue/entity" +) + +// RequestSummaryStore persists the authoritative request-ID materialized view. +type RequestSummaryStore interface { + // Create inserts summary and returns ErrAlreadyExists when RequestID already exists. + // The caller owns retry identity and decides whether an existing row is an identical retry or a conflict. + Create(ctx context.Context, summary entity.RequestSummary) error + + // Get returns the summary for requestID, or ErrNotFound when absent. + Get(ctx context.Context, requestID string) (entity.RequestSummary, error) + + // Update conditionally replaces the mutable status fields when the persisted projection version equals oldVersion. + // The store writes newVersion exactly as supplied and returns ErrVersionMismatch when the guard does not match. + Update(ctx context.Context, summary entity.RequestSummary, oldVersion, newVersion int32) error +} diff --git a/submitqueue/extension/storage/request_uri_store.go b/submitqueue/extension/storage/request_uri_store.go new file mode 100644 index 00000000..9ac1c024 --- /dev/null +++ b/submitqueue/extension/storage/request_uri_store.go @@ -0,0 +1,32 @@ +// 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 storage + +//go:generate mockgen -source=request_uri_store.go -destination=mock/request_uri_store_mock.go -package=mock + +import ( + "context" + + "github.com/uber/submitqueue/submitqueue/entity" +) + +// RequestURIStore persists the immutable change-URI reverse mapping. +type RequestURIStore interface { + // Create inserts mapping and returns ErrAlreadyExists when its full primary key already exists. + Create(ctx context.Context, mapping entity.RequestURI) error + + // ListByURI returns at most limit mappings ordered by received_at_ms descending, then request_id descending. + ListByURI(ctx context.Context, changeURI string, limit int) ([]entity.RequestURI, error) +} diff --git a/submitqueue/extension/storage/storage.go b/submitqueue/extension/storage/storage.go index a02bef73..1dc5141a 100644 --- a/submitqueue/extension/storage/storage.go +++ b/submitqueue/extension/storage/storage.go @@ -65,6 +65,15 @@ type Storage interface { // GetRequestLogStore returns the RequestLogStore instance. GetRequestLogStore() RequestLogStore + // GetRequestSummaryStore returns the RequestSummaryStore instance. + GetRequestSummaryStore() RequestSummaryStore + + // GetRequestQueueSummaryStore returns the RequestQueueSummaryStore instance. + GetRequestQueueSummaryStore() RequestQueueSummaryStore + + // GetRequestURIStore returns the RequestURIStore instance. + GetRequestURIStore() RequestURIStore + // Close closes the storage and all underlying connections. Should only be called once at the end of the program. Close() error } diff --git a/test/integration/submitqueue/extension/storage/suite.go b/test/integration/submitqueue/extension/storage/suite.go index 23984461..9260f6cd 100644 --- a/test/integration/submitqueue/extension/storage/suite.go +++ b/test/integration/submitqueue/extension/storage/suite.go @@ -381,3 +381,83 @@ func (s *StorageContractSuite) TestStorage_ChangeCreate_EmptyDetails() { require.Len(t, got, 1) assert.Equal(t, entity.ChangeDetails{}, got[0].Details) } + +func (s *StorageContractSuite) TestStorage_RequestSummaryCreateGetAndCAS() { + t := s.T() + ctx := s.ctx + summary := entity.RequestSummary{ + RequestID: "summary/1", Queue: "summary-q", ChangeURIs: nil, ReceivedAtMs: 100, + Status: entity.RequestStatusAccepted, StatusTimestampMs: 100, ProjectionVersion: 1, Metadata: nil, + } + + require.NoError(t, s.storage.GetRequestSummaryStore().Create(ctx, summary)) + require.ErrorIs(t, s.storage.GetRequestSummaryStore().Create(ctx, summary), storage.ErrAlreadyExists) + + got, err := s.storage.GetRequestSummaryStore().Get(ctx, summary.RequestID) + require.NoError(t, err) + assert.NotNil(t, got.ChangeURIs) + assert.NotNil(t, got.Metadata) + + got.Status = entity.RequestStatusLanded + got.RequestVersion = 2 + got.StatusTimestampMs = 200 + require.NoError(t, s.storage.GetRequestSummaryStore().Update(ctx, got, 1, 2)) + require.ErrorIs(t, s.storage.GetRequestSummaryStore().Update(ctx, got, 1, 3), storage.ErrVersionMismatch) + + updated, err := s.storage.GetRequestSummaryStore().Get(ctx, summary.RequestID) + require.NoError(t, err) + assert.Equal(t, int32(2), updated.ProjectionVersion) + assert.Equal(t, entity.RequestStatusLanded, updated.Status) +} + +func (s *StorageContractSuite) TestStorage_RequestQueueSummaryListAndCursor() { + t := s.T() + ctx := s.ctx + store := s.storage.GetRequestQueueSummaryStore() + rows := []entity.RequestQueueSummary{ + {RequestID: "queue-summary/1", Queue: "queue-summary", ChangeURIs: nil, ReceivedAtMs: 100, Status: entity.RequestStatusAccepted, ProjectionVersion: 1, Metadata: nil}, + {RequestID: "queue-summary/2", Queue: "queue-summary", ChangeURIs: []string{"uri/2"}, ReceivedAtMs: 200, Status: entity.RequestStatusLanded, ProjectionVersion: 1, Metadata: map[string]string{}}, + {RequestID: "queue-summary/3", Queue: "queue-summary", ChangeURIs: []string{"uri/3"}, ReceivedAtMs: 200, Status: entity.RequestStatusError, ProjectionVersion: 1, Metadata: map[string]string{}}, + } + for _, row := range rows { + require.NoError(t, store.Create(ctx, row)) + } + require.ErrorIs(t, store.Create(ctx, rows[0]), storage.ErrAlreadyExists) + + firstPage, err := store.List(ctx, storage.RequestQueueSummaryQuery{ + Queue: "queue-summary", ReceivedAtOrAfterMs: 50, ReceivedBeforeMs: 250, Limit: 2, + }) + require.NoError(t, err) + require.Len(t, firstPage, 2) + assert.Equal(t, []string{"queue-summary/3", "queue-summary/2"}, []string{firstPage[0].RequestID, firstPage[1].RequestID}) + + secondPage, err := store.List(ctx, storage.RequestQueueSummaryQuery{ + Queue: "queue-summary", ReceivedAtOrAfterMs: 50, ReceivedBeforeMs: 250, Limit: 2, + HasCursor: true, Cursor: storage.RequestQueueSummaryCursor{ReceivedAtMs: 200, RequestID: "queue-summary/2"}, + }) + require.NoError(t, err) + require.Len(t, secondPage, 1) + assert.Equal(t, "queue-summary/1", secondPage[0].RequestID) + assert.NotNil(t, secondPage[0].ChangeURIs) + assert.NotNil(t, secondPage[0].Metadata) +} + +func (s *StorageContractSuite) TestStorage_RequestURIListIsBoundedAndOrdered() { + t := s.T() + ctx := s.ctx + store := s.storage.GetRequestURIStore() + rows := []entity.RequestURI{ + {ChangeURI: "uri/shared", ReceivedAtMs: 100, RequestID: "uri/1"}, + {ChangeURI: "uri/shared", ReceivedAtMs: 200, RequestID: "uri/2"}, + {ChangeURI: "uri/shared", ReceivedAtMs: 200, RequestID: "uri/3"}, + } + for _, row := range rows { + require.NoError(t, store.Create(ctx, row)) + } + require.ErrorIs(t, store.Create(ctx, rows[0]), storage.ErrAlreadyExists) + + got, err := store.ListByURI(ctx, "uri/shared", 2) + require.NoError(t, err) + require.Len(t, got, 2) + assert.Equal(t, []string{"uri/3", "uri/2"}, []string{got[0].RequestID, got[1].RequestID}) +}