fix(subscriptions): harden the worker dispose failure path - #563
Merged
Conversation
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 <noreply@anthropic.com>
Contributor
PR Summary by QodoHarden channel worker dispose: always share shutdown task and guarantee CTS cleanup
AI Description
Diagram
High-Level Assessment
Files changed (2)
|
There was a problem hiding this comment.
Pull request overview
This PR hardens the shutdown/dispose failure path in subscription infrastructure to prevent unobserved task exceptions during teardown and to ensure worker cancellation/cleanup always runs, even when the graceful stop path throws.
Changes:
- Make
ChannelWorkerBase.DisposeAsync()consistently return the shared_disposed.Taskso shutdown failures are always observed by awaiters. - Move worker cleanup (cancel, reader drain, CTS dispose) into a
finallyblock to avoid leaking timers/tokens whenChannelExtensions.Stopthrows. - Add a second
Stoppingtoken cancellation check inside the resubscribe background task to reduce a race withUnsubscribe.
Reviewed changes
Copilot reviewed 2 out of 2 changed files in this pull request and generated 1 comment.
| File | Description |
|---|---|
src/Core/src/Eventuous.Subscriptions/EventSubscription.cs |
Adds a second cancellation check inside the resubscribe task to reduce a scheduling race with shutdown. |
src/Core/src/Eventuous.Subscriptions/Channels/ChannelWorkerBase.cs |
Refactors disposal to always return a shared completion task and ensures CTS/reader cleanup runs in finally. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
+51
to
+55
| // 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(); |
Contributor
Author
There was a problem hiding this comment.
CancellationTokenSource.CancelAsync() returns Task, not ValueTask — it was added in .NET 8 as public Task CancelAsync(). Verified by reflection on the runtime this builds against:
CancelAsync returns: System.Threading.Tasks.Task
Is Task: True Is ValueTask: False
So NoThrow(this Task) binds without a conversion, and there's nothing to overload.
The compile claim is also refuted by this PR's own checks: Build and test core (8.0), (9.0) and (10.0) are all green on the reviewed commit, and dotnet build Eventuous.slnx is clean locally across all three target frameworks. No change made.
![github-code-quality[bot]](https://avatars.githubusercontent.com/u/9919?s=60&v=4){kind=small}
Contributor
Code Review by Qodo🐞 Bugs (0) 📘 Rule violations (0) 📎 Requirement gaps (0)
Great, no issues found!Qodo reviewed your code and found no material issues that require reviewTo customize comments, go to the Qodo configuration screen, or learn more in the docs. |
Test Results 46 files + 24 46 suites +24 12m 54s ⏱️ -50s Results for commit 6b410b6. ± Comparison against base commit 90257a1. This pull request removes 5 and adds 9 tests. Note that renamed tests count towards both.♻️ This comment has been updated with latest results. |
This was referenced Aug 3, 2026
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 <noreply@anthropic.com>
alexeyzimarev
added a commit
that referenced
this pull request
Aug 3, 2026
Conflict in ChannelWorkerBase: kept the idempotent dispose gate from dev (#562/#563) as-is, on top of the cleanup's collection-expression reader initialisation. EventSubscription merged clean to dev's version, and the checkpoint subscription keeps the cleanup's NoContext on the commit handler dispose. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Follow-up to #562, which merged before these review findings were addressed. Two bot reviewers flagged the dispose gate; both were right, though one's stated mechanism turned out to be wrong.
Every caller gets the shared task
The winner awaited
StopWorker()directly while theTaskCompletionSourcewas faulted separately. A failed shutdown with no second caller therefore left a faulted task nobody observed, which resurfaces throughTaskScheduler.UnobservedTaskExceptionas exactly the kind of spurious shutdown warning #562 set out to remove.DisposeAsyncnow returns_disposed.Taskto everyone, so the failure is observed exactly once and there is only ever one task.Cleanup moved into a
finallyA throw from the graceful stop skipped the cancel, the reader drain and
_cts.Dispose(), leaking the ten-second timerChannelExtensions.Stoparms viaCancelAfterand leaving the readers holding a live token. That gap predates #562, but the dispose gate makes it permanent, since_disposingnever resets. Both cleanup awaits suppress, so a throwing cancellation callback can't skip disposal either.Worth recording that the reviewer's premise — "
channel.Stoppropagates reader-task faults" — is mostly wrong.Stopfilters to!r.IsCompleted, and a faulted task is completed, so a reader that has already faulted is silently skipped. I confirmed this empirically: driving the commit worker's processor into a throw and then disposing produces no exception at all. The reachable window is narrower than claimed — a reader faulting whileStopawaits it, orCancelAsyncthrowing — but it is not empty, and the invariant "dispose always releases the CTS" is worth holding unconditionally.Re-check the stopping token inside the resubscribe task
Unsubscribecan cancel between the check inDroppedand the task being scheduled. This does not close the race —EventSubscriptionWithCheckpoint.Resubscribedisposes the commit handler before it looks at the token, which is why the idempotent dispose in #562 is the actual fix — but it keeps the common case out of it.Tests
No new tests. The two
ChannelWorkerBasechanges affect an error path whose consequences (a released timer, an unobserved task) have no seam to assert on: reaching the throw needs a reader to fault mid-Stop, and observing the fix needs either private fields orTaskScheduler.UnobservedTaskExceptionplus forced GC — a global, GC-timed assertion that would itself be the flaky test this work is meant to eliminate. I verified the path with a throwaway probe instead, which is what turned up theIsCompletedfiltering above. The token re-check is inherently racy and equally untestable; the deterministic pre-check it backs up is already covered byDrop_after_shutdown_started_does_not_resubscribe.Verification
Eventuous.Tests.Subscriptions38/38,Eventuous.Tests26/26,Eventuous.Tests.Application21/21 (net10.0)Eventuous.Tests.KurrentDB60/60, including the drop/resubscribe testdotnet build Eventuous.slnxclean across net8.0/net9.0/net10.0; tests ran locally on net10.0 only (net8/net9 runtimes absent on the dev machine)🤖 Generated with Claude Code