%%%------------------------------------------------------------------- %%% @doc %%% Peer Connector - Establishes direct QUIC connections to remote peers (v0.8.0+). %%% %%% This module enables peer-to-peer communication by establishing outbound %%% QUIC connections to arbitrary peers. Used by DHT for STORE/FIND_VALUE %%% message propagation and by RPC/PubSub for direct delivery. %%% %%% == Overview == %%% %%% Pattern: Connection-pooled utility module with NAT-aware fallback %%% - Uses `macula_peer_connection_pool' for connection reuse %%% - Falls back to NAT-aware routing (hole punch, relay) on failure %%% - Fire-and-forget message sending %%% %%% == Connection Strategy (v0.12.0+) == %%% %%% 1. Try pooled connection (fastest, cached) %%% 2. Try direct QUIC connection (new connection) %%% 3. Fall back to NAT-aware routing via `macula_nat_connector': %%% a. Direct connection (if NAT allows) %%% b. Hole punch (coordinated NAT traversal) %%% c. Relay via gateway (guaranteed fallback) %%% %%% == Usage == %%% %%% Used internally by: %%% - `macula_pubsub_dht': Direct pub/sub delivery to discovered subscribers %%% - `macula_service_registry': DHT STORE propagation to k=20 nodes %%% - Future: Multi-hop RPC routing %%% %%% ``` %%% %% Send a DHT STORE message to a peer %%% Endpoint = <<"192.168.1.100:9443">>, %%% Message = #{ %%% key => <<"service.calculator.add">>, %%% value => <<"192.168.1.50:9443">>, %%% ttl => 300 %%% }, %%% ok = macula_peer_connector:send_message(Endpoint, dht_store, Message). %%% ''' %%% %%% == Performance Characteristics == %%% %%% v0.8.0: Fire-and-forget pattern (now legacy fallback) %%% - Creates new connection per message %%% - Simple but inefficient for high-frequency messaging %%% %%% v0.10.0: Connection pooling (current) %%% - Reuses existing connections via macula_peer_connection_pool %%% - 1.5-2x latency improvement for repeated messaging %%% %%% v0.12.0: NAT-aware routing (current) %%% - Automatic fallback to hole punch and relay %%% - Works across all NAT types %%% %%% @end %%%------------------------------------------------------------------- -module(macula_peer_connector). -include_lib("kernel/include/logger.hrl"). -include_lib("quicer/include/quicer.hrl"). %% API -export([ send_message/3, send_message/4, send_message_and_wait/4, % For request-response patterns like NAT_PROBE send_message_nat_aware/4, send_message_nat_aware/5 ]). %%%=================================================================== %%% API %%%=================================================================== %% @doc Send a message to a remote peer (fire-and-forget). %% Uses connection pool for efficiency, falls back to direct connection. -spec send_message(binary(), atom(), map()) -> ok | {error, term()}. send_message(Endpoint, MessageType, Message) -> send_message(Endpoint, MessageType, Message, 5000). %% @doc Send a message to a remote peer with custom timeout. -spec send_message(binary(), atom(), map(), timeout()) -> ok | {error, term()}. send_message(Endpoint, MessageType, Message, _Timeout) -> %% Encode message MessageBinary = macula_protocol_encoder:encode(MessageType, Message), %% Try to use connection pool first PoolPid = whereis(macula_peer_connection_pool), send_via_connection(PoolPid, Endpoint, MessageBinary). %% @doc Send a message and wait for a reply (request-response pattern). %% This is used for messages like NAT_PROBE that expect a reply. %% Returns {ok, ReplyMessage} on success, {error, timeout} if no reply received. -spec send_message_and_wait(binary(), atom(), map(), timeout()) -> {ok, {atom(), map()}} | {error, term()}. send_message_and_wait(Endpoint, MessageType, Message, Timeout) -> %% Encode message MessageBinary = macula_protocol_encoder:encode(MessageType, Message), %% Parse endpoint to get host and port case parse_endpoint(Endpoint) of {ok, Host, Port} -> send_and_wait_quic(Host, Port, MessageBinary, Timeout); {error, Reason} -> {error, {invalid_endpoint, Reason}} end. %% @private Send message and wait for reply using direct QUIC connection. send_and_wait_quic(Host, Port, MessageBinary, Timeout) -> ?LOG_INFO("[PeerConnector] send_and_wait to ~s:~p (~p bytes)", [Host, Port, byte_size(MessageBinary)]), ConnectOpts = [ {alpn, ["macula"]}, {verify, none}, {idle_timeout_ms, 60000}, {keep_alive_interval_ms, 20000}, {handshake_idle_timeout_ms, 30000} ], case macula_quic:connect(Host, Port, ConnectOpts, Timeout) of {ok, Conn} -> send_and_wait_stream(Conn, MessageBinary, Timeout); {error, transport_down, _Details} -> {error, {connect_failed, transport_down}}; {error, Reason} -> {error, {connect_failed, Reason}} end. %% @private Open stream, send message, wait for reply. send_and_wait_stream(Conn, MessageBinary, Timeout) -> case macula_quic:open_stream(Conn) of {ok, Stream} -> %% Set stream to active mode to receive reply case quicer:setopt(Stream, active, true) of ok -> send_and_wait_reply(Conn, Stream, MessageBinary, Timeout); {error, Reason} -> quicer:async_shutdown_connection(Conn, ?QUIC_CONNECTION_SHUTDOWN_FLAG_NONE, 0), {error, {setopt_failed, Reason}} end; {error, Reason} -> quicer:async_shutdown_connection(Conn, ?QUIC_CONNECTION_SHUTDOWN_FLAG_NONE, 0), {error, {stream_failed, Reason}} end. %% @private Send message and wait for reply data. send_and_wait_reply(Conn, Stream, MessageBinary, Timeout) -> case macula_quic:send(Stream, MessageBinary) of ok -> %% Wait for reply with loop to ignore control messages wait_for_data_reply(Conn, Stream, Timeout, erlang:monotonic_time(millisecond)); {error, Reason} -> quicer:async_shutdown_stream(Stream, ?QUIC_STREAM_SHUTDOWN_FLAG_ABORT, 0), quicer:async_shutdown_connection(Conn, ?QUIC_CONNECTION_SHUTDOWN_FLAG_NONE, 0), {error, {send_failed, Reason}} end. %% @private Wait for binary data reply, ignoring QUIC control messages and stale stream data. %% NOTE: Data may arrive on OTHER streams (from previous/concurrent connections). %% We ignore data on other streams and keep waiting for data on OUR stream. wait_for_data_reply(Conn, Stream, Timeout, StartTime) -> RemainingTime = max(0, Timeout - (erlang:monotonic_time(millisecond) - StartTime)), receive {quic, Data, Stream, _Flags} when is_binary(Data) -> ?LOG_INFO("[PeerConnector] Received reply: ~p bytes", [byte_size(Data)]), ReplyResult = macula_protocol_decoder:decode(Data), ?LOG_INFO("[PeerConnector] Decoded reply: ~p", [ReplyResult]), quicer:async_shutdown_stream(Stream, ?QUIC_STREAM_SHUTDOWN_FLAG_GRACEFUL, 0), quicer:async_shutdown_connection(Conn, ?QUIC_CONNECTION_SHUTDOWN_FLAG_NONE, 0), ReplyResult; {quic, Data, OtherStream, _Flags} when is_binary(Data) -> %% Data on different stream - ignore and keep waiting for OUR stream %% This happens with concurrent connections or stale streams ?LOG_DEBUG("[PeerConnector] Ignoring data on other stream ~p (waiting for ~p)", [OtherStream, Stream]), wait_for_data_reply(Conn, Stream, Timeout, StartTime); %% Ignore QUIC control messages and continue waiting {quic, streams_available, _ConnRef, _Info} -> wait_for_data_reply(Conn, Stream, Timeout, StartTime); {quic, peer_needs_streams, _ConnRef, _Info} -> wait_for_data_reply(Conn, Stream, Timeout, StartTime); {quic, shutdown, Stream, _Reason} -> %% Only treat shutdown of OUR stream as error {error, connection_shutdown}; {quic, shutdown, _OtherHandle, _Reason} -> %% Shutdown of other streams - ignore wait_for_data_reply(Conn, Stream, Timeout, StartTime); {quic, closed, Stream, _Reason} -> %% Only treat closure of OUR stream as error {error, connection_closed}; {quic, closed, _OtherHandle, _Reason} -> %% Closure of other streams - ignore wait_for_data_reply(Conn, Stream, Timeout, StartTime) after RemainingTime -> quicer:async_shutdown_stream(Stream, ?QUIC_STREAM_SHUTDOWN_FLAG_ABORT, 0), quicer:async_shutdown_connection(Conn, ?QUIC_CONNECTION_SHUTDOWN_FLAG_NONE, 0), {error, timeout} end. %% @doc Send a message using NAT-aware routing (hole punch, relay fallback). %% Use this when sending to peers that may be behind NAT. %% LocalNodeId is required for hole punch coordination. -spec send_message_nat_aware(binary(), binary(), atom(), map()) -> ok | {error, term()}. send_message_nat_aware(LocalNodeId, TargetNodeId, MessageType, Message) -> send_message_nat_aware(LocalNodeId, TargetNodeId, MessageType, Message, #{}). %% @doc Send a message using NAT-aware routing with options. %% Options: %% - endpoint: Target endpoint (if known, skips DHT lookup) %% - relay_endpoint: Specific relay to use %% - skip_hole_punch: true to skip hole punch attempts -spec send_message_nat_aware(binary(), binary(), atom(), map(), map()) -> ok | {error, term()}. send_message_nat_aware(LocalNodeId, TargetNodeId, MessageType, Message, Opts) -> MessageBinary = macula_protocol_encoder:encode(MessageType, Message), %% Try endpoint from options first, then fall back to NAT-aware connection case maps:get(endpoint, Opts, undefined) of undefined -> send_via_nat_connector(LocalNodeId, TargetNodeId, MessageBinary, Opts); Endpoint -> %% Try direct first, then NAT-aware case send_message(Endpoint, MessageType, Message) of ok -> ok; {error, _Reason} -> send_via_nat_connector(LocalNodeId, TargetNodeId, MessageBinary, Opts) end end. %% @private Pool not running - fall back to direct connection send_via_connection(undefined, Endpoint, MessageBinary) -> send_via_direct_connection(Endpoint, MessageBinary); %% @private Pool available - use pooled connection send_via_connection(_Pid, Endpoint, MessageBinary) -> send_via_pool(Endpoint, MessageBinary). %%%=================================================================== %%% Internal Functions %%%=================================================================== %% @private %% @doc Send message using connection pool (preferred, 1.5-2x faster). send_via_pool(Endpoint, MessageBinary) -> ConnResult = macula_peer_connection_pool:get_connection(Endpoint), do_pool_send(ConnResult, Endpoint, MessageBinary). %% @private Pool connection with no stream (seeded from macula_connection) — %% open a fresh stream on the existing QUIC connection, send, close stream. do_pool_send({ok, Conn, undefined}, Endpoint, MessageBinary) -> ?LOG_DEBUG("[PeerConnector] Pool hit (seeded conn, no stream) for ~s, opening temp stream", [Endpoint]), do_pool_send_temp_stream(macula_quic:open_stream(Conn), Conn, Endpoint, MessageBinary); %% @private Pool connection acquired - attempt send do_pool_send({ok, Conn, Stream}, Endpoint, MessageBinary) -> SendResult = macula_quic:send(Stream, MessageBinary), handle_pool_send_result(SendResult, Conn, Stream, Endpoint, MessageBinary); %% @private Pool connection failed - fall back to direct do_pool_send({error, Reason}, Endpoint, MessageBinary) -> ?LOG_DEBUG("Pool connection failed: ~p, using direct", [Reason]), send_via_direct_connection(Endpoint, MessageBinary). %% @private Temp stream opened on seeded connection — send and close do_pool_send_temp_stream({ok, TempStream}, _Conn, _Endpoint, MessageBinary) -> SendResult = macula_quic:send(TempStream, MessageBinary), %% Brief delay then close the temp stream (don't interfere with main stream) timer:sleep(50), catch quicer:async_shutdown_stream(TempStream, ?QUIC_STREAM_SHUTDOWN_FLAG_GRACEFUL, 0), case SendResult of ok -> ok; {error, Reason} -> ?LOG_WARNING("[PeerConnector] Temp stream send failed: ~p", [Reason]), {error, {send_failed, Reason}} end; %% @private Temp stream open failed — fall back to direct do_pool_send_temp_stream({error, Reason}, _Conn, Endpoint, MessageBinary) -> ?LOG_WARNING("[PeerConnector] Failed to open temp stream on seeded conn: ~p, using direct", [Reason]), send_via_direct_connection(Endpoint, MessageBinary). %% @private Send succeeded - return connection to pool handle_pool_send_result(ok, Conn, Stream, Endpoint, _MessageBinary) -> macula_peer_connection_pool:return_connection(Endpoint, {Conn, Stream}), ok; %% @private Send failed - invalidate and retry with direct handle_pool_send_result({error, Reason}, _Conn, _Stream, Endpoint, MessageBinary) -> macula_peer_connection_pool:invalidate(Endpoint), ?LOG_WARNING("Pool send failed: ~p, falling back to direct", [Reason]), send_via_direct_connection(Endpoint, MessageBinary). %% @private %% @doc Send message via direct connection (fallback, creates new connection). send_via_direct_connection(Endpoint, MessageBinary) -> ParseResult = parse_endpoint(Endpoint), do_direct_send(ParseResult, Endpoint, MessageBinary). %% @private Endpoint parsed successfully do_direct_send({ok, Host, Port}, _Endpoint, MessageBinary) -> send_via_quic(Host, Port, MessageBinary, 5000); %% @private Invalid endpoint format do_direct_send({error, Reason}, Endpoint, _MessageBinary) -> ?LOG_ERROR("Invalid endpoint ~p: ~p", [Endpoint, Reason]), {error, {invalid_endpoint, Reason}}. %% @private %% @doc Parse endpoint string into host and port. %% Supports formats: "host:port", "https://host:port", "http://host:port" parse_endpoint(Endpoint) when is_binary(Endpoint) -> parse_endpoint(binary_to_list(Endpoint)); parse_endpoint(Endpoint) when is_list(Endpoint) -> %% Strip protocol prefix if present Stripped = strip_protocol(Endpoint), %% Split from the end to handle IPv6 addresses correctly %% For "host:port", splits into ["host", "port"] case string:split(Stripped, ":", trailing) of [Host, PortStr] -> parse_port(Host, PortStr); _ -> {error, invalid_format} end. %% @private Strip protocol prefix (https://, http://) from endpoint strip_protocol("https://" ++ Rest) -> Rest; strip_protocol("http://" ++ Rest) -> Rest; strip_protocol(Endpoint) -> Endpoint. %% @private Parse port string to integer parse_port(Host, PortStr) -> parse_port_result(Host, catch list_to_integer(PortStr)). %% @private Port parsed successfully parse_port_result(Host, Port) when is_integer(Port), Port > 0, Port < 65536 -> {ok, Host, Port}; %% @private Invalid port value or parse error parse_port_result(_Host, _) -> {error, invalid_port}. %% @private %% @doc Send message via direct QUIC connection (legacy fallback). send_via_quic(Host, Port, MessageBinary, Timeout) -> ConnectOpts = [ {alpn, ["macula"]}, {verify, none}, {idle_timeout_ms, 60000}, {keep_alive_interval_ms, 20000}, {handshake_idle_timeout_ms, 30000} ], ConnResult = macula_quic:connect(Host, Port, ConnectOpts, Timeout), do_quic_connect(ConnResult, MessageBinary). %% @private Connection established - open stream do_quic_connect({ok, Conn}, MessageBinary) -> StreamResult = macula_quic:open_stream(Conn), do_quic_stream(StreamResult, Conn, MessageBinary); %% @private Transport down do_quic_connect({error, transport_down, _Details}, _MessageBinary) -> {error, {connect_failed, transport_down}}; %% @private Connection failed do_quic_connect({error, Reason}, _MessageBinary) -> {error, {connect_failed, Reason}}. %% @private Stream opened - send message do_quic_stream({ok, Stream}, Conn, MessageBinary) -> SendResult = macula_quic:send(Stream, MessageBinary), do_quic_send(SendResult, Conn, Stream); %% @private Stream open failed do_quic_stream({error, Reason}, Conn, _MessageBinary) -> quicer:async_shutdown_connection(Conn, ?QUIC_CONNECTION_SHUTDOWN_FLAG_NONE, 0), {error, {stream_failed, Reason}}. %% @private Send succeeded - graceful shutdown do_quic_send(ok, Conn, Stream) -> timer:sleep(50), quicer:async_shutdown_stream(Stream, ?QUIC_STREAM_SHUTDOWN_FLAG_GRACEFUL, 0), quicer:async_shutdown_connection(Conn, ?QUIC_CONNECTION_SHUTDOWN_FLAG_NONE, 0), ok; %% @private Send failed - abort stream do_quic_send({error, Reason}, Conn, Stream) -> quicer:async_shutdown_stream(Stream, ?QUIC_STREAM_SHUTDOWN_FLAG_ABORT, 0), quicer:async_shutdown_connection(Conn, ?QUIC_CONNECTION_SHUTDOWN_FLAG_NONE, 0), {error, {send_failed, Reason}}. %%%=================================================================== %%% NAT-Aware Routing (v0.12.0+) %%%=================================================================== %% @private %% @doc Send message via NAT-aware connector (hole punch, relay fallback). -spec send_via_nat_connector(binary(), binary(), binary(), map()) -> ok | {error, term()}. send_via_nat_connector(LocalNodeId, TargetNodeId, MessageBinary, Opts) -> ?LOG_DEBUG("NAT-aware send: ~s -> ~s", [LocalNodeId, TargetNodeId]), %% Use macula_nat_connector for intelligent routing ConnectResult = macula_nat_connector:connect(LocalNodeId, TargetNodeId, Opts), send_on_nat_connection(ConnectResult, MessageBinary, TargetNodeId). %% @private NAT connection established - send message send_on_nat_connection({ok, Conn, Strategy}, MessageBinary, TargetNodeId) -> ?LOG_INFO("NAT connection established to ~s via ~p", [TargetNodeId, Strategy]), StreamResult = macula_quic:open_stream(Conn), send_on_nat_stream(StreamResult, Conn, MessageBinary, Strategy); %% @private NAT connection failed send_on_nat_connection({error, Reason}, _MessageBinary, TargetNodeId) -> ?LOG_WARNING("NAT connection to ~s failed: ~p", [TargetNodeId, Reason]), {error, {nat_connection_failed, Reason}}. %% @private Stream opened - send message send_on_nat_stream({ok, Stream}, Conn, MessageBinary, Strategy) -> SendResult = macula_quic:send(Stream, MessageBinary), handle_nat_send_result(SendResult, Conn, Stream, Strategy); %% @private Stream open failed send_on_nat_stream({error, Reason}, Conn, _MessageBinary, _Strategy) -> macula_nat_connector:disconnect(Conn), {error, {stream_failed, Reason}}. %% @private Send succeeded - graceful shutdown and return connection to pool handle_nat_send_result(ok, Conn, Stream, Strategy) -> timer:sleep(50), quicer:async_shutdown_stream(Stream, ?QUIC_STREAM_SHUTDOWN_FLAG_GRACEFUL, 0), %% Don't disconnect - keep connection for potential reuse ?LOG_DEBUG("NAT-aware send succeeded via ~p", [Strategy]), maybe_cache_nat_connection(Conn, Strategy), ok; %% @private Send failed handle_nat_send_result({error, Reason}, Conn, Stream, _Strategy) -> quicer:async_shutdown_stream(Stream, ?QUIC_STREAM_SHUTDOWN_FLAG_ABORT, 0), macula_nat_connector:disconnect(Conn), {error, {send_failed, Reason}}. %% @private Cache successful NAT connections for reuse maybe_cache_nat_connection(_Conn, relay) -> %% Don't cache relay connections - they may have per-message overhead ok; maybe_cache_nat_connection(Conn, _Strategy) -> %% For direct and hole_punch, could cache the connection %% For now, just disconnect to keep it simple macula_nat_connector:disconnect(Conn), ok.