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, 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); + } +}