diff --git a/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/Cluster/TwoNodeClusterFixture.cs b/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/Cluster/TwoNodeClusterFixture.cs new file mode 100644 index 00000000..ede98c32 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/Cluster/TwoNodeClusterFixture.cs @@ -0,0 +1,120 @@ +using Akka.Actor; +using Akka.Cluster; +using Akka.Configuration; +using ZB.MOM.WW.ScadaBridge.ClusterInfrastructure; +using ZB.MOM.WW.ScadaBridge.Host; +using ZB.MOM.WW.ScadaBridge.Host.Actors; + +namespace ZB.MOM.WW.ScadaBridge.IntegrationTests.Cluster; + +/// +/// Forms a REAL two-node Akka cluster in-process from the production +/// BuildHocon output. NodeA is always started first (=> oldest member / +/// singleton host). CrashNode() terminates a node WITHOUT cluster-leave +/// (coordinated-shutdown by-terminate disabled on both nodes) to simulate +/// a hard crash — the scenario review 01 found untested. +/// +public sealed class TwoNodeClusterFixture : IAsyncDisposable +{ + public ActorSystem NodeA { get; private set; } = null!; + public ActorSystem NodeB { get; private set; } = null!; + public int PortA { get; private set; } + public int PortB { get; private set; } + + public static async Task StartAsync( + string role = "Central", TimeSpan? stableAfter = null) + { + var f = new TwoNodeClusterFixture(); + f.PortA = GetFreeTcpPort(); + f.PortB = GetFreeTcpPort(); + f.NodeA = f.StartNode(f.PortA, role, stableAfter); + await WaitForMembersUp(f.NodeA, 1, TimeSpan.FromSeconds(20)); + f.NodeB = f.StartNode(f.PortB, role, stableAfter); + await WaitForMembersUp(f.NodeA, 2, TimeSpan.FromSeconds(20)); + await WaitForMembersUp(f.NodeB, 2, TimeSpan.FromSeconds(20)); + return f; + } + + /// Starts a node from production HOCON; used by StartAsync and by restart-scenarios. + public ActorSystem StartNode(int port, string role, TimeSpan? stableAfter = null) + { + var nodeOptions = new NodeOptions { Role = role, NodeHostname = "127.0.0.1", RemotingPort = port }; + var clusterOptions = new ClusterOptions + { + SeedNodes = new List + { + $"akka.tcp://scadabridge@127.0.0.1:{PortA}", + $"akka.tcp://scadabridge@127.0.0.1:{PortB}", + }, + SplitBrainResolverStrategy = "keep-oldest", + StableAfter = stableAfter ?? TimeSpan.FromSeconds(3), + HeartbeatInterval = TimeSpan.FromMilliseconds(500), + FailureDetectionThreshold = TimeSpan.FromSeconds(2), + MinNrOfMembers = 1, + DownIfAlone = true, + }; + var hocon = AkkaHostedService.BuildHocon( + nodeOptions, clusterOptions, new[] { role }, + TimeSpan.FromSeconds(1), TimeSpan.FromSeconds(3)); + // Crash simulation: Terminate() must NOT run CoordinatedShutdown + // (no Leave gossip) or the test degenerates to the graceful path. + var config = ConfigurationFactory + .ParseString("akka.coordinated-shutdown.run-by-actor-system-terminate = off") + .WithFallback(ConfigurationFactory.ParseString(hocon)); + return ActorSystem.Create("scadabridge", config); + } + + /// Hard-crash a node: abrupt terminate, no cluster leave. + public static Task CrashNode(ActorSystem node) => node.Terminate(); + + public static async Task WaitForMembersUp(ActorSystem system, int count, TimeSpan timeout) + { + var cluster = Akka.Cluster.Cluster.Get(system); + var deadline = DateTime.UtcNow + timeout; + while (DateTime.UtcNow < deadline) + { + if (cluster.State.Members.Count(m => m.Status == MemberStatus.Up) == count + && !cluster.State.Unreachable.Any()) + return; + await Task.Delay(200); + } + throw new TimeoutException( + $"Cluster on {cluster.SelfAddress} did not reach {count} Up members within {timeout}. " + + $"Members: [{string.Join(", ", cluster.State.Members)}], " + + $"Unreachable: [{string.Join(", ", cluster.State.Unreachable)}]"); + } + + public static async Task WaitForMemberRemoved(ActorSystem survivor, Address removed, TimeSpan timeout) + { + var cluster = Akka.Cluster.Cluster.Get(survivor); + var deadline = DateTime.UtcNow + timeout; + while (DateTime.UtcNow < deadline) + { + if (cluster.State.Members.All(m => m.Address != removed) + && cluster.State.Unreachable.All(m => m.Address != removed)) + return; + await Task.Delay(200); + } + throw new TimeoutException($"Member {removed} was never removed from {cluster.SelfAddress}'s view — SBR did not down it."); + } + + private static int GetFreeTcpPort() + { + var listener = new System.Net.Sockets.TcpListener(System.Net.IPAddress.Loopback, 0); + listener.Start(); + var port = ((System.Net.IPEndPoint)listener.LocalEndpoint).Port; + listener.Stop(); + return port; + } + + public async ValueTask DisposeAsync() + { + foreach (var sys in new[] { NodeA, NodeB }) + { + if (sys is { WhenTerminated.IsCompleted: false }) + { + try { await sys.Terminate().WaitAsync(TimeSpan.FromSeconds(10)); } catch { /* teardown */ } + } + } + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/Cluster/TwoNodeClusterFixtureTests.cs b/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/Cluster/TwoNodeClusterFixtureTests.cs new file mode 100644 index 00000000..abf71bcd --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/Cluster/TwoNodeClusterFixtureTests.cs @@ -0,0 +1,18 @@ +using Xunit; + +namespace ZB.MOM.WW.ScadaBridge.IntegrationTests.Cluster; + +public class TwoNodeClusterFixtureTests +{ + [Fact] + public async Task Fixture_FormsTwoNodeCluster_NodeAIsOldest() + { + await using var cluster = await TwoNodeClusterFixture.StartAsync(); + var a = Akka.Cluster.Cluster.Get(cluster.NodeA); + var b = Akka.Cluster.Cluster.Get(cluster.NodeB); + Assert.Equal(2, a.State.Members.Count); + Assert.Equal(2, b.State.Members.Count); + // NodeA started first => oldest (the singleton host). + Assert.True(a.SelfMember.IsOlderThan(b.State.Members.First(m => m.Address == b.SelfAddress))); + } +}