Commit filtered AllStreamSubscription checkpoint at the caught-up transition (#559) - #560
Conversation
…ught-up transition (#559) The filtered AllStreamSubscription only advanced its checkpoint on received events and server checkpoint messages, which the server sends every CheckpointInterval x max search window scanned events. On stores smaller than that interval, or once the subscription goes live and the server falls quiet, the stored checkpoint parks below the $all head forever - or is never stored at all when the filter matches nothing on a fresh store. Restarts re-scan everything since the parked position, and consumers comparing the checkpoint to the head see a phantom, never-closing lag. The callback-based SubscribeToAllAsync client API silently discards the CaughtUp notification, so the subscription now uses the Messages-based SubscribeToAll API with its own message pump, preserving the confirmation, dispatch, and drop semantics of the callback wrapper. On CaughtUp - which carries no position in the current client - the $all head read just before subscribing is routed through the existing ordered HandleCheckpointReached path from #554: everything at or below it has provably been scanned by the time CaughtUp arrives. The commit is skipped whenever a position at or past the head is already known (stored checkpoint or anything delivered this run), since the commit machinery is sequence-gated, not position-monotonic, and would otherwise regress the stored checkpoint. Also points the ARM64 test container at the 26.1.1-experimental-arm64-10.0-noble image; the 8.0-jammy tag no longer exists on Docker Hub, so every KurrentDB test failed instantly on Apple Silicon. Closes #559 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
PR Summary by QodoCommit filtered $all checkpoint on CaughtUp transition
AI Description
Diagram
High-Level Assessment
Files changed (3)
|
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: cf9e7df8a9
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| if (headPosition is { } head && (_lastScannedPosition is not { } lastScanned || head > lastScanned)) { | ||
| _lastScannedPosition = head; | ||
| await HandleCheckpointReached(new(head, head), cancellationToken).NoContext(); |
There was a problem hiding this comment.
When a live filtered subscription falls behind and catches up again after scanning fewer than CheckpointInterval non-matching events, _lastScannedPosition is unchanged and this branch compares it only with the head captured during the original subscription. The later CaughtUp therefore commits nothing, leaving the checkpoint permanently behind the newly scanned tail until another checkpoint message arrives or the process restarts. Track a new safe head for each FellBehind → CaughtUp cycle rather than reusing the initial value.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Confirmed — the candidate was frozen at the initial subscribe, so later cycles could never commit the newly scanned tail. Fixed in 32d032a by refreshing the candidate with a $all head read when FellBehind is processed: ordered delivery guarantees every match at or below that read arrives before the next CaughtUp, so it carries the same safety invariant as the pre-subscribe read. The refresh deliberately happens at FellBehind rather than at CaughtUp — a head read after the caught-up message races matches that are still in flight between the server's live transition and the read, and committing past them would skip them on a crash-restart.
| } catch (Exception ex) when (ex is SubscriptionException or DeserializationException) { | ||
| Dropped(DropReason.SubscriptionError, ex); | ||
| } catch (Exception ex) { | ||
| Dropped(DropReason.ServerError, ex); |
There was a problem hiding this comment.
With ThrowOnError enabled, DeserializeData catches a serializer failure but rethrows the original exception, not the newly constructed DeserializationException. Consequently this filter does not match typical malformed-payload failures and the following catch reports them as ServerError, whereas the previous callback API classified exceptions from the event callback as SubscriptionError. This changes the observable drop reason for a bad event and incorrectly attributes consumer data failures to the transport.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Confirmed — DeserializeData rethrows the original serializer exception (throw;), constructing the DeserializationException only for logging, so the type filter never matched malformed payloads. Fixed in 32d032a by dropping the type matching entirely and classifying by source, the same split the callback wrapper used: an inner try around message handling maps any handler failure to SubscriptionError, and only MoveNextAsync failures reach the outer ServerError path.
Code Review by Qodo
1.
|
![qodo-free-for-open-source-projects[bot]](https://avatars.githubusercontent.com/in/1068000?s=60&v=4){kind=small}
There was a problem hiding this comment.
Pull request overview
Note
Copilot couldn't run its full agentic review because it didn't start before the timeout. Make sure your repository has a runner available, or add a copilot-code-review.yml file specifying one with the runs-on attribute. See the docs for more details.
This PR ensures filtered AllStreamSubscription checkpoints are committed when a subscription transitions to CaughtUp, addressing cases where the server never emits checkpoint-reached messages (small stores / idle tails) and preventing perpetual “phantom lag” and unnecessary re-scans on restart.
Changes:
- Switch
AllStreamSubscriptionfrom callback-basedSubscribeToAllAsyncto message-basedSubscribeToAllwith a custom message pump to observeStreamMessage.CaughtUp. - Commit the pre-subscribe
$allhead onCaughtUp, guarded to avoid checkpoint regression. - Add regression tests for small-store and
StartFrom = Latestcaught-up scenarios; update ARM64 test container image tag.
Reviewed changes
Copilot reviewed 3 out of 3 changed files in this pull request and generated 4 comments.
| File | Description |
|---|---|
| src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/AllStreamSubscription.cs | Introduces message pump to handle CaughtUp and commit the pre-subscribe $all head safely. |
| src/KurrentDB/test/Eventuous.Tests.KurrentDB/Subscriptions/CaughtUpCheckpointTests.cs | Adds regression tests validating checkpoint commits on caught-up transitions with never-matching filters. |
| src/KurrentDB/test/Eventuous.Tests.KurrentDB/Fixtures/KurrentDBContainer.cs | Updates ARM64 container image tag to an existing Docker Hub tag. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| async Task<ulong?> GetAllStreamHead(CancellationToken cancellationToken) { | ||
| var lastEvent = await Client | ||
| .ReadAllAsync(Direction.Backwards, Position.End, 1, userCredentials: Options.Credentials, cancellationToken: cancellationToken) | ||
| .ToArrayAsync(cancellationToken) | ||
| .NoContext(); | ||
|
|
||
| return lastEvent.Length == 0 ? null : lastEvent[0].Event.Position.CommitPosition; | ||
| } |
| } catch (Exception) when (cancellationToken.IsCancellationRequested) { | ||
| // Normal shutdown: the subscription got disposed or the token got cancelled mid-read | ||
| } catch (Exception ex) when (ex is SubscriptionException or DeserializationException) { | ||
| Dropped(DropReason.SubscriptionError, ex); | ||
| } catch (Exception ex) { | ||
| Dropped(DropReason.ServerError, ex); | ||
| } finally { | ||
| // Double disposal on the unsubscribe path is fine; on the dropped path this is the only | ||
| // cleanup of the underlying call before Resubscribe replaces the subscription. | ||
| await messages.DisposeAsync().NoContext(); | ||
| subscription.Dispose(); | ||
| } |
| async Task<ulong> GetLastAllStreamPosition(CancellationToken cancellationToken) { | ||
| var lastEvent = await Client.ReadAllAsync(Direction.Backwards, Position.End, 1, cancellationToken: cancellationToken).ToArrayAsync(cancellationToken); | ||
|
|
||
| return lastEvent.Length == 0 ? 0 : lastEvent[0].Event.Position.CommitPosition; | ||
| } |
| async Task<ulong> GetLastAllStreamPosition(CancellationToken cancellationToken) { | ||
| var lastEvent = await Client.ReadAllAsync(Direction.Backwards, Position.End, 1, cancellationToken: cancellationToken).ToArrayAsync(cancellationToken); | ||
|
|
||
| return lastEvent.Length == 0 ? 0 : lastEvent[0].Event.Position.CommitPosition; | ||
| } |
… classify drops by source Review follow-ups on #560: - The caught-up commit candidate was captured once before subscribing, so later FellBehind -> CaughtUp cycles could never commit the newly scanned tail. The candidate is now refreshed with a $all head read when FellBehind is processed - safe because every match at or below that read is delivered before the next CaughtUp, unlike a read at CaughtUp time which races in-flight live matches. - Drop reasons are now classified by source (handler failures -> SubscriptionError, transport failures -> ServerError) instead of exception types: DeserializeData rethrows the raw serializer exception, so type matching misattributed malformed payloads to the server. - A clean message-stream completion without cancellation now triggers a ServerError drop so the subscription resubscribes instead of staying silently dead. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
![github-code-quality[bot]](https://avatars.githubusercontent.com/u/9919?s=60&v=4){kind=small}
| [RequiresDynamicCode(AttrConstants.DynamicSerializationMessage)] | ||
| [RequiresUnreferencedCode(AttrConstants.DynamicSerializationMessage)] | ||
| async Task PumpMessages( | ||
| KurrentDBClient.StreamSubscriptionResult subscription, |
| } catch (Exception ex) { | ||
| Dropped(DropReason.ServerError, ex); | ||
| } finally { |
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 3 out of 3 changed files in this pull request and generated no new comments.
Comments suppressed due to low confidence (1)
src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/AllStreamSubscription.cs:170
- In
StreamMessage.FellBehind, failures fromGetAllStreamHeadare currently caught by the generic per-message handler and reported asDropReason.SubscriptionError. SinceGetAllStreamHeadis a server I/O operation (not a consumer/handler/deserialization failure), it should be classified asDropReason.ServerErrorto preserve the intended drop categorization and avoid misleading drop reporting.
case StreamMessage.FellBehind:
// Falling behind re-enters catch-up mode, making the current head the new
// caught-up commit candidate: every match at or below it is delivered before
// the next caught-up notification, exactly like the pre-subscribe head on the
// initial catch-up. Reading the head on the caught-up message instead would
// be unsafe — matches between the server's live transition and the read could
// still be in flight, and committing past them skips them on restart.
headPosition = await GetAllStreamHead(cancellationToken).NoContext();
break;
Test Results 46 files + 24 46 suites +24 12m 56s ⏱️ - 3m 15s Results for commit 1329dc4. ± Comparison against base commit 7827d2f. This pull request removes 5 and adds 12 tests. Note that renamed tests count towards both.♻️ This comment has been updated with latest results. |
… drop FellBehind read to transport errors Review flow follow-ups on #560: - Resolved link events flowed the resolved target's position into the consume context and checkpoint. A link created after the caught-up commit can resolve to an event arbitrarily older than the committed head, and the commit machinery is sequence-gated, not position-monotonic, so acking such a link regressed the stored checkpoint to the target's position. The context position is now the delivered record's own $all position (OriginalPosition, falling back to the original event's record position). Covered by a red-checked integration test that writes a link after the caught-up commit. - The FellBehind head refresh is a server read, but ran inside the per-message try, so its failures were labelled SubscriptionError. It now runs outside that try and reaches the transport (ServerError) path. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
| } catch (Exception) { | ||
| // Nothing to see here | ||
| } |
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 4 out of 4 changed files in this pull request and generated no new comments.
Comments suppressed due to low confidence (1)
src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/AllStreamSubscription.cs:168
StreamMessage.Eventhandling assumesresolvedEvent.Eventis non-null, but KurrentDBResolvedEvent.Eventcan be null for deleted/tombstone records (see the existing guard inStreamSubscription). In that case this code will throw (inGetContextPosition/CreateContext) and incorrectly drop the subscription asSubscriptionError.
Add the same null guard here to safely ignore such records.
case StreamMessage.Event(var resolvedEvent):
_lastScannedPosition = GetContextPosition(resolvedEvent);
await HandleInternal(CreateContext(resolvedEvent, cancellationToken)).NoContext();
Problem
Closes #559.
#554 advances the filtered
AllStreamSubscriptioncheckpoint on servercheckpointReachedmessages, but the server only emits those everyCheckpointInterval× max search window scanned events (~320 with defaults). Two gaps remained:$allhead wedge permanently on fresh stores.Fix
The callback-based
SubscribeToAllAsyncclient API consumes the underlying message stream internally and silently discardsStreamMessage.CaughtUp, so the subscription now uses theMessages-basedSubscribeToAllAPI with its own message pump. The pump keeps the callback wrapper's semantics — awaits the subscription confirmation before returning fromSubscribe, dispatches events andAllStreamCheckpointReachedto the same handlers as before, and mirrors the drop classification (shutdown → silent, handler/deserialization errors →SubscriptionError, transport errors →ServerError).On
CaughtUp— positionless in KurrentDB.Client 1.3.0 — the$allhead read just before subscribing is routed through the existing orderedHandleCheckpointReachedpath from #554: everything at or below that position has provably been scanned by the timeCaughtUparrives, and the contiguous-sequence gate still holds the commit behind any unacknowledged matched event. This establishes the invariant that a caught-up subscription always has a stored checkpoint at or past the head it caught up to, and re-fires naturally on every FellBehind → CaughtUp cycle.One safety detail beyond the issue text: the commit machinery is sequence-gated but not position-monotonic, so blindly committing the head could regress the stored checkpoint (e.g. resuming at a stored scan position past the head, or events written between the head read and the subscribe). The pump tracks the highest scanned position — seeded from the stored checkpoint, advanced by every event and checkpoint message — and skips the caught-up commit unless the head actually advances it.
StreamSubscriptionis left on the callback API, as the issue notes it commits per received event and doesn't need this.Tests
New
CaughtUpCheckpointTests(written first, red → green): per the issue's regression note they keep the store far below the checkpoint interval (never-matching filter,CheckpointInterval = 4096) so no server checkpoint message can mask a missing caught-up commit.CaughtUpCheckpointOnSmallStoreTests— before the fix, parked below the head (22941 vs 29336); now commits at/past the head with zero events handled.CaughtUpCheckpointFromLatestTests—StartFrom = Latest; before the fix no checkpoint was ever stored; now the immediate caught-up transition commits the pre-subscribe head.Full KurrentDB suite (59), Mongo projections (9), and SignalR integration (3) all pass on the final code; full solution builds clean.
Also fixes the ARM64 test container image: the
26.1.1-experimental-arm64-8.0-jammytag no longer exists on Docker Hub (every KurrentDB test failed instantly on Apple Silicon); now uses26.1.1-experimental-arm64-10.0-noble.🤖 Generated with Claude Code