test(historian): replace the recorder tests' fixed sleeps with bounded watches on drain-pass and truncation observables

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
     }
 
+    /// <summary>
+    ///     Waits until the recorder has run <paramref name="additionalPasses"/> further drain passes past
+    ///     <paramref name="baseline"/>, and returns how many it actually observed.
+    /// </summary>
+    /// <remarks>
+    ///     The drain's own clock, for the negative assertions that need "several more passes have happened"
+    ///     — the recorder is single-flight, so <see cref="InMemoryOutbox.PeekBatchCallCount"/> 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.
+    ///     <para>
+    ///         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
+    ///         <c>FasterLogHistorizationOutboxTests.ReadCommittedWhenVisibleAsync</c>.
+    ///     </para>
+    /// </remarks>
+    private static async Task<int> 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;
+    }
+
     /// <summary>
     ///     <see cref="IHistorianValueWriter"/> double that BLOCKS every call until <see cref="Release"/>,
     ///     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;
 
         /// <summary>When non-null, <see cref="AppendBatchAsync"/> durably appends the first
         /// <see cref="AppendBatchFaultAfter"/> entries and then throws this — the production outbox's
@@ -1531,6 +1601,21 @@ public sealed class ContinuousHistorizationRecorderTests : TestKit
             }
         }
 
+        /// <summary>Lifetime count of <see cref="RemoveAsync"/> 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
+        /// <see cref="CountAsync"/> reading cannot distinguish from "the fold never ran".</summary>
+        public int RemoveCallCount
+        {
+            get
+            {
+                lock (_gate)
+                {
+                    return _removeCalls;
+                }
+            }
+        }
+
         /// <param name="capacity">Drop-oldest capacity; 0 means unbounded.</param>
         /// <param name="yieldOnAppend">
         ///     When <c>true</c>, <see cref="AppendAsync"/> 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)
                 {