diff --git a/archreview/plans/R3-02-stale-good-family-plan.md b/archreview/plans/R3-02-stale-good-family-plan.md index b947bb25..df7da694 100644 --- a/archreview/plans/R3-02-stale-good-family-plan.md +++ b/archreview/plans/R3-02-stale-good-family-plan.md @@ -669,3 +669,12 @@ test could only pass once the production change landed. - 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. + +### T7 — `OpcUaPublishActor.AttributeStatusBatchUpdate` + +- RED: `--filter "…OpcUaPublishActorStatusBatchTests"` ⇒ **build failure**, `CS0426: The type name + 'AttributeStatusBatchUpdate' does not exist in the type 'OpcUaPublishActor'` (×3). +- GREEN: 58 `OpcUaPublishActor*` tests pass (3 new). +- Metered `OpcUaSinkWrite kind:"status"` so a status annotation is separable from a value write on the + same counter; the recording fake asserts **zero** value writes, since a handler falling through to the + value path would null every node. diff --git a/archreview/plans/R3-02-stale-good-family-plan.md.tasks.json b/archreview/plans/R3-02-stale-good-family-plan.md.tasks.json index 6afcd7d4..035ba90a 100644 --- a/archreview/plans/R3-02-stale-good-family-plan.md.tasks.json +++ b/archreview/plans/R3-02-stale-good-family-plan.md.tasks.json @@ -47,7 +47,7 @@ { "id": "T7", "subject": "OpcUaPublishActor.AttributeStatusBatchUpdate handler -> sink.WriteStatuses (metered, partial-failure Warning)", - "status": "pending", + "status": "done", "blockedBy": [ "T6" ] 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 f0c9a193..b7f8b2af 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/OpcUa/OpcUaPublishActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/OpcUa/OpcUaPublishActor.cs @@ -56,6 +56,15 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers /// their realm each (); status codes ride verbatim (#519). public sealed record AttributeValueBatchUpdate(IReadOnlyList Items); + /// Status-only batch: every named node keeps its CURRENT value and only its status + + /// timestamp change (archreview 03/S15). The comms-loss fan-out sends one of these covering every + /// NodeId a disconnected driver owns — the raw node and each referencing UNS node — so a client sees + /// over the last known value instead of a stale Good. + /// Deliberately a separate message from rather than a + /// flag on it: the host holds no value cache (v3's single-source design), so it has nothing to put in + /// a , and only the node knows what value to retain. + public sealed record AttributeStatusBatchUpdate(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 @@ -292,6 +301,7 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers Receive(HandleAttributeUpdate); Receive(HandleAttributeBatchUpdate); + Receive(HandleAttributeStatusBatchUpdate); Receive(HandleAlarmUpdate); Receive(HandleAlarmQualityUpdate); Receive(HandleRebuild); @@ -368,6 +378,36 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers } } + /// Routes a status-only batch to — the + /// comms-loss annotation (03/S15). Same shape as : metered per + /// ITEM (batching is a transport change, so dashboards must not drop by the batch factor), a + /// is a partial failure that still counts what landed and names + /// the offender, and any other throw is swallowed so a sink fault cannot restart the publish actor and + /// drop live traffic. Metered under kind:"status" so a status annotation is separable from a + /// value write on the same counter. + private void HandleAttributeStatusBatchUpdate(AttributeStatusBatchUpdate msg) + { + try + { + _sink.WriteStatuses(msg.Items); + Interlocked.Add(ref _writes, msg.Items.Count); + OtOpcUaTelemetry.OpcUaSinkWrite.Add(msg.Items.Count, new KeyValuePair("kind", "status")); + } + catch (ValueWriteBatchException ex) + { + _log.Warning(ex, + "OpcUaPublish: sink.WriteStatuses 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", "status")); + } + catch (Exception ex) + { + _log.Warning(ex, "OpcUaPublish: sink.WriteStatuses 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/OpcUaPublishActorStatusBatchTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorStatusBatchTests.cs new file mode 100644 index 00000000..1174e5f9 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorStatusBatchTests.cs @@ -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; + +/// +/// — the status-only route to the sink +/// that the comms-loss fan-out uses (archreview 03/S15). It must reach +/// as exactly one call, never decompose into +/// value writes (which would null every node's value), and survive a throwing sink. +/// +public sealed class OpcUaPublishActorStatusBatchTests : RuntimeActorTestBase +{ + /// + /// One message, one WriteStatuses call carrying every item verbatim — and zero value + /// writes. The second half is load-bearing: a handler that fell through to WriteValues (or + /// to the interface's default, which loops WriteValue with a null value) would satisfy "the + /// status arrived" while erasing the last known value the annotation exists to keep. + /// + [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); + } + + /// + /// 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. + /// + [Fact] + public void Partial_failure_still_counts_the_items_that_landed() + { + var sink = new PartialFailureSink(); + var actor = ActorOfAsTestActorRef(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"); + } + + /// + /// 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. + /// + [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().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; + } + + /// 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. + private class RecordingSink : IOpcUaAddressSpaceSink + { + private readonly object _gate = new(); + private readonly List> _statusBatches = new(); + private readonly List> _valueBatches = new(); + private readonly List _singles = new(); + + public IReadOnlyList> StatusBatches { get { lock (_gate) return _statusBatches.ToArray(); } } + public IReadOnlyList> ValueBatches { get { lock (_gate) return _valueBatches.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)); + } + + public void WriteValues(IReadOnlyList values) + { + lock (_gate) _valueBatches.Add(values.ToArray()); + } + + // Overrides the interface's default (which loops WriteValue with a NULL value). + public virtual void WriteStatuses(IReadOnlyList 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 notifierFolderNodeIds, AddressSpaceRealm notifierFolderRealm) { } + public void AddReference(string sourceNodeId, AddressSpaceRealm sourceRealm, string targetNodeId, AddressSpaceRealm targetRealm, string referenceType = "Organizes") { } + } + + /// Throws a PLAIN exception out of the status batch — drives the handler's general catch. + private sealed class ThrowOnWriteStatusesSink : RecordingSink + { + private int _threw; + + public bool Threw => Volatile.Read(ref _threw) != 0; + + public override void WriteStatuses(IReadOnlyList statuses) + { + Interlocked.Exchange(ref _threw, 1); + throw new InvalidOperationException("simulated status-write fault"); + } + } + + /// Models the real sink's PARTIAL failure: applies every item with a non-empty NodeId, then + /// reports the aggregate as a . + private sealed class PartialFailureSink : RecordingSink + { + private int _threw; + + public bool Threw => Volatile.Read(ref _threw) != 0; + + public override void WriteStatuses(IReadOnlyList 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")); + } + } +}