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..e56040515 --- /dev/null +++ b/src/Testing/CoreTests/Transports/Sending/SendingAgentDisposalTests.cs @@ -0,0 +1,107 @@ +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(); + + // 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()); + + // 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(); + + // 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()); + + // 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..6cfd7b001 100644 --- a/src/Wolverine/Transports/Sending/CircuitWatcher.cs +++ b/src/Wolverine/Transports/Sending/CircuitWatcher.cs @@ -15,42 +15,57 @@ 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)) + try { - try + using var timer = new PeriodicTimer(_senderCircuit.RetryInterval); + while (await timer.WaitForNextTickAsync(_cancellation.Token)) { - var pinged = await _senderCircuit.TryToResumeAsync(_cancellation); + 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); - 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 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);