From 9c190162e3f9fb5e376a3b3c957356d9f939897c Mon Sep 17 00:00:00 2001
From: Florian Lemaitre <flemaitre@aneo.fr>
Date: Mon, 6 Oct 2025 14:41:03 +0200
Subject: [PATCH] Adjust credits manually before each pull

---
 Adaptors/Amqp/src/PullQueueStorage.cs | 34 ++++++++++++++++++++++-----
 Adaptors/QueueCommon/src/Amqp.cs      | 16 +++++++++----
 2 files changed, 39 insertions(+), 11 deletions(-)

diff --git a/Adaptors/Amqp/src/PullQueueStorage.cs b/Adaptors/Amqp/src/PullQueueStorage.cs
index ab789792d..b761cd62a 100644
--- a/Adaptors/Amqp/src/PullQueueStorage.cs
+++ b/Adaptors/Amqp/src/PullQueueStorage.cs
@@ -97,6 +97,21 @@ public async IAsyncEnumerable<IQueueMessageHandler> PullMessagesAsync(string
                                                                                                                            $"{partitionId}###SenderLink{i}",
                                                                                                                            $"{partitionId}###q{i}")))
                                                         .ToArray());
+    var receiversSet = new bool[receiversArray.Length];
+    await using var rollbackCredits = new Deferrer(async () =>
+                                                   {
+                                                     // Rollback credits for all receivers whose credits have been set during the pull
+                                                     for (var i = 0; i < receiversArray.Length; i++)
+                                                     {
+                                                       if (receiversSet[i])
+                                                       {
+                                                         var receiver = await receiversArray[i];
+                                                         receiver.SetCredit(Options.LinkCredit,
+                                                                            false);
+                                                       }
+                                                     }
+                                                   });
+
     while (nbPulledMessage < nbMessages)
     {
       var currentNbMessages = nbPulledMessage;
@@ -111,6 +126,16 @@ public async IAsyncEnumerable<IQueueMessageHandler> PullMessagesAsync(string
           try
           {
             receiver = await receiversArray[i];
+
+            // Do not set credits if they have been set for this receiver in the same pull
+            if (!receiversSet[i])
+            {
+              // Set credit to number of messages to fetch + prefetch
+              receiver.SetCredit(nbMessages - nbPulledMessage + Options.LinkCredit,
+                                 false);
+              receiversSet[i] = true;
+            }
+
             message = await receiver.ReceiveAsync(TimeSpan.FromMilliseconds(100))
                                     .ConfigureAwait(false);
             break;
@@ -177,12 +202,9 @@ private AsyncLazy<IReceiverLink> CreateReceiver(string          partitionId,
                                        $"{partitionId}###ReceiverLink{link}",
                                        $"{partitionId}###q{link}");
 
-             /* linkCredit_: the maximum number of messages the
-                                       * remote peer can send to the receiver.
-                                       * With the goal of minimizing/deactivating
-                                       * prefetching, a value of 1 gave us the desired
-                                       * behavior. We pick a default value of 2 to have "some cache". */
-             rl.SetCredit(Options.LinkCredit);
+             // Prefetch
+             rl.SetCredit(Options.LinkCredit,
+                          false);
              return rl;
            });
 }
diff --git a/Adaptors/QueueCommon/src/Amqp.cs b/Adaptors/QueueCommon/src/Amqp.cs
index 7cc5113a3..19a333874 100644
--- a/Adaptors/QueueCommon/src/Amqp.cs
+++ b/Adaptors/QueueCommon/src/Amqp.cs
@@ -78,12 +78,18 @@ public class Amqp
   public int MaxRetries { get; set; }
 
   /// <summary>
-  ///   Link credit for flow control in the AMQP connection. The minimum valued supported is 1
-  ///   For more details see:
-  ///   <a
-  ///     href="https: //www.rabbitmq.com/blog/2024/09/02/amqp-flow-control">
-  ///   </a>
+  ///   Link credit for flow control in the AMQP connection. The minimum valued supported is 0
   /// </summary>
+  /// <remarks>
+  ///   <p>
+  ///     Credits are set to LinkCredit + number of messages to pull each time a pull occur.
+  ///     Consequently, the option LinkCredit represent the number of messages to prefetch by the AMQP driver.
+  ///   </p>
+  ///   <p>
+  ///     For more details see:
+  ///     <a href="https: //www.rabbitmq.com/blog/2024/09/02/amqp-flow-control" />
+  ///   </p>
+  /// </remarks>
   public int LinkCredit { get; set; }
 
   /// <summary>