diff --git a/src/Core/test/Eventuous.Tests.Subscriptions.Base/Fixtures/SubscriptionFixtureBase.cs b/src/Core/test/Eventuous.Tests.Subscriptions.Base/Fixtures/SubscriptionFixtureBase.cs index 2a70b300c..dad909875 100644 --- a/src/Core/test/Eventuous.Tests.Subscriptions.Base/Fixtures/SubscriptionFixtureBase.cs +++ b/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!; + /// + /// 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. + /// + protected internal SubscriptionHealthCheck Health { get; } = new(); + + /// + /// True when the subscription has detected a drop and is trying to resubscribe. + /// + public bool IsDropped => ((EventSubscription)Subscription).IsDropped; + + /// + /// Returns the subscription's end-of-stream measure delegate (requires an ). + /// + 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); diff --git a/src/Core/test/Eventuous.Tests.Subscriptions.Base/SubscriptionDropBase.cs b/src/Core/test/Eventuous.Tests.Subscriptions.Base/SubscriptionDropBase.cs new file mode 100644 index 000000000..b1f35abfc --- /dev/null +++ b/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; + +/// +/// 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 +/// while dropped and 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). +/// +public abstract class SubscriptionDropBase( + SubscriptionFixtureBase fixture + ) : SubscriptionTestBase(fixture) + where TContainer : DockerContainer + where TSubscription : EventSubscription + 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); + + /// + /// 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. + /// + 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 GetHealthStatus(CancellationToken cancellationToken) { + var result = await fixture.Health.CheckHealthAsync(new(), cancellationToken); + + return result.Status; + } + + static Task WaitUntil(Func condition, TimeSpan timeout, CancellationToken cancellationToken) + => WaitUntil(() => Task.FromResult(condition()), timeout, cancellationToken); + + static async Task WaitUntil(Func> 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(); + } + } +} diff --git a/src/Core/test/Eventuous.Tests.Subscriptions.Base/SubscriptionMeasureBase.cs b/src/Core/test/Eventuous.Tests.Subscriptions.Base/SubscriptionMeasureBase.cs new file mode 100644 index 000000000..d60e45311 --- /dev/null +++ b/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; + +/// +/// 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 NULL returned by MAX(...) over an empty table. +/// +public abstract class SubscriptionMeasureBase( + SubscriptionFixtureBase fixture + ) : SubscriptionTestBase(fixture) + where TContainer : DockerContainer + where TSubscription : EventSubscription + 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(); + } + } +} diff --git a/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Subscriptions/SubscriptionDropTests.cs b/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Subscriptions/SubscriptionDropTests.cs new file mode 100644 index 000000000..6188eb189 --- /dev/null +++ b/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( + new CatchUpSubscriptionFixture( + _ => { }, + new("$all"), + false + ) + ) { + [Test] + [Retry(3)] + public async Task Esdb_ShouldResubscribeAfterConnectionDrop(CancellationToken cancellationToken) { + await ShouldResubscribeAfterConnectionDrop(cancellationToken); + } +} diff --git a/src/Postgres/test/Eventuous.Tests.Postgres/Subscriptions/SubscriptionDropTests.cs b/src/Postgres/test/Eventuous.Tests.Postgres/Subscriptions/SubscriptionDropTests.cs new file mode 100644 index 000000000..d129e9d04 --- /dev/null +++ b/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( + new SubscriptionFixture( + _ => { }, + false + ) + ) { + [Test] + public async Task Postgres_ShouldResubscribeAfterConnectionDrop(CancellationToken cancellationToken) { + await ShouldResubscribeAfterConnectionDrop(cancellationToken); + } +} diff --git a/src/Postgres/test/Eventuous.Tests.Postgres/Subscriptions/SubscriptionMeasureTests.cs b/src/Postgres/test/Eventuous.Tests.Postgres/Subscriptions/SubscriptionMeasureTests.cs new file mode 100644 index 000000000..ac25eae0a --- /dev/null +++ b/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( + new SubscriptionFixture( + _ => { }, + false + ) + ) { + [Test] + public async Task Postgres_ShouldMeasureEndOfStream(CancellationToken cancellationToken) { + await ShouldMeasureEndOfStream(cancellationToken); + } +} diff --git a/src/Relational/src/Eventuous.Sql.Base/Subscriptions/SqlSubscriptionBase.cs b/src/Relational/src/Eventuous.Sql.Base/Subscriptions/SqlSubscriptionBase.cs index f63bc6830..343e44965 100644 --- a/src/Relational/src/Eventuous.Sql.Base/Subscriptions/SqlSubscriptionBase.cs +++ b/src/Relational/src/Eventuous.Sql.Base/Subscriptions/SqlSubscriptionBase.cs @@ -341,16 +341,20 @@ async ValueTask 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; } diff --git a/src/SqlServer/test/Eventuous.Tests.SqlServer/Subscriptions/SubscriptionDropTests.cs b/src/SqlServer/test/Eventuous.Tests.SqlServer/Subscriptions/SubscriptionDropTests.cs new file mode 100644 index 000000000..f869749f7 --- /dev/null +++ b/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( + new SubscriptionFixture( + _ => { }, + false + ) + ) { + [Test] + public async Task SqlServer_ShouldResubscribeAfterConnectionDrop(CancellationToken cancellationToken) { + await ShouldResubscribeAfterConnectionDrop(cancellationToken); + } +} diff --git a/src/SqlServer/test/Eventuous.Tests.SqlServer/Subscriptions/SubscriptionMeasureTests.cs b/src/SqlServer/test/Eventuous.Tests.SqlServer/Subscriptions/SubscriptionMeasureTests.cs new file mode 100644 index 000000000..95b28a6cc --- /dev/null +++ b/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( + new SubscriptionFixture( + _ => { }, + false + ) + ) { + [Test] + public async Task SqlServer_ShouldMeasureEndOfStream(CancellationToken cancellationToken) { + await ShouldMeasureEndOfStream(cancellationToken); + } +}