ScadaBridge/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/Cluster/TwoNodeClusterFixture.cs

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;

/// <summary>
/// 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.
///
/// Timing knobs default to scaled test values; pass production values
/// (2s heartbeat / 10s failure-detection / 15s stable-after) to measure the
/// real failover envelope (FailoverTimingTests).
/// </summary>
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<TwoNodeClusterFixture> StartAsync(
        string role = "Central", TimeSpan? stableAfter = null,
        int? portA = null, int? portB = null,
        TimeSpan? heartbeatInterval = null, TimeSpan? failureDetectionThreshold = null)
    {
        var f = new TwoNodeClusterFixture();
        f.PortA = portA ?? GetFreeTcpPort();
        f.PortB = portB ?? GetFreeTcpPort();
        f.NodeA = f.StartNode(f.PortA, role, stableAfter, heartbeatInterval, failureDetectionThreshold);
        await WaitForMembersUp(f.NodeA, 1, TimeSpan.FromSeconds(20));
        f.NodeB = f.StartNode(f.PortB, role, stableAfter, heartbeatInterval, failureDetectionThreshold);
        await WaitForMembersUp(f.NodeA, 2, TimeSpan.FromSeconds(20));
        await WaitForMembersUp(f.NodeB, 2, TimeSpan.FromSeconds(20));
        return f;
    }

    /// <summary>Starts a node from production HOCON; used by StartAsync and by restart-scenarios.</summary>
    public ActorSystem StartNode(int port, string role, TimeSpan? stableAfter = null,
        TimeSpan? heartbeatInterval = null, TimeSpan? failureDetectionThreshold = null)
    {
        var nodeOptions = new NodeOptions { Role = role, NodeHostname = "127.0.0.1", RemotingPort = port };
        var clusterOptions = new ClusterOptions
        {
            SeedNodes = new List<string>
            {
                $"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 = heartbeatInterval ?? TimeSpan.FromMilliseconds(500),
            FailureDetectionThreshold = 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);
    }

    /// <summary>Hard-crash a node: abrupt terminate, no cluster leave.</summary>
    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.");
    }

    public 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 */ }
            }
        }
    }
}