suspension fixes

Author: kjnilsson

src/ra_server.erl

@@ -52,6 +52,7 @@
          checkpoint/3,
          persist_last_applied/1,
          peers/1,
+         peer_status/2,
          update_peer/3,
          update_disconnected_peers/3,
          handle_down/5,
@@ -2548,6 +2549,14 @@ new_peer_with(Map) ->
 peers(#{cfg := #cfg{id = Id}, cluster := Peers}) ->
     maps:remove(Id, Peers).
 
+peer_status(PeerId, #{cluster := Peers}) ->
+    case Peers of
+        #{PeerId := #{status := Status}} ->
+            Status;
+        _ ->
+            undefined
+    end.
+
 %% remove any peers that are currently receiving a snapshot
 peers_with_normal_status(State) ->
     maps:filter(fun (_, #{status := normal}) -> true;

src/ra_server_proc.erl

@@ -579,8 +579,13 @@ leader(info, {Status, Node, InfoList}, State0)
   when Status =:= nodedown orelse
        Status =:= nodeup ->
     handle_node_status_change(Node, Status, InfoList, ?FUNCTION_NAME, State0);
-leader(info, {update_peer, PeerId, Update}, State0) ->
-    State = update_peer(PeerId, Update, State0),
+leader(info, {unsuspend_peer, PeerId}, State0) ->
+    State = case ra_server:peer_status(PeerId, State0#state.server_state) of
+                suspended ->
+                    update_peer(PeerId, #{status => normal}, State0);
+                _ ->
+                    State0
+            end,
     {keep_state, State, []};
 leader(_, tick_timeout, State0) ->
     {State1, RpcEffs} = make_rpcs(State0),
@@ -1393,13 +1398,15 @@ handle_effects(RaftState, Effects0, EvtType, State0, Actions0) ->
     {State, lists:reverse(Actions)}.
 
 handle_effect(_RaftState, {send_rpc, To, Rpc}, _,
-              #state{conf = Conf} = State0, Actions) ->
+              #state{conf = Conf,
+                     server_state = SS} = State0, Actions) ->
     % fully qualified use only so that we can mock it for testing
     % TODO: review / refactor to remove the mod call here
+    PeerStatus = ra_server:peer_status(To, SS),
     case ?MODULE:send_rpc(To, Rpc, State0) of
         ok ->
             {State0, Actions};
-        nosuspend ->
+        nosuspend when PeerStatus == normal ->
             %% update peer status to suspended and spawn a process
             %% to send the rpc without nosuspend so that it will block until
             %% the data can get through
@@ -1410,11 +1417,13 @@ handle_effect(_RaftState, {send_rpc, To, Rpc}, _,
                                  %% the peer status back to normal
                                  ok = gen_statem:cast(To, Rpc),
                                  incr_counter(Conf, ?C_RA_SRV_MSGS_SENT, 1),
-                                 Self ! {update_peer, To, #{status => normal}}
+                                 Self ! {unsuspend_peer, To}
                          end),
-            ?DEBUG("~ts: temporarily suspending peer ~w due to full distribution buffer",
-                   [log_id(State0), To]),
+            % ?DEBUG("~ts: temporarily suspending peer ~w due to full distribution buffer ~W",
+            %        [log_id(State0), To, Rpc, 5]),
             {update_peer(To, #{status => suspended}, State0), Actions};
+        nosuspend ->
+            {State0, Actions};
         noconnect ->
             %% for noconnects just allow it to pipeline and catch up later
             {State0, Actions}
@@ -1976,6 +1985,8 @@ send_snapshots(Id, Term, {_, ToNode} = To, ChunkSize,
             Result = read_chunks_and_send_rpc(RPC, To, ReadState, 1,
                                               ChunkSize, InstallTimeout,
                                               SnapState),
+            ?DEBUG("~ts: sending snapshot to ~w completed",
+                   [LogId, To]),
             ok = gen_statem:cast(Id, {To, Result})
     end.
 

test/ra_kv_SUITE.erl

@@ -71,8 +71,8 @@ snapshot_replication_interrupted(_Config) ->
     Data = crypto:strong_rand_bytes(100_000),
     %% write 10k entries of the same key
     [{ok, #{}} = ra_kv:put(KvId, term_to_binary(I), Data, 5000)
-     || I <- lists:seq(1, 50_000)],
-    ?assertMatch({ok, #{machine := #{num_keys := 50_000}}, KvId},
+     || I <- lists:seq(1, 10_000)],
+    ?assertMatch({ok, #{machine := #{num_keys := 10_000}}, KvId},
                  ra:member_overview(KvId)),
 
     ra_log_wal:force_roll_over(ra_log_wal),
@@ -99,7 +99,7 @@ snapshot_replication_interrupted(_Config) ->
     ct:pal("ra_state ~p", [ets:tab2list(ra_state)]),
     ok = ra:stop_server(?SYS, KvId3),
     [{ok, #{}} = ra_kv:put(KvId, term_to_binary(I), Data, 5000)
-     || I <- lists:seq(50_001, 50_010)],
+     || I <- lists:seq(10_001, 10_010)],
     ok = ra:restart_server(?SYS, KvId3),
     {ok, #{log := #{last_index := Kv1LastIndex  }}, _} = ra:member_overview(KvId),
     ok = ra_lib:retry(

test/ra_server_SUITE.erl

@@ -273,7 +273,7 @@ recover_restores_cluster_changes(_Config) ->
 
     % n2 joins
     {leader, #{cluster := Cluster,
-               log := Log0}, _} =
+               log := Log0}, _Effs} =
         ra_server:handle_leader({command, {'$ra_join', meta(),
                                            N2, await_consensus}}, State),
     {LIdx, _} = ra_log:last_index_term(Log0),