Merge branch 'master' into worktree-agent-a720278a57ae41792

Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk
This commit is contained in:
Joseph Doherty
2026-11-11 13:25:58 -05:00
69 changed files with 6709 additions and 915 deletions
@@ -0,0 +1,117 @@
using Microsoft.EntityFrameworkCore;
using ZB.MOM.WW.OtOpcUa.Commons.Interfaces;
using ZB.MOM.WW.OtOpcUa.Configuration;
namespace ZB.MOM.WW.OtOpcUa.AdminUI.Clients;
/// <summary>
/// Cluster-wide counting for the <c>/fleet/diagnostics</c> "Faulted devices" tile (#786). Pure, so it is
/// unit-testable without bUnit.
/// </summary>
/// <remarks>
/// <para>
/// <b>Owner decision 2026-11-11:</b> the tile counts DISTINCT devices across the nodes of one application
/// cluster (a redundant pair), and only devices in <c>Faulted</c> — <c>Stopped</c> is a separate per-node
/// column. Both nodes of a pair run the same driver instances against the same devices, so summing the two
/// nodes' per-driver counts reported every faulted device twice.
/// </para>
/// <para>
/// <b>The device identity key is (driver-instance id, host name)</b>, both compared case-insensitively:
/// <list type="bullet">
/// <item>
/// The driver-instance id is <see cref="DriverInstanceDiagnostics.Name"/> — the string id from the
/// deployed artifact, identical on both nodes of a pair (the row Guid is identical too, but can be
/// <see cref="Guid.Empty"/> on a node with no ConfigDb row behind its spec, so the string id is the
/// sturdier key and the one every other surface uses).
/// </item>
/// <item>
/// The host name is the driver's own <c>HostConnectivityStatus.HostName</c> (e.g. <c>host:port</c>
/// for Modbus/S7, a Platform/Engine name for Galaxy) — driver-defined from the same authored config,
/// so the same device yields the same name on both nodes. It is scoped by the driver id because two
/// different drivers can legitimately name the same endpoint and are counted as two devices.
/// </item>
/// </list>
/// A single-host driver (no <c>IHostConnectivityProbe</c>) reports no host names: the driver instance itself
/// is the device. Its faulted figure (0 or 1) — and any count a pre-#786 sender reports without names —
/// is "unnamed", and unnamed counts for one driver are combined by MAX across nodes, never summed: the
/// same unnamed device on both nodes counts once, which is the most the snapshot can prove distinct.
/// </para>
/// <para>
/// A node that did not answer (the client degrades to an empty snapshot) contributes nothing, so the figure
/// is a lower bound over the nodes that did — the page shows how many answered.
/// </para>
/// </remarks>
public static class FleetDiagnosticsAggregation
{
/// <summary>Counts the distinct faulted devices across a set of node snapshots (normally one cluster's nodes).</summary>
/// <param name="snapshots">One snapshot per node; duplicates of a node are harmless (the count is a set union).</param>
/// <returns>The number of distinct (driver id, host name) devices faulted on at least one node.</returns>
public static int DistinctFaultedDevices(IEnumerable<NodeDiagnosticsSnapshot> snapshots)
{
ArgumentNullException.ThrowIfNull(snapshots);
var named = new HashSet<(string Driver, string Host)>(DeviceKeyComparer.Instance);
var unnamedMax = new Dictionary<string, int>(StringComparer.OrdinalIgnoreCase);
foreach (var snapshot in snapshots)
{
foreach (var driver in snapshot.Drivers)
{
var names = driver.FaultedHostNames ?? Array.Empty<string>();
var namedHere = 0;
foreach (var host in names.Where(n => !string.IsNullOrWhiteSpace(n)).Distinct(StringComparer.OrdinalIgnoreCase))
{
named.Add((driver.Name, host));
namedHere++;
}
var unnamed = Math.Max(0, driver.FaultedDevices - namedHere);
if (unnamed > 0)
unnamedMax[driver.Name] = Math.Max(unnamedMax.GetValueOrDefault(driver.Name), unnamed);
}
}
return named.Count + unnamedMax.Values.Sum();
}
/// <summary>
/// The node ids (<c>host:port</c>) of every enabled node in the same application cluster as
/// <paramref name="nodeId"/>, the node itself included. A node with no <c>ClusterNode</c> row (or an empty
/// ConfigDb) yields just itself, so the tile degrades to a single-node count rather than failing.
/// </summary>
/// <param name="db">The config DB context.</param>
/// <param name="nodeId">The selected node's id, as <c>/fleet</c> shows it.</param>
/// <param name="ct">Cancellation token.</param>
/// <returns>The cluster id (null when unknown) and its node ids, the selected node first.</returns>
public static async Task<(string? ClusterId, IReadOnlyList<string> NodeIds)> LoadClusterNodeIdsAsync(
OtOpcUaConfigDbContext db, string nodeId, CancellationToken ct = default)
{
ArgumentNullException.ThrowIfNull(db);
var self = await db.ClusterNodes.AsNoTracking()
.Where(n => n.NodeId == nodeId)
.Select(n => n.ClusterId)
.FirstOrDefaultAsync(ct);
if (self is null) return (null, new[] { nodeId });
var peers = await db.ClusterNodes.AsNoTracking()
.Where(n => n.ClusterId == self && n.Enabled && n.NodeId != nodeId)
.Select(n => n.NodeId)
.ToListAsync(ct);
peers.Sort(StringComparer.OrdinalIgnoreCase);
return (self, new[] { nodeId }.Concat(peers).ToList());
}
private sealed class DeviceKeyComparer : IEqualityComparer<(string Driver, string Host)>
{
public static readonly DeviceKeyComparer Instance = new();
public bool Equals((string Driver, string Host) x, (string Driver, string Host) y) =>
StringComparer.OrdinalIgnoreCase.Equals(x.Driver, y.Driver)
&& StringComparer.OrdinalIgnoreCase.Equals(x.Host, y.Host);
public int GetHashCode((string Driver, string Host) key) =>
HashCode.Combine(
StringComparer.OrdinalIgnoreCase.GetHashCode(key.Driver),
StringComparer.OrdinalIgnoreCase.GetHashCode(key.Host));
}
}
@@ -86,9 +86,17 @@ else
<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>
@* #786: distinct devices in Faulted across every node of the selected node's cluster (a device faulted on
both nodes of a redundant pair counts once) — Stopped is not a fault and is a per-node column below.
See FleetDiagnosticsAggregation for the identity key. *@
<div class="agg-card alert"
title="Distinct devices in Faulted state across the nodes of @(_clusterId ?? "this node's cluster") that answered (@_clusterAnswered of @_clusterNodeCount). Stopped devices are not counted; see the per-driver Stopped column.">
<div class="agg-label">Faulted devices@(_clusterId is { } cid ? $" (cluster {cid})" : "")</div>
<div class="agg-value numeric">@_clusterFaulted</div>
@if (_clusterNodeCount > 1)
{
<div class="small text-muted">@_clusterAnswered of @_clusterNodeCount nodes answered</div>
}
</div>
</section>
@@ -124,6 +132,7 @@ else
<th>State</th>
<th class="numeric">Connected devices</th>
<th class="numeric">Faulted devices</th>
<th class="numeric" title="Hosts the driver reports Stopped (e.g. taken off scan) — not counted as faulted">Stopped 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>
@@ -144,6 +153,7 @@ else
<td><span class="chip @StateChipClass(d.State)">@d.State</span></td>
<td class="numeric">@d.ConnectedDevices</td>
<td class="numeric">@d.FaultedDevices</td>
<td class="numeric">@d.StoppedDevices</td>
<td>@d.LastChangeUtc.ToString("u")</td>
</tr>
}
@@ -165,6 +175,12 @@ else
private NodeDiagnosticsSnapshot? _snapshot;
private IReadOnlyDictionary<string, string> _displayNames = new Dictionary<string, string>();
private bool _loading;
// #786 cluster-wide faulted-device tile: the selected node's cluster, how many of its nodes answered, and the
// distinct faulted-device count across them.
private string? _clusterId;
private int _clusterNodeCount;
private int _clusterAnswered;
private int _clusterFaulted;
private DateTime? _fetchedUtc;
// Page-lifetime cancellation: a fetch in flight when the circuit goes away is abandoned.
@@ -222,6 +238,7 @@ else
// 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;
await LoadClusterFaultedAsync(_snapshot);
_displayNames = await LoadDisplayNamesAsync(_snapshot);
}
catch (Exception ex) when (ex is OperationCanceledException or ObjectDisposedException)
@@ -235,6 +252,48 @@ else
}
}
// #786: fetch the selected node's cluster peers too and count distinct faulted devices across all of them.
// The ConfigDb lookup is enrichment: on failure the tile falls back to the selected node alone.
private async Task LoadClusterFaultedAsync(NodeDiagnosticsSnapshot selected)
{
string? clusterId = null;
IReadOnlyList<string> nodeIds = new[] { selected.NodeId.Value };
try
{
await using var db = await DbFactory.CreateDbContextAsync(_cts.Token);
(clusterId, nodeIds) = await FleetDiagnosticsAggregation.LoadClusterNodeIdsAsync(db, selected.NodeId.Value, _cts.Token);
}
catch (Exception ex) when (ex is not OperationCanceledException and not ObjectDisposedException)
{
// ConfigDb hiccup — count this node only.
}
var snapshots = new List<NodeDiagnosticsSnapshot> { selected };
var peers = nodeIds.Where(n => !string.Equals(n, selected.NodeId.Value, StringComparison.OrdinalIgnoreCase)).ToList();
var peerSnapshots = await Task.WhenAll(peers.Select(FetchPeerAsync));
snapshots.AddRange(peerSnapshots.OfType<NodeDiagnosticsSnapshot>());
_clusterId = clusterId;
_clusterNodeCount = nodeIds.Count;
// The client degrades an unreachable node to an empty snapshot; one with drivers clearly answered. An empty
// answer is indistinguishable from "no answer", so it is counted as not answered — it contributes nothing either way.
_clusterAnswered = 1 + peerSnapshots.Count(p => p is { Drivers.Count: > 0 });
_clusterFaulted = FleetDiagnosticsAggregation.DistinctFaultedDevices(snapshots);
}
private async Task<NodeDiagnosticsSnapshot?> FetchPeerAsync(string nodeId)
{
try
{
return await Diagnostics.GetDiagnosticsAsync(NodeId.Parse(nodeId), _cts.Token);
}
catch (Exception ex) when (ex is not OperationCanceledException and not ObjectDisposedException)
{
// A malformed peer row (NodeId.Parse) — the peer just does not contribute.
return null;
}
}
// The authored name for a driver row, else its id (no ConfigDb row, or the name lookup failed).
private string DisplayNameOf(DriverInstanceDiagnostics d) =>
_displayNames.TryGetValue(d.Name, out var name) ? name : d.Name;
@@ -111,6 +111,30 @@
}
</InputSelect>
</div>
@* #816 / #837: opt-in by owner decision (unset = off). Kept in the Security panel so it is
found next to the mode/policy it depends on; "Default (off)" leaves the key absent. *@
<div class="col-md-4">
<label class="form-label" for="opcuaValidateHostName">Server certificate host-name check</label>
<InputSelect id="opcuaValidateHostName" @bind-Value="_form.OpcUa.ValidateServerHostName" @bind-Value:after="EmitAsync" class="form-select form-select-sm">
<option value="">Default (off)</option>
<option value="true">Validate configured host</option>
<option value="false">Off (NAT / port publish)</option>
</InputSelect>
</div>
<div class="col-12">
<div class="form-text">
<strong>Opt-in (off unless set).</strong> When validating, the server certificate must name the
endpoint host <em>you configured</em>; a miss refuses the connection
(<code>BadCertificateHostNameInvalid</code>). Applies to Sign / SignAndEncrypt only, never to
<code>None</code>, and auto-accept accepts a mismatch. Caveats:
<ul class="mb-0">
<li><strong>NAT / port publish / connecting by IP:</strong> fails unless the certificate lists that exact host or IP — leave off.</li>
<li><strong>IPv6 zone ids</strong> (<code>fe80::1%eth0</code>, link-local) never match — leave off.</li>
<li><strong>Loose match:</strong> the first DNS label alone counts (<code>plc1.site-b.corp</code> passes a certificate for <code>PLC1</code>), so use fully-qualified, site-unique certificate names.</li>
</ul>
Most useful with CA-issued trust; adds little over a pinned self-signed certificate.
</div>
</div>
</div>
</div>
</section>
@@ -195,15 +219,6 @@
</div>
<div class="form-text">Revoked certificates are always rejected.</div>
</div>
<div class="col-md-3">
<label class="form-label">Server host-name check</label>
<InputSelect @bind-Value="_form.OpcUa.ValidateServerHostName" @bind-Value:after="EmitAsync" class="form-select form-select-sm">
<option value="">Default (off)</option>
<option value="true">Validate configured host</option>
<option value="false">Off (NAT / port publish)</option>
</InputSelect>
<div class="form-text">Sign / SignAndEncrypt only (never on None): the server certificate must name the endpoint host you configured. Off for link-local IPv6.</div>
</div>
<div class="col-md-6">
<label class="form-label">Issuer store / CRL directory</label>
<InputText @bind-Value="_form.OpcUa.IssuerStorePath" @bind-Value:after="EmitAsync" class="form-control form-control-sm mono" />
@@ -52,8 +52,32 @@ namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Coordinators;
/// <c>ServerCluster.Enabled</c> is <b>not</b> consulted — nothing else in the codebase consults it,
/// so honouring it here would invent semantics. Disable the nodes, not the cluster row.
/// </para>
/// <para>
/// <b>Recovery load is asynchronous and retried (lmxopcua#868).</b> The in-flight recovery read used to run
/// synchronously in <c>PreStart</c>. When the ConfigDb was unreachable as the singleton started, the throw
/// became an <c>ActorInitializationException</c>, the default strategy STOPPED the actor, and the
/// <c>ClusterSingletonManager</c> never re-creates a singleton that terminated on its own — so every later
/// deployment was told to a dead singleton and hung. The read now runs off the actor thread, a failure
/// retries with exponential backoff (<see cref="DefaultLoadRetryMin"/> doubling to
/// <see cref="DefaultLoadRetryMax"/>, forever), and until it succeeds every message — deployments and acks
/// alike — is STASHED and replayed in order once the recovered state is in place, so a deploy arriving
/// during the outage is queued, never dropped. <see cref="ConfigPublishCoordinatorReadiness"/> reports the
/// loading state (and the last error) to the node's health endpoint as Degraded.
/// </para>
/// <para>
/// <b>The stash is node-local.</b> A singleton hand-over while the coordinator is still Loading discards what it
/// queued (the old incarnation stops with its mailbox). That loses nothing durable: a queued dispatch's row is
/// <c>Dispatching</c> with no expected-ack rows, which the new incarnation's recovery (or AdminOps' lost-dispatch
/// closure) re-drives or closes; a queued ack is the one exception — its node does not resend, so that node is
/// named in the deadline's timeout reason.
/// </para>
/// <para>
/// <b>An ack is held in memory only once it is persisted.</b> <c>NodeDeploymentState</c> is the system of record
/// (driver-only nodes never write their own row), so a failed <c>PersistNodeAck</c> re-queues the ack on a timer
/// instead of throwing — a throw would restart the actor and drop it.
/// </para>
/// </summary>
public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers
public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers, IWithUnboundedStash
{
public const string DeploymentsTopic = "deployments";
public const string DeploymentAcksTopic = "deployment-acks";
@@ -61,8 +85,35 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers
/// apply-outcome timeout is derived from.</summary>
public static readonly TimeSpan DefaultApplyDeadline = DeployTimings.ApplyDeadline;
/// <summary>First retry delay after a failed recovery load (lmxopcua#868).</summary>
public static readonly TimeSpan DefaultLoadRetryMin = TimeSpan.FromSeconds(1);
/// <summary>Ceiling of the recovery-load retry backoff (lmxopcua#868).</summary>
public static readonly TimeSpan DefaultLoadRetryMax = TimeSpan.FromSeconds(30);
private readonly IDbContextFactory<OtOpcUaConfigDbContext> _dbFactory;
private readonly TimeSpan _applyDeadline;
private readonly TimeSpan _loadRetryMin;
private readonly TimeSpan _loadRetryMax;
private readonly ConfigPublishCoordinatorReadiness? _readiness;
private int _failedLoads;
/// <summary>Deployments whose dispatch was queued while Loading (never sent: recovery re-dispatches them).</summary>
private readonly HashSet<DeploymentId> _queuedDispatches = new();
/// <summary>
/// Acks of the current deployment that arrived but could not be persisted yet (review M2), one per node — the
/// newest wins. They are retried on a backoff, block the deadline from deciding without them, and are
/// re-sent to Self across a restart (<see cref="PreRestart"/>).
/// </summary>
private readonly Dictionary<NodeId, PendingPersist> _unpersisted = new(NodeIdComparer);
private string? _lastPersistError;
/// <summary>When the current deadline was first deferred on an unpersisted ack (bounds the deferral).</summary>
private DateTime? _deadlineDeferredSinceUtc;
/// <summary>How long a deadline may wait on unpersisted acks before deciding from memory (10 × the retry ceiling).</summary>
private TimeSpan MaxDeadlineDeferral => TimeSpan.FromTicks(_loadRetryMax.Ticks * 10);
/// <summary>
/// Where central→node commands are handed off (per-cluster mesh Phase 2), or
@@ -115,20 +166,30 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers
/// <summary>Gets the timer scheduler for managing apply deadlines.</summary>
public ITimerScheduler Timers { get; set; } = null!;
/// <summary>Holds every message that arrives before the recovery load completes (lmxopcua#868).</summary>
public IStash Stash { get; set; } = null!;
/// <summary>Creates actor props for the ConfigPublishCoordinator with the specified configuration.</summary>
/// <param name="dbFactory">The database context factory for accessing configuration data.</param>
/// <param name="applyDeadline">The timeout for waiting for apply acknowledgments from cluster nodes.</param>
/// <returns>The Akka.NET <see cref="Akka.Actor.Props"/> used to spawn this actor.</returns>
/// <param name="meshRouter">
/// Central comm actor the dispatch is handed to, or <see langword="null"/> to publish on the
/// DistributedPubSub mediator directly (the original behaviour).
/// </param>
/// <param name="readiness">Where the recovery-load state is reported for health, or <see langword="null"/>.</param>
/// <param name="loadRetryMin">First recovery-load retry delay; <see langword="null"/> = <see cref="DefaultLoadRetryMin"/>.</param>
/// <param name="loadRetryMax">Recovery-load retry ceiling; <see langword="null"/> = <see cref="DefaultLoadRetryMax"/>.</param>
/// <returns>The Akka.NET <see cref="Akka.Actor.Props"/> used to spawn this actor.</returns>
public static Props Props(
IDbContextFactory<OtOpcUaConfigDbContext> dbFactory,
TimeSpan? applyDeadline = null,
IActorRef? meshRouter = null) =>
Akka.Actor.Props.Create(() =>
new ConfigPublishCoordinator(dbFactory, applyDeadline ?? DefaultApplyDeadline, meshRouter));
IActorRef? meshRouter = null,
ConfigPublishCoordinatorReadiness? readiness = null,
TimeSpan? loadRetryMin = null,
TimeSpan? loadRetryMax = null) =>
Akka.Actor.Props.Create(() => new ConfigPublishCoordinator(
dbFactory, applyDeadline ?? DefaultApplyDeadline, meshRouter, readiness,
loadRetryMin ?? DefaultLoadRetryMin, loadRetryMax ?? DefaultLoadRetryMax));
/// <summary>
/// Initializes a new instance of the <see cref="ConfigPublishCoordinator"/> class.
@@ -136,45 +197,200 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers
/// <param name="dbFactory">The database context factory for persisting deployment state.</param>
/// <param name="applyDeadline">The timeout for waiting for per-node apply acknowledgments.</param>
/// <param name="meshRouter">Central comm actor, or <see langword="null"/> for the mediator.</param>
/// <param name="readiness">Where the recovery-load state is reported, or <see langword="null"/>.</param>
/// <param name="loadRetryMin">First recovery-load retry delay.</param>
/// <param name="loadRetryMax">Recovery-load retry ceiling.</param>
public ConfigPublishCoordinator(
IDbContextFactory<OtOpcUaConfigDbContext> dbFactory,
TimeSpan applyDeadline,
IActorRef? meshRouter = null)
IActorRef? meshRouter = null,
ConfigPublishCoordinatorReadiness? readiness = null,
TimeSpan? loadRetryMin = null,
TimeSpan? loadRetryMax = null)
{
_dbFactory = dbFactory;
_applyDeadline = applyDeadline;
_meshRouter = meshRouter;
_readiness = readiness;
_loadRetryMin = loadRetryMin ?? DefaultLoadRetryMin;
_loadRetryMax = loadRetryMax ?? DefaultLoadRetryMax;
Loading();
}
/// <summary>
/// The state before the recovery load has succeeded: only the load's own outcome and retry tick are
/// handled; everything else is stashed and replayed, in arrival order, once <see cref="Ready"/> is in force.
/// </summary>
private void Loading()
{
Receive<RecoveryLoaded>(HandleRecoveryLoaded);
Receive<RecoveryLoadFailed>(HandleRecoveryLoadFailed);
Receive<RetryRecoveryLoad>(_ => StartRecoveryLoad());
Receive<SubscribeAck>(_ => { /* DPS subscribe confirmation */ });
Receive<DispatchDeployment>(msg =>
{
_log.Info(
"DispatchDeployment {Id} queued: the coordinator has not yet loaded its recovery state from the " +
"ConfigDb ({Failed} failed attempt(s) so far); it is dispatched as soon as the load succeeds",
msg.DeploymentId, _failedLoads);
_queuedDispatches.Add(msg.DeploymentId);
Stash.Stash();
});
ReceiveAny(_ => Stash.Stash());
}
/// <summary>The normal operating state, entered once the recovery load has been applied.</summary>
private void Ready()
{
Receive<DispatchDeployment>(HandleDispatch);
Receive<ApplyAck>(HandleAck);
Receive<RetryAckPersist>(HandleRetryAckPersist);
Receive<DeadlineElapsed>(HandleDeadline);
Receive<SubscribeAck>(_ => { /* DPS subscribe confirmation */ });
// A late load outcome / retry tick cannot normally arrive once Ready; ignored, never stashed.
Receive<RecoveryLoaded>(_ => { });
Receive<RecoveryLoadFailed>(_ => { });
Receive<RetryRecoveryLoad>(_ => { });
}
/// <inheritdoc />
protected override void PreStart()
{
// On startup recover any deployment that was mid-flight when a prior singleton instance
// died. We re-derive _expectedAcks from NodeDeploymentState, replay the ACKs
// that already landed in the DB, and resume the deadline timer.
// Subscribe to per-node ApplyAck broadcasts so DriverHostActors on remote members can
// route their ACKs to whichever node currently hosts this singleton.
DistributedPubSub.Get(Context.System).Mediator.Tell(new Subscribe(DeploymentAcksTopic, Self));
using var db = _dbFactory.CreateDbContext();
// lmxopcua#868: NO database access here. A throw from PreStart stops the actor for good (the
// singleton manager does not re-create it), so the recovery read runs asynchronously and retries.
_readiness?.MarkLoading(_failedLoads, lastError: null);
StartRecoveryLoad();
}
/// <inheritdoc />
protected override void PreRestart(Exception reason, object message)
{
// Timers die with this incarnation; the mailbox survives. Re-send every ack that is still unpersisted so the
// next incarnation (which stashes while Loading) persists and counts it — its node will not resend (M2).
foreach (var pending in _unpersisted.Values) Self.Tell(pending.Ack);
base.PreRestart(reason, message);
}
/// <inheritdoc />
protected override void PostStop()
{
// A restart re-enters PreStart (Loading again); a stop or hand-over leaves this node not hosting it.
_readiness?.MarkNotHosted();
base.PostStop();
}
/// <summary>
/// Reads the deployment that was mid-flight when a prior singleton instance died, off the actor thread,
/// and pipes the outcome back. Touches no actor state.
/// </summary>
private void StartRecoveryLoad()
{
LoadRecoveryAsync(_dbFactory).PipeTo(
Self,
success: loaded => loaded,
failure: ex => new RecoveryLoadFailed(ex is AggregateException { InnerException: { } inner } ? inner : ex));
}
private static async Task<RecoveryLoaded> LoadRecoveryAsync(IDbContextFactory<OtOpcUaConfigDbContext> dbFactory)
{
// Yield first so a factory or provider that throws synchronously still faults the TASK (piped back as
// RecoveryLoadFailed) rather than the caller on the actor thread.
await Task.Yield();
await using var db = await dbFactory.CreateDbContextAsync().ConfigureAwait(false);
// Projected: the row carries the whole ArtifactBlob (varbinary(max)), which this recovery never
// reads — and this query runs on every singleton start and hand-over.
var inflight = db.Deployments
var inflight = await db.Deployments
.Where(d => d.Status == DeploymentStatus.Dispatching || d.Status == DeploymentStatus.AwaitingApplyAcks)
.OrderByDescending(d => d.CreatedAtUtc)
.Select(d => new { d.DeploymentId, d.RevisionHash, d.CreatedAtUtc })
.FirstOrDefault();
.FirstOrDefaultAsync()
.ConfigureAwait(false);
if (inflight is null) return new RecoveryLoaded(null);
var nodeStates = await db.NodeDeploymentStates
.Where(x => x.DeploymentId == inflight.DeploymentId)
.AsNoTracking()
.Select(x => new RecoveredNodeState(x.NodeId, x.Status, x.FailureReason))
.ToListAsync()
.ConfigureAwait(false);
return new RecoveryLoaded(new RecoveredDeployment(
inflight.DeploymentId, inflight.RevisionHash, inflight.CreatedAtUtc, nodeStates));
}
private void HandleRecoveryLoadFailed(RecoveryLoadFailed msg)
{
_failedLoads++;
// Exponential backoff, capped: min * 2^(n-1), with the shift bounded so it never overflows.
var delay = Backoff(_failedLoads);
_readiness?.MarkLoading(_failedLoads, msg.Error.Message);
_log.Warning(msg.Error,
"ConfigPublishCoordinator could not load its recovery state from the ConfigDb (attempt {Attempt}); " +
"retrying in {Delay}. Deployments and acks are queued until it succeeds",
_failedLoads, delay);
Timers.StartSingleTimer(LoadRetryTimerKey, RetryRecoveryLoad.Instance, delay);
}
/// <summary>
/// Applies the recovered state, then switches to <see cref="Ready"/> and replays every stashed message. A
/// decision that must see the queued messages first (the deadline of a recovered deployment that is already
/// past it) is SELF-SENT, so it lands behind the replay; one that must precede them (re-driving a lost
/// dispatch, so a queued duplicate of it is recognised) runs inline. A throw while applying is treated as a
/// failed load — the actor stays Loading and retries on the backoff, never a tight restart loop.
/// </summary>
private void HandleRecoveryLoaded(RecoveryLoaded msg)
{
Timers.Cancel(LoadRetryTimerKey);
try
{
ApplyRecovery(msg.Inflight);
}
catch (Exception ex)
{
_log.Error(ex, "ConfigPublishCoordinator failed to apply its recovered state; retrying the load");
ResetForNext();
HandleRecoveryLoadFailed(new RecoveryLoadFailed(ex));
return;
}
if (_failedLoads > 0)
_log.Info("ConfigPublishCoordinator loaded its recovery state after {Failed} failed attempt(s)", _failedLoads);
_failedLoads = 0;
_queuedDispatches.Clear();
Become(Ready);
_readiness?.MarkReady();
Stash.UnstashAll();
}
/// <summary>
/// Recovers the deployment that was mid-flight when a prior singleton instance died: re-derives
/// <c>_expectedAcks</c> from <c>NodeDeploymentState</c>, replays the acks that already landed, and resumes
/// the deadline timer.
/// </summary>
private void ApplyRecovery(RecoveredDeployment? inflight)
{
if (inflight is null) return;
_current = new DeploymentId(inflight.DeploymentId);
var nodeStates = db.NodeDeploymentStates
.Where(x => x.DeploymentId == inflight.DeploymentId)
.AsNoTracking()
.ToList();
var nodeStates = inflight.NodeStates;
// Review L5: a row recovery cannot interpret (blank revision on a row it must re-dispatch, blank node id) would
// throw on every load and keep the coordinator Loading forever. Close it with a reason and carry on.
var malformed = DescribeMalformed(inflight);
if (malformed is not null)
{
_log.Error("Coordinator cannot recover deployment {Id}: {Problem}. Closing it as PartiallyFailed", _current, malformed);
using var db = _dbFactory.CreateDbContext();
UpdateDeploymentStatus(db, _current.Value, DeploymentStatus.PartiallyFailed, onlyIfInFlight: true);
SetFailureReason(db, Truncate(
$"The deploy coordinator could not recover this deployment: {malformed}. Redeploy."));
db.SaveChanges();
ResetForNext();
return;
}
// Resume the deadline timer using the remaining time. The deadline runs from when the
// deployment was first marked AwaitingApplyAcks (Deployment.CreatedAtUtc is a close enough
@@ -194,19 +410,38 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers
// re-driven through the normal handler (bounded: CreatedAtUtc never moves, so a dispatch that
// keeps throwing ends up on the other branch once the deadline passes); past it the row is
// closed as TimedOut with a reason that names the condition.
if (remaining > TimeSpan.Zero)
if (_queuedDispatches.Contains(_current.Value))
{
// Review L4: its dispatch was QUEUED here while the load was failing — it was never sent, so however
// long the outage lasted it is not a lost dispatch. Send it now with a full, fresh deadline.
_log.Info(
"Coordinator dispatching deployment {Id}, queued while its recovery state was loading",
_current);
HandleDispatch(new DispatchDeployment(
_current.Value, RevisionHash.Parse(inflight.RevisionHash), CorrelationId.NewId()));
}
else if (remaining > TimeSpan.Zero)
{
_log.Warning(
"Coordinator recovered deployment {Id} with NO expected-ack rows — the dispatch was lost " +
"before any node was expected (coordinator restart or singleton hand-over mid-dispatch); " +
"re-dispatching with {Remaining} of the apply deadline left",
"re-dispatching with a fresh apply deadline ({Remaining} of the original was left)",
_current, remaining);
Self.Tell(new DispatchDeployment(
// A FULL deadline (third review #3, superseding the second review's "remaining time"): a node needs its
// whole outcome timeout (DeployTimings.ApplyDeadline - ApplyOutcomeSafetyMargin) plus that margin to
// report — i.e. the full ApplyDeadline — so ANY remaining time shorter than that let the nodes apply
// and the deployment time out moments later. Bounded anyway: there is no dispatched-at column, so a
// SECOND restart recomputes from CreatedAtUtc and, past the deadline by then, closes the row instead.
HandleDispatch(new DispatchDeployment(
_current.Value, RevisionHash.Parse(inflight.RevisionHash), CorrelationId.NewId()));
}
else
{
Self.Tell(new DeadlineElapsed(_current.Value));
// Deliberately INLINE, unlike the populated-set deadline below: with no expected set no queued ack
// can count, so there is nothing to let the replay apply first — and a queued dispatch of this very
// row must meet the TimedOut row (terminal-state guard ignores it), not publish to every node and
// be timed out the moment a self-sent deadline caught up with it.
HandleDeadline(new DeadlineElapsed(_current.Value));
}
return;
}
@@ -220,20 +455,39 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers
if (IsDegraded(s.Status, s.FailureReason)) _degraded.Add(NodeId.Parse(s.NodeId));
}
// Every expected ack is already in the DB (a seal whose SaveChanges failed, or a fused node's own
// DriverHostActor row — which that actor writes Applied only once the address-space outcome is known, so a
// non-Applying row is always a FINAL outcome; review M1): finish it now. Waiting for the deadline would leave it in flight — the deadline
// can only time out an INCOMPLETE set.
if (_acks.Count == _expectedAcks.Count)
{
_log.Info("Coordinator recovered deployment {Id} with every ack already landed; completing it", _current);
CompleteFromAcks();
return;
}
if (remaining <= TimeSpan.Zero)
{
// Self-sent, so acks queued during the load (which arrived in time) are applied first.
Self.Tell(new DeadlineElapsed(_current.Value));
return;
}
else
{
Timers.StartSingleTimer(DeadlineTimerKey, new DeadlineElapsed(_current.Value), remaining);
}
Timers.StartSingleTimer(DeadlineTimerKey, new DeadlineElapsed(_current.Value), remaining);
_log.Info("Coordinator recovered in-flight deployment {Id} ({Acked}/{Total} acks landed)",
_current, _acks.Count, _expectedAcks.Count);
}
private void HandleDispatch(DispatchDeployment msg)
{
// A duplicate of the deployment already in flight with a seeded expected set (a dispatch queued while the
// recovery load ran, for the very row that load re-drove) must not reset the acks already gathered.
if (_current == msg.DeploymentId && _expectedAcks.Count > 0)
{
_log.Debug("Ignoring duplicate DispatchDeployment {Id}: already in flight", msg.DeploymentId);
return;
}
// Terminal-state guard (fix round 1): a late or re-delivered dispatch for a row that is already
// Sealed / PartiallyFailed / TimedOut — e.g. one AdminOps closed as a lost dispatch while this message
// was still queued — must never flip it back to AwaitingApplyAcks or seed an expected set for it.
@@ -355,20 +609,121 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers
return;
}
// Persist FIRST: the row is the system of record (driver-only nodes never write their own, and nodes do not
// resend). A throw here used to restart the actor and drop the ack; it is re-queued instead.
if (!TryPersistAck(msg)) return;
if (_acks.Count < _expectedAcks.Count) return;
CompleteFromAcks();
}
/// <summary>
/// Persists an ack and, only on success, counts it (and drops any older unpersisted ack of that node, with its
/// retry timer — review L3). On failure the ack is held in <see cref="_unpersisted"/> and retried on a backoff.
/// </summary>
/// <returns><see langword="true"/> when the ack is persisted and counted.</returns>
private bool TryPersistAck(ApplyAck msg)
{
try
{
PersistNodeAck(msg);
}
catch (Exception ex) when (IsPoison(ex))
{
// Third review #1: a write the DB REJECTS (constraint, truncation, …) fails identically on every retry;
// retrying it forever wedged the deployment (the deadline deferred on it) and every later deploy. The
// node DID ack, so count it from memory and say loudly that its row is not recorded.
_log.Error(ex,
"The ApplyAck from {Node} for {Id} ({Outcome}) was REJECTED by the ConfigDb and will not be retried; " +
"counting it from memory. Its NodeDeploymentState row stays unrecorded",
msg.NodeId, msg.DeploymentId, msg.Outcome);
DropUnpersisted(msg.NodeId);
CountAck(msg);
return true;
}
catch (Exception ex)
{
var attempts = _unpersisted.TryGetValue(msg.NodeId, out var prior) ? prior.Attempts + 1 : 1;
_unpersisted[msg.NodeId] = new PendingPersist(msg, attempts);
_lastPersistError = ex.Message;
var delay = Backoff(attempts);
// Review L7: the stack trace once, then one line per retry.
if (attempts == 1)
_log.Warning(ex, "Could not persist the ApplyAck from {Node} for {Id}; retrying in {Delay}",
msg.NodeId, msg.DeploymentId, delay);
else
_log.Warning("Still cannot persist the ApplyAck from {Node} for {Id} (attempt {Attempt}: {Error}); retrying in {Delay}",
msg.NodeId, msg.DeploymentId, attempts, ex.Message, delay);
Timers.StartSingleTimer(AckRetryTimerKey(msg.NodeId), new RetryAckPersist(msg), delay);
ReportPersistBacklog();
return false;
}
DropUnpersisted(msg.NodeId);
CountAck(msg);
return true;
}
private void CountAck(ApplyAck msg)
{
_acks[msg.NodeId] = msg.Outcome;
// An old node never sets Degraded, so its ack takes the clean branch exactly as before.
if (IsDegradedAck(msg)) _degraded.Add(msg.NodeId);
else _degraded.Remove(msg.NodeId);
PersistNodeAck(msg);
}
if (_acks.Count < _expectedAcks.Count) return;
private void DropUnpersisted(NodeId node)
{
if (!_unpersisted.Remove(node)) return;
Timers.Cancel(AckRetryTimerKey(node));
if (_unpersisted.Count == 0) _lastPersistError = null;
ReportPersistBacklog();
}
/// <summary>
/// A non-transient rejection of the write itself: an EF <see cref="DbUpdateException"/> (not a concurrency
/// conflict) with no transient database error, timeout or socket failure underneath. Connectivity failures
/// surface as other exception types (or as transient <see cref="System.Data.Common.DbException"/>s) and stay
/// retryable.
/// </summary>
internal static bool IsPoison(Exception ex)
{
if (ex is not DbUpdateException || ex is DbUpdateConcurrencyException) return false;
for (var e = ex.InnerException; e is not null; e = e.InnerException)
{
if (e is System.Data.Common.DbException { IsTransient: true }
or TimeoutException or System.Net.Sockets.SocketException or System.IO.IOException)
return false;
}
return true;
}
/// <summary>A retry tick: only the node's CURRENT unpersisted ack is retried; a superseded one is dropped (L3).</summary>
private void HandleRetryAckPersist(RetryAckPersist msg)
{
if (!_unpersisted.TryGetValue(msg.Ack.NodeId, out var pending) || !ReferenceEquals(pending.Ack, msg.Ack)) return;
HandleAck(msg.Ack);
}
private void ReportPersistBacklog() => _readiness?.MarkAckPersistBacklog(_unpersisted.Count, _lastPersistError);
/// <summary>min * 2^(attempt-1), capped at the max — shared by the load and ack-persist retries.</summary>
private TimeSpan Backoff(int attempt)
{
var factor = 1L << Math.Min(Math.Max(attempt, 1) - 1, 20);
return TimeSpan.FromTicks(Math.Min(_loadRetryMax.Ticks, _loadRetryMin.Ticks * factor));
}
/// <summary>Seals (all Applied) or marks partially failed — called once every expected ack is in.</summary>
private void CompleteFromAcks()
{
if (_acks.Values.All(o => o == ApplyAckOutcome.Applied))
SealDeployment();
else
MarkPartiallyFailed();
}
private static string AckRetryTimerKey(NodeId node) => $"ack-retry:{node.Value.ToUpperInvariant()}";
private void PersistNodeAck(ApplyAck msg)
{
using var db = _dbFactory.CreateDbContext();
@@ -386,24 +741,23 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers
? Truncate(DegradedReasonPrefix + (string.IsNullOrWhiteSpace(msg.DegradedReason)
? "address-space apply degraded (no reason given)"
: msg.DegradedReason))
: msg.FailureReason;
// Third review #1: a Failed ack's reason is the node's free text (often an exception message) and was
// written UNBOUNDED — on SQL Server an overflow refuses the SaveChanges forever.
: FailureReasons.Truncate(msg.FailureReason);
db.SaveChanges();
}
private static bool IsDegradedAck(ApplyAck msg) => msg.Outcome == ApplyAckOutcome.Applied && msg.Degraded;
// FailureReason is nvarchar(2048) on both tables (OtOpcUaConfigDbContext); a reason built from a node's
// free text must never fail the SaveChanges that records the ack.
private const int FailureReasonMaxLength = 2048;
private static string Truncate(string s) =>
s.Length <= FailureReasonMaxLength ? s : s[..(FailureReasonMaxLength - 1)] + "…";
// free text must never fail the SaveChanges that records the ack. One shared bound (FailureReasons).
private static string Truncate(string s) => FailureReasons.Truncate(s)!;
private void SealDeployment()
{
if (_current is null) return;
using var db = _dbFactory.CreateDbContext();
UpdateDeploymentStatus(db, _current.Value, DeploymentStatus.Sealed, sealNow: true);
UpdateDeploymentStatus(db, _current.Value, DeploymentStatus.Sealed, sealNow: true, onlyIfInFlight: true);
if (_degraded.Count > 0)
{
// Still Sealed: every node DID apply this revision and is running it, and "latest sealed" is
@@ -432,7 +786,7 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers
{
if (_current is null) return;
using var db = _dbFactory.CreateDbContext();
UpdateDeploymentStatus(db, _current.Value, DeploymentStatus.PartiallyFailed);
UpdateDeploymentStatus(db, _current.Value, DeploymentStatus.PartiallyFailed, onlyIfInFlight: true);
db.SaveChanges();
_log.Warning("Deployment {Id} partially failed; acks={Acks}", _current.Value,
string.Join(",", _acks.Select(kv => $"{kv.Key.Value}={kv.Value}")));
@@ -447,16 +801,49 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers
msg.DeploymentId, _current);
return;
}
// Review M2: an ack that ARRIVED but is not yet persisted must never be timed out as "no ack from" its node.
// Persist the backlog now; if the DB still refuses, defer the decision (the terminal write needs it too).
if (_unpersisted.Count > 0)
{
foreach (var pending in _unpersisted.Values.ToList()) TryPersistAck(pending.Ack);
_deadlineDeferredSinceUtc ??= DateTime.UtcNow;
if (_unpersisted.Count > 0 && DateTime.UtcNow - _deadlineDeferredSinceUtc.Value >= MaxDeadlineDeferral)
{
// Third review #1: bounded. Past the cap the outcome is decided from what was RECEIVED — the nodes did
// ack — rather than waiting on rows that may never be writable.
_log.Error(
"Deployment {Id}: {Count} received ack(s) still could not be persisted after deferring the deadline " +
"{Cap}; deciding the outcome from the acks held in memory (last error: {Error})",
_current.Value, _unpersisted.Count, MaxDeadlineDeferral, _lastPersistError);
foreach (var pending in _unpersisted.Values.ToList())
{
DropUnpersisted(pending.Ack.NodeId);
CountAck(pending.Ack);
}
}
if (_unpersisted.Count > 0)
{
var delay = Backoff(_unpersisted.Values.Min(p => p.Attempts));
_log.Warning(
"Deployment {Id} reached its apply deadline with {Count} received ack(s) not yet persisted; deferring " +
"the outcome {Delay} until they are", _current.Value, _unpersisted.Count, delay);
Timers.StartSingleTimer(DeadlineTimerKey, msg, delay);
return;
}
}
// Only a NON-EMPTY expected set can be "fully acked": with nothing expected, 0 == 0 is not the
// everyone-acked race, it is a lost dispatch (see PreStart) that must be closed, not skipped.
if (_expectedAcks.Count > 0 && _acks.Count == _expectedAcks.Count)
{
// Race: every node acked just as the deadline fired. Already sealed/failed elsewhere.
// Every ack is in but nothing completed it (a completion normally resets _current, so the stale-id
// check above catches the ack-vs-deadline race). Complete it rather than leave it in flight forever.
CompleteFromAcks();
return;
}
using var db = _dbFactory.CreateDbContext();
UpdateDeploymentStatus(db, _current.Value, DeploymentStatus.TimedOut);
UpdateDeploymentStatus(db, _current.Value, DeploymentStatus.TimedOut, onlyIfInFlight: true);
if (_expectedAcks.Count == 0)
{
SetFailureReason(db, Truncate(
@@ -512,6 +899,14 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers
private void ResetForNext()
{
Timers.Cancel(DeadlineTimerKey);
_deadlineDeferredSinceUtc = null;
foreach (var node in _unpersisted.Keys) Timers.Cancel(AckRetryTimerKey(node));
if (_unpersisted.Count > 0)
{
_unpersisted.Clear();
_lastPersistError = null;
ReportPersistBacklog();
}
_current = null;
_expectedAcks.Clear();
_acks.Clear();
@@ -519,9 +914,43 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers
}
private const string DeadlineTimerKey = "apply-deadline";
private const string LoadRetryTimerKey = "recovery-load-retry";
/// <summary>An ack whose persist failed, re-delivered by its retry timer.</summary>
private sealed record RetryAckPersist(ApplyAck Ack);
/// <summary>A received ack not yet persisted, and how many persists of it have failed.</summary>
private sealed record PendingPersist(ApplyAck Ack, int Attempts);
/// <summary>Why recovery cannot interpret a row (review L5), or null when it can.</summary>
private static string? DescribeMalformed(RecoveredDeployment d)
{
if (d.NodeStates.Count == 0 && string.IsNullOrWhiteSpace(d.RevisionHash))
return "its RevisionHash is blank, so it cannot be re-dispatched";
if (d.NodeStates.Any(n => string.IsNullOrWhiteSpace(n.NodeId)))
return "one of its NodeDeploymentState rows has a blank NodeId";
return null;
}
public sealed record DeadlineElapsed(DeploymentId DeploymentId);
/// <summary>The recovery read's result: the newest in-flight deployment, or null when none.</summary>
private sealed record RecoveryLoaded(RecoveredDeployment? Inflight);
/// <summary>The recovery read failed (ConfigDb unreachable, ...); retried with backoff.</summary>
private sealed record RecoveryLoadFailed(Exception Error);
/// <summary>The backoff timer's tick: run the recovery read again.</summary>
private sealed class RetryRecoveryLoad
{
public static readonly RetryRecoveryLoad Instance = new();
}
private sealed record RecoveredDeployment(
Guid DeploymentId, string RevisionHash, DateTime CreatedAtUtc, IReadOnlyList<RecoveredNodeState> NodeStates);
private sealed record RecoveredNodeState(string NodeId, NodeDeploymentStatus Status, string? FailureReason);
/// <summary>
/// Stages a status change WITHOUT materialising the row: the row carries the whole <c>ArtifactBlob</c>
/// (varbinary(max), multi-MB on a real fleet), which no status transition reads (fix round 1). A stub
@@ -530,9 +959,10 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers
/// stub stays tracked so <see cref="SetFailureReason"/> can add to the same UPDATE.
/// </summary>
private static void UpdateDeploymentStatus(
OtOpcUaConfigDbContext db, DeploymentId id, DeploymentStatus status, bool sealNow = false, byte[]? rowVersion = null)
OtOpcUaConfigDbContext db, DeploymentId id, DeploymentStatus status, bool sealNow = false, byte[]? rowVersion = null,
bool onlyIfInFlight = false)
{
var entry = TrackStub(db, id, rowVersion);
var entry = TrackStub(db, id, rowVersion, onlyIfInFlight);
if (entry is null) return;
entry.Entity.Status = status;
entry.Property(x => x.Status).IsModified = true;
@@ -547,7 +977,7 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers
private void SetFailureReason(OtOpcUaConfigDbContext db, string reason)
{
if (_current is null) return;
var entry = TrackStub(db, _current.Value);
var entry = TrackStub(db, _current.Value, onlyIfInFlight: true);
if (entry is null) return;
entry.Entity.FailureReason = reason;
entry.Property(x => x.FailureReason).IsModified = true;
@@ -561,7 +991,7 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers
/// guard); <see langword="null"/> reads the current one.
/// </param>
private static Microsoft.EntityFrameworkCore.ChangeTracking.EntityEntry<Deployment>? TrackStub(
OtOpcUaConfigDbContext db, DeploymentId id, byte[]? rowVersion = null)
OtOpcUaConfigDbContext db, DeploymentId id, byte[]? rowVersion = null, bool onlyIfInFlight = false)
{
var tracked = db.ChangeTracker.Entries<Deployment>().FirstOrDefault(e => e.Entity.DeploymentId == id.Value);
if (tracked is not null) return tracked;
@@ -570,9 +1000,13 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers
var stamp = db.Deployments
.AsNoTracking()
.Where(x => x.DeploymentId == id.Value)
.Select(x => new { x.RowVersion })
.Select(x => new { x.RowVersion, x.Status })
.FirstOrDefault();
if (stamp is null) return null;
// Review L8: a terminal outcome is written once. A row something else already closed (AdminOps' lost-
// dispatch closure, a previous incarnation) is left as it is; the RowVersion check covers the race.
if (onlyIfInFlight && stamp.Status is not (DeploymentStatus.Dispatching or DeploymentStatus.AwaitingApplyAcks))
return null;
rowVersion = stamp.RowVersion;
}
return db.Deployments.Attach(new Deployment
@@ -0,0 +1,65 @@
namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Coordinators;
/// <summary>Where the <see cref="ConfigPublishCoordinator"/> singleton stands on this node (lmxopcua#868).</summary>
public enum ConfigPublishCoordinatorState
{
/// <summary>The singleton is not running on this node (the partner hosts it, or it has not started yet).</summary>
NotHosted,
/// <summary>Running here, but its recovery state has not loaded from the ConfigDb yet; deployments are queued.</summary>
Loading,
/// <summary>Running here with its recovery state loaded; deployments are dispatched.</summary>
Ready,
}
/// <summary>One immutable observation of the coordinator's state.</summary>
/// <param name="State">The state.</param>
/// <param name="FailedLoadAttempts">Consecutive failed recovery loads (0 unless <see cref="ConfigPublishCoordinatorState.Loading"/>).</param>
/// <param name="LastError">The last load failure's message, or null.</param>
/// <param name="SinceUtc">When the state was entered.</param>
/// <param name="PendingAckPersists">Received acks the coordinator could not persist yet and is retrying.</param>
/// <param name="LastPersistError">The last ack-persist failure's message, or null.</param>
public sealed record ConfigPublishCoordinatorSnapshot(
ConfigPublishCoordinatorState State, int FailedLoadAttempts, string? LastError, DateTime SinceUtc,
int PendingAckPersists = 0, string? LastPersistError = null);
/// <summary>
/// Thread-safe state bridge from the <see cref="ConfigPublishCoordinator"/> actor to the node's health endpoint
/// (one DI singleton per admin node). The actor writes; the health check reads.
/// </summary>
public sealed class ConfigPublishCoordinatorReadiness
{
private volatile ConfigPublishCoordinatorSnapshot _current =
new(ConfigPublishCoordinatorState.NotHosted, 0, null, DateTime.UtcNow);
/// <summary>The latest observation.</summary>
public ConfigPublishCoordinatorSnapshot Current => _current;
/// <summary>The coordinator is loading (or retrying the load of) its recovery state.</summary>
/// <param name="failedAttempts">Failed attempts so far.</param>
/// <param name="lastError">The last failure's message, or null.</param>
public void MarkLoading(int failedAttempts, string? lastError)
{
var since = _current.State == ConfigPublishCoordinatorState.Loading ? _current.SinceUtc : DateTime.UtcNow;
_current = _current with
{
State = ConfigPublishCoordinatorState.Loading, FailedLoadAttempts = failedAttempts, LastError = lastError, SinceUtc = since,
};
}
/// <summary>The ack-persist backlog changed (review L6): acks received but not yet written, and why.</summary>
/// <param name="pending">Acks awaiting a persist retry.</param>
/// <param name="lastError">The last persist failure's message, or null.</param>
public void MarkAckPersistBacklog(int pending, string? lastError) =>
_current = _current with { PendingAckPersists = pending, LastPersistError = pending > 0 ? lastError : null };
/// <summary>The recovery state is loaded.</summary>
public void MarkReady() => _current = _current with
{
State = ConfigPublishCoordinatorState.Ready, FailedLoadAttempts = 0, LastError = null, SinceUtc = DateTime.UtcNow,
};
/// <summary>The singleton stopped on this node.</summary>
public void MarkNotHosted() => _current = new(ConfigPublishCoordinatorState.NotHosted, 0, null, DateTime.UtcNow);
}
@@ -112,7 +112,11 @@ public static class ServiceCollectionExtensions
// coordinator. Deliberately NOT an ActorSelection.ResolveOne here: that blocks
// inside a props factory and races the spawn it is waiting for.
var meshRouter = registry.Get<CentralCommunicationActorKey>();
return ConfigPublishCoordinator.Props(dbFactory, applyDeadline: DeployTimings.ApplyDeadline, meshRouter: meshRouter);
// lmxopcua#868: optional — through IServiceProvider for GetService's null-for-absent contract
// (the Akka resolver makes no such promise), so harnesses that never register it still boot.
var readiness = resolver.GetService<IServiceProvider>()?.GetService<ConfigPublishCoordinatorReadiness>();
return ConfigPublishCoordinator.Props(
dbFactory, applyDeadline: DeployTimings.ApplyDeadline, meshRouter: meshRouter, readiness: readiness);
},
singletonOptions);
@@ -0,0 +1,51 @@
using Microsoft.Extensions.Diagnostics.HealthChecks;
using ZB.MOM.WW.OtOpcUa.ControlPlane.Coordinators;
namespace ZB.MOM.WW.OtOpcUa.Host.Health;
/// <summary>
/// The deploy coordinator's readiness (lmxopcua#868): <b>Degraded</b> while the
/// <see cref="ConfigPublishCoordinator"/> singleton on this node has not yet loaded its recovery state from the
/// ConfigDb (deployments are queued, not dispatched), Healthy otherwise — including on the admin node that does
/// not host the singleton. Never Unhealthy: the node still serves (the configdb probe owns "DB down").
/// </summary>
/// <param name="readiness">The coordinator's state bridge; null on graphs that never register one.</param>
public sealed class ConfigPublishCoordinatorHealthCheck(ConfigPublishCoordinatorReadiness? readiness) : IHealthCheck
{
/// <summary>The check's registration name.</summary>
public const string Name = "config-publish";
/// <inheritdoc />
public Task<HealthCheckResult> CheckHealthAsync(HealthCheckContext context, CancellationToken cancellationToken = default)
{
if (readiness is null)
return Task.FromResult(HealthCheckResult.Healthy("Deploy coordinator readiness is not tracked on this node."));
var s = readiness.Current;
var data = new Dictionary<string, object>(StringComparer.Ordinal)
{
["state"] = s.State.ToString(),
["sinceUtc"] = s.SinceUtc.ToString("O"),
["failedLoadAttempts"] = s.FailedLoadAttempts,
};
if (s.LastError is not null) data["lastError"] = s.LastError;
data["pendingAckPersists"] = s.PendingAckPersists;
if (s.LastPersistError is not null) data["lastPersistError"] = s.LastPersistError;
return Task.FromResult(s.State switch
{
ConfigPublishCoordinatorState.Ready when s.PendingAckPersists > 0 => HealthCheckResult.Degraded(
$"Deploy coordinator cannot persist {s.PendingAckPersists} received deploy ack(s) to the ConfigDb " +
$"(last: {s.LastPersistError}); retrying with backoff. The deployment's outcome waits for them.",
data: data),
ConfigPublishCoordinatorState.Loading => HealthCheckResult.Degraded(
s.FailedLoadAttempts == 0
? "Deploy coordinator is loading its recovery state from the ConfigDb; deployments are queued until it completes."
: $"Deploy coordinator cannot load its recovery state from the ConfigDb ({s.FailedLoadAttempts} failed attempt(s), " +
$"last: {s.LastError}); retrying with backoff. Deployments are queued until it succeeds.",
data: data),
ConfigPublishCoordinatorState.Ready => HealthCheckResult.Healthy("Deploy coordinator ready.", data),
_ => HealthCheckResult.Healthy("Deploy coordinator singleton is not hosted on this node.", data),
});
}
}
@@ -87,6 +87,15 @@ public static class HealthEndpoints
{
ProbeQuery = static (db, ct) => db.Deployments.AsNoTracking().Take(1).ToListAsync(ct),
});
// lmxopcua#868: Degraded (still 200) while the deploy coordinator singleton on this node is retrying
// its ConfigDb recovery load and queueing deployments; Healthy otherwise. Ready tier only.
checks.Add(new HealthCheckRegistration(
ConfigPublishCoordinatorHealthCheck.Name,
sp => new ConfigPublishCoordinatorHealthCheck(
sp.GetService<ZB.MOM.WW.OtOpcUa.ControlPlane.Coordinators.ConfigPublishCoordinatorReadiness>()),
failureStatus: null,
tags: new[] { ZbHealthTags.Ready }));
}
checks
@@ -176,7 +176,12 @@ else
builder.Services.Configure<AclOptions>(builder.Configuration.GetSection(AclOptions.SectionName));
if (hasAdmin)
{
builder.Services.AddOtOpcUaConfigDb(builder.Configuration);
// lmxopcua#868: the deploy coordinator singleton reports its ConfigDb recovery-load state here; the
// config-publish health check (AddOtOpcUaHealth) reads it — Degraded while deployments are queued.
builder.Services.AddSingleton<ZB.MOM.WW.OtOpcUa.ControlPlane.Coordinators.ConfigPublishCoordinatorReadiness>();
}
else
Log.Information("Driver-only node — no ConfigDb registered; configuration via ConfigSource:Mode=FetchAndCache");
@@ -6,10 +6,20 @@ namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.Compatibility.Aggregates;
/// <summary>Adapts the versioned baseline algorithms to the SDK 2 calculator protocol without a second SDK runtime.</summary>
internal static class CompatibilityAggregateFactory
{
/// <remarks>
/// PercentGood / DurationGood are served by the application-owned <see cref="GoodOnlyStatusAggregateCalculator"/>
/// (#809: Uncertain time is not Good, as in Part 13 and HistoryRead), TimeAverage / Total by
/// <see cref="CoveredTimeAverageAggregateCalculator"/> (#821: no start bound ⇒ only the covered part of the interval,
/// even when the first value is Uncertain); every other id is the unchanged baseline.
/// </remarks>
internal static IAggregateCalculator Create(NodeId id, DateTimeUtc start, DateTimeUtc end, double interval,
bool stepped, AggregateConfiguration configuration, ITelemetryContext telemetry) =>
new Adapter(Aggregators.CreateStandardCalculator(id, (DateTime)start, (DateTime)end, interval, stepped, configuration, telemetry)
?? throw new ArgumentOutOfRangeException(nameof(id), "Unsupported baseline aggregate."));
new Adapter(GoodOnlyStatusAggregateCalculator.Overrides(id)
? new GoodOnlyStatusAggregateCalculator(id, (DateTime)start, (DateTime)end, interval, stepped, configuration, telemetry)
: CoveredTimeAverageAggregateCalculator.Overrides(id)
? new CoveredTimeAverageAggregateCalculator(id, (DateTime)start, (DateTime)end, interval, stepped, configuration, telemetry)
: Aggregators.CreateStandardCalculator(id, (DateTime)start, (DateTime)end, interval, stepped, configuration, telemetry)
?? throw new ArgumentOutOfRangeException(nameof(id), "Unsupported baseline aggregate."));
/// <summary>Replaces exactly the existing standard factories; no aggregate IDs are added or removed.</summary>
internal static async ValueTask RegisterAsync(AggregateManager manager, CancellationToken ct)
@@ -0,0 +1,115 @@
using Opc.Ua;
using DataValue = ZB.MOM.WW.OtOpcUa.OpcUaServer.Compatibility.Aggregates.LegacyDataValue;
namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.Compatibility.Aggregates;
/// <summary>
/// Application-owned override (#821) of the baseline <see cref="AverageAggregateCalculator"/> for <c>TimeAverage</c>
/// and <c>Total</c> only: the time before the first usable value (no start bound) is never integrated, whatever the
/// quality of that first value. Part 13 §5.4.3.6/§5.4.3.8 "No Start Bound: Calculate Partial Interval" — the rule the
/// HistoryRead evaluator (<c>TimeWeightedAggregator</c>, SqlServer + QuestDB) already follows, and the one the baseline
/// itself follows when the first value is Good.
/// </summary>
/// <remarks>
/// <para>
/// With no usable value before the interval, the baseline's interpolated start bound is <c>BadNoData</c> (value 0),
/// so the first region <c>[start, first value)</c> is Bad and excluded. But <c>GetRegionsInValueSet</c>
/// (ignoreBadData, sloped) marks the region before an Uncertain value <c>UncertainDataSubNormal</c> without the
/// not-Bad guard every other branch has — so an <b>Uncertain</b> first value turned the Bad region non-Bad and the
/// uncovered span was integrated as a ramp from 0 (QP: Total 188.96 for Part 13's 163.96, TimeAverage 9.448 for
/// 10.931). SDK 2.0.1's stock calculator does the same.
/// </para>
/// <para>
/// This class leaves the copied file untouched and restores exactly that one overwritten status: a region that
/// starts at a Bad point (only the interpolated start bound can — the region machinery skips Bad raw values)
/// keeps that point's Bad status. Every other region, the values, the result status (Uncertain either way: the
/// region is non-Good), timestamp, value type and Calculated / Partial bits are the baseline's.
/// <c>TimeAverage2</c> / <c>Total2</c> (simple bounds, no ignoreBadData) and stepped variables never reach that
/// branch and fall through to the baseline unchanged.
/// </para>
/// </remarks>
internal sealed class CoveredTimeAverageAggregateCalculator : AverageAggregateCalculator
{
/// <summary>True for the two aggregate ids this class overrides.</summary>
internal static bool Overrides(NodeId aggregateId) =>
aggregateId == ObjectIds.AggregateFunction_TimeAverage || aggregateId == ObjectIds.AggregateFunction_Total;
/// <summary>Initializes the calculator; same contract as the baseline constructor.</summary>
public CoveredTimeAverageAggregateCalculator(
NodeId aggregateId,
DateTime startTime,
DateTime endTime,
double processingInterval,
bool stepped,
AggregateConfiguration configuration,
ITelemetryContext? telemetry)
: base(aggregateId, startTime, endTime, processingInterval, stepped, configuration, telemetry)
{
}
/// <inheritdoc />
protected override DataValue ComputeValue(TimeSlice slice)
{
if (AggregateId == ObjectIds.AggregateFunction_TimeAverage) return ComputeCovered(slice, total: false);
if (AggregateId == ObjectIds.AggregateFunction_Total) return ComputeCovered(slice, total: true);
return base.ComputeValue(slice);
}
/// <summary>
/// The baseline <c>ComputeTimeAverage(slice, useSimpleBounds: false, valueType)</c> with a region that starts at a
/// Bad point kept Bad.
/// </summary>
private DataValue ComputeCovered(TimeSlice slice, bool total)
{
List<DataValue>? values = GetValuesWithInterpolatedBounds(slice);
if (values == null || values.Count == 0)
{
return GetNoDataValue(slice);
}
List<SubRegion> regions = GetRegionsInValueSet(values, true, Stepped);
double area = 0;
double totalDuration = 0;
bool nonGoodRegionsExists = false;
foreach (var region in regions)
{
// #821: the one change — undo the baseline's Bad → Uncertain overwrite of a region starting at a Bad point.
if (StatusCode.IsBad(region.DataPoint.StatusCode) && StatusCode.IsNotBad(region.StatusCode))
{
region.StatusCode = region.DataPoint.StatusCode;
}
double duration = region.Duration / 1000.0;
if (StatusCode.IsNotBad(region.StatusCode))
{
area += (region.StartValue + region.EndValue) * duration / 2;
totalDuration += duration;
}
if (StatusCode.IsNotGood(region.StatusCode))
{
nonGoodRegionsExists = true;
}
}
if (totalDuration == 0)
{
return GetNoDataValue(slice);
}
double result = total ? area : area / totalDuration;
var value = new DataValue
{
WrappedValue = SdkValueBoundary.FromClr(result),
SourceTimestamp = GetTimestamp(slice),
ServerTimestamp = GetTimestamp(slice),
StatusCode = nonGoodRegionsExists ? StatusCodes.UncertainDataSubNormal : StatusCodes.Good
};
value.StatusCode = value.StatusCode.WithAggregateBits(AggregateBits.Calculated);
return value;
}
}
@@ -0,0 +1,83 @@
using Opc.Ua;
using DataValue = ZB.MOM.WW.OtOpcUa.OpcUaServer.Compatibility.Aggregates.LegacyDataValue;
namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.Compatibility.Aggregates;
/// <summary>
/// Application-owned override (#809) of the baseline <see cref="StatusAggregateCalculator"/> for
/// <c>PercentGood</c> and <c>DurationGood</c> only: a region counts toward the Good time only when its status is
/// Good. Uncertain time is in neither Good nor (unless <c>TreatUncertainAsBad</c>) Bad, per Part 13
/// §5.4.3.31/§5.4.3.33 (Mantis 0006166) — the rule the HistoryRead evaluators (<c>TimeWeightedAggregator</c>,
/// SqlServer + QuestDB) already follow.
/// </summary>
/// <remarks>
/// The baseline 1.5.378.182 <c>ComputeDurationGoodBad</c> counts Uncertain regions as Good when
/// <c>TreatUncertainAsBad</c> = false. This class leaves the copied file untouched and replaces only the Good
/// branch: the slice values, simple bounds, stepped regions, empty-slice <c>BadNoData</c>, timestamp, value type
/// and Calculated status bit are the baseline's. <c>DurationBad</c> / <c>PercentBad</c> / <c>WorstQuality*</c>
/// fall through to the baseline unchanged.
/// </remarks>
internal sealed class GoodOnlyStatusAggregateCalculator : StatusAggregateCalculator
{
/// <summary>True for the two aggregate ids this class overrides.</summary>
internal static bool Overrides(NodeId aggregateId) =>
aggregateId == ObjectIds.AggregateFunction_PercentGood || aggregateId == ObjectIds.AggregateFunction_DurationGood;
/// <summary>Initializes the calculator; same contract as the baseline constructor.</summary>
public GoodOnlyStatusAggregateCalculator(
NodeId aggregateId,
DateTime startTime,
DateTime endTime,
double processingInterval,
bool stepped,
AggregateConfiguration configuration,
ITelemetryContext? telemetry)
: base(aggregateId, startTime, endTime, processingInterval, stepped, configuration, telemetry)
{
}
/// <inheritdoc />
protected override DataValue ComputeValue(TimeSlice slice)
{
if (AggregateId == ObjectIds.AggregateFunction_DurationGood) return ComputeGoodOnly(slice, usePercent: false);
if (AggregateId == ObjectIds.AggregateFunction_PercentGood) return ComputeGoodOnly(slice, usePercent: true);
return base.ComputeValue(slice);
}
/// <summary>The baseline <c>ComputeDurationGoodBad(slice, isBad: false, usePercent)</c> with Good-only counting.</summary>
private DataValue ComputeGoodOnly(TimeSlice slice, bool usePercent)
{
List<DataValue>? values = GetValuesWithSimpleBounds(slice);
if (values == null || values.Count == 0)
{
return GetNoDataValue(slice);
}
List<SubRegion> regions = GetRegionsInValueSet(values, false, true);
double duration = 0;
double total = 0;
foreach (var region in regions)
{
total += region.Duration;
if (StatusCode.IsGood(region.StatusCode))
{
duration += region.Duration;
}
}
if (usePercent)
{
duration = duration / total * 100;
}
var value = new DataValue
{
WrappedValue = SdkValueBoundary.FromClr(duration),
SourceTimestamp = GetTimestamp(slice),
ServerTimestamp = GetTimestamp(slice)
};
value.StatusCode = value.StatusCode.WithAggregateBits(AggregateBits.Calculated);
return value;
}
}
@@ -17,3 +17,23 @@ The isolated `scripts/validation/opcua-aggregate-baseline` process pins only SDK
Run the generator with a **new** output path, then compare it with `tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/Fixtures/opcua-1.5.378.182-aggregates.json`. It refuses to overwrite evidence. Never regenerate expectations from the compatibility implementation. Existing historian parity tests retain their application-specific documented divergences from the baseline SDK, while new compatibility tests check the complete old-calculator output.
This preserves older behavior deliberately; it does not claim that all older SDK algorithms implement the newest OPC UA interpretation. A future semantic migration needs its own consumer-impact decision and versioned tests.
## Recorded semantic decisions
### #809 — PercentGood / DurationGood count Good time only (owner-approved 2026-11-11)
The baseline `StatusAggregateCalculator.ComputeDurationGoodBad` counts Uncertain regions as **Good** when `TreatUncertainAsBad = false` (1.5.378.106 and SDK 2.0.1 count Good only). Part 13 §5.4.3.31/§5.4.3.33 (Mantis 0006166) and the HistoryRead evaluators (`TimeWeightedAggregator`, SqlServer + QuestDB) count Good regions only, so live AggregateFilter results disagreed with HistoryRead whenever Uncertain data was present. ScadaBridge uses neither AggregateFilter nor these aggregates.
- The copied upstream files above are **unchanged** (their SHA-256s still match `baseline-sources.json`).
- `GoodOnlyStatusAggregateCalculator.cs` (application code, not a baseline copy) subclasses the baseline `StatusAggregateCalculator` and replaces only the PercentGood / DurationGood computation: the same slice values, simple bounds, stepped regions, empty-slice `BadNoData`, timestamps, value type and Calculated bit, with a region counted as Good only when its status is Good — whatever `TreatUncertainAsBad` says (when true, Uncertain is Bad, so it was never Good). `CompatibilityAggregateFactory.Create` routes exactly those two IDs to it; the other 35 IDs, including DurationBad / PercentBad / WorstQuality, are the baseline unchanged.
- Golden test: the 64 vectors this changes (38 PercentGood + 26 DurationGood, all with an effective `TreatUncertainAsBad = false` over data with Uncertain regions) are exempted **exactly** — the PercentGood / DurationGood vectors whose baseline output differs from the same case's baseline vector with `TreatUncertainAsBad = true`, where 1.5.378.182 already counted Good regions only — and each is pinned to that twin's output. The pin is therefore baseline evidence, never regenerated from the application's calculator; any further drift in any of the 6,144 cases fails. `Sdk2AggregateCompatibilityTests.Issue809_*` also pins a hand-computed case.
- Live-vs-history parity: `SdkAggregateParityTests.Live_AggregateFilter_PercentGood_and_DurationGood_equal_HistoryRead_over_Uncertain_samples` compares the registered calculator with `TimeWeightedAggregator` over Good / Uncertain / Bad mixes with no masks, and checks that the data discriminates against the pre-#809 baseline.
### #821 — TimeAverage / Total integrate only the covered part when there is no start bound (owner-approved 2026-11-11)
When the data starts inside an interval (no usable value before it) the baseline's interpolated start bound is `BadNoData`, so the region before the first value is Bad and left out — Part 13 §5.4.3.6/§5.4.3.8 "No Start Bound: Calculate Partial Interval", and what the HistoryRead evaluator (`TimeWeightedAggregator`, SqlServer + QuestDB) does. But when that first value is **Uncertain**, `AggregateCalculator.GetRegionsInValueSet` (ignoreBadData, sloped) overwrites the region's Bad status with `UncertainDataSubNormal` (the one branch without the not-Bad guard), so `AverageAggregateCalculator` integrated the uncovered span as a ramp from 0 and divided TimeAverage by the whole interval (QP: Total 188.96 for 163.96, TimeAverage 9.448 for 10.931). SDK 2.0.1's stock calculator does the same. ScadaBridge uses neither AggregateFilter nor these aggregates.
- The copied upstream files above are **unchanged** (their SHA-256s still match `baseline-sources.json`).
- `CoveredTimeAverageAggregateCalculator.cs` (application code) subclasses the baseline `AverageAggregateCalculator` and recomputes only TimeAverage / Total: the baseline's interpolated-bounds values and regions, with a region that starts at a Bad point (only the start bound can) kept Bad. Values change only for such an interval; status (Uncertain either way, except `BadNoData` when nothing at all is covered), timestamps, value type and Calculated / Partial bits are the baseline's. Stepped variables and `TreatUncertainAsBad = true` never reached the defect and are unchanged. `CompatibilityAggregateFactory.Create` routes exactly those two IDs to it; TimeAverage2 / Total2 and the other 31 IDs are the baseline unchanged.
- Golden test: **no** fixture vector changes — every fixture data set has a usable value at −10 s, before every window, so no fixture interval lacks a start bound; all 384 TimeAverage / Total vectors stay held to their own output (`Issue821FixtureExemptions = 0`). The change is pinned over late-start variants of every data set (cut at +5 s and +25 s, first remaining non-Bad sample made Uncertain, every direction / stepping / configuration / incremental mode): 150 and 142 of the 384 change, each held to a baseline-derived oracle — the untouched baseline's value on the same variant with that sample Good, its status / timestamps / bits on the variant itself. `Sdk2AggregateCompatibilityTests.Issue821_*` also pins a hand-computed case.
- Live-vs-history parity: `SdkAggregateParityTests.Live_AggregateFilter_TimeAverage_and_Total_equal_HistoryRead_over_an_Uncertain_first_late_start` compares the registered calculator with `TimeWeightedAggregator`, forward and reverse, with no masks, and checks that the data discriminates against the pre-#821 baseline; `QualityMixConformance` now includes TimeAverage / Total over QP on both SQL backends.
@@ -1476,7 +1476,14 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
/// synchronous: the walk completes inline while every channel has room.
/// </para>
/// <para>
/// <b>Deferral.</b> When a channel is full the walk continues asynchronously (holding no thread or lock),
/// <b>Since Gitea #855 no notifier of ours is a channel.</b> At a notifier this manager owns, the walk serves
/// the items itself (<see cref="DeliverToNotifierItems"/>) instead of writing to the SDK's
/// <c>MonitoredNode2</c> channel, so it always completes inline and the deferral below is reached only through
/// a hop that is not ours (none exists today) — or the test stall (<see cref="NotifierDeliveryStall"/>). It is
/// kept as the safety net, and the refresh cut (#854) covers folder items too.
/// </para>
/// <para>
/// <b>Deferral.</b> When a hop suspends the walk continues asynchronously (holding no thread or lock),
/// and the condition gets a queue: its later events wait behind it, so ONE condition's events stay in
/// order, while other conditions whose channels have room are not held up (per-condition, not global).
/// Events of different conditions may interleave differently than reported — each carries its own
@@ -1522,13 +1529,29 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
if (delivery.IsCompletedSuccessfully) return;
// #855 review: a delivery that FAILED inline is over, not suspended. Booking it as a deferral used to open and close
// a deferral episode synchronously (moving the episode a concurrent plain refresh checks, and clearing cuts under
// the event-report gate while Lock was held). Only an incomplete delivery is deferred.
if (delivery.IsCompleted)
{
_logger.LogError(delivery.AsTask().Exception?.GetBaseException(),
"OtOpcUaNodeManager: failed to report Part 9 condition event for {NodeId}", condition.NodeId);
return;
}
DeferredConditionReports queue;
lock (_eventReportGate)
{
var queue = new DeferredConditionReports();
queue = new DeferredConditionReports();
_deferredReports[condition] = queue;
CountDeferred(+1, condition);
_ = DrainDeferredAsync(condition, queue, delivery.AsTask());
}
// #855 review 2: started on the pool, never inline. The suspended delivery may complete between the check above and
// here; an inline drain would then run (and end the episode, and clear cuts over every item gate) on this thread,
// under Lock. The queue is registered first, so this condition's later reports wait behind it either way.
var inFlight = delivery.AsTask();
_ = Task.Run(() => DrainDeferredAsync(condition, queue, inFlight));
}
/// <summary>Queues a later event behind a condition's deferred one, coalescing past the budget. Under the gate.</summary>
@@ -1615,7 +1638,42 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
node.OnReportEvent?.Invoke(SystemContext, node, e);
if (node.OnReportEventAsync is { } handler)
await handler(SystemContext, node, e, CancellationToken.None).ConfigureAwait(false);
{
if (OwnNotifierFanOut(node, handler) is { } monitored)
{
// Gitea #855: one of our notifiers. Its items are served here, per item and per event, exactly as the
// Server-object items are (OnConditionReportEvent) — never through the SDK's channel, whose consumer
// needed Lock (#838) and gave no per-event, per-item point at which a refresh cut could apply.
if (NotifierDeliveryStall?.Invoke(node) is { IsCompleted: false } stall)
{
MarkDeferred(e);
await stall.ConfigureAwait(false);
}
// Re-checked: a stall may have outlived the condition (the liveness check above ran before it).
if (IsLiveCondition(condition))
{
// #855 review 3: one hop's failure must not cost the hops after it (before #855 this hop was a channel
// write, which cannot throw; the SDK isolated failures per item).
try
{
DeliverToNotifierItems(monitored, e);
}
catch (Exception ex)
{
_logger.LogWarning(ex, "OtOpcUaNodeManager: failed to deliver a condition event at notifier {NodeId}", node.NodeId);
}
}
}
else
{
// Not a fan-out of ours (none exists today): the SDK path, which may suspend. Whatever is delivered after
// the suspension runs off Lock, so it must be judged as a deferred event (#855).
var pending = handler(SystemContext, node, e, CancellationToken.None);
if (!pending.IsCompleted) MarkDeferred(e);
await pending.ConfigureAwait(false);
}
}
var notifiers = new List<NodeState.Notifier>();
node.GetNotifiers(SystemContext, notifiers);
@@ -1631,6 +1689,101 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
await DeliverConditionEventAsync(condition, parent, e).ConfigureAwait(false);
}
/// <summary>Marks an event whose remaining delivery will run off <c>Lock</c> (#855) — before the walk suspends.</summary>
private static void MarkDeferred(IFilterTarget e)
{
if (e is ConditionEventSnapshot snapshot) snapshot.Deferred = true;
}
/// <summary>
/// The SDK <see cref="MonitoredNode2"/> serving event items on <paramref name="node"/>, when that node is one of
/// ours and <paramref name="handler"/> is the fan-out's own <c>OnReportEventAsync</c> (what
/// <c>MonitoredNode2.Add(IEventMonitoredItem)</c> installs); otherwise null (Gitea #855).
/// </summary>
private MonitoredNode2? OwnNotifierFanOut(NodeState node, NodeStateReportEventAsyncHandler handler) =>
handler.Target is MonitoredNode2 monitored && ReferenceEquals(monitored.Node, node) && OwnsNamespace(node.NodeId)
? monitored
: null;
/// <summary>
/// Test probe (#855): consulted before a condition event is delivered to the items of one of our notifiers; an
/// incomplete task is awaited first, so the walk suspends there exactly as it would behind a slow notifier — the
/// way tests produce a deferral now that no notifier of ours can block. Null in production.
/// </summary>
internal Func<NodeState, Task?>? NotifierDeliveryStall { get; set; }
/// <summary>
/// Delivers one condition event to the event items attached to one of this manager's notifiers (an equipment or
/// raw device folder), item by item (Gitea #855) — the folder twin of <see cref="OnConditionReportEvent"/>.
/// </summary>
/// <param name="monitored">The notifier's SDK fan-out; only its attached-item list is used.</param>
/// <param name="e">The condition's snapshot.</param>
/// <remarks>
/// <para>
/// Mirrors the SDK's <c>MonitoredNode2.ProcessEventForItemAsync</c> — its audit guard, then the verdict through
/// <see cref="INodeManager3.ValidateEventRolePermissions"/> (the SDK's type + SourceNode <c>ReceiveEvents</c>
/// check and, under enforcement, #826's live-notifier + grant check), then <c>QueueEvent</c> — but per event:
/// nothing is memoised, so a deploy, revocation or reference removal decides the very next event. The item's
/// gate and refresh cut apply exactly as on the Server object (<see cref="QueueConditionEventToItem"/>).
/// </para>
/// <para>
/// Synchronous. On a report it runs on the reporter's thread under <c>Lock</c> (as the Server-object delivery
/// does), and <c>QueueEvent</c> never waits for a consumer — the item's own queue discards oldest at its
/// <c>QueueSize</c> — so no notifier of ours defers any more. Off <c>Lock</c> (a deferred event) the verdict,
/// which takes <c>Lock</c>, is computed before the item gate is taken: nothing under a gate takes <c>Lock</c>.
/// A stood-down item (#562) is not in the attached list and receives nothing, as before.
/// </para>
/// </remarks>
private void DeliverToNotifierItems(MonitoredNode2 monitored, IFilterTarget e)
{
var deferred = e is ConditionEventSnapshot { Deferred: true } d ? d : null;
// #855 review 2: the #826 decision's item-independent part — the live condition, whether this notifier is still a
// legitimate notifier of it, and its visibility scopes — is resolved ONCE per hop, not per item (it is what
// IsNotifierEventVisibleTo resolves for an item attached here). Only the grant check and the SDK's role check
// remain per item.
var enforced = AclEnforced;
List<NodeId>? scopes = null;
if (enforced)
{
if (EventConditionOf(e) is not { } named
|| !_alarmConditions.TryGetValue(MapKey(named.NodeId), out var live)
|| !IsLiveNotifierOf(live, monitored.Node))
{
return; // fail closed for every item on this notifier
}
scopes = ConditionVisibilityScopes(live);
}
foreach (var item in monitored.EventMonitoredItems.Values)
{
if (item is null) continue;
try
{
if (IsAuditEvent(e) && !IsAuditDeliverableTo(item)) continue;
if (ServiceResult.IsBad(ValidateEventRolePermissions(item, e))) continue;
if (scopes is not null && !IsConditionVisibleTo(item, scopes)) continue;
QueueConditionEventToItem(item, e, deferred, DeferredFolderDeliveryProbe);
}
catch (Exception ex)
{
// As the SDK's per-item delivery: one item's failure must not cost the others (or the rest of the walk).
_logger.LogWarning(ex, "OtOpcUaNodeManager: failed to deliver a condition event to event item {ItemId}", item.Id);
}
}
}
/// <summary>The SDK's audit test from <c>MonitoredNode2.ProcessEventForItemAsync</c>: the event, or the snapshot's handle.</summary>
private static bool IsAuditEvent(IFilterTarget e) =>
e is AuditEventState || e is InstanceStateSnapshot { Handle: AuditEventState };
/// <summary>
/// Test probe (#855): the folder twin of <see cref="DeferredServerObjectDeliveryProbe"/> — runs between a deferred
/// event's supersession check and its queueing to a folder item, inside the item's gate. Null in production.
/// </summary>
internal Action? DeferredFolderDeliveryProbe { get; set; }
/// <summary>Whether <paramref name="condition"/> is the instance the address space currently holds for its NodeId.</summary>
private bool IsLiveCondition(NodeState condition) =>
_alarmConditions.TryGetValue(MapKey(condition.NodeId), out var live) && ReferenceEquals(live, condition);
@@ -1653,9 +1806,9 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
{
_lastDeferralWarningUtc = now;
_logger.LogWarning(
"OtOpcUaNodeManager: condition events are being deferred — an event notifier's delivery channel is "
+ "full (first: {NodeId}). Folder event subscribers are not keeping up, or Lock is held across a large "
+ "burst; events stay ordered per condition, and past {PerCondition} waiting per condition the oldest "
"OtOpcUaNodeManager: condition events are being deferred — a notifier hop the node manager does not "
+ "serve itself suspended (first: {NodeId}); since #855 no folder of ours does, so this is unexpected. "
+ "Events stay ordered per condition, and past {PerCondition} waiting per condition the oldest "
+ "give way to the newest (Gitea #838)",
condition.NodeId, MaxDeferredEventsPerCondition);
}
@@ -1711,18 +1864,27 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
public bool Retain { get; } = retain;
private volatile bool _deferred;
/// <summary>
/// Set (under the event-report gate, before the drain can dequeue it) when the event waits in a deferral queue:
/// only such an event reaches Server-object items off <c>Lock</c>, which is the only way a refresh can be overtaken.
/// Set when the event's delivery continues off <c>Lock</c>: under the event-report gate, before the drain can
/// dequeue it, when it waits in a deferral queue; or by the walk itself, before it suspends at a hop (#855), for
/// the in-flight event whose remaining hops then run on another thread. Only such an event reaches an item off
/// <c>Lock</c>, which is the only way a refresh can be overtaken.
/// </summary>
public bool Deferred { get; set; }
public bool Deferred
{
get => _deferred;
set => _deferred = value;
}
}
/// <summary>
/// Per Server-object item state (#854): the gate that orders a deferred delivery to the item against a refresh of
/// the item, and the cut of the item's latest refresh while deferred events may still be older than it.
/// Per event item state — Server-object (#854) and, since #855, folder items: the gate that orders a deferred
/// delivery to the item against a refresh of the item, and the cut of the item's latest refresh while deferred
/// events may still be older than it.
/// </summary>
private sealed class AllEventsItemState
private sealed class EventItemState
{
/// <summary>Held across "is this deferred event superseded? → QueueEvent", and across a refresh's
/// "QueueEvent the replay → install the cut". Only ever taken innermost (nothing but the item's own lock under it).</summary>
@@ -1749,21 +1911,21 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
private int _refreshGeneration;
/// <summary>
/// One <see cref="AllEventsItemState"/> per Server-object item for the item's whole lifetime (#854 review): keyed by
/// One <see cref="EventItemState"/> per event item for the item's whole lifetime (#854 review): keyed by
/// the item object itself, so an unsubscribe + resubscribe of the same item between a refresh and the drain keeps its
/// cut and its gate, and a drain holding the state always holds the one a refresh installs into.
/// </summary>
private readonly ConditionalWeakTable<IEventMonitoredItem, AllEventsItemState> _allEventsItemStates = new();
private readonly ConditionalWeakTable<IEventMonitoredItem, EventItemState> _eventItemStates = new();
/// <summary>Server-object items currently holding a refresh cut (test observation, #854 review).</summary>
/// <summary>Server-object and folder event items currently holding a refresh cut (test observation, #854 review).</summary>
internal int ActiveRefreshCuts
{
get
{
var n = 0;
foreach (var item in _allEventsItems.Keys)
foreach (var item in TrackedEventItems())
{
if (!_allEventsItemStates.TryGetValue(item, out var state)) continue;
if (!_eventItemStates.TryGetValue(item, out var state)) continue;
lock (state.Gate) n += state.Cut is null ? 0 : 1;
}
@@ -1771,8 +1933,15 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
}
}
private AllEventsItemState AllEventsItemStateOf(IEventMonitoredItem item) =>
_allEventsItemStates.GetValue(item, static _ => new AllEventsItemState());
private EventItemState EventItemStateOf(IEventMonitoredItem item) =>
_eventItemStates.GetValue(item, static _ => new EventItemState());
/// <summary>
/// Every live event item this manager tracks: the Server-object items and the direct (folder) items (#855). An
/// item that left both maps keeps its state in <see cref="_eventItemStates"/> until collected, but any cut it holds
/// is from an ended episode by the time it could deliver again, and so inert.
/// </summary>
private IEnumerable<IEventMonitoredItem> TrackedEventItems() => _allEventsItems.Keys.Concat(_eventItems.Keys);
/// <summary>
/// Advanced (under the event-report gate) each time the deferral count returns to zero: the end of a deferral
@@ -1780,6 +1949,9 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
/// </summary>
private long _deferralEpisode;
/// <summary>Deferral episodes ended since start (test observation, #855 review).</summary>
internal long DeferralEpisodes => Interlocked.Read(ref _deferralEpisode);
/// <summary>What a Server-object item's latest ConditionRefresh conveyed (#854).</summary>
/// <param name="Watermark">Every condition event with a sequence at or below this was reported before the replay was read.</param>
/// <param name="Episode">The deferral episode the cut was taken in; once it ends the cut is stale and inert.</param>
@@ -1810,9 +1982,9 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
private void ClearStaleRefreshCuts()
{
var current = Interlocked.Read(ref _deferralEpisode);
foreach (var item in _allEventsItems.Keys)
foreach (var item in TrackedEventItems())
{
if (!_allEventsItemStates.TryGetValue(item, out var state)) continue;
if (!_eventItemStates.TryGetValue(item, out var state)) continue;
lock (state.Gate)
{
if (state.Cut is { } cut && cut.Episode < current) state.Cut = null; // episodes only grow: never clear a newer cut (#854 re-review)
@@ -5655,54 +5827,87 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
// through its OLD equipment folder (#808).
if (!IsLiveCondition(node)) return;
var scopes = AclEnforced ? ConditionVisibilityScopes(node) : null;
List<NodeId>? scopes;
try
{
scopes = AclEnforced ? ConditionVisibilityScopes(node) : null;
}
catch (Exception ex)
{
// #855 review 3: withheld from the Server-object items (fail closed), but the walk goes on to the folders.
_logger.LogWarning(ex, "OtOpcUaNodeManager: failed to decide a condition event for Server-object items ({NodeId})", node.NodeId);
return;
}
var reportingSession = (context as ISessionSystemContext)?.SessionId;
var deferred = e is ConditionEventSnapshot { Deferred: true } d ? d : null;
foreach (var item in _allEventsItems.Keys)
{
if (e is AuditEventState && !IsAuditDeliverableTo(item)) continue;
if (scopes is not null && !IsConditionVisibleTo(item, scopes)) continue;
if (ServiceResult.IsBad(ValidateEventRolePermissions(item, e))) continue;
if (reportingSession is { IsNull: false } reportingId
&& item.Session?.Id is { } own
&& !own.Equals(reportingId))
try
{
continue;
}
// #854: a synchronous event (under Lock) needs the gate only while a refresh's replay is pending, and the
// pending marker is installed under Lock, so an item no refresh has touched has nothing to order. A deferred
// event always takes the gate — its state is created here if need be, so a refresh racing it shares it.
AllEventsItemState? state;
if (deferred is not null) state = AllEventsItemStateOf(item);
else if (!_allEventsItemStates.TryGetValue(item, out state) || state.PendingReplay == 0 || e is not ConditionEventSnapshot)
{
item.QueueEvent(e);
continue;
}
// A deferred event reaches this item off Lock, so a ConditionRefresh of the item may have replayed a newer
// state since it was reported; and while a refresh's frozen replay is being queued, an event newer than it must
// not be followed by it. Both are decided and queued under the item's gate, which the refresh also holds for
// each replayed event it queues — so a deferred event lands before the replay or is judged against the cut.
lock (state.Gate)
{
// A cut from an episode that has ended is stale (only waiting for ClearStaleRefreshCuts).
if (deferred is not null
&& state.Cut is { } cut && cut.Episode == Interlocked.Read(ref _deferralEpisode) && cut.Supersedes(deferred))
if (e is AuditEventState && !IsAuditDeliverableTo(item)) continue;
if (scopes is not null && !IsConditionVisibleTo(item, scopes)) continue;
if (ServiceResult.IsBad(ValidateEventRolePermissions(item, e))) continue;
if (reportingSession is { IsNull: false } reportingId
&& item.Session?.Id is { } own
&& !own.Equals(reportingId))
{
Interlocked.Increment(ref _refreshSupersededDeliveries);
OtOpcUaTelemetry.ConditionEventDeliveriesSuperseded.Add(1);
continue;
}
if (state.PendingReplay != 0 && e is ConditionEventSnapshot { } ce && ce.Sequence > state.PendingWatermark)
state.NewerSinceCut.Add(ce.ConditionId);
if (deferred is not null) DeferredServerObjectDeliveryProbe?.Invoke();
item.QueueEvent(e);
QueueConditionEventToItem(item, e, deferred, DeferredServerObjectDeliveryProbe);
}
catch (Exception ex)
{
// #855 review: one item's failure (an item disposed under us, the SDK's operation guard at shutdown) must
// not cost the other Server-object items, nor abort the walk before the folder hops.
_logger.LogWarning(ex, "OtOpcUaNodeManager: failed to deliver a condition event to Server-object item {ItemId}", item.Id);
}
}
}
/// <summary>
/// Queues a condition event to one event item — Server-object or folder (#855) — under the item's refresh gate and
/// cut (#854). The caller has already decided the item may receive it, and holds no item gate.
/// </summary>
/// <param name="item">The item.</param>
/// <param name="e">The event.</param>
/// <param name="deferred">The event, when its delivery runs off <c>Lock</c> (<see cref="ConditionEventSnapshot.Deferred"/>).</param>
/// <param name="probe">Test probe run inside the gate, before queueing a deferred event.</param>
private void QueueConditionEventToItem(IEventMonitoredItem item, IFilterTarget e, ConditionEventSnapshot? deferred, Action? probe)
{
// #854: a synchronous event (under Lock) needs the gate only while a refresh's replay is pending, and the
// pending marker is installed under Lock, so an item no refresh has touched has nothing to order. A deferred
// event always takes the gate — its state is created here if need be, so a refresh racing it shares it.
EventItemState? state;
if (deferred is not null) state = EventItemStateOf(item);
else if (!_eventItemStates.TryGetValue(item, out state) || state.PendingReplay == 0 || e is not ConditionEventSnapshot)
{
item.QueueEvent(e);
return;
}
// A deferred event reaches this item off Lock, so a ConditionRefresh of the item may have replayed a newer
// state since it was reported; and while a refresh's frozen replay is being queued, an event newer than it must
// not be followed by it. Both are decided and queued under the item's gate, which the refresh also holds for
// each replayed event it queues — so a deferred event lands before the replay or is judged against the cut.
lock (state.Gate)
{
// A cut from an episode that has ended is stale (only waiting for ClearStaleRefreshCuts).
if (deferred is not null
&& state.Cut is { } cut && cut.Episode == Interlocked.Read(ref _deferralEpisode) && cut.Supersedes(deferred))
{
Interlocked.Increment(ref _refreshSupersededDeliveries);
OtOpcUaTelemetry.ConditionEventDeliveriesSuperseded.Add(1);
return;
}
if (state.PendingReplay != 0 && e is ConditionEventSnapshot { } ce && ce.Sequence > state.PendingWatermark)
state.NewerSinceCut.Add(ce.ConditionId);
if (deferred is not null) probe?.Invoke();
item.QueueEvent(e);
}
}
@@ -6080,8 +6285,11 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
/// would be.
/// </para>
/// <para>
/// Runs on the SDK's per-node delivery task, where the inherited role-permission check already takes
/// <c>Lock</c> (via <c>GetNodeMetadata</c>); the scope resolution here takes it the same way.
/// Runs per event, uncached, from the per-item folder delivery (#855) — on the reporter's thread under
/// <c>Lock</c>, or off it for a deferred event (before the item gate is taken) — and, for the events that still
/// travel the SDK channel (a condition's method audits), on the SDK's per-node delivery task, where the
/// inherited role-permission check already takes <c>Lock</c> (via <c>GetNodeMetadata</c>); the scope
/// resolution here takes it the same way.
/// </para>
/// </remarks>
private bool IsNotifierEventVisibleTo(IEventMonitoredItem item, IFilterTarget e)
@@ -6146,13 +6354,15 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
/// <returns>Good to deliver; otherwise the veto.</returns>
/// <remarks>
/// <para>
/// The SDK's <c>MonitoredNode2</c> asks this, per event and per item, before it queues an event to an item
/// attached to one of this manager's notifiers. The Server-object path (#586) and ConditionRefresh decide
/// the same question themselves, so a folder's subscribers stop receiving a moved alarm (or a native alarm
/// whose reference was dropped) the moment the generation says so, not when the rebuild lands.
/// Condition events to an item attached to one of this manager's notifiers are decided here per event and per
/// item by <see cref="DeliverToNotifierItems"/> (#855, uncached); the SDK's <c>MonitoredNode2</c> still asks it
/// for what travels its channel (a condition's method audit events). The Server-object path (#586) and
/// ConditionRefresh decide the same question themselves, so a folder's subscribers stop receiving a moved
/// alarm (or a native alarm whose reference was dropped) the moment the generation says so, not when the
/// rebuild lands.
/// </para>
/// <para>
/// ⚠️ The SDK MEMOIZES the verdict per <c>(item, event type, SourceNode)</c> until the monitored node's
/// ⚠️ On the SDK channel path the SDK MEMOIZES the verdict per <c>(item, event type, SourceNode)</c> until the monitored node's
/// permission generation moves, so <see cref="InvalidateFolderEventVerdicts"/> must run whenever the inputs
/// change: a new authority generation or authorizer, the revalidation tick, and the end of every apply. A
/// condition's SourceNode is the condition itself, so a verdict is never shared between two conditions.
@@ -6278,9 +6488,9 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
// client connecting after a transition gets the current alarm picture. Those items are always refreshed
// here (RefreshServerObjectItems): with enforcement on through the same IsConditionVisibleTo decision live
// delivery uses, and in both modes so the refresh can cut off deferred events it supersedes (#854).
// Items on one of this manager's own notifiers keep the base behaviour with enforcement off (a stood-down
// item is off its notifier's fan-out, which the base already treats as "not refreshed"), and
// RefreshDirectItems with it on.
// Items on one of this manager's own notifiers go through RefreshDirectItems in both modes since #855 — the
// base's loop (a stood-down item is off its notifier's fan-out, so not refreshed) plus the #826 decision with
// enforcement on, and the same cut against deferred events as the Server object.
/// <inheritdoc />
public override ServiceResult ConditionRefresh(OperationContext context, IList<IEventMonitoredItem> monitoredItems)
{
@@ -6301,11 +6511,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
var enforced = AclEnforced;
var systemContext = SystemContext.Copy(context);
if (direct.Count > 0)
{
if (enforced) RefreshDirectItems(systemContext, direct);
else base.ConditionRefresh(context, direct);
}
if (direct.Count > 0) RefreshDirectItems(systemContext, direct, enforced);
if (serverItems.Count > 0) RefreshServerObjectItems(systemContext, serverItems, enforced);
return ServiceResult.Good;
@@ -6339,7 +6545,9 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
/// in its deferral queue (#838) and reach Server-object items only when the drain gets to them, off
/// <c>Lock</c>. A refresh in that window replays the condition's CURRENT state between RefreshStart and
/// RefreshEnd; the waiting events, all older, then followed RefreshEnd — a stale state, or a condition the
/// refresh had just dropped re-added, until the newest drained.
/// refresh had just dropped re-added, until the newest drained. (Since #855 folder items are no longer a
/// channel, so nothing of ours defers today; the cut stays as the safety net, and folder items share it —
/// <see cref="RefreshDirectItems"/>.)
/// </para>
/// <para>
/// <b>Common case — nothing deferred: unchanged, and off <c>Lock</c>.</b> The replay is the notifier walk the SDK
@@ -6353,10 +6561,10 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
/// <b>Deferring — the cut.</b> In ONE hold of <c>Lock</c> (under which every condition event is reported): the
/// watermark (<see cref="_conditionEventSequence"/>), the episode, and a frozen snapshot of each Retain
/// condition read from <see cref="_alarmConditions"/> (O(conditions), no tree walk), then queued to each item
/// with its <see cref="RefreshCut"/> installed under the item's <see cref="AllEventsItemState.Gate"/>. The ACL /
/// with its <see cref="RefreshCut"/> installed under the item's <see cref="EventItemState.Gate"/>. The ACL /
/// role verdicts are computed BEFORE taking <c>Lock</c>, against the live conditions (they depend on the event
/// type and source, not state). The cut and a "replay pending" marker are installed per item under its
/// <see cref="AllEventsItemState.Gate"/> inside that <c>Lock</c> hold; the replay is then queued OFF
/// <see cref="EventItemState.Gate"/> inside that <c>Lock</c> hold; the replay is then queued OFF
/// <c>Lock</c>, one event per hold of the gate. While the marker is set every condition event to the item takes
/// the gate: one newer than the cut marks its condition so its (older) frozen replay entry is dropped instead of
/// following it; a deferred one at or below the cut is judged against the cut (<see cref="RefreshCut.Supersedes"/>)
@@ -6406,17 +6614,12 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
{
// A cut-path replay of this item still being queued by a concurrent refresh is older than this live one:
// stop it, so it cannot follow this replay with frozen states.
if (_allEventsItemStates.TryGetValue(item, out var pending))
{
lock (pending.Gate)
{
pending.PendingReplay = 0;
pending.NewerSinceCut.Clear();
}
}
StopPendingReplay(item);
foreach (var (e, _, scopes) in replay)
{
// #855 review 3: never replay a condition a deploy removed or replaced since the walk collected it.
if (ConditionOf(e) is AlarmConditionState walked && !IsLiveCondition(walked)) continue;
if (scopes is not null && !IsConditionVisibleTo(item, scopes)) continue;
if (ServiceResult.IsBad(ValidateEventRolePermissions(item, e))) continue;
item.QueueEvent(e);
@@ -6472,7 +6675,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
}
}
var states = items.Select(AllEventsItemStateOf).ToList();
var states = items.Select(EventItemStateOf).ToList();
var perItem = new List<(IFilterTarget Event, AlarmConditionState Condition)>[items.Count];
var generation = Interlocked.Increment(ref _refreshGeneration);
// #854 re-review: a throw between installing the "replay pending" markers and finishing the replay must not leave
@@ -6540,20 +6743,69 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
}
}
// Queued off Lock, one replayed event per hold of the item gate: an event that reached the item since the cut is
// newer than the frozen replay of its condition, so that replay entry is dropped rather than queued after it (the
// newer event is already inside RefreshStart..RefreshEnd). A later refresh of the item supersedes this one.
ServerObjectRefreshProbe?.Invoke(true, Monitor.IsEntered(Lock));
// #855 review: the verdicts above were taken before Lock; a deploy may have revoked a grant since. Each replay entry
// is re-decided right before it is queued (scopes resolved once per condition).
var scopeCache = new Dictionary<NodeId, List<NodeId>>();
QueueFrozenReplay(
items, states, perItem.Select(entries => entries.Select(x => (x.Event, x.Condition.NodeId)).ToList()).ToArray(),
generation,
(item, e) =>
{
// #855 review 2: never replay a condition a deploy has removed or replaced since the cut — its frozen
// state would be a ghost (and, with the ACL on, a replaced scripted alarm would be judged on its old folder).
if (ConditionOf(e) is { } frozen && !IsLiveCondition(frozen)) return false;
if (enforced)
{
if (ConditionOf(e) is not { } condition) return false;
if (!scopeCache.TryGetValue(condition.NodeId, out var scopes))
scopeCache[condition.NodeId] = scopes = ConditionVisibilityScopes(condition);
if (!IsConditionVisibleTo(item, scopes)) return false;
}
return !ServiceResult.IsBad(ValidateEventRolePermissions(item, e));
});
}
finally
{
ClearPendingReplays(states, generation);
}
}
/// <summary>
/// Queues a cut-path refresh's frozen replay to its items OFF <c>Lock</c>, one replayed event per hold of the item
/// gate (#854; folder items too since #855): an event that reached the item since the cut is newer than the frozen
/// replay of its condition, so that replay entry is dropped rather than queued after it (the newer event is already
/// inside RefreshStart..RefreshEnd). A later refresh of the item supersedes this one.
/// </summary>
/// <param name="allowed">
/// The item's verdict for one entry, asked right before the entry's gate hold (never under it: it takes <c>Lock</c>),
/// against the CURRENT generation — the verdicts the cut was built on were taken earlier (#855 review). An entry
/// denied here is dropped AND removed from the cut's re-sent set, so the condition counts as "omitted for the
/// grants" and its later events are left to the ACL rather than withheld as superseded.
/// </param>
private static void QueueFrozenReplay(
List<IEventMonitoredItem> items, List<EventItemState> states, List<(IFilterTarget Event, NodeId ConditionId)>[] perItem,
int generation, Func<IEventMonitoredItem, IFilterTarget, bool> allowed)
{
for (var i = 0; i < items.Count; i++)
{
var item = items[i];
var state = states[i];
foreach (var (e, condition) in perItem[i])
foreach (var (e, conditionId) in perItem[i])
{
if (state.PendingReplay != generation) break;
var permitted = allowed(item, e);
lock (state.Gate)
{
if (state.PendingReplay != generation) break;
if (state.NewerSinceCut.Contains(condition.NodeId)) continue;
if (!permitted)
{
state.Cut?.Refreshed.Remove(conditionId);
continue;
}
if (state.NewerSinceCut.Contains(conditionId)) continue;
item.QueueEvent(e);
}
}
@@ -6567,51 +6819,234 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
}
}
}
}
finally
}
/// <summary>
/// #854 re-review: a throw between installing the "replay pending" markers and finishing the replay must not leave a
/// marker set (every sync event to that item would keep taking the gate until its next refresh).
/// </summary>
private static void ClearPendingReplays(List<EventItemState> states, int generation)
{
foreach (var state in states)
{
foreach (var state in states)
lock (state.Gate)
{
lock (state.Gate)
{
if (state.PendingReplay != generation) continue;
state.PendingReplay = 0;
state.NewerSinceCut.Clear();
}
if (state.PendingReplay != generation) continue;
state.PendingReplay = 0;
state.NewerSinceCut.Clear();
}
}
}
/// <summary>
/// ConditionRefresh for event items on this manager's own notifiers, under enforcement (Gitea #826): the SDK
/// base's loop exactly — an item no longer attached (stood down by revalidation, or tombstoned) is not
/// refreshed — plus the per-event decision live delivery makes (<see cref="IsNotifierEventVisibleTo"/>), so a
/// refresh cannot replay to a folder's subscriber an alarm that folder's live events withhold.
/// Stops a cut-path replay of <paramref name="item"/> still being queued by a concurrent refresh: a live replay that
/// is about to be queued is newer, and the frozen one must not follow it.
/// </summary>
private void StopPendingReplay(IEventMonitoredItem item)
{
if (!_eventItemStates.TryGetValue(item, out var pending)) return;
lock (pending.Gate)
{
pending.PendingReplay = 0;
pending.NewerSinceCut.Clear();
}
}
/// <summary>
/// ConditionRefresh (and ConditionRefresh2) for event items on this manager's own notifiers — equipment and raw device
/// folders — flag on and off: never overtaken by a deferred event older than what it replays (Gitea #855), the folder
/// twin of <see cref="RefreshServerObjectItems"/>.
/// </summary>
/// <param name="systemContext">The refresh's system context.</param>
/// <param name="items">The direct (non-Server-object) items being refreshed.</param>
private void RefreshDirectItems(ServerSystemContext systemContext, List<IEventMonitoredItem> items)
/// <param name="enforced">Whether the data-plane ACL is enforced.</param>
/// <remarks>
/// <para>
/// <b>Nothing deferred (the only case today — no notifier of ours defers since #855): unchanged, off
/// <c>Lock</c>.</b> The SDK base's loop — an item no longer attached (stood down by revalidation, or tombstoned)
/// is not refreshed; the item's notifier is walked and each live condition queued through the SDK's role check —
/// plus, under enforcement, the per-event decision live delivery makes (<see cref="IsNotifierEventVisibleTo"/>,
/// #826), so a refresh cannot replay to a folder's subscriber an alarm that folder's live events withhold. It
/// stands only if no deferral episode existed at any moment of it; the closing check takes <c>Lock</c> once, so a
/// report whose walk suspended (it marks the event before suspending, under <c>Lock</c>) has been counted by then.
/// Otherwise the cut path runs as well, and its replay — later inside the same RefreshStart..RefreshEnd —
/// supersedes anything that slipped in between.
/// </para>
/// <para>
/// <b>Deferring — the cut</b>, exactly as on the Server object: verdicts per (item, condition) first, off
/// <c>Lock</c>, from a live walk; then one <c>Lock</c> hold takes the watermark, the episode, the Retain set and,
/// per item, the same walk as frozen snapshots, and installs the cut and the "replay pending" marker under the
/// item's gate; the replay is queued off <c>Lock</c> (<see cref="QueueFrozenReplay"/>). Deferred events reaching
/// the item afterwards are judged against the cut (<see cref="QueueConditionEventToItem"/>).
/// </para>
/// </remarks>
private void RefreshDirectItems(ServerSystemContext systemContext, List<IEventMonitoredItem> items, bool enforced)
{
foreach (var item in items)
long episodeBefore;
bool deferringBefore;
lock (_eventReportGate)
{
NodeState? notifier;
lock (Lock)
episodeBefore = _deferralEpisode;
deferringBefore = _deferredEventReports > 0;
}
if (!deferringBefore)
{
foreach (var item in items)
{
if (!MonitoredItems.ContainsKey(item.Id)) continue;
notifier = (item.ManagerHandle as NodeHandle)?.Node;
var notifier = AttachedNotifierOf(item);
if (notifier is null) continue;
StopPendingReplay(item);
var events = new List<IFilterTarget>();
notifier.ConditionRefresh(systemContext, events, true);
DirectRefreshProbe?.Invoke(false, Monitor.IsEntered(Lock));
foreach (var e in events)
{
// #855 review 3: never replay a condition a deploy removed or replaced since the walk collected it.
if (ConditionOf(e) is AlarmConditionState walked && !IsLiveCondition(walked)) continue;
if (!IsDirectRefreshEventAllowed(item, e, enforced)) continue;
item.QueueEvent(e);
}
}
if (notifier is null) continue;
// Lock first: a report that suspended under it has registered its deferral by the time Lock is free.
lock (Lock)
{
lock (_eventReportGate)
{
if (_deferralEpisode == episodeBefore && _deferredEventReports == 0) return;
}
}
}
RefreshDirectItemsWithCut(systemContext, items, enforced);
}
/// <summary>The notifier an attached direct item subscribes to, or null for one that is not attached (not refreshed).</summary>
private NodeState? AttachedNotifierOf(IEventMonitoredItem item)
{
lock (Lock)
{
return MonitoredItems.ContainsKey(item.Id) ? (item.ManagerHandle as NodeHandle)?.Node : null;
}
}
/// <summary>A direct item's refresh verdict for one replayed event: the #826 decision under enforcement, then the SDK's.</summary>
private bool IsDirectRefreshEventAllowed(IEventMonitoredItem item, IFilterTarget e, bool enforced) =>
(!enforced || IsNotifierEventVisibleTo(item, e)) && !ServiceResult.IsBad(ValidateEventRolePermissions(item, e));
/// <summary>
/// Test probe (#855): called by a direct-item refresh with whether it took the cut path and whether the calling
/// thread held <c>Lock</c> while it queued the replay. Null in production.
/// </summary>
internal Action<bool, bool>? DirectRefreshProbe { get; set; }
/// <summary>The cut path of <see cref="RefreshDirectItems"/> — see there.</summary>
private void RefreshDirectItemsWithCut(ServerSystemContext systemContext, List<IEventMonitoredItem> items, bool enforced)
{
// Verdicts per (item, condition) before Lock, from a live walk: they depend on the event type, source, notifier
// wiring and grants, not on the condition's state.
var allowed = new HashSet<(IEventMonitoredItem, NodeState)>();
// Per (item, condition): a condition one item's walk saw is still undecided for another item (#855 review 4).
var known = new HashSet<(IEventMonitoredItem, NodeState)>();
foreach (var item in items)
{
if (AttachedNotifierOf(item) is not { } notifier) continue;
var events = new List<IFilterTarget>();
notifier.ConditionRefresh(systemContext, events, true);
foreach (var e in events)
{
if (!IsNotifierEventVisibleTo(item, e)) continue;
if (ServiceResult.IsBad(ValidateEventRolePermissions(item, e))) continue;
item.QueueEvent(e);
if (ConditionOf(e) is not { } node) continue;
known.Add((item, node));
if (IsDirectRefreshEventAllowed(item, e, enforced)) allowed.Add((item, node));
}
}
var states = items.Select(EventItemStateOf).ToList();
var perItem = new List<(IFilterTarget Event, NodeId ConditionId)>[items.Count];
var generation = Interlocked.Increment(ref _refreshGeneration);
try
{
lock (Lock)
{
var watermark = Interlocked.Read(ref _conditionEventSequence);
long episode;
bool deferring;
lock (_eventReportGate)
{
episode = _deferralEpisode;
deferring = _deferredEventReports > 0;
}
var retained = new HashSet<NodeId>();
foreach (var condition in _alarmConditions.Values)
{
if (condition.Retain?.Value == true) retained.Add(condition.NodeId);
}
for (var i = 0; i < items.Count; i++)
{
var item = items[i];
var entries = new List<(IFilterTarget Event, NodeId ConditionId)>();
perItem[i] = entries;
var state = states[i];
// Lock is held, so this is the attachment the walk below sees. An unattached item is not refreshed and
// gets no cut (it receives nothing until re-attached, and then every event is decided as it arrives).
var notifier = MonitoredItems.ContainsKey(item.Id) ? (item.ManagerHandle as NodeHandle)?.Node : null;
if (notifier is null)
{
lock (state.Gate)
{
state.PendingReplay = 0;
state.NewerSinceCut.Clear();
}
continue;
}
// Frozen at the watermark: the replay is exactly the state every event at or below it describes.
var live = new List<IFilterTarget>();
notifier.ConditionRefresh(SystemContext, live, true);
var refreshed = new HashSet<NodeId>();
foreach (var target in live)
{
if (target is not BaseInstanceState instance) continue;
// A condition materialised since the verdicts were taken is decided here (rare).
if (!allowed.Contains((item, instance)))
{
if (known.Contains((item, instance))) continue;
if (!IsDirectRefreshEventAllowed(item, target, enforced)) continue;
}
var snapshot = new InstanceStateSnapshot();
snapshot.Initialize(SystemContext, instance);
entries.Add((snapshot, instance.NodeId));
refreshed.Add(instance.NodeId);
}
lock (state.Gate)
{
state.Cut = deferring ? new RefreshCut(watermark, episode, refreshed, retained) : null;
state.PendingWatermark = watermark;
state.NewerSinceCut.Clear();
state.PendingReplay = generation;
}
}
}
DirectRefreshProbe?.Invoke(true, Monitor.IsEntered(Lock));
// #855 review: re-decided per entry right before queueing — a deploy may have revoked since the verdicts.
QueueFrozenReplay(
items, states, perItem, generation,
(item, e) => (ConditionOf(e) is not AlarmConditionState frozen || IsLiveCondition(frozen))
&& IsDirectRefreshEventAllowed(item, e, enforced));
}
finally
{
ClearPendingReplays(states, generation);
}
}
/// <summary>The condition a refresh event describes, or null when it names none (withheld: fail closed).</summary>
@@ -10,7 +10,9 @@ namespace ZB.MOM.WW.OtOpcUa.OpcUaServer;
/// <para>
/// The aggregate evaluators (<c>AggregateEvaluator</c> / <c>TimeWeightedAggregator</c> in
/// <c>Driver.Historian.Sql.Common</c>) are written against fixed settings, not parameterised: Uncertain time counts as
/// Good (<see cref="TreatUncertainAsBad"/> = false) in the time-weighted family; the per-sample Average / Minimum /
/// Good (<see cref="TreatUncertainAsBad"/> = false) in TimeAverage2 / Total2, while PercentGood / DurationGood count
/// Good time only and PercentBad / DurationBad Bad time only — Uncertain in neither (Part 13 §5.4.3.31-34, Mantis
/// 0006166); the per-sample Average / Minimum /
/// Maximum / Range / *ActualTime / Count are defined by Part 13 over Good values whatever this setting, as the
/// SDK computes them under it; an interval is Good only when all of
/// it is (<see cref="PercentDataGood"/> = 100) and Bad only when none of it is non-Bad
@@ -23,6 +25,20 @@ namespace ZB.MOM.WW.OtOpcUa.OpcUaServer;
/// the SDK manager so live AggregateFilter subscriptions agree with history reads; a test pins both sides.
/// </para>
/// <para>
/// The live calculators are the application's frozen 1.5.378.182 copies (<c>Compatibility/Aggregates</c>), whose
/// <c>ComputeDurationGoodBad</c> counts Uncertain time as Good under <see cref="TreatUncertainAsBad"/> = false. That
/// made live PercentGood / DurationGood disagree with HistoryRead whenever Uncertain data was present. #809
/// (owner-approved 2026-11-11) overrides exactly those two (<c>GoodOnlyStatusAggregateCalculator</c>), so live and
/// HistoryRead now agree on them — pinned by
/// <c>SdkAggregateParityTests.Live_AggregateFilter_PercentGood_and_DurationGood_equal_HistoryRead_over_Uncertain_samples</c>.
/// Likewise the frozen <c>AverageAggregateCalculator</c> integrated the span before an <b>Uncertain</b> first value
/// as a ramp from 0 when the data starts inside an interval (no start bound) — live TimeAverage / Total disagreed
/// with HistoryRead and Part 13 ("No Start Bound: Calculate Partial Interval"). #821 (owner-approved 2026-11-11)
/// overrides exactly those two (<c>CoveredTimeAverageAggregateCalculator</c>) — pinned by
/// <c>SdkAggregateParityTests.Live_AggregateFilter_TimeAverage_and_Total_equal_HistoryRead_over_an_Uncertain_first_late_start</c>.
/// The remaining documented live-vs-HistoryRead differences are listed in <c>docs/Historian.md</c>.
/// </para>
/// <para>
/// A request asking for <c>UseServerCapabilitiesDefaults</c>, or spelling out exactly these values, is served;
/// any other configuration is refused per node with <c>BadAggregateConfigurationRejected</c> rather than silently
/// answered under settings the client did not ask for (it used to be ignored).
@@ -295,7 +295,8 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
/// idempotent "already at this revision" short-circuit: the short-circuit would ack a clean Applied
/// for an address space known to be partial, and the re-apply is what repairs it (a degraded
/// non-rebuild apply leaves the publish actor's composition un-advanced, so it re-diffs the missing
/// nodes). Reset by every clean outcome and by every boot path that restores a revision.
/// nodes). Reset by every clean outcome and by the ConfigDb boot path (an Applied row is a final outcome);
/// SET by the cache boot paths, whose cached revision may never have reported an outcome (lmxopcua#868).
/// </summary>
private bool _currentRevisionDegraded;
@@ -1434,6 +1435,10 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
}
_currentRevision = revision;
// Unconfirmed (lmxopcua#868 third review #2): the cache is written BEFORE the address-space outcome, so a
// restart in that window boots a revision central never got an ack for. Marking it unconfirmed makes a
// re-dispatch of it RE-APPLY (and ack the real outcome) instead of short-circuiting a clean Applied.
_currentRevisionDegraded = true;
BecomeSteady();
_log.Info(
@@ -1514,6 +1519,8 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
// Stale would have started does not run here, so recovery rides on the next dispatch —
// matching how a normally-booted node behaves.
_currentRevision = revision;
// Unconfirmed, as in BootstrapFromCache (third review #2): a re-dispatch of it re-applies.
_currentRevisionDegraded = true;
BecomeSteady();
_isRunningFromCache = true;
@@ -2759,9 +2766,10 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
/// 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
/// statuses when it is an <c>IHostConnectivityProbe</c> (Running = connected; Faulted = faulted;
/// Stopped = stopped, counted separately since #786; Unknown = none) and the faulted hosts' names ride
/// along so the AdminUI can de-duplicate a device across a redundant pair; 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)
@@ -2772,20 +2780,40 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
if (health is null)
return new DriverInstanceDiagnostics(entry.Spec.DriverInstanceRowId, name, "Connecting", 0, 0, entry.SpawnedUtc);
int connected, faulted;
return DiagnosticsFromHealth(entry.Spec.DriverInstanceRowId, name, health);
}
/// <summary>The device counts of a child that has published health (#786: Faulted and Stopped split).</summary>
/// <param name="rowId">The driver instance's ConfigDb row id.</param>
/// <param name="name">The string driver-instance id.</param>
/// <param name="health">The child's last-published health.</param>
/// <returns>The diagnostics row.</returns>
internal static DriverInstanceDiagnostics DiagnosticsFromHealth(Guid rowId, string name, ChildHealthRecord health)
{
int connected, faulted, stopped;
IReadOnlyList<string> faultedHostNames;
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);
// #786: Faulted only. A Stopped host (e.g. a Galaxy engine an operator took off scan) is a
// deliberate, separately-reported state, not a fault.
var faultedHosts = hosts.Where(h => h.State == HostState.Faulted).ToArray();
faulted = faultedHosts.Length;
stopped = hosts.Count(h => h.State == HostState.Stopped);
faultedHostNames = faultedHosts.Select(h => h.HostName).ToArray();
}
else
{
connected = health.State is DriverState.Healthy or DriverState.Degraded ? 1 : 0;
faulted = health.State == DriverState.Faulted ? 1 : 0;
stopped = 0;
// A single-host driver is its own device: no host name; the AdminUI keys it on the instance id.
faultedHostNames = Array.Empty<string>();
}
return new DriverInstanceDiagnostics(
entry.Spec.DriverInstanceRowId, name, health.State.ToString(), connected, faulted, health.PublishedUtc);
rowId, name, health.State.ToString(), connected, faulted, health.PublishedUtc,
StoppedDevices: stopped, FaultedHostNames: faultedHostNames);
}
private void HandleDispatchFromSteady(DispatchDeployment msg)
@@ -2967,7 +2995,13 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
}
_currentRevision = msg.Revision;
UpsertNodeDeploymentState(msg.DeploymentId, NodeDeploymentStatus.Applied, failureReason: null);
// lmxopcua#868 review M1: the own row stays APPLYING (written by BeginDirectApply) until the
// address-space outcome is known — AwaitAddressSpaceOutcomeThenAck / ReleasePendingAck write the
// final Applied (clean, or with the degraded reason). Writing Applied/null here, up to ~105 s before
// that outcome, made the row read as a CLEAN ack: a coordinator restart in that window recovered it
// as complete and sealed the deployment clean while this node was about to report degraded (#601).
// A restart of THIS node in the window now replays the orphan Applying row (re-applies), which is
// the honest answer to "outcome unknown".
// The Applied ack is NOT sent here any more. It used to go out right now, before the
// address-space apply below had even been dequeued, so a degraded apply (swallowed sink faults,
// _lastApplied not advanced) still sealed the deployment green fleet-wide. The ack is now held
@@ -3052,6 +3086,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
if (_opcUaPublishActor is null)
{
_currentRevisionDegraded = false;
UpsertNodeDeploymentState(deploymentId, NodeDeploymentStatus.Applied, failureReason: null);
SendAck(deploymentId, ApplyAckOutcome.Applied, failureReason: null, correlation);
return;
}
@@ -3109,13 +3144,14 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
if (_currentRevision is { } cur && cur == pending.Revision)
_currentRevisionDegraded = degraded;
// The own row has been Applying since the apply began (lmxopcua#868 review M1); the FINAL outcome is
// written here, clean or degraded. Central adds the reason from the ack only when it ACCEPTS the ack (a
// timed-out or superseded deployment's ack is discarded), so a node with a ConfigDb records it itself; a
// driver-only node no-ops here.
UpsertNodeDeploymentState(pending.DeploymentId, NodeDeploymentStatus.Applied,
degraded ? ApplyAck.DegradedReasonPrefix + (reason ?? "address-space apply degraded") : null);
if (degraded)
{
// The apply wrote this node's own row Applied/null before the outcome was known. Central
// adds the reason from the ack only when it ACCEPTS the ack (a timed-out or superseded deployment's ack
// is discarded), so a node with a ConfigDb records it itself; a driver-only node no-ops here.
UpsertNodeDeploymentState(pending.DeploymentId, NodeDeploymentStatus.Applied,
ApplyAck.DegradedReasonPrefix + (reason ?? "address-space apply degraded"));
OtOpcUaTelemetry.DeploymentApplied.Add(1, new KeyValuePair<string, object?>("outcome", "degraded"));
_log.Warning(
"DriverHost {Node}: deployment {Id} (rev {Rev}) applied DEGRADED — {Reason}. Acking Applied+degraded; re-dispatching the same revision re-applies it",
@@ -4311,6 +4347,9 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
private void UpsertNodeDeploymentState(DeploymentId deploymentId, NodeDeploymentStatus status, string? failureReason)
{
// nvarchar(2048): an exception message or a degraded reason can exceed it, and SQL Server then refuses the
// write — the row would stay Applying forever (lmxopcua#868 third review #1).
failureReason = FailureReasons.Truncate(failureReason);
// Per-cluster mesh Phase 4: a driver-only node has NO ConfigDb. Central is the ack system of
// record — ConfigPublishCoordinator.PersistNodeAck already writes NodeDeploymentState from every
// ApplyAck it receives (and seeds the Applying rows at dispatch) — so this node's own upsert is