diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/TelemetryOptions.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/TelemetryOptions.cs index 779228b9..3e6f5143 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/TelemetryOptions.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/TelemetryOptions.cs @@ -72,7 +72,13 @@ public sealed class TelemetryDialOptions /// How often central refreshes its set of dialable node contacts, in seconds. public int ContactRefreshSeconds { get; set; } = 60; - /// Per-call deadline for a gRPC telemetry dial, in seconds. + /// + /// 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. + /// public int CallTimeoutSeconds { get; set; } = 30; } diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Browsing/DiscoveryDriverBrowser.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Browsing/DiscoveryDriverBrowser.cs index acf34e10..a4b62ec5 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Browsing/DiscoveryDriverBrowser.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Browsing/DiscoveryDriverBrowser.cs @@ -188,6 +188,13 @@ public sealed class DiscoveryDriverBrowser : IUniversalDriverBrowser } } + /// + /// Final teardown of a throwaway capture/probe driver: a bounded + /// followed by a bounded dispose. The dispose is NOT optional and runs even after a failed or timed-out + /// shutdown: a Galaxy driver's ShutdownAsync releases none of its gRPC runtime (channels, MX + /// session, stream loops) — only DisposeAsync does — so a browse that stopped at shutdown leaked a + /// full runtime per /raw browse and per probe (02/L1, 06/G-1). + /// private async Task BoundedShutdownAsync(IDriver driver, string driverType) { try @@ -199,7 +206,25 @@ public sealed class DiscoveryDriverBrowser : IUniversalDriverBrowser catch (Exception ex) { _logger.LogWarning(ex, - "Browse-capture shutdown for {DriverType} failed/timed out; instance abandoned", driverType); + "Browse-capture shutdown for {DriverType} failed/timed out; disposing the instance anyway", driverType); + } + + try + { + switch (driver) + { + case IAsyncDisposable asyncDisposable: + await asyncDisposable.DisposeAsync().AsTask().WaitAsync(_shutdownTimeout).ConfigureAwait(false); + break; + case IDisposable disposable: + disposable.Dispose(); + break; + } + } + catch (Exception ex) + { + _logger.LogWarning(ex, + "Browse-capture dispose for {DriverType} failed/timed out; instance abandoned", driverType); } } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Clients/FleetDiagnosticsClient.cs b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Clients/FleetDiagnosticsClient.cs index 81a0b6cc..71169d81 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Clients/FleetDiagnosticsClient.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Clients/FleetDiagnosticsClient.cs @@ -1,4 +1,6 @@ using Akka.Actor; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; using Microsoft.Extensions.Options; using ZB.MOM.WW.OtOpcUa.Cluster; using ZB.MOM.WW.OtOpcUa.Commons.Interfaces; @@ -13,7 +15,9 @@ namespace ZB.MOM.WW.OtOpcUa.AdminUI.Clients; /// Sends ; expects a reply. /// /// On timeout or any other failure (peer down, GetDiagnostics handler missing) returns an -/// empty snapshot so the UI degrades to "no data" instead of throwing. +/// empty snapshot so the UI degrades to "no data" instead of throwing. The failure is logged at +/// Warning, since the snapshot shape cannot carry it: an empty snapshot reads the same as a node with +/// no drivers. Consumed by the /fleet/diagnostics page. /// public sealed class FleetDiagnosticsClient : IFleetDiagnosticsClient { @@ -21,29 +25,38 @@ public sealed class FleetDiagnosticsClient : IFleetDiagnosticsClient private readonly ActorSystem _system; private readonly string _systemName; + private readonly ILogger _log; /// Initializes a new FleetDiagnosticsClient with the given actor system and cluster options. /// The Akka actor system. /// Cluster configuration options. - public FleetDiagnosticsClient(ActorSystem system, IOptions options) + /// Logger for the degraded-to-empty path; defaults to a no-op logger. + public FleetDiagnosticsClient( + ActorSystem system, IOptions options, ILogger? log = null) { _system = system; _systemName = options.Value.SystemName; + _log = log ?? NullLogger.Instance; } /// public async Task GetDiagnosticsAsync(NodeId nodeId, CancellationToken ct) { - var selection = _system.ActorSelection($"akka.tcp://{_systemName}@{nodeId.Value}/user/driver-host"); try { + // Inside the try: a malformed node value makes ActorSelection throw, and that must degrade to + // "unavailable" like a dead node does, not surface through the page's error boundary. + var selection = _system.ActorSelection($"akka.tcp://{_systemName}@{nodeId.Value}/user/driver-host"); using var linked = CancellationTokenSource.CreateLinkedTokenSource(ct); linked.CancelAfter(AskTimeout); return await selection.Ask( new GetDiagnostics(CorrelationId.NewId()), AskTimeout, linked.Token); } - catch (Exception) + catch (Exception ex) when (!ct.IsCancellationRequested) { + _log.LogWarning(ex, + "Fleet diagnostics for node {NodeId} unavailable (no answer within {Timeout}s); returning an empty snapshot", + nodeId.Value, AskTimeout.TotalSeconds); return new NodeDiagnosticsSnapshot( nodeId, CurrentRevision: null, diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Pages/FleetDiagnostics.razor b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Pages/FleetDiagnostics.razor new file mode 100644 index 00000000..1b958048 --- /dev/null +++ b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Pages/FleetDiagnostics.razor @@ -0,0 +1,238 @@ +@page "/fleet/diagnostics" +@* Per-node driver diagnostics, fetched on demand from the selected driver node's DriverHostActor over + Akka (IFleetDiagnosticsClient → GetDiagnostics → NodeDiagnosticsSnapshot). The node answers from + its live child set + each child's last-published health; the "running from cache" flag is the one + fact /hosts cannot show (a node booted from its artifact cache looks healthy but is frozen). + Reach it as /fleet/diagnostics?node= (the NodeId shown on /fleet and /hosts), or pick + a driver-role member from the selector. *@ +@attribute [Authorize(Policy = AdminUiPolicies.AuthenticatedRead)] +@rendermode RenderMode.InteractiveServer +@using ZB.MOM.WW.OtOpcUa.Cluster +@using ZB.MOM.WW.OtOpcUa.Commons.Interfaces +@using ZB.MOM.WW.OtOpcUa.Commons.Types +@inject IClusterRoleInfo Cluster +@inject IFleetDiagnosticsClient Diagnostics +@inject NavigationManager Nav +@implements IDisposable + + +
+

Node diagnostics

+ Back to fleet status +
+ +
+ + + + + + + Last fetched: @(_fetchedUtc?.ToString("HH:mm:ss 'UTC'") ?? "—") + +
+ +@if (string.IsNullOrEmpty(_selectedNode)) +{ +
+ Pick a driver-role node to fetch its driver diagnostics. The node answers from its own + DriverHostActor, so this works for any node reachable over Akka remoting from this admin node. +
+} +else if (_snapshot is null) +{ +

@(_loading ? "Fetching…" : "No snapshot yet.")

+} +else +{ +
+
+
Node
+
@_snapshot.NodeId.Value
+
+
+
Revision
+
+ @(_snapshot.CurrentRevision is { } rev ? DisplayText.Abbreviate(rev.Value, 12) : "—") +
+
+
+
Config source
+
@(_snapshot.RunningFromCache ? "cache (frozen)" : "live")
+
+
+
Drivers
+
@_snapshot.Drivers.Count
+
+
+
Faulted devices
+
@_snapshot.Drivers.Sum(d => d.FaultedDevices)
+
+
+ + @if (_snapshot.RunningFromCache) + { +
+ This node booted its configuration from its local artifact cache because central was unreachable. + It serves its last-known-good address space, but no deployment can reach it until central is back. +
+ } + +
+
+ Drivers + as of @_snapshot.AsOfUtc.ToString("u") +
+ @if (_snapshot.Drivers.Count == 0) + { +
+ Empty snapshot: the node has no driver children — or it did not answer within the client's + 3 s budget (an unreachable node degrades to an empty snapshot; see the AdminUI log for the + FleetDiagnosticsClient warning). +
+ } + else + { +
+ + + + + + + + + + + + + @foreach (var d in _snapshot.Drivers.OrderBy(d => d.Name, StringComparer.OrdinalIgnoreCase)) + { + + + + + + + + + } + +
Driver instanceRow idStateConnected devicesFaulted devicesLast health publish
@d.Name@(d.DriverInstanceId == Guid.Empty ? "—" : DisplayText.Abbreviate(d.DriverInstanceId.ToString(), 13))@d.State@d.ConnectedDevices@d.FaultedDevices@d.LastChangeUtc.ToString("u")
+
+ } +
+} + + +@code { + [SupplyParameterFromQuery(Name = "node")] + public string? Node { get; set; } + + private IReadOnlyList _members = Array.Empty(); + private string _selectedNode = ""; + private string _typedNode = ""; + private NodeDiagnosticsSnapshot? _snapshot; + private bool _loading; + private DateTime? _fetchedUtc; + + // Page-lifetime cancellation: a fetch in flight when the circuit goes away is abandoned. + private readonly CancellationTokenSource _cts = new(); + private volatile bool _disposed; + + protected override async Task OnParametersSetAsync() + { + _members = Cluster.MembersWithRole(RoleParser.Driver) + .Select(n => n.Value) + .OrderBy(s => s, StringComparer.OrdinalIgnoreCase) + .ToList(); + var requested = Node?.Trim() ?? ""; + if (!string.Equals(requested, _selectedNode, StringComparison.Ordinal)) + { + _selectedNode = requested; + _snapshot = null; + if (!string.IsNullOrEmpty(_selectedNode)) await LoadAsync(); + } + } + + private Task OnNodeSelected(ChangeEventArgs e) + { + var value = e.Value?.ToString() ?? ""; + Navigate(value); + return Task.CompletedTask; + } + + private Task UseTypedNodeAsync() + { + Navigate(_typedNode.Trim()); + return Task.CompletedTask; + } + + private Task OnTypedKeyDown(KeyboardEventArgs e) + { + if (e.Key == "Enter") Navigate(_typedNode.Trim()); + return Task.CompletedTask; + } + + // The node id lives in the URL so a diagnostics view is linkable from /fleet and survives a reload. + private void Navigate(string node) => + Nav.NavigateTo(string.IsNullOrEmpty(node) + ? "/fleet/diagnostics" + : "/fleet/diagnostics?node=" + Uri.EscapeDataString(node)); + + private async Task RefreshAsync() => await LoadAsync(); + + private async Task LoadAsync() + { + if (_loading || string.IsNullOrEmpty(_selectedNode)) return; + _loading = true; + try + { + // The client is fail-safe by contract: a node that does not answer yields an empty snapshot. + _snapshot = await Diagnostics.GetDiagnosticsAsync(NodeId.Parse(_selectedNode), _cts.Token); + _fetchedUtc = DateTime.UtcNow; + } + catch (Exception ex) when (ex is OperationCanceledException or ObjectDisposedException) + { + // Circuit gone mid-fetch — a clean exit, not a fault. + } + finally + { + _loading = false; + if (!_disposed) StateHasChanged(); + } + } + + private static string StateChipClass(string state) => state switch + { + "Healthy" => "chip-ok", + "Degraded" or "Connecting" or "Reconnecting" => "chip-caution", + "Faulted" => "chip-alert", + _ => "chip-idle", + }; + + public void Dispose() + { + _disposed = true; + _cts.Cancel(); + _cts.Dispose(); + } +} 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 d11b1f9d..4d542e7a 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Telemetry/TelemetryDialSupervisor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Telemetry/TelemetryDialSupervisor.cs @@ -129,6 +129,7 @@ public sealed class TelemetryDialSupervisor : ReceiveActor, IWithTimers Receive(HandleTelemetryReceived); Receive(HandleStreamStopped); Receive(HandleReconnect); + Receive(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. @@ -251,6 +252,9 @@ public sealed class TelemetryDialSupervisor : ReceiveActor, IWithTimers return; } + // The stream is over; a still-armed establishment deadline for it must not fire into the reconnect. + Timers.Cancel(DeadlineTimerKey(msg.NodeId)); + if (dialer.Connected) { dialer.Connected = false; @@ -294,6 +298,34 @@ public sealed class TelemetryDialSupervisor : ReceiveActor, IWithTimers } } + /// + /// TelemetryDial:CallTimeoutSeconds: a dial that has delivered no event by the deadline is cut and + /// treated as a transport failure (reconnect with a failure streak). Connected 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. + /// + 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 +338,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 +391,7 @@ public sealed class TelemetryDialSupervisor : ReceiveActor, IWithTimers { dialer.Cancel(); Timers.Cancel(ReconnectTimerKey(nodeId)); + Timers.Cancel(DeadlineTimerKey(nodeId)); RecomputePill(); } } @@ -433,6 +468,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}"; + /// Per-node dialer bookkeeping. Mutated only on the actor thread. private sealed class NodeDialer(TelemetryDialTarget target) { @@ -496,4 +533,10 @@ public sealed class TelemetryDialSupervisor : ReceiveActor, IWithTimers /// Backoff timer tick: start a fresh dialer for the node. /// The node id to reconnect. public sealed record Reconnect(string NodeId); + + /// Dial-establishment deadline tick (TelemetryDial:CallTimeoutSeconds); see + /// . + /// The node whose dial is being checked. + /// The dialer generation the deadline was armed for. + public sealed record DialDeadline(string NodeId, int Generation); } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/ChildHealthLedger.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/ChildHealthLedger.cs new file mode 100644 index 00000000..ec722a2c --- /dev/null +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/ChildHealthLedger.cs @@ -0,0 +1,73 @@ +using System.Collections.Concurrent; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Drivers; + +/// The last health snapshot a driver child published, as kept by . +/// The driver's reported state. +/// When the driver last read successfully, if ever. +/// The driver's last error, if any. +/// Per-host connectivity for a multi-host driver, or null when the driver has none. +/// When the snapshot was published — the child's dedup means this is the last CHANGE. +public sealed record ChildHealthRecord( + DriverState State, + DateTime? LastSuccessfulReadUtc, + string? LastError, + IReadOnlyList? HostStatuses, + DateTime PublishedUtc); + +/// +/// An decorator that remembers the last snapshot per driver instance +/// and forwards every publish to the real publisher. wraps the publisher it +/// hands its children in one of these so GetDiagnostics can answer from the children's own health +/// (state, host counts, last change) instead of a placeholder — the children publish on their own actor +/// threads, hence the concurrent map. +/// +/// +/// Deliberately NOT a message to the host: the children already publish through this seam on every +/// change, so a parallel Tell would double every health transition's traffic for one consumer. +/// Entries are removed by the host when it stops a child; a late publish from a stopping child may re-add +/// one, which the host filters against its live child set when it answers. +/// +public sealed class ChildHealthLedger : IDriverHealthPublisher +{ + private readonly IDriverHealthPublisher _inner; + private readonly ConcurrentDictionary _last = new(StringComparer.Ordinal); + + /// Wraps . + /// The publisher every snapshot is forwarded to. + public ChildHealthLedger(IDriverHealthPublisher inner) + { + ArgumentNullException.ThrowIfNull(inner); + _inner = inner; + } + + /// + public void Publish( + string clusterId, + string driverInstanceId, + DriverHealth health, + int errorCount5Min, + DateTime? rediscoveryNeededUtc = null, + string? rediscoveryReason = null, + IReadOnlyList? hostStatuses = null, + SkippedTagReport? skippedTags = null, + IReadOnlyDictionary? diagnostics = null) + { + _last[driverInstanceId] = new ChildHealthRecord( + health.State, health.LastSuccessfulRead, health.LastError, hostStatuses, DateTime.UtcNow); + _inner.Publish( + clusterId, driverInstanceId, health, errorCount5Min, rediscoveryNeededUtc, rediscoveryReason, + hostStatuses, skippedTags, diagnostics); + } + + /// The last snapshot published for , or null if none yet. + /// The driver instance. + /// The record, or null. + public ChildHealthRecord? TryGet(string driverInstanceId) => + _last.TryGetValue(driverInstanceId, out var record) ? record : null; + + /// Forgets 's snapshot (the child was stopped). + /// The driver instance. + public void Forget(string driverInstanceId) => _last.TryRemove(driverInstanceId, out _); +} diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs index 2d1bec44..76c82ebf 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs @@ -195,7 +195,9 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS private readonly IReadOnlySet _localRoles; private readonly IActorRef? _dependencyMux; private readonly IActorRef? _opcUaPublishActor; - private readonly IDriverHealthPublisher _healthPublisher; + /// The publisher handed to every child — a over the configured one, so + /// answers from the children's own last-published health. + private readonly ChildHealthLedger _healthPublisher; private readonly IVirtualTagEvaluator _virtualTagEvaluator; private readonly IHistoryWriter _historyWriter; private readonly IActorRef? _virtualTagHostOverride; @@ -594,6 +596,9 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS private sealed record ChildEntry(IActorRef Actor, DriverInstanceSpec Spec, bool Stubbed) { + /// When this child was spawned — the diagnostics' last-change time until it publishes health. + public DateTime SpawnedUtc { get; } = DateTime.UtcNow; + // Convenience accessors for sites that don't need the full spec. /// Gets the driver type name from the child's spec. public string DriverType => Spec.DriverType; @@ -886,7 +891,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS _localRoles = localRoles ?? new HashSet(StringComparer.Ordinal); _dependencyMux = dependencyMux; _opcUaPublishActor = opcUaPublishActor; - _healthPublisher = healthPublisher ?? NullDriverHealthPublisher.Instance; + _healthPublisher = new ChildHealthLedger(healthPublisher ?? NullDriverHealthPublisher.Instance); _virtualTagEvaluator = virtualTagEvaluator ?? NullVirtualTagEvaluator.Instance; _historyWriter = historyWriter ?? NullHistoryWriter.Instance; _virtualTagHostOverride = virtualTagHostOverride; @@ -2489,6 +2494,24 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS // A child connectivity transition while the host is Stale has no live address space to annotate — drop // it (the post-recovery rebuild re-materialises conditions, and the child re-announces on its next entry). Receive(_ => { }); + // Likewise a child's value / alarm burst: Stale is entered from Bootstrap's catch before any child is + // spawned, so none should arrive — but the orphan-replay BeginDirectApply in that same try block has + // already started a pipe-back before the catch can run, so "cannot happen" rests on ordering. Explicit + // drops, so a burst that does land is a logged drop rather than an Unhandled dead-letter. + Receive(m => + _log.Debug("DriverHost {Node}: dropping a {Count}-item value burst from {Id} while Stale (no address space)", + _localNode, m.Items.Count, m.DriverInstanceId)); + Receive(m => + _log.Debug("DriverHost {Node}: dropping a native alarm transition ({Cond}) from {Id} while Stale", + _localNode, m.Args.ConditionId, m.DriverInstanceId)); + // A late artifact pipe-back from the apply that was in flight when the host fell Stale: that apply is + // lost (the coordinator's deadline and the boot-time orphan replay recover it). Warning, not Debug — + // an apply outcome is being discarded. + Receive(m => + _log.Warning("DriverHost {Node}: dropping the artifact for {Id} (rev {Rev}) that arrived while Stale; the apply is abandoned", + _localNode, m.DeploymentId, m.Revision)); + Receive(m => + _log.Debug("DriverHost {Node}: late apply-wedge tick for {Id} while Stale; dropping", _localNode, m.DeploymentId)); // Likewise the 03/S15 recovery audit: a timer armed before the host went Stale has nothing to report // on (no latch was set while Stale, and the rebuild re-establishes every node's status anyway). Receive(_ => { }); @@ -2496,16 +2519,16 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS Timers.StartPeriodicTimer("retry-db", RetryConfigDbConnection.Instance, ReconnectInterval); } + /// + /// Answers from the live child set and each child's last-published health + /// (). Used to be a placeholder — Guid.Empty, "Spawned", zero + /// devices, "now" — that looked live in every behaviour and was read by nobody (02/U1); the AdminUI + /// fleet-diagnostics page is its consumer. + /// private void HandleGetDiagnostics(GetDiagnostics msg) { var drivers = _children - .Select(kv => new DriverInstanceDiagnostics( - DriverInstanceId: Guid.Empty, - Name: kv.Key, - State: kv.Value.Stubbed ? "Stubbed" : "Spawned", - ConnectedDevices: 0, - FaultedDevices: 0, - LastChangeUtc: DateTime.UtcNow)) + .Select(kv => ToDiagnostics(kv.Key, kv.Value, _healthPublisher.TryGet(kv.Key))) .ToArray(); var snapshot = new NodeDiagnosticsSnapshot( NodeId: _localNode, @@ -2516,6 +2539,39 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS Sender.Tell(snapshot); } + /// + /// One child's diagnostics row. Name carries the string driver-instance id — the identity every + /// other surface (/hosts, telemetry, the artifact) keys on — and DriverInstanceId the DB row Guid; + /// the wire record has no slot for the authored display name. Device counts come from the driver's host + /// statuses when it is an IHostConnectivityProbe (Running = connected; Stopped/Faulted = faulted; + /// Unknown = neither); a single-host driver counts as one device, connected while it reports Healthy or + /// Degraded and faulted while it reports Faulted. A child that has not published yet reports + /// "Connecting" (or "Stubbed") with its spawn time as the last change. + /// + private static DriverInstanceDiagnostics ToDiagnostics(string driverInstanceId, ChildEntry entry, ChildHealthRecord? health) + { + var name = driverInstanceId; + if (entry.Stubbed) + return new DriverInstanceDiagnostics(entry.Spec.DriverInstanceRowId, name, "Stubbed", 0, 0, entry.SpawnedUtc); + if (health is null) + return new DriverInstanceDiagnostics(entry.Spec.DriverInstanceRowId, name, "Connecting", 0, 0, entry.SpawnedUtc); + + int connected, faulted; + if (health.HostStatuses is { Count: > 0 } hosts) + { + connected = hosts.Count(h => h.State == HostState.Running); + faulted = hosts.Count(h => h.State is HostState.Stopped or HostState.Faulted); + } + else + { + connected = health.State is DriverState.Healthy or DriverState.Degraded ? 1 : 0; + faulted = health.State == DriverState.Faulted ? 1 : 0; + } + + return new DriverInstanceDiagnostics( + entry.Spec.DriverInstanceRowId, name, health.State.ToString(), connected, faulted, health.PublishedUtc); + } + private void HandleDispatchFromSteady(DispatchDeployment msg) { // The short-circuit claims a CLEAN Applied, so it is only honest while the current revision's @@ -3771,6 +3827,12 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS // id lifts it (the tracker via _invokerFactory.Create, the hub via Readmit). _invokerFactory.Release(driverInstanceId); _telemetryHub?.Evict(driverInstanceId); + // The reprovision rate-limit stamp belongs to the incarnation history that just ended. Kept, it grew the + // map by one entry per id ever reprovisioned-then-removed AND made a later re-add of the same id inherit a + // backoff it never earned (its first crash was deferred as a "crash loop"). + _lastReprovisionUtc.Remove(driverInstanceId); + Timers.Cancel(ReprovisionTimerKey(driverInstanceId)); + _healthPublisher.Forget(driverInstanceId); _log.Info("DriverHost {Node}: stopped driver child {Id}", _localNode, driverInstanceId); } @@ -3821,6 +3883,8 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS StopDependencyAdapter(msg.DriverInstanceId); Context.Stop(entry.Actor); _children.Remove(msg.DriverInstanceId); + // The old incarnation's health must not read as the new one's until it publishes its own. + _healthPublisher.Forget(msg.DriverInstanceId); // Respawn from the same spec the last reconcile used — preserves RowId, Name, ClusterId. SpawnChild(entry.Spec); diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs index 19616fc6..60e989ce 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs @@ -1,3 +1,4 @@ +using System.Diagnostics; using Akka.Actor; using Akka.Event; using ZB.MOM.WW.OtOpcUa.Cluster; @@ -184,6 +185,52 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers /// public static readonly TimeSpan DefaultCommsLossTimeout = TimeSpan.FromSeconds(90); + /// + /// Bound on the driver's (and its dispose) when the child stops or + /// restarts, and on the teardown that precedes every reconnect attempt. PostStop runs on a + /// dispatcher thread and blocks on it, so a driver whose shutdown waits on a dead socket / COM proxy used + /// to pin that thread for as long as the wait took — a few such stops starved the default dispatcher and + /// wedged CoordinatedShutdown and the host's own restart cascade. A shutdown that overruns is + /// abandoned (logged), never awaited further. + /// + public static readonly TimeSpan DefaultShutdownTimeout = TimeSpan.FromSeconds(5); + + /// + /// Default cap on the writes this actor runs against its driver at once. Writes run OFF the mailbox + /// (a ReceiveAsync used to suspend it for the whole device round-trip, so writes serialised and + /// every data change queued behind them — the second of two operator writes to a slow device missed the + /// host's 8 s Ask and was reverted although the device took it). The cap keeps a burst from flooding a + /// device with concurrent commands; writes beyond it queue in order and start as slots free. + /// + public const int DefaultMaxWritesInFlight = 4; + + /// + /// How long a queued write may wait for a slot before it is failed BadTimeout instead of being + /// written. The host's write Ask is 8 s: a write older than that has already been reported as a timeout + /// (and its node reverted) upstream, so writing it late would push a stale value to the device. + /// + public static readonly TimeSpan DefaultQueuedWriteBudget = TimeSpan.FromSeconds(8); + + /// + /// Pessimistic bound on one attempt. The invoker's tier timeout is + /// token-only (optimistic), so a driver whose initialise ignores its token would otherwise hold the + /// single-flight gate forever and wedge the reconnect loop. 2× the longest tier (Tier C: 15 s). + /// + public static readonly TimeSpan DefaultInitializeTimeout = TimeSpan.FromSeconds(30); + + /// + /// Pessimistic bound on one write round-trip, on top of the 5 s token-based write cap. A driver whose + /// ignores its token would otherwise hold its slot AND its tag forever, + /// piling every later write to that tag up in the queue. 2× the write cap. + /// + public static readonly TimeSpan DefaultWriteTimeout = TimeSpan.FromSeconds(10); + + /// Self-sent by a write task once it has replied, releasing its slot and its tag. + /// The tag the write held. + /// The actor incarnation that launched it — a late completion from a previous + /// incarnation (after a supervision restart) must not free the new incarnation's slot or tag. + private sealed record WriteCompleted(string TagId, int Incarnation); + private readonly IDriver _driver; /// Phase 6.1 resilience seam: every guarded driver-capability call this actor makes is @@ -204,6 +251,47 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers /// Comms-loss watchdog threshold; or negative disables it. private readonly TimeSpan _commsLossTimeout; + /// Bound on the driver shutdown / dispose; see . + private readonly TimeSpan _shutdownTimeout; + + /// Cap on concurrent writes; see . + private readonly int _maxWritesInFlight; + + /// Writes currently inside the driver. Actor-thread only (incremented on launch, decremented on + /// the self-sent ). + private int _writesInFlight; + + /// Writes waiting for a slot (or for their tag), in arrival order, stamped with arrival time. + /// Failed fast on every exit from Connected, so a queued write never reaches a driver that is + /// reconnecting; failed BadTimeout at dequeue once older than . + private readonly LinkedList<(WriteAttribute Write, IActorRef ReplyTo, DateTime EnqueuedUtc)> _pendingWrites = new(); + + /// Tags with a write currently inside the driver. A second write to the same tag waits in the + /// queue behind it — two concurrent writes v1, v2 to one tag could otherwise land as v1. + private readonly HashSet _tagsInFlight = new(StringComparer.Ordinal); + + /// See . + private readonly TimeSpan _queuedWriteBudget; + + /// See . + private readonly TimeSpan _initializeTimeout; + + /// See . + private readonly TimeSpan _writeTimeout; + + /// This actor object's identity for its own off-mailbox completions (drawn from the process-wide + /// generation source, so it is unique across supervision restarts of the same child). + private readonly int _incarnation; + + /// + /// Set by so the it triggers skips the dispose. A + /// supervision restart recreates the actor object around the SAME (the Props closure + /// captured it) and the host then re-initialises that driver — disposing it here would hand the new + /// incarnation a dead object. Only a real stop (deploy removal, operator Restart, respawn on config change, + /// host shutdown) disposes. + /// + private bool _restarting; + /// When the actor last entered Connected. The comms-loss watchdog's baseline for a /// driver that has never reported a successful read, so "connected but never read anything" is /// measurable rather than indefinitely undefined. @@ -218,6 +306,26 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers /// on the actor thread, so no lock is needed. private int _initGeneration; + /// Process-wide source of attempt generations. A supervision restart recreates the actor object + /// (fresh fields) while a prior incarnation's attempt may still be running on the same driver; a per-actor + /// counter restarting at 0 would let that late result collide with the new incarnation's first attempt. + private static int s_initGenerationSource; + + /// + /// True while an attempt is running (set when it is launched, cleared by its + /// piped /). The Reconnecting retry + /// timer fires on its cadence regardless of whether the previous attempt finished, and the generation + /// token only discards an older attempt's RESULT — so with an Initialize bound longer than the cadence + /// (Tier C: 15 s vs. 10 s) attempt N+1's teardown ran while attempt N was still inside the driver: the + /// dispose-before-overwrite race the drivers' rule file forbids, once per overlap. A tick that lands while + /// an attempt is in flight is skipped; the next tick after the result retries. Actor-thread only. + /// + private bool _initInFlight; + + /// Cancellation for the in-flight attempt, so can cancel it before the + /// shutdown rather than let it race the teardown on a restart. + private CancellationTokenSource? _initCts; + /// /// Timestamps of recent Faulted-state transitions; used to compute the 5-minute error count. /// No lock needed — every read/write site runs inside an Akka message handler, which is @@ -307,6 +415,17 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers /// Optional comms-loss watchdog threshold; defaults to /// . disables the watchdog — used by /// tests that need a Connected driver to stay Connected while reporting non-Healthy. + /// Optional bound on the driver's shutdown / dispose on stop and on the + /// pre-reconnect teardown; defaults to . Exists so a test can prove a + /// hung shutdown is abandoned without waiting 5 s. + /// Optional cap on concurrent writes against the driver; defaults to + /// . Writes beyond it queue in arrival order. + /// Optional bound on how long a queued write may wait before it is failed + /// BadTimeout; defaults to . + /// Optional pessimistic bound on one initialise attempt; defaults to + /// . + /// Optional pessimistic bound on one write round-trip; defaults to + /// . /// A instance configured to create the wrapped . public static Props Props( IDriver driver, @@ -316,7 +435,12 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers string? clusterId = null, IDriverCapabilityInvoker? invoker = null, TimeSpan? healthPollInterval = null, - TimeSpan? commsLossTimeout = null) => + TimeSpan? commsLossTimeout = null, + TimeSpan? shutdownTimeout = null, + int? maxWritesInFlight = null, + TimeSpan? queuedWriteBudget = null, + TimeSpan? initializeTimeout = null, + TimeSpan? writeTimeout = null) => Akka.Actor.Props.Create(() => new DriverInstanceActor( driver, reconnectInterval ?? DefaultReconnectInterval, @@ -325,7 +449,12 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers clusterId ?? string.Empty, invoker, healthPollInterval, - commsLossTimeout)); + commsLossTimeout, + shutdownTimeout, + maxWritesInFlight, + queuedWriteBudget, + initializeTimeout, + writeTimeout)); /// /// Returns true when the driver should boot in DEV-STUB mode based on host platform and @@ -362,6 +491,16 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers /// subscription reconcile; defaults to when null. /// Comms-loss watchdog threshold; defaults to /// when null, and is disabled when zero or negative. + /// Bound on the driver's shutdown / dispose; defaults to + /// when null. + /// Cap on concurrent writes; defaults to + /// when null or non-positive. + /// Bound on a queued write's wait; defaults to + /// when null or non-positive. + /// Pessimistic bound on one initialise attempt; defaults to + /// when null or non-positive. + /// Pessimistic bound on one write round-trip; defaults to + /// when null or non-positive. public DriverInstanceActor( IDriver driver, TimeSpan reconnectInterval, @@ -370,8 +509,18 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers string? clusterId = null, IDriverCapabilityInvoker? invoker = null, TimeSpan? healthPollInterval = null, - TimeSpan? commsLossTimeout = null) + TimeSpan? commsLossTimeout = null, + TimeSpan? shutdownTimeout = null, + int? maxWritesInFlight = null, + TimeSpan? queuedWriteBudget = null, + TimeSpan? initializeTimeout = null, + TimeSpan? writeTimeout = null) { + _writeTimeout = writeTimeout is { } wt && wt > TimeSpan.Zero ? wt : DefaultWriteTimeout; + _incarnation = Interlocked.Increment(ref s_initGenerationSource); + _maxWritesInFlight = maxWritesInFlight is > 0 ? maxWritesInFlight.Value : DefaultMaxWritesInFlight; + _queuedWriteBudget = queuedWriteBudget is { } qb && qb > TimeSpan.Zero ? qb : DefaultQueuedWriteBudget; + _initializeTimeout = initializeTimeout is { } it && it > TimeSpan.Zero ? it : DefaultInitializeTimeout; _driver = driver; _invoker = invoker ?? NullDriverCapabilityInvoker.Instance; _driverInstanceId = driver.DriverInstanceId; @@ -380,6 +529,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers _reconnectInterval = reconnectInterval; _healthPollInterval = healthPollInterval ?? HealthPollInterval; _commsLossTimeout = commsLossTimeout ?? DefaultCommsLossTimeout; + _shutdownTimeout = shutdownTimeout is { } bound && bound > TimeSpan.Zero ? bound : DefaultShutdownTimeout; OtOpcUaTelemetry.DriverInstanceLifecycle.Add(1, new KeyValuePair("event", startStubbed ? "spawn_stub" : "spawn"), new KeyValuePair("driver_type", driver.DriverType)); @@ -421,6 +571,18 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers Context.Parent.Tell(new ReprovisionRequested(_driverInstanceId, reason.GetType().Name)); } + /// + /// Flags the restart before the base implementation runs , so that PostStop + /// shuts the driver down (the host re-initialises it on ) but does + /// NOT dispose it — see . + /// + /// + protected override void PreRestart(Exception reason, object message) + { + _restarting = true; + base.PreRestart(reason, message); + } + private void Stubbed() { // Stubbed drivers accept the standard message contracts but return deterministic @@ -448,6 +610,8 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers // "write timeout". Synchronous Receive: Sender.Tell on the actor thread is safe. Receive(_ => Sender.Tell(new WriteAttributeResult(false, "driver not connected"))); + // A write launched while Connected can finish after the transition out of it; its slot still frees. + Receive(OnWriteCompleted); // An ack arriving while still connecting can't reach the upstream alarm system; drop it (the ack is // fire-and-forget — no reply to surface — and the OPC UA condition state already committed locally). Receive(_ => @@ -455,6 +619,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers Receive(msg => { if (msg.Generation != _initGeneration) return; + CompleteInitAttempt(); _log.Info("DriverInstance {Id}: connected", _driverInstanceId); Become(Connected); PublishHealthSnapshot(); @@ -465,6 +630,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers Receive(msg => { if (msg.Generation != _initGeneration) return; + CompleteInitAttempt(); _log.Warning("DriverInstance {Id}: initialize failed: {Reason}", _driverInstanceId, msg.Reason); RecordFault(); Become(Reconnecting); @@ -514,10 +680,14 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers { _log.Info("DriverInstance {Id}: ForceReconnect requested by admin; re-entering Reconnecting", _driverInstanceId); DetachSubscription(); + FailQueuedWrites("driver not connected"); Become(Reconnecting); PublishHealthSnapshot(); }); - ReceiveAsync(HandleWriteAsync); + // Synchronous Receive: the device call runs off the mailbox (see StartOrQueueWrite), so data changes + // and further writes are never head-of-line blocked behind a slow device. + Receive(msg => StartOrQueueWrite(msg, Sender)); + Receive(OnWriteCompleted); ReceiveAsync(HandleAcknowledgeAsync); ReceiveAsync(HandleSubscribeAsync); ReceiveAsync(_ => UnsubscribeAsync()); @@ -745,6 +915,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers // timeout on an inbound write to a transiently-down driver). Synchronous Receive. Receive(_ => Sender.Tell(new WriteAttributeResult(false, "driver not connected"))); + Receive(OnWriteCompleted); // An ack arriving while reconnecting can't reach the upstream alarm system; drop it (fire-and-forget, // no reply — the OPC UA condition state already committed locally on the Part 9 ack). Receive(_ => @@ -752,6 +923,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers Receive(msg => { if (msg.Generation != _initGeneration) return; + CompleteInitAttempt(); Timers.Cancel("retry-connect"); _log.Info("DriverInstance {Id}: reconnected", _driverInstanceId); Become(Connected); @@ -760,9 +932,13 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers AttachAlarmSource(); SubscribeDesiredAlarms(); }); - // A failure here is a no-op regardless of generation — the retry timer keeps trying the - // current config; only a (generation-matched) InitializeSucceeded transitions state. - Receive(_ => { /* keep retrying via timer */ }); + // A failure here changes no state — the retry timer keeps trying the current config; only a + // (generation-matched) InitializeSucceeded transitions. It DOES release the single-flight gate, or + // no later tick could ever retry. + Receive(msg => + { + if (msg.Generation == _initGeneration) CompleteInitAttempt(); + }); Receive(StoreDesiredSubscriptions); Receive(_ => { /* already reconnecting — no-op */ }); Receive(IgnoreConnectionLoss); @@ -807,12 +983,28 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers private void InitializeAsync(string driverConfigJson, bool teardownFirst = false) { _currentConfigJson = driverConfigJson; - var generation = ++_initGeneration; + if (_initInFlight) + { + // SINGLE-FLIGHT (02/H2). The previous attempt is still inside the driver; starting another would run + // its teardown against the transport that attempt is building. Skip — the next tick after the + // result retries, and a config handed over meanwhile is already stored for it. + _log.Debug("DriverInstance {Id}: initialise attempt {Gen} still in flight; skipping this one", + _driverInstanceId, _initGeneration); + return; + } + + _initInFlight = true; + var generation = Interlocked.Increment(ref s_initGenerationSource); + _initGeneration = generation; + var initCts = new CancellationTokenSource(); + _initCts = initCts; var self = Self; var log = _log; var driverInstanceId = _driverInstanceId; var driver = _driver; var invoker = _invoker; + var shutdownTimeout = _shutdownTimeout; + var initializeTimeout = _initializeTimeout; _ = Task.Run(async () => { try @@ -823,7 +1015,16 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers { // Deliberately UNWRAPPED: teardown is best-effort cleanup of an already-dead // transport — a shutdown failure must not count toward the Initialize breaker. - await driver.ShutdownAsync(CancellationToken.None); + // BOUNDED: a teardown wedged on the dead transport must not hold the retry loop + // (and this thread) for longer than the bound; it is abandoned and the init proceeds. + using var teardownCts = new CancellationTokenSource(shutdownTimeout); + await driver.ShutdownAsync(teardownCts.Token).WaitAsync(shutdownTimeout); + } + catch (TimeoutException) + { + log.Warning( + "DriverInstance {Id}: ShutdownAsync did not complete within {Bound} before a reconnect " + + "attempt; abandoning it and initialising anyway", driverInstanceId, shutdownTimeout); } catch (Exception ex) { @@ -841,33 +1042,103 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers // attempt with BrokenCircuitException, which lands in the catch below exactly like a // real connect failure: InitializeFailed → keep retrying on the timer. The 15 s break // window is shorter than two 10 s timer ticks, so recovery costs at most one extra tick. - await invoker.ExecuteAsync( + // The outer token is this attempt's own: PostStop cancels it so a stop/restart does not run its + // shutdown underneath an attempt still inside the driver. + // PESSIMISTIC bound on top of the invoker's token-only timeout: a driver that ignores its token + // would otherwise hold the single-flight gate forever. An overrun is abandoned (the task keeps + // running, its eventual outcome is observed and discarded), reported as a failure so the retry + // loop continues, and its late Tell is dropped by the generation guard anyway. + var attempt = invoker.ExecuteAsync( DriverCapability.Initialize, driverInstanceId, async ct => await driver.InitializeAsync(driverConfigJson, ct), - CancellationToken.None); + initCts.Token).AsTask(); + try + { + await attempt.WaitAsync(initializeTimeout); + } + catch (TimeoutException) + { + log.Warning( + "DriverInstance {Id}: initialise attempt {Gen} did not complete within {Bound} (the driver ignored " + + "its cancellation token); attempt abandoned, the retry loop continues", + driverInstanceId, generation, initializeTimeout); + try { initCts.Cancel(); } catch (ObjectDisposedException) { } + _ = attempt.ContinueWith(t => _ = t.Exception, TaskContinuationOptions.OnlyOnFaulted); + self.Tell(new InitializeFailed($"initialise did not complete within {initializeTimeout}", generation)); + return; + } self.Tell(new InitializeSucceeded(generation)); } catch (Exception ex) { self.Tell(new InitializeFailed(ex.Message, generation)); } + finally + { + // The CTS outlives an abandoned attempt (it may still be inside the driver); dispose only once + // nothing can cancel it any more — the actor nulls _initCts when it honours the result. + if (!initCts.IsCancellationRequested) initCts.Dispose(); + } }); } - private async Task HandleWriteAsync(WriteAttribute msg) + /// Releases the single-flight gate once the in-flight attempt's result has been honoured. + private void CompleteInitAttempt() + { + _initInFlight = false; + _initCts = null; + } + + /// + /// Launches a write against the driver off the mailbox, or queues it when + /// writes are already inside the driver. Runs on the actor thread; the + /// only actor state it touches is the in-flight count and the queue. The device round-trip itself + /// () replies to the captured sender from the thread pool and then + /// self-sends to free its slot. + /// + private void StartOrQueueWrite(WriteAttribute msg, IActorRef replyTo) { - var replyTo = Sender; if (_driver is not IWritable writable) { replyTo.Tell(new WriteAttributeResult(false, "Driver does not implement IWritable")); return; } + if (_writesInFlight >= _maxWritesInFlight || _tagsInFlight.Contains(msg.TagId)) + { + _pendingWrites.AddLast((msg, replyTo, DateTime.UtcNow)); + return; + } + + LaunchWrite(writable, msg, replyTo); + } + + private void LaunchWrite(IWritable writable, WriteAttribute msg, IActorRef replyTo) + { + _writesInFlight++; + _tagsInFlight.Add(msg.TagId); + // Everything the task needs is captured HERE, on the actor thread — ResolveHost reads the driver, + // and Self must not be touched off-context. + var self = Self; + var hostKey = ResolveHost(msg.TagId); + var invoker = _invoker; + _ = RunWriteAsync(writable, invoker, hostKey, msg, replyTo, self, _writeTimeout, _incarnation, _log); + } + + private static async Task RunWriteAsync( + IWritable writable, IDriverCapabilityInvoker invoker, string hostKey, + WriteAttribute msg, IActorRef replyTo, IActorRef self, TimeSpan writeTimeout, int incarnation, + ILoggingAdapter log) + { + // Off the actor context from here on: the first await yields, and nothing below reads actor state. + await Task.Yield(); var request = new[] { new WriteRequest(msg.TagId, msg.Value) }; - // Bound the write so a hung backend can't pin this actor forever — retry stays - // off by default, but a stalled call still needs an answer. - using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5)); + // Bound the write so a hung backend can't hold a slot forever — retry stays + // off by default, but a stalled call still needs an answer. Disposed below, except when the write is + // abandoned (the orphan may still be holding the token). + var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5)); + var abandoned = false; try { // Route through the resilience invoker (Write pipeline: timeout, per-host breaker). Writes @@ -877,11 +1148,31 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers // therefore authoritative regardless of any configured Write retry. Per-tag opt-in to retries via // WriteIdempotentAttribute is the unshipped forward path (see WriteIdempotentAttribute). The outer // cts is retained as a hard backstop above the (typically tighter) tier timeout. - var results = await _invoker.ExecuteWriteAsync( - ResolveHost(msg.TagId), + // PESSIMISTIC bound on top of the token-based cap: a driver that ignores its token would otherwise + // hold the slot and the tag forever, piling every later write to that tag up in the queue. + var attempt = invoker.ExecuteWriteAsync( + hostKey, isIdempotent: false, async ct => await writable.WriteAsync(request, ct), - cts.Token); + cts.Token).AsTask(); + IReadOnlyList results; + try + { + results = await attempt.WaitAsync(writeTimeout); + } + catch (TimeoutException) + { + abandoned = true; + log.Warning( + "DriverInstance write of {Tag} did not complete within {Bound} (the driver ignored its cancellation " + + "token); abandoned — its slot and tag are released, its eventual outcome is discarded", + msg.TagId, writeTimeout); + try { cts.Cancel(); } catch (ObjectDisposedException) { } + _ = attempt.ContinueWith(t => _ = t.Exception, TaskContinuationOptions.OnlyOnFaulted); + replyTo.Tell(new WriteAttributeResult(false, + $"StatusCode=0x800A0000 (BadTimeout: write did not complete within {writeTimeout.TotalSeconds:F0}s)")); + return; + } if (results is { Count: 1 } && IsGoodStatus(results[0].StatusCode)) { replyTo.Tell(new WriteAttributeResult(true, null)); @@ -898,6 +1189,62 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers { replyTo.Tell(new WriteAttributeResult(false, ex.Message)); } + finally + { + if (!abandoned) cts.Dispose(); + self.Tell(new WriteCompleted(msg.TagId, incarnation)); + } + } + + /// Frees the write's slot and tag and drains the queue: the oldest entries whose tag is free are + /// launched (in arrival order) while slots remain; entries older than the budget are failed instead. A + /// completion stamped with another incarnation (a write launched before a supervision restart) is ignored — + /// it never held a slot or tag of THIS incarnation. + private void OnWriteCompleted(WriteCompleted msg) + { + if (msg.Incarnation != _incarnation) return; + if (_writesInFlight > 0) _writesInFlight--; + _tagsInFlight.Remove(msg.TagId); + DrainWriteQueue(); + } + + private void DrainWriteQueue() + { + if (_driver is not IWritable writable) return; + var now = DateTime.UtcNow; + var node = _pendingWrites.First; + while (node is not null && _writesInFlight < _maxWritesInFlight) + { + var next = node.Next; + var (write, replyTo, enqueuedUtc) = node.Value; + if (now - enqueuedUtc > _queuedWriteBudget) + { + // Already reported as a timeout upstream (the host's Ask is the budget) and the node reverted — + // writing it now would push a stale value to the device. BadTimeout, never written. + _pendingWrites.Remove(node); + replyTo.Tell(new WriteAttributeResult(false, + $"StatusCode=0x800A0000 (BadTimeout: queued for {(now - enqueuedUtc).TotalSeconds:F1}s, over the {_queuedWriteBudget.TotalSeconds:F0}s write budget)")); + } + else if (!_tagsInFlight.Contains(write.TagId)) + { + _pendingWrites.Remove(node); + LaunchWrite(writable, write, replyTo); + } + + node = next; + } + } + + /// Replies to every queued write with ; called on each exit from + /// Connected. In-flight writes are left to finish — they already hold the device and reply with its + /// real outcome. + private void FailQueuedWrites(string reason) + { + while (_pendingWrites.First is { } node) + { + _pendingWrites.RemoveFirst(); + node.Value.ReplyTo.Tell(new WriteAttributeResult(false, reason)); + } } /// @@ -1145,6 +1492,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers { _log.Warning("DriverInstance {Id}: disconnect observed ({Reason}); reconnecting", _driverInstanceId, reason); DetachSubscription(); + FailQueuedWrites("driver not connected"); RecordFault(); Become(Reconnecting); PublishHealthSnapshot(); @@ -1542,15 +1890,91 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers } DetachSubscription(); + FailQueuedWrites("driver stopped"); // MUST happen: the IDriver instance can outlive this actor (the host respawns a child around the // same driver object), so a missing unsubscribe accumulates a handler per respawn holding a dead Self. DetachRediscoverySource(); DetachHostStatusSource(); DetachConnectionLossSource(); - try { _driver.ShutdownAsync(CancellationToken.None).GetAwaiter().GetResult(); } - catch (Exception ex) { _log.Warning(ex, "DriverInstance {Id}: ShutdownAsync threw on PostStop", _driverInstanceId); } + + // An attempt still inside the driver is cancelled before the shutdown below runs underneath it (on a + // restart the next incarnation re-initialises the same object). Best-effort: the task owns the CTS + // and may have disposed it already. + try { _initCts?.Cancel(); } + catch (ObjectDisposedException) { /* the attempt already finished */ } + + // BOUNDED by ONE shared budget across shutdown + dispose. PostStop runs on a dispatcher thread and this + // blocks it, so an unbounded wait on a driver whose shutdown hangs against a dead socket / COM proxy pinned + // that thread for the whole wait, and a few such stops starved the dispatcher. A shutdown that overruns is + // abandoned, and the dispose gets whatever is left of the same budget (never a second full one). + var budget = Stopwatch.StartNew(); + using (var cts = new CancellationTokenSource(_shutdownTimeout)) + { + try { _driver.ShutdownAsync(cts.Token).WaitAsync(_shutdownTimeout).GetAwaiter().GetResult(); } + catch (TimeoutException) + { + _log.Warning("DriverInstance {Id}: ShutdownAsync did not complete within {Bound} on PostStop; abandoning it", + _driverInstanceId, _shutdownTimeout); + } + catch (Exception ex) { _log.Warning(ex, "DriverInstance {Id}: ShutdownAsync threw on PostStop", _driverInstanceId); } + } + + // Dispose ONLY on a real stop. A supervision restart (PreRestart → this PostStop) re-uses the same driver + // object for the next incarnation, which the host re-initialises. Everything else that stops a child — + // deploy removal, operator Restart, respawn on a config change, host shutdown — is final, and the driver + // is never touched again: before this, nothing in the runtime ever disposed an IDriver, so a Calculation + // driver's compiled-script AssemblyLoadContexts and an OpcUaClient driver's LoggerFactory / gate / CTS + // leaked once per respawn (02/L1). Bounded like the shutdown, and runs even after a timed-out shutdown — + // leaking the object is strictly worse than disposing one whose shutdown is still wedged. + if (!_restarting) + { + var remaining = _shutdownTimeout - budget.Elapsed; + DisposeDriver(remaining > MinDisposeBudget ? remaining : MinDisposeBudget); + } + OtOpcUaTelemetry.DriverInstanceLifecycle.Add(1, new KeyValuePair("event", "stop"), new KeyValuePair("driver_type", _driver.DriverType)); } + + /// Floor on the dispose's share of the PostStop budget once a timed-out shutdown has used it up — + /// enough for an idempotent dispose to run, not enough to double the stop's thread time. + private static readonly TimeSpan MinDisposeBudget = TimeSpan.FromMilliseconds(250); + + /// Final teardown of the wrapped driver: preferred, else + /// , never both. A driver implementing neither needs nothing here. Runs even after + /// a timed-out shutdown, so a driver's dispose must be safe CONCURRENTLY with its own still-running + /// . + /// The remaining PostStop budget available to an async dispose. + private void DisposeDriver(TimeSpan bound) + { + try + { + switch (_driver) + { + case IAsyncDisposable asyncDisposable: + asyncDisposable.DisposeAsync().AsTask().WaitAsync(bound).GetAwaiter().GetResult(); + break; + case IDisposable disposable: + disposable.Dispose(); + break; + default: + return; + } + + _log.Info("DriverInstance {Id}: disposed {Type} driver on stop", _driverInstanceId, _driver.DriverType); + OtOpcUaTelemetry.DriverInstanceLifecycle.Add(1, + new KeyValuePair("event", "dispose"), + new KeyValuePair("driver_type", _driver.DriverType)); + } + catch (TimeoutException) + { + _log.Warning("DriverInstance {Id}: DisposeAsync did not complete within {Bound} on PostStop; abandoning it", + _driverInstanceId, bound); + } + catch (Exception ex) + { + _log.Warning(ex, "DriverInstance {Id}: disposing the driver threw on PostStop", _driverInstanceId); + } + } } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Health/DbHealthProbeActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Health/DbHealthProbeActor.cs index 5ea6c44f..8134c914 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Health/DbHealthProbeActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Health/DbHealthProbeActor.cs @@ -7,21 +7,37 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.Health; /// /// Single-flight cached health probe against the ConfigDb. Reads cached state via -/// Ask<DbHealthStatus>; a single SELECT 1 runs at most every RefreshInterval. +/// Ask<DbHealthStatus>; a single SELECT runs at most every RefreshInterval. /// Consumed by both the host's /health/ready endpoint and /// RedundancyStateActor's stale calc. /// +/// +/// The probe runs OFF the mailbox (a thread-pool task that pipes its outcome back), with a bounded +/// command timeout. It used to run synchronously on the actor thread, relying on SqlClient's implicit +/// 15 s connect / 30 s command timeouts — so a hanging SQL Server pinned a dispatcher thread for up to +/// 30 s per tick, later ticks queued behind it, and the cached answer was +/// unreachable for exactly as long as the database was. Single-flight: a tick that lands while a probe +/// is still running is skipped rather than stacked. +/// public sealed class DbHealthProbeActor : ReceiveActor, IWithTimers { public static readonly TimeSpan RefreshInterval = TimeSpan.FromSeconds(5); + /// Bound on the probe's SQL command, relational providers only (the in-memory provider the + /// actor tests run against has no command timeout). + public static readonly TimeSpan CommandTimeout = TimeSpan.FromSeconds(2); + public sealed class GetStatus { public static readonly GetStatus Instance = new(); private GetStatus() { } } public sealed record DbHealthStatus(bool Reachable, DateTime AsOfUtc, string? LastError); public sealed class Tick { public static readonly Tick Instance = new(); private Tick() { } } + /// Piped back by the off-thread probe with its outcome. + private sealed record ProbeCompleted(DbHealthStatus Status); + private readonly IDbContextFactory _dbFactory; private readonly ILoggingAdapter _log = Context.GetLogger(); private DbHealthStatus _last = new(false, DateTime.MinValue, "not probed yet"); + private bool _probeInFlight; /// Gets or sets the timer scheduler for periodic health probes. public ITimerScheduler Timers { get; set; } = null!; @@ -38,28 +54,54 @@ public sealed class DbHealthProbeActor : ReceiveActor, IWithTimers { _dbFactory = dbFactory; Receive(_ => Sender.Tell(_last)); - Receive(_ => RunProbe()); + Receive(_ => StartProbe()); + Receive(msg => + { + _probeInFlight = false; + _last = msg.Status; + if (!msg.Status.Reachable) + _log.Warning("DbHealthProbe: probe failed: {Error}", msg.Status.LastError); + }); } /// protected override void PreStart() { - RunProbe(); + StartProbe(); Timers.StartPeriodicTimer("probe", Tick.Instance, RefreshInterval); } - private void RunProbe() + private void StartProbe() + { + if (_probeInFlight) + { + _log.Debug("DbHealthProbe: previous probe still running; skipping this tick"); + return; + } + + _probeInFlight = true; + var factory = _dbFactory; + // Never touches actor state; the outcome comes back as a message. + Task.Run(() => Probe(factory)).PipeTo( + Self, + success: status => new ProbeCompleted(status), + failure: ex => new ProbeCompleted(new DbHealthStatus(false, DateTime.UtcNow, ex.Message))); + } + + private static DbHealthStatus Probe(IDbContextFactory factory) { try { - using var db = _dbFactory.CreateDbContext(); + using var db = factory.CreateDbContext(); + // Local to this lease — the pool resets it on return (see DriverHostActor.LoadArtifactBlob). + if (db.Database.IsRelational()) + db.Database.SetCommandTimeout(CommandTimeout); _ = db.Deployments.AsNoTracking().Take(1).ToList(); - _last = new DbHealthStatus(true, DateTime.UtcNow, null); + return new DbHealthStatus(true, DateTime.UtcNow, null); } catch (Exception ex) { - _last = new DbHealthStatus(false, DateTime.UtcNow, ex.Message); - _log.Warning(ex, "DbHealthProbe: probe failed"); + return new DbHealthStatus(false, DateTime.UtcNow, ex.Message); } } } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Telemetry/TelemetryLocalHub.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Telemetry/TelemetryLocalHub.cs index 57d11973..1e58c54b 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Telemetry/TelemetryLocalHub.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Telemetry/TelemetryLocalHub.cs @@ -27,7 +27,27 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.Telemetry; /// public sealed class TelemetryLocalHub : ITelemetryLocalHub { + /// + /// How long an eviction tombstone (see ) is kept. It only has to outlive the + /// removed child's last late write — an in-flight capability call that completes after + /// StopChild, bounded by the longest capability timeout (seconds) plus the child's own stop — + /// so ten minutes is a wide margin. Without a sweep a driver removed for good kept its id forever. + /// + public static readonly TimeSpan EvictionTombstoneTtl = TimeSpan.FromMinutes(10); + private readonly object _gate = new(); + private readonly TimeProvider _time; + + /// Initializes a hub on the system clock. + public TelemetryLocalHub() : this(TimeProvider.System) { } + + /// Initializes a hub on — the tombstone-sweep test seam. + /// The clock the eviction tombstones are stamped and swept by. + public TelemetryLocalHub(TimeProvider time) + { + ArgumentNullException.ThrowIfNull(time); + _time = time; + } // Plain Dictionary, NOT ConcurrentDictionary: every access is already under _gate, and // ConcurrentDictionary.Values would materialize a fresh List (and take its own internal lock) on @@ -42,9 +62,10 @@ public sealed class TelemetryLocalHub : ITelemetryLocalHub private readonly ConcurrentDictionary _healthCache = new(); private readonly ConcurrentDictionary<(string InstanceId, string HostName), TelemetryItem.Resilience> _resilienceCache = new(); - // Instances evicted and not since readmitted. Their late Health/Resilience items are dropped, so a - // removed driver's still-stopping child cannot re-prime the cache it was just evicted from. Under _gate. - private readonly HashSet _evicted = new(StringComparer.Ordinal); + // Instances evicted and not since readmitted, stamped with the eviction time. Their late Health/Resilience + // items are dropped, so a removed driver's still-stopping child cannot re-prime the cache it was just + // evicted from. Swept past EvictionTombstoneTtl on the next Evict (never on the Emit hot path). Under _gate. + private readonly Dictionary _evicted = new(StringComparer.Ordinal); /// public void Emit(TelemetryItem item) @@ -55,9 +76,9 @@ public sealed class TelemetryLocalHub : ITelemetryLocalHub { switch (item) { - case TelemetryItem.Health h when _evicted.Contains(h.E.DriverInstanceId): + case TelemetryItem.Health h when _evicted.ContainsKey(h.E.DriverInstanceId): return; - case TelemetryItem.Resilience r when _evicted.Contains(r.E.DriverInstanceId): + case TelemetryItem.Resilience r when _evicted.ContainsKey(r.E.DriverInstanceId): return; case TelemetryItem.Health h: _healthCache[h.E.DriverInstanceId] = h; @@ -80,7 +101,9 @@ public sealed class TelemetryLocalHub : ITelemetryLocalHub lock (_gate) { - _evicted.Add(driverInstanceId); + var now = _time.GetUtcNow(); + SweepExpiredTombstones(now); + _evicted[driverInstanceId] = now; _healthCache.TryRemove(driverInstanceId, out _); foreach (var key in _resilienceCache.Keys) if (string.Equals(key.InstanceId, driverInstanceId, StringComparison.Ordinal)) @@ -97,6 +120,24 @@ public sealed class TelemetryLocalHub : ITelemetryLocalHub _evicted.Remove(driverInstanceId); } + /// Drops tombstones older than . Called under + /// from only — the set is bounded by distinct removed ids, so a linear pass per + /// eviction is cheap, and Emit never pays for it. + private void SweepExpiredTombstones(DateTimeOffset now) + { + if (_evicted.Count == 0) return; + List? expired = null; + foreach (var (id, evictedAt) in _evicted) + { + if (now - evictedAt >= EvictionTombstoneTtl) + (expired ??= new List()).Add(id); + } + + if (expired is null) return; + foreach (var id in expired) + _evicted.Remove(id); + } + /// public ITelemetrySubscription Subscribe(int boundedCapacity) { diff --git a/tests/Core/ZB.MOM.WW.OtOpcUa.Commons.Tests/Browsing/DiscoveryDriverBrowserDisposeTests.cs b/tests/Core/ZB.MOM.WW.OtOpcUa.Commons.Tests/Browsing/DiscoveryDriverBrowserDisposeTests.cs new file mode 100644 index 00000000..5db9edd8 --- /dev/null +++ b/tests/Core/ZB.MOM.WW.OtOpcUa.Commons.Tests/Browsing/DiscoveryDriverBrowserDisposeTests.cs @@ -0,0 +1,125 @@ +using Microsoft.Extensions.Logging.Abstractions; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.Browsing; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.Commons.Tests.Browsing; + +/// +/// Review 02 L1 (host context, 06/G-1): the browser's throwaway capture driver must be disposed, not +/// only shut down. A Galaxy driver's ShutdownAsync left its whole gRPC runtime alive until +/// DisposeAsync, which nothing called — so every /raw browse leaked two channels, an MX session +/// and two stream loops. +/// +public sealed class DiscoveryDriverBrowserDisposeTests +{ + private static readonly TimeSpan ShortOpen = TimeSpan.FromSeconds(5); + private static readonly TimeSpan ShortShutdown = TimeSpan.FromMilliseconds(300); + + private static DiscoveryDriverBrowser Browser(IDriverFactory factory) => + new(factory, NullLogger.Instance, ShortOpen, ShortShutdown); + + [Fact] + public async Task OpenAsync_disposes_the_capture_driver_after_shutdown() + { + var driver = new DisposableFakeDiscoveryDriver(); + var browser = Browser(new FakeDriverFactory((_, _, _) => driver)); + + var session = await browser.OpenAsync("Fake", "{}", CancellationToken.None); + + session.ShouldNotBeNull(); + driver.Inner.Trace.ShouldBe(new[] { "init", "discover", "shutdown" }); + driver.DisposeCalls.ShouldBe(1); + driver.DisposedAfterShutdown.ShouldBeTrue("dispose is the final teardown, after the shutdown"); + } + + [Fact] + public async Task OpenAsync_disposes_the_capture_driver_even_when_discovery_throws() + { + var driver = new DisposableFakeDiscoveryDriver(new FakeDiscoveryDriver + { + OnDiscover = (_, _) => throw new InvalidOperationException("boom-discovery"), + }); + var browser = Browser(new FakeDriverFactory((_, _, _) => driver)); + + await Should.ThrowAsync( + () => browser.OpenAsync("Fake", "{}", CancellationToken.None)); + + driver.DisposeCalls.ShouldBe(1); + } + + [Fact] + public async Task OpenAsync_disposes_the_capture_driver_even_when_its_shutdown_hangs() + { + var never = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var driver = new DisposableFakeDiscoveryDriver(new FakeDiscoveryDriver { OnShutdown = _ => never.Task }); + var browser = Browser(new FakeDriverFactory((_, _, _) => driver)); + + await browser.OpenAsync("Fake", "{}", CancellationToken.None); + + driver.DisposeCalls.ShouldBe(1, "a shutdown that timed out must not skip the dispose"); + } + + [Fact] + public async Task CanBrowse_disposes_its_probe_driver() + { + var driver = new DisposableFakeDiscoveryDriver(); + var browser = Browser(new FakeDriverFactory((_, _, _) => driver)); + + browser.CanBrowse("Fake", "{}").ShouldBeTrue(); + + // The CanBrowse teardown is fire-and-forget; wait for it. + await driver.Disposed.Task.WaitAsync(TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken); + driver.DisposeCalls.ShouldBe(1); + } + + /// A wrapped with + + /// (both, so the "prefer the async one, never both" rule is observable: exactly ONE dispose call lands). + private sealed class DisposableFakeDiscoveryDriver : IDriver, ITagDiscovery, IAsyncDisposable, IDisposable + { + private int _disposeCalls; + + public DisposableFakeDiscoveryDriver(FakeDiscoveryDriver? inner = null) => Inner = inner ?? new FakeDiscoveryDriver(); + + public FakeDiscoveryDriver Inner { get; } + public int DisposeCalls => Volatile.Read(ref _disposeCalls); + public bool DisposedAfterShutdown { get; private set; } + public TaskCompletionSource Disposed { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); + + public string DriverInstanceId => Inner.DriverInstanceId; + public string DriverType => Inner.DriverType; + public bool SupportsOnlineDiscovery => Inner.SupportsOnlineDiscovery; + public DiscoveryRediscoverPolicy RediscoverPolicy => Inner.RediscoverPolicy; + + public Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken) => + Inner.InitializeAsync(driverConfigJson, cancellationToken); + + public Task ReinitializeAsync(string driverConfigJson, CancellationToken cancellationToken) => + Inner.ReinitializeAsync(driverConfigJson, cancellationToken); + + public Task ShutdownAsync(CancellationToken cancellationToken) => Inner.ShutdownAsync(cancellationToken); + + public DriverHealth GetHealth() => Inner.GetHealth(); + +#pragma warning disable OTOPCUA0001 // test double delegating to a test double — no resilience pipeline applies + public Task DiscoverAsync(IAddressSpaceBuilder builder, CancellationToken cancellationToken) => + Inner.DiscoverAsync(builder, cancellationToken); +#pragma warning restore OTOPCUA0001 + + public ValueTask DisposeAsync() + { + Record(); + return ValueTask.CompletedTask; + } + + public void Dispose() => Record(); + + private void Record() + { + Interlocked.Increment(ref _disposeCalls); + DisposedAfterShutdown = Inner.ShutdownCalls > 0; + Disposed.TrySetResult(); + } + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/Telemetry/TelemetryDialDeadlineTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/Telemetry/TelemetryDialDeadlineTests.cs new file mode 100644 index 00000000..d19506ca --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/Telemetry/TelemetryDialDeadlineTests.cs @@ -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; + +/// +/// Review 02 U2: TelemetryDial:CallTimeoutSeconds was bound, defaulted and documented, and consumed +/// nowhere. It now bounds a dial's establishment: 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. +/// +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>( + 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(); + } + + /// Records each invocation; optionally delivers one health event on open; ends only when cancelled. + private sealed class FakeDialLoop + { + private readonly ConcurrentQueue _invocations = new(); + + public bool DeliverOnOpen { get; init; } + + public IReadOnlyList 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 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 AlarmBroadcaster { get; } = new(); + public NullBroadcaster ScriptBroadcaster { get; } = new(); + public RecordingHealthStore HealthStore { get; } = new(); + public NullResilienceStore ResilienceStore { get; } = new(); + } + + private sealed class NullBroadcaster : IInProcessBroadcaster + { + public event Action? Received; + public event Action? 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 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 NullResilienceStore : IDriverResilienceStatusStore + { + public event Action? SnapshotChanged; + public void Upsert(DriverResilienceStatusChanged snapshot) => SnapshotChanged?.Invoke(snapshot); + public IReadOnlyList GetForInstance(string driverInstanceId) => Array.Empty(); + public IReadOnlyCollection GetAll() => Array.Empty(); + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/CalculationDependencyFlowTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/CalculationDependencyFlowTests.cs index 008004d5..d342023c 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/CalculationDependencyFlowTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/CalculationDependencyFlowTests.cs @@ -64,7 +64,7 @@ public sealed class CalculationDependencyFlowTests : RuntimeActorTestBase { published.Enqueue(e); var quality = e.Snapshot.StatusCode == 0u ? OpcUaStatus.Good : OpcUaStatus.Bad; - mux.Tell(new DriverInstanceActor.AttributeValuePublished( + mux.Tell(ValuePublishes.Single( "calc-drv", e.FullReference, e.Snapshot.Value, quality, Ts)); }; @@ -75,7 +75,7 @@ public sealed class CalculationDependencyFlowTests : RuntimeActorTestBase // compute dedups) until the computed value has flowed all the way through. AwaitAssert(() => { - mux.Tell(new DriverInstanceActor.AttributeValuePublished( + mux.Tell(ValuePublishes.Single( "src-drv", "src/a", 21.0, OpcUaStatus.Good, Ts)); // Leg 1 — dep flow: src/a=21 → A computes 42 and publishes on calc/engine/out. @@ -118,14 +118,14 @@ public sealed class CalculationDependencyFlowTests : RuntimeActorTestBase { published.Enqueue(e); var quality = e.Snapshot.StatusCode == 0u ? OpcUaStatus.Good : OpcUaStatus.Bad; - mux.Tell(new DriverInstanceActor.AttributeValuePublished("calc-drv", e.FullReference, e.Snapshot.Value, quality, Ts)); + mux.Tell(ValuePublishes.Single("calc-drv", e.FullReference, e.Snapshot.Value, quality, Ts)); }; var adapter = Sys.ActorOf(DependencyConsumerMuxAdapter.Props(driver, mux), "dep-adapter2"); // Leg 1 — before redeploy: src/a=2 → keep=20, remove=7. AwaitAssert(() => { - mux.Tell(new DriverInstanceActor.AttributeValuePublished("src-drv", "src/a", 2.0, OpcUaStatus.Good, Ts)); + mux.Tell(ValuePublishes.Single("src-drv", "src/a", 2.0, OpcUaStatus.Good, Ts)); published.ShouldContain(e => e.FullReference == "calc/keep" && Equals(e.Snapshot.Value, 20.0)); published.ShouldContain(e => e.FullReference == "calc/remove" && Equals(e.Snapshot.Value, 7.0)); }, PresenceBudget, TimeSpan.FromMilliseconds(100)); @@ -150,8 +150,8 @@ public sealed class CalculationDependencyFlowTests : RuntimeActorTestBase // new result; the REMOVED tag never publishes a value derived from the post-redeploy input. AwaitAssert(() => { - mux.Tell(new DriverInstanceActor.AttributeValuePublished("src-drv", "src/b", 50.0, OpcUaStatus.Good, Ts)); - mux.Tell(new DriverInstanceActor.AttributeValuePublished("src-drv", "src/a", 3.0, OpcUaStatus.Good, Ts)); + mux.Tell(ValuePublishes.Single("src-drv", "src/b", 50.0, OpcUaStatus.Good, Ts)); + mux.Tell(ValuePublishes.Single("src-drv", "src/a", 3.0, OpcUaStatus.Good, Ts)); // (a) new dep B flows through the RE-REGISTERED interest → new tag computes: 50 + 100 = 150. published.ShouldContain(e => e.FullReference == "calc/add" && Equals(e.Snapshot.Value, 150.0)); diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorDiagnosticsTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorDiagnosticsTests.cs new file mode 100644 index 00000000..86ca424b --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorDiagnosticsTests.cs @@ -0,0 +1,79 @@ +using Akka.Actor; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.Interfaces; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Fleet; +using ZB.MOM.WW.OtOpcUa.Commons.Types; +using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers; + +/// +/// Review 02 U1: GetDiagnostics answered a placeholder — Guid.Empty, "Spawned", zero devices and +/// "now" for every child, in every behaviour — so the snapshot looked live and said nothing. It now carries +/// each child's row id, instance id, last-published health state, device counts and last change time. +/// +public sealed class DriverHostActorDiagnosticsTests : RuntimeActorTestBase +{ + private static readonly RevisionHash RevA = RevisionHash.Parse(new string('a', 64)); + private static readonly NodeId TestNode = NodeId.Parse("node-1:4053"); + private const string DriverId = "DI-1"; + + [Fact] + public async Task A_connected_child_reports_its_row_id_name_health_state_and_device_count() + { + var db = NewInMemoryDbFactory(); + var rowId = Guid.NewGuid(); + var before = DateTime.UtcNow.AddSeconds(-1); + var factory = new RecordingSubscribableDriverFactory("Modbus"); + var deploymentId = V3ArtifactSeeder.SeedV3Deployment( + db, RevA, new[] { (DriverId, "Line1-Modbus", "dev1", "speed", "Float", false) }, + Array.Empty<(string, string, string, string, string?)>(), + enabled: true, driverRowIds: new Dictionary { [DriverId] = rowId }); + var (host, _, _) = V3ArtifactSeeder.SpawnHostAndApply(this, db, TestNode, deploymentId, RevA, factory); + + var snapshot = await AwaitDiagnosticsAsync(host, s => s.Drivers.Any(d => d.State == "Healthy")); + + snapshot.NodeId.ShouldBe(TestNode); + snapshot.CurrentRevision.ShouldBe(RevA); + snapshot.RunningFromCache.ShouldBeFalse(); + var driver = snapshot.Drivers.ShouldHaveSingleItem(); + driver.DriverInstanceId.ShouldBe(rowId, "the real row id, not Guid.Empty"); + driver.Name.ShouldBe(DriverId, "the string instance id every other surface keys on"); + driver.State.ShouldBe("Healthy"); + driver.ConnectedDevices.ShouldBe(1); + driver.FaultedDevices.ShouldBe(0); + driver.LastChangeUtc.ShouldBeGreaterThan(before); + driver.LastChangeUtc.ShouldBeLessThanOrEqualTo(DateTime.UtcNow); + } + + [Fact] + public async Task A_stubbed_child_reports_Stubbed_with_no_devices() + { + var db = NewInMemoryDbFactory(); + // The factory answers for a type the artifact does not use, so the host stubs the child. + var factory = new RecordingSubscribableDriverFactory("NotModbus"); + var deploymentId = V3ArtifactSeeder.SeedV3Deployment( + db, RevA, new[] { (DriverId, "Modbus", "dev1", "speed", "Float", false) }, + Array.Empty<(string, string, string, string, string?)>(), enabled: true); + var (host, _, _) = V3ArtifactSeeder.SpawnHostAndApply(this, db, TestNode, deploymentId, RevA, factory); + + var snapshot = await AwaitDiagnosticsAsync(host, s => s.Drivers.Count == 1); + + var driver = snapshot.Drivers.Single(); + driver.State.ShouldBe("Stubbed"); + driver.ConnectedDevices.ShouldBe(0); + driver.FaultedDevices.ShouldBe(0); + } + + private async Task AwaitDiagnosticsAsync(IActorRef host, Func ready) + { + NodeDiagnosticsSnapshot? snapshot = null; + await AwaitAssertAsync(async () => + { + snapshot = await host.Ask(new GetDiagnostics(CorrelationId.NewId()), PresenceBudget); + ready(snapshot).ShouldBeTrue(); + }, PresenceBudget); + return snapshot!; + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorReprovisionStampTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorReprovisionStampTests.cs new file mode 100644 index 00000000..509e7ea9 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorReprovisionStampTests.cs @@ -0,0 +1,63 @@ +using Akka.Actor; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.Types; +using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers; + +/// +/// Review 02 L2: DriverHostActor._lastReprovisionUtc was never pruned — a driver removed by a deploy +/// after a supervision crash kept its stamp, so the map grew with every id ever reprovisioned-then-removed, +/// and (the observable half) a driver removed and later re-added inherited the OLD stamp: its first crash +/// after the re-add was deferred by the backoff as if it were crash-looping, although the new incarnation had +/// never crashed. StopChild now drops the stamp. +/// +public sealed class DriverHostActorReprovisionStampTests : RuntimeActorTestBase +{ + private static readonly RevisionHash RevA = RevisionHash.Parse(new string('a', 64)); + private static readonly RevisionHash RevB = RevisionHash.Parse(new string('b', 64)); + private static readonly RevisionHash RevC = RevisionHash.Parse(new string('c', 64)); + private static readonly NodeId TestNode = NodeId.Parse("node-1:4053"); + private const string DriverId = "DI-1"; + private const string SpeedPath = "Plant/Modbus/dev1/speed"; + + [Fact] + public void A_driver_removed_and_re_added_does_not_inherit_the_old_incarnations_reprovision_stamp() + { + var db = NewInMemoryDbFactory(); + // Far longer than the presence budget below: an inherited stamp would defer the re-added driver's + // reprovision past it, which is the failure this test reports. + var backoff = TimeSpan.FromSeconds(60); + var factory = new RecordingSubscribableDriverFactory("Modbus"); + var withDriver = new[] { (DriverId, "Modbus", "dev1", "speed", "Float", false) }; + var none = Array.Empty<(string, string, string, string, string?)>(); + + var revA = V3ArtifactSeeder.SeedV3Deployment(db, RevA, withDriver, none, enabled: true); + var (host, _, _, coordinator) = V3ArtifactSeeder.SpawnHostAndApplyWithCoordinator( + this, db, TestNode, revA, RevA, factory, reprovisionBackoff: backoff); + AwaitAssert(() => factory.Created.Count.ShouldBe(1), PresenceBudget); + AwaitAssert(() => factory.Created[0].SubscribeCount.ShouldBe(1), PresenceBudget); + + // Crash 1 on the first incarnation: reprovisioned immediately, which stamps the id. + factory.Created[0].FireMalformedDataChange(SpeedPath); + AwaitAssert(() => factory.Created[0].InitializeCount.ShouldBe(2), PresenceBudget); + + // Deploy B removes the driver (StopChild), deploy C re-adds it: a brand-new driver object. + var revB = V3ArtifactSeeder.SeedV3Deployment( + db, RevB, Array.Empty<(string, string, string, string, string, bool)>(), none, enabled: true); + V3ArtifactSeeder.Dispatch(host, coordinator, revB, RevB); + var revC = V3ArtifactSeeder.SeedV3Deployment(db, RevC, withDriver, none, enabled: true); + V3ArtifactSeeder.Dispatch(host, coordinator, revC, RevC); + AwaitAssert(() => factory.Created.Count.ShouldBe(2), PresenceBudget); + var reAdded = factory.Created[1]; + AwaitAssert(() => reAdded.SubscribeCount.ShouldBe(1), PresenceBudget); + + // Its FIRST crash must reprovision immediately — it has no history of its own. + reAdded.FireMalformedDataChange(SpeedPath); + AwaitAssert( + () => reAdded.InitializeCount.ShouldBe( + 2, "the re-added driver's first reprovision was deferred by the REMOVED incarnation's stamp"), + duration: TimeSpan.FromSeconds(5)); + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorStaleDropTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorStaleDropTests.cs new file mode 100644 index 00000000..442f3909 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorStaleDropTests.cs @@ -0,0 +1,66 @@ +using Akka.Actor; +using Akka.Event; +using Microsoft.EntityFrameworkCore; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.OpcUa; +using ZB.MOM.WW.OtOpcUa.Commons.Types; +using ZB.MOM.WW.OtOpcUa.Configuration; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Runtime.Drivers; +using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers; + +/// +/// Review 02 H5: the Stale behaviour registered no receive for a driver child's value/alarm bursts (nor +/// for a late ArtifactForApply / ApplyWedged pipe-back), so any that landed there hit +/// Unhandled — dead-letter noise and a dropped message with no log naming why. They are now explicit +/// drops. The two private pipe-back records cannot be constructed from a test; the two public child +/// messages are pinned here, and the private pair by the same registration pattern (compile-time). +/// +public sealed class DriverHostActorStaleDropTests : RuntimeActorTestBase +{ + private static readonly NodeId TestNode = NodeId.Parse("stale-drop-test"); + + [Fact] + public void Child_value_and_alarm_bursts_landing_in_Stale_are_dropped_not_unhandled() + { + var unhandled = CreateTestProbe(); + Sys.EventStream.Subscribe(unhandled.Ref, typeof(UnhandledMessage)); + + // ThrowingDbFactory ⇒ Bootstrap's catch ⇒ Become(Stale). + var host = Sys.ActorOf(DriverHostActor.Props( + new ThrowingDbFactory(), TestNode, CreateTestProbe().Ref, + localRoles: new HashSet { "driver" })); + // Barrier: Stale answers RouteNodeWrite synchronously, so a reply proves the host is in Stale. + var asker = CreateTestProbe(); + host.Tell(new DriverHostActor.RouteNodeWrite("eq-x/speed", 0.0, AddressSpaceRealm.Uns), asker.Ref); + asker.ExpectMsg(PresenceBudget).Reason!.ShouldContain("stale"); + + // Positive control: an unknown message IS reported as unhandled by this host. + host.Tell("not a message the host knows"); + unhandled.ExpectMsg(m => ReferenceEquals(m.Recipient, host) || m.Recipient.Equals(host), PresenceBudget); + + host.Tell(new DriverInstanceActor.AttributeValueBatchPublished( + "DI-1", new[] { new DriverInstanceActor.AttributeValueItem("Plant/x/y", 1.0, OpcUaStatus.Good, DateTime.UtcNow) })); + host.Tell(new DriverInstanceActor.AttributeAlarmPublished( + "DI-1", new AlarmEventArgs(new Handle(), "Plant/x/alarm", "Plant/x/alarm", "AlarmCondition", "test", + AlarmSeverity.High, DateTime.UtcNow))); + // Barrier again, then the absence window. + host.Tell(new DriverHostActor.RouteNodeWrite("eq-x/speed", 0.0, AddressSpaceRealm.Uns), asker.Ref); + asker.ExpectMsg(PresenceBudget); + + unhandled.ExpectNoMsg(TimeSpan.FromMilliseconds(500)); + } + + private sealed class Handle : IAlarmSubscriptionHandle + { + public string DiagnosticId => "stale-test"; + } + + private sealed class ThrowingDbFactory : IDbContextFactory + { + public OtOpcUaConfigDbContext CreateDbContext() => throw new InvalidOperationException("db unreachable"); + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorConcurrentWriteTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorConcurrentWriteTests.cs new file mode 100644 index 00000000..f6f99002 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorConcurrentWriteTests.cs @@ -0,0 +1,288 @@ +using System.Collections.Concurrent; +using System.Diagnostics; +using Akka.Actor; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.OpcUa; +using ZB.MOM.WW.OtOpcUa.Commons.Types; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Runtime.Drivers; +using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers; + +/// +/// Review 02 P2: WriteAttribute was a ReceiveAsync, which suspends the mailbox for the whole +/// device round-trip (up to the 5 s bound). Writes therefore serialised — the SECOND of two operator writes to +/// a ~3 s device missed the host's 8 s Ask and was reverted although the device took it — and every data +/// change queued behind the write. Writes now run off the mailbox under a per-driver in-flight cap; the +/// subscribe/unsubscribe pair that must serialise against the subscription handle stays ReceiveAsync. +/// +public sealed class DriverInstanceActorConcurrentWriteTests : RuntimeActorTestBase +{ + private static readonly TimeSpan WriteLatency = TimeSpan.FromMilliseconds(1200); + + private (IActorRef Actor, Akka.TestKit.TestProbe Parent) Connect( + IDriver driver, int? maxWritesInFlight = null, TimeSpan? queuedWriteBudget = null, TimeSpan? writeTimeout = null) + { + var parent = CreateTestProbe(); + var actor = parent.ChildActorOf(DriverInstanceActor.Props( + driver, commsLossTimeout: TimeSpan.Zero, maxWritesInFlight: maxWritesInFlight, + queuedWriteBudget: queuedWriteBudget, writeTimeout: writeTimeout)); + actor.Tell(new DriverInstanceActor.InitializeRequested("{}")); + parent.ExpectMsg(m => m.Connected, PresenceBudget); + return (actor, parent); + } + + [Fact] + public async Task Two_concurrent_slow_writes_both_complete_within_one_device_round_trip() + { + var driver = new SlowWritableDriver { Latency = WriteLatency }; + var (actor, _) = Connect(driver); + + var sw = Stopwatch.StartNew(); + var askBudget = TimeSpan.FromSeconds(8); // the host's Ask + var first = actor.Ask( + new DriverInstanceActor.WriteAttribute("tag-1", 1), askBudget); + var second = actor.Ask( + new DriverInstanceActor.WriteAttribute("tag-2", 2), askBudget); + var results = await Task.WhenAll(first, second); + sw.Stop(); + + results.ShouldAllBe(r => r.Success); + driver.Writes.Select(w => w.FullReference).ShouldBe(new[] { "tag-1", "tag-2" }, ignoreOrder: true); + // Both were inside the driver at once — that is the proof; the wall clock is only a sanity ceiling + // (serialised writes would take two latencies, and the host's Ask is 8 s). + driver.MaxConcurrentWrites.ShouldBe(2, "the second write waited for the first — writes are still serialised on the mailbox"); + sw.Elapsed.ShouldBeLessThan(TimeSpan.FromSeconds(8)); + } + + [Fact] + public async Task Data_changes_are_delivered_while_a_write_is_pending() + { + var driver = new SlowWritableDriver { HoldWrites = true }; + var (actor, parent) = Connect(driver); + await actor.Ask( + new DriverInstanceActor.Subscribe(new[] { "tag-a" }, TimeSpan.FromMilliseconds(100)), PresenceBudget); + + var pending = actor.Ask( + new DriverInstanceActor.WriteAttribute("tag-a", 7), PresenceBudget); + AwaitCondition(() => driver.WritesEntered == 1, PresenceBudget); + + driver.FireDataChange("tag-a", 3.5); + var published = parent.ExpectMsg(PresenceBudget); + published.Items.Single().Value.ShouldBe(3.5); + pending.IsCompleted.ShouldBeFalse("the write is still held — the data change did not wait behind it"); + + // Positive control: releasing the device completes the write with its real outcome. + driver.ReleaseWrites(); + (await pending).Success.ShouldBeTrue(); + } + + [Fact] + public async Task Writes_beyond_the_in_flight_cap_queue_and_drain_in_order() + { + var driver = new SlowWritableDriver { HoldWrites = true }; + var (actor, _) = Connect(driver, maxWritesInFlight: 2); + + var asks = Enumerable.Range(0, 4) + .Select(i => actor.Ask( + new DriverInstanceActor.WriteAttribute($"tag-{i}", i), PresenceBudget)) + .ToArray(); + + AwaitCondition(() => driver.WritesEntered == 2, PresenceBudget); + // Settle: a third write entering the driver would show up here. + await Task.Delay(300); + driver.WritesEntered.ShouldBe(2, "the cap let a third write into the driver"); + + driver.ReleaseWrites(); + var results = await Task.WhenAll(asks); + results.ShouldAllBe(r => r.Success); + // Entry order: the first two slots went to the first two writes, the queued pair entered only after a + // slot freed. Within each pair the order is unspecified — distinct tags run concurrently on the pool; + // only SAME-tag order is guaranteed (pinned by the same-tag test below). + var entered = driver.EnteredRefs; + entered.Count.ShouldBe(4); + entered.Take(2).ShouldBe(new[] { "tag-0", "tag-1" }, ignoreOrder: true); + entered.Skip(2).ShouldBe(new[] { "tag-2", "tag-3" }, ignoreOrder: true); + } + + /// + /// Fix round 1: writes to the SAME tag must keep their order — two concurrent writes v1, v2 could land as + /// v1. A same-tag write waits behind its tag while other tags still run concurrently. + /// + [Fact] + public async Task Writes_to_the_same_tag_are_serialised_in_order_while_other_tags_run_concurrently() + { + var driver = new SlowWritableDriver { HoldWrites = true }; + var (actor, _) = Connect(driver, maxWritesInFlight: 4); + + var first = actor.Ask( + new DriverInstanceActor.WriteAttribute("tag-a", 1), PresenceBudget); + AwaitCondition(() => driver.WritesEntered == 1, PresenceBudget); + var second = actor.Ask( + new DriverInstanceActor.WriteAttribute("tag-a", 2), PresenceBudget); + // Positive control: a DIFFERENT tag is not held behind tag-a. + var other = actor.Ask( + new DriverInstanceActor.WriteAttribute("tag-b", 9), PresenceBudget); + AwaitCondition(() => driver.WritesEntered == 2, PresenceBudget); + driver.EnteredRefs.ShouldBe(new[] { "tag-a", "tag-b" }); + + await Task.Delay(300); + driver.WritesEntered.ShouldBe(2, "the second tag-a write entered the driver while the first was still inside it"); + second.IsCompleted.ShouldBeFalse(); + + driver.ReleaseWrites(); + (await Task.WhenAll(first, second, other)).ShouldAllBe(r => r.Success); + driver.Writes.Where(w => w.FullReference == "tag-a").Select(w => w.Value).ShouldBe(new object[] { 1, 2 }); + } + + /// + /// Fix round 1: a write that sat in the queue longer than the host's Ask budget has already been reported + /// as a timeout (and reverted) upstream — writing it late would push a stale value to the device. + /// + [Fact] + public async Task A_queued_write_older_than_the_budget_is_failed_BadTimeout_instead_of_written_late() + { + var driver = new SlowWritableDriver { HoldWrites = true }; + var (actor, _) = Connect(driver, maxWritesInFlight: 1, queuedWriteBudget: TimeSpan.FromMilliseconds(300)); + + var inFlight = actor.Ask( + new DriverInstanceActor.WriteAttribute("tag-0", 0), PresenceBudget); + AwaitCondition(() => driver.WritesEntered == 1, PresenceBudget); + var stale = actor.Ask( + new DriverInstanceActor.WriteAttribute("tag-1", 1), PresenceBudget); + + await Task.Delay(600); + driver.ReleaseWrites(); + + (await inFlight).Success.ShouldBeTrue(); + var staleResult = await stale; + staleResult.Success.ShouldBeFalse(); + staleResult.Reason!.ShouldContain("800A0000", customMessage: "BadTimeout"); + driver.EnteredRefs.ShouldBe(new[] { "tag-0" }, "the stale write must never reach the device"); + } + + /// + /// Fix round 2: a token-ignoring hung write held its tag forever, so every later write to that tag piled up + /// in the queue. The write is bounded pessimistically; on overrun it answers BadTimeout and releases its + /// slot and tag, and a later write to the same tag reaches the device. + /// + [Fact] + public async Task A_hung_token_ignoring_write_is_abandoned_and_releases_its_tag_for_a_later_write() + { + var driver = new SlowWritableDriver { HoldWrites = true, IgnoreCancellation = true }; + var (actor, _) = Connect(driver, maxWritesInFlight: 1, writeTimeout: TimeSpan.FromMilliseconds(300)); + + var hung = actor.Ask( + new DriverInstanceActor.WriteAttribute("tag-a", 1), PresenceBudget); + AwaitCondition(() => driver.WritesEntered == 1, PresenceBudget); + + var hungResult = await hung; + hungResult.Success.ShouldBeFalse(); + hungResult.Reason!.ShouldContain("800A0000", customMessage: "BadTimeout"); + driver.MaxConcurrentWrites.ShouldBe(1); + + // The tag (and the single slot) are free again: a later tag-a write reaches the device even though the + // abandoned one is still inside it. + var later = actor.Ask( + new DriverInstanceActor.WriteAttribute("tag-a", 2), PresenceBudget); + AwaitCondition(() => driver.WritesEntered == 2, PresenceBudget); + driver.EnteredRefs.ShouldBe(new[] { "tag-a", "tag-a" }); + driver.MaxConcurrentWrites.ShouldBe(2, "positive control: the abandoned write really is still wedged inside the driver"); + + driver.ReleaseWrites(); + (await later).Success.ShouldBeTrue(); + } + + [Fact] + public async Task Queued_writes_fail_fast_when_the_driver_leaves_Connected() + { + var driver = new SlowWritableDriver { HoldWrites = true }; + var (actor, parent) = Connect(driver, maxWritesInFlight: 1); + + var inFlight = actor.Ask( + new DriverInstanceActor.WriteAttribute("tag-0", 0), PresenceBudget); + AwaitCondition(() => driver.WritesEntered == 1, PresenceBudget); + var queued = actor.Ask( + new DriverInstanceActor.WriteAttribute("tag-1", 1), PresenceBudget); + + actor.Tell(new DriverInstanceActor.ForceReconnect()); + parent.ExpectMsg(m => !m.Connected, PresenceBudget); + + var queuedResult = await queued; + queuedResult.Success.ShouldBeFalse(); + queuedResult.Reason.ShouldBe("driver not connected"); + driver.WritesEntered.ShouldBe(1, "a queued write must never reach a driver that is reconnecting"); + + // The in-flight one still answers with the device's real outcome once it is released. + driver.ReleaseWrites(); + (await inFlight).Success.ShouldBeTrue(); + } + + /// A writable + subscribable driver whose writes either take or hold until released. + private sealed class SlowWritableDriver : IDriver, IWritable, ISubscribable + { + private readonly StubHandle _handle = new(); + private readonly ConcurrentQueue _writes = new(); + private readonly ConcurrentQueue _enteredRefs = new(); + private TaskCompletionSource _release = new(TaskCreationOptions.RunContinuationsAsynchronously); + private int _entered; + private int _inside; + private int _maxInside; + + public TimeSpan Latency { get; init; } + public bool HoldWrites { get; init; } + /// When true a held write ignores its cancellation token — a wedged driver. + public bool IgnoreCancellation { get; init; } + + public IReadOnlyList Writes => _writes.ToArray(); + public IReadOnlyList EnteredRefs => _enteredRefs.ToArray(); + public int WritesEntered => Volatile.Read(ref _entered); + public int MaxConcurrentWrites => Volatile.Read(ref _maxInside); + + public string DriverInstanceId => "slow-writer-1"; + public string DriverType => "SlowWriter"; + + public void ReleaseWrites() => _release.TrySetResult(); + + public Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken) => Task.CompletedTask; + public Task ReinitializeAsync(string driverConfigJson, CancellationToken cancellationToken) => Task.CompletedTask; + public Task ShutdownAsync(CancellationToken cancellationToken) => Task.CompletedTask; + public DriverHealth GetHealth() => new(DriverState.Healthy, DateTime.UtcNow, null); + + public async Task> WriteAsync(IReadOnlyList writes, CancellationToken cancellationToken) + { + Interlocked.Increment(ref _entered); + foreach (var w in writes) _enteredRefs.Enqueue(w.FullReference); + var inside = Interlocked.Increment(ref _inside); + int seen; + while (inside > (seen = Volatile.Read(ref _maxInside))) + Interlocked.CompareExchange(ref _maxInside, inside, seen); + try + { + if (HoldWrites && IgnoreCancellation) await _release.Task; + else if (HoldWrites) await _release.Task.WaitAsync(cancellationToken); + else if (Latency > TimeSpan.Zero) await Task.Delay(Latency, cancellationToken); + foreach (var w in writes) _writes.Enqueue(w); + return writes.Select(_ => new WriteResult(OpcUaStatus.Good)).ToList(); + } + finally + { + Interlocked.Decrement(ref _inside); + } + } + + public event EventHandler? OnDataChange; + + public Task SubscribeAsync( + IReadOnlyList fullReferences, TimeSpan publishingInterval, CancellationToken cancellationToken) => + Task.FromResult(_handle); + + public Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken) => Task.CompletedTask; + + public void FireDataChange(string fullRef, object? value) => + OnDataChange?.Invoke(this, new DataChangeEventArgs( + _handle, fullRef, new DataValueSnapshot(value, OpcUaStatus.Good, DateTime.UtcNow, DateTime.UtcNow))); + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorDisposeTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorDisposeTests.cs new file mode 100644 index 00000000..815575fb --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorDisposeTests.cs @@ -0,0 +1,173 @@ +using System.Diagnostics; +using Akka.Actor; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.OpcUa; +using ZB.MOM.WW.OtOpcUa.Commons.Types; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Runtime.Drivers; +using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers; + +/// +/// Review 02 L1 / P4: a driver child that stops for good must dispose its +/// (the runtime only ever called ShutdownAsync, so a Calculation driver's compiled-script ALCs and an +/// OpcUaClient driver's LoggerFactory + gate + CTS leaked on every respawn), its ShutdownAsync must be +/// bounded (it used to block a dispatcher thread for as long as a dead socket took to answer), and a +/// supervision restart must NOT dispose — the Props closure re-uses the same driver object for the +/// restarted incarnation, which then has to serve through it. +/// +public sealed class DriverInstanceActorDisposeTests : RuntimeActorTestBase +{ + [Fact] + public void Stopping_the_child_shuts_the_driver_down_and_then_disposes_it_exactly_once() + { + var driver = new DisposableRecordingDriver(); + var parent = CreateTestProbe(); + parent.IgnoreMessages(m => m is DriverInstanceActor.ConnectivityChanged); + var actor = parent.ChildActorOf(DriverInstanceActor.Props(driver)); + actor.Tell(new DriverInstanceActor.InitializeRequested("{}")); + AwaitCondition(() => driver.InitializeCount == 1, PresenceBudget); + + Watch(actor); + Sys.Stop(actor); + ExpectTerminated(actor, PresenceBudget); + + driver.ShutdownCount.ShouldBe(1); + driver.DisposeCount.ShouldBe(1, "a child stopped for good must dispose the driver it wrapped"); + driver.Trace.ToArray().ShouldBe(new[] { "init", "shutdown", "dispose" }, "shutdown must land before dispose"); + } + + [Fact] + public async Task A_supervision_restart_keeps_the_driver_undisposed_and_the_restarted_child_still_serves() + { + var driver = new DisposableRecordingDriver(); + var parent = CreateTestProbe(); + // The probe parent's own strategy STOPS a failing child; the production host (no strategy anywhere under + // src/Server) gets Akka's default decider, which RESTARTS it — the route under test. + var actor = parent.ChildActorOf( + DriverInstanceActor.Props(driver), new OneForOneStrategy(_ => Directive.Restart)); + actor.Tell(new DriverInstanceActor.InitializeRequested("{}")); + // Barrier: Subscribe is only handled once Connected (it is dropped in Connecting). + parent.ExpectMsg(m => m.Connected, PresenceBudget); + await actor.Ask( + new DriverInstanceActor.Subscribe(new[] { "tag-a" }, TimeSpan.FromMilliseconds(100)), PresenceBudget); + + // The crash seam (see DriverHostActorRestartTests): a null snapshot NREs on the actor thread, and the + // probe parent's default decider restarts the child in place around the SAME driver object. + driver.FireMalformedDataChange("tag-a"); + parent.FishForMessage(m => m is DriverInstanceActor.ReprovisionRequested, PresenceBudget); + + driver.ShutdownCount.ShouldBe(1, "PreRestart runs PostStop, which shuts the driver down"); + driver.DisposeCount.ShouldBe(0, "a restart re-uses the driver object — disposing it would kill the new incarnation"); + + // Positive control: the restarted incarnation serves through the undisposed driver. The host would + // re-provision it exactly like this (InitializeRequested + the desired set). + actor.Tell(new DriverInstanceActor.InitializeRequested("{}")); + parent.FishForMessage(m => m is DriverInstanceActor.ConnectivityChanged { Connected: true }, PresenceBudget); + await actor.Ask( + new DriverInstanceActor.Subscribe(new[] { "tag-a" }, TimeSpan.FromMilliseconds(100)), PresenceBudget); + driver.FireDataChange("tag-a", 42.0); + var published = (DriverInstanceActor.AttributeValueBatchPublished)parent.FishForMessage( + m => m is DriverInstanceActor.AttributeValueBatchPublished, PresenceBudget); + published.Items.Single().FullReference.ShouldBe("tag-a"); + published.Items.Single().Value.ShouldBe(42.0); + + // ...and the eventual real stop still disposes it — once. + Watch(actor); + Sys.Stop(actor); + ExpectTerminated(actor, PresenceBudget); + driver.DisposeCount.ShouldBe(1); + } + + [Fact] + public void A_hung_ShutdownAsync_is_abandoned_at_the_bound_and_the_driver_is_still_disposed() + { + var never = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + // Ignores its token on purpose: a wedged socket/COM proxy does not honour cancellation either. + var driver = new DisposableRecordingDriver { ShutdownBehavior = _ => never.Task }; + var bound = TimeSpan.FromMilliseconds(300); + var parent = CreateTestProbe(); + parent.IgnoreMessages(m => m is DriverInstanceActor.ConnectivityChanged); + var actor = parent.ChildActorOf(DriverInstanceActor.Props(driver, shutdownTimeout: bound)); + actor.Tell(new DriverInstanceActor.InitializeRequested("{}")); + AwaitCondition(() => driver.InitializeCount == 1, PresenceBudget); + + Watch(actor); + var sw = Stopwatch.StartNew(); + Sys.Stop(actor); + ExpectTerminated(actor, PresenceBudget); + sw.Stop(); + + driver.ShutdownCount.ShouldBe(1, "the shutdown was attempted (positive control for the bound)"); + never.Task.IsCompleted.ShouldBeFalse("the shutdown is still hung — the stop did not wait for it"); + sw.Elapsed.ShouldBeLessThan(TimeSpan.FromSeconds(5), "the stop must not be pinned by the hung shutdown"); + driver.DisposeCount.ShouldBe(1, "a shutdown that timed out must not skip the dispose"); + } + + /// + /// A subscribable driver that records its lifecycle and is . Deliberately + /// NOT also : the actor prefers the async dispose when both exist, and a double + /// implementation would hide a double-dispose. + /// + private sealed class DisposableRecordingDriver : IDriver, ISubscribable, IAsyncDisposable + { + private readonly StubHandle _handle = new(); + private int _initializeCount; + private int _shutdownCount; + private int _disposeCount; + + public Func? ShutdownBehavior { get; init; } + + public List Trace { get; } = new(); + + public int InitializeCount => Volatile.Read(ref _initializeCount); + public int ShutdownCount => Volatile.Read(ref _shutdownCount); + public int DisposeCount => Volatile.Read(ref _disposeCount); + + public string DriverInstanceId => "disposable-1"; + public string DriverType => "Disposable"; + + public Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken) + { + Interlocked.Increment(ref _initializeCount); + lock (Trace) Trace.Add("init"); + return Task.CompletedTask; + } + + public Task ReinitializeAsync(string driverConfigJson, CancellationToken cancellationToken) => Task.CompletedTask; + + public Task ShutdownAsync(CancellationToken cancellationToken) + { + Interlocked.Increment(ref _shutdownCount); + lock (Trace) Trace.Add("shutdown"); + return ShutdownBehavior is { } behavior ? behavior(cancellationToken) : Task.CompletedTask; + } + + public DriverHealth GetHealth() => new(DriverState.Healthy, DateTime.UtcNow, null); + + public ValueTask DisposeAsync() + { + Interlocked.Increment(ref _disposeCount); + lock (Trace) Trace.Add("dispose"); + return ValueTask.CompletedTask; + } + + public event EventHandler? OnDataChange; + + public Task SubscribeAsync( + IReadOnlyList fullReferences, TimeSpan publishingInterval, CancellationToken cancellationToken) => + Task.FromResult(_handle); + + public Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken) => Task.CompletedTask; + + public void FireDataChange(string fullRef, object? value) => + OnDataChange?.Invoke(this, new DataChangeEventArgs( + _handle, fullRef, new DataValueSnapshot(value, OpcUaStatus.Good, DateTime.UtcNow, DateTime.UtcNow))); + + /// The supervision crash seam — see . + public void FireMalformedDataChange(string fullRef) => + OnDataChange?.Invoke(this, new DataChangeEventArgs(_handle, fullRef, null!)); + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorReconnectSingleFlightTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorReconnectSingleFlightTests.cs new file mode 100644 index 00000000..51914d87 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorReconnectSingleFlightTests.cs @@ -0,0 +1,160 @@ +using Akka.Actor; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.Types; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Runtime.Drivers; +using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers; + +/// +/// Review 02 H2: the Reconnecting retry timer fired every reconnectInterval whether or not the +/// previous attempt was still running — _initGeneration only discarded the older attempt's RESULT. With an +/// Initialize bound longer than the cadence (Tier C's 15 s vs. the 10 s timer, or the test harness's unbounded +/// invoker) attempt N+1's ShutdownAsync ran while attempt N's InitializeAsync was still inside the +/// driver: the dispose-before-overwrite race the drivers' rule file forbids, once per overlap, for as long as +/// the device stayed down. The retry is now single-flight. +/// +public sealed class DriverInstanceActorReconnectSingleFlightTests : RuntimeActorTestBase +{ + [Fact] + public void Shutdown_and_Initialize_never_overlap_while_the_init_outlasts_the_retry_cadence() + { + // A 400 ms failing init against a 50 ms retry cadence: without the gate the timer starts a new + // Shutdown+Initialize pair every 50 ms on top of the one still running (eight-deep overlap). + var driver = new InterleavingDriver(initDuration: TimeSpan.FromMilliseconds(400), failUntilAttempt: 4); + var parent = CreateTestProbe(); + var actor = parent.ChildActorOf(DriverInstanceActor.Props( + driver, reconnectInterval: TimeSpan.FromMilliseconds(50), commsLossTimeout: TimeSpan.Zero)); + + actor.Tell(new DriverInstanceActor.InitializeRequested("{}")); + + // Positive control: the gate skips ticks, it does not stop retrying — the 4th attempt succeeds and the + // actor connects. + parent.FishForMessage(m => m is DriverInstanceActor.ConnectivityChanged { Connected: true }, PresenceBudget); + + driver.Attempts.ShouldBeGreaterThanOrEqualTo(4); + driver.Overlaps.ShouldBe(0, + $"ShutdownAsync entered while an InitializeAsync was still running. Interleaving: {driver.Interleaving}"); + } + + [Fact] + public void A_pre_reconnect_teardown_that_hangs_is_abandoned_at_the_bound_and_the_init_still_runs() + { + var never = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var driver = new InterleavingDriver(initDuration: TimeSpan.Zero, failUntilAttempt: 2) + { + // Only the SECOND shutdown (the reconnect teardown) hangs; the first attempt has none. + ShutdownBehavior = _ => never.Task, + }; + var parent = CreateTestProbe(); + var actor = parent.ChildActorOf(DriverInstanceActor.Props( + driver, reconnectInterval: TimeSpan.FromMilliseconds(50), commsLossTimeout: TimeSpan.Zero, + shutdownTimeout: TimeSpan.FromMilliseconds(200))); + + actor.Tell(new DriverInstanceActor.InitializeRequested("{}")); + + parent.FishForMessage(m => m is DriverInstanceActor.ConnectivityChanged { Connected: true }, PresenceBudget); + never.Task.IsCompleted.ShouldBeFalse("the teardown is still hung — the attempt did not wait for it"); + driver.Attempts.ShouldBe(2); + } + + /// + /// Fix round 1: the single-flight gate only cleared on the attempt's RESULT, and the invoker's timeout is + /// token-only — a driver whose InitializeAsync ignores its token wedged the reconnect loop forever. The + /// attempt is now bounded pessimistically (initializeTimeout); an overrun is abandoned, reported + /// as a failure, and the next tick retries. + /// + [Fact] + public void A_token_ignoring_never_completing_init_is_abandoned_and_a_later_attempt_connects() + { + var driver = new InterleavingDriver(initDuration: TimeSpan.Zero, failUntilAttempt: 1) + { + // Attempt 1 hangs forever and ignores cancellation; attempt 2 completes. + HangOnAttempt = 1, + }; + var parent = CreateTestProbe(); + var actor = parent.ChildActorOf(DriverInstanceActor.Props( + driver, reconnectInterval: TimeSpan.FromMilliseconds(50), commsLossTimeout: TimeSpan.Zero, + initializeTimeout: TimeSpan.FromMilliseconds(300))); + + actor.Tell(new DriverInstanceActor.InitializeRequested("{}")); + + parent.FishForMessage(m => m is DriverInstanceActor.ConnectivityChanged { Connected: true }, PresenceBudget); + driver.Attempts.ShouldBe(2); + driver.HungAttemptStillRunning.ShouldBeTrue("positive control: the first attempt really was wedged"); + } + + /// Records every Shutdown/Initialize entry and exit so an overlap is a fact, not an inference. + private sealed class InterleavingDriver : IDriver + { + private readonly TimeSpan _initDuration; + private readonly int _failUntilAttempt; + private readonly Lock _gate = new(); + private readonly List _events = new(); + private int _attempts; + private int _initsInProgress; + private int _overlaps; + + public InterleavingDriver(TimeSpan initDuration, int failUntilAttempt) + { + _initDuration = initDuration; + _failUntilAttempt = failUntilAttempt; + } + + public Func? ShutdownBehavior { get; init; } + + /// The 1-based attempt that never completes and ignores its token (0 = none). + public int HangOnAttempt { get; init; } + + private readonly TaskCompletionSource _never = new(TaskCreationOptions.RunContinuationsAsynchronously); + + public bool HungAttemptStillRunning => !_never.Task.IsCompleted && Volatile.Read(ref _initsInProgress) > 0; + + public int Attempts => Volatile.Read(ref _attempts); + public int Overlaps => Volatile.Read(ref _overlaps); + public string Interleaving { get { lock (_gate) return string.Join(" ", _events); } } + + public string DriverInstanceId => "interleaving-1"; + public string DriverType => "Interleaving"; + + public async Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken) + { + var attempt = Interlocked.Increment(ref _attempts); + Interlocked.Increment(ref _initsInProgress); + Record($"init{attempt}>"); + try + { + if (attempt == HangOnAttempt) await _never.Task; + if (_initDuration > TimeSpan.Zero) await Task.Delay(_initDuration, CancellationToken.None); + if (attempt < _failUntilAttempt) throw new InvalidOperationException($"init attempt {attempt} refused"); + } + finally + { + Record($" Task.CompletedTask; + + public async Task ShutdownAsync(CancellationToken cancellationToken) + { + if (Volatile.Read(ref _initsInProgress) > 0) Interlocked.Increment(ref _overlaps); + Record("shutdown>"); + try + { + if (ShutdownBehavior is { } behavior) await behavior(cancellationToken); + } + finally + { + Record(" new(DriverState.Healthy, DateTime.UtcNow, null); + + private void Record(string e) { lock (_gate) _events.Add(e); } + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Harness/V3ArtifactSeeder.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Harness/V3ArtifactSeeder.cs index e928b1b2..d5edc255 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Harness/V3ArtifactSeeder.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Harness/V3ArtifactSeeder.cs @@ -47,6 +47,8 @@ public static class V3ArtifactSeeder /// is parsed out of TagConfig.alarm by TagConfigIntent), which the alarm-ref half of the /// restart suite needs — an alarm-bearing raw tag is excluded from the VALUE subscription set and carried /// in the alarm one instead, so the two travel through different child code paths. + /// Optional DriverInstanceRowId (the DB row Guid) per driver id; absent + /// ids get , which is what an artifact without the property parses to. /// Extra DriverInstance ids carrying NO devices and NO tags. The /// only way to express a driver that is a running child yet absent from both the value- and alarm-ref /// maps — the _children.Keys arm of ApplyDesiredSubscriptions's union, which must still @@ -58,14 +60,16 @@ public static class V3ArtifactSeeder (string Area, string Line, string Equip, string BackingTag, string? Effective)[] unsRefs, bool enabled = false, IReadOnlyDictionary? tagConfigByTag = null, - string[]? taglessDrivers = null) + string[]? taglessDrivers = null, + IReadOnlyDictionary? driverRowIds = null) { + Guid RowId(string id) => driverRowIds is not null && driverRowIds.TryGetValue(id, out var g) ? g : Guid.Empty; var drivers = rawTags .Select(t => (t.Driver, t.DriverName)) .Distinct() - .Select(d => new { DriverInstanceId = d.Driver, RawFolderId = "rf-plant", Name = d.DriverName, DriverType = "Modbus", DriverConfig = "{}", ClusterId = "c1", Enabled = enabled }) + .Select(d => new { DriverInstanceRowId = RowId(d.Driver), DriverInstanceId = d.Driver, RawFolderId = "rf-plant", Name = d.DriverName, DriverType = "Modbus", DriverConfig = "{}", ClusterId = "c1", Enabled = enabled }) .Concat((taglessDrivers ?? Array.Empty()) - .Select(id => new { DriverInstanceId = id, RawFolderId = "rf-plant", Name = id, DriverType = "Modbus", DriverConfig = "{}", ClusterId = "c1", Enabled = enabled })) + .Select(id => new { DriverInstanceRowId = RowId(id), DriverInstanceId = id, RawFolderId = "rf-plant", Name = id, DriverType = "Modbus", DriverConfig = "{}", ClusterId = "c1", Enabled = enabled })) .ToArray(); var devices = rawTags .Select(t => (t.Driver, t.Device)) @@ -201,7 +205,37 @@ public static class V3ArtifactSeeder return (actor, mux, coordinator); } - private static void Dispatch(IActorRef actor, TestProbe coordinator, DeploymentId deploymentId, RevisionHash rev) + /// The same arrangement as , also returning the coordinator probe so a + /// test can FURTHER deployments to the same host (remove a driver, re-add it). + /// The calling TestKit, supplying the ActorSystem + probes. + /// The factory whose deployment was seeded by . + /// The host's local node id. + /// The deployment to dispatch. + /// The revision hash the deployment was sealed under. + /// Optional driver factory forwarded to DriverHostActor.Props. + /// Optional per-driver supervision-reprovision rate limit. + /// The host actor plus the publish, mux and coordinator probes wired into it. + public static (IActorRef Actor, TestProbe Publish, TestProbe Mux, TestProbe Coordinator) SpawnHostAndApplyWithCoordinator( + TestKitBase kit, IDbContextFactory db, + NodeId localNode, DeploymentId deploymentId, RevisionHash rev, + IDriverFactory? driverFactory = null, + TimeSpan? reprovisionBackoff = null) + { + var publish = kit.CreateCleanOutcomePublishProbe(); + var (actor, mux, coordinator) = Spawn(kit, db, localNode, publish.Ref, driverFactory, reprovisionBackoff); + + Dispatch(actor, coordinator, deploymentId, rev); + publish.ExpectMsg(RuntimeActorTestBase.PresenceBudget); + + return (actor, publish, mux, coordinator); + } + + /// Dispatches to and waits for its Applied ack. + /// The host. + /// The coordinator probe the host acks to. + /// The deployment to dispatch. + /// Its revision hash. + public static void Dispatch(IActorRef actor, TestProbe coordinator, DeploymentId deploymentId, RevisionHash rev) { actor.Tell(new DispatchDeployment(deploymentId, rev, CorrelationId.NewId())); coordinator.ExpectMsg(RuntimeActorTestBase.PresenceBudget) diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Harness/ValuePublishes.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Harness/ValuePublishes.cs new file mode 100644 index 00000000..d0582eff --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Harness/ValuePublishes.cs @@ -0,0 +1,23 @@ +using ZB.MOM.WW.OtOpcUa.Runtime.Drivers; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness; + +/// +/// Builds the ONE message shape a driver child sends for its values — a +/// — for tests that push a single value into the +/// dependency mux by hand. The per-value has no +/// production sender (02/U4); tests should not keep one alive either. +/// +public static class ValuePublishes +{ + /// A one-item batch for . + /// The publishing driver's id. + /// The value's reference (RawPath). + /// The value. + /// The full OPC UA status code. + /// The source timestamp. + /// The batch message. + public static DriverInstanceActor.AttributeValueBatchPublished Single( + string driverInstanceId, string fullReference, object? value, uint statusCode, DateTime timestampUtc) => + new(driverInstanceId, new[] { new DriverInstanceActor.AttributeValueItem(fullReference, value, statusCode, timestampUtc) }); +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Health/DbHealthProbeOffThreadTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Health/DbHealthProbeOffThreadTests.cs new file mode 100644 index 00000000..c10e24f2 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Health/DbHealthProbeOffThreadTests.cs @@ -0,0 +1,73 @@ +using Akka.Actor; +using Microsoft.EntityFrameworkCore; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Configuration; +using ZB.MOM.WW.OtOpcUa.Runtime.Health; +using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Health; + +/// +/// Review 02 P7: ran its SELECT synchronously on the actor thread, so a +/// hanging SQL connect/command held the dispatcher thread for the implicit 15 s / 30 s SqlClient timeouts and +/// queued every later tick (and every GetStatus Ask) behind it. The probe now runs off the mailbox +/// and pipes its outcome back, so the cached answer stays available while a probe hangs. +/// +public sealed class DbHealthProbeOffThreadTests : RuntimeActorTestBase +{ + [Fact] + public async Task GetStatus_answers_from_cache_while_a_probe_is_hung() + { + var factory = new HangingDbFactory(); + try + { + var actor = Sys.ActorOf(DbHealthProbeActor.Props(factory)); + AwaitCondition(() => factory.Entered > 0, PresenceBudget); + + var status = await actor.Ask( + DbHealthProbeActor.GetStatus.Instance, TimeSpan.FromSeconds(3)); + + status.Reachable.ShouldBeFalse(); + status.LastError.ShouldBe("not probed yet"); + factory.Entered.ShouldBe(1, "a second tick must not stack a second probe behind a hung one"); + } + finally + { + factory.Release(); + } + } + + [Fact] + public async Task A_probe_that_completes_updates_the_cached_status() + { + // Positive control for the off-thread path: the piped-back outcome lands in the cache. + var db = NewInMemoryDbFactory(); + var actor = Sys.ActorOf(DbHealthProbeActor.Props(db)); + + await AwaitAssertAsync(async () => + { + var status = await actor.Ask( + DbHealthProbeActor.GetStatus.Instance, TimeSpan.FromSeconds(3)); + status.Reachable.ShouldBeTrue(); + }, PresenceBudget); + } + + /// Blocks every CreateDbContext until released — a connect that never answers. + private sealed class HangingDbFactory : IDbContextFactory + { + private readonly ManualResetEventSlim _release = new(false); + private int _entered; + + public int Entered => Volatile.Read(ref _entered); + + public void Release() => _release.Set(); + + public OtOpcUaConfigDbContext CreateDbContext() + { + Interlocked.Increment(ref _entered); + _release.Wait(); + throw new InvalidOperationException("released without a database"); + } + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Health/HealthProbeActorTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Health/HealthProbeActorTests.cs index cf5863cd..0dacfebf 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Health/HealthProbeActorTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Health/HealthProbeActorTests.cs @@ -15,18 +15,22 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Health; public sealed class HealthProbeActorTests : RuntimeActorTestBase { - /// Verifies that the DB health probe actor returns reachable status against an in-memory database. + /// Verifies that the DB health probe actor returns reachable status against an in-memory database. + /// The probe runs off the mailbox (02/P7), so the first Ask may still see "not probed yet" — poll. [Fact] public async Task DbHealthProbeActor_returns_reachable_against_in_memory_db() { var db = NewInMemoryDbFactory(); var actor = Sys.ActorOf(DbHealthProbeActor.Props(db)); - var status = await actor.Ask( - DbHealthProbeActor.GetStatus.Instance, PresenceBudget); + await AwaitAssertAsync(async () => + { + var status = await actor.Ask( + DbHealthProbeActor.GetStatus.Instance, PresenceBudget); - status.Reachable.ShouldBeTrue(); - status.LastError.ShouldBeNull(); + status.Reachable.ShouldBeTrue(); + status.LastError.ShouldBeNull(); + }, PresenceBudget); } /// Verifies that the peer OPC UA probe actor reports Ok true against a live listener. diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Telemetry/TelemetryLocalHubTombstoneSweepTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Telemetry/TelemetryLocalHubTombstoneSweepTests.cs new file mode 100644 index 00000000..6e0b8c64 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Telemetry/TelemetryLocalHubTombstoneSweepTests.cs @@ -0,0 +1,69 @@ +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Drivers; +using ZB.MOM.WW.OtOpcUa.Runtime.Telemetry; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Telemetry; + +/// +/// Review 02 L4: 's tombstone set only ever shrank on a re-spawn of the +/// SAME id, so a driver removed for good kept its id forever. The tombstone is load-bearing (#609 — it holds +/// back the removed child's late writes), so it cannot simply be cleared; it is now timestamped and swept +/// once older than , which outlives any in-flight +/// capability call by a wide margin. +/// +public sealed class TelemetryLocalHubTombstoneSweepTests +{ + private static DriverHealthChanged Health(string instanceId) => + new("cluster-a", instanceId, "Healthy", null, null, 0, DateTime.UtcNow); + + private static List Drain(ITelemetrySubscription sub) + { + var items = new List(); + while (sub.Reader.TryRead(out var item)) items.Add(item); + return items; + } + + [Fact] + public void A_tombstone_holds_late_writes_until_the_ttl_then_is_swept() + { + var clock = new ManualTime { Now = new DateTimeOffset(2026, 10, 22, 12, 0, 0, TimeSpan.Zero) }; + var hub = new TelemetryLocalHub(clock); + hub.Emit(new TelemetryItem.Health(Health("drv-a"))); + hub.Evict("drv-a"); + + // Inside the TTL: the late write is dropped (the #609 contract) — and another eviction's sweep does + // not lift a tombstone that is still young. + clock.Now += TelemetryLocalHub.EvictionTombstoneTtl / 2; + hub.Evict("drv-b"); + hub.Emit(new TelemetryItem.Health(Health("drv-a"))); + using (var sub = hub.Subscribe(16)) + Drain(sub).ShouldBeEmpty("a late write inside the TTL must still be held back"); + + // Past the TTL: the next eviction sweeps drv-a's tombstone and its writes land again. + clock.Now += TelemetryLocalHub.EvictionTombstoneTtl; + hub.Evict("drv-c"); + hub.Emit(new TelemetryItem.Health(Health("drv-a"))); + using (var sub = hub.Subscribe(16)) + Drain(sub).OfType().Single().E.DriverInstanceId.ShouldBe("drv-a"); + } + + [Fact] + public void Readmit_lifts_a_tombstone_before_its_ttl() + { + var clock = new ManualTime { Now = DateTimeOffset.UtcNow }; + var hub = new TelemetryLocalHub(clock); + hub.Evict("drv-a"); + hub.Readmit("drv-a"); + + hub.Emit(new TelemetryItem.Health(Health("drv-a"))); + using var sub = hub.Subscribe(16); + Drain(sub).Count.ShouldBe(1); + } + + private sealed class ManualTime : TimeProvider + { + public DateTimeOffset Now { get; set; } + public override DateTimeOffset GetUtcNow() => Now; + } +}