diff --git a/src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/AllStreamSubscription.cs b/src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/AllStreamSubscription.cs index cb8d88dd..64c324ab 100644 --- a/src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/AllStreamSubscription.cs +++ b/src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/AllStreamSubscription.cs @@ -8,8 +8,6 @@ using Eventuous.Subscriptions.Filters; using Eventuous.Tools; -// ReSharper disable ConvertClosureToMethodGroup - namespace Eventuous.KurrentDB.Subscriptions; /// @@ -78,6 +76,15 @@ public AllStreamSubscription( /// internal const string CheckpointReachedMessageType = "$checkpoint-reached"; + KurrentDBClient.StreamSubscriptionResult? _subscription; + Task? _messagePump; + + // The highest $all position known to be scanned by the server in the current run: seeded from the + // stored checkpoint on (re)subscribe, advanced by every received event and checkpoint message. The + // caught-up commit must never go below it — the commit machinery is gated by sequence, not by + // position, so an older position submitted later would regress the stored checkpoint. + ulong? _lastScannedPosition; + /// /// Starts the subscription /// @@ -85,26 +92,35 @@ public AllStreamSubscription( [RequiresDynamicCode(AttrConstants.DynamicSerializationMessage)] [RequiresUnreferencedCode(AttrConstants.DynamicSerializationMessage)] protected override async ValueTask Subscribe(CancellationToken cancellationToken) { - var filterOptions = new SubscriptionFilterOptions( - Options.EventFilter ?? EventTypeFilter.ExcludeSystemEvents(), - Options.CheckpointInterval, - (_, position, ct) => HandleCheckpointReached(position, ct) - ); + var filterOptions = new SubscriptionFilterOptions(Options.EventFilter ?? EventTypeFilter.ExcludeSystemEvents(), Options.CheckpointInterval); var (_, position) = await GetCheckpoint(cancellationToken).NoContext(); + // The $all head, read before subscribing: by the time the server reports the subscription as + // caught up, everything at or below this position has provably been scanned, so it can be + // committed even if no event or checkpoint message ever surfaced it (small stores never cross + // the checkpoint interval, idle tails park up to one interval below the head). + var headPosition = await GetAllStreamHead(cancellationToken).NoContext(); + _lastScannedPosition = position; + var fromAll = GetPosition(); - Subscription = await Client.SubscribeToAllAsync( - fromAll, - (_, @event, ct) => HandleEvent(@event, ct), - Options.ResolveLinkTos, - HandleDrop, - filterOptions, - Options.Credentials, - cancellationToken - ) - .NoContext(); + var subscription = Client.SubscribeToAll(fromAll, Options.ResolveLinkTos, filterOptions, Options.Credentials, cancellationToken); + var messages = subscription.Messages.GetAsyncEnumerator(cancellationToken); + + try { + if (!await messages.MoveNextAsync().NoContext() || messages.Current is not StreamMessage.SubscriptionConfirmation) { + throw new InvalidOperationException($"Subscription {Options.SubscriptionId} to $all could not be confirmed"); + } + } catch { + await messages.DisposeAsync().NoContext(); + subscription.Dispose(); + + throw; + } + + _subscription = subscription; + _messagePump = Task.Run(() => PumpMessages(subscription, messages, headPosition, cancellationToken), CancellationToken.None); return; @@ -113,14 +129,127 @@ protected override async ValueTask Subscribe(CancellationToken cancellationToken null => FromAll.Start, _ => FromAll.After(new(position.Value, position.Value)) }; + } + + /// + /// Consumes the subscription messages, dispatching events and server checkpoints to the same + /// handlers as before, plus the caught-up notification, which the callback-based client API + /// silently discards. The message-based API is used precisely to observe that notification. + /// + [RequiresDynamicCode(AttrConstants.DynamicSerializationMessage)] + [RequiresUnreferencedCode(AttrConstants.DynamicSerializationMessage)] + async Task PumpMessages( + KurrentDBClient.StreamSubscriptionResult subscription, + IAsyncEnumerator messages, + ulong? headPosition, + CancellationToken cancellationToken + ) { + try { + while (await messages.MoveNextAsync().NoContext()) { + // Falling behind re-enters catch-up mode, making the current head the new caught-up + // commit candidate: every match at or below it is delivered before the next caught-up + // notification, exactly like the pre-subscribe head on the initial catch-up. Reading the + // head on the caught-up message instead would be unsafe — matches between the server's + // live transition and the read could still be in flight, and committing past them skips + // them on restart. Handled outside the inner try because the read is a server call: its + // failures are transport failures and must reach the outer catch, not get labelled as + // consumer errors. + if (messages.Current is StreamMessage.FellBehind) { + headPosition = await GetAllStreamHead(cancellationToken).NoContext(); + + continue; + } + + try { + switch (messages.Current) { + case StreamMessage.Event(var resolvedEvent): + _lastScannedPosition = GetContextPosition(resolvedEvent); + await HandleInternal(CreateContext(resolvedEvent, cancellationToken)).NoContext(); - Task HandleEvent(ResolvedEvent re, CancellationToken ct) - => HandleInternal(CreateContext(re, ct)).AsTask(); + break; + case StreamMessage.AllStreamCheckpointReached(var checkpointPosition): + _lastScannedPosition = checkpointPosition.CommitPosition; + await HandleCheckpointReached(checkpointPosition, cancellationToken).NoContext(); - void HandleDrop(global::KurrentDB.Client.StreamSubscription _, SubscriptionDroppedReason reason, Exception? ex) - => Dropped(KurrentDBMappings.AsDropReason(reason), ex); + break; + case StreamMessage.CaughtUp: + // The server reached the live edge, so the commit candidate — the head read + // before (re-)entering catch-up mode — has been scanned even though no + // checkpoint message reported it. The client's caught-up message carries no + // position, so that read is the best provably scanned position available; + // skip it once something newer is already known. + if (headPosition is { } head && (_lastScannedPosition is not { } lastScanned || head > lastScanned)) { + _lastScannedPosition = head; + await HandleCheckpointReached(new(head, head), cancellationToken).NoContext(); + } + + break; + } + } catch (Exception ex) when (!cancellationToken.IsCancellationRequested) { + // Handling a message failed: the transport is fine, the consumer is not — same + // classification the callback-based API gave to errors thrown by its callbacks. + // DeserializeData rethrows the raw serializer exception, so matching on exception + // types here would misattribute malformed payloads to the server. + Dropped(DropReason.SubscriptionError, ex); + + return; + } + } + + // The server ended the message stream without an error and without being asked to stop: + // treat it as a drop, so the subscription resubscribes instead of staying silently dead + if (!cancellationToken.IsCancellationRequested) { + Dropped(DropReason.ServerError, new InvalidOperationException($"Subscription {Options.SubscriptionId} message stream ended unexpectedly")); + } + } catch (Exception) when (cancellationToken.IsCancellationRequested) { + // Normal shutdown: the subscription got disposed or the token got cancelled mid-read + } catch (Exception ex) { + Dropped(DropReason.ServerError, ex); + } finally { + // Double disposal on the unsubscribe path is fine; on the dropped path this is the only + // cleanup of the underlying call before Resubscribe replaces the subscription. + await messages.DisposeAsync().NoContext(); + subscription.Dispose(); + } + } + + async Task GetAllStreamHead(CancellationToken cancellationToken) { + var lastEvent = await Client + .ReadAllAsync(Direction.Backwards, Position.End, 1, userCredentials: Options.Credentials, cancellationToken: cancellationToken) + .ToArrayAsync(cancellationToken) + .NoContext(); + + return lastEvent.Length == 0 ? null : lastEvent[0].Event.Position.CommitPosition; + } + + /// + /// Stops the subscription + /// + /// + protected override async ValueTask Unsubscribe(CancellationToken cancellationToken) { + try { + Stopping.Cancel(false); + _subscription?.Dispose(); + _subscription = null; + + if (_messagePump is { } pump) { + await Task.WhenAny(pump, Task.Delay(100, cancellationToken)).NoContext(); + _messagePump = null; + } + } catch (Exception) { + // Nothing to see here + } } + /// + /// The delivered record's own position in $all — the link's position for a resolved link event, + /// never the resolved target's. The target can be arbitrarily older than the subscription cursor + /// (a link created after the caught-up commit can point far behind the committed head), and this + /// position flows into the checkpoint on ack, so using the target's position would regress the + /// stored checkpoint. + /// + static ulong GetContextPosition(ResolvedEvent re) => (re.OriginalPosition ?? re.OriginalEvent.Position).CommitPosition; + [RequiresDynamicCode(AttrConstants.DynamicSerializationMessage)] [RequiresUnreferencedCode(AttrConstants.DynamicSerializationMessage)] MessageConsumeContext CreateContext(ResolvedEvent re, CancellationToken cancellationToken) { @@ -139,7 +268,7 @@ MessageConsumeContext CreateContext(ResolvedEvent re, CancellationToken cancella re.Event.EventStreamId, re.Event.EventNumber, re.OriginalEventNumber, - re.Event.Position.CommitPosition, + GetContextPosition(re), Sequence++, re.Event.Created, evt, @@ -150,12 +279,13 @@ 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. + /// Handles a position known to be fully scanned by the server — a reported checkpoint, or the + /// pre-subscribe head on the caught-up transition — 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)] diff --git a/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Fixtures/KurrentDBContainer.cs b/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Fixtures/KurrentDBContainer.cs index fd8c44de..d7ee16e0 100644 --- a/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Fixtures/KurrentDBContainer.cs +++ b/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Fixtures/KurrentDBContainer.cs @@ -6,7 +6,7 @@ namespace Eventuous.Tests.KurrentDB.Fixtures; public static class KurrentDBContainer { public static KurrentDbContainer Create() { var image = RuntimeInformation.ProcessArchitecture == Architecture.Arm64 - ? "kurrentplatform/kurrentdb:26.1.1-experimental-arm64-8.0-jammy" + ? "kurrentplatform/kurrentdb:26.1.1-experimental-arm64-10.0-noble" : "kurrentplatform/kurrentdb:26.1.1"; return new KurrentDbBuilder() diff --git a/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Subscriptions/CaughtUpCheckpointTests.cs b/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Subscriptions/CaughtUpCheckpointTests.cs new file mode 100644 index 00000000..a7a7f9d9 --- /dev/null +++ b/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Subscriptions/CaughtUpCheckpointTests.cs @@ -0,0 +1,204 @@ +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; +using EventTypeFilter = KurrentDB.Client.EventTypeFilter; + +// ReSharper disable MethodHasAsyncOverload + +namespace Eventuous.Tests.KurrentDB.Subscriptions; + +/// +/// Covers the caught-up commit for a filtered : on a store smaller +/// than the server's checkpoint interval, no checkpoint message is ever sent during catch-up, so a +/// subscription whose filter matches nothing would otherwise never store any checkpoint at all — a +/// restart re-scans the whole log, and consumers comparing the checkpoint to the $all head see a +/// permanent phantom lag. The caught-up transition must commit the pre-subscribe $all head instead. +/// +public class CaughtUpCheckpointOnSmallStoreTests : 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 CaughtUpCheckpointOnSmallStoreTests() : base(LogLevel.Information) { + AutoStart = false; + TypeMapper.RegisterKnownEventTypes(typeof(TestEvent).Assembly); + } + + [Test] + [Category("Special cases")] + [Timeout(60_000)] + public async Task CheckpointCommittedOnCaughtUp(CancellationToken cancellationToken) { + // Far fewer events than the checkpoint interval, so the server never sends a checkpoint + // message during catch-up: only the caught-up transition can advance the checkpoint. + const int count = 20; + + var testEvents = TestEvent.CreateMany(count); + await _producer.Produce(_stream, testEvents, new(), cancellationToken: cancellationToken); + + var lastPosition = await GetLastAllStreamPosition(cancellationToken); + + await Start(); + + 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 catching up committed a checkpoint at or past the head observed before subscribing. + 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, and a checkpoint + // interval far larger than anything this test writes, so no server checkpoint + // message can mask a missing caught-up commit. + o.EventFilter = EventTypeFilter.Prefix("definitely-does-not-match-anything"); + o.CheckpointInterval = 4096; + + 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 caught-up commit composing with StartFrom = Latest: +/// subscribing from the end of $all means the caught-up notification arrives immediately, and the +/// head observed before subscribing must be committed even though nothing is ever delivered. +/// +public class CaughtUpCheckpointFromLatestTests : 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 CaughtUpCheckpointFromLatestTests() : base(LogLevel.Information) { + AutoStart = false; + TypeMapper.RegisterKnownEventTypes(typeof(TestEvent).Assembly); + } + + [Test] + [Category("Special cases")] + [Timeout(60_000)] + public async Task CheckpointCommittedOnImmediateCaughtUp(CancellationToken cancellationToken) { + const int count = 20; + + var testEvents = TestEvent.CreateMany(count); + await _producer.Produce(_stream, testEvents, new(), cancellationToken: cancellationToken); + + var lastPosition = await GetLastAllStreamPosition(cancellationToken); + + await Start(); + + var checkpoint = await PollUntilCheckpointReaches(lastPosition, TimeSpan.FromSeconds(30), cancellationToken); + + await DisposeAsync(); + + // Starting from Latest, nothing written before the start is ever delivered... + _handler.Count.ShouldBe(0); + // ...yet the immediate caught-up transition committed the head observed before subscribing. + 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 => { + // Never-matching filter so live noise (stats, system events) can't advance the + // checkpoint through regular acks and mask a missing caught-up commit. + o.EventFilter = EventTypeFilter.Prefix("definitely-does-not-match-anything"); + o.CheckpointInterval = 4096; + o.StartFrom = InitialPosition.Latest; + + 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); + } +} diff --git a/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Subscriptions/ResolvedLinkCheckpointTests.cs b/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Subscriptions/ResolvedLinkCheckpointTests.cs new file mode 100644 index 00000000..63cfe4b5 --- /dev/null +++ b/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Subscriptions/ResolvedLinkCheckpointTests.cs @@ -0,0 +1,128 @@ +using System.Text; +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; +using EventTypeFilter = KurrentDB.Client.EventTypeFilter; + +// ReSharper disable MethodHasAsyncOverload + +namespace Eventuous.Tests.KurrentDB.Subscriptions; + +/// +/// Covers the checkpoint position used for resolved link events: the subscription cursor is the +/// link's position in $all, not the resolved target's, which can be arbitrarily older. A link +/// created after the caught-up commit and pointing at an old event would otherwise flow the +/// target's stale position into the checkpoint on ack — the commit machinery is sequence-gated, +/// not position-monotonic, so the stored checkpoint would regress below the committed head and +/// never reach the link. +/// +public class ResolvedLinkCheckpointTests : StoreFixture { + readonly string _subscriptionId = $"test-{Guid.NewGuid():N}"; + readonly StreamName _stream = new($"test-{Guid.NewGuid():N}"); + readonly string _linkStream = $"link-{Guid.NewGuid():N}"; + IProducer _producer = null!; + ICheckpointStore _checkpointStore = null!; + TestEventHandler _handler = null!; + + public ResolvedLinkCheckpointTests() : base(LogLevel.Information) { + AutoStart = false; + TypeMapper.RegisterKnownEventTypes(typeof(TestEvent).Assembly); + } + + [Test] + [Category("Special cases")] + [Timeout(60_000)] + public async Task CheckpointAdvancesToLinkPositionNotTargetPosition(CancellationToken cancellationToken) { + await _producer.Produce(_stream, TestEvent.Create(), new(), cancellationToken: cancellationToken); + + var headBeforeStart = await GetLastAllStreamPosition(cancellationToken); + + await Start(); + + // Precondition: the subscription caught up and committed at or past the pre-start head + var caughtUp = await PollUntilCheckpointReaches(headBeforeStart, TimeSpan.FromSeconds(30), cancellationToken); + caughtUp.Position.ShouldNotBeNull(); + caughtUp.Position!.Value.ShouldBeGreaterThanOrEqualTo(headBeforeStart); + + // A link written after the caught-up commit, resolving to the much older first event: its ack + // must commit the link's own position, not drag the checkpoint back to the target's + var linkResult = await Client.AppendToStreamAsync( + _linkStream, + StreamState.Any, + [new EventData(Uuid.NewUuid(), "$>", Encoding.UTF8.GetBytes($"0@{_stream}"), contentType: "application/octet-stream")], + cancellationToken: cancellationToken + ); + + var linkPosition = linkResult.LogPosition.CommitPosition; + + var checkpoint = await PollUntilCheckpointReaches(linkPosition, TimeSpan.FromSeconds(30), cancellationToken); + + await DisposeAsync(); + + // The event reached the handler directly and through the manual link at minimum — system + // projections on a fresh server emit further $>-typed links ($ce-*, $et-*) resolving to the + // same events, so the exact count is server-dependent... + _handler.Count.ShouldBeGreaterThanOrEqualTo(2); + // ...and the checkpoint advanced to the link's position instead of regressing to the target's. + checkpoint.Position.ShouldNotBeNull(); + checkpoint.Position!.Value.ShouldBeGreaterThanOrEqualTo(linkPosition); + } + + 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 => { + // Matches the produced test events and link records; the large checkpoint + // interval keeps server checkpoint messages from masking a checkpoint that + // moved through the wrong (target instead of link) position. + o.EventFilter = EventTypeFilter.Prefix(TestEvent.TypeName, "$>"); + o.CheckpointInterval = 4096; + o.ResolveLinkTos = true; + + 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); + } +}