Reconcile QQ node dead during delete and redeclare

LoisSotoLopez · gomoripeti · LoisSotoLopez · commit 3293a246b09d · 2025-09-29T15:05:55.000+02:00
Co-authored-by: Péter Gömöri &lt;gomoripeti@users.noreply.github.com&gt;
diff --git a/deps/rabbit/src/amqqueue.erl b/deps/rabbit/src/amqqueue.erl
@@ -30,7 +30,6 @@
          % exclusive_owner
          get_exclusive_owner/1,
          get_leader_node/1,
-         get_nodes/1,
          % name (#resource)
          get_name/1,
          set_name/2,
@@ -425,15 +424,6 @@ get_leader_node(#amqqueue{pid = {_, Leader}}) -> Leader;
 get_leader_node(#amqqueue{pid = none}) -> none;
 get_leader_node(#amqqueue{pid = Pid}) -> node(Pid).
 
--spec get_nodes(amqqueue_v2()) -> [node(),...].
-
-get_nodes(Q) ->
-    case amqqueue:get_type_state(Q) of
-        #{nodes := Nodes} ->
-            Nodes;
-        _ ->
-            [get_leader_node(Q)]
-    end.
 
 % operator_policy
 
diff --git a/deps/rabbit/src/rabbit_amqp_management.erl b/deps/rabbit/src/rabbit_amqp_management.erl
@@ -471,7 +471,7 @@ encode_queue(Q, NumMsgs, NumConsumers) ->
     {Leader :: node() | none, Replicas :: [node(),...]}.
 queue_topology(Q) ->
     Leader = amqqueue:get_leader_node(Q),
-    Replicas = amqqueue:get_nodes(Q),
+    Replicas = rabbit_amqqueue:get_nodes(Q),
     {Leader, Replicas}.
 
 decode_exchange({map, KVList}) ->
diff --git a/deps/rabbit/src/rabbit_amqqueue.erl b/deps/rabbit/src/rabbit_amqqueue.erl
@@ -23,6 +23,7 @@
 -export([list/0, list_durable/0, list/1, info_keys/0, info/1, info/2, info_all/1, info_all/2,
          emit_info_all/5, list_local/1, info_local/1,
          emit_info_local/4, emit_info_down/4]).
+-export([get_nodes/1]).
 -export([count/0]).
 -export([list_down/1, list_down/2, list_all/1,
          count/1, list_names/0, list_names/1, list_local_names/0,
@@ -1226,6 +1227,12 @@ list() ->
 count() ->
     rabbit_db_queue:count().
 
+-spec get_nodes(amqqueue:amqqueue_v2()) -> [node(),...].
+
+get_nodes(Q) ->
+    [{members, Nodes}] = info(Q, [members]),
+    Nodes.
+
 -spec list_names() -> [name()].
 
 list_names() ->
@@ -2035,12 +2042,7 @@ pseudo_queue(#resource{kind = queue} = QueueName, Pid, Durable)
                 ).
 
 get_quorum_nodes(Q) ->
-    case amqqueue:get_type_state(Q) of
-        #{nodes := Nodes} ->
-            Nodes;
-        _ ->
-            []
-    end.
+    rabbit_amqqueue:get_nodes(Q).
 
 -spec prepend_extra_bcc(Qs) ->
     Qs when Qs :: [amqqueue:target() | {amqqueue:target(), route_infos()}].
diff --git a/deps/rabbit/src/rabbit_core_ff.erl b/deps/rabbit/src/rabbit_core_ff.erl
@@ -219,3 +219,10 @@
       depends_on    => ['rabbitmq_4.1.0'],
       callbacks     => #{enable => {rabbit_khepri, enable_feature_flag}}
      }}).
+
+-rabbit_feature_flag(
+   {'track_qq_members_uids',
+    #{desc         => "Track queue members UIDs in the metadata store",
+      stability     => stable,
+      depends_on    => []
+     }}).
diff --git a/deps/rabbit/src/rabbit_queue_location.erl b/deps/rabbit/src/rabbit_queue_location.erl
@@ -143,7 +143,7 @@ select_members(Size, _, AllNodes, RunningNodes, _, _, GetQueues) ->
     Counters0 = maps:from_list([{N, 0} || N <- lists:delete(?MODULE:node(), AllNodes)]),
     Queues = GetQueues(),
     Counters = lists:foldl(fun(Q, Acc) ->
-                                   #{nodes := Nodes} = amqqueue:get_type_state(Q),
+                                   Nodes = rabbit_amqqueue:get_nodes(Q),
                                    lists:foldl(fun(N, A)
                                                      when is_map_key(N, A) ->
                                                        maps:update_with(N, fun(C) -> C+1 end, A);
diff --git a/deps/rabbit/src/rabbit_quorum_queue.erl b/deps/rabbit/src/rabbit_quorum_queue.erl
@@ -270,9 +270,17 @@ start_cluster(Q) ->
     {LeaderNode, FollowerNodes} =
         rabbit_queue_location:select_leader_and_followers(Q, QuorumSize),
     LeaderId = {RaName, LeaderNode},
+    UIDs = maps:from_list([{Node, ra:new_uid(ra_lib:to_binary(RaName))}
+                           || Node <- [LeaderNode | FollowerNodes]]),
     NewQ0 = amqqueue:set_pid(Q, LeaderId),
-    NewQ1 = amqqueue:set_type_state(NewQ0,
-                                    #{nodes => [LeaderNode | FollowerNodes]}),
+    NewQ1 = case rabbit_feature_flags:is_enabled(track_qq_members_uids) of
+        false ->
+            amqqueue:set_type_state(NewQ0,
+                                    #{nodes => [LeaderNode | FollowerNodes]});
+        true ->
+            amqqueue:set_type_state(NewQ0,
+                                    #{nodes => UIDs})
+    end,
 
     Versions = [V || {ok, V} <- erpc:multicall(FollowerNodes,
                                                rabbit_fifo, version, [],
@@ -717,7 +725,7 @@ repair_amqqueue_nodes(Q0) ->
     {Name, _} = amqqueue:get_pid(Q0),
     Members = ra_leaderboard:lookup_members(Name),
     RaNodes = [N || {_, N} <- Members],
-    #{nodes := Nodes} = amqqueue:get_type_state(Q0),
+    Nodes = get_nodes(Q0),
     case lists:sort(RaNodes) =:= lists:sort(Nodes) of
         true ->
             %% up to date
@@ -726,7 +734,18 @@ repair_amqqueue_nodes(Q0) ->
             %% update amqqueue record
             Fun = fun (Q) ->
                           TS0 = amqqueue:get_type_state(Q),
-                          TS = TS0#{nodes => RaNodes},
+                          TS = case rabbit_feature_flags:is_enabled(track_qq_members_uids)
+                                    andalso has_uuid_tracking(TS0)
+                               of
+                                   false ->
+                                       TS0#{nodes => RaNodes};
+                                   true ->
+                                       RaUids = maps:from_list([{N, erpc:call(N, ra_directory, uid_of,
+                                                               [?RA_SYSTEM, Name],
+                                                               ?RPC_TIMEOUT)}
+                                                 || N <- RaNodes]),
+                                       TS0#{nodes => RaUids}
+                               end,
                           amqqueue:set_type_state(Q, TS)
                   end,
             _ = rabbit_amqqueue:update(QName, Fun),
@@ -790,6 +809,23 @@ recover(_Vhost, Queues) ->
          ServerId = {Name, node()},
          QName = amqqueue:get_name(Q0),
          MutConf = make_mutable_config(Q0),
+         RaUId = ra_directory:uid_of(?RA_SYSTEM, Name),
+         #{nodes := Nodes} = QTypeState0 = amqqueue:get_type_state(Q0),
+         QTypeState = case Nodes of
+             List when is_list(List) ->
+                 %% Queue is not aware of node to uid mapping, do nothing
+                 QTypeState0;
+             #{node() := RaUId} ->
+                 %% Queue is aware and uid for current node is correct, do nothing
+                 QTypeState0;
+             _ ->
+                 %% Queue is aware but either current node has no UId or it
+                 %% does not match the one returned by ra_directory, regen uid
+                 maybe_delete_data_dir(RaUId),
+                 NewRaUId = ra:new_uid(ra_lib:to_binary(Name)),
+                 QTypeState0#{nodes := Nodes#{node() => NewRaUId}}
+         end,
+         Q = amqqueue:set_type_state(Q0, QTypeState),
          Res = case ra:restart_server(?RA_SYSTEM, ServerId, MutConf) of
                    ok ->
                        % queue was restarted, good
@@ -802,7 +838,7 @@ recover(_Vhost, Queues) ->
                                           [rabbit_misc:rs(QName), Err1]),
                        % queue was never started on this node
                        % so needs to be started from scratch.
-                       case start_server(make_ra_conf(Q0, ServerId)) of
+                       case start_server(make_ra_conf(Q, ServerId)) of
                            ok -> ok;
                            Err2 ->
                                ?LOG_WARNING("recover: quorum queue ~w could not"
@@ -824,8 +860,7 @@ recover(_Vhost, Queues) ->
          %% present in the rabbit_queue table and not just in
          %% rabbit_durable_queue
          %% So many code paths are dependent on this.
-         ok = rabbit_db_queue:set_dirty(Q0),
-         Q = Q0,
+         ok = rabbit_db_queue:set_dirty(Q),
          case Res of
              ok ->
                  {[Q | R0], F0};
@@ -1208,12 +1243,17 @@ cleanup_data_dir() ->
 maybe_delete_data_dir(UId) ->
     _ = ra_directory:unregister_name(?RA_SYSTEM, UId),
     Dir = ra_env:server_data_dir(?RA_SYSTEM, UId),
-    {ok, Config} = ra_log:read_config(Dir),
-    case maps:get(machine, Config) of
-        {module, rabbit_fifo, _} ->
-            ra_lib:recursive_delete(Dir);
-        _ ->
-            ok
+    case filelib:is_dir(Dir) of
+        false ->
+            ok;
+        true ->
+            {ok, Config} = ra_log:read_config(Dir),
+            case maps:get(machine, Config) of
+                {module, rabbit_fifo, _} ->
+                    ra_lib:recursive_delete(Dir);
+                _ ->
+                    ok
+            end
     end.
 
 policy_changed(Q) ->
@@ -1378,16 +1418,29 @@ add_member(Q, Node, Membership) ->
     do_add_member(Q, Node, Membership, ?MEMBER_CHANGE_TIMEOUT).
 
 
-do_add_member(Q, Node, Membership, Timeout)
-  when ?is_amqqueue(Q) andalso
-       ?amqqueue_is_quorum(Q) andalso
+do_add_member(Q0, Node, Membership, Timeout)
+  when ?is_amqqueue(Q0) andalso
+       ?amqqueue_is_quorum(Q0) andalso
        is_atom(Node) ->
-    {RaName, _} = amqqueue:get_pid(Q),
-    QName = amqqueue:get_name(Q),
+    {RaName, _} = amqqueue:get_pid(Q0),
+    QName = amqqueue:get_name(Q0),
     %% TODO parallel calls might crash this, or add a duplicate in quorum_nodes
     ServerId = {RaName, Node},
-    Members = members(Q),
-
+    Members = members(Q0),
+    QTypeState0 = #{nodes := Nodes} = amqqueue:get_type_state(Q0),
+    NewRaUId = ra:new_uid(ra_lib:to_binary(RaName)),
+    QTypeState = case Nodes of
+        L when is_list(L) ->
+            %% Queue is not aware of node to uid mapping, just add the new node
+            QTypeState0#{nodes => lists:usort([Node | Nodes])};
+        #{Node := _} ->
+            %% Queue is aware and uid for targeted node exists, do nothing
+            QTypeState0;
+        _ ->
+            %% Queue is aware but current node has no UId, regen uid
+            QTypeState0#{nodes => Nodes#{Node => NewRaUId}}
+    end,
+    Q = amqqueue:set_type_state(Q0, QTypeState),
     MachineVersion = erpc_call(Node, rabbit_fifo, version, [], infinity),
     Conf = make_ra_conf(Q, ServerId, Membership, MachineVersion),
     case ra:start_server(?RA_SYSTEM, Conf) of
@@ -1403,8 +1456,12 @@ do_add_member(Q, Node, Membership, Timeout)
                 {ok, {RaIndex, RaTerm}, Leader} ->
                     Fun = fun(Q1) ->
                                   Q2 = update_type_state(
-                                         Q1, fun(#{nodes := Nodes} = Ts) ->
-                                                     Ts#{nodes => lists:usort([Node | Nodes])}
+                                         Q1, fun(#{nodes := NodesList} = Ts) when is_list(NodesList) ->
+                                                     Ts#{nodes => lists:usort([Node | NodesList])};
+                                                (#{nodes := #{Node := _}} = Ts) ->
+                                                     Ts;
+                                                (#{nodes := NodesMap} = Ts) when is_map(NodesMap) ->
+                                                     Ts#{nodes => maps:put(Node, NewRaUId, NodesMap)}
                                              end),
                                   amqqueue:set_pid(Q2, Leader)
                           end,
@@ -1477,8 +1534,10 @@ delete_member(Q, Node) when ?amqqueue_is_quorum(Q) ->
                     Fun = fun(Q1) ->
                                   update_type_state(
                                     Q1,
-                                    fun(#{nodes := Nodes} = Ts) ->
-                                            Ts#{nodes => lists:delete(Node, Nodes)}
+                                    fun(#{nodes := Nodes} = Ts) when is_list(Nodes) ->
+                                            Ts#{nodes => lists:delete(Node, Nodes)};
+                                       (#{nodes := Nodes} = Ts) when is_map(Nodes) ->
+                                            Ts#{nodes => maps:remove(Node, Nodes)}
                                     end)
                           end,
                     _ = rabbit_amqqueue:update(QName, Fun),
@@ -1999,7 +2058,15 @@ make_ra_conf(Q, ServerId, TickTimeout,
     #resource{name = QNameBin} = QName,
     RaMachine = ra_machine(Q),
     [{ClusterName, _} | _] = Members = members(Q),
-    UId = ra:new_uid(ra_lib:to_binary(ClusterName)),
+    {_, Node} = ServerId,
+    UId = case amqqueue:get_type_state(Q) of
+        #{nodes := #{Node := Id}} ->
+            Id;
+        _ ->
+            %% Queue was declared on an older version of RabbitMQ
+            %% and does not have the node to uid mappings
+            ra:new_uid(ra_lib:to_binary(ClusterName))
+    end,
     FName = rabbit_misc:rs(QName),
     Formatter = {?MODULE, format_ra_event, [QName]},
     LogCfg = #{uid => UId,
@@ -2031,7 +2098,12 @@ make_mutable_config(Q) ->
 
 get_nodes(Q) when ?is_amqqueue(Q) ->
     #{nodes := Nodes} = amqqueue:get_type_state(Q),
-    Nodes.
+    case Nodes of
+        List when is_list(List) ->
+            List;
+        Map when is_map(Map) ->
+            maps:keys(Map)
+    end.
 
 get_connected_nodes(Q) when ?is_amqqueue(Q) ->
     ErlangNodes = [node() | nodes()],
@@ -2138,7 +2210,7 @@ force_checkpoint_on_queue(QName) ->
         {ok, Q} when ?amqqueue_is_quorum(Q) ->
             {RaName, _} = amqqueue:get_pid(Q),
             ?LOG_DEBUG("Sending command to force ~ts to take a checkpoint", [QNameFmt]),
-            Nodes = amqqueue:get_nodes(Q),
+            Nodes = rabbit_amqqueue:get_nodes(Q),
             _ = [ra:cast_aux_command({RaName, Node}, force_checkpoint)
                  || Node <- Nodes],
             ok;
@@ -2408,3 +2480,8 @@ queue_vm_stats_sups() ->
 queue_vm_ets() ->
     {[quorum_ets],
      [[ra_log_ets]]}.
+
+has_uuid_tracking(#{nodes := Nodes}) when is_map(Nodes) ->
+    true;
+has_uuid_tracking(_QTypeState) ->
+    false.
diff --git a/deps/rabbit/src/rabbit_stream_coordinator.erl b/deps/rabbit/src/rabbit_stream_coordinator.erl
@@ -153,8 +153,8 @@ stop() ->
 
 new_stream(Q, LeaderNode)
   when ?is_amqqueue(Q) andalso is_atom(LeaderNode) ->
-    #{name := StreamId,
-      nodes := Nodes} = amqqueue:get_type_state(Q),
+    #{name := StreamId} = amqqueue:get_type_state(Q),
+    Nodes = rabbit_amqqueue:get_nodes(Q),
     %% assertion leader is in nodes configuration
     true = lists:member(LeaderNode, Nodes),
     process_command({new_stream, StreamId,
diff --git a/deps/rabbit/test/quorum_queue_SUITE.erl b/deps/rabbit/test/quorum_queue_SUITE.erl
diff --git a/deps/rabbitmq_ct_helpers/src/queue_utils.erl b/deps/rabbitmq_ct_helpers/src/queue_utils.erl
diff --git a/deps/rabbitmq_prometheus/src/collectors/prometheus_rabbitmq_core_metrics_collector.erl b/deps/rabbitmq_prometheus/src/collectors/prometheus_rabbitmq_core_metrics_collector.erl
