perf(historian): recorder appends off its mailbox; Periodic commit default (lmxopcua#746)

The continuous-historization recorder awaited every outbox append (an fsync under PerEntry) on its mailbox, so ingestion was capped at 1/fsync_latency bursts per second while the mux kept telling into an unbounded, unmetered mailbox. GetStatus also suspended the mailbox on an unbounded outbox count. - Gated ingest bursts go to a bounded channel (64 bursts) drained by one ordered append loop, which commits every burst waiting behind a slow append in a single AppendBatchAsync (FIFO kept). Only a full queue holds the mailbox (backpressure, bounded memory). Counters are Interlocked; the loop nudges the drain. PostStop flushes queued bursts within the shared 5 s stop budget. - GetStatus reads counters on the actor thread and pipes a count bounded by a 10 s timeout (-1 when it does not answer); it no longer suspends the mailbox. RecorderStatus gains PendingAppends, MailboxDepth and AppendBackpressureWaits; two gauges publish pending appends and mailbox depth. - ContinuousHistorization:CommitMode defaults to Periodic (100 ms). The Host's appsettings.json, its unparseable-value fallback and docker-dev still say PerEntry: filed as a Host follow-up (outside this lane). Rebased from the unpushed remed/lane-H2 branch (originally authored 2026-10-22 in session_01Ub9hGmYLQvn9XETzw7CMsT) onto current master. Claude-Session: https://claude.ai/code/session_01GT67AEHd2Acv1YPwGfXEoV
2026-10-22 17:30:58 -04:00
parent f4907d098c
commit a5557375ee
7 changed files with 407 additions and 87 deletions
@@ -49,7 +49,8 @@ public sealed class ContinuousHistorizationOptionsTests
    {
        var options = new ContinuousHistorizationOptions();

-        Assert.Equal("PerEntry", options.CommitMode);
+        Assert.Equal("Periodic", options.CommitMode);   // 100 ms group commits — PerEntry capped ingest at 1 / fsync latency
+        Assert.Equal(100, options.CommitIntervalMs);
        Assert.Equal(1024, options.DrainBatchSize);
        Assert.Equal(8, options.DrainConcurrency);
        Assert.Equal(500_000, options.Capacity);
@@ -0,0 +1,144 @@
+using Akka.Actor;
+using Akka.TestKit.Xunit2;
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+using ZB.MOM.WW.OtOpcUa.Runtime.Historian;
+using ZB.MOM.WW.OtOpcUa.Runtime.VirtualTags;
+
+namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Historian;
+
+/// <summary>
+///     The recorder's durable append runs OFF the mailbox: an ingest burst is queued to a single ordered append loop (one
+///     fsync covers every burst waiting behind it) with a bounded number in flight, so a slow fsync no longer serialises
+///     ingestion at <c>1 / fsync_latency</c> bursts/s while the mux keeps telling into an unbounded mailbox; only a full
+///     queue suspends the mailbox (backpressure, bounded memory). <c>GetStatus</c> no longer suspends the mailbox on the
+///     outbox count, and is bounded by a timeout. Pending appends and the mailbox depth are reported.
+/// </summary>
+public sealed class ContinuousHistorizationRecorderAppendPipelineTests : TestKit
+{
+    private static readonly DateTime T0 = new(2026, 10, 22, 8, 0, 0, DateTimeKind.Utc);
+    private static readonly TimeSpan Wait = TimeSpan.FromSeconds(10);
+
+    private static VirtualTagActor.DependencyValueBatch Burst(int n, int offset = 0) =>
+        new(Enumerable.Range(offset, n).Select(i => new VirtualTagActor.DependencyValueChanged("A", (double)i, T0.AddSeconds(i))).ToList());
+
+    private IActorRef Start(LoggingOutbox outbox, IHistorianValueWriter writer, int maxAppendsInFlight = 64, TimeSpan? statusTimeout = null) =>
+        Sys.ActorOf(ContinuousHistorizationRecorder.Props(
+            CreateTestProbe().Ref, writer, outbox, historizedRefs: ["A"],
+            drainInterval: TimeSpan.FromSeconds(30), maxAppendsInFlight: maxAppendsInFlight, statusTimeout: statusTimeout));
+
+    private Task<ContinuousHistorizationRecorder.RecorderStatus> StatusAsync(IActorRef rec, TimeSpan? timeout = null) =>
+        rec.Ask<ContinuousHistorizationRecorder.RecorderStatus>(ContinuousHistorizationRecorder.GetStatus.Instance, timeout ?? TimeSpan.FromSeconds(3));
+
+    [Fact]
+    public async Task A_slow_append_does_not_suspend_the_mailbox()
+    {
+        var log = new EventLog();
+        var outbox = new LoggingOutbox(log) { AppendGate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously) };
+        var rec = Start(outbox, new LoggingPassWriter(log));
+
+        rec.Tell(Burst(3));
+        await outbox.AppendEntered.Task.WaitAsync(Wait);
+
+        // The fsync is "stuck": the actor still answers — the append is not on the mailbox.
+        var during = await StatusAsync(rec);
+        during.PendingAppends.ShouldBe(1);
+        during.TotalRecorded.ShouldBe(0);
+
+        outbox.AppendGate.SetResult();
+        await AwaitAssertAsync(async () => (await StatusAsync(rec)).TotalRecorded.ShouldBe(3), Wait);
+        (await StatusAsync(rec)).PendingAppends.ShouldBe(0);
+    }
+
+    [Fact]
+    public async Task Bursts_queued_behind_a_slow_append_are_recorded_in_order_with_fewer_fsyncs()
+    {
+        var log = new EventLog();
+        var outbox = new LoggingOutbox(log) { AppendGate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously) };
+        var rec = Start(outbox, new LoggingPassWriter(log));
+
+        rec.Tell(Burst(2, 0));
+        await outbox.AppendEntered.Task.WaitAsync(Wait);
+        for (var b = 1; b < 6; b++) rec.Tell(Burst(2, b * 2));
+        await AwaitAssertAsync(async () => (await StatusAsync(rec)).PendingAppends.ShouldBe(6), Wait);
+
+        outbox.AppendGate.SetResult();
+        await AwaitAssertAsync(async () => (await StatusAsync(rec)).TotalRecorded.ShouldBe(12), Wait);
+
+        outbox.Appended.Select(e => e.Value.Number).ShouldBe(Enumerable.Range(0, 12).Select(i => (double)i), "FIFO across bursts");
+        outbox.AppendBatchCalls.ShouldBeLessThan(6, "the bursts waiting behind the slow append share its successor's commit");
+    }
+
+    [Fact]
+    public async Task A_full_append_queue_backpressures_the_mailbox_and_loses_nothing()
+    {
+        var log = new EventLog();
+        var outbox = new LoggingOutbox(log) { AppendGate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously) };
+        var rec = Start(outbox, new LoggingPassWriter(log), maxAppendsInFlight: 2);
+
+        rec.Tell(Burst(1, 0));
+        await outbox.AppendEntered.Task.WaitAsync(Wait);
+        for (var b = 1; b < 5; b++) rec.Tell(Burst(1, b));
+
+        // Two in flight (one in the stuck append, one queued), the third burst's write waits on the queue — and with it
+        // the mailbox: a status request is not answered until the queue drains.
+        var status = StatusAsync(rec, TimeSpan.FromSeconds(10));
+        await Task.Delay(300);
+        status.IsCompleted.ShouldBeFalse("a full append queue must hold the mailbox (bounded memory)");
+
+        outbox.AppendGate.SetResult();
+        (await status).ShouldNotBeNull();
+        await AwaitAssertAsync(async () => (await StatusAsync(rec)).TotalRecorded.ShouldBe(5), Wait);
+        outbox.Appended.Select(e => e.Value.Number).ShouldBe([0d, 1d, 2d, 3d, 4d]);
+    }
+
+    [Fact]
+    public async Task GetStatus_does_not_suspend_the_mailbox_on_a_hung_outbox_count_and_is_bounded()
+    {
+        var log = new EventLog();
+        var outbox = new LoggingOutbox(log) { CountGate = new TaskCompletionSource() };
+        var rec = Start(outbox, new LoggingPassWriter(log), statusTimeout: TimeSpan.FromMilliseconds(300));
+
+        var first = StatusAsync(rec);
+        // The mailbox keeps moving while the count hangs: a burst sent now is appended.
+        rec.Tell(Burst(2));
+        await AwaitAssertAsync(() => outbox.Appended.Count.ShouldBe(2), Wait);
+
+        var answered = await first;
+        answered.QueuedDepth.ShouldBe(-1, "an outbox count that did not answer within the timeout is unknown");
+
+        // Positive control: a count that answers is reported.
+        outbox.CountGate.SetResult();
+        await AwaitAssertAsync(async () => (await StatusAsync(rec)).QueuedDepth.ShouldBeGreaterThanOrEqualTo(0), Wait);
+    }
+
+    [Fact]
+    public async Task The_mailbox_depth_is_reported()
+    {
+        var log = new EventLog();
+        var outbox = new LoggingOutbox(log);
+        var rec = Start(outbox, new LoggingPassWriter(log));
+        var status = await StatusAsync(rec);
+        status.MailboxDepth.ShouldBeGreaterThanOrEqualTo(0);
+    }
+
+    [Fact]
+    public async Task Stopping_flushes_bursts_still_queued_for_append()
+    {
+        var log = new EventLog();
+        var outbox = new LoggingOutbox(log) { AppendGate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously) };
+        var rec = Start(outbox, new LoggingPassWriter(log));
+        rec.Tell(Burst(1, 0));
+        await outbox.AppendEntered.Task.WaitAsync(Wait);
+        rec.Tell(Burst(1, 1));
+        await AwaitAssertAsync(async () => (await StatusAsync(rec)).PendingAppends.ShouldBe(2), Wait);
+
+        _ = Task.Run(async () => { await Task.Delay(200); outbox.AppendGate.SetResult(); });
+        Watch(rec);
+        Sys.Stop(rec);
+        ExpectTerminated(rec, Wait);
+
+        outbox.Appended.Select(e => e.Value.Number).ShouldBe([0d, 1d], "the queued burst is appended within the stop budget, not dropped");
+    }
+}
@@ -23,6 +23,7 @@ internal sealed class LoggingOutbox(EventLog log) : IHistorizationOutbox
 {
    private readonly Lock _gate = new();
    private readonly List<HistorizationOutboxEntry> _entries = new();
+    private readonly List<HistorizationOutboxEntry> _appended = new();
    private int _appendBatchCalls;

    /// <summary>When set, every AppendBatchAsync waits on it (on a pool thread) before committing — a slow fsync.</summary>
@@ -40,6 +41,9 @@ internal sealed class LoggingOutbox(EventLog log) : IHistorizationOutbox

    public IReadOnlyList<HistorizationOutboxEntry> Entries { get { lock (_gate) return _entries.ToArray(); } }

+    /// <summary>Every entry ever appended, in append order (acks do not remove from it).</summary>
+    public IReadOnlyList<HistorizationOutboxEntry> Appended { get { lock (_gate) return _appended.ToArray(); } }
+
    public ValueTask AppendAsync(HistorizationOutboxEntry entry, CancellationToken ct) => AppendBatchAsync([entry], ct);

    public async ValueTask AppendBatchAsync(IReadOnlyList<HistorizationOutboxEntry> entries, CancellationToken ct)
@@ -48,7 +52,11 @@ internal sealed class LoggingOutbox(EventLog log) : IHistorizationOutbox
        AppendEntered.TrySetResult();
        await Task.Delay(1, ct).ConfigureAwait(false); // complete on a pool thread, as a real fsync does
        if (AppendGate is { } gate) await gate.Task.WaitAsync(ct).ConfigureAwait(false);
-        lock (_gate) _entries.AddRange(entries);
+        lock (_gate)
+        {
+            _entries.AddRange(entries);
+            _appended.AddRange(entries);
+        }
        log.Add($"append:{entries.Count}");
    }