feat(stovepipe): reconcile dead-lettered build signals and free the slot

doc/rfc/stovepipe/steps/buildsignal.md

@@ -122,13 +122,24 @@ Per `platform/errs`'s non-retryable-by-default rule (see [platform/errs/README.m
 
 | Failure | Disposition | Why |
 |---|---|---|
-| `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. |
+| `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. **This means the `BuildRunner` backend has to classify**: an unclassified transport or HTTP error gets the non-retryable default, so one proxy blip ends the poll chain (see below). |
 | `Update` CAS conflict (`ErrVersionMismatch`) | declaration-level retryable | A concurrent (redelivered) writer moved the row; reload and re-check converges. |
 
 `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 `record` publish — 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. The poll loop itself no longer has a publish to fail: holding is a local outcome, and a failed postpone write in the framework lapses into a normal visibility-timeout redelivery, so the loop's liveness never rides on an enqueue succeeding.
 
+### What it costs when a backend does not classify `Status` errors
+
+Leaving `Status` to the classifier only works if the backend classifies. A `BuildRunner` whose transport returns plain `fmt.Errorf` values gets the non-retryable default, and here that default is expensive: dead-lettering ends the *only* poll chain for a build that is still running, and the request keeps holding one of the queue's `in_flight_count` build slots until reconciliation gives it back. A single `502` from a proxy in front of the build API then looks exactly like "this build can never be polled".
+
+Two things keep a blip from stalling a queue, and a backend needs both:
+
+- **The backend classifies its own failures.** Transport errors and 5xx/429/408 responses are `errs.NewRetryableDependencyError`. A 4xx about the request itself — unknown build, forbidden — is `errs.NewDependencyError`. Only the layer that sees the status code can tell these apart, which is why the table above leaves the call to it.
+- **The retry budget is worth something.** Retryable means nack, and a nacked message comes back on the next poll, so `Retry.MaxAttempts` counts attempts rather than time — the default three are spent in a few hundred milliseconds. Raising `MaxAttempts` on this subscription buys a little more, but each attempt is another request at a dependency that is already failing, so it does not stretch to cover a proxy restart. Until nacks are spaced by the configured retry backoff, it is the reconciler below rather than the retry budget that keeps a longer outage from costing the queue a slot.
+
+When the budget does run out the message dead-letters, and the buildsignal DLQ reconciler (`stovepipe/controller/dlq/buildsignal.go`) is what makes that recoverable: it maps the build back to its request, releases the slot, and marks the request `failed`. A deployment that registers the primary consumers but not that reconciler has no fail-closed path for this stage, and loses a slot for good every time this happens.
+
 ## Idempotency
 
 Every branch is safe under at-least-once redelivery:

service/stovepipe/server/main.go

@@ -450,6 +450,12 @@ func registerDLQControllers(
 	}
 	count++
 
+	buildSignalDLQController := dlq.NewBuildSignalController(logger, scope, store, dlq.TopicKey(stovepipemq.TopicKeyBuildSignal), "stovepipe-buildsignal-dlq")
+	if err := c.Register(buildSignalDLQController); err != nil {
+		return count, fmt.Errorf("failed to register buildsignal dlq controller: %w", err)
+	}
+	count++
+
 	return count, nil
 }
 
@@ -499,6 +505,12 @@ func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRe
 			Queue:        q,
 			Subscription: extqueue.DLQSubscriptionConfig(subscriberName, "stovepipe-process-dlq"),
 		},
+		{
+			Key:          dlq.TopicKey(stovepipemq.TopicKeyBuildSignal),
+			Name:         "buildsignal_dlq",
+			Queue:        q,
+			Subscription: extqueue.DLQSubscriptionConfig(subscriberName, "stovepipe-buildsignal-dlq"),
+		},
 	})
 }
 

stovepipe/controller/dlq/BUILD.bazel

@@ -3,6 +3,7 @@ load("@rules_go//go:def.bzl", "go_library", "go_test")
 go_library(
     name = "go_default_library",
     srcs = [
+        "buildsignal.go",
         "dlq.go",
         "request.go",
     ],
@@ -21,7 +22,10 @@ go_library(
 
 go_test(
     name = "go_default_test",
-    srcs = ["dlq_test.go"],
+    srcs = [
+        "buildsignal_test.go",
+        "dlq_test.go",
+    ],
     embed = [":go_default_library"],
     deps = [
         "//platform/base/messagequeue:go_default_library",

stovepipe/controller/dlq/buildsignal.go

@@ -0,0 +1,167 @@
+// Copyright (c) 2025 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 dlq
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"github.com/uber-go/tally"
+	"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/extension/storage"
+	"go.uber.org/zap"
+)
+
+// _buildSignalOpName is the metric operation name shared by every emit in this file.
+const _buildSignalOpName = "buildsignal_dlq"
+
+// BuildSignalController is the DLQ reconciler for the buildsignal stage. The
+// payload names a build, not a request, so it takes one more step than the
+// process reconciler: read the build to get its RequestID, then fail that
+// request via failRequest.
+//
+// This DLQ is the one that matters most. A request only reaches buildsignal
+// after process admitted it, so it holds one of the queue's in_flight_count
+// build slots, and buildsignal's terminal path is the only thing that gives that
+// slot back. Once a poll message dead-letters — a Status call that stayed broken
+// through every retry, an unknown build id, a storage write that kept failing —
+// nothing else in the pipeline will look at that build again. Without this
+// reconciler the request stays processing for good and the slot is never
+// returned, so the queue loses one slot per incident until it has none left and
+// stops admitting work.
+//
+// The Build row keeps whatever non-terminal status the runner last reported.
+// There is nothing useful to fix: record decides greenness from Request.State,
+// not Build.Status, and writing a terminal status here would claim we saw an
+// outcome we never saw.
+type BuildSignalController struct {
+	logger        *zap.SugaredLogger
+	metricsScope  tally.Scope
+	stores        storage.Factory
+	topicKey      consumer.TopicKey
+	consumerGroup string
+}
+
+// Verify BuildSignalController implements consumer.Controller at compile time.
+var _ consumer.Controller = (*BuildSignalController)(nil)
+
+// NewBuildSignalController creates a DLQ controller for the buildsignal stage's
+// dead-letter topic. topicKey is typically
+// dlq.TopicKey(stovepipemq.TopicKeyBuildSignal).
+func NewBuildSignalController(
+	logger *zap.SugaredLogger,
+	scope tally.Scope,
+	stores storage.Factory,
+	topicKey consumer.TopicKey,
+	consumerGroup string,
+) *BuildSignalController {
+	return &BuildSignalController{
+		logger:        logger.Named("buildsignal_dlq_controller"),
+		metricsScope:  scope.SubScope("buildsignal_dlq_controller"),
+		stores:        stores,
+		topicKey:      topicKey,
+		consumerGroup: consumerGroup,
+	}
+}
+
+// Process reconciles a single DLQ delivery for the buildsignal topic. Returns nil
+// to ack (success) or an error to nack (retry) — pair this controller only with a
+// consumer wired with errs.AlwaysRetryableProcessor so a transient reconcile
+// failure retries instead of dead-lettering the DLQ message itself.
+func (c *BuildSignalController) Process(ctx context.Context, delivery consumer.Delivery) error {
+	msg := delivery.Message()
+
+	sig := &stovepipemq.BuildSignal{}
+	if err := stovepipemq.Unmarshal(msg.Payload, sig); err != nil {
+		metrics.NamedCounter(c.metricsScope, _buildSignalOpName, "deserialize_errors", 1)
+		// Retried rather than acked, for the same deployment-skew reason the
+		// process reconciler gives: a newer producer's payload decodes fine once
+		// the rollout finishes, and acking here would skip the slot release
+		// without saying so.
+		return fmt.Errorf("failed to decode dlq payload: %w", err)
+	}
+	if sig.Id == "" {
+		metrics.NamedCounter(c.metricsScope, _buildSignalOpName, "empty_id_errors", 1)
+		return fmt.Errorf("dlq payload decoded to empty build id")
+	}
+
+	store, err := c.stores.For(storage.Config{QueueName: sig.GetQueueName()})
+	if err != nil {
+		metrics.NamedCounter(c.metricsScope, _buildSignalOpName, "storage_resolve_errors", 1)
+		// Non-retryable: a missing or unresolvable queue is a malformed message.
+		return fmt.Errorf("failed to resolve storage for queue %q: %w", sig.GetQueueName(), err)
+	}
+
+	dmeta := delivery.Metadata()
+	c.logger.Warnw("dlq message received",
+		"build_id", sig.Id,
+		"attempt", delivery.Attempt(),
+		"dlq_original_topic", dmeta["dlq.original_topic"],
+		"dlq_failure_count", dmeta["dlq.failure_count"],
+		"dlq_last_error", dmeta["dlq.last_error"],
+	)
+
+	build, err := store.GetBuildStore().Get(ctx, sig.Id)
+	if err != nil {
+		if errors.Is(err, storage.ErrNotFound) {
+			// The build row was never written — a crash between Trigger and
+			// Create. There is no request to recover from this payload; the build
+			// stage's own DLQ handles the request that triggered it.
+			c.logger.Warnw("dlq reconcile: build not found, skipping", "build_id", sig.Id)
+			metrics.NamedCounter(c.metricsScope, _buildSignalOpName, "build_not_found", 1)
+			return nil
+		}
+		metrics.NamedCounter(c.metricsScope, _buildSignalOpName, "build_store_errors", 1)
+		return fmt.Errorf("failed to get build %s: %w", sig.Id, err)
+	}
+
+	if build.RequestID == "" {
+		// Defensive: a build with no request has nothing to reconcile and no slot
+		// to release. Ack it so the DLQ does not grow forever.
+		c.logger.Errorw("dlq reconcile: build has empty request id, skipping", "build_id", sig.Id)
+		metrics.NamedCounter(c.metricsScope, _buildSignalOpName, "build_missing_request", 1)
+		return nil
+	}
+
+	// Every request reachable from a build row is either still processing, and holding
+	// the slot failRequest releases, or already terminal, and past releasing it: build
+	// triggers only once process has written the strategy, which lands in the same CAS
+	// as accepted→processing, and processing exits only to a terminal outcome.
+	if err := failRequest(ctx, store, c.logger, build.RequestID); err != nil {
+		metrics.NamedCounter(c.metricsScope, _buildSignalOpName, "reconcile_errors", 1)
+		return err
+	}
+
+	metrics.NamedCounter(c.metricsScope, _buildSignalOpName, "reconciled", 1)
+	return nil
+}
+
+// Name returns the controller name for logging and metrics.
+func (c *BuildSignalController) Name() string {
+	return "buildsignal_dlq"
+}
+
+// TopicKey returns the topic key this controller subscribes to.
+func (c *BuildSignalController) TopicKey() consumer.TopicKey {
+	return c.topicKey
+}
+
+// ConsumerGroup returns the consumer group for offset tracking.
+func (c *BuildSignalController) ConsumerGroup() string {
+	return c.consumerGroup
+}