lmxopcua/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Communication/CentralCommunicationActor.cs

using System.Collections.Immutable;
using Akka.Actor;
using Akka.Cluster.Tools.Client;
using Akka.Cluster.Tools.PublishSubscribe;
using Akka.Event;
using Grpc.Core;
using Microsoft.EntityFrameworkCore;
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.Protos.MeshTransport.V1;
using ZB.MOM.WW.OtOpcUa.Commons.Types;
using ZB.MOM.WW.OtOpcUa.Configuration;
// Grpc.Core (imported for RpcException on the dial path) declares a Status of its own. This file's
// bare Status is Akka's — the Status.Failure a faulted piped task arrives as; the gRPC one is only
// ever reached through an RpcException instance.
using Status = Akka.Actor.Status;

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

/// <summary>
///     Central-side end of the central↔node command boundary (per-cluster mesh Phase 2). Routes
///     every central→node command out — over DistributedPubSub, ClusterClient or gRPC, per
///     <see cref="MeshTransportOptions.Mode"/> — and forwards inbound <see cref="ApplyAck"/>s to the
///     deploy coordinator singleton.
/// </summary>
/// <remarks>
///     <para>
///         <b>Registered per node, NOT as a cluster singleton.</b> A driver node's ClusterClient
///         rotates across contact points and must find a live comm actor at whichever central node
///         answers. A singleton would be reachable only through the one node hosting it, and the
///         rotation would fail silently against the other.
///     </para>
///     <para>
///         <b>One ClusterClient per application <c>Cluster</c> (Phase 6).</b> The fleet is now one
///         Akka mesh per cluster, so a single fleet-wide client's <c>SendToAll</c> would reach only
///         the one mesh whose receptionist answered. This actor therefore holds a client per cluster
///         and fans <c>SendToAll</c> across all of them. See <see cref="RebuildClient"/> — this is the
///         single most important thing to understand before changing this actor.
///     </para>
///     <para>
///         <b>The gRPC branch fans over the same recipient set, node by node</b>
///         (ClusterClient→gRPC Phase 1A). <c>SendToAll</c> reaches every node registered on a mesh's
///         receptionist, and the contact load already selects exactly the enabled, non-maintenance
///         rows those nodes come from; the gRPC branch dials that same row set directly, one call per
///         node, so the recipient set is identical by construction rather than by coincidence. It is
///         deliberately NOT narrower: the node side has no <c>ClusterId</c> or node filter at all —
///         scoping happens later, inside the artifact — so a node left out here silently keeps its
///         old configuration while the deployment seals green.
///     </para>
///     <para>
///         <b>Fire-and-forget, on every transport.</b> The gRPC branch pipes per-node results back to
///         itself for logging ONLY: a failure names the node and the status and is dropped. No retry,
///         no queue, no ordering guarantee across nodes. That is not an omission — it is the
///         ClusterClient semantic preserved exactly, and the backstops are unchanged (the deploy
///         coordinator's apply deadline, and a node's boot-time replay).
///     </para>
/// </remarks>
public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers
{
    private const string RefreshTimerKey = "mesh-contact-refresh";

    private readonly IDbContextFactory<OtOpcUaConfigDbContext> _dbFactory;
    private readonly MeshTransportOptions _options;
    private readonly IMeshClusterClientFactory _clientFactory;
    private readonly Func<IActorRef?> _coordinator;

    // Resolved in PreStart, never at construction. See the Props doc: the DI graph behind these two
    // reaches actor-registry entries that do not exist yet while the Akka builder's starters are
    // still running, and touching it there is what stopped every central node from booting.
    private readonly Func<IMeshCommandDialer?>? _dialerFactory;
    private readonly Func<IMeshPskProvider?>? _pskProviderFactory;
    private IMeshCommandDialer? _dialer;
    private IMeshPskProvider? _pskProvider;

    private readonly ILoggingAdapter _log = Context.GetLogger();
    private readonly bool _useClusterClient;
    private readonly bool _useGrpc;

    // One client per application Cluster, keyed by ClusterId. A ClusterClient reaches only the mesh
    // whose receptionist it dialled, and after Phase 6 every cluster is its own mesh — so central must
    // hold a client per cluster and fan SendToAll across all of them.
    private readonly Dictionary<string, (IActorRef Client, ImmutableHashSet<string> Contacts)> _clients =
        new();

    // Node ids already warned about for a null GrpcPort. A persistently portless row would otherwise
    // produce one warning per refresh forever; same warn-once shape as TelemetryNodeSource. Actor
    // thread only.
    private readonly HashSet<string> _warnedPortlessNodes = new(StringComparer.Ordinal);

    // The gRPC dispatch set, replaced wholesale on every contact load. Actor thread only; the fan
    // captures the list reference, so a refresh mid-fan cannot mutate a dispatch already under way.
    private IReadOnlyList<MeshNodeTarget> _targets = [];

    private CancellationTokenSource _lifecycle = new();

    /// <summary>Gets the timer scheduler driving the periodic contact refresh.</summary>
    public ITimerScheduler Timers { get; set; } = null!;

    /// <summary>Creates the props for this actor.</summary>
    /// <param name="dbFactory">Factory for the config database holding <c>ClusterNode</c> rows.</param>
    /// <param name="options">The bound mesh-transport options.</param>
    /// <param name="clientFactory">Creates the ClusterClient; substituted in tests.</param>
    /// <param name="coordinator">
    ///     Resolves the deploy-coordinator singleton proxy. A <see cref="Func{TResult}"/> rather than
    ///     an <see cref="IActorRef"/> so this actor never depends on registration order at wiring
    ///     time — it resolves on first use and caches.
    /// </param>
    /// <param name="dialer">
    ///     Resolves the thing that places one command call on one node, under
    ///     <see cref="MeshTransportOptions.ModeGrpc"/>. Optional so the two older transports need
    ///     nothing new wired; a <see langword="null"/> dialer under <c>Grpc</c> is a mis-wiring and
    ///     is reported loudly rather than silently dropping every command.
    /// </param>
    /// <param name="pskProvider">
    ///     Resolves central's per-cluster key lookup, told on every contact refresh which clusters
    ///     exist (the known-cluster gate) and to drop its cache (so a rotated key takes effect within
    ///     one refresh). Optional for the same reason as <paramref name="dialer"/>.
    /// </param>
    /// <returns>The props.</returns>
    /// <remarks>
    ///     <para>
    ///         <b>Both dependencies arrive as <see cref="Func{TResult}"/>, and that is not a style
    ///         choice.</b> Passing instances meant the caller resolved them while the Akka builder's
    ///         starters were still running, and the DI graph behind them reaches back into the actor
    ///         registry: dialer → <c>IMeshPskProvider</c> → <c>ISecretResolver</c> → <c>AddZbSecrets</c>
    ///         → <c>IAuditWriter</c> → <c>ActorAuditWriter</c>, whose constructor calls
    ///         <c>registry.Get&lt;AuditWriterActorKey&gt;()</c> — a singleton registered dozens of
    ///         lines LATER in the very same builder. Every central node failed to boot, in every
    ///         transport mode, on a <c>MissingActorRegistryEntryException</c>. Deferring to
    ///         <see cref="PreStart"/> moves the first resolve to a point where the host is fully
    ///         built, and matches the <see cref="Func{TResult}"/> the coordinator proxy already uses
    ///         one argument earlier for the same class of reason.
    ///     </para>
    ///     <para>
    ///         This is the third instance of the same failure mode in this family — eager resolution
    ///         inside an <c>AddAkka</c> configurator, previously <c>scadaproj#1</c> (a DI cycle that
    ///         deadlocked) and mesh Phase 6's <c>IClusterRoleInfo</c> boot <c>StackOverflow</c>. The
    ///         rule it keeps re-teaching: inside an Akka builder, resolve nothing you can defer.
    ///     </para>
    /// </remarks>
    public static Props Props(
        IDbContextFactory<OtOpcUaConfigDbContext> dbFactory,
        MeshTransportOptions options,
        IMeshClusterClientFactory clientFactory,
        Func<IActorRef?> coordinator,
        Func<IMeshCommandDialer?>? dialer = null,
        Func<IMeshPskProvider?>? pskProvider = null) =>
        Akka.Actor.Props.Create(() =>
            new CentralCommunicationActor(
                dbFactory, options, clientFactory, coordinator, dialer, pskProvider));

    /// <summary>Initializes a new instance of the <see cref="CentralCommunicationActor"/> class.</summary>
    /// <param name="dbFactory">Factory for the config database.</param>
    /// <param name="options">The bound mesh-transport options.</param>
    /// <param name="clientFactory">Creates the ClusterClient.</param>
    /// <param name="coordinator">Lazily resolves the deploy-coordinator singleton proxy.</param>
    /// <param name="dialer">Resolves the gRPC command dialer; invoked in <see cref="PreStart"/>.</param>
    /// <param name="pskProvider">Resolves the PSK provider; invoked in <see cref="PreStart"/>.</param>
    public CentralCommunicationActor(
        IDbContextFactory<OtOpcUaConfigDbContext> dbFactory,
        MeshTransportOptions options,
        IMeshClusterClientFactory clientFactory,
        Func<IActorRef?> coordinator,
        Func<IMeshCommandDialer?>? dialer = null,
        Func<IMeshPskProvider?>? pskProvider = null)
    {
        _dbFactory = dbFactory;
        _options = options;
        _clientFactory = clientFactory;
        _coordinator = coordinator;
        _dialerFactory = dialer;
        _pskProviderFactory = pskProvider;
        _useClusterClient = string.Equals(
            options.Mode, MeshTransportOptions.ModeClusterClient, StringComparison.OrdinalIgnoreCase);
        _useGrpc = string.Equals(
            options.Mode, MeshTransportOptions.ModeGrpc, StringComparison.OrdinalIgnoreCase);

        Receive<MeshCommand>(HandleMeshCommand);
        Receive<ApplyAck>(HandleApplyAck);
        Receive<RefreshContacts>(_ => LoadContactsFromDb());
        Receive<ContactsLoaded>(HandleContactsLoaded);
        Receive<CommandDispatched>(HandleCommandDispatched);

        // A faulted LoadContactsFromDb task is piped here as a Status.Failure. Without this handler
        // the failure is an unhandled message (debug level only) and the refresh fails silently —
        // an operator cannot distinguish "no nodes configured" from "the database is down", and the
        // contact set silently freezes at whatever it last held.
        //
        // The freeze extends to the known-cluster gate, deliberately: a failed load must neither
        // widen the gate nor narrow it, and the previous set is the last thing central actually knew.
        // The second line exists because that freeze is INVISIBLE otherwise, and it is fail-OPEN
        // before the first successful load — a central that has never reached its ConfigDb refuses
        // no cluster at all.
        Receive<Status.Failure>(f =>
        {
            _log.Warning(
                f.Cause,
                "Failed to load ClusterNode contact points; the mesh contact set was NOT refreshed "
                + "and may be stale or empty. Commands may be dropped until the next refresh");
            _log.Warning(
                "The known-cluster gate was NOT refreshed either: it still holds the set from the "
                + "last successful load, and if none has succeeded yet (boot, or central SQL Server "
                + "unreachable since start) it is INACTIVE — every cluster id reaches the secret "
                + "store, which is the pre-authentication cost the gate exists to remove");
        });

        Receive<SubscribeAck>(_ => { /* DPS subscribe confirmation */ });
    }

    /// <inheritdoc />
    protected override void PreStart()
    {
        // First point at which the host is fully built — see the Props remark. Re-runs on every
        // actor restart, which re-reads the container rather than caching a stale reference; both
        // dependencies are container singletons, so a restart gets the same instances back.
        _dialer = Resolve(_dialerFactory, nameof(IMeshCommandDialer));
        _pskProvider = Resolve(_pskProviderFactory, nameof(IMeshPskProvider));

        if (!_useClusterClient && !_useGrpc)
        {
            _log.Info(
                "Mesh transport is {Mode}: commands are published on DistributedPubSub and no "
                + "ClusterClient is created. The contact refresh still runs — it is what keeps the "
                + "known-cluster gate current, and that gate belongs to the inbound ack endpoint, "
                + "which is mapped by port and does not care about this mode", _options.Mode);
        }

        if (_useGrpc && _dialer is null)
        {
            // Not a fatal throw: the ack path (HandleApplyAck) is transport-independent and still
            // works, so taking the actor down would turn a command-plane mis-wiring into an ack
            // outage as well. Every dispatch logs its own error, so this cannot go unnoticed.
            _log.Error(
                "Mesh transport is Grpc but no command dialer was supplied to this actor; EVERY "
                + "central→node command will be dropped. Check the ControlPlane registration wires "
                + "IMeshCommandDialer");
        }

        if (_pskProvider is null)
        {
            // Mode-independent, like the gate itself. A silent absence here is the worst of the
            // three outcomes in this method: nothing malfunctions, so nothing is ever investigated,
            // while the gate stays permanently inactive and a rotated key is never picked up.
            _log.Error(
                "No IMeshPskProvider was supplied to this actor; the known-cluster gate will NEVER "
                + "be published and cached per-cluster keys will NEVER be dropped. Inbound mesh acks "
                + "stay exposed to the pre-authentication secret-store cost the gate removes, and a "
                + "rotated key needs a central restart. Check the ControlPlane registration");
        }

        // Started in EVERY mode. The refresh is not transport state: it is central's ClusterNode
        // inventory, and the known-cluster gate it feeds protects an endpoint mapped by PORT, not by
        // MeshTransport:Mode. Gating the timer on the mode would mean a dark-switch rollback to
        // ClusterClient or Dps left that endpoint live with the gate permanently inactive — silently
        // reopening both the timing oracle and the pre-auth store/audit flood. Under Dps this is a
        // ConfigDb query every ContactRefreshSeconds that did not run before; central always has a
        // ConfigDb, and one indexed read a minute is the cheaper half of the trade.
        Timers.StartPeriodicTimer(
            RefreshTimerKey,
            new RefreshContacts(),
            TimeSpan.Zero,
            TimeSpan.FromSeconds(_options.ContactRefreshSeconds));
    }

    /// <summary>
    ///     Invokes a dependency factory, converting a resolution failure into a logged absence.
    /// </summary>
    /// <typeparam name="T">The dependency's type.</typeparam>
    /// <param name="factory">The factory, or <see langword="null"/> when nothing was supplied.</param>
    /// <param name="name">The dependency's name, for the log.</param>
    /// <returns>The resolved dependency, or <see langword="null"/>.</returns>
    /// <remarks>
    ///     Catching is the point. A throw out of <see cref="PreStart"/> is an actor failure, and the
    ///     supervisor's answer to that is to restart — which calls <see cref="PreStart"/> again, and
    ///     a broken DI graph is not a transient condition. That is an unbounded restart loop on the
    ///     node's command boundary, taking the transport-independent ack path down with it. Logging
    ///     and continuing leaves the ack path serving and hands the two null guards below a genuine
    ///     absence to report, which is the same outcome as never having been wired.
    /// </remarks>
    private T? Resolve<T>(Func<T?>? factory, string name)
        where T : class
    {
        if (factory is null)
            return null;

        try
        {
            return factory();
        }
        catch (Exception ex)
        {
            _log.Error(
                ex,
                "Failed to resolve {Dependency} for the mesh command boundary; continuing without "
                + "it. The guards below report what that costs", name);
            return null;
        }
    }

    /// <inheritdoc />
    protected override void PostStop()
    {
        foreach (var (client, _) in _clients.Values) Context.Stop(client);
        _clients.Clear();

        // Cancelling the lifecycle token also cancels every gRPC dial in flight. The dialer itself is
        // deliberately NOT disposed here: it is a container-owned singleton shared with anything else
        // that may dial, and an actor restart must not tear its channel pool down.
        _lifecycle.Cancel();
        _lifecycle.Dispose();
    }

    private void HandleMeshCommand(MeshCommand cmd)
    {
        // Checked before the DPS arm: _useClusterClient is false under Grpc too, so the order is
        // what keeps a Grpc-mode command off DistributedPubSub.
        if (_useGrpc)
        {
            DispatchOverGrpc(cmd);
            return;
        }

        if (!_useClusterClient)
        {
            DistributedPubSub.Get(Context.System).Mediator.Tell(new Publish(cmd.Topic, cmd.Message));
            return;
        }

        if (_clients.Count == 0)
        {
            // App-level drop, matching the sister project's decision and this repo's buffer-size = 0.
            // The command is NOT queued and will NOT be retried: a deploy fails at the coordinator's
            // apply deadline naming the silent nodes, which is a better operator signal than a
            // command arriving minutes late against a config the DB already recorded as failed.
            _log.Warning(
                "No mesh ClusterClient — dropping {MessageType}. The last contact refresh produced "
                + "no usable contact point from the enabled, non-maintenance ClusterNode rows in any "
                + "cluster; the command is not buffered and will not be retried",
                cmd.Message.GetType().Name);
            return;
        }

        // SendToAll, NEVER Send, fanned across EVERY cluster's client. Today's DPS publish reaches
        // EVERY DriverHostActor — there is no ClusterId or node filter on the node side at all; scoping
        // happens later, inside the artifact (DeploymentArtifact.ResolveClusterScope). Each ClusterClient
        // reaches only the one mesh it dialled, so central must fan across all of them to preserve the
        // fleet-wide reach DPS gives. ClusterClient.Send delivers to exactly ONE registered actor, which
        // would deploy to a single node while every other node silently kept its old configuration — and
        // the deployment would still seal green if the ack set happened to be satisfied.
        foreach (var (client, _) in _clients.Values)
            client.Tell(new ClusterClient.SendToAll(MeshPaths.NodeCommunication, cmd.Message));
    }

    /// <summary>
    ///     Encodes one command once and fans it to every target node, one unary gRPC call each.
    /// </summary>
    /// <param name="cmd">The command envelope carrying the domain record.</param>
    /// <remarks>
    ///     <para>
    ///         <b>Encoded once, before the fan.</b> Not per node — partly because the wire form is
    ///         identical for every recipient, but mostly because <c>AlarmCommand</c> has no
    ///         correlation member of its own and the envelope mints one: minting inside the loop
    ///         would give one operator action N unrelated correlation ids and make it untraceable.
    ///     </para>
    ///     <para>
    ///         <b>Nothing captured from the actor.</b> Each dial is a static call taking only locals,
    ///         so no per-node continuation can touch actor state off the actor's thread. Results come
    ///         back as ordinary messages via <c>PipeTo</c> and are handled on the actor thread like
    ///         anything else.
    ///     </para>
    ///     <para>
    ///         <b>Every dial starts on the thread pool, not here.</b> An <c>async</c> method runs
    ///         synchronously up to its first <c>await</c>, and the dialer's pre-await work includes
    ///         <c>IMeshPskProvider.TryGetPsk</c>, which blocks on the secret store (bounded, but
    ///         seconds) the first time a cluster is looked up — and again after every refresh, since
    ///         the refresh drops the cache to pick up rotations. Calling straight into it would park
    ///         this actor's mailbox, and deployment acks queue behind that mailbox.
    ///     </para>
    /// </remarks>
    private void DispatchOverGrpc(MeshCommand cmd)
    {
        var commandType = cmd.Message.GetType().Name;

        var dialer = _dialer;
        if (dialer is null)
        {
            _log.Error(
                "Mesh transport is Grpc but no command dialer is wired — dropping {MessageType}. The "
                + "command is not buffered and will not be retried",
                commandType);
            return;
        }

        // Captured before the fan: a contact refresh between here and the last dial must not change
        // the recipient set of a dispatch already under way.
        var targets = _targets;
        if (targets.Count == 0)
        {
            // App-level drop, exactly as the ClusterClient arm does with no client. The command is
            // NOT queued and will NOT be retried: a deploy fails at the coordinator's apply deadline
            // naming the silent nodes, which is a better operator signal than a command arriving
            // minutes late against a config the DB already recorded as failed.
            _log.Warning(
                "No mesh gRPC targets — dropping {MessageType}. The last contact refresh produced no "
                + "enabled, non-maintenance ClusterNode row carrying a GrpcPort in any cluster; the "
                + "command is not buffered and will not be retried",
                commandType);
            return;
        }

        CommandEnvelope envelope;
        try
        {
            envelope = MeshTransportDtoMapper.ToEnvelope(cmd.Message, new CorrelationId(Guid.NewGuid()));
        }
        catch (Exception ex)
        {
            // The mapper refuses a payload the far side could not decode. Dropping here is the same
            // outcome the old transport gave for such a message (nothing on the node matched it), but
            // loud instead of silent.
            _log.Error(
                ex,
                "Cannot encode {MessageType} for the gRPC command plane; the command is dropped and "
                + "was NOT delivered to any node",
                commandType);
            return;
        }

        CancellationToken ct;
        try
        {
            ct = _lifecycle.Token;
        }
        catch (ObjectDisposedException)
        {
            return; // Stopping.
        }

        var self = Self;
        var correlationId = envelope.CorrelationId;
        foreach (var target in targets)
        {
            var node = target;

            // CancellationToken.None on Task.Run, deliberately: cancelling the SCHEDULING would make
            // the task complete as canceled, which PipeTo converts into a Status.Failure — the one
            // message this actor already reads as "the contact load failed". The lifecycle token is
            // passed into the dial instead, where it cancels the call and comes back as an ordinary
            // per-node outcome.
            Task.Run(
                    () => DispatchOneAsync(dialer, node, envelope, commandType, correlationId, ct),
                    CancellationToken.None)
                .PipeTo(self);
        }
    }

    /// <summary>
    ///     Dials one node and turns every possible outcome into a <see cref="CommandDispatched"/>.
    /// </summary>
    /// <param name="dialer">The dialer seam.</param>
    /// <param name="target">The node to dial.</param>
    /// <param name="envelope">The already-encoded command.</param>
    /// <param name="commandType">CLR type name of the command, for the log.</param>
    /// <param name="correlationId">Transport correlation, for the log.</param>
    /// <param name="ct">The actor's lifecycle token.</param>
    /// <returns>The outcome, always — this method does not fault.</returns>
    /// <remarks>
    ///     Catching everything is what keeps the result on ONE channel. A faulting task piped to an
    ///     actor arrives as <c>Status.Failure</c>, which this actor already handles as "the contact
    ///     load failed" — so an unhandled dial fault would log a database-outage warning about a node
    ///     that simply did not answer.
    /// </remarks>
    private static async Task<CommandDispatched> DispatchOneAsync(
        IMeshCommandDialer dialer,
        MeshNodeTarget target,
        CommandEnvelope envelope,
        string commandType,
        string correlationId,
        CancellationToken ct)
    {
        try
        {
            await dialer.DispatchAsync(target.Endpoint, envelope, target.ClusterId, ct)
                .ConfigureAwait(false);
            return new CommandDispatched(target.NodeId, commandType, correlationId, null, null);
        }
        catch (RpcException ex)
        {
            return new CommandDispatched(
                target.NodeId,
                commandType,
                correlationId,
                ex.StatusCode.ToString(),
                Truncate(ex.Status.Detail));
        }
        catch (OperationCanceledException)
        {
            // Named rather than swept into the generic arm: this is the ordinary shape of a node that
            // did not answer in time, and "Unknown / TaskCanceledException" would send an operator
            // looking for a bug instead of at DispatchTimeoutSeconds.
            return new CommandDispatched(
                target.NodeId,
                commandType,
                correlationId,
                "Cancelled",
                "the dial did not complete within MeshTransport:DispatchTimeoutSeconds, or central "
                + "is shutting down");
        }
        catch (Exception ex)
        {
            // Type name only, never the message: this path also covers whatever the dialer's channel
            // construction can throw, and a message is the one place a misconfigured endpoint string
            // could carry something an operator did not mean to put in a log.
            return new CommandDispatched(
                target.NodeId, commandType, correlationId, "Unknown", ex.GetType().Name);
        }
    }

    /// <summary>
    ///     Bounds a status detail before it reaches the log. The string comes from the far side, and
    ///     the far side is a node — trusted, but not a reason to let it choose how many lines of
    ///     central's log one failed command occupies.
    /// </summary>
    /// <param name="detail">The received status detail.</param>
    /// <returns>The detail, at most 200 characters.</returns>
    private static string Truncate(string? detail)
        => detail is null || detail.Length <= 200 ? detail ?? "" : detail[..200] + "…";

    private void HandleCommandDispatched(CommandDispatched result)
    {
        if (result.FailureStatus is null)
        {
            _log.Debug(
                "Mesh: {CommandType} (correlationId={CorrelationId}) accepted by {NodeId} over gRPC",
                result.CommandType, result.CorrelationId, result.NodeId);
            return;
        }

        _log.Warning(
            "Mesh: {CommandType} (correlationId={CorrelationId}) was NOT delivered to {NodeId} "
            + "({Status}: {Detail}). It is dropped and will NOT be retried — the deployment apply "
            + "deadline and the node's boot-time replay are the backstops",
            result.CommandType, result.CorrelationId, result.NodeId,
            result.FailureStatus, result.FailureDetail);
    }

    private void HandleApplyAck(ApplyAck ack)
    {
        var coordinator = _coordinator();
        if (coordinator is null)
        {
            _log.Warning(
                "Received ApplyAck for {DeploymentId} from {NodeId} but the deploy coordinator is "
                + "not resolvable on this node; the ack is lost and the deployment will time out "
                + "naming a node that applied successfully",
                ack.DeploymentId, ack.NodeId);
            return;
        }

        // Forward, not Tell: preserves the original sender across the hop so the coordinator sees
        // the node rather than this relay.
        coordinator.Forward(ack);
    }

    private void LoadContactsFromDb()
    {
        var self = Self;
        CancellationToken ct;
        try
        {
            ct = _lifecycle.Token;
        }
        catch (ObjectDisposedException)
        {
            return; // Stopping.
        }

        var systemName = Context.System.Name;
        var dbFactory = _dbFactory;

        // Off the actor thread, result piped back as a message. A synchronous DB read here would
        // block every command dispatch behind a slow or unreachable SQL Server.
        Task.Run(async () =>
        {
            await using var db = await dbFactory.CreateDbContextAsync(ct).ConfigureAwait(false);
            // UNFILTERED, with the eligibility flags projected and applied per row below. The two
            // outputs of this read want different row sets: the contact/dispatch sets are the
            // enabled, non-maintenance nodes central talks TO, while the known-cluster gate answers
            // "is this a cluster central has ever heard of" and must include a cluster whose every
            // node is disabled or in maintenance — such a cluster can still send an ack, and denying
            // it would make maintenance mode an authentication switch.
            var rows = await db.ClusterNodes
                .AsNoTracking()
                .Select(n => new
                {
                    n.NodeId, n.ClusterId, n.Host, n.AkkaPort, n.GrpcPort, n.Enabled, n.MaintenanceMode,
                })
                .ToListAsync(ct)
                .ConfigureAwait(false);

            // Grouped BY ClusterId: central now dials one ClusterClient per application Cluster, so the
            // contact set must be partitioned the same way. A row's contacts join only its own cluster's
            // client.
            var byCluster = new Dictionary<string, List<string>>();
            var malformed = new List<string>();

            // The gRPC dispatch set, built from the SAME row set in the SAME pass. Sharing the read is
            // what makes "the gRPC fan reaches exactly the ClusterClient recipient set" structural
            // rather than a claim two separate queries would have to keep agreeing on.
            var targets = new List<MeshNodeTarget>(rows.Count);
            var portless = new List<string>();

            // Every cluster with ANY ClusterNode row — the known-cluster gate's membership set, and
            // deliberately a superset of the fan set. Independent of Enabled, MaintenanceMode and
            // GrpcPort alike: none of those describe whether the cluster exists.
            var clusterIds = new HashSet<string>(StringComparer.OrdinalIgnoreCase);

            foreach (var row in rows)
            {
                clusterIds.Add(row.ClusterId);

                // The eligibility filter that used to be the query's WHERE clause. Everything below
                // this line — receptionist contacts and gRPC targets — sees exactly the rows it saw
                // before.
                if (!row.Enabled || row.MaintenanceMode)
                    continue;

                if (row.GrpcPort is { } grpcPort)
                    targets.Add(new MeshNodeTarget(
                        row.NodeId, row.ClusterId, $"http://{row.Host}:{grpcPort}"));
                else
                    portless.Add(row.NodeId);

                var address = $"akka.tcp://{systemName}@{row.Host}:{row.AkkaPort}/system/receptionist";

                // Parse up front, per row, inside the loop's own guard. A single malformed row must
                // not abort the whole refresh and leave the contact set half-built — the sister
                // project shipped that bug and its regression test deliberately orders the bad row
                // first.
                if (ActorPath.TryParse(address, out _))
                {
                    if (!byCluster.TryGetValue(row.ClusterId, out var list))
                        byCluster[row.ClusterId] = list = new List<string>();
                    list.Add(address);
                }
                else
                {
                    malformed.Add($"{row.NodeId} ({row.ClusterId}) -> {address}");
                }
            }

            var contactsByCluster = byCluster.ToDictionary(
                kv => kv.Key, kv => (IReadOnlyList<string>)kv.Value);
            return new ContactsLoaded(
                contactsByCluster, malformed, targets, portless, clusterIds.ToList());
        }, ct).PipeTo(self);
    }

    private void HandleContactsLoaded(ContactsLoaded msg)
    {
        // Order is load-bearing and is pinned by test: the dispatch set is adopted BEFORE the gate is
        // published, so anything that observes the gate can rely on the targets already being current.
        if (_useGrpc)
            AdoptGrpcTargets(msg);

        // EVERY mode. See PreStart — the gate protects an endpoint mapped by port, not by mode.
        PublishKnownClusters(msg);

        if (!_useClusterClient)
            return; // Dps or Grpc: no ClusterClient to rebuild.

        foreach (var bad in msg.Malformed)
        {
            _log.Warning(
                "ClusterNode {Row} does not yield a usable receptionist address; skipping it in this "
                + "refresh (other nodes are unaffected). Check the row's Host and AkkaPort", bad);
        }

        var byCluster = msg.ContactsByCluster.ToDictionary(
            kv => kv.Key, kv => kv.Value.ToImmutableHashSet());

        // Clusters that dropped out of the DB entirely, or whose only rows were malformed this refresh:
        // stop and forget their clients so a vanished cluster stops receiving. LoadContactsFromDb only
        // creates a ByCluster key when at least one of its rows parsed, so an all-malformed cluster has
        // NO key at all and is caught here by the absent-key (!TryGetValue) arm — not by an empty value.
        var gone = _clients.Keys
            .Where(id => !byCluster.TryGetValue(id, out var c) || c.IsEmpty)
            .ToList();
        foreach (var clusterId in gone)
        {
            _log.Info(
                "Cluster {ClusterId} no longer yields any usable contact point; stopping its "
                + "ClusterClient", clusterId);
            Context.Stop(_clients[clusterId].Client);
            _clients.Remove(clusterId);
        }

        var anyUsable = false;
        foreach (var (clusterId, contacts) in byCluster)
        {
            if (contacts.IsEmpty)
            {
                // Defensive: LoadContactsFromDb never emits an empty-but-present cluster (a key is
                // created only on a first successfully-parsed row), so this arm is unreachable for the
                // DB producer — an all-malformed cluster is handled by the absent-key branch above. It
                // guards a ContactsLoaded (a public record) constructed elsewhere with an empty list.
                _log.Warning(
                    "Cluster {ClusterId} has no usable ClusterNode contact points; no ClusterClient "
                    + "was created for it and its nodes' commands will be dropped until a refresh finds "
                    + "one", clusterId);
                continue;
            }

            anyUsable = true;

            if (_clients.TryGetValue(clusterId, out var existing) && existing.Contacts.SetEquals(contacts))
                continue;

            RebuildClient(clusterId, contacts);
        }

        if (!anyUsable && _clients.Count == 0)
        {
            _log.Warning(
                "No usable ClusterNode contact points in any cluster; no mesh ClusterClient was "
                + "created and every central→node command will be dropped until a refresh finds one");
        }
    }

    /// <summary>
    ///     Adopts a contact load's gRPC dispatch set. <see cref="MeshTransportOptions.ModeGrpc"/> only
    ///     — the other two transports have no use for it.
    /// </summary>
    /// <param name="msg">The load result.</param>
    /// <remarks>
    ///     The malformed-receptionist-address warnings are deliberately NOT logged here: nothing on
    ///     this transport dials a receptionist, and reporting an unusable Akka address as a
    ///     command-plane problem would send an operator after the wrong column. A host that is
    ///     unusable as a gRPC endpoint surfaces instead as a named per-node dial failure on every
    ///     dispatch, which is a truer signal.
    /// </remarks>
    private void AdoptGrpcTargets(ContactsLoaded msg)
    {
        // Rebuilt against this refresh's skip list, not merely added to. A node that gained a port
        // and later lost it again must warn a second time — the operator's mental model is "the last
        // refresh told me", not "you were told once, in a log that has since rotated away".
        _warnedPortlessNodes.IntersectWith(msg.PortlessNodes);

        foreach (var nodeId in msg.PortlessNodes)
        {
            // Warn-once per node id per continuous skip run: a persistently portless row is
            // re-skipped every refresh, and that is the repetition TelemetryNodeSource suppresses.
            if (_warnedPortlessNodes.Add(nodeId))
            {
                _log.Warning(
                    "ClusterNode {NodeId} has no GrpcPort; it exposes no command endpoint and is "
                    + "skipped in this dispatch set (further skips of this node are silent). Commands "
                    + "will not reach it until the row carries a port", nodeId);
            }
        }

        _targets = msg.Targets;

        if (_targets.Count == 0)
        {
            _log.Warning(
                "No enabled, non-maintenance ClusterNode row carries a GrpcPort in any cluster; there "
                + "are no gRPC command targets and every central→node command will be dropped until a "
                + "refresh finds one");
        }
    }

    /// <summary>
    ///     Hands the PSK provider this refresh's cluster inventory and drops its key cache. Runs in
    ///     <b>every</b> transport mode.
    /// </summary>
    /// <param name="msg">The load result.</param>
    /// <remarks>
    ///     <para>
    ///         <b>Not gated on <see cref="MeshTransportOptions.Mode"/>, on purpose.</b> The gate feeds
    ///         <c>CentralCommandAuthInterceptor</c>, which is registered on every admin node, and the
    ///         ack service it protects is mapped by listener port — neither consults the mode. Gating
    ///         this on <c>Grpc</c> would mean a dark-switch rollback to <c>ClusterClient</c> or
    ///         <c>Dps</c> left that endpoint serving with a permanently inactive gate: the timing
    ///         oracle and the pre-authentication store/audit flood would both come back, silently,
    ///         as a side effect of a rollback nobody would connect to authentication.
    ///     </para>
    ///     <para>
    ///         <b><c>InvalidateAll</c>, not per-cluster <c>Invalidate</c>.</b> The requirement is that
    ///         a rotated key takes effect within one refresh, and only a whole-cache drop delivers
    ///         that: a key can rotate without the cluster's membership changing at all, so
    ///         invalidating only departed clusters would serve the pre-rotation key forever. The cost
    ///         is one store read per cluster per refresh — the cache holds a single entry per cluster.
    ///     </para>
    /// </remarks>
    private void PublishKnownClusters(ContactsLoaded msg)
    {
        _pskProvider?.SetKnownClusters(msg.ClusterIds);
        _pskProvider?.InvalidateAll();
    }

    /// <summary>Stops a cluster's existing client and creates a replacement for its new contact set.</summary>
    /// <remarks>
    ///     <para>
    ///         <b>One client per application <c>Cluster</c> (delivered in Phase 6).</b> This actor now
    ///         holds one <c>ClusterClient</c> per cluster and <see cref="HandleMeshCommand"/> fans
    ///         <c>SendToAll</c> across all of them. Once the fleet is split into one Akka mesh per
    ///         cluster, a <c>ClusterClientReceptionist</c> serves only <i>its own</i> mesh, so a single
    ///         client's <c>SendToAll</c> would reach the one mesh whose receptionist answered and leave
    ///         every other cluster silently on its old configuration. Fanning across a client per
    ///         cluster restores the fleet-wide reach the pre-split DPS broadcast gave.
    ///     </para>
    ///     <para>
    ///         Why this is not N× duplicate dispatch: each client dials only its own cluster's
    ///         receptionists, so a command is delivered <b>once per mesh</b>, not once per cluster into
    ///         every mesh. (While the fleet was a single mesh, the same code would have duplicated —
    ///         which is exactly why the fan-out could not land until the meshes were partitioned.)
    ///     </para>
    ///     <para>
    ///         Corollary worth stating because it is easy to assume otherwise: <b>the contact set
    ///         does not scope delivery within a mesh.</b> Excluding a maintenance-mode node from a
    ///         cluster's contacts does not stop it receiving commands — its receptionist still delivers
    ///         to every registered node-comm actor in that mesh. The filter is about which receptionists
    ///         are worth <i>dialling</i> (a node in maintenance may be switched off), not about who gets
    ///         the message.
    ///     </para>
    /// </remarks>
    /// <param name="clusterId">The application <c>Cluster</c> whose client is being (re)built.</param>
    /// <param name="contacts">The receptionist addresses to dial for that cluster.</param>
    private void RebuildClient(string clusterId, ImmutableHashSet<string> contacts)
    {
        if (_clients.TryGetValue(clusterId, out var existing))
        {
            _log.Info("Contact set for cluster {ClusterId} changed; replacing its ClusterClient", clusterId);
            Context.Stop(existing.Client);

            // Removed now: if the create below throws, a stale ref would route commands into a
            // stopping actor and they would vanish without a warning.
            _clients.Remove(clusterId);
        }

        try
        {
            var paths = contacts.Select(ActorPath.Parse).ToImmutableHashSet();
            var client = _clientFactory.Create(Context.System, clusterId, paths);
            _clients[clusterId] = (client, contacts);
            _log.Info(
                "Mesh ClusterClient for cluster {ClusterId} created with {Count} contact point(s)",
                clusterId, contacts.Count);
        }
        catch (Exception ex)
        {
            _log.Error(ex,
                "Failed to create the mesh ClusterClient for cluster {ClusterId}; its nodes' commands "
                + "are dropped until the next contact refresh succeeds", clusterId);
        }
    }

    /// <summary>Internal tick asking for a contact-point refresh.</summary>
    public sealed record RefreshContacts;

    /// <summary>Result of a contact-point load, piped back from the DB read.</summary>
    /// <param name="ContactsByCluster">
    ///     Receptionist addresses that parsed cleanly, grouped by the owning application
    ///     <c>Cluster</c>'s id. One <c>ClusterClient</c> is built per key.
    /// </param>
    /// <param name="Malformed">Rows that did not parse, described for the log.</param>
    /// <param name="Targets">
    ///     The gRPC dispatch set: one entry per enabled, non-maintenance row that carries a
    ///     <c>GrpcPort</c>. Built from the same rows as <paramref name="ContactsByCluster"/>.
    /// </param>
    /// <param name="PortlessNodes">
    ///     Node ids skipped because their row has no <c>GrpcPort</c>. Carried rather than logged in
    ///     the load task so the warn-once bookkeeping stays on the actor's thread.
    /// </param>
    /// <param name="ClusterIds">
    ///     Every application <c>Cluster</c> with <b>any</b> <c>ClusterNode</c> row — the PSK
    ///     provider's known-cluster gate membership, and deliberately a superset of
    ///     <paramref name="Targets"/>: a cluster whose nodes are all disabled or in maintenance can
    ///     still send an ack, and refusing it would make maintenance mode an authentication switch.
    /// </param>
    public sealed record ContactsLoaded(
        IReadOnlyDictionary<string, IReadOnlyList<string>> ContactsByCluster,
        IReadOnlyList<string> Malformed,
        IReadOnlyList<MeshNodeTarget> Targets,
        IReadOnlyList<string> PortlessNodes,
        IReadOnlyList<string> ClusterIds);

    /// <summary>One node the gRPC command plane can dial.</summary>
    /// <param name="NodeId">The <c>ClusterNode</c> row's id — what a failure log names.</param>
    /// <param name="ClusterId">
    ///     The node's application <c>Cluster</c>. Selects the pre-shared key presented to it: a node
    ///     dialled with another cluster's id is graded against the wrong key and refuses the command.
    /// </param>
    /// <param name="Endpoint">The h2c base address, <c>http://{Host}:{GrpcPort}</c>.</param>
    public sealed record MeshNodeTarget(string NodeId, string ClusterId, string Endpoint);

    /// <summary>
    ///     The outcome of one node's dial, piped back to the actor <b>for logging only</b>. Nothing
    ///     retries, counts or waits on these — a deployment's completion is judged by the acks nodes
    ///     send back, never by whether central's outbound call succeeded.
    /// </summary>
    /// <param name="NodeId">The node dialled.</param>
    /// <param name="CommandType">CLR type name of the command.</param>
    /// <param name="CorrelationId">The envelope's transport correlation.</param>
    /// <param name="FailureStatus">
    ///     <see langword="null"/> on success; otherwise the gRPC status code's name, or
    ///     <c>Unknown</c> for a non-gRPC fault.
    /// </param>
    /// <param name="FailureDetail">
    ///     The gRPC status detail, or the exception's type name for a non-gRPC fault. Never a key,
    ///     and never a raw exception message.
    /// </param>
    public sealed record CommandDispatched(
        string NodeId,
        string CommandType,
        string CorrelationId,
        string? FailureStatus,
        string? FailureDetail);
}