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);
+ }
+ }
}