fix(opcuaclient): upstream session loss goes Bad and recovers (#670)

A bad keep-alive now degrades health (LastSuccessfulRead preserved) and,
once per outage, publishes BadNoCommunication (value retained) for every
subscribed ref plus a quality-only annotation on subscribed alarm
conditions. Good keep-alives and data/event notifications advance
LastSuccessfulRead, so an idle-but-alive subscription never looks dead
to the #527 watchdog. A keep-alive that recovers on the same session
re-reads the Bad refs and republishes them Good; a session that does not
come back is recovered by the watchdog's ShutdownAsync + InitializeAsync
+ re-subscribe.

Deletes the SessionReconnectHandler wiring (it retried forever in SDK
1.5.378, cloned subscriptions the driver never re-tracked and kept health
Healthy), OnReconnectComplete, TryRearmReconnect and the reconnect_*
counters (new keepalive_recovered). Teardown of a lost session skips
wire deletes and bounds CloseSession. SDK warnings reach the driver log.
Wire test reworked: restart => Bad, then watchdog-style reinit => Good.
This commit is contained in:
Joseph Doherty
2026-10-11 08:27:04 -04:00
parent db6a53036b
commit 2a8f9af5e4
13 changed files with 751 additions and 283 deletions
+3 -1
View File
@@ -302,7 +302,9 @@ non-`Healthy` state **and** its `LastSuccessfulRead` has been frozen for 90 s. T
success. Each conjunct has a test that goes red when the *other* is deleted. success. Each conjunct has a test that goes red when the *other* is deleted.
- **The mechanism rests on every driver reaching a non-Healthy state on sustained failure.** Surveyed across - **The mechanism rests on every driver reaching a non-Healthy state on sustained failure.** Surveyed across
all 12: nine poll drivers via `Degraded`, MQTT via live connection state, Galaxy via its reconnect all 12: nine poll drivers via `Degraded`, MQTT via live connection state, Galaxy via its reconnect
supervisor. `CalculationDriver` reports `Healthy` while running and so can never trip — correct, as it has supervisor, OpcUaClient via a bad session keep-alive (#670 — it used to stay `Healthy` forever; its SDK
`SessionReconnectHandler` is deleted, a lost session publishes `BadNoCommunication` once per outage and the
watchdog's re-init is the recovery; good keep-alives + notifications advance `LastSuccessfulRead`). `CalculationDriver` reports `Healthy` while running and so can never trip — correct, as it has
no transport. Re-check this if a new driver lands. no transport. Re-check this if a new driver lands.
- **The reconnect retry MUST tear down before re-initialising — FIXED in both drivers, keep it that way.** - **The reconnect retry MUST tear down before re-initialising — FIXED in both drivers, keep it that way.**
`ModbusDriver` (`_transport`) and `MqttDriver` (`_connection`) used to overwrite an `IAsyncDisposable` `ModbusDriver` (`_transport`) and `MqttDriver` (`_connection`) used to overwrite an `IAsyncDisposable`
+4 -1
View File
@@ -422,7 +422,10 @@ Two consequences worth knowing:
- **`CalculationDriver` can never trip it, correctly.** It reports `Healthy` whenever it is running and has - **`CalculationDriver` can never trip it, correctly.** It reports `Healthy` whenever it is running and has
no transport to be degraded about. Every other driver reaches a non-Healthy state on sustained failure: no transport to be degraded about. Every other driver reaches a non-Healthy state on sustained failure:
the nine poll drivers via `Degraded`, MQTT via its live connection state, Galaxy via its reconnect the nine poll drivers via `Degraded`, MQTT via its live connection state, Galaxy via its reconnect
supervisor. This was surveyed across all 12 before relying on it — it is the assumption the whole supervisor, OpcUaClient via a bad session keep-alive (`Degraded`, since Gitea #670 — before that its health
only moved on `ReadAsync`, which a subscription-only driver never calls, so it stayed `Healthy` through any
outage and this survey was wrong about it; good keep-alives and notifications now advance its read stamp, so
an idle-but-alive subscription does not look dead either). This was surveyed across all 12 before relying on it — it is the assumption the whole
mechanism rests on. mechanism rests on.
- **A device that accepts a connection but serves no data will cycle**, reconnecting once per threshold. - **A device that accepts a connection but serves no data will cycle**, reconnecting once per threshold.
That is the correct reading of a broken device, it is rate-limited, and it is visible in the health That is the correct reading of a broken device, it is rate-limited, and it is visible in the health
+36 -11
View File
@@ -45,22 +45,46 @@ out-of-process isolation.
``` ```
`EndpointUrls` (a list) takes precedence over the single-URL `EndpointUrl` and `EndpointUrls` (a list) takes precedence over the single-URL `EndpointUrl` and
provides ordered **failover** — the driver tries each candidate in turn at init provides ordered **failover** — the driver tries each candidate in turn at every
and on session drop, and the first to connect wins (e.g. a hot-standby pair on `InitializeAsync` (including the comms-loss watchdog's re-initialise), and the first to connect wins (e.g. a hot-standby pair on
4840 / 4841). See 4840 / 4841). See
`src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Contracts/OpcUaClientDriverOptions.cs` `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Contracts/OpcUaClientDriverOptions.cs`
for every field (security policy/mode, auth type, session timeout, keep-alive, for every field (security policy/mode, auth type, session timeout, keep-alive,
reconnect period, browse root, node/depth caps). browse root, node/depth caps). `ReconnectPeriod` is still parsed but no longer used (#670).
### Session lifetime ### Session lifetime
A single `Session` per driver instance; subscriptions multiplex onto it. The A single `Session` per driver instance; subscriptions multiplex onto it. Stored NodeIds embed the server-stable namespace **URI** (not the
SDK reconnect handler takes the session down and brings it back on remote-server
restart, re-sending subscriptions on reconnect so monitored-item handles don't
dangle. Stored NodeIds embed the server-stable namespace **URI** (not the
session-relative `ns=N` index) so a remote namespace-table reorder across a session-relative `ns=N` index) so a remote namespace-table reorder across a
restart doesn't silently re-point references at the wrong namespace. restart doesn't silently re-point references at the wrong namespace.
### Upstream outage and recovery (Gitea #670)
The session keep-alive (`KeepAliveInterval`, default 5 s) is the liveness signal:
- **Good keep-alive / any data or event notification** advances `DriverHealth.LastSuccessfulRead` — a
subscription-only driver never calls `ReadAsync`, so without this its read stamp froze at connect time.
- **Bad keep-alive** ⇒ health `Degraded` (`LastError = "keep-alive: <status>"`, read stamp preserved), host status
`Stopped`, and — **once per outage**, not per keep-alive — every subscribed ref is published `BadNoCommunication`
with its last value retained, plus a quality-only `BadNoCommunication` annotation (#626 pattern, no alarm-state
change, no `/alerts` row) on every subscribed alarm condition.
- **The session survives** (a network blip; the keep-alive turns Good again) ⇒ recovered in place: health `Healthy`,
the refs still Bad are re-read once and published Good (a server whose subscription survived does not resend an
unchanged value, and a `(value, status)` dedup downstream lets the identical value through), alarm quality goes
back Good and a `ConditionRefresh` re-reports anything missed. A notification that lands first wins over the read.
- **The session does not come back** (an upstream restart: the server forgot the session) ⇒ the driver stays
`Degraded` with its read stamp frozen, which is exactly the #527 comms-loss watchdog's trip condition. After 90 s
`DriverInstanceActor` runs `ShutdownAsync` + `InitializeAsync` (retried every 10 s while the server is down) and
re-subscribes data + alarms; the new subscription's initial values come back Good. Alarm last-told state survives
the re-initialise (#619), so the post-subscribe `ConditionRefresh` raises no un-ack flicker; the host's reconnect
annotation restores condition quality.
There is **no in-driver reconnect** any more. The SDK's `SessionReconnectHandler` (removed) retried forever in SDK
1.5.378 (its "exhausted" path was unreachable), recreated sessions by cloning subscriptions the driver never
re-tracked, and kept health `Healthy` throughout — so on docker-dev a stopped opc-plc left nodes Good on their last
value indefinitely and they never recovered. A teardown against a lost session skips the wire deletes and bounds
`CloseSession` to 5 s. SDK warnings (keep-alive / publish / channel errors) now reach the driver's own logger.
### Namespace assignment (retired, Gitea #648) ### Namespace assignment (retired, Gitea #648)
Earlier versions required a per-instance `TargetNamespaceKind` (`Equipment` + Earlier versions required a per-instance `TargetNamespaceKind` (`Equipment` +
@@ -107,11 +131,11 @@ initialise (the AdminUI form drops them on the next save). Pinned by
> from whatever the host retained: an inactive+acked one goes out as `Clear` + `Acknowledge` (#657; it used to be an > from whatever the host retained: an inactive+acked one goes out as `Clear` + `Acknowledge` (#657; it used to be an
> `Unspecified` the host wrote as "Activated", and a retained-Active host condition stayed Active). So the host writes no > `Unspecified` the host wrote as "Activated", and a retained-Active host condition stayed Active). So the host writes no
> duplicate row when its condition already shows that state (see `docs/AlarmTracking.md`). Condition **branch** events (non-null `BranchId`) are ignored. > duplicate row when its condition already shows that state (see `docs/AlarmTracking.md`). Condition **branch** events (non-null `BranchId`) are ignored.
> **`ConditionRefresh`** is called after the event item is created, after every in-place reconnect, and on a > **`ConditionRefresh`** is called after the event item is created, after a keep-alive outage the session survived, and on a
> `RefreshRequired` event, so a condition already active upstream is reported at subscribe; `RefreshStart`/`RefreshEnd` > `RefreshRequired` event, so a condition already active upstream is reported at subscribe; `RefreshStart`/`RefreshEnd`
> markers are dropped. A server without it (`BadNotSupported`/`BadMethodInvalid`/…) is logged once and tolerated. > markers are dropped. A server without it (`BadNotSupported`/`BadMethodInvalid`/…) is logged once and tolerated.
> Last-told state survives a re-initialise (the host re-subscribes the same set over nodes it did not rebuild) and is > Last-told state survives a re-initialise (the host re-subscribes the same set over nodes it did not rebuild) and is
> forgotten when the subscribed set changes (the host rebuilt, recreating the nodes inactive). Index keys are server-stable (`nsu=`), so a namespace reorder across a reconnect keeps > forgotten when the subscribed set changes (the host rebuilt, recreating the nodes inactive). Index keys are server-stable (`nsu=`), so a namespace reorder across a re-initialise keeps
> matching. docker-dev seeds an opc-plc driver (`MAIN-opcua`); the alarm tag is > matching. docker-dev seeds an opc-plc driver (`MAIN-opcua`); the alarm tag is
> created only when `@OpcPlcAlarmSourceNodeId` is set in `docker-dev/seed/seed-clusters.sql`. > created only when `@OpcPlcAlarmSourceNodeId` is set in `docker-dev/seed/seed-clusters.sql`.
@@ -192,8 +216,9 @@ every `InitializeAsync`, including comms-loss reconnects). The discovery channel
Session/publish events are published on the shared OTel meter as the counter Session/publish events are published on the shared OTel meter as the counter
`otopcua.driver.opcuaclient.events`, tagged `driver_instance` (the driver instance id) and `event` = `otopcua.driver.opcuaclient.events`, tagged `driver_instance` (the driver instance id) and `event` =
`publish_response`, `notification` (items), `data_change`, `bad_status_value`, `publish_error`, `publish_response`, `notification` (items), `data_change`, `bad_status_value`, `publish_error`,
`keepalive_miss`, `reconnect_attempt`, `reconnect_success`, `reconnect_exhausted`. A climbing `keepalive_miss` / `keepalive_miss`, `keepalive_recovered` (a bad keep-alive followed by a good one on the same session). A climbing
`reconnect_attempt` rate is the signature of a flapping upstream. They are not part of `DriverHealth` or the `keepalive_miss` / `keepalive_recovered` rate is the signature of a flapping upstream (`reconnect_*` were removed with
the reconnect handler, #670). They are not part of `DriverHealth` or the
`/hosts` panel. `/hosts` panel.
## Testing ## Testing
@@ -80,11 +80,11 @@ public static class OtOpcUaTelemetry
/// Session/publish-level events of an OPC UA Client driver instance (#276): the operator's view of a /// Session/publish-level events of an OPC UA Client driver instance (#276): the operator's view of a
/// flapping upstream. Tagged <see cref="PollOverrunDriverInstanceTag"/> (<c>driver_instance</c>) + /// flapping upstream. Tagged <see cref="PollOverrunDriverInstanceTag"/> (<c>driver_instance</c>) +
/// <c>event</c> = <c>publish_response|notification|data_change|bad_status_value|publish_error| /// <c>event</c> = <c>publish_response|notification|data_change|bad_status_value|publish_error|
/// keepalive_miss|reconnect_attempt|reconnect_success|reconnect_exhausted</c>. /// keepalive_miss|keepalive_recovered</c>.
/// </summary> /// </summary>
public static readonly Counter<long> OpcUaClientEvents = public static readonly Counter<long> OpcUaClientEvents =
Meter.CreateCounter<long>("otopcua.driver.opcuaclient.events", unit: "{event}", Meter.CreateCounter<long>("otopcua.driver.opcuaclient.events", unit: "{event}",
description: "OPC UA Client driver session/publish events (event=publish_response|notification|data_change|bad_status_value|publish_error|keepalive_miss|reconnect_attempt|reconnect_success|reconnect_exhausted)."); description: "OPC UA Client driver session/publish events (event=publish_response|notification|data_change|bad_status_value|publish_error|keepalive_miss|keepalive_recovered).");
// ---------------- VirtualTag / ScriptedAlarm engines ---------------- // ---------------- VirtualTag / ScriptedAlarm engines ----------------
@@ -101,7 +101,10 @@ public sealed record OpcUaClientDriverOptions
/// <summary>Client-side keep-alive interval.</summary> /// <summary>Client-side keep-alive interval.</summary>
public TimeSpan KeepAliveInterval { get; init; } = TimeSpan.FromSeconds(5); public TimeSpan KeepAliveInterval { get; init; } = TimeSpan.FromSeconds(5);
/// <summary>Initial reconnect delay after a session drop.</summary> /// <summary>
/// Retained for config compatibility only — no longer used (Gitea #670). The driver has no in-driver reconnect
/// handler; a lost session is recovered by the host's comms-loss watchdog re-initialising the driver.
/// </summary>
public TimeSpan ReconnectPeriod { get; init; } = TimeSpan.FromSeconds(5); public TimeSpan ReconnectPeriod { get; init; } = TimeSpan.FromSeconds(5);
/// <summary> /// <summary>
@@ -22,12 +22,8 @@ internal sealed class OpcUaClientDiagnostics
public const string PublishError = "publish_error"; public const string PublishError = "publish_error";
/// <summary>A keep-alive came back Bad.</summary> /// <summary>A keep-alive came back Bad.</summary>
public const string KeepAliveMiss = "keepalive_miss"; public const string KeepAliveMiss = "keepalive_miss";
/// <summary>The reconnect handler was armed.</summary> /// <summary>A keep-alive came back Good after a bad one on the same session (an in-place recovery, #670).</summary>
public const string ReconnectAttempt = "reconnect_attempt"; public const string KeepAliveRecovered = "keepalive_recovered";
/// <summary>A reconnect landed a new session.</summary>
public const string ReconnectSuccess = "reconnect_success";
/// <summary>A reconnect handler exhausted its retry window.</summary>
public const string ReconnectExhausted = "reconnect_exhausted";
private readonly string _driverInstanceId; private readonly string _driverInstanceId;
@@ -23,10 +23,10 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient;
/// </para> /// </para>
/// <para> /// <para>
/// <b>Session lifetime</b>: a single <see cref="Session"/> per driver instance. /// <b>Session lifetime</b>: a single <see cref="Session"/> per driver instance.
/// Subscriptions multiplex onto that session; SDK reconnect handler takes the session /// Subscriptions multiplex onto that session. A lost session is reported (health Degraded,
/// down and brings it back up on remote-server restart — the driver must re-send /// subscribed values BadNoCommunication — Gitea #670) and recovered by the host's comms-loss
/// subscriptions + TransferSubscriptions on reconnect to avoid dangling /// watchdog re-initialising the driver and re-subscribing; a session whose keep-alive comes
/// monitored-item handles. That mechanic lands in PR 69. /// back on its own recovers in place.
/// </para> /// </para>
/// </remarks> /// </remarks>
public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWritable, ISubscribable, IHostConnectivityProbe, IAlarmSource, ISkippedTagReporter, IDisposable, IAsyncDisposable public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWritable, ISubscribable, IHostConnectivityProbe, IAlarmSource, ISkippedTagReporter, IDisposable, IAsyncDisposable
@@ -82,6 +82,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
_diag = new OpcUaClientDiagnostics(driverInstanceId); _diag = new OpcUaClientDiagnostics(driverInstanceId);
_logger = logger ?? NullLogger<OpcUaClientDriver>.Instance; _logger = logger ?? NullLogger<OpcUaClientDriver>.Instance;
_secretResolver = secretResolver ?? NullSecretResolver.Instance; _secretResolver = secretResolver ?? NullSecretResolver.Instance;
_sdkTelemetry = logger is null ? SdkTelemetry.Context : SdkTelemetry.ForLogger(logger);
} }
/// <summary>Mutable so <see cref="InitializeAsync"/> can adopt a re-parsed config on reinitialize /// <summary>Mutable so <see cref="InitializeAsync"/> can adopt a re-parsed config on reinitialize
@@ -140,6 +141,8 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
private const uint StatusBadNodeIdInvalid = 0x80330000u; private const uint StatusBadNodeIdInvalid = 0x80330000u;
private const uint StatusBadInternalError = 0x80020000u; private const uint StatusBadInternalError = 0x80020000u;
private const uint StatusBadCommunicationError = 0x80050000u; private const uint StatusBadCommunicationError = 0x80050000u;
private const uint StatusBadNoCommunication = 0x80310000u;
private const uint StatusGood = 0x00000000u;
private readonly SemaphoreSlim _gate = new(1, 1); private readonly SemaphoreSlim _gate = new(1, 1);
@@ -150,20 +153,73 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
internal SemaphoreSlim Gate => _gate; internal SemaphoreSlim Gate => _gate;
/// <summary> /// <summary>
/// Test-only seam to swap the active <see cref="Session"/> the way /// Test-only seam to swap the active <see cref="Session"/> the way a re-initialise does at
/// <see cref="OnReconnectComplete"/> does at runtime, so the stale-session race the /// runtime, so the stale-session race the read/history paths guard against (re-read Session
/// read/history paths guard against (re-read Session inside <c>_gate</c>, not before) /// inside <c>_gate</c>, not before) can be exercised deterministically without a live connect.
/// can be exercised deterministically without standing up a live reconnect handler. /// Production code never calls this — <see cref="InitializeAsync"/> owns the real swap.
/// Production code never calls this — the SDK reconnect handler owns the real swap.
/// </summary> /// </summary>
/// <param name="session">The session instance to install, or <c>null</c> to clear it.</param> /// <param name="session">The session instance to install, or <c>null</c> to clear it.</param>
internal void SetSessionForTest(ISession? session) => Session = session; internal void SetSessionForTest(ISession? session) => Session = session;
/// <summary>Test-only seam (#670): install a health snapshot as a completed connect would.</summary>
/// <param name="health">The health to install.</param>
internal void SetHealthForTest(DriverHealth health) => _health = health;
/// <summary>Test-only seam (#670): raise the session keep-alive callback as the SDK timer does.</summary>
/// <param name="status">The keep-alive status.</param>
/// <param name="sender">The raising session; defaults to the current one.</param>
internal void RaiseKeepAliveForTest(StatusCode status, ISession? sender = null) =>
OnKeepAlive(sender ?? Session!, new KeepAliveEventArgs(new ServiceResult(status), ServerState.Running, DateTime.UtcNow));
/// <summary> /// <summary>
/// Test-only seam to install the <see cref="NamespaceMap"/> that connect/reconnect builds at /// Test-only seam (#670): register a data subscription over <paramref name="refs"/> as
/// <see cref="SubscribeAsync"/> would, without the wire create.
/// </summary>
/// <param name="refs">The RawPaths subscribed.</param>
/// <returns>The subscription handle.</returns>
internal ISubscriptionHandle AddSubscriptionForTest(IReadOnlyList<string> refs)
{
var id = Interlocked.Increment(ref _nextSubscriptionId);
var handle = new OpcUaSubscriptionHandle(id);
var subscription = new Subscription(telemetry: null!, new SubscriptionOptions { DisplayName = $"opcua-sub-{id}" });
var items = new List<MonitoredItemNotificationHandle>();
foreach (var r in refs)
{
var item = new MonitoredItem(telemetry: null!, new MonitoredItemOptions { DisplayName = r }) { Handle = r };
MonitoredItemNotificationEventHandler h = (mi, args) => OnMonitoredItemNotification(handle, mi, args);
item.Notification += h;
items.Add(new MonitoredItemNotificationHandle(item, h));
}
_subscriptions[id] = new RemoteSubscription(subscription, handle, items);
return handle;
}
/// <summary>Test-only seam (#670): deliver one data notification as the SDK publish thread would.</summary>
/// <param name="handle">The subscription handle.</param>
/// <param name="fullRef">The RawPath.</param>
/// <param name="value">The delivered value.</param>
internal void DeliverNotificationForTest(ISubscriptionHandle handle, string fullRef, DataValue value) =>
PublishNotification((OpcUaSubscriptionHandle)handle, fullRef, value);
/// <summary>Test-only seam (#670): register an alarm subscription over <paramref name="refs"/> as
/// <see cref="SubscribeAlarmsAsync"/> would, without the wire create.</summary>
/// <param name="refs">The alarm RawPaths subscribed.</param>
internal void AddAlarmSubscriptionForTest(IReadOnlyList<string> refs)
{
var id = Interlocked.Increment(ref _nextAlarmSubscriptionId);
var handle = new OpcUaAlarmSubscriptionHandle(id);
BeginAlarmSubscription(refs);
var item = new MonitoredItem(telemetry: null!, new MonitoredItemOptions { DisplayName = "Server/Events" });
_alarmSubscriptions[id] = new RemoteAlarmSubscription(
new Subscription(telemetry: null!, new SubscriptionOptions { DisplayName = $"{AlarmSubscriptionNamePrefix}{id}" }),
handle, item, (_, _) => { });
}
/// <summary>
/// Test-only seam to install the <see cref="NamespaceMap"/> that connect builds at
/// runtime, so the v3 RawPath → upstream node id resolution can be exercised without a live /// runtime, so the v3 RawPath → upstream node id resolution can be exercised without a live
/// connect (which is what populates <see cref="_namespaceMap"/>). Production code never calls /// connect (which is what populates <see cref="_namespaceMap"/>). Production code never calls
/// this — <see cref="InitializeAsync"/> / <see cref="OnReconnectComplete"/> own the real build. /// this — <see cref="InitializeAsync"/> owns the real build.
/// </summary> /// </summary>
/// <param name="map">The namespace map to install, or <c>null</c> to clear it.</param> /// <param name="map">The namespace map to install, or <c>null</c> to clear it.</param>
internal void SetNamespaceMapForTest(NamespaceMap? map) => _namespaceMap = map; internal void SetNamespaceMapForTest(NamespaceMap? map) => _namespaceMap = map;
@@ -191,13 +247,38 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
/// <summary>The <see cref="CertificateValidator"/> that owns <see cref="_certValidationHandler"/>.</summary> /// <summary>The <see cref="CertificateValidator"/> that owns <see cref="_certValidationHandler"/>.</summary>
private CertificateValidator? _certValidatorRef; private CertificateValidator? _certValidatorRef;
/// <summary> /// <summary>
/// SDK-provided reconnect handler that owns the retry loop + session-transfer machinery /// Gitea #670 — 1 while the CURRENT session's keep-alive is bad (an outage is in progress), else 0. The
/// when the session's keep-alive channel reports a bad status. Null outside the /// 0→1 edge publishes the outage once (Bad per subscribed ref + a quality-only alarm annotation); the 1→0
/// reconnecting window; constructed lazily inside the keep-alive handler. Guarded by /// edge on a good keep-alive recovers in place. Reset whenever the session is released, so a fresh session
/// <see cref="_probeLock"/> — keep-alive callbacks fire from the SDK timer thread and /// reports its own loss. There is deliberately no in-driver reconnect any more: a session that never comes
/// can race a check-then-set if left unsynchronized. /// back keeps health Degraded with <see cref="DriverHealth.LastSuccessfulRead"/> frozen, which is exactly what
/// the host's #527 comms-loss watchdog trips on (→ ShutdownAsync + InitializeAsync + re-subscribe).
/// </summary> /// </summary>
private SessionReconnectHandler? _reconnectHandler; private int _sessionLost;
/// <summary>#670 — the last value delivered per subscribed RawPath, so an outage can publish
/// <c>BadNoCommunication</c> with the value retained (the same shape the host's own comms-loss fan uses).</summary>
private readonly System.Collections.Concurrent.ConcurrentDictionary<string, DataValueSnapshot> _lastValues =
new(StringComparer.Ordinal);
/// <summary>#670 — refs published Bad by the current outage and not yet re-published Good. A notification
/// clears its ref (fresh data beats the recovery read); the recovery read publishes only what is left.</summary>
private readonly System.Collections.Concurrent.ConcurrentDictionary<string, byte> _awaitingRecovery =
new(StringComparer.Ordinal);
/// <summary>Ticks of the last <see cref="DriverHealth.LastSuccessfulRead"/> advance from a notification —
/// throttles the per-notification health write to at most one per second.</summary>
private long _lastEvidenceStampTicks;
/// <summary>Clock for liveness stamps. Test seam (#670); production uses <see cref="TimeProvider.System"/>.</summary>
internal TimeProvider Clock { get; set; } = TimeProvider.System;
/// <summary>
/// SDK telemetry for this instance's sessions: SDK warnings (keep-alive / publish / channel errors) go to the
/// driver's own logger (#670 — the shared <see cref="SdkTelemetry.Context"/> has no provider, which made every
/// SDK-side session error invisible).
/// </summary>
private readonly ITelemetryContext _sdkTelemetry;
/// <summary> /// <summary>
/// Bidirectional namespace map built at connect time from <c>session.NamespaceUris</c>. /// Bidirectional namespace map built at connect time from <c>session.NamespaceUris</c>.
@@ -314,12 +395,11 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
"Tried:\n " + string.Join("\n ", attemptErrors), "Tried:\n " + string.Join("\n ", attemptErrors),
attemptErrors.Select(e => new InvalidOperationException(e))); attemptErrors.Select(e => new InvalidOperationException(e)));
// Wire the session's keep-alive channel into HostState + the reconnect trigger. // Wire the session's keep-alive channel into HostState + health (#670). OPC UA keep-alives
// OPC UA keep-alives are authoritative for session liveness: the SDK pings on // are authoritative for session liveness: the SDK pings on KeepAliveInterval. A good one is
// KeepAliveInterval and sets KeepAliveStopped when N intervals elapse without a // liveness evidence (advances LastSuccessfulRead); a bad one degrades health and publishes the
// response. On a bad keep-alive the driver spins up a SessionReconnectHandler // outage. Recovery from a session that does not come back is the host's comms-loss watchdog
// which transparently retries + swaps the underlying session. Subscriptions move // (ShutdownAsync + InitializeAsync + re-subscribe) — there is no in-driver reconnect handler.
// via TransferSubscriptions so local MonitoredItem handles stay valid.
_keepAliveHandler = OnKeepAlive; _keepAliveHandler = OnKeepAlive;
session.KeepAlive += _keepAliveHandler; session.KeepAlive += _keepAliveHandler;
AttachDiagnostics(session); AttachDiagnostics(session);
@@ -355,9 +435,9 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
/// </summary> /// </summary>
private async Task<ApplicationConfiguration> BuildApplicationConfigurationAsync(CancellationToken ct) private async Task<ApplicationConfiguration> BuildApplicationConfigurationAsync(CancellationToken ct)
{ {
// The ITelemetryContext overload is the non-obsolete ctor; an empty builder keeps SDK // The ITelemetryContext overload is the non-obsolete ctor; SDK warnings reach the driver's
// internals off any logging provider (mirrors Client.Shared's ClientTelemetry). // logger (#670) when one was supplied.
var app = new ApplicationInstance(SdkTelemetry.Context) var app = new ApplicationInstance(_sdkTelemetry)
{ {
ApplicationName = _options.SessionName, ApplicationName = _options.SessionName,
ApplicationType = ApplicationType.Client, ApplicationType = ApplicationType.Client,
@@ -681,7 +761,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
endpointConfig.OperationTimeout = (int)_options.Timeout.TotalMilliseconds; endpointConfig.OperationTimeout = (int)_options.Timeout.TotalMilliseconds;
var endpoint = new ConfiguredEndpoint(null, selected, endpointConfig); var endpoint = new ConfiguredEndpoint(null, selected, endpointConfig);
var session = await new DefaultSessionFactory(SdkTelemetry.Context).CreateAsync( var session = await new DefaultSessionFactory(_sdkTelemetry).CreateAsync(
appConfig, appConfig,
endpoint, endpoint,
false, // updateBeforeConnect false, // updateBeforeConnect
@@ -795,30 +875,15 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
} }
/// <summary> /// <summary>
/// Cancels any in-flight reconnect, detaches the keep-alive handler and closes + disposes the current /// Detaches the keep-alive handler and closes + disposes the current session. The session part of
/// session. The session part of <see cref="ShutdownAsync"/>; also run at the top of /// <see cref="ShutdownAsync"/>; also run at the top of <see cref="InitializeAsync"/> so a re-initialise on a
/// <see cref="InitializeAsync"/> so a re-initialise on a live driver (#630) does not orphan the /// live driver (#630) does not orphan the previous session. Subscriptions are NOT touched here. Clears the
/// previous session and its reconnect handler. Subscriptions are NOT touched here. /// #670 outage latch, so the next session reports its own loss.
/// </summary> /// </summary>
private async Task ReleaseSessionAsync(CancellationToken cancellationToken) private async Task ReleaseSessionAsync(CancellationToken cancellationToken)
{ {
// Abort any in-flight reconnect attempts before touching the session — BeginReconnect's // Take the session reference under _probeLock and clear Session to null so any concurrent caller that
// retry loop holds a reference to the current session and would fight Session.CloseAsync // checks inside _gate sees null immediately after release begins.
// if left spinning. Take the handler under _probeLock so a keep-alive callback racing
// through OnKeepAlive can't arm a fresh handler after we've torn this one down.
SessionReconnectHandler? handlerToCancel;
lock (_probeLock)
{
handlerToCancel = _reconnectHandler;
_reconnectHandler = null;
}
try { handlerToCancel?.CancelReconnect(); } catch { }
handlerToCancel?.Dispose();
// Take the session reference under _probeLock before touching it, so we can't race
// an OnReconnectComplete that is simultaneously swapping to a new session. We clear
// Session to null here so any concurrent caller that checks inside _gate sees null
// immediately after shutdown begins.
ISession? sessionToClose; ISession? sessionToClose;
lock (_probeLock) lock (_probeLock)
{ {
@@ -832,11 +897,25 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
Session = null; Session = null;
} }
try { if (sessionToClose is Session s) await s.CloseAsync(cancellationToken).ConfigureAwait(false); } // #670: bound the close — against a dead upstream (the comms-loss watchdog's whole use case) an unbounded
// CloseSession would sit out the full operation timeout on every reconnect attempt.
try
{
if (sessionToClose is Session s)
{
using var closeCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
closeCts.CancelAfter(SessionCloseTimeout);
await s.CloseAsync(closeCts.Token).ConfigureAwait(false);
}
}
catch { /* best-effort */ } catch { /* best-effort */ }
try { sessionToClose?.Dispose(); } catch { } try { sessionToClose?.Dispose(); } catch { }
Interlocked.Exchange(ref _sessionLost, 0);
} }
/// <summary>Upper bound on the best-effort wire calls a teardown makes against a possibly-dead upstream.</summary>
private static readonly TimeSpan SessionCloseTimeout = TimeSpan.FromSeconds(5);
/// <inheritdoc /> /// <inheritdoc />
public async Task ShutdownAsync(CancellationToken cancellationToken) public async Task ShutdownAsync(CancellationToken cancellationToken)
{ {
@@ -846,11 +925,13 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
// after close anyway. Before deleting each subscription we detach the Notification // after close anyway. Before deleting each subscription we detach the Notification
// handlers we attached at subscribe time so the SDK's invocation list no longer // handlers we attached at subscribe time so the SDK's invocation list no longer
// holds the driver instance through the closure. // holds the driver instance through the closure.
// #670: against a session already known lost (bad keep-alive) the wire deletes cannot succeed and would each
// sit out the operation timeout — skip them; disposing the session releases the local objects.
var sessionAlive = Volatile.Read(ref _sessionLost) == 0;
foreach (var rs in _subscriptions.Values) foreach (var rs in _subscriptions.Values)
{ {
DetachNotificationHandlers(rs.ItemHandlers); DetachNotificationHandlers(rs.ItemHandlers);
try { await rs.Subscription.DeleteAsync(silent: true, cancellationToken).ConfigureAwait(false); } if (sessionAlive) await DeleteSubscriptionBestEffortAsync(rs.Subscription, cancellationToken).ConfigureAwait(false);
catch { /* best-effort */ }
} }
_subscriptions.Clear(); _subscriptions.Clear();
@@ -858,10 +939,13 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
{ {
try { ras.EventItem.Notification -= ras.Handler; } try { ras.EventItem.Notification -= ras.Handler; }
catch { /* best-effort */ } catch { /* best-effort */ }
try { await ras.Subscription.DeleteAsync(silent: true, cancellationToken).ConfigureAwait(false); } if (sessionAlive) await DeleteSubscriptionBestEffortAsync(ras.Subscription, cancellationToken).ConfigureAwait(false);
catch { /* best-effort */ }
} }
_alarmSubscriptions.Clear(); _alarmSubscriptions.Clear();
// The host re-subscribes after a reinit and the new subscriptions deliver initial values; nothing of the
// previous connection's value bookkeeping applies to them.
_lastValues.Clear();
_awaitingRecovery.Clear();
// #619 re-review: the last-told alarm states (and ack targets) deliberately SURVIVE a shutdown. A // #619 re-review: the last-told alarm states (and ack targets) deliberately SURVIVE a shutdown. A
// ReinitializeAsync (comms-loss watchdog / Force Reconnect) is followed by the host re-subscribing the // ReinitializeAsync (comms-loss watchdog / Force Reconnect) is followed by the host re-subscribing the
// same set over condition nodes it did NOT rebuild; forgetting here would make the post-subscribe // same set over condition nodes it did NOT rebuild; forgetting here would make the post-subscribe
@@ -885,6 +969,18 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
_health = new DriverHealth(DriverState.Unknown, _health.LastSuccessfulRead, null); _health = new DriverHealth(DriverState.Unknown, _health.LastSuccessfulRead, null);
} }
/// <summary>Best-effort, bounded delete of one subscription (shutdown path).</summary>
private static async Task DeleteSubscriptionBestEffortAsync(Subscription subscription, CancellationToken cancellationToken)
{
try
{
using var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
cts.CancelAfter(SessionCloseTimeout);
await subscription.DeleteAsync(silent: true, cts.Token).ConfigureAwait(false);
}
catch { /* best-effort */ }
}
/// <inheritdoc /> /// <inheritdoc />
public DriverHealth GetHealth() => _health; public DriverHealth GetHealth() => _health;
@@ -899,14 +995,25 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
// call to this reference — a reconnect can swap Session while we wait on _gate. The // call to this reference — a reconnect can swap Session while we wait on _gate. The
// session actually used is re-read inside the gate. // session actually used is re-read inside the gate.
_ = RequireSession(); _ = RequireSession();
var (results, _) = await ReadCoreAsync(fullReferences, cancellationToken).ConfigureAwait(false);
return results;
}
/// <summary>
/// The read itself, also used by the #670 in-place recovery. <c>TransportOk</c> is false when the wire read
/// did not happen or threw (every result is then a local <c>BadCommunicationError</c>, not upstream data).
/// </summary>
private async Task<(DataValueSnapshot[] Results, bool TransportOk)> ReadCoreAsync(
IReadOnlyList<string> fullReferences, CancellationToken cancellationToken)
{
var results = new DataValueSnapshot[fullReferences.Count]; var results = new DataValueSnapshot[fullReferences.Count];
var now = DateTime.UtcNow; var now = Clock.GetUtcNow().UtcDateTime;
await _gate.WaitAsync(cancellationToken).ConfigureAwait(false); await _gate.WaitAsync(cancellationToken).ConfigureAwait(false);
try try
{ {
// Re-read Session inside the critical section: if a reconnect completed while we // Re-read Session inside the critical section: if a reconnect completed while we
// were blocked on _gate, OnReconnectComplete has already swapped in the new // were blocked on _gate, a re-initialise has already swapped in the new
// session. NodeId parsing is namespace-relative, so it must also use the current // session. NodeId parsing is namespace-relative, so it must also use the current
// session's namespace table. // session's namespace table.
var session = Session; var session = Session;
@@ -914,7 +1021,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
{ {
for (var i = 0; i < fullReferences.Count; i++) for (var i = 0; i < fullReferences.Count; i++)
results[i] = new DataValueSnapshot(null, StatusBadCommunicationError, null, now); results[i] = new DataValueSnapshot(null, StatusBadCommunicationError, null, now);
return results; return (results, false);
} }
// Resolve each RawPath reference (v3) to a live NodeId against the current session. // Resolve each RawPath reference (v3) to a live NodeId against the current session.
@@ -934,7 +1041,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
indexMap.Add(i); indexMap.Add(i);
} }
if (toSend.Count == 0) return results; if (toSend.Count == 0) return (results, true);
try try
{ {
@@ -973,10 +1080,11 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
results[r] = new DataValueSnapshot(null, StatusBadCommunicationError, null, now); results[r] = new DataValueSnapshot(null, StatusBadCommunicationError, null, now);
} }
_health = new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, ex.Message); _health = new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, ex.Message);
return (results, false);
} }
} }
finally { _gate.Release(); } finally { _gate.Release(); }
return results; return (results, true);
} }
// ---- IWritable ---- // ---- IWritable ----
@@ -1473,7 +1581,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
try try
{ {
// Re-read Session inside the critical section: a reconnect completing while we were // Re-read Session inside the critical section: a reconnect completing while we were
// blocked on _gate has already swapped in the new session via OnReconnectComplete. // blocked on _gate has already swapped in the new session via a re-initialise.
var session = Session var session = Session
?? throw new InvalidOperationException( ?? throw new InvalidOperationException(
"OpcUaClient session was lost before the subscription could be created."); "OpcUaClient session was lost before the subscription could be created.");
@@ -1557,17 +1665,44 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
if (args.NotificationValue is not MonitoredItemNotification mn) return; if (args.NotificationValue is not MonitoredItemNotification mn) return;
var dv = mn.Value; var dv = mn.Value;
if (dv is null) return; if (dv is null) return;
var fullRef = (item.Handle as string) ?? item.DisplayName ?? string.Empty; PublishNotification(handle, RefOf(item), dv);
}
private static string RefOf(MonitoredItem item) => (item.Handle as string) ?? item.DisplayName ?? string.Empty;
private void PublishNotification(OpcUaSubscriptionHandle handle, string fullRef, DataValue dv)
{
_diag.Record(OpcUaClientDiagnostics.DataChange); _diag.Record(OpcUaClientDiagnostics.DataChange);
if (StatusCode.IsBad(dv.StatusCode)) _diag.Record(OpcUaClientDiagnostics.BadStatusValue); if (StatusCode.IsBad(dv.StatusCode)) _diag.Record(OpcUaClientDiagnostics.BadStatusValue);
// #670: a delivered notification is liveness evidence — the subscription path never calls ReadAsync, so
// without this a busy subscription-only driver's LastSuccessfulRead stayed at its connect time.
RecordLivenessEvidence();
var snapshot = new DataValueSnapshot( var snapshot = new DataValueSnapshot(
Value: dv.Value, Value: dv.Value,
StatusCode: dv.StatusCode.Code, StatusCode: dv.StatusCode.Code,
SourceTimestampUtc: dv.SourceTimestamp == DateTime.MinValue ? null : dv.SourceTimestamp, SourceTimestampUtc: dv.SourceTimestamp == DateTime.MinValue ? null : dv.SourceTimestamp,
ServerTimestampUtc: dv.ServerTimestamp == DateTime.MinValue ? DateTime.UtcNow : dv.ServerTimestamp); ServerTimestampUtc: dv.ServerTimestamp == DateTime.MinValue ? DateTime.UtcNow : dv.ServerTimestamp);
_lastValues[fullRef] = snapshot;
_awaitingRecovery.TryRemove(fullRef, out _);
OnDataChange?.Invoke(this, new DataChangeEventArgs(handle, fullRef, snapshot)); OnDataChange?.Invoke(this, new DataChangeEventArgs(handle, fullRef, snapshot));
} }
/// <summary>
/// #670 — advance <see cref="DriverHealth.LastSuccessfulRead"/> on upstream traffic (a notification), at most
/// once per second. The state is left as it is: a notification is evidence of comms, but the keep-alive
/// is what declares an outage over.
/// </summary>
private void RecordLivenessEvidence()
{
var now = Clock.GetUtcNow().UtcDateTime;
var last = Interlocked.Read(ref _lastEvidenceStampTicks);
if (now.Ticks - last < TimeSpan.TicksPerSecond && last != 0) return;
Interlocked.Exchange(ref _lastEvidenceStampTicks, now.Ticks);
var h = _health;
if (h.State is DriverState.Healthy or DriverState.Degraded)
_health = h with { LastSuccessfulRead = now };
}
/// <summary> /// <summary>
/// Live data-change subscription bookkeeping. Holds the SDK <see cref="Subscription"/>, /// Live data-change subscription bookkeeping. Holds the SDK <see cref="Subscription"/>,
/// the local handle, and the per-MonitoredItem (item, handler) pairs so /// the local handle, and the per-MonitoredItem (item, handler) pairs so
@@ -1628,8 +1763,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
// match in O(1) without re-parsing on every event. // match in O(1) without re-parsing on every event.
// #619: under v3 the host passes RawPaths, never upstream node ids. Translate each to the upstream // #619: under v3 the host passes RawPaths, never upstream node ids. Translate each to the upstream
// source id its tag authored (TagConfig.nodeId) so the per-event match compares like with like. // source id its tag authored (TagConfig.nodeId) so the per-event match compares like with like.
// Keys are server-stable (AlarmIndexKey) so a namespace reorder across an in-place SDK reconnect — // Keys are server-stable (AlarmIndexKey) so a remote namespace reorder cannot orphan them.
// which transfers this subscription without re-running SubscribeAlarmsAsync — cannot orphan them.
var sessionForResolve = Session; var sessionForResolve = Session;
var mapForResolve = _namespaceMap; var mapForResolve = _namespaceMap;
var sourceFilter = BuildAlarmSourceIndex(sourceNodeIds, raw => var sourceFilter = BuildAlarmSourceIndex(sourceNodeIds, raw =>
@@ -1709,7 +1843,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
try try
{ {
// Re-read Session inside the critical section: a reconnect completing while we were // Re-read Session inside the critical section: a reconnect completing while we were
// blocked on _gate has already swapped in the new session via OnReconnectComplete. // blocked on _gate has already swapped in the new session via a re-initialise.
var session = Session var session = Session
?? throw new InvalidOperationException( ?? throw new InvalidOperationException(
"OpcUaClient session was lost before the alarm subscription could be created."); "OpcUaClient session was lost before the alarm subscription could be created.");
@@ -1815,7 +1949,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
try try
{ {
// Re-read Session inside the critical section: a reconnect completing while we were // Re-read Session inside the critical section: a reconnect completing while we were
// blocked on _gate has already swapped in the new session via OnReconnectComplete. // blocked on _gate has already swapped in the new session via a re-initialise.
// ConditionId parsing is namespace-relative, so it must use the current session's // ConditionId parsing is namespace-relative, so it must use the current session's
// namespace table too. // namespace table too.
var session = Session; var session = Session;
@@ -1914,9 +2048,10 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
{ {
if (args.NotificationValue is not EventFieldList efl) return; if (args.NotificationValue is not EventFieldList efl) return;
// #662 review: only a subscription SubscribeAlarmsAsync completed (and nothing has since released) may // #662 review: only a subscription SubscribeAlarmsAsync completed (and nothing has since released) may
// deliver. A half-built one whose cleanup failed against a broken server can survive on the session and // deliver. A half-built one whose cleanup failed against a broken server can survive on the session;
// be transferred by an SDK reconnect; without this it would feed RouteAlarmEvent alongside the real one. // without this it would feed RouteAlarmEvent alongside the real one.
if (!_alarmSubscriptions.ContainsKey(handle.Id)) return; if (!_alarmSubscriptions.ContainsKey(handle.Id)) return;
RecordLivenessEvidence();
RouteAlarmEvent(efl.EventFields, sourceFilter, handle); RouteAlarmEvent(efl.EventFields, sourceFilter, handle);
} }
@@ -2063,10 +2198,9 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
} }
/// <summary> /// <summary>
/// #619 — ConditionRefresh every alarm subscription on the current session: after an in-place reconnect /// #619 — ConditionRefresh every alarm subscription on the current session (matched by display name): after
/// (transferred or SDK-recreated subscriptions — matched by display name, since a recreate clones the /// a keep-alive outage the session survived (#670) and on a RefreshRequired event. Events missed during the
/// <see cref="Subscription"/> objects) and on a RefreshRequired event. Events missed during the outage /// outage reach the host through the normal deduplicated path.
/// reach the host through the normal deduplicated path.
/// </summary> /// </summary>
/// <param name="cancellationToken">Cancellation.</param> /// <param name="cancellationToken">Cancellation.</param>
/// <returns>A task completing when every refresh call has returned.</returns> /// <returns>A task completing when every refresh call has returned.</returns>
@@ -2269,7 +2403,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
internal readonly record struct AlarmAckTarget(string ConditionRef, byte[] EventId); internal readonly record struct AlarmAckTarget(string ConditionRef, byte[] EventId);
/// <summary>#619 — the key both sides of the alarm index use: server-stable (<c>nsu=</c>) when a map is /// <summary>#619 — the key both sides of the alarm index use: server-stable (<c>nsu=</c>) when a map is
/// available, so a remote namespace-table reorder across an in-place reconnect cannot orphan it.</summary> /// available, so a remote namespace-table reorder cannot orphan it.</summary>
/// <param name="map">The current session's namespace map, or null.</param> /// <param name="map">The current session's namespace map, or null.</param>
/// <param name="nodeId">The NodeId to key.</param> /// <param name="nodeId">The NodeId to key.</param>
/// <returns>The index key.</returns> /// <returns>The index key.</returns>
@@ -2367,174 +2501,159 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
}; };
/// <summary> /// <summary>
/// Session keep-alive handler. On a healthy ping, bumps HostState back to Running /// Session keep-alive handler (Gitea #670). A good keep-alive is liveness evidence: it advances
/// (typical bounce after a transient network blip). On a bad ping, starts the SDK's /// <see cref="DriverHealth.LastSuccessfulRead"/>, and after a bad one it ends the outage in place (the session
/// <see cref="SessionReconnectHandler"/> which retries on the configured period + /// survived — e.g. a network blip): health back to Healthy and the outage's Bad refs re-read and re-published
/// fires <see cref="OnReconnectComplete"/> when it lands a new session. /// Good, because a server whose subscription survived will not resend values that did not change. A bad
/// keep-alive degrades health (stamp preserved) and, once per outage, publishes <c>BadNoCommunication</c> for
/// every subscribed ref (value retained) plus a quality-only annotation on every subscribed alarm condition.
/// </summary> /// </summary>
/// <remarks>
/// There is deliberately no <c>SessionReconnectHandler</c> any more. In SDK 1.5.378 it retried forever, its
/// session-recreate path cloned subscriptions the driver never re-tracked, and it kept health Healthy while
/// it did so — the #670 symptom. A session that does not come back keeps health Degraded with the read stamp
/// frozen, which is exactly what the host's #527 comms-loss watchdog trips on: ShutdownAsync + InitializeAsync
/// + re-subscribe, whose initial values come back Good.
/// </remarks>
private void OnKeepAlive(ISession sender, KeepAliveEventArgs e) private void OnKeepAlive(ISession sender, KeepAliveEventArgs e)
{ {
// A late callback from a session already released (re-initialise) says nothing about the current one.
if (!ReferenceEquals(sender, Session)) return;
var now = Clock.GetUtcNow().UtcDateTime;
if (!ServiceResult.IsBad(e.Status)) if (!ServiceResult.IsBad(e.Status))
{ {
if (Interlocked.Exchange(ref _sessionLost, 0) == 1)
{
_health = new DriverHealth(DriverState.Healthy, now, null);
TransitionTo(HostState.Running);
_diag.Record(OpcUaClientDiagnostics.KeepAliveRecovered);
_logger.LogInformation(
"OpcUaClientDriver '{DriverInstanceId}': upstream keep-alive recovered on the same session; " +
"re-reading the refs published Bad during the outage.", _driverInstanceId);
_ = Task.Run(RecoverAfterOutageAsync);
return;
}
var h = _health;
if (h.State is DriverState.Healthy or DriverState.Degraded)
_health = h with { LastSuccessfulRead = now };
TransitionTo(HostState.Running); TransitionTo(HostState.Running);
return; return;
} }
TransitionTo(HostState.Stopped); TransitionTo(HostState.Stopped);
_diag.Record(OpcUaClientDiagnostics.KeepAliveMiss); _diag.Record(OpcUaClientDiagnostics.KeepAliveMiss);
_health = new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, $"keep-alive: {e.Status}");
if (Interlocked.Exchange(ref _sessionLost, 1) == 1) return;
// Kick off the SDK's reconnect loop exactly once per drop. Keep-alive callbacks fire _logger.LogWarning(
// from the SDK keep-alive timer thread and the SDK can fire this handler repeatedly "OpcUaClientDriver '{DriverInstanceId}': upstream session keep-alive failed ({Status}); subscribed values " +
// while the channel stays down — the check-then-set must be atomic, otherwise two "are BadNoCommunication until it recovers or the host's comms-loss watchdog reconnects the driver.",
// callbacks both observe null, both construct a SessionReconnectHandler, and the _driverInstanceId, e.Status);
// second assignment leaks the first (its retry loop keeps running, unreferenced and PublishCommsLoss(now);
// never disposed). Guard with _probeLock.
SessionReconnectHandler handler;
lock (_probeLock)
{
if (_reconnectHandler is not null || _disposed) return;
handler = new SessionReconnectHandler(telemetry: null!,
reconnectAbort: false,
maxReconnectPeriod: (int)TimeSpan.FromMinutes(2).TotalMilliseconds);
_reconnectHandler = handler;
}
// BeginReconnect is started outside the lock — it does no _reconnectHandler mutation
// and we don't want to hold _probeLock across an SDK call.
_diag.Record(OpcUaClientDiagnostics.ReconnectAttempt);
handler.BeginReconnect(
sender,
(int)_options.ReconnectPeriod.TotalMilliseconds,
OnReconnectComplete);
} }
/// <summary> /// <summary>
/// Called by <see cref="SessionReconnectHandler"/> when its retry loop has either /// #670 — publish the start of an outage: <c>BadNoCommunication</c> once per (subscription, ref) with the last
/// successfully swapped to a new session or given up. Reads the new session off /// delivered value retained, and a quality-only annotation on every subscribed alarm condition (no state
/// <c>handler.Session</c>, unwires the old keep-alive hook, rewires for the new /// change — an outage says nothing about whether the alarm is active).
/// one, and tears down the handler. Subscription migration is already handled
/// inside the SDK via <c>TransferSubscriptions</c> (the SDK calls it automatically
/// when <see cref="Session.TransferSubscriptionsOnReconnect"/> is <c>true</c>,
/// which is the default).
/// </summary> /// </summary>
private void OnReconnectComplete(object? sender, EventArgs e) private void PublishCommsLoss(DateTime now)
{ {
if (sender is not SessionReconnectHandler handler) return; foreach (var rs in _subscriptions.Values)
var newSession = handler.Session;
// All mutations to Session and _reconnectHandler run under _probeLock so
// OnReconnectComplete, OnKeepAlive, and ShutdownAsync cannot race each other:
// a session swap visible to concurrent ReadAsync/WriteAsync/DiscoverAsync callers
// (which re-read Session inside _gate) must be atomic w.r.t. disposal and
// re-arming.
ISession? oldSession;
lock (_probeLock)
{ {
oldSession = Session; foreach (var pair in rs.ItemHandlers)
// Rewire keep-alive before swapping the reference so a hot keep-alive can't
// fire against the old session after we've already assigned the new one.
if (oldSession is not null && _keepAliveHandler is not null)
{ {
try { oldSession.KeepAlive -= _keepAliveHandler; } catch { } var fullRef = RefOf(pair.Item);
} _lastValues.TryGetValue(fullRef, out var last);
if (newSession is not null && _keepAliveHandler is not null) _awaitingRecovery[fullRef] = 0;
{ RaiseDataChange(new DataChangeEventArgs(rs.Handle, fullRef, new DataValueSnapshot(
newSession.KeepAlive += _keepAliveHandler; last?.Value, StatusBadNoCommunication, last?.SourceTimestampUtc, now)));
}
// Detach-then-attach: the SDK may hand back the SAME session object, and -= on a handler that
// is not attached is a no-op, so this never double-counts.
if (oldSession is not null) DetachDiagnostics(oldSession);
if (newSession is not null) AttachDiagnostics(newSession);
Session = newSession;
// Retire the handler that just finished.
if (ReferenceEquals(_reconnectHandler, handler))
{
_reconnectHandler.Dispose();
_reconnectHandler = null;
} }
} }
AnnotateAlarmQuality(StatusBadNoCommunication, now);
if (newSession is not null)
{
// Reconnect succeeded. Rebuild the namespace map from the *new* session: the
// remote server may have reordered its namespace table across the restart that
// caused the drop (driver-specs.md §8). Stable nsu= references stored in the
// address space re-resolve correctly against this fresh map. v3: re-ingest the
// authored RawTags so RawPath resolution survives the reconnect too.
_namespaceMap = NamespaceMap.FromSession(newSession, _options.RawTags);
RecordSkippedTags(_namespaceMap);
TransitionTo(HostState.Running);
_health = new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null);
_diag.Record(OpcUaClientDiagnostics.ReconnectSuccess);
// #619: transitions during the outage were missed (a recreated subscription starts empty; a transferred
// one may have overflowed). ConditionRefresh re-reports current states; unchanged ones dedup to nothing.
if (!_alarmSubscriptions.IsEmpty)
_ = Task.Run(() => RefreshAllAlarmSubscriptionsAsync(CancellationToken.None));
return;
}
// The reconnect handler gave up — its retry loop exhausted the 2-minute
// maxReconnectPeriod and invoked the callback with a null Session. Without an
// explicit Faulted signal the driver is permanently wedged: no session, no live
// keep-alive to re-trigger OnKeepAlive, and the Core never learns it must offer an
// operator reinitialize. Surface Faulted so the Core fans out Bad quality and
// ReinitializeAsync becomes available, and arm a fresh reconnect attempt against
// the last-known session for an always-on gateway rather than abandoning recovery
// entirely.
TransitionTo(HostState.Faulted);
_diag.Record(OpcUaClientDiagnostics.ReconnectExhausted);
_health = new DriverHealth(
DriverState.Faulted, _health.LastSuccessfulRead,
"OPC UA session reconnect exhausted its retry window without recovering. " +
"The remote server is unreachable; reinitialize the driver once it is back.");
if (oldSession is not null && !_disposed)
TryRearmReconnect(handler, oldSession);
} }
/// <summary> /// <summary>
/// Arm a fresh reconnect attempt after a previous handler gave up. The OPC UA Client /// #670 — the session survived the outage: re-read the refs still awaiting recovery and publish them (a ref
/// driver gateways an always-on remote server, so abandoning recovery permanently is /// a notification already refreshed is skipped — fresh data beats the read), restore alarm-condition quality
/// the wrong default — a new <see cref="SessionReconnectHandler"/> keeps retrying so /// and ConditionRefresh so transitions missed during the outage are re-reported (deduplicated). A failed
/// the driver self-heals when the remote returns, while the Faulted health set by the /// read re-arms the outage so the next good keep-alive tries again.
/// caller still lets an operator force a clean reinitialize in the meantime.
/// </summary> /// </summary>
private void TryRearmReconnect(SessionReconnectHandler exhausted, ISession lastSession) private async Task RecoverAfterOutageAsync()
{ {
SessionReconnectHandler handler;
lock (_probeLock)
{
// Only re-arm if no other handler took over and we aren't shutting down.
if (_disposed || (_reconnectHandler is not null && !ReferenceEquals(_reconnectHandler, exhausted)))
return;
handler = new SessionReconnectHandler(telemetry: null!,
reconnectAbort: false,
maxReconnectPeriod: (int)TimeSpan.FromMinutes(2).TotalMilliseconds);
_reconnectHandler = handler;
}
try try
{ {
handler.BeginReconnect( var refs = _awaitingRecovery.Keys.ToList();
lastSession, if (refs.Count > 0 && Session is not null)
(int)_options.ReconnectPeriod.TotalMilliseconds,
OnReconnectComplete);
}
catch
{
// If the SDK refuses to re-arm (e.g. the last session is fully torn down), drop
// the handler so a later operator ReinitializeAsync isn't blocked by a stale one.
lock (_probeLock)
{ {
if (ReferenceEquals(_reconnectHandler, handler)) var (results, transportOk) = await ReadCoreAsync(refs, CancellationToken.None).ConfigureAwait(false);
if (!transportOk)
{ {
_reconnectHandler.Dispose(); Interlocked.Exchange(ref _sessionLost, 1);
_reconnectHandler = null; return;
}
for (var i = 0; i < refs.Count; i++)
{
if (!_awaitingRecovery.TryRemove(refs[i], out _)) continue; // a notification got there first
_lastValues[refs[i]] = results[i];
foreach (var rs in _subscriptions.Values)
if (rs.ItemHandlers.Any(p => RefOf(p.Item) == refs[i]))
RaiseDataChange(new DataChangeEventArgs(rs.Handle, refs[i], results[i]));
} }
} }
AnnotateAlarmQuality(StatusGood, Clock.GetUtcNow().UtcDateTime);
if (!_alarmSubscriptions.IsEmpty)
await RefreshAllAlarmSubscriptionsAsync(CancellationToken.None).ConfigureAwait(false);
}
catch (Exception ex)
{
_logger.LogWarning(ex, "OpcUaClientDriver '{DriverInstanceId}': in-place recovery after an outage failed.",
_driverInstanceId);
}
}
/// <summary>Quality-only annotation (#626 pattern) of every subscribed alarm condition.</summary>
private void AnnotateAlarmQuality(uint quality, DateTime now)
{
if (_alarmSubscriptions.IsEmpty || OnAlarmEvent is null) return;
var handle = _alarmSubscriptions.Values.First().Handle;
lock (_alarmStateLock)
{
if (_lastAlarmRefSet is null) return;
foreach (var conditionId in _lastAlarmRefSet)
{
_alarmStateByRawPath.TryGetValue(conditionId, out var told);
var e = new AlarmEventArgs(handle,
SourceNodeId: conditionId,
ConditionId: conditionId,
AlarmType: "ConditionType",
Message: told.Message ?? string.Empty,
Severity: told.Severity is { } sev ? MapSeverity((ushort)sev) : AlarmSeverity.Low,
SourceTimestampUtc: now,
Kind: AlarmTransitionKind.Unspecified,
OpcUaSeverity: told.Severity,
QualityStatusCode: quality);
try { OnAlarmEvent?.Invoke(this, e); }
catch (Exception ex)
{
_logger.LogWarning(ex, "OpcUaClientDriver '{DriverInstanceId}': alarm subscriber faulted on a quality " +
"annotation for {ConditionId}.", _driverInstanceId, conditionId);
}
}
}
}
/// <summary>Raise <see cref="OnDataChange"/>, isolating a faulting subscriber from the SDK callback thread.</summary>
private void RaiseDataChange(DataChangeEventArgs args)
{
try { OnDataChange?.Invoke(this, args); }
catch (Exception ex)
{
_logger.LogWarning(ex, "OpcUaClientDriver '{DriverInstanceId}': data subscriber faulted for {Ref}.",
_driverInstanceId, args.FullReference);
} }
} }
@@ -2552,7 +2671,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
} }
/// <summary> /// <summary>
/// Synchronous disposal. Cancels the reconnect handler and detaches the keep-alive /// Synchronous disposal. Detaches the keep-alive
/// hook synchronously (no async work on this hot path), then fires the cert-validation /// hook synchronously (no async work on this hot path), then fires the cert-validation
/// handler detach. The async session-close is intentionally skipped — it requires a /// handler detach. The async session-close is intentionally skipped — it requires a
/// live session + network round-trip and is unsafe to block-on from a potentially /// live session + network round-trip and is unsafe to block-on from a potentially
@@ -2564,12 +2683,8 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
if (_disposed) return; if (_disposed) return;
_disposed = true; _disposed = true;
// Cancel any in-flight reconnect handler.
SessionReconnectHandler? handlerToCancel;
lock (_probeLock) lock (_probeLock)
{ {
handlerToCancel = _reconnectHandler;
_reconnectHandler = null;
// Detach keep-alive and null Session so in-flight gated callers see null // Detach keep-alive and null Session so in-flight gated callers see null
// after their next _gate.WaitAsync — they return BadCommunicationError cleanly. // after their next _gate.WaitAsync — they return BadCommunicationError cleanly.
if (_keepAliveHandler is not null && Session is not null) if (_keepAliveHandler is not null && Session is not null)
@@ -2579,8 +2694,6 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
_keepAliveHandler = null; _keepAliveHandler = null;
Session = null; Session = null;
} }
try { handlerToCancel?.CancelReconnect(); } catch { }
handlerToCancel?.Dispose();
// Detach the cert-validation handler registered during InitializeAsync so the // Detach the cert-validation handler registered during InitializeAsync so the
// CertificateValidator (which may be process-shared) doesn't hold a reference to // CertificateValidator (which may be process-shared) doesn't hold a reference to
@@ -1,16 +1,35 @@
using Microsoft.Extensions.Logging;
using Opc.Ua; using Opc.Ua;
namespace ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient; namespace ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient;
/// <summary> /// <summary>
/// The one <see cref="ITelemetryContext" /> the driver hands the SDK's non-obsolete (telemetry-taking) /// The <see cref="ITelemetryContext" /> the driver hands the SDK's non-obsolete (telemetry-taking)
/// overloads. An empty builder keeps SDK internals off any logging provider (mirrors Client.Shared's /// overloads. <see cref="Context"/> has no logging provider (static helpers and tests);
/// <c>ClientTelemetry</c>). /// <see cref="ForLogger"/> routes the SDK's own Warning-and-above output (keep-alive, publish and
/// channel errors) to a driver instance's logger — without it every SDK-side session error was
/// invisible (Gitea #670).
/// </summary> /// </summary>
internal static class SdkTelemetry internal static class SdkTelemetry
{ {
private static readonly Lazy<ITelemetryContext> Instance = new(() => DefaultTelemetry.Create(_ => { })); private static readonly Lazy<ITelemetryContext> Instance = new(() => DefaultTelemetry.Create(_ => { }));
/// <summary>Gets the shared telemetry context.</summary> /// <summary>Gets the shared, provider-less telemetry context.</summary>
public static ITelemetryContext Context => Instance.Value; public static ITelemetryContext Context => Instance.Value;
/// <summary>Creates a telemetry context whose SDK log output at Warning and above goes to <paramref name="logger"/>.</summary>
/// <param name="logger">The driver instance's logger.</param>
/// <returns>The telemetry context.</returns>
public static ITelemetryContext ForLogger(ILogger logger) =>
DefaultTelemetry.Create(builder => builder
.SetMinimumLevel(LogLevel.Warning)
.AddProvider(new ForwardingLoggerProvider(logger)));
/// <summary>Hands every SDK logger category the one driver logger.</summary>
private sealed class ForwardingLoggerProvider(ILogger target) : ILoggerProvider
{
public ILogger CreateLogger(string categoryName) => target;
public void Dispose() { }
}
} }
@@ -579,8 +579,8 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
/// came back — live-confirmed against a stopped Modbus fixture, which needed a node restart. Both /// came back — live-confirmed against a stopped Modbus fixture, which needed a node restart. Both
/// comms-loss quality paths (#477 native conditions, #481 scripted-alarm inputs) hang off /// comms-loss quality paths (#477 native conditions, #481 scripted-alarm inputs) hang off
/// <see cref="ConnectivityChanged"/>, so both were wired to a signal that never fired. /// <see cref="ConnectivityChanged"/>, so both were wired to a signal that never fired.
/// <c>OpcUaClientDriver</c>'s own reconnect-exhausted branch already assumed this mechanism existed /// Since Gitea #670 this is also <c>OpcUaClientDriver</c>'s only recovery path for a lost upstream
/// ("the Core never learns it must offer an operator reinitialize") — it did not. /// session (its in-driver SDK reconnect handler was deleted).
/// </para> /// </para>
/// <para> /// <para>
/// <b>Why the trip condition is a conjunction, and NOT simply "non-Healthy for N polls".</b> /// <b>Why the trip condition is a conjunction, and NOT simply "non-Healthy for N polls".</b>
@@ -57,9 +57,7 @@ public sealed class OpcUaClientDiagnosticsTests
d.Record(OpcUaClientDiagnostics.BadStatusValue); d.Record(OpcUaClientDiagnostics.BadStatusValue);
d.Record(OpcUaClientDiagnostics.PublishError); d.Record(OpcUaClientDiagnostics.PublishError);
d.Record(OpcUaClientDiagnostics.KeepAliveMiss); d.Record(OpcUaClientDiagnostics.KeepAliveMiss);
d.Record(OpcUaClientDiagnostics.ReconnectAttempt); d.Record(OpcUaClientDiagnostics.KeepAliveRecovered);
d.Record(OpcUaClientDiagnostics.ReconnectSuccess);
d.Record(OpcUaClientDiagnostics.ReconnectExhausted);
cap[OpcUaClientDiagnostics.PublishResponse].ShouldBe(1); cap[OpcUaClientDiagnostics.PublishResponse].ShouldBe(1);
cap[OpcUaClientDiagnostics.Notification].ShouldBe(3); cap[OpcUaClientDiagnostics.Notification].ShouldBe(3);
@@ -67,9 +65,7 @@ public sealed class OpcUaClientDiagnosticsTests
cap[OpcUaClientDiagnostics.BadStatusValue].ShouldBe(1); cap[OpcUaClientDiagnostics.BadStatusValue].ShouldBe(1);
cap[OpcUaClientDiagnostics.PublishError].ShouldBe(1); cap[OpcUaClientDiagnostics.PublishError].ShouldBe(1);
cap[OpcUaClientDiagnostics.KeepAliveMiss].ShouldBe(1); cap[OpcUaClientDiagnostics.KeepAliveMiss].ShouldBe(1);
cap[OpcUaClientDiagnostics.ReconnectAttempt].ShouldBe(1); cap[OpcUaClientDiagnostics.KeepAliveRecovered].ShouldBe(1);
cap[OpcUaClientDiagnostics.ReconnectSuccess].ShouldBe(1);
cap[OpcUaClientDiagnostics.ReconnectExhausted].ShouldBe(1);
} }
[Fact] [Fact]
@@ -4,9 +4,10 @@ using Xunit;
namespace ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests; namespace ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests;
/// <summary> /// <summary>
/// Scaffold tests for <see cref="SessionReconnectHandler"/> wiring. Wire-level /// Scaffold tests for the reconnect options. Since Gitea #670 the driver has no in-driver reconnect
/// disconnect-reconnect-resume coverage against a live upstream server lands with the /// handler (session loss is reported and the host's comms-loss watchdog re-initialises the driver);
/// in-process fixture — too much machinery for a unit-test-only lane. /// <c>ReconnectPeriod</c> is retained for config compatibility only. Outage behaviour is covered by
/// <see cref="OpcUaClientSessionLossTests"/> and, over the wire, <c>OpcUaClientDriverReconnectWireTests</c>.
/// </summary> /// </summary>
[Trait("Category", "Unit")] [Trait("Category", "Unit")]
public sealed class OpcUaClientReconnectTests public sealed class OpcUaClientReconnectTests
@@ -26,13 +27,10 @@ public sealed class OpcUaClientReconnectTests
opts.ReconnectPeriod.ShouldBe(TimeSpan.FromMilliseconds(500)); opts.ReconnectPeriod.ShouldBe(TimeSpan.FromMilliseconds(500));
} }
/// <summary>Verifies that the driver starts with no reconnect handler active before initialization.</summary> /// <summary>Verifies that the driver starts in the Unknown state before initialization.</summary>
[Fact] [Fact]
public void Driver_starts_with_no_reconnect_handler_active_pre_init() public void Driver_starts_unknown_pre_init()
{ {
// The reconnect handler is lazy — spun up only when a bad keep-alive fires. Pre-init
// there's no session to reconnect, so the field must be null (indirectly verified by
// the lifecycle-shape test suite catching any accidental construction).
using var drv = new OpcUaClientDriver(new OpcUaClientDriverOptions(), "opcua-reconnect"); using var drv = new OpcUaClientDriver(new OpcUaClientDriverOptions(), "opcua-reconnect");
drv.GetHealth().State.ShouldBe(Core.Abstractions.DriverState.Unknown); drv.GetHealth().State.ShouldBe(Core.Abstractions.DriverState.Unknown);
} }
@@ -0,0 +1,295 @@
using System.Text.Json;
using Moq;
using Opc.Ua;
using Opc.Ua.Client;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
namespace ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests;
/// <summary>
/// Gitea #670 — upstream session loss must reach the driver's health and its subscribers. Before the fix a bad
/// keep-alive only flipped the host status: the subscribed values stayed Good forever, health stayed Healthy (so the
/// #527 comms-loss watchdog could never trip — its first conjunct is "non-Healthy"), and <c>LastSuccessfulRead</c>
/// only moved on init / <c>ReadAsync</c>, which a subscription-only driver never calls.
/// </summary>
[Trait("Category", "Unit")]
public sealed class OpcUaClientSessionLossTests
{
private const uint BadNoCommunication = 0x80310000u;
private const string SpeedRef = "Plant/dev/Speed";
private const string CounterRef = "Plant/dev/Counter";
private sealed class ManualClock(DateTimeOffset start) : TimeProvider
{
public DateTimeOffset Now { get; set; } = start;
public override DateTimeOffset GetUtcNow() => Now;
public void Advance(TimeSpan by) => Now += by;
}
private static RawTagEntry Tag(string rawPath, string nodeId) =>
new(rawPath, JsonSerializer.Serialize(new { nodeId }), WriteIdempotent: true);
private static readonly IReadOnlyList<RawTagEntry> Tags =
[
Tag(SpeedRef, "ns=0;s=Speed"),
Tag(CounterRef, "ns=0;s=Counter"),
];
private static Mock<ISession> NewSession()
{
var mock = new Mock<ISession>(MockBehavior.Loose);
mock.SetupGet(s => s.MessageContext)
.Returns(new ServiceMessageContext(SdkTelemetry.Context));
mock.SetupGet(s => s.NamespaceUris).Returns(new NamespaceTable());
return mock;
}
private static (OpcUaClientDriver Driver, Mock<ISession> Session, ManualClock Clock) NewConnectedDriver()
{
var drv = new OpcUaClientDriver(new OpcUaClientDriverOptions { RawTags = Tags }, "opcua-670");
var session = NewSession();
var clock = new ManualClock(new DateTimeOffset(2026, 10, 11, 12, 0, 0, TimeSpan.Zero));
drv.Clock = clock;
drv.SetSessionForTest(session.Object);
drv.SetNamespaceMapForTest(NamespaceMap.FromTable(new NamespaceTable(), Tags));
drv.SetHealthForTest(new DriverHealth(DriverState.Healthy, clock.Now.UtcDateTime, null));
return (drv, session, clock);
}
private static DataValue Good(object value, DateTime ts) =>
new(new Variant(value), StatusCodes.Good, ts, ts);
private static void SetupRead(Mock<ISession> session, params object[] values) =>
session.Setup(s => s.ReadAsync(
It.IsAny<RequestHeader>(), It.IsAny<double>(), It.IsAny<TimestampsToReturn>(),
It.IsAny<ReadValueIdCollection>(), It.IsAny<CancellationToken>()))
.ReturnsAsync((RequestHeader _, double _, TimestampsToReturn _, ReadValueIdCollection nodes, CancellationToken _) =>
new ReadResponse
{
ResponseHeader = new ResponseHeader(),
Results = [.. nodes.Select((_, i) => new DataValue(new Variant(values[i]), StatusCodes.Good, DateTime.UtcNow, DateTime.UtcNow))],
DiagnosticInfos = [],
});
private static async Task WaitForAsync(Func<bool> condition, string because)
{
var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(10);
while (DateTime.UtcNow < deadline)
{
if (condition()) return;
await Task.Delay(10);
}
condition().ShouldBeTrue(because);
}
[Fact]
public void A_bad_keep_alive_degrades_health_and_preserves_the_last_successful_read()
{
var (drv, _, clock) = NewConnectedDriver();
using var _ = drv;
var before = drv.GetHealth().LastSuccessfulRead;
clock.Advance(TimeSpan.FromSeconds(5));
drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication);
var h = drv.GetHealth();
h.State.ShouldNotBe(DriverState.Healthy, "a lost session must be visible to the #527 watchdog's first conjunct");
h.State.ShouldBe(DriverState.Degraded);
h.LastSuccessfulRead.ShouldBe(before, "the watchdog measures silence from this stamp, so loss must not move it");
h.LastError.ShouldNotBeNull().ShouldContain("keep-alive");
}
[Fact]
public void A_bad_keep_alive_publishes_BadNoCommunication_once_per_ref_retaining_the_value()
{
var (drv, _, clock) = NewConnectedDriver();
using var _ = drv;
var handle = drv.AddSubscriptionForTest([SpeedRef, CounterRef]);
var t0 = clock.Now.UtcDateTime;
drv.DeliverNotificationForTest(handle, SpeedRef, Good(42.5f, t0));
drv.DeliverNotificationForTest(handle, CounterRef, Good(7, t0));
var seen = new List<DataChangeEventArgs>();
drv.OnDataChange += (_, e) => seen.Add(e);
drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication);
seen.Count.ShouldBe(2, "exactly one Bad per subscribed ref");
var speed = seen.Single(e => e.FullReference == SpeedRef);
speed.Snapshot.StatusCode.ShouldBe(BadNoCommunication);
speed.Snapshot.Value.ShouldBe(42.5f, "the value is retained — only the quality changes (as the host does)");
speed.SubscriptionHandle.ShouldBe(handle);
seen.Single(e => e.FullReference == CounterRef).Snapshot.Value.ShouldBe(7);
// The SDK re-raises a bad keep-alive every interval while the channel stays down: no further publishes.
drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication);
drv.RaiseKeepAliveForTest(StatusCodes.BadConnectionClosed);
seen.Count.ShouldBe(2, "a second bad keep-alive in the same outage must publish nothing");
}
[Fact]
public void Good_keep_alives_and_notifications_advance_the_last_successful_read()
{
var (drv, _, clock) = NewConnectedDriver();
using var _ = drv;
var handle = drv.AddSubscriptionForTest([SpeedRef]);
clock.Advance(TimeSpan.FromSeconds(30));
drv.RaiseKeepAliveForTest(StatusCodes.Good);
drv.GetHealth().LastSuccessfulRead.ShouldBe(clock.Now.UtcDateTime);
clock.Advance(TimeSpan.FromSeconds(30));
drv.DeliverNotificationForTest(handle, SpeedRef, Good(1.0f, clock.Now.UtcDateTime));
drv.GetHealth().LastSuccessfulRead.ShouldBe(clock.Now.UtcDateTime);
drv.GetHealth().State.ShouldBe(DriverState.Healthy);
}
/// <summary>
/// Falsification control for the R2 half of the fix: a long-running, idle, subscription-only driver whose
/// server is perfectly alive must never look dead to the #527 watchdog (non-Healthy AND silent ≥ 90 s).
/// Without good keep-alives advancing <c>LastSuccessfulRead</c>, fixing R1 alone would trip it on the first
/// bad tick of any connection older than 90 s.
/// </summary>
[Fact]
public void An_idle_healthy_subscription_driver_never_looks_dead_to_the_watchdog()
{
var (drv, _, clock) = NewConnectedDriver();
using var _ = drv;
drv.AddSubscriptionForTest([SpeedRef]); // static value — no notifications ever arrive
for (var i = 0; i < 60; i++) // 5 minutes of 5 s keep-alives
{
clock.Advance(TimeSpan.FromSeconds(5));
drv.RaiseKeepAliveForTest(StatusCodes.Good);
var h = drv.GetHealth();
h.State.ShouldBe(DriverState.Healthy);
(clock.Now.UtcDateTime - h.LastSuccessfulRead!.Value).ShouldBeLessThan(TimeSpan.FromSeconds(90));
}
}
[Fact]
public async Task Recovery_on_the_same_session_republishes_the_identical_value_Good()
{
var (drv, session, clock) = NewConnectedDriver();
using var _ = drv;
var handle = drv.AddSubscriptionForTest([SpeedRef]);
drv.DeliverNotificationForTest(handle, SpeedRef, Good(42.5f, clock.Now.UtcDateTime));
SetupRead(session, 42.5f); // the server still holds the SAME value — it will not resend it
var seen = new List<DataChangeEventArgs>();
drv.OnDataChange += (_, e) => { lock (seen) seen.Add(e); };
drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication);
clock.Advance(TimeSpan.FromSeconds(20));
drv.RaiseKeepAliveForTest(StatusCodes.Good);
drv.GetHealth().State.ShouldBe(DriverState.Healthy);
drv.GetHealth().LastSuccessfulRead.ShouldBe(clock.Now.UtcDateTime);
await WaitForAsync(() => { lock (seen) return seen.Count >= 2; }, "the recovery read must publish");
lock (seen)
{
seen.Count.ShouldBe(2);
seen[0].Snapshot.StatusCode.ShouldBe(BadNoCommunication);
seen[1].FullReference.ShouldBe(SpeedRef);
seen[1].SubscriptionHandle.ShouldBe(handle);
seen[1].Snapshot.StatusCode.ShouldBe(0u);
seen[1].Snapshot.Value.ShouldBe(42.5f, "the IDENTICAL value must come back Good — a (value,status) dedup lets it through");
}
}
[Fact]
public async Task A_notification_during_recovery_wins_over_the_recovery_read()
{
var (drv, session, clock) = NewConnectedDriver();
using var _ = drv;
var handle = drv.AddSubscriptionForTest([SpeedRef]);
drv.DeliverNotificationForTest(handle, SpeedRef, Good(1.0f, clock.Now.UtcDateTime));
drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication);
// A fresh notification arrives before the keep-alive recovers: the read must not overwrite it.
drv.DeliverNotificationForTest(handle, SpeedRef, Good(2.0f, clock.Now.UtcDateTime));
SetupRead(session, 1.0f);
var seen = new List<DataChangeEventArgs>();
drv.OnDataChange += (_, e) => { lock (seen) seen.Add(e); };
drv.RaiseKeepAliveForTest(StatusCodes.Good);
await Task.Delay(200, TestContext.Current.CancellationToken);
lock (seen) seen.ShouldBeEmpty();
}
[Fact]
public void A_bad_keep_alive_annotates_subscribed_alarm_conditions_quality_only_once()
{
var (drv, _, _) = NewConnectedDriver();
using var _ = drv;
drv.AddAlarmSubscriptionForTest([SpeedRef]);
var seen = new List<AlarmEventArgs>();
drv.OnAlarmEvent += (_, e) => seen.Add(e);
drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication);
drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication);
var e = seen.ShouldHaveSingleItem();
e.ConditionId.ShouldBe(SpeedRef);
e.IsQualityOnly.ShouldBeTrue("an outage says nothing about the alarm state — quality only, no transition");
e.QualityStatusCode.ShouldBe(BadNoCommunication);
}
[Fact]
public async Task Recovery_restores_alarm_condition_quality_Good()
{
var (drv, session, _) = NewConnectedDriver();
using var _ = drv;
SetupRead(session);
drv.AddAlarmSubscriptionForTest([SpeedRef]);
var seen = new List<AlarmEventArgs>();
drv.OnAlarmEvent += (_, e) => { lock (seen) seen.Add(e); };
drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication);
drv.RaiseKeepAliveForTest(StatusCodes.Good);
await WaitForAsync(() => { lock (seen) return seen.Count >= 2; }, "recovery must re-annotate Good");
lock (seen)
{
seen[1].IsQualityOnly.ShouldBeTrue();
seen[1].QualityStatusCode.ShouldBe(0u);
}
}
[Fact]
public void A_keep_alive_from_a_superseded_session_is_ignored()
{
var (drv, _, _) = NewConnectedDriver();
using var _ = drv;
var stale = NewSession();
drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication, stale.Object);
drv.GetHealth().State.ShouldBe(DriverState.Healthy);
}
[Fact]
public async Task Shutdown_clears_the_outage_so_a_fresh_session_reports_its_own_loss()
{
var (drv, _, _) = NewConnectedDriver();
using var _ = drv;
drv.AddSubscriptionForTest([SpeedRef]);
drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication);
await drv.ShutdownAsync(TestContext.Current.CancellationToken);
// As after a watchdog reinit: a new session + subscription, and a new outage must be reported again.
var fresh = NewSession();
drv.SetSessionForTest(fresh.Object);
drv.SetNamespaceMapForTest(NamespaceMap.FromTable(new NamespaceTable(), Tags));
drv.AddSubscriptionForTest([SpeedRef]);
var seen = new List<DataChangeEventArgs>();
drv.OnDataChange += (_, e) => seen.Add(e);
drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication);
seen.ShouldHaveSingleItem().Snapshot.StatusCode.ShouldBe(BadNoCommunication);
}
}
@@ -6,14 +6,19 @@ using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests; namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests;
/// <summary> /// <summary>
/// Reconnect + resume leg of the OpcUaClient driver's wire coverage (Gitea #622): the in-process server is /// Outage + recovery leg of the OpcUaClient driver's wire coverage (Gitea #622, reworked for #670): the in-process
/// stopped and a fresh one is booted on the SAME port / application URI / PKI store — a process restart as the /// server is stopped and a fresh one is booted on the SAME port / application URI / PKI store — a process restart
/// driver sees it, with an empty address space the test then re-creates. The driver must detect the loss, /// as the driver sees it, with an empty address space the test then re-creates. The driver must REPORT the loss
/// rebuild a session, and its existing subscription must deliver again with no caller action. /// (subscribed value BadNoCommunication with the value retained, health non-Healthy — what the host's #527
/// comms-loss watchdog trips on). Recovery is the watchdog's sequence, replayed here as <c>DriverInstanceActor</c>
/// runs it: ShutdownAsync, InitializeAsync, re-subscribe — after which the value comes back Good. There is no
/// in-driver reconnect any more (#670: the SDK reconnect handler kept everything Good and Healthy forever).
/// </summary> /// </summary>
[Trait("Category", "Integration")] [Trait("Category", "Integration")]
public sealed class OpcUaClientDriverReconnectWireTests public sealed class OpcUaClientDriverReconnectWireTests
{ {
private const uint BadNoCommunication = 0x80310000u;
private static async Task WaitAsync(Func<bool> condition, string because, TimeSpan? timeout = null) private static async Task WaitAsync(Func<bool> condition, string because, TimeSpan? timeout = null)
{ {
var deadline = DateTime.UtcNow + (timeout ?? TimeSpan.FromSeconds(60)); var deadline = DateTime.UtcNow + (timeout ?? TimeSpan.FromSeconds(60));
@@ -26,53 +31,66 @@ public sealed class OpcUaClientDriverReconnectWireTests
} }
[Fact] [Fact]
public async Task After_a_server_restart_the_driver_reconnects_and_the_subscription_resumes_delivering() public async Task A_server_restart_goes_Bad_then_the_watchdog_reinit_brings_the_value_back_Good()
{ {
var ct = TestContext.Current.CancellationToken; var ct = TestContext.Current.CancellationToken;
var rig = await OpcUaClientDriverWireTests.StartRigAsync("drv-reconnect", ct); var rig = await OpcUaClientDriverWireTests.StartRigAsync("drv-reconnect", ct);
await using var _rig = rig; await using var _rig = rig;
OpcUaClientDriverWireTests.Publish(rig, OpcUaClientDriverWireTests.SpeedId, 10.0f); OpcUaClientDriverWireTests.Publish(rig, OpcUaClientDriverWireTests.SpeedId, 10.0f);
await using var drv = OpcUaClientDriverWireTests.NewDriver( await using var drv = OpcUaClientDriverWireTests.NewDriver(rig, "wire-reconnect");
rig, "wire-reconnect", reconnectPeriod: TimeSpan.FromSeconds(1));
await drv.InitializeAsync("{}", ct); await drv.InitializeAsync("{}", ct);
var seen = new List<float>(); var seen = new List<(uint Status, object? Value)>();
drv.OnDataChange += (_, e) => drv.OnDataChange += (_, e) =>
{ {
if (e.FullReference == OpcUaClientDriverWireTests.SpeedRef && e.Snapshot.StatusCode == 0u) if (e.FullReference == OpcUaClientDriverWireTests.SpeedRef)
lock (seen) seen.Add(Convert.ToSingle(e.Snapshot.Value)); lock (seen) seen.Add((e.Snapshot.StatusCode, e.Snapshot.Value));
}; };
var states = new List<HostState>();
drv.OnHostStatusChanged += (_, e) => { lock (states) states.Add(e.NewState); };
await drv.SubscribeAsync([OpcUaClientDriverWireTests.SpeedRef], TimeSpan.FromMilliseconds(100), ct); await drv.SubscribeAsync([OpcUaClientDriverWireTests.SpeedRef], TimeSpan.FromMilliseconds(100), ct);
await WaitAsync(() => { lock (seen) return seen.Contains(10.0f); }, "initial value must arrive before the outage"); await WaitAsync(() => { lock (seen) return seen.Any(s => s is (0u, 10.0f)); },
"initial value must arrive before the outage");
// Outage: the loss must be DETECTED (host leaves Running) while the server is genuinely down. // Outage: the loss must be REPORTED while the server is genuinely down — Bad on the value with the value
// retained, and a health state the watchdog can act on.
await rig.RestartAsync(ct, whileDown: async () => await rig.RestartAsync(ct, whileDown: async () =>
{ {
await WaitAsync( await WaitAsync(() => { lock (seen) return seen.Any(s => s.Status == BadNoCommunication); },
() => { lock (states) return states.Any(s => s != HostState.Running); }, "a lost session must publish BadNoCommunication for the subscribed ref", TimeSpan.FromSeconds(30));
"the driver must notice the server is gone", TimeSpan.FromSeconds(30)); lock (seen) seen.Single(s => s.Status == BadNoCommunication).Value.ShouldBe(10.0f);
drv.GetHealth().State.ShouldNotBe(DriverState.Healthy);
drv.GetHostStatuses().Single().State.ShouldNotBe(HostState.Running);
}); });
// The new server is empty: redeploy the nodes and publish a value no earlier delivery could have carried. // The restarted server knows nothing of the old session, so the driver stays down on its own …
await Task.Delay(TimeSpan.FromSeconds(3), ct);
drv.GetHealth().State.ShouldNotBe(DriverState.Healthy, "nothing in the driver may paper over a dead session");
lock (seen) seen.Count(s => s.Status == BadNoCommunication).ShouldBe(1, "one Bad per outage, not per keep-alive");
// … until the host's comms-loss watchdog re-initialises it (DriverInstanceActor: Reconnecting → ShutdownAsync +
// InitializeAsync → Connected → re-subscribe). The new server is empty: redeploy the nodes first, holding the
// SAME value as before the outage.
rig.EnsureRawVariable("Speed", "Float", writable: true); rig.EnsureRawVariable("Speed", "Float", writable: true);
rig.EnsureRawVariable("Counter", "Int32", writable: false); rig.EnsureRawVariable("Counter", "Int32", writable: false);
OpcUaClientDriverWireTests.Publish(rig, OpcUaClientDriverWireTests.SpeedId, 77.0f); OpcUaClientDriverWireTests.Publish(rig, OpcUaClientDriverWireTests.SpeedId, 10.0f);
await WaitAsync(() => drv.GetHostStatuses().Single().State == HostState.Running, await drv.ShutdownAsync(ct);
"the driver must reconnect after the server is back"); await drv.InitializeAsync("{}", ct);
await WaitAsync(() => { lock (seen) return seen.Contains(77.0f); }, drv.GetHealth().State.ShouldBe(DriverState.Healthy);
"the pre-outage subscription must resume and deliver the post-restart value"); int before;
lock (seen) before = seen.Count;
await drv.SubscribeAsync([OpcUaClientDriverWireTests.SpeedRef], TimeSpan.FromMilliseconds(100), ct);
// The IDENTICAL value comes back Good — a (value, status) dedup downstream lets it through.
await WaitAsync(() => { lock (seen) return seen.Skip(before).Any(s => s is (0u, 10.0f)); },
"the re-subscribed ref must come back Good with its value");
// Ongoing delivery, not just a one-off initial value.
OpcUaClientDriverWireTests.Publish(rig, OpcUaClientDriverWireTests.SpeedId, 88.0f); OpcUaClientDriverWireTests.Publish(rig, OpcUaClientDriverWireTests.SpeedId, 88.0f);
await WaitAsync(() => { lock (seen) return seen.Contains(88.0f); }, await WaitAsync(() => { lock (seen) return seen.Any(s => s is (0u, 88.0f)); },
"a later change must also reach the resumed subscription"); "a later change must also reach the new subscription");
drv.GetHostStatuses().Single().State.ShouldBe(HostState.Running);
// And the data plane works for plain reads on the new session.
var back = await drv.ReadAsync([OpcUaClientDriverWireTests.SpeedRef], ct); var back = await drv.ReadAsync([OpcUaClientDriverWireTests.SpeedRef], ct);
back[0].StatusCode.ShouldBe(0u); back[0].StatusCode.ShouldBe(0u);
Convert.ToSingle(back[0].Value).ShouldBe(88.0f); Convert.ToSingle(back[0].Value).ShouldBe(88.0f);