diff --git a/src/Core/test/Eventuous.Tests.Subscriptions/CancelledMessageTests.cs b/src/Core/test/Eventuous.Tests.Subscriptions/CancelledMessageTests.cs
index 41d8df62..26f82ced 100644
--- a/src/Core/test/Eventuous.Tests.Subscriptions/CancelledMessageTests.cs
+++ b/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();
diff --git a/src/Core/test/Eventuous.Tests.Subscriptions/ResubscribeConcurrencyTests.cs b/src/Core/test/Eventuous.Tests.Subscriptions/ResubscribeConcurrencyTests.cs
index 67f83028..7d144e09 100644
--- a/src/Core/test/Eventuous.Tests.Subscriptions/ResubscribeConcurrencyTests.cs
+++ b/src/Core/test/Eventuous.Tests.Subscriptions/ResubscribeConcurrencyTests.cs
@@ -14,7 +14,7 @@ namespace Eventuous.Tests.Subscriptions;
///
public class ResubscribeConcurrencyTests {
///
- /// 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.
///
[Test]
public async Task Burst_of_nacks_produces_a_single_resubscribe(CancellationToken ct) {
@@ -22,7 +22,8 @@ public async Task Burst_of_nacks_produces_a_single_resubscribe(CancellationToken
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() {
///
/// Holds all deliveries until the test releases the burst of failures.
///
- 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 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);
diff --git a/src/Mongo/test/Eventuous.Tests.Projections.MongoDB/ProjectWithBuilder.cs b/src/Mongo/test/Eventuous.Tests.Projections.MongoDB/ProjectWithBuilder.cs
index 251fe6fe..0dc018dc 100644
--- a/src/Mongo/test/Eventuous.Tests.Projections.MongoDB/ProjectWithBuilder.cs
+++ b/src/Mongo/test/Eventuous.Tests.Projections.MongoDB/ProjectWithBuilder.cs
@@ -13,7 +13,6 @@ namespace Eventuous.Tests.Projections.MongoDB;
[ClassDataSource]
public class ProjectWithBuilder(IntegrationFixture fixture) {
[Test]
- [Retry(3)]
[MethodDataSource(typeof(CollectionSource), nameof(CollectionSource.TestOptions))]
public async Task ShouldProjectImported(MongoProjectionOptions? options, CancellationToken cancellationToken) {
var evt = DomainFixture.CreateImportBookingEvent();
diff --git a/src/Postgres/test/Eventuous.Tests.Postgres/Eventuous.Tests.Postgres.csproj b/src/Postgres/test/Eventuous.Tests.Postgres/Eventuous.Tests.Postgres.csproj
index 0763cbe9..63fcb946 100644
--- a/src/Postgres/test/Eventuous.Tests.Postgres/Eventuous.Tests.Postgres.csproj
+++ b/src/Postgres/test/Eventuous.Tests.Postgres/Eventuous.Tests.Postgres.csproj
@@ -16,7 +16,7 @@
-
+
diff --git a/src/Postgres/test/Eventuous.Tests.Postgres/Projections/ProjectorTests.cs b/src/Postgres/test/Eventuous.Tests.Postgres/Projections/ProjectorTests.cs
index 345d6467..6954da0d 100644
--- a/src/Postgres/test/Eventuous.Tests.Postgres/Projections/ProjectorTests.cs
+++ b/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);