fix(stovepipe): separate record dlq handler

doc/rfc/stovepipe/request-log.md

@@ -2,11 +2,12 @@
 
 ## Summary
 
-Stovepipe retains an append-only request log for each validation request. Its internal `RequestLog` is the counterpart of SubmitQueue's `RequestLog`: both retain request status changes and explanatory lifecycle events, while Stovepipe calls its materializer directly instead of sending records through a cross-service log topic. The public API presents these records as request history. The log records every durable `Request.State` transition plus three asynchronous milestones needed to explain those transitions and the public verdict:
+Stovepipe retains an append-only request log for each validation request. Its internal `RequestLog` is the counterpart of SubmitQueue's `RequestLog`: both retain request status changes and explanatory lifecycle events, while Stovepipe calls its materializer directly instead of sending records through a cross-service log topic. The public API presents these records as request history. The log records every durable `Request.State` transition plus asynchronous milestones needed to explain those transitions, the public verdict, and abandoned terminal-stage work:
 
 - `build_triggered`;
 - `build_finished`;
-- `validation_fact_recorded`.
+- `validation_fact_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.
 
@@ -122,6 +123,7 @@ 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 |
+| `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.
 
@@ -208,6 +210,7 @@ Request creation, Build changes, and fact creation use the same source-write, lo
 | 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 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. |
 
 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.

doc/rfc/stovepipe/steps/record.md

@@ -202,7 +202,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. The `record_dlq` consumer is this same controller on the dead-letter topic, so it re-runs this identical idempotent algorithm and the republish is its own recovery path: the fact is already durable, and only the notification was outstanding.
+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.
 
 ## Request lifecycle
 
@@ -219,7 +219,7 @@ Phase 2 broadens "complete" to "all planned facts recorded", which needs a marke
 
 There is no `Update`. The first fact written for an identity is the permanent answer, and a caller that needs to know whether it won the race reads `ErrAlreadyExists` and then loads the winner.
 
-The topic key, the message, and the consumer all exist. The DLQ consumer does not (see [DLQ and fail-closed behavior](#dlq-and-fail-closed-behavior)).
+The topic key, message, primary consumer, and DLQ consumer all exist.
 
 
 | Topic key | Message                  | Producer      | Consumer | Partition key | Message id |
@@ -271,13 +271,13 @@ Every effect is recognize-and-skip, so a redelivery after a complete run re-runs
 
 ## DLQ and fail-closed behavior
 
-**Neither** `record_dlq` **nor** `build_dlq` **has a consumer today, and both topics are already receiving messages.** Every primary subscription comes from `DefaultSubscriptionConfig`, which enables dead-lettering with the `_dlq` suffix, so a message that is rejected outright *or* runs out of retries moves to its stage's dead-letter topic. The wiring registers only `process_dlq` and `buildsignal_dlq`, so messages pile up unread on the other two.
+Every primary subscription comes from `DefaultSubscriptionConfig`, which enables dead-lettering with the `_dlq` suffix, so a message that is rejected outright or runs out of retries moves to its stage's dead-letter topic. Stovepipe registers a reconciler for every pipeline DLQ, including `record_dlq`.
 
 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, which costs the queue an incremental baseline and forces a full build at the next head. A lost notification joins that list, and unlike the fact it gets no second chance from a later commit.
 
-This is the same failure shape [buildsignal.md](buildsignal.md#what-it-costs-when-a-backend-does-not-classify-status-errors) describes for a deployment that registers primary consumers without their reconciler. When the reconciler is built it should re-run this same idempotent algorithm from the request id, under `errs.AlwaysRetryableProcessor`: write and publish the immutable fact as usual if the Request carries a build outcome, keep retrying if Request storage is temporarily unavailable, and treat a malformed payload or a permanently missing Request as poison, which needs an operational alert rather than more retries.
+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.
 
 ## Future Items
 

service/stovepipe/server/main.go

@@ -296,7 +296,7 @@ func run() error {
 	if err != nil {
 		return err
 	}
-	dlqCount, err := registerDLQControllers(dlqConsumer, logger.Sugar(), scope, storageFty, materializer, registry, sourceControl)
+	dlqCount, err := registerDLQControllers(dlqConsumer, logger.Sugar(), scope, storageFty, materializer)
 	if err != nil {
 		return err
 	}
@@ -462,8 +462,6 @@ func registerDLQControllers(
 	scope tally.Scope,
 	store storage.Factory,
 	materializer requestlog.Materializer,
-	registry consumer.TopicRegistry,
-	sourceControl sourcecontrol.Factory,
 ) (int, error) {
 	var count int
 
@@ -485,7 +483,7 @@ func registerDLQControllers(
 	}
 	count++
 
-	recordDLQController := record.NewController(logger, scope, store, materializer, sourceControl, registry, dlq.TopicKey(stovepipemq.TopicKeyRecord), "stovepipe-record-dlq")
+	recordDLQController := record.NewDLQController(logger, scope, store, materializer, 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

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

stovepipe/controller/record/BUILD.bazel

@@ -2,7 +2,10 @@ load("@rules_go//go:def.bzl", "go_library", "go_test")
 
 go_library(
     name = "go_default_library",
-    srcs = ["record.go"],
+    srcs = [
+        "dlq.go",
+        "record.go",
+    ],
     importpath = "github.com/uber/submitqueue/stovepipe/controller/record",
     visibility = ["//visibility:public"],
     deps = [
@@ -25,10 +28,14 @@ go_library(
 
 go_test(
     name = "go_default_test",
-    srcs = ["record_test.go"],
+    srcs = [
+        "dlq_test.go",
+        "record_test.go",
+    ],
     embed = [":go_default_library"],
     deps = [
         "//api/base/hook:go_default_library",
+        "//platform/base/failure:go_default_library",
         "//platform/base/messagequeue:go_default_library",
         "//platform/consumer:go_default_library",
         "//platform/consumer/mock:go_default_library",

stovepipe/controller/record/dlq.go

@@ -0,0 +1,160 @@
+// 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 record
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"github.com/uber-go/tally"
+	entityqueue "github.com/uber/submitqueue/platform/base/messagequeue"
+	"github.com/uber/submitqueue/platform/consumer"
+	"github.com/uber/submitqueue/platform/metrics"
+	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/storage"
+	"go.uber.org/zap"
+)
+
+const _dlqOpName = "record_dlq"
+
+// DLQController records that record-stage work was abandoned without replaying
+// any of the stage's durable writes or outbound calls.
+type DLQController struct {
+	logger        *zap.SugaredLogger
+	metricsScope  tally.Scope
+	stores        storage.Factory
+	materializer  requestlog.Materializer
+	topicKey      consumer.TopicKey
+	consumerGroup string
+}
+
+var _ consumer.Controller = (*DLQController)(nil)
+
+// NewDLQController creates a controller for abandoned record work.
+func NewDLQController(
+	logger *zap.SugaredLogger,
+	scope tally.Scope,
+	stores storage.Factory,
+	materializer requestlog.Materializer,
+	topicKey consumer.TopicKey,
+	consumerGroup string,
+) *DLQController {
+	name := string(topicKey) + "_controller"
+	return &DLQController{
+		logger:        logger.Named(name),
+		metricsScope:  scope.SubScope(name),
+		stores:        stores,
+		materializer:  materializer,
+		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.
+func (c *DLQController) Process(ctx context.Context, delivery consumer.Delivery) error {
+	msg := delivery.Message()
+	rec := &stovepipemq.Record{}
+	if err := stovepipemq.Unmarshal(msg.Payload, rec); err != nil {
+		metrics.NamedCounter(c.metricsScope, _dlqOpName, "deserialize_errors", 1, metrics.TagsFromContext(ctx)...)
+		c.logger.Errorw("discarding malformed record dlq message",
+			"message_id", msg.ID,
+			"error", err,
+		)
+		return nil
+	}
+	if err := entityqueue.ValidatePayloadQueue(msg, rec.GetQueueName()); err != nil {
+		metrics.NamedCounter(c.metricsScope, _dlqOpName, "queue_identity_errors", 1, metrics.TagsFromContext(ctx)...)
+		c.logger.Errorw("discarding record dlq message with invalid queue identity",
+			"message_id", msg.ID,
+			"request_id", rec.GetId(),
+			"error", err,
+		)
+		return nil
+	}
+	if rec.GetId() == "" {
+		metrics.NamedCounter(c.metricsScope, _dlqOpName, "empty_id_errors", 1, metrics.TagsFromContext(ctx)...)
+		c.logger.Errorw("discarding record dlq message with empty request id",
+			"message_id", msg.ID,
+			"queue", rec.GetQueueName(),
+		)
+		return nil
+	}
+
+	store, err := c.stores.For(storage.Config{QueueName: rec.GetQueueName()})
+	if err != nil {
+		metrics.NamedCounter(c.metricsScope, _dlqOpName, "storage_resolve_errors", 1, metrics.TagsFromContext(ctx)...)
+		c.logger.Errorw("discarding record dlq message for unresolvable queue",
+			"message_id", msg.ID,
+			"request_id", rec.GetId(),
+			"queue", rec.GetQueueName(),
+			"error", err,
+		)
+		return nil
+	}
+
+	request, err := store.GetRequestStore().Get(ctx, rec.GetId())
+	if err != nil {
+		if errors.Is(err, storage.ErrNotFound) {
+			metrics.NamedCounter(c.metricsScope, _dlqOpName, "request_not_found", 1, metrics.TagsFromContext(ctx)...)
+			c.logger.Errorw("discarding record dlq message for missing request",
+				"message_id", msg.ID,
+				"request_id", rec.GetId(),
+				"queue", rec.GetQueueName(),
+			)
+			return nil
+		}
+		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)
+	}
+	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)
+	}
+
+	metrics.NamedCounter(c.metricsScope, _dlqOpName, "requests_abandoned", 1, metrics.TagsFromContext(ctx)...)
+	fields := []any{
+		"message_id", msg.ID,
+		"request_id", request.ID,
+		"queue", request.Queue,
+		"request_state", request.State,
+		"history_event", entity.RequestEventRecordAbandoned,
+		"attempt", delivery.Attempt(),
+	}
+	if hasFailure {
+		fields = append(fields,
+			"failure", originalFailure.Message,
+			"failure_subjects", originalFailure.Subjects,
+			"failure_detail", originalFailure.Detail,
+		)
+	}
+	c.logger.Errorw("abandoned record work after retaining failure history", fields...)
+	return nil
+}
+
+// Name returns the controller's name.
+func (c *DLQController) Name() string { return string(c.topicKey) }
+
+// TopicKey returns the controller's topic key.
+func (c *DLQController) TopicKey() consumer.TopicKey { return c.topicKey }
+
+// ConsumerGroup returns the controller's consumer group.
+func (c *DLQController) ConsumerGroup() string { return c.consumerGroup }