feat(runtime): audit refs still BadNoCommunication after a reconnect (dedup-trap tripwire)

This commit is contained in:
Joseph Doherty
2026-08-20 10:16:01 -04:00
parent 97f4ea0c9e
commit 7d98ef9332
6 changed files with 368 additions and 8 deletions
@@ -690,3 +690,20 @@ test could only pass once the production change landed.
this method, but that helper is T9's — calling it here would not compile. The call is added in T9, this method, but that helper is T9's — calling it here would not compile. The call is added in T9,
where the latch and its audit timer land together. R3-03 boundary respected: this is a new method plus where the latch and its audit timer land together. R3-03 boundary respected: this is a new method plus
one branch in `OnDriverConnectivityChanged`; `HandleRestartDriver` and the watchdog are untouched. one branch in `OnDriverConnectivityChanged`; `HandleRestartDriver` and the watchdog are untouched.
### T9 — recovery audit (`_commsLatched` + timer + `otopcua.driver.comms_latched_refs`)
- RED: `--filter "…DriverHostActorCommsLossValueNodesTests"` ⇒ **build failure**, `CS1739: The best
overload for 'Props' does not have a parameter named 'commsRecoveryAuditDelay'`.
- GREEN: 6/6 in the suite; 20 passed across it + `DriverHostActorCommsLossTests` +
`DriverHostActorBatchFanOutTests` + `DriverHostActorPrimaryGateTests`.
- **Falsification of the control** (the plan's discipline, applied to the audit rather than to a dedup
key): deleting the prune loop in `ForwardBatchToMux` ⇒
`A_driver_that_republishes_every_ref_trips_no_audit` **FAILED** with `Received 1 message too many.
Expected 0 messages but received 1 that matched filter [Warning when Message contains "still
BadNoCommunication"]`. Restored, green. So the "no audit fires" assertion is falsifiable and the audit
is not merely dormant.
- Meter assertions filter on a per-test **driver tag**, not on the instrument name alone (#503: a
`MeterListener` sees every measurement in the process).
- The latch counts **RawPaths**, so the raw node and its UNS mirrors report as one ref — the operator is
told how many tags did not come back, not how many nodes.
@@ -63,7 +63,7 @@
{ {
"id": "T9", "id": "T9",
"subject": "Recovery audit: _commsLatched per driver, pruned by routed values, CommsRecoveryAudit timer + otopcua.driver.comms_latched_refs meter (dedup-trap tripwire)", "subject": "Recovery audit: _commsLatched per driver, pruned by routed values, CommsRecoveryAudit timer + otopcua.driver.comms_latched_refs meter (dedup-trap tripwire)",
"status": "pending", "status": "done",
"blockedBy": [ "blockedBy": [
"T8" "T8"
] ]
@@ -106,6 +106,15 @@ public static class OtOpcUaTelemetry
/// <summary>Inbound operations denied by the Primary data-plane gate (archreview 03/S4) — /// <summary>Inbound operations denied by the Primary data-plane gate (archreview 03/S4) —
/// site=write|ack|alarm-emit, reason=secondary|detached|role-unknown.</summary> /// site=write|ack|alarm-emit, reason=secondary|detached|role-unknown.</summary>
/// <summary>Archreview 03/S15 — tag refs a driver left <c>BadNoCommunication</c> after it reconnected:
/// the host marked them Bad on the outage, and the driver never re-published them. Non-zero means a
/// driver that reconnects without a full re-publish (a push driver missing its initial push, or a
/// value-only dedup swallowing an unchanged value), which leaves those nodes Bad indefinitely.
/// Tagged driver + driver_type.</summary>
public static readonly Counter<long> DriverCommsLatchedRefs =
Meter.CreateCounter<long>("otopcua.driver.comms_latched_refs", unit: "{ref}",
description: "Refs still BadNoCommunication after their driver reconnected (driver, driver_type).");
public static readonly Counter<long> PrimaryGateDenied = public static readonly Counter<long> PrimaryGateDenied =
Meter.CreateCounter<long>("otopcua.redundancy.primary_gate_denied", unit: "{denial}", Meter.CreateCounter<long>("otopcua.redundancy.primary_gate_denied", unit: "{denial}",
description: "Operations denied by the Primary gate (site=write|ack|alarm-emit, reason=secondary|detached|role-unknown)."); description: "Operations denied by the Primary gate (site=write|ack|alarm-emit, reason=secondary|detached|role-unknown).");
@@ -82,6 +82,14 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
/// </remarks> /// </remarks>
public static readonly TimeSpan DefaultApplyWedgeTimeout = TimeSpan.FromMinutes(5); public static readonly TimeSpan DefaultApplyWedgeTimeout = TimeSpan.FromMinutes(5);
/// <summary>
/// How long after a driver reconnects the 03/S15 recovery audit waits before reporting refs still
/// latched <see cref="OpcUaStatus.BadNoCommunication"/>. Sized so an ordinary poll driver has had
/// several poll cycles — and a slow one at least one — to re-publish before anything is said; the
/// audit's job is to name a driver that will NEVER re-publish, not to race a slow first poll.
/// </summary>
public static readonly TimeSpan DefaultCommsRecoveryAuditDelay = TimeSpan.FromSeconds(30);
/// <summary> /// <summary>
/// Cache key used when an artifact carries no cluster scoping (single-cluster or unscoped /// Cache key used when an artifact carries no cluster scoping (single-cluster or unscoped
/// deployments). A literal rather than an empty string so the row is visibly deliberate when /// deployments). A literal rather than an empty string so the row is visibly deliberate when
@@ -401,6 +409,27 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
/// <see cref="DefaultApplyWedgeTimeout"/>).</summary> /// <see cref="DefaultApplyWedgeTimeout"/>).</summary>
private readonly TimeSpan _applyWedgeTimeout; private readonly TimeSpan _applyWedgeTimeout;
/// <summary>How long after a driver reconnects the 03/S15 recovery audit waits before reporting refs
/// still latched <see cref="OpcUaStatus.BadNoCommunication"/> (see <see cref="CommsRecoveryAudit"/>).</summary>
private readonly TimeSpan _commsRecoveryAuditDelay;
/// <summary>
/// Archreview <b>03/S15</b> — RawPaths this host has stamped <c>BadNoCommunication</c> on the value
/// nodes of, per driver, and has not yet seen a value for. Set by
/// <see cref="FanCommsLossToValueNodes"/>, pruned by <see cref="ForwardBatchToMux"/> as each ref
/// re-publishes, and read once by <see cref="CommsRecoveryAudit"/> after the driver reconnects.
/// <para><b>It is a tripwire, not a mechanism.</b> Nothing depends on it being correct for the data
/// plane to behave: the host publishes nothing Good on reconnect either way. What it buys is that a
/// driver which reconnects and then fails to re-publish — the pre-STAB-18 MQTT shape, and what any
/// future push driver without an initial push will do — produces a Warning and a metered count
/// instead of a subscription that silently reports Bad forever.</para>
/// <para>Keyed by RawPath, so a raw node and its UNS mirrors count once: the operator is being told
/// how many <i>tags</i> did not come back, not how many address-space nodes they project onto. The
/// entry is removed as soon as it empties, so the steady-state cost in
/// <see cref="ForwardBatchToMux"/> is one dictionary miss per burst.</para>
/// </summary>
private readonly Dictionary<string, HashSet<string>> _commsLatched = new(StringComparer.Ordinal);
/// <summary> /// <summary>
/// Host of this node's LocalDb replication partner — the only node that holds a copy of this /// Host of this node's LocalDb replication partner — the only node that holds a copy of this
/// node's alarm queue, and therefore the only node it may ever stand down in favour of. /// node's alarm queue, and therefore the only node it may ever stand down in favour of.
@@ -449,6 +478,12 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
/// <param name="OperatorUser">The authenticated principal performing the acknowledge.</param> /// <param name="OperatorUser">The authenticated principal performing the acknowledge.</param>
public sealed record RouteNativeAlarmAck(string ConditionNodeId, string? Comment, string OperatorUser); public sealed record RouteNativeAlarmAck(string ConditionNodeId, string? Comment, string OperatorUser);
/// <summary>Self-sent by the 03/S15 audit timer, <see cref="_commsRecoveryAuditDelay"/> after a driver
/// reconnected: report any of its refs still latched <c>BadNoCommunication</c>. See
/// <see cref="HandleCommsRecoveryAudit"/>.</summary>
/// <param name="DriverInstanceId">The driver whose recovery is being audited.</param>
public sealed record CommsRecoveryAudit(string DriverInstanceId);
private sealed record ChildEntry(IActorRef Actor, DriverInstanceSpec Spec, bool Stubbed) private sealed record ChildEntry(IActorRef Actor, DriverInstanceSpec Spec, bool Stubbed)
{ {
// Convenience accessors for sites that don't need the full spec. // Convenience accessors for sites that don't need the full spec.
@@ -514,6 +549,9 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
/// VirtualTag + ScriptedAlarm hosts. Defaults to null (admin-only nodes / tests skip the emit).</param> /// VirtualTag + ScriptedAlarm hosts. Defaults to null (admin-only nodes / tests skip the emit).</param>
/// <param name="applyWedgeTimeout">How long the host may sit in <see cref="Applying"/> before the /// <param name="applyWedgeTimeout">How long the host may sit in <see cref="Applying"/> before the
/// wedge watchdog logs a Warning. Defaults to <see cref="DefaultApplyWedgeTimeout"/>.</param> /// wedge watchdog logs a Warning. Defaults to <see cref="DefaultApplyWedgeTimeout"/>.</param>
/// <param name="commsRecoveryAuditDelay">Archreview 03/S15: how long after a driver reconnects the
/// recovery audit waits before reporting refs still latched <c>BadNoCommunication</c>. Defaults to
/// <see cref="DefaultCommsRecoveryAuditDelay"/>; tests shorten it.</param>
/// <returns>The Akka.NET <see cref="Akka.Actor.Props"/> used to spawn this actor.</returns> /// <returns>The Akka.NET <see cref="Akka.Actor.Props"/> used to spawn this actor.</returns>
public static Props Props( public static Props Props(
IDbContextFactory<OtOpcUaConfigDbContext>? dbFactory, IDbContextFactory<OtOpcUaConfigDbContext>? dbFactory,
@@ -542,7 +580,8 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
IDriverPrimaryRoleSource? primaryRoleSource = null, IDriverPrimaryRoleSource? primaryRoleSource = null,
IHistorizationRoleView? historizationRoleView = null, IHistorizationRoleView? historizationRoleView = null,
Func<CommonsNodeId, bool>? sharesHistorizationScopeWithLocal = null, Func<CommonsNodeId, bool>? sharesHistorizationScopeWithLocal = null,
TimeSpan? applyWedgeTimeout = null) => TimeSpan? applyWedgeTimeout = null,
TimeSpan? commsRecoveryAuditDelay = null) =>
// WARNING: this forwarding list is POSITIONAL, and Props.Create compiles it into an // WARNING: this forwarding list is POSITIONAL, and Props.Create compiles it into an
// expression tree. Six IActorRef? parameters and several interface-typed ones mean a // expression tree. Six IActorRef? parameters and several interface-typed ones mean a
// mis-ordered argument is usually type-compatible and therefore compiles clean, then binds // mis-ordered argument is usually type-compatible and therefore compiles clean, then binds
@@ -554,7 +593,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
loggerFactory, scriptRootLogger, scriptedAlarmHostOverride, invokerFactory, driverMemberCountProvider, loggerFactory, scriptRootLogger, scriptedAlarmHostOverride, invokerFactory, driverMemberCountProvider,
deploymentArtifactCache, redundancyRoleView, replicationPeerHost, fetchAndCacheMode, artifactFetcher, deploymentArtifactCache, redundancyRoleView, replicationPeerHost, fetchAndCacheMode, artifactFetcher,
alarmStateStore, telemetryHub, primaryRoleSource, historizationRoleView, alarmStateStore, telemetryHub, primaryRoleSource, historizationRoleView,
sharesHistorizationScopeWithLocal, applyWedgeTimeout)); sharesHistorizationScopeWithLocal, applyWedgeTimeout, commsRecoveryAuditDelay));
/// <summary>Initializes a new DriverHostActor with the specified dependencies.</summary> /// <summary>Initializes a new DriverHostActor with the specified dependencies.</summary>
/// <param name="dbFactory">Database context factory for configuration database access.</param> /// <param name="dbFactory">Database context factory for configuration database access.</param>
@@ -613,6 +652,9 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
/// <param name="applyWedgeTimeout">How long the host may sit in <see cref="Applying"/> before the /// <param name="applyWedgeTimeout">How long the host may sit in <see cref="Applying"/> before the
/// wedge watchdog logs a Warning. Defaults to <see cref="DefaultApplyWedgeTimeout"/>; a test shortens /// wedge watchdog logs a Warning. Defaults to <see cref="DefaultApplyWedgeTimeout"/>; a test shortens
/// it so the watchdog is observable in a test's lifetime.</param> /// it so the watchdog is observable in a test's lifetime.</param>
/// <param name="commsRecoveryAuditDelay">Archreview 03/S15: how long after a driver reconnects the
/// recovery audit waits before reporting refs still latched <c>BadNoCommunication</c>. Defaults to
/// <see cref="DefaultCommsRecoveryAuditDelay"/>; a test shortens it for the same reason.</param>
public DriverHostActor( public DriverHostActor(
IDbContextFactory<OtOpcUaConfigDbContext>? dbFactory, IDbContextFactory<OtOpcUaConfigDbContext>? dbFactory,
CommonsNodeId localNode, CommonsNodeId localNode,
@@ -640,9 +682,11 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
IDriverPrimaryRoleSource? primaryRoleSource = null, IDriverPrimaryRoleSource? primaryRoleSource = null,
IHistorizationRoleView? historizationRoleView = null, IHistorizationRoleView? historizationRoleView = null,
Func<CommonsNodeId, bool>? sharesHistorizationScopeWithLocal = null, Func<CommonsNodeId, bool>? sharesHistorizationScopeWithLocal = null,
TimeSpan? applyWedgeTimeout = null) TimeSpan? applyWedgeTimeout = null,
TimeSpan? commsRecoveryAuditDelay = null)
{ {
_applyWedgeTimeout = applyWedgeTimeout ?? DefaultApplyWedgeTimeout; _applyWedgeTimeout = applyWedgeTimeout ?? DefaultApplyWedgeTimeout;
_commsRecoveryAuditDelay = commsRecoveryAuditDelay ?? DefaultCommsRecoveryAuditDelay;
_historizationRoleView = historizationRoleView; _historizationRoleView = historizationRoleView;
_sharesHistorizationScopeWithLocal = sharesHistorizationScopeWithLocal; _sharesHistorizationScopeWithLocal = sharesHistorizationScopeWithLocal;
_telemetryHub = telemetryHub; _telemetryHub = telemetryHub;
@@ -1073,6 +1117,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
Receive<DriverInstanceActor.AttributeValueBatchPublished>(ForwardBatchToMux); Receive<DriverInstanceActor.AttributeValueBatchPublished>(ForwardBatchToMux);
Receive<DriverInstanceActor.AttributeAlarmPublished>(ForwardNativeAlarm); Receive<DriverInstanceActor.AttributeAlarmPublished>(ForwardNativeAlarm);
Receive<DriverInstanceActor.ConnectivityChanged>(OnDriverConnectivityChanged); Receive<DriverInstanceActor.ConnectivityChanged>(OnDriverConnectivityChanged);
Receive<CommsRecoveryAudit>(HandleCommsRecoveryAudit);
Receive<DriverInstanceActor.DeltaApplied>(HandleDeltaApplied); Receive<DriverInstanceActor.DeltaApplied>(HandleDeltaApplied);
Receive<DriverInstanceActor.ApplyResult>(HandleApplyResult); Receive<DriverInstanceActor.ApplyResult>(HandleApplyResult);
Receive<RestartDriver>(HandleRestartDriver); Receive<RestartDriver>(HandleRestartDriver);
@@ -1128,6 +1173,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
Receive<DriverInstanceActor.AttributeValueBatchPublished>(ForwardBatchToMux); Receive<DriverInstanceActor.AttributeValueBatchPublished>(ForwardBatchToMux);
Receive<DriverInstanceActor.AttributeAlarmPublished>(ForwardNativeAlarm); Receive<DriverInstanceActor.AttributeAlarmPublished>(ForwardNativeAlarm);
Receive<DriverInstanceActor.ConnectivityChanged>(OnDriverConnectivityChanged); Receive<DriverInstanceActor.ConnectivityChanged>(OnDriverConnectivityChanged);
Receive<CommsRecoveryAudit>(HandleCommsRecoveryAudit);
Receive<DriverInstanceActor.DeltaApplied>(HandleDeltaApplied); Receive<DriverInstanceActor.DeltaApplied>(HandleDeltaApplied);
Receive<DriverInstanceActor.ApplyResult>(HandleApplyResult); Receive<DriverInstanceActor.ApplyResult>(HandleApplyResult);
Receive<RestartDriver>(HandleRestartDriver); Receive<RestartDriver>(HandleRestartDriver);
@@ -1168,6 +1214,15 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
// One envelope per burst; the mux filters per item (it registered a batch receive in Task 9). // One envelope per burst; the mux filters per item (it registered a batch receive in Task 9).
_dependencyMux?.Tell(msg); _dependencyMux?.Tell(msg);
// 03/S15 recovery audit: clear whatever this burst re-published from the driver's Bad latch. Guarded
// by ONE dictionary probe per burst — with no outage outstanding (the steady state) this is a single
// miss, not per-item work.
if (_commsLatched.ContainsKey(msg.DriverInstanceId))
{
for (var i = 0; i < msg.Items.Count; i++)
TrackCommsLatchPrune(msg.DriverInstanceId, msg.Items[i].FullReference);
}
if (_opcUaPublishActor is null) return; if (_opcUaPublishActor is null) return;
// Indexed for-loop, not foreach: IReadOnlyList<AttributeValueItem> enumerates through an interface // Indexed for-loop, not foreach: IReadOnlyList<AttributeValueItem> enumerates through an interface
@@ -1266,6 +1321,12 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
// 03/S15: and the EXTERNAL data plane, which #481 stopped one seam short of. // 03/S15: and the EXTERNAL data plane, which #481 stopped one seam short of.
FanCommsLossToValueNodes(msg.DriverInstanceId); FanCommsLossToValueNodes(msg.DriverInstanceId);
} }
else
{
// 03/S15: nothing Good is published here — instead, check LATER whether the driver republished
// what this host marked Bad. See HandleCommsRecoveryAudit.
TrackCommsLatchArmAudit(msg.DriverInstanceId);
}
if (_opcUaPublishActor is null) return; if (_opcUaPublishActor is null) return;
@@ -1352,12 +1413,105 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
if (writes.Count == 0) return; if (writes.Count == 0) return;
_opcUaPublishActor.Tell(new OpcUa.OpcUaPublishActor.AttributeStatusBatchUpdate(writes)); _opcUaPublishActor.Tell(new OpcUa.OpcUaPublishActor.AttributeStatusBatchUpdate(writes));
TrackCommsLatchSet(driverInstanceId, entries);
_log.Info( _log.Info(
"DriverHost {Node}: driver {Driver} disconnected — {Count} value node(s) marked BadNoCommunication (value retained)", "DriverHost {Node}: driver {Driver} disconnected — {Count} value node(s) marked BadNoCommunication (value retained)",
_localNode, driverInstanceId, writes.Count); _localNode, driverInstanceId, writes.Count);
} }
/// <summary>
/// Records every RawPath just latched <c>BadNoCommunication</c> for <paramref name="driverInstanceId"/>
/// (03/S15's recovery audit — see <see cref="_commsLatched"/>). Replaces any previous latch wholesale:
/// a second outage re-arms from the current ref set rather than accumulating refs a redeploy removed.
/// </summary>
/// <param name="driverInstanceId">The driver that just went comms-lost.</param>
/// <param name="entries">Its value refs, from the per-driver index.</param>
private void TrackCommsLatchSet(
string driverInstanceId,
List<(string RawPath, HashSet<NodeRealmRef> Nodes)> entries)
{
var latched = new HashSet<string>(entries.Count, StringComparer.Ordinal);
foreach (var (rawPath, _) in entries) latched.Add(rawPath);
_commsLatched[driverInstanceId] = latched;
// A previous reconnect's audit is now meaningless — this driver is down again.
Timers.Cancel(CommsAuditTimerKey(driverInstanceId));
}
/// <summary>
/// Clears <paramref name="fullReference"/> from <paramref name="driverInstanceId"/>'s latch, and drops
/// the driver's entry once it empties. Called per routed value item, so the steady-state cost with no
/// latch outstanding is one dictionary miss per burst — not per item.
/// </summary>
/// <param name="driverInstanceId">The publishing driver.</param>
/// <param name="fullReference">The RawPath that just published a value.</param>
private void TrackCommsLatchPrune(string driverInstanceId, string fullReference)
{
if (!_commsLatched.TryGetValue(driverInstanceId, out var latched)) return;
if (!latched.Remove(fullReference)) return;
if (latched.Count == 0) _commsLatched.Remove(driverInstanceId);
}
/// <summary>
/// Arms the recovery audit for a driver that has just reconnected: after
/// <see cref="_commsRecoveryAuditDelay"/>, <see cref="HandleCommsRecoveryAudit"/> reports whatever is
/// still latched. Nothing is armed when the driver has no latch — it never went Bad here.
/// </summary>
/// <param name="driverInstanceId">The driver that just reconnected.</param>
private void TrackCommsLatchArmAudit(string driverInstanceId)
{
if (!_commsLatched.ContainsKey(driverInstanceId)) return;
Timers.StartSingleTimer(
CommsAuditTimerKey(driverInstanceId),
new CommsRecoveryAudit(driverInstanceId),
_commsRecoveryAuditDelay);
}
private static string CommsAuditTimerKey(string driverInstanceId) => $"comms-audit:{driverInstanceId}";
/// <summary>
/// The 03/S15 tripwire firing: refs this host latched <c>BadNoCommunication</c> that the driver has
/// still not re-published <see cref="_commsRecoveryAuditDelay"/> after it reconnected.
/// </summary>
/// <remarks>
/// <para>
/// <b>It reports, it does not repair.</b> Writing Good here would assert a quality nothing has
/// observed — the very thing this whole path refuses to do. What the operator gets instead is a
/// named driver and a count, plus the <c>otopcua.driver.comms_latched_refs</c> counter, which is
/// the operator-visible signature of a driver that reconnects without re-publishing (the
/// pre-STAB-18 MQTT shape, or any future push driver that forgets its initial push).
/// </para>
/// <para>
/// The latch is cleared afterwards either way: it has been reported, and leaving it would make
/// the next outage's audit report refs from this one.
/// </para>
/// </remarks>
/// <param name="msg">Names the driver whose audit is due.</param>
private void HandleCommsRecoveryAudit(CommsRecoveryAudit msg)
{
if (!_commsLatched.TryGetValue(msg.DriverInstanceId, out var latched) || latched.Count == 0)
{
_commsLatched.Remove(msg.DriverInstanceId);
return;
}
var driverType = _children.TryGetValue(msg.DriverInstanceId, out var child) ? child.DriverType : "unknown";
OtOpcUaTelemetry.DriverCommsLatchedRefs.Add(
latched.Count,
new KeyValuePair<string, object?>("driver", msg.DriverInstanceId),
new KeyValuePair<string, object?>("driver_type", driverType));
_log.Warning(
"DriverHost {Node}: driver {Driver} ({Type}) — {Count} ref(s) still BadNoCommunication {Delay} after reconnect; "
+ "the driver did not re-publish them (missing initial push, or a value-only dedup swallowed an unchanged value). "
+ "First: {First}",
_localNode, msg.DriverInstanceId, driverType, latched.Count, _commsRecoveryAuditDelay,
latched.OrderBy(r => r, StringComparer.Ordinal).First());
_commsLatched.Remove(msg.DriverInstanceId);
}
/// <summary> /// <summary>
/// #481 (Layer 4) — push a synthetic <see cref="OpcUaStatus.BadNoCommunication"/> into the dependency /// #481 (Layer 4) — push a synthetic <see cref="OpcUaStatus.BadNoCommunication"/> into the dependency
/// mux for every value ref the just-disconnected driver owns, so scripted alarms and virtual tags see /// mux for every value ref the just-disconnected driver owns, so scripted alarms and virtual tags see
@@ -1807,6 +1961,9 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
// A child connectivity transition while the host is Stale has no live address space to annotate — drop // A child connectivity transition while the host is Stale has no live address space to annotate — drop
// it (the post-recovery rebuild re-materialises conditions, and the child re-announces on its next entry). // it (the post-recovery rebuild re-materialises conditions, and the child re-announces on its next entry).
Receive<DriverInstanceActor.ConnectivityChanged>(_ => { }); Receive<DriverInstanceActor.ConnectivityChanged>(_ => { });
// Likewise the 03/S15 recovery audit: a timer armed before the host went Stale has nothing to report
// on (no latch was set while Stale, and the rebuild re-establishes every node's status anyway).
Receive<CommsRecoveryAudit>(_ => { });
// A late DeltaApplied (an apply completed just before the DB went Stale) — re-register the driver's // A late DeltaApplied (an apply completed just before the DB went Stale) — re-register the driver's
// mux adapter anyway; it simply re-reads the driver's current refs (harmless, no DB access). // mux adapter anyway; it simply re-reads the driver's current refs (harmless, no DB access).
Receive<DriverInstanceActor.DeltaApplied>(HandleDeltaApplied); Receive<DriverInstanceActor.DeltaApplied>(HandleDeltaApplied);
@@ -1,6 +1,8 @@
using System.Diagnostics.Metrics;
using Akka.Actor; using Akka.Actor;
using Shouldly; using Shouldly;
using Xunit; using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.Observability;
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa; using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
using ZB.MOM.WW.OtOpcUa.Commons.Types; using ZB.MOM.WW.OtOpcUa.Commons.Types;
using ZB.MOM.WW.OtOpcUa.Runtime.Drivers; using ZB.MOM.WW.OtOpcUa.Runtime.Drivers;
@@ -32,6 +34,17 @@ public sealed class DriverHostActorCommsLossValueNodesTests : RuntimeActorTestBa
{ {
private static readonly NodeId TestNode = NodeId.Parse("driver-commsloss-values"); private static readonly NodeId TestNode = NodeId.Parse("driver-commsloss-values");
private static readonly RevisionHash RevA = RevisionHash.Parse(new string('c', 64)); private static readonly RevisionHash RevA = RevisionHash.Parse(new string('c', 64));
private static readonly DateTime TsA = new(2026, 8, 20, 9, 0, 0, DateTimeKind.Utc);
/// <summary>Audit delay used by the T9 cases — long enough that a same-turn prune wins the race,
/// short enough to observe.</summary>
private static readonly TimeSpan AuditDelay = TimeSpan.FromMilliseconds(400);
// Distinct driver ids per metered test: the meter tag is what the recorder filters on, and a
// MeterListener sees every measurement in the process — including a parallel test's (#503).
private const string SilentDriver = "drv-audit-silent";
private const string RepublishingDriver = "drv-audit-republish";
private const string PartialDriver = "drv-audit-partial";
/// <summary> /// <summary>
/// A disconnect produces ONE status batch naming exactly the disconnected driver's raw nodes plus /// A disconnect produces ONE status batch naming exactly the disconnected driver's raw nodes plus
@@ -102,6 +115,122 @@ public sealed class DriverHostActorCommsLossValueNodesTests : RuntimeActorTestBa
publish.ExpectNoMsg(TimeSpan.FromMilliseconds(500)); publish.ExpectNoMsg(TimeSpan.FromMilliseconds(500));
} }
// ---------------------------------------------------------------------------------
// T9 — the recovery audit (the dedup-trap tripwire)
// ---------------------------------------------------------------------------------
/// <summary>
/// The host publishes nothing Good on reconnect, so the only way a stuck-Bad node becomes visible
/// is this audit: refs still latched some time after the driver reconnected are logged and metered.
/// Here the driver re-publishes nothing at all — the pre-STAB-18 MQTT shape, and the exact failure
/// any future push driver without an initial push would produce.
/// </summary>
[Fact]
public void Refs_still_latched_after_a_reconnect_are_logged_and_metered()
{
var db = NewInMemoryDbFactory();
var deploymentId = SeedV3Deployment(db, RevA,
rawTags: new[]
{
(Driver: SilentDriver, DriverName: "Modbus", Device: "dev1", Tag: "speed", DataType: "Double", Writable: false),
(Driver: SilentDriver, DriverName: "Modbus", Device: "dev1", Tag: "temp", DataType: "Double", Writable: false),
},
unsRefs: new[] { (Area: "filling", Line: "line1", Equip: "station1", BackingTag: "speed", Effective: (string?)null) });
var (actor, publish, _) = SpawnHostAndApply(this, db, TestNode, deploymentId, RevA, AuditDelay);
using var meter = new MeterRecorder("otopcua.driver.comms_latched_refs");
actor.Tell(new DriverInstanceActor.ConnectivityChanged(SilentDriver, Connected: false));
publish.ExpectMsg<OpcUaPublishActor.AttributeStatusBatchUpdate>(PresenceBudget);
EventFilter.Warning(contains: "still BadNoCommunication").ExpectOne(TimeSpan.FromSeconds(10), () =>
actor.Tell(new DriverInstanceActor.ConnectivityChanged(SilentDriver, Connected: true)));
// Two REFS, not three nodes: the latch is keyed by RawPath, and the UNS mirror shares its
// backing tag's ref — the operator is being told how many tags did not come back, not how many
// address-space nodes they project onto.
AwaitCondition(() => meter.WithTag("driver", SilentDriver) > 0, TimeSpan.FromSeconds(5));
meter.SumWithTag("driver", SilentDriver).ShouldBe(2);
}
/// <summary>
/// <b>The control.</b> A driver that DOES re-publish every latched ref on reconnect trips nothing —
/// without this, the test above would pass against an audit that fires unconditionally, which would
/// make the Warning noise rather than signal.
/// </summary>
[Fact]
public void A_driver_that_republishes_every_ref_trips_no_audit()
{
var db = NewInMemoryDbFactory();
var deploymentId = SeedV3Deployment(db, RevA,
rawTags: new[]
{
(Driver: RepublishingDriver, DriverName: "Modbus", Device: "dev1", Tag: "speed", DataType: "Double", Writable: false),
(Driver: RepublishingDriver, DriverName: "Modbus", Device: "dev1", Tag: "temp", DataType: "Double", Writable: false),
},
unsRefs: new[] { (Area: "filling", Line: "line1", Equip: "station1", BackingTag: "speed", Effective: (string?)null) });
var (actor, publish, _) = SpawnHostAndApply(this, db, TestNode, deploymentId, RevA, AuditDelay);
using var meter = new MeterRecorder("otopcua.driver.comms_latched_refs");
actor.Tell(new DriverInstanceActor.ConnectivityChanged(RepublishingDriver, Connected: false));
publish.ExpectMsg<OpcUaPublishActor.AttributeStatusBatchUpdate>(PresenceBudget);
EventFilter.Warning(contains: "still BadNoCommunication").Expect(0, TimeSpan.FromSeconds(5), () =>
{
actor.Tell(new DriverInstanceActor.ConnectivityChanged(RepublishingDriver, Connected: true));
// The driver's own re-publish — identical values, which is precisely the case a value-only
// dedup would swallow; here it is the driver that force-raised, so both refs are pruned.
actor.Tell(new DriverInstanceActor.AttributeValueBatchPublished(RepublishingDriver, new[]
{
new DriverInstanceActor.AttributeValueItem("Plant/Modbus/dev1/speed", 42.0, OpcUaStatus.Good, TsA),
new DriverInstanceActor.AttributeValueItem("Plant/Modbus/dev1/temp", 7.5, OpcUaStatus.Good, TsA),
}));
publish.ExpectMsg<OpcUaPublishActor.AttributeValueBatchUpdate>(PresenceBudget);
});
meter.WithTag("driver", RepublishingDriver).ShouldBe(0);
}
/// <summary>
/// A partial recovery is still reported, and reports only what is missing: one ref comes back, the
/// other does not, and the audit counts exactly one. An audit that reported the whole latch (or
/// nothing at all once anything arrived) would be useless on precisely the mixed case an operator
/// needs it for.
/// </summary>
[Fact]
public void A_partial_recovery_reports_only_the_refs_that_did_not_come_back()
{
var db = NewInMemoryDbFactory();
var deploymentId = SeedV3Deployment(db, RevA,
rawTags: new[]
{
(Driver: PartialDriver, DriverName: "Modbus", Device: "dev1", Tag: "speed", DataType: "Double", Writable: false),
(Driver: PartialDriver, DriverName: "Modbus", Device: "dev1", Tag: "temp", DataType: "Double", Writable: false),
},
unsRefs: Array.Empty<(string, string, string, string, string?)>());
var (actor, publish, _) = SpawnHostAndApply(this, db, TestNode, deploymentId, RevA, AuditDelay);
using var meter = new MeterRecorder("otopcua.driver.comms_latched_refs");
actor.Tell(new DriverInstanceActor.ConnectivityChanged(PartialDriver, Connected: false));
publish.ExpectMsg<OpcUaPublishActor.AttributeStatusBatchUpdate>(PresenceBudget);
EventFilter.Warning(contains: "still BadNoCommunication").ExpectOne(TimeSpan.FromSeconds(10), () =>
{
actor.Tell(new DriverInstanceActor.ConnectivityChanged(PartialDriver, Connected: true));
actor.Tell(new DriverInstanceActor.AttributeValueBatchPublished(PartialDriver, new[]
{
new DriverInstanceActor.AttributeValueItem("Plant/Modbus/dev1/speed", 42.0, OpcUaStatus.Good, TsA),
}));
publish.ExpectMsg<OpcUaPublishActor.AttributeValueBatchUpdate>(PresenceBudget);
});
AwaitCondition(() => meter.WithTag("driver", PartialDriver) > 0, TimeSpan.FromSeconds(5));
meter.SumWithTag("driver", PartialDriver).ShouldBe(1, "only the ref that never came back is reported");
}
/// <summary> /// <summary>
/// A driver that owns no value refs produces no message at all — an empty batch would cost a hop /// A driver that owns no value refs produces no message at all — an empty batch would cost a hop
/// and a sink call for nothing on every flap of an alarm-only or freshly-added driver. /// and a sink call for nothing on every flap of an alarm-only or freshly-added driver.
@@ -123,4 +252,47 @@ public sealed class DriverHostActorCommsLossValueNodesTests : RuntimeActorTestBa
publish.ExpectNoMsg(TimeSpan.FromMilliseconds(500)); publish.ExpectNoMsg(TimeSpan.FromMilliseconds(500));
} }
/// <summary>
/// Sums one counter's measurements, <b>filtered by tag</b>. A <see cref="MeterListener"/> observes
/// every measurement in the process, so filtering on instrument name alone would fold in a parallel
/// test's (#503) — each test here therefore uses its own driver id and asserts on that tag.
/// </summary>
private sealed class MeterRecorder : IDisposable
{
private readonly MeterListener _listener;
private readonly List<(long Value, KeyValuePair<string, object?>[] Tags)> _measurements = new();
private readonly object _gate = new();
public MeterRecorder(string instrumentName)
{
_listener = new MeterListener
{
InstrumentPublished = (instrument, listener) =>
{
if (instrument.Meter.Name == OtOpcUaTelemetry.MeterName && instrument.Name == instrumentName)
listener.EnableMeasurementEvents(instrument);
},
};
_listener.SetMeasurementEventCallback<long>((_, value, tags, _) =>
{
lock (_gate) _measurements.Add((value, tags.ToArray()));
});
_listener.Start();
}
public int WithTag(string key, string value)
{
lock (_gate) return _measurements.Count(m => m.Tags.Any(t => t.Key == key && Equals(t.Value, value)));
}
public long SumWithTag(string key, string value)
{
lock (_gate)
return _measurements.Where(m => m.Tags.Any(t => t.Key == key && Equals(t.Value, value)))
.Sum(m => m.Value);
}
public void Dispose() => _listener.Dispose();
}
} }
@@ -114,13 +114,17 @@ public static class V3ArtifactSeeder
/// <param name="localNode">The host's local node id.</param> /// <param name="localNode">The host's local node id.</param>
/// <param name="deploymentId">The deployment to dispatch.</param> /// <param name="deploymentId">The deployment to dispatch.</param>
/// <param name="rev">The revision hash the deployment was sealed under.</param> /// <param name="rev">The revision hash the deployment was sealed under.</param>
/// <param name="commsRecoveryAuditDelay">Optional 03/S15 audit delay — how long after a reconnect the
/// host checks for refs still latched <c>BadNoCommunication</c>. Null keeps the production default,
/// which is far longer than any test's lifetime.</param>
/// <returns>The host actor plus the publish and mux probes wired into it.</returns> /// <returns>The host actor plus the publish and mux probes wired into it.</returns>
public static (IActorRef Actor, TestProbe Publish, TestProbe Mux) SpawnHostAndApply( public static (IActorRef Actor, TestProbe Publish, TestProbe Mux) SpawnHostAndApply(
TestKitBase kit, IDbContextFactory<OtOpcUaConfigDbContext> db, TestKitBase kit, IDbContextFactory<OtOpcUaConfigDbContext> db,
NodeId localNode, DeploymentId deploymentId, RevisionHash rev) NodeId localNode, DeploymentId deploymentId, RevisionHash rev,
TimeSpan? commsRecoveryAuditDelay = null)
{ {
var publish = kit.CreateTestProbe(); var publish = kit.CreateTestProbe();
var (actor, mux, coordinator) = Spawn(kit, db, localNode, publish.Ref); var (actor, mux, coordinator) = Spawn(kit, db, localNode, publish.Ref, commsRecoveryAuditDelay);
Dispatch(actor, coordinator, deploymentId, rev); Dispatch(actor, coordinator, deploymentId, rev);
publish.ExpectMsg<OpcUaPublishActor.RebuildAddressSpace>(RuntimeActorTestBase.PresenceBudget); publish.ExpectMsg<OpcUaPublishActor.RebuildAddressSpace>(RuntimeActorTestBase.PresenceBudget);
@@ -147,7 +151,7 @@ public static class V3ArtifactSeeder
private static (IActorRef Actor, TestProbe Mux, TestProbe Coordinator) Spawn( private static (IActorRef Actor, TestProbe Mux, TestProbe Coordinator) Spawn(
TestKitBase kit, IDbContextFactory<OtOpcUaConfigDbContext> db, NodeId localNode, TestKitBase kit, IDbContextFactory<OtOpcUaConfigDbContext> db, NodeId localNode,
IActorRef? opcUaPublishActor) IActorRef? opcUaPublishActor, TimeSpan? commsRecoveryAuditDelay = null)
{ {
var coordinator = kit.CreateTestProbe(); var coordinator = kit.CreateTestProbe();
var mux = kit.CreateTestProbe(); var mux = kit.CreateTestProbe();
@@ -159,7 +163,8 @@ public static class V3ArtifactSeeder
dependencyMux: mux.Ref, dependencyMux: mux.Ref,
opcUaPublishActor: opcUaPublishActor, opcUaPublishActor: opcUaPublishActor,
virtualTagEvaluator: NullVirtualTagEvaluator.Instance, virtualTagEvaluator: NullVirtualTagEvaluator.Instance,
virtualTagHostOverride: vtHost.Ref)); virtualTagHostOverride: vtHost.Ref,
commsRecoveryAuditDelay: commsRecoveryAuditDelay));
return (actor, mux, coordinator); return (actor, mux, coordinator);
} }