diff --git a/CLAUDE.md b/CLAUDE.md index baa899c2..033898ab 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -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. - **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 - 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. - **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` diff --git a/docs/DriverLifecycle.md b/docs/DriverLifecycle.md index a84a2310..df8ab2ec 100644 --- a/docs/DriverLifecycle.md +++ b/docs/DriverLifecycle.md @@ -422,7 +422,10 @@ Two consequences worth knowing: - **`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: 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. - **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 diff --git a/docs/drivers/OpcUaClient.md b/docs/drivers/OpcUaClient.md index 376a265e..b9abdd31 100644 --- a/docs/drivers/OpcUaClient.md +++ b/docs/drivers/OpcUaClient.md @@ -45,22 +45,46 @@ out-of-process isolation. ``` `EndpointUrls` (a list) takes precedence over the single-URL `EndpointUrl` and -provides ordered **failover** — the driver tries each candidate in turn at init -and on session drop, and the first to connect wins (e.g. a hot-standby pair on +provides ordered **failover** — the driver tries each candidate in turn at every +`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 `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Contracts/OpcUaClientDriverOptions.cs` 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 -A single `Session` per driver instance; subscriptions multiplex onto it. 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 +A single `Session` per driver instance; subscriptions multiplex onto it. Stored NodeIds embed the server-stable namespace **URI** (not the session-relative `ns=N` index) so a remote namespace-table reorder across a 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: "`, 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) 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 > `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. -> **`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` > 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 -> 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 > 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 `otopcua.driver.opcuaclient.events`, tagged `driver_instance` (the driver instance id) and `event` = `publish_response`, `notification` (items), `data_change`, `bad_status_value`, `publish_error`, -`keepalive_miss`, `reconnect_attempt`, `reconnect_success`, `reconnect_exhausted`. A climbing `keepalive_miss` / -`reconnect_attempt` rate is the signature of a flapping upstream. They are not part of `DriverHealth` or the +`keepalive_miss`, `keepalive_recovered` (a bad keep-alive followed by a good one on the same session). A climbing +`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. ## Testing diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Observability/OtOpcUaTelemetry.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Observability/OtOpcUaTelemetry.cs index bade4e94..bb3817dc 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Observability/OtOpcUaTelemetry.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Observability/OtOpcUaTelemetry.cs @@ -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 /// flapping upstream. Tagged (driver_instance) + /// event = publish_response|notification|data_change|bad_status_value|publish_error| - /// keepalive_miss|reconnect_attempt|reconnect_success|reconnect_exhausted. + /// keepalive_miss|keepalive_recovered. /// public static readonly Counter OpcUaClientEvents = Meter.CreateCounter("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 ---------------- diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Contracts/OpcUaClientDriverOptions.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Contracts/OpcUaClientDriverOptions.cs index ae424d2d..b4bb7902 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Contracts/OpcUaClientDriverOptions.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Contracts/OpcUaClientDriverOptions.cs @@ -101,7 +101,10 @@ public sealed record OpcUaClientDriverOptions /// Client-side keep-alive interval. public TimeSpan KeepAliveInterval { get; init; } = TimeSpan.FromSeconds(5); - /// Initial reconnect delay after a session drop. + /// + /// 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. + /// public TimeSpan ReconnectPeriod { get; init; } = TimeSpan.FromSeconds(5); /// diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDiagnostics.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDiagnostics.cs index 2d379a63..16c993e8 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDiagnostics.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDiagnostics.cs @@ -22,12 +22,8 @@ internal sealed class OpcUaClientDiagnostics public const string PublishError = "publish_error"; /// A keep-alive came back Bad. public const string KeepAliveMiss = "keepalive_miss"; - /// The reconnect handler was armed. - public const string ReconnectAttempt = "reconnect_attempt"; - /// A reconnect landed a new session. - public const string ReconnectSuccess = "reconnect_success"; - /// A reconnect handler exhausted its retry window. - public const string ReconnectExhausted = "reconnect_exhausted"; + /// A keep-alive came back Good after a bad one on the same session (an in-place recovery, #670). + public const string KeepAliveRecovered = "keepalive_recovered"; private readonly string _driverInstanceId; diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriver.cs index 42739295..671ed97a 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriver.cs @@ -23,10 +23,10 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient; /// /// /// Session lifetime: a single per driver instance. -/// Subscriptions multiplex onto that session; SDK reconnect handler takes the session -/// down and brings it back up on remote-server restart — the driver must re-send -/// subscriptions + TransferSubscriptions on reconnect to avoid dangling -/// monitored-item handles. That mechanic lands in PR 69. +/// Subscriptions multiplex onto that session. A lost session is reported (health Degraded, +/// subscribed values BadNoCommunication — Gitea #670) and recovered by the host's comms-loss +/// watchdog re-initialising the driver and re-subscribing; a session whose keep-alive comes +/// back on its own recovers in place. /// /// 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); _logger = logger ?? NullLogger.Instance; _secretResolver = secretResolver ?? NullSecretResolver.Instance; + _sdkTelemetry = logger is null ? SdkTelemetry.Context : SdkTelemetry.ForLogger(logger); } /// Mutable so 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 StatusBadInternalError = 0x80020000u; private const uint StatusBadCommunicationError = 0x80050000u; + private const uint StatusBadNoCommunication = 0x80310000u; + private const uint StatusGood = 0x00000000u; private readonly SemaphoreSlim _gate = new(1, 1); @@ -150,20 +153,73 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit internal SemaphoreSlim Gate => _gate; /// - /// Test-only seam to swap the active the way - /// does at runtime, so the stale-session race the - /// read/history paths guard against (re-read Session inside _gate, not before) - /// can be exercised deterministically without standing up a live reconnect handler. - /// Production code never calls this — the SDK reconnect handler owns the real swap. + /// Test-only seam to swap the active the way a re-initialise does at + /// runtime, so the stale-session race the read/history paths guard against (re-read Session + /// inside _gate, not before) can be exercised deterministically without a live connect. + /// Production code never calls this — owns the real swap. /// /// The session instance to install, or null to clear it. internal void SetSessionForTest(ISession? session) => Session = session; + /// Test-only seam (#670): install a health snapshot as a completed connect would. + /// The health to install. + internal void SetHealthForTest(DriverHealth health) => _health = health; + + /// Test-only seam (#670): raise the session keep-alive callback as the SDK timer does. + /// The keep-alive status. + /// The raising session; defaults to the current one. + internal void RaiseKeepAliveForTest(StatusCode status, ISession? sender = null) => + OnKeepAlive(sender ?? Session!, new KeepAliveEventArgs(new ServiceResult(status), ServerState.Running, DateTime.UtcNow)); + /// - /// Test-only seam to install the that connect/reconnect builds at + /// Test-only seam (#670): register a data subscription over as + /// would, without the wire create. + /// + /// The RawPaths subscribed. + /// The subscription handle. + internal ISubscriptionHandle AddSubscriptionForTest(IReadOnlyList 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(); + 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; + } + + /// Test-only seam (#670): deliver one data notification as the SDK publish thread would. + /// The subscription handle. + /// The RawPath. + /// The delivered value. + internal void DeliverNotificationForTest(ISubscriptionHandle handle, string fullRef, DataValue value) => + PublishNotification((OpcUaSubscriptionHandle)handle, fullRef, value); + + /// Test-only seam (#670): register an alarm subscription over as + /// would, without the wire create. + /// The alarm RawPaths subscribed. + internal void AddAlarmSubscriptionForTest(IReadOnlyList 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, (_, _) => { }); + } + + /// + /// Test-only seam to install the that connect builds at /// runtime, so the v3 RawPath → upstream node id resolution can be exercised without a live /// connect (which is what populates ). Production code never calls - /// this — / own the real build. + /// this — owns the real build. /// /// The namespace map to install, or null to clear it. internal void SetNamespaceMapForTest(NamespaceMap? map) => _namespaceMap = map; @@ -191,13 +247,38 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit /// The that owns . private CertificateValidator? _certValidatorRef; /// - /// SDK-provided reconnect handler that owns the retry loop + session-transfer machinery - /// when the session's keep-alive channel reports a bad status. Null outside the - /// reconnecting window; constructed lazily inside the keep-alive handler. Guarded by - /// — keep-alive callbacks fire from the SDK timer thread and - /// can race a check-then-set if left unsynchronized. + /// Gitea #670 — 1 while the CURRENT session's keep-alive is bad (an outage is in progress), else 0. The + /// 0→1 edge publishes the outage once (Bad per subscribed ref + a quality-only alarm annotation); the 1→0 + /// edge on a good keep-alive recovers in place. Reset whenever the session is released, so a fresh session + /// reports its own loss. There is deliberately no in-driver reconnect any more: a session that never comes + /// back keeps health Degraded with frozen, which is exactly what + /// the host's #527 comms-loss watchdog trips on (→ ShutdownAsync + InitializeAsync + re-subscribe). /// - private SessionReconnectHandler? _reconnectHandler; + private int _sessionLost; + + /// #670 — the last value delivered per subscribed RawPath, so an outage can publish + /// BadNoCommunication with the value retained (the same shape the host's own comms-loss fan uses). + private readonly System.Collections.Concurrent.ConcurrentDictionary _lastValues = + new(StringComparer.Ordinal); + + /// #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. + private readonly System.Collections.Concurrent.ConcurrentDictionary _awaitingRecovery = + new(StringComparer.Ordinal); + + /// Ticks of the last advance from a notification — + /// throttles the per-notification health write to at most one per second. + private long _lastEvidenceStampTicks; + + /// Clock for liveness stamps. Test seam (#670); production uses . + internal TimeProvider Clock { get; set; } = TimeProvider.System; + + /// + /// SDK telemetry for this instance's sessions: SDK warnings (keep-alive / publish / channel errors) go to the + /// driver's own logger (#670 — the shared has no provider, which made every + /// SDK-side session error invisible). + /// + private readonly ITelemetryContext _sdkTelemetry; /// /// Bidirectional namespace map built at connect time from session.NamespaceUris. @@ -314,12 +395,11 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit "Tried:\n " + string.Join("\n ", attemptErrors), attemptErrors.Select(e => new InvalidOperationException(e))); - // Wire the session's keep-alive channel into HostState + the reconnect trigger. - // OPC UA keep-alives are authoritative for session liveness: the SDK pings on - // KeepAliveInterval and sets KeepAliveStopped when N intervals elapse without a - // response. On a bad keep-alive the driver spins up a SessionReconnectHandler - // which transparently retries + swaps the underlying session. Subscriptions move - // via TransferSubscriptions so local MonitoredItem handles stay valid. + // Wire the session's keep-alive channel into HostState + health (#670). OPC UA keep-alives + // are authoritative for session liveness: the SDK pings on KeepAliveInterval. A good one is + // liveness evidence (advances LastSuccessfulRead); a bad one degrades health and publishes the + // outage. Recovery from a session that does not come back is the host's comms-loss watchdog + // (ShutdownAsync + InitializeAsync + re-subscribe) — there is no in-driver reconnect handler. _keepAliveHandler = OnKeepAlive; session.KeepAlive += _keepAliveHandler; AttachDiagnostics(session); @@ -355,9 +435,9 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit /// private async Task BuildApplicationConfigurationAsync(CancellationToken ct) { - // The ITelemetryContext overload is the non-obsolete ctor; an empty builder keeps SDK - // internals off any logging provider (mirrors Client.Shared's ClientTelemetry). - var app = new ApplicationInstance(SdkTelemetry.Context) + // The ITelemetryContext overload is the non-obsolete ctor; SDK warnings reach the driver's + // logger (#670) when one was supplied. + var app = new ApplicationInstance(_sdkTelemetry) { ApplicationName = _options.SessionName, ApplicationType = ApplicationType.Client, @@ -681,7 +761,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit endpointConfig.OperationTimeout = (int)_options.Timeout.TotalMilliseconds; var endpoint = new ConfiguredEndpoint(null, selected, endpointConfig); - var session = await new DefaultSessionFactory(SdkTelemetry.Context).CreateAsync( + var session = await new DefaultSessionFactory(_sdkTelemetry).CreateAsync( appConfig, endpoint, false, // updateBeforeConnect @@ -795,30 +875,15 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit } /// - /// Cancels any in-flight reconnect, detaches the keep-alive handler and closes + disposes the current - /// session. The session part of ; also run at the top of - /// so a re-initialise on a live driver (#630) does not orphan the - /// previous session and its reconnect handler. Subscriptions are NOT touched here. + /// Detaches the keep-alive handler and closes + disposes the current session. The session part of + /// ; also run at the top of so a re-initialise on a + /// live driver (#630) does not orphan the previous session. Subscriptions are NOT touched here. Clears the + /// #670 outage latch, so the next session reports its own loss. /// private async Task ReleaseSessionAsync(CancellationToken cancellationToken) { - // Abort any in-flight reconnect attempts before touching the session — BeginReconnect's - // retry loop holds a reference to the current session and would fight Session.CloseAsync - // 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. + // Take the session reference under _probeLock and clear Session to null so any concurrent caller that + // checks inside _gate sees null immediately after release begins. ISession? sessionToClose; lock (_probeLock) { @@ -832,11 +897,25 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit 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 */ } try { sessionToClose?.Dispose(); } catch { } + Interlocked.Exchange(ref _sessionLost, 0); } + /// Upper bound on the best-effort wire calls a teardown makes against a possibly-dead upstream. + private static readonly TimeSpan SessionCloseTimeout = TimeSpan.FromSeconds(5); + /// 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 // handlers we attached at subscribe time so the SDK's invocation list no longer // 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) { DetachNotificationHandlers(rs.ItemHandlers); - try { await rs.Subscription.DeleteAsync(silent: true, cancellationToken).ConfigureAwait(false); } - catch { /* best-effort */ } + if (sessionAlive) await DeleteSubscriptionBestEffortAsync(rs.Subscription, cancellationToken).ConfigureAwait(false); } _subscriptions.Clear(); @@ -858,10 +939,13 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit { try { ras.EventItem.Notification -= ras.Handler; } catch { /* best-effort */ } - try { await ras.Subscription.DeleteAsync(silent: true, cancellationToken).ConfigureAwait(false); } - catch { /* best-effort */ } + if (sessionAlive) await DeleteSubscriptionBestEffortAsync(ras.Subscription, cancellationToken).ConfigureAwait(false); } _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 // 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 @@ -885,6 +969,18 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit _health = new DriverHealth(DriverState.Unknown, _health.LastSuccessfulRead, null); } + /// Best-effort, bounded delete of one subscription (shutdown path). + 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 */ } + } + /// 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 // session actually used is re-read inside the gate. _ = RequireSession(); + var (results, _) = await ReadCoreAsync(fullReferences, cancellationToken).ConfigureAwait(false); + return results; + } + + /// + /// The read itself, also used by the #670 in-place recovery. TransportOk is false when the wire read + /// did not happen or threw (every result is then a local BadCommunicationError, not upstream data). + /// + private async Task<(DataValueSnapshot[] Results, bool TransportOk)> ReadCoreAsync( + IReadOnlyList fullReferences, CancellationToken cancellationToken) + { var results = new DataValueSnapshot[fullReferences.Count]; - var now = DateTime.UtcNow; + var now = Clock.GetUtcNow().UtcDateTime; await _gate.WaitAsync(cancellationToken).ConfigureAwait(false); try { // 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's namespace table. var session = Session; @@ -914,7 +1021,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit { for (var i = 0; i < fullReferences.Count; i++) 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. @@ -934,7 +1041,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit indexMap.Add(i); } - if (toSend.Count == 0) return results; + if (toSend.Count == 0) return (results, true); try { @@ -973,10 +1080,11 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit results[r] = new DataValueSnapshot(null, StatusBadCommunicationError, null, now); } _health = new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, ex.Message); + return (results, false); } } finally { _gate.Release(); } - return results; + return (results, true); } // ---- IWritable ---- @@ -1473,7 +1581,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit try { // 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 ?? throw new InvalidOperationException( "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; var dv = mn.Value; 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); 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( Value: dv.Value, StatusCode: dv.StatusCode.Code, SourceTimestampUtc: dv.SourceTimestamp == DateTime.MinValue ? null : dv.SourceTimestamp, ServerTimestampUtc: dv.ServerTimestamp == DateTime.MinValue ? DateTime.UtcNow : dv.ServerTimestamp); + _lastValues[fullRef] = snapshot; + _awaitingRecovery.TryRemove(fullRef, out _); OnDataChange?.Invoke(this, new DataChangeEventArgs(handle, fullRef, snapshot)); } + /// + /// #670 — advance 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. + /// + 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 }; + } + /// /// Live data-change subscription bookkeeping. Holds the SDK , /// 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. // #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. - // Keys are server-stable (AlarmIndexKey) so a namespace reorder across an in-place SDK reconnect — - // which transfers this subscription without re-running SubscribeAlarmsAsync — cannot orphan them. + // Keys are server-stable (AlarmIndexKey) so a remote namespace reorder cannot orphan them. var sessionForResolve = Session; var mapForResolve = _namespaceMap; var sourceFilter = BuildAlarmSourceIndex(sourceNodeIds, raw => @@ -1709,7 +1843,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit try { // 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 ?? throw new InvalidOperationException( "OpcUaClient session was lost before the alarm subscription could be created."); @@ -1815,7 +1949,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit try { // 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 // namespace table too. var session = Session; @@ -1914,9 +2048,10 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit { if (args.NotificationValue is not EventFieldList efl) return; // #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 - // be transferred by an SDK reconnect; without this it would feed RouteAlarmEvent alongside the real one. + // deliver. A half-built one whose cleanup failed against a broken server can survive on the session; + // without this it would feed RouteAlarmEvent alongside the real one. if (!_alarmSubscriptions.ContainsKey(handle.Id)) return; + RecordLivenessEvidence(); RouteAlarmEvent(efl.EventFields, sourceFilter, handle); } @@ -2063,10 +2198,9 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit } /// - /// #619 — ConditionRefresh every alarm subscription on the current session: after an in-place reconnect - /// (transferred or SDK-recreated subscriptions — matched by display name, since a recreate clones the - /// objects) and on a RefreshRequired event. Events missed during the outage - /// reach the host through the normal deduplicated path. + /// #619 — ConditionRefresh every alarm subscription on the current session (matched by display name): after + /// a keep-alive outage the session survived (#670) and on a RefreshRequired event. Events missed during the + /// outage reach the host through the normal deduplicated path. /// /// Cancellation. /// A task completing when every refresh call has returned. @@ -2269,7 +2403,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit internal readonly record struct AlarmAckTarget(string ConditionRef, byte[] EventId); /// #619 — the key both sides of the alarm index use: server-stable (nsu=) when a map is - /// available, so a remote namespace-table reorder across an in-place reconnect cannot orphan it. + /// available, so a remote namespace-table reorder cannot orphan it. /// The current session's namespace map, or null. /// The NodeId to key. /// The index key. @@ -2367,174 +2501,159 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit }; /// - /// Session keep-alive handler. On a healthy ping, bumps HostState back to Running - /// (typical bounce after a transient network blip). On a bad ping, starts the SDK's - /// which retries on the configured period + - /// fires when it lands a new session. + /// Session keep-alive handler (Gitea #670). A good keep-alive is liveness evidence: it advances + /// , and after a bad one it ends the outage in place (the session + /// survived — e.g. a network blip): health back to Healthy and the outage's Bad refs re-read and re-published + /// 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 BadNoCommunication for + /// every subscribed ref (value retained) plus a quality-only annotation on every subscribed alarm condition. /// + /// + /// There is deliberately no SessionReconnectHandler 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. + /// 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 (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); return; } TransitionTo(HostState.Stopped); _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 - // from the SDK keep-alive timer thread and the SDK can fire this handler repeatedly - // while the channel stays down — the check-then-set must be atomic, otherwise two - // callbacks both observe null, both construct a SessionReconnectHandler, and the - // second assignment leaks the first (its retry loop keeps running, unreferenced and - // 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); + _logger.LogWarning( + "OpcUaClientDriver '{DriverInstanceId}': upstream session keep-alive failed ({Status}); subscribed values " + + "are BadNoCommunication until it recovers or the host's comms-loss watchdog reconnects the driver.", + _driverInstanceId, e.Status); + PublishCommsLoss(now); } /// - /// Called by when its retry loop has either - /// successfully swapped to a new session or given up. Reads the new session off - /// handler.Session, unwires the old keep-alive hook, rewires for the new - /// one, and tears down the handler. Subscription migration is already handled - /// inside the SDK via TransferSubscriptions (the SDK calls it automatically - /// when is true, - /// which is the default). + /// #670 — publish the start of an outage: BadNoCommunication once per (subscription, ref) with the last + /// delivered value retained, and a quality-only annotation on every subscribed alarm condition (no state + /// change — an outage says nothing about whether the alarm is active). /// - private void OnReconnectComplete(object? sender, EventArgs e) + private void PublishCommsLoss(DateTime now) { - if (sender is not SessionReconnectHandler handler) return; - 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) + foreach (var rs in _subscriptions.Values) { - oldSession = Session; - - // 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) + foreach (var pair in rs.ItemHandlers) { - try { oldSession.KeepAlive -= _keepAliveHandler; } catch { } - } - if (newSession is not null && _keepAliveHandler is not null) - { - newSession.KeepAlive += _keepAliveHandler; - } - // 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; + var fullRef = RefOf(pair.Item); + _lastValues.TryGetValue(fullRef, out var last); + _awaitingRecovery[fullRef] = 0; + RaiseDataChange(new DataChangeEventArgs(rs.Handle, fullRef, new DataValueSnapshot( + last?.Value, StatusBadNoCommunication, last?.SourceTimestampUtc, 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); + AnnotateAlarmQuality(StatusBadNoCommunication, now); } /// - /// Arm a fresh reconnect attempt after a previous handler gave up. The OPC UA Client - /// driver gateways an always-on remote server, so abandoning recovery permanently is - /// the wrong default — a new keeps retrying so - /// the driver self-heals when the remote returns, while the Faulted health set by the - /// caller still lets an operator force a clean reinitialize in the meantime. + /// #670 — the session survived the outage: re-read the refs still awaiting recovery and publish them (a ref + /// a notification already refreshed is skipped — fresh data beats the read), restore alarm-condition quality + /// and ConditionRefresh so transitions missed during the outage are re-reported (deduplicated). A failed + /// read re-arms the outage so the next good keep-alive tries again. /// - 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 { - handler.BeginReconnect( - lastSession, - (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) + var refs = _awaitingRecovery.Keys.ToList(); + if (refs.Count > 0 && Session is not null) { - if (ReferenceEquals(_reconnectHandler, handler)) + var (results, transportOk) = await ReadCoreAsync(refs, CancellationToken.None).ConfigureAwait(false); + if (!transportOk) { - _reconnectHandler.Dispose(); - _reconnectHandler = null; + Interlocked.Exchange(ref _sessionLost, 1); + 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); + } + } + + /// Quality-only annotation (#626 pattern) of every subscribed alarm condition. + 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); + } + } + } + } + + /// Raise , isolating a faulting subscriber from the SDK callback thread. + 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 } /// - /// 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 /// 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 @@ -2564,12 +2683,8 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit if (_disposed) return; _disposed = true; - // Cancel any in-flight reconnect handler. - SessionReconnectHandler? handlerToCancel; lock (_probeLock) { - handlerToCancel = _reconnectHandler; - _reconnectHandler = null; // Detach keep-alive and null Session so in-flight gated callers see null // after their next _gate.WaitAsync — they return BadCommunicationError cleanly. if (_keepAliveHandler is not null && Session is not null) @@ -2579,8 +2694,6 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit _keepAliveHandler = null; Session = null; } - try { handlerToCancel?.CancelReconnect(); } catch { } - handlerToCancel?.Dispose(); // Detach the cert-validation handler registered during InitializeAsync so the // CertificateValidator (which may be process-shared) doesn't hold a reference to diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/SdkTelemetry.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/SdkTelemetry.cs index 02d2b8f5..d8c7a200 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/SdkTelemetry.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/SdkTelemetry.cs @@ -1,16 +1,35 @@ +using Microsoft.Extensions.Logging; using Opc.Ua; namespace ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient; /// -/// The one 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 -/// ClientTelemetry). +/// The the driver hands the SDK's non-obsolete (telemetry-taking) +/// overloads. has no logging provider (static helpers and tests); +/// 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). /// internal static class SdkTelemetry { private static readonly Lazy Instance = new(() => DefaultTelemetry.Create(_ => { })); - /// Gets the shared telemetry context. + /// Gets the shared, provider-less telemetry context. public static ITelemetryContext Context => Instance.Value; + + /// Creates a telemetry context whose SDK log output at Warning and above goes to . + /// The driver instance's logger. + /// The telemetry context. + public static ITelemetryContext ForLogger(ILogger logger) => + DefaultTelemetry.Create(builder => builder + .SetMinimumLevel(LogLevel.Warning) + .AddProvider(new ForwardingLoggerProvider(logger))); + + /// Hands every SDK logger category the one driver logger. + private sealed class ForwardingLoggerProvider(ILogger target) : ILoggerProvider + { + public ILogger CreateLogger(string categoryName) => target; + + public void Dispose() { } + } } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs index b5e84b76..0d691c0e 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs @@ -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 /// comms-loss quality paths (#477 native conditions, #481 scripted-alarm inputs) hang off /// , so both were wired to a signal that never fired. - /// OpcUaClientDriver's own reconnect-exhausted branch already assumed this mechanism existed - /// ("the Core never learns it must offer an operator reinitialize") — it did not. + /// Since Gitea #670 this is also OpcUaClientDriver's only recovery path for a lost upstream + /// session (its in-driver SDK reconnect handler was deleted). /// /// /// Why the trip condition is a conjunction, and NOT simply "non-Healthy for N polls". diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientDiagnosticsTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientDiagnosticsTests.cs index acb3497b..fa296091 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientDiagnosticsTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientDiagnosticsTests.cs @@ -57,9 +57,7 @@ public sealed class OpcUaClientDiagnosticsTests d.Record(OpcUaClientDiagnostics.BadStatusValue); d.Record(OpcUaClientDiagnostics.PublishError); d.Record(OpcUaClientDiagnostics.KeepAliveMiss); - d.Record(OpcUaClientDiagnostics.ReconnectAttempt); - d.Record(OpcUaClientDiagnostics.ReconnectSuccess); - d.Record(OpcUaClientDiagnostics.ReconnectExhausted); + d.Record(OpcUaClientDiagnostics.KeepAliveRecovered); cap[OpcUaClientDiagnostics.PublishResponse].ShouldBe(1); cap[OpcUaClientDiagnostics.Notification].ShouldBe(3); @@ -67,9 +65,7 @@ public sealed class OpcUaClientDiagnosticsTests cap[OpcUaClientDiagnostics.BadStatusValue].ShouldBe(1); cap[OpcUaClientDiagnostics.PublishError].ShouldBe(1); cap[OpcUaClientDiagnostics.KeepAliveMiss].ShouldBe(1); - cap[OpcUaClientDiagnostics.ReconnectAttempt].ShouldBe(1); - cap[OpcUaClientDiagnostics.ReconnectSuccess].ShouldBe(1); - cap[OpcUaClientDiagnostics.ReconnectExhausted].ShouldBe(1); + cap[OpcUaClientDiagnostics.KeepAliveRecovered].ShouldBe(1); } [Fact] diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientReconnectTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientReconnectTests.cs index c567ca09..88114079 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientReconnectTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientReconnectTests.cs @@ -4,9 +4,10 @@ using Xunit; namespace ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests; /// -/// Scaffold tests for wiring. Wire-level -/// disconnect-reconnect-resume coverage against a live upstream server lands with the -/// in-process fixture — too much machinery for a unit-test-only lane. +/// Scaffold tests for the reconnect options. Since Gitea #670 the driver has no in-driver reconnect +/// handler (session loss is reported and the host's comms-loss watchdog re-initialises the driver); +/// ReconnectPeriod is retained for config compatibility only. Outage behaviour is covered by +/// and, over the wire, OpcUaClientDriverReconnectWireTests. /// [Trait("Category", "Unit")] public sealed class OpcUaClientReconnectTests @@ -26,13 +27,10 @@ public sealed class OpcUaClientReconnectTests opts.ReconnectPeriod.ShouldBe(TimeSpan.FromMilliseconds(500)); } - /// Verifies that the driver starts with no reconnect handler active before initialization. + /// Verifies that the driver starts in the Unknown state before initialization. [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"); drv.GetHealth().State.ShouldBe(Core.Abstractions.DriverState.Unknown); } diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientSessionLossTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientSessionLossTests.cs new file mode 100644 index 00000000..3cde5d9c --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientSessionLossTests.cs @@ -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; + +/// +/// 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 LastSuccessfulRead +/// only moved on init / ReadAsync, which a subscription-only driver never calls. +/// +[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 Tags = + [ + Tag(SpeedRef, "ns=0;s=Speed"), + Tag(CounterRef, "ns=0;s=Counter"), + ]; + + private static Mock NewSession() + { + var mock = new Mock(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 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 session, params object[] values) => + session.Setup(s => s.ReadAsync( + It.IsAny(), It.IsAny(), It.IsAny(), + It.IsAny(), It.IsAny())) + .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 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(); + 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); + } + + /// + /// 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 LastSuccessfulRead, fixing R1 alone would trip it on the first + /// bad tick of any connection older than 90 s. + /// + [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(); + 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(); + 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(); + 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(); + 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(); + drv.OnDataChange += (_, e) => seen.Add(e); + drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication); + + seen.ShouldHaveSingleItem().Snapshot.StatusCode.ShouldBe(BadNoCommunication); + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/OpcUaClientDriverReconnectWireTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/OpcUaClientDriverReconnectWireTests.cs index 1f2413da..97c22141 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/OpcUaClientDriverReconnectWireTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/OpcUaClientDriverReconnectWireTests.cs @@ -6,14 +6,19 @@ using ZB.MOM.WW.OtOpcUa.Core.Abstractions; namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests; /// -/// Reconnect + resume leg of the OpcUaClient driver's wire coverage (Gitea #622): the in-process server is -/// stopped and a fresh one is booted on the SAME port / application URI / PKI store — a process restart as the -/// driver sees it, with an empty address space the test then re-creates. The driver must detect the loss, -/// rebuild a session, and its existing subscription must deliver again with no caller action. +/// Outage + recovery leg of the OpcUaClient driver's wire coverage (Gitea #622, reworked for #670): the in-process +/// server is stopped and a fresh one is booted on the SAME port / application URI / PKI store — a process restart +/// as the driver sees it, with an empty address space the test then re-creates. The driver must REPORT the loss +/// (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 DriverInstanceActor +/// 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). /// [Trait("Category", "Integration")] public sealed class OpcUaClientDriverReconnectWireTests { + private const uint BadNoCommunication = 0x80310000u; + private static async Task WaitAsync(Func condition, string because, TimeSpan? timeout = null) { var deadline = DateTime.UtcNow + (timeout ?? TimeSpan.FromSeconds(60)); @@ -26,53 +31,66 @@ public sealed class OpcUaClientDriverReconnectWireTests } [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 rig = await OpcUaClientDriverWireTests.StartRigAsync("drv-reconnect", ct); await using var _rig = rig; OpcUaClientDriverWireTests.Publish(rig, OpcUaClientDriverWireTests.SpeedId, 10.0f); - await using var drv = OpcUaClientDriverWireTests.NewDriver( - rig, "wire-reconnect", reconnectPeriod: TimeSpan.FromSeconds(1)); + await using var drv = OpcUaClientDriverWireTests.NewDriver(rig, "wire-reconnect"); await drv.InitializeAsync("{}", ct); - var seen = new List(); + var seen = new List<(uint Status, object? Value)>(); drv.OnDataChange += (_, e) => { - if (e.FullReference == OpcUaClientDriverWireTests.SpeedRef && e.Snapshot.StatusCode == 0u) - lock (seen) seen.Add(Convert.ToSingle(e.Snapshot.Value)); + if (e.FullReference == OpcUaClientDriverWireTests.SpeedRef) + lock (seen) seen.Add((e.Snapshot.StatusCode, e.Snapshot.Value)); }; - var states = new List(); - drv.OnHostStatusChanged += (_, e) => { lock (states) states.Add(e.NewState); }; 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 WaitAsync( - () => { lock (states) return states.Any(s => s != HostState.Running); }, - "the driver must notice the server is gone", TimeSpan.FromSeconds(30)); + await WaitAsync(() => { lock (seen) return seen.Any(s => s.Status == BadNoCommunication); }, + "a lost session must publish BadNoCommunication for the subscribed ref", 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("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, - "the driver must reconnect after the server is back"); - await WaitAsync(() => { lock (seen) return seen.Contains(77.0f); }, - "the pre-outage subscription must resume and deliver the post-restart value"); + await drv.ShutdownAsync(ct); + await drv.InitializeAsync("{}", ct); + drv.GetHealth().State.ShouldBe(DriverState.Healthy); + 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); - await WaitAsync(() => { lock (seen) return seen.Contains(88.0f); }, - "a later change must also reach the resumed subscription"); + await WaitAsync(() => { lock (seen) return seen.Any(s => s is (0u, 88.0f)); }, + "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); back[0].StatusCode.ShouldBe(0u); Convert.ToSingle(back[0].Value).ShouldBe(88.0f);