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

feat(deploy): coordinator sources its expected-ack set from ClusterNode rows

Browse Source 
Cuts ConfigPublishCoordinator's last genuinely mesh-bound dependency: central
must be able to name the nodes a deployment is for without sharing a gossip
ring with them, because Phase 2 splits the fleet into one mesh per Cluster.

Behaviour change, confirmed before implementing: a configured node that is
switched off is now expected, so deploying while it is down fails at the apply
deadline instead of sealing green without it. Under the membership rule the
operator was told the fleet was deployed when it was not. ClusterNode.Enabled
= 0 is the maintenance hatch, and the deadline log now names the silent nodes
and points at that hatch — "4/5 acks landed" would leave an operator reading
logs on five machines.

Two assumptions are now documented on the class rather than left implicit:
every ClusterNode row is a driver node (the DB has no role column and
deliberately does not gain one — it would drift from Cluster:Roles), and
ServerCluster.Enabled is not consulted because nothing else consults it.

Dropped from the plan: cluster-scope filtering of the expected set. There is
no cluster-scoped deployment to filter on — Deployment has no ClusterId,
ConfigComposer always snapshots the whole DB, and ResolveClusterScope is
node-side self-scoping of a fleet-wide artifact.

Positive control: reverting DiscoverDriverNodes to the membership scan turns
three of the four new tests red. The fourth pins the seal-empty branch and is
documented as derivation-insensitive rather than left to look like coverage.

ControlPlane.Tests 90/90.

Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW
This commit is contained in:
Joseph Doherty
2026-07-22 08:30:46 -04:00
parent da74ebd696
commit d88e245503
2 changed files with 290 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Coordinators/ConfigPublishCoordinator.cs
+71 -25
View File
@@ -1,5 +1,4 @@
using Akka.Actor;
using Akka.Cluster;
using Akka.Cluster.Tools.PublishSubscribe;
using Akka.Event;
using Microsoft.EntityFrameworkCore;
@@ -18,8 +17,32 @@ namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Coordinators;
/// has acked Applied. Per-node ACKs are persisted in <c>NodeDeploymentState</c> so a failover of
/// this singleton can recover in-flight state from the DB.
///
/// Discovery of the "expected ACK set" comes from <c>Akka.Cluster.State.Members</c> filtered by
/// the <c>driver</c> role — the DB does not own per-node role assignment.
/// Discovery of the "expected ACK set" comes from the <b>enabled <c>ClusterNode</c> rows</b>, not
/// from <c>Akka.Cluster.State.Members</c> (per-cluster mesh Phase 1). Central must be able to name
/// the nodes a deployment is for without sharing a gossip ring with them — Phase 2 splits the fleet
/// into one mesh per <c>Cluster</c>, after which central can no longer see a site node's membership
/// at all. This was the coordinator's last genuinely mesh-bound dependency.
///
/// <para>
/// <b>Behaviour change:</b> a configured node that is switched off is now <i>expected</i>, so a
/// deployment dispatched while it is down fails at the apply deadline instead of sealing green
/// without it. That is deliberate — under the membership rule the operator was told the fleet was
/// deployed when it was not. <c>ClusterNode.Enabled = 0</c> is the escape hatch for a node taken
/// down for maintenance.
/// </para>
/// <para>
/// <b>Every <c>ClusterNode</c> row is assumed to be a driver node.</b> The membership rule filtered
/// on the <c>driver</c> role; the DB has no per-node role column, and deliberately does not gain one
/// here (a second declaration of node roles would drift from <c>Cluster:Roles</c> in the node's own
/// appsettings). The assumption holds because the entity's whole shape — <c>Host</c>,
/// <c>OpcUaPort</c>, <c>ApplicationUri</c>, <c>ServiceLevelBase</c> — describes an OPC UA server
/// node. An admin-only node must not be given a <c>ClusterNode</c> row: it would be expected to ack
/// a deployment it has no <c>DriverHostActor</c> to apply, and every deploy would time out.
/// </para>
/// <para>
/// <c>ServerCluster.Enabled</c> is <b>not</b> consulted — nothing else in the codebase consults it,
/// so honouring it here would invent semantics. Disable the nodes, not the cluster row.
/// </para>
/// </summary>
public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers
{
@@ -119,11 +142,15 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers
{
_current = msg.DeploymentId;
_acks.Clear();
_expectedAcks = DiscoverDriverNodes();
// Seed NodeDeploymentState rows so a failover knows which nodes were expected to ack.
using (var db = _dbFactory.CreateDbContext())
{
// Discovery reuses this context rather than opening a second one — it reads the same
// ClusterNode table the NodeDeploymentState rows below are FK-bound to, so a single
// context keeps the expected set and the seeded rows consistent by construction.
_expectedAcks = DiscoverDriverNodes(db);
foreach (var node in _expectedAcks)
{
db.NodeDeploymentStates.Add(new NodeDeploymentState
@@ -142,9 +169,14 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers
if (_expectedAcks.Count == 0)
{
// No driver-role members. Seal immediately — the alternative is hanging forever
// waiting for ACKs that will never come.
_log.Warning("DispatchDeployment {Id}: no driver-role members in cluster; sealing empty",
// No enabled ClusterNode rows. Seal immediately — the alternative is hanging forever
// waiting for ACKs that will never come. Note this is now a *configuration* error
// rather than a topology observation: under the old membership rule an empty set meant
// "no driver nodes have joined yet", which could resolve itself; an empty set here means
// the fleet has no enabled nodes registered, which cannot.
_log.Warning(
"DispatchDeployment {Id}: no enabled ClusterNode rows registered; sealing empty. " +
"Nothing will receive this deployment — check the fleet's node configuration",
msg.DeploymentId);
SealDeployment();
}
@@ -233,8 +265,15 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers
using var db = _dbFactory.CreateDbContext();
UpdateDeploymentStatus(db, _current.Value, DeploymentStatus.TimedOut);
db.SaveChanges();
_log.Warning("Deployment {Id} timed out after {Deadline} ({Acked}/{Total} acks landed)",
_current.Value, _applyDeadline, _acks.Count, _expectedAcks.Count);
// Name the silent nodes, not just the counts. Since the expected set became DB-derived this
// is the failure an operator meets after deploying while a configured node is switched off,
// so it has to say which node — "4/5 acks landed" leaves them reading logs on five machines.
var missing = _expectedAcks.Where(n => !_acks.ContainsKey(n)).Select(n => n.Value).Order().ToList();
_log.Warning(
"Deployment {Id} timed out after {Deadline} ({Acked}/{Total} acks landed). " +
"No ack from: {MissingNodes}. Each is an enabled ClusterNode row — start the node, or " +
"set ClusterNode.Enabled = 0 if it is down for maintenance, then redeploy",
_current.Value, _applyDeadline, _acks.Count, _expectedAcks.Count, string.Join(", ", missing));
ResetForNext();
}
@@ -259,22 +298,29 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers
if (sealNow) d.SealedAtUtc = DateTime.UtcNow;
}
private HashSet<NodeId> DiscoverDriverNodes()
{
var cluster = Akka.Cluster.Cluster.Get(Context.System);
var nodes = new HashSet<NodeId>(NodeIdComparer);
foreach (var member in cluster.State.Members)
{
if (member.Status is not (MemberStatus.Up or MemberStatus.Joining)) continue;
if (!member.Roles.Contains("driver")) continue;
var host = member.Address.Host;
if (string.IsNullOrWhiteSpace(host)) continue;
// Match ClusterRoleInfo's NodeId derivation (host:port) so DriverHostActor's
// self-identification and the coordinator's expected-ack set agree.
nodes.Add(NodeId.Parse($"{host}:{member.Address.Port ?? 0}"));
}
return nodes;
}
/// <summary>
/// The set of nodes expected to ack this deployment: every <b>enabled</b> <c>ClusterNode</c>
/// row. See the class remarks for why this is DB-derived rather than membership-derived, and
/// for the two assumptions it rests on (every row is a driver node; <c>Enabled</c> is the
/// maintenance hatch).
/// </summary>
/// <remarks>
/// <c>ClusterNode.NodeId</c> is already in the <c>host:port</c> identifier space the
/// membership rule derived — the rig seeds <c>central-1:4053</c> and so on, and
/// <c>NodeDeploymentState.NodeId</c> is FK-bound to this column, so the fact that the
/// membership-derived rows have always satisfied that FK is standing proof the two sets
/// agree. It also has to stay that way: <c>DriverHostActor</c> self-identifies from
/// <c>ClusterRoleInfo</c>'s <c>host:port</c>, so a row whose NodeId is anything else is a
/// node whose acks will never match its expected-ack entry.
/// </remarks>
private static HashSet<NodeId> DiscoverDriverNodes(OtOpcUaConfigDbContext db) =>
db.ClusterNodes
.AsNoTracking()
.Where(n => n.Enabled)
.Select(n => n.NodeId)
.ToList()
.Select(NodeId.Parse)
.ToHashSet(NodeIdComparer);
/// <summary>Case-insensitive <see cref="NodeId"/> equality (by <see cref="NodeId.Value"/>),
/// matching the case-insensitive scoping in <c>DeploymentArtifact.ResolveClusterScope</c> so the
tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/ConfigPublishCoordinatorDbSourcedAcksTests.cs
+219
View File
@@ -0,0 +1,219 @@
using Akka.Actor;
using Microsoft.EntityFrameworkCore;
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>
/// Per-cluster mesh Phase 1: the coordinator's expected-ack set comes from enabled
/// <c>ClusterNode</c> rows, not from <c>Akka.Cluster.State.Members</c> filtered by role.
/// </summary>
/// <remarks>
/// <para>
/// <b>Why every test here seeds ClusterNode rows and asserts a non-empty expected set.</b>
/// The harness's ActorSystem self-joins with role <c>admin</c> and no <c>driver</c> members,
/// so the OLD derivation returned an empty set for every one of these scenarios and sealed
/// immediately. Any test that merely observes "the deployment did not seal" would therefore
/// have passed against the old code too. Each test below distinguishes the two by requiring
/// the coordinator to WAIT for a specific node and then seal when exactly that node acks —
/// behaviour the membership rule cannot produce in this harness at all.
/// </para>
/// <para>
/// Verified by reverting <c>DiscoverDriverNodes</c> to the membership scan: the first three
/// go red. <see cref="No_enabled_rows_seals_empty"/> stays green under both derivations and
/// is <b>not</b> a test of the derivation — the harness has no driver members AND no enabled
/// rows, so both rules produce an empty set. It pins the seal-empty branch and its reworded
/// reason, nothing more.
/// </para>
/// </remarks>
public sealed class ConfigPublishCoordinatorDbSourcedAcksTests : ControlPlaneActorTestBase
{
private static readonly RevisionHash TestRevision = RevisionHash.Parse(new string('b', 64));
private static readonly TimeSpan NoTimeout = TimeSpan.FromMinutes(5);
/// <summary>
/// An enabled row is expected to ack; the deployment waits for it and seals when it arrives.
/// This is the DB-derived set doing something the membership set demonstrably cannot here.
/// </summary>
[Fact]
public void Enabled_ClusterNode_rows_are_the_expected_ack_set()
{
var dbFactory = NewInMemoryDbFactory();
var deploymentId = SeedDispatchingDeployment(dbFactory);
SeedNodes(dbFactory, ("site-a-1:4053", true));
var actor = Sys.ActorOf(ConfigPublishCoordinator.Props(dbFactory, NoTimeout));
actor.Tell(new DispatchDeployment(deploymentId, TestRevision, CorrelationId.NewId()));
// The seeded row must have produced a NodeDeploymentState row — proof the coordinator
// expected it, independent of whether it later sealed.
AwaitAssert(() =>
{
using var db = dbFactory.CreateDbContext();
db.NodeDeploymentStates.Select(s => s.NodeId).ShouldBe(new[] { "site-a-1:4053" });
db.Deployments.Single().Status.ShouldBe(DeploymentStatus.AwaitingApplyAcks);
}, duration: TimeSpan.FromSeconds(3));
actor.Tell(new ApplyAck(deploymentId, NodeId.Parse("site-a-1:4053"),
ApplyAckOutcome.Applied, null, CorrelationId.NewId()));
AwaitAssert(() =>
{
using var db = dbFactory.CreateDbContext();
db.Deployments.Single().Status.ShouldBe(DeploymentStatus.Sealed);
}, duration: TimeSpan.FromSeconds(3));
}
/// <summary>
/// The <c>Enabled = 0</c> escape hatch: a disabled row is NOT expected, so a deployment seals
/// without it. This is what makes the "a stopped node now fails the deploy" behaviour change
/// acceptable to ship — without it, a node down for maintenance blocks every deployment.
/// </summary>
[Fact]
public void Disabled_ClusterNode_rows_are_not_expected_to_ack()
{
var dbFactory = NewInMemoryDbFactory();
var deploymentId = SeedDispatchingDeployment(dbFactory);
SeedNodes(dbFactory, ("site-a-1:4053", true), ("site-a-2:4053", false));
var actor = Sys.ActorOf(ConfigPublishCoordinator.Props(dbFactory, NoTimeout));
actor.Tell(new DispatchDeployment(deploymentId, TestRevision, CorrelationId.NewId()));
// Only the enabled node is expected — the disabled one gets no NodeDeploymentState row.
AwaitAssert(() =>
{
using var db = dbFactory.CreateDbContext();
db.NodeDeploymentStates.Select(s => s.NodeId).ShouldBe(new[] { "site-a-1:4053" });
}, duration: TimeSpan.FromSeconds(3));
// ...and the enabled node's ack alone seals it. If the disabled row were expected this would
// sit at AwaitingApplyAcks until the deadline.
actor.Tell(new ApplyAck(deploymentId, NodeId.Parse("site-a-1:4053"),
ApplyAckOutcome.Applied, null, CorrelationId.NewId()));
AwaitAssert(() =>
{
using var db = dbFactory.CreateDbContext();
db.Deployments.Single().Status.ShouldBe(DeploymentStatus.Sealed);
}, duration: TimeSpan.FromSeconds(3));
}
/// <summary>
/// An ack from a node with no <c>ClusterNode</c> row is discarded — it is not in the expected
/// set, so it must not count toward the seal. Under the membership rule a running-but-
/// unregistered node WAS expected, and then threw FK 547 when its state row was written.
/// </summary>
[Fact]
public void Ack_from_a_node_absent_from_ClusterNode_is_discarded()
{
var dbFactory = NewInMemoryDbFactory();
var deploymentId = SeedDispatchingDeployment(dbFactory);
SeedNodes(dbFactory, ("site-a-1:4053", true));
var actor = Sys.ActorOf(ConfigPublishCoordinator.Props(dbFactory, NoTimeout));
actor.Tell(new DispatchDeployment(deploymentId, TestRevision, CorrelationId.NewId()));
AwaitAssert(() =>
{
using var db = dbFactory.CreateDbContext();
db.Deployments.Single().Status.ShouldBe(DeploymentStatus.AwaitingApplyAcks);
}, duration: TimeSpan.FromSeconds(3));
// A node nobody registered acks. One expected node, one ack — if it were counted, the
// coordinator would seal here.
actor.Tell(new ApplyAck(deploymentId, NodeId.Parse("ghost-9:4053"),
ApplyAckOutcome.Applied, null, CorrelationId.NewId()));
ExpectNoMsg(TimeSpan.FromMilliseconds(400));
using (var db = dbFactory.CreateDbContext())
{
db.Deployments.Single().Status.ShouldBe(DeploymentStatus.AwaitingApplyAcks);
}
// Positive control on the wait itself: the real node's ack DOES seal it, so the assertion
// above measured the discard rather than a coordinator that had simply stopped responding.
actor.Tell(new ApplyAck(deploymentId, NodeId.Parse("site-a-1:4053"),
ApplyAckOutcome.Applied, null, CorrelationId.NewId()));
AwaitAssert(() =>
{
using var db = dbFactory.CreateDbContext();
db.Deployments.Single().Status.ShouldBe(DeploymentStatus.Sealed);
}, duration: TimeSpan.FromSeconds(3));
}
/// <summary>
/// No enabled rows at all seals empty. Same outcome as the old empty-membership branch, but
/// it now means "the fleet has no enabled nodes registered" — a configuration error that
/// cannot resolve itself — rather than "no driver node has joined yet".
/// </summary>
[Fact]
public void No_enabled_rows_seals_empty()
{
var dbFactory = NewInMemoryDbFactory();
var deploymentId = SeedDispatchingDeployment(dbFactory);
SeedNodes(dbFactory, ("site-a-1:4053", false), ("site-a-2:4053", false));
var actor = Sys.ActorOf(ConfigPublishCoordinator.Props(dbFactory, NoTimeout));
actor.Tell(new DispatchDeployment(deploymentId, TestRevision, CorrelationId.NewId()));
AwaitAssert(() =>
{
using var db = dbFactory.CreateDbContext();
db.Deployments.Single().Status.ShouldBe(DeploymentStatus.Sealed);
db.NodeDeploymentStates.ShouldBeEmpty();
}, duration: TimeSpan.FromSeconds(3));
}
private static void SeedNodes(
IDbContextFactory<OtOpcUaConfigDbContext> dbFactory,
params (string NodeId, bool Enabled)[] 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, enabled) in nodes)
{
db.ClusterNodes.Add(new ClusterNode
{
NodeId = nodeId,
ClusterId = "SITE-A",
Host = nodeId.Split(':')[0],
ApplicationUri = $"urn:OtOpcUa:{nodeId}",
Enabled = enabled,
CreatedBy = "test",
});
}
db.SaveChanges();
}
private static DeploymentId SeedDispatchingDeployment(
IDbContextFactory<OtOpcUaConfigDbContext> dbFactory)
{
var id = DeploymentId.NewId();
using var db = dbFactory.CreateDbContext();
db.Deployments.Add(new Deployment
{
DeploymentId = id.Value,
RevisionHash = TestRevision.Value,
Status = DeploymentStatus.Dispatching,
CreatedBy = "test",
});
db.SaveChanges();
return id;
}
}