diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/ContinuousHistorizationRecorderTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/ContinuousHistorizationRecorderTests.cs
index 1ca1c182..cc3ac255 100644
--- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/ContinuousHistorizationRecorderTests.cs
+++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/ContinuousHistorizationRecorderTests.cs
@@ -1,3 +1,4 @@
+using System.Diagnostics;
using Akka.Actor;
using Akka.TestKit.Xunit2;
using Xunit;
@@ -445,11 +446,15 @@ public sealed class ContinuousHistorizationRecorderTests : TestKit
}, TimeSpan.FromSeconds(10));
// And the healthy tag was submitted a BOUNDED number of times — once per blocked pass plus the
- // pass that unblocked, never indefinitely. Snapshot it, let several more drain intervals elapse,
- // and prove it has stopped growing: an unbounded re-write would still be climbing.
+ // pass that unblocked, never indefinitely. Snapshot it, wait on the DRAIN'S OWN clock until three
+ // more passes have demonstrably run (a wall-clock sleep only hopes they did — on a loaded agent it
+ // can elapse with the drain never scheduled, and the negative assertion below then passes
+ // vacuously), and prove the count has stopped growing: an unbounded re-write would still climb.
int rewrites = writer.Snapshot().Count(w => w.Tag == "Good.Tag");
Assert.InRange(rewrites, 1, maxAttempts + 1);
- await Task.Delay(TimeSpan.FromMilliseconds(300));
+
+ int passes = await WaitForFurtherDrainPassesAsync(outbox, outbox.PeekBatchCallCount, 3, TimeSpan.FromSeconds(5));
+ Assert.True(passes >= 3, $"expected >= 3 further drain passes within 5s, observed {passes}");
Assert.Equal(rewrites, writer.Snapshot().Count(w => w.Tag == "Good.Tag"));
}
@@ -598,8 +603,23 @@ public sealed class ContinuousHistorizationRecorderTests : TestKit
await writer.Cancelled.WaitAsync(TimeSpan.FromSeconds(10));
- // Give the (unfixed) fold every chance to book the failure and truncate before we assert it did not.
- await Task.Delay(TimeSpan.FromMilliseconds(500));
+ // Bounded negative watch — red is immediate when the fold truncates; green costs the window once.
+ // This guards a NEGATIVE ("the pass unwound without acking"), so instead of sleeping and hoping the
+ // unwind finished, poll for the BUG's own positive signal: the fold calling RemoveAsync at all, or
+ // the queued count moving off 2. Either fires the moment the unfixed recorder books the failure and
+ // truncates, so a regression fails in milliseconds rather than after a fixed sleep; only the passing
+ // case pays the full window.
+ var watch = Stopwatch.StartNew();
+ while (watch.Elapsed < TimeSpan.FromSeconds(2))
+ {
+ Assert.True(
+ outbox.RemoveCallCount == 0,
+ $"the cancelled pass acked: RemoveAsync ran {outbox.RemoveCallCount}x, truncating the queue");
+ int queued = await outbox.CountAsync(default);
+ Assert.True(queued == 2, $"the cancelled pass changed the queue depth: expected 2, saw {queued}");
+ await Task.Delay(25);
+ }
+
Assert.Equal(2, await outbox.CountAsync(default));
}
@@ -853,8 +873,28 @@ public sealed class ContinuousHistorizationRecorderTests : TestKit
},
TimeSpan.FromSeconds(10));
- // …and stays alone: with a window of 8 the other two groups would have entered by now.
- await Task.Delay(TimeSpan.FromMilliseconds(300));
+ // …and stays alone: with a window of 8 the other two groups would have entered by now. Bounded
+ // negative watch rather than a sleep — red is immediate the moment a second group enters; green
+ // costs the window once.
+ //
+ // The drain-pass clock the other negative watches in this file use is NOT available here: the
+ // recorder is single-flight (`_draining`), and a blocked pass re-arms no drain timer, so
+ // PeekBatchCallCount is pinned at 1 for as long as the writer is gated. What makes the short
+ // window sufficient is that admission is not deferred work — the drain starts every group's task
+ // eagerly and a free semaphore permit is taken synchronously, so under a window of 8 groups B and
+ // C would be inside the writer in the same synchronous stretch as A, before the precondition above
+ // could even observe CallCount >= 1. The genuinely clock-anchored pin is the post-Release
+ // assertion at the end, which reads the peak after all three groups have demonstrably run.
+ var watch = Stopwatch.StartNew();
+ while (watch.Elapsed < TimeSpan.FromMilliseconds(500))
+ {
+ Assert.True(
+ writer.MaxObservedConcurrency == 1 && writer.CallCount == 1,
+ $"a second group entered under a window of 1: calls={writer.CallCount}, "
+ + $"peak={writer.MaxObservedConcurrency}");
+ await Task.Delay(25);
+ }
+
Assert.Equal(1, writer.MaxObservedConcurrency);
Assert.Equal(1, writer.CallCount);
@@ -1231,6 +1271,35 @@ public sealed class ContinuousHistorizationRecorderTests : TestKit
mux.ExpectNoMsg(TimeSpan.FromMilliseconds(300)); // no PreStart re-register == no restart
}
+ ///
+ /// Waits until the recorder has run further drain passes past
+ /// , and returns how many it actually observed.
+ ///
+ ///
+ /// The drain's own clock, for the negative assertions that need "several more passes have happened"
+ /// — the recorder is single-flight, so advances once
+ /// per COMPLETED pass. A wall-clock sleep only hopes the passes ran: on a loaded agent it can elapse
+ /// with the drain never scheduled, and the negative assertion it guards then passes vacuously.
+ ///
+ /// On cap-out the last observation is RETURNED rather than thrown on, so the caller's assert
+ /// reports the actual shortfall instead of a bare timeout — the same shape as
+ /// FasterLogHistorizationOutboxTests.ReadCommittedWhenVisibleAsync.
+ ///
+ ///
+ private static async Task WaitForFurtherDrainPassesAsync(
+ InMemoryOutbox outbox, int baseline, int additionalPasses, TimeSpan cap)
+ {
+ var elapsed = Stopwatch.StartNew();
+ int observed = outbox.PeekBatchCallCount - baseline;
+ while (observed < additionalPasses && elapsed.Elapsed < cap)
+ {
+ await Task.Delay(25);
+ observed = outbox.PeekBatchCallCount - baseline;
+ }
+
+ return observed;
+ }
+
///
/// double that BLOCKS every call until ,
/// recording the peak number of calls in flight simultaneously. That peak is what distinguishes a
@@ -1503,6 +1572,7 @@ public sealed class ContinuousHistorizationRecorderTests : TestKit
private readonly bool _yieldOnAppend;
private int _peekCalls;
+ private int _removeCalls;
/// When non-null, durably appends the first
/// entries and then throws this — the production outbox's
@@ -1531,6 +1601,21 @@ public sealed class ContinuousHistorizationRecorderTests : TestKit
}
}
+ /// Lifetime count of calls — one per acked drain pass, so a
+ /// non-zero value is positive evidence that the outcome fold decided to TRUNCATE. A count is the
+ /// only way to observe an ack that removed nothing (an id already gone), which a bare
+ /// reading cannot distinguish from "the fold never ran".
+ public int RemoveCallCount
+ {
+ get
+ {
+ lock (_gate)
+ {
+ return _removeCalls;
+ }
+ }
+ }
+
/// Drop-oldest capacity; 0 means unbounded.
///
/// When true, genuinely yields before recording the entry —
@@ -1701,6 +1786,7 @@ public sealed class ContinuousHistorizationRecorderTests : TestKit
{
lock (_gate)
{
+ _removeCalls++;
int idx = _entries.FindIndex(e => e.Id == id);
if (idx >= 0)
{