From bb3ccf264b7f0fc7042d62b8163660f03c2da903 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Sat, 15 Aug 2026 12:09:27 -0400 Subject: [PATCH] feat(commons): batched WriteValues on the address-space sink --- .../OpcUa/DeferredAddressSpaceSink.cs | 6 + .../OpcUa/IOpcUaAddressSpaceSink.cs | 20 ++++ .../AddressSpaceSinkBatchTests.cs | 112 ++++++++++++++++++ .../DeferredSinkForwardingReflectionTests.cs | 18 ++- 4 files changed, 155 insertions(+), 1 deletion(-) create mode 100644 tests/Core/ZB.MOM.WW.OtOpcUa.Commons.Tests/AddressSpaceSinkBatchTests.cs diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/OpcUa/DeferredAddressSpaceSink.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/OpcUa/DeferredAddressSpaceSink.cs index b0fd6325..5c917dc0 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/OpcUa/DeferredAddressSpaceSink.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/OpcUa/DeferredAddressSpaceSink.cs @@ -26,6 +26,12 @@ public sealed class DeferredAddressSpaceSink : IOpcUaAddressSpaceSink, ISurgical public void WriteValue(string nodeId, object? value, uint statusCode, DateTime sourceTimestampUtc, AddressSpaceRealm realm) => _inner.WriteValue(nodeId, value, statusCode, sourceTimestampUtc, realm); + /// + // Forward the batch to the inner sink. Without this the interface's default per-item loop runs here and + // the one-lock-per-batch win is silently degraded back to one lock per value on every driver-role host + // (actors inject THIS wrapper, not the inner SdkAddressSpaceSink) — same trap as WireAlarmNotifiers above. + public void WriteValues(IReadOnlyList values) => _inner.WriteValues(values); + /// public void WriteAlarmCondition(string alarmNodeId, AlarmConditionSnapshot state, DateTime sourceTimestampUtc, AddressSpaceRealm realm) => _inner.WriteAlarmCondition(alarmNodeId, state, sourceTimestampUtc, realm); 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 78d2f720..35107c13 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/OpcUa/IOpcUaAddressSpaceSink.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/OpcUa/IOpcUaAddressSpaceSink.cs @@ -22,6 +22,21 @@ public interface IOpcUaAddressSpaceSink /// calls this once per NodeId (raw + every referencing UNS node) with the matching realm. void WriteValue(string nodeId, object? value, uint statusCode, DateTime sourceTimestampUtc, AddressSpaceRealm realm); + /// 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. + /// The value writes to apply, in order. + void WriteValues(IReadOnlyList values) + { + ArgumentNullException.ThrowIfNull(values); + for (var i = 0; i < values.Count; i++) + { + var v = values[i]; + WriteValue(v.NodeId, v.Value, v.StatusCode, v.SourceTimestampUtc, v.Realm); + } + } + /// Write an alarm-condition's full Part 9 state. When a real condition node has been /// materialised for via , /// this projects the whole @@ -165,6 +180,11 @@ public interface IOpcUaAddressSpaceSink string referenceType = "Organizes"); } +/// One value write within a batch. +/// Same semantics per field as (#519: status verbatim). +public readonly record struct ValueWrite( + string NodeId, object? Value, uint StatusCode, DateTime SourceTimestampUtc, AddressSpaceRealm Realm); + /// 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/tests/Core/ZB.MOM.WW.OtOpcUa.Commons.Tests/AddressSpaceSinkBatchTests.cs b/tests/Core/ZB.MOM.WW.OtOpcUa.Commons.Tests/AddressSpaceSinkBatchTests.cs new file mode 100644 index 00000000..ec1b5a5a --- /dev/null +++ b/tests/Core/ZB.MOM.WW.OtOpcUa.Commons.Tests/AddressSpaceSinkBatchTests.cs @@ -0,0 +1,112 @@ +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.OpcUa; + +namespace ZB.MOM.WW.OtOpcUa.Commons.Tests; + +/// +/// The batched seam. It exists so an SDK-backed sink +/// can take its internal lock ONCE per batch instead of once per value, and it ships as a default +/// interface method so every existing sink and test fake keeps compiling unchanged — this suite +/// pins both halves of that bargain: the default loops +/// verbatim (#519: the status code rides through untouched), and +/// forwards the batch to an overriding inner rather than +/// silently degrading it to per-item calls on every driver-role host. +/// +public class AddressSpaceSinkBatchTests +{ + [Fact] + public void Default_WriteValues_loops_WriteValue() + { + // A sink that implements ONLY the pre-existing interface members — exactly the shape of every + // sink and fake that predates the batch seam. The default implementation must fan the batch out + // to WriteValue, in order, with every field carried through unchanged. + var fake = new PerItemOnlySink(); + var t0 = new DateTime(2026, 8, 15, 10, 0, 0, DateTimeKind.Utc); + + ((IOpcUaAddressSpaceSink)fake).WriteValues(new[] + { + new ValueWrite("Plant/Modbus/dev1/Speed", 42, OpcUaStatus.Good, t0, AddressSpaceRealm.Raw), + new ValueWrite("Area/Line/Eq/Speed", 43.5, OpcUaStatus.BadNoCommunication, t0.AddSeconds(1), AddressSpaceRealm.Uns), + new ValueWrite("Plant/Modbus/dev1/Temp", null, OpcUaStatus.Bad, t0.AddSeconds(2), AddressSpaceRealm.Raw), + }); + + fake.Calls.Count.ShouldBe(3); + + fake.Calls[0].ShouldBe(("Plant/Modbus/dev1/Speed", (object?)42, OpcUaStatus.Good, t0, AddressSpaceRealm.Raw)); + fake.Calls[1].ShouldBe(("Area/Line/Eq/Speed", (object?)43.5, OpcUaStatus.BadNoCommunication, t0.AddSeconds(1), AddressSpaceRealm.Uns)); + fake.Calls[2].ShouldBe(("Plant/Modbus/dev1/Temp", (object?)null, OpcUaStatus.Bad, t0.AddSeconds(2), AddressSpaceRealm.Raw)); + } + + [Fact] + public void Deferred_sink_forwards_WriteValues_to_inner_once() + { + // The whole point of the seam is the ONE-lock-per-batch win, and actors inject the deferred + // wrapper — not the inner SdkAddressSpaceSink. A missing forward here would fall back to the + // interface default and re-explode the batch into per-item WriteValue calls on every + // driver-role host: the batching would ship inert, exactly the F10b / PR#423 trap class. + var inner = new BatchingSpySink(); + var sink = new DeferredAddressSpaceSink(); + sink.SetSink(inner); + + var t0 = new DateTime(2026, 8, 15, 11, 0, 0, DateTimeKind.Utc); + var batch = new[] + { + new ValueWrite("a", 1, OpcUaStatus.Good, t0, AddressSpaceRealm.Raw), + new ValueWrite("b", 2, OpcUaStatus.Good, t0, AddressSpaceRealm.Uns), + new ValueWrite("c", 3, OpcUaStatus.Good, t0, AddressSpaceRealm.Raw), + }; + + sink.WriteValues(batch); + + inner.Batches.Count.ShouldBe(1, "the batch must reach the inner sink as ONE call, not be re-looped"); + inner.Batches[0].Count.ShouldBe(3); + inner.Batches[0].Select(v => v.NodeId).ShouldBe(new[] { "a", "b", "c" }); + inner.PerItemCalls.ShouldBe(0, "the inner sink's per-item WriteValue must not be reached at all"); + } + + // ---- test doubles ---- + + /// Implements only the members that existed before the batch seam, so the interface's + /// default WriteValues is what runs. + private sealed class PerItemOnlySink : IOpcUaAddressSpaceSink + { + public List<(string NodeId, object? Value, uint StatusCode, DateTime SourceTimestampUtc, AddressSpaceRealm Realm)> Calls { get; } = new(); + + public void WriteValue(string nodeId, object? value, uint statusCode, DateTime sourceTimestampUtc, AddressSpaceRealm realm) + => Calls.Add((nodeId, value, statusCode, sourceTimestampUtc, realm)); + + public void WriteAlarmCondition(string alarmNodeId, AlarmConditionSnapshot state, DateTime sourceTimestampUtc, AddressSpaceRealm realm) { } + public void WriteAlarmQuality(string alarmNodeId, uint statusCode, DateTime sourceTimestampUtc, AddressSpaceRealm realm) { } + public void MaterialiseAlarmCondition(string alarmNodeId, string equipmentNodeId, string displayName, string alarmType, int severity, AddressSpaceRealm realm, bool isNative = false) { } + public void WireAlarmNotifiers(string alarmNodeId, AddressSpaceRealm alarmRealm, IReadOnlyList notifierFolderNodeIds, AddressSpaceRealm notifierFolderRealm) { } + public void EnsureFolder(string folderNodeId, string? parentNodeId, string displayName, AddressSpaceRealm realm) { } + 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) { } + public void AddReference(string sourceNodeId, AddressSpaceRealm sourceRealm, string targetNodeId, AddressSpaceRealm targetRealm, string referenceType = "Organizes") { } + } + + /// Overrides WriteValues the way a real SDK-backed sink does — records the batch and + /// does NOT loop — so a wrapper that dropped the forward is distinguishable from one that kept it. + private sealed class BatchingSpySink : IOpcUaAddressSpaceSink + { + public List> Batches { get; } = new(); + public int PerItemCalls { get; private set; } + + public void WriteValue(string nodeId, object? value, uint statusCode, DateTime sourceTimestampUtc, AddressSpaceRealm realm) + => PerItemCalls++; + + public void WriteValues(IReadOnlyList values) => Batches.Add(values); + + public void WriteAlarmCondition(string alarmNodeId, AlarmConditionSnapshot state, DateTime sourceTimestampUtc, AddressSpaceRealm realm) { } + public void WriteAlarmQuality(string alarmNodeId, uint statusCode, DateTime sourceTimestampUtc, AddressSpaceRealm realm) { } + public void MaterialiseAlarmCondition(string alarmNodeId, string equipmentNodeId, string displayName, string alarmType, int severity, AddressSpaceRealm realm, bool isNative = false) { } + public void WireAlarmNotifiers(string alarmNodeId, AddressSpaceRealm alarmRealm, IReadOnlyList notifierFolderNodeIds, AddressSpaceRealm notifierFolderRealm) { } + public void EnsureFolder(string folderNodeId, string? parentNodeId, string displayName, AddressSpaceRealm realm) { } + 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) { } + public void AddReference(string sourceNodeId, AddressSpaceRealm sourceRealm, string targetNodeId, AddressSpaceRealm targetRealm, string referenceType = "Organizes") { } + } +} diff --git a/tests/Core/ZB.MOM.WW.OtOpcUa.Commons.Tests/OpcUa/DeferredSinkForwardingReflectionTests.cs b/tests/Core/ZB.MOM.WW.OtOpcUa.Commons.Tests/OpcUa/DeferredSinkForwardingReflectionTests.cs index 5a005ee0..7a455cec 100644 --- a/tests/Core/ZB.MOM.WW.OtOpcUa.Commons.Tests/OpcUa/DeferredSinkForwardingReflectionTests.cs +++ b/tests/Core/ZB.MOM.WW.OtOpcUa.Commons.Tests/OpcUa/DeferredSinkForwardingReflectionTests.cs @@ -86,12 +86,28 @@ public class DeferredSinkForwardingReflectionTests { if (method.Name == nameof(IOpcUaAddressSpaceSink.RebuildAddressSpace)) continue; - method.GetParameters().Any(p => p.ParameterType == typeof(AddressSpaceRealm)).ShouldBeTrue( + method.GetParameters().Any(p => CarriesRealm(p.ParameterType)).ShouldBeTrue( $"{method.DeclaringType!.Name}.{method.Name} names a node but has no AddressSpaceRealm parameter — " + "a bare node id is ambiguous across the Raw and UNS namespaces (B4-WP2). Add the realm discriminator."); } } + /// Whether a parameter type supplies the B4-WP2 realm discriminator — either directly, or + /// carried per element by a batch parameter (WriteValues(IReadOnlyList<ValueWrite>) is + /// realm-qualified per item, since one batch legitimately spans both namespaces). Looking THROUGH the + /// element type rather than exempting the method by name keeps the guard mechanical: a future batch + /// item struct that forgets its realm still trips it. + private static bool CarriesRealm(Type t) + { + if (t == typeof(AddressSpaceRealm)) return true; + + if (t.IsGenericType && t.GetGenericArguments().Any(CarriesRealm)) return true; + + return t.Namespace == typeof(IOpcUaAddressSpaceSink).Namespace + && (t.GetProperties().Any(p => p.PropertyType == typeof(AddressSpaceRealm)) + || t.GetFields().Any(f => f.FieldType == typeof(AddressSpaceRealm))); + } + [Fact] public void Every_IServiceLevelPublisher_method_reaches_the_inner_publisher() {