From 712c5117eeed8cfd9b9e58152b5a7187bae27c2a Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Fri, 13 Nov 2026 03:06:45 -0500 Subject: [PATCH] feat(dcl): ServiceLevel-aware OPC UA failover for primary/backup pairs (#138) Prerequisite for OtOpcUa Warm redundancy (lmxopcua#717): a Warm standby keeps a healthy session but serves BadOutOfService at a standby ServiceLevel (90/50, boot 1), so a connection that lands there sat on Bad quality forever. - OPC UA connections with a backup poll Server_ServiceLevel (i=2267) on the connected endpoint (IServiceLevelReadable, implemented by OpcUaDataConnection; never raises Disconnected). A level < 200 sustained for the degraded window (default 10 s), or the first sample after any (re)connect, triggers a probe of the other endpoint (short-lived session: connect, read, dispose). Switch only when it reports a Healthy (>= 200) higher level, and not within the minimum dwell (default 60 s) of the previous ServiceLevel switch. Both degraded, other unreachable or level unknown: stay put. 250/240 (today's OtOpcUa) never starts a streak. - The switch reuses the failover path: SwitchToOtherEndpoint() now shared by all three triggers, then BecomeReconnecting(ReconnectCause.ServiceLevelFailover) - zero delay, no unstable accounting, ReSubscribeAll on the new endpoint. - Config: OpcUaEndpointConfig.ServiceLevelFailover {enabled, degradedWindowSeconds, minDwellSeconds}, read from the PRIMARY endpoint's JSON only; null = defaults (enabled, 10, 60). Inside the encrypted protocol JSON, so no schema change or migration. Central UI: fields in the Backup card next to Failover Retry Count. Global: ScadaBridge:DataConnection:ServiceLevelPollInterval (2 s) and ServiceLevelProbeTimeout (15 s). - Tests: actor tests per trigger/non-trigger with positive controls, an unsolicited promotion RefreshStart/RefreshEnd test at client and actor level, adapter tests, and a real in-process OPC UA server pair end to end. All 15 mutants killed. - Docs: Component-DataConnectionLayer, components/DataConnectionLayer, Component-CLI. Claude-Session: https://claude.ai/code/session_01GT67ACmpMR7b9LGRPLijMi --- docs/components/DataConnectionLayer.md | 10 +- docs/requirements/Component-CLI.md | 2 + .../Component-DataConnectionLayer.md | 36 +- .../Pages/Design/DataConnectionForm.razor | 62 +++ .../Protocol/IServiceLevelReadable.cs | 22 + .../OpcUaEndpointConfigSerializer.cs | 27 ++ .../DataConnections/OpcUaEndpointConfig.cs | 8 + .../OpcUaServiceLevelFailoverConfig.cs | 46 +++ .../OpcUaEndpointConfigValidator.cs | 8 + .../DataConnectionActor.ServiceLevel.cs | 320 +++++++++++++++ .../Actors/DataConnectionActor.cs | 99 +++-- .../Adapters/OpcUaDataConnection.cs | 38 +- .../DataConnectionOptions.cs | 15 + .../DataConnectionOptionsValidator.cs | 6 + .../DataConnectionFormTests.cs | 70 ++++ .../OpcUaEndpointConfigSerializerTests.cs | 54 +++ .../OpcUaEndpointConfigValidatorTests.cs | 26 ++ .../DataConnectionPromotionRefreshTests.cs | 110 +++++ ...DataConnectionServiceLevelFailoverTests.cs | 377 ++++++++++++++++++ .../Actors/FakeServiceLevelConnection.cs | 119 ++++++ .../Adapters/OpcUaServiceLevelTests.cs | 211 ++++++++++ .../RealOpcUaPromotionRefreshTests.cs | 112 ++++++ .../Adapters/SecureOpcUaTestServer.cs | 18 +- 23 files changed, 1760 insertions(+), 36 deletions(-) create mode 100644 src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Protocol/IServiceLevelReadable.cs create mode 100644 src/ZB.MOM.WW.ScadaBridge.Commons/Types/DataConnections/OpcUaServiceLevelFailoverConfig.cs create mode 100644 src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionActor.ServiceLevel.cs create mode 100644 tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Actors/DataConnectionPromotionRefreshTests.cs create mode 100644 tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Actors/DataConnectionServiceLevelFailoverTests.cs create mode 100644 tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Actors/FakeServiceLevelConnection.cs create mode 100644 tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Adapters/OpcUaServiceLevelTests.cs create mode 100644 tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Adapters/RealOpcUaPromotionRefreshTests.cs diff --git a/docs/components/DataConnectionLayer.md b/docs/components/DataConnectionLayer.md index 58a20aca..75982926 100644 --- a/docs/components/DataConnectionLayer.md +++ b/docs/components/DataConnectionLayer.md @@ -119,6 +119,8 @@ if (result.Success) When `BackupConnectionDetails` is provided, the actor tracks consecutive failures and unstable disconnects. After `FailoverRetryCount` consecutive failures (or unstable connections shorter than `StableConnectionThreshold`), the actor disposes the current adapter, creates a fresh one with the other endpoint's config via `DataConnectionFactory.Create`, and re-arms the connect timer. Failover is round-robin: primary → backup → primary. There is no auto-failback; the connection stays on whichever endpoint is currently working. +**ServiceLevel-aware failover (OPC UA, Gitea #138).** A third trigger, for redundant OPC UA pairs whose standby keeps a healthy session but serves Bad data (OtOpcUa Warm, lmxopcua#717). It lives in `Actors/DataConnectionActor.ServiceLevel.cs`. While Connected, and only when the adapter implements `IServiceLevelReadable`, the connection has a backup and the primary config's `ServiceLevelFailover.Enabled` is true, a periodic timer reads `Server_ServiceLevel` (`i=2267`). A level below 200 starts a degraded streak; a Healthy or unreadable sample ends it. After the degraded window (default 10 s) — or at once on the first sample after a (re)connect — the actor probes the other endpoint with a throwaway adapter from `DataConnectionFactory.Create` (connect, read, dispose; bounded by `ServiceLevelProbeTimeout`). It switches only if that endpoint reports ≥ 200 and higher, and not within the minimum dwell (default 60 s) of the previous ServiceLevel switch; otherwise it stays and re-arms the window. The switch is `SwitchToOtherEndpoint()` — the helper the two connectivity triggers now share — followed by `BecomeReconnecting(ReconnectCause.ServiceLevelFailover)` (zero delay, no unstable accounting). Poll and probe results carry the adapter generation, so a result from a session that has ended is dropped. Today's OtOpcUa (250/240) never starts a streak. Full rule set: `docs/requirements/Component-DataConnectionLayer.md` → *ServiceLevel-aware failover*. + ### Write path Writes are fire-and-forget from the script's view only in the sense that there is no store-and-forward. The `DataConnectionActor` pipes `_adapter.WriteAsync` back to the original sender: @@ -206,7 +208,7 @@ private void SwapAdapter(IDictionary newConfig) } ``` -It then re-enters the reconnect cycle immediately (`BecomeReconnecting(afterReconfigure: true)` — zero delay, and the unstable-disconnect/failover accounting suppressed, since the link was dropped on purpose), and `ReSubscribeAll` / `ReSubscribeAllAlarms` re-issue everything from the untouched subscription registries. A reconfigure carrying the configuration already in force is a no-op. +It then re-enters the reconnect cycle immediately (`BecomeReconnecting(ReconnectCause.Reconfigured)` — zero delay, and the unstable-disconnect/failover accounting suppressed, since the link was dropped on purpose), and `ReSubscribeAll` / `ReSubscribeAllAlarms` re-issue everything from the untouched subscription registries. A reconfigure carrying the configuration already in force is a no-op. ### Subscribing and receiving values @@ -253,6 +255,8 @@ All settings under `DataConnectionLayer` in `appsettings.json` bind to `DataConn | `TagResolutionRetryInterval` | `00:00:10` | Retry interval for tag paths that could not be resolved on the device. | | `WriteTimeout` | `00:00:30` | Timeout applied to each `WriteAsync` call. A hung write times out and returns an error to the calling script. | | `StableConnectionThreshold` | `00:01:00` | A connection that drops before this duration is counted as an unstable disconnect toward `FailoverRetryCount`. | +| `ServiceLevelPollInterval` | `00:00:02` | Read interval for the connected OPC UA endpoint's ServiceLevel when ServiceLevel-aware failover is in force (#138). | +| `ServiceLevelProbeTimeout` | `00:00:15` | Budget for one ServiceLevel read and for probing the other endpoint (connect + read). | ### OPC UA global settings (`OpcUa` section) @@ -286,6 +290,9 @@ Stored in the `DataConnection.PrimaryConfiguration` JSON dict, parsed by `OpcUaE | `QueueSize` | `10` | Per-item notification buffer size. | | `SecurityMode` | `None` | `None`, `Sign`, or `SignAndEncrypt`. | | `AutoAcceptUntrustedCerts` | `true` | Accept untrusted server certificates. | +| `ServiceLevelFailover.Enabled` | `true` | ServiceLevel-aware failover (#138). Primary endpoint only; needs a backup. JSON `serviceLevelFailover` object; absent = defaults. | +| `ServiceLevelFailover.DegradedWindowSeconds` | `10` | Seconds below ServiceLevel 200 before the other endpoint is probed. | +| `ServiceLevelFailover.MinDwellSeconds` | `60` | Minimum seconds between ServiceLevel-driven switches. | ### Per-connection settings (MxGateway) @@ -310,6 +317,7 @@ Carried on `CreateConnectionCommand` from the deployment artifact. |-------|---------|-------------| | `BackupConnectionDetails` | `null` | If absent, the connection retries its single endpoint indefinitely. | | `FailoverRetryCount` | `3` | Consecutive failures (or unstable disconnects) before switching endpoints. | +| `ServiceLevelFailover` (OPC UA, inside the primary's configuration JSON) | enabled, 10 s, 60 s | ServiceLevel-aware failover (#138); see the OPC UA per-connection table. Lives in the protocol JSON rather than a column, so it needs no schema change and is encrypted at rest with the rest of the configuration. | ## Dependencies & Interactions diff --git a/docs/requirements/Component-CLI.md b/docs/requirements/Component-CLI.md index d3471754..424057e9 100644 --- a/docs/requirements/Component-CLI.md +++ b/docs/requirements/Component-CLI.md @@ -190,6 +190,8 @@ scadabridge data-connection delete --id scadabridge data-connection move --id --site-id ``` +For an OPC UA connection with a backup, ServiceLevel-aware failover (Gitea #138) is set inside `--primary-config` as `"serviceLevelFailover": {"enabled": true, "degradedWindowSeconds": 10, "minDwellSeconds": 60}`; omitted means those defaults. See `Component-DataConnectionLayer.md` → *ServiceLevel-aware failover*. + #### Site-routed OPC UA operations (browse / search / verify-endpoint / certs) CLI parity for the site-routed data-connection operations the ManagementActor dispatches diff --git a/docs/requirements/Component-DataConnectionLayer.md b/docs/requirements/Component-DataConnectionLayer.md index e2bf89ad..0c6b81a4 100644 --- a/docs/requirements/Component-DataConnectionLayer.md +++ b/docs/requirements/Component-DataConnectionLayer.md @@ -137,7 +137,7 @@ flowchart TD ``` - **Round-robin**: primary → backup → primary → backup. No preferred endpoint after first failover — the connection stays on whichever endpoint is working. -- **No auto-failback**: The connection remains on the active endpoint until it fails. +- **No auto-failback**: The connection remains on the active endpoint until it fails — or, for OPC UA, until its ServiceLevel says it is no longer the server to be on (see *ServiceLevel-aware failover* below). - **Single-endpoint connections** (no backup): Retry indefinitely on the same endpoint, preserving existing behavior. - **Adapter lifecycle on failover**: The actor disposes the current `IDataConnection` adapter and creates a fresh one via `DataConnectionFactory.Create()` with the other endpoint's configuration. Clean slate — no stale state. @@ -152,6 +152,34 @@ flowchart TD See [`2026-03-22-primary-backup-data-connections-design.md`](../plans/2026-03-22-primary-backup-data-connections-design.md) for the full design. +### ServiceLevel-aware failover (OPC UA, Gitea #138) + +The failover above is driven only by the link misbehaving. A redundant OPC UA server pair in **Warm** mode (OtOpcUa lmxopcua#717, design `OtOpcUa/docs/plans/2026-11-11-warm-redundancy-717-design.md` §4.3/§5) breaks that assumption: the standby keeps a perfectly healthy session but serves `BadOutOfService` data and advertises a standby ServiceLevel, so a connection that lands on it would sit on Bad quality forever. OPC UA connections with a backup therefore also watch `Server_ServiceLevel` (`i=2267`). + +| Band (OPC UA Part 4 §6.6.2.4.2) | Levels | OtOpcUa | +|---|---|---| +| Healthy | 200–255 | Active: 250 Primary / 240 Secondary (today and under Warm), 200 Active-stale | +| Degraded | 100–199, and anything below 200 here | Active critically degraded 100; Warm standby 90 (healthy) / 50 (degraded) | +| NoData / Maintenance | 1 / 0 | Warm booting 1; down / endpoint failed 0 | + +**Rule.** While Connected, the actor reads the connected endpoint's ServiceLevel every `ServiceLevelPollInterval` (2 s). A level **below 200** starts a *degraded streak*; any level of 200 or more, or an unreadable level (Bad status, no such node, read failure), ends it. When the streak has lasted the connection's **degraded window** (default 10 s), the actor probes the **other** endpoint — a short-lived session built from that endpoint's configuration: connect, read `i=2267`, close, bounded by `ServiceLevelProbeTimeout` (15 s). It fails over only when the other endpoint answers with a **Healthy level (≥ 200) higher than the connected one**. Otherwise (other endpoint unreachable, or also below 200) it **stays put** and the window re-arms, so the other endpoint is probed at most once per window. + +- **Connect-time preference.** The first sample after every (re)connect is evaluated at once, without the window: a connection that comes up on a Degraded endpoint while the other is Healthy moves straight over. Because the trigger needs the connected level below 200, a connection to a 240 Secondary is never moved to a 250 Primary. +- **Anti-flap.** A ServiceLevel switch is not repeated within the connection's **minimum dwell** (default 60 s), connect-time preference included. Together with the "Healthy and higher" requirement and the unbroken-streak rule (one Healthy sample resets it) this stops a pair that flips roles from bouncing the connection. +- **Same switch as the connectivity failover.** The switch flips the active endpoint and swaps the adapter through the same helper as the two connection-driven triggers, resets both failure counters, then re-enters the reconnect cycle immediately with the unstable-disconnect accounting suppressed (it is a deliberate swap, like an in-place reconfigure). `ReSubscribeAll` / `ReSubscribeAllAlarms` then re-issue every subscription on the new endpoint; tags go Bad for the swap exactly as on any failover. Site event log: `Failover from Primary to Backup (ServiceLevel)` (Warning) with the two levels. +- **Scope.** Only adapters implementing `IServiceLevelReadable` (the OPC UA adapter) are watched, and only when the connection has a backup and the setting is enabled. MxGateway connections, single-endpoint connections and disabled connections never read a ServiceLevel. The ServiceLevel read never raises the adapter's `Disconnected` event: a dead session stays the keep-alive's business. +- **Behaviour against today's servers.** A pair that only ever reports Healthy levels (OtOpcUa 250/240 without Warm, or a server that always reports 255) never starts a streak: no probe, no switch — the default-on setting changes nothing there. It does act on a non-Warm OtOpcUa node that has dropped to 100/110 or 0/10 while its peer is ≥ 200, which is the desired move. + +**Configuration** (per connection, on the **primary** endpoint's OPC UA JSON — the backup's copy is ignored; edited in the Central UI's *Backup endpoint* card next to *Failover Retry Count*; via the CLI it is part of `--primary-config`): + +```json +"serviceLevelFailover": { "enabled": true, "degradedWindowSeconds": 10, "minDwellSeconds": 60 } +``` + +Absent or `null` means the defaults (enabled, 10 s, 60 s), so stored connections need no migration; the object travels inside the protocol configuration, which is encrypted at rest with the rest of it. Validation: window 1–3600 s, dwell 0–86400 s. + +**Alarms across a promotion.** On Standby → Active the promoted OtOpcUa node runs a ConditionRefresh on every live event item on its own (design §4.5). The existing items therefore receive an **unsolicited** RefreshStart, the retained conditions, RefreshEnd. `RealOpcUaClient` frames it like the refresh it triggers on subscribe — `SnapshotStart` / `Snapshot`… / `SnapshotComplete` under a fresh epoch, live derivation resuming after RefreshEnd — and the actor's epoch-scoped snapshot cache replaces the source's state with it, dropping conditions the promoted node no longer retains. Covered by `RealOpcUaPromotionRefreshTests` and `DataConnectionPromotionRefreshTests`. + ### In-place reconfiguration (register row 97) A `ReconfigureConnectionCommand` replaces a **live** connection's configuration without stopping the connection actor. It is the same adapter swap the failover path above performs — the two share one `SwapAdapter` helper — so the diagram applies verbatim from *dispose adapter* onwards: @@ -186,6 +214,9 @@ All settings are parsed from the data connection's configuration JSON dictionari | `MaxMonitoredItemsPerSubscription` | int | `5000` | Monitored-item budget per OPC UA subscription. Above it the adapter **shards** onto an additional subscription on the same session (37,500 tags → 8 shards). Per endpoint, because item-count ceilings are a property of the target server. | | `SecurityMode` | string | `None` | Preferred endpoint security: `None`, `Sign`, or `SignAndEncrypt` | | `AutoAcceptUntrustedCerts` | bool | `true` | Accept untrusted server certificates | +| `ServiceLevelFailover.Enabled` | bool | `true` | ServiceLevel-aware failover (#138). Read from the **primary** endpoint only; needs a backup. JSON: `serviceLevelFailover.enabled` | +| `ServiceLevelFailover.DegradedWindowSeconds` | int | `10` | How long the connected endpoint's ServiceLevel must stay below 200 before the other endpoint is probed (1–3600) | +| `ServiceLevelFailover.MinDwellSeconds` | int | `60` | Minimum time between ServiceLevel-driven switches (0–86400) | ### OPC UA application identity (deployment-wide, `ScadaBridge:OpcUa`) @@ -242,6 +273,8 @@ These are configured via `DataConnectionOptions` in `appsettings.json`, not per- | `SubscribeBatchDelay` | 50ms | Delay BETWEEN reconnect re-subscribe chunks. Not applied on the instance-driven subscribe path — the Deployment Manager already staggers instance startup there. | | `QualityFlushInterval` | 1s | Coalescing window for pushing tag-quality counters to the health collector | | `MxSupervisoryAdviseParallelism` | 16 | Max in-flight supervisory advise commands on the MxGateway bulk-subscribe path when the endpoint has no write-user context | +| `ServiceLevelPollInterval` | 2s | How often the connected OPC UA endpoint's `Server_ServiceLevel` is read when ServiceLevel-aware failover is in force (#138) | +| `ServiceLevelProbeTimeout` | 15s | Budget for one ServiceLevel read, and for probing the other endpoint (connect + read); no answer in time = unreachable, no switch | | `MxAlarmStreamReopenDebounce` | 200ms | Window over which MxGateway alarm subscribes are coalesced into ONE reopen of the shared alarm stream (0–5 s; `0` = reopen per subscribe, the pre-2026-10-21 behaviour). Every new source still receives a fresh standing-state snapshot. | ## Subscription Management @@ -347,6 +380,7 @@ The registered `SourceReference` is a **NodeId** for OPC UA (the picker, CSV and - **State gating**: `SubscribeAlarmsRequest` is handled only in the **Connected** state; requests arriving while **Connecting**/**Reconnecting** are coalesced to the latest request per subscriber actor/source and processed on entering Connected. Unsubscribes and subscriber termination are handled in every state. - **Capability check**: if `_adapter is not IAlarmSubscribableConnection`, the actor replies `SubscribeAlarmsResponse(Success = false, ...)`. - **Reconnect handling**: on entering **Reconnecting**, the actor pushes a `NativeAlarmSourceUnavailable` to every alarm subscriber (consumers mark mirrored alarms uncertain rather than clearing them). On successful reconnection it re-subscribes the feed; the adapter re-emits a snapshot, reconciling state. +- **Server-initiated refresh**: a RefreshStart/RefreshEnd pair the server sends on its own on an existing feed (OtOpcUa's promotion refresh under Warm redundancy, #138) takes the same snapshot path — forwarded in order under its own epoch, and it replaces the cached state for the source. - **Shared condition filter (last-subscriber-wins)**: because the feed is opened once per source, it carries a single condition filter. A second subscriber that registers a *different* filter for the same source overwrites it (last writer wins) and the actor logs a warning — co-subscribers to one source are expected to agree on the filter. - **Routing index**: subscribed sources are bucketed by the **first path segment** of their reference (everything before the first `.`), and a transition is matched only against the bucket for its own first segment, plus a residue list of sources that carry no separator at all (a prefix shorter than one segment, which can match other buckets). This is sound because a source reference containing a separator can only prefix a reference sharing its entire first segment. The per-source `StartsWith` test and the condition-type gate inside the bucket are unchanged, so routing decisions are identical to the previous linear scan over every subscribed source; snapshot boundaries bypass the condition filter, but OPC UA boundaries are sent only to the exact source binding. Connection-wide Mx boundaries broadcast to all subscribers while preserving their refresh epoch. - **Gateway union filter (MxGateway)**: `StreamAlarms` carries a single `alarm_filter_prefix`, so the adapter opens the stream on the **longest common prefix** of the currently subscribed source references and restarts it for each new source, including one covered by the existing prefix, so every newly bound source receives standing state. Restarts are **coalesced** (review 2026-10-21): subscribes arriving within `MxAlarmStreamReopenDebounce` (200 ms) of the first share ONE restart on the union prefix, so a staggered deploy batch of N alarm-bound instances costs one close/reopen and one connection-wide snapshot replay instead of N (each of which was broadcast to every already-subscribed source — O(N·M) boundary messages). A reconnect, teardown or last unsubscribe invalidates a pending restart. Each transport refresh has connection-wide start/end markers and a distinct epoch. An unsubscribe never restarts the stream — the prefix it leaves behind is at worst too broad, which may increase bandwidth. Correctness also depends on the epoch-scoped snapshot cache merging concurrent live updates and rejecting stale snapshot ends (DC-5); a broad filter alone provides no such guarantee. The prefix is a bandwidth optimisation only; the actor's per-source + condition-type gate remains authoritative, mirroring the OPC UA server-side WhereClause stance. diff --git a/src/ZB.MOM.WW.ScadaBridge.CentralUI/Components/Pages/Design/DataConnectionForm.razor b/src/ZB.MOM.WW.ScadaBridge.CentralUI/Components/Pages/Design/DataConnectionForm.razor index e9fb772e..1e9879f4 100644 --- a/src/ZB.MOM.WW.ScadaBridge.CentralUI/Components/Pages/Design/DataConnectionForm.razor +++ b/src/ZB.MOM.WW.ScadaBridge.CentralUI/Components/Pages/Design/DataConnectionForm.razor @@ -165,6 +165,40 @@ min="1" max="20" @bind="_formFailoverRetryCount" />
Retries before failing over to backup endpoint.
+ @if (_protocol != "MxGateway") + { +
+
+ + +
+
+ Fail over when the connected endpoint's ServiceLevel stays below 200 (Degraded) + for the window below and the other endpoint reports a Healthy (200+), higher level. + Needed for an OtOpcUa pair in Warm redundancy; a 250/240 pair never triggers it. +
+ @if (_formSlEnabled) + { +
+
+ + + @RenderPrimaryError("ServiceLevelFailover.DegradedWindowSeconds") +
+
+ + + @RenderPrimaryError("ServiceLevelFailover.MinDwellSeconds") +
+
+ } +
+ }
@@ -207,6 +241,10 @@ private bool _backupIsLegacy; private bool _showBackup; private int _formFailoverRetryCount = 3; + // ScadaBridge#138: connection-level ServiceLevel failover, stored on the PRIMARY OPC UA endpoint. + private bool _formSlEnabled = true; + private int _formSlWindowSeconds = OpcUaServiceLevelFailoverConfig.DefaultDegradedWindowSeconds; + private int _formSlDwellSeconds = OpcUaServiceLevelFailoverConfig.DefaultMinDwellSeconds; private ValidationResult? _primaryErrors; private ValidationResult? _backupErrors; private string? _formError; @@ -276,6 +314,10 @@ { (_primaryConfig, _primaryIsLegacy) = OpcUaEndpointConfigSerializer.Deserialize(conn.PrimaryConfiguration); + var sl = _primaryConfig.ServiceLevelFailover ?? new OpcUaServiceLevelFailoverConfig(); + _formSlEnabled = sl.Enabled; + _formSlWindowSeconds = sl.DegradedWindowSeconds; + _formSlDwellSeconds = sl.MinDwellSeconds; if (!string.IsNullOrWhiteSpace(conn.BackupConfiguration)) { (_backupConfig, _backupIsLegacy) = @@ -313,6 +355,15 @@ } else { + // Only meaningful with a backup; without one the setting is not stored at all. + _primaryConfig.ServiceLevelFailover = _showBackup + ? new OpcUaServiceLevelFailoverConfig + { + Enabled = _formSlEnabled, + DegradedWindowSeconds = _formSlWindowSeconds, + MinDwellSeconds = _formSlDwellSeconds, + } + : null; _primaryErrors = OpcUaEndpointConfigValidator.Validate(_primaryConfig, "Primary."); _backupErrors = _showBackup ? OpcUaEndpointConfigValidator.Validate(_backupConfig, "Backup.") @@ -350,6 +401,17 @@ _backupMx = new MxGatewayEndpointConfig(); _backupIsLegacy = false; _formFailoverRetryCount = 3; + _formSlEnabled = true; + _formSlWindowSeconds = OpcUaServiceLevelFailoverConfig.DefaultDegradedWindowSeconds; + _formSlDwellSeconds = OpcUaServiceLevelFailoverConfig.DefaultMinDwellSeconds; + } + + private RenderFragment? RenderPrimaryError(string field) + { + var match = _primaryErrors?.Errors.FirstOrDefault(e => e.EntityName == "Primary." + field); + return match is null + ? null + : @
@match.Message
; } private void GoBack() => NavigationManager.NavigateTo("/design/connections"); diff --git a/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Protocol/IServiceLevelReadable.cs b/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Protocol/IServiceLevelReadable.cs new file mode 100644 index 00000000..9fd1d2ef --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Protocol/IServiceLevelReadable.cs @@ -0,0 +1,22 @@ +namespace ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Protocol; + +/// +/// Optional capability for an whose server publishes a redundancy +/// ServiceLevel (OPC UA Server_ServiceLevel, i=2267). The DataConnectionActor polls it +/// on the connected endpoint, and probes the other endpoint with it, to drive ServiceLevel-aware +/// failover (ScadaBridge#138). An adapter without the capability is never switched on ServiceLevel. +/// +public interface IServiceLevelReadable +{ + /// + /// Reads the server's current ServiceLevel. Never raises the connection's + /// event: a lost session is the reconnect machine's + /// business, not this read's. + /// + /// Bounds the read. + /// + /// The ServiceLevel (0–255), or null when it is unknown: not connected, a Bad status, a value + /// that is not a byte, or a failed read. + /// + Task ReadServiceLevelAsync(CancellationToken cancellationToken = default); +} diff --git a/src/ZB.MOM.WW.ScadaBridge.Commons/Serialization/OpcUaEndpointConfigSerializer.cs b/src/ZB.MOM.WW.ScadaBridge.Commons/Serialization/OpcUaEndpointConfigSerializer.cs index fea3f727..8128efa6 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Commons/Serialization/OpcUaEndpointConfigSerializer.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Commons/Serialization/OpcUaEndpointConfigSerializer.cs @@ -83,6 +83,15 @@ public readonly record struct OpcUaConfigParseResult /// public static class OpcUaEndpointConfigSerializer { + /// Flat-dict key for . + public const string ServiceLevelFailoverEnabledKey = "ServiceLevelFailover.Enabled"; + + /// Flat-dict key for . + public const string ServiceLevelFailoverWindowKey = "ServiceLevelFailover.DegradedWindowSeconds"; + + /// Flat-dict key for . + public const string ServiceLevelFailoverDwellKey = "ServiceLevelFailover.MinDwellSeconds"; + private static readonly JsonSerializerOptions JsonOpts = new() { PropertyNamingPolicy = JsonNamingPolicy.CamelCase, @@ -216,6 +225,12 @@ public static class OpcUaEndpointConfigSerializer dict["Deadband.Type"] = db.Type.ToString(); dict["Deadband.Value"] = db.Value.ToString(System.Globalization.CultureInfo.InvariantCulture); } + if (config.ServiceLevelFailover is { } sl) + { + dict[ServiceLevelFailoverEnabledKey] = sl.Enabled.ToString(); + dict[ServiceLevelFailoverWindowKey] = sl.DegradedWindowSeconds.ToString(System.Globalization.CultureInfo.InvariantCulture); + dict[ServiceLevelFailoverDwellKey] = sl.MinDwellSeconds.ToString(System.Globalization.CultureInfo.InvariantCulture); + } return dict; } @@ -291,6 +306,18 @@ public static class OpcUaEndpointConfigSerializer c.Deadband = db; } + if (dict.ContainsKey(ServiceLevelFailoverEnabledKey) + || dict.ContainsKey(ServiceLevelFailoverWindowKey) + || dict.ContainsKey(ServiceLevelFailoverDwellKey)) + { + var sl = new OpcUaServiceLevelFailoverConfig(); + if (dict.TryGetValue(ServiceLevelFailoverEnabledKey, out var slEn) && bool.TryParse(slEn, out var slEnabled)) + sl.Enabled = slEnabled; + TryAssignInt(dict, ServiceLevelFailoverWindowKey, v => sl.DegradedWindowSeconds = v); + TryAssignInt(dict, ServiceLevelFailoverDwellKey, v => sl.MinDwellSeconds = v); + c.ServiceLevelFailover = sl; + } + return c; } diff --git a/src/ZB.MOM.WW.ScadaBridge.Commons/Types/DataConnections/OpcUaEndpointConfig.cs b/src/ZB.MOM.WW.ScadaBridge.Commons/Types/DataConnections/OpcUaEndpointConfig.cs index 9627e76b..2ada84d5 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Commons/Types/DataConnections/OpcUaEndpointConfig.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Commons/Types/DataConnections/OpcUaEndpointConfig.cs @@ -96,4 +96,12 @@ public sealed class OpcUaEndpointConfig /// Heartbeat configuration for connection monitoring. /// public OpcUaHeartbeatConfig? Heartbeat { get; set; } + + // Redundancy (optional; null = defaults, see OpcUaServiceLevelFailoverConfig) + /// + /// ServiceLevel-aware failover settings (ScadaBridge#138). Connection-level: only the primary + /// endpoint's value is read, and only when the connection has a backup. Null means the + /// defaults (enabled, 10 s window, 60 s dwell). + /// + public OpcUaServiceLevelFailoverConfig? ServiceLevelFailover { get; set; } } diff --git a/src/ZB.MOM.WW.ScadaBridge.Commons/Types/DataConnections/OpcUaServiceLevelFailoverConfig.cs b/src/ZB.MOM.WW.ScadaBridge.Commons/Types/DataConnections/OpcUaServiceLevelFailoverConfig.cs new file mode 100644 index 00000000..b538394c --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.Commons/Types/DataConnections/OpcUaServiceLevelFailoverConfig.cs @@ -0,0 +1,46 @@ +namespace ZB.MOM.WW.ScadaBridge.Commons.Types.DataConnections; + +/// +/// ServiceLevel-aware failover for an OPC UA connection with a primary AND a backup endpoint +/// (ScadaBridge#138; OtOpcUa Warm redundancy, lmxopcua#717). A connection-level setting: only the +/// primary endpoint's copy is read, and it has no effect on a connection without a backup. +/// +/// +/// The connected endpoint's Server_ServiceLevel (i=2267) is polled. When it stays in +/// the Degraded band (below 200, OPC UA Part 4 §6.6.2.4.2) for +/// and the other endpoint is reachable and reports a Healthy-band level (200 or more) that is also +/// higher, the connection fails over to it. A switch is not repeated within +/// . Both endpoints degraded, or the other one unreachable, means the +/// connection stays where it is. On connect, a Degraded endpoint is left immediately (no window) for +/// a Healthy, higher other endpoint, subject to the same dwell. +/// +/// +/// +/// A stored configuration without this object behaves as the defaults: enabled, 10 s window, +/// 60 s dwell. With today's OtOpcUa pair (250 Primary / 240 Secondary, both Healthy band) it never +/// triggers. +/// +/// +public sealed class OpcUaServiceLevelFailoverConfig +{ + /// Default for . + public const int DefaultDegradedWindowSeconds = 10; + + /// Default for . + public const int DefaultMinDwellSeconds = 60; + + /// Whether ServiceLevel can trigger a failover. Default true. + public bool Enabled { get; set; } = true; + + /// + /// How long, in seconds, the connected endpoint must report a Degraded-band ServiceLevel + /// without a break before the other endpoint is checked. Default 10. + /// + public int DegradedWindowSeconds { get; set; } = DefaultDegradedWindowSeconds; + + /// + /// Minimum time, in seconds, after a ServiceLevel-driven switch before another one may happen + /// (anti-flap). Default 60. + /// + public int MinDwellSeconds { get; set; } = DefaultMinDwellSeconds; +} diff --git a/src/ZB.MOM.WW.ScadaBridge.Commons/Validators/OpcUaEndpointConfigValidator.cs b/src/ZB.MOM.WW.ScadaBridge.Commons/Validators/OpcUaEndpointConfigValidator.cs index 97692fb2..1e7efbb6 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Commons/Validators/OpcUaEndpointConfigValidator.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Commons/Validators/OpcUaEndpointConfigValidator.cs @@ -74,6 +74,14 @@ public static class OpcUaEndpointConfigValidator if (config.Deadband is { } db && db.Value <= 0) errors.Add(Err("Deadband.Value", "Must be > 0.")); + if (config.ServiceLevelFailover is { } sl) + { + if (sl.DegradedWindowSeconds < 1 || sl.DegradedWindowSeconds > 3600) + errors.Add(Err("ServiceLevelFailover.DegradedWindowSeconds", "Must be between 1 and 3600.")); + if (sl.MinDwellSeconds < 0 || sl.MinDwellSeconds > 86400) + errors.Add(Err("ServiceLevelFailover.MinDwellSeconds", "Must be between 0 and 86400.")); + } + return errors.Count == 0 ? ValidationResult.Success() : ValidationResult.FromErrors(errors.ToArray()); diff --git a/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionActor.ServiceLevel.cs b/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionActor.ServiceLevel.cs new file mode 100644 index 00000000..bcff598c --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionActor.ServiceLevel.cs @@ -0,0 +1,320 @@ +using System.Globalization; +using Akka.Actor; +using Akka.Event; +using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Protocol; +using ZB.MOM.WW.ScadaBridge.Commons.Serialization; +using ZB.MOM.WW.ScadaBridge.Commons.Types.DataConnections; + +namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Actors; + +/// +/// ScadaBridge#138 — ServiceLevel-aware failover between a primary and a backup endpoint. +/// +/// +/// Why. Until now the connection failed over only when the link itself misbehaved +/// (FailoverRetryCount consecutive connect failures or unstable disconnects). Under OtOpcUa +/// Warm redundancy (lmxopcua#717) the standby keeps a perfectly healthy session but serves +/// BadOutOfService data at a standby ServiceLevel (90 healthy / 50 degraded / 1 booting), so a +/// connection that lands there would sit on Bad quality indefinitely. +/// +/// +/// +/// What. While Connected, the connected endpoint's Server_ServiceLevel is polled every +/// . A level below +/// (200, the bottom of the Part 4 Healthy band) starts a +/// degraded streak; any Healthy or unreadable sample ends it. Once the streak has lasted the +/// connection's degraded window — or immediately on the first sample after a (re)connect, which is +/// the "prefer the better endpoint on connect" rule — the other endpoint is probed with a +/// short-lived session. The connection switches only when that probe reads a Healthy level that is +/// also higher than the connected one, and never within the dwell time of the previous +/// ServiceLevel switch. Otherwise it stays put and the window re-arms. +/// +/// +/// +/// Unchanged behaviour. A level of 200 or more never starts a streak, so today's OtOpcUa pair +/// (250 Primary / 240 Secondary) never triggers anything; an adapter without +/// (MxGateway) and a connection without a backup are never +/// monitored at all. The switch itself reuses and the reconnect +/// cycle, exactly like the connection-driven failovers. +/// +/// +public partial class DataConnectionActor +{ + /// The lowest ServiceLevel in the OPC UA Healthy band (Part 4 §6.6.2.4.2). + internal const byte HealthyServiceLevelMinimum = 200; + + private const string ServiceLevelPollTimerKey = "servicelevel-poll"; + + private ServiceLevelFailoverPolicy _serviceLevelPolicy; + + /// When the current unbroken run of Degraded samples began; null when not degraded. + private DateTimeOffset? _degradedSince; + + /// When the last ServiceLevel-driven switch happened (dwell / anti-flap). + private DateTimeOffset _lastServiceLevelSwitchAt = DateTimeOffset.MinValue; + + private bool _serviceLevelReadInFlight; + private bool _serviceLevelProbeInFlight; + + /// True until the first sample of a session is evaluated (connect-time preference). + private bool _serviceLevelConnectCheckPending; + + /// Why is re-establishing the link. + internal enum ReconnectCause + { + /// The link was lost (disconnect, connection-level failure). + ConnectionLost, + + /// An in-place reconfiguration swapped the adapter. + Reconfigured, + + /// A sustained Degraded ServiceLevel swapped to the other endpoint (#138). + ServiceLevelFailover, + } + + /// + /// True when this connection's ServiceLevel is watched: the policy is enabled, there is a + /// backup to go to, and the live adapter can read a ServiceLevel. + /// + private bool ServiceLevelFailoverInForce => + _serviceLevelPolicy.Enabled && _backupConfig != null && _adapter is IServiceLevelReadable; + + /// Called on entering Connected. Arms the poll (first tick immediately). + private void StartServiceLevelMonitoring() + { + ResetServiceLevelSession(); + if (!ServiceLevelFailoverInForce) + return; + + _serviceLevelConnectCheckPending = true; + Timers.StartPeriodicTimer(ServiceLevelPollTimerKey, ServiceLevelPollTick.Instance, + TimeSpan.Zero, _options.ServiceLevelPollInterval); + } + + /// Called on leaving Connected. Results still in flight are fenced by generation. + private void StopServiceLevelMonitoring() + { + Timers.Cancel(ServiceLevelPollTimerKey); + ResetServiceLevelSession(); + } + + private void ResetServiceLevelSession() + { + _degradedSince = null; + _serviceLevelReadInFlight = false; + _serviceLevelProbeInFlight = false; + _serviceLevelConnectCheckPending = false; + } + + private void HandleServiceLevelPollTick() + { + // One read at a time, and none while the other endpoint is being probed: the probe's + // verdict is taken against the level it started from. + if (!ServiceLevelFailoverInForce || _serviceLevelReadInFlight || _serviceLevelProbeInFlight) + return; + + _serviceLevelReadInFlight = true; + var generation = _adapterGeneration; + var reader = (IServiceLevelReadable)_adapter; + var timeout = _options.ServiceLevelProbeTimeout; + Task.Run(() => reader.ReadServiceLevelAsync().WaitAsync(timeout)) + .ContinueWith(t => new ServiceLevelSampled(generation, + t.IsCompletedSuccessfully ? t.Result : null), TaskScheduler.Default) + .PipeTo(_self); + } + + private void HandleServiceLevelSampled(ServiceLevelSampled sampled) + { + if (sampled.AdapterGeneration != _adapterGeneration) + return; + + _serviceLevelReadInFlight = false; + var connectCheck = _serviceLevelConnectCheckPending; + _serviceLevelConnectCheckPending = false; + + if (sampled.Level is not { } level || level >= HealthyServiceLevelMinimum) + { + // Healthy, or unknown: either way the run of Degraded samples is broken. Unknown never + // counts toward a switch — acting on missing information is how a pair flaps. + if (_degradedSince != null) + _log.Info("[{0}] {1} endpoint ServiceLevel is {2} — degraded watch cleared", + _connectionName, _activeEndpoint, sampled.Level?.ToString(CultureInfo.InvariantCulture) ?? "unknown"); + _degradedSince = null; + return; + } + + var now = DateTimeOffset.UtcNow; + if (_degradedSince is null) + { + _degradedSince = now; + _log.Warning("[{0}] {1} endpoint reports ServiceLevel {2} (Degraded band, < {3}) — " + + "will check the other endpoint {4}", + _connectionName, _activeEndpoint, level, HealthyServiceLevelMinimum, + connectCheck ? "now (connect-time preference)" + : $"if it persists for {_serviceLevelPolicy.DegradedWindow.TotalSeconds:F0}s"); + } + + if (!connectCheck && now - _degradedSince.Value < _serviceLevelPolicy.DegradedWindow) + return; + + if (now - _lastServiceLevelSwitchAt < _serviceLevelPolicy.MinDwell) + { + _log.Debug("[{0}] ServiceLevel {1} still Degraded but the last ServiceLevel switch was {2:F0}s ago " + + "(dwell {3:F0}s) — staying", _connectionName, level, + (now - _lastServiceLevelSwitchAt).TotalSeconds, _serviceLevelPolicy.MinDwell.TotalSeconds); + return; + } + + StartServiceLevelProbe(level); + } + + /// + /// Reads the OTHER endpoint's ServiceLevel through a short-lived adapter built from its + /// configuration — connect, read, dispose — off the actor thread, bounded by + /// . The live adapter is untouched. + /// + private void StartServiceLevelProbe(byte currentLevel) + { + _serviceLevelProbeInFlight = true; + var generation = _adapterGeneration; + var otherConfig = _activeEndpoint == ActiveEndpoint.Primary ? _backupConfig! : _primaryConfig; + var factory = _factory; + var protocol = _protocolType; + var timeout = _options.ServiceLevelProbeTimeout; + + Task.Run(() => ProbeServiceLevelAsync(factory, protocol, otherConfig).WaitAsync(timeout)) + .ContinueWith(t => t.IsCompletedSuccessfully + ? new ServiceLevelProbed(generation, currentLevel, t.Result.Level, t.Result.Error) + : new ServiceLevelProbed(generation, currentLevel, null, + t.Exception?.GetBaseException() is TimeoutException + ? $"no answer within {timeout.TotalSeconds:F0}s" + : t.Exception?.GetBaseException().Message ?? "probe cancelled"), + TaskScheduler.Default) + .PipeTo(_self); + } + + private static async Task<(byte? Level, string? Error)> ProbeServiceLevelAsync( + IDataConnectionFactory factory, string protocol, IDictionary config) + { + IDataConnection? probe = null; + try + { + probe = factory.Create(protocol, config); + if (probe is not IServiceLevelReadable reader) + return (null, "the adapter cannot read a ServiceLevel"); + + await probe.ConnectAsync(config).ConfigureAwait(false); + var level = await reader.ReadServiceLevelAsync().ConfigureAwait(false); + return (level, level is null ? "ServiceLevel unreadable" : null); + } + catch (Exception ex) + { + return (null, ex.GetBaseException().Message); + } + finally + { + // Fire-and-forget: closing a session against a black-holed server can block for the + // operation timeout, and the probe's verdict must not wait for it. + if (probe != null) + _ = probe.DisposeAsync().AsTask().ContinueWith( + t => _ = t.Exception, TaskContinuationOptions.OnlyOnFaulted); + } + } + + private void HandleServiceLevelProbed(ServiceLevelProbed probed) + { + if (probed.AdapterGeneration != _adapterGeneration) + return; + + _serviceLevelProbeInFlight = false; + if (!ServiceLevelFailoverInForce || _degradedSince is null) + return; + + var other = _activeEndpoint == ActiveEndpoint.Primary ? ActiveEndpoint.Backup : ActiveEndpoint.Primary; + var now = DateTimeOffset.UtcNow; + + if (probed.OtherLevel is not { } otherLevel) + { + _log.Warning("[{0}] {1} endpoint ServiceLevel {2} is Degraded but the {3} endpoint is not " + + "available ({4}) — staying", _connectionName, _activeEndpoint, probed.CurrentLevel, other, + probed.Error); + _degradedSince = now; // re-arm: look again after another full window + return; + } + + if (otherLevel < HealthyServiceLevelMinimum || otherLevel <= probed.CurrentLevel) + { + _log.Warning("[{0}] {1} endpoint ServiceLevel {2} is Degraded and the {3} endpoint reports {4} " + + "(not a Healthy, higher level) — staying", _connectionName, _activeEndpoint, + probed.CurrentLevel, other, otherLevel); + _degradedSince = now; + return; + } + + var degradedFor = now - _degradedSince.Value; + var previousEndpoint = SwitchToOtherEndpoint(); + _consecutiveFailures = 0; + _consecutiveUnstableDisconnects = 0; + _lastServiceLevelSwitchAt = now; + + _log.Warning("[{0}] Failing over from {1} to {2} (ServiceLevel {3} Degraded for {4:F0}s; {2} reports {5})", + _connectionName, previousEndpoint, _activeEndpoint, probed.CurrentLevel, + degradedFor.TotalSeconds, otherLevel); + + if (_siteEventLogger != null) + { + _ = _siteEventLogger.LogEventAsync( + "connection", "Warning", null, _connectionName, + $"Failover from {previousEndpoint} to {_activeEndpoint} (ServiceLevel)", + $"{previousEndpoint} reported ServiceLevel {probed.CurrentLevel} for {degradedFor.TotalSeconds:F0}s; " + + $"{_activeEndpoint} reports {otherLevel}"); + } + + BecomeReconnecting(ReconnectCause.ServiceLevelFailover); + } + + /// Poll tick for the connected endpoint's ServiceLevel. + internal sealed record ServiceLevelPollTick + { + /// The single instance. + public static readonly ServiceLevelPollTick Instance = new(); + } + + /// One ServiceLevel sample of the connected endpoint (null = unknown). + internal sealed record ServiceLevelSampled(int AdapterGeneration, byte? Level); + + /// The other endpoint's ServiceLevel (null = unreachable / unreadable, see Error). + internal sealed record ServiceLevelProbed(int AdapterGeneration, byte CurrentLevel, byte? OtherLevel, string? Error); +} + +/// +/// The connection-level ServiceLevel failover settings, read from the PRIMARY endpoint's flattened +/// configuration (). Absent keys mean the defaults. +/// +internal readonly record struct ServiceLevelFailoverPolicy(bool Enabled, TimeSpan DegradedWindow, TimeSpan MinDwell) +{ + /// Parses the policy; malformed or out-of-range values fall back to safe values. + /// The primary endpoint's flattened configuration. + /// The policy in force. + public static ServiceLevelFailoverPolicy From(IDictionary? primaryConfig) + { + var enabled = true; + var window = OpcUaServiceLevelFailoverConfig.DefaultDegradedWindowSeconds; + var dwell = OpcUaServiceLevelFailoverConfig.DefaultMinDwellSeconds; + + if (primaryConfig != null) + { + if (primaryConfig.TryGetValue(OpcUaEndpointConfigSerializer.ServiceLevelFailoverEnabledKey, out var e) + && bool.TryParse(e, out var parsedEnabled)) + enabled = parsedEnabled; + if (primaryConfig.TryGetValue(OpcUaEndpointConfigSerializer.ServiceLevelFailoverWindowKey, out var w) + && int.TryParse(w, NumberStyles.Integer, CultureInfo.InvariantCulture, out var parsedWindow)) + window = Math.Max(1, parsedWindow); + if (primaryConfig.TryGetValue(OpcUaEndpointConfigSerializer.ServiceLevelFailoverDwellKey, out var d) + && int.TryParse(d, NumberStyles.Integer, CultureInfo.InvariantCulture, out var parsedDwell)) + dwell = Math.Max(0, parsedDwell); + } + + return new ServiceLevelFailoverPolicy(enabled, TimeSpan.FromSeconds(window), TimeSpan.FromSeconds(dwell)); + } +} diff --git a/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionActor.cs b/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionActor.cs index 216d6b49..93819772 100644 --- a/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionActor.cs @@ -29,7 +29,7 @@ namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Actors; /// Health reporting (connection status + tag resolution counts). /// Subscription lifecycle (register on create, cleanup on stop). /// -public class DataConnectionActor : UntypedActor, IWithTimers +public partial class DataConnectionActor : UntypedActor, IWithTimers { public enum ActiveEndpoint { Primary, Backup } @@ -328,6 +328,7 @@ public class DataConnectionActor : UntypedActor, IWithTimers _siteEventLogger = siteEventLogger; _connectionDetails = _primaryConfig; _tagResolutionInterval = _options.TagResolutionRetryInterval; + _serviceLevelPolicy = ServiceLevelFailoverPolicy.From(_primaryConfig); } /// @@ -491,6 +492,9 @@ public class DataConnectionActor : UntypedActor, IWithTimers // No session yet — a probe would fail for every tag. ReSubscribeAll's chunk // completions re-arm the timer once the link is up. break; + case ServiceLevelPollTick or ServiceLevelSampled or ServiceLevelProbed: + // ScadaBridge#138: belongs to a session that is over; BecomeConnected restarts it. + break; case QualityFlushTick: FlushQualityCountersIfDirty(); break; @@ -538,6 +542,8 @@ public class DataConnectionActor : UntypedActor, IWithTimers // Register row 54 — the only transition that raises the connected gauge. ScadaBridgeTelemetry.RecordDataConnectionState(_connectionName, ConnectionHealth.Connected); Become(Connected); + // ScadaBridge#138: watch this endpoint's ServiceLevel (no-op unless in force). + StartServiceLevelMonitoring(); foreach (var pending in _pendingTagSubscriptions.Values) Self.Tell(pending.Request, pending.Owner); _pendingTagSubscriptions.Clear(); @@ -592,6 +598,15 @@ public class DataConnectionActor : UntypedActor, IWithTimers case QualityFlushTick: FlushQualityCountersIfDirty(); break; + case ServiceLevelPollTick: + HandleServiceLevelPollTick(); + break; + case ServiceLevelSampled sampled: + HandleServiceLevelSampled(sampled); + break; + case ServiceLevelProbed probed: + HandleServiceLevelProbed(probed); + break; case AdapterDisconnected disconnected: // Fence the disconnect on its adapter generation. The outgoing adapter's // event can fire concurrently with SwapAdapter's unhook, so a Disconnected from @@ -629,28 +644,36 @@ public class DataConnectionActor : UntypedActor, IWithTimers // ── Reconnecting State ── - /// - /// True when this transition was caused by an in-place reconfiguration - /// () rather than by losing the link. It suppresses the - /// unstable-disconnect accounting and the failover check — the adapter has just been - /// swapped ON PURPOSE against a brand-new configuration, so counting it as an unstable - /// cycle would let a reconfigure trip a failover onto the new backup before the new primary - /// has been tried even once — and it reconnects immediately instead of waiting out + /// + /// Why the link is being re-established. (an + /// in-place reconfiguration, ) and + /// (ScadaBridge#138) both mean the adapter + /// has just been swapped ON PURPOSE rather than the link lost. They suppress the + /// unstable-disconnect accounting and the failover check — counting a deliberate swap as an + /// unstable cycle would let it trip a second, connection-driven failover before the new + /// endpoint has been tried even once — and reconnect immediately instead of waiting out /// , since there is nothing to wait for. /// - private void BecomeReconnecting(bool afterReconfigure = false) + private void BecomeReconnecting(ReconnectCause cause = ReconnectCause.ConnectionLost) { + var deliberateSwap = cause != ReconnectCause.ConnectionLost; _log.Warning("[{0}] Entering Reconnecting state{1}", _connectionName, - afterReconfigure ? " (reconfigured in place)" : string.Empty); - // The session is over: abandon its re-seed (02/L4). + cause switch + { + ReconnectCause.Reconfigured => " (reconfigured in place)", + ReconnectCause.ServiceLevelFailover => " (ServiceLevel failover)", + _ => string.Empty, + }); + // The session is over: abandon its re-seed (02/L4) and its ServiceLevel watch (#138). CancelSessionWork(); + StopServiceLevelMonitoring(); _healthCollector.UpdateConnectionHealth(_connectionName, ConnectionHealth.Disconnected); _healthCollector.UpdateConnectionEndpoint(_connectionName, "Disconnected"); // Register row 54 — drops this connection out of the connected gauge while the // total holds, which is exactly the connected < total alert condition. ScadaBridgeTelemetry.RecordDataConnectionState(_connectionName, ConnectionHealth.Disconnected); - if (!afterReconfigure) + if (!deliberateSwap) { // Track unstable connections toward failover. // If we were connected for less than the stability threshold, this counts @@ -671,15 +694,10 @@ public class DataConnectionActor : UntypedActor, IWithTimers // Failover if we keep connecting and going stale repeatedly if (_backupConfig != null && _consecutiveUnstableDisconnects >= _failoverRetryCount) { - var previousEndpoint = _activeEndpoint; - _activeEndpoint = _activeEndpoint == ActiveEndpoint.Primary - ? ActiveEndpoint.Backup - : ActiveEndpoint.Primary; + var previousEndpoint = SwitchToOtherEndpoint(); _consecutiveUnstableDisconnects = 0; _consecutiveFailures = 0; - SwapAdapter(_activeEndpoint == ActiveEndpoint.Primary ? _primaryConfig : _backupConfig); - _log.Warning("[{0}] Failing over from {1} to {2} (unstable connection pattern)", _connectionName, previousEndpoint, _activeEndpoint); @@ -697,10 +715,13 @@ public class DataConnectionActor : UntypedActor, IWithTimers if (_siteEventLogger != null) { _ = _siteEventLogger.LogEventAsync( - "connection", afterReconfigure ? "Info" : "Warning", null, _connectionName, - afterReconfigure - ? "Connection reconfigured — reconnecting on the new configuration" - : "Connection lost — entering reconnect cycle", + "connection", cause == ReconnectCause.ConnectionLost ? "Warning" : "Info", null, _connectionName, + cause switch + { + ReconnectCause.Reconfigured => "Connection reconfigured — reconnecting on the new configuration", + ReconnectCause.ServiceLevelFailover => $"ServiceLevel failover — reconnecting on the {_activeEndpoint} endpoint", + _ => "Connection lost — entering reconnect cycle", + }, null); } @@ -716,7 +737,25 @@ public class DataConnectionActor : UntypedActor, IWithTimers // Schedule reconnect attempt. StartSingleTimer REPLACES any reconnect already armed, // so a reconfigure landing mid-backoff cannot leave two AttemptConnect messages racing. Timers.StartSingleTimer("reconnect", new AttemptConnect(), - afterReconfigure ? TimeSpan.Zero : _options.ReconnectInterval); + deliberateSwap ? TimeSpan.Zero : _options.ReconnectInterval); + } + + /// + /// Flips to the other configured endpoint and swaps the adapter + /// onto it — the one endpoint switch shared by every failover trigger: consecutive connect + /// failures (), the unstable-connection pattern + /// () and a sustained Degraded ServiceLevel (ScadaBridge#138). + /// Callers reset whichever counters their trigger owns. Requires a backup configuration. + /// + /// The endpoint that was active before the switch. + private ActiveEndpoint SwitchToOtherEndpoint() + { + var previousEndpoint = _activeEndpoint; + _activeEndpoint = _activeEndpoint == ActiveEndpoint.Primary + ? ActiveEndpoint.Backup + : ActiveEndpoint.Primary; + SwapAdapter(_activeEndpoint == ActiveEndpoint.Primary ? _primaryConfig : _backupConfig!); + return previousEndpoint; } /// @@ -845,6 +884,8 @@ public class DataConnectionActor : UntypedActor, IWithTimers _primaryConfig = command.PrimaryConnectionDetails; _backupConfig = command.BackupConnectionDetails; _failoverRetryCount = command.FailoverRetryCount; + _serviceLevelPolicy = ServiceLevelFailoverPolicy.From(_primaryConfig); + _lastServiceLevelSwitchAt = DateTimeOffset.MinValue; _activeEndpoint = ActiveEndpoint.Primary; _consecutiveFailures = 0; @@ -852,7 +893,7 @@ public class DataConnectionActor : UntypedActor, IWithTimers SwapAdapter(_primaryConfig); - BecomeReconnecting(afterReconfigure: true); + BecomeReconnecting(ReconnectCause.Reconfigured); } /// @@ -930,6 +971,9 @@ public class DataConnectionActor : UntypedActor, IWithTimers // Ignore — the adapter has no live session; ReSubscribeAll's chunk // completions re-arm the retry timer after reconnect. break; + case ServiceLevelPollTick or ServiceLevelSampled or ServiceLevelProbed: + // ScadaBridge#138: belongs to a session that is over; BecomeConnected restarts it. + break; case QualityFlushTick: FlushQualityCountersIfDirty(); break; @@ -1219,14 +1263,9 @@ public class DataConnectionActor : UntypedActor, IWithTimers // Failover: switch endpoint after exhausting retry count (only if backup is configured) if (_backupConfig != null && _consecutiveFailures >= _failoverRetryCount) { - var previousEndpoint = _activeEndpoint; - _activeEndpoint = _activeEndpoint == ActiveEndpoint.Primary - ? ActiveEndpoint.Backup - : ActiveEndpoint.Primary; + var previousEndpoint = SwitchToOtherEndpoint(); _consecutiveFailures = 0; - SwapAdapter(_activeEndpoint == ActiveEndpoint.Primary ? _primaryConfig : _backupConfig); - _log.Warning("[{0}] Failing over from {1} to {2}", _connectionName, previousEndpoint, _activeEndpoint); diff --git a/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Adapters/OpcUaDataConnection.cs b/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Adapters/OpcUaDataConnection.cs index 25dd20fb..279e8caf 100644 --- a/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Adapters/OpcUaDataConnection.cs +++ b/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Adapters/OpcUaDataConnection.cs @@ -21,8 +21,11 @@ namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Adapters; /// public class OpcUaDataConnection : IDataConnection, IBrowsableDataConnection, IAlarmSubscribableConnection, IAddressSpaceSearchable, - IBatchSubscribableConnection + IBatchSubscribableConnection, IServiceLevelReadable { + /// The standard Server_ServiceLevel node (ns=0), OPC UA Part 5. + internal const string ServiceLevelNodeId = "i=2267"; + private readonly IOpcUaClientFactory _clientFactory; private readonly ILogger _logger; private IOpcUaClient? _client; @@ -291,6 +294,39 @@ public class OpcUaDataConnection } } + /// + /// + /// ScadaBridge#138. Unlike , a failure here never raises + /// : the poll runs every few seconds alongside the keep-alive, and a + /// dead session is already detected (and reconnected) there. A failure only means "unknown". + /// + public async Task ReadServiceLevelAsync(CancellationToken cancellationToken = default) + { + var client = _client; + if (client == null || !client.IsConnected) + return null; + + try + { + var (value, _, statusCode) = await client.ReadValueAsync(ServiceLevelNodeId, cancellationToken); + if (MapStatusCode(statusCode) == QualityCode.Bad) + return null; + return value switch + { + byte b => b, + sbyte or short or ushort or int or uint or long or ulong + when Convert.ToInt64(value, System.Globalization.CultureInfo.InvariantCulture) is >= 0 and <= 255 + => (byte)Convert.ToInt64(value, System.Globalization.CultureInfo.InvariantCulture), + _ => null, + }; + } + catch (Exception ex) when (ex is not OperationCanceledException) + { + _logger.LogDebug(ex, "OPC UA ServiceLevel read failed for {Endpoint}", _endpointUrl); + return null; + } + } + /// public async Task> ReadBatchAsync(IEnumerable tagPaths, CancellationToken cancellationToken = default) { diff --git a/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/DataConnectionOptions.cs b/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/DataConnectionOptions.cs index 21cac519..14586faa 100644 --- a/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/DataConnectionOptions.cs +++ b/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/DataConnectionOptions.cs @@ -119,4 +119,19 @@ public class DataConnectionOptions /// (SeedReadMaxAttempts - 1) × SeedReadRetryDelay regardless of tag count. /// public TimeSpan SeedReadRetryDelay { get; set; } = TimeSpan.FromMilliseconds(250); + + /// + /// ScadaBridge#138: how often a connected endpoint's ServiceLevel (i=2267) is read when + /// ServiceLevel-aware failover is in force (OPC UA connection with a backup and the + /// connection's ServiceLevelFailover.Enabled). One single-node Read per poll. Must be + /// shorter than the smallest configured degraded window to make that window meaningful. + /// + public TimeSpan ServiceLevelPollInterval { get; set; } = TimeSpan.FromSeconds(2); + + /// + /// ScadaBridge#138: budget for reading a ServiceLevel — the poll on the connected endpoint, + /// and the probe of the other endpoint (connect a short-lived session, read, close). A probe + /// that does not finish in time counts as "other endpoint unreachable", so no switch happens. + /// + public TimeSpan ServiceLevelProbeTimeout { get; set; } = TimeSpan.FromSeconds(15); } diff --git a/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/DataConnectionOptionsValidator.cs b/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/DataConnectionOptionsValidator.cs index 4b70abf6..086c4921 100644 --- a/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/DataConnectionOptionsValidator.cs +++ b/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/DataConnectionOptionsValidator.cs @@ -64,5 +64,11 @@ public sealed class DataConnectionOptionsValidator : OptionsValidatorBase= TimeSpan.Zero && options.MxAlarmStreamReopenDebounce <= TimeSpan.FromSeconds(5), $"ScadaBridge:DataConnection:MxAlarmStreamReopenDebounce must be between 0 and 5 seconds (was {options.MxAlarmStreamReopenDebounce})."); + + builder.RequireThat(options.ServiceLevelPollInterval > TimeSpan.Zero, + $"ScadaBridge:DataConnection:ServiceLevelPollInterval must be a positive duration (was {options.ServiceLevelPollInterval})."); + + builder.RequireThat(options.ServiceLevelProbeTimeout > TimeSpan.Zero, + $"ScadaBridge:DataConnection:ServiceLevelProbeTimeout must be a positive duration (was {options.ServiceLevelProbeTimeout})."); } } diff --git a/tests/ZB.MOM.WW.ScadaBridge.CentralUI.Tests/DataConnectionFormTests.cs b/tests/ZB.MOM.WW.ScadaBridge.CentralUI.Tests/DataConnectionFormTests.cs index ac15fb9a..712c7c70 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.CentralUI.Tests/DataConnectionFormTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.CentralUI.Tests/DataConnectionFormTests.cs @@ -167,4 +167,74 @@ public class DataConnectionFormTests : BunitContext Assert.Equal(60000, doc.RootElement.GetProperty("sessionTimeoutMs").GetInt32()); } + + /// + /// ScadaBridge#138: with a backup, the ServiceLevel failover fields shown next to Failover + /// Retry Count are stored on the PRIMARY endpoint's JSON; the backup's copy stays null. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task Save_WithBackup_StoresServiceLevelFailoverOnThePrimary() + { + DataConnection? captured = null; + await _siteRepo.AddDataConnectionAsync(Arg.Do(d => captured = d)); + + var cut = RenderForCreateSite(1); + cut.FindAll("input[type='text']") + .First(i => !i.HasAttribute("readonly") + && (i.GetAttribute("placeholder") is null + || !i.GetAttribute("placeholder")!.StartsWith("opc.tcp"))) + .Change("OtOpcUa-Pair"); + cut.FindAll("input[type='text']") + .First(i => i.GetAttribute("placeholder")?.StartsWith("opc.tcp") == true) + .Change("opc.tcp://otopcua-a:4840"); + await cut.FindAll("button").First(b => b.TextContent.Trim() == "Add Backup Endpoint").ClickAsync(new()); + cut.FindAll("input[type='text']") + .Where(i => i.GetAttribute("placeholder")?.StartsWith("opc.tcp") == true) + .ElementAt(1) + .Change("opc.tcp://otopcua-b:4840"); + + // Defaults are on screen, then edited. + Assert.Equal("10", cut.Find("#sl-failover-window").GetAttribute("value")); + Assert.Equal("60", cut.Find("#sl-failover-dwell").GetAttribute("value")); + cut.Find("#sl-failover-window").Change("15"); + cut.Find("#sl-failover-dwell").Change("90"); + + await cut.FindAll("button").First(b => b.TextContent.Trim() == "Save").ClickAsync(new()); + + Assert.NotNull(captured); + using var primary = JsonDocument.Parse(captured!.PrimaryConfiguration!); + var sl = primary.RootElement.GetProperty("serviceLevelFailover"); + Assert.True(sl.GetProperty("enabled").GetBoolean()); + Assert.Equal(15, sl.GetProperty("degradedWindowSeconds").GetInt32()); + Assert.Equal(90, sl.GetProperty("minDwellSeconds").GetInt32()); + + using var backup = JsonDocument.Parse(captured.BackupConfiguration!); + Assert.Equal(JsonValueKind.Null, backup.RootElement.GetProperty("serviceLevelFailover").ValueKind); + } + + /// ScadaBridge#138: without a backup the setting is neither shown nor stored. + /// A task that represents the asynchronous operation. + [Fact] + public async Task Save_WithoutBackup_StoresNoServiceLevelFailover() + { + DataConnection? captured = null; + await _siteRepo.AddDataConnectionAsync(Arg.Do(d => captured = d)); + + var cut = RenderForCreateSite(1); + Assert.Empty(cut.FindAll("#sl-failover-enabled")); + cut.FindAll("input[type='text']") + .First(i => !i.HasAttribute("readonly") + && (i.GetAttribute("placeholder") is null + || !i.GetAttribute("placeholder")!.StartsWith("opc.tcp"))) + .Change("PLC-1"); + cut.FindAll("input[type='text']") + .First(i => i.GetAttribute("placeholder")?.StartsWith("opc.tcp") == true) + .Change("opc.tcp://plant-a:4840"); + await cut.FindAll("button").First(b => b.TextContent.Trim() == "Save").ClickAsync(new()); + + Assert.NotNull(captured); + using var primary = JsonDocument.Parse(captured!.PrimaryConfiguration!); + Assert.Equal(JsonValueKind.Null, primary.RootElement.GetProperty("serviceLevelFailover").ValueKind); + } } diff --git a/tests/ZB.MOM.WW.ScadaBridge.Commons.Tests/Types/DataConnections/OpcUaEndpointConfigSerializerTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Commons.Tests/Types/DataConnections/OpcUaEndpointConfigSerializerTests.cs index 16037647..26bf4963 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Commons.Tests/Types/DataConnections/OpcUaEndpointConfigSerializerTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Commons.Tests/Types/DataConnections/OpcUaEndpointConfigSerializerTests.cs @@ -504,4 +504,58 @@ public class OpcUaEndpointConfigSerializerTests Assert.NotNull(config.UserIdentity); Assert.Equal(OpcUaUserTokenType.Anonymous, config.UserIdentity!.TokenType); } + + /// + /// ScadaBridge#138: the ServiceLevel failover object round-trips through the stored JSON and + /// through the flat dict the site hands the adapter, under its documented keys. + /// + [Fact] + public void ServiceLevelFailover_RoundTrips_JsonAndFlatDict() + { + var original = new OpcUaEndpointConfig + { + EndpointUrl = "opc.tcp://otopcua-a:4840", + ServiceLevelFailover = new OpcUaServiceLevelFailoverConfig + { + Enabled = false, DegradedWindowSeconds = 25, MinDwellSeconds = 120, + }, + }; + + var json = OpcUaEndpointConfigSerializer.Serialize(original); + Assert.Contains("\"serviceLevelFailover\":{\"enabled\":false,\"degradedWindowSeconds\":25,\"minDwellSeconds\":120}", json); + var (round, isLegacy) = OpcUaEndpointConfigSerializer.Deserialize(json); + Assert.False(isLegacy); + Assert.NotNull(round.ServiceLevelFailover); + Assert.False(round.ServiceLevelFailover!.Enabled); + Assert.Equal(25, round.ServiceLevelFailover.DegradedWindowSeconds); + Assert.Equal(120, round.ServiceLevelFailover.MinDwellSeconds); + + var flat = OpcUaEndpointConfigSerializer.ToFlatDict(round); + Assert.Equal("False", flat["ServiceLevelFailover.Enabled"]); + Assert.Equal("25", flat["ServiceLevelFailover.DegradedWindowSeconds"]); + Assert.Equal("120", flat["ServiceLevelFailover.MinDwellSeconds"]); + var back = OpcUaEndpointConfigSerializer.FromFlatDict(flat).ServiceLevelFailover; + Assert.NotNull(back); + Assert.False(back!.Enabled); + Assert.Equal(25, back.DegradedWindowSeconds); + Assert.Equal(120, back.MinDwellSeconds); + } + + /// + /// ScadaBridge#138, upgrade safety: a row stored before the setting existed has no + /// serviceLevelFailover; it deserializes as Typed (not Malformed) with the object null, + /// and its flat dict gains no new keys — so existing connections are not seen as changed. + /// + [Fact] + public void ServiceLevelFailover_AbsentFromAnExistingRow_IsNull_AndAddsNoFlatKeys() + { + const string stored = "{\"endpointUrl\":\"opc.tcp://plant-a:4840\",\"securityMode\":\"none\",\"heartbeat\":null}"; + var result = OpcUaEndpointConfigSerializer.Deserialize(stored); + Assert.Equal(OpcUaConfigParseStatus.Typed, result.Status); + Assert.Null(result.Config.ServiceLevelFailover); + + var flat = OpcUaEndpointConfigSerializer.ToFlatDict(result.Config); + Assert.DoesNotContain(flat.Keys, k => k.StartsWith("ServiceLevelFailover.", StringComparison.Ordinal)); + Assert.Null(OpcUaEndpointConfigSerializer.FromFlatDict(flat).ServiceLevelFailover); + } } diff --git a/tests/ZB.MOM.WW.ScadaBridge.Commons.Tests/Validators/OpcUaEndpointConfigValidatorTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Commons.Tests/Validators/OpcUaEndpointConfigValidatorTests.cs index 040c76b9..7073a354 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Commons.Tests/Validators/OpcUaEndpointConfigValidatorTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Commons.Tests/Validators/OpcUaEndpointConfigValidatorTests.cs @@ -220,4 +220,30 @@ public class OpcUaEndpointConfigValidatorTests var r = OpcUaEndpointConfigValidator.Validate(c, fieldPrefix: "Primary."); Assert.Contains(r.Errors, e => e.EntityName == "Primary.UserIdentity.Username"); } + + /// ScadaBridge#138: the ServiceLevel failover window and dwell are range-checked, prefixed. + /// Degraded window, seconds. + /// Minimum dwell, seconds. + /// The field expected to fail, or null when valid. + [Theory] + [InlineData(10, 60, null)] + [InlineData(1, 0, null)] + [InlineData(3600, 86400, null)] + [InlineData(0, 60, "Primary.ServiceLevelFailover.DegradedWindowSeconds")] + [InlineData(3601, 60, "Primary.ServiceLevelFailover.DegradedWindowSeconds")] + [InlineData(10, -1, "Primary.ServiceLevelFailover.MinDwellSeconds")] + [InlineData(10, 86401, "Primary.ServiceLevelFailover.MinDwellSeconds")] + public void Validate_ServiceLevelFailoverRanges(int window, int dwell, string? badField) + { + var c = Valid(); + c.ServiceLevelFailover = new OpcUaServiceLevelFailoverConfig + { + DegradedWindowSeconds = window, MinDwellSeconds = dwell, + }; + var r = OpcUaEndpointConfigValidator.Validate(c, fieldPrefix: "Primary."); + if (badField is null) + Assert.True(r.IsValid); + else + Assert.Contains(r.Errors, e => e.EntityName == badField); + } } diff --git a/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Actors/DataConnectionPromotionRefreshTests.cs b/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Actors/DataConnectionPromotionRefreshTests.cs new file mode 100644 index 00000000..9900460d --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Actors/DataConnectionPromotionRefreshTests.cs @@ -0,0 +1,110 @@ +using Akka.Actor; +using Akka.TestKit.Xunit2; +using NSubstitute; +using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Protocol; +using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.DataConnection; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Alarms; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; +using ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Actors; +using ZB.MOM.WW.ScadaBridge.HealthMonitoring; + +namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests.Actors; + +/// +/// ScadaBridge#138 / OtOpcUa Warm design §5 "Alarms are compatible", at the actor: the promotion +/// refresh OtOpcUa sends on its own (an unsolicited RefreshStart / retained conditions / RefreshEnd +/// on an EXISTING alarm feed, already framed by RealOpcUaClient as SnapshotStart / Snapshot / +/// SnapshotComplete under a new epoch — see RealOpcUaPromotionRefreshTests) goes through the +/// existing snapshot handling: it is forwarded in order to the subscriber, and it RECONCILES the +/// connection's alarm cache — a condition the promoted node no longer retains is gone from what a +/// later subscriber is replayed. +/// +public class DataConnectionPromotionRefreshTests : TestKit +{ + private const string Source = "Tank01"; + + private readonly ISiteHealthCollector _health = Substitute.For(); + private readonly IDataConnectionFactory _factory = Substitute.For(); + private readonly DataConnectionOptions _options = new() + { + ReconnectInterval = TimeSpan.FromMilliseconds(100), + TagResolutionRetryInterval = TimeSpan.FromMilliseconds(200), + WriteTimeout = TimeSpan.FromSeconds(5), + }; + + private static NativeAlarmTransition Boundary(AlarmTransitionKind kind, Guid epoch) => new( + "", Source, "", kind, + new AlarmConditionState(false, true, null, AlarmShelveState.Unshelved, false, 0), + "", "", "", "", "", null, DateTimeOffset.UtcNow, "", "", + SnapshotSourceReference: Source, SnapshotId: epoch); + + private static NativeAlarmTransition Condition( + string condition, AlarmTransitionKind kind, Guid? epoch, bool active = true, bool acked = false) => new( + $"{Source}.{condition}", Source, "AlarmConditionType", kind, + new AlarmConditionState(active, acked, null, AlarmShelveState.Unshelved, false, 500), + "", "", "msg", "", "", null, DateTimeOffset.UtcNow, "", "", + SnapshotSourceReference: epoch is null ? null : Source, SnapshotId: epoch); + + /// + /// Subscribe-time snapshot {Hi, Lo}; then the unsolicited promotion snapshot {Hi} (Lo cleared + /// while the node was Standby). The subscriber sees the promotion frame in order, and a second + /// subscriber that joins afterwards is replayed Hi only. + /// + [Fact] + public void UnsolicitedPromotionSnapshot_IsForwardedInOrder_AndReconcilesTheCache() + { + AlarmTransitionCallback? cb = null; + var adapter = Substitute.For(); + adapter.ConnectAsync(Arg.Any>(), Arg.Any()) + .Returns(Task.CompletedTask); + ((IAlarmSubscribableConnection)adapter) + .SubscribeAlarmsAsync(Arg.Any(), Arg.Any(), + Arg.Do(c => cb = c), Arg.Any()) + .Returns(Task.FromResult("alarm-sub-1")); + var actor = Sys.ActorOf(Props.Create(() => new DataConnectionActor( + "conn", adapter, _options, _health, _factory, "OpcUa"))); + + var first = CreateTestProbe(); + actor.Tell(new SubscribeAlarmsRequest("c1", "inst1", "conn", Source, null, DateTimeOffset.UtcNow), first); + first.ExpectMsg(m => m.Success); + Assert.NotNull(cb); + + // Subscribe-time refresh on the then-Active node. + var subscribeEpoch = Guid.NewGuid(); + cb!(Boundary(AlarmTransitionKind.SnapshotStart, subscribeEpoch)); + cb(Condition("Hi", AlarmTransitionKind.Snapshot, subscribeEpoch)); + cb(Condition("Lo", AlarmTransitionKind.Snapshot, subscribeEpoch)); + cb(Boundary(AlarmTransitionKind.SnapshotComplete, subscribeEpoch)); + first.ReceiveN(4, TimeSpan.FromSeconds(3)); + + // Promotion on the other node: an unsolicited refresh on this existing feed. + var promotionEpoch = Guid.NewGuid(); + cb(Boundary(AlarmTransitionKind.SnapshotStart, promotionEpoch)); + cb(Condition("Hi", AlarmTransitionKind.Snapshot, promotionEpoch)); + cb(Boundary(AlarmTransitionKind.SnapshotComplete, promotionEpoch)); + + var frame = first.ReceiveN(3, TimeSpan.FromSeconds(3)).Cast().ToList(); + Assert.Equal( + [AlarmTransitionKind.SnapshotStart, AlarmTransitionKind.Snapshot, AlarmTransitionKind.SnapshotComplete], + frame.Select(u => u.Transition.Kind)); + Assert.All(frame, u => Assert.Equal(promotionEpoch, u.Transition.SnapshotId)); + Assert.Equal($"{Source}.Hi", frame[1].Transition.SourceReference); + + // A live event after the promotion is still routed live. + cb(Condition("Hi", AlarmTransitionKind.Acknowledge, epoch: null, acked: true)); + first.ExpectMsg(u => u.Transition.Kind == AlarmTransitionKind.Acknowledge); + + // A subscriber joining now is replayed the reconciled state: Hi only, Lo is gone. + var second = CreateTestProbe(); + actor.Tell(new SubscribeAlarmsRequest("c2", "inst2", "conn", Source, null, DateTimeOffset.UtcNow), second); + var replay = second.ReceiveN(4, TimeSpan.FromSeconds(3)); + var transitions = replay.OfType().Select(u => u.Transition).ToList(); + Assert.Single(replay.OfType(), r => r.Success); + Assert.Equal( + [AlarmTransitionKind.SnapshotStart, AlarmTransitionKind.Snapshot, AlarmTransitionKind.SnapshotComplete], + transitions.Select(t => t.Kind)); + Assert.Equal([$"{Source}.Hi"], transitions.Where(t => t.Kind == AlarmTransitionKind.Snapshot).Select(t => t.SourceReference)); + second.ExpectNoMsg(TimeSpan.FromMilliseconds(300)); + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Actors/DataConnectionServiceLevelFailoverTests.cs b/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Actors/DataConnectionServiceLevelFailoverTests.cs new file mode 100644 index 00000000..5fc988cc --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Actors/DataConnectionServiceLevelFailoverTests.cs @@ -0,0 +1,377 @@ +using System.Diagnostics; +using Akka.Actor; +using Akka.TestKit.Xunit2; +using NSubstitute; +using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Protocol; +using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.DataConnection; +using ZB.MOM.WW.ScadaBridge.Commons.Serialization; +using ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Actors; +using ZB.MOM.WW.ScadaBridge.HealthMonitoring; + +namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests.Actors; + +/// +/// ScadaBridge#138 — ServiceLevel-aware failover for a primary/backup OPC UA connection, the +/// ScadaBridge prerequisite for OtOpcUa Warm redundancy (lmxopcua#717). Under Warm the standby keeps +/// a healthy session but serves BadOutOfService data at a standby ServiceLevel (90 / 50 / +/// boot 1); the Active node is 200 or more (250 Primary / 240 Secondary today). +/// +/// +/// Each trigger has a matching non-trigger case, and every "no switch" assertion is paired with a +/// positive control proving the evaluation it relies on actually ran (the ServiceLevel was read, or +/// the other endpoint was probed) — an inert feature would otherwise pass them all. +/// +/// +public class DataConnectionServiceLevelFailoverTests : TestKit +{ + private const string Primary = "opc.tcp://otopcua-a:4840"; + private const string Backup = "opc.tcp://otopcua-b:4840"; + private const string Tag = "nsu=https://zb.com/otopcua/uns;s=Line1.Pump.Speed"; + + private readonly ServiceLevelRig _rig = new(); + private readonly ISiteHealthCollector _health = Substitute.For(); + private readonly IDataConnectionFactory _factory = Substitute.For(); + private readonly DataConnectionOptions _options = new() + { + ReconnectInterval = TimeSpan.FromMilliseconds(100), + TagResolutionRetryInterval = TimeSpan.FromMilliseconds(200), + WriteTimeout = TimeSpan.FromSeconds(5), + SeedReadMaxAttempts = 1, + SeedReadRetryDelay = TimeSpan.FromMilliseconds(10), + ServiceLevelPollInterval = TimeSpan.FromMilliseconds(50), + ServiceLevelProbeTimeout = TimeSpan.FromSeconds(2), + }; + + /// Wires the factory to build a rig-bound fake for whichever endpoint it is asked for. + public DataConnectionServiceLevelFailoverTests() + : base(@"akka.loglevel = DEBUG") + { + _factory.Create(Arg.Any(), Arg.Any>()) + .Returns(ci => new FakeServiceLevelConnection(_rig, ((IDictionary)ci[1])["Endpoint"])); + } + + private static Dictionary Config( + string endpoint, bool? enabled = null, int? windowSeconds = null, int? dwellSeconds = null) + { + var config = new Dictionary { ["Endpoint"] = endpoint }; + if (enabled is { } e) config[OpcUaEndpointConfigSerializer.ServiceLevelFailoverEnabledKey] = e.ToString(); + if (windowSeconds is { } w) config[OpcUaEndpointConfigSerializer.ServiceLevelFailoverWindowKey] = w.ToString(); + if (dwellSeconds is { } d) config[OpcUaEndpointConfigSerializer.ServiceLevelFailoverDwellKey] = d.ToString(); + return config; + } + + private IActorRef Start( + string name, IDictionary primaryConfig, IDictionary? backupConfig, + IDataConnection? initialAdapter = null) + { + var adapter = initialAdapter ?? new FakeServiceLevelConnection(_rig, Primary); + var actor = Sys.ActorOf(Props.Create(() => new DataConnectionActor( + name, adapter, _options, _health, _factory, "OpcUa", primaryConfig, backupConfig, 3, null)), name); + actor.Tell(new SubscribeTagsRequest("c1", "inst1", name, [Tag], DateTimeOffset.UtcNow)); + ExpectMsg(TimeSpan.FromSeconds(5)); + return actor; + } + + private string ActiveEndpoint(IActorRef actor) => + actor.Ask(new DataConnectionActor.GetHealthReport(), TimeSpan.FromSeconds(3)) + .GetAwaiter().GetResult().ActiveEndpoint; + + private bool LiveOn(string endpoint) => _rig.Subscribes.Contains($"{endpoint}|{Tag}"); + + /// + /// Warm promotion seen from a connection on the old Active: it was Healthy (250), then the + /// node becomes the Standby (90) while the other node is promoted (250). The switch waits for + /// the degraded window, then lands on the backup, where every subscription is re-issued. + /// + [Fact] + public void Sustained_Degraded_FailsOver_AfterTheWindow_AndResubscribesOnTheOtherEndpoint() + { + _rig.SetLevel(Primary, 250); + _rig.SetLevel(Backup, 240); + var actor = Start("sl-window", Config(Primary, windowSeconds: 2, dwellSeconds: 0), Config(Backup)); + AwaitCondition(() => _rig.ReadCount(Primary) >= 3, TimeSpan.FromSeconds(5)); + + var degradedAt = Stopwatch.StartNew(); + _rig.SetLevel(Primary, 90); + _rig.SetLevel(Backup, 250); + + // Not before the window: half-way through, still on the primary and nothing probed. + Thread.Sleep(1000); + Assert.Equal("Primary", ActiveEndpoint(actor)); + Assert.Equal(0, _rig.ConnectCount(Backup)); + + AwaitCondition(() => LiveOn(Backup), TimeSpan.FromSeconds(5)); + Assert.True(degradedAt.Elapsed >= TimeSpan.FromSeconds(1.9), + $"switched after {degradedAt.Elapsed.TotalSeconds:F2}s, before the 2s window"); + Assert.Equal("Backup", ActiveEndpoint(actor)); + } + + /// + /// Connect-time preference: the connection starts on a primary that is the Warm standby (90) + /// while the backup is Active (250). It leaves at once — no 30 s window — because the first + /// sample of a session is evaluated immediately. + /// + [Fact] + public void Connect_OnDegradedEndpoint_PrefersTheHealthierOther_WithoutWaitingTheWindow() + { + _rig.SetLevel(Primary, 90); + _rig.SetLevel(Backup, 250); + var sw = Stopwatch.StartNew(); + var actor = Start("sl-connect", Config(Primary, windowSeconds: 30), Config(Backup)); + + AwaitCondition(() => LiveOn(Backup), TimeSpan.FromSeconds(5)); + Assert.True(sw.Elapsed < TimeSpan.FromSeconds(10), $"took {sw.Elapsed.TotalSeconds:F1}s"); + Assert.Equal("Backup", ActiveEndpoint(actor)); + } + + /// Booting node (ServiceLevel 1) on connect, Active other → leave it too. + [Fact] + public void Connect_OnBootingEndpoint_ServiceLevel1_PrefersTheActiveOther() + { + _rig.SetLevel(Primary, 1); + _rig.SetLevel(Backup, 240); + var actor = Start("sl-boot", Config(Primary, windowSeconds: 30), Config(Backup)); + + AwaitCondition(() => LiveOn(Backup), TimeSpan.FromSeconds(5)); + Assert.Equal("Backup", ActiveEndpoint(actor)); + } + + /// + /// Today's OtOpcUa pair: both nodes Active-band, 250/240 in either order. Nothing is ever + /// probed and nothing switches — a 240 Secondary never triggers. Positive control: the + /// connected endpoint's ServiceLevel WAS read many times across more than a full window. + /// + /// The connected (primary) endpoint's ServiceLevel. + /// The backup endpoint's ServiceLevel. + [Theory] + [InlineData(240, 250)] + [InlineData(250, 240)] + [InlineData(200, 250)] + public void ActiveBandPair_NeverSwitches_NorProbes(byte primaryLevel, byte backupLevel) + { + _rig.SetLevel(Primary, primaryLevel); + _rig.SetLevel(Backup, backupLevel); + var actor = Start($"sl-today-{primaryLevel}-{backupLevel}", Config(Primary, windowSeconds: 1, dwellSeconds: 0), Config(Backup)); + + Thread.Sleep(2000); + Assert.True(_rig.ReadCount(Primary) >= 10, $"only {_rig.ReadCount(Primary)} reads — the watch did not run"); + Assert.Equal(0, _rig.ConnectCount(Backup)); + Assert.Equal("Primary", ActiveEndpoint(actor)); + } + + /// + /// Both endpoints degraded (Warm standby 90 vs standby-degraded 50, either way round): stay + /// put. Positive control: the other endpoint WAS probed, so the verdict, not a dead watch, + /// kept the connection where it is. + /// + /// The connected (primary) endpoint's ServiceLevel. + /// The backup endpoint's ServiceLevel. + [Theory] + [InlineData(50, 90)] + [InlineData(90, 50)] + [InlineData(90, 199)] + public void BothDegraded_StaysPut_AfterProbingTheOther(byte primaryLevel, byte backupLevel) + { + _rig.SetLevel(Primary, primaryLevel); + _rig.SetLevel(Backup, backupLevel); + var actor = Start($"sl-both-{primaryLevel}-{backupLevel}", Config(Primary, windowSeconds: 1, dwellSeconds: 0), Config(Backup)); + + AwaitCondition(() => _rig.ReadCount(Backup) >= 2, TimeSpan.FromSeconds(5)); + Assert.Equal("Primary", ActiveEndpoint(actor)); + Assert.False(LiveOn(Backup)); + } + + /// + /// The other endpoint is down: stay on the degraded one (it still answers). Positive control: + /// sessions to the backup were attempted. + /// + [Fact] + public void OtherEndpointUnreachable_StaysPut() + { + _rig.SetLevel(Primary, 90); + _rig.SetLevel(Backup, 250); + _rig.SetReachable(Backup, false); + var actor = Start("sl-unreachable", Config(Primary, windowSeconds: 1, dwellSeconds: 0), Config(Backup)); + + AwaitCondition(() => _rig.ConnectCount(Backup) >= 2, TimeSpan.FromSeconds(5)); + Assert.Equal("Primary", ActiveEndpoint(actor)); + Assert.False(LiveOn(Backup)); + + // ...and once it comes back, the next window moves the connection. + _rig.SetReachable(Backup, true); + AwaitCondition(() => LiveOn(Backup), TimeSpan.FromSeconds(5)); + } + + /// + /// An unreadable ServiceLevel (Bad status, server without the node) is unknown, never + /// Degraded: no probe, no switch. Positive control: the reads happened. + /// + [Fact] + public void UnknownServiceLevel_IsNotDegraded() + { + _rig.SetLevel(Primary, (byte?)null); + _rig.SetLevel(Backup, 250); + var actor = Start("sl-unknown", Config(Primary, windowSeconds: 1, dwellSeconds: 0), Config(Backup)); + + Thread.Sleep(2000); + Assert.True(_rig.ReadCount(Primary) >= 10); + Assert.Equal(0, _rig.ConnectCount(Backup)); + Assert.Equal("Primary", ActiveEndpoint(actor)); + } + + /// + /// Hysteresis: a level that keeps dipping below 200 but recovers between samples never forms + /// an unbroken window, so it never switches. Positive control: Degraded samples were seen. + /// + [Fact] + public void FlickeringLevel_NeverCompletesTheWindow() + { + var reads = 0; + // Healthy for the first samples, so the connect-time check is not what is exercised; then + // three Degraded samples out of every four. + _rig.SetLevel(Primary, () => + { + var n = Interlocked.Increment(ref reads); + return n <= 3 || n % 4 == 0 ? (byte)250 : (byte)90; + }); + _rig.SetLevel(Backup, 250); + var actor = Start("sl-flicker", Config(Primary, windowSeconds: 1, dwellSeconds: 0), Config(Backup)); + + Thread.Sleep(2500); + Assert.True(Volatile.Read(ref reads) >= 20); + Assert.Equal("Primary", ActiveEndpoint(actor)); + Assert.False(LiveOn(Backup)); + } + + /// + /// Anti-flap: right after a ServiceLevel switch the roles flip again. The connection does not + /// bounce straight back — not even through the connect-time check — until the dwell has passed; + /// then it does. + /// + [Fact] + public void Dwell_BlocksAnImmediateSwitchBack_ThenAllowsIt() + { + _rig.SetLevel(Primary, 90); + _rig.SetLevel(Backup, 250); + var actor = Start("sl-dwell", Config(Primary, windowSeconds: 1, dwellSeconds: 8), Config(Backup)); + AwaitCondition(() => LiveOn(Backup), TimeSpan.FromSeconds(10)); + var switchedAt = Stopwatch.StartNew(); + + // Roles flip back at once. The dwell runs from the switch decision, which comes before the + // observed re-subscribe on the backup by a reconnect + subscribe round trip, so the + // "blocked" window checked here (3 s) leaves wide slack inside the 8 s dwell under load. + _rig.SetLevel(Primary, 250); + _rig.SetLevel(Backup, 90); + var primaryConnectsAfterSwitch = _rig.ConnectCount(Primary); + + Thread.Sleep(3000); + Assert.Equal("Backup", ActiveEndpoint(actor)); + Assert.Equal(primaryConnectsAfterSwitch, _rig.ConnectCount(Primary)); // not even probed + + // Positive control: once the dwell is over, it does switch back. + AwaitCondition(() => ActiveEndpoint(actor) == "Primary", TimeSpan.FromSeconds(20)); + Assert.True(switchedAt.Elapsed >= TimeSpan.FromSeconds(5), + $"switched back after {switchedAt.Elapsed.TotalSeconds:F2}s, well inside the 8s dwell"); + } + + /// + /// ServiceLevelFailover.Enabled=False on the primary: the ServiceLevel is never even + /// read, whatever it is. + /// + [Fact] + public void Disabled_NeverReadsTheServiceLevel() + { + _rig.SetLevel(Primary, 90); + _rig.SetLevel(Backup, 250); + var actor = Start("sl-disabled", Config(Primary, enabled: false, windowSeconds: 1), Config(Backup)); + + Thread.Sleep(1500); + Assert.Equal(0, _rig.ReadCount(Primary)); + Assert.Equal(0, _rig.ConnectCount(Backup)); + Assert.Equal("Primary", ActiveEndpoint(actor)); + } + + /// Only the PRIMARY's copy of the setting counts: disabling it on the backup changes nothing. + [Fact] + public void BackupCopyOfTheSetting_IsIgnored() + { + _rig.SetLevel(Primary, 90); + _rig.SetLevel(Backup, 250); + Start("sl-backup-copy", Config(Primary, windowSeconds: 1), Config(Backup, enabled: false)); + + AwaitCondition(() => LiveOn(Backup), TimeSpan.FromSeconds(5)); + } + + /// A connection without a backup has nowhere to go: the ServiceLevel is never read. + [Fact] + public void NoBackup_NeverReadsTheServiceLevel() + { + _rig.SetLevel(Primary, 90); + var actor = Start("sl-nobackup", Config(Primary, windowSeconds: 1), backupConfig: null); + + Thread.Sleep(1500); + Assert.True(LiveOn(Primary)); // positive control: the session is up and subscribed + Assert.Equal(0, _rig.ReadCount(Primary)); + Assert.Equal("Primary (no backup)", ActiveEndpoint(actor)); + } + + /// + /// Symmetry: sitting on the backup (after a connectivity failover) when the backup becomes the + /// Warm standby moves the connection back to the primary. + /// + [Fact] + public void OnBackup_DegradedBackup_FailsBackToAHealthyPrimary() + { + _rig.SetLevel(Primary, 90); + _rig.SetLevel(Backup, 250); + var actor = Start("sl-symmetric", Config(Primary, windowSeconds: 1, dwellSeconds: 0), Config(Backup)); + AwaitCondition(() => LiveOn(Backup), TimeSpan.FromSeconds(5)); + var primarySubscribes = _rig.Subscribes.Count(s => s.StartsWith(Primary + "|", StringComparison.Ordinal)); + + _rig.SetLevel(Primary, 250); + _rig.SetLevel(Backup, 90); + + AwaitCondition(() => _rig.Subscribes.Count(s => s.StartsWith(Primary + "|", StringComparison.Ordinal)) > primarySubscribes, + TimeSpan.FromSeconds(6)); + Assert.Equal("Primary", ActiveEndpoint(actor)); + } + + /// + /// An adapter without (MxGateway, or any plain adapter) is + /// never watched and never probed, even with a backup and the feature enabled. + /// + [Fact] + public void AdapterWithoutTheCapability_IsNeverWatched() + { + var plain = Substitute.For(); + plain.ConnectAsync(Arg.Any>(), Arg.Any()).Returns(Task.CompletedTask); + plain.SubscribeAsync(Arg.Any(), Arg.Any(), Arg.Any()) + .Returns("sub-1"); + plain.ReadAsync(Arg.Any(), Arg.Any()) + .Returns(new ReadResult(true, new TagValue(1, QualityCode.Good, DateTimeOffset.UtcNow), null)); + _rig.SetLevel(Backup, 250); + var actor = Start("sl-plain", Config(Primary, windowSeconds: 1, dwellSeconds: 0), Config(Backup), plain); + + Thread.Sleep(1500); + Assert.Equal(0, _rig.ConnectCount(Backup)); + Assert.Equal("Primary", ActiveEndpoint(actor)); + } + + /// The policy parser: defaults when absent, and clamps nonsense to safe values. + [Fact] + public void Policy_DefaultsAndClamps() + { + var defaults = ServiceLevelFailoverPolicy.From(new Dictionary { ["Endpoint"] = Primary }); + Assert.True(defaults.Enabled); + Assert.Equal(TimeSpan.FromSeconds(10), defaults.DegradedWindow); + Assert.Equal(TimeSpan.FromSeconds(60), defaults.MinDwell); + + var clamped = ServiceLevelFailoverPolicy.From(Config(Primary, enabled: false, windowSeconds: 0, dwellSeconds: -5)); + Assert.False(clamped.Enabled); + Assert.Equal(TimeSpan.FromSeconds(1), clamped.DegradedWindow); + Assert.Equal(TimeSpan.Zero, clamped.MinDwell); + + Assert.True(ServiceLevelFailoverPolicy.From(null).Enabled); + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Actors/FakeServiceLevelConnection.cs b/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Actors/FakeServiceLevelConnection.cs new file mode 100644 index 00000000..0e67a667 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Actors/FakeServiceLevelConnection.cs @@ -0,0 +1,119 @@ +using System.Collections.Concurrent; +using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Protocol; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; + +namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests.Actors; + +/// +/// The two OPC UA servers of a redundant pair as the ScadaBridge#138 tests see them: a +/// ServiceLevel per endpoint (changeable at any time, null = unreadable), whether each endpoint +/// accepts a session, and a record of what the actor did to them. +/// +internal sealed class ServiceLevelRig +{ + private readonly ConcurrentDictionary> _levels = new(); + private readonly ConcurrentDictionary _unreachable = new(); + + /// Every endpoint a session was opened on, in order — live sessions and probes alike. + public ConcurrentQueue Connects { get; } = new(); + + /// Every tag subscribed, as "endpoint|tag" — only a live session receives these. + public ConcurrentQueue Subscribes { get; } = new(); + + /// Number of ServiceLevel reads, per endpoint. + public ConcurrentDictionary Reads { get; } = new(); + + /// Sets a fixed ServiceLevel for . + public void SetLevel(string endpoint, byte? level) => _levels[endpoint] = () => level; + + /// Sets a ServiceLevel source evaluated on every read of . + public void SetLevel(string endpoint, Func level) => _levels[endpoint] = level; + + /// Makes refuse (or accept again) new sessions. + public void SetReachable(string endpoint, bool reachable) => _unreachable[endpoint] = !reachable; + + public bool IsReachable(string endpoint) => !_unreachable.GetValueOrDefault(endpoint); + + public byte? Read(string endpoint) + { + Reads.AddOrUpdate(endpoint, 1, (_, n) => n + 1); + return _levels.TryGetValue(endpoint, out var level) ? level() : null; + } + + public int ConnectCount(string endpoint) => Connects.Count(e => e == endpoint); + + public int ReadCount(string endpoint) => Reads.GetValueOrDefault(endpoint); +} + +/// +/// An with the capability, bound to +/// one endpoint of a . Subscriptions and reads always succeed. +/// +internal sealed class FakeServiceLevelConnection : IDataConnection, IServiceLevelReadable +{ + private readonly ServiceLevelRig _rig; + private volatile bool _connected; + + public FakeServiceLevelConnection(ServiceLevelRig rig, string endpoint) + { + _rig = rig; + Endpoint = endpoint; + } + + public string Endpoint { get; } + + public ConnectionHealth Status => _connected ? ConnectionHealth.Connected : ConnectionHealth.Disconnected; + + public event Action? Disconnected; + + public Task ConnectAsync(IDictionary connectionDetails, CancellationToken cancellationToken = default) + { + _rig.Connects.Enqueue(Endpoint); + if (!_rig.IsReachable(Endpoint)) + return Task.FromException(new InvalidOperationException($"{Endpoint} refused the session")); + _connected = true; + return Task.CompletedTask; + } + + public Task DisconnectAsync(CancellationToken cancellationToken = default) + { + _connected = false; + return Task.CompletedTask; + } + + public Task ReadServiceLevelAsync(CancellationToken cancellationToken = default) => + Task.FromResult(_connected ? _rig.Read(Endpoint) : null); + + public Task SubscribeAsync(string tagPath, SubscriptionCallback callback, CancellationToken cancellationToken = default) + { + _rig.Subscribes.Enqueue($"{Endpoint}|{tagPath}"); + return Task.FromResult($"{Endpoint}:{tagPath}"); + } + + public Task UnsubscribeAsync(string subscriptionId, CancellationToken cancellationToken = default) => Task.CompletedTask; + + public Task ReadAsync(string tagPath, CancellationToken cancellationToken = default) => + Task.FromResult(new ReadResult(true, new TagValue(1, QualityCode.Good, DateTimeOffset.UtcNow), null)); + + public Task> ReadBatchAsync(IEnumerable tagPaths, CancellationToken cancellationToken = default) => + Task.FromResult>(tagPaths.ToDictionary( + t => t, _ => new ReadResult(true, new TagValue(1, QualityCode.Good, DateTimeOffset.UtcNow), null))); + + public Task WriteAsync(string tagPath, object? value, CancellationToken cancellationToken = default) => + Task.FromResult(new WriteResult(true, null)); + + public Task> WriteBatchAsync(IDictionary values, CancellationToken cancellationToken = default) => + Task.FromResult>(values.ToDictionary(v => v.Key, _ => new WriteResult(true, null))); + + public void RaiseDisconnected() + { + _connected = false; + Disconnected?.Invoke(); + } + + public ValueTask DisposeAsync() + { + _connected = false; + return ValueTask.CompletedTask; + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Adapters/OpcUaServiceLevelTests.cs b/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Adapters/OpcUaServiceLevelTests.cs new file mode 100644 index 00000000..19ce98fb --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Adapters/OpcUaServiceLevelTests.cs @@ -0,0 +1,211 @@ +using Akka.Actor; +using Akka.TestKit.Xunit2; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; +using NSubstitute; +using NSubstitute.ExceptionExtensions; +using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.DataConnection; +using ZB.MOM.WW.ScadaBridge.Commons.Serialization; +using ZB.MOM.WW.ScadaBridge.Commons.Types.DataConnections; +using ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Actors; +using ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Adapters; +using ZB.MOM.WW.ScadaBridge.HealthMonitoring; + +namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests.Adapters; + +/// +/// ScadaBridge#138 at the OPC UA adapter: +/// against a substitute client (status / failure mapping) and against real in-process OPC UA +/// servers (the actual i=2267 node), plus the whole failover through the real +/// against a real server pair. +/// +public sealed class OpcUaServiceLevelTests : TestKit +{ + private readonly string _clientPki = + Path.Combine(Path.GetTempPath(), "sb138-client", Guid.NewGuid().ToString("N")); + + /// Initializes the test kit. + public OpcUaServiceLevelTests() : base(@"akka.loglevel = DEBUG") { } + + /// + protected override void Dispose(bool disposing) + { + base.Dispose(disposing); + try { Directory.Delete(_clientPki, recursive: true); } catch { /* best-effort */ } + } + + private OpcUaGlobalOptions ClientOptions() => new() + { + ApplicationName = "SB138Client", + ApplicationUri = "urn:localhost:SB138Client", + OwnCertificateStorePath = Path.Combine(_clientPki, "own"), + TrustedPeerStorePath = Path.Combine(_clientPki, "trusted"), + TrustedIssuerStorePath = Path.Combine(_clientPki, "issuers"), + RejectedCertificateStorePath = Path.Combine(_clientPki, "rejected"), + AutoCreateApplicationCertificate = true, + }; + + private static IDictionary Flat(string url, OpcUaServiceLevelFailoverConfig? sl = null) => + OpcUaEndpointConfigSerializer.ToFlatDict(new OpcUaEndpointConfig + { + EndpointUrl = url, + SecurityMode = OpcUaSecurityMode.None, + ServiceLevelFailover = sl, + }); + + private static (OpcUaDataConnection Adapter, IOpcUaClient Client) WithSubstituteClient() + { + var client = Substitute.For(); + client.IsConnected.Returns(true); + var factory = Substitute.For(); + factory.Create().Returns(client); + return (new OpcUaDataConnection(factory, NullLogger.Instance), client); + } + + // ── Status / failure mapping (substitute client) ── + + /// The value comes from i=2267; a Good read of a byte is that byte. + [Fact] + public async Task Read_GoodByte_IsTheLevel_FromTheServiceLevelNode() + { + var (adapter, client) = WithSubstituteClient(); + client.ReadValueAsync("i=2267", Arg.Any()) + .Returns(((object?)(byte)90, DateTime.UtcNow, 0u)); + await adapter.ConnectAsync(new Dictionary()); + + Assert.Equal((byte)90, await adapter.ReadServiceLevelAsync()); + } + + /// A Bad status, or a value that is not a byte-ranged integer, is unknown (null). + [Fact] + public async Task Read_BadStatusOrNonByteValue_IsUnknown() + { + var (adapter, client) = WithSubstituteClient(); + await adapter.ConnectAsync(new Dictionary()); + + client.ReadValueAsync("i=2267", Arg.Any()) + .Returns(((object?)(byte)250, DateTime.UtcNow, 0x80340000u)); // BadNodeIdUnknown + Assert.Null(await adapter.ReadServiceLevelAsync()); + + client.ReadValueAsync("i=2267", Arg.Any()) + .Returns(((object?)"250", DateTime.UtcNow, 0u)); + Assert.Null(await adapter.ReadServiceLevelAsync()); + + client.ReadValueAsync("i=2267", Arg.Any()) + .Returns(((object?)300, DateTime.UtcNow, 0u)); + Assert.Null(await adapter.ReadServiceLevelAsync()); + + client.ReadValueAsync("i=2267", Arg.Any()) + .Returns(((object?)240, DateTime.UtcNow, 0u)); // Int32 in range → accepted + Assert.Equal((byte)240, await adapter.ReadServiceLevelAsync()); + } + + /// + /// A failed read is unknown and — unlike — never + /// raises Disconnected: the poll must not drive the reconnect machine. + /// + [Fact] + public async Task Read_Failure_IsUnknown_AndNeverRaisesDisconnected() + { + var (adapter, client) = WithSubstituteClient(); + client.ReadValueAsync("i=2267", Arg.Any()) + .ThrowsAsync(new InvalidOperationException("session dropped")); + var disconnected = 0; + adapter.Disconnected += () => disconnected++; + await adapter.ConnectAsync(new Dictionary()); + + Assert.Null(await adapter.ReadServiceLevelAsync()); + Assert.Equal(0, disconnected); + + // Positive control: the ordinary read path DOES raise it for the same failure. + client.ReadValueAsync("ns=2;s=x", Arg.Any()) + .ThrowsAsync(new InvalidOperationException("session dropped")); + await Assert.ThrowsAsync(() => adapter.ReadAsync("ns=2;s=x")); + Assert.Equal(1, disconnected); + } + + /// Before connect (no client) the level is unknown, and no read is attempted. + [Fact] + public async Task Read_NotConnected_IsUnknown() + { + var (adapter, client) = WithSubstituteClient(); + Assert.Null(await adapter.ReadServiceLevelAsync()); + + await adapter.ConnectAsync(new Dictionary()); + client.IsConnected.Returns(false); + Assert.Null(await adapter.ReadServiceLevelAsync()); + await client.DidNotReceive().ReadValueAsync(Arg.Any(), Arg.Any()); + } + + // ── Real OPC UA servers ── + + /// The real adapter reads the real server's Server_ServiceLevel as it changes. + [Fact] + public async Task RealServer_ReadServiceLevel_TracksTheServersValue() + { + await using var server = await SecureOpcUaTestServer.StartAsync(); + await using var adapter = new OpcUaDataConnection( + new RealOpcUaClientFactory(ClientOptions()), NullLogger.Instance); + await adapter.ConnectAsync(Flat(server.Url)); + + foreach (var level in new byte[] { 250, 240, 90, 50, 1 }) + { + server.SetServiceLevel(level); + Assert.Equal(level, await adapter.ReadServiceLevelAsync()); + } + } + + /// + /// End to end on two real servers through the production factory: the connection starts on a + /// primary that is the Warm standby (90) while the backup is Active (250), and ends up on the + /// backup. Then the mirror case on a fresh connection — today's 250/240 pair — stays on the + /// primary. + /// + [Fact] + public async Task RealServerPair_StandbyPrimary_FailsOverToTheActiveBackup_ActivePair_StaysPut() + { + await using var a = await SecureOpcUaTestServer.StartAsync(); + await using var b = await SecureOpcUaTestServer.StartAsync(); + var factory = new DataConnectionFactory(NullLoggerFactory.Instance, Options.Create(ClientOptions())); + var options = new DataConnectionOptions + { + ReconnectInterval = TimeSpan.FromMilliseconds(200), + ServiceLevelPollInterval = TimeSpan.FromMilliseconds(100), + ServiceLevelProbeTimeout = TimeSpan.FromSeconds(20), + }; + var sl = new OpcUaServiceLevelFailoverConfig { DegradedWindowSeconds = 1, MinDwellSeconds = 0 }; + + async Task Endpoint(IActorRef actor) => + (await actor.Ask(new DataConnectionActor.GetHealthReport(), TimeSpan.FromSeconds(5))) + .ActiveEndpoint; + + // Warm: A is the standby, B the Active. + a.SetServiceLevel(90); + b.SetServiceLevel(250); + var health = Substitute.For(); + var primary = Flat(a.Url, sl); + var warm = Sys.ActorOf(Props.Create(() => new DataConnectionActor( + "warm", factory.Create("OpcUa", primary), options, health, factory, "OpcUa", + primary, Flat(b.Url), 3, null))); + await AwaitConditionAsync(async () => await Endpoint(warm) == "Backup", TimeSpan.FromSeconds(30)); + AwaitCondition(() => health.ReceivedCalls().Any(c => + c.GetMethodInfo().Name == nameof(ISiteHealthCollector.UpdateConnectionEndpoint) + && c.GetArguments() is ["warm", "Connected to backup"]), TimeSpan.FromSeconds(30)); + Sys.Stop(warm); + + // Today: A is the Secondary (240), B the Primary (250) — both Active band. + a.SetServiceLevel(240); + var today = Sys.ActorOf(Props.Create(() => new DataConnectionActor( + "today", factory.Create("OpcUa", primary), options, health, factory, "OpcUa", + primary, Flat(b.Url), 3, null))); + AwaitCondition(() => health.ReceivedCalls().Any(c => + c.GetMethodInfo().Name == nameof(ISiteHealthCollector.UpdateConnectionEndpoint) + && c.GetArguments() is ["today", "Connected to primary"]), TimeSpan.FromSeconds(30)); + await Task.Delay(TimeSpan.FromSeconds(2.5)); // > window, ~25 polls + Assert.Equal("Primary", await Endpoint(today)); + Assert.DoesNotContain(health.ReceivedCalls(), c => + c.GetMethodInfo().Name == nameof(ISiteHealthCollector.UpdateConnectionEndpoint) + && c.GetArguments() is ["today", "Connected to backup"]); + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Adapters/RealOpcUaPromotionRefreshTests.cs b/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Adapters/RealOpcUaPromotionRefreshTests.cs new file mode 100644 index 00000000..2bfb9a75 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Adapters/RealOpcUaPromotionRefreshTests.cs @@ -0,0 +1,112 @@ +using System.Collections.Concurrent; +using System.Reflection; +using Opc.Ua; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Alarms; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; +using ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Adapters; + +namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests.Adapters; + +/// +/// ScadaBridge#138 / OtOpcUa Warm design §4.5 + §5 "Alarms are compatible": when a Warm standby +/// is promoted, OtOpcUa runs a ConditionRefresh on every live event item ON ITS OWN — the client +/// never asked. The existing items therefore receive an unsolicited RefreshStart, the retained +/// conditions, RefreshEnd, then new events. ScadaBridge must treat that exactly like the refresh it +/// triggers itself on subscribe: one framed snapshot with its own epoch, and live derivation +/// before and after it. +/// +public class RealOpcUaPromotionRefreshTests +{ + private const string Handle = "h1"; + private const string Source = "ns=2;s=Tank01"; + + private static readonly MethodInfo Handler = + typeof(RealOpcUaClient).GetMethod("HandleAlarmEvent", BindingFlags.Instance | BindingFlags.NonPublic)!; + + /// What CreateAlarmSubscriptionAsync sets up for a live handle: its per-condition state map. + private static void SeedSubscribedHandle(RealOpcUaClient client) + { + var field = typeof(RealOpcUaClient).GetField("_alarmLastState", BindingFlags.Instance | BindingFlags.NonPublic)!; + var map = (ConcurrentDictionary>)field.GetValue(client)!; + map[Handle] = new Dictionary(StringComparer.Ordinal); + } + + private static void Boundary(RealOpcUaClient client, NodeId type, List sink) => + Handler.Invoke(client, [Handle, Source, + new EventFieldList { EventFields = new Variant[] { Variant.From(type), Variant.Null, Variant.Null, Variant.Null, Variant.Null, Variant.Null } }, + (Action)sink.Add]); + + private static void Condition(RealOpcUaClient client, string condition, bool active, bool acked, List sink) + { + var fields = new Variant[] + { + Variant.From(ObjectTypeIds.AlarmConditionType), // 0 EventType + Variant.From(new NodeId("Tank01", 2)), // 1 SourceNode + Variant.From("Tank01"), // 2 SourceName + Variant.From(DateTime.UtcNow), // 3 Time + Variant.From(new LocalizedText("msg")), // 4 Message + Variant.From((ushort)500), // 5 Severity + Variant.From(active), // 6 ActiveState/Id + Variant.From(acked), // 7 AckedState/Id + Variant.Null, // 8 ConfirmedState/Id + Variant.From(false), // 9 SuppressedOrShelved + Variant.Null, // 10 ShelvingState + Variant.From(condition), // 11 ConditionName + }; + Handler.Invoke(client, [Handle, Source, new EventFieldList { EventFields = fields }, + (Action)sink.Add]); + } + + /// + /// Subscribe-time refresh, live traffic, then the UNSOLICITED promotion refresh on the same + /// item, then live traffic again. The promotion replay is framed SnapshotStart → Snapshot… → + /// SnapshotComplete under a fresh epoch (distinct from the subscribe-time one), and live events + /// after it are derived live again — the replay does not leave the item stuck in Snapshot mode. + /// + [Fact] + public void UnsolicitedPromotionRefresh_IsFramedAsItsOwnSnapshot_AndLiveDerivationResumes() + { + var client = new RealOpcUaClient(); + SeedSubscribedHandle(client); + var sink = new List(); + + // Subscribe-time refresh (the one ScadaBridge triggers): Hi is active. + Boundary(client, ObjectTypeIds.RefreshStartEventType, sink); + Condition(client, "Hi", active: true, acked: false, sink); + Boundary(client, ObjectTypeIds.RefreshEndEventType, sink); + + // Live while the old Active serves: Lo raises. + Condition(client, "Lo", active: true, acked: false, sink); + var liveBefore = sink[^1]; + + // Promotion on the other node: an unsolicited refresh on the existing item. + Boundary(client, ObjectTypeIds.RefreshStartEventType, sink); + Condition(client, "Hi", active: true, acked: false, sink); + Condition(client, "Lo", active: true, acked: false, sink); + Boundary(client, ObjectTypeIds.RefreshEndEventType, sink); + + // Live after promotion: Lo is acknowledged. + Condition(client, "Lo", active: true, acked: true, sink); + var liveAfter = sink[^1]; + + Assert.Equal( + ["SnapshotStart", "Snapshot", "SnapshotComplete", liveBefore.Kind.ToString(), + "SnapshotStart", "Snapshot", "Snapshot", "SnapshotComplete", liveAfter.Kind.ToString()], + sink.Select(t => t.Kind.ToString())); + + // Live kinds really are live (not Snapshot) on both sides of the promotion refresh. + Assert.Equal(AlarmTransitionKind.Raise, liveBefore.Kind); + Assert.Equal(AlarmTransitionKind.Acknowledge, liveAfter.Kind); + Assert.Null(liveBefore.SnapshotId); + Assert.Null(liveAfter.SnapshotId); + + // The promotion replay has its own epoch, shared by its boundaries and its conditions. + var subscribeEpoch = sink[0].SnapshotId; + var promotionEpoch = sink[4].SnapshotId; + Assert.NotNull(subscribeEpoch); + Assert.NotNull(promotionEpoch); + Assert.NotEqual(subscribeEpoch, promotionEpoch); + Assert.All(sink.Skip(4).Take(4), t => Assert.Equal(promotionEpoch, t.SnapshotId)); + Assert.All(sink.Skip(4).Take(4), t => Assert.Equal(Source, t.SnapshotSourceReference)); + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Adapters/SecureOpcUaTestServer.cs b/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Adapters/SecureOpcUaTestServer.cs index b932ce82..c12554a5 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Adapters/SecureOpcUaTestServer.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Adapters/SecureOpcUaTestServer.cs @@ -18,11 +18,14 @@ namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests.Adapters; internal sealed class SecureOpcUaTestServer : IAsyncDisposable { private readonly ApplicationInstance _app; + private readonly StandardServer _server; private readonly string _pkiRoot; - private SecureOpcUaTestServer(ApplicationInstance app, string pkiRoot, string url, X509Certificate2 certificate) + private SecureOpcUaTestServer( + ApplicationInstance app, StandardServer server, string pkiRoot, string url, X509Certificate2 certificate) { _app = app; + _server = server; _pkiRoot = pkiRoot; Url = url; Certificate = certificate; @@ -110,16 +113,25 @@ internal sealed class SecureOpcUaTestServer : IAsyncDisposable ApplicationType = ApplicationType.Server, }; await app.CheckApplicationInstanceCertificatesAsync(silent: true, lifeTimeInMonths: 12); - await app.StartAsync(new TestServer(telemetry, apiKey)); + var server = new TestServer(telemetry, apiKey); + await app.StartAsync(server); // SDK 2: the loaded application certificate lives in the certificate manager, not on the identifier. using var entry = config.CertificateManager.AcquireApplicationCertificateByType( ObjectTypeIds.RsaSha256ApplicationCertificateType) ?? throw new InvalidOperationException("The test server has no application certificate."); var cert = X509CertificateLoader.LoadCertificate(entry.Certificate.RawData); - return new SecureOpcUaTestServer(app, pkiRoot, url, cert); + return new SecureOpcUaTestServer(app, server, pkiRoot, url, cert); } + /// + /// Sets the server's Server_ServiceLevel (i=2267) — what a redundant server + /// publishes for its role (ScadaBridge#138: 250/240 Active, 90/50 Warm standby, 1 booting). + /// + /// The ServiceLevel to publish. + public void SetServiceLevel(byte level) => + _server.CurrentInstance.ServerObject.ServiceLevel!.Value = level; + /// public async ValueTask DisposeAsync() {