test: subscription drop/resubscribe + health; fix relational end-of-stream measure (#308, #548)

src/Core/test/Eventuous.Tests.Subscriptions.Base/Fixtures/SubscriptionFixtureBase.cs

@@ -1,6 +1,7 @@
 using DotNet.Testcontainers.Containers;
 using Eventuous.Subscriptions;
 using Eventuous.Subscriptions.Checkpoints;
+using Eventuous.Subscriptions.Diagnostics;
 using Eventuous.Sut.Domain;
 using Eventuous.Tests.Persistence.Base.Fixtures;
 using Microsoft.Extensions.DependencyInjection;
@@ -30,9 +31,36 @@ protected SubscriptionFixtureBase(bool autoStart = true, LogLevel logLevel = Log
     IMessageSubscription                Subscription    { get; set; }         = null!;
     protected internal ILoggerFactory   LoggerFactory   { get; set; }         = null!;
 
+    /// <summary>
+    /// Health check fed by the subscription's subscribed/dropped callbacks. Lets tests assert that a dropped
+    /// subscription reports unhealthy and that it recovers to healthy after resubscription.
+    /// </summary>
+    protected internal SubscriptionHealthCheck Health { get; } = new();
+
+    /// <summary>
+    /// True when the subscription has detected a drop and is trying to resubscribe.
+    /// </summary>
+    public bool IsDropped => ((EventSubscription<TSubscriptionOptions>)Subscription).IsDropped;
+
+    /// <summary>
+    /// Returns the subscription's end-of-stream measure delegate (requires an <see cref="IMeasuredSubscription"/>).
+    /// </summary>
+    protected internal GetSubscriptionEndOfStream GetMeasure() => ((IMeasuredSubscription)Subscription).GetMeasure();
+
     public string SubscriptionId { get; } = $"test-{Guid.NewGuid():N}";
 
-    protected internal ValueTask StartSubscription() => Subscription.SubscribeWithLog(Log);
+    protected internal ValueTask StartSubscription()
+        => Subscription.Subscribe(
+            id => {
+                Health.ReportHealthy(id);
+                Log.LogInformation("{Subscription} subscribed", id);
+            },
+            (id, reason, ex) => {
+                Health.ReportUnhealthy(id, ex);
+                Log.LogWarning(ex, "{Subscription} dropped {Reason}", id, reason);
+            },
+            CancellationToken.None
+        );
 
     protected internal ValueTask StopSubscription() => Subscription.UnsubscribeWithLog(Log);
 

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

@@ -0,0 +1,114 @@
+using DotNet.Testcontainers.Containers;
+using Eventuous.Subscriptions;
+using Eventuous.Subscriptions.Checkpoints;
+using Eventuous.Sut.App;
+using Eventuous.Tests.Persistence.Base.Fixtures;
+using Microsoft.Extensions.Diagnostics.HealthChecks;
+
+namespace Eventuous.Tests.Subscriptions.Base;
+
+/// <summary>
+/// Base test that verifies a subscription drops and resubscribes when the underlying infrastructure
+/// connection is lost and later restored, and that the subscription health check reports
+/// <see cref="HealthStatus.Unhealthy"/> while dropped and <see cref="HealthStatus.Healthy"/> again after
+/// recovery. The connection loss is simulated by pausing the infrastructure container (Docker pause), which
+/// freezes the database process while keeping the published port intact, so the same connection string keeps
+/// working once the container is unpaused. Reproduces the scenario from GitHub #308 (and #307).
+/// </summary>
+public abstract class SubscriptionDropBase<TContainer, TSubscription, TSubscriptionOptions, TCheckpointStore>(
+        SubscriptionFixtureBase<TContainer, TSubscription, TSubscriptionOptions, TCheckpointStore, TestEventHandler> fixture
+    ) : SubscriptionTestBase(fixture)
+    where TContainer : DockerContainer
+    where TSubscription : EventSubscription<TSubscriptionOptions>
+    where TSubscriptionOptions : SubscriptionOptions
+    where TCheckpointStore : class, ICheckpointStore {
+    const int BatchSize = 5;
+
+    // The poll/connection failure that follows a pause surfaces only after the provider's command/connection
+    // timeout elapses, so give detection (and recovery) a generous window.
+    static readonly TimeSpan DropTimeout = TimeSpan.FromSeconds(90);
+
+    /// <summary>
+    /// Produces and consumes events, drops the connection by pausing the container, asserts the subscription is
+    /// reported unhealthy, restores the connection, then asserts that the subscription resubscribes, reports
+    /// healthy again, and resumes processing newly produced events.
+    /// </summary>
+    protected async Task ShouldResubscribeAfterConnectionDrop(CancellationToken cancellationToken) {
+        // 1. Produce and consume an initial batch; the subscription must be healthy.
+        await GenerateAndHandleCommands(BatchSize);
+        await fixture.StartSubscription();
+        var consumedInitial = await WaitUntil(() => fixture.Handler.Count >= BatchSize, DropTimeout, cancellationToken);
+        await Assert.That(consumedInitial).IsTrue();
+        await Assert.That(await GetHealthStatus(cancellationToken)).IsEqualTo(HealthStatus.Healthy);
+
+        // 2. Drop the connection by pausing the container.
+        WriteLine("Pausing the container to drop the connection");
+        await fixture.Container.PauseAsync(cancellationToken);
+
+        // 3. The subscription must detect the drop and report unhealthy.
+        var dropped = await WaitUntil(() => fixture.IsDropped, DropTimeout, cancellationToken);
+        await Assert.That(dropped).IsTrue();
+        var unhealthy = await WaitUntil(async () => await GetHealthStatus(cancellationToken) == HealthStatus.Unhealthy, DropTimeout, cancellationToken);
+        await Assert.That(unhealthy).IsTrue();
+        WriteLine("Subscription dropped and reported unhealthy");
+
+        // 4. Restore the connection.
+        WriteLine("Unpausing the container to restore the connection");
+        await fixture.Container.UnpauseAsync(cancellationToken);
+
+        // 5. The subscription must resubscribe and report healthy again.
+        var recovered = await WaitUntil(() => !fixture.IsDropped, DropTimeout, cancellationToken);
+        await Assert.That(recovered).IsTrue();
+        var healthy = await WaitUntil(async () => await GetHealthStatus(cancellationToken) == HealthStatus.Healthy, DropTimeout, cancellationToken);
+        await Assert.That(healthy).IsTrue();
+        WriteLine("Subscription resubscribed and reported healthy");
+
+        // 6. Events produced after recovery must be processed.
+        var countBeforeRecovery = fixture.Handler.Count;
+        await GenerateAndHandleCommands(BatchSize);
+        var resumed = await WaitUntil(() => fixture.Handler.Count >= countBeforeRecovery + BatchSize, DropTimeout, cancellationToken);
+
+        await fixture.StopSubscription();
+
+        await Assert.That(resumed).IsTrue();
+        WriteLine("Processed {0} events after recovery", fixture.Handler.Count - countBeforeRecovery);
+    }
+
+    async Task<HealthStatus> GetHealthStatus(CancellationToken cancellationToken) {
+        var result = await fixture.Health.CheckHealthAsync(new(), cancellationToken);
+
+        return result.Status;
+    }
+
+    static Task<bool> WaitUntil(Func<bool> condition, TimeSpan timeout, CancellationToken cancellationToken)
+        => WaitUntil(() => Task.FromResult(condition()), timeout, cancellationToken);
+
+    static async Task<bool> WaitUntil(Func<Task<bool>> condition, TimeSpan timeout, CancellationToken cancellationToken) {
+        using var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
+        cts.CancelAfter(timeout);
+
+        try {
+            while (!await condition()) {
+                await Task.Delay(200, cts.Token);
+            }
+
+            return true;
+        } catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested) {
+            return false;
+        }
+    }
+
+    async Task GenerateAndHandleCommands(int count) {
+        var commands = Enumerable
+            .Range(0, count)
+            .Select(_ => DomainFixture.CreateImportBooking())
+            .ToList();
+
+        var service = new BookingService(fixture.EventStore);
+
+        foreach (var cmd in commands) {
+            var result = await service.Handle(cmd, default);
+            result.ThrowIfError();
+        }
+    }
+}

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

@@ -0,0 +1,53 @@
+using DotNet.Testcontainers.Containers;
+using Eventuous.Subscriptions;
+using Eventuous.Subscriptions.Checkpoints;
+using Eventuous.Subscriptions.Diagnostics;
+using Eventuous.Sut.App;
+using Eventuous.Tests.Persistence.Base.Fixtures;
+
+namespace Eventuous.Tests.Subscriptions.Base;
+
+/// <summary>
+/// Base test for the subscription end-of-stream measure used by the gap/lag diagnostics. Verifies that the
+/// measure reports a valid position both on an empty store and after events are appended. Reproduces GitHub
+/// #548, where the relational implementation queried the wrong column, threw on 32-bit position columns, and
+/// failed on the <c>NULL</c> returned by <c>MAX(...)</c> over an empty table.
+/// </summary>
+public abstract class SubscriptionMeasureBase<TContainer, TSubscription, TSubscriptionOptions, TCheckpointStore>(
+        SubscriptionFixtureBase<TContainer, TSubscription, TSubscriptionOptions, TCheckpointStore, TestEventHandler> fixture
+    ) : SubscriptionTestBase(fixture)
+    where TContainer : DockerContainer
+    where TSubscription : EventSubscription<TSubscriptionOptions>
+    where TSubscriptionOptions : SubscriptionOptions
+    where TCheckpointStore : class, ICheckpointStore {
+    protected async Task ShouldMeasureEndOfStream(CancellationToken cancellationToken) {
+        var measure = fixture.GetMeasure();
+
+        // An empty store must yield a valid measure at position zero, not EndOfStream.Invalid.
+        var empty = await measure(cancellationToken);
+        await Assert.That(empty.SubscriptionId).IsEqualTo(fixture.SubscriptionId);
+        await Assert.That(empty.Position).IsEqualTo(0ul);
+
+        // After appending events, the measure must report the global end position.
+        await GenerateAndHandleCommands(10);
+        var last = await fixture.GetLastPosition();
+
+        var measured = await measure(cancellationToken);
+        await Assert.That(measured.SubscriptionId).IsEqualTo(fixture.SubscriptionId);
+        await Assert.That(measured.Position).IsEqualTo(last);
+    }
+
+    async Task GenerateAndHandleCommands(int count) {
+        var commands = Enumerable
+            .Range(0, count)
+            .Select(_ => DomainFixture.CreateImportBooking())
+            .ToList();
+
+        var service = new BookingService(fixture.EventStore);
+
+        foreach (var cmd in commands) {
+            var result = await service.Handle(cmd, default);
+            result.ThrowIfError();
+        }
+    }
+}

src/KurrentDB/test/Eventuous.Tests.KurrentDB/Subscriptions/SubscriptionDropTests.cs

@@ -0,0 +1,21 @@
+using Eventuous.KurrentDB.Subscriptions;
+using Eventuous.Tests.KurrentDB.Subscriptions.Fixtures;
+using Eventuous.Tests.Subscriptions.Base;
+using Testcontainers.KurrentDb;
+
+namespace Eventuous.Tests.KurrentDB.Subscriptions;
+
+public class SubscriptionDrop()
+    : SubscriptionDropBase<KurrentDbContainer, AllStreamSubscription, AllStreamSubscriptionOptions, TestCheckpointStore>(
+        new CatchUpSubscriptionFixture<AllStreamSubscription, AllStreamSubscriptionOptions, TestEventHandler>(
+            _ => { },
+            new("$all"),
+            false
+        )
+    ) {
+    [Test]
+    [Retry(3)]
+    public async Task Esdb_ShouldResubscribeAfterConnectionDrop(CancellationToken cancellationToken) {
+        await ShouldResubscribeAfterConnectionDrop(cancellationToken);
+    }
+}

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

@@ -0,0 +1,20 @@
+using Eventuous.Postgresql;
+using Eventuous.Postgresql.Subscriptions;
+using Eventuous.Tests.Subscriptions.Base;
+using Testcontainers.PostgreSql;
+
+namespace Eventuous.Tests.Postgres.Subscriptions;
+
+[NotInParallel]
+public class SubscriptionDrop()
+    : SubscriptionDropBase<PostgreSqlContainer, PostgresAllStreamSubscription, PostgresAllStreamSubscriptionOptions, PostgresCheckpointStore>(
+        new SubscriptionFixture<PostgresStore, PostgresAllStreamSubscription, PostgresAllStreamSubscriptionOptions, TestEventHandler>(
+            _ => { },
+            false
+        )
+    ) {
+    [Test]
+    public async Task Postgres_ShouldResubscribeAfterConnectionDrop(CancellationToken cancellationToken) {
+        await ShouldResubscribeAfterConnectionDrop(cancellationToken);
+    }
+}

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

@@ -0,0 +1,20 @@
+using Eventuous.Postgresql;
+using Eventuous.Postgresql.Subscriptions;
+using Eventuous.Tests.Subscriptions.Base;
+using Testcontainers.PostgreSql;
+
+namespace Eventuous.Tests.Postgres.Subscriptions;
+
+[NotInParallel]
+public class SubscriptionMeasure()
+    : SubscriptionMeasureBase<PostgreSqlContainer, PostgresAllStreamSubscription, PostgresAllStreamSubscriptionOptions, PostgresCheckpointStore>(
+        new SubscriptionFixture<PostgresStore, PostgresAllStreamSubscription, PostgresAllStreamSubscriptionOptions, TestEventHandler>(
+            _ => { },
+            false
+        )
+    ) {
+    [Test]
+    public async Task Postgres_ShouldMeasureEndOfStream(CancellationToken cancellationToken) {
+        await ShouldMeasureEndOfStream(cancellationToken);
+    }
+}

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

@@ -341,16 +341,20 @@ async ValueTask<EndOfStream> GetSubscriptionEndOfStream(CancellationToken cancel
             cmd.CommandType = CommandType.Text;
 
             cmd.CommandText = Kind switch {
-                SubscriptionKind.All    => GetEndOfStream,
-                SubscriptionKind.Stream => GetEndOfAll
+                SubscriptionKind.All    => GetEndOfAll,
+                SubscriptionKind.Stream => GetEndOfStream
             };
             await using var reader = await cmd.ExecuteReaderAsync(cancellationToken).NoContext();
 
-            var position = await reader.ReadAsync(cancellationToken).NoContext() ? reader.GetInt64(0) : 0;
+            // 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.
+            var position = await reader.ReadAsync(cancellationToken).NoContext() && reader[0] is not DBNull
+                ? Convert.ToInt64(reader[0])
+                : 0;
 
             return new(SubscriptionId, (ulong)position, DateTime.UtcNow);
-        } catch (Exception) {
-            Log.WarnLog?.Log("Failed to get end of stream");
+        } catch (Exception e) {
+            Log.WarnLog?.Log(e, "Failed to get end of stream");
 
             return EndOfStream.Invalid;
         }

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

@@ -0,0 +1,19 @@
+using Eventuous.SqlServer.Subscriptions;
+using Eventuous.Tests.Subscriptions.Base;
+using Testcontainers.MsSql;
+
+namespace Eventuous.Tests.SqlServer.Subscriptions;
+
+[NotInParallel]
+public class SubscriptionDrop()
+    : SubscriptionDropBase<MsSqlContainer, SqlServerAllStreamSubscription, SqlServerAllStreamSubscriptionOptions, SqlServerCheckpointStore>(
+        new SubscriptionFixture<SqlServerAllStreamSubscription, SqlServerAllStreamSubscriptionOptions, TestEventHandler>(
+            _ => { },
+            false
+        )
+    ) {
+    [Test]
+    public async Task SqlServer_ShouldResubscribeAfterConnectionDrop(CancellationToken cancellationToken) {
+        await ShouldResubscribeAfterConnectionDrop(cancellationToken);
+    }
+}

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

@@ -0,0 +1,19 @@
+using Eventuous.SqlServer.Subscriptions;
+using Eventuous.Tests.Subscriptions.Base;
+using Testcontainers.MsSql;
+
+namespace Eventuous.Tests.SqlServer.Subscriptions;
+
+[NotInParallel]
+public class SubscriptionMeasure()
+    : SubscriptionMeasureBase<MsSqlContainer, SqlServerAllStreamSubscription, SqlServerAllStreamSubscriptionOptions, SqlServerCheckpointStore>(
+        new SubscriptionFixture<SqlServerAllStreamSubscription, SqlServerAllStreamSubscriptionOptions, TestEventHandler>(
+            _ => { },
+            false
+        )
+    ) {
+    [Test]
+    public async Task SqlServer_ShouldMeasureEndOfStream(CancellationToken cancellationToken) {
+        await ShouldMeasureEndOfStream(cancellationToken);
+    }
+}