From 05aba70a938645c52ace8827ac7aa0ad828a384c Mon Sep 17 00:00:00 2001 From: Marko Lahma Date: Mon, 6 Jul 2026 20:56:20 +0300 Subject: [PATCH 1/2] fix: CircuitWatcher.Dispose() does not stop the ping-until-reconnected loop CircuitWatcher only stored the caller's CancellationToken and disposed the Task wrapper on Dispose() -- neither actually stops pingUntilConnectedAsync, which keeps running against the still-live caller token. SendingAgent's own DisposeAsync() never called circuit watcher disposal either, so a sender whose circuit had tripped kept pinging a dead destination forever after the owning host was disposed. Reproduced with a Kafka producer pointed at a permanently unreachable broker: disposing the WebApplicationFactory-hosted app never stopped the background reconnect/ping loop, keeping the test process alive indefinitely regardless of how quickly each connection attempt failed. - CircuitWatcher now links the caller's token into its own CancellationTokenSource and cancels it in Dispose(), so Dispose() can stop the loop on its own instead of only releasing the Task wrapper. - SendingAgent.DisposeAsync() now disposes the circuit watcher before tearing down the sender. Added regression tests exercising both the isolated CircuitWatcher and a BufferedSendingAgent end-to-end; both fail (ping count keeps climbing after Dispose()) without this fix and pass with it. --- .../Sending/CircuitWatcherTester.cs | 6 +- .../Sending/SendingAgentDisposalTests.cs | 99 +++++++++++++++++++ .../Transports/Sending/CircuitWatcher.cs | 17 ++-- .../Transports/Sending/SendingAgent.cs | 5 + 4 files changed, 119 insertions(+), 8 deletions(-) create mode 100644 src/Testing/CoreTests/Transports/Sending/SendingAgentDisposalTests.cs diff --git a/src/Testing/CoreTests/Transports/Sending/CircuitWatcherTester.cs b/src/Testing/CoreTests/Transports/Sending/CircuitWatcherTester.cs index b62ecb839..c06a83fac 100644 --- a/src/Testing/CoreTests/Transports/Sending/CircuitWatcherTester.cs +++ b/src/Testing/CoreTests/Transports/Sending/CircuitWatcherTester.cs @@ -42,9 +42,11 @@ public StubCircuit(int failureCount, ManualResetEvent completed) public bool SupportsNativeScheduledSend => true; + public int CallCount => _count; + public Task TryToResumeAsync(CancellationToken cancellationToken) { - _count++; + Interlocked.Increment(ref _count); if (_count < _failureCount) { @@ -60,7 +62,7 @@ public Task ResumeAsync(CancellationToken cancellationToken) return Task.CompletedTask; } - public TimeSpan RetryInterval { get; } = 50.Milliseconds(); + public TimeSpan RetryInterval { get; set; } = 50.Milliseconds(); public void Dispose() { diff --git a/src/Testing/CoreTests/Transports/Sending/SendingAgentDisposalTests.cs b/src/Testing/CoreTests/Transports/Sending/SendingAgentDisposalTests.cs new file mode 100644 index 000000000..50d7c91f0 --- /dev/null +++ b/src/Testing/CoreTests/Transports/Sending/SendingAgentDisposalTests.cs @@ -0,0 +1,99 @@ +using JasperFx.Core; +using Microsoft.Extensions.Logging.Abstractions; +using NSubstitute; +using Wolverine.ComplianceTests; +using Wolverine.Logging; +using Wolverine.Transports.Local; +using Wolverine.Transports.Sending; +using Xunit; + +namespace CoreTests.Transports.Sending; + +public class SendingAgentDisposalTests +{ + [Fact] + public async Task circuit_watcher_dispose_stops_the_ping_loop() + { + using var completed = new ManualResetEvent(false); + var circuit = new StubCircuit(int.MaxValue, completed) { RetryInterval = 20.Milliseconds() }; + + var watcher = new CircuitWatcher(circuit, default); + + await waitUntilAsync(() => circuit.CallCount > 0); + + watcher.Dispose(); + + var countAtDispose = circuit.CallCount; + await Task.Delay(200.Milliseconds()); + + // Before the fix, Dispose() only released the Task wrapper -- pingUntilConnectedAsync kept + // running against the caller's (still live) token, so this count kept climbing forever. + circuit.CallCount.ShouldBe(countAtDispose); + } + + [Fact] + public async Task disposing_a_sending_agent_stops_its_circuit_watcher() + { + var sender = new AlwaysFailingSender(); + var endpoint = new LocalQueue("disposing-a-sending-agent-stops-its-circuit-watcher") + { + FailuresBeforeCircuitBreaks = 1, + PingIntervalForCircuitResume = 20.Milliseconds() + }; + + var agent = new BufferedSendingAgent( + NullLogger.Instance, + Substitute.For(), + sender, + new DurabilitySettings(), + endpoint); + + // The one guaranteed failure trips the circuit breaker and starts the CircuitWatcher. + await agent.EnqueueOutgoingAsync(ObjectMother.Envelope()); + + await waitUntilAsync(() => sender.PingCount > 0); + + await agent.DisposeAsync(); + + var pingCountAtDispose = sender.PingCount; + await Task.Delay(200.Milliseconds()); + + // Before the fix, SendingAgent.DisposeAsync() never touched the CircuitWatcher, so a sender + // pointed at a permanently unreachable destination (e.g. Kafka against a dead broker) kept + // pinging forever in the background -- keeping the owning process alive indefinitely. + sender.PingCount.ShouldBe(pingCountAtDispose); + } + + private static async Task waitUntilAsync(Func condition) + { + for (var i = 0; i < 200; i++) + { + if (condition()) + { + return; + } + + await Task.Delay(10); + } + + throw new TimeoutException("Condition was never met"); + } +} + +internal class AlwaysFailingSender : ISender +{ + private int _pingCount; + + public int PingCount => _pingCount; + + public bool SupportsNativeScheduledSend => false; + public Uri Destination { get; } = "local://always-failing".ToUri(); + + public Task PingAsync() + { + Interlocked.Increment(ref _pingCount); + return Task.FromResult(false); + } + + public ValueTask SendAsync(Envelope envelope) => throw new Exception("Nope!"); +} diff --git a/src/Wolverine/Transports/Sending/CircuitWatcher.cs b/src/Wolverine/Transports/Sending/CircuitWatcher.cs index 76688454c..bed38dfea 100644 --- a/src/Wolverine/Transports/Sending/CircuitWatcher.cs +++ b/src/Wolverine/Transports/Sending/CircuitWatcher.cs @@ -15,35 +15,40 @@ internal interface ISenderCircuit : ICircuitTester internal class CircuitWatcher : IDisposable { - private readonly CancellationToken _cancellation; + // Linked (not just stored) so Dispose() can stop the loop on its own, independent of whether the + // caller's token has been cancelled yet. Without this, Dispose() only released the Task wrapper -- + // pingUntilConnectedAsync kept running against the still-live caller token. + private readonly CancellationTokenSource _cancellation; private readonly ISenderCircuit _senderCircuit; private readonly Task _task; public CircuitWatcher(ISenderCircuit senderCircuit, CancellationToken cancellation) { _senderCircuit = senderCircuit; - _cancellation = cancellation; + _cancellation = CancellationTokenSource.CreateLinkedTokenSource(cancellation); - _task = Task.Run(pingUntilConnectedAsync, _cancellation); + _task = Task.Run(pingUntilConnectedAsync, _cancellation.Token); } public void Dispose() { + _cancellation.Cancel(); _task.SafeDispose(); + _cancellation.Dispose(); } private async Task pingUntilConnectedAsync() { using var timer=new PeriodicTimer(_senderCircuit.RetryInterval); - while (await timer.WaitForNextTickAsync(_cancellation)) + while (await timer.WaitForNextTickAsync(_cancellation.Token)) { try { - var pinged = await _senderCircuit.TryToResumeAsync(_cancellation); + var pinged = await _senderCircuit.TryToResumeAsync(_cancellation.Token); if (pinged) { - await _senderCircuit.ResumeAsync(_cancellation); + await _senderCircuit.ResumeAsync(_cancellation.Token); return; } } diff --git a/src/Wolverine/Transports/Sending/SendingAgent.cs b/src/Wolverine/Transports/Sending/SendingAgent.cs index fcf89e60a..7a7840107 100644 --- a/src/Wolverine/Transports/Sending/SendingAgent.cs +++ b/src/Wolverine/Transports/Sending/SendingAgent.cs @@ -52,6 +52,11 @@ public SendingAgent(ILogger logger, IMessageTracker messageLogger, ISender sende public virtual async ValueTask DisposeAsync() { + // Stop the circuit-breaker ping loop before tearing down the sender it pings through -- + // otherwise it keeps running against a disposed/unreachable destination indefinitely. + _circuitWatcher?.SafeDispose(); + _circuitWatcher = null; + if (_sender is IAsyncDisposable ad) { await ad.DisposeAsync().ConfigureAwait(false); From 1590bfdb95378c3df1ea16dd7415c287e1a0a25d Mon Sep 17 00:00:00 2001 From: Marko Lahma Date: Mon, 6 Jul 2026 21:21:20 +0300 Subject: [PATCH 2/2] address review feedback: swallow shutdown races, de-flake tests - pingUntilConnectedAsync now wraps the whole loop (not just the inner ping) in a try/catch: cancelling and disposing the linked CancellationTokenSource in Dispose() can surface as OperationCanceledException from the timer wait, or ObjectDisposedException if a token registration races the dispose. Both are expected shutdown outcomes, not faults. - Both new tests settle for one retry interval after Dispose()/DisposeAsync() before taking the "count so far" baseline, so an already-in-flight ping can't tick over during the assertion window and cause a spurious failure. --- .../Sending/SendingAgentDisposalTests.cs | 8 +++++ .../Transports/Sending/CircuitWatcher.cs | 32 ++++++++++++------- 2 files changed, 29 insertions(+), 11 deletions(-) diff --git a/src/Testing/CoreTests/Transports/Sending/SendingAgentDisposalTests.cs b/src/Testing/CoreTests/Transports/Sending/SendingAgentDisposalTests.cs index 50d7c91f0..e56040515 100644 --- a/src/Testing/CoreTests/Transports/Sending/SendingAgentDisposalTests.cs +++ b/src/Testing/CoreTests/Transports/Sending/SendingAgentDisposalTests.cs @@ -23,6 +23,11 @@ public async Task circuit_watcher_dispose_stops_the_ping_loop() watcher.Dispose(); + // Dispose() cancels the loop but does not wait for an already-in-flight tick to observe + // that cancellation, so settle first before taking the baseline -- otherwise a tick that + // was already running when Dispose() was called could tick over during the assertion + // window below and cause a spurious failure. + await Task.Delay(200.Milliseconds()); var countAtDispose = circuit.CallCount; await Task.Delay(200.Milliseconds()); @@ -55,6 +60,9 @@ public async Task disposing_a_sending_agent_stops_its_circuit_watcher() await agent.DisposeAsync(); + // See circuit_watcher_dispose_stops_the_ping_loop above: settle before taking the baseline + // so an already-in-flight ping can't tick over during the assertion window below. + await Task.Delay(200.Milliseconds()); var pingCountAtDispose = sender.PingCount; await Task.Delay(200.Milliseconds()); diff --git a/src/Wolverine/Transports/Sending/CircuitWatcher.cs b/src/Wolverine/Transports/Sending/CircuitWatcher.cs index bed38dfea..6cfd7b001 100644 --- a/src/Wolverine/Transports/Sending/CircuitWatcher.cs +++ b/src/Wolverine/Transports/Sending/CircuitWatcher.cs @@ -39,23 +39,33 @@ public void Dispose() private async Task pingUntilConnectedAsync() { - using var timer=new PeriodicTimer(_senderCircuit.RetryInterval); - while (await timer.WaitForNextTickAsync(_cancellation.Token)) + try { - try + using var timer = new PeriodicTimer(_senderCircuit.RetryInterval); + while (await timer.WaitForNextTickAsync(_cancellation.Token)) { - var pinged = await _senderCircuit.TryToResumeAsync(_cancellation.Token); + try + { + var pinged = await _senderCircuit.TryToResumeAsync(_cancellation.Token); - if (pinged) + if (pinged) + { + await _senderCircuit.ResumeAsync(_cancellation.Token); + return; + } + } + // ReSharper disable once EmptyGeneralCatchClause + catch (Exception) { - await _senderCircuit.ResumeAsync(_cancellation.Token); - return; } } - // ReSharper disable once EmptyGeneralCatchClause - catch (Exception) - { - } + } + catch (Exception) + { + // Expected on Dispose(): cancelling and disposing the linked CTS can surface as + // OperationCanceledException from the timer wait, or ObjectDisposedException if a + // token registration races the dispose. Either way, shutting down is correct -- + // don't let this show up as an unobserved/faulted background task. } } } \ No newline at end of file