test(communication)+test(sitecallaudit): cap parallelism, delta/collection-serialize gauge asserts, replace sleep-then-assert, stop leaked 100ms tickers (arch-review 2026-08-19 P8-5 + P8-6, register rows 40/82/95/96/102 patterns)

This commit is contained in:
Joseph Doherty
2026-08-19 15:03:03 -04:00
parent 9a3f006721
commit d9d72f1bc5
7 changed files with 422 additions and 227 deletions
@@ -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 /// dedup by AlarmKey, placeholder seeded-but-not-on-stream coherence, NodeA↔NodeB reconnect
/// re-seed, and periodic reconcile authoritative-replace (drift correction). /// re-seed, and periodic reconcile authoritative-replace (drift correction).
/// </summary> /// </summary>
/// <remarks>
/// Serialized with <c>SiteAlarmLiveCacheTelemetryTests</c> (register row 102): every actor
/// this class spins up moves the process-global
/// <c>scadabridge.site.alarm_cache.aggregators.active</c> gauge via
/// <see cref="SiteAlarmAggregatorActor"/>'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.
/// </remarks>
[Collection("SiteAlarmAggregatorGauge")]
public class SiteAlarmAggregatorActorTests : TestKit public class SiteAlarmAggregatorActorTests : TestKit
{ {
private const string SiteId = "site-alpha"; private const string SiteId = "site-alpha";
@@ -188,6 +197,28 @@ public class SiteAlarmAggregatorActorTests : TestKit
private static AlarmStateChanged? Find(IReadOnlyList<AlarmStateChanged>? rows, string alarmName, string sourceRef) => private static AlarmStateChanged? Find(IReadOnlyList<AlarmStateChanged>? rows, string alarmName, string sourceRef) =>
rows?.FirstOrDefault(r => r.AlarmName == alarmName && r.SourceReference == sourceRef); rows?.FirstOrDefault(r => r.AlarmName == alarmName && r.SourceReference == sourceRef);
/// <summary>
/// Row-40 house pattern for an ABSENCE assertion: proves <paramref name="checks"/> keep
/// holding throughout <paramref name="window"/> 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.
/// </summary>
private static void AssertHoldsFor(TimeSpan window, params (Func<bool> 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 ─────────────────────────────────────────────────────────────────── // ── Tests ───────────────────────────────────────────────────────────────────
[Fact] [Fact]
@@ -226,8 +257,11 @@ public class SiteAlarmAggregatorActorTests : TestKit
seed.CompleteNext(Alarm("PumpFault", "", 500, tSeed)); seed.CompleteNext(Alarm("PumpFault", "", 500, tSeed));
AwaitCondition(() => sink.Count >= 1, TimeSpan.FromSeconds(3)); AwaitCondition(() => sink.Count >= 1, TimeSpan.FromSeconds(3));
// Give a beat to ensure no late (dropped) delta sneaks in. // Prove no late (dropped) delta sneaks in over the window, not just at one sample.
Thread.Sleep(150); 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 latest = sink.Latest!;
var pumpFault = Find(latest, "PumpFault", ""); var pumpFault = Find(latest, "PumpFault", "");
Assert.NotNull(pumpFault); Assert.NotNull(pumpFault);
@@ -382,8 +416,8 @@ public class SiteAlarmAggregatorActorTests : TestKit
seed.FaultNext(); seed.FaultNext();
// A failed INITIAL seed must not publish (page keeps its poll fallback — not live yet). // A failed INITIAL seed must not publish (page keeps its poll fallback — not live yet).
Thread.Sleep(200); AssertHoldsFor(TimeSpan.FromMilliseconds(200),
Assert.Equal(0, sink.Count); (() => 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 // The periodic reconcile is the backstop: it re-runs the fan-out (flag was cleared
// on failure), completes, and the actor publishes → now live. // 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 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. // Pre-fix this burns retry budget and opens a THIRD stream; post-fix it is ignored.
Thread.Sleep(400); AssertHoldsFor(TimeSpan.FromMilliseconds(400),
Assert.Equal(2, TotalSubs()); (() => TotalSubs() == 2, "a stale gen-1 error must not open a third stream"));
} }
// ── WP1.1: graceful (status OK) stream completion is a reconnect trigger ── // ── 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 // … and without refunding it either: the next error is the 4th, so it exceeds
// MaxRetries and the stream is given up rather than reconnected. // MaxRetries and the stream is given up rather than reconnected.
factory.ClientFor(GrpcNodeB).Subs[2].OnError(new Exception("4")); factory.ClientFor(GrpcNodeB).Subs[2].OnError(new Exception("4"));
Thread.Sleep(400); AssertHoldsFor(TimeSpan.FromMilliseconds(400),
Assert.Equal(before, TotalSubs()); (() => TotalSubs() == before, "budget exhausted — the 4th error must not reconnect"));
} }
[Fact] [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 // 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). // gen-2 stream (which would go deltaless until the next reconcile tick).
firstSub.OnCompleted(); firstSub.OnCompleted();
Thread.Sleep(300); AssertHoldsFor(TimeSpan.FromMilliseconds(300),
Assert.DoesNotContain("corr-1", factory.ClientFor(GrpcNodeB).Unsubscribed); (() => !factory.ClientFor(GrpcNodeB).Unsubscribed.Contains("corr-1"),
Assert.Single(factory.ClientFor(GrpcNodeB).Subs); "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) ── // ── R2 T10: live-delta publish coalescing (N6) ──
@@ -621,15 +657,15 @@ public class SiteAlarmAggregatorActorTests : TestKit
factory.ClientFor(GrpcNodeA).Subs[1].OnError(new Exception("3")); factory.ClientFor(GrpcNodeA).Subs[1].OnError(new Exception("3"));
AwaitCondition(() => factory.ClientFor(GrpcNodeB).Subs.Count == 2, TimeSpan.FromSeconds(5)); AwaitCondition(() => factory.ClientFor(GrpcNodeB).Subs.Count == 2, TimeSpan.FromSeconds(5));
Thread.Sleep(300); AssertHoldsFor(TimeSpan.FromMilliseconds(300),
Assert.Equal(1, seed.CallCount); // pre-fix: 4 (() => 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. // The stream finally comes up - exactly ONE re-seed, on the connect.
factory.ClientFor(GrpcNodeB).Subs[1].OnConnected!(); factory.ClientFor(GrpcNodeB).Subs[1].OnConnected!();
AwaitCondition(() => seed.CallCount == 2, TimeSpan.FromSeconds(5)); AwaitCondition(() => seed.CallCount == 2, TimeSpan.FromSeconds(5));
seed.CompleteNext(); seed.CompleteNext();
Thread.Sleep(300); AssertHoldsFor(TimeSpan.FromMilliseconds(300),
Assert.Equal(2, seed.CallCount); // and no second one (() => seed.CallCount == 2, "the successful connect must re-seed exactly once"));
} }
[Fact] [Fact]
@@ -644,8 +680,8 @@ public class SiteAlarmAggregatorActorTests : TestKit
Thread.Sleep(200); Thread.Sleep(200);
actor.Tell(new RunReconcile()); // skipped - the initial seed covered this window actor.Tell(new RunReconcile()); // skipped - the initial seed covered this window
Thread.Sleep(300); AssertHoldsFor(TimeSpan.FromMilliseconds(300),
Assert.Equal(1, seed.CallCount); (() => seed.CallCount == 1, "a tick inside the just-seeded window must be skipped"));
actor.Tell(new RunReconcile()); // and the next one runs actor.Tell(new RunReconcile()); // and the next one runs
AwaitCondition(() => seed.CallCount == 2, TimeSpan.FromSeconds(3)); AwaitCondition(() => seed.CallCount == 2, TimeSpan.FromSeconds(3));
@@ -668,8 +704,8 @@ public class SiteAlarmAggregatorActorTests : TestKit
AwaitCondition(() => seed.CallCount == 2, TimeSpan.FromSeconds(3)); AwaitCondition(() => seed.CallCount == 2, TimeSpan.FromSeconds(3));
seed.CompleteNext(Alarm("PumpFault", "", 500, t)); // identical snapshot seed.CompleteNext(Alarm("PumpFault", "", 500, t)); // identical snapshot
Thread.Sleep(300); AssertHoldsFor(TimeSpan.FromMilliseconds(300),
Assert.Equal(1, sink.Count); // no second publish (() => sink.Count == 1, "an unchanged reconcile snapshot must not republish"));
// A genuinely changed snapshot still publishes. // A genuinely changed snapshot still publishes.
actor.Tell(new RunReconcile()); actor.Tell(new RunReconcile());
@@ -733,8 +769,8 @@ public class SiteAlarmAggregatorActorTests : TestKit
Thread.Sleep(150); Thread.Sleep(150);
actor.Tell(new RunReconcile()); // consumed by the initial seed's flag actor.Tell(new RunReconcile()); // consumed by the initial seed's flag
Thread.Sleep(200); AssertHoldsFor(TimeSpan.FromMilliseconds(200),
Assert.Equal(1, seed.CallCount); (() => 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. // From here on, with no reconnect in between, EVERY tick must fan out.
for (var expected = 2; expected <= 5; expected++) 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… // That re-seed covers this window: the next tick is skipped…
actor.Tell(new RunReconcile()); actor.Tell(new RunReconcile());
Thread.Sleep(250); AssertHoldsFor(TimeSpan.FromMilliseconds(250),
Assert.Equal(3, seed.CallCount); (() => seed.CallCount == 3, "the connect-driven re-seed must stand the next tick down"));
// …and the one after it fans out again. // …and the one after it fans out again.
actor.Tell(new RunReconcile()); actor.Tell(new RunReconcile());
@@ -87,12 +87,25 @@ public class PendingDeploymentPurgeActorTests : TestKit
// Fast interval so the self-scheduling timer fires repeatedly; the actor must // Fast interval so the self-scheduling timer fires repeatedly; the actor must
// swallow each per-tick exception and keep ticking (best-effort hygiene — a // swallow each per-tick exception and keep ticking (best-effort hygiene — a
// crash here would drop the singleton until failover). // crash here would drop the singleton until failover).
CreateActor(repo, TimeSpan.FromMilliseconds(100)); var actor = CreateActor(repo, TimeSpan.FromMilliseconds(100));
AwaitAssert( try
() => Assert.True(Volatile.Read(ref calls) >= 2, {
$"expected >= 2 ticks despite failures, got {Volatile.Read(ref calls)}"), AwaitAssert(
duration: TimeSpan.FromSeconds(5), () => Assert.True(Volatile.Read(ref calls) >= 2,
interval: TimeSpan.FromMilliseconds(50)); $"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));
}
} }
} }
@@ -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 /// relative to a baseline so the process-wide static instruments are robust to any
/// parallel test interleaving. /// parallel test interleaving.
/// </summary> /// </summary>
/// <remarks>
/// 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 <c>baseline + N</c>
/// unreachable forever. <see cref="SiteAlarmAggregatorActor"/> calls
/// <see cref="ScadaBridgeTelemetry.LiveAlarmAggregatorStarted"/>/
/// <see cref="ScadaBridgeTelemetry.LiveAlarmAggregatorStopped"/> from its own
/// PreStart/PostStop, so every actor <c>Grpc.SiteAlarmAggregatorActorTests</c> spins up
/// moves this exact gauge — serialized into the same collection (row-96 pattern) so the
/// two classes never interleave.
/// </remarks>
[Collection("SiteAlarmAggregatorGauge")]
public class SiteAlarmLiveCacheTelemetryTests public class SiteAlarmLiveCacheTelemetryTests
{ {
private static long ReadGauge(string instrumentName) private static long ReadGauge(string instrumentName)
@@ -22,6 +22,10 @@
<Using Include="Xunit" /> <Using Include="Xunit" />
</ItemGroup> </ItemGroup>
<ItemGroup>
<Content Include="xunit.runner.json" CopyToOutputDirectory="PreserveNewest" />
</ItemGroup>
<ItemGroup> <ItemGroup>
<ProjectReference Include="../../src/ZB.MOM.WW.ScadaBridge.Communication/ZB.MOM.WW.ScadaBridge.Communication.csproj" /> <ProjectReference Include="../../src/ZB.MOM.WW.ScadaBridge.Communication/ZB.MOM.WW.ScadaBridge.Communication.csproj" />
<ProjectReference Include="../../src/ZB.MOM.WW.ScadaBridge.Commons/ZB.MOM.WW.ScadaBridge.Commons.csproj" /> <ProjectReference Include="../../src/ZB.MOM.WW.ScadaBridge.Commons/ZB.MOM.WW.ScadaBridge.Commons.csproj" />
@@ -0,0 +1,4 @@
{
"$schema": "https://xunit.net/schema/current/xunit.runner.schema.json",
"maxParallelThreads": 4
}
@@ -114,6 +114,20 @@ public class SiteCallAuditPurgeTests : TestKit
NullLogger<SiteCallAuditActor>.Instance))); NullLogger<SiteCallAuditActor>.Instance)));
} }
/// <summary>
/// Row-82/row-95 house pattern: every test in this file arms a
/// <c>PurgeIntervalOverride = 100ms</c> 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.
/// </summary>
private void StopTicker(IActorRef actor)
{
Watch(actor);
Sys.Stop(actor);
ExpectTerminated(actor, TimeSpan.FromSeconds(3));
}
// --------------------------------------------------------------------- // ---------------------------------------------------------------------
// 1. PurgeTick_CallsPurgeTerminal_WithRetentionThreshold // 1. PurgeTick_CallsPurgeTerminal_WithRetentionThreshold
// --------------------------------------------------------------------- // ---------------------------------------------------------------------
@@ -124,21 +138,27 @@ public class SiteCallAuditPurgeTests : TestKit
var repo = new RecordingRepo { RowsDeletedPerCall = 7 }; var repo = new RecordingRepo { RowsDeletedPerCall = 7 };
// Non-default retention (30 days) so the assertion isn't accidentally // Non-default retention (30 days) so the assertion isn't accidentally
// satisfied by the 365-day default. // 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( // The threshold the actor passed must be ~UtcNow - 30 days. 1-minute
() => Assert.True(repo.PurgeThresholds.Count >= 1, // slack covers scheduling jitter between the tick firing and the assert.
$"expected >= 1 PurgeTerminalAsync call, got {repo.PurgeThresholds.Count}"), var threshold = repo.PurgeThresholds[0];
duration: TimeSpan.FromSeconds(3), var expected = DateTime.UtcNow - TimeSpan.FromDays(30);
interval: TimeSpan.FromMilliseconds(50)); Assert.True(
Math.Abs((threshold - expected).TotalMinutes) < 1.0,
// The threshold the actor passed must be ~UtcNow - 30 days. 1-minute $"purge threshold {threshold:o} should be within 1 minute of {expected:o}");
// slack covers scheduling jitter between the tick firing and the assert. }
var threshold = repo.PurgeThresholds[0]; finally
var expected = DateTime.UtcNow - TimeSpan.FromDays(30); {
Assert.True( StopTicker(actor);
Math.Abs((threshold - expected).TotalMinutes) < 1.0, }
$"purge threshold {threshold:o} should be within 1 minute of {expected:o}");
} }
// --------------------------------------------------------------------- // ---------------------------------------------------------------------
@@ -149,18 +169,24 @@ public class SiteCallAuditPurgeTests : TestKit
public void PurgeTick_DefaultRetention_Uses365DayThreshold() public void PurgeTick_DefaultRetention_Uses365DayThreshold()
{ {
var repo = new RecordingRepo(); 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( var threshold = repo.PurgeThresholds[0];
() => Assert.True(repo.PurgeThresholds.Count >= 1), var expected = DateTime.UtcNow - TimeSpan.FromDays(365);
duration: TimeSpan.FromSeconds(3), Assert.True(
interval: TimeSpan.FromMilliseconds(50)); 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); finally
Assert.True( {
Math.Abs((threshold - expected).TotalMinutes) < 1.0, StopTicker(actor);
$"purge threshold {threshold:o} should be within 1 minute of {expected:o}"); }
} }
// --------------------------------------------------------------------- // ---------------------------------------------------------------------
@@ -171,16 +197,22 @@ public class SiteCallAuditPurgeTests : TestKit
public void PurgeTick_PurgeThrows_ActorStaysAlive_RetriesNextTick() public void PurgeTick_PurgeThrows_ActorStaysAlive_RetriesNextTick()
{ {
var repo = new PurgeThrowingRepo(); var repo = new PurgeThrowingRepo();
CreateActor(repo, FastPurgeOptions()); 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 singleton must NOT die on a purge fault — a second tick must still
// the first throw and the timer kept ticking. // arrive (continue-on-error). Two purge calls prove the actor survived
AwaitAssert( // the first throw and the timer kept ticking.
() => Assert.True(repo.PurgeCallCount >= 2, AwaitAssert(
$"expected >= 2 purge attempts (actor survived the throw), got {repo.PurgeCallCount}"), () => Assert.True(repo.PurgeCallCount >= 2,
duration: TimeSpan.FromSeconds(3), $"expected >= 2 purge attempts (actor survived the throw), got {repo.PurgeCallCount}"),
interval: TimeSpan.FromMilliseconds(50)); duration: TimeSpan.FromSeconds(3),
interval: TimeSpan.FromMilliseconds(50));
}
finally
{
StopTicker(actor);
}
} }
// --------------------------------------------------------------------- // ---------------------------------------------------------------------
@@ -206,24 +238,30 @@ public class SiteCallAuditPurgeTests : TestKit
.BuildServiceProvider(); .BuildServiceProvider();
var options = FastPurgeOptions(retentionDays: 30); var options = FastPurgeOptions(retentionDays: 30);
Sys.ActorOf(Props.Create(() => new SiteCallAuditActor( var actor = Sys.ActorOf(Props.Create(() => new SiteCallAuditActor(
provider, provider,
options, options,
NullLogger<SiteCallAuditActor>.Instance))); NullLogger<SiteCallAuditActor>.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 var threshold = repo.PurgeThresholds[0];
// must still fire on the production path. var expected = DateTime.UtcNow - TimeSpan.FromDays(30);
AwaitAssert( Assert.True(
() => Assert.True(repo.PurgeThresholds.Count >= 1, Math.Abs((threshold - expected).TotalMinutes) < 1.0,
"purge timer must arm even when the reconciliation collaborators are absent " $"purge threshold {threshold:o} should be within 1 minute of {expected:o}");
+ $"(SiteCallAudit-007), got {repo.PurgeThresholds.Count} purge calls"), }
duration: TimeSpan.FromSeconds(3), finally
interval: TimeSpan.FromMilliseconds(50)); {
StopTicker(actor);
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}");
} }
} }
@@ -251,6 +251,20 @@ public class SiteCallAuditReconciliationTests : TestKit
NullLogger<SiteCallAuditActor>.Instance))); NullLogger<SiteCallAuditActor>.Instance)));
} }
/// <summary>
/// Row-82/row-95 house pattern: every actor built above arms a fast-repeating
/// reconciliation (and, via <see cref="SiteCallAuditActor"/>, 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.
/// </summary>
private void StopTicker(IActorRef actor)
{
Watch(actor);
Sys.Stop(actor);
ExpectTerminated(actor, TimeSpan.FromSeconds(3));
}
// --------------------------------------------------------------------- // ---------------------------------------------------------------------
// 1. AbsentRow_PulledFromSite_IsUpserted // 1. AbsentRow_PulledFromSite_IsUpserted
// --------------------------------------------------------------------- // ---------------------------------------------------------------------
@@ -267,18 +281,24 @@ public class SiteCallAuditReconciliationTests : TestKit
new PullSiteCallsResponse(new[] { row }, MoreAvailable: false)); new PullSiteCallsResponse(new[] { row }, MoreAvailable: false));
var repo = new RecordingRepo(); var repo = new RecordingRepo();
CreateActor(sites, client, repo, FastTickOptions()); var actor = CreateActor(sites, client, repo, FastTickOptions());
try
AwaitAssert( {
() => AwaitAssert(
{ () =>
Assert.True(repo.Upserted.ContainsKey(id), {
"reconciliation tick should upsert the row present at the site but absent centrally"); Assert.True(repo.Upserted.ContainsKey(id),
Assert.Equal("Parked", repo.Upserted[id].Status); "reconciliation tick should upsert the row present at the site but absent centrally");
Assert.Equal(siteId, repo.Upserted[id].SourceSite); Assert.Equal("Parked", repo.Upserted[id].Status);
}, Assert.Equal(siteId, repo.Upserted[id].SourceSite);
duration: TimeSpan.FromSeconds(3), },
interval: TimeSpan.FromMilliseconds(50)); 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)); new PullSiteCallsResponse(new[] { r1, r2, r3 }, MoreAvailable: false));
var repo = new RecordingRepo(); 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.Equal(DateTime.MinValue, client.Calls[0].SinceUtc);
() => Assert.True(client.Calls.Count >= 2, Assert.Equal(t3, client.Calls[1].SinceUtc);
$"need at least 2 pulls to assert cursor advancement, got {client.Calls.Count}"), // The batch size flows through from options.
duration: TimeSpan.FromSeconds(5), Assert.Equal(500, client.Calls[0].BatchSize);
interval: TimeSpan.FromMilliseconds(50)); }
finally
Assert.Equal(DateTime.MinValue, client.Calls[0].SinceUtc); {
Assert.Equal(t3, client.Calls[1].SinceUtc); StopTicker(actor);
// The batch size flows through from options. }
Assert.Equal(500, client.Calls[0].BatchSize);
} }
// --------------------------------------------------------------------- // ---------------------------------------------------------------------
@@ -337,19 +363,25 @@ public class SiteCallAuditReconciliationTests : TestKit
.Script(siteB, new PullSiteCallsResponse(new[] { bRow }, MoreAvailable: false)); .Script(siteB, new PullSiteCallsResponse(new[] { bRow }, MoreAvailable: false));
var repo = new RecordingRepo(); var repo = new RecordingRepo();
CreateActor(sites, client, repo, FastTickOptions()); var actor = CreateActor(sites, client, repo, FastTickOptions());
try
AwaitAssert( {
() => AwaitAssert(
{ () =>
// siteA was attempted (and threw) yet siteB's row still landed — {
// one offline site must not sink the rest of the tick. // siteA was attempted (and threw) yet siteB's row still landed —
Assert.Contains(client.Calls, c => c.SiteId == "siteA"); // one offline site must not sink the rest of the tick.
Assert.True(repo.Upserted.ContainsKey(bId), Assert.Contains(client.Calls, c => c.SiteId == "siteA");
"siteB must be reconciled even though siteA threw"); Assert.True(repo.Upserted.ContainsKey(bId),
}, "siteB must be reconciled even though siteA threw");
duration: TimeSpan.FromSeconds(3), },
interval: TimeSpan.FromMilliseconds(50)); 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. // this actor upserted nothing on its own.
var repo = new RecordingRepo(); var repo = new RecordingRepo();
var bareProvider = TestSupport.StubRepositoryProvider.Build(repo); var bareProvider = TestSupport.StubRepositoryProvider.Build(repo);
Sys.ActorOf(Props.Create(() => new SiteCallAuditActor( var bareActor = Sys.ActorOf(Props.Create(() => new SiteCallAuditActor(
bareProvider, bareProvider,
FastTickOptions(), FastTickOptions(),
NullLogger<SiteCallAuditActor>.Instance))); NullLogger<SiteCallAuditActor>.Instance)));
@@ -379,20 +411,28 @@ public class SiteCallAuditReconciliationTests : TestKit
// collaborator-less actor would have ticked too, had it been wired. // collaborator-less actor would have ticked too, had it been wired.
var liveClient = new ScriptedPullClient(); var liveClient = new ScriptedPullClient();
var liveRepo = new RecordingRepo(); var liveRepo = new RecordingRepo();
CreateActor( var liveActor = CreateActor(
new StaticEnumerator(new SiteEntry("siteX", "http://siteX:8083")), new StaticEnumerator(new SiteEntry("siteX", "http://siteX:8083")),
liveClient, liveClient,
liveRepo, liveRepo,
FastTickOptions()); FastTickOptions());
AwaitAssert( try
() => Assert.True(liveClient.Calls.Count >= 1), {
duration: TimeSpan.FromSeconds(3), AwaitAssert(
interval: TimeSpan.FromMilliseconds(50)); () => 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 // The collaborator-less actor never reconciles: it has no client to pull
// with, so it upserts nothing on its own. // with, so it upserts nothing on its own.
Assert.Equal(0, repo.UpsertCallCount); Assert.Equal(0, repo.UpsertCallCount);
}
finally
{
StopTicker(liveActor);
StopTicker(bareActor);
}
} }
// --------------------------------------------------------------------- // ---------------------------------------------------------------------
@@ -431,26 +471,32 @@ public class SiteCallAuditReconciliationTests : TestKit
ReconciliationBatchSize = 2, 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( // 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}");
// All three rows reconciled — including the page-2 row that only a Assert.Equal(DateTime.MinValue, client.Calls[0].SinceUtc);
// within-tick continuation pull could have fetched. Assert.Equal(t2, client.Calls[1].SinceUtc);
Assert.True(repo.Upserted.ContainsKey(p1a.TrackedOperationId)); }
Assert.True(repo.Upserted.ContainsKey(p1b.TrackedOperationId)); finally
Assert.True(repo.Upserted.ContainsKey(p2.TrackedOperationId), {
"the page-2 row must be reconciled within the same tick via the MoreAvailable continuation drain"); StopTicker(actor);
}, }
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);
} }
// --------------------------------------------------------------------- // ---------------------------------------------------------------------
@@ -484,35 +530,41 @@ public class SiteCallAuditReconciliationTests : TestKit
ReconciliationBatchSize = 2, 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( // The rows it saw were still upserted (idempotent mirror refresh).
() => Assert.True(client.CallCount >= 1, "the first reconciliation tick should have pulled"), // Awaited, not asserted flat: the upsert follows the pull asynchronously,
duration: TimeSpan.FromSeconds(3), // so gating on CallCount alone races the repo write under parallel load
interval: TimeSpan.FromMilliseconds(50)); // (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). // Critical SiteCallAudit-009 invariant: the within-tick drain BROKE on the
// Awaited, not asserted flat: the upsert follows the pull asynchronously, // no-progress pin rather than looping to the 50-page ceiling. With a 2s
// so gating on CallCount alone races the repo write under parallel load // tick interval, only the first tick has fired in the window, so the pull
// (observed once in a full-suite run, 2026-08-17). // count reflects ONE tick's within-loop behaviour. A correct break yields
AwaitAssert( // 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,
Assert.True(repo.Upserted.ContainsKey(r1.TrackedOperationId)); $"a single-timestamp saturation pin must break the within-tick drain, not spin to the "
Assert.True(repo.Upserted.ContainsKey(r2.TrackedOperationId)); + $"page ceiling; got {client.CallCount} pulls (an unbounded within-tick loop would be 50+)");
}, }
duration: TimeSpan.FromSeconds(3), finally
interval: TimeSpan.FromMilliseconds(50)); {
StopTicker(actor);
// 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+)");
} }
// --------------------------------------------------------------------- // ---------------------------------------------------------------------
@@ -546,24 +598,30 @@ public class SiteCallAuditReconciliationTests : TestKit
ReconciliationBatchSize = 2, ReconciliationBatchSize = 2,
}; };
CreateActor(sites, client, repo, options); var actor = CreateActor(sites, client, repo, options);
try
AwaitAssert( {
() => AwaitAssert(
{ () =>
foreach (var r in rows)
{ {
Assert.True(repo.Upserted.ContainsKey(r.TrackedOperationId), foreach (var r in rows)
"every row in the single-timestamp burst must reconcile via the " {
+ "composite keyset cursor within one tick"); 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)); },
duration: TimeSpan.FromSeconds(3),
interval: TimeSpan.FromMilliseconds(50));
// Two pages (2 + 2) drained the burst; well under the 50-page ceiling. // Two pages (2 + 2) drained the burst; well under the 50-page ceiling.
Assert.True(client.CallCount is >= 2 and < 10, Assert.True(client.CallCount is >= 2 and < 10,
$"expected the burst to drain in a couple of keyset pages, got {client.CallCount}"); $"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, ReconciliationBatchSize = 2,
}; };
CreateActor(sites, client, repo, options); var actor = CreateActor(sites, client, repo, options);
try
var evt = ExpectMsg<SiteCallReconciliationPinnedChanged>(TimeSpan.FromSeconds(5)); {
Assert.Equal(siteId, evt.SiteId); var evt = ExpectMsg<SiteCallReconciliationPinnedChanged>(TimeSpan.FromSeconds(5));
Assert.True(evt.Pinned, "a legacy site that ignores after_id must publish Pinned=true"); 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()); var actor = CreateActor(sites, client, repo, FastTickOptions());
// Wait until the drain is genuinely in flight and blocked inside PullAsync. try
await client.Entered.WaitAsync(TimeSpan.FromSeconds(5)); {
// 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 // 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 // fails at the pre-fix behaviour (the reply only arrives once the pull
// unblocks, which never happens until Release below). // unblocks, which never happens until Release below).
var id = TrackedOperationId.New(); var id = TrackedOperationId.New();
var reply = await actor.Ask<UpsertSiteCallReply>( var reply = await actor.Ask<UpsertSiteCallReply>(
new UpsertSiteCallCommand(NewRow(id, sourceSite: siteId)), new UpsertSiteCallCommand(NewRow(id, sourceSite: siteId)),
TimeSpan.FromSeconds(3)); TimeSpan.FromSeconds(3));
Assert.True(reply.Accepted); Assert.True(reply.Accepted);
Assert.Equal(id, reply.TrackedOperationId); Assert.Equal(id, reply.TrackedOperationId);
}
// Let the drain finish so the actor shuts down cleanly. finally
client.Release(); {
// 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] [Fact]
@@ -678,16 +750,25 @@ public class SiteCallAuditReconciliationTests : TestKit
var client = new CountingBlockingPullClient(); var client = new CountingBlockingPullClient();
var repo = new RecordingRepo(); 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. // Several tick intervals elapse while the first pass is still blocked.
await Task.Delay(TimeSpan.FromMilliseconds(600)); await Task.Delay(TimeSpan.FromMilliseconds(600));
Assert.Equal(1, client.CallCount); Assert.Equal(1, client.CallCount);
}
client.Release(); 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);
}
} }
/// <summary> /// <summary>
@@ -715,25 +796,32 @@ public class SiteCallAuditReconciliationTests : TestKit
var actor = CreateActor(sites, client, repo, FastTickOptions()); var actor = CreateActor(sites, client, repo, FastTickOptions());
// Once the pull has been served the drain is upserting; flood the mailbox try
// with ingest commands so the two writers overlap in wall-clock time. {
await client.Entered.WaitAsync(TimeSpan.FromSeconds(5)); // 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) var asks = Enumerable.Range(0, 6)
.Select(_ => actor.Ask<UpsertSiteCallReply>( .Select(_ => actor.Ask<UpsertSiteCallReply>(
new UpsertSiteCallCommand(NewRow(TrackedOperationId.New(), siteId)), new UpsertSiteCallCommand(NewRow(TrackedOperationId.New(), siteId)),
TimeSpan.FromSeconds(10))) TimeSpan.FromSeconds(10)))
.ToArray(); .ToArray();
await Task.WhenAll(asks); await Task.WhenAll(asks);
Assert.All(asks, ask => Assert.True(ask.Result.Accepted)); Assert.All(asks, ask => Assert.True(ask.Result.Accepted));
// The drain's own upserts must have run in the same window. // The drain's own upserts must have run in the same window.
AwaitAssert( AwaitAssert(
() => Assert.True(repo.UpsertCount >= 12, $"expected both writers to have run; saw {repo.UpsertCount}"), () => Assert.True(repo.UpsertCount >= 12, $"expected both writers to have run; saw {repo.UpsertCount}"),
TimeSpan.FromSeconds(5)); TimeSpan.FromSeconds(5));
Assert.Equal(1, repo.MaxObservedConcurrency); Assert.Equal(1, repo.MaxObservedConcurrency);
}
finally
{
StopTicker(actor);
}
} }
/// <summary> /// <summary>