diff --git a/docs/AlarmHistorian.md b/docs/AlarmHistorian.md index 26dc8efd..991d01b2 100644 --- a/docs/AlarmHistorian.md +++ b/docs/AlarmHistorian.md @@ -93,6 +93,16 @@ Two postures are deliberate: - **Unset (the default) means drain.** So does an unpublished view. A deployment that runs no redundancy never publishes a role, and defaulting closed would silently stop its alarm history forever. +- **Only a shared queue is gated.** Standing down is safe only because a peer + holds the same rows, so the gate applies only when + `LocalDbReplicationPolicy.SharesRowsWithPeer` holds: + `LocalDb:Replication:ApiKey` **AND** (`LocalDb:Replication:PeerAddress` **OR** + `LocalDb:SyncListenPort > 0`) — an authenticated sync path in either + direction. Any other node always drains its own queue. This is deliberately + stricter than the table-registration predicate (`PeerAddress` OR `ApiKey`): + a key with no dial and no listener (e.g. a fleet-wide secret on an unpaired + node) captures changes but shares them with nobody, and gating it would park + its events behind a Primary that does not hold them (review R1 m3). - **A gate that throws means "not now", never permission.** Draining on a gate fault would put a pair into dual delivery exactly when its redundancy signal is sick. diff --git a/docs/AlarmTracking.md b/docs/AlarmTracking.md index 85f6365c..445037d0 100644 --- a/docs/AlarmTracking.md +++ b/docs/AlarmTracking.md @@ -350,15 +350,16 @@ is client-visible. > (ScadaBridge included) that acks against the Secondary now gets a Bad result > instead of a Good that later reverts — so ack against the Primary (the higher > ServiceLevel). Scripted conditions are **not** gated: their engine runs on -> both nodes, so an ack on the Secondary is real. Residual: the published -> verdict changes only on a redundancy snapshot, so in the boot window of a -> multi-node cluster (no snapshot yet) the node still accepts, while -> `DriverHostActor`'s live check may drop the route — the old flap, limited to -> that window. +> both nodes, so an ack on the Secondary is real. The published verdict +> mirrors `DriverHostActor`'s live check from the moment the host starts, not +> only from the first redundancy snapshot: the host publishes it once at start +> and again whenever cluster membership changes, so in the boot window of a +> pair (role still unknown, a driver peer Up) the gate already denies, exactly +> as the route does (review R1 m2 — previously the view sat on its single-node +> "allow" seed until the first snapshot, reopening the flap for that window). > **Best-effort during reconnect.** If a native ack arrives while the owning -> driver is `Connecting`/`Reconnecting` (or, in the boot window above, on a -> node whose role is not yet known), it is dropped with a debug log and **not** +> driver is `Connecting`/`Reconnecting`, it is dropped with a debug log and **not** > retried upstream — the OPC UA Part 9 `AlarmConditionState` was already > committed locally at step 1, so the rare outcome is "acked locally but not > propagated to AVEVA." Re-issue the ack once the driver is reconnected if @@ -408,8 +409,10 @@ On allow, the handler publishes a `Commons.OpcUa.AlarmCommand` onto the action is a settable `Action` injected at boot by the hosted service. -`OnTimedUnshelve` (the SDK's automatic unshelve timer) bypasses the -operator gate — it is system-initiated. +`OnTimedUnshelve` (the SDK's automatic unshelve timer) is a **no-op** — it +routes nothing (review R1 M2). The engine owns timed-shelve expiry (its 5 s +shelving check records `AutoUnshelve` by `system`), and its re-projection +unshelves the node; see §"`UnshelveTime` is the real remaining time" below. ### One condition event per method call (Gitea #587) @@ -448,7 +451,14 @@ reports nothing. AddComment always reports (the comment is the change), and so does a re-ack **carrying a comment** (Gitea #597): the SDK writes that comment to the node, so it is reported once, with Acked/Confirmed unchanged. Before #597 the - comment landed on the node but no subscriber ever saw it. + comment landed on the node but no subscriber ever saw it. For a **scripted** + condition the comment is also durable (review R1 m1): the routed Acknowledge + reaches an engine that is already acknowledged, and + `Part9StateMachine.ApplyAcknowledge` records a commented re-ack exactly as an + AddComment (kind `AddComment`, the caller as user) — persisted and replicated + with the condition state, the original ack's `LastAck*` fields untouched. An + uncommented re-ack is still a no-op in the engine. The same applies to a + commented re-ack from the AdminUI `/alerts` page. - **A native re-ack is still routed to the driver** (deliberate, #597). The route is fire-and-forget, so a node showing Acked does not prove the upstream system acknowledged — an ack dropped during a reconnect leaves @@ -471,10 +481,14 @@ reports nothing. engine's projection carries its expiry (`AlarmConditionSnapshot.UnshelveAtUtc`), so `ShelvingState/UnshelveTime` counts down to the engine's expiry whichever path shelved it (the AdminUI `/alerts` timed shelve included). This arms the - SDK's own unshelve timer at the real instant; it changes no node state and - only routes an `Unshelve` (user `system`) back to the engine — a no-op if the - engine's 5 s expiry sweep got there first, otherwise it simply lands the - expiry on time. + SDK's own unshelve timer at the real instant, and `OnTimedUnshelve` — the + timer's callback — is deliberately a **no-op** (review R1 M2): it used to + route an operator-style `Unshelve` by `system`, which, firing before the + engine's 5 s expiry sweep, pre-empted the engine's `AutoUnshelve` on every + expiry, from both pair nodes, rewriting the audit trail as a manual unshelve. + The SDK's timer changes no shelving state itself, so the node stays + TimedShelved (with `UnshelveTime` at zero) for up to one sweep after expiry, + until the engine's re-projection reports the one expiry event. - **Accepted, not fixed — the SDK's `EventsMonitored` unit for late conditions** (Gitea #597 item 6, from the #586 review). The SDK's `AreEventsMonitored` counter never reaches a condition materialised under a folder that already had diff --git a/docs/Configuration.md b/docs/Configuration.md index 4f2d478b..58cf17c3 100644 --- a/docs/Configuration.md +++ b/docs/Configuration.md @@ -218,7 +218,7 @@ See the [ClusterClient→gRPC migration design](plans/2026-08-06-clusterclient-t | `GrpcListenPort` | int | `0` | Dedicated **h2c-only** Kestrel port for the artifact gRPC service. `0` = disabled (nothing bound). Must differ from the main HTTP port and from `LocalDb:SyncListenPort` (h2c cannot share a cleartext HTTP/1 port). Mapped only on admin-role nodes. | | `ApiKey` | string | `""` | Shared bearer key the serve-side interceptor checks (constant-time, **fail-closed**: an unset key rejects every call). **Supply via env `ConfigServe__ApiKey` — never commit.** | -**It is a dark switch; the serve side is always wired.** Central maps the `DeploymentArtifactService` behind the `ConfigServeAuthInterceptor` (path-scoped, `FixedTimeEquals`, fail-closed) whenever `ConfigServe:GrpcListenPort > 0`, in both modes — so flipping a node to `FetchAndCache` is a config change, not a redeploy. A `FetchAndCache` node that cannot fetch (all endpoints down, `NotFound`, SHA mismatch) **fails the apply and keeps last-known-good** — a null fetch is "no answer," never an empty configuration (the [#485](plans/2026-07-15-v3-batch4-address-space-plan.md) contract). At boot it restores served state from the LocalDb pointer with no SQL read; an empty or unreadable cache lands Steady-with-no-revision (the first dispatch fetches), never Stale. +**It is a dark switch; the serve side is always wired.** Central maps the `DeploymentArtifactService` behind the `ConfigServeAuthInterceptor` (path-scoped, `FixedTimeEquals`, fail-closed) whenever `ConfigServe:GrpcListenPort > 0`, in both modes — so flipping a node to `FetchAndCache` is a config change, not a redeploy. A `FetchAndCache` node that cannot fetch (all endpoints down, `NotFound`, SHA mismatch) **fails the apply and keeps last-known-good** — a null fetch is "no answer," never an empty configuration (the [#485](plans/2026-07-15-v3-batch4-address-space-plan.md) contract). At boot it restores served state from the LocalDb pointer with no SQL read; an empty or unreadable cache — including a cached row whose bytes are malformed (review R1 m4) — lands Steady-with-no-revision (the first dispatch fetches, even of the same revision), never Stale. **The listener is dedicated h2c and coexists with the LocalDb-sync listener.** A fused admin+driver node with **both** `ConfigServe:GrpcListenPort` and `LocalDb:SyncListenPort` set re-binds its existing HTTP surface exactly once and adds both dedicated HTTP/2-only ports on top — see the same Kestrel-takeover caveat under [`LocalDb`](#localdb-pair-local-config-cache--driver-role-only). @@ -431,7 +431,10 @@ replication stays inert until the sync port + peer are set. `PeerAddress` **OR** `ApiKey` non-blank — the OR because only the *dialling* half sets `PeerAddress`, so keying on it alone would strip change capture from every passive node and leave the pair converging in one direction only. `SyncListenPort` is deliberately **not** part of it (a port with no `ApiKey` is fail-closed -and syncs with nobody). A configured node registers all four tables with `baselineExistingRows: true`, so +and syncs with nobody). (The alarm-historian **drain gate** uses a stricter predicate on purpose — +`ApiKey` AND (`PeerAddress` OR `SyncListenPort > 0`), `LocalDbReplicationPolicy.SharesRowsWithPeer` — because +gating a node that shares its queue with nobody strands its alarm history; see `docs/AlarmHistorian.md` +§"The drain gate".) A configured node registers all four tables with `baselineExistingRows: true`, so turning replication on converges the rows **already in the file** and not merely subsequent writes; an unconfigured node **deregisters** them at boot, dropping capture triggers an earlier build installed. ⚠️ The sync handshake compares registered-table digests **fail-closed**, so flipping either key is a diff --git a/docs/Redundancy.md b/docs/Redundancy.md index e310c190..c6e73f8b 100644 --- a/docs/Redundancy.md +++ b/docs/Redundancy.md @@ -526,13 +526,13 @@ Three data-plane surfaces are **Primary-gated** so a warm standby never touches Before archreview 03/S4 the unknown-role case defaulted to *allow* unconditionally, opening a dual-primary window: a freshly-booted secondary on a multi-node cluster serviced shared-device writes and emitted duplicate fleet-wide alerts for up to the ~10 s `RedundancyStateActor` heartbeat (and indefinitely if the snapshot's NodeId never matched this node). Membership is authoritative much earlier than DPS delivery, so an unknown role on a **multi-driver** cluster now default-**denies**. The driver member count is read live from `Cluster.State.Members` (Up members with the `driver` role); a non-cluster ActorRefProvider yields 0 ⇒ single-node posture. If a snapshot arrives that never mentions this node while a driver peer exists (the 03/S5 identity-mismatch shape), `DriverHostActor` logs a one-time Warning so the silent mismatch is diagnosable. -**What a client sees during the boot window (multi-node, role unknown):** an inbound write is **rejected** (not queued — OT actuation seconds late is a surprise). The reject rides the sanctioned optimistic-write self-correction surface: the node reverts to its prior value with a transient Bad-quality blip and a Part 8 `AuditWriteUpdateEvent`; the client's synchronous write call still returned Good (exactly as any failed device write behaves since the #5 write-outcome self-correction). The rejection reason distinguishes the boot window (`"not primary (role unknown)"`) from a steady-state Secondary (`"not primary"`). Acks are dropped (Warning-logged when role-unknown). The alerts emit is skipped (the Primary peer publishes the single fleet copy). The window closes on the first delivered snapshot. +**What a client sees during the boot window (multi-node, role unknown):** an inbound write is **rejected** (not queued — OT actuation seconds late is a surprise). The reject rides the sanctioned optimistic-write self-correction surface: the node reverts to its prior value with a transient Bad-quality blip and a Part 8 `AuditWriteUpdateEvent`; the client's synchronous write call still returned Good (exactly as any failed device write behaves since the #5 write-outcome self-correction). The rejection reason distinguishes the boot window (`"not primary (role unknown)"`) from a steady-state Secondary (`"not primary"`). Acks are dropped (Warning-logged when role-unknown) — and since review R1 m2 a native OPC UA Acknowledge is refused up front (`BadInvalidState`) in the same window: `DriverHostActor` publishes its Primary verdict to `IDriverPrimaryRoleSource` (the native-ack gate, and the MQTT Sparkplug primary-host claim) once at start and again on every cluster membership change, not only on redundancy snapshots, so the published verdict denies as soon as a driver peer is Up rather than sitting on the single-node "allow" seed until the first snapshot. The alerts emit is skipped (the Primary peer publishes the single fleet copy). The window closes on the first delivered snapshot. Under warm/hot redundancy both cluster nodes run `ScriptedAlarmHostActor` and evaluate scripted alarms, keeping each node's address space and engine state warm for instant failover. To avoid duplicate rows on `/alerts` and duplicate historian writes, only the Primary node publishes externally: - **`alerts` topic emission** — `ScriptedAlarmHostActor` and `DriverHostActor.ForwardNativeAlarm` subscribe to the `redundancy-state` DPS topic and cache the local node's `RedundancyRole`, then gate the cluster `alerts` publish through `PrimaryGatePolicy` (table above). The OPC UA condition-node write and inbound ack/shelve command processing remain **ungated** on both nodes so the secondary is always ready to serve clients after a failover. - **`HistorianAdapterActor` historization** — likewise Primary-gated so alarm historization is exactly-once across all alarm sources. The actor subscribes to the `alerts` DPS topic and translates each `AlarmTransitionEvent` → `AlarmHistorianEvent` before enqueuing it on the sink; scripted alarms therefore historize exactly once regardless of cluster size. -- **The alarm sink's DRAIN** — a second, independent gate on the same decision. The enqueue gate above is not sufficient any more: since Phase 2 the store-and-forward buffer lives in the replicated LocalDb, so the Secondary holds a full copy of the Primary's queued rows and an ungated drain there would re-deliver every event continuously. `LocalDbStoreAndForwardSink` takes a `drainGate` fed from `IRedundancyRoleView` — a singleton `DriverHostActor` publishes its `PrimaryGatePolicy` verdict to, because the drain runs on a timer and cannot receive `RedundancyStateChanged` itself. A gated-off node reports `HistorianDrainState.NotPrimary`. Delivery is **at-least-once across a failover** by design: rows the old Primary delivered just before dying may not have replicated their deletes yet. See [AlarmHistorian.md](AlarmHistorian.md). +- **The alarm sink's DRAIN** — a second, independent gate on the same decision. The enqueue gate above is not sufficient any more: since Phase 2 the store-and-forward buffer lives in the replicated LocalDb, so the Secondary holds a full copy of the Primary's queued rows and an ungated drain there would re-deliver every event continuously. `LocalDbStoreAndForwardSink` takes a `drainGate` fed from `IRedundancyRoleView` — a singleton `DriverHostActor` publishes its `PrimaryGatePolicy` verdict to, because the drain runs on a timer and cannot receive `RedundancyStateChanged` itself. A gated-off node reports `HistorianDrainState.NotPrimary`. The gate applies only to a node whose queue a peer actually holds — `LocalDbReplicationPolicy.SharesRowsWithPeer`: `LocalDb:Replication:ApiKey` AND (`PeerAddress` OR `LocalDb:SyncListenPort` > 0); any other node always drains (review R1 m3). Delivery is **at-least-once across a failover** by design: rows the old Primary delivered just before dying may not have replicated their deletes yet. See [AlarmHistorian.md](AlarmHistorian.md). - **Continuous-historization value CAPTURE** (Gitea #537) — a third gate on the same role rule, and the one that gates the *opposite* half of its pipeline. Both pair nodes serve the same address space and see @@ -622,7 +622,10 @@ and [AlarmHistorian.md](AlarmHistorian.md). Phase 1 caches the **deployed-config artifact** (chunked). Its job is a single failure mode redundancy does not otherwise cover: a driver node restarting into a **central-SQL-Server outage** would come up with no configuration at all. With the cache, it **boots from the last artifact it applied** instead, and logs a running-from-cache -signal. +signal. A cached artifact whose bytes are empty or malformed is **no answer** (#601): nothing is applied, +the node does **not** adopt that revision or claim running-from-cache, and it goes **Stale** (retrying the +ConfigDb) — so a later dispatch of that same revision really applies instead of being acked as already +current (review R1 m4). A `FetchAndCache` node lands Steady-with-no-revision instead, as for an empty cache. - **What replicates.** The cache can *optionally* replicate to the node's redundant pair peer over the LocalDb library's gRPC sync (HLC-stamped, last-writer-wins). Two tables replicate: diff --git a/docs/ScriptedAlarms.md b/docs/ScriptedAlarms.md index 9b2c467c..b7b61e4f 100644 --- a/docs/ScriptedAlarms.md +++ b/docs/ScriptedAlarms.md @@ -118,7 +118,8 @@ rationale is recorded in the engine's class doc so it is not re-proposed: - `ApplyOneShotShelve` / `ApplyTimedShelve(unshelveAtUtc)` / `ApplyUnshelve` — shelving transitions. `Timed` requires `unshelveAtUtc > nowUtc`. - `ApplyEnable` / `ApplyDisable` — operator enable/disable. Disabled alarms ignore predicate results until re-enabled; on enable, `ActiveState` is re-derived from the next evaluation. - `ApplyAddComment(text)` — append-only audit entry, no state change. -- `ApplyShelvingCheck(nowUtc)` — called by the 5s timer; promotes expired `Timed` shelving to `Unshelved` with a `system / AutoUnshelve` audit entry. +- `ApplyShelvingCheck(nowUtc)` — called by the 5s timer; promotes expired `Timed` shelving to `Unshelved` with a `system / AutoUnshelve` audit entry. The sole owner of timed-shelve expiry — the OPC UA node's own SDK timer does nothing. +- `ApplyAcknowledge` on an already-acknowledged alarm is a no-op **unless it carries a comment**, which is recorded as an `AddComment` (emission `CommentAdded`; `LastAck*` untouched) — the OPC UA node reports a commented re-ack, so the system of record must hold the comment too. Two invariants the machine enforces: @@ -354,11 +355,11 @@ On allow, the handler publishes a `Commons.OpcUa.AlarmCommand` (containing comma - **Scripted conditions** — all eight Part 9 operations (including Enable/Disable) route through `AlarmCommandRouter` onto the `alarm-commands` topic, which `ScriptedAlarmHostActor` dispatches to the engine (`EnableAsync` / `DisableAsync` / `AcknowledgeAsync` / …). On enable, `ActiveState` is re-derived from the next predicate evaluation. - **Native (driver-fed) conditions** — `OnAcknowledge` branches to `NativeAlarmAckRouter` and routes the ack to the owning driver rather than the scripted engine (see §[Native-alarm OPC UA operator operations](#native-alarm-opc-ua-operator-operations) above). `OnEnableDisable`, `OnConfirm` and `OnShelve` return `BadNotSupported` immediately — a native condition supports Acknowledge and AddComment only (no engine-side surface, no upstream path, and the driver projection owns its Confirmed/Shelving; Gitea #587 review). -`OnTimedUnshelve` (the SDK's internal auto-unshelve timer) bypasses the client gate — it is system-initiated and not subject to operator role checks. +`OnTimedUnshelve` (the SDK's internal auto-unshelve timer, armed for the real remaining time since Gitea #597) is a **no-op**: the engine owns timed-shelve expiry via `ApplyShelvingCheck` (`system / AutoUnshelve`), and routing an `Unshelve` from the timer pre-empted that on every expiry as an operator-style unshelve from both pair nodes (review R1 M2). ### Delta-gate de-duplication -`WriteAlarmCondition` fires a Part 9 condition event only when the incoming state differs from the node's current live state. For an OPC UA method call (Acknowledge, Confirm, AddComment, the shelve methods, Enable/Disable) the SDK applies the new state to the node, and the node manager then reports **the one** condition event for it from the method path (Gitea #587 — the SDK reports none of its own, since `AutoReportStateChanges` stays false; the shelve methods' state is applied by the node manager, because the SDK's never applies it; a call that changed nothing — e.g. an uncommented re-ack — reports nothing; a re-ack carrying a comment reports that comment once, Gitea #597). Native (driver-fed) conditions support only Acknowledge and AddComment; their Confirm, shelve and Enable/Disable methods return `BadNotSupported`, and their Acknowledge is refused with `BadInvalidState` on a node that is not the redundancy Primary (Gitea #597). A timed shelve's `UnshelveTime` counts down to the engine's expiry (the projection carries `UnshelveAtUtc`). The engine's later re-projection of the same transition finds the node already in that state and is suppressed, so each accepted call yields exactly one event. See `docs/AlarmTracking.md` §"One condition event per method call". +`WriteAlarmCondition` fires a Part 9 condition event only when the incoming state differs from the node's current live state. For an OPC UA method call (Acknowledge, Confirm, AddComment, the shelve methods, Enable/Disable) the SDK applies the new state to the node, and the node manager then reports **the one** condition event for it from the method path (Gitea #587 — the SDK reports none of its own, since `AutoReportStateChanges` stays false; the shelve methods' state is applied by the node manager, because the SDK's never applies it; a call that changed nothing — e.g. an uncommented re-ack — reports nothing; a re-ack carrying a comment reports that comment once, Gitea #597, and the engine records it as an `AddComment` so it is persisted and replicated, review R1 m1). Native (driver-fed) conditions support only Acknowledge and AddComment; their Confirm, shelve and Enable/Disable methods return `BadNotSupported`, and their Acknowledge is refused with `BadInvalidState` on a node that is not the redundancy Primary (Gitea #597). A timed shelve's `UnshelveTime` counts down to the engine's expiry (the projection carries `UnshelveAtUtc`). The engine's later re-projection of the same transition finds the node already in that state and is suppressed, so each accepted call yields exactly one event. See `docs/AlarmTracking.md` §"One condition event per method call". ### AdminUI path diff --git a/docs/Telemetry.md b/docs/Telemetry.md index 79d00c5a..b4430173 100644 --- a/docs/Telemetry.md +++ b/docs/Telemetry.md @@ -255,6 +255,11 @@ driver node**: reconnect loses whatever transitions happened while disconnected, same as a DPS resubscribe would. The cache is evicted per instance (`ITelemetryLocalHub.Evict`) when `DriverHostActor.StopChild` removes a driver from the plan (#609), so a decommissioned instance's last state is not replayed to new subscribers. + `StopChild` first releases the instance's resilience state (`IDriverCapabilityInvokerFactory.Release` → + pipeline-builder + `DriverResilienceStatusTracker.Invalidate`): `DriverResilienceStatusPublisherService` + re-emits the whole tracker every 5 s, so evicting only the hub was undone within one tick (review R1 M1). + Residual: a Health emit already in flight from the stopping child can land after the eviction and + re-cache that one snapshot — a narrow race, not a recurring replay. - **Connection indicator** — the `/alerts` and `/script-log` live/disconnected pill is driven by **aggregate** stream health across all dialers: connected when **at least one** node stream is up, disconnected when all are down. This matches today's DPS `SubscribeAck`/`PostStop` pill diff --git a/docs/operations/2026-07-20-localdb-pair-replication.md b/docs/operations/2026-07-20-localdb-pair-replication.md index 2b30f253..b61a94ed 100644 --- a/docs/operations/2026-07-20-localdb-pair-replication.md +++ b/docs/operations/2026-07-20-localdb-pair-replication.md @@ -92,7 +92,9 @@ Replication has two knobs, both under `LocalDb`: > 2026-08-18 `LocalDbSetup.OnReady` installs the change-capture triggers only when the node is > configured to replicate — the predicate is **`PeerAddress` OR `ApiKey` non-blank** (the OR because > the passive node has no `PeerAddress`; `SyncListenPort` is deliberately not part of it, since a -> port with no key is fail-closed and syncs with nobody). Consequences at enablement time: +> port with no key is fail-closed and syncs with nobody). The alarm-historian drain gate uses the +> stricter `ApiKey` AND (`PeerAddress` OR `SyncListenPort > 0`) — so set the key **and** the port/peer +> together, or the node keeps draining its own queue. Consequences at enablement time: > > - A configured node registers with **`baselineExistingRows: true`**, so the rows **already in the > file** are seeded into the row-version ledger at the LWW floor (HLC `0`) and actually travel. diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/ClusterRoleInfo.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/ClusterRoleInfo.cs index 9cf82799..44833637 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/ClusterRoleInfo.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/ClusterRoleInfo.cs @@ -76,7 +76,7 @@ public sealed class ClusterRoleInfo : IClusterRoleInfo, IDisposable } } - SeedFromCurrentState(); + SeedFrom(_cluster.State); _subscriber = system.ActorOf(Props.Create(() => new SubscriberActor(this)), "clusterroleinfo-subscriber"); } @@ -121,12 +121,19 @@ public sealed class ClusterRoleInfo : IClusterRoleInfo, IDisposable /// Occurs when the leader for a role changes. public event EventHandler? RoleLeaderChanged; - private void SeedFromCurrentState() + /// Seeds the member/leader maps from a cluster-state snapshot (the constructor passes the live + /// Cluster.State; exposed for tests). + /// The cluster-state snapshot. + internal void SeedFrom(ClusterEvent.CurrentClusterState snapshot) { - var snapshot = _cluster.State; lock (_lock) { - foreach (var member in snapshot.Members) + // Cluster-002 (review R1 m6): seed only USABLE members. A member already Leaving / Exiting / Down + // (or still Joining) at seed time is not one, and a departing member never fires a later + // MemberLeft / MemberExited / MemberDowned that HandleMemberEvent would drop it on — it would + // linger until MemberRemoved, the very delay the member-event handling exists to avoid. A + // Joining member is added by its MemberUp. + foreach (var member in snapshot.Members.Where(m => m.Status is MemberStatus.Up or MemberStatus.WeaklyUp)) foreach (var role in member.Roles) { if (!_membersByRole.TryGetValue(role, out var set)) diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/IDriverCapabilityInvokerFactory.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/IDriverCapabilityInvokerFactory.cs index 6c28bdae..c27ea515 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/IDriverCapabilityInvokerFactory.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/IDriverCapabilityInvokerFactory.cs @@ -28,6 +28,16 @@ public interface IDriverCapabilityInvokerFactory /// /// A per-instance . IDriverCapabilityInvoker Create(string driverInstanceId, string driverType, string? resilienceConfigJson = null); + + /// + /// Release every piece of shared resilience state held for — called + /// by the host when the instance leaves the deployed plan (#609). A concrete factory drops the + /// instance's cached pipelines and its status-tracker snapshots, so the periodic resilience publisher + /// stops re-emitting (and re-priming new telemetry subscribers with) a decommissioned driver's last + /// breaker/failure counters. Default: no-op (a factory that holds no state). + /// + /// The driver instance that was removed. + void Release(string driverInstanceId) { } } /// diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms/Part9StateMachine.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms/Part9StateMachine.cs index 1de8baf3..e8498c1c 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms/Part9StateMachine.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms/Part9StateMachine.cs @@ -105,7 +105,17 @@ public static class Part9StateMachine throw new ArgumentException("User identity required for audit.", nameof(user)); if (current.Acked == AlarmAckedState.Acknowledged) - return TransitionResult.NoOp(current, "already acknowledged"); + { + // A COMMENTED re-ack carries a real comment even though nothing is acknowledged (Gitea #597 review + // m1): the OPC UA node reports it — the SDK writes the comment on a re-ack — so the system of + // record must hold it too, or the event advertises a comment that exists only on the node that + // served the call (never persisted, never replicated to the pair peer). Recorded exactly like + // AddComment; the ack fields (LastAck*) stay those of the real acknowledge. An uncommented re-ack + // is still a no-op. + return string.IsNullOrWhiteSpace(comment) + ? TransitionResult.NoOp(current, "already acknowledged") + : ApplyAddComment(current, user, comment, nowUtc); + } var audit = AppendComment(current.Comments, nowUtc, user, "Acknowledge", comment); var next = current with diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core/Resilience/DriverCapabilityInvokerFactory.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core/Resilience/DriverCapabilityInvokerFactory.cs index e0a0584c..44ffb21b 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core/Resilience/DriverCapabilityInvokerFactory.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core/Resilience/DriverCapabilityInvokerFactory.cs @@ -90,4 +90,16 @@ public sealed class DriverCapabilityInvokerFactory : IDriverCapabilityInvokerFac statusTracker: _statusTracker, optionsGeneration: generation); } + + /// + /// + /// #609: without this the tracker kept a removed instance's snapshots forever (only + /// invalidated, and a removed instance is never re-created), and the resilience publisher re-emitted + /// them every tick — undoing the host's telemetry-hub eviction within one interval. + /// + public void Release(string driverInstanceId) + { + _builder.Invalidate(driverInstanceId); + _statusTracker.Invalidate(driverInstanceId); + } } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/LocalDbRegistration.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/LocalDbRegistration.cs index a26ac5bb..20173d34 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/LocalDbRegistration.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/LocalDbRegistration.cs @@ -46,7 +46,7 @@ public static class LocalDbRegistration /// Configuration key carrying the dedicated h2c sync listener's port. Zero or absent means /// no listener is bound and the passive sync endpoint is not mapped. /// - public const string SyncListenPortKey = "LocalDb:SyncListenPort"; + public const string SyncListenPortKey = Runtime.LocalDbReplicationPolicy.SyncListenPortKey; /// /// The port the dedicated cleartext-HTTP/2 sync listener should bind, or 0 when the diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/LocalDbSetup.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/LocalDbSetup.cs index 7bc8954a..198b99b9 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/LocalDbSetup.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/LocalDbSetup.cs @@ -170,9 +170,9 @@ public static class LocalDbSetup /// true when this node is configured to replicate. public static bool ReplicationIsConfigured(IConfiguration configuration) { - // Shared with the alarm-historian drain gate (#603): one predicate, so the nodes that capture - // changes and the nodes that stand down for a peer can never disagree. - return Runtime.LocalDbReplicationPolicy.IsConfigured(configuration); + // Defined beside the alarm-historian drain gate's stricter predicate (#603, review R1 m3): every node + // the gate treats as sharing rows also captures, so the two never disagree in the harmful direction. + return Runtime.LocalDbReplicationPolicy.CapturesChanges(configuration); } /// Opts every replicated table into change capture, baselining what is already there. diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs index 6f7e00e0..7636529b 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs @@ -749,7 +749,10 @@ if (LogPipelineDiagnostics.CheckFileSinkGrowth(app.Configuration, await app.RunAsync(); } -catch (Exception ex) +// HostAbortedException is how design-time tooling (EF `dotnet ef … --startup-project`, OpenAPI generators — +// HostFactoryResolver) stops Program once it has the built host. It is not a failure: letting it reach the +// Fatal + Environment.Exit(1) below would kill the tool's own process (#618, review R1 m5). +catch (Exception ex) when (ex is not HostAbortedException) { // Log.Logger is still the silent bootstrap logger when the failure happened before AddZbSerilog // ran (e.g. EffectiveRoles.Resolve), so ALSO write to stderr — the one channel that always exists. diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs index f233abeb..0b22c4ea 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs @@ -874,8 +874,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 // (shelved, oneShot) flag pair. Timed shelving's expiry is owned by the engine; the time // passed here is the engine's remaining time, so the node's UnshelveTime counts down to the // real expiry (Gitea #597 — a 0 here reads as MaxTimeShelved, ~24.8 days). The SDK timer it - // arms only routes an Unshelve back to the engine (OnTimedUnshelve), a no-op once the - // engine's own expiry has run. + // arms does nothing (OnTimedUnshelve is a no-op): the engine's own expiry re-projects. condition.SetShelvingState( SystemContext, shelved: state.Shelving != AlarmShelvingKind.Unshelved, @@ -1375,25 +1374,17 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 : ("TimedShelve", DateTime.UtcNow + TimeSpan.FromMilliseconds(shelvingTime)); return HandleAlarmCommand(context, condition, operation, comment: null, unshelveAt); }; - // The auto-unshelve timer callback is SDK-initiated (the TimedShelve duration expired); the SDK - // fires it with the node manager's system context — there is NO session and NO user identity. - // Routing through HandleAlarmCommand would hit the AlarmAck gate and return BadUserAccessDenied, - // leaving the alarm permanently shelved. Instead, bypass the client gate, extract the AlarmId the - // same way HandleAlarmCommand does, and route an Unshelve command so the engine clears its shelve - // state. The manual-client Unshelve path goes through OnShelve(shelving:false) and stays gated. - alarm.OnTimedUnshelve = (context, condition) => - { - var alarmId = condition.NodeId.Identifier?.ToString() ?? string.Empty; - // User MUST be non-empty: the engine's Part9StateMachine.ApplyUnshelve rejects a - // null/whitespace user (ArgumentException "User required."), and that exception is swallowed - // downstream — an empty user would make the timed auto-unshelve silently no-op, so a - // TimedShelve would never auto-expire. There is no client principal on a system-timer - // unshelve, so label it the canonical engine-internal "system" user (matching the engine's - // own AutoUnshelve audit user and the codebase-wide "system" system-actor convention). The - // AlarmAck gate bypass is intentional and preserved — this is a session-less SDK timer. - AlarmCommandRouter?.Invoke(new AlarmCommand(alarmId, "Unshelve", "system", null, null)); - return ServiceResult.Good; - }; + // The SDK's auto-unshelve timer (armed by SetShelvingState for the real remaining time since Gitea + // #597) fires with a session-less system context. It is deliberately a NO-OP: the engine owns + // timed-shelve expiry (ScriptedAlarmEngine's shelving check → Part9StateMachine.ApplyShelvingCheck, + // recorded as "AutoUnshelve" by "system" with its "Timed shelving expired" comment), and its + // re-projection is what unshelves this node. Routing an operator-style "Unshelve" from here used to + // pre-empt that check on every expiry — from BOTH pair nodes, rewriting the audit trail as a manual + // unshelve — and a timer callback already queued when a re-shelve disposed it could unshelve the + // fresh shelve. The SDK's own OnTimerExpired changes no shelving state itself (it only refreshes + // UnshelveTime after this callback), so returning without routing leaves the node shelved until + // the engine's expiry lands. Native conditions are never timed-shelved (BadNotSupported above). + alarm.OnTimedUnshelve = (_, _) => ServiceResult.Good; // H4 — inbound Part 9 Enable/Disable (the condition type's built-in Enable/Disable methods route // here via this delegate). The engine handles Enable/Disable for SCRIPTED alarms @@ -1596,9 +1587,11 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 /// Only a real change is reported. The condition's gate-relevant state is read before and after /// the SDK handler, and nothing is reported when it did not move. The SDK accepts acknowledging an /// acknowledged condition and sets Confirmed = false again; a re-ack therefore also restores the - /// pre-call Confirmed and Retain, so a scripted node never drifts from its engine (which ignores a - /// re-ack) and the call is silent — unless it carried a comment, which the SDK writes and which is then - /// reported once like AddComment (Gitea #597). AddComment always reports: adding a comment is itself + /// pre-call Confirmed and Retain, so a scripted node never drifts from its engine (which ignores an + /// uncommented re-ack) and the call is silent — unless it carried a comment, which the SDK writes and + /// which is then reported once like AddComment (Gitea #597). The scripted engine records that comment + /// too (Part9StateMachine.ApplyAcknowledge treats a commented re-ack as AddComment), so the + /// reported comment is persisted and replicated, not node-local. AddComment always reports: adding a comment is itself /// the change. A NATIVE re-ack is still routed to the driver: the route is fire-and-forget, so a node /// showing Acked does not prove the upstream system acknowledged, and re-acking is the documented retry. /// @@ -1635,7 +1628,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 // The requested duration, so UnshelveTime counts down to the real expiry (Gitea #597 — 0 reads as // MaxTimeShelved, ~24.8 days). The scripted engine still owns the expiry and re-projects Unshelved // when it passes (with its own remaining time meanwhile — WriteAlarmCondition); the SDK timer this - // arms only routes an Unshelve back to it (OnTimedUnshelve), a no-op once the engine has expired it. + // arms does nothing (OnTimedUnshelve is a no-op — the engine's AutoUnshelve owns the expiry). timedShelve.OnCall = (context, method, objectId, shelvingTime) => ReportAfterMethod( () => timed(context, method, objectId, shelvingTime), () => alarm.SetShelvingState(SystemContext, shelved: true, oneShot: false, shelvingTime: shelvingTime)); 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 21a6d84e..f9c3b292 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs @@ -788,9 +788,91 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS // Same rationale for the ScriptedAlarm host — the bootstrap-restore path Tells it // ApplyScriptedAlarms through the same PushDesiredSubscriptions pass. SpawnScriptedAlarmHost(); + StartPrimaryVerdictPublishing(); Bootstrap(); } + /// + /// Review R1 m2 (Gitea #597): the published Primary data-plane verdict ( + /// — the native-ack gate and the Sparkplug primary-host claim) must mirror + /// from the moment this host starts, not only from the first redundancy snapshot. Before this, the view sat + /// on its single-node seed (allow) through the whole boot window, so on a pair BOTH nodes accepted a native + /// ack while this host — which evaluates the member count live — dropped the upstream route on the + /// non-Primary: the acked → unacked revert #597 targets. So publish once now, and again whenever cluster + /// membership changes (a driver peer coming Up turns an unknown role from allow into deny before any + /// snapshot names this node). A non-cluster ActorRefProvider (legacy/test harnesses) just keeps the + /// initial publish. + /// + private void StartPrimaryVerdictPublishing() + { + if (_primaryRoleSource is null) return; + PublishPrimaryVerdict(); + try + { + Akka.Cluster.Cluster.Get(Context.System).Subscribe( + Self, Akka.Cluster.ClusterEvent.SubscriptionInitialStateMode.InitialStateAsSnapshot, + typeof(Akka.Cluster.ClusterEvent.IMemberEvent)); + _subscribedToMemberEvents = true; + } + catch (Exception) + { + // Non-cluster ActorRefProvider — no membership to follow; DriverMemberCount() reads 0 there too. + } + } + + /// A cluster member changed status: re-publish the Primary verdict now and once more shortly after. + /// The deferred pass covers the cluster read-view (what reads) updating on its + /// own subscriber AFTER this event was delivered here — the two are not ordered. + private void OnDriverMembershipChanged(Akka.Cluster.ClusterEvent.IMemberEvent _) + { + PublishPrimaryVerdictIfChanged(); + // One keyed timer: a burst of member events collapses into a single deferred pass. + Timers.StartSingleTimer("republish-primary-verdict", RepublishPrimaryVerdict.Instance, MembershipRepublishDelay); + } + + /// Publishes to the primary-role source (no-op when none is wired). + /// The snapshot path publishes on every snapshot, as the other verdicts do. + private void PublishPrimaryVerdict() + { + if (_primaryRoleSource is null) return; + var verdict = ShouldServiceAsPrimary(); + _lastPublishedPrimaryVerdict = verdict; + _primaryRoleSource.PublishPrimaryRole(verdict); + } + + /// The membership path publishes only a CHANGED verdict — member events are frequent and carry no + /// role information, so republishing an unchanged verdict on each would be noise. + private void PublishPrimaryVerdictIfChanged() + { + if (_primaryRoleSource is null || ShouldServiceAsPrimary() == _lastPublishedPrimaryVerdict) return; + PublishPrimaryVerdict(); + } + + /// + protected override void PostStop() + { + if (_subscribedToMemberEvents) + { + try { Akka.Cluster.Cluster.Get(Context.System).Unsubscribe(Self); } + catch (Exception) { /* system shutting down */ } + } + base.PostStop(); + } + + /// Self-message: re-evaluate and re-publish the Primary verdict (see ). + internal sealed class RepublishPrimaryVerdict + { + /// The shared instance. + public static readonly RepublishPrimaryVerdict Instance = new(); + private RepublishPrimaryVerdict() { } + } + + /// How long after a membership event the Primary verdict is re-published a second time. + private static readonly TimeSpan MembershipRepublishDelay = TimeSpan.FromSeconds(1); + + private bool _subscribedToMemberEvents; + private bool? _lastPublishedPrimaryVerdict; + /// /// Spawns the single child that owns the Equipment-namespace /// VirtualTagActors and bridges their results onto the OPC UA publish actor. A test-supplied @@ -990,12 +1072,24 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS var deploymentId = DeploymentId.Parse(cached.DeploymentId); var revision = RevisionHash.Parse(cached.RevisionHash); - _currentRevision = revision; - BecomeSteady(); // Restore drivers + address space + subscriptions from the cached bytes — no SQL read, no // re-ack (the deployment is already Applied). Same in-hand-bytes apply as the Direct-mode // cache boot, but this is the steady-state config source, not an outage fallback. - ApplyCachedArtifact(deploymentId, cached.Artifact); + if (!ApplyCachedArtifact(deploymentId, cached.Artifact)) + { + // #601 / review R1 m4: unusable cached bytes applied nothing, so they must not become the + // current revision — that would ack a re-dispatch of the same revision as already applied + // without ever applying it. Steady-with-no-revision, exactly as for an empty cache. + _log.Warning( + "DriverHost {Node}: FetchAndCache boot — the cached artifact for deployment {Id} (rev {Rev}) " + + "is unreadable; entering Steady (no revision). The next dispatch will fetch.", + _localNode, deploymentId, revision); + BecomeSteady(); + return; + } + + _currentRevision = revision; + BecomeSteady(); _log.Info( "DriverHost {Node}: FetchAndCache boot — restored served state for deployment {Id} " + @@ -1059,14 +1153,24 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS var deploymentId = DeploymentId.Parse(cached.DeploymentId); var revision = RevisionHash.Parse(cached.RevisionHash); + // #601 / review R1 m4: unusable cached bytes applied nothing — the node is NOT serving a + // configuration, so it must not claim RUNNING FROM CACHE, adopt the revision (a re-dispatch of + // it would then be acked without being applied) or skip Stale's ConfigDb retry loop. + if (!ApplyCachedArtifact(deploymentId, cached.Artifact)) + { + _log.Error( + "DriverHost {Node}: the cached artifact for deployment {Id} (rev {Rev}) is unreadable; " + + "not booting from it — entering Stale with no configuration.", + _localNode, deploymentId, revision); + return false; + } + // Steady, not Stale: this node is serving a real configuration. The retry-db timer that // Stale would have started does not run here, so recovery rides on the next dispatch — // matching how a normally-booted node behaves. _currentRevision = revision; BecomeSteady(); - ApplyCachedArtifact(deploymentId, cached.Artifact); - _isRunningFromCache = true; _log.Warning( "DriverHost {Node}: RUNNING FROM CACHE — central ConfigDb is unreachable, so this node " + @@ -1094,16 +1198,16 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS /// UpsertNodeDeploymentState and SendAck for the same reason: both write to or /// depend on central. /// - private void ApplyCachedArtifact(DeploymentId deploymentId, byte[] blob) + /// when the bytes were empty or malformed and nothing was applied — the + /// caller must not adopt the cached revision (#601, review R1 m4). + private bool ApplyCachedArtifact(DeploymentId deploymentId, byte[] blob) { var correlation = CorrelationId.NewId(); - // #601: null = the cached bytes were empty or malformed — no answer, not an empty configuration. - // Keep last-known-good: no address-space rebuild and no subscription push from unusable bytes. // #601: null = the cached bytes were empty or malformed — no answer, not an empty configuration. // Keep last-known-good: no address-space rebuild and no subscription push from unusable bytes. if (ReconcileDriversFromBlob(deploymentId, blob) is null) - return; + return false; // Hand the cached blob to the rebuild so it materialises the client-facing address space from // it directly. Passing only the deploymentId would drive a ConfigDb read — unreachable here by @@ -1113,6 +1217,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS new ZB.MOM.WW.OtOpcUa.Runtime.OpcUa.OpcUaPublishActor.RebuildAddressSpace(correlation, deploymentId, blob)); PushDesiredSubscriptionsFromArtifact(deploymentId, blob); + return true; } /// @@ -1170,6 +1275,9 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS Receive(HandleRouteNodeWrite); Receive(HandleRouteNativeAlarmAck); Receive(OnRedundancyStateChanged); + Receive(OnDriverMembershipChanged); + Receive(_ => PublishPrimaryVerdictIfChanged()); + Receive(_ => PublishPrimaryVerdictIfChanged()); // The redundancy-state topic also carries OpcUaProbeResult (OpcUaPublishActor peer-probes). // We don't consume it here — drop it so it doesn't dead-letter (matches PeerProbeSupervisor). Receive(_ => { }); @@ -1228,6 +1336,9 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS Receive(HandleRouteNodeWrite); Receive(HandleRouteNativeAlarmAck); Receive(OnRedundancyStateChanged); + Receive(OnDriverMembershipChanged); + Receive(_ => PublishPrimaryVerdictIfChanged()); + Receive(_ => PublishPrimaryVerdictIfChanged()); // The redundancy-state topic also carries OpcUaProbeResult (OpcUaPublishActor peer-probes). // We don't consume it here — drop it so it doesn't dead-letter (matches PeerProbeSupervisor). Receive(_ => { }); @@ -1975,7 +2086,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS // drain verdict published immediately above. A driver reading this claims an exclusive identity // on a shared external system (Gitea #511: the Sparkplug host id), where two claimants is the // harm, so an unknown role with a driver peer present must deny. - _primaryRoleSource?.PublishPrimaryRole(ShouldServiceAsPrimary()); + PublishPrimaryVerdict(); } private void Stale() @@ -1998,6 +2109,9 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS Receive(m => ReprovisionChild(m.Request, m.Child)); Receive(HandleReconnectDriver); Receive(OnRedundancyStateChanged); + Receive(OnDriverMembershipChanged); + Receive(_ => PublishPrimaryVerdictIfChanged()); + Receive(_ => PublishPrimaryVerdictIfChanged()); // The redundancy-state topic also carries OpcUaProbeResult (OpcUaPublishActor peer-probes). // We don't consume it here — drop it so it doesn't dead-letter (matches PeerProbeSupervisor). Receive(_ => { }); @@ -3233,7 +3347,10 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS Context.Stop(entry.Actor); _children.Remove(driverInstanceId); // #609: the instance left the plan (StopChild is only reached via plan.ToStop), so its cached - // Health/Resilience snapshots must not replay to new telemetry subscribers. + // Health/Resilience snapshots must not replay to new telemetry subscribers. Release the shared + // resilience state FIRST: the resilience publisher re-emits the tracker's snapshots every tick, so + // evicting the hub alone was undone within one interval. + _invokerFactory.Release(driverInstanceId); _telemetryHub?.Evict(driverInstanceId); _log.Info("DriverHost {Node}: stopped driver child {Id}", _localNode, driverInstanceId); } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/LocalDbReplicationPolicy.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/LocalDbReplicationPolicy.cs index e4106f55..da09c3e0 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/LocalDbReplicationPolicy.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/LocalDbReplicationPolicy.cs @@ -3,26 +3,36 @@ using Microsoft.Extensions.Configuration; namespace ZB.MOM.WW.OtOpcUa.Runtime; /// -/// The ONE definition of "this node's LocalDb is replicated to a peer" (#603). Two consumers must agree -/// on it, and used to carry separate predicates that diverged: the Host's table-registration step -/// (LocalDbSetup.ReplicationIsConfigured - which nodes capture changes) and the alarm-historian -/// drain gate (AddAlarmHistorian - which nodes stand down in favour of a peer). Lives in Runtime -/// because Host references Runtime and not the reverse. +/// The definitions of "this node's LocalDb is replicated to a peer" (#603) — one place, two deliberately +/// different predicates for the two consumers: the Host's table-registration step +/// (LocalDbSetup.ReplicationIsConfigured → : which nodes capture +/// changes) and the alarm-historian drain gate (AddAlarmHistorian → : +/// which nodes stand down in favour of a peer). Lives in Runtime because Host references Runtime and not +/// the reverse. /// /// /// -/// Predicate: LocalDb:Replication:PeerAddress OR LocalDb:Replication:ApiKey is -/// non-blank. The OR is required: only the dialling half of a pair sets PeerAddress, while -/// the passive half carries the shared ApiKey (plus a listen port) and nothing else. +/// Why two predicates (review R1 m3). The two answers cost differently when wrong. Registering a +/// node that has no working sync path only costs capture triggers. Gating the drain of a node that +/// shares its queue with nobody parks its alarm events behind a Primary that does not hold them — +/// undelivered until the node is promoted, and drop-oldest at Capacity meanwhile. So +/// registration stays permissive and the gate is strict. Every node the gate treats as sharing also +/// captures, so the two never disagree in the harmful direction. /// /// -/// Why the drain gate takes the same predicate (it previously used PeerAddress OR -/// SyncListenPort). The gate's question is "does a peer hold the same rows?". The passive -/// sync endpoint is fail-closed on the key, so a node with a listen port but no ApiKey rejects -/// every inbound stream and shares its queue with nobody - gating it would park its alarm events -/// behind a Primary that cannot deliver them. Conversely an ApiKey-only passive node IS -/// registered and does receive its peer's rows, so it must be gated or both halves drain. The port -/// is not evidence of a peer; the key is. +/// Registration: LocalDb:Replication:PeerAddress OR LocalDb:Replication:ApiKey. +/// The OR is required: only the dialling half of a pair sets PeerAddress, while the passive +/// half carries the shared ApiKey (plus a listen port) and nothing else. Keying on +/// PeerAddress alone would strip capture from every passive node. +/// +/// +/// Drain gate: ApiKey AND (PeerAddress OR LocalDb:SyncListenPort > 0). A +/// peer holds this node's rows only over a working sync path: a dial (PeerAddress) or a +/// listener (SyncListenPort), each authenticated by the key. Both auth directions fail closed, +/// so a path without the key shares nothing, and a key with neither path (e.g. a fleet-wide +/// secret/env ApiKey on a node with no peer) shares nothing either. Every docker-dev shape +/// gives the same answer under both predicates: site-a-1 (PeerAddress + ApiKey + port) and site-a-2 +/// (ApiKey + port) are gated; central and site-b (neither) are not. /// /// public static class LocalDbReplicationPolicy @@ -30,14 +40,31 @@ public static class LocalDbReplicationPolicy /// Configuration section holding the LocalDb replication options. public const string ReplicationSectionPath = "LocalDb:Replication"; - /// Whether this node's LocalDb replicates to a pair peer. + /// The dedicated passive-sync listener port key (same key as LocalDbRegistration.SyncListenPortKey). + public const string SyncListenPortKey = "LocalDb:SyncListenPort"; + + /// Whether this node registers its replicated tables for change capture (table registration). /// The application configuration. /// true when PeerAddress or ApiKey is set under the replication section. - public static bool IsConfigured(IConfiguration configuration) + public static bool CapturesChanges(IConfiguration configuration) { ArgumentNullException.ThrowIfNull(configuration); var section = configuration.GetSection(ReplicationSectionPath); return !string.IsNullOrWhiteSpace(section["PeerAddress"]) || !string.IsNullOrWhiteSpace(section["ApiKey"]); } + + /// Whether a pair peer actually holds this node's replicated rows (the alarm drain gate). + /// The application configuration. + /// true when ApiKey is set AND either PeerAddress is set or + /// LocalDb:SyncListenPort is a positive port. + public static bool SharesRowsWithPeer(IConfiguration configuration) + { + ArgumentNullException.ThrowIfNull(configuration); + var section = configuration.GetSection(ReplicationSectionPath); + if (string.IsNullOrWhiteSpace(section["ApiKey"])) return false; + var dials = !string.IsNullOrWhiteSpace(section["PeerAddress"]); + var listens = int.TryParse(configuration[SyncListenPortKey], out var port) && port > 0; + return dials || listens; + } } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs index a530ce62..c50deff9 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs @@ -177,11 +177,13 @@ public static class ServiceCollectionExtensions // possibly deliver its events. Scoping the gate to "is my queue actually shared?" is what // keeps the two scopes from disagreeing. // BOTH halves of a pair share the queue, but only one of them dials: the initiator sets - // Replication:PeerAddress, its partner carries the shared ApiKey and waits. The predicate is - // LocalDbReplicationPolicy - the same one LocalDbSetup uses to decide which nodes capture - // changes at all. It deliberately does NOT count SyncListenPort: the passive endpoint is - // fail-closed on the key, so a port with no ApiKey shares nothing and must keep draining. - var replicated = LocalDbReplicationPolicy.IsConfigured(configuration); + // Replication:PeerAddress, its partner carries the shared ApiKey plus SyncListenPort and waits. + // The predicate is LocalDbReplicationPolicy.SharesRowsWithPeer: ApiKey AND (PeerAddress OR + // SyncListenPort > 0) - an authenticated sync path in either direction. Both auth directions + // fail closed, so a path without the key, or a key with no path (a fleet-wide secret on an + // unpaired node), shares nothing and must keep draining. Deliberately stricter than the + // registration predicate (CapturesChanges) - see the policy's remarks. + var replicated = LocalDbReplicationPolicy.SharesRowsWithPeer(configuration); if (!replicated) { diff --git a/tests/Core/ZB.MOM.WW.OtOpcUa.Cluster.Tests/ClusterRoleInfoMemberEventTests.cs b/tests/Core/ZB.MOM.WW.OtOpcUa.Cluster.Tests/ClusterRoleInfoMemberEventTests.cs index 024b182c..4253ea10 100644 --- a/tests/Core/ZB.MOM.WW.OtOpcUa.Cluster.Tests/ClusterRoleInfoMemberEventTests.cs +++ b/tests/Core/ZB.MOM.WW.OtOpcUa.Cluster.Tests/ClusterRoleInfoMemberEventTests.cs @@ -97,4 +97,30 @@ public sealed class ClusterRoleInfoMemberEventTests return Task.CompletedTask; }); } + + /// + /// Review R1 m6: the initial seed lists only Up / WeaklyUp members. A member already departing at seed + /// time never fires a later Left/Exited/Downed event, so seeding it would list it until MemberRemoved. + /// The Up and WeaklyUp members are the positive control (the seed does list members). + /// + [Fact] + public async Task Initial_seed_lists_only_up_and_weakly_up_members() + { + await WithInfoAsync(info => + { + var members = ImmutableSortedSet.Create( + NewMember(21001, MemberStatus.Up), + NewMember(21002, MemberStatus.WeaklyUp), + NewMember(21003, MemberStatus.Leaving), + NewMember(21004, MemberStatus.Exiting), + NewMember(21005, MemberStatus.Down), + NewMember(21006, MemberStatus.Joining)); + + info.SeedFrom(new ClusterEvent.CurrentClusterState().Copy(members: members)); + + info.MembersWithRole("driver").Select(n => n.Value).OrderBy(v => v).ShouldBe( + ["127.0.0.1:21001", "127.0.0.1:21002"]); + return Task.CompletedTask; + }); + } } diff --git a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms.Tests/Part9StateMachineTests.cs b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms.Tests/Part9StateMachineTests.cs index 2ebe9fb4..dd8ea3b7 100644 --- a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms.Tests/Part9StateMachineTests.cs +++ b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms.Tests/Part9StateMachineTests.cs @@ -79,6 +79,28 @@ public sealed class Part9StateMachineTests r.Emission.ShouldBe(EmissionKind.None); } + /// + /// Gitea #597 review m1 — a COMMENTED re-ack records its comment (as AddComment would) so the comment the + /// OPC UA event advertised is durable; the original acknowledge's ack fields are untouched. + /// + [Fact] + public void Commented_reack_records_the_comment_without_reacknowledging() + { + var active = Part9StateMachine.ApplyPredicate(Fresh(), true, T0.AddSeconds(1)).State; + var acked = Part9StateMachine.ApplyAcknowledge(active, "alice", "first look", T0.AddSeconds(2)).State; + + var r = Part9StateMachine.ApplyAcknowledge(acked, "bob", "still on it", T0.AddSeconds(3)); + + r.Emission.ShouldBe(EmissionKind.CommentAdded); + r.State.Comments.Count.ShouldBe(2); + r.State.Comments[1].User.ShouldBe("bob"); + r.State.Comments[1].Kind.ShouldBe("AddComment"); + r.State.Comments[1].Text.ShouldBe("still on it"); + r.State.LastAckUser.ShouldBe("alice"); + r.State.LastAckComment.ShouldBe("first look"); + r.State.LastAckUtc.ShouldBe(T0.AddSeconds(2)); + } + /// Verifies that acknowledging without a user throws ArgumentException. [Fact] public void Acknowledge_without_user_throws() diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/ConditionMethodEventTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/ConditionMethodEventTests.cs index a20c6c2c..d0ebede2 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/ConditionMethodEventTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/ConditionMethodEventTests.cs @@ -263,14 +263,18 @@ public sealed class ConditionMethodEventTests /// /// A scripted TimedShelve expires: the engine's expiry comes back as an Unshelved re-projection, which - /// reports exactly one unshelve event after the one shelve event. + /// reports exactly one unshelve event after the one shelve event — and the SDK's own expiry timer routes + /// nothing (review R1 M2). /// /// /// The expiry is the engine's (computed from the command's UnshelveAtUtc). Since Gitea #597 the node - /// is shelved for the real duration, so the SDK's own timer fires at the same instant — but it only routes - /// an Unshelve back to the engine (ignored by this stand-in, a no-op on an unshelved engine alarm) and - /// changes no node state, so it adds no event. The stand-in engine here schedules its expiry from the real - /// UnshelveAtUtc the method handler computed. + /// is shelved for the real duration, so the SDK's own timer fires at that instant. It used to route an + /// operator-style Unshelve by system, which — because the real engine only checks expiry every + /// 5 s — pre-empted the engine's AutoUnshelve on every expiry. This stand-in engine therefore + /// (a) expires LATE, like the real one, so the SDK timer demonstrably fires first; and (b) ACTS on a routed + /// Unshelve as the real engine would, recording who unshelved it. The wrapped + /// OnTimedUnshelve is the positive control: the SDK timer did fire, and still nothing was routed and + /// the unshelve on record is the engine's own. /// [Fact] public async Task A_scripted_timed_shelve_expires_with_exactly_one_unshelve_event() @@ -283,21 +287,47 @@ public sealed class ConditionMethodEventTests rig.MaterialiseScripted(); var current = PreState("TimedShelve", native: false) with { Message = "base" }; Task? expiry = null; + var routed = new System.Collections.Concurrent.ConcurrentQueue(); + string? unshelvedBy = null; + var engineLag = TimeSpan.FromMilliseconds(1500); // the real engine's 5 s shelving-check granularity rig.Nm.AlarmCommandRouter = cmd => { + routed.Enqueue(cmd.Operation); + if (cmd.Operation == "Unshelve") + { + // What the real engine does with a routed Unshelve: an operator-style unshelve. + unshelvedBy ??= $"Unshelve:{cmd.User}"; + current = current with { Shelving = AlarmShelvingKind.Unshelved, UnshelveAtUtc = null }; + rig.ProjectLater(native: false, current); + return; + } if (cmd.Operation != "TimedShelve") return; - current = current with { Shelving = AlarmShelvingKind.Timed }; + // The real engine's projection carries its expiry, so the re-projection keeps the SDK timer armed + // for the real remaining time (without it, it would re-arm for MaxTimeShelved and never fire here). + current = current with { Shelving = AlarmShelvingKind.Timed, UnshelveAtUtc = cmd.UnshelveAtUtc }; rig.ProjectLater(native: false, current); var unshelveAt = cmd.UnshelveAtUtc!.Value; expiry = Task.Run(async () => { - var wait = unshelveAt - DateTime.UtcNow; + var wait = unshelveAt + engineLag - DateTime.UtcNow; if (wait > TimeSpan.Zero) await Task.Delay(wait); - current = current with { Shelving = AlarmShelvingKind.Unshelved }; + if (current.Shelving != AlarmShelvingKind.Timed) return; // ApplyShelvingCheck: Timed only + unshelvedBy ??= "AutoUnshelve:system"; + current = current with { Shelving = AlarmShelvingKind.Unshelved, UnshelveAtUtc = null }; rig.Project(native: false, current); }); }; + // Count the SDK's own expiry-timer firings without changing what the production handler does. + var condition = rig.Nm.TryGetAlarmCondition(ScriptedAlarmId)!; + var sdkTimerFired = 0; + var production = condition.OnTimedUnshelve!; + condition.OnTimedUnshelve = (context, c) => + { + Interlocked.Increment(ref sdkTimerFired); + return production(context, c); + }; + using var session = await rig.ConnectAsync(anonymous: false, ct); var events = new RecordedEvents(); var subscription = await RecordedEvents.SubscribeAsync(session, events, ct); @@ -315,6 +345,10 @@ public sealed class ConditionMethodEventTests rig.Project(false, current = current with { Message = "sentinel" }); await RecordedEvents.WaitUntilAsync(() => events.Has(item, "sentinel")); + Volatile.Read(ref sdkTimerFired).ShouldBeGreaterThanOrEqualTo(1, "the SDK's timed-unshelve timer fired"); + routed.ShouldNotContain("Unshelve", "the SDK timer must not route an operator-style Unshelve"); + unshelvedBy.ShouldBe("AutoUnshelve:system", "the engine's own expiry owns the unshelve"); + var shelving = events.ConditionEvents(item, "base").Select(e => e[RecordedEvents.ShelvingField].Value).ToList(); shelving.ShouldBe( [ObjectIds.ShelvedStateMachineType_Unshelved, ObjectIds.ShelvedStateMachineType_TimedShelved, diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/AlarmCommandRouterTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/AlarmCommandRouterTests.cs index 210c15c2..9972bd68 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/AlarmCommandRouterTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/AlarmCommandRouterTests.cs @@ -268,19 +268,15 @@ public sealed class AlarmCommandRouterTests : IDisposable await host.DisposeAsync(); } - /// OnTimedUnshelve fires with the SDK's system context (no session, no user identity) — - /// the real SDK path when a TimedShelve duration expires. The gate must NOT veto: the result must be - /// Good, the router must be invoked exactly once with Operation == "Unshelve", and no UnshelveAtUtc is - /// carried. - /// - /// Regression (OpcUaServer-007): the routed command's must be a - /// NON-EMPTY system user ("system"). The engine's Part9StateMachine.ApplyUnshelve rejects - /// a null/whitespace user with ArgumentException("User required."); that exception is swallowed - /// downstream, so an empty user made the timed auto-unshelve SILENTLY no-op — a TimedShelve would never - /// auto-expire. Asserting a non-empty user here pins the value the state machine requires. - /// + /// OnTimedUnshelve fires with the SDK's system context (no session, no user identity) when the + /// SDK's own TimedShelve timer expires. It must return Good and route NOTHING (review R1 M2, Gitea #597): + /// the scripted engine owns timed-shelve expiry (its shelving check records an "AutoUnshelve" by + /// "system"), and since #597 arms the SDK timer for the real remaining time, an "Unshelve" routed from + /// here pre-empted that check on every expiry, from both pair nodes, as an operator-style unshelve. + /// Positive control: on the same condition and router, a client Unshelve (OnShelve with + /// shelving:false) DOES route — so the empty capture is not a disconnected router. [Fact] - public async Task OnTimedUnshelve_with_system_context_returns_good_and_routes_unshelve() + public async Task OnTimedUnshelve_with_system_context_returns_good_and_routes_nothing() { var (host, server) = await BootAsync(); var nm = server.NodeManager!; @@ -299,14 +295,13 @@ public sealed class AlarmCommandRouterTests : IDisposable var result = condition.OnTimedUnshelve!(ctx, condition); result.ShouldBe(ServiceResult.Good); - captured.Count.ShouldBe(1); - var cmd = captured[0]; - cmd.AlarmId.ShouldBe("alm-tu"); // == ScriptedAlarmId / condition NodeId identifier - cmd.Operation.ShouldBe("Unshelve"); - // System-initiated (no client principal), but the user MUST be non-empty so ApplyUnshelve accepts it. - cmd.User.ShouldNotBeNullOrWhiteSpace(); - cmd.User.ShouldBe("system"); - cmd.UnshelveAtUtc.ShouldBeNull(); + captured.ShouldBeEmpty(); + + // Positive control: the router is live for a genuine operator Unshelve on the same condition. + var operatorCtx = SessionContext(server, "grace", OpcUaDataPlaneRoles.AlarmAck); + condition.OnShelve!(operatorCtx, condition, shelving: false, oneShot: false, shelvingTime: 0.0) + .ShouldBe(ServiceResult.Good); + captured.Select(c => c.Operation).ShouldBe(["Unshelve"]); await host.DisposeAsync(); } diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorBootFromCacheTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorBootFromCacheTests.cs index ecfb2311..11da96f3 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorBootFromCacheTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorBootFromCacheTests.cs @@ -102,8 +102,14 @@ public sealed class DriverHostActorBootFromCacheTests : RuntimeActorTestBase // Barrier: diagnostics are served only after PreStart's boot work ran, so the absence below is // asserted after the boot path had its chance to Tell (positive control is the valid-bytes test). - AskDiagnostics(actor); + var snapshot = AskDiagnostics(actor); publishProbe.ExpectNoMsg(TimeSpan.FromMilliseconds(500)); + + // Review R1 m4: nothing was applied, so the node is NOT running from cache and has NOT adopted the + // cached revision (Stale with no configuration, retrying the ConfigDb) — the valid-bytes test above + // is the control that both flip when the bytes are usable. + snapshot.RunningFromCache.ShouldBeFalse(); + snapshot.CurrentRevision.ShouldBeNull(); } [Fact] diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorFetchAndCacheBootstrapTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorFetchAndCacheBootstrapTests.cs index f6d8487d..f7e113b6 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorFetchAndCacheBootstrapTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorFetchAndCacheBootstrapTests.cs @@ -111,6 +111,39 @@ public sealed class DriverHostActorFetchAndCacheBootstrapTests : RuntimeActorTes fetcher.Calls.ShouldBe(1); } + /// + /// #601 / review R1 m4 — a cache row whose bytes are malformed applied nothing, so its revision must not + /// become current: the boot lands Steady-with-no-revision (as for an empty cache), and a re-dispatch of + /// the SAME revision really applies (fetches) instead of being acked by the "already at this rev" + /// short-circuit. is the + /// control: valid bytes do adopt the revision. + /// + [Fact] + public void WithMalformedCachedBytes_TheRevisionIsNotAdopted_AndARedispatchOfItActuallyApplies() + { + var cached = new CachedDeploymentArtifact( + DeploymentId.NewId().ToString(), CachedRev.Value, + System.Text.Encoding.UTF8.GetBytes("{\"DriverInstances\": ["), DateTimeOffset.UtcNow.AddHours(-1)); + var fetcher = new RecordingFetcher(Artifact()); + var coordinator = CreateTestProbe(); + + var actor = Sys.ActorOf(DriverHostActor.Props( + new ThrowingDbFactory(), TestNode, coordinator.Ref, + localRoles: new HashSet { "driver" }, + deploymentArtifactCache: new StubArtifactCache(cached), + fetchAndCacheMode: true, + artifactFetcher: fetcher)); + + var snapshot = AskDiagnostics(actor); + snapshot.CurrentRevision.ShouldBeNull(); + snapshot.RunningFromCache.ShouldBeFalse(); + + actor.Tell(new DispatchDeployment(DeploymentId.Parse(cached.DeploymentId), CachedRev, CorrelationId.NewId())); + coordinator.ExpectMsg(TimeSpan.FromSeconds(5)).Outcome.ShouldBe(ApplyAckOutcome.Applied); + fetcher.Calls.ShouldBe(1); + AskDiagnostics(actor).CurrentRevision.ShouldBe(CachedRev); + } + private NodeDiagnosticsSnapshot AskDiagnostics(IActorRef actor) { var probe = CreateTestProbe(); diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorPrimaryRoleSourceTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorPrimaryRoleSourceTests.cs new file mode 100644 index 00000000..b846dea9 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorPrimaryRoleSourceTests.cs @@ -0,0 +1,109 @@ +using Akka.Actor; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Redundancy; +using ZB.MOM.WW.OtOpcUa.Commons.Types; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Runtime.Drivers; +using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers; + +/// +/// Review R1 m2 (Gitea #597) — the Primary data-plane verdict published to +/// (read by the native-ack gate and the Sparkplug primary-host claim) mirrors the host's own live gate from the +/// moment the host starts, not only from the first redundancy snapshot. +/// +/// +/// RedundancyRoleView is seeded with the single-node verdict (allow). Before this fix nothing published to +/// it until a snapshot arrived, so on a pair both nodes accepted a native ack through the whole boot window while +/// DriverHostActor — which evaluates the driver member count live — denied the upstream route on the +/// non-Primary. Assertions are on what was PUBLISHED: the seed already reads true, so a current-value +/// assertion of the single-node case would pass with no publish at all. +/// +public sealed class DriverHostActorPrimaryRoleSourceTests : RuntimeActorTestBase +{ + private static readonly NodeId TestNode = NodeId.Parse("driver-primary-source-test"); + private static readonly TimeSpan Timeout = TimeSpan.FromSeconds(5); + + /// On a pair, an unknown role publishes DENY at start — before any snapshot. The later Primary snapshot + /// is the positive control: the source is wired, and the deny was the verdict, not a missing publish. + [Fact] + public void A_paired_host_publishes_deny_before_the_first_snapshot() + { + var source = new RecordingPrimaryRoleSource(); + var host = SpawnHost(source, () => 2); + + AwaitAssert(() => source.Published.ShouldBe([false]), duration: Timeout); + + host.Tell(Snapshot(RedundancyRole.Primary)); + AwaitAssert(() => source.Published.ShouldBe([false, true]), duration: Timeout); + } + + /// A single-driver node keeps the allow posture — it is published, not merely seeded. + [Fact] + public void A_single_node_host_publishes_allow_at_start() + { + var source = new RecordingPrimaryRoleSource(); + SpawnHost(source, () => 1); + + AwaitAssert(() => source.Published.ShouldBe([true]), duration: Timeout); + } + + /// A driver peer coming Up while the role is still unknown flips the published verdict to deny, with no + /// snapshot involved — the membership re-publish path. + [Fact] + public void A_peer_joining_before_any_snapshot_flips_the_published_verdict_to_deny() + { + var source = new RecordingPrimaryRoleSource(); + var members = 1; + var host = SpawnHost(source, () => Volatile.Read(ref members)); + AwaitAssert(() => source.Published.ShouldBe([true]), duration: Timeout); + + Volatile.Write(ref members, 2); + host.Tell(DriverHostActor.RepublishPrimaryVerdict.Instance); + + AwaitAssert(() => source.Published.ShouldBe([true, false]), duration: Timeout); + } + + private IActorRef SpawnHost(IDriverPrimaryRoleSource source, Func driverMemberCount) => + Sys.ActorOf(DriverHostActor.Props( + NewInMemoryDbFactory(), + TestNode, + ackRouter: null, + driverMemberCountProvider: driverMemberCount, + primaryRoleSource: source)); + + private static RedundancyStateChanged Snapshot(RedundancyRole role) => + new( + [ + new NodeRedundancyState(TestNode, role, + IsClusterLeader: role == RedundancyRole.Primary, + IsDriverPrimary: role == RedundancyRole.Primary, + AsOfUtc: DateTime.UtcNow), + ], + CorrelationId.NewId()); + + /// Remembers every published verdict, so tests assert on the actor's action, not on the seed. + private sealed class RecordingPrimaryRoleSource : IDriverPrimaryRoleSource + { + private readonly List _published = []; + private readonly Lock _gate = new(); + + /// Every value published so far, in order. + public IReadOnlyList Published + { + get { lock (_gate) return _published.ToArray(); } + } + + public bool ShouldActAsPrimary + { + get { lock (_gate) return _published.Count == 0 || _published[^1]; } + } + + public void PublishPrimaryRole(bool shouldActAsPrimary) + { + lock (_gate) _published.Add(shouldActAsPrimary); + } + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorUnreadableArtifactTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorUnreadableArtifactTests.cs index f1cd00cd..ee918813 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorUnreadableArtifactTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorUnreadableArtifactTests.cs @@ -227,6 +227,77 @@ public sealed class DriverHostActorUnreadableArtifactTests : RuntimeActorTestBas replayed.ShouldContain("other-node-driver"); } + /// + /// #609 (review R1 M1) — the Resilience half of the eviction must SURVIVE the next resilience-publisher + /// tick. DriverResilienceStatusPublisherService re-emits tracker.Snapshot() into the hub + /// every interval, so evicting only the hub cache was undone within one tick: the host must also + /// release the instance's tracker state. The tick below is the publisher's own body + /// (BuildMessages → hub.Emit) minus the DPS publish, run against the REAL factory + + /// tracker the Host wires. The neighbour instance is the positive control: it is still emitted by + /// the same tick, so an absent drv-1 is not an empty tick. + /// + [Fact] + public void Removing_a_driver_instance_releases_its_resilience_state_so_the_next_publisher_tick_omits_it() + { + var db = NewInMemoryDbFactory(); + var factory = new SubscribingDriverFactory("Modbus"); + var hub = new ZB.MOM.WW.OtOpcUa.Runtime.Telemetry.TelemetryLocalHub(); + var tracker = new ZB.MOM.WW.OtOpcUa.Core.Resilience.DriverResilienceStatusTracker(); + var invokerFactory = new ZB.MOM.WW.OtOpcUa.Core.Resilience.DriverCapabilityInvokerFactory( + new ZB.MOM.WW.OtOpcUa.Core.Resilience.DriverResiliencePipelineBuilder(statusTracker: tracker), + tracker, + _ => ZB.MOM.WW.OtOpcUa.Core.Abstractions.DriverTier.A); + var coordinator = CreateTestProbe(); + var actor = Sys.ActorOf(DriverHostActor.Props( + db, TestNode, coordinator.Ref, + driverFactory: factory, + localRoles: new HashSet { "driver" }, + invokerFactory: invokerFactory, + telemetryHub: hub)); + + actor.Tell(new DispatchDeployment(SeedV3Deployment(db, RevA), RevA, CorrelationId.NewId())); + coordinator.ExpectMsg(Timeout).Outcome.ShouldBe(ApplyAckOutcome.Applied); + + // Resilience counters for the running driver and for an instance this host does not own. + tracker.RecordBreakerOpen("drv-1", "dev1", DateTime.UtcNow); + tracker.RecordFailure("neighbour-drv", "dev9", DateTime.UtcNow); + PublisherTick(tracker, hub); + ReplayedResilienceIds(hub).ShouldContain("drv-1"); // control: drv-1 is primed before the removal + + actor.Tell(new DispatchDeployment(SeedDriverlessDeployment(db, RevB), RevB, CorrelationId.NewId())); + coordinator.ExpectMsg(Timeout).Outcome.ShouldBe(ApplyAckOutcome.Applied); + AwaitAssert( + () => AskDiagnostics(actor).Drivers.Select(d => d.Name).ShouldNotContain("drv-1"), + duration: Timeout); + + // The publisher's next tick, after the removal. + PublisherTick(tracker, hub); + + var replayed = ReplayedResilienceIds(hub); + replayed.ShouldNotContain("drv-1"); + replayed.ShouldContain("neighbour-drv"); + + static void PublisherTick( + ZB.MOM.WW.OtOpcUa.Core.Resilience.DriverResilienceStatusTracker t, + ZB.MOM.WW.OtOpcUa.Runtime.Telemetry.TelemetryLocalHub h) + { + var now = DateTime.UtcNow; + foreach (var e in t.Snapshot()) + h.Emit(new ZB.MOM.WW.OtOpcUa.Runtime.Telemetry.TelemetryItem.Resilience(new( + e.DriverInstanceId, e.HostName, e.Snapshot.IsBreakerOpen, e.Snapshot.ConsecutiveFailures, + e.Snapshot.CurrentInFlight, e.Snapshot.LastBreakerOpenUtc, e.Snapshot.LastSampledUtc, now))); + } + + static List ReplayedResilienceIds(ZB.MOM.WW.OtOpcUa.Runtime.Telemetry.TelemetryLocalHub h) + { + using var sub = h.Subscribe(64); + var ids = new List(); + while (sub.Reader.TryRead(out var i)) + if (i is ZB.MOM.WW.OtOpcUa.Runtime.Telemetry.TelemetryItem.Resilience r) ids.Add(r.E.DriverInstanceId); + return ids; + } + } + /// Spawns the host with the subscribing factory, dispatches and /// waits for the Applied ACK so the child + the initial SubscribeBulk pass are in place. private (IActorRef Actor, Akka.TestKit.TestProbe Coordinator) SpawnHostAndApply( diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/AlarmHistorianRegistrationTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/AlarmHistorianRegistrationTests.cs index 91f655da..6b806a17 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/AlarmHistorianRegistrationTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/AlarmHistorianRegistrationTests.cs @@ -109,7 +109,8 @@ public sealed class AlarmHistorianRegistrationTests /// /// /// Standing down is only safe because a peer holds the same rows and will send them - /// instead. Without LocalDb:Replication:PeerAddress there is no such peer: the rows + /// instead. Without an authenticated sync path (ApiKey plus a dial or a listener — + /// LocalDbReplicationPolicy.SharesRowsWithPeer) there is no such peer: the rows /// are here and nowhere else, so any deferral means nobody ever delivers them. /// /// @@ -124,14 +125,18 @@ public sealed class AlarmHistorianRegistrationTests // Neither key: an unpaired node. Its rows exist nowhere else. [InlineData(null, null, null, true)] // The dialing half of a pair — gated, because its partner holds the same rows. - [InlineData("http://peer:9001", null, null, false)] + [InlineData("http://peer:9001", null, "k", false)] // The LISTENING half. Same shared queue, but it never dials, so a PeerAddress-only test would - // wrongly class it as unpaired and leave it permanently ungated. It is identified by the shared - // ApiKey (the same predicate LocalDbSetup registers tables on), not by its listen port. + // wrongly class it as unpaired and leave it permanently ungated. Key + listen port = a peer. [InlineData(null, "9001", "k", false)] // A listen port with NO key: the fail-closed passive endpoint rejects every inbound stream, so the // queue is shared with nobody and the node must keep draining (#603: this used to be gated). [InlineData(null, "9001", null, true)] + // A dial with NO key: the peer's fail-closed endpoint rejects it — shares nothing, keeps draining. + [InlineData("http://peer:9001", null, null, true)] + // A key with NEITHER path (review R1 m3: e.g. a fleet-wide secret on an unpaired node) — no listener, + // no dial, so no peer holds these rows. Registered for capture, but must keep draining. + [InlineData(null, null, "k", true)] public async Task Only_an_unreplicated_node_ignores_the_role_view( string? peerAddress, string? syncListenPort, string? apiKey, bool expectDrain) { diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/LocalDbReplicationPolicyTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/LocalDbReplicationPolicyTests.cs index 0ff37296..654f4bf2 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/LocalDbReplicationPolicyTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/LocalDbReplicationPolicyTests.cs @@ -4,29 +4,64 @@ using Xunit; namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Historian; -/// #603: the single "is this node's LocalDb replicated" predicate shared by table registration and the alarm drain gate. +/// +/// #603 + review R1 m3: the two "is this node's LocalDb replicated" predicates — permissive for table +/// registration (), strict for the alarm drain gate +/// (). +/// public sealed class LocalDbReplicationPolicyTests { + private const string Peer = "LocalDb:Replication:PeerAddress"; + private const string Key = "LocalDb:Replication:ApiKey"; + private const string Port = "LocalDb:SyncListenPort"; + private static IConfiguration Config(params (string Key, string? Value)[] pairs) => new ConfigurationBuilder() .AddInMemoryCollection(pairs.Select(p => new KeyValuePair(p.Key, p.Value))) .Build(); - /// PeerAddress OR ApiKey counts; the listen port alone and blank values do not. + /// Registration: PeerAddress OR ApiKey counts; the listen port alone and blank values do not. [Theory] - [InlineData("LocalDb:Replication:PeerAddress", "http://peer:9001", true)] - [InlineData("LocalDb:Replication:ApiKey", "k", true)] - [InlineData("LocalDb:SyncListenPort", "9001", false)] - [InlineData("LocalDb:Replication:ApiKey", " ", false)] - public void IsConfigured_follows_the_PeerAddress_or_ApiKey_rule(string key, string value, bool expected) + [InlineData(Peer, "http://peer:9001", true)] + [InlineData(Key, "k", true)] + [InlineData(Port, "9001", false)] + [InlineData(Key, " ", false)] + public void CapturesChanges_follows_the_PeerAddress_or_ApiKey_rule(string key, string value, bool expected) { - LocalDbReplicationPolicy.IsConfigured(Config((key, value))).ShouldBe(expected); + LocalDbReplicationPolicy.CapturesChanges(Config((key, value))).ShouldBe(expected); } - /// No replication keys at all is an unpaired node. - [Fact] - public void IsConfigured_is_false_for_an_empty_configuration() + /// Drain gate: ApiKey AND (PeerAddress OR SyncListenPort > 0). + [Theory] + [InlineData("http://peer:9001", "k", null, true)] // dialling half of a pair + [InlineData(null, "k", "9001", true)] // listening half of a pair + [InlineData("http://peer:9001", "k", "9001", true)] // both + [InlineData(null, "k", null, false)] // key, no path: shares nothing (review R1 m3) + [InlineData(null, "k", "0", false)] // port 0 = no listener + [InlineData(null, "k", "nope", false)] // unparseable port = no listener + [InlineData("http://peer:9001", null, "9001", false)] // paths, no key: both auth directions fail closed + [InlineData(null, " ", "9001", false)] // blank key + public void SharesRowsWithPeer_needs_a_key_and_a_sync_path(string? peer, string? key, string? port, bool expected) { - LocalDbReplicationPolicy.IsConfigured(Config()).ShouldBeFalse(); + LocalDbReplicationPolicy.SharesRowsWithPeer(Config((Peer, peer), (Key, key), (Port, port))).ShouldBe(expected); + } + + /// Every node the gate treats as sharing rows also captures — never a gated node with no capture. + [Theory] + [InlineData("http://peer:9001", "k", null)] + [InlineData(null, "k", "9001")] + public void A_node_sharing_rows_always_captures(string? peer, string? key, string? port) + { + var config = Config((Peer, peer), (Key, key), (Port, port)); + LocalDbReplicationPolicy.SharesRowsWithPeer(config).ShouldBeTrue(); + LocalDbReplicationPolicy.CapturesChanges(config).ShouldBeTrue(); + } + + /// No replication keys at all is an unpaired node under both predicates. + [Fact] + public void Both_predicates_are_false_for_an_empty_configuration() + { + LocalDbReplicationPolicy.CapturesChanges(Config()).ShouldBeFalse(); + LocalDbReplicationPolicy.SharesRowsWithPeer(Config()).ShouldBeFalse(); } }