From 6c3a754be99c74b778858bb637e648f9d1e35e24 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 15:19:48 -0400 Subject: [PATCH] fix(opcuaclient): lifecycle generation, close the local session on init throw, one teardown budget (R06:O-2, O-9) - Every InitializeAsync/ShutdownAsync bumps a lifecycle generation. The session, namespace map and endpoint are installed under _probeLock only while the init's generation is current, so an init the host abandoned that finishes after the next shutdown discards its session instead of resurrecting it. Every health write (lifecycle, keep-alive, publish-error, read path) is generation-stamped, so a late writer can no longer overwrite newer state. - O-2: the init catch closes and disposes the LOCAL session it opened (it closed the still-null Session property, orphaning an activated session + keep-alive timer). - O-9: ShutdownAsync spends ONE budget (4 s, inside the host's 5 s PostStop budget) on wire calls: every subscription delete concurrently, then the session close with what remains; waits are abandoned when the budget fires even if the SDK ignores the token. It used to be 5 s per subscription, serially. Refs #743 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT --- .../OpcUaClientDriver.cs | 271 +++++++++++++++--- .../OpcUaClientLifecycleGenerationTests.cs | 220 ++++++++++++++ 2 files changed, 444 insertions(+), 47 deletions(-) create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientLifecycleGenerationTests.cs 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 99cbb907..a052a286 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriver.cs @@ -449,11 +449,20 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit /// public async Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken) { + // Every lifecycle transition (initialise / shutdown) takes a new generation. Everything this attempt + // publishes — the session, the namespace map, health — is published only while its generation is still + // the current one, so an init the host abandoned (30 s bound) that finishes after the next ShutdownAsync + // or InitializeAsync cannot resurrect a session, or overwrite the newer health, behind its back. + var generation = Interlocked.Increment(ref _lifecycleGeneration); + // InitializeAsync is reachable on a live driver with no teardown first (the IDriver contract // allows it, and a reconnect retry can overlap a running attempt); `Session = session` below would otherwise orphan the previous session, // its keep-alive handler and any reconnect handler. - await ReleaseSessionAsync(cancellationToken).ConfigureAwait(false); - _health = new DriverHealth(DriverState.Initializing, null, null); + await ReleaseSessionAsync(generation, cancellationToken).ConfigureAwait(false); + WriteLifecycleHealth(generation, _ => new DriverHealth(DriverState.Initializing, null, null)); + ISession? session = null; + KeepAliveEventHandler? keepAliveHandler = null; + var installed = false; try { // re-parse the supplied DriverConfig so a config change delivered through the IDriver @@ -512,15 +521,16 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit // aggregate exception names every URL that was tried and why — critical diag for // operators debugging 'why did the failover pick #3?'. var attemptErrors = new List(candidates.Count); - ISession? session = null; string? connectedUrl = null; foreach (var url in candidates) { try { - session = await OpenSessionOnEndpointAsync( - appConfig, url, _options.SecurityPolicy, _options.SecurityMode, - identity, cancellationToken).ConfigureAwait(false); + session = OpenSessionForTest is { } open + ? await open(url, cancellationToken).ConfigureAwait(false) + : await OpenSessionOnEndpointAsync( + appConfig, url, _options.SecurityPolicy, _options.SecurityMode, + identity, cancellationToken).ConfigureAwait(false); connectedUrl = url; break; } @@ -541,8 +551,10 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit // 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; + // The handler is attached to THIS session only; OnKeepAlive ignores any sender that is not the + // installed session, so a session discarded below never reaches health. + keepAliveHandler = OnKeepAlive; + session.KeepAlive += keepAliveHandler; AttachDiagnostics(session); // Build the bidirectional namespace map from the freshly negotiated session's @@ -551,23 +563,98 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit // reorder across a restart can't silently misaddress nodes. v3: the same map also // ingests the authored RawTags so read/write/subscribe/history references (RawPaths) // resolve to their upstream node ids. - _namespaceMap = NamespaceMap.FromSession(session, _options.RawTags); - RecordSkippedTags(_namespaceMap); + var map = NamespaceMap.FromSession(session, _options.RawTags); - Session = session; - _connectedEndpointUrl = connectedUrl; - _health = new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null); - TransitionTo(HostState.Running); + // Publish the session only while this attempt is still the current lifecycle generation. The + // generation check and the install share _probeLock with ReleaseSessionAsync's take, so a + // ShutdownAsync either sees (and releases) this session or makes this attempt discard it. + lock (_probeLock) + { + if (Volatile.Read(ref _lifecycleGeneration) == generation) + { + _namespaceMap = map; + Session = session; + _keepAliveHandler = keepAliveHandler; + _connectedEndpointUrl = connectedUrl; + installed = true; + } + } + + if (!installed) + throw new OperationCanceledException( + $"OpcUaClientDriver '{_driverInstanceId}': initialise was superseded by a later lifecycle " + + "transition (shutdown or re-initialise); its session was discarded."); + + RecordSkippedTags(map); + WriteLifecycleHealth(generation, _ => new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null)); + if (Volatile.Read(ref _lifecycleGeneration) == generation) TransitionTo(HostState.Running); } catch (Exception ex) { - try { if (Session is Session s) await s.CloseAsync().ConfigureAwait(false); } catch { } - Session = null; - _health = new DriverHealth(DriverState.Faulted, null, ex.Message); + // R06:O-2 — close the LOCAL session this attempt opened (the property is still null, or belongs to a + // newer attempt), so a throw after the session activated (AttachDiagnostics, the namespace map, the + // superseded check) does not orphan it and its keep-alive timer until GC. + if (session is not null && !installed) + await DiscardSessionAsync(session, keepAliveHandler).ConfigureAwait(false); + if (Volatile.Read(ref _lifecycleGeneration) == generation) + { + if (installed) + await ReleaseSessionAsync(generation, CancellationToken.None).ConfigureAwait(false); + WriteLifecycleHealth(generation, + h => new DriverHealth(DriverState.Faulted, h.LastSuccessfulRead, ex.Message)); + } throw; } } + /// + /// Test seam: opens the session for one endpoint URL in place of the real discovery + session create, so + /// the post-connect half of (and its superseded / failure paths) can be driven + /// against a mock session. Production leaves it null. + /// + internal Func>? OpenSessionForTest { get; set; } + + /// + /// Close and dispose a session this driver never installed (a failed or superseded initialise): detach the + /// handlers attached to it, close it within , dispose it. Best-effort. + /// + /// The session to discard. + /// The keep-alive handler attached to it, if any. + private async Task DiscardSessionAsync(ISession session, KeepAliveEventHandler? keepAliveHandler) + { + if (keepAliveHandler is not null) + { + try { session.KeepAlive -= keepAliveHandler; } catch { /* best-effort */ } + } + DetachDiagnostics(session); + try + { + using var closeCts = new CancellationTokenSource(SessionCloseTimeout); + await CloseBoundedAsync(session, closeCts.Token).ConfigureAwait(false); + } + catch { /* best-effort */ } + try { session.Dispose(); } catch { /* best-effort */ } + } + + /// + /// Close a session, abandoning the wait when fires even if the SDK call itself ignores + /// the token (a half-dead upstream must not hold a teardown past its budget). + /// + private static async Task CloseBoundedAsync(ISession session, CancellationToken ct) + { + var close = session.CloseAsync((int)SessionCloseTimeout.TotalMilliseconds, true, ct) ?? Task.FromResult(StatusCodes.Good); + await close.WaitAsync(ct).ConfigureAwait(false); + } + + /// + /// Write a lifecycle health snapshot only while is still the current lifecycle + /// generation (CAS against concurrent keep-alive / data-path writers, re-checking the generation each try). + /// + /// The generation of the init / shutdown making the write. + /// Maps the current health to the one to write. + private void WriteLifecycleHealth(long generation, Func next) => + UpdateHealth(generation, next); + /// /// Build a minimal in-memory . Certificates live /// under the OS user profile — on Windows that's %LocalAppData%\OtOpcUa\pki @@ -1050,15 +1137,20 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit /// 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 does not orphan the previous session. Subscriptions are NOT touched here. Clears the - /// outage latch, so the next session reports its own loss. + /// outage latch, so the next session reports its own loss. Does nothing when is + /// no longer the current lifecycle generation — the session installed then belongs to a NEWER init, which a + /// late (abandoned) shutdown must not close. /// - private async Task ReleaseSessionAsync(CancellationToken cancellationToken) + /// The lifecycle generation of the caller. + /// Bounds the close (on top of ). + private async Task ReleaseSessionAsync(long generation, CancellationToken cancellationToken) { // 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) { + if (Volatile.Read(ref _lifecycleGeneration) != generation) return; sessionToClose = Session; if (_keepAliveHandler is not null && sessionToClose is not null) { @@ -1075,11 +1167,11 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit var sessionKnownLost = Volatile.Read(ref _sessionLost) == 1; try { - if (sessionToClose is Session s && !sessionKnownLost) + if (sessionToClose is not null && !sessionKnownLost) { using var closeCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); closeCts.CancelAfter(SessionCloseTimeout); - await s.CloseAsync(closeCts.Token).ConfigureAwait(false); + await CloseBoundedAsync(sessionToClose, closeCts.Token).ConfigureAwait(false); } } catch { /* best-effort */ } @@ -1107,9 +1199,29 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit /// Upper bound on the best-effort wire calls a teardown makes against a possibly-dead upstream. private static readonly TimeSpan SessionCloseTimeout = TimeSpan.FromSeconds(5); + /// + /// R06:O-9 — the ONE budget the whole of spends on wire calls (every subscription + /// delete, concurrently, then the session close). It used to be 5 s per subscription, serially, so a + /// half-dead upstream held the actor's PostStop for N×5 s. Under the host's 5 s PostStop budget for + /// shutdown + dispose together, so the local teardown still runs inside it. Test seam. + /// + internal TimeSpan TeardownBudget { get; set; } = TimeSpan.FromSeconds(4); + + /// Test seam: the wire delete of one subscription. Production calls the SDK's silent delete. + internal Func DeleteSubscriptionAsync { get; set; } = + static (subscription, ct) => subscription.DeleteAsync(silent: true, ct); + /// public async Task ShutdownAsync(CancellationToken cancellationToken) { + // A new lifecycle generation: an init still running (abandoned by the host) can no longer install its + // session or write health, and keep-alive / data-path health writes of the old session stop landing. + var generation = Interlocked.Increment(ref _lifecycleGeneration); + + // R06:O-9 — one budget for every wire call below, linked to the caller's token. + using var budget = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); + budget.CancelAfter(TeardownBudget); + // Tear down remote subscriptions first — otherwise Session.Close will try and may fail // with BadSubscriptionIdInvalid noise in the upstream log. _subscriptions is cleared // whether or not the wire-side delete succeeds since the local handles are useless @@ -1119,20 +1231,23 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit // 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) + var deletes = new List(); + foreach (var key in _subscriptions.Keys) { + if (!_subscriptions.TryRemove(key, out var rs)) continue; DetachNotificationHandlers(rs.ItemHandlers); - if (sessionAlive) await DeleteSubscriptionBestEffortAsync(rs.Subscription, cancellationToken).ConfigureAwait(false); + if (sessionAlive) deletes.Add(DeleteSubscriptionBestEffortAsync(rs.Subscription, budget.Token)); } - _subscriptions.Clear(); - foreach (var ras in _alarmSubscriptions.Values) + foreach (var key in _alarmSubscriptions.Keys) { + if (!_alarmSubscriptions.TryRemove(key, out var ras)) continue; try { ras.EventItem.Notification -= ras.Handler; } catch { /* best-effort */ } - if (sessionAlive) await DeleteSubscriptionBestEffortAsync(ras.Subscription, cancellationToken).ConfigureAwait(false); + if (sessionAlive) deletes.Add(DeleteSubscriptionBestEffortAsync(ras.Subscription, budget.Token)); } - _alarmSubscriptions.Clear(); + // Concurrently, under the one budget (each delete already swallows its own failure). + await Task.WhenAll(deletes).ConfigureAwait(false); // 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(); @@ -1151,9 +1266,17 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit // ConditionRefresh re-raise every active+acked alarm (an un-ack flicker). BeginAlarmSubscription forgets // them when the set changes, which is when the host did rebuild. - await ReleaseSessionAsync(cancellationToken).ConfigureAwait(false); - _namespaceMap = null; - _connectedEndpointUrl = null; + // The session close spends what is left of the same budget (none left ⇒ dispose without closing). + await ReleaseSessionAsync(generation, budget.Token).ConfigureAwait(false); + lock (_probeLock) + { + // A newer initialise may have installed its own map / endpoint meanwhile — leave those alone. + if (Volatile.Read(ref _lifecycleGeneration) == generation) + { + _namespaceMap = null; + _connectedEndpointUrl = null; + } + } // Detach the cert-validation handler so the (potentially process-shared) // CertificateValidator doesn't hold a delegate to a shutting-down driver. @@ -1164,18 +1287,19 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit _certValidatorRef = null; } - TransitionTo(HostState.Unknown); - _health = new DriverHealth(DriverState.Unknown, _health.LastSuccessfulRead, null); + if (Volatile.Read(ref _lifecycleGeneration) == generation) TransitionTo(HostState.Unknown); + WriteLifecycleHealth(generation, h => new DriverHealth(DriverState.Unknown, h.LastSuccessfulRead, null)); } - /// Best-effort, bounded delete of one subscription (shutdown path). - private static async Task DeleteSubscriptionBestEffortAsync(Subscription subscription, CancellationToken cancellationToken) + /// + /// Best-effort delete of one subscription (shutdown path), bounded by the shutdown's budget: the wait is + /// abandoned when the budget fires even if the SDK call ignores the token. + /// + private async Task DeleteSubscriptionBestEffortAsync(Subscription subscription, CancellationToken budget) { try { - using var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); - cts.CancelAfter(SessionCloseTimeout); - await subscription.DeleteAsync(silent: true, cts.Token).ConfigureAwait(false); + await DeleteSubscriptionAsync(subscription, budget).WaitAsync(budget).ConfigureAwait(false); } catch { /* best-effort */ } } @@ -1216,7 +1340,9 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit // Re-read Session inside the critical section: if a reconnect completed while we // 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. + // session's namespace table. The generation is read first: a shutdown / re-initialise that releases this + // session while the read is in flight bumps it, so the read's health outcome never lands on the next lifecycle. + var generation = Volatile.Read(ref _lifecycleGeneration); var session = Session; if (session is null || (expectedSession is not null && !ReferenceEquals(session, expectedSession))) { @@ -1267,7 +1393,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit SourceTimestampUtc: dv.SourceTimestamp == DateTime.MinValue ? null : dv.SourceTimestamp, ServerTimestampUtc: dv.ServerTimestamp == DateTime.MinValue ? now : dv.ServerTimestamp); } - _health = new DriverHealth(DriverState.Healthy, now, null); + RecordDeviceAnswered(generation, now); } catch (Exception ex) { @@ -1280,7 +1406,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit var r = indexMap[w]; results[r] = new DataValueSnapshot(null, StatusBadCommunicationError, null, now); } - _health = new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, ex.Message); + RecordCommsFault(generation, ex.Message); return (results, false); } } @@ -2062,6 +2188,29 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit UpdateHealth(h => h.State is DriverState.Healthy or DriverState.Degraded ? h with { LastSuccessfulRead = now } : null); } + /// + /// A service call got a RESPONSE from the upstream server (whatever per-node statuses it carried — a refusal + /// is still the server answering): health Healthy, advanced. Not + /// while a tracked subscription is known dead — that latch is what lets the comms-loss watchdog re-initialise + /// a driver whose session answers but whose data plane is gone. + /// + /// The lifecycle generation the call ran under. + /// When the response arrived. + private void RecordDeviceAnswered(long generation, DateTime now) + { + if (Volatile.Read(ref _subscriptionLost) == 1) return; + UpdateHealth(generation, _ => new DriverHealth(DriverState.Healthy, now, null)); + } + + /// + /// A service call got NO response (transport failure / timeout): health Degraded with + /// preserved, so the comms-loss watchdog measures the silence. + /// + /// The lifecycle generation the call ran under. + /// The failure, for . + private void RecordCommsFault(long generation, string error) => + UpdateHealth(generation, h => new DriverHealth(DriverState.Degraded, h.LastSuccessfulRead, error)); + /// /// Read-modify-write of as a CAS loop: the keep-alive timer, publish /// threads and the recovery task all derive the next health from the current one. @@ -2078,6 +2227,27 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit } } + /// + /// stamped with a lifecycle generation: the + /// write lands only while is still current, re-checked on every CAS attempt. Every + /// lifecycle transition bumps the generation BEFORE it writes its own health, so a superseded writer (a late + /// keep-alive of a released session, a read that outlived a shutdown, an abandoned init) either loses the + /// generation check or is overwritten by the transition's own later write. + /// + /// The lifecycle generation the writer belongs to. + /// Maps the current health to the next one, or null to leave it unchanged. + private void UpdateHealth(long generation, Func next) => + UpdateHealth(h => Volatile.Read(ref _lifecycleGeneration) == generation ? next(h) : null); + + /// + /// Bumped by every and . Session install, the + /// namespace-map / endpoint publish and every health write carry the generation they were started under. + /// + private long _lifecycleGeneration; + + /// Test seam: the current lifecycle generation. + internal long LifecycleGenerationForTest => Volatile.Read(ref _lifecycleGeneration); + /// /// Live data-change subscription bookkeeping. Holds the SDK , /// the local handle, and the per-MonitoredItem (item, handler) pairs so @@ -2906,6 +3076,9 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit /// private void OnKeepAlive(ISession sender, KeepAliveEventArgs e) { + // Captured BEFORE the session check: a shutdown / re-initialise that releases this session after the check + // has bumped the generation, so every health write below lands only for the session it was raised for. + var generation = Volatile.Read(ref _lifecycleGeneration); // 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; @@ -2923,7 +3096,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit if (Interlocked.Exchange(ref _sessionLost, 0) == 1) { - UpdateHealth(_ => new DriverHealth(DriverState.Healthy, now, null)); + UpdateHealth(generation, _ => new DriverHealth(DriverState.Healthy, now, null)); // A fresh grace: every subscription reports stopped right after an outage until its next message. Interlocked.Exchange(ref _publishingStoppedSinceTicks, 0); var guardUntil = ArmReplayGuard(now); @@ -2941,19 +3114,19 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit if (FindStalledSubscription(now) is { } stalled) { - DeclareSubscriptionLost($"subscription '{stalled}' stopped publishing while the session stayed up", now, + DeclareSubscriptionLost(generation, $"subscription '{stalled}' stopped publishing while the session stayed up", now, serverReported: false); return; } - UpdateHealth(h => h.State is DriverState.Healthy or DriverState.Degraded ? h with { LastSuccessfulRead = now } : null); + UpdateHealth(generation, h => h.State is DriverState.Healthy or DriverState.Degraded ? h with { LastSuccessfulRead = now } : null); return; } Interlocked.Exchange(ref _publishingStoppedSinceTicks, 0); TransitionTo(HostState.Stopped); _diag.Record(OpcUaClientDiagnostics.KeepAliveMiss); - UpdateHealth(h => new DriverHealth(DriverState.Degraded, h.LastSuccessfulRead, $"keep-alive: {e.Status}")); + UpdateHealth(generation, h => new DriverHealth(DriverState.Degraded, h.LastSuccessfulRead, $"keep-alive: {e.Status}")); // the first bad keep-alive is often ambiguous and a definitive one follows, so this is checked on EVERY // bad keep-alive — but only after the outage below has been published, so the Bad reaches subscribers first. var keepAliveCode = e.Status?.StatusCode ?? StatusCodes.Good; @@ -3058,15 +3231,16 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit /// frozen (sticky until the session is released), and the outage published once, so the host's watchdog /// re-initialises and re-subscribes. /// + /// The lifecycle generation the detection belongs to. /// Why, for the log and health. /// Detection time. /// True when the SERVER said the subscription is gone (BadNoSubscription / /// BadSubscriptionIdInvalid) — definitive, so is raised. False for the publishing-stall /// heuristic, an inference from silence that stays with the comms-loss watchdog. - private void DeclareSubscriptionLost(string reason, DateTime now, bool serverReported) + private void DeclareSubscriptionLost(long generation, string reason, DateTime now, bool serverReported) { if (Interlocked.Exchange(ref _subscriptionLost, 1) == 1) return; - UpdateHealth(h => new DriverHealth(DriverState.Degraded, h.LastSuccessfulRead, $"subscription lost: {reason}")); + UpdateHealth(generation, h => new DriverHealth(DriverState.Degraded, h.LastSuccessfulRead, $"subscription lost: {reason}")); _logger.LogWarning( "OpcUaClientDriver '{DriverInstanceId}': {Reason}; subscribed values are BadNoCommunication until the " + "host re-initialises the driver and re-subscribes.", _driverInstanceId, reason); @@ -3087,6 +3261,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit private void HandlePublishError(object? sender, StatusCode code, uint subscriptionId) { _diag.Record(OpcUaClientDiagnostics.PublishError); + var generation = Volatile.Read(ref _lifecycleGeneration); if (sender is not null && !ReferenceEquals(sender, Session)) return; bool tracked; if (code == StatusCodes.BadNoSubscription) @@ -3097,7 +3272,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit else return; if (tracked) - DeclareSubscriptionLost($"server reported {code} for a tracked subscription", Clock.GetUtcNow().UtcDateTime, + DeclareSubscriptionLost(generation, $"server reported {code} for a tracked subscription", Clock.GetUtcNow().UtcDateTime, serverReported: true); } @@ -3349,6 +3524,8 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit { if (_disposed) return; _disposed = true; + // Supersede any init still running, so it discards rather than installs its session. + Interlocked.Increment(ref _lifecycleGeneration); lock (_probeLock) { diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientLifecycleGenerationTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientLifecycleGenerationTests.cs new file mode 100644 index 00000000..c7641b2f --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientLifecycleGenerationTests.cs @@ -0,0 +1,220 @@ +using System.Diagnostics; +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; + +/// +/// Review 2026-10-21 R06:O-2 / O-9 and the lifecycle-generation contract: an initialise that throws after its +/// session activated closes that LOCAL session; an initialise the host abandoned that finishes after the next +/// shutdown discards its session instead of installing it; a read that outlives a shutdown does not overwrite the +/// shutdown's health; and the whole shutdown spends one bounded budget on wire calls, issued concurrently. +/// +[Trait("Category", "Unit")] +public sealed class OpcUaClientLifecycleGenerationTests +{ + private const string SpeedRef = "Plant/dev/Speed"; + + private static readonly IReadOnlyList Tags = + [new(SpeedRef, JsonSerializer.Serialize(new { nodeId = "ns=0;s=Speed" }), WriteIdempotent: true)]; + + private static OpcUaClientDriverOptions Options() => + new() { EndpointUrl = "opc.tcp://127.0.0.1:1", RawTags = Tags }; + + 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 ReadResponse GoodRead(ReadValueIdCollection nodes) => new() + { + ResponseHeader = new ResponseHeader(), + Results = [.. nodes.Select(_ => new DataValue(new Variant(1), 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); + } + + /// O-2: a throw after the session opened (here: building the namespace map) closes + disposes it. + [Fact] + public async Task An_init_that_throws_after_the_session_opened_closes_and_disposes_that_session() + { + using var drv = new OpcUaClientDriver(Options(), "opcua-o2"); + var session = NewSession(); + session.SetupGet(s => s.NamespaceUris).Throws(new InvalidOperationException("namespace table unavailable")); + drv.OpenSessionForTest = (_, _) => Task.FromResult(session.Object); + + await Should.ThrowAsync(() => + drv.InitializeAsync("{}", TestContext.Current.CancellationToken)); + + session.Verify(s => s.CloseAsync(It.IsAny(), true, It.IsAny()), Times.Once); + session.Verify(s => s.Dispose(), Times.Once, "the activated session must not be orphaned until GC"); + drv.Session.ShouldBeNull(); + drv.GetHealth().State.ShouldBe(DriverState.Faulted); + } + + /// Positive control for the init seam: an uninterrupted init installs the session and goes Healthy. + [Fact] + public async Task Positive_control_an_uninterrupted_init_installs_its_session() + { + using var drv = new OpcUaClientDriver(Options(), "opcua-gen-ok"); + var session = NewSession(); + drv.OpenSessionForTest = (_, _) => Task.FromResult(session.Object); + + await drv.InitializeAsync("{}", TestContext.Current.CancellationToken); + + drv.Session.ShouldBeSameAs(session.Object); + drv.GetHealth().State.ShouldBe(DriverState.Healthy); + session.Verify(s => s.Dispose(), Times.Never); + } + + /// + /// An init abandoned by the host whose connect completes AFTER the next ShutdownAsync must not install its + /// session (it would be live with nothing to ever close it) nor overwrite the shutdown's health. + /// + [Fact] + public async Task An_init_superseded_by_a_shutdown_discards_its_session_and_leaves_the_shutdown_health() + { + var drv = new OpcUaClientDriver(Options(), "opcua-gen-superseded"); + using var _ = drv; + var session = NewSession(); + var opened = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + drv.OpenSessionForTest = async (_, _) => + { + opened.TrySetResult(); + await release.Task; + return session.Object; + }; + + var init = drv.InitializeAsync("{}", TestContext.Current.CancellationToken); + await opened.Task.WaitAsync(TimeSpan.FromSeconds(30), TestContext.Current.CancellationToken); + + await drv.ShutdownAsync(TestContext.Current.CancellationToken); + drv.GetHealth().State.ShouldBe(DriverState.Unknown); + + release.SetResult(); + await Should.ThrowAsync(() => init); + + drv.Session.ShouldBeNull("a superseded init must not resurrect a session behind the shutdown"); + session.Verify(s => s.Dispose(), Times.Once); + drv.GetHealth().State.ShouldBe(DriverState.Unknown, "the abandoned init must not overwrite the newer health"); + } + + /// A read whose response arrives after a shutdown does not write Healthy over the shutdown. + [Fact] + public async Task A_read_that_outlives_a_shutdown_does_not_overwrite_the_shutdown_health() + { + var (drv, session, respond) = ConnectedDriverWithParkedRead(); + using var _ = drv; + + var read = drv.ReadAsync([SpeedRef], TestContext.Current.CancellationToken); + await WaitForAsync(() => respond.Count == 1, "the read must reach the wire"); + + await drv.ShutdownAsync(TestContext.Current.CancellationToken); + respond.Single().SetResult(); + await read; + + drv.GetHealth().State.ShouldBe(DriverState.Unknown); + } + + /// Positive control: the same parked read with no shutdown in between writes Healthy + a new stamp. + [Fact] + public async Task Positive_control_a_read_with_no_intervening_shutdown_writes_healthy() + { + var (drv, _, respond) = ConnectedDriverWithParkedRead(); + using var __ = drv; + drv.SetHealthForTest(new DriverHealth(DriverState.Degraded, new DateTime(2020, 1, 1, 0, 0, 0, DateTimeKind.Utc), "x")); + + var read = drv.ReadAsync([SpeedRef], TestContext.Current.CancellationToken); + await WaitForAsync(() => respond.Count == 1, "the read must reach the wire"); + respond.Single().SetResult(); + await read; + + drv.GetHealth().State.ShouldBe(DriverState.Healthy); + drv.GetHealth().LastSuccessfulRead.ShouldNotBeNull().Year.ShouldBeGreaterThan(2020); + } + + private static (OpcUaClientDriver Driver, Mock Session, System.Collections.Concurrent.ConcurrentQueue Respond) + ConnectedDriverWithParkedRead() + { + var drv = new OpcUaClientDriver(Options(), "opcua-gen-read"); + var session = NewSession(); + var respond = new System.Collections.Concurrent.ConcurrentQueue(); + session.Setup(s => s.ReadAsync( + It.IsAny(), It.IsAny(), It.IsAny(), + It.IsAny(), It.IsAny())) + .Returns(async (RequestHeader _, double _, TimestampsToReturn _, ReadValueIdCollection nodes, CancellationToken _) => + { + var gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + respond.Enqueue(gate); + await gate.Task; + return GoodRead(nodes); + }); + drv.SetSessionForTest(session.Object); + drv.SetNamespaceMapForTest(NamespaceMap.FromTable(new NamespaceTable(), Tags)); + drv.SetHealthForTest(new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null)); + return (drv, session, respond); + } + + /// + /// O-9: against a half-dead upstream every subscription delete and the session close hang. They are issued + /// concurrently under ONE budget, so the shutdown ends near the budget rather than N × 5 s. + /// + [Fact] + public async Task Shutdown_spends_one_budget_on_concurrent_subscription_deletes_and_the_close() + { + using var drv = new OpcUaClientDriver(Options(), "opcua-o9"); + var session = NewSession(); + // A close that ignores its token entirely — only the driver's own abandonment can bound it. + session.Setup(s => s.CloseAsync(It.IsAny(), It.IsAny(), It.IsAny())) + .Returns(new TaskCompletionSource().Task); + drv.SetSessionForTest(session.Object); + drv.TeardownBudget = TimeSpan.FromSeconds(1); + + var inFlight = 0; + var maxInFlight = 0; + drv.DeleteSubscriptionAsync = async (_, ct) => + { + var now = Interlocked.Increment(ref inFlight); + InterlockedMax(ref maxInFlight, now); + try { await Task.Delay(Timeout.Infinite, ct); } + finally { Interlocked.Decrement(ref inFlight); } + }; + for (var i = 0; i < 4; i++) drv.AddSubscriptionForTest([SpeedRef]); + + var sw = Stopwatch.StartNew(); + await drv.ShutdownAsync(TestContext.Current.CancellationToken); + sw.Stop(); + + maxInFlight.ShouldBe(4, "the four deletes must be in flight together, not one after another"); + sw.Elapsed.ShouldBeLessThan(TimeSpan.FromSeconds(3.5), + "one 1 s budget bounds the whole teardown; serial deletes alone would take 4 × the budget"); + session.Verify(s => s.Dispose(), Times.Once, "the session is still disposed when the close was abandoned"); + drv.GetHealth().State.ShouldBe(DriverState.Unknown); + } + + private static void InterlockedMax(ref int target, int value) + { + int current; + while ((current = Volatile.Read(ref target)) < value) + if (Interlocked.CompareExchange(ref target, value, current) == current) return; + } +}