quic

%%% -*- erlang -*-
%%%
%%% QUIC Distribution Module
%%% Erlang Distribution over QUIC (RFC 9000)
%%%
%%% Copyright (c) 2024-2026 Benoit Chesneau
%%% Apache License 2.0
%%%
%%% @doc Erlang distribution protocol implementation over QUIC.
%%%
%%% This module implements the Erlang distribution protocol callbacks
%%% using QUIC as the transport layer. It provides:
%%%
%%% - Connection establishment via TLS 1.3 (built into QUIC)
%%% - Multiple streams for parallel message delivery
%%% - Head-of-line blocking avoidance
%%% - Connection migration for NAT traversal
%%% - 0-RTT reconnection for fast session resumption
%%%
%%% == Configuration ==
%%%
%%% Enable QUIC distribution in vm.args:
%%% ```
%%% -proto_dist quic
%%% -epmd_module quic_epmd
%%% -start_epmd false
%%% '''
%%%
%%% Configure in sys.config:
%%% ```
%%% {quic, [
%%%   {dist, [
%%%     {cert_file, "/path/to/cert.pem"},
%%%     {key_file, "/path/to/key.pem"},
%%%     {cacert_file, "/path/to/ca.pem"},
%%%     {verify, verify_peer}
%%%   ]}
%%% ]}
%%% '''
%%%
%%% @end

-module(quic_dist).

-include("quic.hrl").
-include("quic_dist.hrl").
-include_lib("kernel/include/dist_util.hrl").
-include_lib("kernel/include/net_address.hrl").

%% Dialyzer suppressions:
%% - accept_connection: handshake functions have complex control flow
-dialyzer({nowarn_function, [accept_connection/5]}).

%% Distribution module callbacks
-export([
    listen/1,
    listen/2,
    accept/1,
    accept_connection/5,
    setup/5,
    close/1,
    select/1,
    address/0,
    is_node_name/1
]).

%% User stream API
-export([
    open_stream/1,
    open_stream/2,
    send/2,
    send/3,
    close_stream/1,
    reset_stream/1,
    reset_stream/2,
    accept_streams/1,
    stop_accepting/1,
    controlling_process/2,
    list_streams/0,
    list_streams/1,
    get_controller/1
]).

%% Keep-alive cadence derivation (exported for testing the net_ticktime logic)
-export([keep_alive_interval/0, keep_alive_interval_for/1]).

%% Types
-export_type([stream_ref/0, stream_info/0, stream_opt/0]).

-type stream_ref() :: {quic_dist_stream, node(), non_neg_integer()}.
-type stream_opt() :: {priority, 16..255}.
-type stream_info() :: #{
    ref => stream_ref(),
    node => node(),
    stream_id => non_neg_integer(),
    owner => pid(),
    priority => 16..255,
    recv_fin => boolean(),
    send_fin => boolean()
}.

%% Internal exports
-export([
    acceptor_loop/2,
    do_setup/6
]).

%% Per-node connect-time option overrides
-export([
    set_connect_options/2,
    get_connect_options/1,
    clear_connect_options/1
]).

%%====================================================================
%% Distribution Module Callbacks
%%====================================================================

%% @doc Check if this distribution module should be used for the given node.
%% Returns true if the node name is valid and we can potentially connect.
-spec select(node()) -> boolean().
select(Node) ->
    case dist_util:split_node(Node) of
        {node, Name, Host} ->
            %% Try to resolve address via EPMD module
            EpmdMod = net_kernel:epmd_module(),
            case catch EpmdMod:address_please(Name, Host, inet) of
                {ok, _Addr} ->
                    true;
                {ok, _Addr, _Port, _Version} ->
                    true;
                _ ->
                    %% Even if address lookup fails, allow local connections
                    %% This is needed during initial node startup
                    true
            end;
        _ ->
            false
    end.

%% @doc Check if a node name is valid.
-spec is_node_name(atom()) -> boolean().
is_node_name(Node) when is_atom(Node) ->
    case split_node(atom_to_list(Node), $@, []) of
        [_, _Host] -> true;
        _ -> false
    end;
is_node_name(_) ->
    false.

%% @private
%% Split node name on separator character.
split_node([Sep | Rest], Sep, Acc) ->
    [lists:reverse(Acc) | split_node(Rest, Sep, [])];
split_node([C | Rest], Sep, Acc) ->
    split_node(Rest, Sep, [C | Acc]);
split_node([], _Sep, Acc) ->
    [lists:reverse(Acc)].

%% @doc Return the address family to use.
-spec address() -> #net_address{}.
address() ->
    {ok, Host} = inet:gethostname(),
    #net_address{
        host = Host,
        protocol = quic,
        family = inet
    }.

%% @doc Start listening for incoming distribution connections.
-spec listen(Name :: atom()) ->
    {ok, {LSocket :: term(), TcpAddress :: term(), Creation :: non_neg_integer()}}
    | {error, Reason :: term()}.
listen(Name) ->
    listen(Name, #{}).

%% @doc Start listening with options.
-spec listen(Name :: atom(), Opts :: map()) ->
    {ok, {LSocket :: term(), TcpAddress :: term(), Creation :: non_neg_integer()}}
    | {error, Reason :: term()}.
listen(Name, ExtraOpts) ->
    %% Ensure quic application is started - distribution callbacks run
    %% very early, before -eval or application start
    case ensure_quic_started() of
        ok ->
            Config = load_config(),
            Port = get_listen_port(),
            case start_quic_server(Name, Port, Config, ExtraOpts) of
                {ok, ServerName, ActualPort} ->
                    Listener = #quic_dist_listener{
                        server_name = ServerName,
                        port = ActualPort,
                        config = Config
                    },
                    Address = #net_address{
                        address = {{0, 0, 0, 0}, ActualPort},
                        host = localhost,
                        family = inet,
                        protocol = quic
                    },
                    case resolve_creation(Name, ActualPort, Config) of
                        {ok, Creation} ->
                            {ok, {Listener, Address, Creation}};
                        {error, EpmdReason} ->
                            close(Listener),
                            {error, EpmdReason}
                    end;
                {error, Reason} ->
                    {error, Reason}
            end;
        {error, Reason} ->
            {error, Reason}
    end.

%% @private
%% Resolve the creation number. Either a synthetic one (default), or the
%% one returned by registering with the configured epmd_module so that
%% external tooling (e.g. `epmd -names') can find this node.
resolve_creation(Name, _Port, #quic_dist_config{register_with_epmd = false}) ->
    {ok, get_creation(Name)};
resolve_creation(Name, Port, #quic_dist_config{
    register_with_epmd = true,
    discovery_module = DiscoveryModule
}) ->
    %% quic_discovery dispatches via erlang:function_exported/3, which
    %% only sees exports of *loaded* modules. listen/1 runs before the
    %% normal application boot, so force the discovery backend in now.
    _ = code:ensure_loaded(DiscoveryModule),
    NameStr =
        case Name of
            N when is_atom(N) -> atom_to_list(N);
            N when is_list(N) -> N
        end,
    EpmdMod = net_kernel:epmd_module(),
    case EpmdMod:register_node(NameStr, Port, inet) of
        {ok, Creation} -> {ok, Creation};
        {error, _} = Err -> Err
    end.

%% @doc Accept a connection from the distribution listener.
-spec accept(Listen :: term()) -> AcceptPid :: pid().
accept(#quic_dist_listener{server_name = ServerName} = Listener) ->
    %% Spawn acceptor process that will handle incoming connections
    AcceptorPid = spawn_link(?MODULE, acceptor_loop, [self(), Listener]),
    %% Register acceptor so handle_new_connection can notify it
    persistent_term:put({quic_dist_acceptor, ServerName}, AcceptorPid),
    AcceptorPid.

%% @doc Handle an accepted connection.
%% Called by net_kernel when a new connection is accepted.
-spec accept_connection(
    AcceptPid :: pid(),
    Socket :: term(),
    MyNode :: node(),
    Allowed :: term(),
    SetupTime :: non_neg_integer()
) -> pid().
accept_connection(_AcceptPid, DistCtrl, MyNode, Allowed, SetupTime) ->
    %% self() is net_kernel here - capture before spawning.
    Kernel = self(),
    spawn_opt(
        fun() ->
            %% The controller already owns the QUIC connection (listener
            %% transferred ownership in handle_new_connection), so there
            %% is no socket-handoff rendezvous to wait for. Proceed
            %% straight to the dist handshake - reaching mark_pending
            %% fast is what lets dist_util resolve simultaneous connects.
            %%
            %% Do NOT trap exits: dist_util:start_timer/1 spawns a linked
            %% timer whose exit must kill this process on timeout.
            ok = quic_dist_controller:set_supervisor(DistCtrl, Kernel),
            Timer = dist_util:start_timer(SetupTime),
            HSData = create_hs_data(DistCtrl, MyNode, Timer, Allowed, Kernel),
            dist_util:handshake_other_started(HSData)
        end,
        [link, {priority, max}]
    ).

%% @doc Set up an outgoing distribution connection.
%% Called by net_kernel to establish a connection to another node.
-spec setup(
    Node :: node(),
    Type :: atom(),
    MyNode :: node(),
    LongOrShortNames :: shortnames | longnames,
    SetupTime :: non_neg_integer()
) -> pid().
setup(Node, Type, MyNode, LongOrShortNames, SetupTime) ->
    spawn_opt(
        ?MODULE,
        do_setup,
        [self(), Node, Type, MyNode, LongOrShortNames, SetupTime],
        [link, {priority, max}]
    ).

%% @doc Close the distribution listener.
-spec close(Listen :: term()) -> ok.
close(#quic_dist_listener{server_name = ServerName}) ->
    %% First try to stop via supervised server
    catch quic:stop_server(ServerName),
    %% Also check for early boot standalone listener
    Key = {quic_dist_early_listener, ServerName},
    case persistent_term:get(Key, undefined) of
        undefined ->
            ok;
        #{pid := Pid} ->
            catch quic_listener:stop(Pid),
            catch persistent_term:erase(Key),
            ok
    end;
close(_) ->
    ok.

%%====================================================================
%% Internal Functions - Server Setup
%%====================================================================

%% @private
%% Ensure minimal QUIC resources are available for distribution.
%%
%% This function handles the tricky boot sequence:
%% - When using -proto_dist quic, listen/1 is called BEFORE applications start
%% - Calling application:ensure_all_started(quic) would deadlock
%% - Instead, we initialize only the minimal required resources:
%%   1. Ensure crypto is loaded (for TLS operations)
%%   2. Create discovery ETS table (for node lookup)
%%   3. Set early boot flag so start_quic_server uses standalone listener
%%
%% Once boot completes and the quic application starts normally, it will
%% detect and adopt these early boot resources.
ensure_quic_started() ->
    case whereis(quic_sup) of
        Pid when is_pid(Pid) ->
            %% quic application is already running, nothing to do
            ok;
        undefined ->
            %% Early boot - initialize minimal resources without starting app
            ensure_quic_minimal()
    end.

%% @private
%% Minimal initialization for early boot (before quic application starts).
ensure_quic_minimal() ->
    %% Ensure crypto is available - needed for TLS
    case code:ensure_loaded(crypto) of
        {module, crypto} ->
            %% Create discovery ETS table if it doesn't exist
            ensure_discovery_table(),
            %% Mark that we're in early boot mode
            put(quic_dist_early_boot, true),
            ok;
        {error, Reason} ->
            {error, {crypto_not_available, Reason}}
    end.

%% @private
%% Create the discovery ETS table used by quic_discovery_static.
%% This is normally created by quic_sup:init/1.
ensure_discovery_table() ->
    ensure_named_set(quic_discovery_static_nodes).

%% @private
%% Idempotently create a named, public ETS set with read_concurrency.
%% Returns ok regardless of whether the table already existed or was
%% created by a racing process.
ensure_named_set(Name) ->
    case ets:info(Name) of
        undefined ->
            try
                _ = ets:new(
                    Name,
                    [named_table, public, set, {read_concurrency, true}]
                ),
                ok
            catch
                error:badarg ->
                    %% Lost the create race; fine.
                    ok
            end;
        _ ->
            ok
    end.

%% @private
%% Load distribution configuration from command-line args and application environment.
%% Command-line arguments take precedence (using -quic_dist key value format).
load_config() ->
    %% First check init arguments (command line), then fall back to app env
    DistOpts = application:get_env(quic, dist, []),

    #quic_dist_config{
        cert_file = get_init_arg(cert, get_opt(cert_file, DistOpts)),
        key_file = get_init_arg(key, get_opt(key_file, DistOpts)),
        cacert_file = get_init_arg(cacert, get_opt(cacert_file, DistOpts)),
        cert = get_opt(cert, DistOpts),
        key = get_opt(key, DistOpts),
        cacert = get_opt(cacert, DistOpts),
        verify = get_verify_opt(get_opt(verify, DistOpts, verify_none)),
        discovery_module = get_opt(discovery_module, DistOpts, quic_discovery_static),
        nodes = get_opt(nodes, DistOpts, []),
        dns_domain = get_opt(dns_domain, DistOpts),
        lb_enabled = get_opt(lb_enabled, DistOpts, false),
        lb_server_id = get_opt(lb_server_id, DistOpts, auto),
        lb_key = get_opt(lb_key, DistOpts),
        %% Backpressure tuning
        congestion_threshold = get_opt(
            congestion_threshold, DistOpts, ?DEFAULT_QUEUE_CONGESTION_THRESHOLD
        ),
        max_pull_per_notification = get_opt(
            max_pull_per_notification, DistOpts, ?DEFAULT_MAX_PULL_PER_NOTIFICATION
        ),
        backpressure_retry_ms = get_opt(
            backpressure_retry_ms, DistOpts, ?DEFAULT_BACKPRESSURE_RETRY_MS
        ),
        %% Pacing
        pacing_enabled = get_opt(pacing_enabled, DistOpts, true),
        %% Optional post-QUIC, pre-dist auth handshake
        auth_callback = parse_auth_callback(
            get_init_arg(auth_callback, get_opt(auth_callback, DistOpts))
        ),
        auth_handshake_timeout = parse_timeout(
            get_init_arg(auth_handshake_timeout, get_opt(auth_handshake_timeout, DistOpts)),
            10000
        ),
        %% Stock-EPMD registration of the listener port
        register_with_epmd = parse_bool(
            get_init_arg(register_with_epmd, get_opt(register_with_epmd, DistOpts)),
            false
        ),
        %% TLS 1.3 external PSK config (see docs/PSK.md)
        psks = validate_psks(get_opt(psks, DistOpts)),
        psk_callback = parse_psk_callback(
            get_init_arg(psk_callback, get_opt(psk_callback, DistOpts))
        ),
        external_psk = validate_external_psk(get_opt(external_psk, DistOpts))
    }.

%% @private
validate_psks(undefined) ->
    undefined;
validate_psks(Map) when is_map(Map) ->
    %% Require all keys/values to be binaries; reject loudly so
    %% dist startup fails rather than silently dropping the list.
    case
        lists:all(
            fun({K, V}) -> is_binary(K) andalso is_binary(V) end, maps:to_list(Map)
        )
    of
        true -> Map;
        false -> error({bad_config, {psks, malformed_entries}})
    end;
validate_psks(Other) ->
    error({bad_config, {psks, Other}}).

%% @private
validate_external_psk(undefined) ->
    undefined;
validate_external_psk({Id, Secret}) when is_binary(Id), is_binary(Secret) ->
    {Id, Secret};
validate_external_psk({Id, Secret, Modes}) when
    is_binary(Id), is_binary(Secret), is_list(Modes), Modes =/= []
->
    {Id, Secret, Modes};
validate_external_psk(Other) ->
    error({bad_config, {external_psk, Other}}).

%% @private
%% Parse psk_callback config. Accepts literal anonymous funs,
%% {Module, Function} tuples, and "Module:Function" strings from
%% vm.args. Mirrors parse_auth_callback/1 but for arity-1 funs.
parse_psk_callback(undefined) ->
    undefined;
parse_psk_callback({M, F}) when is_atom(M), is_atom(F) ->
    {M, F};
parse_psk_callback(F) when is_function(F, 1) ->
    F;
parse_psk_callback(Str) when is_list(Str) ->
    case string:split(Str, ":") of
        [M, F] when M =/= "", F =/= "" ->
            {list_to_atom(M), list_to_atom(F)};
        _ ->
            undefined
    end;
parse_psk_callback(_) ->
    undefined.

%% @private
parse_auth_callback(undefined) ->
    undefined;
parse_auth_callback({M, F}) when is_atom(M), is_atom(F) ->
    {M, F};
parse_auth_callback(F) when is_function(F, 3) ->
    F;
parse_auth_callback(Str) when is_list(Str) ->
    case string:split(Str, ":") of
        [M, F] when M =/= "", F =/= "" ->
            {list_to_atom(M), list_to_atom(F)};
        _ ->
            undefined
    end;
parse_auth_callback(_) ->
    undefined.

%% @private
parse_timeout(undefined, Default) ->
    Default;
parse_timeout(N, _Default) when is_integer(N), N > 0 ->
    N;
parse_timeout(infinity, _Default) ->
    infinity;
parse_timeout("infinity", _Default) ->
    infinity;
parse_timeout(Str, Default) when is_list(Str) ->
    try list_to_integer(Str) of
        N when N > 0 -> N;
        _ -> Default
    catch
        _:_ -> Default
    end;
parse_timeout(_, Default) ->
    Default.

%% @private
parse_bool(undefined, Default) -> Default;
parse_bool(true, _) -> true;
parse_bool(false, _) -> false;
parse_bool("true", _) -> true;
parse_bool("false", _) -> false;
parse_bool(_, Default) -> Default.

%% @private
%% Get value from init argument -quic_dist_Key Value
get_init_arg(Key, Default) ->
    ArgName = list_to_atom("quic_dist_" ++ atom_to_list(Key)),
    case init:get_argument(ArgName) of
        {ok, [[Value]]} ->
            Value;
        _ ->
            %% Also try plain -quic_dist with key value pairs
            case init:get_argument(quic_dist) of
                {ok, Args} ->
                    find_in_args(atom_to_list(Key), Args, Default);
                _ ->
                    Default
            end
    end.

%% @private
find_in_args(_Key, [], Default) -> Default;
find_in_args(Key, [[Key, Value] | _], _Default) -> Value;
find_in_args(Key, [_ | Rest], Default) -> find_in_args(Key, Rest, Default).

%% @private
get_verify_opt(verify_peer) -> verify_peer;
get_verify_opt(verify_none) -> verify_none;
get_verify_opt("verify_peer") -> verify_peer;
get_verify_opt("verify_none") -> verify_none;
get_verify_opt(_) -> verify_none.

%% @private
get_opt(Key, Opts) ->
    get_opt(Key, Opts, undefined).

get_opt(Key, Opts, Default) when is_list(Opts) ->
    proplists:get_value(Key, Opts, Default);
get_opt(Key, Opts, Default) when is_map(Opts) ->
    maps:get(Key, Opts, Default).

%% @private
%% Get the port to listen on from init arguments or config.
get_listen_port() ->
    case init:get_argument(quic_dist_port) of
        {ok, [[PortStr]]} ->
            list_to_integer(PortStr);
        _ ->
            application:get_env(quic, dist_port, ?QUIC_DIST_DEFAULT_PORT)
    end.

%% @private
%% @doc QUIC keep-alive interval for distribution connections, in milliseconds.
%%
%% net_kernel declares a peer down after net_ticktime with no received QUIC
%% packets (the transport's getstat reports packets_received), so the PING
%% keep-alive — which bypasses stream flow control — must fire well within that
%% window even when a post-burst flow-control stall delays the dist tick. We pace
%% it at net_ticktime/4, matching net_kernel's own tick cadence (~4 chances per
%% window), so a healthy link (latency well under net_ticktime) never trips a
%% false timeout.
%%
%% net_ticktime is read from the kernel application env, NOT via
%% net_kernel:get_net_ticktime/0: this runs inside the dist listen/connect
%% callback, which executes in the net_kernel process, so calling net_kernel
%% would deadlock ({calling_self,...}). The env holds the configured value
%% net_kernel itself uses at startup (a runtime set_net_ticktime/1 is not
%% reflected, which is acceptable for picking a keep-alive cadence).
-spec keep_alive_interval() -> pos_integer().
keep_alive_interval() ->
    keep_alive_interval_for(application:get_env(kernel, net_ticktime, 60)).

%% @private Pure derivation, split out so the net_ticktime cases are testable.
-spec keep_alive_interval_for(term()) -> pos_integer().
keep_alive_interval_for(Ticktime) when is_integer(Ticktime), Ticktime > 0 ->
    max(?QUIC_DIST_KEEP_ALIVE_MIN, (Ticktime * 1000) div 4);
keep_alive_interval_for(_Other) ->
    %% net_ticktime unset or invalid - use the configured fallback.
    ?QUIC_DIST_KEEP_ALIVE_INTERVAL.

%% Start QUIC server for distribution.
%%
%% Two modes of operation:
%% 1. Early boot mode: Start standalone quic_listener directly (no supervision)
%% 2. Normal mode: Use quic:start_server through the supervisor tree
start_quic_server(Name, Port, Config, _ExtraOpts) ->
    %% Load certificate and key (or fall through to PSK-only auth)
    case load_credentials(Config) of
        {ok, Cert, Key, _CACert} ->
            CongestionThreshold = Config#quic_dist_config.congestion_threshold,
            BaseOpts0 = #{
                alpn => [?QUIC_DIST_ALPN],
                idle_timeout => ?QUIC_DIST_IDLE_TIMEOUT,
                %% QUIC-level keep-alive via PING frames (bypasses flow control).
                %% Paced off net_ticktime so net_kernel never sees a stale
                %% connection under load.
                keep_alive_interval => keep_alive_interval(),
                %% Use aggressive initial cwnd for distribution bulk transfers
                initial_window => ?INITIAL_WINDOW_AGGRESSIVE,
                %% Keep a higher congestion floor to avoid liveness stalls
                %% on bursty virtual networks (e.g., Docker bridge).
                minimum_window => ?MINIMUM_WINDOW_DISTRIBUTION,
                %% Higher flow control limits for distribution to avoid blocking
                %% during large message transfers (code loading, large terms)
                max_data => ?DIST_INITIAL_MAX_DATA,
                max_stream_data_bidi_local => ?DIST_INITIAL_MAX_STREAM_DATA,
                max_stream_data_bidi_remote => ?DIST_INITIAL_MAX_STREAM_DATA,
                max_stream_data_uni => ?DIST_INITIAL_MAX_STREAM_DATA,
                %% Backpressure threshold for congestion detection
                congestion_threshold => CongestionThreshold,
                %% Pacing spreads packet sends to avoid bursts
                pacing_enabled => Config#quic_dist_config.pacing_enabled,
                %% Longer recovery duration for virtual network packet reordering
                min_recovery_duration => ?MIN_RECOVERY_DURATION_DISTRIBUTION,
                %% Use known-safe MTU for LAN (1452 bytes, IPv4/IPv6 compatible)
                %% instead of PMTU probing which adds overhead
                max_udp_payload_size => ?DIST_MAX_UDP_PAYLOAD_SIZE,
                pmtu_enabled => false,
                connection_handler => fun(Conn) ->
                    handle_new_connection(Conn)
                end
            },
            %% Add cert/key when present (PSK-only listeners omit them).
            BaseOpts1 =
                case {Cert, Key} of
                    {undefined, undefined} -> BaseOpts0;
                    {_, _} -> BaseOpts0#{cert => Cert, key => Key}
                end,
            %% Add PSK auth when configured.
            Opts = add_psk_listener_opts(BaseOpts1, Config),

            case whereis(quic_sup) of
                Pid when is_pid(Pid) ->
                    %% Normal mode - use supervised server
                    start_supervised_server(Name, Port, Opts);
                undefined ->
                    %% Early boot mode - start standalone listener
                    start_standalone_listener(Name, Port, Opts)
            end;
        {error, Reason} ->
            {error, {credentials, Reason}}
    end.

%% @private
%% Start QUIC server through the normal supervisor tree.
start_supervised_server(Name, Port, Opts) ->
    ServerName = dist_server_name(Name),
    case quic:start_server(ServerName, Port, Opts) of
        {ok, _Pid} ->
            %% Get actual port (may differ if Port was 0)
            case quic:get_server_port(ServerName) of
                {ok, ActualPort} ->
                    {ok, ServerName, ActualPort};
                Error ->
                    quic:stop_server(ServerName),
                    Error
            end;
        Error ->
            Error
    end.

%% @private
%% Start a standalone QUIC listener during early boot.
%% This bypasses the supervisor tree since quic_sup isn't running yet.
%% The listener is NOT linked to the distribution process to avoid
%% crashing if distribution restarts.
start_standalone_listener(Name, Port, Opts) ->
    ServerName = dist_server_name(Name),
    case quic_listener:start(Port, Opts) of
        {ok, ListenerPid} ->
            %% Get actual port
            ActualPort = quic_listener:get_port(ListenerPid),
            %% Register the listener for later adoption by quic_sup
            register_early_boot_listener(ServerName, ListenerPid, ActualPort),
            {ok, ServerName, ActualPort};
        {error, Reason} ->
            {error, Reason}
    end.

%% @private
%% Register an early boot listener so it can be found and adopted
%% when the quic application starts.
register_early_boot_listener(Name, Pid, Port) ->
    %% Store in persistent_term for cross-process access
    Key = {quic_dist_early_listener, Name},
    persistent_term:put(Key, #{pid => Pid, port => Port, name => Name}),
    ok.

%% @private
dist_server_name(Name) ->
    list_to_atom("quic_dist_" ++ atom_to_list(Name)).

%% @private
%% Get creation number (1-3) for the node.
%% Different instances should have different creation numbers.
get_creation(Name) ->
    (erlang:phash2(Name) + erlang:system_time(second)) rem 3 + 1.

%% @private
%% Load TLS credentials from files or config.
load_credentials(#quic_dist_config{cert = Cert, key = Key, cacert = CACert}) when
    Cert =/= undefined, Key =/= undefined
->
    {ok, Cert, Key, CACert};
load_credentials(#quic_dist_config{
    cert_file = CertFile,
    key_file = KeyFile,
    cacert_file = CACertFile
}) when CertFile =/= undefined, KeyFile =/= undefined ->
    try
        {ok, CertPem} = file:read_file(CertFile),
        {ok, KeyPem} = file:read_file(KeyFile),

        %% Decode PEM to DER
        [{'Certificate', CertDer, _}] = public_key:pem_decode(CertPem),

        %% Decode private key - must be fully decoded record for crypto:sign
        KeyDer =
            case public_key:pem_decode(KeyPem) of
                [{'RSAPrivateKey', Der, not_encrypted}] ->
                    public_key:der_decode('RSAPrivateKey', Der);
                [{'ECPrivateKey', Der, not_encrypted}] ->
                    public_key:der_decode('ECPrivateKey', Der);
                [{'PrivateKeyInfo', Der, not_encrypted}] ->
                    %% PKCS#8 format - decode and extract the key
                    public_key:der_decode('PrivateKeyInfo', Der);
                [{Type, Der, not_encrypted}] ->
                    %% Fallback - try to decode as the specified type
                    public_key:der_decode(Type, Der);
                [Entry] ->
                    Entry
            end,

        %% Load CA certificate if provided
        CACertDer =
            case CACertFile of
                undefined ->
                    undefined;
                _ ->
                    {ok, CACertPem} = file:read_file(CACertFile),
                    [{'Certificate', CADer, _}] = public_key:pem_decode(CACertPem),
                    CADer
            end,

        {ok, CertDer, KeyDer, CACertDer}
    catch
        _:Reason ->
            {error, {load_credentials, Reason}}
    end;
load_credentials(#quic_dist_config{} = Config) ->
    %% PSK-only configuration: either psks or psk_callback configured
    %% and no cert/key path supplied. Return undefined slots so the
    %% caller (start_quic_server / connect_to_node) builds Opts
    %% without cert/key keys.
    case psk_only_credentials_ok(Config) of
        true -> {ok, undefined, undefined, undefined};
        false -> {error, no_credentials}
    end.

%% @private
psk_only_credentials_ok(#quic_dist_config{
    psks = Psks, psk_callback = Cb
}) when Psks =/= undefined; Cb =/= undefined ->
    true;
psk_only_credentials_ok(_) ->
    false.

%% @private
%% Add psk_callback and/or psks to a listener Opts map when configured.
add_psk_listener_opts(Opts, #quic_dist_config{psks = Psks, psk_callback = Cb}) ->
    Opts1 =
        case Psks of
            undefined -> Opts;
            _ -> Opts#{psks => Psks}
        end,
    case Cb of
        undefined -> Opts1;
        _ -> Opts1#{psk_callback => resolve_psk_callback(Cb)}
    end.

%% @private
%% Add external_psk to a client Opts map when configured.
add_psk_client_opts(Opts, #quic_dist_config{external_psk = undefined}) ->
    Opts;
add_psk_client_opts(Opts, #quic_dist_config{external_psk = Ext}) ->
    Opts#{external_psk => Ext}.

%% @private
%% parse_auth_callback stores {Module, Function} or fun/3 — we need
%% to expose a fun/1 to the TLS layer. Wrap the {M, F} case.
resolve_psk_callback({Mod, Fun}) when is_atom(Mod), is_atom(Fun) ->
    fun(Identity) -> Mod:Fun(Identity) end;
resolve_psk_callback(Fn) when is_function(Fn, 1) ->
    Fn;
resolve_psk_callback(_) ->
    undefined.

%% @private
%% Handle a new incoming QUIC connection.
%%
%% Two paths:
%% - No auth_callback configured (default): start the dist controller
%%   immediately, hand it ownership of the QUIC connection, and notify
%%   the acceptor.
%% - auth_callback configured: spawn a short-lived gatekeeper process
%%   that takes ownership, waits for `{quic, Conn, {connected, _}}',
%%   runs the callback, and only on `{ok, _}' starts the dist
%%   controller. Any QUIC events the gatekeeper buffered between the
%%   `connected' event and the ownership swap are forwarded to the
%%   controller so the first stream-data event is not lost.
handle_new_connection(Conn) ->
    Config = load_config(),
    case Config#quic_dist_config.auth_callback of
        undefined ->
            handle_new_connection_direct(Conn);
        Callback ->
            handle_new_connection_with_auth(
                Conn, Callback, Config#quic_dist_config.auth_handshake_timeout
            )
    end.

%% @private
handle_new_connection_direct(Conn) ->
    case quic_dist_controller:start_link(Conn, server) of
        {ok, ControllerPid} ->
            notify_acceptor(ControllerPid),
            {ok, ControllerPid};
        Error ->
            logger:error("quic_dist: failed to start controller: ~p~n", [Error]),
            Error
    end.

%% @private
%% Spawn the gatekeeper unlinked: a crash here must not propagate to
%% the listener. It owns the connection, so if it dies the QUIC
%% connection process will clean up on owner DOWN.
handle_new_connection_with_auth(Conn, Callback, Timeout) ->
    Gatekeeper = spawn(fun() ->
        gatekeeper(Conn, Callback, Timeout)
    end),
    {ok, Gatekeeper}.

%% @private
gatekeeper(Conn, Callback, Timeout) ->
    receive
        {quic, Conn, {connected, _Info}} ->
            case run_auth_callback(Callback, Conn, server, Timeout) of
                {ok, _} ->
                    finalize_server_handoff(Conn);
                {error, Reason} ->
                    catch quic:close(Conn, normal),
                    exit({auth_failed, Reason})
            end;
        {quic, Conn, {closed, Reason}} ->
            exit({connection_closed, Reason});
        {quic, Conn, {transport_error, Code, Reason}} ->
            exit({transport_error, Code, Reason})
    after Timeout ->
        catch quic:close(Conn, normal),
        exit({auth_failed, handshake_timeout})
    end.

%% @private
%% Start the dist controller (which takes ownership of the connection),
%% drain any QUIC events that arrived in our mailbox before the
%% ownership swap and forward them to the controller, then notify the
%% acceptor and exit normally.
finalize_server_handoff(Conn) ->
    case quic_dist_controller:start_link(Conn, server) of
        {ok, ControllerPid} ->
            drain_quic_events(Conn, ControllerPid),
            notify_acceptor(ControllerPid);
        {error, Reason} ->
            catch quic:close(Conn, normal),
            exit({controller_failed, Reason})
    end.

%% @private
drain_quic_events(Conn, Target) ->
    receive
        {quic, Conn, _} = Msg ->
            Target ! Msg,
            drain_quic_events(Conn, Target)
    after 0 ->
        ok
    end.

%% @private
notify_acceptor(ControllerPid) ->
    NodeName = node(),
    ShortName =
        case NodeName of
            nonode@nohost ->
                nonode;
            _ ->
                NodeStr = atom_to_list(NodeName),
                case string:split(NodeStr, "@") of
                    [Name, _Host] -> list_to_atom(Name);
                    [Name] -> list_to_atom(Name)
                end
        end,
    ServerName = dist_server_name(ShortName),
    case persistent_term:get({quic_dist_acceptor, ServerName}, undefined) of
        undefined ->
            %% No acceptor registered yet (early boot). net_kernel will
            %% eventually call accept/1.
            ok;
        AcceptorPid when is_pid(AcceptorPid) ->
            AcceptorPid ! {accept, ControllerPid, undefined},
            ok
    end.

%% @private
%% Invoke the configured auth callback. Crashes are turned into
%% `{error, {crash, _}}' so a buggy callback cannot bring down the
%% process running it. Callers must filter out `undefined' before
%% calling.
run_auth_callback({Mod, Fun}, Conn, Side, Timeout) when is_atom(Mod), is_atom(Fun) ->
    safe_call(fun() -> Mod:Fun(Conn, Side, Timeout) end);
run_auth_callback(F, Conn, Side, Timeout) when is_function(F, 3) ->
    safe_call(fun() -> F(Conn, Side, Timeout) end).

%% @private
safe_call(Thunk) ->
    try Thunk() of
        {ok, _} = Ok -> Ok;
        {error, _} = Err -> Err;
        Other -> {error, {auth_callback_bad_return, Other}}
    catch
        Class:Reason:Stack ->
            {error, {auth_callback_crash, Class, Reason, Stack}}
    end.

%%====================================================================
%% Internal Functions - Acceptor
%%====================================================================

%% @private
%% Acceptor loop - forwards new QUIC connections to net_kernel.
%%
%% The spawn created by accept_connection/5 handshakes directly with
%% dist_util; we do not serialize controller handoff through this loop.
%% net_kernel still sends `{self(), controller, Pid}' on every accept
%% (see net_kernel.erl handle_info({accept, ...})); those messages
%% are ignored by the catch-all clause below so they don't pile up.
acceptor_loop(Kernel, #quic_dist_listener{} = Listener) ->
    receive
        {accept, DistCtrl, _NodeName} ->
            Kernel ! {accept, self(), DistCtrl, inet, quic},
            acceptor_loop(Kernel, Listener);
        stop ->
            ok;
        _Other ->
            acceptor_loop(Kernel, Listener)
    end.

%%====================================================================
%% Internal Functions - Setup Outgoing Connection
%%====================================================================

%% @private
%% Set up outgoing connection to a node.
do_setup(Kernel, Node, Type, MyNode, LongOrShortNames, SetupTime) ->
    %% Trap exits so we can handle the setup timer timeout properly
    process_flag(trap_exit, true),

    %% Ensure quic application is started
    case ensure_quic_started() of
        ok -> ok;
        {error, AppReason} -> ?shutdown2(Node, {quic_app_start_failed, AppReason})
    end,

    %% Start setup timer
    Timer = dist_util:start_timer(SetupTime),

    %% Parse target node name
    case parse_node_name(Node, LongOrShortNames) of
        {ok, Host} ->
            %% Look up node address via discovery
            case discover_node(Node, Host) of
                {ok, IP, Port} ->
                    connect_to_node(Kernel, Node, IP, Port, MyNode, Type, Timer);
                {error, Reason} ->
                    ?shutdown2(Node, {discovery_failed, Reason})
            end;
        {error, Reason} ->
            ?shutdown2(Node, Reason)
    end.

%% @private
parse_node_name(Node, LongOrShortNames) ->
    case dist_util:split_node(Node) of
        {node, Name, Host} when Name =/= "", Host =/= "" ->
            case LongOrShortNames of
                shortnames ->
                    %% Short name - host should not have dots
                    case lists:member($., Host) of
                        true -> {error, shortnames_with_fqdn};
                        false -> {ok, Host}
                    end;
                longnames ->
                    {ok, Host}
            end;
        {host, _Host} ->
            {error, invalid_node_name};
        _ ->
            {error, invalid_node_name}
    end.

%% @private
%% Discover node address using configured discovery module.
discover_node(Node, Host) ->
    Config = load_config(),
    DiscoveryModule = Config#quic_dist_config.discovery_module,

    %% First check static configuration
    case lists:keyfind(Node, 1, Config#quic_dist_config.nodes) of
        {Node, {IP, Port}} when is_tuple(IP) ->
            {ok, IP, Port};
        {Node, {IPStr, Port}} when is_list(IPStr) ->
            case inet:parse_address(IPStr) of
                {ok, IP} -> {ok, IP, Port};
                _ -> resolve_and_lookup(DiscoveryModule, Node, Host)
            end;
        false ->
            resolve_and_lookup(DiscoveryModule, Node, Host)
    end.

%% @private
resolve_and_lookup(DiscoveryModule, Node, Host) ->
    %% Try discovery module
    case code:ensure_loaded(DiscoveryModule) of
        {module, DiscoveryModule} ->
            case DiscoveryModule:lookup(Node, Host) of
                {ok, {IP, Port}} ->
                    {ok, IP, Port};
                {error, not_found} ->
                    %% Fall back to DNS resolution with default port
                    resolve_host(Host);
                Error ->
                    Error
            end;
        _ ->
            %% Discovery module not available, use DNS
            resolve_host(Host)
    end.

%% @private
resolve_host(Host) ->
    case inet:getaddr(Host, inet) of
        {ok, IP} ->
            {ok, IP, ?QUIC_DIST_DEFAULT_PORT};
        {error, _} ->
            case inet:getaddr(Host, inet6) of
                {ok, IP} ->
                    {ok, IP, ?QUIC_DIST_DEFAULT_PORT};
                Error ->
                    Error
            end
    end.

%% @private
%% Convert IP address to host string for QUIC connect.
%% Handles IP tuples, binary strings, and list strings.
ip_to_host(IP) when is_tuple(IP) ->
    inet:ntoa(IP);
ip_to_host(IP) when is_binary(IP) ->
    binary_to_list(IP);
ip_to_host(IP) when is_list(IP) ->
    IP.

%% @doc Register per-node connect-time option overrides for the next
%% `setup/5' attempt against `Node'. The map is merged on top of the
%% defaults that `connect_to_node/7' builds, so callers can override
%% any key, including `socket_backend' and `socket_adapter' to route
%% the underlying UDP packets through a custom transport (for example
%% a MASQUE CONNECT-UDP tunnel), or `external_psk' to authenticate
%% this peer with a different identity/secret than the cluster-wide
%% default. See docs/PSK.md for the PSK option shapes.
%%
%% The entry is consumed once (the first matching `connect_to_node'
%% call clears it). Use `clear_connect_options/1' to drop it without
%% triggering a connect.
-spec set_connect_options(node(), map()) -> ok.
set_connect_options(Node, Opts) when is_atom(Node), is_map(Opts) ->
    ensure_connect_opts_table(),
    true = ets:insert(quic_dist_connect_opts, {Node, Opts}),
    ok.

%% @doc Look up the pending connect-option overrides for `Node' without
%% consuming them. Returns an empty map if none are registered.
-spec get_connect_options(node()) -> map().
get_connect_options(Node) when is_atom(Node) ->
    case ets:info(quic_dist_connect_opts) of
        undefined ->
            #{};
        _ ->
            case ets:lookup(quic_dist_connect_opts, Node) of
                [{Node, Opts}] -> Opts;
                [] -> #{}
            end
    end.

%% @doc Drop any pending connect-option overrides for `Node'.
-spec clear_connect_options(node()) -> ok.
clear_connect_options(Node) when is_atom(Node) ->
    case ets:info(quic_dist_connect_opts) of
        undefined ->
            ok;
        _ ->
            _ = ets:delete(quic_dist_connect_opts, Node),
            ok
    end.

%% @private
%% Atomically read and remove the override for `Node'. Used by the
%% setup path so a registered override applies once.
take_connect_options(Node) ->
    case ets:info(quic_dist_connect_opts) of
        undefined ->
            #{};
        _ ->
            case ets:take(quic_dist_connect_opts, Node) of
                [{Node, Opts}] -> Opts;
                [] -> #{}
            end
    end.

%% @private
%% Create the per-node connect-options table on first use.
ensure_connect_opts_table() ->
    ensure_named_set(quic_dist_connect_opts).

%% @private
%% Connect to the target node.
connect_to_node(Kernel, Node, IP, Port, MyNode, Type, Timer) ->
    Config = load_config(),

    %% Prepare QUIC connection options
    case load_credentials(Config) of
        {ok, Cert, Key, _CACert} ->
            CongestionThreshold = Config#quic_dist_config.congestion_threshold,
            BaseOpts0 = #{
                alpn => [?QUIC_DIST_ALPN],
                idle_timeout => ?QUIC_DIST_IDLE_TIMEOUT,
                %% QUIC-level keep-alive via PING frames (bypasses flow control).
                %% Paced off net_ticktime so net_kernel never sees a stale
                %% connection under load.
                keep_alive_interval => keep_alive_interval(),
                %% Use aggressive initial cwnd for distribution bulk transfers
                initial_window => ?INITIAL_WINDOW_AGGRESSIVE,
                %% Keep a higher congestion floor to avoid liveness stalls
                %% on bursty virtual networks (e.g., Docker bridge).
                minimum_window => ?MINIMUM_WINDOW_DISTRIBUTION,
                %% Higher flow control limits for distribution to avoid blocking
                %% during large message transfers (code loading, large terms)
                max_data => ?DIST_INITIAL_MAX_DATA,
                max_stream_data_bidi_local => ?DIST_INITIAL_MAX_STREAM_DATA,
                max_stream_data_bidi_remote => ?DIST_INITIAL_MAX_STREAM_DATA,
                max_stream_data_uni => ?DIST_INITIAL_MAX_STREAM_DATA,
                %% Backpressure threshold for congestion detection
                congestion_threshold => CongestionThreshold,
                %% Pacing spreads packet sends to avoid bursts
                pacing_enabled => Config#quic_dist_config.pacing_enabled,
                %% Longer recovery duration for virtual network packet reordering
                min_recovery_duration => ?MIN_RECOVERY_DURATION_DISTRIBUTION,
                %% Use known-safe MTU for LAN (1452 bytes, IPv4/IPv6 compatible)
                %% instead of PMTU probing which adds overhead
                max_udp_payload_size => ?DIST_MAX_UDP_PAYLOAD_SIZE,
                pmtu_enabled => false,
                % TODO: Enable proper verification
                verify => false
            },
            %% Add cert/key when present (PSK-only clients omit them).
            BaseOpts1 =
                case {Cert, Key} of
                    {undefined, undefined} -> BaseOpts0;
                    {_, _} -> BaseOpts0#{cert => Cert, key => Key}
                end,
            %% Add external_psk when configured.
            BaseOpts = add_psk_client_opts(BaseOpts1, Config),

            %% Per-node overrides registered via set_connect_options/2.
            %% Merged on top of the defaults so callers can swap the
            %% socket backend, adjust flow control, etc.
            Overrides = take_connect_options(Node),
            Opts = maps:merge(BaseOpts, Overrides),

            %% Convert IP to host format expected by QUIC
            Host = ip_to_host(IP),

            %% Attempt connection
            case quic:connect(Host, Port, Opts, self()) of
                {ok, Conn} ->
                    %% Wait for connection to be established
                    wait_for_connection(Kernel, Node, Conn, MyNode, Type, Timer, Config);
                {error, Reason} ->
                    ?shutdown2(Node, {connect_failed, Reason})
            end;
        {error, Reason} ->
            ?shutdown2(Node, {credentials, Reason})
    end.

%% @private
wait_for_connection(Kernel, Node, Conn, MyNode, Type, Timer, Config) ->
    receive
        {quic, Conn, {connected, _Info}} ->
            %% Optional auth handshake before the dist controller
            %% takes ownership.
            case maybe_run_client_auth(Conn, Config) of
                ok ->
                    start_client_controller(Kernel, Node, Conn, MyNode, Type, Timer);
                {error, AuthReason} ->
                    catch quic:close(Conn, normal),
                    ?shutdown2(Node, {auth_failed, AuthReason})
            end;
        {quic, Conn, {closed, Reason}} ->
            ?shutdown2(Node, {closed, Reason});
        {quic, Conn, {transport_error, Code, Reason}} ->
            ?shutdown2(Node, {transport_error, Code, Reason});
        {'EXIT', Timer, setup_timer_timeout} ->
            quic:close(Conn, timeout),
            ?shutdown2(Node, connect_timeout)
    end.

%% @private
maybe_run_client_auth(_Conn, #quic_dist_config{auth_callback = undefined}) ->
    ok;
maybe_run_client_auth(Conn, #quic_dist_config{
    auth_callback = Callback,
    auth_handshake_timeout = Timeout
}) ->
    case run_auth_callback(Callback, Conn, client, Timeout) of
        {ok, _} -> ok;
        {error, _} = Err -> Err
    end.

%% @private
start_client_controller(Kernel, Node, Conn, MyNode, Type, Timer) ->
    case quic_dist_controller:start_link(Conn, client) of
        {ok, DistCtrl} ->
            quic_dist_controller:set_supervisor(DistCtrl, Kernel),
            quic_dist_controller:set_node(DistCtrl, Node),
            HSData = create_hs_data_setup(Kernel, DistCtrl, Node, MyNode, Type, Timer),
            dist_util:handshake_we_started(HSData);
        {error, Reason} ->
            quic:close(Conn, normal),
            ?shutdown2(Node, {controller_failed, Reason})
    end.

%%====================================================================
%% Internal Functions - Handshake Data
%%====================================================================

%% @private
%% Create handshake data structure for accepted connections.
create_hs_data(DistCtrl, MyNode, Timer, Allowed, Kernel) ->
    %% Capture SetupPid (self) for dist_ctrlr message
    SetupPid = self(),
    logger:info(
        "create_hs_data: Kernel=~p, DistCtrl=~p, SetupPid=~p~n",
        [Kernel, DistCtrl, SetupPid]
    ),
    #hs_data{
        kernel_pid = Kernel,
        other_node = undefined,
        this_node = MyNode,
        socket = DistCtrl,
        timer = Timer,
        this_flags = 0,
        other_flags = 0,
        %% Reject the connection-wide atom cache so it cannot create
        %% cross-stream decoder-state dependencies. Fragments stay on
        %% (routed per SeqId by quic_dist_dispatch). strict_order_flags/0
        %% returns DFLAG_DIST_HDR_ATOM_CACHE (0x2000).
        reject_flags = dist_util:strict_order_flags(),
        f_send = fun(Ctrl, Data) -> quic_dist_controller:send(Ctrl, Data) end,
        f_recv = fun(Ctrl, Len, Timeout) ->
            %% Receive data and try to extract node name if this is the name message
            Result = quic_dist_controller:recv(Ctrl, Len, Timeout),
            case Result of
                {ok, Data} ->
                    %% Try to parse name message and store node in controller
                    maybe_extract_node(Data, Ctrl),
                    Result;
                _ ->
                    Result
            end
        end,
        f_setopts_pre_nodeup = fun(Ctrl) ->
            %% Just log and return ok - inet_tcp_dist doesn't do anything special here
            StoredNode = get_stored_node(Ctrl),
            logger:info(
                "f_setopts_pre_nodeup (accept): Ctrl=~p, Node=~p, SetupPid=~p, linked=~p~n",
                [
                    Ctrl,
                    StoredNode,
                    SetupPid,
                    lists:member(Ctrl, element(2, process_info(self(), links)))
                ]
            ),
            ok
        end,
        f_setopts_post_nodeup = fun(_Ctrl) -> ok end,
        f_getll = fun(Ctrl) -> {ok, Ctrl} end,
        f_address = fun(Ctrl, Node) ->
            quic_dist_controller:get_address(Ctrl, Node)
        end,
        mf_tick = fun(Ctrl) -> quic_dist_controller:tick(Ctrl) end,
        mf_getstat = fun(Ctrl) -> quic_dist_controller:getstat(Ctrl) end,
        request_type = normal,
        mf_setopts = fun(_Ctrl, _Opts) -> ok end,
        mf_getopts = fun(_Ctrl, Opts) -> {ok, [{O, 0} || O <- Opts]} end,
        allowed = Allowed,
        f_handshake_complete = fun(Ctrl, HsNode, DHandle) ->
            logger:info(
                "f_handshake_complete (accept): Ctrl=~p, Node=~p, DHandle=~p~n",
                [Ctrl, HsNode, DHandle]
            ),
            %% Notify controller that handshake is complete
            %% Pass DHandle so controller can use dist_ctrl_* functions
            Ctrl ! {handshake_complete, HsNode, DHandle},
            ok
        end
    }.

%% @private
%% Try to extract node name from name message and store in controller.
%% The name message format depends on protocol version:
%% Protocol 6: <<$N, Flags:64/big, Creation:32/big, NameLen:16/big, Name/binary>>
%% Older: <<$n, Version:16/big, Flags:32/big, Name/binary>>
maybe_extract_node([H | Rest], Ctrl) when H =:= $N; H =:= $n ->
    try
        case H of
            $N ->
                %% Protocol version 6 format
                RestBin = list_to_binary(Rest),
                <<_Flags:64/big, _Creation:32/big, NameLen:16/big, NameBin:NameLen/binary,
                    _/binary>> = RestBin,
                Node = binary_to_atom(NameBin, utf8),
                quic_dist_controller:set_node(Ctrl, Node);
            $n ->
                %% Older protocol format
                RestBin = list_to_binary(Rest),
                <<_Version:16/big, _Flags:32/big, NameBin/binary>> = RestBin,
                Node = binary_to_atom(NameBin, utf8),
                quic_dist_controller:set_node(Ctrl, Node)
        end
    catch
        _:_ ->
            %% Failed to parse, not a name message or malformed
            ok
    end;
maybe_extract_node(_, _) ->
    ok.

%% @private
%% Get the stored node from controller, with fallback.
get_stored_node(Ctrl) ->
    case quic_dist_controller:get_node(Ctrl) of
        {ok, Node} -> Node;
        undefined -> undefined
    end.

%% @private
%% Create handshake data structure for outgoing connections.
create_hs_data_setup(Kernel, DistCtrl, Node, MyNode, Type, Timer) ->
    %% Capture SetupPid (self) for dist_ctrlr message
    SetupPid = self(),
    logger:info(
        "create_hs_data_setup: Kernel=~p, DistCtrl=~p, Node=~p, SetupPid=~p~n",
        [Kernel, DistCtrl, Node, SetupPid]
    ),
    #hs_data{
        kernel_pid = Kernel,
        other_node = Node,
        this_node = MyNode,
        socket = DistCtrl,
        timer = Timer,
        this_flags = 0,
        other_flags = 0,
        %% Reject the connection-wide atom cache — see create_hs_data/5 comment.
        reject_flags = dist_util:strict_order_flags(),
        f_send = fun(Ctrl, Data) -> quic_dist_controller:send(Ctrl, Data) end,
        f_recv = fun(Ctrl, Len, Timeout) -> quic_dist_controller:recv(Ctrl, Len, Timeout) end,
        f_setopts_pre_nodeup = fun(Ctrl) ->
            %% Just log and return ok - inet_tcp_dist doesn't do anything special here
            logger:info(
                "f_setopts_pre_nodeup (setup): Ctrl=~p, Node=~p, SetupPid=~p, linked=~p~n",
                [Ctrl, Node, SetupPid, lists:member(Ctrl, element(2, process_info(self(), links)))]
            ),
            ok
        end,
        f_setopts_post_nodeup = fun(_Ctrl) -> ok end,
        f_getll = fun(Ctrl) -> {ok, Ctrl} end,
        f_address = fun(Ctrl, N) ->
            quic_dist_controller:get_address(Ctrl, N)
        end,
        mf_tick = fun(Ctrl) -> quic_dist_controller:tick(Ctrl) end,
        mf_getstat = fun(Ctrl) -> quic_dist_controller:getstat(Ctrl) end,
        request_type = Type,
        mf_setopts = fun(_Ctrl, _Opts) -> ok end,
        mf_getopts = fun(_Ctrl, Opts) -> {ok, [{O, 0} || O <- Opts]} end,
        f_handshake_complete = fun(Ctrl, HsNode, DHandle) ->
            logger:info(
                "f_handshake_complete (setup): Ctrl=~p, Node=~p, DHandle=~p~n",
                [Ctrl, HsNode, DHandle]
            ),
            %% Notify controller that handshake is complete
            %% Pass DHandle so controller can use dist_ctrl_* functions
            Ctrl ! {handshake_complete, HsNode, DHandle},
            ok
        end
    }.

%%====================================================================
%% User Stream API
%%====================================================================

%% @doc Open a bidirectional user stream to a connected node.
%% Returns {ok, StreamRef} on success where StreamRef can be used with send/2,3 and close_stream/1.
%% The caller becomes the stream owner.
-spec open_stream(Node :: node()) -> {ok, stream_ref()} | {error, term()}.
open_stream(Node) ->
    open_stream(Node, []).

%% @doc Open a bidirectional user stream with options.
%% Options:
%%   {priority, 16..255} - Stream priority (default: 128, lower = higher priority)
%%                         Note: priorities 0-15 are reserved for distribution
-spec open_stream(Node :: node(), Options :: [stream_opt()]) ->
    {ok, stream_ref()} | {error, term()}.
open_stream(Node, Options) ->
    case get_controller(Node) of
        {ok, Ctrl} ->
            case quic_dist_controller:open_user_stream(Ctrl, self(), Options) of
                {ok, StreamId} ->
                    {ok, {quic_dist_stream, Node, StreamId}};
                {error, Reason} ->
                    {error, Reason}
            end;
        {error, Reason} ->
            {error, Reason}
    end.

%% @doc Send data on a user stream.
%% Equivalent to send(StreamRef, Data, false).
-spec send(StreamRef :: stream_ref(), Data :: iodata()) -> ok | {error, term()}.
send(StreamRef, Data) ->
    send(StreamRef, Data, false).

%% @doc Send data on a user stream.
%% When Fin is true, this marks the end of data on this stream (half-close).
-spec send(StreamRef :: stream_ref(), Data :: iodata(), Fin :: boolean()) -> ok | {error, term()}.
send({quic_dist_stream, Node, StreamId}, Data, Fin) ->
    case get_controller(Node) of
        {ok, Ctrl} ->
            quic_dist_controller:send_user_data(Ctrl, StreamId, Data, Fin);
        {error, Reason} ->
            {error, Reason}
    end.

%% @doc Close a user stream gracefully.
%% This sends a FIN to the peer. When both sides have sent FIN, the owner
%% receives {quic_dist_stream, StreamRef, closed}.
-spec close_stream(StreamRef :: stream_ref()) -> ok | {error, term()}.
close_stream({quic_dist_stream, Node, StreamId}) ->
    case get_controller(Node) of
        {ok, Ctrl} ->
            quic_dist_controller:close_user_stream(Ctrl, StreamId);
        {error, Reason} ->
            {error, Reason}
    end.

%% @doc Reset/cancel a user stream immediately (notifies peer).
%% Uses default error code 0.
-spec reset_stream(StreamRef :: stream_ref()) -> ok | {error, term()}.
reset_stream(StreamRef) ->
    reset_stream(StreamRef, 0).

%% @doc Reset/cancel a user stream with a specific error code.
%% The peer receives the reset notification immediately.
-spec reset_stream(StreamRef :: stream_ref(), ErrorCode :: non_neg_integer()) ->
    ok | {error, term()}.
reset_stream({quic_dist_stream, Node, StreamId}, ErrorCode) ->
    case get_controller(Node) of
        {ok, Ctrl} ->
            quic_dist_controller:reset_user_stream(Ctrl, StreamId, ErrorCode);
        {error, Reason} ->
            {error, Reason}
    end.

%% @doc Register to accept incoming user streams from a node.
%% Joins the acceptor pool for the node. Multiple processes can register as acceptors.
%% Incoming streams are assigned to acceptors using round-robin selection.
%%
%% When a new stream arrives, one acceptor receives:
%%   {quic_dist_stream, StreamRef, {data, Data, Fin}}
%%
%% The acceptor automatically becomes the stream owner (implicit ownership).
%% Use controlling_process/2 to transfer ownership to a worker process.
%%
%% If no acceptors are registered, incoming streams are refused with RESET.
-spec accept_streams(Node :: node()) -> ok | {error, term()}.
accept_streams(Node) ->
    case get_controller(Node) of
        {ok, Ctrl} ->
            quic_dist_controller:accept_user_streams(Ctrl, self());
        {error, Reason} ->
            {error, Reason}
    end.

%% @doc Stop accepting incoming user streams from a node.
%% Removes the calling process from the acceptor pool.
-spec stop_accepting(Node :: node()) -> ok | {error, term()}.
stop_accepting(Node) ->
    case get_controller(Node) of
        {ok, Ctrl} ->
            quic_dist_controller:stop_accepting_streams(Ctrl);
        {error, Reason} ->
            {error, Reason}
    end.

%% @doc Transfer stream ownership to another process.
%% The new owner will receive all subsequent messages for this stream.
-spec controlling_process(StreamRef :: stream_ref(), NewOwner :: pid()) -> ok | {error, term()}.
controlling_process({quic_dist_stream, Node, StreamId}, NewOwner) ->
    case get_controller(Node) of
        {ok, Ctrl} ->
            quic_dist_controller:controlling_process(Ctrl, StreamId, NewOwner);
        {error, Reason} ->
            {error, Reason}
    end.

%% @doc List all user streams across all connected nodes.
-spec list_streams() -> [stream_info()].
list_streams() ->
    try
        DistCtrls = erlang:system_info(dist_ctrl),
        lists:flatmap(
            fun
                ({_Node, Ctrl}) when is_pid(Ctrl) ->
                    quic_dist_controller:list_user_streams(Ctrl);
                ({_Node, _Port}) ->
                    []
            end,
            DistCtrls
        )
    catch
        _:_ ->
            []
    end.

%% @doc List user streams for a specific connected node.
-spec list_streams(Node :: node()) -> [stream_info()].
list_streams(Node) ->
    case get_controller(Node) of
        {ok, Ctrl} ->
            quic_dist_controller:list_user_streams(Ctrl);
        {error, _} ->
            []
    end.

%% @doc Get the distribution controller for a connected node.
%% Returns {ok, ControllerPid} if the node is connected, {error, not_connected} otherwise.
-spec get_controller(Node :: node()) -> {ok, pid()} | {error, not_connected | not_quic_connection}.
get_controller(Node) ->
    %% Use erlang:system_info(dist_ctrl) to get the list of distribution controllers
    %% This returns [{Node, CtrlPid}] for all connected nodes
    try
        DistCtrls = erlang:system_info(dist_ctrl),
        case lists:keyfind(Node, 1, DistCtrls) of
            {Node, Ctrl} when is_pid(Ctrl) ->
                {ok, Ctrl};
            {Node, Port} when is_port(Port) ->
                %% TCP distribution uses ports, not pids - not supported
                {error, not_quic_connection};
            _ ->
                {error, not_connected}
        end
    catch
        _:_ ->
            {error, not_connected}
    end.