Fix resubscribe race - #571
Conversation
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>
Test Results 46 files + 24 46 suites +24 11m 52s ⏱️ -39s Results for commit 6aa1cac. ± Comparison against base commit 3cb68c2. This pull request removes 8 and adds 84 tests. Note that renamed tests count towards both. |
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
1 participant
No description provided.