fix(subscriptions): stop double-dispose crashing host shutdown

src/Core/src/Eventuous.Subscriptions/Channels/ChannelWorkerBase.cs

@@ -6,9 +6,12 @@
 namespace Eventuous.Subscriptions.Channels;
 
 abstract class ChannelWorkerBase<T> : IAsyncDisposable {
-    readonly CancellationTokenSource _cts = new();
+    readonly CancellationTokenSource _cts      = new();
+    readonly TaskCompletionSource    _disposed = new(TaskCreationOptions.RunContinuationsAsynchronously);
     readonly Task[]                  _readerTasks;
 
+    int _disposing;
+
     public Func<CancellationToken, ValueTask>? OnDispose { get; set; }
 
     public ValueTask Write(T element, CancellationToken cancellationToken)
@@ -24,16 +27,33 @@ protected ChannelWorkerBase(Channel<T> channel, Func<CancellationToken, Task> pr
         _readerTasks = Enumerable.Range(0, concurrencyLevel).Select(_ => Task.Run(() => processor(_cts.Token))).ToArray();
     }
 
-    public async ValueTask DisposeAsync() {
-        _stopping = true;
-        await _channel.Stop(_cts, _readerTasks, OnDispose).NoContext();
+    /// <summary>
+    /// Idempotent. The commit handler worker is disposed by both the resubscribe and the shutdown
+    /// paths, which can run concurrently, so a second call is expected rather than a programming
+    /// error. It must not re-enter the shutdown: by then the CTS is disposed, and cancelling it
+    /// again throws <see cref="ObjectDisposedException"/> out of host shutdown. The second caller
+    /// awaits the first call's shutdown, so it can't return before the final checkpoint flush.
+    /// </summary>
+    public ValueTask DisposeAsync() => Interlocked.Exchange(ref _disposing, 1) == 0 ? new(StopWorker()) : new(_disposed.Task);
+
+    async Task StopWorker() {
+        try {
+            _stopping = true;
+            await _channel.Stop(_cts, _readerTasks, OnDispose).NoContext();
 #if NET8_0_OR_GREATER
-        await _cts.CancelAsync().NoContext();
+            await _cts.CancelAsync().NoContext();
 #else
-        _cts.Cancel();
+            _cts.Cancel();
 #endif
-        await Task.WhenAll(_readerTasks).NoThrow();
-        _cts.Dispose();
-        GC.SuppressFinalize(this);
+            await Task.WhenAll(_readerTasks).NoThrow();
+            _cts.Dispose();
+            GC.SuppressFinalize(this);
+            _disposed.TrySetResult();
+        } catch (Exception e) {
+            // Don't let a waiter see a clean shutdown that didn't happen.
+            _disposed.TrySetException(e);
+
+            throw;
+        }
     }
 }

src/Core/src/Eventuous.Subscriptions/EventSubscription.cs

@@ -212,12 +212,27 @@ protected void Dropped(DropReason reason, Exception? exception) {
         IsDropped = true;
         _onDropped?.Invoke(Options.SubscriptionId, reason, exception);
 
+        // Read the token here rather than inside the background task below: Unsubscribe disposes
+        // Stopping, and reading .Token from a disposed source throws, which the task would surface as
+        // a spurious warning plus an unobserved exception. A token captured before the dispose stays
+        // usable afterwards, so hoisting the read is what makes the resubscribe safe. Losing the race
+        // outright means shutdown already got there, and there's nothing left to resubscribe to.
+        CancellationToken stopping;
+
+        try { stopping = Stopping.Token; } catch (ObjectDisposedException) { return; }
+
+        // Same reasoning for a token that's merely cancelled, which is the state Unsubscribe leaves it
+        // in for most of shutdown. Resubscribing from there can't succeed, and it isn't free: the
+        // checkpoint subscription's Resubscribe disposes the commit handler before it ever looks at the
+        // token, putting a second disposer in the race with Finalize.
+        if (stopping.IsCancellationRequested) return;
+
         Task.Run(
             async () => {
                 var delay = reason == DropReason.Stopped ? TimeSpan.FromSeconds(10) : TimeSpan.FromSeconds(2);
                 Log.SubscriptionWillResubscribe(delay);
 
-                try { await Resubscribe(delay, Stopping.Token).NoContext(); } catch (Exception e) {
+                try { await Resubscribe(delay, stopping).NoContext(); } catch (Exception e) {
                     Log.WarnLog?.Log(e.Message);
 
                     throw;

src/Core/src/Eventuous.Subscriptions/EventSubscriptionWithCheckpoint.cs

@@ -142,7 +142,10 @@ protected override async Task Resubscribe(TimeSpan delay, CancellationToken canc
     protected override async ValueTask Finalize(CancellationToken cancellationToken) => await DisposeCommitHandler();
 
     async ValueTask DisposeCommitHandler() {
-        // Swap to null first so the concurrent path (Resubscribe vs Finalize) sees null.
+        // Swap to null first so the concurrent path (Resubscribe vs Finalize) sees null. The read and
+        // the write aren't atomic, so both paths can still come away with the same handler — that stays
+        // safe because the commit worker's dispose is idempotent, and the second caller awaits the first
+        // one's shutdown rather than re-entering it and cancelling an already-disposed CTS (AI-1699).
         var handler = CheckpointCommitHandler;
         CheckpointCommitHandler = null;
 

src/Core/test/Eventuous.Tests.Subscriptions/CheckpointCommitHandlerDisposeTests.cs

@@ -0,0 +1,58 @@
+using Eventuous.Subscriptions.Checkpoints;
+using Shouldly;
+
+namespace Eventuous.Tests.Subscriptions;
+
+/// <summary>
+/// Guards against AI-1699, where host shutdown failed with
+/// <c>ObjectDisposedException: The CancellationTokenSource has been disposed</c> thrown from
+/// <c>ChannelWorkerBase.DisposeAsync</c> via <see cref="CheckpointCommitHandler.DisposeAsync"/>.
+/// The commit handler can be disposed twice — <c>Resubscribe</c> and <c>Finalize</c> both call
+/// <c>DisposeCommitHandler</c>, and the second call must be a no-op rather than a crash.
+/// </summary>
+public class CheckpointCommitHandlerDisposeTests {
+    [Test]
+    public async Task Dispose_is_idempotent() {
+        var store   = new NoOpCheckpointStore();
+        var handler = new CheckpointCommitHandler("test-dispose-twice", store, TimeSpan.FromMilliseconds(10));
+
+        await handler.DisposeAsync();
+
+        await Should.NotThrowAsync(async () => await handler.DisposeAsync());
+    }
+
+    /// <summary>
+    /// The loser of the dispose race must not report the worker stopped before it actually is —
+    /// host shutdown continues on that return, and the final checkpoint flush is still in flight.
+    /// </summary>
+    [Test]
+    public async Task Second_dispose_awaits_the_first(CancellationToken ct) {
+        var storeEntered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        var release      = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+
+        var handler = new CheckpointCommitHandler(
+            "test-dispose-awaits",
+            async (checkpoint, _, _) => {
+                storeEntered.TrySetResult();
+                await release.Task;
+
+                return checkpoint;
+            },
+            TimeSpan.FromMilliseconds(1)
+        );
+
+        // Park the commit worker inside the checkpoint store so the first dispose can't complete.
+        await handler.Commit(new CommitPosition(0, 0, DateTime.UtcNow), ct);
+        await storeEntered.Task.WaitAsync(TimeSpan.FromSeconds(10), ct);
+
+        var first  = handler.DisposeAsync();
+        var second = handler.DisposeAsync();
+
+        await Task.Delay(200, ct);
+        second.IsCompleted.ShouldBeFalse("the second dispose must await the shutdown in flight, not skip past it");
+
+        release.SetResult();
+        await first;
+        await second;
+    }
+}

src/Core/test/Eventuous.Tests.Subscriptions/SubscriptionShutdownTests.cs

@@ -0,0 +1,123 @@
+using System.Collections.Concurrent;
+using Eventuous.Subscriptions;
+using Eventuous.Subscriptions.Filters;
+using Microsoft.Extensions.Logging;
+using Shouldly;
+
+namespace Eventuous.Tests.Subscriptions;
+
+/// <summary>
+/// A dropped subscription schedules the resubscribe on a background task, and that task used to read
+/// <c>Stopping.Token</c> long after <c>Unsubscribe</c> got to the source. A drop that races shutdown is
+/// benign — there is nothing left to resubscribe to — but it used to cost a spurious warning, an
+/// unobserved exception, and a commit-handler dispose racing the one in <c>Finalize</c> (AI-1699).
+/// </summary>
+public class SubscriptionShutdownTests {
+    /// <summary>
+    /// The KurrentDB subscriptions cancel <c>Stopping</c> at the top of their <c>Unsubscribe</c>, so a
+    /// drop during shutdown normally finds the token cancelled. Resubscribing from there is pure waste:
+    /// <c>EventSubscriptionWithCheckpoint.Resubscribe</c> disposes the commit handler before it ever
+    /// looks at the token, which is what put a second disposer in the race with <c>Finalize</c>.
+    /// </summary>
+    [Test]
+    public async Task Drop_after_shutdown_started_does_not_resubscribe(CancellationToken ct) {
+        var subscription = new TestSubscription(new() { SubscriptionId = "test-drop-when-cancelled" }, new ConsumePipe(), new CapturingLoggerFactory());
+
+        await subscription.Subscribe(_ => { }, (_, _, _) => { }, ct);
+
+        subscription.DropAfterCancellingStopping();
+
+        var resubscribed = await subscription.WaitForResubscribe(TimeSpan.FromSeconds(2));
+
+        resubscribed.ShouldBeFalse("a subscription that is already stopping has nothing to resubscribe to");
+    }
+
+    [Test]
+    public async Task Drop_racing_unsubscribe_does_not_report_a_disposed_cts(CancellationToken ct) {
+        var logs = new CapturingLoggerFactory();
+
+        var subscription = new TestSubscription(new() { SubscriptionId = "test-drop-race" }, new ConsumePipe(), logs);
+
+        await subscription.Subscribe(_ => { }, (_, _, _) => { }, ct);
+
+        // Reproduce the losing side of the race: Unsubscribe has already disposed Stopping while the
+        // subscription still believes it's running, which is exactly what lets Dropped reach the token.
+        subscription.DropAfterDisposingStopping();
+
+        var reported = await logs.WaitForWarning("CancellationTokenSource has been disposed", TimeSpan.FromSeconds(2));
+
+        reported.ShouldBeFalse("dropping while Unsubscribe disposes Stopping is a benign shutdown race, not an error");
+    }
+
+    record TestSubscriptionOptions : SubscriptionOptions;
+
+    /// <summary>
+    /// A subscription that does nothing but expose the drop path. <c>Stopping</c> is protected, so both
+    /// shutdown states can be reproduced without any test-only hooks in the production class.
+    /// </summary>
+    class TestSubscription(TestSubscriptionOptions options, ConsumePipe pipe, ILoggerFactory loggerFactory)
+        : EventSubscription<TestSubscriptionOptions>(options, pipe, loggerFactory, null) {
+        readonly TaskCompletionSource _resubscribed = new(TaskCreationOptions.RunContinuationsAsynchronously);
+
+        public void DropAfterCancellingStopping() {
+            Stopping.Cancel(false);
+            Drop();
+        }
+
+        public void DropAfterDisposingStopping() {
+            Stopping.Dispose();
+            Drop();
+        }
+
+        public async Task<bool> WaitForResubscribe(TimeSpan timeout)
+            => await Task.WhenAny(_resubscribed.Task, Task.Delay(timeout)) == _resubscribed.Task;
+
+        protected override Task Resubscribe(TimeSpan delay, CancellationToken cancellationToken) {
+            _resubscribed.TrySetResult();
+
+            return Task.CompletedTask;
+        }
+
+        protected override ValueTask Subscribe(CancellationToken cancellationToken) => default;
+
+        protected override ValueTask Unsubscribe(CancellationToken cancellationToken) => default;
+
+        void Drop() => Dropped(DropReason.SubscriptionError, new InvalidOperationException("Simulated drop during shutdown"));
+    }
+
+    sealed class CapturingLoggerFactory : ILoggerFactory {
+        readonly ConcurrentQueue<string> _warnings = [];
+
+        /// <summary>
+        /// Polls rather than waiting out the full timeout, so the failing case reports in milliseconds.
+        /// The resubscribe runs on a fire-and-forget task, so there is nothing to await on.
+        /// </summary>
+        public async Task<bool> WaitForWarning(string contains, TimeSpan timeout) {
+            var deadline = DateTime.UtcNow + timeout;
+
+            while (DateTime.UtcNow < deadline) {
+                if (_warnings.Any(w => w.Contains(contains))) return true;
+
+                await Task.Delay(20);
+            }
+
+            return false;
+        }
+
+        public ILogger CreateLogger(string categoryName) => new CapturingLogger(_warnings);
+
+        public void AddProvider(ILoggerProvider provider) { }
+
+        public void Dispose() { }
+
+        sealed class CapturingLogger(ConcurrentQueue<string> warnings) : ILogger {
+            public IDisposable? BeginScope<TState>(TState state) where TState : notnull => null;
+
+            public bool IsEnabled(LogLevel logLevel) => true;
+
+            public void Log<TState>(LogLevel logLevel, EventId eventId, TState state, Exception? exception, Func<TState, Exception?, string> formatter) {
+                if (logLevel >= LogLevel.Warning) warnings.Enqueue(formatter(state, exception));
+            }
+        }
+    }
+}