Refactor repair_amqqueue_nodes

- Only query node uids via RPC for missing members
- Update nodes in state to map if FF is enabled
- Properly return whether queue repaired or not

Co-authored-by: Péter Gömöri <[email protected]>

Author: LoisSotoLopez

deps/rabbit/src/rabbit_quorum_queue.erl

@@ -724,43 +724,100 @@ repair_amqqueue_nodes(Q0) ->
     {Name, _} = amqqueue:get_pid(Q0),
     Members = ra_leaderboard:lookup_members(Name),
     RaNodes = [N || {_, N} <- Members],
-    Nodes = get_nodes(Q0),
-    case lists:sort(RaNodes) =:= lists:sort(Nodes) of
+    case rabbit_feature_flags:is_enabled(track_qq_members_uids) of
+        false ->
+            Nodes = get_nodes(Q0),
+            case lists:sort(RaNodes) =:= lists:sort(Nodes) of
+                true ->
+                    %% up to date
+                    ok;
+                false ->
+                    %% update amqqueue record
+                    Fun = fun (Q) ->
+                                  TS0 = amqqueue:get_type_state(Q),
+                                  TS = TS0#{nodes => RaNodes},
+                                  amqqueue:set_type_state(Q, TS)
+                          end,
+                    _ = rabbit_amqqueue:update(QName, Fun),
+                    repaired
+            end;
+        true ->
+            {ok, Q0} = rabbit_amqqueue:lookup(QName),
+            OldTypeState = amqqueue:get_type_state(Q0),
+            case OldTypeState of
+                #{nodes := List} when is_list(List) ->
+                    repair_with_list_nodes(QName, Name, RaNodes, OldTypeState);
+                #{nodes := Map} when is_map(Map) ->
+                    repair_with_map_nodes(QName, Name, RaNodes, Map)
+            end
+    end.
+
+%% @doc Repair logic when OldTypeState has a list as nodes value.
+%% Only updates the queue state if ALL nodes return valid UIDs.
+repair_with_list_nodes(QName, Name, RaNodes, _OldTypeState) ->
+    case gather_node_uids(QName, Name, RaNodes) of
+        {NewNodesUids, _ErrorList = []} ->
+            %% All nodes returned valid UIDs, proceed with update
+            Fun = fun (Q) ->
+                          Ts0 = amqqueue:get_type_state(Q),
+                          Ts = Ts0#{nodes => NewNodesUids},
+                          amqqueue:set_type_state(Q, Ts)
+                  end,
+            _ = rabbit_amqqueue:update(QName, Fun),
+            repaired;
+        _ ->
+            %% Fetching UID for at least some nodes failed
+            %% Do not update the queue state
+            ok
+    end.
+
+%% @doc Repair logic when OldTypeState has a map as nodes value.
+%% Only adds new nodes that return valid UIDs.
+repair_with_map_nodes(QName, Name, RaNodes, PreviousUidsMap) ->
+    PrevNodes = maps:keys(PreviousUidsMap),
+    case lists:sort(PrevNodes) == lists:sort(RaNodes) of
         true ->
-            %% up to date
             ok;
         false ->
-            %% update amqqueue record
+            NodesToAdd = RaNodes -- PrevNodes,
+            {AddedNodesUids, _ErrorList} = gather_node_uids(QName, Name, NodesToAdd),
+            RemainingNodesUids = maps:with(RaNodes, PreviousUidsMap),
+            NewNodes = maps:merge(RemainingNodesUids, AddedNodesUids),
             Fun = fun (Q) ->
-                          TS0 = amqqueue:get_type_state(Q),
-                          TS = case rabbit_feature_flags:is_enabled(track_qq_members_uids) of
-                                   false ->
-                                       TS0#{nodes => RaNodes};
-                                   true ->
-                                       RaUidsList = [begin
-                                                         Uid = erpc:call(N, ra_directory, uid_of,
-                                                                         [?RA_SYSTEM, Name],
-                                                                         ?RPC_TIMEOUT),
-                                                         case Uid of
-                                                             undefined ->
-                                                                 ?LOG_WARNING("Unexpected undefined uuid from node ~p for quorum queue ~ts during repair_amqqueue_nodes",
-                                                                              [N, rabbit_misc:rs(QName)]);
-                                                             _ ->
-                                                                 ok
-                                                         end,
-                                                         {N, Uid}
-                                                     end
-                                                     || N <- RaNodes],
-
-                                       RaUids = maps:from_list(RaUidsList),
-                                       TS0#{nodes => RaUids}
-                               end,
-                          amqqueue:set_type_state(Q, TS)
+                          Ts0 = amqqueue:get_type_state(Q),
+                          Ts = Ts0#{nodes => NewNodes},
+                          amqqueue:set_type_state(Q, Ts)
                   end,
             _ = rabbit_amqqueue:update(QName, Fun),
             repaired
     end.
 
+gather_node_uids(QName, Name, RaNodes) ->
+    RPCRes = erpc:multicall(RaNodes, ra_directory, uid_of, [?RA_SYSTEM, Name], ?RPC_TIMEOUT),
+    NewNodesUidsList0 = lists:zip(RaNodes, RPCRes),
+
+    %% Check if all nodes returned valid UIDs
+    {ValidList, ErrorList} =
+        lists:partition(
+          fun({_Node, {ok, UId}}) when UId =/= undefined ->
+                  true;
+             (_) ->
+                  false
+          end, NewNodesUidsList0),
+    NewNodesUidsList = [{Node, UId} || {Node, {ok, UId}} <- ValidList],
+
+    lists:foreach(fun({Node, {ok, undefined}}) ->
+                          ?LOG_WARNING("Unexpected undefined uuid from node ~p "
+                                       "for quorum ~ts during repair_amqqueue_nodes",
+                                       [Node, rabbit_misc:rs(QName)]);
+                     ({Node, CaughtCallException}) ->
+                          ?LOG_WARNING("Call exception while retrieving uuid from node ~p "
+                                       "for quorum ~ts during repair_amqqueue_nodes: ~p",
+                                       [Node, rabbit_misc:rs(QName), CaughtCallException])
+                  end, ErrorList),
+
+    {maps:from_list(NewNodesUidsList), ErrorList}.
+
 reductions(Name) ->
     try
         {reductions, R} = process_info(whereis(Name), reductions),
@@ -821,7 +878,7 @@ recover(_Vhost, Queues) ->
          RaUId = ra_directory:uid_of(?RA_SYSTEM, Name),
          case RaUId of
              undefined ->
-                 ?LOG_WARNING("Unexpected undefined uuid for current node for quorum queue ~ts during recover",
+                 ?LOG_WARNING("Unexpected undefined uuid for current node for quorum ~ts during recover",
                               [rabbit_misc:rs(QName)]);
              _ ->
                  ok
@@ -838,7 +895,7 @@ recover(_Vhost, Queues) ->
              #{node() := _NewRaUId} ->
                  %% Queue is aware but it does not match the one returned by
                  %% ra_directory
-                 rabbit_log:info("Quorum queue ~ts: detected node uuid change, "
+                 rabbit_log:info("Quorum ~ts: detected node uuid change, "
                                  "deleting old data directory", [rabbit_misc:rs(QName)]),
                  maybe_delete_data_dir(RaUId)
          end,

deps/rabbit/test/quorum_queue_SUITE.erl

@@ -104,6 +104,10 @@ groups() ->
                                             force_checkpoint_on_queue,
                                             force_checkpoint,
                                             policy_repair,
+                                            repair_metadata_nodes_list_to_map,
+                                            repair_metadata_nodes_added_member,
+                                            repair_metadata_nodes_removed_member,
+                                            repair_metadata_nodes_added_removed_member,
                                             gh_12635,
                                             replica_states,
                                             restart_after_queue_reincarnation,
@@ -1522,6 +1526,66 @@ force_checkpoint(Config) ->
     % Result should only have quorum queue
     ?assertEqual(ExpectedRes, ForceCheckpointRes).
 
+repair_metadata_nodes_list_to_map(Config) ->
+    %% After feature flag `track_qq_members_uids` is enabled, quorum
+    %% queues will convert their type state in metadata store
+    %% from nodes list to node=>uid mappings
+    UpdateFun =
+        fun(QueueRec) ->
+                #{nodes := NodesMap} = TypeState = amqqueue:get_type_state(QueueRec),
+                amqqueue:set_type_state(QueueRec, TypeState#{nodes => maps:keys(NodesMap)})
+        end,
+    repair_metadata_nodes(Config, UpdateFun).
+
+repair_metadata_nodes_added_member(Config) ->
+    Server1 = rabbit_ct_broker_helpers:get_node_config(Config, 0, nodename),
+    UpdateFun =
+        fun(QueueRec) ->
+                #{nodes := NodesMap} = TypeState = amqqueue:get_type_state(QueueRec),
+                amqqueue:set_type_state(QueueRec, TypeState#{nodes => maps:remove(Server1, NodesMap)})
+        end,
+    repair_metadata_nodes(Config, UpdateFun).
+
+repair_metadata_nodes_removed_member(Config) ->
+    UpdateFun =
+        fun(QueueRec) ->
+                #{nodes := NodesMap} = TypeState = amqqueue:get_type_state(QueueRec),
+                amqqueue:set_type_state(QueueRec, TypeState#{nodes => NodesMap#{'rabbit@foo' => <<"dummy_uid">>}})
+        end,
+    repair_metadata_nodes(Config, UpdateFun).
+
+repair_metadata_nodes_added_removed_member(Config) ->
+    Server1 = rabbit_ct_broker_helpers:get_node_config(Config, 0, nodename),
+    UpdateFun =
+        fun(QueueRec) ->
+                #{nodes := NodesMap} = TypeState = amqqueue:get_type_state(QueueRec),
+                NewNodeMap = maps:remove(Server1, NodesMap#{'rabbit@foo' => <<"dummy_uid">>}),
+                amqqueue:set_type_state(QueueRec, TypeState#{nodes => NewNodeMap})
+        end,
+    repair_metadata_nodes(Config, UpdateFun).
+
+repair_metadata_nodes(Config, UpdateFun) ->
+    Server = rabbit_ct_broker_helpers:get_node_config(Config, 0, nodename),
+
+    Ch = rabbit_ct_client_helpers:open_channel(Config, Server),
+    QQ = ?config(queue_name, Config),
+    QQName = rabbit_misc:r(<<"/">>, queue, QQ),
+
+    declare(Ch, QQ, [{<<"x-queue-type">>, longstr, <<"quorum">>}]),
+
+
+    QueueRecBefore = rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_amqqueue, lookup, [QQName]),
+
+    rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_amqqueue, update, [QQName, UpdateFun]),
+
+    ?assertEqual(repaired, rpc:call(Server, rabbit_quorum_queue, repair_amqqueue_nodes,
+                                    [QQName])),
+
+    QueueRecAfter = rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_amqqueue, lookup, [QQName]),
+
+    ?assertEqual(QueueRecBefore, QueueRecAfter),
+    ok.
+
 % Tests that, if the process of a QQ is dead in the moment of declaring a policy
 % that affects such queue, when the process is made available again, the policy
 % will eventually get applied. (https://github.com/rabbitmq/rabbitmq-server/issues/7863)