Fix failing mixed version test

ansd · ansd · commit 65c8456f2c27 · 2024-04-03T17:51:53.000+02:00
Prior to this commit, mixed version test classic_queue_on_old_node of amqp_client_SUITE was failing. Commit 02c29ac must make sure that the new (4.0) AMQP 1.0 classic queue client continues to convey RabbitMQ internal credit flow control information back to the old (3.13.1) classic queue server. Otherwise, the old classic queue server will stop sending more messages to the new classic queue client after exactly 200 messages, which caused this mixed version test to fail.
diff --git a/deps/rabbit/src/rabbit_amqp_session.erl b/deps/rabbit/src/rabbit_amqp_session.erl
@@ -133,6 +133,9 @@
 -record(pending_transfer, {
           frames :: iolist(),
           queue_ack_required :: boolean(),
+          %% Queue that sent us this message.
+          %% When feature flag credit_api_v2 becomes required, this field should be deleted.
+          queue_pid :: pid() | credit_api_v2,
           delivery_id :: delivery_number(),
           outgoing_unsettled :: #outgoing_unsettled{}
          }).
@@ -1253,10 +1256,31 @@ send_pending(#state{remote_incoming_window = Space,
             Flow = session_flow_fields(Flow0, State0),
             rabbit_amqp_writer:send_command(WriterPid, Ch, Flow),
             send_pending(State0#state{outgoing_pending = Buf});
-        {{value, #pending_transfer{frames = Frames} = Pending}, Buf1}
+        {{value, #pending_transfer{
+                    frames = Frames,
+                    queue_pid = QPid,
+                    outgoing_unsettled = #outgoing_unsettled{queue_name = QName}
+                   } = Pending}, Buf1}
           when Space > 0 ->
+            SendFun = case QPid of
+                          credit_api_v2 ->
+                              send_fun(WriterPid, Ch);
+                          _ ->
+                              case rabbit_queue_type:module(QName, State0#state.queue_states) of
+                                  {ok, rabbit_classic_queue} ->
+                                      %% Classic queue client and classic queue process that
+                                      %% communicate via credit API v1 use RabbitMQ internal
+                                      %% credit flow control.
+                                      fun(Transfer, Sections) ->
+                                              rabbit_amqp_writer:send_command_and_notify(
+                                                WriterPid, Ch, QPid, self(), Transfer, Sections)
+                                      end;
+                                  {ok, _QType} ->
+                                      send_fun(WriterPid, Ch)
+                              end
+                      end,
             {NumTransfersSent, Buf, State1} =
-            case send_frames(WriterPid, Ch, Frames, Space) of
+            case send_frames(SendFun, Frames, Space) of
                 {sent_all, SpaceLeft} ->
                     {Space - SpaceLeft,
                      Buf1,
@@ -1271,8 +1295,9 @@ send_pending(#state{remote_incoming_window = Space,
             send_pending(State);
         {{value, Pending = #pending_management_transfer{frames = Frames}}, Buf1}
           when Space > 0 ->
+            SendFun = send_fun(WriterPid, Ch),
             {NumTransfersSent, Buf} =
-            case send_frames(WriterPid, Ch, Frames, Space) of
+            case send_frames(SendFun, Frames, Space) of
                 {sent_all, SpaceLeft} ->
                     {Space - SpaceLeft, Buf1};
                 {sent_some, Rest} ->
@@ -1285,13 +1310,18 @@ send_pending(#state{remote_incoming_window = Space,
             State0
     end.
 
-send_frames(_, _, [], SpaceLeft) ->
+send_frames(_, [], SpaceLeft) ->
     {sent_all, SpaceLeft};
-send_frames(_, _, Rest, 0) ->
+send_frames(_, Rest, 0) ->
     {sent_some, Rest};
-send_frames(Writer, Ch, [[Transfer, Sections] | Rest], SpaceLeft) ->
-    rabbit_amqp_writer:send_command(Writer, Ch, Transfer, Sections),
-    send_frames(Writer, Ch, Rest, SpaceLeft - 1).
+send_frames(SendFun, [[Transfer, Sections] | Rest], SpaceLeft) ->
+    SendFun(Transfer, Sections),
+    send_frames(SendFun, Rest, SpaceLeft - 1).
+
+send_fun(WriterPid, Ch) ->
+    fun(Transfer, Sections) ->
+            rabbit_amqp_writer:send_command(WriterPid, Ch, Transfer, Sections)
+    end.
 
 record_outgoing_unsettled(#pending_transfer{queue_ack_required = true,
                                             delivery_id = DeliveryId,
@@ -1532,7 +1562,7 @@ handle_queue_actions(Actions, State) ->
       end, State, Actions).
 
 handle_deliver(ConsumerTag, AckRequired,
-               Msg = {QName, _QPid, MsgId, Redelivered, Mc0},
+               Msg = {QName, QPid0, MsgId, Redelivered, Mc0},
                State = #state{outgoing_pending = Pending,
                               outgoing_delivery_id = DeliveryId,
                               outgoing_links = OutgoingLinks0,
@@ -1565,20 +1595,23 @@ handle_deliver(ConsumerTag, AckRequired,
             Frames = transfer_frames(Transfer, Sections, MaxFrameSize),
             messages_delivered(Redelivered, QType),
             rabbit_trace:tap_out(Msg, ConnName, ChannelNum, Username, Trace),
-            OutgoingLinks = case DelCount of
-                                credit_api_v2 ->
-                                    OutgoingLinks0;
-                                {credit_api_v1, C} ->
-                                    Link = Link0#outgoing_link{delivery_count = {credit_api_v1, add(C, 1)}},
-                                    maps:update(Handle, Link, OutgoingLinks0)
-                            end,
+            {OutgoingLinks, QPid
+            } = case DelCount of
+                    credit_api_v2 ->
+                        {OutgoingLinks0, credit_api_v2};
+                    {credit_api_v1, C} ->
+                        Link = Link0#outgoing_link{delivery_count = {credit_api_v1, add(C, 1)}},
+                        OutgoingLinks1 = maps:update(Handle, Link, OutgoingLinks0),
+                        {OutgoingLinks1, QPid0}
+                end,
             Del = #outgoing_unsettled{
                      msg_id = MsgId,
                      consumer_tag = ConsumerTag,
                      queue_name = QName},
             PendingTransfer = #pending_transfer{
                                  frames = Frames,
                                  queue_ack_required = AckRequired,
+                                 queue_pid = QPid,
                                  delivery_id = DeliveryId,
                                  outgoing_unsettled = Del},
             State#state{outgoing_pending = queue:in(PendingTransfer, Pending),
diff --git a/deps/rabbit/src/rabbit_amqp_writer.erl b/deps/rabbit/src/rabbit_amqp_writer.erl
@@ -15,6 +15,7 @@
          send_command/3,
          send_command/4,
          send_command_sync/3,
+         send_command_and_notify/6,
          internal_send_command/3]).
 
 %% gen_server callbacks
@@ -70,6 +71,17 @@ send_command_sync(Writer, ChannelNum, MethodRecord) ->
     Request = {send_command, ChannelNum, MethodRecord},
     gen_server:call(Writer, Request, ?CALL_TIMEOUT).
 
+%% Delete this function when feature flag credit_api_v2 becomes required.
+-spec send_command_and_notify(pid(),
+                              rabbit_types:channel_number(),
+                              pid(),
+                              pid(),
+                              rabbit_framing:amqp_method_record(),
+                              rabbit_types:content()) -> ok.
+send_command_and_notify(Writer, ChannelNum, QueuePid, SessionPid, MethodRecord, Content) ->
+    Request = {send_command_and_notify, ChannelNum, QueuePid, SessionPid, MethodRecord, Content},
+    gen_server:cast(Writer, Request).
+
 -spec internal_send_command(rabbit_net:socket(),
                             rabbit_framing:amqp_method_record(),
                             amqp10_framing | rabbit_amqp_sasl) -> ok.
@@ -95,6 +107,11 @@ handle_cast({send_command, ChannelNum, MethodRecord}, State0) ->
     no_reply(State);
 handle_cast({send_command, ChannelNum, MethodRecord, Content}, State0) ->
     State = internal_send_command_async(ChannelNum, MethodRecord, Content, State0),
+    no_reply(State);
+%% Delete below function clause when feature flag credit_api_v2 becomes required.
+handle_cast({send_command_and_notify, ChannelNum, QueuePid, SessionPid, MethodRecord, Content}, State0) ->
+    State = internal_send_command_async(ChannelNum, MethodRecord, Content, State0),
+    rabbit_amqqueue:notify_sent(QueuePid, SessionPid),
     no_reply(State).
 
 handle_call({send_command, ChannelNum, MethodRecord}, _From, State0) ->
@@ -104,7 +121,11 @@ handle_call({send_command, ChannelNum, MethodRecord}, _From, State0) ->
 
 handle_info(timeout, State0) ->
     State = flush(State0),
-    {noreply, State}.
+    {noreply, State};
+%% Delete below function clause when feature flag credit_api_v2 becomes required.
+handle_info({'DOWN', _MRef, process, QueuePid, _Reason}, State) ->
+    rabbit_amqqueue:notify_sent_queue_down(QueuePid),
+    no_reply(State).
 
 format_status(Status) ->
     maps:update_with(
diff --git a/deps/rabbit/src/rabbit_queue_consumers.erl b/deps/rabbit/src/rabbit_queue_consumers.erl
@@ -48,7 +48,7 @@
              %% The limiter itself
              limiter,
              %% Internal flow control for queue -> writer
-             unsent_message_count,
+             unsent_message_count :: non_neg_integer(),
              link_states :: #{rabbit_types:ctag() => #link_state{}}
             }).
 
@@ -271,11 +271,11 @@ deliver(FetchFun, QName, ConsumersChanged,
             end
     end.
 
-deliver_to_consumer(FetchFun, E = {ChPid, Consumer}, QName) ->
-    C = lookup_ch(ChPid),
+deliver_to_consumer(FetchFun,
+                    E = {ChPid, Consumer = #consumer{tag = CTag}},
+                    QName) ->
+    C = #cr{link_states = LinkStates} = lookup_ch(ChPid),
     ChBlocked = is_ch_blocked(C),
-    LinkStates = C#cr.link_states,
-    CTag = Consumer#consumer.tag,
     case LinkStates of
         #{CTag := #link_state{delivery_count = DeliveryCount0,
                               credit = Credit} = LinkState0} ->
