diff --git a/CLAUDE.md b/CLAUDE.md index e72a1e37..39254018 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -210,6 +210,38 @@ nothing ever writes a `DriverHostStatus` row (the table was re-created in the v3 confirm intent before deleting). The separable `OnHostStatusChanged` event is the useful half. Tracked as Gitea **#521**. +### Comms-loss detection: `GetHealth()` is load-bearing (Gitea #527, closed) + +Until #527, **`DriverInstanceActor.DisconnectObserved` had no production sender.** The only paths into +`Reconnecting` were a failed *initial* connect and an operator's Force Reconnect, so a driver that connected +and then lost its device stayed `Connected` indefinitely — publishing Bad for every tag, never +re-initialising, and **not recovering when the device returned** (live-confirmed: a Modbus driver against a +stopped fixture needed a node restart). `PublishHealthSnapshot` read `GetHealth()` for *reporting only*. + +The `Connected` health tick now runs a watchdog that self-sends `DisconnectObserved` when a driver reports a +non-`Healthy` state **and** its `LastSuccessfulRead` has been frozen for 90 s. Things to know before touching it: + +- ⚠️ **The conjunction is the design — "non-Healthy for N polls" is wrong and was the issue's own proposal.** + `Degraded` means *connected but some data is bad*, and several drivers write health **per tag inside the + read loop** (`ModbusDriver.ReadAsync`), so one unreadable address latches `Degraded` forever against a + perfectly reachable device. State-only tripping would reconnect that driver every 90 s indefinitely. + `LastSuccessfulRead` is the honest evidence — preserved across `Degraded`/`Faulted` writes, advanced by any + success. Each conjunct has a test that goes red when the *other* is deleted. +- **The mechanism rests on every driver reaching a non-Healthy state on sustained failure.** Surveyed across + all 12: nine poll drivers via `Degraded`, MQTT via live connection state, Galaxy via its reconnect + supervisor. `CalculationDriver` reports `Healthy` while running and so can never trip — correct, as it has + no transport. Re-check this if a new driver lands. +- ⚠️ **The reconnect retry MUST tear down before re-initialising.** `ModbusDriver` (`_transport`) and + `MqttDriver` (`_connection`) overwrite an `IAsyncDisposable` field inside `InitializeAsync` without + disposing the old one, so re-initialising without `ShutdownAsync` leaks a socket per attempt — ~360 for an + hour-long outage at the 10 s retry interval. Latent while only a manual Force Reconnect reached this path. +- The threshold is hardcoded (`DefaultCommsLossTimeout`), matching `HealthPollInterval` / + `DefaultReconnectInterval`. Unlike the #523 drift detector this generates **no** traffic on a healthy + system — it is a local `GetHealth()` read, and only reaches the network once a driver has already stopped + reading. + +See `docs/DriverLifecycle.md` §"The comms-loss watchdog". + ## mxaccessgw The gateway lives in a sibling repo at `~/Desktop/MxAccessGateway/` (`mxaccessgw` — see [Related Projects](#related-projects-scadaproj-umbrella)). See `docs/v2/Galaxy.ParityRig.md` for the gw setup recipe (build, API key provisioning via `apikey create-key`, env-var overrides for HTTP/2 cleartext + worker path). The gw's MXAccess Toolkit reference (its `gateway.md`) is the canonical MxAccess API doc; the standalone `mxaccess_documentation.md` previously kept in this repo retired in PR 7.3. diff --git a/docs/AlarmTracking.md b/docs/AlarmTracking.md index 2dd68cfd..ce776a35 100644 --- a/docs/AlarmTracking.md +++ b/docs/AlarmTracking.md @@ -183,15 +183,22 @@ Two things worth knowing before changing this: and is the correct reading of "this input's source is unreachable", but it is a behaviour change beyond scripted alarms. -⚠️ **The signal this hangs off does not currently fire for a mid-run device loss — see #527.** The #481 live -gate on docker-dev found that `DriverInstanceActor.DisconnectObserved` has **no production sender**: the only -paths into `Reconnecting` are a failed *initial* connect and an operator Force Reconnect. A Modbus driver -whose fixture was stopped stayed `Connected` for >10 minutes publishing per-tag Bad, and did not recover when -the fixture came back until the node was restarted. So the fan-out above — and the native `#477` Layer 2 -annotation it mirrors — are correctly wired to a signal that a dying device never raises. What partially -covers the gap today is that Modbus publishes Bad *per tag* rather than going silent, so #478 Layer 3 does -carry Bad into scripted input quality for that driver; that is a per-driver accident, not a guarantee, and it -does nothing for native conditions or for recovery. Closing #527 is what makes this path reachable. +**The signal this hangs off now fires for a mid-run device loss (#527, closed).** The #481 live gate on +docker-dev found that `DriverInstanceActor.DisconnectObserved` had **no production sender**: the only paths +into `Reconnecting` were a failed *initial* connect and an operator Force Reconnect. A Modbus driver whose +fixture was stopped stayed `Connected` for >10 minutes publishing per-tag Bad, and did not recover when the +fixture came back until the node was restarted. So the fan-out above — and the native #477 Layer 2 +annotation it mirrors — were correctly wired to a signal that a dying device never raised. #527 added the +comms-loss watchdog on the Connected health tick, which self-sends `DisconnectObserved` once a driver has +gone 90 s without a successful read *while* reporting itself non-Healthy. See +[`DriverLifecycle.md`](DriverLifecycle.md) §"The comms-loss watchdog" for why the trip condition is that +conjunction and not simply "non-Healthy for N polls" — the naive rule would reconnect a healthy device's +driver every 90 s forever whenever one authored address is unreadable. + +Note the partial cover that existed before, because it explains why *some* of this appeared to work: Modbus +publishes Bad *per tag* rather than going silent, so #478 Layer 3 did carry Bad into scripted input quality +for that driver. That was a per-driver accident, not a guarantee, and it did nothing for native conditions +or for recovery. **Cold start vs. a driver reporting Bad (#478 Finding 3, closed by #481).** `WorstInputStatus` used to skip every input whose *value* was null as "no data yet". A comms-lost driver produces exactly that shape — Bad diff --git a/docs/DriverLifecycle.md b/docs/DriverLifecycle.md index 5bfd6186..d92babec 100644 --- a/docs/DriverLifecycle.md +++ b/docs/DriverLifecycle.md @@ -219,6 +219,57 @@ So a driver whose `InitializeAsync` disposes anything on its failure path must b `MTConnectDriver` does today, because it is the only one that tears down inside a `catch`. The shutdown paths deliberately keep the single trailing write. +### The comms-loss watchdog (Gitea #527) — `GetHealth()` is now load-bearing + +`GetHealth()` used to be **reporting-only**: `DriverInstanceActor.PublishHealthSnapshot` read it every 30 s +and forwarded it to the AdminUI, and nothing ever acted on a driver that reported itself unhealthy. That +left `DisconnectObserved` with **no production sender at all** — the only paths into `Reconnecting` were a +failed *initial* connect and an operator's Force Reconnect. A driver that connected and then lost its +device stayed `Connected` forever: publishing Bad for every tag, never re-initialising, and **not +recovering when the device came back** (live-confirmed against a stopped Modbus fixture, which needed a +node restart). Both comms-loss quality paths — #477's native-condition annotation and #481's +scripted-alarm input fan-out — hang off `ConnectivityChanged`, so both were wired to a signal that never +fired. + +The `Connected`-state health tick now also evaluates a watchdog, and self-sends `DisconnectObserved` when +**both** of these hold: + +1. the driver reports a state other than `Healthy`, **and** +2. `DriverHealth.LastSuccessfulRead` has not advanced for `DefaultCommsLossTimeout` (90 s = three ticks), + measured from the Connected entry when the driver has never reported a successful read. + +**The conjunction is the whole design, and "non-Healthy for N polls" is the wrong rule.** `Degraded` means +*connected but some data is bad* — the enum says so. Several drivers publish health **per tag inside their +read loop** (`ModbusDriver.ReadAsync` writes `Healthy` or `Degraded` once per reference), so a single +unreadable address leaves the driver latched `Degraded` whenever it happens to be read last, indefinitely, +against a device that is perfectly reachable. A state-only watchdog would tear that driver down every 90 s +forever — strictly worse than the bug it fixes. `LastSuccessfulRead` is the honest liveness evidence: every +driver advances it on any successful read and *preserves* it across `Degraded`/`Faulted` writes, so one bad +tag among good ones keeps it moving while a device that is genuinely gone freezes it. The state check is +still needed as the second conjunct, because a `Healthy` driver with nothing subscribed legitimately never +reads and must not be declared dead for being idle. Both halves are pinned by their own test in +`DriverInstanceActorCommsLossWatchdogTests`, each of which goes red when the *other* conjunct is deleted. + +Two consequences worth knowing: + +- **`CalculationDriver` can never trip it, correctly.** It reports `Healthy` whenever it is running and has + no transport to be degraded about. Every other driver reaches a non-Healthy state on sustained failure: + the nine poll drivers via `Degraded`, MQTT via its live connection state, Galaxy via its reconnect + supervisor. This was surveyed across all 12 before relying on it — it is the assumption the whole + mechanism rests on. +- **A device that accepts a connection but serves no data will cycle**, reconnecting once per threshold. + That is the correct reading of a broken device, it is rate-limited, and it is visible in the health + snapshot's error count — but do not mistake it for a bug when reading logs. + +**The reconnect retry now tears down before re-initialising.** `Reconnecting`'s `RetryConnect` calls +`ShutdownAsync` before `InitializeAsync` — the sequence every driver's own `ReinitializeAsync` already +uses — because several drivers **overwrite** an `IAsyncDisposable` transport field inside `InitializeAsync` +without disposing the previous one (`ModbusDriver._transport`, `MqttDriver._connection`). Re-initialising +without a teardown leaks one socket per attempt. That was latent while only a manual Force Reconnect +re-entered this path; the watchdog makes it automatic and unbounded, so a device down for an hour would +have leaked ~360 transports at the 10 s retry interval. The shutdown is best-effort: a driver that throws +on it still gets its initialise attempt. + --- ## IDriverConfigEditor — custom AdminUI config editor (plug-point) diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs index 7a2c60ce..a55ac0f3 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs @@ -141,6 +141,23 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers /// Interval between periodic health-poll heartbeats sent to the snapshot store. public static readonly TimeSpan HealthPollInterval = TimeSpan.FromSeconds(30); + /// + /// How long a Connected driver may go without a single successful read — while also reporting + /// itself non-Healthy — before the actor declares comms lost (Gitea #527). Evaluated on the + /// existing tick, so the effective detection granularity is + /// 30 s and the worst-case detection latency is this value plus one tick. + /// + /// + /// 90 s = three health-poll ticks. Long enough that a device answering slowly under load is not + /// torn down, short enough that an operator watching a dead PLC sees it inside two minutes. + /// Hardcoded rather than exposed in appsettings, matching and + /// . Unlike the periodic drift-detector removed by Gitea + /// #523, this generates no outbound traffic on a healthy system: the watchdog is a local + /// read of , and it only reaches the network once a driver has + /// already stopped reading successfully. + /// + public static readonly TimeSpan DefaultCommsLossTimeout = TimeSpan.FromSeconds(90); + private readonly IDriver _driver; /// Phase 6.1 resilience seam: every guarded driver-capability call this actor makes is @@ -158,6 +175,14 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers private readonly TimeSpan _healthPollInterval; + /// Comms-loss watchdog threshold (#527); or negative disables it. + private readonly TimeSpan _commsLossTimeout; + + /// When the actor last entered Connected. The comms-loss watchdog's baseline for a + /// driver that has never reported a successful read, so "connected but never read anything" is + /// measurable rather than indefinitely undefined. + private DateTime _connectedSinceUtc; + private readonly ILoggingAdapter _log = Context.GetLogger(); private string? _currentConfigJson; @@ -229,6 +254,9 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers /// Optional period of the health-poll heartbeat, which also drives the /// subscription reconcile (); defaults to /// . Exists so a test can drive the reconcile without waiting 30 s. + /// Optional comms-loss watchdog threshold (#527); defaults to + /// . disables the watchdog — used by + /// tests that need a Connected driver to stay Connected while reporting non-Healthy. /// A instance configured to create the wrapped . public static Props Props( IDriver driver, @@ -237,7 +265,8 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers IDriverHealthPublisher? healthPublisher = null, string? clusterId = null, IDriverCapabilityInvoker? invoker = null, - TimeSpan? healthPollInterval = null) => + TimeSpan? healthPollInterval = null, + TimeSpan? commsLossTimeout = null) => Akka.Actor.Props.Create(() => new DriverInstanceActor( driver, reconnectInterval ?? DefaultReconnectInterval, @@ -245,7 +274,8 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers healthPublisher ?? NullDriverHealthPublisher.Instance, clusterId ?? string.Empty, invoker, - healthPollInterval)); + healthPollInterval, + commsLossTimeout)); /// /// Returns true when the driver should boot in DEV-STUB mode based on host platform and @@ -278,6 +308,8 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers /// defaults to (pass-through) when null. /// Period of the health-poll heartbeat, which also drives the /// subscription reconcile; defaults to when null. + /// Comms-loss watchdog threshold (#527); defaults to + /// when null, and is disabled when zero or negative. public DriverInstanceActor( IDriver driver, TimeSpan reconnectInterval, @@ -285,7 +317,8 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers IDriverHealthPublisher? healthPublisher = null, string? clusterId = null, IDriverCapabilityInvoker? invoker = null, - TimeSpan? healthPollInterval = null) + TimeSpan? healthPollInterval = null, + TimeSpan? commsLossTimeout = null) { _driver = driver; _invoker = invoker ?? NullDriverCapabilityInvoker.Instance; @@ -294,6 +327,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers _healthPublisher = healthPublisher ?? NullDriverHealthPublisher.Instance; _reconnectInterval = reconnectInterval; _healthPollInterval = healthPollInterval ?? HealthPollInterval; + _commsLossTimeout = commsLossTimeout ?? DefaultCommsLossTimeout; OtOpcUaTelemetry.DriverInstanceLifecycle.Add(1, new KeyValuePair("event", startStubbed ? "spawn_stub" : "spawn"), new KeyValuePair("driver_type", driver.DriverType)); @@ -403,6 +437,10 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers // non-Good "waiting for initial data" quality at materialise, and this is what confirms them Good. Context.Parent.Tell(new ConnectivityChanged(_driverInstanceId, Connected: true)); + // #527 — baseline for the comms-loss watchdog. Re-stamped on every Connected entry so a driver that + // reconnects but never manages a read is measured from THIS connect, not from a stale earlier one. + _connectedSinceUtc = DateTime.UtcNow; + ReceiveAsync(HandleApplyDeltaAsync); Receive(msg => { @@ -456,11 +494,76 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers Receive(_ => PublishHealthSnapshot()); Receive(_ => { - PublishHealthSnapshot(); + // One GetHealth observation feeds both the snapshot publish and the watchdog — reading it + // twice would let the two disagree about the same tick. + var health = PublishHealthSnapshot(); + EvaluateCommsLossWatchdog(health); ReconcileSubscription(); }); } + /// + /// Gitea #527 — the comms-loss watchdog. Declares a Connected driver disconnected once it has gone + /// without a single successful read while also reporting + /// itself non-Healthy, by self-sending . + /// + /// + /// + /// Why this exists. had no production sender at all. The only + /// ways into Reconnecting were a failed initial connect and an operator's Force + /// Reconnect, so a driver that connected successfully and then lost its device stayed Connected + /// forever: it published Bad for every tag, never re-initialised, and did not recover when the device + /// came back — live-confirmed against a stopped Modbus fixture, which needed a node restart. Both + /// comms-loss quality paths (#477 native conditions, #481 scripted-alarm inputs) hang off + /// , so both were wired to a signal that never fired. + /// OpcUaClientDriver's own reconnect-exhausted branch already assumed this mechanism existed + /// ("the Core never learns it must offer an operator reinitialize") — it did not. + /// + /// + /// Why the trip condition is a conjunction, and NOT simply "non-Healthy for N polls". + /// means connected but some data is bad — the enum says + /// so. Several drivers publish health per-tag inside their read loop (ModbusDriver.ReadAsync + /// writes Healthy or Degraded once per reference), so a single unreadable address leaves the driver + /// Degraded whenever it happens to be the last tag read, indefinitely, against a perfectly healthy + /// device. Tripping on state alone would tear that driver down every watchdog period forever — + /// strictly worse than the bug being fixed. + /// + /// + /// is the honest liveness evidence: every driver + /// advances it on any successful read and preserves it across Degraded/Faulted writes. One + /// bad tag among good ones keeps it moving; a device that is genuinely gone freezes it. The state + /// check is still required as the second conjunct, because a Healthy driver with nothing subscribed + /// legitimately never reads at all and must not be declared dead for being idle. + /// + /// + /// Drivers this cannot fire for, by construction. CalculationDriver reports Healthy + /// whenever it is running (it has no transport to be degraded about), so it never trips — correct, + /// since there is nothing to reconnect to. Every other driver in the tree reaches a non-Healthy + /// state on sustained failure: the nine poll drivers via Degraded, MQTT via its live connection + /// state, Galaxy via its reconnect supervisor. + /// + /// + /// The health observed on this tick, or null when + /// threw — a failed observation is not evidence of comms loss, so it never advances the watchdog. + private void EvaluateCommsLossWatchdog(DriverHealth? health) + { + if (_commsLossTimeout <= TimeSpan.Zero) return; + 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; + var silence = DateTime.UtcNow - lastEvidence; + if (silence < _commsLossTimeout) return; + + _log.Warning( + "DriverInstance {Id}: no successful read for {Silence} while reporting {State} ({Error}) — " + + "declaring comms lost and reconnecting", + _driverInstanceId, silence, observed.State, observed.LastError ?? "no error reported"); + Self.Tell(new DisconnectObserved( + $"no successful read for {silence.TotalSeconds:F0}s while reporting {observed.State}")); + } + /// /// Periodic desired-vs-actual subscription reconcile (#488), riding the existing health-poll /// tick. While Connected, a driver that has a desired subscription set but no live handle is @@ -514,7 +617,8 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers // (a comms-lost driver emits no alarm events, so this is the ONLY signal that the source is unreachable). Context.Parent.Tell(new ConnectivityChanged(_driverInstanceId, Connected: false)); - Receive(_ => InitializeAsync(_currentConfigJson ?? "{}")); + // #527 — tear the dead transport down before re-initialising, or every retry leaks one. + Receive(_ => InitializeAsync(_currentConfigJson ?? "{}", teardownFirst: true)); // Fast-fail writes while reconnecting (same reason as Connecting — avoids the 8s host Ask // timeout on an inbound write to a transiently-down driver). Synchronous Receive. Receive(_ => @@ -562,16 +666,45 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers Timers.StartPeriodicTimer("retry-connect", RetryConnect.Instance, _reconnectInterval); } - private void InitializeAsync(string driverConfigJson) + /// The config blob to initialise the driver with. + /// + /// When true, is awaited before + /// — the sequence every driver's own ReinitializeAsync uses. Set on the Reconnecting + /// retry path (#527), because several drivers overwrite their transport field in + /// InitializeAsync without disposing the previous one: ModbusDriver assigns + /// _transport = _transportFactory(...) and MqttDriver assigns _connection, both + /// over an . Re-initialising without tearing down therefore leaks one + /// socket (and, for Modbus, a poll engine) per attempt. That was latent while nothing but an + /// operator's Force Reconnect ever re-entered this path; the watchdog makes it automatic and + /// unbounded — a device down for an hour would leak ~360 transports at the 10 s retry interval. + /// Best-effort and never fatal: a driver that throws on shutdown still gets its initialise attempt. + /// + private void InitializeAsync(string driverConfigJson, bool teardownFirst = false) { _currentConfigJson = driverConfigJson; var generation = ++_initGeneration; var self = Self; + var log = _log; + var driverInstanceId = _driverInstanceId; + var driver = _driver; _ = Task.Run(async () => { try { - await _driver.InitializeAsync(driverConfigJson, CancellationToken.None); + if (teardownFirst) + { + try + { + await driver.ShutdownAsync(CancellationToken.None); + } + catch (Exception ex) + { + log.Warning(ex, + "DriverInstance {Id}: ShutdownAsync threw before a reconnect attempt; " + + "initialising anyway", driverInstanceId); + } + } + await driver.InitializeAsync(driverConfigJson, CancellationToken.None); self.Tell(new InitializeSucceeded(generation)); } catch (Exception ex) @@ -1006,11 +1139,16 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers /// every 30s when nothing has changed. Newly-joined SignalR clients still get the current /// snapshot via DriverStatusHub.JoinDriver which reads the store directly. /// - private void PublishHealthSnapshot() + /// The health observed on this call, or null when threw. + /// Returned so the caller can reuse the same observation — the #527 comms-loss watchdog rides this + /// call rather than polling the driver a second time and risking a different answer. + private DriverHealth? PublishHealthSnapshot() { + DriverHealth? observed = null; try { var health = _driver.GetHealth(); + observed = health; var errorCount = ErrorCount5Min(); var hostStatuses = CurrentHostStatuses(); // _rediscoveryNeededUtc is PART OF THE FINGERPRINT on purpose. A rediscovery raise on an @@ -1027,7 +1165,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers var fingerprint = (health.State, health.LastSuccessfulRead, health.LastError, errorCount, _rediscoveryNeededUtc, HostStatusDigest(hostStatuses)); if (_lastPublishedFingerprint is { } prev && prev.Equals(fingerprint)) - return; + return observed; _lastPublishedFingerprint = fingerprint; _healthPublisher.Publish( _clusterId, _driverInstanceId, health, errorCount, _rediscoveryNeededUtc, _rediscoveryReason, @@ -1037,6 +1175,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers { _log.Warning(ex, "DriverInstance {Id}: GetHealth threw during health publish; skipping", _driverInstanceId); } + return observed; } /// Order-insensitive value digest of a host-status list for the publish fingerprint. Null in, diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorCommsLossWatchdogTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorCommsLossWatchdogTests.cs new file mode 100644 index 00000000..f12f3b54 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorCommsLossWatchdogTests.cs @@ -0,0 +1,330 @@ +using Akka.Actor; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Runtime.Drivers; +using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers; + +/// +/// Gitea #527 — the comms-loss watchdog that gives +/// its first production sender. +/// +/// +/// +/// The defect. Nothing in src/ ever sent DisconnectObserved. A driver that +/// connected and then lost its device stayed Connected indefinitely — publishing Bad for +/// every tag, never re-initialising, and not recovering when the device returned. Live-confirmed +/// against a stopped Modbus fixture, which needed a node restart. +/// +/// +/// What these tests are really pinning is the trip CONDITION, not the plumbing. The obvious +/// design — "non-Healthy for N consecutive polls" — is wrong, and +/// is the test that +/// says so. means connected but some data is bad, and +/// several drivers write health per-tag inside their read loop, so one unreadable address can leave +/// a perfectly healthy device's driver Degraded forever. Tripping on state alone would reconnect +/// that driver every 90 s in production — worse than the bug being fixed. The conjunction with a +/// frozen is what makes the signal honest, and +/// pins the other half. +/// +/// +/// Every absence assertion carries a positive control — a proven number of health-poll ticks +/// actually processed () — so "the watchdog +/// did not fire" can never pass merely because nothing ran. +/// +/// +public sealed class DriverInstanceActorCommsLossWatchdogTests : RuntimeActorTestBase +{ + private static readonly TimeSpan FastPoll = TimeSpan.FromMilliseconds(50); + private static readonly TimeSpan FastTimeout = TimeSpan.FromMilliseconds(250); + + /// Drives the actor to Connected and returns once the connect has been announced. + private (IActorRef Actor, Akka.TestKit.TestProbe Parent) ConnectDriver( + HealthControlledStubDriver driver, + TimeSpan? commsLossTimeout = null, + TimeSpan? reconnectInterval = null) + { + var parent = CreateTestProbe(); + var actor = parent.ChildActorOf(DriverInstanceActor.Props( + driver, + reconnectInterval: reconnectInterval ?? TimeSpan.FromSeconds(30), + healthPollInterval: FastPoll, + commsLossTimeout: commsLossTimeout ?? FastTimeout)); + + actor.Tell(new DriverInstanceActor.InitializeRequested("{}")); + // Connected entry announces Connected=true; consuming it here leaves the probe holding only + // whatever the watchdog does (or does not) send next. + parent.ExpectMsg(TimeSpan.FromSeconds(3)) + .Connected.ShouldBeTrue(); + return (actor, parent); + } + + /// + /// The core of #527: a Connected driver whose device has gone away — non-Healthy, and no successful + /// read since well before the threshold — is declared disconnected and announces comms loss to the + /// host, which is the signal #477 and #481 both hang off. + /// + [Fact] + public void A_connected_driver_that_stops_reading_successfully_is_declared_disconnected() + { + var driver = new HealthControlledStubDriver + { + // The live shape: reads fail, the driver reports Degraded, and LastSuccessfulRead is frozen + // at the instant the device died. + Health = new DriverHealth(DriverState.Degraded, DateTime.UtcNow.AddMinutes(-5), "socket timeout"), + }; + var (_, parent) = ConnectDriver(driver); + + parent.ExpectMsg(TimeSpan.FromSeconds(3)) + .Connected.ShouldBeFalse(); + } + + /// + /// The falsifying control for the trip condition, and the reason it is not "non-Healthy for N polls". + /// This driver is permanently Degraded — modelling one unreadable address among many, which + /// ModbusDriver.ReadAsync leaves latched whenever the bad tag is read last — but it is still + /// reading successfully, so LastSuccessfulRead keeps advancing. It must be left alone. + /// A state-only watchdog would tear this driver down on every threshold, forever. + /// + [Fact] + public void A_degraded_driver_that_is_still_reading_successfully_is_left_alone() + { + var driver = new HealthControlledStubDriver + { + // Recomputed per call: Degraded, but LastSuccessfulRead is always fresh. + HealthFactory = () => new DriverHealth(DriverState.Degraded, DateTime.UtcNow, "one bad tag"), + }; + var (_, parent) = ConnectDriver(driver); + + // Positive control: prove the watchdog genuinely ran, many times, over a span far longer than the + // trip threshold. Without this the absence assertion below would pass against a watchdog that + // never executed at all. + AwaitCondition( + () => driver.HealthReadCount >= 12, + TimeSpan.FromSeconds(5), + "health-poll ticks did not run — the absence assertion below would be vacuous"); + driver.ObservedSpan.ShouldBeGreaterThan(FastTimeout, "the observation window never exceeded the trip threshold"); + + parent.ExpectNoMsg(TimeSpan.FromMilliseconds(200)); + } + + /// + /// The other conjunct. A Healthy driver with nothing subscribed legitimately never reads, so its + /// LastSuccessfulRead is arbitrarily stale. Staleness alone must not declare it dead. + /// + [Fact] + public void An_idle_healthy_driver_with_a_stale_last_read_is_left_alone() + { + var driver = new HealthControlledStubDriver + { + Health = new DriverHealth(DriverState.Healthy, DateTime.UtcNow.AddHours(-2), null), + }; + var (_, parent) = ConnectDriver(driver); + + AwaitCondition( + () => driver.HealthReadCount >= 12, + TimeSpan.FromSeconds(5), + "health-poll ticks did not run — the absence assertion below would be vacuous"); + + parent.ExpectNoMsg(TimeSpan.FromMilliseconds(200)); + } + + /// + /// A driver that connects but never manages a single read reports LastSuccessfulRead = null, + /// so staleness has no natural baseline. The watchdog measures from the Connected entry instead — + /// otherwise a never-read driver would be the one case that could never be declared dead. + /// + [Fact] + public void A_driver_that_never_reads_after_connecting_is_declared_disconnected() + { + var driver = new HealthControlledStubDriver + { + Health = new DriverHealth(DriverState.Degraded, LastSuccessfulRead: null, "never read"), + }; + var (_, parent) = ConnectDriver(driver); + + parent.ExpectMsg(TimeSpan.FromSeconds(3)) + .Connected.ShouldBeFalse(); + } + + /// + /// A that throws is a failed observation, not evidence that the + /// device is gone. Treating it as a trip would let a buggy health accessor cycle a working driver. + /// + [Fact] + public void A_throwing_GetHealth_is_not_evidence_of_comms_loss() + { + var driver = new HealthControlledStubDriver { GetHealthShouldThrow = true }; + var (_, parent) = ConnectDriver(driver); + + AwaitCondition( + () => driver.HealthReadCount >= 12, + TimeSpan.FromSeconds(5), + "health-poll ticks did not run — the absence assertion below would be vacuous"); + + parent.ExpectNoMsg(TimeSpan.FromMilliseconds(200)); + } + + /// + /// #527's second acceptance criterion, end to end: the driver recovers by itself once the device + /// returns, with no node restart. Trip → Reconnecting → retry re-initialises → Connected, and the + /// host is told connectivity is back. + /// + [Fact] + public void The_driver_recovers_on_its_own_once_the_device_returns() + { + var driver = new HealthControlledStubDriver + { + Health = new DriverHealth(DriverState.Degraded, DateTime.UtcNow.AddMinutes(-5), "socket timeout"), + }; + var (_, parent) = ConnectDriver(driver, reconnectInterval: TimeSpan.FromMilliseconds(100)); + + parent.ExpectMsg(TimeSpan.FromSeconds(3)) + .Connected.ShouldBeFalse(); + + // The device comes back: the next initialise attempt succeeds and reads resume. + driver.Health = new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null); + + parent.ExpectMsg(TimeSpan.FromSeconds(5)) + .Connected.ShouldBeTrue(); + driver.InitializeCount.ShouldBeGreaterThan(1, "the driver was never re-initialised"); + } + + /// + /// The leak guard. Several drivers overwrite an transport field + /// inside InitializeAsync without disposing the previous one (ModbusDriver._transport, + /// MqttDriver._connection), so re-initialising without a teardown leaks one socket per + /// attempt. Latent while only an operator's Force Reconnect re-entered this path; the watchdog + /// makes it automatic and unbounded, so the retry must shut down first. + /// + [Fact] + public void A_reconnect_attempt_tears_the_dead_transport_down_before_re_initialising() + { + var driver = new HealthControlledStubDriver + { + Health = new DriverHealth(DriverState.Degraded, DateTime.UtcNow.AddMinutes(-5), "socket timeout"), + }; + var (_, parent) = ConnectDriver(driver, reconnectInterval: TimeSpan.FromMilliseconds(100)); + + parent.ExpectMsg(TimeSpan.FromSeconds(3)) + .Connected.ShouldBeFalse(); + + AwaitCondition( + () => driver.InitializeCount >= 2, + TimeSpan.FromSeconds(5), + "the reconnect retry never re-initialised the driver"); + + // Ordering, not just counts: EVERY re-initialise must be immediately preceded by a shutdown, since + // it is the initialise that overwrites the transport field. Asserting ShutdownCount > 0 alone would + // also pass on a teardown that happened afterwards, which frees nothing before the overwrite. + // + // The sequence is deliberately not pinned to a fixed length: this driver reports stale health even + // after a successful re-init, so the watchdog trips again and the cycle repeats — the real + // behaviour for a device that accepts a connection but serves no data. Every additional cycle is + // another chance for the invariant to be violated, so a longer run is a stronger test, not a + // flakier one. + var lifecycle = driver.Lifecycle; + lifecycle[0].ShouldBe("init", "the first connect must not tear down a driver that was never up"); + for (var i = 1; i < lifecycle.Count; i++) + { + if (lifecycle[i] != "init") continue; + lifecycle[i - 1].ShouldBe( + "shutdown", + $"re-initialise at index {i} was not preceded by a teardown — the previous transport leaks. " + + $"Full sequence: [{string.Join(", ", lifecycle)}]"); + } + } + + /// + /// The escape hatch is real: a zero threshold disables the watchdog entirely, leaving the + /// pre-#527 behaviour. Pinned because tests elsewhere rely on a Connected driver staying Connected + /// while reporting non-Healthy. + /// + [Fact] + public void A_zero_threshold_disables_the_watchdog() + { + var driver = new HealthControlledStubDriver + { + Health = new DriverHealth(DriverState.Degraded, DateTime.UtcNow.AddMinutes(-5), "socket timeout"), + }; + var (_, parent) = ConnectDriver(driver, commsLossTimeout: TimeSpan.Zero); + + AwaitCondition( + () => driver.HealthReadCount >= 12, + TimeSpan.FromSeconds(5), + "health-poll ticks did not run — the absence assertion below would be vacuous"); + + parent.ExpectNoMsg(TimeSpan.FromMilliseconds(200)); + } +} + +/// +/// A whose health is fully driven by the test, counting observations so an +/// absence assertion can prove the watchdog actually ran, and recording lifecycle call ORDER so the +/// teardown-before-reinitialise guard can assert sequence rather than mere counts. +/// +internal sealed class HealthControlledStubDriver : StubDriver +{ + private readonly Lock _gate = new(); + private int _healthReadCount; + private DateTime _firstReadUtc; + private DateTime _lastReadUtc; + + /// The health returned by every call; ignored when + /// is set. + public DriverHealth Health { get; set; } = new(DriverState.Healthy, DateTime.UtcNow, null); + + /// Per-call health, for modelling a driver whose LastSuccessfulRead keeps advancing. + public Func? HealthFactory { get; set; } + + /// When true, throws — a failed observation. + public bool GetHealthShouldThrow { get; set; } + + /// Number of calls; the positive control for absence assertions. + public int HealthReadCount => Volatile.Read(ref _healthReadCount); + + /// Wall-clock span between the first and last observation — proves an absence window + /// genuinely exceeded the trip threshold rather than merely counting fast ticks. + public TimeSpan ObservedSpan + { + get { lock (_gate) return _lastReadUtc - _firstReadUtc; } + } + + /// Ordered lifecycle log: "init" / "shutdown", in call order. + public IReadOnlyList Lifecycle + { + get { lock (_gate) return _lifecycle.ToArray(); } + } + + private readonly List _lifecycle = new(); + + /// + public override async Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken) + { + lock (_gate) _lifecycle.Add("init"); + await base.InitializeAsync(driverConfigJson, cancellationToken); + } + + /// + public override Task ShutdownAsync(CancellationToken cancellationToken) + { + lock (_gate) _lifecycle.Add("shutdown"); + return base.ShutdownAsync(cancellationToken); + } + + /// + public override DriverHealth GetHealth() + { + Interlocked.Increment(ref _healthReadCount); + lock (_gate) + { + var now = DateTime.UtcNow; + if (_firstReadUtc == default) _firstReadUtc = now; + _lastReadUtc = now; + } + if (GetHealthShouldThrow) throw new InvalidOperationException("stub-health-fail"); + return HealthFactory is { } factory ? factory() : Health; + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/StubDrivers.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/StubDrivers.cs index ea8c5c53..25d459f4 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/StubDrivers.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/StubDrivers.cs @@ -33,7 +33,7 @@ internal class StubDriver : IDriver /// Initializes the driver with the specified configuration JSON. /// The driver configuration JSON. /// Cancellation token for the operation. - public async Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken) + public virtual async Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken) { Interlocked.Increment(ref InitializeCount); lock (_initConfigsLock) InitConfigs.Add(driverConfigJson); @@ -56,13 +56,20 @@ internal class StubDriver : IDriver /// Shuts down the driver. /// Cancellation token for the operation. - public Task ShutdownAsync(CancellationToken cancellationToken) + public virtual Task ShutdownAsync(CancellationToken cancellationToken) { Interlocked.Increment(ref ShutdownCount); return Task.CompletedTask; } /// Gets the health status of the driver. - public DriverHealth GetHealth() => new(DriverState.Healthy, DateTime.UtcNow, null); + /// + /// Virtual so the #527 comms-loss watchdog suite can drive health directly. Note the + /// : this stub is permanently Healthy with an always-fresh + /// LastSuccessfulRead, so the watchdog can never fire for it — which is why every existing + /// test built on is unaffected by the watchdog, and equally why a watchdog + /// test built on it would be vacuous. + /// + public virtual DriverHealth GetHealth() => new(DriverState.Healthy, DateTime.UtcNow, null); } ///