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..473c86f09 100644 --- a/src/Core/src/Eventuous.Subscriptions/Registrations/SubscriptionRegistrationExtensions.cs +++ b/src/Core/src/Eventuous.Subscriptions/Registrations/SubscriptionRegistrationExtensions.cs @@ -73,9 +73,14 @@ 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; - return subscription!.GetMeasure(); + var measure = subscription.GetMeasure(); + + return subscription.EndOfStreamSourceKey is { } key + ? 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 new file mode 100644 index 000000000..572388f9f --- /dev/null +++ b/src/Core/test/Eventuous.Tests.Subscriptions/SubscriptionGapMetricsTests.cs @@ -0,0 +1,195 @@ +// 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); + } + + [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) + 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..a10caec7f 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), 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 8106abd9c..6e2ecea00 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), (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 2354cc9af..7d62a69ba 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), SourceClient, 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..317e231e2 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), (object)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..518932a45 --- /dev/null +++ b/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Metrics/SharedSourceGapMetricsTests.cs @@ -0,0 +1,119 @@ +// 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 Eventuous.Tools; +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).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().NoContext(); + var second = await Append().NoContext(); + var tail = await Append().NoContext(); + + 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).NoContext(); + await Commit("second", second).NoContext(); + await Commit("caught-up", tail).NoContext(); + + 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().NoContext(); + var newTail = await Append().NoContext(); + newTail.ShouldBeGreaterThan(tail); + await Commit("second", advanced).NoContext(); + 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 + ).NoContext(); + 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).NoContext()).ShouldBeTrue(); + await committed.Task.WaitAsync(TimeSpan.FromSeconds(10), cancellationToken).NoContext(); + } + } + + 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).NoContext(); + } + } +} 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..98b981d37 --- /dev/null +++ b/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Subscriptions/EndOfStreamSourceKeyTests.cs @@ -0,0 +1,69 @@ +// 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 KurrentDBPersistentSubscriptionsClient = global::KurrentDB.Client.KurrentDBPersistentSubscriptionsClient; +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(); + } + + [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( + 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/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/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..6f99ac373 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(), 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/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 new file mode 100644 index 000000000..227cf4fdd --- /dev/null +++ b/src/Postgres/test/Eventuous.Tests.Postgres/Metrics/SharedSourceGapMetricsTests.cs @@ -0,0 +1,140 @@ +// 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; +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; + +// Select the store helper explicitly: Persistence exposes the same internal extensions. +using static Postgres::Eventuous.Tools.TaskExtensions; + +namespace Eventuous.Tests.Postgres.Metrics; + +[NotInParallel] +public class SharedSourceGapMetricsTests { + [Test] + [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).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).NoContext(); + 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, dataSourceLifetime: dataSourceLifetime); + 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).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).NoContext(); + await Commit("second", 7).NoContext(); + await Commit("caught-up", 10).NoContext(); + + 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).NoContext(); + await Commit("second", 12).NoContext(); + 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).NoContext()).IsTrue(); + await committed.Task.WaitAsync(TimeSpan.FromSeconds(10), cancellationToken).NoContext(); + } + } + + 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..53066a53b --- /dev/null +++ b/src/Postgres/test/Eventuous.Tests.Postgres/Subscriptions/EndOfStreamSourceKeyTests.cs @@ -0,0 +1,102 @@ +// 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 Microsoft.Extensions.Configuration; +using Microsoft.Extensions.DependencyInjection; +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] + [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"); + 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..68a033fd7 --- /dev/null +++ b/src/SqlServer/test/Eventuous.Tests.SqlServer/Metrics/SharedSourceGapMetricsTests.cs @@ -0,0 +1,138 @@ +// 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 Eventuous.Tools; +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).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).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).NoContext(); + + 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).NoContext(); + await Commit("second", 7).NoContext(); + await Commit("caught-up", 10).NoContext(); + + 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).NoContext(); + await Commit("second", 12).NoContext(); + 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).NoContext()).ShouldBeTrue(); + await committed.Task.WaitAsync(TimeSpan.FromSeconds(10), cancellationToken).NoContext(); + } + } + + 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/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 new file mode 100644 index 000000000..ceeac3793 --- /dev/null +++ b/src/Sqlite/test/Eventuous.Tests.Sqlite/Metrics/SharedSourceGapMetricsTests.cs @@ -0,0 +1,158 @@ +// 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; +using Eventuous.Subscriptions.Checkpoints; +using Eventuous.Subscriptions.Diagnostics; +using Eventuous.Subscriptions.Registrations; +using Microsoft.Data.Sqlite; +using Microsoft.Extensions.DependencyInjection; +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] +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).NoContext(); + await using (var command = keeper.CreateCommand()) { + command.CommandText = "SELECT file FROM pragma_database_list WHERE name = 'main'"; + (await command.ExecuteScalarAsync(cancellationToken).NoContext()).ShouldBe(""); + } + 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).NoContext(); + + 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).NoContext(); + await Commit("second", 7).NoContext(); + await Commit("caught-up", 10).NoContext(); + + 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).NoContext(); + await Commit("second", 12).NoContext(); + 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).NoContext()).ShouldBeTrue(); + await committed.Task.WaitAsync(TimeSpan.FromSeconds(10), cancellationToken).NoContext(); + } + } + + public record GapEvent(int Value); + + sealed class TailQueryCounter : IAsyncDisposable { + readonly SqliteConnection _connection; + readonly sqlite3 _handle; + int _count; + + 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); + } + + public static async Task Create(string connectionString, string commandText, CancellationToken cancellationToken) { + var connection = new SqliteConnection(connectionString); + try { + await connection.OpenAsync(cancellationToken).NoContext(); + var counter = new TailQueryCounter(connection, commandText); + // Production opens/closes connections sequentially and reuses this native handle. + await connection.CloseAsync().NoContext(); + return counter; + } catch { + SqliteConnection.ClearPool(connection); + await connection.DisposeAsync().NoContext(); + throw; + } + } + + public int Count => Volatile.Read(ref _count); + + public async ValueTask DisposeAsync() { + try { + 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().NoContext(); + } + } + } +} 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..7fc047b54 --- /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.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 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()); +}