Transactionally delete all exchanges during vhost deletion

Currently we delete each exchange one-by-one which requires three
commands: the delete itself plus a put and delete for a runtime
parameter that acts as a lock to prevent a client from declaring an
exchange while it's being deleted. The lock is unnecessary during vhost
deletion because permissions are cleared for the vhost before any
resources are deleted.

We can use a transaction to delete all exchanges and bindings for a
vhost in a single command against the Khepri store. This minimizes the
number of commands we need to send against the store and therefore the
latency of the deletion.

In a quick test with a vhost containing only 10,000 exchanges (no
bindings, queues, users, etc.), this is an order of magnitude speedup:
the prior commit takes 22s to delete the vhost while with this commit
the vhost is deleted in 2s.

Author: the-mikedavis

deps/rabbit/src/rabbit_db_exchange.erl

@@ -26,6 +26,7 @@
          peek_serial/1,
          next_serial/1,
          delete/2,
+         delete_all/1,
          delete_serial/1,
          recover/1,
          match/1,
@@ -657,6 +658,69 @@ delete_in_khepri(X = #exchange{name = XName}, OnlyDurable, RemoveBindingsForSour
     ok = khepri_tx:delete(khepri_exchange_path(XName)),
     rabbit_db_binding:delete_all_for_exchange_in_khepri(X, OnlyDurable, RemoveBindingsForSource).
 
+%% -------------------------------------------------------------------
+%% delete_all().
+%% -------------------------------------------------------------------
+
+-spec delete_all(VHostName) -> Ret when
+      VHostName :: vhost:name(),
+      Deletions :: rabbit_binding:deletions(),
+      Ret :: {ok, Deletions}.
+%% @doc Deletes all exchanges for a given vhost.
+%%
+%% @returns an `{ok, Deletions}' tuple containing the {@link
+%% rabbit_binding:deletions()} caused by deleting the exchanges under the given
+%% vhost.
+%%
+%% @private
+
+delete_all(VHostName) ->
+    rabbit_khepri:handle_fallback(
+      #{mnesia => fun() -> delete_all_in_mnesia(VHostName) end,
+        khepri => fun() -> delete_all_in_khepri(VHostName) end
+       }).
+
+delete_all_in_mnesia(VHostName) ->
+    rabbit_mnesia:execute_mnesia_transaction(
+      fun() ->
+              delete_all_in_mnesia_tx(VHostName)
+      end).
+
+delete_all_in_mnesia_tx(VHostName) ->
+    Match = #exchange{name = rabbit_misc:r(VHostName, exchange), _ = '_'},
+    Xs = mnesia:match_object(?MNESIA_TABLE, Match, write),
+    Deletions =
+    lists:foldl(
+      fun(X, Acc) ->
+              {deleted, #exchange{name = XName}, Bindings, XDeletions} =
+              unconditional_delete_in_mnesia( X, false),
+              XDeletions1 = rabbit_binding:add_deletion(
+                              XName, {X, deleted, Bindings}, XDeletions),
+              rabbit_binding:combine_deletions(Acc, XDeletions1)
+      end, rabbit_binding:new_deletions(), Xs),
+    {ok, Deletions}.
+
+delete_all_in_khepri(VHostName) ->
+    rabbit_khepri:transaction(
+      fun() ->
+              delete_all_in_khepri_tx(VHostName)
+      end, rw, #{timeout => infinity}).
+
+delete_all_in_khepri_tx(VHostName) ->
+    Pattern = khepri_exchange_path(VHostName, ?KHEPRI_WILDCARD_STAR),
+    {ok, NodeProps} = khepri_tx_adv:delete_many(Pattern),
+    Deletions =
+    maps:fold(
+      fun(_Path, #{data := X}, Deletions) ->
+              {deleted, #exchange{name = XName}, Bindings, XDeletions} =
+                rabbit_db_binding:delete_all_for_exchange_in_khepri(
+                  X, false, true),
+              Deletions1 = rabbit_binding:add_deletion(
+                             XName, {X, deleted, Bindings}, XDeletions),
+              rabbit_binding:combine_deletions(Deletions, Deletions1)
+      end, rabbit_binding:new_deletions(), NodeProps),
+    {ok, Deletions}.
+
 %% -------------------------------------------------------------------
 %% delete_serial().
 %% -------------------------------------------------------------------

deps/rabbit/src/rabbit_exchange.erl

@@ -14,7 +14,7 @@
          update_scratch/3, update_decorators/2, immutable/1,
          info_keys/0, info/1, info/2, info_all/1, info_all/2, info_all/4,
          route/2, route/3, delete/3, validate_binding/2, count/0,
-         ensure_deleted/3]).
+         ensure_deleted/3, delete_all/2]).
 -export([list_names/0]).
 -export([serialise_events/1]).
 -export([serial/1, peek_serial/1]).
@@ -484,6 +484,17 @@ delete(XName, IfUnused, Username) ->
                                         XName#resource.name, Username)
     end.
 
+-spec delete_all(VHostName, ActingUser) -> Ret when
+      VHostName :: vhost:name(),
+      ActingUser :: rabbit_types:username(),
+      Ret :: ok.
+
+delete_all(VHostName, ActingUser) ->
+    {ok, Deletions} = rabbit_db_exchange:delete_all(VHostName),
+    Deletions1 = rabbit_binding:process_deletions(Deletions),
+    rabbit_binding:notify_deletions(Deletions1, ActingUser),
+    ok.
+
 process_deletions({error, _} = E) ->
     E;
 process_deletions({deleted, #exchange{name = XName} = X, Bs, Deletions}) ->

deps/rabbit/src/rabbit_vhost.erl

@@ -299,8 +299,7 @@ delete(VHost, ActingUser) ->
          assert_benign(rabbit_amqqueue:with(Name, QDelFun), ActingUser)
      end || Q <- rabbit_amqqueue:list(VHost)],
     rabbit_log:info("Deleting exchanges in vhost '~ts' because it's being deleted", [VHost]),
-    [ok = rabbit_exchange:ensure_deleted(Name, false, ActingUser) ||
-        #exchange{name = Name} <- rabbit_exchange:list(VHost)],
+    ok = rabbit_exchange:delete_all(VHost, ActingUser),
     rabbit_log:info("Clearing policies and runtime parameters in vhost '~ts' because it's being deleted", [VHost]),
     _ = rabbit_runtime_parameters:clear_vhost(VHost, ActingUser),
     rabbit_log:debug("Removing vhost '~ts' from the metadata storage because it's being deleted", [VHost]),