feat(stovepipe): Record project issue attribution

doc/rfc/stovepipe/steps/record.md

@@ -1,6 +1,6 @@
 # Record stage
 
-`record` turns a terminal build outcome into a durable validation fact.
+`record` turns a terminal build outcome into durable validation facts and per-request issue/project associations.
 
 - Phase 1 records whole-repository greenness and optional project facts. On green it promotes the commit onto the Queue's promotion ref if the Request still holds the last-green bookmark, then announces the outcome as a hook event. This is implemented: see [stovepipe/controller/record/record.go](../../../../stovepipe/controller/record/record.go).
 - Phase 2 records greenness per project instead of per repository. Sketched here, to be expanded before implementation.
@@ -9,7 +9,7 @@ Notifying downstream systems rides the cross-domain hook framework rather than a
 
 See [workflow.md](../workflow.md) for the whole pipeline, [build.md](build.md) for how builds are created, and [buildsignal.md](buildsignal.md) for the terminal-only handoff into this stage.
 
-`record` owns the validation facts and the promotion ref. `buildsignal` advances the last-green build baseline before handing off to this stage. `record` does not decide build scope, poll a build, release the Queue's build slot, interpret a target graph, or map targets to projects. Those belong to `process`, `buildsignal`, and `analyze`.
+`record` owns validation facts, issue/project attribution recording, and the promotion ref. `buildsignal` advances the last-green build baseline before handing off to this stage. Repository-specific project discovery and dependency traversal belong to the project-status and issue-attribution extensions, not the controller.
 
 ## Phase 1 algorithm
 
@@ -39,8 +39,11 @@ For a delivery carrying a `Record` payload:
    - superseded -> ack, no fact. Unreachable in practice.
    - accepted / processing / anything else -> return a non-retryable invariant error.
 
-6. Resolve optional project results using R and its TerminalBuildID, and validate
-   the entire result set before writing facts. A resolver error returns raw.
+6. Record issue/project attribution after terminal issue analysis completes
+   (see "Issue/project attribution"). While analysis is pending, hold the delivery
+   for 30 seconds without consuming its retry budget.
+   Then resolve optional project results using R and its TerminalBuildID, and
+   validate the entire result set before writing facts. Extension errors return raw.
 
 7. Map R.State and the project results to a whole-repository degree and create the
    ValidationFact keyed by (R.URI, empty project) in the queue-bound fact store.
@@ -75,6 +78,16 @@ Every decision after step 7 uses the persisted fact, not the outcome read from t
 
 Project-status processing delays fact creation and promotion, while the build baseline is already available from `buildsignal`. All fact writes and history complete before promotion and the recorded notification. The fact writes, bookmark advancement, and ref promotion remain separate operations.
 
+## Issue/project attribution
+
+The record stage calls the queue's issue attributer with the request. The attributer loads the confirmed issues from its terminal build and determines their affected projects. Recording waits for that build's `build_analysis_completed` occurrence, because the request becomes terminal before final issue analysis necessarily finishes. Completion for a different build does not release the wait. Abandoned analysis returns an error instead of waiting indefinitely; the existing record DLQ path handles recovery.
+
+Each returned issue/project pair goes through the attribution materializer, which retains `request_projects_by_issue` before creating `request_issues_by_project`. Only after every pair is durable in both directions does recording retain `project_issues_recorded`, identified by the terminal build and carrying its build ID. An empty successful result also records completion, without association rows. An extension error discards all returned results.
+
+Replays skip attribution when completion is already retained. Before completion, retries may recompute attribution and replay any already-written pairs; attributers must return stable results for the same request. A failed pair or completion write returns an error and cannot mark the request's attribution complete. These are historical per-request associations; closing an issue later does not remove them.
+
+Cancelled and superseded requests do not run attribution. A failed request without a recorded CI outcome has no terminal build to attribute and skips this step. Record DLQ recovery does not compute attribution or claim it completed; any incomplete association rows remain incomplete, with `record_abandoned` identifying the stopped recording attempt.
+
 ## Validation facts
 
 A fact answers "how broken was this scope at this Queue URI?" Its identity is `(queue, uri, project)`, where `project` is empty for whole-repository greenness and a stable project id in Phase 2.

service/stovepipe/server/BUILD.bazel

@@ -41,6 +41,7 @@ go_library(
         "//stovepipe/extension/buildrunner/fake:go_default_library",
         "//stovepipe/extension/fixer:go_default_library",
         "//stovepipe/extension/fixer/noop:go_default_library",
+        "//stovepipe/extension/projectissues/noop:go_default_library",
         "//stovepipe/extension/projectstatus/noop:go_default_library",
         "//stovepipe/extension/queueconfig/default:go_default_library",
         "//stovepipe/extension/sourcecontrol:go_default_library",

service/stovepipe/server/main.go

@@ -63,6 +63,7 @@ import (
 	buildrunnerfake "github.com/uber/submitqueue/stovepipe/extension/buildrunner/fake"
 	"github.com/uber/submitqueue/stovepipe/extension/fixer"
 	fixernoop "github.com/uber/submitqueue/stovepipe/extension/fixer/noop"
+	projectissuesnoop "github.com/uber/submitqueue/stovepipe/extension/projectissues/noop"
 	projectstatusnoop "github.com/uber/submitqueue/stovepipe/extension/projectstatus/noop"
 	queueconfigdefault "github.com/uber/submitqueue/stovepipe/extension/queueconfig/default"
 	"github.com/uber/submitqueue/stovepipe/extension/sourcecontrol"
@@ -435,7 +436,7 @@ func registerPrimaryControllers(
 	}
 	count++
 
-	recordController := record.NewController(logger, scope, store, materializer, projectstatusnoop.New(), sourceControl, registry, stovepipemq.TopicKeyRecord, "stovepipe-record")
+	recordController := record.NewController(logger, scope, store, materializer, projectstatusnoop.New(), projectissuesnoop.New(), sourceControl, registry, stovepipemq.TopicKeyRecord, "stovepipe-record")
 	if err := c.Register(recordController); err != nil {
 		return count, fmt.Errorf("failed to register record controller: %w", err)
 	}

stovepipe/controller/record/BUILD.bazel

@@ -18,8 +18,10 @@ go_library(
         "//stovepipe/core/hookevent:go_default_library",
         "//stovepipe/core/loader:go_default_library",
         "//stovepipe/core/messagequeue:go_default_library",
+        "//stovepipe/core/projectissues:go_default_library",
         "//stovepipe/core/requestlog:go_default_library",
         "//stovepipe/entity:go_default_library",
+        "//stovepipe/extension/projectissues:go_default_library",
         "//stovepipe/extension/projectstatus:go_default_library",
         "//stovepipe/extension/sourcecontrol:go_default_library",
         "//stovepipe/extension/storage:go_default_library",
@@ -48,6 +50,9 @@ go_test(
         "//stovepipe/core/requestlog:go_default_library",
         "//stovepipe/core/requestlog/mock:go_default_library",
         "//stovepipe/entity:go_default_library",
+        "//stovepipe/extension/projectissues:go_default_library",
+        "//stovepipe/extension/projectissues/mock:go_default_library",
+        "//stovepipe/extension/projectissues/noop:go_default_library",
         "//stovepipe/extension/projectstatus:go_default_library",
         "//stovepipe/extension/projectstatus/mock:go_default_library",
         "//stovepipe/extension/projectstatus/noop:go_default_library",

stovepipe/controller/record/record.go

@@ -43,8 +43,10 @@ import (
 	"github.com/uber/submitqueue/stovepipe/core/hookevent"
 	"github.com/uber/submitqueue/stovepipe/core/loader"
 	stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue"
+	projectissuescore "github.com/uber/submitqueue/stovepipe/core/projectissues"
 	"github.com/uber/submitqueue/stovepipe/core/requestlog"
 	"github.com/uber/submitqueue/stovepipe/entity"
+	"github.com/uber/submitqueue/stovepipe/extension/projectissues"
 	"github.com/uber/submitqueue/stovepipe/extension/projectstatus"
 	"github.com/uber/submitqueue/stovepipe/extension/sourcecontrol"
 	"github.com/uber/submitqueue/stovepipe/extension/storage"
@@ -55,21 +57,26 @@ import (
 // when that fact is green and holds the queue's last-green bookmark promotes
 // the commit. Implements consumer.Controller.
 type Controller struct {
-	logger                       *zap.SugaredLogger
-	metricsScope                 tally.Scope
-	materializer                 requestlog.Materializer
-	registry                     consumer.TopicRegistry
-	stores                       storage.Factory
-	projectStatusResolverFactory projectstatus.ResolverFactory
-	sourceControl                sourcecontrol.Factory
-	topicKey                     consumer.TopicKey
-	consumerGroup                string
+	logger                         *zap.SugaredLogger
+	metricsScope                   tally.Scope
+	materializer                   requestlog.Materializer
+	registry                       consumer.TopicRegistry
+	stores                         storage.Factory
+	projectStatusResolverFactory   projectstatus.ResolverFactory
+	projectIssuesAttributerFactory projectissues.AttributerFactory
+	projectIssuesMaterializer      projectissuescore.Materializer
+	sourceControl                  sourcecontrol.Factory
+	topicKey                       consumer.TopicKey
+	consumerGroup                  string
 }
 
 // Verify Controller implements consumer.Controller interface at compile time.
 var _ consumer.Controller = (*Controller)(nil)
 
-const _opName = "record"
+const (
+	_opName                    = "record"
+	_analysisWaitDelayMs int64 = 30000
+)
 
 // wholeRepositoryProject is the project component of a fact covering the whole
 // repository rather than one project within it.
@@ -82,22 +89,25 @@ func NewController(
 	stores storage.Factory,
 	materializer requestlog.Materializer,
 	projectStatusResolverFactory projectstatus.ResolverFactory,
+	projectIssuesAttributerFactory projectissues.AttributerFactory,
 	sourceControl sourcecontrol.Factory,
 	registry consumer.TopicRegistry,
 	topicKey consumer.TopicKey,
 	consumerGroup string,
 ) *Controller {
 	name := string(topicKey) + "_controller"
 	return &Controller{
-		logger:                       logger.Named(name),
-		metricsScope:                 scope.SubScope(name),
-		materializer:                 materializer,
-		registry:                     registry,
-		stores:                       stores,
-		projectStatusResolverFactory: projectStatusResolverFactory,
-		sourceControl:                sourceControl,
-		topicKey:                     topicKey,
-		consumerGroup:                consumerGroup,
+		logger:                         logger.Named(name),
+		metricsScope:                   scope.SubScope(name),
+		materializer:                   materializer,
+		registry:                       registry,
+		stores:                         stores,
+		projectStatusResolverFactory:   projectStatusResolverFactory,
+		projectIssuesAttributerFactory: projectIssuesAttributerFactory,
+		projectIssuesMaterializer:      projectissuescore.NewMaterializer(scope),
+		sourceControl:                  sourceControl,
+		topicKey:                       topicKey,
+		consumerGroup:                  consumerGroup,
 	}
 }
 
@@ -143,6 +153,14 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 
 	switch request.State {
 	case entity.RequestStateSucceeded, entity.RequestStateFailed:
+		complete, err := c.recordProjectIssues(ctx, store, request)
+		if err != nil {
+			return err
+		}
+		if !complete {
+			delivery.Hold(_analysisWaitDelayMs)
+			return nil
+		}
 		results, err := c.resolveProjectResults(ctx, request)
 		if err != nil {
 			return err
@@ -183,6 +201,70 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 	}
 }
 
+func (c *Controller) recordProjectIssues(ctx context.Context, store storage.Storage, request entity.Request) (complete bool, retErr error) {
+	op := metrics.Begin(c.metricsScope, "record_project_issues", metrics.LongLatencyBuckets, metrics.TagsFromContext(ctx)...)
+	defer func() { op.Complete(retErr) }()
+
+	if request.TerminalBuildID == "" {
+		return true, nil
+	}
+	history := store.GetRequestLogStore()
+	completion := requestlog.NewRequestEventLog(request, entity.RequestEventProjectIssuesRecorded, request.TerminalBuildID,
+		map[string]string{requestlog.MetadataKeyBuildID: request.TerminalBuildID})
+	_, err := history.Get(ctx, request.ID, completion.ID)
+	if err == nil {
+		return true, nil
+	}
+	if !errors.Is(err, storage.ErrNotFound) {
+		return false, fmt.Errorf("read project issue completion: %w", err)
+	}
+
+	// The terminal request can be visible before analysis has saved its final issues.
+	analysis := requestlog.NewRequestEventLog(request, entity.RequestEventBuildAnalysisCompleted, request.TerminalBuildID, nil)
+	_, err = history.Get(ctx, request.ID, analysis.ID)
+	if errors.Is(err, storage.ErrNotFound) {
+		abandoned := requestlog.NewRequestEventLog(request, entity.RequestEventBuildAnalysisAbandoned, request.TerminalBuildID, nil)
+		_, err = history.Get(ctx, request.ID, abandoned.ID)
+		if errors.Is(err, storage.ErrNotFound) {
+			return false, nil
+		}
+		if err != nil {
+			return false, fmt.Errorf("read abandoned build analysis: %w", err)
+		}
+		return false, fmt.Errorf("cannot attribute issues for build %q: build analysis was abandoned", request.TerminalBuildID)
+	}
+	if err != nil {
+		return false, fmt.Errorf("read completed build analysis: %w", err)
+	}
+
+	attributer, err := c.projectIssuesAttributerFactory.For(projectissues.AttributerConfig{QueueName: request.Queue})
+	if err != nil {
+		return false, fmt.Errorf("resolve project issue attributer for queue %q: %w", request.Queue, err)
+	}
+	attributions, err := attributer.Attribute(ctx, request)
+	if err != nil {
+		return false, fmt.Errorf("attribute issues for request %q: %w", request.ID, err)
+	}
+	for _, attribution := range attributions {
+		if attribution.TaskID == "" {
+			return false, fmt.Errorf("project issue attribution has an empty task ID")
+		}
+		if attribution.UnitID == "" {
+			return false, fmt.Errorf("project issue attribution requires a unit-level issue")
+		}
+		for _, project := range attribution.Projects {
+			entry := entity.ProjectIssueAttribution{TaskID: attribution.TaskID, UnitID: attribution.UnitID, Project: project}
+			if err := c.projectIssuesMaterializer.PersistAttribution(ctx, store, request.ID, entry); err != nil {
+				return false, err
+			}
+		}
+	}
+	if err := c.materializer.PersistLog(ctx, store, completion); err != nil {
+		return false, fmt.Errorf("record project issue completion: %w", err)
+	}
+	return true, nil
+}
+
 func (c *Controller) resolveProjectResults(ctx context.Context, request entity.Request) (results []projectstatus.Result, retErr error) {
 	op := metrics.Begin(c.metricsScope, "resolve_project_results", metrics.LongLatencyBuckets, metrics.TagsFromContext(ctx)...)
 	defer func() { op.Complete(retErr) }()