feat(stovepipe): add and wire the buildsignal stage controller (#409)

## What?
Adds `stovepipe/controller/buildsignal` plus the new `Record` queue
message/proto and `TopicKeyRecord`. `Process` consumes a `BuildSignal`
(build id), loads `Build`+`Request`, no-ops if the request is terminal,
polls `buildRunner.Status`, persists a real status transition via CAS
(`BuildStore.Update`) with terminal status **write-once** (a later poll
of a flaky backend can never overwrite an already-committed terminal
status), and either reschedules itself via `PublishAfter` (5s while
`accepted`, 2s while `running`) or, once terminal, publishes to
`record`. Wires the controller into `main.go` and registers the
`buildsignal` (subscribed) and `record` (publish-only, no consumer yet)
topics.

## Test Plan
`make build`, `make test` - all clean.

## Issue
<!-- 
Link the issue here. 
- Use 'Closes #123' if this is the final fix. 
- Use 'Part of #123' or just '#123' if the feature is still in progress.
-->

Author: roychying

doc/rfc/stovepipe/steps/buildsignal.md

@@ -20,14 +20,12 @@ For a delivery carrying build id `B`:
 
 ```
 1. Load Build B from the build store.
-   - ErrNotFound       -> retryable (build's Create not visible yet; redelivery converges).
+   - ErrNotFound       -> return raw; non-retryable (storage is read-after-write consistent; see [storage README](stovepipe/extension/storage/README.md)).
    - other store error -> return raw; classifier decides.
 
 2. Load Request R = store.Get(Build.RequestID) — needed for R.Queue to resolve the build-runner.
-   - ErrNotFound -> retryable, like step 1: the Build's existence proves the Request write is older,
-     so a miss here is almost certainly a lagging read; redelivery converges. A genuinely orphaned
-     Build (integrity fault) still dead-letters at MaxAttempts — the same terminal outcome, without
-     rejecting straight to DLQ on a stale read.
+   - ErrNotFound -> return raw; non-retryable, same as step 1 — the Build's existence proves the
+     Request write is already committed, so a miss here is a storage defect, not a lagging read.
 
 3. If R.State is terminal (superseded / recorded-green / recorded-not-green): ack and return.
    - the Request is done (record already ran, or the head was superseded); stop polling.
@@ -102,19 +100,19 @@ Per `platform/errs`'s non-retryable-by-default rule (see [platform/errs/README.m
 
 | Failure | Disposition | Why |
 |---|---|---|
-| `Build` not found | retryable (`errs.NewRetryableError`) | `build`'s `Create` not visible yet; redelivery converges. |
-| `Request` not found | retryable (`errs.NewRetryableError`) | The Build's existence proves the Request write is older, so a miss is a stale read; a genuine orphan still dead-letters at `MaxAttempts`. |
 | `Status` call | raw error; classifier decides | Deliberately left open rather than fixed either way — runner timeout/connection is transient, "runner not deployed for this queue" is not, and only a backend classifier can tell them apart. |
 | `Update` CAS conflict (`ErrVersionMismatch`) | retryable | A concurrent (redelivered) writer moved the row; reload and re-check converges. |
 | `PublishAfter` re-poll | retryable | The poll heartbeat; it runs only after status/persist/record all succeeded, so a transient enqueue blip is worth replaying to `MaxAttempts` before dead-lettering. |
 
+`Build`/`Request` not found (`storage.ErrNotFound`) are **not** in this table: storage is required to be read-after-write consistent (see [storage README](stovepipe/extension/storage/README.md)), so a miss here is already the correct default (non-retryable, straight to DLQ) rather than a departure worth overriding.
+
 Everything else — factory lookup, an `Update` store error other than a CAS conflict, and the publish to `record` — is returned raw with no override, because the default is already correct: a queue with no registered runner is a config error, and storage/queue failures dead-letter and let DLQ reconciliation recover.
 
 ## Idempotency
 
 Every branch is safe under at-least-once redelivery:
 
-- **Build not found** — retryable; converges as the row becomes visible.
+- **Build not found** — non-retryable; storage's read-after-write guarantee means a miss here is a storage defect, not a lag condition to retry through.
 - **Status already persisted** — a redelivery re-runs the whole algorithm from step 1, including a redundant `Status` poll (harmless — the runner reports the same thing); step 6 no-ops on the unchanged status, and the delivery proceeds to re-schedule the poll (non-terminal) or republish to `record` (terminal, idempotent). No corruption.
 - **Terminal already published** — a redelivery reloads, re-polls, no-ops at step 6, republishes the same terminal signal to `record` (idempotent), and acks. Harmless.
 - **`PublishAfter` failed, then retried** — the nacked delivery re-runs from step 1; there is no way to resume mid-algorithm, so it re-polls the runner too, but the row already carries the non-terminal status and step 6 no-ops. Only the final enqueue does new work.

service/stovepipe/server/BUILD.bazel

@@ -16,6 +16,7 @@ go_library(
         "//service/stovepipe/server/mapper:go_default_library",
         "//stovepipe/controller:go_default_library",
         "//stovepipe/controller/build:go_default_library",
+        "//stovepipe/controller/buildsignal:go_default_library",
         "//stovepipe/controller/dlq:go_default_library",
         "//stovepipe/controller/process:go_default_library",
         "//stovepipe/core/messagequeue:go_default_library",

service/stovepipe/server/main.go

@@ -38,6 +38,7 @@ import (
 	"github.com/uber/submitqueue/service/stovepipe/server/mapper"
 	"github.com/uber/submitqueue/stovepipe/controller"
 	"github.com/uber/submitqueue/stovepipe/controller/build"
+	"github.com/uber/submitqueue/stovepipe/controller/buildsignal"
 	"github.com/uber/submitqueue/stovepipe/controller/dlq"
 	"github.com/uber/submitqueue/stovepipe/controller/process"
 	stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue"
@@ -382,6 +383,12 @@ func registerPrimaryControllers(
 	}
 	count++
 
+	buildSignalController := buildsignal.NewController(logger, scope, store, brf, registry, stovepipemq.TopicKeyBuildSignal, "stovepipe-buildsignal")
+	if err := c.Register(buildSignalController); err != nil {
+		return count, fmt.Errorf("failed to register buildsignal controller: %w", err)
+	}
+	count++
+
 	return count, nil
 }
 
@@ -407,8 +414,10 @@ func registerDLQControllers(
 
 // 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. The buildsignal topic is added
-// once the buildsignal controller lands to consume it.
+// 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; it
+// has no Subscription yet since no consumer for it exists until the record stage lands.
 func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRegistry, error) {
 	return consumer.NewTopicRegistry([]consumer.TopicConfig{
 		{
@@ -427,6 +436,19 @@ func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRe
 				subscriberName, "stovepipe-build",
 			),
 		},
+		{
+			Key:   stovepipemq.TopicKeyBuildSignal,
+			Name:  "buildsignal",
+			Queue: q,
+			Subscription: extqueue.DefaultSubscriptionConfig(
+				subscriberName, "stovepipe-buildsignal",
+			),
+		},
+		{
+			Key:   stovepipemq.TopicKeyRecord,
+			Name:  "record",
+			Queue: q,
+		},
 		{
 			Key:          dlq.TopicKey(stovepipemq.TopicKeyProcess),
 			Name:         "process_dlq",

stovepipe/controller/build/BUILD.bazel

@@ -10,6 +10,7 @@ go_library(
         "//platform/consumer:go_default_library",
         "//platform/errs: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/buildrunner:go_default_library",

stovepipe/controller/build/build.go

@@ -28,6 +28,7 @@ import (
 	"github.com/uber/submitqueue/platform/consumer"
 	"github.com/uber/submitqueue/platform/errs"
 	"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/buildrunner"
@@ -149,11 +150,7 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r
 
 // loadRequest returns the request for id.
 func (c *Controller) loadRequest(ctx context.Context, id string) (entity.Request, error) {
-	got, err := c.store.GetRequestStore().Get(ctx, id)
-	if err != nil {
-		return entity.Request{}, fmt.Errorf("BuildController failed to load request %s: %w", id, err)
-	}
-	return got, nil
+	return loader.ByID(ctx, id, c.store.GetRequestStore().Get, "BuildController", "request")
 }
 
 // publishBuildSignal publishes buildID to the buildsignal stage, partitioned by

stovepipe/controller/buildsignal/BUILD.bazel

@@ -0,0 +1,44 @@
+load("@rules_go//go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["buildsignal.go"],
+    importpath = "github.com/uber/submitqueue/stovepipe/controller/buildsignal",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//platform/base/messagequeue:go_default_library",
+        "//platform/consumer:go_default_library",
+        "//platform/errs: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/buildrunner: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 = ["buildsignal_test.go"],
+    embed = [":go_default_library"],
+    deps = [
+        "//platform/base/messagequeue:go_default_library",
+        "//platform/consumer:go_default_library",
+        "//platform/errs:go_default_library",
+        "//platform/extension/messagequeue/mock:go_default_library",
+        "//stovepipe/core/messagequeue:go_default_library",
+        "//stovepipe/entity:go_default_library",
+        "//stovepipe/extension/buildrunner:go_default_library",
+        "//stovepipe/extension/buildrunner/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",
+    ],
+)