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

Merge master into remed/lane-F

Browse Source 
Refs #735

Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
This commit is contained in:
Joseph Doherty
2026-10-22 13:29:15 -04:00
parent 77ca10aa37 d8c39933ed
commit 8a0acc4649
51 changed files with 3658 additions and 128 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.AdminUI/ZB.MOM.WW.OtOpcUa.AdminUI.csproj
+6
View File
@@ -9,6 +9,12 @@
<ItemGroup>
<FrameworkReference Include="Microsoft.AspNetCore.App"/>
<PackageReference Include="Microsoft.AspNetCore.SignalR.Client"/>
<!-- Surgical transitive pin — see the Microsoft.AspNetCore.SignalR.Common block in
Directory.Packages.props (GHSA-334f-vvp4-7g6p on SignalR.Common 10.0.0). Inside AdminUI
the shared framework already supplies it (hence NU1510), but the PackageReference flows
to the non-Web consumers (Host.Tests, AdminUI.Tests, OpcUaServer.IntegrationTests) where
SignalR.Client 10.0.0 would otherwise resolve the vulnerable 10.0.0. -->
<PackageReference Include="Microsoft.AspNetCore.SignalR.Common" NoWarn="NU1510"/>
<PackageReference Include="ZB.MOM.WW.Theme"/>
<PackageReference Include="ZB.MOM.WW.Secrets.Ui"/>
</ItemGroup>
src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/HistorianBackendRegistry.cs
+2 -2
View File
@@ -45,8 +45,8 @@ public static class HistorianBackendRegistry
new Dictionary<string, Func<ServerHistorianOptions, IConfiguration, IHistorianBackendProvider>>(StringComparer.OrdinalIgnoreCase)
{
[ServerHistorianOptions.GatewayBackendName] = (o, _) => new GatewayHistorianProvider(o),
[SqlServerBackendName] = (o, c) => new SqlServerHistorianProvider(BindSqlServer(c), o.MaxTimeWeightedRawRows),
[QuestDbBackendName] = (o, c) => new QuestDbHistorianProvider(BindQuestDb(c), o.MaxTimeWeightedRawRows),
[SqlServerBackendName] = (o, c) => new SqlServerHistorianProvider(BindSqlServer(c), o.MaxTimeWeightedRawRows, o.EffectiveMaxRawValuesPerRead),
[QuestDbBackendName] = (o, c) => new QuestDbHistorianProvider(BindQuestDb(c), o.MaxTimeWeightedRawRows, o.EffectiveMaxRawValuesPerRead),
};
/// <summary>The registered backend names, for validation messages.</summary>
src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/HistoryPaging.cs
+52 -12
View File
@@ -1,3 +1,5 @@
using Opc.Ua;
using Opc.Ua.Server;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
namespace ZB.MOM.WW.OtOpcUa.OpcUaServer;
@@ -110,24 +112,62 @@ internal enum HistoryCursorKind
/// <c>NumValuesPerNode</c> samples with a null continuation point — so paging is synthesised here,
/// time-based:
/// <list type="bullet">
/// <item>A page that returns EXACTLY the requested cap (<c>NumValuesPerNode &gt; 0</c>) MAY have
/// more behind it ⇒ emit a continuation point.</item>
/// <item>A short page (fewer than the cap) is the last page ⇒ no continuation point.</item>
/// <item><c>NumValuesPerNode == 0</c> means "all values, no limit" (OPC UA Part 11) ⇒ never page;
/// return everything in one shot.</item>
/// <item>A page whose read returned the lookahead row past it MAY have more behind it ⇒ emit a continuation
/// point (<see cref="IsFullPage"/> judges a read against what was asked for).</item>
/// <item>A short read (no lookahead row) is the last page ⇒ no continuation point.</item>
/// <item><c>NumValuesPerNode == 0</c> means "all values" (OPC UA Part 11) — but not "in one shot": the page is
/// sized to the backend's <see cref="IHistorianDataSource.MaxRawValuesPerRead"/> by <see cref="RawPageLimit"/>
/// (the server raw cap <c>ServerHistorian:MaxRawValuesPerRead</c>, default 10 000, and/or a strict gateway's
/// learned ceiling; review R08:1.1, #736), and a full page carries a continuation point like any other. Only
/// with no limit at all (<c>0</c>, the opt-out) is everything returned in one shot.</item>
/// </list>
/// All methods are static + pure so they unit-test without a server, a session, or the SDK.
/// All paging methods are static + pure so they unit-test without a server, a session, or the SDK;
/// <see cref="AdvertiseMaxReturnDataValues"/> is the one SDK-facing member.
/// </summary>
internal static class HistoryPaging
{
/// <summary>
/// Decide whether a just-returned page is "full" and therefore MAY be followed by more data —
/// the signal to emit a continuation point. A page is full when the client asked for a finite
/// cap (<paramref name="numValuesPerNode"/> &gt; 0) and the backend returned exactly that many
/// samples. A short page (or an unlimited <c>0</c> request) is terminal.
/// Advertise the server raw cap as <c>HistoryServerCapabilities/MaxReturnDataValues</c> (i=11273; Part 11 §5.4.2:
/// "the maximum number of values that can be returned by the Server for each Node", 0 = no limit). The cap is
/// the most values one HistoryRead-Raw page returns (<c>ServerHistorian:MaxRawValuesPerRead</c>), so a client can
/// size its <c>NumValuesPerNode</c> and expect a continuation point past it. Idempotent; a re-sync with a
/// different cap overwrites. Never throws: a missing diagnostics manager or node, or any SDK failure writing the
/// property, leaves the SDK default (0) and returns <c>false</c> for the caller to log — HistoryRead itself still
/// enforces the cap.
/// </summary>
/// <param name="returnedCount">The number of samples the backend returned for this page.</param>
/// <param name="numValuesPerNode">The client's per-page cap; <c>0</c> means "no limit".</param>
/// <param name="server">The running server.</param>
/// <param name="maxRawValuesPerRead">The cap (&lt;= 0 advertises 0 = no limit).</param>
/// <returns><c>true</c> when the property was written.</returns>
public static bool AdvertiseMaxReturnDataValues(IServerInternal? server, int maxRawValuesPerRead)
{
if (server?.DiagnosticsNodeManager is not DiagnosticsNodeManager diagnostics) return false;
try
{
lock (diagnostics.Lock)
{
var capabilities = diagnostics.GetDefaultHistoryCapabilities();
if (capabilities?.MaxReturnDataValues is not { } property) return false;
property.Value = (uint)Math.Max(0, maxRawValuesPerRead);
property.ClearChangeMasks(diagnostics.SystemContext, false);
return true;
}
}
catch (Exception)
{
return false; // advisory only: the caller logs; the cap is enforced by HistoryRead regardless
}
}
/// <summary>
/// Decide whether a just-returned read is "full" and therefore MAY be followed by more data —
/// the signal to emit a continuation point. A read is full when it was issued with a finite request
/// cap (<paramref name="numValuesPerNode"/> &gt; 0 — the <c>RawPagePlan.RequestCap</c>: page + boundary ties +
/// lookahead, which under a backend limit or the server raw cap is always finite, uncapped client reads included)
/// and the backend returned at least that many samples. A short read is terminal; so is a read issued with a
/// request cap of <c>0</c>, which only happens with no limit at all (one shot of the whole window).
/// </summary>
/// <param name="returnedCount">The number of samples the backend returned for this read.</param>
/// <param name="numValuesPerNode">The request cap the read was issued with; <c>0</c> means "no limit".</param>
/// <returns><c>true</c> when a continuation point should be emitted; otherwise <c>false</c>.</returns>
public static bool IsFullPage(int returnedCount, uint numValuesPerNode) =>
numValuesPerNode > 0 && returnedCount >= numValuesPerNode;
src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/RawPageLimit.cs
+11 -6
View File
@@ -11,17 +11,22 @@ internal readonly record struct RawPagePlan(int Limit, uint PageCap, uint Reques
/// <summary>
/// sizes a HistoryRead-Raw page to what the backend can return in one read
/// (<see cref="IHistorianDataSource.MaxRawValuesPerRead"/> = L — a strict HistorianGateway's learned row ceiling, less
/// its boundary-row allowance). With a limit the read — page P + boundary ties k + the lookahead row — is at most L,
/// so P = min(want, L − k − 1) (all values: P = L − k − 1); a full page carries a continuation point like any other, as
/// Part 11 allows fewer values than <c>NumValuesPerNode</c> with one. Previously a cap above a strict gateway's ceiling
/// (or an uncapped read of a window holding more) failed <c>BadTooManyOperations</c>.
/// (<see cref="IHistorianDataSource.MaxRawValuesPerRead"/> = L — the server raw cap
/// <c>ServerHistorian:MaxRawValuesPerRead</c>, default 10 000, which every backend reports (review R08:1.1, #736),
/// and/or a strict HistorianGateway's learned row ceiling less its boundary-row allowance, whichever is lower). With
/// a limit the read — page P + boundary ties k + the lookahead row — is at most L, so P = min(want, L − k − 1) (all
/// values: P = L − k − 1, i.e. a first page of 9 999 at the default cap); a full page carries a continuation point like
/// any other, as Part 11 allows fewer values than <c>NumValuesPerNode</c> with one. Previously a cap above a strict
/// gateway's ceiling (or an uncapped read of a window holding more) failed <c>BadTooManyOperations</c>, and before
/// #736 an uncapped read on a backend without a ceiling read the whole window in one shot.
/// </summary>
/// <remarks>
/// <para>
/// <b>Never a silent truncation.</b> The backend still answers a read in full or refuses it; only the server's page is
/// smaller, and a page is judged full against what was asked for. A read the backend cannot serve at all (a tie
/// cluster larger than one read) stays <c>BadTooManyOperations</c>.
/// cluster larger than one read) stays <c>BadTooManyOperations</c>. The backends enforce the same cap themselves
/// (a read asking for more than L, or for everything, is refused when the window holds more than L rows), so this
/// planner is what keeps the server's own reads inside it: with L &gt; 0 no read it plans ever asks for 0 or more than L.
/// </para>
/// <para>
/// <b>One re-read, planned again.</b> A refused read (<see cref="HistoryReadLimitExceededException"/>) is re-planned and
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/Historian/ServerHistorianOptions.cs
+40
View File
@@ -154,6 +154,42 @@ public sealed class ServerHistorianOptions
/// </summary>
public int MaxTimeWeightedRawRows { get; init; } = 1_000_000;
/// <summary>The default of <see cref="MaxRawValuesPerRead"/>: 10 000 values per raw page / backend read.</summary>
public const int DefaultMaxRawValuesPerRead = 10_000;
/// <summary>
/// The smallest usable <see cref="MaxRawValuesPerRead"/>: a resumed page's read must carry at least one value, the
/// boundary tie it re-reads and trims, AND the lookahead row that tells the server more data follows. At 2 a resumed
/// page has no room at all and every value costs an empty page with a continuation point (the #692 defect, re-created
/// by configuration), so 1 and 2 are raised to 3 (<see cref="EffectiveMaxRawValuesPerRead"/>).
/// </summary>
public const int MinMaxRawValuesPerRead = 3;
/// <summary>
/// Every backend: the most raw values the server returns on ONE HistoryRead-Raw page, and the most rows one
/// backend raw read may materialise — the server's own row ceiling (review R08:1.1, #736). A client's
/// <c>NumValuesPerNode = 0</c> ("all values") no longer reads a whole window into memory: it is served as pages of at
/// most this many values, each with a continuation point (Part 11 allows fewer values than requested with one), through
/// the same tie-safe paging a capped read uses (<c>RawPageLimit</c>: page + boundary ties + lookahead ≤ cap, so a first
/// page holds cap − 1 values). A <c>NumValuesPerNode</c> above it is paged at it too. Advertised as
/// <c>HistoryServerCapabilities/MaxReturnDataValues</c>. The backends' own <c>TOP</c> / <c>LIMIT</c> / <c>max_values</c>
/// are bounded by it independently: a raw read asking for more than this (or for everything) reads at most cap + 1 rows
/// and is REFUSED (<c>BadTooManyOperations</c>) when the window holds more — never answered truncated — so a caller that
/// bypasses the node manager's paging cannot bypass the bound. On the Gateway backend a strict gateway's learned row
/// ceiling (less its boundary-row allowance) still applies when lower. The tie-cluster bound of a resumed page is also
/// at most cap − 1 (as under a gateway ceiling; <see cref="MaxTieClusterOverfetch"/> above that is inert), and a
/// page whose boundary tie cluster is that large steps past it with one empty page. Default 10 000. Non-positive =
/// unbounded, the pre-#736 behaviour (a <see cref="Validate"/> warning); below <see cref="MinMaxRawValuesPerRead"/>
/// (3) is raised to it.
/// </summary>
public int MaxRawValuesPerRead { get; init; } = DefaultMaxRawValuesPerRead;
/// <summary>
/// <see cref="MaxRawValuesPerRead"/> as the backends and the node manager apply it: <c>0</c> (unbounded) when
/// non-positive, otherwise at least <see cref="MinMaxRawValuesPerRead"/>. The one place the rule lives.
/// </summary>
public int EffectiveMaxRawValuesPerRead => MaxRawValuesPerRead <= 0 ? 0 : Math.Max(MinMaxRawValuesPerRead, MaxRawValuesPerRead);
/// <summary>
/// how far into the future an OPC UA HistoryUpdate (Insert / Replace / Update) item's
/// <c>SourceTimestamp</c> may lie, measured from the server clock. An item beyond <c>now + skew</c> is refused
@@ -208,6 +244,10 @@ public sealed class ServerHistorianOptions
warnings.Add($"ServerHistorian:MaxConcurrentHistoryReadBatches is {MaxConcurrentHistoryReadBatches} — must be > 0; the process-wide HistoryRead batch limiter is disabled.");
if (MaxTimeWeightedRawRows <= 0)
warnings.Add($"ServerHistorian:MaxTimeWeightedRawRows is {MaxTimeWeightedRawRows} — must be > 0; time-weighted aggregate reads on the SqlServer/QuestDb backends stream an unbounded number of raw samples.");
if (MaxRawValuesPerRead <= 0)
warnings.Add($"ServerHistorian:MaxRawValuesPerRead is {MaxRawValuesPerRead} — must be > 0; HistoryRead-Raw pages and backend raw reads are unbounded, so a client asking for a whole window (NumValuesPerNode = 0) materialises every sample in memory.");
else if (MaxRawValuesPerRead < MinMaxRawValuesPerRead)
warnings.Add($"ServerHistorian:MaxRawValuesPerRead is {MaxRawValuesPerRead} — must be at least {MinMaxRawValuesPerRead} (a value, its boundary tie and the lookahead row); treated as {MinMaxRawValuesPerRead}.");
if (EventReadHorizon < TimeSpan.Zero)
warnings.Add($"ServerHistorian:EventReadHorizon is {EventReadHorizon} — negative is treated as no horizon (00:00:00); HistoryReadEvents reads every window in full.");
if (HistoryReadDeadline <= TimeSpan.Zero)
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)
{