address copilot review: dispose stale channel even when recreate fails

framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/ChannelPool.cs

@@ -64,32 +64,36 @@ public class ChannelPool : IChannelPool, ISingletonDependency
         if (poolItem.Channel.IsClosed)
         {
             ChannelPoolItem? staleItem = null;
-
-            using (await Semaphore.LockAsync())
+            try
             {
-                if (Channels.TryGetValue(channelName, out var currentChannelPoolItem) &&
-                    ReferenceEquals(currentChannelPoolItem, poolItem))
-                {
-                    staleItem = poolItem;
-                    Channels.TryRemove(channelName, out _);
-
-                    poolItem = new ChannelPoolItem(await CreateChannelAsync(channelName, connectionName));
-                    Channels.TryAdd(channelName, poolItem);
-                }
-                else if (currentChannelPoolItem != null)
+                using (await Semaphore.LockAsync())
                 {
-                    poolItem = currentChannelPoolItem;
-                }
-                else
-                {
-                    poolItem = new ChannelPoolItem(await CreateChannelAsync(channelName, connectionName));
-                    Channels.TryAdd(channelName, poolItem);
+                    if (Channels.TryGetValue(channelName, out var currentChannelPoolItem) &&
+                        ReferenceEquals(currentChannelPoolItem, poolItem))
+                    {
+                        staleItem = poolItem;
+                        Channels.TryRemove(channelName, out _);
+
+                        poolItem = new ChannelPoolItem(await CreateChannelAsync(channelName, connectionName));
+                        Channels.TryAdd(channelName, poolItem);
+                    }
+                    else if (currentChannelPoolItem != null)
+                    {
+                        poolItem = currentChannelPoolItem;
+                    }
+                    else
+                    {
+                        poolItem = new ChannelPoolItem(await CreateChannelAsync(channelName, connectionName));
+                        Channels.TryAdd(channelName, poolItem);
+                    }
                 }
             }
-
-            if (staleItem != null)
+            finally
             {
-                await staleItem.DisposeAsync();
+                if (staleItem != null)
+                {
+                    await staleItem.DisposeAsync();
+                }
             }
         }
 

framework/test/Volo.Abp.RabbitMQ.Tests/Volo/Abp/RabbitMQ/ChannelPool_Tests.cs

@@ -242,6 +242,42 @@ public class ChannelPool_Tests
         attempts.ShouldBe(2);
     }
 
+    [Fact]
+    public async Task AcquireAsync_Should_Dispose_Stale_Channel_Even_When_Recreate_Fails()
+    {
+        var staleChannel = Substitute.For<IChannel>();
+        staleChannel.IsClosed.Returns(false);
+
+        var connection = Substitute.For<IConnection>();
+        var attempts = 0;
+        connection
+            .CreateChannelAsync(Arg.Any<CreateChannelOptions?>(), Arg.Any<CancellationToken>())
+            .Returns(_ =>
+            {
+                var n = Interlocked.Increment(ref attempts);
+                if (n == 1)
+                {
+                    return Task.FromResult(staleChannel);
+                }
+                throw new InvalidOperationException("broker still down");
+            });
+
+        var connectionPool = Substitute.For<IConnectionPool>();
+        connectionPool.GetAsync(Arg.Any<string?>()).Returns(Task.FromResult(connection));
+
+        var channelPool = new TestChannelPool(connectionPool);
+
+        using (await channelPool.AcquireAsync("q"))
+        {
+        }
+
+        staleChannel.IsClosed.Returns(true);
+
+        await Should.ThrowAsync<InvalidOperationException>(() => channelPool.AcquireAsync("q"));
+
+        await staleChannel.Received(1).DisposeAsync();
+    }
+
     [Fact]
     public async Task AcquireAsync_Should_Throw_After_Pool_Disposed()
     {