Advance filtered AllStreamSubscription checkpoints on server checkpointReached

src/Core/src/Eventuous.Subscriptions/Checkpoints/CheckpointCommitHandler.cs

@@ -52,7 +52,12 @@ public CheckpointCommitHandler(
         _loggerFactory    = loggerFactory;
         var channel = Channel.CreateBounded<CommitPosition>(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)

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,

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;
+
+/// <summary>
+/// Pins the channel-backpressure behaviour <see cref="CheckpointCommitHandler"/> switched to
+/// (instead of throwing) when the commit queue fills up while the checkpoint store is slow: a
+/// dropped <see cref="CommitPosition"/> is poison — a gap in the sequence stalls checkpoint
+/// progression permanently — so overflow must throttle the caller, not fail the commit.
+/// </summary>
+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<ulong>          committed    = [];
+        TaskCompletionSource storeGate    = new(TaskCreationOptions.RunContinuationsAsynchronously);
+        TaskCompletionSource storeEntered = new(TaskCreationOptions.RunContinuationsAsynchronously);
+
+        async ValueTask<Checkpoint> 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<ulong> 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<Checkpoint> 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<ChannelClosedException>("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)));
+        }
+    }
+}

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) { }
 
+    /// <summary>
+    /// 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.
+    /// </summary>
+    internal const string CheckpointReachedMessageType = "$checkpoint-reached";
+
     /// <summary>
     /// Starts the subscription
     /// </summary>
@@ -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
         );
     }
 
+    /// <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();
+    }
+
     /// <summary>
     /// Returns a measure delegate for the subscription
     /// </summary>