using Akka.Actor; using Akka.Cluster; using Akka.TestKit.Xunit; using Shouldly; using Xunit; using ZB.MOM.WW.Health.Akka; using ZB.MOM.WW.OtOpcUa.Cluster; using ZB.MOM.WW.OtOpcUa.Commons.OpcUa; using ZB.MOM.WW.OtOpcUa.Runtime.Redundancy; using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness; namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Redundancy; /// /// lmxopcua#762 on a real two-node driver mesh: a Primary whose OPC UA endpoint failed to start hands the role to /// its peer by leaving the mesh — but only once the peer reports a listening endpoint of its own. /// /// /// Node A is a second that forms the cluster first, so it is the OLDEST Up driver /// member — the node the election () /// names Primary. This class's own is node B, which joins second; it survives /// every test, so the base class disposes it normally. /// public sealed class OpcUaEndpointStepDownActorTests : TestKit { private const string Hocon = @" akka { loglevel = ""WARNING"" test.single-expect-default = 15s extensions = [ ""Akka.Cluster.Tools.PublishSubscribe.DistributedPubSubExtensionProvider, Akka.Cluster.Tools"" ] actor.provider = ""Akka.Cluster.ClusterActorRefProvider, Akka.Cluster"" remote.dot-netty.tcp { hostname = ""127.0.0.1"" port = 0 } cluster { seed-nodes = [] roles = [""driver""] min-nr-of-members = 1 run-coordinated-shutdown-when-down = off pub-sub.gossip-interval = 200ms } }"; private static readonly TimeSpan Tick = TimeSpan.FromMilliseconds(100); /// How long a "does NOT leave" assertion watches — many ticks, and well past pub-sub convergence. private static readonly TimeSpan AbsenceWindow = TimeSpan.FromSeconds(3); private readonly TestKit _nodeA; public OpcUaEndpointStepDownActorTests() : base(Hocon + TestTimeFactor.Hocon) { _nodeA = new TestKit(Hocon + TestTimeFactor.Hocon); var clusterA = Akka.Cluster.Cluster.Get(_nodeA.Sys); clusterA.Join(clusterA.SelfAddress); AwaitCondition(() => UpCount(clusterA) == 1, RuntimeActorTestBase.PresenceBudget, RuntimeActorTestBase.PollInterval); var clusterB = Akka.Cluster.Cluster.Get(Sys); clusterB.Join(clusterA.SelfAddress); AwaitCondition(() => UpCount(clusterA) == 2 && UpCount(clusterB) == 2, RuntimeActorTestBase.PresenceBudget, RuntimeActorTestBase.PollInterval); // Fixture sanity: A must be the Primary going in, or every assertion below is about the wrong node. ClusterActiveNode.SelfIsActive(clusterA, RoleParser.Driver).ShouldBeTrue("node A formed the cluster, so it is the oldest Up driver member"); ClusterActiveNode.SelfIsActive(clusterB, RoleParser.Driver).ShouldBeFalse(); } /// protected override void Dispose(bool disposing) { if (disposing) _nodeA.Dispose(); base.Dispose(disposing); } /// /// The issue's test: A's endpoint failed to start, so A steps down and B becomes Primary. Also pins the /// ping-pong guard first — while B's own endpoint is not listening, A holds the role (handing it over would /// only move the outage and restart A for nothing); the moment B reports listening, A leaves. /// [Fact] public async Task Failed_primary_hands_the_role_to_a_peer_once_the_peer_reports_listening() { var stateA = Failed(); var stateB = new OpcUaEndpointState(); // NotStarted: not listening yet _nodeA.Sys.ActorOf(OpcUaEndpointStepDownActor.Props(stateA, Tick), "step-down"); Sys.ActorOf(OpcUaEndpointStepDownActor.Props(stateB, Tick), "step-down"); // Guard: no listening peer -> A stays Up and Primary. await Task.Delay(AbsenceWindow, TestContext.Current.CancellationToken); _nodeA.Sys.WhenTerminated.IsCompleted.ShouldBeFalse("A must not leave while no peer can serve either"); ClusterActiveNode.SelfIsActive(Akka.Cluster.Cluster.Get(_nodeA.Sys), RoleParser.Driver).ShouldBeTrue(); // B's endpoint comes up -> A leaves (its ActorSystem terminates, which in production exits the process)... stateB.MarkStarted(); await _nodeA.Sys.WhenTerminated.WaitAsync(RuntimeActorTestBase.PresenceBudget, TestContext.Current.CancellationToken); // ... and B is now the oldest Up driver member, i.e. what RedundancyStateActor elects Primary. var clusterB = Akka.Cluster.Cluster.Get(Sys); AwaitCondition(() => ClusterActiveNode.SelfIsActive(clusterB, RoleParser.Driver) && clusterB.State.Members.Count(m => m.Status == MemberStatus.Up) == 1, RuntimeActorTestBase.PresenceBudget, RuntimeActorTestBase.PollInterval, cancellationToken: TestContext.Current.CancellationToken); } /// A failed endpoint on the SECONDARY changes nothing: there is no role to hand over. [Fact] public async Task Failed_secondary_does_not_leave() { var stateA = new OpcUaEndpointState(); stateA.MarkStarted(); _nodeA.Sys.ActorOf(OpcUaEndpointStepDownActor.Props(stateA, Tick), "step-down"); Sys.ActorOf(OpcUaEndpointStepDownActor.Props(Failed(), Tick), "step-down"); await Task.Delay(AbsenceWindow, TestContext.Current.CancellationToken); Sys.WhenTerminated.IsCompleted.ShouldBeFalse(); _nodeA.Sys.WhenTerminated.IsCompleted.ShouldBeFalse(); UpCount(Akka.Cluster.Cluster.Get(Sys)).ShouldBe(2); ClusterActiveNode.SelfIsActive(Akka.Cluster.Cluster.Get(_nodeA.Sys), RoleParser.Driver).ShouldBeTrue(); } /// A healthy Primary never leaves, even with a listening peer (positive control for the trigger). [Fact] public async Task Healthy_primary_does_not_leave() { var stateA = new OpcUaEndpointState(); stateA.MarkStarted(); var stateB = new OpcUaEndpointState(); stateB.MarkStarted(); _nodeA.Sys.ActorOf(OpcUaEndpointStepDownActor.Props(stateA, Tick), "step-down"); Sys.ActorOf(OpcUaEndpointStepDownActor.Props(stateB, Tick), "step-down"); await Task.Delay(AbsenceWindow, TestContext.Current.CancellationToken); _nodeA.Sys.WhenTerminated.IsCompleted.ShouldBeFalse(); UpCount(Akka.Cluster.Cluster.Get(Sys)).ShouldBe(2); } private static OpcUaEndpointState Failed() { var state = new OpcUaEndpointState(); state.MarkFailed("port 4840 in use"); return state; } private static int UpCount(Akka.Cluster.Cluster cluster) => cluster.State.Members.Count(m => m.Status == MemberStatus.Up); }