From a8b64066d3e0857903225bf3ddbdd40981c01967 Mon Sep 17 00:00:00 2001 From: "prath.shenoy" Date: Mon, 9 Nov 2026 03:51:10 +0000 Subject: [PATCH] feat(stovepipe): Add request project issue lookup ## Summary **What**: - Add storage for listing a project's attributed issues within a request. **Why**: - Support viewing the failures behind a project's result for a particular validation. Attribution is recorded by issue, so finding one project's issues otherwise requires reading all of the request's associations. A project-first lookup makes that listing a bounded read. ## Test Plan - [x] Add unit tests. - [ ] Run the MySQL integration tests when Docker is available. ## Revert Plan - Revert this PR. No runtime producer or API reads the new lookup yet. ## Issue Links - [CODEM-607](https://linear.app/uber/issue/CODEM-607) --- stovepipe/extension/storage/BUILD.bazel | 1 + stovepipe/extension/storage/README.md | 6 + stovepipe/extension/storage/mock/BUILD.bazel | 1 + .../request_issues_by_project_store_mock.go | 71 ++++++++ .../extension/storage/mock/storage_mock.go | 14 ++ stovepipe/extension/storage/mysql/BUILD.bazel | 2 + .../mysql/request_issues_by_project_store.go | 119 ++++++++++++ .../request_issues_by_project_store_test.go | 170 ++++++++++++++++++ .../mysql/request_projects_by_issue_store.go | 10 +- .../schema/request_issues_by_project.sql | 8 + stovepipe/extension/storage/mysql/storage.go | 6 + .../extension/storage/mysql/storage_test.go | 8 + .../request_issues_by_project_store.go | 34 ++++ stovepipe/extension/storage/storage.go | 2 + .../extension/storage/mysql/BUILD.bazel | 1 + .../mysql/request_issues_by_project_test.go | 103 +++++++++++ .../extension/storage/mysql/storage_test.go | 11 ++ 17 files changed, 562 insertions(+), 5 deletions(-) create mode 100644 stovepipe/extension/storage/mock/request_issues_by_project_store_mock.go create mode 100644 stovepipe/extension/storage/mysql/request_issues_by_project_store.go create mode 100644 stovepipe/extension/storage/mysql/request_issues_by_project_store_test.go create mode 100644 stovepipe/extension/storage/mysql/schema/request_issues_by_project.sql create mode 100644 stovepipe/extension/storage/request_issues_by_project_store.go create mode 100644 test/integration/stovepipe/extension/storage/mysql/request_issues_by_project_test.go diff --git a/stovepipe/extension/storage/BUILD.bazel b/stovepipe/extension/storage/BUILD.bazel index 8e18e2701..42618493f 100644 --- a/stovepipe/extension/storage/BUILD.bazel +++ b/stovepipe/extension/storage/BUILD.bazel @@ -16,6 +16,7 @@ go_library( "queue_summary_store.go", "request_acceptance_store.go", "request_issue_summary_store.go", + "request_issues_by_project_store.go", "request_log_store.go", "request_projects_by_issue_store.go", "request_store.go", diff --git a/stovepipe/extension/storage/README.md b/stovepipe/extension/storage/README.md index a8d81680c..b56cc8ca2 100644 --- a/stovepipe/extension/storage/README.md +++ b/stovepipe/extension/storage/README.md @@ -34,6 +34,12 @@ Attribution is assumed stable for a request. Retries recreate the same pairs and Listing fixes the queue/request prefix and returns bounded pages in ascending bytewise task/unit/project order. The cursor excludes its key. An empty page ends that range; it does not establish that attribution completed. The MySQL key uses the issue stores' task/unit capacities and a 1020-byte project name, matching the byte capacity of existing project names. Create the source table before deploying writers. This store does not maintain current project-issue lookups. +## Request issues by project + +`RequestIssuesByProjectStore` is a derived lookup keyed by `(queue, request_id, project, task_id, unit_id)`. It reuses the attribution row's issue/project identity with project-first ordering within the request. Entries describe the failure associations recorded for that validation; they do not change when an issue later closes. + +Create returns `ErrAlreadyExists` for a repeated entry. Listing fixes the queue/request/project prefix and returns bounded pages in ascending bytewise task/unit order. The zero cursor starts the range; other cursors must match the project and exclude their task/unit key. An empty page ends the range, not evidence that attribution recording is complete. There are no update or delete operations. + ## Build issue observations `BuildIssueStore` accumulates observations identified by build, task, and unit within the bound queue. The build ID is supplied separately from the detected issue. Repeated observations return `ErrAlreadyExists`, which callers may treat as successful persistence. Build and task IDs must be nonempty; empty unit IDs represent task-wide observations. diff --git a/stovepipe/extension/storage/mock/BUILD.bazel b/stovepipe/extension/storage/mock/BUILD.bazel index 18bfa988c..34e532dc3 100644 --- a/stovepipe/extension/storage/mock/BUILD.bazel +++ b/stovepipe/extension/storage/mock/BUILD.bazel @@ -16,6 +16,7 @@ go_library( "queue_summary_store_mock.go", "request_acceptance_store_mock.go", "request_issue_summary_store_mock.go", + "request_issues_by_project_store_mock.go", "request_log_store_mock.go", "request_projects_by_issue_store_mock.go", "request_store_mock.go", diff --git a/stovepipe/extension/storage/mock/request_issues_by_project_store_mock.go b/stovepipe/extension/storage/mock/request_issues_by_project_store_mock.go new file mode 100644 index 000000000..2132e8f36 --- /dev/null +++ b/stovepipe/extension/storage/mock/request_issues_by_project_store_mock.go @@ -0,0 +1,71 @@ +// Code generated by MockGen. DO NOT EDIT. +// Source: request_issues_by_project_store.go +// +// Generated by this command: +// +// mockgen -source=request_issues_by_project_store.go -destination=mock/request_issues_by_project_store_mock.go -package=mock +// + +// Package mock is a generated GoMock package. +package mock + +import ( + context "context" + reflect "reflect" + + entity "github.com/uber/submitqueue/stovepipe/entity" + gomock "go.uber.org/mock/gomock" +) + +// MockRequestIssuesByProjectStore is a mock of RequestIssuesByProjectStore interface. +type MockRequestIssuesByProjectStore struct { + ctrl *gomock.Controller + recorder *MockRequestIssuesByProjectStoreMockRecorder + isgomock struct{} +} + +// MockRequestIssuesByProjectStoreMockRecorder is the mock recorder for MockRequestIssuesByProjectStore. +type MockRequestIssuesByProjectStoreMockRecorder struct { + mock *MockRequestIssuesByProjectStore +} + +// NewMockRequestIssuesByProjectStore creates a new mock instance. +func NewMockRequestIssuesByProjectStore(ctrl *gomock.Controller) *MockRequestIssuesByProjectStore { + mock := &MockRequestIssuesByProjectStore{ctrl: ctrl} + mock.recorder = &MockRequestIssuesByProjectStoreMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *MockRequestIssuesByProjectStore) EXPECT() *MockRequestIssuesByProjectStoreMockRecorder { + return m.recorder +} + +// Create mocks base method. +func (m *MockRequestIssuesByProjectStore) Create(ctx context.Context, requestID string, entry entity.ProjectIssueAttribution) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Create", ctx, requestID, entry) + ret0, _ := ret[0].(error) + return ret0 +} + +// Create indicates an expected call of Create. +func (mr *MockRequestIssuesByProjectStoreMockRecorder) Create(ctx, requestID, entry any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Create", reflect.TypeOf((*MockRequestIssuesByProjectStore)(nil).Create), ctx, requestID, entry) +} + +// List mocks base method. +func (m *MockRequestIssuesByProjectStore) List(ctx context.Context, requestID, project string, after entity.ProjectIssueAttribution, limit int) ([]entity.ProjectIssueAttribution, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "List", ctx, requestID, project, after, limit) + ret0, _ := ret[0].([]entity.ProjectIssueAttribution) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// List indicates an expected call of List. +func (mr *MockRequestIssuesByProjectStoreMockRecorder) List(ctx, requestID, project, after, limit any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "List", reflect.TypeOf((*MockRequestIssuesByProjectStore)(nil).List), ctx, requestID, project, after, limit) +} diff --git a/stovepipe/extension/storage/mock/storage_mock.go b/stovepipe/extension/storage/mock/storage_mock.go index c6b62b686..05f3df487 100644 --- a/stovepipe/extension/storage/mock/storage_mock.go +++ b/stovepipe/extension/storage/mock/storage_mock.go @@ -261,6 +261,20 @@ func (mr *MockStorageMockRecorder) GetRequestIssueSummaryStore() *gomock.Call { return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetRequestIssueSummaryStore", reflect.TypeOf((*MockStorage)(nil).GetRequestIssueSummaryStore)) } +// GetRequestIssuesByProjectStore mocks base method. +func (m *MockStorage) GetRequestIssuesByProjectStore() storage.RequestIssuesByProjectStore { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "GetRequestIssuesByProjectStore") + ret0, _ := ret[0].(storage.RequestIssuesByProjectStore) + return ret0 +} + +// GetRequestIssuesByProjectStore indicates an expected call of GetRequestIssuesByProjectStore. +func (mr *MockStorageMockRecorder) GetRequestIssuesByProjectStore() *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetRequestIssuesByProjectStore", reflect.TypeOf((*MockStorage)(nil).GetRequestIssuesByProjectStore)) +} + // GetRequestLogStore mocks base method. func (m *MockStorage) GetRequestLogStore() storage.RequestLogStore { m.ctrl.T.Helper() diff --git a/stovepipe/extension/storage/mysql/BUILD.bazel b/stovepipe/extension/storage/mysql/BUILD.bazel index 5d04a4b74..d8aa56ce7 100644 --- a/stovepipe/extension/storage/mysql/BUILD.bazel +++ b/stovepipe/extension/storage/mysql/BUILD.bazel @@ -16,6 +16,7 @@ go_library( "queue_summary_store.go", "request_acceptance_store.go", "request_issue_summary_store.go", + "request_issues_by_project_store.go", "request_log_store.go", "request_projects_by_issue_store.go", "request_store.go", @@ -54,6 +55,7 @@ go_test( "queue_summary_store_test.go", "request_acceptance_store_test.go", "request_issue_summary_store_test.go", + "request_issues_by_project_store_test.go", "request_log_store_test.go", "request_projects_by_issue_store_test.go", "request_store_test.go", diff --git a/stovepipe/extension/storage/mysql/request_issues_by_project_store.go b/stovepipe/extension/storage/mysql/request_issues_by_project_store.go new file mode 100644 index 000000000..ebc71d02b --- /dev/null +++ b/stovepipe/extension/storage/mysql/request_issues_by_project_store.go @@ -0,0 +1,119 @@ +// Copyright (c) 2026 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package mysql + +import ( + "context" + "database/sql" + "fmt" + + "github.com/uber-go/tally" + "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/storage" +) + +const listRequestIssuesByProjectQuery = ` + SELECT task_id, unit_id, project + FROM request_issues_by_project WHERE queue = ? AND request_id = ? AND project = ? + ORDER BY task_id ASC, unit_id ASC LIMIT ?` + +const listRequestIssuesByProjectAfterCursorQuery = ` + SELECT task_id, unit_id, project + FROM request_issues_by_project WHERE queue = ? AND request_id = ? AND project = ? + AND (task_id > ? OR (task_id = ? AND unit_id > ?)) + ORDER BY task_id ASC, unit_id ASC LIMIT ?` + +type requestIssuesByProjectStore struct { + db *sql.DB + scope tally.Scope + queue string +} + +// NewRequestIssuesByProjectStore binds the per-request project lookup to a queue. +func NewRequestIssuesByProjectStore(db *sql.DB, scope tally.Scope, queue string) storage.RequestIssuesByProjectStore { + return &requestIssuesByProjectStore{db: db, scope: scope, queue: queue} +} + +func (s *requestIssuesByProjectStore) Create(ctx context.Context, requestID string, entry entity.ProjectIssueAttribution) (retErr error) { + op := metrics.Begin(s.scope, "create", metrics.StorageLatencyBuckets) + defer func() { op.Complete(retErr) }() + + if err := validateProjectIssueRequest(s.queue, requestID); err != nil { + return err + } + if err := validateProjectIssueAttribution(entry); err != nil { + return err + } + _, err := s.db.ExecContext(ctx, + `INSERT INTO request_issues_by_project (queue, request_id, project, task_id, unit_id) + VALUES (?, ?, ?, ?, ?)`, + s.queue, requestID, entry.Project, entry.TaskID, entry.UnitID) + if isDuplicateEntry(err) { + return fmt.Errorf("request project issue already retained: %w", storage.ErrAlreadyExists) + } + if err != nil { + return fmt.Errorf("failed to create request project issue: %w", err) + } + return nil +} + +func (s *requestIssuesByProjectStore) List(ctx context.Context, requestID, project string, after entity.ProjectIssueAttribution, limit int) (ret []entity.ProjectIssueAttribution, retErr error) { + op := metrics.Begin(s.scope, "list", metrics.StorageLatencyBuckets) + defer func() { op.Complete(retErr) }() + + if err := validateProjectIssueRequest(s.queue, requestID); err != nil { + return nil, err + } + if project == "" { + return nil, fmt.Errorf("request project issue project must not be empty") + } + if len(project) > 1020 { + return nil, fmt.Errorf("request project issue project exceeds 1020-byte limit") + } + if limit <= 0 { + return nil, fmt.Errorf("request project issue limit must be positive") + } + query := listRequestIssuesByProjectQuery + args := []any{s.queue, requestID, project} + if after != (entity.ProjectIssueAttribution{}) { + if err := validateProjectIssueAttribution(after); err != nil { + return nil, err + } + if after.Project != project { + return nil, fmt.Errorf("request project issue cursor must match project") + } + query = listRequestIssuesByProjectAfterCursorQuery + args = append(args, after.TaskID, after.TaskID, after.UnitID) + } + args = append(args, limit) + rows, err := s.db.QueryContext(ctx, query, args...) + if err != nil { + return nil, fmt.Errorf("failed to list request project issues: %w", err) + } + defer rows.Close() + + for rows.Next() { + var entry entity.ProjectIssueAttribution + if err := rows.Scan(&entry.TaskID, &entry.UnitID, &entry.Project); err != nil { + return nil, fmt.Errorf("failed to scan request project issue: %w", err) + } + ret = append(ret, entry) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("failed to iterate request project issues: %w", err) + } + return ret, nil +} diff --git a/stovepipe/extension/storage/mysql/request_issues_by_project_store_test.go b/stovepipe/extension/storage/mysql/request_issues_by_project_store_test.go new file mode 100644 index 000000000..479c660c5 --- /dev/null +++ b/stovepipe/extension/storage/mysql/request_issues_by_project_store_test.go @@ -0,0 +1,170 @@ +// Copyright (c) 2026 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package mysql + +import ( + "context" + "errors" + "regexp" + "strings" + "testing" + + "github.com/DATA-DOG/go-sqlmock" + "github.com/go-sql-driver/mysql" + "github.com/stretchr/testify/require" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/storage" +) + +func TestRequestIssuesByProjectStoreCreate(t *testing.T) { + writeErr := errors.New("connection reset") + for _, tt := range []struct { + name string + dbErr error + wantErr error + }{ + {name: "created"}, + {name: "duplicate", dbErr: &mysql.MySQLError{Number: mysqlErrDuplicateEntry}, wantErr: storage.ErrAlreadyExists}, + {name: "write failure", dbErr: writeErr, wantErr: writeErr}, + } { + t.Run(tt.name, func(t *testing.T) { + db, mock, err := sqlmock.New() + require.NoError(t, err) + defer db.Close() + entry := entity.ProjectIssueAttribution{TaskID: "test", UnitID: "//example:test", Project: "service"} + query := mock.ExpectExec("INSERT INTO request_issues_by_project"). + WithArgs("queue", "1", entry.Project, entry.TaskID, entry.UnitID) + if tt.dbErr == nil { + query.WillReturnResult(sqlmock.NewResult(0, 1)) + } else { + query.WillReturnError(tt.dbErr) + } + store := NewRequestIssuesByProjectStore(db, testMetrics(), "queue") + err = store.Create(context.Background(), "1", entry) + if tt.wantErr == nil { + require.NoError(t, err) + } else { + require.ErrorIs(t, err, tt.wantErr) + } + require.NoError(t, mock.ExpectationsWereMet()) + }) + } +} + +func TestRequestIssuesByProjectStoreList(t *testing.T) { + columns := []string{"task_id", "unit_id", "project"} + entry := entity.ProjectIssueAttribution{TaskID: "test", UnitID: "//example:test", Project: "service"} + readErr := errors.New("connection reset") + for _, after := range []entity.ProjectIssueAttribution{ + {}, + {TaskID: "test", UnitID: "//example:before", Project: "service"}, + } { + for _, tt := range []struct { + name string + rows *sqlmock.Rows + queryErr error + want []entity.ProjectIssueAttribution + wantErr bool + }{ + {name: "found", rows: sqlmock.NewRows(columns).AddRow(entry.TaskID, entry.UnitID, entry.Project), want: []entity.ProjectIssueAttribution{entry}}, + {name: "empty", rows: sqlmock.NewRows(columns)}, + {name: "query failure", queryErr: readErr, wantErr: true}, + {name: "scan failure", rows: sqlmock.NewRows(columns).AddRow(nil, entry.UnitID, entry.Project), wantErr: true}, + {name: "iteration failure", rows: sqlmock.NewRows(columns).AddRow(entry.TaskID, entry.UnitID, entry.Project).RowError(0, readErr), wantErr: true}, + } { + t.Run(tt.name+"/"+after.UnitID, func(t *testing.T) { + db, mock, err := sqlmock.New() + require.NoError(t, err) + defer db.Close() + var query *sqlmock.ExpectedQuery + if after == (entity.ProjectIssueAttribution{}) { + query = mock.ExpectQuery(regexp.QuoteMeta(listRequestIssuesByProjectQuery)). + WithArgs("queue", "1", "service", 2) + } else { + query = mock.ExpectQuery(regexp.QuoteMeta(listRequestIssuesByProjectAfterCursorQuery)). + WithArgs("queue", "1", "service", after.TaskID, after.TaskID, after.UnitID, 2) + } + if tt.queryErr == nil { + query.WillReturnRows(tt.rows).RowsWillBeClosed() + } else { + query.WillReturnError(tt.queryErr) + } + store := NewRequestIssuesByProjectStore(db, testMetrics(), "queue") + got, err := store.List(context.Background(), "1", "service", after, 2) + if tt.wantErr { + require.Error(t, err) + } else { + require.NoError(t, err) + require.Equal(t, tt.want, got) + } + if tt.queryErr != nil { + require.ErrorIs(t, err, tt.queryErr) + } + require.NoError(t, mock.ExpectationsWereMet()) + }) + } + } +} + +func TestRequestIssuesByProjectStoreRejectsInvalidInput(t *testing.T) { + db, mock, err := sqlmock.New() + require.NoError(t, err) + defer db.Close() + ctx := context.Background() + entry := entity.ProjectIssueAttribution{TaskID: "test", UnitID: "unit", Project: "service"} + for _, tt := range []struct { + name string + queue string + requestID string + }{ + {name: "missing queue", requestID: "1"}, + {name: "long queue", queue: strings.Repeat("q", 256), requestID: "1"}, + {name: "missing request", queue: "queue"}, + {name: "noncanonical request", queue: "queue", requestID: "01"}, + {name: "invalid request", queue: "queue", requestID: "request"}, + } { + t.Run(tt.name, func(t *testing.T) { + store := NewRequestIssuesByProjectStore(db, testMetrics(), tt.queue) + require.Error(t, store.Create(ctx, tt.requestID, entry)) + _, err := store.List(ctx, tt.requestID, "service", entity.ProjectIssueAttribution{}, 1) + require.Error(t, err) + }) + } + store := NewRequestIssuesByProjectStore(db, testMetrics(), "queue") + for _, invalid := range []entity.ProjectIssueAttribution{ + {UnitID: "unit", Project: "service"}, + {TaskID: "test", Project: "service"}, + {TaskID: "test", UnitID: "unit"}, + {Project: "service"}, + {TaskID: strings.Repeat("t", 256), UnitID: "unit", Project: "service"}, + {TaskID: "test", UnitID: strings.Repeat("u", 1025), Project: "service"}, + {TaskID: "test", UnitID: "unit", Project: strings.Repeat("p", 1021)}, + } { + require.Error(t, store.Create(ctx, "1", invalid)) + _, err := store.List(ctx, "1", "service", invalid, 1) + require.Error(t, err) + } + _, err = store.List(ctx, "1", "service", entity.ProjectIssueAttribution{TaskID: "test", UnitID: "unit", Project: "other"}, 1) + require.Error(t, err) + for _, project := range []string{"", strings.Repeat("p", 1021)} { + _, err := store.List(ctx, "1", project, entity.ProjectIssueAttribution{}, 1) + require.Error(t, err) + } + for _, limit := range []int{0, -1} { + _, err := store.List(ctx, "1", "service", entity.ProjectIssueAttribution{}, limit) + require.Error(t, err) + } + require.NoError(t, mock.ExpectationsWereMet()) +} diff --git a/stovepipe/extension/storage/mysql/request_projects_by_issue_store.go b/stovepipe/extension/storage/mysql/request_projects_by_issue_store.go index 5a095a9fb..80c8e0ebe 100644 --- a/stovepipe/extension/storage/mysql/request_projects_by_issue_store.go +++ b/stovepipe/extension/storage/mysql/request_projects_by_issue_store.go @@ -53,7 +53,7 @@ func (s *requestProjectsByIssueStore) Create(ctx context.Context, requestID stri op := metrics.Begin(s.scope, "create", metrics.StorageLatencyBuckets) defer func() { op.Complete(retErr) }() - if err := s.validatePrefix(requestID); err != nil { + if err := validateProjectIssueRequest(s.queue, requestID); err != nil { return err } if err := validateProjectIssueAttribution(entry); err != nil { @@ -76,7 +76,7 @@ func (s *requestProjectsByIssueStore) List(ctx context.Context, requestID string op := metrics.Begin(s.scope, "list", metrics.StorageLatencyBuckets) defer func() { op.Complete(retErr) }() - if err := s.validatePrefix(requestID); err != nil { + if err := validateProjectIssueRequest(s.queue, requestID); err != nil { return nil, err } if limit <= 0 { @@ -111,14 +111,14 @@ func (s *requestProjectsByIssueStore) List(ctx context.Context, requestID string return ret, nil } -func (s *requestProjectsByIssueStore) validatePrefix(requestID string) error { +func validateProjectIssueRequest(queue, requestID string) error { if err := id.Validate(requestID); err != nil { return fmt.Errorf("invalid project issue attribution request ID: %w", err) } - if s.queue == "" { + if queue == "" { return fmt.Errorf("project issue attribution queue must not be empty") } - if len(s.queue) > 255 { + if len(queue) > 255 { return fmt.Errorf("project issue attribution queue exceeds 255-byte limit") } return nil diff --git a/stovepipe/extension/storage/mysql/schema/request_issues_by_project.sql b/stovepipe/extension/storage/mysql/schema/request_issues_by_project.sql new file mode 100644 index 000000000..533644712 --- /dev/null +++ b/stovepipe/extension/storage/mysql/schema/request_issues_by_project.sql @@ -0,0 +1,8 @@ +CREATE TABLE IF NOT EXISTS request_issues_by_project ( + queue VARBINARY(255) NOT NULL, + request_id VARBINARY(255) NOT NULL, + project VARBINARY(1020) NOT NULL, + task_id VARBINARY(255) NOT NULL, + unit_id VARBINARY(1024) NOT NULL, + PRIMARY KEY (queue, request_id, project, task_id, unit_id) +) ENGINE=InnoDB; diff --git a/stovepipe/extension/storage/mysql/storage.go b/stovepipe/extension/storage/mysql/storage.go index c54f0cd60..383760ed9 100644 --- a/stovepipe/extension/storage/mysql/storage.go +++ b/stovepipe/extension/storage/mysql/storage.go @@ -65,6 +65,7 @@ func (s *Storage) For(queueName string) (storage.Storage, error) { validationFactStore: NewValidationFactStore(s.db, queueScope.SubScope("validation_fact_store"), queueName), projectFactByDegreeStore: NewProjectFactByDegreeStore(s.db, queueScope.SubScope("project_fact_by_degree_store"), queueName), requestProjectsByIssueStore: NewRequestProjectsByIssueStore(s.db, queueScope.SubScope("request_projects_by_issue_store"), queueName), + requestIssuesByProjectStore: NewRequestIssuesByProjectStore(s.db, queueScope.SubScope("request_issues_by_project_store"), queueName), }, nil } @@ -94,6 +95,7 @@ type mysqlStorage struct { validationFactStore storage.ValidationFactStore projectFactByDegreeStore storage.ProjectFactByDegreeStore requestProjectsByIssueStore storage.RequestProjectsByIssueStore + requestIssuesByProjectStore storage.RequestIssuesByProjectStore } // Verify mysqlStorage implements the queue-scoped aggregate at compile time. @@ -147,6 +149,10 @@ func (f *mysqlStorage) GetRequestProjectsByIssueStore() storage.RequestProjectsB return f.requestProjectsByIssueStore } +func (f *mysqlStorage) GetRequestIssuesByProjectStore() storage.RequestIssuesByProjectStore { + return f.requestIssuesByProjectStore +} + func (f *mysqlStorage) GetQueueSummaryStore() storage.QueueSummaryStore { return f.queueSummaryStore } func (f *mysqlStorage) GetQueueLogStore() storage.QueueLogStore { diff --git a/stovepipe/extension/storage/mysql/storage_test.go b/stovepipe/extension/storage/mysql/storage_test.go index a2d4dbd72..2321c5988 100644 --- a/stovepipe/extension/storage/mysql/storage_test.go +++ b/stovepipe/extension/storage/mysql/storage_test.go @@ -206,6 +206,14 @@ func TestStorageForQueueMetrics(t *testing.T) { return err }, }, + { + name: "request_issues_by_project_store", + op: "list", + read: func(bound storage.Storage, _ string) error { + _, err := bound.GetRequestIssuesByProjectStore().List(context.Background(), "1", "service", entity.ProjectIssueAttribution{}, 1) + return err + }, + }, } for _, tt := range tests { diff --git a/stovepipe/extension/storage/request_issues_by_project_store.go b/stovepipe/extension/storage/request_issues_by_project_store.go new file mode 100644 index 000000000..430404c51 --- /dev/null +++ b/stovepipe/extension/storage/request_issues_by_project_store.go @@ -0,0 +1,34 @@ +// Copyright (c) 2026 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package storage + +//go:generate mockgen -source=request_issues_by_project_store.go -destination=mock/request_issues_by_project_store_mock.go -package=mock + +import ( + "context" + + "github.com/uber/submitqueue/stovepipe/entity" +) + +// RequestIssuesByProjectStore holds immutable project-to-issue lookups within a request. +type RequestIssuesByProjectStore interface { + // Create puts one lookup entry; duplicates return ErrAlreadyExists. + Create(ctx context.Context, requestID string, entry entity.ProjectIssueAttribution) error + + // List scans the request/project prefix in ascending bytewise task/unit order. + // after is exclusive; its zero value starts the range. A nonzero cursor must match project. + // limit must be positive. An empty page ends the range, not evidence of completed attribution. + List(ctx context.Context, requestID, project string, after entity.ProjectIssueAttribution, limit int) ([]entity.ProjectIssueAttribution, error) +} diff --git a/stovepipe/extension/storage/storage.go b/stovepipe/extension/storage/storage.go index a438b5133..dcfd87ea6 100644 --- a/stovepipe/extension/storage/storage.go +++ b/stovepipe/extension/storage/storage.go @@ -121,4 +121,6 @@ type Storage interface { GetProjectFactByDegreeStore() ProjectFactByDegreeStore GetRequestProjectsByIssueStore() RequestProjectsByIssueStore + + GetRequestIssuesByProjectStore() RequestIssuesByProjectStore } diff --git a/test/integration/stovepipe/extension/storage/mysql/BUILD.bazel b/test/integration/stovepipe/extension/storage/mysql/BUILD.bazel index f512cab85..2f7c2d5c5 100644 --- a/test/integration/stovepipe/extension/storage/mysql/BUILD.bazel +++ b/test/integration/stovepipe/extension/storage/mysql/BUILD.bazel @@ -19,6 +19,7 @@ go_test( "request_acceptance_test.go", "request_issue_summary_materializer_test.go", "request_issue_summary_test.go", + "request_issues_by_project_test.go", "request_projects_by_issue_test.go", "storage_test.go", ], diff --git a/test/integration/stovepipe/extension/storage/mysql/request_issues_by_project_test.go b/test/integration/stovepipe/extension/storage/mysql/request_issues_by_project_test.go new file mode 100644 index 000000000..94ee4576e --- /dev/null +++ b/test/integration/stovepipe/extension/storage/mysql/request_issues_by_project_test.go @@ -0,0 +1,103 @@ +// Copyright (c) 2026 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package mysql + +import ( + "context" + "strings" + "testing" + + "github.com/stretchr/testify/require" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/storage" + mysqlstorage "github.com/uber/submitqueue/stovepipe/extension/storage/mysql" +) + +func testRequestIssuesByProject(t *testing.T, ctx context.Context, backend *mysqlstorage.Storage) { + bound, err := backend.For("projects/main") + require.NoError(t, err) + store := bound.GetRequestIssuesByProjectStore() + entries := []entity.ProjectIssueAttribution{ + {TaskID: "test", UnitID: "b", Project: "service"}, + {TaskID: "Test", UnitID: "x", Project: "service"}, + {TaskID: "test", UnitID: "a", Project: "service"}, + {TaskID: "test", UnitID: "a\x00", Project: "service"}, + {TaskID: "z", UnitID: "end", Project: "service"}, + {TaskID: "test", UnitID: "a", Project: "Service"}, + {TaskID: "test", UnitID: "a", Project: "other"}, + } + for _, entry := range entries { + require.NoError(t, bound.GetRequestProjectsByIssueStore().Create(ctx, "1", entry)) + } + page, err := store.List(ctx, "1", "service", entity.ProjectIssueAttribution{}, 2) + require.NoError(t, err) + require.Empty(t, page) + for _, entry := range entries { + require.NoError(t, store.Create(ctx, "1", entry)) + require.ErrorIs(t, store.Create(ctx, "1", entry), storage.ErrAlreadyExists) + } + + other, err := backend.For("other") + require.NoError(t, err) + unrelated := entity.ProjectIssueAttribution{TaskID: "other", UnitID: "unit", Project: "service"} + require.NoError(t, other.GetRequestIssuesByProjectStore().Create(ctx, "1", unrelated)) + require.NoError(t, store.Create(ctx, "2", unrelated)) + + want := []entity.ProjectIssueAttribution{entries[1], entries[2], entries[3], entries[0], entries[4]} + for _, pageSize := range []int{1, 2, 3, 20} { + var got []entity.ProjectIssueAttribution + var after entity.ProjectIssueAttribution + for { + page, err := store.List(ctx, "1", "service", after, pageSize) + require.NoError(t, err) + require.LessOrEqual(t, len(page), pageSize) + if len(page) == 0 { + break + } + got = append(got, page...) + require.LessOrEqual(t, len(got), len(want)) + after = page[len(page)-1] + } + require.Equal(t, want, got) + } + page, err = store.List(ctx, "1", "Service", entity.ProjectIssueAttribution{}, 2) + require.NoError(t, err) + require.Equal(t, []entity.ProjectIssueAttribution{entries[5]}, page) + page, err = store.List(ctx, "1", "service", entity.ProjectIssueAttribution{TaskID: "test", UnitID: "az", Project: "service"}, 2) + require.NoError(t, err) + require.Equal(t, want[3:], page) + page, err = store.List(ctx, "1", "missing", entity.ProjectIssueAttribution{}, 2) + require.NoError(t, err) + require.Empty(t, page) + page, err = store.List(ctx, "3", "service", entity.ProjectIssueAttribution{}, 2) + require.NoError(t, err) + require.Empty(t, page) +} + +func testRequestIssuesByProjectKeyLimits(t *testing.T, ctx context.Context, backend *mysqlstorage.Storage) { + bound, err := backend.For(strings.Repeat("q", 255)) + require.NoError(t, err) + store := bound.GetRequestIssuesByProjectStore() + entry := entity.ProjectIssueAttribution{ + TaskID: strings.Repeat("t", 255), UnitID: strings.Repeat("u", 1024), Project: strings.Repeat("界", 340), + } + require.NoError(t, store.Create(ctx, "9223372036854775807", entry)) + page, err := store.List(ctx, "9223372036854775807", entry.Project, entity.ProjectIssueAttribution{}, 1) + require.NoError(t, err) + require.Equal(t, []entity.ProjectIssueAttribution{entry}, page) + page, err = store.List(ctx, "9223372036854775807", entry.Project, entry, 1) + require.NoError(t, err) + require.Empty(t, page) +} diff --git a/test/integration/stovepipe/extension/storage/mysql/storage_test.go b/test/integration/stovepipe/extension/storage/mysql/storage_test.go index b2af12820..ce1d9506f 100644 --- a/test/integration/stovepipe/extension/storage/mysql/storage_test.go +++ b/test/integration/stovepipe/extension/storage/mysql/storage_test.go @@ -105,6 +105,16 @@ func TestMySQLStorage(t *testing.T) { testRequestProjectsByIssueKeyLimits(t, ctx, backend) }) + t.Run("RequestIssuesByProject", func(t *testing.T) { + resetStorage(t, db) + testRequestIssuesByProject(t, ctx, backend) + }) + + t.Run("RequestIssuesByProjectKeyLimits", func(t *testing.T) { + resetStorage(t, db) + testRequestIssuesByProjectKeyLimits(t, ctx, backend) + }) + t.Run("BuildIssueList", func(t *testing.T) { resetStorage(t, db) testBuildIssueList(t, ctx, factory) @@ -238,6 +248,7 @@ func resetStorage(t *testing.T, db *sql.DB) { "TRUNCATE TABLE validation_fact", "TRUNCATE TABLE project_fact_by_degree", "TRUNCATE TABLE request_projects_by_issue", + "TRUNCATE TABLE request_issues_by_project", } { _, err := db.ExecContext(context.Background(), statement) require.NoError(t, err)