StreamRefs

StreamRefs are one-shot streaming handles. A local stream can materialize a SourceRef<T> or SinkRef<T>, pass that handle across an actor boundary, and let the other side attach exactly one stream endpoint while preserving completion, failure, cancellation, and backpressure.

SourceRef

StreamRefs::source_ref::<T>() is a Sink<T, SourceRef<T>>: materializing it consumes a local source and returns a SourceRef<T>. The receiving side calls SourceRef::source() to turn that handle back into a Source<T, NotUsed>.

Use StreamRefs::source_ref_with_settings(settings) to override the default settings.

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

let settings = StreamRefSettings::default()
    .with_buffer_capacity(4)
    .with_subscription_timeout(Duration::from_secs(5))
    .with_demand_redelivery_interval(Duration::from_millis(250));

let source_ref = Source::from_iter([1_u64, 2, 3])
    .run_with(StreamRefs::source_ref_with_settings(settings))
    .unwrap();

let values = source_ref.source().run_collect().unwrap();

The ref only feeds elements as the source_ref.source() side demands them. The configured buffer allows a bounded amount of run-ahead.

SinkRef

StreamRefs::sink_ref::<T>() is the dual: it is a Source<T, SinkRef<T>>. Materializing it returns a SinkRef<T>, and the other side uses SinkRef::sink() to feed the local source.

use datum::{Keep, Sink, Source, StreamRefs};

let (sink_ref, completion) = StreamRefs::sink_ref::<u64>()
    .to_mat(Sink::collect(), Keep::both)
    .run()
    .unwrap();

Source::from_iter([4_u64, 5, 6])
    .run_with(sink_ref.sink())
    .unwrap()
    .wait()
    .unwrap();

let received = completion.wait().unwrap();

Use StreamRefs::sink_ref_with_settings(settings) when the local source endpoint needs non-default subscription or buffering behavior.

Settings

StreamRefSettings::default() mirrors Akka StreamRefs defaults:

Setting	Default	Method
Buffer capacity	32 elements	with_buffer_capacity(capacity)
Subscription timeout	30 seconds	with_subscription_timeout(timeout)
Demand redelivery interval	1 second	with_demand_redelivery_interval(interval)

Demand is cumulative, so redelivery is idempotent. buffer_capacity must be greater than zero.

One-shot pairings

Each SourceRef or SinkRef can be attached once. A second SourceRef::source() materialization, or a second SinkRef::sink() materialization, fails with StreamError::Failed.

Failures propagate across the ref instead of panicking:

use datum::{Source, StreamError, StreamRefs};

let source_ref = Source::<u64>::failed(StreamError::Failed("boom".to_owned()))
    .run_with(StreamRefs::source_ref())
    .unwrap();

assert_eq!(
    source_ref.source().run_collect(),
    Err(StreamError::Failed("boom".to_owned()))
);
assert!(matches!(
    source_ref.source().run_collect(),
    Err(StreamError::Failed(message)) if message.contains("already")
));

Cancellation

Cancellation also crosses the ref. If the local endpoint stops early, the producer attached through the ref observes cancellation.

use datum::{Keep, Sink, Source, StreamError, StreamRefs};

let (sink_ref, completion) = StreamRefs::sink_ref::<u64>()
    .take(1)
    .to_mat(Sink::collect(), Keep::both)
    .run()
    .unwrap();

let producer = Source::repeat(1_u64)
    .run_with(sink_ref.sink())
    .unwrap()
    .wait();

assert_eq!(completion.wait().unwrap(), vec![1]);
assert_eq!(producer, Err(StreamError::Cancelled));

Backpressure across the ref

The receiving side sends cumulative demand. The producer can run ahead only by the configured buffer capacity.

use datum::testkit::TestSink;
use datum::{Source, StreamRefSettings, StreamRefs};
use std::sync::{
    Arc,
    atomic::{AtomicUsize, Ordering},
};
use std::time::{Duration, Instant};

fn wait_until(timeout: Duration, mut condition: impl FnMut() -> bool) -> bool {
    let deadline = Instant::now() + timeout;
    while Instant::now() < deadline {
        if condition() {
            return true;
        }
        std::thread::park_timeout(Duration::from_millis(1));
    }
    condition()
}

fn assert_condition_holds(timeout: Duration, mut condition: impl FnMut() -> bool) {
    let deadline = Instant::now() + timeout;
    while Instant::now() < deadline {
        assert!(condition());
        std::thread::park_timeout(Duration::from_millis(1));
    }
    assert!(condition());
}

let pulled = Arc::new(AtomicUsize::new(0));
let pulled_for_source = Arc::clone(&pulled);
let source = Source::unfold(0_u64, move |next| {
    pulled_for_source.fetch_add(1, Ordering::SeqCst);
    Some((next + 1, next))
});
let settings = StreamRefSettings::default()
    .with_buffer_capacity(1)
    .with_subscription_timeout(Duration::from_secs(5))
    .with_demand_redelivery_interval(Duration::from_millis(20));
let source_ref = source
    .run_with(StreamRefs::source_ref_with_settings(settings))
    .unwrap();
let mut probe = source_ref.source().run_with(TestSink::probe()).unwrap();

probe.request(1);
probe.assert_next(0);
assert!(wait_until(Duration::from_secs(1), || {
    pulled.load(Ordering::SeqCst) >= 2
}));
assert_condition_holds(Duration::from_millis(50), || {
    pulled.load(Ordering::SeqCst) <= 2
});

probe.request(1);
probe.assert_next(1);
assert!(wait_until(Duration::from_secs(1), || {
    pulled.load(Ordering::SeqCst) >= 3
}));
assert_condition_holds(Duration::from_millis(50), || {
    pulled.load(Ordering::SeqCst) <= 3
});

probe.cancel();

Remote transport: TCP & QUIC

datum-core owns a first-class StreamRefs protobuf protocol and a transport-agnostic producer/consumer seam (the StreamRefProtoEndpoint, StreamRefProtoProducer, and StreamRefProtoConsumer in stream_ref_proto.rs). datum-net carries the frames over two transports:

Carrier	API entry points	Encrypted	Use case
Plaintext TCP	serve_source_ref_over_tcp / source_ref_over_tcp / serve_sink_ref_over_tcp / sink_ref_over_tcp	No	Trusted/loopback; Akka Artery-TCP-equivalent baseline.
Encrypted QUIC	serve_source_ref_over_quic / source_ref_over_quic / serve_sink_ref_over_quic / sink_ref_over_quic	Yes (TLS 1.3)	Untrusted/WAN.

The same-process direct splice remains the fast local path (~6.5× faster than Akka) and is untouched by the remote transport layer. When both sides run in the same process, Datum keeps the direct actor-boundary handoff and never serializes.

Protocol

The protocol mirrors Akka StreamRefs: OnSubscribeHandshake, cumulative demand, sequenced OnNext, completion, failure, cancel, and terminal Ack. Sequence numbers start at zero, the default receive window is 32 elements, and demand refills when remaining credit reaches the low-watermark of 16. Akka redelivers cumulative demand for lossy remoting; Datum does not redeliver on TCP or QUIC because both carriers are reliable and ordered.

Steady-state data uses a versioned compact carrier frame (high-bit length prefix, batched SequencedOnNext payloads with an element count and shared validation). Control messages (handshake, completion, ack) stay protobuf. The batch path shares the existing sequence/window/completion validation; there is no second state machine.

Element payloads are encoded through StreamRefPayload. Datum includes built-in payload codecs for primitive numbers, bool, String, and Vec<u8>; custom element types can implement the trait.

SourceRef Examples

Run the two-process SourceRef example over TCP from the workspace root:

cargo run -p datum-net --example streamref_quic_node -- --serve 127.0.0.1:5005 --count 1000

cargo run -p datum-net --example streamref_quic_node -- --connect 127.0.0.1:5005 --count 1000

The client prints deterministic output:

RECEIVED 1000 CHECKSUM 499500

Use --slow-ms <d> on the client to throttle downstream pulls and observe backpressure across the remote ref. The TCP variants mirror the same API shape:

use datum_net::stream_ref::serve_source_ref_over_tcp;

Remote SourceRefs are covered by in-process loopback tests for both TCP and QUIC: success/checksum, origin failure propagation, downstream cancellation with Ack, bounded backpressure, and first-demand subscription timeout.

SinkRef over TCP / QUIC

The SinkRef direction is symmetric with the SourceRef carrier, with the producer/consumer roles swapped: the remote/sender side returns a Sink<T, StreamCompletion<NotUsed>> whose incoming elements are framed and sent over the carrier (producer seam, lazy input); the local/receiver side surfaces inbound elements as a Source<T> (consumer seam) which the caller runs into a local Sink.

Because the consumer sends the OnSubscribeHandshake and cumulative demand first, the receiver must open the stream so those frames establish the connection before the sender accepts — the reverse of the SourceRef carrier. The sender accepts the receiver-opened stream and waits for demand before sending elements.

Run the two-process SinkRef example:

cargo run -p datum-net --example streamref_quic_node -- --serve 127.0.0.1:5005 --sink --count 1000

cargo run -p datum-net --example streamref_quic_node -- --connect 127.0.0.1:5005 --sink --count 1000

The server (receiver) collects inbound elements and prints:

RECEIVED 1000 CHECKSUM 499500

The client (sender) prints SENT 1000. SinkRef carriers are covered by the same in-process loopback test matrix as the SourceRef carriers: success/checksum, sender-failure propagation, receiver cancel reaching the sender, bounded backpressure, and subscription timeout.

Honest performance (vs forced-remote Akka)

After correcting the Akka comparison harness to force real Artery-TCP remoting (the prior table measured an in-JVM ActorRef/SourceRef handoff — 18.7× faster than actual remote), the honest forced-remote benchmark (1,024 u64/Long elements, fold-to-sum, 5× warmup + 5× measurement, whole-process CPU) yields:

Shape	Datum wall	Akka wall	Datum CPU	Akka CPU	Datum alloc	Akka alloc
TCP plaintext	1,951 µs/op	18,612 µs/op	3,149	46,293	133 KB/op	4.77 MB/op
TCP reuse	2,072 µs/op	18,895 µs/op	3,220	47,326	135 KB/op	4.90 MB/op
QUIC encrypted	6,611 µs/op	17,530 µs/op	10,013	44,788	3.38 MB/op	4.90 MB/op
QUIC reuse	5,169 µs/op	18,848 µs/op	7,546	47,634	188 KB/op	4.91 MB/op

Per-element (N-sweep fit, forced-remote TCP, fold-to-sum):

Side	Wall per element	CPU per element
Datum TCP	1.27 µs/elem	1.99 µs/elem
Akka forced-remote	17.0 µs/elem	36.7 µs/elem

On the fair plaintext TCP baseline, Datum is ~10× faster wall, ~15× lower CPU, and ~35× lower allocation than warmed forced-remote Akka. The QUIC carrier is TLS-encrypted (always; QUIC mandates TLS 1.3) and is not a plaintext-vs-plaintext comparison — the TCP row is the Artery-TCP-equivalent baseline.

The full benchmark record, including the N-sweep, the in-JVM-vs-forced-remote Akka delta, and per-axis verdicts, lives in roadmap/benchmarks/net.md.

Cross-process transport

SourceRef<T> and SinkRef<T> are direct in-memory handles inside one process. For cross-process handoff, use the TCP or QUIC StreamRefs carriers in datum-net; they drive Datum's protobuf StreamRefs protocol with StreamRefPayload codecs and do not depend on Ractor cluster support.

The old datum-core cluster/ractor_cluster feature and its BytesConvertable ref serialization path have been removed. The multi-node membership and placement layer is being built separately as datum-cluster for v0.10; remote StreamRefs over datum-net remain the data-plane path across process boundaries.

Next steps

• Actors Interop - actor sources, sinks, ask, and pub/sub
• Backpressure - Datum's demand model
• Configuration - StreamRef settings reference
• Network Carriers - datum-net TLS, UDP, QUIC, sharding, and io_uring