Use nodes from SAC state to detect stale nodes

Author: acogoluegnes

deps/rabbit/src/rabbit_stream_coordinator.erl

@@ -837,7 +837,7 @@ members() ->
             end
     end.
 
-maybe_resize_coordinator_cluster(LeaderPid, MachineVersion) ->
+maybe_resize_coordinator_cluster(LeaderPid, SacNodes, MachineVersion) ->
     spawn(fun() ->
                   RabbitIsRunning = rabbit:is_running(),
                   case members() of
@@ -872,20 +872,20 @@ maybe_resize_coordinator_cluster(LeaderPid, MachineVersion) ->
                                                       "deleting: ~w", [?MODULE, Old]),
                                       remove_member(Leader, Members, Old)
                               end,
-                          maybe_handle_stale_nodes(MemberNodes, RabbitNodes,
+                          maybe_handle_stale_nodes(SacNodes, RabbitNodes,
                                                    LeaderPid,
                                                    MachineVersion);
                       _ ->
                           ok
                   end
           end).
 
-maybe_handle_stale_nodes(MemberNodes, ExpectedNodes,
+maybe_handle_stale_nodes(SacNodes, BrokerNodes,
                          LeaderPid, MachineVersion) when MachineVersion > 4 ->
-    case MemberNodes -- ExpectedNodes of
+    case SacNodes -- BrokerNodes of
         [] ->
             ok;
-        Stale when length(ExpectedNodes) > 0 ->
+        Stale when length(BrokerNodes) > 0 ->
             rabbit_log:debug("Stale nodes detected in stream SAC "
                              "coordinator: ~w. Purging state.",
                              [Stale]),
@@ -981,7 +981,8 @@ handle_aux(leader, _, maybe_resize_coordinator_cluster,
            #aux{resizer = undefined} = Aux, RaAux) ->
     Leader = ra_aux:leader_id(RaAux),
     MachineVersion = ra_aux:effective_machine_version(RaAux),
-    Pid = maybe_resize_coordinator_cluster(Leader, MachineVersion),
+    SacNodes = sac_list_nodes(ra_aux:machine_state(RaAux), MachineVersion),
+    Pid = maybe_resize_coordinator_cluster(Leader, SacNodes, MachineVersion),
     {no_reply, Aux#aux{resizer = Pid}, RaAux, [{monitor, process, aux, Pid}]};
 handle_aux(leader, _, maybe_resize_coordinator_cluster,
            AuxState, RaAux) ->
@@ -2451,3 +2452,8 @@ sac_make_update_conf(Conf) ->
 
 sac_check_conf_change(SacState) ->
     rabbit_stream_sac_coordinator:check_conf_change(SacState).
+
+sac_list_nodes(State, MachineVersion) when MachineVersion > 4 ->
+    rabbit_stream_sac_coordinator:list_nodes(sac_state(State));
+sac_list_nodes(_, _) ->
+    [].

deps/rabbit/src/rabbit_stream_sac_coordinator.erl

@@ -49,7 +49,8 @@
          group_consumers/5,
          overview/1,
          import_state/2,
-         check_conf_change/1]).
+         check_conf_change/1,
+         list_nodes/1]).
 -export([make_purge_nodes/1,
          make_update_conf/1]).
 
@@ -187,7 +188,7 @@ group_consumers(VirtualHost, Stream, Reference, InfoKeys) ->
         {timeout, _} -> {error, timeout}
     end.
 
--spec overview(state()) -> map().
+-spec overview(state() | undefined) -> map() | undefined.
 overview(undefined) ->
     undefined;
 overview(#?MODULE{groups = Groups}) ->
@@ -862,7 +863,7 @@ import_state(4, #{<<"groups">> := Groups, <<"pids_groups">> := PidsGroups}) ->
              conf = #{disconnected_timeout => ?DISCONNECTED_TIMEOUT_MS}}.
 
 -spec check_conf_change(state() | term()) -> {new, conf()} | unchanged.
-check_conf_change(State) when is_record(State, ?MODULE)->
+check_conf_change(State) when is_record(State, ?MODULE) ->
     #?MODULE{conf = Conf} = State,
     DisconTimeout = lookup_disconnected_timeout(),
     case Conf of
@@ -877,7 +878,22 @@ check_conf_change(State) when is_record(State, ?MODULE)->
 check_conf_change(_) ->
     unchanged.
 
-- spec make_purge_nodes([node()]) -> {sac, command()}.
+-spec list_nodes(state()) -> [node()].
+list_nodes(#?MODULE{groups = Groups}) ->
+    Nodes = maps:fold(fun(_, G, Acc) ->
+                              GNodes = nodes_from_group(G),
+                              maps:merge(Acc, GNodes)
+                      end, #{}, Groups),
+    maps:keys(Nodes).
+
+nodes_from_group(#group{consumers = Cs}) when is_list(Cs) ->
+    lists:foldl(fun(#consumer{pid = Pid}, Acc) ->
+                    Acc#{node(Pid) => true}
+                end, #{}, Cs);
+nodes_from_group(_) ->
+    #{}.
+
+-spec make_purge_nodes([node()]) -> {sac, command()}.
 make_purge_nodes(Nodes) ->
     wrap_cmd(#command_purge_nodes{nodes = Nodes}).