Pipeline management requests

ansd · ansd · commit e21007f6bc9b · 2024-03-23T14:06:46.000+01:00
diff --git a/deps/rabbit/src/rabbit_amqp_session.erl b/deps/rabbit/src/rabbit_amqp_session.erl
@@ -48,6 +48,7 @@
 %% An even better approach in future would be to dynamically grow (or shrink) the link credit
 %% we grant depending on how fast target queue(s) actually confirm messages.
 -define(LINK_CREDIT_RCV, 128).
+-define(MANAGEMENT_LINK_CREDIT_RCV, 8).
 -define(MANAGEMENT_NODE_ADDRESS, <<"/management">>).
 
 -export([start_link/8,
@@ -90,8 +91,6 @@
 -record(management_link, {
           name :: binary(),
           delivery_count :: sequence_no(),
-          %% Credit on an incoming management link is always 1 since management
-          %% requests are synchronous and we grant 1 credit when sending a response.
           credit :: non_neg_integer(),
           max_message_size :: unlimited | pos_integer()
          }).
@@ -767,11 +766,10 @@ handle_control(#'v1_0.attach'{
                                                    outgoing_half = unattached},
                              Pairs0)
             end,
-    Credit = 1,
     MaxMessageSize = persistent_term:get(max_message_size),
     Link = #management_link{name = LinkName,
                             delivery_count = DeliveryCountInt,
-                            credit = Credit,
+                            credit = ?MANAGEMENT_LINK_CREDIT_RCV,
                             max_message_size = MaxMessageSize},
     State = State0#state{management_link_pairs = Pairs,
                          incoming_management_links = maps:put(HandleInt, Link, Links)},
@@ -788,7 +786,7 @@ handle_control(#'v1_0.attach'{
                properties = Properties},
     Flow = #'v1_0.flow'{handle = Handle,
                         delivery_count = DeliveryCount,
-                        link_credit = ?UINT(Credit)},
+                        link_credit = ?UINT(?MANAGEMENT_LINK_CREDIT_RCV)},
     reply0([Reply, Flow], State);
 
 handle_control(#'v1_0.attach'{
@@ -1649,22 +1647,33 @@ incoming_mgmt_link_transfer(
                         {ok, Link} ->
                             Link;
                         error ->
-                            protocol_error(?V_1_0_AMQP_ERROR_ILLEGAL_STATE,
-                                           "Unknown link handle: ~p", [IncomingHandleInt])
+                            protocol_error(
+                              ?V_1_0_AMQP_ERROR_ILLEGAL_STATE,
+                              "Unknown link handle: ~p", [IncomingHandleInt])
                     end,
     #management_link{name = Name,
                      delivery_count = IncomingDeliveryCount0,
-                     credit = 1,
+                     credit = IncomingCredit0,
                      max_message_size = IncomingMaxMessageSize
                     } = IncomingLink0,
+    case IncomingCredit0 > 0 of
+        true ->
+            ok;
+        false ->
+            protocol_error(
+              ?V_1_0_LINK_ERROR_TRANSFER_LIMIT_EXCEEDED,
+              "insufficient credit (~b) for management link from client to RabbitMQ",
+              [IncomingCredit0])
+    end,
     #management_link_pair{
        incoming_half = IncomingHandleInt,
        outgoing_half = OutgoingHandleInt
       } = maps:get(Name, LinkPairs),
     OutgoingLink0 = case OutgoingHandleInt of
                         unattached ->
-                            protocol_error(?V_1_0_AMQP_ERROR_PRECONDITION_FAILED,
-                                           "received transfer on half open management link pair", []);
+                            protocol_error(
+                              ?V_1_0_AMQP_ERROR_PRECONDITION_FAILED,
+                              "received transfer on half open management link pair", []);
                         _ ->
                             maps:get(OutgoingHandleInt, OutgoingLinks)
                     end,
@@ -1676,9 +1685,10 @@ incoming_mgmt_link_transfer(
         true ->
             ok;
         false ->
-            protocol_error(?V_1_0_AMQP_ERROR_PRECONDITION_FAILED,
-                           "insufficient credit (~b) for management link from server to client",
-                           [OutgoingCredit])
+            protocol_error(
+              ?V_1_0_AMQP_ERROR_PRECONDITION_FAILED,
+              "insufficient credit (~b) for management link from RabbitMQ to client",
+              [OutgoingCredit])
     end,
     Settled = default(MaybeSettled, false),
     validate_transfer_rcv_settle_mode(RcvSettleMode, Settled),
@@ -1698,17 +1708,19 @@ incoming_mgmt_link_transfer(
     Frames = transfer_frames(Transfer, Response, MaxFrameSize),
     PendingTransfer = #pending_management_transfer{frames = Frames},
     IncomingDeliveryCount = add(IncomingDeliveryCount0, 1),
-    IncomingLink = IncomingLink0#management_link{delivery_count = IncomingDeliveryCount},
+    IncomingCredit1 = IncomingCredit0 - 1,
+    {IncomingCredit, Reply} = maybe_grant_mgmt_link_credit(
+                                IncomingCredit1, IncomingDeliveryCount, IncomingHandle),
+    IncomingLink = IncomingLink0#management_link{delivery_count = IncomingDeliveryCount,
+                                                 credit = IncomingCredit},
     OutgoingLink = OutgoingLink0#management_link{delivery_count = add(OutgoingDeliveryCount, 1),
                                                  credit = OutgoingCredit - 1},
     State = State0#state{
               outgoing_delivery_id = add(OutgoingDeliveryId, 1),
               outgoing_pending = queue:in(PendingTransfer, Pending),
               incoming_management_links = maps:update(IncomingHandleInt, IncomingLink, IncomingLinks),
               outgoing_management_links = maps:update(OutgoingHandleInt, OutgoingLink, OutgoingLinks)},
-
-    Flow = flow(IncomingHandle, IncomingDeliveryCount, 1),
-    {[Flow], State}.
+    {Reply, State}.
 
 incoming_link_transfer(
   #'v1_0.transfer'{more = true,
@@ -1908,6 +1920,13 @@ grant_link_credit(Credit, NumUnconfirmed) ->
     Credit =< ?LINK_CREDIT_RCV / 2 andalso
     NumUnconfirmed < ?LINK_CREDIT_RCV.
 
+maybe_grant_mgmt_link_credit(Credit, DeliveryCount, Handle)
+  when Credit =< ?MANAGEMENT_LINK_CREDIT_RCV / 2 ->
+    {?MANAGEMENT_LINK_CREDIT_RCV,
+     [flow(Handle, DeliveryCount, ?MANAGEMENT_LINK_CREDIT_RCV)]};
+maybe_grant_mgmt_link_credit(Credit, _, _) ->
+    {Credit, []}.
+
 %% TODO default-outcome and outcomes, dynamic lifetimes
 ensure_target(#'v1_0.target'{dynamic = true}, _, _) ->
     protocol_error(?V_1_0_AMQP_ERROR_NOT_IMPLEMENTED,
@@ -2660,6 +2679,9 @@ format_status(
                     outgoing_delivery_id = OutgoingDeliveryId,
                     incoming_links = IncomingLinks,
                     outgoing_links = OutgoingLinks,
+                    management_link_pairs = ManagementLinks,
+                    incoming_management_links = IncomingManagementLinks,
+                    outgoing_management_links = OutgoingManagementLinks,
                     outgoing_unsettled_map = OutgoingUnsettledMap,
                     stashed_rejected = StashedRejected,
                     stashed_settled = StashedSettled,
@@ -2676,6 +2698,9 @@ format_status(
               outgoing_delivery_id => OutgoingDeliveryId,
               incoming_links => IncomingLinks,
               outgoing_links => OutgoingLinks,
+              management_link_pairs => ManagementLinks,
+              incoming_management_links => IncomingManagementLinks,
+              outgoing_management_links => OutgoingManagementLinks,
               outgoing_unsettled_map => OutgoingUnsettledMap,
               stashed_rejected => StashedRejected,
               stashed_settled => StashedSettled,
diff --git a/deps/rabbitmq_amqp_client/test/management_SUITE.erl b/deps/rabbitmq_amqp_client/test/management_SUITE.erl
@@ -72,7 +72,10 @@ groups() ->
        bind_destination_line_feed,
        bind_missing_queue,
        exclusive_queue,
-       purge_stream
+       purge_stream,
+       pipeline,
+       multiple_link_pairs,
+       link_attach_order
       ]},
      {cluster_size_3, [shuffle],
       [classic_queue_stopped,
@@ -786,6 +789,135 @@ queue_topology(Config) ->
     {ok, _} = rabbitmq_amqp_client:delete_queue(LinkPair2, SQName),
     ok = cleanup(Init2).
 
+%% Even though RabbitMQ processes management requests synchronously (one at a time),
+%% the client should be able to send multiple requests at once before receiving a response.
+pipeline(Config) ->
+    Init = {_, _, LinkPair} = init(Config),
+    flush(attached),
+
+    %% We should be able to send 8 management requests at once
+    %% 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),
+    flush(queues_created),
+
+    pipeline0(Num, LinkPair, <<"DELETE">>, null),
+    eventually(?_assertEqual(0, rpc(Config, rabbit_amqqueue, count, [])), 200, 20),
+    flush(queues_deleted),
+
+    ok = cleanup(Init).
+
+pipeline0(Num,
+          #link_pair{outgoing_link = OutgoingLink,
+                     incoming_link = IncomingLink},
+          HttpMethod,
+          Body) ->
+    ok = amqp10_client:flow_link_credit(IncomingLink, Num, never),
+    [begin
+         Request0 = amqp10_msg:new(<<>>, #'v1_0.amqp_value'{content = Body}, true),
+         Bin = integer_to_binary(N),
+         Props = #{subject => HttpMethod,
+                   to => <<"/queues/q-", Bin/binary>>,
+                   message_id => {binary, Bin},
+                   reply_to => <<"$me">>},
+         Request =  amqp10_msg:set_properties(Props, Request0),
+         ok = amqp10_client:send_msg(OutgoingLink, Request)
+     end || N <- lists:seq(1, Num)].
+
+%% RabbitMQ allows attaching multiple link pairs.
+multiple_link_pairs(Config) ->
+    OpnConf = connection_config(Config),
+    {ok, Connection} = amqp10_client:open_connection(OpnConf),
+    {ok, Session} = amqp10_client:begin_session_sync(Connection),
+    {ok, LinkPair1} = rabbitmq_amqp_client:attach_management_link_pair_sync(Session, <<"link pair 1">>),
+    {ok, LinkPair2} = rabbitmq_amqp_client:attach_management_link_pair_sync(Session, <<"link pair 2">>),
+
+    [SessionPid] = rpc(Config, rabbit_amqp_session, list_local, []),
+    #{management_link_pairs := Pairs0,
+      incoming_management_links := Incoming0,
+      outgoing_management_links := Outgoing0} = gen_server_state(SessionPid),
+    ?assertEqual(2, maps:size(Pairs0)),
+    ?assertEqual(2, maps:size(Incoming0)),
+    ?assertEqual(2, maps:size(Outgoing0)),
+
+    QName = <<"q">>,
+    {ok, #{}} = rabbitmq_amqp_client:declare_queue(LinkPair1, QName, #{}),
+    {ok, #{}} = rabbitmq_amqp_client:delete_queue(LinkPair2, QName),
+
+    ok = rabbitmq_amqp_client:detach_management_link_pair_sync(LinkPair1),
+    ok = rabbitmq_amqp_client:detach_management_link_pair_sync(LinkPair2),
+
+    %% Assert that the server cleaned up its state.
+    #{management_link_pairs := Pairs,
+      incoming_management_links := Incoming,
+      outgoing_management_links := Outgoing} = gen_server_state(SessionPid),
+    ?assertEqual(0, maps:size(Pairs)),
+    ?assertEqual(0, maps:size(Incoming)),
+    ?assertEqual(0, maps:size(Outgoing)),
+
+    ok = amqp10_client:end_session(Session),
+    ok = amqp10_client:close_connection(Connection).
+
+%% Attaching (and detaching) either the sender or the receiver link first should both work.
+link_attach_order(Config) ->
+    PairName1 = <<"link pair 1">>,
+    PairName2 = <<"link pair 2">>,
+
+    OpnConf = connection_config(Config),
+    {ok, Connection} = amqp10_client:open_connection(OpnConf),
+    {ok, Session} = amqp10_client:begin_session_sync(Connection),
+
+    Terminus = #{address => <<"/management">>,
+                 durable => none},
+    OutgoingAttachArgs1 = #{name => PairName1,
+                            role => {sender, Terminus},
+                            snd_settle_mode => settled,
+                            rcv_settle_mode => first,
+                            properties => #{<<"paired">> => true}},
+    IncomingAttachArgs1 = OutgoingAttachArgs1#{role := {receiver, Terminus, self()},
+                                               filter => #{}},
+    OutgoingAttachArgs2 = OutgoingAttachArgs1#{name := PairName2},
+    IncomingAttachArgs2 = IncomingAttachArgs1#{name := PairName2},
+
+    %% Attach sender before receiver.
+    {ok, OutgoingRef1} = amqp10_client:attach_link(Session, OutgoingAttachArgs1),
+    {ok, IncomingRef1} = amqp10_client:attach_link(Session, IncomingAttachArgs1),
+    %% Attach receiver before sender.
+    {ok, IncomingRef2} = amqp10_client:attach_link(Session, IncomingAttachArgs2),
+    {ok, OutgoingRef2} = amqp10_client:attach_link(Session, OutgoingAttachArgs2),
+
+    Refs = [OutgoingRef1,
+            OutgoingRef2,
+            IncomingRef1,
+            IncomingRef2],
+
+    [ok = wait_for_event(Ref, attached) || Ref <- Refs],
+    flush(attached),
+
+    LinkPair1 = #link_pair{session = Session,
+                           outgoing_link = OutgoingRef1,
+                           incoming_link = IncomingRef1},
+    LinkPair2 = #link_pair{session = Session,
+                           outgoing_link = OutgoingRef2,
+                           incoming_link = IncomingRef2},
+
+    QName = <<"test queue">>,
+    {ok, #{}} = rabbitmq_amqp_client:declare_queue(LinkPair1, QName, #{}),
+    {ok, #{}} = rabbitmq_amqp_client:delete_queue(LinkPair2, QName),
+
+    %% Detach sender before receiver.
+    ok = amqp10_client:detach_link(OutgoingRef1),
+    ok = amqp10_client:detach_link(IncomingRef1),
+    %% Detach receiver before sender.
+    ok = amqp10_client:detach_link(IncomingRef2),
+    ok = amqp10_client:detach_link(OutgoingRef2),
+
+    [ok = wait_for_event(Ref, {detached, normal}) || Ref <- Refs],
+    flush(detached),
+    ok = amqp10_client:end_session(Session),
+    ok = amqp10_client:close_connection(Connection).
+
 init(Config) ->
     init(Config, 0).
 
@@ -846,3 +978,16 @@ wait_for_settlement(Tag, State) ->
               flush(Reason),
               ct:fail(Reason)
     end.
+
+wait_for_event(Ref, Event) ->
+    receive {amqp10_event, {link, Ref, Event}} -> ok
+    after 5000 -> ct:fail({missing_event, Ref, Event})
+    end.
+
+%% Return the formatted state of a gen_server via sys:get_status/1.
+%% (sys:get_state/1 is unformatted)
+gen_server_state(Pid) ->
+    {status, _, _, L0} = sys:get_status(Pid, 20_000),
+    L1 = lists:last(L0),
+    {data, L2} = lists:last(L1),
+    proplists:get_value("State", L2).
