Handle rabbit_amqqueue:internal_delete/3 failures in classic queues

The design of `rabbit_amqqueue_process` makes this change challenging.
The old implementation of the handler of the `{delete,_,_,_}` command
simply stopped the process and any cleanup was done in `gen_server2`'s
`terminate` callback. This makes it impossible to pass any error back
to the caller if the record can't be deleted from the metadata store
before a timeout.

The strategy taken here slightly mirrors an existing
`{shutdown, missing_owner}` termination value which can be returned from
`init_it2/3`. We pass the `ReplyTo` for the call with the state. We then
optionally reply to this `ReplyTo` if it is set in `terminate_delete/4`
with the result of `rabbit_amqqueue:internal_delete/3`. So deletion of
a classic queue will terminate the process but may return an error to
the caller if the record can't be removed from the metadata store
before the timeout.

(cherry picked from commit 2302eb9)

Author: the-mikedavis

deps/rabbit/src/rabbit_amqqueue.erl

@@ -1669,6 +1669,11 @@ delete_with(QueueName, ConnPid, IfUnused, IfEmpty, Username, CheckExclusive) whe
         {error, {exit, _, _}} ->
             %% delete()/delegate:invoke might return {error, {exit, _, _}}
             {ok, 0};
+        {error, timeout} ->
+            rabbit_misc:protocol_error(
+              internal_error,
+              "The operation to delete the queue from the metadata store "
+              "timed out", []);
         {ok, Count} ->
             {ok, Count};
         {protocol_error, Type, Reason, ReasonArgs} ->

deps/rabbit/src/rabbit_amqqueue_process.erl

@@ -297,7 +297,7 @@ terminate(shutdown = R,      State = #q{backing_queue = BQ, q = Q0}) ->
     end, State);
 terminate({shutdown, missing_owner = Reason}, {{reply_to, From}, #q{q = Q} = State}) ->
     %% if the owner was missing then there will be no queue, so don't emit stats
-    State1 = terminate_shutdown(terminate_delete(false, Reason, State), State),
+    State1 = terminate_shutdown(terminate_delete(false, Reason, none, State), State),
     send_reply(From, {owner_died, Q}),
     State1;
 terminate({shutdown, _} = R, State = #q{backing_queue = BQ}) ->
@@ -310,18 +310,22 @@ terminate(normal, State = #q{status = {terminated_by, auto_delete}}) ->
     %% thousands of queues. A optimisation introduced by server#1513
     %% needs to be reverted by this case, avoiding to guard the delete
     %% operation on `rabbit_durable_queue`
-    terminate_shutdown(terminate_delete(true, auto_delete, State), State);
-terminate(normal,            State) -> %% delete case
-    terminate_shutdown(terminate_delete(true, normal, State), State);
+    terminate_shutdown(terminate_delete(true, auto_delete, none, State), State);
+terminate(normal, {{reply_to, ReplyTo}, State}) -> %% delete case
+    terminate_shutdown(terminate_delete(true, normal, ReplyTo, State), State);
+terminate(normal, State) ->
+    terminate_shutdown(terminate_delete(true, normal, none, State), State);
 %% If we crashed don't try to clean up the BQS, probably best to leave it.
 terminate(_Reason,           State = #q{q = Q}) ->
     terminate_shutdown(fun (BQS) ->
                                Q2 = amqqueue:set_state(Q, crashed),
+                               %% When mnesia is removed this update can become
+                               %% an async Khepri command.
                                _ = rabbit_amqqueue:store_queue(Q2),
                                BQS
                        end, State).
 
-terminate_delete(EmitStats, Reason0,
+terminate_delete(EmitStats, Reason0, ReplyTo,
                  State = #q{q = Q,
                             backing_queue = BQ,
                             status = Status}) ->
@@ -332,19 +336,24 @@ terminate_delete(EmitStats, Reason0,
                      missing_owner -> normal;
                      Any -> Any
                  end,
+        Len = BQ:len(BQS),
         BQS1 = BQ:delete_and_terminate(Reason, BQS),
         if EmitStats -> rabbit_event:if_enabled(State, #q.stats_timer,
                                                 fun() -> emit_stats(State) end);
            true      -> ok
         end,
         %% This try-catch block transforms throws to errors since throws are not
-        %% logged.
-        try
-            %% don't care if the internal delete doesn't return 'ok'.
-            rabbit_amqqueue:internal_delete(Q, ActingUser, Reason0)
-        catch
-            {error, ReasonE} -> error(ReasonE)
-        end,
+        %% logged. When mnesia is removed this `try` can be removed: Khepri
+        %% returns errors as error tuples instead.
+        Reply = try rabbit_amqqueue:internal_delete(Q, ActingUser, Reason0) of
+                    ok ->
+                        {ok, Len};
+                    {error, _} = Err ->
+                        Err
+                catch
+                    {error, ReasonE} -> error(ReasonE)
+                end,
+        send_reply(ReplyTo, Reply),
         BQS1
     end.
 
@@ -1396,15 +1405,16 @@ handle_call(stat, _From, State) ->
         ensure_expiry_timer(State),
     reply({ok, BQ:len(BQS), rabbit_queue_consumers:count()}, State1);
 
-handle_call({delete, IfUnused, IfEmpty, ActingUser}, _From,
+handle_call({delete, IfUnused, IfEmpty, ActingUser}, From,
             State = #q{backing_queue_state = BQS, backing_queue = BQ}) ->
     IsEmpty  = BQ:is_empty(BQS),
     IsUnused = is_unused(State),
     if
         IfEmpty  and not(IsEmpty)  -> reply({error, not_empty}, State);
         IfUnused and not(IsUnused) -> reply({error,    in_use}, State);
-        true                       -> stop({ok, BQ:len(BQS)},
-                                           State#q{status = {terminated_by, ActingUser}})
+        true ->
+            State1 = State#q{status = {terminated_by, ActingUser}},
+            stop({{reply_to, From}, State1})
     end;
 
 handle_call(purge, _From, State = #q{backing_queue       = BQ,