feat(stovepipe): Materialize project issue lookup

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",
+    ],
+)

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.

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)

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

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",

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

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

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:

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 {

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",