feat(stovepipe): Record incomplete build analysis

doc/rfc/stovepipe/request-log.md

@@ -2,13 +2,14 @@
 
 ## 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 asynchronous milestones needed to explain those transitions, the public verdict, and abandoned terminal-stage work:
+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 stage work:
 
 - `build_triggered`;
 - `build_finished`;
 - `validation_fact_recorded`;
 - `project_facts_recorded`;
-- `record_abandoned`.
+- `record_abandoned`;
+- `build_analysis_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.
 
@@ -126,6 +127,7 @@ Immutable Request context such as URI, build strategy, base URI, and terminal bu
 | `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_analysis_abandoned` | Build analysis could not complete; issue observations may be incomplete. This is not a validation verdict. | Build ID metadata and 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. Individual project facts remain outside the vocabulary; one batch-completion event summarizes their durable recording.
 
@@ -214,6 +216,7 @@ Request creation, Build changes, and fact creation use the same source-write, lo
 | 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. |
+| 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.

stovepipe/controller/buildanalysis/BUILD.bazel

@@ -2,7 +2,10 @@ load("@rules_go//go:def.bzl", "go_library", "go_test")
 
 go_library(
     name = "go_default_library",
-    srcs = ["buildanalysis.go"],
+    srcs = [
+        "buildanalysis.go",
+        "dlq.go",
+    ],
     importpath = "github.com/uber/submitqueue/stovepipe/controller/buildanalysis",
     visibility = ["//visibility:public"],
     deps = [
@@ -11,6 +14,7 @@ go_library(
         "//platform/metrics:go_default_library",
         "//stovepipe/core/loader:go_default_library",
         "//stovepipe/core/messagequeue:go_default_library",
+        "//stovepipe/core/requestlog:go_default_library",
         "//stovepipe/entity:go_default_library",
         "//stovepipe/extension/buildissue:go_default_library",
         "//stovepipe/extension/storage:go_default_library",
@@ -21,14 +25,21 @@ go_library(
 
 go_test(
     name = "go_default_test",
-    srcs = ["buildanalysis_test.go"],
+    srcs = [
+        "buildanalysis_test.go",
+        "dlq_test.go",
+    ],
     embed = [":go_default_library"],
     deps = [
+        "//platform/base/failure:go_default_library",
         "//platform/base/messagequeue:go_default_library",
+        "//platform/consumer:go_default_library",
         "//platform/consumer/mock:go_default_library",
         "//platform/errs:go_default_library",
         "//platform/metrics:go_default_library",
         "//stovepipe/core/messagequeue:go_default_library",
+        "//stovepipe/core/requestlog:go_default_library",
+        "//stovepipe/core/requestlog/mock:go_default_library",
         "//stovepipe/entity:go_default_library",
         "//stovepipe/extension/buildissue:go_default_library",
         "//stovepipe/extension/buildissue/mock:go_default_library",

stovepipe/controller/buildanalysis/README.md

@@ -18,4 +18,12 @@ Status polling and updates remain the responsibility of buildsignal. Analysis ne
 
 Detection, persistence, configuration, state-loading, and message-validation errors are returned to the consumer for its normal retry or DLQ policy, whether the Build is running or terminal. Analysis errors do not use the two-minute polling hold. The consumer classifies errors and applies its configured retry limits and backoff; a final analysis can still fail and dead-letter.
 
-The message and controller are defined without runtime producers or consumer registration. Host integration and detector implementations are separate work.
+## DLQ recovery
+
+The recovery controller records a `build_analysis_abandoned` event in the Request's history, with the build ID, to indicate that analysis could not complete and issue observations may be incomplete. It does not rerun detection, delete earlier observations, change Request or Build state, or release a build slot. Original failure details are logged rather than included in the immutable history event.
+
+History identity is stable per build, so repeated recovery deliveries and separate analysis notices for the same build retain one event. Different builds retain different events, even for the same Request. A later successful analysis does not erase the historical occurrence.
+
+State-read and history-write errors are returned for retry. Hosts must run recovery with the DLQ consumer's always-retryable processor and unlimited-retry subscription; only recovery work is retried, not analysis. Unusable payloads, invalid or unresolvable queues, missing Build or Request records, and queue mismatches are logged and acknowledged rather than retried indefinitely.
+
+The controllers are defined without runtime producers or consumer registration. Host integration and detector implementations are separate work.

stovepipe/controller/buildanalysis/dlq.go

@@ -0,0 +1,185 @@
+// 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 buildanalysis
+
+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 = "buildanalysis_dlq"
+
+// DLQController records incomplete analysis without rerunning detection or
+// changing the validation outcome.
+type DLQController struct {
+	logger        *zap.SugaredLogger
+	metricsScope  tally.Scope
+	stores        storage.Factory
+	materializer  requestlog.Materializer
+	topicKey      consumer.TopicKey
+	consumerGroup string
+}
+
+// NewDLQController creates a controller for incomplete build analysis.
+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 incomplete-analysis history and acknowledges unusable messages.
+// State-read and history-write failures are returned for DLQ recovery retry.
+func (c *DLQController) Process(ctx context.Context, delivery consumer.Delivery) error {
+	msg := delivery.Message()
+	analysis := &stovepipemq.BuildAnalysis{}
+	if err := stovepipemq.Unmarshal(msg.Payload, analysis); err != nil {
+		metrics.NamedCounter(c.metricsScope, _dlqOpName, "deserialize_errors", 1, metrics.TagsFromContext(ctx)...)
+		c.logger.Errorw("discarding malformed analysis dlq message",
+			"message_id", msg.ID,
+			"error", err,
+		)
+		return nil
+	}
+
+	if err := entityqueue.ValidatePayloadQueue(msg, analysis.GetQueueName()); err != nil {
+		metrics.NamedCounter(c.metricsScope, _dlqOpName, "queue_identity_errors", 1, metrics.TagsFromContext(ctx)...)
+		c.logger.Errorw("discarding analysis dlq message with invalid queue identity",
+			"message_id", msg.ID,
+			"build_id", analysis.GetId(),
+			"error", err,
+		)
+		return nil
+	}
+
+	if analysis.GetId() == "" {
+		metrics.NamedCounter(c.metricsScope, _dlqOpName, "empty_id_errors", 1, metrics.TagsFromContext(ctx)...)
+		c.logger.Errorw("discarding analysis dlq message with empty build id",
+			"message_id", msg.ID,
+			"queue", analysis.GetQueueName(),
+		)
+		return nil
+	}
+
+	store, err := c.stores.For(storage.Config{QueueName: analysis.GetQueueName()})
+	if err != nil {
+		metrics.NamedCounter(c.metricsScope, _dlqOpName, "storage_resolve_errors", 1, metrics.TagsFromContext(ctx)...)
+		c.logger.Errorw("discarding analysis dlq message for unresolvable queue",
+			"message_id", msg.ID,
+			"build_id", analysis.GetId(),
+			"queue", analysis.GetQueueName(),
+			"error", err,
+		)
+		return nil
+	}
+
+	build, err := store.GetBuildStore().Get(ctx, analysis.GetId())
+	if err != nil {
+		if errors.Is(err, storage.ErrNotFound) {
+			metrics.NamedCounter(c.metricsScope, _dlqOpName, "build_not_found", 1, metrics.TagsFromContext(ctx)...)
+			c.logger.Errorw("discarding analysis dlq message for missing build",
+				"message_id", msg.ID, "build_id", analysis.GetId(), "queue", analysis.GetQueueName())
+			return nil
+		}
+
+		return fmt.Errorf("failed to load build %s for analysis dlq: %w", analysis.GetId(), err)
+	}
+
+	if build.RequestID == "" {
+		metrics.NamedCounter(c.metricsScope, _dlqOpName, "empty_request_id", 1, metrics.TagsFromContext(ctx)...)
+		c.logger.Errorw("discarding analysis dlq message for build without a request",
+			"message_id", msg.ID, "build_id", analysis.GetId())
+		return nil
+	}
+
+	request, err := store.GetRequestStore().Get(ctx, build.RequestID)
+	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 analysis dlq message for missing request",
+				"message_id", msg.ID, "build_id", analysis.GetId(), "request_id", build.RequestID)
+			return nil
+		}
+
+		return fmt.Errorf("failed to load request %s for analysis dlq: %w", build.RequestID, err)
+	}
+
+	if request.Queue != analysis.GetQueueName() {
+		metrics.NamedCounter(c.metricsScope, _dlqOpName, "queue_identity_errors", 1, metrics.TagsFromContext(ctx)...)
+		c.logger.Errorw("discarding analysis dlq message for request in another queue",
+			"message_id", msg.ID, "request_id", request.ID, "queue", request.Queue,
+			"message_queue", analysis.GetQueueName())
+		return nil
+	}
+
+	log := requestlog.NewRequestEventLog(request, entity.RequestEventBuildAnalysisAbandoned, analysis.GetId(),
+		map[string]string{requestlog.MetadataKeyBuildID: analysis.GetId()})
+	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 incomplete build analysis for request %s: %w", request.ID, err)
+	}
+
+	originalFailure, hasFailure := delivery.Failure()
+	metrics.NamedCounter(c.metricsScope, _dlqOpName, "analysis_abandoned", 1, metrics.TagsFromContext(ctx)...)
+	fields := []any{
+		"message_id", msg.ID,
+		"request_id", request.ID,
+		"build_id", analysis.GetId(),
+		"queue", request.Queue,
+		"request_state", request.State,
+		"history_event", entity.RequestEventBuildAnalysisAbandoned,
+		"attempt", delivery.Attempt(),
+	}
+	if hasFailure {
+		fields = append(fields,
+			"failure", originalFailure.Message,
+			"failure_subjects", originalFailure.Subjects,
+			"failure_detail", originalFailure.Detail,
+		)
+	}
+	c.logger.Errorw("build analysis could not complete; issue observations may be incomplete", fields...)
+	return nil
+}
+
+func (c *DLQController) Name() string { return string(c.topicKey) }
+
+func (c *DLQController) TopicKey() consumer.TopicKey { return c.topicKey }
+
+func (c *DLQController) ConsumerGroup() string { return c.consumerGroup }
+
+var _ consumer.Controller = (*DLQController)(nil)