From 2909201696b24e9f307e07b30efe98acbb752566 Mon Sep 17 00:00:00 2001 From: "prath.shenoy" Date: Mon, 9 Nov 2026 17:00:21 +0000 Subject: [PATCH] feat(stovepipe): Add scoped attribution reads ## Summary **What**: - Add bounded reads for the projects attributed to one issue within a request. **Why**: - Avoid reading every association in a request just to find one issue's affected projects. Restricting the read to that issue provides the efficient lookup needed by the read controller. ## Test Plan - [x] Add unit tests. ## Revert Plan - Revert this PR. It adds a read operation without changing schemas or stored data. ## Issue Links - [CODEM-632](https://linear.app/uber/issue/CODEM-632/projects-issues) --- .../request_projects_by_issue_store_mock.go | 15 +++++ .../mysql/request_projects_by_issue_store.go | 33 +++++++++ .../request_projects_by_issue_store_test.go | 67 +++++++++++++++++++ .../request_issues_by_project_store.go | 1 + .../request_projects_by_issue_store.go | 5 ++ .../mysql/request_projects_by_issue_test.go | 18 +++++ 6 files changed, 139 insertions(+) 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 index 07f23e360..0ad20fd97 100644 --- a/stovepipe/extension/storage/mock/request_projects_by_issue_store_mock.go +++ b/stovepipe/extension/storage/mock/request_projects_by_issue_store_mock.go @@ -69,3 +69,18 @@ func (mr *MockRequestProjectsByIssueStoreMockRecorder) List(ctx, requestID, afte mr.mock.ctrl.T.Helper() return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "List", reflect.TypeOf((*MockRequestProjectsByIssueStore)(nil).List), ctx, requestID, after, limit) } + +// ListByIssue mocks base method. +func (m *MockRequestProjectsByIssueStore) ListByIssue(ctx context.Context, requestID, taskID, unitID, afterProject string, limit int) ([]entity.ProjectIssueAttribution, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "ListByIssue", ctx, requestID, taskID, unitID, afterProject, limit) + ret0, _ := ret[0].([]entity.ProjectIssueAttribution) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// ListByIssue indicates an expected call of ListByIssue. +func (mr *MockRequestProjectsByIssueStoreMockRecorder) ListByIssue(ctx, requestID, taskID, unitID, afterProject, limit any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ListByIssue", reflect.TypeOf((*MockRequestProjectsByIssueStore)(nil).ListByIssue), ctx, requestID, taskID, unitID, afterProject, limit) +} 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 80c8e0ebe..9664be5fe 100644 --- a/stovepipe/extension/storage/mysql/request_projects_by_issue_store.go +++ b/stovepipe/extension/storage/mysql/request_projects_by_issue_store.go @@ -38,6 +38,12 @@ const listRequestProjectsByIssueAfterCursorQuery = ` OR (task_id = ? AND unit_id = ? AND project > ?)) ORDER BY task_id ASC, unit_id ASC, project ASC LIMIT ?` +const listProjectsForRequestIssueQuery = ` + SELECT task_id, unit_id, project + FROM request_projects_by_issue + WHERE queue = ? AND request_id = ? AND task_id = ? AND unit_id = ? AND project > ? + ORDER BY project ASC LIMIT ?` + type requestProjectsByIssueStore struct { db *sql.DB scope tally.Scope @@ -92,12 +98,39 @@ func (s *requestProjectsByIssueStore) List(ctx context.Context, requestID string args = append(args, after.TaskID, after.TaskID, after.UnitID, after.TaskID, after.UnitID, after.Project) } args = append(args, limit) + return s.list(ctx, query, args...) +} + +func (s *requestProjectsByIssueStore) ListByIssue(ctx context.Context, requestID, taskID, unitID, afterProject string, limit int) (ret []entity.ProjectIssueAttribution, retErr error) { + op := metrics.Begin(s.scope, "list_by_issue", metrics.StorageLatencyBuckets) + defer func() { op.Complete(retErr) }() + + if err := validateProjectIssueRequest(s.queue, requestID); err != nil { + return nil, err + } + if taskID == "" || len(taskID) > 255 { + return nil, fmt.Errorf("project issue attribution requires a task ID of at most 255 bytes") + } + if unitID == "" || len(unitID) > 1024 { + return nil, fmt.Errorf("project issue attribution requires a unit ID of at most 1024 bytes") + } + if len(afterProject) > 1020 { + return nil, fmt.Errorf("project issue attribution cursor exceeds 1020-byte project limit") + } + if limit <= 0 { + return nil, fmt.Errorf("project issue attribution limit must be positive") + } + return s.list(ctx, listProjectsForRequestIssueQuery, s.queue, requestID, taskID, unitID, afterProject, limit) +} + +func (s *requestProjectsByIssueStore) list(ctx context.Context, query string, args ...any) ([]entity.ProjectIssueAttribution, error) { 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() + var ret []entity.ProjectIssueAttribution for rows.Next() { var entry entity.ProjectIssueAttribution if err := rows.Scan(&entry.TaskID, &entry.UnitID, &entry.Project); err != 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 index 979d311e7..5393c5c46 100644 --- a/stovepipe/extension/storage/mysql/request_projects_by_issue_store_test.go +++ b/stovepipe/extension/storage/mysql/request_projects_by_issue_store_test.go @@ -162,3 +162,70 @@ func TestRequestProjectsByIssueStoreRejectsInvalidInput(t *testing.T) { } require.NoError(t, mock.ExpectationsWereMet()) } + +func TestRequestProjectsByIssueStoreListByIssue(t *testing.T) { + entry := entity.ProjectIssueAttribution{TaskID: "test", UnitID: "unit", Project: "app"} + for _, after := range []string{"", "before"} { + for _, tt := range []struct { + name string + rows *sqlmock.Rows + err error + want []entity.ProjectIssueAttribution + wantErr bool + }{ + {name: "found", rows: sqlmock.NewRows([]string{"task_id", "unit_id", "project"}).AddRow(entry.TaskID, entry.UnitID, entry.Project), want: []entity.ProjectIssueAttribution{entry}}, + {name: "empty", rows: sqlmock.NewRows([]string{"task_id", "unit_id", "project"})}, + {name: "query error", err: errors.New("query failed"), wantErr: true}, + {name: "scan error", rows: sqlmock.NewRows([]string{"task_id", "unit_id", "project"}).AddRow(nil, entry.UnitID, entry.Project), wantErr: true}, + {name: "iteration error", rows: sqlmock.NewRows([]string{"task_id", "unit_id", "project"}).AddRow(entry.TaskID, entry.UnitID, entry.Project).RowError(0, errors.New("read failed")), wantErr: true}, + } { + t.Run(tt.name+"/"+after, func(t *testing.T) { + db, mock, err := sqlmock.New() + require.NoError(t, err) + defer db.Close() + query := mock.ExpectQuery(regexp.QuoteMeta(listProjectsForRequestIssueQuery)).WithArgs("queue", "7", "test", "unit", after, 2) + if tt.err != nil { + query.WillReturnError(tt.err) + } else { + query.WillReturnRows(tt.rows).RowsWillBeClosed() + } + got, err := NewRequestProjectsByIssueStore(db, testMetrics(), "queue").ListByIssue(context.Background(), "7", "test", "unit", after, 2) + if tt.wantErr { + require.Error(t, err) + } else { + require.NoError(t, err) + require.Equal(t, tt.want, got) + } + if tt.err != nil { + require.ErrorIs(t, err, tt.err) + } + require.NoError(t, mock.ExpectationsWereMet()) + }) + } + } +} + +func TestRequestProjectsByIssueStoreListByIssueInvalidInput(t *testing.T) { + db, mock, err := sqlmock.New() + require.NoError(t, err) + defer db.Close() + for _, tt := range []struct { + queue, request, task, unit, after string + limit int + }{ + {"", "7", "test", "unit", "", 1}, + {strings.Repeat("q", 256), "7", "test", "unit", "", 1}, + {"queue", "07", "test", "unit", "", 1}, + {"queue", "7", "", "unit", "", 1}, + {"queue", "7", strings.Repeat("t", 256), "unit", "", 1}, + {"queue", "7", "test", "", "", 1}, + {"queue", "7", "test", strings.Repeat("u", 1025), "", 1}, + {"queue", "7", "test", "unit", strings.Repeat("p", 1021), 1}, + {"queue", "7", "test", "unit", "", 0}, + {"queue", "7", "test", "unit", "", -1}, + } { + _, err := NewRequestProjectsByIssueStore(db, testMetrics(), tt.queue).ListByIssue(context.Background(), tt.request, tt.task, tt.unit, tt.after, tt.limit) + require.Error(t, err) + } + require.NoError(t, mock.ExpectationsWereMet()) +} diff --git a/stovepipe/extension/storage/request_issues_by_project_store.go b/stovepipe/extension/storage/request_issues_by_project_store.go index 430404c51..4e10766d7 100644 --- a/stovepipe/extension/storage/request_issues_by_project_store.go +++ b/stovepipe/extension/storage/request_issues_by_project_store.go @@ -30,5 +30,6 @@ type RequestIssuesByProjectStore interface { // 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. + // Pages are full unless the selected prefix is exhausted. 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 index 47f77dde5..774bf5fa4 100644 --- a/stovepipe/extension/storage/request_projects_by_issue_store.go +++ b/stovepipe/extension/storage/request_projects_by_issue_store.go @@ -30,4 +30,9 @@ type RequestProjectsByIssueStore interface { // 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) + + // ListByIssue scans the request/task/unit prefix in ascending bytewise project order. + // afterProject is exclusive; empty starts the range. limit must be positive. + // Pages are full unless the selected prefix is exhausted. + ListByIssue(ctx context.Context, requestID, taskID, unitID, afterProject string, limit int) ([]entity.ProjectIssueAttribution, error) } 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 index 8538fcd75..201fa1feb 100644 --- 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 @@ -73,6 +73,24 @@ func testRequestProjectsByIssue(t *testing.T, ctx context.Context, backend *mysq } require.Equal(t, want, got) } + for _, pageSize := range []int{1, 2, 3, 20} { + var got []entity.ProjectIssueAttribution + after := "" + for { + page, err := store.ListByIssue(ctx, "1", "test", "x", after, pageSize) + require.NoError(t, err) + if len(page) == 0 { + break + } + got = append(got, page...) + require.LessOrEqual(t, len(got), 3) + after = page[len(page)-1].Project + } + require.Equal(t, []entity.ProjectIssueAttribution{entries[4], entries[2], entries[1]}, got) + } + missing, err := store.ListByIssue(ctx, "1", "test", "missing", "", 2) + require.NoError(t, err) + require.Empty(t, missing) 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)