feat(mesh-phase5): node-side TelemetryStreamService (hub -> server-streaming)

Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW
2026-07-23 15:48:06 -04:00
parent 909d75357b
commit e742fee452
6 changed files with 601 additions and 0 deletions
@@ -0,0 +1,114 @@
+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.
+            }
+
+            _log.LogDebug("Telemetry stream closed (correlationId={CorrelationId})", correlationId);
+        }
+        finally
+        {
+            Interlocked.Decrement(ref _activeStreams);
+        }
+    }
+}