diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationMetrics.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationMetrics.cs index 8e0f752f..6f702e78 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationMetrics.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationMetrics.cs @@ -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 — completes synchronously (it /// just reads an in-memory FIFO count) and 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 GetStatus Ask for a health hook, but are /// not surfaced as gauges here (Ask-in-gauge is the awkward path the plan calls out). /// 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 0626b2b4..7b2667ea 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationRecorder.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationRecorder.cs @@ -11,25 +11,34 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.Historian; /// values the mux fans for those refs — in production a /// per driver burst, with the per-value /// receive retained for direct senders and tests: -/// value → coerce to numeric → append to the durable (the crash +/// value → coerce to a typed (kind-filtered) → append to the durable (the crash /// boundary) → a self-paced drain writes the outbox through 's /// live-value path and acks (truncates) on success. /// /// /// -/// Message path (grounded against the real actors). The mux fans each value — carrying -/// only (TagId, Value, TimestampUtc) and no quality — to every registered -/// subscriber, batched as a on the production -/// path and as a bare from a direct sender -/// or a test. -/// The recorder reuses that existing fan-out (no change to DriverHostActor), 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 -/// . +/// Message path (grounded against the real actors). The mux fans each value as +/// (TagId, Value, TimestampUtc, StatusCode) — forwards +/// the driver publish's own — to +/// every registered subscriber, batched as a on +/// the production path and as a bare from a +/// direct sender or a test. The recorder reuses that existing fan-out (no change to +/// DriverHostActor) and records each sample with its REAL status code, so a driver's own +/// Bad/Uncertain publish (usually a null value ⇒ ) is +/// historized as such when the backend can store it. /// /// -/// Numeric-only gate. 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. +/// What is NOT historized: the comms-loss watchdog's BadNoCommunication stamps. +/// DriverHostActor.FanCommsLossToValueNodes 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. +/// +/// +/// Kind gate. Each value is coerced with ; a +/// value it cannot carry (arrays, references, structures) or whose kind is not in the backend's +/// supported set (; default +/// ) is dropped and metered as +/// rather than appended. /// /// /// Batched ingest, non-blocking drain. A mux burst arrives as one @@ -45,16 +54,10 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.Historian; /// public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers { - /// 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). - /// - /// Note (archreview 06/C-7): the historian gateway's WriteLiveValues SQL path discards - /// quality (server-managed column, 0.2.0 Contracts C-002 — see - /// GatewayHistorianValueWriter's remarks). This constant is therefore what a future - /// quality-honoring gateway path would receive; capturing real per-node quality here will - /// NOT persist until the gateway honors the field. - /// - public const ushort GoodQuality = 192; + /// 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). + public static readonly IReadOnlySet DefaultSupportedKinds = + new HashSet { 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. - /// Lifetime count of values dropped for not being numeric-coercible. + /// Lifetime count of values dropped because they are not historizable + /// (arrays, references) or their kind is not one the backend supports. /// Lifetime count of entries the outbox dropped on capacity overflow. /// 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 /// 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 + /// threw — distinct from , which counts values the backend + /// cannot carry. Non-zero here means an ingest dependency is faulting (most /// plausibly the supplied capture gate), not that the data was unsuitable. 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. private readonly int _drainConcurrency; + /// The value kinds the backend can store; any other kind is dropped at capture. + private readonly IReadOnlySet _supportedKinds; + /// 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 /// , which keeps single-flight. @@ -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. /// Max historian tags one drain pass writes concurrently (must be /// positive; 1 is the fully sequential drain). + /// The value kinds the backend can store + /// (); null ⇒ . /// Props for the actor. public static Props Props( IActorRef dependencyMux, @@ -216,10 +225,11 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers TimeSpan? maxBackoff = null, Func? captureGate = null, int maxWriteAttempts = 5, - int drainConcurrency = 8) => + int drainConcurrency = 8, + IReadOnlySet? supportedKinds = null) => Akka.Actor.Props.Create(() => new ContinuousHistorizationRecorder( dependencyMux, writer, outbox, historizedRefs, drainBatchSize, drainInterval, minBackoff, maxBackoff, - captureGate, maxWriteAttempts, drainConcurrency)); + captureGate, maxWriteAttempts, drainConcurrency, supportedKinds)); /// Initializes a new instance of the class. /// The per-node dependency mux to register historized-ref interest with. @@ -237,6 +247,8 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers /// least 1. /// Max historian tags one drain pass writes concurrently. Clamped to /// at least 1; 1 is the fully sequential drain. + /// The value kinds the backend can store; null ⇒ + /// . public ContinuousHistorizationRecorder( IActorRef dependencyMux, IHistorianValueWriter writer, @@ -248,7 +260,8 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers TimeSpan? maxBackoff = null, Func? captureGate = null, int maxWriteAttempts = 5, - int drainConcurrency = 8) + int drainConcurrency = 8, + IReadOnlySet? 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(OnValueChangedAsync); ReceiveAsync(OnValueBatchAsync); @@ -337,8 +351,8 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers /// /// /// - /// 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 + /// The gates stay per value. 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 The_capture_gate_is_re_read_per_value_so_promotion_resumes_capture). /// 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, diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs index 516be128..94288e35 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs @@ -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()?.SupportedKinds), ContinuousHistorizationRecorderActorName); registry.Register(continuousRecorder); } 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 f5c0479d..10bce23b 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 @@ -13,14 +13,14 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Historian; /// Verifies the : it registers historized-ref /// interest with the dependency mux on start, appends mux-fanned /// 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). /// /// /// Adapted from the plan's Task 17: the recorder handles the REAL fan-out message the mux emits — -/// (TagId/Value/TimestampUtc, no quality) — -/// not DriverInstanceActor.AttributeValuePublished. The mux drops quality, so the recorder -/// records Good-quality (the same convention the scripted-alarm host uses for mux values). +/// (TagId/Value/TimestampUtc/StatusCode) — +/// not DriverInstanceActor.AttributeValuePublished. The typed-value / real-status behaviour is +/// pinned in . /// public sealed class ContinuousHistorizationRecorderTests : TestKit { @@ -334,7 +334,7 @@ public sealed class ContinuousHistorizationRecorderTests : TestKit { var status = await rec.Ask( 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); /// In-memory double honouring the FIFO-truncate /// contract (remove the id plus any older entries diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/ContinuousHistorizationRecorderTypedValueTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/ContinuousHistorizationRecorderTypedValueTests.cs new file mode 100644 index 00000000..ecca2015 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/ContinuousHistorizationRecorderTypedValueTests.cs @@ -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; + +/// +/// The recorder captures the TYPED value and the REAL status code the mux fans +/// (), and drops any kind the +/// backend does not support — metered as . +/// +public sealed class ContinuousHistorizationRecorderTypedValueTests : TestKit +{ + private const string Ref = "Pump1.Temp"; + private const uint BadStatus = 0x80000000u; + + private static readonly IReadOnlySet AllKinds = new HashSet + { + HistorianValueKind.Empty, HistorianValueKind.Double, HistorianValueKind.Int64, + HistorianValueKind.Boolean, HistorianValueKind.String, HistorianValueKind.DateTime, + }; + + private static readonly IReadOnlySet NumericKinds = new HashSet + { + 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? 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 StatusAsync(IActorRef rec) => + await rec.Ask( + ContinuousHistorizationRecorder.GetStatus.Instance, TimeSpan.FromSeconds(3)); + + private async Task 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); + }); + + /// 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. + 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); + } + + /// Records every value written; completes on a POOL thread, as a real gateway write does + /// (never synchronously, never via Task.Yield — Gitea #491). + 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 WriteLiveValuesAsync( + string tag, IReadOnlyList values, CancellationToken ct) + { + await Task.Delay(1, ct).ConfigureAwait(false); + lock (_gate) + { + foreach (HistorizationValue v in values) + { + _written.Add((tag, v)); + } + } + + return true; + } + } + + /// Minimal FIFO-truncate outbox; appends complete on a pool thread like a real fsync. + private sealed class SimpleOutbox : IHistorizationOutbox + { + private readonly Lock _gate = new(); + private readonly List _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 entries, CancellationToken ct) + { + await Task.Delay(1, ct).ConfigureAwait(false); + lock (_gate) + { + _entries.AddRange(entries); + } + } + + public ValueTask> PeekBatchAsync(int max, CancellationToken ct) + { + lock (_gate) + { + return ValueTask.FromResult>(_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 CountAsync(CancellationToken ct) + { + lock (_gate) + { + return ValueTask.FromResult(_entries.Count); + } + } + + public void Dispose() + { + } + } +}