Skip to content

Commit 6159f66

Browse files
JamyDevclaude
JamyDev
and
claude
committed
feat(platform): add StageContext to pipeline.Stage constructors
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 <noreply@anthropic.com>
1 parent b49618c commit 6159f66

2 files changed

Lines changed: 98 additions & 30 deletions

File tree

‎platform/pipeline/pipeline.go‎

Lines changed: 35 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -49,16 +49,33 @@ type Stage[D any] struct {
4949
// (e.g. "orchestrator-start").
5050
ConsumerGroup string
5151

52-
// New builds the stage's controller from the service's Deps. The engine
53-
// calls it once, eagerly, inside Construct — so a nil/missing dependency
54-
// fails at boot with the stage's name on it, never mid-delivery.
55-
New func(D) (consumer.Controller, error)
52+
// New builds the stage's controller from the service's Deps and engine-
53+
// provided StageContext. The engine calls it once, eagerly, inside
54+
// Construct — so a nil/missing dependency fails at boot with the stage's
55+
// name on it, never mid-delivery.
56+
New func(D, StageContext) (consumer.Controller, error)
5657

5758
// DLQ, when non-nil, declares "this stage dead-letters". The engine then
5859
// derives the paired DLQ topic (<topic>_dlq, retry budget, DLQ-of-DLQ
5960
// disabled) AND registers this reconciler on the DLQ consumer. Declaring
6061
// one without getting the other is impossible — that's the invariant.
61-
DLQ func(D) (consumer.Controller, error)
62+
DLQ func(D, StageContext) (consumer.Controller, error)
63+
}
64+
65+
// StageContext carries engine-produced values that controllers need at
66+
// construction time but the host does not own: the assembled topic
67+
// registry (for publishing to downstream stages), the stage's own topic
68+
// key, and its consumer group.
69+
type StageContext struct {
70+
// Registry is the fully assembled TopicRegistry. Controllers use it to
71+
// look up topic names and queue backends for publishing downstream.
72+
Registry consumer.TopicRegistry
73+
74+
// TopicKey is this stage's logical topic key.
75+
TopicKey consumer.TopicKey
76+
77+
// ConsumerGroup is this stage's consumer group name.
78+
ConsumerGroup string
6279
}
6380

6481
// PublishOnlyTopic declares a topic the service publishes to but does not
@@ -166,7 +183,13 @@ func Construct[D any](
166183

167184
// Eagerly construct and register all controllers.
168185
for _, s := range stages {
169-
ctl, err := s.New(deps)
186+
sc := StageContext{
187+
Registry: registry,
188+
TopicKey: s.Key,
189+
ConsumerGroup: s.ConsumerGroup,
190+
}
191+
192+
ctl, err := s.New(deps, sc)
170193
if err != nil {
171194
return nil, fmt.Errorf("pipeline: stage %s: failed to create controller: %w", s.Key, err)
172195
}
@@ -175,7 +198,12 @@ func Construct[D any](
175198
}
176199

177200
if s.DLQ != nil {
178-
rec, err := s.DLQ(deps)
201+
dlqSC := StageContext{
202+
Registry: registry,
203+
TopicKey: dlqTopicKey(s.Key),
204+
ConsumerGroup: s.ConsumerGroup + "-dlq",
205+
}
206+
rec, err := s.DLQ(deps, dlqSC)
179207
if err != nil {
180208
return nil, fmt.Errorf("pipeline: stage %s dlq: failed to create controller: %w", s.Key, err)
181209
}

‎platform/pipeline/pipeline_test.go‎

Lines changed: 63 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -62,8 +62,8 @@ func TestConstruct_SingleStage_NoDLQ(t *testing.T) {
6262
Key: "start",
6363
Name: "start",
6464
ConsumerGroup: "orchestrator-start",
65-
New: func(d testDeps) (consumer.Controller, error) {
66-
return &fakeController{key: "start", group: "orchestrator-start"}, nil
65+
New: func(d testDeps, sc StageContext) (consumer.Controller, error) {
66+
return &fakeController{key: sc.TopicKey, group: sc.ConsumerGroup}, nil
6767
},
6868
},
6969
}
@@ -85,11 +85,11 @@ func TestConstruct_WithDLQ(t *testing.T) {
8585
Key: "start",
8686
Name: "start",
8787
ConsumerGroup: "orchestrator-start",
88-
New: func(d testDeps) (consumer.Controller, error) {
89-
return &fakeController{key: "start", group: "orchestrator-start"}, nil
88+
New: func(d testDeps, sc StageContext) (consumer.Controller, error) {
89+
return &fakeController{key: sc.TopicKey, group: sc.ConsumerGroup}, nil
9090
},
91-
DLQ: func(d testDeps) (consumer.Controller, error) {
92-
return &fakeController{key: "start_dlq", group: "orchestrator-start-dlq"}, nil
91+
DLQ: func(d testDeps, sc StageContext) (consumer.Controller, error) {
92+
return &fakeController{key: sc.TopicKey, group: sc.ConsumerGroup}, nil
9393
},
9494
},
9595
}
@@ -111,19 +111,19 @@ func TestConstruct_MultipleStages(t *testing.T) {
111111
Key: "start",
112112
Name: "start",
113113
ConsumerGroup: "orchestrator-start",
114-
New: func(d testDeps) (consumer.Controller, error) {
115-
return &fakeController{key: "start", group: "orchestrator-start"}, nil
114+
New: func(d testDeps, sc StageContext) (consumer.Controller, error) {
115+
return &fakeController{key: sc.TopicKey, group: sc.ConsumerGroup}, nil
116116
},
117117
},
118118
{
119119
Key: "validate",
120120
Name: "validate",
121121
ConsumerGroup: "orchestrator-validate",
122-
New: func(d testDeps) (consumer.Controller, error) {
123-
return &fakeController{key: "validate", group: "orchestrator-validate"}, nil
122+
New: func(d testDeps, sc StageContext) (consumer.Controller, error) {
123+
return &fakeController{key: sc.TopicKey, group: sc.ConsumerGroup}, nil
124124
},
125-
DLQ: func(d testDeps) (consumer.Controller, error) {
126-
return &fakeController{key: "validate_dlq", group: "orchestrator-validate-dlq"}, nil
125+
DLQ: func(d testDeps, sc StageContext) (consumer.Controller, error) {
126+
return &fakeController{key: sc.TopicKey, group: sc.ConsumerGroup}, nil
127127
},
128128
},
129129
}
@@ -155,7 +155,7 @@ func TestConstruct_ControllerCreationFailure(t *testing.T) {
155155
Key: "start",
156156
Name: "start",
157157
ConsumerGroup: "orchestrator-start",
158-
New: func(d testDeps) (consumer.Controller, error) {
158+
New: func(d testDeps, sc StageContext) (consumer.Controller, error) {
159159
return nil, fmt.Errorf("missing dependency")
160160
},
161161
},
@@ -179,10 +179,10 @@ func TestConstruct_DLQControllerCreationFailure(t *testing.T) {
179179
Key: "start",
180180
Name: "start",
181181
ConsumerGroup: "orchestrator-start",
182-
New: func(d testDeps) (consumer.Controller, error) {
183-
return &fakeController{key: "start", group: "orchestrator-start"}, nil
182+
New: func(d testDeps, sc StageContext) (consumer.Controller, error) {
183+
return &fakeController{key: sc.TopicKey, group: sc.ConsumerGroup}, nil
184184
},
185-
DLQ: func(d testDeps) (consumer.Controller, error) {
185+
DLQ: func(d testDeps, sc StageContext) (consumer.Controller, error) {
186186
return nil, fmt.Errorf("dlq dependency missing")
187187
},
188188
},
@@ -206,8 +206,8 @@ func TestConstruct_WithPublishOnly(t *testing.T) {
206206
Key: "start",
207207
Name: "start",
208208
ConsumerGroup: "orchestrator-start",
209-
New: func(d testDeps) (consumer.Controller, error) {
210-
return &fakeController{key: "start", group: "orchestrator-start"}, nil
209+
New: func(d testDeps, sc StageContext) (consumer.Controller, error) {
210+
return &fakeController{key: sc.TopicKey, group: sc.ConsumerGroup}, nil
211211
},
212212
},
213213
}
@@ -234,8 +234,8 @@ func TestConstruct_WithTopicNameOverrides(t *testing.T) {
234234
Key: "start",
235235
Name: "start",
236236
ConsumerGroup: "orchestrator-start",
237-
New: func(d testDeps) (consumer.Controller, error) {
238-
return &fakeController{key: "start", group: "orchestrator-start"}, nil
237+
New: func(d testDeps, sc StageContext) (consumer.Controller, error) {
238+
return &fakeController{key: sc.TopicKey, group: sc.ConsumerGroup}, nil
239239
},
240240
},
241241
}
@@ -249,6 +249,46 @@ func TestConstruct_WithTopicNameOverrides(t *testing.T) {
249249
assert.NotNil(t, comp)
250250
}
251251

252+
func TestConstruct_StageContext_Populated(t *testing.T) {
253+
ctrl := gomock.NewController(t)
254+
q := mqmock.NewMockQueue(ctrl)
255+
q.EXPECT().Subscriber().Return(mqmock.NewMockSubscriber(ctrl)).AnyTimes()
256+
q.EXPECT().Publisher().Return(mqmock.NewMockPublisher(ctrl)).AnyTimes()
257+
258+
deps := testDeps{logger: newTestLogger()}
259+
260+
var primarySC, dlqSC StageContext
261+
stages := []Stage[testDeps]{
262+
{
263+
Key: "start",
264+
Name: "start",
265+
ConsumerGroup: "orchestrator-start",
266+
New: func(d testDeps, sc StageContext) (consumer.Controller, error) {
267+
primarySC = sc
268+
return &fakeController{key: sc.TopicKey, group: sc.ConsumerGroup}, nil
269+
},
270+
DLQ: func(d testDeps, sc StageContext) (consumer.Controller, error) {
271+
dlqSC = sc
272+
return &fakeController{key: sc.TopicKey, group: sc.ConsumerGroup}, nil
273+
},
274+
},
275+
}
276+
277+
_, err := Construct(deps.logger, tally.NoopScope, q, "test-sub", deps, stages)
278+
require.NoError(t, err)
279+
280+
// Primary StageContext should have the stage's own key and group.
281+
assert.Equal(t, consumer.TopicKey("start"), primarySC.TopicKey)
282+
assert.Equal(t, "orchestrator-start", primarySC.ConsumerGroup)
283+
284+
// DLQ StageContext should have the derived DLQ key and group.
285+
assert.Equal(t, consumer.TopicKey("start_dlq"), dlqSC.TopicKey)
286+
assert.Equal(t, "orchestrator-start-dlq", dlqSC.ConsumerGroup)
287+
288+
// Both should share the same registry.
289+
assert.Equal(t, primarySC.Registry, dlqSC.Registry)
290+
}
291+
252292
func TestResolveTopicName(t *testing.T) {
253293
tests := []struct {
254294
name string
@@ -308,14 +348,14 @@ func TestBuildTopicConfigs(t *testing.T) {
308348
Key: "start",
309349
Name: "start",
310350
ConsumerGroup: "orchestrator-start",
311-
New: func(d testDeps) (consumer.Controller, error) { return nil, nil },
312-
DLQ: func(d testDeps) (consumer.Controller, error) { return nil, nil },
351+
New: func(d testDeps, sc StageContext) (consumer.Controller, error) { return nil, nil },
352+
DLQ: func(d testDeps, sc StageContext) (consumer.Controller, error) { return nil, nil },
313353
},
314354
{
315355
Key: "validate",
316356
Name: "validate",
317357
ConsumerGroup: "orchestrator-validate",
318-
New: func(d testDeps) (consumer.Controller, error) { return nil, nil },
358+
New: func(d testDeps, sc StageContext) (consumer.Controller, error) { return nil, nil },
319359
// No DLQ for this stage.
320360
},
321361
}

0 commit comments

Comments
 (0)