%% ============================================================================= %% SPDX-FileCopyrightText: 2016 Christopher Meiklejohn %% SPDX-FileCopyrightText: 2021 - 2025 Alejandro Ramallo %% SPDX-License-Identifier: Apache-2.0 %% ============================================================================= -module(partisan_peer_service_client). -author("Christopher S. Meiklejohn "). -behaviour(gen_server). -include("partisan.hrl"). -include("partisan_logger.hrl"). -include("partisan_peer_socket.hrl"). -define(REPORT, #{ channel => get({?MODULE, channel}), channel_opts => get({?MODULE, channel_opts}), listen_addr => get({?MODULE, listen_addr}), peer => get({?MODULE, peer}) } ). -define(REPORT(Report), maps:merge(Report, ?REPORT) ). -record(state, { socket :: optional(partisan_peer_socket:t()), listen_addr :: partisan:listen_addr(), channel :: partisan:channel(), channel_opts :: partisan:channel_opts(), encoding_opts :: list(), from :: pid(), peer :: partisan:node_spec(), ping_idle_timeout :: non_neg_integer(), ping_tref :: optional(partisan_remote_ref:r()), ping_retry :: optional(partisan_retry:t()), ping_payload :: binary() }). -type state() :: #state{}. %% Macros. -define(TIMEOUT, 1000). %% API -export([start_link/5]). %% gen_server callbacks -export([init/1]). -export([handle_call/3]). -export([handle_cast/2]). -export([handle_info/2]). -export([terminate/2]). -export([code_change/3]). %% ============================================================================= %% API %% ============================================================================= %% ----------------------------------------------------------------------------- %% @doc Start and link to calling process. %% If the process is started and can get a connection it returns `{ok, pid()}'. %% Otherwise if it fails with `{error, Reason :: any()}'. %% @end %% ----------------------------------------------------------------------------- -spec start_link( Peer :: partisan:node_spec(), ListenAddr :: partisan:listen_addr(), Channel :: partisan:channel(), ChannelOpts :: partisan:channel_opts(), From :: pid()) -> {ok, pid()} | ignore | {error, Reason :: any()}. start_link(Peer, ListenAddr, Channel, ChannelOpts, From) -> gen_server:start_link( ?MODULE, [Peer, ListenAddr, Channel, ChannelOpts, From], [] ). %% ============================================================================= %% GEN_SERVER CALLBACKS %% ============================================================================= -spec init(Args :: list()) -> {ok, state()} | {stop, Reason :: any()}. init([Peer, ListenAddr, Channel, ChannelOpts, From]) -> case connect(ListenAddr, Channel, ChannelOpts) of {ok, Socket} -> ?LOG_INFO(#{ description => "Connection established", peer => Peer, listen_addr => ListenAddr, channel => Channel }), %% For debugging, store information in the process dictionary. EgressDelay = partisan_config:get(egress_delay, 0), put({?MODULE, from}, From), put({?MODULE, listen_addr}, ListenAddr), put({?MODULE, channel}, Channel), put({?MODULE, channel_opts}, ChannelOpts), put({?MODULE, peer}, Peer), put({?MODULE, egress_delay}, EgressDelay), EncodeOpts = case maps:get(compression, ChannelOpts, false) of true -> [compressed]; N when N >= 0, N =< 9 -> [{compressed, N}]; _ -> [] end, State0 = #state{ from = From, listen_addr = ListenAddr, channel = Channel, channel_opts = ChannelOpts, encoding_opts = EncodeOpts, socket = Socket, peer = Peer }, State = maybe_enable_ping( State0, partisan_config:get(connection_ping, #{}) ), {ok, State}; {error, Reason} -> ?LOG_TRACE(#{ description => "Unable to establish connection with peer", peer => Peer, listen_addr => ListenAddr, channel => Channel, reason => Reason }), %% We use shutdown to avoid a crash report {stop, normal} end. -spec handle_call(term(), {pid(), term()}, state()) -> {reply, term(), state()}. handle_call({send_message, Msg}, _From, #state{} = State) -> case get({?MODULE, egress_delay}) of 0 -> ok; Other -> timer:sleep(Other) end, Data = partisan_util:encode(Msg, State#state.encoding_opts), case send(State#state.socket, Data) of ok -> ?LOG_TRACE("Dispatched message: ~p", [Msg]), {reply, ok, State}; Error -> ?LOG_DEBUG("Message ~p failed to send: ~p", [Msg, Error]), {reply, Error, State} end; handle_call(Event, _From, State) -> ?LOG_WARNING(#{description => "Unhandled call event", event => Event}), {reply, ok, State}. -spec handle_cast(term(), state()) -> {noreply, state()}. handle_cast({send_message, Msg}, #state{} = State) -> ?LOG_TRACE("Received cast: ~p", [Msg]), case get({?MODULE, egress_delay}) of 0 -> ok; Other -> timer:sleep(Other) end, Data = partisan_util:encode(Msg, State#state.encoding_opts), case send(State#state.socket, Data) of ok -> ?LOG_TRACE("Dispatched message: ~p", [Msg]), ok; Error -> ?LOG_ERROR(#{ description => "Failed to send message", data => Msg, error => Error }) end, {noreply, State}; handle_cast(Event, State) -> ?LOG_WARNING(#{description => "Unhandled cast event", event => Event}), {noreply, State}. -spec handle_info(term(), state()) -> {noreply, state()} | {stop, normal, state()}. handle_info({Tag, _Socket, Data}, State) when ?DATA_MSG(Tag), is_binary(Data) -> Msg = binary_to_term(Data), ?LOG_TRACE("Received tcp data at ~p: ~p", [self(), Msg]), ok = reset_socket_opts(State), handle_inbound(Msg, State); handle_info({Tag, _Socket}, #state{} = State) when ?CLOSED_MSG(Tag) -> ?LOG_TRACE( "Connection to ~p has been closed for pid ~p", [State#state.peer, self()] ), {stop, normal, State}; handle_info( {timeout, Ref, ping_idle_timeout}, #state{ping_tref = Ref} = State) -> ?LOG_INFO(#{ description => "Connection idle, sending ping", peer => State#state.peer, channel => State#state.channel, attempt => partisan_retry:count(State#state.ping_retry) }), maybe_send_ping(State); handle_info({timeout, Ref, ping_timeout}, #state{ping_tref = Ref} = State) -> ?LOG_INFO(#{ description => "Ping timeout, retrying", attempts => partisan_retry:count(State#state.ping_retry) }), maybe_send_ping(State); handle_info(Event, State) -> ?LOG_WARNING(#{description => "Unhandled info event", event => Event}), {noreply, State}. -spec terminate(term(), state()) -> term(). terminate(Reason, #state{} = State) when Reason == normal orelse Reason == shutdown orelse (is_tuple(Reason) andalso element(1, Reason) == shutdown) -> ?LOG_INFO(?REPORT(#{ description => "Connection closed", reason => Reason })), ok = close_socket(State#state.socket); terminate(Reason, #state{} = State) -> ?LOG_ERROR(?REPORT(#{ description => "Connection closed", reason => Reason })), close_socket(State#state.socket). -spec code_change(term() | {down, term()}, state(), term()) -> {ok, state()}. code_change(_OldVsn, State, _Extra) -> {ok, State}. %% ============================================================================= %% PRIVATE %% ============================================================================= %% @doc Test harness specific. %% %% If we're running a local test, we have to use the same IP address for %% every bind operation, but a different port instead of the standard %% port. %% connect(Node, Channel, ChannelOpts) when is_atom(Node), is_atom(Channel), is_map(ChannelOpts) -> %% Used for testing when connect_disterl is enabled partisan_config:get(connect_disterl, false) orelse error(disterl_not_enabled), case rpc:call(Node, partisan_config, get, [listen_addrs]) of {badrpc, Reason} -> {error, Reason}; [] -> {error, no_listen_addr}; [ListenAddr|_] -> %% Use first address connect(ListenAddr, Channel, ChannelOpts) end; connect(#{ip := Address, port := Port}, Channel, ChannelOpts) when is_atom(Channel), is_map(ChannelOpts) -> SocketOpts = [ binary, {active, once}, {packet, 4}, {keepalive, true} ], Opts = [{monotonic, maps:get(monotonic, ChannelOpts, false)}], Result = partisan_peer_socket:connect( Address, Port, SocketOpts, ?TIMEOUT, Opts ), case Result of {ok, Socket} -> {ok, Socket}; {error, Reason} when ?IS_INET_POSIX(Reason) -> %% TODO We do not want to log here becuase this can be high frequency %% so what we should do is store the latest error on the %% partisan_peer_connections as a status that is cleanedup when we %% establish the connection {error, Reason}; {error, Reason} -> ?LOG_ERROR(#{ description => "Error while trying to establish connection", channel => Channel, reason => Reason, ip => Address, port => Port }), {error, Reason} end. %% @private send(undefined, _) -> {error, no_socket}; send(Socket, Data) -> partisan_peer_socket:send(Socket, Data). %% @private close_socket(undefined) -> ok; close_socket(Socket) -> partisan_peer_socket:close(Socket). %% @private handle_inbound({state, Tag, LocalState}, #state{} = State) -> #state{ peer = Peer, channel = Channel, from = From } = State, %% Notify peer service manager we are done. case LocalState of %% TODO: Anything using a three tuple will be caught here. %% TODO: This format is specific to the HyParView manager. {state, _Active, Epoch} -> From ! {connected, Peer, Channel, Tag, Epoch, LocalState}; _Other -> From ! {connected, Peer, Channel, Tag, LocalState} end, {noreply, reset_ping(State)}; handle_inbound({hello, Node}, #state{peer = #{name := Node}} = State) -> Socket = State#state.socket, Msg = term_to_binary({hello, partisan:node(), State#state.channel}), case send(Socket, Msg) of ok -> {noreply, reset_ping(State)}; {error, Reason} = Error -> ?LOG_INFO(#{ description => "Failed to send hello message to node", node => Node, reason => Reason }), {stop, Error, State} end; handle_inbound({hello, A}, #state{peer = #{name := B}} = State) when A =/= B -> %% Peer isn't who it should be, abort. ?LOG_ERROR(#{ description => "Unexpected peer, aborting", got => A, expected => B }), {stop, {unexpected_peer, A, B}, State}; handle_inbound({ping, Node, Payload}, #state{peer = Node} = State) -> send(State#state.socket, term_to_binary({pong, Node, Payload})), {noreply, reset_ping(State)}; handle_inbound({ping, _, _}, #state{} = State) -> {stop, invalid_ping, State}; handle_inbound({pong, Node, _}, #state{} = State) -> case Node == partisan:node() of true -> {noreply, reset_ping(State)}; false -> {stop, invalid_ping_response, State} end; handle_inbound(Msg, State) -> ?LOG_WARNING(#{ description => "Received invalid message", message => Msg }), {stop, normal, State}. %% @private reset_socket_opts(State) -> partisan_peer_socket:setopts(State#state.socket, [{active, once}]). %% ============================================================================= %% PRIVATE: KEEP ALIVE PING %% ============================================================================= %% @private maybe_enable_ping(State, #{enabled := true} = PingOpts) -> IdleTimeout = maps:get(idle_timeout, PingOpts), Timeout = maps:get(timeout, PingOpts), Attempts = maps:get(max_attempts, PingOpts), Retry = partisan_retry:init( ping_timeout, #{ deadline => 0, % disable, use max_retries only interval => Timeout, max_retries => Attempts, backoff_enabled => false } ), State#state{ ping_idle_timeout = IdleTimeout, ping_payload = partisan:make_ref(), ping_retry = Retry }; maybe_enable_ping(#{enabled := false}, State) -> State. %% @private reset_ping(#state{ping_retry = undefined} = State) -> State; reset_ping(#state{ping_tref = undefined} = State) -> Time = State#state.ping_idle_timeout, Ref = erlang:start_timer(Time, self(), ping_idle_timeout), State#state{ping_tref = Ref}; reset_ping(#state{} = State) -> _ = erlang:cancel_timer(State#state.ping_tref), {_, Retry} = partisan_retry:succeed(State#state.ping_retry), Time = State#state.ping_idle_timeout, Ref = erlang:start_timer(Time, self(), ping_idle_timeout), State#state{ ping_retry = Retry, ping_tref = Ref }. %% @private maybe_send_ping(#state{ping_idle_timeout = undefined} = State) -> {noreply, State}; maybe_send_ping(#state{} = State) -> {Result, Retry} = partisan_retry:fail(State#state.ping_retry), maybe_send_ping(Result, State#state{ping_retry = Retry}). %% @private maybe_send_ping(Limit, State) when Limit == deadline orelse Limit == max_retries -> {stop, {shutdown, ping_timeout}, State}; maybe_send_ping(_Time, #state{} = State0) -> Data = term_to_binary({ping, partisan:node(), State0#state.ping_payload}), ok = send(State0#state.socket, Data), %% We schedule the next retry Ref = partisan_retry:fire(State0#state.ping_retry), State = State0#state{ping_tref = Ref}, {noreply, State}.