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/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>

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;
+
+/// <summary>
+/// Covers the fix for the checkpoint of a filtered <see cref="AllStreamSubscription"/> 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.
+/// </summary>
+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<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;
+    }
+
+    // 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<Checkpoint> 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<KurrentDBProducer>();
+
+        services.AddSubscription<AllStreamSubscription, AllStreamSubscriptionOptions>(
+            _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<TestCheckpointStore>()
+                .AddEventHandler<TestEventHandler>()
+        );
+    }
+
+    protected override void GetDependencies(IServiceProvider provider) {
+        base.GetDependencies(provider);
+        _producer        = provider.GetRequiredService<IProducer>();
+        _checkpointStore = provider.GetRequiredKeyedService<TestCheckpointStore>(_subscriptionId);
+        _handler         = provider.GetRequiredKeyedService<TestEventHandler>(_subscriptionId);
+    }
+}