Fix resubscribe race - #571
Conversation
Test Results 44 files + 21 44 suites +21 12m 40s ⏱️ - 1m 4s Results for commit 89c5c68. ± Comparison against base commit 40c387c. This pull request removes 29 and adds 93 tests. Note that renamed tests count towards both.♻️ This comment has been updated with latest results. |
PR Summary by QodoFix subscription resubscribe race via per-run supervisor loop
AI Description
Diagram
High-Level Assessment
Files changed (58)
|
Code Review by Qodo
1. Delay upper bound unchecked
|
![qodo-free-for-open-source-projects[bot]](https://avatars.githubusercontent.com/in/1068000?s=60&v=4){kind=small}
| // InfiniteTimeSpan is exempt: both Task.Delay and CancellationTokenSource accept it as "never". | ||
| if (retryDelay < TimeSpan.Zero && retryDelay != Timeout.InfiniteTimeSpan) { | ||
| log.SubscriptionRetryDelayInvalid(retryDelay, SubscriptionOptions.DefaultRetryDelay); | ||
| retryDelay = SubscriptionOptions.DefaultRetryDelay; |
There was a problem hiding this comment.
1. Delay upper bound unchecked 🐞 Bug ☼ Reliability
SupervisorSettings.From only guards negative RetryDelay/TeardownTimeout; overly-large finite values can throw in Task.Delay(settings.RetryDelay) or new CancellationTokenSource(settings.TeardownTimeout), which terminates RunSubscriptionLoop and leaves the subscription down. Because the supervisor’s outer catch only logs/report-drops and then exits (no retry), this becomes a permanent outage until restart/recreate.
Agent Prompt
## Issue description
`SupervisorSettings.From` validates only negative `RetryDelay`/`TeardownTimeout` (excluding `Timeout.InfiniteTimeSpan`), but does not validate **overly-large finite** values. Those values are later passed to `Task.Delay(settings.RetryDelay, ...)` and `new CancellationTokenSource(settings.TeardownTimeout)`, which can throw `ArgumentOutOfRangeException` for values above the runtime-supported finite maximum (milliseconds > `int.MaxValue`). The exception is caught by the supervisor’s outer catch, which logs and exits the loop, leaving the subscription down permanently.
## Issue Context
- `SubscriptionOptions` exposes both properties as user-configurable `TimeSpan`s.
- `RunSubscriptionLoop` uses them directly.
## Fix Focus Areas
- src/Core/src/Eventuous.Subscriptions/EventSubscription.cs[357-374]
- src/Core/src/Eventuous.Subscriptions/EventSubscription.cs[154-170]
- src/Core/src/Eventuous.Subscriptions/EventSubscription.cs[207-210]
## Suggested change
- Extend validation in `SupervisorSettings.From` to also reject/clamp finite values greater than the maximum supported by `Task.Delay`/`CancellationTokenSource`.
- Keep allowing `Timeout.InfiniteTimeSpan`.
- Use a max like `TimeSpan.FromMilliseconds(int.MaxValue)`.
- If configured value is invalid (too large), log (reuse existing `SubscriptionRetryDelayInvalid` / `SubscriptionTeardownTimeoutInvalid` or add dedicated log methods) and fall back to `SubscriptionOptions.DefaultRetryDelay` / `DefaultTeardownTimeout`.
## Acceptance criteria
- No `ArgumentOutOfRangeException` can be thrown by `Task.Delay(settings.RetryDelay, ...)` or `new CancellationTokenSource(settings.TeardownTimeout)` due to user configuration.
- Invalid values are surfaced once per Subscribe via log warning and then normalized.
ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools
A seconds-long store blip wedged twenty subscriptions for seven hours, and nothing in the suite could have caught it. Twelve of these fail on every run and the thirteenth whenever the race lands, so the fix has something to prove and the next regression has somewhere to land. The provider suites carry the restart contract itself: a fake transport can show the framework calls teardown once, but only real infrastructure can show a given broker survives being restarted. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Subscribe returns while the work carries on, so a transport that is naturally a loop had to spawn one, and that loop's failure could not propagate — it went sideways through Dropped(), callable from any thread at any time. The state machine, the gate, the run generation and the drop cycle all existed to collapse those writers back into one. Moving the loop's lifetime from the drop to the subscription removes the need for them. A transport now says only how to connect. Whatever it acquires is registered on the run as it is acquired and released in reverse when the run stops, so a failure or an acknowledgement arriving late names the run that produced it instead of reaching whichever run is current. Pumps belong to transports, which report their own death. Teardown is one policy: a graceful attempt on TeardownTimeout, then whatever is left is started and abandoned rather than skipped; Unsubscribe's token bounds only the caller's wait, never the stopping. Also fixes: a supervisor dying after connecting reported no drop, leaving health green; handlers cancelled by teardown were acknowledged, advancing the checkpoint past everything in flight; RabbitMQ delivered contexts with a default token, so the stopping guard never matched, and a nack on a closed channel killed a filter reader for good; $all disposed its subscription twice and released without honouring the teardown budget; a faulted channel reader skipped the final forced checkpoint commit; test fixtures abandoned their containers. Deletes SubscriptionLifecycle, TaskRunner, CheckpointRun, ChannelFullException, Dropped, Resubscribe, Stopping, Generation, ResetSequence, IsDropped, MonitorSubscriberTask and DropReason.Stopped. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
6aa1cac to
fe9c254
Compare
Settings validation caught negatives but not values above what Task.Delay and CancellationTokenSource accept, which is 4294967294ms on every framework we target. A RetryDelay past it throws where the supervisor waits, and a TeardownTimeout past it throws in the graceful stop that runs on the first drop and on a failed first connect — either way the loop exits through its outer catch and the subscription stays down until the host restarts. Both now fall back to the default with the same warning a negative gets. The upper bound belongs to the runtime, not to us, so a test pins it on each target framework: what the validator accepts, Task.Delay and CancellationTokenSource accept too. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
- Recheck disposal after publishing the session in Subscribe, so a subscribe racing DisposeAsync is refused or stopped instead of leaving a live supervisor delivering into a disposed pipe, with a regression test - Join the Pub/Sub pump in a finally, so a StopAsync that throws on a forced hard stop cannot let the replacement client start alongside the old pump - Add missing .NoContext() on CreatePersistentSubscription - Document the Unsubscribe-timeout boundary of flush-then-read in the design doc Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
…away With ThrowOnError, a failing handler threw twice over in RabbitMQ and was heard neither time. From the delivery callback the throw reached the client's consumer dispatcher, which routes it to CallbackException — an event nothing subscribes to — and took the next delivery. From the nack it reached the handling filter's channel worker, where the only thing holding the reader task is a Task.WhenAll nobody awaits until dispose: the reader died, the subscription kept its connection and its healthy status, and consumed nothing until the host restarted. Either way no drop was reported and the supervisor stayed parked on run.Ended. RabbitMQ now tells the run, so the supervisor replaces the connection and the broker redelivers what was left unacked. The exception travels as the drop reason, which is logged and passed to OnDropped, where health is wired — more visible than the throw ever was. Its failure handler runs whatever ThrowOnError says, because deciding the delivery is what it exists for and leaving that to the channel close would requeue the message no matter what a custom handler was configured to do with it; it runs while the channel is still open, and the run ends in a finally, since a failure handler that throws would otherwise skip it. Failing the run being how a transport ends a run, the handling filter has no business propagating a nack's exception either: it logs it and reads the next message, so no transport can take the shared reader down. Its test is synthetic on purpose — after the RabbitMQ fix nothing in the repo throws from a nack, which is exactly why the guard needs one. Also drops the receivedMessage context item, which nothing has read since Ack and Nack started capturing the delivery, and which cost a dictionary allocation per message; and disposes the Pub/Sub client the run created rather than leaving its release to whatever the stop path happens to do internally. Verified against a broker and the Pub/Sub emulator, in both directions: reverting the run failure, or the finally around it, loses the event the handler failed on. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
The disposal recheck, the failed-first-connect catch, and the supervisor's finally all retired the session with the same unpublish, wake, release steps; RetireSession keeps that lifecycle invariant in one place. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2 participants
No description provided.