diff --git a/docs/Historian.md b/docs/Historian.md index 92df06cb6..2d1c665ee 100644 --- a/docs/Historian.md +++ b/docs/Historian.md @@ -266,7 +266,7 @@ and is registered the same way for every backend. |---|---|---|---| | `Enabled` | bool | `false` | `true` **and** a configured `ServerHistorian` wires the FasterLog outbox + the selected backend's value writer and spawns the `ContinuousHistorizationRecorder`. `false` → no recorder; driver tag values are not historized. | | `OutboxPath` | string | `""` | **Directory** holding the FasterLog segment + commit files. Required when `Enabled`. In production set an **absolute** path on durable storage — a relative path resolves against the host's working directory. | -| `CommitMode` | string | `PerEntry` | `PerEntry` fsyncs before the append returns (no loss window); `Periodic` batches commits onto a background timer every `CommitIntervalMs`. Parsed case-insensitively; an unrecognized value falls back to `PerEntry`. **See the batch-commit note below — `PerEntry` is one fsync per ingest *burst*, not per value.** | +| `CommitMode` | string | `Periodic` | ⚠️ **Default changed from `PerEntry`.** `Periodic` commits on a background timer every `CommitIntervalMs` (a crash can lose at most the last interval of captured values); `PerEntry` fsyncs before the append returns (no loss window). Under `PerEntry` ingestion was capped at `1 / fsync_latency` appends/s. Parsed case-insensitively. ⚠️ The shipped `appsettings.json` / docker-dev still restate `PerEntry` (and the Host falls back to `PerEntry` for an unparseable value) until the Host-side follow-up lands — set `Periodic` explicitly to adopt it now. | | `CommitIntervalMs` | int | `100` | Periodic-mode commit cadence (ms); must be `> 0` under `Periodic`, ignored under `PerEntry`. | | `DrainBatchSize` | int | `1024` | Outbox entries peeked + written per drain pass. **Raised from 64**: the peek is an in-memory scan of the outbox's live index (widening it costs allocation, not I/O), and a pass is now bounded-parallel over `DrainConcurrency` rather than sequential, so a wide batch no longer costs `tags × latency`. At 64 a drain drew down at most 64 entries every `DrainIntervalSeconds` — below the arrival rate of even a modest historized tag set — so any gateway hiccup left the outbox permanently behind, growing toward `Capacity` and shedding oldest samples instead of catching up. | | `DrainIntervalSeconds` | double | `2` | Steady drain cadence (and the post-success reschedule). | @@ -277,7 +277,16 @@ and is registered the same way for every backend. | `MinBackoffSeconds` | double | `1` | Initial retry backoff after a failed drain pass. | | `MaxBackoffSeconds` | double | `30` | Cap on the exponential retry backoff. | -> **`PerEntry` now means one fsync per ingest *burst*, not per value.** The recorder appends a whole +> **The append is off the recorder's mailbox.** An ingest burst is queued to one ordered append loop, which commits +> every burst waiting behind it with a single `AppendBatchAsync` (FIFO kept). Up to **64 bursts** may be queued or +> appending; only a full queue holds the recorder's mailbox until room frees (backpressure on the mux, bounded memory) — the +> append used to be awaited on the mailbox, so a slow fsync throttled ingestion while the mux kept telling into an unbounded +> mailbox. Stopping the recorder flushes the queued bursts within the shared 5 s stop budget. `GetStatus` reports +> `PendingAppends`, `MailboxDepth` and `AppendBackpressureWaits`, no longer suspends the mailbox on the outbox count, and +> answers `QueuedDepth = -1` when the count does not return within 10 s. Gauges: `otopcua.historization.recorder.pending_appends` +> and `otopcua.historization.recorder.mailbox_depth`. + +> **`PerEntry` means one fsync per append, not per value.** The recorder appends a whole > value burst through `IHistorizationOutbox.AppendBatchAsync`, which enqueues every entry and then > issues a **single** group commit — so a burst of N values costs one fsync rather than N. The > durability contract is unchanged: the call does not return until that commit completes, so nothing diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationMetrics.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationMetrics.cs index 9333ab1ca..09be12f5a 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationMetrics.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationMetrics.cs @@ -42,6 +42,28 @@ public static class ContinuousHistorizationMetrics () => _outbox?.UnreadableCount ?? 0L, unit: "{entry}", description: "Lifetime undecodable records the continuous-historization outbox skipped."); + OtOpcUaTelemetry.Meter.CreateObservableGauge( + "otopcua.historization.recorder.pending_appends", + () => (long)Volatile.Read(ref _pendingAppends), + unit: "{burst}", + description: "Ingest bursts queued for, or in, the recorder's durable append at its last status (backpressure nears at the in-flight cap)."); + OtOpcUaTelemetry.Meter.CreateObservableGauge( + "otopcua.historization.recorder.mailbox_depth", + () => (long)Volatile.Read(ref _mailboxDepth), + unit: "{message}", + description: "Messages waiting in the continuous-historization recorder's mailbox at its last status (-1 = unknown)."); + } + + private static int _pendingAppends; + private static int _mailboxDepth = -1; + + /// Publishes the recorder's append-queue and mailbox depth for the gauges (called on each status). + /// Bursts queued for or in the durable append. + /// Messages waiting in the recorder's mailbox (-1 = unknown). + public static void ReportRecorder(int pendingAppends, int mailboxDepth) + { + Volatile.Write(ref _pendingAppends, pendingAppends); + Volatile.Write(ref _mailboxDepth, mailboxDepth); } /// Binds the process outbox the gauges observe. Called once by the Host when continuous diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationOptions.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationOptions.cs index 3029a3ef5..026fc424e 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationOptions.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationOptions.cs @@ -38,22 +38,20 @@ public sealed class ContinuousHistorizationOptions public string OutboxPath { get; init; } = ""; /// - /// Outbox commit cadence: PerEntry (the default) fsyncs the log before each append - /// returns (safest, no loss window); Periodic batches commits onto a background timer - /// every ms (higher throughput, a bounded worst-case loss window). - /// Parsed case-insensitively against HistorizationCommitMode; an unrecognized value falls - /// back to PerEntry. + /// Outbox commit cadence: Periodic (the default) commits on a background timer every + /// ms (100 ms by default — a crash can lose at most the last interval of captured + /// values); PerEntry fsyncs before each append returns (no loss window, one fsync per append). Parsed + /// case-insensitively against HistorizationCommitMode. /// - /// PerEntry now means one fsync per ingest burst, not per value. The - /// recorder appends a whole value burst through - /// IHistorizationOutbox.AppendBatchAsync, which enqueues every entry and then issues a - /// single group commit — so a burst of N values costs one fsync instead of N. The - /// durability contract is unchanged: AppendBatchAsync does not return until that - /// commit completes, so nothing the caller has been told is buffered can be lost to an - /// unclean restart. + /// Why Periodic. Under PerEntry every ingest append waits for an fsync, so ingestion is capped at + /// 1 / fsync_latency appends per second (50 drivers × 1 Hz on a 10 ms-fsync disk kept the recorder half + /// busy; at 100+ bursts/s it fell behind for good). The recorder now appends off its mailbox and coalesces + /// every burst waiting behind a slow append into one group commit, which helps PerEntry too — but it is + /// still one fsync per group on the critical path. PerEntry remains available where no loss window is + /// acceptable. /// /// - public string CommitMode { get; init; } = "PerEntry"; + public string CommitMode { get; init; } = "Periodic"; /// Periodic-mode commit cadence in milliseconds; must be positive when /// is Periodic. Ignored under PerEntry. 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 833f3df18..2943ae4ca 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationRecorder.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationRecorder.cs @@ -1,4 +1,5 @@ using System.Collections.Frozen; +using System.Threading.Channels; using Akka.Actor; using Akka.Event; using ZB.MOM.WW.OtOpcUa.Core.Abstractions; @@ -47,15 +48,16 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.Historian; /// rather than appended. /// /// -/// Batched ingest, non-blocking drain. A mux burst arrives as one -/// and is committed with ONE -/// group commit rather than N per-entry -/// fsyncs. Appends run on the actor thread (awaited, so appends stay serialized — the durable -/// boundary completes before the next message), but the drain runs off the actor thread and -/// pipes its outcome back to Self () so the mailbox is never -/// blocked on the gateway write. A drain never throws into the actor: the drain task catches -/// everything and reports a failed , which backs the next attempt off -/// (exponential, capped). +/// Batched, off-mailbox ingest; non-blocking drain. A mux burst arrives as one +/// ; its gated entries are queued to ONE ordered append loop +/// (FIFO preserved), which commits every burst waiting behind it with a single +/// group commit. Up to maxAppendsInFlight bursts may be +/// queued or appending; only a full queue makes the handler wait (backpressure on the mux, bounded memory). The +/// append used to be awaited on the mailbox, so ingestion was capped at 1 / fsync_latency bursts/s while the +/// mux kept telling into an unbounded mailbox. The drain runs off the actor thread too and pipes its outcome back to +/// Self (). Neither ever throws into the actor: failures are counted and the next +/// drain backs off (exponential, capped). Stopping flushes the queued bursts within +/// . /// /// public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers @@ -128,6 +130,14 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers /// threw — distinct from , which counts values the backend /// cannot carry. Non-zero here means an ingest dependency is faulting (most /// plausibly the supplied capture gate), not that the data was unsuitable. + /// Ingest bursts queued for, or in, the durable append (not yet recorded). Near + /// maxAppendsInFlight means the outbox cannot keep up and ingestion is being backpressured. + /// Messages waiting in the recorder's mailbox when the status was taken (-1 = unknown). + /// A growing value means values arrive faster than the recorder accepts them. + /// Lifetime count of bursts that found the append queue full and held the + /// mailbox until room freed. + /// is -1 when the outbox did not report its count within the status + /// timeout. public sealed record RecorderStatus( int QueuedDepth, long TotalRecorded, @@ -137,7 +147,22 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers bool LastDrainSucceeded, long SuppressedNotPrimary = 0, long AbandonedEntries = 0, - long IngestFaults = 0); + long IngestFaults = 0, + int PendingAppends = 0, + int MailboxDepth = -1, + long AppendBackpressureWaits = 0); + + /// The default cap on ingest bursts queued for or in the durable append. + public const int DefaultMaxAppendsInFlight = 64; + + /// The default bound on 's outbox count. + public static readonly TimeSpan DefaultStatusTimeout = TimeSpan.FromSeconds(10); + + /// + /// How long stopping waits for the append loop to flush the bursts still queued (the runtime's shared actor-stop + /// budget). Past it the remainder is abandoned (logged) — the same loss an unclean stop always had. + /// + public static readonly TimeSpan AppendFlushBudget = TimeSpan.FromSeconds(5); private readonly IActorRef _dependencyMux; private readonly IHistorianValueWriter _writer; @@ -183,6 +208,17 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers private readonly ILoggingAdapter _log = Context.GetLogger(); private readonly CancellationTokenSource _lifetime = new(); + /// The ordered queue of gated ingest bursts awaiting the durable append (bounded: backpressure). + private readonly Channel> _appendQueue; + + /// The single append loop draining (started in ). + private Task _appendLoop = Task.CompletedTask; + + private readonly TimeSpan _statusTimeout; + private int _pendingAppends; + private long _appendBackpressureWaits; + private volatile bool _stopping; + private bool _draining; private TimeSpan _currentBackoff; private DateTime _nextAllowedDrainUtc = DateTime.MinValue; @@ -230,6 +266,10 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers /// positive; 1 is the fully sequential drain). /// The value kinds the backend can store /// (); null ⇒ . + /// Ingest bursts that may be queued for or in the durable append before the handler + /// waits for room (backpressure); clamped to at least 1. Default . + /// Bound on 's outbox count (past it the depth is reported -1); + /// default . /// Props for the actor. public static Props Props( IActorRef dependencyMux, @@ -243,10 +283,12 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers Func? captureGate = null, int maxWriteAttempts = 5, int drainConcurrency = 8, - IReadOnlySet? supportedKinds = null) => + IReadOnlySet? supportedKinds = null, + int maxAppendsInFlight = DefaultMaxAppendsInFlight, + TimeSpan? statusTimeout = null) => Akka.Actor.Props.Create(() => new ContinuousHistorizationRecorder( dependencyMux, writer, outbox, historizedRefs, drainBatchSize, drainInterval, minBackoff, maxBackoff, - captureGate, maxWriteAttempts, drainConcurrency, supportedKinds)); + captureGate, maxWriteAttempts, drainConcurrency, supportedKinds, maxAppendsInFlight, statusTimeout)); /// Initializes a new instance of the class. /// The per-node dependency mux to register historized-ref interest with. @@ -266,6 +308,8 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers /// at least 1; 1 is the fully sequential drain. /// The value kinds the backend can store; null ⇒ /// . + /// Ingest bursts queued for or in the durable append before the handler waits. + /// Bound on 's outbox count. public ContinuousHistorizationRecorder( IActorRef dependencyMux, IHistorianValueWriter writer, @@ -278,8 +322,18 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers Func? captureGate = null, int maxWriteAttempts = 5, int drainConcurrency = 8, - IReadOnlySet? supportedKinds = null) + IReadOnlySet? supportedKinds = null, + int maxAppendsInFlight = DefaultMaxAppendsInFlight, + TimeSpan? statusTimeout = null) { + _appendQueue = Channel.CreateBounded>( + new BoundedChannelOptions(Math.Max(1, maxAppendsInFlight)) + { + SingleReader = true, + SingleWriter = true, + FullMode = BoundedChannelFullMode.Wait, + }); + _statusTimeout = statusTimeout is { } st && st > TimeSpan.Zero ? st : DefaultStatusTimeout; _dependencyMux = dependencyMux ?? throw new ArgumentNullException(nameof(dependencyMux)); _writer = writer ?? throw new ArgumentNullException(nameof(writer)); _passWriter = writer as IHistorianPassWriter; @@ -314,14 +368,10 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers Receive(OnUpdateHistorizedRefs); Receive(_ => OnDrainTick()); Receive(OnDrainResult); - ReceiveAsync(async _ => - { - // Capture Sender before the await: although Akka restores the actor context across awaits, - // capturing first is the robust idiom (Sender after an await is brittle). - IActorRef replyTo = Sender; - RecorderStatus status = await BuildStatusAsync().ConfigureAwait(false); - replyTo.Tell(status); - }); + // Not a ReceiveAsync: awaiting the outbox count used to suspend the mailbox for as long as the count took (a hung + // SQLite/FasterLog lock stalled ingestion with it). The counters are read here, the count runs off the mailbox, + // bounded by the status timeout, and the reply is piped to the asker. + Receive(_ => BuildStatusAsync().PipeTo(Sender)); } /// @@ -332,6 +382,8 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers _dependencyMux.Tell(new DependencyMuxActor.RegisterInterest(_refMap.Keys.ToList(), Self)); // Seed the steady drain cadence; appends also nudge a prompt drain (see IngestAsync). Timers.StartSingleTimer(DrainTimerKey, DrainTick.Instance, _drainInterval); + IActorRef self = Self; + _appendLoop = Task.Run(() => AppendLoopAsync(self)); base.PreStart(); } @@ -341,6 +393,20 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers // Drop our mux interest eagerly (mirrors VirtualTagActor.PostStop) so the mux stops fanning to // us immediately, closing the dead-letter window between this stop and the mux's Terminated. _dependencyMux.Tell(new DependencyMuxActor.UnregisterInterest(Self)); + // Flush the bursts still queued for the append (they were accepted from the mux) within the shared stop budget, + // then cancel whatever is left. Bounded: a stuck fsync cannot pin this dispatcher thread past the budget. + _stopping = true; + _appendQueue.Writer.TryComplete(); + try + { + if (!_appendLoop.Wait(AppendFlushBudget)) + _log.Warning("ContinuousHistorization: {Count} ingest burst(s) were still waiting for the outbox when the " + + "recorder stopped; abandoned after {Seconds} s.", Volatile.Read(ref _pendingAppends), AppendFlushBudget.TotalSeconds); + } + catch (AggregateException) + { + // The loop never throws; defensive. + } _lifetime.Cancel(); _lifetime.Dispose(); base.PostStop(); @@ -411,8 +477,6 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers // throw escaped the handler into supervision: the actor restarted, and because production seeds an // EMPTY ref set (the deployed set arrives later via UpdateHistorizedRefs), the restart silently // dropped every historized ref — so historization stopped dead after the FIRST value. - IActorRef self = Self; - List? entries = null; for (int i = 0; i < items.Count; i++) { @@ -461,49 +525,92 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers return; } - // Durable boundary: ONE group commit for the whole burst, awaited so appends stay serialized, - // BEFORE the values are considered captured. The outbox drops the oldest entry on capacity - // overflow and tracks DroppedCount. - try + // Durable boundary, OFF the mailbox: the burst joins the ordered append queue. Only a full queue holds the handler + // (and with it the mailbox) until room frees — backpressure instead of an unbounded mailbox. The append loop + // records the outcome and nudges a drain. + var pending = Interlocked.Increment(ref _pendingAppends); + // The gauges' source: the queue depth and the mailbox depth as this burst is accepted (the actor thread is the + // only place the mailbox can be read). + ContinuousHistorizationMetrics.ReportRecorder(pending, Context is ActorCell cell ? cell.NumberOfMessages : -1); + if (!_appendQueue.Writer.TryWrite(entries)) { - await _outbox.AppendBatchAsync(entries, _lifetime.Token).ConfigureAwait(false); - _totalRecorded += entries.Count; - } - catch (OperationCanceledException) when (_lifetime.IsCancellationRequested) - { - // Normal shutdown raced the append — not a fault. Drop quietly, and do not nudge a drain on - // an actor that is stopping. - return; - } - catch (Exception ex) - { - // A durable-boundary failure (e.g. an fsync hitting disk-full / I/O error) must NEVER propagate - // out of the handler — that would trip Akka supervision into a restart, and under a persistent - // disk fault the actor would restart-loop (re-register → next burst → append fails → restart → - // …). Mirror the drain path's catch-all: meter the failure (category only, no value content) - // and carry on. - // - // The meter is ONE per failed burst, whatever the burst's size — a deliberate granularity change - // from the per-value era, where it was one per value. And it is NOT a count of values lost: the - // outbox commits a PREFIX or nothing (never an orphan), so an unknown number of this burst's - // entries may already be durable. Hence the message says "failed to append", not "dropped" — - // claiming total loss here would be wrong about the prefix the drain nudge below is about to - // flush. TotalRecorded is likewise not credited for that prefix: the outbox does not report how - // far it got, so the counter under-counts rather than over-claims. - _outboxAppendFailures++; - _log.Warning( - "ContinuousHistorization: outbox batch append failed ({Exception}); {Count} value(s) in the " - + "burst failed to append. Any prefix the outbox committed before the failure stays queued " - + "and drains normally; the remainder are lost.", - ex.GetType().Name, entries.Count); + Interlocked.Increment(ref _appendBackpressureWaits); + try + { + await _appendQueue.Writer.WriteAsync(entries, _lifetime.Token).ConfigureAwait(false); + } + catch (Exception ex) when (ex is OperationCanceledException or ChannelClosedException) + { + // Normal shutdown raced the enqueue — not a fault. + Interlocked.Decrement(ref _pendingAppends); + } } + } - // Nudge a prompt drain attempt; the DrainTick handler de-dups (already draining) and honours - // any active backoff window, so this never bypasses a failure cooldown. Sent even after a failed - // append, because the outbox contract leaves any durably-appended prefix tracked and ackable — - // it should drain now rather than wait for the steady timer. Uses the pre-await capture, never - // `Self` — see the note above. - self.Tell(DrainTick.Instance); + /// + /// The single, ordered durable-append loop: takes the next queued burst plus every burst already waiting behind it, + /// commits them with ONE (FIFO kept), records the outcome, and + /// nudges a drain. Never throws. + /// + /// + /// Append failure is batch-wide, and it is NOT a whole-burst loss. A failed group commit is metered ONCE per + /// burst it carried (OutboxAppendFailures) and logged by category. What is actually lost is only what the + /// outbox never committed: its contract is "prefix, never orphan", so anything it durably appended before the + /// failure stays tracked and ackable and the drain nudge flushes it. That prefix does NOT credit + /// TotalRecorded (the outbox does not report how far it got): the counter under-counts rather than + /// over-claims. Runs off the actor thread, so it touches no actor state but the Interlocked counters and the + /// captured . + /// + private async Task AppendLoopAsync(IActorRef self) + { + var reader = _appendQueue.Reader; + while (true) + { + try + { + if (!await reader.WaitToReadAsync(CancellationToken.None).ConfigureAwait(false)) return; + } + catch (ChannelClosedException) + { + return; + } + + var bursts = 0; + var merged = new List(); + while (reader.TryRead(out var burst)) + { + merged.AddRange(burst); + bursts++; + } + if (bursts == 0) continue; + + try + { + await _outbox.AppendBatchAsync(merged, _lifetime.Token).ConfigureAwait(false); + Interlocked.Add(ref _totalRecorded, merged.Count); + } + catch (OperationCanceledException) when (_lifetime.IsCancellationRequested) + { + // Shutdown past the flush budget raced the append — not a fault. + Interlocked.Add(ref _pendingAppends, -bursts); + return; + } + catch (Exception ex) + { + // A durable-boundary failure (e.g. an fsync hitting disk-full / I/O error) never stops the loop: meter it + // (category only, no value content) and carry on. One count per burst, whatever its size. + Interlocked.Add(ref _outboxAppendFailures, bursts); + _log.Warning( + "ContinuousHistorization: outbox batch append failed ({Exception}); {Count} value(s) in {Bursts} burst(s) " + + "failed to append. Any prefix the outbox committed before the failure stays queued and drains normally; " + + "the remainder are lost.", ex.GetType().Name, merged.Count, bursts); + } + + Interlocked.Add(ref _pendingAppends, -bursts); + // Sent even after a failed append: any durably-appended prefix should drain now rather than wait for the + // steady timer. The DrainTick handler de-dups and honours an active backoff. Not while stopping (dead letters). + if (!_stopping) self.Tell(DrainTick.Instance); + } } /// @@ -928,19 +1035,50 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers Timers.StartSingleTimer(DrainTimerKey, DrainTick.Instance, delay); } - private async Task BuildStatusAsync() + /// + /// The status reply: every counter read now (on the actor thread), the outbox count awaited off the mailbox and + /// bounded by the status timeout — a count that does not answer in time is reported -1 (logged). + /// + private Task BuildStatusAsync() { - int depth = await _outbox.CountAsync(_lifetime.Token).ConfigureAwait(false); - return new RecorderStatus( - depth, - _totalRecorded, + var mailbox = Context is ActorCell cell ? cell.NumberOfMessages : -1; + var snapshot = new RecorderStatus( + 0, + Interlocked.Read(ref _totalRecorded), _droppedUnsupportedKind, _outbox.DroppedCount, - _outboxAppendFailures, + Interlocked.Read(ref _outboxAppendFailures), _lastDrainSucceeded, _suppressedNotPrimary, _abandonedEntries, - _ingestFaults); + _ingestFaults, + Volatile.Read(ref _pendingAppends), + mailbox, + Interlocked.Read(ref _appendBackpressureWaits)); + ContinuousHistorizationMetrics.ReportRecorder(snapshot.PendingAppends, mailbox); + var token = _lifetime.Token; + var timeout = _statusTimeout; + var log = _log; + return CountAsync(); + + async Task CountAsync() + { + try + { + var depth = await _outbox.CountAsync(token).AsTask().WaitAsync(timeout, token).ConfigureAwait(false); + return snapshot with { QueuedDepth = depth }; + } + catch (TimeoutException) + { + log.Warning("ContinuousHistorization: the outbox did not report its depth within {Seconds} s; status depth unknown.", + timeout.TotalSeconds); + return snapshot with { QueuedDepth = -1 }; + } + catch (OperationCanceledException) + { + return snapshot with { QueuedDepth = -1 }; + } + } } private static TimeSpan Min(TimeSpan a, TimeSpan b) => a < b ? a : b; diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/ContinuousHistorizationOptionsTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/ContinuousHistorizationOptionsTests.cs index 40deedf41..9e1bd4e40 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/ContinuousHistorizationOptionsTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/ContinuousHistorizationOptionsTests.cs @@ -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); diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/ContinuousHistorizationRecorderAppendPipelineTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/ContinuousHistorizationRecorderAppendPipelineTests.cs new file mode 100644 index 000000000..4759dc87e --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/ContinuousHistorizationRecorderAppendPipelineTests.cs @@ -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; + +/// +/// 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 1 / fsync_latency bursts/s while the mux keeps telling into an unbounded mailbox; only a full +/// queue suspends the mailbox (backpressure, bounded memory). GetStatus no longer suspends the mailbox on the +/// outbox count, and is bounded by a timeout. Pending appends and the mailbox depth are reported. +/// +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 StatusAsync(IActorRef rec, TimeSpan? timeout = null) => + rec.Ask(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"); + } +} 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 index 8d4e948d2..0a15c846f 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/RecorderTestDoubles.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/RecorderTestDoubles.cs @@ -23,6 +23,7 @@ internal sealed class LoggingOutbox(EventLog log) : IHistorizationOutbox { private readonly Lock _gate = new(); private readonly List _entries = new(); + private readonly List _appended = new(); private int _appendBatchCalls; /// When set, every AppendBatchAsync waits on it (on a pool thread) before committing — a slow fsync. @@ -40,6 +41,9 @@ internal sealed class LoggingOutbox(EventLog log) : IHistorizationOutbox public IReadOnlyList Entries { get { lock (_gate) return _entries.ToArray(); } } + /// Every entry ever appended, in append order (acks do not remove from it). + public IReadOnlyList Appended { get { lock (_gate) return _appended.ToArray(); } } + public ValueTask AppendAsync(HistorizationOutboxEntry entry, CancellationToken ct) => AppendBatchAsync([entry], ct); public async ValueTask AppendBatchAsync(IReadOnlyList 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}"); }