From 05d04ead1ecd151d726f1f04a2039dd60628750e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Daniel=20Gra=C3=B1a?= Date: Thu, 17 Sep 2026 09:27:48 -0300 Subject: [PATCH] Generated handlers: Reduce is ReduceOutput The output reducer on a pipeline's handler interface is ReduceOutput, pairing with ReduceFailure; the fold helper follows as ReduceXxxOutput. --- README.md | 4 ++-- cmd/protoc-gen-durable/internal/gen/gen.go | 12 ++++++------ docs/tour.md | 2 +- durabletest/doc.go | 2 +- examples/machines/handlers.go | 2 +- examples/machines/machinespb/machines_durable.pb.go | 12 ++++++------ examples/release-train/deploy.go | 2 +- examples/release-train/legacy.go | 2 +- examples/release-train/legacypb/legacy_durable.pb.go | 12 ++++++------ .../release-train/releasepb/release_durable.pb.go | 12 ++++++------ examples/snapshots/main.go | 2 +- examples/snapshots/main_test.go | 4 ++-- .../snapshots/snapshotspb/snapshots_durable.pb.go | 12 ++++++------ examples/tracing-otel/main.go | 2 +- examples/tracing-otel/orderspb/orders_durable.pb.go | 12 ++++++------ spec/02-authoring.md | 4 ++-- spec/05-codegen.md | 6 +++--- 17 files changed, 52 insertions(+), 52 deletions(-) diff --git a/README.md b/README.md index 276018d..d492433 100644 --- a/README.md +++ b/README.md @@ -64,7 +64,7 @@ message ProvisionMachine { `protoc-gen-durable` turns that into one interface, `ProvisionMachineHandlers`: a method per step, `Unwind` for each step that unwinds, and -`Reduce` for the output. One type implements the pipeline, so its +`ReduceOutput` for the output. One type implements the pipeline, so its dependencies are declared once, and a step it lacks — a step added to the proto included — is a compile error naming the method: @@ -106,7 +106,7 @@ func (h *handlers) CreateMachine(ctx context.Context, inv machinespb.ProvisionMa return &machinespb.ProvisionMachine_CreateMachine{MachineId: id}, nil } -func (h *handlers) Reduce(p *machinespb.ProvisionMachine) *machinespb.ProvisionMachineOutput { +func (h *handlers) ReduceOutput(p *machinespb.ProvisionMachine) *machinespb.ProvisionMachineOutput { m, _ := p.State(machinespb.ProvisionMachine_CreateMachineStep) return &machinespb.ProvisionMachineOutput{MachineId: m.GetMachineId()} } diff --git a/cmd/protoc-gen-durable/internal/gen/gen.go b/cmd/protoc-gen-durable/internal/gen/gen.go index f8c15ac..63311fd 100644 --- a/cmd/protoc-gen-durable/internal/gen/gen.go +++ b/cmd/protoc-gen-durable/internal/gen/gen.go @@ -162,7 +162,7 @@ func Generate(p *protogen.Plugin) error { methods[method] = by } if pl.output != nil { - claim("Reduce", "the output reducer") + claim("ReduceOutput", "the output reducer") } if pl.failureOutput != nil { claim("ReduceFailure", "the failure reducer") @@ -278,7 +278,7 @@ func emitPipeline(g *protogen.GeneratedFile, pl *pipelineDecl) { emitInvocationAlias(g, pl) emitHandlers(g, pl) if pl.output != nil { - emitReduceHelper(g, pl, "", pl.output) + emitReduceHelper(g, pl, "Output", pl.output) } if pl.failureOutput != nil { emitReduceHelper(g, pl, "Failure", pl.failureOutput) @@ -387,10 +387,10 @@ func emitHandlers(g *protogen.GeneratedFile, pl *pipelineDecl) { } } if pl.output != nil { - g.P("// Reduce produces the pipeline output from the immutable input and") + g.P("// ReduceOutput produces the pipeline output from the immutable input and") g.P("// committed step states on success. It must be pure: deterministic,") g.P("// side-effect free, synchronous, and non-failing.") - g.P("Reduce(*", g.QualifiedGoIdent(pl.msg.GoIdent), ") *", g.QualifiedGoIdent(pl.output.GoIdent)) + g.P("ReduceOutput(*", g.QualifiedGoIdent(pl.msg.GoIdent), ") *", g.QualifiedGoIdent(pl.output.GoIdent)) } if pl.failureOutput != nil { g.P("// ReduceFailure produces the pipeline failure output from the immutable") @@ -403,7 +403,7 @@ func emitHandlers(g *protogen.GeneratedFile, pl *pipelineDecl) { _ = name } -// emitReduceHelper emits Reduce[Failure](h, view): the fold the +// emitReduceHelper emits ReduceOutput/Failure(h, view): the fold the // engine reduces through and a reducer unit test calls with a // durabletest.NewInvocation (which is also a durable.ReduceView). func emitReduceHelper(g *protogen.GeneratedFile, pl *pipelineDecl, kind string, out *protogen.Message) { @@ -500,7 +500,7 @@ func emitDefinition(g *protogen.GeneratedFile, pl *pipelineDecl) { } if pl.output != nil { g.P("Reduce: func(view ", g.QualifiedGoIdent(durablePkg.Ident("ReduceView")), ") ", protoMsg, " {") - g.P("return Reduce", name, "(h, view)") + g.P("return Reduce", name, "Output(h, view)") g.P("},") } if pl.failureOutput != nil { diff --git a/docs/tour.md b/docs/tour.md index bcd60ee..3e16838 100644 --- a/docs/tour.md +++ b/docs/tour.md @@ -91,7 +91,7 @@ eng := engine.New(st) deploy, err := deploypb.NewDeployService(&deployer{db: db, lb: lb}).Bind(eng) // deployer has a method per step — ProvisionEnv, RunMigrations, // ShiftTraffic — plus UnwindProvisionEnv, UnwindRunMigrations, and -// Reduce; a step it lacks is a compile error. +// ReduceOutput; a step it lacks is a compile error. // bind every pipeline, then: eng.Start(ctx) diff --git a/durabletest/doc.go b/durabletest/doc.go index d1a17f4..7c3478c 100644 --- a/durabletest/doc.go +++ b/durabletest/doc.go @@ -2,6 +2,6 @@ // fake Clock for the engine's time, and a fake Invocation (NewInvocation) // for unit-testing handlers without an engine or a store. Generated code // accepts the fake through its NewXxxInvocation constructors and folds it -// as a reducer view through XxxReducer.Reduce. The in-memory store lives +// as a reducer view through the generated ReduceXxxOutput. The in-memory store lives // in store/mem: it is a real driver, not a test double. package durabletest diff --git a/examples/machines/handlers.go b/examples/machines/handlers.go index dfa4b06..7839ef3 100644 --- a/examples/machines/handlers.go +++ b/examples/machines/handlers.go @@ -110,7 +110,7 @@ func (h *handlers) CreateMachine(ctx context.Context, inv machinespb.ProvisionMa return &machinespb.ProvisionMachine_CreateMachine{MachineId: h.cloud.id("machine")}, nil } -func (h *handlers) Reduce(p *machinespb.ProvisionMachine) *machinespb.ProvisionMachineOutput { +func (h *handlers) ReduceOutput(p *machinespb.ProvisionMachine) *machinespb.ProvisionMachineOutput { machine, ok := p.State(machinespb.ProvisionMachine_CreateMachineStep) if !ok { panic("successful pipeline missing create-machine state") diff --git a/examples/machines/machinespb/machines_durable.pb.go b/examples/machines/machinespb/machines_durable.pb.go index 33e48e6..f76b476 100644 --- a/examples/machines/machinespb/machines_durable.pb.go +++ b/examples/machines/machinespb/machines_durable.pb.go @@ -56,20 +56,20 @@ type ProvisionMachineHandlers interface { UnwindReserveCapacity(ctx context.Context, inv ProvisionMachineInvocation) error // CreateMachine runs step "create-machine/v1". CreateMachine(ctx context.Context, inv ProvisionMachineInvocation) (*ProvisionMachine_CreateMachine, error) - // Reduce produces the pipeline output from the immutable input and + // ReduceOutput produces the pipeline output from the immutable input and // committed step states on success. It must be pure: deterministic, // side-effect free, synchronous, and non-failing. - Reduce(*ProvisionMachine) *ProvisionMachineOutput + ReduceOutput(*ProvisionMachine) *ProvisionMachineOutput } -// ReduceProvisionMachine folds view through h.Reduce: the marker the reducer +// ReduceProvisionMachineOutput folds view through h.ReduceOutput: the marker the reducer // receives reads its Input, States, and failures from view for the // duration of the call. -func ReduceProvisionMachine(h ProvisionMachineHandlers, view durable.ReduceView) *ProvisionMachineOutput { +func ReduceProvisionMachineOutput(h ProvisionMachineHandlers, view durable.ReduceView) *ProvisionMachineOutput { x := &ProvisionMachine{} provisionMachineViews.Store(x, view) defer provisionMachineViews.Delete(x) - return h.Reduce(x) + return h.ReduceOutput(x) } var provisionMachineViews sync.Map @@ -119,7 +119,7 @@ func NewProvisionMachine(h ProvisionMachineHandlers) *ProvisionMachineDefinition Mutexes: []string{"machine-lifecycle"}, NewInput: func() proto.Message { return &ProvisionMachineInput{} }, Reduce: func(view durable.ReduceView) proto.Message { - return ReduceProvisionMachine(h, view) + return ReduceProvisionMachineOutput(h, view) }, Steps: []pipelinedef.Step{ { diff --git a/examples/release-train/deploy.go b/examples/release-train/deploy.go index f86b277..fceaf86 100644 --- a/examples/release-train/deploy.go +++ b/examples/release-train/deploy.go @@ -57,6 +57,6 @@ func (h *deploy) ShiftTraffic(ctx context.Context, inv releasepb.DeployServiceIn return &releasepb.DeployService_ShiftTraffic{LbGeneration: inv.Input().GetImage()}, nil } -func (h *deploy) Reduce(d *releasepb.DeployService) *releasepb.DeployServiceOutput { +func (h *deploy) ReduceOutput(d *releasepb.DeployService) *releasepb.DeployServiceOutput { return &releasepb.DeployServiceOutput{Url: "https://" + d.Input().GetService() + ".example.com"} } diff --git a/examples/release-train/legacy.go b/examples/release-train/legacy.go index e99f668..13f5504 100644 --- a/examples/release-train/legacy.go +++ b/examples/release-train/legacy.go @@ -46,6 +46,6 @@ func (h *legacyDeploy) ShiftTraffic(ctx context.Context, inv legacypb.DeployServ return &legacypb.DeployService_ShiftTraffic{LbGeneration: inv.Input().GetImage()}, nil } -func (h *legacyDeploy) Reduce(d *legacypb.DeployService) *legacypb.DeployServiceOutput { +func (h *legacyDeploy) ReduceOutput(d *legacypb.DeployService) *legacypb.DeployServiceOutput { return &legacypb.DeployServiceOutput{Url: "https://" + d.Input().GetService() + ".example.com"} } diff --git a/examples/release-train/legacypb/legacy_durable.pb.go b/examples/release-train/legacypb/legacy_durable.pb.go index 9df1705..5cdacda 100644 --- a/examples/release-train/legacypb/legacy_durable.pb.go +++ b/examples/release-train/legacypb/legacy_durable.pb.go @@ -52,20 +52,20 @@ type DeployServiceHandlers interface { UnwindRunMigrations(ctx context.Context, inv DeployServiceInvocation) error // ShiftTraffic runs step "shift-traffic/v1". ShiftTraffic(ctx context.Context, inv DeployServiceInvocation) (*DeployService_ShiftTraffic, error) - // Reduce produces the pipeline output from the immutable input and + // ReduceOutput produces the pipeline output from the immutable input and // committed step states on success. It must be pure: deterministic, // side-effect free, synchronous, and non-failing. - Reduce(*DeployService) *DeployServiceOutput + ReduceOutput(*DeployService) *DeployServiceOutput } -// ReduceDeployService folds view through h.Reduce: the marker the reducer +// ReduceDeployServiceOutput folds view through h.ReduceOutput: the marker the reducer // receives reads its Input, States, and failures from view for the // duration of the call. -func ReduceDeployService(h DeployServiceHandlers, view durable.ReduceView) *DeployServiceOutput { +func ReduceDeployServiceOutput(h DeployServiceHandlers, view durable.ReduceView) *DeployServiceOutput { x := &DeployService{} deployServiceViews.Store(x, view) defer deployServiceViews.Delete(x) - return h.Reduce(x) + return h.ReduceOutput(x) } var deployServiceViews sync.Map @@ -114,7 +114,7 @@ func NewDeployService(h DeployServiceHandlers) *DeployServiceDefinition { ID: "deploy-service", NewInput: func() proto.Message { return &DeployServiceInput{} }, Reduce: func(view durable.ReduceView) proto.Message { - return ReduceDeployService(h, view) + return ReduceDeployServiceOutput(h, view) }, Steps: []pipelinedef.Step{ { diff --git a/examples/release-train/releasepb/release_durable.pb.go b/examples/release-train/releasepb/release_durable.pb.go index f4b252e..2e2e2e3 100644 --- a/examples/release-train/releasepb/release_durable.pb.go +++ b/examples/release-train/releasepb/release_durable.pb.go @@ -58,20 +58,20 @@ type DeployServiceHandlers interface { CanaryAnalysis(ctx context.Context, inv DeployServiceInvocation) (*DeployService_CanaryAnalysis, error) // ShiftTraffic runs step "shift-traffic/v1". ShiftTraffic(ctx context.Context, inv DeployServiceInvocation) (*DeployService_ShiftTraffic, error) - // Reduce produces the pipeline output from the immutable input and + // ReduceOutput produces the pipeline output from the immutable input and // committed step states on success. It must be pure: deterministic, // side-effect free, synchronous, and non-failing. - Reduce(*DeployService) *DeployServiceOutput + ReduceOutput(*DeployService) *DeployServiceOutput } -// ReduceDeployService folds view through h.Reduce: the marker the reducer +// ReduceDeployServiceOutput folds view through h.ReduceOutput: the marker the reducer // receives reads its Input, States, and failures from view for the // duration of the call. -func ReduceDeployService(h DeployServiceHandlers, view durable.ReduceView) *DeployServiceOutput { +func ReduceDeployServiceOutput(h DeployServiceHandlers, view durable.ReduceView) *DeployServiceOutput { x := &DeployService{} deployServiceViews.Store(x, view) defer deployServiceViews.Delete(x) - return h.Reduce(x) + return h.ReduceOutput(x) } var deployServiceViews sync.Map @@ -120,7 +120,7 @@ func NewDeployService(h DeployServiceHandlers) *DeployServiceDefinition { ID: "deploy-service", NewInput: func() proto.Message { return &DeployServiceInput{} }, Reduce: func(view durable.ReduceView) proto.Message { - return ReduceDeployService(h, view) + return ReduceDeployServiceOutput(h, view) }, Steps: []pipelinedef.Step{ { diff --git a/examples/snapshots/main.go b/examples/snapshots/main.go index 3422d0b..8cd6215 100644 --- a/examples/snapshots/main.go +++ b/examples/snapshots/main.go @@ -167,7 +167,7 @@ func (h *snapshotter) RegisterSnapshot(ctx context.Context, inv snapshotspb.Crea return &snapshotspb.CreateSnapshot_RegisterSnapshot{SnapshotId: id}, nil } -func (h *snapshotter) Reduce(p *snapshotspb.CreateSnapshot) *snapshotspb.CreateSnapshotOutput { +func (h *snapshotter) ReduceOutput(p *snapshotspb.CreateSnapshot) *snapshotspb.CreateSnapshotOutput { reg, _ := p.State(snapshotspb.CreateSnapshot_RegisterSnapshotStep) up, _ := p.State(snapshotspb.CreateSnapshot_UploadSnapshotStep) return &snapshotspb.CreateSnapshotOutput{SnapshotId: reg.GetSnapshotId(), ObjectKey: up.GetObjectKey()} diff --git a/examples/snapshots/main_test.go b/examples/snapshots/main_test.go index 6acdc26..74ef9d4 100644 --- a/examples/snapshots/main_test.go +++ b/examples/snapshots/main_test.go @@ -96,7 +96,7 @@ func TestSnapshotUnwindsOnCatalogFull(t *testing.T) { } // The closures are unit-testable without an engine: the generated -// NewXxxInvocation wraps the durabletest fake, and XxxReducer.Reduce +// NewXxxInvocation wraps the durabletest fake, and ReduceXxxOutput // folds it as a reducer view. func TestUploadUnwindDeletesObject(t *testing.T) { @@ -153,7 +153,7 @@ func TestReduceCreateSnapshot(t *testing.T) { snapshotspb.CreateSnapshot_RegisterSnapshotStep.ID(): &snapshotspb.CreateSnapshot_RegisterSnapshot{SnapshotId: "snap-1"}, }, }) - out := snapshotspb.ReduceCreateSnapshot(&snapshotter{}, view) + out := snapshotspb.ReduceCreateSnapshotOutput(&snapshotter{}, view) if out.GetSnapshotId() != "snap-1" || out.GetObjectKey() != "k" { t.Fatalf("Reduce = %+v", out) } diff --git a/examples/snapshots/snapshotspb/snapshots_durable.pb.go b/examples/snapshots/snapshotspb/snapshots_durable.pb.go index 8a522c0..8893f81 100644 --- a/examples/snapshots/snapshotspb/snapshots_durable.pb.go +++ b/examples/snapshots/snapshotspb/snapshots_durable.pb.go @@ -58,24 +58,24 @@ type CreateSnapshotHandlers interface { ThawVolume(ctx context.Context, inv CreateSnapshotInvocation) error // RegisterSnapshot runs step "register-snapshot/v1". RegisterSnapshot(ctx context.Context, inv CreateSnapshotInvocation) (*CreateSnapshot_RegisterSnapshot, error) - // Reduce produces the pipeline output from the immutable input and + // ReduceOutput produces the pipeline output from the immutable input and // committed step states on success. It must be pure: deterministic, // side-effect free, synchronous, and non-failing. - Reduce(*CreateSnapshot) *CreateSnapshotOutput + ReduceOutput(*CreateSnapshot) *CreateSnapshotOutput // ReduceFailure produces the pipeline failure output from the immutable // input, the committed step states, the run's Failure, and the permanent // unwind failures once the unwind completes. It must be pure. ReduceFailure(*CreateSnapshot) *CreateSnapshotFailure } -// ReduceCreateSnapshot folds view through h.Reduce: the marker the reducer +// ReduceCreateSnapshotOutput folds view through h.ReduceOutput: the marker the reducer // receives reads its Input, States, and failures from view for the // duration of the call. -func ReduceCreateSnapshot(h CreateSnapshotHandlers, view durable.ReduceView) *CreateSnapshotOutput { +func ReduceCreateSnapshotOutput(h CreateSnapshotHandlers, view durable.ReduceView) *CreateSnapshotOutput { x := &CreateSnapshot{} createSnapshotViews.Store(x, view) defer createSnapshotViews.Delete(x) - return h.Reduce(x) + return h.ReduceOutput(x) } // ReduceCreateSnapshotFailure folds view through h.ReduceFailure: the marker the reducer @@ -134,7 +134,7 @@ func NewCreateSnapshot(h CreateSnapshotHandlers) *CreateSnapshotDefinition { ID: "create-snapshot", NewInput: func() proto.Message { return &CreateSnapshotInput{} }, Reduce: func(view durable.ReduceView) proto.Message { - return ReduceCreateSnapshot(h, view) + return ReduceCreateSnapshotOutput(h, view) }, ReduceFailure: func(view durable.ReduceView) proto.Message { return ReduceCreateSnapshotFailure(h, view) diff --git a/examples/tracing-otel/main.go b/examples/tracing-otel/main.go index 03c7b70..441fdb9 100644 --- a/examples/tracing-otel/main.go +++ b/examples/tracing-otel/main.go @@ -105,7 +105,7 @@ func (h *fulfillment) Ship(ctx context.Context, inv orderspb.FulfillOrderInvocat durable.WithUserKind(), durable.WithReason("invalid-address")) } -func (h *fulfillment) Reduce(o *orderspb.FulfillOrder) *orderspb.FulfillOrderOutput { +func (h *fulfillment) ReduceOutput(o *orderspb.FulfillOrder) *orderspb.FulfillOrderOutput { s, _ := o.State(orderspb.FulfillOrder_ShipStep) return &orderspb.FulfillOrderOutput{ShipmentId: s.GetShipmentId()} } diff --git a/examples/tracing-otel/orderspb/orders_durable.pb.go b/examples/tracing-otel/orderspb/orders_durable.pb.go index e4f1e98..9cec240 100644 --- a/examples/tracing-otel/orderspb/orders_durable.pb.go +++ b/examples/tracing-otel/orderspb/orders_durable.pb.go @@ -52,20 +52,20 @@ type FulfillOrderHandlers interface { UnwindChargePayment(ctx context.Context, inv FulfillOrderInvocation) error // Ship runs step "ship/v1". Ship(ctx context.Context, inv FulfillOrderInvocation) (*FulfillOrder_Ship, error) - // Reduce produces the pipeline output from the immutable input and + // ReduceOutput produces the pipeline output from the immutable input and // committed step states on success. It must be pure: deterministic, // side-effect free, synchronous, and non-failing. - Reduce(*FulfillOrder) *FulfillOrderOutput + ReduceOutput(*FulfillOrder) *FulfillOrderOutput } -// ReduceFulfillOrder folds view through h.Reduce: the marker the reducer +// ReduceFulfillOrderOutput folds view through h.ReduceOutput: the marker the reducer // receives reads its Input, States, and failures from view for the // duration of the call. -func ReduceFulfillOrder(h FulfillOrderHandlers, view durable.ReduceView) *FulfillOrderOutput { +func ReduceFulfillOrderOutput(h FulfillOrderHandlers, view durable.ReduceView) *FulfillOrderOutput { x := &FulfillOrder{} fulfillOrderViews.Store(x, view) defer fulfillOrderViews.Delete(x) - return h.Reduce(x) + return h.ReduceOutput(x) } var fulfillOrderViews sync.Map @@ -114,7 +114,7 @@ func NewFulfillOrder(h FulfillOrderHandlers) *FulfillOrderDefinition { ID: "fulfill-order", NewInput: func() proto.Message { return &FulfillOrderInput{} }, Reduce: func(view durable.ReduceView) proto.Message { - return ReduceFulfillOrder(h, view) + return ReduceFulfillOrderOutput(h, view) }, Steps: []pipelinedef.Step{ { diff --git a/spec/02-authoring.md b/spec/02-authoring.md index b30d634..9b8a1ac 100644 --- a/spec/02-authoring.md +++ b/spec/02-authoring.md @@ -259,7 +259,7 @@ type ProvisionMachineHandlers interface { ReserveCapacity(context.Context, ProvisionMachineInvocation) (*ReserveCapacity, error) UnwindReserveCapacity(context.Context, ProvisionMachineInvocation) error - Reduce(*ProvisionMachine) *ProvisionMachineOutput + ReduceOutput(*ProvisionMachine) *ProvisionMachineOutput } func NewProvisionMachine(h ProvisionMachineHandlers) *ProvisionMachineDefinition @@ -317,7 +317,7 @@ annotations, park memory) that satisfies both `durable.Invocation` and would invalidate the Run for. Generated code wraps it the same way it wraps the Engine's: `NewProvisionMachineInvocation(fake)` is what a `ProvisionMachineHandlers` method takes, and -`ReduceProvisionMachine(h, fake)` folds the reducer over it. +`ReduceProvisionMachineOutput(h, fake)` folds the reducer over it. ```go inv := durabletest.NewInvocation(durabletest.InvocationConfig{ diff --git a/spec/05-codegen.md b/spec/05-codegen.md index 9188995..f02c17e 100644 --- a/spec/05-codegen.md +++ b/spec/05-codegen.md @@ -51,10 +51,10 @@ Published protobuf extensions MUST use globally allocated extension numbers. (`durable.NoInput` for an Input-less pipeline), with a `NewXxxInvocation(core)` constructor for engine-free handler tests, - one handler interface, `XxxHandlers`: a method per Step named after - the Step, `Unwind` for each Step that unwinds, and `Reduce` / + the Step, `Unwind` for each Step that unwinds, and `ReduceOutput` / `ReduceFailure` when the pipeline declares outputs, - the Pipeline constructor, `NewXxx(h XxxHandlers)`, -- `ReduceXxx(h, view)` and `ReduceXxxFailure(h, view)`, the folds the +- `ReduceXxxOutput(h, view)` and `ReduceXxxFailure(h, view)`, the folds the engine reduces through and reducer tests call, - runtime methods on the Pipeline marker type, - the bound Pipeline handle, @@ -97,7 +97,7 @@ Generated APIs MUST make these compile-time errors where possible: - a missing Step method, a Step added to the pipeline included, - a wrong method signature, - a missing `Unwind`, -- an invalid `Reduce` signature, +- an invalid `ReduceOutput` or `ReduceFailure` signature, - passing a stateless `StepRef` to `State`. Generation MUST also reject a pipeline whose Step and reducer method