dohertj2/lmxopcua
1
0
Fork 0
Code Issues 43 Pull Requests Actions Packages Projects Releases Wiki Activity

Merge lane A: runtime driver lifecycle (#731)

Browse Source 
Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
This commit is contained in:
Joseph Doherty
2026-10-22 10:44:22 -04:00
parent d422eb747b 7ae74ef147
commit d8c39933ed
24 changed files with 2344 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Clients/FleetDiagnosticsClient.cs
+17 -4
View File
@@ -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 <see cref="GetDiagnostics"/>; expects a <see cref="NodeDiagnosticsSnapshot"/> 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 <c>/fleet/diagnostics</c> page.
/// </summary>
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<FleetDiagnosticsClient> _log;
/// <summary>Initializes a new FleetDiagnosticsClient with the given actor system and cluster options.</summary>
/// <param name="system">The Akka actor system.</param>
/// <param name="options">Cluster configuration options.</param>
public FleetDiagnosticsClient(ActorSystem system, IOptions<AkkaClusterOptions> options)
/// <param name="log">Logger for the degraded-to-empty path; defaults to a no-op logger.</param>
public FleetDiagnosticsClient(
ActorSystem system, IOptions<AkkaClusterOptions> options, ILogger<FleetDiagnosticsClient>? log = null)
{
_system = system;
_systemName = options.Value.SystemName;
_log = log ?? NullLogger<FleetDiagnosticsClient>.Instance;
}
/// <inheritdoc />
public async Task<NodeDiagnosticsSnapshot> 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<NodeDiagnosticsSnapshot>(
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,
src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Pages/FleetDiagnostics.razor
+238
View File
@@ -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=<host:port> (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
<PageErrorBoundary>
<div class="d-flex justify-content-between align-items-center mb-3">
<h4 class="mb-0">Node diagnostics</h4>
<a class="btn btn-sm btn-outline-secondary" href="/fleet">Back to fleet status</a>
</div>
<div class="d-flex align-items-center mb-3 gap-2 flex-wrap">
<label class="small text-muted mb-0" for="diag-node">Node</label>
<select id="diag-node" class="form-select form-select-sm" style="width:auto"
value="@_selectedNode" @onchange="OnNodeSelected">
<option value="">— select a driver node —</option>
@foreach (var member in _members)
{
<option value="@member">@member</option>
}
@if (!string.IsNullOrEmpty(_selectedNode) && !_members.Contains(_selectedNode))
{
<option value="@_selectedNode">@_selectedNode (not an Up member)</option>
}
</select>
<input type="text" class="form-control form-control-sm" style="width:220px"
placeholder="or type host:port…" @bind="_typedNode" @bind:event="oninput"
@onkeydown="OnTypedKeyDown" />
<button class="btn btn-sm btn-outline-primary" @onclick="UseTypedNodeAsync"
disabled="@(string.IsNullOrWhiteSpace(_typedNode))">Go</button>
<button class="btn btn-sm btn-outline-primary" @onclick="RefreshAsync"
disabled="@(_loading || string.IsNullOrEmpty(_selectedNode))">
@if (_loading) { <span class="spinner-border spinner-border-sm me-1" /> }
Refresh
</button>
<span class="text-muted small">
Last fetched: @(_fetchedUtc?.ToString("HH:mm:ss 'UTC'") ?? "—")
</span>
</div>
@if (string.IsNullOrEmpty(_selectedNode))
{
<section class="panel notice rise" style="animation-delay:.02s">
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.
</section>
}
else if (_snapshot is null)
{
<p>@(_loading ? "Fetching…" : "No snapshot yet.")</p>
}
else
{
<section class="agg-grid rise" style="animation-delay:.02s">
<div class="agg-card">
<div class="agg-label">Node</div>
<div class="agg-value mono">@_snapshot.NodeId.Value</div>
</div>
<div class="agg-card">
<div class="agg-label">Revision</div>
<div class="agg-value mono" title="@(_snapshot.CurrentRevision?.Value ?? "")">
@(_snapshot.CurrentRevision is { } rev ? DisplayText.Abbreviate(rev.Value, 12) : "—")
</div>
</div>
<div class="agg-card @(_snapshot.RunningFromCache ? "caution" : "")">
<div class="agg-label">Config source</div>
<div class="agg-value">@(_snapshot.RunningFromCache ? "cache (frozen)" : "live")</div>
</div>
<div class="agg-card">
<div class="agg-label">Drivers</div>
<div class="agg-value numeric">@_snapshot.Drivers.Count</div>
</div>
<div class="agg-card alert">
<div class="agg-label">Faulted devices</div>
<div class="agg-value numeric">@_snapshot.Drivers.Sum(d => d.FaultedDevices)</div>
</div>
</section>
@if (_snapshot.RunningFromCache)
{
<section class="panel notice rise" style="animation-delay:.05s">
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.
</section>
}
<section class="panel rise" style="animation-delay:.08s">
<div class="panel-head">
Drivers
<span class="text-muted small ms-2">as of @_snapshot.AsOfUtc.ToString("u")</span>
</div>
@if (_snapshot.Drivers.Count == 0)
{
<div class="p-3 text-muted">
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).
</div>
}
else
{
<div class="table-wrap">
<table class="data-table">
<thead>
<tr>
<th>Driver instance</th>
<th>Row id</th>
<th>State</th>
<th class="numeric">Connected devices</th>
<th class="numeric">Faulted devices</th>
<th title="When the driver child last published a changed health snapshot (its spawn time until the first)">Last health publish</th>
</tr>
</thead>
<tbody>
@foreach (var d in _snapshot.Drivers.OrderBy(d => d.Name, StringComparer.OrdinalIgnoreCase))
{
<tr>
<td>@d.Name</td>
<td><span class="mono" title="@d.DriverInstanceId">@(d.DriverInstanceId == Guid.Empty ? "—" : DisplayText.Abbreviate(d.DriverInstanceId.ToString(), 13))</span></td>
<td><span class="chip @StateChipClass(d.State)">@d.State</span></td>
<td class="numeric">@d.ConnectedDevices</td>
<td class="numeric">@d.FaultedDevices</td>
<td>@d.LastChangeUtc.ToString("u")</td>
</tr>
}
</tbody>
</table>
</div>
}
</section>
}
</PageErrorBoundary>
@code {
[SupplyParameterFromQuery(Name = "node")]
public string? Node { get; set; }
private IReadOnlyList<string> _members = Array.Empty<string>();
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();
}
}
src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Telemetry/TelemetryDialSupervisor.cs
+43
View File
@@ -129,6 +129,7 @@ public sealed class TelemetryDialSupervisor : ReceiveActor, IWithTimers
Receive<TelemetryReceived>(HandleTelemetryReceived);
Receive<StreamStopped>(HandleStreamStopped);
Receive<Reconnect>(HandleReconnect);
Receive<DialDeadline>(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
}
}
/// <summary>
/// <c>TelemetryDial:CallTimeoutSeconds</c>: a dial that has delivered no event by the deadline is cut and
/// treated as a transport failure (reconnect with a failure streak). <c>Connected</c> 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.
/// </summary>
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}";
/// <summary>Per-node dialer bookkeeping. Mutated only on the actor thread.</summary>
private sealed class NodeDialer(TelemetryDialTarget target)
{
@@ -496,4 +533,10 @@ public sealed class TelemetryDialSupervisor : ReceiveActor, IWithTimers
/// <summary>Backoff timer tick: start a fresh dialer for the node.</summary>
/// <param name="NodeId">The node id to reconnect.</param>
public sealed record Reconnect(string NodeId);
/// <summary>Dial-establishment deadline tick (<c>TelemetryDial:CallTimeoutSeconds</c>); see
/// <see cref="HandleDialDeadline"/>.</summary>
/// <param name="NodeId">The node whose dial is being checked.</param>
/// <param name="Generation">The dialer generation the deadline was armed for.</param>
public sealed record DialDeadline(string NodeId, int Generation);
}
src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/ChildHealthLedger.cs
+73
View File
@@ -0,0 +1,73 @@
using System.Collections.Concurrent;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
namespace ZB.MOM.WW.OtOpcUa.Runtime.Drivers;
/// <summary>The last health snapshot a driver child published, as kept by <see cref="ChildHealthLedger"/>.</summary>
/// <param name="State">The driver's reported state.</param>
/// <param name="LastSuccessfulReadUtc">When the driver last read successfully, if ever.</param>
/// <param name="LastError">The driver's last error, if any.</param>
/// <param name="HostStatuses">Per-host connectivity for a multi-host driver, or null when the driver has none.</param>
/// <param name="PublishedUtc">When the snapshot was published — the child's dedup means this is the last CHANGE.</param>
public sealed record ChildHealthRecord(
DriverState State,
DateTime? LastSuccessfulReadUtc,
string? LastError,
IReadOnlyList<HostConnectivityStatus>? HostStatuses,
DateTime PublishedUtc);
/// <summary>
/// An <see cref="IDriverHealthPublisher"/> decorator that remembers the last snapshot per driver instance
/// and forwards every publish to the real publisher. <see cref="DriverHostActor"/> wraps the publisher it
/// hands its children in one of these so <c>GetDiagnostics</c> 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.
/// </summary>
/// <remarks>
/// Deliberately NOT a message to the host: the children already publish through this seam on every
/// change, so a parallel <c>Tell</c> 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.
/// </remarks>
public sealed class ChildHealthLedger : IDriverHealthPublisher
{
private readonly IDriverHealthPublisher _inner;
private readonly ConcurrentDictionary<string, ChildHealthRecord> _last = new(StringComparer.Ordinal);
/// <summary>Wraps <paramref name="inner"/>.</summary>
/// <param name="inner">The publisher every snapshot is forwarded to.</param>
public ChildHealthLedger(IDriverHealthPublisher inner)
{
ArgumentNullException.ThrowIfNull(inner);
_inner = inner;
}
/// <inheritdoc />
public void Publish(
string clusterId,
string driverInstanceId,
DriverHealth health,
int errorCount5Min,
DateTime? rediscoveryNeededUtc = null,
string? rediscoveryReason = null,
IReadOnlyList<HostConnectivityStatus>? hostStatuses = null,
SkippedTagReport? skippedTags = null,
IReadOnlyDictionary<string, string>? 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);
}
/// <summary>The last snapshot published for <paramref name="driverInstanceId"/>, or null if none yet.</summary>
/// <param name="driverInstanceId">The driver instance.</param>
/// <returns>The record, or null.</returns>
public ChildHealthRecord? TryGet(string driverInstanceId) =>
_last.TryGetValue(driverInstanceId, out var record) ? record : null;
/// <summary>Forgets <paramref name="driverInstanceId"/>'s snapshot (the child was stopped).</summary>
/// <param name="driverInstanceId">The driver instance.</param>
public void Forget(string driverInstanceId) => _last.TryRemove(driverInstanceId, out _);
}
src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs
+73 -9
View File
@@ -195,7 +195,9 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
private readonly IReadOnlySet<string> _localRoles;
private readonly IActorRef? _dependencyMux;
private readonly IActorRef? _opcUaPublishActor;
private readonly IDriverHealthPublisher _healthPublisher;
/// <summary>The publisher handed to every child — a <see cref="ChildHealthLedger"/> over the configured one, so
/// <see cref="HandleGetDiagnostics"/> answers from the children's own last-published health.</summary>
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)
{
/// <summary>When this child was spawned — the diagnostics' last-change time until it publishes health.</summary>
public DateTime SpawnedUtc { get; } = DateTime.UtcNow;
// Convenience accessors for sites that don't need the full spec.
/// <summary>Gets the driver type name from the child's spec.</summary>
public string DriverType => Spec.DriverType;
@@ -886,7 +891,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
_localRoles = localRoles ?? new HashSet<string>(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<DriverInstanceActor.ConnectivityChanged>(_ => { });
// 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<DriverInstanceActor.AttributeValueBatchPublished>(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<DriverInstanceActor.AttributeAlarmPublished>(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<ArtifactForApply>(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<ApplyWedged>(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<CommsRecoveryAudit>(_ => { });
@@ -2496,16 +2519,16 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
Timers.StartPeriodicTimer("retry-db", RetryConfigDbConnection.Instance, ReconnectInterval);
}
/// <summary>
/// Answers <see cref="GetDiagnostics"/> from the live child set and each child's last-published health
/// (<see cref="ChildHealthLedger"/>). Used to be a placeholder — <c>Guid.Empty</c>, "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.
/// </summary>
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);
}
/// <summary>
/// One child's diagnostics row. <c>Name</c> carries the string driver-instance id — the identity every
/// other surface (/hosts, telemetry, the artifact) keys on — and <c>DriverInstanceId</c> 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 <c>IHostConnectivityProbe</c> (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.
/// </summary>
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);
src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs
+445 -21
View File
@@ -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
/// </remarks>
public static readonly TimeSpan DefaultCommsLossTimeout = TimeSpan.FromSeconds(90);
/// <summary>
/// Bound on the driver's <see cref="IDriver.ShutdownAsync"/> (and its dispose) when the child stops or
/// restarts, and on the teardown that precedes every reconnect attempt. <c>PostStop</c> 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 <c>CoordinatedShutdown</c> and the host's own restart cascade. A shutdown that overruns is
/// abandoned (logged), never awaited further.
/// </summary>
public static readonly TimeSpan DefaultShutdownTimeout = TimeSpan.FromSeconds(5);
/// <summary>
/// Default cap on the writes this actor runs against its driver at once. Writes run OFF the mailbox
/// (a <c>ReceiveAsync</c> 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.
/// </summary>
public const int DefaultMaxWritesInFlight = 4;
/// <summary>
/// How long a queued write may wait for a slot before it is failed <c>BadTimeout</c> 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.
/// </summary>
public static readonly TimeSpan DefaultQueuedWriteBudget = TimeSpan.FromSeconds(8);
/// <summary>
/// Pessimistic bound on one <see cref="IDriver.InitializeAsync"/> 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).
/// </summary>
public static readonly TimeSpan DefaultInitializeTimeout = TimeSpan.FromSeconds(30);
/// <summary>
/// Pessimistic bound on one write round-trip, on top of the 5 s token-based write cap. A driver whose
/// <see cref="IWritable.WriteAsync"/> 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.
/// </summary>
public static readonly TimeSpan DefaultWriteTimeout = TimeSpan.FromSeconds(10);
/// <summary>Self-sent by a write task once it has replied, releasing its slot and its tag.</summary>
/// <param name="TagId">The tag the write held.</param>
/// <param name="Incarnation">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.</param>
private sealed record WriteCompleted(string TagId, int Incarnation);
private readonly IDriver _driver;
/// <summary>Phase 6.1 resilience seam: every guarded driver-capability call this actor makes is
@@ -204,6 +251,47 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
/// <summary>Comms-loss watchdog threshold; <see cref="TimeSpan.Zero"/> or negative disables it.</summary>
private readonly TimeSpan _commsLossTimeout;
/// <summary>Bound on the driver shutdown / dispose; see <see cref="DefaultShutdownTimeout"/>.</summary>
private readonly TimeSpan _shutdownTimeout;
/// <summary>Cap on concurrent writes; see <see cref="DefaultMaxWritesInFlight"/>.</summary>
private readonly int _maxWritesInFlight;
/// <summary>Writes currently inside the driver. Actor-thread only (incremented on launch, decremented on
/// the self-sent <see cref="WriteCompleted"/>).</summary>
private int _writesInFlight;
/// <summary>Writes waiting for a slot (or for their tag), in arrival order, stamped with arrival time.
/// Failed fast on every exit from <c>Connected</c>, so a queued write never reaches a driver that is
/// reconnecting; failed <c>BadTimeout</c> at dequeue once older than <see cref="_queuedWriteBudget"/>.</summary>
private readonly LinkedList<(WriteAttribute Write, IActorRef ReplyTo, DateTime EnqueuedUtc)> _pendingWrites = new();
/// <summary>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.</summary>
private readonly HashSet<string> _tagsInFlight = new(StringComparer.Ordinal);
/// <summary>See <see cref="DefaultQueuedWriteBudget"/>.</summary>
private readonly TimeSpan _queuedWriteBudget;
/// <summary>See <see cref="DefaultInitializeTimeout"/>.</summary>
private readonly TimeSpan _initializeTimeout;
/// <summary>See <see cref="DefaultWriteTimeout"/>.</summary>
private readonly TimeSpan _writeTimeout;
/// <summary>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).</summary>
private readonly int _incarnation;
/// <summary>
/// Set by <see cref="PreRestart"/> so the <see cref="PostStop"/> it triggers skips the dispose. A
/// supervision restart recreates the actor object around the SAME <see cref="IDriver"/> (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.
/// </summary>
private bool _restarting;
/// <summary>When the actor last entered <c>Connected</c>. 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.</summary>
@@ -218,6 +306,26 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
/// on the actor thread, so no lock is needed.</summary>
private int _initGeneration;
/// <summary>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.</summary>
private static int s_initGenerationSource;
/// <summary>
/// True while an <see cref="InitializeAsync"/> attempt is running (set when it is launched, cleared by its
/// piped <see cref="InitializeSucceeded"/>/<see cref="InitializeFailed"/>). The <c>Reconnecting</c> 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.
/// </summary>
private bool _initInFlight;
/// <summary>Cancellation for the in-flight attempt, so <see cref="PostStop"/> can cancel it before the
/// shutdown rather than let it race the teardown on a restart.</summary>
private CancellationTokenSource? _initCts;
/// <summary>
/// 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
/// <param name="commsLossTimeout">Optional comms-loss watchdog threshold; defaults to
/// <see cref="DefaultCommsLossTimeout"/>. <see cref="TimeSpan.Zero"/> disables the watchdog — used by
/// tests that need a Connected driver to stay Connected while reporting non-Healthy.</param>
/// <param name="shutdownTimeout">Optional bound on the driver's shutdown / dispose on stop and on the
/// pre-reconnect teardown; defaults to <see cref="DefaultShutdownTimeout"/>. Exists so a test can prove a
/// hung shutdown is abandoned without waiting 5 s.</param>
/// <param name="maxWritesInFlight">Optional cap on concurrent writes against the driver; defaults to
/// <see cref="DefaultMaxWritesInFlight"/>. Writes beyond it queue in arrival order.</param>
/// <param name="queuedWriteBudget">Optional bound on how long a queued write may wait before it is failed
/// <c>BadTimeout</c>; defaults to <see cref="DefaultQueuedWriteBudget"/>.</param>
/// <param name="initializeTimeout">Optional pessimistic bound on one initialise attempt; defaults to
/// <see cref="DefaultInitializeTimeout"/>.</param>
/// <param name="writeTimeout">Optional pessimistic bound on one write round-trip; defaults to
/// <see cref="DefaultWriteTimeout"/>.</param>
/// <returns>A <see cref="Akka.Actor.Props"/> instance configured to create the wrapped <see cref="DriverInstanceActor"/>.</returns>
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));
/// <summary>
/// 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 <see cref="HealthPollInterval"/> when null.</param>
/// <param name="commsLossTimeout">Comms-loss watchdog threshold; defaults to
/// <see cref="DefaultCommsLossTimeout"/> when null, and is disabled when zero or negative.</param>
/// <param name="shutdownTimeout">Bound on the driver's shutdown / dispose; defaults to
/// <see cref="DefaultShutdownTimeout"/> when null.</param>
/// <param name="maxWritesInFlight">Cap on concurrent writes; defaults to
/// <see cref="DefaultMaxWritesInFlight"/> when null or non-positive.</param>
/// <param name="queuedWriteBudget">Bound on a queued write's wait; defaults to
/// <see cref="DefaultQueuedWriteBudget"/> when null or non-positive.</param>
/// <param name="initializeTimeout">Pessimistic bound on one initialise attempt; defaults to
/// <see cref="DefaultInitializeTimeout"/> when null or non-positive.</param>
/// <param name="writeTimeout">Pessimistic bound on one write round-trip; defaults to
/// <see cref="DefaultWriteTimeout"/> when null or non-positive.</param>
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<string, object?>("event", startStubbed ? "spawn_stub" : "spawn"),
new KeyValuePair<string, object?>("driver_type", driver.DriverType));
@@ -421,6 +571,18 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
Context.Parent.Tell(new ReprovisionRequested(_driverInstanceId, reason.GetType().Name));
}
/// <summary>
/// Flags the restart before the base implementation runs <see cref="PostStop"/>, so that PostStop
/// shuts the driver down (the host re-initialises it on <see cref="ReprovisionRequested"/>) but does
/// NOT dispose it — see <see cref="_restarting"/>.
/// </summary>
/// <inheritdoc />
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<WriteAttribute>(_ =>
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<WriteCompleted>(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<RouteAlarmAck>(_ =>
@@ -455,6 +619,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
Receive<InitializeSucceeded>(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<InitializeFailed>(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<WriteAttribute>(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<WriteAttribute>(msg => StartOrQueueWrite(msg, Sender));
Receive<WriteCompleted>(OnWriteCompleted);
ReceiveAsync<RouteAlarmAck>(HandleAcknowledgeAsync);
ReceiveAsync<Subscribe>(HandleSubscribeAsync);
ReceiveAsync<Unsubscribe>(_ => UnsubscribeAsync());
@@ -745,6 +915,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
// timeout on an inbound write to a transiently-down driver). Synchronous Receive.
Receive<WriteAttribute>(_ =>
Sender.Tell(new WriteAttributeResult(false, "driver not connected")));
Receive<WriteCompleted>(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<RouteAlarmAck>(_ =>
@@ -752,6 +923,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
Receive<InitializeSucceeded>(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<InitializeFailed>(_ => { /* 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<InitializeFailed>(msg =>
{
if (msg.Generation == _initGeneration) CompleteInitAttempt();
});
Receive<SetDesiredSubscriptions>(StoreDesiredSubscriptions);
Receive<ForceReconnect>(_ => { /* already reconnecting — no-op */ });
Receive<ConnectionLossRaised>(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)
/// <summary>Releases the single-flight gate once the in-flight attempt's result has been honoured.</summary>
private void CompleteInitAttempt()
{
_initInFlight = false;
_initCts = null;
}
/// <summary>
/// Launches a write against the driver off the mailbox, or queues it when
/// <see cref="_maxWritesInFlight"/> 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
/// (<see cref="RunWriteAsync"/>) replies to the captured sender from the thread pool and then
/// self-sends <see cref="WriteCompleted"/> to free its slot.
/// </summary>
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<WriteResult> 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));
}
}
/// <summary>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.</summary>
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;
}
}
/// <summary>Replies to every queued write with <paramref name="reason"/>; called on each exit from
/// <c>Connected</c>. In-flight writes are left to finish — they already hold the device and reply with its
/// real outcome.</summary>
private void FailQueuedWrites(string reason)
{
while (_pendingWrites.First is { } node)
{
_pendingWrites.RemoveFirst();
node.Value.ReplyTo.Tell(new WriteAttributeResult(false, reason));
}
}
/// <summary>
@@ -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<string, object?>("event", "stop"),
new KeyValuePair<string, object?>("driver_type", _driver.DriverType));
}
/// <summary>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.</summary>
private static readonly TimeSpan MinDisposeBudget = TimeSpan.FromMilliseconds(250);
/// <summary>Final teardown of the wrapped driver: <see cref="IAsyncDisposable"/> preferred, else
/// <see cref="IDisposable"/>, 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
/// <see cref="IDriver.ShutdownAsync"/>.</summary>
/// <param name="bound">The remaining PostStop budget available to an async dispose.</param>
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<string, object?>("event", "dispose"),
new KeyValuePair<string, object?>("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);
}
}
}
src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Health/DbHealthProbeActor.cs
+50 -8
View File
@@ -7,21 +7,37 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.Health;
/// <summary>
/// Single-flight cached health probe against the ConfigDb. Reads cached state via
/// <c>Ask&lt;DbHealthStatus&gt;</c>; a single SELECT 1 runs at most every <c>RefreshInterval</c>.
/// <c>Ask&lt;DbHealthStatus&gt;</c>; a single SELECT runs at most every <c>RefreshInterval</c>.
/// Consumed by both the host's <c>/health/ready</c> endpoint and
/// <c>RedundancyStateActor</c>'s stale calc.
/// </summary>
/// <remarks>
/// 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 <see cref="GetStatus"/> 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.
/// </remarks>
public sealed class DbHealthProbeActor : ReceiveActor, IWithTimers
{
public static readonly TimeSpan RefreshInterval = TimeSpan.FromSeconds(5);
/// <summary>Bound on the probe's SQL command, relational providers only (the in-memory provider the
/// actor tests run against has no command timeout).</summary>
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() { } }
/// <summary>Piped back by the off-thread probe with its outcome.</summary>
private sealed record ProbeCompleted(DbHealthStatus Status);
private readonly IDbContextFactory<OtOpcUaConfigDbContext> _dbFactory;
private readonly ILoggingAdapter _log = Context.GetLogger();
private DbHealthStatus _last = new(false, DateTime.MinValue, "not probed yet");
private bool _probeInFlight;
/// <summary>Gets or sets the timer scheduler for periodic health probes.</summary>
public ITimerScheduler Timers { get; set; } = null!;
@@ -38,28 +54,54 @@ public sealed class DbHealthProbeActor : ReceiveActor, IWithTimers
{
_dbFactory = dbFactory;
Receive<GetStatus>(_ => Sender.Tell(_last));
Receive<Tick>(_ => RunProbe());
Receive<Tick>(_ => StartProbe());
Receive<ProbeCompleted>(msg =>
{
_probeInFlight = false;
_last = msg.Status;
if (!msg.Status.Reachable)
_log.Warning("DbHealthProbe: probe failed: {Error}", msg.Status.LastError);
});
}
/// <inheritdoc />
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<OtOpcUaConfigDbContext> 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);
}
}
}
src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Telemetry/TelemetryLocalHub.cs
+47 -6
View File
@@ -27,7 +27,27 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.Telemetry;
/// </remarks>
public sealed class TelemetryLocalHub : ITelemetryLocalHub
{
/// <summary>
/// How long an eviction tombstone (see <see cref="_evicted"/>) is kept. It only has to outlive the
/// removed child's last late write — an in-flight capability call that completes after
/// <c>StopChild</c>, 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.
/// </summary>
public static readonly TimeSpan EvictionTombstoneTtl = TimeSpan.FromMinutes(10);
private readonly object _gate = new();
private readonly TimeProvider _time;
/// <summary>Initializes a hub on the system clock.</summary>
public TelemetryLocalHub() : this(TimeProvider.System) { }
/// <summary>Initializes a hub on <paramref name="time"/> — the tombstone-sweep test seam.</summary>
/// <param name="time">The clock the eviction tombstones are stamped and swept by.</param>
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<string, TelemetryItem.Health> _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<string> _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<string, DateTimeOffset> _evicted = new(StringComparer.Ordinal);
/// <inheritdoc />
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);
}
/// <summary>Drops tombstones older than <see cref="EvictionTombstoneTtl"/>. Called under <see cref="_gate"/>
/// from <see cref="Evict"/> only — the set is bounded by distinct removed ids, so a linear pass per
/// eviction is cheap, and Emit never pays for it.</summary>
private void SweepExpiredTombstones(DateTimeOffset now)
{
if (_evicted.Count == 0) return;
List<string>? expired = null;
foreach (var (id, evictedAt) in _evicted)
{
if (now - evictedAt >= EvictionTombstoneTtl)
(expired ??= new List<string>()).Add(id);
}
if (expired is null) return;
foreach (var id in expired)
_evicted.Remove(id);
}
/// <inheritdoc />
public ITelemetrySubscription Subscribe(int boundedCapacity)
{