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<NodeId, StatusCode> 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 <marcschier@users.noreply.github.com>
This commit is contained in:
Wolfgang Colsman
2026-10-23 05:58:57 +02:00
committed by GitHub
co-authored by Marc Schier
parent be9ccf912c
commit 97cc145811
15 changed files with 2149 additions and 61 deletions
@@ -44,5 +44,13 @@ namespace Opc.Ua.Client
/// <see cref="RedundancySupport.None"/> when the server is not redundant.
/// </summary>
RedundancySupport RedundancySupport { get; }
/// <summary>
/// The server redundancy snapshot of the last refresh: the redundancy mode, the <c>ServiceLevel</c> 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. <see langword="null"/> without a redundancy
/// handler or before the first refresh.
/// </summary>
ServerRedundancyInfo? RedundancyInfo { get; }
}
}
@@ -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
{
/// <summary>
/// Client side failover across the members of a non-transparent redundant
/// Server set (OPC 10000-4 §6.6.2).
/// </summary>
public partial class ManagedSession
{
/// <summary>
/// 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. <see cref="ConnectionStateChanged"/>
/// reports <see cref="ConnectionState.Failover"/> and then
/// <see cref="ConnectionState.Connected"/>.
/// </summary>
/// <remarks>
/// When the failover fails the session transitions to
/// <see cref="ConnectionState.Reconnecting"/> and the reconnect policy
/// restores it. Cancelling <paramref name="ct"/> stops waiting; a
/// failover in progress completes.
/// </remarks>
/// <param name="target">
/// The endpoint to fail over to, or <see langword="null"/> to fail over
/// to the target the redundancy handler selects from a refreshed
/// redundancy snapshot.
/// </param>
/// <param name="ct">Cancellation token.</param>
/// <exception cref="ServiceResultException">
/// <see cref="StatusCodes.BadNotSupported"/> when no redundancy handler
/// is configured, <see cref="StatusCodes.BadInvalidState"/> when the
/// session is not connected, <see cref="StatusCodes.BadNothingToDo"/>
/// when the handler selects no target, or the error of the failed
/// failover.
/// </exception>
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<ConfiguredEndpoint?> 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);
}
/// <summary>
/// Runs a failover to <paramref name="target"/> in the Failover state of
/// the state machine and waits until the session is connected again.
/// </summary>
private async Task RunRequestedFailoverAsync(ConfiguredEndpoint target, CancellationToken ct)
{
var outcome = new TaskCompletionSource<ServiceResult>(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<ServiceResult> 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);
}
}
/// <summary>
/// Reads the <c>ServiceLevel</c> of the Server in use at
/// <paramref name="interval"/> and fails over when the redundancy
/// handler decides that a failover is warranted.
/// </summary>
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);
}
/// <summary>
/// Reads the <c>ServiceLevel</c> of the Server in use.
/// </summary>
/// <returns>
/// <see langword="true"/> when the read succeeds and the value is in the
/// Healthy subrange.
/// </returns>
private async Task<bool> 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;
}
}
/// <summary>
/// Switches the session to <paramref name="target"/> by recreating it in
/// place, so subscriptions are transferred or recreated on the target.
/// With <paramref name="retirePreviousSession"/>, a session newly created
/// on the target closes the previous session on the Server in use.
/// </summary>
private async Task<ServiceResult> 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;
}
/// <summary>
/// After the initial connect, switches to the member of the redundant
/// Server set with the highest probed <c>ServiceLevel</c> when it is
/// higher than the <c>ServiceLevel</c> of the Server connected to.
/// </summary>
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);
}
/// <summary>
/// Selects the running member with a read <c>ServiceLevel</c> that is
/// higher than the <c>ServiceLevel</c> of the Server in use.
/// </summary>
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;
}
/// <summary>
/// Reads <c>ServiceLevel</c> and <c>ServerStatus.State</c> of every
/// other member with a resolved endpoint.
/// </summary>
private async Task<ServerRedundancyInfo> 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<RedundantServer>[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<RedundantServer> 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<DataValue> values, ArrayOf<ServiceResult> 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<byte>(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);
}
}
}
+82 -34
View File
@@ -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));
}
}
/// <summary>
@@ -583,6 +592,9 @@ namespace Opc.Ua.Client
/// <inheritdoc/>
public RedundancySupport RedundancySupport => m_redundancyInfo?.Mode ?? RedundancySupport.None;
/// <inheritdoc/>
public ServerRedundancyInfo? RedundancyInfo => Volatile.Read(ref m_redundancyInfo);
/// <inheritdoc/>
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
}
}
/// <summary>
/// Resolves the endpoints of the members in the background and, with
/// <paramref name="probeMembers"/>, reads their ServiceLevel. Probing
/// only runs for a selection or failover decision, never for a
/// routine refresh.
/// </summary>
private Task<ServerRedundancyInfo> 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
/// <summary>
/// Recreates the session in place, against a copy of
/// <paramref name="endpoint"/> 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
/// <paramref name="retirePreviousSession"/>, a newly created session
/// closes the session it replaces on the previous Server.
/// </summary>
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<ServerRedundancyInfo>? m_redundancyRefreshTask;
private Task<ServerRedundancyInfo>? 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);
}
}
@@ -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<ServiceResult>? m_requestedFailoverOutcome;
private readonly Queue<ConnectionStateChangedEventArgs> m_stateChanges = new();
private bool m_dispatchingStateChanges;
@@ -380,6 +382,49 @@ namespace Opc.Ua.Client
return true;
}
/// <summary>
/// Starts a failover of the connected session that runs
/// <paramref name="operation"/> in the <see cref="ConnectionState.Failover"/>
/// state. When the operation fails the machine transitions to
/// <see cref="ConnectionState.Reconnecting"/>, so the reconnect policy
/// restores the session.
/// </summary>
/// <param name="operation">The failover operation.</param>
/// <param name="outcome">
/// Completed with the result of the failover including the subscription
/// recovery that follows <paramref name="operation"/>, or cancelled when
/// the failover does not run to completion.
/// </param>
/// <returns><see langword="false"/> when the session is not connected.</returns>
internal bool RequestFailover(
ConnectionStateBudgetOperation operation,
TaskCompletionSource<ServiceResult>? 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;
}
/// <summary>
/// Request closing the session.
/// </summary>
@@ -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<ServiceResult>? 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.
/// </summary>
private async Task<ServiceResult> 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);
@@ -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;
}
/// <summary>
/// Resolves <paramref name="serverUri"/> like <see cref="ResolveAsync"/> 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.
/// </summary>
/// <exception cref="ArgumentException"><paramref name="serverUri"/> is null or empty.</exception>
/// <exception cref="ArgumentNullException"><paramref name="currentEndpoint"/> is null.</exception>
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<ConfiguredEndpoint?> 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<ConfiguredEndpoint?> 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<string, ApplicationDescription> m_discoveredServers =
new(StringComparer.Ordinal);
/// <summary>
/// The first discovery failure of one resolution.
/// </summary>
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)
};
}
}
}
/// <summary>
@@ -135,6 +135,12 @@ namespace Opc.Ua.Client
await session.ReadValuesAsync(nodeIds, ct).ConfigureAwait(false);
ct.ThrowIfCancellationRequested();
var readStatuses = new Dictionary<NodeId, StatusCode>(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<ConfiguredEndpoint?> 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<string> ReadCurrentServerIdAsync(
ISession session,
Dictionary<NodeId, StatusCode> 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;
@@ -49,9 +49,16 @@ namespace Opc.Ua.Client
public byte ServiceLevel { get; init; }
/// <summary>
/// Gets a value indicating whether <see cref="ServiceLevel"/> was read from the Server.
/// Gets a value indicating whether <see cref="ServiceLevel"/> was reported for the Server, either in
/// the <c>RedundantServerArray</c> of the Server in use or by reading the Server's own
/// <c>ServiceLevel</c>.
/// </summary>
internal bool ServiceLevelKnown { get; init; } = true;
/// <remarks>
/// A Server known only from <c>ServerUriArray</c> reports <see cref="ServiceLevels.NoData"/> and
/// <see langword="false"/> until its <c>ServiceLevel</c> is read, see
/// <see cref="ServerRedundancyOptions.ProbeMemberServiceLevel"/>.
/// </remarks>
public bool ServiceLevelKnown { get; init; } = true;
/// <summary>
/// The current state of the Server.
@@ -62,5 +69,11 @@ namespace Opc.Ua.Client
/// The resolved endpoint for this redundant server.
/// </summary>
public ConfiguredEndpoint? Endpoint { get; init; }
/// <summary>
/// The error that prevented resolving <see cref="Endpoint"/> or reading the Server's
/// <c>ServiceLevel</c> during the last refresh, or <see langword="null"/> when no attempt failed.
/// </summary>
public ServiceResult? ResolutionError { get; init; }
}
}
@@ -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.
/// </summary>
public string CurrentServerId { get; init; } = string.Empty;
/// <summary>
/// The status of each Node read from the Server in use to build this snapshot, keyed by its NodeId
/// (for example <c>Server.ServerRedundancy.ServerUriArray</c>). A bad status explains why the
/// corresponding value is empty or defaulted.
/// </summary>
public IReadOnlyDictionary<NodeId, StatusCode> ReadStatuses { get; init; } = s_noReadStatuses;
private static readonly IReadOnlyDictionary<NodeId, StatusCode> s_noReadStatuses =
new ReadOnlyDictionary<NodeId, StatusCode>(new Dictionary<NodeId, StatusCode>());
}
}
@@ -65,15 +65,51 @@ namespace Opc.Ua.Client
public TimeSpan PeerDiscoveryTimeout { get; init; } = DefaultTimeout;
/// <summary>
/// Validates that both bounds are usable.
/// When <see langword="true"/>, a <see cref="ManagedSession"/> reads <c>Server.ServiceLevel</c> and
/// <c>Server.ServerStatus.State</c> 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
/// <c>ServiceLevel</c> (OPC 10000-4 §6.6.2.4.5). After the initial connect the session switches to
/// the member with the highest <c>ServiceLevel</c> when it is higher than the <c>ServiceLevel</c> of
/// the Server it connected to. A member that cannot be probed keeps the values reported for it and
/// carries the failure in <see cref="RedundantServer.ResolutionError"/>. Each probe is bounded by
/// <see cref="PeerDiscoveryTimeout"/>. Default <see langword="false"/>.
/// </summary>
public bool ProbeMemberServiceLevel { get; init; }
/// <summary>
/// When set, a <see cref="ManagedSession"/> reads <c>Server.ServiceLevel</c> of the Server in use at
/// this interval while it is connected to a non-transparent redundant Server set. When the
/// <c>ServiceLevel</c> leaves the Healthy subrange (OPC 10000-4 §6.6.2.4.2) the redundancy snapshot is
/// refreshed, including the member probe when <see cref="ProbeMemberServiceLevel"/> is set, and a
/// failover decision that warrants it starts <see cref="ManagedSession.FailoverAsync"/> to the target
/// the redundancy handler selects. Default <see langword="null"/>: the session fails over only when
/// the reconnect policy gives up or the application calls <see cref="ManagedSession.FailoverAsync"/>.
/// </summary>
public TimeSpan? ServiceLevelMonitoringInterval { get; init; }
/// <summary>
/// Validates that both bounds and the monitoring interval are usable.
/// </summary>
/// <exception cref="ArgumentOutOfRangeException">
/// Thrown when a bound is neither positive nor <see cref="Timeout.InfiniteTimeSpan"/>.
/// Thrown when a bound is neither positive nor <see cref="Timeout.InfiniteTimeSpan"/>, or when
/// <see cref="ServiceLevelMonitoringInterval"/> is set and not positive.
/// </exception>
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)
@@ -397,6 +397,96 @@ namespace Opc.Ua.Client
private sealed record PendingSubscriptionRecovery(NodeId PreviousSessionId, bool ReusedSession);
/// <summary>
/// A session this client replaced on another channel, still allocated on its Server.
/// </summary>
private sealed record RetiredServerSession(
NodeId SessionId,
NodeId AuthenticationToken,
ITransportChannel Channel);
/// <summary>
/// Closes <paramref name="retired"/> once no subscription recovery is pending, so
/// subscriptions can still be transferred from it.
/// </summary>
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);
}
}
/// <summary>
/// Sends CloseSession for <paramref name="retired"/> on the channel it is bound to
/// and deletes its subscriptions there.
/// </summary>
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();
}
}
/// <summary>
/// Identity fences completion from an earlier recovery on the same lease.
/// </summary>
@@ -472,6 +562,7 @@ namespace Opc.Ua.Client
private IClientChannelManager? m_channelManager;
private IManagedTransportChannel? m_managedChannel;
private PendingSubscriptionRecovery? m_pendingSubscriptionRecovery;
private RetiredServerSession? m_retiredSession;
/// <summary>
/// Single-flight gate for <see cref="CompleteSessionRecoveryAsync"/>: the
+79 -5
View File
@@ -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);
}
/// <summary>
/// Recreate the server-side session in place, optionally over a
/// waiting reverse connection.
/// </summary>
/// <param name="endpoint">Optional replacement endpoint.</param>
/// <param name="budget">The outer reconnect retry budget.</param>
/// <param name="connection">Optional waiting reverse connection.</param>
/// <param name="ct">Cancellation token.</param>
/// <param name="retirePreviousSession">
/// When <see langword="true"/> 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.
/// </param>
/// <exception cref="ArgumentNullException"><paramref name="budget"/> is <c>null</c>.</exception>
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);
}
}
@@ -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<ServiceResult>(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()
{
@@ -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
{
/// <summary>
/// Runs a managed session against a Warm redundant Server set of two
/// in-process reference servers (OPC 10000-4 §6.6.2).
/// </summary>
[TestFixture]
[NonParallelizable]
[Category("Client")]
[Category("Integration")]
public sealed partial class ManagedSessionServerRedundancyTests
{
/// <summary>
/// Without probing, the session stays on the Server it connected to and
/// a member known only from ServerUriArray keeps an unknown ServiceLevel.
/// </summary>
[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);
});
}
/// <summary>
/// With probing, the initial connect switches to the member with the
/// higher ServiceLevel.
/// </summary>
[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<byte>(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.");
}
/// <summary>
/// A member whose resolved endpoint does not keep the security mode and
/// policy of the session in use is not probed.
/// </summary>
[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);
});
}
/// <summary>
/// A probe works on a copy of the resolved endpoint and of the identity in
/// use, so it changes neither.
/// </summary>
[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));
});
}
/// <summary>
/// While the Server in use is Healthy, ServiceLevel monitoring does not
/// probe the other members.
/// </summary>
[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));
});
}
/// <summary>
/// A member that refuses the ServiceLevel read keeps an unknown
/// ServiceLevel and reports the exact status.
/// </summary>
[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));
});
}
/// <summary>
/// A member that refuses the ServerStatus.State read reports the exact
/// status and is not selected.
/// </summary>
[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));
}
/// <summary>
/// A redundancy node the Server refuses reports its exact status.
/// </summary>
[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 }));
});
}
/// <summary>
/// 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.
/// </summary>
[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<ConnectionState>();
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.");
}
/// <summary>
/// FailoverAsync switches a connected session to an explicit target
/// without a connection loss and closes the session on the Server left.
/// </summary>
[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<ConnectionState>();
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<byte>(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));
}
/// <summary>
/// FailoverAsync without a target fails over to the member the
/// redundancy handler selects.
/// </summary>
[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));
}
/// <summary>
/// FailoverAsync requires a redundancy handler.
/// </summary>
[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<ServiceResultException>()
.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<bool> condition, CancellationToken ct)
{
while (!condition())
{
await Task.Delay(20, ct).ConfigureAwait(false);
}
}
private static async Task<bool> EventuallyAsync(Func<bool> 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<ServerRedundancyInfo> 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);
}
}
/// <summary>
/// Two reference servers that publish each other in ServerUriArray as a
/// Warm redundant Server set, and the client configuration to reach them.
/// </summary>
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<RedundantSet> 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<UaManagedSession> 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<string, ConfiguredEndpoint>
{
[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;
}
/// <summary>
/// One in-process reference server with its own ApplicationUri and an
/// explicitly controlled ServiceLevel.
/// </summary>
private sealed class RedundantMember : IAsyncDisposable
{
private RedundantMember(ServerFixture<ReferenceServer> 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!;
/// <summary>
/// The sessions currently allocated on the server.
/// </summary>
public int SessionCount => m_server.CurrentInstance.SessionManager.GetSessions().Count;
/// <summary>
/// The sessions created on the server since it started.
/// </summary>
public int CreatedSessionCount => Volatile.Read(ref m_createdSessions);
public static async Task<RedundantMember> StartAsync(string root, string name)
{
var fixture = new ServerFixture<ReferenceServer>(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);
}
/// <summary>
/// Promotes Server.ServerRedundancy to NonTransparentRedundancyType and
/// publishes Warm redundancy with the given peers.
/// </summary>
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<QualifiedName, NodeId>
{
[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);
}
/// <summary>
/// Makes a variable of the Server object unreadable for every user.
/// </summary>
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<ReferenceServer> m_fixture;
private readonly ReferenceServer m_server;
private readonly Action<byte> m_setServiceLevel;
private int m_createdSessions;
}
/// <summary>
/// Records the endpoint and identity of every session the factory creates
/// without a reverse connection.
/// </summary>
private sealed class RecordingSessionFactory : DefaultSessionFactory
{
public RecordingSessionFactory(ITelemetryContext telemetry)
: base(telemetry)
{
}
public ConcurrentQueue<(ConfiguredEndpoint Endpoint, IUserIdentity? Identity)> Calls { get; } = new();
public override Task<ISession> CreateAsync(
ApplicationConfiguration configuration,
ConfiguredEndpoint endpoint,
bool updateBeforeConnect,
bool checkDomain,
string sessionName,
uint sessionTimeout,
IUserIdentity? identity,
ArrayOf<string> preferredLocales,
CancellationToken ct = default)
{
Calls.Enqueue((endpoint, identity));
return base.CreateAsync(
configuration,
endpoint,
updateBeforeConnect,
checkDomain,
sessionName,
sessionTimeout,
identity,
preferredLocales,
ct);
}
}
/// <summary>
/// Records the source timestamps of the data changes of a subscription.
/// </summary>
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<StatusCode> Errors { get; } = new();
public ValueTask OnDataChangeNotificationAsync(
ISubscription subscription,
uint sequenceNumber,
DateTime publishTime,
ReadOnlyMemory<DataValueChange> notification,
PublishState publishStateMask,
IReadOnlyList<string> 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<EventNotification> notification,
PublishState publishStateMask,
IReadOnlyList<string> 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;
}
/// <summary>
/// Resolves the members of the set to the endpoints read from their
/// discovery endpoints.
/// </summary>
private sealed class MemberEndpointResolver : IRedundantServerEndpointResolver
{
public MemberEndpointResolver(IReadOnlyDictionary<string, ConfiguredEndpoint> endpoints)
{
m_endpoints = endpoints;
}
public ValueTask<ConfiguredEndpoint?> ResolveAsync(
string serverUri,
ConfiguredEndpoint currentEndpoint,
CancellationToken ct = default)
{
return new ValueTask<ConfiguredEndpoint?>(
m_endpoints.TryGetValue(serverUri, out ConfiguredEndpoint? endpoint) ? endpoint : null);
}
private readonly IReadOnlyDictionary<string, ConfiguredEndpoint> m_endpoints;
}
}
}
@@ -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);
}
/// <summary>
/// Verifies that a ServiceLevel monitoring interval must be positive and is off by default.
/// </summary>
[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<ArgumentOutOfRangeException>());
Assert.That(
() => new ServerRedundancyOptions { ServiceLevelMonitoringInterval = Timeout.InfiniteTimeSpan }.Validate(),
Throws.TypeOf<ArgumentOutOfRangeException>());
Assert.That(
() => new ServerRedundancyOptions { ServiceLevelMonitoringInterval = TimeSpan.FromSeconds(1) }.Validate(),
Throws.Nothing);
}
/// <summary>
/// Verifies that a failed endpoint resolution keeps the exact status of the failure.
/// </summary>
[Test]
public async Task FetchRedundancyInfoReportsResolutionErrorOfFailedPeerAsync()
{
m_resolver.Setup(r => r.ResolveAsync(
"urn:refusing",
It.IsAny<ConfiguredEndpoint>(),
It.IsAny<CancellationToken>()))
.ThrowsAsync(new ServiceResultException(StatusCodes.BadCertificateUntrusted));
Mock<ISession> 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));
}
/// <summary>
/// Verifies that a discovery failure of the default resolver is reported instead of BadNotFound.
/// </summary>
[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<ISession> 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));
}
/// <summary>
/// Verifies that the status of every redundancy node read is reported, including refused nodes.
/// </summary>
[Test]
public async Task FetchRedundancyInfoReportsReadStatusOfEachNodeAsync()
{
Mock<ISession> 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));
}
/// <summary>
/// Verifies that a URI-only peer can be selected when its service level is unknown.
/// </summary>
@@ -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<ArrayOf<ApplicationDescription>> FindServersAsync(
Uri discoveryUri,
EndpointConfiguration configuration,
string serverUri,
ITelemetryContext telemetry,
CancellationToken ct)
{
throw new ServiceResultException(m_statusCode);
}
public ValueTask<ArrayOf<EndpointDescription>> GetEndpointsAsync(
Uri discoveryUri,
EndpointConfiguration configuration,
ITelemetryContext telemetry,
CancellationToken ct)
{
throw new ServiceResultException(m_statusCode);
}
private readonly StatusCode m_statusCode;
}
}
}
@@ -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!;