rabbit_feature_flags: Use cluster members hint for cluster sync

[Why]
During peer discovery, when the feature flags state is synchronized on a
starting node that joins a cluster thanks to peer discovery, the list of
nodes returned by `rabbit_nodes:list_running()` is incorrect because
Mnesia is not initiliazed yet.

Because of that, the synchronization works on the wrong inventory of
feature flags. In the end, the states of feature flags is incorrect
across the cluster.

[How]
`rabbit_mnesia` passes a list of nodes to
`rabbit_feature_flags:sync_feature_flags_with_cluster/2`. We can use
this list, as we did in feature flags V1. This makes sure that the
synchronization works with a valid list of cluster members, in case the
cluster state is not ready yet.

V2: Filter the given list of nodes to only keep those where `rabbit` is
    running. This avoids trying to collect inventory from nodes which
    are stopped.

(cherry picked from commit 7c53958)
(cherry picked from commit a4bde35)

deps/rabbit/src/rabbit_feature_flags.erl

@@ -1822,10 +1822,19 @@ sync_feature_flags_with_cluster(Nodes, NodeIsVirgin) ->
 sync_feature_flags_with_cluster(Nodes, NodeIsVirgin, Timeout) ->
     Clustered = Nodes =/= [],
     case is_enabled(feature_flags_v2) of
-        true when Clustered    -> rabbit_ff_controller:sync_cluster();
-        true when NodeIsVirgin -> rabbit_ff_controller:enable_default();
-        true                   -> ok;
-        false                  -> sync_cluster_v1(Nodes, NodeIsVirgin, Timeout)
+        true when Clustered ->
+            %% We don't use `rabbit_nodes:filter_running()' here because the given
+            %% `Nodes' list may contain nodes which are not members yet (the cluster
+            %% could be being created or expanded).
+            Nodes1 = [N || N <- Nodes, rabbit:is_running(N)],
+            Nodes2 = lists:usort([node() | Nodes1]),
+            rabbit_ff_controller:sync_cluster(Nodes2);
+        true when NodeIsVirgin ->
+            rabbit_ff_controller:enable_default();
+        true ->
+            ok;
+        false ->
+            sync_cluster_v1(Nodes, NodeIsVirgin, Timeout)
     end.
 
 sync_cluster_v1([], NodeIsVirgin, _) ->

deps/rabbit/src/rabbit_ff_controller.erl

@@ -35,7 +35,7 @@
          enable/1,
          enable_default/0,
          check_node_compatibility/1,
-         sync_cluster/0,
+         sync_cluster/1,
          refresh_after_app_load/0,
          get_forced_feature_flag_names/0]).
 
@@ -134,21 +134,21 @@ check_node_compatibility(RemoteNode) ->
     %% feature flags.
     check_node_compatibility_task(ThisNode, RemoteNode).
 
-sync_cluster() ->
+sync_cluster(Nodes) ->
     ?LOG_DEBUG(
        "Feature flags: SYNCING FEATURE FLAGS in cluster...",
        [],
        #{domain => ?RMQLOG_DOMAIN_FEAT_FLAGS}),
     case erlang:whereis(?LOCAL_NAME) of
         Pid when is_pid(Pid) ->
             %% The function is called while `rabbit' is running.
-            gen_statem:call(?LOCAL_NAME, sync_cluster);
+            gen_statem:call(?LOCAL_NAME, {sync_cluster, Nodes});
         undefined ->
             %% The function is called while `rabbit' is stopped. We need to
             %% start a one-off controller, again to make sure concurrent
             %% changes are blocked.
             {ok, Pid} = start_link(),
-            Ret = gen_statem:call(Pid, sync_cluster),
+            Ret = gen_statem:call(Pid, {sync_cluster, Nodes}),
             gen_statem:stop(Pid),
             Ret
     end.
@@ -273,8 +273,8 @@ proceed_with_task({enable, FeatureNames}) ->
     enable_task(FeatureNames);
 proceed_with_task(enable_default) ->
     enable_default_task();
-proceed_with_task(sync_cluster) ->
-    sync_cluster_task();
+proceed_with_task({sync_cluster, Nodes}) ->
+    sync_cluster_task(Nodes);
 proceed_with_task(refresh_after_app_load) ->
     refresh_after_app_load_task().
 
@@ -645,6 +645,15 @@ get_forced_feature_flag_names_from_config() ->
       Reason :: term().
 
 sync_cluster_task() ->
+    Nodes = running_nodes(),
+    sync_cluster_task(Nodes).
+
+-spec sync_cluster_task(Nodes) -> Ret when
+      Nodes :: [node()],
+      Ret :: ok | {error, Reason},
+      Reason :: term().
+
+sync_cluster_task(Nodes) ->
     %% We assume that a feature flag can only be enabled, not disabled.
     %% Therefore this synchronization searches for feature flags enabled on
     %% some nodes but not all, and make sure they are enabled everywhere.
@@ -657,7 +666,6 @@ sync_cluster_task() ->
     %% would make sure a feature flag isn't enabled while there is a network
     %% partition. On the other hand, this would require that all nodes are
     %% running before we can expand the cluster...
-    Nodes = running_nodes(),
     ?LOG_DEBUG(
        "Feature flags: synchronizing feature flags on nodes: ~p",
        [Nodes],

deps/rabbit/test/feature_flags_v2_SUITE.erl

@@ -202,6 +202,7 @@ setup_slave_node(Config, Testcase) ->
         _ ->
             ok = maybe_enable_feature_flags_v2(Config)
     end,
+    _ = catch rabbit_boot_state:set(ready),
     ok.
 
 setup_logger() ->