* Reproduce issue by artifically never acquiring rate limiter leases.

* Fix issue by catching `SemaphoreFullException` and checking that the exception was thrown in a valid failure case.

Author: lukebakken

projects/RabbitMQ.Client/Impl/Channel.PublisherConfirms.cs

@@ -344,7 +344,24 @@ private async Task MaybeEndPublisherConfirmationTracking(PublisherConfirmationIn
         {
             if (_publisherConfirmationsEnabled)
             {
-                _confirmSemaphore.Release();
+                try
+                {
+                    _confirmSemaphore.Release();
+                }
+                catch (SemaphoreFullException ex)
+                {
+                    /*
+                     * rabbitmq/rabbitmq-dotnet-client-1793
+                     * If MaybeStartPublisherConfirmationTracking throws an exception *prior* to acquiring
+                     * _confirmSemaphore, the above Release() call will throw SemaphoreFullException.
+                     * In "normal" cases, publisherConfirmationInfo will thus be null, but if not, throw
+                     * a "bug found" exception here.
+                     */
+                    if (publisherConfirmationInfo is not null)
+                    {
+                        throw new InvalidOperationException(InternalConstants.BugFound, ex);
+                    }
+                }
 
                 if (publisherConfirmationInfo is not null)
                 {

projects/Test/Integration/GH/ThrowsExceptionRateLimiter.cs renamed to projects/Test/Integration/GH/NeverAcquiredRateLimiter.cs

@@ -30,19 +30,39 @@
 //---------------------------------------------------------------------------
 
 using System;
+using System.Collections.Generic;
 using System.Threading;
 using System.Threading.RateLimiting;
 using System.Threading.Tasks;
 
 namespace Integration.GH
 {
-    public class ThrowsExceptionRateLimiter : RateLimiter
+    public class NeverAcquiredRateLimiter : RateLimiter
     {
         public override TimeSpan? IdleDuration => throw new NotImplementedException();
         public override RateLimiterStatistics GetStatistics() => throw new NotImplementedException();
+
         protected override ValueTask<RateLimitLease> AcquireAsyncCore(int permitCount, CancellationToken cancellationToken)
-            => throw new NotImplementedException();
+        {
+            return new ValueTask<RateLimitLease>(new NotAcquiredRateLimitLease());
+        }
+
         protected override RateLimitLease AttemptAcquireCore(int permitCount)
-            => throw new NotImplementedException();
+        {
+            return new NotAcquiredRateLimitLease();
+        }
+    }
+
+    public class NotAcquiredRateLimitLease : RateLimitLease
+    {
+        public override bool IsAcquired => false;
+
+        public override IEnumerable<string> MetadataNames => [];
+
+        public override bool TryGetMetadata(string metadataName, out object metadata)
+        {
+            metadata = string.Empty;
+            return true;
+        }
     }
 }

projects/Test/Integration/GH/TestGitHubIssues.cs

@@ -346,8 +346,7 @@ public async Task MaybeSomethingUpWithRateLimiter_GH1793()
             var channelOpts = new CreateChannelOptions(
                 publisherConfirmationsEnabled: true,
                 publisherConfirmationTrackingEnabled: true,
-                // outstandingPublisherConfirmationsRateLimiter: new ThrottlingRateLimiter(messageCount)
-                outstandingPublisherConfirmationsRateLimiter: new ThrowsExceptionRateLimiter()
+                outstandingPublisherConfirmationsRateLimiter: new NeverAcquiredRateLimiter()
             );
 
             _channel = await _conn.CreateChannelAsync(channelOpts);
@@ -357,39 +356,32 @@ public async Task MaybeSomethingUpWithRateLimiter_GH1793()
                 DeliveryMode = DeliveryModes.Persistent
             };
 
-            async Task PublishMessagesAsync()
+            for (int i = 0; i < messageCount; i++)
             {
-                for (int i = 0; i < messageCount; i++)
+                int retryCount = 0;
+                const int maxRetries = 3;
+                while (retryCount <= maxRetries)
                 {
-                    int retryCount = 0;
-                    const int maxRetries = 3;
-                    while (retryCount <= maxRetries)
+                    try
                     {
-                        try
+                        byte[] bytes = Encoding.UTF8.GetBytes("message");
+                        await Assert.ThrowsAnyAsync<InvalidOperationException>(async () =>
                         {
-                            byte[] bytes = Encoding.UTF8.GetBytes("message");
                             await _channel.BasicPublishAsync(string.Empty, string.Empty, true, properties, bytes);
-                            break;
-                        }
-                        catch (SemaphoreFullException ex0)
-                        {
-                            _output.WriteLine("{0} ex: {1}", _testDisplayName, ex0);
-                            retryCount++;
-                        }
-                        catch (PublishException)
-                        {
-                            retryCount++;
-                        }
+                        });
+                        break;
+                    }
+                    catch (SemaphoreFullException ex0)
+                    {
+                        _output.WriteLine("{0} ex: {1}", _testDisplayName, ex0);
+                        retryCount++;
+                    }
+                    catch (PublishException)
+                    {
+                        retryCount++;
                     }
                 }
             }
-
-            var publishTasks = new List<Task>();
-            for (int i = 0; i < messageCount; i++)
-            {
-                publishTasks.Add(Task.Run(PublishMessagesAsync));
-            }
-            await Task.WhenAll(publishTasks);
         }
     }
 }