feat(stovepipe): recover record DLQ work

service/stovepipe/server/main.go

@@ -282,15 +282,15 @@ func run() error {
 	// Each factory is constructed once and threaded through every consumer of
 	// it, so a real (stateful) backend introduced later is shared rather than
 	// silently duplicated across controllers.
-	scf := fakeSourceControlFactory{}
+	sourceControl := fakeSourceControlFactory{}
 	brf := fakeBuildRunnerFactory{}
 
 	storageFty := storageFactory{backend: store}
-	primaryCount, err := registerPrimaryControllers(primaryConsumer, logger.Sugar(), scope, storageFty, registry, scf, brf)
+	primaryCount, err := registerPrimaryControllers(primaryConsumer, logger.Sugar(), scope, storageFty, registry, sourceControl, brf)
 	if err != nil {
 		return err
 	}
-	dlqCount, err := registerDLQControllers(dlqConsumer, logger.Sugar(), scope, storageFty, registry)
+	dlqCount, err := registerDLQControllers(dlqConsumer, logger.Sugar(), scope, storageFty, registry, sourceControl)
 	if err != nil {
 		return err
 	}
@@ -318,7 +318,7 @@ func run() error {
 		logger.Sugar(),
 		scope,
 		newInMemoryCounterFactory(),
-		scf,
+		sourceControl,
 		storageFty,
 		registry,
 	)
@@ -398,7 +398,7 @@ func registerPrimaryControllers(
 	scope tally.Scope,
 	store storage.Factory,
 	registry consumer.TopicRegistry,
-	scf sourcecontrol.Factory,
+	sourceControl sourcecontrol.Factory,
 	brf buildrunner.Factory,
 ) (int, error) {
 	var count int
@@ -408,7 +408,7 @@ func registerPrimaryControllers(
 		scope,
 		store,
 		queueconfigdefault.NewStore(),
-		scf,
+		sourceControl,
 		registry,
 		stovepipemq.TopicKeyProcess,
 		"stovepipe-process",
@@ -430,7 +430,7 @@ func registerPrimaryControllers(
 	}
 	count++
 
-	recordController := record.NewController(logger, scope, store, scf, stovepipemq.TopicKeyRecord, "stovepipe-record")
+	recordController := record.NewController(logger, scope, store, sourceControl, stovepipemq.TopicKeyRecord, "stovepipe-record")
 	if err := c.Register(recordController); err != nil {
 		return count, fmt.Errorf("failed to register record controller: %w", err)
 	}
@@ -447,6 +447,7 @@ func registerDLQControllers(
 	scope tally.Scope,
 	store storage.Factory,
 	registry consumer.TopicRegistry,
+	sourceControl sourcecontrol.Factory,
 ) (int, error) {
 	var count int
 
@@ -468,6 +469,12 @@ func registerDLQControllers(
 	}
 	count++
 
+	recordDLQController := record.NewController(logger, scope, store, sourceControl, dlq.TopicKey(stovepipemq.TopicKeyRecord), "stovepipe-record-dlq")
+	if err := c.Register(recordDLQController); err != nil {
+		return count, fmt.Errorf("failed to register record dlq controller: %w", err)
+	}
+	count++
+
 	return count, nil
 }
 
@@ -529,6 +536,12 @@ func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRe
 			Queue:        q,
 			Subscription: extqueue.DLQSubscriptionConfig(subscriberName, "stovepipe-buildsignal-dlq"),
 		},
+		{
+			Key:          dlq.TopicKey(stovepipemq.TopicKeyRecord),
+			Name:         "record_dlq",
+			Queue:        q,
+			Subscription: extqueue.DLQSubscriptionConfig(subscriberName, "stovepipe-record-dlq"),
+		},
 	})
 }
 

stovepipe/controller/record/BUILD.bazel

@@ -24,6 +24,7 @@ go_test(
     embed = [":go_default_library"],
     deps = [
         "//platform/base/messagequeue:go_default_library",
+        "//platform/consumer:go_default_library",
         "//platform/consumer/mock:go_default_library",
         "//stovepipe/core/messagequeue:go_default_library",
         "//stovepipe/entity:go_default_library",

stovepipe/controller/record/record.go

@@ -44,12 +44,12 @@ import (
 // when that fact is green advances the queue's last-green bookmark and promotes
 // the commit. Implements consumer.Controller.
 type Controller struct {
-	logger         *zap.SugaredLogger
-	metricsScope   tally.Scope
-	stores         storage.Factory
-	sourceControls sourcecontrol.Factory
-	topicKey       consumer.TopicKey
-	consumerGroup  string
+	logger        *zap.SugaredLogger
+	metricsScope  tally.Scope
+	stores        storage.Factory
+	sourceControl sourcecontrol.Factory
+	topicKey      consumer.TopicKey
+	consumerGroup string
 }
 
 // Verify Controller implements consumer.Controller interface at compile time.
@@ -68,17 +68,18 @@ func NewController(
 	logger *zap.SugaredLogger,
 	scope tally.Scope,
 	stores storage.Factory,
-	sourceControls sourcecontrol.Factory,
+	sourceControl sourcecontrol.Factory,
 	topicKey consumer.TopicKey,
 	consumerGroup string,
 ) *Controller {
+	name := string(topicKey) + "_controller"
 	return &Controller{
-		logger:         logger.Named("record_controller"),
-		metricsScope:   scope.SubScope("record_controller"),
-		stores:         stores,
-		sourceControls: sourceControls,
-		topicKey:       topicKey,
-		consumerGroup:  consumerGroup,
+		logger:        logger.Named(name),
+		metricsScope:  scope.SubScope(name),
+		stores:        stores,
+		sourceControl: sourceControl,
+		topicKey:      topicKey,
+		consumerGroup: consumerGroup,
 	}
 }
 
@@ -246,7 +247,7 @@ func (c *Controller) reportFailureDetectionLatency(ctx context.Context, request
 		return
 	}
 
-	sourceControl, err := c.sourceControls.For(sourcecontrol.Config{QueueName: request.Queue})
+	sourceControl, err := c.sourceControl.For(sourcecontrol.Config{QueueName: request.Queue})
 	if err != nil {
 		c.failureDetectionUnobserved(request, "resolve_source_control", err)
 		return
@@ -365,7 +366,7 @@ func (c *Controller) advanceLastGreen(ctx context.Context, store storage.Storage
 func (c *Controller) emitLastGreenTimestamp(ctx context.Context, request entity.Request) {
 	queueTag := metrics.NewTag("queue", request.Queue)
 
-	sourceControl, err := c.sourceControls.For(sourcecontrol.Config{QueueName: request.Queue})
+	sourceControl, err := c.sourceControl.For(sourcecontrol.Config{QueueName: request.Queue})
 	if err != nil {
 		metrics.NamedCounter(c.metricsScope, _opName, "last_green_timestamp_resolve_errors", 1, queueTag)
 		c.logger.Warnw("failed to resolve source control to report the last green timestamp",
@@ -420,7 +421,7 @@ func (c *Controller) emitLastGreenTimestamp(ctx context.Context, request entity.
 // harmlessly. A commit that a rewritten history dropped from the ref cannot be
 // promoted by any retry, so that case is counted and skipped rather than failed.
 func (c *Controller) promote(ctx context.Context, request entity.Request) error {
-	sc, err := c.sourceControls.For(sourcecontrol.Config{QueueName: request.Queue})
+	sc, err := c.sourceControl.For(sourcecontrol.Config{QueueName: request.Queue})
 	if err != nil {
 		metrics.NamedCounter(c.metricsScope, _opName, "source_control_errors", 1,
 			metrics.NewTag("stage", "resolve"),
@@ -477,7 +478,7 @@ func (c *Controller) loadRequest(ctx context.Context, store storage.Storage, id
 
 // Name returns the controller name for logging and metrics.
 func (c *Controller) Name() string {
-	return "record"
+	return string(c.topicKey)
 }
 
 // TopicKey returns the topic key this controller subscribes to.

stovepipe/controller/record/record_test.go

@@ -24,6 +24,7 @@ import (
 	"github.com/stretchr/testify/require"
 	"github.com/uber-go/tally"
 	entityqueue "github.com/uber/submitqueue/platform/base/messagequeue"
+	"github.com/uber/submitqueue/platform/consumer"
 	consumermock "github.com/uber/submitqueue/platform/consumer/mock"
 	stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue"
 	"github.com/uber/submitqueue/stovepipe/entity"
@@ -94,6 +95,10 @@ func (failingSourceControlFactory) For(sourcecontrol.Config) (sourcecontrol.Sour
 }
 
 func newController(t *testing.T, ctrl *gomock.Controller) (*Controller, recordMocks) {
+	return newControllerForTopic(t, ctrl, stovepipemq.TopicKeyRecord, "stovepipe-record")
+}
+
+func newControllerForTopic(t *testing.T, ctrl *gomock.Controller, topicKey consumer.TopicKey, consumerGroup string) (*Controller, recordMocks) {
 	t.Helper()
 
 	scope := tally.NewTestScope("", nil)
@@ -115,12 +120,20 @@ func newController(t *testing.T, ctrl *gomock.Controller) (*Controller, recordMo
 		scope,
 		staticStorageFactory{store: store},
 		staticSourceControlFactory{sourceControl: m.sourceControl},
-		stovepipemq.TopicKeyRecord,
-		"stovepipe-record",
+		topicKey,
+		consumerGroup,
 	)
 	return c, m
 }
 
+func TestControllerIdentity(t *testing.T) {
+	c, _ := newControllerForTopic(t, gomock.NewController(t), consumer.TopicKey("record_dlq"), "stovepipe-record-dlq")
+
+	assert.Equal(t, "record_dlq", c.Name())
+	assert.Equal(t, consumer.TopicKey("record_dlq"), c.TopicKey())
+	assert.Equal(t, "stovepipe-record-dlq", c.ConsumerGroup())
+}
+
 func delivery(t *testing.T, ctrl *gomock.Controller, payload []byte) *consumermock.MockDelivery {
 	t.Helper()
 	d := consumermock.NewMockDelivery(ctrl)
@@ -283,7 +296,7 @@ func TestProcess_TimestampReportingFailureDoesNotFailRecord(t *testing.T) {
 func TestProcess_UnresolvableSourceControlCountsTimestampFailure(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	c, m := newController(t, ctrl)
-	c.sourceControls = failingSourceControlFactory{}
+	c.sourceControl = failingSourceControlFactory{}
 
 	m.reqStore.EXPECT().Get(gomock.Any(), testID).
 		Return(requestWithState(entity.RequestStateSucceeded), nil)
@@ -385,7 +398,7 @@ func TestProcess_UnobservableDetectionLatencyDoesNotFailRecord(t *testing.T) {
 			name: "source control cannot be resolved",
 			step: "resolve_source_control",
 			setup: func(c *Controller, _ recordMocks) {
-				c.sourceControls = failingSourceControlFactory{}
+				c.sourceControl = failingSourceControlFactory{}
 			},
 		},
 		{
@@ -566,7 +579,7 @@ func TestProcess_PromotionErrorsPropagate(t *testing.T) {
 		{
 			name: "source control resolve fails",
 			setup: func(c *Controller, _ recordMocks) {
-				c.sourceControls = failingSourceControlFactory{}
+				c.sourceControl = failingSourceControlFactory{}
 			},
 		},
 		{