mem3_rpc.erl

% Copyright 2013 Cloudant
%
% Licensed 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(mem3_rpc).

-export([
    find_common_seq/4,
    get_missing_revs/4,
    update_docs/4,
    pull_replication/1,
    load_checkpoint/4,
    load_checkpoint/5,
    save_checkpoint/6,

    load_purge_infos/4,
    save_purge_checkpoint/4,
    purge_docs/4,

    replicate/4
]).

% Private RPC callbacks
-export([
    find_common_seq_rpc/3,
    load_checkpoint_rpc/3,
    pull_replication_rpc/1,
    load_checkpoint_rpc/4,
    save_checkpoint_rpc/5,

    load_purge_infos_rpc/3,
    save_purge_checkpoint_rpc/3,

    replicate_rpc/2
]).

-include("mem3.hrl").
-include_lib("couch/include/couch_db.hrl").

-define(BATCH_SIZE, 1000).
-define(REXI_CALL_TIMEOUT_MSEC, 600000).

% "Pull" is a bit of a misnomer here, as what we're actually doing is
% issuing an RPC request and telling the remote node to push updates to
% us. This lets us reuse all of the battle-tested machinery of mem3_rpc.
pull_replication(Seed) ->
    rexi_call(Seed, {mem3_rpc, pull_replication_rpc, [node()]}).

get_missing_revs(Node, DbName, IdsRevs, Options) ->
    rexi_call(Node, {fabric_rpc, get_missing_revs, [DbName, IdsRevs, Options]}).

update_docs(Node, DbName, Docs, Options) ->
    rexi_call(Node, {fabric_rpc, update_docs, [DbName, Docs, Options]}).

load_checkpoint(Node, DbName, SourceNode, SourceUUID, <<>>) ->
    % Upgrade clause for a mixed cluster for old nodes that don't have
    % load_checkpoint_rpc/4 yet. FilterHash is currently not
    % used and so defaults to <<>> everywhere
    load_checkpoint(Node, DbName, SourceNode, SourceUUID);
load_checkpoint(Node, DbName, SourceNode, SourceUUID, FilterHash) ->
    Args = [DbName, SourceNode, SourceUUID, FilterHash],
    rexi_call(Node, {mem3_rpc, load_checkpoint_rpc, Args}).

load_checkpoint(Node, DbName, SourceNode, SourceUUID) ->
    Args = [DbName, SourceNode, SourceUUID],
    rexi_call(Node, {mem3_rpc, load_checkpoint_rpc, Args}).

save_checkpoint(Node, DbName, DocId, Seq, Entry, History) ->
    Args = [DbName, DocId, Seq, Entry, History],
    rexi_call(Node, {mem3_rpc, save_checkpoint_rpc, Args}).

find_common_seq(Node, DbName, SourceUUID, SourceEpochs) ->
    Args = [DbName, SourceUUID, SourceEpochs],
    rexi_call(Node, {mem3_rpc, find_common_seq_rpc, Args}).

load_purge_infos(Node, DbName, SourceUUID, Count) ->
    Args = [DbName, SourceUUID, Count],
    rexi_call(Node, {mem3_rpc, load_purge_infos_rpc, Args}).

save_purge_checkpoint(Node, DbName, PurgeDocId, Body) ->
    Args = [DbName, PurgeDocId, Body],
    rexi_call(Node, {mem3_rpc, save_purge_checkpoint_rpc, Args}).

purge_docs(Node, DbName, PurgeInfos, Options) ->
    rexi_call(Node, {fabric_rpc, purge_docs, [DbName, PurgeInfos, Options]}).

replicate(Source, Target, DbName, Timeout) when
    is_atom(Source), is_atom(Target), is_binary(DbName)
->
    Args = [DbName, Target],
    rexi_call(Source, {mem3_rpc, replicate_rpc, Args}, Timeout).

load_checkpoint_rpc(DbName, SourceNode, SourceUUID) ->
    load_checkpoint_rpc(DbName, SourceNode, SourceUUID, <<>>).

load_checkpoint_rpc(DbName, SourceNode, SourceUUID, FilterHash) ->
    erlang:put(io_priority, {internal_repl, DbName}),
    case get_or_create_db(DbName, [?ADMIN_CTX]) of
        {ok, Db} ->
            TargetUUID = couch_db:get_uuid(Db),
            NewId = mem3_rep:make_local_id(SourceUUID, TargetUUID, FilterHash),
            case couch_db:open_doc(Db, NewId, []) of
                {ok, Doc} ->
                    rexi:reply({ok, {NewId, Doc}});
                {not_found, _} ->
                    OldId = mem3_rep:make_local_id(SourceNode, node()),
                    case couch_db:open_doc(Db, OldId, []) of
                        {ok, Doc} ->
                            rexi:reply({ok, {NewId, Doc}});
                        {not_found, _} ->
                            rexi:reply({ok, {NewId, #doc{id = NewId}}})
                    end
            end;
        Error ->
            rexi:reply(Error)
    end.

save_checkpoint_rpc(DbName, Id, SourceSeq, NewEntry0, History0) ->
    erlang:put(io_priority, {internal_repl, DbName}),
    case get_or_create_db(DbName, [?ADMIN_CTX]) of
        {ok, Db} ->
            NewEntry = {
                [
                    {<<"target_node">>, atom_to_binary(node(), utf8)},
                    {<<"target_uuid">>, couch_db:get_uuid(Db)},
                    {<<"target_seq">>, couch_db:get_update_seq(Db)}
                ] ++ NewEntry0
            },
            Body =
                {[
                    {<<"seq">>, SourceSeq},
                    {<<"target_uuid">>, couch_db:get_uuid(Db)},
                    {<<"history">>, add_checkpoint(NewEntry, History0)}
                ]},
            Doc = #doc{id = Id, body = Body},
            rexi:reply(
                try couch_db:update_doc(Db, Doc, []) of
                    {ok, _} ->
                        {ok, Body};
                    Else ->
                        {error, Else}
                catch
                    Exception ->
                        Exception;
                    error:Reason ->
                        {error, Reason}
                end
            );
        Error ->
            rexi:reply(Error)
    end.

find_common_seq_rpc(DbName, SourceUUID, SourceEpochs) ->
    erlang:put(io_priority, {internal_repl, DbName}),
    case get_or_create_db(DbName, [?ADMIN_CTX]) of
        {ok, Db} ->
            case couch_db:get_uuid(Db) of
                SourceUUID ->
                    TargetEpochs = couch_db:get_epochs(Db),
                    Seq = compare_epochs(SourceEpochs, TargetEpochs),
                    rexi:reply({ok, Seq});
                _Else ->
                    rexi:reply({ok, 0})
            end;
        Error ->
            rexi:reply(Error)
    end.

pull_replication_rpc(Target) ->
    Dbs = mem3_sync:local_dbs(),
    Opts = [{batch_size, 1000}, {batch_count, 50}],
    Repl = fun(Db) -> {Db, mem3_rep:go(Db, Target, Opts)} end,
    rexi:reply({ok, lists:map(Repl, Dbs)}).

load_purge_infos_rpc(DbName, SrcUUID, BatchSize) ->
    erlang:put(io_priority, {internal_repl, DbName}),
    case get_or_create_db(DbName, [?ADMIN_CTX]) of
        {ok, Db} ->
            TgtUUID = couch_db:get_uuid(Db),
            PurgeDocId = mem3_rep:make_purge_id(SrcUUID, TgtUUID),
            StartSeq =
                case couch_db:open_doc(Db, PurgeDocId, []) of
                    {ok, #doc{body = {Props}}} ->
                        couch_util:get_value(<<"purge_seq">>, Props);
                    {not_found, _} ->
                        Oldest = couch_db:get_oldest_purge_seq(Db),
                        erlang:max(0, Oldest - 1)
                end,
            FoldFun = fun({PSeq, UUID, Id, Revs}, {Count, Infos, _}) ->
                NewCount = Count + length(Revs),
                NewInfos = [{UUID, Id, Revs} | Infos],
                Status =
                    if
                        NewCount < BatchSize -> ok;
                        true -> stop
                    end,
                {Status, {NewCount, NewInfos, PSeq}}
            end,
            InitAcc = {0, [], StartSeq},
            {ok, {_, PurgeInfos, ThroughSeq}} =
                couch_db:fold_purge_infos(Db, StartSeq, FoldFun, InitAcc),
            PurgeSeq = couch_db:get_purge_seq(Db),
            Remaining = PurgeSeq - ThroughSeq,
            rexi:reply({ok, {PurgeDocId, PurgeInfos, ThroughSeq, Remaining}});
        Else ->
            rexi:reply(Else)
    end.

save_purge_checkpoint_rpc(DbName, PurgeDocId, Body) ->
    erlang:put(io_priority, {internal_repl, DbName}),
    case get_or_create_db(DbName, [?ADMIN_CTX]) of
        {ok, Db} ->
            Doc = #doc{id = PurgeDocId, body = Body},
            Resp =
                try couch_db:update_doc(Db, Doc, []) of
                    Resp0 -> Resp0
                catch
                    T:R ->
                        {T, R}
                end,
            rexi:reply(Resp);
        Error ->
            rexi:reply(Error)
    end.

replicate_rpc(DbName, Target) ->
    rexi:reply(
        try
            Opts = [{batch_size, ?BATCH_SIZE}, {batch_count, all}],
            {ok, mem3_rep:go(DbName, Target, Opts)}
        catch
            Tag:Error ->
                {Tag, Error}
        end
    ).

%% @doc Return the sequence where two files with the same UUID diverged.
compare_epochs(SourceEpochs, TargetEpochs) ->
    compare_rev_epochs(
        lists:reverse(SourceEpochs),
        lists:reverse(TargetEpochs)
    ).

compare_rev_epochs([{Node, Seq} | SourceRest], [{Node, Seq} | TargetRest]) ->
    % Common history, fast-forward
    compare_epochs(SourceRest, TargetRest);
compare_rev_epochs([], [{_, TargetSeq} | _]) ->
    % Source has not moved, start from seq just before the target took over
    TargetSeq - 1;
compare_rev_epochs([{_, SourceSeq} | _], []) ->
    % Target has not moved, start from seq where source diverged
    SourceSeq;
compare_rev_epochs([{_, SourceSeq} | _], [{_, TargetSeq} | _]) ->
    % The source was moved to a new location independently, take the minimum
    erlang:min(SourceSeq, TargetSeq) - 1.

%% @doc This adds a new update sequence checkpoint to the replication
%%      history. Checkpoints are keyed by the source node so that we
%%      aren't mixing history between source shard moves.
add_checkpoint({Props}, {History}) ->
    % Extract the source and target seqs for reference
    SourceSeq = couch_util:get_value(<<"source_seq">>, Props),
    TargetSeq = couch_util:get_value(<<"target_seq">>, Props),

    % Get the history relevant to the source node.
    SourceNode = couch_util:get_value(<<"source_node">>, Props),
    SourceHistory = couch_util:get_value(SourceNode, History, []),

    % If either the source or target shard has been truncated
    % we need to filter out any history that was stored for
    % any larger update seq than we're currently recording.
    FilteredHistory = filter_history(SourceSeq, TargetSeq, SourceHistory),

    % Re-bucket our history based on the most recent source
    % sequence. This is where we drop old checkpoints to
    % maintain the exponential distribution.
    {_, RebucketedHistory} = rebucket(FilteredHistory, SourceSeq, 0),
    NewSourceHistory = [{Props} | RebucketedHistory],

    % Finally update the source node history and we're done.
    NodeRemoved = lists:keydelete(SourceNode, 1, History),
    {[{SourceNode, NewSourceHistory} | NodeRemoved]}.

filter_history(SourceSeqThresh, TargetSeqThresh, History) ->
    SourceFilter = fun({Entry}) ->
        SourceSeq = couch_util:get_value(<<"source_seq">>, Entry),
        SourceSeq < SourceSeqThresh
    end,
    TargetFilter = fun({Entry}) ->
        TargetSeq = couch_util:get_value(<<"target_seq">>, Entry),
        TargetSeq < TargetSeqThresh
    end,
    SourceFiltered = lists:filter(SourceFilter, History),
    lists:filter(TargetFilter, SourceFiltered).

%% @doc This function adjusts our history to maintain a
%% history of checkpoints that follow an exponentially
%% increasing age from the most recent checkpoint.
%%
%% The terms newest and oldest used in these comments
%% refers to the (NewSeq - CurSeq) difference where smaller
%% values are considered newer.
%%
%% It works by assigning each entry to a bucket and keeping
%% the newest and oldest entry in each bucket. Keeping
%% both the newest and oldest means that we won't end up
%% with empty buckets as checkpoints are promoted to new
%% buckets.
%%
%% The return value of this function is a two-tuple of the
%% form `{BucketId, History}` where BucketId is the id of
%% the bucket for the first entry in History. This is used
%% when recursing to detect the oldest value in a given
%% bucket.
%%
%% This function expects the provided history to be sorted
%% in descending order of source_seq values.
rebucket([], _NewSeq, Bucket) ->
    {Bucket + 1, []};
rebucket([{Entry} | RestHistory], NewSeq, Bucket) ->
    CurSeq = couch_util:get_value(<<"source_seq">>, Entry),
    case find_bucket(NewSeq, CurSeq, Bucket) of
        Bucket ->
            % This entry is in an existing bucket which means
            % we will only keep it if its the oldest value
            % in the bucket. To detect this we rebucket the
            % rest of the list and only include Entry if the
            % rest of the list is in a bigger bucket.
            case rebucket(RestHistory, NewSeq, Bucket) of
                {Bucket, NewHistory} ->
                    % There's another entry in this bucket so we drop the
                    % current entry.
                    {Bucket, NewHistory};
                {NextBucket, NewHistory} when NextBucket > Bucket ->
                    % The rest of the history was rebucketed into a larger
                    % bucket so this is the oldest entry in the current
                    % bucket.
                    {Bucket, [{Entry} | NewHistory]}
            end;
        NextBucket when NextBucket > Bucket ->
            % This entry is the newest in NextBucket so we add it
            % to our history and continue rebucketing.
            {_, NewHistory} = rebucket(RestHistory, NewSeq, NextBucket),
            {NextBucket, [{Entry} | NewHistory]}
    end.

%% @doc Find the bucket id for the given sequence pair.
find_bucket(NewSeq, CurSeq, Bucket) ->
    % The +1 constant in this comparison is a bit subtle. The
    % reason for it is to make sure that the first entry in
    % the history is guaranteed to have a BucketId of 1. This
    % also relies on never having a duplicated update
    % sequence so adding 1 here guarantees a difference >= 2.
    if
        (NewSeq - CurSeq + 1) > (2 bsl Bucket) ->
            find_bucket(NewSeq, CurSeq, Bucket + 1);
        true ->
            Bucket
    end.

rexi_call(Node, MFA) ->
    rexi_call(Node, MFA, ?REXI_CALL_TIMEOUT_MSEC).

rexi_call(Node, MFA, Timeout) ->
    Mon = rexi_monitor:start([rexi_utils:server_pid(Node)]),
    Ref = rexi:cast(Node, self(), MFA, [sync]),
    try
        receive
            {Ref, {ok, Reply}} ->
                Reply;
            {Ref, Error} ->
                erlang:error(Error);
            {rexi_DOWN, Mon, _, Reason} ->
                erlang:error({rexi_DOWN, {Node, Reason}})
        after Timeout ->
            erlang:error(timeout)
        end
    after
        rexi_monitor:stop(Mon)
    end.

get_or_create_db(DbName, Options) ->
    mem3_util:get_or_create_db_int(DbName, Options).

-ifdef(TEST).
-include_lib("eunit/include/eunit.hrl").

-define(SNODE, <<"src@localhost">>).
-define(TNODE, <<"tgt@localhost">>).
-define(SNODE_KV, {<<"source_node">>, ?SNODE}).
-define(TNODE_KV, {<<"target_node">>, ?TNODE}).
-define(SSEQ, <<"source_seq">>).
-define(TSEQ, <<"target_seq">>).
-define(ENTRY(S, T), {[?SNODE_KV, {?SSEQ, S}, ?TNODE_KV, {?TSEQ, T}]}).

filter_history_data() ->
    [
        ?ENTRY(13, 15),
        ?ENTRY(10, 9),
        ?ENTRY(2, 3)
    ].

filter_history_remove_none_test() ->
    ?assertEqual(filter_history(20, 20, filter_history_data()), [
        ?ENTRY(13, 15),
        ?ENTRY(10, 9),
        ?ENTRY(2, 3)
    ]).

filter_history_remove_all_test() ->
    ?assertEqual(filter_history(1, 1, filter_history_data()), []).

filter_history_remove_equal_test() ->
    ?assertEqual(filter_history(10, 10, filter_history_data()), [
        ?ENTRY(2, 3)
    ]),
    ?assertEqual(filter_history(11, 9, filter_history_data()), [
        ?ENTRY(2, 3)
    ]).

filter_history_remove_for_source_and_target_test() ->
    ?assertEqual(filter_history(11, 20, filter_history_data()), [
        ?ENTRY(10, 9),
        ?ENTRY(2, 3)
    ]),
    ?assertEqual(filter_history(14, 14, filter_history_data()), [
        ?ENTRY(10, 9),
        ?ENTRY(2, 3)
    ]).

filter_history_remove_for_both_test() ->
    ?assertEqual(filter_history(11, 11, filter_history_data()), [
        ?ENTRY(10, 9),
        ?ENTRY(2, 3)
    ]).

filter_history_remove_for_both_again_test() ->
    ?assertEqual(filter_history(3, 4, filter_history_data()), [
        ?ENTRY(2, 3)
    ]).

add_first_checkpoint_test() ->
    History = {[]},
    ?assertEqual(
        add_checkpoint(?ENTRY(2, 3), History),
        {[
            {?SNODE, [
                ?ENTRY(2, 3)
            ]}
        ]}
    ).

add_first_checkpoint_to_empty_test() ->
    History = {[{?SNODE, []}]},
    ?assertEqual(
        add_checkpoint(?ENTRY(2, 3), History),
        {[
            {?SNODE, [
                ?ENTRY(2, 3)
            ]}
        ]}
    ).

add_second_checkpoint_test() ->
    History = {[{?SNODE, [?ENTRY(2, 3)]}]},
    ?assertEqual(
        add_checkpoint(?ENTRY(10, 9), History),
        {[
            {?SNODE, [
                ?ENTRY(10, 9),
                ?ENTRY(2, 3)
            ]}
        ]}
    ).

add_third_checkpoint_test() ->
    History =
        {[
            {?SNODE, [
                ?ENTRY(10, 9),
                ?ENTRY(2, 3)
            ]}
        ]},
    ?assertEqual(
        add_checkpoint(?ENTRY(11, 10), History),
        {[
            {?SNODE, [
                ?ENTRY(11, 10),
                ?ENTRY(10, 9),
                ?ENTRY(2, 3)
            ]}
        ]}
    ).

add_fourth_checkpoint_test() ->
    History =
        {[
            {?SNODE, [
                ?ENTRY(11, 10),
                ?ENTRY(10, 9),
                ?ENTRY(2, 3)
            ]}
        ]},
    ?assertEqual(
        add_checkpoint(?ENTRY(12, 13), History),
        {[
            {?SNODE, [
                ?ENTRY(12, 13),
                ?ENTRY(11, 10),
                ?ENTRY(10, 9),
                ?ENTRY(2, 3)
            ]}
        ]}
    ).

add_checkpoint_with_replacement_test() ->
    History =
        {[
            {?SNODE, [
                ?ENTRY(12, 13),
                ?ENTRY(11, 10),
                ?ENTRY(10, 9),
                ?ENTRY(2, 3)
            ]}
        ]},
    % Picking a source_seq of 16 to force 10, 11, and 12
    % into the same bucket to show we drop the 11 entry.
    ?assertEqual(
        add_checkpoint(?ENTRY(16, 16), History),
        {[
            {?SNODE, [
                ?ENTRY(16, 16),
                ?ENTRY(12, 13),
                ?ENTRY(10, 9),
                ?ENTRY(2, 3)
            ]}
        ]}
    ).

add_checkpoint_drops_redundant_checkpoints_test() ->
    % I've added comments showing the bucket ID based
    % on the ?ENTRY passed to add_checkpoint
    History =
        {[
            {?SNODE, [
                % Bucket 0
                ?ENTRY(15, 15),
                % Bucket 1
                ?ENTRY(14, 14),
                % Bucket 1
                ?ENTRY(13, 13),
                % Bucket 2
                ?ENTRY(12, 12),
                % Bucket 2
                ?ENTRY(11, 11),
                % Bucket 2
                ?ENTRY(10, 10),
                % Bucket 2
                ?ENTRY(9, 9),
                % Bucket 3
                ?ENTRY(8, 8),
                % Bucket 3
                ?ENTRY(7, 7),
                % Bucket 3
                ?ENTRY(6, 6),
                % Bucket 3
                ?ENTRY(5, 5),
                % Bucket 3
                ?ENTRY(4, 4),
                % Bucket 3
                ?ENTRY(3, 3),
                % Bucket 3
                ?ENTRY(2, 2),
                % Bucket 3
                ?ENTRY(1, 1)
            ]}
        ]},
    ?assertEqual(
        add_checkpoint(?ENTRY(16, 16), History),
        {[
            {?SNODE, [
                % Bucket 0
                ?ENTRY(16, 16),
                % Bucket 0
                ?ENTRY(15, 15),
                % Bucket 1
                ?ENTRY(14, 14),
                % Bucket 1
                ?ENTRY(13, 13),
                % Bucket 2
                ?ENTRY(12, 12),
                % Bucket 2
                ?ENTRY(9, 9),
                % Bucket 3
                ?ENTRY(8, 8),
                % Bucket 3
                ?ENTRY(1, 1)
            ]}
        ]}
    ).

add_checkpoint_show_not_always_a_drop_test() ->
    % Depending on the edge conditions of buckets we
    % may not always drop values when adding new
    % checkpoints. In this case 12 stays because there's
    % no longer a value for 10 or 11.
    %
    % I've added comments showing the bucket ID based
    % on the ?ENTRY passed to add_checkpoint
    History =
        {[
            {?SNODE, [
                % Bucket 0
                ?ENTRY(16, 16),
                % Bucket 1
                ?ENTRY(15, 15),
                % Bucket 1
                ?ENTRY(14, 14),
                % Bucket 2
                ?ENTRY(13, 13),
                % Bucket 2
                ?ENTRY(12, 12),
                % Bucket 3
                ?ENTRY(9, 9),
                % Bucket 3
                ?ENTRY(8, 8),
                % Bucket 4
                ?ENTRY(1, 1)
            ]}
        ]},
    ?assertEqual(
        add_checkpoint(?ENTRY(17, 17), History),
        {[
            {?SNODE, [
                % Bucket 0
                ?ENTRY(17, 17),
                % Bucket 0
                ?ENTRY(16, 16),
                % Bucket 1
                ?ENTRY(15, 15),
                % Bucket 1
                ?ENTRY(14, 14),
                % Bucket 2
                ?ENTRY(13, 13),
                % Bucket 2
                ?ENTRY(12, 12),
                % Bucket 3
                ?ENTRY(9, 9),
                % Bucket 3
                ?ENTRY(8, 8),
                % Bucket 4
                ?ENTRY(1, 1)
            ]}
        ]}
    ).

add_checkpoint_big_jump_show_lots_drop_test() ->
    % I've added comments showing the bucket ID based
    % on the ?ENTRY passed to add_checkpoint
    History =
        {[
            {?SNODE, [
                % Bucket 4
                ?ENTRY(16, 16),
                % Bucket 4
                ?ENTRY(15, 15),
                % Bucket 4
                ?ENTRY(14, 14),
                % Bucket 4
                ?ENTRY(13, 13),
                % Bucket 4
                ?ENTRY(12, 12),
                % Bucket 4
                ?ENTRY(9, 9),
                % Bucket 4
                ?ENTRY(8, 8),
                % Bucket 4
                ?ENTRY(1, 1)
            ]}
        ]},
    ?assertEqual(
        add_checkpoint(?ENTRY(32, 32), History),
        {[
            {?SNODE, [
                % Bucket 0
                ?ENTRY(32, 32),
                % Bucket 4
                ?ENTRY(16, 16),
                % Bucket 4
                ?ENTRY(1, 1)
            ]}
        ]}
    ).

add_checkpoint_show_filter_history_test() ->
    History =
        {[
            {?SNODE, [
                ?ENTRY(16, 16),
                ?ENTRY(15, 15),
                ?ENTRY(14, 14),
                ?ENTRY(13, 13),
                ?ENTRY(12, 12),
                ?ENTRY(9, 9),
                ?ENTRY(8, 8),
                ?ENTRY(1, 1)
            ]}
        ]},
    % Drop for both
    ?assertEqual(
        add_checkpoint(?ENTRY(10, 10), History),
        {[
            {?SNODE, [
                ?ENTRY(10, 10),
                ?ENTRY(9, 9),
                ?ENTRY(8, 8),
                ?ENTRY(1, 1)
            ]}
        ]}
    ),
    % Drop four source
    ?assertEqual(
        add_checkpoint(?ENTRY(10, 200), History),
        {[
            {?SNODE, [
                ?ENTRY(10, 200),
                ?ENTRY(9, 9),
                ?ENTRY(8, 8),
                ?ENTRY(1, 1)
            ]}
        ]}
    ),
    % Drop for target. Obviously a source_seq of 200
    % will end up droping the 8 entry.
    ?assertEqual(
        add_checkpoint(?ENTRY(200, 10), History),
        {[
            {?SNODE, [
                ?ENTRY(200, 10),
                ?ENTRY(9, 9),
                ?ENTRY(1, 1)
            ]}
        ]}
    ).

add_checkpoint_from_other_node_test() ->
    History =
        {[
            {<<"not_the_source">>, [
                ?ENTRY(12, 13),
                ?ENTRY(11, 10),
                ?ENTRY(10, 9),
                ?ENTRY(2, 3)
            ]}
        ]},
    % No filtering
    ?assertEqual(
        add_checkpoint(?ENTRY(1, 1), History),
        {[
            {?SNODE, [
                ?ENTRY(1, 1)
            ]},
            {<<"not_the_source">>, [
                ?ENTRY(12, 13),
                ?ENTRY(11, 10),
                ?ENTRY(10, 9),
                ?ENTRY(2, 3)
            ]}
        ]}
    ),
    % No dropping
    ?assertEqual(
        add_checkpoint(?ENTRY(200, 200), History),
        {[
            {?SNODE, [
                ?ENTRY(200, 200)
            ]},
            {<<"not_the_source">>, [
                ?ENTRY(12, 13),
                ?ENTRY(11, 10),
                ?ENTRY(10, 9),
                ?ENTRY(2, 3)
            ]}
        ]}
    ).

-endif.