diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteAlarmAggregatorActorTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteAlarmAggregatorActorTests.cs index b427520f..0c36cadd 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteAlarmAggregatorActorTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteAlarmAggregatorActorTests.cs @@ -15,6 +15,15 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Tests.Grpc; /// dedup by AlarmKey, placeholder seeded-but-not-on-stream coherence, NodeA↔NodeB reconnect /// re-seed, and periodic reconcile authoritative-replace (drift correction). /// +/// +/// Serialized with SiteAlarmLiveCacheTelemetryTests (register row 102): every actor +/// this class spins up moves the process-global +/// scadabridge.site.alarm_cache.aggregators.active gauge via +/// 's PreStart/PostStop, which the telemetry test's +/// fixed-baseline assertions require to hold still (or only increase) between its capture +/// and its checks. Same collection ⇒ never concurrent. +/// +[Collection("SiteAlarmAggregatorGauge")] public class SiteAlarmAggregatorActorTests : TestKit { private const string SiteId = "site-alpha"; @@ -188,6 +197,28 @@ public class SiteAlarmAggregatorActorTests : TestKit private static AlarmStateChanged? Find(IReadOnlyList? rows, string alarmName, string sourceRef) => rows?.FirstOrDefault(r => r.AlarmName == alarmName && r.SourceReference == sourceRef); + /// + /// Row-40 house pattern for an ABSENCE assertion: proves keep + /// holding throughout by polling, rather than a single blind + /// Thread.Sleep(window) followed by one end-of-window sample. A bare sleep-then-assert + /// cannot distinguish "the mechanism correctly stayed quiet" from "the mechanism just + /// hadn't run yet when the one sample was taken", and it also can't catch a value that + /// flips wrong and then flips back before that single sample. This fails immediately on + /// the first tick any check goes false, and only returns once the full window has + /// elapsed with every check still true. + /// + private static void AssertHoldsFor(TimeSpan window, params (Func Holds, string Because)[] checks) + { + var deadline = DateTime.UtcNow + window; + while (true) + { + foreach (var (holds, because) in checks) + Assert.True(holds(), because); + if (DateTime.UtcNow >= deadline) return; + Thread.Sleep(20); + } + } + // ── Tests ─────────────────────────────────────────────────────────────────── [Fact] @@ -226,8 +257,11 @@ public class SiteAlarmAggregatorActorTests : TestKit seed.CompleteNext(Alarm("PumpFault", "", 500, tSeed)); AwaitCondition(() => sink.Count >= 1, TimeSpan.FromSeconds(3)); - // Give a beat to ensure no late (dropped) delta sneaks in. - Thread.Sleep(150); + // Prove no late (dropped) delta sneaks in over the window, not just at one sample. + AssertHoldsFor(TimeSpan.FromMilliseconds(150), + (() => Find(sink.Latest, "PumpFault", "") is not null, "seed row must still be present"), + (() => Find(sink.Latest, "PumpFault", "")?.Priority == 500, + "seed value kept; older buffered delta must stay dropped")); var latest = sink.Latest!; var pumpFault = Find(latest, "PumpFault", ""); Assert.NotNull(pumpFault); @@ -382,8 +416,8 @@ public class SiteAlarmAggregatorActorTests : TestKit seed.FaultNext(); // A failed INITIAL seed must not publish (page keeps its poll fallback — not live yet). - Thread.Sleep(200); - Assert.Equal(0, sink.Count); + AssertHoldsFor(TimeSpan.FromMilliseconds(200), + (() => sink.Count == 0, "a failed initial seed must not publish")); // The periodic reconcile is the backstop: it re-runs the fan-out (flag was cleared // on failure), completes, and the actor publishes → now live. @@ -464,8 +498,8 @@ public class SiteAlarmAggregatorActorTests : TestKit firstSub.OnError(new Exception("late zombie fault")); // stale gen-1 error races in // Pre-fix this burns retry budget and opens a THIRD stream; post-fix it is ignored. - Thread.Sleep(400); - Assert.Equal(2, TotalSubs()); + AssertHoldsFor(TimeSpan.FromMilliseconds(400), + (() => TotalSubs() == 2, "a stale gen-1 error must not open a third stream")); } // ── WP1.1: graceful (status OK) stream completion is a reconnect trigger ── @@ -520,8 +554,8 @@ public class SiteAlarmAggregatorActorTests : TestKit // … and without refunding it either: the next error is the 4th, so it exceeds // MaxRetries and the stream is given up rather than reconnected. factory.ClientFor(GrpcNodeB).Subs[2].OnError(new Exception("4")); - Thread.Sleep(400); - Assert.Equal(before, TotalSubs()); + AssertHoldsFor(TimeSpan.FromMilliseconds(400), + (() => TotalSubs() == before, "budget exhausted — the 4th error must not reconnect")); } [Fact] @@ -539,9 +573,11 @@ public class SiteAlarmAggregatorActorTests : TestKit // A completion racing out of the dead gen-1 stream must not tear down the live // gen-2 stream (which would go deltaless until the next reconcile tick). firstSub.OnCompleted(); - Thread.Sleep(300); - Assert.DoesNotContain("corr-1", factory.ClientFor(GrpcNodeB).Unsubscribed); - Assert.Single(factory.ClientFor(GrpcNodeB).Subs); + AssertHoldsFor(TimeSpan.FromMilliseconds(300), + (() => !factory.ClientFor(GrpcNodeB).Unsubscribed.Contains("corr-1"), + "a late gen-1 completion must not unsubscribe the live gen-2 stream"), + (() => factory.ClientFor(GrpcNodeB).Subs.Count == 1, + "the live gen-2 stream must not be torn down")); } // ── R2 T10: live-delta publish coalescing (N6) ── @@ -621,15 +657,15 @@ public class SiteAlarmAggregatorActorTests : TestKit factory.ClientFor(GrpcNodeA).Subs[1].OnError(new Exception("3")); AwaitCondition(() => factory.ClientFor(GrpcNodeB).Subs.Count == 2, TimeSpan.FromSeconds(5)); - Thread.Sleep(300); - Assert.Equal(1, seed.CallCount); // pre-fix: 4 + AssertHoldsFor(TimeSpan.FromMilliseconds(300), + (() => seed.CallCount == 1, "failed reconnect attempts must not re-seed (pre-fix: 4)")); // The stream finally comes up - exactly ONE re-seed, on the connect. factory.ClientFor(GrpcNodeB).Subs[1].OnConnected!(); AwaitCondition(() => seed.CallCount == 2, TimeSpan.FromSeconds(5)); seed.CompleteNext(); - Thread.Sleep(300); - Assert.Equal(2, seed.CallCount); // and no second one + AssertHoldsFor(TimeSpan.FromMilliseconds(300), + (() => seed.CallCount == 2, "the successful connect must re-seed exactly once")); } [Fact] @@ -644,8 +680,8 @@ public class SiteAlarmAggregatorActorTests : TestKit Thread.Sleep(200); actor.Tell(new RunReconcile()); // skipped - the initial seed covered this window - Thread.Sleep(300); - Assert.Equal(1, seed.CallCount); + AssertHoldsFor(TimeSpan.FromMilliseconds(300), + (() => seed.CallCount == 1, "a tick inside the just-seeded window must be skipped")); actor.Tell(new RunReconcile()); // and the next one runs AwaitCondition(() => seed.CallCount == 2, TimeSpan.FromSeconds(3)); @@ -668,8 +704,8 @@ public class SiteAlarmAggregatorActorTests : TestKit AwaitCondition(() => seed.CallCount == 2, TimeSpan.FromSeconds(3)); seed.CompleteNext(Alarm("PumpFault", "", 500, t)); // identical snapshot - Thread.Sleep(300); - Assert.Equal(1, sink.Count); // no second publish + AssertHoldsFor(TimeSpan.FromMilliseconds(300), + (() => sink.Count == 1, "an unchanged reconcile snapshot must not republish")); // A genuinely changed snapshot still publishes. actor.Tell(new RunReconcile()); @@ -733,8 +769,8 @@ public class SiteAlarmAggregatorActorTests : TestKit Thread.Sleep(150); actor.Tell(new RunReconcile()); // consumed by the initial seed's flag - Thread.Sleep(200); - Assert.Equal(1, seed.CallCount); + AssertHoldsFor(TimeSpan.FromMilliseconds(200), + (() => seed.CallCount == 1, "the tick right after the initial seed must be skipped")); // From here on, with no reconnect in between, EVERY tick must fan out. for (var expected = 2; expected <= 5; expected++) @@ -778,8 +814,8 @@ public class SiteAlarmAggregatorActorTests : TestKit // That re-seed covers this window: the next tick is skipped… actor.Tell(new RunReconcile()); - Thread.Sleep(250); - Assert.Equal(3, seed.CallCount); + AssertHoldsFor(TimeSpan.FromMilliseconds(250), + (() => seed.CallCount == 3, "the connect-driven re-seed must stand the next tick down")); // …and the one after it fans out again. actor.Tell(new RunReconcile()); diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/PendingDeploymentPurgeActorTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/PendingDeploymentPurgeActorTests.cs index 6f3f69aa..a9e5f893 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/PendingDeploymentPurgeActorTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/PendingDeploymentPurgeActorTests.cs @@ -87,12 +87,25 @@ public class PendingDeploymentPurgeActorTests : TestKit // Fast interval so the self-scheduling timer fires repeatedly; the actor must // swallow each per-tick exception and keep ticking (best-effort hygiene — a // crash here would drop the singleton until failover). - CreateActor(repo, TimeSpan.FromMilliseconds(100)); + var actor = CreateActor(repo, TimeSpan.FromMilliseconds(100)); - AwaitAssert( - () => Assert.True(Volatile.Read(ref calls) >= 2, - $"expected >= 2 ticks despite failures, got {Volatile.Read(ref calls)}"), - duration: TimeSpan.FromSeconds(5), - interval: TimeSpan.FromMilliseconds(50)); + try + { + AwaitAssert( + () => Assert.True(Volatile.Read(ref calls) >= 2, + $"expected >= 2 ticks despite failures, got {Volatile.Read(ref calls)}"), + duration: TimeSpan.FromSeconds(5), + interval: TimeSpan.FromMilliseconds(50)); + } + finally + { + // Row-95/row-82 pattern: a 100ms self-scheduling ticker left running strands + // later tests in the same assembly (it keeps firing — and keeps calling the + // now-disposed repo/service-provider — for the rest of the process). Stop it + // deterministically before the test ends. + Watch(actor); + Sys.Stop(actor); + ExpectTerminated(actor, TimeSpan.FromSeconds(3)); + } } } diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/SiteAlarmLiveCacheTelemetryTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/SiteAlarmLiveCacheTelemetryTests.cs index 67729234..90bca491 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/SiteAlarmLiveCacheTelemetryTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/SiteAlarmLiveCacheTelemetryTests.cs @@ -11,6 +11,18 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Tests; /// relative to a baseline so the process-wide static instruments are robust to any /// parallel test interleaving. /// +/// +/// Register row 102's shape: a fixed-baseline delta only survives contributors that hold +/// constant or increase — a parallel class's actors moving the SAME process-global gauge +/// and then STOPPING after this test captures its baseline makes baseline + N +/// unreachable forever. calls +/// / +/// from its own +/// PreStart/PostStop, so every actor Grpc.SiteAlarmAggregatorActorTests spins up +/// moves this exact gauge — serialized into the same collection (row-96 pattern) so the +/// two classes never interleave. +/// +[Collection("SiteAlarmAggregatorGauge")] public class SiteAlarmLiveCacheTelemetryTests { private static long ReadGauge(string instrumentName) diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/ZB.MOM.WW.ScadaBridge.Communication.Tests.csproj b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/ZB.MOM.WW.ScadaBridge.Communication.Tests.csproj index 1b62bdce..258a470d 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/ZB.MOM.WW.ScadaBridge.Communication.Tests.csproj +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/ZB.MOM.WW.ScadaBridge.Communication.Tests.csproj @@ -22,6 +22,10 @@ + + + + diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/xunit.runner.json b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/xunit.runner.json new file mode 100644 index 00000000..6366089b --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/xunit.runner.json @@ -0,0 +1,4 @@ +{ + "$schema": "https://xunit.net/schema/current/xunit.runner.schema.json", + "maxParallelThreads": 4 +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.SiteCallAudit.Tests/SiteCallAuditPurgeTests.cs b/tests/ZB.MOM.WW.ScadaBridge.SiteCallAudit.Tests/SiteCallAuditPurgeTests.cs index d4e8e524..3cfede59 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.SiteCallAudit.Tests/SiteCallAuditPurgeTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.SiteCallAudit.Tests/SiteCallAuditPurgeTests.cs @@ -114,6 +114,20 @@ public class SiteCallAuditPurgeTests : TestKit NullLogger.Instance))); } + /// + /// Row-82/row-95 house pattern: every test in this file arms a + /// PurgeIntervalOverride = 100ms ticker. Left running past the test's own + /// assertions it keeps firing against the recording repo — and, worse, against a + /// scope-per-tick DI container whose lifetime the test no longer controls — for the + /// rest of the assembly's run, stranding later tests. Stop it deterministically. + /// + private void StopTicker(IActorRef actor) + { + Watch(actor); + Sys.Stop(actor); + ExpectTerminated(actor, TimeSpan.FromSeconds(3)); + } + // --------------------------------------------------------------------- // 1. PurgeTick_CallsPurgeTerminal_WithRetentionThreshold // --------------------------------------------------------------------- @@ -124,21 +138,27 @@ public class SiteCallAuditPurgeTests : TestKit var repo = new RecordingRepo { RowsDeletedPerCall = 7 }; // Non-default retention (30 days) so the assertion isn't accidentally // satisfied by the 365-day default. - CreateActor(repo, FastPurgeOptions(retentionDays: 30)); + var actor = CreateActor(repo, FastPurgeOptions(retentionDays: 30)); + try + { + AwaitAssert( + () => Assert.True(repo.PurgeThresholds.Count >= 1, + $"expected >= 1 PurgeTerminalAsync call, got {repo.PurgeThresholds.Count}"), + duration: TimeSpan.FromSeconds(3), + interval: TimeSpan.FromMilliseconds(50)); - AwaitAssert( - () => Assert.True(repo.PurgeThresholds.Count >= 1, - $"expected >= 1 PurgeTerminalAsync call, got {repo.PurgeThresholds.Count}"), - duration: TimeSpan.FromSeconds(3), - interval: TimeSpan.FromMilliseconds(50)); - - // The threshold the actor passed must be ~UtcNow - 30 days. 1-minute - // slack covers scheduling jitter between the tick firing and the assert. - var threshold = repo.PurgeThresholds[0]; - var expected = DateTime.UtcNow - TimeSpan.FromDays(30); - Assert.True( - Math.Abs((threshold - expected).TotalMinutes) < 1.0, - $"purge threshold {threshold:o} should be within 1 minute of {expected:o}"); + // The threshold the actor passed must be ~UtcNow - 30 days. 1-minute + // slack covers scheduling jitter between the tick firing and the assert. + var threshold = repo.PurgeThresholds[0]; + var expected = DateTime.UtcNow - TimeSpan.FromDays(30); + Assert.True( + Math.Abs((threshold - expected).TotalMinutes) < 1.0, + $"purge threshold {threshold:o} should be within 1 minute of {expected:o}"); + } + finally + { + StopTicker(actor); + } } // --------------------------------------------------------------------- @@ -149,18 +169,24 @@ public class SiteCallAuditPurgeTests : TestKit public void PurgeTick_DefaultRetention_Uses365DayThreshold() { var repo = new RecordingRepo(); - CreateActor(repo, FastPurgeOptions()); // default 365 days + var actor = CreateActor(repo, FastPurgeOptions()); // default 365 days + try + { + AwaitAssert( + () => Assert.True(repo.PurgeThresholds.Count >= 1), + duration: TimeSpan.FromSeconds(3), + interval: TimeSpan.FromMilliseconds(50)); - AwaitAssert( - () => Assert.True(repo.PurgeThresholds.Count >= 1), - duration: TimeSpan.FromSeconds(3), - interval: TimeSpan.FromMilliseconds(50)); - - var threshold = repo.PurgeThresholds[0]; - var expected = DateTime.UtcNow - TimeSpan.FromDays(365); - Assert.True( - Math.Abs((threshold - expected).TotalMinutes) < 1.0, - $"purge threshold {threshold:o} should be within 1 minute of {expected:o}"); + var threshold = repo.PurgeThresholds[0]; + var expected = DateTime.UtcNow - TimeSpan.FromDays(365); + Assert.True( + Math.Abs((threshold - expected).TotalMinutes) < 1.0, + $"purge threshold {threshold:o} should be within 1 minute of {expected:o}"); + } + finally + { + StopTicker(actor); + } } // --------------------------------------------------------------------- @@ -171,16 +197,22 @@ public class SiteCallAuditPurgeTests : TestKit public void PurgeTick_PurgeThrows_ActorStaysAlive_RetriesNextTick() { var repo = new PurgeThrowingRepo(); - CreateActor(repo, FastPurgeOptions()); - - // The singleton must NOT die on a purge fault — a second tick must still - // arrive (continue-on-error). Two purge calls prove the actor survived - // the first throw and the timer kept ticking. - AwaitAssert( - () => Assert.True(repo.PurgeCallCount >= 2, - $"expected >= 2 purge attempts (actor survived the throw), got {repo.PurgeCallCount}"), - duration: TimeSpan.FromSeconds(3), - interval: TimeSpan.FromMilliseconds(50)); + var actor = CreateActor(repo, FastPurgeOptions()); + try + { + // The singleton must NOT die on a purge fault — a second tick must still + // arrive (continue-on-error). Two purge calls prove the actor survived + // the first throw and the timer kept ticking. + AwaitAssert( + () => Assert.True(repo.PurgeCallCount >= 2, + $"expected >= 2 purge attempts (actor survived the throw), got {repo.PurgeCallCount}"), + duration: TimeSpan.FromSeconds(3), + interval: TimeSpan.FromMilliseconds(50)); + } + finally + { + StopTicker(actor); + } } // --------------------------------------------------------------------- @@ -206,24 +238,30 @@ public class SiteCallAuditPurgeTests : TestKit .BuildServiceProvider(); var options = FastPurgeOptions(retentionDays: 30); - Sys.ActorOf(Props.Create(() => new SiteCallAuditActor( + var actor = Sys.ActorOf(Props.Create(() => new SiteCallAuditActor( provider, options, NullLogger.Instance))); + try + { + // No reconciliation collaborators were registered, yet the purge tick + // must still fire on the production path. + AwaitAssert( + () => Assert.True(repo.PurgeThresholds.Count >= 1, + "purge timer must arm even when the reconciliation collaborators are absent " + + $"(SiteCallAudit-007), got {repo.PurgeThresholds.Count} purge calls"), + duration: TimeSpan.FromSeconds(3), + interval: TimeSpan.FromMilliseconds(50)); - // No reconciliation collaborators were registered, yet the purge tick - // must still fire on the production path. - AwaitAssert( - () => Assert.True(repo.PurgeThresholds.Count >= 1, - "purge timer must arm even when the reconciliation collaborators are absent " - + $"(SiteCallAudit-007), got {repo.PurgeThresholds.Count} purge calls"), - duration: TimeSpan.FromSeconds(3), - interval: TimeSpan.FromMilliseconds(50)); - - var threshold = repo.PurgeThresholds[0]; - var expected = DateTime.UtcNow - TimeSpan.FromDays(30); - Assert.True( - Math.Abs((threshold - expected).TotalMinutes) < 1.0, - $"purge threshold {threshold:o} should be within 1 minute of {expected:o}"); + var threshold = repo.PurgeThresholds[0]; + var expected = DateTime.UtcNow - TimeSpan.FromDays(30); + Assert.True( + Math.Abs((threshold - expected).TotalMinutes) < 1.0, + $"purge threshold {threshold:o} should be within 1 minute of {expected:o}"); + } + finally + { + StopTicker(actor); + } } } diff --git a/tests/ZB.MOM.WW.ScadaBridge.SiteCallAudit.Tests/SiteCallAuditReconciliationTests.cs b/tests/ZB.MOM.WW.ScadaBridge.SiteCallAudit.Tests/SiteCallAuditReconciliationTests.cs index cd5acd5b..cc054a25 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.SiteCallAudit.Tests/SiteCallAuditReconciliationTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.SiteCallAudit.Tests/SiteCallAuditReconciliationTests.cs @@ -251,6 +251,20 @@ public class SiteCallAuditReconciliationTests : TestKit NullLogger.Instance))); } + /// + /// Row-82/row-95 house pattern: every actor built above arms a fast-repeating + /// reconciliation (and, via , purge) ticker — as fast + /// as 100ms in most of the tests below. Left running past the owning test it keeps + /// firing against test doubles the test has stopped controlling, stranding later + /// tests in the same assembly. Stop it deterministically before the test ends. + /// + private void StopTicker(IActorRef actor) + { + Watch(actor); + Sys.Stop(actor); + ExpectTerminated(actor, TimeSpan.FromSeconds(3)); + } + // --------------------------------------------------------------------- // 1. AbsentRow_PulledFromSite_IsUpserted // --------------------------------------------------------------------- @@ -267,18 +281,24 @@ public class SiteCallAuditReconciliationTests : TestKit new PullSiteCallsResponse(new[] { row }, MoreAvailable: false)); var repo = new RecordingRepo(); - CreateActor(sites, client, repo, FastTickOptions()); - - AwaitAssert( - () => - { - Assert.True(repo.Upserted.ContainsKey(id), - "reconciliation tick should upsert the row present at the site but absent centrally"); - Assert.Equal("Parked", repo.Upserted[id].Status); - Assert.Equal(siteId, repo.Upserted[id].SourceSite); - }, - duration: TimeSpan.FromSeconds(3), - interval: TimeSpan.FromMilliseconds(50)); + var actor = CreateActor(sites, client, repo, FastTickOptions()); + try + { + AwaitAssert( + () => + { + Assert.True(repo.Upserted.ContainsKey(id), + "reconciliation tick should upsert the row present at the site but absent centrally"); + Assert.Equal("Parked", repo.Upserted[id].Status); + Assert.Equal(siteId, repo.Upserted[id].SourceSite); + }, + duration: TimeSpan.FromSeconds(3), + interval: TimeSpan.FromMilliseconds(50)); + } + finally + { + StopTicker(actor); + } } // --------------------------------------------------------------------- @@ -304,18 +324,24 @@ public class SiteCallAuditReconciliationTests : TestKit new PullSiteCallsResponse(new[] { r1, r2, r3 }, MoreAvailable: false)); var repo = new RecordingRepo(); - CreateActor(sites, client, repo, FastTickOptions()); + var actor = CreateActor(sites, client, repo, FastTickOptions()); + try + { + AwaitAssert( + () => Assert.True(client.Calls.Count >= 2, + $"need at least 2 pulls to assert cursor advancement, got {client.Calls.Count}"), + duration: TimeSpan.FromSeconds(5), + interval: TimeSpan.FromMilliseconds(50)); - AwaitAssert( - () => Assert.True(client.Calls.Count >= 2, - $"need at least 2 pulls to assert cursor advancement, got {client.Calls.Count}"), - duration: TimeSpan.FromSeconds(5), - interval: TimeSpan.FromMilliseconds(50)); - - Assert.Equal(DateTime.MinValue, client.Calls[0].SinceUtc); - Assert.Equal(t3, client.Calls[1].SinceUtc); - // The batch size flows through from options. - Assert.Equal(500, client.Calls[0].BatchSize); + Assert.Equal(DateTime.MinValue, client.Calls[0].SinceUtc); + Assert.Equal(t3, client.Calls[1].SinceUtc); + // The batch size flows through from options. + Assert.Equal(500, client.Calls[0].BatchSize); + } + finally + { + StopTicker(actor); + } } // --------------------------------------------------------------------- @@ -337,19 +363,25 @@ public class SiteCallAuditReconciliationTests : TestKit .Script(siteB, new PullSiteCallsResponse(new[] { bRow }, MoreAvailable: false)); var repo = new RecordingRepo(); - CreateActor(sites, client, repo, FastTickOptions()); - - AwaitAssert( - () => - { - // siteA was attempted (and threw) yet siteB's row still landed — - // one offline site must not sink the rest of the tick. - Assert.Contains(client.Calls, c => c.SiteId == "siteA"); - Assert.True(repo.Upserted.ContainsKey(bId), - "siteB must be reconciled even though siteA threw"); - }, - duration: TimeSpan.FromSeconds(3), - interval: TimeSpan.FromMilliseconds(50)); + var actor = CreateActor(sites, client, repo, FastTickOptions()); + try + { + AwaitAssert( + () => + { + // siteA was attempted (and threw) yet siteB's row still landed — + // one offline site must not sink the rest of the tick. + Assert.Contains(client.Calls, c => c.SiteId == "siteA"); + Assert.True(repo.Upserted.ContainsKey(bId), + "siteB must be reconciled even though siteA threw"); + }, + duration: TimeSpan.FromSeconds(3), + interval: TimeSpan.FromMilliseconds(50)); + } + finally + { + StopTicker(actor); + } } // --------------------------------------------------------------------- @@ -369,7 +401,7 @@ public class SiteCallAuditReconciliationTests : TestKit // this actor upserted nothing on its own. var repo = new RecordingRepo(); var bareProvider = TestSupport.StubRepositoryProvider.Build(repo); - Sys.ActorOf(Props.Create(() => new SiteCallAuditActor( + var bareActor = Sys.ActorOf(Props.Create(() => new SiteCallAuditActor( bareProvider, FastTickOptions(), NullLogger.Instance))); @@ -379,20 +411,28 @@ public class SiteCallAuditReconciliationTests : TestKit // collaborator-less actor would have ticked too, had it been wired. var liveClient = new ScriptedPullClient(); var liveRepo = new RecordingRepo(); - CreateActor( + var liveActor = CreateActor( new StaticEnumerator(new SiteEntry("siteX", "http://siteX:8083")), liveClient, liveRepo, FastTickOptions()); - AwaitAssert( - () => Assert.True(liveClient.Calls.Count >= 1), - duration: TimeSpan.FromSeconds(3), - interval: TimeSpan.FromMilliseconds(50)); + try + { + AwaitAssert( + () => Assert.True(liveClient.Calls.Count >= 1), + duration: TimeSpan.FromSeconds(3), + interval: TimeSpan.FromMilliseconds(50)); - // The collaborator-less actor never reconciles: it has no client to pull - // with, so it upserts nothing on its own. - Assert.Equal(0, repo.UpsertCallCount); + // The collaborator-less actor never reconciles: it has no client to pull + // with, so it upserts nothing on its own. + Assert.Equal(0, repo.UpsertCallCount); + } + finally + { + StopTicker(liveActor); + StopTicker(bareActor); + } } // --------------------------------------------------------------------- @@ -431,26 +471,32 @@ public class SiteCallAuditReconciliationTests : TestKit ReconciliationBatchSize = 2, }; - CreateActor(sites, client, repo, options); + var actor = CreateActor(sites, client, repo, options); + try + { + AwaitAssert( + () => + { + // All three rows reconciled — including the page-2 row that only a + // within-tick continuation pull could have fetched. + Assert.True(repo.Upserted.ContainsKey(p1a.TrackedOperationId)); + Assert.True(repo.Upserted.ContainsKey(p1b.TrackedOperationId)); + Assert.True(repo.Upserted.ContainsKey(p2.TrackedOperationId), + "the page-2 row must be reconciled within the same tick via the MoreAvailable continuation drain"); + }, + duration: TimeSpan.FromSeconds(3), + interval: TimeSpan.FromMilliseconds(50)); - AwaitAssert( - () => - { - // All three rows reconciled — including the page-2 row that only a - // within-tick continuation pull could have fetched. - Assert.True(repo.Upserted.ContainsKey(p1a.TrackedOperationId)); - Assert.True(repo.Upserted.ContainsKey(p1b.TrackedOperationId)); - Assert.True(repo.Upserted.ContainsKey(p2.TrackedOperationId), - "the page-2 row must be reconciled within the same tick via the MoreAvailable continuation drain"); - }, - duration: TimeSpan.FromSeconds(3), - interval: TimeSpan.FromMilliseconds(50)); - - // Exactly two pulls happened (page 1 + the continuation page 2) and the - // second pull's `since` cursor advanced to the page-1 max (t2). - Assert.True(client.Calls.Count >= 2, $"expected >= 2 pulls within the tick, got {client.Calls.Count}"); - Assert.Equal(DateTime.MinValue, client.Calls[0].SinceUtc); - Assert.Equal(t2, client.Calls[1].SinceUtc); + // Exactly two pulls happened (page 1 + the continuation page 2) and the + // second pull's `since` cursor advanced to the page-1 max (t2). + Assert.True(client.Calls.Count >= 2, $"expected >= 2 pulls within the tick, got {client.Calls.Count}"); + Assert.Equal(DateTime.MinValue, client.Calls[0].SinceUtc); + Assert.Equal(t2, client.Calls[1].SinceUtc); + } + finally + { + StopTicker(actor); + } } // --------------------------------------------------------------------- @@ -484,35 +530,41 @@ public class SiteCallAuditReconciliationTests : TestKit ReconciliationBatchSize = 2, }; - CreateActor(sites, client, repo, options); + var actor = CreateActor(sites, client, repo, options); + try + { + AwaitAssert( + () => Assert.True(client.CallCount >= 1, "the first reconciliation tick should have pulled"), + duration: TimeSpan.FromSeconds(3), + interval: TimeSpan.FromMilliseconds(50)); - AwaitAssert( - () => Assert.True(client.CallCount >= 1, "the first reconciliation tick should have pulled"), - duration: TimeSpan.FromSeconds(3), - interval: TimeSpan.FromMilliseconds(50)); + // The rows it saw were still upserted (idempotent mirror refresh). + // Awaited, not asserted flat: the upsert follows the pull asynchronously, + // so gating on CallCount alone races the repo write under parallel load + // (observed once in a full-suite run, 2026-08-17). + AwaitAssert( + () => + { + Assert.True(repo.Upserted.ContainsKey(r1.TrackedOperationId)); + Assert.True(repo.Upserted.ContainsKey(r2.TrackedOperationId)); + }, + duration: TimeSpan.FromSeconds(3), + interval: TimeSpan.FromMilliseconds(50)); - // The rows it saw were still upserted (idempotent mirror refresh). - // Awaited, not asserted flat: the upsert follows the pull asynchronously, - // so gating on CallCount alone races the repo write under parallel load - // (observed once in a full-suite run, 2026-08-17). - AwaitAssert( - () => - { - Assert.True(repo.Upserted.ContainsKey(r1.TrackedOperationId)); - Assert.True(repo.Upserted.ContainsKey(r2.TrackedOperationId)); - }, - duration: TimeSpan.FromSeconds(3), - interval: TimeSpan.FromMilliseconds(50)); - - // Critical SiteCallAudit-009 invariant: the within-tick drain BROKE on the - // no-progress pin rather than looping to the 50-page ceiling. With a 2s - // tick interval, only the first tick has fired in the window, so the pull - // count reflects ONE tick's within-loop behaviour. A correct break yields - // 1 pull for that tick; we allow a small margin for a possible second tick - // edge, but it must be far below the 50-page within-tick ceiling. - Assert.True(client.CallCount < 10, - $"a single-timestamp saturation pin must break the within-tick drain, not spin to the " - + $"page ceiling; got {client.CallCount} pulls (an unbounded within-tick loop would be 50+)"); + // Critical SiteCallAudit-009 invariant: the within-tick drain BROKE on the + // no-progress pin rather than looping to the 50-page ceiling. With a 2s + // tick interval, only the first tick has fired in the window, so the pull + // count reflects ONE tick's within-loop behaviour. A correct break yields + // 1 pull for that tick; we allow a small margin for a possible second tick + // edge, but it must be far below the 50-page within-tick ceiling. + Assert.True(client.CallCount < 10, + $"a single-timestamp saturation pin must break the within-tick drain, not spin to the " + + $"page ceiling; got {client.CallCount} pulls (an unbounded within-tick loop would be 50+)"); + } + finally + { + StopTicker(actor); + } } // --------------------------------------------------------------------- @@ -546,24 +598,30 @@ public class SiteCallAuditReconciliationTests : TestKit ReconciliationBatchSize = 2, }; - CreateActor(sites, client, repo, options); - - AwaitAssert( - () => - { - foreach (var r in rows) + var actor = CreateActor(sites, client, repo, options); + try + { + AwaitAssert( + () => { - Assert.True(repo.Upserted.ContainsKey(r.TrackedOperationId), - "every row in the single-timestamp burst must reconcile via the " - + "composite keyset cursor within one tick"); - } - }, - duration: TimeSpan.FromSeconds(3), - interval: TimeSpan.FromMilliseconds(50)); + foreach (var r in rows) + { + Assert.True(repo.Upserted.ContainsKey(r.TrackedOperationId), + "every row in the single-timestamp burst must reconcile via the " + + "composite keyset cursor within one tick"); + } + }, + duration: TimeSpan.FromSeconds(3), + interval: TimeSpan.FromMilliseconds(50)); - // Two pages (2 + 2) drained the burst; well under the 50-page ceiling. - Assert.True(client.CallCount is >= 2 and < 10, - $"expected the burst to drain in a couple of keyset pages, got {client.CallCount}"); + // Two pages (2 + 2) drained the burst; well under the 50-page ceiling. + Assert.True(client.CallCount is >= 2 and < 10, + $"expected the burst to drain in a couple of keyset pages, got {client.CallCount}"); + } + finally + { + StopTicker(actor); + } } // --------------------------------------------------------------------- @@ -593,11 +651,17 @@ public class SiteCallAuditReconciliationTests : TestKit ReconciliationBatchSize = 2, }; - CreateActor(sites, client, repo, options); - - var evt = ExpectMsg(TimeSpan.FromSeconds(5)); - Assert.Equal(siteId, evt.SiteId); - Assert.True(evt.Pinned, "a legacy site that ignores after_id must publish Pinned=true"); + var actor = CreateActor(sites, client, repo, options); + try + { + var evt = ExpectMsg(TimeSpan.FromSeconds(5)); + Assert.Equal(siteId, evt.SiteId); + Assert.True(evt.Pinned, "a legacy site that ignores after_id must publish Pinned=true"); + } + finally + { + StopTicker(actor); + } } // --------------------------------------------------------------------- @@ -648,22 +712,30 @@ public class SiteCallAuditReconciliationTests : TestKit var actor = CreateActor(sites, client, repo, FastTickOptions()); - // Wait until the drain is genuinely in flight and blocked inside PullAsync. - await client.Entered.WaitAsync(TimeSpan.FromSeconds(5)); + try + { + // Wait until the drain is genuinely in flight and blocked inside PullAsync. + await client.Entered.WaitAsync(TimeSpan.FromSeconds(5)); - // The mailbox must still be serving. A generous-but-finite budget: this - // fails at the pre-fix behaviour (the reply only arrives once the pull - // unblocks, which never happens until Release below). - var id = TrackedOperationId.New(); - var reply = await actor.Ask( - new UpsertSiteCallCommand(NewRow(id, sourceSite: siteId)), - TimeSpan.FromSeconds(3)); + // The mailbox must still be serving. A generous-but-finite budget: this + // fails at the pre-fix behaviour (the reply only arrives once the pull + // unblocks, which never happens until Release below). + var id = TrackedOperationId.New(); + var reply = await actor.Ask( + new UpsertSiteCallCommand(NewRow(id, sourceSite: siteId)), + TimeSpan.FromSeconds(3)); - Assert.True(reply.Accepted); - Assert.Equal(id, reply.TrackedOperationId); - - // Let the drain finish so the actor shuts down cleanly. - client.Release(); + Assert.True(reply.Accepted); + Assert.Equal(id, reply.TrackedOperationId); + } + finally + { + // Unblock the drain FIRST so the actor is free to process the Stop rather + // than parked inside the blocked pull, then stop it deterministically + // rather than leaving its 100ms ticker running (row-95). + client.Release(); + StopTicker(actor); + } } [Fact] @@ -678,16 +750,25 @@ public class SiteCallAuditReconciliationTests : TestKit var client = new CountingBlockingPullClient(); var repo = new RecordingRepo(); - CreateActor(sites, client, repo, FastTickOptions()); + var actor = CreateActor(sites, client, repo, FastTickOptions()); - await client.Entered.WaitAsync(TimeSpan.FromSeconds(5)); + try + { + await client.Entered.WaitAsync(TimeSpan.FromSeconds(5)); - // Several tick intervals elapse while the first pass is still blocked. - await Task.Delay(TimeSpan.FromMilliseconds(600)); + // Several tick intervals elapse while the first pass is still blocked. + await Task.Delay(TimeSpan.FromMilliseconds(600)); - Assert.Equal(1, client.CallCount); - - client.Release(); + Assert.Equal(1, client.CallCount); + } + finally + { + // Unblock the drain FIRST so the actor is free to process the Stop, then + // stop it deterministically rather than leaving its 100ms ticker running + // (row-95). + client.Release(); + StopTicker(actor); + } } /// @@ -715,25 +796,32 @@ public class SiteCallAuditReconciliationTests : TestKit var actor = CreateActor(sites, client, repo, FastTickOptions()); - // Once the pull has been served the drain is upserting; flood the mailbox - // with ingest commands so the two writers overlap in wall-clock time. - await client.Entered.WaitAsync(TimeSpan.FromSeconds(5)); + try + { + // Once the pull has been served the drain is upserting; flood the mailbox + // with ingest commands so the two writers overlap in wall-clock time. + await client.Entered.WaitAsync(TimeSpan.FromSeconds(5)); - var asks = Enumerable.Range(0, 6) - .Select(_ => actor.Ask( - new UpsertSiteCallCommand(NewRow(TrackedOperationId.New(), siteId)), - TimeSpan.FromSeconds(10))) - .ToArray(); + var asks = Enumerable.Range(0, 6) + .Select(_ => actor.Ask( + new UpsertSiteCallCommand(NewRow(TrackedOperationId.New(), siteId)), + TimeSpan.FromSeconds(10))) + .ToArray(); - await Task.WhenAll(asks); - Assert.All(asks, ask => Assert.True(ask.Result.Accepted)); + await Task.WhenAll(asks); + Assert.All(asks, ask => Assert.True(ask.Result.Accepted)); - // The drain's own upserts must have run in the same window. - AwaitAssert( - () => Assert.True(repo.UpsertCount >= 12, $"expected both writers to have run; saw {repo.UpsertCount}"), - TimeSpan.FromSeconds(5)); + // The drain's own upserts must have run in the same window. + AwaitAssert( + () => Assert.True(repo.UpsertCount >= 12, $"expected both writers to have run; saw {repo.UpsertCount}"), + TimeSpan.FromSeconds(5)); - Assert.Equal(1, repo.MaxObservedConcurrency); + Assert.Equal(1, repo.MaxObservedConcurrency); + } + finally + { + StopTicker(actor); + } } ///