diff --git a/src/Core/src/Eventuous.Subscriptions/Checkpoints/CheckpointCommitHandler.cs b/src/Core/src/Eventuous.Subscriptions/Checkpoints/CheckpointCommitHandler.cs index b0ed302e9..aef53c1cf 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/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/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..280d9fecd --- /dev/null +++ b/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Subscriptions/CheckpointReachedTests.cs @@ -0,0 +1,368 @@ +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 + +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); + } +} + +/// +/// 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; + + 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("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(); + } + } +}