`pipeline` keyword — supervised multi-stage data flows

[gregwinn]

Summary

Declarative multi-stage data pipelines as a language construct. Each stage runs as a supervised BEAM process with configurable concurrency, retry, timeout, and backpressure. Like Elixir's GenStage/Flow but zero boilerplate.

Syntax

pipeline OrderProcessor
  stage :validate do |order|
    Changeset.validate(order)
  end

  stage :charge, retry: 3, timeout: 5000 do |order|
    PaymentService.charge(order)
  end

  stage :fulfill, concurrency: 5 do |order|
    WarehouseService.ship(order)
  end

  stage :notify, async: true do |result|
    Email.send(result.user, result)
  end
end

# Start supervised pipeline
OrderProcessor.start()
OrderProcessor.push(order)

Design

• Each stage is an agent (agent keyword — stateful actors as a language primitive #102) — stateful, supervised, hot-reloadable
• Backpressure via demand-driven message passing between stages
• Options: retry, timeout, concurrency (parallel workers per stage), async
• Supervision tree auto-generated: one supervisor per pipeline, one child per stage

Dependencies

• Requires agent keyword — stateful actors as a language primitive #102 (agent keyword) as foundation
• Requires Reactive on/emit event system — language-level pub/sub #103 (on/emit events) for inter-stage communication patterns
• Phase 3 of 3 for first-class concurrency primitives