diff --git a/src/Core/src/Eventuous.Subscriptions/Channels/ChannelWorkerBase.cs b/src/Core/src/Eventuous.Subscriptions/Channels/ChannelWorkerBase.cs index 3439e6e2..67d10b6f 100644 --- a/src/Core/src/Eventuous.Subscriptions/Channels/ChannelWorkerBase.cs +++ b/src/Core/src/Eventuous.Subscriptions/Channels/ChannelWorkerBase.cs @@ -31,29 +31,37 @@ protected ChannelWorkerBase(Channel channel, Func 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 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 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. /// - 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; } } } diff --git a/src/Core/src/Eventuous.Subscriptions/EventSubscription.cs b/src/Core/src/Eventuous.Subscriptions/EventSubscription.cs index a3b67685..75b320ba 100644 --- a/src/Core/src/Eventuous.Subscriptions/EventSubscription.cs +++ b/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);