dohertj2/ScadaBridge
1
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity

test(auditlog): close register row 95 — stop the three remaining leaked fast-tick actors

Browse Source 
Row 82 (da23f2dc) fixed this leak shape in PartitionPurgeTests and
NotifyDispatcherAuditTrailTests; three classes in the same assembly still had
it. Each creates an actor with a fast periodic tick (100 ms purge override,
~0 s reconciliation override) against a shared MsSqlMigrationFixture database
and never stops it deterministically, relying on TestKit's Dispose -> Shutdown
— which is timeout-bounded and does NOT fail when it expires. A survivor keeps
mutating the fixture DB while LATER test classes run.

All three now use the same Watch(actor) / Sys.Stop(actor) /
ExpectTerminated(actor) helper da23f2dc established, applied at EVERY
actor-creation site — 11 in AuditLogPurgeActorTests, 10 in
SiteAuditReconciliationActorTests, 2 in OutageReconciliationTests.

The two real-DB cases are the ones that could corrupt another class:
AuditLogPurgeActorTests' EndToEnd_RealPartition_… keeps switching partitions
out of the shared DB, and SiteAuditReconciliationActorTests'
Tick_Duplicates_NotDoubleInserted holds a live AuditLogRepository over a
DbContext the test is about to dispose. The stub-repo cases are cheaper but not
free: each burns a DI scope and a thread every 100 ms for the rest of the
assembly run.

OutageReconciliationTests needed more than the helper. It already called
Sys.Stop, but (a) fire-and-forget, so the call returns before the in-flight tick
has finished writing to MSSQL, and (b) from the LAST line of the enclosing
`await using` block, so any earlier assertion failure skipped it entirely. Both
stops move to immediately after the last step that needs the actor alive —
before the read-back that counts rows — and the AuditLogIngestActor that
BuildCentralPipeline creates is now stopped too (test 2 was discarding its ref
outright, leaving it holding a service provider the block then disposed).

Where a test's remaining assertions read a collection the actor appends to
(client.Calls indexing, Assert.Empty on a switched-boundaries list), the stop is
placed BEFORE them so they no longer read a moving target.

AuditLog.Tests: 383/383 green (1 m 51 s), 0 skipped.
This commit is contained in:
Joseph Doherty
2026-08-17 05:27:37 -04:00
parent 26e3a1b9a6
commit 34a8079d15
3 changed files with 184 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/AuditLogPurgeActorTests.cs
+71 -11
View File
@@ -138,6 +138,38 @@ public class AuditLogPurgeActorTests : TestKit, IClassFixture<MsSqlMigrationFixt
NullLogger<AuditLogPurgeActor>.Instance)));
}
/// <summary>
/// Stops a purge actor and blocks until it is actually dead.
/// </summary>
/// <remarks>
/// <para>
/// Deferred-work register row 95, the same leak shape row 82 fixed in
/// <c>PartitionPurgeTests</c> / <c>NotifyDispatcherAuditTrailTests</c>. Every actor this
/// class creates ticks every <c>100 ms</c>
/// (<see cref="AuditLogPurgeOptions.IntervalOverride"/>) and was only ever stopped
/// implicitly by <see cref="TestKit"/>'s <c>Dispose</c> → <c>Shutdown</c> — which is
/// bounded by a timeout and does NOT fail when that timeout expires. So on a loaded
/// machine an actor kept ticking while LATER test classes ran: the stub-repo tests burn a
/// DI scope and a thread every 100 ms for the rest of the assembly run, and
/// <c>EndToEnd_RealPartition_…</c>'s actor keeps switching partitions out of the shared
/// <see cref="MsSqlMigrationFixture"/> database that other classes in this assembly seed
/// against.
/// </para>
/// <para>
/// Stopping the actor inside the test that created it removes the dependency on
/// ActorSystem-shutdown timing entirely.
/// </para>
/// </remarks>
private void StopPurgeActor(IActorRef actor)
{
Watch(actor);
Sys.Stop(actor);
// A stop is processed after the in-flight tick completes, so this also guarantees no
// partition switch is still running against the fixture DB.
ExpectTerminated(actor, TimeSpan.FromSeconds(30));
Unwatch(actor);
}
private static AuditLogPurgeOptions FastTickOptions(TimeSpan? interval = null) => new()
{
IntervalHours = 24,
@@ -163,7 +195,7 @@ public class AuditLogPurgeActorTests : TestKit, IClassFixture<MsSqlMigrationFixt
public void Tick_Fires_OnDailyInterval()
{
var repo = new RecordingRepo();
CreateActor(repo, FastTickOptions());
var actor = CreateActor(repo, FastTickOptions());
// The first scheduled tick fires after the configured interval. We
// assert the visible side effect (the enumerator was called) rather
@@ -173,6 +205,8 @@ public class AuditLogPurgeActorTests : TestKit, IClassFixture<MsSqlMigrationFixt
$"expected >= 1 enumerator call, got {repo.ThresholdQueries.Count}"),
duration: TimeSpan.FromSeconds(3),
interval: TimeSpan.FromMilliseconds(50));
StopPurgeActor(actor);
}
// ---------------------------------------------------------------------
@@ -192,7 +226,7 @@ public class AuditLogPurgeActorTests : TestKit, IClassFixture<MsSqlMigrationFixt
RowsPerBoundary = _ => 42L,
};
CreateActor(repo, FastTickOptions());
var actor = CreateActor(repo, FastTickOptions());
AwaitAssert(
() =>
@@ -202,6 +236,8 @@ public class AuditLogPurgeActorTests : TestKit, IClassFixture<MsSqlMigrationFixt
},
duration: TimeSpan.FromSeconds(3),
interval: TimeSpan.FromMilliseconds(50));
StopPurgeActor(actor);
}
// ---------------------------------------------------------------------
@@ -217,7 +253,7 @@ public class AuditLogPurgeActorTests : TestKit, IClassFixture<MsSqlMigrationFixt
// and asserting the actor switched nothing despite the tick firing.
var repo = new RecordingRepo { Boundaries = new List<DateTime>() };
CreateActor(repo, FastTickOptions());
var actor = CreateActor(repo, FastTickOptions());
// Wait for at least one tick (visible via the enumerator call) then
// assert no switch happened.
@@ -226,6 +262,10 @@ public class AuditLogPurgeActorTests : TestKit, IClassFixture<MsSqlMigrationFixt
duration: TimeSpan.FromSeconds(3),
interval: TimeSpan.FromMilliseconds(50));
// Stop before the negative assertion: a still-ticking actor is exactly what would
// make it read a moving target (register row 95).
StopPurgeActor(actor);
Assert.Empty(repo.SwitchedBoundaries);
}
@@ -244,9 +284,10 @@ public class AuditLogPurgeActorTests : TestKit, IClassFixture<MsSqlMigrationFixt
};
var probe = SubscribePurged();
CreateActor(repo, FastTickOptions());
var actor = CreateActor(repo, FastTickOptions());
var msg = probe.ExpectMsg<AuditLogPurgedEvent>(TimeSpan.FromSeconds(5));
StopPurgeActor(actor);
Assert.Equal(boundary, msg.MonthBoundary);
Assert.Equal(1234L, msg.RowsDeleted);
Assert.True(msg.DurationMs >= 0,
@@ -269,7 +310,7 @@ public class AuditLogPurgeActorTests : TestKit, IClassFixture<MsSqlMigrationFixt
BoundaryException = new InvalidOperationException("simulated switch failure for poison boundary"),
};
CreateActor(repo, FastTickOptions());
var actor = CreateActor(repo, FastTickOptions());
AwaitAssert(
() =>
@@ -281,6 +322,8 @@ public class AuditLogPurgeActorTests : TestKit, IClassFixture<MsSqlMigrationFixt
},
duration: TimeSpan.FromSeconds(3),
interval: TimeSpan.FromMilliseconds(50));
StopPurgeActor(actor);
}
// ---------------------------------------------------------------------
@@ -349,7 +392,7 @@ public class AuditLogPurgeActorTests : TestKit, IClassFixture<MsSqlMigrationFixt
};
var probe = SubscribePurged();
Sys.ActorOf(Props.Create(() => new AuditLogPurgeActor(
var actor = Sys.ActorOf(Props.Create(() => new AuditLogPurgeActor(
sp,
Options.Create(purgeOptions),
Options.Create(auditOptions),
@@ -363,6 +406,11 @@ public class AuditLogPurgeActorTests : TestKit, IClassFixture<MsSqlMigrationFixt
isMessage: m => m.MonthBoundary == janBoundary,
max: TimeSpan.FromSeconds(30));
// This is the one actor in the class that mutates the SHARED fixture database, so it
// must not outlive the test and start switching partitions out from under the next
// class's seed rows (register row 95).
StopPurgeActor(actor);
Assert.True(matched.RowsDeleted >= 1,
$"Expected RowsDeleted >= 1 for the Jan-2026 partition; got {matched.RowsDeleted}.");
@@ -394,7 +442,7 @@ public class AuditLogPurgeActorTests : TestKit, IClassFixture<MsSqlMigrationFixt
// or some other accidental default. We use a non-default retention
// (30 days) so the assertion isn't satisfied by the 365 default.
var repo = new RecordingRepo();
CreateActor(
var actor = CreateActor(
repo,
FastTickOptions(),
auditOptions: new AuditLogOptions { RetentionDays = 30 });
@@ -404,6 +452,8 @@ public class AuditLogPurgeActorTests : TestKit, IClassFixture<MsSqlMigrationFixt
duration: TimeSpan.FromSeconds(3),
interval: TimeSpan.FromMilliseconds(50));
StopPurgeActor(actor);
var threshold = repo.ThresholdQueries[0];
var expected = DateTime.UtcNow - TimeSpan.FromDays(30);
// 1-minute slack covers test-thread scheduling jitter between the
@@ -437,7 +487,7 @@ public class AuditLogPurgeActorTests : TestKit, IClassFixture<MsSqlMigrationFixt
var purgeOptionsWrapped = Options.Create(purgeOptions);
var sp = BuildScopedProvider(repo);
Sys.ActorOf(Props.Create(() => new AuditLogPurgeActor(
var actor = Sys.ActorOf(Props.Create(() => new AuditLogPurgeActor(
sp,
purgeOptionsWrapped,
auditOptions,
@@ -448,6 +498,8 @@ public class AuditLogPurgeActorTests : TestKit, IClassFixture<MsSqlMigrationFixt
duration: TimeSpan.FromSeconds(3),
interval: TimeSpan.FromMilliseconds(50));
StopPurgeActor(actor);
var purge = repo.ChannelPurges.First(p => p.Channel == "ApiOutbound");
Assert.Equal(1234, purge.BatchSize);
@@ -483,7 +535,7 @@ public class AuditLogPurgeActorTests : TestKit, IClassFixture<MsSqlMigrationFixt
var purgeOptions = Options.Create(FastTickOptions());
var sp = BuildScopedProvider(repo);
Sys.ActorOf(Props.Create(() => new AuditLogPurgeActor(
var actor = Sys.ActorOf(Props.Create(() => new AuditLogPurgeActor(
sp,
purgeOptions,
auditOptions,
@@ -496,6 +548,10 @@ public class AuditLogPurgeActorTests : TestKit, IClassFixture<MsSqlMigrationFixt
duration: TimeSpan.FromSeconds(3),
interval: TimeSpan.FromMilliseconds(50));
// Stop before the negative assertion: a still-ticking actor is exactly what would
// make it read a moving target (register row 95).
StopPurgeActor(actor);
Assert.Empty(repo.ChannelPurges);
}
@@ -527,7 +583,7 @@ public class AuditLogPurgeActorTests : TestKit, IClassFixture<MsSqlMigrationFixt
});
var sp = BuildScopedProvider(repo);
Sys.ActorOf(Props.Create(() => new AuditLogPurgeActor(
var actor = Sys.ActorOf(Props.Create(() => new AuditLogPurgeActor(
sp,
Options.Create(purgeOptions),
auditOptions,
@@ -542,6 +598,8 @@ public class AuditLogPurgeActorTests : TestKit, IClassFixture<MsSqlMigrationFixt
},
duration: TimeSpan.FromSeconds(3),
interval: TimeSpan.FromMilliseconds(50));
StopPurgeActor(actor);
}
// ---------------------------------------------------------------------
@@ -572,7 +630,7 @@ public class AuditLogPurgeActorTests : TestKit, IClassFixture<MsSqlMigrationFixt
var probe = CreateTestProbe();
Sys.EventStream.Subscribe(probe.Ref, typeof(AuditLogPurgeFailedEvent));
Sys.ActorOf(Props.Create(() => new AuditLogPurgeActor(
var actor = Sys.ActorOf(Props.Create(() => new AuditLogPurgeActor(
sp,
Options.Create(FastTickOptions()),
Options.Create(new AuditLogOptions()),
@@ -588,5 +646,7 @@ public class AuditLogPurgeActorTests : TestKit, IClassFixture<MsSqlMigrationFixt
$"expected PurgeFailures >= 1, got {snapshot.PurgeFailures}"),
duration: TimeSpan.FromSeconds(3),
interval: TimeSpan.FromMilliseconds(50));
StopPurgeActor(actor);
}
}
tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/SiteAuditReconciliationActorTests.cs
+68 -11
View File
@@ -252,6 +252,37 @@ public class SiteAuditReconciliationActorTests : TestKit, IClassFixture<MsSqlMig
NullLogger<SiteAuditReconciliationActor>.Instance)));
}
/// <summary>
/// Stops a reconciliation actor and blocks until it is actually dead.
/// </summary>
/// <remarks>
/// <para>
/// Deferred-work register row 95, the same leak shape row 82 fixed in
/// <c>PartitionPurgeTests</c> / <c>NotifyDispatcherAuditTrailTests</c>. Every actor this
/// class creates ticks on a ~0 s repeat interval (<see cref="FastTickOptions"/>) and was
/// only ever stopped implicitly by <see cref="TestKit"/>'s <c>Dispose</c> →
/// <c>Shutdown</c>, which is bounded by a timeout and does NOT fail when it expires. A
/// survivor keeps pulling and INSERTING audit rows: in
/// <c>Tick_Duplicates_NotDoubleInserted</c> it holds a real
/// <see cref="AuditLogRepository"/> over the shared <see cref="MsSqlMigrationFixture"/>
/// database (and over a <c>DbContext</c> the test is about to dispose), so it writes into
/// the DB LATER test classes read from.
/// </para>
/// <para>
/// Stopping the actor inside the test that created it removes the dependency on
/// ActorSystem-shutdown timing entirely.
/// </para>
/// </remarks>
private void StopReconciliationActor(IActorRef actor)
{
Watch(actor);
Sys.Stop(actor);
// A stop is processed after the in-flight tick completes, so this also guarantees no
// pull or insert is still running against the fixture DB.
ExpectTerminated(actor, TimeSpan.FromSeconds(30));
Unwatch(actor);
}
/// <summary>
/// Subscribes to the EventStream and collects every
/// <see cref="SiteAuditTelemetryStalledChanged"/> publication into a list
@@ -278,7 +309,7 @@ public class SiteAuditReconciliationActorTests : TestKit, IClassFixture<MsSqlMig
var repo = new RecordingRepo();
var opts = FastTickOptions();
CreateActor(sites, client, repo, opts);
var actor = CreateActor(sites, client, repo, opts);
// The first scheduled tick fires after `ReconciliationIntervalSeconds`,
// which is 0 for the test — Akka's scheduler still respects the
@@ -289,6 +320,8 @@ public class SiteAuditReconciliationActorTests : TestKit, IClassFixture<MsSqlMig
() => Assert.True(client.Calls.Count >= 1, $"expected >= 1 pull call, got {client.Calls.Count}"),
duration: TimeSpan.FromSeconds(3),
interval: TimeSpan.FromMilliseconds(50));
StopReconciliationActor(actor);
}
// ---------------------------------------------------------------------
@@ -304,7 +337,7 @@ public class SiteAuditReconciliationActorTests : TestKit, IClassFixture<MsSqlMig
var client = new ScriptedPullClient();
var repo = new RecordingRepo();
CreateActor(sites, client, repo, FastTickOptions());
var actor = CreateActor(sites, client, repo, FastTickOptions());
AwaitAssert(() =>
{
@@ -313,6 +346,8 @@ public class SiteAuditReconciliationActorTests : TestKit, IClassFixture<MsSqlMig
},
duration: TimeSpan.FromSeconds(3),
interval: TimeSpan.FromMilliseconds(50));
StopReconciliationActor(actor);
}
// ---------------------------------------------------------------------
@@ -329,11 +364,14 @@ public class SiteAuditReconciliationActorTests : TestKit, IClassFixture<MsSqlMig
new PullAuditEventsResponse(new[] { e1, e2 }, MoreAvailable: false));
var repo = new RecordingRepo();
CreateActor(sites, client, repo, FastTickOptions());
var actor = CreateActor(sites, client, repo, FastTickOptions());
AwaitAssert(() => Assert.Equal(2, repo.InsertCallCount),
duration: TimeSpan.FromSeconds(3),
interval: TimeSpan.FromMilliseconds(50));
StopReconciliationActor(actor);
Assert.Contains(repo.Inserted, e => e.EventId == e1.EventId);
Assert.Contains(repo.Inserted, e => e.EventId == e2.EventId);
}
@@ -371,13 +409,18 @@ public class SiteAuditReconciliationActorTests : TestKit, IClassFixture<MsSqlMig
await using var context = CreateContext();
var repo = new AuditLogRepository(context);
CreateActor(sites, client, repo, FastTickOptions());
var actor = CreateActor(sites, client, repo, FastTickOptions());
// Wait for the actor to ingest both rows.
await Task.Delay(TimeSpan.FromSeconds(1));
AwaitAssert(() => Assert.True(client.Calls.Count >= 1),
duration: TimeSpan.FromSeconds(3));
// Stop before reading: this actor holds a real repository over the SHARED fixture
// database (and over the DbContext this test is about to dispose), so it must not
// outlive the test (register row 95).
StopReconciliationActor(actor);
// Even though the pull returned 2 events, only 1 fresh row should
// exist in MSSQL alongside the pre-existing one — InsertIfNotExistsAsync
// is first-write-wins on EventId.
@@ -413,7 +456,7 @@ public class SiteAuditReconciliationActorTests : TestKit, IClassFixture<MsSqlMig
new PullAuditEventsResponse(new[] { e1, e2, e3 }, MoreAvailable: false));
var repo = new RecordingRepo();
CreateActor(sites, client, repo, FastTickOptions());
var actor = CreateActor(sites, client, repo, FastTickOptions());
// Wait until we have at least two pulls — the second one must use t3
// as its `since` argument because that was the max OccurredAtUtc in
@@ -423,6 +466,10 @@ public class SiteAuditReconciliationActorTests : TestKit, IClassFixture<MsSqlMig
duration: TimeSpan.FromSeconds(5),
interval: TimeSpan.FromMilliseconds(50));
// Stop before indexing into client.Calls: the list must not grow mid-assertion
// (register row 95).
StopReconciliationActor(actor);
Assert.Equal(DateTime.MinValue, client.Calls[0].SinceUtc);
Assert.Equal(t3, client.Calls[1].SinceUtc);
@@ -451,13 +498,15 @@ public class SiteAuditReconciliationActorTests : TestKit, IClassFixture<MsSqlMig
new PullAuditEventsResponse(new[] { low, high, mid }, MoreAvailable: true));
var repo = new RecordingRepo();
CreateActor(sites, client, repo, FastTickOptions());
var actor = CreateActor(sites, client, repo, FastTickOptions());
AwaitAssert(() => Assert.True(client.Calls.Count >= 2,
$"need at least 2 pulls, got {client.Calls.Count}"),
duration: TimeSpan.FromSeconds(5),
interval: TimeSpan.FromMilliseconds(50));
StopReconciliationActor(actor);
Assert.Equal(t, client.Calls[1].SinceUtc);
Assert.Equal(high.EventId.ToString(), client.Calls[1].AfterId);
}
@@ -475,13 +524,15 @@ public class SiteAuditReconciliationActorTests : TestKit, IClassFixture<MsSqlMig
new PullAuditEventsResponse(new[] { evt }, MoreAvailable: false));
var repo = new AlwaysThrowingRepo();
CreateActor(sites, client, repo, FastTickOptions());
var actor = CreateActor(sites, client, repo, FastTickOptions());
AwaitAssert(() => Assert.True(client.Calls.Count >= 2,
$"need at least 2 pulls, got {client.Calls.Count}"),
duration: TimeSpan.FromSeconds(5),
interval: TimeSpan.FromMilliseconds(50));
StopReconciliationActor(actor);
Assert.Equal(DateTime.MinValue, client.Calls[1].SinceUtc);
Assert.Null(client.Calls[1].AfterId);
}
@@ -543,7 +594,7 @@ public class SiteAuditReconciliationActorTests : TestKit, IClassFixture<MsSqlMig
new PullAuditEventsResponse(new[] { bEvent }, MoreAvailable: false));
var repo = new RecordingRepo();
CreateActor(sites, client, repo, FastTickOptions());
var actor = CreateActor(sites, client, repo, FastTickOptions());
AwaitAssert(() =>
{
@@ -552,6 +603,8 @@ public class SiteAuditReconciliationActorTests : TestKit, IClassFixture<MsSqlMig
},
duration: TimeSpan.FromSeconds(3),
interval: TimeSpan.FromMilliseconds(50));
StopReconciliationActor(actor);
}
// ---------------------------------------------------------------------
@@ -574,12 +627,13 @@ public class SiteAuditReconciliationActorTests : TestKit, IClassFixture<MsSqlMig
var repo = new RecordingRepo();
var (probe, _) = SubscribeStalled();
CreateActor(sites, client, repo, FastTickOptions(stalledAfter: 2));
var actor = CreateActor(sites, client, repo, FastTickOptions(stalledAfter: 2));
// Expect Stalled=true after the second non-draining tick. The probe
// waits with its own timeout (a few seconds gives the 0 s repeat
// interval ample slack).
var msg = probe.ExpectMsg<SiteAuditTelemetryStalledChanged>(TimeSpan.FromSeconds(5));
StopReconciliationActor(actor);
Assert.Equal("siteA", msg.SiteId);
Assert.True(msg.Stalled);
}
@@ -606,7 +660,7 @@ public class SiteAuditReconciliationActorTests : TestKit, IClassFixture<MsSqlMig
var repo = new RecordingRepo();
var (probe, _) = SubscribeStalled();
CreateActor(sites, client, repo, FastTickOptions(stalledAfter: 2));
var actor = CreateActor(sites, client, repo, FastTickOptions(stalledAfter: 2));
// First publication is the stalled=true transition; second is the
// back-to-draining flip. The actor publishes ONLY on transitions so we
@@ -615,6 +669,7 @@ public class SiteAuditReconciliationActorTests : TestKit, IClassFixture<MsSqlMig
Assert.True(first.Stalled);
var second = probe.ExpectMsg<SiteAuditTelemetryStalledChanged>(TimeSpan.FromSeconds(5));
StopReconciliationActor(actor);
Assert.False(second.Stalled);
Assert.Equal("siteA", second.SiteId);
}
@@ -643,7 +698,7 @@ public class SiteAuditReconciliationActorTests : TestKit, IClassFixture<MsSqlMig
var repo = new PoisonInsertRepo(
poison.EventId, new InvalidOperationException("poison row will never insert"));
CreateActor(sites, client, repo, FastTickOptions());
var actor = CreateActor(sites, client, repo, FastTickOptions());
// After the 5th failed attempt the actor abandons the row and writes ONE
// synthetic ReconciliationAbandoned row (fresh EventId) whose details carry
@@ -669,5 +724,7 @@ public class SiteAuditReconciliationActorTests : TestKit, IClassFixture<MsSqlMig
() => Assert.Contains(client.Calls, c => c.SinceUtc == poisonTime),
duration: TimeSpan.FromSeconds(5),
interval: TimeSpan.FromMilliseconds(100));
StopReconciliationActor(actor);
}
}
tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Integration/OutageReconciliationTests.cs
+45 -7
View File
@@ -165,6 +165,37 @@ public class OutageReconciliationTests : TestKit, IClassFixture<MsSqlMigrationFi
return (sp, ingest);
}
/// <summary>
/// Stops an actor and blocks until it is actually dead.
/// </summary>
/// <remarks>
/// <para>
/// Deferred-work register row 95, the same leak shape row 82 fixed in
/// <c>PartitionPurgeTests</c> / <c>NotifyDispatcherAuditTrailTests</c>. Both tests here
/// already called <c>Sys.Stop</c>, but that is fire-and-forget: it returns before the
/// actor has processed the stop, so an in-flight 100 ms tick could still be pulling and
/// INSERTING into the shared <see cref="MsSqlMigrationFixture"/> database — and into the
/// <see cref="IServiceProvider"/> the enclosing <c>await using</c> is about to dispose.
/// Worse, the call sat at the very END of the block, so any earlier assertion failure
/// skipped it entirely and left the actor ticking for the rest of the assembly run,
/// stopped only by <see cref="TestKit"/>'s <c>Dispose</c> → <c>Shutdown</c>, which is
/// timeout-bounded and does not fail when it expires.
/// </para>
/// <para>
/// Watch/Stop/ExpectTerminated makes the stop synchronous, and each call site now sits
/// immediately after the last step that needs the actor alive rather than at the end.
/// </para>
/// </remarks>
private void StopActor(IActorRef actor)
{
Watch(actor);
Sys.Stop(actor);
// A stop is processed after the in-flight tick completes, so this also guarantees no
// pull or insert is still running against the fixture DB.
ExpectTerminated(actor, TimeSpan.FromSeconds(30));
Unwatch(actor);
}
private static SiteAuditReconciliationOptions FastTickOptions(int batchSize = 256) => new()
{
ReconciliationIntervalSeconds = 300,
@@ -265,6 +296,12 @@ public class OutageReconciliationTests : TestKit, IClassFixture<MsSqlMigrationFi
duration: TimeSpan.FromSeconds(10),
interval: TimeSpan.FromMilliseconds(100));
// Both actors have done everything the test needs; stop them HERE, before the
// read-back, so nothing is still writing to the shared fixture DB while step 5
// counts rows and so a step-5 failure cannot leak them (register row 95).
StopActor(reconciliationActor);
StopActor(ingest);
// Step 5: assert no duplicates by EventId — central must have
// exactly the 200 rows we wrote at the site (one row per EventId).
await using var verify = CreateContext();
@@ -277,10 +314,6 @@ public class OutageReconciliationTests : TestKit, IClassFixture<MsSqlMigrationFi
// And every EventId we wrote at the site is present centrally.
Assert.True(written.All(w => centralIds.Contains(w.EventId)),
"every site-written EventId should be present centrally.");
// Tear the actor down before disposing the harness; the actor's
// PostStop cancels its scheduled timer.
Sys.Stop(reconciliationActor);
}
}
@@ -303,7 +336,7 @@ public class OutageReconciliationTests : TestKit, IClassFixture<MsSqlMigrationFi
await sqliteWriter.WriteAsync(NewEvent(siteId, baseOccurred.AddMilliseconds(i)));
}
var (sp, _) = BuildCentralPipeline();
var (sp, ingest) = BuildCentralPipeline();
await using (sp as IAsyncDisposable ?? throw new InvalidOperationException())
{
var pullClient = new QueueBackedPullClient(sqliteWriter);
@@ -338,6 +371,13 @@ public class OutageReconciliationTests : TestKit, IClassFixture<MsSqlMigrationFi
Assert.True(pullClient.CallCount > callsAfterDrain,
$"expected additional pull calls after drain to validate idempotency, got {pullClient.CallCount} after {callsAfterDrain}");
// Both actors have done everything the test needs; stop them HERE, before the
// read-back, so nothing is still writing to the shared fixture DB while the
// duplicate check counts rows and so a failure below cannot leak them
// (register row 95).
StopActor(reconciliationActor);
StopActor(ingest);
// Central count must still be exactly totalEvents — no duplicates
// even though the cursor + read-Reconciled-too semantics could
// theoretically re-fetch on the second cycle.
@@ -347,8 +387,6 @@ public class OutageReconciliationTests : TestKit, IClassFixture<MsSqlMigrationFi
.ToListAsync();
Assert.Equal(totalEvents, rows.Count);
Assert.Equal(totalEvents, rows.Select(r => r.EventId).Distinct().Count());
Sys.Stop(reconciliationActor);
}
}
}