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
2026-11-13 03:06:45 -05:00
parent 16209831ca
commit 712c5117ee
23 changed files with 1760 additions and 36 deletions
@@ -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<string, string> 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
@@ -190,6 +190,8 @@ scadabridge data-connection delete --id <id>
 scadabridge data-connection move --id <id> --site-id <destination-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
@@ -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.
@@ -165,6 +165,40 @@
                               min="1" max="20" @bind="_formFailoverRetryCount" />
                        <div class="form-text">Retries before failing over to backup endpoint.</div>
                    </div>
                    @if (_protocol != "MxGateway")
                    {
                        <div class="mb-2" data-test="servicelevel-failover">
                            <div class="form-check">
                                <input type="checkbox" class="form-check-input" id="sl-failover-enabled"
                                       @bind="_formSlEnabled" />
                                <label class="form-check-label small" for="sl-failover-enabled">
                                    ServiceLevel failover
                                </label>
                            </div>
                            <div class="form-text">
                                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.
                            </div>
                            @if (_formSlEnabled)
                            {
                                <div class="row g-2 mt-1">
                                    <div class="col-auto">
                                        <label class="form-label small" for="sl-failover-window">Degraded window (s)</label>
                                        <input type="number" id="sl-failover-window" class="form-control form-control-sm"
                                               style="max-width: 120px;" min="1" max="3600" @bind="_formSlWindowSeconds" />
                                        @RenderPrimaryError("ServiceLevelFailover.DegradedWindowSeconds")
                                    </div>
                                    <div class="col-auto">
                                        <label class="form-label small" for="sl-failover-dwell">Minimum dwell (s)</label>
                                        <input type="number" id="sl-failover-dwell" class="form-control form-control-sm"
                                               style="max-width: 120px;" min="0" max="86400" @bind="_formSlDwellSeconds" />
                                        @RenderPrimaryError("ServiceLevelFailover.MinDwellSeconds")
                                    </div>
                                </div>
                            }
                        </div>
                    }
                    <div class="mb-0">
                        <button type="button" class="btn btn-outline-danger btn-sm"
                                @onclick="RemoveBackup">Remove Backup</button>
@@ -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
            : @<div class="text-danger small">@match.Message</div>;
    }
    private void GoBack() => NavigationManager.NavigateTo("/design/connections");
@@ -0,0 +1,22 @@
 namespace ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Protocol;
 /// <summary>
 /// Optional capability for an <see cref="IDataConnection"/> whose server publishes a redundancy
 /// ServiceLevel (OPC UA <c>Server_ServiceLevel</c>, <c>i=2267</c>). 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.
 /// </summary>
 public interface IServiceLevelReadable
 {
    /// <summary>
    /// Reads the server's current ServiceLevel. Never raises the connection's
    /// <see cref="IDataConnection.Disconnected"/> event: a lost session is the reconnect machine's
    /// business, not this read's.
    /// </summary>
    /// <param name="cancellationToken">Bounds the read.</param>
    /// <returns>
    /// 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.
    /// </returns>
    Task<byte?> ReadServiceLevelAsync(CancellationToken cancellationToken = default);
 }
@@ -83,6 +83,15 @@ public readonly record struct OpcUaConfigParseResult
 /// </summary>
 public static class OpcUaEndpointConfigSerializer
 {
    /// <summary>Flat-dict key for <see cref="OpcUaServiceLevelFailoverConfig.Enabled"/>.</summary>
    public const string ServiceLevelFailoverEnabledKey = "ServiceLevelFailover.Enabled";
    /// <summary>Flat-dict key for <see cref="OpcUaServiceLevelFailoverConfig.DegradedWindowSeconds"/>.</summary>
    public const string ServiceLevelFailoverWindowKey = "ServiceLevelFailover.DegradedWindowSeconds";
    /// <summary>Flat-dict key for <see cref="OpcUaServiceLevelFailoverConfig.MinDwellSeconds"/>.</summary>
    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;
    }
@@ -96,4 +96,12 @@ public sealed class OpcUaEndpointConfig
    /// Heartbeat configuration for connection monitoring.
    /// </summary>
    public OpcUaHeartbeatConfig? Heartbeat { get; set; }
    // Redundancy (optional; null = defaults, see OpcUaServiceLevelFailoverConfig)
    /// <summary>
    /// 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).
    /// </summary>
    public OpcUaServiceLevelFailoverConfig? ServiceLevelFailover { get; set; }
 }
@@ -0,0 +1,46 @@
 namespace ZB.MOM.WW.ScadaBridge.Commons.Types.DataConnections;
 /// <summary>
 /// 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
 /// <b>primary</b> endpoint's copy is read, and it has no effect on a connection without a backup.
 ///
 /// <para>
 /// The connected endpoint's <c>Server_ServiceLevel</c> (<c>i=2267</c>) is polled. When it stays in
 /// the Degraded band (below 200, OPC UA Part 4 §6.6.2.4.2) for <see cref="DegradedWindowSeconds"/>
 /// 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
 /// <see cref="MinDwellSeconds"/>. 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.
 /// </para>
 ///
 /// <para>
 /// 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.
 /// </para>
 /// </summary>
 public sealed class OpcUaServiceLevelFailoverConfig
 {
    /// <summary>Default for <see cref="DegradedWindowSeconds"/>.</summary>
    public const int DefaultDegradedWindowSeconds = 10;
    /// <summary>Default for <see cref="MinDwellSeconds"/>.</summary>
    public const int DefaultMinDwellSeconds = 60;
    /// <summary>Whether ServiceLevel can trigger a failover. Default true.</summary>
    public bool Enabled { get; set; } = true;
    /// <summary>
    /// How long, in seconds, the connected endpoint must report a Degraded-band ServiceLevel
    /// without a break before the other endpoint is checked. Default 10.
    /// </summary>
    public int DegradedWindowSeconds { get; set; } = DefaultDegradedWindowSeconds;
    /// <summary>
    /// Minimum time, in seconds, after a ServiceLevel-driven switch before another one may happen
    /// (anti-flap). Default 60.
    /// </summary>
    public int MinDwellSeconds { get; set; } = DefaultMinDwellSeconds;
 }
@@ -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());
@@ -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;
 /// <summary>
 /// ScadaBridge#138 — ServiceLevel-aware failover between a primary and a backup endpoint.
 ///
 /// <para>
 /// <b>Why.</b> Until now the connection failed over only when the link itself misbehaved
 /// (<c>FailoverRetryCount</c> consecutive connect failures or unstable disconnects). Under OtOpcUa
 /// Warm redundancy (lmxopcua#717) the standby keeps a perfectly healthy session but serves
 /// <c>BadOutOfService</c> data at a standby ServiceLevel (90 healthy / 50 degraded / 1 booting), so a
 /// connection that lands there would sit on Bad quality indefinitely.
 /// </para>
 ///
 /// <para>
 /// <b>What.</b> While Connected, the connected endpoint's <c>Server_ServiceLevel</c> is polled every
 /// <see cref="DataConnectionOptions.ServiceLevelPollInterval"/>. A level below
 /// <see cref="HealthyServiceLevelMinimum"/> (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.
 /// </para>
 ///
 /// <para>
 /// <b>Unchanged behaviour.</b> 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
 /// <see cref="IServiceLevelReadable"/> (MxGateway) and a connection without a backup are never
 /// monitored at all. The switch itself reuses <see cref="SwitchToOtherEndpoint"/> and the reconnect
 /// cycle, exactly like the connection-driven failovers.
 /// </para>
 /// </summary>
 public partial class DataConnectionActor
 {
    /// <summary>The lowest ServiceLevel in the OPC UA Healthy band (Part 4 §6.6.2.4.2).</summary>
    internal const byte HealthyServiceLevelMinimum = 200;
    private const string ServiceLevelPollTimerKey = "servicelevel-poll";
    private ServiceLevelFailoverPolicy _serviceLevelPolicy;
    /// <summary>When the current unbroken run of Degraded samples began; null when not degraded.</summary>
    private DateTimeOffset? _degradedSince;
    /// <summary>When the last ServiceLevel-driven switch happened (dwell / anti-flap).</summary>
    private DateTimeOffset _lastServiceLevelSwitchAt = DateTimeOffset.MinValue;
    private bool _serviceLevelReadInFlight;
    private bool _serviceLevelProbeInFlight;
    /// <summary>True until the first sample of a session is evaluated (connect-time preference).</summary>
    private bool _serviceLevelConnectCheckPending;
    /// <summary>Why <see cref="BecomeReconnecting"/> is re-establishing the link.</summary>
    internal enum ReconnectCause
    {
        /// <summary>The link was lost (disconnect, connection-level failure).</summary>
        ConnectionLost,
        /// <summary>An in-place reconfiguration swapped the adapter.</summary>
        Reconfigured,
        /// <summary>A sustained Degraded ServiceLevel swapped to the other endpoint (#138).</summary>
        ServiceLevelFailover,
    }
    /// <summary>
    /// 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.
    /// </summary>
    private bool ServiceLevelFailoverInForce =>
        _serviceLevelPolicy.Enabled && _backupConfig != null && _adapter is IServiceLevelReadable;
    /// <summary>Called on entering Connected. Arms the poll (first tick immediately).</summary>
    private void StartServiceLevelMonitoring()
    {
        ResetServiceLevelSession();
        if (!ServiceLevelFailoverInForce)
            return;
        _serviceLevelConnectCheckPending = true;
        Timers.StartPeriodicTimer(ServiceLevelPollTimerKey, ServiceLevelPollTick.Instance,
            TimeSpan.Zero, _options.ServiceLevelPollInterval);
    }
    /// <summary>Called on leaving Connected. Results still in flight are fenced by generation.</summary>
    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);
    }
    /// <summary>
    /// Reads the OTHER endpoint's ServiceLevel through a short-lived adapter built from its
    /// configuration — connect, read, dispose — off the actor thread, bounded by
    /// <see cref="DataConnectionOptions.ServiceLevelProbeTimeout"/>. The live adapter is untouched.
    /// </summary>
    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<string, string> 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);
    }
    /// <summary>Poll tick for the connected endpoint's ServiceLevel.</summary>
    internal sealed record ServiceLevelPollTick
    {
        /// <summary>The single instance.</summary>
        public static readonly ServiceLevelPollTick Instance = new();
    }
    /// <summary>One ServiceLevel sample of the connected endpoint (null = unknown).</summary>
    internal sealed record ServiceLevelSampled(int AdapterGeneration, byte? Level);
    /// <summary>The other endpoint's ServiceLevel (null = unreachable / unreadable, see Error).</summary>
    internal sealed record ServiceLevelProbed(int AdapterGeneration, byte CurrentLevel, byte? OtherLevel, string? Error);
 }
 /// <summary>
 /// The connection-level ServiceLevel failover settings, read from the PRIMARY endpoint's flattened
 /// configuration (<see cref="OpcUaServiceLevelFailoverConfig"/>). Absent keys mean the defaults.
 /// </summary>
 internal readonly record struct ServiceLevelFailoverPolicy(bool Enabled, TimeSpan DegradedWindow, TimeSpan MinDwell)
 {
    /// <summary>Parses the policy; malformed or out-of-range values fall back to safe values.</summary>
    /// <param name="primaryConfig">The primary endpoint's flattened configuration.</param>
    /// <returns>The policy in force.</returns>
    public static ServiceLevelFailoverPolicy From(IDictionary<string, string>? 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));
    }
 }
@@ -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).
 /// </summary>
-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);
    }
    /// <summary>
@@ -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 ──
-    /// <param name="afterReconfigure">
+    /// <param name="cause">
-    /// True when this transition was caused by an in-place reconfiguration
+    /// Why the link is being re-established. <see cref="ReconnectCause.Reconfigured"/> (an
-    /// (<see cref="HandleReconfigure"/>) rather than by losing the link. It suppresses the
+    /// in-place reconfiguration, <see cref="HandleReconfigure"/>) and
-    /// unstable-disconnect accounting and the failover check — the adapter has just been
+    /// <see cref="ReconnectCause.ServiceLevelFailover"/> (ScadaBridge#138) both mean the adapter
-    /// swapped ON PURPOSE against a brand-new configuration, so counting it as an unstable
+    /// has just been swapped ON PURPOSE rather than the link lost. They suppress the
-    /// cycle would let a reconfigure trip a failover onto the new backup before the new primary
+    /// unstable-disconnect accounting and the failover check — counting a deliberate swap as an
-    /// has been tried even once — and it reconnects immediately instead of waiting out
+    /// 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
    /// <see cref="DataConnectionOptions.ReconnectInterval"/>, since there is nothing to wait for.
    /// </param>
-    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);
+            cause switch
-        // The session is over: abandon its re-seed (02/L4).
+            {
                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;
+                var previousEndpoint = SwitchToOtherEndpoint();
                _activeEndpoint = _activeEndpoint == ActiveEndpoint.Primary
                    ? ActiveEndpoint.Backup
                    : ActiveEndpoint.Primary;
                _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,
+                "connection", cause == ReconnectCause.ConnectionLost ? "Warning" : "Info", null, _connectionName,
-                afterReconfigure
+                cause switch
-                    ? "Connection reconfigured — reconnecting on the new configuration"
+                {
-                    : "Connection lost — entering reconnect cycle",
+                    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);
    }
    /// <summary>
    /// Flips <see cref="_activeEndpoint"/> to the other configured endpoint and swaps the adapter
    /// onto it — the one endpoint switch shared by every failover trigger: consecutive connect
    /// failures (<see cref="CountFailureAndMaybeFailover"/>), the unstable-connection pattern
    /// (<see cref="BecomeReconnecting"/>) and a sustained Degraded ServiceLevel (ScadaBridge#138).
    /// Callers reset whichever counters their trigger owns. Requires a backup configuration.
    /// </summary>
    /// <returns>The endpoint that was active before the switch.</returns>
    private ActiveEndpoint SwitchToOtherEndpoint()
    {
        var previousEndpoint = _activeEndpoint;
        _activeEndpoint = _activeEndpoint == ActiveEndpoint.Primary
            ? ActiveEndpoint.Backup
            : ActiveEndpoint.Primary;
        SwapAdapter(_activeEndpoint == ActiveEndpoint.Primary ? _primaryConfig : _backupConfig!);
        return previousEndpoint;
    }
    /// <summary>
@@ -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);
    }
    /// <summary>
@@ -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;
+            var previousEndpoint = SwitchToOtherEndpoint();
            _activeEndpoint = _activeEndpoint == ActiveEndpoint.Primary
                ? ActiveEndpoint.Backup
                : ActiveEndpoint.Primary;
            _consecutiveFailures = 0;
            SwapAdapter(_activeEndpoint == ActiveEndpoint.Primary ? _primaryConfig : _backupConfig);
            _log.Warning("[{0}] Failing over from {1} to {2}",
                _connectionName, previousEndpoint, _activeEndpoint);
@@ -21,8 +21,11 @@ namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Adapters;
 /// </summary>
 public class OpcUaDataConnection
    : IDataConnection, IBrowsableDataConnection, IAlarmSubscribableConnection, IAddressSpaceSearchable,
-      IBatchSubscribableConnection
+      IBatchSubscribableConnection, IServiceLevelReadable
 {
    /// <summary>The standard <c>Server_ServiceLevel</c> node (ns=0), OPC UA Part 5.</summary>
    internal const string ServiceLevelNodeId = "i=2267";
    private readonly IOpcUaClientFactory _clientFactory;
    private readonly ILogger<OpcUaDataConnection> _logger;
    private IOpcUaClient? _client;
@@ -291,6 +294,39 @@ public class OpcUaDataConnection
        }
    }
    /// <inheritdoc />
    /// <remarks>
    /// ScadaBridge#138. Unlike <see cref="ReadAsync"/>, a failure here never raises
    /// <see cref="Disconnected"/>: 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".
    /// </remarks>
    public async Task<byte?> 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;
        }
    }
    /// <inheritdoc />
    public async Task<IReadOnlyDictionary<string, ReadResult>> ReadBatchAsync(IEnumerable<string> tagPaths, CancellationToken cancellationToken = default)
    {
@@ -119,4 +119,19 @@ public class DataConnectionOptions
    /// <c>(SeedReadMaxAttempts - 1) × SeedReadRetryDelay</c> regardless of tag count.
    /// </summary>
    public TimeSpan SeedReadRetryDelay { get; set; } = TimeSpan.FromMilliseconds(250);
    /// <summary>
    /// ScadaBridge#138: how often a connected endpoint's ServiceLevel (<c>i=2267</c>) is read when
    /// ServiceLevel-aware failover is in force (OPC UA connection with a backup and the
    /// connection's <c>ServiceLevelFailover.Enabled</c>). One single-node Read per poll. Must be
    /// shorter than the smallest configured degraded window to make that window meaningful.
    /// </summary>
    public TimeSpan ServiceLevelPollInterval { get; set; } = TimeSpan.FromSeconds(2);
    /// <summary>
    /// 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.
    /// </summary>
    public TimeSpan ServiceLevelProbeTimeout { get; set; } = TimeSpan.FromSeconds(15);
 }
@@ -64,5 +64,11 @@ public sealed class DataConnectionOptionsValidator : OptionsValidatorBase<DataCo
        builder.RequireThat(options.MxAlarmStreamReopenDebounce >= 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}).");
    }
 }
@@ -167,4 +167,74 @@ public class DataConnectionFormTests : BunitContext
        Assert.Equal(60000,
            doc.RootElement.GetProperty("sessionTimeoutMs").GetInt32());
    }
    /// <summary>
    /// 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.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task Save_WithBackup_StoresServiceLevelFailoverOnThePrimary()
    {
        DataConnection? captured = null;
        await _siteRepo.AddDataConnectionAsync(Arg.Do<DataConnection>(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);
    }
    /// <summary>ScadaBridge#138: without a backup the setting is neither shown nor stored.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task Save_WithoutBackup_StoresNoServiceLevelFailover()
    {
        DataConnection? captured = null;
        await _siteRepo.AddDataConnectionAsync(Arg.Do<DataConnection>(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);
    }
 }
@@ -504,4 +504,58 @@ public class OpcUaEndpointConfigSerializerTests
        Assert.NotNull(config.UserIdentity);
        Assert.Equal(OpcUaUserTokenType.Anonymous, config.UserIdentity!.TokenType);
    }
    /// <summary>
    /// 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.
    /// </summary>
    [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);
    }
    /// <summary>
    /// ScadaBridge#138, upgrade safety: a row stored before the setting existed has no
    /// <c>serviceLevelFailover</c>; 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.
    /// </summary>
    [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);
    }
 }
@@ -220,4 +220,30 @@ public class OpcUaEndpointConfigValidatorTests
        var r = OpcUaEndpointConfigValidator.Validate(c, fieldPrefix: "Primary.");
        Assert.Contains(r.Errors, e => e.EntityName == "Primary.UserIdentity.Username");
    }
    /// <summary>ScadaBridge#138: the ServiceLevel failover window and dwell are range-checked, prefixed.</summary>
    /// <param name="window">Degraded window, seconds.</param>
    /// <param name="dwell">Minimum dwell, seconds.</param>
    /// <param name="badField">The field expected to fail, or null when valid.</param>
    [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);
    }
 }
@@ -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;
 /// <summary>
 /// 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 <c>RealOpcUaClient</c> as SnapshotStart / Snapshot /
 /// SnapshotComplete under a new epoch — see <c>RealOpcUaPromotionRefreshTests</c>) 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.
 /// </summary>
 public class DataConnectionPromotionRefreshTests : TestKit
 {
    private const string Source = "Tank01";
    private readonly ISiteHealthCollector _health = Substitute.For<ISiteHealthCollector>();
    private readonly IDataConnectionFactory _factory = Substitute.For<IDataConnectionFactory>();
    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);
    /// <summary>
    /// 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.
    /// </summary>
    [Fact]
    public void UnsolicitedPromotionSnapshot_IsForwardedInOrder_AndReconcilesTheCache()
    {
        AlarmTransitionCallback? cb = null;
        var adapter = Substitute.For<IDataConnection, IAlarmSubscribableConnection>();
        adapter.ConnectAsync(Arg.Any<IDictionary<string, string>>(), Arg.Any<CancellationToken>())
            .Returns(Task.CompletedTask);
        ((IAlarmSubscribableConnection)adapter)
            .SubscribeAlarmsAsync(Arg.Any<string>(), Arg.Any<string?>(),
                Arg.Do<AlarmTransitionCallback>(c => cb = c), Arg.Any<CancellationToken>())
            .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<SubscribeAlarmsResponse>(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<NativeAlarmTransitionUpdate>().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<NativeAlarmTransitionUpdate>(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<NativeAlarmTransitionUpdate>().Select(u => u.Transition).ToList();
        Assert.Single(replay.OfType<SubscribeAlarmsResponse>(), 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));
    }
 }
@@ -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;
 /// <summary>
 /// 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 <c>BadOutOfService</c> data at a standby ServiceLevel (90 / 50 /
 /// boot 1); the Active node is 200 or more (250 Primary / 240 Secondary today).
 ///
 /// <para>
 /// 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.
 /// </para>
 /// </summary>
 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<ISiteHealthCollector>();
    private readonly IDataConnectionFactory _factory = Substitute.For<IDataConnectionFactory>();
    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),
    };
    /// <summary>Wires the factory to build a rig-bound fake for whichever endpoint it is asked for.</summary>
    public DataConnectionServiceLevelFailoverTests()
        : base(@"akka.loglevel = DEBUG")
    {
        _factory.Create(Arg.Any<string>(), Arg.Any<IDictionary<string, string>>())
            .Returns(ci => new FakeServiceLevelConnection(_rig, ((IDictionary<string, string>)ci[1])["Endpoint"]));
    }
    private static Dictionary<string, string> Config(
        string endpoint, bool? enabled = null, int? windowSeconds = null, int? dwellSeconds = null)
    {
        var config = new Dictionary<string, string> { ["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<string, string> primaryConfig, IDictionary<string, string>? 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<SubscribeTagsResponse>(TimeSpan.FromSeconds(5));
        return actor;
    }
    private string ActiveEndpoint(IActorRef actor) =>
        actor.Ask<DataConnectionHealthReport>(new DataConnectionActor.GetHealthReport(), TimeSpan.FromSeconds(3))
            .GetAwaiter().GetResult().ActiveEndpoint;
    private bool LiveOn(string endpoint) => _rig.Subscribes.Contains($"{endpoint}|{Tag}");
    /// <summary>
    /// 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.
    /// </summary>
    [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));
    }
    /// <summary>
    /// 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.
    /// </summary>
    [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));
    }
    /// <summary>Booting node (ServiceLevel 1) on connect, Active other → leave it too.</summary>
    [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));
    }
    /// <summary>
    /// 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.
    /// </summary>
    /// <param name="primaryLevel">The connected (primary) endpoint's ServiceLevel.</param>
    /// <param name="backupLevel">The backup endpoint's ServiceLevel.</param>
    [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));
    }
    /// <summary>
    /// 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.
    /// </summary>
    /// <param name="primaryLevel">The connected (primary) endpoint's ServiceLevel.</param>
    /// <param name="backupLevel">The backup endpoint's ServiceLevel.</param>
    [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));
    }
    /// <summary>
    /// The other endpoint is down: stay on the degraded one (it still answers). Positive control:
    /// sessions to the backup were attempted.
    /// </summary>
    [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));
    }
    /// <summary>
    /// An unreadable ServiceLevel (Bad status, server without the node) is unknown, never
    /// Degraded: no probe, no switch. Positive control: the reads happened.
    /// </summary>
    [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));
    }
    /// <summary>
    /// 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.
    /// </summary>
    [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));
    }
    /// <summary>
    /// 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.
    /// </summary>
    [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");
    }
    /// <summary>
    /// <c>ServiceLevelFailover.Enabled=False</c> on the primary: the ServiceLevel is never even
    /// read, whatever it is.
    /// </summary>
    [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));
    }
    /// <summary>Only the PRIMARY's copy of the setting counts: disabling it on the backup changes nothing.</summary>
    [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));
    }
    /// <summary>A connection without a backup has nowhere to go: the ServiceLevel is never read.</summary>
    [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));
    }
    /// <summary>
    /// Symmetry: sitting on the backup (after a connectivity failover) when the backup becomes the
    /// Warm standby moves the connection back to the primary.
    /// </summary>
    [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));
    }
    /// <summary>
    /// An adapter without <see cref="IServiceLevelReadable"/> (MxGateway, or any plain adapter) is
    /// never watched and never probed, even with a backup and the feature enabled.
    /// </summary>
    [Fact]
    public void AdapterWithoutTheCapability_IsNeverWatched()
    {
        var plain = Substitute.For<IDataConnection>();
        plain.ConnectAsync(Arg.Any<IDictionary<string, string>>(), Arg.Any<CancellationToken>()).Returns(Task.CompletedTask);
        plain.SubscribeAsync(Arg.Any<string>(), Arg.Any<SubscriptionCallback>(), Arg.Any<CancellationToken>())
            .Returns("sub-1");
        plain.ReadAsync(Arg.Any<string>(), Arg.Any<CancellationToken>())
            .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));
    }
    /// <summary>The policy parser: defaults when absent, and clamps nonsense to safe values.</summary>
    [Fact]
    public void Policy_DefaultsAndClamps()
    {
        var defaults = ServiceLevelFailoverPolicy.From(new Dictionary<string, string> { ["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);
    }
 }
@@ -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;
 /// <summary>
 /// 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.
 /// </summary>
 internal sealed class ServiceLevelRig
 {
    private readonly ConcurrentDictionary<string, Func<byte?>> _levels = new();
    private readonly ConcurrentDictionary<string, bool> _unreachable = new();
    /// <summary>Every endpoint a session was opened on, in order — live sessions and probes alike.</summary>
    public ConcurrentQueue<string> Connects { get; } = new();
    /// <summary>Every tag subscribed, as "endpoint|tag" — only a live session receives these.</summary>
    public ConcurrentQueue<string> Subscribes { get; } = new();
    /// <summary>Number of ServiceLevel reads, per endpoint.</summary>
    public ConcurrentDictionary<string, int> Reads { get; } = new();
    /// <summary>Sets a fixed ServiceLevel for <paramref name="endpoint"/>.</summary>
    public void SetLevel(string endpoint, byte? level) => _levels[endpoint] = () => level;
    /// <summary>Sets a ServiceLevel source evaluated on every read of <paramref name="endpoint"/>.</summary>
    public void SetLevel(string endpoint, Func<byte?> level) => _levels[endpoint] = level;
    /// <summary>Makes <paramref name="endpoint"/> refuse (or accept again) new sessions.</summary>
    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);
 }
 /// <summary>
 /// An <see cref="IDataConnection"/> with the <see cref="IServiceLevelReadable"/> capability, bound to
 /// one endpoint of a <see cref="ServiceLevelRig"/>. Subscriptions and reads always succeed.
 /// </summary>
 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<string, string> 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<byte?> ReadServiceLevelAsync(CancellationToken cancellationToken = default) =>
        Task.FromResult(_connected ? _rig.Read(Endpoint) : null);
    public Task<string> 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<ReadResult> ReadAsync(string tagPath, CancellationToken cancellationToken = default) =>
        Task.FromResult(new ReadResult(true, new TagValue(1, QualityCode.Good, DateTimeOffset.UtcNow), null));
    public Task<IReadOnlyDictionary<string, ReadResult>> ReadBatchAsync(IEnumerable<string> tagPaths, CancellationToken cancellationToken = default) =>
        Task.FromResult<IReadOnlyDictionary<string, ReadResult>>(tagPaths.ToDictionary(
            t => t, _ => new ReadResult(true, new TagValue(1, QualityCode.Good, DateTimeOffset.UtcNow), null)));
    public Task<WriteResult> WriteAsync(string tagPath, object? value, CancellationToken cancellationToken = default) =>
        Task.FromResult(new WriteResult(true, null));
    public Task<IReadOnlyDictionary<string, WriteResult>> WriteBatchAsync(IDictionary<string, object?> values, CancellationToken cancellationToken = default) =>
        Task.FromResult<IReadOnlyDictionary<string, WriteResult>>(values.ToDictionary(v => v.Key, _ => new WriteResult(true, null)));
    public void RaiseDisconnected()
    {
        _connected = false;
        Disconnected?.Invoke();
    }
    public ValueTask DisposeAsync()
    {
        _connected = false;
        return ValueTask.CompletedTask;
    }
 }
@@ -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;
 /// <summary>
 /// ScadaBridge#138 at the OPC UA adapter: <see cref="OpcUaDataConnection.ReadServiceLevelAsync"/>
 /// against a substitute client (status / failure mapping) and against real in-process OPC UA
 /// servers (the actual <c>i=2267</c> node), plus the whole failover through the real
 /// <see cref="DataConnectionFactory"/> against a real server pair.
 /// </summary>
 public sealed class OpcUaServiceLevelTests : TestKit
 {
    private readonly string _clientPki =
        Path.Combine(Path.GetTempPath(), "sb138-client", Guid.NewGuid().ToString("N"));
    /// <summary>Initializes the test kit.</summary>
    public OpcUaServiceLevelTests() : base(@"akka.loglevel = DEBUG") { }
    /// <inheritdoc />
    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<string, string> 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<IOpcUaClient>();
        client.IsConnected.Returns(true);
        var factory = Substitute.For<IOpcUaClientFactory>();
        factory.Create().Returns(client);
        return (new OpcUaDataConnection(factory, NullLogger<OpcUaDataConnection>.Instance), client);
    }
    // ── Status / failure mapping (substitute client) ──
    /// <summary>The value comes from <c>i=2267</c>; a Good read of a byte is that byte.</summary>
    [Fact]
    public async Task Read_GoodByte_IsTheLevel_FromTheServiceLevelNode()
    {
        var (adapter, client) = WithSubstituteClient();
        client.ReadValueAsync("i=2267", Arg.Any<CancellationToken>())
            .Returns(((object?)(byte)90, DateTime.UtcNow, 0u));
        await adapter.ConnectAsync(new Dictionary<string, string>());
        Assert.Equal((byte)90, await adapter.ReadServiceLevelAsync());
    }
    /// <summary>A Bad status, or a value that is not a byte-ranged integer, is unknown (null).</summary>
    [Fact]
    public async Task Read_BadStatusOrNonByteValue_IsUnknown()
    {
        var (adapter, client) = WithSubstituteClient();
        await adapter.ConnectAsync(new Dictionary<string, string>());
        client.ReadValueAsync("i=2267", Arg.Any<CancellationToken>())
            .Returns(((object?)(byte)250, DateTime.UtcNow, 0x80340000u)); // BadNodeIdUnknown
        Assert.Null(await adapter.ReadServiceLevelAsync());
        client.ReadValueAsync("i=2267", Arg.Any<CancellationToken>())
            .Returns(((object?)"250", DateTime.UtcNow, 0u));
        Assert.Null(await adapter.ReadServiceLevelAsync());
        client.ReadValueAsync("i=2267", Arg.Any<CancellationToken>())
            .Returns(((object?)300, DateTime.UtcNow, 0u));
        Assert.Null(await adapter.ReadServiceLevelAsync());
        client.ReadValueAsync("i=2267", Arg.Any<CancellationToken>())
            .Returns(((object?)240, DateTime.UtcNow, 0u)); // Int32 in range → accepted
        Assert.Equal((byte)240, await adapter.ReadServiceLevelAsync());
    }
    /// <summary>
    /// A failed read is unknown and — unlike <see cref="OpcUaDataConnection.ReadAsync"/> — never
    /// raises <c>Disconnected</c>: the poll must not drive the reconnect machine.
    /// </summary>
    [Fact]
    public async Task Read_Failure_IsUnknown_AndNeverRaisesDisconnected()
    {
        var (adapter, client) = WithSubstituteClient();
        client.ReadValueAsync("i=2267", Arg.Any<CancellationToken>())
            .ThrowsAsync(new InvalidOperationException("session dropped"));
        var disconnected = 0;
        adapter.Disconnected += () => disconnected++;
        await adapter.ConnectAsync(new Dictionary<string, string>());
        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<CancellationToken>())
            .ThrowsAsync(new InvalidOperationException("session dropped"));
        await Assert.ThrowsAsync<InvalidOperationException>(() => adapter.ReadAsync("ns=2;s=x"));
        Assert.Equal(1, disconnected);
    }
    /// <summary>Before connect (no client) the level is unknown, and no read is attempted.</summary>
    [Fact]
    public async Task Read_NotConnected_IsUnknown()
    {
        var (adapter, client) = WithSubstituteClient();
        Assert.Null(await adapter.ReadServiceLevelAsync());
        await adapter.ConnectAsync(new Dictionary<string, string>());
        client.IsConnected.Returns(false);
        Assert.Null(await adapter.ReadServiceLevelAsync());
        await client.DidNotReceive().ReadValueAsync(Arg.Any<string>(), Arg.Any<CancellationToken>());
    }
    // ── Real OPC UA servers ──
    /// <summary>The real adapter reads the real server's <c>Server_ServiceLevel</c> as it changes.</summary>
    [Fact]
    public async Task RealServer_ReadServiceLevel_TracksTheServersValue()
    {
        await using var server = await SecureOpcUaTestServer.StartAsync();
        await using var adapter = new OpcUaDataConnection(
            new RealOpcUaClientFactory(ClientOptions()), NullLogger<OpcUaDataConnection>.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());
        }
    }
    /// <summary>
    /// 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.
    /// </summary>
    [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<string> Endpoint(IActorRef actor) =>
            (await actor.Ask<DataConnectionHealthReport>(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<ISiteHealthCollector>();
        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"]);
    }
 }
@@ -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;
 /// <summary>
 /// 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.
 /// </summary>
 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)!;
    /// <summary>What <c>CreateAlarmSubscriptionAsync</c> sets up for a live handle: its per-condition state map.</summary>
    private static void SeedSubscribedHandle(RealOpcUaClient client)
    {
        var field = typeof(RealOpcUaClient).GetField("_alarmLastState", BindingFlags.Instance | BindingFlags.NonPublic)!;
        var map = (ConcurrentDictionary<string, Dictionary<string, (bool Active, bool Acked)>>)field.GetValue(client)!;
        map[Handle] = new Dictionary<string, (bool Active, bool Acked)>(StringComparer.Ordinal);
    }
    private static void Boundary(RealOpcUaClient client, NodeId type, List<NativeAlarmTransition> 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<NativeAlarmTransition>)sink.Add]);
    private static void Condition(RealOpcUaClient client, string condition, bool active, bool acked, List<NativeAlarmTransition> 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<NativeAlarmTransition>)sink.Add]);
    }
    /// <summary>
    /// 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.
    /// </summary>
    [Fact]
    public void UnsolicitedPromotionRefresh_IsFramedAsItsOwnSnapshot_AndLiveDerivationResumes()
    {
        var client = new RealOpcUaClient();
        SeedSubscribedHandle(client);
        var sink = new List<NativeAlarmTransition>();
        // 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));
    }
 }
@@ -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);
    }
    /// <summary>
    /// Sets the server's <c>Server_ServiceLevel</c> (<c>i=2267</c>) — what a redundant server
    /// publishes for its role (ScadaBridge#138: 250/240 Active, 90/50 Warm standby, 1 booting).
    /// </summary>
    /// <param name="level">The ServiceLevel to publish.</param>
    public void SetServiceLevel(byte level) =>
        _server.CurrentInstance.ServerObject.ServiceLevel!.Value = level;
    /// <inheritdoc />
    public async ValueTask DisposeAsync()
    {