From 98a520f10c88078d52788a0cf107186d6d52172d Mon Sep 17 00:00:00 2001 From: jamy Date: Mon, 20 Jul 2026 19:02:37 +0000 Subject: [PATCH 1/2] feat(platform): add StageContext to pipeline.Stage constructors MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Pass engine-produced values (TopicRegistry, TopicKey, ConsumerGroup) to Stage.New and Stage.DLQ via StageContext. Controllers need the registry for publishing to downstream stages and the topic key/group for their own identity — these are engine-owned, not host-provided Deps. DLQ StageContext automatically derives the DLQ topic key (appending "_dlq") and consumer group (appending "-dlq") from the primary stage's values. Co-Authored-By: Claude Opus 4.6 --- platform/pipeline/pipeline.go | 42 ++++++++++++--- platform/pipeline/pipeline_test.go | 86 ++++++++++++++++++++++-------- 2 files changed, 98 insertions(+), 30 deletions(-) diff --git a/platform/pipeline/pipeline.go b/platform/pipeline/pipeline.go index a4ff8c5e..e3a73538 100644 --- a/platform/pipeline/pipeline.go +++ b/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 (_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) } diff --git a/platform/pipeline/pipeline_test.go b/platform/pipeline/pipeline_test.go index 9280b145..a5412ce7 100644 --- a/platform/pipeline/pipeline_test.go +++ b/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. }, } From 538faeaeec50e5049c037fc4f399978fd8aa612b Mon Sep 17 00:00:00 2001 From: jamy Date: Mon, 20 Jul 2026 19:05:50 +0000 Subject: [PATCH 2/2] feat(orchestrator): add pipeline self-declaration MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Introduce `submitqueue/orchestrator/pipeline.go` with three exports: - `Deps` struct: the full set of dependencies (logger, scope, storage, counter, and four extension factories) the orchestrator pipeline needs. This struct IS the service's public API toward deployers. - `Stages` slice: the complete 12-stage pipeline topology as a typed table of `pipeline.Stage[Deps]`. Each row declares a primary controller constructor and its DLQ reconciler. Adding a stage = adding one row. - `PublishOnlyTopics`: topics the orchestrator publishes to but does not consume (log, merge-conflict-check, runway-merge). - `Controllers` struct + `NewControllers`: RPC-facing controllers (currently Ping), NOT bound to any wire contract. Pure addition alongside the existing main.go wiring — no behavioral changes. Ref: doc/rfc/submitqueue/modular-queue-wiring.md (Step 3) Co-Authored-By: Claude Opus 4.6 --- submitqueue/orchestrator/BUILD.bazel | 37 ++++ submitqueue/orchestrator/pipeline.go | 254 +++++++++++++++++++++++++++ 2 files changed, 291 insertions(+) create mode 100644 submitqueue/orchestrator/BUILD.bazel create mode 100644 submitqueue/orchestrator/pipeline.go diff --git a/submitqueue/orchestrator/BUILD.bazel b/submitqueue/orchestrator/BUILD.bazel new file mode 100644 index 00000000..37baf78a --- /dev/null +++ b/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", + ], +) diff --git a/submitqueue/orchestrator/pipeline.go b/submitqueue/orchestrator/pipeline.go new file mode 100644 index 00000000..1d649f8f --- /dev/null +++ b/submitqueue/orchestrator/pipeline.go @@ -0,0 +1,254 @@ +// 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 orchestrator declares the SubmitQueue orchestrator's pipeline +// topology, extension seams, and controller set. The host (main.go) fills +// Deps and passes Stages to pipeline.Construct; no assembly logic lives here. +package orchestrator + +import ( + "github.com/uber-go/tally" + runwaymq "github.com/uber/submitqueue/api/runway/messagequeue" + "github.com/uber/submitqueue/platform/consumer" + "github.com/uber/submitqueue/platform/extension/counter" + "github.com/uber/submitqueue/platform/pipeline" + "github.com/uber/submitqueue/submitqueue/core/topickey" + "github.com/uber/submitqueue/submitqueue/extension/buildrunner" + "github.com/uber/submitqueue/submitqueue/extension/changeprovider" + "github.com/uber/submitqueue/submitqueue/extension/conflict" + "github.com/uber/submitqueue/submitqueue/extension/scorer" + "github.com/uber/submitqueue/submitqueue/extension/storage" + "github.com/uber/submitqueue/submitqueue/extension/validator" + "github.com/uber/submitqueue/submitqueue/orchestrator/controller" + "github.com/uber/submitqueue/submitqueue/orchestrator/controller/batch" + "github.com/uber/submitqueue/submitqueue/orchestrator/controller/build" + "github.com/uber/submitqueue/submitqueue/orchestrator/controller/buildsignal" + "github.com/uber/submitqueue/submitqueue/orchestrator/controller/cancel" + "github.com/uber/submitqueue/submitqueue/orchestrator/controller/conclude" + "github.com/uber/submitqueue/submitqueue/orchestrator/controller/dlq" + "github.com/uber/submitqueue/submitqueue/orchestrator/controller/merge" + "github.com/uber/submitqueue/submitqueue/orchestrator/controller/mergeconflictsignal" + "github.com/uber/submitqueue/submitqueue/orchestrator/controller/mergesignal" + "github.com/uber/submitqueue/submitqueue/orchestrator/controller/score" + "github.com/uber/submitqueue/submitqueue/orchestrator/controller/speculate" + "github.com/uber/submitqueue/submitqueue/orchestrator/controller/start" + "github.com/uber/submitqueue/submitqueue/orchestrator/controller/validate" + "go.uber.org/zap" +) + +// Deps is the full set of dependencies the orchestrator pipeline needs. +// This struct IS the service's public API toward deployers: fill every +// field, pass it and Stages to pipeline.Construct, and you get a running +// orchestrator pipeline. +type Deps struct { + // Logger is the structured logger for all controllers. + Logger *zap.SugaredLogger + + // Scope is the metrics scope for all controllers. + Scope tally.Scope + + // Storage provides request, batch, and change stores. + Storage storage.Storage + + // Counter provides distributed batch counters. + Counter counter.Counter + + // BuildRunner resolves the build runner for each queue. + BuildRunner buildrunner.Factory + + // ChangeProvider resolves the change provider for each queue. + ChangeProvider changeprovider.Factory + + // Scorer resolves the scorer for each queue. + Scorer scorer.Factory + + // Analyzer resolves the conflict analyzer for each queue. + Analyzer conflict.Factory + + // Validator resolves the validator for each queue. + Validator validator.Factory +} + +// Stages is the orchestrator's pipeline topology as a typed table. +// Adding a stage = adding one row. Nothing else, anywhere. +// +// Pipeline: +// +// start → cancel → validate ⇢ (runway) ⇢ mergeconflictsignal → batch → score → speculate → build → buildsignal ─┐ +// ↑ ↘ ↻ poll │ +// │ merge → conclude │ +// │ │ │ +// └────────┴───────────────────────┘ +var Stages = []pipeline.Stage[Deps]{ + { + Key: topickey.TopicKeyStart, + Name: "start", + ConsumerGroup: "orchestrator-start", + New: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) { + return start.NewController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil + }, + DLQ: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) { + return dlq.NewDLQRequestController(d.Logger, d.Scope, d.Storage, sc.Registry, dlq.DecodeLandRequestID, sc.TopicKey, sc.ConsumerGroup), nil + }, + }, + { + Key: topickey.TopicKeyCancel, + Name: "cancel", + ConsumerGroup: "orchestrator-cancel", + New: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) { + return cancel.NewController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil + }, + DLQ: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) { + return dlq.NewDLQRequestController(d.Logger, d.Scope, d.Storage, sc.Registry, dlq.DecodeCancelRequestID, sc.TopicKey, sc.ConsumerGroup), nil + }, + }, + { + Key: topickey.TopicKeyValidate, + Name: "validate", + ConsumerGroup: "orchestrator-validate", + New: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) { + return validate.NewController(d.Logger, d.Scope, d.Storage, sc.Registry, d.ChangeProvider, d.Validator, runwaymq.TopicKeyMergeConflictCheck, sc.TopicKey, sc.ConsumerGroup), nil + }, + DLQ: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) { + return dlq.NewDLQRequestController(d.Logger, d.Scope, d.Storage, sc.Registry, dlq.DecodeRequestID, sc.TopicKey, sc.ConsumerGroup), nil + }, + }, + { + Key: runwaymq.TopicKeyMergeConflictCheckSignal, + Name: "merge-conflict-check-signal", + ConsumerGroup: "orchestrator-mergeconflictsignal", + New: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) { + return mergeconflictsignal.NewController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil + }, + DLQ: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) { + return dlq.NewDLQMergeConflictSignalController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil + }, + }, + { + Key: topickey.TopicKeyBatch, + Name: "batch", + ConsumerGroup: "orchestrator-batch", + New: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) { + return batch.NewController(d.Logger, d.Scope, sc.Registry, d.Counter, d.Storage, d.Analyzer, sc.TopicKey, sc.ConsumerGroup), nil + }, + DLQ: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) { + return dlq.NewDLQRequestController(d.Logger, d.Scope, d.Storage, sc.Registry, dlq.DecodeRequestID, sc.TopicKey, sc.ConsumerGroup), nil + }, + }, + { + Key: topickey.TopicKeyScore, + Name: "score", + ConsumerGroup: "orchestrator-score", + New: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) { + return score.NewController(d.Logger, d.Scope, d.Storage, d.Scorer, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil + }, + DLQ: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) { + return dlq.NewDLQBatchController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil + }, + }, + { + Key: topickey.TopicKeySpeculate, + Name: "speculate", + ConsumerGroup: "orchestrator-speculate", + New: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) { + return speculate.NewController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil + }, + DLQ: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) { + return dlq.NewDLQBatchController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil + }, + }, + { + Key: topickey.TopicKeyBuild, + Name: "build", + ConsumerGroup: "orchestrator-build", + New: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) { + return build.NewController(d.Logger, d.Scope, d.Storage, d.BuildRunner, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil + }, + DLQ: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) { + return dlq.NewDLQBatchController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil + }, + }, + { + Key: topickey.TopicKeyBuildSignal, + Name: "buildsignal", + ConsumerGroup: "orchestrator-buildsignal", + New: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) { + return buildsignal.NewController(d.Logger, d.Scope, d.Storage, d.BuildRunner, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil + }, + DLQ: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) { + return dlq.NewDLQBuildSignalController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil + }, + }, + { + Key: topickey.TopicKeyMerge, + Name: "submitqueue-merge", + ConsumerGroup: "orchestrator-merge", + New: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) { + return merge.NewController(d.Logger, d.Scope, d.Storage, sc.Registry, runwaymq.TopicKeyMerge, sc.TopicKey, sc.ConsumerGroup), nil + }, + DLQ: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) { + return dlq.NewDLQBatchController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil + }, + }, + { + Key: runwaymq.TopicKeyMergeSignal, + Name: "merge-signal", + ConsumerGroup: "orchestrator-mergesignal", + New: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) { + return mergesignal.NewController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil + }, + DLQ: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) { + return dlq.NewDLQMergeSignalController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil + }, + }, + { + Key: topickey.TopicKeyConclude, + Name: "conclude", + ConsumerGroup: "orchestrator-conclude", + New: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) { + return conclude.NewController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil + }, + DLQ: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) { + return dlq.NewDLQBatchController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil + }, + }, +} + +// PublishOnlyTopics declares topics the orchestrator publishes to but does +// not consume. These are registered in the TopicRegistry so controllers +// can look up topic names for publishing. +var PublishOnlyTopics = []pipeline.PublishOnlyTopic{ + // Log: the orchestrator emits request-log entries; the gateway consumes them. + {Key: topickey.TopicKeyLog, Name: "log"}, + // Merge-conflict check: the orchestrator publishes check requests to runway. + {Key: runwaymq.TopicKeyMergeConflictCheck, Name: "merge-conflict-check"}, + // Merge: the orchestrator publishes merge requests to runway. + {Key: runwaymq.TopicKeyMerge, Name: "runway-merge"}, +} + +// Controllers holds the orchestrator's RPC-facing controllers, constructed +// but NOT bound to any wire contract. Binding to a proto service + transport +// is host glue, because deployers may use different protos or transports. +type Controllers struct { + // Ping is the health-check controller. + Ping *controller.PingController +} + +// NewControllers creates the orchestrator's RPC controllers from the given Deps. +// The PingController takes a base *zap.Logger, so we desugar the SugaredLogger. +func NewControllers(d Deps) Controllers { + return Controllers{ + Ping: controller.NewPingController(d.Logger.Desugar(), d.Scope), + } +}