Keep exclusive queues with Khepri + network partition

dumbbell · dumbbell · commit 3986f35c0198 · 2025-09-22T17:32:07.000+02:00
[Why] With Mnesia, when the network partition strategy is set to `pause_minority`, nodes on the "minority side" are stopped. Thus, the exclusive queues that were hosted by nodes on that minority side are lost: * Consumers connected on these nodes are disconnected because the nodes are stopped. * Queue records on the majority side are deleted from the metadata store. This was ok with Mnesia and how this network partition handling strategy is implemented. However, it does not work with Khepri because the nodes on the "minority side" continue to run and serve clients. Therefore the cluster ends up in a weird situation: 1. The "majority side" deleted the queue records. 2. When the network partition is solved, the "minority side" gets the record deletion, but the queue processes continue to run. [How] With Khepri, we stop to delete transient queue records in general, just because there is a node going down. Thanks to this, an exclusive queue and its consumer are not affected by a network partition: they continue to work. However, if a node is really lost, we need to clean up dead queue records. This was already done for durable queues with both Mnesia and Khepri. But with Khepri, transient queue records persist in the store like durable queue records (unlike with Mnesia). That's why this commit changes the clean-up function, `rabbit_amqqueue:forget_all_durable/1` into `rabbit_amqqueue:forget_all/1` which deletes all queue records of queues that were hosted on the given node, regardless if they are transient or durable. Fixes #12949, #12597.
diff --git a/deps/rabbit/src/rabbit_amqqueue.erl b/deps/rabbit/src/rabbit_amqqueue.erl
@@ -9,7 +9,7 @@
 
 -export([recover/1, stop/1, start/1, declare/6, declare/7,
          delete_immediately/1, delete_exclusive/2, delete/4, purge/1,
-         forget_all_durable/1]).
+         forget_all/1]).
 -export([pseudo_queue/2, pseudo_queue/3]).
 -export([exists/1, lookup/1, lookup/2, lookup_many/1, lookup_durable_queue/1,
          not_found_or_absent_dirty/1,
@@ -1890,19 +1890,19 @@ internal_delete(Queue, ActingUser, Reason) ->
                                       {user_who_performed_action, ActingUser}])
     end.
 
--spec forget_all_durable(node()) -> 'ok'.
+-spec forget_all(node()) -> 'ok'.
 
-%% TODO this is used by `rabbit_mnesia:remove_node_if_mnesia_running`
-%% Does it make any sense once mnesia is not used/removed?
-forget_all_durable(Node) ->
-    ?LOG_INFO("Will remove all classic queues from node ~ts. The node is likely being removed from the cluster.", [Node]),
+%% This is used by `rabbit_mnesia:remove_node_if_mnesia_running/1' and
+%% `rabbit_khepri:remove_*_member/1'.
+forget_all(Node) ->
+    ?LOG_INFO("Will remove all queues from node ~ts. The node is likely being removed from the cluster.", [Node]),
     UpdateFun = fun(Q) ->
                         forget_node_for_queue(Q)
                 end,
     FilterFun = fun(Q) ->
                         is_local_to_node(amqqueue:get_pid(Q), Node)
                 end,
-    rabbit_db_queue:foreach_durable(UpdateFun, FilterFun).
+    rabbit_db_queue:foreach(UpdateFun, FilterFun).
 
 forget_node_for_queue(Q)
   when ?amqqueue_is_quorum(Q) ->
@@ -1950,21 +1950,17 @@ on_node_up(_Node) ->
 -spec on_node_down(node()) -> 'ok'.
 
 on_node_down(Node) ->
-    case delete_transient_queues_on_node(Node) of
-        ok ->
+    case rabbit_khepri:is_enabled() of
+        true ->
+            %% With Khepri, we don't delete transient/exclusive queues. There
+            %% may be a network partition and the node will be reachable again
+            %% after the partition is repaired.
+            %%
+            %% If the node will never come back, it will likely be removed from
+            %% the cluster. We take care of transient queues at that time.
             ok;
-        {error, timeout} ->
-            %% This case is possible when running Khepri. The node going down
-            %% could leave the cluster in a minority so the command to delete
-            %% the transient queue records would fail. Also see
-            %% `rabbit_khepri:init/0': we also try this deletion when the node
-            %% restarts - a time that the cluster is very likely to have a
-            %% majority - to ensure these records are deleted.
-            ?LOG_WARNING("transient queues for node '~ts' could not be "
-                               "deleted because of a timeout. These queues "
-                               "will be removed when node '~ts' restarts or "
-                               "is removed from the cluster.", [Node, Node]),
-            ok
+        false ->
+            ok = delete_transient_queues_on_node(Node)
     end.
 
 -spec delete_transient_queues_on_node(Node) -> Ret when
diff --git a/deps/rabbit/src/rabbit_db_queue.erl b/deps/rabbit/src/rabbit_db_queue.erl
@@ -32,7 +32,8 @@
          delete/2,
          update/2,
          update_decorators/2,
-         exists/1
+         exists/1,
+         foreach/2
         ]).
 
 %% Once mnesia is removed, all transient entities will be deleted. These can be replaced
@@ -57,8 +58,7 @@
 %% Only used by rabbit_amqqueue:forget_node_for_queue, which is only called
 %% by `rabbit_mnesia:remove_node_if_mnesia_running`. Thus, once mnesia and/or
 %% HA queues are removed it can be deleted.
--export([foreach_durable/2,
-         internal_delete/3]).
+-export([internal_delete/3]).
 
 %% Storing it on Khepri is not needed, this function is just used in
 %% rabbit_quorum_queue to ensure the queue is present in the rabbit_queue
@@ -1250,20 +1250,26 @@ foreach_transient_in_khepri(UpdateFun) ->
     end.
 
 %% -------------------------------------------------------------------
-%% foreach_durable().
+%% foreach().
 %% -------------------------------------------------------------------
 
--spec foreach_durable(UpdateFun, FilterFun) -> ok when
+-spec foreach(UpdateFun, FilterFun) -> ok when
       UpdateFun :: fun((Queue) -> any()),
       FilterFun :: fun((Queue) -> boolean()).
-%% @doc Applies `UpdateFun' to all durable queue records that match `FilterFun'.
+%% @doc Applies `UpdateFun' to all queue records that match `FilterFun'.
+%%
+%% With Mnesia, only durable queues are considered because we use the durable
+%% queues table.
+%%
+%% With Khepri, all queues are considered because they are all in the same
+%% "table".
 %%
 %% @private
 
-foreach_durable(UpdateFun, FilterFun) ->
+foreach(UpdateFun, FilterFun) ->
     rabbit_khepri:handle_fallback(
       #{mnesia => fun() -> foreach_durable_in_mnesia(UpdateFun, FilterFun) end,
-        khepri => fun() -> foreach_durable_in_khepri(UpdateFun, FilterFun) end
+        khepri => fun() -> foreach_in_khepri(UpdateFun, FilterFun) end
        }).
 
 foreach_durable_in_mnesia(UpdateFun, FilterFun) ->
@@ -1279,11 +1285,8 @@ foreach_durable_in_mnesia(UpdateFun, FilterFun) ->
           end),
     ok.
 
-foreach_durable_in_khepri(UpdateFun, FilterFun) ->
-    Path = khepri_queue_path(
-             ?KHEPRI_WILDCARD_STAR,
-             #if_data_matches{
-                pattern = amqqueue:pattern_match_on_durable(true)}),
+foreach_in_khepri(UpdateFun, FilterFun) ->
+    Path = khepri_queue_path(?KHEPRI_WILDCARD_STAR, ?KHEPRI_WILDCARD_STAR),
     case rabbit_khepri:filter(Path, fun(_, #{data := Q}) ->
                                             FilterFun(Q)
                                     end) of
diff --git a/deps/rabbit/src/rabbit_khepri.erl b/deps/rabbit/src/rabbit_khepri.erl
@@ -670,7 +670,7 @@ remove_reachable_member(NodeToRemove) ->
             NodeToRemove, khepri_cluster, reset, [?RA_CLUSTER_NAME]),
     case Ret of
         ok ->
-            rabbit_amqqueue:forget_all_durable(NodeToRemove),
+            rabbit_amqqueue:forget_all(NodeToRemove),
             ?LOG_DEBUG(
                "Node ~s removed from Khepri cluster \"~s\"",
                [NodeToRemove, ?RA_CLUSTER_NAME],
@@ -692,7 +692,7 @@ remove_down_member(NodeToRemove) ->
     Ret = ra:remove_member(ServerRef, ServerId, Timeout),
     case Ret of
         {ok, _, _} ->
-            rabbit_amqqueue:forget_all_durable(NodeToRemove),
+            rabbit_amqqueue:forget_all(NodeToRemove),
             ?LOG_DEBUG(
                "Node ~s removed from Khepri cluster \"~s\"",
                [NodeToRemove, ?RA_CLUSTER_NAME],
diff --git a/deps/rabbit/src/rabbit_mnesia.erl b/deps/rabbit/src/rabbit_mnesia.erl
@@ -916,7 +916,7 @@ remove_node_if_mnesia_running(Node) ->
             case mnesia:del_table_copy(schema, Node) of
                 {atomic, ok} ->
                     rabbit_node_monitor:notify_left_cluster(Node),
-                    rabbit_amqqueue:forget_all_durable(Node),
+                    rabbit_amqqueue:forget_all(Node),
                     ok;
                 {aborted, Reason} ->
                     {error, {failed_to_remove_node, Node, Reason}}
diff --git a/deps/rabbit/test/bindings_SUITE.erl b/deps/rabbit/test/bindings_SUITE.erl
@@ -96,25 +96,36 @@ end_per_group(_, Config) ->
                                 rabbit_ct_broker_helpers:teardown_steps()).
 
 init_per_testcase(Testcase, Config) ->
-    Config1 = rabbit_ct_helpers:testcase_started(Config, Testcase),
-    rabbit_ct_broker_helpers:rpc(Config, 0, ?MODULE, delete_queues, []),
-    Name = rabbit_data_coercion:to_binary(Testcase),
-    rabbit_ct_broker_helpers:rpc(Config, 0, ?MODULE, delete_exchange, [Name]),
-    Config2 = rabbit_ct_helpers:set_config(Config1,
-                                           [{queue_name, Name},
-                                            {alt_queue_name, <<Name/binary, "_alt">>},
-                                            {exchange_name, Name}
-                                           ]),
-    rabbit_ct_helpers:run_steps(Config2, rabbit_ct_client_helpers:setup_steps()).
+    case {Testcase, rabbit_ct_broker_helpers:configured_metadata_store(Config)} of
+        {transient_queue_on_node_down, khepri} ->
+            {skip, "Test irrelevant with Khepri"};
+        _ ->
+            Config1 = rabbit_ct_helpers:testcase_started(Config, Testcase),
+            rabbit_ct_broker_helpers:rpc(Config, 0, ?MODULE, delete_queues, []),
+            Name = rabbit_data_coercion:to_binary(Testcase),
+            rabbit_ct_broker_helpers:rpc(Config, 0, ?MODULE, delete_exchange, [Name]),
+            Config2 = rabbit_ct_helpers:set_config(
+                        Config1,
+                        [{queue_name, Name},
+                         {alt_queue_name, <<Name/binary, "_alt">>},
+                         {exchange_name, Name}
+                        ]),
+            rabbit_ct_helpers:run_steps(Config2, rabbit_ct_client_helpers:setup_steps())
+    end.
 
 end_per_testcase(Testcase, Config) ->
-    rabbit_ct_broker_helpers:rpc(Config, 0, ?MODULE, delete_queues, []),
-    rabbit_ct_broker_helpers:rpc(Config, 0, ?MODULE, delete_exchange,
-                                 [?config(exchange_name, Config)]),
-    Config1 = rabbit_ct_helpers:run_steps(
-                Config,
-                rabbit_ct_client_helpers:teardown_steps()),
-    rabbit_ct_helpers:testcase_finished(Config1, Testcase).
+    case {Testcase, rabbit_ct_broker_helpers:configured_metadata_store(Config)} of
+        {transient_queue_on_node_down, khepri} ->
+            Config;
+        _ ->
+            rabbit_ct_broker_helpers:rpc(Config, 0, ?MODULE, delete_queues, []),
+            rabbit_ct_broker_helpers:rpc(Config, 0, ?MODULE, delete_exchange,
+                                         [?config(exchange_name, Config)]),
+            Config1 = rabbit_ct_helpers:run_steps(
+                        Config,
+                        rabbit_ct_client_helpers:teardown_steps()),
+            rabbit_ct_helpers:testcase_finished(Config1, Testcase)
+    end.
 
 %% -------------------------------------------------------------------
 %% Testcases.
diff --git a/deps/rabbit/test/clustering_recovery_SUITE.erl b/deps/rabbit/test/clustering_recovery_SUITE.erl
diff --git a/deps/rabbit/test/rabbit_db_queue_SUITE.erl b/deps/rabbit/test/rabbit_db_queue_SUITE.erl
