From 5571c7723cda793f31d0d1a896e91059a66ac9c4 Mon Sep 17 00:00:00 2001 From: Alexey Zimarev Date: Mon, 3 Aug 2026 12:34:32 +0200 Subject: [PATCH 1/2] fix(subscriptions): harden the worker dispose failure path Review follow-ups on the dispose gate. Return the shared task to every caller, including the first. The winner used to await StopWorker() directly while the TaskCompletionSource was faulted separately, so a failed shutdown with no second caller left a faulted task nobody observed -- resurfacing later through TaskScheduler.UnobservedTaskException as exactly the kind of spurious shutdown warning this branch removes. Cancel the CTS, drain the readers and dispose in a finally. A throw from the graceful stop used to skip all of it, leaking the ten-second timer Stop arms via CancelAfter and leaving the readers holding a live token. That gap predates the dispose gate, but the gate makes it permanent since _disposing never resets. Both cleanup awaits suppress, so disposal can't be skipped by a cancellation callback throwing. Also re-check the stopping token inside the resubscribe task. Unsubscribe can cancel between the check in Dropped and the task being scheduled. It doesn't close the race -- Resubscribe disposes the commit handler before it looks at the token -- but it keeps the common case out of it. Co-Authored-By: Claude Opus 5 --- .../Channels/ChannelWorkerBase.cs | 42 ++++++++++--------- .../EventSubscription.cs | 5 +++ 2 files changed, 28 insertions(+), 19 deletions(-) diff --git a/src/Core/src/Eventuous.Subscriptions/Channels/ChannelWorkerBase.cs b/src/Core/src/Eventuous.Subscriptions/Channels/ChannelWorkerBase.cs index 3439e6e2..9e46dba8 100644 --- a/src/Core/src/Eventuous.Subscriptions/Channels/ChannelWorkerBase.cs +++ b/src/Core/src/Eventuous.Subscriptions/Channels/ChannelWorkerBase.cs @@ -31,29 +31,33 @@ 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); - _disposed.TrySetResult(); - } catch (Exception e) { - // Don't let a waiter see a clean shutdown that didn't happen. - _disposed.TrySetException(e); + 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); + } - throw; - } + _disposed.TrySetResult(); + } catch (Exception e) { _disposed.TrySetException(e); } } } 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); From 6b410b678cd26b65b10775a23027f3d8f4793d08 Mon Sep 17 00:00:00 2001 From: Alexey Zimarev Date: Mon, 3 Aug 2026 13:38:01 +0200 Subject: [PATCH 2/2] docs(subscriptions): record why the dispose catch is broad The catch transfers the outcome onto the shared completion, which is the only thing that releases waiters. Narrowing it would strand every caller of DisposeAsync, so the breadth is deliberate rather than sloppy. Co-Authored-By: Claude Opus 5 --- .../Eventuous.Subscriptions/Channels/ChannelWorkerBase.cs | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/src/Core/src/Eventuous.Subscriptions/Channels/ChannelWorkerBase.cs b/src/Core/src/Eventuous.Subscriptions/Channels/ChannelWorkerBase.cs index 9e46dba8..67d10b6f 100644 --- a/src/Core/src/Eventuous.Subscriptions/Channels/ChannelWorkerBase.cs +++ b/src/Core/src/Eventuous.Subscriptions/Channels/ChannelWorkerBase.cs @@ -58,6 +58,10 @@ async Task StopWorker() { } _disposed.TrySetResult(); - } catch (Exception e) { _disposed.TrySetException(e); } + } catch (Exception e) { + // 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); + } } }