diff --git a/archreview/plans/R2-04-failure-visibility-plan.md b/archreview/plans/R2-04-failure-visibility-plan.md index 1278a33d..3b31e0be 100644 --- a/archreview/plans/R2-04-failure-visibility-plan.md +++ b/archreview/plans/R2-04-failure-visibility-plan.md @@ -537,3 +537,11 @@ T14 anytime; T15 last (after T10–T13 merge-ready) ``` Overall effort: **~1.5–2 dev-days code+unit, plus the rig session**. The two HIGH-RISK slices (T1 record change, T10 gate change) each land behind their full regression suites before anything else stacks on them. + +## Execution deviations (R2-04) + +- **T1–T3 test increments built incrementally.** The three tasks edit the same two files (`AddressSpaceApplier.cs` + a new test file). The `OpcUaServer.Tests` project sets `TreatWarningsAsErrors=true`, so an unused test helper would break the build. To keep every per-task commit green, the new `AddressSpaceApplierFailureSurfaceTests.cs` grew one task's tests at a time (T1 RebuildFailed, T2 FailedNodes removal-pass, T3 Materialise* int returns) rather than landing all tests up-front. +- **T11 write-path assertions made retry-tolerant.** `Unknown_role_single_driver_services_write` and the Primary arm of `Known_role_wins_over_member_count` transiently hit a pre-existing driver-connect race ("driver not connected" reply while the `RecordingDriver` child's async init lags the `ApplyAck`) — orthogonal to the gate. The gate correctly ALLOWED (reason was never "not primary"). Wrapped the write in an `AwaitAssert` retry (as a real client retries), asserting the invariant under test: the gate never denies. Verified deterministic across two full-class runs. +- **T13 (in-process 2-node delivery) implemented as a gate-reason poll, not a tag write.** Rather than deploy a known tag and drive its NodeId (brittle: the harness's fixed deploy config + the DriverHostActor path), the test resolves both nodes' `DriverHostActor` via the `ActorRegistry` (`DriverHostActorKey`) and Asks `RouteNodeWrite` for an intentionally-unmapped node, polling until the DELIVERED snapshot promotes exactly one node to Primary (the Secondary rejects with `"not primary"`; the Primary passes the gate and rejects only with a MAPPING reason). This is a faithful delivery-path assertion with a built-in negative control (a broken redundancy-topic subscribe leaves both stuck `"not primary (role unknown)"` and times out). Build-verified; marked `deferred-live` (Host.IntegrationTests memory leak + needs SQL/2-node rig — controller runs it in the serialized heavy pass). +- **T8 / T13 / T15 marked `deferred-live`.** T8 Galaxy live smoke skips cleanly offline (its live run is the operator's, matching the plan's acceptance) — kept `completed` (authored + skip-verified). T13 (in-process 2-node integration) and T15 (2-node docker-dev rig gate) require the live rig / heavy suites the plan reserves for the serialized pass. +- **`ServiceCollectionExtensions.DriverRole` reused for the driver-member-count read** in both `DriverHostActor.DriverMemberCount()` and `ScriptedAlarmHostActor.DriverMemberCount()` (the canonical `"driver"` role constant), rather than a new literal. diff --git a/archreview/plans/R2-04-failure-visibility-plan.md.tasks.json b/archreview/plans/R2-04-failure-visibility-plan.md.tasks.json index 1d922cbe..07d3a34a 100644 --- a/archreview/plans/R2-04-failure-visibility-plan.md.tasks.json +++ b/archreview/plans/R2-04-failure-visibility-plan.md.tasks.json @@ -2,20 +2,20 @@ "planPath": "archreview/plans/R2-04-failure-visibility-plan.md", "lastUpdated": "2026-07-12", "tasks": [ - { "id": "T1", "subject": "Extend AddressSpaceApplyOutcome with RebuildFailed/FailedNodes + SafeRebuild returns bool (01/S-1)", "status": "pending", "blockedBy": [] }, - { "id": "T2", "subject": "Safe* helpers return bool; Apply tallies removal-pass failures into FailedNodes", "status": "pending", "blockedBy": ["T1"] }, - { "id": "T3", "subject": "Materialise* passes return swallowed-failure counts (void -> int, five methods)", "status": "pending", "blockedBy": ["T2"] }, - { "id": "T4", "subject": "OpcUaApplyFailed counter + OpcUaPublishActor Error/meter surfacing of degraded applies", "status": "pending", "blockedBy": ["T3"] }, - { "id": "T5", "subject": "Galaxy internal IMxWriteOps seam under GatewayGalaxyDataWriter (sealed SDK session untestable) (06/S-1)", "status": "pending", "blockedBy": [] }, - { "id": "T6", "subject": "Galaxy fail-closed write on AdviseSupervisory failure -> BadCommunicationError, no raw write", "status": "pending", "blockedBy": ["T5"] }, - { "id": "T7", "subject": "Galaxy counters galaxy.writes.advise_failed + galaxy.writes.unconfirmed (empty-statuses Good metered)", "status": "pending", "blockedBy": ["T6"] }, - { "id": "T8", "subject": "Galaxy skip-gated live smoke: normal advised write still returns Good post-change", "status": "pending", "blockedBy": ["T6"] }, - { "id": "T9", "subject": "PrimaryGatePolicy pure class + truth-table tests + otopcua.redundancy.primary_gate_denied counter (03/S4)", "status": "pending", "blockedBy": [] }, - { "id": "T10", "subject": "Wire DriverHostActor gates (:977/:1039/:1095) through PrimaryGatePolicy with member-count provider + S5 log-once", "status": "pending", "blockedBy": ["T9"] }, - { "id": "T11", "subject": "DriverHostActorPrimaryGateTests: deny-on-unknown multi-node, allow single-node, ack/emit variants, meter tags", "status": "pending", "blockedBy": ["T10"] }, - { "id": "T12", "subject": "ScriptedAlarmHostActor alerts-emit gate adopts PrimaryGatePolicy (consistency extension)", "status": "pending", "blockedBy": ["T9"] }, - { "id": "T13", "subject": "In-process 2-node delivery-path test (TwoNodeClusterHarness): DPS-delivered snapshot drives the gate", "status": "pending", "blockedBy": ["T11"] }, - { "id": "T14", "subject": "docs/Redundancy.md: gate policy table, boot-window client experience, new meters, Galaxy fail-closed semantics", "status": "pending", "blockedBy": [] }, - { "id": "T15", "subject": "LIVE GATE: 2-node docker-dev rig verify (ServiceLevel 240/100, boot-window write rejected+reverted, single /alerts row)", "status": "pending", "blockedBy": ["T10", "T11", "T12", "T13"] } + { "id": "T1", "subject": "Extend AddressSpaceApplyOutcome with RebuildFailed/FailedNodes + SafeRebuild returns bool (01/S-1)", "status": "completed", "blockedBy": [] }, + { "id": "T2", "subject": "Safe* helpers return bool; Apply tallies removal-pass failures into FailedNodes", "status": "completed", "blockedBy": ["T1"] }, + { "id": "T3", "subject": "Materialise* passes return swallowed-failure counts (void -> int, five methods)", "status": "completed", "blockedBy": ["T2"] }, + { "id": "T4", "subject": "OpcUaApplyFailed counter + OpcUaPublishActor Error/meter surfacing of degraded applies", "status": "completed", "blockedBy": ["T3"] }, + { "id": "T5", "subject": "Galaxy internal IMxWriteOps seam under GatewayGalaxyDataWriter (sealed SDK session untestable) (06/S-1)", "status": "completed", "blockedBy": [] }, + { "id": "T6", "subject": "Galaxy fail-closed write on AdviseSupervisory failure -> BadCommunicationError, no raw write", "status": "completed", "blockedBy": ["T5"] }, + { "id": "T7", "subject": "Galaxy counters galaxy.writes.advise_failed + galaxy.writes.unconfirmed (empty-statuses Good metered)", "status": "completed", "blockedBy": ["T6"] }, + { "id": "T8", "subject": "Galaxy skip-gated live smoke: normal advised write still returns Good post-change", "status": "completed", "blockedBy": ["T6"] }, + { "id": "T9", "subject": "PrimaryGatePolicy pure class + truth-table tests + otopcua.redundancy.primary_gate_denied counter (03/S4)", "status": "completed", "blockedBy": [] }, + { "id": "T10", "subject": "Wire DriverHostActor gates (:977/:1039/:1095) through PrimaryGatePolicy with member-count provider + S5 log-once", "status": "completed", "blockedBy": ["T9"] }, + { "id": "T11", "subject": "DriverHostActorPrimaryGateTests: deny-on-unknown multi-node, allow single-node, ack/emit variants, meter tags", "status": "completed", "blockedBy": ["T10"] }, + { "id": "T12", "subject": "ScriptedAlarmHostActor alerts-emit gate adopts PrimaryGatePolicy (consistency extension)", "status": "completed", "blockedBy": ["T9"] }, + { "id": "T13", "subject": "In-process 2-node delivery-path test (TwoNodeClusterHarness): DPS-delivered snapshot drives the gate", "status": "deferred-live", "note": "rig/live; Host.IntegrationTests (~16GB/run leak + needs SQL/2-node rig) — implemented + build-verified, controller runs it in the serialized heavy pass", "blockedBy": ["T11"] }, + { "id": "T14", "subject": "docs/Redundancy.md: gate policy table, boot-window client experience, new meters, Galaxy fail-closed semantics", "status": "completed", "blockedBy": [] }, + { "id": "T15", "subject": "LIVE GATE: 2-node docker-dev rig verify (ServiceLevel 240/100, boot-window write rejected+reverted, single /alerts row)", "status": "deferred-live", "note": "rig/live; 2-node docker-dev rig session (rebuild both centrals) — merge-blocking gate the controller runs in the serialized heavy pass", "blockedBy": ["T10", "T11", "T12", "T13"] } ] } diff --git a/docs/Redundancy.md b/docs/Redundancy.md index bd58859a..f16e7248 100644 --- a/docs/Redundancy.md +++ b/docs/Redundancy.md @@ -179,11 +179,24 @@ survives removing the typed option, and only breaks under an explicit `NoDowning There is no operator-driven role swap during a partition. Failover is what the cluster does automatically. -## Primary-gated alarm emission and historization +## Primary data-plane gate (writes, acks, alerts emit) -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. However, to avoid duplicate rows on `/alerts` and duplicate historian writes, only the Primary node publishes externally: +Three data-plane surfaces are **Primary-gated** so a warm standby never touches a shared field device or duplicates a fleet-wide publish: inbound operator **writes** (`DriverHostActor.HandleRouteNodeWrite`), native-alarm **acks** (`HandleRouteNativeAlarmAck`), and the native/scripted **alerts emit** (`ForwardNativeAlarm` / `ScriptedAlarmHostActor.OnEngineEmission`). All three route through one policy — `PrimaryGatePolicy.ShouldServiceAsPrimary(localRole, driverMemberCount)` (`OtOpcUa.Runtime.Drivers`): -- **`alerts` topic emission** — `ScriptedAlarmHostActor` subscribes to the `redundancy-state` DPS topic and caches the local node's `RedundancyRole`. Each alarm transition is published to the cluster `alerts` topic **only when the node's role is `Primary`**. The default behaviour before any `redundancy-state` message arrives is to emit, so single-node deployments and the boot window never drop transitions. 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. +| Local role (last redundancy snapshot) | Cluster driver members | Decision | +|---|---|---| +| `Primary` | any | **service** | +| `Secondary` / `Detached` | any | **deny** | +| unknown (no snapshot yet, or snapshot omitted this node) | ≤ 1 (single-node / boot on a lone driver) | **service** (boot-window / single-node posture) | +| unknown | ≥ 2 (a real driver peer exists) | **deny** (default-deny — a Primary peer exists; don't act until a snapshot proves this node is it) | + +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. + +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. Net effect: each alarm transition appears **once** on `/alerts` and would historize once, not once per node. @@ -201,8 +214,14 @@ The OtOpcUa Client CLI at `src/Client/ZB.MOM.WW.OtOpcUa.Client.CLI` supports `-F | Metric | Type | Notes | |---|---|---| | `otopcua.redundancy.service_level_change` | Counter (`{change}`) | OPC UA `Server.ServiceLevel` transitions emitted by the redundancy state. Tagged with `level` = the new byte. | +| `otopcua.redundancy.primary_gate_denied` | Counter (`{denial}`) | Operations denied by the Primary data-plane gate (archreview 03/S4). Tagged `site` = `write` \| `ack` \| `alarm-emit` and `reason` = `secondary` \| `detached` \| `role-unknown`. A non-zero `role-unknown` rate on a healthy multi-node cluster flags a stuck boot window (redundancy snapshot not delivering). | +| `otopcua.opcua.apply.failed` | Counter (`{apply}`) | Address-space apply/materialise passes that swallowed at least one sink failure (archreview 01/S-1). Tagged `kind` = `rebuild` (the rebuild threw) \| `nodes` (per-node materialise failures). A non-zero rate means the running server holds a stale or partial address space despite a reported-successful deploy — `OpcUaPublishActor` also logs the degraded apply at **Error**. | +| `galaxy.writes.advise_failed` | Counter (`{write}`) | Galaxy writes short-circuited to `Bad` because `AdviseSupervisory` failed — the value could not have committed, so the write **fails closed** (archreview 06/S-1) and the Bad status fires the #5 node revert instead of leaving a phantom-Good node. On the Galaxy driver meter `ZB.MOM.WW.OtOpcUa.Driver.Galaxy`. | +| `galaxy.writes.unconfirmed` | Counter (`{write}`) | Galaxy writes reported `Good` off an **empty** gateway statuses array (command accepted; the COM-side commit is unconfirmed pending the mxaccessgw `WriteComplete` correlation follow-up). The honest signal for the optimistic-Good rate until that cross-repo fix lands. On the Galaxy driver meter. | -The meter is defined on `OtOpcUaTelemetry` (`src/Core/ZB.MOM.WW.OtOpcUa.Commons/Observability/OtOpcUaTelemetry.cs`); it surfaces through whatever OpenTelemetry exporter the host configures. +The redundancy/apply meters are defined on `OtOpcUaTelemetry` (`src/Core/ZB.MOM.WW.OtOpcUa.Commons/Observability/OtOpcUaTelemetry.cs`); the Galaxy write meters hang off the driver's own meter (`EventPump.MeterName`). All surface through whatever OpenTelemetry exporter the host configures. + +**Galaxy fail-closed write semantics (archreview 06/S-1):** a non-secured Galaxy write commits only when its item is `AdviseSupervisory`-advised first (the writer runs with no login). If the advise fails, the write is **not** issued — `GatewayGalaxyDataWriter` returns `BadCommunicationError` (a knowingly-lost write is never reported as Good). An empty statuses array on the write reply is still treated as `Good` but metered as `galaxy.writes.unconfirmed`. See [Historian.md](Historian.md) / the Galaxy driver notes for the write path. ## Depth reference diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Observability/OtOpcUaTelemetry.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Observability/OtOpcUaTelemetry.cs index 668521fc..7b9d7933 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Observability/OtOpcUaTelemetry.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Observability/OtOpcUaTelemetry.cs @@ -58,10 +58,24 @@ public static class OtOpcUaTelemetry Meter.CreateCounter("otopcua.opcua.sink.write", unit: "{write}", description: "Writes that landed in IOpcUaAddressSpaceSink (kind=value|alarm|rebuild)."); + /// Address-space apply/materialise passes that swallowed at least one sink failure — + /// a failed rebuild (kind=rebuild) or per-node materialise failures (kind=nodes). A non-zero rate + /// means the running server holds a stale or partially-materialised address space despite a + /// reported-successful deploy (archreview 01/S-1). + public static readonly Counter OpcUaApplyFailed = + Meter.CreateCounter("otopcua.opcua.apply.failed", unit: "{apply}", + description: "Apply/materialise passes with swallowed sink failures (kind=rebuild|nodes)."); + public static readonly Counter ServiceLevelChange = Meter.CreateCounter("otopcua.redundancy.service_level_change", unit: "{change}", description: "OPC UA Server.ServiceLevel transitions emitted by the redundancy state."); + /// Inbound operations denied by the Primary data-plane gate (archreview 03/S4) — + /// site=write|ack|alarm-emit, reason=secondary|detached|role-unknown. + public static readonly Counter PrimaryGateDenied = + Meter.CreateCounter("otopcua.redundancy.primary_gate_denied", unit: "{denial}", + description: "Operations denied by the Primary gate (site=write|ack|alarm-emit, reason=secondary|detached|role-unknown)."); + // ---------------- Convenience helpers ---------------- /// diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/GatewayGalaxyDataWriter.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/GatewayGalaxyDataWriter.cs index 1389b51c..282803ec 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/GatewayGalaxyDataWriter.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/GatewayGalaxyDataWriter.cs @@ -1,4 +1,5 @@ using System.Collections.Concurrent; +using System.Diagnostics.Metrics; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging.Abstractions; using ZB.MOM.WW.MxGateway.Client; @@ -37,6 +38,18 @@ public sealed class GatewayGalaxyDataWriter : IGalaxyDataWriter private readonly ConcurrentDictionary _supervisedHandles = new(); private int _addItemCallCount; + // Write failure-visibility meters (archreview 06/S-1). Reuse the pump's meter name so the single host + // listener catches both. advise_failed = writes short-circuited to Bad because AdviseSupervisory failed + // (the value could not have committed); unconfirmed = writes reported Good off an EMPTY gateway statuses + // array (command accepted, COM commit unconfirmed pending the gateway WriteComplete correlation follow-up). + private static readonly Meter WriterMeter = new(EventPump.MeterName); + private static readonly Counter WriteAdviseFailed = + WriterMeter.CreateCounter("galaxy.writes.advise_failed", unit: "{write}", + description: "Writes short-circuited to Bad because AdviseSupervisory failed (value could not have committed)."); + private static readonly Counter WriteUnconfirmed = + WriterMeter.CreateCounter("galaxy.writes.unconfirmed", unit: "{write}", + description: "Writes reported Good off an EMPTY gateway statuses array (command accepted; commit unconfirmed)."); + /// Initializes a new Galaxy data writer. /// The MXAccess gateway session. /// The user ID for write operations. @@ -129,23 +142,39 @@ public sealed class GatewayGalaxyDataWriter : IGalaxyDataWriter "GalaxyMxSession is not connected. Call ConnectAsync before issuing writes."); var serverHandle = _session.ServerHandle; + // Bind the per-batch MX write operations behind the internal IMxWriteOps seam (see below) so the + // whole write pipeline is unit-testable without the sealed SDK session (archreview 06/S-1). + var ops = new SessionMxWriteOps(session, serverHandle); + var results = new WriteResult[writes.Count]; for (var i = 0; i < writes.Count; i++) { - results[i] = await WriteOneAsync(session, serverHandle, writes[i], + results[i] = await WriteOneAsync(ops, writes[i], securityResolver(writes[i].FullReference), cancellationToken) .ConfigureAwait(false); } return results; } + /// Test seam: drive the real single-write pipeline against a fake + /// so the fail-closed / unconfirmed behaviour (archreview 06/S-1) is covered offline. Behaviourally + /// identical to the per-entry path runs. + /// The (fake) MX write operations. + /// The write request. + /// The tag's security classification. + /// The cancellation token. + /// The translated write result. + internal Task WriteOneForTestAsync( + IMxWriteOps ops, WriteRequest request, SecurityClassification classification, CancellationToken ct) + => WriteOneAsync(ops, request, classification, ct); + private async Task WriteOneAsync( - MxGatewaySession session, int serverHandle, WriteRequest request, + IMxWriteOps ops, WriteRequest request, SecurityClassification classification, CancellationToken ct) { try { - var itemHandle = await EnsureItemHandleAsync(session, serverHandle, request.FullReference, ct) + var itemHandle = await EnsureItemHandleAsync(ops, request.FullReference, ct) .ConfigureAwait(false); var mxValue = MxValueEncoder.Encode(request.Value); @@ -154,7 +183,7 @@ public sealed class GatewayGalaxyDataWriter : IGalaxyDataWriter { // SecuredWrite/VerifiedWrite tags carry their own ArchestrA user identity // (current/verifier user), so they don't use the supervisory path. - reply = await InvokeWriteSecuredAsync(session, serverHandle, itemHandle, mxValue, ct) + reply = await InvokeWriteSecuredAsync(ops, itemHandle, mxValue, ct) .ConfigureAwait(false); } else @@ -162,9 +191,18 @@ public sealed class GatewayGalaxyDataWriter : IGalaxyDataWriter // A raw Write runs with NO user login (WriteUserId is typically 0), so MXAccess // only COMMITS the value when the item is advised in SUPERVISORY mode. Without it // the gateway's Write call doesn't throw (reply looks OK) but the value never - // reaches the galaxy. AdviseSupervisory once per handle, then Write. - await EnsureSupervisoryAdvisedAsync(session, serverHandle, itemHandle, ct).ConfigureAwait(false); - reply = await session.WriteRawAsync(serverHandle, itemHandle, mxValue, _writeUserId, ct) + // reaches the galaxy — a knowingly-lost write. FAIL-CLOSED (archreview 06/S-1): when + // AdviseSupervisory fails we do NOT issue the raw Write; we return Bad so the + // #5 write-outcome self-correction reverts the phantom-Good node instead of leaving it. + // The real fix is a gateway-side WriteComplete correlation (mxaccessgw backlog) so the + // unary reply's Statuses carry the actual COM commit outcome; until then advise-status + + // the galaxy.writes.* meters are the honest signal. + if (!await EnsureSupervisoryAdvisedAsync(ops, itemHandle, ct).ConfigureAwait(false)) + { + WriteAdviseFailed.Add(1); + return new WriteResult(StatusCodeMap.BadCommunicationError); + } + reply = await ops.WriteRawAsync(itemHandle, mxValue, _writeUserId, ct) .ConfigureAwait(false); } @@ -189,10 +227,10 @@ public sealed class GatewayGalaxyDataWriter : IGalaxyDataWriter classification is SecurityClassification.SecuredWrite or SecurityClassification.VerifiedWrite; private async Task EnsureItemHandleAsync( - MxGatewaySession session, int serverHandle, string fullRef, CancellationToken ct) + IMxWriteOps ops, string fullRef, CancellationToken ct) { if (TryResolveCachedOrBorrowed(fullRef) is int resolved) return resolved; - var handle = await session.AddItemAsync(serverHandle, fullRef, ct).ConfigureAwait(false); + var handle = await ops.AddItemAsync(fullRef, ct).ConfigureAwait(false); Interlocked.Increment(ref _addItemCallCount); _itemHandles[fullRef] = handle; return handle; @@ -210,36 +248,45 @@ public sealed class GatewayGalaxyDataWriter : IGalaxyDataWriter /// session reconnect via , so a fresh session always /// re-advises — the supervisory state never outlives the handle it was taken against. /// - private async Task EnsureSupervisoryAdvisedAsync( - MxGatewaySession session, int serverHandle, int itemHandle, CancellationToken ct) + /// Ensure the item is supervisory-advised so a no-login raw Write can commit. Returns + /// true when the item is advised (already-advised this session, or the advise round-trip + /// succeeded), false when the advise failed — the caller then FAILS THE WRITE CLOSED + /// (archreview 06/S-1) rather than issuing a raw Write that cannot commit. + /// true when advised; false when the advise failed (caller must not write). + private async Task EnsureSupervisoryAdvisedAsync( + IMxWriteOps ops, int itemHandle, CancellationToken ct) { - if (!_supervisedHandles.TryAdd(itemHandle, 0)) return; + // Already advised this session (a prior successful advise on this handle) — nothing to do. + if (!_supervisedHandles.TryAdd(itemHandle, 0)) return true; var request = new MxCommandRequest { - SessionId = session.SessionId, + SessionId = ops.SessionId, ClientCorrelationId = Guid.NewGuid().ToString("N"), Command = new MxCommand { Kind = MxCommandKind.AdviseSupervisory, AdviseSupervisory = new AdviseSupervisoryCommand { - ServerHandle = serverHandle, + ServerHandle = ops.ServerHandle, ItemHandle = itemHandle, }, }, }; - var reply = await session.InvokeAsync(request, ct).ConfigureAwait(false); + var reply = await ops.InvokeAsync(request, ct).ConfigureAwait(false); if (reply.ProtocolStatus is { } proto && proto.Code != ProtocolStatusCode.Ok) { - // Supervisory advise failed — forget it so the next write retries, and let the - // write proceed (it surfaces its own status via TranslateReply). + // Supervisory advise failed — forget it so the next write retries the advise, and report + // failure so the caller fails the write closed (the raw Write would not commit). _supervisedHandles.TryRemove(itemHandle, out _); _logger.LogWarning( "GalaxyDriver supervisory advise failed for item {ItemHandle}: {Code} {Message}", itemHandle, proto.Code, proto.Message); + return false; } + + return true; } /// @@ -250,14 +297,14 @@ public sealed class GatewayGalaxyDataWriter : IGalaxyDataWriter /// interprets the underlying MXAccess command kind. /// private static Task InvokeWriteSecuredAsync( - MxGatewaySession session, int serverHandle, int itemHandle, MxValue value, CancellationToken ct) + IMxWriteOps ops, int itemHandle, MxValue value, CancellationToken ct) { var command = new MxCommand { Kind = MxCommandKind.WriteSecured, WriteSecured = new WriteSecuredCommand { - ServerHandle = serverHandle, + ServerHandle = ops.ServerHandle, ItemHandle = itemHandle, Value = value, CurrentUserId = 0, @@ -266,17 +313,22 @@ public sealed class GatewayGalaxyDataWriter : IGalaxyDataWriter }; var request = new MxCommandRequest { - SessionId = session.SessionId, + SessionId = ops.SessionId, ClientCorrelationId = Guid.NewGuid().ToString("N"), Command = command, }; - return session.InvokeAsync(request, ct); + return ops.InvokeAsync(request, ct); } /// /// Translate a gateway into an OPC UA /// . Honours the protocol-level Status field first /// (transport / dispatch failures), then the first MXAccess status row. + /// An EMPTY statuses array is treated as Good but is PROVISIONAL: the reply proves + /// worker-side command acceptance, not the COM-side commit (the gateway does not yet correlate a + /// WriteComplete row onto the unary reply — mxaccessgw backlog). Each such reply increments + /// galaxy.writes.unconfirmed so the unconfirmed-write rate is operator-visible today + /// (archreview 06/S-1). /// private WriteResult TranslateReply(MxCommandReply reply, string fullRef) { @@ -298,6 +350,81 @@ public sealed class GatewayGalaxyDataWriter : IGalaxyDataWriter return new WriteResult(StatusCodeMap.FromMxStatus(status, _logger)); } + // Empty statuses ⇒ provisional Good. Meter it so the unconfirmed-write rate is visible until the + // gateway-side WriteComplete correlation lands (archreview 06/S-1 cross-repo follow-up). + WriteUnconfirmed.Add(1); return new WriteResult(StatusCodeMap.Good); } } + +/// +/// The per-write MXAccess gateway operations the writer needs, extracted behind an internal seam so the +/// fail-closed write pipeline (archreview 06/S-1) is unit-testable — the SDK MxGatewaySession +/// types are sealed with internal ctors and cannot be faked. Production binds +/// over the real (MxGatewaySession, serverHandle); tests inject a +/// fake that records calls and returns canned replies. +/// +internal interface IMxWriteOps +{ + /// The gateway session id stamped onto every . + string SessionId { get; } + + /// The MXAccess server handle carried in each item/advise/write command. + int ServerHandle { get; } + + /// Add an MXAccess item and return its item handle. + /// The dotted tag full reference. + /// The cancellation token. + /// The MXAccess item handle. + Task AddItemAsync(string fullRef, CancellationToken ct); + + /// Invoke a raw (AdviseSupervisory / WriteSecured) and return the reply. + /// The command request. + /// The cancellation token. + /// The gateway reply. + Task InvokeAsync(MxCommandRequest request, CancellationToken ct); + + /// Issue a raw (non-secured) Write for the given item handle. + /// The MXAccess item handle to write. + /// The encoded MX value. + /// The write user id (typically 0 — the supervisory path commits it). + /// The cancellation token. + /// The gateway reply. + Task WriteRawAsync(int itemHandle, MxValue value, int userId, CancellationToken ct); +} + +/// Production — a thin adapter over a live +/// and its server handle. Behaviour is bit-identical to the previous +/// inline (session, serverHandle) calls. +internal sealed class SessionMxWriteOps : IMxWriteOps +{ + private readonly MxGatewaySession _session; + private readonly int _serverHandle; + + /// Initializes the adapter over a connected session + server handle. + /// The connected gateway session. + /// The MXAccess server handle for this session. + public SessionMxWriteOps(MxGatewaySession session, int serverHandle) + { + _session = session; + _serverHandle = serverHandle; + } + + /// + public string SessionId => _session.SessionId; + + /// + public int ServerHandle => _serverHandle; + + /// + public Task AddItemAsync(string fullRef, CancellationToken ct) + => _session.AddItemAsync(_serverHandle, fullRef, ct); + + /// + public Task InvokeAsync(MxCommandRequest request, CancellationToken ct) + => _session.InvokeAsync(request, ct); + + /// + public Task WriteRawAsync(int itemHandle, MxValue value, int userId, CancellationToken ct) + => _session.WriteRawAsync(_serverHandle, itemHandle, value, userId, ct); +} diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/AddressSpaceApplier.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/AddressSpaceApplier.cs index 3fba3d99..eff4d63d 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/AddressSpaceApplier.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/AddressSpaceApplier.cs @@ -82,14 +82,17 @@ public sealed class AddressSpaceApplier var ts = DateTime.UtcNow; var removedCount = 0; + // Swallowed removal-pass condition-write failures are tallied here (not lost to a per-node Warning) + // so a degraded removal is operator-visible via AddressSpaceApplyOutcome.FailedNodes (archreview 01/S-1). + var failedNodes = 0; foreach (var eq in plan.RemovedEquipment) { - SafeWriteAlarmCondition(eq.EquipmentId, RemovedConditionState, ts); + if (!SafeWriteAlarmCondition(eq.EquipmentId, RemovedConditionState, ts)) failedNodes++; removedCount++; } foreach (var alarm in plan.RemovedAlarms) { - SafeWriteAlarmCondition(alarm.ScriptedAlarmId, RemovedConditionState, ts); + if (!SafeWriteAlarmCondition(alarm.ScriptedAlarmId, RemovedConditionState, ts)) failedNodes++; removedCount++; } // Removed equipment tags / VirtualTags are plain variable nodes (no Part 9 condition to write @@ -146,10 +149,11 @@ public sealed class AddressSpaceApplier // covered for free and need no separate surgical pass. var renamedFolders = plan.RenamedFolders; var rebuilt = false; + var rebuildFailed = false; if (structuralRebuild) { - SafeRebuild(); + rebuildFailed = !SafeRebuild(); rebuilt = true; } else if (surgicalTagDeltas.Count > 0 || renamedFolders.Count > 0) @@ -188,12 +192,12 @@ public sealed class AddressSpaceApplier } if (!ok) { allApplied = false; break; } } - if (!allApplied) { SafeRebuild(); rebuilt = true; } + if (!allApplied) { rebuildFailed = !SafeRebuild(); rebuilt = true; } } else { // Sink lacks the surgical capability ⇒ rebuild (safe default). - SafeRebuild(); + rebuildFailed = !SafeRebuild(); rebuilt = true; } } @@ -212,7 +216,7 @@ public sealed class AddressSpaceApplier // currently-historized set. Same non-blocking + throw-safe discipline as the provisioning hook. FeedHistorizedRefs(plan); - return new AddressSpaceApplyOutcome(removedCount, addedCount, changedCount, rebuilt); + return new AddressSpaceApplyOutcome(removedCount, addedCount, changedCount, rebuilt, rebuildFailed, failedNodes); } /// @@ -370,15 +374,22 @@ public sealed class AddressSpaceApplier string.IsNullOrWhiteSpace(tag.HistorianTagname) ? tag.FullName : tag.HistorianTagname) : null; - private void SafeRebuild() + /// Rebuild the sink's address space, swallowing (and Error-logging) any fault. + /// Returns true on success, false when the sink threw — the caller threads the + /// result into so a broken rebuild is + /// visible instead of reported as optimistic success (archreview 01/S-1). + /// true when the rebuild completed; false when the sink threw. + private bool SafeRebuild() { try { _sink.RebuildAddressSpace(); + return true; } catch (Exception ex) { _logger.LogError(ex, "AddressSpaceApplier: sink.RebuildAddressSpace threw"); + return false; } } @@ -390,28 +401,32 @@ public sealed class AddressSpaceApplier /// present, so re-applies are cheap. /// /// The composition result containing the hierarchy to materialise. - public void MaterialiseHierarchy(AddressSpaceComposition composition) + /// The count of swallowed sink failures in this pass (0 on a clean apply) — the publish actor + /// sums it into the degraded-apply surface (archreview 01/S-1). + public int MaterialiseHierarchy(AddressSpaceComposition composition) { ArgumentNullException.ThrowIfNull(composition); + var failed = 0; foreach (var area in composition.UnsAreas) { - SafeEnsureFolder(area.UnsAreaId, parentNodeId: null, displayName: area.DisplayName); + if (!SafeEnsureFolder(area.UnsAreaId, parentNodeId: null, displayName: area.DisplayName)) failed++; } foreach (var line in composition.UnsLines) { - SafeEnsureFolder(line.UnsLineId, parentNodeId: line.UnsAreaId, displayName: line.DisplayName); + if (!SafeEnsureFolder(line.UnsLineId, parentNodeId: line.UnsAreaId, displayName: line.DisplayName)) failed++; } foreach (var equipment in composition.EquipmentNodes) { // Equipment with no UnsLineId (legacy / dev rows) hang under the root. var parent = string.IsNullOrWhiteSpace(equipment.UnsLineId) ? null : equipment.UnsLineId; - SafeEnsureFolder(equipment.EquipmentId, parentNodeId: parent, displayName: equipment.DisplayName); + if (!SafeEnsureFolder(equipment.EquipmentId, parentNodeId: parent, displayName: equipment.DisplayName)) failed++; } _logger.LogInformation( - "AddressSpaceApplier: hierarchy materialised (areas={Areas}, lines={Lines}, equipment={Equipment})", - composition.UnsAreas.Count, composition.UnsLines.Count, composition.EquipmentNodes.Count); + "AddressSpaceApplier: hierarchy materialised (areas={Areas}, lines={Lines}, equipment={Equipment}, failed={Failed})", + composition.UnsAreas.Count, composition.UnsLines.Count, composition.EquipmentNodes.Count, failed); + return failed; } /// @@ -429,11 +444,14 @@ public sealed class AddressSpaceApplier /// takes a plain string dataType (not a DriverAttributeInfo). /// /// The composition result containing the equipment tags to materialise. - public void MaterialiseEquipmentTags(AddressSpaceComposition composition) + /// The count of swallowed sink failures in this pass (0 on a clean apply) — the publish actor + /// sums it into the degraded-apply surface (archreview 01/S-1). + public int MaterialiseEquipmentTags(AddressSpaceComposition composition) { ArgumentNullException.ThrowIfNull(composition); - if (composition.EquipmentTags.Count == 0) return; + if (composition.EquipmentTags.Count == 0) return 0; + var failed = 0; // Sub-folders first — a tag's FolderPath becomes one folder UNDER its equipment folder // (deduped per distinct equipment+path). Tags with no FolderPath hang directly under the // equipment folder, which MaterialiseHierarchy already created (never re-create @@ -444,7 +462,7 @@ public sealed class AddressSpaceApplier if (string.IsNullOrWhiteSpace(tag.FolderPath)) continue; var folderNodeId = EquipmentNodeIds.SubFolder(tag.EquipmentId, tag.FolderPath); if (!foldersCreated.Add(folderNodeId)) continue; - SafeEnsureFolder(folderNodeId, parentNodeId: tag.EquipmentId, displayName: tag.FolderPath); + if (!SafeEnsureFolder(folderNodeId, parentNodeId: tag.EquipmentId, displayName: tag.FolderPath)) failed++; } // Variables: NodeId is FOLDER-SCOPED ("/"), NOT the raw FullName — a driver @@ -463,7 +481,7 @@ public sealed class AddressSpaceApplier { // Native alarm tag → a real Part 9 condition node (reuses the scripted-alarm path), // NOT a value variable. Parent is the sub-folder when set, else the equipment folder. - SafeMaterialiseAlarmCondition(nodeId, parent, tag.Name, tag.Alarm.AlarmType, tag.Alarm.Severity, isNative: true); + if (!SafeMaterialiseAlarmCondition(nodeId, parent, tag.Name, tag.Alarm.AlarmType, tag.Alarm.Severity, isNative: true)) failed++; } else { @@ -477,14 +495,16 @@ public sealed class AddressSpaceApplier // even if authored ReadWrite, so a client write cannot reach the driver write path which // does not handle arrays (e.g. S7 BoxValueForWrite would crash). var writable = tag.Writable && !tag.IsArray; - SafeEnsureVariable(nodeId, parent, tag.Name, tag.DataType, writable, historianTagname, tag.IsArray, tag.ArrayLength); + if (!SafeEnsureVariable(nodeId, parent, tag.Name, tag.DataType, writable, historianTagname, tag.IsArray, tag.ArrayLength)) failed++; } } _logger.LogInformation( - "AddressSpaceApplier: equipment tags materialised (tags={Tags}, equipment={Equipment})", + "AddressSpaceApplier: equipment tags materialised (tags={Tags}, equipment={Equipment}, failed={Failed})", composition.EquipmentTags.Count, - composition.EquipmentTags.Select(t => t.EquipmentId).Distinct(StringComparer.Ordinal).Count()); + composition.EquipmentTags.Select(t => t.EquipmentId).Distinct(StringComparer.Ordinal).Count(), + failed); + return failed; } /// @@ -498,7 +518,9 @@ public sealed class AddressSpaceApplier /// NodeAdded model-change is announced under this node. /// The discovered folders to ensure (parent-first by depth). /// The discovered variables to ensure (read-only value nodes). - public void MaterialiseDiscoveredNodes( + /// The count of swallowed sink failures in this pass (0 on a clean apply) — the publish actor + /// surfaces a non-zero count as a degraded discovered-node injection (archreview 01/S-1). + public int MaterialiseDiscoveredNodes( string equipmentRootNodeId, IReadOnlyList folders, IReadOnlyList variables) @@ -506,25 +528,27 @@ public sealed class AddressSpaceApplier ArgumentException.ThrowIfNullOrEmpty(equipmentRootNodeId); ArgumentNullException.ThrowIfNull(folders); ArgumentNullException.ThrowIfNull(variables); - if (folders.Count == 0 && variables.Count == 0) return; + if (folders.Count == 0 && variables.Count == 0) return 0; + var failed = 0; // Parent-first: a child folder's parent must exist before it. Ordering by '/' count == depth. foreach (var f in folders.OrderBy(f => f.NodeId.Count(c => c == '/'))) - SafeEnsureFolder(f.NodeId, f.ParentNodeId, f.DisplayName); + if (!SafeEnsureFolder(f.NodeId, f.ParentNodeId, f.DisplayName)) failed++; foreach (var v in variables) { // Mirror MaterialiseEquipmentTags: arrays forced read-only (the driver write path can't handle arrays). var writable = v.Writable && !v.IsArray; - SafeEnsureVariable(v.NodeId, v.ParentNodeId, v.DisplayName, v.DataType, writable, - historianTagname: null, isArray: v.IsArray, arrayLength: v.ArrayLength); + if (!SafeEnsureVariable(v.NodeId, v.ParentNodeId, v.DisplayName, v.DataType, writable, + historianTagname: null, isArray: v.IsArray, arrayLength: v.ArrayLength)) failed++; } _sink.RaiseNodesAddedModelChange(equipmentRootNodeId); _logger.LogInformation( - "AddressSpaceApplier: discovered nodes materialised under {Equipment} (folders={Folders}, vars={Vars})", - equipmentRootNodeId, folders.Count, variables.Count); + "AddressSpaceApplier: discovered nodes materialised under {Equipment} (folders={Folders}, vars={Vars}, failed={Failed})", + equipmentRootNodeId, folders.Count, variables.Count, failed); + return failed; } /// @@ -540,11 +564,14 @@ public sealed class AddressSpaceApplier /// Idempotent (per-variable idempotency relies on the sink's own EnsureVariable). /// /// The composition result containing the equipment VirtualTags to materialise. - public void MaterialiseEquipmentVirtualTags(AddressSpaceComposition composition) + /// The count of swallowed sink failures in this pass (0 on a clean apply) — the publish actor + /// sums it into the degraded-apply surface (archreview 01/S-1). + public int MaterialiseEquipmentVirtualTags(AddressSpaceComposition composition) { ArgumentNullException.ThrowIfNull(composition); - if (composition.EquipmentVirtualTags.Count == 0) return; + if (composition.EquipmentVirtualTags.Count == 0) return 0; + var failed = 0; // Sub-folders first — a VirtualTag's FolderPath becomes one folder UNDER its equipment folder // (deduped per distinct equipment+path). VirtualTags with no FolderPath hang directly under the // equipment folder, which MaterialiseHierarchy already created (never re-create it here). @@ -554,7 +581,7 @@ public sealed class AddressSpaceApplier if (string.IsNullOrWhiteSpace(v.FolderPath)) continue; var folderNodeId = EquipmentNodeIds.SubFolder(v.EquipmentId, v.FolderPath); if (!foldersCreated.Add(folderNodeId)) continue; - SafeEnsureFolder(folderNodeId, parentNodeId: v.EquipmentId, displayName: v.FolderPath); + if (!SafeEnsureFolder(folderNodeId, parentNodeId: v.EquipmentId, displayName: v.FolderPath)) failed++; } // Variables: NodeId is FOLDER-SCOPED ("/"), mirroring the equipment-tag pass. @@ -570,13 +597,15 @@ public sealed class AddressSpaceApplier : EquipmentNodeIds.SubFolder(v.EquipmentId, v.FolderPath); var nodeId = EquipmentNodeIds.Variable(v.EquipmentId, v.FolderPath, v.Name); // VirtualTags are computed outputs — read-only nodes (no inbound write). - SafeEnsureVariable(nodeId, parent, v.Name, v.DataType, writable: false); + if (!SafeEnsureVariable(nodeId, parent, v.Name, v.DataType, writable: false)) failed++; } _logger.LogInformation( - "AddressSpaceApplier: equipment virtualtags materialised (vtags={Vtags}, equipment={Equipment})", + "AddressSpaceApplier: equipment virtualtags materialised (vtags={Vtags}, equipment={Equipment}, failed={Failed})", composition.EquipmentVirtualTags.Count, - composition.EquipmentVirtualTags.Select(v => v.EquipmentId).Distinct(StringComparer.Ordinal).Count()); + composition.EquipmentVirtualTags.Select(v => v.EquipmentId).Distinct(StringComparer.Ordinal).Count(), + failed); + return failed; } /// @@ -589,36 +618,49 @@ public sealed class AddressSpaceApplier /// MaterialiseAlarmCondition re-creates cleanly on re-apply). /// /// The composition result containing the scripted alarms to materialise. - public void MaterialiseScriptedAlarms(AddressSpaceComposition composition) + /// The count of swallowed sink failures in this pass (0 on a clean apply) — the publish actor + /// sums it into the degraded-apply surface (archreview 01/S-1). + public int MaterialiseScriptedAlarms(AddressSpaceComposition composition) { ArgumentNullException.ThrowIfNull(composition); - if (composition.EquipmentScriptedAlarms.Count == 0) return; + if (composition.EquipmentScriptedAlarms.Count == 0) return 0; var materialised = 0; + var failed = 0; foreach (var alarm in composition.EquipmentScriptedAlarms) { if (!alarm.Enabled) continue; - SafeMaterialiseAlarmCondition(alarm.ScriptedAlarmId, alarm.EquipmentId, alarm.Name, alarm.AlarmType, alarm.Severity, isNative: false); + if (!SafeMaterialiseAlarmCondition(alarm.ScriptedAlarmId, alarm.EquipmentId, alarm.Name, alarm.AlarmType, alarm.Severity, isNative: false)) failed++; materialised++; } _logger.LogInformation( - "AddressSpaceApplier: scripted alarms materialised (alarms={Alarms}, equipment={Equipment})", + "AddressSpaceApplier: scripted alarms materialised (alarms={Alarms}, equipment={Equipment}, failed={Failed})", materialised, composition.EquipmentScriptedAlarms.Where(a => a.Enabled) - .Select(a => a.EquipmentId).Distinct(StringComparer.Ordinal).Count()); + .Select(a => a.EquipmentId).Distinct(StringComparer.Ordinal).Count(), + failed); + return failed; } - private void SafeEnsureFolder(string nodeId, string? parentNodeId, string displayName) + /// Ensure a folder node, swallowing (and Warning-logging) any sink fault. Returns true + /// on success, false when the sink threw — callers tally the false into their pass's failed-node + /// count so a swallowed materialise failure is operator-visible (archreview 01/S-1). + /// true when the folder was ensured; false when the sink threw. + private bool SafeEnsureFolder(string nodeId, string? parentNodeId, string displayName) { - try { _sink.EnsureFolder(nodeId, parentNodeId, displayName); } - catch (Exception ex) { _logger.LogWarning(ex, "AddressSpaceApplier: EnsureFolder threw for {Node}", nodeId); } + try { _sink.EnsureFolder(nodeId, parentNodeId, displayName); return true; } + catch (Exception ex) { _logger.LogWarning(ex, "AddressSpaceApplier: EnsureFolder threw for {Node}", nodeId); return false; } } - private void SafeEnsureVariable(string nodeId, string? parentNodeId, string displayName, string dataType, bool writable, string? historianTagname = null, bool isArray = false, uint? arrayLength = null) + /// Ensure a variable node, swallowing (and Warning-logging) any sink fault. Returns true + /// on success, false when the sink threw — callers tally the false into their pass's failed-node + /// count (archreview 01/S-1). + /// true when the variable was ensured; false when the sink threw. + private bool SafeEnsureVariable(string nodeId, string? parentNodeId, string displayName, string dataType, bool writable, string? historianTagname = null, bool isArray = false, uint? arrayLength = null) { - try { _sink.EnsureVariable(nodeId, parentNodeId, displayName, dataType, writable, historianTagname, isArray, arrayLength); } - catch (Exception ex) { _logger.LogWarning(ex, "AddressSpaceApplier: EnsureVariable threw for {Node}", nodeId); } + try { _sink.EnsureVariable(nodeId, parentNodeId, displayName, dataType, writable, historianTagname, isArray, arrayLength); return true; } + catch (Exception ex) { _logger.LogWarning(ex, "AddressSpaceApplier: EnsureVariable threw for {Node}", nodeId); return false; } } // A VirtualTag's materialised OPC UA node (MaterialiseEquipmentVirtualTags) is derived ONLY from @@ -674,22 +716,39 @@ public sealed class AddressSpaceApplier Severity: 0, Message: string.Empty); - private void SafeWriteAlarmCondition(string nodeId, AlarmConditionSnapshot state, DateTime ts) + /// Write an alarm-condition snapshot, swallowing (and Warning-logging) any sink fault. + /// Returns true on success, false when the sink threw — the removal pass tallies the + /// false into (archreview 01/S-1). + /// true when the write landed; false when the sink threw. + private bool SafeWriteAlarmCondition(string nodeId, AlarmConditionSnapshot state, DateTime ts) { - try { _sink.WriteAlarmCondition(nodeId, state, ts); } - catch (Exception ex) { _logger.LogWarning(ex, "AddressSpaceApplier: WriteAlarmCondition threw for {Node}", nodeId); } + try { _sink.WriteAlarmCondition(nodeId, state, ts); return true; } + catch (Exception ex) { _logger.LogWarning(ex, "AddressSpaceApplier: WriteAlarmCondition threw for {Node}", nodeId); return false; } } - private void SafeMaterialiseAlarmCondition(string alarmNodeId, string equipmentNodeId, string displayName, string alarmType, int severity, bool isNative) + /// Materialise an alarm-condition node, swallowing (and Warning-logging) any sink fault. + /// Returns true on success, false when the sink threw — callers tally the false into + /// their pass's failed-node count (archreview 01/S-1). + /// true when the condition was materialised; false when the sink threw. + private bool SafeMaterialiseAlarmCondition(string alarmNodeId, string equipmentNodeId, string displayName, string alarmType, int severity, bool isNative) { - try { _sink.MaterialiseAlarmCondition(alarmNodeId, equipmentNodeId, displayName, alarmType, severity, isNative); } - catch (Exception ex) { _logger.LogWarning(ex, "AddressSpaceApplier: MaterialiseAlarmCondition threw for {Node}", alarmNodeId); } + try { _sink.MaterialiseAlarmCondition(alarmNodeId, equipmentNodeId, displayName, alarmType, severity, isNative); return true; } + catch (Exception ex) { _logger.LogWarning(ex, "AddressSpaceApplier: MaterialiseAlarmCondition threw for {Node}", alarmNodeId); return false; } } } -/// Summary of one apply pass. Useful for tests + audit-log entries on the deploy path. +/// Summary of one apply pass. Useful for tests + audit-log entries on the deploy path. +/// means a rebuild was ATTEMPTED; means the +/// attempt threw (the sink's address space is now stale/partial). counts +/// swallowed per-node sink failures in 's own passes (the removal +/// alarm-condition writes). The Materialise* passes report their own failed-node tallies via their +/// int returns — the publish actor sums both channels so a degraded apply is operator-visible +/// (Error log + otopcua.opcua.apply.failed meter) instead of reported as optimistic success +/// (archreview 01/S-1). New trailing fields are defaulted so every existing construction compiles. public sealed record AddressSpaceApplyOutcome( int RemovedNodes, int AddedNodes, int ChangedNodes, - bool RebuildCalled); + bool RebuildCalled, + bool RebuildFailed = false, + int FailedNodes = 0); 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 42ffbe84..97e3eb68 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs @@ -192,14 +192,24 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers /// /// Cached local from the latest - /// snapshot (null = unknown until the first snapshot arrives, or no local node match). The inbound - /// write gate in reuses this signal — the SAME one the - /// scripted-alarm emit gate uses (ScriptedAlarmHostActor._localRole): only the Primary - /// services writes, default-allow while unknown so single-node deploys + the boot window never - /// reject (a node is the sole Primary until told otherwise). + /// snapshot (null = unknown until the first snapshot arrives, or no local node match). The Primary + /// data-plane gates (, , + /// ) resolve this through : a KNOWN + /// role wins outright; an UNKNOWN role is resolved by cluster membership — a single-driver cluster + /// stays default-ALLOW (boot-window / single-node), a multi-driver cluster default-DENIES until a + /// snapshot proves this node is Primary (archreview 03/S4 — closes the dual-primary boot window). /// private RedundancyRole? _localRole; + /// Test seam (archreview 03/S4): overrides the count of Up driver-role cluster members + /// the Primary gate reads while the role is unknown. Null ⇒ read the live cluster state (0 on a + /// non-cluster ActorRefProvider). See . + private readonly Func? _driverMemberCountProvider; + + /// Debounces the S5 "snapshot omitted this node" Warning to once per process — a persistent + /// identity mismatch (03/S5) would otherwise log on every snapshot. + private bool _warnedSnapshotMissingLocalNode; + /// Cached cluster DistributedPubSub mediator, resolved once in (on the /// actor thread) and reused for the Primary-gated native-alarm alerts fan-out in /// instead of re-resolving it per-publish. Mirrors @@ -308,11 +318,12 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers ILoggerFactory? loggerFactory = null, ScriptRootLogger? scriptRootLogger = null, IActorRef? scriptedAlarmHostOverride = null, - IDriverCapabilityInvokerFactory? invokerFactory = null) => + IDriverCapabilityInvokerFactory? invokerFactory = null, + Func? driverMemberCountProvider = null) => Akka.Actor.Props.Create(() => new DriverHostActor( dbFactory, localNode, coordinator, driverFactory, localRoles, dependencyMux, opcUaPublishActor, healthPublisher, virtualTagEvaluator, historyWriter, virtualTagHostOverride, - loggerFactory, scriptRootLogger, scriptedAlarmHostOverride, invokerFactory)); + loggerFactory, scriptRootLogger, scriptedAlarmHostOverride, invokerFactory, driverMemberCountProvider)); /// Initializes a new DriverHostActor with the specified dependencies. /// Database context factory for configuration database access. @@ -338,6 +349,9 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers /// Phase 6.1 resilience-invoker factory used to attach an /// to each spawned driver; defaults to /// (pass-through) when null. + /// Test seam (archreview 03/S4): overrides the count of Up + /// driver-role cluster members the Primary gate reads while the role is unknown. When null the + /// default reads Cluster.Get(Context.System).State.Members (0 on a non-cluster ActorRefProvider). public DriverHostActor( IDbContextFactory dbFactory, CommonsNodeId localNode, @@ -353,13 +367,15 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers ILoggerFactory? loggerFactory = null, ScriptRootLogger? scriptRootLogger = null, IActorRef? scriptedAlarmHostOverride = null, - IDriverCapabilityInvokerFactory? invokerFactory = null) + IDriverCapabilityInvokerFactory? invokerFactory = null, + Func? driverMemberCountProvider = null) { _dbFactory = dbFactory; _localNode = localNode; _coordinatorOverride = coordinator; _driverFactory = driverFactory ?? NullDriverFactory.Instance; _invokerFactory = invokerFactory ?? NullDriverCapabilityInvokerFactory.Instance; + _driverMemberCountProvider = driverMemberCountProvider; _localRoles = localRoles ?? new HashSet(StringComparer.Ordinal); _dependencyMux = dependencyMux; _opcUaPublishActor = opcUaPublishActor; @@ -964,17 +980,29 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers _localNode, msg.DriverInstanceId, msg.Args.ConditionId); return; } + // PRIMARY GATE (archreview 03/S4) — decided ONCE for this transition: the OPC UA condition write + // below stays UNGATED (a Secondary keeps its address space warm for failover), but only the Primary + // publishes the single cluster-wide alerts copy. Denied on a Secondary/Detached, OR on a multi-driver + // cluster while the role is unknown (default-deny closes the dual-primary boot window that would + // otherwise historize duplicate AVEVA alarm rows). Metered + Debug-logged once when denied. + var serviceAlertsAsPrimary = ShouldServiceAsPrimary(); + if (!serviceAlertsAsPrimary) + { + OtOpcUaTelemetry.PrimaryGateDenied.Add(1, + new KeyValuePair("site", "alarm-emit"), + new KeyValuePair("reason", PrimaryGateDenyReason())); + _log.Debug("DriverHost {Node}: native-alarm alerts publish suppressed for ({Driver},{Ref}) — not primary", + _localNode, msg.DriverInstanceId, msg.Args.ConditionId); + } + foreach (var nodeId in nodeIds) { var snapshot = _nativeAlarmProjector.Project(nodeId, msg.Args); _opcUaPublishActor.Tell(new ZB.MOM.WW.OtOpcUa.Runtime.OpcUa.OpcUaPublishActor.AlarmStateUpdate( nodeId, snapshot, msg.Args.SourceTimestampUtc)); - // Warm-standby dedup: the OPC UA condition write above is UNGATED (the secondary keeps its - // address space warm), but only the Primary publishes the cluster-wide alerts transition. - // Default-emit until told we are Secondary/Detached so single-node deploys + the boot window - // never drop transitions — the SAME signal HandleRouteNodeWrite gates writes on. - if (_localRole is RedundancyRole.Secondary or RedundancyRole.Detached) continue; + // Only the Primary publishes the cluster-wide alerts transition (see the gate above). + if (!serviceAlertsAsPrimary) continue; var meta = _alarmMetaByNodeId.TryGetValue(nodeId, out var m) ? m : (EquipmentId: nodeId, Name: nodeId, AlarmType: "AlarmCondition", HistorizeToAveva: (bool?)null); @@ -1034,11 +1062,18 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers /// private void HandleRouteNodeWrite(RouteNodeWrite msg) { - // PRIMARY GATE FIRST — only the Primary services operator writes (same signal as the alarm-emit - // gate; unknown role ⇒ treated as Primary so single-node deploys + the boot window aren't blocked). - if (_localRole is RedundancyRole.Secondary or RedundancyRole.Detached) + // PRIMARY GATE FIRST (archreview 03/S4) — only the Primary services operator writes. A KNOWN + // Secondary/Detached is denied "not primary" (unchanged); an UNKNOWN role denies ONLY on a + // multi-driver cluster ("not primary (role unknown)"), while a single-driver/boot-window cluster is + // still serviced. The rejection rides the established optimistic-write self-correction surface (the + // node reverts to its prior value with a Bad blip + a Part 8 audit event) — writes are NOT queued. + if (!ShouldServiceAsPrimary()) { - Sender.Tell(new NodeWriteResult(false, "not primary")); + var reason = _localRole is null ? "not primary (role unknown)" : "not primary"; + OtOpcUaTelemetry.PrimaryGateDenied.Add(1, + new KeyValuePair("site", "write"), + new KeyValuePair("reason", PrimaryGateDenyReason())); + Sender.Tell(new NodeWriteResult(false, reason)); return; } @@ -1090,12 +1125,25 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers /// private void HandleRouteNativeAlarmAck(RouteNativeAlarmAck msg) { - // PRIMARY GATE FIRST — only the Primary services operator acks (same signal as the inbound-write + - // alarm-emit gates; unknown role ⇒ treated as Primary so single-node deploys + the boot window aren't blocked). - if (_localRole is RedundancyRole.Secondary or RedundancyRole.Detached) + // PRIMARY GATE FIRST (archreview 03/S4) — only the Primary pushes the ack upstream. Same policy as + // the inbound-write gate. A KNOWN Secondary/Detached is dropped at Debug (unchanged); an UNKNOWN role + // on a multi-driver cluster is dropped at WARNING (an operator ack silently not reaching the upstream + // alarm system is a swallowed failure worth surfacing). + if (!ShouldServiceAsPrimary()) { - _log.Debug("DriverHost {Node}: dropping native-alarm ack for {Cond} — not primary", - _localNode, msg.ConditionNodeId); + OtOpcUaTelemetry.PrimaryGateDenied.Add(1, + new KeyValuePair("site", "ack"), + new KeyValuePair("reason", PrimaryGateDenyReason())); + if (_localRole is null) + { + _log.Warning("DriverHost {Node}: dropping native-alarm ack for {Cond} — not primary (role unknown on multi-driver cluster)", + _localNode, msg.ConditionNodeId); + } + else + { + _log.Debug("DriverHost {Node}: dropping native-alarm ack for {Cond} — not primary", + _localNode, msg.ConditionNodeId); + } return; } @@ -1119,9 +1167,43 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers entry.Actor.Tell(new DriverInstanceActor.RouteAlarmAck(target.FullName, msg.Comment, msg.OperatorUser)); } - /// Caches this node's from a cluster redundancy snapshot so - /// can gate inbound writes to the Primary. A snapshot that doesn't - /// mention this node leaves the cached role unchanged ⇒ default-allow. Mirrors + /// Whether this node should service a Primary-only data-plane operation right now — the single + /// gate decision (archreview 03/S4), delegating to with the live driver + /// member count. + /// true to service as Primary; false to deny. + private bool ShouldServiceAsPrimary() => + PrimaryGatePolicy.ShouldServiceAsPrimary(_localRole, DriverMemberCount()); + + /// Count of Up cluster members carrying the driver role. Uses the injected test seam when + /// present; otherwise reads the live cluster state guarded by try/catch so a non-cluster + /// ActorRefProvider (legacy/test harnesses) yields 0 ⇒ the single-node default-allow posture. + /// The number of Up driver-role members, or 0 when the cluster extension is unavailable. + private int DriverMemberCount() + { + if (_driverMemberCountProvider is not null) return _driverMemberCountProvider(); + try + { + return Akka.Cluster.Cluster.Get(Context.System).State.Members + .Count(m => m.Status == Akka.Cluster.MemberStatus.Up && m.Roles.Contains(ServiceCollectionExtensions.DriverRole)); + } + catch (Exception) + { + return 0; // non-cluster ActorRefProvider (legacy harnesses) ⇒ single-node posture + } + } + + /// The Primary-gate deny reason tag for the denial meter (secondary|detached|role-unknown). + private string PrimaryGateDenyReason() => _localRole switch + { + RedundancyRole.Secondary => "secondary", + RedundancyRole.Detached => "detached", + _ => "role-unknown", + }; + + /// Caches this node's from a cluster redundancy snapshot so the + /// Primary data-plane gates can resolve their decision. A snapshot that doesn't mention this node leaves + /// the cached role unchanged; when this node has a real driver peer (count > 1) that omission is the + /// 03/S5 identity-mismatch shape, logged ONCE at Warning. Mirrors /// ScriptedAlarmHostActor.OnRedundancyStateChanged / OpcUaPublishActor. private void OnRedundancyStateChanged(RedundancyStateChanged msg) { @@ -1129,6 +1211,18 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers if (local is not null) { _localRole = local.Role; + return; + } + + // S5 diagnostic: the snapshot never mentioned this node. On a multi-driver cluster that means the + // Primary gate stays default-DENY indefinitely for an unknown role — surface it once so a silent + // identity mismatch (e.g. an ApplicationUri/NodeId skew) is diagnosable. + if (!_warnedSnapshotMissingLocalNode && DriverMemberCount() > 1) + { + _warnedSnapshotMissingLocalNode = true; + _log.Warning( + "DriverHost {Node}: redundancy snapshot omitted this node (snapshotNodes={SnapshotNodes}); Primary data-plane gate stays default-DENY while role is unknown on a multi-driver cluster", + _localNode, string.Join(",", msg.Nodes.Select(n => n.NodeId))); } } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/PrimaryGatePolicy.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/PrimaryGatePolicy.cs new file mode 100644 index 00000000..90486a19 --- /dev/null +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/PrimaryGatePolicy.cs @@ -0,0 +1,28 @@ +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Redundancy; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Drivers; + +/// +/// Single decision point for the Primary data-plane gates (inbound device writes, native alarm acks, +/// fleet-wide alerts publishes). A KNOWN role wins outright; an UNKNOWN role (no redundancy snapshot +/// yet, or the snapshot never mentioned this node — the 03/S5 identity-mismatch shape) is resolved by +/// cluster membership: a single-driver cluster stays default-ALLOW (no peer exists — the +/// boot-window / single-node posture), a multi-driver cluster is default-DENY (a real Primary peer +/// exists; do not touch the shared field device until the redundancy snapshot proves this node is it). +/// Fixes archreview 03/S4 (primary gate defaulted to allow while role unknown → dual-primary window). +/// +public static class PrimaryGatePolicy +{ + /// Decide whether this node should service a Primary-only data-plane operation. + /// This node's last-known redundancy role, or null when unknown + /// (no snapshot received yet, or the snapshot omitted this node). + /// Count of Up cluster members carrying the driver role. + /// true to service as Primary; false to deny. + public static bool ShouldServiceAsPrimary(RedundancyRole? localRole, int driverMemberCount) => + localRole switch + { + RedundancyRole.Primary => true, + RedundancyRole.Secondary or RedundancyRole.Detached => false, + _ => driverMemberCount <= 1, // unknown: allow only when no driver peer exists + }; +} diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/OpcUa/OpcUaPublishActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/OpcUa/OpcUaPublishActor.cs index a22ffcad..b6648b2d 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/OpcUa/OpcUaPublishActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/OpcUa/OpcUaPublishActor.cs @@ -335,27 +335,43 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers var outcome = _applier.Apply(plan); _lastApplied = composition; - _applier.MaterialiseHierarchy(composition); + // Sum swallowed per-node materialise failures across every pass together with Apply's own + // removal-pass tally (archreview 01/S-1). A degraded apply used to vanish into per-node + // Warnings + an optimistic Info line; now it surfaces at Error + a dedicated meter. + var failedNodes = outcome.FailedNodes; + failedNodes += _applier.MaterialiseHierarchy(composition); // T14 — scripted alarms get their own pass right after the hierarchy so the equipment // folders they parent under already exist. Materialises real Part 9 AlarmConditionState // nodes (keyed by ScriptedAlarmId so AlarmStateUpdate writes target them); disabled // alarms are skipped. - _applier.MaterialiseScriptedAlarms(composition); + failedNodes += _applier.MaterialiseScriptedAlarms(composition); // Equipment-namespace tags get their own pass: ensures each signal's Variable (and any // FolderPath sub-folder) exists under its already-materialised equipment folder so // clients can browse them. Live values are pushed by DriverHostActor.ForwardToMux after // each subscription cycle; variables show BadWaitingForInitialData only until the first // publish interval fires. - _applier.MaterialiseEquipmentTags(composition); + failedNodes += _applier.MaterialiseEquipmentTags(composition); // Equipment-namespace VirtualTags get their own pass right after the equipment tags: // ensures each computed signal's Variable (and any FolderPath sub-folder) exists under its // equipment folder with a folder-scoped NodeId. VirtualTagHostActor.OnResult pushes live // values once the first dependency update arrives; until then variables show BadWaitingForInitialData. - _applier.MaterialiseEquipmentVirtualTags(composition); + failedNodes += _applier.MaterialiseEquipmentVirtualTags(composition); OtOpcUaTelemetry.OpcUaSinkWrite.Add(1, new KeyValuePair("kind", "rebuild")); - _log.Info("OpcUaPublish: applied rebuild (correlation={Correlation}, added={Added}, removed={Removed}, changed={Changed}, rebuild={Rebuild})", - msg.Correlation, outcome.AddedNodes, outcome.RemovedNodes, outcome.ChangedNodes, outcome.RebuildCalled); + if (outcome.RebuildFailed || failedNodes > 0) + { + OtOpcUaTelemetry.OpcUaApplyFailed.Add(1, + new KeyValuePair("kind", outcome.RebuildFailed ? "rebuild" : "nodes")); + _log.Error( + "OpcUaPublish: address-space apply DEGRADED (correlation={Correlation}, rebuildFailed={RebuildFailed}, failedNodes={FailedNodes}, added={Added}, removed={Removed}, changed={Changed}) — the running address space may be stale or partial", + msg.Correlation, outcome.RebuildFailed, failedNodes, + outcome.AddedNodes, outcome.RemovedNodes, outcome.ChangedNodes); + } + else + { + _log.Info("OpcUaPublish: applied rebuild (correlation={Correlation}, added={Added}, removed={Removed}, changed={Changed}, rebuild={Rebuild})", + msg.Correlation, outcome.AddedNodes, outcome.RemovedNodes, outcome.ChangedNodes, outcome.RebuildCalled); + } } catch (Exception ex) { @@ -412,7 +428,16 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers _log.Debug("OpcUaPublish: no applier wired — discarding MaterialiseDiscoveredNodes for {Equipment}", msg.EquipmentRootNodeId); return; } - _applier.MaterialiseDiscoveredNodes(msg.EquipmentRootNodeId, msg.Folders, msg.Variables); + var failedNodes = _applier.MaterialiseDiscoveredNodes(msg.EquipmentRootNodeId, msg.Folders, msg.Variables); + if (failedNodes > 0) + { + // archreview 01/S-1: a swallowed discovered-node injection failure surfaces at Error + the + // dedicated meter instead of vanishing into per-node Warnings. + OtOpcUaTelemetry.OpcUaApplyFailed.Add(1, new KeyValuePair("kind", "nodes")); + _log.Error( + "OpcUaPublish: discovered-node injection DEGRADED for {Equipment} (failedNodes={FailedNodes}) — some discovered nodes may be missing", + msg.EquipmentRootNodeId, failedNodes); + } } private void HandleServiceLevelChanged(ServiceLevelChanged msg) diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ScriptedAlarms/ScriptedAlarmHostActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ScriptedAlarms/ScriptedAlarmHostActor.cs index 7d93e51d..a8d790b8 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ScriptedAlarms/ScriptedAlarmHostActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ScriptedAlarms/ScriptedAlarmHostActor.cs @@ -3,11 +3,13 @@ using Akka.Cluster.Tools.PublishSubscribe; using Akka.Event; using ZB.MOM.WW.OtOpcUa.Commons.Messages.Alerts; using ZB.MOM.WW.OtOpcUa.Commons.Messages.Redundancy; +using ZB.MOM.WW.OtOpcUa.Commons.Observability; using ZB.MOM.WW.OtOpcUa.Commons.OpcUa; using ZB.MOM.WW.OtOpcUa.Commons.Types; using ZB.MOM.WW.OtOpcUa.Core.Abstractions; using ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms; using ZB.MOM.WW.OtOpcUa.OpcUaServer; +using ZB.MOM.WW.OtOpcUa.Runtime.Drivers; using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa; using ZB.MOM.WW.OtOpcUa.Runtime.VirtualTags; @@ -104,10 +106,16 @@ public sealed class ScriptedAlarmHostActor : ReceiveActor /// Cached local from the latest /// snapshot (null = unknown until the first snapshot arrives, or no wired). The - /// cluster-wide alerts publish in is gated on this: only the Primary - /// publishes (default-emit while unknown so single-node deploys + the boot window never drop transitions). + /// cluster-wide alerts publish in resolves this through + /// : only the Primary publishes; an unknown role default-emits on a + /// single-driver cluster but default-DENIES on a multi-driver one, so the dual-primary boot window can't + /// historize duplicate AVEVA alarm rows (archreview 03/S4). private RedundancyRole? _localRole; + /// Test seam (archreview 03/S4): overrides the count of Up driver-role cluster members the + /// alerts-emit gate reads while the role is unknown. Null ⇒ read the live cluster state. + private readonly Func? _driverMemberCountProvider; + /// Monotonic load generation, bumped on every . The continuation that /// pipes back an captures the generation it was started under; a stale /// completion (an earlier generation arriving after a newer apply) is discarded in @@ -136,8 +144,9 @@ public sealed class ScriptedAlarmHostActor : ReceiveActor IActorRef? mux, DependencyMuxTagUpstreamSource upstream, ScriptedAlarmEngine engine, - NodeId? localNode = null) => - Akka.Actor.Props.Create(() => new ScriptedAlarmHostActor(publishActor, mux, upstream, engine, localNode)); + NodeId? localNode = null, + Func? driverMemberCountProvider = null) => + Akka.Actor.Props.Create(() => new ScriptedAlarmHostActor(publishActor, mux, upstream, engine, localNode, driverMemberCountProvider)); /// Initializes a new instance of the class. /// The OPC UA publish actor emissions are bridged to. @@ -147,12 +156,16 @@ public sealed class ScriptedAlarmHostActor : ReceiveActor /// The local cluster node id, used to read this node's /// from the redundancy-state topic so only the Primary publishes the cluster-wide alerts /// transition. Null leaves the role unknown ⇒ default-emit (single-node deploys + tests). + /// Test seam (archreview 03/S4): overrides the count of Up + /// driver-role cluster members the alerts-emit gate reads while the role is unknown. Null reads + /// the live cluster state (0 on a non-cluster ActorRefProvider ⇒ single-node default-emit). public ScriptedAlarmHostActor( IActorRef publishActor, IActorRef? mux, DependencyMuxTagUpstreamSource upstream, ScriptedAlarmEngine engine, - NodeId? localNode = null) + NodeId? localNode = null, + Func? driverMemberCountProvider = null) { ArgumentNullException.ThrowIfNull(publishActor); ArgumentNullException.ThrowIfNull(upstream); @@ -162,6 +175,7 @@ public sealed class ScriptedAlarmHostActor : ReceiveActor _upstream = upstream; _engine = engine; _localNode = localNode; + _driverMemberCountProvider = driverMemberCountProvider; // OnEvent fires on the engine's worker thread. NEVER touch Context / actor state here — // marshal onto the actor thread via the thread-safe Self.Tell. Keep the handler in a field @@ -307,18 +321,45 @@ public sealed class ScriptedAlarmHostActor : ReceiveActor // suppress only the sink write. This publish (and the live `/alerts` fan-out) is NOT gated on it. HistorizeToAveva: e.HistorizeToAveva); - // Warm-standby dedup: only the Primary (driver-role leader) publishes the cluster-wide - // transition. Default-emit until told we are Secondary/Detached so single-node deploys + the - // boot window never drop transitions. The OPC UA node write above + inbound command processing - // stay ungated (the secondary keeps its address space + engine state warm for failover). - if (_localRole is RedundancyRole.Secondary or RedundancyRole.Detached) + // Warm-standby dedup: only the Primary publishes the cluster-wide transition (archreview 03/S4 — + // same PrimaryGatePolicy as DriverHostActor). A KNOWN Secondary/Detached is denied; an UNKNOWN role + // default-emits on a single-driver cluster but default-DENIES on a multi-driver one (so the + // dual-primary boot window can't historize duplicate AVEVA alarm rows). The OPC UA node write above + + // inbound command processing stay ungated (the secondary keeps its state warm for failover). + if (!PrimaryGatePolicy.ShouldServiceAsPrimary(_localRole, DriverMemberCount())) { + OtOpcUaTelemetry.PrimaryGateDenied.Add(1, + new KeyValuePair("site", "alarm-emit"), + new KeyValuePair("reason", _localRole switch + { + RedundancyRole.Secondary => "secondary", + RedundancyRole.Detached => "detached", + _ => "role-unknown", + })); return; } _mediator.Tell(new Publish(AlertsTopic, evt)); } + /// Count of Up cluster members carrying the driver role, for the alerts-emit gate. Uses + /// the injected test seam when present; otherwise reads the live cluster state guarded by try/catch so a + /// non-cluster ActorRefProvider yields 0 ⇒ the single-node default-emit posture (archreview 03/S4). + /// The number of Up driver-role members, or 0 when the cluster extension is unavailable. + private int DriverMemberCount() + { + if (_driverMemberCountProvider is not null) return _driverMemberCountProvider(); + try + { + return Akka.Cluster.Cluster.Get(Context.System).State.Members + .Count(m => m.Status == Akka.Cluster.MemberStatus.Up && m.Roles.Contains(ServiceCollectionExtensions.DriverRole)); + } + catch (Exception) + { + return 0; + } + } + /// Caches this node's from a cluster redundancy snapshot so /// can gate the cluster-wide alerts publish to the Primary. A /// snapshot that doesn't mention (or no local node wired) leaves the cached diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GatewayGalaxyDataWriterFailClosedTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GatewayGalaxyDataWriterFailClosedTests.cs new file mode 100644 index 00000000..5b514fba --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GatewayGalaxyDataWriterFailClosedTests.cs @@ -0,0 +1,211 @@ +using ZB.MOM.WW.MxGateway.Contracts.Proto; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Runtime; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests.Runtime; + +/// +/// archreview 06/S-1: Galaxy writes must fail CLOSED. When AdviseSupervisory fails, the raw +/// Write (which the file comment notes "doesn't throw but the value never reaches the galaxy") must +/// NOT be issued — the writer returns BadCommunicationError so the #5 node-revert fires instead +/// of leaving a phantom-Good node. These tests drive the real write pipeline through the internal +/// seam (the SDK session types are sealed + internal-ctor and cannot be faked). +/// +public sealed class GatewayGalaxyDataWriterFailClosedTests +{ + private static GatewayGalaxyDataWriter NewWriter() + => new(new GalaxyMxSession(new Config.GalaxyMxAccessOptions(ClientName: "OtOpcUa-Test")), writeUserId: 0); + + // ---------------- T5: seam ---------------- + + /// A secured-write classification routes through WriteSecured and never advises (regression via the seam). + [Fact] + public async Task WriteOne_SecuredClassification_RoutesThroughWriteSecured_NoAdvise() + { + var ops = new FakeMxWriteOps(); + var writer = NewWriter(); + + var result = await writer.WriteOneForTestAsync( + ops, new WriteRequest("Tag.Sec", true), SecurityClassification.SecuredWrite, CancellationToken.None); + + result.StatusCode.ShouldBe(StatusCodeMap.Good); + ops.Invokes.Count(r => r.Command.Kind == MxCommandKind.WriteSecured).ShouldBe(1); + ops.Invokes.Any(r => r.Command.Kind == MxCommandKind.AdviseSupervisory).ShouldBeFalse(); + ops.RawWriteHandles.ShouldBeEmpty(); + } + + // ---------------- T6: fail-closed on advise failure ---------------- + + /// When AdviseSupervisory fails (non-OK protocol status), the raw Write must NOT be issued and + /// the writer returns Bad (which fires the #5 node revert). The handle must not linger in the supervised + /// cache so the next write retries the advise. + [Fact] + public async Task WriteOne_AdviseFails_ReturnsBadCommunicationError_AndNeverIssuesRawWrite() + { + var ops = new FakeMxWriteOps + { + InvokeResponder = _ => new MxCommandReply + { + ProtocolStatus = new ProtocolStatus { Code = ProtocolStatusCode.MxaccessFailure, Message = "advise boom" }, + }, + }; + var writer = NewWriter(); + + var result = await writer.WriteOneForTestAsync( + ops, new WriteRequest("Tag.A", true), SecurityClassification.FreeAccess, CancellationToken.None); + + result.StatusCode.ShouldBe(StatusCodeMap.BadCommunicationError); + ops.Invokes.Count(r => r.Command.Kind == MxCommandKind.AdviseSupervisory).ShouldBe(1); + ops.RawWriteHandles.ShouldBeEmpty(); // the knowingly-lost raw write was NOT issued + writer.CachedSupervisedHandleCount.ShouldBe(0); // advise failure forgot the handle → next write retries + } + + /// When advise succeeds, the raw Write proceeds and a clean reply returns Good (regression). + [Fact] + public async Task WriteOne_AdviseSucceeds_RawWriteProceeds() + { + var ops = new FakeMxWriteOps + { + InvokeResponder = _ => new MxCommandReply { ProtocolStatus = new ProtocolStatus { Code = ProtocolStatusCode.Ok } }, + }; + var writer = NewWriter(); + + var result = await writer.WriteOneForTestAsync( + ops, new WriteRequest("Tag.B", 42), SecurityClassification.FreeAccess, CancellationToken.None); + + result.StatusCode.ShouldBe(StatusCodeMap.Good); + ops.Invokes.Count(r => r.Command.Kind == MxCommandKind.AdviseSupervisory).ShouldBe(1); + ops.RawWriteHandles.ShouldHaveSingleItem(); // the raw write was issued after the successful advise + writer.CachedSupervisedHandleCount.ShouldBe(1); + } + + // ---------------- T7: counters ---------------- + + /// A fail-closed advise increments galaxy.writes.advise_failed (and not unconfirmed). + [Fact] + public async Task WriteOne_AdviseFails_Increments_AdviseFailed_Meter() + { + using var advise = new MeterRecorder("galaxy.writes.advise_failed"); + using var unconfirmed = new MeterRecorder("galaxy.writes.unconfirmed"); + var ops = new FakeMxWriteOps + { + InvokeResponder = _ => new MxCommandReply { ProtocolStatus = new ProtocolStatus { Code = ProtocolStatusCode.MxaccessFailure } }, + }; + var writer = NewWriter(); + + await writer.WriteOneForTestAsync(ops, new WriteRequest("Tag.A", true), SecurityClassification.FreeAccess, CancellationToken.None); + + advise.Total.ShouldBe(1); + unconfirmed.Total.ShouldBe(0); + } + + /// An advised write whose reply carries an EMPTY statuses array returns Good but increments + /// galaxy.writes.unconfirmed (the commit is unconfirmed pending gateway WriteComplete correlation). + [Fact] + public async Task WriteOne_EmptyStatuses_ReturnsGood_And_Increments_Unconfirmed_Meter() + { + using var advise = new MeterRecorder("galaxy.writes.advise_failed"); + using var unconfirmed = new MeterRecorder("galaxy.writes.unconfirmed"); + var ops = new FakeMxWriteOps + { + InvokeResponder = _ => new MxCommandReply { ProtocolStatus = new ProtocolStatus { Code = ProtocolStatusCode.Ok } }, + RawWriteResponder = () => new MxCommandReply(), // no ProtocolStatus, EMPTY statuses ⇒ Good + }; + var writer = NewWriter(); + + var result = await writer.WriteOneForTestAsync(ops, new WriteRequest("Tag.B", 1), SecurityClassification.FreeAccess, CancellationToken.None); + + result.StatusCode.ShouldBe(StatusCodeMap.Good); + unconfirmed.Total.ShouldBe(1); + advise.Total.ShouldBe(0); + } + + /// A reply carrying an MX status ROW increments neither meter (not empty-statuses, advise OK). + [Fact] + public async Task WriteOne_StatusRowReply_Increments_Neither_Meter() + { + using var advise = new MeterRecorder("galaxy.writes.advise_failed"); + using var unconfirmed = new MeterRecorder("galaxy.writes.unconfirmed"); + var okStatusReply = new MxCommandReply { ProtocolStatus = new ProtocolStatus { Code = ProtocolStatusCode.Ok } }; + okStatusReply.Statuses.Add(new MxStatusProxy { Success = 1, Category = MxStatusCategory.Ok }); + var ops = new FakeMxWriteOps + { + InvokeResponder = _ => new MxCommandReply { ProtocolStatus = new ProtocolStatus { Code = ProtocolStatusCode.Ok } }, + RawWriteResponder = () => okStatusReply, + }; + var writer = NewWriter(); + + var result = await writer.WriteOneForTestAsync(ops, new WriteRequest("Tag.C", 2), SecurityClassification.FreeAccess, CancellationToken.None); + + result.StatusCode.ShouldBe(StatusCodeMap.Good); // Success/Ok status row ⇒ Good + advise.Total.ShouldBe(0); + unconfirmed.Total.ShouldBe(0); // NOT empty-statuses ⇒ not metered unconfirmed + } + + // ---------------- fakes ---------------- + + /// Listens to a single instrument by name on the Galaxy driver meter and sums the values. + private sealed class MeterRecorder : IDisposable + { + private readonly string _name; + private readonly System.Diagnostics.Metrics.MeterListener _listener; + private long _total; + private readonly object _gate = new(); + + public MeterRecorder(string instrumentName) + { + _name = instrumentName; + _listener = new System.Diagnostics.Metrics.MeterListener + { + InstrumentPublished = (instrument, listener) => + { + if (instrument.Meter.Name == EventPump.MeterName && instrument.Name == _name) + listener.EnableMeasurementEvents(instrument); + } + }; + _listener.SetMeasurementEventCallback((_, value, _, _) => { lock (_gate) _total += value; }); + _listener.Start(); + } + + public long Total { get { lock (_gate) return _total; } } + + public void Dispose() => _listener.Dispose(); + } + + /// A fake that records every call and returns canned replies. + private sealed class FakeMxWriteOps : IMxWriteOps + { + public string SessionId => "sess-1"; + public int ServerHandle => 100; + public int NextHandle { get; init; } = 7; + + public List AddItemRefs { get; } = new(); + public List Invokes { get; } = new(); + public List RawWriteHandles { get; } = new(); + + /// Responder for ; null ⇒ an OK (empty) reply. + public Func? InvokeResponder { get; init; } + /// Responder for ; null ⇒ an empty (Good) reply. + public Func? RawWriteResponder { get; init; } + + public Task AddItemAsync(string fullRef, CancellationToken ct) + { + AddItemRefs.Add(fullRef); + return Task.FromResult(NextHandle); + } + + public Task InvokeAsync(MxCommandRequest request, CancellationToken ct) + { + Invokes.Add(request); + return Task.FromResult(InvokeResponder?.Invoke(request) ?? new MxCommandReply()); + } + + public Task WriteRawAsync(int itemHandle, MxValue value, int userId, CancellationToken ct) + { + RawWriteHandles.Add(itemHandle); + return Task.FromResult(RawWriteResponder?.Invoke() ?? new MxCommandReply()); + } + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GatewayGalaxyLiveReopenAndWriteTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GatewayGalaxyLiveReopenAndWriteTests.cs index f2a3cb1e..95a2e68c 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GatewayGalaxyLiveReopenAndWriteTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GatewayGalaxyLiveReopenAndWriteTests.cs @@ -173,6 +173,34 @@ public sealed class GatewayGalaxyLiveReopenAndWriteTests $"borrow smoke: subscribed {WriteRef} -> handle {match.ItemHandle}; borrowed write Good with AddItemCallCount=0; control AddItemCallCount=1"); } + /// + /// archreview 06/S-1 guard — after the fail-closed advise change, a NORMAL advised no-login write + /// must STILL return Good (0). Guards against over-eager fail-closing: the advise round-trip succeeds + /// on a healthy gateway, so the raw Write proceeds and commits. If this ever returns Bad on a healthy + /// gateway, the fail-closed branch is mis-judging a successful advise. Skip-gated like the others. + /// + [Fact] + public async Task Live_normal_advised_write_still_returns_good_under_failclosed_advise() + { + var (endpoint, apiKey) = RequireLiveGatewayOrSkip(); + var ct = TestContext.Current.CancellationToken; + var clientOptions = BuildClientOptions(endpoint, apiKey); + + await using var session = new GalaxyMxSession(new GalaxyMxAccessOptions(ClientName: "OtOpcUaFailClosedGuard")); + await session.ConnectAsync(clientOptions, ct); + var writer = new GatewayGalaxyDataWriter(session, writeUserId: 0); + + var result = await writer.WriteAsync( + [new WriteRequest(WriteRef, 5151.0f)], _ => SecurityClassification.FreeAccess, ct); + + result.ShouldHaveSingleItem().StatusCode.ShouldBe(0u, + "a normal advised write must still return Good after the fail-closed advise change (no over-eager fail-close)"); + writer.CachedSupervisedHandleCount.ShouldBe(1, "the healthy advise should have supervised the handle"); + + TestContext.Current.SendDiagnosticMessage( + $"fail-closed guard smoke: normal advised write {WriteRef}=5151 returned Good; handle supervised"); + } + private static (string Endpoint, string ApiKey) RequireLiveGatewayOrSkip() { var endpoint = Environment.GetEnvironmentVariable("MXGW_ENDPOINT"); diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/PrimaryGateFailoverTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/PrimaryGateFailoverTests.cs new file mode 100644 index 00000000..c9ca2b92 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/PrimaryGateFailoverTests.cs @@ -0,0 +1,75 @@ +using Akka.Actor; +using Akka.Hosting; +using Microsoft.Extensions.DependencyInjection; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Runtime; +using ZB.MOM.WW.OtOpcUa.Runtime.Drivers; + +namespace ZB.MOM.WW.OtOpcUa.Host.IntegrationTests; + +/// +/// archreview 03/S4 delivery-path guard — the "unit tests can't catch delivery" half. Proves that the +/// REAL DistributedPubSub-delivered redundancy snapshot (not a test-injected RedundancyStateChanged) +/// actually drives the Primary write gate end-to-end across a live 2-node Akka cluster: after both nodes +/// join and the snapshot converges, exactly ONE node's rejects a +/// with "not primary" (the Secondary) while the other +/// passes the gate (the Primary — its rejection, if any, is about node MAPPING, never the gate). +/// +/// Before the snapshot delivers, BOTH nodes have an unknown role on a 2-driver cluster and so both +/// default-DENY with "not primary (role unknown)" (archreview 03/S4). The poll below therefore +/// waits for the delivered snapshot to promote exactly one node to Primary — a broken redundancy-topic +/// subscribe (the historical double-break) would leave both stuck "role unknown" and time this out, which +/// is exactly the negative control this test provides over the pure-TestKit guards. +/// +[Trait("Category", "Failover")] +public sealed class PrimaryGateFailoverTests +{ + private static CancellationToken Ct => TestContext.Current.CancellationToken; + + // A NodeId that is guaranteed to have no driver mapping — so a node that PASSES the primary gate + // rejects it with a MAPPING reason ("no driver mapping for node …"), never a gate reason. + private const string UnmappedProbeNode = "__pgate_delivery_probe__"; + + [Fact] + public async Task Delivered_redundancy_snapshot_drives_the_primary_write_gate_across_the_cluster() + { + await using var harness = await TwoNodeClusterHarness.StartAsync(); + + var driverHostA = harness.NodeA.Services.GetRequiredService().Get(); + var driverHostB = harness.NodeB.Services.GetRequiredService().Get(); + + // Poll until the DELIVERED snapshot has promoted exactly one node to Primary: one host passes the + // gate (reason not "not primary*") and the other is gated ("not primary"). Deadline covers the + // ~250ms-debounced initial publish plus cluster convergence margin. + var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(45); + (bool aGated, bool bGated, string? aReason, string? bReason) last = default; + while (DateTime.UtcNow < deadline) + { + var a = await RouteProbeWriteAsync(driverHostA); + var b = await RouteProbeWriteAsync(driverHostB); + last = (IsGateReject(a), IsGateReject(b), a.Reason, b.Reason); + + // Exactly one node gated ⇒ the snapshot converged (one Primary, one Secondary). + if (last.aGated ^ last.bGated) break; + await Task.Delay(500, Ct); + } + + // Exactly one node is the Secondary (gate-rejected); the other is the Primary (passed the gate — + // its reject, if any, is a MAPPING reason). This proves the delivered snapshot drives the gate. + (last.aGated ^ last.bGated).ShouldBeTrue( + $"exactly one node should be gated once the redundancy snapshot converges (A gated={last.aGated} reason='{last.aReason}', B gated={last.bGated} reason='{last.bReason}')"); + + var primaryReason = last.aGated ? last.bReason : last.aReason; + primaryReason.ShouldNotBeNull(); + primaryReason!.StartsWith("not primary", StringComparison.Ordinal).ShouldBeFalse( + $"the Primary must PASS the gate — its reject reason should be a mapping reason, not '{primaryReason}'"); + } + + private static async Task RouteProbeWriteAsync(IActorRef driverHost) + => await driverHost.Ask( + new DriverHostActor.RouteNodeWrite(UnmappedProbeNode, 0.0), TimeSpan.FromSeconds(10)); + + private static bool IsGateReject(DriverHostActor.NodeWriteResult r) + => !r.Success && r.Reason is not null && r.Reason.StartsWith("not primary", StringComparison.Ordinal); +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/AddressSpaceApplierFailureSurfaceTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/AddressSpaceApplierFailureSurfaceTests.cs new file mode 100644 index 00000000..c4a1b75d --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/AddressSpaceApplierFailureSurfaceTests.cs @@ -0,0 +1,185 @@ +using Microsoft.Extensions.Logging.Abstractions; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.OpcUa; + +namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests; + +/// +/// Guards the failure-visibility surface added for archreview 01/S-1: the applier must report a +/// failed rebuild () and tally swallowed per-node +/// sink failures ( for Apply's own passes; the +/// Materialise* passes return their own failed-node counts) instead of reporting optimistic +/// success while the running address space is stale or partial. +/// +public sealed class AddressSpaceApplierFailureSurfaceTests +{ + // ---------------- T1: RebuildFailed ---------------- + + /// A rebuild whose sink call throws is reported as attempted-but-failed, not optimistic success. + [Fact] + public void Apply_WhenRebuildThrows_ReportsRebuildFailed() + { + var sink = new ConfigurableThrowingSink { ThrowOnRebuild = true }; + var applier = new AddressSpaceApplier(sink, NullLogger.Instance); + + var outcome = applier.Apply(AddedEquipmentPlan("new")); + + outcome.RebuildCalled.ShouldBeTrue(); + outcome.RebuildFailed.ShouldBeTrue(); + } + + /// A clean apply reports no failure on either channel — the regression guard for the new fields. + [Fact] + public void Apply_HappyPath_ReportsNoFailure() + { + var sink = new ConfigurableThrowingSink(); + var applier = new AddressSpaceApplier(sink, NullLogger.Instance); + + var outcome = applier.Apply(AddedEquipmentPlan("new")); + + outcome.RebuildCalled.ShouldBeTrue(); + outcome.RebuildFailed.ShouldBeFalse(); + outcome.FailedNodes.ShouldBe(0); + } + + // ---------------- T2: FailedNodes (removal-pass tally) ---------------- + + /// A removal-pass alarm-condition write that throws is counted into FailedNodes (swallowed today). + [Fact] + public void Apply_WhenRemovalConditionWriteThrows_CountsFailedNodes() + { + var sink = new ConfigurableThrowingSink { ThrowOnAlarmWrite = true }; + var applier = new AddressSpaceApplier(sink, NullLogger.Instance); + + var outcome = applier.Apply(EquipmentRemovalPlan("eq-1")); + + outcome.RemovedNodes.ShouldBe(1); + outcome.FailedNodes.ShouldBe(1); + } + + // ---------------- T3: Materialise* passes return failed-node counts ---------------- + + /// Every EnsureVariable throwing in MaterialiseEquipmentTags is tallied into the pass's int return. + [Fact] + public void MaterialiseEquipmentTags_WhenEnsureVariableThrows_ReturnsFailedCount() + { + var sink = new ConfigurableThrowingSink { ThrowOnEnsureVariable = true }; + var applier = new AddressSpaceApplier(sink, NullLogger.Instance); + + var composition = CompositionWithTags( + new EquipmentTagPlan("t1", "eq-1", "drv", FolderPath: "", Name: "A", DataType: "Float", FullName: "40001", Writable: false, Alarm: null), + new EquipmentTagPlan("t2", "eq-1", "drv", FolderPath: "", Name: "B", DataType: "Float", FullName: "40002", Writable: false, Alarm: null), + new EquipmentTagPlan("t3", "eq-1", "drv", FolderPath: "", Name: "C", DataType: "Float", FullName: "40003", Writable: false, Alarm: null)); + + applier.MaterialiseEquipmentTags(composition).ShouldBe(3); + } + + /// All five Materialise* passes return 0 failures on a clean sink — the happy-path regression. + [Fact] + public void Materialise_HappyPath_AllPassesReturnZero() + { + var sink = new ConfigurableThrowingSink(); + var applier = new AddressSpaceApplier(sink, NullLogger.Instance); + + var tags = CompositionWithTags( + new EquipmentTagPlan("t1", "eq-1", "drv", FolderPath: "", Name: "A", DataType: "Float", FullName: "40001", Writable: false, Alarm: null)); + var vtags = CompositionWithVirtualTags( + new EquipmentVirtualTagPlan("v1", "eq-1", FolderPath: "", Name: "VA", DataType: "Float", Expression: "1", DependencyRefs: Array.Empty(), Historize: false)); + var alarms = CompositionWithScriptedAlarms( + new EquipmentScriptedAlarmPlan("a1", "eq-1", "Alarm", "OffNormalAlarm", 500, "msg", "pred", "true", + Array.Empty(), HistorizeToAveva: false, Retain: true, Enabled: true)); + + applier.MaterialiseHierarchy(tags).ShouldBe(0); + applier.MaterialiseEquipmentTags(tags).ShouldBe(0); + applier.MaterialiseEquipmentVirtualTags(vtags).ShouldBe(0); + applier.MaterialiseScriptedAlarms(alarms).ShouldBe(0); + applier.MaterialiseDiscoveredNodes( + "eq-1", + Array.Empty(), + new[] { new DiscoveredVariable("eq-1/D", "eq-1", "D", "Float", Writable: false, IsArray: false, ArrayLength: null) }) + .ShouldBe(0); + } + + // ---------------- fixtures ---------------- + + private static AddressSpaceComposition CompositionWithTags(params EquipmentTagPlan[] tags) => + new(Array.Empty(), Array.Empty(), Array.Empty()) + { + EquipmentTags = tags, + }; + + private static AddressSpaceComposition CompositionWithVirtualTags(params EquipmentVirtualTagPlan[] vtags) => + new(Array.Empty(), Array.Empty(), Array.Empty()) + { + EquipmentVirtualTags = vtags, + }; + + private static AddressSpaceComposition CompositionWithScriptedAlarms(params EquipmentScriptedAlarmPlan[] alarms) => + new(Array.Empty(), Array.Empty(), Array.Empty()) + { + EquipmentScriptedAlarms = alarms, + }; + + private static AddressSpacePlan AddedEquipmentPlan(string id) => new( + AddedEquipment: new[] { new EquipmentNode(id, id, "line-1") }, + RemovedEquipment: Array.Empty(), + ChangedEquipment: Array.Empty(), + AddedDrivers: Array.Empty(), + RemovedDrivers: Array.Empty(), + ChangedDrivers: Array.Empty(), + AddedAlarms: Array.Empty(), + RemovedAlarms: Array.Empty(), + ChangedAlarms: Array.Empty()); + + private static AddressSpacePlan EquipmentRemovalPlan(params string[] ids) => new( + AddedEquipment: Array.Empty(), + RemovedEquipment: ids.Select(id => new EquipmentNode(id, id, "line-1")).ToArray(), + ChangedEquipment: Array.Empty(), + AddedDrivers: Array.Empty(), + RemovedDrivers: Array.Empty(), + ChangedDrivers: Array.Empty(), + AddedAlarms: Array.Empty(), + RemovedAlarms: Array.Empty(), + ChangedAlarms: Array.Empty()); + + /// An whose every sink call can be independently made to + /// throw, so tests can drive each Safe* helper's catch branch. Non-throwing calls are no-ops. + private sealed class ConfigurableThrowingSink : IOpcUaAddressSpaceSink + { + public bool ThrowOnRebuild { get; init; } + public bool ThrowOnEnsureVariable { get; init; } + public bool ThrowOnEnsureFolder { get; init; } + public bool ThrowOnAlarmWrite { get; init; } + public bool ThrowOnMaterialiseAlarm { get; init; } + + public void WriteValue(string nodeId, object? value, OpcUaQuality quality, DateTime sourceTimestampUtc) { } + + public void WriteAlarmCondition(string alarmNodeId, AlarmConditionSnapshot state, DateTime sourceTimestampUtc) + { + if (ThrowOnAlarmWrite) throw new InvalidOperationException("simulated WriteAlarmCondition fault"); + } + + public void MaterialiseAlarmCondition(string alarmNodeId, string equipmentNodeId, string displayName, string alarmType, int severity, bool isNative = false) + { + if (ThrowOnMaterialiseAlarm) throw new InvalidOperationException("simulated MaterialiseAlarmCondition fault"); + } + + public void EnsureFolder(string folderNodeId, string? parentNodeId, string displayName) + { + if (ThrowOnEnsureFolder) throw new InvalidOperationException("simulated EnsureFolder fault"); + } + + public void EnsureVariable(string variableNodeId, string? parentFolderNodeId, string displayName, string dataType, bool writable, string? historianTagname = null, bool isArray = false, uint? arrayLength = null) + { + if (ThrowOnEnsureVariable) throw new InvalidOperationException("simulated EnsureVariable fault"); + } + + public void RebuildAddressSpace() + { + if (ThrowOnRebuild) throw new InvalidOperationException("simulated RebuildAddressSpace fault"); + } + + public void RaiseNodesAddedModelChange(string affectedNodeId) { } + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorPrimaryGateTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorPrimaryGateTests.cs new file mode 100644 index 00000000..036a9007 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorPrimaryGateTests.cs @@ -0,0 +1,425 @@ +using System.Collections.Concurrent; +using System.Diagnostics.Metrics; +using System.Text.Json; +using Akka.Actor; +using Akka.Cluster.Tools.PublishSubscribe; +using Akka.TestKit; +using Microsoft.EntityFrameworkCore; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.Engines; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Alerts; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Fleet; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Redundancy; +using ZB.MOM.WW.OtOpcUa.Commons.Observability; +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; +using ZB.MOM.WW.OtOpcUa.Runtime.ScriptedAlarms; +using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers; + +/// +/// archreview 03/S4 — the boot-window primary-gate guard. On a MULTI-driver cluster (member count > 1) +/// while the role is UNKNOWN, the Primary data-plane gates (inbound write, native ack, native alerts emit) +/// default-DENY; on a SINGLE-driver cluster (count <= 1) they default-ALLOW (boot-window / single-node +/// posture). Driver member count is injected via the driverMemberCountProvider Props seam. +/// +public sealed class DriverHostActorPrimaryGateTests : RuntimeActorTestBase +{ + private static readonly NodeId TestNode = NodeId.Parse("driver-pgate-test"); + private static readonly RevisionHash RevA = RevisionHash.Parse(new string('a', 64)); + private static readonly DateTime Ts = new(2026, 7, 13, 10, 0, 0, DateTimeKind.Utc); + private static readonly TimeSpan Timeout = TimeSpan.FromSeconds(5); + + // ---------------- write gate ---------------- + + /// Role unknown + a real driver peer (count 2) ⇒ RouteNodeWrite is DENIED with the + /// boot-window reason and the driver sees no write. + [Fact] + public void Unknown_role_multi_driver_denies_write_with_role_unknown_reason() + { + var db = NewInMemoryDbFactory(); + var recorder = new RecordingDriverFactory("Modbus"); + var dep = SeedWriteTagDeployment(db); + var actor = SpawnWriteHost(db, dep, recorder, driverMemberCount: 2); + + var asker = CreateTestProbe(); + actor.Tell(new DriverHostActor.RouteNodeWrite("eq-1/speed", 123.0), asker.Ref); + + var result = asker.ExpectMsg(Timeout); + result.Success.ShouldBeFalse(); + result.Reason.ShouldBe("not primary (role unknown)"); + recorder.Writes.ShouldBeEmpty(); + } + + /// Role unknown + no driver peer (count 1) ⇒ the write is SERVICED (single-node / boot-window + /// posture preserved). + [Fact] + public void Unknown_role_single_driver_services_write() + { + var db = NewInMemoryDbFactory(); + var recorder = new RecordingDriverFactory("Modbus"); + var dep = SeedWriteTagDeployment(db); + var actor = SpawnWriteHost(db, dep, recorder, driverMemberCount: 1); + + // The gate must ALLOW (count 1, role unknown). The driver child's async connect can lag the ApplyAck, + // so a first write may transiently report "driver not connected" — retry until connected (a real + // client retries too). The point under test is that the gate never denies (never "not primary"). + AwaitAssert(() => + { + var asker = CreateTestProbe(); + actor.Tell(new DriverHostActor.RouteNodeWrite("eq-1/speed", 123.0), asker.Ref); + var res = asker.ExpectMsg(Timeout); + res.Reason.ShouldNotBe("not primary"); + res.Reason.ShouldNotBe("not primary (role unknown)"); + res.Success.ShouldBeTrue(res.Reason); + }, duration: TimeSpan.FromSeconds(10)); + recorder.Writes.ShouldNotBeEmpty(); + } + + /// A Primary snapshot services even at count 2; a Secondary snapshot denies with the steady-state + /// reason (distinct from the boot-window reason). + [Fact] + public void Known_role_wins_over_member_count() + { + var db = NewInMemoryDbFactory(); + var recorder = new RecordingDriverFactory("Modbus"); + var dep = SeedWriteTagDeployment(db); + var actor = SpawnWriteHost(db, dep, recorder, driverMemberCount: 2); + + // Primary snapshot ⇒ serviced even with a driver peer. Retry through the transient driver-connect + // window; the invariant under test is that the gate never denies a Primary. + TellRole(actor, RedundancyRole.Primary); + AwaitAssert(() => + { + var asker1 = CreateTestProbe(); + actor.Tell(new DriverHostActor.RouteNodeWrite("eq-1/speed", 1.0), asker1.Ref); + var res = asker1.ExpectMsg(Timeout); + res.Reason.ShouldNotBe("not primary"); + res.Success.ShouldBeTrue(res.Reason); + }, duration: TimeSpan.FromSeconds(10)); + + // Secondary snapshot ⇒ denied with the steady-state reason. + TellRole(actor, RedundancyRole.Secondary); + var asker2 = CreateTestProbe(); + actor.Tell(new DriverHostActor.RouteNodeWrite("eq-1/speed", 2.0), asker2.Ref); + var denied = asker2.ExpectMsg(Timeout); + denied.Success.ShouldBeFalse(); + denied.Reason.ShouldBe("not primary"); + } + + /// The denial meter carries the site + reason tags. + [Fact] + public void Denied_write_increments_primary_gate_denied_meter_with_tags() + { + using var recorder = new MeterRecorder("otopcua.redundancy.primary_gate_denied"); + var db = NewInMemoryDbFactory(); + var factory = new RecordingDriverFactory("Modbus"); + var dep = SeedWriteTagDeployment(db); + var actor = SpawnWriteHost(db, dep, factory, driverMemberCount: 2); + + var asker = CreateTestProbe(); + actor.Tell(new DriverHostActor.RouteNodeWrite("eq-1/speed", 9.0), asker.Ref); + asker.ExpectMsg(Timeout).Success.ShouldBeFalse(); + + AwaitAssert(() => + { + recorder.Total.ShouldBeGreaterThanOrEqualTo(1); + recorder.WithTag("site", "write").ShouldBeGreaterThanOrEqualTo(1); + recorder.WithTag("reason", "role-unknown").ShouldBeGreaterThanOrEqualTo(1); + }, duration: Timeout); + } + + // ---------------- ack gate ---------------- + + /// Role unknown + count 2 ⇒ a native-alarm ack is dropped at Warning and the denial meter + /// increments with site=ack. + [Fact] + public void Unknown_role_multi_driver_drops_native_ack_with_warning_and_meter() + { + using var meter = new MeterRecorder("otopcua.redundancy.primary_gate_denied"); + var db = NewInMemoryDbFactory(); + var dep = SeedAlarmTagDeployment(db); + var (actor, _) = SpawnAlarmHost(db, dep, driverMemberCount: 2); + + EventFilter.Warning(contains: "role unknown").ExpectOne(() => + actor.Tell(new DriverHostActor.RouteNativeAlarmAck("eq-1/temp_hi", Comment: null, OperatorUser: "op"))); + + AwaitAssert(() => + { + meter.WithTag("site", "ack").ShouldBeGreaterThanOrEqualTo(1); + }, duration: Timeout); + } + + // ---------------- alarm-emit gate ---------------- + + /// Role unknown + count 2 ⇒ ForwardNativeAlarm still writes the (ungated) OPC UA condition update + /// but publishes NO cluster-wide alerts transition. + [Fact] + public void Unknown_role_multi_driver_suppresses_alerts_emit_but_updates_condition() + { + var db = NewInMemoryDbFactory(); + var dep = SeedAlarmTagDeployment(db); + var alerts = CreateTestProbe(); + SubscribeToAlerts(alerts); + var (actor, publish) = SpawnAlarmHost(db, dep, driverMemberCount: 2); + + actor.Tell(RaiseAlarm()); + + // The ungated OPC UA condition write still arrives. + publish.ExpectMsg(Timeout).AlarmNodeId.ShouldBe("eq-1/temp_hi"); + // The cluster-wide alerts publish is suppressed (a real Primary peer publishes the single copy). + alerts.ExpectNoMsg(TimeSpan.FromMilliseconds(500)); + } + + /// Role unknown + count 1 ⇒ ForwardNativeAlarm publishes the alerts transition (single-node + /// posture preserved). + [Fact] + public void Unknown_role_single_driver_publishes_alerts_emit() + { + var db = NewInMemoryDbFactory(); + var dep = SeedAlarmTagDeployment(db); + var alerts = CreateTestProbe(); + SubscribeToAlerts(alerts); + var (actor, publish) = SpawnAlarmHost(db, dep, driverMemberCount: 1); + + actor.Tell(RaiseAlarm()); + + publish.ExpectMsg(Timeout); + alerts.ExpectMsg(Timeout).AlarmId.ShouldBe("eq-1/temp_hi"); + } + + // ---------------- helpers ---------------- + + private static DriverInstanceActor.AttributeAlarmPublished RaiseAlarm() => + new("drv-1", new AlarmEventArgs( + new StubAlarmHandle(), + SourceNodeId: "Temp", + ConditionId: "Temp.HiHi", + AlarmType: "OffNormalAlarm", + Message: "temperature high", + Severity: AlarmSeverity.High, + SourceTimestampUtc: Ts, + Kind: AlarmTransitionKind.Raise)); + + private static void TellRole(IActorRef host, RedundancyRole role) => + host.Tell(new RedundancyStateChanged( + new[] + { + new NodeRedundancyState(TestNode, role, + IsClusterLeader: role == RedundancyRole.Primary, + IsRoleLeaderForDriver: role == RedundancyRole.Primary, + AsOfUtc: DateTime.UtcNow), + }, + CorrelationId.NewId())); + + private IActorRef SpawnWriteHost( + IDbContextFactory db, DeploymentId dep, IDriverFactory factory, int driverMemberCount) + { + var coordinator = CreateTestProbe(); + var actor = Sys.ActorOf(DriverHostActor.Props( + db, TestNode, coordinator.Ref, + driverFactory: factory, + localRoles: new HashSet { "driver" }, + driverMemberCountProvider: () => driverMemberCount)); + + actor.Tell(new DispatchDeployment(dep, RevA, CorrelationId.NewId())); + coordinator.ExpectMsg(Timeout).Outcome.ShouldBe(ApplyAckOutcome.Applied); + return actor; + } + + private (IActorRef Actor, TestProbe Publish) SpawnAlarmHost( + IDbContextFactory db, DeploymentId dep, int driverMemberCount) + { + var coordinator = CreateTestProbe(); + var publish = CreateTestProbe(); + var mux = CreateTestProbe(); + var vtHost = CreateTestProbe(); + + var actor = Sys.ActorOf(DriverHostActor.Props( + db, TestNode, coordinator.Ref, + localRoles: new HashSet { "driver" }, + dependencyMux: mux.Ref, + opcUaPublishActor: publish.Ref, + virtualTagEvaluator: NullVirtualTagEvaluator.Instance, + virtualTagHostOverride: vtHost.Ref, + driverMemberCountProvider: () => driverMemberCount)); + + actor.Tell(new DispatchDeployment(dep, RevA, CorrelationId.NewId())); + coordinator.ExpectMsg(Timeout).Outcome.ShouldBe(ApplyAckOutcome.Applied); + publish.ExpectMsg(Timeout); + return (actor, publish); + } + + private void SubscribeToAlerts(TestProbe probe) + { + DistributedPubSub.Get(Sys).Mediator.Tell( + new Subscribe(ScriptedAlarmHostActor.AlertsTopic, probe.Ref), probe.Ref); + probe.ExpectMsg(Timeout); + } + + private static DeploymentId SeedWriteTagDeployment(IDbContextFactory db) + { + var artifact = JsonSerializer.SerializeToUtf8Bytes(new + { + Namespaces = new[] { new { NamespaceId = "ns-eq", Kind = 0 } }, + DriverInstances = new[] + { + new + { + DriverInstanceRowId = Guid.NewGuid(), + DriverInstanceId = "drv-1", + Name = "drv-1", + DriverType = "Modbus", + Enabled = true, + DriverConfig = "{}", + NamespaceId = "ns-eq", + }, + }, + Tags = new[] + { + new + { + TagId = "tag-0", + EquipmentId = "eq-1", + DriverInstanceId = "drv-1", + Name = "speed", + FolderPath = (string?)null, + DataType = "Double", + TagConfig = JsonSerializer.Serialize(new { FullName = "40001" }), + }, + }, + }); + return SealArtifact(db, artifact); + } + + private static DeploymentId SeedAlarmTagDeployment(IDbContextFactory db) + { + var artifact = JsonSerializer.SerializeToUtf8Bytes(new + { + Namespaces = new[] { new { NamespaceId = "ns-eq", Kind = 0 } }, + DriverInstances = new[] { new { DriverInstanceId = "drv-1", NamespaceId = "ns-eq" } }, + Tags = new[] + { + new + { + TagId = "tag-0", + EquipmentId = "eq-1", + DriverInstanceId = "drv-1", + Name = "temp_hi", + FolderPath = (string?)null, + DataType = "Boolean", + TagConfig = JsonSerializer.Serialize(new + { + FullName = "Temp.HiHi", + alarm = new { alarmType = "OffNormalAlarm", severity = 700 }, + }), + }, + }, + }); + return SealArtifact(db, artifact); + } + + private static DeploymentId SealArtifact(IDbContextFactory db, byte[] artifact) + { + var id = DeploymentId.NewId(); + using var ctx = db.CreateDbContext(); + ctx.Deployments.Add(new Deployment + { + DeploymentId = id.Value, + RevisionHash = RevA.Value, + Status = DeploymentStatus.Sealed, + CreatedBy = "test", + SealedAtUtc = DateTime.UtcNow, + ArtifactBlob = artifact, + }); + ctx.SaveChanges(); + return id; + } + + private sealed class StubAlarmHandle : IAlarmSubscriptionHandle + { + public string DiagnosticId => "stub-alarm-sub"; + } + + private sealed class RecordingDriverFactory : IDriverFactory + { + private readonly string _supportedType; + private readonly RecordingDriver _driver = new(); + public RecordingDriverFactory(string supportedType) { _supportedType = supportedType; } + public IReadOnlyList Writes => _driver.Writes; + + public IDriver? TryCreate(string driverType, string driverInstanceId, string driverConfigJson) + { + if (!string.Equals(driverType, _supportedType, StringComparison.Ordinal)) return null; + _driver.Bind(driverInstanceId, driverType); + return _driver; + } + + public IReadOnlyCollection SupportedTypes => new[] { _supportedType }; + } + + private sealed class RecordingDriver : IDriver, IWritable + { + private readonly ConcurrentQueue _writes = new(); + public string DriverInstanceId { get; private set; } = string.Empty; + public string DriverType { get; private set; } = string.Empty; + public IReadOnlyList Writes => _writes.ToArray(); + public void Bind(string id, string type) { DriverInstanceId = id; DriverType = type; } + public Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken) => Task.CompletedTask; + public Task ReinitializeAsync(string driverConfigJson, CancellationToken cancellationToken) => Task.CompletedTask; + public Task ShutdownAsync(CancellationToken cancellationToken) => Task.CompletedTask; + public DriverHealth GetHealth() => new(DriverState.Healthy, DateTime.UtcNow, LastError: null); + public long GetMemoryFootprint() => 0; + public Task FlushOptionalCachesAsync(CancellationToken cancellationToken) => Task.CompletedTask; + public Task> WriteAsync( + IReadOnlyList writes, CancellationToken cancellationToken) + { + foreach (var w in writes) _writes.Enqueue(w); + return Task.FromResult>(writes.Select(_ => new WriteResult(0u)).ToArray()); + } + } + + private sealed class MeterRecorder : IDisposable + { + private readonly string _name; + private readonly MeterListener _listener; + private long _total; + private readonly List[]> _tagSets = new(); + private readonly object _gate = new(); + + public MeterRecorder(string instrumentName) + { + _name = instrumentName; + _listener = new MeterListener + { + InstrumentPublished = (instrument, listener) => + { + if (instrument.Meter.Name == OtOpcUaTelemetry.MeterName && instrument.Name == _name) + listener.EnableMeasurementEvents(instrument); + } + }; + _listener.SetMeasurementEventCallback((_, value, tags, _) => + { + lock (_gate) { _total += value; _tagSets.Add(tags.ToArray()); } + }); + _listener.Start(); + } + + public long Total { get { lock (_gate) return _total; } } + + public int WithTag(string key, string value) + { + lock (_gate) return _tagSets.Count(set => set.Any(t => t.Key == key && Equals(t.Value, value))); + } + + public void Dispose() => _listener.Dispose(); + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/PrimaryGatePolicyTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/PrimaryGatePolicyTests.cs new file mode 100644 index 00000000..07dac1d6 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/PrimaryGatePolicyTests.cs @@ -0,0 +1,38 @@ +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Redundancy; +using ZB.MOM.WW.OtOpcUa.Runtime.Drivers; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers; + +/// +/// archreview 03/S4 — the single Primary data-plane gate decision. A KNOWN role wins outright; an +/// UNKNOWN role (no snapshot yet, or the snapshot never mentioned this node) is resolved by cluster +/// membership: a single-driver cluster stays default-ALLOW (boot-window / single-node posture), a +/// multi-driver cluster is default-DENY (a real Primary peer exists — don't touch the shared field +/// device until the snapshot proves this node is Primary). +/// +public sealed class PrimaryGatePolicyTests +{ + [Theory] + // Primary → always service, regardless of member count. + [InlineData(RedundancyRole.Primary, 0, true)] + [InlineData(RedundancyRole.Primary, 1, true)] + [InlineData(RedundancyRole.Primary, 2, true)] + // Secondary / Detached → never service, regardless of member count. + [InlineData(RedundancyRole.Secondary, 0, false)] + [InlineData(RedundancyRole.Secondary, 2, false)] + [InlineData(RedundancyRole.Detached, 0, false)] + [InlineData(RedundancyRole.Detached, 2, false)] + public void Known_role_wins_regardless_of_member_count(RedundancyRole role, int members, bool expected) + => PrimaryGatePolicy.ShouldServiceAsPrimary(role, members).ShouldBe(expected); + + [Theory] + // Unknown role: allow only when no driver peer exists (count <= 1). + [InlineData(0, true)] + [InlineData(1, true)] + [InlineData(2, false)] + [InlineData(3, false)] + public void Unknown_role_resolves_by_membership(int members, bool expected) + => PrimaryGatePolicy.ShouldServiceAsPrimary(null, members).ShouldBe(expected); +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorApplyFailureTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorApplyFailureTests.cs new file mode 100644 index 00000000..2d401aaf --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorApplyFailureTests.cs @@ -0,0 +1,170 @@ +using System.Diagnostics.Metrics; +using System.Text.Json; +using Akka.Actor; +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.Logging.Abstractions; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.Observability; +using ZB.MOM.WW.OtOpcUa.Commons.OpcUa; +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.OpcUaServer; +using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa; +using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.OpcUa; + +/// +/// archreview 01/S-1: when the applier reports a degraded apply (a rebuild that threw, or per-node +/// materialise failures), must surface it — increment the +/// otopcua.opcua.apply.failed meter and log at Error — instead of the optimistic Info line. The +/// happy path must stay Info-only (zero increments). +/// +public sealed class OpcUaPublishActorApplyFailureTests : RuntimeActorTestBase +{ + /// A rebuild whose sink throws increments otopcua.opcua.apply.failed (kind=rebuild). + [Fact] + public void Rebuild_when_sink_throws_increments_apply_failed_meter() + { + using var recorder = new MeterRecorder("otopcua.opcua.apply.failed"); + var db = NewInMemoryDbFactory(); + var sink = new ThrowOnRebuildSink(); + var applier = new AddressSpaceApplier(sink, NullLogger.Instance); + SeedEquipmentDeployment(db, "eq-1"); + + var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(sink: sink, dbFactory: db, applier: applier)); + actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId())); + + AwaitAssert(() => + { + recorder.Total.ShouldBeGreaterThanOrEqualTo(1); + recorder.WithTag("kind", "rebuild").ShouldBeGreaterThanOrEqualTo(1); + }, duration: TimeSpan.FromSeconds(2)); + } + + /// A clean rebuild does NOT increment the apply-failed meter (Info-only happy path). + [Fact] + public void Rebuild_happy_path_does_not_increment_apply_failed_meter() + { + using var recorder = new MeterRecorder("otopcua.opcua.apply.failed"); + var db = NewInMemoryDbFactory(); + var sink = new NoopSink(); + var applier = new AddressSpaceApplier(sink, NullLogger.Instance); + SeedEquipmentDeployment(db, "eq-1"); + + var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(sink: sink, dbFactory: db, applier: applier)); + actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId())); + + // Give the actor time to process, then confirm the failed meter never fired. + Thread.Sleep(400); + recorder.Total.ShouldBe(0); + } + + private void AwaitAssert(Action assertion, TimeSpan duration) + { + var deadline = DateTime.UtcNow + duration; + Exception? last = null; + while (DateTime.UtcNow < deadline) + { + try { assertion(); return; } + catch (Exception ex) { last = ex; Thread.Sleep(25); } + } + if (last is not null) throw last; + } + + private static void SeedEquipmentDeployment(IDbContextFactory dbFactory, params string[] equipmentIds) + { + var artifact = JsonSerializer.SerializeToUtf8Bytes(new + { + Equipment = equipmentIds.Select(id => new + { + EquipmentId = id, + MachineCode = id.ToUpperInvariant(), + UnsLineId = "line-1", + Name = id, + }).ToArray(), + DriverInstances = Array.Empty(), + ScriptedAlarms = Array.Empty(), + }); + + using var ctx = dbFactory.CreateDbContext(); + ctx.Deployments.Add(new Deployment + { + DeploymentId = Guid.NewGuid(), + RevisionHash = new string('a', 64), + Status = DeploymentStatus.Sealed, + CreatedBy = "test", + SealedAtUtc = DateTime.UtcNow, + ArtifactBlob = artifact, + }); + ctx.SaveChanges(); + } + + /// A sink whose RebuildAddressSpace throws (drives the applier's SafeRebuild catch → RebuildFailed). + private sealed class ThrowOnRebuildSink : IOpcUaAddressSpaceSink + { + public void WriteValue(string nodeId, object? value, OpcUaQuality quality, DateTime sourceTimestampUtc) { } + public void WriteAlarmCondition(string alarmNodeId, AlarmConditionSnapshot state, DateTime occurredUtc) { } + public void MaterialiseAlarmCondition(string alarmNodeId, string equipmentNodeId, string displayName, string alarmType, int severity, bool isNative = false) { } + public void EnsureFolder(string folderNodeId, string? parentNodeId, string displayName) { } + public void EnsureVariable(string variableNodeId, string? parentFolderNodeId, string displayName, string dataType, bool writable, string? historianTagname = null, bool isArray = false, uint? arrayLength = null) { } + public void RebuildAddressSpace() => throw new InvalidOperationException("simulated rebuild fault"); + public void RaiseNodesAddedModelChange(string affectedNodeId) { } + } + + /// A no-op sink — a clean apply (no failures). + private sealed class NoopSink : IOpcUaAddressSpaceSink + { + public void WriteValue(string nodeId, object? value, OpcUaQuality quality, DateTime sourceTimestampUtc) { } + public void WriteAlarmCondition(string alarmNodeId, AlarmConditionSnapshot state, DateTime occurredUtc) { } + public void MaterialiseAlarmCondition(string alarmNodeId, string equipmentNodeId, string displayName, string alarmType, int severity, bool isNative = false) { } + public void EnsureFolder(string folderNodeId, string? parentNodeId, string displayName) { } + public void EnsureVariable(string variableNodeId, string? parentFolderNodeId, string displayName, string dataType, bool writable, string? historianTagname = null, bool isArray = false, uint? arrayLength = null) { } + public void RebuildAddressSpace() { } + public void RaiseNodesAddedModelChange(string affectedNodeId) { } + } + + /// Listens to a single instrument by name on the central meter and tallies value + tags. + private sealed class MeterRecorder : IDisposable + { + private readonly string _name; + private readonly MeterListener _listener; + private long _total; + private readonly List[]> _tagSets = new(); + private readonly object _gate = new(); + + public MeterRecorder(string instrumentName) + { + _name = instrumentName; + _listener = new MeterListener + { + InstrumentPublished = (instrument, listener) => + { + if (instrument.Meter.Name == OtOpcUaTelemetry.MeterName && instrument.Name == _name) + listener.EnableMeasurementEvents(instrument); + } + }; + _listener.SetMeasurementEventCallback((_, value, tags, _) => + { + lock (_gate) + { + _total += value; + _tagSets.Add(tags.ToArray()); + } + }); + _listener.Start(); + } + + public long Total { get { lock (_gate) return _total; } } + + public int WithTag(string key, string value) + { + lock (_gate) return _tagSets.Count(set => set.Any(t => t.Key == key && Equals(t.Value, value))); + } + + public void Dispose() => _listener.Dispose(); + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/ScriptedAlarms/ScriptedAlarmHostActorTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/ScriptedAlarms/ScriptedAlarmHostActorTests.cs index 94e7a78c..2bd182bd 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/ScriptedAlarms/ScriptedAlarmHostActorTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/ScriptedAlarms/ScriptedAlarmHostActorTests.cs @@ -519,6 +519,58 @@ public sealed class ScriptedAlarmHostActorTests : RuntimeActorTestBase evt.TransitionKind.ShouldBe("Activated"); } + /// Boot-window multi-driver suppression (archreview 03/S4): role unknown + a real driver peer + /// (member count 2) ⇒ the alerts publish is DENIED (default-deny closes the dual-primary window that would + /// historize duplicate AVEVA rows), while the OPC UA node write stays ungated. + [Fact] + public void Unknown_role_multi_driver_suppresses_alerts_but_still_writes_opcua() + { + var publish = CreateTestProbe(); + var mux = CreateTestProbe(); + var alerts = CreateTestProbe(); + SubscribeToAlerts(alerts); + + var (host, _) = SpawnWithMemberCount(publish, mux, LocalNode, driverMemberCount: 2); + host.Tell(new ScriptedAlarmHostActor.ApplyScriptedAlarms(new[] { Plan(severity: 800) })); + mux.ExpectMsg(Timeout); + + // No snapshot sent ⇒ role unknown; count 2 ⇒ default-DENY. + host.Tell(new VirtualTagActor.DependencyValueChanged("M.T", 99, DateTime.UtcNow)); + + publish.FishForMessage(m => m.State.Active, Timeout); + alerts.ExpectNoMsg(TimeSpan.FromMilliseconds(500)); + } + + /// Boot-window single-driver emit (archreview 03/S4): role unknown + no driver peer (member + /// count 1) ⇒ the alerts publish proceeds (single-node posture preserved). + [Fact] + public void Unknown_role_single_driver_publishes_alerts() + { + var publish = CreateTestProbe(); + var mux = CreateTestProbe(); + var alerts = CreateTestProbe(); + SubscribeToAlerts(alerts); + + var (host, _) = SpawnWithMemberCount(publish, mux, LocalNode, driverMemberCount: 1); + host.Tell(new ScriptedAlarmHostActor.ApplyScriptedAlarms(new[] { Plan(severity: 800) })); + mux.ExpectMsg(Timeout); + + host.Tell(new VirtualTagActor.DependencyValueChanged("M.T", 99, DateTime.UtcNow)); + + publish.FishForMessage(m => m.State.Active, Timeout); + alerts.ExpectMsg(Timeout).AlarmId.ShouldBe("alm-1"); + } + + private (IActorRef Host, DependencyMuxTagUpstreamSource Upstream) SpawnWithMemberCount( + TestProbe publish, TestProbe mux, NodeId localNode, int driverMemberCount) + { + var upstream = new DependencyMuxTagUpstreamSource(); + var engine = BuildEngine(upstream); + var host = Sys.ActorOf(ScriptedAlarmHostActor.Props( + publish.Ref, mux.Ref, upstream, engine, localNode, driverMemberCountProvider: () => driverMemberCount)); + return (host, upstream); + } + /// Inbound command ungated by role (T1): the alerts-publish gate must NOT affect inbound /// command processing. Under a Secondary role, an AlarmCommand("Acknowledge") for an owned, active /// alarm still drives the engine — observed via the resulting AlarmStateUpdate(Acknowledged=true)