From 0aec95a6e9133bbe9c364351549e032c5dc3313d Mon Sep 17 00:00:00 2001 From: Neal Mummau Date: Thu, 8 Oct 2026 22:21:57 -0400 Subject: [PATCH 1/7] fix(subscriptions): deduplicate gap metrics source reads Share source-position reads within each metrics collection while preserving per-subscription checkpoints and refreshing on the next collection. Keep custom subclasses independent unless they opt in, and floor gaps at zero when checkpoints advance beyond the observed source position. Add core and provider-key regression coverage plus real SQL Server, PostgreSQL, SQLite, and KurrentDB query/request-count tests. --- .../Diagnostics/SubscriptionGapMeasure.cs | 11 +- .../Diagnostics/SubscriptionMetrics.cs | 51 ++++-- .../IMessageSubscription.cs | 8 + .../SubscriptionRegistrationExtensions.cs | 6 +- .../SubscriptionGapMetricsTests.cs | 171 ++++++++++++++++++ .../AllPersistentSubscription.cs | 5 + .../Subscriptions/AllStreamSubscription.cs | 5 + .../StreamPersistentSubscription.cs | 5 + .../Subscriptions/StreamSubscription.cs | 5 + .../Fixtures/KurrentDBContainer.cs | 7 +- .../Metrics/SharedSourceGapMetricsTests.cs | 118 ++++++++++++ .../EndOfStreamSourceKeyTests.cs | 48 +++++ .../PostgresAllStreamSubscription.cs | 4 + .../PostgresStreamSubscription.cs | 4 + .../Subscriptions/PostgresSubscriptionBase.cs | 4 + .../Metrics/SharedSourceGapMetricsTests.cs | 133 ++++++++++++++ .../EndOfStreamSourceKeyTests.cs | 56 ++++++ .../Subscriptions/SqlSubscriptionBase.cs | 3 + .../SqlServerAllStreamSubscription.cs | 4 + .../SqlServerStreamSubscription.cs | 4 + .../SqlServerSubscriptionBase.cs | 4 + .../Metrics/SharedSourceGapMetricsTests.cs | 137 ++++++++++++++ .../EndOfStreamSourceKeyTests.cs | 55 ++++++ .../SqliteAllStreamSubscription.cs | 4 + .../Subscriptions/SqliteStreamSubscription.cs | 4 + .../Subscriptions/SqliteSubscriptionBase.cs | 4 + .../Metrics/SharedSourceGapMetricsTests.cs | 141 +++++++++++++++ .../EndOfStreamSourceKeyTests.cs | 98 ++++++++++ 28 files changed, 1081 insertions(+), 18 deletions(-) create mode 100644 src/Core/test/Eventuous.Tests.Subscriptions/SubscriptionGapMetricsTests.cs create mode 100644 src/KurrentDB/test/Eventuous.Tests.KurrentDB/Metrics/SharedSourceGapMetricsTests.cs create mode 100644 src/KurrentDB/test/Eventuous.Tests.KurrentDB/Subscriptions/EndOfStreamSourceKeyTests.cs create mode 100644 src/Postgres/test/Eventuous.Tests.Postgres/Metrics/SharedSourceGapMetricsTests.cs create mode 100644 src/Postgres/test/Eventuous.Tests.Postgres/Subscriptions/EndOfStreamSourceKeyTests.cs create mode 100644 src/SqlServer/test/Eventuous.Tests.SqlServer/Metrics/SharedSourceGapMetricsTests.cs create mode 100644 src/SqlServer/test/Eventuous.Tests.SqlServer/Subscriptions/EndOfStreamSourceKeyTests.cs create mode 100644 src/Sqlite/test/Eventuous.Tests.Sqlite/Metrics/SharedSourceGapMetricsTests.cs create mode 100644 src/Sqlite/test/Eventuous.Tests.Sqlite/Subscriptions/EndOfStreamSourceKeyTests.cs diff --git a/src/Core/src/Eventuous.Subscriptions/Diagnostics/SubscriptionGapMeasure.cs b/src/Core/src/Eventuous.Subscriptions/Diagnostics/SubscriptionGapMeasure.cs index d6449be07..681a77745 100644 --- a/src/Core/src/Eventuous.Subscriptions/Diagnostics/SubscriptionGapMeasure.cs +++ b/src/Core/src/Eventuous.Subscriptions/Diagnostics/SubscriptionGapMeasure.cs @@ -11,4 +11,13 @@ namespace Eventuous.Subscriptions.Diagnostics; [StructLayout(LayoutKind.Auto)] public readonly record struct EndOfStream(string SubscriptionId, ulong Position, DateTime Timestamp) { public static readonly EndOfStream Invalid = new("error", 0, DateTime.MinValue); -} \ No newline at end of file +} + +// Carry registration metadata on the delegate target, preserving the existing delegate-based +// constructor and DI registrations. Directly registered delegates continue to work independently. +internal sealed class RegisteredSubscriptionMeasure(string subscriptionId, object sourceKey, GetSubscriptionEndOfStream measure) { + public string SubscriptionId { get; } = subscriptionId; + public object SourceKey { get; } = sourceKey; + + public ValueTask GetEndOfStream(CancellationToken cancellationToken) => measure(cancellationToken); +} diff --git a/src/Core/src/Eventuous.Subscriptions/Diagnostics/SubscriptionMetrics.cs b/src/Core/src/Eventuous.Subscriptions/Diagnostics/SubscriptionMetrics.cs index 8deebfa10..f66cee342 100644 --- a/src/Core/src/Eventuous.Subscriptions/Diagnostics/SubscriptionMetrics.cs +++ b/src/Core/src/Eventuous.Subscriptions/Diagnostics/SubscriptionMetrics.cs @@ -1,6 +1,7 @@ // Copyright (C) Eventuous HQ OÜ. All rights reserved // Licensed under the Apache License, Version 2.0. +using System.Collections.Concurrent; using System.Diagnostics; using System.Diagnostics.Metrics; using Eventuous.Diagnostics; @@ -37,7 +38,7 @@ public sealed class SubscriptionMetrics : IWithCustomTags, IDisposable { public SubscriptionMetrics(IEnumerable measures) { var getGaps = measures.ToArray(); - Dictionary streams = new(); + ConcurrentDictionary streams = new(); ObserveMetric observeGapValues = () => ObserveGapValues(getGaps); @@ -74,11 +75,39 @@ IEnumerable> ObserveTimeValues() x => Measure(Math.Round((_checkpointMetrics.GetLastTimestamp(x.SubscriptionId) - x.Timestamp).TotalSeconds), x.SubscriptionId) ); - IEnumerable> ObserveGapValues(GetSubscriptionEndOfStream[] getEndOfStreams) - => getEndOfStreams - .Select(endOfStream => GetGap(endOfStream)) - .Where(x => x.Item1 != EndOfStream.Invalid) - .Select(x => Measure((long)(x.Item1.Position - x.Item2), x.Item1.SubscriptionId)); + IEnumerable> ObserveGapValues(GetSubscriptionEndOfStream[] getEndOfStreams) { + // Collection-local: neither successful reads nor failures survive into the next scrape. + Dictionary sources = new(); + List> values = []; + + foreach (var measure in getEndOfStreams) { + var registration = measure.Target as RegisteredSubscriptionMeasure; + EndOfStream endOfStream; + + if (registration is null) { + endOfStream = ReadEndOfStream(measure); + } else { + if (!sources.TryGetValue(registration.SourceKey, out endOfStream)) { + endOfStream = ReadEndOfStream(measure); + sources.Add(registration.SourceKey, endOfStream); + } + + if (endOfStream != EndOfStream.Invalid) + endOfStream = endOfStream with { SubscriptionId = registration.SubscriptionId }; + } + + if (endOfStream == EndOfStream.Invalid) continue; + + streams[endOfStream.SubscriptionId] = endOfStream; + var lastProcessed = _checkpointMetrics.GetLastCommitPosition(endOfStream.SubscriptionId); + // Checkpoints can advance beyond the shared tail while other sources are read. + // Such a subscription is caught up to this observation, not negatively behind. + var gap = endOfStream.Position > lastProcessed ? endOfStream.Position - lastProcessed : 0; + values.Add(Measure((long)gap, endOfStream.SubscriptionId)); + } + + return values; + } Measurement Measure(T value, string subscriptionId) where T : struct { if (_customTags.Length == 0) { @@ -106,21 +135,17 @@ TagList GetTags(SubscriptionMetricsContext ctx) { return tags; } - (EndOfStream, ulong) GetGap(GetSubscriptionEndOfStream getEndOfStream) { + EndOfStream ReadEndOfStream(GetSubscriptionEndOfStream getEndOfStream) { using var cts = new CancellationTokenSource(500); try { var t = getEndOfStream(cts.Token); - var endOfStream = t.IsCompletedSuccessfully ? t.Result : t.NoContext().GetAwaiter().GetResult(); - streams[endOfStream.SubscriptionId] = endOfStream; - var lastProcessed = _checkpointMetrics.GetLastCommitPosition(endOfStream.SubscriptionId); - - return (endOfStream, lastProcessed); + return t.IsCompletedSuccessfully ? t.Result : t.NoContext().GetAwaiter().GetResult(); } catch (Exception e) { Log.MetricCollectionFailed("Subscription Gap", e); - return (EndOfStream.Invalid, 0); + return EndOfStream.Invalid; } } } diff --git a/src/Core/src/Eventuous.Subscriptions/IMessageSubscription.cs b/src/Core/src/Eventuous.Subscriptions/IMessageSubscription.cs index 7ddd90980..504169126 100644 --- a/src/Core/src/Eventuous.Subscriptions/IMessageSubscription.cs +++ b/src/Core/src/Eventuous.Subscriptions/IMessageSubscription.cs @@ -29,4 +29,12 @@ public interface IMessageSubscription { public interface IMeasuredSubscription { GetSubscriptionEndOfStream GetMeasure(); + + /// + /// Identifies the source read by . Equal keys share one end-of-stream + /// read per metric collection. Include the store/client, schema and stream (and any other + /// setting that changes the result). Keys must be immutable and must not contain subscription IDs. + /// Null keeps independent reads for existing implementations. Keys are never exported as tags. + /// + object? EndOfStreamSourceKey => null; } diff --git a/src/Core/src/Eventuous.Subscriptions/Registrations/SubscriptionRegistrationExtensions.cs b/src/Core/src/Eventuous.Subscriptions/Registrations/SubscriptionRegistrationExtensions.cs index a9804c7fa..ac6491abc 100644 --- a/src/Core/src/Eventuous.Subscriptions/Registrations/SubscriptionRegistrationExtensions.cs +++ b/src/Core/src/Eventuous.Subscriptions/Registrations/SubscriptionRegistrationExtensions.cs @@ -75,7 +75,11 @@ Action> configureSubscription GetSubscriptionEndOfStream GetEndOfStream(IServiceProvider sp) { var subscription = GetBuilder(sp).ResolveSubscription(sp) as IMeasuredSubscription; - return subscription!.GetMeasure(); + var measure = subscription!.GetMeasure(); + + return subscription.EndOfStreamSourceKey is { } key + ? new RegisteredSubscriptionMeasure(subscriptionId, key, measure).GetEndOfStream + : measure; } } } diff --git a/src/Core/test/Eventuous.Tests.Subscriptions/SubscriptionGapMetricsTests.cs b/src/Core/test/Eventuous.Tests.Subscriptions/SubscriptionGapMetricsTests.cs new file mode 100644 index 000000000..c2ca9408c --- /dev/null +++ b/src/Core/test/Eventuous.Tests.Subscriptions/SubscriptionGapMetricsTests.cs @@ -0,0 +1,171 @@ +// Copyright (C) Eventuous HQ OÜ. All rights reserved +// Licensed under the Apache License, Version 2.0. + +using System.Diagnostics; +using System.Diagnostics.Metrics; +using Eventuous.Subscriptions; +using Eventuous.Subscriptions.Checkpoints; +using Eventuous.Subscriptions.Diagnostics; +using Eventuous.Subscriptions.Filters; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging.Abstractions; +using Shouldly; + +namespace Eventuous.Tests.Subscriptions; + +[NotInParallel] +public class SubscriptionGapMetricsTests { + [Test] + public async Task Shared_source_keeps_individual_checkpoints_and_refreshes_each_collection() { + await using var services = CreateServices(("first", "shared"), ("second", "shared"), ("other", "other")); + using var metrics = new SubscriptionMetrics(services.GetServices()); + metrics.SetCustomTags(new TagList { { "custom", "value" } }); + using var diagnostic = new DiagnosticListener(CheckpointCommitHandler.DiagnosticName); + var timestamp = DateTime.UtcNow; + diagnostic.Write(CheckpointCommitHandler.CommitOperation, new CheckpointCommitHandler.CommitEvent("first", new(3, 0, timestamp), null)); + diagnostic.Write(CheckpointCommitHandler.CommitOperation, new CheckpointCommitHandler.CommitEvent("second", new(7, 0, timestamp), null)); + + using var listener = new MeterListener(); + Dictionary values = new(); + Dictionary times = new(); + listener.InstrumentPublished = (instrument, owner) => { + if (instrument.Meter.Name == SubscriptionMetrics.MeterName) owner.EnableMeasurementEvents(instrument); + }; + listener.SetMeasurementEventCallback((instrument, value, tags, _) => { + if (instrument.Name != SubscriptionMetrics.GapCountMetricName) return; + var copied = tags.ToArray(); + copied.ShouldContain(x => x.Key == "custom" && Equals(x.Value, "value")); + values[(string)copied.Single(x => x.Key == SubscriptionMetrics.SubscriptionIdTag).Value!] = value; + }); + listener.SetMeasurementEventCallback((instrument, value, tags, _) => { + if (instrument.Name == SubscriptionMetrics.GapTimeMetricName) + times[(string)tags.ToArray().Single(x => x.Key == SubscriptionMetrics.SubscriptionIdTag).Value!] = value; + }); + listener.Start(); + listener.RecordObservableInstruments(); + + values["first"].ShouldBe(7); + values["second"].ShouldBe(3); + values["other"].ShouldBe(10); + times.Keys.ShouldContain("first"); + times.Keys.ShouldContain("second"); + times["first"].ShouldBe(times["second"]); + var subscriptions = services.GetServices().ToArray(); + subscriptions.Sum(x => x.Reads).ShouldBe(2); + + foreach (var subscription in subscriptions) subscription.Position = 20; + listener.RecordObservableInstruments(); + values["first"].ShouldBe(17); + values["second"].ShouldBe(13); + subscriptions.Sum(x => x.Reads).ShouldBe(4); + } + + [Test] + [Arguments(false)] + [Arguments(true)] + public async Task Failed_shared_source_is_read_once_and_retried_next_collection(bool throws) { + await using var services = CreateServices(("first", "shared"), ("second", "shared"), ("other", "other")); + var subscriptions = services.GetServices().ToArray(); + subscriptions[0].Fail = true; + subscriptions[0].Throw = throws; + using var metrics = new SubscriptionMetrics(services.GetServices()); + using var listener = Listen(out var values); + listener.RecordObservableInstruments(); + values.Keys.ShouldBe(new[] { "other" }); + subscriptions.Sum(x => x.Reads).ShouldBe(2); + + subscriptions[0].Fail = false; + listener.RecordObservableInstruments(); + values.Count.ShouldBe(3); + subscriptions.Sum(x => x.Reads).ShouldBe(4); + } + + [Test] + public async Task Unkeyed_subscriptions_and_direct_delegates_keep_independent_reads() { + await using var services = CreateServices(("first", null), ("second", null)); + var directReads = 0; + GetSubscriptionEndOfStream direct = _ => { + directReads++; + return new(new EndOfStream("direct", 42, DateTime.UtcNow)); + }; + using var metrics = new SubscriptionMetrics([.. services.GetServices(), direct]); + using var listener = Listen(out var values); + listener.RecordObservableInstruments(); + values.Count.ShouldBe(3); + values["direct"].ShouldBe(42); + directReads.ShouldBe(1); + services.GetServices().ShouldAllBe(x => x.Reads == 1); + ((IMeasuredSubscription)new LegacyMeasure()).EndOfStreamSourceKey.ShouldBeNull(); + } + + [Test] + public async Task Checkpoint_advancing_past_shared_tail_reports_zero_then_refreshes() { + await using var services = CreateServices(("first", "shared"), ("other", "other"), ("second", "shared")); + var subscriptions = services.GetServices().ToArray(); + using var metrics = new SubscriptionMetrics(services.GetServices()); + using var diagnostic = new DiagnosticListener(CheckpointCommitHandler.DiagnosticName); + // Processing continues while an unrelated source is read between the two shared subscriptions. + subscriptions[1].BeforeRead = () => { + foreach (var subscription in subscriptions) subscription.Position = 30; + diagnostic.Write(CheckpointCommitHandler.CommitOperation, + new CheckpointCommitHandler.CommitEvent("second", new(30, 0, DateTime.UtcNow), null)); + }; + using var listener = Listen(out var values); + listener.RecordObservableInstruments(); + values["second"].ShouldBe(0); + subscriptions.Sum(x => x.Reads).ShouldBe(2); + + subscriptions[1].BeforeRead = null; + foreach (var subscription in subscriptions) subscription.Position = 40; + listener.RecordObservableInstruments(); + values["second"].ShouldBe(10); + subscriptions.Sum(x => x.Reads).ShouldBe(4); + } + + static ServiceProvider CreateServices(params (string Id, string? Key)[] subscriptions) { + var services = new ServiceCollection(); + foreach (var (id, key) in subscriptions) + services.AddSubscription(id, builder => builder.Configure(x => x.SourceKey = key)); + return services.BuildServiceProvider(); + } + + static MeterListener Listen(out Dictionary values) { + Dictionary collected = new(); + values = collected; + var listener = new MeterListener { + InstrumentPublished = (instrument, owner) => { + if (instrument.Meter.Name == SubscriptionMetrics.MeterName && instrument.Name == SubscriptionMetrics.GapCountMetricName) + owner.EnableMeasurementEvents(instrument); + } + }; + listener.SetMeasurementEventCallback((_, value, tags, _) => + collected[(string)tags.ToArray().Single(x => x.Key == SubscriptionMetrics.SubscriptionIdTag).Value!] = value); + listener.Start(); + return listener; + } + + public record MeasureOptions : SubscriptionOptions { + public string? SourceKey { get; set; } + } + + public class MeasuredSubscription(MeasureOptions options, ConsumePipe pipe) + : EventSubscription(options, pipe, NullLoggerFactory.Instance, null), IMeasuredSubscription { + public int Reads { get; private set; } + public ulong Position { get; set; } = 10; + public bool Fail { get; set; } + public bool Throw { get; set; } + public Action? BeforeRead { get; set; } + public object? EndOfStreamSourceKey => Options.SourceKey; + protected override ValueTask Connect(SubscriptionRun run) => default; + public GetSubscriptionEndOfStream GetMeasure() => _ => { + Reads++; + BeforeRead?.Invoke(); + if (Fail && Throw) throw new InvalidOperationException("Unavailable source"); + return new(Fail ? EndOfStream.Invalid : new(SubscriptionId, Position, DateTime.UtcNow)); + }; + } + + sealed class LegacyMeasure : IMeasuredSubscription { + public GetSubscriptionEndOfStream GetMeasure() => _ => new(EndOfStream.Invalid); + } +} diff --git a/src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/AllPersistentSubscription.cs b/src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/AllPersistentSubscription.cs index 9bbd66a88..8939ebf16 100644 --- a/src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/AllPersistentSubscription.cs +++ b/src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/AllPersistentSubscription.cs @@ -120,6 +120,11 @@ CancellationToken cance /// protected override ulong GetContextStreamPosition(ResolvedEvent re) => re.Event.Position.CommitPosition; + /// + // Derived subscriptions must opt in explicitly: their measure may read a different source. + public virtual object? EndOfStreamSourceKey + => GetType() == typeof(AllPersistentSubscription) ? (typeof(AllStreamSubscriptionMeasure), Client) : null; + /// /// Returns a measure callback for the subscription /// diff --git a/src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/AllStreamSubscription.cs b/src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/AllStreamSubscription.cs index 8106abd9c..0eb12a993 100644 --- a/src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/AllStreamSubscription.cs +++ b/src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/AllStreamSubscription.cs @@ -277,6 +277,11 @@ Task HandleCheckpointReached(SubscriptionRun run, global::KurrentDB.Client.Posit return HandleInternal(run, context).AsTask(); } + /// + // Derived subscriptions must opt in explicitly: their measure may read a different source. + public virtual object? EndOfStreamSourceKey + => GetType() == typeof(AllStreamSubscription) ? (typeof(AllStreamSubscriptionMeasure), Client) : null; + /// /// Returns a measure delegate for the subscription /// diff --git a/src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/StreamPersistentSubscription.cs b/src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/StreamPersistentSubscription.cs index 2354cc9af..53083dbb9 100644 --- a/src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/StreamPersistentSubscription.cs +++ b/src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/StreamPersistentSubscription.cs @@ -109,6 +109,11 @@ CancellationToken cance /// protected override ulong GetContextStreamPosition(ResolvedEvent re) => re.Event.EventNumber; + /// + // Derived subscriptions must opt in explicitly: their measure may read a different source. + public virtual object? EndOfStreamSourceKey + => GetType() == typeof(StreamPersistentSubscription) ? (typeof(StreamSubscriptionMeasure), Client, Options.StreamName) : null; + /// public GetSubscriptionEndOfStream GetMeasure() => new StreamSubscriptionMeasure(Options.SubscriptionId, Options.StreamName, Client).GetEndOfStream; diff --git a/src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/StreamSubscription.cs b/src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/StreamSubscription.cs index d32e18b2d..03a4616c0 100644 --- a/src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/StreamSubscription.cs +++ b/src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/StreamSubscription.cs @@ -161,6 +161,11 @@ MessageConsumeContext CreateContext(SubscriptionRun run, ResolvedEvent re, Cance ); } + /// + // Derived subscriptions must opt in explicitly: their measure may read a different source. + public virtual object? EndOfStreamSourceKey + => GetType() == typeof(StreamSubscription) ? (typeof(StreamSubscriptionMeasure), Client, Options.StreamName) : null; + /// /// Returns a measure delegate for this subscription /// diff --git a/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Fixtures/KurrentDBContainer.cs b/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Fixtures/KurrentDBContainer.cs index d7ee16e0f..235f200a7 100644 --- a/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Fixtures/KurrentDBContainer.cs +++ b/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Fixtures/KurrentDBContainer.cs @@ -4,14 +4,15 @@ namespace Eventuous.Tests.KurrentDB.Fixtures; public static class KurrentDBContainer { - public static KurrentDbContainer Create() { + public static KurrentDbContainer Create() => CreateBuilder().Build(); + + public static KurrentDbBuilder CreateBuilder() { var image = RuntimeInformation.ProcessArchitecture == Architecture.Arm64 ? "kurrentplatform/kurrentdb:26.1.1-experimental-arm64-10.0-noble" : "kurrentplatform/kurrentdb:26.1.1"; return new KurrentDbBuilder() .WithImage(image) - .WithEnvironment("KURRENTDB_ENABLE_ATOM_PUB_OVER_HTTP", "true") - .Build(); + .WithEnvironment("KURRENTDB_ENABLE_ATOM_PUB_OVER_HTTP", "true"); } } diff --git a/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Metrics/SharedSourceGapMetricsTests.cs b/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Metrics/SharedSourceGapMetricsTests.cs new file mode 100644 index 000000000..77e72636d --- /dev/null +++ b/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Metrics/SharedSourceGapMetricsTests.cs @@ -0,0 +1,118 @@ +// Copyright (C) Eventuous HQ OÜ. All rights reserved +// Licensed under the Apache License, Version 2.0. + +using System.Diagnostics.Metrics; +using System.Net.Http; +using Eventuous.KurrentDB.Subscriptions; +using Eventuous.Subscriptions.Diagnostics; +using Eventuous.Subscriptions.Registrations; +using global::KurrentDB.Client; +using Microsoft.Extensions.DependencyInjection; +using Shouldly; + +namespace Eventuous.Tests.KurrentDB.Metrics; + +[NotInParallel] +public class SharedSourceGapMetricsTests { + [Test] + public async Task All_stream_subscriptions_read_shared_tail_once_per_collection(CancellationToken cancellationToken) { + // Standard projections append link events asynchronously, making the global tail nondeterministic. + await using var container = KurrentDBContainer.CreateBuilder() + .WithEnvironment("KURRENTDB_RUN_PROJECTIONS", "None") + .WithEnvironment("KURRENTDB_START_STANDARD_PROJECTIONS", "false") + .Build(); + await container.StartAsync(cancellationToken); + using var writer = new KurrentDBClient(KurrentDBClientSettings.Create(container.GetConnectionString())); + var stream = $"shared-gap-{Guid.NewGuid():N}"; + // KurrentDB global positions are byte offsets, not event counts. Use actual append positions. + var first = await Append(); + var second = await Append(); + var tail = await Append(); + + using var requests = new TailReadCounter(); + var settings = KurrentDBClientSettings.Create(container.GetConnectionString()); + settings.CreateHttpMessageHandler = () => requests; + using var client = new KurrentDBClient(settings); + var services = new ServiceCollection(); + services.AddSingleton(client); + foreach (var id in new[] { "first", "second", "caught-up" }) { + services.AddSubscription(id, builder => builder + .UseCheckpointStore()); + } + await using var provider = services.BuildServiceProvider(); + var subscriptions = provider.GetServices().ToArray(); + subscriptions.Length.ShouldBe(3); + subscriptions.Distinct().Count().ShouldBe(3); + using var metrics = new SubscriptionMetrics(provider.GetServices()); + // Do not start the subscriptions: advance deterministic checkpoints through the real commit handler. + await Commit("first", first); + await Commit("second", second); + await Commit("caught-up", tail); + + Dictionary gaps = new(); + using var listener = new MeterListener { + InstrumentPublished = (instrument, owner) => { + if (instrument.Meter.Name == SubscriptionMetrics.MeterName && instrument.Name == SubscriptionMetrics.GapCountMetricName) + owner.EnableMeasurementEvents(instrument); + } + }; + listener.SetMeasurementEventCallback((_, value, tags, _) => + gaps.Add((string)tags.ToArray().Single(x => x.Key == SubscriptionMetrics.SubscriptionIdTag).Value!, value)); + listener.Start(); + + // This client's only Streams/Read calls are the real source-position reads triggered by collection. + // Appends use a separate client; feature discovery is excluded by the exact gRPC method path. + requests.Count.ShouldBe(0); + listener.RecordObservableInstruments(); + gaps.Count.ShouldBe(3); + gaps["first"].ShouldBe(checked((long)(tail - first))); + gaps["second"].ShouldBe(checked((long)(tail - second))); + gaps["caught-up"].ShouldBe(0); + requests.Count.ShouldBe(1); + + var advanced = await Append(); + var newTail = await Append(); + newTail.ShouldBeGreaterThan(tail); + await Commit("second", advanced); + gaps.Clear(); + listener.RecordObservableInstruments(); + gaps.Count.ShouldBe(3); + gaps["first"].ShouldBe(checked((long)(newTail - first))); + gaps["second"].ShouldBe(checked((long)(newTail - advanced))); + gaps["caught-up"].ShouldBe(checked((long)(newTail - tail))); + requests.Count.ShouldBe(2); + + async Task Append() { + var result = await writer.AppendToStreamAsync( + stream, StreamState.Any, [new EventData(Uuid.NewUuid(), "shared-gap-event", "{}"u8.ToArray())], + cancellationToken: cancellationToken + ); + return result.LogPosition.CommitPosition; + } + + async Task Commit(string id, ulong position) { + var committed = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + await using var handler = new CheckpointCommitHandler(id, (checkpoint, _, _) => { + committed.TrySetResult(); + return new ValueTask(checkpoint); + }, TimeSpan.FromMilliseconds(10)); + (await handler.Commit(new(position, 0, DateTime.UtcNow), cancellationToken)).ShouldBeTrue(); + await committed.Task.WaitAsync(TimeSpan.FromSeconds(10), cancellationToken); + } + } + + sealed class TailReadCounter : DelegatingHandler { + int _count; + + public TailReadCounter() : base(new SocketsHttpHandler()) { } + + public int Count => Volatile.Read(ref _count); + + protected override async Task SendAsync(HttpRequestMessage request, CancellationToken cancellationToken) { + if (request.RequestUri?.AbsolutePath == "/event_store.client.streams.Streams/Read") + Interlocked.Increment(ref _count); + // Count transport attempts, including failures/retries, and forward the actual request unchanged. + return await base.SendAsync(request, cancellationToken); + } + } +} diff --git a/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Subscriptions/EndOfStreamSourceKeyTests.cs b/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Subscriptions/EndOfStreamSourceKeyTests.cs new file mode 100644 index 000000000..973aa26bb --- /dev/null +++ b/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Subscriptions/EndOfStreamSourceKeyTests.cs @@ -0,0 +1,48 @@ +// Copyright (C) Eventuous HQ OÜ. All rights reserved +// Licensed under the Apache License, Version 2.0. + +using Eventuous.KurrentDB.Subscriptions; +using Eventuous.Subscriptions; +using Eventuous.Subscriptions.Checkpoints; +using KurrentDBClient = global::KurrentDB.Client.KurrentDBClient; +using KurrentDBClientSettings = global::KurrentDB.Client.KurrentDBClientSettings; + +namespace Eventuous.Tests.KurrentDB.Subscriptions; + +public class EndOfStreamSourceKeyTests { + [Test] + public async Task Derived_subscriptions_keep_independent_reads() { + using var client = new KurrentDBClient(KurrentDBClientSettings.Create("esdb://localhost:2113?tls=false")); + await using var all = new DerivedAll(client); + await using var stream = new DerivedStream(client); + await using var persistentAll = new DerivedPersistentAll(client); + await using var persistentStream = new DerivedPersistentStream(client); + foreach (IMeasuredSubscription subscription in new IMeasuredSubscription[] { all, stream, persistentAll, persistentStream }) + await Assert.That(subscription.EndOfStreamSourceKey).IsNull(); + } + + sealed class DerivedAll(KurrentDBClient client) : AllStreamSubscription(client, "derived-all", new NoOpCheckpointStore(), new()); + sealed class DerivedPersistentAll(KurrentDBClient client) : AllPersistentSubscription(client, "derived-persistent-all", new()); + sealed class DerivedStream(KurrentDBClient client) : StreamSubscription( + client, new() { SubscriptionId = "derived-stream", StreamName = new("one") }, new NoOpCheckpointStore(), new()); + sealed class DerivedPersistentStream(KurrentDBClient client) : StreamPersistentSubscription( + client, new() { SubscriptionId = "derived-persistent-stream", StreamName = new("one") }, new()); + + [Test] + public async Task Keys_share_catchup_and_persistent_sources_but_separate_clients_and_streams() { + using var client = new KurrentDBClient(KurrentDBClientSettings.Create("esdb://localhost:2113?tls=false")); + using var otherClient = new KurrentDBClient(KurrentDBClientSettings.Create("esdb://localhost:2114?tls=false")); + await using var all = new AllStreamSubscription(client, "all", new NoOpCheckpointStore(), new()); + await using var persistentAll = new AllPersistentSubscription(client, "persistent-all", new()); + await using var other = new AllStreamSubscription(otherClient, "other", new NoOpCheckpointStore(), new()); + await using var stream = new StreamSubscription(client, new StreamSubscriptionOptions { SubscriptionId = "stream", StreamName = new("one") }, new NoOpCheckpointStore(), new()); + await using var persistentStream = new StreamPersistentSubscription(client, new StreamPersistentSubscriptionOptions { SubscriptionId = "persistent-stream", StreamName = new("one") }, new()); + await using var otherStream = new StreamSubscription(client, new StreamSubscriptionOptions { SubscriptionId = "other-stream", StreamName = new("two") }, new NoOpCheckpointStore(), new()); + + await Assert.That(all.EndOfStreamSourceKey).IsEqualTo(persistentAll.EndOfStreamSourceKey); + await Assert.That(all.EndOfStreamSourceKey).IsNotEqualTo(other.EndOfStreamSourceKey); + await Assert.That(all.EndOfStreamSourceKey).IsNotEqualTo(stream.EndOfStreamSourceKey); + await Assert.That(stream.EndOfStreamSourceKey).IsEqualTo(persistentStream.EndOfStreamSourceKey); + await Assert.That(stream.EndOfStreamSourceKey).IsNotEqualTo(otherStream.EndOfStreamSourceKey); + } +} diff --git a/src/Postgres/src/Eventuous.Postgresql/Subscriptions/PostgresAllStreamSubscription.cs b/src/Postgres/src/Eventuous.Postgresql/Subscriptions/PostgresAllStreamSubscription.cs index 8d2cd7ca8..44a6a3527 100644 --- a/src/Postgres/src/Eventuous.Postgresql/Subscriptions/PostgresAllStreamSubscription.cs +++ b/src/Postgres/src/Eventuous.Postgresql/Subscriptions/PostgresAllStreamSubscription.cs @@ -34,6 +34,10 @@ public class PostgresAllStreamSubscription( metaSerializer, storeOptions ) { + // Derived subscriptions may customize the query or connection; sharing must be explicit. + public override object? EndOfStreamSourceKey + => GetType() == typeof(PostgresAllStreamSubscription) ? GetEndOfStreamSourceKey(null) : null; + protected override NpgsqlCommand PrepareCommand(NpgsqlConnection connection, long start) => connection.GetCommand(Schema.ReadAllForwards) .Add("_from_position", NpgsqlDbType.Bigint, start + 1) diff --git a/src/Postgres/src/Eventuous.Postgresql/Subscriptions/PostgresStreamSubscription.cs b/src/Postgres/src/Eventuous.Postgresql/Subscriptions/PostgresStreamSubscription.cs index a8faeeaf3..0b2909ca8 100644 --- a/src/Postgres/src/Eventuous.Postgresql/Subscriptions/PostgresStreamSubscription.cs +++ b/src/Postgres/src/Eventuous.Postgresql/Subscriptions/PostgresStreamSubscription.cs @@ -33,6 +33,10 @@ public class PostgresStreamSubscription( metaSerializer, storeOptions ) { + // Derived subscriptions may customize the query or connection; sharing must be explicit. + public override object? EndOfStreamSourceKey + => GetType() == typeof(PostgresStreamSubscription) ? GetEndOfStreamSourceKey(_streamName) : null; + protected override NpgsqlCommand PrepareCommand(NpgsqlConnection connection, long start) => connection.GetCommand(Schema.ReadStreamSub) .Add("_stream_id", NpgsqlDbType.Integer, _streamId) diff --git a/src/Postgres/src/Eventuous.Postgresql/Subscriptions/PostgresSubscriptionBase.cs b/src/Postgres/src/Eventuous.Postgresql/Subscriptions/PostgresSubscriptionBase.cs index de64436d2..bd7b0051e 100644 --- a/src/Postgres/src/Eventuous.Postgresql/Subscriptions/PostgresSubscriptionBase.cs +++ b/src/Postgres/src/Eventuous.Postgresql/Subscriptions/PostgresSubscriptionBase.cs @@ -40,6 +40,10 @@ public abstract class PostgresSubscriptionBase( protected NpgsqlDataSource DataSource { get; } = dataSource; + // The concrete type also isolates custom subscriptions that change the end-of-stream query. + protected object GetEndOfStreamSourceKey(string? streamName) + => (GetType(), DataSource, Schema.Name, streamName); + protected override async ValueTask OpenConnection(CancellationToken cancellationToken) => await DataSource.OpenConnectionAsync(cancellationToken).NoContext(); diff --git a/src/Postgres/test/Eventuous.Tests.Postgres/Metrics/SharedSourceGapMetricsTests.cs b/src/Postgres/test/Eventuous.Tests.Postgres/Metrics/SharedSourceGapMetricsTests.cs new file mode 100644 index 000000000..d70c55163 --- /dev/null +++ b/src/Postgres/test/Eventuous.Tests.Postgres/Metrics/SharedSourceGapMetricsTests.cs @@ -0,0 +1,133 @@ +// Copyright (C) Eventuous HQ OÜ. All rights reserved +// Licensed under the Apache License, Version 2.0. + +using System.Diagnostics; +using System.Diagnostics.Metrics; +using Eventuous.Postgresql; +using Eventuous.Postgresql.Subscriptions; +using Eventuous.Subscriptions.Checkpoints; +using Eventuous.Subscriptions.Diagnostics; +using Eventuous.Subscriptions.Registrations; +using Eventuous.Tests.Postgres.Fixtures; +using Microsoft.Extensions.DependencyInjection; + +namespace Eventuous.Tests.Postgres.Metrics; + +[NotInParallel] +public class SharedSourceGapMetricsTests { + [Test] + public async Task All_stream_subscriptions_read_shared_tail_once_per_collection(CancellationToken cancellationToken) { + await using var container = PostgresContainer.Create(); + await container.StartAsync(cancellationToken); + var connectionString = container.GetConnectionString(); + var schema = new Schema($"metrics_{Guid.NewGuid():N}"); + var storeOptions = new PostgresStoreOptions { + ConnectionString = connectionString, Schema = schema.Name, InitializeDatabase = true + }; + await new SchemaInitializer(storeOptions).StartAsync(cancellationToken); + var mapper = new TypeMapper(); + mapper.AddType("shared-gap-event"); + using var queries = new TailQueryCounter($"select max(global_position) from {schema.Name}.messages"); + var services = new ServiceCollection(); + services.AddEventuousPostgres(connectionString, schema.Name); + foreach (var id in new[] { "first", "second", "caught-up" }) { + services.AddSubscription(id, builder => builder + .Configure(options => { + options.Schema = schema.Name; + }) + .UseCheckpointStore()); + } + await using var provider = services.BuildServiceProvider(); + var store = new PostgresStore(provider.GetRequiredService(), storeOptions, + new DefaultEventSerializer(new(), mapper)); + // PostgreSQL global positions start at one: ten events leave the tail at ten. + await Append(10); + var subscriptions = provider.GetServices().ToArray(); + await Assert.That(subscriptions.Length).IsEqualTo(3); + await Assert.That(subscriptions.Distinct().Count()).IsEqualTo(3); + using var metrics = new SubscriptionMetrics(provider.GetServices()); + // Keep polling stopped; commit deterministic positions through the production handler. + await Commit("first", 3); + await Commit("second", 7); + await Commit("caught-up", 10); + + Dictionary gaps = new(); + using var listener = new MeterListener { + InstrumentPublished = (instrument, owner) => { + if (instrument.Meter.Name == SubscriptionMetrics.MeterName && instrument.Name == SubscriptionMetrics.GapCountMetricName) + owner.EnableMeasurementEvents(instrument); + } + }; + listener.SetMeasurementEventCallback((_, value, tags, _) => + gaps.Add((string)tags.ToArray().Single(x => x.Key == SubscriptionMetrics.SubscriptionIdTag).Value!, value)); + listener.Start(); + + // Count completed production Npgsql commands, not callbacks or source-key comparisons. + // Exact SQL with a unique schema excludes setup, appends and unrelated queries. + await Assert.That(queries.Count).IsEqualTo(0); + listener.RecordObservableInstruments(); + await Assert.That(gaps.Count).IsEqualTo(3); + await Assert.That(gaps["first"]).IsEqualTo(7); + await Assert.That(gaps["second"]).IsEqualTo(3); + await Assert.That(gaps["caught-up"]).IsEqualTo(0); + await Assert.That(queries.Count).IsEqualTo(1); + await Assert.That(queries.FailedCount).IsEqualTo(0); + + await Append(10); + await Commit("second", 12); + gaps.Clear(); + listener.RecordObservableInstruments(); + await Assert.That(gaps.Count).IsEqualTo(3); + await Assert.That(gaps["first"]).IsEqualTo(17); + await Assert.That(gaps["second"]).IsEqualTo(8); + await Assert.That(gaps["caught-up"]).IsEqualTo(10); + await Assert.That(queries.Count).IsEqualTo(2); + await Assert.That(queries.FailedCount).IsEqualTo(0); + + Task Append(int count) => store.AppendEvents( + new StreamName("shared-source"), ExpectedStreamVersion.Any, + Enumerable.Range(0, count).Select(x => new NewStreamEvent(Guid.NewGuid(), new GapEvent(x), new())).ToArray(), + cancellationToken + ); + + async Task Commit(string id, ulong position) { + var committed = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + await using var handler = new CheckpointCommitHandler(id, (checkpoint, _, _) => { + committed.TrySetResult(); + return new ValueTask(checkpoint); + }, TimeSpan.FromMilliseconds(10)); + await Assert.That(await handler.Commit(new(position, 0, DateTime.UtcNow), cancellationToken)).IsTrue(); + await committed.Task.WaitAsync(TimeSpan.FromSeconds(10), cancellationToken); + } + } + + public record GapEvent(int Value); + + sealed class TailQueryCounter : IDisposable { + readonly ActivityListener _listener; + int _count; + int _failedCount; + + public TailQueryCounter(string commandText) { + _listener = new ActivityListener { + ShouldListenTo = source => source.Name == "Npgsql", + Sample = (ref ActivityCreationOptions _) => ActivitySamplingResult.AllData, + SampleUsingParentId = (ref ActivityCreationOptions _) => ActivitySamplingResult.AllData, + ActivityStopped = activity => { + if (!Equals(activity.GetTagItem("db.query.text"), commandText)) return; + // Count only completed, non-error production commands; also reject any failures. + if (activity.Status == ActivityStatusCode.Error) + Interlocked.Increment(ref _failedCount); + else + Interlocked.Increment(ref _count); + } + }; + ActivitySource.AddActivityListener(_listener); + } + + public int Count => Volatile.Read(ref _count); + public int FailedCount => Volatile.Read(ref _failedCount); + + public void Dispose() => _listener.Dispose(); + } +} diff --git a/src/Postgres/test/Eventuous.Tests.Postgres/Subscriptions/EndOfStreamSourceKeyTests.cs b/src/Postgres/test/Eventuous.Tests.Postgres/Subscriptions/EndOfStreamSourceKeyTests.cs new file mode 100644 index 000000000..01b6d27ee --- /dev/null +++ b/src/Postgres/test/Eventuous.Tests.Postgres/Subscriptions/EndOfStreamSourceKeyTests.cs @@ -0,0 +1,56 @@ +// Copyright (C) Eventuous HQ OÜ. All rights reserved +// Licensed under the Apache License, Version 2.0. + +using Eventuous.Postgresql; +using Eventuous.Postgresql.Subscriptions; +using Eventuous.Subscriptions; +using Eventuous.Subscriptions.Checkpoints; +using Npgsql; + +namespace Eventuous.Tests.Postgres.Subscriptions; + +public class EndOfStreamSourceKeyTests { + [Test] + public async Task Keys_separate_data_sources_schemas_and_streams() { + await using var source = NpgsqlDataSource.Create("Host=localhost;Database=one"); + await using var otherSource = NpgsqlDataSource.Create("Host=localhost;Database=two"); + await using var first = All(source, "first", "schema"); + await using var same = All(source, "second", "schema"); + await using var other = All(otherSource, "third", "schema"); + await using var otherSchema = All(source, "fourth", "other"); + await using var overridden = new PostgresAllStreamSubscription(source, + new() { SubscriptionId = "override", Schema = "ignored" }, new NoOpCheckpointStore(), new(), storeOptions: new("schema")); + await using var stream = Stream(source, "stream1", "one"); + await using var sameStream = Stream(source, "stream2", "one"); + await using var otherStream = Stream(source, "stream3", "two"); + + await Assert.That(first.EndOfStreamSourceKey).IsEqualTo(same.EndOfStreamSourceKey); + await Assert.That(first.EndOfStreamSourceKey).IsEqualTo(overridden.EndOfStreamSourceKey); + await Assert.That(first.EndOfStreamSourceKey).IsNotEqualTo(other.EndOfStreamSourceKey); + await Assert.That(first.EndOfStreamSourceKey).IsNotEqualTo(otherSchema.EndOfStreamSourceKey); + await Assert.That(first.EndOfStreamSourceKey).IsNotEqualTo(stream.EndOfStreamSourceKey); + await Assert.That(stream.EndOfStreamSourceKey).IsEqualTo(sameStream.EndOfStreamSourceKey); + await Assert.That(stream.EndOfStreamSourceKey).IsNotEqualTo(otherStream.EndOfStreamSourceKey); + } + + [Test] + public async Task Derived_subscriptions_keep_independent_reads() { + await using var source = NpgsqlDataSource.Create("Host=localhost;Database=one"); + await using var all = new DerivedAll(source); + await using var stream = new DerivedStream(source); + await Assert.That(((IMeasuredSubscription)all).EndOfStreamSourceKey).IsNull(); + await Assert.That(((IMeasuredSubscription)stream).EndOfStreamSourceKey).IsNull(); + } + + sealed class DerivedAll(NpgsqlDataSource source) : PostgresAllStreamSubscription( + source, new() { SubscriptionId = "derived-all" }, new NoOpCheckpointStore(), new()); + + sealed class DerivedStream(NpgsqlDataSource source) : PostgresStreamSubscription( + source, new() { SubscriptionId = "derived-stream", Stream = new("one") }, new NoOpCheckpointStore(), new()); + + static PostgresAllStreamSubscription All(NpgsqlDataSource source, string id, string schema) + => new(source, new() { SubscriptionId = id, Schema = schema }, new NoOpCheckpointStore(), new()); + + static PostgresStreamSubscription Stream(NpgsqlDataSource source, string id, string stream) + => new(source, new() { SubscriptionId = id, Schema = "schema", Stream = new(stream) }, new NoOpCheckpointStore(), new()); +} diff --git a/src/Relational/src/Eventuous.Sql.Base/Subscriptions/SqlSubscriptionBase.cs b/src/Relational/src/Eventuous.Sql.Base/Subscriptions/SqlSubscriptionBase.cs index e742bbbd3..8b659613d 100644 --- a/src/Relational/src/Eventuous.Sql.Base/Subscriptions/SqlSubscriptionBase.cs +++ b/src/Relational/src/Eventuous.Sql.Base/Subscriptions/SqlSubscriptionBase.cs @@ -327,6 +327,9 @@ MessageConsumeContext AsContext(SubscriptionRun run, PersistedEvent evt, object? [StructLayout(LayoutKind.Auto)] readonly record struct PollingResult(bool Continue, bool Retry, int ReceivedEvents); + /// Override to share end-of-stream reads for subscriptions with the same source. + public virtual object? EndOfStreamSourceKey => null; + GetSubscriptionEndOfStream IMeasuredSubscription.GetMeasure() => GetSubscriptionEndOfStream; /// diff --git a/src/SqlServer/src/Eventuous.SqlServer/Subscriptions/SqlServerAllStreamSubscription.cs b/src/SqlServer/src/Eventuous.SqlServer/Subscriptions/SqlServerAllStreamSubscription.cs index 8eca37132..e77b270af 100644 --- a/src/SqlServer/src/Eventuous.SqlServer/Subscriptions/SqlServerAllStreamSubscription.cs +++ b/src/SqlServer/src/Eventuous.SqlServer/Subscriptions/SqlServerAllStreamSubscription.cs @@ -33,6 +33,10 @@ public class SqlServerAllStreamSubscription( metaSerializer, connectionOptions ) { + // Derived subscriptions may customize the query or connection; sharing must be explicit. + public override object? EndOfStreamSourceKey + => GetType() == typeof(SqlServerAllStreamSubscription) ? GetEndOfStreamSourceKey(null) : null; + protected override SqlCommand PrepareCommand(SqlConnection connection, long start) => connection.GetStoredProcCommand(Schema.ReadAllForwards) .Add("@from_position", SqlDbType.BigInt, start + 1) diff --git a/src/SqlServer/src/Eventuous.SqlServer/Subscriptions/SqlServerStreamSubscription.cs b/src/SqlServer/src/Eventuous.SqlServer/Subscriptions/SqlServerStreamSubscription.cs index 6ef2ae5b5..07a89eace 100644 --- a/src/SqlServer/src/Eventuous.SqlServer/Subscriptions/SqlServerStreamSubscription.cs +++ b/src/SqlServer/src/Eventuous.SqlServer/Subscriptions/SqlServerStreamSubscription.cs @@ -33,6 +33,10 @@ public class SqlServerStreamSubscription( metaSerializer, connectionOptions ) { + // Derived subscriptions may customize the query or connection; sharing must be explicit. + public override object? EndOfStreamSourceKey + => GetType() == typeof(SqlServerStreamSubscription) ? GetEndOfStreamSourceKey(_streamName) : null; + protected override SqlCommand PrepareCommand(SqlConnection connection, long start) => connection.GetStoredProcCommand(Schema.ReadStreamSub) .Add("@stream_id", SqlDbType.Int, _streamId) diff --git a/src/SqlServer/src/Eventuous.SqlServer/Subscriptions/SqlServerSubscriptionBase.cs b/src/SqlServer/src/Eventuous.SqlServer/Subscriptions/SqlServerSubscriptionBase.cs index 78cdc2291..76717a71a 100644 --- a/src/SqlServer/src/Eventuous.SqlServer/Subscriptions/SqlServerSubscriptionBase.cs +++ b/src/SqlServer/src/Eventuous.SqlServer/Subscriptions/SqlServerSubscriptionBase.cs @@ -33,6 +33,10 @@ protected SqlServerSubscriptionBase( _connectionString = Ensure.NotEmptyString(connectionString); } + // The concrete type also isolates custom subscriptions that change the end-of-stream query. + protected object GetEndOfStreamSourceKey(string? streamName) + => (GetType(), _connectionString, Schema.SchemaName, streamName); + protected override async ValueTask OpenConnection(CancellationToken cancellationToken) => await ConnectionFactory.GetConnection(_connectionString, cancellationToken).NoContext(); diff --git a/src/SqlServer/test/Eventuous.Tests.SqlServer/Metrics/SharedSourceGapMetricsTests.cs b/src/SqlServer/test/Eventuous.Tests.SqlServer/Metrics/SharedSourceGapMetricsTests.cs new file mode 100644 index 000000000..31ac6361b --- /dev/null +++ b/src/SqlServer/test/Eventuous.Tests.SqlServer/Metrics/SharedSourceGapMetricsTests.cs @@ -0,0 +1,137 @@ +// Copyright (C) Eventuous HQ OÜ. All rights reserved +// Licensed under the Apache License, Version 2.0. + +using System.Diagnostics; +using System.Diagnostics.Metrics; +using Eventuous.SqlServer; +using Eventuous.SqlServer.Subscriptions; +using Eventuous.Subscriptions.Checkpoints; +using Eventuous.Subscriptions.Diagnostics; +using Eventuous.Subscriptions.Registrations; +using Eventuous.Tests.SqlServer.Fixtures; +using Microsoft.Data.SqlClient.Diagnostics; +using Microsoft.Extensions.DependencyInjection; +using Shouldly; + +namespace Eventuous.Tests.SqlServer.Metrics; + +[NotInParallel] +public class SharedSourceGapMetricsTests { + [Test] + public async Task All_stream_subscriptions_read_shared_tail_once_per_collection(CancellationToken cancellationToken) { + await using var container = SqlContainer.Create(); + await container.StartAsync(cancellationToken); + var connectionString = container.GetConnectionString(); + var schema = new Schema($"metrics_{Guid.NewGuid():N}"); + var storeOptions = new SqlServerStoreOptions { + ConnectionString = connectionString, Schema = schema.SchemaName, InitializeDatabase = true + }; + // Use the same startup retry as the metrics fixture: login readiness can precede tempdb readiness. + await new SchemaInitializer(storeOptions).StartAsync(cancellationToken); + var mapper = new TypeMapper(); + mapper.AddType("shared-gap-event"); + var store = new SqlServerStore(storeOptions, new DefaultEventSerializer(new(), mapper)); + // Global positions start at zero: appending eleven events leaves the tail at ten. + await Append(11); + + var services = new ServiceCollection(); + foreach (var id in new[] { "first", "second", "caught-up" }) { + services.AddSubscription(id, builder => builder + .Configure(options => { + options.ConnectionString = connectionString; + options.Schema = schema.SchemaName; + }) + .UseCheckpointStore()); + } + await using var provider = services.BuildServiceProvider(); + var subscriptions = provider.GetServices().ToArray(); + subscriptions.Length.ShouldBe(3); + subscriptions.Distinct().Count().ShouldBe(3); + using var metrics = new SubscriptionMetrics(provider.GetServices()); + // Keep polling stopped; commit deterministic positions through the production handler. + await Commit("first", 3); + await Commit("second", 7); + await Commit("caught-up", 10); + + Dictionary gaps = new(); + using var listener = new MeterListener { + InstrumentPublished = (instrument, owner) => { + if (instrument.Meter.Name == SubscriptionMetrics.MeterName && instrument.Name == SubscriptionMetrics.GapCountMetricName) + owner.EnableMeasurementEvents(instrument); + } + }; + listener.SetMeasurementEventCallback((_, value, tags, _) => + gaps.Add((string)tags.ToArray().Single(x => x.Key == SubscriptionMetrics.SubscriptionIdTag).Value!, value)); + listener.Start(); + + // Count successful real SqlClient executions, not callbacks or source-key comparisons. + // The unique schema excludes setup, appends and unrelated tests' SQL commands. + using var queries = new TailQueryCounter($"SELECT MAX(GlobalPosition) FROM {schema.SchemaName}.Messages"); + listener.RecordObservableInstruments(); + gaps.Count.ShouldBe(3); + gaps["first"].ShouldBe(7); + gaps["second"].ShouldBe(3); + gaps["caught-up"].ShouldBe(0); + queries.Count.ShouldBe(1); + + await Append(10); + await Commit("second", 12); + gaps.Clear(); + listener.RecordObservableInstruments(); + gaps.Count.ShouldBe(3); + gaps["first"].ShouldBe(17); + gaps["second"].ShouldBe(8); + gaps["caught-up"].ShouldBe(10); + queries.Count.ShouldBe(2); + + Task Append(int count) => store.AppendEvents( + new StreamName("shared-source"), ExpectedStreamVersion.Any, + Enumerable.Range(0, count).Select(x => new NewStreamEvent(Guid.NewGuid(), new GapEvent(x), new())).ToArray(), + cancellationToken + ); + + async Task Commit(string id, ulong position) { + var committed = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + await using var handler = new CheckpointCommitHandler(id, (checkpoint, _, _) => { + committed.TrySetResult(); + return new ValueTask(checkpoint); + }, TimeSpan.FromMilliseconds(10)); + (await handler.Commit(new(position, 0, DateTime.UtcNow), cancellationToken)).ShouldBeTrue(); + await committed.Task.WaitAsync(TimeSpan.FromSeconds(10), cancellationToken); + } + } + + public record GapEvent(int Value); + + sealed class TailQueryCounter : IObserver, IObserver>, IDisposable { + readonly string _commandText; + readonly IDisposable _allListeners; + readonly List _subscriptions = []; + int _count; + + public TailQueryCounter(string commandText) { + _commandText = commandText; + _allListeners = DiagnosticListener.AllListeners.Subscribe(this); + } + + public int Count => Volatile.Read(ref _count); + + public void OnNext(DiagnosticListener listener) { + if (listener.Name == "SqlClientDiagnosticListener") + _subscriptions.Add(listener.Subscribe(this, name => name == "Microsoft.Data.SqlClient.WriteCommandAfter")); + } + + public void OnNext(KeyValuePair value) { + if (value.Value is SqlClientCommandAfter after && after.Command.CommandText == _commandText) + Interlocked.Increment(ref _count); + } + + public void OnCompleted() { } + public void OnError(Exception error) { } + + public void Dispose() { + _allListeners.Dispose(); + foreach (var subscription in _subscriptions) subscription.Dispose(); + } + } +} diff --git a/src/SqlServer/test/Eventuous.Tests.SqlServer/Subscriptions/EndOfStreamSourceKeyTests.cs b/src/SqlServer/test/Eventuous.Tests.SqlServer/Subscriptions/EndOfStreamSourceKeyTests.cs new file mode 100644 index 000000000..594e1e442 --- /dev/null +++ b/src/SqlServer/test/Eventuous.Tests.SqlServer/Subscriptions/EndOfStreamSourceKeyTests.cs @@ -0,0 +1,55 @@ +// Copyright (C) Eventuous HQ OÜ. All rights reserved +// Licensed under the Apache License, Version 2.0. + +using Eventuous.SqlServer.Projections; +using Eventuous.SqlServer.Subscriptions; +using Eventuous.Subscriptions; +using Eventuous.Subscriptions.Checkpoints; +using Shouldly; + +namespace Eventuous.Tests.SqlServer.Subscriptions; + +public class EndOfStreamSourceKeyTests { + [Test] + public async Task Keys_use_effective_connection_schema_and_stream() { + await using var first = All("first", "connection", "schema"); + await using var same = All("second", "connection", "schema"); + await using var otherConnection = All("third", "other", "schema"); + await using var otherSchema = All("fourth", "connection", "other"); + await using var stream = Stream("stream1", "one"); + await using var sameStream = Stream("stream2", "one"); + await using var otherStream = Stream("stream3", "two"); + await using var overridden = new SqlServerAllStreamSubscription( + new() { SubscriptionId = "override", ConnectionString = "ignored", Schema = "ignored" }, + new NoOpCheckpointStore(), new(), connectionOptions: new SqlServerConnectionOptions("connection", "schema") + ); + + first.EndOfStreamSourceKey.ShouldBe(same.EndOfStreamSourceKey); + first.EndOfStreamSourceKey.ShouldBe(overridden.EndOfStreamSourceKey); + first.EndOfStreamSourceKey.ShouldNotBe(otherConnection.EndOfStreamSourceKey); + first.EndOfStreamSourceKey.ShouldNotBe(otherSchema.EndOfStreamSourceKey); + first.EndOfStreamSourceKey.ShouldNotBe(stream.EndOfStreamSourceKey); + stream.EndOfStreamSourceKey.ShouldBe(sameStream.EndOfStreamSourceKey); + stream.EndOfStreamSourceKey.ShouldNotBe(otherStream.EndOfStreamSourceKey); + } + + [Test] + public async Task Derived_subscriptions_keep_independent_reads() { + await using var all = new DerivedAll(); + await using var stream = new DerivedStream(); + ((IMeasuredSubscription)all).EndOfStreamSourceKey.ShouldBeNull(); + ((IMeasuredSubscription)stream).EndOfStreamSourceKey.ShouldBeNull(); + } + + sealed class DerivedAll() : SqlServerAllStreamSubscription( + new() { SubscriptionId = "derived-all", ConnectionString = "connection" }, new NoOpCheckpointStore(), new()); + + sealed class DerivedStream() : SqlServerStreamSubscription( + new() { SubscriptionId = "derived-stream", ConnectionString = "connection", Stream = new("one") }, new NoOpCheckpointStore(), new()); + + static SqlServerAllStreamSubscription All(string id, string connection, string schema) + => new(new() { SubscriptionId = id, ConnectionString = connection, Schema = schema }, new NoOpCheckpointStore(), new()); + + static SqlServerStreamSubscription Stream(string id, string stream) + => new(new() { SubscriptionId = id, ConnectionString = "connection", Schema = "schema", Stream = new(stream) }, new NoOpCheckpointStore(), new()); +} diff --git a/src/Sqlite/src/Eventuous.Sqlite/Subscriptions/SqliteAllStreamSubscription.cs b/src/Sqlite/src/Eventuous.Sqlite/Subscriptions/SqliteAllStreamSubscription.cs index 88cee7e27..781f1c879 100644 --- a/src/Sqlite/src/Eventuous.Sqlite/Subscriptions/SqliteAllStreamSubscription.cs +++ b/src/Sqlite/src/Eventuous.Sqlite/Subscriptions/SqliteAllStreamSubscription.cs @@ -33,6 +33,10 @@ public class SqliteAllStreamSubscription( metaSerializer, connectionOptions ) { + // Derived subscriptions may customize the query or connection; sharing must be explicit. + public override object? EndOfStreamSourceKey + => GetType() == typeof(SqliteAllStreamSubscription) ? GetEndOfStreamSourceKey(null) : null; + protected override SqliteCommand PrepareCommand(SqliteConnection connection, long start) => connection.GetTextCommand( $""" diff --git a/src/Sqlite/src/Eventuous.Sqlite/Subscriptions/SqliteStreamSubscription.cs b/src/Sqlite/src/Eventuous.Sqlite/Subscriptions/SqliteStreamSubscription.cs index 8edeb79af..a43449070 100644 --- a/src/Sqlite/src/Eventuous.Sqlite/Subscriptions/SqliteStreamSubscription.cs +++ b/src/Sqlite/src/Eventuous.Sqlite/Subscriptions/SqliteStreamSubscription.cs @@ -33,6 +33,10 @@ public class SqliteStreamSubscription( metaSerializer, connectionOptions ) { + // Derived subscriptions may customize the query or connection; sharing must be explicit. + public override object? EndOfStreamSourceKey + => GetType() == typeof(SqliteStreamSubscription) ? GetEndOfStreamSourceKey(_streamName) : null; + protected override SqliteCommand PrepareCommand(SqliteConnection connection, long start) => connection.GetTextCommand( $""" diff --git a/src/Sqlite/src/Eventuous.Sqlite/Subscriptions/SqliteSubscriptionBase.cs b/src/Sqlite/src/Eventuous.Sqlite/Subscriptions/SqliteSubscriptionBase.cs index c4ceaa28a..736c02478 100644 --- a/src/Sqlite/src/Eventuous.Sqlite/Subscriptions/SqliteSubscriptionBase.cs +++ b/src/Sqlite/src/Eventuous.Sqlite/Subscriptions/SqliteSubscriptionBase.cs @@ -33,6 +33,10 @@ protected SqliteSubscriptionBase( _connectionString = Ensure.NotEmptyString(connectionString); } + // The concrete type also isolates custom subscriptions that change the end-of-stream query. + protected object GetEndOfStreamSourceKey(string? streamName) + => (GetType(), _connectionString, Schema.MessagesTable, streamName); + protected override async ValueTask OpenConnection(CancellationToken cancellationToken) => await ConnectionFactory.GetConnection(_connectionString, cancellationToken).NoContext(); diff --git a/src/Sqlite/test/Eventuous.Tests.Sqlite/Metrics/SharedSourceGapMetricsTests.cs b/src/Sqlite/test/Eventuous.Tests.Sqlite/Metrics/SharedSourceGapMetricsTests.cs new file mode 100644 index 000000000..f68edcdf0 --- /dev/null +++ b/src/Sqlite/test/Eventuous.Tests.Sqlite/Metrics/SharedSourceGapMetricsTests.cs @@ -0,0 +1,141 @@ +// Copyright (C) Eventuous HQ OÜ. All rights reserved +// Licensed under the Apache License, Version 2.0. + +using System.Diagnostics.Metrics; +using Eventuous.Sqlite; +using Eventuous.Sqlite.Subscriptions; +using Eventuous.Subscriptions.Checkpoints; +using Eventuous.Subscriptions.Diagnostics; +using Eventuous.Subscriptions.Registrations; +using Microsoft.Data.Sqlite; +using Microsoft.Extensions.DependencyInjection; +using Shouldly; +using SQLitePCL; + +namespace Eventuous.Tests.Sqlite.Metrics; + +[NotInParallel] +public class SharedSourceGapMetricsTests { + [Test] + public async Task All_stream_subscriptions_read_shared_tail_once_per_collection(CancellationToken cancellationToken) { + var schema = new Schema($"metrics_{Guid.NewGuid():N}"); + // URI mode=memory keeps this a real in-memory database while allowing pooling. + // Mode=Memory and Data Source=:memory: disable Microsoft.Data.Sqlite pooling. + // An isolated pool lets the native profiler observe the actual production connection, + // without overriding subscriptions or adding a production connection factory hook. + var connectionString = $"Data Source=file:{schema.SchemaName}?mode=memory&cache=shared;Pooling=True"; + await using var keeper = new SqliteConnection(connectionString + ";Pooling=False"); + await keeper.OpenAsync(cancellationToken); + await using (var command = keeper.CreateCommand()) { + command.CommandText = "SELECT file FROM pragma_database_list WHERE name = 'main'"; + (await command.ExecuteScalarAsync(cancellationToken)).ShouldBe(""); + } + using var queries = new TailQueryCounter(connectionString, $"SELECT MAX(global_position) FROM {schema.MessagesTable}"); + await schema.CreateSchema(connectionString, null, cancellationToken); + var mapper = new TypeMapper(); + mapper.AddType("shared-gap-event"); + var store = new SqliteStore(new() { ConnectionString = connectionString, Schema = schema.SchemaName }, new DefaultEventSerializer(new(), mapper)); + // SQLite global positions start at one: ten events leave the tail at ten. + await Append(10); + + var services = new ServiceCollection(); + foreach (var id in new[] { "first", "second", "caught-up" }) { + services.AddSubscription(id, builder => builder + .Configure(options => { + options.ConnectionString = connectionString; + options.Schema = schema.SchemaName; + }) + .UseCheckpointStore()); + } + await using var provider = services.BuildServiceProvider(); + var subscriptions = provider.GetServices().ToArray(); + subscriptions.Length.ShouldBe(3); + subscriptions.Distinct().Count().ShouldBe(3); + using var metrics = new SubscriptionMetrics(provider.GetServices()); + // Keep polling stopped; commit deterministic positions through the production handler. + await Commit("first", 3); + await Commit("second", 7); + await Commit("caught-up", 10); + + Dictionary gaps = new(); + using var listener = new MeterListener { + InstrumentPublished = (instrument, owner) => { + if (instrument.Meter.Name == SubscriptionMetrics.MeterName && instrument.Name == SubscriptionMetrics.GapCountMetricName) + owner.EnableMeasurementEvents(instrument); + } + }; + listener.SetMeasurementEventCallback((_, value, tags, _) => + gaps.Add((string)tags.ToArray().Single(x => x.Key == SubscriptionMetrics.SubscriptionIdTag).Value!, value)); + listener.Start(); + + queries.Count.ShouldBe(0); + listener.RecordObservableInstruments(); + gaps.Count.ShouldBe(3); + gaps["first"].ShouldBe(7); + gaps["second"].ShouldBe(3); + gaps["caught-up"].ShouldBe(0); + queries.Count.ShouldBe(1); + + await Append(10); + await Commit("second", 12); + queries.Count.ShouldBe(1); + gaps.Clear(); + listener.RecordObservableInstruments(); + gaps.Count.ShouldBe(3); + gaps["first"].ShouldBe(17); + gaps["second"].ShouldBe(8); + gaps["caught-up"].ShouldBe(10); + queries.Count.ShouldBe(2); + + Task Append(int count) => store.AppendEvents( + new StreamName("shared-source"), ExpectedStreamVersion.Any, + Enumerable.Range(0, count).Select(x => new NewStreamEvent(Guid.NewGuid(), new GapEvent(x), new())).ToArray(), + cancellationToken + ); + + async Task Commit(string id, ulong position) { + var committed = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + await using var handler = new CheckpointCommitHandler(id, (checkpoint, _, _) => { + committed.TrySetResult(); + return new ValueTask(checkpoint); + }, TimeSpan.FromMilliseconds(10)); + (await handler.Commit(new(position, 0, DateTime.UtcNow), cancellationToken)).ShouldBeTrue(); + await committed.Task.WaitAsync(TimeSpan.FromSeconds(10), cancellationToken); + } + } + + public record GapEvent(int Value); + + sealed class TailQueryCounter : IDisposable { + readonly SqliteConnection _connection; + readonly sqlite3 _handle; + int _count; + + public TailQueryCounter(string connectionString, string commandText) { + _connection = new(connectionString); + _connection.Open(); + _handle = _connection.Handle!; + // SQLitePCLRaw is already a transitive dependency. The native profile callback + // fires on statement execution completion, not preparation or source callbacks. + raw.sqlite3_profile(_handle, (strdelegate_profile)((_, sql, _) => { + if (sql == commandText) Interlocked.Increment(ref _count); + }), null); + // Production opens/closes connections sequentially and reuses this native handle. + _connection.Close(); + } + + public int Count => Volatile.Read(ref _count); + + public void Dispose() { + try { + _connection.Open(); + _connection.Handle.ShouldBeSameAs(_handle); + raw.sqlite3_profile(_handle, (strdelegate_profile)null!, null); + } finally { + // Clear only this test's uniquely named pool, never other tests' pools. + SqliteConnection.ClearPool(_connection); + _connection.Dispose(); + } + } + } +} diff --git a/src/Sqlite/test/Eventuous.Tests.Sqlite/Subscriptions/EndOfStreamSourceKeyTests.cs b/src/Sqlite/test/Eventuous.Tests.Sqlite/Subscriptions/EndOfStreamSourceKeyTests.cs new file mode 100644 index 000000000..e68020e82 --- /dev/null +++ b/src/Sqlite/test/Eventuous.Tests.Sqlite/Subscriptions/EndOfStreamSourceKeyTests.cs @@ -0,0 +1,98 @@ +// Copyright (C) Eventuous HQ OÜ. All rights reserved +// Licensed under the Apache License, Version 2.0. + +using System.Diagnostics.Metrics; +using Eventuous.Subscriptions; +using Eventuous.Subscriptions.Diagnostics; +using Eventuous.Subscriptions.Filters; +using Microsoft.Data.Sqlite; +using Microsoft.Extensions.DependencyInjection; +using Eventuous.Sqlite.Projections; +using Eventuous.Sqlite.Subscriptions; +using Eventuous.Subscriptions.Checkpoints; +using Shouldly; + +namespace Eventuous.Tests.Sqlite.Subscriptions; + +public class EndOfStreamSourceKeyTests { + [Test] + public async Task Keys_use_effective_connection_schema_and_stream() { + await using var first = All("first", "connection", "schema"); + await using var same = All("second", "connection", "schema"); + await using var otherConnection = All("third", "other", "schema"); + await using var otherSchema = All("fourth", "connection", "other"); + await using var stream = Stream("stream1", "one"); + await using var sameStream = Stream("stream2", "one"); + await using var otherStream = Stream("stream3", "two"); + await using var overridden = new SqliteAllStreamSubscription( + new() { SubscriptionId = "override", ConnectionString = "ignored", Schema = "ignored" }, + new NoOpCheckpointStore(), new(), connectionOptions: new SqliteConnectionOptions("connection", "schema") + ); + + first.EndOfStreamSourceKey.ShouldBe(same.EndOfStreamSourceKey); + first.EndOfStreamSourceKey.ShouldBe(overridden.EndOfStreamSourceKey); + first.EndOfStreamSourceKey.ShouldNotBe(otherConnection.EndOfStreamSourceKey); + first.EndOfStreamSourceKey.ShouldNotBe(otherSchema.EndOfStreamSourceKey); + first.EndOfStreamSourceKey.ShouldNotBe(stream.EndOfStreamSourceKey); + stream.EndOfStreamSourceKey.ShouldBe(sameStream.EndOfStreamSourceKey); + stream.EndOfStreamSourceKey.ShouldNotBe(otherStream.EndOfStreamSourceKey); + } + + [Test] + [NotInParallel] + public async Task Existing_custom_queries_keep_independent_measurements() { + var registrations = new ServiceCollection(); + registrations.AddSingleton(); + foreach (var id in new[] { "custom-a", "custom-b" }) + registrations.AddSubscription(id, builder => + builder.Configure(options => { options.ConnectionString = "Data Source=:memory:"; options.Schema = "events"; })); + await using var services = registrations.BuildServiceProvider(); + using var metrics = new SubscriptionMetrics(services.GetServices()); + Dictionary values = new(); + using var listener = new MeterListener { + InstrumentPublished = (instrument, owner) => { + if (instrument.Meter.Name == SubscriptionMetrics.MeterName && instrument.Name == SubscriptionMetrics.GapCountMetricName) + owner.EnableMeasurementEvents(instrument); + } + }; + listener.SetMeasurementEventCallback((_, value, tags, _) => + values[(string)tags.ToArray().Single(x => x.Key == SubscriptionMetrics.SubscriptionIdTag).Value!] = value); + listener.Start(); + listener.RecordObservableInstruments(); + values["custom-a"].ShouldBe(10); + values["custom-b"].ShouldBe(20); + } + + [Test] + public async Task Derived_subscriptions_are_independent_unless_they_opt_in() { + await using var all = new CustomQuery(new() { SubscriptionId = "custom-a", ConnectionString = "connection" }, new NoOpCheckpointStore(), new()); + await using var stream = new DerivedStream(); + await using var optedIn = new OptedInQuery(new() { SubscriptionId = "custom-b", ConnectionString = "connection" }, new NoOpCheckpointStore(), new()); + ((IMeasuredSubscription)all).EndOfStreamSourceKey.ShouldBeNull(); + ((IMeasuredSubscription)stream).EndOfStreamSourceKey.ShouldBeNull(); + ((IMeasuredSubscription)optedIn).EndOfStreamSourceKey.ShouldBe("explicit-custom-source"); + } + + public class CustomQuery(SqliteAllStreamSubscriptionOptions options, ICheckpointStore checkpoints, ConsumePipe pipe) + : SqliteAllStreamSubscription(options, checkpoints, pipe) { + protected override SqliteCommand PrepareEndOfStreamCommand(SqliteConnection connection) { + var command = connection.CreateCommand(); + command.CommandText = SubscriptionId == "custom-a" ? "SELECT 10" : "SELECT 20"; + return command; + } + } + + sealed class OptedInQuery(SqliteAllStreamSubscriptionOptions options, ICheckpointStore checkpoints, ConsumePipe pipe) + : CustomQuery(options, checkpoints, pipe) { + public override object EndOfStreamSourceKey => "explicit-custom-source"; + } + + sealed class DerivedStream() : SqliteStreamSubscription( + new() { SubscriptionId = "derived", ConnectionString = "connection", Stream = new("stream") }, new NoOpCheckpointStore(), new()); + + static SqliteAllStreamSubscription All(string id, string connection, string schema) + => new(new() { SubscriptionId = id, ConnectionString = connection, Schema = schema }, new NoOpCheckpointStore(), new()); + + static SqliteStreamSubscription Stream(string id, string stream) + => new(new() { SubscriptionId = id, ConnectionString = "connection", Schema = "schema", Stream = new(stream) }, new NoOpCheckpointStore(), new()); +} From 3f717b38f9caaeec8dcbf270af6334646d78df7d Mon Sep 17 00:00:00 2001 From: Neal Mummau Date: Thu, 8 Oct 2026 23:32:13 -0400 Subject: [PATCH 2/7] test(subscriptions): align gap metrics helpers with async conventions --- .../Metrics/SharedSourceGapMetricsTests.cs | 3 +- .../Metrics/SharedSourceGapMetricsTests.cs | 32 +++++++++++++------ .../EndOfStreamSourceKeyTests.cs | 8 ++--- 3 files changed, 28 insertions(+), 15 deletions(-) diff --git a/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Metrics/SharedSourceGapMetricsTests.cs b/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Metrics/SharedSourceGapMetricsTests.cs index 77e72636d..cf9be696a 100644 --- a/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Metrics/SharedSourceGapMetricsTests.cs +++ b/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Metrics/SharedSourceGapMetricsTests.cs @@ -6,6 +6,7 @@ using Eventuous.KurrentDB.Subscriptions; using Eventuous.Subscriptions.Diagnostics; using Eventuous.Subscriptions.Registrations; +using Eventuous.Tools; using global::KurrentDB.Client; using Microsoft.Extensions.DependencyInjection; using Shouldly; @@ -112,7 +113,7 @@ protected override async Task SendAsync(HttpRequestMessage if (request.RequestUri?.AbsolutePath == "/event_store.client.streams.Streams/Read") Interlocked.Increment(ref _count); // Count transport attempts, including failures/retries, and forward the actual request unchanged. - return await base.SendAsync(request, cancellationToken); + return await base.SendAsync(request, cancellationToken).NoContext(); } } } diff --git a/src/Sqlite/test/Eventuous.Tests.Sqlite/Metrics/SharedSourceGapMetricsTests.cs b/src/Sqlite/test/Eventuous.Tests.Sqlite/Metrics/SharedSourceGapMetricsTests.cs index f68edcdf0..af4c079c0 100644 --- a/src/Sqlite/test/Eventuous.Tests.Sqlite/Metrics/SharedSourceGapMetricsTests.cs +++ b/src/Sqlite/test/Eventuous.Tests.Sqlite/Metrics/SharedSourceGapMetricsTests.cs @@ -30,7 +30,7 @@ public async Task All_stream_subscriptions_read_shared_tail_once_per_collection( command.CommandText = "SELECT file FROM pragma_database_list WHERE name = 'main'"; (await command.ExecuteScalarAsync(cancellationToken)).ShouldBe(""); } - using var queries = new TailQueryCounter(connectionString, $"SELECT MAX(global_position) FROM {schema.MessagesTable}"); + await using var queries = await TailQueryCounter.Create(connectionString, $"SELECT MAX(global_position) FROM {schema.MessagesTable}", cancellationToken); await schema.CreateSchema(connectionString, null, cancellationToken); var mapper = new TypeMapper(); mapper.AddType("shared-gap-event"); @@ -106,35 +106,47 @@ async Task Commit(string id, ulong position) { public record GapEvent(int Value); - sealed class TailQueryCounter : IDisposable { + sealed class TailQueryCounter : IAsyncDisposable { readonly SqliteConnection _connection; readonly sqlite3 _handle; int _count; - public TailQueryCounter(string connectionString, string commandText) { - _connection = new(connectionString); - _connection.Open(); + TailQueryCounter(SqliteConnection connection, string commandText) { + _connection = connection; _handle = _connection.Handle!; // SQLitePCLRaw is already a transitive dependency. The native profile callback // fires on statement execution completion, not preparation or source callbacks. raw.sqlite3_profile(_handle, (strdelegate_profile)((_, sql, _) => { if (sql == commandText) Interlocked.Increment(ref _count); }), null); - // Production opens/closes connections sequentially and reuses this native handle. - _connection.Close(); + } + + public static async Task Create(string connectionString, string commandText, CancellationToken cancellationToken) { + var connection = new SqliteConnection(connectionString); + try { + await connection.OpenAsync(cancellationToken); + var counter = new TailQueryCounter(connection, commandText); + // Production opens/closes connections sequentially and reuses this native handle. + await connection.CloseAsync(); + return counter; + } catch { + SqliteConnection.ClearPool(connection); + await connection.DisposeAsync(); + throw; + } } public int Count => Volatile.Read(ref _count); - public void Dispose() { + public async ValueTask DisposeAsync() { try { - _connection.Open(); + await _connection.OpenAsync(); _connection.Handle.ShouldBeSameAs(_handle); raw.sqlite3_profile(_handle, (strdelegate_profile)null!, null); } finally { // Clear only this test's uniquely named pool, never other tests' pools. SqliteConnection.ClearPool(_connection); - _connection.Dispose(); + await _connection.DisposeAsync(); } } } diff --git a/src/Sqlite/test/Eventuous.Tests.Sqlite/Subscriptions/EndOfStreamSourceKeyTests.cs b/src/Sqlite/test/Eventuous.Tests.Sqlite/Subscriptions/EndOfStreamSourceKeyTests.cs index e68020e82..7fc047b54 100644 --- a/src/Sqlite/test/Eventuous.Tests.Sqlite/Subscriptions/EndOfStreamSourceKeyTests.cs +++ b/src/Sqlite/test/Eventuous.Tests.Sqlite/Subscriptions/EndOfStreamSourceKeyTests.cs @@ -2,14 +2,14 @@ // Licensed under the Apache License, Version 2.0. using System.Diagnostics.Metrics; -using Eventuous.Subscriptions; +using Eventuous.Sqlite.Projections; +using Eventuous.Sqlite.Subscriptions; +using Eventuous.Subscriptions.Checkpoints; using Eventuous.Subscriptions.Diagnostics; using Eventuous.Subscriptions.Filters; +using Eventuous.Subscriptions; using Microsoft.Data.Sqlite; using Microsoft.Extensions.DependencyInjection; -using Eventuous.Sqlite.Projections; -using Eventuous.Sqlite.Subscriptions; -using Eventuous.Subscriptions.Checkpoints; using Shouldly; namespace Eventuous.Tests.Sqlite.Subscriptions; From ebe3c930847177cd25d6954f6bedc6bc23e436ed Mon Sep 17 00:00:00 2001 From: Neal Mummau Date: Thu, 8 Oct 2026 23:32:16 -0400 Subject: [PATCH 3/7] fix(subscriptions): use runtime subscription IDs in gap metrics --- .../SubscriptionRegistrationExtensions.cs | 7 +++--- .../SubscriptionGapMetricsTests.cs | 24 +++++++++++++++++++ 2 files changed, 28 insertions(+), 3 deletions(-) diff --git a/src/Core/src/Eventuous.Subscriptions/Registrations/SubscriptionRegistrationExtensions.cs b/src/Core/src/Eventuous.Subscriptions/Registrations/SubscriptionRegistrationExtensions.cs index ac6491abc..473c86f09 100644 --- a/src/Core/src/Eventuous.Subscriptions/Registrations/SubscriptionRegistrationExtensions.cs +++ b/src/Core/src/Eventuous.Subscriptions/Registrations/SubscriptionRegistrationExtensions.cs @@ -73,12 +73,13 @@ Action> configureSubscription SubscriptionBuilder GetBuilder(IServiceProvider sp) => sp.GetSubscriptionBuilder(subscriptionId); GetSubscriptionEndOfStream GetEndOfStream(IServiceProvider sp) { - var subscription = GetBuilder(sp).ResolveSubscription(sp) as IMeasuredSubscription; + var resolved = GetBuilder(sp).ResolveSubscription(sp); + var subscription = (IMeasuredSubscription)resolved; - var measure = subscription!.GetMeasure(); + var measure = subscription.GetMeasure(); return subscription.EndOfStreamSourceKey is { } key - ? new RegisteredSubscriptionMeasure(subscriptionId, key, measure).GetEndOfStream + ? new RegisteredSubscriptionMeasure(resolved.SubscriptionId, key, measure).GetEndOfStream : measure; } } diff --git a/src/Core/test/Eventuous.Tests.Subscriptions/SubscriptionGapMetricsTests.cs b/src/Core/test/Eventuous.Tests.Subscriptions/SubscriptionGapMetricsTests.cs index c2ca9408c..572388f9f 100644 --- a/src/Core/test/Eventuous.Tests.Subscriptions/SubscriptionGapMetricsTests.cs +++ b/src/Core/test/Eventuous.Tests.Subscriptions/SubscriptionGapMetricsTests.cs @@ -122,6 +122,30 @@ public async Task Checkpoint_advancing_past_shared_tail_reports_zero_then_refres subscriptions.Sum(x => x.Reads).ShouldBe(4); } + [Test] + public async Task Shared_source_uses_configured_subscription_ids_for_checkpoints_and_tags() { + var registrations = new ServiceCollection(); + foreach (var id in new[] { "first", "second" }) + registrations.AddSubscription(id, builder => builder.Configure(options => { + options.SubscriptionId = $"runtime-{id}"; + options.SourceKey = "shared"; + })); + await using var services = registrations.BuildServiceProvider(); + using var metrics = new SubscriptionMetrics(services.GetServices()); + using var diagnostic = new DiagnosticListener(CheckpointCommitHandler.DiagnosticName); + diagnostic.Write(CheckpointCommitHandler.CommitOperation, + new CheckpointCommitHandler.CommitEvent("runtime-first", new(3, 0, DateTime.UtcNow), null)); + diagnostic.Write(CheckpointCommitHandler.CommitOperation, + new CheckpointCommitHandler.CommitEvent("runtime-second", new(7, 0, DateTime.UtcNow), null)); + using var listener = Listen(out var values); + listener.RecordObservableInstruments(); + + values.Count.ShouldBe(2); + values["runtime-first"].ShouldBe(7); + values["runtime-second"].ShouldBe(3); + services.GetServices().Sum(x => x.Reads).ShouldBe(1); + } + static ServiceProvider CreateServices(params (string Id, string? Key)[] subscriptions) { var services = new ServiceCollection(); foreach (var (id, key) in subscriptions) From f5cc41f89fadfa029ab62b938834b44b222ccaf2 Mon Sep 17 00:00:00 2001 From: Neal Mummau Date: Thu, 8 Oct 2026 23:32:29 -0400 Subject: [PATCH 4/7] fix(kurrentdb): share gap source keys for persistent clients --- .../AllPersistentSubscription.cs | 2 +- .../Subscriptions/AllStreamSubscription.cs | 2 +- .../PersistentSubscriptionBase.cs | 5 +++++ .../StreamPersistentSubscription.cs | 2 +- .../Subscriptions/StreamSubscription.cs | 2 +- .../EndOfStreamSourceKeyTests.cs | 21 +++++++++++++++++++ 6 files changed, 30 insertions(+), 4 deletions(-) diff --git a/src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/AllPersistentSubscription.cs b/src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/AllPersistentSubscription.cs index 8939ebf16..a10caec7f 100644 --- a/src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/AllPersistentSubscription.cs +++ b/src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/AllPersistentSubscription.cs @@ -123,7 +123,7 @@ CancellationToken cance /// // Derived subscriptions must opt in explicitly: their measure may read a different source. public virtual object? EndOfStreamSourceKey - => GetType() == typeof(AllPersistentSubscription) ? (typeof(AllStreamSubscriptionMeasure), Client) : null; + => GetType() == typeof(AllPersistentSubscription) ? (typeof(AllStreamSubscriptionMeasure), SourceClient) : null; /// /// Returns a measure callback for the subscription diff --git a/src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/AllStreamSubscription.cs b/src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/AllStreamSubscription.cs index 0eb12a993..6e2ecea00 100644 --- a/src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/AllStreamSubscription.cs +++ b/src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/AllStreamSubscription.cs @@ -280,7 +280,7 @@ Task HandleCheckpointReached(SubscriptionRun run, global::KurrentDB.Client.Posit /// // Derived subscriptions must opt in explicitly: their measure may read a different source. public virtual object? EndOfStreamSourceKey - => GetType() == typeof(AllStreamSubscription) ? (typeof(AllStreamSubscriptionMeasure), Client) : null; + => GetType() == typeof(AllStreamSubscription) ? (typeof(AllStreamSubscriptionMeasure), (object)Client) : null; /// /// Returns a measure delegate for the subscription diff --git a/src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/PersistentSubscriptionBase.cs b/src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/PersistentSubscriptionBase.cs index b2707181b..7a8e079b0 100644 --- a/src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/PersistentSubscriptionBase.cs +++ b/src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/PersistentSubscriptionBase.cs @@ -40,6 +40,9 @@ public abstract class PersistentSubscriptionBase : EventSubscription where /// protected IMetadataSerializer MetadataSerializer { get; } + /// Identity of the supplied client used to share source-position reads. + protected object SourceClient { get; } + readonly HandleEventProcessingFailure _handleEventProcessingFailure; /// @@ -60,6 +63,7 @@ protected PersistentSubscriptionBase( IMetadataSerializer? metaSerializer ) : base(options, consumePipe, loggerFactory, eventSerializer) { + SourceClient = client; Client = client; MetadataSerializer = metaSerializer ?? DefaultMetadataSerializer.Instance; @@ -89,6 +93,7 @@ protected PersistentSubscriptionBase( IMetadataSerializer? metaSerializer ) : base(options, consumePipe, loggerFactory, eventSerializer) { + SourceClient = client; SubscriptionClient = client; MetadataSerializer = metaSerializer ?? DefaultMetadataSerializer.Instance; var settings = client.GetSettings().Copy(); diff --git a/src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/StreamPersistentSubscription.cs b/src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/StreamPersistentSubscription.cs index 53083dbb9..7d62a69ba 100644 --- a/src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/StreamPersistentSubscription.cs +++ b/src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/StreamPersistentSubscription.cs @@ -112,7 +112,7 @@ CancellationToken cance /// // Derived subscriptions must opt in explicitly: their measure may read a different source. public virtual object? EndOfStreamSourceKey - => GetType() == typeof(StreamPersistentSubscription) ? (typeof(StreamSubscriptionMeasure), Client, Options.StreamName) : null; + => GetType() == typeof(StreamPersistentSubscription) ? (typeof(StreamSubscriptionMeasure), SourceClient, Options.StreamName) : null; /// public GetSubscriptionEndOfStream GetMeasure() diff --git a/src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/StreamSubscription.cs b/src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/StreamSubscription.cs index 03a4616c0..317e231e2 100644 --- a/src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/StreamSubscription.cs +++ b/src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/StreamSubscription.cs @@ -164,7 +164,7 @@ MessageConsumeContext CreateContext(SubscriptionRun run, ResolvedEvent re, Cance /// // Derived subscriptions must opt in explicitly: their measure may read a different source. public virtual object? EndOfStreamSourceKey - => GetType() == typeof(StreamSubscription) ? (typeof(StreamSubscriptionMeasure), Client, Options.StreamName) : null; + => GetType() == typeof(StreamSubscription) ? (typeof(StreamSubscriptionMeasure), (object)Client, Options.StreamName) : null; /// /// Returns a measure delegate for this subscription diff --git a/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Subscriptions/EndOfStreamSourceKeyTests.cs b/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Subscriptions/EndOfStreamSourceKeyTests.cs index 973aa26bb..98b981d37 100644 --- a/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Subscriptions/EndOfStreamSourceKeyTests.cs +++ b/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Subscriptions/EndOfStreamSourceKeyTests.cs @@ -5,6 +5,7 @@ using Eventuous.Subscriptions; using Eventuous.Subscriptions.Checkpoints; using KurrentDBClient = global::KurrentDB.Client.KurrentDBClient; +using KurrentDBPersistentSubscriptionsClient = global::KurrentDB.Client.KurrentDBPersistentSubscriptionsClient; using KurrentDBClientSettings = global::KurrentDB.Client.KurrentDBClientSettings; namespace Eventuous.Tests.KurrentDB.Subscriptions; @@ -21,6 +22,26 @@ public async Task Derived_subscriptions_keep_independent_reads() { await Assert.That(subscription.EndOfStreamSourceKey).IsNull(); } + [Test] + public async Task Persistent_client_keys_share_sources_but_separate_clients_and_streams() { + using var client = new KurrentDBPersistentSubscriptionsClient(KurrentDBClientSettings.Create("esdb://localhost:2113?tls=false")); + using var otherClient = new KurrentDBPersistentSubscriptionsClient(KurrentDBClientSettings.Create("esdb://localhost:2113?tls=false")); + await using var all = new AllPersistentSubscription(client, new() { SubscriptionId = "first" }, new()); + await using var sameAll = new AllPersistentSubscription(client, new() { SubscriptionId = "second" }, new()); + await using var otherAll = new AllPersistentSubscription(otherClient, new() { SubscriptionId = "other" }, new()); + await using var stream = new StreamPersistentSubscription(client, new() { SubscriptionId = "stream", StreamName = new("one") }, new()); + await using var sameStream = new StreamPersistentSubscription(client, new() { SubscriptionId = "same-stream", StreamName = new("one") }, new()); + await using var otherStream = new StreamPersistentSubscription(client, new() { SubscriptionId = "other-stream", StreamName = new("two") }, new()); + await using var otherClientStream = new StreamPersistentSubscription(otherClient, new() { SubscriptionId = "other-client-stream", StreamName = new("one") }, new()); + + await Assert.That(all.EndOfStreamSourceKey).IsEqualTo(sameAll.EndOfStreamSourceKey); + await Assert.That(all.EndOfStreamSourceKey).IsNotEqualTo(otherAll.EndOfStreamSourceKey); + await Assert.That(all.EndOfStreamSourceKey).IsNotEqualTo(stream.EndOfStreamSourceKey); + await Assert.That(stream.EndOfStreamSourceKey).IsEqualTo(sameStream.EndOfStreamSourceKey); + await Assert.That(stream.EndOfStreamSourceKey).IsNotEqualTo(otherStream.EndOfStreamSourceKey); + await Assert.That(stream.EndOfStreamSourceKey).IsNotEqualTo(otherClientStream.EndOfStreamSourceKey); + } + sealed class DerivedAll(KurrentDBClient client) : AllStreamSubscription(client, "derived-all", new NoOpCheckpointStore(), new()); sealed class DerivedPersistentAll(KurrentDBClient client) : AllPersistentSubscription(client, "derived-persistent-all", new()); sealed class DerivedStream(KurrentDBClient client) : StreamSubscription( From d89dd67711191cf2ed3de6ce146c8ad3c54b851d Mon Sep 17 00:00:00 2001 From: Neal Mummau Date: Thu, 8 Oct 2026 23:57:27 -0400 Subject: [PATCH 5/7] test(kurrentdb): apply NoContext to gap metrics append helper --- .../Metrics/SharedSourceGapMetricsTests.cs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Metrics/SharedSourceGapMetricsTests.cs b/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Metrics/SharedSourceGapMetricsTests.cs index cf9be696a..20bc3b377 100644 --- a/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Metrics/SharedSourceGapMetricsTests.cs +++ b/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Metrics/SharedSourceGapMetricsTests.cs @@ -87,7 +87,7 @@ async Task Append() { var result = await writer.AppendToStreamAsync( stream, StreamState.Any, [new EventData(Uuid.NewUuid(), "shared-gap-event", "{}"u8.ToArray())], cancellationToken: cancellationToken - ); + ).NoContext(); return result.LogPosition.CommitPosition; } From 35ed1b0b619efcc2d7a901b99992971fcb367d1f Mon Sep 17 00:00:00 2001 From: Neal Mummau Date: Thu, 8 Oct 2026 23:58:11 -0400 Subject: [PATCH 6/7] fix(postgres): share gap reads across equivalent transient sources --- .../Extensions/RegistrationExtensions.cs | 17 +++++-- .../PostgresSourceIdentity.cs | 20 ++++++++ .../Subscriptions/PostgresSubscriptionBase.cs | 2 +- .../Metrics/SharedSourceGapMetricsTests.cs | 6 ++- .../EndOfStreamSourceKeyTests.cs | 46 +++++++++++++++++++ 5 files changed, 84 insertions(+), 7 deletions(-) create mode 100644 src/Postgres/src/Eventuous.Postgresql/PostgresSourceIdentity.cs diff --git a/src/Postgres/src/Eventuous.Postgresql/Extensions/RegistrationExtensions.cs b/src/Postgres/src/Eventuous.Postgresql/Extensions/RegistrationExtensions.cs index 7909e50f7..ad882dd7e 100644 --- a/src/Postgres/src/Eventuous.Postgresql/Extensions/RegistrationExtensions.cs +++ b/src/Postgres/src/Eventuous.Postgresql/Extensions/RegistrationExtensions.cs @@ -48,7 +48,8 @@ public IServiceCollection AddEventuousPostgres( configureBuilder?.Invoke(sp, builder); }, connectionLifetime, - dataSourceLifetime + dataSourceLifetime, + shareSourceIdentity: configureBuilder is null ); services.AddSingleton(options); services.AddSingleton(); @@ -83,7 +84,8 @@ public IServiceCollection AddEventuousPostgres( configureBuilder?.Invoke(sp, builder); }, connectionLifetime, - dataSourceLifetime + dataSourceLifetime, + shareSourceIdentity: configureBuilder is null ); services.AddSingleton(); @@ -96,8 +98,10 @@ void AddNpgsqlDataSourceCore( Func getConnectionString, Action? configureDataSource, ServiceLifetime connectionLifetime, - ServiceLifetime dataSourceLifetime + ServiceLifetime dataSourceLifetime, + bool shareSourceIdentity ) { + var registrationIdentity = new object(); services.TryAdd( new ServiceDescriptor( typeof(NpgsqlDataSource), @@ -106,7 +110,12 @@ ServiceLifetime dataSourceLifetime dataSourceBuilder.UseLoggerFactory(sp.GetService()); configureDataSource?.Invoke(sp, dataSourceBuilder); - return dataSourceBuilder.Build(); + var dataSource = dataSourceBuilder.Build(); + // A user callback may change session state or credentials on each resolution. + // Only the built-in configuration is known to produce equivalent reads. + if (shareSourceIdentity) + PostgresSourceIdentity.Register(dataSource, registrationIdentity, dataSourceBuilder.ConnectionString); + return dataSource; }, dataSourceLifetime ) diff --git a/src/Postgres/src/Eventuous.Postgresql/PostgresSourceIdentity.cs b/src/Postgres/src/Eventuous.Postgresql/PostgresSourceIdentity.cs new file mode 100644 index 000000000..02ac20614 --- /dev/null +++ b/src/Postgres/src/Eventuous.Postgresql/PostgresSourceIdentity.cs @@ -0,0 +1,20 @@ +// Copyright (C) Eventuous HQ OÜ. All rights reserved +// Licensed under the Apache License, Version 2.0. + +using System.Runtime.CompilerServices; + +namespace Eventuous.Postgresql; + +// Attach identities only to data sources built by a known-equivalent registration. +// Weak keys avoid extending data-source lifetimes; unregistered sources remain independent. +static class PostgresSourceIdentity { + static readonly ConditionalWeakTable Identities = new(); + + public static void Register(NpgsqlDataSource source, object registration, string connectionString) + => Identities.Add(source, new(registration, connectionString)); + + public static object Get(NpgsqlDataSource source) + => Identities.TryGetValue(source, out var identity) ? identity : source; + + sealed record Identity(object Registration, string ConnectionString); +} diff --git a/src/Postgres/src/Eventuous.Postgresql/Subscriptions/PostgresSubscriptionBase.cs b/src/Postgres/src/Eventuous.Postgresql/Subscriptions/PostgresSubscriptionBase.cs index bd7b0051e..6f99ac373 100644 --- a/src/Postgres/src/Eventuous.Postgresql/Subscriptions/PostgresSubscriptionBase.cs +++ b/src/Postgres/src/Eventuous.Postgresql/Subscriptions/PostgresSubscriptionBase.cs @@ -42,7 +42,7 @@ public abstract class PostgresSubscriptionBase( // The concrete type also isolates custom subscriptions that change the end-of-stream query. protected object GetEndOfStreamSourceKey(string? streamName) - => (GetType(), DataSource, Schema.Name, streamName); + => (GetType(), PostgresSourceIdentity.Get(DataSource), Schema.Name, streamName); protected override async ValueTask OpenConnection(CancellationToken cancellationToken) => await DataSource.OpenConnectionAsync(cancellationToken).NoContext(); diff --git a/src/Postgres/test/Eventuous.Tests.Postgres/Metrics/SharedSourceGapMetricsTests.cs b/src/Postgres/test/Eventuous.Tests.Postgres/Metrics/SharedSourceGapMetricsTests.cs index d70c55163..0f3b7e0b2 100644 --- a/src/Postgres/test/Eventuous.Tests.Postgres/Metrics/SharedSourceGapMetricsTests.cs +++ b/src/Postgres/test/Eventuous.Tests.Postgres/Metrics/SharedSourceGapMetricsTests.cs @@ -16,7 +16,9 @@ namespace Eventuous.Tests.Postgres.Metrics; [NotInParallel] public class SharedSourceGapMetricsTests { [Test] - public async Task All_stream_subscriptions_read_shared_tail_once_per_collection(CancellationToken cancellationToken) { + [Arguments(ServiceLifetime.Singleton)] + [Arguments(ServiceLifetime.Transient)] + public async Task All_stream_subscriptions_read_shared_tail_once_per_collection(ServiceLifetime dataSourceLifetime, CancellationToken cancellationToken) { await using var container = PostgresContainer.Create(); await container.StartAsync(cancellationToken); var connectionString = container.GetConnectionString(); @@ -29,7 +31,7 @@ public async Task All_stream_subscriptions_read_shared_tail_once_per_collection( mapper.AddType("shared-gap-event"); using var queries = new TailQueryCounter($"select max(global_position) from {schema.Name}.messages"); var services = new ServiceCollection(); - services.AddEventuousPostgres(connectionString, schema.Name); + services.AddEventuousPostgres(connectionString, schema.Name, dataSourceLifetime: dataSourceLifetime); foreach (var id in new[] { "first", "second", "caught-up" }) { services.AddSubscription(id, builder => builder .Configure(options => { diff --git a/src/Postgres/test/Eventuous.Tests.Postgres/Subscriptions/EndOfStreamSourceKeyTests.cs b/src/Postgres/test/Eventuous.Tests.Postgres/Subscriptions/EndOfStreamSourceKeyTests.cs index 01b6d27ee..53066a53b 100644 --- a/src/Postgres/test/Eventuous.Tests.Postgres/Subscriptions/EndOfStreamSourceKeyTests.cs +++ b/src/Postgres/test/Eventuous.Tests.Postgres/Subscriptions/EndOfStreamSourceKeyTests.cs @@ -5,6 +5,8 @@ using Eventuous.Postgresql.Subscriptions; using Eventuous.Subscriptions; using Eventuous.Subscriptions.Checkpoints; +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.DependencyInjection; using Npgsql; namespace Eventuous.Tests.Postgres.Subscriptions; @@ -33,6 +35,50 @@ public async Task Keys_separate_data_sources_schemas_and_streams() { await Assert.That(stream.EndOfStreamSourceKey).IsNotEqualTo(otherStream.EndOfStreamSourceKey); } + [Test] + [Arguments(false, false)] + [Arguments(true, false)] + [Arguments(false, true)] + [Arguments(true, true)] + public async Task Transient_registration_shares_only_known_equivalent_sources(bool fromConfiguration, bool customConfiguration) { + var services = new ServiceCollection(); + // Even a matching connection string cannot prove an arbitrary callback is equivalent. + Action? configure = customConfiguration ? (_, _) => { } : null; + if (fromConfiguration) { + var configuration = new ConfigurationBuilder().AddInMemoryCollection(new Dictionary { + ["ConnectionString"] = "Host=localhost;Database=one" + }).Build(); + services.AddEventuousPostgres(configuration, configureBuilder: configure, dataSourceLifetime: ServiceLifetime.Transient); + } else { + services.AddEventuousPostgres("Host=localhost;Database=one", configureBuilder: configure, dataSourceLifetime: ServiceLifetime.Transient); + } + await using var provider = services.BuildServiceProvider(); + var firstSource = provider.GetRequiredService(); + var secondSource = provider.GetRequiredService(); + await Assert.That(ReferenceEquals(firstSource, secondSource)).IsFalse(); + await using var first = All(firstSource, "first", "schema"); + await using var second = All(secondSource, "second", "schema"); + await using var stream = Stream(firstSource, "stream", "one"); + await using var sameStream = Stream(secondSource, "same-stream", "one"); + await using var otherStream = Stream(secondSource, "other-stream", "two"); + await using var otherSchema = All(secondSource, "other-schema", "other"); + + await Assert.That(Equals(first.EndOfStreamSourceKey, second.EndOfStreamSourceKey)).IsEqualTo(!customConfiguration); + await Assert.That(Equals(stream.EndOfStreamSourceKey, sameStream.EndOfStreamSourceKey)).IsEqualTo(!customConfiguration); + await Assert.That(first.EndOfStreamSourceKey).IsNotEqualTo(otherSchema.EndOfStreamSourceKey); + await Assert.That(first.EndOfStreamSourceKey).IsNotEqualTo(stream.EndOfStreamSourceKey); + await Assert.That(stream.EndOfStreamSourceKey).IsNotEqualTo(otherStream.EndOfStreamSourceKey); + } + + [Test] + public async Task Independently_created_sources_with_matching_connection_strings_remain_separate() { + await using var source = NpgsqlDataSource.Create("Host=localhost;Database=one"); + await using var otherSource = NpgsqlDataSource.Create("Host=localhost;Database=one"); + await using var first = All(source, "first", "schema"); + await using var second = All(otherSource, "second", "schema"); + await Assert.That(first.EndOfStreamSourceKey).IsNotEqualTo(second.EndOfStreamSourceKey); + } + [Test] public async Task Derived_subscriptions_keep_independent_reads() { await using var source = NpgsqlDataSource.Create("Host=localhost;Database=one"); From 7437033df6e297188e049d76eae7d36f3bf2942e Mon Sep 17 00:00:00 2001 From: Neal Mummau Date: Fri, 9 Oct 2026 13:16:43 -0400 Subject: [PATCH 7/7] fix(tests): configure awaits in shared-source gap metrics tests Apply NoContext() to I/O and checkpoint awaits, including SQLite setup and cleanup, while preserving assertion awaits. Disambiguate PostgreSQL and SQLite task helpers with assembly aliases. --- .../Metrics/SharedSourceGapMetricsTests.cs | 24 ++++++------ .../Eventuous.Tests.Postgres.csproj | 2 +- .../Metrics/SharedSourceGapMetricsTests.cs | 25 +++++++----- .../Metrics/SharedSourceGapMetricsTests.cs | 21 +++++----- .../Eventuous.Tests.Sqlite.csproj | 2 +- .../Metrics/SharedSourceGapMetricsTests.cs | 39 +++++++++++-------- 6 files changed, 62 insertions(+), 51 deletions(-) diff --git a/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Metrics/SharedSourceGapMetricsTests.cs b/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Metrics/SharedSourceGapMetricsTests.cs index 20bc3b377..518932a45 100644 --- a/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Metrics/SharedSourceGapMetricsTests.cs +++ b/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Metrics/SharedSourceGapMetricsTests.cs @@ -22,13 +22,13 @@ public async Task All_stream_subscriptions_read_shared_tail_once_per_collection( .WithEnvironment("KURRENTDB_RUN_PROJECTIONS", "None") .WithEnvironment("KURRENTDB_START_STANDARD_PROJECTIONS", "false") .Build(); - await container.StartAsync(cancellationToken); + await container.StartAsync(cancellationToken).NoContext(); using var writer = new KurrentDBClient(KurrentDBClientSettings.Create(container.GetConnectionString())); var stream = $"shared-gap-{Guid.NewGuid():N}"; // KurrentDB global positions are byte offsets, not event counts. Use actual append positions. - var first = await Append(); - var second = await Append(); - var tail = await Append(); + var first = await Append().NoContext(); + var second = await Append().NoContext(); + var tail = await Append().NoContext(); using var requests = new TailReadCounter(); var settings = KurrentDBClientSettings.Create(container.GetConnectionString()); @@ -46,9 +46,9 @@ public async Task All_stream_subscriptions_read_shared_tail_once_per_collection( subscriptions.Distinct().Count().ShouldBe(3); using var metrics = new SubscriptionMetrics(provider.GetServices()); // Do not start the subscriptions: advance deterministic checkpoints through the real commit handler. - await Commit("first", first); - await Commit("second", second); - await Commit("caught-up", tail); + await Commit("first", first).NoContext(); + await Commit("second", second).NoContext(); + await Commit("caught-up", tail).NoContext(); Dictionary gaps = new(); using var listener = new MeterListener { @@ -71,10 +71,10 @@ public async Task All_stream_subscriptions_read_shared_tail_once_per_collection( gaps["caught-up"].ShouldBe(0); requests.Count.ShouldBe(1); - var advanced = await Append(); - var newTail = await Append(); + var advanced = await Append().NoContext(); + var newTail = await Append().NoContext(); newTail.ShouldBeGreaterThan(tail); - await Commit("second", advanced); + await Commit("second", advanced).NoContext(); gaps.Clear(); listener.RecordObservableInstruments(); gaps.Count.ShouldBe(3); @@ -97,8 +97,8 @@ async Task Commit(string id, ulong position) { committed.TrySetResult(); return new ValueTask(checkpoint); }, TimeSpan.FromMilliseconds(10)); - (await handler.Commit(new(position, 0, DateTime.UtcNow), cancellationToken)).ShouldBeTrue(); - await committed.Task.WaitAsync(TimeSpan.FromSeconds(10), cancellationToken); + (await handler.Commit(new(position, 0, DateTime.UtcNow), cancellationToken).NoContext()).ShouldBeTrue(); + await committed.Task.WaitAsync(TimeSpan.FromSeconds(10), cancellationToken).NoContext(); } } diff --git a/src/Postgres/test/Eventuous.Tests.Postgres/Eventuous.Tests.Postgres.csproj b/src/Postgres/test/Eventuous.Tests.Postgres/Eventuous.Tests.Postgres.csproj index 63fcb946b..4307bd9b2 100644 --- a/src/Postgres/test/Eventuous.Tests.Postgres/Eventuous.Tests.Postgres.csproj +++ b/src/Postgres/test/Eventuous.Tests.Postgres/Eventuous.Tests.Postgres.csproj @@ -16,7 +16,7 @@ - + diff --git a/src/Postgres/test/Eventuous.Tests.Postgres/Metrics/SharedSourceGapMetricsTests.cs b/src/Postgres/test/Eventuous.Tests.Postgres/Metrics/SharedSourceGapMetricsTests.cs index 0f3b7e0b2..227cf4fdd 100644 --- a/src/Postgres/test/Eventuous.Tests.Postgres/Metrics/SharedSourceGapMetricsTests.cs +++ b/src/Postgres/test/Eventuous.Tests.Postgres/Metrics/SharedSourceGapMetricsTests.cs @@ -1,6 +1,8 @@ // Copyright (C) Eventuous HQ OÜ. All rights reserved // Licensed under the Apache License, Version 2.0. +extern alias Postgres; + using System.Diagnostics; using System.Diagnostics.Metrics; using Eventuous.Postgresql; @@ -11,6 +13,9 @@ using Eventuous.Tests.Postgres.Fixtures; using Microsoft.Extensions.DependencyInjection; +// Select the store helper explicitly: Persistence exposes the same internal extensions. +using static Postgres::Eventuous.Tools.TaskExtensions; + namespace Eventuous.Tests.Postgres.Metrics; [NotInParallel] @@ -20,13 +25,13 @@ public class SharedSourceGapMetricsTests { [Arguments(ServiceLifetime.Transient)] public async Task All_stream_subscriptions_read_shared_tail_once_per_collection(ServiceLifetime dataSourceLifetime, CancellationToken cancellationToken) { await using var container = PostgresContainer.Create(); - await container.StartAsync(cancellationToken); + await container.StartAsync(cancellationToken).NoContext(); var connectionString = container.GetConnectionString(); var schema = new Schema($"metrics_{Guid.NewGuid():N}"); var storeOptions = new PostgresStoreOptions { ConnectionString = connectionString, Schema = schema.Name, InitializeDatabase = true }; - await new SchemaInitializer(storeOptions).StartAsync(cancellationToken); + await new SchemaInitializer(storeOptions).StartAsync(cancellationToken).NoContext(); var mapper = new TypeMapper(); mapper.AddType("shared-gap-event"); using var queries = new TailQueryCounter($"select max(global_position) from {schema.Name}.messages"); @@ -43,15 +48,15 @@ public async Task All_stream_subscriptions_read_shared_tail_once_per_collection( var store = new PostgresStore(provider.GetRequiredService(), storeOptions, new DefaultEventSerializer(new(), mapper)); // PostgreSQL global positions start at one: ten events leave the tail at ten. - await Append(10); + await Append(10).NoContext(); var subscriptions = provider.GetServices().ToArray(); await Assert.That(subscriptions.Length).IsEqualTo(3); await Assert.That(subscriptions.Distinct().Count()).IsEqualTo(3); using var metrics = new SubscriptionMetrics(provider.GetServices()); // Keep polling stopped; commit deterministic positions through the production handler. - await Commit("first", 3); - await Commit("second", 7); - await Commit("caught-up", 10); + await Commit("first", 3).NoContext(); + await Commit("second", 7).NoContext(); + await Commit("caught-up", 10).NoContext(); Dictionary gaps = new(); using var listener = new MeterListener { @@ -75,8 +80,8 @@ public async Task All_stream_subscriptions_read_shared_tail_once_per_collection( await Assert.That(queries.Count).IsEqualTo(1); await Assert.That(queries.FailedCount).IsEqualTo(0); - await Append(10); - await Commit("second", 12); + await Append(10).NoContext(); + await Commit("second", 12).NoContext(); gaps.Clear(); listener.RecordObservableInstruments(); await Assert.That(gaps.Count).IsEqualTo(3); @@ -98,8 +103,8 @@ async Task Commit(string id, ulong position) { committed.TrySetResult(); return new ValueTask(checkpoint); }, TimeSpan.FromMilliseconds(10)); - await Assert.That(await handler.Commit(new(position, 0, DateTime.UtcNow), cancellationToken)).IsTrue(); - await committed.Task.WaitAsync(TimeSpan.FromSeconds(10), cancellationToken); + await Assert.That(await handler.Commit(new(position, 0, DateTime.UtcNow), cancellationToken).NoContext()).IsTrue(); + await committed.Task.WaitAsync(TimeSpan.FromSeconds(10), cancellationToken).NoContext(); } } diff --git a/src/SqlServer/test/Eventuous.Tests.SqlServer/Metrics/SharedSourceGapMetricsTests.cs b/src/SqlServer/test/Eventuous.Tests.SqlServer/Metrics/SharedSourceGapMetricsTests.cs index 31ac6361b..68a033fd7 100644 --- a/src/SqlServer/test/Eventuous.Tests.SqlServer/Metrics/SharedSourceGapMetricsTests.cs +++ b/src/SqlServer/test/Eventuous.Tests.SqlServer/Metrics/SharedSourceGapMetricsTests.cs @@ -9,6 +9,7 @@ using Eventuous.Subscriptions.Diagnostics; using Eventuous.Subscriptions.Registrations; using Eventuous.Tests.SqlServer.Fixtures; +using Eventuous.Tools; using Microsoft.Data.SqlClient.Diagnostics; using Microsoft.Extensions.DependencyInjection; using Shouldly; @@ -20,19 +21,19 @@ public class SharedSourceGapMetricsTests { [Test] public async Task All_stream_subscriptions_read_shared_tail_once_per_collection(CancellationToken cancellationToken) { await using var container = SqlContainer.Create(); - await container.StartAsync(cancellationToken); + await container.StartAsync(cancellationToken).NoContext(); var connectionString = container.GetConnectionString(); var schema = new Schema($"metrics_{Guid.NewGuid():N}"); var storeOptions = new SqlServerStoreOptions { ConnectionString = connectionString, Schema = schema.SchemaName, InitializeDatabase = true }; // Use the same startup retry as the metrics fixture: login readiness can precede tempdb readiness. - await new SchemaInitializer(storeOptions).StartAsync(cancellationToken); + await new SchemaInitializer(storeOptions).StartAsync(cancellationToken).NoContext(); var mapper = new TypeMapper(); mapper.AddType("shared-gap-event"); var store = new SqlServerStore(storeOptions, new DefaultEventSerializer(new(), mapper)); // Global positions start at zero: appending eleven events leaves the tail at ten. - await Append(11); + await Append(11).NoContext(); var services = new ServiceCollection(); foreach (var id in new[] { "first", "second", "caught-up" }) { @@ -49,9 +50,9 @@ public async Task All_stream_subscriptions_read_shared_tail_once_per_collection( subscriptions.Distinct().Count().ShouldBe(3); using var metrics = new SubscriptionMetrics(provider.GetServices()); // Keep polling stopped; commit deterministic positions through the production handler. - await Commit("first", 3); - await Commit("second", 7); - await Commit("caught-up", 10); + await Commit("first", 3).NoContext(); + await Commit("second", 7).NoContext(); + await Commit("caught-up", 10).NoContext(); Dictionary gaps = new(); using var listener = new MeterListener { @@ -74,8 +75,8 @@ public async Task All_stream_subscriptions_read_shared_tail_once_per_collection( gaps["caught-up"].ShouldBe(0); queries.Count.ShouldBe(1); - await Append(10); - await Commit("second", 12); + await Append(10).NoContext(); + await Commit("second", 12).NoContext(); gaps.Clear(); listener.RecordObservableInstruments(); gaps.Count.ShouldBe(3); @@ -96,8 +97,8 @@ async Task Commit(string id, ulong position) { committed.TrySetResult(); return new ValueTask(checkpoint); }, TimeSpan.FromMilliseconds(10)); - (await handler.Commit(new(position, 0, DateTime.UtcNow), cancellationToken)).ShouldBeTrue(); - await committed.Task.WaitAsync(TimeSpan.FromSeconds(10), cancellationToken); + (await handler.Commit(new(position, 0, DateTime.UtcNow), cancellationToken).NoContext()).ShouldBeTrue(); + await committed.Task.WaitAsync(TimeSpan.FromSeconds(10), cancellationToken).NoContext(); } } diff --git a/src/Sqlite/test/Eventuous.Tests.Sqlite/Eventuous.Tests.Sqlite.csproj b/src/Sqlite/test/Eventuous.Tests.Sqlite/Eventuous.Tests.Sqlite.csproj index b8ca40d56..329d817cd 100644 --- a/src/Sqlite/test/Eventuous.Tests.Sqlite/Eventuous.Tests.Sqlite.csproj +++ b/src/Sqlite/test/Eventuous.Tests.Sqlite/Eventuous.Tests.Sqlite.csproj @@ -14,7 +14,7 @@ - + diff --git a/src/Sqlite/test/Eventuous.Tests.Sqlite/Metrics/SharedSourceGapMetricsTests.cs b/src/Sqlite/test/Eventuous.Tests.Sqlite/Metrics/SharedSourceGapMetricsTests.cs index af4c079c0..ceeac3793 100644 --- a/src/Sqlite/test/Eventuous.Tests.Sqlite/Metrics/SharedSourceGapMetricsTests.cs +++ b/src/Sqlite/test/Eventuous.Tests.Sqlite/Metrics/SharedSourceGapMetricsTests.cs @@ -1,6 +1,8 @@ // Copyright (C) Eventuous HQ OÜ. All rights reserved // Licensed under the Apache License, Version 2.0. +extern alias Sqlite; + using System.Diagnostics.Metrics; using Eventuous.Sqlite; using Eventuous.Sqlite.Subscriptions; @@ -12,6 +14,9 @@ using Shouldly; using SQLitePCL; +// Select the store helper explicitly: Persistence exposes the same internal extensions. +using static Sqlite::Eventuous.Tools.TaskExtensions; + namespace Eventuous.Tests.Sqlite.Metrics; [NotInParallel] @@ -25,18 +30,18 @@ public async Task All_stream_subscriptions_read_shared_tail_once_per_collection( // without overriding subscriptions or adding a production connection factory hook. var connectionString = $"Data Source=file:{schema.SchemaName}?mode=memory&cache=shared;Pooling=True"; await using var keeper = new SqliteConnection(connectionString + ";Pooling=False"); - await keeper.OpenAsync(cancellationToken); + await keeper.OpenAsync(cancellationToken).NoContext(); await using (var command = keeper.CreateCommand()) { command.CommandText = "SELECT file FROM pragma_database_list WHERE name = 'main'"; - (await command.ExecuteScalarAsync(cancellationToken)).ShouldBe(""); + (await command.ExecuteScalarAsync(cancellationToken).NoContext()).ShouldBe(""); } - await using var queries = await TailQueryCounter.Create(connectionString, $"SELECT MAX(global_position) FROM {schema.MessagesTable}", cancellationToken); - await schema.CreateSchema(connectionString, null, cancellationToken); + await using var queries = await TailQueryCounter.Create(connectionString, $"SELECT MAX(global_position) FROM {schema.MessagesTable}", cancellationToken).NoContext(); + await schema.CreateSchema(connectionString, null, cancellationToken).NoContext(); var mapper = new TypeMapper(); mapper.AddType("shared-gap-event"); var store = new SqliteStore(new() { ConnectionString = connectionString, Schema = schema.SchemaName }, new DefaultEventSerializer(new(), mapper)); // SQLite global positions start at one: ten events leave the tail at ten. - await Append(10); + await Append(10).NoContext(); var services = new ServiceCollection(); foreach (var id in new[] { "first", "second", "caught-up" }) { @@ -53,9 +58,9 @@ public async Task All_stream_subscriptions_read_shared_tail_once_per_collection( subscriptions.Distinct().Count().ShouldBe(3); using var metrics = new SubscriptionMetrics(provider.GetServices()); // Keep polling stopped; commit deterministic positions through the production handler. - await Commit("first", 3); - await Commit("second", 7); - await Commit("caught-up", 10); + await Commit("first", 3).NoContext(); + await Commit("second", 7).NoContext(); + await Commit("caught-up", 10).NoContext(); Dictionary gaps = new(); using var listener = new MeterListener { @@ -76,8 +81,8 @@ public async Task All_stream_subscriptions_read_shared_tail_once_per_collection( gaps["caught-up"].ShouldBe(0); queries.Count.ShouldBe(1); - await Append(10); - await Commit("second", 12); + await Append(10).NoContext(); + await Commit("second", 12).NoContext(); queries.Count.ShouldBe(1); gaps.Clear(); listener.RecordObservableInstruments(); @@ -99,8 +104,8 @@ async Task Commit(string id, ulong position) { committed.TrySetResult(); return new ValueTask(checkpoint); }, TimeSpan.FromMilliseconds(10)); - (await handler.Commit(new(position, 0, DateTime.UtcNow), cancellationToken)).ShouldBeTrue(); - await committed.Task.WaitAsync(TimeSpan.FromSeconds(10), cancellationToken); + (await handler.Commit(new(position, 0, DateTime.UtcNow), cancellationToken).NoContext()).ShouldBeTrue(); + await committed.Task.WaitAsync(TimeSpan.FromSeconds(10), cancellationToken).NoContext(); } } @@ -124,14 +129,14 @@ sealed class TailQueryCounter : IAsyncDisposable { public static async Task Create(string connectionString, string commandText, CancellationToken cancellationToken) { var connection = new SqliteConnection(connectionString); try { - await connection.OpenAsync(cancellationToken); + await connection.OpenAsync(cancellationToken).NoContext(); var counter = new TailQueryCounter(connection, commandText); // Production opens/closes connections sequentially and reuses this native handle. - await connection.CloseAsync(); + await connection.CloseAsync().NoContext(); return counter; } catch { SqliteConnection.ClearPool(connection); - await connection.DisposeAsync(); + await connection.DisposeAsync().NoContext(); throw; } } @@ -140,13 +145,13 @@ public static async Task Create(string connectionString, strin public async ValueTask DisposeAsync() { try { - await _connection.OpenAsync(); + await _connection.OpenAsync().NoContext(); _connection.Handle.ShouldBeSameAs(_handle); raw.sqlite3_profile(_handle, (strdelegate_profile)null!, null); } finally { // Clear only this test's uniquely named pool, never other tests' pools. SqliteConnection.ClearPool(_connection); - await _connection.DisposeAsync(); + await _connection.DisposeAsync().NoContext(); } } }