track global non_voter status from the leader

Alex Valiushko · Alex Valiushko · commit e6c474f4cb29 · 2023-09-11T16:42:02.000-07:00
diff --git a/src/ra.erl b/src/ra.erl
@@ -23,6 +23,8 @@
          %% queries
          members/1,
          members/2,
+         cluster/1,
+         cluster/2,
          initial_members/1,
          initial_members/2,
          local_query/2,
@@ -558,10 +560,9 @@ delete_cluster(ServerIds, Timeout) ->
 %% affect said cluster's availability characteristics (by increasing quorum node count).
 %%
 %% @param ServerLoc the ra server or servers to try to send the command to
-%% @param ServerId the ra server id of the new server, or a map with server id and settings.
+%% @param ServerId the ra server id of the new server.
 %% @end
--spec add_member(ra_server_id() | [ra_server_id()],
-                 ra_server_id() | ra_new_server()) ->
+-spec add_member(ra_server_id() | [ra_server_id()], ra_server_id()) ->
     ra_cmd_ret() |
     {error, already_member} |
     {error, cluster_change_not_permitted}.
@@ -572,8 +573,7 @@ add_member(ServerLoc, ServerId) ->
 %% @see add_member/2
 %% @end
 -spec add_member(ra_server_id() | [ra_server_id()],
-                 ra_server_id() | ra_new_server(),
-                 timeout()) ->
+                 ra_server_id(), timeout()) ->
     ra_cmd_ret() |
     {error, already_member} |
     {error, cluster_change_not_permitted}.
@@ -1038,6 +1038,44 @@ members({local, ServerId}, Timeout) ->
 members(ServerId, Timeout) ->
     ra_server_proc:state_query(ServerId, members, Timeout).
 
+%% @doc Returns a map of cluster members to their status.
+%%
+%% Except if `{local, ServerId}' is passed, the query is sent to the specified
+%% server which may redirect it to the leader if it is a follower. It may
+%% timeout if there is currently no leader (i.e. an election is in progress).
+%%
+%% With `{local, ServerId}', the query is always handled by the specified
+%% server. It means the returned list might be out-of-date compared to what the
+%% leader would have returned.
+%%
+%% @param ServerId the Ra server(s) to send the query to
+%% @end
+-spec cluster(ra_server_id() | [ra_server_id()]) ->
+    ra_server_proc:ra_leader_call_ret(ra_cluster()).
+cluster(ServerId) ->
+    cluster(ServerId, ?DEFAULT_TIMEOUT).
+
+%% @doc Returns a map of cluster members to their status.
+%%
+%% Except if `{local, ServerId}' is passed, the query is sent to the specified
+%% server which may redirect it to the leader if it is a follower. It may
+%% timeout if there is currently no leader (i.e. an election is in progress).
+%%
+%% With `{local, ServerId}', the query is always handled by the specified
+%% server. It means the returned list might be out-of-date compared to what the
+%% leader would have returned.
+%%
+%% @param ServerId the Ra server(s) to send the query to
+%% @param Timeout the timeout to use
+%% @end
+-spec cluster(ra_server_id() | [ra_server_id()] | {local, ra_server_id()},
+              timeout()) ->
+    ra_server_proc:ra_leader_call_ret(ra_cluster()).
+cluster({local, ServerId}, Timeout) ->
+    ra_server_proc:local_state_query(ServerId, cluster, Timeout);
+cluster(ServerId, Timeout) ->
+    ra_server_proc:state_query(ServerId, cluster, Timeout).
+
 %% @doc Returns a list of initial (seed) cluster members.
 %%
 %% This allows Ra-based systems with dynamic cluster membership
diff --git a/src/ra.hrl b/src/ra.hrl
@@ -39,10 +39,6 @@
 %% after node restart). Pids are not stable in this sense.
 -type ra_server_id() :: {Name :: atom(), Node :: node()}.
 
-%% Specifies server configuration for a new cluster member.
--type ra_new_server() :: #{id := ra_server_id(),
-                           voter := boolean()}.
-
 -type ra_peer_status() :: normal |
                           {sending_snapshot, pid()} |
                           suspended |
diff --git a/src/ra_server.erl b/src/ra_server.erl
@@ -2513,25 +2513,16 @@ append_log_leader({'$ra_join', From,
             Cluster = OldCluster#{JoiningNode => new_peer_with(#{voter_status => Voter})},
             append_cluster_change(Cluster, From, ReplyMode, State)
     end;
-append_log_leader({'$ra_join', From, #{id := JoiningNode,
-                                       voter := WantVoter}, ReplyMode},
-                  State) ->
-    % Shortcut to compute non-voter status
-    VoterStatus = case WantVoter of
-                      true -> voter;
-                      false -> new_nonvoter(State)
-                  end,
-    append_log_leader({'$ra_join', From, #{id => JoiningNode,
-                                           voter_status => VoterStatus}, ReplyMode},
-                      State);
 append_log_leader({'$ra_join', From, JoiningNode, ReplyMode},
                   State = #{cluster := OldCluster}) ->
-    % Legacy $ra_join, join as full voter iff no such member in the cluster.
+    % Legacy $ra_join, join as non voter iff no such member in the cluster.
     case OldCluster of
         #{JoiningNode := _} ->
             already_member(State);
         _ ->
-            append_log_leader({'$ra_join', From, #{id => JoiningNode, voter => true}, ReplyMode},
+            append_log_leader({'$ra_join', From,
+                               #{id => JoiningNode, voter_status => new_nonvoter(State)},
+                               ReplyMode},
                               State)
     end;
 append_log_leader({'$ra_leave', From, LeavingServer, ReplyMode},
@@ -2583,17 +2574,7 @@ append_cluster_change(Cluster, From, ReplyMode,
                                 cluster := PrevCluster,
                                 cluster_index_term := {PrevCITIdx, PrevCITTerm},
                                 current_term := Term}) ->
-    %% TODO A bit dense.
-    maps:foreach(fun(K, #{voter_status := TargetStatus}) ->
-                     case maps:get(K, PrevCluster, #{voter_status => voter}) of
-                       #{voter_status := TargetStatus} ->
-                         ok;
-                       #{voter_status := voter} when TargetStatus =/= voter ->
-                         ets:insert(ra_non_voters, K);
-                       #{voter_status := {nonvoter, _}} when TargetStatus =:= voter ->
-                         ets:delete_object(ra_non_voters, K)
-                     end
-                 end, Cluster),
+    update_known_non_voters(PrevCluster, Cluster),
     % turn join command into a generic cluster change command
     % that include the new cluster configuration
     Command = {'$ra_cluster_change', From, Cluster, ReplyMode},
@@ -2609,6 +2590,26 @@ append_cluster_change(Cluster, From, ReplyMode,
             cluster_index_term => IdxTerm,
             previous_cluster => {PrevCITIdx, PrevCITTerm, PrevCluster}}}.
 
+update_known_non_voters(OldCluster, NewCluster) ->
+  %% ra_non_voters table is used to report known non-voters from the leader perspecrtive.
+  maps:foreach(fun(K, V) ->
+                   OldV = maps:get(K, OldCluster, undefined),
+                   update_known_non_voter(K, OldV, V)
+               end, NewCluster),
+  Removed = maps:to_list(OldCluster) -- maps:to_list(NewCluster),
+  lists:foreach(fun({K, OldV})->
+                   update_known_non_voter(K, OldV, undefined)
+                end, Removed).
+
+update_known_non_voter(_Id, Peer, Peer) ->
+  ok;
+update_known_non_voter(Id, _OldPeer, undefined) ->
+  ets:delete_object(ra_non_voters, Id);
+update_known_non_voter(Id, _OldPeer, #{voter_status := voter}) ->
+  ets:delete_object(ra_non_voters, Id);
+update_known_non_voter(Id, _OldPeer, #{voter_status := _}) ->
+  ets:insert(ra_non_voters, Id).
+
 mismatch_append_entries_reply(Term, CommitIndex, State0) ->
     {CITerm, State} = fetch_term(CommitIndex, State0),
     % assert CITerm is found
diff --git a/src/ra_server_proc.erl b/src/ra_server_proc.erl
@@ -184,6 +184,7 @@ log_fold(ServerId, Fun, InitialState, Timeout) ->
                   all |
                   overview |
                   members |
+                  cluster |
                   initial_members |
                   machine, timeout()) ->
     ra_leader_call_ret(term()).
@@ -194,6 +195,7 @@ state_query(ServerLoc, Spec, Timeout) ->
                         all |
                         overview |
                         members |
+                        cluster |
                         initial_members |
                         machine, timeout()) ->
     ra_local_call_ret(term()).
@@ -1514,6 +1516,8 @@ do_state_query(machine, #{machine_state := MacState}) ->
     MacState;
 do_state_query(members, #{cluster := Cluster}) ->
     maps:keys(Cluster);
+do_state_query(cluster, #{cluster := Cluster}) ->
+    Cluster;
 do_state_query(initial_members, #{log := Log}) ->
     case ra_log:read_config(Log) of
         {ok, #{initial_members := InitialMembers}} ->
diff --git a/test/coordination_SUITE.erl b/test/coordination_SUITE.erl
@@ -399,7 +399,7 @@ disconnected_node_catches_up(Config) ->
 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]],
+    [{_, ANode} = 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)),
@@ -408,25 +408,25 @@ nonvoter_catches_up(Config) ->
      || N <- lists:seq(1, 10000)],
     {ok, _, _} = ra:process_command(Leader, banana),
 
-    New = #{id => C, voter => false},
-    {ok, _, _} = ra:add_member(A, New),
+    {ok, _, _} = ra:add_member(A, C),
     ok = ra:start_server(?SYS, ClusterName, C, Machine, [A, B]),
-    ?assertMatch({ok, #{cluster := #{C := #{voter_status := {nonvoter, _}}}}, _},
-                 ra:member_overview(A)),
+    ?assertMatch({ok, #{C := #{voter_status := {nonvoter, _}}}, _}, ra:cluster(C)),
+    ?assertMatch(#{known_non_voters := [C]}, rpc:call(ANode, ra, overview,[?SYS])),
 
     await_condition(
       fun () ->
-          {ok, #{cluster := #{C := Peer}}, _} = ra:member_overview(A),
-          voter == maps:get(voter_status, Peer)
+          {ok, #{C := #{voter_status := Voter}}, _} = ra:cluster(C),
+          voter =:= Voter
       end, 200),
+    ?assertMatch(#{known_non_voters := []}, rpc:call(ANode, ra, overview,[?SYS])),
 
     [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]],
+    [{_, ANode} = 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)),
@@ -435,27 +435,27 @@ nonvoter_catches_up_after_restart(Config) ->
      || N <- lists:seq(1, 10000)],
     {ok, _, _} = ra:process_command(Leader, banana),
 
-    New = #{id => C, voter => false},
-    {ok, _, _} = ra:add_member(A, New),
+    {ok, _, _} = ra:add_member(A, C),
     ok = ra:start_server(?SYS, ClusterName, C, Machine, [A, B]),
-    ?assertMatch({ok, #{cluster := #{C := #{voter_status := {nonvoter, _}}}}, _},
-                 ra:member_overview(A)),
+    ?assertMatch({ok, #{C := #{voter_status := {nonvoter, _}}}, _}, ra:cluster(C)),
+    ?assertMatch(#{known_non_voters := [C]}, rpc:call(ANode, ra, overview,[?SYS])),
     ok = ra:stop_server(?SYS, C),
     ok = ra:restart_server(?SYS, C),
 
     await_condition(
       fun () ->
-          {ok, #{cluster := #{C := Peer}}, _} = ra:member_overview(A),
-          voter == maps:get(voter_status, Peer)
+          {ok, #{C := #{voter_status := Voter}}, _} = ra:cluster(C),
+          voter =:= Voter
       end, 200),
+    ?assertMatch(#{known_non_voters := []}, rpc:call(ANode, ra, overview,[?SYS])),
 
     [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]],
+    [{_, ANode} = 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)),
@@ -464,19 +464,19 @@ nonvoter_catches_up_after_leader_restart(Config) ->
      || N <- lists:seq(1, 10000)],
     {ok, _, _} = ra:process_command(Leader, banana),
 
-    New = #{id => C, voter => false},
-    {ok, _, _} = ra:add_member(A, New),
+    {ok, _, _} = ra:add_member(A, C),
     ok = ra:start_server(?SYS, ClusterName, C, Machine, [A, B]),
-    ?assertMatch({ok, #{cluster := #{C := #{voter_status := {nonvoter, _}}}}, _},
-                 ra:member_overview(A)),
+    ?assertMatch({ok, #{C := #{voter_status := {nonvoter, _}}}, _}, ra:cluster(C)),
+    ?assertMatch(#{known_non_voters := [C]}, rpc:call(ANode, ra, overview,[?SYS])),
     ok = ra:stop_server(?SYS, Leader),
     ok = ra:restart_server(?SYS, Leader),
 
     await_condition(
       fun () ->
-          {ok, #{cluster := #{C := Peer}}, _} = ra:member_overview(A),
-          voter == maps:get(voter_status, Peer)
+          {ok, #{C := #{voter_status := Voter}}, _} = ra:cluster(C),
+          voter =:= Voter
       end, 200),
+    ?assertMatch(#{known_non_voters := []}, rpc:call(ANode, ra, overview,[?SYS])),
 
     [ok = slave:stop(S) || {_, S} <- ServerIds],
     ok.
@@ -730,3 +730,14 @@ await_condition(Fun, Attempts) ->
             timer:sleep(100),
             await_condition(Fun, Attempts - 1)
     end.
+
+fail_on_condition(_Fun, 0) ->
+    ok;
+fail_on_condition(Fun, Attempts) ->
+    case catch Fun() of
+        true -> 
+            exit(condition_materialised);
+        _ ->
+            timer:sleep(100),
+            fail_on_condition(Fun, Attempts - 1)
+    end.
diff --git a/test/ra_2_SUITE.erl b/test/ra_2_SUITE.erl
@@ -45,7 +45,6 @@ all_tests() ->
      external_reader,
      add_member_without_quorum,
      force_start_follower_as_single_member,
-     force_start_follower_as_single_member_nonvoter,
      initial_members_query
     ].
 
@@ -674,46 +673,6 @@ force_start_follower_as_single_member(Config) ->
     UId4 = ?config(uid4, Config),
     Conf4 = conf(ClusterName, UId4, ServerId4, PrivDir, [ServerId3]),
     {ok, _, _} = ra:add_member(ServerId3, ServerId4),
-    {timeout,_} = ra:process_command(ServerId3, {enq, banana}),
-    %% start new member
-    ok = ra:start_server(?SYS, Conf4),
-    {ok, _, ServerId3} = ra:members(ServerId4),
-    ok = enqueue(ServerId3, msg3),
-
-    ok.
-
-force_start_follower_as_single_member_nonvoter(Config) ->
-    ok = logger:set_primary_config(level, all),
-    %% ra:start_server should fail if the node already exists
-    ClusterName = ?config(cluster_name, Config),
-    PrivDir = ?config(priv_dir, Config),
-    ServerId1 = ?config(server_id, Config),
-    ServerId2 = ?config(server_id2, Config),
-    ServerId3 = ?config(server_id3, Config),
-    InitialCluster = [ServerId1, ServerId2, ServerId3],
-    ok = start_cluster(ClusterName, InitialCluster),
-    timer:sleep(100),
-    %% stop majority to simulate permanent outage
-    ok = ra:stop_server(?SYS, ServerId1),
-    ok = ra:stop_server(?SYS, ServerId2),
-
-    timer:sleep(100),
-    %% force the remaining node to change it's membership
-    ok = ra_server_proc:force_shrink_members_to_current_member(ServerId3),
-    {ok, [_], ServerId3} = ra:members(ServerId3),
-    ok = enqueue(ServerId3, msg1),
-
-    %% test that it works after restart
-    ok = ra:stop_server(?SYS, ServerId3),
-    ok = ra:restart_server(?SYS, ServerId3),
-    {ok, [_], ServerId3} = ra:members(ServerId3),
-    ok = enqueue(ServerId3, msg2),
-
-    %% add a member
-    ServerId4 = ?config(server_id4, Config),
-    UId4 = ?config(uid4, Config),
-    Conf4 = conf(ClusterName, UId4, ServerId4, PrivDir, [ServerId3]),
-    {ok, _, _} = ra:add_member(ServerId3, #{id => ServerId4, voter => false}),
     %% the membership has changed but member not running yet
     %% it is nonvoter and does not affect quorum size
     {ok, _, _} = ra:process_command(ServerId3, {enq, banana}),
diff --git a/test/ra_SUITE.erl b/test/ra_SUITE.erl
diff --git a/test/ra_server_SUITE.erl b/test/ra_server_SUITE.erl
