From 521e1eb89ba03548f87186c7884c3f34c4adc151 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 8 Oct 2026 17:19:57 -0400 Subject: [PATCH] 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 --- .../Messages/Deploy/ApplyAck.cs | 17 +- .../Messages/Mesh/MeshTransportDtoMapper.cs | 10 +- .../Protos/mesh_transport.proto | 5 + .../OpcUa/OpcUaPublishActor.cs | 56 ++++++- .../MeshTransportDtoMapperTests.cs | 55 +++++++ .../OpcUaPublishActorApplyOutcomeTests.cs | 154 ++++++++++++++++++ 6 files changed, 286 insertions(+), 11 deletions(-) create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorApplyOutcomeTests.cs diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Messages/Deploy/ApplyAck.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Messages/Deploy/ApplyAck.cs index 10ff6f7d..6cb593cd 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Messages/Deploy/ApplyAck.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Messages/Deploy/ApplyAck.cs @@ -7,9 +7,24 @@ public enum ApplyAckOutcome { Applied, Failed } /// /// Per-node acknowledgment returned by DriverHostActor to the dispatching coordinator. /// +/// The deployment being acknowledged. +/// The acknowledging node. +/// Whether the node applied the deployment. +/// Why the apply failed; null on . +/// The dispatch's correlation id. +/// +/// Gitea #601: 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 +/// . 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. +/// +/// A short, operator-readable description of the degradation, or null. public sealed record ApplyAck( DeploymentId DeploymentId, NodeId NodeId, ApplyAckOutcome Outcome, string? FailureReason, - CorrelationId CorrelationId); + CorrelationId CorrelationId, + bool Degraded = false, + string? DegradedReason = null); diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Messages/Mesh/MeshTransportDtoMapper.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Messages/Mesh/MeshTransportDtoMapper.cs index f3d78255..4fe43816 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Messages/Mesh/MeshTransportDtoMapper.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Messages/Mesh/MeshTransportDtoMapper.cs @@ -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 diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Protos/mesh_transport.proto b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Protos/mesh_transport.proto index 1dd09580..848d45d2 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Protos/mesh_transport.proto +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Protos/mesh_transport.proto @@ -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 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 27a321f7..391bd4e3 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/OpcUa/OpcUaPublishActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/OpcUa/OpcUaPublishActor.cs @@ -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. /// + /// + /// Gitea #601 — when , the actor replies to the sender with a + /// once the apply outcome is known (clean or degraded), on EVERY path + /// through the handler. DriverHostActor 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. + /// public sealed record RebuildAddressSpace( - CorrelationId Correlation, DeploymentId? DeploymentId = null, byte[]? Artifact = null); + CorrelationId Correlation, DeploymentId? DeploymentId = null, byte[]? Artifact = null, + bool ReportOutcome = false); + + /// + /// Gitea #601 — the reply to a carrying + /// : whether the address-space apply was clean. + /// + /// The rebuild's correlation id, echoed. + /// The rebuild's deployment id, echoed (null for a legacy caller). + /// + /// 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. + /// + /// A short operator-readable reason; null when clean. + 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)); + } + + /// The rebuild pipeline. Returns whether the apply was degraded, and why (Gitea #601). + 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}"); } } diff --git a/tests/Core/ZB.MOM.WW.OtOpcUa.Commons.Tests/MeshTransportDtoMapperTests.cs b/tests/Core/ZB.MOM.WW.OtOpcUa.Commons.Tests/MeshTransportDtoMapperTests.cs index aa2adcee..ce19d0da 100644 --- a/tests/Core/ZB.MOM.WW.OtOpcUa.Commons.Tests/MeshTransportDtoMapperTests.cs +++ b/tests/Core/ZB.MOM.WW.OtOpcUa.Commons.Tests/MeshTransportDtoMapperTests.cs @@ -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() { diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorApplyOutcomeTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorApplyOutcomeTests.cs new file mode 100644 index 00000000..37d0773d --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorApplyOutcomeTests.cs @@ -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; + +/// +/// 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 (): +/// 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). +/// +/// +/// In the opcua-publish-meters collection because the degraded cases increment +/// otopcua.opcua.apply.failed, which the sibling classes assert stays at zero. +/// +[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.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(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.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(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.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(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.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(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(PresenceBudget).Degraded.ShouldBeFalse(); + } + + [Fact] + public void Without_ReportOutcome_nothing_is_replied() + { + var sink = new FaultingSink(throwOnEnsureFolder: true); + var applier = new AddressSpaceApplier(sink, NullLogger.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(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(), + Namespaces = Array.Empty(), + Tags = Array.Empty(), + ScriptedAlarms = Array.Empty(), + }); + + /// A sink that optionally throws from EnsureFolder (a swallowed per-node fault). + 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 notifierFolderNodeIds, AddressSpaceRealm notifierFolderRealm) { } + public void AddReference(string sourceNodeId, AddressSpaceRealm sourceRealm, string targetNodeId, AddressSpaceRealm targetRealm, string referenceType = "Organizes") { } + } +}