lmxopcua/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/ManualFailoverServiceTests.cs

using Akka.Actor;
using Akka.Cluster;
using Akka.Configuration;
using Microsoft.Extensions.Logging.Abstractions;
using Shouldly;
using Xunit;
using ZB.MOM.WW.Audit;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Redundancy;
using ZB.MOM.WW.OtOpcUa.ControlPlane.Redundancy;

namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Tests;

/// <summary>
/// Pins <b>which</b> node a manual failover removes, on a real two-node cluster built so that the
/// oldest member is not the lowest-addressed one.
/// </summary>
/// <remarks>
///     <para>
///         The fixture mirrors <see cref="RedundancyPrimaryElectionTests"/> deliberately, and for the
///         same reason: age order and address order coincide on a freshly-formed cluster and diverge
///         after any restart. A failover that targeted the <i>role leader</i> (lowest address) rather
///         than the oldest member would, after a restart, remove the node that is <b>not</b> hosting
///         the singletons — leaving the Primary exactly where it was while taking an innocent node
///         down. Node A therefore binds the <b>higher</b> port and joins first.
///     </para>
///     <para>
///         The parity test is the durable one: it pins the service's target to the member
///         <see cref="RedundancyStateActor"/> actually names Primary, so the two five-line queries
///         cannot drift apart silently.
///     </para>
/// </remarks>
public sealed class ManualFailoverServiceTests : IAsyncLifetime
{
    // Distinct from RedundancyPrimaryElectionTests' 19530/19531 — xunit may run the two classes in
    // parallel, and both need deterministic (non-zero) ports.
    private const int OldestPort = 19_541;   // joins FIRST  -> oldest, but the HIGHER address
    private const int YoungestPort = 19_540; // joins SECOND -> role leader, the LOWER address

    private ActorSystem? _oldest;
    private ActorSystem? _youngest;

    private static Config NodeConfig(int port, string systemName, int seedPort) =>
        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://{{systemName}}@127.0.0.1:{{seedPort}}"]
                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("manual-failover", NodeConfig(OldestPort, "manual-failover", OldestPort));
        await WaitForUpAsync(_oldest, expectedMembers: 1);

        _youngest = ActorSystem.Create("manual-failover", NodeConfig(YoungestPort, "manual-failover", OldestPort));
        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)
        {
            if (cluster.State.Members.Count(m => m.Status == MemberStatus.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)})");
    }

    private static (ManualFailoverService Service, RecordingAuditWriter Audit) BuildService(ActorSystem system)
    {
        var audit = new RecordingAuditWriter();
        return (new ManualFailoverService(() => system, audit, NullLogger<ManualFailoverService>.Instance), audit);
    }

    /// <summary>
    /// Fixture sanity check: if age order and address order did not actually diverge, the assertion
    /// below would pass for the wrong reason.
    /// </summary>
    [Fact]
    public void Fixture_actually_produces_divergent_age_and_address_ordering()
    {
        Akka.Cluster.Cluster.Get(_oldest!).State.RoleLeader("driver")!.Port.ShouldBe(
            YoungestPort,
            "the fixture is only meaningful if the lowest-addressed node is the YOUNGER one");
    }

    /// <summary>
    /// The failover removes the oldest driver member — the node hosting the singletons — not the
    /// role leader, and the survivor is then elected Primary.
    /// </summary>
    [Fact]
    public async Task FailOver_targets_the_oldest_driver_member_not_the_role_leader()
    {
        // Driven from the survivor, as production does: the admin node asks the Primary to leave.
        var (service, audit) = BuildService(_youngest!);

        var target = await service.FailOverDriverPrimaryAsync("tester");

        target.ShouldNotBeNull();
        target.ShouldContain(
            $":{OldestPort}",
            Case.Sensitive,
            "the failover must remove the OLDEST driver member (which holds the singletons), not the role leader");

        // The leaving node runs CoordinatedShutdown and terminates — that is what makes the process
        // exit and be restarted as the youngest member.
        await _oldest!.WhenTerminated.WaitAsync(TimeSpan.FromSeconds(30));

        // ... and the survivor is now what RedundancyStateActor names Primary.
        var snapshot = await CaptureSnapshotAsync(_youngest!);
        var primaries = snapshot.Where(n => n.Role == RedundancyRole.Primary).ToList();
        primaries.Count.ShouldBe(1, "exactly one node may be Primary after the swap");
        primaries[0].NodeId.Value.ShouldBe($"127.0.0.1:{YoungestPort}");

        audit.Events.Count.ShouldBe(1, "a manual failover must be audited exactly once");
        audit.Events[0].Action.ShouldBe(ManualFailoverService.AuditAction);
        audit.Events[0].Actor.ShouldBe("tester");
        audit.Events[0].SourceNode.ShouldNotBeNull().ShouldContain($":{OldestPort}");
    }

    /// <summary>
    /// With no peer to take over, a failover is a disguised shutdown. It must refuse — and must not
    /// audit, because nothing happened.
    /// </summary>
    [Fact]
    public async Task FailOver_refuses_when_no_driver_peer_exists()
    {
        const int LonePort = 19_542;
        var lone = ActorSystem.Create("manual-failover-lone", NodeConfig(LonePort, "manual-failover-lone", LonePort));
        try
        {
            await WaitForUpAsync(lone, expectedMembers: 1);
            var (service, audit) = BuildService(lone);

            var result = await service.FailOverDriverPrimaryAsync("tester");

            result.ShouldBeNull("a single driver member has nowhere to fail over to");
            audit.Events.ShouldBeEmpty("a refused failover changed nothing and must not be audited");

            // Positive assert: the node is still a live Up member — the refusal did not half-leave it.
            Akka.Cluster.Cluster.Get(lone).State.Members
                .Count(m => m.Status == MemberStatus.Up)
                .ShouldBe(1);
            lone.WhenTerminated.IsCompleted.ShouldBeFalse();
        }
        finally
        {
            await lone.Terminate();
        }
    }

    /// <summary>
    /// The service's target and <see cref="RedundancyStateActor.SelectDriverPrimary"/> are two copies
    /// of the same query. This pins them together, so a change to one that is not made to the other
    /// fails here rather than in production, where it would mean failing over the wrong node.
    /// </summary>
    [Fact]
    public async Task Parity_with_SelectDriverPrimary()
    {
        var elected = RedundancyStateActor.SelectDriverPrimary(
            Akka.Cluster.Cluster.Get(_oldest!).State.Members);

        var target = ManualFailoverService.FailOverCore(_oldest!, dryRun: true);
        var snapshotPrimary = BuildService(_oldest!).Service.GetSnapshot().PrimaryAddress;

        elected.ShouldNotBeNull();
        target.ShouldBe(elected, "the node acted on must be exactly the node the election names Primary");
        snapshotPrimary.ShouldBe(elected.ToString(), "the UI panel must name the same node the button acts on");

        // The dry run really was dry.
        await Task.Delay(500);
        Akka.Cluster.Cluster.Get(_oldest!).State.Members
            .Count(m => m.Status == MemberStatus.Up)
            .ShouldBe(2, "a dry run must not remove anyone");
    }

    private static async Task<IReadOnlyList<NodeRedundancyState>> CaptureSnapshotAsync(ActorSystem system)
    {
        var tcs = new TaskCompletionSource<RedundancyStateChanged>(
            TaskCreationOptions.RunContinuationsAsynchronously);

        system.ActorOf(
            RedundancyStateActor.Props(broadcast: msg =>
            {
                if (msg is RedundancyStateChanged changed) tcs.TrySetResult(changed);
            }),
            $"redundancy-{Guid.NewGuid():N}");

        return (await tcs.Task.WaitAsync(TimeSpan.FromSeconds(15))).Nodes;
    }

    private sealed class RecordingAuditWriter : IAuditWriter
    {
        public List<AuditEvent> Events { get; } = new();

        public Task WriteAsync(AuditEvent auditEvent, CancellationToken cancellationToken = default)
        {
            Events.Add(auditEvent);
            return Task.CompletedTask;
        }
    }
}