dohertj2/lmxopcua
1
0
Fork 0
Code Issues 32 Pull Requests Actions Packages Projects Releases Wiki Activity

perf(historian): batch ingest — one group commit per value burst

Browse Source
This commit is contained in:
Joseph Doherty
2026-08-16 04:33:35 -04:00
parent aad7583b79
commit 384568b47a
4 changed files with 455 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationRecorder.cs
+176 -73
View File
@@ -28,8 +28,10 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.Historian;
/// non-coercible value (string/null/reference) is dropped and metered rather than appended.
/// </para>
/// <para>
/// <b>Non-blocking drain.</b> Appends run on the actor thread (an awaited
/// <see cref="IHistorizationOutbox.AppendAsync"/> so appends stay serialized — the durable
/// <b>Batched ingest, non-blocking drain.</b> A mux burst arrives as one
/// <see cref="VirtualTagActor.DependencyValueBatch"/> and is committed with ONE
/// <see cref="IHistorizationOutbox.AppendBatchAsync"/> 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 <c>Self</c> (<see cref="DrainResult"/>) 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).</param>
/// <param name="AbandonedEntries">Lifetime count of entries discarded because their historian tag
/// kept failing while other tags were writing successfully (Gitea #538).</param>
/// <param name="IngestFaults">Lifetime count of values dropped because the per-value gate/coercion step
/// threw — distinct from <paramref name="DroppedNonNumeric"/>, 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.</param>
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;
/// <summary>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 <see cref="IngestAsync"/>.</summary>
private bool _ingestFaultLogged;
/// <summary>Last observed state of <see cref="_captureGate"/>, so a change is logged once rather
/// than on every value. Seeded open: a gate that starts closed logs its first suppression.</summary>
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
/// <summary>
/// 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).
/// <para>
/// ⚠️ The await below deliberately does NOT use <c>ConfigureAwait(false)</c>. <see
/// cref="OnValueChangedAsync"/> reads <c>Self</c> 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 <c>NotSupportedException</c> — 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.
/// </para>
/// burst. The whole burst is ingested as ONE durable append (see <see cref="IngestAsync"/>), so a
/// 50-value scan costs one group commit instead of 50 per-entry fsyncs.
/// </summary>
/// <param name="msg">The batch of matched dependency values.</param>
private async Task OnValueBatchAsync(VirtualTagActor.DependencyValueBatch msg)
/// <returns>A task that completes when the burst has been appended (or dropped).</returns>
private Task OnValueBatchAsync(VirtualTagActor.DependencyValueBatch msg) => IngestAsync(msg.Items);
/// <summary>
/// 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.
/// </summary>
/// <param name="msg">The matched dependency value.</param>
/// <returns>A task that completes when the value has been appended (or dropped).</returns>
private Task OnValueChangedAsync(VirtualTagActor.DependencyValueChanged msg) => IngestAsync([msg]);
/// <summary>
/// Ingest one burst of mux-fanned values: run the per-value gates over every item, then append the
/// survivors with a SINGLE <see cref="IHistorizationOutbox.AppendBatchAsync"/> group commit.
/// </summary>
/// <remarks>
/// <para>
/// <b>The gates stay per value.</b> 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 <c>The_capture_gate_is_re_read_per_value_so_promotion_resumes_capture</c>).
/// Only the durable append is batched.
/// </para>
/// <para>
/// <b>One bad item cannot drop the tail.</b> The per-item gate/coercion step is wrapped: an
/// unexpected throw (most plausibly a faulting <c>captureGate</c> delegate) is metered as an
/// <c>IngestFaults</c>, 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).
/// </para>
/// <para>
/// <b>Append failure is batch-wide.</b> A failed group commit is metered once
/// (<c>OutboxAppendFailures</c>), 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.
/// </para>
/// </remarks>
/// <param name="items">The burst's matched dependency values, in fan-out order.</param>
/// <returns>A task that completes when the burst has been appended (or dropped).</returns>
private async Task IngestAsync(IReadOnlyList<VirtualTagActor.DependencyValueChanged> items)
{
IReadOnlyList<VirtualTagActor.DependencyValueChanged> 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<HistorizationOutboxEntry>? 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)
/// <summary>
/// Runs the per-value gates for one mux-fanned value and, if it survives them, adds one outbox
/// entry per historian name to <paramref name="entries"/> (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.
/// </summary>
/// <param name="msg">The matched dependency value.</param>
/// <param name="entries">The burst's accumulating entry list; null until the first survivor.</param>
private void CollectEntries(
VirtualTagActor.DependencyValueChanged msg,
ref List<HistorizationOutboxEntry>? 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<HistorizationOutboxEntry>();
// 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);
}
/// <summary>
@@ -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;
src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ScriptedAlarms/ScriptedAlarmHostActor.cs
+12
View File
@@ -401,6 +401,18 @@ public sealed class ScriptedAlarmHostActor : ReceiveActor
/// <summary>One batched mux delivery — every item already matched this host's registered interest.
/// Loops the per-value handler, preserving in-burst order.</summary>
/// <remarks>
/// <b>Mid-batch exception policy: unguarded, deliberately.</b> An exception escaping
/// <see cref="OnDependencyChanged"/> 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
/// <c>IUpstreamValueSource.Push</c> 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
/// <c>ContinuousHistorizationRecorder.IngestAsync</c>, which DOES catch per item because its
/// per-item step runs value coercion.
/// </remarks>
/// <param name="msg">The batch of matched dependency values.</param>
private void OnDependencyBatch(VirtualTagActor.DependencyValueBatch msg)
{
var items = msg.Items;
src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/DependencyConsumerMuxAdapter.cs
+10
View File
@@ -73,6 +73,16 @@ public sealed class DependencyConsumerMuxAdapter : ReceiveActor
/// <summary>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.</summary>
/// <remarks>
/// <b>Mid-batch exception policy: unguarded, matching the single-value path.</b> An exception
/// escaping <see cref="IDependencyConsumer.OnDependencyValue"/> 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 <c>OnDependencyValue</c>.
/// </remarks>
/// <param name="msg">The batch of matched dependency values.</param>
private void OnDependencyValueBatch(VirtualTagActor.DependencyValueBatch msg)
{
var items = msg.Items;
tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/ContinuousHistorizationRecorderTests.cs
+257
View File
@@ -718,6 +718,225 @@ public sealed class ContinuousHistorizationRecorderTests : TestKit
Assert.Equal(1, writer.MaxObservedConcurrency);
}
// ---------------- Batched ingest: ONE group commit per value burst ----------------
/// <summary>
/// A whole driver burst delivered as one <see cref="VirtualTagActor.DependencyValueBatch"/> costs
/// exactly ONE durable append, and every value in it still reaches the writer.
/// </summary>
/// <remarks>
/// The load-bearing assertion is <c>AppendBatchCallCount == 1</c> paired with
/// <c>AppendCallCount == 0</c>. "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 <c>FasterLogHistorizationOutbox</c> under <c>PerEntry</c> commit mode)
/// collapsed into one group commit.
/// </remarks>
[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<DependencyMuxActor.RegisterInterest>();
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.RecorderStatus>(
ContinuousHistorizationRecorder.GetStatus.Instance, TimeSpan.FromSeconds(3));
Assert.Equal(0, status.QueuedDepth);
Assert.Equal(5, status.TotalRecorded);
}, TimeSpan.FromSeconds(10));
}
/// <summary>
/// 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.
/// </summary>
/// <remarks>
/// Mirrors <see cref="A_yielding_append_neither_faults_the_actor_nor_loses_the_value"/>, and for
/// the same reason: the ingest handler awaits
/// <see cref="IHistorizationOutbox.AppendBatchAsync"/> with <c>ConfigureAwait(false)</c>, so a
/// <c>Self</c> read after that await finds no ActorContext, throws
/// <see cref="NotSupportedException"/> into supervision, and RESTARTS the actor — which in
/// production discards every historized ref (the deployed set arrives later via
/// <see cref="ContinuousHistorizationRecorder.UpdateHistorizedRefs"/>).
/// <para>
/// The outbox double must complete on a POOL thread (<c>Task.Delay(1).ConfigureAwait(false)</c>),
/// which is what a real group-commit fsync does. A synchronously-completing fake — or a bare
/// <c>Task.Yield()</c>, which posts back to the CAPTURED ActorTaskScheduler context — leaves the
/// context installed and the defect invisible.
/// </para>
/// </remarks>
[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<DependencyMuxActor.RegisterInterest>();
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.RecorderStatus>(
ContinuousHistorizationRecorder.GetStatus.Instance, TimeSpan.FromSeconds(3));
Assert.Equal(3, status.TotalRecorded);
mux.ExpectNoMsg(TimeSpan.FromMilliseconds(300)); // no PreStart re-register == no restart
}
/// <summary>
/// A closed capture gate suppresses every item of a batch, and appends NOTHING (Gitea #537 on the
/// batched path).
/// </summary>
/// <remarks>
/// Extends <see cref="A_closed_capture_gate_suppresses_the_value"/> 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. <c>SuppressedNotPrimary == 3</c> is the positive observation — "the outbox is
/// empty" is equally true of a recorder that never ran.
/// </remarks>
[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.RecorderStatus>(
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);
}
/// <summary>
/// One item whose gate step THROWS is skipped; the rest of the burst still records, and the actor
/// does not restart.
/// </summary>
/// <remarks>
/// 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
/// <c>captureGate</c>, which is what this drives: the gate throws on exactly the SECOND of three
/// items.
/// <para>
/// 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.
/// </para>
/// </remarks>
[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<DependencyMuxActor.RegisterInterest>();
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.RecorderStatus>(
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
}
/// <summary>
/// <see cref="IHistorianValueWriter"/> double that BLOCKS every call until <see cref="Release"/>,
/// 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<HistorizationOutboxEntry> _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
}
}
/// <summary>Lifetime count of single-entry <see cref="AppendAsync"/> 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.</summary>
public int AppendCallCount
{
get
{
lock (_gate)
{
return _appendCalls;
}
}
}
/// <summary>Lifetime count of <see cref="AppendBatchAsync"/> calls — one per group commit.</summary>
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<HistorizationOutboxEntry> entries, CancellationToken ct)
{
lock (_gate)
{
_appendBatchCalls++;
}
if (entries.Count == 0)
{
return ValueTask.CompletedTask;