rabbit_db_queue: Separate transient queue fetching from deletion

Author: the-mikedavis

deps/rabbit/src/rabbit_db_queue.erl

@@ -980,23 +980,30 @@ delete_transient_in_khepri(FilterFun) ->
         [?KHEPRI_WILDCARD_STAR,
          #if_data_matches{
             pattern = amqqueue:pattern_match_on_durable(false)}],
-    Ret = rabbit_khepri:fold(
-            PathPattern,
-            fun(Path, #{data := Queue}, Acc) ->
-                    case FilterFun(Queue) of
-                        true ->
-                            QueueName = khepri_queue_path_to_name(Path),
-                            case delete_in_khepri(QueueName, false) of
-                                ok        -> Acc;
-                                Deletions -> [{QueueName, Deletions} | Acc]
-                            end;
-                        false ->
-                            Acc
-                    end
-            end, [],
-            #{copy_tree_and_run_from_caller => true}),
-    case Ret of
-        {ok, Items} ->
+    %% The `FilterFun' might try to determine if the queue's process is alive.
+    %% This can cause a `calling_self' exception if we use the `FilterFun'
+    %% within the function passed to `khepri:fold/5' since the Khepri server
+    %% process might call itself. Instead we can fetch all of the transient
+    %% queues with `get_many' and then filter and fold the results outside of
+    %% Khepri's Ra server process.
+    case rabbit_khepri:get_many(PathPattern) of
+        {ok, Qs} ->
+            Items = maps:fold(
+                     fun(Path, #{data := Queue}, Acc) ->
+                             case FilterFun(Queue) of
+                                 true ->
+                                     QueueName = khepri_queue_path_to_name(
+                                                   Path),
+                                     case delete_in_khepri(QueueName, false) of
+                                         ok ->
+                                             Acc;
+                                         Deletions ->
+                                             [{QueueName, Deletions} | Acc]
+                                     end;
+                                 false ->
+                                     Acc
+                             end
+                     end, [], Qs),
             {QueueNames, Deletions} = lists:unzip(Items),
             {QueueNames, lists:flatten(Deletions)};
         {error, _} = Error ->

deps/rabbit/src/rabbit_khepri.erl

@@ -38,6 +38,7 @@
 
          get/1,
          get/2,
+         get_many/1,
          adv_get/1,
          match/1,
          match/2,
@@ -684,6 +685,9 @@ get(Path) ->
 get(Path, Options) ->
     khepri:get(?STORE_ID, Path, Options).
 
+get_many(PathPattern) ->
+    khepri:get_many(?STORE_ID, PathPattern, #{favor => low_latency}).
+
 adv_get(Path) ->
     khepri_adv:get(?STORE_ID, Path, #{favor => low_latency}).