feat(stovepipe): Add scoped attribution reads

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 {

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())
+}

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)
 }

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)
 }

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)