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

using System.Collections.Concurrent;
using Grpc.Core;
using Grpc.Net.Client;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
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;

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

/// <summary>
///     Posts this node's <see cref="ApplyAck"/> to central's <c>CentralCommandService</c> as a unary
///     gRPC call (mesh gRPC command plane), the <c>MeshTransport:Mode = Grpc</c>
///     cross-mesh ack transport.
/// </summary>
/// <remarks>
///     <para>
///         <b>One ordered pass, stopping at the first success.</b>
///         <see cref="MeshTransportOptions.CentralGrpcEndpoints"/> is an availability list, not a
///         fan-out list — the single-target semantic this leg owes (one ack per node per dispatch,
///         deliberately not <c>SendToAll</c>).
///     </para>
///     <para>
///         <b>Delivery is at-most-once per SUCCESSFUL attempt, but the walk as a whole is
///         at-least-once</b>, and that is fine. An endpoint that processes the ack and then exceeds
///         its deadline before answering is indistinguishable here from one that never received it, so
///         the next endpoint is tried and central may see the ack twice.
///         <c>ConfigPublishCoordinator.HandleAck</c> absorbs that idempotently: acks land in a
///         <c>NodeId</c>-keyed dictionary behind an expected-set membership guard and a
///         stale-deployment guard, so a duplicate overwrites this node's own entry rather than
///         inflating the count, and the row upsert behind it is idempotent too. Duplicates therefore
///         cannot seal a deployment other nodes never applied — the single-endpoint rule is defence in
///         depth and a bound on wasted work, not a correctness dependency.
///     </para>
///     <para>
///         <b>No retry beyond that single pass, and no queue.</b> The transport buffers nothing
///         and retried nothing, and this is a like-for-like replacement, not an upgrade to delivery
///         guarantees. The deploy path's own apply deadline stays the only recovery mechanism; a
///         retry loop here would instead pile acks up behind an unreachable central and deliver a
///         burst of stale ones when it returned.
///     </para>
///     <para>
///         <b>All endpoints failing is a DROP — but a loud one.</b> That is the same outcome the
///         node has when it cannot reach central; what changes is that the operator
///         now gets the node, the deployment, the correlation id and the status each endpoint
///         answered with, which is what distinguishes "central is down" from "central rejected this
///         node's key".
///     </para>
///     <para>
///         <b>The bearer never reaches a log.</b> It travels in the call headers on every attempt and
///         the failure log names endpoints and status codes only — a per-attempt dump of the
///         <see cref="Metadata"/> would write the cluster's pre-shared key into the node log of every
///         deployment that could not reach central.
///     </para>
///     <para>
///         <b>The cluster id comes from this node's own <c>cluster-&lt;id&gt;</c> Akka role</b>, read
///         through <see cref="RoleParser"/> off the same bound <c>Cluster:Roles</c> that
///         <c>ClusterRoleInfo</c> reads (the <c>OTOPCUA_ROLES</c> fallback is already applied to those
///         options by <c>AddOtOpcUaCluster</c>'s <c>PostConfigure</c>, so nothing is re-resolved
///         here). The two must agree: central grades the bearer against the key of the cluster named
///         in the header, so a node that authenticated as a different cluster than it gossips would be
///         rejected on every ack.
///     </para>
/// </remarks>
public sealed class GrpcApplyAckSender : IApplyAckSender, IDisposable
{
    /// <summary>The cluster the caller claims to be; central grades the bearer against ITS key.</summary>
    private const string ClusterHeader = "x-otopcua-cluster";

    private readonly MeshTransportOptions _options;
    private readonly string? _clusterId;
    private readonly ILogger<GrpcApplyAckSender> _log;
    private readonly Func<string, CentralCommandService.CentralCommandServiceClient> _clientFactory;
    private readonly ConcurrentDictionary<string, GrpcChannel> _channels = new(StringComparer.Ordinal);

    /// <summary>Initializes a new instance of the <see cref="GrpcApplyAckSender"/> class.</summary>
    /// <param name="options">Mesh-transport options; supplies the endpoints, the PSK and the deadline.</param>
    /// <param name="clusterOptions">
    ///     Cluster options, read only for this node's roles — the <c>cluster-&lt;id&gt;</c> one names
    ///     the cluster this node authenticates as.
    /// </param>
    /// <param name="log">Logger.</param>
    /// <param name="clientFactory">
    ///     Seam for tests: builds the generated client for an endpoint. Defaults to a real
    ///     <see cref="GrpcChannel"/> (one cached per endpoint).
    /// </param>
    public GrpcApplyAckSender(
        IOptions<MeshTransportOptions> options,
        IOptions<AkkaClusterOptions> clusterOptions,
        ILogger<GrpcApplyAckSender> log,
        Func<string, CentralCommandService.CentralCommandServiceClient>? clientFactory = null)
    {
        ArgumentNullException.ThrowIfNull(options);
        ArgumentNullException.ThrowIfNull(clusterOptions);
        ArgumentNullException.ThrowIfNull(log);

        _options = options.Value;
        _log = log;
        _clusterId = RoleParser.ClusterIdFromRoles(clusterOptions.Value.Roles);
        _clientFactory = clientFactory ?? DefaultClientFactory;

        if (_clusterId is null)
        {
            // Said once, at construction, rather than on every ack. Central cannot grade a bearer it
            // has no cluster for, so every ack this node posts will come back PermissionDenied — and
            // without this line that reads as a key mismatch rather than a missing role.
            _log.LogWarning(
                "This node carries no '{Prefix}<ClusterId>' role, so its deployment acks will be "
                + "posted without the {Header} header and central will reject every one of them. Add "
                + "the cluster-scoped role to Cluster:Roles (or OTOPCUA_ROLES).",
                RoleParser.ClusterRolePrefix, ClusterHeader);
        }
    }

    /// <inheritdoc />
    public async Task SendAsync(ApplyAck ack, CancellationToken cancellationToken = default)
    {
        try
        {
            // INSIDE the try, not before it. This is an async method, so a guard clause above the try
            // does not throw to the caller — it faults the returned task, which the caller discards.
            // That would break this method's "never faults" promise in the one case a caller could
            // never see.
            ArgumentNullException.ThrowIfNull(ack);

            await PostAsync(ack, cancellationToken);
        }
        catch (Exception ex)
        {
            // Deliberately unbounded, and deliberately not a rethrow. The caller discards this task
            // (see IApplyAckSender), so anything escaping would be swallowed by the runtime and
            // resurface as an unobserved exception at GC — a drop with no line in the node log, which
            // is precisely the failure mode this transport replaces. Encoding faults and non-RpcException
            // transport faults both land here.
            //
            // The ack is null-conditional because the guard above is one of the things this catches.
            _log.LogWarning(
                ex,
                "Dropping the ApplyAck for {DeploymentId} (correlation {CorrelationId}) from {NodeId}: "
                + "posting it to central failed unexpectedly. It is not buffered and will not be "
                + "retried, so the deployment will time out at central naming this node even though it "
                + "applied successfully.",
                ack?.DeploymentId, ack?.CorrelationId, ack?.NodeId);
        }
    }

    private async Task PostAsync(ApplyAck ack, CancellationToken cancellationToken)
    {
        var endpoints = _options.CentralGrpcEndpoints;
        if (endpoints.Length == 0)
        {
            _log.LogWarning(
                "Dropping the ApplyAck for {DeploymentId} from {NodeId}: no "
                + "MeshTransport:CentralGrpcEndpoints are configured, so there is nowhere to post it. "
                + "The deployment will time out at central naming this node even though it applied.",
                ack.DeploymentId, ack.NodeId);
            return;
        }

        var message = MeshTransportDtoMapper.ToProto(ack);

        // Hoisted: read-only, and the walk is strictly sequential, so one Metadata is shared safely
        // across the attempts. The DEADLINE emphatically is not hoisted — see below.
        var headers = new Metadata { { "authorization", $"Bearer {_options.GrpcPsk}" } };
        if (_clusterId is not null)
            headers.Add(ClusterHeader, _clusterId);

        var failures = new List<string>(endpoints.Length);
        foreach (var endpoint in endpoints)
        {
            try
            {
                // Computed FRESH per endpoint, which is what MeshTransport:AckTimeoutSeconds documents
                // ("applied per endpoint, not per ack"). A single absolute deadline hoisted out of this
                // loop defeats failover against the characteristic WAN failure: a central that HANGS
                // (SYN black-hole, frozen host) rather than refusing. The first endpoint burns the whole
                // timeout, and the second is then dialled with a deadline already in the past — it fails
                // instantly as DeadlineExceeded without a packet leaving the node, so the surviving
                // central is never actually tried.
                var deadline = _options.AckTimeoutSeconds > 0
                    ? DateTime.UtcNow.AddSeconds(_options.AckTimeoutSeconds)
                    : (DateTime?)null;

                var client = _clientFactory(endpoint);
                using var call = client.ReportApplyAsync(message, headers, deadline, cancellationToken);
                await call;

                if (failures.Count == 0)
                {
                    _log.LogDebug(
                        "Posted the ApplyAck for {DeploymentId} ({Outcome}) to {Endpoint}.",
                        ack.DeploymentId, ack.Outcome, endpoint);
                }
                else
                {
                    // Information, not Debug, because this is a FAILOVER: the ack landed, but only
                    // after one or more endpoints cost their full deadline. The deploy simply took
                    // that much longer, and without this line the only evidence is the absence of the
                    // all-endpoints-failed Warning — which reads identically to a healthy first-try.
                    _log.LogInformation(
                        "Posted the ApplyAck for {DeploymentId} ({Outcome}) to {Endpoint} after "
                        + "{FailedCount} endpoint(s) failed ({Failures}).",
                        ack.DeploymentId, ack.Outcome, endpoint, failures.Count, string.Join("; ", failures));
                }

                // The single-endpoint rule, and it is this `return`: the walk stops at the first
                // endpoint that accepted the ack and never goes back to one that did not.
                return;
            }
            catch (RpcException ex)
            {
                // Status code only — never the exception's own message and never the headers, either
                // of which could echo credential material back into the log.
                failures.Add($"{endpoint} => {ex.StatusCode}");
            }
        }

        _log.LogWarning(
            "Dropping the ApplyAck for {DeploymentId} (correlation {CorrelationId}) from {NodeId}: all "
            + "{Count} central endpoint(s) refused it ({Failures}). It is not buffered and will not be "
            + "retried, so the deployment will time out at central naming this node even though it "
            + "applied successfully.",
            ack.DeploymentId, ack.CorrelationId, ack.NodeId, endpoints.Length, string.Join("; ", failures));
    }

    private CentralCommandService.CentralCommandServiceClient DefaultClientFactory(string endpoint)
    {
        // h2c: GrpcChannel over an http:// address negotiates prior-knowledge HTTP/2. One channel per
        // endpoint, cached and reused (creating a channel per ack would leak a socket per deployment).
        var channel = _channels.GetOrAdd(endpoint, ep => GrpcChannel.ForAddress(ep));
        return new CentralCommandService.CentralCommandServiceClient(channel);
    }

    /// <summary>Disposes every cached gRPC channel.</summary>
    public void Dispose()
    {
        foreach (var channel in _channels.Values)
            channel.Dispose();
        _channels.Clear();
    }
}