feat(opcua): IOpcUaAddressSpaceSink.WriteStatuses — status-only batch write retaining the node value (03/S15)

This commit is contained in:
Joseph Doherty
2026-08-20 10:00:47 -04:00
parent 6874a562f3
commit ec5b48bfcd
7 changed files with 398 additions and 1 deletions
@@ -651,3 +651,21 @@ test could only pass once the production change landed.
- Docs: `docs/drivers/Mqtt.md` §"Quality on connection loss and re-subscribe" (links the contract in
`DriverLifecycle.md`, records the two distinct "not communicating" codes and the operator-visible
behaviour change).
### T6 — `IOpcUaAddressSpaceSink.WriteStatuses` (interface default + node manager + both sinks)
- RED (the guard, first): adding the member to the interface and running
`--filter "…DeferredSinkForwardingReflectionTests"` ⇒ **`Every_forwarding_interface_method_reaches_the_inner_sink`
FAILED**. Note the failure surfaced as `System.ArgumentNullException : Value cannot be null. (Parameter
'statuses')` rather than the guard's "did NOT forward" message — the reflection harness passes `null`
for an `IReadOnlyList<>` parameter, so the *interface default* ran on the wrapper and threw. Red for
the right reason (the member was not forwarded); green once `DeferredAddressSpaceSink.WriteStatuses`
forwards. This is the memory'd "Deferred sink doesn't forward the new member" trap, and here the
default is not merely slower — it writes a **null value**, erasing the very value the annotation exists
to keep.
- RED (node manager): `--filter "…OtOpcUaNodeManagerWriteStatusesTests"` ⇒ **build failure**, `CS1061:
'OtOpcUaNodeManager' does not contain a definition for 'WriteStatuses'` (×5).
- GREEN: 5/5 new; full `OpcUaServer.Tests` 495 passed / 1 skipped; `Commons.Tests` 348/348.
- An unmaterialised NodeId is **skipped and not counted as a failure** (Debug tally via the SDK static
trace — the manager has no `ILogger`): the caller's per-driver index racing a surgical remove is an
expected shape, not a fault.
@@ -41,7 +41,7 @@
{
"id": "T6",
"subject": "IOpcUaAddressSpaceSink.WriteStatuses (StatusWrite, default impl) + OtOpcUaNodeManager status-only batch retaining value + SdkAddressSpaceSink/DeferredAddressSpaceSink forwards (03/S15)",
"status": "pending",
"status": "done",
"blockedBy": []
},
{
@@ -32,6 +32,14 @@ public sealed class DeferredAddressSpaceSink : IOpcUaAddressSpaceSink, ISurgical
// (actors inject THIS wrapper, not the inner SdkAddressSpaceSink) — same trap as WireAlarmNotifiers above.
public void WriteValues(IReadOnlyList<ValueWrite> values) => _inner.WriteValues(values);
/// <inheritdoc />
// Forward the status-only batch (03/S15). Unforwarded, the interface's DEFAULT runs here instead —
// and that default writes a NULL value, so a comms-loss annotation would erase the last known value
// of every node on the driver rather than retaining it, on every driver-role host (actors inject
// THIS wrapper, not the inner SdkAddressSpaceSink). Same trap as WriteValues / WireAlarmNotifiers
// above, except this one degrades to wrong data rather than to slow.
public void WriteStatuses(IReadOnlyList<StatusWrite> statuses) => _inner.WriteStatuses(statuses);
/// <inheritdoc />
public void WriteAlarmCondition(string alarmNodeId, AlarmConditionSnapshot state, DateTime sourceTimestampUtc, AddressSpaceRealm realm)
=> _inner.WriteAlarmCondition(alarmNodeId, state, sourceTimestampUtc, realm);
@@ -42,6 +42,31 @@ public interface IOpcUaAddressSpaceSink
}
}
/// <summary>Batched <b>status-only</b> write: each node keeps its CURRENT value and only its
/// StatusCode + SourceTimestamp change. This is the comms-loss annotation for value variables
/// (archreview 03/S15) — a driver that has stopped communicating leaves the node holding its last
/// known value (the Part 8 convention SCADA clients expect) under a
/// <see cref="OpcUaStatus.BadNoCommunication"/> status, rather than nulling it.
/// <para>A sink that HOLDS the node's value must retain it — the SDK-backed sink does, because the
/// node itself is where the value lives. This DEFAULT implementation cannot: the host holds no value
/// cache (v3's single-source design), so it degrades to <see cref="WriteValue"/> with a null value,
/// which is the honest statement for a value-less sink and keeps every fake compiling unchanged.</para>
/// <para>Same per-item isolation contract as <see cref="WriteValues"/>. An unknown or
/// unmaterialised NodeId is a <b>no-op, never a create</b>: a status carries nothing to create a node
/// from, and the caller's index can legitimately name a node a surgical remove has just dropped.</para></summary>
/// <param name="statuses">The status writes to apply, in order.</param>
/// <exception cref="ValueWriteBatchException">A batching implementation throws this when some — but not
/// necessarily all — items failed; every other item has been applied.</exception>
void WriteStatuses(IReadOnlyList<StatusWrite> statuses)
{
ArgumentNullException.ThrowIfNull(statuses);
for (var i = 0; i < statuses.Count; i++)
{
var s = statuses[i];
WriteValue(s.NodeId, null, s.StatusCode, s.SourceTimestampUtc, s.Realm);
}
}
/// <summary>Write an alarm-condition's full Part 9 state. When a real condition node has been
/// materialised for <paramref name="alarmNodeId"/> via <see cref="MaterialiseAlarmCondition"/>,
/// this projects the whole <see cref="AlarmConditionSnapshot"/>
@@ -190,6 +215,16 @@ public interface IOpcUaAddressSpaceSink
public readonly record struct ValueWrite(
string NodeId, object? Value, uint StatusCode, DateTime SourceTimestampUtc, AddressSpaceRealm Realm);
/// <summary>One status-only write within a <see cref="IOpcUaAddressSpaceSink.WriteStatuses"/> batch: the
/// node's current VALUE is retained and only <paramref name="StatusCode"/> +
/// <paramref name="SourceTimestampUtc"/> change.</summary>
/// <param name="NodeId">The OPC UA node id of the variable, within <paramref name="Realm"/>.</param>
/// <param name="StatusCode">The full status code, sub-code included, passed to the SDK verbatim (#519).</param>
/// <param name="SourceTimestampUtc">The source timestamp in UTC.</param>
/// <param name="Realm">Which of the two v3 namespaces <paramref name="NodeId"/> lives in.</param>
public readonly record struct StatusWrite(
string NodeId, uint StatusCode, DateTime SourceTimestampUtc, AddressSpaceRealm Realm);
/// <summary>Aggregate outcome of a partially-failed <see cref="IOpcUaAddressSpaceSink.WriteValues"/> batch:
/// every non-failing item HAS been applied; this carries what the caller needs to log and meter honestly.</summary>
/// <remarks>A batched sink must NOT abandon the tail of a batch when one item throws: the poll engine
@@ -640,6 +640,109 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
}
}
/// <summary>Apply a batch of <b>status-only</b> writes: each node keeps its CURRENT value and only its
/// StatusCode + Timestamp change (archreview 03/S15 — the comms-loss annotation for value variables).
/// Same shape as <see cref="WriteValues"/>: an outside-the-lock validate/resolve pre-pass, then
/// <see cref="MaxWritesPerLockAcquisition"/>-sized chunks under one <c>Lock</c> each, per-item isolation,
/// and a <see cref="ValueWriteBatchException"/> aggregate.
/// <para><b>It never creates a node.</b> A status carries nothing to create one from, and the caller's
/// per-driver ref index can legitimately name a node a surgical remove has just dropped — so an unknown
/// id is skipped, not materialised, and is not counted as a failure either (it is an expected race, not
/// a fault).</para></summary>
/// <param name="statuses">The status writes to apply, in order.</param>
/// <exception cref="ValueWriteBatchException">When one or more items failed validation or threw. Every
/// other item HAS been applied.</exception>
public void WriteStatuses(IReadOnlyList<StatusWrite> statuses)
{
ArgumentNullException.ThrowIfNull(statuses);
EnsureAddressSpaceCreated();
if (statuses.Count == 0) return;
var resolved = new ResolvedWrite[statuses.Count];
string? firstFailedNodeId = null;
Exception? firstFailure = null;
var failed = 0;
for (var i = 0; i < statuses.Count; i++)
{
var nodeId = statuses[i].NodeId;
if (string.IsNullOrEmpty(nodeId))
{
failed++;
firstFailedNodeId ??= nodeId ?? string.Empty;
firstFailure ??= new ArgumentException(
$"WriteStatuses item {i} has a null or empty NodeId.", nameof(statuses));
continue;
}
resolved[i] = new ResolvedWrite(
nodeId, NamespaceIndexForRealm(statuses[i].Realm), MapKey(statuses[i].Realm, nodeId), IsValid: true);
}
var succeeded = 0;
var skipped = 0;
for (var start = 0; start < statuses.Count; start += MaxWritesPerLockAcquisition)
{
var end = Math.Min(start + MaxWritesPerLockAcquisition, statuses.Count);
lock (Lock)
{
for (var i = start; i < end; i++)
{
if (!resolved[i].IsValid) continue;
var s = statuses[i];
try
{
if (WriteStatusLocked(resolved[i].Key, s.StatusCode, s.SourceTimestampUtc)) succeeded++;
else skipped++;
}
catch (Exception ex)
{
failed++;
firstFailedNodeId ??= s.NodeId;
firstFailure ??= ex;
}
}
}
}
if (skipped > 0)
{
// Debug, not Warning: the caller's index racing a surgical remove is expected, and a driver
// whose nodes were all just removed would otherwise log once per ref. This manager carries no
// ILogger, so log via the SDK's static trace (see ServeNodeAsync).
#pragma warning disable CS0618 // Type or member is obsolete
Utils.LogDebug(
"OtOpcUaNodeManager: WriteStatuses skipped {0} of {1} item(s) — node not materialised.",
skipped,
statuses.Count);
#pragma warning restore CS0618
}
if (failed > 0)
{
throw new ValueWriteBatchException(
firstFailedNodeId ?? string.Empty, failed, succeeded, firstFailure!);
}
}
/// <summary>The body of a single status-only write. <b>Caller MUST hold <c>Lock</c></b>.</summary>
/// <param name="key">The realm-qualified map key.</param>
/// <param name="statusCode">The full OPC UA status code, sub-code included.</param>
/// <param name="sourceTimestampUtc">The timestamp of the annotation in UTC.</param>
/// <returns><see langword="true"/> when the node existed and was stamped; <see langword="false"/> when
/// no such node is materialised (skipped — deliberately not a create; see <see cref="WriteStatuses"/>).</returns>
private bool WriteStatusLocked(RealmKey key, uint statusCode, DateTime sourceTimestampUtc)
{
if (!_variables.TryGetValue(key, out var variable)) return false;
// Value deliberately untouched — that IS the feature: the node keeps its last known value under a
// Bad status (OPC UA Part 8's last-known-value convention).
variable.StatusCode = new StatusCode(statusCode);
variable.Timestamp = sourceTimestampUtc;
variable.ClearChangeMasks(SystemContext, includeChildren: false);
return true;
}
/// <summary>One <see cref="WriteValues"/> item after the outside-the-lock pre-pass: its namespace index and
/// map key already resolved, or <paramref name="IsValid"/> false when the item failed validation.</summary>
/// <param name="NodeId">The item's bare <c>s=</c> identifier.</param>
@@ -31,6 +31,13 @@ public sealed class SdkAddressSpaceSink : IOpcUaAddressSpaceSink, ISurgicalAddre
public void WriteValues(IReadOnlyList<ValueWrite> values)
=> _nodeManager.WriteValues(values);
/// <inheritdoc />
/// <remarks>Overrides the interface's default on purpose, and here the default is not merely slower but
/// <b>wrong</b>: it writes a null value (a value-less sink has nothing else to write), which would erase
/// the last known value this annotation exists to preserve. The manager's own entry point retains it.</remarks>
public void WriteStatuses(IReadOnlyList<StatusWrite> statuses)
=> _nodeManager.WriteStatuses(statuses);
/// <inheritdoc />
public void WriteAlarmCondition(string alarmNodeId, AlarmConditionSnapshot state, DateTime sourceTimestampUtc, AddressSpaceRealm realm)
=> _nodeManager.WriteAlarmCondition(alarmNodeId, state, sourceTimestampUtc, realm);
@@ -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 */ }
}
}
}