Hacktoberfest 2026：メンテナが10月に向けて印を付けた、オープンで初心者向けの issue。 Hacktoberfest の issue を見る

MongoCheckpointStore leaks a batching pipeline on every GetLastCheckpoint call

オープン
  • 何か問題がありますか？
#616 コメント 1 件 リアクション 0 件 担当者 0 名 GitHub で見る

メンテナーはふだん 7 日以内に返信

まだ誰も着手していません。

評価

難易度
4/5
見積もり時間
3〜5日
初心者へのやさしさ
48/100
issue の種類
バグ
明瞭さ
おおむね明確
活発さ
活発
技術スタック
csharp, mongodb
領域
backend, database

調査の方向性

Start with src/Mongo/src/Eventuous.Projections.MongoDB/MongoCheckpointStore.cs and trace how GetLastCheckpoint, StoreCheckpoint, and GetSubject handle subjects, batching, and errors. Then read EventSubscriptionWithCheckpoint.GetCheckpoint and the checkpoint store registration to understand call frequency and lifetime. Add a regression test for repeated reads or resubscription, then verify that checkpoint writes remain ordered and failures do not stop later writes.

索引モデルが issue の本文から書いたものです。

説明

bug mongo

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.

主要言語
C#
スター
512
フォーク
99
平均マージ
5日 6時間
マージ済み PR（30日）
5

環境構築

はじめの一歩

  1. issue を最後まで読み、次にプロジェクトのコントリビューションガイドを読みます。
  2. 着手することを issue にコメントします — 二人が同じ作業をするのを防げます。
  3. リポジトリをフォークし、ブランチを切って変更します。
  4. issue 番号を参照したプルリクエストを送ります。

Eventuous/eventuous のほかの issue

Eventuous/eventuous の issue をすべて見る

似ている issue

C# の issue をもっと見る

新しい issue をメールで受け取る

初心者向けの GitHub issue を短くまとめたダイジェスト。