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

fix(historian): record the real status code and typed values; backend-supported kinds gate capture

Browse Source 
The recorder now appends the typed HistorianValue and the mux's own StatusCode instead of a
double + hard-coded Good, and drops any kind not in the backend's SupportedKinds (resolved
optionally from IHistorianBackendProvider; absent => {Double, Int64, Boolean}, today's behaviour).
RecorderStatus.DroppedNonNumeric -> DroppedUnsupportedKind; GoodQuality const removed; stale
"mux carries no quality" remarks rewritten (comms-loss node stamps are not historized).

Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
This commit is contained in:
Joseph Doherty
2026-10-09 12:49:05 -04:00
parent ff7c6c531a
commit 04b882f5c6
5 changed files with 326 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationMetrics.cs
+1 -1
View File
@@ -13,7 +13,7 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.Historian;
/// thread on the actor mailbox. The outbox exposes both gauge sources cheaply and
/// non-blockingly — <see cref="IHistorizationOutbox.CountAsync"/> completes synchronously (it
/// just reads an in-memory FIFO count) and <see cref="IHistorizationOutbox.DroppedCount"/> is a
/// plain property. The recorder's other counters (TotalRecorded / DroppedNonNumeric /
/// plain property. The recorder's other counters (TotalRecorded / DroppedUnsupportedKind /
/// OutboxAppendFailures) remain available via its <c>GetStatus</c> Ask for a health hook, but are
/// not surfaced as gauges here (Ask-in-gauge is the awkward path the plan calls out).
/// </para>
src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationRecorder.cs
+58 -43
View File
@@ -11,25 +11,34 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.Historian;
/// values the mux fans for those refs — in production a
/// <see cref="VirtualTagActor.DependencyValueBatch"/> per driver burst, with the per-value
/// <see cref="VirtualTagActor.DependencyValueChanged"/> receive retained for direct senders and tests:
/// value → coerce to numeric → append to the durable <see cref="IHistorizationOutbox"/> (the crash
/// value → coerce to a typed <see cref="HistorianValue"/> (kind-filtered) → append to the durable <see cref="IHistorizationOutbox"/> (the crash
/// boundary) → a self-paced drain writes the outbox through <see cref="IHistorianValueWriter"/>'s
/// live-value path and acks (truncates) on success.
/// </summary>
/// <remarks>
/// <para>
/// <b>Message path (grounded against the real actors).</b> The mux fans each value — carrying
/// only <c>(TagId, Value, TimestampUtc)</c> and <em>no quality</em> — to every registered
/// subscriber, batched as a <see cref="VirtualTagActor.DependencyValueBatch"/> on the production
/// path and as a bare <see cref="VirtualTagActor.DependencyValueChanged"/> from a direct sender
/// or a test.
/// The recorder reuses that existing fan-out (no change to <c>DriverHostActor</c>), so values
/// arrive Good-quality by the same convention the scripted-alarm host uses (the mux only
/// forwards driver-published values); historized samples are recorded with
/// <see cref="GoodQuality"/>.
/// <b>Message path (grounded against the real actors).</b> The mux fans each value as
/// <c>(TagId, Value, TimestampUtc, StatusCode)</c> — <see cref="DependencyMuxActor"/> forwards
/// the driver publish's own <see cref="VirtualTagActor.DependencyValueChanged.StatusCode"/> — to
/// every registered subscriber, batched as a <see cref="VirtualTagActor.DependencyValueBatch"/> on
/// the production path and as a bare <see cref="VirtualTagActor.DependencyValueChanged"/> from a
/// direct sender or a test. The recorder reuses that existing fan-out (no change to
/// <c>DriverHostActor</c>) and records each sample with its REAL status code, so a driver's own
/// Bad/Uncertain publish (usually a <c>null</c> value ⇒ <see cref="HistorianValueKind.Empty"/>) is
/// historized as such when the backend can store it.
/// </para>
/// <para>
/// <b>Numeric-only gate.</b> The SQL analog live-value write path is numeric-only, so a
/// non-coercible value (string/null/reference) is dropped and metered rather than appended.
/// <b>What is NOT historized:</b> the comms-loss watchdog's <c>BadNoCommunication</c> stamps.
/// <c>DriverHostActor.FanCommsLossToValueNodes</c> writes those to the address-space NODES only
/// (status-only sink write) and never reaches the mux, so the recorder never sees them. Only a
/// value a driver itself publishes through the mux is recorded.
/// </para>
/// <para>
/// <b>Kind gate.</b> Each value is coerced with <see cref="HistorianValue.TryFromObject"/>; a
/// value it cannot carry (arrays, references, structures) or whose kind is not in the backend's
/// supported set (<see cref="IHistorianBackendProvider.SupportedKinds"/>; default
/// <see cref="DefaultSupportedKinds"/>) is dropped and metered as
/// <see cref="RecorderStatus.DroppedUnsupportedKind"/> rather than appended.
/// </para>
/// <para>
/// <b>Batched ingest, non-blocking drain.</b> A mux burst arrives as one
@@ -45,16 +54,10 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.Historian;
/// </remarks>
public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers
{
/// <summary>OPC-DA quality byte recorded for mux-fanned values (the mux drops quality; driver-published
/// values are Good by the same convention the scripted-alarm host applies).
/// <para>
/// Note (archreview 06/C-7): the historian gateway's <c>WriteLiveValues</c> SQL path <b>discards</b>
/// quality (server-managed column, 0.2.0 Contracts C-002 — see
/// <c>GatewayHistorianValueWriter</c>'s remarks). This constant is therefore what a future
/// quality-honoring gateway path <em>would</em> receive; capturing real per-node quality here will
/// NOT persist until the gateway honors the field.
/// </para></summary>
public const ushort GoodQuality = 192;
/// <summary>The value kinds captured when no backend provider states its own: the numeric kinds the
/// gateway's SQL analog live-value path has always carried (Double, Int64, Boolean).</summary>
public static readonly IReadOnlySet<HistorianValueKind> DefaultSupportedKinds =
new HashSet<HistorianValueKind> { HistorianValueKind.Double, HistorianValueKind.Int64, HistorianValueKind.Boolean };
private const string DrainTimerKey = "drain";
@@ -100,7 +103,8 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers
/// 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.</param>
/// <param name="DroppedNonNumeric">Lifetime count of values dropped for not being numeric-coercible.</param>
/// <param name="DroppedUnsupportedKind">Lifetime count of values dropped because they are not historizable
/// (arrays, references) or their kind is not one the backend supports.</param>
/// <param name="OutboxDropped">Lifetime count of entries the outbox dropped on capacity overflow.</param>
/// <param name="OutboxAppendFailures">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
@@ -113,13 +117,13 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers
/// <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
/// threw — distinct from <paramref name="DroppedUnsupportedKind"/>, 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.</param>
public sealed record RecorderStatus(
int QueuedDepth,
long TotalRecorded,
long DroppedNonNumeric,
long DroppedUnsupportedKind,
long OutboxDropped,
long OutboxAppendFailures,
bool LastDrainSucceeded,
@@ -157,6 +161,9 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers
/// many are outstanding at once so a wide batch cannot flood the gateway.</summary>
private readonly int _drainConcurrency;
/// <summary>The value kinds the backend can store; any other kind is dropped at capture.</summary>
private readonly IReadOnlySet<HistorianValueKind> _supportedKinds;
/// <summary>Consecutive failed write passes per historian tag, counted only across passes that
/// proved the backend reachable. Reset on the tag's first success. Mutated exclusively inside
/// <see cref="DrainOnceAsync"/>, which <see cref="_draining"/> keeps single-flight.</summary>
@@ -169,7 +176,7 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers
private TimeSpan _currentBackoff;
private DateTime _nextAllowedDrainUtc = DateTime.MinValue;
private long _totalRecorded;
private long _droppedNonNumeric;
private long _droppedUnsupportedKind;
private long _outboxAppendFailures;
private long _suppressedNotPrimary;
private long _abandonedEntries;
@@ -204,6 +211,8 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers
/// accumulate, while other tags are succeeding, before its entries are abandoned.</param>
/// <param name="drainConcurrency">Max historian tags one drain pass writes concurrently (must be
/// positive; 1 is the fully sequential drain).</param>
/// <param name="supportedKinds">The value kinds the backend can store
/// (<see cref="IHistorianBackendProvider.SupportedKinds"/>); null ⇒ <see cref="DefaultSupportedKinds"/>.</param>
/// <returns>Props for the actor.</returns>
public static Props Props(
IActorRef dependencyMux,
@@ -216,10 +225,11 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers
TimeSpan? maxBackoff = null,
Func<bool>? captureGate = null,
int maxWriteAttempts = 5,
int drainConcurrency = 8) =>
int drainConcurrency = 8,
IReadOnlySet<HistorianValueKind>? supportedKinds = null) =>
Akka.Actor.Props.Create(() => new ContinuousHistorizationRecorder(
dependencyMux, writer, outbox, historizedRefs, drainBatchSize, drainInterval, minBackoff, maxBackoff,
captureGate, maxWriteAttempts, drainConcurrency));
captureGate, maxWriteAttempts, drainConcurrency, supportedKinds));
/// <summary>Initializes a new instance of the <see cref="ContinuousHistorizationRecorder"/> class.</summary>
/// <param name="dependencyMux">The per-node dependency mux to register historized-ref interest with.</param>
@@ -237,6 +247,8 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers
/// least 1.</param>
/// <param name="drainConcurrency">Max historian tags one drain pass writes concurrently. Clamped to
/// at least 1; 1 is the fully sequential drain.</param>
/// <param name="supportedKinds">The value kinds the backend can store; null ⇒
/// <see cref="DefaultSupportedKinds"/>.</param>
public ContinuousHistorizationRecorder(
IActorRef dependencyMux,
IHistorianValueWriter writer,
@@ -248,7 +260,8 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers
TimeSpan? maxBackoff = null,
Func<bool>? captureGate = null,
int maxWriteAttempts = 5,
int drainConcurrency = 8)
int drainConcurrency = 8,
IReadOnlySet<HistorianValueKind>? supportedKinds = null)
{
_dependencyMux = dependencyMux ?? throw new ArgumentNullException(nameof(dependencyMux));
_writer = writer ?? throw new ArgumentNullException(nameof(writer));
@@ -276,6 +289,7 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers
_captureGate = captureGate ?? (static () => true);
_maxWriteAttempts = maxWriteAttempts > 0 ? maxWriteAttempts : 1;
_drainConcurrency = drainConcurrency > 0 ? drainConcurrency : 1;
_supportedKinds = supportedKinds ?? DefaultSupportedKinds;
ReceiveAsync<VirtualTagActor.DependencyValueChanged>(OnValueChangedAsync);
ReceiveAsync<VirtualTagActor.DependencyValueBatch>(OnValueBatchAsync);
@@ -337,8 +351,8 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers
/// </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
/// <b>The gates stay per value.</b> The ref filter, the Gitea #537 capture gate and the typed-value
/// coercion + kind gate 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.
@@ -393,10 +407,10 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers
// 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
// Metered under its own counter rather than folded into DroppedUnsupportedKind: 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.
// unsupported-kind 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
@@ -524,13 +538,14 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers
+ "resuming value capture.");
}
if (!(HistorianValue.TryFromObject(msg.Value, out HistorianValue historianValue)
&& historianValue.TryGetDouble(out double numeric)))
if (!HistorianValue.TryFromObject(msg.Value, out HistorianValue historianValue)
|| !_supportedKinds.Contains(historianValue.Kind))
{
// The SQL analog live-value path is numeric-only — drop non-coercible values (string/null/
// reference) and meter the drop. No value content is logged.
_droppedNonNumeric++;
_log.Debug("ContinuousHistorization: dropped non-numeric value for a historized ref.");
// Not historizable at all (array/reference/structure), or a kind this backend cannot store
// (e.g. String/Empty on the gateway's numeric-only SQL analog path) — drop and meter. No value
// content is logged.
_droppedUnsupportedKind++;
_log.Debug("ContinuousHistorization: dropped a value of a kind the historian backend does not support.");
return;
}
@@ -549,8 +564,8 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers
entries.Add(new HistorizationOutboxEntry(
Guid.NewGuid(),
historianName,
HistorianValue.FromDouble(numeric),
0u, // Good
historianValue,
msg.StatusCode, // the REAL status the driver published, not an assumed Good
timestampUtc));
}
}
@@ -870,7 +885,7 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers
return new RecorderStatus(
depth,
_totalRecorded,
_droppedNonNumeric,
_droppedUnsupportedKind,
_outbox.DroppedCount,
_outboxAppendFailures,
_lastDrainSucceeded,
src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs
+5 -1
View File
@@ -509,7 +509,11 @@ public static class ServiceCollectionExtensions
maxWriteAttempts: continuousOptions.MaxWriteAttempts,
// Bounded-parallel per-tag drain: the groups in one pass are independent
// gateway round-trips, so a sequential pass costs `tags × latency`.
drainConcurrency: continuousOptions.DrainConcurrency),
drainConcurrency: continuousOptions.DrainConcurrency,
// The backend states which value kinds it can store; the recorder drops the
// rest at capture. No provider registered ⇒ null ⇒ the recorder's numeric
// default {Double, Int64, Boolean} (the gateway's historic behaviour).
supportedKinds: resolver.GetService<IHistorianBackendProvider>()?.SupportedKinds),
ContinuousHistorizationRecorderActorName);
registry.Register<ContinuousHistorizationRecorderKey>(continuousRecorder);
}
tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/ContinuousHistorizationRecorderTests.cs
+9 -9
View File
@@ -13,14 +13,14 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Historian;
/// Verifies the <see cref="ContinuousHistorizationRecorder"/>: it registers historized-ref
/// interest with the dependency mux on start, appends mux-fanned
/// <see cref="VirtualTagActor.DependencyValueChanged"/> values to the durable outbox then drains
/// them to the live-value writer, retains entries when the writer fails, and drops non-numeric
/// values (the SQL analog write path is numeric-only).
/// them to the live-value writer, retains entries when the writer fails, and drops values of a kind
/// the backend does not support (the default kinds are the numeric ones).
/// </summary>
/// <remarks>
/// Adapted from the plan's Task 17: the recorder handles the REAL fan-out message the mux emits —
/// <see cref="VirtualTagActor.DependencyValueChanged"/> (TagId/Value/TimestampUtc, no quality) —
/// not <c>DriverInstanceActor.AttributeValuePublished</c>. The mux drops quality, so the recorder
/// records Good-quality (the same convention the scripted-alarm host uses for mux values).
/// <see cref="VirtualTagActor.DependencyValueChanged"/> (TagId/Value/TimestampUtc/StatusCode) —
/// not <c>DriverInstanceActor.AttributeValuePublished</c>. The typed-value / real-status behaviour is
/// pinned in <see cref="ContinuousHistorizationRecorderTypedValueTests"/>.
/// </remarks>
public sealed class ContinuousHistorizationRecorderTests : TestKit
{
@@ -334,7 +334,7 @@ public sealed class ContinuousHistorizationRecorderTests : TestKit
{
var status = await rec.Ask<ContinuousHistorizationRecorder.RecorderStatus>(
ContinuousHistorizationRecorder.GetStatus.Instance, TimeSpan.FromSeconds(3));
Assert.Equal(1, status.DroppedNonNumeric);
Assert.Equal(1, status.DroppedUnsupportedKind);
Assert.Equal(0, status.QueuedDepth);
});
Assert.Empty(writer.Snapshot());
@@ -1146,7 +1146,7 @@ public sealed class ContinuousHistorizationRecorderTests : TestKit
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);
Assert.Equal(0, status.DroppedUnsupportedKind);
mux.ExpectNoMsg(TimeSpan.FromMilliseconds(300)); // no PreStart re-register == no restart
}
@@ -1467,7 +1467,7 @@ public sealed class ContinuousHistorizationRecorderTests : TestKit
_calls++;
foreach (HistorizationValue v in values)
{
_written.Add(new WrittenValue(tag, v.Value.Number, (ushort)v.StatusCode, v.TimestampUtc));
_written.Add(new WrittenValue(tag, v.Value.Number, v.StatusCode, v.TimestampUtc));
}
// A throwing tag overrides every ack rule. Thrown synchronously from inside the awaited
@@ -1569,7 +1569,7 @@ public sealed class ContinuousHistorizationRecorderTests : TestKit
}
}
private sealed record WrittenValue(string Tag, double Value, ushort Quality, DateTime? TimestampUtc);
private sealed record WrittenValue(string Tag, double Value, uint Quality, DateTime? TimestampUtc);
/// <summary>In-memory <see cref="IHistorizationOutbox"/> double honouring the FIFO-truncate
/// <see cref="IHistorizationOutbox.RemoveAsync"/> contract (remove the id plus any older entries
tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/ContinuousHistorizationRecorderTypedValueTests.cs
+253
View File
@@ -0,0 +1,253 @@
using Akka.Actor;
using Akka.TestKit.Xunit2;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
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;
/// <summary>
/// The recorder captures the TYPED value and the REAL status code the mux fans
/// (<see cref="VirtualTagActor.DependencyValueChanged.StatusCode"/>), and drops any kind the
/// backend does not support — metered as <see cref="ContinuousHistorizationRecorder.RecorderStatus.DroppedUnsupportedKind"/>.
/// </summary>
public sealed class ContinuousHistorizationRecorderTypedValueTests : TestKit
{
private const string Ref = "Pump1.Temp";
private const uint BadStatus = 0x80000000u;
private static readonly IReadOnlySet<HistorianValueKind> AllKinds = new HashSet<HistorianValueKind>
{
HistorianValueKind.Empty, HistorianValueKind.Double, HistorianValueKind.Int64,
HistorianValueKind.Boolean, HistorianValueKind.String, HistorianValueKind.DateTime,
};
private static readonly IReadOnlySet<HistorianValueKind> NumericKinds = new HashSet<HistorianValueKind>
{
HistorianValueKind.Double, HistorianValueKind.Int64, HistorianValueKind.Boolean,
};
[Fact]
public async Task A_bad_numeric_sample_is_written_with_its_real_status_code()
{
var writer = new PoolThreadValueWriter();
IActorRef rec = Spawn(writer, supportedKinds: null); // default kinds
rec.Tell(new VirtualTagActor.DependencyValueChanged(Ref, 42.5, DateTime.UtcNow, BadStatus));
HistorizationValue written = await AwaitSingleWriteAsync(writer);
written.StatusCode.ShouldBe(BadStatus);
written.Value.Kind.ShouldBe(HistorianValueKind.Double);
written.Value.Number.ShouldBe(42.5);
}
[Fact]
public async Task A_good_integer_sample_keeps_its_Int64_kind_and_Good_status()
{
var writer = new PoolThreadValueWriter();
IActorRef rec = Spawn(writer, supportedKinds: null);
rec.Tell(new VirtualTagActor.DependencyValueChanged(Ref, 7, DateTime.UtcNow));
HistorizationValue written = await AwaitSingleWriteAsync(writer);
written.StatusCode.ShouldBe(OpcUaStatus.Good);
written.Value.Kind.ShouldBe(HistorianValueKind.Int64);
written.Value.Integer.ShouldBe(7);
written.Value.Number.ShouldBe(7d); // the double view the gateway's numeric path writes
}
[Fact]
public async Task A_string_sample_is_written_when_the_backend_supports_String()
{
var writer = new PoolThreadValueWriter();
IActorRef rec = Spawn(writer, AllKinds);
rec.Tell(new VirtualTagActor.DependencyValueChanged(Ref, "running", DateTime.UtcNow));
HistorizationValue written = await AwaitSingleWriteAsync(writer);
written.Value.Kind.ShouldBe(HistorianValueKind.String);
written.Value.Text.ShouldBe("running");
(await StatusAsync(rec)).DroppedUnsupportedKind.ShouldBe(0);
}
[Fact]
public async Task A_string_sample_is_dropped_and_counted_when_the_backend_does_not_support_String()
{
var writer = new PoolThreadValueWriter();
IActorRef rec = Spawn(writer, NumericKinds);
rec.Tell(new VirtualTagActor.DependencyValueChanged(Ref, "running", DateTime.UtcNow));
await AwaitDroppedAsync(rec, 1);
await AssertNothingWrittenAfterPositiveControlAsync(rec, writer);
}
[Fact]
public async Task A_null_BadNoCommunication_sample_is_written_as_Empty_when_the_backend_supports_Empty()
{
var writer = new PoolThreadValueWriter();
IActorRef rec = Spawn(writer, AllKinds);
rec.Tell(new VirtualTagActor.DependencyValueChanged(Ref, null, DateTime.UtcNow, OpcUaStatus.BadNoCommunication));
HistorizationValue written = await AwaitSingleWriteAsync(writer);
written.Value.Kind.ShouldBe(HistorianValueKind.Empty);
written.StatusCode.ShouldBe(OpcUaStatus.BadNoCommunication);
}
[Fact]
public async Task A_null_BadNoCommunication_sample_is_dropped_when_the_backend_does_not_support_Empty()
{
var writer = new PoolThreadValueWriter();
IActorRef rec = Spawn(writer, NumericKinds);
rec.Tell(new VirtualTagActor.DependencyValueChanged(Ref, null, DateTime.UtcNow, OpcUaStatus.BadNoCommunication));
await AwaitDroppedAsync(rec, 1);
await AssertNothingWrittenAfterPositiveControlAsync(rec, writer);
}
[Fact]
public async Task An_array_sample_is_always_dropped_even_when_every_kind_is_supported()
{
var writer = new PoolThreadValueWriter();
IActorRef rec = Spawn(writer, AllKinds);
rec.Tell(new VirtualTagActor.DependencyValueChanged(Ref, new[] { 1, 2, 3 }, DateTime.UtcNow));
await AwaitDroppedAsync(rec, 1);
await AssertNothingWrittenAfterPositiveControlAsync(rec, writer);
}
private IActorRef Spawn(IHistorianValueWriter writer, IReadOnlySet<HistorianValueKind>? supportedKinds)
{
var mux = CreateTestProbe();
return Sys.ActorOf(ContinuousHistorizationRecorder.Props(
mux.Ref,
writer,
new SimpleOutbox(),
new[] { Ref },
drainInterval: TimeSpan.FromMilliseconds(50),
supportedKinds: supportedKinds));
}
private static async Task<ContinuousHistorizationRecorder.RecorderStatus> StatusAsync(IActorRef rec) =>
await rec.Ask<ContinuousHistorizationRecorder.RecorderStatus>(
ContinuousHistorizationRecorder.GetStatus.Instance, TimeSpan.FromSeconds(3));
private async Task<HistorizationValue> AwaitSingleWriteAsync(PoolThreadValueWriter writer)
{
await AwaitConditionAsync(() => Task.FromResult(writer.Snapshot().Count > 0), TimeSpan.FromSeconds(5));
IReadOnlyList<(string Tag, HistorizationValue Value)> written = writer.Snapshot();
written.Count.ShouldBe(1);
written[0].Tag.ShouldBe(Ref);
return written[0].Value;
}
private async Task AwaitDroppedAsync(IActorRef rec, long expected) =>
await AwaitAssertAsync(async () =>
{
ContinuousHistorizationRecorder.RecorderStatus status = await StatusAsync(rec);
status.DroppedUnsupportedKind.ShouldBe(expected);
status.QueuedDepth.ShouldBe(0);
});
/// <summary>An absence assertion needs a positive control: send a value that MUST be written and
/// prove it is the only one that was — so "nothing else written" is not vacuous.</summary>
private async Task AssertNothingWrittenAfterPositiveControlAsync(IActorRef rec, PoolThreadValueWriter writer)
{
rec.Tell(new VirtualTagActor.DependencyValueChanged(Ref, 1.0, DateTime.UtcNow));
HistorizationValue written = await AwaitSingleWriteAsync(writer);
written.Value.Kind.ShouldBe(HistorianValueKind.Double);
written.Value.Number.ShouldBe(1.0);
}
/// <summary>Records every value written; completes on a POOL thread, as a real gateway write does
/// (never synchronously, never via <c>Task.Yield</c> — Gitea #491).</summary>
private sealed class PoolThreadValueWriter : IHistorianValueWriter
{
private readonly Lock _gate = new();
private readonly List<(string Tag, HistorizationValue Value)> _written = new();
public IReadOnlyList<(string Tag, HistorizationValue Value)> Snapshot()
{
lock (_gate)
{
return _written.ToArray();
}
}
public async Task<bool> WriteLiveValuesAsync(
string tag, IReadOnlyList<HistorizationValue> values, CancellationToken ct)
{
await Task.Delay(1, ct).ConfigureAwait(false);
lock (_gate)
{
foreach (HistorizationValue v in values)
{
_written.Add((tag, v));
}
}
return true;
}
}
/// <summary>Minimal FIFO-truncate outbox; appends complete on a pool thread like a real fsync.</summary>
private sealed class SimpleOutbox : IHistorizationOutbox
{
private readonly Lock _gate = new();
private readonly List<HistorizationOutboxEntry> _entries = new();
public long DroppedCount => 0;
public async ValueTask AppendAsync(HistorizationOutboxEntry entry, CancellationToken ct) =>
await AppendBatchAsync(new[] { entry }, ct).ConfigureAwait(false);
public async ValueTask AppendBatchAsync(IReadOnlyList<HistorizationOutboxEntry> entries, CancellationToken ct)
{
await Task.Delay(1, ct).ConfigureAwait(false);
lock (_gate)
{
_entries.AddRange(entries);
}
}
public ValueTask<IReadOnlyList<HistorizationOutboxEntry>> PeekBatchAsync(int max, CancellationToken ct)
{
lock (_gate)
{
return ValueTask.FromResult<IReadOnlyList<HistorizationOutboxEntry>>(_entries.Take(max).ToArray());
}
}
public ValueTask RemoveAsync(Guid id, CancellationToken ct)
{
lock (_gate)
{
int idx = _entries.FindIndex(e => e.Id == id);
if (idx >= 0)
{
_entries.RemoveRange(0, idx + 1);
}
}
return ValueTask.CompletedTask;
}
public ValueTask<int> CountAsync(CancellationToken ct)
{
lock (_gate)
{
return ValueTask.FromResult(_entries.Count);
}
}
public void Dispose()
{
}
}
}