Set a floor of zero for incoming-window

Prior to this commit, when the sending client overshot RabbitMQ's incoming-window
(which is allowed in the event of a cluster wide memory or disk alarm),
and RabbitMQ sent a FLOW frame to the client, RabbitMQ sent a negative
incoming-window field in the FLOW frame causing the following crash in
the writer proc:
```
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)
```

This commit fixes this crash by maintaning a floor of zero for
incoming-window in the FLOW frame.

Fixes #12816

Author: ansd

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,

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,51 @@ 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),
+    {ok, Sender} = amqp10_client:attach_sender_link(Session, <<"sender">>, Address, mixed),
+    ok = wait_for_credit(Sender),
+
+    %% Send many messages while a memory alarm kicks in.
+    spawn_link(
+      fun() ->
+              ct:pal("sending first message..."),
+              [begin
+                   Bin = integer_to_binary(I),
+                   _ = amqp10_client:send_msg(Sender, amqp10_msg:new(Bin, Bin, true))
+               end || I <- lists:seq(1, 50_000)],
+              ct:pal("sent last message")
+      end),
+    DefaultWatermark = rpc(Config, vm_memory_monitor, get_vm_memory_high_watermark, []),
+    ct:pal("setting memory alarm..."),
+    ok = rpc(Config, vm_memory_monitor, set_vm_memory_high_watermark, [0]),
+    ct:pal("set memory alarm"),
+    timer:sleep(100),
+    %% At some point, RabbitMQ blocked our client sending messages.
+    %% However, once the memory alarm clears, our client should be able to send a message again.
+    ct:pal("clearing memory alarm..."),
+    ok = rpc(Config, vm_memory_monitor, set_vm_memory_high_watermark, [DefaultWatermark]),
+    ct:pal("cleared memory alarm"),
+    timer:sleep(100),
+    ?assertEqual(ok,
+                 amqp10_client:send_msg(Sender, amqp10_msg:new(<<"tag">>, <<"msg">>, false))),
+    ok = wait_for_accepted(<<"tag">>),
+
+    ok = amqp10_client:detach_link(Sender),
+    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, []),