From d565bc97fa061f2d536ec4b99b202e3f9bee92af Mon Sep 17 00:00:00 2001 From: Albert Wu Date: Tue, 7 Jul 2026 14:02:23 -0700 Subject: [PATCH] feat(summary): project request logs from admission context --- submitqueue/core/request/BUILD.bazel | 2 + submitqueue/core/request/store.go | 161 +++++++++++++ submitqueue/core/request/store_test.go | 101 +++++++++ submitqueue/entity/BUILD.bazel | 1 + submitqueue/entity/request_summary.go | 32 +++ submitqueue/extension/storage/BUILD.bazel | 1 + .../extension/storage/mock/BUILD.bazel | 1 + .../mock/request_summary_store_mock.go | 101 +++++++++ .../extension/storage/mock/storage_mock.go | 14 ++ .../extension/storage/mysql/BUILD.bazel | 1 + .../storage/mysql/request_summary_store.go | 211 ++++++++++++++++++ .../extension/storage/mysql/schema/README.md | 4 + .../storage/mysql/schema/request_summary.sql | 17 ++ .../extension/storage/mysql/storage.go | 7 + .../storage/request_summary_store.go | 62 +++++ submitqueue/extension/storage/storage.go | 3 + 16 files changed, 719 insertions(+) create mode 100644 submitqueue/core/request/store.go create mode 100644 submitqueue/core/request/store_test.go create mode 100644 submitqueue/entity/request_summary.go create mode 100644 submitqueue/extension/storage/mock/request_summary_store_mock.go create mode 100644 submitqueue/extension/storage/mysql/request_summary_store.go create mode 100644 submitqueue/extension/storage/mysql/schema/request_summary.sql create mode 100644 submitqueue/extension/storage/request_summary_store.go diff --git a/submitqueue/core/request/BUILD.bazel b/submitqueue/core/request/BUILD.bazel index c2a6e589..488df0eb 100644 --- a/submitqueue/core/request/BUILD.bazel +++ b/submitqueue/core/request/BUILD.bazel @@ -5,6 +5,7 @@ go_library( srcs = [ "log.go", "request.go", + "store.go", ], importpath = "github.com/uber/submitqueue/submitqueue/core/request", visibility = ["//visibility:public"], @@ -22,6 +23,7 @@ go_test( srcs = [ "log_test.go", "request_test.go", + "store_test.go", ], embed = [":request"], deps = [ diff --git a/submitqueue/core/request/store.go b/submitqueue/core/request/store.go new file mode 100644 index 00000000..f4b72e03 --- /dev/null +++ b/submitqueue/core/request/store.go @@ -0,0 +1,161 @@ +// 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 request + +import ( + "context" + "errors" + "fmt" + + "github.com/uber/submitqueue/submitqueue/entity" + "github.com/uber/submitqueue/submitqueue/extension/storage" +) + +const maxSummaryProjectionAttempts = 8 + +// CreateContext creates immutable admission data and the initial List projection. +func CreateContext(ctx context.Context, store storage.Storage, requestContext entity.RequestContext) error { + if err := store.GetRequestContextStore().Create(ctx, requestContext); err != nil { + if !errors.Is(err, storage.ErrAlreadyExists) { + return fmt.Errorf("failed to create request context request_id=%s: %w", requestContext.RequestID, err) + } + existing, getErr := store.GetRequestContextStore().Get(ctx, requestContext.RequestID) + if getErr != nil { + return fmt.Errorf("failed to verify existing request context request_id=%s: %w", requestContext.RequestID, getErr) + } + if !equalRequestContext(existing, requestContext) { + return fmt.Errorf("request context conflicts with existing immutable context request_id=%s", requestContext.RequestID) + } + } + + summary := entity.RequestSummary{ + RequestID: requestContext.RequestID, + Queue: requestContext.Queue, + ChangeURIs: append([]string(nil), requestContext.ChangeURIs...), + Status: entity.RequestStatusAccepted, + Metadata: map[string]string{}, + StartedAtMs: requestContext.AdmittedAtMs, + UpdatedAtMs: requestContext.AdmittedAtMs, + StatusTimestampMs: requestContext.AdmittedAtMs, + Version: 1, + } + if err := store.GetRequestSummaryStore().Create(ctx, summary); err != nil && !errors.Is(err, storage.ErrAlreadyExists) { + return fmt.Errorf("failed to create request summary request_id=%s: %w", requestContext.RequestID, err) + } + return nil +} + +// PersistLog appends one immutable status event and updates its summary projection. A projection error is returned so queue delivery retries. +func PersistLog(ctx context.Context, store storage.Storage, log entity.RequestLog) error { + if err := store.GetRequestLogStore().Insert(ctx, log); err != nil { + return fmt.Errorf("failed to insert request log request_id=%s: %w", log.RequestID, err) + } + if err := ProjectLog(ctx, store, log); err != nil { + return fmt.Errorf("failed to project request log request_id=%s: %w", log.RequestID, err) + } + return nil +} + +// ProjectLog applies a status event to an existing summary using optimistic concurrency. +func ProjectLog(ctx context.Context, store storage.Storage, log entity.RequestLog) error { + requestContext, err := store.GetRequestContextStore().Get(ctx, log.RequestID) + if err != nil { + return fmt.Errorf("failed to get request context request_id=%s: %w", log.RequestID, err) + } + + for attempt := 0; attempt < maxSummaryProjectionAttempts; attempt++ { + existing, err := store.GetRequestSummaryStore().Get(ctx, requestContext.Queue, log.RequestID) + if err != nil { + return fmt.Errorf("failed to get request summary request_id=%s: %w", log.RequestID, err) + } + next := MergeSummary(existing, log) + newVersion := existing.Version + 1 + if err := store.GetRequestSummaryStore().Update(ctx, next, existing.Version, newVersion); err == nil { + return nil + } else if !errors.Is(err, storage.ErrVersionMismatch) { + return fmt.Errorf("failed to update request summary request_id=%s: %w", log.RequestID, err) + } + } + return fmt.Errorf("request summary projection did not converge request_id=%s: %w", log.RequestID, storage.ErrVersionMismatch) +} + +// MergeSummary returns the summary that results when log is reconciled against existing. +func MergeSummary(existing entity.RequestSummary, log entity.RequestLog) entity.RequestSummary { + next := existing + if !shouldReplaceWinner(existing, log) { + return next + } + next.Status = log.Status + next.LastError = log.LastError + next.Metadata = cloneMetadata(log.Metadata) + next.UpdatedAtMs = log.TimestampMs + next.RequestVersion = log.RequestVersion + next.StatusTimestampMs = log.TimestampMs + next.WinnerTerminalVersion = isTerminalVersion(log) + if entity.IsRequestStateTerminal(entity.RequestState(string(log.Status))) { + next.CompletedAtMs = log.TimestampMs + } else { + next.CompletedAtMs = 0 + } + return next +} + +func shouldReplaceWinner(existing entity.RequestSummary, log entity.RequestLog) bool { + incomingTerminalVersion := isTerminalVersion(log) + if incomingTerminalVersion { + if !existing.WinnerTerminalVersion { + return true + } + return log.RequestVersion > existing.RequestVersion || (log.RequestVersion == existing.RequestVersion && log.TimestampMs > existing.StatusTimestampMs) + } + if existing.WinnerTerminalVersion { + return false + } + return log.TimestampMs > existing.StatusTimestampMs +} + +func isTerminalVersion(log entity.RequestLog) bool { + return log.RequestVersion > 0 && entity.IsRequestStateTerminal(entity.RequestState(string(log.Status))) +} + +func cloneMetadata(metadata map[string]string) map[string]string { + if metadata == nil { + return map[string]string{} + } + clone := make(map[string]string, len(metadata)) + for key, value := range metadata { + clone[key] = value + } + return clone +} + +func equalRequestContext(left, right entity.RequestContext) bool { + return left.RequestID == right.RequestID && + left.Queue == right.Queue && + left.AdmittedAtMs == right.AdmittedAtMs && + equalChangeURIs(left.ChangeURIs, right.ChangeURIs) +} + +func equalChangeURIs(left, right []string) bool { + if len(left) != len(right) { + return false + } + for index := range left { + if left[index] != right[index] { + return false + } + } + return true +} diff --git a/submitqueue/core/request/store_test.go b/submitqueue/core/request/store_test.go new file mode 100644 index 00000000..8854df65 --- /dev/null +++ b/submitqueue/core/request/store_test.go @@ -0,0 +1,101 @@ +// 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 request + +import ( + "context" + "testing" + + "github.com/stretchr/testify/require" + "github.com/uber/submitqueue/submitqueue/entity" + "github.com/uber/submitqueue/submitqueue/extension/storage" + storagemock "github.com/uber/submitqueue/submitqueue/extension/storage/mock" + "go.uber.org/mock/gomock" +) + +func TestCreateContext(t *testing.T) { + requestContext := entity.RequestContext{ + RequestID: "q/1", + Queue: "q", + ChangeURIs: []string{"github://uber/repo/pull/1/abc"}, + AdmittedAtMs: 100, + } + + tests := []struct { + name string + stored entity.RequestContext + wantErr bool + setupCreate func(*storagemock.MockRequestContextStore) + }{ + { + name: "new context", + setupCreate: func(contextStore *storagemock.MockRequestContextStore) { + contextStore.EXPECT().Create(gomock.Any(), requestContext).Return(nil) + }, + }, + { + name: "identical retry", + stored: requestContext, + setupCreate: func(contextStore *storagemock.MockRequestContextStore) { + contextStore.EXPECT().Create(gomock.Any(), requestContext).Return(storage.ErrAlreadyExists) + contextStore.EXPECT().Get(gomock.Any(), requestContext.RequestID).Return(requestContext, nil) + }, + }, + { + name: "conflicting retry", + stored: entity.RequestContext{ + RequestID: requestContext.RequestID, + Queue: "other", + }, + wantErr: true, + setupCreate: func(contextStore *storagemock.MockRequestContextStore) { + contextStore.EXPECT().Create(gomock.Any(), requestContext).Return(storage.ErrAlreadyExists) + contextStore.EXPECT().Get(gomock.Any(), requestContext.RequestID).Return(entity.RequestContext{RequestID: requestContext.RequestID, Queue: "other"}, nil) + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + contextStore := storagemock.NewMockRequestContextStore(ctrl) + tt.setupCreate(contextStore) + summaryStore := storagemock.NewMockRequestSummaryStore(ctrl) + if !tt.wantErr { + summaryStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil) + } + store := storagemock.NewMockStorage(ctrl) + store.EXPECT().GetRequestContextStore().Return(contextStore).AnyTimes() + if !tt.wantErr { + store.EXPECT().GetRequestSummaryStore().Return(summaryStore) + } + + err := CreateContext(context.Background(), store, requestContext) + if tt.wantErr { + require.Error(t, err) + return + } + require.NoError(t, err) + }) + } +} + +func TestEqualRequestContext_NormalizesEmptyChangeURIs(t *testing.T) { + left := entity.RequestContext{RequestID: "q/1", Queue: "q", AdmittedAtMs: 100} + right := left + right.ChangeURIs = []string{} + + require.True(t, equalRequestContext(left, right)) +} diff --git a/submitqueue/entity/BUILD.bazel b/submitqueue/entity/BUILD.bazel index f997cfab..26383df9 100644 --- a/submitqueue/entity/BUILD.bazel +++ b/submitqueue/entity/BUILD.bazel @@ -18,6 +18,7 @@ go_library( "request.go", "request_context.go", "request_log.go", + "request_summary.go", "speculation_tree.go", ], importpath = "github.com/uber/submitqueue/submitqueue/entity", diff --git a/submitqueue/entity/request_summary.go b/submitqueue/entity/request_summary.go new file mode 100644 index 00000000..04ef983e --- /dev/null +++ b/submitqueue/entity/request_summary.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 entity + +// RequestSummary is the gateway-owned current List projection. +type RequestSummary struct { + RequestID string + Queue string + ChangeURIs []string + Status RequestStatus + LastError string + Metadata map[string]string + StartedAtMs int64 + UpdatedAtMs int64 + CompletedAtMs int64 + RequestVersion int32 + StatusTimestampMs int64 + WinnerTerminalVersion bool + Version int64 +} diff --git a/submitqueue/extension/storage/BUILD.bazel b/submitqueue/extension/storage/BUILD.bazel index da7a8357..86297154 100644 --- a/submitqueue/extension/storage/BUILD.bazel +++ b/submitqueue/extension/storage/BUILD.bazel @@ -10,6 +10,7 @@ go_library( "request_context_store.go", "request_log_store.go", "request_store.go", + "request_summary_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 8c7afd4b..218e6325 100644 --- a/submitqueue/extension/storage/mock/BUILD.bazel +++ b/submitqueue/extension/storage/mock/BUILD.bazel @@ -10,6 +10,7 @@ go_library( "request_context_store_mock.go", "request_log_store_mock.go", "request_store_mock.go", + "request_summary_store_mock.go", "speculation_tree_store_mock.go", "storage_mock.go", ], 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..070e0045 --- /dev/null +++ b/submitqueue/extension/storage/mock/request_summary_store_mock.go @@ -0,0 +1,101 @@ +// 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" + storage "github.com/uber/submitqueue/submitqueue/extension/storage" + 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, queue, requestID string) (entity.RequestSummary, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Get", ctx, queue, 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, queue, requestID any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Get", reflect.TypeOf((*MockRequestSummaryStore)(nil).Get), ctx, queue, requestID) +} + +// List mocks base method. +func (m *MockRequestSummaryStore) List(ctx context.Context, opts storage.RequestSummaryListOptions) (storage.RequestSummaryListResult, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "List", ctx, opts) + ret0, _ := ret[0].(storage.RequestSummaryListResult) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// List indicates an expected call of List. +func (mr *MockRequestSummaryStoreMockRecorder) List(ctx, opts any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "List", reflect.TypeOf((*MockRequestSummaryStore)(nil).List), ctx, opts) +} + +// Update mocks base method. +func (m *MockRequestSummaryStore) Update(ctx context.Context, summary entity.RequestSummary, oldVersion, newVersion int64) 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/storage_mock.go b/submitqueue/extension/storage/mock/storage_mock.go index e3b89ce6..7b60b949 100644 --- a/submitqueue/extension/storage/mock/storage_mock.go +++ b/submitqueue/extension/storage/mock/storage_mock.go @@ -152,6 +152,20 @@ func (mr *MockStorageMockRecorder) GetRequestStore() *gomock.Call { return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetRequestStore", reflect.TypeOf((*MockStorage)(nil).GetRequestStore)) } +// GetRequestSummaryStore mocks base method. +func (m *MockStorage) GetRequestSummaryStore() storage.RequestSummaryStore { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "GetRequestSummaryStore") + ret0, _ := ret[0].(storage.RequestSummaryStore) + return ret0 +} + +// GetRequestSummaryStore indicates an expected call of GetRequestSummaryStore. +func (mr *MockStorageMockRecorder) GetRequestSummaryStore() *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetRequestSummaryStore", reflect.TypeOf((*MockStorage)(nil).GetRequestSummaryStore)) +} + // 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 e58c7af1..ceef201a 100644 --- a/submitqueue/extension/storage/mysql/BUILD.bazel +++ b/submitqueue/extension/storage/mysql/BUILD.bazel @@ -10,6 +10,7 @@ go_library( "request_context_store.go", "request_log_store.go", "request_store.go", + "request_summary_store.go", "speculation_tree_store.go", "storage.go", ], 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..20e9dea5 --- /dev/null +++ b/submitqueue/extension/storage/mysql/request_summary_store.go @@ -0,0 +1,211 @@ +// 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" + "strings" + + 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 requestSummaryStore struct { + db *sql.DB + scope tally.Scope +} + +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) + if err != nil { + return err + } + _, err = s.db.ExecContext(ctx, + "INSERT INTO request_summary (request_id, queue, change_uri, status, request_version, status_timestamp_ms, winner_terminal_version, last_error, metadata, started_at_ms, updated_at_ms, completed_at_ms, version) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + summary.RequestID, summary.Queue, changeURIsJSON, summary.Status, summary.RequestVersion, summary.StatusTimestampMs, summary.WinnerTerminalVersion, summary.LastError, metadataJSON, summary.StartedAtMs, summary.UpdatedAtMs, summary.CompletedAtMs, summary.Version, + ) + if err != nil { + var mysqlErr *mysqlDriver.MySQLError + if errors.As(err, &mysqlErr) && mysqlErr.Number == 1062 { + return fmt.Errorf("request summary request_id=%s: %w", summary.RequestID, storage.ErrAlreadyExists) + } + return fmt.Errorf("failed to create request summary request_id=%s: %w", summary.RequestID, err) + } + return nil +} + +func (s *requestSummaryStore) Get(ctx context.Context, queue, requestID string) (ret entity.RequestSummary, retErr error) { + op := metrics.Begin(s.scope, "get") + defer func() { op.Complete(retErr) }() + + row := s.db.QueryRowContext(ctx, + "SELECT request_id, queue, change_uri, status, request_version, status_timestamp_ms, winner_terminal_version, last_error, metadata, started_at_ms, updated_at_ms, completed_at_ms, version FROM request_summary WHERE queue = ? AND request_id = ?", + queue, requestID, + ) + ret, err := scanRequestSummary(row) + if errors.Is(err, sql.ErrNoRows) { + return entity.RequestSummary{}, fmt.Errorf("request summary request_id=%s: %w", requestID, storage.ErrNotFound) + } + if err != nil { + return entity.RequestSummary{}, err + } + return ret, nil +} + +func (s *requestSummaryStore) Update(ctx context.Context, summary entity.RequestSummary, oldVersion, newVersion int64) (retErr error) { + op := metrics.Begin(s.scope, "update") + defer func() { op.Complete(retErr) }() + + changeURIsJSON, metadataJSON, err := marshalSummaryJSON(summary) + if err != nil { + return err + } + result, err := s.db.ExecContext(ctx, + "UPDATE request_summary SET change_uri = ?, status = ?, request_version = ?, status_timestamp_ms = ?, winner_terminal_version = ?, last_error = ?, metadata = ?, started_at_ms = ?, updated_at_ms = ?, completed_at_ms = ?, version = ? WHERE queue = ? AND request_id = ? AND version = ?", + changeURIsJSON, summary.Status, summary.RequestVersion, summary.StatusTimestampMs, summary.WinnerTerminalVersion, summary.LastError, metadataJSON, summary.StartedAtMs, summary.UpdatedAtMs, summary.CompletedAtMs, newVersion, summary.Queue, summary.RequestID, oldVersion, + ) + if err != nil { + return fmt.Errorf("failed to update request summary request_id=%s: %w", summary.RequestID, err) + } + affected, err := result.RowsAffected() + if err != nil { + return fmt.Errorf("failed to read updated request summary rows request_id=%s: %w", summary.RequestID, err) + } + if affected == 0 { + return fmt.Errorf("request summary request_id=%s old_version=%d: %w", summary.RequestID, oldVersion, storage.ErrVersionMismatch) + } + return nil +} + +func (s *requestSummaryStore) List(ctx context.Context, opts storage.RequestSummaryListOptions) (ret storage.RequestSummaryListResult, retErr error) { + op := metrics.Begin(s.scope, "list") + defer func() { op.Complete(retErr) }() + if opts.Limit <= 0 { + return storage.RequestSummaryListResult{}, fmt.Errorf("request summary list requires a positive limit") + } + cursorClause, orderBy, err := listSortSQL(opts.Sort) + if err != nil { + return storage.RequestSummaryListResult{}, err + } + args := []any{opts.Queue, opts.StartTimeMs, opts.EndTimeMs} + clauses := []string{"queue = ?", "started_at_ms >= ?", "started_at_ms < ?"} + if len(opts.Statuses) > 0 { + placeholders := make([]string, len(opts.Statuses)) + for index, status := range opts.Statuses { + placeholders[index] = "?" + args = append(args, status) + } + clauses = append(clauses, "status IN ("+strings.Join(placeholders, ", ")+")") + } + if opts.Cursor != nil { + clauses = append(clauses, cursorClause) + args = append(args, opts.Cursor.StartedAtMs, opts.Cursor.StartedAtMs, opts.Cursor.RequestID) + } + args = append(args, opts.Limit+1) + query := "SELECT request_id, queue, change_uri, status, request_version, status_timestamp_ms, winner_terminal_version, last_error, metadata, started_at_ms, updated_at_ms, completed_at_ms, version FROM request_summary WHERE " + strings.Join(clauses, " AND ") + orderBy + " LIMIT ?" + rows, err := s.db.QueryContext(ctx, query, args...) + if err != nil { + return storage.RequestSummaryListResult{}, fmt.Errorf("failed to list request summaries for queue=%s: %w", opts.Queue, err) + } + defer rows.Close() + var summaries []entity.RequestSummary + for rows.Next() { + summary, err := scanRequestSummary(rows) + if err != nil { + return storage.RequestSummaryListResult{}, err + } + summaries = append(summaries, summary) + } + if err := rows.Err(); err != nil { + return storage.RequestSummaryListResult{}, fmt.Errorf("failed to iterate request summaries for queue=%s: %w", opts.Queue, err) + } + result := storage.RequestSummaryListResult{Requests: summaries} + if len(result.Requests) > opts.Limit { + result.Requests = result.Requests[:opts.Limit] + last := result.Requests[len(result.Requests)-1] + result.NextCursor = &storage.RequestSummaryCursor{StartedAtMs: last.StartedAtMs, RequestID: last.RequestID} + } + return result, nil +} + +func listSortSQL(sort storage.RequestSummarySort) (string, string, error) { + switch sort { + case "", storage.RequestSummarySortAdmittedAsc: + return "(started_at_ms > ? OR (started_at_ms = ? AND request_id > ?))", " ORDER BY started_at_ms ASC, request_id ASC", nil + case storage.RequestSummarySortAdmittedDesc: + return "(started_at_ms < ? OR (started_at_ms = ? AND request_id < ?))", " ORDER BY started_at_ms DESC, request_id DESC", nil + default: + return "", "", fmt.Errorf("unsupported request summary sort %q", sort) + } +} + +type summaryScanner interface{ Scan(...any) error } + +func scanRequestSummary(scanner summaryScanner) (ret entity.RequestSummary, retErr error) { + var changeURIsJSON, metadataJSON []byte + var completedAtMs int64 + if err := scanner.Scan(&ret.RequestID, &ret.Queue, &changeURIsJSON, &ret.Status, &ret.RequestVersion, &ret.StatusTimestampMs, &ret.WinnerTerminalVersion, &ret.LastError, &metadataJSON, &ret.StartedAtMs, &ret.UpdatedAtMs, &completedAtMs, &ret.Version); err != nil { + return entity.RequestSummary{}, fmt.Errorf("failed to scan request summary: %w", err) + } + if err := json.Unmarshal(changeURIsJSON, &ret.ChangeURIs); err != nil { + return entity.RequestSummary{}, fmt.Errorf("failed to unmarshal request summary change URIs request_id=%s: %w", ret.RequestID, err) + } + if err := json.Unmarshal(metadataJSON, &ret.Metadata); err != nil { + return entity.RequestSummary{}, fmt.Errorf("failed to unmarshal request summary metadata request_id=%s: %w", ret.RequestID, err) + } + if ret.ChangeURIs == nil { + ret.ChangeURIs = []string{} + } + if ret.Metadata == nil { + ret.Metadata = map[string]string{} + } + ret.CompletedAtMs = completedAtMs + return ret, nil +} + +func marshalSummaryJSON(summary entity.RequestSummary) ([]byte, []byte, error) { + changeURIs := summary.ChangeURIs + if changeURIs == nil { + changeURIs = []string{} + } + metadata := summary.Metadata + if metadata == nil { + metadata = map[string]string{} + } + changeURIsJSON, err := json.Marshal(changeURIs) + if err != nil { + return nil, nil, fmt.Errorf("failed to marshal request summary change URIs request_id=%s: %w", summary.RequestID, err) + } + metadataJSON, err := json.Marshal(metadata) + if err != nil { + return nil, nil, fmt.Errorf("failed to marshal request summary metadata request_id=%s: %w", summary.RequestID, err) + } + return changeURIsJSON, metadataJSON, nil +} diff --git a/submitqueue/extension/storage/mysql/schema/README.md b/submitqueue/extension/storage/mysql/schema/README.md index 03bf0086..48da28c4 100644 --- a/submitqueue/extension/storage/mysql/schema/README.md +++ b/submitqueue/extension/storage/mysql/schema/README.md @@ -29,3 +29,7 @@ The `change` table records per-URI claims by in-flight requests. `request_id` is ## request_log table `request_log` stores immutable request status records. + +## request_summary table + +`request_summary` stores the current gateway-owned List projection. Its primary key supports point reads and conditional updates. The `(queue, started_at_ms, request_id)` index supports the bounded admission-time range and admitted-order keyset scans used by List; any additional status index must be justified with representative `EXPLAIN` plans. diff --git a/submitqueue/extension/storage/mysql/schema/request_summary.sql b/submitqueue/extension/storage/mysql/schema/request_summary.sql new file mode 100644 index 00000000..562a80ae --- /dev/null +++ b/submitqueue/extension/storage/mysql/schema/request_summary.sql @@ -0,0 +1,17 @@ +CREATE TABLE IF NOT EXISTS request_summary ( + request_id VARCHAR(255) NOT NULL, + queue VARCHAR(255) NOT NULL, + change_uri JSON NOT NULL, + status VARCHAR(64) NOT NULL, + request_version INT NOT NULL, + status_timestamp_ms BIGINT NOT NULL, + winner_terminal_version BOOLEAN NOT NULL, + last_error TEXT NOT NULL, + metadata JSON NOT NULL, + started_at_ms BIGINT NOT NULL, + updated_at_ms BIGINT NOT NULL, + completed_at_ms BIGINT NOT NULL, + version BIGINT NOT NULL, + PRIMARY KEY (queue, request_id), + KEY idx_queue_started (queue, started_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 81c37a35..8146382d 100644 --- a/submitqueue/extension/storage/mysql/storage.go +++ b/submitqueue/extension/storage/mysql/storage.go @@ -33,6 +33,7 @@ type mysqlStorage struct { speculationTreeStore storage.SpeculationTreeStore requestLogStore storage.RequestLogStore requestContextStore storage.RequestContextStore + requestSummaryStore storage.RequestSummaryStore } // NewStorage creates a new MySQL storage. @@ -47,6 +48,7 @@ func NewStorage(db *sql.DB, scope tally.Scope) (storage.Storage, error) { speculationTreeStore: NewSpeculationTreeStore(db, scope.SubScope("speculation_tree_store")), requestLogStore: NewRequestLogStore(db, scope.SubScope("request_log_store")), requestContextStore: NewRequestContextStore(db, scope.SubScope("request_context_store")), + requestSummaryStore: NewRequestSummaryStore(db, scope.SubScope("request_summary_store")), }, nil } @@ -90,6 +92,11 @@ func (f *mysqlStorage) GetRequestContextStore() storage.RequestContextStore { return f.requestContextStore } +// GetRequestSummaryStore returns the MySQL-backed RequestSummaryStore. +func (f *mysqlStorage) GetRequestSummaryStore() storage.RequestSummaryStore { + return f.requestSummaryStore +} + // Close closes the underlying database connection. func (f *mysqlStorage) Close() error { return f.db.Close() diff --git a/submitqueue/extension/storage/request_summary_store.go b/submitqueue/extension/storage/request_summary_store.go new file mode 100644 index 00000000..05b0e52d --- /dev/null +++ b/submitqueue/extension/storage/request_summary_store.go @@ -0,0 +1,62 @@ +// 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" +) + +// RequestSummarySort defines deterministic request-summary page orderings. +type RequestSummarySort string + +const ( + RequestSummarySortAdmittedAsc RequestSummarySort = "admitted_asc" + RequestSummarySortAdmittedDesc RequestSummarySort = "admitted_desc" +) + +// RequestSummaryStore persists and queries the gateway List read model. Projection decisions belong to callers. +type RequestSummaryStore interface { + Create(ctx context.Context, summary entity.RequestSummary) error + Get(ctx context.Context, queue, requestID string) (entity.RequestSummary, error) + Update(ctx context.Context, summary entity.RequestSummary, oldVersion, newVersion int64) error + List(ctx context.Context, opts RequestSummaryListOptions) (RequestSummaryListResult, error) +} + +// RequestSummaryListOptions defines a queue-scoped admission-time page query. Queue, positive StartTimeMs and EndTimeMs values satisfying StartTimeMs < EndTimeMs, and a positive Limit are required. Empty Statuses disables status filtering. Empty Sort selects RequestSummarySortAdmittedAsc. Nil Cursor selects the first page; a non-nil Cursor continues strictly after its complete (StartedAtMs, RequestID) ordering key. +type RequestSummaryListOptions struct { + Queue string + StartTimeMs int64 + EndTimeMs int64 + Statuses []entity.RequestStatus + Sort RequestSummarySort + // Cursor is nil for the first page; otherwise it is the last-seen ordering key. + Cursor *RequestSummaryCursor + Limit int +} + +// RequestSummaryCursor is the complete last-seen (started_at_ms, request_id) ordering key. +type RequestSummaryCursor struct { + StartedAtMs int64 + RequestID string +} + +type RequestSummaryListResult struct { + Requests []entity.RequestSummary + NextCursor *RequestSummaryCursor +} diff --git a/submitqueue/extension/storage/storage.go b/submitqueue/extension/storage/storage.go index 6a5db931..6773f1f0 100644 --- a/submitqueue/extension/storage/storage.go +++ b/submitqueue/extension/storage/storage.go @@ -68,6 +68,9 @@ type Storage interface { // GetRequestContextStore returns the RequestContextStore instance. GetRequestContextStore() RequestContextStore + // GetRequestSummaryStore returns the RequestSummaryStore instance. + GetRequestSummaryStore() RequestSummaryStore + // Close closes the storage and all underlying connections. Should only be called once at the end of the program. Close() error }