test: restore queued nack coverage and simplify projection tests

src/Core/test/Eventuous.Tests.Subscriptions/CancelledMessageTests.cs

@@ -100,6 +100,7 @@ public async Task Handler_self_cancellation_is_an_ordinary_failure_and_is_skippe
 
         await subscription.Unsubscribe(_ => { }, ct);
     }
+
     static async Task WaitFor(Task signal, string phase, CancellationToken ct) {
         try {
             await signal.WaitAsync(RecoveryTimeout, ct).NoContext();

src/Core/test/Eventuous.Tests.Subscriptions/ResubscribeConcurrencyTests.cs

@@ -14,15 +14,16 @@ namespace Eventuous.Tests.Subscriptions;
 /// </summary>
 public class ResubscribeConcurrencyTests {
     /// <summary>
-    /// A whole page of messages is in flight before any handler fails, exercising one concurrent drop window.
+    /// A whole page is queued before any handler fails, including messages dequeued after the run fails.
     /// </summary>
     [Test]
     public async Task Burst_of_nacks_produces_a_single_resubscribe(CancellationToken ct) {
         var logs = new CapturingLoggerFactory(LogLevel.Trace);
 
         const int messageCount = 8;
 
-        var handler = new BurstFailingHandler(messageCount, ct);
+        var delivered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        var handler   = new BurstFailingHandler(ct);
 
         await using var subscription = new PumpingSubscription(
             new() {
@@ -34,11 +35,13 @@ public async Task Burst_of_nacks_produces_a_single_resubscribe(CancellationToken
             new NoOpCheckpointStore(),
             new ConsumePipe().AddDefaultConsumer(handler),
             logs,
-            concurrencyLimit: messageCount,
+            concurrencyLimit: 4,
             // Only the first run delivers — a replacement redelivering the same messages would open a second drop window.
             pump: async (sub, transport, start, run) => {
                 if (transport.Index == 0) {
                     for (var i = 0; i < messageCount; i++) await sub.Deliver(run, start + (ulong)i).NoContext();
+
+                    delivered.TrySetResult();
                 }
 
                 await Task.Delay(Timeout.Infinite, run.Token).NoContext();
@@ -48,12 +51,17 @@ public async Task Burst_of_nacks_produces_a_single_resubscribe(CancellationToken
         try {
             await subscription.Subscribe(_ => { }, (_, _, _) => { }, ct);
 
-            // A first nack can cancel the pump. Get every delivery into its handler before allowing any nack.
-            await handler.AllStarted.WaitAsync(TimeSpan.FromSeconds(5), ct).NoContext();
+            // Queue the whole page before a nack can cancel the pump, keeping half the messages behind the four workers.
+            try {
+                await delivered.Task.WaitAsync(TimeSpan.FromSeconds(5), ct).NoContext();
+            } catch (TimeoutException exception) {
+                throw new TimeoutException($"Timed out after 5 seconds waiting for the pump to queue all {messageCount} messages.", exception);
+            }
+
             handler.Release();
 
             (await Wait.Until(() => handler.FailedCount == messageCount, TimeSpan.FromSeconds(5)))
-                .ShouldBeTrue($"all {messageCount} in-flight handlers should have failed, got {handler.FailedCount}");
+                .ShouldBeTrue($"all {messageCount} queued messages should have failed, got {handler.FailedCount}");
 
             (await Wait.Until(() => subscription.SubscribeCalls > 1, TimeSpan.FromSeconds(5))).ShouldBeTrue("the subscription should have resubscribed");
 
@@ -815,21 +823,16 @@ void TrackPumpStarted() {
     /// <summary>
     /// Holds all deliveries until the test releases the burst of failures.
     /// </summary>
-    sealed class BurstFailingHandler(int messageCount, CancellationToken testCancellation) : BaseEventHandler {
-        readonly TaskCompletionSource _allStarted = new(TaskCreationOptions.RunContinuationsAsynchronously);
-        readonly TaskCompletionSource _release    = new(TaskCreationOptions.RunContinuationsAsynchronously);
-        int _started;
+    sealed class BurstFailingHandler(CancellationToken testCancellation) : BaseEventHandler {
+        readonly TaskCompletionSource _release = new(TaskCreationOptions.RunContinuationsAsynchronously);
         int _failed;
 
-        public Task AllStarted => _allStarted.Task;
         public int FailedCount => Volatile.Read(ref _failed);
 
         public void Release() => _release.TrySetResult();
 
         public override async ValueTask<EventHandlingStatus> HandleEvent(IMessageConsumeContext context) {
-            if (Interlocked.Increment(ref _started) == messageCount) _allStarted.TrySetResult();
-
-            // All in-flight operations must fail even if the first nack cancels their subscription run.
+            // Queued messages must fail too, even if the first nack has already cancelled their subscription run.
             await _release.Task.WaitAsync(testCancellation).NoContext();
             Interlocked.Increment(ref _failed);
 

src/Mongo/test/Eventuous.Tests.Projections.MongoDB/ProjectWithBuilder.cs

@@ -13,7 +13,6 @@ namespace Eventuous.Tests.Projections.MongoDB;
 [ClassDataSource<IntegrationFixture>]
 public class ProjectWithBuilder(IntegrationFixture fixture) {
     [Test]
-    [Retry(3)]
     [MethodDataSource(typeof(CollectionSource), nameof(CollectionSource.TestOptions))]
     public async Task ShouldProjectImported(MongoProjectionOptions<BookingDocument>? options, CancellationToken cancellationToken) {
         var evt               = DomainFixture.CreateImportBookingEvent();

src/Postgres/test/Eventuous.Tests.Postgres/Eventuous.Tests.Postgres.csproj

@@ -16,7 +16,7 @@
         </PackageReference>
     </ItemGroup>
     <ItemGroup>
-        <ProjectReference Include="$(LocalRoot)\Eventuous.Postgresql\Eventuous.Postgresql.csproj" Aliases="global,Postgresql"/>
+        <ProjectReference Include="$(LocalRoot)\Eventuous.Postgresql\Eventuous.Postgresql.csproj"/>
         <ProjectReference Include="$(DiagRoot)\Eventuous.Diagnostics.Logging\Eventuous.Diagnostics.Logging.csproj"/>
         <ProjectReference Include="$(ExtRoot)\Eventuous.Extensions.DependencyInjection\Eventuous.Extensions.DependencyInjection.csproj"/>
         <ProjectReference Include="$(CoreTestRoot)\Eventuous.Tests.Persistence.Base\Eventuous.Tests.Persistence.Base.csproj"/>

src/Postgres/test/Eventuous.Tests.Postgres/Projections/ProjectorTests.cs

@@ -1,5 +1,3 @@
-extern alias Postgresql;
-
 using Eventuous.Postgresql;
 using Eventuous.Postgresql.Projections;
 using Eventuous.Postgresql.Subscriptions;
@@ -8,7 +6,6 @@
 using Eventuous.Tests.Persistence.Base.Fixtures;
 using Eventuous.Tests.Postgres.Subscriptions;
 using Npgsql;
-using static Postgresql::Eventuous.Tools.TaskExtensions;
 using Assert = TUnit.Assertions.Assert;
 
 namespace Eventuous.Tests.Postgres.Projections;
@@ -29,7 +26,7 @@ public async Task ProjectImportedBookingsToTable(CancellationToken cancellationT
         await CreateSchema();
         var commands = await GenerateAndProduceEvents(100);
 
-        await WaitForBookings(commands.Count, cancellationToken).NoContext();
+        await WaitForBookings(commands.Count, cancellationToken);
 
         await using var connection = await _fixture.DataSource.OpenConnectionAsync(cancellationToken);
 
@@ -39,7 +36,7 @@ public async Task ProjectImportedBookingsToTable(CancellationToken cancellationT
             await using var cmd = new NpgsqlCommand(select, connection);
             cmd.Parameters.AddWithValue("@bookingId", command.BookingId);
             await using var reader = await cmd.ExecuteReaderAsync(cancellationToken);
-            await Assert.That(await reader.ReadAsync(cancellationToken).NoContext()).IsTrue();
+            await Assert.That(await reader.ReadAsync(cancellationToken)).IsTrue();
             await Assert.That(reader["checkin_date"]).IsEqualTo(command.CheckIn.ToDateTimeUnspecified());
             await Assert.That(reader["price"]).IsEqualTo((decimal)command.Price);
         }
@@ -52,14 +49,14 @@ async Task WaitForBookings(int expectedCount, CancellationToken cancellationToke
         long projectedCount = 0;
 
         try {
-            await using var connection = await _fixture.DataSource.OpenConnectionAsync(cts.Token).NoContext();
+            await using var connection = await _fixture.DataSource.OpenConnectionAsync(cts.Token);
             await using var cmd        = new NpgsqlCommand($"select count(*) from {_fixture.SchemaName}.bookings", connection);
 
             while (true) {
-                projectedCount = (long)(await cmd.ExecuteScalarAsync(cts.Token).NoContext())!;
+                projectedCount = (long)(await cmd.ExecuteScalarAsync(cts.Token))!;
                 if (projectedCount == expectedCount) return;
 
-                await Task.Delay(100, cts.Token).NoContext();
+                await Task.Delay(100, cts.Token);
             }
         } catch (OperationCanceledException ex) when (cts.IsCancellationRequested && !cancellationToken.IsCancellationRequested) {
             throw new TimeoutException($"Expected {expectedCount} projected bookings within 30 seconds, but observed {projectedCount}.", ex);