From 2e70ef88aac78834b2f4e757448e741cb3cd81bb Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 23 Jul 2026 16:02:26 -0400 Subject: [PATCH] =?UTF-8?q?fix(mesh-phase5):=20telemetry=20service=20?= =?UTF-8?q?=E2=80=94=20graceful=20client-disconnect=20+=20null-safe=20requ?= =?UTF-8?q?ired-string=20mapping?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW --- .../Grpc/TelemetryProtoMapNode.cs | 34 +++++---- .../Grpc/TelemetryStreamGrpcService.cs | 9 +++ .../Grpc/TelemetryStreamGrpcServiceTests.cs | 70 +++++++++++++++++++ 3 files changed, 98 insertions(+), 15 deletions(-) diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Grpc/TelemetryProtoMapNode.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Grpc/TelemetryProtoMapNode.cs index 900ee4a2..8682a16f 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Grpc/TelemetryProtoMapNode.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Grpc/TelemetryProtoMapNode.cs @@ -67,17 +67,21 @@ public static class TelemetryProtoMapNode private static AlarmTransition MapAlarm(AlarmTransitionEvent e) { + // Required (non-optional) proto string setters throw ArgumentNullException on null, and the + // domain records carry no runtime null guard (nullable-ref annotations are compile-time only). + // Coalesce every required string to "" so a stray null can never throw out of the pump and kill + // the stream. The `optional` fields below keep their null-vs-"" presence guards. var msg = new AlarmTransition { - AlarmId = e.AlarmId, - EquipmentPath = e.EquipmentPath, - AlarmName = e.AlarmName, - TransitionKind = e.TransitionKind, + AlarmId = e.AlarmId ?? "", + EquipmentPath = e.EquipmentPath ?? "", + AlarmName = e.AlarmName ?? "", + TransitionKind = e.TransitionKind ?? "", Severity = e.Severity, - Message = e.Message, - User = e.User, + Message = e.Message ?? "", + User = e.User ?? "", TimestampUtc = ToUtcTimestamp(e.TimestampUtc), - AlarmTypeName = e.AlarmTypeName, + AlarmTypeName = e.AlarmTypeName ?? "", }; if (e.Comment is not null) @@ -93,9 +97,9 @@ public static class TelemetryProtoMapNode { var msg = new ScriptLog { - ScriptId = e.ScriptId, - Level = e.Level, - Message = e.Message, + ScriptId = e.ScriptId ?? "", + Level = e.Level ?? "", + Message = e.Message ?? "", TimestampUtc = ToUtcTimestamp(e.TimestampUtc), }; @@ -113,9 +117,9 @@ public static class TelemetryProtoMapNode { var msg = new DriverHealth { - ClusterId = e.ClusterId, - DriverInstanceId = e.DriverInstanceId, - State = e.State, + ClusterId = e.ClusterId ?? "", + DriverInstanceId = e.DriverInstanceId ?? "", + State = e.State ?? "", ErrorCount5Min = e.ErrorCount5Min, PublishedUtc = ToUtcTimestamp(e.PublishedUtc), }; @@ -132,8 +136,8 @@ public static class TelemetryProtoMapNode { var msg = new DriverResilienceStatus { - DriverInstanceId = e.DriverInstanceId, - HostName = e.HostName, + DriverInstanceId = e.DriverInstanceId ?? "", + HostName = e.HostName ?? "", BreakerOpen = e.BreakerOpen, ConsecutiveFailures = e.ConsecutiveFailures, CurrentInFlight = e.CurrentInFlight, diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Grpc/TelemetryStreamGrpcService.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Grpc/TelemetryStreamGrpcService.cs index 20dbc490..b2e7c634 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Grpc/TelemetryStreamGrpcService.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Grpc/TelemetryStreamGrpcService.cs @@ -103,6 +103,15 @@ public sealed class TelemetryStreamGrpcService : GeneratedServiceBase // 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); } diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Grpc/TelemetryStreamGrpcServiceTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Grpc/TelemetryStreamGrpcServiceTests.cs index f98e9c58..b3b60e21 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Grpc/TelemetryStreamGrpcServiceTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Grpc/TelemetryStreamGrpcServiceTests.cs @@ -210,6 +210,62 @@ public sealed class TelemetryStreamGrpcServiceTests evt.DriverHealth.LastSuccessfulReadUtc.ToDateTime().ShouldBe(expectedUtc); } + [Fact] + public async Task Client_disconnect_mid_stream_ends_cleanly_without_leaking_a_slot() + { + // A broken pipe surfaces as IOException out of WriteAsync. It must be swallowed as a routine + // disconnect (not escape), and the finally must still decrement the counter. Looping well past + // the cap proves no leak: a single leaked slot per pass would trip ResourceExhausted by pass 101. + var hub = new TelemetryLocalHub(); + var service = NewService(hub); + + // Cached snapshot ⇒ every fresh subscription is deterministically handed one item to write. + hub.Emit(new TelemetryItem.Health(new DriverHealthChanged( + "cluster-a", "drv-1", "Healthy", null, null, 0, DateTime.UtcNow))); + + for (var i = 0; i < 150; i++) + { + using var cts = new CancellationTokenSource(); + var pump = service.Subscribe( + new TelemetryStreamRequest { CorrelationId = $"disc-{i}" }, + new ThrowingStreamWriter(new IOException("connection reset")), + Ctx(cts.Token)); + + // No exception escapes Subscribe, and no ResourceExhausted ever fires (proves the slot is freed). + await Should.NotThrowAsync(() => pump); + } + + // Counter is back at its prior value: a normal stream still opens and pumps. + using var okCts = new CancellationTokenSource(); + var writer = new CapturingStreamWriter(); + var okPump = service.Subscribe( + new TelemetryStreamRequest { CorrelationId = "disc-after" }, writer, Ctx(okCts.Token)); + await WaitUntilAsync(() => writer.Count >= 1, "the cached snapshot to stream after the disconnect loop"); + okCts.Cancel(); + await okPump; + } + + [Fact] + public void ToProto_null_required_string_does_not_throw_and_maps_to_empty() + { + // Required proto string setters throw on null; the domain records carry no runtime guard. + var alarm = new AlarmTransitionEvent( + AlarmId: null!, EquipmentPath: null!, AlarmName: null!, TransitionKind: null!, + Severity: 100, Message: null!, User: null!, TimestampUtc: DateTime.UtcNow, + AlarmTypeName: null!, Comment: null, HistorizeToAveva: null, ReferencingEquipmentPaths: null); + + AlarmTransition msg = null!; + Should.NotThrow(() => msg = TelemetryProtoMapNode.ToProto(new TelemetryItem.Alarm(alarm), "c").AlarmTransition); + + msg.AlarmId.ShouldBe(""); + msg.EquipmentPath.ShouldBe(""); + msg.AlarmName.ShouldBe(""); + msg.TransitionKind.ShouldBe(""); + msg.Message.ShouldBe(""); + msg.User.ShouldBe(""); + msg.AlarmTypeName.ShouldBe(""); + } + [Fact] public void ToProto_alarm_omits_absent_nullables() { @@ -284,4 +340,18 @@ public sealed class TelemetryStreamGrpcServiceTests return [.. _items]; } } + + /// A stream writer that always throws the given exception — models a broken client pipe. + private sealed class ThrowingStreamWriter(Exception toThrow) : IServerStreamWriter + { + public WriteOptions? WriteOptions { get; set; } + + public Task WriteAsync(TelemetryEvent message) => throw toThrow; + + public Task WriteAsync(TelemetryEvent message, CancellationToken cancellationToken) + { + cancellationToken.ThrowIfCancellationRequested(); + return WriteAsync(message); + } + } }