fix(sql): scope stream subscription end-of-stream measure to the subscribed stream (#586)

src/Core/test/Eventuous.Tests.Subscriptions.Base/SubscriptionMeasureBase.cs

@@ -36,6 +36,43 @@ protected async Task ShouldMeasureEndOfStream(CancellationToken cancellationToke
         await Assert.That(measured.Position).IsEqualTo(last);
     }
 
+    /// <summary>
+    /// For a stream subscription, the measure must report the tail of the subscribed stream only. Reproduces GitHub
+    /// #586, where the relational implementation returned <c>MAX(stream_position)</c> across every stream in the
+    /// store, so any longer stream inflated the gap of a fully caught-up subscription.
+    /// </summary>
+    protected async Task ShouldMeasureEndOfSubscribedStream(StreamName streamName, CancellationToken cancellationToken) {
+        const int otherStreamLength      = 20;
+        const int subscribedStreamLength = 5;
+
+        var measure = fixture.GetMeasure();
+
+        // Invoked before the subscription has ever connected, as the metrics observer does. The subscribed stream
+        // doesn't exist yet, which must read as an empty stream rather than EndOfStream.Invalid.
+        var empty = await measure(cancellationToken);
+        await Assert.That(empty.SubscriptionId).IsEqualTo(fixture.SubscriptionId);
+        await Assert.That(empty.Position).IsEqualTo(0ul);
+
+        // A longer, unrelated stream: its tail must not leak into the subscribed stream's measure.
+        await fixture.AppendEvents(new($"other-{Guid.NewGuid():N}"), [.. fixture.CreateEvents(otherStreamLength)], ExpectedStreamVersion.NoStream);
+
+        var events = fixture.CreateEvents(subscribedStreamLength).ToList();
+        await fixture.AppendEvents(streamName, [.. events], ExpectedStreamVersion.NoStream);
+
+        var measured = await measure(cancellationToken);
+        await Assert.That(measured.SubscriptionId).IsEqualTo(fixture.SubscriptionId);
+        await Assert.That(measured.Position).IsEqualTo((ulong)(subscribedStreamLength - 1));
+
+        await fixture.StartSubscription();
+        await fixture.Handler.AssertCollection(TimeSpan.FromSeconds(2), [.. events]).Validate(cancellationToken);
+        await fixture.StopSubscription();
+
+        // Once caught up, the gap the metrics derive from this measure and the checkpoint must be zero.
+        var checkpoint = await fixture.CheckpointStore.GetLastCheckpoint(fixture.SubscriptionId, cancellationToken);
+        var caughtUp   = await measure(cancellationToken);
+        await Assert.That(caughtUp.Position - checkpoint.Position!.Value).IsEqualTo(0ul);
+    }
+
     async Task GenerateAndHandleCommands(int count) {
         var commands = Enumerable
             .Range(0, count)

src/Diagnostics/test/Eventuous.Tests.OpenTelemetry/Fakes/TestHandler.cs

@@ -2,7 +2,7 @@
 
 namespace Eventuous.Tests.OpenTelemetry.Fakes;
 
-class TestHandler(MessageCounter counter, ILogger<TestHandler> log) : BaseEventHandler {
+public class TestHandler(MessageCounter counter, ILogger<TestHandler> log) : BaseEventHandler {
     public override async ValueTask<EventHandlingStatus> HandleEvent(IMessageConsumeContext context) {
         await Task.Delay(10, context.CancellationToken);
         counter.Increment();

src/Diagnostics/test/Eventuous.Tests.OpenTelemetry/MetricsTests.cs

@@ -62,6 +62,67 @@ public void Teardown() {
     MetricValue[]? _values;
 }
 
+/// <summary>
+/// Observes <c>eventuous.subscription.gap.count</c> through the real pipeline — the meter registered by
+/// <c>AddEventuousSubscriptions</c>, the measure resolved from DI, the checkpoint-commit listener and the gap
+/// arithmetic in <see cref="SubscriptionMetrics"/> — rather than calling the subscription's measure directly.
+/// Reproduces GitHub #586 at the metric level: a longer, unrelated stream must not inflate the gap reported for a
+/// caught-up stream subscription.
+/// </summary>
+public abstract class SubscriptionGapMetricsTestsBase(IMetricsSubscriptionFixtureBase fixture) {
+    protected async Task ShouldReportZeroGapWhenCaughtUp(CancellationToken cancellationToken) {
+        var other = new StreamName($"other-{Guid.NewGuid():N}");
+        await fixture.Producer.Produce(other, TestEvent.CreateMany(fixture.Count * 2), new(), cancellationToken: cancellationToken);
+        await fixture.Producer.Produce(fixture.Stream, TestEvent.CreateMany(fixture.Count), new(), cancellationToken: cancellationToken);
+
+        await WaitUntil(() => fixture.Counter.Count >= fixture.Count, TimeSpan.FromSeconds(30), cancellationToken);
+        await Assert.That(fixture.Counter.Count).IsEqualTo(fixture.Count);
+
+        // The checkpoint behind the gap commits on a batch or a delay, so the gauge converges on zero rather than
+        // reading it the moment the last event is handled.
+        var gap = await WaitForGapCount(0, TimeSpan.FromSeconds(15), cancellationToken);
+
+        await Assert.That(gap).IsNotNull();
+        await Assert.That(gap!.Value).IsEqualTo(0d);
+        await gap.CheckTag(SubscriptionMetrics.SubscriptionIdTag, fixture.SubscriptionId);
+        await gap.CheckTag(fixture.DefaultTagKey, fixture.DefaultTagValue);
+    }
+
+    /// <summary>
+    /// Polls the exporter until the gap gauge reads <paramref name="expected"/> or the timeout elapses, returning the
+    /// last observation either way so a failure shows the value that was actually reported.
+    /// </summary>
+    async Task<MetricValue?> WaitForGapCount(double expected, TimeSpan timeout, CancellationToken cancellationToken) {
+        MetricValue? gap      = null;
+        var          deadline = DateTime.UtcNow + timeout;
+
+        while (DateTime.UtcNow < deadline) {
+            fixture.Exporter.Collect(Timeout.Infinite);
+            gap = fixture.Exporter.CollectValues().FirstOrDefault(x => x.Name == SubscriptionMetrics.GapCountMetricName);
+            TestContext.Current?.OutputWriter.WriteLine($"Gap: {gap?.Value.ToString() ?? "not reported"}");
+
+            if (gap?.Value == expected) break;
+
+            await Task.Delay(200, cancellationToken);
+        }
+
+        return gap;
+    }
+
+    static async Task WaitUntil(Func<bool> condition, TimeSpan timeout, CancellationToken cancellationToken) {
+        var deadline = DateTime.UtcNow + timeout;
+
+        while (!condition() && DateTime.UtcNow < deadline) {
+            await Task.Delay(100, cancellationToken);
+        }
+    }
+
+    [After(Test)]
+    public void Teardown() => _es.Dispose();
+
+    readonly TestEventListener _es = new(null, "OpenTelemetry");
+}
+
 static class TagExtensions {
     extension(MetricValue metric) {
         public async Task CheckTag(string tag, string expectedValue) {

src/KurrentDB/test/Eventuous.Tests.KurrentDB/Metrics/MetricsTests.cs

@@ -11,3 +11,12 @@ public async Task ShouldMeasureSubscriptionGapCountBase_Esdb() {
         await ShouldMeasureSubscriptionGapCountBase();
     }
 }
+
+[ClassDataSource<MetricsFixture>]
+[NotInParallel]
+public class SubscriptionGapMetricsTests(MetricsFixture fixture) : SubscriptionGapMetricsTestsBase(fixture) {
+    [Test]
+    public async Task ShouldReportZeroGapWhenCaughtUp_Esdb(CancellationToken cancellationToken) {
+        await ShouldReportZeroGapWhenCaughtUp(cancellationToken);
+    }
+}

src/Postgres/src/Eventuous.Postgresql/Subscriptions/PostgresAllStreamSubscription.cs

@@ -38,6 +38,9 @@ protected override NpgsqlCommand PrepareCommand(NpgsqlConnection connection, lon
         => connection.GetCommand(Schema.ReadAllForwards)
             .Add("_from_position", NpgsqlDbType.Bigint, start + 1)
             .Add("_count", NpgsqlDbType.Integer, Options.MaxPageSize);
+
+    protected override NpgsqlCommand PrepareEndOfStreamCommand(NpgsqlConnection connection)
+        => connection.GetCommand($"select max(global_position) from {Schema.Name}.messages");
 }
 
 public record PostgresAllStreamSubscriptionOptions : PostgresSubscriptionBaseOptions;

src/Postgres/src/Eventuous.Postgresql/Subscriptions/PostgresStreamSubscription.cs

@@ -40,6 +40,19 @@ protected override NpgsqlCommand PrepareCommand(NpgsqlConnection connection, lon
             .Add("_from_position", NpgsqlDbType.Integer, (int)start + 1)
             .Add("_count", NpgsqlDbType.Integer, Options.MaxPageSize);
 
+    // Filtered by name rather than the id resolved in BeforeSubscribe, so the measure is valid before the
+    // subscription connects; a stream that doesn't exist yet reads as empty.
+    protected override NpgsqlCommand PrepareEndOfStreamCommand(NpgsqlConnection connection)
+        => connection.GetCommand(
+                $"""
+                 select max(m.stream_position)
+                 from {Schema.Name}.messages m
+                 inner join {Schema.Name}.streams s on s.stream_id = m.stream_id
+                 where s.stream_name = @_stream_name
+                 """
+            )
+            .Add("_stream_name", NpgsqlDbType.Varchar, _streamName);
+
     protected override async Task BeforeSubscribe(CancellationToken cancellationToken) {
         await using var connection = await DataSource.OpenConnectionAsync(cancellationToken).NoContext();
 

src/Postgres/src/Eventuous.Postgresql/Subscriptions/PostgresSubscriptionBase.cs

@@ -45,9 +45,6 @@ protected override async ValueTask<NpgsqlConnection> OpenConnection(Cancellation
 
     protected override bool IsTransient(Exception exception) => exception is PostgresException { IsTransient: true };
 
-    protected override string GetEndOfStream { get; } = $"select max(stream_position) from {options.Schema}.messages";
-    protected override string GetEndOfAll    { get; } = $"select max(global_position) from {options.Schema}.messages";
-
     protected override async ValueTask HandleGapTimeout(long gapPosition, long currentStart, CancellationToken cancellationToken) {
         try {
             await using var connection = await DataSource.OpenConnectionAsync(cancellationToken).NoContext();

src/Postgres/test/Eventuous.Tests.Postgres/Metrics/MetricsTests.cs

@@ -11,3 +11,12 @@ public async Task ShouldMeasureSubscriptionGapCountBase_Postgres() {
         await ShouldMeasureSubscriptionGapCountBase();
     }
 }
+
+[ClassDataSource<MetricsFixture>]
+[NotInParallel]
+public class SubscriptionGapMetricsTests(MetricsFixture fixture) : SubscriptionGapMetricsTestsBase(fixture) {
+    [Test]
+    public async Task ShouldReportZeroGapWhenCaughtUp_Postgres(CancellationToken cancellationToken) {
+        await ShouldReportZeroGapWhenCaughtUp(cancellationToken);
+    }
+}

src/Postgres/test/Eventuous.Tests.Postgres/Subscriptions/SubscriptionMeasureTests.cs

@@ -18,3 +18,18 @@ public async Task Postgres_ShouldMeasureEndOfStream(CancellationToken cancellati
         await ShouldMeasureEndOfStream(cancellationToken);
     }
 }
+
+[ClassDataSource<StreamNameFixture>(Shared = SharedType.None)]
+[NotInParallel]
+public class StreamSubscriptionMeasure(StreamNameFixture streamNameFixture)
+    : SubscriptionMeasureBase<PostgreSqlContainer, PostgresStreamSubscription, PostgresStreamSubscriptionOptions, PostgresCheckpointStore>(
+        new SubscriptionFixture<PostgresStore, PostgresStreamSubscription, PostgresStreamSubscriptionOptions, TestEventHandler>(
+            opt => opt.Stream = streamNameFixture.StreamName,
+            false
+        )
+    ) {
+    [Test]
+    public async Task Postgres_ShouldMeasureEndOfSubscribedStream(CancellationToken cancellationToken) {
+        await ShouldMeasureEndOfSubscribedStream(streamNameFixture.StreamName, cancellationToken);
+    }
+}

src/Relational/src/Eventuous.Sql.Base/Subscriptions/SqlSubscriptionBase.cs

@@ -1,7 +1,6 @@
 // Copyright (C) Eventuous HQ OÜ. All rights reserved
 // Licensed under the Apache License, Version 2.0.
 
-using System.Data;
 using System.Data.Common;
 using System.Runtime.InteropServices;
 using System.Text;
@@ -321,26 +320,23 @@ MessageConsumeContext AsContext(SubscriptionRun run, PersistedEvent evt, object?
     GetSubscriptionEndOfStream IMeasuredSubscription.GetMeasure() => GetSubscriptionEndOfStream;
 
     /// <summary>
-    /// Get SQL statement to get the end of the stream
+    /// Prepares a command that returns the position the subscription gap is measured against: the last global
+    /// position for an <see cref="SubscriptionKind.All"/> subscription, or the last position within the subscribed
+    /// stream for a <see cref="SubscriptionKind.Stream"/> one. The position is read from the first column of the
+    /// first row and may be <c>NULL</c> when there is nothing to measure yet.
     /// </summary>
-    protected abstract string GetEndOfStream { get; }
-
-    /// <summary>
-    /// Get SQL statement to get the end of the global log
-    /// </summary>
-    protected abstract string GetEndOfAll { get; }
+    /// <param name="connection">Connection that can be used to create the command</param>
+    /// <remarks>
+    /// The measure is also handed to the metrics observer, which can invoke it before the subscription has ever
+    /// connected, so the command must not depend on state resolved in <see cref="BeforeSubscribe"/>.
+    /// </remarks>
+    protected abstract DbCommand PrepareEndOfStreamCommand(TConnection connection);
 
     async ValueTask<EndOfStream> GetSubscriptionEndOfStream(CancellationToken cancellationToken) {
         try {
             await using var connection = await OpenConnection(cancellationToken).NoContext();
-            await using var cmd        = connection.CreateCommand();
-            cmd.CommandType = CommandType.Text;
-
-            cmd.CommandText = Kind switch {
-                SubscriptionKind.All    => GetEndOfAll,
-                SubscriptionKind.Stream => GetEndOfStream
-            };
-            await using var reader = await cmd.ExecuteReaderAsync(cancellationToken).NoContext();
+            await using var cmd        = PrepareEndOfStreamCommand(connection);
+            await using var reader     = await cmd.ExecuteReaderAsync(cancellationToken).NoContext();
 
             // MAX(...) returns NULL on an empty table, and providers may return the position column as either
             // Int32 or Int64, so guard against DBNull and convert rather than calling the strict GetInt64.

src/SqlServer/src/Eventuous.SqlServer/Subscriptions/SqlServerAllStreamSubscription.cs

@@ -37,6 +37,9 @@ protected override SqlCommand PrepareCommand(SqlConnection connection, long star
         => connection.GetStoredProcCommand(Schema.ReadAllForwards)
             .Add("@from_position", SqlDbType.BigInt, start + 1)
             .Add("@count", SqlDbType.Int, Options.MaxPageSize);
+
+    protected override SqlCommand PrepareEndOfStreamCommand(SqlConnection connection)
+        => connection.GetTextCommand($"SELECT MAX(GlobalPosition) FROM {Schema.SchemaName}.Messages");
 }
 
 public record SqlServerAllStreamSubscriptionOptions : SqlServerSubscriptionBaseOptions;

src/SqlServer/src/Eventuous.SqlServer/Subscriptions/SqlServerStreamSubscription.cs

@@ -40,6 +40,19 @@ protected override SqlCommand PrepareCommand(SqlConnection connection, long star
             .Add("@from_position", SqlDbType.Int, (int)start + 1)
             .Add("@count", SqlDbType.Int, Options.MaxPageSize);
 
+    // Filtered by name rather than the id resolved in BeforeSubscribe, so the measure is valid before the
+    // subscription connects; a stream that doesn't exist yet reads as empty.
+    protected override SqlCommand PrepareEndOfStreamCommand(SqlConnection connection)
+        => connection.GetTextCommand(
+                $"""
+                 SELECT MAX(m.StreamPosition)
+                 FROM {Schema.SchemaName}.Messages m
+                 INNER JOIN {Schema.SchemaName}.Streams s ON s.StreamId = m.StreamId
+                 WHERE s.StreamName = @stream_name
+                 """
+            )
+            .Add("@stream_name", SqlDbType.NVarChar, _streamName);
+
     protected override async Task BeforeSubscribe(CancellationToken cancellationToken) {
         await using var connection = await OpenConnection(cancellationToken).NoContext();
 

src/SqlServer/src/Eventuous.SqlServer/Subscriptions/SqlServerSubscriptionBase.cs

@@ -31,8 +31,6 @@ protected SqlServerSubscriptionBase(
         );
         var connectionString = connectionOptions?.ConnectionString ?? options.ConnectionString;
         _connectionString = Ensure.NotEmptyString(connectionString);
-        GetEndOfStream    = $"SELECT MAX(StreamPosition) FROM {options.Schema}.Messages";
-        GetEndOfAll       = $"SELECT MAX(GlobalPosition) FROM {options.Schema}.Messages";
     }
 
     protected override async ValueTask<SqlConnection> OpenConnection(CancellationToken cancellationToken)
@@ -51,9 +49,6 @@ protected override bool IsStopping(Exception exception)
                 sqlException.Number == 3980 && sqlException.Message.Contains("Operation cancelled by user."),
             _ => false
         };
-
-    protected override string GetEndOfStream { get; }
-    protected override string GetEndOfAll    { get; }
 }
 
 public abstract record SqlServerSubscriptionBaseOptions : SqlSubscriptionOptionsBase {

src/SqlServer/test/Eventuous.Tests.SqlServer/Metrics/MetricsTests.cs

@@ -11,3 +11,12 @@ public async Task ShouldMeasureSubscriptionGapCountBase_SqlServer() {
         await ShouldMeasureSubscriptionGapCountBase();
     }
 }
+
+[ClassDataSource<MetricsFixture>]
+[NotInParallel]
+public class SubscriptionGapMetricsTests(MetricsFixture fixture) : SubscriptionGapMetricsTestsBase(fixture) {
+    [Test]
+    public async Task ShouldReportZeroGapWhenCaughtUp_SqlServer(CancellationToken cancellationToken) {
+        await ShouldReportZeroGapWhenCaughtUp(cancellationToken);
+    }
+}

src/SqlServer/test/Eventuous.Tests.SqlServer/Subscriptions/ConnectionStringTests.cs

@@ -145,6 +145,14 @@ protected override SqlCommand PrepareCommand(SqlConnection connection, long star
 
         return command;
     }
+
+    protected override SqlCommand PrepareEndOfStreamCommand(SqlConnection connection) {
+        var command = connection.CreateCommand();
+        command.CommandType = CommandType.Text;
+        command.CommandText = "SELECT MAX(GlobalPosition) FROM dbo.Messages";
+
+        return command;
+    }
 }
 
 /// <summary>

src/SqlServer/test/Eventuous.Tests.SqlServer/Subscriptions/SubscriptionMeasureTests.cs

@@ -17,3 +17,18 @@ public async Task SqlServer_ShouldMeasureEndOfStream(CancellationToken cancellat
         await ShouldMeasureEndOfStream(cancellationToken);
     }
 }
+
+[ClassDataSource<StreamNameFixture>(Shared = SharedType.None)]
+[NotInParallel]
+public class StreamSubscriptionMeasure(StreamNameFixture streamNameFixture)
+    : SubscriptionMeasureBase<MsSqlContainer, SqlServerStreamSubscription, SqlServerStreamSubscriptionOptions, SqlServerCheckpointStore>(
+        new SubscriptionFixture<SqlServerStreamSubscription, SqlServerStreamSubscriptionOptions, TestEventHandler>(
+            opt => opt.Stream = streamNameFixture.StreamName,
+            false
+        )
+    ) {
+    [Test]
+    public async Task SqlServer_ShouldMeasureEndOfSubscribedStream(CancellationToken cancellationToken) {
+        await ShouldMeasureEndOfSubscribedStream(streamNameFixture.StreamName, cancellationToken);
+    }
+}

src/Sqlite/src/Eventuous.Sqlite/Subscriptions/SqliteAllStreamSubscription.cs

@@ -47,6 +47,9 @@ LIMIT @count
             )
             .Add("@from_position", start + 1)
             .Add("@count", Options.MaxPageSize);
+
+    protected override SqliteCommand PrepareEndOfStreamCommand(SqliteConnection connection)
+        => connection.GetTextCommand($"SELECT MAX(global_position) FROM {Schema.MessagesTable}");
 }
 
 public record SqliteAllStreamSubscriptionOptions : SqliteSubscriptionBaseOptions;

src/Sqlite/src/Eventuous.Sqlite/Subscriptions/SqliteStreamSubscription.cs

@@ -49,6 +49,19 @@ LIMIT @count
             .Add("@from_position", (int)start + 1)
             .Add("@count", Options.MaxPageSize);
 
+    // Filtered by name rather than the id resolved in BeforeSubscribe, so the measure is valid before the
+    // subscription connects; a stream that doesn't exist yet reads as empty.
+    protected override SqliteCommand PrepareEndOfStreamCommand(SqliteConnection connection)
+        => connection.GetTextCommand(
+                $"""
+                 SELECT MAX(m.stream_position)
+                 FROM {Schema.MessagesTable} m
+                 INNER JOIN {Schema.StreamsTable} s ON s.stream_id = m.stream_id
+                 WHERE s.stream_name = @stream_name
+                 """
+            )
+            .Add("@stream_name", _streamName);
+
     protected override async Task BeforeSubscribe(CancellationToken cancellationToken) {
         await using var connection = await OpenConnection(cancellationToken).NoContext();
 

src/Sqlite/src/Eventuous.Sqlite/Subscriptions/SqliteSubscriptionBase.cs

@@ -31,8 +31,6 @@ protected SqliteSubscriptionBase(
         );
         var connectionString = connectionOptions?.ConnectionString ?? options.ConnectionString;
         _connectionString = Ensure.NotEmptyString(connectionString);
-        GetEndOfStream    = $"SELECT MAX(stream_position) FROM {Schema.MessagesTable}";
-        GetEndOfAll       = $"SELECT MAX(global_position) FROM {Schema.MessagesTable}";
     }
 
     protected override async ValueTask<SqliteConnection> OpenConnection(CancellationToken cancellationToken)
@@ -42,9 +40,6 @@ protected override async ValueTask<SqliteConnection> OpenConnection(Cancellation
 
     protected override bool IsStopping(Exception exception)
         => exception is OperationCanceledException;
-
-    protected override string GetEndOfStream { get; }
-    protected override string GetEndOfAll    { get; }
 }
 
 public abstract record SqliteSubscriptionBaseOptions : SqlSubscriptionOptionsBase {