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

using Akka.Actor;
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 (pair-local) or gRPC (cross-mesh),
///     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 posts its ack to
///         whichever central node answers first, so a live comm actor must exist on both. A singleton
///         would be reachable only through the one node hosting it, and the other central node's ack
///         ingress would have nothing to forward to.
///     </para>
///     <para>
///         <b>The gRPC branch fans node by node over the contact load's target set.</b> That load
///         selects exactly the enabled, non-maintenance <c>ClusterNode</c> rows carrying a
///         <c>GrpcPort</c>, and the fan dials every one of them, one call per node. 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 both transports.</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
///         DistributedPubSub 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 const string BootstrapRetryTimerKey = "mesh-contact-bootstrap-retry";

    // While the known-cluster gate is unpublished (PSK provider unresolved, or the first
    // ClusterNode load failed) a single-shot RefreshContacts is re-armed every
    // DefaultBootstrapRetryInterval, at most MaxBootstrapRetries times (~30 s), stopping at the
    // first publish. Constants, not config. Steady state costs nothing extra.
    private static readonly TimeSpan DefaultBootstrapRetryInterval = TimeSpan.FromSeconds(2);
    private const int MaxBootstrapRetries = 15;
    private readonly TimeSpan _bootstrapRetryInterval;
    private int _bootstrapRetries;
    private int _refreshAttempts;
    private bool _gatePublished;

    private readonly IDbContextFactory<OtOpcUaConfigDbContext> _dbFactory;
    private readonly MeshTransportOptions _options;
    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;

    // Memoize-on-first-use, NOT PreStart-only: PreStart runs on a dispatcher thread concurrently
    // with the builder's remaining starters, so its resolve can lose a millisecond-wide race
    // against AuditWriterActorKey's registration (the fourth starter) and Resolve() would swallow that
    // into a silent null — the known-cluster gate unpublished, fail-open. These properties
    // retry on first use, but the first RefreshContacts is sent from PreStart (TimeSpan.Zero) and
    // can ALSO land inside the starter window (a fast DB load finishes in ms), so first-use is not
    // enough: while the gate is unpublished a bounded fast retry (ScheduleBootstrapRetry) re-arms
    // RefreshContacts every couple of seconds until it publishes. Actor-thread
    // only, so ??= needs no synchronisation; both are container singletons, so a repeat resolve
    // is a dictionary hit.
    private IMeshPskProvider? PskProvider =>
        _pskProvider ??= Resolve(_pskProviderFactory, nameof(IMeshPskProvider), firstUse: true);

    private IMeshCommandDialer? Dialer =>
        _dialer ??= Resolve(_dialerFactory, nameof(IMeshCommandDialer), firstUse: true);

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

    // 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);

    // Set once PublishKnownClusters has reported a genuinely-absent PSK provider (a factory WAS
    // supplied and still resolves null after the first-use retry). PublishKnownClusters runs on
    // the contact-refresh timer for the actor's whole life, so an unguarded Error there would
    // repeat every ContactRefreshSeconds forever. Same warn-once shape as _warnedPortlessNodes;
    // actor thread only.
    private bool _pskProviderAbsenceReported;

    // 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="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 Dps transport needs 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>
    /// <param name="bootstrapRetryInterval">
    ///     Interval of the bounded fast retry run while the known-cluster gate is unpublished
    ///     (default 2 s, at most 15 attempts).
    /// </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,
        Func<IActorRef?> coordinator,
        Func<IMeshCommandDialer?>? dialer = null,
        Func<IMeshPskProvider?>? pskProvider = null,
        TimeSpan? bootstrapRetryInterval = null) =>
        Akka.Actor.Props.Create(() =>
            new CentralCommunicationActor(
                dbFactory, options, coordinator, dialer, pskProvider, bootstrapRetryInterval));

    /// <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="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>
    /// <param name="bootstrapRetryInterval">Fast-retry interval while the gate is unpublished; default 2 s.</param>
    public CentralCommunicationActor(
        IDbContextFactory<OtOpcUaConfigDbContext> dbFactory,
        MeshTransportOptions options,
        Func<IActorRef?> coordinator,
        Func<IMeshCommandDialer?>? dialer = null,
        Func<IMeshPskProvider?>? pskProvider = null,
        TimeSpan? bootstrapRetryInterval = null)
    {
        _bootstrapRetryInterval = bootstrapRetryInterval ?? DefaultBootstrapRetryInterval;
        _dbFactory = dbFactory;
        _options = options;
        _coordinator = coordinator;
        _dialerFactory = dialer;
        _pskProviderFactory = pskProvider;
        _useGrpc = string.Equals(
            options.Mode, MeshTransportOptions.ModeGrpc, StringComparison.OrdinalIgnoreCase);

        Receive<MeshCommand>(HandleMeshCommand);
        Receive<ApplyAck>(HandleApplyAck);
        Receive<RefreshContacts>(_ =>
        {
            _refreshAttempts++;
            LoadContactsFromDb();
        });
        Receive<ContactsLoaded>(HandleContactsLoaded);
        Receive<PskRefreshed>(msg => _log.Debug("Mesh PSK refresh completed for {Count} cluster(s)", msg.ClusterCount));
        Receive<PskRefreshFailed>(msg => _log.Warning(msg.Error,
            "Mesh PSK refresh failed; previously resolved keys stay in service until the next refresh"));
        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. Until the first publish a bounded fast retry (above) heals it.
        Receive<Status.Failure>(f =>
        {
            // Gate still unpublished (boot with ConfigDb briefly down): retry fast, bounded.
            if (!_gatePublished)
                ScheduleBootstrapRetry();
            _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.
        //
        // Quiet (firstUse: false — see Resolve): PreStart runs on a dispatcher thread concurrently
        // with the builder's remaining starters and loses a millisecond-wide race against
        // AuditWriterActorKey's registration (the fourth starter) on essentially every admin boot (live
        // gates: 1A 2/3 boots, 1B 4/4 boots). That is the ORDINARY case, not a malfunction — the
        // memoized PskProvider/Dialer properties retry on first use, and the bounded bootstrap
        // retry re-arms RefreshContacts until the gate publishes (the first refresh itself can
        // land inside the starter window). Logging it as
        // Warning/Error twice a boot is what made a soak error-grep on a healthy rig impossible.
        _dialer = Resolve(_dialerFactory, nameof(IMeshCommandDialer), firstUse: false);
        _pskProvider = Resolve(_pskProviderFactory, nameof(IMeshPskProvider), firstUse: false);

        if (!_useGrpc)
        {
            _log.Info(
                "Mesh transport is {Mode}: commands are published on DistributedPubSub and nothing "
                + "is dialled. 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);
        }

        // Deliberately NO null guard for either dependency here. At this point a still-null value
        // is indistinguishable from the starter-ordering race Resolve just logged quietly above,
        // so a loud report here cannot tell a lost race from a genuine absence — that decision
        // belongs where a still-null result after the memoized properties' first-use retry
        // actually costs something: DispatchOverGrpc for the dialer (every dropped command), and
        // PublishKnownClusters for the PSK provider (once per incarnation, since that handler runs
        // on the contact-refresh timer for the actor's whole life).

        // 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
        // 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>Re-arms a single-shot <see cref="RefreshContacts"/> while the gate is unpublished.</summary>
    /// <returns><see langword="false"/> when the retry budget is spent and nothing was scheduled.</returns>
    private bool ScheduleBootstrapRetry()
    {
        if (_bootstrapRetries >= MaxBootstrapRetries)
            return false;
        _bootstrapRetries++;
        Timers.StartSingleTimer(BootstrapRetryTimerKey, new RefreshContacts(), _bootstrapRetryInterval);
        return true;
    }

    /// <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>
    /// <param name="firstUse">
    ///     <see langword="false"/> for the eager <see cref="PreStart"/> call, <see langword="true"/>
    ///     for the retry the memoized <see cref="Dialer"/>/<see cref="PskProvider"/> properties make
    ///     on first use. Controls wording only, deliberately at the SAME (Debug) severity either
    ///     way — see the remarks on why the loud signal lives elsewhere.
    /// </param>
    /// <returns>The resolved dependency, or <see langword="null"/>.</returns>
    /// <remarks>
    ///     <para>
    ///         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.
    ///     </para>
    ///     <para>
    ///         <b>Never logs above Debug, on EITHER call site.</b> From inside this method, a
    ///         PreStart resolve losing the starter-ordering race and a genuinely broken factory are
    ///         indistinguishable — both return null. Only the CALLER can tell them apart, because
    ///         only the caller knows whether a value came back after the retry the memoized
    ///         properties perform. So the loud, operator-facing signal is deliberately NOT here: it
    ///         is at the two places that consume a still-null result after that retry —
    ///         <see cref="DispatchOverGrpc"/> for the dialer (fires every dropped command) and
    ///         <see cref="PublishKnownClusters"/> for the PSK provider (fires once per incarnation,
    ///         because that call site runs on a timer forever, where <see cref="DispatchOverGrpc"/>
    ///         only runs once per command).
    ///     </para>
    /// </remarks>
    private T? Resolve<T>(Func<T?>? factory, string name, bool firstUse)
        where T : class
    {
        if (factory is null)
            return null; // Never supplied — not a resolve failure, nothing to report at either call site.

        T? result;
        try
        {
            result = factory();
        }
        catch (Exception ex)
        {
            LogUnresolved(name, firstUse, ex);
            return null;
        }

        if (result is null)
            LogUnresolved(name, firstUse, null);

        return result;
    }

    private void LogUnresolved(string name, bool firstUse, Exception? ex)
    {
        if (firstUse)
        {
            _log.Debug(
                ex,
                "{Dependency} is still not resolvable on the first-use retry; continuing without "
                + "it for now. Whether this is worth an operator-facing signal is the caller's call",
                name);
        }
        else
        {
            _log.Debug(
                ex,
                "{Dependency} was not yet resolvable at PreStart — likely a starter-ordering race "
                + "with a later starter that has not finished registering yet; will retry on first "
                + "use", name);
        }
    }

    /// <inheritdoc />
    protected override void PostStop()
    {
        // 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 first, so a Grpc-mode command never reaches DistributedPubSub. DPS is mesh-wide and
        // cannot cross an application-Cluster boundary at all in the split topology.
        if (_useGrpc)
        {
            DispatchOverGrpc(cmd);
            return;
        }

        DistributedPubSub.Get(Context.System).Mediator.Tell(new Publish(cmd.Topic, 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. 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 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 dispatch set is 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.GrpcPort, n.Enabled, n.MaintenanceMode,
                })
                .ToListAsync(ct)
                .ConfigureAwait(false);

            // The gRPC dispatch set: one entry per eligible row that carries a port.
            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 — the 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);
            }

            return new ContactsLoaded(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);
    }

    /// <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>
    ///     A host that is unusable as a gRPC endpoint is deliberately NOT screened here: it surfaces
    ///     as a named per-node dial failure on every dispatch, which is a truer signal than a
    ///     refresh-time warning about a string that may well be fine.
    /// </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;

        // R03:1.3 — release the channels of endpoints that left the dispatch set (host renamed, port
        // moved). Only the resolved dialer is asked; nothing is resolved just to prune it.
        if (_dialer is { } dialer)
        {
            try
            {
                dialer.RetainOnly(_targets.Select(t => t.Endpoint).ToHashSet(StringComparer.Ordinal));
            }
            catch (Exception ex)
            {
                _log.Warning(ex, "Mesh dialer failed to release channels for retired endpoints; they stay cached until the next refresh");
            }
        }

        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>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>
    ///     <para>
    ///         <b>This is the PSK provider's "first use" call site.</b> <see cref="PskProvider"/> is
    ///         evaluated exactly once per call and its result reused below; evaluating it
    ///         per-statement would double the memoized property's retry rate for no benefit. A
    ///         still-null result here means the property's own retry, on top of PreStart's original
    ///         attempt, has ALSO failed — the genuinely permanent absence — so this is where that
    ///         gets reported loudly, guarded to once per incarnation because this handler runs on
    ///         the contact-refresh timer for the actor's entire life and an unguarded Error here
    ///         would repeat every <c>ContactRefreshSeconds</c> forever.
    ///     </para>
    /// </remarks>
    private void PublishKnownClusters(ContactsLoaded msg)
    {
        var pskProvider = PskProvider;
        if (pskProvider is null)
        {
            // Gated on a factory having been supplied at all: a never-supplied factory is a
            // deliberate wiring/test choice Resolve() already stayed silent about (see its
            // remarks), not a resolve failure, so it earns no Error here either.
            // Error only when the scheduler REFUSES (budget spent and this final attempt also
            // failed), so a healed boot race stays quiet.
            var budgetSpent = _pskProviderFactory is not null && !_gatePublished
                && !ScheduleBootstrapRetry();
            if (budgetSpent && !_pskProviderAbsenceReported)
            {
                _pskProviderAbsenceReported = true;
                _log.Error(
                    "No IMeshPskProvider is resolvable for this actor, even after a first-use retry and "
                    + "the bounded bootstrap retries; 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 that the ControlPlane registration wires up "
                    + "IMeshPskProvider (this will not be logged again this incarnation)");
            }

            return;
        }

        pskProvider.SetKnownClusters(msg.ClusterIds);

        // R03:3.4 — refresh the per-cluster keys OFF the request path and swap them in atomically,
        // instead of dropping the cache and making the first ack per cluster re-read the store on a
        // gRPC thread. A provider that does not pre-resolve (the interface default, and every fake)
        // still drops its cache here, synchronously, exactly as before.
        CancellationToken ct;
        try
        {
            ct = _lifecycle.Token;
        }
        catch (ObjectDisposedException)
        {
            return; // Stopping.
        }
        var self = Self;
        Task refresh;
        try
        {
            refresh = pskProvider.RefreshAsync(msg.ClusterIds, ct);
        }
        catch (Exception ex)
        {
            _log.Warning(ex, "Mesh PSK refresh threw synchronously; cached keys are unchanged until the next refresh");
            refresh = Task.CompletedTask;
        }
        if (!refresh.IsCompletedSuccessfully)
        {
            refresh.PipeTo(self,
                success: () => new PskRefreshed(msg.ClusterIds.Count),
                failure: ex => new PskRefreshFailed(ex));
        }

        if (!_gatePublished)
        {
            _gatePublished = true;
            Timers.Cancel(BootstrapRetryTimerKey);
            _log.Info(
                "Known-cluster gate published ({Count} clusters) after {Attempts} refresh attempt(s)",
                msg.ClusterIds.Count, _refreshAttempts);
        }
    }

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

    /// <summary>The asynchronous PSK refresh (R03:3.4) completed.</summary>
    /// <param name="ClusterCount">How many clusters were refreshed.</param>
    public sealed record PskRefreshed(int ClusterCount);

    /// <summary>The asynchronous PSK refresh (R03:3.4) threw; the previous keys stay in service.</summary>
    /// <param name="Error">The failure.</param>
    public sealed record PskRefreshFailed(Exception Error);

    /// <summary>Result of a contact-point load, piped back from the DB read.</summary>
    /// <param name="Targets">
    ///     The gRPC dispatch set: one entry per enabled, non-maintenance row that carries a
    ///     <c>GrpcPort</c>.
    /// </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(
        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);
}