diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/TelemetryOptions.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/TelemetryOptions.cs index 779228b9..3e6f5143 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/TelemetryOptions.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/TelemetryOptions.cs @@ -72,7 +72,13 @@ public sealed class TelemetryDialOptions /// How often central refreshes its set of dialable node contacts, in seconds. public int ContactRefreshSeconds { get; set; } = 60; - /// Per-call deadline for a gRPC telemetry dial, in seconds. + /// + /// Deadline, in seconds, for a gRPC telemetry dial to deliver its first event. A live node answers a + /// fresh subscribe at once (it replays its health/resilience snapshots), so a dial still silent at the + /// deadline is one that never connected — it is cancelled and re-dialled. Bounds establishment only; + /// an established stream is kept alive by HTTP/2 keepalive pings, not by this value. Values below 1 + /// are treated as 1. + /// public int CallTimeoutSeconds { get; set; } = 30; } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Telemetry/TelemetryDialSupervisor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Telemetry/TelemetryDialSupervisor.cs index d11b1f9d..3d6af8fe 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Telemetry/TelemetryDialSupervisor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Telemetry/TelemetryDialSupervisor.cs @@ -129,6 +129,7 @@ public sealed class TelemetryDialSupervisor : ReceiveActor, IWithTimers Receive(HandleTelemetryReceived); Receive(HandleStreamStopped); Receive(HandleReconnect); + Receive(HandleDialDeadline); // A faulted node-source task pipes here. Log and wait for the next periodic refresh — a stale // node set is far better than crashing the supervisor and dropping every live dialer. @@ -294,6 +295,34 @@ public sealed class TelemetryDialSupervisor : ReceiveActor, IWithTimers } } + /// + /// TelemetryDial:CallTimeoutSeconds: a dial that has delivered no event by the deadline is cut and + /// treated as a transport failure (reconnect with a failure streak). Connected flips on the first + /// routed event, on this thread, so a dial that connected in time is left alone — the deadline bounds + /// ESTABLISHMENT only. A live node answers a fresh subscribe at once (the node-local hub replays its + /// health/resilience snapshots), so what this catches is a dial that never connects: a black-holed host, + /// where the socket connect itself has no timeout and the stream would otherwise hang forever without a + /// reconnect. An established stream is never bounded here — the wire has no heartbeat, keepalive pings own + /// its liveness, and an idle cut would churn every quiet node. + /// + private void HandleDialDeadline(DialDeadline msg) + { + if (!_dialers.TryGetValue(msg.NodeId, out var dialer) || msg.Generation != dialer.Generation || dialer.Connected) + { + return; + } + + _log.Warning( + "Telemetry dial to node {NodeId} ({Endpoint}) delivered nothing within {Deadline}s; cancelling it and re-dialing", + msg.NodeId, dialer.Target.Endpoint, DialDeadlineFor().TotalSeconds); + dialer.Cancel(); + HandleStreamStopped(new StreamStopped( + msg.NodeId, msg.Generation, + new TimeoutException($"no telemetry event within {DialDeadlineFor().TotalSeconds:F0}s of dialing"))); + } + + private TimeSpan DialDeadlineFor() => TimeSpan.FromSeconds(Math.Max(1, _options.CallTimeoutSeconds)); + private void StartDialer(string nodeId) { var dialer = _dialers[nodeId]; @@ -306,6 +335,8 @@ public sealed class TelemetryDialSupervisor : ReceiveActor, IWithTimers var cts = new CancellationTokenSource(); dialer.Cts = cts; dialer.Connected = false; + // One deadline per generation; a superseded tick is dropped by the generation guard in the handler. + Timers.StartSingleTimer(DeadlineTimerKey(nodeId), new DialDeadline(nodeId, generation), DialDeadlineFor()); var self = Self; var target = dialer.Target; @@ -357,6 +388,7 @@ public sealed class TelemetryDialSupervisor : ReceiveActor, IWithTimers { dialer.Cancel(); Timers.Cancel(ReconnectTimerKey(nodeId)); + Timers.Cancel(DeadlineTimerKey(nodeId)); RecomputePill(); } } @@ -433,6 +465,8 @@ public sealed class TelemetryDialSupervisor : ReceiveActor, IWithTimers private static string ReconnectTimerKey(string nodeId) => $"telemetry-reconnect-{nodeId}"; + private static string DeadlineTimerKey(string nodeId) => $"telemetry-dial-deadline-{nodeId}"; + /// Per-node dialer bookkeeping. Mutated only on the actor thread. private sealed class NodeDialer(TelemetryDialTarget target) { @@ -496,4 +530,10 @@ public sealed class TelemetryDialSupervisor : ReceiveActor, IWithTimers /// Backoff timer tick: start a fresh dialer for the node. /// The node id to reconnect. public sealed record Reconnect(string NodeId); + + /// Dial-establishment deadline tick (TelemetryDial:CallTimeoutSeconds); see + /// . + /// The node whose dial is being checked. + /// The dialer generation the deadline was armed for. + public sealed record DialDeadline(string NodeId, int Generation); } diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/Telemetry/TelemetryDialDeadlineTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/Telemetry/TelemetryDialDeadlineTests.cs new file mode 100644 index 00000000..d19506ca --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/Telemetry/TelemetryDialDeadlineTests.cs @@ -0,0 +1,153 @@ +using System.Collections.Concurrent; +using Akka.Actor; +using Akka.TestKit.Xunit2; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.AdminUI.Hubs; +using ZB.MOM.WW.OtOpcUa.AdminUI.Telemetry; +using ZB.MOM.WW.OtOpcUa.Cluster; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Alerts; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Drivers; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Logging; + +namespace ZB.MOM.WW.OtOpcUa.AdminUI.Tests.Telemetry; + +/// +/// Review 02 U2: TelemetryDial:CallTimeoutSeconds was bound, defaulted and documented, and consumed +/// nowhere. It now bounds a dial's establishment: a dial that has delivered no event by the deadline is +/// cut and re-dialled with a failure streak. A live node answers a fresh subscribe at once (the node-local hub +/// replays its health/resilience snapshots), so the deadline bites only a dial that never connects — a +/// black-holed host, where the underlying socket connect has no timeout of its own. An established stream +/// is deliberately NOT bounded: HTTP/2 keepalive pings own its liveness, and the wire has no heartbeat, so an +/// idle timeout would churn every quiet node. +/// +public sealed class TelemetryDialDeadlineTests : TestKit +{ + private static readonly TimeSpan Timeout = TimeSpan.FromSeconds(8); + + [Fact] + public void A_dial_that_delivers_no_event_by_the_deadline_is_cancelled_and_redialled() + { + var dial = new FakeDialLoop(); + Spawn(dial, callTimeoutSeconds: 1); + + var first = dial.WaitForCount(1, Timeout).Single(); + first.Ct.IsCancellationRequested.ShouldBeFalse("the deadline has not elapsed yet"); + + // The stalled dial is cut at the deadline... + AwaitCondition(() => first.Ct.IsCancellationRequested, Timeout); + // ...and the supervisor re-dials (a stalled dial is a transport failure, never a silent end). + var invocations = dial.WaitForCount(2, Timeout); + invocations[1].CorrelationId.ShouldBe("central-node-1-2"); + } + + [Fact] + public async Task A_dial_that_delivers_an_event_before_the_deadline_is_left_open() + { + var dial = new FakeDialLoop { DeliverOnOpen = true }; + var sinks = new Sinks(); + Spawn(dial, callTimeoutSeconds: 1, sinks); + + var first = dial.WaitForCount(1, Timeout).Single(); + AwaitCondition(() => sinks.HealthStore.Upserts.Count == 1, Timeout); + + // Positive control: well past the deadline, the established stream is untouched. + await Task.Delay(TimeSpan.FromSeconds(2.5)); + first.Ct.IsCancellationRequested.ShouldBeFalse("an established stream must never be cut by the dial deadline"); + dial.Invocations.Count.ShouldBe(1); + } + + private IActorRef Spawn(FakeDialLoop dial, int callTimeoutSeconds, Sinks? sinks = null) + { + sinks ??= new Sinks(); + var options = new TelemetryDialOptions { ContactRefreshSeconds = 3600, CallTimeoutSeconds = callTimeoutSeconds }; + return Sys.ActorOf(TelemetryDialSupervisor.Props( + () => Task.FromResult>( + new[] { new TelemetryDialTarget("node-1", "http://h1:5300") }), + dial.Loop, + sinks.AlarmBroadcaster, + sinks.ScriptBroadcaster, + sinks.HealthStore, + sinks.ResilienceStore, + options)); + } + + private static DriverHealthChanged Health(string id) => + new("C1", id, "Healthy", DateTime.UtcNow, null, 0, DateTime.UtcNow); + + private sealed class DialInvocation + { + private readonly TaskCompletionSource _completion = new(TaskCreationOptions.RunContinuationsAsynchronously); + public required string CorrelationId { get; init; } + public required CancellationToken Ct { get; init; } + public Task Completion => _completion.Task; + public void Complete() => _completion.TrySetResult(); + } + + /// Records each invocation; optionally delivers one health event on open; ends only when cancelled. + private sealed class FakeDialLoop + { + private readonly ConcurrentQueue _invocations = new(); + + public bool DeliverOnOpen { get; init; } + + public IReadOnlyList Invocations => _invocations.ToArray(); + + public TelemetryDialLoop Loop => async (_, correlationId, onMapped, _, ct) => + { + var invocation = new DialInvocation { CorrelationId = correlationId, Ct = ct }; + _invocations.Enqueue(invocation); + if (DeliverOnOpen) onMapped(Health("drv-1")); + await using var reg = ct.Register(() => invocation.Complete()); + await invocation.Completion; + }; + + public IReadOnlyList WaitForCount(int count, TimeSpan timeout) + { + var deadline = DateTime.UtcNow + timeout; + while (DateTime.UtcNow < deadline) + { + var snapshot = _invocations.ToArray(); + if (snapshot.Length >= count) return snapshot; + Thread.Sleep(10); + } + + throw new Xunit.Sdk.XunitException( + $"Expected at least {count} dial invocations within {timeout}, saw {_invocations.Count}."); + } + } + + private sealed class Sinks + { + public NullBroadcaster AlarmBroadcaster { get; } = new(); + public NullBroadcaster ScriptBroadcaster { get; } = new(); + public RecordingHealthStore HealthStore { get; } = new(); + public NullResilienceStore ResilienceStore { get; } = new(); + } + + private sealed class NullBroadcaster : IInProcessBroadcaster + { + public event Action? Received; + public event Action? ConnectionStateChanged; + public bool IsConnected { get; private set; } + public void Publish(T item) => Received?.Invoke(item); + public void SetConnected(bool connected) { IsConnected = connected; ConnectionStateChanged?.Invoke(connected); } + } + + private sealed class RecordingHealthStore : IDriverStatusSnapshotStore + { + public ConcurrentQueue Upserts { get; } = new(); + public event Action? SnapshotChanged; + public void Upsert(DriverHealthChanged snapshot) { Upserts.Enqueue(snapshot); SnapshotChanged?.Invoke(snapshot); } + public bool TryGet(string driverInstanceId, out DriverHealthChanged snapshot) { snapshot = null!; return false; } + public IReadOnlyCollection GetAll() => Upserts.ToArray(); + } + + private sealed class NullResilienceStore : IDriverResilienceStatusStore + { + public event Action? SnapshotChanged; + public void Upsert(DriverResilienceStatusChanged snapshot) => SnapshotChanged?.Invoke(snapshot); + public IReadOnlyList GetForInstance(string driverInstanceId) => Array.Empty(); + public IReadOnlyCollection GetAll() => Array.Empty(); + } +}