From 6ffc3e7dd6536a75e51b8d917a1f5a63357c1c24 Mon Sep 17 00:00:00 2001 From: "Jeremy D. Miller" Date: Wed, 22 Jul 2026 06:10:17 -0500 Subject: [PATCH] #5022: kill the shutdown-telemetry race in extended_progression_batch_write The new extended_progression_batch_write tests (jasperfx#553 / #5008) started a live daemon, waited for it to catch up, then StopAllAsync()'d it before asserting on the mt_event_progression rows. Daemon shutdown itself emits a "Stopped" extended-progression heartbeat asynchronously and does not drain it before the daemon is considered stopped, so it races back in AFTER the explicit WriteExtendedProgressionAsync batch write and clobbers the rows under assertion (the intermittent `should be "Paused" but was "Stopped"` CI failure). The helper's own comment had it backwards: stopping the daemon was the source of the racing write, not a guard against it. Nothing in these tests needs a running daemon. Seed the two committed progression rows directly via mt_mark_event_progression so the only writer left against those rows is the WriteExtendedProgressionAsync call under test. This removes the race (and the shutdown-path ObjectDisposedException) entirely and makes the tests deterministic, unblocking #5008. Test-only change (issue #5022 direction 1). The product-side drain of in-flight shutdown heartbeats before disposing the JasperFx.Events shutdown semaphore (Symptom 2) lives in JasperFx.Events.Daemon.GroupedProjectionExecution and is tracked separately as the cross-repo fix. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../extended_progression_batch_write.cs | 40 ++++++++++--------- 1 file changed, 22 insertions(+), 18 deletions(-) diff --git a/src/DaemonTests/extended_progression_batch_write.cs b/src/DaemonTests/extended_progression_batch_write.cs index 41d47a448f..6fa048c348 100644 --- a/src/DaemonTests/extended_progression_batch_write.cs +++ b/src/DaemonTests/extended_progression_batch_write.cs @@ -1,7 +1,7 @@ using System; using System.Threading.Tasks; using DaemonTests.TestingSupport; -using JasperFx.Core; +using JasperFx.Events; using JasperFx.Events.Daemon; using JasperFx.Events.Projections; using Marten.Events.Aggregation; @@ -70,7 +70,15 @@ private async Task countRowsAsync() return Convert.ToInt64(raw); } - private async Task startCaughtUpDaemonAsync() + // #5022: the original helper started a live daemon, waited for it to catch up, then + // StopAllAsync()'d it before the tests asserted on the rows. Daemon shutdown itself emits a + // "Stopped" extended-progression heartbeat ASYNCHRONOUSLY, which is not drained before the daemon + // is considered stopped — so it races back in and clobbers the very rows these tests assert on + // (the intermittent `should be "Paused" but was "Stopped"` failure), and can reach an + // already-disposed shutdown SemaphoreSlim on the JasperFx.Events side. Nothing here actually needs + // a running daemon: seed the committed progression rows directly via mt_mark_event_progression so + // the ONLY writer left against these rows is the WriteExtendedProgressionAsync call under test. + private async Task seedProgressionRowsAsync() { StoreOptions(x => { @@ -79,23 +87,19 @@ private async Task startCaughtUpDaemonAsync() x.Projections.Add(new OtherBatchTelemetryProjection(), ProjectionLifecycle.Async); }); - var daemon = await StartDaemon(); + // Build the event storage (mt_event_progression + the mt_mark_event_progression* functions) + // without ever starting a daemon. + var database = (MartenDatabase)theStore.Storage.Database; + await database.EnsureStorageExistsAsync(typeof(IEvent)); - await using (var session = theStore.LightweightSession()) + await using var session = theStore.LightweightSession(); + foreach (var shard in new[] { "BatchTelemetryStream:All", "OtherBatchTelemetry:All" }) { - for (var i = 0; i < 10; i++) - { - session.Events.Append(Guid.NewGuid(), new BatchTelemetryEvent()); - } - - await session.SaveChangesAsync(); + session.QueueSqlCommand( + $"select {theStore.Events.DatabaseSchemaName}.mt_mark_event_progression(?, ?)", shard, 10L); } - await daemon.Tracker.WaitForShardState(new ShardState("BatchTelemetryStream:All", 10), 30.Seconds()); - await daemon.Tracker.WaitForShardState(new ShardState("OtherBatchTelemetry:All", 10), 30.Seconds()); - - // Stop the daemon so nothing else races telemetry writes into the rows we assert on - await daemon.StopAllAsync(); + await session.SaveChangesAsync(); } private static ShardState telemetry(string shard, string status, string? reason = null, int? node = null) @@ -113,7 +117,7 @@ private static ShardState telemetry(string shard, string status, string? reason [Fact] public async Task updates_every_existing_row_in_one_batch() { - await startCaughtUpDaemonAsync(); + await seedProgressionRowsAsync(); var database = (MartenDatabase)theStore.Storage.Database; @@ -137,7 +141,7 @@ await database.WriteExtendedProgressionAsync([ [Fact] public async Task never_inserts_a_row_and_never_touches_progression() { - await startCaughtUpDaemonAsync(); + await seedProgressionRowsAsync(); var database = (MartenDatabase)theStore.Storage.Database; var rowsBefore = await countRowsAsync(); @@ -161,7 +165,7 @@ await database.WriteExtendedProgressionAsync([ [Fact] public async Task an_empty_batch_is_a_no_op_and_a_single_state_batch_delegates() { - await startCaughtUpDaemonAsync(); + await seedProgressionRowsAsync(); var database = (MartenDatabase)theStore.Storage.Database;