Configuration

Datum's runtime and materialization surface is intentionally small: there is no HOCON file, no actor-system configuration, and no dispatcher table. Configuration happens through the Runtime type and the Attributes API.

---

Runtime and Materializer

Runtime is the entry point for all stream execution. Materializer is a type alias for Runtime — they are the same type:

pub type Materializer = Runtime;

You can use either name; Materializer is the conventional alias when passing the runtime to graph constructors, mirroring Akka's naming.

Creating a runtime

use datum::Runtime;

let runtime = Runtime::new();

Runtime::new() starts a shared timer thread (datum-tmr-N) and a process-wide lazy thread pool (datum-stream-runtime). The timer is used by time-aware operators (delay, throttle, grouped_within, timeouts, etc.). The thread pool runs each materialized stream's drain loop on a dedicated worker thread.

The process-wide thread pool is shared across all Runtime instances. Workers are spawned on demand (one per concurrent materialized stream), parked when idle, and reaped after a 10-second idle timeout. A spawn failure falls back to running the stream inline on the caller's thread.

Configuring a runtime

Runtime is Clone. Both methods return a new Runtime that shares the same underlying timer and thread pool:

let runtime = Runtime::new()
    .with_name_prefix("my-app")          // names worker threads "my-app-N"
    .with_attributes(Attributes::named("root"));  // default attributes for all materialized streams

Method	Effect
with_name_prefix(prefix)	Prefix for thread and stream names (default: "datum-stream")
with_attributes(attrs)	Default Attributes applied to every materialized stream; merged with per-graph attributes
name_prefix()	Returns the current prefix
attributes()	Returns the current default attributes
effective_attributes(local)	Merges default + local attributes (later attributes win)

Materializing a graph

Use materializer.materialize(graph) to run a RunnableGraph explicitly:

let completion = materializer.materialize(&runnable_graph)?;

Source::run_with_materializer(sink, materializer) and RunnableGraph::run_with_materializer(mat) are convenience shortcuts that do the same thing.

Lifecycle

Method	Effect
shutdown()	Sets the shutdown flag; the timer stops and no new streams can be materialized
is_shutdown()	Returns true after shutdown() is called
active_streams()	Number of streams currently running under this runtime

A runtime that has been shut down returns Err(StreamError::AbruptTermination) on any new materialization attempt. There is no restart — create a new Runtime instance after shutdown.

Timers

Runtime exposes three timer methods. All return a Cancellable handle:

let cancellable = runtime.schedule_once(Duration::from_secs(5), || println!("fired"));

let cancellable = runtime.schedule_with_fixed_delay(
    Duration::from_millis(100),  // initial delay
    Duration::from_secs(1),      // delay between completions
    || println!("tick"),
);

let cancellable = runtime.schedule_at_fixed_rate(
    Duration::from_millis(100),  // initial delay
    Duration::from_secs(1),      // interval between firings
    || println!("rate tick"),
);

These are the same timer primitives used internally by tick, delay, throttle, and the timeout operators. schedule_with_fixed_delay waits the given delay after each task completes; schedule_at_fixed_rate fires at the given interval regardless of task duration.

---

Attributes and Attribute

Attributes is a list of Attribute values that annotate a blueprint or stage. They are hints to the materializer — not enforced by the type system — and are merged when a graph is materialized.

Available attributes

pub enum Attribute {
    Name(Arc<str>),
    InputBuffer { initial: usize, max: usize },
    Dispatcher(Arc<str>),
}

Attribute variant	Constructor	Effect
Attribute::Name	Attributes::named("my-stage")	Labels a stage or graph segment; visible in thread names and debug output
Attribute::InputBuffer { initial, max }	Attributes::input_buffer(initial, max)	Hint for internal buffer sizing on detached stages
Attribute::Dispatcher	Attributes::dispatcher("my-dispatcher")	Named dispatcher hint; currently advisory — dispatcher selection is not implemented beyond the default thread pool

Creating and composing attributes

use datum::Attributes;

// Single attribute
let attrs = Attributes::named("ingestion");
let attrs = Attributes::input_buffer(16, 64);
let attrs = Attributes::dispatcher("blocking");

// Combine two Attributes (right-hand side wins on conflict)
let combined = Attributes::named("foo").and(Attributes::input_buffer(8, 32));

// Empty (no-op)
let none = Attributes::none();

Applying attributes to a blueprint

Every Source, Flow, Sink, and RunnableGraph has three attribute methods:

// Replace all attributes
source.with_attributes(attrs)

// Append attributes (merged with existing)
source.add_attributes(Attributes::named("debug"))

// Shorthand for a name attribute
source.named("my-source")

Attributes set on a Runtime (via with_attributes) act as defaults; per-graph or per-stage attributes are merged on top using effective_attributes(local).

Reading attributes

let attrs = source.attributes();

// Get the first name, if any
let name: Option<&str> = attrs.name();

// Get the input buffer hint, if any
let hint: Option<(usize, usize)> = attrs.input_buffer_hint();

// Get the dispatcher hint, if any
let dispatcher: Option<&str> = attrs.dispatcher_hint();

// Inspect the raw list
let list: &[Attribute] = attrs.attribute_list();

---

Async boundaries

In the async-boundary execution path, an async boundary decouples two fused regions with a bounded handoff. In linear streams, use Source::async_boundary() or Flow::async_boundary(); .r#async() is available as the raw-identifier alias for Akka's .async name. Use async_boundary_with_config(AsyncBoundaryExecutionConfig { ... }) when you need to set the bounded handoff size.

use datum::{AsyncBoundaryExecutionConfig, Sink, Source};

let values: Vec<u64> = Source::from_iter(1_u64..=3)
    .map(|item| item + 1)
    .async_boundary_with_config(AsyncBoundaryExecutionConfig {
        buffer_size: 4,
        ..AsyncBoundaryExecutionConfig::default()
    })
    .map(|item| item * 2)
    .run_with(Sink::collect())
    .unwrap()
    .wait()
    .unwrap();

In GraphDsl, use the AsyncBoundary<T>::new() graph stage:

use datum::{
    AsyncBoundary, AsyncBoundaryExecutionConfig, Attributes, FusedExecutionConfig, GraphDsl,
    GraphFlowShape, MapStage,
};

let graph = GraphDsl::try_create(|builder| {
    let first = builder.add(MapStage::new(|item: u64| item + 1));
    let boundary = builder
        .add_with_attributes(AsyncBoundary::<u64>::new(), Attributes::input_buffer(4, 8));
    let second = builder.add(MapStage::new(|item: u64| item * 2));

    builder.connect(first.outlet(), boundary.inlet())?;
    builder.connect(boundary.outlet(), second.inlet())?;

    Ok(GraphFlowShape::new(first.inlet(), second.outlet()))
})
.unwrap();

let output = graph.run_with_input([1, 2, 3]).unwrap();
let report = graph
    .run_async_boundary_count_with_input_report(
        1_u64..=3,
        AsyncBoundaryExecutionConfig {
            fused: FusedExecutionConfig { event_limit: 1024 },
            buffer_size: 4,
        },
    )
    .unwrap();

Attach Attributes::input_buffer(initial, max) with GraphBuilder::add_with_attributes when the boundary stage should carry an input-buffer hint. The explicit async-boundary execution/report path takes AsyncBoundaryExecutionConfig; its buffer_size controls the bounded handoff queue, and its fused field controls each fused region's event limit.

---

Fused executor event limit

Every fused stream region has an event budget, configured via FusedExecutionConfig::event_limit (default: 100_000_000). The executor bumps an atomic counter on each event (push/pull/complete); when the limit is reached, the stream fails with StreamError::EventLimitExceeded. The counter is reset when an AsyncBoundary hand-off transfers work to a new fused region.

This is a safety cap, not a throughput throttle. The 100M default is adequate for normal workloads. Tighten it for bounded loops or when you suspect a non-productive feedback cycle (a Merge + Broadcast loop with no exit will hit the limit instead of hanging). Increase it if you have legitimately long-running single-region streams.

The event limit is surfaced on FusedExecutionConfig, which is passed to benchmark/report execution paths (run_with_input_report, run_typed_linear_with_input_report, etc.) and to AsyncBoundaryExecutionConfig::fused.

StreamRef settings

StreamRef endpoints (StreamRefs::source_ref / sink_ref) accept an optional StreamRefSettings, configured via builder methods:

use datum::{StreamRefs, StreamRefSettings};
use std::time::Duration;

let settings = StreamRefSettings::default()
    .with_buffer_capacity(64)
    .with_subscription_timeout(Duration::from_secs(10))
    .with_demand_redelivery_interval(Duration::from_millis(500));

let sink = StreamRefs::source_ref_with_settings::<u64>(settings);

Method	Default	Effect
with_buffer_capacity(n)	32	Receive buffer size on the remote end
with_subscription_timeout(d)	30 s	Maximum time to wait for the remote peer to subscribe
with_demand_redelivery_interval(d)	1 s	Interval between demand re-signals (cumulative; idempotent)

The defaults mirror Akka StreamRefs. Use the *_with_settings constructors to override them; the plain source_ref() / sink_ref() constructors use StreamRefSettings::default().

No configuration file

Datum has no HOCON file and no environment-variable dispatcher configuration. There is no application.conf, no named dispatcher table, and no fork-join pool configuration. This is a deliberate design choice: the crate targets embedded use cases and benchmarks where transparent, predictable behavior matters more than runtime reconfigurability.

If you need to bound concurrency, use map_async(parallelism, ...). If you need to isolate a slow stage, insert an AsyncBoundary with an InputBuffer hint. If you need a different thread pool, materialize on a custom Runtime with a distinct name prefix.