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

Merge branch 'master' into fix/r3-02-stale-good-family

Browse Source 
# Conflicts:
#	src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs
#	tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Harness/V3ArtifactSeeder.cs
This commit is contained in:
Joseph Doherty
2026-08-20 11:03:02 -04:00
parent 5dc8d34472 d79d51dd4f
commit 1d251cc037
46 changed files with 5411 additions and 208 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs
+190 -19
View File
@@ -82,6 +82,14 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
/// </remarks>
public static readonly TimeSpan DefaultApplyWedgeTimeout = TimeSpan.FromMinutes(5);
/// <summary>Minimum interval between two supervision reprovisions of the SAME driver child (archreview
/// review F3). Akka's default decider restarts a faulting child immediately and without limit, and the
/// reprovision hook makes every restart do ShutdownAsync + InitializeAsync + SubscribeAsync — so an
/// un-limited hook turns a driver that faults on every value into a reconnect storm against its device.
/// Matched to <see cref="DriverInstanceActor.DefaultReconnectInterval"/>: the reprovision drives the same
/// connect work, so it should not run more often than an ordinary reconnect attempt.</summary>
public static readonly TimeSpan DefaultReprovisionBackoff = TimeSpan.FromSeconds(10);
/// <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
@@ -352,9 +360,25 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
/// the config (#516). Bounded by the number of driver children with a delta in flight.</summary>
private readonly Dictionary<CorrelationId, DriverInstanceSpec> _pendingDelta = new();
/// <summary>The composition from the most-recent apply (set at the END of
/// <see cref="PushDesiredSubscriptions"/>). Null until the first apply.</summary>
private AddressSpaceComposition? _lastComposition;
/// <summary>One driver child's desired subscription payload — exactly the
/// <see cref="DriverInstanceActor.SetDesiredSubscriptions"/> the apply's bulk push sent it.</summary>
/// <param name="ValueRefs">The non-alarm raw tags' RawPaths this driver should keep subscribed.</param>
/// <param name="AlarmRefs">The alarm-bearing raw tags' RawPaths that un-gate this driver's
/// <see cref="IAlarmSource"/> feed.</param>
private readonly record struct DesiredSubscriptionSet(
IReadOnlyList<string> ValueRefs,
IReadOnlyList<string> AlarmRefs);
/// <summary>Per-driver desired subscription sets from the most-recent apply. THE authority: the bulk
/// push, <see cref="SpawnChild"/> and the supervision-reprovision handler all send from here, so a
/// child obtained by ANY route carries the same set. Cleared and repopulated whole on every apply,
/// alongside the routing maps, so renames/removals can never leave a stale entry. Empty until the
/// first apply — a child spawned before any apply is handed nothing, which is what it should have.
/// <para>It is only ever rebuilt by a SUCCESSFUL apply, so after a failed one it deliberately lingers on
/// the last-good composition — matching the address space, drivers and subscriptions that also keep
/// serving through a failed apply (#485/#486). A respawn or reprovision in that window therefore hands
/// the child the last-known-good set, not an empty one.</para></summary>
private readonly Dictionary<string, DesiredSubscriptionSet> _desiredByDriver = new(StringComparer.Ordinal);
/// <summary>
/// Cached local <see cref="RedundancyRole"/> from the latest <see cref="RedundancyStateChanged"/>
@@ -430,6 +454,15 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
/// </summary>
private readonly Dictionary<string, HashSet<string>> _commsLatched = new(StringComparer.Ordinal);
/// <summary>Per-driver rate limit for the supervision-reprovision hook (see
/// <see cref="DefaultReprovisionBackoff"/>).</summary>
private readonly TimeSpan _reprovisionBackoff;
/// <summary>When each driver was last reprovisioned after a supervision restart, so a crash loop is
/// paced rather than amplified. Bounded by the number of driver children; an entry for a driver that is
/// later removed simply stops being consulted (the handler resolves <c>_children</c> first).</summary>
private readonly Dictionary<string, DateTime> _lastReprovisionUtc = new(StringComparer.Ordinal);
/// <summary>
/// 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.
@@ -581,6 +614,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
IHistorizationRoleView? historizationRoleView = null,
Func<CommonsNodeId, bool>? sharesHistorizationScopeWithLocal = null,
TimeSpan? applyWedgeTimeout = null,
TimeSpan? reprovisionBackoff = null,
TimeSpan? commsRecoveryAuditDelay = null) =>
// 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
@@ -593,7 +627,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
loggerFactory, scriptRootLogger, scriptedAlarmHostOverride, invokerFactory, driverMemberCountProvider,
deploymentArtifactCache, redundancyRoleView, replicationPeerHost, fetchAndCacheMode, artifactFetcher,
alarmStateStore, telemetryHub, primaryRoleSource, historizationRoleView,
sharesHistorizationScopeWithLocal, applyWedgeTimeout, commsRecoveryAuditDelay));
sharesHistorizationScopeWithLocal, applyWedgeTimeout, reprovisionBackoff, commsRecoveryAuditDelay));
/// <summary>Initializes a new DriverHostActor with the specified dependencies.</summary>
/// <param name="dbFactory">Database context factory for configuration database access.</param>
@@ -683,9 +717,11 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
IHistorizationRoleView? historizationRoleView = null,
Func<CommonsNodeId, bool>? sharesHistorizationScopeWithLocal = null,
TimeSpan? applyWedgeTimeout = null,
TimeSpan? reprovisionBackoff = null,
TimeSpan? commsRecoveryAuditDelay = null)
{
_applyWedgeTimeout = applyWedgeTimeout ?? DefaultApplyWedgeTimeout;
_reprovisionBackoff = reprovisionBackoff ?? DefaultReprovisionBackoff;
_commsRecoveryAuditDelay = commsRecoveryAuditDelay ?? DefaultCommsRecoveryAuditDelay;
_historizationRoleView = historizationRoleView;
_sharesHistorizationScopeWithLocal = sharesHistorizationScopeWithLocal;
@@ -1121,6 +1157,8 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
Receive<DriverInstanceActor.DeltaApplied>(HandleDeltaApplied);
Receive<DriverInstanceActor.ApplyResult>(HandleApplyResult);
Receive<RestartDriver>(HandleRestartDriver);
Receive<DriverInstanceActor.ReprovisionRequested>(HandleReprovisionRequested);
Receive<ReprovisionDeferred>(m => ReprovisionChild(m.Request, m.Child));
Receive<ReconnectDriver>(HandleReconnectDriver);
Receive<RouteNodeWrite>(HandleRouteNodeWrite);
Receive<RouteNativeAlarmAck>(HandleRouteNativeAlarmAck);
@@ -1177,6 +1215,8 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
Receive<DriverInstanceActor.DeltaApplied>(HandleDeltaApplied);
Receive<DriverInstanceActor.ApplyResult>(HandleApplyResult);
Receive<RestartDriver>(HandleRestartDriver);
Receive<DriverInstanceActor.ReprovisionRequested>(HandleReprovisionRequested);
Receive<ReprovisionDeferred>(m => ReprovisionChild(m.Request, m.Child));
Receive<ReconnectDriver>(HandleReconnectDriver);
Receive<RouteNodeWrite>(HandleRouteNodeWrite);
Receive<RouteNativeAlarmAck>(HandleRouteNativeAlarmAck);
@@ -1943,6 +1983,8 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
Receive<GetDiagnostics>(HandleGetDiagnostics);
Receive<RetryConfigDbConnection>(_ => TryRecoverFromStale());
Receive<RestartDriver>(HandleRestartDriver);
Receive<DriverInstanceActor.ReprovisionRequested>(HandleReprovisionRequested);
Receive<ReprovisionDeferred>(m => ReprovisionChild(m.Request, m.Child));
Receive<ReconnectDriver>(HandleReconnectDriver);
Receive<RedundancyStateChanged>(OnRedundancyStateChanged);
// The redundancy-state topic also carries OpcUaProbeResult (OpcUaPublishActor peer-probes).
@@ -2885,23 +2927,25 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
}
}
// One authored-only push (value refs + alarm refs from the maps built above), shared by the bulk loop AND
// the dropped-driver fallback so the two CANNOT drift: the fallback's correctness depends on sending the
// SAME payload the bulk loop would have, so it's a shared helper (structural), not a comment-maintained
// invariant. An EMPTY set is valid — the child's Connected handler routes it to Unsubscribe (dropping a
// stale handle) rather than a spurious subscribe. Returns the value-ref count for the bulk-loop log total.
int SendAuthoredOnly(IActorRef actor, string driverId)
// Cache the per-driver payload as THE authority, in the same clear-and-repopulate pass as the routing
// maps above so a rename/removal can never leave a stale entry behind. Keyed over the UNION of the two
// ref maps AND the running children: a child with no authored tags must get an explicit EMPTY set (which
// it routes to Unsubscribe), not be absent from the authority — absence and empty must not differ here.
_desiredByDriver.Clear();
foreach (var driverId in refsByDriver.Keys
.Concat(alarmRefsByDriver.Keys)
.Concat(_children.Keys)
.Distinct(StringComparer.Ordinal))
{
var refs = refsByDriver.TryGetValue(driverId, out var r) ? r : Array.Empty<string>();
var alarmRefs = alarmRefsByDriver.TryGetValue(driverId, out var ar) ? ar : Array.Empty<string>();
actor.Tell(new DriverInstanceActor.SetDesiredSubscriptions(refs, SubscriptionPublishingInterval, alarmRefs));
return refs.Count;
_desiredByDriver[driverId] = new DesiredSubscriptionSet(
refsByDriver.TryGetValue(driverId, out var r) ? r : Array.Empty<string>(),
alarmRefsByDriver.TryGetValue(driverId, out var ar) ? ar : Array.Empty<string>());
}
var total = 0;
foreach (var (driverId, entry) in _children)
{
total += SendAuthoredOnly(entry.Actor, driverId);
total += SendDesiredSubscriptions(entry.Actor, driverId);
}
if (total > 0)
@@ -2936,11 +2980,34 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
_log.Info("DriverHost {Node}: applied {Count} Equipment ScriptedAlarm(s) to the ScriptedAlarm host",
_localNode, composition.EquipmentScriptedAlarms.Count);
}
}
// Cache the applied composition LAST. Set here (not in HandleArtifactForApply) so both the fresh-apply and
// bootstrap-restore paths — which both route through this method — leave a current composition.
_lastComposition = composition;
/// <summary>
/// Send <paramref name="driverId"/>'s desired subscription payload to <paramref name="actor"/>.
/// </summary>
/// <remarks>
/// The ONE place a <see cref="DriverInstanceActor.SetDesiredSubscriptions"/> is built, so the apply's
/// bulk push, a respawn (<see cref="SpawnChild"/>) and a supervision reprovision
/// (<c>HandleReprovisionRequested</c>) cannot drift. That is structural rather than
/// comment-maintained on purpose: a second sender that assembled its own payload is exactly how
/// 03/S16 happened — the restart path got a child without the set that makes a child useful.
/// <para>
/// An EMPTY set is valid and MEANINGFUL — the child routes it to <c>Unsubscribe</c> (dropping a
/// stale handle) rather than to a spurious subscribe — so a driver missing from
/// <see cref="_desiredByDriver"/> is sent an empty set, never skipped.
/// </para>
/// </remarks>
/// <param name="actor">The driver child to hand the set to.</param>
/// <param name="driverId">The driver instance whose set to send.</param>
/// <returns>The value-ref count, for the bulk loop's log total.</returns>
private int SendDesiredSubscriptions(IActorRef actor, string driverId)
{
var set = _desiredByDriver.TryGetValue(driverId, out var s)
? s
: new DesiredSubscriptionSet(Array.Empty<string>(), Array.Empty<string>());
actor.Tell(new DriverInstanceActor.SetDesiredSubscriptions(
set.ValueRefs, SubscriptionPublishingInterval, set.AlarmRefs));
return set.ValueRefs.Count;
}
private void SpawnChild(DriverInstanceSpec spec)
@@ -3014,6 +3081,29 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
_children[spec.DriverInstanceId] = new ChildEntry(child, spec, stub);
// EVERY spawn hands the child its desired set — deploy reconcile, operator Restart (03/S16), and the
// supervision reprovision. Sending it HERE rather than at each call site is the point: a child that
// exists without its subscription set is a dead driver, so the two facts are established together.
// Before this, the only sender was the apply's bulk push, so an operator's Restart produced a child
// that connected and then read nothing until the next deployment — the #488 periodic reconcile cannot
// help, because it early-returns on an empty desired set.
//
// ORDERING, and its one accepted cost: ApplyDriverSpecs (which spawns) runs BEFORE
// ApplyDesiredSubscriptions (which rebuilds _desiredByDriver), so on a deploy that RESPAWNS a child
// (#516 — the merged DriverConfig folds in the raw tags, so a tag-set edit changes the spec) the new
// child is handed the PREVIOUS apply's set, and nothing at all on the first deploy. That transient is
// superseded a moment later by the bulk push, on the same sender→receiver pair, so Akka's per-pair
// ordering guarantees the child CONVERGES on the new set — but it is NOT guaranteed to be inert on
// the way there: the child runs concurrently with the rest of this handler, so a driver whose
// InitializeAsync completes quickly can reach Connected and subscribe the stale set before the bulk
// push arrives, then re-subscribe. The cost is one extra subscribe/unsubscribe round-trip on a
// respawning deploy (HandleSubscribeAsync drops the prior subscription before establishing the new
// one, so it converges); a real device driver, whose initialise opens a socket, rarely wins that race.
// Deliberately not "fixed" by withholding the send: a child that never gets a set is the 03/S16
// defect, and the operator-Restart and supervision paths have no bulk push coming at all.
// Sent after InitializeRequested above, which keeps init-before-subscribe ordering.
SendDesiredSubscriptions(child, spec.DriverInstanceId);
// v3 (WP7): if the (real, non-stub) driver consumes other tags' live values, spawn a mux-adapter
// child that registers its dependency refs on the per-node dependency mux and forwards changes
// into the driver. Requires a wired mux (dev/None deployments have none — then calc inputs simply
@@ -3164,6 +3254,87 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
SpawnChild(entry.Spec);
}
/// <summary>
/// Re-provision a driver child that Akka's supervision just RESTARTED in place (archreview 03/S6,
/// 03/S16) — the third and least visible of the three incarnation routes.
/// </summary>
/// <remarks>
/// A restart recreates the actor object around the SAME driver instance (the Props closure captured
/// it) without going through <see cref="SpawnChild"/>, so neither the <c>InitializeRequested</c> nor
/// the desired-subscription send that a spawn performs ever happens — and the child's
/// <see cref="IActorRef"/> is unchanged, so <c>_children</c> needs no maintenance and the host has no
/// other notification that a new incarnation exists. Its <c>PreRestart</c> ran <c>PostStop</c>, which
/// called <c>ShutdownAsync</c> on the driver, so re-initialising is not optional: without it the
/// child sits in <c>Connecting</c> forever around a shut-down driver.
/// <para>
/// The dependency-consumer mux adapter is a SEPARATE child of this host and is unaffected by the
/// instance actor restarting, so it is deliberately not touched here.
/// </para>
/// </remarks>
/// <param name="msg">The restarted child's self-announcement.</param>
private void HandleReprovisionRequested(DriverInstanceActor.ReprovisionRequested msg) =>
ReprovisionChild(msg, Sender);
/// <summary>A reprovision held back by <see cref="_reprovisionBackoff"/>, carrying the incarnation it
/// was requested by so the identity check still applies when it is re-delivered by the timer.</summary>
/// <param name="Request">The original request.</param>
/// <param name="Child">The incarnation that asked; re-validated against <c>_children</c> on delivery,
/// so a driver restarted or removed in the meantime discards it.</param>
private sealed record ReprovisionDeferred(
DriverInstanceActor.ReprovisionRequested Request, IActorRef Child);
private static string ReprovisionTimerKey(string driverInstanceId) => "reprovision-" + driverInstanceId;
private void ReprovisionChild(DriverInstanceActor.ReprovisionRequested msg, IActorRef child)
{
// Already stopped (a restart racing a reconcile that removed the driver) — nothing to provision.
if (!_children.TryGetValue(msg.DriverInstanceId, out var entry)) return;
// Route by INCARNATION, not just by id. This message is delivered asynchronously, so one can still
// be in flight from an OLD incarnation after HandleRestartDriver stopped that child and spawned a
// replacement under the same DriverInstanceId. Delivered to the new child — which is in Connecting,
// where InitializeRequested IS handled — it would start a second, concurrent InitializeAsync,
// racing the dispose-before-overwrite that #527's teardown exists to serialise
// (ModbusDriver._transport, MqttDriver._connection each leak a transport on that race).
if (!Equals(child, entry.Actor))
{
_log.Debug(
"DriverHost {Node}: dropping a reprovision for {Id} from a superseded incarnation ({Stale})",
_localNode, msg.DriverInstanceId, child);
return;
}
// Rate-limit per driver (review F3). DEFERRED, never dropped: the last request must still be
// honoured, or a child whose final crash fell inside the window is left un-provisioned forever —
// exactly the 03/S16 failure this hook exists to close. One timer key per driver, so a burst
// collapses to a single pending reprovision rather than a queue of them.
var now = DateTime.UtcNow;
if (_lastReprovisionUtc.TryGetValue(msg.DriverInstanceId, out var last) &&
now - last < _reprovisionBackoff)
{
var delay = _reprovisionBackoff - (now - last);
_log.Warning(
"DriverHost {Node}: driver child {Id} was restarted by supervision again after {Reason} " +
"within {Backoff} — deferring the re-provision by {Delay} (crash loop?)",
_localNode, msg.DriverInstanceId, msg.Reason, _reprovisionBackoff, delay);
Timers.StartSingleTimer(
ReprovisionTimerKey(msg.DriverInstanceId), new ReprovisionDeferred(msg, child), delay);
return;
}
_lastReprovisionUtc[msg.DriverInstanceId] = now;
// A crash that Akka silently restarted is otherwise invisible to operators; name it.
_log.Warning(
"DriverHost {Node}: driver child {Id} was restarted by supervision after {Reason} — re-provisioning",
_localNode, msg.DriverInstanceId, msg.Reason);
// A stub child has no real driver to initialise (its Stubbed behaviour no-ops InitializeRequested
// anyway), but it still stores a desired set, so only the initialise is conditional.
if (!entry.Stubbed) entry.Actor.Tell(new DriverInstanceActor.InitializeRequested(entry.Spec.DriverConfig));
SendDesiredSubscriptions(entry.Actor, msg.DriverInstanceId);
}
private void HandleReconnectDriver(ReconnectDriver msg)
{
// DPS broadcast — only act if this node hosts the requested instance.