%%%------------------------------------------------------------------- %%% @doc Macula streaming RPC — single-stream state machine. %%% %%% Owns one streaming RPC's state. Each `call_stream', `open_stream', %%% or server-side handler invocation gets its own `macula_stream' %%% gen_server. The state machine itself is carrier-agnostic; the %%% peer shape (`{local, _}' or `{remote_via_link, _, _}') decides %%% how chunks reach the wire. %%% %%% Two carriers route through `forward_to_peer/2': %%% %%% %%% Renamed from `macula_stream_v1' in 3.17.0; the V1 mesh_client %%% carrier (`{remote, _, _}') was retired alongside the rest of the %%% V1 surface in the same release. The module now spans the LOCAL %%% carrier and the V2 station_link carrier only. %%% @end %%%------------------------------------------------------------------- -module(macula_stream). -behaviour(gen_server). %% Public API -export([ start_link/1, pair/2, attach_to_link/3, send/2, send/3, recv/1, recv/2, close/1, close_send/1, await_reply/1, await_reply/2, set_reply/2, set_error/2, abort/3, controlling_process/2, info/1 ]). %% The check of the stream functions a supervised wrapper is given. -export([stream_io/2]). %% Peer-to-peer protocol — drives inbound deliveries from carrier %% modules (`macula_stream_local' for LOCAL pairs, `macula_station_link' %% for V2 station-link pairs). -export([ deliver_chunk/3, deliver_end/2, deliver_error/3, deliver_reply/2, deliver_frame/2 ]). %% gen_server callbacks -export([ init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, format_status/1 ]). -type role() :: client | server. -type mode() :: server_stream | client_stream | bidi. -type encoding() :: raw | msgpack. -type chunk() :: binary() | {raw, binary()} | {term, term()}. -type stream_id() :: binary(). -type result() :: {ok, term()} | {error, term()}. %% Peer shape: %% undefined — unpaired %% {local, Pid} — in-process pairing %% {remote_via_link, L, Sid}: the station_link carrier. The stream %% signs its frames and hands the link %% their bytes. -type peer() :: undefined | {local, pid()} | {remote_via_link, pid(), stream_id()}. -export_type([role/0, mode/0, encoding/0, chunk/0, stream_id/0, result/0, peer/0]). %% The functions a supervised stream wrapper, such as `macula_stream_sink', %% opens, reads, writes and ends its stream with, by key. Each wrapper %% calls some of them; `stream_io/2' checks the ones it is given. -type stream_io() :: #{call_stream => fun((macula:pool(), macula:realm(), macula:procedure(), term(), map()) -> {ok, pid()} | {error, term()}), recv => fun((pid(), timeout()) -> {chunk, binary()} | {data, term()} | eof | {error, term()}), send => fun((pid(), binary() | term(), encoding()) -> ok | {error, term()}), close_send => fun((pid()) -> term()), close => fun((pid()) -> term()), close_stream => fun((pid()) -> term()), abort => fun((pid(), binary(), binary()) -> term()), set_reply => fun((pid(), term()) -> term()), set_error => fun((pid(), term()) -> term()), await_reply => fun((pid()) -> result()), controlling_process => fun((pid(), pid()) -> ok | {error, not_owner})}. -export_type([stream_io/0]). %% How a session ended, as the owner is told it. -type ended() :: closed | peer_down | {error, {binary(), binary()}} | {error, {transport, term()}}. %% The bytes of chunks no reader has taken a stream keeps by default, the %% same as a QUIC stream's default receive window. -define(MAX_INBOX_BYTES, 16#1000000). %% A STREAM_ERROR message is text for people of at most 256 bytes, as a %% GOODBYE reason is. -define(MAX_ERROR_TEXT_BYTES, 256). %% The code an abort sends in place of one a STREAM_ERROR cannot carry. -define(ABORTED_CODE, <<"aborted">>). -record(state, { id :: stream_id(), role :: role(), mode :: mode(), owner :: pid(), owner_ref :: reference(), peer :: peer(), %% Recv side: inbound chunks queued, waiting recv/2 callers, eof flag inbox = queue:new() :: queue:queue({encoding(), term()}), %% The bytes of the queued chunks, and the most it may reach inbox_bytes = 0 :: non_neg_integer(), max_inbox_bytes = ?MAX_INBOX_BYTES :: pos_integer(), waiters = queue:new() :: queue:queue({{pid(), reference()}, reference()}), closed_recv = false :: boolean(), %% Send side closed_send = false :: boolean(), seq_out = 0 :: non_neg_integer(), seq_in = 0 :: non_neg_integer(), %% Terminal reply (for client-stream / bidi) reply = undefined :: undefined | result(), reply_waiters = [] :: [{pid(), reference()}], %% How the session ended, once it has and the owner has been told ended = undefined :: undefined | ended(), %% A link-carried stream: the loader of the node identity key it signs %% with, called each time it signs, so the stream never holds the key; %% its verified STREAM_OPEN, the peering connection it reports refused %% frames to, the crypto profile, and what it has verified of the %% peer's frames so far. key :: fun(() -> macula_node_keys:node_key()) | undefined, open :: macula_frame:verified_request() | undefined, conn :: pid() | undefined, profile :: macula_crypto_profile:profile() | undefined, verifier :: macula_frame:stream_state() | undefined }). %%%=================================================================== %%% Public API %%%=================================================================== %% @doc Start a stream gen_server. %% %% Required opts: id, role, mode, owner. Optional: max_inbox_bytes, the %% bytes of chunks no reader has taken that the stream keeps; a chunk past %% them ends the session (default 16 MiB). A stream carried by a %% `macula_station_link' also takes `key' (a function that returns the %% node identity key it signs with, called each time it signs, so the %% stream never holds the key), `open' (the verified STREAM_OPEN), %% `conn' (the peering connection that carries it) and `profile'. A %% stream given the key itself as `key' does not start, and start_link %% returns `{error, {key, not_a_loader}}'. -spec start_link(map()) -> {ok, pid()} | {error, term()}. start_link(Opts) -> gen_server:start_link(?MODULE, Opts, []). %% @doc Pair two stream processes as peers (Phase 1 local dispatch). -spec pair(pid(), pid()) -> ok. pair(A, B) when is_pid(A), is_pid(B) -> ok = gen_server:call(A, {pair_local, B}), ok = gen_server:call(B, {pair_local, A}), ok. %% @doc Attach a `macula_station_link' peer to this stream. The stream %% hands the link the bytes of each frame it signs, and the link %% forwards the peer's STREAM_* frames into it through %% `deliver_frame/2'. -spec attach_to_link(pid(), pid(), stream_id()) -> ok. attach_to_link(StreamPid, LinkPid, StreamId) when is_pid(StreamPid), is_pid(LinkPid), is_binary(StreamId) -> gen_server:call(StreamPid, {pair_via_link, LinkPid, StreamId}). %% @doc Send a binary chunk on the stream. -spec send(pid(), binary()) -> ok | {error, term()}. send(Pid, Bin) when is_binary(Bin) -> send(Pid, Bin, raw). -spec send(pid(), binary() | term(), encoding()) -> ok | {error, term()}. send(Pid, Body, raw) when is_binary(Body) -> gen_server:call(Pid, {send, raw, Body}); send(Pid, Body, msgpack) -> gen_server:call(Pid, {send, msgpack, Body}). %% @doc Receive the next chunk (blocks indefinitely). -spec recv(pid()) -> {chunk, binary()} | {data, term()} | eof | {error, term()}. recv(Pid) -> recv(Pid, infinity). -spec recv(pid(), timeout()) -> {chunk, binary()} | {data, term()} | eof | {error, term()}. recv(Pid, Timeout) -> %% Long timeouts allowed because the wait is on inbound network %% data, not on the gen_server's processing time. GsTimeout = case Timeout of infinity -> infinity; N when is_integer(N) -> N + 100 end, gen_server:call(Pid, {recv, Timeout}, GsTimeout). %% @doc Half-close the write side. Recv side stays open. -spec close_send(pid()) -> ok. close_send(Pid) -> gen_server:call(Pid, close_send). %% @doc Close both sides. Idempotent. -spec close(pid()) -> ok. close(Pid) -> gen_server:call(Pid, close). %% @doc Wait for the terminal reply (client-stream / bidi). -spec await_reply(pid()) -> result(). await_reply(Pid) -> await_reply(Pid, infinity). -spec await_reply(pid(), timeout()) -> result() | {error, timeout}. await_reply(Pid, Timeout) -> GsTimeout = case Timeout of infinity -> infinity; N when is_integer(N) -> N + 100 end, gen_server:call(Pid, {await_reply, Timeout}, GsTimeout). %% @doc Server-side: emit the terminal reply. -spec set_reply(pid(), term()) -> ok. set_reply(Pid, Result) -> gen_server:call(Pid, {set_reply, {ok, Result}}). %% @doc Server-side: emit a terminal error as the reply value. -spec set_error(pid(), term()) -> ok. set_error(Pid, Reason) -> gen_server:call(Pid, {set_reply, {error, Reason}}). %% @doc Abort the stream with a STREAM_ERROR frame. Both sides close; %% any pending recv/await_reply waiters receive {error, {Code, Message}}. %% A code a STREAM_ERROR cannot carry, over 64 bytes or not UTF-8, still %% aborts the stream, with the code `aborted', and the caller gets the %% code's refusal by name, so a call to abort always stops the stream. -spec abort(pid(), binary(), binary()) -> ok | {error, {text_too_long | invalid_text, code}}. abort(Pid, Code, Message) when is_binary(Code), is_binary(Message) -> aborted(macula_frame:text_checked(code, Code), Pid, Code, Message). aborted(ok, Pid, Code, Message) -> gen_server:call(Pid, {abort, Code, Message}); aborted({error, _} = Refused, Pid, _Code, Message) -> ok = gen_server:call(Pid, {abort, ?ABORTED_CODE, Message}), Refused. %% @doc Hand the stream to `NewOwner'. A stream ends when its owner ends; %% after this it ends when `NewOwner' does, and `NewOwner' is told when the %% session ends, as `{macula_stream, ended, Stream, How}', or at once if it %% already has. Only the stream's current owner can hand it over; any other %% caller gets `{error, not_owner}' and the stream stays with its owner. -spec controlling_process(pid(), pid()) -> ok | {error, not_owner}. controlling_process(Pid, NewOwner) when is_pid(Pid), is_pid(NewOwner) -> gen_server:call(Pid, {controlling_process, NewOwner}). %% @doc Inspect stream state (debugging). -spec info(pid()) -> map(). info(Pid) -> gen_server:call(Pid, info). %% @doc The stream functions a wrapper runs on. `Defaults' are the %% functions the wrapper calls, by key, and `Given' the `stream_io' its %% caller gave, or `undefined' for none, which gives `Defaults'. A given %% set has every key in `Defaults', each function at the arity its key %% takes, and may carry other `stream_io()' functions; any other is %% refused with `function_clause', in the calling process. -spec stream_io(stream_io(), stream_io() | undefined) -> stream_io(). stream_io(Defaults, undefined) when is_map(Defaults) -> Defaults; stream_io(Defaults, Given) when is_map(Defaults), is_map(Given) -> ok = maps:foreach(fun stream_function/2, Given), ok = lists:foreach(fun(Key) -> given_key(Key, Given) end, maps:keys(Defaults)), Given. stream_function(call_stream, Fun) when is_function(Fun, 5) -> ok; stream_function(recv, Fun) when is_function(Fun, 2) -> ok; stream_function(send, Fun) when is_function(Fun, 3) -> ok; stream_function(close_send, Fun) when is_function(Fun, 1) -> ok; stream_function(close, Fun) when is_function(Fun, 1) -> ok; stream_function(close_stream, Fun) when is_function(Fun, 1) -> ok; stream_function(abort, Fun) when is_function(Fun, 3) -> ok; stream_function(set_reply, Fun) when is_function(Fun, 2) -> ok; stream_function(set_error, Fun) when is_function(Fun, 2) -> ok; stream_function(await_reply, Fun) when is_function(Fun, 1) -> ok; stream_function(controlling_process, Fun) when is_function(Fun, 2) -> ok. given_key(Key, Given) when is_map_key(Key, Given) -> ok. %%%=================================================================== %%% Peer-to-peer protocol %%%=================================================================== %% @doc Deliver a chunk frame from the peer. -spec deliver_chunk(pid(), encoding(), term()) -> ok. deliver_chunk(Pid, Encoding, Body) -> gen_server:cast(Pid, {peer_chunk, Encoding, Body}). %% @doc Deliver a STREAM_END frame from the peer. -spec deliver_end(pid(), send | both) -> ok. deliver_end(Pid, Role) -> gen_server:cast(Pid, {peer_end, Role}). %% @doc Deliver a STREAM_ERROR frame from the peer. -spec deliver_error(pid(), binary(), binary()) -> ok. deliver_error(Pid, Code, Message) -> gen_server:cast(Pid, {peer_error, Code, Message}). %% @doc Deliver a STREAM_REPLY frame from the peer. -spec deliver_reply(pid(), result()) -> ok. deliver_reply(Pid, Result) -> gen_server:cast(Pid, {peer_reply, Result}). %% @doc Deliver a STREAM_DATA, STREAM_END, STREAM_ERROR or STREAM_REPLY %% frame of a link-carried stream from the peer. The stream verifies it %% against its STREAM_OPEN before it takes effect. A refused frame is %% dropped and reported to the stream's peering connection, and the %% stream carries on. A frame of a type that belongs on the control %% stream rejects the connection with malformed_frame and ends the %% stream, in either profile. -spec deliver_frame(pid(), macula_frame:frame()) -> ok. deliver_frame(Pid, Frame) when is_map(Frame) -> gen_server:cast(Pid, {peer_frame, Frame}). %%%=================================================================== %%% gen_server callbacks %%%=================================================================== init(#{key := Key}) when not is_function(Key, 0) -> {error, {key, not_a_loader}}; init(Opts) -> Id = maps:get(id, Opts), Role = maps:get(role, Opts), Mode = maps:get(mode, Opts), Owner = maps:get(owner, Opts), OwnerRef = erlang:monitor(process, Owner), {ok, carried(Opts, #state{ id = Id, role = Role, mode = Mode, owner = Owner, owner_ref = OwnerRef, max_inbox_bytes = maps:get(max_inbox_bytes, Opts, ?MAX_INBOX_BYTES) })}. %% --- pair -------------------------------------------------------------- handle_call({pair_local, Peer}, _From, State) -> _ = erlang:monitor(process, Peer), {reply, ok, State#state{peer = {local, Peer}}}; handle_call({pair_via_link, LinkPid, StreamId}, _From, State) -> _ = erlang:monitor(process, LinkPid), {reply, ok, State#state{peer = {remote_via_link, LinkPid, StreamId}}}; %% --- send -------------------------------------------------------------- handle_call({send, _Encoding, _Body}, _From, #state{closed_send = true} = State) -> {reply, {error, send_closed}, State}; handle_call({send, Encoding, Body}, _From, #state{role = Role, mode = Mode} = State) -> send_chunk(may_send(Role, Mode), Encoding, Body, State); %% --- recv -------------------------------------------------------------- handle_call({recv, Timeout}, From, State) -> handle_recv(From, Timeout, State); %% --- close_send -------------------------------------------------------- handle_call(close_send, _From, #state{closed_send = true} = State) -> {reply, ok, State}; handle_call(close_send, _From, State) -> {_Sent, State1} = forward_to_peer(State, {end_stream, send}), {reply, ok, ended_when_both_closed(State1#state{closed_send = true})}; %% --- close ------------------------------------------------------------- handle_call(close, _From, State) -> {_Sent, State0} = forward_to_peer(State, {end_stream, both}), State1 = State0#state{closed_send = true, closed_recv = true}, State2 = drain_waiters(eof, State1), {reply, ok, session_ended(closed, State2)}; %% --- await_reply ------------------------------------------------------- handle_call({await_reply, _Timeout}, _From, #state{reply = {ok, _} = R} = State) -> {reply, R, State}; handle_call({await_reply, _Timeout}, _From, #state{reply = {error, _} = R} = State) -> {reply, R, State}; handle_call({await_reply, Timeout}, From, State) -> Ref = case Timeout of infinity -> undefined; N -> erlang:send_after(N, self(), {reply_timeout, From}) end, Waiters = [{From, Ref} | State#state.reply_waiters], {noreply, State#state{reply_waiters = Waiters}}; %% --- set_reply --------------------------------------------------------- handle_call({set_reply, Result}, _From, #state{reply = undefined} = State) -> replied(forward_to_peer(State, {reply, Result}), Result); handle_call({set_reply, _Result}, _From, State) -> {reply, ok, State}; handle_call({abort, Code, Message}, _From, State) -> {reply, ok, abort_session(Code, Message, State)}; %% --- controlling_process ----------------------------------------------- %% Only the owner hands the stream over. handle_call({controlling_process, NewOwner}, {Owner, _Tag}, #state{owner = Owner} = State) -> {reply, ok, hand_over(NewOwner, State)}; handle_call({controlling_process, _NewOwner}, _From, State) -> {reply, {error, not_owner}, State}; %% --- info -------------------------------------------------------------- handle_call(info, _From, State) -> Map = #{ id => State#state.id, role => State#state.role, mode => State#state.mode, peer => State#state.peer, inbox_size => queue:len(State#state.inbox), inbox_bytes => State#state.inbox_bytes, waiters => queue:len(State#state.waiters), closed_recv => State#state.closed_recv, closed_send => State#state.closed_send, seq_out => State#state.seq_out, seq_in => State#state.seq_in, reply => State#state.reply }, {reply, Map, State}; handle_call(_Msg, _From, State) -> {reply, {error, unknown}, State}. %% --- peer-delivered events -------------------------------------------- handle_cast({peer_chunk, Encoding, Body}, State) -> {noreply, chunk_arrived(Encoding, Body, State)}; handle_cast({peer_end, Role}, State) when Role =:= send; Role =:= both -> {noreply, end_arrived(Role, State)}; handle_cast({peer_error, Code, Message}, State) -> {noreply, error_arrived(Code, Message, State)}; handle_cast({peer_reply, Result}, State) -> {noreply, reply_arrived(Result, State)}; handle_cast({peer_frame, #{frame_type := Type} = Frame}, State) -> {noreply, peer_frame(macula_frame:control_frame(Type), Frame, State)}; handle_cast({peer_frame, Frame}, State) -> {noreply, peer_frame(false, Frame, State)}; handle_cast(_Msg, State) -> {noreply, State}. %% --- info / monitors / timers ----------------------------------------- handle_info({recv_timeout, From}, State) -> %% Drop this waiter and reply timeout — only if it's still queued {Replied, NewQ} = drop_waiter_and_reply(From, {error, timeout}, State#state.waiters), case Replied of true -> ok; false -> ok % already served end, {noreply, State#state{waiters = NewQ}}; handle_info({reply_timeout, From}, State) -> NewWaiters = lists:filter( fun({F, _Ref}) when F =:= From -> gen_server:reply(F, {error, timeout}), false; (_) -> true end, State#state.reply_waiters), {noreply, State#state{reply_waiters = NewWaiters}}; handle_info({'DOWN', Ref, process, Pid, _Reason}, State) -> IsOwner = Ref =:= State#state.owner_ref andalso Pid =:= State#state.owner, handle_down(IsOwner, Pid, State); %% A write of this stream's bytes failed on its link. The stream ends %% here with a transport failure, which its readers and reply waiters %% receive. It is not a refusal, so its connection hears nothing. handle_info({stream_write_failed, Sid, Reason}, #state{id = Sid} = State) -> {noreply, transport_failed({error, {transport, Reason}}, State)}; handle_info(_Msg, State) -> {noreply, State}. terminate(_Reason, _State) -> ok. %% A link-carried stream holds its key's loader: status output and crash %% reports show it as a printed function, and any key they reach redacted. format_status(Status) -> macula_node_keys:redacted(Status). %%%=================================================================== %%% Internal helpers %%%=================================================================== %% @private Dispatch a stream-level action to the peer, returning what %% the send gave and the stream's next state. %% %% Peer-shape-aware: %% {local, Pid}: an in-process pair; cast the symmetric deliver_* %% helper directly. %% {remote_via_link, L, Sid}: sign the frame with the stream's own key %% and number, and hand the link its bytes. A side sends %% nothing after its own STREAM_END. %% %% Action shapes: %% {chunk, Encoding, Body} %% {end_stream, send | both} %% {error, Code, Message} %% {reply, Result} forward_to_peer(#state{peer = undefined} = S, _Action) -> {{error, no_peer}, S}; forward_to_peer(#state{peer = {local, Pid}} = S, {chunk, Encoding, Body}) -> {deliver_chunk(Pid, Encoding, Body), S#state{seq_out = S#state.seq_out + 1}}; forward_to_peer(#state{peer = {local, Pid}} = S, {end_stream, Role}) -> {deliver_end(Pid, Role), S}; forward_to_peer(#state{peer = {local, Pid}} = S, {error, Code, Message}) -> {deliver_error(Pid, Code, Message), S}; forward_to_peer(#state{peer = {local, Pid}} = S, {reply, Result}) -> {deliver_reply(Pid, Result), S}; forward_to_peer(#state{peer = {remote_via_link, _Link, _Sid}, closed_send = true} = S, _Action) -> {{error, send_closed}, S}; forward_to_peer(#state{peer = {remote_via_link, Link, Sid}, role = Role, mode = Mode} = S, Action) -> #{frame_type := Type} = Spec = frame_spec(Action, S#state.seq_out), sent_via_link(allowed(Role, Mode, Type), encodable(Spec), Spec, Link, Sid, S). %% @private The frame a link-carried stream sends for an action, with its %% next sequence number. An error reply travels as STREAM_ERROR with code %% `error'. frame_spec({chunk, Encoding, Body}, Seq) -> #{frame_type => stream_data, seq => Seq, encoding => Encoding, body => Body}; frame_spec({end_stream, Role}, Seq) -> #{frame_type => stream_end, seq => Seq, role => Role}; frame_spec({error, Code, Message}, Seq) -> #{frame_type => stream_error, seq => Seq, code => Code, message => error_text(Message)}; frame_spec({reply, {ok, Value}}, Seq) -> #{frame_type => stream_reply, seq => Seq, payload => Value}; frame_spec({reply, {error, Reason}}, Seq) -> #{frame_type => stream_error, seq => Seq, code => <<"error">>, message => error_text(Reason)}. %% @private A reason as STREAM_ERROR message text: a binary, or an atom's %% name, that is valid UTF-8 of at most 256 bytes. Anything else sends an %% empty message, so no other term is rendered onto the wire. error_text(Reason) when is_atom(Reason) -> error_text(atom_to_binary(Reason)); error_text(Reason) when is_binary(Reason), byte_size(Reason) =< ?MAX_ERROR_TEXT_BYTES -> valid_text(unicode:characters_to_binary(Reason)); error_text(_Reason) -> <<>>. valid_text(Text) when is_binary(Text) -> Text; valid_text(_Invalid) -> <<>>. %% @private A caller sends no STREAM_REPLY, and no STREAM_DATA in a %% server_stream. allowed(client, _Mode, stream_reply) -> false; allowed(client, server_stream, stream_data) -> false; allowed(_Role, _Mode, _Type) -> true. encodable(#{encoding := msgpack, body := Body}) -> macula_frame:check_payload(Body); encodable(#{payload := Payload}) -> macula_frame:check_payload(Payload); encodable(_Spec) -> ok. sent_via_link(false, _Encodable, _Spec, _Link, _Sid, S) -> {{error, not_allowed}, S}; sent_via_link(true, {error, _} = Unsendable, _Spec, _Link, _Sid, S) -> {Unsendable, S}; sent_via_link(true, ok, Spec, Link, Sid, #state{seq_out = Seq} = S) -> Bytes = macula_frame:encode(signed_frame(Spec, S)), {macula_station_link:send_stream_bytes(Link, Sid, Bytes, last_frame(Spec)), S#state{seq_out = Seq + 1}}. signed_frame(Spec, #state{role = server, key = Load, open = Open}) -> macula_frame:provider_stream(Spec, Load(), Open); signed_frame(Spec, #state{role = client, key = Load, open = Open}) -> macula_frame:caller_stream(Spec, Load(), Open). %% @private The last frame from a side, after which its link forgets the %% stream. last_frame(#{frame_type := stream_end, role := both}) -> true; last_frame(#{frame_type := stream_error}) -> true; last_frame(#{frame_type := stream_reply}) -> true; last_frame(_Spec) -> false. %% @private A reply the side may not send is refused and not recorded. replied({{error, not_allowed} = Refused, State}, _Result) -> {reply, Refused, State}; replied({_Sent, State}, Result) -> {reply, ok, State#state{reply = Result}}. %% @private A stream started with the key loader and STREAM_OPEN of a %% link-carried stream signs, numbers and verifies its frames; a local %% pair has none of these. carried(#{key := Key, open := Open, conn := Conn, profile := Profile}, State) -> State#state{key = Key, open = Open, conn = Conn, profile = Profile, verifier = macula_frame:open_stream(Open)}; carried(_LocalPair, State) -> State. %% @private Owner DOWN → stop. Otherwise check whether the dead pid %% was our peer (or our peer's mesh_client for remote peers) and, if %% so, surface as a stream error to any local readers / reply waiters. handle_down(true, _Pid, State) -> {stop, normal, State}; handle_down(false, Pid, #state{peer = {local, Pid}} = State) -> propagate_peer_down(State); handle_down(false, Pid, #state{peer = {remote_via_link, Pid, _Sid}} = State) -> propagate_peer_down(State); handle_down(false, _Pid, State) -> {noreply, State}. propagate_peer_down(State) -> Err = {error, peer_down}, State1 = State#state{closed_recv = true, closed_send = true, peer = undefined, reply = first_reply(State#state.reply, Err)}, State2 = drain_waiters(Err, State1), State3 = settle_reply_waiters_with(Err, State2), {noreply, session_ended(peer_down, State3)}. %% @private In server_stream only the server sends chunks, and in client_stream %% only the client does; bidi takes them both ways. may_send(client, server_stream) -> false; may_send(server, client_stream) -> false; may_send(_Role, _Mode) -> true. peer_may_send(Role, Mode) -> may_send(peer_role(Role), Mode). peer_role(client) -> server; peer_role(server) -> client. %% @private A chunk this side's mode lets it send goes to the peer. Any other %% send is refused, and nothing reaches the peer. send_chunk(true, Encoding, Body, State) -> {Sent, State1} = forward_to_peer(State, {chunk, Encoding, Body}), {reply, Sent, State1}; send_chunk(false, _Encoding, _Body, #state{mode = Mode} = State) -> {reply, {error, {send_not_allowed, Mode}}, State}. %% @private A chunk the mode lets the peer send is delivered or queued. Any %% other chunk ends the session with a stream protocol error, and nothing of it %% is kept. take_chunk(true, Encoding, Body, State) -> State1 = enqueue_or_deliver(Encoding, Body, State), State1#state{seq_in = State1#state.seq_in + 1}; take_chunk(false, _Encoding, _Body, State) -> abort_session(<<"stream_protocol_error">>, <<"the peer sent a chunk its stream mode does not allow">>, State). %% @private End the session with an error: the peer is sent it, both directions %% close, waiting readers and reply waiters get it, a reply already set stays, %% and the owner is told. abort_session(Code, Message, State) -> Err = {error, {Code, Message}}, {_Sent, State0} = forward_to_peer(State, {error, Code, Message}), State1 = State0#state{closed_recv = true, closed_send = true, reply = first_reply(State0#state.reply, Err)}, State2 = drain_waiters(Err, State1), State3 = settle_reply_waiters_with(Err, State2), session_ended(Err, State3). first_reply(undefined, Err) -> Err; first_reply(Reply, _Err) -> Reply. %% @private A session has ended once both of its directions are closed. ended_when_both_closed(#state{closed_recv = true, closed_send = true} = State) -> session_ended(closed, State); ended_when_both_closed(State) -> State. %% @private The owner is told once how the session ended: `closed', %% `{error, {Code, Message}}' or `peer_down'. The stream itself stays %% until its owner ends. session_ended(_How, #state{ended = Ended} = State) when Ended =/= undefined -> State; session_ended(How, #state{owner = Owner} = State) -> Owner ! {macula_stream, ended, self(), How}, State#state{ended = How}. %% @private The new owner is monitored before the old one is let go, so the %% stream always has an owner it ends with; a new owner that is already gone %% ends the stream at once. A new owner is told at once if the session has %% already ended. hand_over(NewOwner, #state{owner_ref = OldRef, ended = Ended} = State) -> NewRef = erlang:monitor(process, NewOwner), true = erlang:demonitor(OldRef, [flush]), ok = tell_new_owner(Ended, NewOwner), State#state{owner = NewOwner, owner_ref = NewRef}. tell_new_owner(undefined, _NewOwner) -> ok; tell_new_owner(How, NewOwner) -> NewOwner ! {macula_stream, ended, self(), How}, ok. %% @private The effects of the peer's frames, the same for both carriers. A %% chunk after the receive side closed is dropped, and any other goes through %% the mode's direction check (`take_chunk/4'). chunk_arrived(_Encoding, _Body, #state{closed_recv = true} = State) -> State; chunk_arrived(Encoding, Body, #state{role = Role, mode = Mode} = State) -> take_chunk(peer_may_send(Role, Mode), Encoding, Body, State). %% Peer half-closed: no more inbound data, and the session has ended once both %% directions are closed. A full close keeps how the session ended as its %% reply, unless a reply is already set, so an await_reply called later %% returns it at once. end_arrived(send, State) -> ended_when_both_closed(drain_waiters(eof, State#state{closed_recv = true})); end_arrived(both, #state{reply = Reply} = State) -> State1 = drain_waiters(eof, State#state{closed_recv = true, closed_send = true, reply = first_reply(Reply, {error, peer_closed})}), session_ended(closed, settle_reply_waiters_with({error, peer_closed}, State1)). %% The peer's STREAM_ERROR ends the session: how it ended stays as the reply, %% unless one is already set, and the owner is told. error_arrived(Code, Message, #state{reply = Reply} = State) -> Err = {error, {Code, Message}}, session_ended(Err, ended_with(Err, State#state{reply = first_reply(Reply, Err)})). %% A reply takes effect while the session runs, or once it has ended without %% a result. A session that ended with a result keeps it, whichever carrier a %% later reply comes by. reply_arrived(_Late, #state{ended = Ended, reply = Reply} = State) when Ended =/= undefined, Reply =/= undefined -> State; reply_arrived(Result, State) -> settle_reply_waiters_with(Result, State#state{reply = Result}). %% Both sides closed, with every reader and reply waiter answered with Err. ended_with(Err, State) -> State1 = drain_waiters(Err, State#state{closed_recv = true, closed_send = true}), settle_reply_waiters_with(Err, State1). %% A transport failure ends the session too, and the owner is told. transport_failed(Err, #state{reply = Reply} = State) -> session_ended(Err, ended_with(Err, State#state{peer = undefined, reply = first_reply(Reply, Err)})). %% @private A frame of a type that belongs on the control stream has no place %% on a dedicated stream, in either profile. It is the connection peer's %% doing, so the connection is rejected with malformed_frame, and this %% stream ends with that transport failure. Once the session has ended, any %% other frame takes no effect and is reported as stream_ended, which is not %% charged. Any other frame is verified. peer_frame(true, _Frame, #state{conn = Conn} = State) -> ok = macula_peering:reject(Conn, malformed_frame), transport_failed({error, {transport, malformed_frame}}, State); peer_frame(false, _Frame, #state{ended = Ended, conn = Conn} = State) when Ended =/= undefined -> ok = macula_peering:object_refused(Conn, stream_ended), State; peer_frame(false, Frame, #state{role = Role, verifier = Verifier, profile = Profile} = State) -> verified_frame(peer_verified(Role, Frame, Verifier, Profile), State). %% @private A link-carried stream's caller side verifies the provider's %% frames, and its provider side the caller's. peer_verified(client, Frame, Verifier, Profile) -> macula_frame:verify_provider_stream(Frame, Verifier, Profile); peer_verified(server, Frame, Verifier, Profile) -> macula_frame:verify_caller_stream(Frame, Verifier, Profile). verified_frame({ok, Fields, Verifier}, State) -> peer_event(Fields, State#state{verifier = Verifier}); verified_frame({error, Refusal}, #state{conn = Conn} = State) -> ok = macula_peering:object_refused(Conn, Refusal), State. peer_event(#{frame_type := stream_data, encoding := Encoding, body := Body}, State) -> chunk_arrived(Encoding, Body, State); peer_event(#{frame_type := stream_end, role := Role}, State) -> end_arrived(Role, State); peer_event(#{frame_type := stream_error, code := Code, message := Message}, State) -> error_arrived(Code, Message, State); peer_event(#{frame_type := stream_reply, payload := Payload}, State) -> reply_arrived({ok, Payload}, State). %% @doc Either deliver a chunk to a waiting recv/2 caller or queue it. enqueue_or_deliver(Encoding, Body, #state{waiters = W0} = State) -> case queue:out(W0) of {{value, {From, Ref}}, W1} -> cancel_timer(Ref), gen_server:reply(From, chunk_to_recv_result(Encoding, Body)), State#state{waiters = W1}; {empty, _} -> Kept = kept(Encoding, Body), enqueue(chunk_bytes(Encoding, Kept), Encoding, Kept, State) end. %% A chunk that would take the memory the queued chunks hold past the %% stream's bound, or a served stream's caller or node past its budget for %% unread bytes, ends the session with resource_exhausted, and nothing of it %% is kept: the receiving side has no room, which a peer that keeps to the %% protocol cannot see coming. enqueue(Bytes, _Encoding, _Kept, #state{inbox_bytes = Queued, max_inbox_bytes = Max} = State) when Queued + Bytes > Max -> abort_session(<<"resource_exhausted">>, <<"the stream keeps no more unread bytes">>, State); enqueue(Bytes, Encoding, Kept, State) -> queue_charged(charge_budget(Bytes, State), Bytes, Encoding, Kept, State). queue_charged(ok, Bytes, Encoding, Kept, #state{inbox = Inbox, inbox_bytes = Queued} = State) -> State#state{inbox = queue:in({Encoding, Kept}, Inbox), inbox_bytes = Queued + Bytes}; queue_charged({error, _Refused}, _Bytes, _Encoding, _Kept, State) -> abort_session(<<"resource_exhausted">>, <<"the node keeps no more unread bytes for this session">>, State). %% A served stream on a station link charges what it keeps unread to its %% caller and the node (`macula_stream_sessions'), and gives it back when a %% reader takes it or the stream ends; any other stream keeps only its own %% bound. charge_budget(Bytes, #state{role = server, peer = {remote_via_link, _Link, _Sid}}) -> macula_stream_sessions:charge(self(), Bytes); charge_budget(_Bytes, _State) -> ok. release_budget(Bytes, #state{role = server, peer = {remote_via_link, _Link, _Sid}}) -> macula_stream_sessions:release(self(), Bytes); release_budget(_Bytes, _State) -> ok. %% A chunk as the inbox keeps it: a copy, so a body that is part of the frame %% it arrived in keeps none of the rest of that frame. kept(raw, Body) when is_binary(Body) -> binary:copy(Body); kept(_Encoding, Body) -> binary_to_term(term_to_binary(Body)). %% The memory a queued chunk holds: its term on the heap, with the tuple and %% the queue cell that hold it (two words), and every binary it keeps off the %% heap. An empty chunk still holds its cell and tuple. chunk_bytes(Encoding, Kept) -> (erts_debug:flat_size({Encoding, Kept}) + 2) * erlang:system_info(wordsize) + off_heap_bytes(Kept). %% Binaries over 64 bytes live off the heap; smaller ones are counted in the %% heap size. off_heap_bytes(Bin) when is_binary(Bin), byte_size(Bin) > 64 -> byte_size(Bin); off_heap_bytes(Tuple) when is_tuple(Tuple) -> off_heap_bytes(tuple_to_list(Tuple)); off_heap_bytes(Map) when is_map(Map) -> off_heap_bytes(maps:to_list(Map)); off_heap_bytes([Head | Tail]) -> off_heap_bytes(Head) + off_heap_bytes(Tail); off_heap_bytes(_Other) -> 0. handle_recv(From, _Timeout, #state{inbox = Inbox} = State) -> case queue:out(Inbox) of {{value, {Encoding, Body}}, Rest} -> take_queued(Encoding, Body, Rest, State); {empty, _} when State#state.closed_recv -> {reply, recv_ended(State#state.ended), State}; {empty, _} -> queue_waiter(From, _Timeout, State) end. %% What a reader gets once the receive side has closed and nothing is queued: %% eof after a clean close, or while only the peer's side has closed, and how %% the session ended after an error, an abort, a transport failure or a lost %% peer, so a cut-off stream never reads as complete. recv_ended({error, _} = Err) -> Err; recv_ended(peer_down) -> {error, peer_down}; recv_ended(_ClosedOrRunning) -> eof. %% A reader takes a queued chunk: its bytes leave the stream's count, and a %% served stream gives them back to its caller's and the node's budget. take_queued(Encoding, Body, Rest, #state{inbox_bytes = Queued} = State) -> Bytes = chunk_bytes(Encoding, Body), ok = release_budget(Bytes, State), {reply, chunk_to_recv_result(Encoding, Body), State#state{inbox = Rest, inbox_bytes = Queued - Bytes}}. queue_waiter(From, Timeout, State) -> Ref = case Timeout of infinity -> undefined; 0 -> immediate; N when is_integer(N) -> erlang:send_after(N, self(), {recv_timeout, From}) end, case Ref of immediate -> {reply, {error, would_block}, State}; _ -> Waiters = queue:in({From, Ref}, State#state.waiters), {noreply, State#state{waiters = Waiters}} end. chunk_to_recv_result(raw, Body) -> {chunk, Body}; chunk_to_recv_result(msgpack, Body) -> {data, Body}; chunk_to_recv_result(Other, Body) -> {data, {Other, Body}}. drain_waiters(Reply, State) -> drain_waiters(Reply, State#state.waiters, State). drain_waiters(Reply, Q, State) -> case queue:out(Q) of {{value, {From, Ref}}, Rest} -> cancel_timer(Ref), gen_server:reply(From, Reply), drain_waiters(Reply, Rest, State#state{waiters = Rest}); {empty, _} -> State#state{waiters = queue:new()} end. settle_reply_waiters_with(Result, State) -> lists:foreach( fun({From, Ref}) -> cancel_timer(Ref), gen_server:reply(From, Result) end, State#state.reply_waiters), State#state{reply_waiters = []}. drop_waiter_and_reply(From, Reply, Q) -> %% Walk the queue once, dropping the matching waiter and replying. L = queue:to_list(Q), {Match, Rest} = lists:partition(fun({F, _R}) -> F =:= From end, L), case Match of [{F, _Ref}] -> gen_server:reply(F, Reply), {true, queue:from_list(Rest)}; _ -> {false, Q} end. cancel_timer(undefined) -> ok; cancel_timer(immediate) -> ok; cancel_timer(Ref) -> erlang:cancel_timer(Ref), ok.