fix(subscriptions): deduplicate gap metrics source reads

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);
-}
+}
+
+// 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<EndOfStream> GetEndOfStream(CancellationToken cancellationToken) => measure(cancellationToken);
+}

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<GetSubscriptionEndOfStream> measures) {
         var getGaps = measures.ToArray();
 
-        Dictionary<string, EndOfStream> streams = new();
+        ConcurrentDictionary<string, EndOfStream> streams = new();
 
         ObserveMetric<long> observeGapValues = () => ObserveGapValues(getGaps);
 
@@ -74,11 +75,39 @@ IEnumerable<Measurement<double>> ObserveTimeValues()
                 x => Measure(Math.Round((_checkpointMetrics.GetLastTimestamp(x.SubscriptionId) - x.Timestamp).TotalSeconds), x.SubscriptionId)
             );
 
-        IEnumerable<Measurement<long>> 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<Measurement<long>> ObserveGapValues(GetSubscriptionEndOfStream[] getEndOfStreams) {
+            // Collection-local: neither successful reads nor failures survive into the next scrape.
+            Dictionary<object, EndOfStream> sources = new();
+            List<Measurement<long>> 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<T> Measure<T>(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;
             }
         }
     }

src/Core/src/Eventuous.Subscriptions/IMessageSubscription.cs

@@ -29,4 +29,12 @@ public interface IMessageSubscription {
 
 public interface IMeasuredSubscription {
     GetSubscriptionEndOfStream GetMeasure();
+
+    /// <summary>
+    /// Identifies the source read by <see cref="GetMeasure"/>. 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.
+    /// </summary>
+    object? EndOfStreamSourceKey => null;
 }

src/Core/src/Eventuous.Subscriptions/Registrations/SubscriptionRegistrationExtensions.cs

@@ -73,9 +73,14 @@ Action<SubscriptionBuilder<T, TOptions>> configureSubscription
             SubscriptionBuilder<T, TOptions> GetBuilder(IServiceProvider sp) => sp.GetSubscriptionBuilder<T, TOptions>(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;
             }
         }
     }

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<GetSubscriptionEndOfStream>());
+        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<string, long> values = new();
+        Dictionary<string, double> times = new();
+        listener.InstrumentPublished = (instrument, owner) => {
+            if (instrument.Meter.Name == SubscriptionMetrics.MeterName) owner.EnableMeasurementEvents(instrument);
+        };
+        listener.SetMeasurementEventCallback<long>((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<double>((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<MeasuredSubscription>().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<MeasuredSubscription>().ToArray();
+        subscriptions[0].Fail = true;
+        subscriptions[0].Throw = throws;
+        using var metrics = new SubscriptionMetrics(services.GetServices<GetSubscriptionEndOfStream>());
+        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<GetSubscriptionEndOfStream>(), direct]);
+        using var listener = Listen(out var values);
+        listener.RecordObservableInstruments();
+        values.Count.ShouldBe(3);
+        values["direct"].ShouldBe(42);
+        directReads.ShouldBe(1);
+        services.GetServices<MeasuredSubscription>().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<MeasuredSubscription>().ToArray();
+        using var metrics = new SubscriptionMetrics(services.GetServices<GetSubscriptionEndOfStream>());
+        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<MeasuredSubscription, MeasureOptions>(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<GetSubscriptionEndOfStream>());
+        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<MeasuredSubscription>().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<MeasuredSubscription, MeasureOptions>(id, builder => builder.Configure(x => x.SourceKey = key));
+        return services.BuildServiceProvider();
+    }
+
+    static MeterListener Listen(out Dictionary<string, long> values) {
+        Dictionary<string, long> 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<long>((_, 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<MeasureOptions>(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);
+    }
+}

src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/AllPersistentSubscription.cs

@@ -120,6 +120,11 @@ CancellationToken                                                          cance
     /// <inheritdoc />
     protected override ulong GetContextStreamPosition(ResolvedEvent re) => re.Event.Position.CommitPosition;
 
+    /// <inheritdoc/>
+    // Derived subscriptions must opt in explicitly: their measure may read a different source.
+    public virtual object? EndOfStreamSourceKey
+        => GetType() == typeof(AllPersistentSubscription) ? (typeof(AllStreamSubscriptionMeasure), SourceClient) : null;
+
     /// <summary>
     /// Returns a measure callback for the subscription
     /// </summary>

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();
     }
 
+    /// <inheritdoc/>
+    // 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;
+
     /// <summary>
     /// Returns a measure delegate for the subscription
     /// </summary>

src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/PersistentSubscriptionBase.cs

@@ -40,6 +40,9 @@ public abstract class PersistentSubscriptionBase<T> : EventSubscription<T> where
     /// </summary>
     protected IMetadataSerializer MetadataSerializer { get; }
 
+    /// <summary>Identity of the supplied client used to share source-position reads.</summary>
+    protected object SourceClient { get; }
+
     readonly HandleEventProcessingFailure _handleEventProcessingFailure;
 
     /// <summary>
@@ -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();

src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/StreamPersistentSubscription.cs

@@ -109,6 +109,11 @@ CancellationToken                                                          cance
     /// <inheritdoc/>
     protected override ulong GetContextStreamPosition(ResolvedEvent re) => re.Event.EventNumber;
 
+    /// <inheritdoc/>
+    // 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;
+
     /// <inheritdoc/>
     public GetSubscriptionEndOfStream GetMeasure()
         => new StreamSubscriptionMeasure(Options.SubscriptionId, Options.StreamName, Client).GetEndOfStream;