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 266aba86..799d9606 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationRecorder.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationRecorder.cs @@ -92,11 +92,17 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers /// A point-in-time snapshot of the recorder's counters. /// Un-acked entries currently held in the durable outbox. - /// Lifetime count of values appended to the outbox. + /// Lifetime count of values durably appended to the outbox. Credited per + /// BURST, on a successful group commit — so it UNDER-counts when a batch append fails after the outbox + /// durably committed a partial prefix: the outbox does not report how far it got, so the recorder cannot + /// credit that prefix. Those entries still drain; they are simply never counted here. /// Lifetime count of values dropped for not being numeric-coercible. /// Lifetime count of entries the outbox dropped on capacity overflow. - /// Lifetime count of durable-boundary append failures (the value - /// was dropped, not recorded, and the actor stayed alive rather than restart-looping). + /// Lifetime count of FAILED BATCH APPENDS at the durable boundary — + /// ONE per failed burst whatever its size, never one per value (the granularity changed with the batched + /// ingest; the per-value era counted one per value). The actor stays alive rather than restart-looping. + /// It is NOT a count of values lost: the outbox contract is "prefix, never orphan", so a failed burst + /// may still have left entries durably queued that drain normally — see . /// Whether the most recent drain pass acked cleanly. /// Lifetime count of values not captured because this node is /// the Secondary of a pair whose Primary records the same refs (Gitea #537). @@ -200,7 +206,7 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers IHistorianValueWriter writer, IHistorizationOutbox outbox, IReadOnlyList historizedRefs, - int drainBatchSize = 64, + int drainBatchSize = 1024, TimeSpan? drainInterval = null, TimeSpan? minBackoff = null, TimeSpan? maxBackoff = null, @@ -232,7 +238,7 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers IHistorianValueWriter writer, IHistorizationOutbox outbox, IReadOnlyList historizedRefs, - int drainBatchSize = 64, + int drainBatchSize = 1024, TimeSpan? drainInterval = null, TimeSpan? minBackoff = null, TimeSpan? maxBackoff = null, @@ -252,7 +258,10 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers AddRef(r, r); } - _drainBatchSize = drainBatchSize > 0 ? drainBatchSize : 64; + // Both this fallback and the ctor/Props defaults above must track ContinuousHistorizationOptions + // .DrainBatchSize (1024): production always passes the option through, so a divergent literal here + // only ever shows up as a quietly different batch size in tests and in degenerate-input callers. + _drainBatchSize = drainBatchSize > 0 ? drainBatchSize : 1024; _drainInterval = drainInterval is { } di && di > TimeSpan.Zero ? di : TimeSpan.FromSeconds(2); _minBackoff = minBackoff is { } mb && mb > TimeSpan.Zero ? mb : TimeSpan.FromSeconds(1); _maxBackoff = maxBackoff is { } xb && xb > TimeSpan.Zero ? xb : TimeSpan.FromSeconds(30); @@ -339,11 +348,14 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers /// #491 note below). /// /// - /// Append failure is batch-wide. A failed group commit is metered once - /// (OutboxAppendFailures), logged by category, and the burst's values are dropped — the - /// same never-restart-loop policy the per-value path had, applied to the batch. Note the outbox - /// contract makes any partially-appended prefix fully tracked and ackable, so the trailing drain - /// nudge still flushes it; nothing is left buffered-but-unackable. + /// Append failure is batch-wide, and it is NOT a whole-burst loss. A failed group commit + /// is metered ONCE (OutboxAppendFailures — one per failed burst whatever its size, never + /// one per value) and logged by category, keeping the same never-restart-loop policy the + /// per-value path had. 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 + /// fully tracked and ackable and the trailing drain nudge flushes it normally — nothing is left + /// buffered-but-unackable. That prefix does NOT credit TotalRecorded, because the outbox + /// does not report how far it got; the counter under-counts rather than over-claims. /// /// /// The burst's matched dependency values, in fan-out order. @@ -433,9 +445,20 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers // 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 drop this burst's values. + // 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) dropped.", + _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); } diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/ContinuousHistorizationRecorderTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/ContinuousHistorizationRecorderTests.cs index d917adc6..b4f44cca 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/ContinuousHistorizationRecorderTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/ContinuousHistorizationRecorderTests.cs @@ -937,6 +937,140 @@ public sealed class ContinuousHistorizationRecorderTests : TestKit mux.ExpectNoMsg(TimeSpan.FromMilliseconds(300)); // no PreStart re-register == no restart } + /// + /// A failed group commit is metered ONCE for the whole burst, credits nothing to + /// TotalRecorded, leaves the actor alive, and STILL nudges the drain — which flushes whatever + /// prefix the outbox durably committed before it failed. + /// + /// + /// Pins the batched ingest's append-failure catch branch, which had no coverage at all: every other + /// test here uses an outbox that always succeeds, so a catch that miscounted, swallowed the drain + /// nudge, or rethrew into supervision would go unnoticed. + /// + /// Three separate claims, each of which is a distinct way the branch can be wrong: + /// + /// + /// (1) OutboxAppendFailures == 1 for a THREE-item burst. The counter is per failed batch, + /// not per value — a deliberate granularity change from the per-value era — so a catch that + /// credited entries.Count would report 3 and make an operator read one bad fsync as three + /// separate faults. + /// + /// + /// (2) TotalRecorded == 0 even though entry "A" IS durable. The outbox does not report how + /// far it got, so the recorder deliberately under-counts rather than over-claims; this asserts + /// the documented direction of the inaccuracy. + /// + /// + /// (3) The durable prefix reaches the writer. This is the load-bearing half — it is the positive + /// observation that the post-failure DrainTick was sent, and it is why the log message + /// must not claim the burst was "dropped". drainInterval is set to 30 s so the steady + /// cadence cannot fire inside the test: the only thing that can produce a drain here is the + /// nudge. Deleting self.Tell(DrainTick.Instance) from the failure path (or returning from + /// the catch) leaves "A" queued forever and turns this red. + /// + /// + [Fact] + public async Task A_failed_batch_append_counts_one_failure_and_still_ticks_the_drain() + { + var mux = CreateTestProbe(); + var writer = new FakeValueWriter { Succeed = true }; + var outbox = new InMemoryOutbox + { + // The partial-prefix shape: "A" commits durably, then the commit fails — so "B" and "C" are + // genuinely lost while "A" is merely un-drained. + AppendBatchFault = new IOException("simulated fsync failure (disk full)"), + AppendBatchFaultAfter = 1, + }; + + IActorRef rec = Sys.ActorOf(ContinuousHistorizationRecorder.Props( + mux.Ref, writer, outbox, historizedRefs: new[] { "A", "B", "C" }, + drainInterval: TimeSpan.FromSeconds(30))); + mux.ExpectMsg(); + + var stamp = new DateTime(2026, 8, 15, 13, 0, 0, DateTimeKind.Utc); + rec.Tell(new VirtualTagActor.DependencyValueBatch(new[] + { + new VirtualTagActor.DependencyValueChanged("A", 1.0, stamp), + new VirtualTagActor.DependencyValueChanged("B", 2.0, stamp), + new VirtualTagActor.DependencyValueChanged("C", 3.0, stamp), + })); + + // Answering GetStatus at all is itself the "actor is still alive" observation — an append failure + // that reached supervision would have restarted (or stopped) it. + await AwaitAssertAsync( + async () => + { + var status = await rec.Ask( + ContinuousHistorizationRecorder.GetStatus.Instance, TimeSpan.FromSeconds(3)); + Assert.Equal(1, status.OutboxAppendFailures); + Assert.Equal(0, status.TotalRecorded); + }, + TimeSpan.FromSeconds(10)); + + await AwaitAssertAsync( + () => + { + Assert.True(outbox.PeekBatchCallCount >= 1, "the failure must still have nudged a drain"); + Assert.Contains(writer.Snapshot(), w => w.Tag == "A" && w.Value == 1.0); + return Task.CompletedTask; + }, + TimeSpan.FromSeconds(10)); + + // What never committed stays gone — the prefix drained, the remainder did not resurrect. + Assert.DoesNotContain(writer.Snapshot(), w => w.Tag == "B"); + Assert.DoesNotContain(writer.Snapshot(), w => w.Tag == "C"); + mux.ExpectNoMsg(TimeSpan.FromMilliseconds(300)); // no PreStart re-register == no restart + } + + /// + /// An that is NOT shutdown is metered as an append failure, + /// not swallowed. + /// + /// + /// The catch that quietly discards a cancelled append is filtered on + /// when (_lifetime.IsCancellationRequested) — i.e. only during the actor's own shutdown. + /// Drop the filter and every cancelled append becomes invisible: no metric, and no drain nudge for + /// the prefix. This drives the observable side of that filter (the lifetime is open here, so the + /// general catch must take it). + /// + /// The shutdown side itself is deliberately not tested: _lifetime is cancelled in + /// PostStop, so reaching that branch means racing an append against a stopping actor — + /// and once it has stopped there is no GetStatus to read the counter from. There is no + /// cheap seam for it; the filter's correctness is pinned from this side instead. + /// + /// + [Fact] + public async Task A_cancellation_that_is_not_shutdown_is_metered_as_an_append_failure() + { + var mux = CreateTestProbe(); + var writer = new FakeValueWriter { Succeed = true }; + var outbox = new InMemoryOutbox + { + AppendBatchFault = new OperationCanceledException("not a shutdown"), + AppendBatchFaultAfter = 0, + }; + + IActorRef rec = Sys.ActorOf(ContinuousHistorizationRecorder.Props( + mux.Ref, writer, outbox, historizedRefs: new[] { "A" }, + drainInterval: TimeSpan.FromSeconds(30))); + mux.ExpectMsg(); + + rec.Tell(new VirtualTagActor.DependencyValueChanged( + "A", 1.0, new DateTime(2026, 8, 15, 14, 0, 0, DateTimeKind.Utc))); + + await AwaitAssertAsync( + async () => + { + var status = await rec.Ask( + ContinuousHistorizationRecorder.GetStatus.Instance, TimeSpan.FromSeconds(3)); + Assert.Equal(1, status.OutboxAppendFailures); + Assert.Equal(0, status.TotalRecorded); + }, + TimeSpan.FromSeconds(10)); + + mux.ExpectNoMsg(TimeSpan.FromMilliseconds(300)); // no PreStart re-register == no restart + } + /// /// double that BLOCKS every call until , /// recording the peak number of calls in flight simultaneously. That peak is what distinguishes a @@ -1097,6 +1231,34 @@ public sealed class ContinuousHistorizationRecorderTests : TestKit private int _appendBatchCalls; private readonly bool _yieldOnAppend; + private int _peekCalls; + + /// When non-null, durably appends the first + /// entries and then throws this — the production outbox's + /// "prefix, never orphan" contract on its failure path (an fsync hitting disk-full mid-commit). + /// The throw happens AFTER a pool-thread yield, as a real group-commit fsync does, so the + /// recorder's ConfigureAwait(false) continuation runs its catch block off the actor + /// thread — the Gitea #491 shape, on the failure path. + public Exception? AppendBatchFault { get; init; } + + /// How many of the batch's entries are durably committed before + /// is thrown. 0 = nothing committed; a positive value models the + /// PARTIAL-PREFIX failure, where some entries are durable and will drain even though the append + /// as a whole reported failure. + public int AppendBatchFaultAfter { get; init; } + + /// Lifetime count of calls — one per drain pass, so a + /// non-zero value is positive evidence that a drain actually RAN. + public int PeekBatchCallCount + { + get + { + lock (_gate) + { + return _peekCalls; + } + } + } /// Drop-oldest capacity; 0 means unbounded. /// @@ -1175,6 +1337,11 @@ public sealed class ContinuousHistorizationRecorderTests : TestKit _appendBatchCalls++; } + if (AppendBatchFault is not null) + { + return AppendBatchFaultingAsync(entries); + } + if (entries.Count == 0) { return ValueTask.CompletedTask; @@ -1193,6 +1360,24 @@ public sealed class ContinuousHistorizationRecorderTests : TestKit return ValueTask.CompletedTask; } + private async ValueTask AppendBatchFaultingAsync(IReadOnlyList entries) + { + // Complete on a POOL thread before faulting (the same requirement AppendYieldingAsync documents): + // a real group commit fsyncs, so the recorder's ConfigureAwait(false) continuation — here, its + // CATCH block — resumes with no ActorContext installed. A synchronous throw would run the catch + // inline on the actor thread and would not exercise that. + await Task.Delay(1).ConfigureAwait(false); + + // "Prefix, never orphan": whatever was committed before the failure is durably tracked and + // ackable, which is exactly what makes the post-failure drain nudge meaningful. + for (var i = 0; i < AppendBatchFaultAfter && i < entries.Count; i++) + { + AppendCore(entries[i]); + } + + throw AppendBatchFault!; + } + private async ValueTask AppendBatchYieldingAsync(IReadOnlyList entries) { // Same pool-thread-completion requirement as AppendYieldingAsync below — a group commit @@ -1235,6 +1420,7 @@ public sealed class ContinuousHistorizationRecorderTests : TestKit { lock (_gate) { + _peekCalls++; IReadOnlyList batch = _entries.Take(max).ToArray(); return ValueTask.FromResult(batch); }