fix(mesh-phase5): dial supervisor — re-dial on endpoint change, pin no-restart invariant, throttle skip warnings

Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW
2026-07-23 16:43:09 -04:00
parent ffb75b7854
commit 84fa2e1e43
3 changed files with 91 additions and 13 deletions
@@ -73,6 +73,7 @@ public sealed class TelemetryDialSupervisor : ReceiveActor, IWithTimers
    private readonly ILoggingAdapter _log = Context.GetLogger();

    private readonly Dictionary<string, NodeDialer> _dialers = new(StringComparer.Ordinal);
+    private readonly HashSet<string> _warnedUnroutedTypes = new(StringComparer.Ordinal);
    private bool _pillConnected;

    /// <summary>Gets the timer scheduler driving the periodic refresh and per-node reconnect backoff.</summary>
@@ -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
    /// <summary>Per-node dialer bookkeeping. Mutated only on the actor thread.</summary>
    private sealed class NodeDialer(TelemetryDialTarget target)
    {
-        /// <summary>The node's dial target.</summary>
-        public TelemetryDialTarget Target { get; } = target;
+        /// <summary>The node's dial target. Reassigned in place when a known node's endpoint changes.</summary>
+        public TelemetryDialTarget Target { get; set; } = target;

        /// <summary>Monotonic stream generation; a message tagged with a stale value is dropped.</summary>
        public int Generation { get; set; }
@@ -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<string>(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;
                }

@@ -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<TelemetryDialTarget>
+        {
+            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]