%%%------------------------------------------------------------------- %%% @doc QUIC Distribution Carrier for Erlang. %%% %%% This module implements the Erlang distribution carrier interface %%% using QUIC transport via quicer. It replaces inet_tcp_dist to enable %%% distributed Erlang over QUIC with: %%% %%% - Built-in TLS 1.3 encryption (mandatory) %%% - NAT-friendly UDP-based transport %%% - Connection migration support %%% - Stream multiplexing for message priorities %%% - Decentralized discovery (no EPMD required) %%% %%% When MACULA_DIST_MODE=relay, distribution is tunneled through the %%% Macula relay mesh via gen_tcp loopback pairs bridged to pub/sub. %%% See `macula_dist_relay' for relay tunnel details. %%% %%% == Usage == %%% %%% Start the VM with: %%% erl -proto_dist macula -no_epmd -start_epmd false %%% %%% Or in vm.args: %%% -proto_dist macula %%% -no_epmd %%% -start_epmd false %%% -macula_dist_port 4433 %%% %%% @copyright 2025 Macula.io Apache-2.0 %%% @end %%%------------------------------------------------------------------- -module(macula_dist). -include_lib("kernel/include/logger.hrl"). -include_lib("kernel/include/net_address.hrl"). -include_lib("kernel/include/dist.hrl"). -include_lib("kernel/include/dist_util.hrl"). %% Distribution carrier callbacks (required by OTP) -export([ listen/1, accept/1, accept_connection/5, setup/5, close/1, select/1, childspecs/0 ]). %% Additional exports for integration -export([ is_node_name/1, splitname/1, address/0 ]). %% Internal exports (used by spawn_link) -export([ acceptor_loop/2, do_accept/3, do_setup/6 ]). -ifdef(TEST). %% Exports for unit tests — pure helpers that are otherwise private. -export([ frame_outgoing/2, packet_mode/1, set_packet_mode/2, extract_frame/4, parse_frames/2 ]). -endif. -define(FAMILY, inet). -define(DRIVER, macula_dist). -define(DEFAULT_PORT, 4433). -define(ALPN, "macula-dist"). -define(HANDSHAKE_TIMEOUT, 30000). -define(CONNECT_TIMEOUT, 15000). %%%=================================================================== %%% Distribution Carrier Callbacks %%%=================================================================== %% @doc Return child specifications for the distribution supervisor. -spec childspecs() -> [supervisor:child_spec()]. childspecs() -> []. %% @doc Listen for incoming distribution connections. %% %% Dispatches on the current mode (set via MACULA_DIST_MODE env var): %% - `relay' (legacy): pub/sub bridge through the station — no local listener %% - `dist_relay': tunnels via the dedicated dist relay — no local listener %% - `direct' (default): raw QUIC listener on the dist port -spec listen(atom()) -> {ok, {term(), #net_address{}, 1..3}} | {error, term()}. listen(NodeName) -> do_listen(dist_mode(), NodeName). do_listen(relay, NodeName) -> ?LOG_INFO("[dist] Relay mode (pub/sub bridge) — no QUIC listener"), {ok, {relay_mode, make_address(NodeName, 0), 1}}; do_listen(dist_relay, NodeName) -> ?LOG_INFO("[dist] dist_relay mode — tunnels via macula-dist-relay"), register_node_with_dist_relay(NodeName), {ok, {dist_relay_mode, make_address(NodeName, 0), 1}}; do_listen(direct, NodeName) -> Port = get_dist_port(NodeName), handle_quic_listen(start_quic_listener(Port), NodeName, Port). handle_quic_listen({ok, Listener}, NodeName, Port) -> {ok, {Listener, make_address(NodeName, Port), 1}}; handle_quic_listen({error, Reason}, _NodeName, _Port) -> ?LOG_ERROR("[dist] Listen failed: ~p", [Reason]), {error, Reason}. register_node_with_dist_relay(NodeName) -> %% The dist_relay_client started on node boot should be up by now. %% If it's not, log + continue — outgoing setup will fail cleanly %% with client_not_running which is better than blocking listen/1. maybe_announce_node(macula_dist_relay_client:whereis_client(), NodeName). maybe_announce_node(undefined, NodeName) -> ?LOG_WARNING("[dist] dist_relay_client not running — dist over relay will " "fail until macula:join_mesh/1 is called with dist_relay opt " "(node=~p)", [NodeName]); maybe_announce_node(_Pid, _NodeName) -> %% Client was started with the NodeName already (via start_link) and %% sent identify during init. Nothing more to do here. ok. %% @doc Accept incoming connections. Called in a loop by net_kernel. %% %% In direct-QUIC mode: acceptor_loop runs accept on the listener handle. %% In relay (pub/sub) / dist_relay modes: acceptor_loop blocks — inbound %% connections arrive via the respective client which delivers the %% `{accept, ...}' messages directly to net_kernel. -spec accept(term()) -> pid(). accept(dist_relay_mode = Sentinel) -> register_kernel_with_dist_relay(macula_dist_relay_client:whereis_client()), spawn_link(?MODULE, acceptor_loop, [self(), Sentinel]); accept(ListenerHandle) -> spawn_link(?MODULE, acceptor_loop, [self(), ListenerHandle]). register_kernel_with_dist_relay(undefined) -> ?LOG_ERROR("[dist] Cannot accept in dist_relay mode — client not running"); register_kernel_with_dist_relay(Client) -> ok = macula_dist_relay_client:set_kernel(Client, self()). %% @doc Accept a distribution connection from a remote node. %% Socket is either {QuicConn, Stream} (direct QUIC) or a gen_tcp port (relay). -spec accept_connection(pid(), term(), node(), term(), term()) -> pid(). accept_connection(AcceptPid, Socket, MyNode, Allowed, SetupTime) when is_port(Socket) -> spawn_link(?MODULE, do_accept, [ {AcceptPid, Socket, Socket, MyNode, Allowed, SetupTime}, self(), unused ]); accept_connection(AcceptPid, {QuicConn, Stream}, MyNode, Allowed, SetupTime) -> spawn_link(?MODULE, do_accept, [ {AcceptPid, QuicConn, Stream, MyNode, Allowed, SetupTime}, self(), unused ]). %% @doc Setup an outgoing distribution connection. %% self() here is net_kernel — captured as Kernel for the spawned process. -spec setup(node(), term(), atom(), term(), term()) -> pid(). setup(Node, Type, MyNode, LongOrShortNames, SetupTime) -> Kernel = self(), spawn_link(?MODULE, do_setup, [ Kernel, Node, Type, MyNode, LongOrShortNames, SetupTime ]). %% @doc Close a distribution connection. -spec close(term()) -> ok. close(Socket) when is_port(Socket) -> catch gen_tcp:close(Socket), ok; close({S, S}) when is_port(S) -> catch gen_tcp:close(S), ok; close({QuicConn, _Stream}) -> catch macula_quic:close_connection(QuicConn), ok; close(QuicConn) when is_reference(QuicConn) -> catch macula_quic:close_connection(QuicConn), ok; close(_) -> ok. %% @doc Check if this module should handle distribution to the given node. -spec select(atom()) -> boolean(). select(Node) -> case splitname(Node) of {_Port, _Host} -> true; false -> false end. %%%=================================================================== %%% Name Handling %%%=================================================================== %% @doc Check if a string is a valid node name. -spec is_node_name(string()) -> boolean(). is_node_name(Name) -> case splitname(Name) of {_Port, _Host} -> true; _ -> false end. %% @doc Split a node name into port and host. %% Supports two formats: %% - port@host (e.g., 4433@192.168.1.100) — explicit port %% - name@host (e.g., test@nanode1.example.com) — uses default port -spec splitname(atom() | string()) -> {integer(), string()} | false. splitname(NodeName) when is_atom(NodeName) -> splitname(atom_to_list(NodeName)); splitname(NodeName) when is_list(NodeName) -> splitname_tokens(string:tokens(NodeName, "@")). splitname_tokens([NameOrPort, Host]) -> splitname_port(catch list_to_integer(NameOrPort), Host); splitname_tokens(_) -> false. splitname_port(Port, Host) when is_integer(Port), Port > 0, Port < 65536 -> {Port, Host}; splitname_port(_, Host) -> DefaultPort = application:get_env(kernel, macula_dist_port, ?DEFAULT_PORT), {DefaultPort, Host}. %% @doc Return address information for this distribution. -spec address() -> #net_address{}. address() -> {ok, Host} = inet:gethostname(), Port = application:get_env(kernel, macula_dist_port, ?DEFAULT_PORT), make_address(Port, Host). %%%=================================================================== %%% Internal — Mode Detection %%%=================================================================== %% @private Distribution transport mode. %% - `relay' (legacy): pub/sub bridge via station (macula-relay). MACULA_DIST_MODE=relay %% - `dist_relay': dedicated dist relay (macula-dist-relay). MACULA_DIST_MODE=dist_relay %% - `direct' (default): raw QUIC on the dist port -spec dist_mode() -> relay | dist_relay | direct. dist_mode() -> mode_from_env(os:getenv("MACULA_DIST_MODE")). mode_from_env("relay") -> relay; mode_from_env("dist_relay") -> dist_relay; mode_from_env(_) -> direct. %%%=================================================================== %%% Internal — QUIC Listener %%%=================================================================== start_quic_listener(Port) -> {CertFile, KeyFile} = get_tls_certs(), ListenerOpts = [ {certfile, CertFile}, {keyfile, KeyFile}, {alpn, [?ALPN]}, {idle_timeout_ms, 60000}, {peer_bidi_stream_count, 100}, {peer_unidi_stream_count, 100} ], case macula_quic:listen(Port, ListenerOpts) of {ok, Listener} -> ?LOG_INFO("[dist] Listening on UDP port ~p", [Port]), {ok, Listener}; {error, Reason} -> {error, Reason} end. %%%=================================================================== %%% Internal — Acceptor Loop (flattened from 5-level nesting) %%%=================================================================== %% @private Acceptor loop. In all currently-supported modes inbound %% connections arrive via the relay or dist-relay client (message %% delivery), not via direct QUIC accept. Direct-listener mode was %% never implemented end-to-end (the Quinn NIF binding for blocking %% accept doesn't exist) and the supervisor today only spawns the %% relay or dist_relay variants, so the loop just blocks on `stop`. acceptor_loop(_Kernel, relay_mode) -> receive stop -> ok end; acceptor_loop(_Kernel, dist_relay_mode) -> receive stop -> ok end; acceptor_loop(_Kernel, _Listener) -> ?LOG_WARNING("[dist] Direct-listener acceptor mode is not implemented; " "use relay or dist_relay modes."), receive stop -> ok end. %% Direct-listener acceptance helpers were removed alongside %% macula_quic:accept/2 (which had only ever been a stub). Inbound %% dist connections now arrive exclusively via relay or dist_relay %% modes, where the relay client message-delivers them rather than %% invoking acceptor_loop. %%%=================================================================== %%% Internal — Accept Incoming Connection %%%=================================================================== %% @private Handle incoming connection setup. %% _Unused kept for arity compatibility with spawn_link call sites. do_accept({AcceptPid, QuicConn, Stream, MyNode, Allowed, SetupTime}, Kernel, _Unused) -> Timer = dist_util:start_timer(SetupTime), receive {AcceptPid, controller, ok} -> ok after ?HANDSHAKE_TIMEOUT -> dist_util:shutdown(?MODULE, ?LINE, control_transfer_timeout) end, Socket = {QuicConn, Stream}, HSData = make_hs_data(Kernel, MyNode, Socket, Timer, Allowed), dist_util:handshake_other_started(HSData). %%%=================================================================== %%% Internal — Setup Outgoing Connection %%%=================================================================== %% @private Setup outgoing distribution connection. do_setup(Kernel, Node, Type, MyNode, _LongOrShortNames, SetupTime) -> Timer = dist_util:start_timer(SetupTime), case splitname(Node) of {Port, Host} -> do_setup_connect(Kernel, Node, Type, MyNode, Timer, Host, Port); false -> ?LOG_WARNING("[dist] Invalid node name: ~p", [Node]), dist_util:shutdown(?MODULE, ?LINE, invalid_node_name) end. do_setup_connect(Kernel, Node, Type, MyNode, Timer, Host, Port) -> Mode = dist_mode(), ?LOG_INFO("[dist] setup ~p mode=~p host=~s port=~p", [Node, Mode, Host, Port]), handle_connect_result(connect_by_mode(Mode, Node, Host, Port), Kernel, Node, Type, MyNode, Timer). connect_by_mode(relay, Node, Host, Port) -> macula_dist_relay:connect(atom_to_list(Node), Host, Port); connect_by_mode(dist_relay, Node, _Host, _Port) -> connect_via_dist_relay(Node); connect_by_mode(direct, _Node, Host, Port) -> connect_quic(Host, Port). connect_via_dist_relay(Node) -> handle_client_lookup(macula_dist_relay_client:whereis_client(), Node). handle_client_lookup(undefined, _Node) -> {error, dist_relay_client_not_running}; handle_client_lookup(Client, Node) -> NodeBin = atom_to_binary(Node), macula_dist_relay_client:request_tunnel(Client, NodeBin). handle_connect_result({ok, Conn, Stream}, Kernel, Node, Type, MyNode, Timer) -> ?LOG_INFO("[dist] Connected to ~p, starting handshake", [Node]), HSData = make_hs_data(Kernel, MyNode, {Conn, Stream}, Timer, undefined), dist_util:handshake_we_started( HSData#hs_data{other_node = Node, request_type = Type}); handle_connect_result({error, Reason}, _Kernel, Node, _Type, _MyNode, _Timer) -> ?LOG_WARNING("[dist] Connection to ~p failed: ~p", [Node, Reason]), dist_util:shutdown(?MODULE, ?LINE, {connect_failed, Reason}). connect_quic(Host, Port) -> {CertFile, KeyFile} = get_tls_certs(), TlsOpts = macula_tls:quic_client_opts(), BaseOpts = [ {alpn, [?ALPN]}, {certfile, CertFile}, {keyfile, KeyFile}, {idle_timeout_ms, 60000} ], ConnOpts = merge_dist_opts(BaseOpts, TlsOpts), connect_quic_result(macula_quic:connect(Host, Port, ConnOpts, ?CONNECT_TIMEOUT)). connect_quic_result({ok, Conn}) -> open_quic_stream(macula_quic:open_stream(Conn, #{active => false}), Conn); connect_quic_result({error, Reason}) -> {error, Reason}. open_quic_stream({ok, Stream}, Conn) -> {ok, Conn, Stream}; open_quic_stream({error, Reason}, Conn) -> macula_quic:close_connection(Conn), {error, {stream_failed, Reason}}. %%%=================================================================== %%% Internal — Shared hs_data construction %%%=================================================================== make_hs_data(Kernel, MyNode, Socket, Timer, Allowed) -> #hs_data{ kernel_pid = Kernel, this_node = MyNode, socket = Socket, timer = Timer, this_flags = 0, other_flags = 0, allowed = Allowed, f_send = fun quic_send/2, f_recv = fun quic_recv/3, f_setopts_pre_nodeup = fun quic_setopts_pre_nodeup/1, f_setopts_post_nodeup = fun quic_setopts_post_nodeup/1, f_getll = fun quic_getll/1, f_address = fun quic_address/2, f_handshake_complete = fun quic_handshake_complete/3, mf_tick = fun quic_tick/1, mf_getstat = fun quic_getstat/1, mf_setopts = fun quic_setopts/2, mf_getopts = fun quic_getopts/2 }. %%%=================================================================== %%% Socket Operations — dispatch on gen_tcp port vs QUIC reference %%%=================================================================== %% --- send --- %% %% QUIC streams are raw bytestreams — no packet framing is done by the %% transport. dist_util expects `{packet, 2}' during handshake and %% `{packet, 4}' post-nodeup (from gen_tcp's perspective). Without framing, %% recv returns arbitrary-sized chunks that don't align with dist frames. %% %% We emulate the packet option per-stream in the dist controller's process %% dictionary. Default is 2 (handshake mode); `quic_setopts_pre_nodeup' %% upgrades to 4 before post-handshake traffic begins. quic_send(Socket, Data) when is_port(Socket) -> gen_tcp:send(Socket, Data); quic_send({S, S}, Data) when is_port(S) -> gen_tcp:send(S, Data); quic_send({_Conn, Stream}, Data) -> Framed = frame_outgoing(packet_mode(Stream), Data), case macula_quic:send(Stream, Framed) of ok -> ok; {ok, _} -> ok; {error, _} = Err -> Err end. frame_outgoing(2, Data) -> Bin = iolist_to_binary(Data), <<(byte_size(Bin)):16, Bin/binary>>; frame_outgoing(4, Data) -> Bin = iolist_to_binary(Data), <<(byte_size(Bin)):32, Bin/binary>>. %% --- recv (dist_util expects {ok, List}) --- quic_recv(Socket, Length, Timeout) when is_port(Socket) -> recv_tcp(Socket, Length, Timeout); quic_recv({S, S}, Length, Timeout) when is_port(S) -> recv_tcp(S, Length, Timeout); quic_recv({_Conn, Stream}, _Length, Timeout) -> recv_frame(Stream, packet_mode(Stream), recv_timeout(Timeout)). %% Pull one complete length-prefixed frame. Accumulates bytes across %% multiple QUIC data events via a per-stream buffer in the proc dict. recv_frame(Stream, Mode, TimeoutMs) -> Buf = erlang:get({macula_dist_recv_buf, Stream}), extract_frame(Stream, Mode, init_buf(Buf), TimeoutMs). init_buf(undefined) -> <<>>; init_buf(B) when is_binary(B) -> B. extract_frame(Stream, 2, <>, _Timeout) -> stash_buf(Stream, Rest), {ok, binary_to_list(Frame)}; extract_frame(Stream, 4, <>, _Timeout) -> stash_buf(Stream, Rest), {ok, binary_to_list(Frame)}; extract_frame(Stream, Mode, Buf, TimeoutMs) -> wait_more_bytes(Stream, Mode, Buf, TimeoutMs). wait_more_bytes(Stream, Mode, Buf, TimeoutMs) -> receive {quic, Data, Stream, _Flags} when is_binary(Data) -> extract_frame(Stream, Mode, <>, TimeoutMs); {quic, stream_closed, Stream, _Flags} -> stash_buf(Stream, Buf), {error, closed}; {quic, peer_send_shutdown, Stream, _Flags} -> stash_buf(Stream, Buf), {error, closed} after TimeoutMs -> stash_buf(Stream, Buf), {error, timeout} end. stash_buf(Stream, <<>>) -> erlang:erase({macula_dist_recv_buf, Stream}); stash_buf(Stream, Buf) -> erlang:put({macula_dist_recv_buf, Stream}, Buf). packet_mode(Stream) -> packet_mode_of(erlang:get({macula_dist_packet_mode, Stream})). packet_mode_of(undefined) -> 2; packet_mode_of(N) when N =:= 2; N =:= 4 -> N. set_packet_mode(Stream, N) when N =:= 2; N =:= 4 -> erlang:put({macula_dist_packet_mode, Stream}, N), ok. recv_tcp(Socket, Length, Timeout) -> case gen_tcp:recv(Socket, Length, recv_timeout(Timeout)) of {ok, Bin} when is_binary(Bin) -> {ok, binary_to_list(Bin)}; Other -> Other end. recv_timeout(infinity) -> 30000; recv_timeout(T) -> T. %% --- setopts (match inet_tcp_dist: {packet,4} post-handshake) --- quic_setopts_pre_nodeup(Socket) when is_port(Socket) -> inet:setopts(Socket, [{active, false}, {packet, 4}]); quic_setopts_pre_nodeup({S, S}) when is_port(S) -> inet:setopts(S, [{active, false}, {packet, 4}]); quic_setopts_pre_nodeup({_Conn, Stream}) -> set_packet_mode(Stream, 4). quic_setopts_post_nodeup(Socket) when is_port(Socket) -> inet:setopts(Socket, [{active, true}, {packet, 4}, {deliver, port}, binary]); quic_setopts_post_nodeup({S, S}) when is_port(S) -> inet:setopts(S, [{active, true}, {packet, 4}, {deliver, port}, binary]); quic_setopts_post_nodeup({_Conn, _Stream}) -> ok. %% --- getll --- quic_getll(Socket) when is_port(Socket) -> {ok, Socket}; quic_getll({S, S}) when is_port(S) -> {ok, S}; quic_getll({_Conn, _Stream}) -> %% dist_util passes the returned value to erlang:setnode as the DistCtrl. %% Must be a port or a pid that implements the dist controller %% protocol (dist_ctrl_get_data / dist_ctrl_put_data). We return self() %% — the dist controller process (do_setup or do_accept) whose %% post-handshake loop shepherds bytes between the QUIC stream and the %% distribution runtime via the dist_ctrl_* BIFs. %% %% Note: dist_util's default con_loop does NOT implement this %% shuffling — a custom loop is wired in via f_handshake_complete. {ok, self()}. %% --- address --- quic_address(Socket, Node) when is_port(Socket) -> make_net_address(inet:peername(Socket), Node); quic_address({S, S}, Node) when is_port(S) -> make_net_address(inet:peername(S), Node); quic_address({Conn, _Stream}, Node) -> make_net_address(macula_quic:peername(Conn), Node). make_net_address({ok, {IP, Port}}, Node) -> #net_address{address = {IP, Port}, host = atom_to_list(Node), protocol = ?DRIVER, family = ?FAMILY}; make_net_address(_, Node) -> #net_address{address = undefined, host = atom_to_list(Node), protocol = ?DRIVER, family = ?FAMILY}. %% --- handshake_complete --- %% %% dist_util calls this AFTER f_setopts_post_nodeup and BEFORE con_loop. %% For QUIC carrier we hijack the controlling process here and never %% return — dist_util's con_loop doesn't understand `dist_data' %% notifications or raw stream reads, so we run our own loop that shuffles %% bytes between the QUIC stream and the distribution runtime via the %% `erlang:dist_ctrl_*' BIFs. %% %% For gen_tcp sockets (when Socket is a port) we fall through to the %% default no-op; the native port handles dist framing itself. quic_handshake_complete(Socket, _Node, _DHandle) when is_port(Socket) -> ok; quic_handshake_complete({S, S}, _Node, _DHandle) when is_port(S) -> ok; quic_handshake_complete({_Conn, Stream}, Node, DHandle) -> ?LOG_INFO("[dist] handshake complete for ~p — entering ctrl loop", [Node]), %% Ask the runtime to return {Size, Data} tuples from dist_ctrl_get_data. %% Returns the PREVIOUS value, so we ignore it rather than asserting. _ = erlang:dist_ctrl_set_opt(DHandle, get_size, true), ok = erlang:dist_ctrl_get_data_notification(DHandle), %% Handshake-phase recv buffer may still hold bytes that arrived AFTER %% the last handshake frame (peer's first post-nodeup packet-4 frame, %% typically). If we drop them here, BEAM's dist parser chokes on the %% following frame because the stream is truncated mid-header. Seed %% the ctrl_loop's buffer with whatever recv_frame stashed. Seed = take_stashed_bytes(Stream), Seeded = put_incoming(DHandle, Seed), %% Don't return — dist_util's con_loop cannot drive the %% `dist_data' / `dist_ctrl_get_data' protocol. Our loop replaces it. ctrl_loop(Node, Stream, DHandle, Seeded). take_stashed_bytes(Stream) -> case erlang:erase({macula_dist_recv_buf, Stream}) of undefined -> <<>>; Bin when is_binary(Bin) -> Bin end. %% @private Dist controller loop — flat clauses, one message class each. %% %% `InBuf' is the accumulator for inbound bytes from the QUIC stream. %% A packet-4 frame may arrive split across several `{quic, ...}' events %% so we buffer until we have a complete frame (or several). ctrl_loop(Node, Stream, DHandle, InBuf) -> receive Msg -> handle_msg(Msg, Node, Stream, DHandle, InBuf) end. handle_msg(dist_data, Node, Stream, DHandle, InBuf) -> drain_out(DHandle, Stream), ok = erlang:dist_ctrl_get_data_notification(DHandle), ctrl_loop(Node, Stream, DHandle, InBuf); handle_msg({quic, Data, Stream, _Flags}, Node, Stream, DHandle, InBuf) when is_binary(Data) -> NewBuf = put_incoming(DHandle, <>), ctrl_loop(Node, Stream, DHandle, NewBuf); handle_msg({quic, peer_send_shutdown, Stream, _}, Node, Stream, _DH, _B) -> exit({shutdown, {Node, peer_send_shutdown}}); handle_msg({quic, stream_closed, Stream, _}, Node, Stream, _DH, _B) -> exit({shutdown, {Node, stream_closed}}); handle_msg({quic, transport_shutdown, _, _}, Node, _Stream, _DH, _B) -> exit({shutdown, {Node, transport_shutdown}}); handle_msg({quic, closed, _, _}, Node, _Stream, _DH, _B) -> exit({shutdown, {Node, closed}}); handle_msg({_Kernel, disconnect}, Node, _Stream, _DH, _B) -> exit({shutdown, {Node, disconnected}}); handle_msg({_Kernel, tick}, Node, Stream, DHandle, InBuf) -> send_tick(Stream), ctrl_loop(Node, Stream, DHandle, InBuf); handle_msg({_Kernel, aux_tick}, Node, Stream, DHandle, InBuf) -> send_tick(Stream), ctrl_loop(Node, Stream, DHandle, InBuf); handle_msg({From, Ref, {setopts, _Opts}}, Node, Stream, DHandle, InBuf) -> From ! {Ref, ok}, ctrl_loop(Node, Stream, DHandle, InBuf); handle_msg({From, Ref, {getopts, _Opts}}, Node, Stream, DHandle, InBuf) -> From ! {Ref, {ok, []}}, ctrl_loop(Node, Stream, DHandle, InBuf); handle_msg({From, get_status}, Node, Stream, DHandle, InBuf) -> {ok, R, W, _} = getstat_dist(DHandle), From ! {self(), get_status, {ok, R, W}}, ctrl_loop(Node, Stream, DHandle, InBuf); handle_msg(_Other, Node, Stream, DHandle, InBuf) -> ctrl_loop(Node, Stream, DHandle, InBuf). %% Zero-length packet-4 frame — the dist-layer keepalive. send_tick(Stream) -> _ = macula_quic:send(Stream, <<0:32>>), ok. getstat_dist(DHandle) -> %% erlang:dist_get_stat returns {ok, Read, Write, PendingWrites}. erlang:dist_get_stat(DHandle). %% --- outbound: drain runtime → QUIC stream --- drain_out(DHandle, Stream) -> send_out(erlang:dist_ctrl_get_data(DHandle), DHandle, Stream). send_out(none, _DHandle, _Stream) -> ok; send_out({Size, Data}, DHandle, Stream) -> Frame = iolist_to_binary([<> | Data]), write_frame(macula_quic:send(Stream, Frame)), drain_out(DHandle, Stream). write_frame(ok) -> ok; write_frame({ok, _}) -> ok; write_frame({error, Reason}) -> ?LOG_ERROR("[dist] ctrl_loop send failed: ~p", [Reason]), exit({dist_send_failed, Reason}). %% --- inbound: QUIC stream → runtime. Return leftover tail. --- put_incoming(DHandle, Buf) -> {Frames, Leftover} = parse_frames(Buf, []), lists:foreach(fun(F) -> deliver_frame(DHandle, F) end, Frames), Leftover. %% Zero-length frames are dist-layer ticks — they must NOT be forwarded to %% dist_ctrl_put_data, which rejects them as corrupt. (ssl_dist applies the %% same `when 0 < Size' guard in read_application_dist_data.) deliver_frame(_DHandle, <<>>) -> ok; deliver_frame(DHandle, Frame) -> erlang:dist_ctrl_put_data(DHandle, Frame). %% Parse as many complete `<>' frames as the %% buffer holds. Returns `{FramesInOrder, Leftover}'. parse_frames(<>, Acc) -> parse_frames(Rest, [Payload | Acc]); parse_frames(Partial, Acc) -> {lists:reverse(Acc), Partial}. %% --- tick --- quic_tick(Socket) when is_port(Socket) -> gen_tcp:send(Socket, <<>>); quic_tick({S, S}) when is_port(S) -> gen_tcp:send(S, <<>>); quic_tick({_Conn, Stream}) -> %% Dist tick = zero-payload packet. Frame with current packet mode %% (post-nodeup so always 4 here, but read the mode to stay honest). Frame = frame_outgoing(packet_mode(Stream), <<>>), macula_quic:send(Stream, Frame). %% --- getstat (dist_util expects {ok, R, W, P} 4-tuple) --- quic_getstat(Socket) when is_port(Socket) -> getstat_tcp(Socket); quic_getstat({S, S}) when is_port(S) -> getstat_tcp(S); quic_getstat({Conn, _Stream}) -> case macula_quic:getstat(Conn, [send_cnt, recv_cnt, send_pend]) of {ok, Stats} -> split_stat(Stats, 0, 0, 0); {error, _} -> {ok, 0, 0, 0} end. getstat_tcp(Socket) -> case inet:getstat(Socket, [recv_cnt, send_cnt, send_pend]) of {ok, Stats} -> split_stat(Stats, 0, 0, 0); Error -> Error end. split_stat([{recv_cnt, R} | Rest], _, W, P) -> split_stat(Rest, R, W, P); split_stat([{send_cnt, W} | Rest], R, _, P) -> split_stat(Rest, R, W, P); split_stat([{send_pend, P} | Rest], R, W, _) -> split_stat(Rest, R, W, P); split_stat([], R, W, P) -> {ok, R, W, P}. %% --- setopts / getopts --- quic_setopts(Socket, Opts) when is_port(Socket) -> inet:setopts(Socket, Opts); quic_setopts({S, S}, Opts) when is_port(S) -> inet:setopts(S, Opts); quic_setopts({_Conn, Stream}, Opts) -> lists:foreach( fun({active, Mode}) -> macula_quic:setopt(Stream, active, Mode); (_) -> ok end, Opts), ok. quic_getopts(Socket, Opts) when is_port(Socket) -> inet:getopts(Socket, Opts); quic_getopts({S, S}, Opts) when is_port(S) -> inet:getopts(S, Opts); quic_getopts({_Conn, _Stream}, _Opts) -> {ok, [{active, true}]}. %%%=================================================================== %%% Utility Functions %%%=================================================================== merge_dist_opts(BaseOpts, TlsOpts) -> lists:foldl( fun({Key, Value}, Acc) -> lists:keystore(Key, 1, Acc, {Key, Value}) end, BaseOpts, TlsOpts). get_dist_port(NodeName) when is_atom(NodeName) -> get_dist_port(atom_to_list(NodeName)); get_dist_port(NodeName) when is_list(NodeName) -> get_dist_port_split(splitname(NodeName), NodeName). get_dist_port_split({Port, _Host}, _NodeName) -> Port; get_dist_port_split(false, NodeName) -> get_dist_port_int(catch list_to_integer(NodeName)). get_dist_port_int(Port) when is_integer(Port), Port > 0, Port < 65536 -> Port; get_dist_port_int(_) -> application:get_env(kernel, macula_dist_port, ?DEFAULT_PORT). get_tls_certs() -> CertDir = application:get_env(kernel, macula_dist_cert_dir, "/tmp/macula_dist"), CertFile = filename:join(CertDir, "cert.pem"), KeyFile = filename:join(CertDir, "key.pem"), case filelib:is_regular(CertFile) andalso filelib:is_regular(KeyFile) of true -> {CertFile, KeyFile}; false -> generate_self_signed_cert(CertDir, CertFile, KeyFile) end. generate_self_signed_cert(_CertDir, CertFile, KeyFile) -> ok = filelib:ensure_dir(CertFile), Cmd = io_lib:format( "openssl req -x509 -newkey rsa:2048 -keyout ~s -out ~s " "-days 365 -nodes -subj '/CN=macula-dist' 2>/dev/null", [KeyFile, CertFile]), case os:cmd(lists:flatten(Cmd)) of "" -> ok; Error -> ?LOG_ERROR("[dist] Cert generation failed: ~s", [Error]) end, {CertFile, KeyFile}. make_address(NodeName, Port) when is_atom(NodeName) -> {ok, Host} = inet:gethostname(), make_address(Port, Host); make_address(Port, Host) when is_integer(Port) -> #net_address{address = {Host, Port}, host = Host, protocol = ?DRIVER, family = ?FAMILY}.