Use initial_machine_version config to avoid initalising

kjnilsson · kjnilsson · commit 0ffb2d0b4c56 · 2025-01-23T10:27:34.000Z
from rabbit_fifo version 0.

The same was also implemented for the stream coordinator.

QQ: avoid dead lock in queue federation.

When processing the queue federation startup even the process
may call back into the ra process causing a deadlock. in this
case we spawn a temporary process to avoid this.
diff --git a/deps/rabbit/src/rabbit_quorum_queue.erl b/deps/rabbit/src/rabbit_quorum_queue.erl
@@ -259,11 +259,16 @@ start_cluster(Q) ->
     NewQ1 = amqqueue:set_type_state(NewQ0,
                                     #{nodes => [LeaderNode | FollowerNodes]}),
 
-    rabbit_log:debug("Will start up to ~w replicas for quorum ~ts with leader on node '~ts'",
-                     [QuorumSize, rabbit_misc:rs(QName), LeaderNode]),
+    Versions = [V || {ok, V} <- erpc:multicall(FollowerNodes,
+                                               rabbit_fifo, version, [])],
+    MinVersion = lists:min([rabbit_fifo:version() | Versions]),
+
+    rabbit_log:debug("Will start up to ~w replicas for quorum queue ~ts with "
+                     "leader on node '~ts', initial machine version ~b",
+                     [QuorumSize, rabbit_misc:rs(QName), LeaderNode, MinVersion]),
     case rabbit_amqqueue:internal_declare(NewQ1, false) of
         {created, NewQ} ->
-            RaConfs = [make_ra_conf(NewQ, ServerId)
+            RaConfs = [make_ra_conf(NewQ, ServerId, voter, MinVersion)
                        || ServerId <- members(NewQ)],
 
             %% khepri projections on remote nodes are eventually consistent
@@ -544,6 +549,10 @@ spawn_deleter(QName) ->
                   delete(Q, false, false, <<"expired">>)
           end).
 
+spawn_notify_decorators(QName, startup = Fun, Args) ->
+    spawn(fun() ->
+                  notify_decorators(QName, Fun, Args)
+          end);
 spawn_notify_decorators(QName, Fun, Args) ->
     %% run in ra process for now
     catch notify_decorators(QName, Fun, Args).
@@ -860,7 +869,7 @@ delete(Q, _IfUnused, _IfEmpty, ActingUser) when ?amqqueue_is_quorum(Q) ->
             notify_decorators(QName, shutdown),
             case delete_queue_data(Q, ActingUser) of
                 ok ->
-                    _ = erpc:call(LeaderNode, rabbit_core_metrics, queue_deleted, [QName],
+                    _ = erpc_call(LeaderNode, rabbit_core_metrics, queue_deleted, [QName],
                                   ?RPC_TIMEOUT),
                     {ok, ReadyMsgs};
                 {error, timeout} = Err ->
@@ -1339,7 +1348,9 @@ add_member(Q, Node, Membership, Timeout) when ?amqqueue_is_quorum(Q) ->
     %% TODO parallel calls might crash this, or add a duplicate in quorum_nodes
     ServerId = {RaName, Node},
     Members = members(Q),
-    Conf = make_ra_conf(Q, ServerId, Membership),
+
+    MachineVersion = erpc_call(Node, rabbit_fifo, version, [], infinity),
+    Conf = make_ra_conf(Q, ServerId, Membership, MachineVersion),
     case ra:start_server(?RA_SYSTEM, Conf) of
         ok ->
             ServerIdSpec  =
@@ -1383,8 +1394,9 @@ add_member(Q, Node, Membership, Timeout) when ?amqqueue_is_quorum(Q) ->
                     E
             end;
         E ->
-          rabbit_log:warning("Could not add a replica of quorum ~ts on node ~ts: ~p", [rabbit_misc:rs(QName), Node, E]),
-          E
+            rabbit_log:warning("Could not add a replica of quorum ~ts on node ~ts: ~p",
+                               [rabbit_misc:rs(QName), Node, E]),
+            E
     end.
 
 delete_member(VHost, Name, Node) ->
@@ -1912,9 +1924,10 @@ format_ra_event(ServerId, Evt, QRef) ->
     {'$gen_cast', {queue_event, QRef, {ServerId, Evt}}}.
 
 make_ra_conf(Q, ServerId) ->
-    make_ra_conf(Q, ServerId, voter).
+    make_ra_conf(Q, ServerId, voter, rabbit_fifo:version()).
 
-make_ra_conf(Q, ServerId, Membership) ->
+make_ra_conf(Q, ServerId, Membership, MacVersion)
+  when is_integer(MacVersion) ->
     TickTimeout = application:get_env(rabbit, quorum_tick_interval,
                                       ?TICK_INTERVAL),
     SnapshotInterval = application:get_env(rabbit, quorum_snapshot_interval,
@@ -1923,10 +1936,12 @@ make_ra_conf(Q, ServerId, Membership) ->
                                              quorum_min_checkpoint_interval,
                                              ?MIN_CHECKPOINT_INTERVAL),
     make_ra_conf(Q, ServerId, TickTimeout,
-                 SnapshotInterval, CheckpointInterval, Membership).
+                 SnapshotInterval, CheckpointInterval,
+                 Membership, MacVersion).
 
 make_ra_conf(Q, ServerId, TickTimeout,
-             SnapshotInterval, CheckpointInterval, Membership) ->
+             SnapshotInterval, CheckpointInterval,
+             Membership, MacVersion) ->
     QName = amqqueue:get_name(Q),
     RaMachine = ra_machine(Q),
     [{ClusterName, _} | _] = Members = members(Q),
@@ -1947,6 +1962,7 @@ make_ra_conf(Q, ServerId, TickTimeout,
                                   log_init_args => LogCfg,
                                   tick_timeout => TickTimeout,
                                   machine => RaMachine,
+                                  initial_machine_version => MacVersion,
                                   ra_event_formatter => Formatter}).
 
 make_mutable_config(Q) ->
diff --git a/deps/rabbit/src/rabbit_stream_coordinator.erl b/deps/rabbit/src/rabbit_stream_coordinator.erl
@@ -3,7 +3,6 @@
 %% file, You can obtain one at https://mozilla.org/MPL/2.0/.
 %%
 %% Copyright (c) 2007-2025 Broadcom. All Rights Reserved. The term “Broadcom” refers to Broadcom Inc. and/or its subsidiaries. All rights reserved.
-%%
 
 -module(rabbit_stream_coordinator).
 
@@ -493,9 +492,17 @@ locally_known_members() ->
 
 start_coordinator_cluster() ->
     Nodes = rabbit_nodes:list_reachable(),
-    rabbit_log:debug("Starting stream coordinator on nodes: ~w", [Nodes]),
     true = Nodes =/= [],
-    case ra:start_cluster(?RA_SYSTEM, [make_ra_conf(Node, Nodes) || Node <-  Nodes]) of
+
+    Versions = [V || {ok, V} <- erpc:multicall(Nodes,
+                                               ?MODULE, version, [])],
+    MinVersion = lists:min([version() | Versions]),
+    rabbit_log:debug("Starting stream coordinator on nodes: ~w, "
+                     "initial machine version ~b",
+                     [Nodes, MinVersion]),
+    case ra:start_cluster(?RA_SYSTEM,
+                          [make_ra_conf(Node, Nodes, MinVersion)
+                           || Node <- Nodes]) of
         {ok, Started, _} ->
             rabbit_log:debug("Started stream coordinator on ~w", [Started]),
             Started;
@@ -813,7 +820,8 @@ maybe_resize_coordinator_cluster() ->
           end).
 
 add_member(Members, Node) ->
-    Conf = make_ra_conf(Node, [N || {_, N} <- Members]),
+    MinMacVersion = erpc:call(Node, ?MODULE, version, []),
+    Conf = make_ra_conf(Node, [N || {_, N} <- Members], MinMacVersion),
     ServerId = {?MODULE, Node},
     case ra:start_server(?RA_SYSTEM, Conf) of
         ok ->
@@ -1255,7 +1263,7 @@ phase_update_mnesia(StreamId, Args, #{reference := QName,
 format_ra_event(ServerId, Evt) ->
     {stream_coordinator_event, ServerId, Evt}.
 
-make_ra_conf(Node, Nodes) ->
+make_ra_conf(Node, Nodes, MinMacVersion) ->
     UId = ra:new_uid(ra_lib:to_binary(?MODULE)),
     Formatter = {?MODULE, format_ra_event, []},
     Members = [{?MODULE, N} || N <- Nodes],
@@ -1270,6 +1278,7 @@ make_ra_conf(Node, Nodes) ->
       log_init_args => #{uid => UId},
       tick_timeout => TickTimeout,
       machine => {module, ?MODULE, #{}},
+      initial_machine_version => MinMacVersion,
       ra_event_formatter => Formatter}.
 
 filter_command(_Meta, {delete_replica, _, #{node := Node}}, #stream{id = StreamId,
diff --git a/deps/rabbit/test/rabbit_fifo_int_SUITE.erl b/deps/rabbit/test/rabbit_fifo_int_SUITE.erl
@@ -76,6 +76,7 @@ init_per_testcase(TestCase, Config) ->
     meck:expect(rabbit_quorum_queue, cancel_consumer_handler, fun (_, _) -> ok end),
     meck:new(rabbit_feature_flags, []),
     meck:expect(rabbit_feature_flags, is_enabled, fun (_) -> true end),
+    meck:expect(rabbit_feature_flags, is_enabled, fun (_, _) -> true end),
     ra_server_sup_sup:remove_all(?RA_SYSTEM),
     ServerName2 = list_to_atom(atom_to_list(TestCase) ++ "2"),
     ServerName3 = list_to_atom(atom_to_list(TestCase) ++ "3"),
@@ -941,10 +942,16 @@ discard_next_delivery(ClusterName, State0, Wait) ->
     end.
 
 start_cluster(ClusterName, ServerIds, RaFifoConfig) ->
-    {ok, Started, _} = ra:start_cluster(?RA_SYSTEM,
-                                        ClusterName#resource.name,
-                                        {module, rabbit_fifo, RaFifoConfig},
-                                        ServerIds),
+    UId = ra:new_uid(ra_lib:to_binary(ClusterName#resource.name)),
+    Confs = [#{id => Id,
+               uid => UId,
+               cluster_name => ClusterName#resource.name,
+               log_init_args => #{uid => UId},
+               initial_members => ServerIds,
+               initial_machine_version => rabbit_fifo:version(),
+               machine => {module, rabbit_fifo, RaFifoConfig}}
+             || Id <- ServerIds],
+    {ok, Started, _} = ra:start_cluster(?RA_SYSTEM, Confs),
     ?assertEqual(length(Started), length(ServerIds)),
     ok.
 
