feat(fleet): reconcile ClusterNode dial targets against live membership

ClusterNode.AkkaPort and the node's own Cluster:Port are the same fact in two places and nothing made them agree. Phase 2 dials the row instead of gossiping, so a node binding 4054 while its row says 4053 becomes unreachable from central — and the symptom is a silent absence of acks rather than an error. That is the shape of the Modbus/ModbusTcp and TwinCat/Focas drifts already in this repo. Since Phase 1 also made the rows the deploy path's expected-ack set, drift already costs a failed deployment today. Implemented as the plan's preferred option: an admin-role singleton comparing rows against the membership an admin node can already see, rather than each driver node asserting its own row — Phase 4 removes the driver nodes' ConfigDb connection, so a self-assertion written there would have to be deleted again. Three shapes, split by severity: a row whose dial target disagrees with its own NodeId and a running node with no row are Errors; an enabled row with no matching member is a Warning, because a node down for maintenance is a legitimate state. Findings are logged only when the set changes — a check that reprints the same warning every sweep trains operators to filter it out. Documented limitation: Phase 2 must revisit this. Once the fleet splits into one mesh per cluster an admin node cannot see site members, and every site row would report EnabledRowNotInCluster forever. Sabotage: removing the change-detection guard turns the repeat test red. ControlPlane.Tests 100/100. Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW
2026-07-22 08:36:44 -04:00
parent d88e245503
commit 90c1302f71
6 changed files with 511 additions and 3 deletions
@@ -0,0 +1,133 @@
+using Akka.Actor;
+using Akka.Cluster;
+using Akka.Event;
+using Microsoft.EntityFrameworkCore;
+using ZB.MOM.WW.OtOpcUa.Configuration;
+
+namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Fleet;
+
+/// <summary>
+///     Admin-role cluster singleton that runs <see cref="ClusterNodeAddressReconciler"/> against live
+///     membership and logs what it finds. A singleton rather than a per-admin-node actor so a
+///     two-admin fleet reports each disagreement once.
+/// </summary>
+/// <remarks>
+///     Recomputes on membership change (debounced) and on a slow periodic sweep, and <b>logs only
+///     when the finding set changes</b>. A node legitimately down would otherwise reprint the same
+///     warning every sweep until someone silences it by ignoring the log, which is the failure mode
+///     this check exists to avoid.
+/// </remarks>
+public sealed class ClusterNodeAddressReconcilerActor : ReceiveActor, IWithTimers
+{
+    /// <summary>Driver role name — only members carrying it are expected to have a ClusterNode row.</summary>
+    public const string DriverRole = "driver";
+
+    /// <summary>Collapses a burst of membership events (a rolling restart) into one reconcile.</summary>
+    public static readonly TimeSpan DebounceWindow = TimeSpan.FromSeconds(2);
+
+    /// <summary>Periodic sweep, so a row edited in the AdminUI is checked without a topology change.</summary>
+    public static readonly TimeSpan DefaultSweepInterval = TimeSpan.FromMinutes(5);
+
+    private readonly IDbContextFactory<OtOpcUaConfigDbContext> _dbFactory;
+    private readonly Akka.Cluster.Cluster _cluster;
+    private readonly TimeSpan _sweepInterval;
+    private readonly ILoggingAdapter _log = Context.GetLogger();
+    private IReadOnlyList<ClusterNodeAddressMismatch>? _lastReported;
+
+    /// <summary>Gets the timer scheduler for this actor.</summary>
+    public ITimerScheduler Timers { get; set; } = null!;
+
+    /// <summary>Creates Props for the reconciler.</summary>
+    /// <param name="dbFactory">Factory for the config database context.</param>
+    /// <param name="sweepInterval">Periodic sweep interval; defaults to <see cref="DefaultSweepInterval"/>.</param>
+    /// <returns>Props for actor creation.</returns>
+    public static Props Props(
+        IDbContextFactory<OtOpcUaConfigDbContext> dbFactory, TimeSpan? sweepInterval = null) =>
+        Akka.Actor.Props.Create(() => new ClusterNodeAddressReconcilerActor(dbFactory, sweepInterval));
+
+    /// <summary>Initializes a new instance of the <see cref="ClusterNodeAddressReconcilerActor"/> class.</summary>
+    /// <param name="dbFactory">Factory for the config database context.</param>
+    /// <param name="sweepInterval">Periodic sweep interval; defaults to <see cref="DefaultSweepInterval"/>.</param>
+    public ClusterNodeAddressReconcilerActor(
+        IDbContextFactory<OtOpcUaConfigDbContext> dbFactory, TimeSpan? sweepInterval = null)
+    {
+        _dbFactory = dbFactory;
+        _cluster = Akka.Cluster.Cluster.Get(Context.System);
+        _sweepInterval = sweepInterval ?? DefaultSweepInterval;
+
+        Receive<ClusterEvent.IMemberEvent>(_ => ScheduleReconcile());
+        Receive<ClusterEvent.CurrentClusterState>(_ => ScheduleReconcile());
+        Receive<ReconcileNow>(_ => Reconcile());
+    }
+
+    /// <inheritdoc />
+    protected override void PreStart()
+    {
+        _cluster.Subscribe(Self, ClusterEvent.InitialStateAsEvents, typeof(ClusterEvent.IMemberEvent));
+        Timers.StartPeriodicTimer("sweep", ReconcileNow.Instance, _sweepInterval, _sweepInterval);
+    }
+
+    /// <inheritdoc />
+    protected override void PostStop() => _cluster.Unsubscribe(Self);
+
+    private void ScheduleReconcile() =>
+        Timers.StartSingleTimer("debounce", ReconcileNow.Instance, DebounceWindow);
+
+    private void Reconcile()
+    {
+        var observed = _cluster.State.Members
+            .Where(m => m.Status == MemberStatus.Up && m.Roles.Contains(DriverRole))
+            .Select(m => (Host: m.Address.Host ?? string.Empty, Port: m.Address.Port ?? 0))
+            .Where(a => !string.IsNullOrWhiteSpace(a.Host) && a.Port > 0)
+            .ToList();
+
+        List<ClusterNodeAddress> rows;
+        List<string> enabled;
+        try
+        {
+            using var db = _dbFactory.CreateDbContext();
+            rows = db.ClusterNodes.AsNoTracking()
+                .Select(n => new ClusterNodeAddress(n.NodeId, n.Host, n.AkkaPort))
+                .ToList();
+            enabled = db.ClusterNodes.AsNoTracking()
+                .Where(n => n.Enabled).Select(n => n.NodeId).ToList();
+        }
+        catch (Exception ex)
+        {
+            // A consistency check must never take the admin node down with it. Central SQL being
+            // briefly unreachable is an operational fact, not a reason to restart this singleton.
+            _log.Warning(ex, "ClusterNode address reconcile skipped — config database unreachable");
+            return;
+        }
+
+        var findings = ClusterNodeAddressReconciler.Reconcile(observed, rows, enabled);
+
+        if (_lastReported is not null && findings.SequenceEqual(_lastReported)) return;
+        _lastReported = findings;
+
+        if (findings.Count == 0)
+        {
+            _log.Info("ClusterNode addresses reconcile with cluster membership ({Count} driver members)",
+                observed.Count);
+            return;
+        }
+
+        foreach (var f in findings)
+        {
+            // Error for the two shapes that are always a misconfiguration; Warning for a row whose
+            // node is merely absent, which is a legitimate state during maintenance.
+            if (f.Kind == AddressMismatchKind.EnabledRowNotInCluster)
+                _log.Warning("ClusterNode address check [{Kind}] {NodeId}: {Detail}", f.Kind, f.NodeId, f.Detail);
+            else
+                _log.Error("ClusterNode address check [{Kind}] {NodeId}: {Detail}", f.Kind, f.NodeId, f.Detail);
+        }
+    }
+
+    /// <summary>Self-message: run a reconcile pass now.</summary>
+    public sealed class ReconcileNow
+    {
+        /// <summary>The singleton instance.</summary>
+        public static readonly ReconcileNow Instance = new();
+        private ReconcileNow() { }
+    }
+}