From e6406c494c7af8edad4a6694afe534333300db6d Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 06:37:23 -0400 Subject: [PATCH] fix(dcl): async OPC UA endpoint discovery bounded by OperationTimeoutMs/verify budget; adapter connect never runs on the actor thread (review 2026-10-21 02/P2) --- .../Component-DataConnectionLayer.md | 4 +- .../Actors/DataConnectionActor.cs | 9 +- .../Adapters/RealOpcUaClient.cs | 50 +++++++---- .../Actors/DataConnectionHousekeepingTests.cs | 36 ++++++++ .../Adapters/OpcUaSecurityDowngradeTests.cs | 14 +-- .../Adapters/RealOpcUaDiscoveryBoundTests.cs | 85 +++++++++++++++++++ 6 files changed, 170 insertions(+), 28 deletions(-) create mode 100644 tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Adapters/RealOpcUaDiscoveryBoundTests.cs diff --git a/docs/requirements/Component-DataConnectionLayer.md b/docs/requirements/Component-DataConnectionLayer.md index 6790d300..719b2b6b 100644 --- a/docs/requirements/Component-DataConnectionLayer.md +++ b/docs/requirements/Component-DataConnectionLayer.md @@ -71,7 +71,7 @@ All protocols produce the same value tuple consumed by Instance Actors. Before t ### OPC UA - Uses the **OPC Foundation .NET Standard Library** (`OPCFoundation.NetStandard.Opc.Ua.Client`). -- Session-based connection with endpoint discovery, certificate handling, and configurable security modes. +- Session-based connection with endpoint discovery, certificate handling, and configurable security modes. Endpoint discovery is **asynchronous** and bounded by the connection's `OperationTimeoutMs` (the discovery client is built from the connection's own `EndpointConfiguration`) and by the caller's cancellation; a cancelled discovery is never mistaken for "discovery unavailable". The `DataConnectionActor` starts every adapter `ConnectAsync` on the thread pool, so no adapter connect work — whatever runs before its first yielding await — can occupy the actor's dispatcher thread (review 2026-10-21; previously a synchronous discovery against a black-holed endpoint held the thread for the SDK's 120 s default on every reconnect tick). - Subscriptions via OPC UA Monitored Items with data change notifications (1000ms sampling, queue size 10, discard-oldest). - Read/Write via OPC UA Read/Write services with StatusCode-based quality mapping. - Disconnect detection via `Session.KeepAlive` event (see Disconnect Detection Pattern below). @@ -282,7 +282,7 @@ IAddressSpaceSearchable Before saving or deploying an OPC UA data connection, the Central UI can ask the target site to **probe** the configured endpoint — this exercises connectivity (and TLS trust) against the live config, including edited-but-unsaved values. -- `DataConnectionManagerActor` handles `VerifyEndpointCommand` (`SiteId`, protocol, config JSON). The site spins up a **temporary** `RealOpcUaClient` from the submitted config, attempts discovery + a session with a short timeout (a few seconds), then disconnects and disposes the client. The reply is a `VerifyEndpointResult` (`Success`, a typed `FailureKind`, an error message, and an optional captured server cert). +- `DataConnectionManagerActor` handles `VerifyEndpointCommand` (`SiteId`, protocol, config JSON). The site spins up a **temporary** `RealOpcUaClient` from the submitted config, attempts discovery + a session with a short timeout (a few seconds — the budget bounds discovery too, not only the session create), then disconnects and disposes the client. The reply is a `VerifyEndpointResult` (`Success`, a typed `FailureKind`, an error message, and an optional captured server cert). - The probe forces **`AutoAcceptUntrustedCerts = false`** and hooks the certificate-validation callback so it can **capture** an untrusted server certificate (Subject / Issuer / Thumbprint / NotBefore / NotAfter / DER) for the UI to display. Critically, the probe **never trusts** the cert — the validation callback always rejects (`Accept = false`); trusting is a separate, explicit, Admin-gated action that writes the cert into the site PKI store (see the cert-trust path in Component-SiteRuntime.md). The probe is read-only with respect to the trust boundary. ## Native Alarm Mirroring diff --git a/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionActor.cs b/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionActor.cs index 87f9dfa2..54fd1376 100644 --- a/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionActor.cs @@ -957,7 +957,14 @@ public class DataConnectionActor : UntypedActor, IWithTimers // is still in flight swaps the adapter and bumps the counter, and the OUTGOING // adapter's answer must never be applied to its replacement. var generation = _adapterGeneration; - _adapter.ConnectAsync(_connectionDetails).ContinueWith(t => + // Started on the thread pool, never inline: everything an adapter's ConnectAsync does + // before its first YIELDING await would otherwise run on this actor's dispatcher thread + // — OPC UA endpoint discovery was synchronous there and parked the thread for the SDK's + // operation timeout on every reconnect tick to a black-holed endpoint (review 2026-10-21 + // 02/P2). Adapter and details are captured here, on the actor thread. + var adapter = _adapter; + var details = _connectionDetails; + Task.Run(() => adapter.ConnectAsync(details)).ContinueWith(t => { if (t.IsCompletedSuccessfully) return new ConnectResult(true, null, generation); diff --git a/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Adapters/RealOpcUaClient.cs b/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Adapters/RealOpcUaClient.cs index edd12df9..b51aa2d8 100644 --- a/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Adapters/RealOpcUaClient.cs +++ b/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Adapters/RealOpcUaClient.cs @@ -96,19 +96,42 @@ public class RealOpcUaClient : IOpcUaClient /// public bool IsConnected => _session?.Connected ?? false; + /// + /// Asynchronous endpoint discovery bounded by 's + /// TransportQuotas.OperationTimeout (the connection's OperationTimeoutMs) and by + /// (review 2026-10-21 02/P2). It replaces the synchronous + /// DiscoveryClient.Create(uri).GetEndpoints(null), which ran on whatever thread called + /// the connect — the actor thread, when ValidateAsync completed synchronously — under + /// the SDK's default 120 s operation timeout, and could not be cancelled by a verify probe's + /// budget. + /// + private static async Task DiscoverEndpointsAsync( + ApplicationConfiguration appConfig, string endpointUrl, CancellationToken ct) + { + var discoveryConfig = EndpointConfiguration.Create(appConfig); + using var discoveryClient = await DiscoveryClient.CreateAsync( + appConfig, new Uri(endpointUrl), discoveryConfig, DiagnosticsMasks.None, ct).ConfigureAwait(false); + return await discoveryClient.GetEndpointsAsync(null, ct).ConfigureAwait(false); + } + /// Raised when the OPC UA connection is lost. public event Action? ConnectionLost; /// Selects exactly the requested protection mode; protected discovery failures fail closed. /// The discovery URL to query. /// The security mode that must be matched exactly. - /// Delegate that returns the server's advertised endpoints. + /// Delegate that returns the server's advertised endpoints, honouring the token. + /// + /// Bounds the discovery. A cancellation is rethrown — never mistaken for "discovery + /// unavailable", which would otherwise fall back to an unsecured endpoint and carry on. + /// /// The endpoint matching the requested security mode. - internal static EndpointDescription DiscoverEndpoint(string url, MessageSecurityMode requested, - Func discover) + internal static async Task DiscoverEndpointAsync(string url, MessageSecurityMode requested, + Func> discover, CancellationToken ct = default) { EndpointDescriptionCollection endpoints; - try { endpoints = discover(); } + try { endpoints = await discover(ct).ConfigureAwait(false); } + catch (OperationCanceledException) when (ct.IsCancellationRequested) { throw; } catch (Exception ex) { if (requested != MessageSecurityMode.None) @@ -163,13 +186,8 @@ public class RealOpcUaClient : IOpcUaClient } // Discover endpoints from the server, pick the preferred security mode - var endpoint = DiscoverEndpoint(endpointUrl, preferredSecurityMode, () => - { -#pragma warning disable CS0618 - using var discoveryClient = DiscoveryClient.Create(new Uri(endpointUrl)); - return discoveryClient.GetEndpoints(null); -#pragma warning restore CS0618 - }); + var endpoint = await DiscoverEndpointAsync(endpointUrl, preferredSecurityMode, + token => DiscoverEndpointsAsync(appConfig, endpointUrl, token), cancellationToken); // The server advertises its own base address, which is frequently unroutable from here // (a 0.0.0.0 wildcard bind, or an internal container/NAT hostname). Swap it back to the @@ -391,13 +409,9 @@ public class RealOpcUaClient : IOpcUaClient await appConfig.ValidateAsync(ApplicationType.Client); // Discover endpoints, pick the preferred security mode (same logic as ConnectAsync). - var endpoint = DiscoverEndpoint(endpointUrl, preferredSecurityMode, () => - { -#pragma warning disable CS0618 - using var discoveryClient = DiscoveryClient.Create(new Uri(endpointUrl)); - return discoveryClient.GetEndpoints(null); -#pragma warning restore CS0618 - }); + // Bounded by the probe's own budget (linkedCts), not just the SDK operation timeout. + var endpoint = await DiscoverEndpointAsync(endpointUrl, preferredSecurityMode, + token => DiscoverEndpointsAsync(appConfig, endpointUrl, token), linkedCts.Token); // Same reachability rewrite as ConnectAsync — a probe against a server advertising a // 0.0.0.0 / NAT base address must dial the reachable host, not the advertised one. diff --git a/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Actors/DataConnectionHousekeepingTests.cs b/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Actors/DataConnectionHousekeepingTests.cs index b36bc97b..de660f30 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Actors/DataConnectionHousekeepingTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Actors/DataConnectionHousekeepingTests.cs @@ -126,4 +126,40 @@ public class DataConnectionHousekeepingTests : TestKit && message.Contains("established", StringComparison.OrdinalIgnoreCase)), TimeSpan.FromSeconds(5)); } + + /// + /// 02/P2: an adapter whose connect does synchronous work before its first yielding await + /// (OPC UA discovery used to) must not park the actor's dispatcher thread — the actor keeps + /// answering while the connect is in progress. + /// + [Fact] + public async Task AdapterConnect_NeverRunsOnTheActorThread() + { + const string name = "connect-off-actor-thread"; + var adapter = Substitute.For(); + using var release = new ManualResetEventSlim(false); + adapter.ConnectAsync(Arg.Any>(), Arg.Any()) + .Returns(_ => + { + release.Wait(TimeSpan.FromSeconds(10)); // synchronous I/O before any await + return Task.CompletedTask; + }); + adapter.Status.Returns(ConnectionHealth.Connecting); + + var actor = Sys.ActorOf(Props.Create(() => new DataConnectionActor( + name, adapter, Options(), _health, _factory, "OpcUa")), name); + AwaitCondition(() => adapter.ReceivedCalls().Any(c => c.GetMethodInfo().Name == nameof(IDataConnection.ConnectAsync)), + TimeSpan.FromSeconds(5)); + + try + { + var report = await actor.Ask( + new DataConnectionActor.GetHealthReport(), TimeSpan.FromSeconds(2)); + Assert.Equal(name, report.ConnectionName); + } + finally + { + release.Set(); + } + } } diff --git a/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Adapters/OpcUaSecurityDowngradeTests.cs b/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Adapters/OpcUaSecurityDowngradeTests.cs index ff7e0009..f3f23699 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Adapters/OpcUaSecurityDowngradeTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Adapters/OpcUaSecurityDowngradeTests.cs @@ -8,19 +8,19 @@ public class OpcUaSecurityDowngradeTests [Theory] [InlineData(false)] [InlineData(true)] - public void DC4_RequestedProtectionNeverFallsBackToNone(bool discoveryFails) + public async Task DC4_RequestedProtectionNeverFallsBackToNone(bool discoveryFails) { - Assert.Throws(() => RealOpcUaClient.DiscoverEndpoint("opc.tcp://test:4840", - MessageSecurityMode.SignAndEncrypt, () => discoveryFails + await Assert.ThrowsAsync(() => RealOpcUaClient.DiscoverEndpointAsync("opc.tcp://test:4840", + MessageSecurityMode.SignAndEncrypt, _ => discoveryFails ? throw new Exception("discovery unavailable") - : new EndpointDescriptionCollection { new("opc.tcp://test:4840") { SecurityMode = MessageSecurityMode.None } })); + : Task.FromResult(new EndpointDescriptionCollection { new("opc.tcp://test:4840") { SecurityMode = MessageSecurityMode.None } }))); } /// Verifies that DC4: exact requested mode is selected. [Fact] - public void DC4_ExactRequestedModeIsSelected() + public async Task DC4_ExactRequestedModeIsSelected() { var encrypted = new EndpointDescription("opc.tcp://test:4840") { SecurityMode = MessageSecurityMode.SignAndEncrypt }; - Assert.Same(encrypted, RealOpcUaClient.DiscoverEndpoint("opc.tcp://test:4840", MessageSecurityMode.SignAndEncrypt, - () => new EndpointDescriptionCollection { new("opc.tcp://test:4840"), encrypted })); + Assert.Same(encrypted, await RealOpcUaClient.DiscoverEndpointAsync("opc.tcp://test:4840", MessageSecurityMode.SignAndEncrypt, + _ => Task.FromResult(new EndpointDescriptionCollection { new("opc.tcp://test:4840"), encrypted }))); } } diff --git a/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Adapters/RealOpcUaDiscoveryBoundTests.cs b/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Adapters/RealOpcUaDiscoveryBoundTests.cs new file mode 100644 index 00000000..94ca76b6 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Adapters/RealOpcUaDiscoveryBoundTests.cs @@ -0,0 +1,85 @@ +using System.Diagnostics; +using System.Net; +using System.Net.Sockets; +using Microsoft.Extensions.Logging.Abstractions; +using ZB.MOM.WW.ScadaBridge.Commons.Types.DataConnections; +using ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Adapters; + +namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests.Adapters; + +/// +/// Review 2026-10-21 02/P2: OPC UA endpoint discovery ran SYNCHRONOUSLY on a discovery client +/// built with the SDK's default configuration, so neither the connection's +/// OperationTimeoutMs nor the verify probe's budget bounded it — a black-holed endpoint held +/// the caller for the SDK's default operation timeout instead. +/// +public class RealOpcUaDiscoveryBoundTests +{ + /// + /// A listener that accepts the TCP connection and then never speaks — the protocol-level + /// black hole. The verify probe must give up within (about) its own budget. + /// + [Fact] + public async Task Verify_AgainstABlackHoledEndpoint_ReturnsWithinItsBudget() + { + var listener = new TcpListener(IPAddress.Loopback, 0); + listener.Start(); + var held = new List(); + using var stop = new CancellationTokenSource(); + _ = Task.Run(async () => + { + try + { + while (!stop.IsCancellationRequested) + held.Add(await listener.AcceptTcpClientAsync(stop.Token)); + } + catch (OperationCanceledException) { } + catch (ObjectDisposedException) { } + }); + + try + { + var port = ((IPEndPoint)listener.LocalEndpoint).Port; + var config = new OpcUaEndpointConfig + { + EndpointUrl = $"opc.tcp://127.0.0.1:{port}", + OperationTimeoutMs = 60_000, + }; + + var sw = Stopwatch.StartNew(); + var verify = RealOpcUaClient.VerifyEndpointAsync( + config, new OpcUaGlobalOptions(), NullLogger.Instance, TimeSpan.FromSeconds(2), CancellationToken.None); + var result = await verify.WaitAsync(TimeSpan.FromSeconds(20)); + sw.Stop(); + + Assert.False(result.Success); + Assert.True(sw.Elapsed < TimeSpan.FromSeconds(10), $"verify took {sw.Elapsed}"); + } + finally + { + stop.Cancel(); + listener.Stop(); + foreach (var c in held) c.Dispose(); + } + } + + /// + /// Cancellation of discovery is never mistaken for "discovery unavailable": the unsecured + /// fallback endpoint is only for a genuine discovery failure, not for an abandoned attempt. + /// + [Fact] + public async Task CancelledDiscovery_IsNotTreatedAsADiscoveryFailure() + { + using var cts = new CancellationTokenSource(); + cts.Cancel(); + + await Assert.ThrowsAnyAsync(() => RealOpcUaClient.DiscoverEndpointAsync( + "opc.tcp://test:4840", Opc.Ua.MessageSecurityMode.None, + async ct => + { + await Task.Delay(Timeout.Infinite, ct); + return new Opc.Ua.EndpointDescriptionCollection(); + }, + cts.Token)); + } +}