diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Telemetry/TelemetryProtoMapCentral.cs b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Telemetry/TelemetryProtoMapCentral.cs index 462994b0..f851f761 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Telemetry/TelemetryProtoMapCentral.cs +++ b/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 , so the nullable domain fields map via /// ?.ToDateTime() and an absent Timestamp becomes . The /// non-nullable domain fields (published / sampled / transition timestamps) are always set by -/// the node mapper, so they map via a direct .ToDateTime() (which yields a +/// the node mapper, so they map via (which yields a /// value). /// /// +/// A required Timestamp is defended, not assumed. Across a mixed-version fleet an +/// out-of-contract sender could leave a required Timestamp unset. throws +/// a clear naming the kind + field instead of letting a +/// bare 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. +/// +/// /// optional-string presence is honoured. A proto3 optional string that the node /// left unset reads back as (via the generated Has… presence), /// distinguishing null from the empty string; likewise the optional bool @@ -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")); } /// Projects a onto a . @@ -126,7 +135,23 @@ public static class TelemetryProtoMapCentral ConsecutiveFailures: msg.ConsecutiveFailures, CurrentInFlight: msg.CurrentInFlight, LastBreakerOpenUtc: msg.LastBreakerOpenUtc?.ToDateTime(), - LastSampledUtc: msg.LastSampledUtc.ToDateTime(), - PublishedUtc: msg.PublishedUtc.ToDateTime()); + LastSampledUtc: Required(msg.LastSampledUtc, "DriverResilienceStatus", "last_sampled_utc"), + PublishedUtc: Required(msg.PublishedUtc, "DriverResilienceStatus", "published_utc")); } + + /// + /// Converts a required proto to a UTC , + /// throwing a clear naming the telemetry kind + proto + /// field if an out-of-contract sender left it unset — rather than letting a bare + /// escape. + /// + /// The required proto Timestamp ( when the sender omitted it). + /// The telemetry sub-message name, for the diagnostic. + /// The proto field name, for the diagnostic. + /// The Timestamp as a value. + /// is unset. + private static DateTime Required(Timestamp? ts, string kind, string field) => + ts is null + ? throw new InvalidOperationException($"telemetry {kind} missing required timestamp {field}") + : ts.ToDateTime(); } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Telemetry/TelemetryStreamClient.cs b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Telemetry/TelemetryStreamClient.cs index fb3acaa8..6240962b 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Telemetry/TelemetryStreamClient.cs +++ b/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; /// /// /// -/// Deliberately dumb. Connect, pump, surface errors — no reconnect, no backoff, no -/// buffering. A normal shutdown (the caller's firing, or the -/// server ending the stream with ) completes -/// without calling onError; any other failure is routed to -/// onError exactly once as the reconnect trigger the supervisor (a later task) owns. +/// Deliberately dumb. Connect, pump, surface transport errors — no reconnect, no +/// backoff, no buffering. A normal shutdown (the caller's +/// firing, or the server ending the stream with ) completes +/// without calling onError. +/// +/// +/// onError is strictly the transport-failure signal. Only a failure from the wire itself +/// — an exception thrown out of call.ResponseStream — reaches onError, 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: +/// +/// +/// Programming faults (null/empty correlationId, use-after-dispose) are +/// validated before the pump and throw synchronously out of +/// — a caller bug must surface as a caller bug, never as "node +/// down". +/// +/// +/// Consumer-callback faults (a poison/unmappable event throwing inside +/// onEvent — e.g. a future oneof case the mapper rejects, or a malformed required +/// Timestamp) are caught per event, logged at Warning, and the pump +/// continues. A code/version defect on one event must not tear down the stream +/// or trigger a reconnect. +/// +/// /// /// /// h2c + Bearer. The endpoint is a prior-knowledge http://host:port 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 _log; + private bool _disposed; /// /// Initializes a new instance of the class dialing a live @@ -39,12 +62,14 @@ public sealed class TelemetryStreamClient : IDisposable /// /// Prior-knowledge http://host:port address of the node's telemetry server. /// Shared node bearer key sent as authorization: Bearer <key>. - public TelemetryStreamClient(string endpoint, string apiKey) + /// Logger for dropped-malformed-event diagnostics; defaults to a no-op logger. + public TelemetryStreamClient(string endpoint, string apiKey, ILogger? log = null) { ArgumentException.ThrowIfNullOrEmpty(endpoint); ArgumentNullException.ThrowIfNull(apiKey); _apiKey = apiKey; + _log = log ?? NullLogger.Instance; _ownedChannel = GrpcChannel.ForAddress(endpoint, new GrpcChannelOptions { HttpHandler = new SocketsHttpHandler @@ -64,34 +89,47 @@ public sealed class TelemetryStreamClient : IDisposable /// /// The generated client to drive (typically a fake in tests). /// Shared node bearer key sent as authorization: Bearer <key>. - public TelemetryStreamClient(TelemetryStreamService.TelemetryStreamServiceClient client, string apiKey) + /// Logger for dropped-malformed-event diagnostics; defaults to a no-op logger. + public TelemetryStreamClient( + TelemetryStreamService.TelemetryStreamServiceClient client, + string apiKey, + ILogger? log = null) { ArgumentNullException.ThrowIfNull(client); ArgumentNullException.ThrowIfNull(apiKey); _client = client; _apiKey = apiKey; + _log = log ?? NullLogger.Instance; _ownedChannel = null; } /// /// Opens the telemetry stream and pumps every envelope to until the /// server ends the stream or fires. A normal shutdown returns without - /// invoking ; any other failure invokes - /// exactly once and returns. + /// invoking ; only a transport failure invokes + /// (exactly once). A consumer-callback exception on one event is + /// logged and the pump continues; programming faults throw synchronously (see the type remarks). /// /// Stream correlation id echoed on every envelope. /// Sink for each received raw . - /// Invoked once on an abnormal stream failure (the reconnect trigger). + /// Invoked once on an abnormal transport failure (the reconnect trigger). /// Cancels the stream (normal shutdown). + /// is null or empty. + /// or is null. + /// This client has been disposed. public async Task RunAsync( string correlationId, Action onEvent, Action 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,7 +139,22 @@ public sealed class TelemetryStreamClient : IDisposable new TelemetryStreamRequest { CorrelationId = correlationId }, headers, cancellationToken: ct); await foreach (var evt in call.ResponseStream.ReadAllAsync(ct)) - onEvent(evt); + { + // 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) { @@ -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); } } /// - public void Dispose() => _ownedChannel?.Dispose(); + public void Dispose() + { + _disposed = true; + _ownedChannel?.Dispose(); + } } diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Telemetry/TelemetryProtoMapCentralTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Telemetry/TelemetryProtoMapCentralTests.cs index d51ab2e5..f5602483 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Telemetry/TelemetryProtoMapCentralTests.cs +++ b/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(() => 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(() => 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(() => 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(() => TelemetryProtoMapCentral.ToResilience(proto)); + ex.Message.ShouldContain("DriverResilienceStatus"); + } + [Fact] public void MapEvent_routes_each_case_to_its_record_type() { diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Telemetry/TelemetryStreamClientTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Telemetry/TelemetryStreamClientTests.cs index 8335f6b4..3c1b6544 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Telemetry/TelemetryStreamClientTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Telemetry/TelemetryStreamClientTests.cs @@ -10,7 +10,9 @@ namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Tests.Telemetry; /// /// Unit tests for 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. /// 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(); + 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(); + 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(() => + 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(() => + 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 () => { }); } - /// Replays a fixed sequence of events, optionally throwing at the end. + /// + /// Replays a fixed sequence of events. Honors the pump's (so a + /// cancelled token surfaces as 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). + /// private sealed class FakeStreamReader : IAsyncStreamReader { private readonly IReadOnlyList _items; private readonly Exception? _throwAtEnd; + private readonly CancellationTokenSource? _cancelAfterFirst; private int _index = -1; - public FakeStreamReader(IReadOnlyList items, Exception? throwAtEnd = null) + public FakeStreamReader( + IReadOnlyList items, + Exception? throwAtEnd = null, + CancellationTokenSource? cancelAfterFirst = null) { _items = items; _throwAtEnd = throwAtEnd; + _cancelAfterFirst = cancelAfterFirst; } public TelemetryEvent Current => _items[_index]; public Task 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;