Files
UA-.NETStandard/Libraries/Opc.Ua.Client/Session/SessionExtensions.cs
Marc Schier f358078b1c [Client] Central channel manager: ref-counted shared channels, coalesced reconnect, IRetryBudget (#3288) (#3852)
# Description

Introduces a central `IClientChannelManager` that owns client-side
transport channels with reference counting, sharing, coalesced
reconnect, and asynchronous participant notification. Sessions,
discovery clients, and registration clients targeting the same
`ConfiguredEndpoint` (with the same reverse-connect identity) now share
a single underlying `ITransportChannel`; reconnect is transparent to
callers; and the previous `AttachChannel`/`DetachChannel` +
`SessionReconnectHandler` patterns are obsoleted.

## What lands

### Core types (`Stack/Opc.Ua.Core/Stack/Client/`)

- `ManagedChannelKey`, `ChannelState`, `ChannelStateChange`,
`ParticipantReconnectResult`
- `IReconnectParticipant`, `IManagedTransportChannel`,
`IClientChannelManager`
- `IChannelReconnectPolicy` + `ExponentialBackoffChannelReconnectPolicy`
(default mirrors historical `SessionReconnectHandler`: 500 ms → 30 s
exponential)
- `IRetryBudget` + `RetryBudget` — shared retry deadline so the
two-level retry no longer compounds multiplicatively
- `ClientChannelManager` — single non-partial sealed class with
SOLID-extracted internal types under
`Stack/Opc.Ua.Core/Stack/Client/Channels/Internal/`: `ChannelEntry`
(behind `IChannelEntryHost`), `ManagedTransportChannelLease`,
`ClientChannel`, `ClientChannelManagerMetrics` (IMeter),
`ClientChannelManagerCertRotation` (auto `ReconnectAllAsync` on
`CertificateManager` rotation), `ClientChannelManagerDiagnostics`
(ActivitySource + EventSource). Refcount, coalesced reconnect, and
transparent faulted-entry recovery via `SwapFaultedEntryAsync` (swaps a
lease's underlying entry on subsequent `ReconnectAsync` when the prior
cycle Faulted/Closed, preserves participant, back-off via policy
`GetDelay(SwapCount)`)

### Three-state lifecycle gate

`Disconnected` → `TransportConnecting` / `TransportReconnecting` →
`TransportConnectedSessionReactivating` → `Ready` (or `Faulted` /
`Closed`). Only `Ready` releases the service-call gate; the manager
bypasses the gate internally for participant `ActivateSession` traffic
during reactivation via an `AsyncLocal` scope.

### Session integration (`Libraries/Opc.Ua.Client/Session/`)

- New `Session.CreateAsync(IClientChannelManager, …)` overload
- `Session` implements `IReconnectParticipant`; reactivation handler
distinguishes `Reactivated` / `RequiresSessionRecreate` /
`TransientFailure` / `FatalForParticipant` / `FatalForChannel`
- `Session.ReconnectAsync(ct)` automatically delegates to the channel
manager when wired
- `Session.RecreateInPlaceAsync` swaps managed-channel leases for
failover-to-different-endpoint via a new atomic
`IClientChannelManager.GetAsync(endpoint, participantFactory, …)`
overload; same-key recreates delegate to manager reconnect;
explicit-channel/connection callers retain legacy behavior
- KA failure routes through
`IClientChannelManager.ReconnectAsync(channel)`

### ManagedSession integration

- `WithChannelManager(...)` builder + ctor param; DI-resolved
automatically
- Two-level retry: channel-mgr handles transparent reconnect; outer
`ConnectionStateMachine` + `IReconnectPolicy` only triggers on terminal
channel `Faulted`. Outer-state churn suppressed during channel-mgr
cycles via `_channelReconnectInProgress` flag wired through
`IManagedTransportChannel.StateChanged`
- New `ManagedSession.ChannelStateChanged` event surfaces transparent
reconnects
- `ConnectionStateChangedEventArgs.UnderlyingChannelState` populated
when outer reconnect was triggered by a channel fault
- Shared `IRetryBudget` propagated so the layers enforce a single
max-total-time

### Discovery / Registration / GDS

Channel-manager-aware `CreateAsync` overloads on `DiscoveryClient`,
`RegistrationClient`, `ServerPushConfigurationClient`,
`LocalDiscoveryServerClient`, `GlobalDiscoveryServerClient`. Each
registers a minimal `IReconnectParticipant`; co-located sessions and
discovery probes targeting the same endpoint share a channel.

### HTTPS transport (`Stack/Opc.Ua.Bindings.Https/`)

- New `IOpcUaHttpClientFactory` abstraction
- HTTPS / OPC-HTTPS channels migrated to `IHttpClientFactory` +
`Microsoft.Extensions.Http.Resilience` (10.6.0; AOT-compatible on net10
via source generators; legacy direct-HTTP path retained for older TFMs /
no-DI consumers)

### DI

`AddOpcUa().AddClient(...)` registers `IClientChannelManager` and the
named HTTPS `HttpClient` + standard resilience handler automatically.
`ManagedSessionBuilder` resolves the manager from DI when present.

### Obsoletions (functional, `[Obsolete]` with guidance)

- `IClientBase.AttachChannel` / `DetachChannel`
- `SessionReconnectHandler`
- `SessionExtensions.ReconnectAsync(connection, ct)` and `(channel, ct)`
legacy overloads

### Tests

- Core: `ClientChannelManagerManagedTests`,
`ClientChannelManagerCertRotationTests`, `RetryBudgetTests`,
`OpcUaHttpClientFactoryTests` — **56 / 56 pass**
- Client: `ManagedSession*Tests` updated; legacy
`SessionReconnectHandlerTests` + `SessionExtensionsTests` preserved with
`#pragma warning disable CS0618` — **129 / 129 pass**
- Sessions integration: `ChannelManagerSharing` / `TransparentReconnect`
/ `SessionLifecycle` / `CertRotation` — **7 pass, 1 skipped**
(`ChannelManagerExhaustionEscalatesAndRecoversWhenServerReturns`
un-`[Ignore]`d and implemented in this PR via `SwapFaultedEntryAsync`;
remaining skip is end-to-end `RequiresSessionRecreate` plumbing —
carry-forward)
- Stress (consolidated `Tests/Opc.Ua.Stress.Tests/Subscriptions/` +
`Channels/`): chaos / contract / soak / lifecycle / leak-accuracy / gap
/ participant-result-aggregation tests merged from the
previously-separate `Opc.Ua.Channels.Stress.Tests` project; nightly run
via `.github/workflows/stress-test.yml`

Builds clean (0 warnings / 0 errors) across all six TFMs: `net472`,
`net48`, `netstandard2.1`, `net8.0`, `net9.0`, `net10.0`.

### Docs

- `Docs/Sessions.md` new § 4 documents the channel manager, three-state
gating, two-level retry with shared `IRetryBudget`, HTTPS resilience
layering
- `Docs/MigrationGuide.md` migration recipes for the four obsoleted API
groups
- `Docs/DependencyInjection.md` channel-manager and HTTPS factory
registration

## Carry-forward (separate PRs)

- End-to-end `ParticipantReconnectResult.RequiresSessionRecreate` →
`Session.RecreateAsync` lifecycle wiring (currently incomplete — flagged
in the 1 remaining skipped integration test)

## Related Issues

- Fixes #3288
- Builds on #3282 (PR that introduced the initial
`ITransportChannelManager` skeleton)

## 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.
- [x] 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.
*(Channel-manager-specific tests + ManagedSession/client tests verified
across all 6 TFMs; full UA.slnx run is pending CI.)*
- [ ] I fixed **all** failing and flaky tests in the CI pipelines and
**all** CodeQL warnings.
- [ ] I have addressed **all** PR feedback received.
2026-06-13 16:34:48 +02:00

443 lines
17 KiB
C#

/* ========================================================================
* Copyright (c) 2005-2025 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.Generic;
using System.IO;
using System.Threading;
using System.Threading.Tasks;
namespace Opc.Ua.Client
{
/// <summary>
/// Extensions to ISession that are not dependent on anything internal
/// to the Session but layer over ISession
/// </summary>
public static class SessionExtensions
{
/// <summary>
/// Establishes a session with the server.
/// </summary>
/// <param name="session">session to use</param>
/// <param name="sessionName">The name to assign to the session.</param>
/// <param name="identity">The user identity.</param>
/// <param name="ct">The cancellation token.</param>
public static Task OpenAsync(
this ISession session,
string sessionName,
IUserIdentity identity,
CancellationToken ct = default)
{
return session.OpenAsync(sessionName, 0, identity, default, ct);
}
/// <summary>
/// Establishes a session with the server.
/// </summary>
/// <param name="session">session to use</param>
/// <param name="sessionName">The name to assign to the session.</param>
/// <param name="sessionTimeout">The session timeout.</param>
/// <param name="identity">The user identity.</param>
/// <param name="preferredLocales">The list of preferred locales.</param>
/// <param name="ct">The cancellation token.</param>
public static Task OpenAsync(
this ISession session,
string sessionName,
uint sessionTimeout,
IUserIdentity identity,
ArrayOf<string> preferredLocales,
CancellationToken ct = default)
{
return session.OpenAsync(
sessionName,
sessionTimeout,
identity,
preferredLocales,
true,
ct);
}
/// <summary>
/// Establishes a session with the server.
/// </summary>
/// <param name="session">session to use</param>
/// <param name="sessionName">The name to assign to the session.</param>
/// <param name="sessionTimeout">The session timeout.</param>
/// <param name="identity">The user identity.</param>
/// <param name="preferredLocales">The list of preferred locales.</param>
/// <param name="checkDomain">If set to <c>true</c> then the
/// domain in the certificate must match the endpoint used.</param>
/// <param name="ct">The cancellation token.</param>
public static Task OpenAsync(
this ISession session,
string sessionName,
uint sessionTimeout,
IUserIdentity identity,
ArrayOf<string> preferredLocales,
bool checkDomain,
CancellationToken ct = default)
{
return session.OpenAsync(
sessionName,
sessionTimeout,
identity,
preferredLocales,
checkDomain,
true,
ct);
}
/// <summary>
/// Reconnects to the server after a network failure.
/// Uses the current channel if possible or creates
/// a new one.
/// </summary>
public static Task ReconnectAsync(
this ISession session,
CancellationToken ct = default)
{
return session.ReconnectAsync(null, null, ct);
}
/// <summary>
/// Reconnects to the server on a waiting connection.
/// </summary>
public static Task ReconnectAsync(
this ISession session,
ITransportWaitingConnection connection,
CancellationToken ct = default)
{
return session.ReconnectAsync(connection, null, ct);
}
/// <summary>
/// Reconnects to the server after a network failure
/// using a new channel.
/// </summary>
public static Task ReconnectAsync(
this ISession session,
ITransportChannel channel,
CancellationToken ct = default)
{
return session.ReconnectAsync(null, channel, ct);
}
/// <summary>
/// Saves all the subscriptions of the session.
/// </summary>
/// <param name="session">session to use</param>
/// <param name="filePath">The file path.</param>
/// <param name="knownTypes">Known types</param>
public static void Save(
this ISession session,
string filePath,
IEnumerable<Type>? knownTypes = null)
{
session.Save(filePath, session.Subscriptions, knownTypes);
}
/// <summary>
/// Load the list of subscriptions saved in a file.
/// </summary>
/// <param name="session">session to use</param>
/// <param name="filePath">The file path.</param>
/// <param name="transferSubscriptions">Load the subscriptions for transfer
/// after load.</param>
/// <param name="knownTypes">Additional known types that may be needed to
/// read the saved subscriptions.</param>
/// <returns>The list of loaded subscriptions</returns>
public static IEnumerable<Subscription> Load(
this ISession session,
string filePath,
bool transferSubscriptions = false,
IEnumerable<Type>? knownTypes = null)
{
using FileStream stream = File.OpenRead(filePath);
return session.Load(stream, transferSubscriptions, knownTypes);
}
/// <summary>
/// Saves a set of subscriptions to a file.
/// </summary>
public static void Save(
this ISession session,
string filePath,
IEnumerable<Subscription> subscriptions,
IEnumerable<Type>? knownTypes = null)
{
using var stream = new FileStream(filePath, FileMode.Create);
session.Save(stream, subscriptions, knownTypes);
}
/// <summary>
/// Close the session with the server and optionally closes the channel.
/// </summary>
public static Task<StatusCode> CloseAsync(
this ISession session,
bool closeChannel,
CancellationToken ct = default)
{
return session.CloseAsync(session.KeepAliveInterval, closeChannel, ct);
}
/// <summary>
/// Disconnects from the server and frees any network resources (closes
/// the channel) with the specified timeout.
/// </summary>
public static Task<StatusCode> CloseAsync(
this ISession session,
int timeout,
CancellationToken ct = default)
{
return session.CloseAsync(timeout, true, ct);
}
/// <summary>
/// Reads a byte string which is too large for the (server side) encoder to handle.
/// </summary>
/// <param name="session">session to use</param>
/// <param name="nodeId">The node id of a byte string variable</param>
/// <param name="ct">Cancellation token to cancel operation with</param>
/// <exception cref="ServiceResultException"></exception>
public static async Task<ByteString> ReadByteStringInChunksAsync(
this ISession session,
NodeId nodeId,
CancellationToken ct = default)
{
int maxByteStringLength = (int)session.ServerCapabilities.MaxByteStringLength;
if (maxByteStringLength <= 1)
{
throw ServiceResultException.Create(
StatusCodes.BadIndexRangeNoData,
"The MaxByteStringLength is not known or too small for reading data in chunks.");
}
return await session.ReadBytesAsync(
nodeId,
maxByteStringLength,
ct).ConfigureAwait(false);
}
/// <summary>
/// Finds the NodeIds for the components for an instance.
/// </summary>
public static async Task<(ArrayOf<NodeId>, ArrayOf<ServiceResult>)> FindComponentIdsAsync(
this ISession session,
NodeId instanceId,
IList<string> componentPaths,
CancellationToken ct = default)
{
var componentIds = new List<NodeId>();
var errors = new List<ServiceResult>();
// build list of paths to translate.
var pathsToTranslateList = new List<BrowsePath>();
for (int ii = 0; ii < componentPaths.Count; ii++)
{
var pathToTranslate = new BrowsePath
{
StartingNode = instanceId,
RelativePath = RelativePath.Parse(componentPaths[ii], session.TypeTree)
};
pathsToTranslateList.Add(pathToTranslate);
}
var pathsToTranslate = pathsToTranslateList.ToArrayOf();
// translate the paths.
TranslateBrowsePathsToNodeIdsResponse response = await session.TranslateBrowsePathsToNodeIdsAsync(
null,
pathsToTranslate,
ct).ConfigureAwait(false);
ArrayOf<BrowsePathResult> results = response.Results;
ArrayOf<DiagnosticInfo> diagnosticInfos = response.DiagnosticInfos;
ResponseHeader responseHeader = response.ResponseHeader;
// verify that the server returned the correct number of results.
ClientBase.ValidateResponse(results, pathsToTranslate);
ClientBase.ValidateDiagnosticInfos(diagnosticInfos, pathsToTranslate);
for (int ii = 0; ii < componentPaths.Count; ii++)
{
componentIds.Add(NodeId.Null);
errors.Add(ServiceResult.Good);
// process any diagnostics associated with any error.
if (StatusCode.IsBad(results[ii].StatusCode))
{
errors[ii] = new ServiceResult(
results[ii].StatusCode,
ii,
diagnosticInfos,
responseHeader.StringTable);
continue;
}
// Expecting exact one NodeId for a local node.
// Report an error if the server returns anything other than that.
if (results[ii].Targets.Count == 0)
{
errors[ii] = ServiceResult.Create(
StatusCodes.BadTargetNodeIdInvalid,
"Could not find target for path: {0}.",
componentPaths[ii]);
continue;
}
if (results[ii].Targets.Count != 1)
{
errors[ii] = ServiceResult.Create(
StatusCodes.BadTooManyMatches,
"Too many matches found for path: {0}.",
componentPaths[ii]);
continue;
}
if (results[ii].Targets[0].RemainingPathIndex != uint.MaxValue)
{
errors[ii] = ServiceResult.Create(
StatusCodes.BadTargetNodeIdInvalid,
"Cannot follow path to external server: {0}.",
componentPaths[ii]);
continue;
}
if (results[ii].Targets[0].TargetId.IsNull)
{
errors[ii] = ServiceResult.Create(
StatusCodes.BadUnexpectedError,
"Server returned a null NodeId for path: {0}.",
componentPaths[ii]);
continue;
}
if (results[ii].Targets[0].TargetId.IsAbsolute)
{
errors[ii] = ServiceResult.Create(
StatusCodes.BadUnexpectedError,
"Server returned a remote node for path: {0}.",
componentPaths[ii]);
continue;
}
// suitable target found.
componentIds[ii] = ExpandedNodeId.ToNodeId(
results[ii].Targets[0].TargetId,
session.NamespaceUris);
}
return (componentIds, errors);
}
/// <summary>
/// Returns the available encodings for a node
/// </summary>
/// <param name="session">The session to use</param>
/// <param name="variableId">The variable node.</param>
/// <param name="ct">Cancellation token to use to cancel the operation</param>
/// <exception cref="ServiceResultException"></exception>
public static async Task<ArrayOf<ReferenceDescription>> ReadAvailableEncodingsAsync(
this ISession session,
NodeId variableId,
CancellationToken ct = default)
{
if (await session.NodeCache.FindAsync(variableId, ct).ConfigureAwait(false)
is not VariableNode variable)
{
throw ServiceResultException.Create(
StatusCodes.BadNodeIdInvalid,
"NodeId does not refer to a valid variable node.");
}
// no encodings available if there was a problem reading the
// data type for the node.
if (variable.DataType.IsNull)
{
return [];
}
// no encodings for non-structures.
if (!await session.NodeCache.IsTypeOfAsync(
variable.DataType,
DataTypeIds.Structure,
ct).ConfigureAwait(false))
{
return [];
}
// look for cached values.
ArrayOf<INode> encodings = await session.NodeCache.FindAsync(
variableId,
ReferenceTypeIds.HasEncoding,
false,
true,
ct).ConfigureAwait(false);
if (encodings.Count > 0)
{
var references = new List<ReferenceDescription>();
foreach (INode encoding in encodings)
{
var reference = new ReferenceDescription
{
ReferenceTypeId = ReferenceTypeIds.HasEncoding,
IsForward = true,
NodeId = encoding.NodeId,
NodeClass = encoding.NodeClass,
BrowseName = encoding.BrowseName,
DisplayName = encoding.DisplayName,
TypeDefinition = encoding.TypeDefinitionId
};
references.Add(reference);
}
return references;
}
var browser = new Browser(session, new BrowserOptions
{
BrowseDirection = BrowseDirection.Forward,
ReferenceTypeId = ReferenceTypeIds.HasEncoding,
IncludeSubtypes = false,
NodeClassMask = 0
});
return await browser.BrowseAsync(variable.DataType, ct).ConfigureAwait(false);
}
}
}