From 019587c917bf152ea3b5e590b027d262c7381003 Mon Sep 17 00:00:00 2001 From: "prath.shenoy" Date: Mon, 9 Nov 2026 01:56:05 +0000 Subject: [PATCH 1/3] feat(stovepipe): Store project issue attribution MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary **What**: - Add immutable storage for per-validation issue-to-project attribution. - Define a request-history event identifying completed attribution recording. **Why**: - Give project–issue views a durable source they can replay. Existing issue history does not retain which projects each issue affects, so these records provide the associations needed for both lookup directions. - Distinguish completed recording from partial writes. A completed request may have no associations, so an empty set of stored rows is not enough to establish completion. ## Test Plan - [x] Add unit tests. - [ ] Run the MySQL integration tests when Docker is available. ## Revert Plan - Revert this PR. No runtime producer or materializer is wired to the new source yet. ## Issue Links - [CODEM-607](https://linear.app/uber/issue/CODEM-607) --- stovepipe/entity/BUILD.bazel | 2 + stovepipe/entity/project_issue_attribution.go | 41 +++++ .../entity/project_issue_attribution_test.go | 42 +++++ stovepipe/entity/request_log.go | 4 +- stovepipe/entity/request_log_test.go | 10 ++ stovepipe/extension/storage/BUILD.bazel | 1 + stovepipe/extension/storage/README.md | 8 + stovepipe/extension/storage/mock/BUILD.bazel | 1 + .../request_projects_by_issue_store_mock.go | 71 ++++++++ .../extension/storage/mock/storage_mock.go | 14 ++ stovepipe/extension/storage/mysql/BUILD.bazel | 2 + .../mysql/request_projects_by_issue_store.go | 145 ++++++++++++++++ .../request_projects_by_issue_store_test.go | 164 ++++++++++++++++++ .../schema/request_projects_by_issue.sql | 8 + stovepipe/extension/storage/mysql/storage.go | 78 +++++---- .../extension/storage/mysql/storage_test.go | 8 + .../request_projects_by_issue_store.go | 33 ++++ stovepipe/extension/storage/storage.go | 2 + .../extension/storage/mysql/BUILD.bazel | 1 + .../mysql/request_projects_by_issue_test.go | 126 ++++++++++++++ .../extension/storage/mysql/storage_test.go | 11 ++ 21 files changed, 735 insertions(+), 37 deletions(-) create mode 100644 stovepipe/entity/project_issue_attribution.go create mode 100644 stovepipe/entity/project_issue_attribution_test.go create mode 100644 stovepipe/extension/storage/mock/request_projects_by_issue_store_mock.go create mode 100644 stovepipe/extension/storage/mysql/request_projects_by_issue_store.go create mode 100644 stovepipe/extension/storage/mysql/request_projects_by_issue_store_test.go create mode 100644 stovepipe/extension/storage/mysql/schema/request_projects_by_issue.sql create mode 100644 stovepipe/extension/storage/request_projects_by_issue_store.go create mode 100644 test/integration/stovepipe/extension/storage/mysql/request_projects_by_issue_test.go diff --git a/stovepipe/entity/BUILD.bazel b/stovepipe/entity/BUILD.bazel index 7a46174b0..b23baa740 100644 --- a/stovepipe/entity/BUILD.bazel +++ b/stovepipe/entity/BUILD.bazel @@ -15,6 +15,7 @@ go_library( "list_open_issues.go", "open_issue.go", "project_fact_by_degree.go", + "project_issue_attribution.go", "project_status.go", "queue.go", "queue_config.go", @@ -45,6 +46,7 @@ go_test( "issue_summary_test.go", "issue_tracking_state_test.go", "project_fact_by_degree_test.go", + "project_issue_attribution_test.go", "queue_state_test.go", "request_id_test.go", "request_issue_summary_test.go", diff --git a/stovepipe/entity/project_issue_attribution.go b/stovepipe/entity/project_issue_attribution.go new file mode 100644 index 000000000..ee0c10502 --- /dev/null +++ b/stovepipe/entity/project_issue_attribution.go @@ -0,0 +1,41 @@ +// 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 entity + +import "fmt" + +// ProjectIssueAttribution is one immutable issue/project pair within an attribution result. +type ProjectIssueAttribution struct { + // TaskID identifies the validation task and must be nonempty. + TaskID string + // UnitID identifies the failed unit within the task and must be nonempty. + UnitID string + // Project is an affected project name and must be nonempty. + Project string +} + +// Validate verifies the identity of a unit-level attribution. +func (a ProjectIssueAttribution) Validate() error { + if a.TaskID == "" { + return fmt.Errorf("project issue attribution task ID must not be empty") + } + if a.UnitID == "" { + return fmt.Errorf("project issue attribution unit ID must not be empty") + } + if a.Project == "" { + return fmt.Errorf("project issue attribution project must not be empty") + } + return nil +} diff --git a/stovepipe/entity/project_issue_attribution_test.go b/stovepipe/entity/project_issue_attribution_test.go new file mode 100644 index 000000000..2cf89d8fd --- /dev/null +++ b/stovepipe/entity/project_issue_attribution_test.go @@ -0,0 +1,42 @@ +// 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 entity + +import ( + "testing" + + "github.com/stretchr/testify/require" +) + +func TestProjectIssueAttributionValidate(t *testing.T) { + for _, tt := range []struct { + name string + entry ProjectIssueAttribution + wantErr bool + }{ + {name: "affected project", entry: ProjectIssueAttribution{TaskID: "test", UnitID: "//example:test", Project: "service"}}, + {name: "missing project", entry: ProjectIssueAttribution{TaskID: "test", UnitID: "//example:test"}, wantErr: true}, + {name: "missing task", entry: ProjectIssueAttribution{UnitID: "//example:test", Project: "service"}, wantErr: true}, + {name: "task-wide failure", entry: ProjectIssueAttribution{TaskID: "test", Project: "service"}, wantErr: true}, + } { + t.Run(tt.name, func(t *testing.T) { + if tt.wantErr { + require.Error(t, tt.entry.Validate()) + } else { + require.NoError(t, tt.entry.Validate()) + } + }) + } +} diff --git a/stovepipe/entity/request_log.go b/stovepipe/entity/request_log.go index c81426b1a..ceaf1263d 100644 --- a/stovepipe/entity/request_log.go +++ b/stovepipe/entity/request_log.go @@ -30,6 +30,8 @@ const ( RequestEventValidationFactRecorded RequestEvent = "validation_fact_recorded" // RequestEventProjectFactsRecorded records that all project validation facts for a request were established. RequestEventProjectFactsRecorded RequestEvent = "project_facts_recorded" + // RequestEventProjectIssuesRecorded records completed project-issue attribution for a request. + RequestEventProjectIssuesRecorded RequestEvent = "project_issues_recorded" // RequestEventRecordAbandoned records that record-stage work stopped after exhausting retries. RequestEventRecordAbandoned RequestEvent = "record_abandoned" // RequestEventBuildAnalysisAbandoned records incomplete build analysis, not a validation outcome. @@ -153,7 +155,7 @@ func (e RequestLog) validateEvent() error { return fmt.Errorf("event log must not contain request-state context") } switch e.Event { - case RequestEventBuildTriggered, RequestEventBuildFinished, RequestEventValidationFactRecorded, RequestEventProjectFactsRecorded, + case RequestEventBuildTriggered, RequestEventBuildFinished, RequestEventValidationFactRecorded, RequestEventProjectFactsRecorded, RequestEventProjectIssuesRecorded, RequestEventRecordAbandoned, RequestEventBuildAnalysisAbandoned, RequestEventBuildAnalysisCompleted, RequestEventIssueTrackingStarted, RequestEventIssuesReconciled, RequestEventIssueTrackingAbandoned, RequestEventFixSubmitted, RequestEventFixSubmissionSkipped, RequestEventFixSubmissionAbandoned: diff --git a/stovepipe/entity/request_log_test.go b/stovepipe/entity/request_log_test.go index 7dcb41263..d31334b92 100644 --- a/stovepipe/entity/request_log_test.go +++ b/stovepipe/entity/request_log_test.go @@ -116,6 +116,16 @@ func TestRequestLogValidate(t *testing.T) { return entry }, }, + { + name: "project issues recorded", + mutate: func(entry RequestLog) RequestLog { + entry.State = RequestStateUnknown + entry.Event = RequestEventProjectIssuesRecorded + entry.RequestVersion = 0 + entry.Metadata = map[string]string{"build_id": "42"} + return entry + }, + }, { name: "build event", mutate: func(entry RequestLog) RequestLog { diff --git a/stovepipe/extension/storage/BUILD.bazel b/stovepipe/extension/storage/BUILD.bazel index 00542e3f3..8e18e2701 100644 --- a/stovepipe/extension/storage/BUILD.bazel +++ b/stovepipe/extension/storage/BUILD.bazel @@ -17,6 +17,7 @@ go_library( "request_acceptance_store.go", "request_issue_summary_store.go", "request_log_store.go", + "request_projects_by_issue_store.go", "request_store.go", "request_summary_store.go", "request_uri_store.go", diff --git a/stovepipe/extension/storage/README.md b/stovepipe/extension/storage/README.md index c04f76d14..a8d81680c 100644 --- a/stovepipe/extension/storage/README.md +++ b/stovepipe/extension/storage/README.md @@ -26,6 +26,14 @@ Same design space as [`submitqueue/extension/storage`](../../../submitqueue/exte `ProjectFactByDegreeStore` is the ordered lookup mapping for named project facts. Its immutable `(queue, URI, degree DESC, project ASC)` primary key stores the original degree and orders by descending degree and ascending bytewise project ID, so listing is a bounded key-range read. It holds no duplicate validation payload; readers hydrate facts by their original keys and verify the mapping. The record controller ensures mappings from persisted winning facts before recording project completion, including retries after partial writes. Create the table before deploying writers. The table starts empty: only results processed by index-aware writers receive mappings, with no historical population or read-time fallback. +## Request projects by issue + +`RequestProjectsByIssueStore` retains source rows keyed by `(queue, request_id, task_id, unit_id, project)`. Each row records one affected project for a unit-level issue; task, unit, and project must be nonempty. Rows are immutable and duplicate creates return `ErrAlreadyExists`. + +Attribution is assumed stable for a request. Retries recreate the same pairs and treat duplicate creates as success. All rows must be written before retaining the request's `project_issues_recorded` occurrence, whose metadata identifies the build. The occurrence establishes completion even when there are no associations. Without it, retained rows are incomplete. + +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. + ## 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 15380ccf8..18bfa988c 100644 --- a/stovepipe/extension/storage/mock/BUILD.bazel +++ b/stovepipe/extension/storage/mock/BUILD.bazel @@ -17,6 +17,7 @@ go_library( "request_acceptance_store_mock.go", "request_issue_summary_store_mock.go", "request_log_store_mock.go", + "request_projects_by_issue_store_mock.go", "request_store_mock.go", "request_summary_store_mock.go", "request_uri_store_mock.go", diff --git a/stovepipe/extension/storage/mock/request_projects_by_issue_store_mock.go b/stovepipe/extension/storage/mock/request_projects_by_issue_store_mock.go new file mode 100644 index 000000000..07f23e360 --- /dev/null +++ b/stovepipe/extension/storage/mock/request_projects_by_issue_store_mock.go @@ -0,0 +1,71 @@ +// Code generated by MockGen. DO NOT EDIT. +// Source: request_projects_by_issue_store.go +// +// Generated by this command: +// +// mockgen -source=request_projects_by_issue_store.go -destination=mock/request_projects_by_issue_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" +) + +// MockRequestProjectsByIssueStore is a mock of RequestProjectsByIssueStore interface. +type MockRequestProjectsByIssueStore struct { + ctrl *gomock.Controller + recorder *MockRequestProjectsByIssueStoreMockRecorder + isgomock struct{} +} + +// MockRequestProjectsByIssueStoreMockRecorder is the mock recorder for MockRequestProjectsByIssueStore. +type MockRequestProjectsByIssueStoreMockRecorder struct { + mock *MockRequestProjectsByIssueStore +} + +// NewMockRequestProjectsByIssueStore creates a new mock instance. +func NewMockRequestProjectsByIssueStore(ctrl *gomock.Controller) *MockRequestProjectsByIssueStore { + mock := &MockRequestProjectsByIssueStore{ctrl: ctrl} + mock.recorder = &MockRequestProjectsByIssueStoreMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *MockRequestProjectsByIssueStore) EXPECT() *MockRequestProjectsByIssueStoreMockRecorder { + return m.recorder +} + +// Create mocks base method. +func (m *MockRequestProjectsByIssueStore) 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 *MockRequestProjectsByIssueStoreMockRecorder) Create(ctx, requestID, entry any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Create", reflect.TypeOf((*MockRequestProjectsByIssueStore)(nil).Create), ctx, requestID, entry) +} + +// List mocks base method. +func (m *MockRequestProjectsByIssueStore) List(ctx context.Context, requestID string, after entity.ProjectIssueAttribution, limit int) ([]entity.ProjectIssueAttribution, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "List", ctx, requestID, after, limit) + ret0, _ := ret[0].([]entity.ProjectIssueAttribution) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// List indicates an expected call of List. +func (mr *MockRequestProjectsByIssueStoreMockRecorder) List(ctx, requestID, after, limit any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "List", reflect.TypeOf((*MockRequestProjectsByIssueStore)(nil).List), ctx, requestID, after, limit) +} diff --git a/stovepipe/extension/storage/mock/storage_mock.go b/stovepipe/extension/storage/mock/storage_mock.go index 1594c3929..c6b62b686 100644 --- a/stovepipe/extension/storage/mock/storage_mock.go +++ b/stovepipe/extension/storage/mock/storage_mock.go @@ -275,6 +275,20 @@ func (mr *MockStorageMockRecorder) GetRequestLogStore() *gomock.Call { return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetRequestLogStore", reflect.TypeOf((*MockStorage)(nil).GetRequestLogStore)) } +// GetRequestProjectsByIssueStore mocks base method. +func (m *MockStorage) GetRequestProjectsByIssueStore() storage.RequestProjectsByIssueStore { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "GetRequestProjectsByIssueStore") + ret0, _ := ret[0].(storage.RequestProjectsByIssueStore) + return ret0 +} + +// GetRequestProjectsByIssueStore indicates an expected call of GetRequestProjectsByIssueStore. +func (mr *MockStorageMockRecorder) GetRequestProjectsByIssueStore() *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetRequestProjectsByIssueStore", reflect.TypeOf((*MockStorage)(nil).GetRequestProjectsByIssueStore)) +} + // GetRequestStore mocks base method. func (m *MockStorage) GetRequestStore() storage.RequestStore { m.ctrl.T.Helper() diff --git a/stovepipe/extension/storage/mysql/BUILD.bazel b/stovepipe/extension/storage/mysql/BUILD.bazel index 258cc5464..5d04a4b74 100644 --- a/stovepipe/extension/storage/mysql/BUILD.bazel +++ b/stovepipe/extension/storage/mysql/BUILD.bazel @@ -17,6 +17,7 @@ go_library( "request_acceptance_store.go", "request_issue_summary_store.go", "request_log_store.go", + "request_projects_by_issue_store.go", "request_store.go", "request_summary_store.go", "request_uri_store.go", @@ -54,6 +55,7 @@ go_test( "request_acceptance_store_test.go", "request_issue_summary_store_test.go", "request_log_store_test.go", + "request_projects_by_issue_store_test.go", "request_store_test.go", "request_summary_store_test.go", "request_uri_store_test.go", diff --git a/stovepipe/extension/storage/mysql/request_projects_by_issue_store.go b/stovepipe/extension/storage/mysql/request_projects_by_issue_store.go new file mode 100644 index 000000000..5a095a9fb --- /dev/null +++ b/stovepipe/extension/storage/mysql/request_projects_by_issue_store.go @@ -0,0 +1,145 @@ +// 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/base/id" + "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/storage" +) + +const listRequestProjectsByIssueQuery = ` + SELECT task_id, unit_id, project + FROM request_projects_by_issue WHERE queue = ? AND request_id = ? + ORDER BY task_id ASC, unit_id ASC, project ASC LIMIT ?` + +const listRequestProjectsByIssueAfterCursorQuery = ` + SELECT task_id, unit_id, project + FROM request_projects_by_issue WHERE queue = ? AND request_id = ? + AND (task_id > ? OR (task_id = ? AND unit_id > ?) + OR (task_id = ? AND unit_id = ? AND project > ?)) + ORDER BY task_id ASC, unit_id ASC, project ASC LIMIT ?` + +type requestProjectsByIssueStore struct { + db *sql.DB + scope tally.Scope + queue string +} + +// NewRequestProjectsByIssueStore binds immutable attribution results to a queue. +func NewRequestProjectsByIssueStore(db *sql.DB, scope tally.Scope, queue string) storage.RequestProjectsByIssueStore { + return &requestProjectsByIssueStore{db: db, scope: scope, queue: queue} +} + +func (s *requestProjectsByIssueStore) 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 := s.validatePrefix(requestID); err != nil { + return err + } + if err := validateProjectIssueAttribution(entry); err != nil { + return err + } + _, err := s.db.ExecContext(ctx, + `INSERT INTO request_projects_by_issue (queue, request_id, task_id, unit_id, project) + VALUES (?, ?, ?, ?, ?)`, + s.queue, requestID, entry.TaskID, entry.UnitID, entry.Project) + if isDuplicateEntry(err) { + return fmt.Errorf("project issue attribution already retained: %w", storage.ErrAlreadyExists) + } + if err != nil { + return fmt.Errorf("failed to create project issue attribution: %w", err) + } + return nil +} + +func (s *requestProjectsByIssueStore) List(ctx context.Context, requestID 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 := s.validatePrefix(requestID); err != nil { + return nil, err + } + if limit <= 0 { + return nil, fmt.Errorf("project issue attribution limit must be positive") + } + query := listRequestProjectsByIssueQuery + args := []any{s.queue, requestID} + if after != (entity.ProjectIssueAttribution{}) { + if err := validateProjectIssueAttribution(after); err != nil { + return nil, err + } + query = listRequestProjectsByIssueAfterCursorQuery + args = append(args, after.TaskID, after.TaskID, after.UnitID, after.TaskID, after.UnitID, after.Project) + } + args = append(args, limit) + rows, err := s.db.QueryContext(ctx, query, args...) + if err != nil { + return nil, fmt.Errorf("failed to list project issue attributions: %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 project issue attribution: %w", err) + } + ret = append(ret, entry) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("failed to iterate project issue attributions: %w", err) + } + return ret, nil +} + +func (s *requestProjectsByIssueStore) validatePrefix(requestID string) error { + if err := id.Validate(requestID); err != nil { + return fmt.Errorf("invalid project issue attribution request ID: %w", err) + } + if s.queue == "" { + return fmt.Errorf("project issue attribution queue must not be empty") + } + if len(s.queue) > 255 { + return fmt.Errorf("project issue attribution queue exceeds 255-byte limit") + } + return nil +} + +func validateProjectIssueAttribution(entry entity.ProjectIssueAttribution) error { + if err := entry.Validate(); err != nil { + return err + } + for _, field := range []struct { + name string + value string + maxBytes int + }{ + {"task ID", entry.TaskID, 255}, + {"unit ID", entry.UnitID, 1024}, + {"project", entry.Project, 1020}, + } { + if len(field.value) > field.maxBytes { + return fmt.Errorf("project issue attribution %s exceeds %d-byte limit", field.name, field.maxBytes) + } + } + return nil +} diff --git a/stovepipe/extension/storage/mysql/request_projects_by_issue_store_test.go b/stovepipe/extension/storage/mysql/request_projects_by_issue_store_test.go new file mode 100644 index 000000000..979d311e7 --- /dev/null +++ b/stovepipe/extension/storage/mysql/request_projects_by_issue_store_test.go @@ -0,0 +1,164 @@ +// 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 TestRequestProjectsByIssueStoreCreate(t *testing.T) { + writeErr := errors.New("connection reset") + for _, tt := range []struct { + name string + dbErr error + wantErr error + }{ + {name: "affected project"}, + {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_projects_by_issue"). + WithArgs("queue", "1", entry.TaskID, entry.UnitID, entry.Project) + if tt.dbErr == nil { + query.WillReturnResult(sqlmock.NewResult(0, 1)) + } else { + query.WillReturnError(tt.dbErr) + } + store := NewRequestProjectsByIssueStore(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 TestRequestProjectsByIssueStoreList(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+"/"+after.Project, 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(listRequestProjectsByIssueQuery)). + WithArgs("queue", "1", 2) + } else { + query = mock.ExpectQuery(regexp.QuoteMeta(listRequestProjectsByIssueAfterCursorQuery)). + WithArgs("queue", "1", after.TaskID, after.TaskID, after.UnitID, after.TaskID, after.UnitID, after.Project, 2) + } + if tt.queryErr == nil { + query.WillReturnRows(tt.rows).RowsWillBeClosed() + } else { + query.WillReturnError(tt.queryErr) + } + store := NewRequestProjectsByIssueStore(db, testMetrics(), "queue") + got, err := store.List(context.Background(), "1", 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 TestRequestProjectsByIssueStoreRejectsInvalidInput(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 := NewRequestProjectsByIssueStore(db, testMetrics(), tt.queue) + require.Error(t, store.Create(ctx, tt.requestID, entry)) + _, err := store.List(ctx, tt.requestID, entity.ProjectIssueAttribution{}, 1) + require.Error(t, err) + }) + } + store := NewRequestProjectsByIssueStore(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", invalid, 1) + require.Error(t, err) + } + for _, limit := range []int{0, -1} { + _, err := store.List(ctx, "1", entity.ProjectIssueAttribution{}, limit) + require.Error(t, err) + } + require.NoError(t, mock.ExpectationsWereMet()) +} diff --git a/stovepipe/extension/storage/mysql/schema/request_projects_by_issue.sql b/stovepipe/extension/storage/mysql/schema/request_projects_by_issue.sql new file mode 100644 index 000000000..1b0a4dc54 --- /dev/null +++ b/stovepipe/extension/storage/mysql/schema/request_projects_by_issue.sql @@ -0,0 +1,8 @@ +CREATE TABLE IF NOT EXISTS request_projects_by_issue ( + queue VARBINARY(255) NOT NULL, + request_id VARBINARY(255) NOT NULL, + task_id VARBINARY(255) NOT NULL, + unit_id VARBINARY(1024) NOT NULL, + project VARBINARY(1020) NOT NULL, + PRIMARY KEY (queue, request_id, task_id, unit_id, project) +) ENGINE=InnoDB; diff --git a/stovepipe/extension/storage/mysql/storage.go b/stovepipe/extension/storage/mysql/storage.go index 2b5f28b0a..c54f0cd60 100644 --- a/stovepipe/extension/storage/mysql/storage.go +++ b/stovepipe/extension/storage/mysql/storage.go @@ -46,24 +46,25 @@ func (s *Storage) For(queueName string) (storage.Storage, error) { } queueScope := s.scope.Tagged(map[string]string{"queue": queueName}) return &mysqlStorage{ - requestStore: NewRequestStore(s.db, queueScope.SubScope("request_store"), queueName), - requestURIStore: NewRequestURIStore(s.db, queueScope.SubScope("request_uri_store"), queueName), - requestLogStore: NewRequestLogStore(s.db, queueScope.SubScope("request_log_store"), queueName), - requestSummaryStore: NewRequestSummaryStore(s.db, queueScope.SubScope("request_summary_store"), queueName), - requestAcceptanceStore: NewRequestAcceptanceStore(s.db, queueScope.SubScope("request_acceptance_store"), queueName), - queueSummaryStore: NewQueueSummaryStore(s.db, queueScope.SubScope("queue_summary_store"), queueName), - queueLogStore: NewQueueLogStore(s.db, queueScope.SubScope("queue_log_store"), queueName), - queueStore: NewQueueStore(s.db, queueScope.SubScope("queue_store"), queueName), - buildStore: NewBuildStore(s.db, queueScope.SubScope("build_store"), queueName), - buildIssueStore: NewBuildIssueStore(s.db, queueScope.SubScope("build_issue_store"), queueName), - requestIssueSummaryStore: NewRequestIssueSummaryStore(s.db, queueScope.SubScope("request_issue_summary_store"), queueName), - issueStateStore: NewIssueStateStore(s.db, queueScope.SubScope("issue_state_store"), queueName), - issueLogStore: NewIssueLogStore(s.db, queueScope.SubScope("issue_log_store"), queueName), - issueSummaryStore: NewIssueSummaryStore(s.db, queueScope.SubScope("issue_summary_store"), queueName), - openIssueStore: NewOpenIssueStore(s.db, queueScope.SubScope("open_issue_store"), queueName), - issueTrackingStore: NewIssueTrackingStore(s.db, queueScope.SubScope("issue_tracking_store"), queueName), - validationFactStore: NewValidationFactStore(s.db, queueScope.SubScope("validation_fact_store"), queueName), - projectFactByDegreeStore: NewProjectFactByDegreeStore(s.db, queueScope.SubScope("project_fact_by_degree_store"), queueName), + requestStore: NewRequestStore(s.db, queueScope.SubScope("request_store"), queueName), + requestURIStore: NewRequestURIStore(s.db, queueScope.SubScope("request_uri_store"), queueName), + requestLogStore: NewRequestLogStore(s.db, queueScope.SubScope("request_log_store"), queueName), + requestSummaryStore: NewRequestSummaryStore(s.db, queueScope.SubScope("request_summary_store"), queueName), + requestAcceptanceStore: NewRequestAcceptanceStore(s.db, queueScope.SubScope("request_acceptance_store"), queueName), + queueSummaryStore: NewQueueSummaryStore(s.db, queueScope.SubScope("queue_summary_store"), queueName), + queueLogStore: NewQueueLogStore(s.db, queueScope.SubScope("queue_log_store"), queueName), + queueStore: NewQueueStore(s.db, queueScope.SubScope("queue_store"), queueName), + buildStore: NewBuildStore(s.db, queueScope.SubScope("build_store"), queueName), + buildIssueStore: NewBuildIssueStore(s.db, queueScope.SubScope("build_issue_store"), queueName), + requestIssueSummaryStore: NewRequestIssueSummaryStore(s.db, queueScope.SubScope("request_issue_summary_store"), queueName), + issueStateStore: NewIssueStateStore(s.db, queueScope.SubScope("issue_state_store"), queueName), + issueLogStore: NewIssueLogStore(s.db, queueScope.SubScope("issue_log_store"), queueName), + issueSummaryStore: NewIssueSummaryStore(s.db, queueScope.SubScope("issue_summary_store"), queueName), + openIssueStore: NewOpenIssueStore(s.db, queueScope.SubScope("open_issue_store"), queueName), + issueTrackingStore: NewIssueTrackingStore(s.db, queueScope.SubScope("issue_tracking_store"), queueName), + 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), }, nil } @@ -74,24 +75,25 @@ func (s *Storage) Close() error { // mysqlStorage is the queue-scoped store aggregate returned by For. type mysqlStorage struct { - requestStore storage.RequestStore - requestURIStore storage.RequestURIStore - requestLogStore storage.RequestLogStore - requestSummaryStore storage.RequestSummaryStore - requestAcceptanceStore storage.RequestAcceptanceStore - queueStore storage.QueueStore - queueSummaryStore storage.QueueSummaryStore - queueLogStore storage.QueueLogStore - buildStore storage.BuildStore - buildIssueStore storage.BuildIssueStore - requestIssueSummaryStore storage.RequestIssueSummaryStore - issueStateStore storage.IssueStateStore - issueLogStore storage.IssueLogStore - issueSummaryStore storage.IssueSummaryStore - openIssueStore storage.OpenIssueStore - issueTrackingStore storage.IssueTrackingStore - validationFactStore storage.ValidationFactStore - projectFactByDegreeStore storage.ProjectFactByDegreeStore + requestStore storage.RequestStore + requestURIStore storage.RequestURIStore + requestLogStore storage.RequestLogStore + requestSummaryStore storage.RequestSummaryStore + requestAcceptanceStore storage.RequestAcceptanceStore + queueStore storage.QueueStore + queueSummaryStore storage.QueueSummaryStore + queueLogStore storage.QueueLogStore + buildStore storage.BuildStore + buildIssueStore storage.BuildIssueStore + requestIssueSummaryStore storage.RequestIssueSummaryStore + issueStateStore storage.IssueStateStore + issueLogStore storage.IssueLogStore + issueSummaryStore storage.IssueSummaryStore + openIssueStore storage.OpenIssueStore + issueTrackingStore storage.IssueTrackingStore + validationFactStore storage.ValidationFactStore + projectFactByDegreeStore storage.ProjectFactByDegreeStore + requestProjectsByIssueStore storage.RequestProjectsByIssueStore } // Verify mysqlStorage implements the queue-scoped aggregate at compile time. @@ -141,6 +143,10 @@ func (f *mysqlStorage) GetProjectFactByDegreeStore() storage.ProjectFactByDegree return f.projectFactByDegreeStore } +func (f *mysqlStorage) GetRequestProjectsByIssueStore() storage.RequestProjectsByIssueStore { + return f.requestProjectsByIssueStore +} + 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 00146fe31..a2d4dbd72 100644 --- a/stovepipe/extension/storage/mysql/storage_test.go +++ b/stovepipe/extension/storage/mysql/storage_test.go @@ -198,6 +198,14 @@ func TestStorageForQueueMetrics(t *testing.T) { return err }, }, + { + name: "request_projects_by_issue_store", + op: "list", + read: func(bound storage.Storage, _ string) error { + _, err := bound.GetRequestProjectsByIssueStore().List(context.Background(), "1", entity.ProjectIssueAttribution{}, 1) + return err + }, + }, } for _, tt := range tests { diff --git a/stovepipe/extension/storage/request_projects_by_issue_store.go b/stovepipe/extension/storage/request_projects_by_issue_store.go new file mode 100644 index 000000000..47f77dde5 --- /dev/null +++ b/stovepipe/extension/storage/request_projects_by_issue_store.go @@ -0,0 +1,33 @@ +// 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 +// +// 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_projects_by_issue_store.go -destination=mock/request_projects_by_issue_store_mock.go -package=mock + +import ( + "context" + + "github.com/uber/submitqueue/stovepipe/entity" +) + +// RequestProjectsByIssueStore retains immutable attribution rows per request within the bound queue. +type RequestProjectsByIssueStore interface { + // Create puts one row by request, task, unit, and project. + // Duplicates return ErrAlreadyExists. + Create(ctx context.Context, requestID string, entry entity.ProjectIssueAttribution) error + + // List scans one request in ascending bytewise task/unit/project order. + // after is exclusive; its zero value starts the range. limit must be positive. + // An empty page ends the range, not evidence that attribution completed. + List(ctx context.Context, requestID string, after entity.ProjectIssueAttribution, limit int) ([]entity.ProjectIssueAttribution, error) +} diff --git a/stovepipe/extension/storage/storage.go b/stovepipe/extension/storage/storage.go index 933e35e75..a438b5133 100644 --- a/stovepipe/extension/storage/storage.go +++ b/stovepipe/extension/storage/storage.go @@ -119,4 +119,6 @@ type Storage interface { GetValidationFactStore() ValidationFactStore GetProjectFactByDegreeStore() ProjectFactByDegreeStore + + GetRequestProjectsByIssueStore() RequestProjectsByIssueStore } diff --git a/test/integration/stovepipe/extension/storage/mysql/BUILD.bazel b/test/integration/stovepipe/extension/storage/mysql/BUILD.bazel index 8a6df3ead..f512cab85 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_projects_by_issue_test.go", "storage_test.go", ], data = [ diff --git a/test/integration/stovepipe/extension/storage/mysql/request_projects_by_issue_test.go b/test/integration/stovepipe/extension/storage/mysql/request_projects_by_issue_test.go new file mode 100644 index 000000000..8538fcd75 --- /dev/null +++ b/test/integration/stovepipe/extension/storage/mysql/request_projects_by_issue_test.go @@ -0,0 +1,126 @@ +// 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 testRequestProjectsByIssue(t *testing.T, ctx context.Context, backend *mysqlstorage.Storage) { + bound, err := backend.For("attribution/main") + require.NoError(t, err) + store := bound.GetRequestProjectsByIssueStore() + entries := []entity.ProjectIssueAttribution{ + {TaskID: "z", UnitID: "last", Project: "last"}, + {TaskID: "test", UnitID: "x", Project: "b"}, + {TaskID: "test", UnitID: "x", Project: "a"}, + {TaskID: "Test", UnitID: "x", Project: "a"}, + {TaskID: "test", UnitID: "x", Project: "A"}, + {TaskID: "test", UnitID: "x\x00", Project: "after"}, + } + require.NoError(t, store.Create(ctx, "1", entries[0])) + _, err = bound.GetRequestLogStore().Get(ctx, "1", "project-issues/all") + require.ErrorIs(t, err, storage.ErrNotFound) + for i, entry := range entries { + err := store.Create(ctx, "1", entry) + if i == 0 { + require.ErrorIs(t, err, storage.ErrAlreadyExists) + } else { + require.NoError(t, err) + } + 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.GetRequestProjectsByIssueStore().Create(ctx, "1", unrelated)) + require.NoError(t, store.Create(ctx, "2", unrelated)) + + want := []entity.ProjectIssueAttribution{entries[3], entries[4], entries[2], entries[1], entries[5], entries[0]} + for _, pageSize := range []int{1, 2, 3, 20} { + var got []entity.ProjectIssueAttribution + var after entity.ProjectIssueAttribution + for { + page, err := store.List(ctx, "1", 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", entity.ProjectIssueAttribution{TaskID: "test", UnitID: "w", Project: "service"}, 2) + require.NoError(t, err) + require.Equal(t, want[1:3], page) + page, err = store.List(ctx, "4", entity.ProjectIssueAttribution{}, 2) + require.NoError(t, err) + require.Empty(t, page) + + log := entity.RequestLog{ + ID: "project-issues/all", Queue: "attribution/main", RequestID: "1", TimestampMs: 1000, + Event: entity.RequestEventProjectIssuesRecorded, + Metadata: map[string]string{"build_id": "build-1"}, + } + require.NoError(t, bound.GetRequestLogStore().Create(ctx, log)) + duplicate := log + duplicate.TimestampMs = 2000 + require.ErrorIs(t, bound.GetRequestLogStore().Create(ctx, duplicate), storage.ErrAlreadyExists) + retained, err := bound.GetRequestLogStore().Get(ctx, log.RequestID, log.ID) + require.NoError(t, err) + require.Equal(t, log, retained) + page, err = store.List(ctx, retained.RequestID, entity.ProjectIssueAttribution{}, 20) + require.NoError(t, err) + require.Equal(t, want, page) + + empty := log + empty.RequestID = "3" + empty.Metadata = map[string]string{"build_id": "build-3"} + require.NoError(t, bound.GetRequestLogStore().Create(ctx, empty)) + retained, err = bound.GetRequestLogStore().Get(ctx, empty.RequestID, empty.ID) + require.NoError(t, err) + page, err = store.List(ctx, retained.RequestID, entity.ProjectIssueAttribution{}, 2) + require.NoError(t, err) + require.Empty(t, page) + _, err = bound.GetRequestLogStore().Get(ctx, "4", empty.ID) + require.ErrorIs(t, err, storage.ErrNotFound) +} + +func testRequestProjectsByIssueKeyLimits(t *testing.T, ctx context.Context, backend *mysqlstorage.Storage) { + bound, err := backend.For(strings.Repeat("q", 255)) + require.NoError(t, err) + store := bound.GetRequestProjectsByIssueStore() + 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", entity.ProjectIssueAttribution{}, 1) + require.NoError(t, err) + require.Equal(t, []entity.ProjectIssueAttribution{entry}, page) + page, err = store.List(ctx, "9223372036854775807", 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 458d4843b..b2af12820 100644 --- a/test/integration/stovepipe/extension/storage/mysql/storage_test.go +++ b/test/integration/stovepipe/extension/storage/mysql/storage_test.go @@ -95,6 +95,16 @@ func TestMySQLStorage(t *testing.T) { testBuildIssueStore(t, ctx, db, factory) }) + t.Run("RequestProjectsByIssue", func(t *testing.T) { + resetStorage(t, db) + testRequestProjectsByIssue(t, ctx, backend) + }) + + t.Run("RequestProjectsByIssueKeyLimits", func(t *testing.T) { + resetStorage(t, db) + testRequestProjectsByIssueKeyLimits(t, ctx, backend) + }) + t.Run("BuildIssueList", func(t *testing.T) { resetStorage(t, db) testBuildIssueList(t, ctx, factory) @@ -227,6 +237,7 @@ func resetStorage(t *testing.T, db *sql.DB) { "TRUNCATE TABLE queue", "TRUNCATE TABLE validation_fact", "TRUNCATE TABLE project_fact_by_degree", + "TRUNCATE TABLE request_projects_by_issue", } { _, err := db.ExecContext(context.Background(), statement) require.NoError(t, err) From a8b64066d3e0857903225bf3ddbdd40981c01967 Mon Sep 17 00:00:00 2001 From: "prath.shenoy" Date: Mon, 9 Nov 2026 03:51:10 +0000 Subject: [PATCH 2/3] 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) From f6db4c734cc7eaa67645715c9fdc0778ec14e21b Mon Sep 17 00:00:00 2001 From: "prath.shenoy" Date: Mon, 9 Nov 2026 14:04:23 +0000 Subject: [PATCH 3/3] feat(stovepipe): Materialize project issue lookup ## Summary **What**: - Add a materializer that retains per-request issue-to-project associations and creates their reverse lookups. **Why**: - Make both query directions recoverable when recording is interrupted. An association can be saved before its reverse lookup is written; replaying the same pair repairs the missing lookup without duplicating either row. This provides the write path for the recording-stage integration. ## Test Plan - [x] Add unit tests. ## Revert Plan - Revert this PR. The helper is not wired into processing yet, and no schemas or existing write paths change. ## Issue Links - [CODEM-607](https://linear.app/uber/issue/CODEM-607) --- stovepipe/core/projectissues/BUILD.bazel | 30 ++++ stovepipe/core/projectissues/README.md | 7 + stovepipe/core/projectissues/materializer.go | 68 +++++++++ .../core/projectissues/materializer_test.go | 136 ++++++++++++++++++ 4 files changed, 241 insertions(+) create mode 100644 stovepipe/core/projectissues/BUILD.bazel create mode 100644 stovepipe/core/projectissues/README.md create mode 100644 stovepipe/core/projectissues/materializer.go create mode 100644 stovepipe/core/projectissues/materializer_test.go diff --git a/stovepipe/core/projectissues/BUILD.bazel b/stovepipe/core/projectissues/BUILD.bazel new file mode 100644 index 000000000..d0a291245 --- /dev/null +++ b/stovepipe/core/projectissues/BUILD.bazel @@ -0,0 +1,30 @@ +load("@rules_go//go:def.bzl", "go_library", "go_test") + +go_library( + name = "go_default_library", + srcs = ["materializer.go"], + importpath = "github.com/uber/submitqueue/stovepipe/core/projectissues", + visibility = ["//visibility:public"], + deps = [ + "//platform/base/id:go_default_library", + "//platform/metrics:go_default_library", + "//stovepipe/entity:go_default_library", + "//stovepipe/extension/storage:go_default_library", + "@com_github_uber_go_tally//:go_default_library", + ], +) + +go_test( + name = "go_default_test", + srcs = ["materializer_test.go"], + embed = [":go_default_library"], + deps = [ + "//stovepipe/entity:go_default_library", + "//stovepipe/extension/storage:go_default_library", + "//stovepipe/extension/storage/mock:go_default_library", + "@com_github_stretchr_testify//assert:go_default_library", + "@com_github_stretchr_testify//require:go_default_library", + "@com_github_uber_go_tally//:go_default_library", + "@org_uber_go_mock//gomock:go_default_library", + ], +) diff --git a/stovepipe/core/projectissues/README.md b/stovepipe/core/projectissues/README.md new file mode 100644 index 000000000..1f31d5dde --- /dev/null +++ b/stovepipe/core/projectissues/README.md @@ -0,0 +1,7 @@ +# Request issue attribution materialization + +The materializer retains each issue/project pair in `request_projects_by_issue` before creating its reverse lookup in `request_issues_by_project`. It uses single-row creates, with no transaction or separate consumer. + +Both rows are immutable and contain only their primary-key fields. A duplicate therefore identifies the same pair; there is no payload to reconcile. Replaying a retained pair still attempts the reverse write, repairing a lookup left missing by an interrupted attempt. Requests have separate keys, so an older request cannot replace a newer request's attribution. + +The helper handles one pair at a time. Its caller owns attribution computation and the request-level completion event, and must retry failed pairs before marking attribution complete. A successful pair does not mean the entire request's attribution is complete. diff --git a/stovepipe/core/projectissues/materializer.go b/stovepipe/core/projectissues/materializer.go new file mode 100644 index 000000000..a9077a5c8 --- /dev/null +++ b/stovepipe/core/projectissues/materializer.go @@ -0,0 +1,68 @@ +// 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 projectissues retains request attribution and materializes its reverse lookup. +package projectissues + +import ( + "context" + "errors" + "fmt" + + "github.com/uber-go/tally" + "github.com/uber/submitqueue/platform/base/id" + "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/storage" +) + +type Materializer interface { + // PersistAttribution retains one pair before projecting its reverse lookup. + // Replaying a retained pair repairs a missing lookup. + PersistAttribution(ctx context.Context, stores storage.Storage, requestID string, entry entity.ProjectIssueAttribution) error +} + +type materializer struct{ scope tally.Scope } + +func NewMaterializer(scope tally.Scope) Materializer { + return &materializer{scope: scope.SubScope("project_issues_materializer")} +} + +func (m *materializer) PersistAttribution(ctx context.Context, stores storage.Storage, requestID string, entry entity.ProjectIssueAttribution) (retErr error) { + op := metrics.Begin(m.scope, "persist_attribution", metrics.StorageLatencyBuckets, metrics.TagsFromContext(ctx)...) + defer func() { op.Complete(retErr) }() + + if err := ctx.Err(); err != nil { + return err + } + if err := id.Validate(requestID); err != nil { + return fmt.Errorf("invalid attribution request ID: %w", err) + } + if err := entry.Validate(); err != nil { + return fmt.Errorf("invalid project issue attribution: %w", err) + } + + // Both rows contain only their immutable key, so a duplicate is the same pair. + err := stores.GetRequestProjectsByIssueStore().Create(ctx, requestID, entry) + if err != nil && !errors.Is(err, storage.ErrAlreadyExists) { + return fmt.Errorf("retain request projects by issue: %w", err) + } + err = stores.GetRequestIssuesByProjectStore().Create(ctx, requestID, entry) + if err != nil && !errors.Is(err, storage.ErrAlreadyExists) { + return fmt.Errorf("materialize request issues by project: %w", err) + } + return nil +} + +var _ Materializer = (*materializer)(nil) diff --git a/stovepipe/core/projectissues/materializer_test.go b/stovepipe/core/projectissues/materializer_test.go new file mode 100644 index 000000000..d7edda522 --- /dev/null +++ b/stovepipe/core/projectissues/materializer_test.go @@ -0,0 +1,136 @@ +// 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 projectissues + +import ( + "context" + "errors" + "fmt" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/uber-go/tally" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/storage" + storagemock "github.com/uber/submitqueue/stovepipe/extension/storage/mock" + "go.uber.org/mock/gomock" +) + +func TestPersistAttribution(t *testing.T) { + for _, tt := range []struct { + name string + sourceErr error + projectionErr error + wantErr error + }{ + {name: "new pair"}, + {name: "retained source", sourceErr: fmt.Errorf("duplicate source: %w", storage.ErrAlreadyExists)}, + {name: "existing projection", projectionErr: fmt.Errorf("duplicate lookup: %w", storage.ErrAlreadyExists)}, + {name: "replay", sourceErr: storage.ErrAlreadyExists, projectionErr: storage.ErrAlreadyExists}, + {name: "source failure", sourceErr: assert.AnError, wantErr: assert.AnError}, + {name: "projection failure", projectionErr: assert.AnError, wantErr: assert.AnError}, + } { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + stores := storagemock.NewMockStorage(ctrl) + source := storagemock.NewMockRequestProjectsByIssueStore(ctrl) + projection := storagemock.NewMockRequestIssuesByProjectStore(ctrl) + entry := entity.ProjectIssueAttribution{TaskID: "test", UnitID: "//app:test", Project: "app"} + + calls := []any{ + stores.EXPECT().GetRequestProjectsByIssueStore().Return(source), + source.EXPECT().Create(gomock.Any(), "42", entry).Return(tt.sourceErr), + } + if tt.sourceErr == nil || errors.Is(tt.sourceErr, storage.ErrAlreadyExists) { + calls = append(calls, + stores.EXPECT().GetRequestIssuesByProjectStore().Return(projection), + projection.EXPECT().Create(gomock.Any(), "42", entry).Return(tt.projectionErr), + ) + } + gomock.InOrder(calls...) + + err := NewMaterializer(tally.NoopScope).PersistAttribution(context.Background(), stores, "42", entry) + if tt.wantErr != nil { + require.ErrorIs(t, err, tt.wantErr) + } else { + require.NoError(t, err) + } + }) + } +} + +func TestPersistAttributionRetriesPartialResults(t *testing.T) { + ctrl := gomock.NewController(t) + stores := storagemock.NewMockStorage(ctrl) + source := storagemock.NewMockRequestProjectsByIssueStore(ctrl) + projection := storagemock.NewMockRequestIssuesByProjectStore(ctrl) + stores.EXPECT().GetRequestProjectsByIssueStore().Return(source).AnyTimes() + stores.EXPECT().GetRequestIssuesByProjectStore().Return(projection).AnyTimes() + entries := []entity.ProjectIssueAttribution{ + {TaskID: "test", UnitID: "//app:test", Project: "app"}, + {TaskID: "test", UnitID: "//app:test", Project: "dependent"}, + {TaskID: "lint", UnitID: "//lib:lint", Project: "library"}, + } + + gomock.InOrder( + source.EXPECT().Create(gomock.Any(), "42", entries[0]).Return(nil), + projection.EXPECT().Create(gomock.Any(), "42", entries[0]).Return(nil), + source.EXPECT().Create(gomock.Any(), "42", entries[1]).Return(nil), + projection.EXPECT().Create(gomock.Any(), "42", entries[1]).Return(assert.AnError), + source.EXPECT().Create(gomock.Any(), "42", entries[0]).Return(storage.ErrAlreadyExists), + projection.EXPECT().Create(gomock.Any(), "42", entries[0]).Return(storage.ErrAlreadyExists), + source.EXPECT().Create(gomock.Any(), "42", entries[1]).Return(storage.ErrAlreadyExists), + projection.EXPECT().Create(gomock.Any(), "42", entries[1]).Return(nil), + source.EXPECT().Create(gomock.Any(), "42", entries[2]).Return(nil), + projection.EXPECT().Create(gomock.Any(), "42", entries[2]).Return(nil), + ) + + m := NewMaterializer(tally.NoopScope) + ctx := context.Background() + require.NoError(t, m.PersistAttribution(ctx, stores, "42", entries[0])) + require.ErrorIs(t, m.PersistAttribution(ctx, stores, "42", entries[1]), assert.AnError) + for _, entry := range entries { + require.NoError(t, m.PersistAttribution(ctx, stores, "42", entry)) + } +} + +func TestPersistAttributionRejectsInvalidInput(t *testing.T) { + for _, tt := range []struct { + name string + requestID string + entry entity.ProjectIssueAttribution + }{ + {name: "missing request", entry: entity.ProjectIssueAttribution{TaskID: "test", UnitID: "//app:test", Project: "app"}}, + {name: "invalid request", requestID: "not-a-request", entry: entity.ProjectIssueAttribution{TaskID: "test", UnitID: "//app:test", Project: "app"}}, + {name: "missing task", requestID: "42", entry: entity.ProjectIssueAttribution{UnitID: "//app:test", Project: "app"}}, + {name: "task wide issue", requestID: "42", entry: entity.ProjectIssueAttribution{TaskID: "test", Project: "app"}}, + {name: "missing project", requestID: "42", entry: entity.ProjectIssueAttribution{TaskID: "test", UnitID: "//app:test"}}, + } { + t.Run(tt.name, func(t *testing.T) { + stores := storagemock.NewMockStorage(gomock.NewController(t)) + err := NewMaterializer(tally.NoopScope).PersistAttribution(context.Background(), stores, tt.requestID, tt.entry) + require.Error(t, err) + }) + } +} + +func TestPersistAttributionCancelled(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + stores := storagemock.NewMockStorage(gomock.NewController(t)) + entry := entity.ProjectIssueAttribution{TaskID: "test", UnitID: "//app:test", Project: "app"} + require.ErrorIs(t, NewMaterializer(tally.NoopScope).PersistAttribution(ctx, stores, "42", entry), context.Canceled) +}