Fix error insufficient_credit for AMQP 1.0 shovel

WHY:

Shovelling from RabbitMQ to Azure Service Bus and Azure Event Hub fails.

Reported in
https://discord.com/channels/1092487794984755311/1092487794984755314/1169894510743011430

Reproduction steps:
1. Follow https://learn.microsoft.com/en-us/azure/service-bus-messaging/service-bus-integrate-with-rabbitmq
2. Publish messages to RabbitMQ:
```
java -jar target/perf-test.jar -x 1 -y 0 -u azure -p -C 100000 -s 1 -c 100000
```

Prior to this commit, after a few seconds and after around 20k messages
arrived in Azure, RabbitMQ errored and logged:
```
{function_clause,
    [{amqp10_client_connection,close_sent,
         [info,
          {'EXIT',<0.949.0>,
              {{badmatch,{error,insufficient_credit}},
               [{rabbit_amqp10_shovel,forward,4,
                    [{file,"rabbit_amqp10_shovel.erl"},
                     {line,334}]},
                {rabbit_shovel_worker,handle_info,2,
                    [{file,"rabbit_shovel_worker.erl"},
                     {line,101}]},
                {gen_server2,handle_msg,2,
                    [{file,"gen_server2.erl"},{line,1056}]},
                {proc_lib,init_p_do_apply,3,
                    [{file,"proc_lib.erl"},{line,241}]}]}},
```

After this commit, all 100k messages get shovelled to Azure Service Bus.

HOW:

1. Fix link credit accounting in Erlang AMQP 1.0 client library. For each
message being published, link credit must be decreased by 1 instead of
being increased by 1.

2. If the shovel plugin runs out of credits, it must wait until the
   receiver (Azure Service Bus) grants more credits to RabbitMQ.

Note that the solution in this commit is rather a naive quick fix for one
obvious bug. AMQP 1.0 integration between RabbitMQ and Azure Service Bus is
not tested and not guaranteed at this point in time.
More work will be needed in the future, some work is done as part of
#9022

Author: ansd

deps/amqp10_client/src/amqp10_client_session.erl

@@ -955,7 +955,7 @@ update_link(Link = #link{output_handle = OutHandle},
             State#state{links = Links#{OutHandle => Link}}.
 
 incr_link_counters(#link{link_credit = LC, delivery_count = DC} = Link) ->
-    Link#link{delivery_count = DC+1, link_credit = LC+1}.
+    Link#link{delivery_count = DC+1, link_credit = LC-1}.
 
 append_partial_transfer(Transfer, Payload,
                         #link{partial_transfers = undefined} = Link) ->

deps/rabbitmq_shovel/src/rabbit_amqp10_shovel.erl

@@ -38,7 +38,7 @@
 -import(rabbit_data_coercion, [to_binary/1]).
 
 -define(INFO(Text, Args), rabbit_log_shovel:info(Text, Args)).
--define(LINK_CREDIT_TIMEOUT, 5000).
+-define(LINK_CREDIT_TIMEOUT, 20_000).
 
 -type state() :: rabbit_shovel_behaviour:state().
 -type uri() :: rabbit_shovel_behaviour:uri().
@@ -173,7 +173,8 @@ dest_endpoint(#{shovel_type := dynamic,
                 dest := #{target_address := Addr}}) ->
     [{dest_address, Addr}].
 
--spec handle_source(Msg :: any(), state()) -> not_handled | state().
+-spec handle_source(Msg :: any(), state()) ->
+    not_handled | state() | {stop, any()}.
 handle_source({amqp10_msg, _LinkRef, Msg}, State) ->
     Tag = amqp10_msg:delivery_id(Msg),
     Payload = amqp10_msg:body_bin(Msg),
@@ -312,7 +313,8 @@ status(_) ->
     ignore.
 
 -spec forward(Tag :: tag(), Props :: #{atom() => any()},
-              Payload :: binary(), state()) -> state().
+              Payload :: binary(), state()) ->
+    state() | {stop, any()}.
 forward(_Tag, _Props, _Payload,
         #{source := #{remaining_unacked := 0}} = State) ->
     State;
@@ -331,17 +333,33 @@ forward(Tag, Props, Payload,
     Msg = add_timestamp_header(
             State, set_message_properties(
                      Props, add_forward_headers(State, Msg0))),
-    ok = amqp10_client:send_msg(Link, Msg),
-    rabbit_shovel_behaviour:decr_remaining_unacked(
-      case AckMode of
-          no_ack ->
-              rabbit_shovel_behaviour:decr_remaining(1, State);
-          on_confirm ->
-              State#{dest => Dst#{unacked => Unacked#{OutTag => Tag}}};
-          on_publish ->
-              State1 = rabbit_shovel_behaviour:ack(Tag, false, State),
-              rabbit_shovel_behaviour:decr_remaining(1, State1)
-      end).
+    case send_msg(Link, Msg) of
+        ok ->
+            rabbit_shovel_behaviour:decr_remaining_unacked(
+              case AckMode of
+                  no_ack ->
+                      rabbit_shovel_behaviour:decr_remaining(1, State);
+                  on_confirm ->
+                      State#{dest => Dst#{unacked => Unacked#{OutTag => Tag}}};
+                  on_publish ->
+                      State1 = rabbit_shovel_behaviour:ack(Tag, false, State),
+                      rabbit_shovel_behaviour:decr_remaining(1, State1)
+              end);
+        Stop ->
+            Stop
+    end.
+
+send_msg(Link, Msg) ->
+    case amqp10_client:send_msg(Link, Msg) of
+        ok ->
+            ok;
+        {error, insufficient_credit} ->
+            receive {amqp10_event, {link, Link, credited}} ->
+                        ok = amqp10_client:send_msg(Link, Msg)
+            after ?LINK_CREDIT_TIMEOUT ->
+                      {stop, credited_timeout}
+            end
+    end.
 
 new_message(Tag, Payload, #{ack_mode := AckMode,
                             dest := #{properties := Props,

deps/rabbitmq_shovel/src/rabbit_shovel_behaviour.erl

@@ -140,7 +140,8 @@ source_endpoint(#{source := #{module := Mod}} = State) ->
 dest_endpoint(#{dest := #{module := Mod}} = State) ->
     Mod:dest_endpoint(State).
 
--spec forward(tag(), #{atom() => any()}, binary(), state()) -> state().
+-spec forward(tag(), #{atom() => any()}, binary(), state()) ->
+    state() | {stop, any()}.
 forward(Tag, Props, Payload, #{dest := #{module := Mod}} = State) ->
     Mod:forward(Tag, Props, Payload, State).