-spec abort(stream(), binary(), binary()) -> ok.

Abort the stream with an error frame.

-spec advertise(pool(), realm(), procedure(), macula_client:handler(), map()) -> ok | {error, term()}.

Advertise a procedure handler on a V2 pool. Fans out to every healthy link and stores in pool state for replay on link respawn. See macula_client:advertise/4.

-spec advertise_stream(procedure(), stream_handler()) -> ok | {error, term()}.

Advertise a LOCAL in-process streaming procedure (default: server_stream).

-spec advertise_stream(procedure(), stream_mode(), stream_handler()) -> ok | {error, term()}.

Advertise a LOCAL in-process streaming procedure with mode.

-spec advertise_stream(pool(), realm(), procedure(), stream_mode(), stream_handler()) ->
                          ok | {error, term()}.

Advertise a streaming procedure on a V2 pool. Fans out to every healthy link and stores in pool state for replay on link respawn. See macula_client:advertise_stream/5.

-spec await_reply(stream()) -> {ok, term()} | {error, term()}.

Wait for the terminal reply (client-stream / bidi).

-spec call(pool(), realm(), procedure(), term(), pos_integer()) -> {ok, term()} | {error, term()}.

Issue a CALL frame against a V2 pool. First-success across the pool's healthy links. See macula_client:call/5.

-spec call_station(pool(), macula_client:seed(), realm(), procedure(), term(), pos_integer()) ->
                      {ok, term()} | {error, term()}.

Issue a CALL to ONE specific station, dialing it directly even if it is not in the pool's seed set. Station is a seed URL (e.g. <<"quic://[::1]:4433">>). The pool reuses an existing link or dials and monitors a new one, waits for the handshake, and calls there. This is the direct-dial data path: resolve a serving_station and its endpoint, then reach it in one hop. See macula_client:call_station/6.

-spec call_station(pool(), macula_client:seed(), realm(), procedure(), term(), pos_integer(), map()) ->
                      {ok, term()} | {error, term()}.

As call_station/6, presenting a capability token to a gated provider via Opts (#{ucan_token => Token}). Empty/absent = none. Slice 7b dual-trust.

-spec call_stream(procedure(), term()) -> {ok, stream()} | {error, term()}.

Open a LOCAL in-process server-stream call. Used for unit tests and same-BEAM dispatch via macula_stream_local.

-spec call_stream(procedure(), term(), map()) -> {ok, stream()} | {error, term()}.

Open a LOCAL in-process server-stream call with options.

-spec call_stream(pool(), realm(), procedure(), term(), map()) -> {ok, stream()} | {error, term()}.

Open a streaming RPC against a V2 pool. Picks the first currently-healthy link and opens the stream there; the returned stream is sticky-to-link (errors with peer_down if the link dies; caller re-opens). See macula_client:call_stream/5.

-spec call_stream_station(pool(), macula_client:seed(), realm(), procedure(), term(), map()) ->
                             {ok, stream()} | {error, term()}.

Open a streaming RPC by DIALING a specific station directly (direct-dial), instead of routing through an existing pool link — the streaming analogue of call_station/6. Compose it with DHT resolution (find_records -> read_procedure_advertisement -> station_endpoint) to reach a stream provider in one hop, exactly as a unary caller does. Opts may set dial_timeout_ms (default 10_000) and a mode.

-spec child_spec(term(), [macula_client:seed()], macula_client:opts()) -> supervisor:child_spec().

OTP child spec to drop a V2 pool into a caller's supervision tree.

-spec close(pool()) -> ok.

Stop a V2 pool. Every subscriber receives a final {macula_event_gone, SubRef, pool_closed} message.

-spec close_send(stream()) -> ok.

Half-close the write side; recv still drains.

-spec close_stream(stream()) -> ok.

Close a V1 stream (both sides). Renamed from close/1 in 3.11.0 because close/1 now refers to the V2 pool surface.

-spec connect([macula_client:seed()], macula_client:opts()) -> {ok, pool()} | {error, term()}.

Connect to the Macula relay mesh and return a pool handle.

Seeds is a list of relay endpoints (URL binaries/strings or #{host, port} maps). The pool spawns one peering link per seed and routes ops with replication, replay, and event dedup. Returns immediately; link handshakes complete asynchronously.

Honored opts (full reference: macula_client:opts()):

• identity — pool's Ed25519 keypair; auto-generated if absent.
• replication_factor — links per PUBLISH (default 1).
• capabilities — per-link bitfield (default 0).
• alpn — QUIC ALPN list (default [<<"macula">>]).
• connect_timeout_ms — per-link CONNECT/HELLO deadline (default 30_000).
• dedup_window_ms, dedup_sweep_ms — inbound-EVENT dedup tunables.

Legacy opts silently dropped (with a one-shot logger:notice): relays (use the Seeds positional argument), realm (V2 is realm-per-call), site (no V2 analog), connections (one link per seed; add more seeds to grow the pool).

See macula_client for the canonical pool implementation and macula_pubsub for the slice module.

-spec ensure_distributed() -> ok | {error, term()}.

Ensure this node is running in distributed mode.

-spec find_content_providers(pool(), mcid()) -> {ok, [map()]} | {error, term()}.

Resolve every host currently announcing an MCID: hosts that stored a chunked put (_content.put_manifest) and got content_announcementd automatically by the station on receipt (macula_content_announcer). get_content/2 already reaches a copy via the connected station's own 1-hop peer relay, so this is for a caller that wants to know WHO holds an MCID, or to dial a specific one directly with call_station/6 — e.g. when the connected station's relay hop budget does not reach the host (a partial-mesh pair with no mutual peer), or to route around a specific host deliberately.

Each entry's signature is verified before its endpoint is trusted; unverifiable or malformed records are dropped, not surfaced as errors. Single-block content (put via _content.put_block alone) is not announced — resolving its MCID returns {ok, []}.

-spec find_record(pool(), record_key()) -> {ok, m_record()} | {error, not_found | term()}.

Fetch a record from the mesh DHT by its macula_record:storage_key/1.

Returns {error, not_found} when no record exists at the key. The returned record's signature should be verified via macula_record:verify/1 before its payload is trusted.

-spec find_records(pool(), record_key()) -> {ok, [m_record()]} | {error, term()}.

Fetch EVERY record stored at Key — the full multi-value set, e.g. every procedure_advertisement under one procedure's storage key. Where find_record/2 returns the first record (or not_found), this returns the whole list, empty when none.

The relay's local store is a signer-deduped multiset: one record per signing key at a storage key, so N providers of one procedure return N records. Each returned record's signature should be verified via macula_record:verify/1 before its payload is trusted.

-spec find_records_by_type(pool(), record_type()) -> {ok, [m_record()]} | {error, term()}.

Return every record of a given type currently visible from the pool's connected stations.

Coverage depends on each station's view of the DHT — a single station sees its local replicas plus whatever its peers have gossiped. Aggregating across the full mesh requires querying multiple stations and deduplicating by record key.

-spec get_content(pool(), mcid()) -> {ok, binary()} | {error, not_found | term()}.

Fetch the bytes for a previously-stored MCID. Returns {error, not_found} if no provider in the pool's reach holds a copy (for chunked content, if any single chunk is unreachable). Dispatches on the MCID's codec byte: 16#55 (raw/single-block) fetches one block, BLAKE3-verified by the station before it leaves the store; 16#56 (manifest) fetches the manifest, then every chunk in order, reassembles, and verifies the whole against the manifest's size and Merkle root before returning.

-spec get_cookie() -> atom().

Get the Erlang cluster cookie.

-spec join_dist_relay(map()) -> ok | {error, term()}.

Enable Erlang distribution over a dedicated dist relay (macula-io/macula-dist-relay).

Different from join_mesh/1: - Connects to a dist relay (port 4434, ALPN macula-dist), NOT the pub/sub station mesh - No mesh_client, no pub/sub subscriptions — only dist traffic - Uses raw QUIC stream routing with no MessagePack overhead

Options: - url (required): <<"quic://relay.example.com:4434">>

After this returns ok, standard OTP distribution (rpc:call/4, gen_server:call/3 across nodes, pg groups, etc.) works across firewalls via the dist relay.

-spec join_mesh(map()) -> ok | {error, term()}.

Join the Macula relay mesh with Erlang distribution.

After calling this, standard OTP distribution works across firewalls. Opts takes:

• relays (required) — list of seed URLs for the V2 pool.
• identity — V2 pool's macula_identity:key_pair(). Default: auto-generated.

Internally builds a V2 macula_client:pool() and registers it with macula_dist_pool as the carrier for _dist.tunnel.* traffic. Dist tunnel frames travel under the all-zeros realm (protocol-internal infrastructure, not bound to any user realm).

-spec links(pool()) -> {ok, [macula_client:link_info()]}.

Per-link snapshot of a V2 pool — one entry per spawned link with its peer station node_id (pubkey), dial host, pid, and connected flag. Use this to resolve a specific station (by pubkey or hostname) to its link for targeted, per-station operations. See macula_client:links/1 for the link_info() shape.

-spec monitor_nodes() -> ok.

Subscribe to node up/down events.

-spec open_stream(procedure(), term(), map()) -> {ok, stream()} | {error, term()}.

Open a LOCAL in-process client-stream or bidi call. Used for unit tests and same-BEAM dispatch via macula_stream_local.

-spec open_stream(procedure(), term(), map(), stream_mode()) -> {ok, stream()} | {error, term()}.

Open a LOCAL in-process stream with explicit mode.

-spec publish(pool(), realm(), topic(), term()) -> ok | {error, term()}.

Publish to (Realm, Topic) on Pool. Equivalent to publish/5 with empty opts.

-spec publish(pool(), realm(), topic(), term(), map()) -> ok | {error, term()}.

Publish to (Realm, Topic) on Pool with options. See macula_pubsub:publish/5 for honored opts.

-spec put_content(pool(), binary()) -> {ok, mcid()} | {error, term()}.

Store Bytes in the mesh's content store and return its MCID (Macula Content ID — 34 bytes: version, codec, then a 32-byte hash). Content that fits in one block (byte_size(Bytes) =< macula_manifest:default_chunk_size/0, 256 KiB) is sent as a single _content.put_block — the MCID is <<1, 16#55, BLAKE3(Bytes)>>, unchanged since v4.2.7. Larger content is split into chunks (macula_manifest:create/1), each chunk sent via its own _content.put_block, then a content_manifest via _content.put_manifest; the returned MCID is the manifest's (<<1, 16#56, _/binary>>), Merkle-rooted over every chunk. Either way the station verifies each block's hash before accepting it.

-spec put_record(pool(), m_record()) -> ok | {error, term()}.

Store a signed record in the mesh DHT via a V2 pool.

Build the record via the typed constructors in macula_record (node_record/3,4, content_announcement/3,4, tombstone/3,4, realm_directory/3,4, procedure_advertisement/3,4, etc.) then sign it with macula_record:sign/2. The relay validates the signature on receipt; an invalid signature returns {error, bad_signature}. Successful stores propagate to the K-nearest peers in the DHT under the record's macula_record:storage_key/1.

-spec recv(stream()) -> {chunk, binary()} | {data, term()} | eof | {error, term()}.

Receive the next chunk (blocks).

-spec send(stream(), binary()) -> ok | {error, term()}.

Send a binary chunk on the stream.

-spec send(stream(), binary() | term(), raw | msgpack) -> ok | {error, term()}.

Send a chunk with explicit encoding.

-spec set_cookie(atom() | binary()) -> ok.

Set the Erlang cluster cookie.

-spec set_reply(stream(), term()) -> ok.

Server-side: emit the terminal reply value.

-spec status(pool()) -> {ok, macula_client:status()}.

Aggregate health snapshot of a V2 pool. Suitable for /health or /status endpoints; not for hot-loop polling. See macula_client:status/1 for the full shape.

-spec subscribe(pool(), realm(), topic(), pid()) -> {ok, reference()}.

Subscribe Subscriber to (Realm, Topic) on Pool. Equivalent to subscribe/5 with empty opts.

-spec subscribe(pool(), realm(), topic(), pid(), map()) -> {ok, reference()}.

Subscribe Subscriber to (Realm, Topic) on Pool with options. The delivery option chooses how a single publisher's out-of-order arrivals are handled:

• ordered (default) — per-publisher FIFO by seq; out-of-order arrivals are buffered and released in order, a genuinely missing seq skipped after order_timeout_ms (a connect/2 option, default 250ms).
• latest_only — deliver only seqs newer than the highest seen for that publisher (drop stale); no buffering, no delay.
• as_arrives — deliver in raw arrival order; the consumer orders it itself.

See macula_pubsub:subscribe/5.

-spec subscribe_callback(pool(), realm(), topic(), macula_pubsub:callback()) ->
                            {ok, reference()} | {error, term()}.

Subscribe with a callback function. The SDK spawns a small receiver process internally and invokes the callback once per inbound event. See macula_pubsub:subscribe_callback/4.

-spec subscribe_records(pool(), record_type(), fun((m_record()) -> any())) ->
                           {ok, reference()} | {error, term()}.

Subscribe to live record-stored events filtered by type.

The callback receives each newly-stored record of the given type. Returns a subscription reference for unsubscribe_records/2. Topic shape is _dht.records.<type>.stored, rendered with the type tag as a decimal integer for log friendliness.

-spec unadvertise(pool(), realm(), procedure()) -> ok.

Stop advertising a procedure on a V2 pool.

-spec unadvertise_stream(procedure()) -> ok.

Stop advertising a LOCAL streaming procedure.

-spec unadvertise_stream(pool(), realm(), procedure()) -> ok.

Stop advertising a streaming procedure on a V2 pool.

-spec unmonitor_nodes() -> ok.

Unsubscribe from node up/down events.

-spec unsubscribe(pool(), reference()) -> ok.

Drop a pool subscription. Idempotent.

-spec unsubscribe_records(pool(), reference()) -> ok.

Cancel a subscribe_records/3 subscription.