From 4d0d5d229dd9def831e329e99daf382467dce956 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 15:12:03 -0400 Subject: [PATCH 01/27] fix(opcuaclient): remove a half-built data subscription when its create throws (R06:O-1) SubscribeAsync left the SDK Subscription on the session (server-side when CreateAsync succeeded, item handlers attached, tracked by nothing) when CreateAsync/CreateItemsAsync threw, so the host's per-tick reconcile retry accumulated one live subscription per retry. Mirror the alarm path: detach the item handlers, bounded RemoveSubscriptionAsync, rethrow. Refs #743 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT --- .../OpcUaClientDriver.cs | 34 ++++++-- .../OpcUaClientSubscribeCleanupTests.cs | 85 +++++++++++++++++++ 2 files changed, 113 insertions(+), 6 deletions(-) create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientSubscribeCleanupTests.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 4e2642ca..99cbb907 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriver.cs @@ -1787,13 +1787,39 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit ?? throw new InvalidOperationException( "OpcUaClient session was lost before the subscription could be created."); session.AddSubscription(subscription); - await subscription.CreateAsync(cancellationToken).ConfigureAwait(false); - // Track each (MonitoredItem, handler) pair so UnsubscribeAsync / ShutdownAsync // can detach the Notification delegate before disposing the session. The lambda // captures `handle`, so we must hold the exact delegate instance returned by // `+=` to be able to remove it. var itemHandlers = new List(); + try + { + await CreateDataItemsAsync(session, itemHandlers).ConfigureAwait(false); + } + catch + { + // R06:O-1 — the host's reconcile retries a failed subscribe every health tick, so a half-built + // subscription left on the session (created server-side when CreateAsync succeeded, its item + // handlers still attached, tracked by nothing) would accumulate one live, publishing subscription + // per retry. Mirror the alarm path: detach, bounded remove, rethrow. + DetachNotificationHandlers(itemHandlers); + try + { + using var removeCts = new CancellationTokenSource(SessionCloseTimeout); + await session.RemoveSubscriptionAsync(subscription, removeCts.Token).ConfigureAwait(false); + } + catch { /* best-effort — the session may be the thing that failed */ } + throw; + } + _subscriptions[id] = new RemoteSubscription(subscription, handle, itemHandlers); + } + finally { _gate.Release(); } + + return handle; + + async Task CreateDataItemsAsync(ISession session, List itemHandlers) + { + await subscription.CreateAsync(cancellationToken).ConfigureAwait(false); foreach (var fullRef in fullReferences) { if (!TryResolveTagNode(session, fullRef, out var nodeId)) continue; @@ -1821,11 +1847,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit } await subscription.CreateItemsAsync(cancellationToken).ConfigureAwait(false); - _subscriptions[id] = new RemoteSubscription(subscription, handle, itemHandlers); } - finally { _gate.Release(); } - - return handle; } /// diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientSubscribeCleanupTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientSubscribeCleanupTests.cs new file mode 100644 index 00000000..b3ee7ce6 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientSubscribeCleanupTests.cs @@ -0,0 +1,85 @@ +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-1 — a data SubscribeAsync whose wire create throws used to leave the +/// half-built on the session (untracked, so neither Unsubscribe nor Shutdown +/// ever deleted it). The host's reconcile retries every health tick, so a server that refused the create +/// accumulated one live subscription per retry. The alarm path already cleaned up; the data path now does too. +/// +/// +/// A Moq session never attaches the subscription (AddSubscription is a no-op), so the SDK's +/// Subscription.CreateAsync throws — the exact "create failed after AddSubscription" shape. +/// +[Trait("Category", "Unit")] +public sealed class OpcUaClientSubscribeCleanupTests +{ + 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 (OpcUaClientDriver Driver, Mock Session) NewConnectedDriver() + { + var drv = new OpcUaClientDriver(new OpcUaClientDriverOptions { RawTags = Tags }, "opcua-o1"); + var session = new Mock(MockBehavior.Loose); + session.SetupGet(s => s.MessageContext).Returns(new ServiceMessageContext(SdkTelemetry.Context)); + session.SetupGet(s => s.NamespaceUris).Returns(new NamespaceTable()); + drv.SetSessionForTest(session.Object); + drv.SetNamespaceMapForTest(NamespaceMap.FromTable(new NamespaceTable(), Tags)); + return (drv, session); + } + + /// A failed data subscribe removes the subscription it added to the session, then rethrows. + [Fact] + public async Task A_failed_data_subscribe_removes_the_half_built_subscription_from_the_session() + { + var (drv, session) = NewConnectedDriver(); + using var _ = drv; + + await Should.ThrowAsync(() => + drv.SubscribeAsync([SpeedRef], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken)); + + session.Verify(s => s.AddSubscription(It.IsAny()), Times.Once); + session.Verify(s => s.RemoveSubscriptionAsync(It.IsAny(), It.IsAny()), Times.Once, + "the subscription added before the failing create must not stay on the session"); + } + + /// + /// Positive control for the observation above: the alarm path (which always cleaned up) is seen removing + /// its subscription through the same mock, so a "Times.Once" pass on the data path is not vacuous. + /// + [Fact] + public async Task Positive_control_the_alarm_path_is_observed_removing_its_half_built_subscription() + { + var (drv, session) = NewConnectedDriver(); + using var _ = drv; + + await Should.ThrowAsync(() => + drv.SubscribeAlarmsAsync([SpeedRef], TestContext.Current.CancellationToken)); + + session.Verify(s => s.RemoveSubscriptionAsync(It.IsAny(), It.IsAny()), Times.Once); + } + + /// Repeated failing subscribes (the host's reconcile retry) each clean up their own subscription. + [Fact] + public async Task Each_retry_of_a_failing_subscribe_cleans_up_after_itself() + { + var (drv, session) = NewConnectedDriver(); + using var _ = drv; + + for (var i = 0; i < 3; i++) + await Should.ThrowAsync(() => + drv.SubscribeAsync([SpeedRef], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken)); + + session.Verify(s => s.AddSubscription(It.IsAny()), Times.Exactly(3)); + session.Verify(s => s.RemoveSubscriptionAsync(It.IsAny(), It.IsAny()), Times.Exactly(3)); + } +} From 6c3a754be99c74b778858bb637e648f9d1e35e24 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 15:19:48 -0400 Subject: [PATCH 02/27] 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; + } +} From afafd483ddd10544ca06922e6c0de43b98d22f57 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 15:21:19 -0400 Subject: [PATCH 03/27] fix(opcuaclient): an acknowledge with no reply fails with its transport status (R06:O-8) AcknowledgeAsync swallowed every non-cancellation transport failure (and returned normally when the session was lost while the ack queued), so the host logged the Part 9 ack as applied while upstream still showed it unacknowledged. Both now throw (ServiceResultException with the transport status / BadNotConnected). A reply, including per-ack refusals, still returns normally and counts as the server answering. Refs #743 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT --- docs/drivers/OpcUaClient.md | 6 ++ .../OpcUaClientDriver.cs | 17 ++-- .../OpcUaClientAcknowledgeFailureTests.cs | 83 +++++++++++++++++++ 3 files changed, 101 insertions(+), 5 deletions(-) create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientAcknowledgeFailureTests.cs diff --git a/docs/drivers/OpcUaClient.md b/docs/drivers/OpcUaClient.md index 13cd6db8..d2d7b523 100644 --- a/docs/drivers/OpcUaClient.md +++ b/docs/drivers/OpcUaClient.md @@ -231,6 +231,12 @@ initialise (the AdminUI form drops them on the next save). Pinned by > 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`. +> +> **An ack that never got a reply fails (review 2026-10-21 O-8).** A transport failure on the `Acknowledge` call +> (closed channel, faulted session) — or a session lost while the ack queued — now throws with its status +> (`ServiceResultException`, e.g. `BadConnectionClosed` / `BadNotConnected`), so the host logs the ack as failed +> instead of applied. A *reply* still returns normally, including per-ack refusals such as +> `BadConditionBranchAlreadyAcked` (logged per ack): a refusal is the server answering. > This driver does **not** implement `IRediscoverable` — there is no > push-driven rediscovery signal from a remote OPC UA server in this driver. 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 a052a286..9a79ac3c 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriver.cs @@ -2496,9 +2496,12 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit // Re-read Session inside the critical section: a reconnect completing while we were // 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; - if (session is null) return; + // namespace table too. A session lost while this ack queued is a failure, not a silent success: the + // host would otherwise log the ack as applied (R06:O-8). + var generation = Volatile.Read(ref _lifecycleGeneration); + var session = Session + ?? throw new ServiceResultException(StatusCodes.BadNotConnected, + "OpcUaClient session was lost before the acknowledgement could be sent."); // OPC UA A&C: call the AcknowledgeableConditionType.Acknowledge method on each // condition node with EventId + Comment arguments. CallAsync accepts a batch — @@ -2543,6 +2546,8 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit requestHeader: null, methodsToCall: callRequests, ct: cancellationToken).ConfigureAwait(false); + // A response — whatever the per-ack statuses say — is the server answering. + RecordDeviceAnswered(generation, Clock.GetUtcNow().UtcDateTime); // Inspect per-ack results — the upstream server can reject individual acks // (BadConditionAlreadyAcked, BadNodeIdUnknown, BadUserAccessDenied) even when @@ -2575,11 +2580,13 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit } catch (Exception ex) { - // Log genuine transport failures rather than swallowing them silently. + // R06:O-8 — a transport failure means NO reply: the acknowledgement was not applied upstream, so the + // caller must see it (with the transport status, e.g. BadConnectionClosed) instead of a success. _logger.LogWarning(ex, "OpcUaClientDriver '{DriverInstanceId}': AcknowledgeAsync failed; " + - "acknowledgements may not have been applied.", + "acknowledgements were not applied.", _driverInstanceId); + throw; } } finally { _gate.Release(); } diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientAcknowledgeFailureTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientAcknowledgeFailureTests.cs new file mode 100644 index 00000000..63e9b62a --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientAcknowledgeFailureTests.cs @@ -0,0 +1,83 @@ +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-8 — AcknowledgeAsync swallowed every non-cancellation transport failure, so +/// the host logged the Part 9 ack as applied while upstream still showed the condition unacknowledged. A +/// transport failure (no reply) now propagates with its status; a reply — even one carrying per-ack refusals — +/// still returns normally (that is the server answering, logged per ack). +/// +[Trait("Category", "Unit")] +public sealed class OpcUaClientAcknowledgeFailureTests +{ + private const string Condition = "ns=2;s=Pump17"; + + 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 AlarmAcknowledgeRequest[] Ack() => [new(Condition, Condition, "operator ack")]; + + /// A transport failure on the Acknowledge call reaches the caller with the transport status. + [Fact] + public async Task A_transport_failure_propagates_with_its_status() + { + using var drv = new OpcUaClientDriver(new OpcUaClientDriverOptions(), "opcua-o8"); + var session = NewSession(); + session.Setup(s => s.CallAsync(It.IsAny(), It.IsAny(), It.IsAny())) + .ThrowsAsync(new ServiceResultException(StatusCodes.BadConnectionClosed)); + drv.SetSessionForTest(session.Object); + + var ex = await Should.ThrowAsync(() => + drv.AcknowledgeAsync(Ack(), TestContext.Current.CancellationToken)); + + ex.StatusCode.ShouldBe(StatusCodes.BadConnectionClosed); + } + + /// Positive control: a reply returns normally and the call reached the wire. + [Fact] + public async Task Positive_control_a_reply_returns_normally() + { + using var drv = new OpcUaClientDriver(new OpcUaClientDriverOptions(), "opcua-o8-ok"); + var session = NewSession(); + session.Setup(s => s.CallAsync(It.IsAny(), It.IsAny(), It.IsAny())) + .ReturnsAsync(new CallResponse + { + ResponseHeader = new ResponseHeader(), + Results = [new CallMethodResult { StatusCode = StatusCodes.BadConditionBranchAlreadyAcked }], + DiagnosticInfos = [], + }); + drv.SetSessionForTest(session.Object); + + await drv.AcknowledgeAsync(Ack(), TestContext.Current.CancellationToken); + + session.Verify(s => s.CallAsync(It.IsAny(), It.IsAny(), It.IsAny()), + Times.Once); + } + + /// A session lost while the ack queued on the gate fails the ack rather than reporting it applied. + [Fact] + public async Task A_session_lost_while_queued_fails_the_ack() + { + var ct = TestContext.Current.CancellationToken; + using var drv = new OpcUaClientDriver(new OpcUaClientDriverOptions(), "opcua-o8-lost"); + drv.SetSessionForTest(NewSession().Object); + + await drv.Gate.WaitAsync(ct); + var call = drv.AcknowledgeAsync(Ack(), ct); + drv.SetSessionForTest(null); + drv.Gate.Release(); + + await Should.ThrowAsync(() => call); + } +} From 66d100a9e94137954cb385dfb108df4d8735129a Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 15:24:40 -0400 Subject: [PATCH 04/27] perf(opcuaclient): split the read gate from subscription management; no gate on writes or the discovery browse (R06:O-6) One SemaphoreSlim serialised reads, writes, subscribes, acks and the whole recursive discovery browse, so a stalled write (up to the operation timeout) blocked every read and subscription create, and a discovery browse would block the data plane for its full duration. Reads/acks keep the read gate; writes and DiscoverAsync are ungated (the SDK session is safe for concurrent service calls, the host bounds write concurrency); subscription management takes its own gate. Dispose drains both (DisposeAsync's drain no longer over-releases on a timed-out wait). Refs #743 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT --- docs/drivers/OpcUaClient.md | 15 ++ .../OpcUaClientDriver.cs | 243 +++++++++--------- .../OpcUaClientGateSplitTests.cs | 153 +++++++++++ 3 files changed, 293 insertions(+), 118 deletions(-) create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientGateSplitTests.cs diff --git a/docs/drivers/OpcUaClient.md b/docs/drivers/OpcUaClient.md index d2d7b523..aa9d4e3f 100644 --- a/docs/drivers/OpcUaClient.md +++ b/docs/drivers/OpcUaClient.md @@ -70,6 +70,21 @@ A single `Session` per driver instance; subscriptions multiplex onto it. Stored session-relative `ns=N` index) so a remote namespace-table reorder across a restart doesn't silently re-point references at the wrong namespace. +**Concurrency (review 2026-10-21 O-6).** Reads (including the post-outage recovery / reconcile reads) and acks +share one gate; **writes and the discovery browse take none** (the SDK session is safe for concurrent service +calls, and the host already bounds write concurrency and serialises per tag); subscription management — data and +alarm subscribe/unsubscribe and ConditionRefresh sweeps — has its own gate. One gate used to serialise all of it, +so a write stalled for the operation timeout blocked every read and subscription create, and a discovery browse +would have blocked the data plane for its whole duration. + +**Lifecycle generation.** Every `InitializeAsync` / `ShutdownAsync` takes a new generation; the session, namespace +map and every health write land only while their generation is current. An init the host abandoned that finishes +after the next shutdown discards (closes + disposes) its session instead of installing it, and a read whose +response arrives after a shutdown does not write `Healthy` over it. A throw after the session activated closes +that local session (O-2). `ShutdownAsync` spends **one** 4 s budget on wire calls — every subscription delete +concurrently, then the close with what is left — and abandons waits the SDK does not cancel (O-9; it used to be +5 s per subscription, serially, inside the actor's `PostStop`). + ### Upstream outage and recovery (Gitea #670) The session keep-alive (`KeepAliveInterval`, default 5 s) is the liveness signal: 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 9a79ac3c..4e757d6d 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriver.cs @@ -158,9 +158,20 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit /// Active OPC UA session. Null until returns cleanly. internal ISession? Session { get; private set; } - /// Per-connection gate. PRs 67+ serialize read/write/browse on this. + /// + /// R06:O-6 — subscription management (data + alarm subscribe/unsubscribe, ConditionRefresh sweeps) has its own + /// gate, so a slow subscription create no longer queues reads, and the reconcile reads never wait on it. + /// + private readonly SemaphoreSlim _subscriptionGate = new(1, 1); + + /// Read gate: serialises reads (incl. the recovery / reconcile reads) and acknowledges. Writes and the + /// discovery browse take no gate (the SDK session is safe for concurrent service calls); subscription management + /// takes . internal SemaphoreSlim Gate => _gate; + /// Test seam: the subscription-management gate. + internal SemaphoreSlim SubscriptionGate => _subscriptionGate; + /// /// 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 @@ -1420,80 +1431,76 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit public async Task> WriteAsync( IReadOnlyList writes, CancellationToken cancellationToken) { - // See ReadAsync — the wire call must use the session current inside the gate, not a - // reference captured before WaitAsync. + // R06:O-6 — writes take NO gate: a write stalled for the operation timeout must not hold up every read + // and subscription create behind it (the host already bounds write concurrency and serialises per tag; + // the SDK session is safe for concurrent service calls). The session is read once, here. _ = RequireSession(); var results = new WriteResult[writes.Count]; - await _gate.WaitAsync(cancellationToken).ConfigureAwait(false); + var session = Session; + if (session is null) + { + // Writes are non-idempotent — but here the request never reached the wire, so + // BadCommunicationError ("definitely did not happen") is the honest code. + for (var i = 0; i < writes.Count; i++) + results[i] = new WriteResult(StatusBadCommunicationError); + return results; + } + + var toSend = new WriteValueCollection(); + var indexMap = new List(writes.Count); + for (var i = 0; i < writes.Count; i++) + { + if (!TryResolveTagNode(session, writes[i].FullReference, out var nodeId)) + { + results[i] = new WriteResult(StatusBadNodeIdInvalid); + continue; + } + toSend.Add(new WriteValue + { + NodeId = nodeId, + AttributeId = Attributes.Value, + Value = new DataValue(new Variant(writes[i].Value)), + }); + indexMap.Add(i); + } + + if (toSend.Count == 0) return results; + try { - var session = Session; - if (session is null) - { - // Writes are non-idempotent — but here the request never reached the wire, so - // BadCommunicationError ("definitely did not happen") is the honest code. - for (var i = 0; i < writes.Count; i++) - results[i] = new WriteResult(StatusBadCommunicationError); - return results; - } + var resp = await session.WriteAsync( + requestHeader: null, + nodesToWrite: toSend, + ct: cancellationToken).ConfigureAwait(false); - var toSend = new WriteValueCollection(); - var indexMap = new List(writes.Count); - for (var i = 0; i < writes.Count; i++) + var codes = resp.Results; + for (var w = 0; w < codes.Count; w++) { - if (!TryResolveTagNode(session, writes[i].FullReference, out var nodeId)) - { - results[i] = new WriteResult(StatusBadNodeIdInvalid); - continue; - } - toSend.Add(new WriteValue - { - NodeId = nodeId, - AttributeId = Attributes.Value, - Value = new DataValue(new Variant(writes[i].Value)), - }); - indexMap.Add(i); - } - - if (toSend.Count == 0) return results; - - try - { - var resp = await session.WriteAsync( - requestHeader: null, - nodesToWrite: toSend, - ct: cancellationToken).ConfigureAwait(false); - - var codes = resp.Results; - for (var w = 0; w < codes.Count; w++) - { - var r = indexMap[w]; - // Pass upstream WriteResult StatusCode through verbatim. Success codes - // include Good (0) and any warning-level Good* variants; anything with - // the severity bits set is a Bad. - results[r] = new WriteResult(codes[w].Code); - } - } - catch (OperationCanceledException) - { - // Timeout / cancellation after the wire request may have been dispatched. - // Writes are non-idempotent — BadTimeout ("outcome unknown, do not blindly - // retry") is more honest than BadCommunicationError ("definitely did not - // happen"). Downstream callers that need retry semantics check for BadTimeout - // and can decide whether to re-issue. - const uint StatusBadTimeout = 0x800A0000u; - for (var w = 0; w < indexMap.Count; w++) - results[indexMap[w]] = new WriteResult(StatusBadTimeout); - } - catch (Exception) - { - // Pre-wire transport failure — the write definitely did not reach the server. - for (var w = 0; w < indexMap.Count; w++) - results[indexMap[w]] = new WriteResult(StatusBadCommunicationError); + var r = indexMap[w]; + // Pass upstream WriteResult StatusCode through verbatim. Success codes + // include Good (0) and any warning-level Good* variants; anything with + // the severity bits set is a Bad. + results[r] = new WriteResult(codes[w].Code); } } - finally { _gate.Release(); } + catch (OperationCanceledException) + { + // Timeout / cancellation after the wire request may have been dispatched. + // Writes are non-idempotent — BadTimeout ("outcome unknown, do not blindly + // retry") is more honest than BadCommunicationError ("definitely did not + // happen"). Downstream callers that need retry semantics check for BadTimeout + // and can decide whether to re-issue. + const uint StatusBadTimeout = 0x800A0000u; + for (var w = 0; w < indexMap.Count; w++) + results[indexMap[w]] = new WriteResult(StatusBadTimeout); + } + catch (Exception) + { + // Pre-wire transport failure — the write definitely did not reach the server. + for (var w = 0; w < indexMap.Count; w++) + results[indexMap[w]] = new WriteResult(StatusBadCommunicationError); + } return results; } @@ -1550,46 +1557,38 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit public async Task DiscoverAsync(IAddressSpaceBuilder builder, CancellationToken cancellationToken) { ArgumentNullException.ThrowIfNull(builder); - // Confirm a session exists before queuing; the session actually browsed is re-read - // inside the gate so a reconnect mid-wait can't leave us browsing a closed session. - _ = RequireSession(); + // R06:O-6 — no gate across the recursive browse: it can run for as long as the remote tree takes to walk, + // and holding the read gate for that long would stall the data plane behind it. The browse reads the session + // once and uses it throughout (the SDK session is safe for concurrent service calls). + var session = Session + ?? throw new InvalidOperationException("OpcUaClientDriver not initialized"); var rootFolder = builder.Folder("Remote", "Remote"); var visited = new HashSet(); var discovered = 0; var pendingVariables = new List(); - await _gate.WaitAsync(cancellationToken).ConfigureAwait(false); - try - { - var session = Session - ?? throw new InvalidOperationException( - "OpcUaClient session was lost before discovery could browse the remote server."); + var root = !string.IsNullOrEmpty(_options.BrowseRoot) + ? NodeId.Parse(session.MessageContext, _options.BrowseRoot) + : ObjectIds.ObjectsFolder; - var root = !string.IsNullOrEmpty(_options.BrowseRoot) - ? NodeId.Parse(session.MessageContext, _options.BrowseRoot) - : ObjectIds.ObjectsFolder; + // Pass 1: browse hierarchy + create folders inline, collect variables into a + // pending list. Defers variable registration until attributes are resolved — the + // address-space builder's Variable call is the one-way commit, so doing it only + // once per variable (with correct DataType/SecurityClass/IsArray) avoids the + // alternative (register with placeholders + mutate later) which the + // IAddressSpaceBuilder contract doesn't expose. + await BrowseRecursiveAsync(session, root, rootFolder, visited, + depth: 0, + discovered: () => discovered, increment: () => discovered++, + pendingVariables: pendingVariables, + ct: cancellationToken).ConfigureAwait(false); - // Pass 1: browse hierarchy + create folders inline, collect variables into a - // pending list. Defers variable registration until attributes are resolved — the - // address-space builder's Variable call is the one-way commit, so doing it only - // once per variable (with correct DataType/SecurityClass/IsArray) avoids the - // alternative (register with placeholders + mutate later) which the - // IAddressSpaceBuilder contract doesn't expose. - await BrowseRecursiveAsync(session, root, rootFolder, visited, - depth: 0, - discovered: () => discovered, increment: () => discovered++, - pendingVariables: pendingVariables, - ct: cancellationToken).ConfigureAwait(false); - - // Pass 2: batch-read DataType + AccessLevel + ValueRank + Historizing per - // variable. One wire request for up to ~N variables; for 10k-node servers this is - // still a couple of hundred ms total since the SDK chunks ReadAsync automatically. - await EnrichAndRegisterVariablesAsync(session, pendingVariables, cancellationToken) - .ConfigureAwait(false); - - } - finally { _gate.Release(); } + // Pass 2: batch-read DataType + AccessLevel + ValueRank + Historizing per + // variable. One wire request for up to ~N variables; for 10k-node servers this is + // still a couple of hundred ms total since the SDK chunks ReadAsync automatically. + await EnrichAndRegisterVariablesAsync(session, pendingVariables, cancellationToken) + .ConfigureAwait(false); } /// @@ -1904,7 +1903,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit TimestampsToReturn = TimestampsToReturn.Both, }); - await _gate.WaitAsync(cancellationToken).ConfigureAwait(false); + await _subscriptionGate.WaitAsync(cancellationToken).ConfigureAwait(false); try { // Re-read Session inside the critical section: a reconnect completing while we were @@ -1939,7 +1938,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit } _subscriptions[id] = new RemoteSubscription(subscription, handle, itemHandlers); } - finally { _gate.Release(); } + finally { _subscriptionGate.Release(); } return handle; @@ -1982,7 +1981,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit if (handle is not OpcUaSubscriptionHandle h) return; if (!_subscriptions.TryRemove(h.Id, out var rs)) return; - await _gate.WaitAsync(cancellationToken).ConfigureAwait(false); + await _subscriptionGate.WaitAsync(cancellationToken).ConfigureAwait(false); try { // Detach Notification handlers BEFORE deleting the subscription so the SDK's @@ -1994,7 +1993,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit try { await rs.Subscription.DeleteAsync(silent: true, cancellationToken).ConfigureAwait(false); } catch { /* best-effort — the subscription may already be gone on reconnect */ } } - finally { _gate.Release(); } + finally { _subscriptionGate.Release(); } } private static void DetachNotificationHandlers(IReadOnlyList items) @@ -2384,7 +2383,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit MonitoredItem? eventItem = null; MonitoredItemNotificationEventHandler? notifHandler = null; - await _gate.WaitAsync(cancellationToken).ConfigureAwait(false); + await _subscriptionGate.WaitAsync(cancellationToken).ConfigureAwait(false); try { // Re-read Session inside the critical section: a reconnect completing while we were @@ -2422,7 +2421,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit // current state through the normal path (deduplicated against the last-told state). await TryConditionRefreshAsync(subscription, cancellationToken).ConfigureAwait(false); } - finally { _gate.Release(); } + finally { _subscriptionGate.Release(); } return handle; @@ -2463,7 +2462,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit if (handle is not OpcUaAlarmSubscriptionHandle h) return; if (!_alarmSubscriptions.TryRemove(h.Id, out var rs)) return; - await _gate.WaitAsync(cancellationToken).ConfigureAwait(false); + await _subscriptionGate.WaitAsync(cancellationToken).ConfigureAwait(false); try { // Detach the Notification handler before deleting the subscription so the SDK's @@ -2474,7 +2473,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit try { await rs.Subscription.DeleteAsync(silent: true, cancellationToken).ConfigureAwait(false); } catch { /* best-effort — session may already be gone across a reconnect */ } } - finally { _gate.Release(); } + finally { _subscriptionGate.Release(); } } /// @@ -2761,7 +2760,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit if (_alarmSubscriptions.IsEmpty) return; try { - await _gate.WaitAsync(cancellationToken).ConfigureAwait(false); + await _subscriptionGate.WaitAsync(cancellationToken).ConfigureAwait(false); } catch (ObjectDisposedException) { return; } try @@ -2781,7 +2780,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit } finally { - try { _gate.Release(); } catch (ObjectDisposedException) { } + try { _subscriptionGate.Release(); } catch (ObjectDisposedException) { } } } @@ -3560,13 +3559,16 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit // DiscoverAsync) has definitely released before we dispose the gate. Without this // drain, a background read that calls _gate.Release() after Dispose throws // ObjectDisposedException. - try + foreach (var gate in new[] { _gate, _subscriptionGate }) { - if (_gate.Wait(TimeSpan.FromSeconds(2))) - _gate.Release(); + try + { + if (gate.Wait(TimeSpan.FromSeconds(2))) + gate.Release(); + } + catch { /* timeout or already disposed — proceed */ } + gate.Dispose(); } - catch { /* timeout or already disposed — proceed */ } - _gate.Dispose(); try { _sessionWorkCts.Cancel(); } catch (ObjectDisposedException) { } SdkTelemetry.Release(_sdkTelemetry); } @@ -3590,13 +3592,18 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit // Drain the gate before disposal so no in-flight _gate.Release() fires after // Dispose. - try + foreach (var gate in new[] { _gate, _subscriptionGate }) { - await _gate.WaitAsync(TimeSpan.FromSeconds(2)).ConfigureAwait(false); - _gate.Release(); + try + { + // Short: the host's PostStop gives shutdown + dispose 5 s together, and ShutdownAsync above + // already spent its own budget; an op still holding a gate finishes on its own token. + if (await gate.WaitAsync(TimeSpan.FromMilliseconds(500)).ConfigureAwait(false)) + gate.Release(); + } + catch { /* timeout or already disposed */ } + gate.Dispose(); } - catch { /* timeout or already disposed */ } - _gate.Dispose(); try { _sessionWorkCts.Cancel(); } catch (ObjectDisposedException) { } SdkTelemetry.Release(_sdkTelemetry); } diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientGateSplitTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientGateSplitTests.cs new file mode 100644 index 00000000..b78ba971 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientGateSplitTests.cs @@ -0,0 +1,153 @@ +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-6 — one SemaphoreSlim serialised reads, writes, subscribes, acks AND the +/// whole recursive discovery browse, so a stalled write (up to the operation timeout) blocked every read and +/// subscription create, and a discovery browse blocked the data plane for its full duration. Reads (and acks) +/// keep their gate; writes and the discovery browse take none (the SDK session is thread-safe for concurrent +/// service calls); subscription management has its own gate. +/// +[Trait("Category", "Unit")] +public sealed class OpcUaClientGateSplitTests +{ + private const string SpeedRef = "Plant/dev/Speed"; + private static readonly TimeSpan Bound = TimeSpan.FromSeconds(10); + + private static readonly IReadOnlyList Tags = + [new(SpeedRef, JsonSerializer.Serialize(new { nodeId = "ns=0;s=Speed" }), WriteIdempotent: true)]; + + private static (OpcUaClientDriver Driver, Mock Session) NewConnectedDriver() + { + var drv = new OpcUaClientDriver(new OpcUaClientDriverOptions { RawTags = Tags }, "opcua-o6"); + var session = new Mock(MockBehavior.Loose); + session.SetupGet(s => s.MessageContext).Returns(new ServiceMessageContext(SdkTelemetry.Context)); + session.SetupGet(s => s.NamespaceUris).Returns(new NamespaceTable()); + 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(_ => new DataValue(new Variant(42), StatusCodes.Good, DateTime.UtcNow, DateTime.UtcNow))], + DiagnosticInfos = [], + }); + drv.SetSessionForTest(session.Object); + drv.SetNamespaceMapForTest(NamespaceMap.FromTable(new NamespaceTable(), Tags)); + drv.SetHealthForTest(new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null)); + return (drv, session); + } + + private static async Task ReadCompletes(OpcUaClientDriver drv, string because) + { + var result = await drv.ReadAsync([SpeedRef], TestContext.Current.CancellationToken).WaitAsync(Bound, TestContext.Current.CancellationToken); + result.Single().Value.ShouldBe(42, because); + } + + /// A write stalled on the wire does not hold up a read. + [Fact] + public async Task A_stalled_write_does_not_block_a_read() + { + var (drv, session) = NewConnectedDriver(); + using var _ = drv; + var writeOnWire = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var releaseWrite = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + session.Setup(s => s.WriteAsync(It.IsAny(), It.IsAny(), It.IsAny())) + .Returns(async (RequestHeader _, WriteValueCollection nodes, CancellationToken _) => + { + writeOnWire.TrySetResult(); + await releaseWrite.Task; + return new WriteResponse + { + ResponseHeader = new ResponseHeader(), + Results = [.. nodes.Select(_ => (StatusCode)StatusCodes.Good)], + DiagnosticInfos = [], + }; + }); + + var write = drv.WriteAsync([new ZB.MOM.WW.OtOpcUa.Core.Abstractions.WriteRequest(SpeedRef, 7)], TestContext.Current.CancellationToken); + await writeOnWire.Task.WaitAsync(Bound, TestContext.Current.CancellationToken); + + await ReadCompletes(drv, "a read must not queue behind a stalled write"); + + releaseWrite.SetResult(); + (await write).Single().StatusCode.ShouldBe(0u); + } + + /// Subscription management queues on its own gate, not the read gate. + [Fact] + public async Task Subscription_management_does_not_share_the_read_gate() + { + var (drv, _) = NewConnectedDriver(); + using var __ = drv; + + await drv.SubscriptionGate.WaitAsync(TestContext.Current.CancellationToken); + var subscribe = drv.SubscribeAsync([SpeedRef], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken); + + // Positive control: the subscribe really is parked on the subscription gate. + subscribe.IsCompleted.ShouldBeFalse(); + await ReadCompletes(drv, "a read must not queue behind subscription management"); + + drv.SubscriptionGate.Release(); + // The mock never attaches the subscription, so the create itself throws — what matters is that it ran. + await Should.ThrowAsync(() => subscribe.WaitAsync(Bound, TestContext.Current.CancellationToken)); + } + + /// A discovery browse in progress does not hold the read gate. + [Fact] + public async Task A_discovery_browse_in_progress_does_not_block_a_read() + { + var (drv, session) = NewConnectedDriver(); + using var _ = drv; + var browseOnWire = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var releaseBrowse = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + session.Setup(s => s.BrowseAsync( + It.IsAny(), It.IsAny(), It.IsAny(), + It.IsAny(), It.IsAny())) + .Returns(async (RequestHeader _, ViewDescription _, uint _, BrowseDescriptionCollection _, CancellationToken _) => + { + browseOnWire.TrySetResult(); + await releaseBrowse.Task; + return new BrowseResponse + { + ResponseHeader = new ResponseHeader(), + Results = [new BrowseResult { References = [], ContinuationPoint = [] }], + DiagnosticInfos = [], + }; + }); + + var discover = drv.DiscoverAsync(new NullAddressSpaceBuilder(), TestContext.Current.CancellationToken); + await browseOnWire.Task.WaitAsync(Bound, TestContext.Current.CancellationToken); + + await ReadCompletes(drv, "a read must not queue behind a recursive discovery browse"); + + releaseBrowse.SetResult(); + await discover.WaitAsync(Bound, TestContext.Current.CancellationToken); + } + + private sealed class NullAddressSpaceBuilder : IAddressSpaceBuilder + { + public IAddressSpaceBuilder Folder(string browseName, string displayName) => this; + + public IVariableHandle Variable(string browseName, string displayName, DriverAttributeInfo attributeInfo) + => new StubHandle(); + + public void AddProperty(string browseName, DriverDataType dataType, object? value) { } + + public void AttachAlarmCondition(IVariableHandle sourceVariable, string alarmName, DriverAttributeInfo alarmInfo) { } + + private sealed class StubHandle : IVariableHandle + { + public string FullReference => "stub"; + public IAlarmConditionSink MarkAsAlarmCondition(AlarmConditionInfo info) => throw new NotSupportedException(); + } + } +} From cba3b2e96025400b08e4bee5ae1e147fe9bcf2d1 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 15:26:43 -0400 Subject: [PATCH 05/27] fix(opcuaclient): writes feed health; a good keep-alive clears a stale Degraded Health contract: a write response (per-node refusals included) is the server answering, so it restores Healthy and advances LastSuccessfulRead; a write with no reply (transport failure / timeout) writes Degraded with the stamp preserved; a caller-cancelled write touches nothing. A good keep-alive with no outage or dead subscription latched now clears a Degraded left by a failed read/write instead of only moving the stamp (a subscription-only driver otherwise stayed Degraded until its next re-initialise). All writes are generation-stamped. Refs #743 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT --- docs/drivers/OpcUaClient.md | 5 + .../OpcUaClientDriver.cs | 20 ++- .../OpcUaClientWriteHealthTests.cs | 120 ++++++++++++++++++ 3 files changed, 142 insertions(+), 3 deletions(-) create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientWriteHealthTests.cs diff --git a/docs/drivers/OpcUaClient.md b/docs/drivers/OpcUaClient.md index aa9d4e3f..d189f57c 100644 --- a/docs/drivers/OpcUaClient.md +++ b/docs/drivers/OpcUaClient.md @@ -91,6 +91,11 @@ 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. + A good keep-alive with no outage latched also **clears a `Degraded` left by a failed read/write** (it used to move + only the stamp, so a subscription-only driver stayed `Degraded` until its next re-initialise). +- **Read / write / ack responses are replies.** Any service response — per-node refusals such as `BadNotWritable` + included — restores `Healthy` and advances the stamp; only a call that got **no** response (transport failure, + timeout) writes `Degraded` with the stamp preserved. A write the caller cancelled touches no health. - **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 and **source + server timestamp = the detection time** (live-gate F1: stamped with the 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 4e757d6d..a6497254 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriver.cs @@ -1437,6 +1437,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit _ = RequireSession(); var results = new WriteResult[writes.Count]; + var generation = Volatile.Read(ref _lifecycleGeneration); var session = Session; if (session is null) { @@ -1483,8 +1484,10 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit // the severity bits set is a Bad. results[r] = new WriteResult(codes[w].Code); } + // A response — refusals included (BadNotWritable, BadTypeMismatch, ...) — is the server answering. + RecordDeviceAnswered(generation, Clock.GetUtcNow().UtcDateTime); } - catch (OperationCanceledException) + catch (OperationCanceledException ex) { // Timeout / cancellation after the wire request may have been dispatched. // Writes are non-idempotent — BadTimeout ("outcome unknown, do not blindly @@ -1494,12 +1497,15 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit const uint StatusBadTimeout = 0x800A0000u; for (var w = 0; w < indexMap.Count; w++) results[indexMap[w]] = new WriteResult(StatusBadTimeout); + // No reply arrived. Only a timeout is a comms fault — a caller that cancelled says nothing about upstream. + if (!cancellationToken.IsCancellationRequested) RecordCommsFault(generation, $"write: {ex.Message}"); } - catch (Exception) + catch (Exception ex) { // Pre-wire transport failure — the write definitely did not reach the server. for (var w = 0; w < indexMap.Count; w++) results[indexMap[w]] = new WriteResult(StatusBadCommunicationError); + RecordCommsFault(generation, $"write: {ex.Message}"); } return results; } @@ -3125,7 +3131,15 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit return; } - UpdateHealth(generation, h => h.State is DriverState.Healthy or DriverState.Degraded ? h with { LastSuccessfulRead = now } : null); + // A good keep-alive is the server answering. With no outage and no dead subscription latched, a Degraded + // here can only be a failed read/write the server has since contradicted, so it clears too — otherwise a + // subscription-only driver (which never reads again) stayed Degraded until its next re-initialise. + UpdateHealth(generation, h => h.State switch + { + DriverState.Healthy => h with { LastSuccessfulRead = now }, + DriverState.Degraded => new DriverHealth(DriverState.Healthy, now, null), + _ => null, + }); return; } diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientWriteHealthTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientWriteHealthTests.cs new file mode 100644 index 00000000..10e4637e --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientWriteHealthTests.cs @@ -0,0 +1,120 @@ +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; + +/// +/// The health contract for the write path and the steady keep-alive: a write RESPONSE — whatever per-node +/// statuses it carries, a refusal included — is the server answering, so it restores Healthy and advances +/// ; a write that got NO response degrades health with the stamp +/// preserved; a caller-cancelled write is not a comms fault. A good keep-alive clears a Degraded left by a failed +/// read/write (it used to only move the stamp, leaving a subscription-only driver Degraded forever). +/// +[Trait("Category", "Unit")] +public sealed class OpcUaClientWriteHealthTests +{ + private const string SpeedRef = "Plant/dev/Speed"; + private static readonly DateTime Stale = new(2020, 1, 1, 0, 0, 0, DateTimeKind.Utc); + + private static readonly IReadOnlyList Tags = + [new(SpeedRef, JsonSerializer.Serialize(new { nodeId = "ns=0;s=Speed" }), WriteIdempotent: true)]; + + private static (OpcUaClientDriver Driver, Mock Session) NewConnectedDriver(DriverHealth health) + { + var drv = new OpcUaClientDriver(new OpcUaClientDriverOptions { RawTags = Tags }, "opcua-write-health"); + var session = new Mock(MockBehavior.Loose); + session.SetupGet(s => s.MessageContext).Returns(new ServiceMessageContext(SdkTelemetry.Context)); + session.SetupGet(s => s.NamespaceUris).Returns(new NamespaceTable()); + drv.SetSessionForTest(session.Object); + drv.SetNamespaceMapForTest(NamespaceMap.FromTable(new NamespaceTable(), Tags)); + drv.SetHealthForTest(health); + return (drv, session); + } + + private static ZB.MOM.WW.OtOpcUa.Core.Abstractions.WriteRequest[] Write() => [new(SpeedRef, 7)]; + + /// A refused write (BadNotWritable) is a reply: Healthy again, stamp advanced. + [Fact] + public async Task A_refused_write_is_a_reply_and_restores_healthy() + { + var (drv, session) = NewConnectedDriver(new DriverHealth(DriverState.Degraded, Stale, "read failed")); + using var _ = drv; + session.Setup(s => s.WriteAsync(It.IsAny(), It.IsAny(), It.IsAny())) + .ReturnsAsync(new WriteResponse + { + ResponseHeader = new ResponseHeader(), + Results = [StatusCodes.BadNotWritable], + DiagnosticInfos = [], + }); + + var results = await drv.WriteAsync(Write(), TestContext.Current.CancellationToken); + + results.Single().StatusCode.ShouldBe(StatusCodes.BadNotWritable, "the upstream refusal still passes through verbatim"); + var h = drv.GetHealth(); + h.State.ShouldBe(DriverState.Healthy); + h.LastSuccessfulRead.ShouldNotBeNull().ShouldBeGreaterThan(Stale); + h.LastError.ShouldBeNull(); + } + + /// A write with no reply (transport failure) degrades health and preserves the stamp. + [Fact] + public async Task A_write_with_no_reply_degrades_and_preserves_the_stamp() + { + var (drv, session) = NewConnectedDriver(new DriverHealth(DriverState.Healthy, Stale, null)); + using var _ = drv; + session.Setup(s => s.WriteAsync(It.IsAny(), It.IsAny(), It.IsAny())) + .ThrowsAsync(new ServiceResultException(StatusCodes.BadConnectionClosed)); + + var results = await drv.WriteAsync(Write(), TestContext.Current.CancellationToken); + + results.Single().StatusCode.ShouldBe(0x80050000u); // BadCommunicationError + var h = drv.GetHealth(); + h.State.ShouldBe(DriverState.Degraded); + h.LastSuccessfulRead.ShouldBe(Stale, "the comms-loss watchdog measures silence from this stamp"); + } + + /// + /// A write the CALLER cancelled says nothing about the upstream: health is left alone. (The positive control is + /// the test above — the same throw-path does write health when the failure is the transport's.) + /// + [Fact] + public async Task A_caller_cancelled_write_leaves_health_alone() + { + var healthy = new DriverHealth(DriverState.Healthy, Stale, null); + var (drv, session) = NewConnectedDriver(healthy); + using var _ = drv; + using var cts = new CancellationTokenSource(); + session.Setup(s => s.WriteAsync(It.IsAny(), It.IsAny(), It.IsAny())) + .Returns(async (RequestHeader _, WriteValueCollection _, CancellationToken ct) => + { + cts.Cancel(); + await Task.Delay(Timeout.Infinite, ct); + return new WriteResponse(); + }); + + var results = await drv.WriteAsync(Write(), cts.Token); + + results.Single().StatusCode.ShouldBe(0x800A0000u); // BadTimeout — outcome unknown + drv.GetHealth().ShouldBe(healthy); + } + + /// A good keep-alive is a reply: it clears a Degraded a failed read left behind. + [Fact] + public void A_good_keep_alive_clears_a_degraded_left_by_a_failed_read() + { + var (drv, _) = NewConnectedDriver(new DriverHealth(DriverState.Degraded, Stale, "read: BadTimeout")); + using var __ = drv; + + drv.RaiseKeepAliveForTest(StatusCodes.Good); + + var h = drv.GetHealth(); + h.State.ShouldBe(DriverState.Healthy); + h.LastSuccessfulRead.ShouldNotBeNull().ShouldBeGreaterThan(Stale); + h.LastError.ShouldBeNull(); + } +} From d98f4a5bf173ef66ba820ec16b9424a51b2d07ba Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 15:28:20 -0400 Subject: [PATCH 06/27] fix(opcuaclient): wire ProbeTimeoutSeconds into the Test Connect probe (R06:O-4) The knob was authored by the AdminUI and parsed, but consumed by nothing. It now caps the probe deadline (shorter of the caller's budget and the config; never extends it; 1 s floor), the same rule as the Galaxy probe. Refs #743 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT --- docs/drivers/OpcUaClient.md | 2 + .../OpcUaClientDriverProbe.cs | 20 ++++++- .../OpcUaClientProbeTimeoutTests.cs | 54 +++++++++++++++++++ 3 files changed, 75 insertions(+), 1 deletion(-) create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientProbeTimeoutTests.cs diff --git a/docs/drivers/OpcUaClient.md b/docs/drivers/OpcUaClient.md index d189f57c..85f6168d 100644 --- a/docs/drivers/OpcUaClient.md +++ b/docs/drivers/OpcUaClient.md @@ -53,6 +53,8 @@ for every field (security policy/mode, auth type, session timeout, keep-alive, browse root, node/depth caps). `ReconnectPeriod` was **deleted** (#676 — unused since #670 removed the SDK reconnect handler, and the AdminUI no longer shows it); stored configs that still carry `reconnectPeriod` keep parsing in the factory, the Test-Connect probe and the browse picker (unknown keys are skipped), and the form drops it on the next save. +`ProbeTimeoutSeconds` (default 15, AdminUI "Probe timeout") caps the Test-Connect probe's deadline — it can shorten +the budget the host hands the probe, never extend it (review 2026-10-21 O-4; it used to be authored and consumed by nothing). ### Endpoint address: the configured host and port win (#676) diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriverProbe.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriverProbe.cs index 748ff0c3..e4695e7a 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriverProbe.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriverProbe.cs @@ -51,7 +51,8 @@ public sealed class OpcUaClientDriverProbe : IDriverProbe // Bound the whole probe (both phases) by the caller timeout, independent of `ct` — a // stalled OPC UA endpoint that accepts TCP but never answers GetEndpoints must not block - // the probe when the caller passes CancellationToken.None. + // the probe when the caller passes CancellationToken.None. The config's ProbeTimeoutSeconds caps it (O-4). + timeout = EffectiveTimeout(timeout, opts.ProbeTimeoutSeconds); using var cts = CancellationTokenSource.CreateLinkedTokenSource(ct); cts.CancelAfter(timeout); @@ -106,6 +107,23 @@ public sealed class OpcUaClientDriverProbe : IDriverProbe } } + /// + /// R06:O-4 — the deadline the probe actually runs under: the caller's budget, tightened by the config's + /// when that is positive and shorter (it can only + /// shorten the budget — the caller's own deadline fires regardless). A non-positive caller budget is floored + /// to 1 s so the probe runs briefly rather than cancelling before it starts. Same rule as the Galaxy probe. + /// + /// The budget the host handed the probe. + /// The config's ProbeTimeoutSeconds; ignored when not positive. + /// The effective probe deadline. + internal static TimeSpan EffectiveTimeout(TimeSpan callerTimeout, int configuredSeconds) + { + var caller = callerTimeout > TimeSpan.Zero ? callerTimeout : TimeSpan.FromSeconds(1); + if (configuredSeconds <= 0) return caller; + var configured = TimeSpan.FromSeconds(configuredSeconds); + return configured < caller ? configured : caller; + } + /// /// Builds a minimal suitable for unauthenticated /// discovery (GetEndpoints). No PKI stores, no session certificate — discovery traffic diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientProbeTimeoutTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientProbeTimeoutTests.cs new file mode 100644 index 00000000..584952b1 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientProbeTimeoutTests.cs @@ -0,0 +1,54 @@ +using System.Diagnostics; +using System.Net; +using System.Net.Sockets; +using Shouldly; +using Xunit; + +namespace ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests; + +/// +/// Review 2026-10-21 R06:O-4 — ProbeTimeoutSeconds was authored by the AdminUI and parsed, but consumed +/// by nothing: the Test Connect probe ran on the caller's budget alone. It now caps the probe deadline (it can +/// shorten the caller's budget, never extend it) — the same rule as the Galaxy probe. +/// +[Trait("Category", "Unit")] +public sealed class OpcUaClientProbeTimeoutTests +{ + /// The effective deadline is the shorter of the caller's budget and the configured cap. + [Theory] + [InlineData(10, 30, 10)] // caller shorter → caller wins + [InlineData(10, 2, 2)] // config shorter → config caps + [InlineData(10, 0, 10)] // unset config → caller + [InlineData(10, -1, 10)] // nonsense config → caller + [InlineData(0, 5, 1)] // no caller budget → 1 s floor beats a longer config + public void EffectiveTimeout_is_the_shorter_of_caller_and_config(int callerSeconds, int configSeconds, int expectedSeconds) + => OpcUaClientDriverProbe.EffectiveTimeout(TimeSpan.FromSeconds(callerSeconds), configSeconds) + .ShouldBe(TimeSpan.FromSeconds(expectedSeconds)); + + /// + /// Against a port that accepts TCP but never speaks OPC UA, a 1 s configured cap returns long before the + /// caller's 30 s budget — the discovery handshake does the waiting, so this exercises the real deadline path. + /// + [Fact] + public async Task Configured_cap_shortens_the_probe_against_a_silent_listener() + { + var listener = new TcpListener(IPAddress.Loopback, 0); + listener.Start(); + try + { + var port = ((IPEndPoint)listener.LocalEndpoint).Port; + var config = $$"""{ "endpointUrl": "opc.tcp://127.0.0.1:{{port}}", "probeTimeoutSeconds": 1 }"""; + + var sw = Stopwatch.StartNew(); + var result = await new OpcUaClientDriverProbe().ProbeAsync(config, TimeSpan.FromSeconds(30), TestContext.Current.CancellationToken); + sw.Stop(); + + result.Ok.ShouldBeFalse(); + sw.Elapsed.ShouldBeLessThan(TimeSpan.FromSeconds(15), $"the 1 s configured cap must bound the probe (message: {result.Message})"); + } + finally + { + listener.Stop(); + } + } +} From fcdc44f9f7940bbfd99c3d6c99006e7cc5c48e98 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 15:33:08 -0400 Subject: [PATCH 07/27] feat(opcuaclient-browser): real AttributesAsync + certificate-auth browse (R06:O-5) - AttributesAsync was a stub returning nothing; it now reads NodeClass / DisplayName / DataType / ValueRank / UserAccessLevel in one request and maps them through UpstreamAttributeMapping, the same mapping discovery registers with (moved into the shared .Session leaf; the driver forwards to it). Objects have no value attributes. - Certificate auth was refused "in v1". The browse now builds its identity through OpcUaUserIdentityFactory (shared .Session leaf, the driver forwards to it), including the secret: resolution of Password / UserCertificatePassword through the host's ISecretResolver (optional ctor param, DI-injected). Fail-closed: a secret: ref with no store throws instead of being sent as the literal (the browse used to send a secret: Password verbatim). The identity is built before any network I/O. Refs #743 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT --- docs/drivers/OpcUaClient.md | 3 +- .../OpcUaClientBrowseSession.cs | 58 ++++++++- .../OpcUaClientDriverBrowser.cs | 40 +++--- .../OpcUaUserIdentityFactory.cs | 115 ++++++++++++++++++ .../UpstreamAttributeMapping.cs | 65 ++++++++++ ....OtOpcUa.Driver.OpcUaClient.Session.csproj | 9 +- .../OpcUaClientDriver.cs | 69 ++--------- .../OpcUaClientSecretResolution.cs | 50 +------- ...OpcUaClientBrowseSessionAttributesTests.cs | 103 ++++++++++++++++ .../OpcUaClientBrowserCertificateAuthTests.cs | 99 +++++++++++++++ .../OpcUaClientDriverBrowserTests.cs | 15 +-- 11 files changed, 492 insertions(+), 134 deletions(-) create mode 100644 src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Session/OpcUaUserIdentityFactory.cs create mode 100644 src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Session/UpstreamAttributeMapping.cs create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Browser.Tests/OpcUaClientBrowseSessionAttributesTests.cs create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Browser.Tests/OpcUaClientBrowserCertificateAuthTests.cs diff --git a/docs/drivers/OpcUaClient.md b/docs/drivers/OpcUaClient.md index 85f6168d..66651886 100644 --- a/docs/drivers/OpcUaClient.md +++ b/docs/drivers/OpcUaClient.md @@ -25,7 +25,8 @@ out-of-process isolation. | Project | Target | Role | |---------|--------|------| | `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/` | net10.0 | In-process driver — session lifetime, read / write / subscribe / alarm / history passthrough | -| `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Browser/` | net10.0 | `IDriverBrowser` — live address-picker browse used by the AdminUI | +| `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Browser/` | net10.0 | `IDriverBrowser` — live address-picker browse used by the AdminUI. Its side-panel `AttributesAsync` reads a variable's DataType / ValueRank / UserAccessLevel and maps them through the same `UpstreamAttributeMapping` discovery uses; it authenticates Anonymous, Username **or Certificate** through the driver's own `OpcUaUserIdentityFactory`, resolving `secret:` credentials through the host's secret store (fail-closed) — review 2026-10-21 O-5 (it was a stub, and Certificate was refused) | +| `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Session/` | net10.0 | The SDK-typed layer the driver AND the browser share: `NamespaceMap`, `UpstreamAttributeMapping`, `OpcUaUserIdentityFactory` | | `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Contracts/` | net10.0 | Config records + enums bound from `DriverConfig` JSON | ## Minimum deployment diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Browser/OpcUaClientBrowseSession.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Browser/OpcUaClientBrowseSession.cs index 42f10018..fa534016 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Browser/OpcUaClientBrowseSession.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Browser/OpcUaClientBrowseSession.cs @@ -54,10 +54,64 @@ internal sealed class OpcUaClientBrowseSession : IBrowseSession } /// - public Task> AttributesAsync(string nodeId, CancellationToken cancellationToken) + // R06:O-5 — reads the node's NodeClass / DisplayName / DataType / ValueRank / UserAccessLevel in one request and + // maps them through the same UpstreamAttributeMapping the runtime driver's discovery registers with, so the + // side-panel shows what a browse-commit would author. An Object (folder) has no value attributes → empty. + // Throws ArgumentException when nodeId cannot be resolved against the live session's namespace table. + public async Task> AttributesAsync(string nodeId, CancellationToken cancellationToken) { + ObjectDisposedException.ThrowIf(_disposed, this); + if (!NamespaceMap.TryResolve(_session, nodeId, out var resolved)) + throw new ArgumentException( + $"Cannot resolve NodeId '{nodeId}' against the live session.", nameof(nodeId)); + + uint[] attributeIds = + [ + Attributes.NodeClass, Attributes.DisplayName, Attributes.DataType, Attributes.ValueRank, + Attributes.UserAccessLevel, + ]; + var toRead = new ReadValueIdCollection(attributeIds.Length); + foreach (var id in attributeIds) toRead.Add(new ReadValueId { NodeId = resolved, AttributeId = id }); + + DataValueCollection values; + await _gate.WaitAsync(cancellationToken).ConfigureAwait(false); + try + { + var resp = await _session.ReadAsync( + requestHeader: null, + maxAge: 0, + timestampsToReturn: TimestampsToReturn.Neither, + nodesToRead: toRead, + ct: cancellationToken).ConfigureAwait(false); + values = resp?.Results ?? []; + } + finally + { + _gate.Release(); + } LastUsedUtc = DateTime.UtcNow; - return Task.FromResult>(Array.Empty()); + + DataValue? At(int i) => i < values.Count && StatusCode.IsGood(values[i].StatusCode) + ? values[i] : null; + + var nodeClass = At(0)?.Value is int nc ? (NodeClass)nc : NodeClass.Unspecified; + if (nodeClass != NodeClass.Variable) return Array.Empty(); + + var name = (At(1)?.Value as LocalizedText)?.Text ?? resolved.ToString() ?? nodeId; + var dataType = At(2)?.Value is NodeId dt + ? UpstreamAttributeMapping.MapDataType(dt) + : ZB.MOM.WW.OtOpcUa.Core.Abstractions.DriverDataType.String; + var valueRank = At(3)?.Value is int vr ? vr : ValueRanks.Scalar; + var access = At(4)?.Value is byte ab ? ab : (byte)0; + + return + [ + new AttributeInfo( + Name: name, + DriverDataType: dataType.ToString(), + IsArray: UpstreamAttributeMapping.IsArray(valueRank), + SecurityClass: UpstreamAttributeMapping.MapAccessLevel(access).ToString()), + ]; } /// Issue a single-level Browse (plus continuation-point follow-ups) under the diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Browser/OpcUaClientDriverBrowser.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Browser/OpcUaClientDriverBrowser.cs index 6573e87d..845d2caa 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Browser/OpcUaClientDriverBrowser.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Browser/OpcUaClientDriverBrowser.cs @@ -7,7 +7,8 @@ using Opc.Ua.Client; using Opc.Ua.Configuration; using ZB.MOM.WW.OtOpcUa.Commons.Browsing; using ZB.MOM.WW.OtOpcUa.Core.Abstractions; -using ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient; // OpcUaClientDriverOptions + NamespaceMap +using ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient; // OpcUaClientDriverOptions + NamespaceMap + OpcUaUserIdentityFactory +using ZB.MOM.WW.Secrets.Abstractions; namespace ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Browser; @@ -33,12 +34,19 @@ public sealed class OpcUaClientDriverBrowser : IDriverBrowser }; private readonly ILogger _logger; + private readonly ISecretResolver? _secretResolver; /// Creates a new browser. Logger defaults to NullLogger when not supplied. /// Optional logger; defaults to . - public OpcUaClientDriverBrowser(ILogger? logger = null) + /// + /// The host's shared secret resolver (DI-injected where one is registered), used to resolve + /// secret:-prefixed Password / UserCertificatePassword exactly as the runtime driver does. + /// Null ⇒ a secret: ref fails closed (it is never sent as the literal). + /// + public OpcUaClientDriverBrowser(ILogger? logger = null, ISecretResolver? secretResolver = null) { _logger = logger ?? NullLogger.Instance; + _secretResolver = secretResolver; } /// @@ -58,9 +66,10 @@ public sealed class OpcUaClientDriverBrowser : IDriverBrowser if (string.IsNullOrWhiteSpace(endpoint)) throw new InvalidOperationException("OpcUaClient browser requires EndpointUrl or EndpointUrls[0]."); - if (opts.AuthType == OpcUaAuthType.Certificate) - throw new InvalidOperationException( - "Browser does not support OpcUaAuthType.Certificate in v1; use Anonymous or Username."); + // R06:O-5 — Certificate auth is supported: the identity is built (and any secret: credential resolved) by the + // same factory the runtime driver uses, BEFORE any network I/O, so a bad certificate path / password fails fast + // with its own message. + var identity = await BuildBrowseUserIdentityAsync(opts, _secretResolver, cancellationToken).ConfigureAwait(false); if (opts.AutoAcceptCertificates) _logger.LogWarning( @@ -70,7 +79,6 @@ public sealed class OpcUaClientDriverBrowser : IDriverBrowser endpoint); var appConfig = await BuildBrowseAppConfigurationAsync(cancellationToken).ConfigureAwait(false); - var identity = BuildBrowseUserIdentity(opts); var perEndpointBudget = TimeSpan.FromSeconds( Math.Clamp(opts.PerEndpointConnectTimeout.TotalSeconds, 5, 30)); @@ -175,16 +183,20 @@ public sealed class OpcUaClientDriverBrowser : IDriverBrowser return config; } - /// Build the OPC UA user identity from the form's auth fields. + /// + /// Build the OPC UA user identity from the form's auth fields — Anonymous, Username or Certificate — through + /// the factory the runtime driver uses, after resolving any secret:-prefixed credential (fail-closed). + /// /// Driver options carrying the form's auth fields. - private static UserIdentity BuildBrowseUserIdentity(OpcUaClientDriverOptions opts) => opts.AuthType switch + /// The secret store, or null when none is available. + /// Cancellation token for the secret resolution. + /// The identity the browse session activates with. + internal static async Task BuildBrowseUserIdentityAsync( + OpcUaClientDriverOptions opts, ISecretResolver? secretResolver, CancellationToken ct) { - OpcUaAuthType.Anonymous => new UserIdentity(new AnonymousIdentityToken()), - OpcUaAuthType.Username => new UserIdentity( - opts.Username ?? string.Empty, - System.Text.Encoding.UTF8.GetBytes(opts.Password ?? string.Empty)), - _ => new UserIdentity(new AnonymousIdentityToken()), - }; + var resolved = await OpcUaUserIdentityFactory.ResolveSecretRefsAsync(opts, secretResolver, ct).ConfigureAwait(false); + return OpcUaUserIdentityFactory.Build(resolved); + } /// Select the endpoint matching the requested SecurityPolicy + SecurityMode pair. /// Application configuration used by the discovery client. diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Session/OpcUaUserIdentityFactory.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Session/OpcUaUserIdentityFactory.cs new file mode 100644 index 00000000..16be7eed --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Session/OpcUaUserIdentityFactory.cs @@ -0,0 +1,115 @@ +using Opc.Ua; +using ZB.MOM.WW.Secrets.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient; + +/// +/// The ONE place the OpcUaClient user identity is built from — shared by +/// the runtime driver's session open and the AdminUI live browse, so a certificate-authenticated upstream is +/// browsable with exactly the credentials the deployed driver will present. Includes the secret: +/// resolution of the two credential fields (fail-closed). +/// +public static class OpcUaUserIdentityFactory +{ + private const string SecretPrefix = "secret:"; + + /// + /// Build the user-identity token from options whose credential fields are already resolved plaintext + /// (see ). + /// + /// Driver options containing authentication configuration. + /// The constructed for session activation. + public static UserIdentity Build(OpcUaClientDriverOptions options) => + options.AuthType switch + { + OpcUaAuthType.Anonymous => new UserIdentity(new AnonymousIdentityToken()), + OpcUaAuthType.Username => new UserIdentity( + options.Username ?? string.Empty, + System.Text.Encoding.UTF8.GetBytes(options.Password ?? string.Empty)), + OpcUaAuthType.Certificate => FromCertificate(options), + _ => new UserIdentity(new AnonymousIdentityToken()), + }; + + /// + /// Build a carrying a client user-authentication certificate loaded from + /// . Fails fast if the path is missing, the file + /// doesn't exist, or the certificate lacks a private key (required to sign the user-token challenge). + /// + /// Driver options containing certificate configuration. + /// The certificate-backed . + public static UserIdentity FromCertificate(OpcUaClientDriverOptions options) + { + if (string.IsNullOrWhiteSpace(options.UserCertificatePath)) + throw new InvalidOperationException( + "OpcUaAuthType.Certificate requires OpcUaClientDriverOptions.UserCertificatePath to be set."); + if (!File.Exists(options.UserCertificatePath)) + throw new FileNotFoundException( + $"User certificate not found at '{options.UserCertificatePath}'.", + options.UserCertificatePath); + + // X509CertificateLoader (new in .NET 9) is the only non-obsolete way to load a PFX + // since the legacy X509Certificate2 ctors are marked obsolete on net10. + var cert = System.Security.Cryptography.X509Certificates.X509CertificateLoader + .LoadPkcs12FromFile(options.UserCertificatePath, options.UserCertificatePassword); + + if (!cert.HasPrivateKey) + throw new InvalidOperationException( + $"User certificate at '{options.UserCertificatePath}' has no private key — " + + "the private key is required to sign the OPC UA user-token challenge at session activation."); + + return new UserIdentity(cert); + } + + /// + /// Return a copy of with any secret:-prefixed + /// / + /// resolved through . Non-secret / null / empty fields pass through unchanged (the + /// reference-equal original is returned when nothing changed). Fail-closed: an absent secret, or a + /// secret: ref with no resolver available, throws rather than sending the literal as a credential. + /// + /// The raw options (credential fields may carry secret: refs). + /// The shared secret resolver, or null when none is available. + /// Cancellation token. + /// An options copy whose credential fields carry resolved plaintext. + public static async Task ResolveSecretRefsAsync( + OpcUaClientDriverOptions options, ISecretResolver? resolver, CancellationToken ct) + { + ArgumentNullException.ThrowIfNull(options); + + var password = await ResolveFieldAsync(options.Password, nameof(options.Password), resolver, ct) + .ConfigureAwait(false); + var certPassword = await ResolveFieldAsync( + options.UserCertificatePassword, nameof(options.UserCertificatePassword), resolver, ct) + .ConfigureAwait(false); + + if (ReferenceEquals(password, options.Password) + && ReferenceEquals(certPassword, options.UserCertificatePassword)) + { + return options; + } + + return options with { Password = password, UserCertificatePassword = certPassword }; + } + + private static async Task ResolveFieldAsync( + string? value, string fieldName, ISecretResolver? resolver, CancellationToken ct) + { + if (string.IsNullOrEmpty(value) + || !value.StartsWith(SecretPrefix, StringComparison.OrdinalIgnoreCase)) + { + return value; + } + + var name = value[SecretPrefix.Length..]; + if (resolver is null) + throw new InvalidOperationException( + $"OpcUaClientDriverOptions.{fieldName}='{value}' references secret '{name}', but no secret store is " + + "available here (fail-closed)."); + var resolved = await resolver.GetAsync(new SecretName(name), ct).ConfigureAwait(false); + return !string.IsNullOrEmpty(resolved) + ? resolved + : throw new InvalidOperationException( + $"OpcUaClientDriverOptions.{fieldName}='{value}' resolves secret '{name}', but it is " + + "absent from the store (fail-closed)."); + } +} diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Session/UpstreamAttributeMapping.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Session/UpstreamAttributeMapping.cs new file mode 100644 index 00000000..046fa2e1 --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Session/UpstreamAttributeMapping.cs @@ -0,0 +1,65 @@ +using Opc.Ua; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient; + +/// +/// The ONE mapping from an upstream variable's OPC UA attributes (DataType, ValueRank, AccessLevel) to the +/// driver-agnostic shape — shared by the runtime driver's discovery enrichment and the AdminUI live browse +/// side-panel, so the picker shows exactly what a browse-commit / discovery would register. +/// +public static class UpstreamAttributeMapping +{ + /// + /// Map an upstream OPC UA built-in DataType NodeId (via DataTypeIds.*) to a + /// . Unknown / custom types fall through to + /// which is the safest passthrough for + /// Variant-wrapped structs + enums + extension objects. + /// + /// The OPC UA data type NodeId to map. + /// The mapped . + public static DriverDataType MapDataType(NodeId dataType) + { + if (dataType == DataTypeIds.Boolean) return DriverDataType.Boolean; + // SByte (signed 8-bit) shares Int16 — DriverDataType has no narrower signed type. + // Byte (unsigned 8-bit) belongs in the unsigned family → UInt16, not Int16 + // (mapping an unsigned 0-255 type onto Int16 misrepresents type metadata and + // confuses range/validation logic keyed off DriverDataType). + if (dataType == DataTypeIds.SByte || dataType == DataTypeIds.Int16) return DriverDataType.Int16; + if (dataType == DataTypeIds.Byte || dataType == DataTypeIds.UInt16) return DriverDataType.UInt16; + if (dataType == DataTypeIds.Int32) return DriverDataType.Int32; + if (dataType == DataTypeIds.UInt32) return DriverDataType.UInt32; + if (dataType == DataTypeIds.Int64) return DriverDataType.Int64; + if (dataType == DataTypeIds.UInt64) return DriverDataType.UInt64; + if (dataType == DataTypeIds.Float) return DriverDataType.Float32; + if (dataType == DataTypeIds.Double) return DriverDataType.Float64; + if (dataType == DataTypeIds.String) return DriverDataType.String; + if (dataType == DataTypeIds.DateTime || dataType == DataTypeIds.UtcTime) + return DriverDataType.DateTime; + return DriverDataType.String; + } + + /// + /// Map an OPC UA AccessLevel/UserAccessLevel attribute value (AccessLevels bitmask) to a + /// : CurrentWrite-capable variables surface as + /// ; read-only as . + /// + /// The OPC UA access level bitmask. + /// The mapped security classification. + public static SecurityClassification MapAccessLevel(byte accessLevel) + { + const byte CurrentWrite = 2; // AccessLevels.CurrentWrite = 0x02 + return (accessLevel & CurrentWrite) != 0 + ? SecurityClassification.Operate + : SecurityClassification.ViewOnly; + } + + /// + /// OPC UA Part 3 ValueRank: -3 ScalarOrOneDimension, -2 Any, -1 Scalar, 0 OneOrMoreDimensions, 1 OneDimension, + /// >1 N dimensions. Anything >= 0 is an array; -3 / -2 are conservatively scalar until the upstream variable + /// carries a concrete dimensioned rank. + /// + /// The ValueRank attribute. + /// Whether the variable is exposed as an array. + public static bool IsArray(int valueRank) => valueRank >= 0; +} diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Session/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Session.csproj b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Session/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Session.csproj index 3ecc9663..41166a40 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Session/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Session.csproj +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Session/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Session.csproj @@ -11,8 +11,9 @@ + +