Advance filtered AllStreamSubscription checkpoints on server checkpointReached

[alexeyzimarev]

Problem

AllStreamSubscription.Subscribe constructs SubscriptionFilterOptions(filter, checkpointInterval) without the third checkpointReached parameter, so the callback defaults to a no-op. A filtered $all subscription's checkpoint therefore only advances when a filter-matched event is processed:

• Sparse filters (e.g. a single category on a busy log) park their stored checkpoint at the last matched event — potentially weeks behind — so every restart re-scans the entire unmatched stretch.
• Any consumer comparing the stored checkpoint to the $all head (including the eventuous.subscription.gap.count metric) sees a phantom, never-closing lag.

Fix

Wire checkpointReached to route the server-reported checkpoint position through the existing ordered commit machinery as a checkpoint-only message: a null-payload MessageConsumeContext (type $checkpoint-reached) with the position and the subscription's Sequence++, handed to HandleInternal. The null-message arm in EventSubscription.Handler already Ignores + Acknowledges such contexts, which lands in CheckpointCommitHandler.Commit with an ordered sequence number — no new commit paths.

Why the ordering is safe

• The KurrentDB client invokes eventAppeared and checkpointReached sequentially on one delivery loop (each awaited before MoveNextAsync), so Sequence++ never races event contexts and sequences reflect exact delivery order.
• A gap context acks inline while real events ack on the async worker — but CheckpointCommitHandler's contiguous-sequence gate (CommitPositionSequence.FirstBeforeGap) holds a gap position back until every lower-sequence event has acked. The checkpoint can never commit past an unprocessed event, so crash/restart semantics are unchanged.
• On resubscribe, gap contexts are fully awaited inline by the dropped delivery loop, so none survive into the new run's reset Sequence.

Test

CheckpointReachedTests.CheckpointAdvancesPastUnmatchedEvents: a filtered subscription whose filter matches nothing, 500 unmatched events appended — asserts the handler saw zero events AND the stored checkpoint advances past the last event's position (bounded 30s poll + 60s timeout). Verified red without the fix, green with it.

Notes

• No public API change; AOT annotations mirror the existing CreateContext.
• Pre-existing (not introduced here): the null-message arm creates an unstarted SubscriptionActivity for async contexts when diagnostics are enabled; gap contexts now hit that arm periodically. Harmless inert allocation — happy to short-circuit it here or in a follow-up if preferred.

🤖 Generated with Claude Code

[qodo-free-for-open-source-projects]

PR Summary by Qodo

Advance filtered $all checkpoints using server checkpointReached callbacks

🐞 Bug fix 🧪 Tests 🕐 20-40 Minutes

AI Description

• Wire KurrentDB checkpointReached into filtered $all subscriptions to prevent checkpoint parking.
• Emit payload-less "checkpoint" contexts through the existing ordered ack/commit pipeline.
• Add a regression test proving checkpoints advance even when the filter matches nothing.

Diagram

graph TD
  S["KurrentDB server"] --> A["AllStreamSubscription"] --> E["Event context"] --> H["HandleInternal"] --> R["Handler (ignore+ack)"] --> C["CheckpointCommitHandler"] --> D[("CheckpointStore")]
  S --> A --> P["Checkpoint ctx"] --> H

Loading

High-Level Assessment

The following are alternative approaches to this PR:

1. Commit directly from checkpointReached callback

• ➕ Simpler control flow (no synthetic consume context)
• ➕ Avoids running through diagnostics/handler paths for null messages
• ➖ Bypasses ordered contiguous-sequence gate; can commit past unacked events if callback timing changes
• ➖ Introduces a second commit path with different crash/restart semantics to maintain

2. Track a separate 'scan position' alongside the matched-event checkpoint

• ➕ Explicitly models filtered-scan progress vs processed progress
• ➕ Could improve observability by exposing both positions
• ➖ Requires checkpoint schema/API changes and dual-position reconciliation logic
• ➖ More invasive change across checkpoint stores and metrics

Recommendation: Keep the PR’s approach: routing checkpointReached through the existing ordered ack/commit machinery preserves current crash/restart semantics and avoids adding a second checkpoint commit pathway. The synthetic payload-less context cleanly reuses the contiguous-sequence guard in CheckpointCommitHandler, which is the core safety requirement for advancing checkpoints in the presence of async event handling.

Files changed (2) +149 / -1

Bug fix (1) +40 / -1

AllStreamSubscription.csRoute checkpointReached positions through ordered checkpoint commits +40/-1

Route checkpointReached positions through ordered checkpoint commits

• Adds a dedicated message type and wires SubscriptionFilterOptions.checkpointReached to create a payload-less MessageConsumeContext carrying the server-reported commit position. The synthetic context is dispatched via HandleInternal so it gets ignored+acked and still participates in ordered checkpoint committing using the subscription sequence counter.

src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/AllStreamSubscription.cs

Tests (1) +109 / -0

CheckpointReachedTests.csRegression test: checkpoint advances with filters that match nothing +109/-0

Regression test: checkpoint advances with filters that match nothing

• Introduces an integration-style test that appends many unmatched events, starts a filtered $all subscription whose filter excludes everything, and polls the checkpoint store until the persisted checkpoint reaches the $all head position. Asserts the handler processed zero events while the checkpoint still advanced beyond the last written position.

src/KurrentDB/test/Eventuous.Tests.KurrentDB/Subscriptions/CheckpointReachedTests.cs

[qodo-free-for-open-source-projects]

Code Review by Qodo

🐞 Bugs (0) 📘 Rule violations (0) 📎 Requirement gaps (0) 🎨 UX issues (0) 🔗 Cross-repo conflicts (0) 📜 Skill insights (0)

1. Commit queue overflow stalls ✓ Resolved 🐞 Bug ☼ Reliability

Description

checkpointReached contexts are ignored and acknowledged immediately, which enqueues a checkpoint
commit for every callback; CheckpointCommitHandler writes to a bounded channel with
throwOnFull=true. If that channel ever throws (burst of checkpointReached + slow store), the
dropped sequence number can permanently block future commits because CheckpointCommitHandler
refuses to commit past sequence gaps.

Code

src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/AllStreamSubscription.cs[R87-92]

   protected override async ValueTask Subscribe(CancellationToken cancellationToken) {
       var filterOptions = new SubscriptionFilterOptions(
           Options.EventFilter ?? EventTypeFilter.ExcludeSystemEvents(),
-            Options.CheckpointInterval
+            Options.CheckpointInterval,
+            (_, position, ct) => HandleCheckpointReached(position, ct)
       );

Evidence

The new wiring passes checkpointReached into HandleInternal. For null messages, Handler
acknowledges immediately; that calls Ack, which always enqueues into CheckpointCommitHandler.
The commit handler writes into a bounded channel configured to throw on full, and its commit logic
explicitly refuses to commit when there is a sequence gap—so any dropped enqueue can block all
future commits.

src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/AllStreamSubscription.cs[87-92]
src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/AllStreamSubscription.cs[152-180]
src/Core/src/Eventuous.Subscriptions/EventSubscription.cs[112-129]
src/Core/src/Eventuous.Subscriptions/Context/AsyncConsumeContext.cs[21-43]
src/Core/src/Eventuous.Subscriptions/EventSubscriptionWithCheckpoint.cs[91-108]
src/Core/src/Eventuous.Subscriptions/Checkpoints/CheckpointCommitHandler.cs[53-107]
src/Core/src/Eventuous.Subscriptions/Channels/ChannelWorkerBase.cs[14-16]
src/Core/src/Eventuous.Subscriptions/Channels/ChannelExtensions.cs[37-41]

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

## Issue description
The PR introduces synthetic checkpoint contexts that are immediately ACKed (no pipeline work), which can enqueue commit positions much faster than before. `CheckpointCommitHandler` uses a bounded channel and is configured to throw on full; a single dropped `CommitPosition` (e.g., `ChannelFullException`) can create a sequence gap that prevents any further commits (`AtGap()` returns `CommitPosition.None`). This can stall checkpoint advancement indefinitely.
### Issue Context
- New wiring of `checkpointReached` into `HandleInternal` increases commit enqueue rate, especially with small `CheckpointInterval` and/or slow checkpoint store.
- The commit handler’s correctness depends on receiving a contiguous sequence of commit positions.
### Fix Focus Areas
- src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/AllStreamSubscription.cs[87-92]
- src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/AllStreamSubscription.cs[152-180]
- src/Core/src/Eventuous.Subscriptions/EventSubscriptionWithCheckpoint.cs[91-108]
- src/Core/src/Eventuous.Subscriptions/Checkpoints/CheckpointCommitHandler.cs[53-107]
- src/Core/src/Eventuous.Subscriptions/Channels/ChannelWorkerBase.cs[14-16]
- src/Core/src/Eventuous.Subscriptions/Channels/ChannelExtensions.cs[37-41]
### Suggested fix
Prefer backpressure over throwing for checkpoint commits:
- Change `CheckpointCommitHandler` to NOT set `throwOnFull=true` (let `WriteAsync` await capacity), **or**
- Catch `ChannelFullException` at the commit enqueue site and retry with a delay until enqueued, ensuring no sequence is skipped, **or**
- Coalesce checkpointReached updates (keep only the latest server position and enqueue at most one pending commit per subscription loop) so the commit queue cannot be flooded.
Keep the invariant: never drop a commit position sequence number, otherwise commit progression can stall at a gap.

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---

2. Async null Activity leak ✓ Resolved 🐞 Bug ➹ Performance

Description

AllStreamSubscription.HandleCheckpointReached routes a payload-less context (Message=null)
through HandleInternal, which hits EventSubscription.Handler’s async null-message branch. With
diagnostics enabled, Handler still allocates a SubscriptionActivity but does not start or
dispose it for async + null-message contexts, so frequent checkpointReached callbacks can create
sustained unnecessary allocations.

Code

src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/AllStreamSubscription.cs[R152-180]

+    /// <summary>
+    /// Handles a server-reported checkpoint position for the filtered subscription by routing it
+    /// through the same ordered commit machinery as real events, as a payload-less context. Without
+    /// this, the stored checkpoint would only advance when a filter-matched event is processed, so a
+    /// long unmatched stretch (sparse filters, quiet servers) leaves the checkpoint parked at the last
+    /// matched event: restarts re-scan everything since then, and consumers comparing the checkpoint to
+    /// the $all head see a phantom, never-closing lag.
+    /// </summary>
+    [RequiresDynamicCode(AttrConstants.DynamicSerializationMessage)]
+    [RequiresUnreferencedCode(AttrConstants.DynamicSerializationMessage)]
+    Task HandleCheckpointReached(global::KurrentDB.Client.Position position, CancellationToken cancellationToken) {
+        var context = new MessageConsumeContext(
+            position.CommitPosition.ToString(),
+            CheckpointReachedMessageType,
+            "",
+            "$all",
+            position.CommitPosition,
+            position.CommitPosition,
+            position.CommitPosition,
+            Sequence++,
+            DateTime.UtcNow,
+            null,
+            null,
+            SubscriptionId,
+            cancellationToken
+        );
+
+        return HandleInternal(context).AsTask();
+    }

Evidence

The new code creates a null-payload context and routes it through the normal handler.
EventSubscription.Handler always allocates an activity when diagnostics are enabled, but only
attaches/starts it for non-null messages; for async null messages it acknowledges inline and never
disposes the allocated activity, making the allocation waste proportional to checkpointReached
frequency.

src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/AllStreamSubscription.cs[87-92]
src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/AllStreamSubscription.cs[152-180]
src/Core/src/Eventuous.Subscriptions/EventSubscription.cs[86-149]
src/Core/src/Eventuous.Subscriptions/Filters/AsyncHandlingFilter.cs[28-33]

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

## Issue description
`EventSubscription.Handler` allocates a `SubscriptionActivity` whenever diagnostics are enabled, but for async contexts where `context.Message == null` it never starts the activity (it isn't put into `context.Items` for `AsyncHandlingFilter` to start) and it also never disposes it (because `activity.Dispose()` is only executed for non-async contexts). The new checkpointReached synthetic contexts are exactly async + null-message, so this becomes a hot allocation path.
### Issue Context
The PR introduces frequent payload-less contexts via `AllStreamSubscription.HandleCheckpointReached`, increasing the frequency of the async null-message handler path.
### Fix Focus Areas
- src/Core/src/Eventuous.Subscriptions/EventSubscription.cs[86-149]
- src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/AllStreamSubscription.cs[152-180]
### Suggested fix
Move `SubscriptionActivity.Create(...)` inside the `if (context.Message != null)` block (or add a `try/finally` that disposes `activity` for the async null-message path). Ensure the activity lifecycle remains correct for the async non-null path where `AsyncHandlingFilter` starts and disposes the activity.

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---