fix(stovepipe): record broken facts from record DLQ

doc/rfc/stovepipe/request-log.md

@@ -217,13 +217,13 @@ The `project_facts_recorded` event also projects its retained timestamp into `Re
 | 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`; 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. |
+| Record DLQ | Retain `record_abandoned`; for failed Requests, create or verify the repository fact and retain `validation_fact_recorded` before publishing its hook. | Existing facts remain authoritative; stable log and hook IDs make replay idempotent without resolving projects or replaying promotion. |
 | Build analysis DLQ | Retain `build_analysis_abandoned` without rerunning detection or changing validation state. | One stable event per build deduplicates recovery deliveries; only state reads and history retention retry. |
 | 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. |
 
 Build running and unchanged polls create no entry. A failed runner trigger that creates no Build creates no event. Cancelled and superseded requests create no validation fact.
 
-No controller infers an outcome reason from `RequestStateFailed` alone, invents a broken fact, or records DLQ as the domain reason.
+No controller infers an outcome reason from `RequestStateFailed` alone or records DLQ as the domain reason. Record DLQ uses that persisted state to recover a missing broken repository fact; it does not infer project verdicts.
 
 ### Repair trigger requirement
 

doc/rfc/stovepipe/steps/record.md

@@ -207,7 +207,7 @@ Ordering is per-subject only and the subject is the Request, so events for *diff
 
 Absence of an event is not a signal. A Request abandoned before any build went terminal never reaches this stage, and a superseded one publishes nothing, so a consumer waiting for one event per ingested commit waits forever on those. Gating keeps treating "no recorded fact" as not green. The converse holds too — an event is not proof the code was tested, since a fail-closed Request can produce a broken fact without a build having failed.
 
-Hooks here must be idempotent on `id`, as everywhere. "Fire-and-forget" describes downstream consumption, not the publish: `record` never waits for a hook, but a failed *publish* fails the delivery. Per `[platform/errs](../../../../platform/errs/README.md)` rule 4 it is not wrapped retryable just because replaying it is convenient, so it dead-letters. Once primary retries are exhausted, `record_dlq` stops trying to complete the original work: it reloads the already-terminal Request, retains `record_abandoned` history, logs and counts the abandonment, and acknowledges. It does not create a fact, advance the bookmark, promote, or publish a hook. Any partial durable effects already written remain authoritative; absence of a fact remains fail-closed, and a later request or operator action may repair external state. Promotion is one example of this case: an outbound call rejected for persistent permissions is not attempted again from the DLQ, and its diagnostic remains in structured logs.
+Hooks here must be idempotent on `id`, as everywhere. A failed publish fails the primary delivery and follows the normal classifier and retry policy. After dead-lettering, `record_dlq` retains `record_abandoned` and ensures a failed Request has a repository fact and a `validation.repository.recorded` notification. Existing immutable facts remain authoritative. See [DLQ and fail-closed behavior](#dlq-and-fail-closed-behavior) for the recovery boundary.
 
 ## Request lifecycle
 
@@ -280,9 +280,11 @@ Every primary subscription comes from `DefaultSubscriptionConfig`, which enables
 
 Two different things put a message there, and only one is a poison payload. A delivery that fails with its retry budget spent is dead-lettered by the nack itself, carrying the reason it actually failed. A delivery that never reaches a nack, because it crashed or because its **ack failed** and the visibility timeout redelivered it, is dead-lettered by the poll loop once `retry_count` reaches `MaxAttempts` (3 by default), without the controller running on that final attempt and with only a generic reason recorded. So a missing reconciler exposes more than malformed messages: a fact can be lost to a storage failure that would have succeeded on a later retry, or to an ack that never landed even though the write did.
 
-Gating stays safe, because everything this stage can lose reads as not-green: a Request with no fact is indistinguishable from one not yet validated. What is lost is the *fact*. A green build whose fact write permanently failed leaves the URI looking unvalidated, while its successful-build baseline remains available to the next head. A lost notification joins that list, and unlike the fact it gets no second chance from a later commit.
+The reconciler retains a stable `record_abandoned` event. For a Request already in `failed`, it creates a `DegreeBroken` whole-repository fact if absent, retains `validation_fact_recorded` history, and publishes `validation.repository.recorded` after both writes succeed. An existing fact owned by the same Request is adopted unchanged, including its degree and timestamp; a fact owned by another Request is an invariant error. Replays repair missing history and repeat the notification with the same event ID.
 
-The reconciler does not re-run the stage. It loads the Request only to identify existing durable state and retain a stable `record_abandoned` event. The DLQ consumer uses `errs.AlwaysRetryableProcessor`, so a transient Request read or history persistence failure keeps retrying until that observable abandonment is durable. A malformed payload, invalid or unresolvable queue identity, or missing Request cannot be repaired by redelivery; those cases are logged, counted, and acknowledged so poison cannot occupy the DLQ indefinitely. No DLQ path creates a fact, advances a bookmark, invokes source control, or publishes a hook.
+This fallback does not resolve project attribution or mark project recording complete. A failed build establishes a repository failure even when no individual project verdict is known. Build-issue analysis and detectors remain independent. The reconciler does not change the Request state, advance the bookmark, promote, or release build slots. Other Request states retain abandonment history only, so a succeeded build whose fact could not be recorded still appears unvalidated.
+
+The DLQ consumer uses `errs.AlwaysRetryableProcessor`: Request reads, fact writes, history persistence, and notification publication retry until recovery succeeds. Malformed payloads, invalid or unresolvable queue identities, queue mismatches, and missing Requests are logged, counted, and acknowledged.
 
 ## Future Items
 

service/stovepipe/server/main.go

@@ -262,7 +262,7 @@ func run() error {
 	if err != nil {
 		return err
 	}
-	dlqCount, err := registerDLQControllers(dlqConsumer, logger.Sugar(), scope, storageFty, materializer)
+	dlqCount, err := registerDLQControllers(dlqConsumer, logger.Sugar(), scope, storageFty, materializer, registry)
 	if err != nil {
 		return err
 	}
@@ -438,6 +438,7 @@ func registerDLQControllers(
 	scope tally.Scope,
 	store storage.Factory,
 	materializer requestlog.Materializer,
+	registry consumer.TopicRegistry,
 ) (int, error) {
 	var count int
 
@@ -465,7 +466,7 @@ func registerDLQControllers(
 	}
 	count++
 
-	recordDLQController := record.NewDLQController(logger, scope, store, materializer, dlq.TopicKey(stovepipemq.TopicKeyRecord), "stovepipe-record-dlq")
+	recordDLQController := record.NewDLQController(logger, scope, store, materializer, registry, dlq.TopicKey(stovepipemq.TopicKeyRecord), "stovepipe-record-dlq")
 	if err := c.Register(recordDLQController); err != nil {
 		return count, fmt.Errorf("failed to register record dlq controller: %w", err)
 	}

service/stovepipe/server/main_test.go

@@ -64,7 +64,7 @@ func registeredControllers(t *testing.T) (consumer.TopicRegistry, []consumer.Con
 		fakeSourceControlFactory{}, fakeBuildRunnerFactory{}, hookResolver{}, buildissuenoop.New())
 	require.NoError(t, err)
 
-	_, err = registerDLQControllers(deadLetter, logger, tally.NoopScope, store, requestlog.NewMaterializer(tally.NoopScope))
+	_, err = registerDLQControllers(deadLetter, logger, tally.NoopScope, store, requestlog.NewMaterializer(tally.NoopScope), registry)
 	require.NoError(t, err)
 
 	return registry, primary.controllers, deadLetter.controllers

stovepipe/controller/record/BUILD.bazel

@@ -4,6 +4,7 @@ go_library(
     name = "go_default_library",
     srcs = [
         "dlq.go",
+        "facts.go",
         "record.go",
     ],
     importpath = "github.com/uber/submitqueue/stovepipe/controller/record",

stovepipe/controller/record/dlq.go

@@ -23,6 +23,7 @@ import (
 	entityqueue "github.com/uber/submitqueue/platform/base/messagequeue"
 	"github.com/uber/submitqueue/platform/consumer"
 	"github.com/uber/submitqueue/platform/metrics"
+	"github.com/uber/submitqueue/stovepipe/core/hookevent"
 	stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue"
 	"github.com/uber/submitqueue/stovepipe/core/requestlog"
 	"github.com/uber/submitqueue/stovepipe/entity"
@@ -32,13 +33,14 @@ import (
 
 const _dlqOpName = "record_dlq"
 
-// DLQController records that record-stage work was abandoned without replaying
-// any of the stage's durable writes or outbound calls.
+// DLQController retains abandoned record work and ensures failed requests have
+// a repository fact without requiring project attribution.
 type DLQController struct {
 	logger        *zap.SugaredLogger
 	metricsScope  tally.Scope
-	stores        storage.Factory
 	materializer  requestlog.Materializer
+	registry      consumer.TopicRegistry
+	stores        storage.Factory
 	topicKey      consumer.TopicKey
 	consumerGroup string
 }
@@ -51,23 +53,25 @@ func NewDLQController(
 	scope tally.Scope,
 	stores storage.Factory,
 	materializer requestlog.Materializer,
+	registry consumer.TopicRegistry,
 	topicKey consumer.TopicKey,
 	consumerGroup string,
 ) *DLQController {
 	name := string(topicKey) + "_controller"
 	return &DLQController{
 		logger:        logger.Named(name),
 		metricsScope:  scope.SubScope(name),
-		stores:        stores,
 		materializer:  materializer,
+		registry:      registry,
+		stores:        stores,
 		topicKey:      topicKey,
 		consumerGroup: consumerGroup,
 	}
 }
 
 // Process retains an observable abandonment occurrence for record work that the
 // primary consumer could not finish. Deterministic poison is acknowledged;
-// failures reading durable state or retaining history are returned for retry.
+// failures recording facts, retaining history, or publishing are retried.
 func (c *DLQController) Process(ctx context.Context, delivery consumer.Delivery) error {
 	msg := delivery.Message()
 	rec := &stovepipemq.Record{}
@@ -123,13 +127,33 @@ func (c *DLQController) Process(ctx context.Context, delivery consumer.Delivery)
 		metrics.NamedCounter(c.metricsScope, _dlqOpName, "request_store_errors", 1, metrics.TagsFromContext(ctx)...)
 		return fmt.Errorf("failed to load request %s for record dlq: %w", rec.GetId(), err)
 	}
+	if rec.GetQueueName() != "" && rec.GetQueueName() != request.Queue {
+		metrics.NamedCounter(c.metricsScope, _dlqOpName, "queue_mismatch", 1, metrics.TagsFromContext(ctx)...)
+		c.logger.Errorw("discarding record dlq message with mismatched request queue",
+			"message_id", msg.ID, "request_id", request.ID,
+			"payload_queue", rec.GetQueueName(), "request_queue", request.Queue)
+		return nil
+	}
 	originalFailure, hasFailure := delivery.Failure()
 	log := requestlog.NewRequestEventLog(request, entity.RequestEventRecordAbandoned, "repository", nil)
 	if err := c.materializer.PersistLog(ctx, store, log); err != nil {
 		metrics.NamedCounter(c.metricsScope, _dlqOpName, "history_errors", 1, metrics.TagsFromContext(ctx)...)
 		return fmt.Errorf("failed to retain abandoned record work for request %s: %w", request.ID, err)
 	}
 
+	if request.State == entity.RequestStateFailed {
+		fact, _, err := recordFact(ctx, c.logger, c.metricsScope, store, request, nil)
+		if err != nil {
+			return err
+		}
+		if err := persistValidationFactRecordedLog(ctx, c.materializer, store, request, fact); err != nil {
+			return err
+		}
+		if err := publishHookEvent(ctx, c.logger, c.metricsScope, c.registry, request, hookevent.NewValidationRepositoryRecorded(request)); err != nil {
+			return err
+		}
+	}
+
 	metrics.NamedCounter(c.metricsScope, _dlqOpName, "requests_abandoned", 1, metrics.TagsFromContext(ctx)...)
 	fields := []any{
 		"message_id", msg.ID,