refactor(stovepipe): align DLQ controller conventions (#619)

## Summary
Intent:
- Align the existing Stovepipe DLQ controllers with the repository-wide
convention used by the parent PR.
- Keep the public construction API consistent as additional DLQ stages
are added.

Changes:
- Rename the process and buildsignal constructors to the
NewDLQ<Stage>Controller pattern.
- Hide controller implementations behind consumer.Controller and derive
controller names from their topic keys.
- Update service wiring and test fixtures for the new constructors.

This PR is stacked on #618, which adds build-stage DLQ reconciliation
using the same convention.

## Test Plan
- Run the Stovepipe DLQ controller test target.
- Build the Stovepipe server target.

## Revert Plan
- Revert this PR to restore the prior Stovepipe DLQ constructor and
implementation type names.

## Issues

Author: mnoah1

service/stovepipe/server/main.go

@@ -450,7 +450,7 @@ func registerDLQControllers(
 ) (int, error) {
 	var count int
 
-	processDLQController := dlq.NewController(logger, scope, store, dlq.TopicKey(stovepipemq.TopicKeyProcess), "stovepipe-process-dlq")
+	processDLQController := dlq.NewDLQRequestController(logger, scope, store, dlq.TopicKey(stovepipemq.TopicKeyProcess), "stovepipe-process-dlq")
 	if err := c.Register(processDLQController); err != nil {
 		return count, fmt.Errorf("failed to register process dlq controller: %w", err)
 	}
@@ -462,7 +462,7 @@ func registerDLQControllers(
 	}
 	count++
 
-	buildSignalDLQController := dlq.NewBuildSignalController(logger, scope, store, dlq.TopicKey(stovepipemq.TopicKeyBuildSignal), "stovepipe-buildsignal-dlq")
+	buildSignalDLQController := dlq.NewDLQBuildSignalController(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)
 	}

stovepipe/controller/dlq/buildsignal.go

@@ -30,7 +30,7 @@ import (
 // _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
+// 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.
@@ -49,30 +49,31 @@ const _buildSignalOpName = "buildsignal_dlq"
 // 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 {
+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)
+// Verify buildSignalController implements consumer.Controller at compile time.
+var _ consumer.Controller = (*buildSignalController)(nil)
 
-// NewBuildSignalController creates a DLQ controller for the buildsignal stage's
+// NewDLQBuildSignalController creates a DLQ controller for the buildsignal stage's
 // dead-letter topic. topicKey is typically
 // dlq.TopicKey(stovepipemq.TopicKeyBuildSignal).
-func NewBuildSignalController(
+func NewDLQBuildSignalController(
 	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"),
+) consumer.Controller {
+	name := string(topicKey) + "_controller"
+	return &buildSignalController{
+		logger:        logger.Named(name),
+		metricsScope:  scope.SubScope(name),
 		stores:        stores,
 		topicKey:      topicKey,
 		consumerGroup: consumerGroup,
@@ -83,7 +84,7 @@ func NewBuildSignalController(
 // 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 {
+func (c *buildSignalController) Process(ctx context.Context, delivery consumer.Delivery) error {
 	msg := delivery.Message()
 
 	sig := &stovepipemq.BuildSignal{}
@@ -152,16 +153,16 @@ func (c *BuildSignalController) Process(ctx context.Context, delivery consumer.D
 }
 
 // Name returns the controller name for logging and metrics.
-func (c *BuildSignalController) Name() string {
-	return "buildsignal_dlq"
+func (c *buildSignalController) Name() string {
+	return string(c.topicKey)
 }
 
 // TopicKey returns the topic key this controller subscribes to.
-func (c *BuildSignalController) TopicKey() consumer.TopicKey {
+func (c *buildSignalController) TopicKey() consumer.TopicKey {
 	return c.topicKey
 }
 
 // ConsumerGroup returns the consumer group for offset tracking.
-func (c *BuildSignalController) ConsumerGroup() string {
+func (c *buildSignalController) ConsumerGroup() string {
 	return c.consumerGroup
 }

stovepipe/controller/dlq/buildsignal_test.go

@@ -38,7 +38,7 @@ type buildSignalDLQMocks struct {
 	buildStore *storagemock.MockBuildStore
 }
 
-func newBuildSignalController(t *testing.T, ctrl *gomock.Controller) (*BuildSignalController, buildSignalDLQMocks) {
+func newBuildSignalController(t *testing.T, ctrl *gomock.Controller) (consumer.Controller, buildSignalDLQMocks) {
 	t.Helper()
 
 	m := buildSignalDLQMocks{
@@ -52,7 +52,7 @@ func newBuildSignalController(t *testing.T, ctrl *gomock.Controller) (*BuildSign
 	store.EXPECT().GetQueueStore().Return(m.queueStore).AnyTimes()
 	store.EXPECT().GetBuildStore().Return(m.buildStore).AnyTimes()
 
-	c := NewBuildSignalController(
+	c := NewDLQBuildSignalController(
 		zap.NewNop().Sugar(),
 		tally.NewTestScope("test", nil),
 		staticStorageFactory{store: store},

stovepipe/controller/dlq/dlq_test.go

@@ -48,7 +48,7 @@ type staticStorageFactory struct{ store storage.Storage }
 // For returns the fixed store aggregate for any queue.
 func (f staticStorageFactory) For(storage.Config) (storage.Storage, error) { return f.store, nil }
 
-func newController(t *testing.T, ctrl *gomock.Controller) (*Controller, dlqMocks) {
+func newController(t *testing.T, ctrl *gomock.Controller) (consumer.Controller, dlqMocks) {
 	t.Helper()
 
 	m := dlqMocks{
@@ -60,7 +60,7 @@ func newController(t *testing.T, ctrl *gomock.Controller) (*Controller, dlqMocks
 	store.EXPECT().GetRequestStore().Return(m.reqStore).AnyTimes()
 	store.EXPECT().GetQueueStore().Return(m.queueStore).AnyTimes()
 
-	c := NewController(zap.NewNop().Sugar(), tally.NewTestScope("test", nil), staticStorageFactory{store: store}, TopicKey(stovepipemq.TopicKeyProcess), "stovepipe-process-dlq")
+	c := NewDLQRequestController(zap.NewNop().Sugar(), tally.NewTestScope("test", nil), staticStorageFactory{store: store}, TopicKey(stovepipemq.TopicKeyProcess), "stovepipe-process-dlq")
 	return c, m
 }
 

stovepipe/controller/dlq/request.go

@@ -26,36 +26,37 @@ import (
 	"go.uber.org/zap"
 )
 
-// Controller is the DLQ reconciler for the process stage. It is registered against the
+// requestController is the DLQ reconciler for the process stage. It is registered against the
 // process topic's DLQ (see TopicKey) and, on each delivery, decodes the request id from
 // the same ProcessRequest payload the primary process controller consumes, then drives
 // the referenced request to a terminal failed state via failRequest.
-type Controller struct {
+type requestController struct {
 	logger        *zap.SugaredLogger
 	metricsScope  tally.Scope
 	stores        storage.Factory
 	topicKey      consumer.TopicKey
 	consumerGroup string
 }
 
-// Verify Controller implements consumer.Controller at compile time.
-var _ consumer.Controller = (*Controller)(nil)
+// Verify requestController implements consumer.Controller at compile time.
+var _ consumer.Controller = (*requestController)(nil)
 
 // _opName is the metric operation name shared by every emit in this file.
 const _opName = "process_dlq"
 
-// NewController creates a new DLQ controller for the process stage's dead-letter topic.
+// NewDLQRequestController creates a DLQ controller for the process stage's dead-letter topic.
 // topicKey is typically dlq.TopicKey(stovepipemq.TopicKeyProcess).
-func NewController(
+func NewDLQRequestController(
 	logger *zap.SugaredLogger,
 	scope tally.Scope,
 	stores storage.Factory,
 	topicKey consumer.TopicKey,
 	consumerGroup string,
-) *Controller {
-	return &Controller{
-		logger:        logger.Named("process_dlq_controller"),
-		metricsScope:  scope.SubScope("process_dlq_controller"),
+) consumer.Controller {
+	name := string(topicKey) + "_controller"
+	return &requestController{
+		logger:        logger.Named(name),
+		metricsScope:  scope.SubScope(name),
 		stores:        stores,
 		topicKey:      topicKey,
 		consumerGroup: consumerGroup,
@@ -66,7 +67,7 @@ func NewController(
 // (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 *Controller) Process(ctx context.Context, delivery consumer.Delivery) error {
+func (c *requestController) Process(ctx context.Context, delivery consumer.Delivery) error {
 	msg := delivery.Message()
 
 	pr := &stovepipemq.ProcessRequest{}
@@ -113,16 +114,16 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 }
 
 // Name returns the controller name for logging and metrics.
-func (c *Controller) Name() string {
-	return "process_dlq"
+func (c *requestController) Name() string {
+	return string(c.topicKey)
 }
 
 // TopicKey returns the topic key this controller subscribes to.
-func (c *Controller) TopicKey() consumer.TopicKey {
+func (c *requestController) TopicKey() consumer.TopicKey {
 	return c.topicKey
 }
 
 // ConsumerGroup returns the consumer group for offset tracking.
-func (c *Controller) ConsumerGroup() string {
+func (c *requestController) ConsumerGroup() string {
 	return c.consumerGroup
 }