feat(stovepipe): Wire build analysis pipeline

Author: prathshenoy

doc/rfc/stovepipe/steps/buildanalysis.md

@@ -0,0 +1,29 @@
+# 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.
+
+## Start and completion notices
+
+After persisting a Build and its trigger occurrence, build publishes to buildsignal and buildanalysis. On completion, buildsignal persists status, releases the build slot, records the Request outcome, and publishes to record and buildanalysis. Each pair of publications attempts both destinations even if one fails. Publication is not atomic: one destination can receive its message while the other publication fails.
+
+Start and completion notices use distinct deterministic message IDs and the same build-ID partition. Retries of one notice deduplicate without suppressing the other. A held start delivery can delay the completion notice in that partition, but it also reloads Build status on every pass and can perform terminal analysis itself. Both notices may analyze the terminal Build; duplicate observations are safe.
+
+Publication errors are returned through the producing stage's retry/DLQ policy. Analysis processing does not gate record, but notification publication is still part of the producing stage: a failed start publication can send the build-stage message to its existing fail-closed recovery path.
+
+## Observations and polling
+
+Each analysis delivery reloads the Build and Request, invokes the detector, and persists observations individually. Observations already stored for the same build, task, and unit within a queue are not persisted again. A persistence error can leave earlier observations stored; retrying does not duplicate them. Detector results accompanying an error are discarded.
+
+After a successful pass, a nonterminal Build holds the delivery for 30 seconds; a terminal Build completes its delivery after persistence. The status read before detection determines whether the pass holds, so completion during detection requires another delivery to analyze the terminal Build. Analysis only accumulates historical observations: it does not maintain open issues, infer recovery from absence, or track issue lifecycle history.
+
+## Requests that end without a build result
+
+If build-status polling cannot recover, its DLQ handler fails a nonterminal Request and releases its slot, but leaves the Build at its last observed status. That recovery path does not publish a build-completion notice.
+
+If analysis observes a terminal Request with a nonterminal Build, it reloads the Build before stopping: completion may have occurred between the initial Build read and the Request read. If the Build is now terminal, analysis proceeds. Otherwise it stops without detection or another hold. Analysis itself never changes Request outcome or releases a build slot.
+
+## Analysis errors and dead letters
+
+Analysis errors are returned to the normal consumer whether the Build is running or terminal; errors do not use the 30-second polling hold. The reference service uses backend error classification and the extended retry configuration: a maximum of 10 attempts, with retry delays of 5, 10, 20, 40, then 60 seconds, capped at 60 seconds. A successful hold resets failure accounting. Non-retryable errors and exhausted retries send the message to the analysis DLQ, so final analysis is attempted rather than guaranteed to succeed.
+
+The analysis DLQ has no subscription or recovery controller. Dead letters remain subject to queue retention and require an explicit investigation or replay decision; they are not automatically retried indefinitely.

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 attempts final analysis after build completion. Record does not wait for analysis to succeed. 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,6 +217,7 @@ 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)
@@ -254,7 +258,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 +377,7 @@ func registerPrimaryControllers(
 	sourceControl sourcecontrol.Factory,
 	brf buildrunner.Factory,
 	hooks hookext.Hooks,
+	detectorFactory buildissue.DetectorFactory,
 ) (int, error) {
 	var count int
 
@@ -404,6 +409,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 +430,7 @@ 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 the validation and hook recovery controllers.
 func registerDLQControllers(
 	c consumer.Consumer,
 	logger *zap.SugaredLogger,
@@ -463,15 +473,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 +547,18 @@ 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"),
+		},
+		{
+			// No subscription: analysis dead letters require an explicit replay decision.
+			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,28 @@ 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,14 @@ 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 +223,23 @@ 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)