Merge branch 'fix/r3-02-stale-good-family' — R3-02: one quality contract at four seams — comms loss marks raw+UNS value variables BadNoCommunication (WriteStatuses sink member, recovery audit), Calculation + VT hold on Bad / worst-of sub-code / (value,status) dedup, MQTT initial push + ConnectionLost stale, Sparkplug birth replay (archreview 03/S15, 05/STAB-19, 05/STAB-18, 02/S14)

This commit is contained in:
Joseph Doherty
2026-08-20 11:49:06 -04:00
46 changed files with 4340 additions and 129 deletions
@@ -0,0 +1,97 @@
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
namespace ZB.MOM.WW.OtOpcUa.Core.Abstractions.Tests;
/// <summary>
/// Truth table for <see cref="InputQuality"/> — the one worst-of-input rule every derived-value
/// seam uses (scripted alarms, virtual tags, the Calculation driver). The quality contract
/// (archreview R3-02) is "a value consumer never reports a quality better than its source's",
/// and this type is the single place that decides what "worse" means, so a divergence here is a
/// divergence in every seam at once.
/// </summary>
public sealed class InputQualityTests
{
private const uint Good = 0x00000000u;
private const uint Uncertain = 0x40000000u;
private const uint UncertainLastUsableValue = 0x40900000u;
private const uint BadNoCommunication = 0x80310000u;
private const uint BadInternalError = 0x80020000u;
private const uint BadWaitingForInitialData = 0x80320000u;
private static DataValueSnapshot Snap(object? value, uint status) =>
new(value, status, SourceTimestampUtc: DateTime.UtcNow, ServerTimestampUtc: DateTime.UtcNow);
/// <summary>An empty input set reports Good — there is nothing to be worse than.</summary>
[Fact]
public void WorstOf_no_inputs_is_Good()
{
InputQuality.WorstOf([]).ShouldBe(Good);
}
/// <summary>All-Good inputs report Good.</summary>
[Fact]
public void WorstOf_all_good_is_Good()
{
InputQuality.WorstOf([Snap(1, Good), Snap(2, Good)]).ShouldBe(Good);
}
/// <summary>A single Uncertain input carries its own sub-code out, not the generic Uncertain.</summary>
[Fact]
public void WorstOf_propagates_the_uncertain_sub_code()
{
InputQuality.WorstOf([Snap(1, Good), Snap(2, UncertainLastUsableValue)])
.ShouldBe(UncertainLastUsableValue);
}
/// <summary>Bad beats Uncertain regardless of the order the inputs are seen in.</summary>
[Theory]
[InlineData(true)]
[InlineData(false)]
public void WorstOf_bad_beats_uncertain(bool badFirst)
{
DataValueSnapshot?[] inputs = badFirst
? [Snap(null, BadNoCommunication), Snap(2, Uncertain)]
: [Snap(2, Uncertain), Snap(null, BadNoCommunication)];
InputQuality.WorstOf(inputs).ShouldBe(BadNoCommunication);
}
/// <summary>
/// A null entry means "this dependency has never been pushed", which is
/// <c>BadWaitingForInitialData</c> — the same code the materialiser stamps on a fresh node.
/// Reporting Good for it would be the exact laundering this helper exists to stop.
/// </summary>
[Fact]
public void WorstOf_a_null_entry_is_BadWaitingForInitialData()
{
InputQuality.WorstOf([Snap(1, Good), null]).ShouldBe(BadWaitingForInitialData);
}
/// <summary>
/// Among equal severity the FIRST seen wins, so a re-evaluation over the same inputs reports
/// the same sub-code rather than oscillating between two equally-bad codes.
/// </summary>
[Fact]
public void WorstOf_equal_severity_keeps_the_first_seen()
{
InputQuality.WorstOf([Snap(null, BadNoCommunication), Snap(null, BadInternalError)])
.ShouldBe(BadNoCommunication);
InputQuality.WorstOf([Snap(null, BadInternalError), Snap(null, BadNoCommunication)])
.ShouldBe(BadInternalError);
}
/// <summary>Good and Uncertain are usable (a script may still inspect the value); Bad is not.</summary>
[Theory]
[InlineData(Good, true)]
[InlineData(Uncertain, true)]
[InlineData(UncertainLastUsableValue, true)]
[InlineData(BadNoCommunication, false)]
[InlineData(BadWaitingForInitialData, false)]
[InlineData(0xC0000000u, false)]
public void IsUsable_follows_Part4_severity(uint statusCode, bool expected)
{
InputQuality.IsUsable(statusCode).ShouldBe(expected);
}
}
@@ -0,0 +1,316 @@
using System.Collections.Concurrent;
using Serilog;
using Serilog.Events;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Logging;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Core.Scripting;
namespace ZB.MOM.WW.OtOpcUa.Driver.Calculation.Tests;
/// <summary>
/// Archreview <b>05/STAB-19</b> — what the Calculation driver does with a dependency's <b>quality</b>.
///
/// <para><b>The defect.</b> The driver stored dependency <i>values</i> only and ignored the status code
/// entirely, so a comms-lost input arrived as <em>arrived, Good, null</em>: the warm-up gate was
/// satisfied by a value that never came, <c>(double)ctx.GetTag("x").Value</c> threw on every trigger
/// (a "script threw" storm), and a null-tolerant script published a <b>Good</b> output computed from a
/// dead input — indefinitely.</para>
///
/// <para><b>The policy pinned here</b> matches <c>ScriptedAlarmEngine</c>'s, which is the in-tree
/// precedent: <b>hold on Bad, propagate the worst input's status; evaluate through Uncertain and stamp
/// the output with it.</b> Holding (rather than evaluating and degrading) is what removes the NRE
/// storm; a Bad sub-code (rather than Uncertain) is what tells the operator the value may be
/// arbitrarily old rather than merely imprecise.</para>
///
/// <para>⚠️ <b>Vacuity rule:</b> the recovery case re-publishes an <b>identical</b> computed value.
/// A recovery whose value differs passes against a value-only dedup and proves nothing — it is the
/// "Bad forever" trap the v3 Batch 4 live gate found.</para>
/// </summary>
public sealed class CalculationDriverInputQualityTests
{
private const uint StatusGood = 0u;
private const uint StatusUncertain = 0x40000000u;
private const uint StatusBadNoCommunication = 0x80310000u;
private const uint StatusBadInternalError = 0x80020000u;
private const uint StatusBadWaitingForInitialData = 0x80320000u;
private const string SumTag = "calc/e/sum";
private const string SumScript = "return (double)ctx.GetTag(\"src/a\").Value + (double)ctx.GetTag(\"src/b\").Value;";
private static RawTagEntry CalcTag(string rawPath, string source)
{
var json = $"{{\"scriptId\":\"s-{rawPath}\",\"changeTriggered\":true," +
$"\"scriptSource\":{System.Text.Json.JsonSerializer.Serialize(source)}}}";
return new RawTagEntry(rawPath, json, WriteIdempotent: false);
}
private static (CalculationDriver Driver, ConcurrentQueue<DataChangeEventArgs> Published, List<ScriptLogEntry> Logs)
Build(params RawTagEntry[] tags)
{
var logs = new List<ScriptLogEntry>();
var sink = new ScriptLogTopicSink(new CapturingPublisher(logs), LogEventLevel.Information);
var scriptRoot = new ScriptRootLogger(new LoggerConfiguration().WriteTo.Sink(sink).CreateLogger());
var driver = new CalculationDriver(new CalculationDriverOptions { RawTags = tags }, "calc-drv", scriptRoot);
var published = new ConcurrentQueue<DataChangeEventArgs>();
driver.OnDataChange += (_, e) => published.Enqueue(e);
return (driver, published, logs);
}
/// <summary>Local copy of the script-log capture sink (the sibling suite's is private to it).</summary>
private sealed class CapturingPublisher(List<ScriptLogEntry> sink) : IScriptLogPublisher
{
private readonly object _lock = new();
public void Publish(ScriptLogEntry entry)
{
lock (_lock) sink.Add(entry);
}
}
private static async Task Settle(CalculationDriver driver)
{
var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(10);
while (DateTime.UtcNow < deadline)
{
if (driver.IsEvaluationIdle) return;
await Task.Delay(10);
}
driver.IsEvaluationIdle.ShouldBeTrue("the calc driver's evaluation pump should have drained");
}
/// <summary>Drives both deps Good and drains the resulting publish, leaving a settled Good output.</summary>
private static async Task<CalculationDriver> WarmedUpAsync(ConcurrentQueue<DataChangeEventArgs> published, CalculationDriver driver)
{
await driver.InitializeAsync("{}", TestContext.Current.CancellationToken);
driver.OnDependencyValue("src/a", 10.0, StatusGood, DateTime.UtcNow);
driver.OnDependencyValue("src/b", 20.0, StatusGood, DateTime.UtcNow);
await Settle(driver);
published.Count.ShouldBe(1);
published.TryDequeue(out var first).ShouldBeTrue();
first!.Snapshot.Value.ShouldBe(30.0);
first.Snapshot.StatusCode.ShouldBe(StatusGood);
return driver;
}
/// <summary>
/// A Bad input holds the script and publishes the input's own sub-code, once, carrying the last
/// known value. Three things are asserted together because each is a separate way the old
/// behaviour was wrong: no evaluation (no NRE storm), the sub-code propagated (not a generic
/// <c>BadInternalError</c>, which would blame the script), and no script-log entry (an input
/// dying is not a script fault).
/// </summary>
[Fact]
public async Task A_bad_input_holds_evaluation_and_publishes_the_inputs_own_sub_code_once()
{
var (driver, published, logs) = Build(CalcTag(SumTag, SumScript));
await WarmedUpAsync(published, driver);
driver.OnDependencyValue("src/a", null, StatusBadNoCommunication, DateTime.UtcNow);
await Settle(driver);
published.Count.ShouldBe(1);
published.TryDequeue(out var bad).ShouldBeTrue();
bad!.Snapshot.StatusCode.ShouldBe(StatusBadNoCommunication);
bad.Snapshot.StatusCode.ShouldNotBe(StatusBadInternalError, "the script did not fail — its input did");
bad.Snapshot.Value.ShouldBe(30.0, "the last known value rides out with the Bad status");
logs.ShouldBeEmpty("a dead input is not a script fault; the script log stays the per-failure diagnostic");
// Transition-once: a second Bad push of the same code changes nothing observable.
driver.OnDependencyValue("src/a", null, StatusBadNoCommunication, DateTime.UtcNow);
await Settle(driver);
published.ShouldBeEmpty();
}
/// <summary>
/// A different Bad sub-code IS republished: the sub-code is the diagnostic, and masking
/// <c>BadNoCommunication</c> behind an earlier <c>BadWaitingForInitialData</c> (or vice versa)
/// hides which input died and why.
/// </summary>
[Fact]
public async Task A_different_bad_sub_code_is_republished()
{
var (driver, published, _) = Build(CalcTag(SumTag, SumScript));
await WarmedUpAsync(published, driver);
driver.OnDependencyValue("src/a", null, StatusBadNoCommunication, DateTime.UtcNow);
await Settle(driver);
published.TryDequeue(out _).ShouldBeTrue();
driver.OnDependencyValue("src/a", null, StatusBadInternalError, DateTime.UtcNow);
await Settle(driver);
published.Count.ShouldBe(1);
published.TryDequeue(out var second).ShouldBeTrue();
second!.Snapshot.StatusCode.ShouldBe(StatusBadInternalError);
}
/// <summary>
/// <b>The vacuity-sensitive assertion.</b> The input recovers with the value it had before the
/// outage, so the computed output is <b>identical</b> to the pre-outage one — and it must still be
/// republished Good, because the node is currently Bad and only a publish can clear it. A
/// value-only dedup swallows exactly this case, which is why a recovery test with a changed value
/// is not the guard.
/// </summary>
[Fact]
public async Task An_identical_value_is_republished_when_recovering_from_a_bad_input()
{
var (driver, published, _) = Build(CalcTag(SumTag, SumScript));
await WarmedUpAsync(published, driver);
driver.OnDependencyValue("src/a", null, StatusBadNoCommunication, DateTime.UtcNow);
await Settle(driver);
published.TryDequeue(out _).ShouldBeTrue();
// The device comes back holding the same reading it had before.
driver.OnDependencyValue("src/a", 10.0, StatusGood, DateTime.UtcNow);
await Settle(driver);
published.Count.ShouldBe(1);
published.TryDequeue(out var recovered).ShouldBeTrue();
recovered!.Snapshot.Value.ShouldBe(30.0, "the same value as before the outage — that is the point");
recovered.Snapshot.StatusCode.ShouldBe(StatusGood);
}
/// <summary>
/// An Uncertain input is still computable, so the script runs and the output inherits the
/// Uncertain code — the alarm engine's rule (Uncertain is "ready"). Holding here would take a
/// perfectly good arithmetic result hostage.
/// </summary>
[Fact]
public async Task An_uncertain_input_is_evaluated_and_stamps_the_output_uncertain()
{
var (driver, published, _) = Build(CalcTag(SumTag, SumScript));
await WarmedUpAsync(published, driver);
driver.OnDependencyValue("src/a", 11.0, StatusUncertain, DateTime.UtcNow);
await Settle(driver);
published.Count.ShouldBe(1);
published.TryDequeue(out var uncertain).ShouldBeTrue();
uncertain!.Snapshot.Value.ShouldBe(31.0, "the script DID run — Uncertain is usable");
uncertain.Snapshot.StatusCode.ShouldBe(StatusUncertain);
// …and returning to Good republishes, even though the computed value changes back to a value the
// tag has held before.
driver.OnDependencyValue("src/a", 10.0, StatusGood, DateTime.UtcNow);
await Settle(driver);
published.TryDequeue(out var good).ShouldBeTrue();
good!.Snapshot.StatusCode.ShouldBe(StatusGood);
good.Snapshot.Value.ShouldBe(30.0);
}
/// <summary>
/// Cold start is unchanged: with one dep never pushed, a Bad arrival on the other publishes
/// nothing at all (the node stays at the materialiser's <c>BadWaitingForInitialData</c>). Only once
/// every dep has arrived does a Bad among them become reportable.
/// </summary>
[Fact]
public async Task Nothing_is_published_while_a_dep_has_never_arrived()
{
var (driver, published, _) = Build(CalcTag(SumTag, SumScript));
await driver.InitializeAsync("{}", TestContext.Current.CancellationToken);
driver.OnDependencyValue("src/a", null, StatusBadNoCommunication, DateTime.UtcNow);
await Settle(driver);
published.ShouldBeEmpty("src/b has never arrived — this is warm-up, not a fault");
driver.OnDependencyValue("src/b", 20.0, StatusGood, DateTime.UtcNow);
await Settle(driver);
published.Count.ShouldBe(1);
published.TryDequeue(out var bad).ShouldBeTrue();
bad!.Snapshot.StatusCode.ShouldBe(StatusBadNoCommunication);
}
/// <summary>
/// A passthrough script over a Bad input must not publish <c>Ok(null)</c> Good — the worst case of
/// the laundering, since the passthrough fast path bypasses the read cache entirely.
/// </summary>
[Fact]
public async Task A_passthrough_over_a_bad_input_is_bad_not_good_null()
{
var (driver, published, _) = Build(CalcTag("calc/e/pass", "return ctx.GetTag(\"src/a\").Value;"));
await driver.InitializeAsync("{}", TestContext.Current.CancellationToken);
driver.OnDependencyValue("src/a", 42.0, StatusGood, DateTime.UtcNow);
await Settle(driver);
published.TryDequeue(out var good).ShouldBeTrue();
good!.Snapshot.Value.ShouldBe(42.0);
driver.OnDependencyValue("src/a", null, StatusBadNoCommunication, DateTime.UtcNow);
await Settle(driver);
published.Count.ShouldBe(1);
published.TryDequeue(out var bad).ShouldBeTrue();
bad!.Snapshot.StatusCode.ShouldBe(StatusBadNoCommunication);
bad.Snapshot.StatusCode.ShouldNotBe(StatusGood);
}
/// <summary>
/// <c>ReadAsync</c> reports what the subscriber was told. A read that answered Good while the
/// subscription carried Bad would leave two clients of the same node disagreeing about it.
/// </summary>
[Fact]
public async Task ReadAsync_reports_the_same_status_the_subscriber_saw()
{
var (driver, published, _) = Build(CalcTag(SumTag, SumScript));
await WarmedUpAsync(published, driver);
driver.OnDependencyValue("src/a", null, StatusBadNoCommunication, DateTime.UtcNow);
await Settle(driver);
published.TryDequeue(out var bad).ShouldBeTrue();
var read = await driver.ReadAsync([SumTag], TestContext.Current.CancellationToken);
read[0].StatusCode.ShouldBe(bad!.Snapshot.StatusCode);
read[0].Value.ShouldBe(bad.Snapshot.Value);
}
/// <summary>
/// A tag whose ONLY publish was inputs-Bad — it never produced a value — must read back that Bad
/// status, not <c>BadWaitingForInitialData</c>. Guarding the read on "has a value" rather than
/// "has published" made a polled read say "never polled" while the subscriber on the same node had
/// been told <c>BadNoCommunication</c>: the two clients disagree, and the read is the one that
/// lies (something HAS been observed about this tag — that its input is dead).
/// </summary>
[Fact]
public async Task ReadAsync_after_an_inputs_Bad_with_no_prior_value_reports_the_bad_status()
{
var (driver, published, _) = Build(CalcTag(SumTag, SumScript));
await driver.InitializeAsync("{}", TestContext.Current.CancellationToken);
// The first thing that ever happens to this tag is a dead input — no successful evaluation, so
// there is no last value to carry.
driver.OnDependencyValue("src/a", null, StatusBadNoCommunication, DateTime.UtcNow);
driver.OnDependencyValue("src/b", 20.0, StatusGood, DateTime.UtcNow);
await Settle(driver);
published.Count.ShouldBe(1);
published.TryDequeue(out var bad).ShouldBeTrue();
bad!.Snapshot.StatusCode.ShouldBe(StatusBadNoCommunication);
bad.Snapshot.Value.ShouldBeNull();
var read = await driver.ReadAsync([SumTag], TestContext.Current.CancellationToken);
read[0].StatusCode.ShouldBe(StatusBadNoCommunication);
read[0].StatusCode.ShouldNotBe(StatusBadWaitingForInitialData,
"something HAS been observed about this tag — that its input is dead");
read[0].Value.ShouldBeNull();
}
/// <summary>A never-evaluated tag still reads BadWaitingForInitialData (regression).</summary>
[Fact]
public async Task An_unevaluated_tag_still_reads_BadWaitingForInitialData()
{
var (driver, _, _) = Build(CalcTag(SumTag, SumScript));
await driver.InitializeAsync("{}", TestContext.Current.CancellationToken);
var read = await driver.ReadAsync([SumTag], TestContext.Current.CancellationToken);
read[0].StatusCode.ShouldBe(StatusBadWaitingForInitialData);
}
}
@@ -4,6 +4,7 @@ using Microsoft.Extensions.Logging.Abstractions;
using Serilog;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Core.Scripting;
namespace ZB.MOM.WW.OtOpcUa.Driver.Calculation.Tests;
@@ -230,6 +231,74 @@ public sealed class CalculationEvaluatorTests
return (int)countProp.GetValue(cache)!;
}
// ── input quality reaches the script (archreview 05/STAB-19) ────────────────────────────
/// <summary>
/// A script reading <c>ctx.GetTag(path).StatusCode</c> sees the input's <b>real</b> status.
/// The read cache used to stamp every dependency Good, so a script that checked quality was lied
/// to — and, since the driver now holds on Bad, an Uncertain input is precisely the case where a
/// script can still make a qualified decision and needs the truth to do it.
/// </summary>
[Fact]
public void The_read_cache_carries_the_inputs_real_status_code()
{
const uint uncertain = 0x40000000u;
using var sut = new CalculationEvaluator(NullLogger<CalculationEvaluator>.Instance, NoOpScriptRoot());
var sourceStamp = new DateTime(2026, 8, 20, 9, 0, 0, DateTimeKind.Utc);
var result = sut.Evaluate(
calcTagId: "calc-quality",
scriptSource: "return (long)ctx.GetTag(\"a\").StatusCode;",
dependencies: new Dictionary<string, DataValueSnapshot>
{
["a"] = new(7.0, uncertain, sourceStamp, sourceStamp),
});
result.Success.ShouldBeTrue(result.Reason);
result.Value.ShouldBe((long)uncertain);
}
/// <summary>
/// The dependency's own source timestamp reaches the script too. The old cache stamped
/// evaluation time on every input, so a script comparing input ages compared its own clock with
/// itself.
/// </summary>
[Fact]
public void The_read_cache_carries_the_inputs_real_source_timestamp()
{
using var sut = new CalculationEvaluator(NullLogger<CalculationEvaluator>.Instance, NoOpScriptRoot());
var sourceStamp = new DateTime(2026, 8, 20, 9, 0, 0, DateTimeKind.Utc);
var result = sut.Evaluate(
calcTagId: "calc-stamp",
scriptSource: "return ctx.GetTag(\"a\").SourceTimestampUtc;",
dependencies: new Dictionary<string, DataValueSnapshot>
{
["a"] = new(7.0, 0u, sourceStamp, DateTime.UtcNow),
});
result.Success.ShouldBeTrue(result.Reason);
result.Value.ShouldBe(sourceStamp);
}
/// <summary>The passthrough fast path takes the same snapshot input and returns its value.</summary>
[Fact]
public void The_passthrough_fast_path_accepts_snapshots()
{
using var sut = new CalculationEvaluator(NullLogger<CalculationEvaluator>.Instance, NoOpScriptRoot());
var result = sut.Evaluate(
calcTagId: "calc-mirror",
scriptSource: "return ctx.GetTag(\"a\").Value;",
dependencies: new Dictionary<string, DataValueSnapshot>
{
["a"] = new(42, 0u, DateTime.UtcNow, DateTime.UtcNow),
});
result.Success.ShouldBeTrue(result.Reason);
result.Value.ShouldBe(42);
}
/// <summary>Minimal <see cref="ILogger{T}"/> that records emitted entries so the SetVirtualTag-drop
/// test can assert the Debug "dropped" line was logged.</summary>
private sealed class CapturingLogger : ILogger<CalculationEvaluator>
@@ -615,6 +615,81 @@ public sealed class MqttConnectionTests
blackhole.AcceptedCount.ShouldBe(acceptedAfterTheAttempt);
}
// ---------------------------------------------------------------------------------
// ConnectionLost — the quality edge (archreview 05/STAB-18 shape a)
// ---------------------------------------------------------------------------------
/// <summary>
/// <see cref="MqttConnection.ConnectionLost"/> fires exactly once per
/// <c>Connected → Reconnecting</c> edge. It is the signal the ingest paths stale their cached
/// values on, so an extra fire is downstream churn and a missing one leaves every tag
/// reporting Good over a dead session.
/// </summary>
[Fact]
public async Task BrokerDropsTheConnection_FiresConnectionLostOncePerEdge()
{
using var broker = new MiniBroker();
await using var conn = new MqttConnection(BrokerOpts(broker.Port), driverId: "t", logger: null);
var losses = 0;
conn.ConnectionLost += () => Interlocked.Increment(ref losses);
await conn.ConnectAsync(CancellationToken.None);
Volatile.Read(ref losses).ShouldBe(0, "a successful connect is not a loss");
broker.DropAll();
(await WaitUntilAsync(() => Volatile.Read(ref losses) >= 1, TimeSpan.FromSeconds(30))).ShouldBeTrue();
(await WaitUntilAsync(() => conn.State == MqttConnectionState.Connected, TimeSpan.FromSeconds(30)))
.ShouldBeTrue();
Volatile.Read(ref losses).ShouldBe(1, "the reconnect itself is not a second loss");
broker.DropAll();
(await WaitUntilAsync(() => Volatile.Read(ref losses) >= 2, TimeSpan.FromSeconds(30)))
.ShouldBeTrue("a second outage is a second edge");
}
/// <summary>
/// An initial connect that never succeeds is not a <i>loss</i> — there was no session to lose,
/// and staling on it would flip a never-observed tag Bad on a shape the cache already reports
/// as <c>BadWaitingForInitialData</c>.
/// </summary>
[Fact]
public async Task ConnectionLost_DoesNotFireOnAnInitialConnectFailure()
{
using var blackhole = new BlackholeBroker();
var opts = new MqttDriverOptions
{
Host = "127.0.0.1", Port = blackhole.Port, UseTls = false, ConnectTimeoutSeconds = 1,
};
await using var conn = new MqttConnection(opts, driverId: "t", logger: null);
var losses = 0;
conn.ConnectionLost += () => Interlocked.Increment(ref losses);
await Should.ThrowAsync<TimeoutException>(async () => await conn.ConnectAsync(CancellationToken.None));
await Task.Delay(TimeSpan.FromMilliseconds(500), TestContext.Current.CancellationToken);
Volatile.Read(ref losses).ShouldBe(0);
}
/// <summary>
/// Dispose tears the client down, which the library reports as a disconnect. That is an
/// intentional teardown, not an outage: firing here would stale every tag on an ordinary
/// redeploy or shutdown.
/// </summary>
[Fact]
public async Task ConnectionLost_DoesNotFireOnDispose()
{
using var broker = new MiniBroker();
var conn = new MqttConnection(BrokerOpts(broker.Port), driverId: "t", logger: null);
var losses = 0;
conn.ConnectionLost += () => Interlocked.Increment(ref losses);
await conn.ConnectAsync(CancellationToken.None);
await conn.DisposeAsync();
await Task.Delay(TimeSpan.FromMilliseconds(500), TestContext.Current.CancellationToken);
Volatile.Read(ref losses).ShouldBe(0);
}
// ---------------------------------------------------------------------------------
// Task 4 — the reconnect supervisor, against a real (minimal) broker
// ---------------------------------------------------------------------------------
@@ -100,6 +100,79 @@ public sealed class MqttDriverSessionLifecycleTests
"the authored 2s budget must still bound the connect, not the 15s default");
}
/// <summary>
/// <b>End-to-end through <see cref="MqttDriver"/> (archreview 05/STAB-18).</b> Deliver a value,
/// let the real broker drop the real session, and the driver publishes
/// <c>BadNoCommunication</c> with the value retained; feed the <b>identical</b> payload back —
/// the retained replay a broker sends on reconnect — and it publishes Good again.
/// </summary>
/// <remarks>
/// <para>
/// <b>What this adds over <c>MqttSubscriptionManagerRecoveryTests</c>:</b> the wiring. Those
/// drive <c>OnConnectionLost()</c> directly, so they stay green even if
/// <c>MqttSubscriptionManager.AttachTo</c> never subscribes to
/// <see cref="MqttConnection.ConnectionLost"/> — an unforwarded seam is this repo's
/// best-documented way to ship an inert fix. Here the edge comes from a genuine socket drop
/// through MQTTnet's own <c>DisconnectedAsync</c>.
/// </para>
/// <para>
/// <b>Vacuity rule:</b> the recovery payload is byte-identical to the pre-outage one. A test
/// that replays a <i>different</i> value passes against a build with no stale at all — the
/// value change alone would defeat the suppression — and is not the guard.
/// </para>
/// <para>
/// Ingest is fed through <c>Subscriptions.HandleMessage</c>, the method MQTTnet's dispatcher
/// calls, because <see cref="MiniBroker"/> answers CONNECT and PINGREQ and nothing else: it
/// can drop a session (which is what this test needs from it) but cannot PUBLISH.
/// </para>
/// </remarks>
[Fact]
public async Task ABrokerDrop_MarksCachedRefsBadNoCommunication_AndTheIdenticalReplayRecovers()
{
const string rawPath = "Plant/Mqtt/dev1/Temp";
const string topic = "factory/oven/temp";
const uint good = 0x00000000u;
const uint badNoCommunication = 0x80310000u;
using var broker = new MiniBroker();
var options = BrokerOptions(broker.Port) with
{
RawTags =
[
new RawTagEntry(
rawPath,
$$"""{"topic":"{{topic}}","payloadFormat":"Json","jsonPath":"$.value","dataType":"Float64"}""",
WriteIdempotent: false),
],
};
await using var driver = new MqttDriver(options, "d", null);
await driver.InitializeAsync(Json(options), TestContext.Current.CancellationToken);
await driver.SubscribeAsync([rawPath], TimeSpan.Zero, TestContext.Current.CancellationToken);
var raised = new List<DataValueSnapshot>();
driver.OnDataChange += (_, e) => { lock (raised) raised.Add(e.Snapshot); };
driver.Subscriptions.HandleMessage(topic, """{"value":21.5}"""u8.ToArray(), retained: false);
lock (raised) raised.Count.ShouldBe(1, "the first value must publish");
broker.DropAll();
await WaitForAsync(
() => { lock (raised) return raised.Count == 2; },
"the broker drop to publish a BadNoCommunication snapshot");
DataValueSnapshot staled;
lock (raised) staled = raised[1];
staled.StatusCode.ShouldBe(badNoCommunication);
staled.Value.ShouldBe(21.5, "the last known value is retained through the outage");
// The reconnect's retained replay — the IDENTICAL payload.
driver.Subscriptions.HandleMessage(topic, """{"value":21.5}"""u8.ToArray(), retained: true);
DataValueSnapshot recovered;
lock (raised) { raised.Count.ShouldBe(3); recovered = raised[2]; }
recovered.StatusCode.ShouldBe(good);
recovered.Value.ShouldBe(21.5);
}
private static async Task WaitForAsync(Func<bool> condition, string what)
{
var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(10);
@@ -0,0 +1,229 @@
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests;
/// <summary>
/// The push-model edge behaviours <see cref="MqttSubscriptionManager"/> lacked (archreview
/// 05/STAB-18): an <b>initial push</b> of the cached snapshot on every (re)subscribe, and a
/// <b>connection-loss stale</b> to <c>BadNoCommunication</c> with the value retained.
///
/// <para><b>Why both, and why the identical payload is load-bearing.</b> The manager's steady-state
/// dedup suppresses a message whose value AND status equal the cached one, and that cache is
/// driver-owned — it survives a reconnect and a redeploy. So a broker that replays its retained
/// message unchanged after an outage raised nothing, and every node the host had latched Bad stayed
/// Bad until the plant value happened to change. A recovery test that replays a <em>different</em>
/// payload passes against the pre-fix code and proves nothing; every recovery assertion here
/// therefore replays the <b>identical</b> payload.</para>
///
/// <para>The steady-state suppression itself must survive — it is what keeps a 10 Hz topic from
/// re-triggering the whole downstream fan-out per message — so each recovery case is paired with a
/// control that asserts an unchanged republish is still suppressed.</para>
/// </summary>
public sealed class MqttSubscriptionManagerRecoveryTests
{
// OPC UA status codes, verified against Opc.Ua.StatusCodes rather than recalled.
private const uint Good = 0x00000000u;
private const uint BadNoCommunication = 0x80310000u;
private const uint BadWaitingForInitialData = 0x80320000u;
private const string TempPath = "Plant/Mqtt/broker1/OvenTemp";
private const string PressPath = "Plant/Mqtt/broker1/OvenPressure";
private const string TempTopic = "f/oven/temp";
private const string PressTopic = "f/oven/press";
private static RawTagEntry Tag(string rawPath, string topic) =>
new(rawPath, $$"""{"topic":"{{topic}}","payloadFormat":"Json","jsonPath":"$.value","dataType":"Float64"}""",
WriteIdempotent: false);
private static MqttSubscriptionManager NewManager()
{
var mgr = new MqttSubscriptionManager(new MqttDriverOptions { Mode = MqttMode.Plain }, "mqtt-1");
mgr.Register([Tag(TempPath, TempTopic), Tag(PressPath, PressTopic)]);
return mgr;
}
private sealed record Raised(string Reference, object? Value, uint StatusCode, ISubscriptionHandle Handle);
private static List<Raised> Record(MqttSubscriptionManager mgr)
{
var raised = new List<Raised>();
mgr.OnDataChange += (_, e) =>
raised.Add(new Raised(e.FullReference, e.Snapshot.Value, e.Snapshot.StatusCode, e.SubscriptionHandle));
return raised;
}
// ---------------------------------------------------------------------------------
// (A) Initial push on (re)subscribe — 05/STAB-18 shape (b)
// ---------------------------------------------------------------------------------
/// <summary>
/// A ref observed before the subscription existed is pushed once, with its cached snapshot,
/// attributed to the new handle. This is the push-model equivalent of the poll engine's
/// <c>forceRaise: true</c> on the first poll of a new subscription, and it is what refills a
/// node the address-space rebuild reseeded <c>BadWaitingForInitialData</c> on a quiet topic.
/// </summary>
[Fact]
public async Task SubscribeAsync_pushes_the_cached_snapshot_for_an_observed_ref()
{
var mgr = NewManager();
mgr.HandleMessage(TempTopic, """{"value":21.5}"""u8.ToArray(), retained: false); // cached; no handle yet
var raised = Record(mgr);
var handle = await mgr.SubscribeAsync([TempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken);
raised.Count.ShouldBe(1);
raised[0].Reference.ShouldBe(TempPath);
raised[0].Value.ShouldBe(21.5);
raised[0].StatusCode.ShouldBe(Good);
raised[0].Handle.ShouldBe(handle);
}
/// <summary>
/// A ref that has never been observed carries only the materialiser's
/// <c>BadWaitingForInitialData</c> placeholder — pushing that would claim an observation the
/// driver has not made, so the initial push skips it.
/// </summary>
[Fact]
public async Task SubscribeAsync_does_not_push_a_never_observed_ref()
{
var mgr = NewManager();
mgr.HandleMessage(TempTopic, """{"value":21.5}"""u8.ToArray(), retained: false);
var raised = Record(mgr);
await mgr.SubscribeAsync([TempPath, PressPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken);
raised.Select(r => r.Reference).ShouldBe([TempPath]);
mgr.Values.Read(PressPath).StatusCode.ShouldBe(BadWaitingForInitialData);
}
/// <summary>
/// <b>Control.</b> The initial push must not weaken steady-state suppression: an identical
/// message arriving after the subscribe is still swallowed. Without this, "recovery works"
/// could simply mean "nothing is ever deduped any more".
/// </summary>
[Fact]
public async Task An_identical_message_after_the_initial_push_is_still_suppressed()
{
var mgr = NewManager();
mgr.HandleMessage(TempTopic, """{"value":21.5}"""u8.ToArray(), retained: false);
var raised = Record(mgr);
await mgr.SubscribeAsync([TempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken);
mgr.HandleMessage(TempTopic, """{"value":21.5}"""u8.ToArray(), retained: false);
raised.Count.ShouldBe(1); // the initial push only
}
/// <summary>
/// The rebuild shape: unsubscribe and subscribe again — as every deploy does through
/// <c>DriverInstanceActor</c>'s unsubscribe-then-subscribe — and the cached value is pushed
/// again onto the new handle rather than leaving the rebuilt node waiting for the plant.
/// </summary>
[Fact]
public async Task A_resubscribe_pushes_again()
{
var mgr = NewManager();
mgr.HandleMessage(TempTopic, """{"value":21.5}"""u8.ToArray(), retained: false);
var first = await mgr.SubscribeAsync([TempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken);
await mgr.UnsubscribeAsync(first, TestContext.Current.CancellationToken);
var raised = Record(mgr);
var second = await mgr.SubscribeAsync([TempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken);
raised.Count.ShouldBe(1);
raised[0].Value.ShouldBe(21.5);
raised[0].Handle.ShouldBe(second);
}
// ---------------------------------------------------------------------------------
// (B) Connection-loss stale — 05/STAB-18 shape (a)
// ---------------------------------------------------------------------------------
/// <summary>
/// A lost broker session makes every observed reference <c>BadNoCommunication</c> — with its
/// last value retained, the Part 8 last-known-value convention a SCADA client expects. MQTT
/// previously marked nothing at all on its own disconnect edge, so the tags stayed Good until
/// the host's 90 s comms-loss watchdog noticed (if it noticed at all — MQTTnet usually
/// reconnects well inside that window).
/// </summary>
[Fact]
public async Task OnConnectionLost_stales_every_observed_ref_keeping_its_value()
{
var mgr = NewManager();
mgr.HandleMessage(TempTopic, """{"value":21.5}"""u8.ToArray(), retained: false);
await mgr.SubscribeAsync([TempPath, PressPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken);
var raised = Record(mgr); // recorded AFTER the subscribe, so the initial push is not counted
mgr.OnConnectionLost();
raised.Count.ShouldBe(1, "the never-observed ref has nothing to stale");
raised[0].Reference.ShouldBe(TempPath);
raised[0].StatusCode.ShouldBe(BadNoCommunication);
raised[0].Value.ShouldBe(21.5, "the last known value is retained, not nulled");
mgr.Values.Read(TempPath).StatusCode.ShouldBe(BadNoCommunication);
mgr.Values.Read(TempPath).Value.ShouldBe(21.5);
}
/// <summary>
/// <b>The contract test for this seam.</b> After the stale, the broker's retained replay of the
/// <b>identical</b> payload is raised Good — because the cached status now differs even though
/// the value does not. Replaying a <em>different</em> payload here would pass against the
/// pre-fix build and prove nothing; the identical payload is the whole point.
/// </summary>
[Fact]
public async Task After_the_stale_the_identical_retained_replay_is_raised_Good()
{
var mgr = NewManager();
mgr.HandleMessage(TempTopic, """{"value":21.5}"""u8.ToArray(), retained: false);
await mgr.SubscribeAsync([TempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken);
mgr.OnConnectionLost();
var raised = Record(mgr);
mgr.HandleMessage(TempTopic, """{"value":21.5}"""u8.ToArray(), retained: true); // the retained seed
raised.Count.ShouldBe(1);
raised[0].Reference.ShouldBe(TempPath);
raised[0].Value.ShouldBe(21.5);
raised[0].StatusCode.ShouldBe(Good);
}
/// <summary>
/// A second loss with no intervening message raises nothing: the ref is already Bad, and a
/// re-stamp would be pure churn on the downstream fan-out.
/// </summary>
[Fact]
public async Task OnConnectionLost_does_not_re_raise_an_already_bad_ref()
{
var mgr = NewManager();
mgr.HandleMessage(TempTopic, """{"value":21.5}"""u8.ToArray(), retained: false);
await mgr.SubscribeAsync([TempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken);
mgr.OnConnectionLost();
var raised = Record(mgr);
mgr.OnConnectionLost();
raised.ShouldBeEmpty();
}
/// <summary>
/// The stale runs on MQTTnet's dispatcher thread, so it must complete synchronously and never
/// await: parking that thread stalls delivery for every other subscription on the connection.
/// </summary>
[Fact]
public async Task OnConnectionLost_completes_on_the_calling_thread()
{
var mgr = NewManager();
mgr.HandleMessage(TempTopic, """{"value":21.5}"""u8.ToArray(), retained: false);
await mgr.SubscribeAsync([TempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken);
var callingThread = Environment.CurrentManagedThreadId;
var raisedOn = 0;
mgr.OnDataChange += (_, _) => raisedOn = Environment.CurrentManagedThreadId;
mgr.OnConnectionLost();
raisedOn.ShouldBe(callingThread);
}
}
@@ -0,0 +1,268 @@
using Google.Protobuf;
using Org.Eclipse.Tahu.Protobuf;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Sparkplug;
using TahuDataType = Org.Eclipse.Tahu.Protobuf.DataType;
namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests;
/// <summary>
/// The two push-model edge behaviours <see cref="SparkplugIngestor"/> lacked (archreview
/// 05/STAB-18), matching <see cref="MqttSubscriptionManagerRecoveryTests"/> on the plain path: an
/// <b>initial push</b> of the cached snapshot on every (re)subscribe, and a
/// <b>connection-loss stale</b> to <c>BadNoCommunication</c> with the value retained.
///
/// <para><b>Where the falsifier actually is on this seam — measured, not assumed.</b> Recovery is
/// proved with an <b>identical</b> DBIRTH value, as on the plain path. But Sparkplug ingestion
/// carries <em>no</em> unchanged-value suppression (it publishes births and data directly rather
/// than through <c>MqttSubscriptionManager.Deliver</c>), so that replay was already raised Good
/// before this change — pinned below by
/// <see cref="An_identical_birth_replay_is_raised_even_without_a_preceding_stale"/>. The
/// load-bearing assertion here is therefore the <b>stale</b>: a Sparkplug tag now reports Bad on
/// the driver's own disconnect edge instead of Good over a dead broker.</para>
///
/// <para><b>A death is not a connection loss.</b> DDEATH/NDEATH is the source announcing it is
/// offline and keeps its own <c>BadCommunicationError</c> stale, unchanged; <c>ConnectionLost</c>
/// is this driver's transport losing the broker and reports <c>BadNoCommunication</c> — the same
/// code the host's comms-loss fan-out writes. The DDEATH regression case below pins that the two
/// paths stayed distinct.</para>
/// </summary>
public sealed class SparkplugIngestorRecoveryTests
{
// OPC UA status codes, verified against Opc.Ua.StatusCodes rather than recalled.
private const uint Good = 0x00000000u;
private const uint BadCommunicationError = 0x80050000u;
private const uint BadNoCommunication = 0x80310000u;
private const uint BadWaitingForInitialData = 0x80320000u;
private const string Group = "Plant1";
private const string Node = "EdgeA";
private const string Device = "Filler1";
private const string TempPath = "Plant/Mqtt/spb/Filler1Temp";
private const string PressPath = "Plant/Mqtt/spb/Filler1Press";
private const ulong TempAlias = 5UL;
private const ulong PressAlias = 6UL;
private sealed record Raised(string Reference, object? Value, uint StatusCode);
private static List<Raised> Record(SparkplugIngestor ing)
{
var raised = new List<Raised>();
ing.OnDataChange += (_, e) => raised.Add(new Raised(e.FullReference, e.Snapshot.Value, e.Snapshot.StatusCode));
return raised;
}
// ---------------------------------------------------------------------------------
// (A) Initial push on (re)subscribe
// ---------------------------------------------------------------------------------
/// <summary>
/// A metric observed before this subscription existed is pushed once from the cache. This is the
/// address-space-rebuild shape: a rebuild reseeds every node <c>BadWaitingForInitialData</c> and
/// triggers no rebirth, so without the push a quiet metric never refills.
/// </summary>
[Fact]
public async Task SubscribeAsync_pushes_the_cached_snapshot_for_an_observed_metric()
{
var ing = await NewSubscribedIngestorAsync();
FeedBirths(ing);
ing.Dispatch(Topic(SparkplugMessageType.DDATA, Device), Ddata(seq: 2, (TempAlias, TahuDataType.Float, 7f)));
var raised = Record(ing);
await ing.SubscribeAsync([TempPath], TimeSpan.Zero, TestContext.Current.CancellationToken);
raised.Count.ShouldBe(1);
raised[0].Reference.ShouldBe(TempPath);
raised[0].Value.ShouldBe(7f);
raised[0].StatusCode.ShouldBe(Good);
}
/// <summary>
/// A metric that has never been observed carries only the cache's
/// <c>BadWaitingForInitialData</c> placeholder, and pushing that would claim an observation the
/// driver has not made.
/// </summary>
[Fact]
public async Task SubscribeAsync_does_not_push_a_never_observed_metric()
{
var ing = await NewSubscribedIngestorAsync();
var raised = Record(ing);
await ing.SubscribeAsync([TempPath, PressPath], TimeSpan.Zero, TestContext.Current.CancellationToken);
raised.ShouldBeEmpty();
ing.Values.Read(TempPath).StatusCode.ShouldBe(BadWaitingForInitialData);
}
// ---------------------------------------------------------------------------------
// (B) Connection-loss stale + the identical-value recovery
// ---------------------------------------------------------------------------------
/// <summary>
/// A lost broker session marks every observed metric <c>BadNoCommunication</c> with its last
/// value retained. <c>ResetSessionState</c>'s previous remark argued the opposite — that the
/// connection's health surface already reports the outage — but health is not quality, and an
/// OPC UA client subscribed to the metric's node sees neither the health page nor the outage.
/// </summary>
[Fact]
public async Task OnConnectionLost_stales_every_observed_metric_keeping_its_value()
{
var ing = await NewSubscribedIngestorAsync();
FeedBirths(ing);
ing.Dispatch(Topic(SparkplugMessageType.DDATA, Device), Ddata(seq: 2, (TempAlias, TahuDataType.Float, 7f)));
var raised = Record(ing);
ing.OnConnectionLost();
raised.ShouldContain(r => r.Reference == TempPath && r.StatusCode == BadNoCommunication && Equals(r.Value, 7f));
ing.Values.Read(TempPath).StatusCode.ShouldBe(BadNoCommunication);
ing.Values.Read(TempPath).Value.ShouldBe(7f);
}
/// <summary>
/// <b>The contract test for this seam.</b> The reconnect replays NBIRTH/DBIRTH carrying the
/// <b>identical</b> metric value, and it must be raised Good. Before the stale, the birth's
/// value equalled the cached one and the whole tag stayed Bad until the plant moved.
/// </summary>
[Fact]
public async Task After_the_stale_an_identical_birth_replay_is_raised_Good()
{
var ing = await NewSubscribedIngestorAsync();
FeedBirths(ing, tempBirthValue: 7f);
ing.OnConnectionLost();
ing.Values.Read(TempPath).StatusCode.ShouldBe(BadNoCommunication);
var raised = Record(ing);
FeedBirths(ing, tempBirthValue: 7f, bdSeq: 2); // the reconnect's replayed births, value unchanged
raised.ShouldContain(r => r.Reference == TempPath && r.StatusCode == Good && Equals(r.Value, 7f));
}
/// <summary>
/// <b>Vacuity control, and it changes what the test above proves.</b> Unlike the plain path,
/// Sparkplug ingestion has no unchanged-value suppression at all — a birth republishes its
/// metrics regardless — so an identical birth replay would have been raised Good even before
/// the stale existed. This asserts that directly. The consequence: on THIS seam the
/// load-bearing assertion is the stale itself (a Sparkplug tag now reports Bad on the driver's
/// own disconnect edge instead of Good over a dead broker), and the replay test guards only
/// that the stale is not sticky.
/// </summary>
[Fact]
public async Task An_identical_birth_replay_is_raised_even_without_a_preceding_stale()
{
var ing = await NewSubscribedIngestorAsync();
FeedBirths(ing, tempBirthValue: 7f);
var raised = Record(ing);
FeedBirths(ing, tempBirthValue: 7f, bdSeq: 2);
raised.ShouldContain(r => r.Reference == TempPath && r.StatusCode == Good && Equals(r.Value, 7f));
}
/// <summary>
/// Regression: a DDEATH still stales with <c>BadCommunicationError</c>, the source-announced
/// code. The connection-loss stale is a second, distinct path — it did not replace this one.
/// </summary>
[Fact]
public async Task DDeath_still_stales_with_BadCommunicationError()
{
var ing = await NewSubscribedIngestorAsync();
FeedBirths(ing, tempBirthValue: 7f);
var raised = Record(ing);
ing.Dispatch(Topic(SparkplugMessageType.DDEATH, Device), Ddeath(seq: 2));
raised.ShouldContain(r => r.Reference == TempPath && r.StatusCode == BadCommunicationError);
ing.Values.Read(TempPath).StatusCode.ShouldBe(BadCommunicationError);
}
// =================================================================================
// Fixtures — deliberately a local copy of SparkplugIngestorTests' helpers (both are
// private to their own class; sharing them would make one suite's edits ripple into
// the other's fixtures).
// =================================================================================
private static RawTagEntry Tag(string rawPath, string tagConfig) => new(rawPath, tagConfig, WriteIdempotent: false);
private static string Blob(string group, string node, string? device, string metric, string dataType)
{
var deviceKey = device is null ? "" : $"""{'"'}deviceId{'"'}:"{device}",""";
return $$"""{"groupId":"{{group}}","edgeNodeId":"{{node}}",{{deviceKey}}"metricName":"{{metric}}","dataType":"{{dataType}}"}""";
}
private static async Task<SparkplugIngestor> NewSubscribedIngestorAsync()
{
var options = new MqttDriverOptions
{
Mode = MqttMode.SparkplugB,
Sparkplug = new MqttSparkplugOptions { GroupId = Group, HostId = "otopcua-host-1" },
};
var ingestor = new SparkplugIngestor(options, "mqtt-spb-1", rebirthDebounce: TimeSpan.Zero);
ingestor.Register(
[
Tag(TempPath, Blob(Group, Node, Device, "Temperature", "Float32")),
Tag(PressPath, Blob(Group, Node, Device, "Pressure", "Float32")),
]);
await ingestor.SubscribeAsync([.. ingestor.AuthoredRawPaths], TimeSpan.Zero, TestContext.Current.CancellationToken);
return ingestor;
}
private static void FeedBirths(SparkplugIngestor ing, float tempBirthValue = 0f, ulong bdSeq = 1)
{
ing.Dispatch(Topic(SparkplugMessageType.NBIRTH), Nbirth(seq: 0, bdSeq));
ing.Dispatch(Topic(SparkplugMessageType.DBIRTH, Device), Dbirth(seq: 1, tempBirthValue));
}
private static SparkplugTopic Topic(SparkplugMessageType type, string? device = null) =>
new(type, Group, Node, device, null);
private static SparkplugPayload Decode(Payload payload) => SparkplugCodec.Decode(payload.ToByteArray());
private static SparkplugPayload Nbirth(ulong seq, ulong bdSeq)
{
var payload = new Payload { Seq = seq, Timestamp = 1721822400000UL };
payload.Metrics.Add(new Payload.Types.Metric
{
Name = "bdSeq",
Datatype = (uint)TahuDataType.Uint64,
LongValue = bdSeq,
});
return Decode(payload);
}
private static SparkplugPayload Dbirth(ulong seq, float tempValue)
{
var payload = new Payload { Seq = seq, Timestamp = 1721822400000UL };
payload.Metrics.Add(BirthMetric("Temperature", TempAlias, tempValue));
payload.Metrics.Add(BirthMetric("Pressure", PressAlias, 0f));
return Decode(payload);
}
private static SparkplugPayload Ddata(ulong seq, params (ulong Alias, TahuDataType Type, float Value)[] metrics)
{
var payload = new Payload { Seq = seq, Timestamp = 1721822405000UL };
foreach (var (alias, _, value) in metrics)
{
// Deliberately NO Name and NO Datatype: the real post-birth wire shape.
payload.Metrics.Add(new Payload.Types.Metric { Alias = alias, FloatValue = value });
}
return Decode(payload);
}
private static SparkplugPayload Ddeath(ulong seq) =>
Decode(new Payload { Seq = seq, Timestamp = 1721822405000UL });
private static Payload.Types.Metric BirthMetric(string name, ulong alias, float value) => new()
{
Name = name,
Alias = alias,
Datatype = (uint)TahuDataType.Float,
FloatValue = value,
};
}
@@ -7,6 +7,7 @@ using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.Engines;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Logging;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Core.Scripting;
using ZB.MOM.WW.OtOpcUa.Host.Engines;
@@ -539,6 +540,98 @@ public sealed class RoslynVirtualTagEvaluatorTests
$"{starved}/{totalEvaluations} evaluations missed the 5s budget");
}
// ── input quality reaches the script (archreview 02/S14) ────────────────────────────────
/// <summary>
/// A script reading <c>ctx.GetTag(path).StatusCode</c> sees the dependency's <b>real</b> status.
/// The read cache used to stamp every input <c>Good</c>, so a script could not detect a degraded
/// input at all — "broken input looks Good", the other door onto the same lie 02/S13 closed for
/// broken scripts.
/// </summary>
[Fact]
public void Evaluate_read_cache_carries_the_inputs_real_status_code()
{
const uint uncertain = 0x40000000u;
using var sut = new RoslynVirtualTagEvaluator(NullLogger<RoslynVirtualTagEvaluator>.Instance, NoOpScriptRoot());
var stamp = new DateTime(2026, 8, 20, 9, 0, 0, DateTimeKind.Utc);
var result = sut.Evaluate(
virtualTagId: "vt-quality",
expression: "return (long)ctx.GetTag(\"a\").StatusCode;",
dependencies: new Dictionary<string, DataValueSnapshot>
{
["a"] = new(7, uncertain, stamp, stamp),
});
result.Success.ShouldBeTrue(result.Reason);
result.Value.ShouldBe((long)uncertain);
}
/// <summary>The dependency's own source timestamp reaches the script, not evaluation time.</summary>
[Fact]
public void Evaluate_read_cache_carries_the_inputs_real_source_timestamp()
{
using var sut = new RoslynVirtualTagEvaluator(NullLogger<RoslynVirtualTagEvaluator>.Instance, NoOpScriptRoot());
var stamp = new DateTime(2026, 8, 20, 9, 0, 0, DateTimeKind.Utc);
var result = sut.Evaluate(
virtualTagId: "vt-stamp",
expression: "return ctx.GetTag(\"a\").SourceTimestampUtc;",
dependencies: new Dictionary<string, DataValueSnapshot>
{
["a"] = new(7, 0u, stamp, DateTime.UtcNow),
});
result.Success.ShouldBeTrue(result.Reason);
result.Value.ShouldBe(stamp);
}
/// <summary>
/// The value-only overload keeps working unchanged — it is what the interface's default
/// implementation and the five test fakes still use, which is the whole reason the new overload is
/// default-implemented rather than abstract.
/// </summary>
[Fact]
public void The_value_only_overload_still_evaluates()
{
using var sut = new RoslynVirtualTagEvaluator(NullLogger<RoslynVirtualTagEvaluator>.Instance, NoOpScriptRoot());
var result = sut.Evaluate(
virtualTagId: "vt-values",
expression: "return (int)ctx.GetTag(\"a\").Value + 1;",
dependencies: new Dictionary<string, object?> { ["a"] = 41 });
result.Success.ShouldBeTrue(result.Reason);
result.Value.ShouldBe(42);
}
/// <summary>
/// A fake implementing ONLY the value-only member still answers the snapshot overload, through the
/// interface's default projection. Without that default, adding the overload would have been a
/// breaking change for every existing <see cref="IVirtualTagEvaluator"/> implementation.
/// </summary>
[Fact]
public void The_snapshot_overload_falls_back_to_a_value_only_implementation()
{
IVirtualTagEvaluator fake = new ValueOnlyEvaluator();
var result = fake.Evaluate("vt-fake", "ignored", new Dictionary<string, DataValueSnapshot>
{
["a"] = new(41, 0u, DateTime.UtcNow, DateTime.UtcNow),
});
result.Success.ShouldBeTrue();
result.Value.ShouldBe(41);
}
/// <summary>An evaluator predating the snapshot overload: implements the value-only member only.</summary>
private sealed class ValueOnlyEvaluator : IVirtualTagEvaluator
{
public VirtualTagEvalResult Evaluate(
string virtualTagId, string expression, IReadOnlyDictionary<string, object?> dependencies)
=> VirtualTagEvalResult.Ok(dependencies["a"]);
}
/// <summary>Reads the count of the private compiled-script cache via reflection. The cache is a
/// <c>CompiledScriptCache&lt;,&gt;</c> which exposes a <c>Count</c> property (not <c>ICollection</c>).</summary>
private static int CompiledCacheCount(RoslynVirtualTagEvaluator sut)
@@ -0,0 +1,226 @@
using Microsoft.Extensions.Logging.Abstractions;
using Opc.Ua;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests;
/// <summary>
/// <see cref="OtOpcUaNodeManager.WriteStatuses"/> — the status-only batch write behind the
/// comms-loss annotation for value variables (archreview 03/S15).
///
/// <para><b>Keeping the value is the whole point.</b> The host holds no value cache (v3's
/// single-source design), so "this driver has stopped communicating, but here is what it last
/// said" can only be expressed at the node, where the value already lives. The Part 8
/// last-known-value convention is what a SCADA client expects, and the interface's default
/// implementation — which nulls the value because a value-less sink has nothing else to write —
/// would silently erase it if the SDK sink ever stopped overriding it.</para>
///
/// <para><b>Never creates.</b> A status carries nothing to create a node from, and the caller's
/// per-driver index can legitimately name a node a surgical remove has just dropped. A missing id
/// is skipped, not materialised — the opposite of <see cref="OtOpcUaNodeManager.WriteValues"/>,
/// which lazily creates.</para>
///
/// <para>Boots a real <see cref="OtOpcUaSdkServer"/>, like
/// <see cref="NodeManagerBatchWriteTests"/>, so the assertions read the SDK node state a client
/// would receive.</para>
/// </summary>
public sealed class OtOpcUaNodeManagerWriteStatusesTests : IDisposable
{
private static CancellationToken Ct => TestContext.Current.CancellationToken;
private readonly string _pkiRoot = Path.Combine(
Path.GetTempPath(),
$"otopcua-writestatuses-{Guid.NewGuid():N}");
/// <summary>
/// The contract: value retained, status + timestamp replaced. Asserted on a node that was first
/// written a real value, because "value retained" is unfalsifiable on a node that never had one.
/// </summary>
[Fact]
public async Task WriteStatuses_replaces_status_and_timestamp_but_keeps_the_value()
{
var (host, server) = await BootAsync();
try
{
var nm = server.NodeManager!;
var t1 = new DateTime(2026, 8, 20, 10, 0, 0, DateTimeKind.Utc);
var t2 = new DateTime(2026, 8, 20, 10, 5, 0, DateTimeKind.Utc);
nm.EnsureVariable("Plant/Modbus/sim/HR5", parentFolderNodeId: null, displayName: "HR5",
dataType: "Int32", writable: false, realm: AddressSpaceRealm.Raw);
nm.WriteValue("Plant/Modbus/sim/HR5", 42, OpcUaStatus.Good, t1, AddressSpaceRealm.Raw);
nm.WriteStatuses([new StatusWrite("Plant/Modbus/sim/HR5", OpcUaStatus.BadNoCommunication, t2, AddressSpaceRealm.Raw)]);
var v = nm.TryGetVariable("Plant/Modbus/sim/HR5", AddressSpaceRealm.Raw);
v.ShouldNotBeNull();
v!.Value.ShouldBe(42, "the last known value must survive the comms-loss annotation");
v.StatusCode.ShouldBe((StatusCode)StatusCodes.BadNoCommunication);
v.StatusCode.ShouldNotBe((StatusCode)StatusCodes.Bad, "the sub-code reaches the SDK verbatim (#519)");
v.Timestamp.ShouldBe(t2);
}
finally
{
await host.DisposeAsync();
}
}
/// <summary>
/// Realm-qualified, like every other node-naming sink call: the same <c>s=</c> id in the other
/// namespace is a different node and must be untouched.
/// </summary>
[Fact]
public async Task WriteStatuses_only_touches_the_named_realm()
{
var (host, server) = await BootAsync();
try
{
var nm = server.NodeManager!;
var ts = new DateTime(2026, 8, 20, 10, 0, 0, DateTimeKind.Utc);
const string sharedId = "Shared/both/realms";
nm.WriteValue(sharedId, 1, OpcUaStatus.Good, ts, AddressSpaceRealm.Raw);
nm.WriteValue(sharedId, 2, OpcUaStatus.Good, ts, AddressSpaceRealm.Uns);
nm.WriteStatuses([new StatusWrite(sharedId, OpcUaStatus.BadNoCommunication, ts, AddressSpaceRealm.Raw)]);
nm.TryGetVariable(sharedId, AddressSpaceRealm.Raw)!.StatusCode
.ShouldBe((StatusCode)StatusCodes.BadNoCommunication);
nm.TryGetVariable(sharedId, AddressSpaceRealm.Uns)!.StatusCode
.ShouldBe((StatusCode)StatusCodes.Good);
}
finally
{
await host.DisposeAsync();
}
}
/// <summary>
/// An unknown NodeId is a no-op: no node is created, nothing throws, and the rest of the batch
/// still lands. A create here would materialise a node nobody authored, holding no value, from a
/// driver-index entry that is stale precisely because the node was removed.
/// </summary>
[Fact]
public async Task An_unknown_node_id_is_skipped_not_created()
{
var (host, server) = await BootAsync();
try
{
var nm = server.NodeManager!;
var ts = new DateTime(2026, 8, 20, 10, 0, 0, DateTimeKind.Utc);
nm.WriteValue("Plant/known", 7, OpcUaStatus.Good, ts, AddressSpaceRealm.Raw);
var before = nm.VariableCount;
Should.NotThrow(() => nm.WriteStatuses(
[
new StatusWrite("Plant/does/not/exist", OpcUaStatus.BadNoCommunication, ts, AddressSpaceRealm.Raw),
new StatusWrite("Plant/known", OpcUaStatus.BadNoCommunication, ts, AddressSpaceRealm.Raw),
]));
nm.VariableCount.ShouldBe(before, "a status write must never create a node");
nm.TryGetVariable("Plant/does/not/exist", AddressSpaceRealm.Raw).ShouldBeNull();
nm.TryGetVariable("Plant/known", AddressSpaceRealm.Raw)!.StatusCode
.ShouldBe((StatusCode)StatusCodes.BadNoCommunication);
}
finally
{
await host.DisposeAsync();
}
}
/// <summary>
/// A malformed item does not drop the tail — the same per-item isolation contract
/// <see cref="OtOpcUaNodeManager.WriteValues"/> carries, reported through the same
/// <see cref="ValueWriteBatchException"/> aggregate.
/// </summary>
[Fact]
public async Task A_failing_item_neither_drops_the_tail_nor_hides_which_node_failed()
{
var (host, server) = await BootAsync();
try
{
var nm = server.NodeManager!;
var ts = new DateTime(2026, 8, 20, 10, 0, 0, DateTimeKind.Utc);
nm.WriteValue("Plant/first", 1, OpcUaStatus.Good, ts, AddressSpaceRealm.Raw);
nm.WriteValue("Plant/third", 3, OpcUaStatus.Good, ts, AddressSpaceRealm.Raw);
var ex = Should.Throw<ValueWriteBatchException>(() => nm.WriteStatuses(
[
new StatusWrite("Plant/first", OpcUaStatus.BadNoCommunication, ts, AddressSpaceRealm.Raw),
new StatusWrite(string.Empty, OpcUaStatus.BadNoCommunication, ts, AddressSpaceRealm.Raw),
new StatusWrite("Plant/third", OpcUaStatus.BadNoCommunication, ts, AddressSpaceRealm.Raw),
]));
ex.FailedCount.ShouldBe(1);
ex.SucceededCount.ShouldBe(2);
ex.FirstFailedNodeId.ShouldBe(string.Empty);
nm.TryGetVariable("Plant/first", AddressSpaceRealm.Raw)!.Value.ShouldBe(1);
nm.TryGetVariable("Plant/third", AddressSpaceRealm.Raw)!.StatusCode
.ShouldBe((StatusCode)StatusCodes.BadNoCommunication, "the item after the failing one must not be dropped");
}
finally
{
await host.DisposeAsync();
}
}
/// <summary>An empty batch neither throws nor touches the address space.</summary>
[Fact]
public async Task Empty_batch_is_a_no_op()
{
var (host, server) = await BootAsync();
try
{
var nm = server.NodeManager!;
var before = nm.VariableCount;
Should.NotThrow(() => nm.WriteStatuses([]));
nm.VariableCount.ShouldBe(before);
}
finally
{
await host.DisposeAsync();
}
}
private async Task<(OpcUaApplicationHost Host, OtOpcUaSdkServer Server)> BootAsync()
{
var host = new OpcUaApplicationHost(
new OpcUaApplicationHostOptions
{
ApplicationName = "OtOpcUa.WriteStatusesTest",
ApplicationUri = $"urn:OtOpcUa.WriteStatusesTest:{Guid.NewGuid():N}",
OpcUaPort = AllocateFreePort(),
PublicHostname = "localhost",
PkiStoreRoot = _pkiRoot,
},
NullLogger<OpcUaApplicationHost>.Instance);
var server = new OtOpcUaSdkServer();
await host.StartAsync(server, Ct);
return (host, server);
}
private static int AllocateFreePort()
{
using var listener = new System.Net.Sockets.TcpListener(System.Net.IPAddress.Loopback, 0);
listener.Start();
var port = ((System.Net.IPEndPoint)listener.LocalEndpoint).Port;
listener.Stop();
return port;
}
/// <summary>Cleans up the PKI root directory.</summary>
public void Dispose()
{
if (Directory.Exists(_pkiRoot))
{
try { Directory.Delete(_pkiRoot, recursive: true); }
catch { /* best-effort cleanup */ }
}
}
}
@@ -0,0 +1,298 @@
using System.Diagnostics.Metrics;
using Akka.Actor;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.Observability;
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
using ZB.MOM.WW.OtOpcUa.Commons.Types;
using ZB.MOM.WW.OtOpcUa.Runtime.Drivers;
using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa;
using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness;
using static ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness.V3ArtifactSeeder;
namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers;
/// <summary>
/// Archreview <b>03/S15</b> — the value-node half of the comms-loss fan-out. Until this, a driver
/// declared comms-lost annotated its Part 9 conditions and pushed a synthetic Bad into the dependency
/// mux, but its <b>raw and UNS value variables</b> — the product's external surface — got nothing: an
/// OPC UA client subscribed to a value node saw the last value at <b>Good</b> quality for the whole
/// outage, for every driver whose failure shape is "reader stops producing" rather than "per-tag Bad
/// snapshot".
///
/// <para><b>Status-only, value retained.</b> The write goes through
/// <see cref="OpcUaPublishActor.AttributeStatusBatchUpdate"/> so the node keeps its last known value
/// under <c>BadNoCommunication</c> (OPC UA Part 8's convention). The host holds no value cache — v3's
/// single-source design — so only the node itself can retain it.</para>
///
/// <para><b>Nothing is published on reconnect</b>, deliberately: publishing Good would assert a
/// quality this actor has not observed. Recovery is the driver's own re-publish (the poll engine's
/// <c>forceRaise</c> on a fresh subscription, or the push-model initial push). That contract is
/// end-to-end tested in <see cref="DriverHostCommsLossRecoveryTests"/>.</para>
/// </summary>
public sealed class DriverHostActorCommsLossValueNodesTests : RuntimeActorTestBase
{
private static readonly NodeId TestNode = NodeId.Parse("driver-commsloss-values");
private static readonly RevisionHash RevA = RevisionHash.Parse(new string('c', 64));
private static readonly DateTime TsA = new(2026, 8, 20, 9, 0, 0, DateTimeKind.Utc);
/// <summary>Audit delay used by the T9 cases — long enough that a same-turn prune wins the race,
/// short enough to observe.</summary>
private static readonly TimeSpan AuditDelay = TimeSpan.FromMilliseconds(400);
// Distinct driver ids per metered test: the meter tag is what the recorder filters on, and a
// MeterListener sees every measurement in the process — including a parallel test's (#503).
private const string SilentDriver = "drv-audit-silent";
private const string RepublishingDriver = "drv-audit-republish";
private const string PartialDriver = "drv-audit-partial";
/// <summary>
/// A disconnect produces ONE status batch naming exactly the disconnected driver's raw nodes plus
/// every referencing UNS mirror, all <c>BadNoCommunication</c>. The second driver's ref is the
/// control: a fan-out that walked every value ref on the node instead of the per-driver index would
/// mark a perfectly healthy driver's nodes Bad, and no assertion about the first driver's nodes
/// would notice.
/// </summary>
[Fact]
public void Disconnect_marks_the_drivers_raw_and_uns_value_nodes_BadNoCommunication()
{
var db = NewInMemoryDbFactory();
var deploymentId = SeedV3Deployment(db, RevA,
rawTags: new[]
{
(Driver: "drv-1", DriverName: "Modbus", Device: "dev1", Tag: "speed", DataType: "Double", Writable: false),
(Driver: "drv-1", DriverName: "Modbus", Device: "dev1", Tag: "temp", DataType: "Double", Writable: false),
(Driver: "drv-2", DriverName: "S7", Device: "dev2", Tag: "other", DataType: "Double", Writable: false),
},
unsRefs: new[] { (Area: "filling", Line: "line1", Equip: "station1", BackingTag: "speed", Effective: (string?)null) });
var (actor, publish, _) = SpawnHostAndApply(this, db, TestNode, deploymentId, RevA);
actor.Tell(new DriverInstanceActor.ConnectivityChanged("drv-1", Connected: false));
var batch = publish.ExpectMsg<OpcUaPublishActor.AttributeStatusBatchUpdate>(PresenceBudget);
batch.Items.Select(i => i.NodeId).ShouldBe(new[]
{
"Plant/Modbus/dev1/speed",
"filling/line1/station1/speed",
"Plant/Modbus/dev1/temp",
}, ignoreOrder: true);
batch.Items.ShouldAllBe(i => i.StatusCode == OpcUaStatus.BadNoCommunication);
batch.Items.Single(i => i.NodeId == "Plant/Modbus/dev1/speed").Realm.ShouldBe(AddressSpaceRealm.Raw);
batch.Items.Single(i => i.NodeId == "filling/line1/station1/speed").Realm.ShouldBe(AddressSpaceRealm.Uns);
// The control: the OTHER driver's node is untouched by drv-1's outage.
batch.Items.ShouldNotContain(i => i.NodeId == "Plant/S7/dev2/other");
// ONE batch, not one message per node.
publish.ExpectNoMsg(TimeSpan.FromMilliseconds(300));
}
/// <summary>
/// A reconnect publishes <b>nothing</b> to the value nodes. Writing Good here would claim a quality
/// the host has not observed — the device may be answering the transport and still returning
/// garbage — and it would race the driver's own first real value.
/// </summary>
[Fact]
public void Reconnect_publishes_no_value_or_status_message()
{
var db = NewInMemoryDbFactory();
var deploymentId = SeedV3Deployment(db, RevA,
rawTags: new[]
{
(Driver: "drv-1", DriverName: "Modbus", Device: "dev1", Tag: "speed", DataType: "Double", Writable: false),
},
unsRefs: Array.Empty<(string, string, string, string, string?)>());
var (actor, publish, _) = SpawnHostAndApply(this, db, TestNode, deploymentId, RevA);
actor.Tell(new DriverInstanceActor.ConnectivityChanged("drv-1", Connected: false));
publish.ExpectMsg<OpcUaPublishActor.AttributeStatusBatchUpdate>(PresenceBudget);
actor.Tell(new DriverInstanceActor.ConnectivityChanged("drv-1", Connected: true));
publish.ExpectNoMsg(TimeSpan.FromMilliseconds(500));
}
// ---------------------------------------------------------------------------------
// T9 — the recovery audit (the dedup-trap tripwire)
// ---------------------------------------------------------------------------------
/// <summary>
/// The host publishes nothing Good on reconnect, so the only way a stuck-Bad node becomes visible
/// is this audit: refs still latched some time after the driver reconnected are logged and metered.
/// Here the driver re-publishes nothing at all — the pre-STAB-18 MQTT shape, and the exact failure
/// any future push driver without an initial push would produce.
/// </summary>
[Fact]
public void Refs_still_latched_after_a_reconnect_are_logged_and_metered()
{
var db = NewInMemoryDbFactory();
var deploymentId = SeedV3Deployment(db, RevA,
rawTags: new[]
{
(Driver: SilentDriver, DriverName: "Modbus", Device: "dev1", Tag: "speed", DataType: "Double", Writable: false),
(Driver: SilentDriver, DriverName: "Modbus", Device: "dev1", Tag: "temp", DataType: "Double", Writable: false),
},
unsRefs: new[] { (Area: "filling", Line: "line1", Equip: "station1", BackingTag: "speed", Effective: (string?)null) });
var (actor, publish, _) = SpawnHostAndApply(this, db, TestNode, deploymentId, RevA, commsRecoveryAuditDelay: AuditDelay);
using var meter = new MeterRecorder("otopcua.driver.comms_latched_refs");
actor.Tell(new DriverInstanceActor.ConnectivityChanged(SilentDriver, Connected: false));
publish.ExpectMsg<OpcUaPublishActor.AttributeStatusBatchUpdate>(PresenceBudget);
EventFilter.Warning(contains: "still BadNoCommunication").ExpectOne(TimeSpan.FromSeconds(10), () =>
actor.Tell(new DriverInstanceActor.ConnectivityChanged(SilentDriver, Connected: true)));
// Two REFS, not three nodes: the latch is keyed by RawPath, and the UNS mirror shares its
// backing tag's ref — the operator is being told how many tags did not come back, not how many
// address-space nodes they project onto.
AwaitCondition(() => meter.WithTag("driver", SilentDriver) > 0, TimeSpan.FromSeconds(5));
meter.SumWithTag("driver", SilentDriver).ShouldBe(2);
}
/// <summary>
/// <b>The control.</b> A driver that DOES re-publish every latched ref on reconnect trips nothing —
/// without this, the test above would pass against an audit that fires unconditionally, which would
/// make the Warning noise rather than signal.
/// </summary>
[Fact]
public void A_driver_that_republishes_every_ref_trips_no_audit()
{
var db = NewInMemoryDbFactory();
var deploymentId = SeedV3Deployment(db, RevA,
rawTags: new[]
{
(Driver: RepublishingDriver, DriverName: "Modbus", Device: "dev1", Tag: "speed", DataType: "Double", Writable: false),
(Driver: RepublishingDriver, DriverName: "Modbus", Device: "dev1", Tag: "temp", DataType: "Double", Writable: false),
},
unsRefs: new[] { (Area: "filling", Line: "line1", Equip: "station1", BackingTag: "speed", Effective: (string?)null) });
var (actor, publish, _) = SpawnHostAndApply(this, db, TestNode, deploymentId, RevA, commsRecoveryAuditDelay: AuditDelay);
using var meter = new MeterRecorder("otopcua.driver.comms_latched_refs");
actor.Tell(new DriverInstanceActor.ConnectivityChanged(RepublishingDriver, Connected: false));
publish.ExpectMsg<OpcUaPublishActor.AttributeStatusBatchUpdate>(PresenceBudget);
EventFilter.Warning(contains: "still BadNoCommunication").Expect(0, TimeSpan.FromSeconds(5), () =>
{
actor.Tell(new DriverInstanceActor.ConnectivityChanged(RepublishingDriver, Connected: true));
// The driver's own re-publish — identical values, which is precisely the case a value-only
// dedup would swallow; here it is the driver that force-raised, so both refs are pruned.
actor.Tell(new DriverInstanceActor.AttributeValueBatchPublished(RepublishingDriver, new[]
{
new DriverInstanceActor.AttributeValueItem("Plant/Modbus/dev1/speed", 42.0, OpcUaStatus.Good, TsA),
new DriverInstanceActor.AttributeValueItem("Plant/Modbus/dev1/temp", 7.5, OpcUaStatus.Good, TsA),
}));
publish.ExpectMsg<OpcUaPublishActor.AttributeValueBatchUpdate>(PresenceBudget);
});
meter.WithTag("driver", RepublishingDriver).ShouldBe(0);
}
/// <summary>
/// A partial recovery is still reported, and reports only what is missing: one ref comes back, the
/// other does not, and the audit counts exactly one. An audit that reported the whole latch (or
/// nothing at all once anything arrived) would be useless on precisely the mixed case an operator
/// needs it for.
/// </summary>
[Fact]
public void A_partial_recovery_reports_only_the_refs_that_did_not_come_back()
{
var db = NewInMemoryDbFactory();
var deploymentId = SeedV3Deployment(db, RevA,
rawTags: new[]
{
(Driver: PartialDriver, DriverName: "Modbus", Device: "dev1", Tag: "speed", DataType: "Double", Writable: false),
(Driver: PartialDriver, DriverName: "Modbus", Device: "dev1", Tag: "temp", DataType: "Double", Writable: false),
},
unsRefs: Array.Empty<(string, string, string, string, string?)>());
var (actor, publish, _) = SpawnHostAndApply(this, db, TestNode, deploymentId, RevA, commsRecoveryAuditDelay: AuditDelay);
using var meter = new MeterRecorder("otopcua.driver.comms_latched_refs");
actor.Tell(new DriverInstanceActor.ConnectivityChanged(PartialDriver, Connected: false));
publish.ExpectMsg<OpcUaPublishActor.AttributeStatusBatchUpdate>(PresenceBudget);
EventFilter.Warning(contains: "still BadNoCommunication").ExpectOne(TimeSpan.FromSeconds(10), () =>
{
actor.Tell(new DriverInstanceActor.ConnectivityChanged(PartialDriver, Connected: true));
actor.Tell(new DriverInstanceActor.AttributeValueBatchPublished(PartialDriver, new[]
{
new DriverInstanceActor.AttributeValueItem("Plant/Modbus/dev1/speed", 42.0, OpcUaStatus.Good, TsA),
}));
publish.ExpectMsg<OpcUaPublishActor.AttributeValueBatchUpdate>(PresenceBudget);
});
AwaitCondition(() => meter.WithTag("driver", PartialDriver) > 0, TimeSpan.FromSeconds(5));
meter.SumWithTag("driver", PartialDriver).ShouldBe(1, "only the ref that never came back is reported");
}
/// <summary>
/// A driver that owns no value refs produces no message at all — an empty batch would cost a hop
/// and a sink call for nothing on every flap of an alarm-only or freshly-added driver.
/// </summary>
[Fact]
public void A_driver_with_no_value_refs_publishes_nothing()
{
var db = NewInMemoryDbFactory();
var deploymentId = SeedV3Deployment(db, RevA,
rawTags: new[]
{
(Driver: "drv-1", DriverName: "Modbus", Device: "dev1", Tag: "speed", DataType: "Double", Writable: false),
},
unsRefs: Array.Empty<(string, string, string, string, string?)>());
var (actor, publish, _) = SpawnHostAndApply(this, db, TestNode, deploymentId, RevA);
actor.Tell(new DriverInstanceActor.ConnectivityChanged("drv-unknown", Connected: false));
publish.ExpectNoMsg(TimeSpan.FromMilliseconds(500));
}
/// <summary>
/// Sums one counter's measurements, <b>filtered by tag</b>. A <see cref="MeterListener"/> observes
/// every measurement in the process, so filtering on instrument name alone would fold in a parallel
/// test's (#503) — each test here therefore uses its own driver id and asserts on that tag.
/// </summary>
private sealed class MeterRecorder : IDisposable
{
private readonly MeterListener _listener;
private readonly List<(long Value, KeyValuePair<string, object?>[] Tags)> _measurements = new();
private readonly object _gate = new();
public MeterRecorder(string instrumentName)
{
_listener = new MeterListener
{
InstrumentPublished = (instrument, listener) =>
{
if (instrument.Meter.Name == OtOpcUaTelemetry.MeterName && instrument.Name == instrumentName)
listener.EnableMeasurementEvents(instrument);
},
};
_listener.SetMeasurementEventCallback<long>((_, value, tags, _) =>
{
lock (_gate) _measurements.Add((value, tags.ToArray()));
});
_listener.Start();
}
public int WithTag(string key, string value)
{
lock (_gate) return _measurements.Count(m => m.Tags.Any(t => t.Key == key && Equals(t.Value, value)));
}
public long SumWithTag(string key, string value)
{
lock (_gate)
return _measurements.Where(m => m.Tags.Any(t => t.Key == key && Equals(t.Value, value)))
.Sum(m => m.Value);
}
public void Dispose() => _listener.Dispose();
}
}
@@ -0,0 +1,237 @@
using System.Diagnostics.Metrics;
using Akka.Actor;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Admin;
using ZB.MOM.WW.OtOpcUa.Commons.Observability;
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
using ZB.MOM.WW.OtOpcUa.Commons.Types;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Runtime.Drivers;
using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa;
using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness;
using static ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness.V3ArtifactSeeder;
namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers;
/// <summary>
/// <b>The 03/S15 contract test: comms loss → value nodes Bad → the driver's own re-publish clears
/// them.</b> A real <see cref="DriverInstanceActor"/> is spawned by a real
/// <see cref="DriverHostActor"/> around a scripted driver, so the whole seam runs: the child's
/// <c>Reconnecting</c> entry announces comms loss, the host stamps the value nodes
/// <c>BadNoCommunication</c>, the child reconnects and re-subscribes, and the driver's initial push is
/// what makes the nodes Good again.
///
/// <para>⚠️ <b>Vacuity rule — the pushed value is IDENTICAL to the pre-outage one.</b> The host writes
/// nothing Good on reconnect by design, so recovery depends entirely on a re-publish landing; and every
/// dedup on that path (a driver's last-value cache, a consumer's last-published memory) keys on value.
/// A recovery test whose driver pushes a <em>different</em> value therefore passes against a build with
/// no force-republish anywhere — the value change alone defeats every dedup — and is not this
/// contract's guard. The same-value push is the whole point.</para>
///
/// <para>The falsification control is
/// <see cref="A_silent_resubscribe_leaves_the_nodes_Bad_and_trips_the_audit"/>: a driver whose
/// <c>SubscribeAsync</c> pushes nothing (the pre-STAB-18 MQTT shape) leaves the nodes Bad and is caught
/// only by the audit. It proves the test above can fail.</para>
/// </summary>
public sealed class DriverHostCommsLossRecoveryTests : RuntimeActorTestBase
{
private static readonly NodeId TestNode = NodeId.Parse("driver-commsloss-recovery");
private static readonly RevisionHash RevA = RevisionHash.Parse(new string('d', 64));
/// <summary>Audit window measured from the RECONNECT, not from the outage — the initial push lands
/// within milliseconds of Connected, so this is generous.</summary>
private static readonly TimeSpan AuditDelay = TimeSpan.FromSeconds(2);
/// <summary>The child's reconnect retry runs on <c>DriverInstanceActor.DefaultReconnectInterval</c>
/// (10 s) and the host exposes no seam to shorten it, so every budget here clears that.</summary>
private static readonly TimeSpan ReconnectBudget = TimeSpan.FromSeconds(25);
private const string SpeedRef = "Plant/Modbus/dev1/speed";
private const string UnsSpeedNode = "filling/line1/station1/speed";
[Fact]
public void An_identical_value_pushed_on_resubscribe_recovers_the_nodes()
{
var driver = new ScriptedSubscribeDriver { InstanceId = "drv-recover", InitialPush = (SpeedRef, 5.0, OpcUaStatus.Good) };
var (host, publish) = Arrange(driver, "drv-recover");
using var meter = new MeterRecorder("otopcua.driver.comms_latched_refs");
// Steady state: the driver's FIRST subscribe already pushed 5 Good, and the host fanned it to
// raw + UNS. (The push on subscribe is the behaviour under test, so it is also how the pre-outage
// value gets there — no separate seeding call, which would leave an extra batch in the probe.)
var first = publish.ExpectMsg<OpcUaPublishActor.AttributeValueBatchUpdate>(PresenceBudget);
first.Items.Select(i => i.NodeId).ShouldBe(new[] { SpeedRef, UnsSpeedNode }, ignoreOrder: true);
first.Items.ShouldAllBe(i => Equals(i.Value, 5.0));
// The outage: the child re-enters Reconnecting and announces comms loss to the host.
host.Tell(new ReconnectDriver("c1", "drv-recover", "test-operator", Guid.NewGuid()));
var staled = publish.ExpectMsg<OpcUaPublishActor.AttributeStatusBatchUpdate>(ReconnectBudget);
staled.Items.Select(i => i.NodeId).ShouldBe(new[] { SpeedRef, UnsSpeedNode }, ignoreOrder: true);
staled.Items.ShouldAllBe(i => i.StatusCode == OpcUaStatus.BadNoCommunication);
// Recovery: the reconnect re-subscribes, and the driver pushes the SAME value it held before.
var recovered = publish.ExpectMsg<OpcUaPublishActor.AttributeValueBatchUpdate>(ReconnectBudget);
recovered.Items.Select(i => i.NodeId).ShouldBe(new[] { SpeedRef, UnsSpeedNode }, ignoreOrder: true);
recovered.Items.ShouldAllBe(i => i.StatusCode == OpcUaStatus.Good);
recovered.Items.ShouldAllBe(i => Equals(i.Value, 5.0));
// …and the audit finds nothing outstanding.
EventFilter.Warning(contains: "still BadNoCommunication")
.Expect(0, AuditDelay + TimeSpan.FromSeconds(3), () => { });
meter.SumWithTag("driver", "drv-recover").ShouldBe(0);
}
/// <summary>
/// <b>The falsification control.</b> Same scenario, but the driver's <c>SubscribeAsync</c> pushes
/// nothing — exactly what MQTT did before 05/STAB-18, and what any push driver without an initial
/// push does. The nodes stay Bad (the host writes no Good of its own), and the ONLY thing that says
/// so is the recovery audit.
/// </summary>
[Fact]
public void A_silent_resubscribe_leaves_the_nodes_Bad_and_trips_the_audit()
{
var driver = new ScriptedSubscribeDriver { InstanceId = "drv-silent", InitialPush = null };
var (host, publish) = Arrange(driver, "drv-silent");
using var meter = new MeterRecorder("otopcua.driver.comms_latched_refs");
driver.FireDataChange(SpeedRef, 5.0, OpcUaStatus.Good);
publish.ExpectMsg<OpcUaPublishActor.AttributeValueBatchUpdate>(PresenceBudget);
EventFilter.Warning(contains: "still BadNoCommunication").ExpectOne(ReconnectBudget, () =>
{
host.Tell(new ReconnectDriver("c1", "drv-silent", "test-operator", Guid.NewGuid()));
publish.ExpectMsg<OpcUaPublishActor.AttributeStatusBatchUpdate>(ReconnectBudget);
// Nothing Good ever arrives — assert it rather than merely not looking.
publish.ExpectNoMsg(TimeSpan.FromSeconds(1));
});
// One ref (the raw tag); its UNS mirror shares it.
AwaitCondition(() => meter.SumWithTag("driver", "drv-silent") == 1, TimeSpan.FromSeconds(5));
driver.SubscribeCount.ShouldBeGreaterThanOrEqualTo(2,
"the child must actually have re-subscribed — otherwise this proves nothing about the push");
}
private (IActorRef Host, Akka.TestKit.TestProbe Publish) Arrange(ScriptedSubscribeDriver driver, string driverId)
{
var db = NewInMemoryDbFactory();
var deploymentId = SeedV3Deployment(db, RevA,
rawTags: new[]
{
(Driver: driverId, DriverName: "Modbus", Device: "dev1", Tag: "speed", DataType: "Double", Writable: false),
},
unsRefs: new[] { (Area: "filling", Line: "line1", Equip: "station1", BackingTag: "speed", Effective: (string?)null) },
enabled: true);
var (host, publish, _) = SpawnHostAndApply(
this, db, TestNode, deploymentId, RevA,
commsRecoveryAuditDelay: AuditDelay,
driverFactory: new SingleDriverFactory(driver));
// The child is spawned + initialised by the apply; wait for its first subscribe so the scripted
// pushes below are attributed to a live subscription.
AwaitCondition(() => driver.SubscribeCount >= 1, PresenceBudget);
return (host, publish);
}
/// <summary>Hands the host one scripted driver, whatever type/instance it asks for.</summary>
private sealed class SingleDriverFactory(IDriver driver) : IDriverFactory
{
public IDriver? TryCreate(string driverType, string driverInstanceId, string driverConfigJson) => driver;
public IReadOnlyCollection<string> SupportedTypes { get; } = ["Modbus"];
}
/// <summary>
/// A subscribable driver whose <see cref="SubscribeAsync"/> optionally performs an <b>initial
/// push</b> — the poll engine's <c>forceRaise: true</c> and MQTT's post-STAB-18 behaviour, modelled
/// at the seam the host actually observes. With <see cref="InitialPush"/> null it is the silent
/// shape the control test needs.
/// </summary>
private sealed class ScriptedSubscribeDriver : IDriver, ISubscribable
{
private readonly StubHandle _handle = new();
private int _subscribeCount;
public event EventHandler<DataChangeEventArgs>? OnDataChange;
/// <summary>The value re-published on every (re)subscribe, or null for a silent driver.</summary>
public (string Ref, object? Value, uint StatusCode)? InitialPush { get; init; }
/// <summary>The instance id the child actor adopts — <c>DriverInstanceActor</c> takes it from the
/// DRIVER, not from the spec, and the host's ref map is keyed by the spec's id. They must agree or
/// every published value is dropped as unmapped.</summary>
public required string InstanceId { get; init; }
public int SubscribeCount => Volatile.Read(ref _subscribeCount);
public string DriverInstanceId => InstanceId;
public string DriverType => "Modbus";
public Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken) => Task.CompletedTask;
public Task ReinitializeAsync(string driverConfigJson, CancellationToken cancellationToken) => Task.CompletedTask;
public Task ShutdownAsync(CancellationToken cancellationToken) => Task.CompletedTask;
// Stable health: the #527 watchdog must not fire on its own here — the outage in these tests is
// driven explicitly, and a health-driven trip would make the timing non-deterministic.
public DriverHealth GetHealth() => new(DriverState.Healthy, DateTime.UtcNow, null);
public Task<ISubscriptionHandle> SubscribeAsync(
IReadOnlyList<string> fullReferences, TimeSpan publishingInterval, CancellationToken cancellationToken)
{
Interlocked.Increment(ref _subscribeCount);
if (InitialPush is { } push && fullReferences.Contains(push.Ref))
FireDataChange(push.Ref, push.Value, push.StatusCode);
return Task.FromResult<ISubscriptionHandle>(_handle);
}
public Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken) => Task.CompletedTask;
public void FireDataChange(string fullRef, object? value, uint statusCode)
{
var snapshot = new DataValueSnapshot(value, statusCode, DateTime.UtcNow, DateTime.UtcNow);
OnDataChange?.Invoke(this, new DataChangeEventArgs(_handle, fullRef, snapshot));
}
}
/// <summary>Tag-filtered counter recorder — see the note in
/// <c>DriverHostActorCommsLossValueNodesTests</c> on why the instrument name alone is not enough.</summary>
private sealed class MeterRecorder : IDisposable
{
private readonly MeterListener _listener;
private readonly List<(long Value, KeyValuePair<string, object?>[] Tags)> _measurements = new();
private readonly object _gate = new();
public MeterRecorder(string instrumentName)
{
_listener = new MeterListener
{
InstrumentPublished = (instrument, listener) =>
{
if (instrument.Meter.Name == OtOpcUaTelemetry.MeterName && instrument.Name == instrumentName)
listener.EnableMeasurementEvents(instrument);
},
};
_listener.SetMeasurementEventCallback<long>((_, value, tags, _) =>
{
lock (_gate) _measurements.Add((value, tags.ToArray()));
});
_listener.Start();
}
public long SumWithTag(string key, string value)
{
lock (_gate)
return _measurements.Where(m => m.Tags.Any(t => t.Key == key && Equals(t.Value, value)))
.Sum(m => m.Value);
}
public void Dispose() => _listener.Dispose();
}
}
@@ -374,4 +374,54 @@ public sealed class DriverInstanceActorTests : RuntimeActorTestBase
AwaitCondition(() => driver.SubscribeCount >= 1, TimeSpan.FromSeconds(3));
}
/// <summary>
/// Archreview <b>03/S15 (T11)</b> — a child that is <b>stopped</b> announces comms loss on its way
/// out, so the host latches its value nodes <c>BadNoCommunication</c>.
/// <para>Every other route into <c>Reconnecting</c> announces it (watchdog trip, failed initial
/// connect, Force Reconnect all run the <c>Reconnecting</c> entry). Stopping did not: an operator's
/// Restart and a supervision respawn tore the child down silently, so its nodes kept serving their
/// last value at Good across the gap. The fresh child's re-subscribe + initial push (03/S16) is what
/// clears them again.</para>
/// </summary>
[Fact]
public void Stopping_a_connected_child_announces_comms_loss_to_its_parent()
{
var parent = CreateTestProbe();
var driver = new StubDriver();
var actor = parent.ChildActorOf(DriverInstanceActor.Props(driver));
actor.Tell(new DriverInstanceActor.InitializeRequested("{}"));
parent.ExpectMsg<DriverInstanceActor.ConnectivityChanged>(PresenceBudget).Connected.ShouldBeTrue();
parent.Watch(actor);
parent.Send(actor, PoisonPill.Instance);
parent.ExpectMsg<DriverInstanceActor.ConnectivityChanged>(PresenceBudget).Connected.ShouldBeFalse();
parent.ExpectTerminated(actor, PresenceBudget);
}
/// <summary>
/// <b>Control.</b> A child that never reached <c>Connected</c> announces nothing when stopped — its
/// nodes are still <c>BadWaitingForInitialData</c>, which is a more precise statement than
/// "communication lost", and overwriting it would claim an outage that never happened.
/// </summary>
[Fact]
public void Stopping_a_never_connected_child_announces_nothing()
{
var parent = CreateTestProbe();
var driver = new StubDriver { InitializeShouldThrow = true };
var actor = parent.ChildActorOf(DriverInstanceActor.Props(
driver, reconnectInterval: TimeSpan.FromSeconds(30)));
actor.Tell(new DriverInstanceActor.InitializeRequested("{}"));
// The failed initial connect announces Connected=false from the Reconnecting entry; consume it, so
// what follows is only what STOPPING produced.
parent.ExpectMsg<DriverInstanceActor.ConnectivityChanged>(PresenceBudget).Connected.ShouldBeFalse();
parent.Watch(actor);
parent.Send(actor, PoisonPill.Instance);
parent.ExpectTerminated(actor, PresenceBudget);
parent.ExpectNoMsg(TimeSpan.FromMilliseconds(300));
}
}
@@ -140,15 +140,20 @@ public static class V3ArtifactSeeder
/// <param name="reprovisionBackoff">Optional per-driver supervision-reprovision rate limit forwarded to
/// <c>DriverHostActor.Props</c>; null keeps the production default. Exists so a test can observe the
/// deferral without waiting the production window.</param>
/// <param name="commsRecoveryAuditDelay">Optional 03/S15 audit delay — how long after a reconnect the
/// host checks for refs still latched <c>BadNoCommunication</c>. Null keeps the production default,
/// which is far longer than any test's lifetime.</param>
/// <returns>The host actor plus the publish and mux probes wired into it.</returns>
public static (IActorRef Actor, TestProbe Publish, TestProbe Mux) SpawnHostAndApply(
TestKitBase kit, IDbContextFactory<OtOpcUaConfigDbContext> db,
NodeId localNode, DeploymentId deploymentId, RevisionHash rev,
IDriverFactory? driverFactory = null,
TimeSpan? reprovisionBackoff = null)
TimeSpan? reprovisionBackoff = null,
TimeSpan? commsRecoveryAuditDelay = null)
{
var publish = kit.CreateTestProbe();
var (actor, mux, coordinator) = Spawn(kit, db, localNode, publish.Ref, driverFactory, reprovisionBackoff);
var (actor, mux, coordinator) =
Spawn(kit, db, localNode, publish.Ref, driverFactory, reprovisionBackoff, commsRecoveryAuditDelay);
Dispatch(actor, coordinator, deploymentId, rev);
publish.ExpectMsg<OpcUaPublishActor.RebuildAddressSpace>(RuntimeActorTestBase.PresenceBudget);
@@ -175,7 +180,8 @@ public static class V3ArtifactSeeder
private static (IActorRef Actor, TestProbe Mux, TestProbe Coordinator) Spawn(
TestKitBase kit, IDbContextFactory<OtOpcUaConfigDbContext> db, NodeId localNode,
IActorRef? opcUaPublishActor, IDriverFactory? driverFactory, TimeSpan? reprovisionBackoff)
IActorRef? opcUaPublishActor, IDriverFactory? driverFactory, TimeSpan? reprovisionBackoff,
TimeSpan? commsRecoveryAuditDelay = null)
{
var coordinator = kit.CreateTestProbe();
var mux = kit.CreateTestProbe();
@@ -189,7 +195,8 @@ public static class V3ArtifactSeeder
opcUaPublishActor: opcUaPublishActor,
virtualTagEvaluator: NullVirtualTagEvaluator.Instance,
virtualTagHostOverride: vtHost.Ref,
reprovisionBackoff: reprovisionBackoff));
reprovisionBackoff: reprovisionBackoff,
commsRecoveryAuditDelay: commsRecoveryAuditDelay));
return (actor, mux, coordinator);
}
@@ -0,0 +1,197 @@
using Akka.Actor;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa;
using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness;
namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.OpcUa;
/// <summary>
/// <see cref="OpcUaPublishActor.AttributeStatusBatchUpdate"/> — the status-only route to the sink
/// that the comms-loss fan-out uses (archreview 03/S15). It must reach
/// <see cref="IOpcUaAddressSpaceSink.WriteStatuses"/> as exactly one call, never decompose into
/// value writes (which would null every node's value), and survive a throwing sink.
/// </summary>
public sealed class OpcUaPublishActorStatusBatchTests : RuntimeActorTestBase
{
/// <summary>
/// One message, one <c>WriteStatuses</c> call carrying every item verbatim — and <b>zero</b> value
/// writes. The second half is load-bearing: a handler that fell through to <c>WriteValues</c> (or
/// to the interface's default, which loops <c>WriteValue</c> with a null value) would satisfy "the
/// status arrived" while erasing the last known value the annotation exists to keep.
/// </summary>
[Fact]
public void Status_batch_reaches_sink_as_one_WriteStatuses_call()
{
var sink = new RecordingSink();
var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(sink: sink));
var t0 = new DateTime(2026, 8, 20, 10, 0, 0, DateTimeKind.Utc);
var items = new[]
{
new StatusWrite("Plant/Modbus/dev1/Speed", OpcUaStatus.BadNoCommunication, t0, AddressSpaceRealm.Raw),
new StatusWrite("AreaA/Line1/Eq1/Speed", OpcUaStatus.BadNoCommunication, t0, AddressSpaceRealm.Uns),
};
actor.Tell(new OpcUaPublishActor.AttributeStatusBatchUpdate(items));
AwaitAssert(() => sink.StatusBatches.Count.ShouldBe(1), PresenceBudget);
var batch = sink.StatusBatches[0];
batch.Count.ShouldBe(2);
batch[0].ShouldBe(items[0]);
batch[1].ShouldBe(items[1]);
sink.ValueBatches.Count.ShouldBe(0, "a status annotation must never be routed as a value write");
sink.SingleWrites.Count.ShouldBe(0);
}
/// <summary>
/// A partial failure is metered honestly, exactly as the value batch is: the tally advances by the
/// items the sink DID apply, not by zero and not by the whole batch.
/// </summary>
[Fact]
public void Partial_failure_still_counts_the_items_that_landed()
{
var sink = new PartialFailureSink();
var actor = ActorOfAsTestActorRef<OpcUaPublishActor>(OpcUaPublishActor.PropsForTests(sink: sink));
var t0 = new DateTime(2026, 8, 20, 10, 0, 0, DateTimeKind.Utc);
actor.Tell(new OpcUaPublishActor.AttributeStatusBatchUpdate(new[]
{
new StatusWrite("Plant/Modbus/dev1/Speed", OpcUaStatus.BadNoCommunication, t0, AddressSpaceRealm.Raw),
new StatusWrite("", OpcUaStatus.BadNoCommunication, t0, AddressSpaceRealm.Raw),
new StatusWrite("Plant/Modbus/dev1/Temp", OpcUaStatus.BadNoCommunication, t0, AddressSpaceRealm.Raw),
}));
sink.Threw.ShouldBeTrue();
actor.UnderlyingActor.WriteCount.ShouldBe(2,
"the two items the sink DID apply must still be counted — not 0 (the general catch) and not 3");
}
/// <summary>
/// A throwing sink is swallowed and logged, so a node manager fault cannot restart the publish
/// actor and drop live traffic. As in the value-batch suite, survival alone proves nothing (Akka's
/// restart preserves the mailbox) — the load-bearing assertions are that no exception reaches
/// supervision and that the handler logs its own Warning.
/// </summary>
[Fact]
public void Sink_throw_is_swallowed_and_logged()
{
var sink = new ThrowOnWriteStatusesSink();
var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(sink: sink));
var t0 = new DateTime(2026, 8, 20, 10, 0, 0, DateTimeKind.Utc);
EventFilter.Exception<InvalidOperationException>().Expect(0, () =>
{
EventFilter.Warning(contains: "sink.WriteStatuses threw").ExpectOne(() =>
actor.Tell(new OpcUaPublishActor.AttributeStatusBatchUpdate(new[]
{
new StatusWrite("Plant/Modbus/dev1/Speed", OpcUaStatus.BadNoCommunication, t0, AddressSpaceRealm.Raw),
})));
// Survival: a follow-up on the untouched per-item value path lands only if the actor is pumping.
actor.Tell(new OpcUaPublishActor.AttributeValueUpdate(
"Plant/Modbus/dev1/Temp", 2, OpcUaStatus.Good, t0, AddressSpaceRealm.Raw));
AwaitAssert(() => sink.SingleWrites.Count.ShouldBe(1), PresenceBudget);
});
sink.Threw.ShouldBeTrue();
}
private void AwaitAssert(Action assertion, TimeSpan duration)
{
var deadline = DateTime.UtcNow + duration;
Exception? last = null;
while (DateTime.UtcNow < deadline)
{
try { assertion(); return; }
catch (Exception ex) { last = ex; Thread.Sleep(25); }
}
if (last is not null) throw last;
}
/// <summary>Records status batches, value batches and per-item value writes separately, so a handler
/// that routed a status annotation down the value path is distinguishable from one that did not.</summary>
private class RecordingSink : IOpcUaAddressSpaceSink
{
private readonly object _gate = new();
private readonly List<IReadOnlyList<StatusWrite>> _statusBatches = new();
private readonly List<IReadOnlyList<ValueWrite>> _valueBatches = new();
private readonly List<ValueWrite> _singles = new();
public IReadOnlyList<IReadOnlyList<StatusWrite>> StatusBatches { get { lock (_gate) return _statusBatches.ToArray(); } }
public IReadOnlyList<IReadOnlyList<ValueWrite>> ValueBatches { get { lock (_gate) return _valueBatches.ToArray(); } }
public IReadOnlyList<ValueWrite> SingleWrites { get { lock (_gate) return _singles.ToArray(); } }
public void WriteValue(string nodeId, object? value, uint statusCode, DateTime sourceTimestampUtc, AddressSpaceRealm realm = AddressSpaceRealm.Uns)
{
lock (_gate) _singles.Add(new ValueWrite(nodeId, value, statusCode, sourceTimestampUtc, realm));
}
public void WriteValues(IReadOnlyList<ValueWrite> values)
{
lock (_gate) _valueBatches.Add(values.ToArray());
}
// Overrides the interface's default (which loops WriteValue with a NULL value).
public virtual void WriteStatuses(IReadOnlyList<StatusWrite> statuses)
{
lock (_gate) _statusBatches.Add(statuses.ToArray());
}
public void WriteAlarmQuality(string alarmNodeId, uint statusCode, DateTime sourceTimestampUtc, AddressSpaceRealm realm) { }
public void WriteAlarmCondition(string alarmNodeId, AlarmConditionSnapshot state, DateTime occurredUtc, AddressSpaceRealm realm = AddressSpaceRealm.Uns) { }
public void MaterialiseAlarmCondition(string alarmNodeId, string equipmentNodeId, string displayName, string alarmType, int severity, AddressSpaceRealm realm, bool isNative = false) { }
public void EnsureFolder(string folderNodeId, string? parentNodeId, string displayName, AddressSpaceRealm realm = AddressSpaceRealm.Uns) { }
public void EnsureVariable(string variableNodeId, string? parentFolderNodeId, string displayName, string dataType, bool writable, AddressSpaceRealm realm, string? historianTagname = null, bool isArray = false, uint? arrayLength = null) { }
public void RebuildAddressSpace() { }
public void RaiseNodesAddedModelChange(string affectedNodeId, AddressSpaceRealm realm = AddressSpaceRealm.Uns) { }
public void WireAlarmNotifiers(string alarmNodeId, AddressSpaceRealm alarmRealm, IReadOnlyList<string> notifierFolderNodeIds, AddressSpaceRealm notifierFolderRealm) { }
public void AddReference(string sourceNodeId, AddressSpaceRealm sourceRealm, string targetNodeId, AddressSpaceRealm targetRealm, string referenceType = "Organizes") { }
}
/// <summary>Throws a PLAIN exception out of the status batch — drives the handler's general catch.</summary>
private sealed class ThrowOnWriteStatusesSink : RecordingSink
{
private int _threw;
public bool Threw => Volatile.Read(ref _threw) != 0;
public override void WriteStatuses(IReadOnlyList<StatusWrite> statuses)
{
Interlocked.Exchange(ref _threw, 1);
throw new InvalidOperationException("simulated status-write fault");
}
}
/// <summary>Models the real sink's PARTIAL failure: applies every item with a non-empty NodeId, then
/// reports the aggregate as a <see cref="ValueWriteBatchException"/>.</summary>
private sealed class PartialFailureSink : RecordingSink
{
private int _threw;
public bool Threw => Volatile.Read(ref _threw) != 0;
public override void WriteStatuses(IReadOnlyList<StatusWrite> statuses)
{
var applied = statuses.Where(s => !string.IsNullOrEmpty(s.NodeId)).ToArray();
var failedCount = statuses.Count - applied.Length;
if (failedCount == 0)
{
base.WriteStatuses(statuses);
return;
}
base.WriteStatuses(applied);
Interlocked.Exchange(ref _threw, 1);
throw new ValueWriteBatchException(
firstFailedNodeId: statuses.First(s => string.IsNullOrEmpty(s.NodeId)).NodeId,
failedCount: failedCount,
succeededCount: applied.Length,
firstFailure: new ArgumentException("empty NodeId"));
}
}
}
@@ -0,0 +1,91 @@
using Akka.Actor;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness;
using ZB.MOM.WW.OtOpcUa.Runtime.VirtualTags;
namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.VirtualTags;
/// <summary>
/// Archreview <b>05/STAB-19</b> — the adapter that carries dependency values from the per-node
/// dependency mux into an <see cref="IDependencyConsumer"/> (today: the Calculation driver).
///
/// <para><b>The defect these pin.</b> The adapter forwarded a hard-coded <c>statusCode: 0u</c> behind
/// a comment claiming "quality is stripped at the mux boundary". That was true once; since #478/#519
/// <see cref="VirtualTagActor.DependencyValueChanged"/> carries the source driver's full status code,
/// sub-code included. The consequence was that a comms-lost input reached the Calculation driver as
/// <em>arrived, Good, null</em> — indistinguishable from a real Good reading of null, which is how a
/// calc output kept publishing Good over a dead input.</para>
/// </summary>
public sealed class DependencyConsumerMuxAdapterTests : RuntimeActorTestBase
{
private const string TagA = "Plant/Modbus/dev1/a";
private const string TagB = "Plant/Modbus/dev1/b";
/// <summary>Single delivery: the status code reaches the consumer verbatim, sub-code and all.</summary>
[Fact]
public void A_single_dependency_value_carries_its_status_code_to_the_consumer()
{
var (consumer, adapter) = Spawn();
adapter.Tell(new VirtualTagActor.DependencyValueChanged(
TagA, Value: null, DateTime.UtcNow, OpcUaStatus.BadNoCommunication));
AwaitCondition(() => consumer.Received.Count == 1, PresenceBudget);
consumer.Received[0].Ref.ShouldBe(TagA);
consumer.Received[0].StatusCode.ShouldBe(OpcUaStatus.BadNoCommunication);
consumer.Received[0].StatusCode.ShouldNotBe(OpcUaStatus.Good,
"a Bad input forwarded as Good is the laundering 05/STAB-19 is about");
}
/// <summary>
/// Batched delivery goes through the same seam, so it must carry status too — the batch path is
/// what a real driver burst uses, and a fix applied only to the single-value path would be inert
/// in production.
/// </summary>
[Fact]
public void A_batched_delivery_carries_each_items_status_code()
{
var (consumer, adapter) = Spawn();
var ts = DateTime.UtcNow;
adapter.Tell(new VirtualTagActor.DependencyValueBatch(new[]
{
new VirtualTagActor.DependencyValueChanged(TagA, 1.0, ts, OpcUaStatus.Good),
new VirtualTagActor.DependencyValueChanged(TagB, null, ts, OpcUaStatus.BadNoCommunication),
}));
AwaitCondition(() => consumer.Received.Count == 2, PresenceBudget);
consumer.Received[0].StatusCode.ShouldBe(OpcUaStatus.Good);
consumer.Received[1].StatusCode.ShouldBe(OpcUaStatus.BadNoCommunication);
}
private (RecordingConsumer Consumer, IActorRef Adapter) Spawn()
{
var consumer = new RecordingConsumer();
var mux = CreateTestProbe();
var adapter = Sys.ActorOf(DependencyConsumerMuxAdapter.Props(consumer, mux.Ref));
mux.ExpectMsg<DependencyMuxActor.RegisterInterest>(PresenceBudget);
return (consumer, adapter);
}
private sealed class RecordingConsumer : IDependencyConsumer
{
private readonly List<(string Ref, object? Value, uint StatusCode)> _received = new();
private readonly object _gate = new();
public IReadOnlyList<(string Ref, object? Value, uint StatusCode)> Received
{
get { lock (_gate) return _received.ToArray(); }
}
public IReadOnlyCollection<string> DependencyRefs { get; } = [TagA, TagB];
public void OnDependencyValue(string fullReference, object? value, uint statusCode, DateTime timestampUtc)
{
lock (_gate) _received.Add((fullReference, value, statusCode));
}
}
}
@@ -0,0 +1,315 @@
using System.Diagnostics.Metrics;
using Akka.Actor;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.Engines;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Logging;
using ZB.MOM.WW.OtOpcUa.Commons.Observability;
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness;
using ZB.MOM.WW.OtOpcUa.Runtime.VirtualTags;
namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.VirtualTags;
/// <summary>
/// Archreview <b>02/S14</b> — what a virtual tag does with its dependencies' <b>quality</b>.
///
/// <para><b>The defect.</b> The actor stored dependency values only, evaluated unconditionally, and
/// the Roslyn read cache stamped every input Good — so a VT over a comms-lost input kept publishing
/// fresh-Good results, and a script author could not detect it because
/// <c>ctx.GetTag(path).StatusCode</c> read the fabricated Good. 02/S13 fixed "broken script looks
/// Good"; this is the same lie through the other door.</para>
///
/// <para><b>The policy</b> is the Calculation driver's and the alarm engine's, stated once in
/// <c>InputQuality</c>: hold on Bad and publish the worst input's sub-code; evaluate through
/// Uncertain and stamp the output with it. The R2-03 decisions are kept, not re-litigated —
/// transition-once, the inputs-ready gate, and force-republish on recovery.</para>
///
/// <para>⚠️ <b>Vacuity rule:</b> the recovery case computes the <b>identical</b> value it had before
/// the outage. A recovery with a different value passes against a value-only dedup key and is not the
/// guard — that dedup is exactly the "Bad forever" trap.</para>
/// </summary>
public sealed class VirtualTagActorInputQualityTests : RuntimeActorTestBase
{
private const uint Uncertain = 0x40000000u;
/// <summary>
/// A Bad input holds the evaluator and publishes that input's own sub-code, once, carrying the
/// last known value. The evaluator-call count is the load-bearing half: "the result is Bad" could
/// also be produced by running the script against a null and failing, which is the NRE storm this
/// replaces.
/// </summary>
[Fact]
public void A_bad_input_holds_evaluation_and_publishes_the_inputs_sub_code()
{
var parent = CreateTestProbe();
var evaluator = new CountingSumEvaluator();
var actor = parent.ChildActorOf(VirtualTagActor.Props(
"vt-1", "a + b", evaluator: evaluator, dependencyRefs: ["a", "b"]));
actor.Tell(new VirtualTagActor.DependencyValueChanged("a", 10, DateTime.UtcNow));
actor.Tell(new VirtualTagActor.DependencyValueChanged("b", 32, DateTime.UtcNow));
parent.ExpectMsg<VirtualTagActor.EvaluationResult>().Value.ShouldBe(10);
parent.ExpectMsg<VirtualTagActor.EvaluationResult>().Value.ShouldBe(42);
var callsBefore = evaluator.Calls;
actor.Tell(new VirtualTagActor.DependencyValueChanged("a", null, DateTime.UtcNow, OpcUaStatus.BadNoCommunication));
var bad = parent.ExpectMsg<VirtualTagActor.EvaluationResult>(PresenceBudget);
bad.StatusCode.ShouldBe(OpcUaStatus.BadNoCommunication);
bad.StatusCode.ShouldNotBe(OpcUaStatus.BadInternalError, "the script did not fail — its input did");
bad.Value.ShouldBe(42, "the last known value rides out with the Bad status");
evaluator.Calls.ShouldBe(callsBefore, "the script must NOT be run against a Bad input");
// Transition-once: the same Bad again changes nothing.
actor.Tell(new VirtualTagActor.DependencyValueChanged("a", null, DateTime.UtcNow, OpcUaStatus.BadNoCommunication));
parent.ExpectNoMsg(TimeSpan.FromMilliseconds(300));
}
/// <summary>A different Bad sub-code republishes — the sub-code is the diagnostic.</summary>
[Fact]
public void A_different_bad_sub_code_is_republished()
{
var parent = CreateTestProbe();
var actor = parent.ChildActorOf(VirtualTagActor.Props(
"vt-1", "a + b", evaluator: new CountingSumEvaluator(), dependencyRefs: ["a"]));
actor.Tell(new VirtualTagActor.DependencyValueChanged("a", 10, DateTime.UtcNow));
parent.ExpectMsg<VirtualTagActor.EvaluationResult>();
actor.Tell(new VirtualTagActor.DependencyValueChanged("a", null, DateTime.UtcNow, OpcUaStatus.BadNoCommunication));
parent.ExpectMsg<VirtualTagActor.EvaluationResult>().StatusCode.ShouldBe(OpcUaStatus.BadNoCommunication);
actor.Tell(new VirtualTagActor.DependencyValueChanged("a", null, DateTime.UtcNow, OpcUaStatus.BadWaitingForInitialData));
parent.ExpectMsg<VirtualTagActor.EvaluationResult>().StatusCode.ShouldBe(OpcUaStatus.BadWaitingForInitialData);
}
/// <summary>
/// <b>The vacuity-sensitive assertion.</b> The input recovers holding the value it had before, so
/// the computed result is identical to the pre-outage one — and it must still be published Good,
/// because the node is currently Bad and only a publish can clear it.
/// </summary>
[Fact]
public void An_identical_value_is_republished_when_recovering_from_a_bad_input()
{
var parent = CreateTestProbe();
var actor = parent.ChildActorOf(VirtualTagActor.Props(
"vt-1", "a", evaluator: new CountingSumEvaluator(), dependencyRefs: ["a"]));
actor.Tell(new VirtualTagActor.DependencyValueChanged("a", 10, DateTime.UtcNow));
parent.ExpectMsg<VirtualTagActor.EvaluationResult>().Value.ShouldBe(10);
actor.Tell(new VirtualTagActor.DependencyValueChanged("a", null, DateTime.UtcNow, OpcUaStatus.BadNoCommunication));
parent.ExpectMsg<VirtualTagActor.EvaluationResult>().StatusCode.ShouldBe(OpcUaStatus.BadNoCommunication);
// Same value as before the outage.
actor.Tell(new VirtualTagActor.DependencyValueChanged("a", 10, DateTime.UtcNow));
var recovered = parent.ExpectMsg<VirtualTagActor.EvaluationResult>(PresenceBudget);
recovered.Value.ShouldBe(10, "the same value as before — that is the point");
recovered.StatusCode.ShouldBe(OpcUaStatus.Good);
}
/// <summary>
/// An Uncertain input is usable: the script runs and the output inherits the Uncertain code. The
/// script also sees the real code — a capturing evaluator asserts it, because a VT author writing
/// a quality-aware script is the reason the read cache had to stop fabricating Good.
/// </summary>
[Fact]
public void An_uncertain_input_is_evaluated_and_stamps_the_output_uncertain()
{
var parent = CreateTestProbe();
var evaluator = new CountingSumEvaluator();
var actor = parent.ChildActorOf(VirtualTagActor.Props(
"vt-1", "a", evaluator: evaluator, dependencyRefs: ["a"]));
actor.Tell(new VirtualTagActor.DependencyValueChanged("a", 10, DateTime.UtcNow));
parent.ExpectMsg<VirtualTagActor.EvaluationResult>();
actor.Tell(new VirtualTagActor.DependencyValueChanged("a", 11, DateTime.UtcNow, Uncertain));
var result = parent.ExpectMsg<VirtualTagActor.EvaluationResult>(PresenceBudget);
result.Value.ShouldBe(11, "the script DID run — Uncertain is usable");
result.StatusCode.ShouldBe(Uncertain);
evaluator.LastSeenStatus.ShouldBe(Uncertain, "the script must observe the input's REAL status");
}
/// <summary>
/// The R2-03 inputs-ready gate is preserved: with one declared dependency never pushed, a Bad
/// arrival on the other publishes nothing — cold start is not a fault.
/// </summary>
[Fact]
public void Nothing_is_published_while_a_declared_dependency_has_never_arrived()
{
var parent = CreateTestProbe();
var actor = parent.ChildActorOf(VirtualTagActor.Props(
"vt-1", "a + b", evaluator: new CountingSumEvaluator(), dependencyRefs: ["a", "b"]));
actor.Tell(new VirtualTagActor.DependencyValueChanged("a", null, DateTime.UtcNow, OpcUaStatus.BadNoCommunication));
parent.ExpectNoMsg(TimeSpan.FromMilliseconds(400));
}
/// <summary>
/// The <c>inputs-bad</c> outcome counter must tick only when a tag actually <b>reports</b> Bad
/// inputs — not on every warm-up trigger, where a declared dependency simply has not arrived yet.
/// Metering above the readiness gate turns ordinary cold start into a stream of "inputs bad"
/// measurements, which is exactly the noise that makes an operator stop trusting the counter.
/// <para>The absence half is paired with a positive control in the same test — the counter DOES
/// tick once the tag reports — so "no measurements" can never pass for want of a working listener.</para>
/// </summary>
[Fact]
public void The_inputs_bad_counter_does_not_tick_during_warm_up()
{
var parent = CreateTestProbe();
var actor = parent.ChildActorOf(VirtualTagActor.Props(
"vt-1", "a + b", evaluator: new CountingSumEvaluator(), dependencyRefs: ["a", "b"]));
using var meter = new OutcomeRecorder();
// Warm-up: "b" has never arrived, so this is cold start, not a fault.
actor.Tell(new VirtualTagActor.DependencyValueChanged("a", null, DateTime.UtcNow, OpcUaStatus.BadNoCommunication));
parent.ExpectNoMsg(TimeSpan.FromMilliseconds(400));
meter.Count("inputs-bad").ShouldBe(0, "warm-up is not an inputs-bad report");
// Positive control: once every dep has arrived, the tag reports — and the counter ticks.
actor.Tell(new VirtualTagActor.DependencyValueChanged("b", 32, DateTime.UtcNow));
parent.ExpectMsg<VirtualTagActor.EvaluationResult>(PresenceBudget)
.StatusCode.ShouldBe(OpcUaStatus.BadNoCommunication);
AwaitCondition(() => meter.Count("inputs-bad") == 1, TimeSpan.FromSeconds(3));
}
/// <summary>
/// A deploy's <c>ReassertValue</c> re-emits the LAST PUBLISHED status, so a VT whose input is
/// comms-lost re-asserts <c>BadNoCommunication</c> onto the freshly-materialised node rather than
/// Good. Re-asserting Good here would re-introduce the stale-Good lie at every deploy.
/// </summary>
[Fact]
public void Reassert_carries_the_last_published_status()
{
var parent = CreateTestProbe();
var actor = parent.ChildActorOf(VirtualTagActor.Props(
"vt-1", "a", evaluator: new CountingSumEvaluator(), dependencyRefs: ["a"]));
actor.Tell(new VirtualTagActor.DependencyValueChanged("a", 10, DateTime.UtcNow));
parent.ExpectMsg<VirtualTagActor.EvaluationResult>();
actor.Tell(new VirtualTagActor.DependencyValueChanged("a", null, DateTime.UtcNow, OpcUaStatus.BadNoCommunication));
parent.ExpectMsg<VirtualTagActor.EvaluationResult>();
actor.Tell(VirtualTagActor.ReassertValue.Instance);
var reassert = parent.ExpectMsg<VirtualTagActor.EvaluationResult>(PresenceBudget);
reassert.IsReassert.ShouldBeTrue();
reassert.StatusCode.ShouldBe(OpcUaStatus.BadNoCommunication);
reassert.Value.ShouldBe(10);
}
/// <summary>
/// The batched mux delivery goes through the same gate: a burst whose worst item is Bad holds the
/// script and publishes that sub-code. A fix applied only to the single-value path would be inert
/// in production, where the mux batches.
/// </summary>
[Fact]
public void A_batch_containing_a_bad_item_holds_evaluation_too()
{
var parent = CreateTestProbe();
var evaluator = new CountingSumEvaluator();
var actor = parent.ChildActorOf(VirtualTagActor.Props(
"vt-1", "a + b", evaluator: evaluator, dependencyRefs: ["a", "b"]));
var ts = DateTime.UtcNow;
actor.Tell(new VirtualTagActor.DependencyValueBatch(new[]
{
new VirtualTagActor.DependencyValueChanged("a", 10, ts),
new VirtualTagActor.DependencyValueChanged("b", 32, ts),
}));
parent.ExpectMsg<VirtualTagActor.EvaluationResult>().Value.ShouldBe(42);
var callsBefore = evaluator.Calls;
actor.Tell(new VirtualTagActor.DependencyValueBatch(new[]
{
new VirtualTagActor.DependencyValueChanged("a", null, ts, OpcUaStatus.BadNoCommunication),
new VirtualTagActor.DependencyValueChanged("b", 33, ts),
}));
parent.ExpectMsg<VirtualTagActor.EvaluationResult>(PresenceBudget)
.StatusCode.ShouldBe(OpcUaStatus.BadNoCommunication);
evaluator.Calls.ShouldBe(callsBefore);
}
/// <summary>
/// Counts <c>otopcua.virtualtag.eval</c> measurements by their <c>outcome</c> tag.
/// <para>A <see cref="MeterListener"/> sees every measurement in the process, and this counter is
/// hot across the assembly — so the assertions above are scoped to the <c>inputs-bad</c> outcome,
/// which only this suite produces (nothing else drives a Bad dependency), and the absence
/// assertion is paired with a positive control rather than standing alone.</para>
/// </summary>
private sealed class OutcomeRecorder : IDisposable
{
private readonly MeterListener _listener;
private readonly List<string> _outcomes = new();
private readonly object _gate = new();
public OutcomeRecorder()
{
_listener = new MeterListener
{
InstrumentPublished = (instrument, listener) =>
{
if (instrument.Meter.Name == OtOpcUaTelemetry.MeterName
&& instrument.Name == "otopcua.virtualtag.eval")
listener.EnableMeasurementEvents(instrument);
},
};
_listener.SetMeasurementEventCallback<long>((_, _, tags, _) =>
{
foreach (var tag in tags)
{
if (tag.Key != "outcome" || tag.Value is not string outcome) continue;
lock (_gate) _outcomes.Add(outcome);
}
});
_listener.Start();
}
public int Count(string outcome)
{
lock (_gate) return _outcomes.Count(o => o == outcome);
}
public void Dispose() => _listener.Dispose();
}
/// <summary>
/// Sums the dependency values, counts its invocations, and records the status code the actor
/// handed it for "a" — so a test can assert both that the script did NOT run and that, when it
/// did, it saw the input's real quality.
/// </summary>
private sealed class CountingSumEvaluator : IVirtualTagEvaluator
{
private int _calls;
public int Calls => Volatile.Read(ref _calls);
public uint? LastSeenStatus { get; private set; }
public VirtualTagEvalResult Evaluate(
string virtualTagId, string expression, IReadOnlyDictionary<string, object?> dependencies)
{
Interlocked.Increment(ref _calls);
return VirtualTagEvalResult.Ok(Sum(dependencies.Values));
}
public VirtualTagEvalResult Evaluate(
string virtualTagId, string expression, IReadOnlyDictionary<string, DataValueSnapshot> dependencies)
{
Interlocked.Increment(ref _calls);
LastSeenStatus = dependencies.TryGetValue("a", out var a) ? a.StatusCode : null;
return VirtualTagEvalResult.Ok(Sum(dependencies.Values.Select(v => v.Value)));
}
private static object Sum(IEnumerable<object?> values) =>
values.Where(v => v is not null).Sum(v => Convert.ToInt32(v));
}
}
@@ -201,6 +201,38 @@ public sealed class VirtualTagHostActorTests : RuntimeActorTestBase
value.ServerTimestampUtc.ShouldBe(ts);
}
/// <summary>
/// Archreview 02/S14 — a result carrying a propagated INPUT status (<c>BadNoCommunication</c>,
/// because the VT held evaluation on a comms-lost dependency) is bridged verbatim to the publish
/// actor <b>and historized with that status</b>, the same way 02/S13's script-failure Bad is.
/// A bridge that special-cased status — or historized it as Good — would put the stale-Good lie
/// back one hop downstream of the fix.
/// </summary>
[Fact]
public void A_result_carrying_a_propagated_input_status_is_bridged_and_historized_with_it()
{
const uint badNoCommunication = 0x80310000u;
var publish = CreateTestProbe();
var writer = new CapturingHistoryWriter();
var host = Sys.ActorOf(VirtualTagHostActor.Props(publish.Ref, mux: null, new StubEvaluator(), writer));
host.Tell(new VirtualTagHostActor.ApplyVirtualTags(
new[] { PlanH("vt-1", "eq-1", "speed-rpm", historize: true) }));
var ts = new DateTime(2026, 8, 20, 12, 0, 0, DateTimeKind.Utc);
host.Tell(new VirtualTagActor.EvaluationResult(
"vt-1", 42.0, ts, CorrelationId.NewId(), badNoCommunication));
var update = publish.ExpectMsg<OpcUaPublishActor.AttributeValueUpdate>();
update.NodeId.ShouldBe("eq-1/speed-rpm");
update.StatusCode.ShouldBe(badNoCommunication);
update.Value.ShouldBe(42.0, "the last known value rides out under the Bad status");
AwaitAssert(() => writer.Calls.Count.ShouldBe(1));
writer.Calls.TryPeek(out var captured).ShouldBeTrue();
captured.Item2.StatusCode.ShouldBe(badNoCommunication);
}
/// <summary>H5c: a result for a vtag whose plan has Historize=false is NOT recorded — the writer
/// is never called — though the value is still published.</summary>
[Fact]