fix: member restarts as nonvoter after catching up

Author: Alex Valiushko

src/ra.hrl

@@ -149,6 +149,7 @@
 -type snapshot_meta() :: #{index := ra_index(),
                            term := ra_term(),
                            cluster := ra_cluster_servers(),
+                           non_voters => ra_cluster_servers(),
                            machine_version := ra_machine:version()}.
 
 -record(install_snapshot_rpc,

src/ra_log.erl

@@ -638,12 +638,17 @@ update_release_cursor0(Idx, Cluster, MacVersion, MacState,
                                 reader = Reader,
                                 snapshot_state = SnapState} = State0) ->
     ClusterServerIds = maps:keys(Cluster),
+    NonvoterIds = maps:fold(fun
+                              (K, #{voter_status := {nonvoter, _}}, Acc) -> [K|Acc];
+                              (_, _, Acc) -> Acc
+                            end, [], Cluster),
     SnapLimit = case ra_snapshot:current(SnapState) of
                     undefined -> SnapInter;
                     {I, _} -> I + SnapInter
                 end,
     Meta = #{index => Idx,
              cluster => ClusterServerIds,
+             non_voters => NonvoterIds,
              machine_version => MacVersion},
     % The release cursor index is the last entry _not_ contributing
     % to the current state. I.e. the last entry that can be discarded.

src/ra_server.erl

@@ -295,21 +295,25 @@ init(#{id := Id,
 
     LatestMacVer = ra_machine:version(Machine),
 
-    {_FirstIndex, Cluster0, MacVer, MacState,
+    {Cluster, VoterStatus,
+     MacVer, MacState,
      {SnapshotIdx, _} = SnapshotIndexTerm} =
         case ra_log:recover_snapshot(Log0) of
             undefined ->
                 InitialMachineState = ra_machine:init(Machine, Name),
-                {0, make_cluster(Id, InitialNodes),
-                 0, InitialMachineState, {0, 0}};
-            {#{index := Idx,
-               term := Term,
+                Voter = case maps:get(voter, Config, true) of
+                          false ->
+                              new_nonvoter(init);
+                          true ->
+                              voter
+                        end,
+                {make_cluster(Id, InitialNodes), Voter, 0, InitialMachineState, {0, 0}};
+            {#{index := Idx, term := Term,
                cluster := ClusterNodes,
-               machine_version := MacVersion}, MacSt} ->
-                Clu = make_cluster(Id, ClusterNodes),
-                %% the snapshot is the last index before the first index
-                %% TODO: should this be Idx + 1?
-                {Idx + 1, Clu, MacVersion, MacSt, {Idx, Term}}
+               machine_version := MacVersion} = Snapshot, MacSt} ->
+                Nonvoters = maps:get(non_voters, Snapshot, []),
+                Clu = make_cluster(Id, ClusterNodes, Nonvoters, snapshot),
+                {Clu, voter_status(Id, Clu), MacVersion, MacSt, {Idx, Term}}
         end,
     MacMod = ra_machine:which_module(Machine, MacVer),
 
@@ -331,18 +335,9 @@ init(#{id := Id,
     put_counter(Cfg, ?C_RA_SVR_METRIC_LAST_APPLIED, SnapshotIdx),
     put_counter(Cfg, ?C_RA_SVR_METRIC_TERM, CurrentTerm),
 
-    VoterStatus = case maps:get(voter, Config, true) of
-                      false ->
-                          {nonvoter, init};
-                      true ->
-                          voter
-                  end,
-    Peer = maps:get(Id, Cluster0),
-    Cluster1 = Cluster0#{Id => Peer#{voter_status => VoterStatus}},
-
     #{cfg => Cfg,
       current_term => CurrentTerm,
-      cluster => Cluster1,
+      cluster => Cluster,
       % There may be scenarios when a single server
       % starts up but hasn't
       % yet re-applied its noop command that we may receive other join
@@ -1296,13 +1291,14 @@ handle_receive_snapshot(#install_snapshot_rpc{term = Term,
                           Cfg0
                   end,
 
-            {#{cluster := ClusterIds}, MacState} = ra_log:recover_snapshot(Log),
+            {#{cluster := ClusterIds} = Snapshot, MacState} = ra_log:recover_snapshot(Log),
+            Nonvoters = maps:get(non_voters, Snapshot, []),
             State = update_term(Term,
                                 State0#{cfg => Cfg,
                                         log => Log,
                                         commit_index => SnapIndex,
                                         last_applied => SnapIndex,
-                                        cluster => make_cluster(Id, ClusterIds),
+                                        cluster => make_cluster(Id, ClusterIds, Nonvoters, remote_snapshot),
                                         machine_state => MacState}),
             %% it was the last snapshot chunk so we can revert back to
             %% follower status
@@ -2245,8 +2241,15 @@ fetch_term(Idx, #{log := Log0} = State) ->
     end.
 
 make_cluster(Self, Nodes) ->
+  make_cluster(Self, Nodes, [], all_voters).
+
+make_cluster(Self, Nodes, Nonvoters, Reason) ->
     case lists:foldl(fun(N, Acc) ->
-                             Acc#{N => new_peer()}
+                         P = case lists:member(N, Nonvoters) of
+                               true -> new_nonvoter(Reason);
+                               false -> new_peer()
+                             end,
+                         Acc#{N =>P}
                      end, #{}, Nodes) of
         #{Self := _} = Cluster ->
             % current server is already in cluster - do nothing
@@ -2895,9 +2898,11 @@ already_member(State) ->
 %%% Voter status helpers
 %%% ====================
 
--spec new_nonvoter(ra_server_state()) -> ra_voter_status().
+-spec new_nonvoter(ra_server_state() | atom()) -> ra_voter_status().
 new_nonvoter(#{commit_index := Target} = _State) ->
-    {nonvoter, #{target => Target}}.
+    {nonvoter, #{target => Target}};
+new_nonvoter(Reason) when is_atom(Reason) ->
+    {nonvoter, Reason}.
 
 -spec maybe_promote_voter(ra_server_id(), ra_server_state(), effects()) -> effects().
 maybe_promote_voter(PeerID, #{cluster := Cluster} = _State, Effects) ->

test/coordination_SUITE.erl

@@ -29,6 +29,9 @@ all() ->
 
 all_tests() ->
     [
+     nonvoter_catches_up,
+     nonvoter_catches_up_after_restart,
+     nonvoter_catches_up_after_leader_restart,
      start_stop_restart_delete_on_remote,
      start_cluster,
      start_or_restart_cluster,
@@ -393,6 +396,91 @@ disconnected_node_catches_up(Config) ->
     [ok = slave:stop(S) || {_, S} <- ServerIds],
     ok.
 
+nonvoter_catches_up(Config) ->
+    PrivDir = ?config(data_dir, Config),
+    ClusterName = ?config(cluster_name, Config),
+    [A, B, C] = ServerIds = [{ClusterName, start_follower(N, PrivDir)} || N <- [s1,s2,s3]],
+    Machine = {module, ?MODULE, #{}},
+    {ok, Started, []} = ra:start_cluster(?SYS, ClusterName, Machine, [A, B]),
+    {ok, _, Leader} = ra:members(hd(Started)),
+
+    [ok = ra:pipeline_command(Leader, N, no_correlation, normal)
+     || N <- lists:seq(1, 10000)],
+    {ok, _, _} = ra:process_command(Leader, banana),
+
+    New = #{id => C, voter => false},
+    {ok, _, _} = ra:add_member(A, New),
+    ok = ra:start_server(?SYS, ClusterName, New, Machine, [A, B]),
+    ?assertMatch({ok, #{voter_status := {nonvoter, _}}, _},
+                 ra:member_overview(C)),
+
+    await_condition(
+      fun () ->
+          {ok, O2, _} = ra:member_overview(C),
+          voter == maps:get(voter_status, O2)
+      end, 200),
+
+    [ok = slave:stop(S) || {_, S} <- ServerIds],
+    ok.
+
+nonvoter_catches_up_after_restart(Config) ->
+    PrivDir = ?config(data_dir, Config),
+    ClusterName = ?config(cluster_name, Config),
+    [A, B, C] = ServerIds = [{ClusterName, start_follower(N, PrivDir)} || N <- [s1,s2,s3]],
+    Machine = {module, ?MODULE, #{}},
+    {ok, Started, []} = ra:start_cluster(?SYS, ClusterName, Machine, [A, B]),
+    {ok, _, Leader} = ra:members(hd(Started)),
+
+    [ok = ra:pipeline_command(Leader, N, no_correlation, normal)
+     || N <- lists:seq(1, 10000)],
+    {ok, _, _} = ra:process_command(Leader, banana),
+
+    New = #{id => C, voter => false},
+    {ok, _, _} = ra:add_member(A, New),
+    ok = ra:start_server(?SYS, ClusterName, New, Machine, [A, B]),
+    ?assertMatch({ok, #{voter_status := {nonvoter, _}}, _},
+                 ra:member_overview(C)),
+    ok = ra:stop_server(?SYS, C),
+    ok = ra:restart_server(?SYS, C),
+
+    await_condition(
+      fun () ->
+          {ok, O2, _} = ra:member_overview(C),
+          voter == maps:get(voter_status, O2)
+      end, 200),
+
+    [ok = slave:stop(S) || {_, S} <- ServerIds],
+    ok.
+
+nonvoter_catches_up_after_leader_restart(Config) ->
+    PrivDir = ?config(data_dir, Config),
+    ClusterName = ?config(cluster_name, Config),
+    [A, B, C] = ServerIds = [{ClusterName, start_follower(N, PrivDir)} || N <- [s1,s2,s3]],
+    Machine = {module, ?MODULE, #{}},
+    {ok, Started, []} = ra:start_cluster(?SYS, ClusterName, Machine, [A, B]),
+    {ok, _, Leader} = ra:members(hd(Started)),
+
+    [ok = ra:pipeline_command(Leader, N, no_correlation, normal)
+     || N <- lists:seq(1, 10000)],
+    {ok, _, _} = ra:process_command(Leader, banana),
+
+    New = #{id => C, voter => false},
+    {ok, _, _} = ra:add_member(A, New),
+    ok = ra:start_server(?SYS, ClusterName, New, Machine, [A, B]),
+    ?assertMatch({ok, #{voter_status := {nonvoter, _}}, _},
+                 ra:member_overview(C)),
+    ok = ra:stop_server(?SYS, Leader),
+    ok = ra:restart_server(?SYS, Leader),
+
+    await_condition(
+      fun () ->
+          {ok, O2, _} = ra:member_overview(C),
+          voter == maps:get(voter_status, O2)
+      end, 200),
+
+    [ok = slave:stop(S) || {_, S} <- ServerIds],
+    ok.
+
 key_metrics(Config) ->
     PrivDir = ?config(data_dir, Config),
     ClusterName = ?config(cluster_name, Config),

test/ra_log_snapshot_SUITE.erl

@@ -71,7 +71,7 @@ end_per_testcase(_TestCase, _Config) ->
 
 roundtrip(Config) ->
     Dir = ?config(dir, Config),
-    SnapshotMeta = meta(33, 94, [{banana, node@jungle}, {banana, node@savanna}]),
+    SnapshotMeta = meta(33, 94, [{banana, node@jungle}, {banana, node@savanna}], [{banana, node@savanna}]),
     SnapshotRef = my_state,
     ok = ra_log_snapshot:write(Dir, SnapshotMeta, SnapshotRef),
     Context = #{can_accept_full_file => true},
@@ -80,7 +80,7 @@ roundtrip(Config) ->
 
 roundtrip_compat(Config) ->
     Dir = ?config(dir, Config),
-    SnapshotMeta = meta(33, 94, [{banana, node@jungle}, {banana, node@savanna}]),
+    SnapshotMeta = meta(33, 94, [{banana, node@jungle}, {banana, node@savanna}], [{banana, node@savanna}]),
     SnapshotRef = my_state,
     ok = ra_log_snapshot:write(Dir, SnapshotMeta, SnapshotRef),
     ?assertEqual({SnapshotMeta, SnapshotRef}, read(Dir)),
@@ -105,7 +105,7 @@ test_accept(Config, Name, DataSize, FullFile, ChunkSize) ->
     Dir = dir(?config(dir, Config), Name),
     AcceptDir = dir(?config(accept_dir, Config), Name),
     ct:pal("test_accept ~w ~b ~w ~b", [Name, DataSize, FullFile, ChunkSize]),
-    SnapshotMeta = meta(33, 94, [{banana, node@jungle}, {banana, node@savanna}]),
+    SnapshotMeta = meta(33, 94, [{banana, node@jungle}, {banana, node@savanna}], [{banana, node@savanna}]),
     SnapshotRef = crypto:strong_rand_bytes(DataSize),
     ok = ra_log_snapshot:write(Dir, SnapshotMeta, SnapshotRef),
     Context = #{can_accept_full_file => FullFile},
@@ -178,24 +178,25 @@ recover_invalid_checksum(Config) ->
 
 read_meta_data(Config) ->
     Dir = ?config(dir, Config),
-    SnapshotMeta = meta(33, 94, [{banana, node@jungle}, {banana, node@savanna}]),
+    SnapshotMeta = meta(33, 94, [{banana, node@jungle}, {banana, node@savanna}], [{banana, node@savanna}]),
     SnapshotRef = my_state,
     ok = ra_log_snapshot:write(Dir, SnapshotMeta, SnapshotRef),
     {ok, SnapshotMeta} = ra_log_snapshot:read_meta(Dir),
     ok.
 
 recover_same_as_read(Config) ->
     Dir = ?config(dir, Config),
-    SnapshotMeta = meta(33, 94, [{banana, node@jungle}, {banana, node@savanna}]),
+    SnapshotMeta = meta(33, 94, [{banana, node@jungle}, {banana, node@savanna}], [{banana, node@savanna}]),
     SnapshotData = my_state,
     ok = ra_log_snapshot:write(Dir, SnapshotMeta, SnapshotData),
     {ok, SnapshotMeta, SnapshotData} = ra_log_snapshot:recover(Dir),
     ok.
 
 %% Utility
 
-meta(Idx, Term, Cluster) ->
+meta(Idx, Term, Cluster, Nonvoters) ->
     #{index => Idx,
       term => Term,
       cluster => Cluster,
+      non_voters => Nonvoters,
       machine_version => 1}.