feat(deploy): additive degraded flag on ApplyAck + publish-actor apply outcome reply (#601)

ApplyAck gains trailing Degraded/DegradedReason (defaulted, so an old node's DPS JSON or gRPC ack decodes exactly as before). mesh_transport.proto adds ApplyAckMessage.degraded = 6 / optional degraded_reason = 7. OpcUaPublishActor replies RebuildCompleted(Degraded, Reason) on every handler path when the RebuildAddressSpace carries ReportOutcome. Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
2026-10-08 17:19:57 -04:00
parent 9b74a65b25
commit 521e1eb89b
6 changed files with 286 additions and 11 deletions
@@ -7,9 +7,24 @@ public enum ApplyAckOutcome { Applied, Failed }
 /// <summary>
 /// Per-node acknowledgment returned by <c>DriverHostActor</c> to the dispatching coordinator.
 /// </summary>
+/// <param name="DeploymentId">The deployment being acknowledged.</param>
+/// <param name="NodeId">The acknowledging node.</param>
+/// <param name="Outcome">Whether the node applied the deployment.</param>
+/// <param name="FailureReason">Why the apply failed; null on <see cref="ApplyAckOutcome.Applied"/>.</param>
+/// <param name="CorrelationId">The dispatch's correlation id.</param>
+/// <param name="Degraded">
+///     Gitea #601: <see langword="true"/> when the node applied the deployment (the revision advanced) but
+///     its OPC UA address-space apply was DEGRADED — swallowed per-node sink faults, a failed rebuild, or no
+///     outcome reported in time — so the served address space may be partial. Only meaningful with
+///     <see cref="ApplyAckOutcome.Applied"/>. Trailing and defaulted so an ack from a node built before
+///     this field existed (DPS JSON or gRPC without field 6) decodes exactly as it always did: clean.
+/// </param>
+/// <param name="DegradedReason">A short, operator-readable description of the degradation, or null.</param>
 public sealed record ApplyAck(
    DeploymentId DeploymentId,
    NodeId NodeId,
    ApplyAckOutcome Outcome,
    string? FailureReason,
-    CorrelationId CorrelationId);
+    CorrelationId CorrelationId,
+    bool Degraded = false,
+    string? DegradedReason = null);
@@ -216,6 +216,11 @@ public static class MeshTransportDtoMapper
        if (message.FailureReason is not null)
            proto.FailureReason = message.FailureReason;

+        // #601 — additive fields; a clean ack leaves both unset, byte-identical to a pre-#601 encode.
+        proto.Degraded = message.Degraded;
+        if (message.DegradedReason is not null)
+            proto.DegradedReason = message.DegradedReason;
+
        return proto;
    }

@@ -249,7 +254,10 @@ public static class MeshTransportDtoMapper
            NodeId.Parse(proto.NodeId),
            outcome,
            proto.HasFailureReason ? proto.FailureReason : null,
-            CorrelationId.Parse(proto.CorrelationId));
+            CorrelationId.Parse(proto.CorrelationId),
+            // #601 — absent on an old node's ack, which decodes as false/null: today's clean ack.
+            Degraded: proto.Degraded,
+            DegradedReason: proto.HasDegradedReason ? proto.DegradedReason : null);
    }

    // ---------------------------------------------------------------- Envelope
@@ -105,6 +105,11 @@ message ApplyAckMessage {
  ApplyOutcome outcome = 3;
  optional string failure_reason = 4;                     // nullable in the record — presence distinguishes null from ""
  string correlation_id = 5;                              // CorrelationId (Guid) in its "N" hex form
+  // Gitea #601 — ADDITIVE. True when the node applied the deployment but its address-space apply was
+  // degraded (sink faults, failed rebuild, or no outcome in time). An old node never sets it, so it decodes
+  // as false: exactly today's clean ack. Only meaningful with APPLY_OUTCOME_APPLIED.
+  bool degraded = 6;
+  optional string degraded_reason = 7;                    // short operator-readable reason; presence = set
 }

 // Mirrors ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy.ApplyAckOutcome, shifted by one so 0 stays
@@ -100,8 +100,31 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
    ///     that cannot succeed, leaving the rebuild a no-op and clients browsing an empty server. Null
    ///     on the normal path, where loading from the ConfigDb by id is correct.
    /// </param>
+    /// <param name="ReportOutcome">
+    ///     Gitea #601 — when <see langword="true"/>, the actor replies to the sender with a
+    ///     <see cref="RebuildCompleted"/> once the apply outcome is known (clean or degraded), on EVERY path
+    ///     through the handler. <c>DriverHostActor</c> sets it on the deploy path and sends the deployment's
+    ///     ack only after that reply, so a degraded apply can no longer seal a deployment green. Off for the
+    ///     boot/restore paths, which re-ack nothing.
+    /// </param>
    public sealed record RebuildAddressSpace(
-        CorrelationId Correlation, DeploymentId? DeploymentId = null, byte[]? Artifact = null);
+        CorrelationId Correlation, DeploymentId? DeploymentId = null, byte[]? Artifact = null,
+        bool ReportOutcome = false);
+
+    /// <summary>
+    ///     Gitea #601 — the reply to a <see cref="RebuildAddressSpace"/> carrying
+    ///     <see cref="RebuildAddressSpace.ReportOutcome"/>: whether the address-space apply was clean.
+    /// </summary>
+    /// <param name="Correlation">The rebuild's correlation id, echoed.</param>
+    /// <param name="DeploymentId">The rebuild's deployment id, echoed (null for a legacy caller).</param>
+    /// <param name="Degraded">
+    ///     <see langword="true"/> when the apply swallowed per-node sink faults, its rebuild failed, the
+    ///     pipeline threw, or there was no usable artifact — i.e. the served address space may be partial or
+    ///     stale. An empty plan (nothing changed) is clean.
+    /// </param>
+    /// <param name="DegradedReason">A short operator-readable reason; null when clean.</param>
+    public sealed record RebuildCompleted(
+        CorrelationId Correlation, DeploymentId? DeploymentId, bool Degraded, string? DegradedReason);

    public sealed record ServiceLevelChanged(byte ServiceLevel);

@@ -437,6 +460,17 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
    }

    private void HandleRebuild(RebuildAddressSpace msg)
+    {
+        // Captured up front: the pipeline below is synchronous on this actor's thread today, but the reply
+        // must reach whoever asked even if a future edit introduces an await into it.
+        var requester = Sender;
+        var (degraded, reason) = HandleRebuildCore(msg);
+        if (msg.ReportOutcome)
+            requester.Tell(new RebuildCompleted(msg.Correlation, msg.DeploymentId, degraded, reason));
+    }
+
+    /// <summary>The rebuild pipeline. Returns whether the apply was degraded, and why (Gitea #601).</summary>
+    private (bool Degraded, string? Reason) HandleRebuildCore(RebuildAddressSpace msg)
    {
        using var span = OtOpcUaTelemetry.StartAddressSpaceRebuildSpan();
        span?.SetTag("otopcua.correlation_id", msg.Correlation.ToString());
@@ -460,8 +494,9 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
            {
                _log.Error(ex, "OpcUaPublish: sink.RebuildAddressSpace threw (correlation={Correlation})",
                    msg.Correlation);
+                return (true, $"address-space rebuild threw: {ex.Message}");
            }
-            return;
+            return (false, null);
        }

        try
@@ -499,7 +534,7 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
                    _log.Debug(
                        "OpcUaPublish: no artifact to materialise yet (deployment={Id}, correlation={Correlation})",
                        msg.DeploymentId, msg.Correlation);
-                return;
+                return (true, "no usable artifact reached the address-space apply; the previously served address space was kept");
            }

            // Parse the artifact ONCE for this rebuild. Every consumer below (data-plane authority, then the
@@ -532,7 +567,7 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
            {
                _log.Debug("OpcUaPublish: rebuild requested but plan is empty (correlation={Correlation})",
                    msg.Correlation);
-                return;
+                return (false, null);
            }

            var outcome = _applier.Apply(plan);
@@ -608,16 +643,19 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
                    "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);
+                return (true, outcome.RebuildFailed
+                    ? $"address-space rebuild failed (failedNodes={failedNodes})"
+                    : $"address-space apply degraded (failedNodes={failedNodes}); not recorded as applied, the next deploy of this revision re-diffs and repairs it");
            }
-            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);
-            }
+
+            _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);
+            return (false, null);
        }
        catch (Exception ex)
        {
            _log.Error(ex, "OpcUaPublish: rebuild pipeline threw (correlation={Correlation})", msg.Correlation);
+            return (true, $"address-space apply threw: {ex.Message}");
        }
    }

@@ -201,6 +201,61 @@ public class MeshTransportDtoMapperTests
        back.FailureReason.ShouldBeEmpty();
    }

+    [Fact]
+    public void ApplyAck_Degraded_RoundTrips_WithItsReason()
+    {
+        // #601: an Applied ack whose address-space apply was DEGRADED must reach central as degraded —
+        // a node that loses the flag on the wire reports a clean apply it did not have.
+        var domain = new ApplyAck(
+            new DeploymentId(DeploymentGuid),
+            new NodeId("site-a-1"),
+            ApplyAckOutcome.Applied,
+            null,
+            new CorrelationId(CorrelationGuid),
+            Degraded: true,
+            DegradedReason: "failedNodes=3");
+
+        var back = MeshTransportDtoMapper.FromProto(Wire(MeshTransportDtoMapper.ToProto(domain)));
+
+        back.ShouldBe(domain);
+        back.Degraded.ShouldBeTrue();
+        back.DegradedReason.ShouldBe("failedNodes=3");
+    }
+
+    [Fact]
+    public void ApplyAck_Clean_EncodesNoDegradedFields()
+    {
+        var proto = MeshTransportDtoMapper.ToProto(new ApplyAck(
+            new DeploymentId(DeploymentGuid), new NodeId("site-a-1"), ApplyAckOutcome.Applied, null,
+            new CorrelationId(CorrelationGuid)));
+
+        proto.Degraded.ShouldBeFalse();
+        proto.HasDegradedReason.ShouldBeFalse();
+    }
+
+    [Fact]
+    public void ApplyAck_FromAnOldNode_WithoutTheDegradedFields_DecodesAsTodaysCleanAck()
+    {
+        // #601 compatibility: a node built before the degraded fields existed sends exactly fields 1-5.
+        // Its ack must decode exactly as it always did — not degraded, no reason.
+        var oldNodeBytes = new ApplyAckMessage
+        {
+            DeploymentId = DeploymentGuid.ToString("N"),
+            NodeId = "site-a-1",
+            Outcome = ApplyOutcome.Applied,
+            CorrelationId = CorrelationGuid.ToString("N"),
+        }.ToByteArray();
+
+        var back = MeshTransportDtoMapper.FromProto(ApplyAckMessage.Parser.ParseFrom(oldNodeBytes));
+
+        back.Outcome.ShouldBe(ApplyAckOutcome.Applied);
+        back.Degraded.ShouldBeFalse();
+        back.DegradedReason.ShouldBeNull();
+        back.ShouldBe(new ApplyAck(
+            new DeploymentId(DeploymentGuid), new NodeId("site-a-1"), ApplyAckOutcome.Applied, null,
+            new CorrelationId(CorrelationGuid)));
+    }
+
    [Fact]
    public void ApplyAck_UnspecifiedOutcome_Throws_RatherThanDecodingToAMeaningfulValue()
    {
@@ -0,0 +1,154 @@
+using System.Text.Json;
+using Akka.Actor;
+using Microsoft.Extensions.Logging.Abstractions;
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
+using ZB.MOM.WW.OtOpcUa.Commons.Types;
+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>
+///     Gitea #601 — the deploy ack is sent only once the address-space apply outcome is known, so the
+///     publish actor must REPORT that outcome when asked (<see cref="OpcUaPublishActor.RebuildAddressSpace.ReportOutcome"/>):
+///     clean for a clean apply, degraded for swallowed per-node sink faults or a pipeline throw, and nothing
+///     at all when the caller did not ask (boot/restore paths re-ack nothing).
+/// </summary>
+/// <remarks>
+///     In the <c>opcua-publish-meters</c> collection because the degraded cases increment
+///     <c>otopcua.opcua.apply.failed</c>, which the sibling classes assert stays at zero.
+/// </remarks>
+[Collection("opcua-publish-meters")]
+public sealed class OpcUaPublishActorApplyOutcomeTests : RuntimeActorTestBase
+{
+    [Fact]
+    public void A_clean_apply_reports_a_clean_outcome_to_the_requester()
+    {
+        var sink = new FaultingSink(throwOnEnsureFolder: false);
+        var applier = new AddressSpaceApplier(sink, NullLogger<AddressSpaceApplier>.Instance);
+        var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(sink: sink, applier: applier));
+        var correlation = CorrelationId.NewId();
+        var deployment = DeploymentId.NewId();
+
+        actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(
+            correlation, deployment, BuildEquipmentArtifact("eq-1"), ReportOutcome: true));
+
+        var done = ExpectMsg<OpcUaPublishActor.RebuildCompleted>(PresenceBudget);
+        done.Correlation.ShouldBe(correlation);
+        done.DeploymentId.ShouldBe(deployment);
+        done.Degraded.ShouldBeFalse();
+        done.DegradedReason.ShouldBeNull();
+    }
+
+    [Fact]
+    public void A_degraded_non_rebuild_apply_reports_degraded_with_the_failed_node_count()
+    {
+        // Every EnsureFolder throws — the applier swallows each and counts it, the classic degraded apply
+        // that leaves _lastApplied un-advanced and used to seal the deployment green.
+        var sink = new FaultingSink(throwOnEnsureFolder: true);
+        var applier = new AddressSpaceApplier(sink, NullLogger<AddressSpaceApplier>.Instance);
+        var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(sink: sink, applier: applier));
+
+        actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(
+            CorrelationId.NewId(), DeploymentId.NewId(), BuildEquipmentArtifact("eq-1"), ReportOutcome: true));
+
+        var done = ExpectMsg<OpcUaPublishActor.RebuildCompleted>(PresenceBudget);
+        done.Degraded.ShouldBeTrue();
+        done.DegradedReason.ShouldNotBeNull();
+        done.DegradedReason.ShouldContain("failedNodes=");
+    }
+
+    [Fact]
+    public void A_pipeline_that_throws_reports_degraded()
+    {
+        var sink = new FaultingSink(throwOnEnsureFolder: false);
+        var applier = new AddressSpaceApplier(sink, NullLogger<AddressSpaceApplier>.Instance);
+        var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(sink: sink, applier: applier));
+
+        // Non-empty bytes that are not JSON: DeploymentArtifact.Open throws inside the rebuild pipeline.
+        actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(
+            CorrelationId.NewId(), DeploymentId.NewId(), "not json"u8.ToArray(), ReportOutcome: true));
+
+        var done = ExpectMsg<OpcUaPublishActor.RebuildCompleted>(PresenceBudget);
+        done.Degraded.ShouldBeTrue();
+        done.DegradedReason.ShouldNotBeNull();
+    }
+
+    [Fact]
+    public void An_unchanged_composition_reports_clean()
+    {
+        var sink = new FaultingSink(throwOnEnsureFolder: false);
+        var applier = new AddressSpaceApplier(sink, NullLogger<AddressSpaceApplier>.Instance);
+        var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(sink: sink, applier: applier));
+        var artifact = BuildEquipmentArtifact("eq-1");
+
+        actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId(), DeploymentId.NewId(), artifact, ReportOutcome: true));
+        ExpectMsg<OpcUaPublishActor.RebuildCompleted>(PresenceBudget).Degraded.ShouldBeFalse();
+
+        // Same composition again ⇒ empty plan ⇒ nothing to apply, nothing degraded.
+        actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId(), DeploymentId.NewId(), artifact, ReportOutcome: true));
+        ExpectMsg<OpcUaPublishActor.RebuildCompleted>(PresenceBudget).Degraded.ShouldBeFalse();
+    }
+
+    [Fact]
+    public void Without_ReportOutcome_nothing_is_replied()
+    {
+        var sink = new FaultingSink(throwOnEnsureFolder: true);
+        var applier = new AddressSpaceApplier(sink, NullLogger<AddressSpaceApplier>.Instance);
+        var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(sink: sink, applier: applier));
+
+        actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(
+            CorrelationId.NewId(), DeploymentId.NewId(), BuildEquipmentArtifact("eq-1")));
+
+        // Positive control first: a follow-up REQUESTED outcome must arrive, which proves the first
+        // message was fully processed before the absence assertion is read.
+        actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(
+            CorrelationId.NewId(), DeploymentId.NewId(), BuildEquipmentArtifact("eq-2"), ReportOutcome: true));
+        var only = ExpectMsg<OpcUaPublishActor.RebuildCompleted>(PresenceBudget);
+        only.Degraded.ShouldBeTrue();
+        ExpectNoMsg(TimeSpan.FromMilliseconds(200));
+    }
+
+    private static byte[] BuildEquipmentArtifact(string equipmentId) =>
+        JsonSerializer.SerializeToUtf8Bytes(new
+        {
+            UnsAreas = new[] { new { UnsAreaId = "area-1", ClusterId = "c1", Name = "Area" } },
+            UnsLines = new[] { new { UnsLineId = "line-1", UnsAreaId = "area-1", Name = "Line" } },
+            Equipment = new[]
+            {
+                new
+                {
+                    EquipmentId = equipmentId,
+                    DriverInstanceId = (string?)null,
+                    UnsLineId = "line-1",
+                    Name = equipmentId,
+                    MachineCode = equipmentId.ToUpperInvariant(),
+                },
+            },
+            DriverInstances = Array.Empty<object>(),
+            Namespaces = Array.Empty<object>(),
+            Tags = Array.Empty<object>(),
+            ScriptedAlarms = Array.Empty<object>(),
+        });
+
+    /// <summary>A sink that optionally throws from <c>EnsureFolder</c> (a swallowed per-node fault).</summary>
+    private sealed class FaultingSink(bool throwOnEnsureFolder) : IOpcUaAddressSpaceSink
+    {
+        public void WriteValue(string nodeId, object? value, uint statusCode, DateTime sourceTimestampUtc, AddressSpaceRealm realm = AddressSpaceRealm.Uns) { }
+        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)
+        {
+            if (throwOnEnsureFolder) throw new InvalidOperationException("simulated sink fault");
+        }
+        public void EnsureVariable(string variableNodeId, string? parentFolderNodeId, string displayName, string dataType, bool writable, AddressSpaceRealm realm, string? historianTagname = null, bool isArray = false, uint? arrayLength = null, bool refuseArrayOnScalar = false) { }
+        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") { }
+    }
+}