ScadaBridge/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/ShutOutRestartTests.cs

using System.Collections.Immutable;
using System.Reflection;
using Akka.Actor;
using Akka.Cluster;
using Akka.Configuration;
using Akka.Remote;
using Akka.TestKit.Xunit2;
using Akka.Util;
using ZB.MOM.WW.ScadaBridge.Host.Actors;

namespace ZB.MOM.WW.ScadaBridge.Host.Tests;

/// <summary>
/// ScadaBridge#132: the decision logic of <see cref="ShutOutRestartBase"/>. Cluster events and
/// <see cref="ThisActorSystemQuarantinedEvent"/> are fed straight to the actor; the restart and
/// probe seams report to the test actor. Covers the single-side shut-out (restart, no tiebreak),
/// the tiebreak in both orders, the both-sides case where exactly one of two nodes restarts, the
/// winner's grace, and the probing that makes a healed partition produce the event at all. The
/// real two-node behaviour is proven in IntegrationTests (<c>ShutOutRestartScenarioTests</c>).
/// </summary>
public class ShutOutRestartTests : TestKit
{
    /// <summary>INFO so the decision-log assertions see every line.</summary>
    public ShutOutRestartTests()
        : base("akka.loglevel = INFO")
    {
    }

    // "a" < "b" < "c" by Akka's member address ordering (host, then port).
    private static readonly Address NodeA = new("akka.tcp", "scadabridge", "a", 8081);
    private static readonly Address NodeB = new("akka.tcp", "scadabridge", "b", 8081);
    private static readonly Address Stranger = new("akka.tcp", "scadabridge", "z", 9999);

    private static readonly ShutOutRestartSettings Fast = new(
        Enabled: true,
        Grace: TimeSpan.FromSeconds(1.5),
        ProbeInterval: TimeSpan.FromMilliseconds(200),
        ProbeMaxInterval: TimeSpan.FromMilliseconds(800));

    private sealed record Restarted(Address Self);

    private sealed record Probed(Address Self, Address Peer);

    /// <summary>Test double: restart and probe report to a probe actor.</summary>
    private sealed class TestShutOutRestart : ShutOutRestartBase
    {
        private readonly IActorRef _probe;
        private readonly Address[] _seeds;

        public TestShutOutRestart(Address self, Address[] seeds, ShutOutRestartSettings settings, IActorRef probe)
            : base(settings)
        {
            SelfAddress = self;
            _seeds = seeds;
            _probe = probe;
        }

        public override Address SelfAddress { get; }

        public override IScheduler Scheduler => Context.System.Scheduler;

        protected override IEnumerable<Address> SeedNodes => _seeds;

        protected override void RestartSelf() => _probe.Tell(new Restarted(SelfAddress));

        protected override void Probe(Address peer) => _probe.Tell(new Probed(SelfAddress, peer));
    }

    private static Member CreateMember(Address address, MemberStatus status)
    {
        var create = typeof(Member).GetMethod(
            "Create",
            BindingFlags.Static | BindingFlags.NonPublic,
            new[] { typeof(UniqueAddress), typeof(int), typeof(MemberStatus), typeof(ImmutableHashSet<string>), typeof(AppVersion) })
            ?? throw new InvalidOperationException("Akka.Cluster.Member.Create(…) not found — Akka internals changed.");
        return (Member)create.Invoke(null, new object[]
        {
            new UniqueAddress(address, 1), 1, status, ImmutableHashSet<string>.Empty, AppVersion.Zero,
        })!;
    }

    private static ClusterEvent.CurrentClusterState StateWith(params Member[] members) =>
        new(
            ImmutableSortedSet.CreateRange(members),
            ImmutableHashSet<Member>.Empty,
            ImmutableHashSet<Address>.Empty,
            null,
            ImmutableDictionary<string, Address>.Empty);

    private IActorRef Node(Address self, Address peer, IActorRef? probe = null, ShutOutRestartSettings? settings = null) =>
        Sys.ActorOf(Props.Create(() => new TestShutOutRestart(
            self, new[] { self, peer }, settings ?? Fast, probe ?? TestActor)));

    /// <summary>Both nodes see each other Up, then each removes the other after downing it (the partition outcome).</summary>
    private static void BothRemovedEachOther(IActorRef node, Address self, Address peer)
    {
        node.Tell(StateWith(CreateMember(self, MemberStatus.Up), CreateMember(peer, MemberStatus.Up)));
        node.Tell(new ClusterEvent.MemberRemoved(CreateMember(peer, MemberStatus.Removed), MemberStatus.Down));
    }

    private static ThisActorSystemQuarantinedEvent ShutOutBy(Address peer, Address self) => new(self, peer);

    /// <summary>S2 as Akka.Remote 1.5.62 surfaces it: the endpoint writer's InvalidAssociation wrapping the protocol refusal.</summary>
    private static AssociationErrorEvent StillRunningRemoved(Address peer, Address self) =>
        new(
            new Exception(
                $"Association failed with {peer}",
                new AggregateException(new Exception("The remote system has a UID that has been quarantined. Association aborted."))),
            self, peer, inbound: false, Akka.Event.LogLevel.WarningLevel);

    /// <summary>Everything <paramref name="probe"/> receives during <paramref name="window"/>.</summary>
    private static List<object> Drain(Akka.TestKit.TestProbe probe, TimeSpan window)
    {
        var received = new List<object>();
        var until = DateTime.UtcNow + window;
        while (true)
        {
            var left = until - DateTime.UtcNow;
            if (left <= TimeSpan.Zero)
            {
                return received;
            }

            if (probe.TryReceiveOne(out var envelope, left))
            {
                received.Add(envelope.Message);
            }
        }
    }

    /// <summary>
    /// Single side (the frozen node): the peer is still a live member in this node's view, so only
    /// this side can be shut out — it restarts at once, whichever address is higher.
    /// </summary>
    /// <param name="selfIsA">Whether this node is the lower-address node.</param>
    [Theory]
    [InlineData(true)]
    [InlineData(false)]
    public void ShutOut_PeerStillAMember_RestartsWithoutTiebreak(bool selfIsA)
    {
        var (self, peer) = selfIsA ? (NodeA, NodeB) : (NodeB, NodeA);
        var node = Node(self, peer);
        node.Tell(StateWith(CreateMember(self, MemberStatus.Up), CreateMember(peer, MemberStatus.Up)));

        EventFilter.Warning(contains: "tiebreak not needed").ExpectOne(() => node.Tell(ShutOutBy(peer, self)));
        ExpectMsg(new Restarted(self));
        ExpectNoMsg(TimeSpan.FromMilliseconds(300));
    }

    /// <summary>Tiebreak, higher address: both sides removed each other; the higher address restarts at once.</summary>
    [Fact]
    public void ShutOut_PeerRemoved_HigherAddress_Restarts()
    {
        var node = Node(NodeB, NodeA);
        BothRemovedEachOther(node, NodeB, NodeA);

        EventFilter.Warning(contains: "Tiebreak says RESTART").ExpectOne(() => node.Tell(ShutOutBy(NodeA, NodeB)));
        ExpectMsg(new Restarted(NodeB));
    }

    /// <summary>
    /// Tiebreak, lower address: it stays; it keeps probing during the grace, and at the end of the
    /// grace — the peer silent, i.e. it restarted — it still stays.
    /// </summary>
    [Fact]
    public void ShutOut_PeerRemoved_LowerAddress_Stays_AndStaysAfterGraceWhenPeerWentQuiet()
    {
        var probe = CreateTestProbe();
        var node = Node(NodeA, NodeB, probe.Ref);
        BothRemovedEachOther(node, NodeA, NodeB);

        EventFilter.Warning(contains: "Tiebreak says STAY").ExpectOne(() => node.Tell(ShutOutBy(NodeB, NodeA)));
        EventFilter.Info(contains: "nothing of its old incarnation has been seen").ExpectOne(TimeSpan.FromSeconds(4), () => { });

        var received = Drain(probe, TimeSpan.FromMilliseconds(500));
        Assert.Contains(new Probed(NodeA, NodeB), received);      // it kept probing its peer
        Assert.DoesNotContain(new Restarted(NodeA), received);    // and never restarted
    }

    /// <summary>
    /// Tiebreak, lower address, but the peer keeps refusing for the whole grace and never rejoins:
    /// the lower node was the stale side after all (single-sided with the lower address) and restarts.
    /// </summary>
    [Fact]
    public async Task ShutOut_PeerRemoved_LowerAddress_StillRefusedAfterGrace_Restarts()
    {
        var probe = CreateTestProbe();
        var node = Node(NodeA, NodeB, probe.Ref);
        BothRemovedEachOther(node, NodeA, NodeB);
        node.Tell(ShutOutBy(NodeB, NodeA));

        var until = DateTime.UtcNow + Fast.Grace + TimeSpan.FromSeconds(1);
        var restarted = false;
        while (DateTime.UtcNow < until && !restarted)
        {
            node.Tell(ShutOutBy(NodeB, NodeA)); // every probe is refused
            await Task.Delay(100);
            while (probe.HasMessages)
            {
                restarted |= probe.ReceiveOne(TimeSpan.Zero) is Restarted;
            }
        }

        Assert.True(restarted, "the lower node never restarted although its peer kept refusing it past the grace");
    }

    /// <summary>Tiebreak, lower address: the peer rejoins during the grace; the lower node stays.</summary>
    [Fact]
    public void ShutOut_PeerRemoved_LowerAddress_PeerRejoins_Stays()
    {
        var probe = CreateTestProbe();
        var node = Node(NodeA, NodeB, probe.Ref);
        BothRemovedEachOther(node, NodeA, NodeB);
        node.Tell(ShutOutBy(NodeB, NodeA));
        node.Tell(ShutOutBy(NodeB, NodeA)); // refused right up to the restart of the peer
        node.Tell(new ClusterEvent.MemberUp(CreateMember(NodeB, MemberStatus.Up)));

        EventFilter.Info(contains: "rejoined").ExpectOne(TimeSpan.FromSeconds(4), () => { });
        Assert.DoesNotContain(new Restarted(NodeA), Drain(probe, TimeSpan.FromMilliseconds(500)));
    }

    /// <summary>
    /// Both sides shut out (a healed partition): two nodes, each having removed the other, both
    /// receive the event — exactly ONE restarts, and it is the higher address. Run in both
    /// delivery orders so the outcome cannot depend on who hears first.
    /// </summary>
    /// <param name="higherHearsFirst">Whether the higher-address node receives its event first.</param>
    [Theory]
    [InlineData(true)]
    [InlineData(false)]
    public void BothSidesShutOut_ExactlyOneRestarts_TheHigherAddress(bool higherHearsFirst)
    {
        var probe = CreateTestProbe();
        var a = Node(NodeA, NodeB, probe.Ref);
        var b = Node(NodeB, NodeA, probe.Ref);
        BothRemovedEachOther(a, NodeA, NodeB);
        BothRemovedEachOther(b, NodeB, NodeA);

        if (higherHearsFirst)
        {
            b.Tell(ShutOutBy(NodeA, NodeB));
            a.Tell(ShutOutBy(NodeB, NodeA));
        }
        else
        {
            a.Tell(ShutOutBy(NodeB, NodeA));
            b.Tell(ShutOutBy(NodeA, NodeB));
        }

        // Within the grace, the only restart is B's.
        var restarts = Drain(probe, Fast.Grace - TimeSpan.FromMilliseconds(300)).OfType<Restarted>().ToList();
        Assert.Equal(new[] { new Restarted(NodeB) }, restarts);
    }

    /// <summary>The same event delivered many times restarts once.</summary>
    [Fact]
    public void RepeatedShutOut_RestartsOnce()
    {
        var node = Node(NodeB, NodeA);
        node.Tell(StateWith(CreateMember(NodeB, MemberStatus.Up), CreateMember(NodeA, MemberStatus.Up)));
        for (var i = 0; i < 5; i++)
        {
            node.Tell(ShutOutBy(NodeA, NodeB));
        }

        ExpectMsg(new Restarted(NodeB));
        ExpectNoMsg(TimeSpan.FromMilliseconds(500));
    }

    /// <summary>A quarantine by an address that is neither a seed nor a member is not this pair's business.</summary>
    [Fact]
    public void ShutOutByStranger_IsIgnored()
    {
        var node = Node(NodeB, NodeA);
        node.Tell(StateWith(CreateMember(NodeB, MemberStatus.Up), CreateMember(NodeA, MemberStatus.Up)));

        EventFilter.Info(contains: "not a seed or member of this cluster").ExpectOne(() => node.Tell(ShutOutBy(Stranger, NodeB)));
        ExpectNoMsg(TimeSpan.FromMilliseconds(500));
    }

    /// <summary>
    /// A peer removed after being DOWNED is probed with a back-off until it is a member again; a
    /// peer that LEFT (Exiting → Removed) is not probed.
    /// </summary>
    [Fact]
    public void DownedPeer_IsProbedUntilItRejoins_LeftPeerIsNot()
    {
        var node = Node(NodeA, NodeB);
        node.Tell(StateWith(CreateMember(NodeA, MemberStatus.Up), CreateMember(NodeB, MemberStatus.Up)));

        node.Tell(new ClusterEvent.MemberRemoved(CreateMember(NodeB, MemberStatus.Removed), MemberStatus.Exiting));
        ExpectNoMsg(TimeSpan.FromMilliseconds(600));

        node.Tell(new ClusterEvent.MemberRemoved(CreateMember(NodeB, MemberStatus.Removed), MemberStatus.Down));
        ExpectMsg(new Probed(NodeA, NodeB), TimeSpan.FromSeconds(1));
        ExpectMsg(new Probed(NodeA, NodeB), TimeSpan.FromSeconds(1));

        node.Tell(new ClusterEvent.MemberJoined(CreateMember(NodeB, MemberStatus.Joining)));
        ReceiveWhile<object>(TimeSpan.FromMilliseconds(100), m => m); // a tick already in flight
        ExpectNoMsg(TimeSpan.FromSeconds(1.5));
    }

    /// <summary>
    /// S2 on the higher node (it removed a peer whose old incarnation is still alive): no restart
    /// during the grace — a frozen peer gets the chance to restart itself — then, still split, the
    /// tiebreak restarts it.
    /// </summary>
    [Fact]
    public async Task StillRunningRemoved_HigherAddress_WaitsGrace_ThenRestartsIfStillSplit()
    {
        var probe = CreateTestProbe();
        var node = Node(NodeB, NodeA, probe.Ref);
        BothRemovedEachOther(node, NodeB, NodeA);

        EventFilter.Warning(contains: "Giving it").ExpectOne(() => node.Tell(StillRunningRemoved(NodeA, NodeB)));
        var started = DateTime.UtcNow;
        var restartedAt = (DateTime?)null;
        while (DateTime.UtcNow - started < Fast.Grace + TimeSpan.FromSeconds(1.5) && restartedAt is null)
        {
            node.Tell(StillRunningRemoved(NodeA, NodeB)); // every probe still meets the old incarnation
            await Task.Delay(100);
            while (probe.HasMessages)
            {
                if (probe.ReceiveOne(TimeSpan.Zero) is Restarted)
                {
                    restartedAt = DateTime.UtcNow;
                }
            }
        }

        Assert.NotNull(restartedAt);
        Assert.True(restartedAt.Value - started >= Fast.Grace - TimeSpan.FromMilliseconds(200),
            $"restarted after {(restartedAt.Value - started).TotalSeconds:0.0}s, before the {Fast.Grace} grace");
    }

    /// <summary>
    /// The freeze survivor: S2 (the frozen node's old incarnation is alive), then the frozen node
    /// restarts itself and its old incarnation goes quiet — the survivor does NOT restart, even
    /// though it has the higher address.
    /// </summary>
    [Fact]
    public void StillRunningRemoved_PeerGoesQuiet_NoRestartEvenWhenHigher()
    {
        var probe = CreateTestProbe();
        var node = Node(NodeB, NodeA, probe.Ref);
        BothRemovedEachOther(node, NodeB, NodeA);
        node.Tell(StillRunningRemoved(NodeA, NodeB));

        EventFilter.Info(contains: "nothing of its old incarnation has been seen").ExpectOne(TimeSpan.FromSeconds(4), () => { });
        Assert.DoesNotContain(new Restarted(NodeB), Drain(probe, TimeSpan.FromMilliseconds(500)));
    }

    /// <summary>S2, then the peer's new incarnation joins during the grace: no restart.</summary>
    [Fact]
    public void StillRunningRemoved_PeerRejoins_NoRestart()
    {
        var probe = CreateTestProbe();
        var node = Node(NodeB, NodeA, probe.Ref);
        BothRemovedEachOther(node, NodeB, NodeA);
        node.Tell(StillRunningRemoved(NodeA, NodeB));
        node.Tell(new ClusterEvent.MemberUp(CreateMember(NodeA, MemberStatus.Up)));
        node.Tell(StillRunningRemoved(NodeA, NodeB)); // stale evidence from before the restart

        EventFilter.Info(contains: "rejoined").ExpectOne(TimeSpan.FromSeconds(4), () => { });
        Assert.DoesNotContain(new Restarted(NodeB), Drain(probe, TimeSpan.FromMilliseconds(500)));
    }

    /// <summary>
    /// A healed partition as Akka 1.5.62 really reports it — S2 on BOTH sides, no S1 — ends with
    /// exactly one restart, the higher address, after the grace; the lower stays.
    /// </summary>
    [Fact]
    public async Task BothSidesStillRunningRemoved_ExactlyOneRestarts_TheHigherAddress()
    {
        var probe = CreateTestProbe();
        var a = Node(NodeA, NodeB, probe.Ref);
        var b = Node(NodeB, NodeA, probe.Ref);
        BothRemovedEachOther(a, NodeA, NodeB);
        BothRemovedEachOther(b, NodeB, NodeA);

        var restarts = new List<Restarted>();
        var until = DateTime.UtcNow + Fast.Grace + Fast.Grace - TimeSpan.FromMilliseconds(400);
        while (DateTime.UtcNow < until)
        {
            if (!restarts.Contains(new Restarted(NodeB)))
            {
                // Until B's old incarnation is gone, each side's probes meet the other's.
                a.Tell(StillRunningRemoved(NodeB, NodeA));
                b.Tell(StillRunningRemoved(NodeA, NodeB));
            }

            await Task.Delay(100);
            while (probe.HasMessages)
            {
                if (probe.ReceiveOne(TimeSpan.Zero) is Restarted r)
                {
                    restarts.Add(r);
                }
            }
        }

        Assert.Equal(new[] { new Restarted(NodeB) }, restarts);
    }

    /// <summary>S1 arriving while an S2 grace runs: the peer has quarantined us too, so the tiebreak runs at once.</summary>
    [Fact]
    public void StillRunningRemoved_ThenShutOut_TiebreaksImmediately()
    {
        var node = Node(NodeB, NodeA);
        BothRemovedEachOther(node, NodeB, NodeA);
        node.Tell(StillRunningRemoved(NodeA, NodeB));

        EventFilter.Warning(contains: "Tiebreak says RESTART").ExpectOne(() => node.Tell(ShutOutBy(NodeA, NodeB)));
        FishForMessage(m => m is Restarted, TimeSpan.FromSeconds(1));
    }

    /// <summary>S2 detection matches Akka 1.5.62's two wordings anywhere in the cause chain, and nothing else.</summary>
    [Fact]
    public void IsQuarantinedUidRefusal_MatchesAkkaWordingOnly()
    {
        Assert.True(ShutOutRestartBase.IsQuarantinedUidRefusal(
            new Exception("Association failed with x", new AggregateException(
                new Exception("The remote system has a UID that has been quarantined. Association aborted.")))));
        Assert.True(ShutOutRestartBase.IsQuarantinedUidRefusal(
            new Exception("The remote system [akka.tcp://scadabridge@b:8081] has a UID [42] that has been quarantined. Association aborted.")));
        Assert.False(ShutOutRestartBase.IsQuarantinedUidRefusal(
            new Exception("Association failed with x", new Exception("Connection refused"))));
        Assert.False(ShutOutRestartBase.IsQuarantinedUidRefusal(
            new Exception("The remote system has quarantined this system. No further associations to the remote system are possible until this system is restarted.")));
        Assert.False(ShutOutRestartBase.IsQuarantinedUidRefusal(null));
    }

    /// <summary>Settings: defaults when keys are missing; the switch and durations are read when present.</summary>
    [Fact]
    public void Settings_FromConfig_DefaultsAndOverrides()
    {
        var defaults = ShutOutRestartSettings.FromConfig(ConfigurationFactory.Empty);
        Assert.Equal(new ShutOutRestartSettings(true, TimeSpan.FromSeconds(60), TimeSpan.FromSeconds(5), TimeSpan.FromSeconds(60)), defaults);

        var set = ShutOutRestartSettings.FromConfig(ConfigurationFactory.ParseString($@"
{ShutOutRestartSettings.EnabledPath} = off
{ShutOutRestartSettings.GracePath} = 2500ms
{ShutOutRestartSettings.ProbeIntervalPath} = 1s
{ShutOutRestartSettings.ProbeMaxIntervalPath} = 4s"));
        Assert.Equal(new ShutOutRestartSettings(false, TimeSpan.FromMilliseconds(2500), TimeSpan.FromSeconds(1), TimeSpan.FromSeconds(4)), set);
    }
}