Working with Graphs

The GraphDSL layer lets you build arbitrary fan-in / fan-out topologies that the linear Source → Flow → Sink API cannot express: diamonds, multi-port junctions, and reusable partial graphs, including bounded feedback loops. It mirrors the Akka Streams GraphDSL API.

Building a static typed graph

For static Rust topologies, prefer GraphDsl::typed_try_create. It uses the consuming TypedGraphBuilder: add_flow and fixed-arity junction helpers return move-only open-port tokens, and connect consumes the two tokens it wires. Reusing the same port token is a Rust use-after-move compile error.

use datum::{GraphDsl, GraphFlowShape, Identity, MapStage};

let graph = GraphDsl::typed_try_create(|builder| {
    let (builder, input, first_out) = builder.add_flow(Identity::<u64>::new());
    let (builder, second_in, output) = builder.add_flow(MapStage::new(|item: u64| item + 1));

    let builder = builder.connect(first_out, second_in)?;

    Ok((builder, GraphFlowShape::new(input.into_port(), output.into_port())))
})?;

Fixed two-port junction helpers cover common acyclic shapes:

use datum::{GraphDsl, GraphFlowShape};

let graph = GraphDsl::typed_try_create(|builder| {
    let (builder, input, [left_out, right_out]) = builder.add_broadcast2::<u64>();
    let (builder, left_in, right_in, output) = builder.add_zip::<u64, u64>();

    let builder = builder.connect(left_out, left_in)?;
    let builder = builder.connect(right_out, right_in)?;

    Ok((builder, GraphFlowShape::new(input.into_port(), output.into_port())))
})?;

The consuming builder produces the same immutable GraphBlueprint as the mutable builder. Execution still starts only when a run_* method is called, and ExecutorMode::Auto still chooses the same typed-linear, typed-junction, cyclic, or erased executor tier from the finished blueprint.

Building a dynamic graph

Use GraphDsl::try_create and the mutable GraphBuilder when the topology is dynamic, cyclic, imported from another graph, or needs erased interop through connect_any/wire. It opens a builder closure and returns StreamResult<GraphBlueprint<S>>:

use datum::{Broadcast, GraphDsl, GraphFlowShape, Merge, WireDsl};

let graph = GraphDsl::try_create(|builder| {
    let bcast = builder.add(Broadcast::<u64>::new(2));
    let merge = builder.add(Merge::<u64>::new(2));

    builder.wire(bcast.to(&merge)).wire(bcast.to(&merge));

    Ok(GraphFlowShape::new(bcast.inlet(), merge.outlet()))
})?;

Inside the mutable-builder closure:

• builder.add(stage) allocates the stage's typed ports and returns its shape.
• builder.wire(shape.to(&other)) wires the first currently unconnected outlet to the first currently unconnected inlet in declaration order; errors are collected and reported at graph creation.
• builder.try_wire(spec)? keeps the same method DSL but returns the wiring error immediately.
• The closure must return a Shape that describes the graph's external ports.

Use explicit cursors when the automatic port choice is not the topology you want:

use datum::{Broadcast, GraphDsl, GraphFlowShape, Zip, WireDsl};

let graph = GraphDsl::try_create(|builder| {
    let bcast = builder.add(Broadcast::<u64>::new(2));
    let zip = builder.add(Zip::<u64, u64>::new());

    builder
        .wire(bcast.out(1).to(&zip.in_(1)))
        .wire(bcast.out(0).to(&zip.in_(0)));

    Ok(GraphFlowShape::new(bcast.inlet(), zip.outlet()))
})?;

MergePreferredShape and BidiShape are explicit-only in the wiring DSL. Use merge.preferred(), merge.secondary(i)?, existing typed accessors, or .out(i) / .in_(i) cursors for those shapes.

Two equivalent wiring styles

wire and connect are both first-class and fully supported on the mutable builder — they share the same runtime validation and produce identical graphs. Pick whichever reads best for dynamic construction, and mix them freely in one builder closure:

• Convenient: builder.wire(shape.to(&other)) auto-selects the next open ports, chains, and defers wiring errors to graph creation. Best for readable, common topologies.
• Explicit: builder.connect(outlet, inlet)? takes typed ports directly and returns an immediate StreamResult<()>. builder.connect_any(outlet, inlet)? is the explicit erased interop path for runtime-typed ports. This original API remains available for dynamic graphs.

The broadcast→merge graph above, written explicitly:

use datum::{Broadcast, GraphDsl, GraphFlowShape, Merge};

let graph = GraphDsl::try_create(|builder| {
    let bcast = builder.add(Broadcast::<u64>::new(2));
    let merge = builder.add(Merge::<u64>::new(2));

    builder.connect(bcast.outlet(0)?, merge.inlet(0)?)?;
    builder.connect(bcast.outlet(1)?, merge.inlet(1)?)?;

    Ok(GraphFlowShape::new(bcast.inlet(), merge.outlet()))
})?;

Shapes

A Shape lists the graph's external inlets and outlets. Common shapes:

Shape	External ports
FlowShape<In, Out>	1 inlet, 1 outlet — use for a graph that replaces a Flow
SourceShape<Out>	0 inlets, 1 outlet — a graph that acts as a source
SinkShape<In>	1 inlet, 0 outlets — a graph that acts as a sink
FanInShape<In, Out>	N inlets, 1 outlet
FanOutShape<In, Out>	1 inlet, N outlets

GraphFlowShape is a re-export alias for FlowShape.

Junctions

All built-in junctions are re-exported from datum:

Junction	Inputs	Outputs	Semantics
Broadcast<T>::new(n)	1	N	Push each element to all N outlets
Balance<T>::new(n)	1	N	Round-robin across N outlets
Merge<T>::new(n)	N	1	Merge from whichever inlet has an element
MergePreferred<T>::new(n_secondary)	1+N	1	Always drain preferred inlet first
MergePrioritized<T>::new(weights)	N	1	Deterministic weighted merge — drains inlet i weights[i] times per cycle (Datum uses a fixed schedule, not Akka's randomized pick)
MergeSequence<T>::new(n, extract_seq)	N	1	Ordered-by-sequence-number merge
MergeSorted<T>::new()	2	1	Sorted merge of two pre-sorted streams
MergeLatest<T>::new(n, eager_complete)	N	1	Emit a Vec<T> snapshot of the latest element per inlet
OrElse<T>::new()	2	1	Emit the primary inlet; fall back to the secondary only if the primary is empty
Zip<L, R>::new()	2	1	Pair elements from two inlets
Unzip<A, B>::new()	1	2	Split (A, B) pairs to two outlets
UnzipWith<In, A, B>::new(f)	1	2	Split each element using f
Partition<T>::new(n, f)	1	N	Route each element to outlet f(&elem)
Concat<T>::new(n)	N	1	Drain inputs in order
Interleave<T>::new(n, segment_size)	N	1	Round-robin in chunks of segment_size
Buffer<T>::new(n, strategy)	1	1	Bounded graph-stage buffer
TakeWhile<T>::new(predicate)	1	1	Pass elements while predicate(&elem) is true, then close

Shape accessors

Each junction returns a shape from builder.add(...). The accessors vary by shape type:

• FanOutShape (Broadcast, Balance, Partition): .inlet(), .outlet(i)?, .outlet_count()
• FanOutShape2 (Unzip, UnzipWith): .inlet(), .out0(), .out1()
• FanInShape (Merge, Concat, Interleave, OrElse, and the Merge* variants): .inlet(i)?, .outlet(), .inlet_count()
• ZipShape: .in0(), .in1(), .outlet()
• MergePreferredShape: .preferred(), .secondary(i)?, .outlet()

Running a graph

A GraphBlueprint<FlowShape<In, Out>> exposes run_with_input(iter) to execute the graph directly from an iterator:

use datum::{Broadcast, GraphDsl, GraphFlowShape, Merge, WireDsl};

// Build a broadcast→merge diamond: each input element is broadcast to two
// Merge inlets, so every element appears twice in the output.
let graph = GraphDsl::try_create(|builder| {
    let bcast = builder.add(Broadcast::<u64>::new(2));
    let merge = builder.add(Merge::<u64>::new(2));

    builder.wire(bcast.to(&merge)).wire(bcast.to(&merge));

    Ok(GraphFlowShape::new(bcast.inlet(), merge.outlet()))
})
.unwrap();

let mut items: Vec<u64> = graph.run_with_input(1_u64..=3).unwrap();
items.sort_unstable();

The broadcast→merge diamond sends each input element to both Merge inlets, so every element appears twice in the output. Order across inlets is not guaranteed; sort if you need a stable assertion.

run_with_input accepts any IntoIterator<Item = In>. Use run_count_with_input to get only the output count, or run_fold_with_input(iter, init, f) to fold the outputs into a single accumulated value without collecting them.

Cycles

Datum supports fused cyclic graph topologies such as MergePreferred/Broadcast feedback loops. Cycles are still bounded by the fused executor event budget: a cycle that cannot make progress surfaces StreamError::EventLimitExceeded instead of hanging the caller.

use datum::{
    Broadcast, Buffer, GraphDsl, GraphFlowShape, MapStage, MergePreferred, OverflowStrategy,
    TakeWhile, WireDsl,
};

// Feedback values count down through a bounded feedback path. TakeWhile
// closes that path at zero, so the cycle completes instead of hanging.
let graph = GraphDsl::try_create(|builder| {
    let merge = builder.add(MergePreferred::<u64>::new(1));
    let bcast = builder.add(Broadcast::<u64>::new(2));
    let buffer = builder.add(Buffer::<u64>::new(8, OverflowStrategy::Backpressure));
    let positive = builder.add(TakeWhile::<u64>::new(|item| *item > 0));
    let decrement = builder.add(MapStage::new(|item: u64| item - 1));

    builder
        .wire(merge.out(0).to(&bcast))
        .wire(bcast.out(1).to(&buffer))
        .wire(buffer.to(&positive))
        .wire(positive.to(&decrement))
        .wire(decrement.to(&merge.preferred()));

    Ok(GraphFlowShape::new(merge.secondary(0)?, bcast.outlet(0)?))
})
.unwrap();

let items = graph.run_with_input([3]).unwrap();

Like Akka, a productive cycle needs a liveness break: elements must leave the loop, the feedback path must close, or a bounded/dropping stage must recover space. MergePreferred keeps feedback moving ahead of secondary inputs, but a naive unbounded Merge + Broadcast loop has no way to recover buffer space; run it with an explicit FusedExecutionConfig event limit if you are probing that behavior.

Partial graphs

GraphDsl::partial builds a reusable fragment that can be imported into multiple parent graphs. The returned PartialGraph<S> is a blueprint that is wired in with builder.import(&partial):

use datum::{GraphDsl, PartialGraph, GraphFlowShape, MergePrioritized};

let fragment: PartialGraph<GraphFlowShape<u64, u64>> = GraphDsl::partial(|builder| {
    let stage = builder.add(MergePrioritized::<u64>::new(vec![3, 1]));
    Ok(GraphFlowShape::new(stage.inlet(0)?, stage.outlet()))
});

// Use in another graph:
let outer = GraphDsl::try_create(|builder| {
    let shape = builder.import(&fragment)?;
    // wire imported shapes with builder.wire(shape.to(&other)) or explicit ports
    Ok(shape)
})?;

Performance

Datum's fused executor has two paths (all numbers are same-host, 2026-06-11; full table in roadmap/benchmarks/graph.md):

• Typed-linear fast path: applies to straight linear chains of type-safe stages. Avoids all DatumValue boxing — 38–44× faster than Akka on the benchmarked typed identity/count shapes; 13–17× on map/fold shapes.
• Auto-typed erased path: WP-18 extended typed plan selection via ExecutorMode::Auto to graphs that previously fell to the erased executor. Rows that formerly measured 0.28–0.31× Akka now auto-select the typed plan at 28–36× (identity) and 13–17× (map). The M1 Tier-1 performance gate is closed.
• Junctions: WP-P1/P1b added typed acyclic dispatcher and typed helper execution, raising junction throughput dramatically:
  • Concat: ~302–340×; MergePreferred: ~137–139×; Broadcast+Zip: ~67–118×; Balance+Merge: ~52–95×; UnzipWith: ~38–49×; MergePrioritized: ~39–41×; Partition: ~34–41×; Interleave: ~29–30×; MergeSorted: ~9–13×; MergeSequence: ~7.5–8.7×; MergeLatest: ~2.2–2.3×.
  • All above Akka parity.
• Cycles (MergePreferred/Broadcast feedback): ~29× Akka after the opt-cycles typed kernel (was 0.16× with the initial correctness-first erased interpreter).
• BidiFlow: join ~3.7–4.0×; atop ~5.0–5.5×.
• Graph build: parity range (0.94–1.68× depending on depth); junction-chain build ahead.

Next steps

• Materialization — how blueprints become running streams
• Buffers & Rate — decoupling producer and consumer speed
• Dynamic Streams — MergeHub, BroadcastHub, and kill switches