From 97cc145811bc2d9cef11e685392696435b10df36 Mon Sep 17 00:00:00 2001 From: Wolfgang Colsman <52833340+wolfgangcolsman@users.noreply.github.com> Date: Fri, 23 Oct 2026 03:58:57 +0000 Subject: [PATCH] Client redundancy: ServiceLevel-driven failover, member ServiceLevel probing and read statuses (#4884) # Description ## Problem OPC 10000-4, clause 6.6.2 "Server redundancy", lets a Client of a non-transparent redundant Server set (Cold, Warm, Hot, HotAndMirrored) connect to the Server with the highest `ServiceLevel` and fail over when the Server in use is lost or its `ServiceLevel` leaves the Healthy subrange (6.6.2.4.2 "ServiceLevel": 200-255 Healthy, 2-199 Degraded, 1 NoData, 0 Maintenance; 6.6.2.4.5 "Client failover behaviour"). The managed session on `master` covers part of this: - `ManagedSession` fails over only in the `Failover` state of `ConnectionStateMachine`, which is entered after the reconnect policy gives up. `DefaultServerRedundancyHandler.ShouldFailover` / `ServerRedundancyHandlerExtensions.ShouldFailover` implement the subrange rules, but nothing in the client calls them: a connected session stays on a Server that reports Degraded or Maintenance until its connection breaks. There is no public member that starts a failover (`HandleFailoverAsync` is private). - Members known only from `ServerUriArray` get `ServiceLevel` NoData with `ServiceLevelKnown = false` (internal). Their `ServiceLevel` is never read, so neither the initial connect nor `SelectFailoverTarget` can prefer the member with the highest `ServiceLevel`. - `ReadRedundancyInfoAsync` maps a bad status of `RedundancySupport`, `ServerUriArray`, `RedundantServerArray` or `CurrentServerId` to an empty or default value, and peer endpoint resolution failures are only logged. An application cannot report which redundancy Node the Server refused or why a member has no endpoint. An application that must fail over on `ServiceLevel` therefore polls the `ServiceLevel` itself, opens its own sessions to read each member, and, lacking a failover trigger, closes the managed session and opens a new one - losing the transfer or recreate of its subscriptions that the managed failover performs. ## API (all additions; defaults keep the current behaviour) ```csharp public partial class ManagedSession { public Task FailoverAsync(ConfiguredEndpoint? target = null, CancellationToken ct = default); public ServerRedundancyInfo? RedundancyInfo { get; } // implements IManagedSession.RedundancyInfo } public interface IManagedSession : ISession { ServerRedundancyInfo? RedundancyInfo { get; } // new interface member } public sealed record ServerRedundancyOptions { public bool ProbeMemberServiceLevel { get; init; } // default false public TimeSpan? ServiceLevelMonitoringInterval { get; init; } // default null (off); must be positive when set } public sealed class RedundantServer { public bool ServiceLevelKnown { get; init; } = true; // was internal public ServiceResult? ResolutionError { get; init; } // new } public sealed class ServerRedundancyInfo { public IReadOnlyDictionary ReadStatuses { get; init; } // new, empty by default } ``` ## Behaviour 1. **Read statuses and resolution errors.** `DefaultServerRedundancyHandler` records the status of every Node it reads from the Server in use (`RedundancySupport`, `ServiceLevel`, `RedundantServerArray`, `ServerUriArray`, `EstimatedReturnTime`, and `CurrentServerId` for Transparent redundancy) in `ServerRedundancyInfo.ReadStatuses`. A peer whose endpoint resolution fails carries the exact status in `RedundantServer.ResolutionError` (the discovery `ServiceResult`, `BadTimeout` when the discovery bound expires, `BadNotFound` when no endpoint matches the security and transport in use). `IManagedSession.RedundancyInfo` exposes the last snapshot. 2. **Member probing** (`ProbeMemberServiceLevel`). After a member's endpoint is resolved, `ManagedSession` reads `Server.ServiceLevel` and `Server.ServerStatus.State` of every other member over a short session with the identity of the session in use, bounded by `PeerDiscoveryTimeout`, and replaces the reported values. A member that cannot be probed keeps the values reported for it and carries the failure in `ResolutionError` (for example `BadNotReadable`). With probing, the initial connect switches to the member with the highest `ServiceLevel` when it is higher than that of the Server it connected to (6.6.2.4.5). 3. **`FailoverAsync`.** Fails over a connected session to `target`, or to the target the redundancy handler selects from a refreshed snapshot, through the existing in-place recreate (subscriptions are transferred or recreated; with token reuse enabled a HotAndMirrored target reactivates the session). The connection state machine gets an internal `RequestFailover` that enters `Failover` with the requested operation; `ConnectionStateChanged` reports `Failover` then `Connected`. Throws `BadNotSupported` without a redundancy handler, `BadInvalidState` when not connected, `BadNothingToDo` when the handler selects no target, and the failover error otherwise. A failed requested failover transitions to `Reconnecting` (not `Disconnected` plus the recovery timer), so the reconnect policy restores the session at once. Cancelling `ct` stops waiting; a failover in progress completes. 4. **ServiceLevel monitoring** (`ServiceLevelMonitoringInterval`). While connected to a non-transparent set, `ManagedSession` reads the `ServiceLevel` of the Server in use at the interval. While it is Healthy nothing else happens (a Healthy Server is not left, 6.6.2.4.3). Otherwise the snapshot is refreshed (with probing when enabled), `ShouldFailover` decides, and a warranted failover runs `FailoverAsync` to the handler's target. The loop ends with the session. The switch logic of the automatic failover (`HandleFailoverAsync`) is factored into a helper shared with the initial selection and `FailoverAsync`; its behaviour is unchanged. ## Tests - `ManagedSessionServerRedundancyTests` (new, in-process Warm set of two `ReferenceServer` fixtures with distinct ApplicationUris, `Server.ServerRedundancy` promoted to `NonTransparentRedundancyType` with each other in `ServerUriArray`, `ServiceLevel` set through `IServerServiceLevelControl`, endpoints resolved by a test `IRedundantServerEndpointResolver`): - without probing the session stays on the configured Server; the peer has `ServiceLevelKnown = false` (defaults unchanged); - with probing the initial connect selects the member with the higher `ServiceLevel` (250 over 210); - a member that refuses the `ServiceLevel` read keeps an unknown `ServiceLevel` and reports `BadNotReadable` in `ResolutionError`; - a refused `RedundantServerArray` reports `BadNotReadable` in `ReadStatuses`; - lowering the `ServiceLevel` of the Server in use to 150 while the peer reports 230 fails over within the monitoring interval, `ConnectionStateChanged` reports only `Failover` and `Connected`, the same subscription object keeps delivering data changes from the new Server (sequence numbers restart there, as the subscription is recreated on another Server); - `FailoverAsync` to an explicit target and to the handler-selected target; - `FailoverAsync` without a redundancy handler throws `BadNotSupported`. - `ServerRedundancyHandlerTests`: resolution error of a refusing and of an unresolved peer, read status of each Node, validation of the monitoring interval. - `ConnectionStateMachineTests`: a requested failover runs its operation in `Failover` and bypasses the failover delegate; a failed requested failover reconnects without `Disconnected`; `RequestFailover` is refused when not connected. ## Compatibility - Source/binary additions only, except that `IManagedSession` gains `RedundancyInfo`; `ManagedSession` is the only implementation in the repository, external implementers of `IManagedSession` must add the property. - `RedundantServer.ServiceLevelKnown` changes from internal to public (same default `true`). - With the default options nothing changes at runtime: no probing, no monitoring, the automatic failover path is unchanged. The only default-visible differences are the new `ReadStatuses`/`ResolutionError` data. - Probing opens one short additional session per member; reverse-connect-only members cannot be probed and report the probe failure in `ResolutionError`. ## Related Issues - No tracking issue; the problem and the design are described above. ## Checklist - [x] I have signed the [CLA](https://opcfoundation.org/license/cla/ContributorLicenseAgreementv1.0.pdf) and read the [CONTRIBUTING](https://github.com/OPCFoundation/UA-.NETStandard/blob/master/CONTRIBUTING.md) doc. - [x] I have added tests that prove my fix is effective or that my feature works and increased code coverage. - [ ] I have added all necessary documentation. - [x] I have verified that my changes do not introduce (new) build or analyzer warnings. - [ ] I ran **all** tests locally using the **UA.slnx** solution against at least .net **framework** and .net **10**, and all passed. (Run locally: the affected test classes on net48 and net10.0 and the whole Opc.Ua.Client.Tests project on net10.0; not the whole UA.slnx.) - [ ] I fixed **all** failing and flaky tests in the CI pipelines and **all** CodeQL warnings. - [ ] I have addressed **all** PR feedback received. --------- Co-authored-by: Marc Schier --- src/Opc.Ua.Client/Session/IManagedSession.cs | 8 + .../Session/ManagedSession.Redundancy.cs | 554 ++++++++++++ src/Opc.Ua.Client/Session/ManagedSession.cs | 116 ++- .../Reconnect/ConnectionStateMachine.cs | 86 +- .../DefaultRedundantServerEndpointResolver.cs | 54 +- .../DefaultServerRedundancyHandler.cs | 74 +- .../Session/Redundancy/RedundantServer.cs | 17 +- .../Redundancy/ServerRedundancyInfo.cs | 12 + .../Redundancy/ServerRedundancyOptions.cs | 40 +- .../Session/Session.ChannelManager.cs | 91 ++ src/Opc.Ua.Client/Session/Session.cs | 84 +- .../Session/ConnectionStateMachineTests.cs | 99 +++ .../ManagedSessionServerRedundancyTests.cs | 835 ++++++++++++++++++ .../Session/ServerRedundancyHandlerTests.cs | 137 +++ .../Session/SessionTests.cs | 3 +- 15 files changed, 2149 insertions(+), 61 deletions(-) create mode 100644 src/Opc.Ua.Client/Session/ManagedSession.Redundancy.cs create mode 100644 tests/Opc.Ua.Client.Tests/Session/ManagedSessionServerRedundancyTests.cs diff --git a/src/Opc.Ua.Client/Session/IManagedSession.cs b/src/Opc.Ua.Client/Session/IManagedSession.cs index 0e6b8eb3a5..049d4cc65d 100644 --- a/src/Opc.Ua.Client/Session/IManagedSession.cs +++ b/src/Opc.Ua.Client/Session/IManagedSession.cs @@ -44,5 +44,13 @@ namespace Opc.Ua.Client /// when the server is not redundant. /// RedundancySupport RedundancySupport { get; } + + /// + /// The server redundancy snapshot of the last refresh: the redundancy mode, the ServiceLevel of + /// the Server in use, the members of the redundant Server set with their resolved endpoints and + /// errors, and the status of each redundancy Node read. without a redundancy + /// handler or before the first refresh. + /// + ServerRedundancyInfo? RedundancyInfo { get; } } } diff --git a/src/Opc.Ua.Client/Session/ManagedSession.Redundancy.cs b/src/Opc.Ua.Client/Session/ManagedSession.Redundancy.cs new file mode 100644 index 0000000000..c38807db93 --- /dev/null +++ b/src/Opc.Ua.Client/Session/ManagedSession.Redundancy.cs @@ -0,0 +1,554 @@ +/* ======================================================================== + * Copyright (c) 2005-2026 The OPC Foundation, Inc. All rights reserved. + * + * OPC Foundation MIT License 1.00 + * + * Permission is hereby granted, free of charge, to any person + * obtaining a copy of this software and associated documentation + * files (the "Software"), to deal in the Software without + * restriction, including without limitation the rights to use, + * copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the + * Software is furnished to do so, subject to the following + * conditions: + * + * The above copyright notice and this permission notice shall be + * included in all copies or substantial portions of the Software. + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, + * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES + * OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND + * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT + * HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, + * WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING + * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR + * OTHER DEALINGS IN THE SOFTWARE. + * + * The complete license agreement can be found here: + * http://opcfoundation.org/License/MIT/1.00/ + * ======================================================================*/ + +using System; +using System.IO; +using System.Net.Sockets; +using System.Threading; +using System.Threading.Tasks; + +namespace Opc.Ua.Client +{ + /// + /// Client side failover across the members of a non-transparent redundant + /// Server set (OPC 10000-4 §6.6.2). + /// + public partial class ManagedSession + { + /// + /// Fails over now to another member of the non-transparent redundant + /// Server set, with the same in-place recreate as a failover after a + /// lost connection: subscriptions are transferred or recreated on the + /// target and, with token reuse enabled, a HotAndMirrored target + /// reactivates the existing session. + /// reports and then + /// . + /// + /// + /// When the failover fails the session transitions to + /// and the reconnect policy + /// restores it. Cancelling stops waiting; a + /// failover in progress completes. + /// + /// + /// The endpoint to fail over to, or to fail over + /// to the target the redundancy handler selects from a refreshed + /// redundancy snapshot. + /// + /// Cancellation token. + /// + /// when no redundancy handler + /// is configured, when the + /// session is not connected, + /// when the handler selects no target, or the error of the failed + /// failover. + /// + public async Task FailoverAsync(ConfiguredEndpoint? target = null, CancellationToken ct = default) + { + ct.ThrowIfCancellationRequested(); + IServerRedundancyHandler handler = m_redundancyHandler ?? throw new ServiceResultException( + StatusCodes.BadNotSupported, "A failover requires a server redundancy handler."); + if (!StateMachine.IsConnected) + { + throw new ServiceResultException( + StatusCodes.BadInvalidState, "The managed session can only fail over while it is connected."); + } + + target ??= await SelectFailoverTargetAsync(handler, ct).ConfigureAwait(false) ?? + throw new ServiceResultException( + StatusCodes.BadNothingToDo, "The redundancy handler selected no failover target."); + await RunRequestedFailoverAsync(target, ct).ConfigureAwait(false); + } + + private async Task SelectFailoverTargetAsync( + IServerRedundancyHandler handler, + CancellationToken ct) + { + await RefreshRedundancyInfoBestEffortAsync(ct).ConfigureAwait(false); + await RefreshRedundancyInfoBestEffortAsync(ct, resolveCachedEndpoints: true).ConfigureAwait(false); + ServerRedundancyInfo? info = m_redundancyInfo; + return info == null ? null : handler.SelectFailoverTarget(info, ConfiguredEndpoint); + } + + /// + /// Runs a failover to in the Failover state of + /// the state machine and waits until the session is connected again. + /// + private async Task RunRequestedFailoverAsync(ConfiguredEndpoint target, CancellationToken ct) + { + var outcome = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + if (!StateMachine.RequestFailover( + (budget, token) => FailoverToRequestedTargetAsync(target, budget, token), + outcome)) + { + throw new ServiceResultException( + StatusCodes.BadInvalidState, "The managed session can only fail over while it is connected."); + } + + using (var waiting = CancellationTokenSource.CreateLinkedTokenSource(ct)) + { + Task closed = StateMachine.WaitForClosedAsync(waiting.Token).AsTask(); + try + { + if (await Task.WhenAny(outcome.Task, closed).ConfigureAwait(false) != outcome.Task || + outcome.Task.IsCanceled) + { + ct.ThrowIfCancellationRequested(); + throw new ServiceResultException( + StatusCodes.BadSessionClosed, "The managed session was closed during the failover."); + } + } + finally + { + // Releases the close wait, which otherwise lasts until the session closes. + await waiting.CancelAsync().ConfigureAwait(false); + } + } + + ServiceResult result = await outcome.Task.ConfigureAwait(false); + if (ServiceResult.IsBad(result)) + { + throw new ServiceResultException(result); + } + await StateMachine.WaitForConnectedAsync(ct).ConfigureAwait(false); + } + + private async Task FailoverToRequestedTargetAsync( + ConfiguredEndpoint target, + IRetryBudget budget, + CancellationToken ct) + { + try + { + m_logger.ManagedSessionFailingOverEndpoint(target.EndpointUrl); + ServiceResult result = await SwitchToRedundantServerAsync( + target, budget, retirePreviousSession: true, ct).ConfigureAwait(false); + if (ServiceResult.IsGood(result)) + { + await RefreshRedundancyInfoBestEffortAsync(ct).ConfigureAwait(false); + m_logger.ManagedSessionFailoverComplete(); + } + return result; + } + catch (Exception exception) when ( + !(exception is OperationCanceledException && ct.IsCancellationRequested)) + { + if (m_redundancyHandler is IServerRedundancyEndpointCache cache) + { + cache.InvalidateEndpoint(target); + } + m_logger.ManagedSessionFailoverFailed(exception); + return ToAttemptFailure(exception); + } + } + + /// + /// Reads the ServiceLevel of the Server in use at + /// and fails over when the redundancy + /// handler decides that a failover is warranted. + /// + private async ValueTask MonitorServiceLevelAsync(TimeSpan interval, CancellationToken shutdown) + { + while (StateMachine.State != ConnectionState.Closed) + { + await m_timeProvider.Delay(interval, shutdown).ConfigureAwait(false); + if (StateMachine.State != ConnectionState.Connected) + { + continue; + } + try + { + await FailOverOnServiceLevelAsync(shutdown).ConfigureAwait(false); + } + catch (OperationCanceledException) when (shutdown.IsCancellationRequested) + { + return; + } + catch (Exception exception) when (exception is not OutOfMemoryException) + { + m_logger.ManagedSessionServiceLevelMonitoringFailed(exception); + } + } + } + + private async Task FailOverOnServiceLevelAsync(CancellationToken ct) + { + IServerRedundancyHandler? handler = m_redundancyHandler; + if (handler == null) + { + return; + } + ServerRedundancyInfo? info = m_redundancyInfo; + if (info == null) + { + await RefreshRedundancyInfoBestEffortAsync(ct).ConfigureAwait(false); + info = m_redundancyInfo; + } + if (info == null || + info.Mode is RedundancySupport.None or RedundancySupport.Transparent) + { + return; + } + + // A Healthy Server is never left (OPC 10000-4 §6.6.2.4.3), so only its + // ServiceLevel is read; the redundant set is refreshed, resolved and + // probed once the Server in use is no longer Healthy. + if (await IsServiceLevelHealthyAsync(ct).ConfigureAwait(false)) + { + return; + } + + await RefreshRedundancyInfoBestEffortAsync(ct).ConfigureAwait(false); + info = m_redundancyInfo; + if (info == null || + info.Mode is RedundancySupport.None or RedundancySupport.Transparent || + (info.ServiceLevelAccessible && ServiceLevels.IsHealthy(info.ServiceLevel))) + { + return; + } + + await RefreshRedundancyInfoBestEffortAsync(ct, resolveCachedEndpoints: true).ConfigureAwait(false); + info = m_redundancyInfo!; + ConfiguredEndpoint current = ConfiguredEndpoint; + ServerFailoverDecision decision = handler.ShouldFailover(info, current); + ConfiguredEndpoint? target = decision.IsFailoverWarranted + ? handler.SelectFailoverTarget(info, current) + : null; + if (target == null || StateMachine.State != ConnectionState.Connected) + { + return; + } + + m_logger.ManagedSessionServiceLevelFailover(info.ServiceLevel, decision.Reason); + await RunRequestedFailoverAsync(target, ct).ConfigureAwait(false); + } + + /// + /// Reads the ServiceLevel of the Server in use. + /// + /// + /// when the read succeeds and the value is in the + /// Healthy subrange. + /// + private async Task IsServiceLevelHealthyAsync(CancellationToken ct) + { + try + { + DataValue value = await this.ReadValueAsync(VariableIds.Server_ServiceLevel, ct) + .ConfigureAwait(false); + return StatusCode.IsGood(value.StatusCode) && + value.WrappedValue.TryGetValue(out byte level) && + ServiceLevels.IsHealthy(level); + } + catch (ServiceResultException) + { + return false; + } + } + + /// + /// Switches the session to by recreating it in + /// place, so subscriptions are transferred or recreated on the target. + /// With , a session newly created + /// on the target closes the previous session on the Server in use. + /// + private async Task SwitchToRedundantServerAsync( + ConfiguredEndpoint target, + IRetryBudget budget, + bool retirePreviousSession, + CancellationToken ct) + { + using (await RecoveryWriterLockAsync(ct) + .ConfigureAwait(false)) + { + Session? session = m_session; + if (session == null) + { + return new ServiceResult( + StatusCodes.BadInvalidState); + } + + // Recreate in place: the inner Session reference is + // preserved across failover so that the V2 + // SubscriptionEngine, SubscriptionManager, and any + // external holders of InnerSession all continue to + // operate on the same Session object. Session + // internals re-run CreateSession+ActivateSession + // against the new endpoint and drive subscription + // recreate/transfer for both unamanged templates and + // the new engine. + using IDisposable recovery = session.DeferSubscriptionRecovery(); + await RecreateInPlaceAndRebindAsync( + session, + target, + budget, + ct, + retirePreviousSession) + .ConfigureAwait(false); + } + + m_reconnectPolicy.Reset(); + return ServiceResult.Good; + } + + /// + /// After the initial connect, switches to the member of the redundant + /// Server set with the highest probed ServiceLevel when it is + /// higher than the ServiceLevel of the Server connected to. + /// + private async Task SelectInitialRedundantServerAsync(CancellationToken ct) + { + ServerRedundancyInfo? info = m_redundancyInfo; + if (info == null || info.Mode is RedundancySupport.None or RedundancySupport.Transparent) + { + return; + } + + try + { + await RefreshRedundancyInfoBestEffortAsync(ct, resolveCachedEndpoints: true) + .ConfigureAwait(false); + } + catch (Exception exception) when ( + exception is not OutOfMemoryException && + !(exception is OperationCanceledException && ct.IsCancellationRequested)) + { + m_logger.ManagedSessionRedundancyDiscoveryFailed(exception); + return; + } + + info = m_redundancyInfo; + RedundantServer? best = info == null ? null : SelectHigherServiceLevelMember(info, ConfiguredEndpoint); + if (best?.Endpoint == null) + { + return; + } + + m_logger.ManagedSessionSelectingRedundantServer( + best.Endpoint.EndpointUrl, + best.ServiceLevel, + info!.ServiceLevel); + ServiceResult result = await SwitchToRedundantServerAsync( + best.Endpoint, CreateRecoveryBudget(), retirePreviousSession: true, ct) + .ConfigureAwait(false); + if (ServiceResult.IsBad(result)) + { + throw new ServiceResultException(result); + } + await RefreshRedundancyInfoBestEffortAsync(ct).ConfigureAwait(false); + } + + /// + /// Selects the running member with a read ServiceLevel that is + /// higher than the ServiceLevel of the Server in use. + /// + private static RedundantServer? SelectHigherServiceLevelMember( + ServerRedundancyInfo info, + ConfiguredEndpoint current) + { + string? currentUri = current.Description?.Server?.ApplicationUri; + byte currentLevel = info.ServiceLevelAccessible ? info.ServiceLevel : ServiceLevels.Maintenance; + RedundantServer? best = null; + for (int ii = 0; ii < info.RedundantServers.Count; ii++) + { + RedundantServer server = info.RedundantServers[ii]; + if (server.Endpoint != null && + server.ServiceLevelKnown && + server.ResolutionError == null && + server.ServerState == ServerState.Running && + server.ServiceLevel > currentLevel && + !string.Equals(server.ServerUri, currentUri, StringComparison.Ordinal) && + (best == null || server.ServiceLevel > best.ServiceLevel)) + { + best = server; + } + } + return best; + } + + /// + /// Reads ServiceLevel and ServerStatus.State of every + /// other member with a resolved endpoint. + /// + private async Task ProbeMemberServiceLevelsAsync( + ServerRedundancyInfo snapshot, + ConfiguredEndpoint current, + CancellationToken ct) + { + if (snapshot.Mode is RedundancySupport.None or RedundancySupport.Transparent || + snapshot.RedundantServers.Count == 0) + { + return snapshot; + } + + string? currentUri = current.Description?.Server?.ApplicationUri; + var probes = new Task[snapshot.RedundantServers.Count]; + for (int ii = 0; ii < probes.Length; ii++) + { + probes[ii] = ProbeMemberAsync(snapshot.RedundantServers[ii], current, currentUri, ct); + } + RedundantServer[] servers = await Task.WhenAll(probes).ConfigureAwait(false); + return new ServerRedundancyInfo + { + Mode = snapshot.Mode, + RedundantServers = servers, + ServiceLevel = snapshot.ServiceLevel, + ServiceLevelAccessible = snapshot.ServiceLevelAccessible, + ServiceLevelSubrange = snapshot.ServiceLevelSubrange, + EstimatedReturnTime = snapshot.EstimatedReturnTime, + CurrentServerId = snapshot.CurrentServerId, + ReadStatuses = snapshot.ReadStatuses + }; + } + + private async Task ProbeMemberAsync( + RedundantServer server, + ConfiguredEndpoint current, + string? currentUri, + CancellationToken ct) + { + ConfiguredEndpoint? endpoint = server.Endpoint; + if (endpoint == null || + string.IsNullOrEmpty(server.ServerUri) || + string.Equals(server.ServerUri, currentUri, StringComparison.Ordinal)) + { + return server; + } + + // The probe authenticates with the identity in use, so it must not + // weaken the security the session is bound to. + ServiceResultException? mismatch = GetSecurityConfigurationMismatch( + current.Description, endpoint.Description); + if (mismatch != null) + { + m_logger.ManagedSessionMemberServiceLevelProbeFailed(mismatch, endpoint.EndpointUrl); + return WithProbeError(server, mismatch.Result); + } + + IUserIdentity? identity = m_session?.Identity ?? m_identity; + using CancellationTokenSource timeout = m_timeProvider.CreateCancellationTokenSource(m_peerProbeTimeout); + using var linked = CancellationTokenSource.CreateLinkedTokenSource(ct, timeout.Token); + ISession? probe = null; + try + { + probe = await SessionFactory.CreateAsync( + m_configuration, + DetachEndpoint(endpoint), + false, + m_checkDomain, + m_sessionName, + m_sessionTimeout, + identity == null ? null : new UserIdentity(identity.TokenHandler), + m_preferredLocales, + linked.Token).ConfigureAwait(false); + (ArrayOf values, ArrayOf errors) = await probe.ReadValuesAsync( + [VariableIds.Server_ServiceLevel, VariableIds.Server_ServerStatus_State], + linked.Token).ConfigureAwait(false); + for (int ii = 0; ii < values.Count; ii++) + { + if (StatusCode.IsBad(errors[ii].StatusCode) || StatusCode.IsBad(values[ii].StatusCode)) + { + return WithProbeError(server, StatusCode.IsBad(errors[ii].StatusCode) + ? errors[ii] + : new ServiceResult(values[ii].StatusCode)); + } + } + return new RedundantServer + { + ServerUri = server.ServerUri, + ServiceLevel = values[0].GetValue(0), + ServiceLevelKnown = true, + ServerState = (ServerState)values[1].GetValue((int)server.ServerState), + Endpoint = endpoint + }; + } + catch (OperationCanceledException) when (ct.IsCancellationRequested) + { + throw; + } + catch (Exception exception) when (exception is ServiceResultException or TimeoutException + or OperationCanceledException or IOException or SocketException) + { + m_logger.ManagedSessionMemberServiceLevelProbeFailed(exception, endpoint.EndpointUrl); + return WithProbeError(server, exception switch + { + ServiceResultException sre => sre.Result, + IOException or SocketException => new ServiceResult(exception), + _ => new ServiceResult(StatusCodes.BadTimeout, new LocalizedText(exception.Message)) + }); + } + finally + { + if (probe != null) + { + await CloseProbeAsync(probe).ConfigureAwait(false); + } + } + } + + private async Task CloseProbeAsync(ISession probe) + { + try + { + int closeTimeout = m_peerProbeTimeout == Timeout.InfiniteTimeSpan + ? 0 + : (int)Math.Min(m_peerProbeTimeout.TotalMilliseconds, int.MaxValue); + await probe.CloseAsync(closeTimeout, true, CancellationToken.None).ConfigureAwait(false); + } + catch (Exception exception) when (exception is not OutOfMemoryException) + { + m_logger.ManagedSessionSessionCloseFailed(exception); + } + finally + { + probe.Dispose(); + } + } + + private static RedundantServer WithProbeError(RedundantServer server, ServiceResult error) + { + return new RedundantServer + { + ServerUri = server.ServerUri, + ServiceLevel = server.ServiceLevel, + ServiceLevelKnown = server.ServiceLevelKnown, + ServerState = server.ServerState, + Endpoint = server.Endpoint, + ResolutionError = error + }; + } + + private RetryBudget CreateRecoveryBudget() + { + return new RetryBudget( + m_maxTotalReconnectTime == TimeSpan.Zero ? Timeout.InfiniteTimeSpan : m_maxTotalReconnectTime, + m_timeProvider); + } + } +} diff --git a/src/Opc.Ua.Client/Session/ManagedSession.cs b/src/Opc.Ua.Client/Session/ManagedSession.cs index bf83f763d3..ceb1c1c222 100644 --- a/src/Opc.Ua.Client/Session/ManagedSession.cs +++ b/src/Opc.Ua.Client/Session/ManagedSession.cs @@ -102,6 +102,8 @@ namespace Opc.Ua.Client ServerRedundancyOptions redundancyOptions = serverRedundancy ?? new ServerRedundancyOptions(); redundancyOptions.Validate(); m_redundancyRefreshTimeout = redundancyOptions.RefreshTimeout; + m_peerProbeTimeout = redundancyOptions.PeerDiscoveryTimeout; + m_probeMemberServiceLevel = redundancyOptions.ProbeMemberServiceLevel; m_logger = logger ?? throw new ArgumentNullException(nameof(logger)); m_identity = identity; @@ -135,6 +137,13 @@ namespace Opc.Ua.Client WireStateMachineCallbacks(); SubscribeCertificateChanges(); + if (redundancyHandler != null && + redundancyOptions.ServiceLevelMonitoringInterval is TimeSpan monitoringInterval) + { + m_backgroundWork.Run( + "MonitorServiceLevel", + shutdown => MonitorServiceLevelAsync(monitoringInterval, shutdown)); + } } /// @@ -583,6 +592,9 @@ namespace Opc.Ua.Client /// public RedundancySupport RedundancySupport => m_redundancyInfo?.Mode ?? RedundancySupport.None; + /// + public ServerRedundancyInfo? RedundancyInfo => Volatile.Read(ref m_redundancyInfo); + /// public string SessionName => m_session?.SessionName ?? m_sessionName; @@ -1616,6 +1628,10 @@ namespace Opc.Ua.Client } await RefreshRedundancyInfoBestEffortAsync(ct).ConfigureAwait(false); + if (m_probeMemberServiceLevel) + { + await SelectInitialRedundantServerAsync(ct).ConfigureAwait(false); + } } catch { @@ -2338,36 +2354,14 @@ namespace Opc.Ua.Client m_logger.ManagedSessionFailingOverEndpoint(failoverEndpoint.EndpointUrl); - using (await RecoveryWriterLockAsync(ct) - .ConfigureAwait(false)) + ServiceResult switched = await SwitchToRedundantServerAsync( + failoverEndpoint, budget, retirePreviousSession: false, ct) + .ConfigureAwait(false); + if (ServiceResult.IsBad(switched)) { - Session? session = m_session; - if (session == null) - { - return new ServiceResult( - StatusCodes.BadInvalidState); - } - - // Recreate in place: the inner Session reference is - // preserved across failover so that the V2 - // SubscriptionEngine, SubscriptionManager, and any - // external holders of InnerSession all continue to - // operate on the same Session object. Session - // internals re-run CreateSession+ActivateSession - // against the new endpoint and drive subscription - // recreate/transfer for both unamanged templates and - // the new engine. - using IDisposable recovery = session.DeferSubscriptionRecovery(); - await RecreateInPlaceAndRebindAsync( - session, - failoverEndpoint, - budget, - ct) - .ConfigureAwait(false); + return switched; } - m_reconnectPolicy.Reset(); - m_logger.ManagedSessionFailoverComplete(); return ServiceResult.Good; @@ -2400,6 +2394,12 @@ namespace Opc.Ua.Client ServerRedundancyInfo? snapshot = m_redundancyInfo; if (handler is not IServerRedundancyEndpointCache cache || snapshot == null) { + if (m_probeMemberServiceLevel && snapshot != null) + { + ServerRedundancyInfo probed = await ProbeMemberServiceLevelsAsync( + snapshot, ConfiguredEndpoint, ct).ConfigureAwait(false); + Interlocked.CompareExchange(ref m_redundancyInfo, probed, snapshot); + } return; } if (m_redundancyEndpointRefreshTask is { IsCompleted: false } pending) @@ -2414,7 +2414,8 @@ namespace Opc.Ua.Client } } snapshot = Volatile.Read(ref m_redundancyInfo) ?? snapshot; - await StartRedundancyEndpointRefresh(cache, snapshot, ct).WaitAsync(ct).ConfigureAwait(false); + await StartRedundancyEndpointRefresh(cache, snapshot, m_probeMemberServiceLevel, ct) + .WaitAsync(ct).ConfigureAwait(false); return; } @@ -2438,7 +2439,7 @@ namespace Opc.Ua.Client if (handler is IServerRedundancyEndpointCache cache && m_redundancyInfo.Mode is not (RedundancySupport.None or RedundancySupport.Transparent)) { - _ = StartRedundancyEndpointRefresh(cache, m_redundancyInfo, ct); + _ = StartRedundancyEndpointRefresh(cache, m_redundancyInfo, probeMembers: false, ct); } } catch (Exception exception) when (exception is OperationCanceledException or TimeoutException) @@ -2456,9 +2457,16 @@ namespace Opc.Ua.Client } } + /// + /// Resolves the endpoints of the members in the background and, with + /// , reads their ServiceLevel. Probing + /// only runs for a selection or failover decision, never for a + /// routine refresh. + /// private Task StartRedundancyEndpointRefresh( IServerRedundancyEndpointCache cache, ServerRedundancyInfo snapshot, + bool probeMembers, CancellationToken ct) { if (m_redundancyEndpointRefreshTask is { IsCompleted: false } previous) @@ -2476,6 +2484,11 @@ namespace Opc.Ua.Client { ServerRedundancyInfo resolved = await cache.ResolveCachedEndpointsAsync( snapshot, current, linked.Token).ConfigureAwait(false); + if (probeMembers) + { + resolved = await ProbeMemberServiceLevelsAsync(resolved, current, linked.Token) + .ConfigureAwait(false); + } linked.Token.ThrowIfCancellationRequested(); Interlocked.CompareExchange(ref m_redundancyInfo, resolved, snapshot); completion.TrySetResult(resolved); @@ -2587,13 +2600,16 @@ namespace Opc.Ua.Client /// /// Recreates the session in place, against a copy of /// when set, which must keep the security - /// mode and policy of the endpoint the session works on. + /// mode and policy of the endpoint the session works on. With + /// , a newly created session + /// closes the session it replaces on the previous Server. /// private async Task RecreateInPlaceAndRebindAsync( Session session, ConfiguredEndpoint? endpoint, IRetryBudget budget, - CancellationToken ct) + CancellationToken ct, + bool retirePreviousSession = false) { if (endpoint == null) { @@ -2605,7 +2621,7 @@ namespace Opc.Ua.Client await SwitchEndpointAsync( session, endpoint, - target => RecreateInPlaceCoreAndRebindAsync(session, target, budget, ct), + target => RecreateInPlaceCoreAndRebindAsync(session, target, budget, ct, retirePreviousSession), ct).ConfigureAwait(false); } @@ -2619,14 +2635,15 @@ namespace Opc.Ua.Client Session session, ConfiguredEndpoint? endpoint, IRetryBudget budget, - CancellationToken ct) + CancellationToken ct, + bool retirePreviousSession = false) { IManagedTransportChannel? previousChannel = session.ManagedChannel; try { ITransportWaitingConnection? connection = await WaitForRecoveryConnectionAsync( endpoint ?? session.ConfiguredEndpoint, ct).ConfigureAwait(false); - await session.RecreateInPlaceAsync(endpoint, budget, connection, ct) + await session.RecreateInPlaceAsync(endpoint, budget, connection, ct, retirePreviousSession) .ConfigureAwait(false); } finally @@ -3459,6 +3476,8 @@ namespace Opc.Ua.Client private Task? m_redundancyRefreshTask; private Task? m_redundancyEndpointRefreshTask; private readonly TimeSpan m_redundancyRefreshTimeout; + private readonly TimeSpan m_peerProbeTimeout; + private readonly bool m_probeMemberServiceLevel; private readonly Lock m_identityRefreshLock = new(); #pragma warning disable CA2213 // Owned and disposed by StopIdentityRefreshLoopAsync. @@ -3607,5 +3626,34 @@ namespace Opc.Ua.Client this ILogger logger, Exception? exception, [NetworkAddress] Uri? endpointUrl); + + [LoggerMessage(EventId = ClientEventIds.ManagedSession + 32, Level = LogLevel.Information, + Message = "ManagedSession: Switching to redundant Server {Endpoint} with ServiceLevel {ServiceLevel}; " + + "the Server in use reports ServiceLevel {CurrentServiceLevel}.")] + public static partial void ManagedSessionSelectingRedundantServer( + this ILogger logger, + [NetworkAddress] Uri? endpoint, + byte serviceLevel, + byte currentServiceLevel); + + [LoggerMessage(EventId = ClientEventIds.ManagedSession + 33, Level = LogLevel.Warning, + Message = "ManagedSession: Reading the ServiceLevel of redundant Server {Endpoint} failed.")] + public static partial void ManagedSessionMemberServiceLevelProbeFailed( + this ILogger logger, + Exception? exception, + [NetworkAddress] Uri? endpoint); + + [LoggerMessage(EventId = ClientEventIds.ManagedSession + 34, Level = LogLevel.Information, + Message = "ManagedSession: The Server in use reports ServiceLevel {ServiceLevel}; failing over: {Reason}")] + public static partial void ManagedSessionServiceLevelFailover( + this ILogger logger, + byte serviceLevel, + string reason); + + [LoggerMessage(EventId = ClientEventIds.ManagedSession + 35, Level = LogLevel.Warning, + Message = "ManagedSession: ServiceLevel monitoring could not complete a failover.")] + public static partial void ManagedSessionServiceLevelMonitoringFailed( + this ILogger logger, + Exception? exception); } } diff --git a/src/Opc.Ua.Client/Session/Reconnect/ConnectionStateMachine.cs b/src/Opc.Ua.Client/Session/Reconnect/ConnectionStateMachine.cs index 2cad6b93d0..628d714e90 100644 --- a/src/Opc.Ua.Client/Session/Reconnect/ConnectionStateMachine.cs +++ b/src/Opc.Ua.Client/Session/Reconnect/ConnectionStateMachine.cs @@ -114,6 +114,8 @@ namespace Opc.Ua.Client private bool m_hasConnected; private readonly ITimer m_recoveryTimer; private ConnectionStateBudgetOperation? m_requestedReconnect; + private ConnectionStateBudgetOperation? m_requestedFailover; + private TaskCompletionSource? m_requestedFailoverOutcome; private readonly Queue m_stateChanges = new(); private bool m_dispatchingStateChanges; @@ -380,6 +382,49 @@ namespace Opc.Ua.Client return true; } + /// + /// Starts a failover of the connected session that runs + /// in the + /// state. When the operation fails the machine transitions to + /// , so the reconnect policy + /// restores the session. + /// + /// The failover operation. + /// + /// Completed with the result of the failover including the subscription + /// recovery that follows , or cancelled when + /// the failover does not run to completion. + /// + /// when the session is not connected. + internal bool RequestFailover( + ConnectionStateBudgetOperation operation, + TaskCompletionSource? outcome = null) + { + lock (m_lock) + { + if (m_worker == null || + m_disposed != 0 || + m_state != ConnectionState.Connected) + { + return false; + } + + m_requestedFailover = operation; + m_requestedFailoverOutcome = outcome; + m_lastError = null; + ClearReconnectBudget(); + m_settled.Reset(); + TransitionTo( + ConnectionState.Failover, + error: null, + reconnectAttempt: 0); + } + + m_trigger.Set(); + DispatchStateChanges(); + return true; + } + /// /// Request closing the session. /// @@ -392,6 +437,9 @@ namespace Opc.Ua.Client { m_recoveryTimer.Change(Timeout.InfiniteTimeSpan, Timeout.InfiniteTimeSpan); m_requestedReconnect = null; + m_requestedFailover = null; + m_requestedFailoverOutcome?.TrySetCanceled(); + m_requestedFailoverOutcome = null; TransitionTo( ConnectionState.Closing, error: null, @@ -911,21 +959,38 @@ namespace Opc.Ua.Client IRetryBudget budget = GetOrCreateReconnectBudget(); + ConnectionStateBudgetOperation? requested; + TaskCompletionSource? outcome; + lock (m_lock) + { + requested = m_requestedFailover; + outcome = m_requestedFailoverOutcome; + m_requestedFailover = null; + m_requestedFailoverOutcome = null; + } + ServiceResult result; using (CancellationTokenSource closeAware = CreateCloseAwareTokenSource(ct)) { try { - result = await InvokeFailoverAsync(budget, closeAware.Token) + result = await InvokeFailoverAsync(requested, budget, closeAware.Token) .ConfigureAwait(false); } catch (OperationCanceledException) when (!ct.IsCancellationRequested) { // A close was requested while failing over: return so the // worker loop handles the Closing state right away. + outcome?.TrySetCanceled(CancellationToken.None); return; } + catch (OperationCanceledException) + { + outcome?.TrySetCanceled(ct); + throw; + } } + outcome?.TrySetResult(result); lock (m_lock) { @@ -946,6 +1011,19 @@ namespace Opc.Ua.Client reconnectAttempt: 0); m_settled.Set(); } + else if (requested != null) + { + // A requested failover leaves a session the reconnect policy can + // restore; it must not wait for the recovery timer. + ClearReconnectBudget(); + m_logger.ConnectionStateMachineFailoverFailedError(result); + m_lastError = result; + TransitionTo( + ConnectionState.Reconnecting, + error: result, + reconnectAttempt: 0); + m_trigger.Set(); + } else { ClearReconnectBudget(); @@ -1089,11 +1167,17 @@ namespace Opc.Ua.Client /// Safely invoke the failover delegate. /// private async Task InvokeFailoverAsync( + ConnectionStateBudgetOperation? requested, IRetryBudget budget, CancellationToken ct) { try { + if (requested != null) + { + ServiceResult result = await requested(budget, ct).ConfigureAwait(false); + return await CompleteRecoveryAfterHandshakeAsync(result, ct).ConfigureAwait(false); + } if (FailoverWithBudgetAsync != null) { ServiceResult result = await FailoverWithBudgetAsync(budget, ct).ConfigureAwait(false); diff --git a/src/Opc.Ua.Client/Session/Redundancy/DefaultRedundantServerEndpointResolver.cs b/src/Opc.Ua.Client/Session/Redundancy/DefaultRedundantServerEndpointResolver.cs index d121c85e1c..5e6546f9fb 100644 --- a/src/Opc.Ua.Client/Session/Redundancy/DefaultRedundantServerEndpointResolver.cs +++ b/src/Opc.Ua.Client/Session/Redundancy/DefaultRedundantServerEndpointResolver.cs @@ -78,6 +78,24 @@ namespace Opc.Ua.Client string serverUri, ConfiguredEndpoint currentEndpoint, CancellationToken ct = default) + { + (ConfiguredEndpoint? endpoint, _) = await ResolveWithDiscoveryErrorAsync(serverUri, currentEndpoint, ct) + .ConfigureAwait(false); + return endpoint; + } + + /// + /// Resolves like and also returns the + /// first discovery failure when no endpoint was resolved, so a caller can tell a failed + /// discovery from a Server without a matching endpoint. + /// + /// is null or empty. + /// is null. + internal async ValueTask<(ConfiguredEndpoint? Endpoint, ServiceResult? DiscoveryError)> + ResolveWithDiscoveryErrorAsync( + string serverUri, + ConfiguredEndpoint currentEndpoint, + CancellationToken ct = default) { if (string.IsNullOrEmpty(serverUri)) { @@ -90,13 +108,14 @@ namespace Opc.Ua.Client } ct.ThrowIfCancellationRequested(); + var failure = new DiscoveryFailure(); if (m_discoveredServers.TryGetValue(serverUri, out ApplicationDescription? cached)) { - ConfiguredEndpoint? resolved = await ResolveServerEndpointsAsync(cached, currentEndpoint, ct) - .ConfigureAwait(false); + ConfiguredEndpoint? resolved = await ResolveServerEndpointsAsync( + cached, currentEndpoint, failure, ct).ConfigureAwait(false); if (resolved != null) { - return resolved; + return (resolved, null); } } foreach (string discoveryUrl in GetDiscoveryUrls(currentEndpoint)) @@ -108,6 +127,7 @@ namespace Opc.Ua.Client serverUri, currentEndpoint, discoveryUrl, + failure, ct).ConfigureAwait(false); } catch (OperationCanceledException) when (ct.IsCancellationRequested) @@ -119,21 +139,23 @@ namespace Opc.Ua.Client { ct.ThrowIfCancellationRequested(); m_logger.ManagedSessionRedundancyDiscoveryFailed(exception); + failure.Record(exception); continue; } if (endpoint != null) { - return endpoint; + return (endpoint, null); } } - return null; + return (null, failure.First); } private async ValueTask ResolveWithDiscoveryUrlAsync( string serverUri, ConfiguredEndpoint currentEndpoint, string discoveryUrl, + DiscoveryFailure failure, CancellationToken ct) { Uri discoveryUri = CoreClientUtils.GetDiscoveryUrl(discoveryUrl); @@ -162,12 +184,13 @@ namespace Opc.Ua.Client // Keep the address mapping even when an offline peer cannot yet answer GetEndpoints. server = CoreUtils.Clone(server)!; m_discoveredServers[serverUri] = server; - return await ResolveServerEndpointsAsync(server, currentEndpoint, ct).ConfigureAwait(false); + return await ResolveServerEndpointsAsync(server, currentEndpoint, failure, ct).ConfigureAwait(false); } private async ValueTask ResolveServerEndpointsAsync( ApplicationDescription server, ConfiguredEndpoint currentEndpoint, + DiscoveryFailure failure, CancellationToken ct) { string serverUri = server.ApplicationUri @@ -200,6 +223,7 @@ namespace Opc.Ua.Client { ct.ThrowIfCancellationRequested(); m_logger.ManagedSessionRedundancyDiscoveryFailed(exception); + failure.Record(exception); } } @@ -284,6 +308,24 @@ namespace Opc.Ua.Client private readonly ConcurrentDictionary m_discoveredServers = new(StringComparer.Ordinal); + + /// + /// The first discovery failure of one resolution. + /// + private sealed class DiscoveryFailure + { + public ServiceResult? First { get; private set; } + + public void Record(Exception exception) + { + First ??= exception switch + { + ServiceResultException sre => sre.Result, + TimeoutException => new ServiceResult(StatusCodes.BadTimeout, new LocalizedText(exception.Message)), + _ => new ServiceResult(exception) + }; + } + } } /// diff --git a/src/Opc.Ua.Client/Session/Redundancy/DefaultServerRedundancyHandler.cs b/src/Opc.Ua.Client/Session/Redundancy/DefaultServerRedundancyHandler.cs index 5a8b6e103b..b78eafd011 100644 --- a/src/Opc.Ua.Client/Session/Redundancy/DefaultServerRedundancyHandler.cs +++ b/src/Opc.Ua.Client/Session/Redundancy/DefaultServerRedundancyHandler.cs @@ -135,6 +135,12 @@ namespace Opc.Ua.Client await session.ReadValuesAsync(nodeIds, ct).ConfigureAwait(false); ct.ThrowIfCancellationRequested(); + var readStatuses = new Dictionary(nodeIds.Count + 1); + for (int ii = 0; ii < nodeIds.Count; ii++) + { + readStatuses[nodeIds[ii]] = GetReadStatus(values[ii], errors[ii]); + } + RedundancySupport mode = RedundancySupport.None; if (StatusCode.IsGood(errors[0].StatusCode)) { @@ -165,7 +171,7 @@ namespace Opc.Ua.Client values[3], errors[3]); currentServerId = mode == RedundancySupport.Transparent - ? await ReadCurrentServerIdAsync(session, ct).ConfigureAwait(false) + ? await ReadCurrentServerIdAsync(session, readStatuses, ct).ConfigureAwait(false) : string.Empty; redundantServers = redundantServers.ConvertAll(server => WithEndpoint( server, m_resolvedEndpoints.TryGetValue( @@ -184,7 +190,8 @@ namespace Opc.Ua.Client ServiceLevelSubrange = ServiceLevels.GetSubrange(serviceLevel), EstimatedReturnTime = estimatedReturnTime, CurrentServerId = currentServerId, - RedundantServers = redundantServers + RedundantServers = redundantServers, + ReadStatuses = readStatuses }; } @@ -297,7 +304,8 @@ namespace Opc.Ua.Client ServiceLevelAccessible = snapshot.ServiceLevelAccessible, ServiceLevelSubrange = snapshot.ServiceLevelSubrange, EstimatedReturnTime = snapshot.EstimatedReturnTime, - CurrentServerId = snapshot.CurrentServerId + CurrentServerId = snapshot.CurrentServerId, + ReadStatuses = snapshot.ReadStatuses }; } @@ -336,6 +344,7 @@ namespace Opc.Ua.Client { ct.ThrowIfCancellationRequested(); ConfiguredEndpoint? endpoint = null; + ServiceResult? error = null; if (!string.IsNullOrEmpty(server.ServerUri)) { using CancellationTokenSource timeout = m_timeProvider.CreateCancellationTokenSource( @@ -348,6 +357,13 @@ namespace Opc.Ua.Client // below would unhook it from the caller's token and leave the discovery running. endpoint = await ResolveEndpointAsync(server.ServerUri, currentEndpoint, linked.Token) .WaitAsync(m_peerDiscoveryTimeout, m_timeProvider, linked.Token).ConfigureAwait(false); + if (endpoint == null) + { + error = new ServiceResult( + StatusCodes.BadNotFound, + new LocalizedText( + "No endpoint of the Server matches the security and transport of the endpoint in use.")); + } } catch (OperationCanceledException) when (ct.IsCancellationRequested) { @@ -358,6 +374,7 @@ namespace Opc.Ua.Client { ct.ThrowIfCancellationRequested(); m_logger.ManagedSessionRedundancyDiscoveryFailed(exception); + error = ToServiceResult(exception); } finally { @@ -365,10 +382,13 @@ namespace Opc.Ua.Client await linked.CancelAsync().ConfigureAwait(false); } } - return WithEndpoint(server, endpoint); + return WithEndpoint(server, endpoint, error); } - private static RedundantServer WithEndpoint(RedundantServer server, ConfiguredEndpoint? endpoint) + private static RedundantServer WithEndpoint( + RedundantServer server, + ConfiguredEndpoint? endpoint, + ServiceResult? resolutionError = null) { return new RedundantServer { @@ -376,10 +396,27 @@ namespace Opc.Ua.Client ServiceLevel = server.ServiceLevel, ServiceLevelKnown = server.ServiceLevelKnown, ServerState = server.ServerState, - Endpoint = endpoint + Endpoint = endpoint, + ResolutionError = resolutionError }; } + private static ServiceResult ToServiceResult(Exception exception) + { + return exception switch + { + ServiceResultException sre => sre.Result, + TimeoutException or OperationCanceledException => new ServiceResult( + StatusCodes.BadTimeout, new LocalizedText(exception.Message)), + _ => new ServiceResult(exception) + }; + } + + private static StatusCode GetReadStatus(DataValue value, ServiceResult error) + { + return StatusCode.IsBad(error.StatusCode) ? error.StatusCode : value.StatusCode; + } + private async Task ResolveEndpointAsync( string serverUri, ConfiguredEndpoint currentEndpoint, @@ -419,8 +456,22 @@ namespace Opc.Ua.Client { try { - ConfiguredEndpoint? endpoint = await m_endpointResolver - .ResolveAsync(key.ServerUri, currentEndpoint, ct).ConfigureAwait(false); + ConfiguredEndpoint? endpoint; + if (m_endpointResolver is DefaultRedundantServerEndpointResolver resolver) + { + (endpoint, ServiceResult? discoveryError) = await resolver + .ResolveWithDiscoveryErrorAsync(key.ServerUri, currentEndpoint, ct).ConfigureAwait(false); + if (endpoint == null && discoveryError != null) + { + // A failed discovery is reported as such, not as a missing endpoint. + throw new ServiceResultException(discoveryError); + } + } + else + { + endpoint = await m_endpointResolver + .ResolveAsync(key.ServerUri, currentEndpoint, ct).ConfigureAwait(false); + } ct.ThrowIfCancellationRequested(); if (endpoint != null) { @@ -533,6 +584,7 @@ namespace Opc.Ua.Client private static async Task ReadCurrentServerIdAsync( ISession session, + Dictionary readStatuses, CancellationToken ct) { try @@ -540,6 +592,7 @@ namespace Opc.Ua.Client DataValue dataValue = await session.ReadValueAsync( VariableIds.Server_ServerRedundancy_CurrentServerId, ct).ConfigureAwait(false); + readStatuses[VariableIds.Server_ServerRedundancy_CurrentServerId] = dataValue.StatusCode; if (StatusCode.IsGood(dataValue.StatusCode) && dataValue.WrappedValue.TryGetValue(out string currentServerId)) @@ -547,9 +600,10 @@ namespace Opc.Ua.Client return currentServerId; } } - catch (ServiceResultException) + catch (ServiceResultException sre) { - // Node may not exist; return empty string. + // The node may not exist; the status reports why the identifier is empty. + readStatuses[VariableIds.Server_ServerRedundancy_CurrentServerId] = sre.StatusCode; } return string.Empty; diff --git a/src/Opc.Ua.Client/Session/Redundancy/RedundantServer.cs b/src/Opc.Ua.Client/Session/Redundancy/RedundantServer.cs index 8988d18833..bafb767d2b 100644 --- a/src/Opc.Ua.Client/Session/Redundancy/RedundantServer.cs +++ b/src/Opc.Ua.Client/Session/Redundancy/RedundantServer.cs @@ -49,9 +49,16 @@ namespace Opc.Ua.Client public byte ServiceLevel { get; init; } /// - /// Gets a value indicating whether was read from the Server. + /// Gets a value indicating whether was reported for the Server, either in + /// the RedundantServerArray of the Server in use or by reading the Server's own + /// ServiceLevel. /// - internal bool ServiceLevelKnown { get; init; } = true; + /// + /// A Server known only from ServerUriArray reports and + /// until its ServiceLevel is read, see + /// . + /// + public bool ServiceLevelKnown { get; init; } = true; /// /// The current state of the Server. @@ -62,5 +69,11 @@ namespace Opc.Ua.Client /// The resolved endpoint for this redundant server. /// public ConfiguredEndpoint? Endpoint { get; init; } + + /// + /// The error that prevented resolving or reading the Server's + /// ServiceLevel during the last refresh, or when no attempt failed. + /// + public ServiceResult? ResolutionError { get; init; } } } diff --git a/src/Opc.Ua.Client/Session/Redundancy/ServerRedundancyInfo.cs b/src/Opc.Ua.Client/Session/Redundancy/ServerRedundancyInfo.cs index 41b6d8fba7..6456652345 100644 --- a/src/Opc.Ua.Client/Session/Redundancy/ServerRedundancyInfo.cs +++ b/src/Opc.Ua.Client/Session/Redundancy/ServerRedundancyInfo.cs @@ -28,6 +28,8 @@ * ======================================================================*/ using System; +using System.Collections.Generic; +using System.Collections.ObjectModel; namespace Opc.Ua.Client { @@ -74,5 +76,15 @@ namespace Opc.Ua.Client /// The current Server identifier for Transparent redundancy. /// public string CurrentServerId { get; init; } = string.Empty; + + /// + /// The status of each Node read from the Server in use to build this snapshot, keyed by its NodeId + /// (for example Server.ServerRedundancy.ServerUriArray). A bad status explains why the + /// corresponding value is empty or defaulted. + /// + public IReadOnlyDictionary ReadStatuses { get; init; } = s_noReadStatuses; + + private static readonly IReadOnlyDictionary s_noReadStatuses = + new ReadOnlyDictionary(new Dictionary()); } } diff --git a/src/Opc.Ua.Client/Session/Redundancy/ServerRedundancyOptions.cs b/src/Opc.Ua.Client/Session/Redundancy/ServerRedundancyOptions.cs index ff5f245354..6add4148ae 100644 --- a/src/Opc.Ua.Client/Session/Redundancy/ServerRedundancyOptions.cs +++ b/src/Opc.Ua.Client/Session/Redundancy/ServerRedundancyOptions.cs @@ -65,15 +65,51 @@ namespace Opc.Ua.Client public TimeSpan PeerDiscoveryTimeout { get; init; } = DefaultTimeout; /// - /// Validates that both bounds are usable. + /// When , a reads Server.ServiceLevel and + /// Server.ServerStatus.State of every other member of a non-transparent redundant Server set + /// over a short session with a copy of the identity of the session in use, when a selection or + /// failover decision needs them: after the initial connect, for a failover, and during ServiceLevel + /// monitoring once the Server in use is no longer Healthy. A member whose resolved endpoint does not + /// keep the security mode and policy of the session in use is not probed. The values replace those + /// reported by the Server in use, so the initial connect, the + /// failover target selection and the failover decision compare the members' current + /// ServiceLevel (OPC 10000-4 §6.6.2.4.5). After the initial connect the session switches to + /// the member with the highest ServiceLevel when it is higher than the ServiceLevel of + /// the Server it connected to. A member that cannot be probed keeps the values reported for it and + /// carries the failure in . Each probe is bounded by + /// . Default . + /// + public bool ProbeMemberServiceLevel { get; init; } + + /// + /// When set, a reads Server.ServiceLevel of the Server in use at + /// this interval while it is connected to a non-transparent redundant Server set. When the + /// ServiceLevel leaves the Healthy subrange (OPC 10000-4 §6.6.2.4.2) the redundancy snapshot is + /// refreshed, including the member probe when is set, and a + /// failover decision that warrants it starts to the target + /// the redundancy handler selects. Default : the session fails over only when + /// the reconnect policy gives up or the application calls . + /// + public TimeSpan? ServiceLevelMonitoringInterval { get; init; } + + /// + /// Validates that both bounds and the monitoring interval are usable. /// /// - /// Thrown when a bound is neither positive nor . + /// Thrown when a bound is neither positive nor , or when + /// is set and not positive. /// public void Validate() { ValidateTimeout(RefreshTimeout, nameof(RefreshTimeout)); ValidateTimeout(PeerDiscoveryTimeout, nameof(PeerDiscoveryTimeout)); + if (ServiceLevelMonitoringInterval is TimeSpan interval && interval <= TimeSpan.Zero) + { + throw new ArgumentOutOfRangeException( + nameof(ServiceLevelMonitoringInterval), + interval, + "The ServiceLevel monitoring interval must be positive."); + } } private static void ValidateTimeout(TimeSpan value, string name) diff --git a/src/Opc.Ua.Client/Session/Session.ChannelManager.cs b/src/Opc.Ua.Client/Session/Session.ChannelManager.cs index f6d356862b..fa0062293c 100644 --- a/src/Opc.Ua.Client/Session/Session.ChannelManager.cs +++ b/src/Opc.Ua.Client/Session/Session.ChannelManager.cs @@ -397,6 +397,96 @@ namespace Opc.Ua.Client private sealed record PendingSubscriptionRecovery(NodeId PreviousSessionId, bool ReusedSession); + /// + /// A session this client replaced on another channel, still allocated on its Server. + /// + private sealed record RetiredServerSession( + NodeId SessionId, + NodeId AuthenticationToken, + ITransportChannel Channel); + + /// + /// Closes once no subscription recovery is pending, so + /// subscriptions can still be transferred from it. + /// + private async Task RetireAfterSubscriptionRecoveryAsync(RetiredServerSession retired) + { + RetiredServerSession? replaced = Interlocked.Exchange(ref m_retiredSession, retired); + if (replaced != null) + { + await RetireServerSessionAsync(replaced, disposeChannel: true).ConfigureAwait(false); + } + if (m_pendingSubscriptionRecovery == null) + { + await RetirePendingServerSessionAsync().ConfigureAwait(false); + } + } + + private async Task RetirePendingServerSessionAsync() + { + RetiredServerSession? retired = Interlocked.Exchange(ref m_retiredSession, null); + if (retired != null) + { + await RetireServerSessionAsync(retired, disposeChannel: true).ConfigureAwait(false); + } + } + + /// + /// Sends CloseSession for on the channel it is bound to + /// and deletes its subscriptions there. + /// + private async Task RetireServerSessionAsync(RetiredServerSession retired, bool disposeChannel) + { + try + { + var request = new CloseSessionRequest + { + RequestHeader = new RequestHeader + { + AuthenticationToken = retired.AuthenticationToken, + RequestHandle = NewRequestHandle(), + Timestamp = DateTime.UtcNow, + TimeoutHint = (uint)GetCloseRequestTimeout(m_keepAliveInterval) + }, + DeleteSubscriptions = true + }; + using CancellationTokenSource timeout = + CreateCloseRequestCancellationTokenSource(m_keepAliveInterval, CancellationToken.None); + await retired.Channel.SendRequestAsync(request, timeout.Token).ConfigureAwait(false); + m_logger.SessionRetiredPreviousSession(retired.SessionId); + } + catch (Exception ex) when (ex is not OutOfMemoryException) + { + m_logger.CleanupCloseSessionAsyncCloseChannelAsyncRaisedException(ex); + } + finally + { + if (disposeChannel) + { + await CloseRetainedChannelAsync(retired.Channel).ConfigureAwait(false); + } + } + } + + private async Task CloseRetainedChannelAsync(ITransportChannel channel) + { + try + { + if (channel is not IManagedTransportChannel) + { + await channel.CloseAsync(CancellationToken.None).ConfigureAwait(false); + } + } + catch (Exception ex) when (ex is not OutOfMemoryException) + { + m_logger.CleanupCloseSessionAsyncCloseChannelAsyncRaisedException(ex); + } + finally + { + channel.Dispose(); + } + } + /// /// Identity fences completion from an earlier recovery on the same lease. /// @@ -472,6 +562,7 @@ namespace Opc.Ua.Client private IClientChannelManager? m_channelManager; private IManagedTransportChannel? m_managedChannel; private PendingSubscriptionRecovery? m_pendingSubscriptionRecovery; + private RetiredServerSession? m_retiredSession; /// /// Single-flight gate for : the diff --git a/src/Opc.Ua.Client/Session/Session.cs b/src/Opc.Ua.Client/Session/Session.cs index 9d1a888df6..19e591e30c 100644 --- a/src/Opc.Ua.Client/Session/Session.cs +++ b/src/Opc.Ua.Client/Session/Session.cs @@ -578,6 +578,8 @@ namespace Opc.Ua.Client m_logger.ErrorClosingSessionDuringDispose(ex); } + await RetirePendingServerSessionAsync().ConfigureAwait(false); + m_defaultSubscription?.Dispose(); m_nodeCache?.Dispose(); @@ -3428,11 +3430,26 @@ namespace Opc.Ua.Client return RecreateInPlaceAsync(endpoint, budget, connection: null, ct); } + /// + /// Recreate the server-side session in place, optionally over a + /// waiting reverse connection. + /// + /// Optional replacement endpoint. + /// The outer reconnect retry budget. + /// Optional waiting reverse connection. + /// Cancellation token. + /// + /// When and a new session is created, the previous session is closed on the + /// previous channel once the subscriptions are restored. Used when the previous Server is still + /// reachable, so its session is not left allocated until it times out. + /// + /// is null. internal Task RecreateInPlaceAsync( ConfiguredEndpoint? endpoint, IRetryBudget budget, ITransportWaitingConnection? connection, - CancellationToken ct) + CancellationToken ct, + bool retirePreviousSession = false) { if (budget == null) { @@ -3444,7 +3461,8 @@ namespace Opc.Ua.Client connection, channel: null, budget, - ct); + ct, + retirePreviousSession: retirePreviousSession); } internal Task ReactivateOnNetworkEndpointAsync( @@ -3515,12 +3533,16 @@ namespace Opc.Ua.Client bool requireTokenReuse = false, SessionClient? recoveryClient = null, bool networkRecovery = false, - bool bindSuppliedChannel = false) + bool bindSuppliedChannel = false, + bool retirePreviousSession = false) { ThrowIfDisposed(); using Activity? activity = CachedActivitySource.StartActivity(); NodeId previousSessionId = SessionId; + NodeId previousAuthenticationToken = AuthenticationToken; + ITransportChannel? retainedChannel = null; + RetiredServerSession? retired = null; IManagedTransportChannel? oldManagedLease = ManagedChannel; IClientChannelManager? manager = oldManagedLease?.Manager ?? ChannelManager; IManagedTransportChannel? newManagedLease = null; @@ -3690,7 +3712,16 @@ namespace Opc.Ua.Client .ConfigureAwait(false); } - TransportChannel = newChannel; + if (retirePreviousSession) + { + // The previous channel stays open to close the previous session on it. + retainedChannel = NullableTransportChannel; + InitializeChannel(newChannel); + } + else + { + TransportChannel = newChannel; + } } // Token-reuse fast reconnect: re-activate the @@ -3761,6 +3792,21 @@ namespace Opc.Ua.Client SessionCreated(default, default); } + if (retirePreviousSession && !previousAuthenticationToken.IsNull) + { + ITransportChannel? previousChannel = manager == null + ? retainedChannel + : ReferenceEquals(oldManagedLease, newManagedLease) ? null : oldManagedLease; + if (previousChannel != null) + { + retired = new RetiredServerSession( + previousSessionId, + previousAuthenticationToken, + previousChannel); + retainedChannel = null; + } + } + UserIdentity? tempIdentity = m_identity == null ? new UserIdentity() : null; @@ -3830,7 +3876,8 @@ namespace Opc.Ua.Client managedLeaseActivated = true; if (newManagedLease != null && oldManagedLease != null && - !ReferenceEquals(oldManagedLease, newManagedLease)) + !ReferenceEquals(oldManagedLease, newManagedLease) && + !ReferenceEquals(retired?.Channel, oldManagedLease)) { oldManagedLease.Dispose(); } @@ -3861,6 +3908,13 @@ namespace Opc.Ua.Client m_logger.SessionRECREATEPLACESessionIdFailed( ex, previousSessionId); + if (retired != null) + { + // The client no longer holds the previous session, so it is closed now. + await RetireServerSessionAsync( + retired, + disposeChannel: !ReferenceEquals(retired.Channel, m_managedChannel)).ConfigureAwait(false); + } throw; } finally @@ -3882,9 +3936,18 @@ namespace Opc.Ua.Client Reconnecting = false; m_reconnectLock.Release(); } + if (retainedChannel != null) + { + await CloseRetainedChannelAsync(retainedChannel).ConfigureAwait(false); + } } } + if (retired != null) + { + await RetireAfterSubscriptionRecoveryAsync(retired).ConfigureAwait(false); + } + if (reused) { ct.ThrowIfCancellationRequested(); @@ -3940,6 +4003,7 @@ namespace Opc.Ua.Client remaining == TimeSpan.MaxValue ? Timeout.InfiniteTimeSpan : remaining); using var linked = CancellationTokenSource.CreateLinkedTokenSource(ct, timeout.Token); CancellationToken recoveryToken = linked.Token; + bool recovered = false; try { if (remaining == TimeSpan.Zero) @@ -3964,6 +4028,7 @@ namespace Opc.Ua.Client StatusCodes.BadSecureChannelClosed, "The subscription recovery deadline expired."); } m_pendingSubscriptionRecovery = null; + recovered = true; } catch (OperationCanceledException) when (timeout.IsCancellationRequested && !ct.IsCancellationRequested) @@ -3975,6 +4040,11 @@ namespace Opc.Ua.Client { await ResumePublishingAfterRecoveryAsync().ConfigureAwait(false); } + + if (recovered) + { + await RetirePendingServerSessionAsync().ConfigureAwait(false); + } } private async Task ResumePublishingAfterRecoveryAsync() @@ -7603,5 +7673,9 @@ namespace Opc.Ua.Client public static partial void RecreationDiscoveryUnavailableUsingStoredSnapshot( this ILogger logger, Exception exception); + + [LoggerMessage(EventId = ClientEventIds.Session + 72, Level = LogLevel.Information, + Message = "Closed the replaced session {SessionId} on the Server it was created on.")] + public static partial void SessionRetiredPreviousSession(this ILogger logger, NodeId? sessionId); } } diff --git a/tests/Opc.Ua.Client.Tests/Session/ConnectionStateMachineTests.cs b/tests/Opc.Ua.Client.Tests/Session/ConnectionStateMachineTests.cs index 453d7293ec..31d8f196c8 100644 --- a/tests/Opc.Ua.Client.Tests/Session/ConnectionStateMachineTests.cs +++ b/tests/Opc.Ua.Client.Tests/Session/ConnectionStateMachineTests.cs @@ -858,6 +858,105 @@ namespace Opc.Ua.Client.Tests.ManagedSession Is.EqualTo(ConnectionState.Disconnected)); } + [Test] + public async Task RequestedFailoverRunsOperationInFailoverStateAsync() + { + await using ConnectionStateMachine sm = CreateMachine(new ReconnectPolicy()); + sm.ConnectAsync = _ => Task.FromResult(ServiceResult.Good); + int handlerFailovers = 0; + sm.FailoverAsync = _ => + { + Interlocked.Increment(ref handlerFailovers); + return Task.FromResult(ServiceResult.Good); + }; + StateTransitionRecorder recorder = AttachRecorder(sm); + + Assert.That(sm.RequestFailover((_, _) => Task.FromResult(ServiceResult.Good)), Is.False); + sm.Start(); + sm.RequestConnect(); + using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(10)); + await sm.WaitForConnectedAsync(timeout.Token).ConfigureAwait(false); + + ConnectionState observed = ConnectionState.Disconnected; + Assert.That(sm.RequestFailover((_, _) => + { + observed = sm.State; + return Task.FromResult(ServiceResult.Good); + }), Is.True); + await sm.WaitForConnectedAsync(timeout.Token).ConfigureAwait(false); + + Assert.That(observed, Is.EqualTo(ConnectionState.Failover)); + Assert.That(recorder.HasVisited(ConnectionState.Failover), Is.True); + Assert.That(recorder.HasVisited(ConnectionState.Reconnecting), Is.False); + Assert.That(handlerFailovers, Is.Zero); + } + + [Test] + public async Task FailedRequestedFailoverReconnectsWithoutDisconnectAsync() + { + var policy = new ReconnectPolicy + { + Strategy = BackoffStrategy.Constant, + InitialDelay = TimeSpan.FromMilliseconds(5), + JitterFactor = 0.0 + }; + await using ConnectionStateMachine sm = CreateMachine(policy); + sm.ConnectAsync = _ => Task.FromResult(ServiceResult.Good); + int reconnects = 0; + sm.ReconnectAsync = _ => + { + Interlocked.Increment(ref reconnects); + return Task.FromResult(ServiceResult.Good); + }; + StateTransitionRecorder recorder = AttachRecorder(sm); + sm.Start(); + sm.RequestConnect(); + using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(10)); + await sm.WaitForConnectedAsync(timeout.Token).ConfigureAwait(false); + + Assert.That(sm.RequestFailover((_, _) => + Task.FromResult(new ServiceResult(StatusCodes.BadServerHalted))), Is.True); + await recorder.WaitForStateAsync(ConnectionState.Reconnecting, 5000).ConfigureAwait(false); + await sm.WaitForConnectedAsync(timeout.Token).ConfigureAwait(false); + + Assert.That(reconnects, Is.EqualTo(1)); + Assert.That(recorder.HasVisited(ConnectionState.Disconnected), Is.False); + Assert.That(sm.State, Is.EqualTo(ConnectionState.Connected)); + } + + [Test] + public async Task RequestedFailoverOutcomeReportsFailedSubscriptionRecoveryAsync() + { + var policy = new ReconnectPolicy + { + Strategy = BackoffStrategy.Constant, + InitialDelay = TimeSpan.FromMilliseconds(5), + JitterFactor = 0.0 + }; + await using ConnectionStateMachine sm = CreateMachine(policy); + sm.ConnectAsync = _ => Task.FromResult(ServiceResult.Good); + sm.ReconnectAsync = _ => Task.FromResult(ServiceResult.Good); + int failRecovery = 0; + sm.CompleteRecoveryAsync = _ => Task.FromResult(Interlocked.Exchange(ref failRecovery, 0) != 0 + ? new ServiceResult(StatusCodes.BadSubscriptionIdInvalid) + : ServiceResult.Good); + sm.Start(); + sm.RequestConnect(); + using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(10)); + await sm.WaitForConnectedAsync(timeout.Token).ConfigureAwait(false); + + Volatile.Write(ref failRecovery, 1); + var outcome = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + Assert.That(sm.RequestFailover((_, _) => Task.FromResult(ServiceResult.Good), outcome), Is.True); + Assert.That( + await Task.WhenAny(outcome.Task, Task.Delay(10000, timeout.Token)).ConfigureAwait(false), + Is.SameAs(outcome.Task)); + ServiceResult result = await outcome.Task.ConfigureAwait(false); + + Assert.That(result.StatusCode, Is.EqualTo((StatusCode)StatusCodes.BadSubscriptionIdInvalid)); + await sm.WaitForConnectedAsync(timeout.Token).ConfigureAwait(false); + } + [Test] public async Task WaitForConnectedAsyncThrowsOriginalErrorWhenConnectGivesUp() { diff --git a/tests/Opc.Ua.Client.Tests/Session/ManagedSessionServerRedundancyTests.cs b/tests/Opc.Ua.Client.Tests/Session/ManagedSessionServerRedundancyTests.cs new file mode 100644 index 0000000000..b4d5c1f665 --- /dev/null +++ b/tests/Opc.Ua.Client.Tests/Session/ManagedSessionServerRedundancyTests.cs @@ -0,0 +1,835 @@ +/* ======================================================================== + * Copyright (c) 2005-2026 The OPC Foundation, Inc. All rights reserved. + * + * OPC Foundation MIT License 1.00 + * + * Permission is hereby granted, free of charge, to any person + * obtaining a copy of this software and associated documentation + * files (the "Software"), to deal in the Software without + * restriction, including without limitation the rights to use, + * copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the + * Software is furnished to do so, subject to the following + * conditions: + * + * The above copyright notice and this permission notice shall be + * included in all copies or substantial portions of the Software. + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, + * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES + * OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND + * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT + * HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, + * WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING + * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR + * OTHER DEALINGS IN THE SOFTWARE. + * + * The complete license agreement can be found here: + * http://opcfoundation.org/License/MIT/1.00/ + * ======================================================================*/ + +using System; +using System.Collections.Concurrent; +using System.Collections.Generic; +using System.IO; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using NUnit.Framework; +using Opc.Ua.Client.Subscriptions; +using Opc.Ua.Client.TestFramework; +using Opc.Ua.Server; +using Opc.Ua.Server.TestFramework; +using Opc.Ua.Tests; +using Quickstarts.ReferenceServer; +using ISubscription = Opc.Ua.Client.Subscriptions.ISubscription; +using ISubscriptionManager = Opc.Ua.Client.Subscriptions.ISubscriptionManager; +using UaManagedSession = Opc.Ua.Client.ManagedSession; + +namespace Opc.Ua.Client.Tests +{ + /// + /// Runs a managed session against a Warm redundant Server set of two + /// in-process reference servers (OPC 10000-4 §6.6.2). + /// + [TestFixture] + [NonParallelizable] + [Category("Client")] + [Category("Integration")] + public sealed partial class ManagedSessionServerRedundancyTests + { + /// + /// Without probing, the session stays on the Server it connected to and + /// a member known only from ServerUriArray keeps an unknown ServiceLevel. + /// + [Test] + public async Task ConnectWithoutProbingStaysOnConfiguredServerAsync() + { + using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(60)); + await using RedundantSet set = await RedundantSet.StartAsync(timeout.Token).ConfigureAwait(false); + set.Primary.SetServiceLevel(210); + set.Backup.SetServiceLevel(250); + + await using UaManagedSession session = await set.ConnectAsync( + new ServerRedundancyOptions(), timeout.Token).ConfigureAwait(false); + + Assert.That(ServerUriOf(session), Is.EqualTo(set.Primary.ApplicationUri)); + ServerRedundancyInfo info = await WaitForResolvedMemberAsync(session, set.Backup, timeout.Token) + .ConfigureAwait(false); + RedundantServer backup = Member(info, set.Backup); + Assert.Multiple(() => + { + Assert.That(info.Mode, Is.EqualTo(RedundancySupport.Warm)); + Assert.That(info.ServiceLevel, Is.EqualTo(210)); + Assert.That(backup.ServiceLevelKnown, Is.False); + Assert.That(backup.ServiceLevel, Is.EqualTo(ServiceLevels.NoData)); + Assert.That(backup.ResolutionError, Is.Null); + }); + } + + /// + /// With probing, the initial connect switches to the member with the + /// higher ServiceLevel. + /// + [Test] + public async Task ConnectWithProbingSelectsMemberWithHigherServiceLevelAsync() + { + using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(60)); + await using RedundantSet set = await RedundantSet.StartAsync(timeout.Token).ConfigureAwait(false); + set.Primary.SetServiceLevel(210); + set.Backup.SetServiceLevel(250); + + await using UaManagedSession session = await set.ConnectAsync( + new ServerRedundancyOptions { ProbeMemberServiceLevel = true }, timeout.Token).ConfigureAwait(false); + + Assert.That(ServerUriOf(session), Is.EqualTo(set.Backup.ApplicationUri)); + ServerRedundancyInfo? info = session.RedundancyInfo; + Assert.That(info, Is.Not.Null); + Assert.That(info!.ServiceLevel, Is.EqualTo(250)); + DataValue level = await session.ReadValueAsync(VariableIds.Server_ServiceLevel, timeout.Token) + .ConfigureAwait(false); + Assert.That(level.GetValue(0), Is.EqualTo(250)); + Assert.That( + await EventuallyAsync(() => set.Primary.SessionCount == 0).ConfigureAwait(false), + Is.True, + "The session replaced on the Server connected to first must be closed there."); + } + + /// + /// A member whose resolved endpoint does not keep the security mode and + /// policy of the session in use is not probed. + /// + [Test] + public async Task ProbeSkipsMemberWithDifferentSecurityAsync() + { + using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(60)); + await using RedundantSet set = await RedundantSet.StartAsync(timeout.Token).ConfigureAwait(false); + set.Primary.SetServiceLevel(210); + set.Backup.SetServiceLevel(250); + ConfiguredEndpoint secured = await set.Client.GetEndpointAsync( + new Uri(set.Backup.Url), SecurityPolicies.Basic256Sha256).ConfigureAwait(false); + Assert.That(secured.Description.SecurityMode, Is.Not.EqualTo(MessageSecurityMode.None)); + + await using UaManagedSession session = await set.ConnectAsync( + new ServerRedundancyOptions { ProbeMemberServiceLevel = true }, + timeout.Token, + backupEndpoint: secured).ConfigureAwait(false); + + Assert.That(ServerUriOf(session), Is.EqualTo(set.Primary.ApplicationUri)); + RedundantServer backup = Member(session.RedundancyInfo!, set.Backup); + Assert.Multiple(() => + { + Assert.That( + backup.ResolutionError?.StatusCode, + Is.EqualTo((StatusCode)StatusCodes.BadSecurityModeRejected)); + Assert.That(backup.ServiceLevelKnown, Is.False); + Assert.That(set.Backup.CreatedSessionCount, Is.Zero); + }); + } + + /// + /// A probe works on a copy of the resolved endpoint and of the identity in + /// use, so it changes neither. + /// + [Test] + public async Task ProbeUsesCopiesOfEndpointAndIdentityAsync() + { + using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(60)); + await using RedundantSet set = await RedundantSet.StartAsync(timeout.Token).ConfigureAwait(false); + set.Primary.SetServiceLevel(ServiceLevels.Maximum); + set.Backup.SetServiceLevel(230); + set.Backup.Endpoint.UpdateBeforeConnect = true; + var factory = new RecordingSessionFactory(set.Telemetry); + + await using UaManagedSession session = await set.ConnectAsync( + new ServerRedundancyOptions { ProbeMemberServiceLevel = true }, + timeout.Token, + sessionFactory: factory, + identity: new UserIdentity()).ConfigureAwait(false); + + Assert.That(ServerUriOf(session), Is.EqualTo(set.Primary.ApplicationUri)); + Assert.That(Member(session.RedundancyInfo!, set.Backup).ServiceLevel, Is.EqualTo(230)); + (ConfiguredEndpoint endpoint, IUserIdentity? identity) = factory.Calls + .First(c => c.Endpoint.Description.Server?.ApplicationUri == set.Backup.ApplicationUri); + Assert.Multiple(() => + { + Assert.That(set.Backup.Endpoint.UpdateBeforeConnect, Is.True); + Assert.That(endpoint, Is.Not.SameAs(set.Backup.Endpoint)); + Assert.That(identity, Is.Not.Null); + Assert.That(identity, Is.Not.SameAs(session.Identity)); + Assert.That(identity!.TokenHandler, Is.Not.SameAs(session.Identity.TokenHandler)); + }); + } + + /// + /// While the Server in use is Healthy, ServiceLevel monitoring does not + /// probe the other members. + /// + [Test] + public async Task MonitoringOfHealthyServerDoesNotProbeMembersAsync() + { + using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(60)); + await using RedundantSet set = await RedundantSet.StartAsync(timeout.Token).ConfigureAwait(false); + set.Primary.SetServiceLevel(ServiceLevels.Maximum); + set.Backup.SetServiceLevel(230); + var options = new ServerRedundancyOptions + { + ProbeMemberServiceLevel = true, + ServiceLevelMonitoringInterval = TimeSpan.FromMilliseconds(100) + }; + + await using UaManagedSession session = await set.ConnectAsync(options, timeout.Token) + .ConfigureAwait(false); + await Task.Delay(500, timeout.Token).ConfigureAwait(false); + int probesAfterConnect = set.Backup.CreatedSessionCount; + await Task.Delay(1500, timeout.Token).ConfigureAwait(false); + + Assert.Multiple(() => + { + Assert.That(ServerUriOf(session), Is.EqualTo(set.Primary.ApplicationUri)); + Assert.That(probesAfterConnect, Is.EqualTo(1)); + Assert.That(set.Backup.CreatedSessionCount, Is.EqualTo(probesAfterConnect)); + }); + } + + /// + /// A member that refuses the ServiceLevel read keeps an unknown + /// ServiceLevel and reports the exact status. + /// + [Test] + public async Task MemberRefusingProbeReportsResolutionErrorAsync() + { + using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(60)); + await using RedundantSet set = await RedundantSet.StartAsync(timeout.Token).ConfigureAwait(false); + set.Primary.SetServiceLevel(210); + set.Backup.SetServiceLevel(250); + set.Backup.RefuseRead(VariableIds.Server_ServiceLevel); + + await using UaManagedSession session = await set.ConnectAsync( + new ServerRedundancyOptions { ProbeMemberServiceLevel = true }, timeout.Token).ConfigureAwait(false); + + Assert.That(ServerUriOf(session), Is.EqualTo(set.Primary.ApplicationUri)); + RedundantServer backup = Member(session.RedundancyInfo!, set.Backup); + Assert.Multiple(() => + { + Assert.That(backup.Endpoint, Is.Not.Null); + Assert.That(backup.ServiceLevelKnown, Is.False); + Assert.That(backup.ResolutionError, Is.Not.Null); + Assert.That(backup.ResolutionError?.StatusCode, Is.EqualTo((StatusCode)StatusCodes.BadNotReadable)); + }); + } + + /// + /// A member that refuses the ServerStatus.State read reports the exact + /// status and is not selected. + /// + [Test] + public async Task MemberRefusingServerStateReportsResolutionErrorAsync() + { + using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(60)); + await using RedundantSet set = await RedundantSet.StartAsync(timeout.Token).ConfigureAwait(false); + set.Primary.SetServiceLevel(210); + set.Backup.SetServiceLevel(250); + set.Backup.RefuseRead(VariableIds.Server_ServerStatus_State); + + await using UaManagedSession session = await set.ConnectAsync( + new ServerRedundancyOptions { ProbeMemberServiceLevel = true }, timeout.Token).ConfigureAwait(false); + + Assert.That(ServerUriOf(session), Is.EqualTo(set.Primary.ApplicationUri)); + RedundantServer backup = Member(session.RedundancyInfo!, set.Backup); + Assert.That(backup.ResolutionError?.StatusCode, Is.EqualTo((StatusCode)StatusCodes.BadNotReadable)); + } + + /// + /// A redundancy node the Server refuses reports its exact status. + /// + [Test] + public async Task RefusedRedundancyNodeReportsReadStatusAsync() + { + using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(60)); + await using RedundantSet set = await RedundantSet.StartAsync(timeout.Token).ConfigureAwait(false); + set.Primary.RefuseRead(VariableIds.Server_ServerRedundancy_RedundantServerArray); + + await using UaManagedSession session = await set.ConnectAsync( + new ServerRedundancyOptions(), timeout.Token).ConfigureAwait(false); + + ServerRedundancyInfo? info = session.RedundancyInfo; + Assert.That(info, Is.Not.Null); + Assert.Multiple(() => + { + Assert.That(info!.Mode, Is.EqualTo(RedundancySupport.Warm)); + Assert.That( + info.ReadStatuses[VariableIds.Server_ServerRedundancy_RedundantServerArray], + Is.EqualTo((StatusCode)StatusCodes.BadNotReadable)); + Assert.That( + info.ReadStatuses[VariableIds.Server_ServerRedundancy_ServerUriArray], + Is.EqualTo((StatusCode)StatusCodes.Good)); + Assert.That( + info.ReadStatuses[VariableIds.Server_ServerRedundancy_RedundancySupport], + Is.EqualTo((StatusCode)StatusCodes.Good)); + Assert.That(info.RedundantServers.ToList().Select(s => s.ServerUri), + Is.EqualTo(new[] { set.Backup.ApplicationUri })); + }); + } + + /// + /// A degraded ServiceLevel of the Server in use fails over to the + /// healthy member while connected, without a connection loss, and the + /// subscription continues on the new Server. + /// + [Test] + public async Task DegradedServiceLevelFailsOverAndSubscriptionContinuesAsync() + { + using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(90)); + await using RedundantSet set = await RedundantSet.StartAsync(timeout.Token).ConfigureAwait(false); + set.Primary.SetServiceLevel(ServiceLevels.Maximum); + set.Backup.SetServiceLevel(230); + var options = new ServerRedundancyOptions + { + ProbeMemberServiceLevel = true, + ServiceLevelMonitoringInterval = TimeSpan.FromMilliseconds(250) + }; + await using UaManagedSession session = await set.ConnectAsync(options, timeout.Token) + .ConfigureAwait(false); + Assert.That(ServerUriOf(session), Is.EqualTo(set.Primary.ApplicationUri)); + var states = new ConcurrentQueue(); + session.ConnectionStateChanged += (_, e) => states.Enqueue(e.NewState); + + var recorder = new DataChangeRecorder(); + ISubscriptionManager manager = session.RequireSubscriptionManager(); + ISubscription subscription = manager.Add( + recorder, + OptionsFactory.Create(new Subscriptions.SubscriptionOptions + { + PublishingEnabled = true, + PublishingInterval = TimeSpan.FromMilliseconds(100), + KeepAliveCount = 10, + LifetimeCount = 100 + })); + Assert.That(subscription.MonitoredItems.TryAdd( + "CurrentTime", + OptionsFactory.Create(new Subscriptions.MonitoredItems.MonitoredItemOptions + { + StartNodeId = VariableIds.Server_ServerStatus_CurrentTime, + SamplingInterval = TimeSpan.FromMilliseconds(100), + QueueSize = 1 + }), + out _), Is.True); + await WaitUntilAsync(() => recorder.Count >= 3, timeout.Token).ConfigureAwait(false); + Assert.That(states, Is.Empty); + + set.Primary.SetServiceLevel(150); + await WaitUntilAsync( + () => ServerUriOf(session) == set.Backup.ApplicationUri && + session.StateMachine.State == ConnectionState.Connected, + timeout.Token).ConfigureAwait(false); + DateTime failedOverAt = DateTime.UtcNow; + await WaitUntilAsync(() => recorder.LatestSourceTimestamp > failedOverAt, timeout.Token) + .ConfigureAwait(false); + + Assert.Multiple(() => + { + Assert.That(states.ToArray(), Is.EqualTo(new[] { ConnectionState.Failover, ConnectionState.Connected })); + Assert.That(manager.Items.Single(), Is.SameAs(subscription)); + Assert.That(session.RedundancyInfo?.ServiceLevel, Is.EqualTo(230)); + Assert.That(recorder.Errors, Is.Empty); + }); + Assert.That( + await EventuallyAsync(() => set.Primary.SessionCount == 0).ConfigureAwait(false), + Is.True, + "The session on the Server left must be closed there."); + } + + /// + /// FailoverAsync switches a connected session to an explicit target + /// without a connection loss and closes the session on the Server left. + /// + [TestCase(false)] + [TestCase(true)] + public async Task FailoverAsyncSwitchesToExplicitTargetAsync(bool useChannelManager) + { + using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(60)); + await using RedundantSet set = await RedundantSet.StartAsync(timeout.Token).ConfigureAwait(false); + await using ClientChannelManager? channels = useChannelManager + ? new ClientChannelManager(set.Client.Config, set.Telemetry) + : null; + set.Primary.SetServiceLevel(ServiceLevels.Maximum); + set.Backup.SetServiceLevel(230); + await using UaManagedSession session = await set.ConnectAsync( + new ServerRedundancyOptions(), timeout.Token, channelManager: channels).ConfigureAwait(false); + Assert.That(set.Primary.SessionCount, Is.EqualTo(1)); + var states = new ConcurrentQueue(); + session.ConnectionStateChanged += (_, e) => states.Enqueue(e.NewState); + + await session.FailoverAsync(set.Backup.Endpoint, timeout.Token).ConfigureAwait(false); + await WaitUntilAsync(() => states.Contains(ConnectionState.Connected), timeout.Token).ConfigureAwait(false); + + Assert.That(ServerUriOf(session), Is.EqualTo(set.Backup.ApplicationUri)); + Assert.That(states.ToArray(), Is.EqualTo(new[] { ConnectionState.Failover, ConnectionState.Connected })); + DataValue level = await session.ReadValueAsync(VariableIds.Server_ServiceLevel, timeout.Token) + .ConfigureAwait(false); + Assert.That(level.GetValue(0), Is.EqualTo(230)); + Assert.That(session.RedundancyInfo?.ServiceLevel, Is.EqualTo(230)); + Assert.That( + await EventuallyAsync(() => set.Primary.SessionCount == 0).ConfigureAwait(false), + Is.True, + "The session on the Server left must be closed there."); + Assert.That(set.Backup.SessionCount, Is.EqualTo(1)); + } + + /// + /// FailoverAsync without a target fails over to the member the + /// redundancy handler selects. + /// + [Test] + public async Task FailoverAsyncSwitchesToHandlerSelectedTargetAsync() + { + using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(60)); + await using RedundantSet set = await RedundantSet.StartAsync(timeout.Token).ConfigureAwait(false); + set.Primary.SetServiceLevel(ServiceLevels.Maximum); + set.Backup.SetServiceLevel(230); + await using UaManagedSession session = await set.ConnectAsync( + new ServerRedundancyOptions(), timeout.Token).ConfigureAwait(false); + + await session.FailoverAsync(ct: timeout.Token).ConfigureAwait(false); + + Assert.That(ServerUriOf(session), Is.EqualTo(set.Backup.ApplicationUri)); + Assert.That(session.StateMachine.State, Is.EqualTo(ConnectionState.Connected)); + } + + /// + /// FailoverAsync requires a redundancy handler. + /// + [Test] + public async Task FailoverAsyncWithoutRedundancyHandlerThrowsAsync() + { + using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(60)); + await using RedundantSet set = await RedundantSet.StartAsync(timeout.Token).ConfigureAwait(false); + await using UaManagedSession session = await set.ConnectAsync( + new ServerRedundancyOptions(), timeout.Token, withRedundancyHandler: false).ConfigureAwait(false); + + await Assert.ThatAsync( + () => session.FailoverAsync(set.Backup.Endpoint, timeout.Token), + Throws.TypeOf() + .With.Property(nameof(ServiceResultException.StatusCode)) + .EqualTo((StatusCode)StatusCodes.BadNotSupported)).ConfigureAwait(false); + + Assert.That(session.RedundancyInfo, Is.Null); + Assert.That(ServerUriOf(session), Is.EqualTo(set.Primary.ApplicationUri)); + } + + private static string? ServerUriOf(UaManagedSession session) + { + return session.ConfiguredEndpoint.Description.Server?.ApplicationUri; + } + + private static async Task WaitUntilAsync(Func condition, CancellationToken ct) + { + while (!condition()) + { + await Task.Delay(20, ct).ConfigureAwait(false); + } + } + + private static async Task EventuallyAsync(Func condition) + { + DateTime deadline = DateTime.UtcNow.AddSeconds(10); + while (!condition()) + { + if (DateTime.UtcNow > deadline) + { + return false; + } + await Task.Delay(20).ConfigureAwait(false); + } + return true; + } + + private static RedundantServer Member(ServerRedundancyInfo info, RedundantMember member) + { + return info.RedundantServers.ToList().Single(s => s.ServerUri == member.ApplicationUri); + } + + private static async Task WaitForResolvedMemberAsync( + UaManagedSession session, + RedundantMember member, + CancellationToken ct) + { + while (true) + { + ServerRedundancyInfo? info = session.RedundancyInfo; + if (info != null && + info.RedundantServers.ToList().Any(s => s.ServerUri == member.ApplicationUri && s.Endpoint != null)) + { + return info; + } + await Task.Delay(50, ct).ConfigureAwait(false); + } + } + + /// + /// Two reference servers that publish each other in ServerUriArray as a + /// Warm redundant Server set, and the client configuration to reach them. + /// + private sealed class RedundantSet : IAsyncDisposable + { + private RedundantSet(string root, ITelemetryContext telemetry) + { + m_root = root; + Telemetry = telemetry; + Client = new ClientFixture(telemetry); + } + + public RedundantMember Primary { get; private set; } = null!; + + public RedundantMember Backup { get; private set; } = null!; + + public ClientFixture Client { get; } + + public ITelemetryContext Telemetry { get; } + + public static async Task StartAsync(CancellationToken ct) + { + ITelemetryContext telemetry = NUnitTelemetryContext.Create(); + string root = Path.Combine(Path.GetTempPath(), "redundancy-" + Guid.NewGuid().ToString("N")[..8]); + var set = new RedundantSet(root, telemetry); + try + { + set.Primary = await RedundantMember.StartAsync(root, "Primary").ConfigureAwait(false); + set.Backup = await RedundantMember.StartAsync(root, "Backup").ConfigureAwait(false); + await set.Primary.PublishSetAsync([set.Backup.ApplicationUri], ct).ConfigureAwait(false); + await set.Backup.PublishSetAsync([set.Primary.ApplicationUri], ct).ConfigureAwait(false); + await set.Client.LoadClientConfigurationAsync(root, "RedundancyClient").ConfigureAwait(false); + set.Primary.Endpoint = await set.Client.GetEndpointAsync( + new Uri(set.Primary.Url), SecurityPolicies.None).ConfigureAwait(false); + set.Backup.Endpoint = await set.Client.GetEndpointAsync( + new Uri(set.Backup.Url), SecurityPolicies.None).ConfigureAwait(false); + return set; + } + catch + { + await set.DisposeAsync().ConfigureAwait(false); + throw; + } + } + + public Task ConnectAsync( + ServerRedundancyOptions options, + CancellationToken ct, + bool withRedundancyHandler = true, + ConfiguredEndpoint? backupEndpoint = null, + ISessionFactory? sessionFactory = null, + IUserIdentity? identity = null, + IClientChannelManager? channelManager = null) + { + var resolver = new MemberEndpointResolver(new Dictionary + { + [Primary.ApplicationUri] = Primary.Endpoint, + [Backup.ApplicationUri] = backupEndpoint ?? Backup.Endpoint + }); + return UaManagedSession.CreateAsync( + Client.Config, + Primary.Endpoint, + sessionFactory ?? Client.SessionFactory, + identity: identity, + redundancyHandler: withRedundancyHandler + ? new DefaultServerRedundancyHandler(resolver, options: options) + : null, + telemetry: Telemetry, + channelManager: channelManager, + serverRedundancy: options, + ct: ct); + } + + public async ValueTask DisposeAsync() + { + await Client.DisposeAsync().ConfigureAwait(false); + if (Backup != null) + { + await Backup.DisposeAsync().ConfigureAwait(false); + } + if (Primary != null) + { + await Primary.DisposeAsync().ConfigureAwait(false); + } + try + { + if (Directory.Exists(m_root)) + { + Directory.Delete(m_root, recursive: true); + } + } + catch (IOException) + { + // A certificate store file may still be open; the temp folder is reclaimed later. + } + } + + private readonly string m_root; + } + + /// + /// One in-process reference server with its own ApplicationUri and an + /// explicitly controlled ServiceLevel. + /// + private sealed class RedundantMember : IAsyncDisposable + { + private RedundantMember(ServerFixture fixture, ReferenceServer server) + { + m_fixture = fixture; + m_server = server; + ApplicationUri = fixture.Config.ApplicationUri!; + Url = $"{Utils.UriSchemeOpcTcp}://localhost:{fixture.Port}"; + m_setServiceLevel = ((IServerServiceLevelControl)server.CurrentInstance).ClaimServiceLevelControl(); + server.CurrentInstance.SessionManager.SessionCreated += + (_, _) => Interlocked.Increment(ref m_createdSessions); + } + + public string ApplicationUri { get; } + + public string Url { get; } + + public ConfiguredEndpoint Endpoint { get; set; } = null!; + + /// + /// The sessions currently allocated on the server. + /// + public int SessionCount => m_server.CurrentInstance.SessionManager.GetSessions().Count; + + /// + /// The sessions created on the server since it started. + /// + public int CreatedSessionCount => Volatile.Read(ref m_createdSessions); + + public static async Task StartAsync(string root, string name) + { + var fixture = new ServerFixture(context => new ReferenceServer(context)) + { + AutoAccept = true, + SecurityNone = true + }; + string pki = Path.Combine(root, name); + await fixture.LoadConfigurationAsync(pki).ConfigureAwait(false); + fixture.Config.ApplicationUri = Utils.ReplaceLocalhost("urn:localhost:UA:RedundantMember:" + name); + ReferenceServer server = await fixture.StartAsync(pki).ConfigureAwait(false); + return new RedundantMember(fixture, server); + } + + public void SetServiceLevel(byte level) + { + m_setServiceLevel(level); + } + + /// + /// Promotes Server.ServerRedundancy to NonTransparentRedundancyType and + /// publishes Warm redundancy with the given peers. + /// + public async Task PublishSetAsync(string[] peerUris, CancellationToken ct) + { + IServerInternal server = m_server.CurrentInstance; + ServerObjectState serverObject = server.ServerObject; + ServerSystemContext context = server.DefaultSystemContext; + var replacer = (IPredefinedNodeSubtypeReplacer)server.DiagnosticsNodeManager; + NonTransparentRedundancyState subtype = context.CreateInstanceOfNonTransparentRedundancyType(); + subtype.AddRedundantServerArray(context); + var replaced = (NonTransparentRedundancyState)await replacer.ReplacePredefinedInstanceSubtypeAsync( + context, + serverObject.ServerRedundancy!, + subtype, + new Dictionary + { + [new QualifiedName(BrowseNames.ServerUriArray, 0)] = + VariableIds.Server_ServerRedundancy_ServerUriArray + }, + node => serverObject.ServerRedundancy = (ServerRedundancyState)node, + ct).ConfigureAwait(false); + replaced.RedundancySupport!.Value = RedundancySupport.Warm; + replaced.ServerUriArray!.Value = peerUris; + replaced.ClearChangeMasks(context, true); + } + + /// + /// Makes a variable of the Server object unreadable for every user. + /// + public void RefuseRead(NodeId variableId) + { + IServerInternal server = m_server.CurrentInstance; + ServerObjectState serverObject = server.ServerObject; + BaseVariableState variable = variableId == VariableIds.Server_ServiceLevel + ? serverObject.ServiceLevel! + : variableId == VariableIds.Server_ServerStatus_State + ? serverObject.ServerStatus!.State! + : serverObject.ServerRedundancy!.RedundantServerArray!; + variable.AccessLevel = AccessLevels.None; + variable.UserAccessLevel = AccessLevels.None; + variable.ClearChangeMasks(server.DefaultSystemContext, false); + } + + public async ValueTask DisposeAsync() + { + await m_fixture.StopAsync().ConfigureAwait(false); + m_server.Dispose(); + } + + private readonly ServerFixture m_fixture; + private readonly ReferenceServer m_server; + private readonly Action m_setServiceLevel; + private int m_createdSessions; + } + + /// + /// Records the endpoint and identity of every session the factory creates + /// without a reverse connection. + /// + private sealed class RecordingSessionFactory : DefaultSessionFactory + { + public RecordingSessionFactory(ITelemetryContext telemetry) + : base(telemetry) + { + } + + public ConcurrentQueue<(ConfiguredEndpoint Endpoint, IUserIdentity? Identity)> Calls { get; } = new(); + + public override Task CreateAsync( + ApplicationConfiguration configuration, + ConfiguredEndpoint endpoint, + bool updateBeforeConnect, + bool checkDomain, + string sessionName, + uint sessionTimeout, + IUserIdentity? identity, + ArrayOf preferredLocales, + CancellationToken ct = default) + { + Calls.Enqueue((endpoint, identity)); + return base.CreateAsync( + configuration, + endpoint, + updateBeforeConnect, + checkDomain, + sessionName, + sessionTimeout, + identity, + preferredLocales, + ct); + } + } + + /// + /// Records the source timestamps of the data changes of a subscription. + /// + private sealed class DataChangeRecorder : ISubscriptionNotificationHandler + { + public int Count => Volatile.Read(ref m_count); + + public DateTime LatestSourceTimestamp => new(Interlocked.Read(ref m_latestTicks), DateTimeKind.Utc); + + public ConcurrentQueue Errors { get; } = new(); + + public ValueTask OnDataChangeNotificationAsync( + ISubscription subscription, + uint sequenceNumber, + DateTime publishTime, + ReadOnlyMemory notification, + PublishState publishStateMask, + IReadOnlyList stringTable) + { + foreach (DataValueChange change in notification.Span) + { + if (StatusCode.IsBad(change.Value.StatusCode)) + { + Errors.Enqueue(change.Value.StatusCode); + continue; + } + long ticks = change.Value.GetValue(DateTime.MinValue).ToUniversalTime().Ticks; + long latest = Interlocked.Read(ref m_latestTicks); + while (ticks > latest) + { + long previous = Interlocked.CompareExchange(ref m_latestTicks, ticks, latest); + if (previous == latest) + { + break; + } + latest = previous; + } + Interlocked.Increment(ref m_count); + } + return default; + } + + public ValueTask OnEventDataNotificationAsync( + ISubscription subscription, + uint sequenceNumber, + DateTime publishTime, + ReadOnlyMemory notification, + PublishState publishStateMask, + IReadOnlyList stringTable) + { + return default; + } + + public ValueTask OnKeepAliveNotificationAsync( + ISubscription subscription, + uint sequenceNumber, + DateTime publishTime, + PublishState publishStateMask) + { + return default; + } + + public ValueTask OnSubscriptionStateChangedAsync( + ISubscription subscription, + Subscriptions.SubscriptionState state, + PublishState publishStateMask, + CancellationToken ct = default) + { + return default; + } + + private long m_latestTicks = DateTime.MinValue.Ticks; + private int m_count; + } + + /// + /// Resolves the members of the set to the endpoints read from their + /// discovery endpoints. + /// + private sealed class MemberEndpointResolver : IRedundantServerEndpointResolver + { + public MemberEndpointResolver(IReadOnlyDictionary endpoints) + { + m_endpoints = endpoints; + } + + public ValueTask ResolveAsync( + string serverUri, + ConfiguredEndpoint currentEndpoint, + CancellationToken ct = default) + { + return new ValueTask( + m_endpoints.TryGetValue(serverUri, out ConfiguredEndpoint? endpoint) ? endpoint : null); + } + + private readonly IReadOnlyDictionary m_endpoints; + } + } +} diff --git a/tests/Opc.Ua.Client.Tests/Session/ServerRedundancyHandlerTests.cs b/tests/Opc.Ua.Client.Tests/Session/ServerRedundancyHandlerTests.cs index e93476e034..c2ecab16b6 100644 --- a/tests/Opc.Ua.Client.Tests/Session/ServerRedundancyHandlerTests.cs +++ b/tests/Opc.Ua.Client.Tests/Session/ServerRedundancyHandlerTests.cs @@ -33,6 +33,7 @@ using System.Threading.Tasks; using Microsoft.Extensions.Time.Testing; using Moq; using NUnit.Framework; +using Opc.Ua.Tests; namespace Opc.Ua.Client.Tests.ManagedSession { @@ -935,9 +936,116 @@ namespace Opc.Ua.Client.Tests.ManagedSession Assert.That(info.RedundantServers, Has.Count.EqualTo(1)); Assert.That(info.RedundantServers[0].Endpoint, Is.Null); + Assert.That(info.RedundantServers[0].ResolutionError, Is.Not.Null); + Assert.That(info.RedundantServers[0].ResolutionError!.StatusCode, Is.EqualTo(StatusCodes.BadNotFound)); Assert.That(target, Is.Null); } + /// + /// Verifies that a ServiceLevel monitoring interval must be positive and is off by default. + /// + [Test] + public void ServiceLevelMonitoringIntervalMustBePositive() + { + var defaults = new ServerRedundancyOptions(); + Assert.That(defaults.ServiceLevelMonitoringInterval, Is.Null); + Assert.That(defaults.ProbeMemberServiceLevel, Is.False); + Assert.That( + () => new ServerRedundancyOptions { ServiceLevelMonitoringInterval = TimeSpan.Zero }.Validate(), + Throws.TypeOf()); + Assert.That( + () => new ServerRedundancyOptions { ServiceLevelMonitoringInterval = Timeout.InfiniteTimeSpan }.Validate(), + Throws.TypeOf()); + Assert.That( + () => new ServerRedundancyOptions { ServiceLevelMonitoringInterval = TimeSpan.FromSeconds(1) }.Validate(), + Throws.Nothing); + } + + /// + /// Verifies that a failed endpoint resolution keeps the exact status of the failure. + /// + [Test] + public async Task FetchRedundancyInfoReportsResolutionErrorOfFailedPeerAsync() + { + m_resolver.Setup(r => r.ResolveAsync( + "urn:refusing", + It.IsAny(), + It.IsAny())) + .ThrowsAsync(new ServiceResultException(StatusCodes.BadCertificateUntrusted)); + Mock mockSession = CreateMockSession( + redundancySupport: (int)RedundancySupport.Warm, + serviceLevel: ServiceLevels.Maximum, + serverUris: ["urn:refusing"]); + + ServerRedundancyInfo info = await m_handler.FetchRedundancyInfoAsync( + mockSession.Object).ConfigureAwait(false); + + Assert.That(info.RedundantServers, Has.Count.EqualTo(1)); + Assert.That(info.RedundantServers[0].Endpoint, Is.Null); + Assert.That(info.RedundantServers[0].ServiceLevelKnown, Is.False); + Assert.That( + info.RedundantServers[0].ResolutionError?.StatusCode, + Is.EqualTo((StatusCode)StatusCodes.BadCertificateUntrusted)); + } + + /// + /// Verifies that a discovery failure of the default resolver is reported instead of BadNotFound. + /// + [Test] + public async Task FetchRedundancyInfoReportsDiscoveryFailureOfDefaultResolverAsync() + { + var resolver = new DefaultRedundantServerEndpointResolver( + NUnitTelemetryContext.Create(), + new FailingDiscovery(StatusCodes.BadCertificateUntrusted)); + var handler = new DefaultServerRedundancyHandler(resolver, new FixedTimeProvider(s_now)); + Mock mockSession = CreateMockSession( + redundancySupport: (int)RedundancySupport.Warm, + serviceLevel: ServiceLevels.Maximum, + serverUris: ["urn:refusing"]); + + ServerRedundancyInfo info = await handler.FetchRedundancyInfoAsync( + mockSession.Object).ConfigureAwait(false); + + Assert.That(info.RedundantServers, Has.Count.EqualTo(1)); + Assert.That(info.RedundantServers[0].Endpoint, Is.Null); + Assert.That( + info.RedundantServers[0].ResolutionError?.StatusCode, + Is.EqualTo((StatusCode)StatusCodes.BadCertificateUntrusted)); + } + + /// + /// Verifies that the status of every redundancy node read is reported, including refused nodes. + /// + [Test] + public async Task FetchRedundancyInfoReportsReadStatusOfEachNodeAsync() + { + Mock mockSession = CreateMockSession( + redundancySupport: (int)RedundancySupport.Transparent, + serviceLevel: ServiceLevels.Maximum, + currentServerId: "urn:current"); + + ServerRedundancyInfo info = await m_handler.FetchRedundancyInfoAsync( + mockSession.Object).ConfigureAwait(false); + + Assert.That(info.CurrentServerId, Is.EqualTo("urn:current")); + Assert.That(info.ReadStatuses, Has.Count.EqualTo(6)); + Assert.That( + info.ReadStatuses[VariableIds.Server_ServerRedundancy_RedundancySupport], + Is.EqualTo((StatusCode)StatusCodes.Good)); + Assert.That( + info.ReadStatuses[VariableIds.Server_ServiceLevel], + Is.EqualTo((StatusCode)StatusCodes.Good)); + Assert.That( + info.ReadStatuses[VariableIds.Server_ServerRedundancy_RedundantServerArray], + Is.EqualTo((StatusCode)StatusCodes.BadNodeIdUnknown)); + Assert.That( + info.ReadStatuses[VariableIds.Server_ServerRedundancy_ServerUriArray], + Is.EqualTo((StatusCode)StatusCodes.BadNodeIdUnknown)); + Assert.That( + info.ReadStatuses[VariableIds.Server_ServerRedundancy_CurrentServerId], + Is.EqualTo((StatusCode)StatusCodes.Good)); + } + /// /// Verifies that a URI-only peer can be selected when its service level is unknown. /// @@ -1275,5 +1383,34 @@ namespace Opc.Ua.Client.Tests.ManagedSession private readonly DateTimeOffset m_utcNow; } + + private sealed class FailingDiscovery : IRedundantServerDiscovery + { + public FailingDiscovery(StatusCode statusCode) + { + m_statusCode = statusCode; + } + + public ValueTask> FindServersAsync( + Uri discoveryUri, + EndpointConfiguration configuration, + string serverUri, + ITelemetryContext telemetry, + CancellationToken ct) + { + throw new ServiceResultException(m_statusCode); + } + + public ValueTask> GetEndpointsAsync( + Uri discoveryUri, + EndpointConfiguration configuration, + ITelemetryContext telemetry, + CancellationToken ct) + { + throw new ServiceResultException(m_statusCode); + } + + private readonly StatusCode m_statusCode; + } } } diff --git a/tests/Opc.Ua.Client.Tests/Session/SessionTests.cs b/tests/Opc.Ua.Client.Tests/Session/SessionTests.cs index 27c70ee61a..58f129ef0d 100644 --- a/tests/Opc.Ua.Client.Tests/Session/SessionTests.cs +++ b/tests/Opc.Ua.Client.Tests/Session/SessionTests.cs @@ -2827,6 +2827,7 @@ namespace Opc.Ua.Client.Tests typeof(bool), typeof(SessionClient), typeof(bool), + typeof(bool), typeof(bool) ], null); @@ -2835,7 +2836,7 @@ namespace Opc.Ua.Client.Tests var task = (Task?)method!.Invoke( session, - [endpoint, null, channel, null, CancellationToken.None, false, requireTokenReuse, null, false, false]); + [endpoint, null, channel, null, CancellationToken.None, false, requireTokenReuse, null, false, false, false]); Assert.That(task, Is.Not.Null); return task!;