diff --git a/submitqueue/entity/BUILD.bazel b/submitqueue/entity/BUILD.bazel index 231dcbd6..f997cfab 100644 --- a/submitqueue/entity/BUILD.bazel +++ b/submitqueue/entity/BUILD.bazel @@ -16,6 +16,7 @@ go_library( "push_result.go", "queue_config.go", "request.go", + "request_context.go", "request_log.go", "speculation_tree.go", ], diff --git a/submitqueue/entity/request_context.go b/submitqueue/entity/request_context.go new file mode 100644 index 00000000..b78c20d0 --- /dev/null +++ b/submitqueue/entity/request_context.go @@ -0,0 +1,27 @@ +// 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 entity + +// RequestContext is immutable gateway-owned admission data for a request. +type RequestContext struct { + // RequestID identifies the request this context belongs to. + RequestID string + // Queue identifies the queue that admitted the request. + Queue string + // ChangeURIs contains the original change URIs submitted with the request. + ChangeURIs []string + // AdmittedAtMs is the request admission time in Unix epoch milliseconds. + AdmittedAtMs int64 +} diff --git a/submitqueue/extension/storage/BUILD.bazel b/submitqueue/extension/storage/BUILD.bazel index 867eef0f..da7a8357 100644 --- a/submitqueue/extension/storage/BUILD.bazel +++ b/submitqueue/extension/storage/BUILD.bazel @@ -7,6 +7,7 @@ go_library( "batch_store.go", "build_store.go", "change_store.go", + "request_context_store.go", "request_log_store.go", "request_store.go", "speculation_tree_store.go", diff --git a/submitqueue/extension/storage/mock/BUILD.bazel b/submitqueue/extension/storage/mock/BUILD.bazel index 55c5d808..8c7afd4b 100644 --- a/submitqueue/extension/storage/mock/BUILD.bazel +++ b/submitqueue/extension/storage/mock/BUILD.bazel @@ -7,6 +7,7 @@ go_library( "batch_store_mock.go", "build_store_mock.go", "change_store_mock.go", + "request_context_store_mock.go", "request_log_store_mock.go", "request_store_mock.go", "speculation_tree_store_mock.go", diff --git a/submitqueue/extension/storage/mock/request_context_store_mock.go b/submitqueue/extension/storage/mock/request_context_store_mock.go new file mode 100644 index 00000000..87906a4a --- /dev/null +++ b/submitqueue/extension/storage/mock/request_context_store_mock.go @@ -0,0 +1,71 @@ +// Code generated by MockGen. DO NOT EDIT. +// Source: request_context_store.go +// +// Generated by this command: +// +// mockgen -source=request_context_store.go -destination=mock/request_context_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" +) + +// MockRequestContextStore is a mock of RequestContextStore interface. +type MockRequestContextStore struct { + ctrl *gomock.Controller + recorder *MockRequestContextStoreMockRecorder + isgomock struct{} +} + +// MockRequestContextStoreMockRecorder is the mock recorder for MockRequestContextStore. +type MockRequestContextStoreMockRecorder struct { + mock *MockRequestContextStore +} + +// NewMockRequestContextStore creates a new mock instance. +func NewMockRequestContextStore(ctrl *gomock.Controller) *MockRequestContextStore { + mock := &MockRequestContextStore{ctrl: ctrl} + mock.recorder = &MockRequestContextStoreMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *MockRequestContextStore) EXPECT() *MockRequestContextStoreMockRecorder { + return m.recorder +} + +// Create mocks base method. +func (m *MockRequestContextStore) Create(ctx context.Context, arg1 entity.RequestContext) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Create", ctx, arg1) + ret0, _ := ret[0].(error) + return ret0 +} + +// Create indicates an expected call of Create. +func (mr *MockRequestContextStoreMockRecorder) Create(ctx, arg1 any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Create", reflect.TypeOf((*MockRequestContextStore)(nil).Create), ctx, arg1) +} + +// Get mocks base method. +func (m *MockRequestContextStore) Get(ctx context.Context, requestID string) (entity.RequestContext, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Get", ctx, requestID) + ret0, _ := ret[0].(entity.RequestContext) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// Get indicates an expected call of Get. +func (mr *MockRequestContextStoreMockRecorder) Get(ctx, requestID any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Get", reflect.TypeOf((*MockRequestContextStore)(nil).Get), ctx, requestID) +} diff --git a/submitqueue/extension/storage/mock/storage_mock.go b/submitqueue/extension/storage/mock/storage_mock.go index 4133bc2a..e3b89ce6 100644 --- a/submitqueue/extension/storage/mock/storage_mock.go +++ b/submitqueue/extension/storage/mock/storage_mock.go @@ -110,6 +110,20 @@ func (mr *MockStorageMockRecorder) GetChangeStore() *gomock.Call { return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetChangeStore", reflect.TypeOf((*MockStorage)(nil).GetChangeStore)) } +// GetRequestContextStore mocks base method. +func (m *MockStorage) GetRequestContextStore() storage.RequestContextStore { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "GetRequestContextStore") + ret0, _ := ret[0].(storage.RequestContextStore) + return ret0 +} + +// GetRequestContextStore indicates an expected call of GetRequestContextStore. +func (mr *MockStorageMockRecorder) GetRequestContextStore() *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetRequestContextStore", reflect.TypeOf((*MockStorage)(nil).GetRequestContextStore)) +} + // GetRequestLogStore mocks base method. func (m *MockStorage) GetRequestLogStore() storage.RequestLogStore { m.ctrl.T.Helper() diff --git a/submitqueue/extension/storage/mysql/BUILD.bazel b/submitqueue/extension/storage/mysql/BUILD.bazel index c78d27f0..e58c7af1 100644 --- a/submitqueue/extension/storage/mysql/BUILD.bazel +++ b/submitqueue/extension/storage/mysql/BUILD.bazel @@ -7,6 +7,7 @@ go_library( "batch_store.go", "build_store.go", "change_store.go", + "request_context_store.go", "request_log_store.go", "request_store.go", "speculation_tree_store.go", diff --git a/submitqueue/extension/storage/mysql/request_context_store.go b/submitqueue/extension/storage/mysql/request_context_store.go new file mode 100644 index 00000000..f410128b --- /dev/null +++ b/submitqueue/extension/storage/mysql/request_context_store.go @@ -0,0 +1,93 @@ +// 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" + + mysqlDriver "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 requestContextStore struct { + db *sql.DB + scope tally.Scope +} + +// NewRequestContextStore creates a MySQL-backed RequestContextStore. +func NewRequestContextStore(db *sql.DB, scope tally.Scope) storage.RequestContextStore { + return &requestContextStore{db: db, scope: scope} +} + +// Create persists immutable request admission data. +func (s *requestContextStore) Create(ctx context.Context, requestContext entity.RequestContext) (retErr error) { + op := metrics.Begin(s.scope, "create") + defer func() { op.Complete(retErr) }() + + changeURIs := requestContext.ChangeURIs + if changeURIs == nil { + changeURIs = []string{} + } + changeURIsJSON, err := json.Marshal(changeURIs) + if err != nil { + return fmt.Errorf("failed to marshal request context change URIs for request_id=%s: %w", requestContext.RequestID, err) + } + + _, err = s.db.ExecContext(ctx, + "INSERT INTO request_context (request_id, queue, change_uri, admitted_at_ms) VALUES (?, ?, ?, ?)", + requestContext.RequestID, requestContext.Queue, changeURIsJSON, requestContext.AdmittedAtMs, + ) + if err != nil { + var mysqlErr *mysqlDriver.MySQLError + if errors.As(err, &mysqlErr) && mysqlErr.Number == 1062 { + return fmt.Errorf("request context request_id=%s: %w", requestContext.RequestID, storage.ErrAlreadyExists) + } + return fmt.Errorf("failed to create request context request_id=%s: %w", requestContext.RequestID, err) + } + return nil +} + +// Get returns immutable request admission data. +func (s *requestContextStore) Get(ctx context.Context, requestID string) (ret entity.RequestContext, retErr error) { + op := metrics.Begin(s.scope, "get") + defer func() { op.Complete(retErr) }() + + var changeURIsJSON []byte + err := s.db.QueryRowContext(ctx, + "SELECT request_id, queue, change_uri, admitted_at_ms FROM request_context WHERE request_id = ?", + requestID, + ).Scan(&ret.RequestID, &ret.Queue, &changeURIsJSON, &ret.AdmittedAtMs) + if errors.Is(err, sql.ErrNoRows) { + return entity.RequestContext{}, fmt.Errorf("request context request_id=%s: %w", requestID, storage.ErrNotFound) + } + if err != nil { + return entity.RequestContext{}, fmt.Errorf("failed to get request context request_id=%s: %w", requestID, err) + } + if err := json.Unmarshal(changeURIsJSON, &ret.ChangeURIs); err != nil { + return entity.RequestContext{}, fmt.Errorf("failed to unmarshal request context change URIs for request_id=%s: %w", requestID, err) + } + if ret.ChangeURIs == nil { + ret.ChangeURIs = []string{} + } + return ret, nil +} diff --git a/submitqueue/extension/storage/mysql/schema/README.md b/submitqueue/extension/storage/mysql/schema/README.md index 9e8c36bf..03bf0086 100644 --- a/submitqueue/extension/storage/mysql/schema/README.md +++ b/submitqueue/extension/storage/mysql/schema/README.md @@ -22,3 +22,10 @@ As the `batch` table grows, the secondary index will grow with it, increasing st The `change` table records per-URI claims by in-flight requests. `request_id` is part of the primary key so that concurrent claims on the same URI by different requests coexist as distinct rows — a same-request retry collides on the PK and is a no-op (`INSERT IGNORE`), while a different-request claim is a new row that `GetByURI` surfaces for overlap detection. `queue` leads the key so queue-scoped lookups are primary-key-prefix scans and the table is shardable by queue. +## request_context table + +`request_context` stores immutable gateway-owned admission data. It is created once by `Land` and supplies queue, original change URIs, and admission time to the List read model. The `(queue, admitted_at_ms, request_id)` index supports queue-scoped admission-order scans and future context backfills. + +## request_log table + +`request_log` stores immutable request status records. diff --git a/submitqueue/extension/storage/mysql/schema/request_context.sql b/submitqueue/extension/storage/mysql/schema/request_context.sql new file mode 100644 index 00000000..c3960a58 --- /dev/null +++ b/submitqueue/extension/storage/mysql/schema/request_context.sql @@ -0,0 +1,8 @@ +CREATE TABLE IF NOT EXISTS request_context ( + request_id VARCHAR(255) NOT NULL, + queue VARCHAR(255) NOT NULL, + change_uri JSON NOT NULL, + admitted_at_ms BIGINT NOT NULL, + PRIMARY KEY (request_id), + KEY idx_queue_admitted (queue, admitted_at_ms, request_id) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; diff --git a/submitqueue/extension/storage/mysql/storage.go b/submitqueue/extension/storage/mysql/storage.go index 4ba0cf41..81c37a35 100644 --- a/submitqueue/extension/storage/mysql/storage.go +++ b/submitqueue/extension/storage/mysql/storage.go @@ -32,6 +32,7 @@ type mysqlStorage struct { buildStore storage.BuildStore speculationTreeStore storage.SpeculationTreeStore requestLogStore storage.RequestLogStore + requestContextStore storage.RequestContextStore } // NewStorage creates a new MySQL storage. @@ -45,6 +46,7 @@ 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")), + requestContextStore: NewRequestContextStore(db, scope.SubScope("request_context_store")), }, nil } @@ -83,6 +85,11 @@ func (f *mysqlStorage) GetRequestLogStore() storage.RequestLogStore { return f.requestLogStore } +// GetRequestContextStore returns the MySQL-backed RequestContextStore. +func (f *mysqlStorage) GetRequestContextStore() storage.RequestContextStore { + return f.requestContextStore +} + // Close closes the underlying database connection. func (f *mysqlStorage) Close() error { return f.db.Close() diff --git a/submitqueue/extension/storage/request_context_store.go b/submitqueue/extension/storage/request_context_store.go new file mode 100644 index 00000000..a0d7c5a2 --- /dev/null +++ b/submitqueue/extension/storage/request_context_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_context_store.go -destination=mock/request_context_store_mock.go -package=mock + +import ( + "context" + + "github.com/uber/submitqueue/submitqueue/entity" +) + +// RequestContextStore persists immutable gateway-owned request admission data. +type RequestContextStore interface { + // Create persists requestContext. Returns ErrAlreadyExists when any context already exists for the request ID; callers own retry idempotency and conflict detection. + Create(ctx context.Context, requestContext entity.RequestContext) error + + // Get returns the context for requestID. Returns ErrNotFound when it does not exist. + Get(ctx context.Context, requestID string) (entity.RequestContext, error) +} diff --git a/submitqueue/extension/storage/storage.go b/submitqueue/extension/storage/storage.go index a02bef73..6a5db931 100644 --- a/submitqueue/extension/storage/storage.go +++ b/submitqueue/extension/storage/storage.go @@ -65,6 +65,9 @@ type Storage interface { // GetRequestLogStore returns the RequestLogStore instance. GetRequestLogStore() RequestLogStore + // GetRequestContextStore returns the RequestContextStore instance. + GetRequestContextStore() RequestContextStore + // Close closes the storage and all underlying connections. Should only be called once at the end of the program. Close() error }