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) +} 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..42618493f 100644 --- a/stovepipe/extension/storage/BUILD.bazel +++ b/stovepipe/extension/storage/BUILD.bazel @@ -16,7 +16,9 @@ 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", "request_summary_store.go", "request_uri_store.go", diff --git a/stovepipe/extension/storage/README.md b/stovepipe/extension/storage/README.md index c04f76d14..b56cc8ca2 100644 --- a/stovepipe/extension/storage/README.md +++ b/stovepipe/extension/storage/README.md @@ -26,6 +26,20 @@ 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. + +## 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 15380ccf8..34e532dc3 100644 --- a/stovepipe/extension/storage/mock/BUILD.bazel +++ b/stovepipe/extension/storage/mock/BUILD.bazel @@ -16,7 +16,9 @@ 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", "request_summary_store_mock.go", "request_uri_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/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..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() @@ -275,6 +289,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..d8aa56ce7 100644 --- a/stovepipe/extension/storage/mysql/BUILD.bazel +++ b/stovepipe/extension/storage/mysql/BUILD.bazel @@ -16,7 +16,9 @@ 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", "request_summary_store.go", "request_uri_store.go", @@ -53,7 +55,9 @@ 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", "request_summary_store_test.go", "request_uri_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 new file mode 100644 index 000000000..80c8e0ebe --- /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 := validateProjectIssueRequest(s.queue, 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 := validateProjectIssueRequest(s.queue, 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 validateProjectIssueRequest(queue, requestID string) error { + if err := id.Validate(requestID); err != nil { + return fmt.Errorf("invalid project issue attribution request ID: %w", err) + } + if queue == "" { + return fmt.Errorf("project issue attribution queue must not be empty") + } + if len(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_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/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..383760ed9 100644 --- a/stovepipe/extension/storage/mysql/storage.go +++ b/stovepipe/extension/storage/mysql/storage.go @@ -46,24 +46,26 @@ 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), + requestIssuesByProjectStore: NewRequestIssuesByProjectStore(s.db, queueScope.SubScope("request_issues_by_project_store"), queueName), }, nil } @@ -74,24 +76,26 @@ 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 + requestIssuesByProjectStore storage.RequestIssuesByProjectStore } // Verify mysqlStorage implements the queue-scoped aggregate at compile time. @@ -141,6 +145,14 @@ func (f *mysqlStorage) GetProjectFactByDegreeStore() storage.ProjectFactByDegree return f.projectFactByDegreeStore } +func (f *mysqlStorage) GetRequestProjectsByIssueStore() storage.RequestProjectsByIssueStore { + 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 00146fe31..2321c5988 100644 --- a/stovepipe/extension/storage/mysql/storage_test.go +++ b/stovepipe/extension/storage/mysql/storage_test.go @@ -198,6 +198,22 @@ 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 + }, + }, + { + 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/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..dcfd87ea6 100644 --- a/stovepipe/extension/storage/storage.go +++ b/stovepipe/extension/storage/storage.go @@ -119,4 +119,8 @@ type Storage interface { GetValidationFactStore() ValidationFactStore 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 8a6df3ead..2f7c2d5c5 100644 --- a/test/integration/stovepipe/extension/storage/mysql/BUILD.bazel +++ b/test/integration/stovepipe/extension/storage/mysql/BUILD.bazel @@ -19,6 +19,8 @@ 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", ], data = [ 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/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..ce1d9506f 100644 --- a/test/integration/stovepipe/extension/storage/mysql/storage_test.go +++ b/test/integration/stovepipe/extension/storage/mysql/storage_test.go @@ -95,6 +95,26 @@ 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("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) @@ -227,6 +247,8 @@ 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", + "TRUNCATE TABLE request_issues_by_project", } { _, err := db.ExecContext(context.Background(), statement) require.NoError(t, err)