From 3f8bb1d80baaa06cc7b561b9d78dc6e25f817828 Mon Sep 17 00:00:00 2001 From: Albert Wu Date: Wed, 8 Jul 2026 15:51:36 -0700 Subject: [PATCH] feat(storage): add request read-model stores Summary: Add mechanical storage contracts and MySQL implementations for authoritative request summaries, queue-ordered summaries, and change URI mappings. Keep retry identity, conflict reconciliation, and projection decisions in callers while covering duplicate, CAS, ordering, cursor, and JSON normalization behavior in the shared storage contract suite. Test Plan: make fmt make mocks make gazelle ./tool/bazel test //test/integration/submitqueue/extension/storage/mysql:go_default_test --test_output=errors Revert Plan: Revert this commit. No production callers use the new stores yet. API Changes: None. Monitoring and Alerts: Storage operation metrics use the existing extension metric pattern. --- submitqueue/extension/storage/BUILD.bazel | 3 + .../extension/storage/mock/BUILD.bazel | 3 + .../mock/request_queue_summary_store_mock.go | 101 +++++++++++ .../mock/request_summary_store_mock.go | 85 +++++++++ .../storage/mock/request_uri_store_mock.go | 71 ++++++++ .../extension/storage/mock/storage_mock.go | 42 +++++ .../extension/storage/mysql/BUILD.bazel | 3 + .../storage/mysql/batch_dependent_store.go | 2 +- .../extension/storage/mysql/batch_store.go | 2 +- .../extension/storage/mysql/build_store.go | 2 +- .../mysql/request_queue_summary_store.go | 160 ++++++++++++++++ .../extension/storage/mysql/request_store.go | 4 +- .../storage/mysql/request_summary_store.go | 171 ++++++++++++++++++ .../storage/mysql/request_uri_store.go | 86 +++++++++ .../storage/mysql/speculation_tree_store.go | 2 +- .../extension/storage/mysql/storage.go | 25 +++ .../storage/request_queue_summary_store.go | 63 +++++++ .../storage/request_summary_store.go | 37 ++++ .../extension/storage/request_uri_store.go | 32 ++++ submitqueue/extension/storage/storage.go | 9 + .../submitqueue/extension/storage/suite.go | 80 ++++++++ 21 files changed, 976 insertions(+), 7 deletions(-) create mode 100644 submitqueue/extension/storage/mock/request_queue_summary_store_mock.go create mode 100644 submitqueue/extension/storage/mock/request_summary_store_mock.go create mode 100644 submitqueue/extension/storage/mock/request_uri_store_mock.go create mode 100644 submitqueue/extension/storage/mysql/request_queue_summary_store.go create mode 100644 submitqueue/extension/storage/mysql/request_summary_store.go create mode 100644 submitqueue/extension/storage/mysql/request_uri_store.go create mode 100644 submitqueue/extension/storage/request_queue_summary_store.go create mode 100644 submitqueue/extension/storage/request_summary_store.go create mode 100644 submitqueue/extension/storage/request_uri_store.go 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}) +}