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; + } +}