fix(comm): shared oldest-Up ActiveNodeEvaluator in Communication; Host evaluator delegates (plan R2-02 T1)
This commit is contained in:
@@ -0,0 +1,39 @@
|
|||||||
|
using Akka.Cluster;
|
||||||
|
|
||||||
|
namespace ZB.MOM.WW.ScadaBridge.Communication.ClusterState;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// 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 <c>cluster.State.Leader</c>.
|
||||||
|
/// <para>
|
||||||
|
/// Lives in Communication (not Host) so BOTH <c>SiteCommunicationActor</c> and
|
||||||
|
/// <c>SiteReplicationActor</c> can default to it — Host cannot be referenced from either.
|
||||||
|
/// The Host's <c>ClusterActivityEvaluator.SelfIsOldest</c> delegates here, so the S&F
|
||||||
|
/// delivery gate (<c>IClusterNodeProvider.SelfIsPrimary</c>), the resync authority checks,
|
||||||
|
/// and the heartbeat IsActive stamp all share one implementation.
|
||||||
|
/// </para>
|
||||||
|
/// </summary>
|
||||||
|
public static class ActiveNodeEvaluator
|
||||||
|
{
|
||||||
|
/// <summary>True when self is Up and no other Up member (in the role scope) is older.</summary>
|
||||||
|
/// <param name="cluster">The Akka cluster to evaluate.</param>
|
||||||
|
/// <param name="role">Optional role scope; when set, only members with this role are considered.</param>
|
||||||
|
/// <returns><c>true</c> when self is Up and the oldest Up member in the role scope.</returns>
|
||||||
|
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));
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -13,23 +13,15 @@ namespace ZB.MOM.WW.ScadaBridge.Host.Health;
|
|||||||
/// </summary>
|
/// </summary>
|
||||||
public static class ClusterActivityEvaluator
|
public static class ClusterActivityEvaluator
|
||||||
{
|
{
|
||||||
/// <summary>True when self is Up and no other Up member (in the role scope) is older.</summary>
|
/// <summary>True when self is Up and no other Up member (in the role scope) is older.
|
||||||
|
/// Delegates to the shared <see cref="Communication.ClusterState.ActiveNodeEvaluator"/> —
|
||||||
|
/// one implementation for the delivery gate, the resync authority checks, and the
|
||||||
|
/// heartbeat IsActive stamp (review 02 round 2, N1).</summary>
|
||||||
/// <param name="cluster">The Akka cluster to evaluate.</param>
|
/// <param name="cluster">The Akka cluster to evaluate.</param>
|
||||||
/// <param name="role">Optional role scope; when set, only members with this role are considered.</param>
|
/// <param name="role">Optional role scope; when set, only members with this role are considered.</param>
|
||||||
/// <returns><c>true</c> when self is Up and the oldest Up member in the role scope.</returns>
|
/// <returns><c>true</c> when self is Up and the oldest Up member in the role scope.</returns>
|
||||||
public static bool SelfIsOldest(Cluster cluster, string? role = null)
|
public static bool SelfIsOldest(Cluster cluster, string? role = null) =>
|
||||||
{
|
Communication.ClusterState.ActiveNodeEvaluator.SelfIsOldestUp(cluster, role);
|
||||||
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));
|
|
||||||
}
|
|
||||||
|
|
||||||
/// <summary>The oldest Up member in the role scope, or null while none is Up. Used for Primary/Standby labelling.</summary>
|
/// <summary>The oldest Up member in the role scope, or null while none is Up. Used for Primary/Standby labelling.</summary>
|
||||||
/// <param name="cluster">The Akka cluster to evaluate.</param>
|
/// <param name="cluster">The Akka cluster to evaluate.</param>
|
||||||
|
|||||||
@@ -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;
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user