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);
}