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

src/Core/src/Eventuous.Subscriptions/Diagnostics/SubscriptionHealth.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 Microsoft.Extensions.Diagnostics.HealthChecks;
 
 namespace Eventuous.Subscriptions.Diagnostics;
@@ -12,7 +13,10 @@ public interface ISubscriptionHealth {
 }
 
 public class SubscriptionHealthCheck : ISubscriptionHealth, IHealthCheck {
-    readonly Dictionary<string, HealthReport> _healthReports = new();
+    // Reports are written from subscription subscribed/dropped callbacks (background threads) while
+    // CheckHealthAsync enumerates them from the health endpoint. A plain Dictionary throws
+    // "Collection was modified" on concurrent access; ConcurrentDictionary makes both sides safe.
+    readonly ConcurrentDictionary<string, HealthReport> _healthReports = new();
 
     public Task<HealthCheckResult> CheckHealthAsync(HealthCheckContext context, CancellationToken cancellationToken = default) {
         var        unhealthy  = new List<string>();

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/Fixtures/TestEventHandler.cs

@@ -1,3 +1,4 @@
+using System.Collections.Concurrent;
 using Bogus;
 using Eventuous.Subscriptions;
 using Eventuous.Subscriptions.Context;
@@ -28,22 +29,33 @@ public TestEventHandler() : this(null) { }
 
     public int Count { get; private set; }
 
-    readonly Observer<object> _observer = new();
+    readonly Observer<object>       _observer = new();
+    readonly ConcurrentQueue<object> _handled = new();
 
     public On<object> AssertThat() => Hypothesis.On(_observer);
 
     public Hypothesis<object> AssertCollection(TimeSpan deadline, List<object> collection)
         => Hypothesis.On(_observer).Timebox(deadline).Exactly(collection.Count).Match(collection.Contains);
 
+    /// <summary>
+    /// Messages handled so far. Backed by a concurrent queue so tests can poll it while the subscription
+    /// keeps handling on background threads.
+    /// </summary>
+    public IReadOnlyCollection<object> Handled => _handled.ToArray();
+
     public override async ValueTask<EventHandlingStatus> HandleEvent(IMessageConsumeContext context) {
         await Task.Delay(_delay);
         await _observer.Add(context.Message!, context.CancellationToken);
+        _handled.Enqueue(context.Message!);
         Count++;
 
         return EventHandlingStatus.Success;
     }
 
-    public void Reset() => Count = 0;
+    public void Reset() {
+        Count = 0;
+        _handled.Clear();
+    }
 }
 
 public record TestEventHandlerOptions(TimeSpan? Delay = null);

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

@@ -0,0 +1,159 @@
+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;
+using static Eventuous.Sut.App.Commands;
+using static Eventuous.Sut.Domain.BookingEvents;
+
+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) {
+        var subscriptionStarted = false;
+        var containerPaused     = false;
+
+        try {
+            // 1. Produce and consume an initial batch; the subscription must be healthy.
+            await GenerateAndHandleCommands(BatchSize);
+            await fixture.StartSubscription();
+            subscriptionStarted = true;
+            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);
+            containerPaused = true;
+
+            // 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);
+            containerPaused = false;
+
+            // 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. Assert the specific post-recovery
+            // events are handled (by identity), so a replay of the initial batch on resubscribe can't
+            // satisfy this — only genuinely new events count. This is provider-agnostic, unlike comparing
+            // the committed checkpoint to GetLastPosition ($all includes system events the subscription
+            // skips, so its checkpoint never reaches that head).
+            var expected = (await GenerateAndHandleCommands(BatchSize)).Select(ToEvent).ToList();
+            var resumed = await WaitUntil(
+                () => {
+                    var handled = fixture.Handler.Handled;
+
+                    return expected.All(e => handled.Contains(e));
+                },
+                DropTimeout,
+                cancellationToken
+            );
+
+            await fixture.StopSubscription();
+            subscriptionStarted = false;
+
+            await Assert.That(resumed).IsTrue();
+            WriteLine("Processed {0} events after recovery", expected.Count);
+        } finally {
+            // Undo the destructive steps even if an assertion fails or the test is cancelled mid-flight:
+            // a container left paused, or a subscription left resubscribing, would contaminate later tests.
+            // These fixtures use autoStart=false, so fixture disposal won't stop the subscription either.
+            if (containerPaused) {
+                try {
+                    await fixture.Container.UnpauseAsync(CancellationToken.None);
+                } catch (Exception ex) {
+                    WriteLine("Cleanup: failed to unpause the container: {0}", ex.Message);
+                }
+            }
+
+            if (subscriptionStarted) {
+                try {
+                    await fixture.StopSubscription();
+                } catch (Exception ex) {
+                    WriteLine("Cleanup: failed to stop the subscription: {0}", ex.Message);
+                }
+            }
+        }
+    }
+
+    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<List<ImportBooking>> 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();
+        }
+
+        return commands;
+    }
+
+    static BookingImported ToEvent(ImportBooking cmd) => new(cmd.RoomId, cmd.Price, cmd.CheckIn, cmd.CheckOut);
+}

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);
+    }
+}