From 1cecb2352087a8fd21d2f655c58a0906a915977d Mon Sep 17 00:00:00 2001 From: Alexey Zimarev Date: Tue, 14 Jul 2026 18:37:39 +0200 Subject: [PATCH 1/4] Fix filtered AllStreamSubscription checkpoint parking on unmatched events Wire up the checkpointReached callback on SubscriptionFilterOptions so server-reported checkpoint positions flow through the same ordered commit machinery as real events, as payload-less contexts (Message is null, so Handler ignores and acks without touching the consume pipe). Without this, a filtered $all subscription's checkpoint only advanced when a filter-matched event was processed. A long unmatched stretch (sparse filters, quiet servers) left the checkpoint parked at the last matched event: restarts re-scanned everything since then, and any consumer comparing the checkpoint to the $all head saw a phantom, never-closing lag. Adds CheckpointReachedTests, which reproduces the bug with a filter that matches nothing and asserts the checkpoint still advances past written events. --- .../Subscriptions/AllStreamSubscription.cs | 41 ++++++- .../Subscriptions/CheckpointReachedTests.cs | 109 ++++++++++++++++++ 2 files changed, 149 insertions(+), 1 deletion(-) create mode 100644 src/KurrentDB/test/Eventuous.Tests.KurrentDB/Subscriptions/CheckpointReachedTests.cs diff --git a/src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/AllStreamSubscription.cs b/src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/AllStreamSubscription.cs index 6acf814a4..cb8d88dd3 100644 --- a/src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/AllStreamSubscription.cs +++ b/src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/AllStreamSubscription.cs @@ -70,6 +70,14 @@ public AllStreamSubscription( IMetadataSerializer? metaSerializer = null ) : base(client, options, checkpointStore, consumePipe, SubscriptionKind.All, loggerFactory, eventSerializer, metaSerializer) { } + /// + /// Message type used for the synthetic, payload-less context created when the server reports + /// a checkpoint position for a filtered subscription that hasn't matched any event in a while. + /// This lets the checkpoint advance past long unmatched stretches instead of parking at the + /// last matched event. + /// + internal const string CheckpointReachedMessageType = "$checkpoint-reached"; + /// /// Starts the subscription /// @@ -79,7 +87,8 @@ public AllStreamSubscription( protected override async ValueTask Subscribe(CancellationToken cancellationToken) { var filterOptions = new SubscriptionFilterOptions( Options.EventFilter ?? EventTypeFilter.ExcludeSystemEvents(), - Options.CheckpointInterval + Options.CheckpointInterval, + (_, position, ct) => HandleCheckpointReached(position, ct) ); var (_, position) = await GetCheckpoint(cancellationToken).NoContext(); @@ -140,6 +149,36 @@ MessageConsumeContext CreateContext(ResolvedEvent re, CancellationToken cancella ); } + /// + /// 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. + /// + [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(); + } + /// /// Returns a measure delegate for the subscription /// diff --git a/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Subscriptions/CheckpointReachedTests.cs b/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Subscriptions/CheckpointReachedTests.cs new file mode 100644 index 000000000..6734725d9 --- /dev/null +++ b/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Subscriptions/CheckpointReachedTests.cs @@ -0,0 +1,109 @@ +using Eventuous.KurrentDB.Producers; +using Eventuous.KurrentDB.Subscriptions; +using Eventuous.Producers; +using Eventuous.Subscriptions.Registrations; +using Eventuous.TestHelpers.TUnit; +using Eventuous.Tests.Subscriptions.Base; +using KurrentDB.Client; +using Microsoft.Extensions.DependencyInjection; +using Shouldly; + +// ReSharper disable MethodHasAsyncOverload + +namespace Eventuous.Tests.KurrentDB.Subscriptions; + +/// +/// Covers the fix for the checkpoint of a filtered parking at the +/// last matched event: with a server-side event filter that never matches, the server-reported +/// checkpoint position must still flow into the checkpoint store, so a restart doesn't re-scan +/// everything since the last match. +/// +public class CheckpointReachedTests : StoreFixture { + readonly string _subscriptionId = $"test-{Guid.NewGuid():N}"; + readonly StreamName _stream = new($"test-{Guid.NewGuid():N}"); + IProducer _producer = null!; + ICheckpointStore _checkpointStore = null!; + TestEventHandler _handler = null!; + + public CheckpointReachedTests() : base(LogLevel.Information) { + AutoStart = false; + TypeMapper.RegisterKnownEventTypes(typeof(TestEvent).Assembly); + } + + [Test] + [Category("Special cases")] + [Timeout(60_000)] + public async Task CheckpointAdvancesPastUnmatchedEvents(CancellationToken cancellationToken) { + // Enough unmatched events, and a small MaxSearchWindow, so the server reports a + // checkpoint position well before it would have scanned the whole write. + const int count = 500; + + var testEvents = TestEvent.CreateMany(count); + await _producer.Produce(_stream, testEvents, new(), cancellationToken: cancellationToken); + + await Start(); + + var lastPosition = await GetLastAllStreamPosition(cancellationToken); + + var checkpoint = await PollUntilCheckpointReaches(lastPosition, TimeSpan.FromSeconds(30), cancellationToken); + + await DisposeAsync(); + + // The filter never matched anything, so no event reached the handler... + _handler.Count.ShouldBe(0); + // ...yet the checkpoint advanced past the last written (unmatched) event. + checkpoint.Position.ShouldNotBeNull(); + checkpoint.Position!.Value.ShouldBeGreaterThanOrEqualTo(lastPosition); + } + + async Task 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; + } + + // Polls until the checkpoint reaches minPosition, or returns the last-seen checkpoint once the + // deadline passes (the assertions in the test produce a clear failure message in that case). + async Task PollUntilCheckpointReaches(ulong minPosition, TimeSpan timeout, CancellationToken cancellationToken) { + var deadline = DateTime.UtcNow + timeout; + var checkpoint = await _checkpointStore.GetLastCheckpoint(_subscriptionId, cancellationToken); + + while (!(checkpoint.Position is { } position && position >= minPosition) && DateTime.UtcNow < deadline) { + await Task.Delay(200.Milliseconds(), cancellationToken); + checkpoint = await _checkpointStore.GetLastCheckpoint(_subscriptionId, cancellationToken); + } + + return checkpoint; + } + + protected override void SetupServices(IServiceCollection services) { + base.SetupServices(services); + services.AddProducer(); + + services.AddSubscription( + _subscriptionId, + c => c + .Configure( + o => { + // A prefix that will never match the produced test events, so the filter + // excludes everything, but with a small enough search window that the + // server still reports checkpoint progress while scanning past them. + o.EventFilter = EventTypeFilter.Prefix(4, "definitely-does-not-match-anything"); + o.CheckpointInterval = 1; + + o.CheckpointCommitBatchSize = 1; + o.CheckpointCommitDelayMs = 100; + } + ) + .UseCheckpointStore() + .AddEventHandler() + ); + } + + protected override void GetDependencies(IServiceProvider provider) { + base.GetDependencies(provider); + _producer = provider.GetRequiredService(); + _checkpointStore = provider.GetRequiredKeyedService(_subscriptionId); + _handler = provider.GetRequiredKeyedService(_subscriptionId); + } +} From 950c5f46b6d24e4c12a4e514cde36a90d6c4b297 Mon Sep 17 00:00:00 2001 From: Alexey Zimarev Date: Tue, 14 Jul 2026 21:23:39 +0200 Subject: [PATCH 2/4] Address review: commit-queue backpressure + skip activity for payload-less contexts Co-Authored-By: Claude Fable 5 --- .../Checkpoints/CheckpointCommitHandler.cs | 7 ++++++- src/Core/src/Eventuous.Subscriptions/EventSubscription.cs | 5 ++++- 2 files changed, 10 insertions(+), 2 deletions(-) diff --git a/src/Core/src/Eventuous.Subscriptions/Checkpoints/CheckpointCommitHandler.cs b/src/Core/src/Eventuous.Subscriptions/Checkpoints/CheckpointCommitHandler.cs index 5c0dd38a3..9624d70c4 100644 --- a/src/Core/src/Eventuous.Subscriptions/Checkpoints/CheckpointCommitHandler.cs +++ b/src/Core/src/Eventuous.Subscriptions/Checkpoints/CheckpointCommitHandler.cs @@ -52,7 +52,12 @@ public CheckpointCommitHandler( _loggerFactory = loggerFactory; var channel = Channel.CreateBounded(batchSize * 1000); - _worker = new(channel, Process, batchSize, delay, true); + // Backpressure, never throw: a dropped CommitPosition is poison — GetCommitPosition refuses + // to commit past a sequence gap, so one lost sequence number stalls checkpoint progression + // permanently (the throw is swallowed by the subscription's handler-error path). Awaiting + // capacity merely throttles the producer while the checkpoint store is slow, and only after + // the batchSize*1000 buffer is exhausted. + _worker = new(channel, Process, batchSize, delay); _worker.OnDispose = async _ => { if (_lastCommit.Valid) diff --git a/src/Core/src/Eventuous.Subscriptions/EventSubscription.cs b/src/Core/src/Eventuous.Subscriptions/EventSubscription.cs index 65b3b9028..443d69010 100644 --- a/src/Core/src/Eventuous.Subscriptions/EventSubscription.cs +++ b/src/Core/src/Eventuous.Subscriptions/EventSubscription.cs @@ -95,7 +95,10 @@ protected async ValueTask Handler(IMessageConsumeContext context) { Logger.Current ??= Log; using (Log.Logger.BeginScope(scope)) { - var activity = EventuousDiagnostics.Enabled + // No activity for payload-less contexts: they are ignored and acknowledged below without + // entering the pipe, so an activity would never be started or disposed on the async path — + // a pure allocation leak, hot since checkpoint-reached contexts arrive payload-less. + var activity = EventuousDiagnostics.Enabled && context.Message != null ? SubscriptionActivity.Create( $"{Constants.Components.Subscription}.{SubscriptionId}/{context.MessageType}", ActivityKind.Internal, From 447fc88a1f0fcaabef48b4523c1ed0ae97bfb3a0 Mon Sep 17 00:00:00 2001 From: Alexey Zimarev Date: Wed, 15 Jul 2026 12:06:28 +0200 Subject: [PATCH 3/4] Add regression tests for checkpoint gate safety and commit backpressure MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds CheckpointGatedByPendingMatchTests, which reproduces the safety half of the checkpointReached fix: a filter-matched event whose ack is held open must block the stored checkpoint from advancing past it, even while checkpoint markers for unmatched noise keep arriving. The commit handler's "Commit" diagnostic (fired before the sequence gate) proves a marker positioned BEYOND the blocked event reached the commit path during the hold — not assumed from a fixed sleep, and not satisfiable by markers for pre-existing activity below the event — and the blocked handler is always released so a failed assertion can't hang fixture disposal. Adds CheckpointCommitHandlerBackpressureTests, which pins the channel backpressure CheckpointCommitHandler switched to instead of throwing when the commit queue overflows, and that disposing with a writer parked on the full channel releases it (ChannelClosedException) and drains the queue without hanging, force-recommitting the last stored position. The store gate always opens and the handler is always disposed on failure paths, so a caught regression surfaces as an assertion failure rather than a timeout. --- ...heckpointCommitHandlerBackpressureTests.cs | 165 +++++++++++ .../Subscriptions/CheckpointReachedTests.cs | 259 ++++++++++++++++++ 2 files changed, 424 insertions(+) create mode 100644 src/Core/test/Eventuous.Tests.Subscriptions/CheckpointCommitHandlerBackpressureTests.cs diff --git a/src/Core/test/Eventuous.Tests.Subscriptions/CheckpointCommitHandlerBackpressureTests.cs b/src/Core/test/Eventuous.Tests.Subscriptions/CheckpointCommitHandlerBackpressureTests.cs new file mode 100644 index 000000000..d609476ca --- /dev/null +++ b/src/Core/test/Eventuous.Tests.Subscriptions/CheckpointCommitHandlerBackpressureTests.cs @@ -0,0 +1,165 @@ +using System.Threading.Channels; +using Eventuous.Subscriptions.Checkpoints; +using Shouldly; + +namespace Eventuous.Tests.Subscriptions; + +/// +/// Pins the channel-backpressure behaviour switched to +/// (instead of throwing) when the commit queue fills up while the checkpoint store is slow: a +/// dropped is poison — a gap in the sequence stalls checkpoint +/// progression permanently — so overflow must throttle the caller, not fail the commit. +/// +public class CheckpointCommitHandlerBackpressureTests { + static CommitPosition Pos(ulong sequence) => new(sequence, sequence, DateTime.UtcNow); + + [Test] + [Timeout(20_000)] + public async Task Commit_awaits_capacity_instead_of_throwing_when_the_channel_is_full(CancellationToken cancellationToken) { + List committed = []; + TaskCompletionSource storeGate = new(TaskCreationOptions.RunContinuationsAsynchronously); + TaskCompletionSource storeEntered = new(TaskCreationOptions.RunContinuationsAsynchronously); + + async ValueTask CommitFn(Checkpoint checkpoint, bool force, CancellationToken ct) { + storeEntered.TrySetResult(); + await storeGate.Task.WaitAsync(ct); + lock (committed) committed.Add(checkpoint.Position!.Value); + + return checkpoint; + } + + await using var handler = new CheckpointCommitHandler("backpressure-sub", CommitFn, TimeSpan.FromMilliseconds(10), batchSize: 1); + + // The gate must open no matter how the test body ends: a failed assertion with the gate + // still closed would leave the worker parked in CommitFn and the `await using` disposal + // hanging until the test timeout — masking the clean assertion failure. + try { + // Sequence 0 is picked up by the worker and blocks inside CommitFn on the stalled gate — + // this is the "current commit in flight" the rest of the batch queues behind. The entered + // signal proves the worker actually dequeued it before we start filling the channel. + await handler.Commit(Pos(0), cancellationToken); + await storeEntered.Task.WaitAsync(TimeSpan.FromSeconds(5), cancellationToken); + + // Fill the bounded channel: capacity is batchSize * 1000 = 1000. + for (ulong sequence = 1; sequence <= 1000; sequence++) { + await handler.Commit(Pos(sequence), cancellationToken).AsTask().WaitAsync(TimeSpan.FromSeconds(5), cancellationToken); + } + + // The channel is now full and the worker is still blocked on sequence 0: the next Commit + // call has nowhere to enqueue to. It must not throw — it should await channel capacity. + var overflow = handler.Commit(Pos(1001), cancellationToken); + await Task.Delay(200, cancellationToken); + overflow.IsCompleted.ShouldBeFalse("Commit should be backpressured (awaiting channel capacity), not completed or throwing"); + + // Let the store recover: the worker drains the backlog, one sequence at a time. + storeGate.TrySetResult(); + + await overflow.AsTask().WaitAsync(TimeSpan.FromSeconds(5), cancellationToken); + + // Poll until the store has recorded sequence 1001, then assert nothing was skipped along + // the way: because the commit batch size is 1 and the run is fully contiguous, every single + // sequence from 0 to 1001 must have gone through CommitFn, in order. + var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(5); + List snapshot; + + while (true) { + lock (committed) snapshot = [..committed]; + + if (snapshot.Count > 0 && snapshot[^1] == 1001) break; + + if (DateTime.UtcNow > deadline) break; + + await Task.Delay(50, cancellationToken); + } + + snapshot.ShouldBe(Enumerable.Range(0, 1002).Select(i => (ulong)i).ToList()); + } finally { + storeGate.TrySetResult(); + } + } + + [Test] + [Timeout(20_000)] + public async Task Dispose_releases_a_backpressured_commit_and_drains_without_hanging(CancellationToken cancellationToken) { + List<(ulong Position, bool Force)> committed = []; + TaskCompletionSource storeGate = new(TaskCreationOptions.RunContinuationsAsynchronously); + TaskCompletionSource storeEntered = new(TaskCreationOptions.RunContinuationsAsynchronously); + + async ValueTask CommitFn(Checkpoint checkpoint, bool force, CancellationToken ct) { + storeEntered.TrySetResult(); + await storeGate.Task.WaitAsync(ct); + lock (committed) committed.Add((checkpoint.Position!.Value, force)); + + return checkpoint; + } + + var handler = new CheckpointCommitHandler("backpressure-dispose-sub", CommitFn, TimeSpan.FromMilliseconds(10), batchSize: 1); + + Task? disposeTask = null; + + // Two guarantees on every exit path: the gate opens (a failed assertion with the gate still + // closed would otherwise leave the worker parked in CommitFn and disposal hanging until the + // test timeout), and the handler's disposal is awaited (bounded) so a failure doesn't leak + // a live worker into the rest of the run. + try { + // Sequence 0 is dequeued by the worker and blocks in CommitFn on the still-stalled gate. + await handler.Commit(Pos(0), cancellationToken); + await storeEntered.Task.WaitAsync(TimeSpan.FromSeconds(5), cancellationToken); + + // Fill the bounded channel (capacity batchSize * 1000 = 1000) and park one more Commit in + // backpressure — a writer genuinely awaiting channel capacity when Dispose begins. + for (ulong sequence = 1; sequence <= 1000; sequence++) { + await handler.Commit(Pos(sequence), cancellationToken).AsTask().WaitAsync(TimeSpan.FromSeconds(5), cancellationToken); + } + + var overflow = handler.Commit(Pos(1001), cancellationToken); + await Task.Delay(200, cancellationToken); + overflow.IsCompleted.ShouldBeFalse("The overflow Commit should be backpressured before Dispose begins"); + + // Dispose while the store is stalled and a writer is parked on the full channel. Dispose + // completes the channel writer, which releases the parked write rather than leaving it + // hanging forever: the pending WriteAsync faults with ChannelClosedException (observed + // behaviour, pinned here). The position is lost, but only because the handler is shutting + // down — the caller is unblocked, not stalled. The outcome is captured first, then + // asserted, so a surprise here still flows through the finally-side cleanup. + disposeTask = handler.DisposeAsync().AsTask(); + + var overflowOutcome = await overflow.AsTask() + .WaitAsync(TimeSpan.FromSeconds(5), cancellationToken) + .ContinueWith(t => t.Exception?.GetBaseException(), TaskContinuationOptions.ExecuteSynchronously); + + overflowOutcome.ShouldBeOfType("Completing the writer should release the parked Commit with ChannelClosedException"); + + // Let the store recover so dispose can drain the queued positions and run its final + // force-commit within its own internal bounds. + storeGate.TrySetResult(); + + var completed = await Task.WhenAny(disposeTask, Task.Delay(TimeSpan.FromSeconds(10), cancellationToken)); + completed.ShouldBe(disposeTask, "DisposeAsync should complete once the store recovers, not hang"); + await disposeTask; + + List<(ulong Position, bool Force)> snapshot; + lock (committed) snapshot = [..committed]; + + snapshot.ShouldNotBeEmpty(); + + // The queued positions (0..1000) drained normally, in order, during dispose; the parked + // overflow write (1001) never entered the channel, so it must not appear. + var normal = snapshot.Where(x => !x.Force).Select(x => x.Position).ToList(); + normal.ShouldBe(Enumerable.Range(0, 1001).Select(i => (ulong)i).ToList()); + + // CheckpointCommitHandler's OnDispose force-recommits whatever it last successfully stored — + // it should match the highest position that drained normally, not something stale or ahead + // of it. + var forced = snapshot.Where(x => x.Force).ToList(); + forced.ShouldNotBeEmpty(); + forced[^1].Position.ShouldBe(normal[^1]); + } finally { + storeGate.TrySetResult(); + + // Bounded so a genuinely hung disposal surfaces the original assertion failure instead + // of stalling the finally block until the test timeout. + await Task.WhenAny(disposeTask ?? handler.DisposeAsync().AsTask(), Task.Delay(TimeSpan.FromSeconds(10))); + } + } +} diff --git a/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Subscriptions/CheckpointReachedTests.cs b/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Subscriptions/CheckpointReachedTests.cs index 6734725d9..8e3622d59 100644 --- a/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Subscriptions/CheckpointReachedTests.cs +++ b/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Subscriptions/CheckpointReachedTests.cs @@ -1,12 +1,15 @@ +using System.Diagnostics; using Eventuous.KurrentDB.Producers; using Eventuous.KurrentDB.Subscriptions; using Eventuous.Producers; +using Eventuous.Subscriptions.Context; using Eventuous.Subscriptions.Registrations; using Eventuous.TestHelpers.TUnit; using Eventuous.Tests.Subscriptions.Base; using KurrentDB.Client; using Microsoft.Extensions.DependencyInjection; using Shouldly; +using EventTypeFilter = KurrentDB.Client.EventTypeFilter; // ReSharper disable MethodHasAsyncOverload @@ -107,3 +110,259 @@ protected override void GetDependencies(IServiceProvider provider) { _handler = provider.GetRequiredKeyedService(_subscriptionId); } } + +/// +/// Covers the safety half of the checkpointReached wiring: a synthetic checkpoint marker for an +/// unmatched event must never let the stored checkpoint pass a filter-matched event whose ack is +/// still pending. 's contiguous-sequence gate means the matched +/// event — produced (and therefore delivered) before any of the unmatched noise this test writes — +/// blocks every later position from that noise, including checkpoint markers, from committing past it +/// until it is acknowledged. If checkpoint markers were ever committed directly instead of flowing +/// through that same gated commit path, this test would observe the checkpoint racing past the +/// matched event's position while its ack is still held open. +/// +public class CheckpointGatedByPendingMatchTests : StoreFixture { + readonly string _subscriptionId = $"test-{Guid.NewGuid():N}"; + readonly StreamName _stream = new($"test-{Guid.NewGuid():N}"); + IProducer _producer = null!; + ICheckpointStore _checkpointStore = null!; + BlockingEventHandler _handler = null!; + + public CheckpointGatedByPendingMatchTests() : base(LogLevel.Information) { + AutoStart = false; + TypeMapper.RegisterKnownEventTypes(typeof(TestEvent).Assembly, typeof(UnmatchedEvent).Assembly); + } + + [Test] + [Category("Special cases")] + [Timeout(60_000)] + public async Task CheckpointDoesNotPassUnacknowledgedMatchedEvent(CancellationToken cancellationToken) { + // Enough unmatched events, and a small MaxSearchWindow, so the server reports plenty of + // checkpoint marker positions above the held-open matched event while we hold it. + const int unmatchedCount = 500; + + // The matched event goes in first, so it gets the lowest sequence among everything our + // filter can see: everything the subscription observes afterwards (real or synthetic) must + // wait behind its ack. (A fresh KurrentDB instance can carry a little bit of unrelated, + // unmatched system/metadata activity before this point — that's fine, and legitimately + // committable on its own, since none of it depends on our event's ack.) + await _producer.Produce(_stream, TestEvent.Create(), new(), cancellationToken: cancellationToken); + var matchedEventPosition = await GetLastAllStreamPosition(cancellationToken); + + await _producer.Produce(_stream, UnmatchedEvent.CreateMany(unmatchedCount), new(), cancellationToken: cancellationToken); + + var lastPosition = await GetLastAllStreamPosition(cancellationToken); + + // Observes CheckpointCommitHandler's "Commit" diagnostic, which fires on every Commit call + // BEFORE the contiguous-sequence gate: it proves positions reached the commit path without + // saying anything about whether they were allowed to commit. + using var commitObserver = new CommitDiagnosticCounter(_subscriptionId); + + await Start(); + + // The handler must Release() no matter how the held-open section ends, otherwise a failed + // assertion leaves the handler blocked and fixture disposal hanging until the test timeout. + try { + var started = await Task.WhenAny(_handler.Started, Task.Delay(TimeSpan.FromSeconds(30), cancellationToken)); + started.ShouldBe(_handler.Started, "The handler should have started processing the matched event before the hold-open assertion"); + + // While the handler is blocked, the matched event's own ack cannot have fired, so every + // observed Commit write is a checkpoint marker. Waiting for a marker whose position is + // strictly beyond the matched event proves the dangerous kind of marker — one that would + // overtake the pending ack if committed directly — actually entered the commit path + // during the hold. A bare count wouldn't: a fresh KurrentDB carries some unmatched + // system/metadata activity below the matched event, whose markers can be observed (and + // legitimately committed) without ever exercising the gate. + var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(30); + + while (!(commitObserver.MaxPosition is { } max && max > matchedEventPosition) && DateTime.UtcNow < deadline) { + await Task.Delay(100, cancellationToken); + } + + (commitObserver.MaxPosition ?? 0).ShouldBeGreaterThan( + matchedEventPosition, + "A checkpoint marker beyond the blocked matched event should reach the commit path while its ack is held open — none arriving means the checkpointReached wiring is broken" + ); + + // A past-the-event marker provably reached the commit path; the sequence gate must still + // have held it (and every other position) behind the unacknowledged matched event. + var stalledCheckpoint = await _checkpointStore.GetLastCheckpoint(_subscriptionId, cancellationToken); + + if (stalledCheckpoint.Position is { } stalledPosition) { + stalledPosition.ShouldBeLessThan( + matchedEventPosition, + "No position at or past the matched (still unacknowledged) event should be committed" + ); + } + } finally { + _handler.Release(); + } + + var checkpoint = await PollUntilCheckpointReaches(lastPosition, TimeSpan.FromSeconds(30), cancellationToken); + + await DisposeAsync(); + + // Only the matched event ever reached the handler... + _handler.Count.ShouldBe(1); + // ...but once it was acknowledged, the checkpoint caught up past all the unmatched events. + checkpoint.Position.ShouldNotBeNull(); + checkpoint.Position!.Value.ShouldBeGreaterThanOrEqualTo(lastPosition); + } + + async Task 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; + } + + // Polls until the checkpoint reaches minPosition, or returns the last-seen checkpoint once the + // deadline passes (the assertions in the test produce a clear failure message in that case). + async Task PollUntilCheckpointReaches(ulong minPosition, TimeSpan timeout, CancellationToken cancellationToken) { + var deadline = DateTime.UtcNow + timeout; + var checkpoint = await _checkpointStore.GetLastCheckpoint(_subscriptionId, cancellationToken); + + while (!(checkpoint.Position is { } position && position >= minPosition) && DateTime.UtcNow < deadline) { + await Task.Delay(200.Milliseconds(), cancellationToken); + checkpoint = await _checkpointStore.GetLastCheckpoint(_subscriptionId, cancellationToken); + } + + return checkpoint; + } + + protected override void SetupServices(IServiceCollection services) { + base.SetupServices(services); + services.AddProducer(); + + services.AddSubscription( + _subscriptionId, + c => c + .Configure( + o => { + // A prefix that matches only the produced TestEvent instances, with a small + // enough search window that the server still reports frequent checkpoint + // progress while scanning past the unmatched noise. + o.EventFilter = EventTypeFilter.Prefix(4, TestEvent.TypeName); + o.CheckpointInterval = 1; + + o.CheckpointCommitBatchSize = 1; + o.CheckpointCommitDelayMs = 100; + } + ) + .UseCheckpointStore() + .AddEventHandler() + ); + } + + protected override void GetDependencies(IServiceProvider provider) { + base.GetDependencies(provider); + _producer = provider.GetRequiredService(); + _checkpointStore = provider.GetRequiredKeyedService(_subscriptionId); + _handler = provider.GetRequiredKeyedService(_subscriptionId); + } +} + +/// +/// A second, distinct event type whose type name doesn't match the "test-event" prefix, used to +/// generate noise that the subscription's server-side filter excludes. +/// +[EventType(TypeName)] +// ReSharper disable once ClassNeverInstantiated.Global +public record UnmatchedEvent(int Number) { + public const string TypeName = "unmatched-event"; + + public static List CreateMany(int count) => Enumerable.Range(0, count).Select(i => new UnmatchedEvent(i)).ToList(); +} + +/// +/// Like , but holds the first event's ack open until +/// is called, so a test can observe checkpoint behaviour while a matched event's ack is still pending. +/// +public class BlockingEventHandler : BaseEventHandler { + readonly TaskCompletionSource _started = new(TaskCreationOptions.RunContinuationsAsynchronously); + readonly TaskCompletionSource _release = new(TaskCreationOptions.RunContinuationsAsynchronously); + + public int Count { get; private set; } + public Task Started => _started.Task; + + public override async ValueTask HandleEvent(IMessageConsumeContext context) { + _started.TrySetResult(); + await _release.Task.WaitAsync(context.CancellationToken); + Count++; + + return EventHandlingStatus.Success; + } + + public void Release() => _release.TrySetResult(); +} + +/// +/// Observes "Commit" diagnostic writes for one subscription, +/// tracking the highest commit position seen. The diagnostic fires on every Commit call +/// before the contiguous-sequence gate, so it observes positions reaching the commit path +/// regardless of whether they're allowed to commit. The payload (an internal CommitEvent) +/// carries the subscription id — read via reflection to keep parallel tests' commit handlers from +/// polluting the observation — and the CommitPosition, whose Position is what gets +/// tracked; if either property is renamed, nothing matches and the test's marker-arrival poll fails +/// loudly rather than silently passing. +/// +sealed class CommitDiagnosticCounter : IObserver, IObserver>, IDisposable { + readonly string _subscriptionId; + readonly IDisposable _allListeners; + readonly List _subscriptions = []; + readonly object _positionLock = new(); + + ulong? _maxPosition; + + public CommitDiagnosticCounter(string subscriptionId) { + _subscriptionId = subscriptionId; + _allListeners = DiagnosticListener.AllListeners.Subscribe(this); + } + + public ulong? MaxPosition { + get { + lock (_positionLock) return _maxPosition; + } + } + + // The commit handler's write is guarded by Diagnostic.IsEnabled(CommitOperation), so the + // subscription must carry an IsEnabled predicate that answers true. + void IObserver.OnNext(DiagnosticListener listener) { + if (listener.Name != CheckpointCommitHandler.DiagnosticName) return; + + lock (_subscriptions) _subscriptions.Add(listener.Subscribe(this, (_, _, _) => true)); + } + + void IObserver>.OnNext(KeyValuePair evt) { + if (evt.Key != CheckpointCommitHandler.CommitOperation) return; + + var payload = evt.Value; + + if (payload?.GetType().GetProperty("Id")?.GetValue(payload) is not string id || id != _subscriptionId) return; + + if (payload.GetType().GetProperty("CommitPosition")?.GetValue(payload) is not { } commitPosition) return; + + if (commitPosition.GetType().GetProperty("Position")?.GetValue(commitPosition) is not ulong position) return; + + lock (_positionLock) { + if (_maxPosition is not { } max || position > max) _maxPosition = position; + } + } + + void IObserver.OnCompleted() { } + + void IObserver.OnError(Exception error) { } + + void IObserver>.OnCompleted() { } + + void IObserver>.OnError(Exception error) { } + + public void Dispose() { + _allListeners.Dispose(); + + lock (_subscriptions) { + foreach (var subscription in _subscriptions) subscription.Dispose(); + + _subscriptions.Clear(); + } + } +} From b26ce79367cbebb060368fee1caae027a0093eb7 Mon Sep 17 00:00:00 2001 From: Alexey Zimarev Date: Wed, 15 Jul 2026 15:12:13 +0200 Subject: [PATCH 4/4] Address review: explicit null guard before reflection in CommitDiagnosticCounter Co-Authored-By: Claude Fable 5 --- .../Subscriptions/CheckpointReachedTests.cs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Subscriptions/CheckpointReachedTests.cs b/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Subscriptions/CheckpointReachedTests.cs index 8e3622d59..280d9fecd 100644 --- a/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Subscriptions/CheckpointReachedTests.cs +++ b/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Subscriptions/CheckpointReachedTests.cs @@ -335,9 +335,9 @@ void IObserver.OnNext(DiagnosticListener listener) { void IObserver>.OnNext(KeyValuePair evt) { if (evt.Key != CheckpointCommitHandler.CommitOperation) return; - var payload = evt.Value; + if (evt.Value is not { } payload) return; - if (payload?.GetType().GetProperty("Id")?.GetValue(payload) is not string id || id != _subscriptionId) return; + if (payload.GetType().GetProperty("Id")?.GetValue(payload) is not string id || id != _subscriptionId) return; if (payload.GetType().GetProperty("CommitPosition")?.GetValue(payload) is not { } commitPosition) return;