From 91209070d0f8b004d3773dd109f976d15282aa97 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Mon, 13 Jul 2026 09:43:22 -0400 Subject: [PATCH] fix(comm): shared oldest-Up ActiveNodeEvaluator in Communication; Host evaluator delegates (plan R2-02 T1) --- .../ClusterState/ActiveNodeEvaluator.cs | 39 +++++++++++++ .../Health/ClusterActivityEvaluator.cs | 20 ++----- .../ActiveNodeEvaluatorTests.cs | 57 +++++++++++++++++++ 3 files changed, 102 insertions(+), 14 deletions(-) create mode 100644 src/ZB.MOM.WW.ScadaBridge.Communication/ClusterState/ActiveNodeEvaluator.cs create mode 100644 tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/ActiveNodeEvaluatorTests.cs diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/ClusterState/ActiveNodeEvaluator.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/ClusterState/ActiveNodeEvaluator.cs new file mode 100644 index 00000000..57d7b425 --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/ClusterState/ActiveNodeEvaluator.cs @@ -0,0 +1,39 @@ +using Akka.Cluster; + +namespace ZB.MOM.WW.ScadaBridge.Communication.ClusterState; + +/// +/// THE single definition of "active node" (review 01 [High]; review 02 round 2 [Critical] N1): +/// a node is active when it is the OLDEST Up member (optionally within a role scope) — i.e. +/// the member the ClusterSingletonManager places singletons on. Cluster LEADERSHIP (lowest +/// address) is an Akka-internal concept that diverges from singleton placement permanently +/// once the original first node restarts and rejoins; every product-level active/standby +/// decision must use this evaluator, never cluster.State.Leader. +/// +/// Lives in Communication (not Host) so BOTH SiteCommunicationActor and +/// SiteReplicationActor can default to it — Host cannot be referenced from either. +/// The Host's ClusterActivityEvaluator.SelfIsOldest delegates here, so the S&F +/// delivery gate (IClusterNodeProvider.SelfIsPrimary), the resync authority checks, +/// and the heartbeat IsActive stamp all share one implementation. +/// +/// +public static class ActiveNodeEvaluator +{ + /// True when self is Up and no other Up member (in the role scope) is older. + /// The Akka cluster to evaluate. + /// Optional role scope; when set, only members with this role are considered. + /// true when self is Up and the oldest Up member in the role scope. + public static bool SelfIsOldestUp(Cluster cluster, string? role = null) + { + var self = cluster.SelfMember; + if (self.Status != MemberStatus.Up) + return false; + if (role != null && !self.HasRole(role)) + return false; + + return cluster.State.Members + .Where(m => m.Status == MemberStatus.Up) + .Where(m => role == null || m.HasRole(role)) + .All(m => m.UniqueAddress.Equals(self.UniqueAddress) || self.IsOlderThan(m)); + } +} diff --git a/src/ZB.MOM.WW.ScadaBridge.Host/Health/ClusterActivityEvaluator.cs b/src/ZB.MOM.WW.ScadaBridge.Host/Health/ClusterActivityEvaluator.cs index 382ade94..b4729cb5 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Host/Health/ClusterActivityEvaluator.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Host/Health/ClusterActivityEvaluator.cs @@ -13,23 +13,15 @@ namespace ZB.MOM.WW.ScadaBridge.Host.Health; /// public static class ClusterActivityEvaluator { - /// True when self is Up and no other Up member (in the role scope) is older. + /// True when self is Up and no other Up member (in the role scope) is older. + /// Delegates to the shared — + /// one implementation for the delivery gate, the resync authority checks, and the + /// heartbeat IsActive stamp (review 02 round 2, N1). /// The Akka cluster to evaluate. /// Optional role scope; when set, only members with this role are considered. /// true when self is Up and the oldest Up member in the role scope. - public static bool SelfIsOldest(Cluster cluster, string? role = null) - { - var self = cluster.SelfMember; - if (self.Status != MemberStatus.Up) - return false; - if (role != null && !self.HasRole(role)) - return false; - - return cluster.State.Members - .Where(m => m.Status == MemberStatus.Up) - .Where(m => role == null || m.HasRole(role)) - .All(m => m.UniqueAddress.Equals(self.UniqueAddress) || self.IsOlderThan(m)); - } + public static bool SelfIsOldest(Cluster cluster, string? role = null) => + Communication.ClusterState.ActiveNodeEvaluator.SelfIsOldestUp(cluster, role); /// The oldest Up member in the role scope, or null while none is Up. Used for Primary/Standby labelling. /// The Akka cluster to evaluate. diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/ActiveNodeEvaluatorTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/ActiveNodeEvaluatorTests.cs new file mode 100644 index 00000000..07589409 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/ActiveNodeEvaluatorTests.cs @@ -0,0 +1,57 @@ +using System.Net; +using System.Net.Sockets; +using Akka.Actor; +using Akka.Configuration; +using Xunit; +using ZB.MOM.WW.ScadaBridge.Communication.ClusterState; + +namespace ZB.MOM.WW.ScadaBridge.Communication.Tests; + +public class ActiveNodeEvaluatorTests : IAsyncLifetime +{ + private ActorSystem? _system; + + public async Task InitializeAsync() + { + var port = FreePort(); + var config = ConfigurationFactory.ParseString($@" +akka {{ + actor.provider = cluster + remote.dot-netty.tcp {{ hostname = ""127.0.0.1"", port = {port} }} + cluster {{ + seed-nodes = [""akka.tcp://ane-test@127.0.0.1:{port}""] + roles = [""site-x""] + min-nr-of-members = 1 + }} +}}"); + _system = ActorSystem.Create("ane-test", config); + var cluster = Akka.Cluster.Cluster.Get(_system); + var deadline = DateTime.UtcNow.AddSeconds(20); + while (cluster.SelfMember.Status != Akka.Cluster.MemberStatus.Up && DateTime.UtcNow < deadline) + await Task.Delay(100); + } + + public async Task DisposeAsync() { if (_system != null) await _system.Terminate(); } + + [Fact] + public void SelfIsOldestUp_SoleUpMember_ReturnsTrue() + { + var cluster = Akka.Cluster.Cluster.Get(_system!); + Assert.True(ActiveNodeEvaluator.SelfIsOldestUp(cluster)); + Assert.True(ActiveNodeEvaluator.SelfIsOldestUp(cluster, "site-x")); + } + + [Fact] + public void SelfIsOldestUp_RoleNotHeld_ReturnsFalse() + { + var cluster = Akka.Cluster.Cluster.Get(_system!); + Assert.False(ActiveNodeEvaluator.SelfIsOldestUp(cluster, "role-nonexistent")); + } + + private static int FreePort() + { + using var l = new TcpListener(IPAddress.Loopback, 0); + l.Start(); + return ((IPEndPoint)l.LocalEndpoint).Port; + } +}