feat(stovepipe): Wire build analysis pipeline

Author: prathshenoy

doc/rfc/stovepipe/steps/buildanalysis.md

@@ -0,0 +1,11 @@
+# Build issue observation
+
+Build analysis runs alongside status polling and recording. Hosts supply a queue-resolved detector factory. The reference service uses a no-op detector so the analysis stage is always wired without inspecting provider output; deployments replace the factory with their integration. No-op results do not establish build success or recovery.
+
+After persisting a Build and its trigger occurrence, build publishes to buildsignal and buildanalysis. On completion, buildsignal persists status, releases the build slot, finalizes the Request, and publishes to record and buildanalysis. At each point, both publications are attempted even if one fails. Any errors are returned to the consumer framework for its retry or DLQ policy. Publication is not atomic: one destination can receive its message while the other publication fails. Start and terminal analysis notices use distinct deterministic message IDs and the same build-ID partition; retries of one notice deduplicate without suppressing the other.
+
+Analysis reloads the Build and Request on each delivery, detects available issues, and creates immutable build-scoped observations. Observations already stored for the same build, task, and unit within a queue are not persisted again. A nonterminal Build holds for 30 seconds after each detection or persistence attempt, including failed attempts. Analysis errors remain visible in logs and metrics but do not consume queue retries while the build is active. A terminal Build receives detection and persistence before the delivery completes; failures use the subscription's normal bounded retry or DLQ policy. Malformed messages, invalid configuration, state-loading failures, and shutdown cancellation still return errors. The decision uses the status loaded before detection, so completion during a failed nonterminal pass leads to terminal processing on the next delivery.
+
+If build-status polling cannot recover, its DLQ handler fails the Request and releases its slot, but leaves the Build at its last observed status because the actual build result is unknown. If analysis observes a terminal Request with a nonterminal Build, it reloads the Build in case completion advanced it between the two reads. If the Build is still nonterminal, analysis stops rather than continuing to poll a request that has ended. The analysis subscription keeps dead-lettering enabled. No consumer automatically replays analysis dead letters; failed messages remain available for investigation and an explicit replay decision. Analysis never changes request outcome or releases a build slot. Record remains independent of analysis success.
+
+This branch records historical observations only; it does not maintain open issues, infer recovery from absence, or track issue lifecycle history.

doc/rfc/stovepipe/workflow.md

@@ -6,6 +6,8 @@ The pipeline is a queue-driven chain of small, single-purpose controllers, in th
 
 > **poll for a new head → ingest → process the build strategy → build → buildsignal → record whole-repo greenness (and any project results the resolver returns) → hook.**
 
+The [build issue observation](steps/buildanalysis.md) branch runs alongside this chain. It accumulates issues while builds run and performs terminal analysis without gating record. The reference host wires a no-op detector; deployments supply a real detector factory.
+
 There is no `analyze` controller and no `analyze` topic. A later project-analysis stage that would map a target graph onto another `build` pass is design only, described under [Designed, not built](#designed-not-built-project-analysis). `record` resolves project results inline on the same delivery that writes the repository fact.
 
 ## What Stovepipe is agnostic about

service/stovepipe/server/BUILD.bazel

@@ -26,6 +26,7 @@ go_library(
         "//service/stovepipe/server/handler:go_default_library",
         "//stovepipe/controller:go_default_library",
         "//stovepipe/controller/build:go_default_library",
+        "//stovepipe/controller/buildanalysis:go_default_library",
         "//stovepipe/controller/buildsignal:go_default_library",
         "//stovepipe/controller/dlq:go_default_library",
         "//stovepipe/controller/process:go_default_library",
@@ -34,6 +35,8 @@ go_library(
         "//stovepipe/core/queuelog:go_default_library",
         "//stovepipe/core/queuestate:go_default_library",
         "//stovepipe/core/requestlog:go_default_library",
+        "//stovepipe/extension/buildissue:go_default_library",
+        "//stovepipe/extension/buildissue/noop:go_default_library",
         "//stovepipe/extension/buildrunner:go_default_library",
         "//stovepipe/extension/buildrunner/fake:go_default_library",
         "//stovepipe/extension/projectstatus/noop:go_default_library",
@@ -89,6 +92,9 @@ go_test(
         "//stovepipe/controller/dlq: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/buildissue/noop: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",

service/stovepipe/server/main.go

@@ -48,6 +48,7 @@ import (
 	"github.com/uber/submitqueue/service/stovepipe/server/handler"
 	"github.com/uber/submitqueue/stovepipe/controller"
 	"github.com/uber/submitqueue/stovepipe/controller/build"
+	"github.com/uber/submitqueue/stovepipe/controller/buildanalysis"
 	"github.com/uber/submitqueue/stovepipe/controller/buildsignal"
 	"github.com/uber/submitqueue/stovepipe/controller/dlq"
 	"github.com/uber/submitqueue/stovepipe/controller/process"
@@ -56,6 +57,8 @@ import (
 	"github.com/uber/submitqueue/stovepipe/core/queuelog"
 	"github.com/uber/submitqueue/stovepipe/core/queuestate"
 	"github.com/uber/submitqueue/stovepipe/core/requestlog"
+	"github.com/uber/submitqueue/stovepipe/extension/buildissue"
+	buildissuenoop "github.com/uber/submitqueue/stovepipe/extension/buildissue/noop"
 	"github.com/uber/submitqueue/stovepipe/extension/buildrunner"
 	buildrunnerfake "github.com/uber/submitqueue/stovepipe/extension/buildrunner/fake"
 	projectstatusnoop "github.com/uber/submitqueue/stovepipe/extension/projectstatus/noop"
@@ -214,18 +217,15 @@ func run() error {
 		subscriberName = fmt.Sprintf("stovepipe-%d", time.Now().Unix())
 	}
 
+	detectorFactory := buildissuenoop.New()
 	registry, err := newTopicRegistry(mysqlQueue, subscriberName)
 	if err != nil {
 		return fmt.Errorf("failed to create topic registry: %w", err)
 	}
 
-	// Two consumers share the topic registry but apply different error classification
-	// policies. The primary consumer runs the standard classifier walk. The DLQ consumer
-	// uses AlwaysRetryableProcessor so every non-nil error from a DLQ controller is
-	// forced retryable — reconciliation must redeliver on any failure because the DLQ
-	// subscription is a final destination (DLQ.Enabled is false on it, so there is no
-	// further DLQ to fall back on). Stovepipe has no gated deployment yet, so both
-	// consumers use the no-op gate.
+	// The primary consumer applies backend classification. The DLQ consumer
+	// retries recovery work without a second-level DLQ; analysis dead letters
+	// are retained without automatic replay. Both consumers use no-op gates.
 	primaryConsumer := consumer.New(logger.Sugar(), scope.SubScope("consumer"), registry,
 		errs.NewClassifierProcessor(
 			genericerrs.Classifier,
@@ -254,7 +254,7 @@ func run() error {
 		}
 	}
 	materializer := requestlog.NewMaterializer(scope)
-	primaryCount, err := registerPrimaryControllers(primaryConsumer, logger.Sugar(), scope, storageFty, materializer, registry, sourceControl, brf, hookResolver{})
+	primaryCount, err := registerPrimaryControllers(primaryConsumer, logger.Sugar(), scope, storageFty, materializer, registry, sourceControl, brf, hookResolver{}, detectorFactory)
 	if err != nil {
 		return err
 	}
@@ -373,6 +373,7 @@ func registerPrimaryControllers(
 	sourceControl sourcecontrol.Factory,
 	brf buildrunner.Factory,
 	hooks hookext.Hooks,
+	detectorFactory buildissue.DetectorFactory,
 ) (int, error) {
 	var count int
 
@@ -404,6 +405,12 @@ func registerPrimaryControllers(
 	}
 	count++
 
+	analysisController := buildanalysis.NewController(logger, scope, store, detectorFactory, stovepipemq.TopicKeyBuildAnalysis, "stovepipe-buildanalysis")
+	if err := c.Register(analysisController); err != nil {
+		return count, fmt.Errorf("failed to register buildanalysis controller: %w", err)
+	}
+	count++
+
 	recordController := record.NewController(logger, scope, store, materializer, projectstatusnoop.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)
@@ -419,8 +426,8 @@ func registerPrimaryControllers(
 	return count, nil
 }
 
-// registerDLQControllers creates one DLQ reconciler per primary stage and
-// registers them with c, returning how many were registered.
+// registerDLQControllers registers recovery for validation and hook stages.
+// Analysis dead letters are retained without an automatic replay consumer.
 func registerDLQControllers(
 	c consumer.Consumer,
 	logger *zap.SugaredLogger,
@@ -463,15 +470,8 @@ func registerDLQControllers(
 	return count, nil
 }
 
-// newTopicRegistry builds the TopicRegistry for Stovepipe's internal pipeline queues. ingest
-// publishes to the process topic and the process consumer subscribes to it; process publishes
-// to the build topic and the build consumer subscribes to it; build publishes to the buildsignal
-// topic and the buildsignal consumer subscribes to it, and also republishes to itself while
-// polling. buildsignal publishes to the record topic once a build reaches a terminal status,
-// and the record consumer subscribes to it.
-//
-// The hook topic name is domain-qualified because its key is shared across domains: two
-// domains pointed at one queue backend would otherwise consume each other's events.
+// newTopicRegistry includes every pipeline topic and its dead-letter destination.
+// Hook topic names are domain-qualified because the hook key is shared across domains.
 func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRegistry, error) {
 	return consumer.NewTopicRegistry([]consumer.TopicConfig{
 		{
@@ -544,6 +544,17 @@ func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRe
 			Queue:        q,
 			Subscription: extqueue.DLQSubscriptionConfig(subscriberName, "stovepipe-hook-dlq"),
 		},
+		{
+			Key:          stovepipemq.TopicKeyBuildAnalysis,
+			Name:         "buildanalysis",
+			Queue:        q,
+			Subscription: extqueue.ExtendedSubscriptionConfig(subscriberName, "stovepipe-buildanalysis"),
+		},
+		{
+			Key:   dlq.TopicKey(stovepipemq.TopicKeyBuildAnalysis),
+			Name:  "buildanalysis_dlq",
+			Queue: q,
+		},
 	})
 }
 

service/stovepipe/server/main_test.go

@@ -28,6 +28,9 @@ import (
 	"github.com/uber/submitqueue/stovepipe/controller/dlq"
 	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/buildissue"
+	buildissuenoop "github.com/uber/submitqueue/stovepipe/extension/buildissue/noop"
 	"go.uber.org/zap/zaptest"
 )
 
@@ -60,7 +63,7 @@ func registeredControllers(t *testing.T) (consumer.TopicRegistry, []consumer.Con
 	deadLetter := &recordingConsumer{}
 
 	_, err = registerPrimaryControllers(primary, logger, tally.NoopScope, store, requestlog.NewMaterializer(tally.NoopScope), registry,
-		fakeSourceControlFactory{}, fakeBuildRunnerFactory{}, hookResolver{})
+		fakeSourceControlFactory{}, fakeBuildRunnerFactory{}, hookResolver{}, buildissuenoop.New())
 	require.NoError(t, err)
 
 	_, err = registerDLQControllers(deadLetter, logger, tally.NoopScope, store, requestlog.NewMaterializer(tally.NoopScope))
@@ -129,6 +132,7 @@ func TestPipelineSubscriptionRetryPolicies(t *testing.T) {
 		{name: "process", topicKey: stovepipemq.TopicKeyProcess, consumerGroup: "stovepipe-process", wantRetry: defaultRetry},
 		{name: "build", topicKey: stovepipemq.TopicKeyBuild, consumerGroup: "stovepipe-build", wantRetry: defaultRetry},
 		{name: "buildsignal", topicKey: stovepipemq.TopicKeyBuildSignal, consumerGroup: "stovepipe-buildsignal", wantRetry: extendedRetry},
+		{name: "buildanalysis", topicKey: stovepipemq.TopicKeyBuildAnalysis, consumerGroup: "stovepipe-buildanalysis", wantRetry: extendedRetry},
 		{name: "record", topicKey: stovepipemq.TopicKeyRecord, consumerGroup: "stovepipe-record", wantRetry: extendedRetry},
 		{name: "hook", topicKey: basehook.TopicKeyHook, consumerGroup: "stovepipe-hook", wantRetry: defaultRetry},
 	}
@@ -154,3 +158,24 @@ func TestDLQSubscriptionRetryPoliciesRemainUnlimited(t *testing.T) {
 		})
 	}
 }
+
+func TestBuildAnalysisIsWiredWithNoopDetector(t *testing.T) {
+	registry, primary, deadLetter := registeredControllers(t)
+	assert.Contains(t, topicKeys(primary), stovepipemq.TopicKeyBuildAnalysis)
+	assert.NotContains(t, topicKeys(deadLetter), dlq.TopicKey(stovepipemq.TopicKeyBuildAnalysis))
+	dlqName, ok := registry.TopicName(dlq.TopicKey(stovepipemq.TopicKeyBuildAnalysis))
+	require.True(t, ok)
+	assert.Equal(t, "buildanalysis_dlq", dlqName)
+	_, subscribed := registry.SubscriptionConfig(dlq.TopicKey(stovepipemq.TopicKeyBuildAnalysis), "stovepipe-buildanalysis-dlq")
+	assert.False(t, subscribed)
+	cfg, ok := registry.SubscriptionConfig(stovepipemq.TopicKeyBuildAnalysis, "stovepipe-buildanalysis")
+	require.True(t, ok)
+	assert.True(t, cfg.DLQ.Enabled)
+	assert.Equal(t, "_dlq", cfg.DLQ.TopicSuffix)
+
+	detector, err := (buildissuenoop.New()).For(buildissue.DetectorConfig{QueueName: "validation/main"})
+	require.NoError(t, err)
+	issues, err := detector.Detect(context.Background(), entity.BuildID{ID: "build/42"})
+	require.NoError(t, err)
+	assert.Empty(t, issues)
+}

stovepipe/controller/build/BUILD.bazel

@@ -24,7 +24,10 @@ go_library(
 
 go_test(
     name = "go_default_test",
-    srcs = ["build_test.go"],
+    srcs = [
+        "build_analysis_test.go",
+        "build_test.go",
+    ],
     embed = [":go_default_library"],
     deps = [
         "//platform/base/messagequeue:go_default_library",
@@ -33,10 +36,15 @@ go_test(
         "//platform/errs:go_default_library",
         "//platform/extension/messagequeue/mock:go_default_library",
         "//platform/metrics:go_default_library",
+        "//platform/publish:go_default_library",
+        "//stovepipe/controller/buildanalysis:go_default_library",
+        "//stovepipe/controller/buildsignal: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/extension/buildrunner:go_default_library",
         "//stovepipe/extension/buildrunner/mock:go_default_library",
         "//stovepipe/extension/storage:go_default_library",

stovepipe/controller/build/build.go

@@ -15,11 +15,12 @@
 // Package build holds the build-stage queue controller. It consumes BuildRequest
 // messages (a request id), reloads the Request, triggers the build-runner for
 // the scope process already decided, persists a Build row, and publishes the
-// build id to buildsignal.
+// build id to buildsignal and buildanalysis.
 package build
 
 import (
 	"context"
+	"errors"
 	"fmt"
 
 	"github.com/uber-go/tally"
@@ -39,7 +40,7 @@ import (
 
 // Controller consumes BuildRequest messages, reloads the referenced Request,
 // triggers a build for its already-decided scope, and publishes the resulting
-// build id to buildsignal. Implements consumer.Controller.
+// build id to status polling and analysis. Implements consumer.Controller.
 type Controller struct {
 	logger        *zap.SugaredLogger
 	metricsScope  tally.Scope
@@ -51,9 +52,6 @@ type Controller struct {
 	consumerGroup string
 }
 
-// Verify Controller implements consumer.Controller interface at compile time.
-var _ consumer.Controller = (*Controller)(nil)
-
 // _opName is the metric operation name shared by every emit in this file.
 const _opName = "build"
 
@@ -155,8 +153,13 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 		return err
 	}
 
-	if err := c.publishBuildSignal(ctx, build.ID, request.Queue); err != nil {
-		return fmt.Errorf("failed to publish build signal for %s: %w", build.ID, err)
+	signalErr := c.publishBuildSignal(ctx, build.ID, request.Queue)
+	if signalErr != nil {
+		signalErr = fmt.Errorf("failed to publish build signal for %s: %w", build.ID, signalErr)
+	}
+	analysisErr := c.publishBuildAnalysis(ctx, build.ID, request.Queue)
+	if err := errors.Join(signalErr, analysisErr); err != nil {
+		return err
 	}
 
 	c.logger.Debugw("triggered build",
@@ -219,3 +222,21 @@ func (c *Controller) TopicKey() consumer.TopicKey {
 func (c *Controller) ConsumerGroup() string {
 	return c.consumerGroup
 }
+
+func (c *Controller) publishBuildAnalysis(ctx context.Context, buildID, queue string) error {
+	payload, err := stovepipemq.Marshal(&stovepipemq.BuildAnalysis{Id: buildID, QueueName: queue})
+	if err != nil {
+		return fmt.Errorf("failed to serialize build analysis: %w", err)
+	}
+	if err := publish.Message(ctx, c.registry, stovepipemq.TopicKeyBuildAnalysis, publish.MessageParams{
+		Tenant:       queue,
+		ID:           publish.IntentID(buildID, "started"),
+		Payload:      payload,
+		PartitionKey: buildID,
+	}); err != nil {
+		return fmt.Errorf("failed to publish build analysis for %s: %w", buildID, err)
+	}
+	return nil
+}
+
+var _ consumer.Controller = (*Controller)(nil)