Ensure batch size is non negative when making a pipelined AER

add assertion

add some logging and a test

add some logging and a test

Author: kjnilsson

src/ra_server.erl

@@ -480,23 +480,26 @@ handle_leader({PeerId, #append_entries_reply{term = Term, success = true,
             State1 = put_peer(PeerId, Peer, State0),
             Effects00 = maybe_promote_peer(PeerId, State1, []),
             {State2, Effects0} = evaluate_quorum(State1, Effects00),
-            {State, Effects1} = process_pending_consistent_queries(State2,
+            {State3, Effects1} = process_pending_consistent_queries(State2,
                                                                    Effects0),
-            Effects = [{next_event, info, pipeline_rpcs} | Effects1],
-            case State of
+            Effects2 = [{next_event, info, pipeline_rpcs} | Effects1],
+            case State3 of
                 #{cluster := #{Id := _}} ->
                     % leader is in the cluster
-                    {leader, State, Effects};
+                    {leader, State3, Effects2};
                 #{commit_index := CI,
                   cluster_index_term := {CITIndex, _}}
                   when CI >= CITIndex ->
                     % leader is not in the cluster and the new cluster
                     % config has been committed
                     % time to say goodbye
                     ?INFO("~ts: leader not in new cluster - goodbye", [LogId]),
+                    %% need to make pipelined rpcs here as cannot use next event
+                    {State, _, Effects} =
+                        make_pipelined_rpc_effects(State3, Effects2),
                     {stop, State, Effects};
                 _ ->
-                    {leader, State, Effects}
+                    {leader, State3, Effects2}
             end
     end;
 handle_leader({PeerId, #append_entries_reply{term = Term}},
@@ -2086,11 +2089,15 @@ make_pipelined_rpc_effects(#{cfg := #cfg{id = Id,
                                    end,
                       %% ensure we don't pass a batch size that would allow
                       %% the peer to go over the max pipeline count
-                      BatchSize = min(MaxBatchSize,
-                                      MaxPipelineCount - NumInFlight),
+                      %% we'd only really get here if Force=true so setting
+                      %% a single entry batch size should be fine
+                      BatchSize = max(1,
+                                      min(MaxBatchSize,
+                                          MaxPipelineCount - NumInFlight)),
                       {NewNextIdx, Eff, S} =
                           make_rpc_effect(PeerId, Peer0, BatchSize, S0,
                                           EntryCache),
+                      ?assert(NewNextIdx >= NextIdx),
                       Peer = Peer0#{next_index => NewNextIdx,
                                     commit_index_sent => CommitIndex},
                       NewNumInFlight = NewNextIdx - MatchIdx - 1,
@@ -2149,12 +2156,17 @@ make_rpc_effect(PeerId, #{next_index := Next}, MaxBatchSize,
                                             PrevTerm, MaxBatchSize,
                                             State#{log => Log},
                                             EntryCache);
-                {LastIdx, _} ->
+                {SnapIdx, _} ->
+                    ?DEBUG("~ts: sending snapshot to ~w as their next index ~b "
+                           "is lower than snapshot index ~b", [log_id(State),
+                                                               PeerId, Next,
+                                                               SnapIdx]),
+                    ?assert(PrevIdx < SnapIdx),
                     SnapState = ra_log:snapshot_state(Log),
                     %% don't increment the next index here as we will do
                     %% that once the snapshot is fully replicated
                     %% and we don't pipeline entries until after snapshot
-                    {LastIdx,
+                    {SnapIdx,
                      {send_snapshot, PeerId, {SnapState, Id, Term}},
                      State#{log => Log}}
             end
@@ -2861,16 +2873,16 @@ apply_with({Idx, Term, {'$ra_cluster_change', CmdMeta, NewCluster, ReplyMode}},
     State = case State0 of
                 #{cluster_index_term := {CI, CT}}
                   when Idx > CI andalso Term >= CT ->
-                    ?DEBUG("~ts: applying ra cluster change to ~w",
-                           [log_id(State0), maps:keys(NewCluster)]),
+                    ?DEBUG("~ts: applying ra cluster change at index ~b to ~w",
+                           [log_id(State0), Idx, maps:keys(NewCluster)]),
                     %% we are recovering and should apply the cluster change
                     State0#{cluster => NewCluster,
                             membership => get_membership(NewCluster, State0),
                             cluster_change_permitted => true,
                             cluster_index_term => {Idx, Term}};
                 _  ->
-                    ?DEBUG("~ts: committing ra cluster change to ~w",
-                           [log_id(State0), maps:keys(NewCluster)]),
+                    ?DEBUG("~ts: committing ra cluster change at index ~b to ~w",
+                           [log_id(State0), Idx, maps:keys(NewCluster)]),
                     %% else just enable further cluster changes again
                     State0#{cluster_change_permitted => true}
             end,
@@ -3078,17 +3090,23 @@ pre_append_log_follower({Idx, Term, Cmd} = Entry,
     % cluster
     case Cmd of
         {'$ra_cluster_change', _, Cluster, _} ->
+            ?DEBUG("~ts: ~ts: follower applying ra cluster change to ~w",
+                   [log_id(State), ?FUNCTION_NAME, maps:keys(Cluster)]),
             State#{cluster => Cluster,
                    cluster_index_term => {Idx, Term}};
         _ ->
             % revert back to previous cluster
             {PrevIdx, PrevTerm, PrevCluster} = maps:get(previous_cluster, State),
+            ?DEBUG("~ts: ~ts: follower reverting cluster change to ~w",
+                   [log_id(State), ?FUNCTION_NAME, maps:keys(PrevCluster)]),
             State1 = State#{cluster => PrevCluster,
                             cluster_index_term => {PrevIdx, PrevTerm}},
             pre_append_log_follower(Entry, State1)
     end;
 pre_append_log_follower({Idx, Term, {'$ra_cluster_change', _, Cluster, _}},
                         State) ->
+    ?DEBUG("~ts: ~ts: follower applying ra cluster change to ~w",
+           [log_id(State), ?FUNCTION_NAME, maps:keys(Cluster)]),
     State#{cluster => Cluster,
            membership => get_membership(Cluster, State),
            cluster_index_term => {Idx, Term}};

src/ra_server_proc.erl

@@ -1142,17 +1142,20 @@ handle_enter(RaftState, OldRaftState,
     {ServerState, Effects} = ra_server:handle_state_enter(RaftState,
                                                           OldRaftState,
                                                           ServerState0),
+    LastApplied = do_state_query(last_applied, State),
     case RaftState == leader orelse OldRaftState == leader of
         true ->
             %% ensure transitions from and to leader are logged at a higher
             %% level
-            ?NOTICE("~ts: ~s -> ~s in term: ~b machine version: ~b",
+            ?NOTICE("~ts: ~s -> ~s in term: ~b machine version: ~b, last applied ~b",
                     [log_id(State), OldRaftState, RaftState,
-                     current_term(State), machine_version(State)]);
+                     current_term(State), machine_version(State),
+                     LastApplied]);
         false ->
-            ?DEBUG("~ts: ~s -> ~s in term: ~b machine version: ~b",
+            ?DEBUG("~ts: ~s -> ~s in term: ~b machine version: ~b, last applied ~b",
                    [log_id(State), OldRaftState, RaftState,
-                    current_term(State), machine_version(State)])
+                    current_term(State), machine_version(State),
+                    LastApplied])
     end,
     handle_effects(RaftState, Effects, cast,
                    State#state{server_state = ServerState}).

test/coordination_SUITE.erl

@@ -43,6 +43,7 @@ all_tests() ->
      start_cluster_majority,
      start_cluster_minority,
      grow_cluster,
+     shrink_cluster_with_snapshot,
      send_local_msg,
      local_log_effect,
      leaderboard,
@@ -382,6 +383,45 @@ grow_cluster(Config) ->
     stop_peers(Peers),
     ok.
 
+shrink_cluster_with_snapshot(Config) ->
+    %% this test removes leaders to ensure the remaining cluster can
+    %% resume activity ok
+    PrivDir = ?config(data_dir, Config),
+    ClusterName = ?config(cluster_name, Config),
+    Peers = start_peers([s1,s2,s3], PrivDir),
+    ServerIds = server_ids(ClusterName, Peers),
+    [A, B, C] = ServerIds,
+
+    Machine = {module, ?MODULE, #{}},
+    {ok, _, []} = ra:start_cluster(?SYS, ClusterName, Machine, ServerIds),
+    {ok, _, Leader1} = ra:members(ServerIds),
+
+    %% run some activity to create a snapshot
+    [_ = ra:process_command(Leader1, {banana, I})
+      || I <- lists:seq(1, 5000)],
+
+    Fun = fun F(L0) ->
+                  {ok, _, L} = ra:process_command(L0, banana),
+                  F(L)
+          end,
+    Pid = spawn(fun () -> Fun(Leader1) end),
+    timer:sleep(100),
+
+    exit(Pid, kill),
+    {ok, _, _} = ra:remove_member(Leader1, Leader1),
+
+
+    timer:sleep(500),
+
+    {ok, _, Leader2} = ra:members(ServerIds),
+
+    ct:pal("old leader ~p, new leader ~p", [Leader1, Leader2]),
+    {ok, O, _} = ra:member_overview(Leader2),
+    ct:pal("overview2 ~p", [O]),
+    stop_peers(Peers),
+    ?assertMatch(#{cluster_change_permitted := true}, O),
+    ok.
+
 send_local_msg(Config) ->
     PrivDir = ?config(data_dir, Config),
     ClusterName = ?config(cluster_name, Config),

test/ra_server_SUITE.erl

@@ -1097,22 +1097,6 @@ append_entries_reply_success_promotes_nonvoter(_Config) ->
       {aux, eval}]} = ra_server:handle_leader({N2, Ack2}, State3),
     ok.
 
-% leader_make_rpcs(_Config) ->
-%     N1 = ?N1, N2 = ?N2, N3 = ?N3,
-%     Cluster = #{N1 => new_peer_with(#{next_index => 1, match_index => 0}),
-%                 N2 => new_peer_with(#{next_index => 4, match_index => 3,
-%                                       commit_index_sent => 3}),
-%                 N3 => new_peer_with(#{next_index => 2, match_index => 1})},
-%     State0 = (base_state(3, ?FUNCTION_NAME))#{commit_index => 3,
-%                                               last_applied => 3,
-%                                               cluster => Cluster,
-%                                               machine_state => <<"hi3">>},
-%     {leader, State1, Effs1} = ra_server:handle_leader(pipeline_rpcs, State0),
-%     % Res = ra_server:make_rpcs(State0),
-%     ct:pal("reas ~p", [ra_server:make_rpcs(State1)]),
-
-%     ok.
-
 follower_aer_dupe(_Config) ->
     N1 = ?N1,
     N2 = ?N2,