partisan_trace_orchestrator.erl

%% -------------------------------------------------------------------
%%
%% Copyright (c) 2018 Christopher S. 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.
%%
%% -------------------------------------------------------------------

%% -----------------------------------------------------------------------------
%% @doc
%% This module requires `disterl'.
%% @end
%% -----------------------------------------------------------------------------
-module(partisan_trace_orchestrator).
-author("Christopher Meiklejohn <christopher.meiklejohn@gmail.com>").

-behaviour(gen_server).

-include("partisan.hrl").
-include("partisan_logger.hrl").

-define(MANAGER, partisan_pluggable_peer_service_manager).

%% API
-export([start_link/0,
         start_link/1,
         stop/0,
         trace/2,
         replay/2,
         reset/0,
         enable/1,
         identify/1,
         print/0,
         perform_preloads/1]).

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

-record(state, {previous_trace=[],
                trace=[],
                enabled=false,
                nodes=[],
                replay=false,
                shrinking=false,
                blocked_processes=[],
                identifier=undefined}).

%%%===================================================================
%%% API
%%%===================================================================

%% @doc Same as start_link([]).
start_link() ->
    start_link([]).

stop() ->
    gen_server:stop({global, ?MODULE}, normal, infinity).

%% @doc Start and link to calling process.
start_link(Args) ->
    gen_server:start_link({global, ?MODULE}, ?MODULE, Args, []).

%% @doc Record trace message.
trace(Type, Message) ->
    gen_server:cast({global, ?MODULE}, {trace, Type, Message}).

%% @doc Replay trace.
replay(Type, Message) ->
    gen_server:call({global, ?MODULE}, {replay, Type, Message}, infinity).

%% @doc Enable trace.
enable(Nodes) ->
    gen_server:call({global, ?MODULE}, {enable, Nodes}, infinity).

%% @doc Reset trace.
reset() ->
    gen_server:call({global, ?MODULE}, reset, infinity).

%% @doc Print trace.
print() ->
    gen_server:call({global, ?MODULE}, print, infinity).

%% @doc Identify trace.
identify(Identifier) ->
    gen_server:call({global, ?MODULE}, {identify, Identifier}, infinity).

%% @doc Perform preloads.
perform_preloads(Nodes) ->
    %% This is a replay, so load the preloads.
    replay_debug("loading preload omissions.", []),
    preload_omissions(Nodes),
    replay_debug("preloads finished.", []),

    ok.

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

%% @private
-spec init([]) -> {ok, #state{}}.
init([]) ->
    debug("test orchestrator started on node: ~p", [partisan:node()]),
    State = initialize_state(),
    {ok, State}.

%% @private
-spec handle_call(term(), {pid(), term()}, #state{}) ->
    {reply, term(), #state{}} | {noreply, #state{}}.

handle_call({enable, Nodes}, _From, #state{enabled=false}=State)
when is_list(Nodes) ->
    ?LOG_INFO("enabling tracing for nodes: ~p", [Nodes]),

    %% Add send and receive pre-interposition functions to enforce message
    %% ordering.
    PreInterpositionFun = fun({Type, OriginNode, OriginalMessage}) ->
        %% TODO: This needs to be fixed: replay and trace need to be done
        %% atomically otherwise processes will race to write trace entry when
        %% they are unblocked from retry: this means that under replay the trace
        %% file might generate small permutations of messages which means it's
        %% technically not the same trace.
        ?LOG_INFO("pre interposition fired for message type: ~p", [Type]),

        %% Under replay ensure they match the trace order (but only for
        %% pre-interposition messages).
        ok = partisan_trace_orchestrator:replay(
            pre_interposition_fun, {node(), Type, OriginNode, OriginalMessage}
        ),

        %% Record message incoming and outgoing messages.
        ok = partisan_trace_orchestrator:trace(
            pre_interposition_fun, {node(), Type, OriginNode, OriginalMessage}
        )
    end,

    lists:foreach(
        fun({_Name, Node}) when is_atom(Node)->
            ?LOG_INFO(
                "Installing $tracing pre-interposition function on node: ~p",
                [Node]
            ),
            ok = rpc:call(
                Node,
                ?MANAGER,
                add_pre_interposition_fun,
                ['$tracing', PreInterpositionFun]
            )
        end,
        Nodes
    ),

    %% Add send and receive post-interposition functions to perform tracing.
    PostInterpositionFun = fun({Type, OriginNode, OriginalMessage}, {Type, OriginNode, RewrittenMessage}) ->
        %% Record outgoing message after transformation.
    ?LOG_INFO("post interposition fired for message type: ~p", [Type]),
        ok = partisan_trace_orchestrator:trace(
            post_interposition_fun,
            {node(), OriginNode, Type, OriginalMessage, RewrittenMessage}
        )
    end,
    lists:foreach(
        fun({_Name, Node}) when is_atom(Node) ->
            ok = rpc:call(
                Node,
                ?MANAGER,
                add_post_interposition_fun,
                ['$tracing', PostInterpositionFun]
            )
        end,
        Nodes
    ),

    {reply, ok, State#state{nodes=Nodes, enabled=true}};

handle_call(Message, _From, #state{enabled=false}=State) ->
    replay_debug("ignoring ~p as tracing is disabled.", [Message]),
    {reply, ok, State};

handle_call({replay, Type, Message}, From, #state{previous_trace=PreviousTrace0, replay=Replay, shrinking=Shrinking, blocked_processes=BlockedProcesses0}=State) ->
    case Replay of
        true ->
            %% Should we enforce trace order during replay?
            ShouldEnforce = case Type of
                pre_interposition_fun ->
                    %% Destructure pre-interposition trace message.
                    {_TracingNode, InterpositionType, _OriginNode, MessagePayload} = Message,

                    case is_membership_strategy_message(InterpositionType, MessagePayload) of
                        true ->
                            membership_strategy_tracing();
                        false ->
                            true
                    end;
                _ ->
                    true
            end,

            case ShouldEnforce of
                false ->
                    {reply, ok, State};
                true ->
                    %% Find next message that should arrive based on the trace.
                    %% Can we process immediately?
                    case can_deliver_based_on_trace(Shrinking, {Type, Message}, PreviousTrace0, BlockedProcesses0) of
                        true ->
                            %% Deliver as much as we can.
                            {PreviousTrace, BlockedProcesses} = trace_deliver(Shrinking, {Type, Message}, PreviousTrace0, BlockedProcesses0),

                            %% Record new trace position and new list of blocked processes.
                            {reply, ok, State#state{blocked_processes=BlockedProcesses, previous_trace=PreviousTrace}};
                        false ->
                            %% If not, store message, block caller until
                            %% processed.
                            BlockedProcesses = [
                                {{Type, Message}, From} | BlockedProcesses0
                            ],
                            %% Block the process.
                            {noreply, State#state{blocked_processes=BlockedProcesses}}
                    end
            end;
        false ->
            {reply, ok, State}
    end;
handle_call(reset, _From, _State) ->
    replay_debug("resetting trace.", []),
    State = initialize_state(),
    {reply, ok, State};
handle_call({identify, Identifier}, _From, State) ->
    replay_debug("identifying trace: ~p", [Identifier]),
    {reply, ok, State#state{identifier=Identifier}};
handle_call(print, _From, #state{trace=Trace}=State) ->
    replay_debug("printing trace", []),

    lists:foreach(fun({Type, Message}) ->
        case Type of
            enter_command ->
                %% Destructure message.
                {TracingNode, Command} = Message,

                %% Format trace accordingly.
                replay_debug("~p entering command: ~p", [TracingNode, Command]);
            exit_command ->
                %% Destructure message.
                {TracingNode, Command} = Message,

                %% Format trace accordingly.
                replay_debug("~p leaving command: ~p", [TracingNode, Command]);
            pre_interposition_fun ->
                %% Destructure message.
                {_TracingNode, _InterpositionType, _OriginNode, _MessagePayload} = Message,

                %% Do nothing.
                ok;
            interposition_fun ->
                %% Destructure message.
                {TracingNode, OriginNode, InterpositionType, MessagePayload} = Message,

                %% Format trace accordingly.
                case InterpositionType of
                    receive_message ->
                        % replay_debug("~p <- ~p: ~p", [TracingNode, OriginNode, MessagePayload]),
                        ok;
                    forward_message ->
                        replay_debug("~p => ~p: ~p", [TracingNode, OriginNode, MessagePayload])
                end;
            post_interposition_fun ->
                %% Destructure message.
                {TracingNode, OriginNode, InterpositionType, MessagePayload, RewrittenMessagePayload} = Message,

                case is_membership_strategy_message(InterpositionType, MessagePayload) andalso not membership_strategy_tracing() of
                    true ->
                        %% Protocol message and we're not tracing protocol messages.
                        ok;
                    _ ->
                        %% Otherwise, format trace accordingly.
                        case MessagePayload =:= RewrittenMessagePayload of
                            true ->
                                case InterpositionType of
                                    receive_message ->
                                        % replay_debug("* ~p <- ~p: ~p", [TracingNode, OriginNode, MessagePayload]),
                                        ok;
                                    forward_message ->
                                        replay_debug("~p => ~p: ~p", [TracingNode, OriginNode, MessagePayload])
                                end;
                            false ->
                                case RewrittenMessagePayload of
                                    undefined ->
                                        case InterpositionType of
                                            receive_message ->
                                                replay_debug("* ~p <- ~p: DROPPED ~p", [TracingNode, OriginNode, MessagePayload]);
                                            forward_message ->
                                                replay_debug("~p => ~p: DROPPED ~p", [TracingNode, OriginNode, MessagePayload])
                                        end;
                                    _ ->
                                        case InterpositionType of
                                            receive_message ->
                                                replay_debug("* ~p <- ~p: REWROTE ~p to ~p", [TracingNode, OriginNode, MessagePayload, RewrittenMessagePayload]);
                                            forward_message ->
                                                replay_debug("~p => ~p: REWROTE ~p to ~p", [TracingNode, OriginNode, MessagePayload, RewrittenMessagePayload])
                                        end
                                end
                        end
                end;
            _ ->
                replay_debug("unknown message type: ~p, message: ~p", [Type, Message])
        end
    end, Trace),

    write_trace(Trace),

    {reply, ok, State};
handle_call(Msg, _From, State) ->
    replay_debug("Unhandled call messages: ~p", [Msg]),
    {reply, ok, State}.

%% @private
-spec handle_cast(term(), #state{}) -> {noreply, #state{}}.
handle_cast({trace, Type, Message}, #state{trace=Trace0}=State) ->
    {noreply, State#state{trace=Trace0++[{Type, Message}]}};
handle_cast(Msg, State) ->
    replay_debug("Unhandled cast messages: ~p", [Msg]),
    {noreply, State}.

%% @private
-spec handle_info(term(), #state{}) -> {noreply, #state{}}.
handle_info(Msg, State) ->
    replay_debug("Unhandled info messages: ~p", [Msg]),
    {noreply, State}.

%% @private
-spec terminate(term(), #state{}) -> term().
terminate(_Reason, #state{nodes=Nodes}) ->
    lists:foreach(fun({_Name, Node}) ->
        replay_debug("removing tracing pre-interposition on node: ~p", [Node]),
        rpc:call(Node, ?MANAGER, remove_pre_interposition_fun, ['$tracing'])
    end, Nodes),

    lists:foreach(fun({_Name, Node}) ->
        replay_debug("removing tracing post-interposition on node: ~p", [Node]),
        rpc:call(Node, ?MANAGER, remove_post_interposition_fun, ['$tracing'])
    end, Nodes),

    ok.

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

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

%% @private
-spec can_deliver_based_on_trace(term(), term(), term(), [term()]) ->
    true | false.

can_deliver_based_on_trace(Shrinking, {Type, Message}, PreviousTrace, BlockedProcesses)  ->
    replay_debug("determining if message ~p: ~p can be delivered.", [Type, Message]),

    case PreviousTrace of
        [{NextType, NextMessage} | _] ->
            replay_debug("waiting for message ~p: ~p ", [NextType, NextMessage]),

            CanDeliver = case {NextType, NextMessage} of
                {Type, Message} ->
                    replay_debug("=> YES!", []),
                    true;
                _ ->
                    %% But, does the message actually exist in the trace?
                    case lists:member({Type, Message}, PreviousTrace) of
                        true ->
                            replay_debug("=> NO, waiting for message: ~p: ~p", [NextType, NextMessage]),
                            false;
                        false ->
                            %% new messages in the middle of the trace.
                            %% this *should* be the common case if we shrink from the start of the trace forward (foldl)
                            case Shrinking of
                                true ->
                                    replay_debug("=> CONDITIONAL YES, message doesn't exist in previous trace, but shrinking: ~p ~p", [Type, Message]),
                                    true;
                                false ->
                                    replay_debug("=> NO, waiting for message: ~p: ~p", [NextType, NextMessage]),
                                    false
                            end
                    end
            end,

            replay_debug("can deliver: ~p", [CanDeliver]),
            replay_debug("blocked processes: ~p", [length(BlockedProcesses)]),

            CanDeliver;
        [] ->
            %% end of trace, but we are still receiving messages.
            %% this *should* be the common case if we shrink from the back of the trace forward (foldr)
            case Shrinking of
                true ->
                    replay_debug("=> CONDITIONAL YES, message doesn't exist in previous trace, but shrinking: ~p ~p", [Type, Message]),
                    true;
                false ->
                    replay_debug("=> message should not have been delivered, blocking.", []),
                    false
            end
    end.

%% @private
trace_deliver(Shrinking, {Type, Message}, [{Type, Message} | Trace], BlockedProcesses) ->
    replay_debug("delivering single message!", []),

    %% Advance the trace, then try to flush the blocked processes.
    trace_deliver_log_flush(Shrinking, Trace, BlockedProcesses);
trace_deliver(_Shrinking, {_, _}, [{_, _} = NextMessage | Trace], BlockedProcesses) ->
    replay_debug("delivering single message (not in the trace)!", []),

    %% Advance the trace, don't flush blocked processes, since we weren't in the trace, nothing is blocked.
    {[NextMessage|Trace], BlockedProcesses};
trace_deliver(_Shrinking, {_, _}, [], BlockedProcesses) ->
    replay_debug("delivering single message (not in the trace -- end of trace)!", []),

    %% No trace to advance.
    {[], BlockedProcesses}.

%% @private
trace_deliver_log_flush(Shrinking, Trace0, BlockedProcesses0) ->
    replay_debug("attempting to flush blocked messages!", []),

    %% Iterate blocked processes in an attempt to remove one.
    {ND, T, BP} = lists:foldl(fun({{NextType, NextMessage}, Pid} = BP, {NumDelivered1, Trace1, BlockedProcesses1}) ->
        case can_deliver_based_on_trace(Shrinking, {NextType, NextMessage}, Trace1, BlockedProcesses0) of
            true ->
                replay_debug("message ~p ~p can be unblocked!", [NextType, NextMessage]),

                %% Advance the trace.
                [{_, _} | RestOfTrace] = Trace1,

                %% Advance the count of delivered messages.
                NewNumDelivered = NumDelivered1 + 1,

                %% Unblock the process.
                gen_server:reply(Pid, ok),

                %% Remove from the blocked processes list.
                NewBlockedProcesses = BlockedProcesses1 -- [BP],

                {NewNumDelivered, RestOfTrace, NewBlockedProcesses};
            false ->
                replay_debug("pid ~p CANNOT be unblocked yet, unmet dependencies!", [Pid]),

                {NumDelivered1, Trace1, BlockedProcesses1}
        end
    end, {0, Trace0, BlockedProcesses0}, BlockedProcesses0),

    %% Did we deliver something? If so, try again.
    case ND > 0 of
        true ->
            %% replay_debug("was able to deliver a message, trying again", []),
            trace_deliver_log_flush(Shrinking, T, BP);
        false ->
            replay_debug("flush attempt finished.", []),
            {T, BP}
    end.

%% @private
preload_omission_file() ->
    case os:getenv("PRELOAD_OMISSIONS_FILE") of
        false ->
            undefined;
        Other ->
            Other
    end.

%% @private
trace_file() ->
    case os:getenv("TRACE_FILE") of
        false ->
            "/tmp/partisan-latest.trace";
        Other ->
            Other
    end.

%% @private
replay_trace_file() ->
    case os:getenv("REPLAY_TRACE_FILE") of
        false ->
            "/tmp/partisan-latest.trace";
        Other ->
            Other
    end.

%% @private
initialize_state() ->
    case os:getenv("REPLAY_DEBUG", "false") of
        "true" ->
            partisan_config:set(replay_debug, true);
        _ ->
            partisan_config:set(replay_debug, false)
    end,

    case os:getenv("REPLAY") of
        false ->
            %% This is not a replay, so store the current trace.
            replay_debug("recording trace to file.", []),
            #state{enabled=false, trace=[], blocked_processes=[], identifier=undefined};
        _ ->
            %% Mark that we are in replay mode.
            partisan_config:set(replaying, true),

            %% This is a replay, so load the previous trace.
            replay_debug("loading previous trace for replay.", []),

            ReplayTraceFile = replay_trace_file(),
            {ok, Lines} = partisan_trace_file:read(ReplayTraceFile),

            lists:foreach(fun(Line) -> replay_debug("~p", [Line]) end, Lines),

            replay_debug("trace loaded.", []),

            Shrinking = case os:getenv("SHRINKING") of
                false ->
                    false;
                _ ->
                    %% Mark that we are in shrinking mode.
                    partisan_config:set(shrinking, true),

                    true
            end,

            #state{enabled=false, trace=[], previous_trace=Lines, replay=true, shrinking=Shrinking, blocked_processes=[], identifier=undefined}
    end.

%% @private
write_trace(Trace) ->
    %% Write trace.
    FilteredTrace = lists:filter(fun({Type, Message}) ->
        case Type of
            pre_interposition_fun ->
                %% Trace all entry points if protocol message, unless tracing enabled.
                {_TracingNode, InterpositionType, _OriginNode, MessagePayload} = Message,

                case is_membership_strategy_message(InterpositionType, MessagePayload) of
                    true ->
                        %% Trace protocol messages only if protocol tracing is enabled.
                        membership_strategy_tracing();
                    false ->
                        %% Always trace non-protocol messages.
                        true
                end;
            enter_command ->
                true;
            exit_command ->
                true;
            _ ->
                false
        end
    end, Trace),

    %% Write the trace file out.
    TraceFile = trace_file(),
    ok = partisan_trace_file:write(TraceFile, FilteredTrace),

    ok.

%% Should we do replay debugging?
replay_debug(Line, Args) ->
    case partisan_config:get(replay_debug) of
        true ->
            ?LOG_INFO("~p: " ++ Line, [?MODULE] ++ Args);
        _ ->
            ok
    end.

debug(Line, Args) ->
    ?LOG_INFO("~p: " ++ Line, [?MODULE] ++ Args).

%% @private
membership_strategy_tracing() ->
    partisan_config:get(membership_strategy_tracing, ?MEMBERSHIP_STRATEGY_TRACING).

%% @private
preload_omissions(Nodes) ->
    PreloadOmissionFile = preload_omission_file(),

    case PreloadOmissionFile of
        undefined ->
            replay_debug("no preload omissions file...", []),
            ok;
        _ ->
            {ok, Omissions} = partisan_trace_file:read(PreloadOmissionFile),

            %% Preload each omission at the correct node.
            lists:foldl(fun({T, Message}, OmissionNodes0) ->
                case T of
                    pre_interposition_fun ->
                        {TracingNode, forward_message, OriginNode, MessagePayload} = Message,

                        replay_debug("enabling preload omission for ~p => ~p: ~p", [TracingNode, OriginNode, MessagePayload]) ,

                        InterpositionFun = fun
                            ({forward_message, N, M}) ->
                                case N of
                                    OriginNode ->
                                        case M of
                                            MessagePayload ->
                                                ?LOG_INFO("~p: dropping packet from ~p to ~p due to preload interposition.", [partisan:node(), TracingNode, OriginNode]),

                                                case partisan_config:get(fauled_for_background) of
                                                    true ->
                                                        ok;
                                                    _ ->
                                                        ?LOG_INFO("~p: setting node ~p to faulted due to preload interposition hit on message: ~p", [partisan:node(), TracingNode, Message]),
                                                        partisan_config:set(fauled_for_background, true)
                                                end,

                                                undefined;
                                            Other ->
                                                ?LOG_INFO("~p: allowing message, doesn't match interposition payload while node matches", [partisan:node()]),
                                                ?LOG_INFO("~p: => expecting: ~p", [partisan:node(), MessagePayload]),
                                                ?LOG_INFO("~p: => got: ~p", [partisan:node(), Other]),
                                                M
                                        end;
                                    OtherNode ->
                                        ?LOG_INFO("~p: allowing message, doesn't match interposition as destination is ~p and not ~p", [partisan:node(), TracingNode, OtherNode]),
                                        M
                                end;
                            ({receive_message, _N, M}) ->
                                M
                        end,

                        %% Install function.
                        ok = rpc:call(TracingNode, ?MANAGER, add_interposition_fun, [{send_omission, OriginNode, Message}, InterpositionFun]),

                        lists:usort(OmissionNodes0 ++ [TracingNode]);
                    Other ->
                        replay_debug("unknown preload: ~p", [Other]),
                        OmissionNodes0
                end
            end, [], Omissions)
    end,

    %% Load background annotations.
    BackgroundAnnotations = background_annotations(),
    debug("background annotations are: ~p", [BackgroundAnnotations]),

    %% Install faulted tracing interposition function.
    lists:foreach(fun({_, Node}) ->
        InterpositionFun = fun({forward_message, _N, M}) ->
            case partisan_config:get(faulted) of
                true ->
                    case M of
                        undefined ->
                            undefined;
                        _ ->
                            replay_debug("~p: faulted during forward_message of background message, message ~p should be dropped.", [partisan:node(), M]),
                            undefined
                    end;
                _ ->
                    M
            end;
            ({receive_message, _N, M}) ->
                case partisan_config:get(faulted) of
                    true ->
                        case M of
                            undefined ->
                                undefined;
                            _ ->
                                replay_debug("~p: faulted during receive_message of background message, message ~p should be dropped.", [partisan:node(), M]),
                                undefined
                        end;
                    _ ->
                        M
                end
        end,

        %% Install function.
        replay_debug("installing faulted pre-interposition for node: ~p", [Node]),
        ok = rpc:call(Node, ?MANAGER, add_interposition_fun, [{faulted, Node}, InterpositionFun])
    end, Nodes),

    %% Install faulted_for_background tracing interposition function.
    lists:foreach(fun({_, Node}) ->
        InterpositionFun = fun({forward_message, _N, M}) ->
            replay_debug("~p: interposition called for message: ~p", [partisan:node(), M]),

            case partisan_config:get(faulted_for_background) of
                true ->
                    case M of
                        undefined ->
                            undefined;
                        _ ->
                            MessageType = message_type(forward_message, M),

                            case lists:member(element(2, MessageType), BackgroundAnnotations) of
                                true ->
                                    ?LOG_INFO("~p: faulted_for_background during forward_message of background message, message ~p should be dropped.", [partisan:node(), M]),
                                    undefined;
                                false ->
                                    ?LOG_INFO("~p: faulted_for_background, but forward_message payload is not background message: ~p, message_type: ~p", [partisan:node(), M, MessageType]),
                                    M
                            end
                    end;
                _ ->
                    M
            end;
            ({receive_message, _N, M}) ->
                case partisan_config:get(faulted_for_background) of
                    true ->
                        case M of
                            undefined ->
                                undefined;
                            _ ->
                                MessageType = message_type(receive_message, M),

                                case lists:member(element(2, MessageType), BackgroundAnnotations) of
                                    true ->
                                        ?LOG_INFO("~p: faulted_for_background during receive_message of background message, message ~p should be dropped.", [partisan:node(), M]),
                                        undefined;
                                    false ->
                                        ?LOG_INFO("~p: faulted_for_background, but receive_message payload is not background message: ~p, message_type: ~p", [partisan:node(), M, MessageType]),
                                        M
                                end
                        end;
                    _ ->
                        M
                end
        end,

        %% Install function.
        replay_debug("installing faulted_for_background pre-interposition for node: ~p", [Node]),
        ok = rpc:call(Node, ?MANAGER, add_interposition_fun, [{faulted_for_background, Node}, InterpositionFun])
    end, Nodes),

    ok.

%% @private
message_type(InterpositionType, MessagePayload) ->
    ?LOG_INFO("interposition_type: ~p, payload: ~p", [InterpositionType, MessagePayload]),

    case InterpositionType of
        forward_message ->
            MessageType1 = element(1, MessagePayload),

            ActualType = case MessageType1 of
                '$gen_cast' ->
                    CastMessage = element(2, MessagePayload),
                    element(1, CastMessage);
                _ ->
                    MessageType1
            end,

            {forward_message, ActualType};
        receive_message ->
            {forward_message, _Module, Payload} = MessagePayload,
            MessageType1 = element(1, Payload),

            ActualType = case MessageType1 of
                '$gen_cast' ->
                    CastMessage = element(2, Payload),
                    element(1, CastMessage);
                _ ->
                    MessageType1
            end,

            {receive_message, ActualType}
    end.


%% @private
background_annotations() ->
    %% Get module as string.
    background_annotations(os:getenv("IMPLEMENTATION_MODULE")).

%% @private
background_annotations(false) ->
    debug("IMPLEMENTATION_MODULE env var undefined", []),
    [];

background_annotations(ModuleString) ->
    %% Open the annotations file.
    AnnotationsFile = "./annotations/partisan-annotations-" ++ ModuleString,

    case filelib:is_file(AnnotationsFile) of
        false ->
            debug("Annotations file doesn't exist: ~p~n", [AnnotationsFile]),
            [];

        true ->
            {ok, [RawAnnotations]} = file:consult(AnnotationsFile),
            % debug("Raw annotations loaded: ~p~n", [RawAnnotations]),
            AllAnnotations = dict:from_list(RawAnnotations),
            % debug("Annotations loaded: ~p~n", [dict:to_list(AllAnnotations)]),

            % {ok, RawCausalityAnnotations} = dict:find(causality, AllAnnotations),
            % debug("Raw causality annotations loaded: ~p~n", [RawCausalityAnnotations]),

            % CausalityAnnotations = dict:from_list(RawCausalityAnnotations),
            % debug("Causality annotations loaded: ~p~n", [dict:to_list(CausalityAnnotations)]),

            {ok, BackgroundAnnotations} = dict:find(background, AllAnnotations),
            % debug("Background annotations loaded: ~p~n", [BackgroundAnnotations]),

            BackgroundAnnotations
    end.

%%%===================================================================
%%% Trace filtering: super hack, until we can refactor these messages.
%%%===================================================================

%% TODO: Find a way to disable distance metrics, which are non-deterministic
%% for the test execution.

%% TODO: Rename protocol messages, find a way to specify an external
%% function for filtering the trace -- do it from the harness.  Maybe
%% some sort of glob function for receives and forwards to filter.

%% TODO: Weird hack, now everything goes through the interposition mechanism
%% which means we can capture everything *good!* but we only want
%% to see a small subset of it -- acks, yes!, membership updates, sometimes!
%% but distances?  never.  so, we need some really terrible hacks here.

%% TODO: Change "protocol" tracing to "membership strategy" tracing.

%% Pre-interposition examples.
is_membership_strategy_message(receive_message, {_, _, {membership_strategy, _}}) ->
    true;

is_membership_strategy_message(forward_message, {membership_strategy, _}) ->
    true;

%% Post-interposition examples.
is_membership_strategy_message(forward_message, {_, _, {membership_strategy, _}}) ->
    true;

is_membership_strategy_message(receive_message, {membership_strategy, _}) ->
    true;

is_membership_strategy_message(_Type, _Message) ->
    false.