Reconcile QQ node dead during delete and redeclare

LoisSotoLopez · gomoripeti · LoisSotoLopez · commit 782fccf62e0b · 2025-08-13T13:22:56.000+02:00
Co-authored-by: Péter Gömöri &lt;gomoripeti@users.noreply.github.com&gt;
diff --git a/deps/rabbit/src/rabbit_quorum_queue.erl b/deps/rabbit/src/rabbit_quorum_queue.erl
@@ -276,9 +276,12 @@ 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]}),
+                                    #{nodes => [LeaderNode | FollowerNodes],
+                                      uids => UIDs}),
 
     Versions = [V || {ok, V} <- erpc:multicall(FollowerNodes,
                                                rabbit_fifo, version, [],
@@ -792,6 +795,24 @@ recover(_Vhost, Queues) ->
          ServerId = {Name, node()},
          QName = amqqueue:get_name(Q0),
          MutConf = make_mutable_config(Q0),
+         RaUId = ra_directory:uid_of(?RA_SYSTEM, Name),
+         QTypeState0 = amqqueue:get_type_state(Q0),
+         RaUIds = maps:get(uids, QTypeState0, undefined),
+         QTypeState = case RaUIds of
+             undefined ->
+                 %% 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#{uids := RaUIds#{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
@@ -804,7 +825,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"
@@ -826,8 +847,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 +1228,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) ->
@@ -1379,16 +1404,30 @@ 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 = amqqueue:get_type_state(Q0),
+    RaUIds = maps:get(uids, QTypeState0, undefined),
+    QTypeState = case RaUIds of
+        undefined ->
+            %% Queue is not aware of node to uid mapping, do nothing
+            QTypeState0;
+        #{Node := _} ->
+            %% Queue is aware and uid for targeted node exists, do nothing
+            QTypeState0;
+        _ ->
+            %% Queue is aware but current node has no UId, regen uid
+            NewRaUId = ra:new_uid(ra_lib:to_binary(RaName)),
+            QTypeState0#{uids := RaUIds#{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
@@ -1478,7 +1517,11 @@ delete_member(Q, Node) when ?amqqueue_is_quorum(Q) ->
                     Fun = fun(Q1) ->
                                   update_type_state(
                                     Q1,
-                                    fun(#{nodes := Nodes} = Ts) ->
+                                    fun(#{nodes := Nodes,
+                                          uids := UIds} = Ts) ->
+                                            Ts#{nodes => lists:delete(Node, Nodes),
+                                                uids => maps:remove(Node, UIds)};
+                                       (#{nodes := Nodes} = Ts) ->
                                             Ts#{nodes => lists:delete(Node, Nodes)}
                                     end)
                           end,
@@ -1986,7 +2029,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
+        #{uids := #{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,
diff --git a/deps/rabbit/test/quorum_queue_SUITE.erl b/deps/rabbit/test/quorum_queue_SUITE.erl
@@ -105,7 +105,9 @@ groups() ->
                                             force_checkpoint,
                                             policy_repair,
                                             gh_12635,
-                                            replica_states
+                                            replica_states,
+                                            restart_after_queue_reincarnation,
+                                            no_messages_after_queue_reincarnation
                                            ]
                        ++ all_tests()},
                       {cluster_size_5, [], [start_queue,
@@ -2792,15 +2794,21 @@ add_member_wrong_type(Config) ->
                           [<<"/">>, SQ, Server, voter, 5000])).
 
 add_member_already_a_member(Config) ->
-    [Server | _] = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
+    [Server, Server2 | _] = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
     Ch = rabbit_ct_client_helpers:open_channel(Config, Server),
     QQ = ?config(queue_name, Config),
     ?assertEqual({'queue.declare_ok', QQ, 0, 0},
                  declare(Ch, QQ, [{<<"x-queue-type">>, longstr, <<"quorum">>}])),
+    R1 = rpc:call(Server, rabbit_amqqueue, lookup, [{resource, <<"/">>, queue, QQ}]),
     %% idempotent by design
     ?assertEqual(ok,
                  rpc:call(Server, rabbit_quorum_queue, add_member,
-                          [<<"/">>, QQ, Server, voter, 5000])).
+                          [<<"/">>, QQ, Server, voter, 5000])),
+    ?assertEqual(R1, rpc:call(Server, rabbit_amqqueue, lookup, [{resource, <<"/">>, queue, QQ}])),
+    ?assertEqual(ok,
+                 rpc:call(Server, rabbit_quorum_queue, add_member,
+                          [<<"/">>, QQ, Server2, voter, 5000])),
+    ?assertEqual(R1, rpc:call(Server, rabbit_amqqueue, lookup, [{resource, <<"/">>, queue, QQ}])).
 
 add_member_not_found(Config) ->
     [Server | _] = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
@@ -4872,6 +4880,140 @@ replica_states(Config) ->
                 end
              end, Result2).
 
+% Testcase motivated by : https://github.com/rabbitmq/rabbitmq-server/discussions/13131
+restart_after_queue_reincarnation(Config) ->
+    [S1, S2, S3] = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
+    Ch = rabbit_ct_client_helpers:open_channel(Config, S1),
+    QName = <<"QQ">>,
+
+    ?assertEqual({'queue.declare_ok', QName, 0, 0},
+                 declare(Ch, QName, [{<<"x-queue-type">>, longstr, <<"quorum">>}])),
+
+    [Q] = rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_amqqueue, list, []),
+    VHost = amqqueue:get_vhost(Q),
+
+    MessagesPublished = 1000,
+    publish_many(Ch, QName, MessagesPublished),
+
+    %% Trigger a snapshot by purging the queue.
+    rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_queue_type, purge, [Q]),
+
+    %% Stop S3
+    rabbit_ct_broker_helpers:mark_as_being_drained(Config, S3),
+    ?assertEqual(ok, rabbit_control_helper:command(stop_app, S3)),
+
+    %% Delete and re-declare queue with the same name.
+    rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_amqqueue, delete, [Q,false,false,<<"dummy_user">>]),
+    ?assertEqual({'queue.declare_ok', QName, 0, 0},
+                 declare(Ch, QName, [{<<"x-queue-type">>, longstr, <<"quorum">>}])),
+
+    % Now S3 should have the old queue state, and S1 and S2 a new one.
+    St1 = rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_quorum_queue, status, [VHost, QName]),
+    Status0 = [{proplists:get_value(<<"Node Name">>, S), S} || S <- St1],
+    S3_Status1 = proplists:get_value(S3, Status0),
+    Others_Status1 = [V || {_K, V} <- proplists:delete(S3, Status0)],
+
+    S3_LastLogIndex = proplists:get_value(<<"Last Log Index">>, S3_Status1),
+    S3_LastWritten = proplists:get_value(<<"Last Written">>, S3_Status1),
+    S3_LastApplied = proplists:get_value(<<"Last Applied">>, S3_Status1),
+    S3_CommitIndex = proplists:get_value(<<"Commit Index">>, S3_Status1),
+    S3_Term = proplists:get_value(<<"Term">>, S3_Status1),
+
+    ?assertEqual(noproc, proplists:get_value(<<"Raft State">>, S3_Status1)),
+    ?assertEqual(unknown, proplists:get_value(<<"Membership">>, S3_Status1)),
+    [begin
+        ?assert(S3_LastLogIndex > proplists:get_value(<<"Last Log Index">>, O)),
+        ?assert(S3_LastWritten > proplists:get_value(<<"Last Written">>, O)),
+        ?assert(S3_LastApplied > proplists:get_value(<<"Last Applied">>, O)),
+        ?assert(S3_CommitIndex > proplists:get_value(<<"Commit Index">>, O)),
+        ?assertEqual(S3_Term, proplists:get_value(<<"Term">>, O))
+     end || O <- Others_Status1],
+
+    %% Bumping term in online nodes
+    rabbit_ct_broker_helpers:rpc(Config, 1, rabbit_quorum_queue, transfer_leadership, [Q, S2]),
+
+    %% Restart S3
+    ?assertEqual(ok, rabbit_control_helper:command(start_app, S3)),
+
+    timer:sleep(1000),
+
+    %% Now all three nodes should have the new state.
+    Status2 = rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_quorum_queue, status, [VHost, QName]),
+    % They are either leader or follower.
+    ?assert(
+       lists:all(
+         fun(NodeStatus) ->
+                 NodeRaftState = proplists:get_value(<<"Raft State">>, NodeStatus),
+                 lists:member(NodeRaftState, [leader, follower])
+         end, Status2)),
+    % Remove "Node Name" and "Raft State" from the status.
+    Status3 = [NE1, NE2, NE3]= [
+        begin
+            R = proplists:delete(<<"Node Name">>, NodeEntry),
+            proplists:delete(<<"Raft State">>, R)
+        end || NodeEntry <- Status2],
+    % Check all other properties have same value on all nodes.
+    ct:pal("Status3: ~tp", [Status3]),
+    [
+     begin
+        ?assertEqual(V, proplists:get_value(K, NE2)),
+        ?assertEqual(V, proplists:get_value(K, NE3))
+     end || {K, V} <- NE1
+    ].
+
+% Testcase motivated by : https://github.com/rabbitmq/rabbitmq-server/issues/12366
+no_messages_after_queue_reincarnation(Config) ->
+    [S1, S2, S3] = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
+    Ch = rabbit_ct_client_helpers:open_channel(Config, S1),
+    QName = <<"QQ">>,
+
+    ?assertEqual({'queue.declare_ok', QName, 0, 0},
+                 declare(Ch, QName, [{<<"x-queue-type">>, longstr, <<"quorum">>}])),
+
+    [Q] = rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_amqqueue, list, []),
+
+    publish(Ch, QName, <<"msg1">>),
+    publish(Ch, QName, <<"msg2">>),
+
+    %% Stop S3
+    rabbit_ct_broker_helpers:mark_as_being_drained(Config, S3),
+    ?assertEqual(ok, rabbit_control_helper:command(stop_app, S3)),
+
+    qos(Ch, 1, false),
+    subscribe(Ch, QName, false, <<"tag0">>, [], 500),
+    DeliveryTag = receive
+        {#'basic.deliver'{delivery_tag = DT}, #amqp_msg{}} ->
+            receive
+                {#'basic.deliver'{consumer_tag = <<"tag0">>}, #amqp_msg{}} ->
+                    ct:fail("did not expect the second one")
+            after 500 ->
+                DT
+            end
+    after 500 ->
+            ct:fail("Expected some delivery, but got none")
+    end,
+
+    %% Delete and re-declare queue with the same name.
+    rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_amqqueue, delete, [Q,false,false,<<"dummy_user">>]),
+    ?assertEqual({'queue.declare_ok', QName, 0, 0},
+                 declare(Ch, QName, [{<<"x-queue-type">>, longstr, <<"quorum">>}])),
+
+    %% Bumping term in online nodes
+    rabbit_ct_broker_helpers:rpc(Config, 1, rabbit_quorum_queue, transfer_leadership, [Q, S2]),
+
+    %% Restart S3
+    ?assertEqual(ok, rabbit_control_helper:command(start_app, S3)),
+
+    ok = amqp_channel:cast(Ch, #'basic.ack'{delivery_tag = DeliveryTag,
+                                       multiple     = false}),
+    %% No message should be delivered after reincarnation
+    receive
+        {#'basic.deliver'{consumer_tag = <<"tag0">>}, #amqp_msg{}} ->
+            ct:fail("Expected no deliveries, but got one")
+    after 500 ->
+            ok
+    end.
+
 %%----------------------------------------------------------------------------
 
 same_elements(L1, L2)
@@ -4941,7 +5083,10 @@ consume_empty(Ch, Queue, NoAck) ->
 subscribe(Ch, Queue, NoAck) ->
     subscribe(Ch, Queue, NoAck, <<"ctag">>, []).
 
+
 subscribe(Ch, Queue, NoAck, Tag, Args) ->
+    subscribe(Ch, Queue, NoAck, Tag, Args, ?TIMEOUT).
+subscribe(Ch, Queue, NoAck, Tag, Args, Timeout) ->
     amqp_channel:subscribe(Ch, #'basic.consume'{queue = Queue,
                                                 no_ack = NoAck,
                                                 arguments = Args,
@@ -4950,7 +5095,7 @@ subscribe(Ch, Queue, NoAck, Tag, Args) ->
     receive
         #'basic.consume_ok'{consumer_tag = Tag} ->
              ok
-    after ?TIMEOUT ->
+    after Timeout ->
               flush(100),
               exit(subscribe_timeout)
     end.
