feat(stovepipe): Add build analysis polling

stovepipe/controller/buildanalysis/BUILD.bazel

@@ -0,0 +1,43 @@
+load("@rules_go//go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["buildanalysis.go"],
+    importpath = "github.com/uber/submitqueue/stovepipe/controller/buildanalysis",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//platform/base/messagequeue:go_default_library",
+        "//platform/consumer:go_default_library",
+        "//platform/metrics:go_default_library",
+        "//stovepipe/core/loader:go_default_library",
+        "//stovepipe/core/messagequeue:go_default_library",
+        "//stovepipe/entity:go_default_library",
+        "//stovepipe/extension/buildissue:go_default_library",
+        "//stovepipe/extension/storage:go_default_library",
+        "@com_github_uber_go_tally//:go_default_library",
+        "@org_uber_go_zap//:go_default_library",
+    ],
+)
+
+go_test(
+    name = "go_default_test",
+    srcs = ["buildanalysis_test.go"],
+    embed = [":go_default_library"],
+    deps = [
+        "//platform/base/messagequeue: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/entity:go_default_library",
+        "//stovepipe/extension/buildissue:go_default_library",
+        "//stovepipe/extension/buildissue/mock:go_default_library",
+        "//stovepipe/extension/storage:go_default_library",
+        "//stovepipe/extension/storage/mock:go_default_library",
+        "@com_github_stretchr_testify//assert:go_default_library",
+        "@com_github_stretchr_testify//require:go_default_library",
+        "@com_github_uber_go_tally//:go_default_library",
+        "@org_uber_go_mock//gomock:go_default_library",
+        "@org_uber_go_zap//:go_default_library",
+    ],
+)

stovepipe/controller/buildanalysis/README.md

@@ -0,0 +1,21 @@
+# Build analysis
+
+## Observations
+
+The controller reloads the queue-scoped Build, invokes the queue's issue detector, and persists each returned observation individually. Observations already stored for the same build, task, and unit within a queue are not persisted again, so repeated polls and retries after partial writes are safe. Detector results accompanying an error are discarded.
+
+This controller only accumulates historical observations. An empty detection result neither deletes earlier observations nor establishes recovery. Open-issue projection, lifecycle history, and resolution are outside its scope.
+
+## Polling and completion
+
+Every delivery reloads the Build and its Request before detection. After a successful pass, a nonterminal Build holds the delivery for two minutes; a terminal Build completes the delivery after persistence. The status read before detection determines whether the pass holds: if the Build finishes during detection, another delivery is needed to analyze it after observing terminal status. More than one delivery may analyze the same terminal Build; observation persistence remains idempotent.
+
+If the Request is terminal but the loaded Build is nonterminal, analysis reloads the Build before deciding to stop. The Build may have finished between the initial Build read and the Request read; reloading avoids skipping its final analysis. If it is still nonterminal, analysis stops without detection or a hold. This can occur when the status-polling DLQ handler fails the Request without learning the actual build result, leaving the Build at its last observed status.
+
+Status polling and updates remain the responsibility of buildsignal. Analysis never changes Request state or releases build slots.
+
+## Errors
+
+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.

stovepipe/controller/buildanalysis/buildanalysis.go

@@ -0,0 +1,162 @@
+// 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 accumulates issue observations during and after builds.
+// Build status is read from storage; status polling remains separate.
+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"
+	"github.com/uber/submitqueue/stovepipe/core/loader"
+	stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue"
+	"github.com/uber/submitqueue/stovepipe/entity"
+	"github.com/uber/submitqueue/stovepipe/extension/buildissue"
+	"github.com/uber/submitqueue/stovepipe/extension/storage"
+	"go.uber.org/zap"
+)
+
+// PollDelayMs is the delay in milliseconds after a successful nonterminal pass.
+// Two minutes limits provider traffic during long-running builds.
+const PollDelayMs int64 = 120000
+
+// Controller accumulates observations while builds run and after completion.
+// Terminal analysis must finish persistence before its delivery completes.
+type Controller struct {
+	logger          *zap.SugaredLogger
+	metricsScope    tally.Scope
+	stores          storage.Factory
+	detectorFactory buildissue.DetectorFactory
+	topicKey        consumer.TopicKey
+	consumerGroup   string
+}
+
+func NewController(
+	logger *zap.SugaredLogger,
+	scope tally.Scope,
+	stores storage.Factory,
+	detectorFactory buildissue.DetectorFactory,
+	topicKey consumer.TopicKey,
+	consumerGroup string,
+) *Controller {
+	return &Controller{
+		logger:          logger.Named("buildanalysis_controller"),
+		metricsScope:    scope.SubScope("buildanalysis_controller"),
+		stores:          stores,
+		detectorFactory: detectorFactory,
+		topicKey:        topicKey,
+		consumerGroup:   consumerGroup,
+	}
+}
+
+// Process holds after a successful nonterminal pass. Errors are returned for
+// normal queue retry or DLQ handling; partial writes are safe to retry.
+func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) {
+	op := metrics.Begin(c.metricsScope, "process", metrics.LongLatencyBuckets, metrics.TagsFromContext(ctx)...)
+	defer func() { op.Complete(retErr) }()
+
+	msg := delivery.Message()
+	analysis := &stovepipemq.BuildAnalysis{}
+	if err := stovepipemq.Unmarshal(msg.Payload, analysis); err != nil {
+		return fmt.Errorf("failed to deserialize build analysis: %w", err)
+	}
+
+	if err := entityqueue.ValidatePayloadQueue(msg, analysis.GetQueueName()); err != nil {
+		return fmt.Errorf("invalid message identity: %w", err)
+	}
+
+	if analysis.GetId() == "" {
+		return fmt.Errorf("build analysis requires a build ID")
+	}
+
+	store, err := c.stores.For(storage.Config{QueueName: analysis.GetQueueName()})
+	if err != nil {
+		return fmt.Errorf("failed to resolve storage for queue %q: %w", analysis.GetQueueName(), err)
+	}
+
+	build, err := loader.ByID(ctx, analysis.GetId(), store.GetBuildStore().Get, "build")
+	if err != nil {
+		return err
+	}
+
+	request, err := loader.ByID(ctx, build.RequestID, store.GetRequestStore().Get, "request")
+	if err != nil {
+		return err
+	}
+
+	if request.Queue != analysis.GetQueueName() {
+		return fmt.Errorf("request queue %q does not match analysis queue %q", request.Queue, analysis.GetQueueName())
+	}
+
+	// The Build may finish between the initial Build read and the Request read.
+	// Recheck its status so stopping here does not skip final analysis.
+	if !build.Status.IsTerminal() && request.State.IsTerminal() {
+		build, err = loader.ByID(ctx, analysis.GetId(), store.GetBuildStore().Get, "build")
+		if err != nil {
+			return err
+		}
+
+		if !build.Status.IsTerminal() {
+			c.logger.Warnw("stopping analysis for a terminal request with nonterminal build",
+				"queue", request.Queue, "build_id", build.ID, "request_id", request.ID, "request_state", string(request.State))
+			return nil
+		}
+	}
+
+	detector, err := c.detectorFactory.For(buildissue.DetectorConfig{QueueName: analysis.GetQueueName()})
+	if err != nil {
+		return fmt.Errorf("failed to resolve issue detector for queue %q: %w", analysis.GetQueueName(), err)
+	}
+
+	issues, err := detector.Detect(ctx, entity.BuildID{ID: build.ID})
+	if err != nil {
+		return fmt.Errorf("failed to detect issues for build %q: %w", build.ID, err)
+	}
+
+	issueStore := store.GetBuildIssueStore()
+	for _, issue := range issues {
+		if err := issueStore.Create(ctx, build.ID, issue); err != nil && !errors.Is(err, storage.ErrAlreadyExists) {
+			return fmt.Errorf("failed to persist build issue build=%q task=%q unit=%q: %w", build.ID, issue.TaskID, issue.UnitID, err)
+		}
+	}
+
+	// A nonterminal status read before detection requires another pass,
+	// even if the build finishes while detection is running.
+	if !build.Status.IsTerminal() {
+		delivery.Hold(PollDelayMs)
+	}
+
+	c.logger.Debugw("build analysis pass complete",
+		"queue", analysis.GetQueueName(),
+		"build_id", build.ID,
+		"status", string(build.Status),
+		"observed_issue_count", len(issues),
+	)
+
+	return nil
+}
+
+func (c *Controller) Name() string { return "buildanalysis" }
+
+func (c *Controller) TopicKey() consumer.TopicKey { return c.topicKey }
+
+func (c *Controller) ConsumerGroup() string { return c.consumerGroup }
+
+var _ consumer.Controller = (*Controller)(nil)