Refactor: Move semaphore ownership to PublisherConfirmationInfo

Move the semaphore ownership and release responsibility from MaybeEndPublisherConfirmationTrackingAsync
to the PublisherConfirmationInfo class itself. This improves encapsulation by making the PublisherConfirmationInfo
responsible for managing the semaphore resource it owns.

Changes:
- Add _semaphore field and _semaphoreReleased flag to PublisherConfirmationInfo
- Add ReleaseSemaphore() method with idempotence check
- Update constructor to accept SemaphoreSlim parameter
- Update Dispose() to call ReleaseSemaphore()
- Simplify control flow in MaybeStartPublisherConfirmationTrackingAsync (early return pattern)
- Update MaybeEndPublisherConfirmationTrackingAsync to use ReleaseSemaphore() from info object

Author: EvheniyHlushko

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

@@ -54,20 +54,34 @@ internal partial class Channel : IChannel, IRecoverable
 
         private sealed class PublisherConfirmationInfo : IDisposable
         {
-            private TaskCompletionSource<bool>? _publisherConfirmationTcs;
+            private readonly SemaphoreSlim _semaphore;
+            private readonly TaskCompletionSource<bool>? _publisherConfirmationTcs;
             private readonly RateLimitLease? _lease;
+            private bool _semaphoreReleased;
 
-            internal PublisherConfirmationInfo(ulong publishSequenceNumber,
+            internal PublisherConfirmationInfo(
+                SemaphoreSlim semaphore,
+                ulong publishSequenceNumber,
                 TaskCompletionSource<bool>? publisherConfirmationTcs,
                 RateLimitLease? lease)
             {
+                _semaphore = semaphore;
                 PublishSequenceNumber = publishSequenceNumber;
                 _publisherConfirmationTcs = publisherConfirmationTcs;
                 _lease = lease;
             }
 
             internal ulong PublishSequenceNumber { get; }
 
+            internal void ReleaseSemaphore()
+            {
+                if (!_semaphoreReleased)
+                {
+                    _semaphoreReleased = true;
+                    _semaphore.Release();
+                }
+            }
+
             internal async Task MaybeWaitForConfirmationAsync(CancellationToken cancellationToken)
             {
                 if (_publisherConfirmationTcs is not null)
@@ -92,6 +106,7 @@ internal bool MaybeHandleException(Exception ex)
 
             public void Dispose()
             {
+                ReleaseSemaphore();
                 _lease?.Dispose();
             }
         }
@@ -291,54 +306,51 @@ await _confirmSemaphore.WaitAsync(reason.CancellationToken)
         [MethodImpl(MethodImplOptions.AggressiveInlining)]
         private async Task<PublisherConfirmationInfo?> MaybeStartPublisherConfirmationTrackingAsync(CancellationToken cancellationToken)
         {
-            if (_publisherConfirmationsEnabled)
+            if (!_publisherConfirmationsEnabled)
             {
-                RateLimitLease? lease = null;
-                if (_publisherConfirmationTrackingEnabled)
-                {
-                    if (_outstandingPublisherConfirmationsRateLimiter is not null)
-                    {
-                        lease = await _outstandingPublisherConfirmationsRateLimiter.AcquireAsync(
-                            cancellationToken: cancellationToken)
-                            .ConfigureAwait(false);
-
-                        if (!lease.IsAcquired)
-                        {
-                            throw new InvalidOperationException("Could not acquire a lease from the rate limiter.");
-                        }
-                    }
-                }
+                return null;
+            }
 
-                try
-                {
-                    await _confirmSemaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
-                }
-                catch (OperationCanceledException)
+            RateLimitLease? lease = null;
+            if (_publisherConfirmationTrackingEnabled && _outstandingPublisherConfirmationsRateLimiter is not null)
+            {
+                lease = await _outstandingPublisherConfirmationsRateLimiter.AcquireAsync(
+                    cancellationToken: cancellationToken)
+                    .ConfigureAwait(false);
+                if (!lease.IsAcquired)
                 {
-                    lease?.Dispose();
-                    throw;
+                    throw new InvalidOperationException("Could not acquire a lease from the rate limiter.");
                 }
+            }
 
-                ulong publishSequenceNumber = _nextPublishSeqNo;
+            try
+            {
+                await _confirmSemaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
+            }
+            catch (OperationCanceledException)
+            {
+                lease?.Dispose();
+                throw;
+            }
 
-                TaskCompletionSource<bool>? publisherConfirmationTcs = null;
-                if (_publisherConfirmationTrackingEnabled)
+            ulong publishSequenceNumber = _nextPublishSeqNo;
+
+            TaskCompletionSource<bool>? publisherConfirmationTcs = null;
+            if (_publisherConfirmationTrackingEnabled)
+            {
+                publisherConfirmationTcs = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+                if (!_confirmsTaskCompletionSources.TryAdd(publishSequenceNumber, publisherConfirmationTcs))
                 {
-                    publisherConfirmationTcs = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
-                    if (!_confirmsTaskCompletionSources.TryAdd(publishSequenceNumber, publisherConfirmationTcs))
-                    {
-                        throw new InvalidOperationException($"Failed to track the publisher confirmation for sequence number '{publishSequenceNumber}' because it already exists.");
-                    }
+                    _confirmSemaphore.Release();
+                    lease?.Dispose();
+                    throw new InvalidOperationException($"Failed to track the publisher confirmation for sequence number '{publishSequenceNumber}' because it already exists.");
                 }
+            }
 
-                _nextPublishSeqNo++;
+            _nextPublishSeqNo++;
+
+            return new PublisherConfirmationInfo(_confirmSemaphore, publishSequenceNumber, publisherConfirmationTcs, lease);
 
-                return new PublisherConfirmationInfo(publishSequenceNumber, publisherConfirmationTcs, lease);
-            }
-            else
-            {
-                return null;
-            }
         }
 
         [MethodImpl(MethodImplOptions.AggressiveInlining)]
@@ -367,28 +379,28 @@ private bool MaybeHandleExceptionWithEnabledPublisherConfirmations(PublisherConf
         private async Task MaybeEndPublisherConfirmationTrackingAsync(PublisherConfirmationInfo? publisherConfirmationInfo,
             CancellationToken cancellationToken)
         {
-            if (_publisherConfirmationsEnabled)
+            if (publisherConfirmationInfo is null)
             {
-                if (publisherConfirmationInfo is not null)
-                {
-                    _confirmSemaphore.Release();
+                return;
+            }
 
-                    try
-                    {
-                        await publisherConfirmationInfo.MaybeWaitForConfirmationAsync(cancellationToken)
-                            .ConfigureAwait(false);
-                    }
-                    catch (OperationCanceledException)
-                    {
-                        _confirmsTaskCompletionSources.TryRemove(publisherConfirmationInfo.PublishSequenceNumber, out _);
-                        throw;
-                    }
-                    finally
-                    {
-                        publisherConfirmationInfo.Dispose();
-                    }
-                }
+            publisherConfirmationInfo.ReleaseSemaphore();
+
+            try
+            {
+                await publisherConfirmationInfo.MaybeWaitForConfirmationAsync(cancellationToken)
+                    .ConfigureAwait(false);
+            }
+            catch (OperationCanceledException)
+            {
+                _confirmsTaskCompletionSources.TryRemove(publisherConfirmationInfo.PublishSequenceNumber, out _);
+                throw;
             }
+            finally
+            {
+                publisherConfirmationInfo.Dispose();
+            }
+
         }
 
         [MethodImpl(MethodImplOptions.AggressiveInlining)]