fix(r2-04): surface degraded address-space applies (Error log + otopcua.opcua.apply.failed)

2026-07-13 10:44:18 -04:00
parent e37abf36a0
commit 759125372a
4 changed files with 211 additions and 8 deletions
@@ -5,7 +5,7 @@
    { "id": "T1", "subject": "Extend AddressSpaceApplyOutcome with RebuildFailed/FailedNodes + SafeRebuild returns bool (01/S-1)", "status": "completed", "blockedBy": [] },
    { "id": "T2", "subject": "Safe* helpers return bool; Apply tallies removal-pass failures into FailedNodes", "status": "completed", "blockedBy": ["T1"] },
    { "id": "T3", "subject": "Materialise* passes return swallowed-failure counts (void -> int, five methods)", "status": "completed", "blockedBy": ["T2"] },
-    { "id": "T4", "subject": "OpcUaApplyFailed counter + OpcUaPublishActor Error/meter surfacing of degraded applies", "status": "pending", "blockedBy": ["T3"] },
+    { "id": "T4", "subject": "OpcUaApplyFailed counter + OpcUaPublishActor Error/meter surfacing of degraded applies", "status": "completed", "blockedBy": ["T3"] },
    { "id": "T5", "subject": "Galaxy internal IMxWriteOps seam under GatewayGalaxyDataWriter (sealed SDK session untestable) (06/S-1)", "status": "pending", "blockedBy": [] },
    { "id": "T6", "subject": "Galaxy fail-closed write on AdviseSupervisory failure -> BadCommunicationError, no raw write", "status": "pending", "blockedBy": ["T5"] },
    { "id": "T7", "subject": "Galaxy counters galaxy.writes.advise_failed + galaxy.writes.unconfirmed (empty-statuses Good metered)", "status": "pending", "blockedBy": ["T6"] },
@@ -58,6 +58,14 @@ public static class OtOpcUaTelemetry
        Meter.CreateCounter<long>("otopcua.opcua.sink.write", unit: "{write}",
            description: "Writes that landed in IOpcUaAddressSpaceSink (kind=value|alarm|rebuild).");

+    /// <summary>Address-space apply/materialise passes that swallowed at least one sink failure —
+    /// a failed rebuild (kind=rebuild) or per-node materialise failures (kind=nodes). A non-zero rate
+    /// means the running server holds a stale or partially-materialised address space despite a
+    /// reported-successful deploy (archreview 01/S-1).</summary>
+    public static readonly Counter<long> OpcUaApplyFailed =
+        Meter.CreateCounter<long>("otopcua.opcua.apply.failed", unit: "{apply}",
+            description: "Apply/materialise passes with swallowed sink failures (kind=rebuild|nodes).");
+
    public static readonly Counter<long> ServiceLevelChange =
        Meter.CreateCounter<long>("otopcua.redundancy.service_level_change", unit: "{change}",
            description: "OPC UA Server.ServiceLevel transitions emitted by the redundancy state.");
@@ -335,27 +335,43 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
            var outcome = _applier.Apply(plan);
            _lastApplied = composition;

-            _applier.MaterialiseHierarchy(composition);
+            // Sum swallowed per-node materialise failures across every pass together with Apply's own
+            // removal-pass tally (archreview 01/S-1). A degraded apply used to vanish into per-node
+            // Warnings + an optimistic Info line; now it surfaces at Error + a dedicated meter.
+            var failedNodes = outcome.FailedNodes;
+            failedNodes += _applier.MaterialiseHierarchy(composition);
            // T14 — scripted alarms get their own pass right after the hierarchy so the equipment
            // folders they parent under already exist. Materialises real Part 9 AlarmConditionState
            // nodes (keyed by ScriptedAlarmId so AlarmStateUpdate writes target them); disabled
            // alarms are skipped.
-            _applier.MaterialiseScriptedAlarms(composition);
+            failedNodes += _applier.MaterialiseScriptedAlarms(composition);
            // Equipment-namespace tags get their own pass: ensures each signal's Variable (and any
            // FolderPath sub-folder) exists under its already-materialised equipment folder so
            // clients can browse them. Live values are pushed by DriverHostActor.ForwardToMux after
            // each subscription cycle; variables show BadWaitingForInitialData only until the first
            // publish interval fires.
-            _applier.MaterialiseEquipmentTags(composition);
+            failedNodes += _applier.MaterialiseEquipmentTags(composition);
            // Equipment-namespace VirtualTags get their own pass right after the equipment tags:
            // ensures each computed signal's Variable (and any FolderPath sub-folder) exists under its
            // equipment folder with a folder-scoped NodeId. VirtualTagHostActor.OnResult pushes live
            // values once the first dependency update arrives; until then variables show BadWaitingForInitialData.
-            _applier.MaterialiseEquipmentVirtualTags(composition);
+            failedNodes += _applier.MaterialiseEquipmentVirtualTags(composition);

            OtOpcUaTelemetry.OpcUaSinkWrite.Add(1, new KeyValuePair<string, object?>("kind", "rebuild"));
-            _log.Info("OpcUaPublish: applied rebuild (correlation={Correlation}, added={Added}, removed={Removed}, changed={Changed}, rebuild={Rebuild})",
-                msg.Correlation, outcome.AddedNodes, outcome.RemovedNodes, outcome.ChangedNodes, outcome.RebuildCalled);
+            if (outcome.RebuildFailed || failedNodes > 0)
+            {
+                OtOpcUaTelemetry.OpcUaApplyFailed.Add(1,
+                    new KeyValuePair<string, object?>("kind", outcome.RebuildFailed ? "rebuild" : "nodes"));
+                _log.Error(
+                    "OpcUaPublish: address-space apply DEGRADED (correlation={Correlation}, rebuildFailed={RebuildFailed}, failedNodes={FailedNodes}, added={Added}, removed={Removed}, changed={Changed}) — the running address space may be stale or partial",
+                    msg.Correlation, outcome.RebuildFailed, failedNodes,
+                    outcome.AddedNodes, outcome.RemovedNodes, outcome.ChangedNodes);
+            }
+            else
+            {
+                _log.Info("OpcUaPublish: applied rebuild (correlation={Correlation}, added={Added}, removed={Removed}, changed={Changed}, rebuild={Rebuild})",
+                    msg.Correlation, outcome.AddedNodes, outcome.RemovedNodes, outcome.ChangedNodes, outcome.RebuildCalled);
+            }
        }
        catch (Exception ex)
        {
@@ -412,7 +428,16 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
            _log.Debug("OpcUaPublish: no applier wired — discarding MaterialiseDiscoveredNodes for {Equipment}", msg.EquipmentRootNodeId);
            return;
        }
-        _applier.MaterialiseDiscoveredNodes(msg.EquipmentRootNodeId, msg.Folders, msg.Variables);
+        var failedNodes = _applier.MaterialiseDiscoveredNodes(msg.EquipmentRootNodeId, msg.Folders, msg.Variables);
+        if (failedNodes > 0)
+        {
+            // archreview 01/S-1: a swallowed discovered-node injection failure surfaces at Error + the
+            // dedicated meter instead of vanishing into per-node Warnings.
+            OtOpcUaTelemetry.OpcUaApplyFailed.Add(1, new KeyValuePair<string, object?>("kind", "nodes"));
+            _log.Error(
+                "OpcUaPublish: discovered-node injection DEGRADED for {Equipment} (failedNodes={FailedNodes}) — some discovered nodes may be missing",
+                msg.EquipmentRootNodeId, failedNodes);
+        }
    }

    private void HandleServiceLevelChanged(ServiceLevelChanged msg)
@@ -0,0 +1,170 @@
+using System.Diagnostics.Metrics;
+using System.Text.Json;
+using Akka.Actor;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.Extensions.Logging.Abstractions;
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Commons.Observability;
+using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
+using ZB.MOM.WW.OtOpcUa.Commons.Types;
+using ZB.MOM.WW.OtOpcUa.Configuration;
+using ZB.MOM.WW.OtOpcUa.Configuration.Entities;
+using ZB.MOM.WW.OtOpcUa.Configuration.Enums;
+using ZB.MOM.WW.OtOpcUa.OpcUaServer;
+using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa;
+using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness;
+
+namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.OpcUa;
+
+/// <summary>
+///     archreview 01/S-1: when the applier reports a degraded apply (a rebuild that threw, or per-node
+///     materialise failures), <see cref="OpcUaPublishActor.HandleRebuild"/> must surface it — increment the
+///     <c>otopcua.opcua.apply.failed</c> meter and log at Error — instead of the optimistic Info line. The
+///     happy path must stay Info-only (zero increments).
+/// </summary>
+public sealed class OpcUaPublishActorApplyFailureTests : RuntimeActorTestBase
+{
+    /// <summary>A rebuild whose sink throws increments otopcua.opcua.apply.failed (kind=rebuild).</summary>
+    [Fact]
+    public void Rebuild_when_sink_throws_increments_apply_failed_meter()
+    {
+        using var recorder = new MeterRecorder("otopcua.opcua.apply.failed");
+        var db = NewInMemoryDbFactory();
+        var sink = new ThrowOnRebuildSink();
+        var applier = new AddressSpaceApplier(sink, NullLogger<AddressSpaceApplier>.Instance);
+        SeedEquipmentDeployment(db, "eq-1");
+
+        var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(sink: sink, dbFactory: db, applier: applier));
+        actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId()));
+
+        AwaitAssert(() =>
+        {
+            recorder.Total.ShouldBeGreaterThanOrEqualTo(1);
+            recorder.WithTag("kind", "rebuild").ShouldBeGreaterThanOrEqualTo(1);
+        }, duration: TimeSpan.FromSeconds(2));
+    }
+
+    /// <summary>A clean rebuild does NOT increment the apply-failed meter (Info-only happy path).</summary>
+    [Fact]
+    public void Rebuild_happy_path_does_not_increment_apply_failed_meter()
+    {
+        using var recorder = new MeterRecorder("otopcua.opcua.apply.failed");
+        var db = NewInMemoryDbFactory();
+        var sink = new NoopSink();
+        var applier = new AddressSpaceApplier(sink, NullLogger<AddressSpaceApplier>.Instance);
+        SeedEquipmentDeployment(db, "eq-1");
+
+        var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(sink: sink, dbFactory: db, applier: applier));
+        actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId()));
+
+        // Give the actor time to process, then confirm the failed meter never fired.
+        Thread.Sleep(400);
+        recorder.Total.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;
+    }
+
+    private static void SeedEquipmentDeployment(IDbContextFactory<OtOpcUaConfigDbContext> dbFactory, params string[] equipmentIds)
+    {
+        var artifact = JsonSerializer.SerializeToUtf8Bytes(new
+        {
+            Equipment = equipmentIds.Select(id => new
+            {
+                EquipmentId = id,
+                MachineCode = id.ToUpperInvariant(),
+                UnsLineId = "line-1",
+                Name = id,
+            }).ToArray(),
+            DriverInstances = Array.Empty<object>(),
+            ScriptedAlarms = Array.Empty<object>(),
+        });
+
+        using var ctx = dbFactory.CreateDbContext();
+        ctx.Deployments.Add(new Deployment
+        {
+            DeploymentId = Guid.NewGuid(),
+            RevisionHash = new string('a', 64),
+            Status = DeploymentStatus.Sealed,
+            CreatedBy = "test",
+            SealedAtUtc = DateTime.UtcNow,
+            ArtifactBlob = artifact,
+        });
+        ctx.SaveChanges();
+    }
+
+    /// <summary>A sink whose RebuildAddressSpace throws (drives the applier's SafeRebuild catch → RebuildFailed).</summary>
+    private sealed class ThrowOnRebuildSink : IOpcUaAddressSpaceSink
+    {
+        public void WriteValue(string nodeId, object? value, OpcUaQuality quality, DateTime sourceTimestampUtc) { }
+        public void WriteAlarmCondition(string alarmNodeId, AlarmConditionSnapshot state, DateTime occurredUtc) { }
+        public void MaterialiseAlarmCondition(string alarmNodeId, string equipmentNodeId, string displayName, string alarmType, int severity, bool isNative = false) { }
+        public void EnsureFolder(string folderNodeId, string? parentNodeId, string displayName) { }
+        public void EnsureVariable(string variableNodeId, string? parentFolderNodeId, string displayName, string dataType, bool writable, string? historianTagname = null, bool isArray = false, uint? arrayLength = null) { }
+        public void RebuildAddressSpace() => throw new InvalidOperationException("simulated rebuild fault");
+        public void RaiseNodesAddedModelChange(string affectedNodeId) { }
+    }
+
+    /// <summary>A no-op sink — a clean apply (no failures).</summary>
+    private sealed class NoopSink : IOpcUaAddressSpaceSink
+    {
+        public void WriteValue(string nodeId, object? value, OpcUaQuality quality, DateTime sourceTimestampUtc) { }
+        public void WriteAlarmCondition(string alarmNodeId, AlarmConditionSnapshot state, DateTime occurredUtc) { }
+        public void MaterialiseAlarmCondition(string alarmNodeId, string equipmentNodeId, string displayName, string alarmType, int severity, bool isNative = false) { }
+        public void EnsureFolder(string folderNodeId, string? parentNodeId, string displayName) { }
+        public void EnsureVariable(string variableNodeId, string? parentFolderNodeId, string displayName, string dataType, bool writable, string? historianTagname = null, bool isArray = false, uint? arrayLength = null) { }
+        public void RebuildAddressSpace() { }
+        public void RaiseNodesAddedModelChange(string affectedNodeId) { }
+    }
+
+    /// <summary>Listens to a single instrument by name on the central meter and tallies value + tags.</summary>
+    private sealed class MeterRecorder : IDisposable
+    {
+        private readonly string _name;
+        private readonly MeterListener _listener;
+        private long _total;
+        private readonly List<KeyValuePair<string, object?>[]> _tagSets = new();
+        private readonly object _gate = new();
+
+        public MeterRecorder(string instrumentName)
+        {
+            _name = instrumentName;
+            _listener = new MeterListener
+            {
+                InstrumentPublished = (instrument, listener) =>
+                {
+                    if (instrument.Meter.Name == OtOpcUaTelemetry.MeterName && instrument.Name == _name)
+                        listener.EnableMeasurementEvents(instrument);
+                }
+            };
+            _listener.SetMeasurementEventCallback<long>((_, value, tags, _) =>
+            {
+                lock (_gate)
+                {
+                    _total += value;
+                    _tagSets.Add(tags.ToArray());
+                }
+            });
+            _listener.Start();
+        }
+
+        public long Total { get { lock (_gate) return _total; } }
+
+        public int WithTag(string key, string value)
+        {
+            lock (_gate) return _tagSets.Count(set => set.Any(t => t.Key == key && Equals(t.Value, value)));
+        }
+
+        public void Dispose() => _listener.Dispose();
+    }
+}