feat(stovepipe): Record project validation facts

Makefile

@@ -579,7 +579,7 @@ local-stovepipe-stop: ## Stop the Stovepipe service
 
 mocks: ## Generate mock files using mockgen
 	@echo "Generating mocks..."
-	@$(BAZEL) run @rules_go//go -- generate ./submitqueue/extension/storage/... ./submitqueue/extension/buildrunner/... ./submitqueue/extension/changeprovider/... ./platform/extension/counter/... ./platform/extension/consumergate/... ./platform/extension/hook/... ./platform/extension/messagequeue/... ./submitqueue/extension/queueconfig/... ./runway/extension/merger/... ./submitqueue/extension/conflict/... ./submitqueue/extension/speculation/... ./submitqueue/extension/validator/... ./platform/consumer/... ./stovepipe/core/requestlog/... ./stovepipe/extension/storage/... ./stovepipe/extension/sourcecontrol/...
+	@$(BAZEL) run @rules_go//go -- generate ./submitqueue/extension/storage/... ./submitqueue/extension/buildrunner/... ./submitqueue/extension/changeprovider/... ./platform/extension/counter/... ./platform/extension/consumergate/... ./platform/extension/hook/... ./platform/extension/messagequeue/... ./submitqueue/extension/queueconfig/... ./runway/extension/merger/... ./submitqueue/extension/conflict/... ./submitqueue/extension/speculation/... ./submitqueue/extension/validator/... ./platform/consumer/... ./stovepipe/core/requestlog/... ./stovepipe/extension/storage/... ./stovepipe/extension/sourcecontrol/... ./stovepipe/extension/projectresult/...
 	@echo "Mocks generated successfully!"
 
 proto: ## Generate protobuf files from .proto definitions

doc/rfc/stovepipe/request-log.md

@@ -7,6 +7,7 @@ Stovepipe retains an append-only request log for each validation request. Its in
 - `build_triggered`;
 - `build_finished`;
 - `validation_fact_recorded`;
+- `project_facts_recorded`;
 - `record_abandoned`.
 
 The model deliberately follows SubmitQueue's distinction between statuses describing where a request is and events describing important activity that does not move it. It remains a bounded request-lifecycle log rather than a generic event bus or an audit of every correlated operation.
@@ -32,7 +33,7 @@ The following remain with their owning entities, metrics, or structured logs:
 
 - Queue latest-request and last-green bookmarks;
 - source-control promotion;
-- project analysis and project facts;
+- project analysis and individual project facts;
 - hooks and downstream notifications;
 - build-slot claims, waits, and releases;
 - queue handoffs, delivery attempts, holds, nacks, and visibility timeouts;
@@ -123,9 +124,10 @@ Immutable Request context such as URI, build strategy, and base URI remains on `
 | `build_triggered` | A runner accepted a build and its Build row became durable. | Build ID metadata and creation time |
 | `build_finished` | The Build first reached a write-once terminal status. | Build ID metadata and status-change time |
 | `validation_fact_recorded` | The immutable whole-repository fact became durable. | Degree metadata and fact creation time |
+| `project_facts_recorded` | All project validation facts returned for a request became durable. | Project-fact count metadata and completion time |
 | `record_abandoned` | Record-stage work stopped after exhausting primary retries. | Event retention time |
 
-Build running and unchanged polls are not retained. Trigger and terminal result explain the request outcome without turning polling into an unbounded log. Project facts remain outside the initial vocabulary.
+Build running and unchanged polls are not retained. Trigger and terminal result explain the request outcome without turning polling into an unbounded log. Individual project facts remain outside the vocabulary; one batch-completion event summarizes their durable recording.
 
 ### Evolution
 
@@ -157,6 +159,7 @@ SubmitQueue applies that identity to the message carrying a log to its materiali
 | Build triggered | Request ID, event kind, and build ID |
 | Build finished | Request ID, event kind, and build ID |
 | Validation fact recorded | Request ID, event kind, and whole-repository fact identity |
+| Project facts recorded | Request ID, event kind, and the completed project-fact batch |
 
 The controller passes the materializer the same queue-scoped storage aggregate used for the source write. The materializer preserves a supplied occurrence time or assigns the current time immediately before the first insertion attempt, then calls `RequestLogStore.Create`. If the ID already exists, it loads the stored record and compares the explicitly designated stable semantic fields. The first successfully retained timestamp is authoritative and is not compared with a later retry's newly sampled time. Metadata keys emitted by both records must agree, while a key present on only one record remains compatible so an additive metadata rollout does not turn retries of older occurrences into conflicts. Compatible content is idempotent success; conflicting content is an internal consistency error, and the stored record is never overwritten or enriched.
 
@@ -209,7 +212,7 @@ Request creation, Build changes, and fact creation use the same source-write, lo
 | Process | CAS to superseded or processing, then retain that state. | An existing state is reconstructed from Request context before ack or build publication. |
 | Build | Create Build after runner acceptance, then retain `build_triggered`. | An identical existing Build ensures the event before buildsignal publication. |
 | Buildsignal | Persist terminal Build and retain `build_finished`; CAS the Request outcome and retain its terminal state. | Existing terminal Build and Request outcome each ensure their own entry before record publication. |
-| Record | Create or verify the whole-repository fact, then retain `validation_fact_recorded`. | An identical fact owned by the Request ensures the event before bookmark or promotion work. |
+| Record | Create or verify the whole-repository fact, then retain `validation_fact_recorded`; after recording project facts, retain one `project_facts_recorded` event. | An identical whole-repository fact and project-fact batch ensure their events before bookmark or promotion work. |
 | Record DLQ | Retain `record_abandoned`, then acknowledge the remaining record work. | The stable event ID makes history retention idempotent without replaying facts, bookmarks, promotion, or hooks. |
 | Reconciler | CAS an unrecoverable non-terminal Request to failed, then retain failed. | An existing terminal Request is repaired from its persisted outcome without relabeling it. |
 

service/stovepipe/server/BUILD.bazel

@@ -33,6 +33,7 @@ go_library(
         "//stovepipe/core/requestlog:go_default_library",
         "//stovepipe/extension/buildrunner:go_default_library",
         "//stovepipe/extension/buildrunner/fake:go_default_library",
+        "//stovepipe/extension/projectresult/noop:go_default_library",
         "//stovepipe/extension/queueconfig/default:go_default_library",
         "//stovepipe/extension/sourcecontrol:go_default_library",
         "//stovepipe/extension/sourcecontrol/fake:go_default_library",

service/stovepipe/server/main.go

@@ -55,6 +55,7 @@ import (
 	"github.com/uber/submitqueue/stovepipe/core/requestlog"
 	"github.com/uber/submitqueue/stovepipe/extension/buildrunner"
 	buildrunnerfake "github.com/uber/submitqueue/stovepipe/extension/buildrunner/fake"
+	projectresultnoop "github.com/uber/submitqueue/stovepipe/extension/projectresult/noop"
 	queueconfigdefault "github.com/uber/submitqueue/stovepipe/extension/queueconfig/default"
 	"github.com/uber/submitqueue/stovepipe/extension/sourcecontrol"
 	sourcecontrolfake "github.com/uber/submitqueue/stovepipe/extension/sourcecontrol/fake"
@@ -440,7 +441,7 @@ func registerPrimaryControllers(
 	}
 	count++
 
-	recordController := record.NewController(logger, scope, store, materializer, sourceControl, registry, stovepipemq.TopicKeyRecord, "stovepipe-record")
+	recordController := record.NewController(logger, scope, store, materializer, projectresultnoop.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

@@ -19,6 +19,7 @@ go_library(
         "//stovepipe/core/messagequeue:go_default_library",
         "//stovepipe/core/requestlog:go_default_library",
         "//stovepipe/entity:go_default_library",
+        "//stovepipe/extension/projectresult:go_default_library",
         "//stovepipe/extension/sourcecontrol:go_default_library",
         "//stovepipe/extension/storage:go_default_library",
         "@com_github_uber_go_tally//:go_default_library",
@@ -46,6 +47,9 @@ go_test(
         "//stovepipe/core/requestlog:go_default_library",
         "//stovepipe/core/requestlog/mock:go_default_library",
         "//stovepipe/entity:go_default_library",
+        "//stovepipe/extension/projectresult:go_default_library",
+        "//stovepipe/extension/projectresult/mock:go_default_library",
+        "//stovepipe/extension/projectresult/noop:go_default_library",
         "//stovepipe/extension/sourcecontrol:go_default_library",
         "//stovepipe/extension/sourcecontrol/mock:go_default_library",
         "//stovepipe/extension/storage:go_default_library",

stovepipe/controller/record/record.go

@@ -33,6 +33,7 @@ import (
 	"encoding/hex"
 	"errors"
 	"fmt"
+	"math"
 	"strconv"
 	"time"
 
@@ -47,6 +48,7 @@ import (
 	stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue"
 	"github.com/uber/submitqueue/stovepipe/core/requestlog"
 	"github.com/uber/submitqueue/stovepipe/entity"
+	"github.com/uber/submitqueue/stovepipe/extension/projectresult"
 	"github.com/uber/submitqueue/stovepipe/extension/sourcecontrol"
 	"github.com/uber/submitqueue/stovepipe/extension/storage"
 	"go.uber.org/zap"
@@ -56,14 +58,15 @@ import (
 // when that fact is green advances the queue's last-green bookmark and promotes
 // the commit. Implements consumer.Controller.
 type Controller struct {
-	logger        *zap.SugaredLogger
-	metricsScope  tally.Scope
-	stores        storage.Factory
-	materializer  requestlog.Materializer
-	sourceControl sourcecontrol.Factory
-	registry      consumer.TopicRegistry
-	topicKey      consumer.TopicKey
-	consumerGroup string
+	logger               *zap.SugaredLogger
+	metricsScope         tally.Scope
+	stores               storage.Factory
+	materializer         requestlog.Materializer
+	projectResultFactory projectresult.Factory
+	sourceControl        sourcecontrol.Factory
+	registry             consumer.TopicRegistry
+	topicKey             consumer.TopicKey
+	consumerGroup        string
 }
 
 // Verify Controller implements consumer.Controller interface at compile time.
@@ -73,8 +76,7 @@ var _ consumer.Controller = (*Controller)(nil)
 const _opName = "record"
 
 // wholeRepositoryProject is the project component of a fact covering the whole
-// repository rather than one project within it. Per-project facts need target-graph
-// attribution that this stage does not do, so every fact it writes is whole-repository.
+// repository rather than one project within it.
 const wholeRepositoryProject = ""
 
 // NewController creates a new record controller.
@@ -83,21 +85,23 @@ func NewController(
 	scope tally.Scope,
 	stores storage.Factory,
 	materializer requestlog.Materializer,
+	projectResultFactory projectresult.Factory,
 	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),
-		stores:        stores,
-		materializer:  materializer,
-		sourceControl: sourceControl,
-		registry:      registry,
-		topicKey:      topicKey,
-		consumerGroup: consumerGroup,
+		logger:               logger.Named(name),
+		metricsScope:         scope.SubScope(name),
+		stores:               stores,
+		materializer:         materializer,
+		projectResultFactory: projectResultFactory,
+		sourceControl:        sourceControl,
+		registry:             registry,
+		topicKey:             topicKey,
+		consumerGroup:        consumerGroup,
 	}
 }
 
@@ -150,6 +154,9 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 		if err := c.persistValidationFactRecordedLog(ctx, store, request, fact); err != nil {
 			return err
 		}
+		if err := c.recordProjectFacts(ctx, store, request); err != nil {
+			return err
+		}
 		if err := c.applyFactToDerivedCaches(ctx, store, request, fact, created); err != nil {
 			return err
 		}
@@ -176,6 +183,72 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 	}
 }
 
+func (c *Controller) recordProjectFacts(ctx context.Context, store storage.Storage, request entity.Request) error {
+	resolver, err := c.projectResultFactory.For(projectresult.Config{QueueName: request.Queue})
+	if err != nil {
+		return fmt.Errorf("failed to resolve project result resolver for queue %q: %w", request.Queue, err)
+	}
+	results, err := resolver.Resolve(ctx, request)
+	if err != nil {
+		return fmt.Errorf("failed to resolve project results for request %q: %w", request.ID, err)
+	}
+
+	if err := validateProjectResults(request, results); err != nil {
+		return err
+	}
+	for _, result := range results {
+		_, _, err := c.recordValidationFact(ctx, store.GetValidationFactStore(), entity.ValidationFact{
+			URI:       request.URI,
+			Project:   result.Project,
+			Degree:    result.Degree,
+			RequestID: request.ID,
+			CreatedAt: time.Now().UnixMilli(),
+		})
+		if err != nil {
+			return err
+		}
+	}
+	if len(results) == 0 {
+		return nil
+	}
+	return c.persistProjectFactsRecordedLog(ctx, store, request, len(results))
+}
+
+func validateProjectResults(request entity.Request, results []projectresult.Result) error {
+	seen := make(map[string]struct{}, len(results))
+	for _, result := range results {
+		if result.Project == "" {
+			return fmt.Errorf("project result for request %q has an empty project", request.ID)
+		}
+		if _, ok := seen[result.Project]; ok {
+			return fmt.Errorf("project result for request %q contains duplicate project %q", request.ID, result.Project)
+		}
+		seen[result.Project] = struct{}{}
+		if math.IsNaN(result.Degree) || result.Degree < entity.DegreeGreen || result.Degree > entity.DegreeBroken {
+			return fmt.Errorf("project result for request %q and project %q has invalid degree %v", request.ID, result.Project, result.Degree)
+		}
+	}
+	return nil
+}
+
+func (c *Controller) persistProjectFactsRecordedLog(
+	ctx context.Context,
+	store storage.Storage,
+	request entity.Request,
+	count int,
+) error {
+	log := requestlog.NewRequestEventLog(
+		request,
+		entity.RequestEventProjectFactsRecorded,
+		"all",
+		map[string]string{requestlog.MetadataKeyProjectFactCount: strconv.Itoa(count)},
+	)
+	if err := c.materializer.PersistLog(ctx, store, log); err != nil {
+		return fmt.Errorf("failed to record project facts for request %s: %w", request.ID, err)
+	}
+	return nil
+}
+
 func (c *Controller) persistValidationFactRecordedLog(
 	ctx context.Context,
 	store storage.Storage,
@@ -241,48 +314,49 @@ func (c *Controller) applyFactToDerivedCaches(
 // second return reports whether this call is the one that wrote the fact, which is
 // how a caller tells the original delivery from a redelivery.
 func (c *Controller) recordFact(ctx context.Context, store storage.Storage, request entity.Request) (entity.ValidationFact, bool, error) {
-	factStore := store.GetValidationFactStore()
-
-	fact := entity.ValidationFact{
+	return c.recordValidationFact(ctx, store.GetValidationFactStore(), entity.ValidationFact{
 		URI:       request.URI,
 		Project:   wholeRepositoryProject,
 		Degree:    degreeFor(request.State),
 		RequestID: request.ID,
 		CreatedAt: time.Now().UnixMilli(),
-	}
+	})
+}
+
+func (c *Controller) recordValidationFact(ctx context.Context, factStore storage.ValidationFactStore, fact entity.ValidationFact) (entity.ValidationFact, bool, error) {
 
 	err := factStore.Create(ctx, fact)
 	switch {
 	case err == nil:
 		metrics.NamedCounter(c.metricsScope, _opName, "fact_created", 1, metrics.TagsFromContext(ctx)...)
 		c.logger.Infow("recorded validation fact",
-			"queue", request.Queue,
-			"request_id", request.ID,
-			"uri", request.URI,
+			"request_id", fact.RequestID,
+			"uri", fact.URI,
+			"project", fact.Project,
 			"degree", fact.Degree,
 		)
 		return fact, true, nil
 
 	case errors.Is(err, storage.ErrAlreadyExists):
-		stored, getErr := factStore.Get(ctx, request.URI, wholeRepositoryProject)
+		stored, getErr := factStore.Get(ctx, fact.URI, fact.Project)
 		if getErr != nil {
 			metrics.NamedCounter(c.metricsScope, _opName, "storage_errors", 1, metrics.TagsFromContext(ctx)...)
-			return entity.ValidationFact{}, false, fmt.Errorf("failed to load the existing fact for uri %s: %w", request.URI, getErr)
+			return entity.ValidationFact{}, false, fmt.Errorf("failed to load the existing fact for uri %s and project %q: %w", fact.URI, fact.Project, getErr)
 		}
-		if stored.RequestID != request.ID {
+		if stored.RequestID != fact.RequestID {
 			// Two requests validating one URI would break the dedup ingest
 			// enforces, so this is a broken invariant rather than a race to
 			// resolve. Non-retryable: the stored fact is immutable.
 			metrics.NamedCounter(c.metricsScope, _opName, "invariant_errors", 1, metrics.TagsFromContext(ctx)...)
 			return entity.ValidationFact{}, false, fmt.Errorf(
-				"fact for uri %s is owned by request %s, not %s", request.URI, stored.RequestID, request.ID)
+				"fact for uri %s and project %q is owned by request %s, not %s", fact.URI, fact.Project, stored.RequestID, fact.RequestID)
 		}
 		metrics.NamedCounter(c.metricsScope, _opName, "fact_exists", 1, metrics.TagsFromContext(ctx)...)
 		return stored, false, nil
 
 	default:
 		metrics.NamedCounter(c.metricsScope, _opName, "storage_errors", 1, metrics.TagsFromContext(ctx)...)
-		return entity.ValidationFact{}, false, fmt.Errorf("failed to create the fact for uri %s: %w", request.URI, err)
+		return entity.ValidationFact{}, false, fmt.Errorf("failed to create the fact for uri %s and project %q: %w", fact.URI, fact.Project, err)
 	}
 }