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

Merge branch 'fix/r3-03-restart-resubscribe-watchdog' — R3-03: SpawnChild always sends the desired set (operator Restart + supervision restart re-provision, stale-alarm-set re-subscribe, incarnation check, 10 s backoff); watchdog grace = max(LastSuccessfulRead, connectedSince) (archreview 03/S16, S17, U6)

Browse Source
This commit is contained in:
Joseph Doherty
2026-08-20 10:40:11 -04:00
parent e475f7b4ab 6d0002889a
commit 851e4cf5e0
9 changed files with 1699 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs
+192 -21
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>
/// 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
@@ -344,9 +352,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"/>
@@ -401,6 +425,15 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
/// <see cref="DefaultApplyWedgeTimeout"/>).</summary>
private readonly TimeSpan _applyWedgeTimeout;
/// <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.
@@ -542,7 +575,8 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
IDriverPrimaryRoleSource? primaryRoleSource = null,
IHistorizationRoleView? historizationRoleView = null,
Func<CommonsNodeId, bool>? sharesHistorizationScopeWithLocal = null,
TimeSpan? applyWedgeTimeout = null) =>
TimeSpan? applyWedgeTimeout = null,
TimeSpan? reprovisionBackoff = 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
// mis-ordered argument is usually type-compatible and therefore compiles clean, then binds
@@ -554,7 +588,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
loggerFactory, scriptRootLogger, scriptedAlarmHostOverride, invokerFactory, driverMemberCountProvider,
deploymentArtifactCache, redundancyRoleView, replicationPeerHost, fetchAndCacheMode, artifactFetcher,
alarmStateStore, telemetryHub, primaryRoleSource, historizationRoleView,
sharesHistorizationScopeWithLocal, applyWedgeTimeout));
sharesHistorizationScopeWithLocal, applyWedgeTimeout, reprovisionBackoff));
/// <summary>Initializes a new DriverHostActor with the specified dependencies.</summary>
/// <param name="dbFactory">Database context factory for configuration database access.</param>
@@ -640,9 +674,11 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
IDriverPrimaryRoleSource? primaryRoleSource = null,
IHistorizationRoleView? historizationRoleView = null,
Func<CommonsNodeId, bool>? sharesHistorizationScopeWithLocal = null,
TimeSpan? applyWedgeTimeout = null)
TimeSpan? applyWedgeTimeout = null,
TimeSpan? reprovisionBackoff = null)
{
_applyWedgeTimeout = applyWedgeTimeout ?? DefaultApplyWedgeTimeout;
_reprovisionBackoff = reprovisionBackoff ?? DefaultReprovisionBackoff;
_historizationRoleView = historizationRoleView;
_sharesHistorizationScopeWithLocal = sharesHistorizationScopeWithLocal;
_telemetryHub = telemetryHub;
@@ -1076,6 +1112,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);
@@ -1131,6 +1169,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);
@@ -1713,6 +1753,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).
@@ -2652,23 +2694,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)
@@ -2703,11 +2747,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)
@@ -2781,6 +2848,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
@@ -2931,6 +3021,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.
src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs
+109 -8
View File
@@ -42,6 +42,25 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
/// restored → Good) — independently of alarm transitions, since a comms-lost driver emits no alarm
/// events. Fire-and-forget, mirroring <see cref="DeltaApplied"/>.</summary>
public sealed record ConnectivityChanged(string DriverInstanceId, bool Connected);
/// <summary>
/// Sent to the parent from <see cref="PostRestart"/> so the host re-provisions THIS incarnation
/// (archreview 03/S6, 03/S16).
/// </summary>
/// <remarks>
/// A supervision restart recreates the actor object in place: the desired subscription sets are gone
/// (they are instance fields), and <c>PreRestart</c> has already run <see cref="PostStop"/> — which
/// called <see cref="IDriver.ShutdownAsync"/> on the driver object the Props closure re-uses. Nothing
/// else re-sends <see cref="InitializeRequested"/>: its only sender is
/// <c>DriverHostActor.SpawnChild</c>, and a restart does NOT respawn (the child's
/// <see cref="IActorRef"/> is unchanged, so the host is not otherwise notified that a new incarnation
/// exists at all). Without this the restarted child sits in <c>Connecting</c> forever around a
/// shut-down driver, with nothing subscribed — silently, since Akka logs the fault but the recovery
/// is invisible.
/// </remarks>
/// <param name="DriverInstanceId">The driver instance that was restarted.</param>
/// <param name="Reason">The type name of the exception that tripped supervision, for the host's log.</param>
public sealed record ReprovisionRequested(string DriverInstanceId, string Reason);
public sealed record ApplyDelta(string DriverConfigJson, CorrelationId Correlation);
public sealed record ApplyResult(bool Success, string? Reason, CorrelationId Correlation);
/// <summary>
@@ -252,6 +271,15 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
/// idempotent across repeated <see cref="SetDesiredSubscriptions"/> pushes.</summary>
private IAlarmSubscriptionHandle? _alarmSubscriptionHandle;
/// <summary>The alarm references the live <see cref="_alarmSubscriptionHandle"/> was actually established
/// with, so a CHANGED desired set can be told apart from a repeat of the same one.
/// <para>Without this the subscription latched: both <see cref="SubscribeDesiredAlarms"/> and
/// <see cref="HandleSubscribeAlarmsAsync"/> gate on "a handle already exists", which is the right guard
/// for an idempotent re-push and the WRONG one for a deploy that widens or narrows the alarm set —
/// drivers that filter on <c>sourceNodeIds</c> (OPC UA Client, AB CIP, FOCAS) would then never deliver a
/// newly-added alarm for the life of the connection. Empty whenever no handle is held.</para></summary>
private IReadOnlyList<string> _subscribedAlarmRefs = Array.Empty<string>();
/// <summary>
/// Gets or sets the timer scheduler for scheduling reconnection attempts.
/// </summary>
@@ -375,6 +403,20 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
Timers.StartPeriodicTimer("health-poll", HealthPollTick.Instance, _healthPollInterval);
}
/// <inheritdoc />
/// <remarks>
/// Ask the host to re-provision this incarnation — see <see cref="ReprovisionRequested"/> for why a
/// supervision restart otherwise leaves a dead child. <c>base.PostRestart</c> runs first because it
/// runs <see cref="PreStart"/>, which re-attaches the rediscovery / host-status sources and restarts
/// the health timer; the reprovision is the part <see cref="PreStart"/> cannot do, since only the
/// host knows this driver's config and desired set.
/// </remarks>
protected override void PostRestart(Exception reason)
{
base.PostRestart(reason);
Context.Parent.Tell(new ReprovisionRequested(_driverInstanceId, reason.GetType().Name));
}
private void Stubbed()
{
// Stubbed drivers accept the standard message contracts but return deterministic
@@ -488,10 +530,14 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
// Native-alarm analogue: un-gate the IAlarmSource feed when alarm tags are (now) present. The
// common live path — a deploy delivers SetDesiredSubscriptions while the driver is already
// Connected — flows through HERE, so the alarm subscribe must happen on this message, not only
// on Connected entry. Unlike the value path above, there is deliberately no unsubscribe-on-empty:
// a removed alarm tag's condition node is torn down on the address-space rebuild and
// DriverHostActor.ForwardNativeAlarm drops any transition whose ConditionId no longer maps, so a
// lingering session-less alarm subscription can never surface a removed alarm.
// on Connected entry. A CHANGED set re-subscribes (release-then-subscribe inside
// HandleSubscribeAlarmsAsync): unlike the value path, "a subscription already exists" is NOT
// sufficient, because several drivers filter their feed on the subscribed sourceNodeIds, so a
// widened set that never re-subscribed would silently never deliver the new alarm.
// Still deliberately no unsubscribe-on-EMPTY: a removed alarm tag's condition node is torn down
// on the address-space rebuild and DriverHostActor.ForwardNativeAlarm drops any transition whose
// ConditionId no longer maps, so a lingering session-less alarm subscription can never surface a
// removed alarm.
SubscribeDesiredAlarms();
});
ReceiveAsync<SubscribeAlarms>(HandleSubscribeAlarmsAsync);
@@ -561,6 +607,19 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
/// state on sustained failure: the nine poll drivers via Degraded, MQTT via its live connection
/// state, Galaxy via its reconnect supervisor.
/// </para>
/// <para>
/// <b>The grace is per CONNECTION (03/S17).</b> The silence baseline is
/// <c>max(LastSuccessfulRead, _connectedSinceUtc)</c>, not <c>LastSuccessfulRead ?? _connectedSinceUtc</c>.
/// Because <c>LastSuccessfulRead</c> is <em>preserved</em> across Degraded/Faulted writes (see above),
/// a device that accepts a connection but serves no data reconnects carrying a stamp that is already
/// older than the threshold, so a <c>??</c> baseline re-trips on the FIRST health tick of every new
/// connection — a reconnect cycle at health-tick cadence rather than the designed threshold. Taking
/// the later of the two gives each connection its own full window, which is exactly what re-stamping
/// <c>_connectedSinceUtc</c> on every Connected entry was for. Note this <b>paces</b> the cycle, it
/// does not end it: a permanently half-broken device still reconnects once per threshold, and a
/// driver that stops reading <i>within</i> a stable connection still trips on schedule (its read
/// stamp is the later term, exactly as before).
/// </para>
/// </remarks>
/// <param name="health">The health observed on this tick, or null when <see cref="IDriver.GetHealth"/>
/// threw — a failed observation is not evidence of comms loss, so it never advances the watchdog.</param>
@@ -570,8 +629,15 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
if (health is not { } observed) return;
if (observed.State == DriverState.Healthy) return;
// A driver that has never reported a successful read is measured from the current Connected entry.
var lastEvidence = observed.LastSuccessfulRead ?? _connectedSinceUtc;
// Measure silence from the LATER of the last successful read and THIS connect — the grace is per
// connection. `??` was wrong: drivers PRESERVE LastSuccessfulRead across Degraded/Faulted writes (that
// preservation is the #527 design), so after a trip + reconnect to a device that accepts the connection
// but still serves no data, the stamp is already older than the threshold and the FIRST tick of the new
// connection re-trips — churn at health-tick cadence instead of the designed grace. `??` only ever helped
// a driver that had never read at all.
var lastEvidence = observed.LastSuccessfulRead is { } read && read > _connectedSinceUtc
? read
: _connectedSinceUtc;
var silence = DateTime.UtcNow - lastEvidence;
if (silence < _commsLossTimeout) return;
@@ -1114,6 +1180,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
// (GalaxyDriver.SubscribeAlarmsAsync clears the set before adding), so handles no longer accumulate
// across reconnects. The gate (Count > 0) and the one-shot fan-out are unchanged.
_alarmSubscriptionHandle = null;
_subscribedAlarmRefs = Array.Empty<string>();
}
/// <summary>Establish the native-alarm subscription that un-gates an <see cref="IAlarmSource"/> driver's
@@ -1123,10 +1190,17 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
/// desired, and no subscription is yet established.</summary>
private void SubscribeDesiredAlarms()
{
if (_driver is IAlarmSource && _desiredAlarmRefs.Count > 0 && _alarmSubscriptionHandle is null)
if (_driver is IAlarmSource && _desiredAlarmRefs.Count > 0 && AlarmSubscriptionIsStale())
Self.Tell(new SubscribeAlarms());
}
/// <summary>True when the desired alarm refs are not the ones the live handle was established with —
/// either because there is no handle yet, or because a deploy changed the set. The second case is the
/// one a bare "handle is null" check misses; see <see cref="_subscribedAlarmRefs"/>.</summary>
private bool AlarmSubscriptionIsStale() =>
_alarmSubscriptionHandle is null ||
!_subscribedAlarmRefs.SequenceEqual(_desiredAlarmRefs, StringComparer.Ordinal);
/// <summary>Calls the driver's <see cref="IAlarmSource.SubscribeAlarmsAsync"/> (bounded) to register the
/// alarm subscription that un-gates its native feed, caching the returned handle. Re-checks the guard
/// (the desired set may have cleared, or another SubscribeAlarms may have already established a handle,
@@ -1134,10 +1208,36 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
/// stays gated until then.</summary>
private async Task HandleSubscribeAlarmsAsync(SubscribeAlarms _)
{
if (_driver is not IAlarmSource src || _desiredAlarmRefs.Count == 0 || _alarmSubscriptionHandle is not null)
if (_driver is not IAlarmSource src || _desiredAlarmRefs.Count == 0 || !AlarmSubscriptionIsStale())
return;
var refs = _desiredAlarmRefs;
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10));
// A handle held against a DIFFERENT set must be released before re-subscribing, or a
// sourceNodeIds-filtering driver accumulates two overlapping subscriptions. Best-effort and
// deliberately non-fatal: the re-subscribe below is what actually matters, and a driver whose
// unsubscribe throws must not be left on the stale set because of it. Cleared FIRST so a throw
// cannot leave a handle that the staleness check would keep treating as live.
if (_alarmSubscriptionHandle is { } stale)
{
_alarmSubscriptionHandle = null;
_subscribedAlarmRefs = Array.Empty<string>();
try
{
await _invoker.ExecuteAsync(
DriverCapability.AlarmSubscribe,
_driverInstanceId,
async ct => await src.UnsubscribeAlarmsAsync(stale, ct),
cts.Token);
}
catch (Exception ex)
{
_log.Warning(ex,
"DriverInstance {Id}: releasing the previous native-alarm subscription ({Diag}) threw; re-subscribing anyway",
_driverInstanceId, stale.DiagnosticId);
}
}
try
{
_alarmSubscriptionHandle = await _invoker.ExecuteAsync(
@@ -1145,6 +1245,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
_driverInstanceId,
async ct => await src.SubscribeAlarmsAsync(refs, ct),
cts.Token);
_subscribedAlarmRefs = refs;
_log.Info("DriverInstance {Id}: native-alarm subscription established for {Count} alarm ref(s) ({Diag})",
_driverInstanceId, refs.Count, _alarmSubscriptionHandle.DiagnosticId);
}