using Akka.Actor; using Akka.Cluster; using Akka.Configuration; using Shouldly; using Xunit; using ZB.MOM.WW.OtOpcUa.Commons.Messages.Redundancy; using ZB.MOM.WW.OtOpcUa.ControlPlane.Redundancy; namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Tests; /// /// Pins which node elects Primary, using a real two-node /// cluster deliberately built so that the oldest member is not the lowest-addressed one. /// /// /// /// The distinction is the whole point. Akka offers two different "first" members and they /// disagree after any restart: /// /// /// /// /// ClusterState.RoleLeader(role) — the lowest-addressed Up member with /// that role. Address order is host, then port; it has nothing to do with time. /// /// /// /// /// Oldest — the Up member with the lowest up-number, i.e. the one that has been /// in the cluster longest. This is what ClusterSingletonManager uses to place /// singletons, and (under keep-oldest) what the split-brain resolver keys on. /// /// /// /// /// They coincide on a freshly-formed cluster, which is why single-node and /// happy-path tests never caught the difference. They diverge the moment a node restarts: the /// restarted node becomes the youngest while keeping its address, so if it happens to /// hold the lower address it becomes role leader — and the actor would have named it Primary /// while the cluster singletons, and every piece of work they own, stayed on the other node. /// The Primary-gated data plane (writes, alarm acks, alerts emit, alarm-history drain) would /// then be enabled on the node that is not hosting the work. /// /// /// This test reproduces that divergence directly rather than waiting for a restart: node A /// binds the higher port and joins first (so it is oldest), node B binds the /// lower port and joins second (so it is role leader). The correct answer is A. /// /// public sealed class RedundancyPrimaryElectionTests : IAsyncLifetime { // Fixed, unusual ports: the test needs a deterministic address ordering, which port 0 cannot give. private const int OldestPort = 19_531; // joins FIRST -> oldest, but the HIGHER address private const int YoungestPort = 19_530; // joins SECOND -> role leader, the LOWER address private ActorSystem? _oldest; private ActorSystem? _youngest; private static Config NodeConfig(int port) => ConfigurationFactory.ParseString($$""" akka { loglevel = "WARNING" actor.provider = "Akka.Cluster.ClusterActorRefProvider, Akka.Cluster" remote.dot-netty.tcp { hostname = "127.0.0.1" port = {{port}} } cluster { seed-nodes = ["akka.tcp://redundancy-election@127.0.0.1:{{OldestPort}}"] roles = ["admin", "driver"] min-nr-of-members = 1 run-coordinated-shutdown-when-down = off downing-provider-class = "" } } """); public async Task InitializeAsync() { // Order matters: the seed forms the cluster and is therefore the oldest member. _oldest = ActorSystem.Create("redundancy-election", NodeConfig(OldestPort)); await WaitForUpAsync(_oldest, expectedMembers: 1); _youngest = ActorSystem.Create("redundancy-election", NodeConfig(YoungestPort)); await WaitForUpAsync(_oldest, expectedMembers: 2); await WaitForUpAsync(_youngest, expectedMembers: 2); } public async Task DisposeAsync() { if (_youngest is not null) await _youngest.Terminate(); if (_oldest is not null) await _oldest.Terminate(); } private static async Task WaitForUpAsync(ActorSystem system, int expectedMembers) { var cluster = Akka.Cluster.Cluster.Get(system); var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(30); while (DateTime.UtcNow < deadline) { var up = cluster.State.Members.Count(m => m.Status == MemberStatus.Up); if (up >= expectedMembers) return; await Task.Delay(100); } throw new TimeoutException( $"cluster on port {cluster.SelfAddress.Port} never saw {expectedMembers} Up members " + $"(saw {cluster.State.Members.Count(m => m.Status == MemberStatus.Up)})"); } /// /// Sanity check on the fixture itself: if the two orderings did not actually diverge, the real /// assertion below would pass for the wrong reason. This asserts the setup produced the /// divergence it claims to — role leader is the younger node, not the oldest one. /// [Fact] public void Fixture_actually_produces_divergent_age_and_address_ordering() { var state = Akka.Cluster.Cluster.Get(_oldest!).State; state.RoleLeader("driver")!.Port.ShouldBe( YoungestPort, "the fixture is only meaningful if the lowest-addressed node is the YOUNGER one"); } /// /// The elected Primary is the oldest driver member — the node that hosts the cluster singletons — /// not the lowest-addressed one. /// [Fact] public async Task Primary_is_the_oldest_driver_member_not_the_role_leader() { var snapshot = await CaptureSnapshotAsync(); var primaries = snapshot.Where(n => n.Role == RedundancyRole.Primary).ToList(); primaries.Count.ShouldBe(1, "exactly one node may be Primary"); primaries[0].NodeId.Value.ShouldBe( $"127.0.0.1:{OldestPort}", "Primary must follow cluster-singleton placement (oldest Up member), or the Primary-gated data plane is enabled on a node that is not hosting the work"); } /// The younger node is Secondary, not a second Primary and not Detached. [Fact] public async Task Younger_driver_member_is_secondary() { var snapshot = await CaptureSnapshotAsync(); snapshot.Single(n => n.NodeId.Value == $"127.0.0.1:{YoungestPort}") .Role.ShouldBe(RedundancyRole.Secondary); } /// /// IsDriverPrimary agrees with Role. They feed different consumers — the OPC UA /// ServiceLevel calculation reads the flag while the data-plane gates read the role — so a /// disagreement would advertise one node as authoritative while gating writes on the other. /// [Fact] public async Task IsDriverPrimary_flag_agrees_with_the_role() { var snapshot = await CaptureSnapshotAsync(); foreach (var node in snapshot) { node.IsDriverPrimary.ShouldBe( node.Role == RedundancyRole.Primary, $"{node.NodeId.Value} disagrees between Role and IsDriverPrimary"); } } private async Task> CaptureSnapshotAsync() { var tcs = new TaskCompletionSource( TaskCreationOptions.RunContinuationsAsynchronously); _oldest!.ActorOf( RedundancyStateActor.Props(broadcast: msg => { if (msg is RedundancyStateChanged changed) tcs.TrySetResult(changed); }), $"redundancy-{Guid.NewGuid():N}"); var published = await tcs.Task.WaitAsync(TimeSpan.FromSeconds(15)); return published.Nodes; } }