From 104b4d0a94f09f9aaff114cb81a812a01d2ba838 Mon Sep 17 00:00:00 2001 From: Liu Ke <56467036+scrapstation@users.noreply.github.com> Date: Wed, 22 Apr 2026 14:39:41 +0000 Subject: [PATCH 1/7] fix(#25310): revent hanging when acquiring closed channel after rabbitmq restart --- .../src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/ChannelPool.cs | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/ChannelPool.cs b/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/ChannelPool.cs index 03d6029b69..c06c0b6a95 100644 --- a/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/ChannelPool.cs +++ b/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/ChannelPool.cs @@ -61,8 +61,6 @@ public class ChannelPool : IChannelPool, ISingletonDependency } } - poolItem.Acquire(); - if (poolItem.Channel.IsClosed) { await poolItem.DisposeAsync(); @@ -80,10 +78,10 @@ public class ChannelPool : IChannelPool, ISingletonDependency Channels.TryAdd(channelName, poolItem); } } - - poolItem.Acquire(); } + poolItem.Acquire(); + return new ChannelAccessor( poolItem.Channel, channelName, From ba4c0bd3d68dfcfc0e8e92c015904115f3f4408f Mon Sep 17 00:00:00 2001 From: maliming Date: Thu, 23 Apr 2026 11:59:59 +0800 Subject: [PATCH 2/7] feat: add RabbitMQ tests and improve channel pool handling --- framework/Volo.Abp.abpmdl | 4 + framework/Volo.Abp.slnx | 1 + .../Volo/Abp/RabbitMQ/ChannelPool.cs | 19 +- .../Volo.Abp.RabbitMQ.Tests.abppkg | 3 + .../Volo.Abp.RabbitMQ.Tests.csproj | 16 + .../Volo/Abp/RabbitMQ/ChannelPool_Tests.cs | 331 ++++++++++++++++++ 6 files changed, 369 insertions(+), 5 deletions(-) create mode 100644 framework/test/Volo.Abp.RabbitMQ.Tests/Volo.Abp.RabbitMQ.Tests.abppkg create mode 100644 framework/test/Volo.Abp.RabbitMQ.Tests/Volo.Abp.RabbitMQ.Tests.csproj create mode 100644 framework/test/Volo.Abp.RabbitMQ.Tests/Volo/Abp/RabbitMQ/ChannelPool_Tests.cs diff --git a/framework/Volo.Abp.abpmdl b/framework/Volo.Abp.abpmdl index afc52ca801..0935fedade 100644 --- a/framework/Volo.Abp.abpmdl +++ b/framework/Volo.Abp.abpmdl @@ -478,6 +478,10 @@ "path": "test/Volo.Abp.ObjectMapping.Tests/Volo.Abp.ObjectMapping.Tests.abppkg", "folder": "test" }, + "Volo.Abp.RabbitMQ.Tests": { + "path": "test/Volo.Abp.RabbitMQ.Tests/Volo.Abp.RabbitMQ.Tests.abppkg", + "folder": "test" + }, "Volo.Abp.Ddd.Application.Contracts": { "path": "src/Volo.Abp.Ddd.Application.Contracts/Volo.Abp.Ddd.Application.Contracts.abppkg", "folder": "src" diff --git a/framework/Volo.Abp.slnx b/framework/Volo.Abp.slnx index 26d462fb4f..df59da0515 100644 --- a/framework/Volo.Abp.slnx +++ b/framework/Volo.Abp.slnx @@ -240,6 +240,7 @@ + diff --git a/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/ChannelPool.cs b/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/ChannelPool.cs index c06c0b6a95..424c1600ad 100644 --- a/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/ChannelPool.cs +++ b/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/ChannelPool.cs @@ -1,5 +1,6 @@ using System; using System.Collections.Concurrent; +using System.Collections.Generic; using System.Diagnostics; using System.Linq; using System.Threading; @@ -63,14 +64,22 @@ public class ChannelPool : IChannelPool, ISingletonDependency if (poolItem.Channel.IsClosed) { - await poolItem.DisposeAsync(); - Channels.TryRemove(channelName, out _); - using (await Semaphore.LockAsync()) { - if (Channels.TryGetValue(channelName, out var existingChannelPoolItem3)) + if (Channels.TryGetValue(channelName, out var currentChannelPoolItem) && + ReferenceEquals(currentChannelPoolItem, poolItem)) + { + // This caller is the first to notice the cached channel is closed; rebuild it. + await poolItem.DisposeAsync(); + Channels.TryRemove(new KeyValuePair(channelName, poolItem)); + + poolItem = new ChannelPoolItem(await CreateChannelAsync(channelName, connectionName)); + Channels.TryAdd(channelName, poolItem); + } + else if (currentChannelPoolItem != null) { - poolItem = existingChannelPoolItem3; + // Another caller already rebuilt the channel; use the fresh one. + poolItem = currentChannelPoolItem; } else { diff --git a/framework/test/Volo.Abp.RabbitMQ.Tests/Volo.Abp.RabbitMQ.Tests.abppkg b/framework/test/Volo.Abp.RabbitMQ.Tests/Volo.Abp.RabbitMQ.Tests.abppkg new file mode 100644 index 0000000000..64c1552e37 --- /dev/null +++ b/framework/test/Volo.Abp.RabbitMQ.Tests/Volo.Abp.RabbitMQ.Tests.abppkg @@ -0,0 +1,3 @@ +{ + "role": "lib.test" +} diff --git a/framework/test/Volo.Abp.RabbitMQ.Tests/Volo.Abp.RabbitMQ.Tests.csproj b/framework/test/Volo.Abp.RabbitMQ.Tests/Volo.Abp.RabbitMQ.Tests.csproj new file mode 100644 index 0000000000..0c4b2a2260 --- /dev/null +++ b/framework/test/Volo.Abp.RabbitMQ.Tests/Volo.Abp.RabbitMQ.Tests.csproj @@ -0,0 +1,16 @@ + + + + + + net10.0 + + + + + + + + + + diff --git a/framework/test/Volo.Abp.RabbitMQ.Tests/Volo/Abp/RabbitMQ/ChannelPool_Tests.cs b/framework/test/Volo.Abp.RabbitMQ.Tests/Volo/Abp/RabbitMQ/ChannelPool_Tests.cs new file mode 100644 index 0000000000..3fe52ae563 --- /dev/null +++ b/framework/test/Volo.Abp.RabbitMQ.Tests/Volo/Abp/RabbitMQ/ChannelPool_Tests.cs @@ -0,0 +1,331 @@ +#nullable enable + +using System; +using System.Threading; +using System.Threading.Tasks; +using NSubstitute; +using RabbitMQ.Client; +using Shouldly; +using Xunit; + +namespace Volo.Abp.RabbitMQ; + +public class ChannelPool_Tests +{ + private static readonly TimeSpan RaceTimeout = TimeSpan.FromSeconds(5); + + // ---- Regression tests for issue #25310 / PR #25311 -------------------- + + /// + /// After a RabbitMQ restart the cached channel becomes closed. Two callers + /// that both observe the stale poolItem must not deadlock while the pool + /// replaces it with a fresh channel. + /// + [Fact] + public async Task AcquireAsync_Should_Not_Hang_When_Channel_Closed_With_Concurrent_Callers() + { + var fixture = BuildConnectionPool(); + var channelPool = new TestChannelPool(fixture.Pool); + + using (await channelPool.AcquireAsync("q")) + { + } + + fixture.Channel1.IsClosed.Returns(true); + fixture.Channel1 + .When(x => x.DisposeAsync()) + .Do(_ => Thread.Sleep(300)); + + using var barrier = new Barrier(2); + + var firstCaller = Task.Run(async () => + { + barrier.SignalAndWait(); + using var accessor = await channelPool.AcquireAsync("q"); + }); + + var secondCaller = Task.Run(async () => + { + barrier.SignalAndWait(); + await Task.Delay(50); + using var accessor = await channelPool.AcquireAsync("q"); + }); + + var completed = Task.WhenAll(firstCaller, secondCaller); + var winner = await Task.WhenAny(completed, Task.Delay(RaceTimeout)); + + winner.ShouldBe( + completed, + $"AcquireAsync is hanging on a stale poolItem. firstCaller={firstCaller.Status}, secondCaller={secondCaller.Status}"); + } + + /// + /// When several callers race through the closed-channel recovery path, only one new + /// channel must be created and the dictionary must keep exactly the channel that callers + /// receive — otherwise the replacement leaks on the broker side. + /// + [Fact] + public async Task AcquireAsync_Should_Create_Only_One_Replacement_Channel_When_Many_Callers_Race() + { + var fixture = BuildConnectionPool(); + var channelPool = new TestChannelPool(fixture.Pool); + + using (await channelPool.AcquireAsync("q")) + { + } + + fixture.Channel1.IsClosed.Returns(true); + + const int callerCount = 8; + using var barrier = new Barrier(callerCount); + var tasks = new Task[callerCount]; + for (var i = 0; i < callerCount; i++) + { + tasks[i] = Task.Run(async () => + { + barrier.SignalAndWait(); + using var accessor = await channelPool.AcquireAsync("q"); + return accessor.Channel; + }); + } + + var channels = await Task.WhenAll(tasks); + + foreach (var channel in channels) + { + channel.ShouldBe(fixture.Channel2, "every caller must receive the single replacement channel"); + } + + fixture.CreateChannelCalls.ShouldBe( + 2, + $"exactly one initial + one replacement channel should be created, but got {fixture.CreateChannelCalls}"); + } + + // ---- Behavior guard-rails: make sure the refactor preserves core semantics ---- + + [Fact] + public async Task AcquireAsync_Should_Return_Channel_From_ConnectionPool() + { + var fixture = BuildConnectionPool(); + var channelPool = new TestChannelPool(fixture.Pool); + + using var accessor = await channelPool.AcquireAsync("q"); + + accessor.Channel.ShouldBe(fixture.Channel1); + accessor.Name.ShouldBe("q"); + fixture.CreateChannelCalls.ShouldBe(1); + } + + [Fact] + public async Task AcquireAsync_Should_Reuse_Cached_Channel_On_Subsequent_Calls() + { + var fixture = BuildConnectionPool(); + var channelPool = new TestChannelPool(fixture.Pool); + + using (var first = await channelPool.AcquireAsync("q")) + { + first.Channel.ShouldBe(fixture.Channel1); + } + using (var second = await channelPool.AcquireAsync("q")) + { + second.Channel.ShouldBe(fixture.Channel1); + } + using (var third = await channelPool.AcquireAsync("q")) + { + third.Channel.ShouldBe(fixture.Channel1); + } + + fixture.CreateChannelCalls.ShouldBe(1, "a healthy cached channel must be reused"); + } + + [Fact] + public async Task AcquireAsync_Should_Keep_Separate_PoolItems_For_Different_ChannelNames() + { + var fixture = BuildConnectionPool(); + var channelPool = new TestChannelPool(fixture.Pool); + + using var a = await channelPool.AcquireAsync("queue-a"); + using var b = await channelPool.AcquireAsync("queue-b"); + + a.Channel.ShouldBe(fixture.Channel1); + b.Channel.ShouldBe(fixture.Channel2); + a.Name.ShouldBe("queue-a"); + b.Name.ShouldBe("queue-b"); + fixture.CreateChannelCalls.ShouldBe(2); + } + + [Fact] + public async Task AcquireAsync_Should_Serialize_Concurrent_Callers_On_Same_Channel() + { + var fixture = BuildConnectionPool(); + var channelPool = new TestChannelPool(fixture.Pool); + + // First caller holds the channel for a while. + var first = await channelPool.AcquireAsync("q"); + var firstAcquiredAt = DateTime.UtcNow; + + // Second caller should block until the first one releases. + var secondTask = Task.Run(async () => + { + var accessor = await channelPool.AcquireAsync("q"); + return (accessor, acquiredAt: DateTime.UtcNow); + }); + + await Task.Delay(100); + secondTask.IsCompleted.ShouldBeFalse("second caller must block while the channel is held"); + + first.Dispose(); + var firstReleasedAt = DateTime.UtcNow; + + var completed = await Task.WhenAny(secondTask, Task.Delay(RaceTimeout)); + completed.ShouldBe(secondTask, "second caller must be unblocked after release"); + + var (secondAccessor, secondAcquiredAt) = await secondTask; + secondAccessor.Channel.ShouldBe(fixture.Channel1, "the cached channel is reused"); + secondAcquiredAt.ShouldBeGreaterThanOrEqualTo(firstReleasedAt); + + secondAccessor.Dispose(); + _ = firstAcquiredAt; + fixture.CreateChannelCalls.ShouldBe(1, "channel is reused, never recreated"); + } + + [Fact] + public async Task AcquireAsync_Should_Not_Rebuild_When_Cached_Channel_Is_Healthy() + { + var fixture = BuildConnectionPool(); + var channelPool = new TestChannelPool(fixture.Pool); + + using (await channelPool.AcquireAsync("q")) + { + } + // Channel stays open, so subsequent calls must not go through the rebuild branch. + using (await channelPool.AcquireAsync("q")) + { + } + + await fixture.Channel1.DidNotReceive().DisposeAsync(); + fixture.CreateChannelCalls.ShouldBe(1); + } + + [Fact] + public async Task AcquireAsync_Should_Propagate_Exception_When_CreateChannel_Fails() + { + var channel = Substitute.For(); + channel.IsClosed.Returns(false); + + var connection = Substitute.For(); + var attempts = 0; + connection + .CreateChannelAsync(Arg.Any(), Arg.Any()) + .Returns(_ => + { + var n = Interlocked.Increment(ref attempts); + if (n == 1) + { + throw new InvalidOperationException("broker down"); + } + return Task.FromResult(channel); + }); + + var connectionPool = Substitute.For(); + connectionPool.GetAsync(Arg.Any()).Returns(Task.FromResult(connection)); + + var channelPool = new TestChannelPool(connectionPool); + + await Should.ThrowAsync(() => channelPool.AcquireAsync("q")); + + // Subsequent call should be able to succeed (broker came back). + using var accessor = await channelPool.AcquireAsync("q"); + accessor.Channel.ShouldBe(channel); + attempts.ShouldBe(2); + } + + [Fact] + public async Task AcquireAsync_Should_Throw_After_Pool_Disposed() + { + var fixture = BuildConnectionPool(); + var channelPool = new TestChannelPool(fixture.Pool); + + await channelPool.DisposeAsync(); + + await Should.ThrowAsync(() => channelPool.AcquireAsync("q")); + } + + [Fact] + public async Task AcquireAsync_Should_Only_Create_One_Channel_Even_When_First_Callers_Race() + { + var fixture = BuildConnectionPool(); + var channelPool = new TestChannelPool(fixture.Pool); + + const int callerCount = 8; + using var barrier = new Barrier(callerCount); + var tasks = new Task[callerCount]; + for (var i = 0; i < callerCount; i++) + { + tasks[i] = Task.Run(async () => + { + barrier.SignalAndWait(); + using var accessor = await channelPool.AcquireAsync("q"); + return accessor.Channel; + }); + } + + var channels = await Task.WhenAll(tasks); + + foreach (var channel in channels) + { + channel.ShouldBe(fixture.Channel1, "the first-creation semaphore must serialize initial creation"); + } + fixture.CreateChannelCalls.ShouldBe(1); + } + + // ---- Fixture --------------------------------------------------------- + + private static ConnectionPoolFixture BuildConnectionPool() + { + var channel1 = Substitute.For(); + channel1.IsClosed.Returns(false); + + var channel2 = Substitute.For(); + channel2.IsClosed.Returns(false); + + var fixture = new ConnectionPoolFixture(channel1, channel2); + + var connection = Substitute.For(); + connection + .CreateChannelAsync(Arg.Any(), Arg.Any()) + .Returns(_ => + { + var n = Interlocked.Increment(ref fixture.CreateChannelCallsField); + return n == 1 ? Task.FromResult(channel1) : Task.FromResult(channel2); + }); + + var connectionPool = Substitute.For(); + connectionPool.GetAsync(Arg.Any()).Returns(Task.FromResult(connection)); + + fixture.Pool = connectionPool; + return fixture; + } + + private sealed class ConnectionPoolFixture + { + public IConnectionPool Pool { get; set; } = default!; + public IChannel Channel1 { get; } + public IChannel Channel2 { get; } + public int CreateChannelCallsField; + public int CreateChannelCalls => CreateChannelCallsField; + + public ConnectionPoolFixture(IChannel channel1, IChannel channel2) + { + Channel1 = channel1; + Channel2 = channel2; + } + } + + private sealed class TestChannelPool : ChannelPool + { + public TestChannelPool(IConnectionPool connectionPool) : base(connectionPool) + { + } + } +} From 4286e3f581568a9a07acbc26c82bedb424396668 Mon Sep 17 00:00:00 2001 From: maliming Date: Thu, 23 Apr 2026 12:16:09 +0800 Subject: [PATCH 3/7] fix: use netstandard-compatible TryRemove overload in ChannelPool --- .../src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/ChannelPool.cs | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/ChannelPool.cs b/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/ChannelPool.cs index 424c1600ad..fc3440bd88 100644 --- a/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/ChannelPool.cs +++ b/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/ChannelPool.cs @@ -1,6 +1,5 @@ using System; using System.Collections.Concurrent; -using System.Collections.Generic; using System.Diagnostics; using System.Linq; using System.Threading; @@ -69,16 +68,14 @@ public class ChannelPool : IChannelPool, ISingletonDependency if (Channels.TryGetValue(channelName, out var currentChannelPoolItem) && ReferenceEquals(currentChannelPoolItem, poolItem)) { - // This caller is the first to notice the cached channel is closed; rebuild it. await poolItem.DisposeAsync(); - Channels.TryRemove(new KeyValuePair(channelName, poolItem)); + Channels.TryRemove(channelName, out _); poolItem = new ChannelPoolItem(await CreateChannelAsync(channelName, connectionName)); Channels.TryAdd(channelName, poolItem); } else if (currentChannelPoolItem != null) { - // Another caller already rebuilt the channel; use the fresh one. poolItem = currentChannelPoolItem; } else From 27edcf5691701f008ef1a0288a3e5d4b06418156 Mon Sep 17 00:00:00 2001 From: maliming Date: Thu, 23 Apr 2026 12:29:47 +0800 Subject: [PATCH 4/7] address copilot review: dispose stale channel outside lock; async delay in tests --- .../Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/ChannelPool.cs | 9 ++++++++- .../Volo/Abp/RabbitMQ/ChannelPool_Tests.cs | 6 ++++-- 2 files changed, 12 insertions(+), 3 deletions(-) diff --git a/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/ChannelPool.cs b/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/ChannelPool.cs index fc3440bd88..16cb580746 100644 --- a/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/ChannelPool.cs +++ b/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/ChannelPool.cs @@ -63,12 +63,14 @@ public class ChannelPool : IChannelPool, ISingletonDependency if (poolItem.Channel.IsClosed) { + ChannelPoolItem? staleItem = null; + using (await Semaphore.LockAsync()) { if (Channels.TryGetValue(channelName, out var currentChannelPoolItem) && ReferenceEquals(currentChannelPoolItem, poolItem)) { - await poolItem.DisposeAsync(); + staleItem = poolItem; Channels.TryRemove(channelName, out _); poolItem = new ChannelPoolItem(await CreateChannelAsync(channelName, connectionName)); @@ -84,6 +86,11 @@ public class ChannelPool : IChannelPool, ISingletonDependency Channels.TryAdd(channelName, poolItem); } } + + if (staleItem != null) + { + await staleItem.DisposeAsync(); + } } poolItem.Acquire(); diff --git a/framework/test/Volo.Abp.RabbitMQ.Tests/Volo/Abp/RabbitMQ/ChannelPool_Tests.cs b/framework/test/Volo.Abp.RabbitMQ.Tests/Volo/Abp/RabbitMQ/ChannelPool_Tests.cs index 3fe52ae563..62596cd038 100644 --- a/framework/test/Volo.Abp.RabbitMQ.Tests/Volo/Abp/RabbitMQ/ChannelPool_Tests.cs +++ b/framework/test/Volo.Abp.RabbitMQ.Tests/Volo/Abp/RabbitMQ/ChannelPool_Tests.cs @@ -33,8 +33,8 @@ public class ChannelPool_Tests fixture.Channel1.IsClosed.Returns(true); fixture.Channel1 - .When(x => x.DisposeAsync()) - .Do(_ => Thread.Sleep(300)); + .DisposeAsync() + .Returns(_ => new ValueTask(Task.Delay(300))); using var barrier = new Barrier(2); @@ -57,6 +57,8 @@ public class ChannelPool_Tests winner.ShouldBe( completed, $"AcquireAsync is hanging on a stale poolItem. firstCaller={firstCaller.Status}, secondCaller={secondCaller.Status}"); + + await completed; } /// From 61f9fb278c0fee070f08a608407e84164536e871 Mon Sep 17 00:00:00 2001 From: maliming Date: Thu, 23 Apr 2026 12:39:37 +0800 Subject: [PATCH 5/7] address copilot review: dispose stale channel even when recreate fails --- .../Volo/Abp/RabbitMQ/ChannelPool.cs | 46 ++++++++++--------- .../Volo/Abp/RabbitMQ/ChannelPool_Tests.cs | 36 +++++++++++++++ 2 files changed, 61 insertions(+), 21 deletions(-) diff --git a/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/ChannelPool.cs b/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/ChannelPool.cs index 16cb580746..7bf4fc5d9b 100644 --- a/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/ChannelPool.cs +++ b/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(); + } } } diff --git a/framework/test/Volo.Abp.RabbitMQ.Tests/Volo/Abp/RabbitMQ/ChannelPool_Tests.cs b/framework/test/Volo.Abp.RabbitMQ.Tests/Volo/Abp/RabbitMQ/ChannelPool_Tests.cs index 62596cd038..f001949ba1 100644 --- a/framework/test/Volo.Abp.RabbitMQ.Tests/Volo/Abp/RabbitMQ/ChannelPool_Tests.cs +++ b/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(); + staleChannel.IsClosed.Returns(false); + + var connection = Substitute.For(); + var attempts = 0; + connection + .CreateChannelAsync(Arg.Any(), Arg.Any()) + .Returns(_ => + { + var n = Interlocked.Increment(ref attempts); + if (n == 1) + { + return Task.FromResult(staleChannel); + } + throw new InvalidOperationException("broker still down"); + }); + + var connectionPool = Substitute.For(); + connectionPool.GetAsync(Arg.Any()).Returns(Task.FromResult(connection)); + + var channelPool = new TestChannelPool(connectionPool); + + using (await channelPool.AcquireAsync("q")) + { + } + + staleChannel.IsClosed.Returns(true); + + await Should.ThrowAsync(() => channelPool.AcquireAsync("q")); + + await staleChannel.Received(1).DisposeAsync(); + } + [Fact] public async Task AcquireAsync_Should_Throw_After_Pool_Disposed() { From 69adc13cbede1dd15b0c9e1f1717b26b52d7653f Mon Sep 17 00:00:00 2001 From: maliming Date: Thu, 23 Apr 2026 12:58:41 +0800 Subject: [PATCH 6/7] address copilot review: remove racy timestamp assertion in serialization test --- .../Volo/Abp/RabbitMQ/ChannelPool_Tests.cs | 14 ++------------ 1 file changed, 2 insertions(+), 12 deletions(-) diff --git a/framework/test/Volo.Abp.RabbitMQ.Tests/Volo/Abp/RabbitMQ/ChannelPool_Tests.cs b/framework/test/Volo.Abp.RabbitMQ.Tests/Volo/Abp/RabbitMQ/ChannelPool_Tests.cs index f001949ba1..9e7b737d74 100644 --- a/framework/test/Volo.Abp.RabbitMQ.Tests/Volo/Abp/RabbitMQ/ChannelPool_Tests.cs +++ b/framework/test/Volo.Abp.RabbitMQ.Tests/Volo/Abp/RabbitMQ/ChannelPool_Tests.cs @@ -162,32 +162,22 @@ public class ChannelPool_Tests var fixture = BuildConnectionPool(); var channelPool = new TestChannelPool(fixture.Pool); - // First caller holds the channel for a while. var first = await channelPool.AcquireAsync("q"); - var firstAcquiredAt = DateTime.UtcNow; - // Second caller should block until the first one releases. - var secondTask = Task.Run(async () => - { - var accessor = await channelPool.AcquireAsync("q"); - return (accessor, acquiredAt: DateTime.UtcNow); - }); + var secondTask = Task.Run(() => channelPool.AcquireAsync("q")); await Task.Delay(100); secondTask.IsCompleted.ShouldBeFalse("second caller must block while the channel is held"); first.Dispose(); - var firstReleasedAt = DateTime.UtcNow; var completed = await Task.WhenAny(secondTask, Task.Delay(RaceTimeout)); completed.ShouldBe(secondTask, "second caller must be unblocked after release"); - var (secondAccessor, secondAcquiredAt) = await secondTask; + var secondAccessor = await secondTask; secondAccessor.Channel.ShouldBe(fixture.Channel1, "the cached channel is reused"); - secondAcquiredAt.ShouldBeGreaterThanOrEqualTo(firstReleasedAt); secondAccessor.Dispose(); - _ = firstAcquiredAt; fixture.CreateChannelCalls.ShouldBe(1, "channel is reused, never recreated"); } From e4ccf01d2375b5c514c85c81918f260558cddfa3 Mon Sep 17 00:00:00 2001 From: maliming Date: Thu, 23 Apr 2026 13:27:43 +0800 Subject: [PATCH 7/7] address copilot review: wait for stale channel release before disposal; add barrier timeouts in tests --- .../Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/ChannelPool.cs | 1 + .../Volo/Abp/RabbitMQ/ChannelPool_Tests.cs | 8 ++++---- 2 files changed, 5 insertions(+), 4 deletions(-) diff --git a/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/ChannelPool.cs b/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/ChannelPool.cs index 7bf4fc5d9b..933e1bf005 100644 --- a/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/ChannelPool.cs +++ b/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/ChannelPool.cs @@ -92,6 +92,7 @@ public class ChannelPool : IChannelPool, ISingletonDependency { if (staleItem != null) { + staleItem.WaitIfInUse(TotalDisposeWaitDuration); await staleItem.DisposeAsync(); } } diff --git a/framework/test/Volo.Abp.RabbitMQ.Tests/Volo/Abp/RabbitMQ/ChannelPool_Tests.cs b/framework/test/Volo.Abp.RabbitMQ.Tests/Volo/Abp/RabbitMQ/ChannelPool_Tests.cs index 9e7b737d74..f631ce2d51 100644 --- a/framework/test/Volo.Abp.RabbitMQ.Tests/Volo/Abp/RabbitMQ/ChannelPool_Tests.cs +++ b/framework/test/Volo.Abp.RabbitMQ.Tests/Volo/Abp/RabbitMQ/ChannelPool_Tests.cs @@ -40,13 +40,13 @@ public class ChannelPool_Tests var firstCaller = Task.Run(async () => { - barrier.SignalAndWait(); + barrier.SignalAndWait(RaceTimeout); using var accessor = await channelPool.AcquireAsync("q"); }); var secondCaller = Task.Run(async () => { - barrier.SignalAndWait(); + barrier.SignalAndWait(RaceTimeout); await Task.Delay(50); using var accessor = await channelPool.AcquireAsync("q"); }); @@ -85,7 +85,7 @@ public class ChannelPool_Tests { tasks[i] = Task.Run(async () => { - barrier.SignalAndWait(); + barrier.SignalAndWait(RaceTimeout); using var accessor = await channelPool.AcquireAsync("q"); return accessor.Channel; }); @@ -292,7 +292,7 @@ public class ChannelPool_Tests { tasks[i] = Task.Run(async () => { - barrier.SignalAndWait(); + barrier.SignalAndWait(RaceTimeout); using var accessor = await channelPool.AcquireAsync("q"); return accessor.Channel; });