GetCheckpoint can create a CheckpointCommitHandler that is never disposed

[alexeyzimarev]

Spotted while fixing the shutdown dispose race in #562 / #563, which addressed disposing the handler twice. This is the mirror image: creating one that nobody disposes.

Describe the bug

GetCheckpoint lazily creates the commit handler with ??= (EventSubscriptionWithCheckpoint.cs:112):

protected async Task<Checkpoint> GetCheckpoint(CancellationToken cancellationToken) {
    CheckpointCommitHandler ??= new(
        options.SubscriptionId,
        checkpointStore,
        TimeSpan.FromMilliseconds(options.CheckpointCommitDelayMs),
        options.CheckpointCommitBatchSize,
        LoggerFactory
    );
    ...

??= is a read followed by a write, not an atomic operation, and CheckpointCommitHandler is a plain auto-property. Two paths reach it concurrently, so a handler can be created that no disposal path will ever see:

• Create racing shutdown. Resubscribe disposes the handler and then calls Subscribe, which calls GetCheckpoint and creates a fresh one. If that creation lands after Finalize has already swapped the field to null, the new handler is orphaned.
• Create racing create. Two GetCheckpoint calls both observe null and both construct, so one instance is overwritten and lost.

An orphaned handler is not inert. Its BatchedChannelWorker starts a reader task in its constructor, so it keeps a live task and an undisposed CancellationTokenSource for the rest of the process, and it can still write checkpoints after the subscription is supposed to have stopped.

Expected behaviour

Either exactly one handler exists per subscription run and shutdown always disposes the one that exists, or a handler created after shutdown began is not created at all.

What makes the create-vs-create case reachable

Dropped has no re-entry guard (EventSubscription.cs:207) — it checks IsRunning but not IsDropped, and each call schedules its own Task.Run(... Resubscribe ...). NackOnAsyncWorker calls Dropped once per failing message (EventSubscriptionWithCheckpoint.cs:80), and with AsyncHandlingFilter's concurrency above 1 several messages can fail at once. So a burst of failures produces several concurrent resubscribe loops, each disposing and recreating the handler.

That re-entry may deserve its own issue — concurrent resubscribe loops also mean concurrent calls into the concrete Subscribe, and e.g. AllStreamSubscription assigns _subscription unconditionally, so one of the two underlying subscriptions would be leaked rather than disposed. I've left it here rather than splitting it, since it's the mechanism that makes this race concurrent.

Notes

#563 makes Dropped bail out when the stopping token is already cancelled, which narrows the create-racing-shutdown window but doesn't close it: a resubscribe loop already past that guard can still be inside Subscribe when Unsubscribe runs.

[linear]

EVE-240