partisan

%% -------------------------------------------------------------------
%%
%% Copyright (c) 2019 Christopher Meiklejohn.  All Rights Reserved.
%%
%% This file is provided to you under the Apache License,
%% Version 2.0 (the "License"); you may not use this file
%% except in compliance with the License.  You may obtain
%% a copy of the License at
%%
%%   http://www.apache.org/licenses/LICENSE-2.0
%%
%% Unless required by applicable law or agreed to in writing,
%% software distributed under the License is distributed on an
%% "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
%% KIND, either express or implied.  See the License for the
%% specific language governing permissions and limitations
%% under the License.
%%
%% -------------------------------------------------------------------

-module(partisan_pluggable_peer_service_manager).

-behaviour(gen_server).
-behaviour(partisan_peer_service_manager).

%% partisan_peer_service_manager callbacks
-export([start_link/0,
         members/0,
         members_for_orchestration/0,
         myself/0,
         get_local_state/0,
         join/1,
         sync_join/1,
         leave/0,
         leave/1,
         update_members/1,
         on_down/2,
         on_up/2,
         send_message/2,
         cast_message/3,
         forward_message/2,
         forward_message/3,
         cast_message/4,
         forward_message/4,
         cast_message/5,
         forward_message/5,
         receive_message/2,
         decode/1,
         reserve/1,
         partitions/0,
         add_pre_interposition_fun/2,
         remove_pre_interposition_fun/1,
         add_interposition_fun/2,
         remove_interposition_fun/1,
         get_interposition_funs/0,
         get_pre_interposition_funs/0,
         add_post_interposition_fun/2,
         remove_post_interposition_fun/1,
         inject_partition/2,
         resolve_partition/1]).

%% gen_server callbacks
-export([init/1,
         handle_call/3,
         handle_cast/2,
         handle_info/2,
         terminate/2,
         code_change/3]).

%% debug callbacks
-export([connections/0]).

-include("partisan.hrl").

-define(SET, state_orset).

-type pending() :: [node_spec()].
-type from() :: {pid(), atom()}.

-record(state, {actor :: actor(),
                distance_metrics :: dict:dict(),
                vclock :: term(),
                pending :: pending(),
                membership :: list(),
                down_functions :: dict:dict(),
                up_functions :: dict:dict(),
                out_links :: [term()],
                pre_interposition_funs :: dict:dict(),
                interposition_funs :: dict:dict(),
                post_interposition_funs :: dict:dict(),
                sync_joins :: [{node_spec(), from()}],
                connections :: partisan_peer_service_connections:t(),
                membership_strategy :: atom(),
                membership_strategy_state :: term()}).

-type state_t() :: #state{}.

%%%===================================================================
%%% partisan_peer_service_manager callbacks
%%%===================================================================

%% @doc Same as start_link([]).
-spec start_link() -> {ok, pid()} | ignore | {error, term()}.
start_link() ->
    gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).

%% @doc Return membership list.
members() ->
    gen_server:call(?MODULE, members, infinity).

%% @doc Return membership list.
members_for_orchestration() ->
    gen_server:call(?MODULE, members_for_orchestration, infinity).

%% @doc Return connections list.
connections() ->
    gen_server:call(?MODULE, connections, infinity).

%% @doc Return myself.
mynode() ->
    partisan_peer_service_manager:mynode().

%% @doc Return myself.
myself() ->
    partisan_peer_service_manager:myself().

%% @doc Update membership.
update_members(Nodes) ->
    gen_server:call(?MODULE, {update_members, Nodes}, infinity).

%% @doc Return local node's view of cluster membership.
get_local_state() ->
    gen_server:call(?MODULE, get_local_state, infinity).

%% @doc Trigger function on connection close for a given node.
on_down(#{name := Name}, Function) ->
    on_down(Name, Function);
on_down(Name, Function) when is_atom(Name) ->
    gen_server:call(?MODULE, {on_down, Name, Function}, infinity).

%% @doc Trigger function on connection open for a given node.
on_up(#{name := Name}, Function) ->
    on_up(Name, Function);
on_up(Name, Function) when is_atom(Name) ->
    gen_server:call(?MODULE, {on_up, Name, Function}, infinity).

%% @doc Send message to a remote manager.
send_message(Name, Message) ->
    gen_server:call(?MODULE, {send_message, Name, ?DEFAULT_CHANNEL, Message}, infinity).

%% @doc Cast a message to a remote gen_server.
cast_message(Name, ServerRef, Message) ->
    cast_message(Name, ?DEFAULT_CHANNEL, ServerRef, Message).

%% @doc Cast a message to a remote gen_server.
cast_message(Name, Channel, ServerRef, Message) ->
    FullMessage = {'$gen_cast', Message},
    forward_message(Name, Channel, ServerRef, FullMessage),
    ok.

%% @doc Cast a message to a remote gen_server.
cast_message(Name, Channel, ServerRef, Message, Options) ->
    FullMessage = {'$gen_cast', Message},
    forward_message(Name, Channel, ServerRef, FullMessage, Options),
    ok.

%% @doc Gensym support for forwarding.
forward_message(Pid, Message) when is_pid(Pid) ->
    forward_message(node(), ?DEFAULT_CHANNEL, Pid, Message);

forward_message({partisan_remote_reference, Name, ServerRef}, Message) ->
    forward_message(Name, ?DEFAULT_CHANNEL, ServerRef, Message).

%% @doc Forward message to registered process on the remote side.
forward_message(Name, ServerRef, Message) ->
    forward_message(Name, ?DEFAULT_CHANNEL, ServerRef, Message).

%% @doc Forward message to registered process on the remote side.
forward_message(Name, Channel, ServerRef, Message) ->
    forward_message(Name, Channel, ServerRef, Message, []).

%% @doc Forward message to registered process on the remote side.
forward_message(Name, Channel, ServerRef, Message, Options) ->
    % lager:info("Entering forward_message at ~p for message: ~p ~p", [node(), Name, Message]),

    %% Attempt to get the partition key, if possible.
    PartitionKey = proplists:get_value(partition_key, Options, ?DEFAULT_PARTITION_KEY),

    %% Use a clock provided by the sender, otherwise, use a generated one.
    Clock = proplists:get_value(clock, Options, undefined),

    %% Should ack?
    ShouldAck = proplists:get_value(ack, Options, false),

    %% Use causality?
    CausalLabel = proplists:get_value(causal_label, Options, undefined),

    %% Get forwarding options and combine with message specific options.
    ForwardOptions = lists:usort(partisan_config:get(forward_options, []) ++ Options),
    % lager:info("Forwarding options for message ~p are ~p", [Message, ForwardOptions]),

    %% Use configuration to disable fast forwarding.
    DisableFastForward = partisan_config:get(disable_fast_forward, false),

    %% Should use fast forwarding?
    %%
    %% Conditions:
    %% - not labeled for causal delivery
    %% - message does not need acknowledgements
    %%
    FastForward = not (CausalLabel =/= undefined) andalso 
                  not ShouldAck andalso 
                  not DisableFastForward,

    %% If attempting to forward to the local node, bypass.
    case partisan_peer_service_manager:mynode() of
        Name ->
            partisan_util:process_forward(ServerRef, Message),
            ok;
        _ ->
            Disterl = partisan_config:get(disterl, false),
            case Disterl of
                true ->
                    partisan_util:process_forward(ServerRef, Message),
                    ok;
                false ->
                    FullMessage = case partisan_config:get(binary_padding, false) of
                        true ->
                            BinaryPadding = partisan_config:get(binary_padding_term, undefined),
                            {forward_message, Name, Channel, Clock, PartitionKey, ServerRef, {'$partisan_padded', BinaryPadding, Message}, ForwardOptions};
                        false ->
                            {forward_message, Name, Channel, Clock, PartitionKey, ServerRef, Message, ForwardOptions}
                    end,

                    case FastForward of
                        true ->
                            %% Attempt to fast-path through the memoized connection cache.
                            case partisan_connection_cache:dispatch(FullMessage) of
                                ok ->
                                    ok;
                                {error, trap} ->
                                    gen_server:call(?MODULE, FullMessage, infinity)
                            end;
                        false ->
                            gen_server:call(?MODULE, FullMessage, infinity)
                    end                
            end
    end.

%% @doc Receive message from a remote manager.
receive_message(Peer, {forward_message, _SourceNode, _MessageClock, _ServerRef, _Message} = FullMessage) ->
    %% Process the message and generate the acknowledgement.
    gen_server:call(?MODULE, {receive_message, Peer, FullMessage}, infinity);
receive_message(Peer, {forward_message, ServerRef, {'$partisan_padded', _Padding, Message}}) ->
    receive_message(Peer, {forward_message, ServerRef, Message});
receive_message(_Peer, {forward_message, _ServerRef, {causal, Label, _, _, _, _, _} = Message}) ->
    partisan_causality_backend:receive_message(Label, Message);
receive_message(Peer, {forward_message, ServerRef, Message} = FullMessage) ->
    % lager:info("in mesage receive at node ~p for peer ~p: ~p", [node(), Peer, FullMessage]),

    case partisan_config:get(disable_fast_receive, false) of
        true ->
            % lager:info("in mesage receive at node ~p for peer ~p FAST RECEIVE DISABLE: ~p", [node(), Peer, Message]),
            gen_server:call(?MODULE, {receive_message, Peer, FullMessage}, infinity);
        false ->
            % lager:info("in mesage receive at node ~p for peer ~p FAST RECEIVE NOT DISABLE", [node(), Peer]),
            partisan_util:process_forward(ServerRef, Message)
    end;
receive_message(Peer, Message) ->
    gen_server:call(?MODULE, {receive_message, Peer, Message}, infinity).

%% @doc Attempt to join a remote node.
sync_join(Node) ->
    gen_server:call(?MODULE, {sync_join, Node}, infinity).

%% @doc Attempt to join a remote node.
join(Node) ->
    gen_server:call(?MODULE, {join, Node}, infinity).

%% @doc Leave the cluster.
leave() ->
    gen_server:call(?MODULE, {leave, partisan_peer_service_manager:myself()}, infinity).

%% @doc Remove another node from the cluster.
leave(Node) ->
    gen_server:call(?MODULE, {leave, Node}, infinity).

%% @doc Decode state.
decode(Membership) ->
    Membership.

%% @doc Reserve a slot for the particular tag.
reserve(Tag) ->
    gen_server:call(?MODULE, {reserve, Tag}, infinity).

%% @doc
add_pre_interposition_fun(Name, PreInterpositionFun) ->
    gen_server:call(?MODULE, {add_pre_interposition_fun, Name, PreInterpositionFun}, infinity).

%% @doc
remove_pre_interposition_fun(Name) ->
    gen_server:call(?MODULE, {remove_pre_interposition_fun, Name}, infinity).

%% @doc
add_interposition_fun(Name, InterpositionFun) ->
    gen_server:call(?MODULE, {add_interposition_fun, Name, InterpositionFun}, infinity).

%% @doc
remove_interposition_fun(Name) ->
    gen_server:call(?MODULE, {remove_interposition_fun, Name}, infinity).

%% @doc
get_interposition_funs() ->
    gen_server:call(?MODULE, get_interposition_funs, infinity).

%% @doc
get_pre_interposition_funs() ->
    gen_server:call(?MODULE, get_pre_interposition_funs, infinity).

%% @doc
add_post_interposition_fun(Name, PostInterpositionFun) ->
    gen_server:call(?MODULE, {add_post_interposition_fun, Name, PostInterpositionFun}, infinity).

%% @doc
remove_post_interposition_fun(Name) ->
    gen_server:call(?MODULE, {remove_post_interposition_fun, Name}, infinity).

%% @doc Inject a partition.
inject_partition(_Origin, _TTL) ->
    {error, not_implemented}.

%% @doc Resolve a partition.
resolve_partition(_Reference) ->
    {error, not_implemented}.

%% @doc Return partitions.
partitions() ->
    {error, not_implemented}.

%%%===================================================================
%%% gen_server callbacks
%%%===================================================================

%% @private
-spec init([]) -> {ok, state_t()}.
init([]) ->
    %% Seed the random number generator.
    partisan_config:seed(),

    case partisan_config:get(binary_padding, false) of
        true ->    
            %% Use 64-byte binary to force shared heap usage to cut down on copying.
            BinaryPaddingTerm = rand_bits(512),
            partisan_config:set(binary_padding_term, BinaryPaddingTerm);
        _ ->
            undefined
    end,

    %% Process connection exits.
    process_flag(trap_exit, true),

    %% Schedule periodic.
    schedule_periodic(),

    %% Schedule instrumentation.
    schedule_instrumentation(),

    %% Schedule distance metric.
    schedule_distance(),

    %% Schedule periodic connections.
    schedule_connections(),

    %% Schedule periodic retransmissionj.
    schedule_retransmit(),

    %% Schedule tree peers refresh.
    schedule_tree_refresh(),

    Actor = gen_actor(),
    VClock = partisan_vclock:fresh(),
    Connections = partisan_peer_service_connections:new(),
    MembershipStrategy = partisan_config:get(membership_strategy),
    {ok, Membership, MembershipStrategyState} = MembershipStrategy:init(Actor),

    DistanceMetrics = dict:new(),

    {ok, #state{actor=Actor,
                pending=[],
                vclock=VClock,
                pre_interposition_funs=dict:new(),
                interposition_funs=dict:new(),
                post_interposition_funs=dict:new(),
                connections=Connections,
                distance_metrics=DistanceMetrics,
                sync_joins=[],
                up_functions=dict:new(),
                down_functions=dict:new(),
                out_links=[],
                membership=Membership,
                membership_strategy=MembershipStrategy,
                membership_strategy_state=MembershipStrategyState}}.

%% @private
-spec handle_call(term(), {pid(), term()}, state_t()) ->
    {reply, term(), state_t()}.

handle_call({reserve, _Tag}, _From, State) ->
    {reply, {error, no_available_slots}, State};

handle_call({on_up, Name, Function},
            _From,
            #state{up_functions=UpFunctions0}=State) ->
    UpFunctions = dict:append(Name, Function, UpFunctions0),
    {reply, ok, State#state{up_functions=UpFunctions}};

handle_call({on_down, Name, Function},
            _From,
            #state{down_functions=DownFunctions0}=State) ->
    DownFunctions = dict:append(Name, Function, DownFunctions0),
    {reply, ok, State#state{down_functions=DownFunctions}};

handle_call({add_pre_interposition_fun, Name, PreInterpositionFun}, _From, #state{pre_interposition_funs=PreInterpositionFuns0}=State) ->
    PreInterpositionFuns = dict:store(Name, PreInterpositionFun, PreInterpositionFuns0),
    {reply, ok, State#state{pre_interposition_funs=PreInterpositionFuns}};

handle_call({remove_pre_interposition_fun, Name}, _From, #state{pre_interposition_funs=PreInterpositionFuns0}=State) ->
    PreInterpositionFuns = dict:erase(Name, PreInterpositionFuns0),
    {reply, ok, State#state{pre_interposition_funs=PreInterpositionFuns}};

handle_call({add_interposition_fun, Name, InterpositionFun}, _From, #state{interposition_funs=InterpositionFuns0}=State) ->
    InterpositionFuns = dict:store(Name, InterpositionFun, InterpositionFuns0),
    {reply, ok, State#state{interposition_funs=InterpositionFuns}};

handle_call({remove_interposition_fun, Name}, _From, #state{interposition_funs=InterpositionFuns0}=State) ->
    InterpositionFuns = dict:erase(Name, InterpositionFuns0),
    {reply, ok, State#state{interposition_funs=InterpositionFuns}};

handle_call(get_interposition_funs, _From, #state{interposition_funs=InterpositionFuns}=State) ->
    {reply, {ok, InterpositionFuns}, State};

handle_call(get_pre_interposition_funs, _From, #state{pre_interposition_funs=PreInterpositionFuns}=State) ->
    {reply, {ok, PreInterpositionFuns}, State};

handle_call({add_post_interposition_fun, Name, PostInterpositionFun}, _From, #state{post_interposition_funs=PostInterpositionFuns0}=State) ->
    PostInterpositionFuns = dict:store(Name, PostInterpositionFun, PostInterpositionFuns0),
    {reply, ok, State#state{post_interposition_funs=PostInterpositionFuns}};

handle_call({remove_post_interposition_fun, Name}, _From, #state{post_interposition_funs=PostInterpositionFuns0}=State) ->
    PostInterpositionFuns = dict:erase(Name, PostInterpositionFuns0),
    {reply, ok, State#state{post_interposition_funs=PostInterpositionFuns}};

%% For compatibility with external membership services.
handle_call({update_members, Nodes}, 
            _From, 
            #state{membership=Membership}=State) ->
    % lager:debug("Updating membership with: ~p", [Nodes]),

    %% Get the current membership.
    CurrentMembership = [N || #{name := N} <- Membership],
    % lager:debug("CurrentMembership: ~p", [CurrentMembership]),

    %% need to support Nodes as a list of maps or atoms
    %% TODO: require each node to be a map
    NodesNames = lists:map(fun(#{name := N}) ->
                                   N;
                              (N) when is_atom(N) ->
                                   N
                           end, Nodes),

    %% Compute leaving list.
    LeavingNodes = lists:filter(fun(N) ->
                                        not lists:member(N, NodesNames)
                                end, CurrentMembership),
    % lager:debug("LeavingNodes: ~p", [LeavingNodes]),

    %% Issue leaves.
    State1 = lists:foldl(fun(N, S) ->
                                 internal_leave(N, S)
                         end, State, LeavingNodes),

    %% Compute joining list.
    JoiningNodes = lists:filter(fun(#{name := N}) ->
                                        not lists:member(N, CurrentMembership);
                                   (N) when is_atom(N) ->
                                        not lists:member(N, CurrentMembership)
                                end, Nodes),
    % lager:debug("JoiningNodes: ~p", [JoiningNodes]),

    %% Issue joins.
    State2=#state{pending=Pending} = lists:foldl(fun(N, S) ->
                                                         internal_join(N, undefined, S)
                                                 end, State1, JoiningNodes),

    %% Compute current pending list.
    Pending1 = lists:filter(fun(#{name := N}) ->
                                    lists:member(N, NodesNames)
                            end, Pending),

    {reply, ok, State2#state{pending=Pending1}};

handle_call({leave, #{name := Name} = Node}, 
            From, 
            State0) ->
    %% Perform leave.
    State = internal_leave(Node, State0),

    case partisan_peer_service_manager:mynode() of
        Name ->
            gen_server:reply(From, ok),

            {stop, normal, State};
        _ ->
            {reply, ok, State}
    end;

handle_call({join, #{name := Name} = Node},
            _From,
            State0) ->
    case partisan_peer_service_manager:mynode() of
        Name ->
            %% Ignoring self join.
            {reply, ok, State0};
        _ ->
            %% Perform join.
            State = internal_join(Node, undefined, State0),

            %% Return.
            {reply, ok, State}
    end;

handle_call({sync_join, #{name := Name} = Node},
            From,
            State0) ->
    lager:debug("Starting synchronous join to ~p from ~p", [Node, partisan_peer_service_manager:mynode()]),

    case partisan_peer_service_manager:mynode() of
        Name ->
            %% Ignoring self join.
            {reply, ok, State0};
        _ ->
            %% Perform join.
            State = internal_join(Node, From, State0),

            %% Return.
            {noreply, State}
    end;

handle_call({send_message, Name, Channel, Message}, _From,
            #state{pre_interposition_funs=PreInterpositionFuns}=State) ->
    schedule_self_message_delivery(Name, Message, Channel, ?DEFAULT_PARTITION_KEY, PreInterpositionFuns),
    {reply, ok, State};

handle_call({forward_message, Name, Channel, Clock, PartitionKey, ServerRef, OriginalMessage, Options}=_FullMessage, 
            From, 
            #state{pre_interposition_funs=PreInterpositionFuns}=State) ->
    %% lager:info("number of forward_message pre_interposition_funs ~p", [length(dict:to_list(PreInterpositionFuns))]),
    % lager:info("called: ~p", [FullMessage]),

    %% Run all interposition functions.
    DeliveryFun = fun() ->
        %% Fire pre-interposition functions.
        PreFoldFun = fun(_Name, PreInterpositionFun, ok) ->
            %% lager:info("firing forward_message preinterposition fun for original message: ~p", [OriginalMessage]),
            PreInterpositionFun({forward_message, Name, OriginalMessage}),
            ok
        end,
        dict:fold(PreFoldFun, ok, PreInterpositionFuns),

        %% Once pre-interposition returns, then schedule for delivery.
        gen_server:cast(?MODULE, {forward_message, From, Name, Channel, Clock, PartitionKey, ServerRef, OriginalMessage, Options})
    end,

    case partisan_config:get(replaying, false) of 
        false ->
            %% Fire all pre-interposition functions, and then deliver, preserving serial order of messages.
            DeliveryFun();
        true ->
            %% Allow the system to proceed, and the message will be delivered once pre-interposition is done.
            spawn_link(DeliveryFun)
    end,

    {noreply, State};

handle_call({receive_message, Peer, OriginalMessage}, 
            From, 
            #state{pre_interposition_funs=PreInterpositionFuns}=State) ->
    %% lager:info("~p: receive message invoked for message from peer ~p: ~p", [node(), Peer, OriginalMessage]),
    %% lager:info("number of receive_message pre_interposition_funs ~p", [length(dict:to_list(PreInterpositionFuns))]),

    %% Run all interposition functions.
    DeliveryFun = fun() ->
        %% Fire pre-interposition functions.
        PreFoldFun = fun(_Name, PreInterpositionFun, ok) ->
            PreInterpositionFun({receive_message, Peer, OriginalMessage}),
            ok
        end,
        dict:fold(PreFoldFun, ok, PreInterpositionFuns),

        %% Once pre-interposition returns, then schedule for delivery.
        gen_server:cast(?MODULE, {receive_message, From, Peer, OriginalMessage})
    end,

    case partisan_config:get(replaying, false) of 
        false ->
            %% Fire all pre-interposition functions, and then deliver, preserving serial order of messages.
            DeliveryFun();
        true ->
            %% Allow the system to proceed, and the message will be delivered once pre-interposition is done.
            spawn_link(DeliveryFun)
    end,

    {noreply, State};

handle_call(members_for_orchestration, _From, #state{membership=Membership}=State) ->
    {reply, {ok, Membership}, State};

handle_call(members, _From, #state{membership=Membership}=State) ->
    Members = [P || #{name := P} <- Membership],
    {reply, {ok, Members}, State};

handle_call(connections, _From, #state{connections=Connections}=State) ->
    {reply, {ok, Connections}, State};

handle_call(get_local_state, _From, #state{membership_strategy_state=MembershipStrategyState}=State) ->
    {reply, {ok, MembershipStrategyState}, State};

handle_call(Msg, _From, State) ->
    lager:warning("Unhandled call messages at module ~p: ~p", [?MODULE, Msg]),
    {reply, ok, State}.

%% @private
-spec handle_cast(term(), state_t()) -> {noreply, state_t()}.
handle_cast({receive_message, From, Peer, OriginalMessage}, 
            #state{interposition_funs=InterpositionFuns,
                   post_interposition_funs=PostInterpositionFuns} = State) ->
    %% lager:info("~p: Inside the receive interposition, message from ~p at node ~p", [node(), Peer, node()]),
    %% lager:info("~p: Count of interposition funs: ~p", [node(), dict:size(InterpositionFuns)]),

    %% Filter messages using interposition functions.
    FoldFun = fun(_InterpositionName, InterpositionFun, M) ->
        InterpositionResult = InterpositionFun({receive_message, Peer, M}),
        % lager:info("result from interposition ~p is ~p", [InterpositionName, InterpositionResult]),
        InterpositionResult
    end,
    Message = dict:fold(FoldFun, OriginalMessage, InterpositionFuns),

    %% Fire post-interposition functions.
    PostFoldFun = fun(_Name, PostInterpositionFun, ok) ->
        PostInterpositionFun({receive_message, Peer, OriginalMessage}, {receive_message, Peer, Message}),
        ok
    end,
    dict:fold(PostFoldFun, ok, PostInterpositionFuns),

    %% lager:info("~p: Message after receive interposition is: ~p", [node(), Message]),

    case Message of
        undefined ->
            gen_server:reply(From, ok),
            {noreply, State};
        {'$delay', NewMessage} ->
            lager:info("Delaying receive_message due to interposition result: ~p", [NewMessage]),
            gen_server:cast(?MODULE, {receive_message, From, Peer, NewMessage}),
            {noreply, State};
        _ ->
            handle_message(Message, From, State)
    end;

handle_cast({forward_message, From, Name, Channel, Clock, PartitionKey, ServerRef, OriginalMessage, Options},
            #state{interposition_funs=InterpositionFuns, 
                   pre_interposition_funs=PreInterpositionFuns, 
                   post_interposition_funs=PostInterpositionFuns, 
                   connections=Connections, 
                   vclock=VClock0}=State) ->
    %% lager:info("~p: Inside the send interposition, message ~p from ~p at node ~p", [node(), OriginalMessage, Name, node()]),
    %% lager:info("~p: Count of interposition funs: ~p", [node(), dict:size(InterpositionFuns)]),

    %% Filter messages using interposition functions.
    FoldFun = fun(_InterpositionName, InterpositionFun, M) ->
        InterpositionResult = InterpositionFun({forward_message, Name, M}),
        % lager:info("result from interposition ~p is ~p", [InterpositionName, InterpositionResult]),
        InterpositionResult
    end,
    Message = dict:fold(FoldFun, OriginalMessage, InterpositionFuns),

    %% Increment the clock.
    VClock = partisan_vclock:increment(mynode(), VClock0),

    %% Are we using causality?
    CausalLabel = proplists:get_value(causal_label, Options, undefined),

    %% Use local information for message unless it's a causal message.
    {MessageClock, FullMessage} = case CausalLabel of
        undefined ->
            %% Generate a message clock or use the provided clock.
            LocalClock = case Clock of
                undefined ->
                    {undefined, VClock};
                Clock ->
                    Clock
            end,

            {LocalClock, Message};
        CausalLabel ->
            case Clock of 
                %% First time through.
                undefined ->
                    %% We don't have a clock yet, get one using the causality backend.
                    {ok, LocalClock0, CausalMessage} = partisan_causality_backend:emit(CausalLabel, Name, ServerRef, Message),

                    %% Wrap the clock wih a scope.
                    %% TODO: Maybe do this wrapping inside of the causality backend.
                    LocalClock = {CausalLabel, LocalClock0},

                    %% Return clock and wrapped message.
                    {LocalClock, CausalMessage};
                %% Retransmission.
                _ ->
                    %% Get the clock and message we used last time.
                    {ok, LocalClock, CausalMessage} = partisan_causality_backend:reemit(CausalLabel, Clock),

                    %% Return clock and wrapped message.
                    {LocalClock, CausalMessage}
            end
    end,

    case Message of
        undefined ->
            %% Store for reliability, if necessary.
            case proplists:get_value(ack, Options, false) of
                false ->
                    ok;
                true ->
                    %% Acknowledgements.
                    case proplists:get_value(retransmission, Options, false) of 
                        false ->
                            RescheduleableMessage = {forward_message, From, Name, Channel, MessageClock, PartitionKey, ServerRef, OriginalMessage, Options},
                            partisan_acknowledgement_backend:store(MessageClock, RescheduleableMessage);
                        true ->
                            ok
                    end
            end,

            %% Fire post-interposition functions.
            PostFoldFun = fun(_Name, PostInterpositionFun, ok) ->
                PostInterpositionFun({forward_message, Name, OriginalMessage}, {forward_message, Name, FullMessage}),
                ok
            end,
            dict:fold(PostFoldFun, ok, PostInterpositionFuns),

            lager:info("~p: Message ~p after send interposition is: ~p", [node(), OriginalMessage, FullMessage]),

            case From of 
                undefined ->
                    ok;
                _ ->
                    gen_server:reply(From, ok)
            end,

            {noreply, State#state{vclock=VClock}};
        {'$delay', NewMessage} ->
            lager:info("Delaying receive_message due to interposition result: ~p", [NewMessage]),
            gen_server:cast(?MODULE, {forward_message, From, Name, Channel, Clock, PartitionKey, ServerRef, NewMessage, Options}),
            {noreply, State};
        _ ->
            %% Store for reliability, if necessary.
            Result = case proplists:get_value(ack, Options, false) of
                false ->
                    %% Tracing.
                    WrappedMessage =  {forward_message, ServerRef, FullMessage},
                    WrappedOriginalMessage =  {forward_message, ServerRef, OriginalMessage},

                    % lager:info("NO acknowledge message: ~p, options: ~p", [WrappedMessage, Options]),

                    %% Fire post-interposition functions -- trace after wrapping!
                    PostFoldFun = fun(_Name, PostInterpositionFun, ok) ->
                        PostInterpositionFun({forward_message, Name, WrappedOriginalMessage}, {forward_message, Name, WrappedMessage}),
                        ok
                    end,
                    dict:fold(PostFoldFun, ok, PostInterpositionFuns),

                    %% lager:info("~p: Message after send interposition is: ~p", [node(), FullMessage]),

                    %% Send message along.
                    do_send_message(Name,
                                    Channel,
                                    PartitionKey,
                                    WrappedMessage,
                                    Connections,
                                    Options,
                                    PreInterpositionFuns);
                true ->
                    %% Tracing.
                    WrappedOriginalMessage = {forward_message, mynode(), MessageClock, ServerRef, OriginalMessage},
                    WrappedMessage = {forward_message, mynode(), MessageClock, ServerRef, FullMessage},

                    lager:info("should acknowledge message: ~p", [WrappedMessage]),

                    %% Fire post-interposition functions -- trace after wrapping!
                    PostFoldFun = fun(_Name, PostInterpositionFun, ok) ->
                        PostInterpositionFun({forward_message, Name, WrappedOriginalMessage}, {forward_message, Name, WrappedMessage}),
                        ok
                    end,
                    dict:fold(PostFoldFun, ok, PostInterpositionFuns),

                    lager:info("~p: Sending message ~p with clock: ~p", [node(), Message, MessageClock]),
                    lager:info("~p: Message after send interposition is: ~p", [node(), Message]),

                    %% Acknowledgements.
                    case proplists:get_value(retransmission, Options, false) of 
                        false ->
                            RescheduleableMessage = {forward_message, From, Name, Channel, MessageClock, PartitionKey, ServerRef, OriginalMessage, Options},
                            partisan_acknowledgement_backend:store(MessageClock, RescheduleableMessage);
                        true ->
                            ok
                    end,

                    %% Send message along.
                    do_send_message(Name,
                                    Channel,
                                    PartitionKey,
                                    WrappedMessage,
                                    Connections,
                                    Options,
                                    PreInterpositionFuns)
            end,

            case From of 
                undefined ->
                    ok;
                _ ->
                    gen_server:reply(From, Result)
            end,

            {noreply, State#state{vclock=VClock}}
    end;
handle_cast(Msg, State) ->
    lager:warning("Unhandled cast messages at module ~p: ~p", [?MODULE, Msg]),
    {noreply, State}.

%% @private
-spec handle_info(term(), state_t()) -> {noreply, state_t()}.
handle_info(tree_refresh, #state{}=State) ->
    %% Get lazily computed outlinks.
    OutLinks = retrieve_outlinks(),

    %% Reschedule.
    schedule_tree_refresh(),

    {noreply, State#state{out_links=OutLinks}};

handle_info(distance, #state{pending=Pending,
                             membership=Membership,
                             connections=Connections0,
                             pre_interposition_funs=PreInterpositionFuns}=State) ->
    %% Establish any new connections.
    Connections = establish_connections(Pending,
                                        Membership,
                                        Connections0),

    %% Record time.
    SourceTime = erlang:timestamp(),

    %% Send distance requests.
    SourceNode = partisan_peer_service_manager:myself(),

    lists:foreach(fun(Peer) ->
        schedule_self_message_delivery(Peer, {ping, SourceNode, Peer, SourceTime}, ?MEMBERSHIP_PROTOCOL_CHANNEL, ?DEFAULT_PARTITION_KEY, PreInterpositionFuns)
    end, Membership),

    schedule_distance(),

    {noreply, State#state{connections=Connections}};

handle_info(instrumentation, State) ->
    MessageQueueLen = process_info(self(), message_queue_len),
    lager:info("message_queue_len: ~p", [MessageQueueLen]),
    schedule_instrumentation(),
    {noreply, State};

handle_info(periodic, #state{pending=Pending,
                             membership_strategy=MembershipStrategy,
                             membership_strategy_state=MembershipStrategyState0,
                             pre_interposition_funs=PreInterpositionFuns,
                             connections=Connections0}=State) ->
    {ok, Membership, OutgoingMessages, MembershipStrategyState} = MembershipStrategy:periodic(MembershipStrategyState0),
    % lager:info("Periodic fired! Generated ~p messages to send.", [length(OutgoingMessages)]),

    %% Establish any new connections.
    Connections = establish_connections(Pending,
                                        Membership,
                                        Connections0),

    %% Send outgoing messages.
    lists:foreach(fun({Peer, Message}) ->
        schedule_self_message_delivery(Peer, Message, ?MEMBERSHIP_PROTOCOL_CHANNEL, ?DEFAULT_PARTITION_KEY, PreInterpositionFuns)
    end, OutgoingMessages),

    schedule_periodic(),

    {noreply, State#state{membership=Membership,
                          membership_strategy_state=MembershipStrategyState,
                          connections=Connections}};

handle_info(retransmit, #state{pre_interposition_funs=PreInterpositionFuns}=State) ->
    RetransmitFun = fun({_, {forward_message, From, Name, Channel, Clock, PartitionKey, ServerRef, Message, Options}}) ->
        Mynode = partisan_peer_service_manager:mynode(),
        lager:info("~p no acknowledgement yet, restranmitting message ~p with clock ~p to ~p", [Mynode, Message, Clock, Name]),

        %% Fire pre-interposition functions.
        PreFoldFun = fun(_Name, PreInterpositionFun, ok) ->
            lager:info("firing preinterposition fun for original message: ~p", [Message]),
            PreInterpositionFun({forward_message, Name, Message}),
            ok
        end,
        dict:fold(PreFoldFun, ok, PreInterpositionFuns),

        %% Schedule message for redelivery.
        RetryOptions = Options ++ [{retransmission, true}],
        gen_server:cast(?MODULE, {forward_message, From, Name, Channel, Clock, PartitionKey, ServerRef, Message, RetryOptions})
    end,

    {ok, Outstanding} = partisan_acknowledgement_backend:outstanding(),
    %% lager:info("~p outstanding messages are: ~p", [node(), Outstanding]),

    case partisan_config:get(replaying, false) of 
        false ->
            %% Fire all pre-interposition functions, and then deliver, preserving serial order of messages.
            lists:foreach(RetransmitFun, Outstanding);
        true ->
            %% Allow the system to proceed, and the message will be delivered once pre-interposition is done.
            lists:foreach(fun(OutstandingMessage) -> 
                spawn_link(fun() -> 
                    RetransmitFun(OutstandingMessage) 
                end) 
            end, Outstanding)
    end,

    %% Reschedule retransmission.
    schedule_retransmit(),

    {noreply, State};

handle_info(connections, #state{pending=Pending,
                                membership=Membership,
                                sync_joins=SyncJoins0, 
                                connections=Connections0}=State) ->
    %% Trigger connection.
    Connections = establish_connections(Pending,
                                        Membership,
                                        Connections0),

    %% Advance sync_join's if we have enough open connections to remote host.
    SyncJoins = lists:foldl(fun({Node, FromPid}, Joins) ->
            case fully_connected(Node, Connections) of
                true ->
                    lager:debug("Node ~p is now fully connected.", [Node]),
                    gen_server:reply(FromPid, ok),
                    Joins;
                _ ->
                    Joins ++ [{Node, FromPid}]
            end
        end, [], SyncJoins0),

    schedule_connections(),

    {noreply, State#state{pending=Pending,
                          sync_joins=SyncJoins,
                          connections=Connections}};

handle_info({'EXIT', From, _Reason}, #state{connections=Connections0}=State) ->
    %% invoke the down callback on each matching entry
    partisan_peer_service_connections:foreach(
          fun(Node, Pids) ->
                case lists:keymember(From, 3, Pids) andalso length(Pids) =:= 1 of
                    true ->
                        down(Node, State);
                    false ->
                        ok
                end
          end, Connections0),
    %% and then prune all node entries containing the exit pid
    {_, Connections} = partisan_peer_service_connections:prune(From, Connections0),
    {noreply, State#state{connections=Connections}};

handle_info({connected, Node, _Tag, RemoteState},
               #state{pending=Pending0,
                      sync_joins=SyncJoins0,
                      connections=Connections,
                      membership_strategy=MembershipStrategy,
                      membership_strategy_state=MembershipStrategyState0,
                      pre_interposition_funs=PreInterpositionFuns}=State0) ->
    lager:debug("Node ~p connected!", [Node]),

    State = case lists:member(Node, Pending0) of
        true ->
            %% Move out of pending.
            Pending = Pending0 -- [Node],

            %% Update membership by joining with remote membership.
            {ok, Membership, OutgoingMessages, MembershipStrategyState} = MembershipStrategy:join(MembershipStrategyState0, Node, RemoteState),

            %% Gossip the new membership.
            lists:foreach(fun({Peer, Message}) ->
                schedule_self_message_delivery(Peer, Message, ?MEMBERSHIP_PROTOCOL_CHANNEL, ?DEFAULT_PARTITION_KEY, PreInterpositionFuns)
            end, OutgoingMessages),

            %% Announce to the peer service.
            partisan_peer_service_events:update(Membership),

            %% Send up notifications.
            partisan_peer_service_connections:foreach(
                fun(N, Pids) ->
                        case length(Pids -- Pending) =:= 1 of
                            true ->
                                up(N, State0);
                            false ->
                                ok
                        end
                end, Connections),

            %% Return.
            State0#state{pending=Pending,
                         membership=Membership,
                         membership_strategy_state=MembershipStrategyState};
        false ->
            State0
    end,

    %% Notify for sync join.
    SyncJoins = case lists:keyfind(Node, 1, SyncJoins0) of
        {Node, FromPid} ->
            case fully_connected(Node, Connections) of
                true ->
                    gen_server:reply(FromPid, ok),
                    lists:keydelete(FromPid, 2, SyncJoins0);
                _ ->
                    SyncJoins0
            end;
        false ->
            SyncJoins0
    end,

    {noreply, State#state{sync_joins=SyncJoins}};

handle_info(Msg, State) ->
    handle_message(Msg, undefined, State).

%% @private
-spec terminate(term(), state_t()) -> term().
terminate(_Reason, #state{connections=Connections}=_State) ->
    Fun =
        fun(_K, Pids) ->
            lists:foreach(
              fun(Pid) ->
                 try
                     gen_server:stop(Pid, normal, infinity)
                 catch
                     _:_ ->
                         ok
                 end
              end, Pids)
         end,
    partisan_peer_service_connections:foreach(Fun, Connections),
    ok.

%% @private
-spec code_change(term() | {down, term()}, state_t(), term()) -> {ok, state_t()}.
code_change(_OldVsn, State, _Extra) ->
    {ok, State}.

%%%===================================================================
%%% Internal functions
%%%===================================================================

%% @private
gen_actor() ->
    Node = atom_to_list(partisan_peer_service_manager:mynode()),
    Unique = erlang:unique_integer([positive]),
    TS = integer_to_list(Unique),
    Term = Node ++ TS,
    crypto:hash(sha, Term).

%% @private
without_me(Members) ->
    lists:filter(fun(#{name := Name}) ->
                         case partisan_peer_service_manager:mynode() of
                             Name ->
                                 false;
                             _ ->
                                 true
                         end
                 end, Members).

%% @private
establish_connections(Pending,
                      Membership,
                      Connections0) ->
    %% Compute list of nodes that should be connected.
    Peers = without_me(Membership ++ Pending),

    %% Reconnect disconnected members and members waiting to join.
    Connections = lists:foldl(fun(Peer, Cs) ->
                                partisan_util:maybe_connect(Peer, Cs)
                              end, Connections0, without_me(Peers)),

    %% Return the updated list of connections.
    Connections.

%% @private
handle_message({ping, SourceNode, DestinationNode, SourceTime},
               From,
               #state{pending=Pending,
                      connections=Connections0,
                      membership=Membership,
                      pre_interposition_funs=PreInterpositionFuns}=State) ->
    %% Establish any new connections.
    Connections = establish_connections(Pending,
                                        Membership,
                                        Connections0),

    %% Send ping response.
    schedule_self_message_delivery(SourceNode, {pong, SourceNode, DestinationNode, SourceTime}, ?MEMBERSHIP_PROTOCOL_CHANNEL, ?DEFAULT_PARTITION_KEY, PreInterpositionFuns),

    optional_gen_server_reply(From, ok),

    {noreply, State#state{connections=Connections}};

handle_message({pong, SourceNode, DestinationNode, SourceTime},
               From,
               #state{distance_metrics=DistanceMetrics0}=State) ->
    %% Compute difference.
    ArrivalTime = erlang:timestamp(),
    Difference = timer:now_diff(ArrivalTime, SourceTime),
    
    case partisan_config:get(tracing, ?TRACING) of 
        true ->
            lager:info("Updating distance metric for node ~p => ~p communication: ~p", [SourceNode, DestinationNode, Difference]);
        false ->
            ok
    end,

    %% Update differences.
    DistanceMetrics = dict:store(DestinationNode, Difference, DistanceMetrics0),

    %% Store in pdict.
    put(distance_metrics, DistanceMetrics),

    optional_gen_server_reply(From, ok),

    {noreply, State#state{distance_metrics=DistanceMetrics}};

handle_message({membership_strategy, ProtocolMessage},
               From,
               #state{pending=Pending,
                      connections=Connections0,
                      membership=Membership0,
                      membership_strategy=MembershipStrategy,
                      membership_strategy_state=MembershipStrategyState0,
                      pre_interposition_funs=PreInterpositionFuns}=State) ->
    %% Process the protocol message.
    {ok, Membership, OutgoingMessages, MembershipStrategyState} = MembershipStrategy:handle_message(MembershipStrategyState0, ProtocolMessage),

    %% Establish any new connections.
    Connections = establish_connections(Pending,
                                        Membership,
                                        Connections0),

    %% Update users of the peer service.
    case Membership of
        Membership0 ->
            ok;
        _ ->
            partisan_peer_service_events:update(Membership)
    end,

    %% Send outgoing messages.
    lists:foreach(fun({Peer, Message}) ->
        schedule_self_message_delivery(Peer, Message, ?MEMBERSHIP_PROTOCOL_CHANNEL, ?DEFAULT_PARTITION_KEY, PreInterpositionFuns)
    end, OutgoingMessages),

    case lists:member(partisan_peer_service_manager:myself(), Membership) of
        false ->
            lager:info("Shutting down: membership doesn't contain us. ~p not in ~p", [partisan_peer_service_manager:myself(), Membership]),
            %% Shutdown if we've been removed from the cluster.
            {stop, normal, State#state{membership=Membership,
                                       connections=Connections,
                                       membership_strategy_state=MembershipStrategyState}};
        true ->
            optional_gen_server_reply(From, ok),

            {noreply, State#state{membership=Membership,
                                  connections=Connections,
                                  membership_strategy_state=MembershipStrategyState}}
    end;

%% Causal and acknowledged messages.
handle_message({forward_message, SourceNode, MessageClock, ServerRef, {causal, Label, _, _, _, _, _} = Message},
               From,
               #state{pre_interposition_funs=PreInterpositionFuns}=State) ->
    %% Send message acknowledgement.
    send_acknowledgement(SourceNode, MessageClock, PreInterpositionFuns),

    case partisan_causality_backend:is_causal_message(Message) of
        true ->
            partisan_causality_backend:receive_message(Label, Message);
        false ->
            %% Attempt message delivery.
            partisan_util:process_forward(ServerRef, Message)
    end,

    optional_gen_server_reply(From, ok),

    {noreply, State};

%% Acknowledged messages.
handle_message({forward_message, SourceNode, MessageClock, ServerRef, Message},
               From,
               #state{pre_interposition_funs=PreInterpositionFuns}=State) ->
    %% Send message acknowledgement.
    send_acknowledgement(SourceNode, MessageClock, PreInterpositionFuns),

    partisan_util:process_forward(ServerRef, Message),

    optional_gen_server_reply(From, ok),

    {noreply, State};

%% Causal messages.
handle_message({forward_message, ServerRef, {causal, Label, _, _, _, _, _} = Message}, 
               From,
               State) ->
    case partisan_causality_backend:is_causal_message(Message) of
        true ->
            partisan_causality_backend:receive_message(Label, Message);
        false ->
            %% Attempt message delivery.
            partisan_util:process_forward(ServerRef, Message)
    end,

    optional_gen_server_reply(From, ok),

    {noreply, State};

%% Best-effort messages.
%% TODO: Maybe remove me.
handle_message({forward_message, ServerRef, Message}, 
               From,
               State) ->
    partisan_util:process_forward(ServerRef, Message),
    optional_gen_server_reply(From, ok),
    {noreply, State};

handle_message({ack, MessageClock}, 
               From,
               State) ->
    % Mynode = partisan_peer_service_manager:mynode(),
    % lager:warning("~p acknowledgement received for message ~p", [Mynode, MessageClock]),
    partisan_acknowledgement_backend:ack(MessageClock),
    optional_gen_server_reply(From, ok),
    {noreply, State};

handle_message(Message,
               _From,
               State) ->
    lager:warning("Unhandled message at module ~p: ~p", [?MODULE, Message]),
    {noreply, State}.

%% @private
schedule_distance() ->
    case partisan_config:get(distance_enabled, false) of 
        true ->
            DistanceInterval = partisan_config:get(distance_interval, 10000),
            erlang:send_after(DistanceInterval, ?MODULE, distance);
        false ->
            ok
    end.

%% @private
schedule_instrumentation() ->
    case partisan_config:get(instrumentation, false) of 
        true ->
            erlang:send_after(1000, ?MODULE, instrumentation);
        _ ->
            ok
    end.

%% @private
schedule_periodic() ->
    case partisan_config:get(periodic_enabled, false) of 
        true ->
            PeriodicInterval = partisan_config:get(periodic_interval, ?PERIODIC_INTERVAL),
            erlang:send_after(PeriodicInterval, ?MODULE, periodic);
        false ->
            ok
    end.

%% @private
schedule_retransmit() ->
    RetransmitInterval = partisan_config:get(retransmit_interval, 1000),
    erlang:send_after(RetransmitInterval, ?MODULE, retransmit).

%% @private
schedule_connections() ->
    ConnectionInterval = partisan_config:get(connection_interval, 1000),
    erlang:send_after(ConnectionInterval, ?MODULE, connections).

%% @private
do_send_message(Node, Channel, PartitionKey, Message, Connections, Options, PreInterpositionFuns) ->
    %% Find a connection for the remote node, if we have one.
    case partisan_peer_service_connections:find(Node, Connections) of
        {ok, []} ->
            %% Tracing.
            case partisan_config:get(tracing, ?TRACING) of 
                true ->
                    lager:info("Node ~p was connected, but is now disconnected!", [Node]);
                false ->
                    ok
            end,

            %% We were connected, but we're not anymore.
            case partisan_config:get(broadcast, false) of
                true ->
                    case proplists:get_value(transitive, Options, false) of
                        true ->
                            lager:info("Performing tree forward from node ~p to node ~p and message: ~p", [node(), Node, Message]),
                            TTL = partisan_config:get(relay_ttl, ?RELAY_TTL),
                            do_tree_forward(Node, Channel, PartitionKey, Message, Connections, Options, TTL, PreInterpositionFuns);
                        false ->
                            ok
                    end;
                false ->
                    %% Node was connected but is now disconnected.
                    {error, disconnected}
            end;
        {ok, Entries} ->
            Pid = partisan_util:dispatch_pid(PartitionKey, Channel, Entries),
            gen_server:cast(Pid, {send_message, Message});
        {error, not_found} ->
            %% Tracing.
            case partisan_config:get(tracing, ?TRACING) of 
                true ->
                    lager:info("Node ~p is not directly connected.", [Node]);
                false ->
                    ok
            end,

            %% We were connected, but we're not anymore.
            case partisan_config:get(broadcast, false) of
                true ->
                    case proplists:get_value(transitive, Options, false) of
                        true ->
                            lager:info("Performing tree forward from node ~p to node ~p and message: ~p", [node(), Node, Message]),
                            TTL = partisan_config:get(relay_ttl, ?RELAY_TTL),
                            do_tree_forward(Node, Channel, PartitionKey, Message, Connections, Options, TTL, PreInterpositionFuns);
                        false ->
                            ok
                    end;
                false ->
                    %% Node has not been connected yet.
                    {error, not_yet_connected}
            end
    end.

%% @private
up(#{name := Name}, State) ->
    up(Name, State);
up(Name, #state{up_functions=UpFunctions}) ->
    case dict:find(Name, UpFunctions) of
        error ->
            ok;
        {ok, Functions} ->
            [Function() || Function <- Functions],
            ok
    end.

%% @private
down(#{name := Name}, State) ->
    down(Name, State);
down(Name, #state{down_functions=DownFunctions}) ->
    case dict:find(Name, DownFunctions) of
        error ->
            ok;
        {ok, Functions} ->
            [Function() || Function <- Functions],
            ok
    end.

%% @private
internal_leave(#{name := Name} = Node,
               #state{pending=Pending,
                      connections=Connections0,
                      membership_strategy=MembershipStrategy,
                      membership_strategy_state=MembershipStrategyState0,
                      pre_interposition_funs=PreInterpositionFuns}=State) ->
    lager:info("Leaving node ~p at node ~p", [Node, partisan_peer_service_manager:mynode()]),

    {ok, Membership, OutgoingMessages, MembershipStrategyState} = MembershipStrategy:leave(MembershipStrategyState0, Node),

    %% Establish any new connections.
    Connections = establish_connections(Pending,
                                        Membership,
                                        Connections0),

    %% Transmit outgoing messages.
    lists:foreach(fun({Peer, Message}) ->
        schedule_self_message_delivery(Peer, Message, ?MEMBERSHIP_PROTOCOL_CHANNEL, ?DEFAULT_PARTITION_KEY, PreInterpositionFuns)
    end, OutgoingMessages),

    case partisan_config:get(connect_disterl) of 
        true ->
            %% call the net_kernel:disconnect(Node) function to leave erlang network explicitly
            net_kernel:disconnect(Name);
        false ->
            ok
    end,

    State#state{membership=Membership,
                connections=Connections,
                membership_strategy_state=MembershipStrategyState}.

%% @private
internal_join(#{name := Name} = Node,
              From,
              #state{pending=Pending0,
                     membership=Membership,
                     sync_joins=SyncJoins0,
                     connections=Connections0}=State) ->
    case partisan_config:get(connect_disterl) of 
        true ->
            %% Maintain disterl connection for control messages.
            _ = net_kernel:connect_node(Name);
        false ->
            ok
    end,

    %% Add to list of pending connections.
    Pending = [Node|Pending0],

    %% Sleep before connecting, to avoid a rush on connections.
    avoid_rush(),

    %% Add to sync joins list.
    SyncJoins = case From of 
        undefined ->
            SyncJoins0;
        _ ->
            SyncJoins0 ++ [{Node, From}]
    end,

    %% Trigger connection.
    Connections = establish_connections(Pending,
                                        Membership,
                                        Connections0),

    State#state{pending=Pending, 
                sync_joins=SyncJoins, 
                connections=Connections}.

%% @private
fully_connected(Node, Connections) ->
    Parallelism = maps:get(parallelism, Node, ?PARALLELISM),

    Channels = case maps:get(channels, Node, [?DEFAULT_CHANNEL]) of
        [] ->
            [?DEFAULT_CHANNEL];
        undefined ->
            [?DEFAULT_CHANNEL];
        Other ->
            lists:usort(Other ++ [?DEFAULT_CHANNEL])
    end,

    case partisan_peer_service_connections:find(Node, Connections) of
        {ok, Conns} ->
            Open = length(Conns),
            Required = length(Channels) * Parallelism,
            Open =:= Required;
        _ ->
            false
    end.

%% @private
rand_bits(Bits) ->
        Bytes = (Bits + 7) div 8,
        <<Result:Bits/bits, _/bits>> = crypto:strong_rand_bytes(Bytes),
        Result.

%% @private
avoid_rush() ->
    %% Sleep before connecting, to avoid a rush on connections.
    ConnectionJitter = partisan_config:get(connection_jitter, ?CONNECTION_JITTER),
    case partisan_config:get(jitter, false) of
        true ->
            timer:sleep(rand:uniform(ConnectionJitter));
        false ->
            timer:sleep(ConnectionJitter)
    end.

%% @private
do_tree_forward(Node, Channel, PartitionKey, Message, _Connections, Options, TTL, PreInterpositionFuns) ->
    case partisan_config:get(tracing, ?TRACING) of
        true ->
            lager:info("Attempting to forward message ~p from ~p to ~p.", 
                    [Message, partisan_peer_service_manager:mynode(), Node]);
        false ->
            ok
    end,

    %% Preempt with user-supplied outlinks.
    UserOutLinks = proplists:get_value(out_links, Options, undefined),

    OutLinks = case UserOutLinks of
        undefined ->
            try retrieve_outlinks() of
                Value ->
                    Value
            catch
                _:Reason ->
                    lager:error("Outlinks retrieval failed, reason: ~p", [Reason]),
                    []
            end;
        OL ->
            OL -- [node()]
    end,

    %% Send messages, but don't attempt to forward again, if we aren't connected.
    lists:foreach(fun(N) ->
        case partisan_config:get(tracing, ?TRACING) of
            true ->
                lager:info("Forwarding relay message ~p to node ~p for node ~p from node ~p", 
                        [Message, N, Node, partisan_peer_service_manager:mynode()]);
            false ->
                ok
        end,

        RelayMessage = {relay_message, Node, Message, TTL - 1},
        schedule_self_message_delivery(N, RelayMessage, Channel, PartitionKey, proplists:delete(transitive, Options), PreInterpositionFuns)
    end, OutLinks),
    ok.

%% @private
retrieve_outlinks() ->
    case partisan_config:get(tracing, ?TRACING) of
        true ->
            lager:info("About to retrieve outlinks...");
        false ->
            ok
    end,

    Root = partisan_peer_service_manager:mynode(),

    OutLinks = try partisan_plumtree_broadcast:debug_get_peers(partisan_peer_service_manager:mynode(), Root, 1000) of
        {EagerPeers, _LazyPeers} ->
            ordsets:to_list(EagerPeers)
    catch
        _:_ ->
            lager:info("Request to get outlinks timed out..."),
            []
    end,

    case partisan_config:get(tracing, ?TRACING) of
        true ->
            lager:info("Finished getting outlinks: ~p", [OutLinks]);
        false ->
            ok
    end,

    OutLinks -- [node()].

%% @private
schedule_tree_refresh() ->
    case partisan_config:get(broadcast, false) of
        true ->
            Period = partisan_config:get(tree_refresh, 1000),
            erlang:send_after(Period, ?MODULE, tree_refresh);
        false ->
            ok
    end.

%% @private
schedule_self_message_delivery(Name, Message, Channel, PartitionKey, PreInterpositionFuns) ->
    schedule_self_message_delivery(Name, Message, Channel, PartitionKey, [], PreInterpositionFuns).

%% @private
schedule_self_message_delivery(Name, Message, Channel, PartitionKey, Options, PreInterpositionFuns) ->
    %% Run all interposition functions.
    DeliveryFun = fun() ->
        %% Fire pre-interposition functions.
        PreFoldFun = fun(_Name, PreInterpositionFun, ok) ->
            lager:info("firing forward_message preinterposition fun for message: ~p", [Message]),
            PreInterpositionFun({forward_message, Name, Message}),
            ok
        end,
        dict:fold(PreFoldFun, ok, PreInterpositionFuns),

        %% Once pre-interposition returns, then schedule for delivery.
        gen_server:cast(?MODULE, {forward_message, undefined, Name, Channel, undefined, PartitionKey, ?MODULE, Message, Options})
    end,

    case partisan_config:get(replaying, false) of 
        false ->
            %% Fire all pre-interposition functions, and then deliver, preserving serial order of messages.
            DeliveryFun();
        true ->
            %% Allow the system to proceed, and the message will be delivered once pre-interposition is done.
            spawn_link(DeliveryFun)
    end,

    ok.

%% @private
send_acknowledgement(Name, MessageClock, PreInterpositionFuns) ->
    %% Generate message.
    Message = {ack, MessageClock},

    %% Send on the default channel.
    schedule_self_message_delivery(Name, Message, ?DEFAULT_CHANNEL, ?DEFAULT_PARTITION_KEY, PreInterpositionFuns).

%% @private
optional_gen_server_reply(From, Response) ->
    case From of
        undefined ->
            ok;
        _ ->
            gen_server:reply(From, Response)
    end.