Repository: rabbitmq/rabbitmq-server (language: javascript)

Issue:
AMQP-1.0 session, `{badarg, [{erlang, iolist_size}]` at high consumption rates
### Describe the bug

I've seen this in two situations so far:
* on `main`, with `max_link_credit` set to a higher value (1700 - for no particular reason, other than to see the impact)
* on `ra-2.16.0` branch without tweaking `max_link_credit`. However, the main benefit of ra-2.16 is a higher consumption rate with many consumers and that's exactly when this issue happened.

This is from a main+high max_link_credit env:
```
reason: {{badarg,
             [{erlang,iolist_size,
                  [[<<112,0,35,30,126>>,
                    [82,-325],
                    <<"pÿÿÿü">>,<<"pÿÿÿÿ">>,67,
                    <<112,0,35,30,126>>,
                    <<112,0,0,6,164>>,
                    64,64,64,64]],
                  [{error_info,#{module => erl_erts_errors}}]},
              {amqp10_binary_generator,generate1,1,
                  [{file,"amqp10_binary_generator.erl"},{line,141}]},
              {amqp10_binary_generator,generate1,1,
                  [{file,"amqp10_binary_generator.erl"},{line,88}]},
              {amqp10_binary_generator,generate,1,
                  [{file,"amqp10_binary_generator.erl"},{line,79}]},
              {rabbit_amqp_writer,assemble_frame,3,
                  [{file,"rabbit_amqp_writer.erl"},{line,206}]},
              {rabbit_amqp_writer,internal_send_command_async,3,
                  [{file,"rabbit_amqp_writer.erl"},{line,189}]},
              {rabbit_amqp_writer,handle_cast,2,
                  [{file,"rabbit_amqp_writer.erl"},{line,110}]},
              {gen_server,try_handle_cast,3,
                  [{file,"gen_server.erl"},{line,1121}]}]},
         {gen_server,call,
             [<0.1759.0>,
              {send_command,0,
                  {'v1_0.end',
                      {'v1_0.error',
                          {symbol,<<"amqp:session:window-violation">>},
                          {utf8,
                              <<"incoming window violation (tolerated excess tranfers: 400)">>},
                          undefined}}},
              300000]}}
```

and this is from a ra-2.16 with the default 170 credits:
```
crasher:
  initial call: rabbit_amqp_writer:init/1
  pid: <0.19353.0>
  registered_name: []
  exception error: bad argument
    in function  iolist_size/1
       called as iolist_size([<<112,0,0,23,120>>,
                              [82,-15],
                              <<"pÿÿÿü">>,<<"pÿÿÿÿ">>,67,
                              <<112,0,0,23,120>>,
                              "Rª",64,64,64,64])
       *** argument 1: not an iodata term
    in call from amqp10_binary_generator:generate1/1 (amqp10_binary_generator.erl, line 141)
    in call from amqp10_binary_generator:generate1/1 (amqp10_binary_generator.erl, line 88)
    in call from amqp10_binary_generator:generate/1 (amqp10_binary_generator.erl, line 79)
    in call from rabbit_amqp_writer:assemble_frame/3 (rabbit_amqp_writer.erl, line 206)
    in call from rabbit_amqp_writer:internal_send_command_async/3 (rabbit_amqp_writer.erl, line 189)
    in call from rabbit_amqp_writer:handle_cast/2 (rabbit_amqp_writer.erl, line 110)
    in call from gen_server:try_handle_cast/3 (gen_server.erl, line 1121)
```


Discussion / hints:
@mkuratczyk what are the steps to repro this issue? How did you invoke `omq` CLI to see this error?
I have no deterministic repro steps. This is the workload that was running when this happened:
```omq amqp -x 50 --amqp-send-settled -r 1000 -y 10 --publish-to /queues/amqp10_consumers_join_late --consume-from /queues/amqp10_consumers_join_late --consumer-credits 1000 --queues quorum --consumer-startup-delay 150s -s 5000```

The issue occurred 40 seconds after the consumers "joined" (started).

NOTE: with the latest `omq` versions, replace `--amqp-send-settled` with `-c 50`

Fix patch:
diff --git a/deps/amqp10_client/src/amqp10_client_session.erl b/deps/amqp10_client/src/amqp10_client_session.erl
index ba7218b84d7e..5738abace5ff 100644
--- a/deps/amqp10_client/src/amqp10_client_session.erl
+++ b/deps/amqp10_client/src/amqp10_client_session.erl
@@ -467,11 +467,6 @@ mapped({call, From},
        #state{remote_incoming_window = Window})
   when Window =< 0 ->
     {keep_state_and_data, {reply, From, {error, remote_incoming_window_exceeded}}};
-mapped({call, From},
-       {transfer, _Transfer, _Sections},
-       #state{remote_incoming_window = Window})
-  when Window =< 0 ->
-    {keep_state_and_data, {reply, From, {error, remote_incoming_window_exceeded}}};
 mapped({call, From = {Pid, _}},
        {transfer, #'v1_0.transfer'{handle = {uint, OutHandle},
                                    delivery_tag = {binary, DeliveryTag},
diff --git a/deps/rabbit/src/rabbit_amqp_session.erl b/deps/rabbit/src/rabbit_amqp_session.erl
index 5a15222c0c76..44f492e94456 100644
--- a/deps/rabbit/src/rabbit_amqp_session.erl
+++ b/deps/rabbit/src/rabbit_amqp_session.erl
@@ -2003,7 +2003,10 @@ session_flow_fields(Frames, State)
 session_flow_fields(Flow = #'v1_0.flow'{},
                     #state{next_outgoing_id = NextOutgoingId,
                            next_incoming_id = NextIncomingId,
-                           incoming_window = IncomingWindow}) ->
+                           incoming_window = IncomingWindow0}) ->
+    %% IncomingWindow0 can be negative when the sending client overshoots our window.
+    %% However, we must set a floor of 0 in the FLOW frame because field incoming-window is an uint.
+    IncomingWindow = max(0, IncomingWindow0),
     Flow#'v1_0.flow'{
            next_outgoing_id = ?UINT(NextOutgoingId),
            outgoing_window = ?UINT_OUTGOING_WINDOW,


Test patch:
diff --git a/deps/rabbit/test/amqp_client_SUITE.erl b/deps/rabbit/test/amqp_client_SUITE.erl
index 97e8e6b079cf..ab4addfd6966 100644
--- a/deps/rabbit/test/amqp_client_SUITE.erl
+++ b/deps/rabbit/test/amqp_client_SUITE.erl
@@ -108,6 +108,7 @@ groups() ->
        detach_requeues_drop_head_classic_queue,
        resource_alarm_before_session_begin,
        resource_alarm_after_session_begin,
+       resource_alarm_send_many,
        max_message_size_client_to_server,
        max_message_size_server_to_client,
        global_counters,
@@ -3207,6 +3208,42 @@ resource_alarm_after_session_begin(Config) ->
     #'queue.delete_ok'{} = amqp_channel:call(Ch, #'queue.delete'{queue = QName}),
     ok = rabbit_ct_client_helpers:close_channel(Ch).
 
+%% Test case for
+%% https://github.com/rabbitmq/rabbitmq-server/issues/12816
+resource_alarm_send_many(Config) ->
+    QName = atom_to_binary(?FUNCTION_NAME),
+    Ch = rabbit_ct_client_helpers:open_channel(Config),
+    #'queue.declare_ok'{} = amqp_channel:call(Ch, #'queue.declare'{queue = QName}),
+    Address = rabbitmq_amqp_address:queue(QName),
+    OpnConf = connection_config(Config),
+    {ok, Connection} = amqp10_client:open_connection(OpnConf),
+    {ok, Session} = amqp10_client:begin_session_sync(Connection),
+
+    %% Send many messages while a memory alarm kicks in.
+    %% Our expectations are:
+    %% 1. At some point, our client's remote-incoming-window should be exceeded because
+    %%    RabbitMQ sets its incoming-window to 0 when the alarm kicks in.
+    %% 2. No crash.
+    {Pid, Ref} = spawn_monitor(?MODULE,
+                               send_until_remote_incoming_window_exceeded,
+                               [Session, Address]),
+    DefaultWatermark = rpc(Config, vm_memory_monitor, get_vm_memory_high_watermark, []),
+    ok = rpc(Config, vm_memory_monitor, set_vm_memory_high_watermark, [0]),
+    receive {'DOWN', Ref, process, Pid, Reason} ->
+                ?assertEqual(normal, Reason)
+    after 30_000 ->
+              ct:fail(send_timeout)
+    end,
+
+    %% Clear memory alarm.
+    ok = rpc(Config, vm_memory_monitor, set_vm_memory_high_watermark, [DefaultWatermark]),
+    timer:sleep(100),
+
+    ok = end_session_sync(Session),
+    ok = amqp10_client:close_connection(Connection),
+    #'queue.delete_ok'{} = amqp_channel:call(Ch, #'queue.delete'{queue = QName}),
+    ok = rabbit_ct_client_helpers:close_channel(Ch).
+
 auth_attempt_metrics(Config) ->
     open_and_close_connection(Config),
     [Attempt1] = rpc(Config, rabbit_core_metrics, get_auth_attempts, []),
@@ -6286,6 +6323,28 @@ count_received_messages0(Receiver, Count) ->
               Count
     end.
 
+send_until_remote_incoming_window_exceeded(Session, Address) ->
+    {ok, Sender} = amqp10_client:attach_sender_link(Session, <<"sender">>, Address, settled),
+    ok = wait_for_credit(Sender),
+    ok = send_until_remote_incoming_window_exceeded0(Sender, 100_000),
+    ok = amqp10_client:detach_link(Sender).
+
+send_until_remote_incoming_window_exceeded0(_Sender, 0) ->
+    ct:fail(remote_incoming_window_never_exceeded);
+send_until_remote_incoming_window_exceeded0(Sender, Left) ->
+    Bin = integer_to_binary(Left),
+    Msg = amqp10_msg:new(Bin, Bin, true),
+    case amqp10_client:send_msg(Sender, Msg) of
+        ok ->
+            send_until_remote_incoming_window_exceeded0(Sender, Left - 1);
+        {error, insufficient_credit} ->
+            ok = wait_for_credit(Sender),
+            send_until_remote_incoming_window_exceeded0(Sender, Left);
+        {error, remote_incoming_window_exceeded = Reason} ->
+            ct:pal("~s: ~b messages left", [Reason, Left]),
+            ok
+    end.
+
 assert_link_credit_runs_out(_Sender, 0) ->
     ct:fail(sufficient_link_credit);
 assert_link_credit_runs_out(Sender, Left) ->