Auto widen session incoming-window in AMQP 1.0 client

This commit fixes a bug in the Erlang AMQP 1.0 client.

Prior to this commit, to repro this bug:
1. Send more than 2^16 messages to a queue.
2. Grant more than a total of 2^16 link credit initially (on a single link
   or across multiple links) on a single session without any
   auto or manual link credit renewal.

The expectation is that thanks to sufficiently granted initial link-credit,
the client will receive all messages.
However, consumption stops after exactly 2^16-1 messages.

That's because the client lib was never sending a flow frame to the server.
So, after the client received all 2^16-1 messages (the initial
incoming-window set by the client), the server's remote-incoming-window
reached 0 causing the server to stop delivering messages.

The expectation is that the client lib automatically handles session
flow control without any manual involvement of the client app.

This commit implements this fix:
* We keep the server's remote-incoming window always large by default as
  explained in https://www.rabbitmq.com/blog/2024/09/02/amqp-flow-control#incoming-window
* Hence, the client lib sets its incoming-window to 100,000 initially.
* The client lib tracks its incoming-window decrementing it by 1 for
  every transfer it received. (This wasn't done prior to this commit.)
* Whenever this window shrinks below 50,000, the client sends a flow
  frame without any link information widening its incoming-window back to 100,000.
* For test cases (maybe later for apps as well), there is a new function
  `amqp10_client_session:flow/3`, which allows for a test case to do manual
  session flow control. Its API is designed very similar to
  `amqp10_client_session:flow_link/4` in that the test can optionally request
  the lib to auto widen the session window whenever it falls below a certain threshold.

(cherry picked from commit 32854e8)

Author: ansd

deps/amqp10_client/src/amqp10_client.erl

@@ -339,7 +339,7 @@ flow_link_credit(#link_ref{role = receiver, session = Session,
        RenewWhenBelow =< Credit) ->
     Flow = #'v1_0.flow'{link_credit = {uint, Credit},
                         drain = Drain},
-    ok = amqp10_client_session:flow(Session, Handle, Flow, RenewWhenBelow).
+    ok = amqp10_client_session:flow_link(Session, Handle, Flow, RenewWhenBelow).
 
 %% @doc Stop a receiving link.
 %% See AMQP 1.0 spec §2.6.10.
@@ -348,7 +348,7 @@ stop_receiver_link(#link_ref{role = receiver,
                              link_handle = Handle}) ->
     Flow = #'v1_0.flow'{link_credit = {uint, 0},
                         echo = true},
-    ok = amqp10_client_session:flow(Session, Handle, Flow, never).
+    ok = amqp10_client_session:flow_link(Session, Handle, Flow, never).
 
 %%% messages
 

deps/amqp10_client/src/amqp10_client_session.erl

@@ -20,10 +20,13 @@
          attach/2,
          detach/2,
          transfer/3,
-         flow/4,
-         disposition/5
+         disposition/5,
+         flow_link/4
         ]).
 
+%% Manual session flow control is currently only used in tests.
+-export([flow/3]).
+
 %% Private API
 -export([start_link/4,
          socket_ready/2
@@ -51,7 +54,8 @@
         [add/2,
          diff/2]).
 
--define(MAX_SESSION_WINDOW_SIZE, 65535).
+%% By default, we want to keep the server's remote-incoming-window large at all times.
+-define(DEFAULT_MAX_INCOMING_WINDOW, 100_000).
 -define(UINT_OUTGOING_WINDOW, {uint, ?UINT_MAX}).
 -define(INITIAL_OUTGOING_DELIVERY_ID, ?UINT_MAX).
 %% "The next-outgoing-id MAY be initialized to an arbitrary value" [2.5.6]
@@ -129,7 +133,8 @@
          available = 0 :: non_neg_integer(),
          drain = false :: boolean(),
          partial_transfers :: undefined | {#'v1_0.transfer'{}, [binary()]},
-         auto_flow :: never | {auto, RenewWhenBelow :: pos_integer(), Credit :: pos_integer()},
+         auto_flow :: never | {RenewWhenBelow :: pos_integer(),
+                               Credit :: pos_integer()},
          incoming_unsettled = #{} :: #{delivery_number() => ok},
          footer_opt :: footer_opt() | undefined
         }).
@@ -140,7 +145,10 @@
 
          %% session flow control, see section 2.5.6
          next_incoming_id :: transfer_number() | undefined,
-         incoming_window = ?MAX_SESSION_WINDOW_SIZE :: non_neg_integer(),
+         %% Can become negative if the peer overshoots our window.
+         incoming_window :: integer(),
+         auto_flow :: never | {RenewWhenBelow :: pos_integer(),
+                               NewWindowSize :: pos_integer()},
          next_outgoing_id = ?INITIAL_OUTGOING_TRANSFER_ID :: transfer_number(),
          remote_incoming_window = 0 :: non_neg_integer(),
          remote_outgoing_window = 0 :: non_neg_integer(),
@@ -200,7 +208,17 @@ transfer(Session, Amqp10Msg, Timeout) ->
     [Transfer | Sections] = amqp10_msg:to_amqp_records(Amqp10Msg),
     gen_statem:call(Session, {transfer, Transfer, Sections}, Timeout).
 
-flow(Session, Handle, Flow, RenewWhenBelow) ->
+-spec flow(pid(), non_neg_integer(), never | pos_integer()) -> ok.
+flow(Session, IncomingWindow, RenewWhenBelow) when
+      %% Check that the RenewWhenBelow value make sense.
+      RenewWhenBelow =:= never orelse
+      is_integer(RenewWhenBelow) andalso
+      RenewWhenBelow > 0 andalso
+      RenewWhenBelow =< IncomingWindow ->
+    gen_statem:cast(Session, {flow_session, IncomingWindow, RenewWhenBelow}).
+
+-spec flow_link(pid(), link_handle(), #'v1_0.flow'{}, never | pos_integer()) -> ok.
+flow_link(Session, Handle, Flow, RenewWhenBelow) ->
     gen_statem:cast(Session, {flow_link, Handle, Flow, RenewWhenBelow}).
 
 %% Sending a disposition on a sender link (with receiver-settle-mode = second)
@@ -239,6 +257,9 @@ init([FromPid, Channel, Reader, ConnConfig]) ->
                    channel = Channel,
                    reader = Reader,
                    connection_config = ConnConfig,
+                   incoming_window = ?DEFAULT_MAX_INCOMING_WINDOW,
+                   auto_flow = {?DEFAULT_MAX_INCOMING_WINDOW div 2,
+                                ?DEFAULT_MAX_INCOMING_WINDOW},
                    early_attach_requests = []},
     {ok, unmapped, State}.
 
@@ -282,15 +303,15 @@ mapped(cast, 'end', State) ->
 mapped(cast, {flow_link, OutHandle, Flow0, RenewWhenBelow}, State0) ->
     State = send_flow_link(OutHandle, Flow0, RenewWhenBelow, State0),
     {keep_state, State};
-mapped(cast, {flow_session, Flow0 = #'v1_0.flow'{incoming_window = {uint, IncomingWindow}}},
-       #state{next_incoming_id = NII,
-              next_outgoing_id = NOI} = State) ->
-    Flow = Flow0#'v1_0.flow'{
-                   next_incoming_id = maybe_uint(NII),
-                   next_outgoing_id = uint(NOI),
-                   outgoing_window = ?UINT_OUTGOING_WINDOW},
-    ok = send(Flow, State),
-    {keep_state, State#state{incoming_window = IncomingWindow}};
+mapped(cast, {flow_session, IncomingWindow, RenewWhenBelow}, State0) ->
+    AutoFlow = case RenewWhenBelow of
+                   never -> never;
+                   _ -> {RenewWhenBelow, IncomingWindow}
+               end,
+    State = State0#state{incoming_window = IncomingWindow,
+                         auto_flow = AutoFlow},
+    send_flow_session(State),
+    {keep_state, State};
 mapped(cast, #'v1_0.end'{} = End, State) ->
     %% We receive the first end frame, reply and terminate.
     _ = send_end(State),
@@ -656,35 +677,44 @@ is_bare_message_section(_Section) ->
 
 send_flow_link(OutHandle,
                #'v1_0.flow'{link_credit = {uint, Credit}} = Flow0, RenewWhenBelow,
-               #state{links = Links,
-                      next_incoming_id = NII,
-                      next_outgoing_id = NOI,
-                      incoming_window = InWin} = State) ->
+               #state{links = Links} = State) ->
     AutoFlow = case RenewWhenBelow of
                    never -> never;
-                   Limit -> {auto, Limit, Credit}
+                   _ -> {RenewWhenBelow, Credit}
                end,
     #{OutHandle := #link{output_handle = H,
                          role = receiver,
                          delivery_count = DeliveryCount,
                          available = Available} = Link} = Links,
-    Flow = Flow0#'v1_0.flow'{
-                   handle = uint(H),
-                   %% "This value MUST be set if the peer has received the begin
-                   %% frame for the session, and MUST NOT be set if it has not." [2.7.4]
-                   next_incoming_id = maybe_uint(NII),
-                   next_outgoing_id = uint(NOI),
-                   outgoing_window = ?UINT_OUTGOING_WINDOW,
-                   incoming_window = uint(InWin),
-                   %% "In the event that the receiving link endpoint has not yet seen the
-                   %% initial attach frame from the sender this field MUST NOT be set." [2.7.4]
-                   delivery_count = maybe_uint(DeliveryCount),
-                   available = uint(Available)},
+    Flow1 = Flow0#'v1_0.flow'{
+                    handle = uint(H),
+                    %% "In the event that the receiving link endpoint has not yet seen the
+                    %% initial attach frame from the sender this field MUST NOT be set." [2.7.4]
+                    delivery_count = maybe_uint(DeliveryCount),
+                    available = uint(Available)},
+    Flow = set_flow_session_fields(Flow1, State),
     ok = send(Flow, State),
     State#state{links = Links#{OutHandle =>
                                Link#link{link_credit = Credit,
                                          auto_flow = AutoFlow}}}.
 
+send_flow_session(State) ->
+    Flow = set_flow_session_fields(#'v1_0.flow'{}, State),
+    ok = send(Flow, State).
+
+set_flow_session_fields(Flow, #state{next_incoming_id = NID,
+                                     incoming_window = IW,
+                                     next_outgoing_id = NOI}) ->
+    Flow#'v1_0.flow'{
+           %% "This value MUST be set if the peer has received the begin
+           %% frame for the session, and MUST NOT be set if it has not." [2.7.4]
+           next_incoming_id = maybe_uint(NID),
+           %% IncomingWindow0 can be negative when the sending server overshoots our window.
+           %% We must set a floor of 0 in the FLOW frame because field incoming-window is an uint.
+           incoming_window = uint(max(0, IW)),
+           next_outgoing_id = uint(NOI),
+           outgoing_window = ?UINT_OUTGOING_WINDOW}.
+
 build_frames(Channel, Trf, Bin, MaxPayloadSize, Acc)
   when byte_size(Bin) =< MaxPayloadSize ->
     T = amqp10_framing:encode_bin(Trf#'v1_0.transfer'{more = false}),
@@ -1059,17 +1089,21 @@ book_transfer_send(Num, #link{output_handle = Handle} = Link,
                 links = Links#{Handle => book_link_transfer_send(Link)}}.
 
 book_partial_transfer_received(#state{next_incoming_id = NID,
-                                      remote_outgoing_window = ROW} = State) ->
-    State#state{next_incoming_id = add(NID, 1),
-                remote_outgoing_window = ROW - 1}.
+                                      incoming_window = IW,
+                                      remote_outgoing_window = ROW} = State0) ->
+    State = State0#state{next_incoming_id = add(NID, 1),
+                         incoming_window = IW - 1,
+                         remote_outgoing_window = ROW - 1},
+    maybe_widen_incoming_window(State).
 
 book_transfer_received(State = #state{connection_config =
                                       #{transfer_limit_margin := Margin}},
                        #link{link_credit = Margin} = Link) ->
     {transfer_limit_exceeded, Link, State};
 book_transfer_received(#state{next_incoming_id = NID,
+                              incoming_window = IW,
                               remote_outgoing_window = ROW,
-                              links = Links} = State,
+                              links = Links} = State0,
                        #link{output_handle = OutHandle,
                              delivery_count = DC,
                              link_credit = LC,
@@ -1079,19 +1113,31 @@ book_transfer_received(#state{next_incoming_id = NID,
                       %% "the receiver MUST maintain a floor of zero in its
                       %% calculation of the value of available" [2.6.7]
                       available = max(0, Avail - 1)},
-    State1 = State#state{links = Links#{OutHandle => Link1},
-                         next_incoming_id = add(NID, 1),
-                         remote_outgoing_window = ROW - 1},
+    State1 = State0#state{links = Links#{OutHandle => Link1},
+                          next_incoming_id = add(NID, 1),
+                          incoming_window = IW - 1,
+                          remote_outgoing_window = ROW - 1},
+    State = maybe_widen_incoming_window(State1),
     case Link1 of
         #link{link_credit = 0,
               auto_flow = never} ->
-            {credit_exhausted, Link1, State1};
+            {credit_exhausted, Link1, State};
         _ ->
-            {ok, Link1, State1}
+            {ok, Link1, State}
     end.
 
+maybe_widen_incoming_window(
+  State0 = #state{incoming_window = IncomingWindow,
+                  auto_flow = {RenewWhenBelow, NewWindowSize}})
+  when IncomingWindow < RenewWhenBelow ->
+    State = State0#state{incoming_window = NewWindowSize},
+    send_flow_session(State),
+    State;
+maybe_widen_incoming_window(State) ->
+    State.
+
 auto_flow(#link{link_credit = LC,
-                auto_flow = {auto, RenewWhenBelow, Credit},
+                auto_flow = {RenewWhenBelow, Credit},
                 output_handle = OutHandle,
                 incoming_unsettled = Unsettled},
           State)
@@ -1230,6 +1276,7 @@ format_status(Status = #{data := Data0}) ->
            remote_channel = RemoteChannel,
            next_incoming_id = NextIncomingId,
            incoming_window = IncomingWindow,
+           auto_flow = SessionAutoFlow,
            next_outgoing_id = NextOutgoingId,
            remote_incoming_window = RemoteIncomingWindow,
            remote_outgoing_window = RemoteOutgoingWindow,
@@ -1294,6 +1341,7 @@ format_status(Status = #{data := Data0}) ->
              remote_channel => RemoteChannel,
              next_incoming_id => NextIncomingId,
              incoming_window => IncomingWindow,
+             auto_flow => SessionAutoFlow,
              next_outgoing_id => NextOutgoingId,
              remote_incoming_window => RemoteIncomingWindow,
              remote_outgoing_window => RemoteOutgoingWindow,