lmxopcua/src/Server/ZB.MOM.WW.OtOpcUa.Host/Grpc/TelemetryStreamGrpcService.cs

using Grpc.Core;
using Microsoft.Extensions.Logging;
using ZB.MOM.WW.OtOpcUa.Commons.Protos.Telemetry.V1;
using ZB.MOM.WW.OtOpcUa.Runtime.Telemetry;
// The generated service container is itself named TelemetryStreamService; alias its nested server
// base so this impl can keep the natural name without colliding with the generated type.
using GeneratedServiceBase =
    ZB.MOM.WW.OtOpcUa.Commons.Protos.Telemetry.V1.TelemetryStreamService.TelemetryStreamServiceBase;

namespace ZB.MOM.WW.OtOpcUa.Host.Grpc;

/// <summary>
///     Node-side live-telemetry streaming server (per-cluster mesh Phase 5). Central dials a driver
///     node and opens <see cref="Subscribe"/>; the node attaches a fresh reader on its
///     <see cref="ITelemetryLocalHub"/> and pumps this node's own telemetry
///     (alarm-transitions / script-logs / driver-health / driver-resilience) to the caller as a
///     sequence of <see cref="TelemetryEvent"/> envelopes until the client cancels or the
///     max-lifetime cap fires.
/// </summary>
/// <remarks>
///     The hub carries ONLY this node's telemetry, so a per-node stream never double-counts a peer's
///     events. Fan-out is lossy under backpressure by design (the per-subscriber channel is
///     bounded/DropOldest inside the hub). A simple process-wide concurrency cap sheds excess dials
///     with <see cref="StatusCode.ResourceExhausted"/>, and every accepted stream is bounded by a
///     linked max-lifetime CTS.
/// </remarks>
public sealed class TelemetryStreamGrpcService : GeneratedServiceBase
{
    /// <summary>Process-wide cap on concurrently-open telemetry streams.</summary>
    private const int MaxConcurrentStreams = 100;

    /// <summary>Upper bound on an accepted correlation id (defends against absurd inputs).</summary>
    private const int MaxCorrelationIdLength = 256;

    /// <summary>Per-subscriber bounded-channel capacity requested from the hub.</summary>
    private const int SubscriptionCapacity = 1000;

    /// <summary>Hard ceiling on a single stream's lifetime; the linked CTS cancels the pump when it fires.</summary>
    private static readonly TimeSpan MaxStreamLifetime = TimeSpan.FromHours(4);

    /// <summary>Process-wide count of currently-open streams; mutated only via <see cref="Interlocked"/>.</summary>
    private static int _activeStreams;

    private readonly ITelemetryLocalHub _hub;
    private readonly ILogger<TelemetryStreamGrpcService> _log;

    /// <summary>Initializes a new instance of the <see cref="TelemetryStreamGrpcService"/> class.</summary>
    /// <param name="hub">The node-local telemetry fan-out hub this stream drains.</param>
    /// <param name="log">Logger.</param>
    public TelemetryStreamGrpcService(ITelemetryLocalHub hub, ILogger<TelemetryStreamGrpcService> log)
    {
        _hub = hub;
        _log = log;
    }

    /// <inheritdoc />
    public override async Task Subscribe(
        TelemetryStreamRequest request,
        IServerStreamWriter<TelemetryEvent> responseStream,
        ServerCallContext context)
    {
        ArgumentNullException.ThrowIfNull(request);
        ArgumentNullException.ThrowIfNull(responseStream);
        ArgumentNullException.ThrowIfNull(context);

        var correlationId = request.CorrelationId;
        if (string.IsNullOrWhiteSpace(correlationId) || correlationId.Length > MaxCorrelationIdLength)
            throw new RpcException(new Status(
                StatusCode.InvalidArgument,
                "correlation_id must be non-empty and at most 256 characters"));

        // Concurrency cap. The increment is ALWAYS paired: on rejection we decrement here and throw
        // before entering the try; on acceptance the finally decrements exactly once. No path
        // decrements without a matching successful increment, and no path double-decrements.
        var active = Interlocked.Increment(ref _activeStreams);
        if (active > MaxConcurrentStreams)
        {
            Interlocked.Decrement(ref _activeStreams);
            _log.LogWarning(
                "Telemetry stream rejected (correlationId={CorrelationId}): {Active} open streams exceeds cap {Cap}",
                correlationId, active - 1, MaxConcurrentStreams);
            throw new RpcException(new Status(
                StatusCode.ResourceExhausted, "too many concurrent telemetry streams"));
        }

        try
        {
            using var lifetime = CancellationTokenSource.CreateLinkedTokenSource(context.CancellationToken);
            lifetime.CancelAfter(MaxStreamLifetime);
            using var sub = _hub.Subscribe(SubscriptionCapacity);

            _log.LogDebug("Telemetry stream opened (correlationId={CorrelationId})", correlationId);

            try
            {
                await foreach (var item in sub.Reader.ReadAllAsync(lifetime.Token).ConfigureAwait(false))
                    await responseStream
                        .WriteAsync(TelemetryProtoMapNode.ToProto(item, correlationId), lifetime.Token)
                        .ConfigureAwait(false);
            }
            catch (OperationCanceledException)
            {
                // Normal termination: the client disconnected (context token) or the max-lifetime cap
                // fired (linked token). Neither is an error — swallow and let the stream close cleanly.
            }
            catch (Exception ex) when (ex is IOException or RpcException)
            {
                // Routine mid-stream disconnect: central's TelemetryStreamClient reconnects on ANY
                // non-Cancelled error, so a broken pipe / connection-reset here is expected churn, not a
                // fault. Swallow at Debug so grpc-dotnet doesn't log it at Error on every reconnect.
                // (A genuinely unexpected exception still escapes to surface as an error.)
                _log.LogDebug(
                    ex, "Telemetry client disconnected mid-stream (correlationId={CorrelationId})", correlationId);
            }

            _log.LogDebug("Telemetry stream closed (correlationId={CorrelationId})", correlationId);
        }
        finally
        {
            Interlocked.Decrement(ref _activeStreams);
        }
    }
}