fix: CircuitWatcher.Dispose() does not stop the ping-until-reconnected loop

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<bool> 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()
     {

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<IMessageTracker>(),
+            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<bool> 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<bool> PingAsync()
+    {
+        Interlocked.Increment(ref _pingCount);
+        return Task.FromResult(false);
+    }
+
+    public ValueTask SendAsync(Envelope envelope) => throw new Exception("Nope!");
+}

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.
         }
     }
 }

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);