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 da8b0fa4..266aba86 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationRecorder.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationRecorder.cs @@ -28,8 +28,10 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.Historian; /// non-coercible value (string/null/reference) is dropped and metered rather than appended. /// /// -/// Non-blocking drain. Appends run on the actor thread (an awaited -/// so appends stay serialized — the durable +/// 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 @@ -100,6 +102,10 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers /// the Secondary of a pair whose Primary records the same refs (Gitea #537). /// Lifetime count of entries discarded because their historian tag /// kept failing while other tags were writing successfully (Gitea #538). + /// Lifetime count of values dropped because the per-value gate/coercion step + /// threw — distinct from , which counts values the historian's + /// numeric-only write path cannot carry. Non-zero here means an ingest dependency is faulting (most + /// plausibly the supplied capture gate), not that the data was unsuitable. public sealed record RecorderStatus( int QueuedDepth, long TotalRecorded, @@ -108,7 +114,8 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers long OutboxAppendFailures, bool LastDrainSucceeded, long SuppressedNotPrimary = 0, - long AbandonedEntries = 0); + long AbandonedEntries = 0, + long IngestFaults = 0); private readonly IActorRef _dependencyMux; private readonly IHistorianValueWriter _writer; @@ -156,8 +163,14 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers private long _outboxAppendFailures; private long _suppressedNotPrimary; private long _abandonedEntries; + private long _ingestFaults; private bool _lastDrainSucceeded = true; + /// Whether the current run of per-item ingest faults has already been reported at Warning, so + /// a per-value fault mode (a throwing capture gate) logs once rather than once per sample. Cleared by + /// the first item that ingests cleanly — see . + private bool _ingestFaultLogged; + /// Last observed state of , so a change is logged once rather /// than on every value. Seeded open: a gate that starts closed logs its first suppression. private bool _captureGateOpen = true; @@ -272,7 +285,7 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers // Register interest by mux ref (the key the mux fans DependencyValueChanged by) so it fans those // tags' values to us. Historian-name overrides are tracked separately for the write side. _dependencyMux.Tell(new DependencyMuxActor.RegisterInterest(_refMap.Keys.ToList(), Self)); - // Seed the steady drain cadence; appends also nudge a prompt drain (see OnValueChangedAsync). + // Seed the steady drain cadence; appends also nudge a prompt drain (see IngestAsync). Timers.StartSingleTimer(DrainTimerKey, DrainTick.Instance, _drainInterval); base.PreStart(); } @@ -290,29 +303,161 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers /// /// The mux's batched delivery: one message carrying this recorder's whole matched slice of a driver - /// burst. Each item is appended by the existing per-value handler, in order — the saving here is the - /// mailbox hop, not the append (a genuinely batched append is a later change). - /// - /// ⚠️ The await below deliberately does NOT use ConfigureAwait(false). reads Self before its own first await, so it must start - /// each item ON the actor thread; a context-dropping await here would leave item 2 onward with - /// no ActorContext and throw NotSupportedException — the Gitea #491 failure, which - /// supervision turns into a restart that silently discards every historized ref. A plain await - /// resumes on Akka's ActorTaskScheduler context, so the context is restored before the next - /// iteration. - /// + /// burst. The whole burst is ingested as ONE durable append (see ), so a + /// 50-value scan costs one group commit instead of 50 per-entry fsyncs. /// /// The batch of matched dependency values. - private async Task OnValueBatchAsync(VirtualTagActor.DependencyValueBatch msg) + /// A task that completes when the burst has been appended (or dropped). + private Task OnValueBatchAsync(VirtualTagActor.DependencyValueBatch msg) => IngestAsync(msg.Items); + + /// + /// A single mux-fanned value. Delegates to the same core as the batch path as a one-item burst, so + /// a direct sender (and every test that Tells one value) behaves identically to a batch of one. + /// + /// The matched dependency value. + /// A task that completes when the value has been appended (or dropped). + private Task OnValueChangedAsync(VirtualTagActor.DependencyValueChanged msg) => IngestAsync([msg]); + + /// + /// Ingest one burst of mux-fanned values: run the per-value gates over every item, then append the + /// survivors with a SINGLE group commit. + /// + /// + /// + /// The gates stay per value. The ref filter, the Gitea #537 capture gate and the numeric + /// coercion are evaluated for each item in turn — the capture gate in particular is a live read + /// per value, never hoisted to once per batch, so a promotion resumes capture on the very next + /// sample (pinned by The_capture_gate_is_re_read_per_value_so_promotion_resumes_capture). + /// Only the durable append is batched. + /// + /// + /// One bad item cannot drop the tail. The per-item gate/coercion step is wrapped: an + /// unexpected throw (most plausibly a faulting captureGate delegate) is metered as an + /// IngestFaults, logged by category, and the loop continues with the next item. Left + /// unguarded it would escape the handler into supervision, discarding every item after it in the + /// burst AND restarting the actor — which in production discards the historized-ref set (see the + /// #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. + /// + /// + /// The burst's matched dependency values, in fan-out order. + /// A task that completes when the burst has been appended (or dropped). + private async Task IngestAsync(IReadOnlyList items) { - IReadOnlyList items = msg.Items; + // Capture Self BEFORE the await below. Self reads ActorBase.Context, and the append is awaited with + // ConfigureAwait(false) — which drops Akka's ActorTaskScheduler synchronization context, so a + // continuation that actually resumes off the actor thread finds NO ActorContext and throws + // NotSupportedException. This is the same idiom the GetStatus handler applies to Sender. + // + // It only bites when the append genuinely YIELDS: the production FasterLog outbox fsyncs per group + // commit (PerEntry commit mode), so it does; a fake outbox returning ValueTask.CompletedTask + // completes synchronously, the continuation runs inline with the context still installed, and the + // bug is invisible. That is exactly how it reached production — see the yielding fake in + // ContinuousHistorizationRecorderTests. Live-caught on the docker-dev rig (Gitea #491), where the + // 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++) { - await OnValueChangedAsync(items[i]); + try + { + CollectEntries(items[i], ref entries); + } + catch (Exception ex) + { + // Defensive: keep one malformed item from dropping the rest of the burst (and from + // restarting the actor). Category only — no value content is logged. + // + // Metered under its own counter rather than folded into DroppedNonNumeric: the realistic + // source is a throwing _captureGate (a supplied delegate reading the redundancy role view), + // not a value the historian cannot carry, and conflating the two would make the + // non-numeric metric lie about a redundancy fault. + // + // Logged once per run of faults (the same latch idiom the capture gate uses above): the + // plausible fault modes are per-VALUE, so an unlatched Warning would emit one line per + // sample — at scan rates that is the log flood, not the diagnosis. + _ingestFaults++; + if (!_ingestFaultLogged) + { + _ingestFaultLogged = true; + _log.Warning( + "ContinuousHistorization: value ingest failed ({Exception}); value dropped. Further " + + "ingest faults are logged at Debug until one succeeds.", ex.GetType().Name); + } + else + { + _log.Debug("ContinuousHistorization: value ingest failed ({Exception}); value dropped.", + ex.GetType().Name); + } + + continue; + } + + // A clean item re-arms the fault log, so a NEW run of faults is reported at Warning again. + _ingestFaultLogged = false; } + + if (entries is null) + { + // Everything was filtered, suppressed or dropped — nothing to commit, and no drain to nudge + // (this burst added nothing; the steady timer still covers any earlier residue). + 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 + { + 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 drop this burst's values. + _outboxAppendFailures++; + _log.Warning("ContinuousHistorization: outbox batch append failed ({Exception}); {Count} value(s) dropped.", + ex.GetType().Name, entries.Count); + } + + // 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); } - private async Task OnValueChangedAsync(VirtualTagActor.DependencyValueChanged msg) + /// + /// Runs the per-value gates for one mux-fanned value and, if it survives them, adds one outbox + /// entry per historian name to (lazily created on first survivor). + /// Purely synchronous — it runs on the actor thread with the mailbox suspended, so it may read and + /// mutate actor state freely. + /// + /// The matched dependency value. + /// The burst's accumulating entry list; null until the first survivor. + private void CollectEntries( + VirtualTagActor.DependencyValueChanged msg, + ref List? entries) { // The mux fans values keyed by the driver ref (msg.TagId == the mux ref). Only historize refs we // registered interest for, and resolve the HISTORIAN NAME(s) to write under — which differ from the @@ -328,9 +473,10 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers // fan-out, so an ungated recorder writes every sample TWICE (measured on the docker-dev central // pair: the gateway's write counter advanced twice per value, and halved when one node stopped). // The gate closes only on a node whose pair partner is the Primary and is recording the same - // refs; it is read PER VALUE rather than latched, so promotion resumes capture on the very next - // sample with no restart. Note what is NOT gated: the drain. Entries already captured are this - // node's to deliver, and a demoted node must still empty its own outbox. + // refs; it is read PER VALUE rather than latched (and never hoisted out of this per-item step), + // so promotion resumes capture on the very next sample with no restart. Note what is NOT gated: + // the drain. Entries already captured are this node's to deliver, and a demoted node must still + // empty its own outbox. if (!_captureGate()) { _suppressedNotPrimary++; @@ -362,67 +508,23 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers var timestampUtc = DateTime.SpecifyKind(msg.TimestampUtc, DateTimeKind.Utc); - // Capture Self BEFORE the first await. Self reads ActorBase.Context, and the awaits below use - // ConfigureAwait(false) — which drops Akka's ActorTaskScheduler synchronization context, so any - // continuation that actually resumes off the actor thread finds NO ActorContext and throws - // NotSupportedException. This is the same idiom the GetStatus handler already applies to Sender. - // - // It only bites when the append genuinely YIELDS: the production FasterLog outbox fsyncs per entry - // (PerEntry commit mode), so it does; a fake outbox returning ValueTask.CompletedTask completes - // synchronously, the continuation runs inline with the context still installed, and the bug is - // invisible. That is exactly how it reached production — see the yielding fake in - // ContinuousHistorizationRecorderTests. Live-caught on the docker-dev rig (Gitea #491), where the - // 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; + entries ??= new List(); // One outbox entry PER historian name: a single mux fan-out must reach every aliased historized tag // (each its own historian name), not just one — collapsing to one name here would silently drop the - // rest. Snapshot the names first: although the mailbox is suspended across the awaits below (so the - // set cannot be mutated concurrently), iterating a snapshot is robust and clearer. - foreach (string historianName in historianNames.ToList()) + // rest. No snapshot copy is needed: this method is synchronous, so the set cannot be mutated while + // it is iterated. + foreach (string historianName in historianNames) { // Record under the historian name (override-or-FullName), NOT the mux ref — the outbox entry + // the drain's WriteLiveValues target the historian tag the EnsureTags hook provisioned. - var entry = new HistorizationOutboxEntry( + entries.Add(new HistorizationOutboxEntry( Guid.NewGuid(), historianName, numeric, GoodQuality, - timestampUtc); - - // Durable boundary: append (awaited so appends stay serialized) BEFORE the value is considered - // captured. The outbox drops the oldest entry on capacity overflow and tracks DroppedCount. - try - { - await _outbox.AppendAsync(entry, _lifetime.Token).ConfigureAwait(false); - } - catch (OperationCanceledException) when (_lifetime.IsCancellationRequested) - { - // Normal shutdown raced the append — not a fault. Drop quietly (abandon the rest too). - return; - } - catch (Exception ex) - { - // A durable-boundary failure (e.g. a PerEntry 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 value → - // append fails → restart → …). Mirror the drain path's catch-all: meter the failure - // (category only, no value content), drop THIS name's value, and continue with the rest. - _outboxAppendFailures++; - _log.Warning("ContinuousHistorization: outbox append failed ({Exception}); value dropped.", - ex.GetType().Name); - continue; - } - - _totalRecorded++; + timestampUtc)); } - - // 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. Uses the pre-await - // capture, never `Self` — see the note above. - self.Tell(DrainTick.Instance); } /// @@ -731,7 +833,8 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers _outboxAppendFailures, _lastDrainSucceeded, _suppressedNotPrimary, - _abandonedEntries); + _abandonedEntries, + _ingestFaults); } private static TimeSpan Min(TimeSpan a, TimeSpan b) => a < b ? a : b; diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ScriptedAlarms/ScriptedAlarmHostActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ScriptedAlarms/ScriptedAlarmHostActor.cs index aaebb6a7..621a0ea4 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ScriptedAlarms/ScriptedAlarmHostActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ScriptedAlarms/ScriptedAlarmHostActor.cs @@ -401,6 +401,18 @@ public sealed class ScriptedAlarmHostActor : ReceiveActor /// One batched mux delivery — every item already matched this host's registered interest. /// Loops the per-value handler, preserving in-burst order. + /// + /// Mid-batch exception policy: unguarded, deliberately. An exception escaping + /// drops the remaining items of this burst and restarts the actor + /// — exactly what it already did when the same values arrived as N separate messages (the tail is + /// then lost to the mailbox flush instead of to the loop), so batching changes the blast radius by + /// nothing. Catching per item is deliberately NOT done: the handler is a single + /// IUpstreamValueSource.Push into an in-memory cache, so a throw means the upstream itself is + /// broken and swallowing it would feed the engine a silently truncated burst. Contrast + /// ContinuousHistorizationRecorder.IngestAsync, which DOES catch per item because its + /// per-item step runs value coercion. + /// + /// The batch of matched dependency values. private void OnDependencyBatch(VirtualTagActor.DependencyValueBatch msg) { var items = msg.Items; diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/DependencyConsumerMuxAdapter.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/DependencyConsumerMuxAdapter.cs index ebe94dc0..74f82867 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/DependencyConsumerMuxAdapter.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/DependencyConsumerMuxAdapter.cs @@ -73,6 +73,16 @@ public sealed class DependencyConsumerMuxAdapter : ReceiveActor /// The mux's batched delivery: one message carrying this adapter's whole matched slice of a /// driver burst. The consumer contract is per value, so dispatch each item in order — the saving is the /// mailbox hop, not the consumer call. + /// + /// Mid-batch exception policy: unguarded, matching the single-value path. An exception + /// escaping drops the remaining items of this + /// burst and restarts the adapter — the same outcome the single-value handler already has (there the + /// tail is lost to the mailbox flush rather than to the loop), so batching does not widen it. + /// Catching per item HERE only would make the batch path quietly more forgiving than the single-value + /// one for the same consumer fault, so the two are deliberately kept identical: a consumer that needs + /// per-value fault tolerance owns it inside its own OnDependencyValue. + /// + /// The batch of matched dependency values. private void OnDependencyValueBatch(VirtualTagActor.DependencyValueBatch msg) { var items = msg.Items; 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 221f7bfa..d917adc6 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 @@ -718,6 +718,225 @@ public sealed class ContinuousHistorizationRecorderTests : TestKit Assert.Equal(1, writer.MaxObservedConcurrency); } + // ---------------- Batched ingest: ONE group commit per value burst ---------------- + + /// + /// A whole driver burst delivered as one costs + /// exactly ONE durable append, and every value in it still reaches the writer. + /// + /// + /// The load-bearing assertion is AppendBatchCallCount == 1 paired with + /// AppendCallCount == 0. "All five values arrived" was already true of the loop-over-the- + /// per-value-handler batch path this replaces, so on its own it does not distinguish the two + /// implementations at all; the call counts are what state the actual change — five per-entry + /// fsyncs (production FasterLogHistorizationOutbox under PerEntry commit mode) + /// collapsed into one group commit. + /// + [Fact] + public async Task A_burst_appends_once_and_drains_all_values() + { + var mux = CreateTestProbe(); + var writer = new FakeValueWriter { Succeed = true }; + var outbox = new InMemoryOutbox(); + + var refs = new[] { "T1", "T2", "T3", "T4", "T5" }; + IActorRef rec = Sys.ActorOf(ContinuousHistorizationRecorder.Props( + mux.Ref, writer, outbox, historizedRefs: refs, + drainInterval: TimeSpan.FromMilliseconds(50))); + mux.ExpectMsg(); + + var stamp = new DateTime(2026, 8, 15, 9, 0, 0, DateTimeKind.Utc); + rec.Tell(new VirtualTagActor.DependencyValueBatch( + refs.Select((r, i) => new VirtualTagActor.DependencyValueChanged(r, (double)(i + 1), stamp)) + .ToArray())); + + await AwaitAssertAsync( + () => + { + var written = writer.Snapshot(); + Assert.Equal(5, written.Count); + Assert.Equal(refs.OrderBy(x => x, StringComparer.Ordinal), + written.Select(w => w.Tag).OrderBy(x => x, StringComparer.Ordinal)); + return Task.CompletedTask; + }, + TimeSpan.FromSeconds(10)); + + Assert.Equal(1, outbox.AppendBatchCallCount); + Assert.Equal(0, outbox.AppendCallCount); + + // …and the pass acked: the whole prefix left the outbox. + await AwaitAssertAsync(async () => + { + var status = await rec.Ask( + ContinuousHistorizationRecorder.GetStatus.Instance, TimeSpan.FromSeconds(3)); + Assert.Equal(0, status.QueuedDepth); + Assert.Equal(5, status.TotalRecorded); + }, TimeSpan.FromSeconds(10)); + } + + /// + /// The batched append must also survive genuinely yielding — the Gitea #491 bug class, on the + /// path that now owns the only await in the ingest handler. + /// + /// + /// Mirrors , and for + /// the same reason: the ingest handler awaits + /// with ConfigureAwait(false), so a + /// Self read after that await finds no ActorContext, throws + /// into supervision, and RESTARTS the actor — which in + /// production discards every historized ref (the deployed set arrives later via + /// ). + /// + /// The outbox double must complete on a POOL thread (Task.Delay(1).ConfigureAwait(false)), + /// which is what a real group-commit fsync does. A synchronously-completing fake — or a bare + /// Task.Yield(), which posts back to the CAPTURED ActorTaskScheduler context — leaves the + /// context installed and the defect invisible. + /// + /// + [Fact] + public async Task A_yielding_batch_append_neither_faults_the_actor_nor_loses_values() + { + var mux = CreateTestProbe(); + var writer = new FakeValueWriter { Succeed = true }; + var outbox = new InMemoryOutbox(yieldOnAppend: true); + + IActorRef rec = Sys.ActorOf(ContinuousHistorizationRecorder.Props( + mux.Ref, writer, outbox, historizedRefs: new[] { "A", "B", "C" }, + drainInterval: TimeSpan.FromMilliseconds(200))); + mux.ExpectMsg(); + + var stamp = new DateTime(2026, 8, 15, 10, 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), + })); + + // Every value in the burst reaches the writer… + await AwaitAssertAsync( + () => + { + Assert.Equal(3, writer.Snapshot().Count); + return Task.CompletedTask; + }, + TimeSpan.FromSeconds(10)); + + // …and — the load-bearing half — the actor is still the SAME incarnation, holding the ref set it + // was seeded with. A restart would reset TotalRecorded and re-register interest from PreStart. + var status = await rec.Ask( + ContinuousHistorizationRecorder.GetStatus.Instance, TimeSpan.FromSeconds(3)); + Assert.Equal(3, status.TotalRecorded); + mux.ExpectNoMsg(TimeSpan.FromMilliseconds(300)); // no PreStart re-register == no restart + } + + /// + /// A closed capture gate suppresses every item of a batch, and appends NOTHING (Gitea #537 on the + /// batched path). + /// + /// + /// Extends to the batch handler: the + /// rewrite runs the gate per item while collecting entries, so a batched path that hoisted the + /// gate read (or skipped it entirely) would silently double-record on the Secondary of every + /// redundant pair. SuppressedNotPrimary == 3 is the positive observation — "the outbox is + /// empty" is equally true of a recorder that never ran. + /// + [Fact] + public async Task The_capture_gate_suppresses_items_within_a_batch() + { + var mux = CreateTestProbe(); + var writer = new FakeValueWriter(); + var outbox = new InMemoryOutbox(); + + IActorRef rec = Sys.ActorOf(ContinuousHistorizationRecorder.Props( + mux.Ref, writer, outbox, historizedRefs: new[] { "A", "B", "C" }, + captureGate: () => false)); + + var stamp = new DateTime(2026, 8, 15, 11, 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), + })); + + await AwaitAssertAsync(async () => + { + var status = await rec.Ask( + ContinuousHistorizationRecorder.GetStatus.Instance, TimeSpan.FromSeconds(3)); + Assert.Equal(3, status.SuppressedNotPrimary); + Assert.Equal(0, status.TotalRecorded); + Assert.Equal(0, status.QueuedDepth); + }); + + Assert.Empty(writer.Snapshot()); + Assert.Equal(0, outbox.AppendBatchCallCount); + Assert.Equal(0, outbox.AppendCallCount); + } + + /// + /// One item whose gate step THROWS is skipped; the rest of the burst still records, and the actor + /// does not restart. + /// + /// + /// Batching turns a per-item throw into a tail-dropper: unguarded, it escapes the handler with the + /// whole burst still uncommitted (the group commit happens after the loop), so every value — + /// including the ones already collected — is lost, and supervision restarts the actor, which in + /// production discards the historized-ref set. The realistic source is a faulting + /// captureGate, which is what this drives: the gate throws on exactly the SECOND of three + /// items. + /// + /// The load-bearing assertion is that BOTH surviving values reached the writer — the one BEFORE + /// the fault and the one AFTER it. The trailing one alone would not distinguish the guard from + /// a restart that happened to re-drain, and the leading one alone is not evidence the loop + /// continued. + /// + /// + [Fact] + public async Task A_faulting_item_does_not_drop_the_rest_of_the_burst() + { + var mux = CreateTestProbe(); + var writer = new FakeValueWriter { Succeed = true }; + var outbox = new InMemoryOutbox(); + var gateCalls = 0; + + IActorRef rec = Sys.ActorOf(ContinuousHistorizationRecorder.Props( + mux.Ref, writer, outbox, historizedRefs: new[] { "A", "B", "C" }, + drainInterval: TimeSpan.FromMilliseconds(50), + captureGate: () => Interlocked.Increment(ref gateCalls) == 2 + ? throw new InvalidOperationException("role view unavailable") + : true)); + mux.ExpectMsg(); + + var stamp = new DateTime(2026, 8, 15, 12, 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), + })); + + await AwaitAssertAsync( + () => + { + var written = writer.Snapshot(); + Assert.Contains(written, w => w.Tag == "A"); + Assert.Contains(written, w => w.Tag == "C"); + Assert.DoesNotContain(written, w => w.Tag == "B"); + return Task.CompletedTask; + }, + TimeSpan.FromSeconds(10)); + + var status = await rec.Ask( + ContinuousHistorizationRecorder.GetStatus.Instance, TimeSpan.FromSeconds(3)); + Assert.Equal(1, status.IngestFaults); + Assert.Equal(2, status.TotalRecorded); + // The fault is NOT a "value the historian cannot carry" — that counter must stay clean, or an + // ingest-dependency outage is indistinguishable from bad data on the /hosts surface. + Assert.Equal(0, status.DroppedNonNumeric); + 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 @@ -874,6 +1093,8 @@ public sealed class ContinuousHistorizationRecorderTests : TestKit private readonly List _entries = new(); private readonly int _capacity; private long _dropped; + private int _appendCalls; + private int _appendBatchCalls; private readonly bool _yieldOnAppend; @@ -905,8 +1126,39 @@ public sealed class ContinuousHistorizationRecorderTests : TestKit } } + /// Lifetime count of single-entry calls. Distinguishes the + /// batched ingest path from the per-value loop it replaced — both record the same values, so only + /// the call shape tells them apart. + public int AppendCallCount + { + get + { + lock (_gate) + { + return _appendCalls; + } + } + } + + /// Lifetime count of calls — one per group commit. + public int AppendBatchCallCount + { + get + { + lock (_gate) + { + return _appendBatchCalls; + } + } + } + public ValueTask AppendAsync(HistorizationOutboxEntry entry, CancellationToken ct) { + lock (_gate) + { + _appendCalls++; + } + if (_yieldOnAppend) { return AppendYieldingAsync(entry); @@ -918,6 +1170,11 @@ public sealed class ContinuousHistorizationRecorderTests : TestKit public ValueTask AppendBatchAsync(IReadOnlyList entries, CancellationToken ct) { + lock (_gate) + { + _appendBatchCalls++; + } + if (entries.Count == 0) { return ValueTask.CompletedTask;