MongoCheckpointStore leaks a batching pipeline on every GetLastCheckpoint call #616

MongoCheckpointStore.GetLastCheckpoint is not a pure read. Every call builds a new Subject with its own Rx batching pipeline and overwrites the previous one for that checkpoint ID, without completing or disposing it:

public async ValueTask<EventuousCheckpoint> GetLastCheckpoint(string checkpointId, CancellationToken cancellationToken = default) {
    // ... read the stored checkpoint ...
    _subjects[checkpointId] = GetSubject();   // replaces whatever was there

    return checkpoint;
}

readonly Dictionary<string, Subject<EventuousCheckpoint>> _subjects = new();

Subject<EventuousCheckpoint> GetSubject() {
    var subject = new Subject<EventuousCheckpoint>();
    // ... Buffer(interval, size) by default ...
    observable
        .Where(x => x.Count > 0)
        .Select(x => Observable.FromAsync(ct => StoreInternal(x.Last(), false, ct)))
        .Concat()
        .Subscribe();                          // IDisposable discarded, no onError

    return subject;
}

MongoCheckpointStore.cs#L32-L66

GetLastCheckpoint is called once per subscription run (EventSubscriptionWithCheckpoint.GetCheckpoint), so this happens on every resubscribe, not only at startup.

Found by reading the code while reviewing #603. None of it is reproduced with a failing test yet.

Consequences

1. One leaked pipeline per resubscribe. The old subject is dropped from the dictionary, but its pipeline is never completed and the IDisposable returned by Subscribe() is thrown away. With the default options (BatchSize = 1, BatchIntervalSec = 5) the pipeline is Buffer(TimeSpan, int), which runs on a timer, so each abandoned pipeline stays alive and wakes every 5 seconds for the rest of the process lifetime. A subscription that drops and recovers N times carries N orphaned timers.

2. Writes from the old and the new pipeline are not ordered against each other. Concat() serialises writes within one pipeline only. Whatever is still buffered or in flight on the old subject is written when its timer or queued write fires, which can be after the replacement run has stored a later position. The stored checkpoint then moves backwards until the next write. In production the window is narrow, because there is one replacement per resubscribe. It was much wider in the MongoDB projection tests, which polled GetLastCheckpoint every 100 ms against a live subscription until #603 changed them to read the collection directly.

3. _subjects is a plain Dictionary on a singleton. AddCheckpointStore<T>() registers the store as a singleton shared by every subscription in the container. GetLastCheckpoint writes the dictionary (L42) while other subscriptions' commit handlers read it in StoreCheckpoint (L75). One subscription resubscribing while another commits is an unsynchronised write and read.

Related, in the same pipeline

.Subscribe() has no error handler (L63). If StoreInternal throws, for example on a transient MongoDB failure, the sequence terminates and later OnNext calls on that subject go nowhere. Non-forced checkpoints then stop being persisted until the next resubscribe replaces the subject, or until shutdown force-writes the last position.

CheckpointCommitHandler cannot see this. StoreCheckpoint(force: false) returns as soon as OnNext does, so the handler records the position as committed and its UnableToCommitPosition log never fires.

Suggested direction

• Make GetLastCheckpoint a pure read. Create the subject lazily, once per checkpoint ID (ConcurrentDictionary.GetOrAdd on the first non-forced store), so there is one pipeline per ID for the lifetime of the store.
• Give each write its own error handling (catch, log, continue) so one failed write does not end the pipeline.
• Route forced writes through the same serialisation as batched ones, so a forced write cannot race a batched write that is still in flight.
• Implement IAsyncDisposable on the store to complete the subjects and flush on shutdown.

Alternatively, consider dropping store-level batching. CheckpointCommitHandler already batches by CheckpointCommitBatchSize and CheckpointCommitDelayMs before it calls the store, so BatchSize and BatchIntervalSec batch a second time.