feat(orchestrator): add pipeline self-declaration

platform/pipeline/pipeline.go

@@ -49,16 +49,33 @@ type Stage[D any] struct {
 	// (e.g. "orchestrator-start").
 	ConsumerGroup string
 
-	// New builds the stage's controller from the service's Deps. The engine
-	// calls it once, eagerly, inside Construct — so a nil/missing dependency
-	// fails at boot with the stage's name on it, never mid-delivery.
-	New func(D) (consumer.Controller, error)
+	// New builds the stage's controller from the service's Deps and engine-
+	// provided StageContext. The engine calls it once, eagerly, inside
+	// Construct — so a nil/missing dependency fails at boot with the stage's
+	// 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 func(D) (consumer.Controller, error)
+	DLQ func(D, StageContext) (consumer.Controller, error)
+}
+
+// StageContext carries engine-produced values that controllers need at
+// construction time but the host does not own: the assembled topic
+// registry (for publishing to downstream stages), the stage's own topic
+// key, and its consumer group.
+type StageContext struct {
+	// Registry is the fully assembled TopicRegistry. Controllers use it to
+	// look up topic names and queue backends for publishing downstream.
+	Registry consumer.TopicRegistry
+
+	// TopicKey is this stage's logical topic key.
+	TopicKey consumer.TopicKey
+
+	// ConsumerGroup is this stage's consumer group name.
+	ConsumerGroup string
 }
 
 // PublishOnlyTopic declares a topic the service publishes to but does not
@@ -166,7 +183,13 @@ func Construct[D any](
 
 	// Eagerly construct and register all controllers.
 	for _, s := range stages {
-		ctl, err := s.New(deps)
+		sc := StageContext{
+			Registry:      registry,
+			TopicKey:      s.Key,
+			ConsumerGroup: s.ConsumerGroup,
+		}
+
+		ctl, err := s.New(deps, sc)
 		if err != nil {
 			return nil, fmt.Errorf("pipeline: stage %s: failed to create controller: %w", s.Key, err)
 		}
@@ -175,7 +198,12 @@ func Construct[D any](
 		}
 
 		if s.DLQ != nil {
-			rec, err := s.DLQ(deps)
+			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)
 			}

platform/pipeline/pipeline_test.go

@@ -62,8 +62,8 @@ func TestConstruct_SingleStage_NoDLQ(t *testing.T) {
 			Key:           "start",
 			Name:          "start",
 			ConsumerGroup: "orchestrator-start",
-			New: func(d testDeps) (consumer.Controller, error) {
-				return &fakeController{key: "start", group: "orchestrator-start"}, nil
+			New: func(d testDeps, sc StageContext) (consumer.Controller, error) {
+				return &fakeController{key: sc.TopicKey, group: sc.ConsumerGroup}, nil
 			},
 		},
 	}
@@ -85,11 +85,11 @@ func TestConstruct_WithDLQ(t *testing.T) {
 			Key:           "start",
 			Name:          "start",
 			ConsumerGroup: "orchestrator-start",
-			New: func(d testDeps) (consumer.Controller, error) {
-				return &fakeController{key: "start", group: "orchestrator-start"}, nil
+			New: func(d testDeps, sc StageContext) (consumer.Controller, error) {
+				return &fakeController{key: sc.TopicKey, group: sc.ConsumerGroup}, nil
 			},
-			DLQ: func(d testDeps) (consumer.Controller, error) {
-				return &fakeController{key: "start_dlq", group: "orchestrator-start-dlq"}, nil
+			DLQ: func(d testDeps, sc StageContext) (consumer.Controller, error) {
+				return &fakeController{key: sc.TopicKey, group: sc.ConsumerGroup}, nil
 			},
 		},
 	}
@@ -111,19 +111,19 @@ func TestConstruct_MultipleStages(t *testing.T) {
 			Key:           "start",
 			Name:          "start",
 			ConsumerGroup: "orchestrator-start",
-			New: func(d testDeps) (consumer.Controller, error) {
-				return &fakeController{key: "start", group: "orchestrator-start"}, nil
+			New: func(d testDeps, sc StageContext) (consumer.Controller, error) {
+				return &fakeController{key: sc.TopicKey, group: sc.ConsumerGroup}, nil
 			},
 		},
 		{
 			Key:           "validate",
 			Name:          "validate",
 			ConsumerGroup: "orchestrator-validate",
-			New: func(d testDeps) (consumer.Controller, error) {
-				return &fakeController{key: "validate", group: "orchestrator-validate"}, nil
+			New: func(d testDeps, sc StageContext) (consumer.Controller, error) {
+				return &fakeController{key: sc.TopicKey, group: sc.ConsumerGroup}, nil
 			},
-			DLQ: func(d testDeps) (consumer.Controller, error) {
-				return &fakeController{key: "validate_dlq", group: "orchestrator-validate-dlq"}, nil
+			DLQ: func(d testDeps, sc StageContext) (consumer.Controller, error) {
+				return &fakeController{key: sc.TopicKey, group: sc.ConsumerGroup}, nil
 			},
 		},
 	}
@@ -155,7 +155,7 @@ func TestConstruct_ControllerCreationFailure(t *testing.T) {
 			Key:           "start",
 			Name:          "start",
 			ConsumerGroup: "orchestrator-start",
-			New: func(d testDeps) (consumer.Controller, error) {
+			New: func(d testDeps, sc StageContext) (consumer.Controller, error) {
 				return nil, fmt.Errorf("missing dependency")
 			},
 		},
@@ -179,10 +179,10 @@ func TestConstruct_DLQControllerCreationFailure(t *testing.T) {
 			Key:           "start",
 			Name:          "start",
 			ConsumerGroup: "orchestrator-start",
-			New: func(d testDeps) (consumer.Controller, error) {
-				return &fakeController{key: "start", group: "orchestrator-start"}, nil
+			New: func(d testDeps, sc StageContext) (consumer.Controller, error) {
+				return &fakeController{key: sc.TopicKey, group: sc.ConsumerGroup}, nil
 			},
-			DLQ: func(d testDeps) (consumer.Controller, error) {
+			DLQ: func(d testDeps, sc StageContext) (consumer.Controller, error) {
 				return nil, fmt.Errorf("dlq dependency missing")
 			},
 		},
@@ -206,8 +206,8 @@ func TestConstruct_WithPublishOnly(t *testing.T) {
 			Key:           "start",
 			Name:          "start",
 			ConsumerGroup: "orchestrator-start",
-			New: func(d testDeps) (consumer.Controller, error) {
-				return &fakeController{key: "start", group: "orchestrator-start"}, nil
+			New: func(d testDeps, sc StageContext) (consumer.Controller, error) {
+				return &fakeController{key: sc.TopicKey, group: sc.ConsumerGroup}, nil
 			},
 		},
 	}
@@ -234,8 +234,8 @@ func TestConstruct_WithTopicNameOverrides(t *testing.T) {
 			Key:           "start",
 			Name:          "start",
 			ConsumerGroup: "orchestrator-start",
-			New: func(d testDeps) (consumer.Controller, error) {
-				return &fakeController{key: "start", group: "orchestrator-start"}, nil
+			New: func(d testDeps, sc StageContext) (consumer.Controller, error) {
+				return &fakeController{key: sc.TopicKey, group: sc.ConsumerGroup}, nil
 			},
 		},
 	}
@@ -249,6 +249,46 @@ func TestConstruct_WithTopicNameOverrides(t *testing.T) {
 	assert.NotNil(t, comp)
 }
 
+func TestConstruct_StageContext_Populated(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()}
+
+	var primarySC, dlqSC StageContext
+	stages := []Stage[testDeps]{
+		{
+			Key:           "start",
+			Name:          "start",
+			ConsumerGroup: "orchestrator-start",
+			New: func(d testDeps, sc StageContext) (consumer.Controller, error) {
+				primarySC = sc
+				return &fakeController{key: sc.TopicKey, group: sc.ConsumerGroup}, nil
+			},
+			DLQ: func(d testDeps, sc StageContext) (consumer.Controller, error) {
+				dlqSC = sc
+				return &fakeController{key: sc.TopicKey, group: sc.ConsumerGroup}, nil
+			},
+		},
+	}
+
+	_, err := Construct(deps.logger, tally.NoopScope, q, "test-sub", deps, stages)
+	require.NoError(t, err)
+
+	// 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)
+
+	// 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)
+
+	// Both should share the same registry.
+	assert.Equal(t, primarySC.Registry, dlqSC.Registry)
+}
+
 func TestResolveTopicName(t *testing.T) {
 	tests := []struct {
 		name      string
@@ -308,14 +348,14 @@ func TestBuildTopicConfigs(t *testing.T) {
 			Key:           "start",
 			Name:          "start",
 			ConsumerGroup: "orchestrator-start",
-			New:           func(d testDeps) (consumer.Controller, error) { return nil, nil },
-			DLQ:           func(d testDeps) (consumer.Controller, error) { return nil, nil },
+			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",
-			New:           func(d testDeps) (consumer.Controller, error) { return nil, nil },
+			New:           func(d testDeps, sc StageContext) (consumer.Controller, error) { return nil, nil },
 			// No DLQ for this stage.
 		},
 	}

submitqueue/orchestrator/BUILD.bazel

@@ -0,0 +1,37 @@
+load("@rules_go//go:def.bzl", "go_library")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["pipeline.go"],
+    importpath = "github.com/uber/submitqueue/submitqueue/orchestrator",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//api/runway/messagequeue:go_default_library",
+        "//platform/consumer:go_default_library",
+        "//platform/extension/counter:go_default_library",
+        "//platform/pipeline:go_default_library",
+        "//submitqueue/core/topickey:go_default_library",
+        "//submitqueue/extension/buildrunner:go_default_library",
+        "//submitqueue/extension/changeprovider:go_default_library",
+        "//submitqueue/extension/conflict:go_default_library",
+        "//submitqueue/extension/scorer:go_default_library",
+        "//submitqueue/extension/storage:go_default_library",
+        "//submitqueue/extension/validator:go_default_library",
+        "//submitqueue/orchestrator/controller:go_default_library",
+        "//submitqueue/orchestrator/controller/batch:go_default_library",
+        "//submitqueue/orchestrator/controller/build:go_default_library",
+        "//submitqueue/orchestrator/controller/buildsignal:go_default_library",
+        "//submitqueue/orchestrator/controller/cancel:go_default_library",
+        "//submitqueue/orchestrator/controller/conclude:go_default_library",
+        "//submitqueue/orchestrator/controller/dlq:go_default_library",
+        "//submitqueue/orchestrator/controller/merge:go_default_library",
+        "//submitqueue/orchestrator/controller/mergeconflictsignal:go_default_library",
+        "//submitqueue/orchestrator/controller/mergesignal:go_default_library",
+        "//submitqueue/orchestrator/controller/score:go_default_library",
+        "//submitqueue/orchestrator/controller/speculate:go_default_library",
+        "//submitqueue/orchestrator/controller/start:go_default_library",
+        "//submitqueue/orchestrator/controller/validate:go_default_library",
+        "@com_github_uber_go_tally//:go_default_library",
+        "@org_uber_go_zap//:go_default_library",
+    ],
+)