Always decode log entries for witnesses so they can be stubbed

Summary: as title

Reviewed By: hsun324

Differential Revision: D77493707

fbshipit-source-id: f70b959fb1c3bb41dfb00878f5c994284f7c27d6

Author: Jose-Angel Herrero Bajo

src/wa_raft_log_catchup.erl

@@ -41,7 +41,7 @@
 
 %% API
 -export([
-    start_catchup_request/5,
+    start_catchup_request/6,
     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), {Peer, FollowerLastIndex, LeaderTerm, LeaderCommitIndex}).
+-define(CATCHUP_REQUEST(Peer, FollowerLastIndex, LeaderTerm, LeaderCommitIndex, Witness), {Peer, FollowerLastIndex, LeaderTerm, LeaderCommitIndex, Witness}).
 
 %% An entry in the catchup ETS table that indicates an in-progress log
 %% catchup to the specified node.
@@ -116,9 +116,10 @@ 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()) -> ok.
-start_catchup_request(Catchup, Peer, FollowerLastIndex, LeaderTerm, LeaderCommitIndex) ->
-    ets:insert(Catchup, ?CATCHUP_REQUEST(Peer, FollowerLastIndex, LeaderTerm, LeaderCommitIndex)),
+                            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)),
     ok.
 
 %% Cancel a request to trigger log catchup for a particular follower.
@@ -206,8 +207,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) = lists:nth(rand:uniform(length(Requests)), Requests),
-            NewState = send_logs(Peer, FollowerLastIndex, LeaderTerm, LeaderCommitIndex, State),
+            ?CATCHUP_REQUEST(Peer, FollowerLastIndex, LeaderTerm, LeaderCommitIndex, Witness) = lists:nth(rand:uniform(length(Requests)), Requests),
+            NewState = send_logs(Peer, FollowerLastIndex, LeaderTerm, LeaderCommitIndex, Witness, State),
             {noreply, NewState, ?CONTINUE_TIMEOUT}
     end;
 handle_info(Info, #state{name = Name} = State) ->
@@ -222,8 +223,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(), #state{}) -> #state{}.
-send_logs(Peer, NextLogIndex, LeaderTerm, LeaderCommitIndex, #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(), boolean(), #state{}) -> #state{}.
+send_logs(Peer, NextLogIndex, LeaderTerm, LeaderCommitIndex, Witness, #state{name = Name, lockouts = Lockouts} = State) ->
     StartMillis = erlang:system_time(millisecond),
     LockoutMillis = maps:get(Peer, Lockouts, 0),
     NewState = case LockoutMillis =< StartMillis of
@@ -233,7 +234,7 @@ send_logs(Peer, NextLogIndex, LeaderTerm, LeaderCommitIndex, #state{name = Name,
                 true ->
                     counters:add(Counters, ?COUNTER_CONCURRENT_CATCHUP, 1),
                     ets:insert(?MODULE, ?CATCHUP_RECORD(Name, Peer)),
-                    try send_logs_impl(Peer, NextLogIndex, LeaderTerm, LeaderCommitIndex, State) catch
+                    try send_logs_impl(Peer, NextLogIndex, LeaderTerm, LeaderCommitIndex, Witness, 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",
@@ -256,15 +257,22 @@ send_logs(Peer, NextLogIndex, LeaderTerm, LeaderCommitIndex, #state{name = Name,
     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(), #state{}) -> term().
-send_logs_impl(#raft_identity{node = PeerNode} = Peer, NextLogIndex, LeaderTerm, LeaderCommitIndex,
+-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,
                #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),
     LogBatchBytes = ?RAFT_CATCHUP_MAX_BYTES_PER_BATCH(App),
-    {ok, Entries} = wa_raft_log:entries(Log, NextLogIndex, LogBatchEntries, LogBatchBytes),
+    Entries = case Witness of
+        true ->
+            {ok, RawEntries} = wa_raft_log:get(Log, NextLogIndex, LogBatchEntries, LogBatchBytes),
+            wa_raft_server:stub_entries_for_witness(RawEntries);
+        false ->
+            {ok, RawEntries} = wa_raft_log:entries(Log, NextLogIndex, LogBatchEntries, LogBatchBytes),
+            RawEntries
+    end,
 
     case Entries of
         [] ->
@@ -278,7 +286,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, FollowerMatchIndex, _)} ->
-                    send_logs_impl(Peer, FollowerMatchIndex + 1, LeaderTerm, LeaderCommitIndex, State);
+                    send_logs_impl(Peer, FollowerMatchIndex + 1, LeaderTerm, LeaderCommitIndex, Witness, State);
                 {LeaderTerm, _, ?APPEND_ENTRIES_RESPONSE(PrevLogIndex, false, _FollowerLastIndex, _)} ->
                     exit(append_failed);
                 {LeaderTerm, _, Other} ->

src/wa_raft_server.erl

@@ -49,6 +49,11 @@
     normalize_config/1
 ]).
 
+% Stubbing log entries for witnesses
+-export([
+    stub_entries_for_witness/1
+]).
+
 %% Modification of cluster configuration
 -export([
     set_config_members/2,
@@ -2722,11 +2727,12 @@ heartbeat(
             MaxLogEntries = ?RAFT_HEARTBEAT_MAX_ENTRIES(App),
             MaxHeartbeatSize = ?RAFT_HEARTBEAT_MAX_BYTES(App),
             Witnesses = config_witnesses(config(State0)),
-            {ok, RawEntries} = wa_raft_log:entries(View, FollowerNextIndex, MaxLogEntries, MaxHeartbeatSize),
             Entries = case lists:member({Name, FollowerId}, Witnesses) of
                 true ->
+                    {ok, RawEntries} = wa_raft_log:get(View, FollowerNextIndex, MaxLogEntries, MaxHeartbeatSize),
                     stub_entries_for_witness(RawEntries);
                 false ->
+                    {ok, RawEntries} = wa_raft_log:entries(View, FollowerNextIndex, MaxLogEntries, MaxHeartbeatSize),
                     RawEntries
             end,
             {ok, PrevLogTerm} = PrevLogTermRes,
@@ -3293,16 +3299,18 @@ request_snapshot_for_follower(
     State :: #raft_state{}
 ) -> ok.
 request_bulk_logs_for_follower(
-    Peer,
+    ?IDENTITY_REQUIRES_MIGRATION(_, FollowerId) = Peer,
     FollowerLastIndex,
     #raft_state{
+        name = Name,
         catchup = Catchup,
         current_term = CurrentTerm,
         commit_index = CommitIndex
     } = State
 ) ->
     ?RAFT_LOG_DEBUG(leader, State, "requesting bulk logs catchup for follower ~0p.", [Peer]),
-    wa_raft_log_catchup:start_catchup_request(Catchup, Peer, FollowerLastIndex, CurrentTerm, CommitIndex).
+    Witness = lists:member({Name, FollowerId}, config_witnesses(config(State))),
+    wa_raft_log_catchup:start_catchup_request(Catchup, Peer, FollowerLastIndex, CurrentTerm, CommitIndex, Witness).
 
 -spec cancel_bulk_logs_for_follower(Peer :: #raft_identity{}, State :: #raft_state{}) -> ok.
 cancel_bulk_logs_for_follower(Peer, #raft_state{catchup = Catchup}) ->