fix(pipeline): make DLQ required and simplify ConsumerGroup convention

Address review feedback:
- ConsumerGroup is now documented as service-wide (e.g. "orchestrator"),
  not per-stage (e.g. "orchestrator-start").
- DLQ handler is now required on every stage — the queue infrastructure
  always creates a DLQ topic, so not processing it silently accumulates
  messages.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: JamyDev

platform/pipeline/pipeline.go

@@ -45,8 +45,8 @@ type Stage[D any] struct {
 	// Used as the default when no TopicNames override is provided.
 	Name string
 
-	// ConsumerGroup is the consumer group suffix for this stage's subscription
-	// (e.g. "orchestrator-start").
+	// ConsumerGroup is the consumer group for this stage's subscription
+	// (e.g. "orchestrator").
 	ConsumerGroup string
 
 	// New builds the stage's controller from the service's Deps and engine-
@@ -55,10 +55,11 @@ type Stage[D any] struct {
 	// name on it, never mid-delivery.
 	New func(D, StageContext) (consumer.Controller, error)
 
-	// DLQ, when non-nil, declares "this stage dead-letters". The engine then
-	// derives the paired DLQ topic (<topic>_dlq, retry budget, DLQ-of-DLQ
-	// disabled) AND registers this reconciler on the DLQ consumer. Declaring
-	// one without getting the other is impossible — that's the invariant.
+	// DLQ builds the stage's dead-letter reconciler. Every stage must declare
+	// one: the queue infrastructure always creates a DLQ topic, so not
+	// processing it silently accumulates messages. The engine derives the
+	// paired DLQ topic (<topic>_dlq, retry budget, DLQ-of-DLQ disabled) and
+	// registers this reconciler on the DLQ consumer automatically.
 	DLQ func(D, StageContext) (consumer.Controller, error)
 }
 
@@ -179,10 +180,12 @@ func Construct[D any](
 	// Create the DLQ consumer with always-retryable processor.
 	dlq := consumer.New(logger, scope, registry, errs.AlwaysRetryableProcessor)
 
-	hasDLQ := false
-
 	// Eagerly construct and register all controllers.
 	for _, s := range stages {
+		if s.DLQ == nil {
+			return nil, fmt.Errorf("pipeline: stage %s: DLQ handler is required (every topic gets a DLQ)", s.Key)
+		}
+
 		sc := StageContext{
 			Registry:      registry,
 			TopicKey:      s.Key,
@@ -197,30 +200,25 @@ func Construct[D any](
 			return nil, fmt.Errorf("pipeline: stage %s: failed to register controller: %w", s.Key, err)
 		}
 
-		if s.DLQ != nil {
-			dlqSC := StageContext{
-				Registry:      registry,
-				TopicKey:      dlqTopicKey(s.Key),
-				ConsumerGroup: s.ConsumerGroup + "-dlq",
-			}
-			rec, err := s.DLQ(deps, dlqSC)
-			if err != nil {
-				return nil, fmt.Errorf("pipeline: stage %s dlq: failed to create controller: %w", s.Key, err)
-			}
-			if err := dlq.Register(rec); err != nil {
-				return nil, fmt.Errorf("pipeline: stage %s dlq: failed to register controller: %w", s.Key, err)
-			}
-			hasDLQ = true
+		dlqSC := StageContext{
+			Registry:      registry,
+			TopicKey:      dlqTopicKey(s.Key),
+			ConsumerGroup: s.ConsumerGroup + "-dlq",
+		}
+		rec, err := s.DLQ(deps, dlqSC)
+		if err != nil {
+			return nil, fmt.Errorf("pipeline: stage %s dlq: failed to create controller: %w", s.Key, err)
+		}
+		if err := dlq.Register(rec); err != nil {
+			return nil, fmt.Errorf("pipeline: stage %s dlq: failed to register controller: %w", s.Key, err)
 		}
 	}
 
 	// Compose the lifecycle group.
 	members := make([]lifecycle.Component, 0, len(o.extraComponents)+2)
 	members = append(members, o.extraComponents...)
 	members = append(members, &consumerComponent{name: "primary", c: primary})
-	if hasDLQ {
-		members = append(members, &consumerComponent{name: "dlq", c: dlq})
-	}
+	members = append(members, &consumerComponent{name: "dlq", c: dlq})
 
 	return lifecycle.NewGroup(members...), nil
 }
@@ -232,7 +230,7 @@ func buildTopicConfigs[D any](
 	stages []Stage[D],
 	o *options,
 ) ([]consumer.TopicConfig, error) {
-	// Pre-size: each stage gets a primary config + optional DLQ config,
+	// Pre-size: each stage gets a primary config + DLQ config,
 	// plus publish-only topics.
 	configs := make([]consumer.TopicConfig, 0, 2*len(stages)+len(o.publishOnly))
 
@@ -248,16 +246,14 @@ func buildTopicConfigs[D any](
 			),
 		})
 
-		if s.DLQ != nil {
-			configs = append(configs, consumer.TopicConfig{
-				Key:   dlqTopicKey(s.Key),
-				Name:  topicName + dlqTopicSuffix,
-				Queue: queue,
-				Subscription: extqueue.DLQSubscriptionConfig(
-					subscriberName, s.ConsumerGroup+"-dlq",
-				),
-			})
-		}
+		configs = append(configs, consumer.TopicConfig{
+			Key:   dlqTopicKey(s.Key),
+			Name:  topicName + dlqTopicSuffix,
+			Queue: queue,
+			Subscription: extqueue.DLQSubscriptionConfig(
+				subscriberName, s.ConsumerGroup+"-dlq",
+			),
+		})
 	}
 
 	for _, p := range o.publishOnly {

platform/pipeline/pipeline_test.go

@@ -50,7 +50,7 @@ func newTestLogger() *zap.SugaredLogger {
 	return l.Sugar()
 }
 
-func TestConstruct_SingleStage_NoDLQ(t *testing.T) {
+func TestConstruct_SingleStage(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	q := mqmock.NewMockQueue(ctrl)
 	q.EXPECT().Subscriber().Return(mqmock.NewMockSubscriber(ctrl)).AnyTimes()
@@ -61,10 +61,13 @@ func TestConstruct_SingleStage_NoDLQ(t *testing.T) {
 		{
 			Key:           "start",
 			Name:          "start",
-			ConsumerGroup: "orchestrator-start",
+			ConsumerGroup: "orchestrator",
 			New: func(d testDeps, sc StageContext) (consumer.Controller, error) {
 				return &fakeController{key: sc.TopicKey, group: sc.ConsumerGroup}, nil
 			},
+			DLQ: func(d testDeps, sc StageContext) (consumer.Controller, error) {
+				return &fakeController{key: sc.TopicKey, group: sc.ConsumerGroup}, nil
+			},
 		},
 	}
 
@@ -84,7 +87,7 @@ func TestConstruct_WithDLQ(t *testing.T) {
 		{
 			Key:           "start",
 			Name:          "start",
-			ConsumerGroup: "orchestrator-start",
+			ConsumerGroup: "orchestrator",
 			New: func(d testDeps, sc StageContext) (consumer.Controller, error) {
 				return &fakeController{key: sc.TopicKey, group: sc.ConsumerGroup}, nil
 			},
@@ -99,6 +102,30 @@ func TestConstruct_WithDLQ(t *testing.T) {
 	assert.NotNil(t, comp)
 }
 
+func TestConstruct_NilDLQ_Error(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	q := mqmock.NewMockQueue(ctrl)
+	q.EXPECT().Subscriber().Return(mqmock.NewMockSubscriber(ctrl)).AnyTimes()
+	q.EXPECT().Publisher().Return(mqmock.NewMockPublisher(ctrl)).AnyTimes()
+
+	deps := testDeps{logger: newTestLogger()}
+	stages := []Stage[testDeps]{
+		{
+			Key:           "start",
+			Name:          "start",
+			ConsumerGroup: "orchestrator",
+			New: func(d testDeps, sc StageContext) (consumer.Controller, error) {
+				return &fakeController{key: sc.TopicKey, group: sc.ConsumerGroup}, nil
+			},
+			// DLQ intentionally omitted.
+		},
+	}
+
+	_, err := Construct(deps.logger, tally.NoopScope, q, "test-sub", deps, stages)
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "DLQ handler is required")
+}
+
 func TestConstruct_MultipleStages(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	q := mqmock.NewMockQueue(ctrl)
@@ -110,15 +137,18 @@ func TestConstruct_MultipleStages(t *testing.T) {
 		{
 			Key:           "start",
 			Name:          "start",
-			ConsumerGroup: "orchestrator-start",
+			ConsumerGroup: "orchestrator",
 			New: func(d testDeps, sc StageContext) (consumer.Controller, error) {
 				return &fakeController{key: sc.TopicKey, group: sc.ConsumerGroup}, nil
 			},
+			DLQ: func(d testDeps, sc StageContext) (consumer.Controller, error) {
+				return &fakeController{key: sc.TopicKey, group: sc.ConsumerGroup}, nil
+			},
 		},
 		{
 			Key:           "validate",
 			Name:          "validate",
-			ConsumerGroup: "orchestrator-validate",
+			ConsumerGroup: "orchestrator",
 			New: func(d testDeps, sc StageContext) (consumer.Controller, error) {
 				return &fakeController{key: sc.TopicKey, group: sc.ConsumerGroup}, nil
 			},
@@ -154,10 +184,13 @@ func TestConstruct_ControllerCreationFailure(t *testing.T) {
 		{
 			Key:           "start",
 			Name:          "start",
-			ConsumerGroup: "orchestrator-start",
+			ConsumerGroup: "orchestrator",
 			New: func(d testDeps, sc StageContext) (consumer.Controller, error) {
 				return nil, fmt.Errorf("missing dependency")
 			},
+			DLQ: func(d testDeps, sc StageContext) (consumer.Controller, error) {
+				return &fakeController{key: sc.TopicKey, group: sc.ConsumerGroup}, nil
+			},
 		},
 	}
 
@@ -178,7 +211,7 @@ func TestConstruct_DLQControllerCreationFailure(t *testing.T) {
 		{
 			Key:           "start",
 			Name:          "start",
-			ConsumerGroup: "orchestrator-start",
+			ConsumerGroup: "orchestrator",
 			New: func(d testDeps, sc StageContext) (consumer.Controller, error) {
 				return &fakeController{key: sc.TopicKey, group: sc.ConsumerGroup}, nil
 			},
@@ -205,10 +238,13 @@ func TestConstruct_WithPublishOnly(t *testing.T) {
 		{
 			Key:           "start",
 			Name:          "start",
-			ConsumerGroup: "orchestrator-start",
+			ConsumerGroup: "orchestrator",
 			New: func(d testDeps, sc StageContext) (consumer.Controller, error) {
 				return &fakeController{key: sc.TopicKey, group: sc.ConsumerGroup}, nil
 			},
+			DLQ: func(d testDeps, sc StageContext) (consumer.Controller, error) {
+				return &fakeController{key: sc.TopicKey, group: sc.ConsumerGroup}, nil
+			},
 		},
 	}
 
@@ -233,10 +269,13 @@ func TestConstruct_WithTopicNameOverrides(t *testing.T) {
 		{
 			Key:           "start",
 			Name:          "start",
-			ConsumerGroup: "orchestrator-start",
+			ConsumerGroup: "orchestrator",
 			New: func(d testDeps, sc StageContext) (consumer.Controller, error) {
 				return &fakeController{key: sc.TopicKey, group: sc.ConsumerGroup}, nil
 			},
+			DLQ: func(d testDeps, sc StageContext) (consumer.Controller, error) {
+				return &fakeController{key: sc.TopicKey, group: sc.ConsumerGroup}, nil
+			},
 		},
 	}
 
@@ -262,7 +301,7 @@ func TestConstruct_StageContext_Populated(t *testing.T) {
 		{
 			Key:           "start",
 			Name:          "start",
-			ConsumerGroup: "orchestrator-start",
+			ConsumerGroup: "orchestrator",
 			New: func(d testDeps, sc StageContext) (consumer.Controller, error) {
 				primarySC = sc
 				return &fakeController{key: sc.TopicKey, group: sc.ConsumerGroup}, nil
@@ -279,11 +318,11 @@ func TestConstruct_StageContext_Populated(t *testing.T) {
 
 	// Primary StageContext should have the stage's own key and group.
 	assert.Equal(t, consumer.TopicKey("start"), primarySC.TopicKey)
-	assert.Equal(t, "orchestrator-start", primarySC.ConsumerGroup)
+	assert.Equal(t, "orchestrator", primarySC.ConsumerGroup)
 
 	// DLQ StageContext should have the derived DLQ key and group.
 	assert.Equal(t, consumer.TopicKey("start_dlq"), dlqSC.TopicKey)
-	assert.Equal(t, "orchestrator-start-dlq", dlqSC.ConsumerGroup)
+	assert.Equal(t, "orchestrator-dlq", dlqSC.ConsumerGroup)
 
 	// Both should share the same registry.
 	assert.Equal(t, primarySC.Registry, dlqSC.Registry)
@@ -347,16 +386,16 @@ func TestBuildTopicConfigs(t *testing.T) {
 		{
 			Key:           "start",
 			Name:          "start",
-			ConsumerGroup: "orchestrator-start",
+			ConsumerGroup: "orchestrator",
 			New:           func(d testDeps, sc StageContext) (consumer.Controller, error) { return nil, nil },
 			DLQ:           func(d testDeps, sc StageContext) (consumer.Controller, error) { return nil, nil },
 		},
 		{
 			Key:           "validate",
 			Name:          "validate",
-			ConsumerGroup: "orchestrator-validate",
+			ConsumerGroup: "orchestrator",
 			New:           func(d testDeps, sc StageContext) (consumer.Controller, error) { return nil, nil },
-			// No DLQ for this stage.
+			DLQ:           func(d testDeps, sc StageContext) (consumer.Controller, error) { return nil, nil },
 		},
 	}
 
@@ -369,29 +408,30 @@ func TestBuildTopicConfigs(t *testing.T) {
 	configs, err := buildTopicConfigs(q, "test-sub", stages, o)
 	require.NoError(t, err)
 
-	// Expected: start (primary + DLQ) + validate (primary only) + log (publish-only) = 4
-	assert.Len(t, configs, 4)
+	// Expected: start (primary + DLQ) + validate (primary + DLQ) + log (publish-only) = 5
+	assert.Len(t, configs, 5)
 
 	// Verify primary stage config.
 	assert.Equal(t, consumer.TopicKey("start"), configs[0].Key)
 	assert.Equal(t, "start", configs[0].Name)
-	assert.Equal(t, "orchestrator-start", configs[0].Subscription.ConsumerGroup)
+	assert.Equal(t, "orchestrator", configs[0].Subscription.ConsumerGroup)
 
 	// Verify DLQ config derived from primary.
 	assert.Equal(t, consumer.TopicKey("start_dlq"), configs[1].Key)
 	assert.Equal(t, "start_dlq", configs[1].Name)
-	assert.Equal(t, "orchestrator-start-dlq", configs[1].Subscription.ConsumerGroup)
+	assert.Equal(t, "orchestrator-dlq", configs[1].Subscription.ConsumerGroup)
 
 	// Verify DLQ subscription has DLQ disabled (no cascade).
-	expected := extqueue.DLQSubscriptionConfig("test-sub", "orchestrator-start-dlq")
+	expected := extqueue.DLQSubscriptionConfig("test-sub", "orchestrator-dlq")
 	assert.Equal(t, expected.DLQ.Enabled, configs[1].Subscription.DLQ.Enabled)
 	assert.Equal(t, expected.Retry.MaxAttempts, configs[1].Subscription.Retry.MaxAttempts)
 
-	// Verify validate stage (no DLQ).
+	// Verify validate stage (primary + DLQ).
 	assert.Equal(t, consumer.TopicKey("validate"), configs[2].Key)
+	assert.Equal(t, consumer.TopicKey("validate_dlq"), configs[3].Key)
 
 	// Verify publish-only topic.
-	assert.Equal(t, consumer.TopicKey("log"), configs[3].Key)
-	assert.Equal(t, "log", configs[3].Name)
-	assert.Equal(t, "", configs[3].Subscription.ConsumerGroup)
+	assert.Equal(t, consumer.TopicKey("log"), configs[4].Key)
+	assert.Equal(t, "log", configs[4].Name)
+	assert.Equal(t, "", configs[4].Subscription.ConsumerGroup)
 }