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

using System.Diagnostics.CodeAnalysis;
using Akka.Actor;
using Akka.Cluster.Tools.PublishSubscribe;
using Microsoft.EntityFrameworkCore;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Cluster;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Mesh;
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
using ZB.MOM.WW.OtOpcUa.Commons.Protos.MeshTransport.V1;
using ZB.MOM.WW.OtOpcUa.Commons.Types;
using ZB.MOM.WW.OtOpcUa.Configuration;
using ZB.MOM.WW.OtOpcUa.Configuration.Entities;
using ZB.MOM.WW.OtOpcUa.Configuration.Enums;
using ZB.MOM.WW.OtOpcUa.ControlPlane.Communication;
using ZB.MOM.WW.OtOpcUa.ControlPlane.Tests.Harness;

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

public class CentralCommunicationActorTests : ControlPlaneActorTestBase
{
    private const string DeploymentsTopic = "deployments";

    private static MeshTransportOptions GrpcMode(int refreshSeconds = 60) => new()
    {
        Mode = MeshTransportOptions.ModeGrpc,
        ContactRefreshSeconds = refreshSeconds,
        DispatchTimeoutSeconds = 10,
    };

    /// <summary>Seeds a cluster whose nodes carry a <c>GrpcPort</c> — the command plane's shape.</summary>
    private static void SeedGrpcCluster(
        IDbContextFactory<OtOpcUaConfigDbContext> factory,
        string clusterId,
        params (string NodeId, string Host, int? GrpcPort, bool Enabled, bool Maintenance)[] nodes)
    {
        using var db = factory.CreateDbContext();
        db.ServerClusters.Add(new ServerCluster
        {
            ClusterId = clusterId,
            Name = clusterId,
            Enterprise = "ZB",
            Site = "Site",
            RedundancyMode = RedundancyMode.None,
            NodeCount = 1,
            CreatedBy = "test",
        });
        foreach (var n in nodes)
        {
            db.ClusterNodes.Add(new ClusterNode
            {
                NodeId = n.NodeId,
                ClusterId = clusterId,
                Host = n.Host,
                AkkaPort = 4053,
                GrpcPort = n.GrpcPort,
                ApplicationUri = $"urn:{n.NodeId}",
                Enabled = n.Enabled,
                MaintenanceMode = n.Maintenance,
                CreatedBy = "test",
            });
        }

        db.SaveChanges();
    }

    private IActorRef Spawn(
        IDbContextFactory<OtOpcUaConfigDbContext> factory,
        MeshTransportOptions options,
        IActorRef? coordinator = null) =>
        Sys.ActorOf(CentralCommunicationActor.Props(factory, options, () => coordinator));

    [Fact]
    public void Dps_mode_publishes_on_the_carried_topic()
    {
        // Dps is the pair-local default and stays in the tree: it is the correct transport for a
        // single-mesh topology, which every integration test and every profile that leaves
        // MeshTransport:Mode unset still runs.
        var factory = NewInMemoryDbFactory();
        var mediatorProbe = CreateTestProbe();
        // Subscribe on the probe's behalf: SubscribeAck goes to the SENDER, so telling the mediator
        // without an explicit sender would ack to TestActor and this wait would hang.
        DistributedPubSub.Get(Sys).Mediator.Tell(
            new Subscribe(DeploymentsTopic, mediatorProbe.Ref), mediatorProbe.Ref);
        mediatorProbe.ExpectMsg<SubscribeAck>(TimeSpan.FromSeconds(5));

        var actor = Spawn(factory, new MeshTransportOptions());
        var dispatch = new DispatchDeployment(
            new DeploymentId(Guid.NewGuid()), new RevisionHash("abc"), new CorrelationId(Guid.NewGuid()));

        actor.Tell(new MeshCommand(DeploymentsTopic, dispatch));

        mediatorProbe.ExpectMsg<DispatchDeployment>(TimeSpan.FromSeconds(5)).ShouldBe(dispatch);
    }

    [Fact]
    public void ApplyAck_is_forwarded_to_the_coordinator_preserving_the_sender()
    {
        var factory = NewInMemoryDbFactory();
        var coordinator = CreateTestProbe();
        var node = CreateTestProbe();
        var actor = Spawn(factory, GrpcMode(), coordinator.Ref);

        var ack = new ApplyAck(
            new DeploymentId(Guid.NewGuid()),
            new NodeId("host-a:4053"),
            ApplyAckOutcome.Applied,
            null,
            new CorrelationId(Guid.NewGuid()));

        actor.Tell(ack, node.Ref);

        coordinator.ExpectMsg<ApplyAck>(TimeSpan.FromSeconds(5)).ShouldBe(ack);
        // Forward, not Tell: the coordinator must see the node, not this relay.
        coordinator.LastSender.ShouldBe(node.Ref);
    }

    [Fact]
    public void An_unresolvable_coordinator_logs_rather_than_losing_the_ack_silently()
    {
        var factory = NewInMemoryDbFactory();
        var actor = Spawn(factory, GrpcMode(), coordinator: null);

        var ack = new ApplyAck(
            new DeploymentId(Guid.NewGuid()),
            new NodeId("host-a:4053"),
            ApplyAckOutcome.Applied,
            null,
            new CorrelationId(Guid.NewGuid()));

        EventFilter.Warning(contains: "not resolvable").ExpectOne(TimeSpan.FromSeconds(5), () =>
            actor.Tell(ack));
    }

    [Fact]
    public void A_faulted_contact_load_warns_and_leaves_the_actor_alive()
    {
        var factory = NewInMemoryDbFactory();
        var actor = Spawn(factory, GrpcMode());

        // Filter on a phrase unique to the DB-failure warning. EventFilter matches case-INSENSITIVELY,
        // so a looser "was NOT" would also catch this actor's other warnings and the assertion would
        // fail with two matches instead of one.
        EventFilter.Warning(contains: "contact set was NOT refreshed").ExpectOne(
            TimeSpan.FromSeconds(5),
            () => actor.Tell(new Status.Failure(new InvalidOperationException("db down"))));

        // Still serving: a transient DB outage must not take the command boundary down with it.
        actor.Tell(new MeshCommand(DeploymentsTopic, "anything"));
        ExpectNoMsg(TimeSpan.FromMilliseconds(200));
    }

    // ------------------------------------------------------------------ Mode = Grpc
    //
    // The cross-mesh transport. The fan-out must reach EVERY enabled, non-maintenance node in every
    // cluster, because the node side has no ClusterId or node filter at all; scoping happens later,
    // inside the artifact. A narrower fan would leave nodes silently on their old configuration
    // while the deployment sealed green.

    [Fact]
    public void Grpc_mode_fans_one_command_to_every_enabled_node_across_all_clusters()
    {
        // THE assertion of this transport: one command, one call per node, across every cluster.
        var factory = NewInMemoryDbFactory();
        SeedGrpcCluster(
            factory,
            "C1",
            ("host-a:4053", "host-a", 4056, true, false),
            ("host-b:4053", "host-b", 4056, true, false));
        SeedGrpcCluster(factory, "C2", ("host-c:4053", "host-c", 4056, true, false));
        var rig = SpawnGrpcRig(factory);
        AwaitContactsLoaded(rig);

        rig.Actor.Tell(new MeshCommand(DeploymentsTopic, NewDispatch()));

        AwaitCondition(() => rig.Dialer.Calls.Count == 3, TimeSpan.FromSeconds(5));
        rig.Dialer.Calls.Select(c => c.Endpoint).ShouldBe(
            new[] { "http://host-a:4056", "http://host-b:4056", "http://host-c:4056" }, ignoreOrder: true);

        // Each node is dialled with ITS OWN cluster's id — that is what selects the pre-shared key,
        // so getting it wrong turns every command into a PermissionDenied at the node.
        rig.Dialer.Calls.Single(c => c.Endpoint.Contains("host-c")).ClusterId.ShouldBe("C2");
        rig.Dialer.Calls.Where(c => !c.Endpoint.Contains("host-c")).ShouldAllBe(c => c.ClusterId == "C1");
    }

    [Fact]
    public void Grpc_mode_publishes_nothing_on_pubsub()
    {
        // DPS is mesh-wide and cannot cross an application-Cluster boundary, so a command that also
        // went out on the topic would be delivered twice inside central's own mesh and not at all
        // anywhere else.
        var factory = NewInMemoryDbFactory();
        SeedGrpcCluster(factory, "C1", ("host-a:4053", "host-a", 4056, true, false));
        var mediatorProbe = CreateTestProbe();
        DistributedPubSub.Get(Sys).Mediator.Tell(
            new Subscribe(DeploymentsTopic, mediatorProbe.Ref), mediatorProbe.Ref);
        mediatorProbe.ExpectMsg<SubscribeAck>(TimeSpan.FromSeconds(5));

        var rig = SpawnGrpcRig(factory);
        AwaitContactsLoaded(rig);

        rig.Actor.Tell(new MeshCommand(DeploymentsTopic, NewDispatch()));

        AwaitCondition(() => rig.Dialer.Calls.Count == 1, TimeSpan.FromSeconds(5));
        mediatorProbe.ExpectNoMsg(TimeSpan.FromMilliseconds(300));
    }

    [Fact]
    public void Grpc_mode_does_not_dial_disabled_or_maintenance_nodes()
    {
        // Disabled and maintenance-mode rows are not nodes central talks to; the same filter the
        // contact load applies, asserted here so the fan cannot drift from it.
        var factory = NewInMemoryDbFactory();
        SeedGrpcCluster(
            factory,
            "C1",
            ("host-a:4053", "host-a", 4056, true, false),
            ("host-b:4053", "host-b", 4056, false, false),
            ("host-c:4053", "host-c", 4056, true, true));
        var rig = SpawnGrpcRig(factory);
        AwaitContactsLoaded(rig);

        rig.Actor.Tell(new MeshCommand(DeploymentsTopic, NewDispatch()));

        AwaitCondition(() => rig.Dialer.Calls.Count == 1, TimeSpan.FromSeconds(5));
        rig.Dialer.Calls[0].Endpoint.ShouldContain("host-a");

        // And no late arrival from the two excluded rows.
        ExpectNoMsg(TimeSpan.FromMilliseconds(300));
        rig.Dialer.Calls.Count.ShouldBe(1);
    }

    [Fact]
    public void A_node_with_no_GrpcPort_is_skipped_with_a_warning_and_the_others_are_still_dialled()
    {
        // Mirrors TelemetryNodeSource: a row with no port exposes no endpoint yet. It must cost that
        // node its commands, never the whole fan.
        var factory = NewInMemoryDbFactory();
        SeedGrpcCluster(
            factory,
            "C1",
            ("host-a:4053", "host-a", null, true, false),
            ("host-b:4053", "host-b", 4056, true, false));

        GrpcRig? rig = null;
        EventFilter.Warning(contains: "has no GrpcPort").ExpectOne(
            TimeSpan.FromSeconds(5), () => rig = SpawnGrpcRig(factory));

        rig.ShouldNotBeNull();
        AwaitContactsLoaded(rig);
        rig.Actor.Tell(new MeshCommand(DeploymentsTopic, NewDispatch()));

        AwaitCondition(() => rig.Dialer.Calls.Count == 1, TimeSpan.FromSeconds(5));
        rig.Dialer.Calls[0].Endpoint.ShouldBe("http://host-b:4056");
    }

    [Fact]
    public void A_node_that_regains_then_loses_its_GrpcPort_warns_again()
    {
        // Warn-once is per continuous skip run, not per process. The suppression exists so a
        // permanently portless row does not repeat every minute; carried across a repair it would
        // silence the SECOND time a node loses its endpoint — the occurrence an operator most needs
        // to see, because it is a regression rather than a known gap.
        var factory = NewInMemoryDbFactory();
        SeedGrpcCluster(factory, "C1", ("host-a:4053", "host-a", null, true, false));
        var rig = SpawnGrpcRig(factory);
        AwaitContactsLoaded(rig);

        SetGrpcPort(factory, "host-a:4053", 4056);
        rig.Actor.Tell(new CentralCommunicationActor.RefreshContacts());
        AwaitCondition(
            () => rig.Psk.KnownClusterPublications.Count >= 2, TimeSpan.FromSeconds(5));

        SetGrpcPort(factory, "host-a:4053", null);
        EventFilter.Warning(contains: "has no GrpcPort").ExpectOne(TimeSpan.FromSeconds(5), () =>
            rig.Actor.Tell(new CentralCommunicationActor.RefreshContacts()));
    }

    [Fact]
    public void One_failing_node_does_not_stop_the_fan_and_is_logged_by_node_id()
    {
        // Fire-and-forget: a failure is logged and DROPPED. No retry, no queue, and above all no
        // abort of the remaining nodes — the deploy path's apply deadline is the backstop.
        var factory = NewInMemoryDbFactory();
        SeedGrpcCluster(
            factory,
            "C1",
            ("host-a:4053", "host-a", 4056, true, false),
            ("host-b:4053", "host-b", 4056, true, false));
        var dialer = new RecordingDialer
        {
            FailEndpointsContaining = "host-a",
            Fault = new Grpc.Core.RpcException(
                new Grpc.Core.Status(Grpc.Core.StatusCode.Unavailable, "node down")),
        };
        var rig = SpawnGrpcRig(factory, dialer);
        AwaitContactsLoaded(rig);

        // The node id, not the endpoint: an operator reading the log is looking at ClusterNode rows.
        EventFilter.Warning(contains: "host-a:4053").ExpectOne(TimeSpan.FromSeconds(5), () =>
            rig.Actor.Tell(new MeshCommand(DeploymentsTopic, NewDispatch())));

        // Both were attempted; the healthy one is unaffected by its neighbour's failure.
        AwaitCondition(() => rig.Dialer.Calls.Count == 2, TimeSpan.FromSeconds(5));

        // Still serving after a per-node failure — no supervision restart, no state loss.
        rig.Actor.Tell(new MeshCommand(DeploymentsTopic, NewDispatch()));
        AwaitCondition(() => rig.Dialer.Calls.Count == 4, TimeSpan.FromSeconds(5));
    }

    [Fact]
    public void A_psk_miss_costs_only_that_node_and_never_becomes_a_fan_level_abort()
    {
        // The dialer refuses a keyless cluster locally, so it arrives here as an ordinary per-node
        // failure. The fan-level assertion is that the other cluster is still dialled.
        var factory = NewInMemoryDbFactory();
        SeedGrpcCluster(factory, "C1", ("host-a:4053", "host-a", 4056, true, false));
        SeedGrpcCluster(factory, "C2", ("host-b:4053", "host-b", 4056, true, false));
        var dialer = new RecordingDialer
        {
            FailClusters = ["C2"],
            Fault = new Grpc.Core.RpcException(
                new Grpc.Core.Status(Grpc.Core.StatusCode.Unauthenticated, "no key")),
        };
        var rig = SpawnGrpcRig(factory, dialer);
        AwaitContactsLoaded(rig);

        EventFilter.Warning(contains: "Unauthenticated").ExpectOne(TimeSpan.FromSeconds(5), () =>
            rig.Actor.Tell(new MeshCommand(DeploymentsTopic, NewDispatch())));

        AwaitCondition(() => rig.Dialer.Calls.Count == 2, TimeSpan.FromSeconds(5));
    }

    [Fact]
    public void A_command_with_no_grpc_target_is_dropped_with_a_warning()
    {
        var factory = NewInMemoryDbFactory();
        var rig = SpawnGrpcRig(factory);
        AwaitContactsLoaded(rig);

        EventFilter.Warning(contains: "dropping").ExpectOne(TimeSpan.FromSeconds(5), () =>
            rig.Actor.Tell(new MeshCommand(DeploymentsTopic, NewDispatch())));

        rig.Dialer.Calls.ShouldBeEmpty();
    }

    [Fact]
    public void A_payload_that_is_not_a_mesh_command_record_is_dropped_with_an_error()
    {
        // The mapper refuses to encode an unknown payload rather than shipping something the node
        // cannot decode. That refusal must cost the command, not the actor.
        var factory = NewInMemoryDbFactory();
        SeedGrpcCluster(factory, "C1", ("host-a:4053", "host-a", 4056, true, false));
        var rig = SpawnGrpcRig(factory);
        AwaitContactsLoaded(rig);

        EventFilter.Error(contains: "Cannot encode").ExpectOne(TimeSpan.FromSeconds(5), () =>
            rig.Actor.Tell(new MeshCommand(DeploymentsTopic, "not a command record")));

        rig.Dialer.Calls.ShouldBeEmpty();

        // Still serving.
        rig.Actor.Tell(new MeshCommand(DeploymentsTopic, NewDispatch()));
        AwaitCondition(() => rig.Dialer.Calls.Count == 1, TimeSpan.FromSeconds(5));
    }

    [Fact]
    public void The_envelope_carries_the_commands_own_correlation_and_the_whole_fan_shares_it()
    {
        var factory = NewInMemoryDbFactory();
        SeedGrpcCluster(
            factory,
            "C1",
            ("host-a:4053", "host-a", 4056, true, false),
            ("host-b:4053", "host-b", 4056, true, false));
        var rig = SpawnGrpcRig(factory);
        AwaitContactsLoaded(rig);

        var correlationId = new CorrelationId(Guid.NewGuid());
        rig.Actor.Tell(new MeshCommand(DeploymentsTopic, new DispatchDeployment(
            new DeploymentId(Guid.NewGuid()), new RevisionHash("abc"), correlationId)));

        AwaitCondition(() => rig.Dialer.Calls.Count == 2, TimeSpan.FromSeconds(5));
        rig.Dialer.Calls.ShouldAllBe(c => c.Envelope.CorrelationId == correlationId.ToString());
        // One envelope object, encoded once and fanned — not re-encoded per node.
        rig.Dialer.Calls.Select(c => c.Envelope).Distinct().Count().ShouldBe(1);
    }

    [Fact]
    public void An_alarm_command_gets_one_minted_transport_correlation_shared_by_the_whole_fan()
    {
        // AlarmCommand is the one record with no correlation member of its own, so the envelope's is
        // the only handle an operator has on it. Minting per fan rather than per node is what makes
        // "this alarm command" one traceable event instead of N unrelated ones.
        var factory = NewInMemoryDbFactory();
        SeedGrpcCluster(
            factory,
            "C1",
            ("host-a:4053", "host-a", 4056, true, false),
            ("host-b:4053", "host-b", 4056, true, false));
        var rig = SpawnGrpcRig(factory);
        AwaitContactsLoaded(rig);

        rig.Actor.Tell(new MeshCommand(
            "alarms", new AlarmCommand("alarm-1", "Acknowledge", "operator", null, null)));

        AwaitCondition(() => rig.Dialer.Calls.Count == 2, TimeSpan.FromSeconds(5));
        rig.Dialer.Calls.Select(c => c.Envelope.CorrelationId).Distinct().Count().ShouldBe(1);
        rig.Dialer.Calls[0].Envelope.CorrelationId.ShouldNotBeNullOrWhiteSpace();
        rig.Dialer.Calls[0].Envelope.AlarmCommand.AlarmId.ShouldBe("alarm-1");
    }

    [Fact]
    public void The_contact_refresh_publishes_the_known_cluster_set_and_drops_cached_psks()
    {
        // Phase 1A security register item 1, wired: the gate is only a control if something keeps it
        // current, and the refresh is the moment central's picture of the fleet is up to date. The
        // InvalidateAll on the same tick is what makes a rotated key take effect within one refresh
        // instead of requiring a central restart.
        var factory = NewInMemoryDbFactory();
        SeedGrpcCluster(factory, "C1", ("host-a:4053", "host-a", 4056, true, false));
        SeedGrpcCluster(factory, "C2", ("host-b:4053", "host-b", null, true, false));
        var rig = SpawnGrpcRig(factory);

        AwaitContactsLoaded(rig);

        // C2 is included even though its only node has no GrpcPort: the gate is about which clusters
        // central KNOWS, not which it can currently dial. Excluding it would deny that cluster's acks.
        rig.Psk.KnownClusterPublications[0].ShouldBe(new[] { "C1", "C2" }, ignoreOrder: true);
        rig.Psk.InvalidateAllCount.ShouldBeGreaterThanOrEqualTo(1);
    }

    [Fact]
    public void A_slow_dial_does_not_park_the_actors_mailbox()
    {
        // The dialer's first act is a PSK lookup, which blocks on the secret store — bounded, but
        // seconds, and repeated after every refresh because the refresh drops the cache. An async
        // method runs synchronously to its first await, so dialing straight from the receive would
        // stall this actor for that long PER NODE, and deployment acks queue behind that mailbox.
        var factory = NewInMemoryDbFactory();
        SeedGrpcCluster(factory, "C1", ("host-a:4053", "host-a", 4056, true, false));
        var coordinator = CreateTestProbe();
        var dialer = new RecordingDialer { BlockBeforeFirstAwait = true };
        var rig = SpawnGrpcRig(factory, dialer, coordinator: coordinator.Ref);
        AwaitContactsLoaded(rig);

        try
        {
            rig.Actor.Tell(new MeshCommand(DeploymentsTopic, NewDispatch()));

            // The ack must be handled while the dial is still parked.
            var ack = new ApplyAck(
                new DeploymentId(Guid.NewGuid()),
                new NodeId("host-a:4053"),
                ApplyAckOutcome.Applied,
                null,
                new CorrelationId(Guid.NewGuid()));
            rig.Actor.Tell(ack);

            coordinator.ExpectMsg<ApplyAck>(TimeSpan.FromSeconds(5)).ShouldBe(ack);
        }
        finally
        {
            dialer.Release();
        }
    }

    [Fact]
    public void A_node_added_after_start_joins_the_fan_on_the_next_refresh()
    {
        // The refresh is not decoration: operators add and retire nodes, and central must pick them
        // up without a restart. Under Grpc that is the ONLY thing keeping the dispatch set current —
        // nothing on the node side announces itself; the DB row is the only source.
        //
        // Driven by telling the actor its own RefreshContacts tick rather than by shortening
        // ContactRefreshSeconds: a one-second timer would run for the whole fact, adding DB load to
        // every test sharing the runner, and would make this assertion depend on wall-clock timing.
        var factory = NewInMemoryDbFactory();
        SeedGrpcCluster(factory, "C1", ("host-a:4053", "host-a", 4056, true, false));
        var rig = SpawnGrpcRig(factory);
        AwaitContactsLoaded(rig);

        using (var db = factory.CreateDbContext())
        {
            db.ClusterNodes.Add(new ClusterNode
            {
                NodeId = "host-b:4053",
                ClusterId = "C1",
                Host = "host-b",
                AkkaPort = 4053,
                GrpcPort = 4056,
                ApplicationUri = "urn:host-b",
                Enabled = true,
                CreatedBy = "test",
            });
            db.SaveChanges();
        }

        rig.Actor.Tell(new CentralCommunicationActor.RefreshContacts());
        AwaitCondition(
            () => rig.Psk.KnownClusterPublications.Count >= 2, TimeSpan.FromSeconds(5));

        rig.Actor.Tell(new MeshCommand(DeploymentsTopic, NewDispatch()));

        AwaitCondition(() => rig.Dialer.Calls.Count == 2, TimeSpan.FromSeconds(5));
        rig.Dialer.Calls.Select(c => c.Endpoint).ShouldBe(
            new[] { "http://host-a:4056", "http://host-b:4056" }, ignoreOrder: true);

        // Once PER REFRESH, not once at startup. A key can rotate without any membership change, so
        // an invalidation that only fired when the cluster set moved would serve the pre-rotation
        // key forever and a rotation would need a central restart.
        rig.Psk.InvalidateAllCount.ShouldBeGreaterThanOrEqualTo(2);
    }

    [Theory]
    [InlineData(MeshTransportOptions.ModeDps)]
    [InlineData(MeshTransportOptions.ModeGrpc)]
    public void The_known_cluster_gate_is_published_in_every_transport_mode(string mode)
    {
        // Replaces an earlier fact asserting the OPPOSITE, whose premise did not survive Phase 1B: the
        // gate feeds CentralCommandAuthInterceptor, which is registered on every admin node, and the
        // ack service it protects is mapped by LISTENER PORT — neither consults MeshTransport:Mode.
        // Gating the publication on Grpc meant a rollback to Dps left that endpoint serving with a
        // permanently inactive gate, silently restoring both the timing oracle and the
        // pre-authentication store/audit flood. The set is central's cluster inventory, not
        // transport state.
        var factory = NewInMemoryDbFactory();
        SeedGrpcCluster(factory, "C1", ("host-a:4053", "host-a", 4056, true, false));
        var psk = new RecordingPskProvider();

        Sys.ActorOf(CentralCommunicationActor.Props(
            factory,
            new MeshTransportOptions { Mode = mode, ContactRefreshSeconds = 60 },
            () => null,
            () => new RecordingDialer(),
            () => psk));

        AwaitCondition(
            () => psk.KnownClusterPublications.Count >= 1, TimeSpan.FromSeconds(5));
        psk.KnownClusterPublications[0].ShouldBe(new[] { "C1" });
        psk.InvalidateAllCount.ShouldBeGreaterThanOrEqualTo(1);
    }

    [Fact]
    public void The_gate_set_includes_a_cluster_the_fan_set_excludes()
    {
        // The two sets answer different questions. The FAN asks "who does central talk to" — enabled,
        // non-maintenance, ported. The GATE asks "is this a cluster central has ever heard of", and
        // must say yes for a cluster whose every node is disabled or in maintenance: such a cluster
        // can still send a deployment ack, and refusing it would quietly turn maintenance mode into
        // an authentication switch.
        var factory = NewInMemoryDbFactory();
        SeedGrpcCluster(factory, "C1", ("host-a:4053", "host-a", 4056, true, false));
        SeedGrpcCluster(factory, "C-DISABLED", ("host-b:4053", "host-b", 4056, false, false));
        SeedGrpcCluster(factory, "C-MAINT", ("host-c:4053", "host-c", 4056, true, true));
        var rig = SpawnGrpcRig(factory);
        AwaitContactsLoaded(rig);

        rig.Psk.KnownClusterPublications[0].ShouldBe(
            new[] { "C1", "C-DISABLED", "C-MAINT" }, ignoreOrder: true);

        rig.Actor.Tell(new MeshCommand(DeploymentsTopic, NewDispatch()));

        AwaitCondition(() => rig.Dialer.Calls.Count == 1, TimeSpan.FromSeconds(5));
        ExpectNoMsg(TimeSpan.FromMilliseconds(300));
        rig.Dialer.Calls.Count.ShouldBe(1);
        rig.Dialer.Calls[0].ClusterId.ShouldBe("C1");
    }

    [Fact]
    public void The_gate_is_published_after_the_dispatch_set_is_adopted_and_before_the_cache_is_dropped()
    {
        // Two orderings in one handler, both load-bearing. SetKnownClusters must precede
        // InvalidateAll, or the first lookup after a refresh re-reads the store under the PREVIOUS
        // membership — briefly resolving a cluster central has just retired. And the dispatch set is
        // adopted first so a command processed immediately after the publication already fans over
        // the current targets.
        var factory = NewInMemoryDbFactory();
        SeedGrpcCluster(factory, "C1", ("host-a:4053", "host-a", 4056, true, false));
        var rig = SpawnGrpcRig(factory);
        AwaitContactsLoaded(rig);

        rig.Psk.Sequence.ShouldBe(new[] { "SetKnownClusters", "InvalidateAll" });

        // Targets-before-gate, observed the only way it can be from outside the actor: the very next
        // message finds the dispatch set already populated.
        rig.Actor.Tell(new MeshCommand(DeploymentsTopic, NewDispatch()));
        AwaitCondition(() => rig.Dialer.Calls.Count == 1, TimeSpan.FromSeconds(5));
    }

    [Fact]
    public void Grpc_mode_with_no_dialer_wired_drops_the_command_with_an_error()
    {
        // Mis-wiring, not misconfiguration: Mode=Grpc validates fine but the DI graph did not supply
        // a dialer. Silence here would be a fleet-wide command outage with no symptom until a deploy
        // stalled at its apply deadline two minutes later.
        var factory = NewInMemoryDbFactory();
        SeedGrpcCluster(factory, "C1", ("host-a:4053", "host-a", 4056, true, false));
        var actor = Sys.ActorOf(CentralCommunicationActor.Props(
            factory,
            GrpcMode(),
            () => null,
            dialer: null,
            pskProvider: null));

        // A never-supplied factory (dialer: null here) is silent at PreStart and on every retry —
        // see Resolve's remarks — so this dispatch-time Error is the ONLY log line the mis-wiring
        // produces. That is what makes the plain "no command dialer is wired" phrase safe to match
        // without a second source racing it, unlike before this actor stopped double-reporting the
        // same absence at two severities.
        EventFilter.Error(contains: "no command dialer is wired").ExpectOne(
            TimeSpan.FromSeconds(5),
            () => actor.Tell(new MeshCommand(DeploymentsTopic, NewDispatch())));
    }

    // ------------------------------------------------------------------ PreStart resolve-race logging
    //
    // PreStart's eager resolve of IMeshCommandDialer/IMeshPskProvider loses a millisecond-wide race
    // against the Akka builder's remaining starters on essentially every admin boot (live gates: 1A
    // 2/3 boots, 1B 4/4 boots) and self-heals at the memoized properties' first use. The two facts
    // below are the headline pin: a healthy boot must produce ZERO Warning/Error from this actor, and
    // a GENUINELY broken factory must still be reported loudly, exactly once.

    [Fact]
    public void A_healthy_boot_produces_no_warning_or_error_even_though_PreStart_loses_the_resolve_race()
    {
        // Models the ordinary boot: both factories throw on their FIRST invocation (PreStart, which
        // loses the race) and succeed on every subsequent one (the memoized properties' first-use
        // retry, long after every starter has finished). This must be silent — no Warning, no Error
        // — from PreStart all the way through the first contact load and the first dispatch, or a
        // soak error-grep on a healthy rig is impossible.
        var factory = NewInMemoryDbFactory();
        SeedGrpcCluster(factory, "C1", ("host-a:4053", "host-a", 4056, true, false));
        var dialer = new RecordingDialer();
        var psk = new RecordingPskProvider();
        var dialerCalls = 0;
        var pskCalls = 0;

        IMeshCommandDialer? DialerFactory()
        {
            dialerCalls++;
            return dialerCalls == 1 ? throw new InvalidOperationException("starter-ordering race") : dialer;
        }

        IMeshPskProvider? PskFactory()
        {
            pskCalls++;
            return pskCalls == 1 ? throw new InvalidOperationException("starter-ordering race") : psk;
        }

        IActorRef? actor = null;
        EventFilter.Warning().And.Error().Expect(0, TimeSpan.FromSeconds(5), () =>
        {
            actor = Sys.ActorOf(CentralCommunicationActor.Props(
                factory,
                GrpcMode(),
                () => null,
                DialerFactory,
                PskFactory));

            // First use of the PSK provider: the contact-refresh timer's initial (TimeSpan.Zero)
            // tick, observed via the gate publish.
            AwaitCondition(() => psk.KnownClusterPublications.Count >= 1, TimeSpan.FromSeconds(5));

            // First use of the dialer: an ordinary dispatch, retried until the contact load (which
            // races this on the same timer tick) has adopted the gRPC target set.
            AwaitCondition(() =>
            {
                actor!.Tell(new MeshCommand(DeploymentsTopic, NewDispatch()));
                return dialer.Calls.Count >= 1;
            }, TimeSpan.FromSeconds(5));
        });

        // Both retries actually happened — this fact is not vacuously true because the factories
        // never got a second chance to run.
        dialerCalls.ShouldBeGreaterThanOrEqualTo(2);
        pskCalls.ShouldBeGreaterThanOrEqualTo(2);
    }

    /// <summary>Counts contact loads and fails the first <c>failFirst</c> of them.</summary>
    private sealed class CountingDbFactory(
        IDbContextFactory<OtOpcUaConfigDbContext> inner, int failFirst = 0)
        : IDbContextFactory<OtOpcUaConfigDbContext>
    {
        private int _calls;

        public int Calls => Volatile.Read(ref _calls);

        public OtOpcUaConfigDbContext CreateDbContext()
        {
            if (Interlocked.Increment(ref _calls) <= failFirst)
                throw new InvalidOperationException("ConfigDb unreachable at boot");
            return inner.CreateDbContext();
        }

        public Task<OtOpcUaConfigDbContext> CreateDbContextAsync(CancellationToken ct = default) =>
            Task.FromResult(CreateDbContext());
    }

    private static MeshTransportOptions ModeOptions(string mode) => new()
    {
        Mode = mode,
        ContactRefreshSeconds = 60,
        DispatchTimeoutSeconds = 10,
    };

    [Theory]
    [InlineData(MeshTransportOptions.ModeGrpc)]
    [InlineData(MeshTransportOptions.ModeDps)]
    public void The_gate_publishes_within_seconds_when_the_provider_resolves_late_under_the_default_refresh(
        string mode)
    {
        // #654: the first RefreshContacts can land before the registry-bound PSK graph is resolvable.
        // The next PERIODIC tick is 60 s away; the bounded bootstrap retry must publish in seconds.
        var factory = NewInMemoryDbFactory();
        SeedGrpcCluster(factory, "C1", ("host-a:4053", "host-a", 4056, true, false));
        var psk = new RecordingPskProvider();
        var calls = 0;

        IMeshPskProvider? PskFactory() =>
            Interlocked.Increment(ref calls) <= 3 ? throw new InvalidOperationException("race") : psk;

        EventFilter.Error().Expect(0, TimeSpan.FromSeconds(1), () =>
        {
            Sys.ActorOf(CentralCommunicationActor.Props(
                factory, ModeOptions(mode), () => null, () => new RecordingDialer(), PskFactory));
            AwaitCondition(() => psk.KnownClusterPublications.Count >= 1, TimeSpan.FromSeconds(10));
        });
    }

    [Fact]
    public void A_failed_first_contact_load_retries_quickly_under_the_default_refresh()
    {
        var inner = NewInMemoryDbFactory();
        SeedGrpcCluster(inner, "C1", ("host-a:4053", "host-a", 4056, true, false));
        var factory = new CountingDbFactory(inner, failFirst: 1);
        var psk = new RecordingPskProvider();

        EventFilter.Warning().Expect(2, TimeSpan.FromSeconds(10), () =>
            Sys.ActorOf(CentralCommunicationActor.Props(
                factory, GrpcMode(), () => null, () => new RecordingDialer(), () => psk)));

        AwaitCondition(() => psk.KnownClusterPublications.Count >= 1, TimeSpan.FromSeconds(10));
    }

    [Fact]
    public void Bootstrap_retries_stop_once_the_gate_is_published()
    {
        // The provider throws on PreStart (call 1) and on the first load's resolve (call 2), which
        // schedules a 1 s retry. A manual refresh then publishes while that retry is still PENDING;
        // stop-on-publish must cancel it. Without the cancel the timer fires, loads again and
        // republishes (loads == 3, publications == 2).
        var inner = NewInMemoryDbFactory();
        SeedGrpcCluster(inner, "C1", ("host-a:4053", "host-a", 4056, true, false));
        var factory = new CountingDbFactory(inner);
        var psk = new RecordingPskProvider();
        var calls = 0;

        IMeshPskProvider? PskFactory() =>
            Interlocked.Increment(ref calls) <= 2 ? throw new InvalidOperationException("race") : psk;

        var actor = Sys.ActorOf(CentralCommunicationActor.Props(
            factory, GrpcMode(), () => null, () => new RecordingDialer(), PskFactory,
            bootstrapRetryInterval: TimeSpan.FromSeconds(1)));

        // Positive control: the failed first resolve happened, so a retry is pending.
        AwaitCondition(() => Volatile.Read(ref calls) >= 2, TimeSpan.FromSeconds(5));
        Thread.Sleep(100);
        actor.Tell(new CentralCommunicationActor.RefreshContacts());
        AwaitCondition(() => psk.KnownClusterPublications.Count >= 1, TimeSpan.FromSeconds(5));

        Thread.Sleep(2500);

        factory.Calls.ShouldBe(2);
        psk.KnownClusterPublications.Count.ShouldBe(1);
    }

    [Fact]
    public void A_permanently_failing_load_retry_budget_is_bounded()
    {
        var inner = NewInMemoryDbFactory();
        var factory = new CountingDbFactory(inner, failFirst: int.MaxValue);
        Sys.ActorOf(CentralCommunicationActor.Props(
            factory, GrpcMode(), () => null, () => new RecordingDialer(), () => new RecordingPskProvider(),
            bootstrapRetryInterval: TimeSpan.FromMilliseconds(20)));

        EventFilter.Warning().Expect(32, TimeSpan.FromSeconds(10), () => { });
        Thread.Sleep(1500);
        // 1 initial + 15 retries; the 60 s periodic tick has not fired again.
        factory.Calls.ShouldBe(16);
    }

    [Fact]
    public void A_permanently_unresolvable_pskProvider_is_reported_once_per_incarnation_not_once_per_refresh()
    {
        // The counterpart pin: a factory that NEVER succeeds is the genuinely permanent absence and
        // must still be loud — but PublishKnownClusters runs on the contact-refresh timer for the
        // actor's whole life, so an unguarded Error would repeat every ContactRefreshSeconds forever.
        // Guarded exactly like _warnedPortlessNodes: once per incarnation, not once per refresh.
        //
        // The factory counts its own invocations rather than relying on a dialer/dispatch side
        // effect: LoadContactsFromDb is fire-and-forget, so a signal that isn't the resolve itself
        // (e.g. a dispatched command landing) can close an AwaitCondition window before the SECOND
        // contact load has actually re-invoked PublishKnownClusters, making the "does not repeat"
        // half true for the wrong reason.
        var factory = NewInMemoryDbFactory();
        SeedGrpcCluster(factory, "C1", ("host-a:4053", "host-a", 4056, true, false));
        var pskCalls = 0;

        IMeshPskProvider? PskFactory()
        {
            pskCalls++;
            throw new InvalidOperationException("permanently broken");
        }

        IActorRef actor = null!;
        EventFilter.Error(contains: "will NEVER be published").ExpectOne(TimeSpan.FromSeconds(10), () =>
        {
            actor = Sys.ActorOf(CentralCommunicationActor.Props(
                factory,
                GrpcMode(),
                () => null,
                () => new RecordingDialer(),
                pskProvider: PskFactory,
                bootstrapRetryInterval: TimeSpan.FromMilliseconds(50)));

            // pskCalls == 1 is PreStart's own attempt; reaching 2 is PROOF the first contact load
            // ran and PublishKnownClusters's property access performed the first-use retry — not
            // merely that the actor was constructed.
            AwaitCondition(() => pskCalls >= 2, TimeSpan.FromSeconds(5));
        });

        // The bootstrap retry is BOUNDED: with a 50 ms interval and a budget of 15 it has long
        // finished, so the call count plateaus instead of growing with elapsed time.
        Thread.Sleep(500);
        var plateau = pskCalls;
        Thread.Sleep(500);
        pskCalls.ShouldBe(plateau);

        var pskCallsAfterFirstLoad = pskCalls;

        // A SECOND contact load must NOT repeat the Error. The pin only means something if that
        // second load actually happened and retried the resolve again — AwaitCondition on a
        // STRICTLY higher pskCalls count, taken INSIDE the very block whose zero-Error assertion
        // depends on it, is what proves that rather than assuming it from elapsed time.
        EventFilter.Error(contains: "will NEVER be published").Expect(0, TimeSpan.FromSeconds(5), () =>
        {
            actor.Tell(new CentralCommunicationActor.RefreshContacts());
            AwaitCondition(() => pskCalls > pskCallsAfterFirstLoad, TimeSpan.FromSeconds(5));
        });
    }

    [Fact]
    public void A_supplied_dialer_factory_that_always_throws_still_drops_each_command_loudly_while_PreStart_stays_quiet()
    {
        // The symmetric case to Grpc_mode_with_no_dialer_wired_drops_the_command_with_an_error
        // above: there the factory was never supplied (Resolve returns null with no exception,
        // silent at every call site); here a factory WAS supplied and never succeeds, so it goes
        // through Resolve's catch on both the PreStart attempt (firstUse: false, Debug) and every
        // first-use retry (firstUse: true, still Debug — see Resolve's remarks on why severity is
        // decided by the consumer, not by Resolve). The loud signal is DispatchOverGrpc's existing
        // per-dispatch Error, unaffected by an exception being the reason the dialer is null rather
        // than a bare absence, and unguarded — unlike the PSK provider, every dispatch reports it.
        var factory = NewInMemoryDbFactory();
        SeedGrpcCluster(factory, "C1", ("host-a:4053", "host-a", 4056, true, false));
        var psk = new RecordingPskProvider();

        IActorRef actor = null!;
        EventFilter.Warning().And.Error().Expect(0, TimeSpan.FromSeconds(5), () =>
        {
            actor = Sys.ActorOf(CentralCommunicationActor.Props(
                factory,
                GrpcMode(),
                () => null,
                dialer: () => throw new InvalidOperationException("permanently broken"),
                pskProvider: () => psk));

            // Proof PreStart's resolve attempt and the first contact load it schedules have
            // actually run, so this block's silence is not vacuous.
            AwaitCondition(() => psk.KnownClusterPublications.Count >= 1, TimeSpan.FromSeconds(5));
        });

        EventFilter.Error(contains: "no command dialer is wired").ExpectOne(TimeSpan.FromSeconds(5), () =>
            actor.Tell(new MeshCommand(DeploymentsTopic, NewDispatch())));

        // A SECOND dispatch must ALSO log it: this signal is per-dispatch and unguarded, the
        // opposite shape from the PSK provider's once-per-incarnation guard above.
        EventFilter.Error(contains: "no command dialer is wired").ExpectOne(TimeSpan.FromSeconds(5), () =>
            actor.Tell(new MeshCommand(DeploymentsTopic, NewDispatch())));
    }

    private static void SetGrpcPort(
        IDbContextFactory<OtOpcUaConfigDbContext> factory, string nodeId, int? grpcPort)
    {
        using var db = factory.CreateDbContext();
        var node = db.ClusterNodes.Single(n => n.NodeId == nodeId);
        node.GrpcPort = grpcPort;
        db.SaveChanges();
    }

    private static DispatchDeployment NewDispatch() => new(
        new DeploymentId(Guid.NewGuid()), new RevisionHash("abc"), new CorrelationId(Guid.NewGuid()));

    /// <summary>A spawned gRPC-mode actor plus the two seams the tests observe it through.</summary>
    private sealed record GrpcRig(IActorRef Actor, RecordingDialer Dialer, RecordingPskProvider Psk);

    private GrpcRig SpawnGrpcRig(
        IDbContextFactory<OtOpcUaConfigDbContext> factory,
        RecordingDialer? dialer = null,
        IActorRef? coordinator = null)
    {
        dialer ??= new RecordingDialer();
        var psk = new RecordingPskProvider();
        var actor = Sys.ActorOf(CentralCommunicationActor.Props(
            factory,
            GrpcMode(),
            () => coordinator,
            () => dialer,
            () => psk));
        return new GrpcRig(actor, dialer, psk);
    }

    /// <summary>
    ///     Waits until the actor has applied a contact load. The known-cluster publication is the
    ///     signal because it happens in the same handler, immediately after the target set is
    ///     adopted — and unlike the target set itself it is observable from outside the actor. A
    ///     command told before this point would be dropped as "no targets" and never retried, which
    ///     is correct behaviour and a flaky test.
    /// </summary>
    private void AwaitContactsLoaded(GrpcRig rig) =>
        AwaitCondition(
            () => rig.Psk.KnownClusterPublications.Count >= 1, TimeSpan.FromSeconds(5));

    /// <summary>Records every dial and answers however the test asks; never opens a socket.</summary>
    private sealed class RecordingDialer : IMeshCommandDialer
    {
        private readonly object _gate = new();
        private readonly List<Dial> _calls = [];
        private readonly ManualResetEventSlim _released = new(false);

        /// <summary>
        ///     When true, every dial blocks its calling thread synchronously until
        ///     <see cref="Release"/>. Models the dialer's pre-await PSK lookup.
        /// </summary>
        public bool BlockBeforeFirstAwait { get; set; }

        /// <summary>Unblocks every dial parked by <see cref="BlockBeforeFirstAwait"/>.</summary>
        public void Release() => _released.Set();

        /// <summary>Every dial made. Copied under the lock — the fan runs on thread-pool threads.</summary>
        public IReadOnlyList<Dial> Calls
        {
            get { lock (_gate) return _calls.ToList(); }
        }

        /// <summary>Endpoints whose substring match makes the dial fault.</summary>
        public string? FailEndpointsContaining { get; set; }

        /// <summary>Cluster ids whose dials fault.</summary>
        public string[] FailClusters { get; set; } = [];

        /// <summary>
        ///     The fault raised for a matching dial. Fully qualified because this file's bare
        ///     <c>Status</c> is Akka's — an existing fact drives the actor's <c>Status.Failure</c>
        ///     handler, and importing Grpc.Core here would make that reference ambiguous.
        /// </summary>
        public Grpc.Core.RpcException? Fault { get; set; }

        public Task DispatchAsync(
            string nodeEndpoint, CommandEnvelope envelope, string clusterId, CancellationToken ct)
        {
            lock (_gate) _calls.Add(new Dial(nodeEndpoint, envelope, clusterId));

            if (BlockBeforeFirstAwait)
            {
                // Blocks the CALLING thread, exactly as IMeshPskProvider.TryGetPsk does on a cold
                // cache: an async method runs synchronously to its first await, so a dial invoked
                // straight from the actor would park the mailbox here.
                _released.Wait(TimeSpan.FromSeconds(30));
            }

            var shouldFail =
                (FailEndpointsContaining is not null && nodeEndpoint.Contains(FailEndpointsContaining))
                || FailClusters.Contains(clusterId, StringComparer.OrdinalIgnoreCase);

            return shouldFail && Fault is not null
                ? Task.FromException(Fault)
                : Task.CompletedTask;
        }

        /// <summary>One recorded dial.</summary>
        public sealed record Dial(string Endpoint, CommandEnvelope Envelope, string ClusterId);
    }

    /// <summary>Records the gate and invalidation calls the contact refresh is supposed to make.</summary>
    private sealed class RecordingPskProvider : IMeshPskProvider
    {
        private readonly object _gate = new();
        private readonly List<IReadOnlyList<string>> _publications = [];
        private readonly List<string> _sequence = [];
        private int _invalidateAllCount;

        /// <summary>Names of the mutating calls made, in the order they arrived.</summary>
        public IReadOnlyList<string> Sequence
        {
            get { lock (_gate) return _sequence.ToList(); }
        }

        /// <summary>Every known-cluster set published, in order.</summary>
        public IReadOnlyList<IReadOnlyList<string>> KnownClusterPublications
        {
            get { lock (_gate) return _publications.ToList(); }
        }

        /// <summary>How many times the whole cache was dropped.</summary>
        public int InvalidateAllCount => Volatile.Read(ref _invalidateAllCount);

        public bool TryGetPsk(string clusterId, [NotNullWhen(true)] out string? psk)
        {
            psk = "psk";
            return true;
        }

        public bool Matches(string clusterId, string presented) => true;

        public void SetKnownClusters(IReadOnlyCollection<string> clusterIds)
        {
            lock (_gate)
            {
                _publications.Add(clusterIds.ToList());
                _sequence.Add(nameof(SetKnownClusters));
            }
        }

        public void Invalidate(string clusterId) { }

        public void InvalidateAll()
        {
            Interlocked.Increment(ref _invalidateAllCount);
            lock (_gate) _sequence.Add(nameof(InvalidateAll));
        }
    }
}