lmxopcua/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorHistorizationRoleViewTests.cs

using System.Collections.Immutable;
using System.Reflection;
using Akka.Actor;
using Akka.Cluster;
using Akka.Util;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Redundancy;
using ZB.MOM.WW.OtOpcUa.Commons.Types;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Runtime.Drivers;
using ZB.MOM.WW.OtOpcUa.Runtime.Redundancy;
using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness;

namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers;

/// <summary>
///     Gitea #537 — the bridge that carries the continuous-historization <b>capture</b> verdict out of
///     the actor and into <see cref="IHistorizationRoleView"/>, where the recorder reads it per value.
/// </summary>
/// <remarks>
///     <para>
///         Both nodes of a redundant pair serve the same address space and see the same dependency-mux
///         fan-out, so an ungated recorder writes every sample twice. Measured on the docker-dev central
///         pair against a real gateway: the write counter advanced twice per value, and halved when one
///         node was stopped.
///     </para>
///     <para>
///         These cases share the role rule with <see cref="DriverHostActorRoleViewTests"/> but differ on
///         which peer counts. The alarm queue replicates, so the relevant peer is whoever holds a copy
///         of this node's rows; historization is fleet-wide by historian tag name, so the relevant peer
///         is whoever records this node's refs — a partner in its own per-cluster mesh.
///     </para>
/// </remarks>
public sealed class DriverHostActorHistorizationRoleViewTests : RuntimeActorTestBase
{
    private static readonly NodeId TestNode = NodeId.Parse("driver-histview-test");
    private static readonly NodeId MeshPeer = NodeId.Parse("mesh-peer:4053");
    private static readonly NodeId ForeignNode = NodeId.Parse("other-cluster:4053");
    private static readonly TimeSpan Timeout = PresenceBudget;

    /// <summary>
    ///     Before any snapshot arrives the view must read open, not closed.
    /// </summary>
    /// <remarks>
    ///     A deployment that runs no redundancy never publishes here at all. Defaulting closed would
    ///     leave it with no trend data whatsoever and nothing to say why.
    /// </remarks>
    [Fact]
    public void An_unpublished_view_captures()
        => new RedundancyRoleView().ShouldRecordHistorizedValues.ShouldBeTrue();

    /// <summary>A Secondary stands down for a Primary in its own mesh — the duplicate-write case.</summary>
    [Fact]
    public void Secondary_stands_down_for_a_primary_in_its_own_mesh()
    {
        var view = new RecordingHistorizationView();
        SpawnHost(view, sharesScopeWith: MeshPeer).Tell(Snapshot(
            (TestNode, RedundancyRole.Secondary), (MeshPeer, RedundancyRole.Primary)));

        AwaitAssert(() => view.Published.ShouldBe([false]), duration: Timeout);
    }

    /// <summary>
    ///     A Secondary whose Primary is in ANOTHER cluster keeps capturing.
    /// </summary>
    /// <remarks>
    ///     That node runs a different configuration and records none of these tags, so standing down for
    ///     it would end this cluster's historization silently — the drain would simply report success on
    ///     an empty outbox forever. Same scope mismatch that once suspended the alarm drain fleet-wide.
    /// </remarks>
    [Fact]
    public void Secondary_keeps_capturing_when_the_primary_records_other_tags()
    {
        var view = new RecordingHistorizationView();
        SpawnHost(view, sharesScopeWith: MeshPeer).Tell(Snapshot(
            (TestNode, RedundancyRole.Secondary), (ForeignNode, RedundancyRole.Primary)));

        AwaitAssert(() => view.Published.ShouldBe([true]), duration: Timeout);
    }

    /// <summary>
    ///     With NO way to establish which peers share this node's address space, it never stands down.
    /// </summary>
    /// <remarks>
    ///     The predicate is null on a node carrying no cluster-scoped role — the legacy, pre-Phase-6
    ///     shape, where the redundancy election is fleet-wide and the elected Primary usually belongs to
    ///     some other cluster. Duplicates are recoverable; a silent gap is not, so unproven means
    ///     capture. Delete the null check in the wiring and this is the case that goes red.
    /// </remarks>
    [Fact]
    public void An_unprovable_scope_never_stands_down()
    {
        var view = new RecordingHistorizationView();
        SpawnHost(view, sharesScopeWith: null).Tell(Snapshot(
            (TestNode, RedundancyRole.Secondary), (MeshPeer, RedundancyRole.Primary)));

        AwaitAssert(() => view.Published.ShouldBe([true]), duration: Timeout);
    }

    /// <summary>Primary captures.</summary>
    [Fact]
    public void Primary_captures()
    {
        var view = new RecordingHistorizationView();
        SpawnHost(view, sharesScopeWith: MeshPeer).Tell(Snapshot(
            (TestNode, RedundancyRole.Primary), (MeshPeer, RedundancyRole.Secondary)));

        // Published, not current — `true` is also the seed, so asserting the property would pass before
        // the actor had processed anything.
        AwaitAssert(() => view.Published.ShouldBe([true]), duration: Timeout);
    }

    /// <summary>
    ///     Unknown role captures — the snapshot never names this node, the documented identity-mismatch
    ///     shape.
    /// </summary>
    [Fact]
    public void Unknown_role_captures()
    {
        var view = new RecordingHistorizationView();
        SpawnHost(view, sharesScopeWith: MeshPeer).Tell(Snapshot(
            (MeshPeer, RedundancyRole.Primary), (ForeignNode, RedundancyRole.Secondary)));

        AwaitAssert(() => view.Published.ShouldBe([true]), duration: Timeout);
    }

    /// <summary>A promotion must MOVE the view, not merely leave it at its open seed.</summary>
    [Fact]
    public void A_promotion_reopens_a_closed_view()
    {
        var view = new RecordingHistorizationView();
        IActorRef host = SpawnHost(view, sharesScopeWith: MeshPeer);

        host.Tell(Snapshot((TestNode, RedundancyRole.Secondary), (MeshPeer, RedundancyRole.Primary)));
        AwaitAssert(() => view.Published.ShouldBe([false]), duration: Timeout);

        host.Tell(Snapshot((TestNode, RedundancyRole.Primary), (MeshPeer, RedundancyRole.Secondary)));
        AwaitAssert(() => view.Published.ShouldBe([false, true]), duration: Timeout);
    }

    // ---------------- Gitea #672: an unreachable recording Primary ----------------
    //
    // A SIGKILLed Primary stays a member (unreachable, not downed) for the failure-detector + auto-down
    // window, and the redundancy singleton usually died with it — so the survivor's cached snapshot keeps
    // naming the dead node Primary and nobody captures. These drive REAL Akka ReachabilityEvents (built by
    // reflection, as the ControlPlane tests do) into the host, so the host:port projection is exercised too.

    private static readonly NodeId Self672 = NodeId.Parse("self-node:4053");
    private static readonly NodeId Peer672 = NodeId.Parse("mesh-peer:4053");
    private static readonly NodeId Foreign672 = NodeId.Parse("other-cluster:4053");

    /// <summary>
    ///     An unreachable mesh-partner Primary opens the capture gate and its return closes it again — and
    ///     neither moves the alarm-drain verdict nor the Primary data-plane verdict.
    /// </summary>
    /// <remarks>
    ///     The drain/data-plane half is the falsification control: a lazier fix that excluded unreachable
    ///     nodes from every peer question (or published the whole verdict set on a reachability change)
    ///     would let a Secondary claim the Sparkplug host id / drain the replicated alarm queue during a
    ///     mere failure-detector blip.
    /// </remarks>
    [Fact]
    public void An_unreachable_mesh_primary_opens_capture_and_its_return_closes_it_without_touching_other_verdicts()
    {
        var hist = new RecordingHistorizationView();
        var others = new RecordingOtherViews();
        IActorRef host = SpawnHost672(hist, others, captureWhilePrimaryUnreachable: true);

        host.Tell(Snapshot((Self672, RedundancyRole.Secondary), (Peer672, RedundancyRole.Primary)));
        AwaitAssert(() => hist.Published.ShouldBe([false]), duration: Timeout);
        AwaitAssert(() => others.Drained.ShouldNotBeEmpty(), duration: Timeout);
        var drainedBefore = others.Drained;
        var primaryBefore = others.Primary;
        drainedBefore[^1].ShouldBeFalse();   // queue-sharing mesh peer is Primary ⇒ drain stands down
        primaryBefore[^1].ShouldBeFalse();   // Secondary ⇒ not the data-plane Primary

        host.Tell(new ClusterEvent.UnreachableMember(MemberAt(Peer672)));
        AwaitAssert(() => hist.Published.ShouldBe([false, true]), duration: Timeout);

        // Same handler run that published `true` above — anything it published elsewhere is already recorded.
        others.Drained.ShouldBe(drainedBefore);
        others.Primary.ShouldBe(primaryBefore);
        others.Current.ShouldDrainAlarmHistory.ShouldBeFalse();
        others.Current.ShouldActAsPrimary.ShouldBeFalse();

        host.Tell(new ClusterEvent.ReachableMember(MemberAt(Peer672)));
        AwaitAssert(() => hist.Published.ShouldBe([false, true, false]), duration: Timeout);
        others.Drained.ShouldBe(drainedBefore);
        others.Primary.ShouldBe(primaryBefore);
    }

    /// <summary>A peer outside this node's mesh going unreachable changes nothing.</summary>
    [Fact]
    public void An_unreachable_out_of_scope_primary_changes_nothing()
    {
        var hist = new RecordingHistorizationView();
        IActorRef host = SpawnHost672(hist, others: null, captureWhilePrimaryUnreachable: true);

        // Both a foreign-cluster Primary (never stood down for) and the mesh Primary (stood down for).
        host.Tell(Snapshot(
            (Self672, RedundancyRole.Secondary), (Peer672, RedundancyRole.Primary), (Foreign672, RedundancyRole.Primary)));
        AwaitAssert(() => hist.Published.ShouldBe([false]), duration: Timeout);

        host.Tell(new ClusterEvent.UnreachableMember(MemberAt(Foreign672)));
        // Positive control: the mesh peer's loss, sent AFTER, must be the only change — had the foreign
        // one published anything it would sit between the two.
        host.Tell(new ClusterEvent.UnreachableMember(MemberAt(Peer672)));
        AwaitAssert(() => hist.Published.ShouldBe([false, true]), duration: Timeout);
    }

    /// <summary>This node's own reachability (as gossiped by others) is not a peer question.</summary>
    [Fact]
    public void Self_unreachable_is_ignored()
    {
        var hist = new RecordingHistorizationView();
        IActorRef host = SpawnHost672(hist, others: null, captureWhilePrimaryUnreachable: true);

        host.Tell(Snapshot((Self672, RedundancyRole.Secondary), (Peer672, RedundancyRole.Primary)));
        AwaitAssert(() => hist.Published.ShouldBe([false]), duration: Timeout);

        host.Tell(new ClusterEvent.UnreachableMember(MemberAt(Self672)));
        host.Tell(new ClusterEvent.UnreachableMember(MemberAt(Peer672)));
        AwaitAssert(() => hist.Published.ShouldBe([false, true]), duration: Timeout);

        // Self "returning" is ignored too, and it does not close what the peer's loss opened. Events are
        // processed in order, so the peer's return (which MUST close the gate) is the only thing that may
        // appear next: had the self event published anything it would sit between `true` and `false`.
        host.Tell(new ClusterEvent.ReachableMember(MemberAt(Self672)));
        host.Tell(new ClusterEvent.ReachableMember(MemberAt(Peer672)));
        AwaitAssert(() => hist.Published.ShouldBe([false, true, false]), duration: Timeout);
    }

    /// <summary>
    ///     A host that starts while its peer is ALREADY unreachable (the <c>InitialStateAsEvents</c> replay
    ///     delivers the <c>UnreachableMember</c> before the first redundancy snapshot) must still treat that
    ///     peer as unreachable when the snapshot naming it Primary finally arrives.
    /// </summary>
    [Fact]
    public void A_peer_unreachable_before_the_first_snapshot_still_opens_capture()
    {
        var hist = new RecordingHistorizationView();
        IActorRef host = SpawnHost672(hist, others: null, captureWhilePrimaryUnreachable: true);

        host.Tell(new ClusterEvent.UnreachableMember(MemberAt(Peer672)));
        host.Tell(Snapshot((Self672, RedundancyRole.Secondary), (Peer672, RedundancyRole.Primary)));

        // Exactly one publication, and it is `true`: no spurious `false` from treating the peer as reachable.
        AwaitAssert(() => hist.Published.ShouldBe([true]), duration: Timeout);
    }

    /// <summary><c>CaptureWhilePrimaryUnreachable = false</c> restores the pre-#672 behaviour.</summary>
    [Fact]
    public void With_the_option_off_an_unreachable_primary_keeps_the_gate_closed()
    {
        var hist = new RecordingHistorizationView();
        IActorRef host = SpawnHost672(hist, others: null, captureWhilePrimaryUnreachable: false);

        host.Tell(Snapshot((Self672, RedundancyRole.Secondary), (Peer672, RedundancyRole.Primary)));
        AwaitAssert(() => hist.Published.ShouldBe([false]), duration: Timeout);

        host.Tell(new ClusterEvent.UnreachableMember(MemberAt(Peer672)));
        // Positive control: a re-sent snapshot publishes on every arrival, so a `true` from the
        // reachability event would show up between the two `false`s.
        host.Tell(Snapshot((Self672, RedundancyRole.Secondary), (Peer672, RedundancyRole.Primary)));
        AwaitAssert(() => hist.Published.ShouldBe([false, false]), duration: Timeout);
    }

    /// <summary>
    ///     A fresh snapshot naming this node Primary — the singleton re-homed after auto-down removed the
    ///     dead peer — keeps the gate open; the removal itself must not flap it closed in between.
    /// </summary>
    [Fact]
    public void A_snapshot_naming_me_primary_after_the_peer_is_removed_keeps_capturing()
    {
        var hist = new RecordingHistorizationView();
        IActorRef host = SpawnHost672(hist, others: null, captureWhilePrimaryUnreachable: true);

        host.Tell(Snapshot((Self672, RedundancyRole.Secondary), (Peer672, RedundancyRole.Primary)));
        AwaitAssert(() => hist.Published.ShouldBe([false]), duration: Timeout);

        host.Tell(new ClusterEvent.UnreachableMember(MemberAt(Peer672)));
        AwaitAssert(() => hist.Published.ShouldBe([false, true]), duration: Timeout);

        host.Tell(new ClusterEvent.MemberRemoved(MemberAt(Peer672, MemberStatus.Removed), MemberStatus.Down));
        host.Tell(Snapshot((Self672, RedundancyRole.Primary)));
        AwaitAssert(() => hist.Published.ShouldBe([false, true, true]), duration: Timeout);
    }

    /// <summary>A re-joined incarnation at the dead peer's address is reachable again.</summary>
    [Fact]
    public void A_rejoined_peer_is_reachable_again()
    {
        var hist = new RecordingHistorizationView();
        IActorRef host = SpawnHost672(hist, others: null, captureWhilePrimaryUnreachable: true);

        host.Tell(Snapshot((Self672, RedundancyRole.Secondary), (Peer672, RedundancyRole.Primary)));
        host.Tell(new ClusterEvent.UnreachableMember(MemberAt(Peer672)));
        AwaitAssert(() => hist.Published.ShouldBe([false, true]), duration: Timeout);

        host.Tell(new ClusterEvent.MemberUp(MemberAt(Peer672)));
        AwaitAssert(() => hist.Published.ShouldBe([false, true, false]), duration: Timeout);
    }

    /// <summary>
    ///     End to end on a REAL two-node cluster: crash the recording Primary and the survivor's capture gate
    ///     opens on failure detection alone — while the dead node is still a member, i.e. well before any
    ///     auto-down could have removed it and re-homed the redundancy singleton.
    /// </summary>
    /// <remarks>
    ///     Downing is disabled outright here, so the dead node stays a member forever: the only signal that
    ///     can open the gate is the failure detector's <c>UnreachableMember</c>. The crash is
    ///     <c>Terminate()</c> with <c>run-by-actor-system-terminate = off</c>, which gossips no Leave — a
    ///     process death, not a graceful drain (pattern from <c>SelfFirstSeedBootstrapTests</c>).
    /// </remarks>
    [Fact]
    public async Task A_crashed_primary_opens_the_survivors_capture_gate_before_it_is_downed()
    {
        var portA = FreePort();
        var portB = FreePort();
        const string systemName = "otopcua-672";
        var sysA = ActorSystem.Create(systemName, TwoNodeHocon(portA, portA));
        var sysB = ActorSystem.Create(systemName, TwoNodeHocon(portB, portA));
        try
        {
            var clusterB = Akka.Cluster.Cluster.Get(sysB);
            AwaitCondition(
                () => clusterB.State.Members.Count(m => m.Status == MemberStatus.Up) == 2,
                TimeSpan.FromSeconds(30));

            var nodeA = NodeId.Parse($"127.0.0.1:{portA}");
            var nodeB = NodeId.Parse($"127.0.0.1:{portB}");
            var hist = new RecordingHistorizationView();
            IActorRef host = sysB.ActorOf(DriverHostActor.Props(
                dbFactory: null,
                nodeB,
                ackRouter: null,
                driverMemberCountProvider: () => 2,
                historizationRoleView: hist,
                sharesHistorizationScopeWithLocal: node => node == nodeA));

            host.Tell(Snapshot((nodeB, RedundancyRole.Secondary), (nodeA, RedundancyRole.Primary)));
            AwaitAssert(() => hist.Published.ShouldBe([false]), duration: Timeout);

            await sysA.Terminate();

            AwaitAssert(() => hist.Published.ShouldBe([false, true]), duration: TimeSpan.FromSeconds(30));
            // Still a member: the gate opened on reachability, not on a removal + fresh snapshot.
            clusterB.State.Members.ShouldContain(m => m.Address.Port == portA);
            clusterB.State.Unreachable.ShouldContain(m => m.Address.Port == portA);
        }
        finally
        {
            await sysB.Terminate();
            if (!sysA.WhenTerminated.IsCompleted) await sysA.Terminate();
        }
    }

    private static int FreePort()
    {
        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;
    }

    private static Akka.Configuration.Config TwoNodeHocon(int selfPort, int seedPort) =>
        Akka.Configuration.ConfigurationFactory.ParseString($@"
akka {{
    loglevel = ""WARNING""
    extensions = [""Akka.Cluster.Tools.PublishSubscribe.DistributedPubSubExtensionProvider, Akka.Cluster.Tools""]
    actor.provider = ""Akka.Cluster.ClusterActorRefProvider, Akka.Cluster""
    remote.dot-netty.tcp {{ hostname = ""127.0.0.1"", port = {selfPort} }}
    coordinated-shutdown.run-by-actor-system-terminate = off
    cluster {{
        seed-nodes = [""akka.tcp://otopcua-672@127.0.0.1:{seedPort}""]
        roles = [""driver""]
        downing-provider-class = """"
        run-coordinated-shutdown-when-down = off
        failure-detector {{ heartbeat-interval = 1s, acceptable-heartbeat-pause = 3s }}
    }}
}}");

    // ---------------- helpers ----------------

    /// <summary>Records the drain and data-plane verdicts the host publishes.</summary>
    private sealed class RecordingOtherViews : IRedundancyRoleView, IDriverPrimaryRoleSource
    {
        private readonly List<bool> _drained = [];
        private readonly List<bool> _primary = [];
        private readonly Lock _gate = new();

        public RedundancyRoleView Current { get; } = new();

        public IReadOnlyList<bool> Drained
        {
            get { lock (_gate) return _drained.ToArray(); }
        }

        public IReadOnlyList<bool> Primary
        {
            get { lock (_gate) return _primary.ToArray(); }
        }

        public bool ShouldDrainAlarmHistory => Current.ShouldDrainAlarmHistory;

        public bool ShouldActAsPrimary => Current.ShouldActAsPrimary;

        public void Publish(bool shouldDrainAlarmHistory)
        {
            lock (_gate) _drained.Add(shouldDrainAlarmHistory);
            Current.Publish(shouldDrainAlarmHistory);
        }

        public void PublishPrimaryRole(bool shouldActAsPrimary)
        {
            lock (_gate) _primary.Add(shouldActAsPrimary);
            Current.PublishPrimaryRole(shouldActAsPrimary);
        }
    }

    private IActorRef SpawnHost672(
        IHistorizationRoleView view, RecordingOtherViews? others, bool captureWhilePrimaryUnreachable) =>
        Sys.ActorOf(DriverHostActor.Props(
            NewInMemoryDbFactory(),
            Self672,
            ackRouter: null,
            driverMemberCountProvider: () => 2,
            redundancyRoleView: others,
            primaryRoleSource: others,
            historizationRoleView: view,
            sharesHistorizationScopeWithLocal: node => node == Peer672 || node == Self672,
            sharesAlarmQueueWithLocal: node => node == Peer672,
            captureWhilePrimaryUnreachable: captureWhilePrimaryUnreachable));

    private static int _memberUid;

    // Akka's Member.Create factories are internal — same reflection idiom as the ControlPlane tests.
    private static readonly MethodInfo MemberCreate = typeof(Member).GetMethod(
            "Create",
            BindingFlags.Static | BindingFlags.NonPublic | BindingFlags.Public,
            binder: null,
            types:
            [
                typeof(UniqueAddress), typeof(int), typeof(MemberStatus),
                typeof(ImmutableHashSet<string>), typeof(AppVersion),
            ],
            modifiers: null)
        ?? throw new InvalidOperationException("Akka.Cluster.Member.Create(5-arg) not found — Akka API changed");

    private static Member MemberAt(NodeId node, MemberStatus status = MemberStatus.Up)
    {
        var text = node.ToString();
        var colon = text.LastIndexOf(':');
        return (Member)MemberCreate.Invoke(null,
        [
            new UniqueAddress(
                new Address("akka.tcp", "otopcua", text[..colon], int.Parse(text[(colon + 1)..])),
                Interlocked.Increment(ref _memberUid)),
            1,
            status,
            ImmutableHashSet.Create("driver"),
            AppVersion.Zero,
        ])!;
    }

    /// <summary>
    ///     A view that remembers every published decision, so a test asserts on the actor's ACTION
    ///     rather than on a state its open seed already satisfies.
    /// </summary>
    private sealed class RecordingHistorizationView : IHistorizationRoleView
    {
        private readonly List<bool> _published = [];
        private readonly Lock _gate = new();
        private readonly RedundancyRoleView _inner = new();

        /// <summary>Every value published so far, in order.</summary>
        public IReadOnlyList<bool> Published
        {
            get { lock (_gate) return _published.ToArray(); }
        }

        public bool ShouldRecordHistorizedValues => _inner.ShouldRecordHistorizedValues;

        public void PublishHistorizationRole(bool shouldRecordHistorizedValues)
        {
            lock (_gate) _published.Add(shouldRecordHistorizedValues);
            _inner.PublishHistorizationRole(shouldRecordHistorizedValues);
        }
    }

    private IActorRef SpawnHost(IHistorizationRoleView view, NodeId? sharesScopeWith) =>
        Sys.ActorOf(DriverHostActor.Props(
            NewInMemoryDbFactory(),
            TestNode,
            ackRouter: null,
            driverMemberCountProvider: () => 2,
            historizationRoleView: view,
            sharesHistorizationScopeWithLocal:
                sharesScopeWith is { } peer ? node => node == peer : null));

    private static RedundancyStateChanged Snapshot(params (NodeId Node, RedundancyRole Role)[] nodes) =>
        new(
            nodes.Select(n => new NodeRedundancyState(
                    n.Node,
                    n.Role,
                    IsClusterLeader: n.Role == RedundancyRole.Primary,
                    IsDriverPrimary: n.Role == RedundancyRole.Primary,
                    AsOfUtc: DateTime.UtcNow))
                .ToList(),
            CorrelationId.NewId());
}