diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Coordinators/ConfigPublishCoordinator.cs b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Coordinators/ConfigPublishCoordinator.cs index b52beb87..c35fd0d5 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Coordinators/ConfigPublishCoordinator.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Coordinators/ConfigPublishCoordinator.cs @@ -83,6 +83,30 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers private static readonly IEqualityComparer NodeIdComparer = new CaseInsensitiveNodeIdComparer(); private readonly Dictionary _acks = new(NodeIdComparer); + /// Gitea #601 — the expected nodes whose Applied ack was DEGRADED (address-space apply partial). + /// Recovered from the DB on failover via . + private readonly HashSet _degraded = new(NodeIdComparer); + + /// + /// Gitea #601 — prefix of NodeDeploymentState.FailureReason on an Applied row whose apply + /// was degraded. The status model is deliberately unchanged (no new enum value, no migration): a + /// degraded node DID apply the revision, so its status stays + /// and the degradation rides the existing reason column, which /fleet already renders. + /// + public const string DegradedReasonPrefix = "Applied degraded: "; + + /// + /// Gitea #601 — whether a NodeDeploymentState row records a degraded apply: status + /// with a non-empty FailureReason. A clean Applied row + /// never carries a reason (neither this coordinator nor DriverHostActor writes one), so an + /// old node's row reads as clean exactly as before. + /// + /// The row's status. + /// The row's failure reason. + /// for an Applied-but-degraded row. + public static bool IsDegraded(NodeDeploymentStatus status, string? failureReason) => + status == NodeDeploymentStatus.Applied && !string.IsNullOrEmpty(failureReason); + private DeploymentId? _current; private HashSet _expectedAcks = new(NodeIdComparer); @@ -149,9 +173,12 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers _expectedAcks = nodeStates.Select(s => NodeId.Parse(s.NodeId)).ToHashSet(NodeIdComparer); foreach (var s in nodeStates.Where(s => s.Status != NodeDeploymentStatus.Applying)) + { _acks[NodeId.Parse(s.NodeId)] = s.Status == NodeDeploymentStatus.Applied ? ApplyAckOutcome.Applied : ApplyAckOutcome.Failed; + if (IsDegraded(s.Status, s.FailureReason)) _degraded.Add(NodeId.Parse(s.NodeId)); + } // Resume the deadline timer using the remaining time. The deadline runs from when the // deployment was first marked AwaitingApplyAcks (Deployment.CreatedAtUtc is a close enough @@ -174,6 +201,7 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers { _current = msg.DeploymentId; _acks.Clear(); + _degraded.Clear(); // Seed NodeDeploymentState rows so a failover knows which nodes were expected to ack. using (var db = _dbFactory.CreateDbContext()) @@ -237,6 +265,9 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers } _acks[msg.NodeId] = msg.Outcome; + // #601 — an old node never sets Degraded, so its ack takes the clean branch exactly as before. + if (IsDegradedAck(msg)) _degraded.Add(msg.NodeId); + else _degraded.Remove(msg.NodeId); PersistNodeAck(msg); if (_acks.Count < _expectedAcks.Count) return; @@ -258,15 +289,51 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers ? NodeDeploymentStatus.Applied : NodeDeploymentStatus.Failed; row.AppliedAtUtc = DateTime.UtcNow; - row.FailureReason = msg.FailureReason; + // #601 — an Applied+degraded ack keeps status Applied and records WHY in the reason column, so the row + // reads as degraded (IsDegraded) rather than clean. A clean or Failed ack writes what it always did. + row.FailureReason = IsDegradedAck(msg) + ? Truncate(DegradedReasonPrefix + (string.IsNullOrWhiteSpace(msg.DegradedReason) + ? "address-space apply degraded (no reason given)" + : msg.DegradedReason)) + : msg.FailureReason; db.SaveChanges(); } + private static bool IsDegradedAck(ApplyAck msg) => msg.Outcome == ApplyAckOutcome.Applied && msg.Degraded; + + // FailureReason is nvarchar(2048) on both tables (OtOpcUaConfigDbContext); a reason built from a node's + // free text must never fail the SaveChanges that records the ack. + private const int FailureReasonMaxLength = 2048; + + private static string Truncate(string s) => + s.Length <= FailureReasonMaxLength ? s : s[..(FailureReasonMaxLength - 1)] + "…"; + private void SealDeployment() { if (_current is null) return; using var db = _dbFactory.CreateDbContext(); UpdateDeploymentStatus(db, _current.Value, DeploymentStatus.Sealed, sealNow: true); + if (_degraded.Count > 0) + { + // #601 — still Sealed: every node DID apply this revision and is running it, and "latest sealed" is + // what drift detection and the publish actor's fallback key on, so a degraded deploy that was not + // sealed would make the fleet look one revision behind what it actually runs. Sealed is not the + // same as clean, though: the deployment's reason names the degraded nodes (per-node detail is on + // each NodeDeploymentState row) and the seal logs at Warning. + var nodes = _degraded.Select(n => n.Value).Order(StringComparer.OrdinalIgnoreCase).ToList(); + var d = db.Deployments.FirstOrDefault(x => x.DeploymentId == _current.Value.Value); + if (d is not null) + d.FailureReason = Truncate( + $"Sealed with a degraded address-space apply on {nodes.Count} node(s): {string.Join(", ", nodes)}. " + + "See each node's NodeDeploymentState.FailureReason; re-deploying the same revision re-applies it."); + db.SaveChanges(); + _log.Warning( + "Deployment {Id} sealed DEGRADED (acks={Count}) — the address-space apply was degraded on: {Nodes}", + _current.Value, _acks.Count, string.Join(", ", nodes)); + ResetForNext(); + return; + } + db.SaveChanges(); _log.Info("Deployment {Id} sealed (acks={Count})", _current.Value, _acks.Count); ResetForNext(); @@ -318,6 +385,7 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers _current = null; _expectedAcks.Clear(); _acks.Clear(); + _degraded.Clear(); } private const string DeadlineTimerKey = "apply-deadline"; diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/ConfigPublishCoordinatorDegradedAckTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/ConfigPublishCoordinatorDegradedAckTests.cs new file mode 100644 index 00000000..f67424b1 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/ConfigPublishCoordinatorDegradedAckTests.cs @@ -0,0 +1,220 @@ +using System.Text; +using Akka.Actor; +using Microsoft.EntityFrameworkCore; +using Newtonsoft.Json.Linq; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy; +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.ControlPlane.Coordinators; +using ZB.MOM.WW.OtOpcUa.ControlPlane.Tests.Harness; + +namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Tests; + +/// +/// Gitea #601 — central's side of the honest deploy ack. A node that applied a deployment but whose +/// address-space apply was DEGRADED acks with +/// . The deployment still SEALS (the revision is what the node now runs, and +/// "latest sealed" drives drift + the publish actor's fallback) but never looks clean: the node's +/// NodeDeploymentState row keeps status Applied with its degraded reason in FailureReason, +/// and the Deployment row's FailureReason names the degraded nodes. An ack from an old node +/// (no degraded field) is handled exactly as before. +/// +public sealed class ConfigPublishCoordinatorDegradedAckTests : ControlPlaneActorTestBase +{ + private static readonly RevisionHash TestRevision = RevisionHash.Parse(new string('c', 64)); + private static readonly TimeSpan NoTimeout = TimeSpan.FromMinutes(5); + private const string NodeA = "site-a-1:4053"; + private const string NodeB = "site-a-2:4053"; + + [Fact] + public void A_degraded_ack_seals_but_is_recorded_as_degraded_on_the_node_and_the_deployment() + { + var dbFactory = NewInMemoryDbFactory(); + var deploymentId = SeedDispatchingDeployment(dbFactory); + SeedNodes(dbFactory, NodeA, NodeB); + var actor = Sys.ActorOf(ConfigPublishCoordinator.Props(dbFactory, NoTimeout)); + actor.Tell(new DispatchDeployment(deploymentId, TestRevision, CorrelationId.NewId())); + AwaitStatus(dbFactory, DeploymentStatus.AwaitingApplyAcks); + + actor.Tell(new ApplyAck(deploymentId, NodeId.Parse(NodeA), ApplyAckOutcome.Applied, null, + CorrelationId.NewId())); + actor.Tell(new ApplyAck(deploymentId, NodeId.Parse(NodeB), ApplyAckOutcome.Applied, null, + CorrelationId.NewId(), Degraded: true, DegradedReason: "address-space apply degraded (failedNodes=3)")); + + AwaitStatus(dbFactory, DeploymentStatus.Sealed); + using var db = dbFactory.CreateDbContext(); + + var deployment = db.Deployments.Single(); + deployment.FailureReason.ShouldNotBeNull(); + deployment.FailureReason.ShouldContain("degraded"); + deployment.FailureReason.ShouldContain(NodeB); + deployment.FailureReason.ShouldNotContain(NodeA); + + var rowB = db.NodeDeploymentStates.Single(s => s.NodeId == NodeB); + rowB.Status.ShouldBe(NodeDeploymentStatus.Applied); + rowB.FailureReason.ShouldNotBeNull(); + rowB.FailureReason.ShouldStartWith(ConfigPublishCoordinator.DegradedReasonPrefix); + rowB.FailureReason.ShouldContain("failedNodes=3"); + ConfigPublishCoordinator.IsDegraded(rowB.Status, rowB.FailureReason).ShouldBeTrue(); + + var rowA = db.NodeDeploymentStates.Single(s => s.NodeId == NodeA); + rowA.Status.ShouldBe(NodeDeploymentStatus.Applied); + rowA.FailureReason.ShouldBeNull(); + ConfigPublishCoordinator.IsDegraded(rowA.Status, rowA.FailureReason).ShouldBeFalse(); + } + + [Fact] + public void Old_node_acks_without_the_degraded_field_seal_clean_exactly_as_before() + { + // The five-argument constructor IS the pre-#601 ack shape: Degraded defaults to false. + var dbFactory = NewInMemoryDbFactory(); + var deploymentId = SeedDispatchingDeployment(dbFactory); + SeedNodes(dbFactory, NodeA, NodeB); + var actor = Sys.ActorOf(ConfigPublishCoordinator.Props(dbFactory, NoTimeout)); + actor.Tell(new DispatchDeployment(deploymentId, TestRevision, CorrelationId.NewId())); + AwaitStatus(dbFactory, DeploymentStatus.AwaitingApplyAcks); + + actor.Tell(new ApplyAck(deploymentId, NodeId.Parse(NodeA), ApplyAckOutcome.Applied, null, CorrelationId.NewId())); + actor.Tell(new ApplyAck(deploymentId, NodeId.Parse(NodeB), ApplyAckOutcome.Applied, null, CorrelationId.NewId())); + + AwaitStatus(dbFactory, DeploymentStatus.Sealed); + using var db = dbFactory.CreateDbContext(); + db.Deployments.Single().FailureReason.ShouldBeNull(); + db.NodeDeploymentStates.All(s => s.Status == NodeDeploymentStatus.Applied && s.FailureReason == null) + .ShouldBeTrue(); + } + + [Fact] + public void A_degraded_ack_alongside_a_failed_one_is_still_PartiallyFailed() + { + var dbFactory = NewInMemoryDbFactory(); + var deploymentId = SeedDispatchingDeployment(dbFactory); + SeedNodes(dbFactory, NodeA, NodeB); + var actor = Sys.ActorOf(ConfigPublishCoordinator.Props(dbFactory, NoTimeout)); + actor.Tell(new DispatchDeployment(deploymentId, TestRevision, CorrelationId.NewId())); + AwaitStatus(dbFactory, DeploymentStatus.AwaitingApplyAcks); + + actor.Tell(new ApplyAck(deploymentId, NodeId.Parse(NodeA), ApplyAckOutcome.Failed, "boom", CorrelationId.NewId())); + actor.Tell(new ApplyAck(deploymentId, NodeId.Parse(NodeB), ApplyAckOutcome.Applied, null, + CorrelationId.NewId(), Degraded: true, DegradedReason: "x")); + + AwaitStatus(dbFactory, DeploymentStatus.PartiallyFailed); + using var db = dbFactory.CreateDbContext(); + db.NodeDeploymentStates.Single(s => s.NodeId == NodeA).FailureReason.ShouldBe("boom"); + db.NodeDeploymentStates.Single(s => s.NodeId == NodeA).Status.ShouldBe(NodeDeploymentStatus.Failed); + } + + [Fact] + public void A_recovered_coordinator_remembers_a_degraded_ack_that_landed_before_failover() + { + // Node B's degraded ack was persisted by a previous singleton instance; this instance recovers the + // in-flight deployment from the DB and must still seal it as degraded when node A acks clean. + var dbFactory = NewInMemoryDbFactory(); + var deploymentId = SeedDispatchingDeployment(dbFactory, DeploymentStatus.AwaitingApplyAcks); + SeedNodes(dbFactory, NodeA, NodeB); + using (var db = dbFactory.CreateDbContext()) + { + db.NodeDeploymentStates.Add(new NodeDeploymentState + { + NodeId = NodeA, DeploymentId = deploymentId.Value, Status = NodeDeploymentStatus.Applying, + }); + db.NodeDeploymentStates.Add(new NodeDeploymentState + { + NodeId = NodeB, DeploymentId = deploymentId.Value, Status = NodeDeploymentStatus.Applied, + FailureReason = ConfigPublishCoordinator.DegradedReasonPrefix + "failedNodes=1", + AppliedAtUtc = DateTime.UtcNow, + }); + db.SaveChanges(); + } + + var actor = Sys.ActorOf(ConfigPublishCoordinator.Props(dbFactory, NoTimeout)); + actor.Tell(new ApplyAck(deploymentId, NodeId.Parse(NodeA), ApplyAckOutcome.Applied, null, CorrelationId.NewId())); + + AwaitStatus(dbFactory, DeploymentStatus.Sealed); + using var verify = dbFactory.CreateDbContext(); + var reason = verify.Deployments.Single().FailureReason; + reason.ShouldNotBeNull(); + reason.ShouldContain(NodeB); + } + + [Fact] + public void An_old_nodes_DPS_ack_without_the_degraded_fields_deserializes_as_a_clean_ack() + { + // The DPS ack path rides Akka's default serializer for the record. Serialize a NEW ack, strip the two + // #601 members from the payload (what an old node sends), and decode it with this build. + var ack = new ApplyAck(DeploymentId.NewId(), NodeId.Parse(NodeA), ApplyAckOutcome.Applied, null, + CorrelationId.NewId(), Degraded: true, DegradedReason: "will be stripped"); + var serializer = Sys.Serialization.FindSerializerFor(ack); + var json = JObject.Parse(Encoding.UTF8.GetString(serializer.ToBinary(ack))); + json.Remove("Degraded").ShouldBeTrue(); + json.Remove("DegradedReason").ShouldBeTrue(); + + var back = (ApplyAck)serializer.FromBinary( + Encoding.UTF8.GetBytes(json.ToString(Newtonsoft.Json.Formatting.None)), typeof(ApplyAck)); + + back.Outcome.ShouldBe(ApplyAckOutcome.Applied); + back.Degraded.ShouldBeFalse(); + back.DegradedReason.ShouldBeNull(); + back.DeploymentId.ShouldBe(ack.DeploymentId); + + // ...and the new fields survive the same serializer when present. + var full = (ApplyAck)serializer.FromBinary(serializer.ToBinary(ack), typeof(ApplyAck)); + full.ShouldBe(ack); + } + + private void AwaitStatus(IDbContextFactory dbFactory, DeploymentStatus expected) => + AwaitAssert(() => + { + using var db = dbFactory.CreateDbContext(); + db.Deployments.Single().Status.ShouldBe(expected); + }, duration: TimeSpan.FromSeconds(5)); + + private static void SeedNodes(IDbContextFactory dbFactory, params string[] nodes) + { + using var db = dbFactory.CreateDbContext(); + db.ServerClusters.Add(new ServerCluster + { + ClusterId = "SITE-A", + Name = "Site A", + Enterprise = "zb", + Site = "site-a", + NodeCount = 2, + RedundancyMode = RedundancyMode.Warm, + CreatedBy = "test", + }); + foreach (var nodeId in nodes) + { + db.ClusterNodes.Add(new ClusterNode + { + NodeId = nodeId, + ClusterId = "SITE-A", + Host = nodeId.Split(':')[0], + ApplicationUri = $"urn:OtOpcUa:{nodeId}", + Enabled = true, + CreatedBy = "test", + }); + } + db.SaveChanges(); + } + + private static DeploymentId SeedDispatchingDeployment( + IDbContextFactory dbFactory, + DeploymentStatus status = DeploymentStatus.Dispatching) + { + var id = DeploymentId.NewId(); + using var db = dbFactory.CreateDbContext(); + db.Deployments.Add(new Deployment + { + DeploymentId = id.Value, + RevisionHash = TestRevision.Value, + Status = status, + CreatedBy = "test", + }); + db.SaveChanges(); + return id; + } +}