lmxopcua/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Communication/NodeCommunicationActor.cs

using Akka.Actor;
using Akka.Cluster.Tools.Client;
using Akka.Event;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Admin;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Mesh;
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;

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

/// <summary>
///     Node-side end of the central↔node command boundary (per-cluster mesh Phase 2), registered
///     with the <c>ClusterClientReceptionist</c> at <see cref="MeshPaths.NodeCommunication"/>.
/// </summary>
/// <remarks>
///     <para>
///         <b>Registered per node, NOT as a cluster singleton</b> — central's ClusterClient rotates
///         across contact points and must find a live comm actor at whichever node answers.
///     </para>
///     <para>
///         <b>Inbound commands are re-emitted on the node's <see cref="EventStream"/>, not on their
///         DistributedPubSub topic.</b> DPS is mesh-wide, and central <c>SendToAll</c>s to every
///         node's comm actor — so a DPS re-publish would deliver N copies of every command to every
///         subscriber, where N is the node count. The EventStream is node-local by construction,
///         which is exactly the needed semantics. It also avoids a registration handshake with
///         actors spawned later: <c>ScriptedAlarmHostActor</c> is a <i>child of</i>
///         <c>DriverHostActor</c> and has no registry key, so there is no ref to hand in at wiring
///         time.
///     </para>
///     <para>
///         <b>Outbound is <see cref="ApplyAck"/> only.</b> There is no Ask across this boundary in
///         Phase 2 — every migrated command is fire-and-forget, and the deploy ack returns as an
///         independent message the coordinator already subscribes for. Nothing here needs sender
///         preservation, which is why this actor is far smaller than the sister project's.
///     </para>
/// </remarks>
public sealed class NodeCommunicationActor : ReceiveActor
{
    private readonly IActorRef? _centralClient;
    private readonly IApplyAckSender? _ackSender;
    private readonly ILoggingAdapter _log = Context.GetLogger();

    /// <summary>Creates the props for this actor.</summary>
    /// <param name="centralClient">
    ///     ClusterClient dialling central's receptionist, or <see langword="null"/> under
    ///     <c>MeshTransport:Mode = Dps</c>, where acks still go out over DistributedPubSub from
    ///     <c>DriverHostActor</c> and this actor never sees them — and under
    ///     <c>Mode = Grpc</c>, where no ClusterClient is created at all.
    /// </param>
    /// <param name="ackSender">
    ///     The gRPC ack transport under <c>MeshTransport:Mode = Grpc</c>, or <see langword="null"/>
    ///     under <c>Dps</c> and <c>ClusterClient</c>.
    /// </param>
    /// <returns>The props.</returns>
    public static Props Props(IActorRef? centralClient, IApplyAckSender? ackSender = null) =>
        Akka.Actor.Props.Create(() => new NodeCommunicationActor(centralClient, ackSender));

    /// <summary>Initializes a new instance of the <see cref="NodeCommunicationActor"/> class.</summary>
    /// <param name="centralClient">ClusterClient dialling central, or <see langword="null"/>.</param>
    /// <param name="ackSender">The gRPC ack transport, or <see langword="null"/>.</param>
    public NodeCommunicationActor(IActorRef? centralClient, IApplyAckSender? ackSender = null)
    {
        _centralClient = centralClient;
        _ackSender = ackSender;

        // Inbound from central. Each type is listed explicitly rather than caught by a ReceiveAny:
        // an unknown type must dead-letter loudly during development rather than be republished
        // blind onto the node's EventStream, where it would be silently ignored by every subscriber.
        Receive<DispatchDeployment>(PublishLocally);
        Receive<RestartDriver>(PublishLocally);
        Receive<ReconnectDriver>(PublishLocally);
        Receive<AlarmCommand>(PublishLocally);

        // Outbound to central.
        Receive<ApplyAck>(HandleApplyAck);
    }

    private void PublishLocally(object msg)
    {
        _log.Debug("Mesh: received {MessageType} from central; publishing node-locally", msg.GetType().Name);
        Context.System.EventStream.Publish(msg);
    }

    private void HandleApplyAck(ApplyAck ack)
    {
        if (_ackSender is not null)
        {
            // Fire-and-forget by construction, exactly as the ClusterClient Tell below is. The sender
            // never throws and never retries (see IApplyAckSender), so there is nothing to await and
            // nothing to pipe back: awaiting would stall this mailbox behind a network call to central,
            // and an Ask would invent a request/response the one-way contract does not have.
            //
            // Checked BEFORE the ClusterClient. The registration arm builds exactly one transport, so
            // both are never set — but were they, posting over both would deliver two acks for one
            // apply and duplicate central's work for no benefit.
            try
            {
                _ = _ackSender.SendAsync(ack);
            }
            catch (Exception ex)
            {
                // "Never throws" is a contract on a public seam, which makes it documentation until
                // something enforces it. An implementation that threw synchronously — a guard clause
                // ahead of the async state machine is the easy way to do it by accident — would fault
                // this Receive, restart the actor, and take the node's whole inbound command path down
                // with it. One dropped ack is a deployment that times out; a restarting comm actor is
                // every command to this node going missing.
                _log.Error(
                    ex,
                    "The ApplyAck sender threw synchronously for {DeploymentId}; dropping the ack. "
                    + "IApplyAckSender must never throw — this is a defect in the registered sender.",
                    ack.DeploymentId);
            }

            return;
        }

        if (_centralClient is null)
        {
            _log.Warning(
                "No central ack transport — dropping ApplyAck for {DeploymentId}. The ack is not "
                + "buffered and will not be retried, so the deployment will time out at central "
                + "naming this node even though it applied successfully. Check "
                + "MeshTransport:CentralContactPoints (ClusterClient mode) or "
                + "MeshTransport:CentralGrpcEndpoints (Grpc mode)",
                ack.DeploymentId);
            return;
        }

        // Send, not SendToAll: the deploy coordinator is a single singleton, and the central comm
        // actor forwards to its proxy. SendToAll would deliver one ack per central node, and the
        // coordinator would count the same node twice.
        _centralClient.Tell(new ClusterClient.Send(MeshPaths.CentralCommunication, ack));
    }
}