Hacktoberfest 2026: những issue maintainer đã đánh dấu cho tháng Mười, đang mở và phù hợp người mới. Xem issue Hacktoberfest

MongoCheckpointStore leaks a batching pipeline on every GetLastCheckpoint call

Đang mở
  • Có gì không ổn?
#616 1 bình luận 0 reaction 0 người được giao Xem trên GitHub

Maintainer thường phản hồi trong vòng 7 ngày

Chưa có ai nhận issue này.

Đánh giá

Độ khó
4/5
Thời gian dự kiến
3-5 ngày
Mức phù hợp với người mới
48/100
Loại issue
Lỗi
Độ rõ ràng
Khá rõ ràng
Mức độ hoạt động
Sôi nổi
Công nghệ
csharp, mongodb
Lĩnh vực
backend, database

Hướng nghiên cứu

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.

Do mô hình lập chỉ mục viết ra từ nội dung của issue.

Mô tả

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.

Ngôn ngữ chính
C#
Star
512
Fork
99
Merge trung bình
5 ngày 6 giờ
Pull request đã merge (30 ngày)
5

Chuẩn bị môi trường

Bắt đầu từ đâu

  1. Đọc hết issue, rồi đọc hướng dẫn đóng góp của dự án.
  2. Bình luận trên issue rằng bạn sẽ nhận — tránh hai người làm cùng một việc.
  3. Fork repository và làm thay đổi trên một nhánh.
  4. Mở pull request có tham chiếu số hiệu của issue.

Issue khác của Eventuous/eventuous

Tất cả issue của Eventuous/eventuous

Issue tương tự

Thêm issue về C#

Nhận issue mới trong hộp thư của bạn

Bản tóm tắt ngắn những issue GitHub phù hợp với người mới.