dohertj2/lmxopcua
1
0
Fork 0
Code Issues 14 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(deploy): coordinator records degraded acks; deployment seals but names degraded nodes (#601)

Browse Source 
An Applied+Degraded ack keeps NodeDeploymentState.Status = Applied with
FailureReason 'Applied degraded: <reason>' (no enum/migration change). The
deployment still seals (the revision is running; latest-sealed drives drift)
but Deployment.FailureReason names the degraded nodes and the seal logs at
Warning. Recovered from the DB on singleton failover. Old-node acks (no field,
DPS JSON or gRPC) behave exactly as before.

Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
This commit is contained in:
Joseph Doherty
2026-10-08 17:32:09 -04:00
parent 90d7f6dfca
commit 49dfabc141
2 changed files with 289 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Coordinators/ConfigPublishCoordinator.cs
+69 -1
View File
@@ -83,6 +83,30 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers
private static readonly IEqualityComparer<NodeId> NodeIdComparer = new CaseInsensitiveNodeIdComparer();
private readonly Dictionary<NodeId, ApplyAckOutcome> _acks = new(NodeIdComparer);
/// <summary>Gitea #601 — the expected nodes whose Applied ack was DEGRADED (address-space apply partial).
/// Recovered from the DB on failover via <see cref="IsDegraded"/>.</summary>
private readonly HashSet<NodeId> _degraded = new(NodeIdComparer);
/// <summary>
/// Gitea #601 — prefix of <c>NodeDeploymentState.FailureReason</c> on an <b>Applied</b> 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 <see cref="NodeDeploymentStatus.Applied"/>
/// and the degradation rides the existing reason column, which <c>/fleet</c> already renders.
/// </summary>
public const string DegradedReasonPrefix = "Applied degraded: ";
/// <summary>
/// Gitea #601 — whether a <c>NodeDeploymentState</c> row records a degraded apply: status
/// <see cref="NodeDeploymentStatus.Applied"/> with a non-empty <c>FailureReason</c>. A clean Applied row
/// never carries a reason (neither this coordinator nor <c>DriverHostActor</c> writes one), so an
/// old node's row reads as clean exactly as before.
/// </summary>
/// <param name="status">The row's status.</param>
/// <param name="failureReason">The row's failure reason.</param>
/// <returns><see langword="true"/> for an Applied-but-degraded row.</returns>
public static bool IsDegraded(NodeDeploymentStatus status, string? failureReason) =>
status == NodeDeploymentStatus.Applied && !string.IsNullOrEmpty(failureReason);
private DeploymentId? _current;
private HashSet<NodeId> _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";
tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/ConfigPublishCoordinatorDegradedAckTests.cs
+220
View File
@@ -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;
/// <summary>
/// Gitea #601 — central's side of the honest deploy ack. A node that applied a deployment but whose
/// address-space apply was DEGRADED acks <see cref="ApplyAckOutcome.Applied"/> with
/// <see cref="ApplyAck.Degraded"/>. 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
/// <c>NodeDeploymentState</c> row keeps status Applied with its degraded reason in <c>FailureReason</c>,
/// and the <c>Deployment</c> row's <c>FailureReason</c> names the degraded nodes. An ack from an old node
/// (no degraded field) is handled exactly as before.
/// </summary>
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<OtOpcUaConfigDbContext> dbFactory, DeploymentStatus expected) =>
AwaitAssert(() =>
{
using var db = dbFactory.CreateDbContext();
db.Deployments.Single().Status.ShouldBe(expected);
}, duration: TimeSpan.FromSeconds(5));
private static void SeedNodes(IDbContextFactory<OtOpcUaConfigDbContext> 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<OtOpcUaConfigDbContext> 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;
}
}