From f0b6b911cf774653a0b83eb8411339a1654b4199 Mon Sep 17 00:00:00 2001 From: Alexey Zimarev Date: Tue, 30 Jun 2026 18:32:27 +0200 Subject: [PATCH 1/4] test: add subscription drop/resubscribe + health tests (#308) Add an infra-agnostic `SubscriptionDropBase` that simulates a transport drop by pausing the infrastructure container, then asserts the subscription is dropped and the health check reports Unhealthy, unpauses, and asserts resubscribe, Healthy again, and continued processing of new events. Wire the subscription's subscribed/dropped callbacks in the shared fixture to a `SubscriptionHealthCheck` (mirroring `SubscriptionHostedService`) and expose `IsDropped` so tests can observe drop state. Covered for the container-backed stores: SQL Server, PostgreSQL, and KurrentDB (ESDB). SQLite is excluded as its fixture isn't container-backed. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../Fixtures/SubscriptionFixtureBase.cs | 25 +++- .../SubscriptionDropBase.cs | 114 ++++++++++++++++++ .../Subscriptions/SubscriptionDropTests.cs | 21 ++++ .../Subscriptions/SubscriptionDropTests.cs | 20 +++ .../Subscriptions/SubscriptionDropTests.cs | 19 +++ 5 files changed, 198 insertions(+), 1 deletion(-) create mode 100644 src/Core/test/Eventuous.Tests.Subscriptions.Base/SubscriptionDropBase.cs create mode 100644 src/KurrentDB/test/Eventuous.Tests.KurrentDB/Subscriptions/SubscriptionDropTests.cs create mode 100644 src/Postgres/test/Eventuous.Tests.Postgres/Subscriptions/SubscriptionDropTests.cs create mode 100644 src/SqlServer/test/Eventuous.Tests.SqlServer/Subscriptions/SubscriptionDropTests.cs 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..b1eb696f6 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,31 @@ 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; + 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/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/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); + } +} From 8c873c5676b16c662cd6c58404d02cd49906e068 Mon Sep 17 00:00:00 2001 From: Alexey Zimarev Date: Tue, 30 Jun 2026 18:40:46 +0200 Subject: [PATCH 2/4] fix: correct relational subscription end-of-stream measure (#548) `SqlSubscriptionBase.GetSubscriptionEndOfStream` had three defects that made the gap/lag diagnostics log "Failed to get end of stream" repeatedly: - Swapped switch arms: an All subscription queried MAX(stream_position) and a Stream subscription queried MAX(global_position). - `reader.GetInt64(0)` threw `InvalidCastException` when the provider returned the position column as Int32. - No DBNull guard, so MAX(...) over an empty table threw. Query the correct column per kind, guard against DBNull, and use `Convert.ToInt64` so either integer width is accepted. Also log the actual exception instead of swallowing it. Add `SubscriptionMeasureBase` plus SQL Server and PostgreSQL subclasses that assert the measure reports position 0 on an empty store and the global end position after events are appended. Verified to fail against the unfixed code. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../Fixtures/SubscriptionFixtureBase.cs | 5 ++ .../SubscriptionMeasureBase.cs | 53 +++++++++++++++++++ .../Subscriptions/SubscriptionMeasureTests.cs | 20 +++++++ .../Subscriptions/SqlSubscriptionBase.cs | 14 +++-- .../Subscriptions/SubscriptionMeasureTests.cs | 19 +++++++ 5 files changed, 106 insertions(+), 5 deletions(-) create mode 100644 src/Core/test/Eventuous.Tests.Subscriptions.Base/SubscriptionMeasureBase.cs create mode 100644 src/Postgres/test/Eventuous.Tests.Postgres/Subscriptions/SubscriptionMeasureTests.cs create mode 100644 src/SqlServer/test/Eventuous.Tests.SqlServer/Subscriptions/SubscriptionMeasureTests.cs 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 b1eb696f6..dad909875 100644 --- a/src/Core/test/Eventuous.Tests.Subscriptions.Base/Fixtures/SubscriptionFixtureBase.cs +++ b/src/Core/test/Eventuous.Tests.Subscriptions.Base/Fixtures/SubscriptionFixtureBase.cs @@ -42,6 +42,11 @@ protected SubscriptionFixtureBase(bool autoStart = true, LogLevel logLevel = Log /// 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() 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/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/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); + } +} From fafd6c7900bcf665fb3682f7d3554aa8a23235e6 Mon Sep 17 00:00:00 2001 From: Alexey Zimarev Date: Wed, 15 Jul 2026 18:10:51 +0200 Subject: [PATCH 3/4] fix(subscriptions): thread-safe health reports; harden drop test MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address review findings on #551: - SubscriptionHealthCheck stored reports in a plain Dictionary that CheckHealthAsync enumerated while subscription callbacks mutated it from background threads — a real "Collection was modified" race on the production singleton. Switch to ConcurrentDictionary (Qodo). - SubscriptionDropBase now waits until the initial batch is committed to the checkpoint before pausing, so replayed events can't spuriously satisfy the post-recovery assertion (Codex). - Wrap the pause/subscription flow in try/finally so a failed assertion or cancellation can't leave the container paused or the subscription resubscribing into later tests — these fixtures use autoStart=false, so disposal won't stop the subscription (Qodo). Co-Authored-By: Claude Opus 4.8 (1M context) --- .../Diagnostics/SubscriptionHealth.cs | 6 +- .../SubscriptionDropBase.cs | 120 ++++++++++++------ 2 files changed, 87 insertions(+), 39 deletions(-) diff --git a/src/Core/src/Eventuous.Subscriptions/Diagnostics/SubscriptionHealth.cs b/src/Core/src/Eventuous.Subscriptions/Diagnostics/SubscriptionHealth.cs index d7cb29863..97d4cab12 100644 --- a/src/Core/src/Eventuous.Subscriptions/Diagnostics/SubscriptionHealth.cs +++ b/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 _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 _healthReports = new(); public Task CheckHealthAsync(HealthCheckContext context, CancellationToken cancellationToken = default) { var unhealthy = new List(); diff --git a/src/Core/test/Eventuous.Tests.Subscriptions.Base/SubscriptionDropBase.cs b/src/Core/test/Eventuous.Tests.Subscriptions.Base/SubscriptionDropBase.cs index b1f35abfc..8694924ea 100644 --- a/src/Core/test/Eventuous.Tests.Subscriptions.Base/SubscriptionDropBase.cs +++ b/src/Core/test/Eventuous.Tests.Subscriptions.Base/SubscriptionDropBase.cs @@ -34,44 +34,88 @@ SubscriptionFixtureBase 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); + 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); + + // Wait until the initial batch is committed to the checkpoint. Otherwise the subscription + // replays those uncommitted events on resubscribe, and the replay could satisfy the + // post-recovery assertion below without any newly produced event ever being processed. + var lastPosition = await fixture.GetLastPosition(); + var checkpointed = await WaitUntil( + async () => { + var checkpoint = await fixture.CheckpointStore.GetLastCheckpoint(fixture.SubscriptionId, cancellationToken); + + return checkpoint.Position >= lastPosition; + }, + DropTimeout, + cancellationToken + ); + await Assert.That(checkpointed).IsTrue(); + + // 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. The initial batch is already + // checkpointed, so it cannot be replayed and this count only advances for new events. + var countBeforeRecovery = fixture.Handler.Count; + await GenerateAndHandleCommands(BatchSize); + var resumed = await WaitUntil(() => fixture.Handler.Count >= countBeforeRecovery + BatchSize, DropTimeout, cancellationToken); + + await fixture.StopSubscription(); + subscriptionStarted = false; + + await Assert.That(resumed).IsTrue(); + WriteLine("Processed {0} events after recovery", fixture.Handler.Count - countBeforeRecovery); + } 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 GetHealthStatus(CancellationToken cancellationToken) { From 1f4457c7675dbd43d364d904fab0304137611595 Mon Sep 17 00:00:00 2001 From: Alexey Zimarev Date: Wed, 15 Jul 2026 19:57:35 +0200 Subject: [PATCH 4/4] fix(test): assert post-recovery events by identity, not checkpoint position The checkpoint-await added in the previous commit timed out for the KurrentDB $all drop test: GetLastPosition reads the $all head, which includes system events the AllStreamSubscription skips, so its committed checkpoint never reaches that position. Relational stores passed only because their GetLastPosition reads the user-event table. Replace it with the provider-agnostic form of the same guard: capture the specific events produced after recovery and wait until the handler has handled those exact events (by record identity). A replay of the initial batch on resubscribe can no longer satisfy the assertion. TestEventHandler now exposes handled messages via a concurrent queue. Verified: Postgres_ShouldResubscribeAfterConnectionDrop passes locally. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../Fixtures/TestEventHandler.cs | 16 ++++++- .../SubscriptionDropBase.cs | 45 ++++++++++--------- 2 files changed, 37 insertions(+), 24 deletions(-) diff --git a/src/Core/test/Eventuous.Tests.Subscriptions.Base/Fixtures/TestEventHandler.cs b/src/Core/test/Eventuous.Tests.Subscriptions.Base/Fixtures/TestEventHandler.cs index 8ef13f177..21092da92 100644 --- a/src/Core/test/Eventuous.Tests.Subscriptions.Base/Fixtures/TestEventHandler.cs +++ b/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 _observer = new(); + readonly Observer _observer = new(); + readonly ConcurrentQueue _handled = new(); public On AssertThat() => Hypothesis.On(_observer); public Hypothesis AssertCollection(TimeSpan deadline, List collection) => Hypothesis.On(_observer).Timebox(deadline).Exactly(collection.Count).Match(collection.Contains); + /// + /// Messages handled so far. Backed by a concurrent queue so tests can poll it while the subscription + /// keeps handling on background threads. + /// + public IReadOnlyCollection Handled => _handled.ToArray(); + public override async ValueTask 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); diff --git a/src/Core/test/Eventuous.Tests.Subscriptions.Base/SubscriptionDropBase.cs b/src/Core/test/Eventuous.Tests.Subscriptions.Base/SubscriptionDropBase.cs index 8694924ea..721ef9eab 100644 --- a/src/Core/test/Eventuous.Tests.Subscriptions.Base/SubscriptionDropBase.cs +++ b/src/Core/test/Eventuous.Tests.Subscriptions.Base/SubscriptionDropBase.cs @@ -4,6 +4,8 @@ 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; @@ -46,21 +48,6 @@ protected async Task ShouldResubscribeAfterConnectionDrop(CancellationToken canc await Assert.That(consumedInitial).IsTrue(); await Assert.That(await GetHealthStatus(cancellationToken)).IsEqualTo(HealthStatus.Healthy); - // Wait until the initial batch is committed to the checkpoint. Otherwise the subscription - // replays those uncommitted events on resubscribe, and the replay could satisfy the - // post-recovery assertion below without any newly produced event ever being processed. - var lastPosition = await fixture.GetLastPosition(); - var checkpointed = await WaitUntil( - async () => { - var checkpoint = await fixture.CheckpointStore.GetLastCheckpoint(fixture.SubscriptionId, cancellationToken); - - return checkpoint.Position >= lastPosition; - }, - DropTimeout, - cancellationToken - ); - await Assert.That(checkpointed).IsTrue(); - // 2. Drop the connection by pausing the container. WriteLine("Pausing the container to drop the connection"); await fixture.Container.PauseAsync(cancellationToken); @@ -85,17 +72,27 @@ protected async Task ShouldResubscribeAfterConnectionDrop(CancellationToken canc await Assert.That(healthy).IsTrue(); WriteLine("Subscription resubscribed and reported healthy"); - // 6. Events produced after recovery must be processed. The initial batch is already - // checkpointed, so it cannot be replayed and this count only advances for new events. - var countBeforeRecovery = fixture.Handler.Count; - await GenerateAndHandleCommands(BatchSize); - var resumed = await WaitUntil(() => fixture.Handler.Count >= countBeforeRecovery + BatchSize, DropTimeout, cancellationToken); + // 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", fixture.Handler.Count - countBeforeRecovery); + 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. @@ -142,7 +139,7 @@ static async Task WaitUntil(Func> condition, TimeSpan timeout, } } - async Task GenerateAndHandleCommands(int count) { + async Task> GenerateAndHandleCommands(int count) { var commands = Enumerable .Range(0, count) .Select(_ => DomainFixture.CreateImportBooking()) @@ -154,5 +151,9 @@ async Task GenerateAndHandleCommands(int count) { 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); }