fix(mesh-phase5): harden telemetry client onError contract (validate/isolate/defend)

Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW

src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Telemetry/TelemetryProtoMapCentral.cs

@@ -1,3 +1,4 @@
+using Google.Protobuf.WellKnownTypes;
 using ZB.MOM.WW.OtOpcUa.Commons.Messages.Alerts;
 using ZB.MOM.WW.OtOpcUa.Commons.Messages.Drivers;
 using ZB.MOM.WW.OtOpcUa.Commons.Messages.Logging;
@@ -18,10 +19,18 @@ namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Telemetry;
 ///         generated property is <see langword="null"/>, so the nullable domain fields map via
 ///         <c>?.ToDateTime()</c> and an absent Timestamp becomes <see langword="null"/>. The
 ///         non-nullable domain fields (published / sampled / transition timestamps) are always set by
-///         the node mapper, so they map via a direct <c>.ToDateTime()</c> (which yields a
+///         the node mapper, so they map via <see cref="Required"/> (which yields a
 ///         <see cref="DateTimeKind.Utc"/> value).
 ///     </para>
 ///     <para>
+///         <b>A required Timestamp is defended, not assumed.</b> Across a mixed-version fleet an
+///         out-of-contract sender could leave a required Timestamp unset. <see cref="Required"/> throws
+///         a clear <see cref="InvalidOperationException"/> naming the kind + field instead of letting a
+///         bare <see cref="NullReferenceException"/> escape — the client's per-event isolation then logs
+///         it as a dropped malformed event and continues, because it is a code/version fault, never a
+///         transport fault.
+///     </para>
+///     <para>
 ///         <b>optional-string presence is honoured.</b> A proto3 <c>optional string</c> that the node
 ///         left unset reads back as <see langword="null"/> (via the generated <c>Has…</c> presence),
 ///         distinguishing null from the empty string; likewise the <c>optional bool</c>
@@ -71,7 +80,7 @@ public static class TelemetryProtoMapCentral
             Severity: msg.Severity,
             Message: msg.Message,
             User: msg.User,
-            TimestampUtc: msg.TimestampUtc.ToDateTime(),
+            TimestampUtc: Required(msg.TimestampUtc, "AlarmTransition", "timestamp_utc"),
             AlarmTypeName: msg.AlarmTypeName,
             Comment: msg.HasComment ? msg.Comment : null,
             HistorizeToAveva: msg.HasHistorizeToAveva ? msg.HistorizeToAveva : null,
@@ -89,7 +98,7 @@ public static class TelemetryProtoMapCentral
             ScriptId: msg.ScriptId,
             Level: msg.Level,
             Message: msg.Message,
-            TimestampUtc: msg.TimestampUtc.ToDateTime(),
+            TimestampUtc: Required(msg.TimestampUtc, "ScriptLog", "timestamp_utc"),
             VirtualTagId: msg.HasVirtualTagId ? msg.VirtualTagId : null,
             AlarmId: msg.HasAlarmId ? msg.AlarmId : null,
             EquipmentId: msg.HasEquipmentId ? msg.EquipmentId : null);
@@ -109,7 +118,7 @@ public static class TelemetryProtoMapCentral
             LastSuccessfulReadUtc: msg.LastSuccessfulReadUtc?.ToDateTime(),
             LastError: msg.HasLastError ? msg.LastError : null,
             ErrorCount5Min: msg.ErrorCount5Min,
-            PublishedUtc: msg.PublishedUtc.ToDateTime());
+            PublishedUtc: Required(msg.PublishedUtc, "DriverHealth", "published_utc"));
     }
 
     /// <summary>Projects a <see cref="DriverResilienceStatus"/> onto a <see cref="DriverResilienceStatusChanged"/>.</summary>
@@ -126,7 +135,23 @@ public static class TelemetryProtoMapCentral
             ConsecutiveFailures: msg.ConsecutiveFailures,
             CurrentInFlight: msg.CurrentInFlight,
             LastBreakerOpenUtc: msg.LastBreakerOpenUtc?.ToDateTime(),
-            LastSampledUtc: msg.LastSampledUtc.ToDateTime(),
+            LastSampledUtc: Required(msg.LastSampledUtc, "DriverResilienceStatus", "last_sampled_utc"),
-            PublishedUtc: msg.PublishedUtc.ToDateTime());
+            PublishedUtc: Required(msg.PublishedUtc, "DriverResilienceStatus", "published_utc"));
     }
+
+    /// <summary>
+    ///     Converts a <b>required</b> proto <see cref="Timestamp"/> to a UTC <see cref="DateTime"/>,
+    ///     throwing a clear <see cref="InvalidOperationException"/> naming the telemetry kind + proto
+    ///     field if an out-of-contract sender left it unset — rather than letting a bare
+    ///     <see cref="NullReferenceException"/> escape.
+    /// </summary>
+    /// <param name="ts">The required proto Timestamp (<see langword="null"/> when the sender omitted it).</param>
+    /// <param name="kind">The telemetry sub-message name, for the diagnostic.</param>
+    /// <param name="field">The proto field name, for the diagnostic.</param>
+    /// <returns>The Timestamp as a <see cref="DateTimeKind.Utc"/> value.</returns>
+    /// <exception cref="InvalidOperationException"><paramref name="ts"/> is unset.</exception>
+    private static DateTime Required(Timestamp? ts, string kind, string field) =>
+        ts is null
+            ? throw new InvalidOperationException($"telemetry {kind} missing required timestamp {field}")
+            : ts.ToDateTime();
 }

src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Telemetry/TelemetryStreamClient.cs

@@ -1,6 +1,7 @@
-using System.Net;
 using Grpc.Core;
 using Grpc.Net.Client;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Logging.Abstractions;
 using ZB.MOM.WW.OtOpcUa.Commons.Protos.Telemetry.V1;
 
 namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Telemetry;
@@ -14,11 +15,31 @@ namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Telemetry;
 /// </summary>
 /// <remarks>
 ///     <para>
-///         <b>Deliberately dumb.</b> Connect, pump, surface errors — no reconnect, no backoff, no
+///         <b>Deliberately dumb.</b> Connect, pump, surface transport errors — no reconnect, no
-///         buffering. A normal shutdown (the caller's <see cref="CancellationToken"/> firing, or the
+///         backoff, no buffering. A normal shutdown (the caller's <see cref="CancellationToken"/>
-///         server ending the stream with <see cref="StatusCode.Cancelled"/>) completes
+///         firing, or the server ending the stream with <see cref="StatusCode.Cancelled"/>) completes
-///         <see cref="RunAsync"/> without calling <c>onError</c>; any other failure is routed to
+///         <see cref="RunAsync"/> without calling <c>onError</c>.
-///         <c>onError</c> exactly once as the reconnect trigger the supervisor (a later task) owns.
+///     </para>
+///     <para>
+///         <b>onError is strictly the transport-failure signal.</b> Only a failure from the wire itself
+///         — an exception thrown out of <c>call.ResponseStream</c> — reaches <c>onError</c>, and exactly
+///         once, so the supervisor's reconnect loop reacts only to a genuinely unreachable/faulted node.
+///         Everything that is NOT a transport fault is kept off that path:
+///         <list type="bullet">
+///             <item>
+///                 <b>Programming faults</b> (null/empty <c>correlationId</c>, use-after-dispose) are
+///                 validated <em>before</em> the pump and throw synchronously out of
+///                 <see cref="RunAsync"/> — a caller bug must surface as a caller bug, never as "node
+///                 down".
+///             </item>
+///             <item>
+///                 <b>Consumer-callback faults</b> (a poison/unmappable event throwing inside
+///                 <c>onEvent</c> — e.g. a future oneof case the mapper rejects, or a malformed required
+///                 Timestamp) are caught <em>per event</em>, logged at Warning, and the pump
+///                 <em>continues</em>. A code/version defect on one event must not tear down the stream
+///                 or trigger a reconnect.
+///             </item>
+///         </list>
 ///     </para>
 ///     <para>
 ///         <b>h2c + Bearer.</b> The endpoint is a prior-knowledge <c>http://host:port</c> address (no
@@ -32,6 +53,8 @@ public sealed class TelemetryStreamClient : IDisposable
     private readonly string _apiKey;
     private readonly GrpcChannel? _ownedChannel;
     private readonly TelemetryStreamService.TelemetryStreamServiceClient _client;
+    private readonly ILogger<TelemetryStreamClient> _log;
+    private bool _disposed;
 
     /// <summary>
     ///     Initializes a new instance of the <see cref="TelemetryStreamClient"/> class dialing a live
@@ -39,12 +62,14 @@ public sealed class TelemetryStreamClient : IDisposable
     /// </summary>
     /// <param name="endpoint">Prior-knowledge <c>http://host:port</c> address of the node's telemetry server.</param>
     /// <param name="apiKey">Shared node bearer key sent as <c>authorization: Bearer &lt;key&gt;</c>.</param>
-    public TelemetryStreamClient(string endpoint, string apiKey)
+    /// <param name="log">Logger for dropped-malformed-event diagnostics; defaults to a no-op logger.</param>
+    public TelemetryStreamClient(string endpoint, string apiKey, ILogger<TelemetryStreamClient>? log = null)
     {
         ArgumentException.ThrowIfNullOrEmpty(endpoint);
         ArgumentNullException.ThrowIfNull(apiKey);
 
         _apiKey = apiKey;
+        _log = log ?? NullLogger<TelemetryStreamClient>.Instance;
         _ownedChannel = GrpcChannel.ForAddress(endpoint, new GrpcChannelOptions
         {
             HttpHandler = new SocketsHttpHandler
@@ -64,34 +89,47 @@ public sealed class TelemetryStreamClient : IDisposable
     /// </summary>
     /// <param name="client">The generated client to drive (typically a fake in tests).</param>
     /// <param name="apiKey">Shared node bearer key sent as <c>authorization: Bearer &lt;key&gt;</c>.</param>
-    public TelemetryStreamClient(TelemetryStreamService.TelemetryStreamServiceClient client, string apiKey)
+    /// <param name="log">Logger for dropped-malformed-event diagnostics; defaults to a no-op logger.</param>
+    public TelemetryStreamClient(
+        TelemetryStreamService.TelemetryStreamServiceClient client,
+        string apiKey,
+        ILogger<TelemetryStreamClient>? log = null)
     {
         ArgumentNullException.ThrowIfNull(client);
         ArgumentNullException.ThrowIfNull(apiKey);
 
         _client = client;
         _apiKey = apiKey;
+        _log = log ?? NullLogger<TelemetryStreamClient>.Instance;
         _ownedChannel = null;
     }
 
     /// <summary>
     ///     Opens the telemetry stream and pumps every envelope to <paramref name="onEvent"/> until the
     ///     server ends the stream or <paramref name="ct"/> fires. A normal shutdown returns without
-    ///     invoking <paramref name="onError"/>; any other failure invokes <paramref name="onError"/>
+    ///     invoking <paramref name="onError"/>; only a transport failure invokes
-    ///     exactly once and returns.
+    ///     <paramref name="onError"/> (exactly once). A consumer-callback exception on one event is
+    ///     logged and the pump continues; programming faults throw synchronously (see the type remarks).
     /// </summary>
     /// <param name="correlationId">Stream correlation id echoed on every envelope.</param>
     /// <param name="onEvent">Sink for each received raw <see cref="TelemetryEvent"/>.</param>
-    /// <param name="onError">Invoked once on an abnormal stream failure (the reconnect trigger).</param>
+    /// <param name="onError">Invoked once on an abnormal transport failure (the reconnect trigger).</param>
     /// <param name="ct">Cancels the stream (normal shutdown).</param>
+    /// <exception cref="ArgumentException"><paramref name="correlationId"/> is null or empty.</exception>
+    /// <exception cref="ArgumentNullException"><paramref name="onEvent"/> or <paramref name="onError"/> is null.</exception>
+    /// <exception cref="ObjectDisposedException">This client has been disposed.</exception>
     public async Task RunAsync(
         string correlationId,
         Action<TelemetryEvent> onEvent,
         Action<Exception> onError,
         CancellationToken ct)
     {
+        // Programming faults are validated BEFORE the pump so they throw synchronously out of RunAsync
+        // rather than funnelling to onError and masquerading as a transport fault (supervisor spin).
+        ArgumentException.ThrowIfNullOrEmpty(correlationId);
         ArgumentNullException.ThrowIfNull(onEvent);
         ArgumentNullException.ThrowIfNull(onError);
+        ObjectDisposedException.ThrowIf(_disposed, this);
 
         var headers = new Metadata { { "authorization", $"Bearer {_apiKey}" } };
 
@@ -101,8 +139,23 @@ public sealed class TelemetryStreamClient : IDisposable
                 new TelemetryStreamRequest { CorrelationId = correlationId }, headers, cancellationToken: ct);
 
             await foreach (var evt in call.ResponseStream.ReadAllAsync(ct))
+            {
+                // Isolate the consumer callback: a poison/unmappable event is a code/version fault, not
+                // "node down". Log and continue — never break the stream, never call onError.
+                try
+                {
                     onEvent(evt);
                 }
+                catch (Exception ex)
+                {
+                    _log.LogWarning(
+                        ex,
+                        "Dropping a malformed/unhandled telemetry event on stream {CorrelationId}; " +
+                        "the consumer callback threw. Continuing the stream (not a transport fault).",
+                        correlationId);
+                }
+            }
+        }
         catch (OperationCanceledException)
         {
             // Normal shutdown: the caller's token fired.
@@ -113,11 +166,15 @@ public sealed class TelemetryStreamClient : IDisposable
         }
         catch (Exception ex)
         {
-            // The reconnect trigger — a later supervisor task decides retry/backoff.
+            // The transport-failure signal — a later supervisor task decides retry/backoff.
             onError(ex);
         }
     }
 
     /// <inheritdoc />
-    public void Dispose() => _ownedChannel?.Dispose();
+    public void Dispose()
+    {
+        _disposed = true;
+        _ownedChannel?.Dispose();
+    }
 }

tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Telemetry/TelemetryProtoMapCentralTests.cs

@@ -233,6 +233,59 @@ public sealed class TelemetryProtoMapCentralTests
         TelemetryProtoMapCentral.ToResilience(proto).LastBreakerOpenUtc.ShouldBeNull();
     }
 
+    [Fact]
+    public void ToAlarm_missing_required_timestamp_throws_clear_InvalidOperation()
+    {
+        var proto = new AlarmTransition
+        {
+            AlarmId = "a", EquipmentPath = "p", AlarmName = "n", TransitionKind = "Activated",
+            Severity = 1, Message = "m", User = "system", AlarmTypeName = "AlarmCondition",
+            // TimestampUtc deliberately unset — an out-of-contract sender.
+        };
+
+        var ex = Should.Throw<InvalidOperationException>(() => TelemetryProtoMapCentral.ToAlarm(proto));
+        ex.Message.ShouldContain("AlarmTransition");
+        ex.Message.ShouldContain("timestamp_utc");
+    }
+
+    [Fact]
+    public void ToScript_missing_required_timestamp_throws_clear_InvalidOperation()
+    {
+        var proto = new ScriptLog { ScriptId = "s", Level = "Information", Message = "m" };
+
+        var ex = Should.Throw<InvalidOperationException>(() => TelemetryProtoMapCentral.ToScript(proto));
+        ex.Message.ShouldContain("ScriptLog");
+        ex.Message.ShouldContain("timestamp_utc");
+    }
+
+    [Fact]
+    public void ToHealth_missing_required_timestamp_throws_clear_InvalidOperation()
+    {
+        var proto = new DriverHealth
+        {
+            ClusterId = "c", DriverInstanceId = "d", State = "Healthy", ErrorCount5Min = 0,
+            // PublishedUtc unset.
+        };
+
+        var ex = Should.Throw<InvalidOperationException>(() => TelemetryProtoMapCentral.ToHealth(proto));
+        ex.Message.ShouldContain("DriverHealth");
+        ex.Message.ShouldContain("published_utc");
+    }
+
+    [Fact]
+    public void ToResilience_missing_required_timestamp_throws_clear_InvalidOperation()
+    {
+        var proto = new DriverResilienceStatus
+        {
+            DriverInstanceId = "d", HostName = "h", BreakerOpen = false,
+            ConsecutiveFailures = 0, CurrentInFlight = 0,
+            // LastSampledUtc + PublishedUtc unset.
+        };
+
+        var ex = Should.Throw<InvalidOperationException>(() => TelemetryProtoMapCentral.ToResilience(proto));
+        ex.Message.ShouldContain("DriverResilienceStatus");
+    }
+
     [Fact]
     public void MapEvent_routes_each_case_to_its_record_type()
     {

tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Telemetry/TelemetryStreamClientTests.cs

@@ -10,7 +10,9 @@ namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Tests.Telemetry;
 /// <summary>
 ///     Unit tests for <see cref="TelemetryStreamClient"/> driving a fake generated client (the ctor
 ///     seam), so the pump/normal-shutdown/error-routing behaviour is provable without a live gRPC
-///     server. The wire itself is covered by the Phase 5 live gate.
+///     server. The wire itself is covered by the Phase 5 live gate. The central assertion throughout is
+///     the onError contract: only a transport fault reaches onError; programming faults throw
+///     synchronously and consumer-callback faults are isolated.
 /// </summary>
 public sealed class TelemetryStreamClientTests
 {
@@ -53,6 +55,27 @@ public sealed class TelemetryStreamClientTests
         errored.ShouldBeNull();
     }
 
+    [Fact]
+    public async Task RunAsync_cancelled_token_ends_without_calling_onError()
+    {
+        // The primary shutdown branch the supervisor's routine reconnect-teardown drives: the token
+        // fires mid-stream, the transport surfaces an OperationCanceledException, and RunAsync returns
+        // cleanly — never onError.
+        using var cts = new CancellationTokenSource();
+        var fake = new FakeClient(new FakeStreamReader(
+            [new TelemetryEvent { ScriptLog = Script("one") }],
+            cancelAfterFirst: cts));
+        using var sut = new TelemetryStreamClient(fake, "key");
+
+        var received = new List<TelemetryEvent>();
+        Exception? errored = null;
+
+        await sut.RunAsync("corr-1", received.Add, ex => errored = ex, cts.Token);
+
+        received.Count.ShouldBe(1);
+        errored.ShouldBeNull();
+    }
+
     [Fact]
     public async Task RunAsync_routes_real_rpc_failure_to_onError()
     {
@@ -67,6 +90,65 @@ public sealed class TelemetryStreamClientTests
         errored.ShouldBeSameAs(boom);
     }
 
+    [Fact]
+    public async Task RunAsync_onEvent_exception_is_isolated_pump_continues_and_onError_not_called()
+    {
+        var events = new[]
+        {
+            new TelemetryEvent { ScriptLog = Script("poison") },
+            new TelemetryEvent { ScriptLog = Script("healthy") },
+        };
+        var fake = new FakeClient(new FakeStreamReader(events));
+        using var sut = new TelemetryStreamClient(fake, "key");
+
+        var received = new List<string>();
+        Exception? errored = null;
+
+        await sut.RunAsync(
+            "corr-1",
+            evt =>
+            {
+                if (evt.ScriptLog.Message == "poison")
+                    throw new InvalidOperationException("unmappable event");
+                received.Add(evt.ScriptLog.Message);
+            },
+            ex => errored = ex,
+            CancellationToken.None);
+
+        // The poison event was dropped, the pump continued to the next event, and onError never fired.
+        received.ShouldBe(["healthy"]);
+        errored.ShouldBeNull();
+    }
+
+    [Fact]
+    public async Task RunAsync_empty_correlationId_throws_synchronously_and_never_calls_onError()
+    {
+        var fake = new FakeClient(new FakeStreamReader([]));
+        using var sut = new TelemetryStreamClient(fake, "key");
+
+        Exception? errored = null;
+
+        await Should.ThrowAsync<ArgumentException>(() =>
+            sut.RunAsync("", _ => { }, ex => errored = ex, CancellationToken.None));
+
+        errored.ShouldBeNull();
+    }
+
+    [Fact]
+    public async Task RunAsync_after_dispose_throws_ObjectDisposed_and_never_calls_onError()
+    {
+        var fake = new FakeClient(new FakeStreamReader([]));
+        var sut = new TelemetryStreamClient(fake, "key");
+        sut.Dispose();
+
+        Exception? errored = null;
+
+        await Should.ThrowAsync<ObjectDisposedException>(() =>
+            sut.RunAsync("corr-1", _ => { }, ex => errored = ex, CancellationToken.None));
+
+        errored.ShouldBeNull();
+    }
+
     private static ScriptLog Script(string message) => new()
     {
         ScriptId = "s",
@@ -95,26 +177,44 @@ public sealed class TelemetryStreamClientTests
                 () => { });
     }
 
-    /// <summary>Replays a fixed sequence of events, optionally throwing at the end.</summary>
+    /// <summary>
+    ///     Replays a fixed sequence of events. Honors the pump's <see cref="CancellationToken"/> (so a
+    ///     cancelled token surfaces as <see cref="OperationCanceledException"/> just as the real
+    ///     transport does), can throw a supplied exception at the end of the sequence, and can cancel a
+    ///     supplied source after delivering the first item (to drive the mid-stream cancel branch).
+    /// </summary>
     private sealed class FakeStreamReader : IAsyncStreamReader<TelemetryEvent>
     {
         private readonly IReadOnlyList<TelemetryEvent> _items;
         private readonly Exception? _throwAtEnd;
+        private readonly CancellationTokenSource? _cancelAfterFirst;
         private int _index = -1;
 
-        public FakeStreamReader(IReadOnlyList<TelemetryEvent> items, Exception? throwAtEnd = null)
+        public FakeStreamReader(
+            IReadOnlyList<TelemetryEvent> items,
+            Exception? throwAtEnd = null,
+            CancellationTokenSource? cancelAfterFirst = null)
         {
             _items = items;
             _throwAtEnd = throwAtEnd;
+            _cancelAfterFirst = cancelAfterFirst;
         }
 
         public TelemetryEvent Current => _items[_index];
 
         public Task<bool> MoveNext(CancellationToken cancellationToken)
         {
+            cancellationToken.ThrowIfCancellationRequested();
+
             _index++;
             if (_index < _items.Count)
+            {
+                // After delivering the first item, request cancellation so the NEXT MoveNext throws
+                // OperationCanceledException at the top — the transport-cancel shape.
+                if (_index == 0)
+                    _cancelAfterFirst?.Cancel();
                 return Task.FromResult(true);
+            }
 
             if (_throwAtEnd is not null)
                 throw _throwAtEnd;