Merge pull request #13732 from rabbitmq/mergify/bp/v4.0.x/pr-13729

Fix concurrent AMQP queue declarations (backport #13727) (backport #13729)

Author: michaelklishin

deps/rabbit/src/rabbit_amqp_management.erl

@@ -127,15 +127,15 @@ handle_http_req(HttpMethod = <<"PUT">>,
     PermCache1 = check_resource_access(QName, configure, User, PermCache0),
     rabbit_core_metrics:queue_declared(QName),
 
-    {Q1, NumMsgs, NumConsumers, StatusCode, PermCache} =
     case rabbit_amqqueue:with(
            QName,
            fun(Q) ->
                    try rabbit_amqqueue:assert_equivalence(
                          Q, Durable, AutoDelete, QArgs, Owner) of
                        ok ->
                            {ok, Msgs, Consumers} = rabbit_amqqueue:stat(Q),
-                           {ok, {Q, Msgs, Consumers, <<"200">>, PermCache1}}
+                           RespPayload = encode_queue(Q, Msgs, Consumers),
+                           {ok, {<<"200">>, RespPayload, {PermCache1, TopicPermCache}}}
                    catch exit:#amqp_error{name = precondition_failed,
                                           explanation = Expl} ->
                              throw(<<"409">>, Expl, []);
@@ -146,23 +146,26 @@ handle_http_req(HttpMethod = <<"PUT">>,
         {ok, Result} ->
             Result;
         {error, not_found} ->
-            PermCache2 = check_dead_letter_exchange(QName, QArgs, User, PermCache1),
+            PermCache = check_dead_letter_exchange(QName, QArgs, User, PermCache1),
+            PermCaches = {PermCache, TopicPermCache},
             try rabbit_amqqueue:declare(
                   QName, Durable, AutoDelete, QArgs, Owner, Username) of
                 {new, Q} ->
                     rabbit_core_metrics:queue_created(QName),
-                    {Q, 0, 0, <<"201">>, PermCache2};
+                    RespPayload = encode_queue(Q, 0, 0),
+                    {<<"201">>, RespPayload, PermCaches};
                 {owner_died, Q} ->
                     %% Presumably our own days are numbered since the
                     %% connection has died. Pretend the queue exists though,
                     %% just so nothing fails.
-                    {Q, 0, 0, <<"201">>, PermCache2};
+                    RespPayload = encode_queue(Q, 0, 0),
+                    {<<"201">>, RespPayload, PermCaches};
                 {absent, Q, Reason} ->
                     absent(Q, Reason);
                 {existing, _Q} ->
                     %% Must have been created in the meantime. Loop around again.
                     handle_http_req(HttpMethod, PathSegments, Query, ReqPayload,
-                                    Vhost, User, ConnPid, {PermCache2, TopicPermCache});
+                                    Vhost, User, ConnPid, PermCaches);
                 {error, queue_limit_exceeded, Reason, ReasonArgs} ->
                     throw(<<"403">>,
                           Reason,
@@ -177,10 +180,7 @@ handle_http_req(HttpMethod = <<"PUT">>,
             end;
         {error, {absent, Q, Reason}} ->
             absent(Q, Reason)
-    end,
-
-    RespPayload = encode_queue(Q1, NumMsgs, NumConsumers),
-    {StatusCode, RespPayload, {PermCache, TopicPermCache}};
+    end;
 
 handle_http_req(<<"PUT">>,
                 [<<"exchanges">>, XNameBinQuoted],

deps/rabbitmq_amqp_client/test/management_SUITE.erl

@@ -52,6 +52,7 @@ groups() ->
        bad_exchange_property,
        bad_exchange_type,
        get_queue_not_found,
+       declare_queues_concurrently,
        declare_queue_default_queue_type,
        declare_queue_empty_name,
        declare_queue_line_feed,
@@ -436,6 +437,40 @@ get_queue_not_found(Config) ->
                  amqp10_msg:body(Resp)),
     ok = cleanup(Init).
 
+declare_queues_concurrently(Config) ->
+    NumQueues = 5,
+    {Pid1, Ref1} = spawn_monitor(?MODULE, declare_queues, [Config, NumQueues]),
+    {Pid2, Ref2} = spawn_monitor(?MODULE, declare_queues, [Config, NumQueues]),
+    receive {'DOWN', Ref1, process, Pid1, Reason1} ->
+                ?assertEqual(normal, Reason1)
+    end,
+    receive {'DOWN', Ref2, process, Pid2, Reason2} ->
+                ?assertEqual(normal, Reason2)
+    end,
+
+    ?assertEqual(NumQueues, count_queues(Config)),
+
+    Init = {_, LinkPair} = init(Config),
+    lists:foreach(fun(N) ->
+                          Bin = integer_to_binary(N),
+                          QName = <<"queue-", Bin/binary>>,
+                          {ok, _} = rabbitmq_amqp_client:delete_queue(LinkPair, QName)
+                  end, lists:seq(1, NumQueues)),
+    ok = cleanup(Init).
+
+declare_queues(Config, Num) ->
+    Init = {_, LinkPair} = init(Config),
+    ok = declare_queues0(LinkPair, Num),
+    ok = cleanup(Init).
+
+declare_queues0(_LinkPair, 0) ->
+    ok;
+declare_queues0(LinkPair, Left) ->
+    Bin = integer_to_binary(Left),
+    QName = <<"queue-", Bin/binary>>,
+    ?assertMatch({ok, _}, rabbitmq_amqp_client:declare_queue(LinkPair, QName, #{})),
+    declare_queues0(LinkPair, Left - 1).
+
 declare_queue_default_queue_type(Config) ->
     Node = get_node_config(Config, 0, nodename),
     Vhost = QName = atom_to_binary(?FUNCTION_NAME),
@@ -864,11 +899,11 @@ pipeline(Config) ->
     %% because RabbitMQ grants us 8 link credits initially.
     Num = 8,
     pipeline0(Num, LinkPair, <<"PUT">>, {map, []}),
-    eventually(?_assertEqual(Num, rpc(Config, rabbit_amqqueue, count, [])), 200, 20),
+    eventually(?_assertEqual(Num, count_queues(Config)), 200, 20),
     flush(queues_created),
 
     pipeline0(Num, LinkPair, <<"DELETE">>, null),
-    eventually(?_assertEqual(0, rpc(Config, rabbit_amqqueue, count, [])), 200, 20),
+    eventually(?_assertEqual(0, count_queues(Config)), 200, 20),
     flush(queues_deleted),
 
     ok = cleanup(Init).
@@ -1120,3 +1155,6 @@ gen_server_state(Pid) ->
     L1 = lists:last(L0),
     {data, L2} = lists:last(L1),
     proplists:get_value("State", L2).
+
+count_queues(Config) ->
+    rpc(Config, rabbit_amqqueue, count, []).