diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Telemetry/TelemetryDialSupervisor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Telemetry/TelemetryDialSupervisor.cs
new file mode 100644
index 00000000..ca793e3f
--- /dev/null
+++ b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Telemetry/TelemetryDialSupervisor.cs
@@ -0,0 +1,468 @@
+using Akka.Actor;
+using Akka.Event;
+using ZB.MOM.WW.OtOpcUa.AdminUI.Hubs;
+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.Telemetry;
+
+///
+/// A single dialable driver node: its logical id and the prior-knowledge
+/// http://host:port gRPC telemetry endpoint central connects to.
+///
+/// ClusterNode.NodeId ).
+/// http://{Host}:{GrpcPort} h2c telemetry endpoint.
+public sealed record TelemetryDialTarget(string NodeId, string Endpoint);
+
+///
+/// The dial-loop seam: opens one telemetry stream to TelemetryStreamClient + TelemetryProtoMapCentral ; tests
+/// substitute a fake that captures the callbacks and drives events/failures by hand.
+///
+/// A task that completes when the stream ends or is cancelled.
+public delegate Task TelemetryDialLoop(
+ TelemetryDialTarget target,
+ string correlationId,
+ Action onMapped,
+ Action onError,
+ CancellationToken ct);
+
+///
+/// Central-side supervisor of the Phase 5 telemetry dials. Maintains one reconnecting dialer per
+/// enabled, non-maintenance driver node and feeds every received record to the same four AdminUI
+/// in-process sinks the DPS bridges feed today — so flipping central from
+/// TelemetryDial:Mode = Dps to Grpc is transparent to the Blazor panels.
+///
+///
+///
+/// Discover → dial → reconnect, forever. On a periodic refresh the supervisor resolves
+/// the current dialable node set (via the injected
+///
+/// All state mutation is on the actor thread. The background dialer tasks only ever
+/// Self.Tell — they never touch the dialer table, the connection pill, or a generation
+/// counter directly. Each dialer carries a monotonic generation ; a message tagged with
+/// a stale generation (a late event or a late failure from a superseded stream) is dropped, so a
+/// reconnect can neither double-fire nor route a stale record.
+///
+///
+public sealed class TelemetryDialSupervisor : ReceiveActor, IWithTimers
+{
+ private const string RefreshTimerKey = "telemetry-node-refresh";
+ private static readonly TimeSpan ReconnectBackoff = TimeSpan.FromSeconds(5);
+
+ private readonly Func>> _nodeSource;
+ private readonly TelemetryDialLoop _dialLoop;
+ private readonly IInProcessBroadcaster _alarmBroadcaster;
+ private readonly IInProcessBroadcaster _scriptBroadcaster;
+ private readonly IDriverStatusSnapshotStore _healthStore;
+ private readonly IDriverResilienceStatusStore _resilienceStore;
+ private readonly TelemetryDialOptions _options;
+ private readonly ILoggingAdapter _log = Context.GetLogger();
+
+ private readonly Dictionary _dialers = new(StringComparer.Ordinal);
+ private bool _pillConnected;
+
+ /// Gets the timer scheduler driving the periodic refresh and per-node reconnect backoff.
+ public ITimerScheduler Timers { get; set; } = null!;
+
+ /// Creates the props for the telemetry dial supervisor.
+ /// The props.
+ public static Props Props(
+ Func>> nodeSource,
+ TelemetryDialLoop dialLoop,
+ IInProcessBroadcaster alarmBroadcaster,
+ IInProcessBroadcaster scriptBroadcaster,
+ IDriverStatusSnapshotStore healthStore,
+ IDriverResilienceStatusStore resilienceStore,
+ TelemetryDialOptions options) =>
+ Akka.Actor.Props.Create(() => new TelemetryDialSupervisor(
+ nodeSource, dialLoop, alarmBroadcaster, scriptBroadcaster, healthStore, resilienceStore, options));
+
+ /// Initializes a new instance of the
+ /// >> nodeSource,
+ TelemetryDialLoop dialLoop,
+ IInProcessBroadcaster alarmBroadcaster,
+ IInProcessBroadcaster scriptBroadcaster,
+ IDriverStatusSnapshotStore healthStore,
+ IDriverResilienceStatusStore resilienceStore,
+ TelemetryDialOptions options)
+ {
+ _nodeSource = nodeSource;
+ _dialLoop = dialLoop;
+ _alarmBroadcaster = alarmBroadcaster;
+ _scriptBroadcaster = scriptBroadcaster;
+ _healthStore = healthStore;
+ _resilienceStore = resilienceStore;
+ _options = options;
+
+ Receive(_ => HandleRefreshNodes());
+ Receive(HandleNodesResolved);
+ Receive(HandleTelemetryReceived);
+ Receive(HandleStreamStopped);
+ Receive(HandleReconnect);
+
+ // 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.
+ Receive(f => _log.Warning(
+ f.Cause,
+ "Telemetry node source failed; the dialable node set was NOT refreshed and may be stale. "
+ + "Existing dialers keep running; the next periodic refresh retries"));
+ }
+
+ /// > task;
+ try
+ {
+ task = _nodeSource();
+ }
+ catch (Exception ex)
+ {
+ _log.Warning(ex, "Telemetry node source threw synchronously; skipping this refresh");
+ return;
+ }
+
+ task.PipeTo(
+ Self,
+ success: targets => new NodesResolved(targets),
+ failure: ex => new Status.Failure(ex));
+ }
+
+ private void HandleNodesResolved(NodesResolved msg)
+ {
+ var incoming = new Dictionary(StringComparer.Ordinal);
+ foreach (var target in msg.Targets)
+ {
+ // Last write wins on a duplicate id; a malformed set is the node source's problem, not ours.
+ incoming[target.NodeId] = target;
+ }
+
+ // Drop dialers whose node is no longer present.
+ foreach (var nodeId in _dialers.Keys.ToList())
+ {
+ if (!incoming.ContainsKey(nodeId))
+ {
+ StopDialer(nodeId);
+ }
+ }
+
+ // Start a dialer for each newly-present node. Existing dialers keep running untouched.
+ foreach (var (nodeId, target) in incoming)
+ {
+ if (!_dialers.ContainsKey(nodeId))
+ {
+ var dialer = new NodeDialer(target);
+ _dialers[nodeId] = dialer;
+ StartDialer(nodeId);
+ }
+ }
+ }
+
+ private void HandleTelemetryReceived(TelemetryReceived msg)
+ {
+ if (!_dialers.TryGetValue(msg.NodeId, out var dialer) || msg.Generation != dialer.Generation)
+ {
+ // Superseded stream (a reconnect already advanced the generation) — drop, never route.
+ return;
+ }
+
+ if (!dialer.Connected)
+ {
+ dialer.Connected = true;
+ dialer.FailureStreak = 0;
+ RecomputePill();
+ }
+
+ Route(msg.NodeId, msg.Record);
+ }
+
+ private void HandleStreamStopped(StreamStopped msg)
+ {
+ if (!_dialers.TryGetValue(msg.NodeId, out var dialer) || msg.Generation != dialer.Generation)
+ {
+ // Late failure/end from a superseded stream — ignore, so a reconnect never double-fires.
+ return;
+ }
+
+ if (dialer.Connected)
+ {
+ dialer.Connected = false;
+ RecomputePill();
+ }
+
+ dialer.FailureStreak++;
+
+ if (msg.Error is not null && (dialer.FailureStreak == 1 || dialer.FailureStreak % 10 == 0))
+ {
+ _log.Warning(
+ msg.Error,
+ "Telemetry stream to node {NodeId} failed (failure streak {Streak}); reconnecting",
+ msg.NodeId, dialer.FailureStreak);
+ }
+ else
+ {
+ _log.Debug(
+ "Telemetry stream to node {NodeId} stopped (failure streak {Streak}); reconnecting",
+ msg.NodeId, dialer.FailureStreak);
+ }
+
+ if (dialer.FailureStreak == 1)
+ {
+ // First retry after a healthy stream is immediate.
+ StartDialer(msg.NodeId);
+ }
+ else
+ {
+ // Subsequent retries back off. A single timer per node — a queued reconnect is replaced.
+ Timers.StartSingleTimer(ReconnectTimerKey(msg.NodeId), new Reconnect(msg.NodeId), ReconnectBackoff);
+ }
+ }
+
+ private void HandleReconnect(Reconnect msg)
+ {
+ // The node may have been removed while the backoff timer was pending.
+ if (_dialers.ContainsKey(msg.NodeId))
+ {
+ StartDialer(msg.NodeId);
+ }
+ }
+
+ private void StartDialer(string nodeId)
+ {
+ var dialer = _dialers[nodeId];
+
+ // Cancel any prior stream and advance the generation so its in-flight signals are ignored.
+ dialer.Cancel();
+ Timers.Cancel(ReconnectTimerKey(nodeId));
+
+ var generation = ++dialer.Generation;
+ var cts = new CancellationTokenSource();
+ dialer.Cts = cts;
+ dialer.Connected = false;
+
+ var self = Self;
+ var target = dialer.Target;
+ var token = cts.Token;
+ var correlationId = $"central-{nodeId}-{generation}";
+ var dialLoop = _dialLoop;
+
+ // The background task ONLY ever Self.Tell()s — it never touches actor state.
+ _ = Task.Run(async () =>
+ {
+ var faulted = false;
+ try
+ {
+ await dialLoop(
+ target,
+ correlationId,
+ onMapped: record => self.Tell(new TelemetryReceived(nodeId, generation, record)),
+ onError: error =>
+ {
+ faulted = true;
+ self.Tell(new StreamStopped(nodeId, generation, error));
+ },
+ token).ConfigureAwait(false);
+ }
+ catch (OperationCanceledException)
+ {
+ // Normal shutdown: our cancellation token fired. No reconnect.
+ return;
+ }
+ catch (Exception ex)
+ {
+ // The dial loop threw instead of routing to onError — treat as a transport failure.
+ self.Tell(new StreamStopped(nodeId, generation, ex));
+ return;
+ }
+
+ // The loop returned. If it did not already fault and we did not cancel it, the server ended
+ // the stream — reconnect. (A faulted loop already told StreamStopped with the error.)
+ if (!faulted && !token.IsCancellationRequested)
+ {
+ self.Tell(new StreamStopped(nodeId, generation, Error: null));
+ }
+ });
+ }
+
+ private void StopDialer(string nodeId)
+ {
+ if (_dialers.Remove(nodeId, out var dialer))
+ {
+ dialer.Cancel();
+ Timers.Cancel(ReconnectTimerKey(nodeId));
+ RecomputePill();
+ }
+ }
+
+ private void Route(string nodeId, object record)
+ {
+ // Defensive: one poison record must never crash the supervisor and take every dialer with it.
+ try
+ {
+ switch (record)
+ {
+ case AlarmTransitionEvent alarm:
+ _alarmBroadcaster.Publish(alarm);
+ break;
+ case ScriptLogEntry script:
+ _scriptBroadcaster.Publish(script);
+ break;
+ case DriverHealthChanged health:
+ _healthStore.Upsert(health);
+ break;
+ case DriverResilienceStatusChanged resilience:
+ _resilienceStore.Upsert(resilience);
+ break;
+ default:
+ _log.Warning(
+ "Telemetry record of unrouted type {RecordType} from node {NodeId}; dropped",
+ record?.GetType().Name ?? "null", nodeId);
+ break;
+ }
+ }
+ catch (Exception ex)
+ {
+ _log.Warning(
+ ex,
+ "Failed to route a telemetry record of type {RecordType} from node {NodeId} to its sink",
+ record?.GetType().Name ?? "null", nodeId);
+ }
+ }
+
+ private void RecomputePill()
+ {
+ var anyConnected = false;
+ foreach (var dialer in _dialers.Values)
+ {
+ if (dialer.Connected)
+ {
+ anyConnected = true;
+ break;
+ }
+ }
+
+ if (anyConnected == _pillConnected)
+ {
+ return;
+ }
+
+ _pillConnected = anyConnected;
+ _alarmBroadcaster.SetConnected(anyConnected);
+ _scriptBroadcaster.SetConnected(anyConnected);
+ }
+
+ private static string ReconnectTimerKey(string nodeId) => $"telemetry-reconnect-{nodeId}";
+
+ /// 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;
+
+ /// Monotonic stream generation; a message tagged with a stale value is dropped.
+ public int Generation { get; set; }
+
+ /// Cancels the current stream's background task.
+ public CancellationTokenSource? Cts { get; set; }
+
+ /// Whether the current stream has delivered at least one event.
+ public bool Connected { get; set; }
+
+ /// Consecutive stream stops since the last successful event (drives immediate vs. backoff).
+ public int FailureStreak { get; set; }
+
+ /// Cancels + disposes the current stream's token source, if any.
+ public void Cancel()
+ {
+ var cts = Cts;
+ Cts = null;
+ if (cts is null)
+ {
+ return;
+ }
+
+ try
+ {
+ cts.Cancel();
+ }
+ catch (ObjectDisposedException)
+ {
+ // Already disposed — nothing to cancel.
+ }
+
+ cts.Dispose();
+ }
+ }
+
+ /// Timer tick: re-resolve the dialable node set.
+ public sealed record RefreshNodes;
+
+ /// The node set resolved off the actor thread.
+ /// Targets);
+
+ /// A mapped domain record received from a node's stream.
+ /// A node's stream stopped — transport failure (
+ /// Backoff timer tick: start a fresh dialer for the node.
+ ///
+/// Production seams for
+public static class TelemetryNodeSource
+{
+ ///
+ /// Builds the node source the supervisor polls: enabled, non-maintenance ClusterNode
+ /// rows carrying a non-null GrpcPort , projected to http://{Host}:{GrpcPort}
+ /// h2c endpoints. A row with a null GrpcPort exposes no telemetry surface yet and is
+ /// skipped with a Warning — mirrors CentralCommunicationActor.LoadContactsFromDb 's
+ /// enabled + non-maintenance read.
+ ///
+ /// ClusterNode rows.
+ /// An async delegate resolving the current dialable node set.
+ public static Func>> Create(
+ IDbContextFactory dbFactory,
+ ILogger logger)
+ {
+ ArgumentNullException.ThrowIfNull(dbFactory);
+ ArgumentNullException.ThrowIfNull(logger);
+
+ return async () =>
+ {
+ await using var db = await dbFactory.CreateDbContextAsync().ConfigureAwait(false);
+ var rows = await db.ClusterNodes
+ .AsNoTracking()
+ .Where(n => n.Enabled && !n.MaintenanceMode)
+ .Select(n => new { n.NodeId, n.Host, n.GrpcPort })
+ .ToListAsync()
+ .ConfigureAwait(false);
+
+ var targets = new List(rows.Count);
+ foreach (var row in rows)
+ {
+ 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);
+ continue;
+ }
+
+ targets.Add(new TelemetryDialTarget(row.NodeId, $"http://{row.Host}:{row.GrpcPort}"));
+ }
+
+ return (IReadOnlyList)targets;
+ };
+ }
+
+ ///
+ /// Builds the production dial loop: one onError , so the supervisor sees a poison event as "dropped",
+ /// never as "node down".
+ ///
+ /// The dial-loop delegate the supervisor invokes per generation.
+ public static TelemetryDialLoop CreateDialLoop(
+ string apiKey,
+ ILogger? clientLogger)
+ {
+ ArgumentNullException.ThrowIfNull(apiKey);
+
+ return async (target, correlationId, onMapped, onError, ct) =>
+ {
+ using var client = new TelemetryStreamClient(target.Endpoint, apiKey, clientLogger);
+ await client.RunAsync(
+ correlationId,
+ onEvent: evt => onMapped(TelemetryProtoMapCentral.MapEvent(evt)),
+ onError: onError,
+ ct).ConfigureAwait(false);
+ };
+ }
+}
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
new file mode 100644
index 00000000..0b24173f
--- /dev/null
+++ b/tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/Telemetry/TelemetryDialSupervisorTests.cs
@@ -0,0 +1,433 @@
+using System.Collections.Concurrent;
+using Akka.Actor;
+using Akka.TestKit.Xunit2;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.Extensions.Logging.Abstractions;
+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;
+using ZB.MOM.WW.OtOpcUa.Configuration;
+using ZB.MOM.WW.OtOpcUa.Configuration.Entities;
+
+namespace ZB.MOM.WW.OtOpcUa.AdminUI.Tests.Telemetry;
+
+///
+/// Unit tests for
+public sealed class TelemetryDialSupervisorTests : TestKit
+{
+ private static readonly TimeSpan Timeout = TimeSpan.FromSeconds(5);
+
+ [Fact]
+ public void Discovery_starts_one_dialer_per_target_with_distinct_correlation_ids()
+ {
+ var dial = new FakeDialLoop();
+ var targets = new[]
+ {
+ new TelemetryDialTarget("node-1", "http://h1:5300"),
+ new TelemetryDialTarget("node-2", "http://h2:5300"),
+ };
+
+ Spawn(dial, () => targets);
+
+ var invocations = dial.WaitForCount(2, Timeout);
+ invocations.Select(i => i.Target.NodeId).ShouldBe(new[] { "node-1", "node-2" }, ignoreOrder: true);
+ invocations.ShouldContain(i => i.CorrelationId == "central-node-1-1");
+ invocations.ShouldContain(i => i.CorrelationId == "central-node-2-1");
+ invocations.ShouldAllBe(i => i.Ct.CanBeCanceled);
+ }
+
+ [Fact]
+ public void Refresh_removes_departed_nodes_and_adds_new_ones()
+ {
+ var dial = new FakeDialLoop();
+ var current = new List
+ {
+ new("node-1", "http://h1:5300"),
+ new("node-2", "http://h2:5300"),
+ };
+
+ var actor = Spawn(dial, () => current.ToArray());
+ dial.WaitForCount(2, Timeout);
+
+ var node1 = dial.Invocations.Single(i => i.Target.NodeId == "node-1");
+
+ // node-1 leaves, node-3 joins.
+ current.Clear();
+ current.Add(new TelemetryDialTarget("node-2", "http://h2:5300"));
+ current.Add(new TelemetryDialTarget("node-3", "http://h3:5300"));
+ actor.Tell(new TelemetryDialSupervisor.RefreshNodes());
+
+ // node-1's dialer is cancelled and dropped.
+ AwaitCondition(() => node1.Ct.IsCancellationRequested, Timeout);
+ // 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");
+ }
+
+ [Fact]
+ public void Each_record_type_lands_on_its_sink()
+ {
+ var dial = new FakeDialLoop();
+ var sinks = new Sinks();
+ Spawn(dial, () => new[] { new TelemetryDialTarget("node-1", "http://h1:5300") }, sinks);
+
+ var inv = dial.WaitForCount(1, Timeout).Single();
+
+ inv.OnMapped(Alarm("a1"));
+ inv.OnMapped(Script("s1"));
+ inv.OnMapped(Health("d1"));
+ inv.OnMapped(Resilience("d1"));
+
+ AwaitAssert(
+ () =>
+ {
+ sinks.Alarms.ShouldHaveSingleItem().AlarmId.ShouldBe("a1");
+ sinks.Scripts.ShouldHaveSingleItem().ScriptId.ShouldBe("s1");
+ sinks.Health.ShouldHaveSingleItem().DriverInstanceId.ShouldBe("d1");
+ sinks.Resilience.ShouldHaveSingleItem().DriverInstanceId.ShouldBe("d1");
+ },
+ Timeout);
+ }
+
+ [Fact]
+ public void Stream_failure_reconnects_with_an_incremented_generation()
+ {
+ var dial = new FakeDialLoop();
+ Spawn(dial, () => new[] { new TelemetryDialTarget("node-1", "http://h1:5300") });
+
+ var first = dial.WaitForCount(1, Timeout).Single();
+ first.CorrelationId.ShouldBe("central-node-1-1");
+
+ // Transport failure → the client would end the stream after onError, so end the fake loop too.
+ first.OnError(new InvalidOperationException("boom"));
+ first.Complete();
+
+ // First retry is immediate; generation is bumped.
+ var invocations = dial.WaitForCount(2, Timeout);
+ invocations[1].CorrelationId.ShouldBe("central-node-1-2");
+ }
+
+ [Fact]
+ public void Late_event_from_a_superseded_generation_is_dropped()
+ {
+ var dial = new FakeDialLoop();
+ var sinks = new Sinks();
+ Spawn(dial, () => new[] { new TelemetryDialTarget("node-1", "http://h1:5300") }, sinks);
+
+ var first = dial.WaitForCount(1, Timeout).Single();
+ first.OnError(new InvalidOperationException("boom"));
+ first.Complete();
+ dial.WaitForCount(2, Timeout); // gen-2 dialer is live.
+
+ // A late event arriving on the OLD (gen-1) stream must be dropped, not routed.
+ first.OnMapped(Alarm("stale"));
+
+ // Give the message a chance to be (wrongly) routed, then assert it was not.
+ ExpectNoMsg(TimeSpan.FromMilliseconds(300));
+ sinks.Alarms.ShouldBeEmpty();
+ }
+
+ [Fact]
+ public void Late_failure_from_a_superseded_generation_does_not_trigger_a_second_reconnect()
+ {
+ var dial = new FakeDialLoop();
+ Spawn(dial, () => new[] { new TelemetryDialTarget("node-1", "http://h1:5300") });
+
+ var first = dial.WaitForCount(1, Timeout).Single();
+ first.OnError(new InvalidOperationException("boom"));
+ first.Complete();
+ dial.WaitForCount(2, Timeout); // gen-2 dialer.
+
+ // A duplicate/late failure from gen-1 must be ignored (no third dial).
+ first.OnError(new InvalidOperationException("late duplicate"));
+
+ ExpectNoMsg(TimeSpan.FromMilliseconds(400));
+ dial.Invocations.Count.ShouldBe(2);
+ }
+
+ [Fact]
+ public void Pill_flips_true_on_first_connection_and_false_only_when_all_down()
+ {
+ var dial = new FakeDialLoop();
+ var sinks = new Sinks();
+ Spawn(
+ dial,
+ () => new[]
+ {
+ new TelemetryDialTarget("node-1", "http://h1:5300"),
+ new TelemetryDialTarget("node-2", "http://h2:5300"),
+ },
+ sinks);
+
+ var inv = dial.WaitForCount(2, Timeout);
+ var node1 = inv.Single(i => i.Target.NodeId == "node-1");
+ var node2 = inv.Single(i => i.Target.NodeId == "node-2");
+
+ // First node connects → both broadcasters flip to connected.
+ node1.OnMapped(Alarm("a1"));
+ AwaitAssert(() => sinks.PillStates.ToArray().ShouldBe(new[] { true }), Timeout);
+
+ // Second node connects → still connected, no extra transition.
+ node2.OnMapped(Alarm("a2"));
+ ExpectNoMsg(TimeSpan.FromMilliseconds(200));
+ sinks.PillStates.ToArray().ShouldBe(new[] { true });
+
+ // node-1 drops but node-2 is still up → pill stays true.
+ node1.OnError(new InvalidOperationException("down"));
+ node1.Complete();
+ ExpectNoMsg(TimeSpan.FromMilliseconds(200));
+ sinks.PillStates.ToArray().ShouldBe(new[] { true });
+
+ // node-2 drops too → all down → pill flips false.
+ node2.OnError(new InvalidOperationException("down"));
+ node2.Complete();
+ AwaitAssert(() => sinks.PillStates.ToArray().ShouldBe(new[] { true, false }), Timeout);
+ }
+
+ [Fact]
+ public void PostStop_cancels_every_dialer()
+ {
+ var dial = new FakeDialLoop();
+ var actor = Spawn(
+ dial,
+ () => new[]
+ {
+ new TelemetryDialTarget("node-1", "http://h1:5300"),
+ new TelemetryDialTarget("node-2", "http://h2:5300"),
+ });
+
+ var inv = dial.WaitForCount(2, Timeout);
+
+ Sys.Stop(actor);
+
+ AwaitCondition(() => inv.All(i => i.Ct.IsCancellationRequested), Timeout);
+ }
+
+ [Fact]
+ public async Task Production_node_source_skips_null_grpc_port_and_maintenance_rows()
+ {
+ var factory = new InMemoryFactory(Guid.NewGuid().ToString());
+ await using (var db = factory.CreateDbContext())
+ {
+ db.ClusterNodes.AddRange(
+ Node("has-port", "h1", grpcPort: 5300, enabled: true, maintenance: false),
+ Node("null-port", "h2", grpcPort: null, enabled: true, maintenance: false),
+ Node("maintenance", "h3", grpcPort: 5300, enabled: true, maintenance: true),
+ Node("disabled", "h4", grpcPort: 5300, enabled: false, maintenance: false));
+ await db.SaveChangesAsync();
+ }
+
+ var source = TelemetryNodeSource.Create(factory, NullLogger.Instance);
+ var targets = await source();
+
+ targets.ShouldHaveSingleItem();
+ targets[0].NodeId.ShouldBe("has-port");
+ targets[0].Endpoint.ShouldBe("http://h1:5300");
+ }
+
+ private IActorRef Spawn(
+ FakeDialLoop dial,
+ Func nodeSource,
+ Sinks? sinks = null)
+ {
+ sinks ??= new Sinks();
+ var options = new TelemetryDialOptions { ContactRefreshSeconds = 3600 };
+ return Sys.ActorOf(TelemetryDialSupervisor.Props(
+ () => Task.FromResult>(nodeSource()),
+ dial.Loop,
+ sinks.AlarmBroadcaster,
+ sinks.ScriptBroadcaster,
+ sinks.HealthStore,
+ sinks.ResilienceStore,
+ options));
+ }
+
+ private static AlarmTransitionEvent Alarm(string id) =>
+ new(id, "Area/Line/Eq", "hi", "Activated", 500, "msg", "system", DateTime.UtcNow);
+
+ private static ScriptLogEntry Script(string id) =>
+ new(id, "Info", "msg", DateTime.UtcNow, null, null, null);
+
+ private static DriverHealthChanged Health(string id) =>
+ new("C1", id, "Healthy", DateTime.UtcNow, null, 0, DateTime.UtcNow);
+
+ private static DriverResilienceStatusChanged Resilience(string id) =>
+ new(id, "host", false, 0, 0, null, DateTime.UtcNow, DateTime.UtcNow);
+
+ private static ClusterNode Node(string id, string host, int? grpcPort, bool enabled, bool maintenance) =>
+ new()
+ {
+ NodeId = id,
+ ClusterId = "C1",
+ Host = host,
+ GrpcPort = grpcPort,
+ Enabled = enabled,
+ MaintenanceMode = maintenance,
+ ApplicationUri = $"urn:{id}",
+ CreatedBy = "test",
+ };
+
+ /// A captured dial-loop invocation with the callbacks the test drives by hand.
+ private sealed class DialInvocation
+ {
+ private readonly TaskCompletionSource _completion = new(TaskCreationOptions.RunContinuationsAsynchronously);
+
+ public required TelemetryDialTarget Target { get; init; }
+ public required string CorrelationId { get; init; }
+ public required Action OnMapped { get; init; }
+ public required Action OnError { get; init; }
+ public required CancellationToken Ct { get; init; }
+
+ public Task Completion => _completion.Task;
+
+ /// Ends the fake loop (as the client returns after a server-ended or post-onError stream).
+ public void Complete() => _completion.TrySetResult();
+ }
+
+ /// A fake
+ private sealed class FakeDialLoop
+ {
+ private readonly ConcurrentQueue _invocations = new();
+
+ public IReadOnlyList Invocations => _invocations.ToArray();
+
+ public TelemetryDialLoop Loop => async (target, correlationId, onMapped, onError, ct) =>
+ {
+ var invocation = new DialInvocation
+ {
+ Target = target,
+ CorrelationId = correlationId,
+ OnMapped = onMapped,
+ OnError = onError,
+ Ct = ct,
+ };
+ _invocations.Enqueue(invocation);
+
+ // Complete when the test ends the loop OR the supervisor cancels us.
+ 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}.");
+ }
+ }
+
+ /// The four fake sinks plus the pill-transition record.
+ private sealed class Sinks
+ {
+ public RecordingBroadcaster AlarmBroadcaster { get; }
+ public RecordingBroadcaster ScriptBroadcaster { get; }
+ public RecordingHealthStore HealthStore { get; } = new();
+ public RecordingResilienceStore ResilienceStore { get; } = new();
+
+ public ConcurrentQueue Alarms { get; } = new();
+ public ConcurrentQueue Scripts { get; } = new();
+ public ConcurrentQueue PillStates { get; } = new();
+
+ public ConcurrentQueue Health => HealthStore.Upserts;
+ public ConcurrentQueue Resilience => ResilienceStore.Upserts;
+
+ public Sinks()
+ {
+ // Both broadcasters share the ONE pill-transition record so a test can assert the alarm and
+ // script pills move together (matching the supervisor calling SetConnected on both).
+ AlarmBroadcaster = new RecordingBroadcaster(Alarms, PillStates);
+ ScriptBroadcaster = new RecordingBroadcaster(Scripts, null);
+ }
+ }
+
+ private sealed class RecordingBroadcaster(ConcurrentQueue published, ConcurrentQueue? pill)
+ : IInProcessBroadcaster
+ {
+ public event Action? Received;
+ public event Action? ConnectionStateChanged;
+
+ public bool IsConnected { get; private set; }
+
+ public void Publish(T item)
+ {
+ published.Enqueue(item);
+ Received?.Invoke(item);
+ }
+
+ public void SetConnected(bool connected)
+ {
+ IsConnected = connected;
+ pill?.Enqueue(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 RecordingResilienceStore : IDriverResilienceStatusStore
+ {
+ public ConcurrentQueue Upserts { get; } = new();
+
+ public event Action? SnapshotChanged;
+
+ public void Upsert(DriverResilienceStatusChanged snapshot)
+ {
+ Upserts.Enqueue(snapshot);
+ SnapshotChanged?.Invoke(snapshot);
+ }
+
+ public IReadOnlyList GetForInstance(string driverInstanceId) =>
+ Upserts.ToArray();
+
+ public IReadOnlyCollection GetAll() => Upserts.ToArray();
+ }
+
+ private sealed class InMemoryFactory(string name) : IDbContextFactory
+ {
+ public OtOpcUaConfigDbContext CreateDbContext() =>
+ new(new DbContextOptionsBuilder()
+ .UseInMemoryDatabase(name)
+ .Options);
+
+ public Task CreateDbContextAsync(CancellationToken cancellationToken = default) =>
+ Task.FromResult(CreateDbContext());
+ }
+}