From 84fa2e1e43f194ab7af6b335fc9c69a9d481a0ab Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 23 Jul 2026 16:43:09 -0400 Subject: [PATCH] =?UTF-8?q?fix(mesh-phase5):=20dial=20supervisor=20?= =?UTF-8?q?=E2=80=94=20re-dial=20on=20endpoint=20change,=20pin=20no-restar?= =?UTF-8?q?t=20invariant,=20throttle=20skip=20warnings?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW --- .../Telemetry/TelemetryDialSupervisor.cs | 51 +++++++++++++++---- .../Telemetry/TelemetryNodeSource.cs | 15 ++++-- .../Telemetry/TelemetryDialSupervisorTests.cs | 38 ++++++++++++++ 3 files changed, 91 insertions(+), 13 deletions(-) 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 ca793e3f..d11b1f9d 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Telemetry/TelemetryDialSupervisor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Telemetry/TelemetryDialSupervisor.cs @@ -73,6 +73,7 @@ public sealed class TelemetryDialSupervisor : ReceiveActor, IWithTimers private readonly ILoggingAdapter _log = Context.GetLogger(); private readonly Dictionary _dialers = new(StringComparer.Ordinal); + private readonly HashSet _warnedUnroutedTypes = new(StringComparer.Ordinal); private bool _pillConnected; /// Gets the timer scheduler driving the periodic refresh and per-node reconnect backoff. @@ -195,15 +196,32 @@ public sealed class TelemetryDialSupervisor : ReceiveActor, IWithTimers } } - // Start a dialer for each newly-present node. Existing dialers keep running untouched. + // Start a dialer for each newly-present node; re-dial one whose endpoint moved. An existing + // node whose endpoint is UNCHANGED keeps its dialer untouched (no churn — a restart would drop + // its stream on every refresh). foreach (var (nodeId, target) in incoming) { - if (!_dialers.ContainsKey(nodeId)) + if (_dialers.TryGetValue(nodeId, out var existing)) { - var dialer = new NodeDialer(target); - _dialers[nodeId] = dialer; - StartDialer(nodeId); + if (!string.Equals(existing.Target.Endpoint, target.Endpoint, StringComparison.Ordinal)) + { + // A re-provisioned node moved host/port under the same id. Update the target in place + // and re-dial — StartDialer bumps the SAME dialer's generation, so the old (dead) + // endpoint's stream is cancelled AND its in-flight messages are dropped by the + // generation guard. Replacing the dialer object would reset the generation and let a + // late message from the old stream slip through. + _log.Info( + "Node {NodeId} telemetry endpoint changed {Old} -> {New}; re-dialing", + nodeId, existing.Target.Endpoint, target.Endpoint); + existing.Target = target; + StartDialer(nodeId); + } + + continue; } + + _dialers[nodeId] = new NodeDialer(target); + StartDialer(nodeId); } } @@ -363,9 +381,22 @@ public sealed class TelemetryDialSupervisor : ReceiveActor, IWithTimers _resilienceStore.Upsert(resilience); break; default: - _log.Warning( - "Telemetry record of unrouted type {RecordType} from node {NodeId}; dropped", - record?.GetType().Name ?? "null", nodeId); + // Warn once per unexpected type — a version-skew event would otherwise Warn per + // event, forever. Subsequent drops of the same type log at Debug. + var typeName = record?.GetType().Name ?? "null"; + if (_warnedUnroutedTypes.Add(typeName)) + { + _log.Warning( + "Telemetry record of unrouted type {RecordType} from node {NodeId}; dropped " + + "(further drops of this type log at Debug)", typeName, nodeId); + } + else + { + _log.Debug( + "Telemetry record of unrouted type {RecordType} from node {NodeId}; dropped", + typeName, nodeId); + } + break; } } @@ -405,8 +436,8 @@ public sealed class TelemetryDialSupervisor : ReceiveActor, IWithTimers /// Per-node dialer bookkeeping. Mutated only on the actor thread. private sealed class NodeDialer(TelemetryDialTarget target) { - /// The node's dial target. - public TelemetryDialTarget Target { get; } = target; + /// The node's dial target. Reassigned in place when a known node's endpoint changes. + public TelemetryDialTarget Target { get; set; } = target; /// Monotonic stream generation; a message tagged with a stale value is dropped. public int Generation { get; set; } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Telemetry/TelemetryNodeSource.cs b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Telemetry/TelemetryNodeSource.cs index 2a083f0b..2c465815 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Telemetry/TelemetryNodeSource.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Telemetry/TelemetryNodeSource.cs @@ -29,6 +29,10 @@ public static class TelemetryNodeSource ArgumentNullException.ThrowIfNull(dbFactory); ArgumentNullException.ThrowIfNull(logger); + // A persistently null-GrpcPort node is skipped on every refresh (every ContactRefreshSeconds); + // warn once per node id so it does not spam the log forever. + var warnedNullPort = new HashSet(StringComparer.Ordinal); + return async () => { await using var db = await dbFactory.CreateDbContextAsync().ConfigureAwait(false); @@ -44,9 +48,14 @@ public static class TelemetryNodeSource { if (row.GrpcPort is null) { - logger.LogWarning( - "ClusterNode {NodeId} has no GrpcPort; it exposes no telemetry stream and is " - + "skipped in this dial refresh", row.NodeId); + if (warnedNullPort.Add(row.NodeId)) + { + logger.LogWarning( + "ClusterNode {NodeId} has no GrpcPort; it exposes no telemetry stream and is " + + "skipped in this dial refresh (further skips of this node are silent)", + row.NodeId); + } + continue; } diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/Telemetry/TelemetryDialSupervisorTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/Telemetry/TelemetryDialSupervisorTests.cs index 0b24173f..4de5f16d 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/Telemetry/TelemetryDialSupervisorTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/Telemetry/TelemetryDialSupervisorTests.cs @@ -70,6 +70,44 @@ public sealed class TelemetryDialSupervisorTests : TestKit // node-3 gets a fresh dialer. var invocations = dial.WaitForCount(3, Timeout); invocations.ShouldContain(i => i.Target.NodeId == "node-3" && i.CorrelationId == "central-node-3-1"); + + // node-2 is present in BOTH refreshes with an unchanged endpoint: it must NOT be restarted, or + // its live stream would be dropped on every refresh. Exactly one dial invocation for it. + ExpectNoMsg(TimeSpan.FromMilliseconds(200)); + dial.Invocations.Count(i => i.Target.NodeId == "node-2").ShouldBe(1); + } + + [Fact] + public void Refresh_redials_a_node_whose_endpoint_changed_but_not_an_unchanged_one() + { + var dial = new FakeDialLoop(); + var current = new List + { + new("node-1", "http://old-host:5300"), + new("node-2", "http://h2:5300"), + }; + + var actor = Spawn(dial, () => current.ToArray()); + dial.WaitForCount(2, Timeout); + var node1Original = dial.Invocations.Single(i => i.Target.NodeId == "node-1"); + + // node-1 re-provisioned to a new host; node-2 unchanged. + current.Clear(); + current.Add(new TelemetryDialTarget("node-1", "http://new-host:5300")); + current.Add(new TelemetryDialTarget("node-2", "http://h2:5300")); + actor.Tell(new TelemetryDialSupervisor.RefreshNodes()); + + // The old node-1 stream is torn down and a fresh dialer opens at the NEW endpoint (gen 2). + AwaitCondition(() => node1Original.Ct.IsCancellationRequested, Timeout); + var invocations = dial.WaitForCount(3, Timeout); + invocations.ShouldContain(i => + i.Target.NodeId == "node-1" + && i.Target.Endpoint == "http://new-host:5300" + && i.CorrelationId == "central-node-1-2"); + + // node-2's endpoint did not change → still exactly one dial invocation (no churn). + ExpectNoMsg(TimeSpan.FromMilliseconds(200)); + dial.Invocations.Count(i => i.Target.NodeId == "node-2").ShouldBe(1); } [Fact]