From 1104785c89209466404dd6455351b7343184d49c Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 23 Jul 2026 16:28:57 -0400 Subject: [PATCH] feat(mesh-phase5): central telemetry dial supervisor (discover + reconnect + feed sinks) Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW --- .../Telemetry/TelemetryDialSupervisor.cs | 468 ++++++++++++++++++ .../Telemetry/TelemetryNodeSource.cs | 86 ++++ .../Telemetry/TelemetryDialSupervisorTests.cs | 433 ++++++++++++++++ 3 files changed, 987 insertions(+) create mode 100644 src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Telemetry/TelemetryDialSupervisor.cs create mode 100644 src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Telemetry/TelemetryNodeSource.cs create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/Telemetry/TelemetryDialSupervisorTests.cs 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. +/// +/// The node's stable logical id (the ClusterNode.NodeId). +/// The http://{Host}:{GrpcPort} h2c telemetry endpoint. +public sealed record TelemetryDialTarget(string NodeId, string Endpoint); + +/// +/// The dial-loop seam: opens one telemetry stream to and pumps mapped +/// domain records to until the stream ends or +/// fires, routing a transport failure to . The production +/// implementation wraps TelemetryStreamClient + TelemetryProtoMapCentral; tests +/// substitute a fake that captures the callbacks and drives events/failures by hand. +/// +/// The node to dial. +/// Per-generation stream correlation id. +/// Sink for each mapped domain record (already projected from the wire). +/// Invoked on a transport failure (the reconnect trigger). +/// Cancels the stream (a normal, supervisor-initiated shutdown). +/// 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 node source), +/// starts a dialer for each new node id, and stops+removes a dialer whose node id has left the +/// set. A dialer whose stream drops reconnects indefinitely (first retry immediate, then a +/// fixed backoff) — this is an observability plane, not a data plane, so it never gives up. +/// +/// +/// 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. + /// Resolves the current dialable node set (off the actor thread). + /// Opens + pumps one node's telemetry stream (the reconnect unit). + /// Sink for alarm-transition records. + /// Sink for script-log records. + /// Sink for driver-health snapshots. + /// Sink for driver-resilience snapshots. + /// The bound central dial options (refresh cadence etc.). + /// 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 class. + /// Resolves the current dialable node set (off the actor thread). + /// Opens + pumps one node's telemetry stream (the reconnect unit). + /// Sink for alarm-transition records. + /// Sink for script-log records. + /// Sink for driver-health snapshots. + /// Sink for driver-resilience snapshots. + /// The bound central dial options (refresh cadence etc.). + public TelemetryDialSupervisor( + Func>> 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")); + } + + /// + protected override void PreStart() + { + Timers.StartPeriodicTimer( + RefreshTimerKey, + new RefreshNodes(), + TimeSpan.Zero, + TimeSpan.FromSeconds(Math.Max(1, _options.ContactRefreshSeconds))); + } + + /// + protected override void PostStop() + { + foreach (var dialer in _dialers.Values) + { + dialer.Cancel(); + } + + _dialers.Clear(); + } + + private void HandleRefreshNodes() + { + Task> 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. + /// The current dialable nodes. + public sealed record NodesResolved(IReadOnlyList Targets); + + /// A mapped domain record received from a node's stream. + /// The source node id. + /// The dialer generation that produced it. + /// The mapped domain record. + public sealed record TelemetryReceived(string NodeId, int Generation, object Record); + + /// A node's stream stopped — transport failure ( set) or server-ended. + /// The node id whose stream stopped. + /// The dialer generation that stopped. + /// The transport failure, or when the server ended the stream. + public sealed record StreamStopped(string NodeId, int Generation, Exception? Error); + + /// Backoff timer tick: start a fresh dialer for the node. + /// The node id to reconnect. + public sealed record Reconnect(string NodeId); +} diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Telemetry/TelemetryNodeSource.cs b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Telemetry/TelemetryNodeSource.cs new file mode 100644 index 00000000..2a083f0b --- /dev/null +++ b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Telemetry/TelemetryNodeSource.cs @@ -0,0 +1,86 @@ +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.Logging; +using ZB.MOM.WW.OtOpcUa.Configuration; +using ZB.MOM.WW.OtOpcUa.ControlPlane.Telemetry; + +namespace ZB.MOM.WW.OtOpcUa.AdminUI.Telemetry; + +/// +/// Production seams for : the DB-backed node source and the +/// gRPC dial loop. Kept out of the actor so the actor stays DB-free and gRPC-free (and therefore +/// unit-testable with fakes). +/// +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. + /// + /// Factory for the config database holding ClusterNode rows. + /// Logger for skipped-row diagnostics. + /// 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 per stream, mapping + /// every wire envelope via before handing it to + /// the supervisor. The client isolates a mapper throw (logs + continues) and routes only + /// transport failures to onError, so the supervisor sees a poison event as "dropped", + /// never as "node down". + /// + /// Shared node bearer key sent on every dial. + /// Logger passed to each per-stream client (dropped-event diagnostics). + /// 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 — the central-side dial supervisor that +/// discovers driver nodes, keeps one reconnecting dialer each, and feeds the four AdminUI +/// in-process sinks. All seams are faked: no real DB, no real gRPC. +/// +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 that records each invocation and blocks until told to end. + 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()); + } +}