Merge pull request #9324 from Ayanda-D/robust-ctl-delete-queue

Make rabbitmqctl delete_queue more robust

Author: michaelklishin

deps/rabbit/app.bzl

@@ -47,6 +47,7 @@ def all_beam_files(name = "all_beam_files"):
             "src/rabbit_access_control.erl",
             "src/rabbit_alarm.erl",
             "src/rabbit_amqqueue.erl",
+            "src/rabbit_amqqueue_control.erl",
             "src/rabbit_amqqueue_process.erl",
             "src/rabbit_amqqueue_sup.erl",
             "src/rabbit_amqqueue_sup_sup.erl",
@@ -297,6 +298,7 @@ def all_test_beam_files(name = "all_test_beam_files"):
             "src/rabbit_access_control.erl",
             "src/rabbit_alarm.erl",
             "src/rabbit_amqqueue.erl",
+            "src/rabbit_amqqueue_control.erl",
             "src/rabbit_amqqueue_process.erl",
             "src/rabbit_amqqueue_sup.erl",
             "src/rabbit_amqqueue_sup_sup.erl",
@@ -559,6 +561,7 @@ def all_srcs(name = "all_srcs"):
             "src/rabbit_access_control.erl",
             "src/rabbit_alarm.erl",
             "src/rabbit_amqqueue.erl",
+            "src/rabbit_amqqueue_control.erl",
             "src/rabbit_amqqueue_process.erl",
             "src/rabbit_amqqueue_sup.erl",
             "src/rabbit_amqqueue_sup_sup.erl",

deps/rabbit/src/rabbit_amqqueue.erl

@@ -53,8 +53,9 @@
 -export([delete_crashed/1,
          delete_crashed/2,
          delete_crashed_internal/2]).
-
+-export([delete_with/4, delete_with/6]).
 -export([pid_of/1, pid_of/2]).
+-export([pid_or_crashed/2]).
 -export([mark_local_durable_queues_stopped/1]).
 
 -export([rebalance/3]).
@@ -71,6 +72,8 @@
 -export([prepend_extra_bcc/1]).
 -export([queue/1, queue_names/1]).
 
+-export([kill_queue/2, kill_queue/3, kill_queue_hard/2, kill_queue_hard/3]).
+
 %% internal
 -export([internal_declare/2, internal_delete/2, run_backing_queue/3,
          set_ram_duration_target/2, set_maximum_since_use/2,
@@ -116,6 +119,7 @@
 -define(CONSUMER_INFO_KEYS,
         [queue_name, channel_pid, consumer_tag, ack_required, prefetch_count,
          active, activity_status, arguments]).
+-define(KILL_QUEUE_DELAY_INTERVAL, 100).
 
 warn_file_limit() ->
     DurableQueues = find_recoverable_queues(),
@@ -1601,6 +1605,51 @@ delete_immediately_by_resource(Resources) ->
 delete(Q, IfUnused, IfEmpty, ActingUser) ->
     rabbit_queue_type:delete(Q, IfUnused, IfEmpty, ActingUser).
 
+-spec delete_with(amqqueue:amqqueue() | name(), boolean(), boolean(), rabbit_types:username()) ->
+    rabbit_types:ok(integer()) | rabbit_misc:channel_or_connection_exit().
+delete_with(QueueName, IfUnused, IfEmpty, ActingUser) ->
+    delete_with(QueueName, undefined, IfUnused, IfEmpty, ActingUser, false).
+
+-spec delete_with(amqqueue:amqqueue() | name(), pid() | undefined, boolean(), boolean(), rabbit_types:username(), boolean()) ->
+    rabbit_types:ok(integer()) | rabbit_misc:channel_or_connection_exit().
+delete_with(AMQQueue, ConnPid, IfUnused, IfEmpty, Username, CheckExclusive) when ?is_amqqueue(AMQQueue) ->
+    QueueName = amqqueue:get_name(AMQQueue),
+    delete_with(QueueName, ConnPid, IfUnused, IfEmpty, Username, CheckExclusive);
+delete_with(QueueName, ConnPid, IfUnused, IfEmpty, Username, CheckExclusive) when is_record(QueueName, resource) ->
+    case with(
+           QueueName,
+           fun (Q) ->
+                if CheckExclusive ->
+                    check_exclusive_access(Q, ConnPid);
+                true ->
+                    ok
+                end,
+                rabbit_queue_type:delete(Q, IfUnused, IfEmpty, Username)
+           end,
+           fun (not_found) ->
+                   {ok, 0};
+               ({absent, Q, crashed}) ->
+                   _ = delete_crashed(Q, Username),
+                   {ok, 0};
+               ({absent, Q, stopped}) ->
+                   _ = delete_crashed(Q, Username),
+                   {ok, 0};
+               ({absent, Q, Reason}) ->
+                   absent(Q, Reason)
+           end) of
+        {error, in_use} ->
+            rabbit_misc:precondition_failed("~ts in use", [rabbit_misc:rs(QueueName)]);
+        {error, not_empty} ->
+            rabbit_misc:precondition_failed("~ts not empty", [rabbit_misc:rs(QueueName)]);
+        {error, {exit, _, _}} ->
+            %% rabbit_amqqueue:delete()/delegate:invoke might return {error, {exit, _, _}}
+            {ok, 0};
+        {ok, Count} ->
+            {ok, Count};
+        {protocol_error, Type, Reason, ReasonArgs} ->
+            rabbit_misc:protocol_error(Type, Reason, ReasonArgs)
+    end.
+
 %% delete_crashed* INCLUDED FOR BACKWARDS COMPATBILITY REASONS
 delete_crashed(Q) when ?amqqueue_is_classic(Q) ->
     rabbit_classic_queue:delete_crashed(Q).
@@ -2061,3 +2110,61 @@ is_queue_args_combination_permitted(Durable, Exclusive) ->
         true ->
             rabbit_deprecated_features:is_permitted(transient_nonexcl_queues)
     end.
+
+-spec kill_queue_hard(node(), name()) -> ok.
+kill_queue_hard(Node, QRes = #resource{kind = queue}) ->
+    kill_queue_hard(Node, QRes, boom).
+
+-spec kill_queue_hard(node(), name(), atom()) -> ok.
+kill_queue_hard(Node, QRes = #resource{kind = queue}, Reason) ->
+    case kill_queue(Node, QRes, Reason) of
+        crashed -> ok;
+        stopped -> ok;
+        NewPid when is_pid(NewPid) ->
+            timer:sleep(?KILL_QUEUE_DELAY_INTERVAL),
+            kill_queue_hard(Node, QRes, Reason);
+        Error -> Error
+    end.
+
+-spec kill_queue(node(), name()) -> pid() | crashed | stopped | rabbit_types:error(term()).
+kill_queue(Node, QRes = #resource{kind = queue}) ->
+    kill_queue(Node, QRes, boom).
+
+-spec kill_queue(node(), name(), atom()) -> pid() | crashed | stopped | rabbit_types:error(term()).
+kill_queue(Node, QRes = #resource{kind = queue}, Reason = shutdown) ->
+    Pid1 = pid_or_crashed(Node, QRes),
+    exit(Pid1, Reason),
+    rabbit_amqqueue_control:await_state(Node, QRes, stopped),
+    stopped;
+kill_queue(Node, QRes = #resource{kind = queue}, Reason) ->
+    case pid_or_crashed(Node, QRes) of
+        Pid1 when is_pid(Pid1) ->
+            exit(Pid1, Reason),
+            rabbit_amqqueue_control:await_new_pid(Node, QRes, Pid1);
+        crashed ->
+            crashed;
+        Error ->
+            Error
+    end.
+
+-spec pid_or_crashed(node(), name()) -> pid() | crashed | rabbit_types:error(term()).
+pid_or_crashed(Node, QRes = #resource{virtual_host = VHost, kind = queue}) ->
+    case rpc:call(Node, rabbit_amqqueue, lookup, [QRes]) of
+        {ok, Q} ->
+            QPid = amqqueue:get_pid(Q),
+            State = amqqueue:get_state(Q),
+            case State of
+                crashed ->
+                    case rabbit_amqqueue_sup_sup:find_for_vhost(VHost, Node) of
+                        {error, {queue_supervisor_not_found, _}} -> {error, no_sup};
+                        {ok, SPid} ->
+                            case rabbit_misc:remote_sup_child(Node, SPid) of
+                            {ok, _}           -> QPid;   %% restarting
+                            {error, no_child} -> crashed %% given up
+                            end
+                    end;
+                _       -> QPid
+            end;
+        Error = {error, _} -> Error;
+        Reason             -> {error, Reason}
+    end.

deps/rabbit/src/rabbit_amqqueue_control.erl

@@ -0,0 +1,57 @@
+% This Source Code Form is subject to the terms of the Mozilla Public
+%% License, v. 2.0. If a copy of the MPL was not distributed with this
+%% file, You can obtain one at https://mozilla.org/MPL/2.0/.
+%%
+%% Copyright (c) 2007-2023 VMware, Inc. or its affiliates.  All rights reserved.
+%%
+
+-module(rabbit_amqqueue_control).
+
+-export([await_new_pid/3, await_state/3, await_state/4]).
+
+-define(DEFAULT_AWAIT_STATE_TIMEOUT,  30000).
+-define(AWAIT_NEW_PID_DELAY_INTERVAL, 10).
+-define(AWAIT_STATE_DELAY_INTERVAL,   100).
+-define(AWAIT_STATE_DELAY_TIME_DELTA, 100).
+
+-include_lib("rabbit_common/include/resource.hrl").
+
+-spec await_new_pid(node(), rabbit_amqqueue:name(), pid()) -> pid().
+await_new_pid(Node, QRes = #resource{kind = queue}, OldPid) ->
+    case rabbit_amqqueue:pid_or_crashed(Node, QRes) of
+        OldPid -> timer:sleep(?AWAIT_NEW_PID_DELAY_INTERVAL),
+                  await_new_pid(Node, QRes, OldPid);
+        New    -> New
+    end.
+
+-spec await_state(node(), rabbit_amqqueue:name() | binary(), atom()) -> 'ok'.
+await_state(Node, QName, State) when is_binary(QName) ->
+    QRes = rabbit_misc:r(<<"/">>, queue, QName),
+    await_state(Node, QRes, State);
+await_state(Node, QRes = #resource{kind = queue}, State) ->
+    await_state(Node, QRes, State, ?DEFAULT_AWAIT_STATE_TIMEOUT).
+
+-spec await_state(node(), rabbit_amqqueue:name() | binary(), atom(), integer()) -> 'ok'.
+await_state(Node, QName, State, Time) when is_binary(QName) ->
+    QRes = rabbit_misc:r(<<"/">>, queue, QName),
+    await_state(Node, QRes, State, Time);
+await_state(Node, QRes = #resource{kind = queue}, State, Time) ->
+    case state(Node, QRes) of
+        State ->
+            ok;
+        Other ->
+            case Time of
+                0 -> exit({timeout_awaiting_state, State, Other});
+                _ -> timer:sleep(?AWAIT_STATE_DELAY_INTERVAL),
+                     await_state(Node, QRes, State, Time - ?AWAIT_STATE_DELAY_TIME_DELTA)
+            end
+    end.
+
+state(Node, QRes = #resource{virtual_host = VHost, kind = queue}) ->
+    Infos = rpc:call(Node, rabbit_amqqueue, info_all, [VHost, [name, state]]),
+    fetch_state(QRes, Infos).
+
+fetch_state(_QRes, [])                                   -> undefined;
+fetch_state(QRes,  [[{name, QRes}, {state, State}] | _]) -> State;
+fetch_state(QRes,  [[{name, _}, {state, _State}] | Rem]) ->
+    fetch_state(QRes, Rem).

deps/rabbit/src/rabbit_channel.erl

@@ -930,15 +930,6 @@ handle_exception(Reason, State = #ch{cfg = #conf{protocol = Protocol,
             {stop, normal, State1}
     end.
 
--spec precondition_failed(string()) -> no_return().
-
-precondition_failed(Format) -> precondition_failed(Format, []).
-
--spec precondition_failed(string(), [any()]) -> no_return().
-
-precondition_failed(Format, Params) ->
-    rabbit_misc:protocol_error(precondition_failed, Format, Params).
-
 return_queue_declare_ok(#resource{name = ActualName},
                         NoWait, MessageCount, ConsumerCount,
                         #ch{cfg = Cfg} = State) ->
@@ -995,15 +986,15 @@ check_user_id_header(#'P_basic'{user_id = Claimed},
                                                   tags     = Tags}}}) ->
     case lists:member(impersonator, Tags) of
         true  -> ok;
-        false -> precondition_failed(
+        false -> rabbit_misc:precondition_failed(
                    "user_id property set to '~ts' but authenticated user was "
                    "'~ts'", [Claimed, Actual])
     end.
 
 check_expiration_header(Props) ->
     case rabbit_basic:parse_expiration(Props) of
         {ok, _}    -> ok;
-        {error, E} -> precondition_failed("invalid expiration '~ts': ~tp",
+        {error, E} -> rabbit_misc:precondition_failed("invalid expiration '~ts': ~tp",
                                           [Props#'P_basic'.expiration, E])
     end.
 
@@ -1074,15 +1065,15 @@ check_msg_size(Content, MaxMessageSize, GCThreshold) ->
                 _ ->
                     "message size ~B is larger than configured max size ~B"
             end,
-            precondition_failed(ErrorMessage,
+            rabbit_misc:precondition_failed(ErrorMessage,
                                 [Size, MaxMessageSize]);
         _ -> ok
     end.
 
 check_vhost_queue_limit(#resource{name = QueueName}, VHost) ->
   case rabbit_vhost_limit:is_over_queue_limit(VHost) of
     false         -> ok;
-    {true, Limit} -> precondition_failed("cannot declare queue '~ts': "
+    {true, Limit} -> rabbit_misc:precondition_failed("cannot declare queue '~ts': "
                                "queue limit in vhost '~ts' (~tp) is reached",
                                [QueueName, VHost, Limit])
 
@@ -1704,7 +1695,7 @@ handle_method(#'queue.purge'{nowait = NoWait} = Method,
     end;
 
 handle_method(#'tx.select'{}, _, #ch{confirm_enabled = true}) ->
-    precondition_failed("cannot switch from confirm to tx mode");
+    rabbit_misc:precondition_failed("cannot switch from confirm to tx mode");
 
 handle_method(#'tx.select'{}, _, State = #ch{tx = none}) ->
     {reply, #'tx.select_ok'{}, State#ch{tx = new_tx()}};
@@ -1713,7 +1704,7 @@ handle_method(#'tx.select'{}, _, State) ->
     {reply, #'tx.select_ok'{}, State};
 
 handle_method(#'tx.commit'{}, _, #ch{tx = none}) ->
-    precondition_failed("channel is not transactional");
+    rabbit_misc:precondition_failed("channel is not transactional");
 
 handle_method(#'tx.commit'{}, _, State = #ch{tx      = {Deliveries, Acks},
                                              limiter = Limiter}) ->
@@ -1731,7 +1722,7 @@ handle_method(#'tx.commit'{}, _, State = #ch{tx      = {Deliveries, Acks},
     {noreply, maybe_complete_tx(State3#ch{tx = committing})};
 
 handle_method(#'tx.rollback'{}, _, #ch{tx = none}) ->
-    precondition_failed("channel is not transactional");
+    rabbit_misc:precondition_failed("channel is not transactional");
 
 handle_method(#'tx.rollback'{}, _, State = #ch{unacked_message_q = UAMQ,
                                                tx = {_Msgs, Acks}}) ->
@@ -1741,7 +1732,7 @@ handle_method(#'tx.rollback'{}, _, State = #ch{unacked_message_q = UAMQ,
                                           tx                = new_tx()}};
 
 handle_method(#'confirm.select'{}, _, #ch{tx = {_, _}}) ->
-    precondition_failed("cannot switch from tx to confirm mode");
+    rabbit_misc:precondition_failed("cannot switch from tx to confirm mode");
 
 handle_method(#'confirm.select'{nowait = NoWait}, _, State) ->
     return_ok(State#ch{confirm_enabled = true},
@@ -1762,7 +1753,7 @@ handle_method(#'basic.credit'{consumer_tag = CTag,
         {ok, {Q, _CParams}} ->
             {ok, QStates, Actions} = rabbit_queue_type:credit(Q, CTag, Credit, Drain, QStates0),
             {noreply, handle_queue_actions(Actions, State#ch{queue_states = QStates})};
-        error -> precondition_failed(
+        error -> rabbit_misc:precondition_failed(
                    "unknown consumer tag '~ts'", [CTag])
     end;
 
@@ -2050,7 +2041,7 @@ collect_acks(AcknowledgedAcc, RemainingAcc, UAMQ, DeliveryTag, Multiple) ->
                                  UAMQTail, DeliveryTag, Multiple)
             end;
         {empty, _} ->
-            precondition_failed("unknown delivery tag ~w", [DeliveryTag])
+            rabbit_misc:precondition_failed("unknown delivery tag ~w", [DeliveryTag])
     end.
 
 %% Settles (acknowledges) messages at the queue replica process level.
@@ -2540,7 +2531,7 @@ handle_method(#'queue.declare'{queue       = QueueNameBin,
                undefined ->
                    ok;
                {error, {invalid_type, Type}} ->
-                    precondition_failed(
+                    rabbit_misc:precondition_failed(
                       "invalid type '~ts' for arg '~ts' in ~ts",
                       [Type, DlxKey, rabbit_misc:rs(QueueName)]);
                DLX ->
@@ -2605,35 +2596,7 @@ handle_method(#'queue.delete'{queue     = QueueNameBin,
     QueueName = qbin_to_resource(StrippedQueueNameBin, VHostPath),
 
     check_configure_permitted(QueueName, User, AuthzContext),
-    case rabbit_amqqueue:with(
-           QueueName,
-           fun (Q) ->
-                   rabbit_amqqueue:check_exclusive_access(Q, ConnPid),
-                   rabbit_queue_type:delete(Q, IfUnused, IfEmpty, Username)
-           end,
-           fun (not_found) ->
-                   {ok, 0};
-               ({absent, Q, crashed}) ->
-                   _ = rabbit_classic_queue:delete_crashed(Q, Username),
-                   {ok, 0};
-               ({absent, Q, stopped}) ->
-                   _ = rabbit_classic_queue:delete_crashed(Q, Username),
-                   {ok, 0};
-               ({absent, Q, Reason}) ->
-                   rabbit_amqqueue:absent(Q, Reason)
-           end) of
-        {error, in_use} ->
-            precondition_failed("~ts in use", [rabbit_misc:rs(QueueName)]);
-        {error, not_empty} ->
-            precondition_failed("~ts not empty", [rabbit_misc:rs(QueueName)]);
-        {error, {exit, _, _}} ->
-            %% rabbit_amqqueue:delete()/delegate:invoke might return {error, {exit, _, _}}
-            {ok, 0};
-        {ok, Count} ->
-            {ok, Count};
-        {protocol_error, Type, Reason, ReasonArgs} ->
-            rabbit_misc:protocol_error(Type, Reason, ReasonArgs)
-    end;
+    rabbit_amqqueue:delete_with(QueueName, ConnPid, IfUnused, IfEmpty, Username, true);
 handle_method(#'exchange.delete'{exchange  = ExchangeNameBin,
                                  if_unused = IfUnused},
               _ConnPid, AuthzContext, _CollectorPid, VHostPath,
@@ -2647,7 +2610,7 @@ handle_method(#'exchange.delete'{exchange  = ExchangeNameBin,
         {error, not_found} ->
             ok;
         {error, in_use} ->
-            precondition_failed("~ts in use", [rabbit_misc:rs(ExchangeName)]);
+            rabbit_misc:precondition_failed("~ts in use", [rabbit_misc:rs(ExchangeName)]);
         ok ->
             ok
     end;
@@ -2689,7 +2652,7 @@ handle_method(#'exchange.declare'{exchange    = ExchangeNameBin,
                 case rabbit_misc:r_arg(VHostPath, exchange, Args, AeKey) of
                     undefined -> ok;
                     {error, {invalid_type, Type}} ->
-                        precondition_failed(
+                        rabbit_misc:precondition_failed(
                           "invalid type '~ts' for arg '~ts' in ~ts",
                           [Type, AeKey, rabbit_misc:rs(ExchangeName)]);
                     AName     -> check_read_permitted(ExchangeName, User, AuthzContext),