diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/OpcUa/IOpcUaAddressSpaceSink.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/OpcUa/IOpcUaAddressSpaceSink.cs
index 35107c13..2e240a6d 100644
--- a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/OpcUa/IOpcUaAddressSpaceSink.cs
+++ b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/OpcUa/IOpcUaAddressSpaceSink.cs
@@ -25,8 +25,13 @@ public interface IOpcUaAddressSpaceSink
/// Write a batch of Variable values. Semantically identical to calling
/// per item; an SDK-backed sink overrides this to take its internal
/// lock ONCE per batch instead of once per value (the hot-path win). The default implementation
- /// loops so existing sinks and fakes behave unchanged.
+ /// loops so existing sinks and fakes behave unchanged.
+ /// A batching implementation MUST isolate per item — one bad item may not drop the tail of the
+ /// batch — and report the aggregate as a ; see that type for why.
+ /// The default loop above predates the contract and is only used by fakes / the no-op sink.
/// The value writes to apply, in order.
+ /// A batching implementation throws this when some — but not
+ /// necessarily all — items failed; every other item has been applied.
void WriteValues(IReadOnlyList values)
{
ArgumentNullException.ThrowIfNull(values);
@@ -185,6 +190,30 @@ public interface IOpcUaAddressSpaceSink
public readonly record struct ValueWrite(
string NodeId, object? Value, uint StatusCode, DateTime SourceTimestampUtc, AddressSpaceRealm Realm);
+/// Aggregate outcome of a partially-failed batch:
+/// every non-failing item HAS been applied; this carries what the caller needs to log and meter honestly.
+/// A batched sink must NOT abandon the tail of a batch when one item throws: the poll engine
+/// updates its last-value cache BEFORE raising, so a dropped item is never re-raised until that tag's value
+/// changes again — the node would serve a stale value indefinitely. Hence the sink isolates per item and
+/// reports the aggregate here rather than letting the first failure escape.
+/// The NodeId of the first item that failed (empty when that item's own
+/// NodeId was null/empty — the validation failure this type most often reports).
+/// How many items of the batch failed.
+/// How many items of the batch were applied.
+/// The exception the first failing item produced; the inner exception.
+public sealed class ValueWriteBatchException(string firstFailedNodeId, int failedCount, int succeededCount, Exception firstFailure)
+ : Exception($"WriteValues batch: {failedCount} of {failedCount + succeededCount} items failed (first: '{firstFailedNodeId}')", firstFailure)
+{
+ /// Gets the NodeId of the first item that failed.
+ public string FirstFailedNodeId { get; } = firstFailedNodeId;
+
+ /// Gets the number of items of the batch that failed.
+ public int FailedCount { get; } = failedCount;
+
+ /// Gets the number of items of the batch that were successfully applied.
+ public int SucceededCount { get; } = succeededCount;
+}
+
/// No-op sink. Bound by default so the actors are safe to run in dev / Mac /
/// integration tests without a real SDK behind them.
public sealed class NullOpcUaAddressSpaceSink : IOpcUaAddressSpaceSink
diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs
index 351cb662..5c9c0643 100644
--- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs
+++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs
@@ -503,8 +503,10 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
///
/// The node identifier of the variable.
/// The new value to write.
- /// The OPC UA quality status code.
+ /// The full OPC UA status code, sub-code included (#519 — passed to the SDK verbatim).
/// The timestamp of the value in UTC.
+ /// Which of the two v3 namespaces the lives in; the
+ /// namespace index is resolved from this rather than parsed out of the id string.
public void WriteValue(string nodeId, object? value, uint statusCode, DateTime sourceTimestampUtc, AddressSpaceRealm realm)
{
ArgumentException.ThrowIfNullOrEmpty(nodeId);
@@ -516,28 +518,105 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
WriteValueLocked(nodeId, ns, key, value, statusCode, sourceTimestampUtc);
}
- /// Batched : one Lock acquisition for the whole batch instead of
- /// one per value — the publish path's per-value lock take was the steady-state throughput ceiling.
- /// Per-item semantics are identical to (create-on-first-write, #519 verbatim
- /// status codes, ClearChangeMasks per node).
+ /// The most items applies under a SINGLE Lock acquisition before
+ /// releasing and re-taking it.
+ /// WHY a cap at all: the lock take is what batching amortises, and the amortisation curve is
+ /// brutally front-loaded — going from one take per value to one per 512 already removes >99.8% of the
+ /// takes, so an unbounded hold buys essentially nothing more. What it DOES buy is a stall: the manager-wide
+ /// Lock is the same lock every SDK service thread (Read / Browse / subscription publish /
+ /// ConditionRefresh) contends for, and the first poll cycle force-raises EVERY tag (the initial-data push),
+ /// so batch #1 is a whole device × its UNS fan-out — thousands of CreateVariable calls, each mutating
+ /// the SDK address space, under one acquisition. 512 is therefore "effectively unbounded for the win,
+ /// bounded for the stall": chunking hands the SDK a scheduling point between chunks.
+ private const int MaxWritesPerLockAcquisition = 512;
+
+ /// Batched : one Lock acquisition per chunk of
+ /// items instead of one per value — the publish path's per-value
+ /// lock take was the steady-state throughput ceiling. Per-item semantics are identical to
+ /// (create-on-first-write, #519 verbatim status codes, ClearChangeMasks per node).
+ /// Items are isolated. A bad item (empty NodeId, or a throw out of the SDK) does not abandon
+ /// the rest of the batch: it is recorded and the loop continues, and the aggregate surfaces afterwards as
+ /// a . Dropping the tail would be silent and durable — the poll
+ /// engine updates its last-value cache BEFORE raising, so a dropped item is not re-raised until that tag's
+ /// value changes again, leaving the node stale indefinitely.
/// The value writes to apply, in order.
+ /// When one or more items failed. Every other item HAS been
+ /// applied; the exception carries the first failing NodeId plus the failed/succeeded split so the caller
+ /// can log and meter honestly.
public void WriteValues(IReadOnlyList values)
{
ArgumentNullException.ThrowIfNull(values);
- if (values.Count == 0) return;
+ // Guard BEFORE the empty-batch early-return, so a pre-start call fails the same legible way every
+ // other mutator does regardless of what it happens to carry.
EnsureAddressSpaceCreated();
+ if (values.Count == 0) return;
- lock (Lock)
+ // Pre-pass, OUTSIDE the lock: validate each item and resolve its namespace index + map key into a
+ // scratch array. Both jobs belong here — validating outside keeps a malformed item from ever reaching
+ // the SDK, and resolving outside moves the MapKey string allocation (a NodeId + ToString per item)
+ // off the critical section entirely.
+ var resolved = new ResolvedWrite[values.Count];
+ string? firstFailedNodeId = null;
+ Exception? firstFailure = null;
+ var failed = 0;
+
+ for (var i = 0; i < values.Count; i++)
{
- for (var i = 0; i < values.Count; i++)
+ var nodeId = values[i].NodeId;
+ if (string.IsNullOrEmpty(nodeId))
{
- var v = values[i];
- WriteValueLocked(v.NodeId, NamespaceIndexForRealm(v.Realm), MapKey(v.Realm, v.NodeId),
- v.Value, v.StatusCode, v.SourceTimestampUtc);
+ failed++;
+ firstFailedNodeId ??= nodeId ?? string.Empty;
+ firstFailure ??= new ArgumentException(
+ $"WriteValues item {i} has a null or empty NodeId.", nameof(values));
+ continue;
+ }
+
+ resolved[i] = new ResolvedWrite(
+ nodeId, NamespaceIndexForRealm(values[i].Realm), MapKey(values[i].Realm, nodeId), IsValid: true);
+ }
+
+ var succeeded = 0;
+ for (var start = 0; start < values.Count; start += MaxWritesPerLockAcquisition)
+ {
+ var end = Math.Min(start + MaxWritesPerLockAcquisition, values.Count);
+ lock (Lock)
+ {
+ for (var i = start; i < end; i++)
+ {
+ if (!resolved[i].IsValid) continue;
+ var v = values[i];
+ try
+ {
+ WriteValueLocked(resolved[i].NodeId, resolved[i].NamespaceIndex, resolved[i].Key,
+ v.Value, v.StatusCode, v.SourceTimestampUtc);
+ succeeded++;
+ }
+ catch (Exception ex)
+ {
+ failed++;
+ firstFailedNodeId ??= v.NodeId;
+ firstFailure ??= ex;
+ }
+ }
}
}
+
+ if (failed > 0)
+ {
+ throw new ValueWriteBatchException(
+ firstFailedNodeId ?? string.Empty, failed, succeeded, firstFailure!);
+ }
}
+ /// One item after the outside-the-lock pre-pass: its namespace index and
+ /// map key already resolved, or false when the item failed validation.
+ /// The item's bare s= identifier.
+ /// The namespace index resolved from the item's realm.
+ /// The realm-qualified map key.
+ /// False when the item failed the pre-pass and must be skipped.
+ private readonly record struct ResolvedWrite(string NodeId, ushort NamespaceIndex, string Key, bool IsValid);
+
/// The body of a single value write. Caller MUST hold Lock — this is factored out
/// solely so can apply a whole batch under one acquisition; it is byte-identical
/// to what has always done inside its own lock.
@@ -2647,7 +2726,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
/// Guard the address-space mutators against a too-early call. _root
/// is only assigned in , which the SDK invokes during
- /// StandardServer start; every public mutator (,
+ /// StandardServer start; every public mutator (, ,
/// , , ,
/// ) and assume it has run.
/// If one is invoked before the server has started (a sink wired or a publish replayed before
diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/OpcUa/OpcUaPublishActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/OpcUa/OpcUaPublishActor.cs
index 631feed5..78bceace 100644
--- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/OpcUa/OpcUaPublishActor.cs
+++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/OpcUa/OpcUaPublishActor.cs
@@ -339,7 +339,12 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
/// Mirrors for a whole burst: ONE
/// call, but the write tally and the
/// otopcua.opcua.sink.write meter still advance PER ITEM — batching is a transport change, so
- /// dashboards must not drop by the batch factor.
+ /// dashboards must not drop by the batch factor.
+ /// A is a PARTIAL failure: the sink applied every item it
+ /// could and is telling us which one it couldn't. So it is caught ahead of the general catch to (a) name
+ /// the offending node in the log — a bare "the batch threw" cannot be acted on — and (b) still count the
+ /// items that DID land, keeping the per-item counting invariant honest instead of silently under-reporting
+ /// a whole batch because one item was bad.
private void HandleAttributeBatchUpdate(AttributeValueBatchUpdate msg)
{
try
@@ -348,6 +353,15 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
Interlocked.Add(ref _writes, msg.Items.Count);
OtOpcUaTelemetry.OpcUaSinkWrite.Add(msg.Items.Count, new KeyValuePair("kind", "value"));
}
+ catch (ValueWriteBatchException ex)
+ {
+ _log.Warning(ex,
+ "OpcUaPublish: sink.WriteValues partially failed for a {Count}-item batch — {Failed} failed " +
+ "(first: {Node}), {Succeeded} applied",
+ msg.Items.Count, ex.FailedCount, ex.FirstFailedNodeId, ex.SucceededCount);
+ Interlocked.Add(ref _writes, ex.SucceededCount);
+ OtOpcUaTelemetry.OpcUaSinkWrite.Add(ex.SucceededCount, new KeyValuePair("kind", "value"));
+ }
catch (Exception ex)
{
_log.Warning(ex, "OpcUaPublish: sink.WriteValues threw for a {Count}-item batch", msg.Items.Count);
diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerBatchWriteTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerBatchWriteTests.cs
index ace33503..6c3e9c19 100644
--- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerBatchWriteTests.cs
+++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerBatchWriteTests.cs
@@ -7,9 +7,10 @@ using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests;
///
-/// Value-pipeline batching (Task 6) — applies a whole
-/// batch under ONE acquisition of the manager-wide SDK Lock instead of one per value (the
-/// per-value take was the steady-state throughput ceiling on the publish path).
+/// Value-pipeline batching (Task 6) — applies a batch under
+/// one acquisition of the manager-wide SDK Lock per 512-item chunk instead of one per value (the
+/// per-value take was the steady-state throughput ceiling on the publish path; the chunk cap keeps the
+/// first poll cycle's force-raise-everything batch from stalling every SDK service thread).
/// These tests pin the observable contract rather than the locking: every item lands with
/// its own Value / StatusCode / Timestamp, an item for an undeclared NodeId is lazily created exactly
/// as does, and an empty batch does nothing at all.
@@ -35,78 +36,174 @@ public sealed class NodeManagerBatchWriteTests : IDisposable
public async Task WriteValues_lands_every_value()
{
var (host, server) = await BootAsync();
- var nm = server.NodeManager!;
-
- foreach (var name in new[] { "a", "b", "c" })
+ try
{
- nm.EnsureVariable($"eq-1/{name}", parentFolderNodeId: null, displayName: name, dataType: "Double",
- writable: false, realm: AddressSpaceRealm.Uns);
+ var nm = server.NodeManager!;
+
+ foreach (var name in new[] { "a", "b", "c" })
+ {
+ nm.EnsureVariable($"eq-1/{name}", parentFolderNodeId: null, displayName: name, dataType: "Double",
+ writable: false, realm: AddressSpaceRealm.Uns);
+ }
+
+ var t1 = new DateTime(2026, 8, 15, 10, 0, 0, DateTimeKind.Utc);
+ var t2 = new DateTime(2026, 8, 15, 10, 0, 1, DateTimeKind.Utc);
+ var t3 = new DateTime(2026, 8, 15, 10, 0, 2, DateTimeKind.Utc);
+
+ nm.WriteValues(new List
+ {
+ new("eq-1/a", 1.5d, OpcUaStatus.Good, t1, AddressSpaceRealm.Uns),
+ new("eq-1/b", null, OpcUaStatus.BadNoCommunication, t2, AddressSpaceRealm.Uns),
+ new("eq-1/c", 3.5d, 0x80740000u /* BadTypeMismatch */, t3, AddressSpaceRealm.Uns),
+ });
+
+ var a = nm.TryGetVariable("eq-1/a");
+ a.ShouldNotBeNull();
+ a!.Value.ShouldBe(1.5d);
+ a.StatusCode.ShouldBe((StatusCode)StatusCodes.Good);
+ a.Timestamp.ShouldBe(t1);
+
+ var b = nm.TryGetVariable("eq-1/b");
+ b.ShouldNotBeNull();
+ b!.Value.ShouldBeNull();
+ b.StatusCode.ShouldBe((StatusCode)StatusCodes.BadNoCommunication);
+ b.StatusCode.ShouldNotBe((StatusCode)StatusCodes.Bad,
+ "the batched path must hand the status sub-code to the SDK verbatim (#519)");
+ b.Timestamp.ShouldBe(t2);
+
+ var c = nm.TryGetVariable("eq-1/c");
+ c.ShouldNotBeNull();
+ c!.Value.ShouldBe(3.5d);
+ c.StatusCode.ShouldBe((StatusCode)StatusCodes.BadTypeMismatch);
+ c.Timestamp.ShouldBe(t3);
}
-
- var t1 = new DateTime(2026, 8, 15, 10, 0, 0, DateTimeKind.Utc);
- var t2 = new DateTime(2026, 8, 15, 10, 0, 1, DateTimeKind.Utc);
- var t3 = new DateTime(2026, 8, 15, 10, 0, 2, DateTimeKind.Utc);
-
- nm.WriteValues(new List
+ finally
{
- new("eq-1/a", 1.5d, OpcUaStatus.Good, t1, AddressSpaceRealm.Uns),
- new("eq-1/b", null, OpcUaStatus.BadNoCommunication, t2, AddressSpaceRealm.Uns),
- new("eq-1/c", 3.5d, 0x80740000u /* BadTypeMismatch */, t3, AddressSpaceRealm.Uns),
- });
-
- var a = nm.TryGetVariable("eq-1/a");
- a.ShouldNotBeNull();
- a!.Value.ShouldBe(1.5d);
- a.StatusCode.ShouldBe((StatusCode)StatusCodes.Good);
- a.Timestamp.ShouldBe(t1);
-
- var b = nm.TryGetVariable("eq-1/b");
- b.ShouldNotBeNull();
- b!.Value.ShouldBeNull();
- b.StatusCode.ShouldBe((StatusCode)StatusCodes.BadNoCommunication);
- b.StatusCode.ShouldNotBe((StatusCode)StatusCodes.Bad,
- "the batched path must hand the status sub-code to the SDK verbatim (#519)");
- b.Timestamp.ShouldBe(t2);
-
- var c = nm.TryGetVariable("eq-1/c");
- c.ShouldNotBeNull();
- c!.Value.ShouldBe(3.5d);
- c.StatusCode.ShouldBe((StatusCode)StatusCodes.BadTypeMismatch);
- c.Timestamp.ShouldBe(t3);
-
- await host.DisposeAsync();
+ await host.DisposeAsync();
+ }
}
/// A batch item naming a NodeId that was never declared lazily creates the variable, exactly
/// as the single-value create-on-first-write path does —
- /// and it does so per realm, so a Raw and a UNS item sharing an s= id create two nodes.
+ /// and it does so PER REALM. Both items here carry the SAME s= id and differ only by realm, so a
+ /// key that ignored the realm would create one node and have the second item overwrite the first. The
+ /// assertion is therefore two nodes with DISTINCT namespace indices, not merely "both reads returned
+ /// something" (which a realm-blind map satisfies, since the shared key resolves for either lookup).
[Fact]
public async Task WriteValues_creates_missing_variables()
{
var (host, server) = await BootAsync();
- var nm = server.NodeManager!;
-
- nm.TryGetVariable("eq-1/lazy").ShouldBeNull();
- nm.TryGetVariable("Plant/dev1/lazy", AddressSpaceRealm.Raw).ShouldBeNull();
-
- var ts = new DateTime(2026, 8, 15, 11, 0, 0, DateTimeKind.Utc);
- nm.WriteValues(new List
+ try
{
- new("eq-1/lazy", 42, OpcUaStatus.Good, ts, AddressSpaceRealm.Uns),
- new("Plant/dev1/lazy", 7, OpcUaStatus.Good, ts, AddressSpaceRealm.Raw),
- });
+ var nm = server.NodeManager!;
- var uns = nm.TryGetVariable("eq-1/lazy");
- uns.ShouldNotBeNull();
- uns!.Value.ShouldBe(42);
- uns.StatusCode.ShouldBe((StatusCode)StatusCodes.Good);
- uns.Timestamp.ShouldBe(ts);
+ const string SharedId = "Shared/lazy/id";
+ nm.TryGetVariable(SharedId, AddressSpaceRealm.Uns).ShouldBeNull();
+ nm.TryGetVariable(SharedId, AddressSpaceRealm.Raw).ShouldBeNull();
- var raw = nm.TryGetVariable("Plant/dev1/lazy", AddressSpaceRealm.Raw);
- raw.ShouldNotBeNull();
- raw!.Value.ShouldBe(7);
+ var ts = new DateTime(2026, 8, 15, 11, 0, 0, DateTimeKind.Utc);
+ nm.WriteValues(new List
+ {
+ new(SharedId, 42, OpcUaStatus.Good, ts, AddressSpaceRealm.Uns),
+ new(SharedId, 7, OpcUaStatus.Good, ts, AddressSpaceRealm.Raw),
+ });
- await host.DisposeAsync();
+ var uns = nm.TryGetVariable(SharedId, AddressSpaceRealm.Uns);
+ uns.ShouldNotBeNull();
+ uns!.Value.ShouldBe(42);
+ uns.StatusCode.ShouldBe((StatusCode)StatusCodes.Good);
+ uns.Timestamp.ShouldBe(ts);
+
+ var raw = nm.TryGetVariable(SharedId, AddressSpaceRealm.Raw);
+ raw.ShouldNotBeNull();
+ raw!.Value.ShouldBe(7);
+
+ // Two DISTINCT nodes, one per namespace — the realm-blind failure mode collapses these to one.
+ raw.NodeId.NamespaceIndex.ShouldNotBe(uns.NodeId.NamespaceIndex);
+ raw.NodeId.NamespaceIndex.ShouldBe(nm.NamespaceIndexForRealm(AddressSpaceRealm.Raw));
+ uns.NodeId.NamespaceIndex.ShouldBe(nm.NamespaceIndexForRealm(AddressSpaceRealm.Uns));
+ }
+ finally
+ {
+ await host.DisposeAsync();
+ }
+ }
+
+ /// A bad item does NOT drop the tail of the batch. Item 2's NodeId is empty; items 1 and 3 must
+ /// still land, and the aggregate surfaces as a naming the offender
+ /// and the failed/succeeded split.
+ /// This is the durability-critical half: the poll engine updates its last-value cache BEFORE
+ /// raising, so an item dropped here is never re-raised until that tag's value changes again — the node
+ /// would serve a stale value indefinitely, with nothing in the log naming it.
+ [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, 15, 12, 0, 0, DateTimeKind.Utc);
+
+ var ex = Should.Throw(() => nm.WriteValues(new List
+ {
+ new("eq-iso/first", 1, OpcUaStatus.Good, ts, AddressSpaceRealm.Uns),
+ new(string.Empty, 2, OpcUaStatus.Good, ts, AddressSpaceRealm.Uns),
+ new("eq-iso/third", 3, OpcUaStatus.Good, ts, AddressSpaceRealm.Uns),
+ }));
+
+ ex.FailedCount.ShouldBe(1);
+ ex.SucceededCount.ShouldBe(2);
+ ex.FirstFailedNodeId.ShouldBe(string.Empty);
+
+ // The load-bearing half: BOTH good items applied — including the one AFTER the failure.
+ var first = nm.TryGetVariable("eq-iso/first");
+ first.ShouldNotBeNull();
+ first!.Value.ShouldBe(1);
+
+ var third = nm.TryGetVariable("eq-iso/third");
+ third.ShouldNotBeNull("the item after the failing one must not be dropped");
+ third!.Value.ShouldBe(3);
+ }
+ finally
+ {
+ await host.DisposeAsync();
+ }
+ }
+
+ /// A batch larger than MaxWritesPerLockAcquisition (512) spans several lock acquisitions;
+ /// chunking must be invisible in the outcome — every item still lands exactly once, with its own value.
+ /// Chunk boundaries are the natural off-by-one hazard, so the size is deliberately not a multiple of 512.
+ ///
+ [Fact]
+ public async Task A_batch_spanning_several_chunks_lands_every_item()
+ {
+ var (host, server) = await BootAsync();
+ try
+ {
+ var nm = server.NodeManager!;
+ var ts = new DateTime(2026, 8, 15, 13, 0, 0, DateTimeKind.Utc);
+
+ const int Count = 1300; // > 2 chunks of 512, and not a multiple of it
+ var items = new List(Count);
+ for (var i = 0; i < Count; i++)
+ {
+ items.Add(new ValueWrite($"eq-chunk/{i}", i, OpcUaStatus.Good, ts, AddressSpaceRealm.Uns));
+ }
+
+ nm.WriteValues(items);
+
+ nm.VariableCount.ShouldBe(Count);
+ foreach (var probe in new[] { 0, 511, 512, 1023, 1024, Count - 1 })
+ {
+ var v = nm.TryGetVariable($"eq-chunk/{probe}");
+ v.ShouldNotBeNull($"item {probe} (a chunk boundary) must have landed");
+ v!.Value.ShouldBe(probe);
+ }
+ }
+ finally
+ {
+ await host.DisposeAsync();
+ }
}
/// An empty batch neither throws nor touches the address space.
@@ -114,13 +211,18 @@ public sealed class NodeManagerBatchWriteTests : IDisposable
public async Task Empty_batch_is_a_no_op()
{
var (host, server) = await BootAsync();
- var nm = server.NodeManager!;
+ try
+ {
+ var nm = server.NodeManager!;
- var before = nm.VariableCount;
- Should.NotThrow(() => nm.WriteValues([]));
- nm.VariableCount.ShouldBe(before);
-
- await host.DisposeAsync();
+ var before = nm.VariableCount;
+ Should.NotThrow(() => nm.WriteValues([]));
+ nm.VariableCount.ShouldBe(before);
+ }
+ finally
+ {
+ await host.DisposeAsync();
+ }
}
private async Task<(OpcUaApplicationHost Host, OtOpcUaSdkServer Server)> BootAsync()
diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerPreStartGuardTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerPreStartGuardTests.cs
index 718aa73e..afcbea51 100644
--- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerPreStartGuardTests.cs
+++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerPreStartGuardTests.cs
@@ -9,7 +9,8 @@ namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests;
///
/// OpcUaServer-004 — a fresh whose CreateAddressSpace has NOT
/// yet run (i.e. the server has not started) has a null _root. Every public address-space mutator
-/// (, ,
+/// (, ,
+/// ,
/// , ,
/// ) must now fail with a legible
/// instead of a bare NRE out of ResolveParentFolder /
@@ -78,6 +79,28 @@ public sealed class NodeManagerPreStartGuardTests : IDisposable
}
}
+ /// The batched mutator guards the same way — and it does so BEFORE its empty-batch early-return,
+ /// so a pre-start caller gets the legible failure regardless of what the batch happens to carry (an empty
+ /// batch returning quietly would make the guard depend on payload rather than on server state).
+ [Fact]
+ public async Task WriteValues_before_CreateAddressSpace_throws_InvalidOperationException()
+ {
+ var (host, nm) = await BuildPreStartNodeManagerAsync();
+ try
+ {
+ Should.Throw(() => nm.WriteValues(new[]
+ {
+ new ValueWrite("eq-1/temp", 1.0, OpcUaStatus.Good, DateTime.UtcNow, AddressSpaceRealm.Uns),
+ }));
+
+ Should.Throw(() => nm.WriteValues([]));
+ }
+ finally
+ {
+ await host.DisposeAsync();
+ }
+ }
+
[Fact]
public async Task MaterialiseAlarmCondition_before_CreateAddressSpace_throws_InvalidOperationException()
{
diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/SdkAddressSpaceSinkTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/SdkAddressSpaceSinkTests.cs
index bd3f8b85..699915a1 100644
--- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/SdkAddressSpaceSinkTests.cs
+++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/SdkAddressSpaceSinkTests.cs
@@ -36,6 +36,83 @@ public sealed class SdkAddressSpaceSinkTests : IDisposable
await host.DisposeAsync();
}
+ /// The sink's override forwards the batch to the
+ /// node manager's own batched entry point — every item lands, with its own value and realm.
+ [Fact]
+ public async Task WriteValues_forwards_the_batch_to_the_node_manager()
+ {
+ var (host, server) = await BootAsync();
+ try
+ {
+ var nm = server.NodeManager!;
+ // Through the INTERFACE on purpose: that is the seam production uses, and it is the only call
+ // shape under which deleting the sink's override silently degrades to the per-value default loop
+ // (a concrete-typed call would simply stop compiling, which is not the regression being pinned).
+ IOpcUaAddressSpaceSink sink = new SdkAddressSpaceSink(nm);
+ var ts = new DateTime(2026, 8, 15, 9, 0, 0, DateTimeKind.Utc);
+
+ sink.WriteValues(new List
+ {
+ new("Plant/dev1/Speed", 42, OpcUaStatus.Good, ts, AddressSpaceRealm.Raw),
+ new("eq-1/Speed", 42, OpcUaStatus.Good, ts, AddressSpaceRealm.Uns),
+ });
+
+ nm.VariableCount.ShouldBe(2);
+
+ var raw = nm.TryGetVariable("Plant/dev1/Speed", AddressSpaceRealm.Raw);
+ raw.ShouldNotBeNull();
+ raw!.Value.ShouldBe(42);
+ raw.Timestamp.ShouldBe(ts);
+
+ var uns = nm.TryGetVariable("eq-1/Speed", AddressSpaceRealm.Uns);
+ uns.ShouldNotBeNull();
+ uns!.Value.ShouldBe(42);
+ }
+ finally
+ {
+ await host.DisposeAsync();
+ }
+ }
+
+ /// Pins that the sink actually OVERRIDES WriteValues rather than inheriting the
+ /// interface's per-value default loop. The values-landed assertion above cannot tell the two apart — the
+ /// default loop lands them too — so this uses the one place their behaviour diverges: the manager's
+ /// batched path isolates per item and reports a aggregate, whereas
+ /// the default loop lets the first item's ArgumentException escape and never reaches the tail.
+ /// A test seam (a recording subclass) is not available here — is sealed
+ /// and its WriteValues is not virtual — so the forward is pinned by its observable contract
+ /// instead, which is the stronger assertion anyway.
+ [Fact]
+ public async Task WriteValues_forward_surfaces_the_node_managers_partial_failure_contract()
+ {
+ var (host, server) = await BootAsync();
+ try
+ {
+ var nm = server.NodeManager!;
+ IOpcUaAddressSpaceSink sink = new SdkAddressSpaceSink(nm);
+ var ts = new DateTime(2026, 8, 15, 9, 30, 0, DateTimeKind.Utc);
+
+ var ex = Should.Throw(() => sink.WriteValues(new List
+ {
+ new("eq-fwd/first", 1, OpcUaStatus.Good, ts, AddressSpaceRealm.Uns),
+ new(string.Empty, 2, OpcUaStatus.Good, ts, AddressSpaceRealm.Uns),
+ new("eq-fwd/third", 3, OpcUaStatus.Good, ts, AddressSpaceRealm.Uns),
+ }));
+
+ ex.FailedCount.ShouldBe(1);
+ ex.SucceededCount.ShouldBe(2);
+
+ // The tail item landed — the default per-value loop would have thrown out on item 2 and skipped it.
+ var third = nm.TryGetVariable("eq-fwd/third", AddressSpaceRealm.Uns);
+ third.ShouldNotBeNull();
+ third!.Value.ShouldBe(3);
+ }
+ finally
+ {
+ await host.DisposeAsync();
+ }
+ }
+
/// Verifies that WriteAlarmCondition (un-materialised fallback) creates a dedicated node
/// distinct from value writes.
[Fact]
diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorBatchTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorBatchTests.cs
index 23a776d0..92444a92 100644
--- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorBatchTests.cs
+++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorBatchTests.cs
@@ -72,6 +72,38 @@ public sealed class OpcUaPublishActorBatchTests : RuntimeActorTestBase
sink.Threw.ShouldBeTrue();
}
+ /// A PARTIAL failure is metered honestly: a sink that applies 2 of 3 items and reports the
+ /// aggregate as a must advance the write tally by the SUCCEEDED
+ /// count, not by zero (the general catch's behaviour) and not by the whole batch. The invariant the
+ /// batching work must not break is that the tally counts per ITEM — so a batch that half-failed reports
+ /// exactly the half that landed.
+ [Fact]
+ public void Partial_failure_still_counts_the_items_that_landed()
+ {
+ var sink = new PartialFailureSink();
+ // TestActorRef so the actor's own WriteCount tally is readable — the counter the handler advances is
+ // the thing under test, and nothing else observes it.
+ var actor = ActorOfAsTestActorRef(OpcUaPublishActor.PropsForTests(sink: sink));
+
+ var t0 = new DateTime(2026, 8, 15, 10, 0, 0, DateTimeKind.Utc);
+ actor.Tell(new OpcUaPublishActor.AttributeValueBatchUpdate(new[]
+ {
+ new ValueWrite("Plant/Modbus/dev1/Speed", 1, OpcUaStatus.Good, t0, AddressSpaceRealm.Raw),
+ new ValueWrite("", 2, OpcUaStatus.Good, t0, AddressSpaceRealm.Raw),
+ new ValueWrite("Plant/Modbus/dev1/Temp", 3, OpcUaStatus.Good, 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");
+
+ // The actor survived the partial failure and keeps pumping.
+ actor.Tell(new OpcUaPublishActor.AttributeValueUpdate(
+ "Plant/Modbus/dev1/Other", 4, OpcUaStatus.Good, t0, AddressSpaceRealm.Raw));
+ sink.SingleWrites.Count.ShouldBe(1);
+ actor.UnderlyingActor.WriteCount.ShouldBe(3);
+ }
+
/// The pre-existing single-item AttributeValueUpdate path is untouched by the batch addition.
[Fact]
public void Single_item_message_still_works()
@@ -133,7 +165,8 @@ public sealed class OpcUaPublishActorBatchTests : RuntimeActorTestBase
public void AddReference(string sourceNodeId, AddressSpaceRealm sourceRealm, string targetNodeId, AddressSpaceRealm targetRealm, string referenceType = "Organizes") { }
}
- /// A sink whose batched write throws — drives the handler's catch.
+ /// A sink whose batched write throws a PLAIN exception — drives the handler's GENERAL catch (an
+ /// unrecognised sink failure, where nothing is known about what landed, so nothing is counted).
private sealed class ThrowOnWriteValuesSink : RecordingSink
{
private int _threw;
@@ -146,4 +179,33 @@ public sealed class OpcUaPublishActorBatchTests : RuntimeActorTestBase
throw new InvalidOperationException("simulated batched-write fault");
}
}
+
+ /// A sink modelling the real SDK-backed one's PARTIAL failure: it applies every item with a
+ /// non-empty NodeId, then reports the aggregate as a . Drives the
+ /// handler's typed catch.
+ private sealed class PartialFailureSink : RecordingSink
+ {
+ private int _threw;
+
+ public bool Threw => Volatile.Read(ref _threw) != 0;
+
+ public override void WriteValues(IReadOnlyList values)
+ {
+ var applied = values.Where(v => !string.IsNullOrEmpty(v.NodeId)).ToArray();
+ var failedCount = values.Count - applied.Length;
+ if (failedCount == 0)
+ {
+ base.WriteValues(values);
+ return;
+ }
+
+ base.WriteValues(applied);
+ Interlocked.Exchange(ref _threw, 1);
+ throw new ValueWriteBatchException(
+ firstFailedNodeId: values.First(v => string.IsNullOrEmpty(v.NodeId)).NodeId,
+ failedCount: failedCount,
+ succeededCount: applied.Length,
+ firstFailure: new ArgumentException("empty NodeId"));
+ }
+ }
}