fix(subscriptions): harden the worker dispose failure path

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

@@ -31,29 +31,37 @@ protected ChannelWorkerBase(Channel<T> channel, Func<CancellationToken, Task> pr
     /// 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.
+    /// again throws <see cref="ObjectDisposedException"/> out of host shutdown. Every caller awaits
+    /// the same task, so none of them returns before the final checkpoint flush, and a shutdown that
+    /// failed is reported to whoever awaits it instead of being left on a task nobody observes.
     /// </summary>
-    public ValueTask DisposeAsync() => Interlocked.Exchange(ref _disposing, 1) == 0 ? new(StopWorker()) : new(_disposed.Task);
+    public ValueTask DisposeAsync() {
+        if (Interlocked.Exchange(ref _disposing, 1) == 0) _ = StopWorker();
+
+        return 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();
-#else
-            _cts.Cancel();
-#endif
-            await Task.WhenAll(_readerTasks).NoThrow();
-            _cts.Dispose();
-            GC.SuppressFinalize(this);
+            try {
+                _stopping = true;
+                await _channel.Stop(_cts, _readerTasks, OnDispose).NoContext();
+            }
+            finally {
+                // Release the readers even when the graceful stop above failed: they hold _cts.Token,
+                // and Stop armed a ten-second timer on it, so both outlive the worker unless cancelled
+                // here. Cancelling runs their callbacks, which is why this can't be allowed to throw.
+                await _cts.CancelAsync().NoThrow();
+                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.
+            // Broad on purpose. DisposeAsync hands _disposed.Task to every caller, so completing it is the
+            // only thing that ever releases them; an exception escaping here would strand all of them.
             _disposed.TrySetException(e);
-
-            throw;
         }
     }
 }

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

@@ -229,6 +229,11 @@ protected void Dropped(DropReason reason, Exception? exception) {
 
         Task.Run(
             async () => {
+                // Check again: Unsubscribe may have cancelled between the check above and this task
+                // getting scheduled. It doesn't close the race — Resubscribe still disposes the commit
+                // handler before it looks at the token — but it keeps the common case out of it.
+                if (stopping.IsCancellationRequested) return;
+
                 var delay = reason == DropReason.Stopped ? TimeSpan.FromSeconds(10) : TimeSpan.FromSeconds(2);
                 Log.SubscriptionWillResubscribe(delay);