feat(runtime): publish actor routes status-only batches to the sink

2026-08-20 10:03:25 -04:00
parent ec5b48bfcd
commit 04019e9414
4 changed files with 247 additions and 1 deletions
@@ -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.
@@ -47,7 +47,7 @@
    {
      "id": "T7",
      "subject": "OpcUaPublishActor.AttributeStatusBatchUpdate handler -> sink.WriteStatuses (metered, partial-failure Warning)",
-      "status": "pending",
+      "status": "done",
      "blockedBy": [
        "T6"
      ]
@@ -56,6 +56,15 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
    /// their realm each (<see cref="ValueWrite"/>); status codes ride verbatim (#519).</summary>
    public sealed record AttributeValueBatchUpdate(IReadOnlyList<ValueWrite> Items);

+    /// <summary>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
+    /// <see cref="OpcUaStatus.BadNoCommunication"/> over the last known value instead of a stale Good.
+    /// <para>Deliberately a separate message from <see cref="AttributeValueBatchUpdate"/> 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 <see cref="ValueWrite"/>, and only the node knows what value to retain.</para></summary>
+    public sealed record AttributeStatusBatchUpdate(IReadOnlyList<StatusWrite> Items);
+
    /// <summary>Carries the full Part 9 condition state for a scripted alarm to the sink. The
    /// <paramref name="State"/> snapshot is the Commons projection the Runtime host maps from the engine's
    /// Core <c>AlarmConditionState</c> + severity/message — the actor stays decoupled from
@@ -292,6 +301,7 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers

        Receive<AttributeValueUpdate>(HandleAttributeUpdate);
        Receive<AttributeValueBatchUpdate>(HandleAttributeBatchUpdate);
+        Receive<AttributeStatusBatchUpdate>(HandleAttributeStatusBatchUpdate);
        Receive<AlarmStateUpdate>(HandleAlarmUpdate);
        Receive<AlarmQualityUpdate>(HandleAlarmQualityUpdate);
        Receive<RebuildAddressSpace>(HandleRebuild);
@@ -368,6 +378,36 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
        }
    }

+    /// <summary>Routes a status-only batch to <see cref="IOpcUaAddressSpaceSink.WriteStatuses"/> — the
+    /// comms-loss annotation (03/S15). Same shape as <see cref="HandleAttributeBatchUpdate"/>: metered per
+    /// ITEM (batching is a transport change, so dashboards must not drop by the batch factor), a
+    /// <see cref="ValueWriteBatchException"/> 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 <c>kind:"status"</c> so a status annotation is separable from a
+    /// value write on the same counter.</summary>
+    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<string, object?>("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<string, object?>("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
@@ -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;
+
+/// <summary>
+///     <see cref="OpcUaPublishActor.AttributeStatusBatchUpdate"/> — the status-only route to the sink
+///     that the comms-loss fan-out uses (archreview 03/S15). It must reach
+///     <see cref="IOpcUaAddressSpaceSink.WriteStatuses"/> as exactly one call, never decompose into
+///     value writes (which would null every node's value), and survive a throwing sink.
+/// </summary>
+public sealed class OpcUaPublishActorStatusBatchTests : RuntimeActorTestBase
+{
+    /// <summary>
+    ///     One message, one <c>WriteStatuses</c> call carrying every item verbatim — and <b>zero</b> value
+    ///     writes. The second half is load-bearing: a handler that fell through to <c>WriteValues</c> (or
+    ///     to the interface's default, which loops <c>WriteValue</c> with a null value) would satisfy "the
+    ///     status arrived" while erasing the last known value the annotation exists to keep.
+    /// </summary>
+    [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);
+    }
+
+    /// <summary>
+    ///     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.
+    /// </summary>
+    [Fact]
+    public void Partial_failure_still_counts_the_items_that_landed()
+    {
+        var sink = new PartialFailureSink();
+        var actor = ActorOfAsTestActorRef<OpcUaPublishActor>(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");
+    }
+
+    /// <summary>
+    ///     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.
+    /// </summary>
+    [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<InvalidOperationException>().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;
+    }
+
+    /// <summary>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.</summary>
+    private class RecordingSink : IOpcUaAddressSpaceSink
+    {
+        private readonly object _gate = new();
+        private readonly List<IReadOnlyList<StatusWrite>> _statusBatches = new();
+        private readonly List<IReadOnlyList<ValueWrite>> _valueBatches = new();
+        private readonly List<ValueWrite> _singles = new();
+
+        public IReadOnlyList<IReadOnlyList<StatusWrite>> StatusBatches { get { lock (_gate) return _statusBatches.ToArray(); } }
+        public IReadOnlyList<IReadOnlyList<ValueWrite>> ValueBatches { get { lock (_gate) return _valueBatches.ToArray(); } }
+        public IReadOnlyList<ValueWrite> 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<ValueWrite> values)
+        {
+            lock (_gate) _valueBatches.Add(values.ToArray());
+        }
+
+        // Overrides the interface's default (which loops WriteValue with a NULL value).
+        public virtual void WriteStatuses(IReadOnlyList<StatusWrite> 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<string> notifierFolderNodeIds, AddressSpaceRealm notifierFolderRealm) { }
+        public void AddReference(string sourceNodeId, AddressSpaceRealm sourceRealm, string targetNodeId, AddressSpaceRealm targetRealm, string referenceType = "Organizes") { }
+    }
+
+    /// <summary>Throws a PLAIN exception out of the status batch — drives the handler's general catch.</summary>
+    private sealed class ThrowOnWriteStatusesSink : RecordingSink
+    {
+        private int _threw;
+
+        public bool Threw => Volatile.Read(ref _threw) != 0;
+
+        public override void WriteStatuses(IReadOnlyList<StatusWrite> statuses)
+        {
+            Interlocked.Exchange(ref _threw, 1);
+            throw new InvalidOperationException("simulated status-write fault");
+        }
+    }
+
+    /// <summary>Models the real sink's PARTIAL failure: applies every item with a non-empty NodeId, then
+    /// reports the aggregate as a <see cref="ValueWriteBatchException"/>.</summary>
+    private sealed class PartialFailureSink : RecordingSink
+    {
+        private int _threw;
+
+        public bool Threw => Volatile.Read(ref _threw) != 0;
+
+        public override void WriteStatuses(IReadOnlyList<StatusWrite> 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"));
+        }
+    }
+}