Skip to content

Commit 816246e

Browse files
the-mikedavisdumbbell
authored andcommitted
rabbit_db_queue: Separate transient queue fetching from deletion
1 parent 99da1e6 commit 816246e

File tree

2 files changed

+28
-17
lines changed

2 files changed

+28
-17
lines changed

deps/rabbit/src/rabbit_db_queue.erl

Lines changed: 24 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -980,23 +980,30 @@ delete_transient_in_khepri(FilterFun) ->
980980
[?KHEPRI_WILDCARD_STAR,
981981
#if_data_matches{
982982
pattern = amqqueue:pattern_match_on_durable(false)}],
983-
Ret = rabbit_khepri:fold(
984-
PathPattern,
985-
fun(Path, #{data := Queue}, Acc) ->
986-
case FilterFun(Queue) of
987-
true ->
988-
QueueName = khepri_queue_path_to_name(Path),
989-
case delete_in_khepri(QueueName, false) of
990-
ok -> Acc;
991-
Deletions -> [{QueueName, Deletions} | Acc]
992-
end;
993-
false ->
994-
Acc
995-
end
996-
end, [],
997-
#{copy_tree_and_run_from_caller => true}),
998-
case Ret of
999-
{ok, Items} ->
983+
%% The `FilterFun' might try to determine if the queue's process is alive.
984+
%% This can cause a `calling_self' exception if we use the `FilterFun'
985+
%% within the function passed to `khepri:fold/5' since the Khepri server
986+
%% process might call itself. Instead we can fetch all of the transient
987+
%% queues with `get_many' and then filter and fold the results outside of
988+
%% Khepri's Ra server process.
989+
case rabbit_khepri:get_many(PathPattern) of
990+
{ok, Qs} ->
991+
Items = maps:fold(
992+
fun(Path, #{data := Queue}, Acc) ->
993+
case FilterFun(Queue) of
994+
true ->
995+
QueueName = khepri_queue_path_to_name(
996+
Path),
997+
case delete_in_khepri(QueueName, false) of
998+
ok ->
999+
Acc;
1000+
Deletions ->
1001+
[{QueueName, Deletions} | Acc]
1002+
end;
1003+
false ->
1004+
Acc
1005+
end
1006+
end, [], Qs),
10001007
{QueueNames, Deletions} = lists:unzip(Items),
10011008
{QueueNames, lists:flatten(Deletions)};
10021009
{error, _} = Error ->

deps/rabbit/src/rabbit_khepri.erl

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,7 @@
3838

3939
get/1,
4040
get/2,
41+
get_many/1,
4142
adv_get/1,
4243
match/1,
4344
match/2,
@@ -684,6 +685,9 @@ get(Path) ->
684685
get(Path, Options) ->
685686
khepri:get(?STORE_ID, Path, Options).
686687

688+
get_many(PathPattern) ->
689+
khepri:get_many(?STORE_ID, PathPattern, #{favor => low_latency}).
690+
687691
adv_get(Path) ->
688692
khepri_adv:get(?STORE_ID, Path, #{favor => low_latency}).
689693

0 commit comments

Comments
 (0)