macula_peering (macula v11.4.0)
View SourceMacula peering — connection state machine API.
Each peer connection is one macula_peering_conn gen_statem under the macula_peering_conn_sup simple_one_for_one supervisor.
Two entry points:
connect/1— outbound dial; worker drives the QUIC connect.accept/2— inbound; caller transfers ownership of an already-establishedmacula_quicconnection (areference()) to a new worker.
The caller passes a controlling_pid in opts; that pid receives peering events as messages:
{macula_peering, connected, ConnPid, PeerNodeId}{macula_peering, frame, ConnPid, Frame}(post-handshake){macula_peering, disconnected, ConnPid, Reason}
PeerNodeId is the peer's node_id, derived from the identity key the handshake verified (plans/DESIGN_PQ_HANDSHAKE_FRAMES.md).
An optional accept_owner pid in opts receives a single {macula_peering, handshake_complete, ConnPid, PeerNodeId} message the moment the worker transitions from handshaking to connected. Used by accept-side listeners that (a) cap concurrent handshaking workers separately from healthy connected peers, and (b) dedupe duplicate dials from the same peer identity by closing prior workers for the same PeerNodeId.
Summary
Functions
Inbound accept. Caller currently owns Conn (e.g. it's the listener owner that just received {quic, new_conn, Conn, _}). The transfer of ownership and the handshake start are sequenced atomically.
Start opening a dedicated stream, as open_dedicated_stream/1 does, and return at once with a reference. The calling process later receives {macula_peering, dedicated_stream_opened, Ref, Stream}, and owns Stream in active mode, or {macula_peering, dedicated_stream_open_failed, Ref, Reason}, where Reason is timeout when the peer allows no further stream within 10 s, not_connected when the connection is not connected, closed when it ends first, or the open's error. A connection process that has already exited sends nothing, so monitor it to learn that.
relay_on_stream/2 without waiting, with the busy and send_ready behaviour of async_send_on_stream/2.
async_relay_on_stream/2 for a frame whose end the calling process hears about, with the notices of async_send_on_stream/3.
send_on_stream/2 without waiting: the bytes are queued with macula_quic:async_send/2, so a peer that stops reading cannot hold the calling process. Returns ok once they are queued. When the stream already holds 1 MiB unwritten, queues nothing and returns {error, busy}, and the calling process later gets {quic, send_ready, Stream, undefined} when it may send again. A relay passes a frame on as the bytes it received.
async_send_on_stream/2 for bytes whose end the calling process hears about. For bytes it queued, that process gets exactly one of {quic, send_complete, Stream, Tag}, once they are written, and {quic, send_incomplete, Stream, {Tag, Reason}}, when the stream is reset, closed or fails first. Tag is the caller's own term, copied into that message, so keep it small.
Initiate a graceful close (sends GOODBYE, drains 5s, terminates). For a peer that was never trusted in the first place (failed an admission check rather than ending a legitimate session), use reject/2 instead — see its own doc for why.
Close a dedicated stream, one obtained from open_dedicated_stream/1 or one the peer opened, gracefully: data already written still goes out, and then the stream ends, as macula_quic:close_stream/1 does.
Outbound connect. Spawns a worker that opens a QUIC connection to target and runs the post-quantum handshake.
Report an object a connection carried that its receiver refused, by the kind of refusal. The connection counts refusals by kind, and counts the ones macula_frame:charged_refusal/1 charges. A kind outside that classification is refused here, where it is reported.
Open a QUIC stream on this connection dedicated to one session (a streaming RPC call, a content transfer) instead of sharing the connection's control stream, waiting for it in the calling process. Ownership transfers to the calling process: it drives the stream directly via send_on_stream/2 and macula_quic:*, and receives the stream's {quic, Bin, Stream, Flags} events straight into its own mailbox — the peering connection process is not in this stream's path at all once this call returns. See PLAN_PER_STREAM_QUIC_ISOLATION.md.
Read the peer's capabilities bitmask as observed in their CONNECT/HELLO frame. Returns {ok, NegotiatedCaps} once the handshake has completed and {error, not_connected} otherwise.
What the handshake verified of the peer: its node_id, its identity key as carried, the profile and its capabilities. Returns {error, not_connected} until the handshake has completed.
The refusals reported on a connection: a count per kind, and how many of them were charged.
Terminate a connection immediately, with no GOODBYE and no drain window — for a peer that failed an admission check (e.g. an S/Kademlia identity puzzle) rather than one ending a legitimate session. close/2 transitions through draining for up to 5s (?DRAIN_TIMEOUT_MS), during which any further inbound data is silently accepted and discarded by design (draining state's "ignore late inbound during drain" clause) — correct for a genuinely-trusted peer whose last few in-flight frames shouldn't cause spurious errors, but for a peer that was never admitted at all, those same 5 seconds are pure exposure: the connection went from "no verdict yet" to "should already be gone" the instant the admission check failed, so there is no legitimate traffic left to drain gracefully. reject/2 skips draining and terminates the state machine directly.
Write a frame a relay received onto a dedicated stream, as the bytes it received: a unit macula_frame:parse_for_relay/2 accepted, and nothing else, so a relay never writes bytes its reader did not accept. Synchronous, as send_on_stream/2.
Send a frame through the peer connection. The frame goes out as its producer built it, and in pq_hybrid the connection adds a neighbour signature to a control frame.
Write one frame's bytes directly onto a dedicated stream obtained from open_dedicated_stream/1, with no peering connection process involved, unlike send_frame/2. The stream's writer builds, signs and encodes the frame itself; this writes the bytes it is given and nothing else. Synchronous: waits in the calling process until the bytes are written, for as long as the peer withholds QUIC flow-control credit, as macula_quic:send/2 does. A process that must not wait on a peer uses async_send_on_stream/2 instead.
Types
-type opts() :: macula_peering_conn:opts().
Functions
Inbound accept. Caller currently owns Conn (e.g. it's the listener owner that just received {quic, new_conn, Conn, _}). The transfer of ownership and the handshake start are sequenced atomically.
Start opening a dedicated stream, as open_dedicated_stream/1 does, and return at once with a reference. The calling process later receives {macula_peering, dedicated_stream_opened, Ref, Stream}, and owns Stream in active mode, or {macula_peering, dedicated_stream_open_failed, Ref, Reason}, where Reason is timeout when the peer allows no further stream within 10 s, not_connected when the connection is not connected, closed when it ends first, or the open's error. A connection process that has already exited sends nothing, so monitor it to learn that.
-spec async_relay_on_stream(reference(), macula_frame:received_frame()) -> ok | {error, term()}.
relay_on_stream/2 without waiting, with the busy and send_ready behaviour of async_send_on_stream/2.
-spec async_relay_on_stream(reference(), macula_frame:received_frame(), term()) -> ok | {error, term()}.
async_relay_on_stream/2 for a frame whose end the calling process hears about, with the notices of async_send_on_stream/3.
send_on_stream/2 without waiting: the bytes are queued with macula_quic:async_send/2, so a peer that stops reading cannot hold the calling process. Returns ok once they are queued. When the stream already holds 1 MiB unwritten, queues nothing and returns {error, busy}, and the calling process later gets {quic, send_ready, Stream, undefined} when it may send again. A relay passes a frame on as the bytes it received.
async_send_on_stream/2 for bytes whose end the calling process hears about. For bytes it queued, that process gets exactly one of {quic, send_complete, Stream, Tag}, once they are written, and {quic, send_incomplete, Stream, {Tag, Reason}}, when the stream is reset, closed or fails first. Tag is the caller's own term, copied into that message, so keep it small.
-spec close(pid()) -> ok.
Initiate a graceful close (sends GOODBYE, drains 5s, terminates). For a peer that was never trusted in the first place (failed an admission check rather than ending a legitimate session), use reject/2 instead — see its own doc for why.
-spec close_dedicated_stream(reference()) -> ok.
Close a dedicated stream, one obtained from open_dedicated_stream/1 or one the peer opened, gracefully: data already written still goes out, and then the stream ends, as macula_quic:close_stream/1 does.
Outbound connect. Spawns a worker that opens a QUIC connection to target and runs the post-quantum handshake.
Report an object a connection carried that its receiver refused, by the kind of refusal. The connection counts refusals by kind, and counts the ones macula_frame:charged_refusal/1 charges. A kind outside that classification is refused here, where it is reported.
Open a QUIC stream on this connection dedicated to one session (a streaming RPC call, a content transfer) instead of sharing the connection's control stream, waiting for it in the calling process. Ownership transfers to the calling process: it drives the stream directly via send_on_stream/2 and macula_quic:*, and receives the stream's {quic, Bin, Stream, Flags} events straight into its own mailbox — the peering connection process is not in this stream's path at all once this call returns. See PLAN_PER_STREAM_QUIC_ISOLATION.md.
Returns {error, timeout} when the peer allows no further stream within 10 s, and {error, closed} when the connection ends first. A process that must keep serving while it waits uses async_open_dedicated_stream/1 instead.
-spec peer_capabilities(pid()) -> {ok, non_neg_integer()} | {error, not_connected}.
Read the peer's capabilities bitmask as observed in their CONNECT/HELLO frame. Returns {ok, NegotiatedCaps} once the handshake has completed and {error, not_connected} otherwise.
Used by relays to tell direct daemon ADVERTISEs from station-to- station gossip relays at frame-dispatch time (see ?CAP_STATION). Daemons send 0; relay stations OR-in ?CAP_STATION. Pre-version peers that don't set the bit are treated as daemons by callers, which matches their actual role.
-spec peer_identity(pid()) -> {ok, #{node_id := <<_:256>>, identity_key := binary(), profile := macula_crypto_profile:profile(), capabilities := non_neg_integer()}} | {error, not_connected}.
What the handshake verified of the peer: its node_id, its identity key as carried, the profile and its capabilities. Returns {error, not_connected} until the handshake has completed.
-spec refusals(pid()) -> #{counts := #{atom() => pos_integer()}, charged := non_neg_integer()}.
The refusals reported on a connection: a count per kind, and how many of them were charged.
Terminate a connection immediately, with no GOODBYE and no drain window — for a peer that failed an admission check (e.g. an S/Kademlia identity puzzle) rather than one ending a legitimate session. close/2 transitions through draining for up to 5s (?DRAIN_TIMEOUT_MS), during which any further inbound data is silently accepted and discarded by design (draining state's "ignore late inbound during drain" clause) — correct for a genuinely-trusted peer whose last few in-flight frames shouldn't cause spurious errors, but for a peer that was never admitted at all, those same 5 seconds are pure exposure: the connection went from "no verdict yet" to "should already be gone" the instant the admission check failed, so there is no legitimate traffic left to drain gracefully. reject/2 skips draining and terminates the state machine directly.
-spec relay_on_stream(reference(), macula_frame:received_frame()) -> ok | {error, term()}.
Write a frame a relay received onto a dedicated stream, as the bytes it received: a unit macula_frame:parse_for_relay/2 accepted, and nothing else, so a relay never writes bytes its reader did not accept. Synchronous, as send_on_stream/2.
-spec send_frame(pid(), macula_frame:frame()) -> ok | {error, term()}.
Send a frame through the peer connection. The frame goes out as its producer built it, and in pq_hybrid the connection adds a neighbour signature to a control frame.
The send is a cast, so encoding happens later, inside the shared connection process. This is therefore the LAST synchronous point at which a caller can be told its frame is unsendable, and every producer — pubsub, RPC calls and results, streaming, advertise, content — funnels through here. Guarding one verb upstream (publish) left the other five able to kill the connection, so the check lives here, where it covers all of them at one seam.
Returns {error, {unsupported_payload_type, Type, Path}} without casting when the frame cannot be encoded. Callers that ignore the return at least no longer take the connection down; callers that check get a structured reason and a path to the offending value.
Write one frame's bytes directly onto a dedicated stream obtained from open_dedicated_stream/1, with no peering connection process involved, unlike send_frame/2. The stream's writer builds, signs and encodes the frame itself; this writes the bytes it is given and nothing else. Synchronous: waits in the calling process until the bytes are written, for as long as the peer withholds QUIC flow-control credit, as macula_quic:send/2 does. A process that must not wait on a peer uses async_send_on_stream/2 instead.