perf(historian): one QuestDB WAL wait per drain pass, with an exponential poll (lmxopcua#746)

The QuestDB live writer waited for WAL visibility after every tag group, and the wait polled wal_tables() every 10 ms. With DrainConcurrency = 8 that was up to eight pollers at ~100 queries/s each against a database whose WAL apply was already behind. New Runtime seam IHistorianPassWriter (WriteLiveValuesInPassAsync + FlushAsync). The recorder writes every group of a pass through it and calls FlushAsync once after all groups report, before the ack, whenever any group wrote (an un-acked pass is redelivered and its insert-if-absent check must see this pass's rows). A flush failure does not fail the pass. The QuestDB writer implements it: in-pass writes only mark a flush as owed; FlushAsync waits once. The poll now backs off 10 -> 20 -> ... -> 200 ms, never past the timeout. Direct WriteLiveValuesAsync callers still wait per call. Also updates the QuestDB cancelled-write integration test to the rethrow contract from the previous commit. 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 16:48:40 -04:00
parent 7c5cc9c7e5
commit ee314d1b9e
9 changed files with 395 additions and 13 deletions
@@ -0,0 +1,100 @@
+using Akka.Actor;
+using Akka.TestKit.Xunit2;
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Runtime.Historian;
+using ZB.MOM.WW.OtOpcUa.Runtime.VirtualTags;
+
+namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Historian;
+
+/// <summary>
+///     A pass-aware writer (<see cref="IHistorianPassWriter"/>) gets every group of a drain pass as an in-pass write and
+///     ONE flush after all of them and before the ack — QuestDB's WAL visibility wait used to run once per tag group,
+///     amplifying load on a database that was already behind.
+/// </summary>
+public sealed class ContinuousHistorizationRecorderPassFlushTests : TestKit
+{
+    private static readonly DateTime T0 = new(2026, 10, 22, 8, 0, 0, DateTimeKind.Utc);
+    private static readonly TimeSpan Wait = TimeSpan.FromSeconds(10);
+
+    private (EventLog Log, LoggingOutbox Outbox, LoggingPassWriter Writer) Start(params string[] tags)
+    {
+        var log = new EventLog();
+        var outbox = new LoggingOutbox(log);
+        var writer = new LoggingPassWriter(log);
+        var rec = Sys.ActorOf(ContinuousHistorizationRecorder.Props(
+            CreateTestProbe().Ref, writer, outbox, historizedRefs: tags,
+            drainInterval: TimeSpan.FromMilliseconds(50), minBackoff: TimeSpan.FromMilliseconds(50),
+            maxBackoff: TimeSpan.FromMilliseconds(100)));
+        rec.Tell(new VirtualTagActor.DependencyValueBatch(
+            tags.Select((t, i) => new VirtualTagActor.DependencyValueChanged(t, (double)i, T0.AddSeconds(i))).ToList()));
+        return (log, outbox, writer);
+    }
+
+    [Fact]
+    public void One_flush_follows_every_group_of_the_pass_and_precedes_the_ack()
+    {
+        var (log, _, _) = Start("A", "B", "C");
+
+        AwaitAssert(() => log.Snapshot().ShouldContain("ack"), Wait);
+        var events = log.Snapshot().SkipWhile(e => !e.StartsWith("write:")).ToList();
+        var flush = events.IndexOf("flush");
+        flush.ShouldBe(3, $"three in-pass writes, then the flush: {string.Join(", ", events)}");
+        events.Take(3).OrderBy(e => e).ShouldBe(["write:A", "write:B", "write:C"]);
+        events[flush + 1].ShouldBe("ack");
+        events.Count(e => e == "flush").ShouldBe(1);
+        log.Snapshot().ShouldNotContain(e => e.StartsWith("plain:"), "a pass writer is never written per call");
+    }
+
+    [Fact]
+    public void A_pass_with_a_failed_group_still_flushes_the_groups_that_wrote()
+    {
+        var log = new EventLog();
+        var outbox = new LoggingOutbox(log);
+        var writer = new LoggingPassWriter(log);
+        writer.Fail("B");
+        var rec = Sys.ActorOf(ContinuousHistorizationRecorder.Props(
+            CreateTestProbe().Ref, writer, outbox, historizedRefs: ["A", "B"],
+            drainInterval: TimeSpan.FromMilliseconds(50), minBackoff: TimeSpan.FromMilliseconds(50),
+            maxBackoff: TimeSpan.FromMilliseconds(100), maxWriteAttempts: 100));
+        rec.Tell(new VirtualTagActor.DependencyValueBatch(
+            [new("A", 1.0, T0), new("B", 2.0, T0)]));
+
+        // The pass cannot ack (B blocks the head), but A's rows must be visible to the next pass's redelivery check.
+        AwaitAssert(() => log.Snapshot().ShouldContain("flush"), Wait);
+        log.Snapshot().ShouldNotContain("ack");
+    }
+
+    [Fact]
+    public void A_pass_where_nothing_wrote_does_not_flush()
+    {
+        var log = new EventLog();
+        var outbox = new LoggingOutbox(log);
+        var writer = new LoggingPassWriter(log);
+        writer.Fail("A");
+        var rec = Sys.ActorOf(ContinuousHistorizationRecorder.Props(
+            CreateTestProbe().Ref, writer, outbox, historizedRefs: ["A"],
+            drainInterval: TimeSpan.FromMilliseconds(50), minBackoff: TimeSpan.FromMilliseconds(50),
+            maxBackoff: TimeSpan.FromMilliseconds(100)));
+        rec.Tell(new VirtualTagActor.DependencyValueBatch([new("A", 1.0, T0)]));
+
+        // Positive control first: the failing write itself happened (twice — a retry ran), so "no flush" is not vacuous.
+        AwaitAssert(() => log.Snapshot().Count(e => e == "write:A").ShouldBeGreaterThanOrEqualTo(2), Wait);
+        log.Snapshot().ShouldNotContain("flush");
+    }
+
+    [Fact]
+    public void A_failing_flush_does_not_fail_the_pass()
+    {
+        var log = new EventLog();
+        var outbox = new LoggingOutbox(log);
+        var writer = new LoggingPassWriter(log) { FlushThrows = new InvalidOperationException("wal_tables() failed") };
+        var rec = Sys.ActorOf(ContinuousHistorizationRecorder.Props(
+            CreateTestProbe().Ref, writer, outbox, historizedRefs: ["A"],
+            drainInterval: TimeSpan.FromMilliseconds(50)));
+        rec.Tell(new VirtualTagActor.DependencyValueBatch([new("A", 1.0, T0)]));
+
+        AwaitAssert(() => log.Snapshot().ShouldContain("ack"), Wait);
+        outbox.Entries.ShouldBeEmpty("the rows are committed; only their visibility is unproven");
+    }
+}
@@ -0,0 +1,110 @@
+using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+using ZB.MOM.WW.OtOpcUa.Runtime.Historian;
+
+namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Historian;
+
+/// <summary>
+///     Shared doubles for the recorder tests added with the pass-flush / off-mailbox-append work: a FIFO outbox and a
+///     pass-aware writer that log every call into one ordered event list, so a test can assert the ORDER of writes,
+///     flushes and acks across both. Thread-safe — the recorder appends and drains off the actor thread.
+/// </summary>
+internal sealed class EventLog
+{
+    private readonly Lock _gate = new();
+    private readonly List<string> _events = new();
+
+    public void Add(string e) { lock (_gate) _events.Add(e); }
+
+    public IReadOnlyList<string> Snapshot() { lock (_gate) return _events.ToArray(); }
+}
+
+/// <summary>FIFO outbox honouring the truncate-through contract; appends can be made to yield or block.</summary>
+internal sealed class LoggingOutbox(EventLog log) : IHistorizationOutbox
+{
+    private readonly Lock _gate = new();
+    private readonly List<HistorizationOutboxEntry> _entries = new();
+    private int _appendBatchCalls;
+
+    /// <summary>When set, every AppendBatchAsync waits on it (on a pool thread) before committing — a slow fsync.</summary>
+    public TaskCompletionSource? AppendGate { get; set; }
+
+    /// <summary>Completed by an append that has entered (before waiting on <see cref="AppendGate"/>).</summary>
+    public TaskCompletionSource AppendEntered { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously);
+
+    /// <summary>When set, CountAsync blocks on it — a hung outbox.</summary>
+    public TaskCompletionSource? CountGate { get; set; }
+
+    public int AppendBatchCalls { get { lock (_gate) return _appendBatchCalls; } }
+
+    public long DroppedCount => 0;
+
+    public IReadOnlyList<HistorizationOutboxEntry> Entries { get { lock (_gate) return _entries.ToArray(); } }
+
+    public ValueTask AppendAsync(HistorizationOutboxEntry entry, CancellationToken ct) => AppendBatchAsync([entry], ct);
+
+    public async ValueTask AppendBatchAsync(IReadOnlyList<HistorizationOutboxEntry> entries, CancellationToken ct)
+    {
+        lock (_gate) _appendBatchCalls++;
+        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);
+        log.Add($"append:{entries.Count}");
+    }
+
+    public ValueTask<IReadOnlyList<HistorizationOutboxEntry>> PeekBatchAsync(int max, CancellationToken ct)
+    {
+        lock (_gate) return ValueTask.FromResult<IReadOnlyList<HistorizationOutboxEntry>>(_entries.Take(max).ToArray());
+    }
+
+    public ValueTask RemoveAsync(Guid id, CancellationToken ct)
+    {
+        lock (_gate)
+        {
+            var i = _entries.FindIndex(e => e.Id == id);
+            if (i >= 0) _entries.RemoveRange(0, i + 1);
+        }
+        log.Add("ack");
+        return ValueTask.CompletedTask;
+    }
+
+    public async ValueTask<int> CountAsync(CancellationToken ct)
+    {
+        if (CountGate is { } gate) await gate.Task.WaitAsync(ct).ConfigureAwait(false);
+        lock (_gate) return _entries.Count;
+    }
+
+    public void Dispose() { }
+}
+
+/// <summary>A pass-aware writer: logs every in-pass write and flush; plain writes are logged distinctly.</summary>
+internal sealed class LoggingPassWriter(EventLog log) : IHistorianPassWriter
+{
+    private readonly HashSet<string> _failTags = new(StringComparer.Ordinal);
+
+    public Exception? FlushThrows { get; set; }
+
+    public void Fail(string tag) { lock (_failTags) _failTags.Add(tag); }
+
+    public Task<bool> WriteLiveValuesAsync(string tag, IReadOnlyList<HistorizationValue> values, CancellationToken ct)
+    {
+        log.Add($"plain:{tag}");
+        return Task.FromResult(true);
+    }
+
+    public async Task<bool> WriteLiveValuesInPassAsync(string tag, IReadOnlyList<HistorizationValue> values, CancellationToken ct)
+    {
+        await Task.Delay(1, ct).ConfigureAwait(false);
+        bool fail;
+        lock (_failTags) fail = _failTags.Contains(tag);
+        log.Add($"write:{tag}");
+        return !fail;
+    }
+
+    public async Task FlushAsync(CancellationToken ct)
+    {
+        await Task.Delay(1, ct).ConfigureAwait(false);
+        log.Add("flush");
+        if (FlushThrows is { } ex) throw ex;
+    }
+}