[Feature] Akka-Streams DSL subset (SourceQueue, MergeHub, BroadcastHub)

[pathosDev]

Size / Priority

• Size: XL — needs its own design phase before implementation.
• Reference: Akka Streams (full library is XL; this issue scopes a 5% subset).
• Legacy tracking: roadmap Akka-Streams-style reactive streaming DSL (Source/Flow/Sink) #54 (now superseded by this issue).

Rationale

Akka apps frequently reach for Streams for three specific patterns:

1. Backpressure-aware queue from non-actor code (Source.queue) — HTTP request handlers push events into a bounded queue; downstream actor consumes at its own pace.
2. Fan-in from many producers (MergeHub) — every WebSocket connection produces events; a single downstream pipeline consumes them merged.
3. Fan-out to many consumers (BroadcastHub) — single upstream event stream; N late-attached consumers each get the live tail.

actor-ts has actor-mailbox-based backpressure but no front-door API for non-actor callers to participate in it. The Streams subset closes that gap without shipping the full Streams library (which would be a 6-month project alone — sources, sinks, flows, async-boundaries, materializers, RestartSource, Throttle, Conflate, etc.).

The non-goal is "Akka Streams API compatibility". The goal is "the 5% of the API surface that 80% of apps actually use", expressed naturally in TypeScript.

Reference: what Akka does

// Source.queue + backpressure
val (queue, completion) = Source
  .queue[Event](bufferSize = 100, overflowStrategy = OverflowStrategy.backpressure)
  .toMat(Sink.foreach(processEvent))(Keep.both)
  .run()

queue.offer(event).onComplete {
  case Success(QueueOfferResult.Enqueued) => ...
  case Success(QueueOfferResult.Dropped) => ...
}

// MergeHub: many producers → one sink
val sink: Sink[Event, NotUsed] = MergeHub.source[Event](perProducerBufferSize = 16)
  .to(Sink.foreach(processEvent))
  .run()
// Each producer:
Source.single(event).runWith(sink)

// BroadcastHub: one source → many consumers
val source: Source[Event, NotUsed] = Source
  .actorRef[Event](bufferSize = 256, overflowStrategy = OverflowStrategy.dropHead)
  .toMat(BroadcastHub.sink[Event](bufferSize = 256))(Keep.right)
  .run()
// Each late consumer:
source.runForeach(processForOneConsumer)

Design sketch — actor-ts equivalents

SourceQueue<T> — push from non-actor code with backpressure semantics:

export interface SourceQueueOptions {
  /** Bounded buffer; 0 = unbuffered (offer blocks until consumer ready). Default: 64. */
  readonly bufferSize?: number;
  /** What happens when bufferSize is reached. Default: 'backpressure'. */
  readonly overflowStrategy?: 'backpressure' | 'drop-head' | 'drop-tail' | 'drop-new' | 'fail';
}

export interface SourceQueue<T> {
  /**
   * Push one element.  Resolves with:
   *   - 'enqueued' if the buffer accepted (or if backpressure waited then accepted).
   *   - 'dropped'  for drop-* strategies when buffer was full.
   * Rejects for 'fail' strategy + full buffer.
   */
  offer(value: T): Promise<'enqueued' | 'dropped'>;

  /** Signal completion to the consumer.  Subsequent offer() rejects. */
  complete(): Promise<void>;

  /** Signal failure to the consumer.  Subsequent offer() rejects. */
  fail(error: Error): Promise<void>;

  /** Future that resolves when the consumer side terminates. */
  readonly watchCompletion: Promise<void>;
}

export function sourceQueue<T>(
  consumer: (value: T) => Promise<void> | void,
  options?: SourceQueueOptions,
): SourceQueue<T>;

Implementation: under the hood, a small wrapper actor with a bounded mailbox. offer is actor.ask({ kind: 'offer', value }); the actor's handler awaits consumer(value) and replies. Backpressure is exactly the mailbox-backpressure we already have.

MergeHub<T> — fan-in from many producers:

export interface MergeHubOptions {
  /** Per-producer buffer; protects against one slow producer blocking others. Default: 16. */
  readonly perProducerBufferSize?: number;
}

export interface MergeHubSink<T> {
  /** Attach a producer.  Returns its push-handle. */
  attach(): SourceQueue<T>;
  /** Stop accepting new producers; in-flight ones drain. */
  complete(): Promise<void>;
}

export function mergeHub<T>(
  consumer: (value: T) => Promise<void> | void,
  options?: MergeHubOptions,
): MergeHubSink<T>;

Implementation: a coordinator actor that holds the consumer; each attach() spawns a child actor with its own bounded queue that drains into the coordinator. Per-producer backpressure isolated.

BroadcastHub<T> — fan-out to many late consumers:

export interface BroadcastHubOptions {
  /** Per-consumer buffer; slow consumers don't block fast ones. Default: 256. */
  readonly perConsumerBufferSize?: number;
  /** What happens for a consumer that's too slow. Default: 'drop-head'. */
  readonly slowConsumerStrategy?: 'drop-head' | 'detach';
}

export interface BroadcastHubSource<T> {
  /** Attach a consumer.  Returns an unsubscribe handle. */
  attach(consumer: (value: T) => Promise<void> | void): () => void;
  /** Push from upstream. */
  push(value: T): Promise<void>;
  /** Signal completion. */
  complete(): Promise<void>;
}

export function broadcastHub<T>(options?: BroadcastHubOptions): BroadcastHubSource<T>;

Implementation: coordinator actor; each attach() adds a child with its own per-consumer buffer. Upstream push fans out to every child's mailbox concurrently.

Integration with existing actor-ts subsystems

• Mailbox: piggy-backs on the existing bounded-mailbox + backpressure semantics. No new transport, no new scheduler.
• Supervision: hubs are normal actors; user can system.actorOf(...) and apply standard supervision strategies.
• Cluster: a SourceQueue whose consumer is a RemoteActorRef works automatically — the backpressure is whatever the remote mailbox provides (today: unbounded; with [Feature] Dispatcher-saturation + mailbox-depth histograms #196 dispatcher-saturation tracking, becomes meaningful).
• Metrics: streams_queue_depth{name} Gauge, streams_offer_dropped_total{name,reason} Counter, streams_consumer_latency_seconds Histogram. Reuse MetricsRegistry.

Out of scope / non-goals

Explicitly NOT in this issue (track separately if demand emerges):

• Full Streams DSL: Source.map, Source.filter, Source.via(Flow), Source.runWith, materialization. We're not building Akka Streams — we're shipping the 3 hub patterns.
• Async boundaries / materializer: actor-ts already has async via actors. No need for a separate execution model.
• RestartSource / RestartFlow: use BackoffSupervisor on the hub actor instead.
• Reactive Streams interop (Publisher / Subscriber / Subscription): track as separate item [Feature] Reactive-Streams interop (Akka-Streams subset interop) #189 (B6.6) — wraps the hubs in the reactive-streams-spec API.
• GraphDSL: the visual-graph composition syntax. Not needed for the 3 hubs.

Migration story (from Akka)

// Akka Scala:
//   Source.queue[Event](100, OverflowStrategy.backpressure).runForeach(processEvent)
// actor-ts:
const queue = sourceQueue<Event>(processEvent, { bufferSize: 100 });
await queue.offer(event);  // returns 'enqueued' | 'dropped'

Cheat-sheet doc page mapping Akka idioms → actor-ts equivalents. ~3 pages.

Open design questions

1. offer return type: Akka returns QueueOfferResult (a sealed trait). TypeScript discriminated union vs. throwing the failure? Current sketch uses a string literal; could promote to { status: 'enqueued' } | { status: 'dropped'; reason: 'buffer-full' } if richer info needed.
2. Materialization: Akka separates "graph definition" from "running graph" via .run(). We start running on construction — simpler, matches actor lifecycle. Is anyone relying on graph-without-running for re-use?
3. MergeHub.attach() lifecycle: how does a producer signal "I'm done"? Probably queue.complete() on the returned SourceQueue. The coordinator stops counting that producer for completion-tracking but other producers continue.
4. BroadcastHub slow-consumer: 'detach' strategy needs a callback to notify the consumer they were detached. New onDetach?: (reason) => void option on attach()?

These are part of the design phase, not pre-decided.

Test plan

1. sourceQueue happy path — push 100 events; consumer receives all 100 in order.
2. sourceQueue backpressure — bufferSize=10, push 100; verify offer blocks once buffer fills.
3. sourceQueue drop- strategies* — each overflow strategy verified independently.
4. mergeHub fan-in — 10 producers × 100 events each = 1000 events; consumer receives all 1000 (any interleaving).
5. mergeHub slow-producer isolation — one producer pauses; others continue without blocking.
6. broadcastHub fan-out — 1 producer × 100 events; 5 late-attached consumers all receive all 100.
7. broadcastHub slow-consumer — one consumer ignores; verify per-consumer buffer fills + drop-head fires.
8. Lifecycle — complete() signals all consumers; fail(err) propagates to all.
9. Stress test — 10K events through each hub; memory + time bounded.
10. Metrics — gauges + counters emitted correctly.
11. Cross-runtime parity — same tests pass under Bun, Node, Deno.

Acceptance criteria

• src/streams/ module: sourceQueue, mergeHub, broadcastHub exported.
• Bounded buffers; backpressure semantics match Akka where applicable.
• Metrics emitted (queue depth, drops, latency).
• Documentation: Akka-to-actor-ts cheat-sheet + 3 worked examples (HTTP queue, WS merge, live broadcast).
• Cross-runtime tests pass.
• CHANGELOG entry under "New: Akka-Streams DSL subset (hubs)".

Pre-implementation checklist

Because this is XL, the implementation should not start without a dedicated planning slot:

• Joint review of the design sketch above.
• Resolve the 4 open design questions.
• Confirm scope cuts (no full Streams library, no async-boundary primitives).
• Pick implementation order: SourceQueue first (foundational) → MergeHub → BroadcastHub.

[pathosDev]

Cross-reference: #54 stays open as the full Source / Flow / Sink DSL track.

This issue remains the near-term ~5% subset (SourceQueue, MergeHub, BroadcastHub) as its body describes. Neither supersedes the other — the "Legacy tracking: roadmap #54" line here refers to the roadmap entry, not to #54 being obsolete.