diff --git a/archreview/plans/R3-03-restart-resubscribe-watchdog-plan.md b/archreview/plans/R3-03-restart-resubscribe-watchdog-plan.md index e872aefd..10648423 100644 --- a/archreview/plans/R3-03-restart-resubscribe-watchdog-plan.md +++ b/archreview/plans/R3-03-restart-resubscribe-watchdog-plan.md @@ -452,3 +452,321 @@ T8 anytime; T9 last (after T3/T5/T6/T7 merge-ready) ``` Overall effort: **~0.75–1 dev-day code+unit, plus the rig session.** The one HIGH-RISK production slice (T3, which changes what every child receives at spawn) lands behind the full 25-suite `DriverHostActor` regression before any new test stacks on it, and behind T2's explicitly behaviour-preserving refactor. + +--- + +## Execution notes + +Branch `fix/r3-03-restart-resubscribe-watchdog`, worktree +`.claude/worktrees/agent-a5bcdd61bfba22e10`, off master `5fd65d19`. TDD throughout: every +behaviour task's test was written first, RUN, and observed red for the predicted reason before the +production edit landed. Red observations are recorded per task below. + +### T1 — 03/S17 watchdog grace (`??` → max) + +**RED observed (as predicted).** With the two new tests in place against unmodified +`DriverInstanceActor`: + +``` +DriverInstanceActorCommsLossWatchdogTests.A_reconnected_driver_gets_a_full_grace_window_before_the_watchdog_trips_again [FAIL] + Failed: Expected no messages during 00:00:00.4000000, instead we received + after 00:00:00.0496590 +Failed! - Failed: 1, Passed: 9, Total: 10 +``` + +The re-trip landed **49.7 ms** after the reconnect — one health tick (`FastPoll` = 50 ms) — against a +`Grace` of 1500 ms, i.e. exactly the "churn at health-tick cadence instead of the designed grace" +the finding describes. `A_half_broken_device_is_still_declared_lost_once_the_grace_elapses` passed +in the same run, which is what makes it the anti-vacuity control: it cannot distinguish the fix from +the bug, so the delay assertion is the one carrying the finding. + +**GREEN after the one-line max at `:574`** (plus the "grace is per connection" remarks paragraph and +the reciprocal falsification-partner xmldoc on the existing +`A_degraded_driver_that_is_still_reading_successfully_is_left_alone`): 10/10 passed, 20 s. +`A_reconnect_attempt_tears_the_dead_transport_down_before_re_initialising` stayed green — as the plan +anticipated, each of its trip cycles now costs a full `FastTimeout` (250 ms) instead of one 50 ms +tick, and its `AwaitCondition(InitializeCount >= 2, 5 s)` still has ample headroom. No budget change +was needed. + +### T2 — desired-subscription payload authority + +Behaviour-preserving refactor, so there is no red observation to record by design — the gate is that +**no test moves**. `--filter "FullyQualifiedName~DriverHostActor"` → **114/114 passed**, unchanged +from the base commit. + +⚠️ One run of that filter reported `DriverHostActorNativeAlarmAckRoutingTests.RouteNativeAlarmAck_on_non_primary_is_dropped` +failing on an `AwaitAssert`. Re-run in isolation: **3/3 passed**; re-run of the whole 114-test filter: +**114/114 passed**. Treated as the known #500 flake set for this assembly (~13 %/run), not a +regression — it touches none of the code this task edits (the alarm-ack Primary gate is a different +method entirely). + +Deviation from the plan's literal step text (cosmetic, recorded for the reviewer): the plan's Finding-1 +step 1 says `_desiredByDriver` *replaces* `_lastComposition` at `:347-349` while T3 step 2 says T3 +deletes `_lastComposition`. Keeping T2 compilable means the two fields coexist for one commit; +`_lastComposition` and its assignment are deleted in T3 as T3's own step text specifies. + +### T3 — `SpawnChild` always sends the desired set; `_lastComposition` deleted + +The falsification for this production slice is **T5**, written and observed red afterwards (below) — +T3 lands the fix first because the plan sequences it that way and the ordering guard needs the whole +25-suite regression before any new test is trusted on top of it. + +**HIGH-RISK regression gate (every child's spawn-time state changes, deploy path included):** the +FULL `Runtime.Tests` project, not just the `DriverHostActor` filter — +**639 passed, 3 skipped, 0 failed, 55 s**. The 3 skips are pre-existing (`DeploymentArtifact*` parity +tests, skipped on the base commit too). + +`_lastComposition` is gone: `grep -c _lastComposition DriverHostActor.cs` → **0**. Its xmldoc had also +drifted (it claimed the field was set at the end of `PushDesiredSubscriptions`; the assignment was in +`ApplyDesiredSubscriptions`) — deleted with the field rather than corrected. The stale +"dropped-driver fallback" comment on the old `SendAuthoredOnly` local function was rewritten as the +`SendDesiredSubscriptions` remarks naming the three real consumers. + +### T4 — test harness (seeder `enabled` flag + recording driver factory) + +Test-only, additive, source-compatible: every pre-existing `SeedV3Deployment` / `SpawnHostAndApply` +call site compiles and behaves identically (the new parameters are trailing and default to today's +values). Gate: `--filter "…~DriverHostActor|…~DriverInstanceActor"` → **179/179 passed** (the plan +named `--filter "…~V3"`, which matches nothing — the seeder's four consumers are all `DriverHostActor*` +classes; the wider filter is the same set plus more). + +**Deviation (recorded).** The plan lists one new seeder parameter (`enabled`); a second was needed and +added: `IReadOnlyDictionary? tagConfigByTag = null`. The seeder hard-codes +`TagConfig = "{}"` on every tag, and a raw tag's alarm is derived *only* from `TagConfig.alarm` via +`TagConfigIntent.Parse` (`AddressSpaceComposer.cs:769-772`) — so without it there is no way to seed the +alarm-bearing raw tag that `Operator_restart_re_subscribes_alarm_refs_too` (a test the plan itself +specifies) requires. It is trailing and optional, so it is the same source-compatible shape as +`enabled`. + +### T5 — `DriverHostActorRestartTests` (operator-restart falsification, 03/U6's first half) + +**RED observed against a T3-reverted build** (the `SendDesiredSubscriptions(child, …)` call in +`SpawnChild` commented out, everything else on the branch): **all 4 FAILED, 0 passed**, each at the +bar the plan predicts — + +``` +Operator_restart_re_subscribes_the_respawned_child [FAIL] + factory.Created[1].SubscribeCount should be greater than or equal to 1 but was 0 +Operator_restart_re_subscribes_alarm_refs_too [FAIL] + factory.Created[1].SubscribeAlarmsCount should be greater than or equal to 1 but was 0 +Restart_of_a_driver_with_an_empty_value_set_still_sends_the_set [FAIL] + respawned.SubscribeAlarmsCount should be greater than or equal to 1 but was 0 +A_child_spawned_by_a_fresh_deploy_ends_with_the_NEW_refs [FAIL] + factory.Created[2].LastSubscribedRefs should be ["Plant/Modbus/dev1/temp"] but was null +``` + +The call was then restored; **4/4 pass**, and the class was run **20× in isolation, 20/20 green** (#500 +discipline for this assembly). + +**Two deviations, both found by running the tests rather than by reading:** + +1. ⚠️ **The plan's claim that the transient spawn-time set "is never acted on" is FALSE, and the + production comment that repeated it has been corrected.** The plan reasons that the child is still + `Connecting` when the stale set arrives, so it is only stored. But the child runs **concurrently** + with the rest of `HandleArtifactForApply` — `Context.ActorOf` returns and the child starts + processing while the host is still building `refsByDriver` — so a driver whose `InitializeAsync` + completes instantly reaches Connected and subscribes the stale set *before* the bulk push arrives. + Measured directly: `SubscribeCount == 2` on the respawned child, first with revision A's ref and + then with B's. What Akka's per-pair ordering does guarantee is **convergence**, and + `HandleSubscribeAsync` drops the prior subscription before establishing the new one, so the cost is + one extra subscribe round-trip on a *respawning* deploy — a real device driver, whose initialise + opens a socket, rarely wins that race. Accepted rather than "fixed": withholding the spawn-time send + is the 03/S16 defect, and the operator-Restart and supervision paths have no bulk push coming at + all. The `SpawnChild` comment now states this instead of the plan's claim, and the test asserts + convergence (the plan's own stated acceptance — "the child's live subscription carries the NEW + refs") rather than pinning the race. +2. **`A_child_spawned_by_a_fresh_deploy_ends_with_the_NEW_refs` had to be re-shaped.** The plan + describes revision B as one "which respawns nothing but re-pushes". It respawns: the raw tags are + folded into the merged `DriverConfig`, so a tag-set edit changes the `DriverInstanceSpec` and trips + the #516 config respawn. That is the *stronger* arrangement for this test's subject (a brand-new + child spawned while the authority still holds A), so the test keeps the same name and asserts the + respawned child converges on B — plus a subsequent operator Restart, which pins that the authority + is repopulated **whole** per apply rather than accumulated. +3. **`Restart_of_a_driver_with_no_authored_tags_sends_an_empty_set_not_nothing` was renamed + `Restart_of_a_driver_with_an_empty_value_set_still_sends_the_set`, with a real positive control.** + The seeder derives `DriverInstances` from the raw-tag rows, so a driver with *zero* tags is not + expressible, and — more importantly — an all-empty set has **no driver-side observable at all** + (the child stores it and, having no handle, does nothing), which would have made the test vacuous: + "it did not subscribe" passes equally against a build that sends the respawned child nothing, i.e. + against the defect itself. The subject is therefore a driver whose only raw tag is alarm-bearing: + its value half is genuinely empty while its alarm half is not, and **both halves ride in one + message** — so `SubscribeAlarmsCount >= 1` proves the message ARRIVED and `SubscribeCount == 0` + then proves the empty value half was honoured. The falsification run above confirms the control + works: with T3 reverted this test fails on the *positive control*, not on the absence. + **Not covered, and worth a reviewer's eye:** the `_desiredByDriver` union's `_children.Keys` arm + (a running child absent from BOTH ref maps) is not reachable through this seeder and is untested. + +### T6 — supervision reprovision hook + +`DriverInstanceActor.PostRestart` (calling `base.PostRestart(reason)` first, so `PreStart` re-attaches +the rediscovery / host-status sources and restarts the health timer) → `ReprovisionRequested` → +`DriverHostActor.HandleReprovisionRequested`, registered in **all three** `Receive` blocks that already +accept `RestartDriver`. The falsification is T7, below. + +Gate: `--filter "…~DriverHostActor|…~DriverInstanceActor"` → **183/183 passed** (179 before this task +plus T5's four), no test moved. + +### T7 — first supervision-restart test + the force-reconnect verified negative (03/U6) + +**Crash seam that landed: the planned one** — `RecordingSubscribableDriver.FireMalformedDataChange` +raises `OnDataChange` with a null `Snapshot`; `DriverInstanceActor.OnDataChangeForward` dereferences +`msg.Snapshot.SourceTimestampUtc` on the actor thread and NREs; Akka's default decider Restarts. No +fallback (`OneForOneStrategy` + an internal test message) was needed, so no production surface was +added for the test. + +**RED observed against a T6-reverted build** (the `Context.Parent.Tell(new ReprovisionRequested(…))` +in `PostRestart` commented out): + +``` +Supervision_restart_reinitialises_and_re_subscribes_the_child [FAIL] + driver.InitializeCount should be 2 but was 1 + "the restarted child was never re-initialised — it is sitting in Connecting around a driver + its own PostStop already shut down (03/S6, the worse half of 03/S16)" +``` + +**Deviation, and the reason for it — the first red run was AMBIGUOUS and the test was strengthened.** +`InitializeCount == 1` is equally consistent with "the child restarted and was not re-provisioned" +and with "the child never crashed at all", so a seam that silently stopped working would read as a +passing regression test forever. A preceding `AwaitAssert(ShutdownCount >= 1)` was added — `PreRestart` +runs `PostStop`, which calls `ShutdownAsync`, so that is the evidence a restart genuinely happened. +On the re-run of the falsification that assertion **passed** and the failure landed squarely on the +re-initialise, which is what makes the red observation above meaningful. + +`Force_reconnect_keeps_the_desired_set_and_re_subscribes` passed in **both** the red and green runs — +correct, and the point: it is the verified negative (`ForceReconnect` reaches the same live actor, so +`_desiredRefs` survive and the reconnect's Connected entry re-applies them). It carries a 30 s budget +rather than `PresenceBudget`, because `SpawnChild` gives the child the production +`DefaultReconnectInterval` (10 s) and the retry is a periodic timer, so the reconnect cannot land +sooner; a budget is only spent on failure. + +After restoring the hook: **6/6 pass**, and the class was run **20× in isolation, 20/20 green**. + +### T8 — docs + +`docs/DriverLifecycle.md`: + +- New §"What a (re)spawn re-provisions" (placed before §IDriverHealthPublisher): `_desiredByDriver` as + the single authority, a table of the three incarnation routes plus the verified-negative Force + Reconnect, the supervision route's invisibility, why replacing the default decider is out of scope, + the respawning-deploy ordering **with the measured caveat that the transient is not inert**, and the + assert-a-function-not-existence rule naming the weaker existing test. +- §"The comms-loss watchdog": conjunct 2 restated as the max, plus a new paragraph on the + per-connection grace — why `??` was wrong, that it **paces** rather than ends the half-broken-device + cycle, that the watchdog is not disabled by the change, and the two tests that pin each half. + +### Final verification + +- `dotnet build ZB.MOM.WW.OtOpcUa.slnx` → **0 errors** (904 warnings, all pre-existing analyzer + `OTOPCUA0001` noise in unrelated driver test projects). +- **Full `Runtime.Tests`: 645 passed, 3 skipped, 0 failed** (1 m 9 s). Baseline on the branch point was + 637 + 3 skipped; the +8 are this plan's new tests (T1 ×2, T5 ×4, T7 ×2). The 3 skips are pre-existing + `DeploymentArtifact*` parity tests. +- The two other test projects that reference `ZB.MOM.WW.OtOpcUa.Runtime` were run as well, since T3 + changes what every driver child receives at spawn: `ControlPlane.Tests` **262/262**, `Host.Tests` + **106/106**. +- T9 (live gate) deliberately left **pending** — run by the orchestrator. + +### Summary of deviations (for the reviewer) + +1. **The plan's "the transient spawn-time set is never acted on" is false** — measured, corrected in + both the production comment and the docs, and accepted with its cost stated (T5, deviation 1). This + is the one behavioural claim in the plan that did not survive contact with the code. +2. `A_child_spawned_by_a_fresh_deploy_ends_with_the_NEW_refs` re-shaped — revision B *does* respawn the + child (raw tags fold into the merged `DriverConfig`) (T5, deviation 2). +3. `Restart_of_a_driver_with_no_authored_tags_sends_an_empty_set_not_nothing` renamed + + re-based on an alarm-only driver, because an all-empty set has no driver-side observable and the + planned test would have been vacuous; the `_children.Keys` union arm remains untested (T5, + deviation 3). +4. Seeder gained a second optional parameter (`tagConfigByTag`) — the only way to seed an alarm-bearing + raw tag, which one of the plan's own tests needs (T4). +5. The supervision test gained a preceding `ShutdownCount >= 1` assertion, because the plan's red + observation was ambiguous with "the child never crashed" (T7). +6. `Force_reconnect_keeps_the_desired_set_and_re_subscribes` uses a 30 s budget, not `PresenceBudget` — + `SpawnChild` gives the child the production 10 s `DefaultReconnectInterval` (T7). +7. `_lastComposition` and `_desiredByDriver` coexist for one commit (T2) so T2 stays compilable; the + deletion lands in T3, as T3's own step text specifies. + +--- + +## Review outcome + +Verdict **FIX-FIRST**, 3 items. All three confirmed against the code, fixed TDD (test first, seen red +for the predicted reason, then fixed), one commit each. + +### F1 [HIGH regression] — a changed alarm set latched (`edd01fcc`) + +**Confirmed.** `SubscribeDesiredAlarms` and `HandleSubscribeAlarmsAsync` both gated on +`_alarmSubscriptionHandle is not null`, which is the right guard for an idempotent re-push and the wrong +one for a deploy that changes the set. The value side converges by construction +(`StoreDesiredSubscriptions` is a full replace; `HandleSubscribeAsync` drops the prior subscription +first) — the alarm side did not. **T3 is what made it reachable**: a tag-set edit forces a respawn, and +a fast-initialising driver subscribes the transient spawn-time set before the bulk push, so the +PREVIOUS revision's alarm refs latched for the connection's life. `sourceNodeIds`-filtering drivers +(OPC UA Client, AB CIP, FOCAS) would silently never deliver the newly-added alarm. + +**RED:** `A_deploy_that_widens_the_alarm_set_re_subscribes_the_new_refs` — after deploying +alarm `{alarmX}` then `{alarmX, alarmY}`, `LastSubscribedAlarmRefs` was still `["Plant/Modbus/dev1/alarmX"]`. + +**Fix:** new `_subscribedAlarmRefs` records what the live handle was established with; +`AlarmSubscriptionIsStale()` (no handle **or** the desired refs differ) replaces the bare null check in +both places; a stale handle is **released before** re-subscribing so a filtering driver cannot +accumulate two overlapping subscriptions. The release clears the handle *first*, so a throwing +unsubscribe cannot leave the staleness check treating a dead handle as live, and it is best-effort +(logged, non-fatal — the re-subscribe is what matters). Unsubscribe-on-**empty** is still deliberately +absent, unchanged. Two notes: the release had to be wrapped in `_invoker.ExecuteAsync` or the repo's +`OTOPCUA0001` analyzer fails the build; and the Connected-handler comment that stated the +value-only convergence claim was corrected, as was the matching bullet in `docs/DriverLifecycle.md`. + +### F2 [MEDIUM] — reprovision routed by id, not incarnation (`2b7100cf`) + +**Confirmed.** Fixed with `if (!Equals(Sender, entry.Actor)) return;` (plus a Debug line naming the +superseded ref). + +⚠️ **The first version of the test PASSED against the unfixed code — the arrangement mattered.** With +the child already **Connected**, `InitializeRequested` is unhandled, so the stale reprovision was +swallowed for the wrong reason and the test was vacuous. The defect only bites a child in +`Connecting`, which is exactly where `InitializeRequested` *is* handled. `RecordingSubscribableDriver` +therefore gained an `InitializeGate` (a `TaskCompletionSource` its `InitializeAsync` awaits) so the +test can hold the child in `Connecting`. **RED with that arrangement:** `InitializeCount == 2` — the +second concurrent `InitializeAsync` racing the #527 dispose-before-overwrite. The absence assertion +carries a positive control (the child is proven alive at `InitializeCount == 1` first) and a +`GetDiagnostics` round-trip so the dropped message is provably processed before the absence is read. + +### F3 [IMPORTANT] — no backoff on the reprovision (`0793b0cb`) + +**Confirmed.** Added `DefaultReprovisionBackoff` (10 s, matched to +`DriverInstanceActor.DefaultReconnectInterval` because the reprovision drives the same connect work), +a per-driver `_lastReprovisionUtc` stamp, and a `reprovisionBackoff` Props parameter following the +existing `applyWedgeTimeout` precedent so a test need not wait the production window. + +A held request is **deferred, not dropped**: one `StartSingleTimer` per driver re-delivers a +`ReprovisionDeferred` carrying the **original incarnation ref**, so F2's identity check still applies +on re-delivery, and a burst collapses to a single pending reprovision rather than a queue. Each +deferral logs a Warning naming the crash type, so a crash loop is visible rather than quiet. + +**RED** (backoff disabled): `InitializeCount == 3` where the test requires 2-then-3 — i.e. the second +reprovision applied immediately. The test drives **two genuine supervision restarts** (not synthetic +messages), waits for the re-subscribe before arming the second crash, and asserts `ShutdownCount >= 2` +so "the second crash tripped supervision" is proven before the backoff is read. + +### Optional adds (both done) + +- `A_running_driver_with_no_authored_tags_is_handed_an_empty_set` — the `_children.Keys` union arm, + previously untested and called out as a gap in the T5 notes. It needed a driver with no tags at all, + which the seeder could not express, so `SeedV3Deployment` gained `taglessDrivers`. The tagless child + is asserted initialised-but-never-subscribed, with its tag-owning sibling's subscribe as the positive + control that the push pass ran. +- `_desiredByDriver` xmldoc now records that it is rebuilt only by a **successful** apply, so it + lingers on the last-known-good composition after a failed one — consistent with the address space, + drivers and subscriptions that also keep serving (#485/#486). + +### Verification after the three fixes + +- `dotnet build ZB.MOM.WW.OtOpcUa.slnx` → **0 errors**. +- **Full `Runtime.Tests`: 649 passed, 3 skipped, 0 failed** (up 4 from 645 — F1, F2, F3 and the + union-arm test). +- **`ControlPlane.Tests`: 262/262.** +- `DriverHostActorRestartTests` (now 10 cases) run **10× in isolation, 10/10 green**. +- T9 (live gate) still **pending** — the orchestrator runs it. Worth noting for that session: F1 is + directly observable on the rig, by deploying a second alarm-bearing raw tag to a driver that already + has one and confirming a transition on the NEW tag reaches `/alerts`. diff --git a/archreview/plans/R3-03-restart-resubscribe-watchdog-plan.md.tasks.json b/archreview/plans/R3-03-restart-resubscribe-watchdog-plan.md.tasks.json index b14841de..5e1d2dab 100644 --- a/archreview/plans/R3-03-restart-resubscribe-watchdog-plan.md.tasks.json +++ b/archreview/plans/R3-03-restart-resubscribe-watchdog-plan.md.tasks.json @@ -5,19 +5,19 @@ { "id": "T1", "subject": "Comms-loss grace is per connection: ?? -> max(LastSuccessfulRead, _connectedSinceUtc) + falsification pair (03/S17)", - "status": "pending", + "status": "done", "blockedBy": [] }, { "id": "T2", "subject": "Desired-subscription payload authority: _desiredByDriver + SendDesiredSubscriptions (behaviour-preserving refactor)", - "status": "pending", + "status": "done", "blockedBy": [] }, { "id": "T3", "subject": "SpawnChild always sends the desired set; delete the write-only _lastComposition (03/S16)", - "status": "pending", + "status": "done", "blockedBy": [ "T2" ] @@ -25,13 +25,13 @@ { "id": "T4", "subject": "Test harness: V3ArtifactSeeder enabled flag + RecordingSubscribableDriver(Factory) with a crash seam", - "status": "pending", + "status": "done", "blockedBy": [] }, { "id": "T5", "subject": "DriverHostActorRestartTests: operator restart re-subscribes the respawned child (falsifiable; 03/S16, U6)", - "status": "pending", + "status": "done", "blockedBy": [ "T3", "T4" @@ -40,7 +40,7 @@ { "id": "T6", "subject": "Supervision reprovision hook: DriverInstanceActor.PostRestart -> ReprovisionRequested -> host re-init + re-subscribe (03/S6, S16)", - "status": "pending", + "status": "done", "blockedBy": [ "T3" ] @@ -48,7 +48,7 @@ { "id": "T7", "subject": "First real supervision-restart test + force-reconnect verified-negative regression (03/U6)", - "status": "pending", + "status": "done", "blockedBy": [ "T4", "T6" @@ -57,7 +57,7 @@ { "id": "T8", "subject": "docs/DriverLifecycle.md: what a (re)spawn re-provisions + per-connection watchdog grace", - "status": "pending", + "status": "done", "blockedBy": [] }, { diff --git a/docs/DriverLifecycle.md b/docs/DriverLifecycle.md index 3e4519db..a157a5d7 100644 --- a/docs/DriverLifecycle.md +++ b/docs/DriverLifecycle.md @@ -164,6 +164,93 @@ mostly reusable — what was always missing is the trigger. --- +## What a (re)spawn re-provisions + +A driver child can come into existence by **three** routes, and until 2026-08-20 only the first of them +handed the child everything it needs to be a working driver. That is archreview **03/S16** (and its worse +half, **03/S6**): a child that exists without its desired subscription set is a *dead* driver — it +connects, reports `Connected`, publishes nothing and raises no alarms, and the #488 periodic reconcile +cannot rescue it (it early-returns on an empty desired set). The fix makes the desired set **part of what +it means to have a child** rather than a later step in the apply pipeline. + +**`DriverHostActor._desiredByDriver` is the single authority.** It is a per-driver +`(ValueRefs, AlarmRefs)` pair, cleared and repopulated **whole** on every apply in the same pass that +rebuilds the routing maps — keyed over the union of the value-ref map, the alarm-ref map and the running +`_children`, so a driver with no authored tags gets an explicit **empty** set rather than being absent. +Everything that hands a child a `SetDesiredSubscriptions` goes through the one helper +`SendDesiredSubscriptions(actor, driverId)`. (An empty set is meaningful, not a no-op to skip: the child +routes it to `Unsubscribe`, dropping a stale handle.) + +| Route | Trigger | What re-provisions it | +|---|---|---| +| **Deploy reconcile** | `ApplyDriverSpecs` → `SpawnChild` | `SpawnChild` sends `InitializeRequested` then the desired set; the apply's bulk push then sends the (freshly rebuilt) set to every child. | +| **Operator Restart** | `/hosts` → Restart → `HandleRestartDriver` | Stops the child, respawns from the cached spec — and `SpawnChild` sends the desired set, because it always does. **This is the 03/S16 fix**: there is no bulk push on this path, and before it the respawned child subscribed to nothing until the next deployment. | +| **Supervision restart** | a child throws ⇒ Akka's **default decider** ⇒ `Restart` | `DriverInstanceActor.PostRestart` Tells the host `ReprovisionRequested`; `HandleReprovisionRequested` logs a Warning naming the crash type, re-sends `InitializeRequested` (non-stub children) and the desired set. | +| **Force Reconnect** | `/hosts` → Force Reconnect → `HandleReconnectDriver` | **Nothing — verified.** It Tells `ForceReconnect` to the *same* child actor, so the instance and its `_desiredRefs`/`_desiredAlarmRefs` survive and the reconnect's Connected entry re-applies them itself. Pinned by `Force_reconnect_keeps_the_desired_set_and_re_subscribes` so a future edit to the restart paths cannot break the one that was already correct. | + +Things to know before touching any of this: + +- ⚠️ **The supervision route is the invisible one, and it was the worst.** There is **no** + `SupervisorStrategy` anywhere under `src/Server`, so a child that throws gets Akka's default `Restart`. + A restart recreates the actor object around the **same** `IDriver` — the Props closure captured it — + and its `PreRestart` runs `PostStop`, which calls `IDriver.ShutdownAsync`. It does **not** run + `SpawnChild`, and the child's `IActorRef` is unchanged, so the host had no notification a new + incarnation existed at all. Net effect before the hook: the child sat in `Connecting` forever around a + shut-down driver, with its desired set gone, and nothing logged the non-recovery. +- **Replacing the default decider is deliberately out of scope.** A `Stop` directive would silently + remove a driver; a bounded retry count would silently stop one after N crashes. Both change failure + semantics for every driver in the fleet and deserve their own plan and live gate. The reprovision hook + makes the *current* infinite-restart behaviour correct instead of dead. +- ⚠️ **…which is exactly why the reprovision is RATE-LIMITED.** The default decider restarts a faulting + child **immediately and without limit**, and the hook now makes every one of those restarts do + `ShutdownAsync` + `InitializeAsync` + `SubscribeAsync` — so a driver that faults on every value would + hammer its device as fast as it can crash. `DriverHostActor` keeps a per-driver last-reprovision stamp + and holds anything inside `DefaultReprovisionBackoff` (10 s, matched to + `DriverInstanceActor.DefaultReconnectInterval` because it drives the same connect work). The held + request is **deferred, not dropped** — one `StartSingleTimer` per driver, so a burst collapses to a + single pending reprovision — because dropping it would leave a child whose *final* crash fell inside + the window un-provisioned forever, which is the 03/S16 failure this hook exists to close. Each + deferral logs a Warning naming the crash type, so a crash loop is visible rather than merely quiet. +- **A reprovision is routed by INCARNATION, not by driver id.** `ReprovisionRequested` 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 id. 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). The handler + therefore checks `Sender` against the child it currently holds and drops anything else; the deferred + path carries the original ref so the same check still applies when the timer re-delivers it. +- **Ordering on a respawning deploy, and its one accepted cost.** `ApplyDriverSpecs` (which spawns) runs + before the pass that rebuilds `_desiredByDriver`, so a child respawned by a deploy — raw tags are folded + into the merged `DriverConfig`, so a tag-set edit changes the spec and trips the #516 config respawn — + is first handed the **previous** apply's set, then the new one by the bulk push. Akka's per-pair + ordering guarantees it **converges**; it does not guarantee the transient is inert, because the child + runs concurrently with the rest of the apply handler and a fast-initialising driver can reach + `Connected` and subscribe the stale set first. The cost is one extra subscribe round-trip + (`HandleSubscribeAsync` drops the prior subscription before establishing the new one); a real device + driver, whose initialise opens a socket, rarely wins that race. Withholding the spawn-time send to avoid + it would simply restore 03/S16, since the Restart and supervision routes have no bulk push coming. +- ⚠️ **Convergence is not free on the ALARM side, and "a subscription already exists" is the wrong guard + there.** The value path converges by construction — `StoreDesiredSubscriptions` is a full replace and + `HandleSubscribeAsync` drops the prior subscription before establishing the new one — but + `SubscribeDesiredAlarms` / `HandleSubscribeAlarmsAsync` used to gate purely on + `_alarmSubscriptionHandle is not null`, so once a handle existed the alarm set **latched for the life of + the connection**. Combined with the transient above, a deploy widening a driver's alarm refs `{X}` → + `{X, Y}` could leave a fast-initialising driver subscribed to `{X}` forever, and the drivers that filter + their feed on the subscribed `sourceNodeIds` (OPC UA Client, AB CIP, FOCAS) would simply never deliver + the new alarm — silently. The guard is now "the desired refs differ from the ones the live handle was + established with" (`_subscribedAlarmRefs`), and a changed set **releases the old subscription before + re-subscribing** so a filtering driver does not accumulate two overlapping ones. There is still + deliberately no unsubscribe-on-**empty**: a removed alarm tag's condition node is torn down on the + address-space rebuild and `ForwardNativeAlarm` drops any transition whose `ConditionId` no longer maps. +- **Restart tests must assert a FUNCTION, never existence.** `RestartDriver_respawns_the_child_without_an_actor_name_collision` + asserts only that a child exists after a restart, and it passed green against the dead driver 03/S16 + describes. It is kept — its own subject, the `InvalidActorNameException`, is real — but the load-bearing + suite is `DriverHostActorRestartTests`, which asserts that the new incarnation *subscribed* and + *re-initialised*. Its file remarks also record the supervision crash seam and why the obvious + alternatives (a throwing health publisher, a throwing `SubscribeAsync`, `Kill.Instance`) do not work. + +--- + ## IDriverHealthPublisher — health pub/sub sink [`Core.Abstractions/IDriverHealthPublisher.cs`](../src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/IDriverHealthPublisher.cs) @@ -236,7 +323,8 @@ The `Connected`-state health tick now also evaluates a watchdog, and self-sends 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. + measured from **the later of that read stamp and the current Connected entry** (see "the grace is per + connection" below). **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 @@ -250,6 +338,26 @@ still needed as the second conjunct, because a `Healthy` driver with nothing sub 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. +**The grace is per CONNECTION — the baseline is `max(LastSuccessfulRead, connectedSince)`, not +`LastSuccessfulRead ?? connectedSince`** (archreview 03/S17, fixed 2026-08-20). The two spellings differ +for exactly the case that matters, and the `??` one collapsed the grace to a single health tick. Because +`LastSuccessfulRead` is *preserved* across `Degraded`/`Faulted` writes — that preservation is the design +described above — a device that accepts a TCP connection but never serves data (a half-broken PLC, a wrong +unit id, a firewalled read path) reconnects carrying a read stamp frozen from before the outage, already +far older than the threshold. Measuring from that stamp alone re-trips on the **first** tick of every new +connection, so the reconnect cycle ran at health-tick cadence (~30 s in production) rather than at the +designed 90 s. Taking the *later* of the two gives each connection its own full window, which is what +re-stamping `connectedSince` on every Connected entry was always for; `??` only ever helped a driver that +had never read at all, and the max still covers that case. + +What this does **not** change: the conjunction, the disable-at-zero hatch, and a driver that stops reading +*within* a stable connection (its read stamp is the later term, so it still trips on schedule measured +from that read). And it **paces** the half-broken-device cycle rather than ending it — expect at most one +reconnect attempt per threshold per connection, not silence. Pinned by +`A_reconnected_driver_gets_a_full_grace_window_before_the_watchdog_trips_again` (the delay) and +`A_half_broken_device_is_still_declared_lost_once_the_grace_elapses` (the anti-vacuity half — without it, +a "fix" that simply disabled the watchdog after the first connection would look correct). + Two consequences worth knowing: - **`CalculationDriver` can never trip it, correctly.** It reports `Healthy` whenever it is running and has diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs index e95ca092..8b5f5c2c 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs @@ -82,6 +82,14 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS /// public static readonly TimeSpan DefaultApplyWedgeTimeout = TimeSpan.FromMinutes(5); + /// 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 : the reprovision drives the same + /// connect work, so it should not run more often than an ordinary reconnect attempt. + public static readonly TimeSpan DefaultReprovisionBackoff = TimeSpan.FromSeconds(10); + /// /// 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. private readonly Dictionary _pendingDelta = new(); - /// The composition from the most-recent apply (set at the END of - /// ). Null until the first apply. - private AddressSpaceComposition? _lastComposition; + /// One driver child's desired subscription payload — exactly the + /// the apply's bulk push sent it. + /// The non-alarm raw tags' RawPaths this driver should keep subscribed. + /// The alarm-bearing raw tags' RawPaths that un-gate this driver's + /// feed. + private readonly record struct DesiredSubscriptionSet( + IReadOnlyList ValueRefs, + IReadOnlyList AlarmRefs); + + /// Per-driver desired subscription sets from the most-recent apply. THE authority: the bulk + /// push, 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. + /// 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. + private readonly Dictionary _desiredByDriver = new(StringComparer.Ordinal); /// /// Cached local from the latest @@ -401,6 +425,15 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS /// ). private readonly TimeSpan _applyWedgeTimeout; + /// Per-driver rate limit for the supervision-reprovision hook (see + /// ). + private readonly TimeSpan _reprovisionBackoff; + + /// 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 _children first). + private readonly Dictionary _lastReprovisionUtc = new(StringComparer.Ordinal); + /// /// 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? 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)); /// Initializes a new DriverHostActor with the specified dependencies. /// Database context factory for configuration database access. @@ -640,9 +674,11 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS IDriverPrimaryRoleSource? primaryRoleSource = null, IHistorizationRoleView? historizationRoleView = null, Func? 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(HandleDeltaApplied); Receive(HandleApplyResult); Receive(HandleRestartDriver); + Receive(HandleReprovisionRequested); + Receive(m => ReprovisionChild(m.Request, m.Child)); Receive(HandleReconnectDriver); Receive(HandleRouteNodeWrite); Receive(HandleRouteNativeAlarmAck); @@ -1131,6 +1169,8 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS Receive(HandleDeltaApplied); Receive(HandleApplyResult); Receive(HandleRestartDriver); + Receive(HandleReprovisionRequested); + Receive(m => ReprovisionChild(m.Request, m.Child)); Receive(HandleReconnectDriver); Receive(HandleRouteNodeWrite); Receive(HandleRouteNativeAlarmAck); @@ -1713,6 +1753,8 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS Receive(HandleGetDiagnostics); Receive(_ => TryRecoverFromStale()); Receive(HandleRestartDriver); + Receive(HandleReprovisionRequested); + Receive(m => ReprovisionChild(m.Request, m.Child)); Receive(HandleReconnectDriver); Receive(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(); - var alarmRefs = alarmRefsByDriver.TryGetValue(driverId, out var ar) ? ar : Array.Empty(); - actor.Tell(new DriverInstanceActor.SetDesiredSubscriptions(refs, SubscriptionPublishingInterval, alarmRefs)); - return refs.Count; + _desiredByDriver[driverId] = new DesiredSubscriptionSet( + refsByDriver.TryGetValue(driverId, out var r) ? r : Array.Empty(), + alarmRefsByDriver.TryGetValue(driverId, out var ar) ? ar : Array.Empty()); } 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; - + /// + /// Send 's desired subscription payload to . + /// + /// + /// The ONE place a is built, so the apply's + /// bulk push, a respawn () and a supervision reprovision + /// (HandleReprovisionRequested) 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. + /// + /// An EMPTY set is valid and MEANINGFUL — the child routes it to Unsubscribe (dropping a + /// stale handle) rather than to a spurious subscribe — so a driver missing from + /// is sent an empty set, never skipped. + /// + /// + /// The driver child to hand the set to. + /// The driver instance whose set to send. + /// The value-ref count, for the bulk loop's log total. + private int SendDesiredSubscriptions(IActorRef actor, string driverId) + { + var set = _desiredByDriver.TryGetValue(driverId, out var s) + ? s + : new DesiredSubscriptionSet(Array.Empty(), Array.Empty()); + 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); } + /// + /// 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. + /// + /// + /// A restart recreates the actor object around the SAME driver instance (the Props closure captured + /// it) without going through , so neither the InitializeRequested nor + /// the desired-subscription send that a spawn performs ever happens — and the child's + /// is unchanged, so _children needs no maintenance and the host has no + /// other notification that a new incarnation exists. Its PreRestart ran PostStop, which + /// called ShutdownAsync on the driver, so re-initialising is not optional: without it the + /// child sits in Connecting forever around a shut-down driver. + /// + /// 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. + /// + /// + /// The restarted child's self-announcement. + private void HandleReprovisionRequested(DriverInstanceActor.ReprovisionRequested msg) => + ReprovisionChild(msg, Sender); + + /// A reprovision held back by , carrying the incarnation it + /// was requested by so the identity check still applies when it is re-delivered by the timer. + /// The original request. + /// The incarnation that asked; re-validated against _children on delivery, + /// so a driver restarted or removed in the meantime discards it. + 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. 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 7c8f5b3d..cdcde90c 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs @@ -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 . public sealed record ConnectivityChanged(string DriverInstanceId, bool Connected); + + /// + /// Sent to the parent from so the host re-provisions THIS incarnation + /// (archreview 03/S6, 03/S16). + /// + /// + /// A supervision restart recreates the actor object in place: the desired subscription sets are gone + /// (they are instance fields), and PreRestart has already run — which + /// called on the driver object the Props closure re-uses. Nothing + /// else re-sends : its only sender is + /// DriverHostActor.SpawnChild, and a restart does NOT respawn (the child's + /// is unchanged, so the host is not otherwise notified that a new incarnation + /// exists at all). Without this the restarted child sits in Connecting forever around a + /// shut-down driver, with nothing subscribed — silently, since Akka logs the fault but the recovery + /// is invisible. + /// + /// The driver instance that was restarted. + /// The type name of the exception that tripped supervision, for the host's log. + 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); /// @@ -252,6 +271,15 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers /// idempotent across repeated pushes. private IAlarmSubscriptionHandle? _alarmSubscriptionHandle; + /// The alarm references the live was actually established + /// with, so a CHANGED desired set can be told apart from a repeat of the same one. + /// Without this the subscription latched: both and + /// 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 sourceNodeIds (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. + private IReadOnlyList _subscribedAlarmRefs = Array.Empty(); + /// /// Gets or sets the timer scheduler for scheduling reconnection attempts. /// @@ -375,6 +403,20 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers Timers.StartPeriodicTimer("health-poll", HealthPollTick.Instance, _healthPollInterval); } + /// + /// + /// Ask the host to re-provision this incarnation — see for why a + /// supervision restart otherwise leaves a dead child. base.PostRestart runs first because it + /// runs , which re-attaches the rediscovery / host-status sources and restarts + /// the health timer; the reprovision is the part cannot do, since only the + /// host knows this driver's config and desired set. + /// + 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(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. /// + /// + /// The grace is per CONNECTION (03/S17). The silence baseline is + /// max(LastSuccessfulRead, _connectedSinceUtc), not LastSuccessfulRead ?? _connectedSinceUtc. + /// Because LastSuccessfulRead is preserved 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 ?? 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 + /// _connectedSinceUtc on every Connected entry was for. Note this paces the cycle, it + /// does not end it: a permanently half-broken device still reconnects once per threshold, and a + /// driver that stops reading within a stable connection still trips on schedule (its read + /// stamp is the later term, exactly as before). + /// /// /// 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. @@ -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(); } /// Establish the native-alarm subscription that un-gates an driver's @@ -1123,10 +1190,17 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers /// desired, and no subscription is yet established. 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()); } + /// 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 . + private bool AlarmSubscriptionIsStale() => + _alarmSubscriptionHandle is null || + !_subscribedAlarmRefs.SequenceEqual(_desiredAlarmRefs, StringComparer.Ordinal); + /// Calls the driver's (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. 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(); + 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); } diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorRestartTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorRestartTests.cs new file mode 100644 index 00000000..414932b0 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorRestartTests.cs @@ -0,0 +1,630 @@ +using Akka.Actor; +using Microsoft.EntityFrameworkCore; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Admin; +using ZB.MOM.WW.OtOpcUa.Commons.Types; +using ZB.MOM.WW.OtOpcUa.Configuration; +using ZB.MOM.WW.OtOpcUa.Runtime.Drivers; +using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers; + +/// +/// The repo's first driver-child restart suite (archreview 03/S16, 03/U6): what a child obtained +/// by each of the three incarnation routes is actually handed. +/// +/// +/// +/// The falsification bar, and why this file exists at all. An actor-restart test must assert a +/// FUNCTION the new incarnation performs — it subscribed, it re-initialised — never that the +/// incarnation EXISTS. Existence is exactly what +/// DriverHostActorReconcileTests.RestartDriver_respawns_the_child_without_an_actor_name_collision +/// asserts (no name collision, factory created a second driver, one child remains), and that test +/// passes green against the 03/S16 defect: a respawned child that connects and then reads nothing +/// forever. It is kept, because its own subject — the InvalidActorNameException — is real; it +/// is simply the weaker half of a pair, and this file is the other half. +/// +/// +/// The three incarnation routes, and what re-provisions each. (1) Deploy reconcile — +/// ApplyDriverSpecs → SpawnChild, then the apply's bulk desired-subscription push. +/// (2) Operator Restart — HandleRestartDriver stops the child and calls +/// SpawnChild with the cached spec; before 03/S16 that was the whole story, and the new child +/// started with an EMPTY desired set that nothing re-sent until the next deployment (the #488 +/// periodic reconcile early-returns on an empty set, so it is structurally inert here). +/// (3) Supervision restart — Akka's default decider recreates the actor object around the +/// SAME driver instance without going through SpawnChild at all, so nothing re-sent +/// InitializeRequested either and the child sat in Connecting forever around a driver +/// its own PostStop had already shut down. A fourth route, Force Reconnect, re-uses the +/// live actor instance and needs nothing — a verified negative, pinned here so a future edit to the +/// restart path cannot break the path that was already correct. +/// +/// +/// The supervision crash seam, and why the obvious ones do not work. The seam used is the +/// driver's own OnDataChange carrying a NULL Snapshot +/// (): DataChangeEventArgs is +/// a positional record with no null validation, and DriverInstanceActor.OnDataChangeForward +/// dereferences msg.Snapshot unguarded on the ACTOR thread, so it NREs inside the child and +/// the default decider Restarts it. Rejected alternatives, each for a specific reason: a throwing +/// IDriverHealthPublisher (PublishHealthSnapshot wraps the publish in a +/// catch (Exception), so it never escapes); a throwing SubscribeAsync (also swallowed — +/// the stub already models it via SubscribeFailuresBeforeSuccess); Kill.Instance (the +/// default decider Stops on ActorKilledException, a different directive and therefore a +/// different code path); and an internal "throw this" test message (a production surface added +/// purely for a test). The fault is arbitrary by design — what is under test is the RECOVERY. If +/// DataChangeEventArgs ever gains null validation, the fallback is an explicit +/// OneForOneStrategy(Directive.Restart) on the host plus an internal test-only throw message. +/// +/// +public sealed class DriverHostActorRestartTests : RuntimeActorTestBase +{ + private static readonly RevisionHash RevA = RevisionHash.Parse(new string('a', 64)); + private static readonly RevisionHash RevB = RevisionHash.Parse(new string('b', 64)); + private static readonly NodeId TestNode = NodeId.Parse("node-1:4053"); + + private const string DriverId = "DI-1"; + private const string SpeedPath = "Plant/Modbus/dev1/speed"; + private const string TempPath = "Plant/Modbus/dev1/temp"; + private const string AlarmXPath = "Plant/Modbus/dev1/alarmX"; + private const string AlarmYPath = "Plant/Modbus/dev1/alarmY"; + + /// TagConfig marking a raw tag as alarm-bearing (parsed by TagConfigIntent). + private const string AlarmConfig = """{"alarm":{"alarmType":"AlarmCondition","severity":500}}"""; + + /// Seeds + applies a v3 deployment whose single Modbus driver is ENABLED, so a real child is + /// spawned around a , and waits until that child has connected + /// and established its first subscription. + private (IActorRef Host, RecordingSubscribableDriverFactory Factory) DeployOneDriver( + IDbContextFactory db, + (string Driver, string DriverName, string Device, string Tag, string DataType, bool Writable)[] rawTags, + IReadOnlyDictionary? tagConfigByTag = null) + { + var factory = new RecordingSubscribableDriverFactory("Modbus"); + var deploymentId = V3ArtifactSeeder.SeedV3Deployment( + db, RevA, rawTags, Array.Empty<(string, string, string, string, string?)>(), + enabled: true, tagConfigByTag: tagConfigByTag); + var (host, _, _) = V3ArtifactSeeder.SpawnHostAndApply(this, db, TestNode, deploymentId, RevA, factory); + return (host, factory); + } + + /// + /// The 03/S16 falsification. An operator's Restart is supposed to recover a sick driver; before + /// the fix it silently killed it — the respawned child came up with _desiredRefs empty, and the + /// only SetDesiredSubscriptions sender in the repo was the apply's bulk push, so the driver + /// subscribed to nothing until the next deployment dispatch. + /// + /// Deliberately NOT the pass criterion: that a second driver was created, or that one child + /// remains. Both are true against the defect. + /// + /// + [Fact] + public void Operator_restart_re_subscribes_the_respawned_child() + { + var db = NewInMemoryDbFactory(); + var (host, factory) = DeployOneDriver(db, new[] { (DriverId, "Modbus", "dev1", "speed", "Float", false) }); + + AwaitAssert( + () => + { + factory.Created.Count.ShouldBe(1); + factory.Created[0].SubscribeCount.ShouldBe(1); + factory.Created[0].LastSubscribedRefs.ShouldBe(new[] { SpeedPath }); + }, + duration: PresenceBudget); + + host.Tell(new RestartDriver("c1", DriverId, "tester", Guid.NewGuid())); + + AwaitAssert( + () => + { + factory.Created.Count.ShouldBe(2, "the restart did not respawn the child around a new driver"); + factory.Created[1].SubscribeCount.ShouldBeGreaterThanOrEqualTo( + 1, + "the RESPAWNED child never subscribed — it is Connected and reading nothing, which is the " + + "03/S16 defect (an operator pressed Restart to recover a driver and silently killed it)"); + factory.Created[1].LastSubscribedRefs.ShouldBe(new[] { SpeedPath }); + }, + duration: PresenceBudget); + } + + /// + /// The value refs and the alarm refs ride in ONE SetDesiredSubscriptions but are applied by two + /// different child code paths (the value set self-Tells Subscribe; the alarm set runs + /// SubscribeDesiredAlarms against the IAlarmSource feed, which stays gated shut until it + /// is called). Sending the message is therefore not evidence that both halves landed, so the alarm + /// half is pinned separately. + /// + [Fact] + public void Operator_restart_re_subscribes_alarm_refs_too() + { + var db = NewInMemoryDbFactory(); + var (host, factory) = DeployOneDriver( + db, + new[] + { + (DriverId, "Modbus", "dev1", "speed", "Float", false), + (DriverId, "Modbus", "dev1", "temp", "Float", false), + }, + // An alarm-bearing raw tag is EXCLUDED from the value set and carried in the alarm set instead. + new Dictionary + { + ["temp"] = """{"alarm":{"alarmType":"AlarmCondition","severity":500}}""", + }); + + AwaitAssert( + () => + { + factory.Created.Count.ShouldBe(1); + factory.Created[0].SubscribeAlarmsCount.ShouldBeGreaterThanOrEqualTo(1); + factory.Created[0].LastSubscribedAlarmRefs.ShouldBe(new[] { TempPath }); + }, + duration: PresenceBudget); + + host.Tell(new RestartDriver("c1", DriverId, "tester", Guid.NewGuid())); + + AwaitAssert( + () => + { + factory.Created.Count.ShouldBe(2); + factory.Created[1].SubscribeAlarmsCount.ShouldBeGreaterThanOrEqualTo( + 1, + "the respawned child never re-subscribed its alarm refs — the IAlarmSource feed stays " + + "gated shut, so no native alarm transition can ever reach the host again"); + factory.Created[1].LastSubscribedAlarmRefs.ShouldBe(new[] { TempPath }); + // The value half must survive the alarm half being present. + factory.Created[1].LastSubscribedRefs.ShouldBe(new[] { SpeedPath }); + }, + duration: PresenceBudget); + } + + /// + /// The ordering guard for the spawn-time send. ApplyDriverSpecs (which spawns) runs BEFORE + /// ApplyDesiredSubscriptions (which rebuilds the authority), so a child spawned by a fresh + /// deploy is momentarily handed the PREVIOUS apply's set. That transient must never be acted on: it + /// is superseded by the bulk push on the same sender→receiver pair (Akka guarantees per-pair + /// ordering) while the child is still Connecting, where SetDesiredSubscriptions only + /// STORES; the subscribe happens on the Connected entry, from whatever was stored last. + /// + /// Revision B swaps the driver's tag, so A's and B's sets are distinguishable. Note the raw tags + /// are folded into the merged DriverConfig, so a tag-set change DOES change the driver + /// spec and therefore respawns the child (#516) — which is precisely the arrangement under test: + /// a brand-new child, spawned while the authority still holds A, must end up subscribed to B. + /// + /// + /// What is asserted is CONVERGENCE, not inertness, and the difference was measured rather + /// than assumed. The child runs concurrently with the rest of the apply handler, so a driver + /// whose InitializeAsync completes instantly (this stub does) can reach Connected and + /// subscribe the transient stale set BEFORE the bulk push arrives — observed here as + /// SubscribeCount == 2. Akka's per-pair ordering still guarantees the final state is B's, + /// and HandleSubscribeAsync drops the prior subscription before establishing the new one, + /// so the cost is one extra round-trip on a respawning deploy. Pinning the count would pin a + /// race; the final refs are the invariant. + /// + /// + [Fact] + public void A_child_spawned_by_a_fresh_deploy_ends_with_the_NEW_refs() + { + var db = NewInMemoryDbFactory(); + var (host, factory) = DeployOneDriver(db, new[] { (DriverId, "Modbus", "dev1", "speed", "Float", false) }); + + AwaitAssert( + () => factory.Created[0].LastSubscribedRefs.ShouldBe(new[] { SpeedPath }), + duration: PresenceBudget); + + var revB = V3ArtifactSeeder.SeedV3Deployment( + db, RevB, + new[] { (DriverId, "Modbus", "dev1", "temp", "Float", false) }, + Array.Empty<(string, string, string, string, string?)>(), + enabled: true); + host.Tell(new Commons.Messages.Deploy.DispatchDeployment(revB, RevB, CorrelationId.NewId())); + + AwaitAssert( + () => + { + factory.Created.Count.ShouldBe(2, "the tag-set change re-spawns the child (#516 config respawn)"); + factory.Created[1].LastSubscribedRefs.ShouldBe( + new[] { TempPath }, + "the newly-spawned child is on the PREVIOUS revision's refs — the spawn-time set was not superseded"); + }, + duration: PresenceBudget); + + // A Restart on top of that must hand the new child B's set, not A's — i.e. the authority really is + // repopulated WHOLE on every apply rather than accumulated. + host.Tell(new RestartDriver("c1", DriverId, "tester", Guid.NewGuid())); + AwaitAssert( + () => + { + factory.Created.Count.ShouldBe(3); + factory.Created[2].LastSubscribedRefs.ShouldBe(new[] { TempPath }); + }, + duration: PresenceBudget); + } + + /// + /// The empty-set semantics are load-bearing and must not degrade into "skip the send": an EMPTY value + /// set is what tells the child to Unsubscribe and drop a stale handle, so it must be SENT. + /// + /// The subject is a driver whose only raw tag is alarm-bearing: alarm tags are excluded from the + /// value set and carried in the alarm one, so its value half is genuinely empty while its alarm + /// half is not — and both halves ride in ONE SetDesiredSubscriptions. That is what makes + /// the absence assertion non-vacuous: SubscribeAlarmsCount >= 1 proves the message + /// ARRIVED, and SubscribeCount == 0 then proves the empty value half was honoured rather + /// than the whole send being skipped. Asserting only "it did not subscribe" would pass equally + /// against a build that sent the respawned child nothing at all — the 03/S16 defect itself. + /// + /// + [Fact] + public void Restart_of_a_driver_with_an_empty_value_set_still_sends_the_set() + { + var db = NewInMemoryDbFactory(); + var (host, factory) = DeployOneDriver( + db, + new[] { (DriverId, "Modbus", "dev1", "temp", "Float", false) }, + new Dictionary + { + ["temp"] = """{"alarm":{"alarmType":"AlarmCondition","severity":500}}""", + }); + + AwaitAssert( + () => + { + factory.Created.Count.ShouldBe(1); + factory.Created[0].SubscribeAlarmsCount.ShouldBeGreaterThanOrEqualTo(1); + }, + duration: PresenceBudget); + + host.Tell(new RestartDriver("c1", DriverId, "tester", Guid.NewGuid())); + + AwaitAssert( + () => + { + factory.Created.Count.ShouldBe(2); + var respawned = factory.Created[1]; + respawned.InitializeCount.ShouldBe(1, "the respawned child was never initialised"); + // Positive control: the SetDesiredSubscriptions genuinely reached the respawned child. + respawned.SubscribeAlarmsCount.ShouldBeGreaterThanOrEqualTo( + 1, "the respawned child received no desired set at all — the absence below would be vacuous"); + respawned.LastSubscribedAlarmRefs.ShouldBe(new[] { TempPath }); + respawned.SubscribeCount.ShouldBe( + 0, "an EMPTY value set must route to Unsubscribe / no-op, never to a spurious subscribe"); + }, + duration: PresenceBudget); + } + + /// + /// Review finding F1 — the alarm subscription must FOLLOW a changed desired set, not latch. + /// The value side converges by construction (HandleSubscribeAsync drops the prior subscription + /// before establishing the new one, and StoreDesiredSubscriptions is a full replace), but the + /// alarm side is guarded by _alarmSubscriptionHandle is not null in BOTH + /// SubscribeDesiredAlarms and HandleSubscribeAlarmsAsync — so once a handle exists it + /// was never re-established for the life of the connection. + /// + /// That became reachable when every spawn started carrying a desired set: a tag-set edit forces a + /// respawn (raw tags fold into the merged DriverConfig, #516), and a fast-initialising + /// driver connects and subscribes the transient spawn-time set — the PREVIOUS revision's alarm + /// refs — before the bulk push lands. The bulk push then early-returns, latching the old set for + /// the connection's life. Drivers that FILTER on sourceNodeIds (OPC UA Client, AB CIP, + /// FOCAS) would silently never deliver the newly-added alarm. + /// + /// + [Fact] + public void A_deploy_that_widens_the_alarm_set_re_subscribes_the_new_refs() + { + var db = NewInMemoryDbFactory(); + var (host, factory) = DeployOneDriver( + db, + new[] { (DriverId, "Modbus", "dev1", "alarmX", "Float", false) }, + new Dictionary { ["alarmX"] = AlarmConfig }); + + AwaitAssert( + () => + { + factory.Created.Count.ShouldBe(1); + factory.Created[0].LastSubscribedAlarmRefs.ShouldBe(new[] { AlarmXPath }, ignoreOrder: true); + }, + duration: PresenceBudget); + + // Revision B WIDENS the alarm set: X stays, Y is added. + var revB = V3ArtifactSeeder.SeedV3Deployment( + db, RevB, + new[] + { + (DriverId, "Modbus", "dev1", "alarmX", "Float", false), + (DriverId, "Modbus", "dev1", "alarmY", "Float", false), + }, + Array.Empty<(string, string, string, string, string?)>(), + enabled: true, + tagConfigByTag: new Dictionary + { + ["alarmX"] = AlarmConfig, + ["alarmY"] = AlarmConfig, + }); + host.Tell(new Commons.Messages.Deploy.DispatchDeployment(revB, RevB, CorrelationId.NewId())); + + AwaitAssert( + () => + { + var current = factory.Created[^1]; + current.LastSubscribedAlarmRefs.ShouldBe( + new[] { AlarmXPath, AlarmYPath }, + ignoreOrder: true, + "the driver is still subscribed to the PREVIOUS revision's alarm refs — the alarm " + + "subscription latched on the transient spawn-time set and never followed the new one"); + }, + duration: PresenceBudget); + } + + /// + /// The repo's first genuine Akka SUPERVISION-restart test (03/U6). Not a stop, not a stub: the + /// driver raises a malformed OnDataChange, the actor NREs on its own thread, and Akka's default + /// decider Restarts the child in place. + /// + /// A restart is strictly worse than the operator-Restart case 03/S16 names. It recreates the actor + /// object around the SAME driver instance without going through SpawnChild, so nothing + /// re-sends InitializeRequested — while PreRestart has already run PostStop, + /// which called ShutdownAsync on that very driver. The child therefore sat in + /// Connecting forever around a shut-down driver, with its desired set gone too, and + /// silently: Akka logs the fault, nothing logs the non-recovery. + /// + /// + /// The assertions are FUNCTIONS of the new incarnation — it re-initialised, it re-subscribed the + /// deployed refs — never that it exists. Created.Count == 1 is what distinguishes this from + /// the operator path: a restart must NOT produce a second driver object. + /// + /// + [Fact] + public void Supervision_restart_reinitialises_and_re_subscribes_the_child() + { + var db = NewInMemoryDbFactory(); + var (_, factory) = DeployOneDriver(db, new[] { (DriverId, "Modbus", "dev1", "speed", "Float", false) }); + + AwaitAssert( + () => + { + factory.Created.Count.ShouldBe(1); + factory.Created[0].SubscribeCount.ShouldBe(1); + }, + duration: PresenceBudget); + + var driver = factory.Created[0]; + driver.InitializeCount.ShouldBe(1); + + // The crash. Subscribing is what attaches the actor's OnDataChange handler, so this only reaches the + // actor thread because the arrangement above got as far as a live subscription. + driver.FireMalformedDataChange(SpeedPath); + + // FIRST: prove the crash seam actually tripped supervision, before asserting anything about the + // recovery. PreRestart runs PostStop, which calls ShutdownAsync on the driver — so a raised + // ShutdownCount is the evidence that a restart happened at all. Without this the assertions below + // would be ambiguous: "the child was not re-provisioned" and "the child never crashed" look + // identical from the outside, and a seam that silently stopped working would read as a passing + // regression test forever. + AwaitAssert( + () => driver.ShutdownCount.ShouldBeGreaterThanOrEqualTo( + 1, "the crash seam did not trip supervision — nothing below is testing a restart"), + duration: PresenceBudget); + + AwaitAssert( + () => + { + factory.Created.Count.ShouldBe( + 1, "a supervision restart must re-use the same driver object, not respawn the child"); + driver.InitializeCount.ShouldBe( + 2, + "the restarted child was never re-initialised — it is sitting in Connecting around a " + + "driver its own PostStop already shut down (03/S6, the worse half of 03/S16)"); + driver.SubscribeCount.ShouldBeGreaterThanOrEqualTo( + 2, "the restarted child never re-subscribed — its desired set died with the old instance"); + driver.LastSubscribedRefs.ShouldBe(new[] { SpeedPath }); + }, + duration: PresenceBudget); + } + + /// + /// Review finding F2 — a reprovision must be routed by INCARNATION, not just by driver id. + /// ReprovisionRequested is delivered asynchronously, so one can still be in flight from an old + /// incarnation when HandleRestartDriver has already stopped that child and swapped in a new + /// one under the same DriverInstanceId. Routing on the id alone delivers it to the NEW child, + /// which is in Connecting around a healthy driver: it triggers a second, concurrent + /// InitializeAsync that races the very dispose-before-overwrite the #527 teardown exists to + /// serialise (ModbusDriver._transport, MqttDriver._connection). + /// + /// A stale sender is simulated directly — a probe standing in for the dead incarnation — because + /// provoking the real race would be timing-dependent. The assertion is an ABSENCE, so it carries + /// a positive control: the current child is proven alive and provisioned first, and its + /// InitializeCount must then not move. + /// + /// + [Fact] + public void A_reprovision_from_a_stale_incarnation_is_ignored() + { + var db = NewInMemoryDbFactory(); + // The child must be in CONNECTING for this defect to bite: InitializeRequested is unhandled in the + // Connected behaviour, so an already-connected child would swallow the stale reprovision for the + // wrong reason and the test would pass vacuously. The gated init holds it in Connecting. + var factory = new RecordingSubscribableDriverFactory("Modbus", gateInitialize: true); + var deploymentId = V3ArtifactSeeder.SeedV3Deployment( + db, RevA, + new[] { (DriverId, "Modbus", "dev1", "speed", "Float", false) }, + Array.Empty<(string, string, string, string, string?)>(), + enabled: true); + var (host, _, _) = V3ArtifactSeeder.SpawnHostAndApply(this, db, TestNode, deploymentId, RevA, factory); + + // Positive control: the current child is alive and has begun its one and only initialise. + AwaitAssert( + () => + { + factory.Created.Count.ShouldBe(1); + factory.Created[0].InitializeCount.ShouldBe(1); + }, + duration: PresenceBudget); + var driver = factory.Created[0]; + + // A ReprovisionRequested that did NOT come from the child the host currently holds. + var staleIncarnation = CreateTestProbe(); + host.Tell( + new DriverInstanceActor.ReprovisionRequested(DriverId, "NullReferenceException"), + staleIncarnation.Ref); + + // Round-trip a message the host must answer, so the reprovision above is provably processed before + // the absence is asserted — otherwise "nothing happened yet" would pass for "it was ignored". + var coordinator = CreateTestProbe(); + host.Tell(new Commons.Messages.Fleet.GetDiagnostics(CorrelationId.NewId()), coordinator.Ref); + coordinator.ExpectMsg(PresenceBudget); + + driver.InitializeCount.ShouldBe( + 1, + "a reprovision from a dead incarnation reached the CURRENT child and started a second, " + + "concurrent InitializeAsync on a driver that was already connected"); + } + + /// + /// Review finding F3 - a crash-looping driver must not become a reconnect storm. There is no + /// SupervisorStrategy anywhere under src/Server, so the default decider restarts a + /// faulting child immediately and without limit. That was survivable while a restart did + /// nothing; the reprovision hook now makes every one of them do + /// ShutdownAsync + InitializeAsync + SubscribeAsync, so a driver that faults on + /// every value would hammer its device as fast as it can crash. + /// + /// The reprovision is therefore rate-limited per driver. Crucially it is deferred, not + /// dropped - the last request must still be honoured, or a child whose final crash fell + /// inside the window would be left un-provisioned forever, which is the very failure this whole + /// plan exists to close. Two genuine back-to-back supervision restarts: the first reprovision + /// applies immediately, the second is held and then lands. + /// + /// + [Fact] + public void A_second_reprovision_inside_the_backoff_window_is_deferred_not_dropped() + { + var db = NewInMemoryDbFactory(); + // Generous relative to the in-memory step time below, so "inside the window" is a decision rather + // than a race, while still far under the presence budget the deferred landing waits on. + var backoff = TimeSpan.FromSeconds(3); + var factory = new RecordingSubscribableDriverFactory("Modbus"); + var deploymentId = V3ArtifactSeeder.SeedV3Deployment( + db, RevA, + new[] { (DriverId, "Modbus", "dev1", "speed", "Float", false) }, + Array.Empty<(string, string, string, string, string?)>(), + enabled: true); + V3ArtifactSeeder.SpawnHostAndApply( + this, db, TestNode, deploymentId, RevA, factory, reprovisionBackoff: backoff); + + AwaitAssert( + () => + { + factory.Created.Count.ShouldBe(1); + factory.Created[0].SubscribeCount.ShouldBe(1); + }, + duration: PresenceBudget); + var driver = factory.Created[0]; + + // Crash 1: the first reprovision has no prior stamp, so it applies immediately. + driver.FireMalformedDataChange(SpeedPath); + AwaitAssert( + () => + { + driver.InitializeCount.ShouldBe(2); + driver.SubscribeCount.ShouldBe(2, "the re-subscribe must land before the second crash is armed"); + }, + duration: PresenceBudget); + + // Crash 2, immediately after - well inside the window. + driver.FireMalformedDataChange(SpeedPath); + AwaitAssert( + () => driver.ShutdownCount.ShouldBeGreaterThanOrEqualTo( + 2, "the second crash seam did not trip supervision - nothing below is testing a backoff"), + duration: PresenceBudget); + + driver.InitializeCount.ShouldBe( + 2, "the second reprovision was applied immediately - the backoff is not rate-limiting anything"); + + // ...but it must still land once the window elapses. + AwaitAssert( + () => driver.InitializeCount.ShouldBe( + 3, "the deferred reprovision was DROPPED, not deferred"), + duration: PresenceBudget); + } + + /// + /// The _children.Keys arm of the desired-set authority's union. A driver with NO authored tags + /// is absent from both the value- and alarm-ref maps, so keying the authority off those maps alone + /// would leave it with no entry at all — and "absent" must not be allowed to differ from "empty", + /// because an empty set is what tells a child to Unsubscribe and drop a stale handle. + /// + /// Asserted the only way it is observable: the tagless child is initialised (so it is provably + /// alive and was provisioned) and never subscribes, while its tag-owning sibling in the same + /// deployment does — the sibling is the positive control proving the apply's push pass actually + /// ran, so "it did not subscribe" cannot pass for "nothing happened". + /// + /// + [Fact] + public void A_running_driver_with_no_authored_tags_is_handed_an_empty_set() + { + var db = NewInMemoryDbFactory(); + var factory = new RecordingSubscribableDriverFactory("Modbus"); + var deploymentId = V3ArtifactSeeder.SeedV3Deployment( + db, RevA, + new[] { (DriverId, "Modbus", "dev1", "speed", "Float", false) }, + Array.Empty<(string, string, string, string, string?)>(), + enabled: true, + taglessDrivers: new[] { "DI-TAGLESS" }); + V3ArtifactSeeder.SpawnHostAndApply(this, db, TestNode, deploymentId, RevA, factory); + + AwaitAssert( + () => + { + factory.Created.Count.ShouldBe(2); + // Positive control: the sibling that DOES have a tag subscribed, so the push pass ran. + factory.Created.Single(d => d.DriverInstanceId == DriverId) + .SubscribeCount.ShouldBe(1); + factory.Created.Single(d => d.DriverInstanceId == "DI-TAGLESS") + .InitializeCount.ShouldBe(1); + }, + duration: PresenceBudget); + + factory.Created.Single(d => d.DriverInstanceId == "DI-TAGLESS") + .SubscribeCount.ShouldBe( + 0, "a driver with no authored tags must be handed an EMPTY set, never a spurious subscribe"); + } + + /// + /// The verified negative, pinned so this plan's edits to the restart paths cannot break the one that + /// was already correct. Force Reconnect (ReconnectDriver) Tells ForceReconnect to the + /// SAME child actor: the instance survives, so _desiredRefs/_desiredAlarmRefs are intact + /// and the reconnect's Connected entry re-applies them by itself. Nothing needs re-provisioning, and + /// no new driver object is created. + /// + /// + /// The budget here is deliberately wider than : + /// SpawnChild gives the child the production DefaultReconnectInterval (10 s) and the + /// retry is a periodic timer, so the reconnect cannot land sooner than that. A budget is spent only + /// when the assertion fails, so widening it costs nothing on the happy path. + /// + [Fact] + public void Force_reconnect_keeps_the_desired_set_and_re_subscribes() + { + var db = NewInMemoryDbFactory(); + var (host, factory) = DeployOneDriver(db, new[] { (DriverId, "Modbus", "dev1", "speed", "Float", false) }); + + AwaitAssert( + () => + { + factory.Created.Count.ShouldBe(1); + factory.Created[0].SubscribeCount.ShouldBe(1); + }, + duration: PresenceBudget); + + var driver = factory.Created[0]; + host.Tell(new ReconnectDriver("c1", DriverId, "tester", Guid.NewGuid())); + + AwaitAssert( + () => + { + factory.Created.Count.ShouldBe(1, "Force Reconnect must NOT respawn the child"); + driver.SubscribeCount.ShouldBeGreaterThanOrEqualTo( + 2, "the reconnected driver never re-subscribed its (surviving) desired set"); + driver.LastSubscribedRefs.ShouldBe(new[] { SpeedPath }); + }, + duration: TimeSpan.FromSeconds(30)); + } +} 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 index f12f3b54..8010dd02 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorCommsLossWatchdogTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorCommsLossWatchdogTests.cs @@ -40,6 +40,10 @@ public sealed class DriverInstanceActorCommsLossWatchdogTests : RuntimeActorTest private static readonly TimeSpan FastPoll = TimeSpan.FromMilliseconds(50); private static readonly TimeSpan FastTimeout = TimeSpan.FromMilliseconds(250); + /// The threshold used by the per-connection grace pair (03/S17). Deliberately an order of + /// magnitude above so the absence window below is a decision, not a race. + private static readonly TimeSpan Grace = TimeSpan.FromMilliseconds(1500); + /// Drives the actor to Connected and returns once the connect has been announced. private (IActorRef Actor, Akka.TestKit.TestProbe Parent) ConnectDriver( HealthControlledStubDriver driver, @@ -87,6 +91,14 @@ public sealed class DriverInstanceActorCommsLossWatchdogTests : RuntimeActorTest /// 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. + /// + /// Falsification partner (03/S17): + /// . + /// The silence baseline is max(LastSuccessfulRead, _connectedSinceUtc), and each test kills + /// one term of that max — this one goes red if the LastSuccessfulRead term is dropped (its + /// connect ages far past the threshold while its reads keep succeeding), that one goes red if the + /// _connectedSinceUtc term is dropped (i.e. if the original ?? is restored). + /// /// [Fact] public void A_degraded_driver_that_is_still_reading_successfully_is_left_alone() @@ -258,6 +270,77 @@ public sealed class DriverInstanceActorCommsLossWatchdogTests : RuntimeActorTest parent.ExpectNoMsg(TimeSpan.FromMilliseconds(200)); } + + /// + /// 03/S17 — the grace is per connection. Drivers deliberately preserve + /// across Degraded/Faulted health writes (that + /// preservation is the #527 design), so for a device that accepts a TCP connection but never serves + /// data the stamp stays frozen from before the outage — already older than the threshold. Measuring + /// silence from that stamp alone (??) makes the first health tick of every new + /// connection re-trip, so the reconnect cycle runs at health-tick cadence (~30 s in production) + /// instead of the designed grace. Measuring from max(LastSuccessfulRead, _connectedSinceUtc) + /// gives each connection its own full window, which is exactly what the _connectedSinceUtc + /// re-stamp on every Connected entry already promises in its comment. + /// + /// Falsification partner: + /// — see its + /// remarks for how the pair covers both terms of the max. + /// + /// + /// The absence window carries the house positive control: a proven number of health-poll ticks + /// processed after the reconnect, so "the watchdog did not re-trip" cannot pass because + /// the actor stopped ticking. It is also deliberately far shorter than , so + /// it asserts a delay, not a disablement — + /// is the anti-vacuity half. + /// + /// + [Fact] + public void A_reconnected_driver_gets_a_full_grace_window_before_the_watchdog_trips_again() + { + var driver = new HealthControlledStubDriver + { + // The half-broken device: the socket connects, every read fails, and the last successful read is + // frozen well before the outage — preserved across the Degraded writes, as every driver does. + Health = new DriverHealth(DriverState.Degraded, DateTime.UtcNow.AddMinutes(-5), "reads time out"), + }; + var (_, parent) = ConnectDriver(driver, commsLossTimeout: Grace, reconnectInterval: TimeSpan.FromMilliseconds(100)); + + // The first trip is correct and expected — the device really has been silent for five minutes. + parent.ExpectMsg(PresenceBudget) + .Connected.ShouldBeFalse(); + // The retry re-initialises successfully (the socket still accepts), so a NEW connection starts here. + parent.ExpectMsg(PresenceBudget) + .Connected.ShouldBeTrue(); + + var ticksAtReconnect = driver.HealthReadCount; + parent.ExpectNoMsg(TimeSpan.FromMilliseconds(400)); + (driver.HealthReadCount - ticksAtReconnect).ShouldBeGreaterThanOrEqualTo( + 4, + "health-poll ticks did not run after the reconnect — the absence assertion above would be vacuous"); + } + + /// + /// The anti-vacuity half of the 03/S17 pair: the per-connection grace delays the watchdog, it + /// does not disable it. The same half-broken device is still declared lost once the grace elapses — + /// without this, a "fix" that simply stopped evaluating the watchdog after the first connection would + /// pass + /// and look correct. + /// + [Fact] + public void A_half_broken_device_is_still_declared_lost_once_the_grace_elapses() + { + var driver = new HealthControlledStubDriver + { + Health = new DriverHealth(DriverState.Degraded, DateTime.UtcNow.AddMinutes(-5), "reads time out"), + }; + var (_, parent) = ConnectDriver(driver, commsLossTimeout: Grace, reconnectInterval: TimeSpan.FromMilliseconds(100)); + + parent.ExpectMsg(PresenceBudget).Connected.ShouldBeFalse(); + parent.ExpectMsg(PresenceBudget).Connected.ShouldBeTrue(); + + // The second connection serves no data either, so once ITS grace elapses the watchdog trips again. + parent.ExpectMsg(PresenceBudget).Connected.ShouldBeFalse(); + } } /// 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 25d459f4..3c4783d0 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 @@ -188,3 +188,215 @@ internal sealed class StubHandle : ISubscriptionHandle /// Gets the diagnostic ID of the subscription. public string DiagnosticId => "stub-sub"; } + +/// +/// A per-instance recording driver for the restart / re-subscribe suite (03/S16, U6): unlike +/// it carries the spec's +/// ( hard-codes a constant on a non-virtual member), and it counts and records the +/// alarm subscription alongside the value one — the two travel in a single +/// SetDesiredSubscriptions but through two different child code paths, so both need observing. +/// +/// +/// +/// The falsification bar these counters exist for: a restart test must assert a FUNCTION the new +/// incarnation performs — it subscribed, it re-initialised — never that the incarnation exists. +/// Existence is what +/// DriverHostActorReconcileTests.RestartDriver_respawns_the_child_without_an_actor_name_collision +/// asserts, and that test passes against the very defect 03/S16 names. +/// +/// +/// is the supervision crash seam — see +/// DriverHostActorRestartTests's file remarks for why the obvious alternatives do not work. +/// +/// +internal sealed class RecordingSubscribableDriver : IDriver, ISubscribable, IAlarmSource +{ + private readonly Lock _gate = new(); + private readonly StubHandle _handle = new(); + private readonly RecordingAlarmHandle _alarmHandle = new(); + private int _initializeCount; + private int _shutdownCount; + private int _subscribeCount; + private int _subscribeAlarmsCount; + private IReadOnlyList? _lastSubscribedRefs; + private IReadOnlyList? _lastSubscribedAlarmRefs; + + /// Initializes a new recording driver stamped with the spec's identity. + /// The driver instance id from the spec that created it. + /// The driver type from the spec that created it. + public RecordingSubscribableDriver(string driverInstanceId, string driverType) + { + DriverInstanceId = driverInstanceId; + DriverType = driverType; + } + + /// + public string DriverInstanceId { get; } + + /// + public string DriverType { get; } + + /// Number of calls. A supervision restart re-uses the SAME driver + /// object (the Props closure captured it), so this going 1 to 2 is how the reprovision is observed. + public int InitializeCount => Volatile.Read(ref _initializeCount); + + /// Number of calls — a restart's PostStop raises it. + public int ShutdownCount => Volatile.Read(ref _shutdownCount); + + /// Number of calls. + public int SubscribeCount => Volatile.Read(ref _subscribeCount); + + /// Number of calls. + public int SubscribeAlarmsCount => Volatile.Read(ref _subscribeAlarmsCount); + + /// The refs passed to the most recent , or null if never called. + public IReadOnlyList? LastSubscribedRefs + { + get { lock (_gate) return _lastSubscribedRefs; } + } + + /// The refs passed to the most recent , or null if never called. + public IReadOnlyList? LastSubscribedAlarmRefs + { + get { lock (_gate) return _lastSubscribedAlarmRefs; } + } + + /// + /// When set, awaits it instead of completing inline — so the actor + /// stays in Connecting for as long as the test wants. Needed because several defects only + /// bite a child that has NOT yet connected (InitializeRequested is unhandled in + /// Connected, so a test built on an already-connected child passes vacuously). + /// + public TaskCompletionSource? InitializeGate { get; set; } + + /// + public async Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken) + { + Interlocked.Increment(ref _initializeCount); + if (InitializeGate is { } gate) await gate.Task.ConfigureAwait(false); + } + + /// + public Task ReinitializeAsync(string driverConfigJson, CancellationToken cancellationToken) => Task.CompletedTask; + + /// + public Task ShutdownAsync(CancellationToken cancellationToken) + { + Interlocked.Increment(ref _shutdownCount); + return Task.CompletedTask; + } + + /// + public DriverHealth GetHealth() => new(DriverState.Healthy, DateTime.UtcNow, null); + + /// + public event EventHandler? OnDataChange; + + /// + public Task SubscribeAsync( + IReadOnlyList fullReferences, TimeSpan publishingInterval, CancellationToken cancellationToken) + { + Interlocked.Increment(ref _subscribeCount); + lock (_gate) _lastSubscribedRefs = fullReferences.ToArray(); + return Task.FromResult(_handle); + } + + /// + public Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken) => Task.CompletedTask; + + /// + public event EventHandler? OnAlarmEvent; + + /// + public Task SubscribeAlarmsAsync( + IReadOnlyList sourceNodeIds, CancellationToken cancellationToken) + { + Interlocked.Increment(ref _subscribeAlarmsCount); + lock (_gate) _lastSubscribedAlarmRefs = sourceNodeIds.ToArray(); + return Task.FromResult(_alarmHandle); + } + + /// + public Task UnsubscribeAlarmsAsync(IAlarmSubscriptionHandle handle, CancellationToken cancellationToken) => Task.CompletedTask; + + /// + public Task AcknowledgeAsync( + IReadOnlyList acknowledgements, CancellationToken cancellationToken) => Task.CompletedTask; + + /// Fires a well-formed data change (the ordinary production path). + /// The reference whose value changed. + /// The new value. + public void FireDataChange(string fullRef, object? value) => + OnDataChange?.Invoke(this, new DataChangeEventArgs( + _handle, fullRef, new DataValueSnapshot(value, 0u, DateTime.UtcNow, DateTime.UtcNow))); + + /// + /// Raises with a NULL Snapshot — the supervision crash seam. + /// DataChangeEventArgs is a positional record with no null validation, and the actor's + /// OnDataChangeForward dereferences msg.Snapshot unguarded ON THE ACTOR THREAD, so this + /// produces a genuine unhandled exception inside the child, which Akka's default decider turns into a + /// Restart. The fault itself is arbitrary; what is under test is the RECOVERY. + /// + /// The reference to name in the malformed event. + public void FireMalformedDataChange(string fullRef) => + OnDataChange?.Invoke(this, new DataChangeEventArgs(_handle, fullRef, null!)); + + /// Raises ; present so the event is not merely written-to-never-read. + /// The condition id to raise. + public void FireAlarm(string conditionId) => + OnAlarmEvent?.Invoke(this, new AlarmEventArgs( + _alarmHandle, conditionId, conditionId, "AlarmCondition", "test", AlarmSeverity.High, DateTime.UtcNow)); + + private sealed class RecordingAlarmHandle : IAlarmSubscriptionHandle + { + /// + public string DiagnosticId => "recording-alarm-sub"; + } +} + +/// +/// An that hands out s and KEEPS +/// every one it created, in creation order. The list — not a single "the driver" reference — is the point: +/// an operator Restart tears the child down and spawns a new one around a NEW driver object, so the +/// assertion subject is Created[1], while a SUPERVISION restart re-uses Created[0] (the Props +/// closure captured it) and part of the assertion is that Created.Count stayed 1. +/// +internal sealed class RecordingSubscribableDriverFactory : IDriverFactory +{ + private readonly string _supportedType; + private readonly Lock _gate = new(); + private readonly List _created = new(); + + private readonly bool _gateInitialize; + + /// Initializes a factory that materialises exactly one driver type. + /// The driver type this factory answers for. + /// When true, every created driver gets an un-completed + /// , so its child actor stays in + /// Connecting until the test completes it. + public RecordingSubscribableDriverFactory(string supportedType, bool gateInitialize = false) + { + _supportedType = supportedType; + _gateInitialize = gateInitialize; + } + + /// Every driver this factory created, in creation order. + public IReadOnlyList Created + { + get { lock (_gate) return _created.ToArray(); } + } + + /// + public IDriver? TryCreate(string driverType, string driverInstanceId, string driverConfigJson) + { + if (!string.Equals(driverType, _supportedType, StringComparison.Ordinal)) return null; + var driver = new RecordingSubscribableDriver(driverInstanceId, driverType); + if (_gateInitialize) + driver.InitializeGate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + lock (_gate) _created.Add(driver); + return driver; + } + + /// + public IReadOnlyCollection SupportedTypes => new[] { _supportedType }; +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Harness/V3ArtifactSeeder.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Harness/V3ArtifactSeeder.cs index 4ffc9a1d..11160db8 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Harness/V3ArtifactSeeder.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Harness/V3ArtifactSeeder.cs @@ -9,6 +9,7 @@ using ZB.MOM.WW.OtOpcUa.Commons.Types; using ZB.MOM.WW.OtOpcUa.Configuration; using ZB.MOM.WW.OtOpcUa.Configuration.Entities; using ZB.MOM.WW.OtOpcUa.Configuration.Enums; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; using ZB.MOM.WW.OtOpcUa.Runtime.Drivers; using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa; @@ -36,16 +37,35 @@ public static class V3ArtifactSeeder /// The revision hash the deployment is sealed under. /// The raw tags to materialise, one per (driver, device, tag). /// UNS references projecting a raw tag (by BackingTag) into an equipment. + /// Whether the seeded DriverInstances are Enabled — i.e. whether applying this + /// artifact actually SPAWNS driver children. Defaults to false, the shape every pre-existing caller + /// relies on: the value-routing suites inject values straight into the host and want no children at all. + /// The restart/re-subscribe suite (03/S16) passes true, because its whole subject is what a spawned + /// child is handed. + /// Optional per-tag TagConfig JSON, keyed by tag name; tags not named + /// here keep the default "{}". The only way to seed an alarm-bearing raw tag (the alarm intent + /// is parsed out of TagConfig.alarm by TagConfigIntent), which the alarm-ref half of the + /// restart suite needs — an alarm-bearing raw tag is excluded from the VALUE subscription set and carried + /// in the alarm one instead, so the two travel through different child code paths. + /// Extra DriverInstance ids carrying NO devices and NO tags. The + /// only way to express a driver that is a running child yet absent from both the value- and alarm-ref + /// maps — the _children.Keys arm of ApplyDesiredSubscriptions's union, which must still + /// hand that child an explicit EMPTY set rather than skipping it. /// The seeded deployment's id. public static DeploymentId SeedV3Deployment( IDbContextFactory db, RevisionHash rev, (string Driver, string DriverName, string Device, string Tag, string DataType, bool Writable)[] rawTags, - (string Area, string Line, string Equip, string BackingTag, string? Effective)[] unsRefs) + (string Area, string Line, string Equip, string BackingTag, string? Effective)[] unsRefs, + bool enabled = false, + IReadOnlyDictionary? tagConfigByTag = null, + string[]? taglessDrivers = null) { var drivers = rawTags .Select(t => (t.Driver, t.DriverName)) .Distinct() - .Select(d => new { DriverInstanceId = d.Driver, RawFolderId = "rf-plant", Name = d.DriverName, DriverType = "Modbus", DriverConfig = "{}", ClusterId = "c1", Enabled = false }) + .Select(d => new { DriverInstanceId = d.Driver, RawFolderId = "rf-plant", Name = d.DriverName, DriverType = "Modbus", DriverConfig = "{}", ClusterId = "c1", Enabled = enabled }) + .Concat((taglessDrivers ?? Array.Empty()) + .Select(id => new { DriverInstanceId = id, RawFolderId = "rf-plant", Name = id, DriverType = "Modbus", DriverConfig = "{}", ClusterId = "c1", Enabled = enabled })) .ToArray(); var devices = rawTags .Select(t => (t.Driver, t.Device)) @@ -60,7 +80,7 @@ public static class V3ArtifactSeeder Name = t.Tag, DataType = t.DataType, AccessLevel = t.Writable ? 1 : 0, // TagAccessLevel.ReadWrite = 1 - TagConfig = "{}", + TagConfig = tagConfigByTag is not null && tagConfigByTag.TryGetValue(t.Tag, out var tc) ? tc : "{}", }).ToArray(); // UNS topology: one area/line per distinct (Area, Line); one equipment per (Area, Line, Equip). @@ -114,13 +134,21 @@ public static class V3ArtifactSeeder /// The host's local node id. /// The deployment to dispatch. /// The revision hash the deployment was sealed under. + /// Optional driver factory forwarded to DriverHostActor.Props. Null (the + /// default, preserving every existing call site) leaves the host on its own default, which materialises no + /// real drivers. Supply one — with enabled: true on the seed — to actually spawn driver children. + /// Optional per-driver supervision-reprovision rate limit forwarded to + /// DriverHostActor.Props; null keeps the production default. Exists so a test can observe the + /// deferral without waiting the production window. /// The host actor plus the publish and mux probes wired into it. public static (IActorRef Actor, TestProbe Publish, TestProbe Mux) SpawnHostAndApply( TestKitBase kit, IDbContextFactory db, - NodeId localNode, DeploymentId deploymentId, RevisionHash rev) + NodeId localNode, DeploymentId deploymentId, RevisionHash rev, + IDriverFactory? driverFactory = null, + TimeSpan? reprovisionBackoff = null) { var publish = kit.CreateTestProbe(); - var (actor, mux, coordinator) = Spawn(kit, db, localNode, publish.Ref); + var (actor, mux, coordinator) = Spawn(kit, db, localNode, publish.Ref, driverFactory, reprovisionBackoff); Dispatch(actor, coordinator, deploymentId, rev); publish.ExpectMsg(RuntimeActorTestBase.PresenceBudget); @@ -140,14 +168,14 @@ public static class V3ArtifactSeeder TestKitBase kit, IDbContextFactory db, NodeId localNode, DeploymentId deploymentId, RevisionHash rev) { - var (actor, mux, coordinator) = Spawn(kit, db, localNode, opcUaPublishActor: null); + var (actor, mux, coordinator) = Spawn(kit, db, localNode, opcUaPublishActor: null, driverFactory: null, reprovisionBackoff: null); Dispatch(actor, coordinator, deploymentId, rev); return (actor, mux); } private static (IActorRef Actor, TestProbe Mux, TestProbe Coordinator) Spawn( TestKitBase kit, IDbContextFactory db, NodeId localNode, - IActorRef? opcUaPublishActor) + IActorRef? opcUaPublishActor, IDriverFactory? driverFactory, TimeSpan? reprovisionBackoff) { var coordinator = kit.CreateTestProbe(); var mux = kit.CreateTestProbe(); @@ -155,11 +183,13 @@ public static class V3ArtifactSeeder var actor = kit.Sys.ActorOf(DriverHostActor.Props( db, localNode, coordinator.Ref, + driverFactory: driverFactory, localRoles: new HashSet { "driver" }, dependencyMux: mux.Ref, opcUaPublishActor: opcUaPublishActor, virtualTagEvaluator: NullVirtualTagEvaluator.Instance, - virtualTagHostOverride: vtHost.Ref)); + virtualTagHostOverride: vtHost.Ref, + reprovisionBackoff: reprovisionBackoff)); return (actor, mux, coordinator); }