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

using System.Collections.Concurrent;
using Akka.Actor;
using Akka.Cluster;
using Akka.Cluster.Tools.Singleton;
using Akka.Configuration;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Redundancy;
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
using ZB.MOM.WW.OtOpcUa.ControlPlane.Redundancy;
using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa;
using CommonsNodeId = ZB.MOM.WW.OtOpcUa.Commons.Types.NodeId;

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

/// <summary>
///     Gitea #717 spec §4.2 / plan T3 on a REAL two-node cluster (pattern: <see cref="RedundancyPrimaryElectionTests"/>):
///     the redundancy singleton runs under a real <see cref="ClusterSingletonManager"/> on both nodes, the old
///     incarnation's last publish is held back through the broadcast seam while the singleton hands over, and the held
///     publish is delivered to a real Warm <see cref="OpcUaPublishActor"/> on the surviving node AFTER the new
///     incarnation's. The subscriber must keep the new Primary. Then the ordering assumption the revision rests on is
///     pinned: a removed node cannot come back with its old up-number — its ActorSystem's cluster is gone for good, and
///     only a NEW ActorSystem rejoins, as a younger member.
/// </summary>
public sealed class WarmSingletonHandoverTests : IAsyncLifetime
{
    // Fixed ports: the rejoin control re-binds the first node's port with a new ActorSystem.
    private const int FirstPort = 19_570;  // seed: forms the cluster, oldest, hosts the first singleton incarnation
    private const int SecondPort = 19_571; // joins second: the survivor
    private const string SystemName = "warm-handover";
    private static readonly TimeSpan Budget = TimeSpan.FromSeconds(30);

    private ActorSystem? _first;
    private ActorSystem? _second;
    private ActorSystem? _rejoined;

    private readonly ConcurrentQueue<(int Port, RedundancyStateChanged Msg)> _published = new();
    private readonly ConcurrentQueue<string> _log = new();
    private volatile bool _holdFirst;
    private IActorRef? _subscriber;

    private static Config NodeConfig(int port, bool seedOnFirst = true) => 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 = [{{(seedOnFirst ? $"\"akka.tcp://{SystemName}@127.0.0.1:{FirstPort}\"" : string.Empty)}}]
                roles = ["admin", "driver"]
                min-nr-of-members = 1
                run-coordinated-shutdown-when-down = off
                downing-provider-class = ""
            }
        }
        """);

    public async ValueTask InitializeAsync()
    {
        _first = ActorSystem.Create(SystemName, NodeConfig(FirstPort));
        await WaitForAsync(() => UpCount(_first) >= 1, "the first node Up");
        _second = ActorSystem.Create(SystemName, NodeConfig(SecondPort));
        await WaitForAsync(() => UpCount(_first) == 2 && UpCount(_second) == 2, "both nodes Up on both");
    }

    public async ValueTask DisposeAsync()
    {
        foreach (var system in new[] { _rejoined, _second, _first })
            if (system is not null) await system.Terminate();
    }

    private static int UpCount(ActorSystem system) =>
        Akka.Cluster.Cluster.Get(system).State.Members.Count(m => m.Status == MemberStatus.Up);

    private static async Task WaitForAsync(Func<bool> condition, string what)
    {
        var deadline = DateTime.UtcNow + Budget;
        while (DateTime.UtcNow < deadline)
        {
            if (condition()) return;
            await Task.Delay(50);
        }

        throw new TimeoutException($"timed out waiting for {what}");
    }

    private static CommonsNodeId NodeOf(int port) => CommonsNodeId.Parse($"127.0.0.1:{port}");

    private static CommonsNodeId? PrimaryOf(RedundancyStateChanged msg) =>
        msg.Nodes.FirstOrDefault(n => n.Role == RedundancyRole.Primary)?.NodeId;

    /// <summary>The singleton's broadcast seam on each node: record, and relay to the subscriber unless held back.</summary>
    private void Broadcast(int port, object msg)
    {
        if (msg is not RedundancyStateChanged changed) return;
        _published.Enqueue((port, changed));
        if (port == FirstPort && _holdFirst) return;
        _subscriber?.Tell(changed);
    }

    private void StartSingleton(ActorSystem system, int port) =>
        system.ActorOf(ClusterSingletonManager.Props(
                RedundancyStateActor.Props(msg => Broadcast(port, msg), TimeSpan.FromMilliseconds(100)),
                PoisonPill.Instance,
                ClusterSingletonManagerSettings.Create(system).WithRole("driver")),
            "redundancy-state");

    private RedundancyStateChanged[] PublishedBy(int port) => _published.Where(p => p.Port == port).Select(p => p.Msg).ToArray();

    [Fact]
    public async Task A_held_back_publish_of_the_old_incarnation_never_demotes_the_new_Primary_after_handover()
    {
        // The subscriber: a real Warm OpcUaPublishActor on the node that will survive.
        _subscriber = _second!.ActorOf(OpcUaPublishActor.PropsForTests(
            serviceLevel: new LogPublisher(_log), localNode: NodeOf(SecondPort), dbLess: true,
            healthTickInterval: TimeSpan.FromMinutes(5), redundancy: new WarmRedundancySettings(OpcUaRedundancyMode.Warm),
            standbyProjection: new LogProjection(_log)));
        StartSingleton(_first!, FirstPort);
        StartSingleton(_second!, SecondPort);

        // 1. The first incarnation (on the oldest node) names it Primary; the survivor is Standby.
        await WaitForAsync(() => PublishedBy(FirstPort).Any(m => PrimaryOf(m) == NodeOf(FirstPort)), "the first incarnation's publish");
        await WaitForAsync(() => _log.Contains("SL:90"), "the survivor publishing its Standby level");
        PublishedBy(SecondPort).ShouldBeEmpty("one singleton incarnation at a time");
        _log.ShouldNotContain("SL:250");

        // 2. Let the old incarnation run well ahead on sequence (heartbeats), then hold back its publishes and keep the
        //    last one.
        await WaitForAsync(() => PublishedBy(FirstPort).Length >= 40, "the old incarnation's heartbeats");
        _holdFirst = true;
        var heldFrom = PublishedBy(FirstPort).Length;
        await WaitForAsync(() => PublishedBy(FirstPort).Length > heldFrom, "a held-back heartbeat of the old incarnation");

        // 3. The oldest node is downed (a crash, seen from the survivor); the singleton hands over to the survivor.
        Akka.Cluster.Cluster.Get(_second!).Down(Akka.Cluster.Cluster.Get(_first!).SelfAddress);
        await WaitForAsync(() => Akka.Cluster.Cluster.Get(_second!).State.Members.All(m => m.Address.Port != FirstPort),
            "the downed node removed from the survivor's view");
        await WaitForAsync(() => PublishedBy(SecondPort).Any(m => PrimaryOf(m) == NodeOf(SecondPort)), "the new incarnation's publish");
        await WaitForAsync(() => _log.LastOrDefault() == "SL:250", "the survivor promoting");
        var held = PublishedBy(FirstPort).Last();
        var newest = PublishedBy(SecondPort).Last();

        // Artifact first: the held publish is a real, revisioned, OLDER snapshot that would demote the survivor.
        PrimaryOf(held).ShouldBe(NodeOf(FirstPort));
        held.Revision.ShouldNotBeNull();
        newest.Revision.ShouldNotBeNull();
        held.Revision!.PublisherUpNumber.ShouldBeLessThan(newest.Revision!.PublisherUpNumber,
            "the old incarnation ran on the older member");
        held.Revision.CompareTo(newest.Revision).ShouldBeLessThan(0);
        held.Revision.Sequence.ShouldBeGreaterThan(newest.Revision.Sequence,
            "the old incarnation published more often, so a sequence-only order would let it win");

        // 4. The late publish arrives after the handover — and is dropped. The survivor stays Active.
        var before = _log.Count;
        _subscriber.Tell(held);
        await Task.Delay(1500, TestContext.Current.CancellationToken); // absence window: ~15 heartbeats of the new incarnation are relayed meanwhile
        var after = _log.Skip(before).ToArray();
        after.ShouldNotContain("delivery:True", "the held publish must not demote the new Primary");
        after.ShouldNotContain("SL:90");
        _log.Last().ShouldBe("SL:250");
    }

    /// <summary>
    ///     The assumption under the up-number order (plan T3): a member the cluster has removed cannot rejoin from the
    ///     same ActorSystem — its cluster extension has shut down — so an incarnation hosted there can never publish
    ///     again with its old, lower up-number. A NEW ActorSystem on the same address joins as a new, younger member
    ///     (positive control) whose up-number exceeds the survivor's.
    /// </summary>
    [Fact]
    public async Task A_removed_node_cannot_rejoin_without_a_new_ActorSystem_and_comes_back_younger()
    {
        var survivor = Akka.Cluster.Cluster.Get(_second!);
        var removed = Akka.Cluster.Cluster.Get(_first!);
        var oldUpNumber = RedundancyStateActor.ReadUpNumber(removed.SelfMember)!.Value;
        var survivorUpNumber = RedundancyStateActor.ReadUpNumber(survivor.SelfMember)!.Value;
        oldUpNumber.ShouldBeLessThan(survivorUpNumber);

        survivor.Down(removed.SelfAddress);
        await WaitForAsync(() => survivor.State.Members.All(m => m.Address.Port != FirstPort), "the node removed");
        await WaitForAsync(() => removed.IsTerminated, "the removed node's cluster extension shutting down");
        _first!.WhenTerminated.IsCompleted.ShouldBeFalse("the ActorSystem itself is still alive (run-coordinated-shutdown-when-down = off)");

        // The same ActorSystem asks to join again: nothing happens.
        try { removed.Join(survivor.SelfAddress); }
        catch (Exception) { /* a terminated cluster may refuse outright; either way it must not come back */ }
        var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(4);
        while (DateTime.UtcNow < deadline)
        {
            survivor.State.Members.ShouldNotContain(m => m.Address.Port == FirstPort,
                "a removed member must never reappear from its old ActorSystem");
            await Task.Delay(100, TestContext.Current.CancellationToken);
        }

        // Positive control: a NEW ActorSystem on the same address does join, as a younger member.
        await _first.Terminate();
        _first = null;
        _rejoined = ActorSystem.Create(SystemName, NodeConfig(FirstPort, seedOnFirst: false));
        Akka.Cluster.Cluster.Get(_rejoined).Join(survivor.SelfAddress);
        await WaitForAsync(() => survivor.State.Members.Any(m => m.Address.Port == FirstPort && m.Status == MemberStatus.Up),
            "the new ActorSystem Up");
        var rejoinedMember = survivor.State.Members.Single(m => m.Address.Port == FirstPort);
        RedundancyStateActor.ReadUpNumber(rejoinedMember)!.Value.ShouldBeGreaterThan(survivorUpNumber,
            "the address comes back only as a younger member, so its publishes outrank nothing of the old incarnation's");
        rejoinedMember.UniqueAddress.Uid.ShouldNotBe(removed.SelfUniqueAddress.Uid);
    }

    private sealed class LogPublisher(ConcurrentQueue<string> log) : IServiceLevelPublisher
    {
        public void Publish(byte serviceLevel) => log.Enqueue($"SL:{serviceLevel}");
    }

    private sealed class LogProjection(ConcurrentQueue<string> log) : IStandbyProjection
    {
        public void SetDataProjection(bool projected) => log.Enqueue($"proj:{projected}");
        public void SetStandbyDelivery(bool standby) => log.Enqueue($"delivery:{standby}");
        public void SetAlarmSuppression(bool suppressed) => log.Enqueue($"alarms:{suppressed}");
        public int QueueCurrentToLiveItems(bool projected) { log.Enqueue($"queue:{projected}"); return 0; }
        public int RefreshAllEventItems() { log.Enqueue("refresh"); return 0; }
    }
}