From ee314d1b9ea1894f55604312a5787198a33ae2fc Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 16:48:40 -0400 Subject: [PATCH] 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 --- docs/Historian.md | 11 +- .../QuestDbHistorianValueWriter.cs | 42 ++++++- .../QuestDbWalVisibility.cs | 31 ++++- .../ContinuousHistorizationRecorder.cs | 29 ++++- .../Historian/IHistorianPassWriter.cs | 36 ++++++ .../WriteTests.cs | 41 ++++++- .../QuestDbWalVisibilityTests.cs | 8 ++ ...uousHistorizationRecorderPassFlushTests.cs | 100 ++++++++++++++++ .../Historian/RecorderTestDoubles.cs | 110 ++++++++++++++++++ 9 files changed, 395 insertions(+), 13 deletions(-) create mode 100644 src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/IHistorianPassWriter.cs create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/ContinuousHistorizationRecorderPassFlushTests.cs create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/RecorderTestDoubles.cs diff --git a/docs/Historian.md b/docs/Historian.md index 3da225773..29823488a 100644 --- a/docs/Historian.md +++ b/docs/Historian.md @@ -2008,8 +2008,15 @@ Because `DEDUP UPSERT` would otherwise let a re-sent sample overwrite the stored QuestDB applies WAL commits asynchronously, so a row just inserted may not be visible to the next query (measured: up to 7 of 20 inserts invisible to an immediate re-read). **Every write path waits** — the live writer, the HistoryUpdate updater (including between DeleteRaw pages), the tag catalog/provisioner and the alarm writer (`QuestDbWalVisibility`): after the INSERT(s) it reads `sequencerTxn` from `wal_tables()` and polls -`writerTxn` about every 10 ms until it catches up, the table is reported `suspended` (Warning, stop), or `VisibilityTimeout` elapses (Warning). -A timeout never changes a result code and never discards data; the rows simply become visible shortly after. Measured: median 0 ms, p95 6 ms. +`writerTxn` — first after 10 ms, then backing off exponentially (20, 40, 80, 160, then every 200 ms) — until it catches up, the table is +reported `suspended` (Warning, stop), or `VisibilityTimeout` elapses (Warning). A timeout never changes a result code and never discards data; +the rows simply become visible shortly after. Measured: median 0 ms, p95 6 ms. + +**The live writer waits once per drain pass, not once per tag group.** The recorder writes every group of a pass through the writer's +`IHistorianPassWriter` path and then calls `FlushAsync` once — after every group has reported, before the ack, whenever any group wrote — +so the pass's rows are visible to the next pass's insert-if-absent check. With `DrainConcurrency = 8` the per-group wait used to run up to +eight pollers at ~100 `wal_tables()` queries/s each against a database whose WAL apply was already behind. A flush failure does not fail +the pass (the rows are committed). Direct callers of `WriteLiveValuesAsync` still wait per call. ### HistoryUpdate (soft delete) diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianValueWriter.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianValueWriter.cs index 3f537daf2..2c04447fd 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianValueWriter.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianValueWriter.cs @@ -6,6 +6,7 @@ using Npgsql; using NpgsqlTypes; using ZB.MOM.WW.OtOpcUa.Core.Abstractions; using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common; +using ZB.MOM.WW.OtOpcUa.Runtime.Historian; namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb; @@ -22,10 +23,14 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb; /// + Bad, an in-batch µs collision keeps the first). RC1: a sample whose time is before 1970 cannot be a /// QuestDB designated timestamp and one such row would fail the whole multi-row INSERT, so those samples are dropped /// (Warning at most once per tag per ; every one counted in ) and the call still succeeds for the rest. -/// R5: after inserting, the writer waits for WAL visibility so an immediate redelivery sees the rows. +/// R5: after inserting, the writer waits for WAL visibility so an immediate redelivery sees the rows — per call +/// through , but ONCE per drain pass through the recorder's +/// path ( + ): with +/// DrainConcurrency groups per pass each running its own poll, a lagging WAL apply was polled by up to eight +/// waiters at once, amplifying load on a database that was already behind. /// Inserts are chunked at rows (8 bound parameters per row, far under the 32,767 limit). /// -internal sealed class QuestDbHistorianValueWriter : IHistorianValueWriter +internal sealed class QuestDbHistorianValueWriter : IHistorianPassWriter { /// Rows per multi-row INSERT (T0 Q2: 1,000 rows is reliable). internal const int BatchRows = 1000; @@ -42,6 +47,7 @@ internal sealed class QuestDbHistorianValueWriter : IHistorianValueWriter private readonly ILogger _log; private readonly Func _utcNow; private long _droppedPre1970; + private int _flushPending; private readonly ConcurrentDictionary _pre1970WarnedUtc = new(StringComparer.Ordinal); /// Creates the writer. @@ -64,7 +70,32 @@ internal sealed class QuestDbHistorianValueWriter : IHistorianValueWriter internal long DroppedPre1970 => Interlocked.Read(ref _droppedPre1970); /// - public async Task WriteLiveValuesAsync(string tag, IReadOnlyList values, CancellationToken ct) + public Task WriteLiveValuesAsync(string tag, IReadOnlyList values, CancellationToken ct) => + WriteAsync(tag, values, waitForVisibility: true, ct); + + /// + public Task WriteLiveValuesInPassAsync(string tag, IReadOnlyList values, CancellationToken ct) => + WriteAsync(tag, values, waitForVisibility: false, ct); + + /// + // One WAL visibility wait for every in-pass insert since the last flush; the wait's timeout / a suspended table does + // not change the outcome (the rows are committed and become visible later). + public async Task FlushAsync(CancellationToken ct) + { + if (Interlocked.Exchange(ref _flushPending, 0) == 0) return; + try + { + await _wal.WaitAsync(_conn.SampleTable, ct).ConfigureAwait(false); + } + catch + { + Volatile.Write(ref _flushPending, 1); // still owed: the next flush retries it + throw; + } + } + + private async Task WriteAsync(string tag, IReadOnlyList values, bool waitForVisibility, + CancellationToken ct) { try { @@ -99,7 +130,10 @@ internal sealed class QuestDbHistorianValueWriter : IHistorianValueWriter for (var i = 0; i < fresh.Count; i += BatchRows) await InsertAsync(tag, fresh.GetRange(i, Math.Min(BatchRows, fresh.Count - i)), serverTs, ct).ConfigureAwait(false); - await _wal.WaitAsync(_conn.SampleTable, ct).ConfigureAwait(false); // timeout/suspension does not change the result (R5) + if (waitForVisibility) + await _wal.WaitAsync(_conn.SampleTable, ct).ConfigureAwait(false); // timeout/suspension does not change the result + else + Volatile.Write(ref _flushPending, 1); // the pass's FlushAsync waits once for every group return true; } catch (OperationCanceledException) when (ct.IsCancellationRequested) diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbWalVisibility.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbWalVisibility.cs index 578ff3bdc..a6712a178 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbWalVisibility.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbWalVisibility.cs @@ -8,12 +8,29 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb; /// The read-your-writes wait (R5, T0 Q4). QuestDB applies WAL commits asynchronously, so a row just inserted may /// not be visible to the next query. After its INSERT(s) a writer calls , which captures /// sequencerTxn from wal_tables() (every autocommitted INSERT is one sequencer transaction; a concurrent -/// writer can only raise the target) and polls writerTxn every ~10 ms until it reaches the target, the table -/// is suspended, or elapses. +/// writer can only raise the target) and polls writerTxn — first after 10 ms, then backing off exponentially +/// (doubling, capped at ) so a lagging WAL apply is not hammered with ~100 queries/s per +/// waiter — until it reaches the target, the table is suspended, or +/// elapses. /// internal sealed class QuestDbWalVisibility { - private static readonly TimeSpan PollInterval = TimeSpan.FromMilliseconds(10); + /// The first poll delay. + internal static readonly TimeSpan FirstPollInterval = TimeSpan.FromMilliseconds(10); + + /// The longest poll delay the backoff reaches. + internal static readonly TimeSpan MaxPollInterval = TimeSpan.FromMilliseconds(200); + + /// The poll delay after : doubled, capped at . + /// The delay just waited. + /// The next delay. + internal static TimeSpan NextPollInterval(TimeSpan current) => + current >= MaxPollInterval / 2 ? MaxPollInterval : current * 2; + + private int _waits; + + /// Waits started so far (lifetime) — test/diagnostic. + internal int Waits => Volatile.Read(ref _waits); private readonly QuestDbConnection _conn; private readonly TimeSpan _timeout; @@ -49,7 +66,9 @@ internal sealed class QuestDbWalVisibility public async Task WaitAsync(string table, CancellationToken ct) { if (Disabled) return true; + Interlocked.Increment(ref _waits); var sw = Stopwatch.StartNew(); + var delay = FirstPollInterval; await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false); long target; @@ -86,7 +105,11 @@ internal sealed class QuestDbWalVisibility table, _timeout.TotalMilliseconds, writer, target); return false; } - await Task.Delay(PollInterval, ct).ConfigureAwait(false); + var remaining = _timeout - sw.Elapsed; + // Never sleep past the timeout (and never a negative delay if it elapsed in between). + await Task.Delay(remaining <= TimeSpan.Zero ? TimeSpan.Zero : delay < remaining ? delay : remaining, ct) + .ConfigureAwait(false); + delay = NextPollInterval(delay); } } } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationRecorder.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationRecorder.cs index 65ec4853f..833f3df18 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationRecorder.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationRecorder.cs @@ -141,6 +141,9 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers private readonly IActorRef _dependencyMux; private readonly IHistorianValueWriter _writer; + + /// as a pass-aware writer (one flush per pass), or null for a plain writer. + private readonly IHistorianPassWriter? _passWriter; private readonly IHistorizationOutbox _outbox; /// The tracked historized tags, keyed by mux ref ( — the @@ -279,6 +282,7 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers { _dependencyMux = dependencyMux ?? throw new ArgumentNullException(nameof(dependencyMux)); _writer = writer ?? throw new ArgumentNullException(nameof(writer)); + _passWriter = writer as IHistorianPassWriter; _outbox = outbox ?? throw new ArgumentNullException(nameof(outbox)); ArgumentNullException.ThrowIfNull(historizedRefs); // The ctor seed is the no-override identity (mux ref == historian name); production seeds an EMPTY @@ -740,7 +744,9 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers bool ok; try { - ok = await _writer.WriteLiveValuesAsync(g.Key, g.Values, ct).ConfigureAwait(false); + ok = _passWriter is { } passWriter + ? await passWriter.WriteLiveValuesInPassAsync(g.Key, g.Values, ct).ConfigureAwait(false) + : await _writer.WriteLiveValuesAsync(g.Key, g.Values, ct).ConfigureAwait(false); } catch (OperationCanceledException) when (ct.IsCancellationRequested) { @@ -776,6 +782,27 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers // where the not-yet-entered tasks fault out of WaitAsync without reaching the finally. (string Tag, int Count, bool Ok)[] outcomes = await Task.WhenAll(tasks).ConfigureAwait(false); + // A pass-aware writer completes the pass ONCE (QuestDB: one WAL visibility wait for every group, instead of + // one per group) — whenever any group wrote, acked or not, because an un-acked pass is redelivered and the + // redelivery's insert-if-absent check must see what this pass wrote. Before the ack. A flush failure does not + // fail the pass: the rows are committed, only their visibility is unproven. + if (_passWriter is { } flushWriter && outcomes.Any(o => o.Ok)) + { + try + { + await flushWriter.FlushAsync(ct).ConfigureAwait(false); + } + catch (OperationCanceledException) when (ct.IsCancellationRequested) + { + throw; + } + catch (Exception ex) + { + _log.Warning("ContinuousHistorization: pass flush failed ({Exception}); the pass's writes are committed.", + ex.GetType().Name); + } + } + // Fold the outcomes on the single logical drain flow — NEVER inside the parallel lambdas. // _tagWriteFailures is actor state (kept single-flight by _draining); mutating it from N // concurrent tasks would corrupt the failure counters. Nothing below observes completion diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/IHistorianPassWriter.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/IHistorianPassWriter.cs new file mode 100644 index 000000000..6e198293c --- /dev/null +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/IHistorianPassWriter.cs @@ -0,0 +1,36 @@ +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Historian; + +/// +/// An whose per-write completion work can be done ONCE per drain pass instead of +/// once per tag group — QuestDB's WAL visibility wait, which a backend needs so that an immediate redelivery's +/// insert-if-absent check sees the rows just written. The continuous-historization recorder writes every group of a +/// pass with and then calls once, after every group +/// has reported and before it acks the pass. +/// +/// +/// The cancellation contract is the recorder's for both members: a cancellation of the caller's token (the pass) is +/// rethrown; any other failure of a write is false. A failure does not fail the pass — +/// the rows are committed; only their visibility to the next pass is unproven. +/// +public interface IHistorianPassWriter : IHistorianValueWriter +{ + /// + /// Writes a batch of live values for a single tag as part of a drain pass, deferring the per-write completion + /// work to . Same result contract as . + /// + /// Fully-qualified historian tag the values are recorded against. + /// The samples to write, in append order. + /// The pass's cancellation token. + /// true when the backend stored the batch; false on a retryable failure. + Task WriteLiveValuesInPassAsync(string tag, IReadOnlyList values, CancellationToken ct); + + /// + /// Completes the pass: does the deferred completion work for every write since the last flush (QuestDB: one WAL + /// visibility wait). A no-op when nothing was written. + /// + /// The pass's cancellation token. + /// A task that completes when the pass's writes are visible (or the wait gave up). + Task FlushAsync(CancellationToken ct); +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/WriteTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/WriteTests.cs index a090346f4..71b6c1a53 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/WriteTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/WriteTests.cs @@ -85,6 +85,40 @@ public sealed class WriteTests(QuestDbFixture fx) } } + [Fact] + public async Task A_drain_pass_waits_for_wal_visibility_once_not_once_per_tag_group() + { + fx.SkipIfUnavailable(); + var o = fx.Options(fx.UniquePrefix()); + var c = new QuestDbConnection(o); + var b = new QuestDbSchemaBootstrapper(c, o); + var catalogWal = new QuestDbWalVisibility(c, o.VisibilityTimeout); + var sampleWal = new QuestDbWalVisibility(c, o.VisibilityTimeout); // the writer's own: counts only sample waits + var w = new QuestDbHistorianValueWriter(c, new QuestDbTagCatalog(c, b, catalogWal), sampleWal, utcNow: () => ServerNow); + + foreach (var tag in new[] { "pass.a", "pass.b", "pass.c" }) + (await w.WriteLiveValuesInPassAsync(tag, [new(T0, HistorianValue.FromDouble(1), 0)], CancellationToken.None)).ShouldBeTrue(); + sampleWal.Waits.ShouldBe(0, "in-pass writes defer the visibility wait"); + + await w.FlushAsync(CancellationToken.None); + sampleWal.Waits.ShouldBe(1, "one wait for the whole pass"); + foreach (var tag in new[] { "pass.a", "pass.b", "pass.c" }) + (await RowsAsync(c, tag)).Count.ShouldBe(1, $"{tag} is visible after the flush"); + + // A redelivery of the pass (the insert-if-absent check reads what the flush made visible) keeps one row each — + // and, having inserted nothing, owes no wait. + foreach (var tag in new[] { "pass.a", "pass.b", "pass.c" }) + (await w.WriteLiveValuesInPassAsync(tag, [new(T0, HistorianValue.FromDouble(2), 0)], CancellationToken.None)).ShouldBeTrue(); + await w.FlushAsync(CancellationToken.None); + foreach (var tag in new[] { "pass.a", "pass.b", "pass.c" }) + (await RowsAsync(c, tag)).Single().N.ShouldBe(1d); + sampleWal.Waits.ShouldBe(1); + + // The plain per-call path still waits per call (positive control). + (await w.WriteLiveValuesAsync("pass.d", [new(T0, HistorianValue.FromDouble(4), 0)], CancellationToken.None)).ShouldBeTrue(); + sampleWal.Waits.ShouldBe(2); + } + [Fact] public async Task A_redelivered_batch_leaves_one_row_and_the_first_value() { @@ -210,13 +244,16 @@ public sealed class WriteTests(QuestDbFixture fx) } [Fact] - public async Task A_cancelled_write_returns_false_without_throwing() + public async Task A_cancelled_write_rethrows_so_the_recorder_does_not_book_a_tag_failure() { + // The recorder's contract: a cancelled drain pass is not this tag's failure (it used to return false, which the + // recorder counted toward the tag's quarantine budget). fx.SkipIfUnavailable(); var (w, _) = Make(); using var cts = new CancellationTokenSource(); cts.Cancel(); - (await w.WriteLiveValuesAsync("x", [new(T0, HistorianValue.FromDouble(1), 0)], cts.Token)).ShouldBeFalse(); + await Should.ThrowAsync( + () => w.WriteLiveValuesAsync("x", [new(T0, HistorianValue.FromDouble(1), 0)], cts.Token)); } private static async Task SeedRowAsync(QuestDbConnection c, QuestDbWalVisibility wal, string tag, DateTime t, double v) diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests/QuestDbWalVisibilityTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests/QuestDbWalVisibilityTests.cs index b4f717140..0ed543c09 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests/QuestDbWalVisibilityTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests/QuestDbWalVisibilityTests.cs @@ -30,4 +30,12 @@ public sealed class QuestDbWalVisibilityTests [InlineData(4, 5, false)] public void Reached_is_writerTxn_at_or_beyond_the_target(long writer, long target, bool expected) => QuestDbWalVisibility.Reached(writer, target).ShouldBe(expected); + + [Fact] + public void The_poll_backs_off_exponentially_from_10_ms_to_a_200_ms_cap() + { + var delays = new List { QuestDbWalVisibility.FirstPollInterval }; + while (delays.Count < 8) delays.Add(QuestDbWalVisibility.NextPollInterval(delays[^1])); + delays.Select(d => (int)d.TotalMilliseconds).ShouldBe([10, 20, 40, 80, 160, 200, 200, 200]); + } } diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/ContinuousHistorizationRecorderPassFlushTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/ContinuousHistorizationRecorderPassFlushTests.cs new file mode 100644 index 000000000..13f27c6d4 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/ContinuousHistorizationRecorderPassFlushTests.cs @@ -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; + +/// +/// A pass-aware writer () 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. +/// +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"); + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/RecorderTestDoubles.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/RecorderTestDoubles.cs new file mode 100644 index 000000000..8d4e948d2 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/RecorderTestDoubles.cs @@ -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; + +/// +/// 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. +/// +internal sealed class EventLog +{ + private readonly Lock _gate = new(); + private readonly List _events = new(); + + public void Add(string e) { lock (_gate) _events.Add(e); } + + public IReadOnlyList Snapshot() { lock (_gate) return _events.ToArray(); } +} + +/// FIFO outbox honouring the truncate-through contract; appends can be made to yield or block. +internal sealed class LoggingOutbox(EventLog log) : IHistorizationOutbox +{ + private readonly Lock _gate = new(); + private readonly List _entries = new(); + private int _appendBatchCalls; + + /// When set, every AppendBatchAsync waits on it (on a pool thread) before committing — a slow fsync. + public TaskCompletionSource? AppendGate { get; set; } + + /// Completed by an append that has entered (before waiting on ). + public TaskCompletionSource AppendEntered { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); + + /// When set, CountAsync blocks on it — a hung outbox. + public TaskCompletionSource? CountGate { get; set; } + + public int AppendBatchCalls { get { lock (_gate) return _appendBatchCalls; } } + + public long DroppedCount => 0; + + public IReadOnlyList Entries { get { lock (_gate) return _entries.ToArray(); } } + + public ValueTask AppendAsync(HistorizationOutboxEntry entry, CancellationToken ct) => AppendBatchAsync([entry], ct); + + public async ValueTask AppendBatchAsync(IReadOnlyList 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> PeekBatchAsync(int max, CancellationToken ct) + { + lock (_gate) return ValueTask.FromResult>(_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 CountAsync(CancellationToken ct) + { + if (CountGate is { } gate) await gate.Task.WaitAsync(ct).ConfigureAwait(false); + lock (_gate) return _entries.Count; + } + + public void Dispose() { } +} + +/// A pass-aware writer: logs every in-pass write and flush; plain writes are logged distinctly. +internal sealed class LoggingPassWriter(EventLog log) : IHistorianPassWriter +{ + private readonly HashSet _failTags = new(StringComparer.Ordinal); + + public Exception? FlushThrows { get; set; } + + public void Fail(string tag) { lock (_failTags) _failTags.Add(tag); } + + public Task WriteLiveValuesAsync(string tag, IReadOnlyList values, CancellationToken ct) + { + log.Add($"plain:{tag}"); + return Task.FromResult(true); + } + + public async Task WriteLiveValuesInPassAsync(string tag, IReadOnlyList 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; + } +}