Merge pull request #15287 from rabbitmq/mergify/bp/v4.1.x/pr-15286

Check exclusive queue owner before deleting a queue (backport #15276) (backport #15286)

Author: michaelklishin

deps/rabbit/src/amqqueue.erl

@@ -72,6 +72,7 @@
          pattern_match_on_durable/1,
          pattern_match_on_type_and_durable/2,
          pattern_match_on_type_and_vhost/2,
+         pattern_match_on_exclusive_owner/1,
          reset_decorators/1,
          set_immutable/1,
          qnode/1,
@@ -594,6 +595,11 @@ pattern_match_on_type_and_durable(Type, IsDurable) ->
 pattern_match_on_type_and_vhost(Type, VHost) ->
     #amqqueue{type = Type, vhost = VHost, _ = '_'}.
 
+-spec pattern_match_on_exclusive_owner(pid() | none) -> amqqueue_pattern().
+
+pattern_match_on_exclusive_owner(Owner) ->
+    #amqqueue{exclusive_owner = Owner, _ = '_'}.
+
 -spec reset_decorators(amqqueue()) -> amqqueue().
 
 reset_decorators(#amqqueue{} = Queue) ->

deps/rabbit/src/rabbit_amqqueue.erl

@@ -79,6 +79,7 @@
 %% Deprecated feature callback.
 -export([are_transient_nonexcl_used/1]).
 
+-include_lib("khepri/include/khepri.hrl").
 -include_lib("rabbit_common/include/rabbit.hrl").
 -include_lib("stdlib/include/qlc.hrl").
 -include("amqqueue.hrl").
@@ -1863,7 +1864,14 @@ internal_delete(Queue, ActingUser) ->
 
 internal_delete(Queue, ActingUser, Reason) ->
     QueueName = amqqueue:get_name(Queue),
-    case rabbit_db_queue:delete(QueueName, Reason) of
+    Conditions = case amqqueue:get_exclusive_owner(Queue) of
+                     none ->
+                         [];
+                     Owner ->
+                         Pattern = amqqueue:pattern_match_on_exclusive_owner(Owner),
+                         [#if_data_matches{pattern = Pattern}]
+                 end,
+    case rabbit_db_queue:delete_if(QueueName, Conditions, Reason) of
         ok ->
             ok;
         {error, timeout} = Err ->

deps/rabbit/src/rabbit_db_queue.erl

@@ -30,6 +30,7 @@
          create_or_get/1,
          set/1,
          delete/2,
+         delete_if/3,
          update/2,
          update_decorators/2,
          exists/1
@@ -384,34 +385,51 @@ list_for_count_in_khepri(VHostName) ->
              rabbit_khepri:timeout_error().
 
 delete(QueueName, Reason) ->
+    delete_if(QueueName, [], Reason).
+
+-spec delete_if(QName, Conditions, Reason) -> Ret when
+      QName :: rabbit_amqqueue:name(),
+      Conditions :: [khepri_condition:condition()],
+      Reason :: atom(),
+      Ret :: ok |
+      Deletions :: rabbit_binding:deletions() |
+      rabbit_khepri:timeout_error().
+
+delete_if(QueueName, Conditions, Reason) ->
     rabbit_khepri:handle_fallback(
-      #{mnesia => fun() -> delete_in_mnesia(QueueName, Reason) end,
-        khepri => fun() -> delete_in_khepri(QueueName) end
+      #{mnesia => fun() -> delete_in_mnesia(QueueName, Conditions, Reason) end,
+        khepri => fun() -> delete_in_khepri(QueueName, Conditions) end
        }).
 
-delete_in_mnesia(QueueName, Reason) ->
+delete_in_mnesia(QueueName, Conditions, Reason) ->
     rabbit_mnesia:execute_mnesia_transaction(
       fun () ->
               case {mnesia:wread({?MNESIA_TABLE, QueueName}),
                     mnesia:wread({?MNESIA_DURABLE_TABLE, QueueName})} of
                   {[], []} ->
                       ok;
-                  _ ->
-                      OnlyDurable = case Reason of
-                                        missing_owner -> true;
-                                        _ -> false
-                                    end,
-                      internal_delete_in_mnesia(QueueName, OnlyDurable, Reason)
+                  {Qs, DurableQs} ->
+                      case queue_matches_conditions(Qs ++ DurableQs, Conditions) of
+                          true ->
+                              OnlyDurable = case Reason of
+                                                missing_owner -> true;
+                                                _ -> false
+                                            end,
+                              internal_delete_in_mnesia(QueueName, OnlyDurable, Reason);
+                          false ->
+                              ok
+                      end
               end
       end).
 
-delete_in_khepri(QueueName) ->
-    delete_in_khepri(QueueName, false).
+delete_in_khepri(QueueName, Conditions) ->
+    delete_in_khepri(QueueName, Conditions, false).
 
-delete_in_khepri(QueueName, OnlyDurable) ->
+delete_in_khepri(QueueName, Conditions, OnlyDurable) ->
+    Path0 = khepri_queue_path(QueueName),
+    Path = khepri_path:combine_with_conditions(Path0, Conditions),
     rabbit_khepri:transaction(
       fun () ->
-              Path = khepri_queue_path(QueueName),
               case khepri_tx_adv:delete(Path) of
                   {ok, #{data := _}} ->
                       %% we want to execute some things, as decided by rabbit_exchange,
@@ -438,7 +456,7 @@ internal_delete(QueueName, OnlyDurable, Reason) ->
     %% HA queues are removed it can be removed.
     rabbit_khepri:handle_fallback(
       #{mnesia => fun() -> internal_delete_in_mnesia(QueueName, OnlyDurable, Reason) end,
-        khepri => fun() -> delete_in_khepri(QueueName, OnlyDurable) end
+        khepri => fun() -> delete_in_khepri(QueueName, [], OnlyDurable) end
        }).
 
 internal_delete_in_mnesia(QueueName, OnlyDurable, Reason) ->
@@ -1360,6 +1378,20 @@ clear_in_khepri() ->
 %% Internal
 %% --------------------------------------------------------------
 
+queue_matches_conditions(_, []) ->
+    true;
+queue_matches_conditions([], _Conditions) ->
+    false;
+queue_matches_conditions([Q | _], [#if_data_matches{pattern = Pattern,
+                                                     conditions = Conditions} | Rest]) ->
+    MatchSpec = ets:match_spec_compile([{Pattern, Conditions, [match]}]),
+    case ets:match_spec_run([Q], MatchSpec) of
+        [match] -> queue_matches_conditions([Q], Rest);
+        [] -> false
+    end;
+queue_matches_conditions(_, [Condition | _]) ->
+    error({unsupported_condition, Condition}).
+
 list_with_possible_retry_in_mnesia(Fun) ->
     %% amqqueue migration:
     %% The `rabbit_queue' or `rabbit_durable_queue' tables

deps/rabbit/test/rabbit_db_queue_SUITE.erl

@@ -7,6 +7,7 @@
 
 -module(rabbit_db_queue_SUITE).
 
+-include_lib("khepri/include/khepri.hrl").
 -include_lib("rabbit_common/include/rabbit.hrl").
 -include_lib("eunit/include/eunit.hrl").
 -include("amqqueue.hrl").
@@ -42,6 +43,7 @@ all_tests() ->
      count_by_vhost,
      set,
      delete,
+     delete_exclusive_queue,
      update,
      update_decorators,
      exists,
@@ -62,7 +64,8 @@ mnesia_tests() ->
      foreach_transient,
      delete_transient,
      update_in_mnesia_tx,
-     get_durable_in_mnesia_tx
+     get_durable_in_mnesia_tx,
+     delete_exclusive_queue
     ].
 
 %% -------------------------------------------------------------------
@@ -323,6 +326,34 @@ delete1(_Config) ->
     ?assertEqual({error, not_found}, rabbit_db_queue:get(QName)),
     passed.
 
+delete_exclusive_queue(Config) ->
+    passed = rabbit_ct_broker_helpers:rpc(
+               Config, 0, ?MODULE, delete_exclusive_queue1, [Config]).
+
+delete_exclusive_queue1(_Config) ->
+    QName = rabbit_misc:r(?VHOST, queue, <<"test-exclusive-queue">>),
+    Owner = spawn_link(fun() -> receive stop -> ok end end),
+    NotOwner = spawn_link(fun() -> receive stop -> ok end end),
+    Q = new_exclusive_queue(QName, rabbit_classic_queue, Owner),
+    ?assertEqual(ok, rabbit_db_queue:set(Q)),
+    ?assertEqual({ok, Q}, rabbit_db_queue:get(QName)),
+    %% Delete with wrong owner should not delete the queue
+    NotOwnerPattern = amqqueue:pattern_match_on_exclusive_owner(NotOwner),
+    NotOwnerConditions = [#if_data_matches{pattern = NotOwnerPattern}],
+    ?assertEqual(ok, rabbit_db_queue:delete_if(QName, NotOwnerConditions, normal)),
+    ?assertEqual({ok, Q}, rabbit_db_queue:get(QName)),
+    %% Delete with correct owner should delete the queue
+    OwnerPattern = amqqueue:pattern_match_on_exclusive_owner(Owner),
+    OwnerConditions = [#if_data_matches{pattern = OwnerPattern}],
+    Deletions = rabbit_db_queue:delete_if(QName, OwnerConditions, normal),
+    ?assertEqual(rabbit_binding:new_deletions(), Deletions),
+    ?assertEqual({error, not_found}, rabbit_db_queue:get(QName)),
+    unlink(Owner),
+    unlink(NotOwner),
+    Owner ! stop,
+    NotOwner ! stop,
+    passed.
+
 update(Config) ->
     passed = rabbit_ct_broker_helpers:rpc(Config, 0, ?MODULE, update1, [Config]).
 
@@ -620,3 +651,6 @@ new_queue(QName, Type) ->
 
 new_queue(#resource{virtual_host = VHost} = QName, Type, Pid) ->
     amqqueue:new(QName, Pid, true, false, none, [], VHost, #{}, Type).
+
+new_exclusive_queue(#resource{virtual_host = VHost} = QName, Type, Owner) ->
+    amqqueue:new(QName, none, true, false, Owner, [], VHost, #{}, Type).