transfer full log entries to witnesses

Reviewed By: jaher

Differential Revision:
D72331130

Privacy Context Container: L1179610

fbshipit-source-id: cb74b67b4ffd11f273e1bf8ae1877beeddf42b4d

Author: Sarah Hassan

src/wa_raft_log.erl

@@ -30,13 +30,11 @@
     fold/5,
     fold/6,
 
-    fold_terms/5,
 
     term/2,
     get/2,
     get/3,
     get/4,
-    get_terms/3,
 
     config/1
 ]).
@@ -437,45 +435,6 @@ fold_impl(Log, First, Last, SizeLimit, Func, AccIn) ->
             {error, Reason}
     end.
 
-%% Folds over the terms in the log view of raw entries from the log provider
-%% between the provided first and last log indices (inclusive).
-%% If there exists a log term between the provided first and last indices then
-%% the accumulator function will be called on at least that term.
-%% This API provides no validation of the log indices and term passed by the
-%% provider to the callback function.
--spec fold_terms(LogOrView :: log() | view(),
-                 First :: log_index(),
-                 Last :: log_index(),
-                 Func :: fun((Index :: log_index(), Term :: log_term(), Acc) -> Acc),
-                 Acc) ->
-    {ok, Acc} | wa_raft:error().
-fold_terms(#log_view{log = Log, first = LogFirst, last = LogLast}, First, Last, Func, Acc) ->
-    fold_terms_impl(Log, max(First, LogFirst), min(Last, LogLast), Func, Acc);
-fold_terms(Log, First, Last, Func, Acc) ->
-    Provider = provider(Log),
-    LogFirst = Provider:first_index(Log),
-    LogLast = Provider:last_index(Log),
-    fold_terms_impl(Log, max(First, LogFirst), min(Last, LogLast), Func, Acc).
-
--spec fold_terms_impl(
-    Log :: log(),
-    First :: log_index(),
-    Last :: log_index(),
-    Func :: fun((Index :: log_index(), Term :: log_term(), Acc) -> Acc),
-    Acc :: term()
-) -> {ok, Acc} | wa_raft:error().
-fold_terms_impl(Log, First, Last, Func, AccIn) ->
-    ?RAFT_COUNT('raft.log.fold_terms'),
-    ?RAFT_COUNTV('raft.log.fold_terms.total', Last - First + 1),
-    Provider = provider(Log),
-    case Provider:fold_terms(Log, First, Last, Func, AccIn) of
-        {ok, AccOut} ->
-            {ok, AccOut};
-        {error, Reason} ->
-            ?RAFT_COUNT('raft.log.fold_terms.error'),
-            {error, Reason}
-    end.
-
 %% Gets the term of entry at the provided log index. When using a log view
 %% this function may return 'not_found' even if the underlying log entry still
 %% exists if the entry is outside of the log view.
@@ -534,25 +493,6 @@ get(LogOrView, First, CountLimit, SizeLimit) ->
             {error, corruption}
     end.
 
--spec get_terms(LogOrView :: log() | view(), First :: log_index(), Limit :: non_neg_integer()) ->
-    {ok, Terms :: [wa_raft_log:log_term()]} | wa_raft:error().
-get_terms(LogOrView, First, Limit) ->
-    try
-        fold_terms(LogOrView, First, First + Limit - 1,
-            fun
-                (Index, Term, {Index, Acc})            -> {Index + 1, [Term | Acc]};
-                (_Index, _Term, {ExpectedIndex, _Acc}) -> throw({missing, ExpectedIndex})
-            end, {First, []})
-    of
-        {ok, {_, TermsRev}} -> {ok, lists:reverse(TermsRev)};
-        {error, Reason} -> {error, Reason}
-    catch
-        throw:{missing, Index} ->
-            ?LOG_WARNING("[~p] detected log is missing index ~p during get of ~p ~~ ~p",
-                [log_name(LogOrView), Index, First, First + Limit - 1], #{domain => [whatsapp, wa_raft]}),
-            {error, corruption}
-    end.
-
 -spec config(LogOrView :: log() | view()) -> {ok, Index :: log_index(), Config :: wa_raft_server:config() | undefined} | not_found.
 config(#log_view{config_index = undefined}) ->
     not_found;

src/wa_raft_log_catchup.erl

@@ -41,7 +41,7 @@
 
 %% API
 -export([
-    start_catchup_request/6,
+    start_catchup_request/5,
     cancel_catchup_request/2,
     is_catching_up/2
 ]).
@@ -81,7 +81,7 @@
 
 %% An entry in the catchup request ETS table representing a request to
 %% trigger log catchup for a particular peer.
--define(CATCHUP_REQUEST(Peer, FollowerLastIndex, LeaderTerm, LeaderCommitIndex, Witness), {Peer, FollowerLastIndex, LeaderTerm, LeaderCommitIndex, Witness}).
+-define(CATCHUP_REQUEST(Peer, FollowerLastIndex, LeaderTerm, LeaderCommitIndex), {Peer, FollowerLastIndex, LeaderTerm, LeaderCommitIndex}).
 
 %% An entry in the catchup ETS table that indicates an in-progress log
 %% catchup to the specified node.
@@ -116,9 +116,9 @@ start_link(#raft_options{log_catchup_name = Name} = Options) ->
 
 %% Submit a request to trigger log catchup for a particular follower starting at the index provided.
 -spec start_catchup_request(Catchup :: atom(), Peer :: #raft_identity{}, FollowerLastIndex :: wa_raft_log:log_index(),
-                            LeaderTerm :: wa_raft_log:log_term(), LeaderCommitIndex :: wa_raft_log:log_index(), Witness :: boolean()) -> ok.
-start_catchup_request(Catchup, Peer, FollowerLastIndex, LeaderTerm, LeaderCommitIndex, Witness) ->
-    ets:insert(Catchup, ?CATCHUP_REQUEST(Peer, FollowerLastIndex, LeaderTerm, LeaderCommitIndex, Witness)),
+                            LeaderTerm :: wa_raft_log:log_term(), LeaderCommitIndex :: wa_raft_log:log_index()) -> ok.
+start_catchup_request(Catchup, Peer, FollowerLastIndex, LeaderTerm, LeaderCommitIndex) ->
+    ets:insert(Catchup, ?CATCHUP_REQUEST(Peer, FollowerLastIndex, LeaderTerm, LeaderCommitIndex)),
     ok.
 
 %% Cancel a request to trigger log catchup for a particular follower.
@@ -206,8 +206,8 @@ handle_info(timeout, #state{name = Name} = State) ->
             {noreply, State, ?IDLE_TIMEOUT};
         Requests ->
             % Select a random log catchup request to process.
-            ?CATCHUP_REQUEST(Peer, FollowerLastIndex, LeaderTerm, LeaderCommitIndex, Witness) = lists:nth(rand:uniform(length(Requests)), Requests),
-            NewState = send_logs(Peer, FollowerLastIndex, LeaderTerm, LeaderCommitIndex, Witness, State),
+            ?CATCHUP_REQUEST(Peer, FollowerLastIndex, LeaderTerm, LeaderCommitIndex) = lists:nth(rand:uniform(length(Requests)), Requests),
+            NewState = send_logs(Peer, FollowerLastIndex, LeaderTerm, LeaderCommitIndex, State),
             % erlint-ignore garbage_collect
             erlang:garbage_collect(),
             {noreply, NewState, ?CONTINUE_TIMEOUT}
@@ -224,8 +224,8 @@ terminate(_Reason, #state{name = Name}) ->
 %% Private functions - Send logs to follower
 %%
 
--spec send_logs(#raft_identity{}, wa_raft_log:log_index(), wa_raft_log:log_term(), wa_raft_log:log_index(), boolean(), #state{}) -> #state{}.
-send_logs(Peer, NextLogIndex, LeaderTerm, LeaderCommitIndex, Witness, #state{name = Name, lockouts = Lockouts} = State) ->
+-spec send_logs(#raft_identity{}, wa_raft_log:log_index(), wa_raft_log:log_term(), wa_raft_log:log_index(), #state{}) -> #state{}.
+send_logs(Peer, NextLogIndex, LeaderTerm, LeaderCommitIndex, #state{name = Name, lockouts = Lockouts} = State) ->
     StartMillis = erlang:system_time(millisecond),
     LockoutMillis = maps:get(Peer, Lockouts, 0),
     NewState = case LockoutMillis =< StartMillis of
@@ -235,7 +235,7 @@ send_logs(Peer, NextLogIndex, LeaderTerm, LeaderCommitIndex, Witness, #state{nam
                 true ->
                     counters:add(Counters, ?COUNTER_CONCURRENT_CATCHUP, 1),
                     ets:insert(?MODULE, ?CATCHUP_RECORD(Name, Peer)),
-                    try send_logs_impl(Peer, NextLogIndex, LeaderTerm, LeaderCommitIndex, Witness, State) catch
+                    try send_logs_impl(Peer, NextLogIndex, LeaderTerm, LeaderCommitIndex, State) catch
                         T:E:S ->
                             ?RAFT_COUNT('raft.catchup.error'),
                             ?LOG_ERROR("Catchup[~p, term ~p] bulk logs transfer to ~0p failed with ~0p ~0p at ~p",
@@ -258,22 +258,15 @@ send_logs(Peer, NextLogIndex, LeaderTerm, LeaderCommitIndex, Witness, #state{nam
     ets:delete(?MODULE, Name),
     NewState.
 
--spec send_logs_impl(#raft_identity{}, wa_raft_log:log_index(), wa_raft_log:log_term(), wa_raft_log:log_index(), boolean(), #state{}) -> term().
-send_logs_impl(#raft_identity{node = PeerNode} = Peer, NextLogIndex, LeaderTerm, LeaderCommitIndex, Witness,
+-spec send_logs_impl(#raft_identity{}, wa_raft_log:log_index(), wa_raft_log:log_term(), wa_raft_log:log_index(), #state{}) -> term().
+send_logs_impl(#raft_identity{node = PeerNode} = Peer, NextLogIndex, LeaderTerm, LeaderCommitIndex,
                #state{application = App, name = Name, self = Self, identifier = Identifier, distribution_module = DistributionModule, server_name = Server, log = Log} = State) ->
     PrevLogIndex = NextLogIndex - 1,
     {ok, PrevLogTerm} = wa_raft_log:term(Log, PrevLogIndex),
 
     LogBatchEntries = ?RAFT_CATCHUP_MAX_ENTRIES_PER_BATCH(App),
-    Entries = case Witness of
-        false ->
-            LogBatchBytes = ?RAFT_CATCHUP_MAX_BYTES_PER_BATCH(App),
-            {ok, E} = wa_raft_log:get(Log, NextLogIndex, LogBatchEntries, LogBatchBytes),
-            E;
-        true ->
-            {ok, T} = wa_raft_log:get_terms(Log, NextLogIndex, min(LogBatchEntries, LeaderCommitIndex - NextLogIndex + 1)),
-            [{Term, []} || Term <- T]
-    end,
+    LogBatchBytes = ?RAFT_CATCHUP_MAX_BYTES_PER_BATCH(App),
+    {ok, Entries} = wa_raft_log:get(Log, NextLogIndex, LogBatchEntries, LogBatchBytes),
 
     case Entries of
         [] ->
@@ -287,7 +280,7 @@ send_logs_impl(#raft_identity{node = PeerNode} = Peer, NextLogIndex, LeaderTerm,
 
             try wa_raft_server:parse_rpc(Self, DistributionModule:call(Dest, Identifier, Command, Timeout)) of
                 {LeaderTerm, _, ?APPEND_ENTRIES_RESPONSE(PrevLogIndex, true, FollowerEndIndex)} ->
-                    send_logs_impl(Peer, FollowerEndIndex + 1, LeaderTerm, LeaderCommitIndex, Witness, State);
+                    send_logs_impl(Peer, FollowerEndIndex + 1, LeaderTerm, LeaderCommitIndex, State);
                 {LeaderTerm, _, ?APPEND_ENTRIES_RESPONSE(PrevLogIndex, false, _FollowerEndIndex)} ->
                     exit(append_failed);
                 {LeaderTerm, _, Other} ->

src/wa_raft_server.erl

@@ -2101,10 +2101,6 @@ max_index_to_apply(MatchIndex, LastIndex, Config) ->
 to_member_list(Mapping, Default, Config) ->
     [maps:get(Node, Mapping, Default) || {_, Node} <- config_membership(Config)].
 
--spec compute_max(Mapping :: #{node() => Value}, Default :: Value, Config :: config()) -> Max :: Value.
-compute_max(Mapping, Default, Config) ->
-    lists:max(to_member_list(Mapping, Default, Config)).
-
 %% Compute the quorum maximum value for the current membership given a config for
 %% the values represented by the given a mapping of peers (see note on config about
 %% RAFT RPC ids) to values assuming a default value for peers who are not represented
@@ -2314,8 +2310,6 @@ heartbeat(?IDENTITY_REQUIRES_MIGRATION(_, FollowerId) = Sender,
     LastFollowerHeartbeatTs = maps:get(FollowerId, LastHeartbeatTs, undefined),
     State1 = State0#raft_state{last_heartbeat_ts = LastHeartbeatTs#{FollowerId => NowTs}, leader_heartbeat_ts = NowTs},
     LastIndex = wa_raft_log:last_index(View),
-    Config = config(State0),
-    Witnesses = config_witnesses(Config),
     case PrevLogTermRes =:= not_found orelse IsCatchingUp of %% catching up, or prep
         true ->
             {ok, LastTerm} = wa_raft_log:term(View, LastIndex),
@@ -2326,20 +2320,9 @@ heartbeat(?IDENTITY_REQUIRES_MIGRATION(_, FollowerId) = Sender,
             LastFollowerHeartbeatTs =/= undefined andalso ?RAFT_GATHER('raft.leader.heartbeat.interval_ms', erlang:monotonic_time(millisecond) - LastFollowerHeartbeatTs),
             State1;
         false ->
-            Entries =
-                case lists:member({Name, FollowerId}, Witnesses) of
-                    true ->
-                        MaxWitnessLogEntries = ?RAFT_HEARTBEAT_MAX_ENTRIES_TO_WITNESS(App),
-                        MaxMatchIndex = compute_max(MatchIndex, 0, Config),
-                        Limit = max(0, min(MaxWitnessLogEntries, MaxMatchIndex - FollowerNextIndex + 1)),
-                        {ok, Terms} = wa_raft_log:get_terms(View, FollowerNextIndex, Limit),
-                        [{Term, []} || Term <- Terms];
-                    false ->
-                        MaxLogEntries = ?RAFT_HEARTBEAT_MAX_ENTRIES(App),
-                        MaxHeartbeatSize = ?RAFT_HEARTBEAT_MAX_BYTES(App),
-                        {ok, Ret} = wa_raft_log:get(View, FollowerNextIndex, MaxLogEntries, MaxHeartbeatSize),
-                        Ret
-                end,
+            MaxLogEntries = ?RAFT_HEARTBEAT_MAX_ENTRIES(App),
+            MaxHeartbeatSize = ?RAFT_HEARTBEAT_MAX_BYTES(App),
+            {ok, Entries} = wa_raft_log:get(View, FollowerNextIndex, MaxLogEntries, MaxHeartbeatSize),
             {ok, PrevLogTerm} = PrevLogTermRes,
             ?RAFT_GATHER('raft.leader.heartbeat.size', length(Entries)),
             ?LOG_DEBUG("Server[~0p, term ~0p, leader] heartbeat to follower ~p from ~p(~p entries). Commit index ~p",
@@ -2707,11 +2690,10 @@ request_snapshot_for_follower(FollowerId, #raft_state{application = App, name =
     end.
 
 -spec request_bulk_logs_for_follower(#raft_identity{}, wa_raft_log:log_index(), #raft_state{}) -> ok.
-request_bulk_logs_for_follower(#raft_identity{node = FollowerId} = Peer, FollowerEndIndex, #raft_state{name = Name, catchup = Catchup, current_term = CurrentTerm, commit_index = CommitIndex} = State) ->
+request_bulk_logs_for_follower(Peer, FollowerEndIndex, #raft_state{name = Name, catchup = Catchup, current_term = CurrentTerm, commit_index = CommitIndex}) ->
     ?LOG_DEBUG("Server[~0p, term ~0p, leader] requesting bulk logs catchup for follower ~0p.",
         [Name, CurrentTerm, Peer], #{domain => [whatsapp, wa_raft]}),
-    Witness = lists:member({Name, FollowerId}, config_witnesses(config(State))),
-    wa_raft_log_catchup:start_catchup_request(Catchup, Peer, FollowerEndIndex, CurrentTerm, CommitIndex, Witness).
+    wa_raft_log_catchup:start_catchup_request(Catchup, Peer, FollowerEndIndex, CurrentTerm, CommitIndex).
 
 -spec cancel_bulk_logs_for_follower(#raft_identity{}, #raft_state{}) -> ok.
 cancel_bulk_logs_for_follower(Peer, #raft_state{catchup = Catchup}) ->