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 63292de0..631feed5 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/OpcUa/OpcUaPublishActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/OpcUa/OpcUaPublishActor.cs @@ -51,6 +51,11 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers /// hands it to the SDK verbatim. public sealed record AttributeValueUpdate(string NodeId, object? Value, uint StatusCode, DateTime TimestampUtc, AddressSpaceRealm Realm); + /// Batched form of : every fan-out write of one driver + /// production burst in one message — the sink takes its lock once for the whole batch. Items carry + /// their realm each (); status codes ride verbatim (#519). + public sealed record AttributeValueBatchUpdate(IReadOnlyList Items); + /// Carries the full Part 9 condition state for a scripted alarm to the sink. The /// snapshot is the Commons projection the Runtime host maps from the engine's /// Core AlarmConditionState + severity/message — the actor stays decoupled from @@ -286,6 +291,7 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers _healthTickInterval = healthTickInterval ?? TimeSpan.FromSeconds(5); Receive(HandleAttributeUpdate); + Receive(HandleAttributeBatchUpdate); Receive(HandleAlarmUpdate); Receive(HandleAlarmQualityUpdate); Receive(HandleRebuild); @@ -330,6 +336,24 @@ 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. + private void HandleAttributeBatchUpdate(AttributeValueBatchUpdate msg) + { + try + { + _sink.WriteValues(msg.Items); + Interlocked.Add(ref _writes, msg.Items.Count); + OtOpcUaTelemetry.OpcUaSinkWrite.Add(msg.Items.Count, new KeyValuePair("kind", "value")); + } + catch (Exception ex) + { + _log.Warning(ex, "OpcUaPublish: sink.WriteValues threw for a {Count}-item batch", msg.Items.Count); + } + } + private void HandleAlarmUpdate(AlarmStateUpdate msg) { try 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 new file mode 100644 index 00000000..23a776d0 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorBatchTests.cs @@ -0,0 +1,149 @@ +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; + +/// +/// Task 7 of the value-pipeline batching work: +/// must reach the sink as exactly ONE call — that single +/// call is the whole point (the SDK-backed sink takes its lock once per batch instead of once per value), +/// so a handler that looped WriteValue would pass a "the values arrived" assertion while delivering +/// none of the win. Hence the fake counts per-item WriteValue calls separately and the assertions +/// require that count to be zero. +/// The per-item receive stays — VirtualTagHostActor +/// and the alarm paths still send it — so it is pinned here too. +/// +public sealed class OpcUaPublishActorBatchTests : RuntimeActorTestBase +{ + /// A 3-item batch lands as one WriteValues call carrying all three items verbatim (field-exact, + /// realm included), and never decomposes into per-item WriteValue calls. + [Fact] + public void Batch_message_reaches_sink_as_one_WriteValues_call() + { + var sink = new RecordingSink(); + var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(sink: sink)); + + var t0 = new DateTime(2026, 8, 15, 10, 0, 0, DateTimeKind.Utc); + var items = new[] + { + new ValueWrite("Plant/Modbus/dev1/Speed", 42, OpcUaStatus.Good, t0, AddressSpaceRealm.Raw), + new ValueWrite("AreaA/Line1/Eq1/Speed", 42, OpcUaStatus.Good, t0, AddressSpaceRealm.Uns), + new ValueWrite("Plant/Modbus/dev1/Temp", 7.5d, OpcUaStatus.BadNoCommunication, + t0.AddSeconds(1), AddressSpaceRealm.Raw), + }; + + actor.Tell(new OpcUaPublishActor.AttributeValueBatchUpdate(items)); + + AwaitAssert(() => sink.Batches.Count.ShouldBe(1), PresenceBudget); + + var batch = sink.Batches[0]; + batch.Count.ShouldBe(3); + batch[0].ShouldBe(items[0]); + batch[1].ShouldBe(items[1]); + batch[2].ShouldBe(items[2]); + + // The load-bearing half: one batched call, NOT three per-item calls. + sink.SingleWrites.Count.ShouldBe(0); + } + + /// A sink that throws out of WriteValues is swallowed and logged — the actor survives and keeps + /// processing (proved by a follow-up message landing), rather than restarting and dropping traffic. + [Fact] + public void Sink_throw_is_swallowed_and_logged() + { + var sink = new ThrowOnWriteValuesSink(); + var actor = Sys.ActorOf(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), + })); + + // Follow-up on the untouched per-item path: it lands only if the actor is still alive and 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(); + } + + /// The pre-existing single-item AttributeValueUpdate path is untouched by the batch addition. + [Fact] + public void Single_item_message_still_works() + { + var sink = new RecordingSink(); + var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(sink: sink)); + + var t0 = new DateTime(2026, 8, 15, 10, 0, 0, DateTimeKind.Utc); + actor.Tell(new OpcUaPublishActor.AttributeValueUpdate( + "AreaA/Line1/Eq1/Speed", 99, OpcUaStatus.Good, t0, AddressSpaceRealm.Uns)); + + AwaitAssert(() => sink.SingleWrites.Count.ShouldBe(1), PresenceBudget); + sink.SingleWrites[0].ShouldBe(new ValueWrite("AreaA/Line1/Eq1/Speed", 99, OpcUaStatus.Good, t0, AddressSpaceRealm.Uns)); + sink.Batches.Count.ShouldBe(0); + } + + 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; + } + + /// Records batched and per-item value writes separately, so a handler that looped WriteValue + /// instead of calling WriteValues once is distinguishable from one that batched. + private class RecordingSink : IOpcUaAddressSpaceSink + { + private readonly object _gate = new(); + private readonly List> _batches = new(); + private readonly List _singles = new(); + + public IReadOnlyList> Batches { get { lock (_gate) return _batches.ToArray(); } } + public IReadOnlyList 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)); + } + + // Overrides the interface's default WriteValues so the batch is observed as one call, not N. + public virtual void WriteValues(IReadOnlyList values) + { + lock (_gate) _batches.Add(values.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 notifierFolderNodeIds, AddressSpaceRealm notifierFolderRealm) { } + 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. + private sealed class ThrowOnWriteValuesSink : RecordingSink + { + private int _threw; + + public bool Threw => Volatile.Read(ref _threw) != 0; + + public override void WriteValues(IReadOnlyList values) + { + Interlocked.Exchange(ref _threw, 1); + throw new InvalidOperationException("simulated batched-write fault"); + } + } +}