Macula SDK — Streaming Protocol

The raw wire primitives underneath macula_streamer / macula_stream_sink.

Audience: building something the supervised wrappers don't fit — custom retry logic, observability, an SDK for another language. Most applications want the Streaming Guide instead — it covers the same capability via macula_streamer/macula_stream_sink, with an addressable pid, cancel, and mesh facts already wired in.

Opening a stream

Two ways to open one, mirroring unary RPC:

• call_stream/5: resolves the procedure's provider through its procedure_advertisement and opens the stream at its serving station, naming the provider as the target, exactly as call/5 does for unary RPC. Good when you don't know or care which station serves it.
• call_stream_station/7 (direct-dial): dials a specific station and opens the stream there in one hop, naming a provider's node_id as its target, exactly like call_station/7 for unary RPC. Use it after resolving a provider's procedure_advertisement and station_endpoint in the DHT (see the RPC Guide), so a stream reaches its provider the same way a unary call does.

Every STREAM_OPEN names its target. A provider serves only an open addressed to its own node_id; one for another node is dropped, and its stream closes with nothing written.

{ok, Stream} = macula:call_stream_station(Pool, StationUrl, ProviderNodeId, Realm,
                                          Procedure, Args, #{advertisement => Advertisement}).

Since 13.0.0 a stream to an explicit station decides to seal exactly as call_station/8 does: from the provider's verified advertisement, or confidential => required | off, and is otherwise refused as {error, {confidentiality, no_signed_state}} (see the RPC Protocol).

Since 13.1.0 macula:stream_report(Stream) tells the caller whether its stream went sealed: {ok, #{sealed := 0 | 1, provider := Provider, seal_key_id => KeyId}}. It settles on the provider's first chunk or reply opened under the stream's key (on a clear stream, its first chunk, reply or end); before that, and on a stream that ended first, an error included, it is {error, not_settled}. A served stream has none ({error, not_a_caller}). report in a stream's options is refused ({error, {invalid_option, report}}): a stream reports only through stream_report/1. See the design, §3.

Opts may set dial_timeout_ms (default 10_000) for the dial and handshake, plus the same per-call TLS trust override as call_station/8: verify and expected_node_id (see the RPC Guide). pin_tls_cert => true is REFUSED with {error, {refused, {pin_tls_cert, no_pin_primitive_for_mldsa87_identity}}}; false and an absent key pass. See macula#15.

This is what macula_streamer/macula_stream_sink wrap — an addressable pid you can monitor and cancel, streaming.*_v1 mesh facts around each session. Reach for the raw calls below directly only if you're building something the wrapper doesn't fit.

Consumer side

server_stream — read a feed

{ok, Stream} = macula:call_stream(Pool, Realm, <<"live.feed">>, Request, #{}),
loop(Stream).

loop(Stream) ->
    case macula:recv(Stream) of
        {chunk, Bin}  -> handle(Bin), loop(Stream);   %% raw bytes
        {data, Term}  -> handle(Term), loop(Stream);  %% decoded (msgpack)
        eof           -> ok;                           %% source stopped
        {error, R}    -> {error, R}                    %% e.g. peer_down -> re-open
    end.

recv/1 blocks for the next chunk; recv/2 takes a timeout. eof means the source closed the stream cleanly.

client_stream — push then await the result

{ok, Stream} = macula:call_stream(Pool, Realm, <<"bulk.ingest">>, Meta,
                                  #{mode => client_stream}),
[ok = macula:send(Stream, Chunk) || Chunk <- Chunks],
ok = macula:close_send(Stream),               %% signal "no more input"
{ok, Result} = macula:await_reply(Stream).    %% the provider's final reply

send/2 sends raw bytes; send/3 takes an encoding (raw | msgpack). close_send/1 half-closes your direction; await_reply/1,2 returns the provider's single final result.

Provider side

Advertise a streaming procedure with a mode and a fun(Stream, Args) handler. The handler drives the stream with the same send / recv primitives, and ends it with set_reply (a final result) or abort (an error).

The handler's process owns the stream, and the stream ends when that process ends: when the handler returns or crashes. A handler that lets another process keep using the stream hands it over first, and the stream then ends with that process:

ok = macula:advertise_stream(
       Pool, Realm, <<"live.feed">>, server_stream,
       fun(Stream, _Args) ->
           Feeder = spawn(fun() -> feed(Stream) end),
           ok = macula_stream:controlling_process(Stream, Feeder)
       end),

When a session ends (both sides closed, an abort, or the link lost), the stream's owner gets {macula_stream, ended, Stream, How} once, with How being closed, {error, {Code, Message}} or peer_down. An owner the stream is handed to after that is told at once. macula_streamer takes its stream over this way and stops on that message.

The stream process stays until its owner ends, so the owner can still read chunks already received and the reply after the end. A handler must end once its session has ended: on that message, or when a read returns eof or an error such as peer_down. One that runs on keeps its stream, and its session slot (below) is not freed until it ends. A caller that goes away ends the session for the provider (peer_down), so only the provider's own handler can keep a slot this way, never a caller (#64 F1, F3).

The provider verifies each STREAM_OPEN's signature against its caller before the handler runs; a STREAM_OPEN that does not verify runs no handler. advertise_stream/6 takes an auth policy in Opts, the same policies as advertise/5. A STREAM_OPEN the policy refuses gets a STREAM_ERROR with code unauthorized and runs no handler. A consumer presents its token with call_stream/5's ucan_token opt. See the Authorization Guide for the policies.

%% server_stream: push N chunks, then CLOSE — that is what produces `eof'
%% for a consumer looping on `recv'. `set_reply' is for client_stream /
%% bidi (see below); a pure push-only server_stream does not use it.
ok = macula:advertise_stream(
       Pool, Realm, <<"live.feed">>, server_stream,
       fun(Stream, _Args) ->
           lists:foreach(fun(Frame) -> macula:send(Stream, Frame) end, frames()),
           macula:close_stream(Stream)
       end),

%% client_stream: drain the consumer's chunks, then reply
ok = macula:advertise_stream(
       Pool, Realm, <<"bulk.ingest">>, client_stream,
       fun(Stream, _Args) ->
           N = drain(Stream, 0),
           macula:set_reply(Stream, #{ingested => N})
       end),

drain(Stream, N) ->
    case macula:recv(Stream) of
        {chunk, Bin} -> store(Bin), drain(Stream, N + 1);
        eof          -> N;
        {error, _}   -> N
    end.

close_stream vs. set_reply — do not mix them for server_stream. close_stream/1 is what makes a consumer's recv loop see eof. set_reply/2 only resolves await_reply/1,2; it does not close the stream. A server_stream handler that calls set_reply without also closing leaves a consumer's recv-until-eof loop waiting forever — use close_stream for a pure push, and reserve set_reply + await_reply for client_stream / bidi, where the consumer already knows to stop sending and ask for the result instead of draining chunks.

Abort with a BOLT#4-style code and message when something goes wrong:

macula:abort(Stream, <<"0F">>, <<"source unavailable">>).

Over the mesh the message is text for people, at most 256 bytes of UTF-8; a longer message, or one that is not valid UTF-8, travels empty. An error reply from macula_stream:set_error/2 travels as the same STREAM_ERROR, with code error and the reason as its message when the reason is a binary or an atom.

A stream takes chunks only from the side its mode lets send: the server in server_stream, the client in client_stream, and both in bidi. A caller's chunk in a server_stream is refused as a malformed frame, charged to the connection that carried it, and reaches no reader; the session goes on. A send the mode does not allow returns {error, {send_not_allowed, Mode}} without sending anything.

A provider serves at most 16 sessions per verified caller and 1000 on the node at once (max_served_sessions_per_caller and max_served_sessions in the macula application env), and one session per dedicated stream.

A provider serves a STREAM_OPEN once per caller and request id. A copy of an admitted open gets a STREAM_ERROR with code request_copy, and an open the node's request admission refuses gets the refusal's name: not_yet_valid or expired for a deadline outside the window it accepts (up to 10 minutes ahead, 5 minutes past), request_id_reused, caller_quota, share_full, admission_full or reply_not_kept. An open past a session cap gets too_many_sessions, one its procedure's policy refuses gets unauthorized, one for a procedure the provider does not advertise gets not_found, one in a mode other than the one its procedure is advertised in gets mode_mismatch, and a second STREAM_OPEN on a stream that already carries a session gets refused. When the node's session counter or its request admission does not answer, an open gets unavailable, and a later try may be served.

Each refusal is the provider's first frame under the open it refuses, signed and verifiable like any other. A refusal on a stream that carries no session also closes that stream, so open each session on a new stream.

A stream a peer opens must start with a STREAM_OPEN that verifies and names this node as its target, and bring it within 10 seconds (dedicated_stream_open_timeout_ms). A stream whose first frame is anything else, whose STREAM_OPEN does not verify or names another node, or that stays silent that long closes without a STREAM_ERROR.

A STREAM_OPEN is at most 1 MiB (max_stream_open_bytes). A longer first frame closes its stream the same way, and call_stream refuses such an open before sending anything, with {error, {open_too_large, Limit}}. Send bulk data as chunks once the stream is open. Each node reads its own setting: a provider whose max_stream_open_bytes is lower than a caller's closes a larger open without an answer, and the caller waits out its deadline, so keep the default unless both sides change it.

A stream holds at most 16 MiB of memory for chunks no reader has taken, counting each chunk's bytes, a decoded term's heap size and the cell that queues it. The streams a provider serves also share a budget for those chunks: one caller's streams together at most 16 MiB (max_served_inbox_bytes_per_caller), and all served streams on the node at most 256 MiB (max_served_inbox_bytes). A chunk that would take a stream past its bound, or a budget past its limit, ends the session with code resource_exhausted. A reader that takes chunks as they come never meets either.

The two codes tell a peer different things. stream_protocol_error says it broke the protocol, and sending the same again fails the same way. resource_exhausted says the receiving side had no room for what it sent, which a well-behaved sender cannot see coming, and a later session may succeed.

A queued chunk is a copy, but a chunk handed straight to a waiting reader can still be part of the frame it arrived in. A handler that keeps chunks after reading them, in a list, a table or its state, should copy what it keeps (binary:copy/1), or it keeps whole frames alive.

Local (in-process) streams

macula:open_stream/3,4, macula:advertise_stream/2,3, and call_stream/2,3 drive streams inside one BEAM (no mesh), backed by macula_stream_local. They are for unit tests and same-node dispatch. The pool forms (call_stream/5, advertise_stream/5,6) are the ones that go over the mesh.

Reference

See also

• STREAMING_GUIDE.md — the supervised wrappers most applications should use instead of these raw primitives.