scadaproj/ZB.MOM.WW.Health/tests/ZB.MOM.WW.Health.Akka.Tests/ClusterActiveNodeTests.cs

using Akka.Actor;
using Akka.Cluster;
using Akka.Configuration;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Diagnostics.HealthChecks;
using ZB.MOM.WW.Health.Akka;

namespace ZB.MOM.WW.Health.Akka.Tests;

/// <summary>
/// Pins <b>which</b> node the active tier names, against a real two-node cluster deliberately built
/// so the oldest member is <i>not</i> the lowest-addressed one.
/// </summary>
/// <remarks>
///     <para>
///         The divergence is the whole point. Akka offers two different "first" members:
///         <c>ClusterState.Leader</c> / <c>RoleLeader</c> is the <b>lowest-addressed</b> Up member
///         (host, then port — nothing to do with time), while <b>oldest</b> is the member with the
///         lowest up-number, which is what <c>ClusterSingletonManager</c> uses to place singletons.
///     </para>
///     <para>
///         They coincide on a freshly-formed cluster, which is why every single-node and happy-path
///         test missed it and why the leader-based check survived to 0.2.1. They diverge the moment a
///         node restarts: the restarted node rejoins as the youngest while keeping its address, so if
///         it holds the lower address it becomes leader while the singletons — and every piece of work
///         they own — stay on the other node. This fixture reproduces that state directly instead of
///         waiting for a restart.
///     </para>
/// </remarks>
public sealed class ClusterActiveNodeTests : IAsyncLifetime
{
    // Fixed, unusual ports: the test needs a deterministic address ordering, which port 0 cannot give.
    // Distinct from OtOpcUa's equivalent fixture (19_530/19_531) so the two suites can run at once.
    private const int OldestPort = 19_541;   // joins FIRST  -> oldest, but the HIGHER address
    private const int YoungestPort = 19_540; // joins SECOND -> cluster leader, the LOWER address

    private const string SystemName = "health-active-node";

    private ActorSystem? _oldest;
    private ActorSystem? _youngest;

    private static Config NodeConfig(int port, string roles) => 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:{{OldestPort}}"]
                roles = [{{roles}}]
                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. The oldest
        // node is fused (admin + driver); the younger one is driver-only, so the role-scoped
        // assertions below can distinguish "oldest of this role" from "oldest overall".
        _oldest = ActorSystem.Create(SystemName, NodeConfig(OldestPort, "\"admin\", \"driver\""));
        await WaitForUpAsync(_oldest, expectedMembers: 1);

        _youngest = ActorSystem.Create(SystemName, NodeConfig(YoungestPort, "\"driver\""));
        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 = 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)})");
    }

    /// <summary>
    /// Sanity check on the fixture itself: if the two orderings did not actually diverge, every
    /// assertion below would pass for the wrong reason.
    /// </summary>
    [Fact]
    public void Fixture_actually_produces_divergent_age_and_address_ordering()
    {
        var state = Cluster.Get(_oldest!).State;

        Assert.Equal(YoungestPort, state.Leader!.Port);
        Assert.Equal(YoungestPort, state.RoleLeader("driver")!.Port);
    }

    [Fact]
    public void OldestUpMember_IsTheOldest_NotTheLeader()
    {
        var members = Cluster.Get(_oldest!).State.Members;

        Assert.Equal(OldestPort, ClusterActiveNode.OldestUpMember(members)!.Address.Port);
        Assert.Equal(OldestPort, ClusterActiveNode.OldestUpMember(members, "driver")!.Address.Port);
    }

    [Fact]
    public void OldestUpMember_IsRoleScoped()
    {
        var members = Cluster.Get(_oldest!).State.Members;

        // Only the fused node carries admin, so it owns the admin slot even though it is not the
        // lowest-addressed member.
        Assert.Equal(OldestPort, ClusterActiveNode.OldestUpMember(members, "admin")!.Address.Port);

        // A role nobody carries has no owner.
        Assert.Null(ClusterActiveNode.OldestUpMember(members, "no-such-role"));
    }

    [Fact]
    public void SelfIsActive_IsTrueOnlyOnTheOldestNode()
    {
        Assert.True(ClusterActiveNode.SelfIsActive(Cluster.Get(_oldest!)));
        Assert.False(ClusterActiveNode.SelfIsActive(Cluster.Get(_youngest!)));
    }

    [Fact]
    public void SelfIsActive_RoleScoped_YoungestOwnsNoSlot()
    {
        // Driver-scoped: the fused node is still the oldest driver, so the driver-only node is standby.
        Assert.True(ClusterActiveNode.SelfIsActive(Cluster.Get(_oldest!), "driver"));
        Assert.False(ClusterActiveNode.SelfIsActive(Cluster.Get(_youngest!), "driver"));

        // Admin-scoped: the driver-only node does not carry admin, so it can never be the admin active
        // node — even though it is the cluster leader.
        Assert.False(ClusterActiveNode.SelfIsActive(Cluster.Get(_youngest!), "admin"));
    }

    [Fact]
    public async Task HealthCheck_IsHealthyOnTheOldestAndUnhealthyOnTheLeader()
    {
        var onOldest = await CheckAsync(_oldest!, new ActiveNodeHealthCheckOptions());
        var onYoungest = await CheckAsync(_youngest!, new ActiveNodeHealthCheckOptions());

        Assert.Equal(HealthStatus.Healthy, onOldest.Status);

        // The node the old leader-based check would have called active. Exactly one 200 per cluster is
        // the property an orchestrator's active-tier routing depends on.
        Assert.Equal(HealthStatus.Unhealthy, onYoungest.Status);
    }

    [Fact]
    public async Task HealthCheck_DataNamesTheActiveNode_EvenFromAStandby()
    {
        // A standby reporting *who* is active is what lets a dashboard render the pair from either
        // node's payload alone (the 0.2.0 per-entry data object).
        var result = await CheckAsync(_youngest!, new ActiveNodeHealthCheckOptions());

        Assert.Equal($"akka.tcp://{SystemName}@127.0.0.1:{OldestPort}", result.Data["activeNode"]);
        Assert.Equal($"akka.tcp://{SystemName}@127.0.0.1:{YoungestPort}", result.Data["selfAddress"]);
    }

    [Fact]
    public async Task HealthCheck_RolePreference_FusedNodeAnswersForAdmin()
    {
        var options = new ActiveNodeHealthCheckOptions { RolePreference = ["admin", "driver"] };

        var onOldest = await CheckAsync(_oldest!, options);
        var onYoungest = await CheckAsync(_youngest!, options);

        Assert.Equal(HealthStatus.Healthy, onOldest.Status);
        Assert.Equal("admin", onOldest.Data["activeRole"]);

        // Resolves to driver (it has no admin role) and loses to the older fused node.
        Assert.Equal(HealthStatus.Unhealthy, onYoungest.Status);
        Assert.Equal("driver", onYoungest.Data["activeRole"]);
    }

    [Fact]
    public async Task HealthCheck_NodeOwningNoneOfThePreferredRoles_IsUnhealthyByDefault()
    {
        var options = new ActiveNodeHealthCheckOptions { RolePreference = ["no-such-role"] };

        var result = await CheckAsync(_oldest!, options);

        // Not "not applicable ⇒ Healthy": a node that owns no active work must stay out of the pool.
        Assert.Equal(HealthStatus.Unhealthy, result.Status);
        Assert.Contains("carries none of the active roles", result.Description);
        Assert.DoesNotContain("activeRole", result.Data.Keys);
    }

    [Fact]
    public async Task HealthCheck_NoActiveRoleStatus_IsOverridable()
    {
        var options = new ActiveNodeHealthCheckOptions
        {
            RolePreference = ["no-such-role"],
            NoActiveRoleStatus = HealthStatus.Healthy,
        };

        Assert.Equal(HealthStatus.Healthy, (await CheckAsync(_oldest!, options)).Status);
    }

    [Fact]
    public void Gate_IsActiveOnlyOnTheOldestNode()
    {
        Assert.True(new AkkaActiveNodeGate(ProviderFor(_oldest!)).IsActiveNode);
        Assert.False(new AkkaActiveNodeGate(ProviderFor(_youngest!)).IsActiveNode);

        // Role-scoped: the driver-only node cannot gate admin work.
        Assert.False(new AkkaActiveNodeGate(ProviderFor(_youngest!), "admin").IsActiveNode);
    }

    private static IServiceProvider ProviderFor(ActorSystem system) =>
        new ServiceCollection().AddSingleton(system).BuildServiceProvider();

    private static async Task<HealthCheckResult> CheckAsync(
        ActorSystem system,
        ActiveNodeHealthCheckOptions options)
    {
        var check = new ActiveNodeHealthCheck(ProviderFor(system), options);
        var context = new HealthCheckContext
        {
            Registration = new HealthCheckRegistration("active-node", check, HealthStatus.Unhealthy, tags: null),
        };
        return await check.CheckHealthAsync(context);
    }
}