dohertj2/lmxopcua
1
0
Fork 0
Code Issues 43 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(telemetry): wire TelemetryDial:CallTimeoutSeconds as the dial-establishment deadline (A5)

Browse Source 
CallTimeoutSeconds was bound, defaulted and documented but consumed
nowhere. TelemetryDialSupervisor now arms a per-generation deadline on
every dial: a dial that has delivered no event by the deadline is
cancelled and treated as a transport failure (reconnect with a failure
streak). A live node replays its snapshots on subscribe, so this bites
only a dial that never connects (a black-holed host, where the socket
connect has no timeout). An established stream is deliberately not
bounded — the wire has no heartbeat, keepalive pings own its liveness.

Review ref: 02/U2.

Refs #731

Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
This commit is contained in:
Joseph Doherty
2026-10-22 06:38:32 -04:00
parent 79d6b0a4b2
commit c6d4b14c93
3 changed files with 200 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Core/ZB.MOM.WW.OtOpcUa.Cluster/TelemetryOptions.cs
+7 -1
View File
@@ -72,7 +72,13 @@ public sealed class TelemetryDialOptions
/// <summary>How often central refreshes its set of dialable node contacts, in seconds.</summary>
public int ContactRefreshSeconds { get; set; } = 60;
/// <summary>Per-call deadline for a gRPC telemetry dial, in seconds.</summary>
/// <summary>
/// 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.
/// </summary>
public int CallTimeoutSeconds { get; set; } = 30;
}
src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Telemetry/TelemetryDialSupervisor.cs
+40
View File
@@ -129,6 +129,7 @@ public sealed class TelemetryDialSupervisor : ReceiveActor, IWithTimers
Receive<TelemetryReceived>(HandleTelemetryReceived);
Receive<StreamStopped>(HandleStreamStopped);
Receive<Reconnect>(HandleReconnect);
Receive<DialDeadline>(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
}
}
/// <summary>
/// <c>TelemetryDial:CallTimeoutSeconds</c>: a dial that has delivered no event by the deadline is cut and
/// treated as a transport failure (reconnect with a failure streak). <c>Connected</c> 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.
/// </summary>
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}";
/// <summary>Per-node dialer bookkeeping. Mutated only on the actor thread.</summary>
private sealed class NodeDialer(TelemetryDialTarget target)
{
@@ -496,4 +530,10 @@ public sealed class TelemetryDialSupervisor : ReceiveActor, IWithTimers
/// <summary>Backoff timer tick: start a fresh dialer for the node.</summary>
/// <param name="NodeId">The node id to reconnect.</param>
public sealed record Reconnect(string NodeId);
/// <summary>Dial-establishment deadline tick (<c>TelemetryDial:CallTimeoutSeconds</c>); see
/// <see cref="HandleDialDeadline"/>.</summary>
/// <param name="NodeId">The node whose dial is being checked.</param>
/// <param name="Generation">The dialer generation the deadline was armed for.</param>
public sealed record DialDeadline(string NodeId, int Generation);
}
tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/Telemetry/TelemetryDialDeadlineTests.cs
+153
View File
@@ -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;
/// <summary>
/// Review 02 U2: <c>TelemetryDial:CallTimeoutSeconds</c> was bound, defaulted and documented, and consumed
/// nowhere. It now bounds a dial's <b>establishment</b>: 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.
/// </summary>
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<IReadOnlyList<TelemetryDialTarget>>(
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();
}
/// <summary>Records each invocation; optionally delivers one health event on open; ends only when cancelled.</summary>
private sealed class FakeDialLoop
{
private readonly ConcurrentQueue<DialInvocation> _invocations = new();
public bool DeliverOnOpen { get; init; }
public IReadOnlyList<DialInvocation> 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<DialInvocation> 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<AlarmTransitionEvent> AlarmBroadcaster { get; } = new();
public NullBroadcaster<ScriptLogEntry> ScriptBroadcaster { get; } = new();
public RecordingHealthStore HealthStore { get; } = new();
public NullResilienceStore ResilienceStore { get; } = new();
}
private sealed class NullBroadcaster<T> : IInProcessBroadcaster<T>
{
public event Action<T>? Received;
public event Action<bool>? 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<DriverHealthChanged> Upserts { get; } = new();
public event Action<DriverHealthChanged>? 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<DriverHealthChanged> GetAll() => Upserts.ToArray();
}
private sealed class NullResilienceStore : IDriverResilienceStatusStore
{
public event Action<DriverResilienceStatusChanged>? SnapshotChanged;
public void Upsert(DriverResilienceStatusChanged snapshot) => SnapshotChanged?.Invoke(snapshot);
public IReadOnlyList<DriverResilienceStatusChanged> GetForInstance(string driverInstanceId) => Array.Empty<DriverResilienceStatusChanged>();
public IReadOnlyCollection<DriverResilienceStatusChanged> GetAll() => Array.Empty<DriverResilienceStatusChanged>();
}
}