/* ========================================================================
* 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.Concurrent;
using System.Collections.Generic;
using System.Diagnostics;
using System.Diagnostics.Metrics;
using System.Globalization;
using System.Linq;
using System.Reflection;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Time.Testing;
using Moq;
using NUnit.Framework;
using Opc.Ua.Bindings;
using Opc.Ua.Security.Certificates;
using Opc.Ua.Tests;
// CA2000: test code; many disposables are ownership-transferred to test fixtures or short-lived.
#pragma warning disable CA2000
namespace Opc.Ua.Core.Tests.Stack.Client
{
///
/// Unit tests for the managed (sharing, refcount, coalesced
/// reconnect, participant notification) behavior of
/// implementing
/// .
///
[TestFixture]
[Category("Client")]
[SetCulture("en-us")]
[SetUICulture("en-us")]
[Parallelizable]
public sealed class ClientChannelManagerManagedTests
{
private static readonly ICertificateFactory s_factory = DefaultCertificateFactory.Instance;
///
/// How long a test waits for an asynchronous operation it has already
/// unblocked to be observed as complete.
///
///
/// This is a hang detector, not a latency assertion. These tests drive a
/// fake clock, so the work under test finishes in microseconds; what is
/// being waited on is purely the thread-pool scheduling of the
/// continuation chain. On a saturated CI agent the pool injects threads
/// at roughly one per second, so a short budget times out while the
/// operation is merely queued - the channel has already faulted and its
/// completion source has already been signalled. Keep this generous
/// enough that it never fires on a healthy run, and far below the
/// blame-hang timeout so a genuine deadlock still fails the job quickly.
///
private static readonly TimeSpan s_completionTimeout = TimeSpan.FromSeconds(60);
[Test]
public void ChannelKeyEqualityIsValueBased()
{
using Certificate serverCert = s_factory.CreateCertificate("CN=server").CreateForRSA();
ConfiguredEndpoint endpoint1 = GetTestEndpoint(serverCert);
ConfiguredEndpoint endpoint2 = GetTestEndpoint(serverCert);
var k1 = ManagedChannelKey.FromEndpoint(endpoint1);
var k2 = ManagedChannelKey.FromEndpoint(endpoint2);
Assert.That(k1, Is.EqualTo(k2));
Assert.That(k1.GetHashCode(), Is.EqualTo(k2.GetHashCode()));
}
[Test]
public void ChannelKeyDistinguishesReverseFromForward()
{
using Certificate serverCert = s_factory.CreateCertificate("CN=server").CreateForRSA();
ConfiguredEndpoint endpoint = GetTestEndpoint(serverCert);
var forward = ManagedChannelKey.FromEndpoint(endpoint);
var reverse = ManagedChannelKey.FromEndpoint(
endpoint,
reverseConnectionIdentity: new object());
Assert.That(forward, Is.Not.EqualTo(reverse));
}
[Test]
public void ChannelKeyDistinguishesDifferentReverseHandles()
{
using Certificate serverCert = s_factory.CreateCertificate("CN=server").CreateForRSA();
ConfiguredEndpoint endpoint = GetTestEndpoint(serverCert);
var r1 = ManagedChannelKey.FromEndpoint(endpoint, reverseConnectionIdentity: new object());
var r2 = ManagedChannelKey.FromEndpoint(endpoint, reverseConnectionIdentity: new object());
Assert.That(r1, Is.Not.EqualTo(r2));
}
///
/// The HTTPS binary, JSON and OpenAPI endpoints of one server share a
/// URL, security mode, policy and certificate but are served by
/// different transports. Leaving the transport profile out of the key
/// gave all three the single channel of whichever transport happened to
/// be created first.
///
[Test]
public void ChannelKeyDistinguishesTransportProfiles()
{
using Certificate serverCert = s_factory.CreateCertificate("CN=server").CreateForRSA();
ConfiguredEndpoint binary = GetTestEndpoint(serverCert);
binary.Description.EndpointUrl = "https://localhost:4843";
binary.Description.TransportProfileUri = Profiles.HttpsBinaryTransport;
ConfiguredEndpoint json = GetTestEndpoint(serverCert);
json.Description.EndpointUrl = "https://localhost:4843";
json.Description.TransportProfileUri = Profiles.HttpsJsonTransport;
var binaryKey = ManagedChannelKey.FromEndpoint(binary);
var jsonKey = ManagedChannelKey.FromEndpoint(json);
Assert.That(binaryKey, Is.Not.EqualTo(jsonKey));
}
///
/// Two endpoints that name the same transport profile still share a key,
/// so the profile was added to the identity rather than replacing it.
///
[Test]
public void ChannelKeyMatchesForTheSameTransportProfile()
{
using Certificate serverCert = s_factory.CreateCertificate("CN=server").CreateForRSA();
ConfiguredEndpoint first = GetTestEndpoint(serverCert);
first.Description.TransportProfileUri = Profiles.UaTcpTransport;
ConfiguredEndpoint second = GetTestEndpoint(serverCert);
second.Description.TransportProfileUri = Profiles.UaTcpTransport;
var firstKey = ManagedChannelKey.FromEndpoint(first);
var secondKey = ManagedChannelKey.FromEndpoint(second);
Assert.That(firstKey, Is.EqualTo(secondKey));
Assert.That(firstKey.GetHashCode(), Is.EqualTo(secondKey.GetHashCode()));
}
///
/// The seven-argument constructor is kept alongside the profile-aware
/// one. Folding the new argument in as an optional parameter would be
/// source compatible but not binary compatible, and an application
/// compiled against the old signature would fail with a
/// MissingMethodException.
///
[Test]
public void ChannelKeyKeepsTheConstructorWithoutATransportProfile()
{
ConstructorInfo? seven = typeof(ManagedChannelKey).GetConstructor(
[
typeof(string),
typeof(string),
typeof(MessageSecurityMode),
typeof(ByteString),
typeof(int),
typeof(ByteString),
typeof(object)
]);
Assert.That(seven, Is.Not.Null);
var key = (ManagedChannelKey)seven!.Invoke(
[
"opc.tcp://localhost:4840",
SecurityPolicies.None,
MessageSecurityMode.None,
default(ByteString),
0,
default(ByteString),
null
]);
Assert.That(key.TransportProfileUri, Is.EqualTo(Profiles.UaTcpTransport));
}
[Test]
public void ExponentialBackoffPolicyDoublesWithCap()
{
var policy = new ExponentialBackoffChannelReconnectPolicy
{
MinDelay = TimeSpan.FromMilliseconds(100),
MaxDelay = TimeSpan.FromSeconds(1)
};
Assert.That(policy.GetDelay(0).TotalMilliseconds, Is.EqualTo(100));
Assert.That(policy.GetDelay(1).TotalMilliseconds, Is.EqualTo(200));
Assert.That(policy.GetDelay(2).TotalMilliseconds, Is.EqualTo(400));
Assert.That(policy.GetDelay(3).TotalMilliseconds, Is.EqualTo(800));
Assert.That(policy.GetDelay(4).TotalMilliseconds, Is.EqualTo(1000)); // capped
Assert.That(policy.GetDelay(10).TotalMilliseconds, Is.EqualTo(1000));
}
[Test]
public void ExponentialBackoffPolicyReturnsInfiniteWhenExhausted()
{
var policy = new ExponentialBackoffChannelReconnectPolicy
{
MaxAttempts = 3
};
Assert.That(policy.GetDelay(2), Is.Not.EqualTo(Timeout.InfiniteTimeSpan));
Assert.That(policy.GetDelay(3), Is.EqualTo(Timeout.InfiniteTimeSpan));
Assert.That(policy.GetDelay(100), Is.EqualTo(Timeout.InfiniteTimeSpan));
}
[Test]
public async Task GetAsyncReturnsManagedChannelWithMatchingKeyAsync()
{
(ClientChannelManager sut, Certificate serverCert, _) = CreateMockedSut();
try
{
ConfiguredEndpoint endpoint = GetTestEndpoint(serverCert);
var participant = new TestParticipant("p1", endpoint);
IManagedTransportChannel channel = await sut.GetAsync(participant, default)
.ConfigureAwait(false);
Assert.That(channel, Is.Not.Null);
Assert.That(channel.Key, Is.EqualTo(ManagedChannelKey.FromEndpoint(endpoint)));
Assert.That(channel.State, Is.EqualTo(ChannelState.Ready));
Assert.That(channel.Manager, Is.SameAs(sut));
channel.Dispose();
}
finally
{
serverCert.Dispose();
}
}
[Test]
public async Task GetAsyncWithFactoryConstructsParticipantInsideLockAsync()
{
(ClientChannelManager sut, Certificate serverCert, _) = CreateMockedSut();
try
{
ConfiguredEndpoint endpoint = GetTestEndpoint(serverCert);
IManagedTransportChannel? observedLease = null;
TestParticipant? participant = null;
int factoryCalls = 0;
IManagedTransportChannel channel = await sut.GetAsync(
endpoint,
lease =>
{
factoryCalls++;
observedLease = lease;
Assert.That(lease.Key, Is.EqualTo(ManagedChannelKey.FromEndpoint(endpoint)));
Assert.That(lease.Manager, Is.SameAs(sut));
var createdParticipant = new TestParticipant("p1", endpoint);
participant = createdParticipant;
return createdParticipant;
},
reverseConnection: null,
default)
.ConfigureAwait(false);
Assert.That(factoryCalls, Is.EqualTo(1));
Assert.That(channel, Is.SameAs(observedLease));
Assert.That(participant, Is.Not.Null);
Assert.That(channel.State, Is.EqualTo(ChannelState.Ready));
channel.Dispose();
}
finally
{
serverCert.Dispose();
}
}
[Test]
public async Task GetAsyncWithFactoryExceptionPropagates()
{
(ClientChannelManager sut, Certificate serverCert, Mock chMock) = CreateMockedSut();
try
{
ConfiguredEndpoint endpoint = GetTestEndpoint(serverCert);
var expected = new InvalidOperationException("factory failed");
Exception? actual = null;
int factoryCalls = 0;
try
{
_ = await sut.GetAsync(
endpoint,
_ =>
{
factoryCalls++;
throw expected;
},
reverseConnection: null,
default)
.ConfigureAwait(false);
}
catch (Exception ex)
{
actual = ex;
}
Assert.That(actual, Is.SameAs(expected));
Assert.That(factoryCalls, Is.EqualTo(1));
var participant = new TestParticipant("p1", endpoint);
IManagedTransportChannel channel = await sut.GetAsync(participant, default)
.ConfigureAwait(false);
chMock.Verify(c => c.OpenAsync(
It.IsAny(),
It.IsAny(),
It.IsAny()),
Times.Exactly(2));
channel.Dispose();
}
finally
{
serverCert.Dispose();
}
}
[Test]
public async Task TwoParticipantsSameEndpointShareUnderlyingChannelAsync()
{
(ClientChannelManager sut, Certificate serverCert, Mock chMock) = CreateMockedSut();
try
{
ConfiguredEndpoint endpoint = GetTestEndpoint(serverCert);
var p1 = new TestParticipant("p1", endpoint);
var p2 = new TestParticipant("p2", endpoint);
IManagedTransportChannel ch1 = await sut.GetAsync(p1, default).ConfigureAwait(false);
IManagedTransportChannel ch2 = await sut.GetAsync(p2, default).ConfigureAwait(false);
Assert.That(ch1.Key, Is.EqualTo(ch2.Key));
Assert.That(ch1, Is.Not.SameAs(ch2)); // distinct lease wrappers
// underlying transport opened exactly once
chMock.Verify(c => c.OpenAsync(
It.IsAny(),
It.IsAny(),
It.IsAny()),
Times.Once);
ch1.Dispose();
ch2.Dispose();
}
finally
{
serverCert.Dispose();
}
}
[Test]
public async Task ReleasingOneLeaseKeepsChannelAliveForOtherAsync()
{
(ClientChannelManager sut, Certificate serverCert, Mock chMock) = CreateMockedSut();
try
{
ConfiguredEndpoint endpoint = GetTestEndpoint(serverCert);
var p1 = new TestParticipant("p1", endpoint);
var p2 = new TestParticipant("p2", endpoint);
IManagedTransportChannel ch1 = await sut.GetAsync(p1, default).ConfigureAwait(false);
IManagedTransportChannel ch2 = await sut.GetAsync(p2, default).ConfigureAwait(false);
ch1.Dispose();
// underlying channel should still be alive
Assert.That(ch2.State, Is.EqualTo(ChannelState.Ready));
chMock.Verify(c => c.CloseAsync(It.IsAny()), Times.Never);
ch2.Dispose();
// give the close fiber a moment to run
await Task.Delay(100).ConfigureAwait(false);
}
finally
{
serverCert.Dispose();
}
}
[Test]
public async Task SyncDisposeReleasesLeaseRefcountBeforeReturningAsync()
{
(ClientChannelManager sut, Certificate serverCert, Mock chMock) = CreateMockedSut();
try
{
ConfiguredEndpoint endpoint = GetTestEndpoint(serverCert);
var pinned = new TestParticipant("pinned", endpoint);
IManagedTransportChannel pinnedLease = await sut.GetAsync(pinned, default)
.ConfigureAwait(false);
// A session failover swaps its lease with a synchronous Dispose and
// reports the failover complete right after, so the released lease
// must stop counting against the shared channel before Dispose
// returns rather than once a background task gets scheduled.
for (int i = 0; i < 50; i++)
{
var swapped = new TestParticipant("swapped" + i, endpoint);
IManagedTransportChannel swappedLease = await sut.GetAsync(swapped, default)
.ConfigureAwait(false);
Assert.That(GetDiagnostic(sut, pinnedLease.Key).Refcount, Is.EqualTo(2));
swappedLease.Dispose();
ManagedChannelDiagnostic diagnostic = GetDiagnostic(sut, pinnedLease.Key);
Assert.That(diagnostic.Refcount, Is.EqualTo(1), $"iteration {i}");
Assert.That(diagnostic.ParticipantCount, Is.EqualTo(1), $"iteration {i}");
Assert.That(diagnostic.State, Is.EqualTo(ChannelState.Ready), $"iteration {i}");
}
chMock.Verify(c => c.CloseAsync(It.IsAny()), Times.Never);
// The last lease still tears the channel down, off the caller's thread.
pinnedLease.Dispose();
await WaitForMockInvocationAsync(
() => chMock.Verify(c => c.CloseAsync(It.IsAny()), Times.Once))
.ConfigureAwait(false);
}
finally
{
await sut.DisposeAsync().ConfigureAwait(false);
serverCert.Dispose();
}
}
[Test]
public async Task GetAsyncRightAfterLastLeaseDisposeNeverReturnsClosedLeaseAsync()
{
(ClientChannelManager sut, Certificate serverCert, _) = CreateMockedSut();
try
{
ConfiguredEndpoint endpoint = GetTestEndpoint(serverCert);
IManagedTransportChannel lease = await sut.GetAsync(
new TestParticipant("p0", endpoint), default).ConfigureAwait(false);
// Releasing the last lease schedules the teardown in the
// background; a lease acquired before it runs must either keep
// the channel or land on a fresh entry, never on one the
// pending teardown closes afterwards.
for (int i = 1; i <= 50; i++)
{
lease.Dispose();
lease = await sut.GetAsync(
new TestParticipant("p" + i, endpoint), default).ConfigureAwait(false);
await Task.Delay(5).ConfigureAwait(false);
Assert.That(lease.State, Is.EqualTo(ChannelState.Ready), $"iteration {i}");
Assert.That(
GetInternalPropertyValue(GetLeaseEntry(lease), "IsClosing"),
Is.False,
$"iteration {i}");
}
lease.Dispose();
}
finally
{
await sut.DisposeAsync().ConfigureAwait(false);
serverCert.Dispose();
}
}
[Test]
public async Task DeferredTeardownKeepsEntryThatGainedLeaseAsync()
{
(ClientChannelManager sut, Certificate serverCert, Mock chMock) = CreateMockedSut();
try
{
ConfiguredEndpoint endpoint = GetTestEndpoint(serverCert);
IManagedTransportChannel lease = await sut.GetAsync(
new TestParticipant("late", endpoint), default).ConfigureAwait(false);
object entry = GetLeaseEntry(lease);
// The teardown a lease release scheduled runs after another
// lease attached: it must leave the entry alone.
MethodInfo? tearDown = entry.GetType().GetMethod(
"TearDownAsync",
BindingFlags.Instance | BindingFlags.NonPublic);
Assert.That(tearDown, Is.Not.Null);
await ((Task)tearDown!.Invoke(
entry,
[ClientChannelManager.ChannelCloseReason.LeaseReleased, true])!)
.ConfigureAwait(false);
Assert.That(GetEntryState(entry), Is.EqualTo(ChannelState.Ready));
Assert.That(GetInternalPropertyValue(entry, "IsClosing"), Is.False);
Assert.That(GetInternalIntProperty(entry, "RefCount"), Is.EqualTo(1));
chMock.Verify(c => c.CloseAsync(It.IsAny()), Times.Never);
lease.Dispose();
await WaitForMockInvocationAsync(
() => chMock.Verify(c => c.CloseAsync(It.IsAny()), Times.Once))
.ConfigureAwait(false);
}
finally
{
await sut.DisposeAsync().ConfigureAwait(false);
serverCert.Dispose();
}
}
[Test]
[CancelAfter(30_000)]
public async Task WaitForReadyAsyncObservesCancellationOnEntryAndWhileParkedAsync(
CancellationToken testCt)
{
// Regression for #4508: the drain unblocks a parked publish worker by
// cancelling its per-attempt token. That only works while this gate
// honours the token, so a refactor must not be able to make the wait
// token-blind without failing here.
(ClientChannelManager sut, Certificate serverCert, Mock chMock) = CreateMockedSut();
try
{
ConfiguredEndpoint endpoint = GetTestEndpoint(serverCert);
using IManagedTransportChannel channel = await sut.GetAsync(
new TestParticipant("parked", endpoint), testCt).ConfigureAwait(false);
object entry = GetLeaseEntry(channel);
MethodInfo waitForReady = entry.GetType()
.GetMethod("WaitForReadyAsync", [typeof(CancellationToken)])!;
// The channel is not ready, so the gate parks instead of
// completing. Nothing in the test ever makes it ready.
SetPrivateField(entry, "m_state", ChannelState.TransportReconnecting);
entry.GetType()
.GetMethod("ResetReadyGate", BindingFlags.Instance | BindingFlags.NonPublic)!
.Invoke(entry, null);
Assert.That(GetEntryState(entry), Is.EqualTo(ChannelState.TransportReconnecting));
using var parkedSource = new CancellationTokenSource();
var parked = (Task)waitForReady.Invoke(entry, [parkedSource.Token])!;
await Task.Delay(50, testCt).ConfigureAwait(false);
Assert.That(parked.IsCompleted, Is.False,
"the gate must park while the channel is not ready.");
parkedSource.Cancel();
// Bounded so a token-blind refactor fails here quickly instead
// of hanging the suite on a wait that can never complete.
Assert.That(
async () => await parked.WaitAsync(TimeSpan.FromSeconds(10), testCt)
.ConfigureAwait(false),
Throws.InstanceOf()
.And.Not.InstanceOf(),
"a parked caller must observe cancellation.");
// A token already cancelled on entry must be observed too, so the
// abort cannot be lost in the window before the wait begins.
using var enteredSource = new CancellationTokenSource();
enteredSource.Cancel();
var onEntry = (Task)waitForReady.Invoke(entry, [enteredSource.Token])!;
Assert.That(
async () => await onEntry.WaitAsync(TimeSpan.FromSeconds(10), testCt)
.ConfigureAwait(false),
Throws.InstanceOf()
.And.Not.InstanceOf(),
"a caller entering with a cancelled token must observe cancellation.");
SetPrivateField(entry, "m_state", ChannelState.Ready);
}
finally
{
await sut.DisposeAsync().ConfigureAwait(false);
serverCert.Dispose();
}
}
[Test]
public async Task GetAsyncReplacesEntryWhoseTeardownIsReservedAsync()
{
(ClientChannelManager sut, Certificate serverCert, Mock chMock) = CreateMockedSut();
try
{
ConfiguredEndpoint endpoint = GetTestEndpoint(serverCert);
IManagedTransportChannel first = await sut.GetAsync(
new TestParticipant("first", endpoint), default).ConfigureAwait(false);
object closingEntry = GetLeaseEntry(first);
// A teardown reserves the entry before it clears the transport
// and before the state reaches Closed.
SetPrivateField(closingEntry, "m_closing", true);
Assert.That(GetEntryState(closingEntry), Is.EqualTo(ChannelState.Ready));
// The readiness gate must not let a caller through on the stale
// Ready state of an entry that is being torn down.
var ready = (Task)closingEntry.GetType()
.GetMethod("WaitForReadyAsync", [typeof(CancellationToken)])!
.Invoke(closingEntry, [CancellationToken.None])!;
ServiceResultException? notReady = Assert.ThrowsAsync(
async () => await ready.ConfigureAwait(false));
Assert.That(notReady!.StatusCode, Is.EqualTo(StatusCodes.BadSecureChannelClosed));
IManagedTransportChannel second = await sut.GetAsync(
new TestParticipant("second", endpoint), default).ConfigureAwait(false);
object freshEntry = GetLeaseEntry(second);
Assert.That(freshEntry, Is.Not.SameAs(closingEntry));
Assert.That(GetEntryState(freshEntry), Is.EqualTo(ChannelState.Ready));
Assert.That(GetInternalIntProperty(freshEntry, "RefCount"), Is.EqualTo(1));
Assert.That(GetInternalIntProperty(closingEntry, "RefCount"), Is.EqualTo(1));
chMock.Verify(c => c.OpenAsync(
It.IsAny(),
It.IsAny(),
It.IsAny()),
Times.Exactly(2));
second.Dispose();
first.Dispose();
}
finally
{
await sut.DisposeAsync().ConfigureAwait(false);
serverCert.Dispose();
}
}
[Test]
public async Task DiscoveryClientCreateAsyncSharesSessionChannelAndReleasesLeaseAsync()
{
(ClientChannelManager sut, Certificate serverCert, Mock chMock) = CreateMockedSut();
try
{
var endpointConfiguration = new EndpointConfiguration
{
OperationTimeout = 6000
};
ConfiguredEndpoint endpoint = GetNoneSecurityEndpoint(endpointConfiguration);
var sessionParticipant = new TestParticipant("session", endpoint);
IManagedTransportChannel sessionChannel = await sut.GetAsync(sessionParticipant, default)
.ConfigureAwait(false);
DiscoveryClient? discoveryClient = null;
try
{
discoveryClient = await DiscoveryClient.CreateAsync(
sut,
new Uri(endpoint.Description.EndpointUrl!),
endpointConfiguration,
NUnitTelemetryContext.Create(),
ct: default).ConfigureAwait(false);
Assert.That(discoveryClient.TransportChannel, Is.InstanceOf());
var discoveryChannel = (IManagedTransportChannel)discoveryClient.TransportChannel;
Assert.That(discoveryChannel.Key, Is.EqualTo(sessionChannel.Key));
Assert.That(discoveryChannel, Is.Not.SameAs(sessionChannel));
chMock.Verify(c => c.OpenAsync(
It.IsAny(),
It.IsAny(),
It.IsAny()),
Times.Once);
discoveryClient.Dispose();
discoveryClient = null;
chMock.Verify(c => c.CloseAsync(It.IsAny()), Times.Never);
Assert.That(sessionChannel.State, Is.EqualTo(ChannelState.Ready));
}
finally
{
discoveryClient?.Dispose();
sessionChannel.Dispose();
}
// lease.Dispose() is non-blocking; poll for the
// CloseAsync invocation before the strict verify.
await WaitForMockInvocationAsync(
() => chMock.Verify(c => c.CloseAsync(It.IsAny()), Times.Once))
.ConfigureAwait(false);
}
finally
{
serverCert.Dispose();
}
}
[Test]
public async Task RegistrationClientCreateAsyncUsesManagedLeaseAndReleasesItAsync()
{
(ClientChannelManager sut, Certificate serverCert, Mock chMock) = CreateMockedSut();
try
{
ConfiguredEndpoint endpoint = GetNoneSecurityEndpoint(new EndpointConfiguration
{
OperationTimeout = 6000
});
RegistrationClient registrationClient = await RegistrationClient.CreateAsync(
sut,
endpoint.Description,
endpoint.Configuration,
NUnitTelemetryContext.Create(),
ct: default).ConfigureAwait(false);
Assert.That(registrationClient.TransportChannel, Is.InstanceOf());
chMock.Verify(c => c.OpenAsync(
It.IsAny(),
It.IsAny(),
It.IsAny()),
Times.Once);
registrationClient.Dispose();
chMock.Verify(c => c.CloseAsync(It.IsAny()), Times.Once);
}
finally
{
serverCert.Dispose();
}
}
[Test]
public async Task ReconnectAsyncNotifiesAttachedParticipantsAsync()
{
(ClientChannelManager sut, Certificate serverCert, Mock chMock) = CreateMockedSut();
try
{
ConfiguredEndpoint endpoint = GetTestEndpoint(serverCert);
var participant = new TestParticipant("p1", endpoint);
IManagedTransportChannel ch = await sut.GetAsync(participant, default).ConfigureAwait(false);
chMock.Setup(c => c.SupportedFeatures).Returns(TransportChannelFeatures.Reconnect);
chMock.Setup(c => c.ReconnectAsync(
It.IsAny(),
It.IsAny()))
.Returns(new ValueTask());
await sut.ReconnectAsync(ch, default).ConfigureAwait(false);
Assert.That(participant.NotificationCount, Is.GreaterThanOrEqualTo(1));
Assert.That(ch.State, Is.EqualTo(ChannelState.Ready));
ch.Dispose();
}
finally
{
serverCert.Dispose();
}
}
[Test]
public async Task ConcurrentReconnectCallsCoalesceIntoOneCycleAsync()
{
(ClientChannelManager sut, Certificate serverCert, Mock chMock) = CreateMockedSut();
try
{
ConfiguredEndpoint endpoint = GetTestEndpoint(serverCert);
var participant = new TestParticipant("p1", endpoint);
IManagedTransportChannel ch = await sut.GetAsync(participant, default).ConfigureAwait(false);
chMock.Setup(c => c.SupportedFeatures).Returns(TransportChannelFeatures.Reconnect);
int reconnectCalls = 0;
chMock.Setup(c => c.ReconnectAsync(
It.IsAny(),
It.IsAny()))
.Returns(async () =>
{
Interlocked.Increment(ref reconnectCalls);
await Task.Delay(100).ConfigureAwait(false);
});
Task t1 = sut.ReconnectAsync(ch, default).AsTask();
Task t2 = sut.ReconnectAsync(ch, default).AsTask();
Task t3 = sut.ReconnectAsync(ch, default).AsTask();
await Task.WhenAll(t1, t2, t3).ConfigureAwait(false);
Assert.That(reconnectCalls, Is.EqualTo(1),
"Concurrent ReconnectAsync calls should coalesce into a single cycle.");
Assert.That(participant.NotificationCount, Is.EqualTo(1));
ch.Dispose();
}
finally
{
serverCert.Dispose();
}
}
[Test]
public async Task ConcurrentInitialOpenAndReconnectDisposesLosingTransportAsync()
{
var initialOpenStarted = new TaskCompletionSource(
TaskCreationOptions.RunContinuationsAsynchronously);
var releaseInitialOpen = new TaskCompletionSource(
TaskCreationOptions.RunContinuationsAsynchronously);
var reconnectOpenStarted = new TaskCompletionSource(
TaskCreationOptions.RunContinuationsAsynchronously);
var initialChannel = new Mock();
var reconnectChannel = new Mock();
initialChannel
.Setup(channel => channel.OpenAsync(
It.IsAny(),
It.IsAny(),
It.IsAny()))
.Callback(() => initialOpenStarted.TrySetResult(true))
.Returns(() => new ValueTask(releaseInitialOpen.Task));
reconnectChannel
.Setup(channel => channel.OpenAsync(
It.IsAny(),
It.IsAny(),
It.IsAny()))
.Callback(() => reconnectOpenStarted.TrySetResult(true))
.Returns(new ValueTask());
initialChannel
.Setup(channel => channel.CloseAsync(It.IsAny()))
.Returns(new ValueTask());
reconnectChannel
.Setup(channel => channel.CloseAsync(It.IsAny()))
.Returns(new ValueTask());
int createCount = 0;
var bindings = new Mock();
bindings
.Setup(factory => factory.Create(
It.IsAny(),
It.IsAny()))
.Returns(() => Interlocked.Increment(ref createCount) == 1
? initialChannel.Object
: reconnectChannel.Object);
ITelemetryContext telemetry = NUnitTelemetryContext.Create();
var configuration = new ApplicationConfiguration(telemetry);
await using var manager = new ClientChannelManager(
configuration,
telemetry,
bindings.Object);
ConfiguredEndpoint endpoint = GetNoneSecurityEndpoint(
new EndpointConfiguration { OperationTimeout = 6000 });
var entry = new ChannelEntry(
manager,
ManagedChannelKey.FromEndpoint(endpoint),
endpoint,
reverseConnection: null);
ManagedTransportChannelLease? lease = null;
try
{
Task initialOpen = entry.OpenInitialAsync(
clientCertificate: null,
clientCertificateChain: null,
clientCertificateVersion: 0,
default);
await initialOpenStarted.Task.WaitAsync(s_completionTimeout)
.ConfigureAwait(false);
lease = entry.AcquireLease(new TestParticipant("p1", endpoint));
Task reconnect = entry.RequestReconnectAsync(default);
await reconnectOpenStarted.Task.WaitAsync(s_completionTimeout)
.ConfigureAwait(false);
Assert.That(
await reconnect.WaitAsync(s_completionTimeout).ConfigureAwait(false),
Is.True);
releaseInitialOpen.TrySetResult(true);
await initialOpen.WaitAsync(s_completionTimeout).ConfigureAwait(false);
Assert.That(entry.Underlying, Is.SameAs(reconnectChannel.Object));
initialChannel.Verify(
channel => channel.CloseAsync(It.IsAny()),
Times.Once);
initialChannel.Verify(channel => channel.Dispose(), Times.Once);
reconnectChannel.Verify(channel => channel.Dispose(), Times.Never);
}
finally
{
releaseInitialOpen.TrySetResult(true);
if (lease != null)
{
await lease.CloseAsync(default).ConfigureAwait(false);
}
await entry.DisposeAsync().ConfigureAwait(false);
}
}
///
/// The caller that creates a shared entry holds no lease while it opens
/// the transport. A participant that attached meanwhile and released its
/// lease closed the entry under that open, which failed the creator with
/// BadSecureChannelClosed instead of retrying on a fresh entry.
///
[Test]
public async Task CreatorRetriesWhenLeaseReleasedDuringInitialOpenAsync()
{
var initialOpenStarted = new TaskCompletionSource(
TaskCreationOptions.RunContinuationsAsynchronously);
var releaseInitialOpen = new TaskCompletionSource(
TaskCreationOptions.RunContinuationsAsynchronously);
var initialChannel = new Mock();
var retryChannel = new Mock();
initialChannel
.Setup(channel => channel.OpenAsync(
It.IsAny(),
It.IsAny(),
It.IsAny()))
.Callback(() => initialOpenStarted.TrySetResult(true))
.Returns(() => new ValueTask(releaseInitialOpen.Task));
retryChannel
.Setup(channel => channel.OpenAsync(
It.IsAny(),
It.IsAny(),
It.IsAny()))
.Returns(new ValueTask());
initialChannel
.Setup(channel => channel.CloseAsync(It.IsAny()))
.Returns(new ValueTask());
retryChannel
.Setup(channel => channel.CloseAsync(It.IsAny()))
.Returns(new ValueTask());
int createCount = 0;
var bindings = new Mock();
bindings
.Setup(factory => factory.Create(
It.IsAny(),
It.IsAny()))
.Returns(() => Interlocked.Increment(ref createCount) == 1
? initialChannel.Object
: retryChannel.Object);
ITelemetryContext telemetry = NUnitTelemetryContext.Create();
var configuration = new ApplicationConfiguration(telemetry);
await using var manager = new ClientChannelManager(
configuration,
telemetry,
bindings.Object);
ConfiguredEndpoint endpoint = GetNoneSecurityEndpoint(
new EndpointConfiguration { OperationTimeout = 6000 });
using var timeout = new CancellationTokenSource(s_completionTimeout);
try
{
ValueTask creating = manager.GetAsync(
new TestParticipant("creator", endpoint), timeout.Token);
await initialOpenStarted.Task.WaitAsync(s_completionTimeout)
.ConfigureAwait(false);
IManagedTransportChannel attached = await manager.GetAsync(
new TestParticipant("attached", endpoint), timeout.Token).ConfigureAwait(false);
object openingEntry = GetLeaseEntry(attached);
await attached.CloseAsync(timeout.Token).ConfigureAwait(false);
Assert.That(GetInternalPropertyValue(openingEntry, "IsClosing"), Is.True);
releaseInitialOpen.TrySetResult(true);
using IManagedTransportChannel lease = await creating.AsTask()
.WaitAsync(s_completionTimeout).ConfigureAwait(false);
Assert.That(GetLeaseEntry(lease), Is.Not.SameAs(openingEntry));
Assert.That(lease.State, Is.EqualTo(ChannelState.Ready));
Assert.That(createCount, Is.EqualTo(2));
initialChannel.Verify(
channel => channel.CloseAsync(It.IsAny()),
Times.Once);
}
finally
{
releaseInitialOpen.TrySetResult(true);
}
}
///
/// A reconnect that starts while the initial open is in flight moves the
/// entry out of its initial states before any transport is installed. A
/// participant waiting for the transport must keep waiting until the
/// reconnect installs one, not read the entry while it has none.
///
[Test]
public async Task WaitForTransportWaitsForReconnectStartedDuringInitialOpenAsync()
{
var releaseInitialOpen = new TaskCompletionSource(
TaskCreationOptions.RunContinuationsAsynchronously);
var initialOpenStarted = new TaskCompletionSource(
TaskCreationOptions.RunContinuationsAsynchronously);
var reconnectOpenStarted = new TaskCompletionSource(
TaskCreationOptions.RunContinuationsAsynchronously);
var releaseReconnectOpen = new TaskCompletionSource(
TaskCreationOptions.RunContinuationsAsynchronously);
var initialChannel = new Mock();
var reconnectChannel = new Mock();
initialChannel
.Setup(channel => channel.OpenAsync(
It.IsAny(),
It.IsAny(),
It.IsAny()))
.Callback(() => initialOpenStarted.TrySetResult(true))
.Returns(() => new ValueTask(releaseInitialOpen.Task));
reconnectChannel
.Setup(channel => channel.OpenAsync(
It.IsAny(),
It.IsAny(),
It.IsAny()))
.Callback(() => reconnectOpenStarted.TrySetResult(true))
.Returns(() => new ValueTask(releaseReconnectOpen.Task));
initialChannel
.Setup(channel => channel.CloseAsync(It.IsAny()))
.Returns(new ValueTask());
reconnectChannel
.Setup(channel => channel.CloseAsync(It.IsAny()))
.Returns(new ValueTask());
int createCount = 0;
var bindings = new Mock();
bindings
.Setup(factory => factory.Create(
It.IsAny(),
It.IsAny()))
.Returns(() => Interlocked.Increment(ref createCount) == 1
? initialChannel.Object
: reconnectChannel.Object);
ITelemetryContext telemetry = NUnitTelemetryContext.Create();
var configuration = new ApplicationConfiguration(telemetry);
await using var manager = new ClientChannelManager(
configuration,
telemetry,
bindings.Object);
ConfiguredEndpoint endpoint = GetNoneSecurityEndpoint(
new EndpointConfiguration { OperationTimeout = 6000 });
var entry = new ChannelEntry(
manager,
ManagedChannelKey.FromEndpoint(endpoint),
endpoint,
reverseConnection: null);
ManagedTransportChannelLease? lease = null;
try
{
Task initialOpen = entry.OpenInitialAsync(
clientCertificate: null,
clientCertificateChain: null,
clientCertificateVersion: 0,
default);
await initialOpenStarted.Task.WaitAsync(s_completionTimeout)
.ConfigureAwait(false);
lease = entry.AcquireLease(new TestParticipant("p1", endpoint));
Task reconnect = entry.RequestReconnectAsync(default);
await reconnectOpenStarted.Task.WaitAsync(s_completionTimeout)
.ConfigureAwait(false);
Assert.That(entry.State, Is.EqualTo(ChannelState.TransportReconnecting));
Assert.That(entry.Underlying, Is.Null);
Task waiting = entry.WaitForTransportAsync(default);
Assert.That(waiting.IsCompleted, Is.False);
releaseReconnectOpen.TrySetResult(true);
await waiting.WaitAsync(s_completionTimeout).ConfigureAwait(false);
Assert.That(entry.Underlying, Is.SameAs(reconnectChannel.Object));
Assert.That(
await reconnect.WaitAsync(s_completionTimeout).ConfigureAwait(false),
Is.True);
releaseInitialOpen.TrySetResult(true);
await initialOpen.WaitAsync(s_completionTimeout).ConfigureAwait(false);
}
finally
{
releaseInitialOpen.TrySetResult(true);
releaseReconnectOpen.TrySetResult(true);
if (lease != null)
{
await lease.CloseAsync(default).ConfigureAwait(false);
}
await entry.DisposeAsync().ConfigureAwait(false);
}
}
///
/// An initial open that fails on its own fails the creator, even after
/// the entry it faulted moved on to Closed; only a teardown that closed
/// the entry under the open lets the creator retry.
///
[Test]
public async Task CreatorDoesNotRetryInitialOpenThatFaultedAsync()
{
var channel = new Mock();
channel
.Setup(c => c.OpenAsync(
It.IsAny(),
It.IsAny(),
It.IsAny()))
.Returns(() => new ValueTask(Task.FromException(
new ServiceResultException(StatusCodes.BadSecureChannelClosed))));
channel
.Setup(c => c.CloseAsync(It.IsAny()))
.Returns(new ValueTask());
int createCount = 0;
var bindings = new Mock();
bindings
.Setup(factory => factory.Create(
It.IsAny(),
It.IsAny()))
.Returns(() =>
{
Interlocked.Increment(ref createCount);
return channel.Object;
});
ITelemetryContext telemetry = NUnitTelemetryContext.Create();
var configuration = new ApplicationConfiguration(telemetry);
await using var manager = new ClientChannelManager(
configuration,
telemetry,
bindings.Object);
ConfiguredEndpoint endpoint = GetNoneSecurityEndpoint(
new EndpointConfiguration { OperationTimeout = 6000 });
var entry = new ChannelEntry(
manager,
ManagedChannelKey.FromEndpoint(endpoint),
endpoint,
reverseConnection: null);
Assert.That(
async () => await entry.OpenInitialAsync(null, null, 0, default).ConfigureAwait(false),
Throws.TypeOf());
await entry.DisposeAsync().ConfigureAwait(false);
Assert.Multiple(() =>
{
Assert.That(entry.State, Is.EqualTo(ChannelState.Closed));
Assert.That(entry.InitialOpenFaulted, Is.True);
Assert.That(
async () => await entry.WaitForTransportAsync(default).ConfigureAwait(false),
Throws.TypeOf()
.With.Property(nameof(ServiceResultException.StatusCode))
.EqualTo(StatusCodes.BadSecureChannelClosed));
});
createCount = 0;
ServiceResultException? sre = Assert.ThrowsAsync(
async () => await manager.GetAsync(
new TestParticipant("creator", endpoint), default).ConfigureAwait(false));
Assert.That(sre!.StatusCode, Is.EqualTo(StatusCodes.BadSecureChannelClosed));
Assert.That(createCount, Is.EqualTo(1));
}
///
/// Fatal participant verdicts fault the channel without starting another cycle for a bounded caller.
///
[TestCase(false)]
[TestCase(true)]
public async Task FatalForChannelTransitionsToFaultedStateAsync(bool boundedCaller)
{
var time = new FakeTimeProvider();
var policy = new ExponentialBackoffChannelReconnectPolicy
{
MinDelay = TimeSpan.Zero,
MaxDelay = TimeSpan.Zero
};
(ClientChannelManager sut, Certificate serverCert, Mock chMock) =
CreateMockedSut(reconnectPolicy: policy, timeProvider: time);
try
{
ConfiguredEndpoint endpoint = GetTestEndpoint(serverCert);
var participant = new TestParticipant(
"p1", endpoint,
onReconnect: (_, _, _) => ParticipantReconnectResult.FatalForChannel);
IManagedTransportChannel ch = await sut.GetAsync(participant, default).ConfigureAwait(false);
chMock.Setup(c => c.SupportedFeatures).Returns(TransportChannelFeatures.Reconnect);
chMock.Setup(c => c.ReconnectAsync(
It.IsAny(),
It.IsAny()))
.Returns(new ValueTask());
if (boundedCaller)
{
await AssertThrowsAsync(
sut.ReconnectAsync(ch, new RetryBudget(TimeSpan.FromSeconds(1), time)).AsTask(),
s_completionTimeout).ConfigureAwait(false);
}
else
{
await sut.ReconnectAsync(ch, default).ConfigureAwait(false);
}
time.Advance(TimeSpan.FromMinutes(1));
Assert.That(ch.State, Is.EqualTo(ChannelState.Faulted));
Assert.That(GetInternalIntProperty(ch, "SwapCount"), Is.Zero);
chMock.Verify(value => value.ReconnectAsync(
It.IsAny(), It.IsAny()), Times.Once);
ch.Dispose();
}
finally
{
await sut.DisposeAsync().ConfigureAwait(false);
serverCert.Dispose();
}
}
[Test]
public async Task FatalForParticipantDetachesOnlyThatParticipantAsync()
{
(ClientChannelManager sut, Certificate serverCert, Mock chMock) = CreateMockedSut();
try
{
ConfiguredEndpoint endpoint = GetTestEndpoint(serverCert);
var p1 = new TestParticipant(
"p1", endpoint,
onReconnect: (_, _, _) => ParticipantReconnectResult.FatalForParticipant);
var p2 = new TestParticipant("p2", endpoint);
IManagedTransportChannel ch1 = await sut.GetAsync(p1, default).ConfigureAwait(false);
IManagedTransportChannel ch2 = await sut.GetAsync(p2, default).ConfigureAwait(false);
chMock.Setup(c => c.SupportedFeatures).Returns(TransportChannelFeatures.Reconnect);
chMock.Setup(c => c.ReconnectAsync(
It.IsAny(),
It.IsAny()))
.Returns(new ValueTask());
await sut.ReconnectAsync(ch1, default).ConfigureAwait(false);
Assert.That(ch1.State, Is.EqualTo(ChannelState.Ready));
Assert.That(ch2.State, Is.EqualTo(ChannelState.Ready));
// p2 still got the notification
Assert.That(p2.NotificationCount, Is.GreaterThanOrEqualTo(1));
ch1.Dispose();
ch2.Dispose();
}
finally
{
serverCert.Dispose();
}
}
[Test]
public async Task ReconnectAllAsyncTriggersAllEntriesAsync()
{
(ClientChannelManager sut, Certificate serverCert, Mock chMock) = CreateMockedSut();
try
{
ConfiguredEndpoint endpoint = GetTestEndpoint(serverCert);
var p1 = new TestParticipant("p1", endpoint);
IManagedTransportChannel ch = await sut.GetAsync(p1, default).ConfigureAwait(false);
chMock.Setup(c => c.SupportedFeatures).Returns(TransportChannelFeatures.Reconnect);
chMock.Setup(c => c.ReconnectAsync(
It.IsAny(),
It.IsAny()))
.Returns(new ValueTask());
await sut.ReconnectAllAsync(default).ConfigureAwait(false);
Assert.That(p1.NotificationCount, Is.GreaterThanOrEqualTo(1));
Assert.That(ch.State, Is.EqualTo(ChannelState.Ready));
ch.Dispose();
}
finally
{
serverCert.Dispose();
}
}
[Test]
public async Task RebindParticipantSwapsParticipantOnLeaseAsync()
{
(ClientChannelManager sut, Certificate serverCert, _) = CreateMockedSut();
try
{
ConfiguredEndpoint endpoint = GetTestEndpoint(serverCert);
var bootstrap = new TestParticipant("boot", endpoint);
IManagedTransportChannel ch = await sut.GetAsync(bootstrap, default)
.ConfigureAwait(false);
var real = new TestParticipant("real", endpoint);
#pragma warning disable CS0618 // Test verifies the obsolete compatibility shim remains functional.
sut.RebindParticipant(ch, real);
#pragma warning restore CS0618
// No exception means swap succeeded; participant list is internal,
// but downstream OnReconnect tests verify behavior.
ch.Dispose();
}
finally
{
serverCert.Dispose();
}
}
[Test]
[NonParallelizable]
public async Task MetricsAreEmittedForChannelLifetimeAsync()
{
using var metrics = new ChannelMetricListener();
ITelemetryContext telemetry = NUnitTelemetryContext.Create();
(ClientChannelManager sut, Certificate serverCert, _) = CreateMockedSut(telemetry);
try
{
ConfiguredEndpoint endpoint = GetTestEndpoint(serverCert);
string endpointUrl = endpoint.Description.EndpointUrl!;
var p1 = new TestParticipant("p1", endpoint);
var p2 = new TestParticipant("p2", endpoint);
IManagedTransportChannel ch1 = await sut.GetAsync(p1, default).ConfigureAwait(false);
IManagedTransportChannel ch2 = await sut.GetAsync(p2, default).ConfigureAwait(false);
metrics.RecordObservableInstruments();
ch1.Dispose();
ch2.Dispose();
// ch.Dispose() is non-blocking (lease teardown runs on
// the threadpool). The teardown emits the close counter
// *before* the active up/down counter, so waiting on the
// close measurement alone still races the active(-1) one
// that the assertions below require: poll for the last
// measurement the teardown produces instead.
await WaitForMeasurementAsync(
metrics,
"opc.ua.channel.active",
-1,
Tag("endpoint", endpointUrl)).ConfigureAwait(false);
Assert.That(metrics.HasMeasurement(
"opc.ua.channel.open",
Tag("endpoint", endpointUrl),
Tag("reverse", false)), Is.True, metrics.FormatMeasurements());
Assert.That(metrics.HasMeasurement(
"opc.ua.channel.active",
1,
Tag("endpoint", endpointUrl)), Is.True, metrics.FormatMeasurements());
Assert.That(metrics.HasMeasurement(
"opc.ua.channel.active",
-1,
Tag("endpoint", endpointUrl)), Is.True, metrics.FormatMeasurements());
Assert.That(metrics.HasMeasurement(
"opc.ua.channel.refcount",
2,
Tag("endpoint", endpointUrl)), Is.True, metrics.FormatMeasurements());
Assert.That(metrics.HasMeasurement(
"opc.ua.channel.participants",
2,
Tag("endpoint", endpointUrl)), Is.True, metrics.FormatMeasurements());
Assert.That(metrics.HasMeasurement(
"opc.ua.channel.close",
Tag("endpoint", endpointUrl),
Tag("reverse", false),
Tag("reason", "lease-released")), Is.True, metrics.FormatMeasurements());
}
finally
{
await sut.DisposeAsync().ConfigureAwait(false);
serverCert.Dispose();
}
}
[Test]
[NonParallelizable]
public async Task MetricsAreEmittedForReconnectAndGateWaitAsync()
{
using var metrics = new ChannelMetricListener();
ITelemetryContext telemetry = NUnitTelemetryContext.Create();
(ClientChannelManager sut, Certificate serverCert, Mock chMock) = CreateMockedSut(telemetry);
try
{
ConfiguredEndpoint endpoint = GetTestEndpoint(serverCert);
string endpointUrl = endpoint.Description.EndpointUrl!;
var participant = new TestParticipant("p1", endpoint);
IManagedTransportChannel ch = await sut.GetAsync(participant, default).ConfigureAwait(false);
var reconnectEntered = new TaskCompletionSource(
TaskCreationOptions.RunContinuationsAsynchronously);
var allowReconnect = new TaskCompletionSource(
TaskCreationOptions.RunContinuationsAsynchronously);
chMock.Setup(c => c.SupportedFeatures).Returns(TransportChannelFeatures.Reconnect);
chMock.Setup(c => c.ReconnectAsync(
It.IsAny(),
It.IsAny()))
.Returns(() => new ValueTask(WaitForReconnectReleaseAsync(
reconnectEntered,
allowReconnect)));
chMock.Setup(c => c.SendRequestAsync(
It.IsAny(),
It.IsAny()))
.Returns(new ValueTask(new ReadResponse
{
ResponseHeader = new ResponseHeader { ServiceResult = StatusCodes.Good }
}));
Task reconnectTask = sut.ReconnectAsync(ch, default).AsTask();
await reconnectEntered.Task.WaitAsync(s_completionTimeout).ConfigureAwait(false);
Task sendTask = ch.SendRequestAsync(
new ReadRequest { RequestHeader = new RequestHeader() },
default).AsTask();
allowReconnect.SetResult(true);
await Task.WhenAll(reconnectTask, sendTask).ConfigureAwait(false);
// RunReconnectCycleAsync.RecordReconnectDuration is emitted from a
// finally block AFTER tcs.TrySetResult(true) returns, so on a fast
// runner the test thread can race past the assertion before the
// measurement lands. Poll briefly for the histogram measurement.
await WaitForMeasurementAsync(
metrics,
"opc.ua.channel.reconnect.duration",
Tag("endpoint", endpointUrl),
Tag("outcome", "success"))
.ConfigureAwait(false);
Assert.That(metrics.HasMeasurement(
"opc.ua.channel.reconnect.attempts",
Tag("endpoint", endpointUrl),
Tag("outcome", "success")), Is.True, metrics.FormatMeasurements());
Assert.That(metrics.HasMeasurement(
"opc.ua.channel.reconnect.duration",
Tag("endpoint", endpointUrl),
Tag("outcome", "success")), Is.True, metrics.FormatMeasurements());
Assert.That(metrics.HasMeasurement(
"opc.ua.channel.gate.wait",
Tag("endpoint", endpointUrl)), Is.True, metrics.FormatMeasurements());
ch.Dispose();
}
finally
{
await sut.DisposeAsync().ConfigureAwait(false);
serverCert.Dispose();
}
}
private static Task WaitForMeasurementAsync(
ChannelMetricListener metrics,
string instrumentName,
params KeyValuePair[] tags)
{
return WaitForMeasurementAsync(metrics, instrumentName, null, tags);
}
private static async Task WaitForMeasurementAsync(
ChannelMetricListener metrics,
string instrumentName,
double? value,
params KeyValuePair[] tags)
{
const int kMaxPollMs = 2000;
const int kPollIntervalMs = 25;
int elapsed = 0;
while (!metrics.HasMeasurement(instrumentName, value, tags) && elapsed < kMaxPollMs)
{
await Task.Delay(kPollIntervalMs).ConfigureAwait(false);
elapsed += kPollIntervalMs;
}
}
private static async Task WaitForMockInvocationAsync(Action verify)
{
// Used to bridge Moq.Verify against state mutated by fire-
// and-forget tasks (e.g. ManagedTransportChannelLease.Dispose
// which posts the actual underlying CloseAsync onto the
// thread pool). Polls the verify until it stops throwing or
// the budget is exhausted; the final invocation is allowed
// to throw and surface the failure to NUnit.
const int kMaxPollMs = 2000;
const int kPollIntervalMs = 25;
int elapsed = 0;
while (elapsed < kMaxPollMs)
{
try
{
verify();
return;
}
catch (MockException)
{
await Task.Delay(kPollIntervalMs).ConfigureAwait(false);
elapsed += kPollIntervalMs;
}
}
verify();
}
private static async Task WaitForConditionAsync(
Func condition,
string description)
{
// Generic test-side poll for state mutated by fire-and-forget
// tasks. Returns once the condition holds; if the budget is
// exhausted the caller's subsequent assertion is allowed to
// run and surface the failure to NUnit.
const int kMaxPollMs = 2000;
const int kPollIntervalMs = 25;
int elapsed = 0;
while (!condition() && elapsed < kMaxPollMs)
{
await Task.Delay(kPollIntervalMs).ConfigureAwait(false);
elapsed += kPollIntervalMs;
}
_ = description;
}
[Test]
[NonParallelizable]
public async Task ActivitySpanIsRecordedForReconnectAsync()
{
using var listener = new ChannelActivityListener();
ITelemetryContext telemetry = NUnitTelemetryContext.Create();
(ClientChannelManager sut, Certificate serverCert, Mock chMock) = CreateMockedSut(telemetry);
try
{
ConfiguredEndpoint endpoint = GetTestEndpoint(serverCert);
string endpointUrl = endpoint.Description.EndpointUrl!;
var participant = new TestParticipant("p1", endpoint);
IManagedTransportChannel ch = await sut.GetAsync(participant, default).ConfigureAwait(false);
chMock.Setup(c => c.SupportedFeatures).Returns(TransportChannelFeatures.Reconnect);
chMock.Setup(c => c.ReconnectAsync(
It.IsAny(),
It.IsAny()))
.Returns(new ValueTask());
await sut.ReconnectAsync(ch, default).ConfigureAwait(false);
Activity activity = await listener
.WaitForStoppedActivityAsync("OpcUaChannelReconnect")
.ConfigureAwait(false);
var tags = activity.TagObjects.ToDictionary(t => t.Key, t => t.Value);
Assert.That(tags, Does.ContainKey("endpoint"));
Assert.That(tags["endpoint"], Is.EqualTo(endpointUrl));
Assert.That(tags, Does.ContainKey("attempt.count"));
Assert.That(tags["attempt.count"], Is.EqualTo(1));
Assert.That(tags, Does.ContainKey("outcome"));
Assert.That(tags["outcome"], Is.EqualTo("success"));
ch.Dispose();
}
finally
{
await sut.DisposeAsync().ConfigureAwait(false);
serverCert.Dispose();
}
}
[Test]
[NonParallelizable]
public async Task StructuredLogsCaptureStateTransitionsAsync()
{
using var loggerProvider = new RecordingLoggerProvider();
ITelemetryContext telemetry = DefaultTelemetry.Create(
builder => builder.AddProvider(loggerProvider));
(ClientChannelManager sut, Certificate serverCert, Mock chMock) = CreateMockedSut(telemetry);
try
{
ConfiguredEndpoint endpoint = GetTestEndpoint(serverCert);
var participant = new TestParticipant("p1", endpoint);
IManagedTransportChannel ch = await sut.GetAsync(participant, default).ConfigureAwait(false);
chMock.Setup(c => c.SupportedFeatures).Returns(TransportChannelFeatures.Reconnect);
chMock.Setup(c => c.ReconnectAsync(
It.IsAny(),
It.IsAny()))
.Returns(new ValueTask());
await sut.ReconnectAsync(ch, default).ConfigureAwait(false);
ch.Dispose();
// ch.Dispose() is non-blocking; teardown logs are emitted by
// the background cleanup task.
await WaitForConditionAsync(
() => loggerProvider.Records.Any(record =>
record.EventId.Name == "ParticipantDetached") &&
loggerProvider.Records.Any(record =>
record.EventId.Name == "ChannelClosed"),
"ParticipantDetached + ChannelClosed events").ConfigureAwait(false);
RecordedLogRecord[] records =
[
.. loggerProvider.Records.Where(record => record.CategoryName == "Opc.Ua.ChannelManager")
];
string formatted = string.Join(
Environment.NewLine,
records.Select(record => $"{record.EventId.Name} {record.Message}"));
string?[] eventNames = [.. records.Select(record => record.EventId.Name)];
Assert.That(eventNames, Does.Contain("StateChanged"), formatted);
Assert.That(eventNames, Does.Contain("ReconnectStarted"), formatted);
Assert.That(eventNames, Does.Contain("ReconnectCompleted"), formatted);
Assert.That(eventNames, Does.Contain("ParticipantAttached"), formatted);
Assert.That(eventNames, Does.Contain("ParticipantDetached"), formatted);
Assert.That(eventNames, Does.Contain("ChannelOpened"), formatted);
Assert.That(eventNames, Does.Contain("ChannelClosed"), formatted);
}
finally
{
await sut.DisposeAsync().ConfigureAwait(false);
serverCert.Dispose();
}
}
[Test]
public async Task GetChannelDiagnosticsReturnsSnapshot()
{
(ClientChannelManager sut, Certificate serverCert, _) = CreateMockedSut();
try
{
ConfiguredEndpoint endpoint1 = GetTestEndpoint(serverCert);
ConfiguredEndpoint endpoint2 = GetTestEndpoint(serverCert);
endpoint2.Description.EndpointUrl = "opc.tcp://localhost:4841";
var p1 = new TestParticipant("p1", endpoint1);
var p2 = new TestParticipant("p2", endpoint2);
IManagedTransportChannel ch1 = await sut.GetAsync(p1, default).ConfigureAwait(false);
IManagedTransportChannel ch2 = await sut.GetAsync(p2, default).ConfigureAwait(false);
IReadOnlyList snapshot = sut.GetChannelDiagnostics();
Assert.That(snapshot, Has.Count.EqualTo(2));
ManagedChannelDiagnostic d1 = snapshot.Single(d => d.Key == ch1.Key);
ManagedChannelDiagnostic d2 = snapshot.Single(d => d.Key == ch2.Key);
AssertChannelDiagnostic(d1, ch1.Key);
AssertChannelDiagnostic(d2, ch2.Key);
ch1.Dispose();
ch2.Dispose();
}
finally
{
await sut.DisposeAsync().ConfigureAwait(false);
serverCert.Dispose();
}
}
///
/// Exhausting the shared retry budget stops further reconnect attempts and faults the channel.
///
[Test]
[NonParallelizable]
public async Task ReconnectAsyncWithBudgetStopsWhenExhaustedAsync()
{
using var listener = new ChannelActivityListener();
ITelemetryContext telemetry = NUnitTelemetryContext.Create(logLevel: LogLevel.Critical);
var timeProvider = new FakeTimeProvider();
var reconnectPolicy = new ExponentialBackoffChannelReconnectPolicy
{
MinDelay = TimeSpan.Zero,
MaxDelay = TimeSpan.Zero
};
(ClientChannelManager sut, Certificate serverCert, Mock chMock) =
CreateMockedSut(telemetry, reconnectPolicy, timeProvider);
try
{
ConfiguredEndpoint endpoint = GetTestEndpoint(serverCert);
var participant = new TestParticipant("p1", endpoint);
IManagedTransportChannel ch = await sut.GetAsync(participant, default).ConfigureAwait(false);
var budget = new RetryBudget(TimeSpan.Zero, timeProvider);
// Await rather than Assert.ThrowsAsync: see AssertThrowsAsync.
Task exhaustedReconnect = sut.ReconnectAsync(ch, budget, default).AsTask();
ServiceResultException ex = await AssertThrowsAsync(
exhaustedReconnect,
s_completionTimeout).ConfigureAwait(false);
Assert.That(ex, Is.Not.Null);
Assert.That(ex!.StatusCode, Is.EqualTo(StatusCodes.BadSecureChannelClosed));
Assert.That(ch.State, Is.EqualTo(ChannelState.Faulted));
Activity activity = await listener
.WaitForStoppedActivityAsync("OpcUaChannelReconnect")
.ConfigureAwait(false);
var tags = activity.TagObjects.ToDictionary(t => t.Key, t => t.Value);
Assert.That(tags["attempt.count"], Is.Zero);
Assert.That(tags["outcome"], Is.EqualTo("deadline-expired"));
Assert.That(tags["error.status_code"], Is.EqualTo("BadSecureChannelClosed"));
Assert.That(
tags["error.message"],
Is.EqualTo("Channel recovery deadline expired after 00:00:00."));
chMock.Verify(c => c.ReconnectAsync(
It.IsAny(),
It.IsAny()),
Times.Never);
ch.Dispose();
}
finally
{
await sut.DisposeAsync().ConfigureAwait(false);
serverCert.Dispose();
}
}
[Test]
public async Task ReconnectAsyncSwapsFaultedLeaseEntryAsync()
{
var timeProvider = new ObservableFakeTimeProvider();
var reconnectPolicy = new ExponentialBackoffChannelReconnectPolicy
{
MinDelay = TimeSpan.FromMilliseconds(100),
MaxDelay = TimeSpan.FromMilliseconds(100),
MaxAttempts = 3
};
(ClientChannelManager sut, Certificate serverCert, _) =
CreateMockedSut(reconnectPolicy: reconnectPolicy, timeProvider: timeProvider);
try
{
ConfiguredEndpoint endpoint = GetTestEndpoint(serverCert);
var participant = new TestParticipant("p1", endpoint);
IManagedTransportChannel ch = await sut.GetAsync(participant, default).ConfigureAwait(false);
object originalEntry = GetLeaseEntry(ch);
var exhaustedBudget = new RetryBudget(TimeSpan.Zero, timeProvider);
var faulted = new TaskCompletionSource(
TaskCreationOptions.RunContinuationsAsynchronously);
var reconnecting = new TaskCompletionSource(
TaskCreationOptions.RunContinuationsAsynchronously);
var ready = new TaskCompletionSource(
TaskCreationOptions.RunContinuationsAsynchronously);
ch.StateChanged += (_, change) =>
{
if (change.NewState == ChannelState.Faulted)
{
faulted.TrySetResult(true);
}
else if (change.NewState == ChannelState.TransportReconnecting)
{
reconnecting.TrySetResult(true);
}
else if (change.NewState == ChannelState.Ready)
{
ready.TrySetResult(true);
}
};
// Await rather than Assert.ThrowsAsync: see AssertThrowsAsync.
// This test drives a fake clock from this thread, so blocking
// here would prevent the clock from ever moving again.
Task faultedReconnect = sut
.ReconnectAsync(ch, exhaustedBudget, default)
.AsTask();
await faulted.Task.WaitAsync(s_completionTimeout).ConfigureAwait(false);
await AssertThrowsAsync(
faultedReconnect,
s_completionTimeout).ConfigureAwait(false);
Assert.That(ch.State, Is.EqualTo(ChannelState.Faulted));
// Arm before starting the reconnect: the waiter counts timers
// created from here, so the earlier faulted reconnect above
// cannot consume the slot this test is waiting on.
Task swapBackoff = timeProvider.WaitForTimersCreatedAsync();
Task reconnectTask = sut.ReconnectAsync(ch, default).AsTask();
await swapBackoff.WaitAsync(s_completionTimeout).ConfigureAwait(false);
Assert.That(reconnectTask.IsCompleted, Is.False, "Swap back-off should delay the reset.");
Task retryBackoff = timeProvider.WaitForTimersCreatedAsync();
timeProvider.Advance(TimeSpan.FromMilliseconds(100));
await reconnecting.Task.WaitAsync(s_completionTimeout).ConfigureAwait(false);
await retryBackoff.WaitAsync(s_completionTimeout).ConfigureAwait(false);
Assert.That(reconnectTask.IsCompleted, Is.False, "Reconnect back-off should delay the retry.");
timeProvider.Advance(TimeSpan.FromMilliseconds(100));
await reconnectTask.WaitAsync(s_completionTimeout).ConfigureAwait(false);
await ready.Task.WaitAsync(s_completionTimeout).ConfigureAwait(false);
object freshEntry = GetLeaseEntry(ch);
ManagedChannelDiagnostic diagnostic = sut.GetChannelDiagnostics()
.Single(d => d.Key.Equals(ch.Key));
Assert.That(freshEntry, Is.Not.SameAs(originalEntry));
Assert.That(GetEntryState(freshEntry), Is.EqualTo(ChannelState.Ready));
Assert.That(GetInternalIntProperty(freshEntry, "RefCount"), Is.EqualTo(1));
Assert.That(GetInternalIntProperty(freshEntry, "ParticipantCount"), Is.EqualTo(1));
Assert.That(GetInternalIntProperty(originalEntry, "RefCount"), Is.Zero);
Assert.That(GetInternalIntProperty(ch, "SwapCount"), Is.EqualTo(1));
Assert.That(diagnostic.Refcount, Is.EqualTo(1));
Assert.That(diagnostic.ParticipantCount, Is.EqualTo(1));
Assert.That(participant.NotificationCount, Is.GreaterThanOrEqualTo(2));
ch.Dispose();
}
finally
{
await sut.DisposeAsync().ConfigureAwait(false);
serverCert.Dispose();
}
}
[Test]
public async Task ReconnectAsyncWithBudgetShrinksDelayToFitRemainingAsync()
{
var timeProvider = new ObservableFakeTimeProvider();
var reconnectPolicy = new ExponentialBackoffChannelReconnectPolicy
{
MinDelay = TimeSpan.FromSeconds(10),
MaxDelay = TimeSpan.FromSeconds(10)
};
(ClientChannelManager sut, Certificate serverCert, Mock chMock) =
CreateMockedSut(reconnectPolicy: reconnectPolicy, timeProvider: timeProvider);
try
{
ConfiguredEndpoint endpoint = GetTestEndpoint(serverCert);
var participant = new TestParticipant("p1", endpoint);
IManagedTransportChannel ch = await sut.GetAsync(participant, default).ConfigureAwait(false);
var budget = new RetryBudget(TimeSpan.FromMilliseconds(100), timeProvider);
var reconnecting = new TaskCompletionSource(
TaskCreationOptions.RunContinuationsAsynchronously);
ch.StateChanged += (_, change) =>
{
if (change.NewState == ChannelState.TransportReconnecting)
{
reconnecting.TrySetResult(true);
}
};
// Arm before starting the reconnect so the waiter cannot be
// satisfied by an unrelated timer created earlier in the run.
Task shrunkBackoff = timeProvider.WaitForTimersCreatedAsync();
Task reconnectTask = sut.ReconnectAsync(ch, budget, default).AsTask();
await reconnecting.Task.WaitAsync(s_completionTimeout).ConfigureAwait(false);
await shrunkBackoff.WaitAsync(s_completionTimeout).ConfigureAwait(false);
Assert.That(reconnectTask.IsCompleted, Is.False);
timeProvider.Advance(TimeSpan.FromMilliseconds(100));
ServiceResultException ex = await AssertThrowsAsync(
reconnectTask,
s_completionTimeout).ConfigureAwait(false);
Assert.That(ex, Is.Not.Null);
Assert.That(ex!.StatusCode, Is.EqualTo(StatusCodes.BadSecureChannelClosed));
Assert.That(ch.State, Is.EqualTo(ChannelState.Faulted));
chMock.Verify(c => c.ReconnectAsync(
It.IsAny(),
It.IsAny()),
Times.Never);
ch.Dispose();
}
finally
{
await sut.DisposeAsync().ConfigureAwait(false);
serverCert.Dispose();
}
}
///
/// Verifies the injected clock expires a stalled reactivation at its timeout
/// and issues one final notification.
///
[Test]
public async Task ReconnectParticipantTimeoutUsesInjectedClockAsync()
{
var timeProvider = new ObservableFakeTimeProvider();
var participantTimeout = TimeSpan.FromMilliseconds(200);
var reconnectPolicy = new ExponentialBackoffChannelReconnectPolicy
{
MinDelay = TimeSpan.Zero,
MaxDelay = TimeSpan.Zero,
MaxAttempts = 1,
ParticipantTimeout = participantTimeout
};
var participantCompletion = new TaskCompletionSource(
TaskCreationOptions.RunContinuationsAsynchronously);
(ClientChannelManager sut, Certificate serverCert, _) =
CreateMockedSut(reconnectPolicy: reconnectPolicy, timeProvider: timeProvider);
try
{
var participant = new Mock();
participant.SetupGet(p => p.Id).Returns("bounded-participant");
participant.SetupGet(p => p.Endpoint).Returns(GetTestEndpoint(serverCert));
participant.Setup(p => p.OnReconnectAsync(
It.IsAny(),
0,
It.IsAny()))
.Returns(() => new ValueTask(participantCompletion.Task));
participant.Setup(p => p.OnReconnectAsync(
It.IsAny(),
-1,
It.IsAny()))
.Returns(new ValueTask(ParticipantReconnectResult.Reactivated));
using IManagedTransportChannel channel = await sut.GetAsync(participant.Object, default)
.ConfigureAwait(false);
Task timerCreated = timeProvider.WaitForTimersCreatedAsync(dueTime: participantTimeout);
var budget = new RetryBudget(TimeSpan.FromSeconds(5), timeProvider);
Task reconnectTask = sut.ReconnectAsync(channel, budget, default).AsTask();
await timerCreated.WaitAsync(s_completionTimeout).ConfigureAwait(false);
timeProvider.Advance(participantTimeout - TimeSpan.FromTicks(1));
Assert.That(reconnectTask.IsCompleted, Is.False);
Assert.That(channel.State, Is.EqualTo(ChannelState.TransportConnectedSessionReactivating));
timeProvider.Advance(TimeSpan.FromTicks(1));
ServiceResultException exception = await AssertThrowsAsync(
reconnectTask, s_completionTimeout).ConfigureAwait(false);
Assert.That(exception.StatusCode, Is.EqualTo(StatusCodes.BadSecureChannelClosed));
Assert.That(channel.State, Is.EqualTo(ChannelState.Faulted));
Assert.That(participantCompletion.Task.IsCompleted, Is.False);
participant.Verify(p => p.OnReconnectAsync(
It.IsAny(),
0,
It.IsAny()), Times.Once);
participant.Verify(p => p.OnReconnectAsync(
It.IsAny(),
-1,
It.IsAny()), Times.Once);
}
finally
{
participantCompletion.TrySetResult(ParticipantReconnectResult.Reactivated);
await sut.DisposeAsync().ConfigureAwait(false);
serverCert.Dispose();
}
}
[TestCase(false, false)]
[TestCase(false, true)]
[TestCase(true, false)]
[TestCase(true, true)]
public async Task LegacyParticipantCanSendDuringReactivationAndRecreationAsync(
bool recreate,
bool infiniteTimeout)
{
using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(10));
var policy = new ExponentialBackoffChannelReconnectPolicy
{
MinDelay = TimeSpan.Zero,
MaxDelay = TimeSpan.Zero,
MaxAttempts = 1,
ParticipantTimeout = infiniteTimeout ? Timeout.InfiniteTimeSpan : TimeSpan.FromSeconds(5)
};
(ClientChannelManager manager, Certificate server, Mock transport) =
CreateMockedSut(reconnectPolicy: policy);
var entered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
IManagedTransportChannel? captured = null;
var response = new ActivateSessionResponse();
transport.Setup(value => value.SendRequestAsync(
It.IsAny(), It.IsAny()))
.Returns(new ValueTask(response));
var participant = new Mock();
participant.SetupGet(value => value.Id).Returns("legacy-recovery");
participant.SetupGet(value => value.Endpoint).Returns(GetTestEndpoint(server));
participant.Setup(value => value.OnReconnectAsync(
It.IsAny(), 0, It.IsAny()))
.Returns(async (IManagedTransportChannel channel, int _, CancellationToken ct) =>
{
captured = channel;
IServiceResponse result = await channel.SendRequestAsync(new ActivateSessionRequest(), ct)
.ConfigureAwait(false);
Assert.That(result, Is.SameAs(response));
if (recreate)
{
return ParticipantReconnectResult.RequiresSessionRecreate;
}
entered.TrySetResult(true);
await release.Task.WaitAsync(ct).ConfigureAwait(false);
return ParticipantReconnectResult.Reactivated;
});
if (recreate)
{
participant.As()
.Setup(value => value.RecreateAsync(It.IsAny()))
.Returns(async (CancellationToken ct) =>
{
IServiceResponse result = await captured!.SendRequestAsync(new CreateSessionRequest(), ct)
.ConfigureAwait(false);
Assert.That(result, Is.SameAs(response));
entered.TrySetResult(true);
await release.Task.WaitAsync(ct).ConfigureAwait(false);
});
}
try
{
using IManagedTransportChannel lease = await manager.GetAsync(participant.Object, timeout.Token)
.ConfigureAwait(false);
Task reconnect = manager.ReconnectAsync(lease, timeout.Token).AsTask();
await entered.Task.WaitAsync(timeout.Token).ConfigureAwait(false);
Assert.That(captured, Is.Not.SameAs(lease));
Assert.That(captured!.Key, Is.EqualTo(lease.Key));
Assert.That(captured.Manager, Is.SameAs(manager));
Assert.That(captured.State, Is.EqualTo(ChannelState.TransportConnectedSessionReactivating));
Task ordinary = lease.SendRequestAsync(new ReadRequest(), timeout.Token).AsTask();
Assert.That(ordinary.IsCompleted, Is.False);
Assert.That(manager.GetChannelDiagnostics().Single().Refcount, Is.EqualTo(1));
release.TrySetResult(true);
await reconnect.WaitAsync(timeout.Token).ConfigureAwait(false);
Assert.That(await ordinary.ConfigureAwait(false), Is.SameAs(response));
Assert.That(lease.State, Is.EqualTo(ChannelState.Ready));
ServiceResultException expired = await AssertThrowsAsync(
captured.SendRequestAsync(new CreateSessionRequest(), timeout.Token).AsTask(),
s_completionTimeout).ConfigureAwait(false);
Assert.That(expired.StatusCode, Is.EqualTo(StatusCodes.BadInvalidState));
await captured.CloseAsync(timeout.Token).ConfigureAwait(false);
captured.Dispose();
Assert.That(manager.GetChannelDiagnostics().Single().Refcount, Is.EqualTo(1));
transport.Verify(value => value.SendRequestAsync(
It.IsAny(), It.IsAny()), Times.Once);
transport.Verify(value => value.SendRequestAsync(
It.IsAny(), It.IsAny()),
recreate ? Times.Once() : Times.Never());
}
finally
{
release.TrySetResult(true);
await manager.DisposeAsync().ConfigureAwait(false);
server.Dispose();
}
}
///
/// Recovery views cancel pending sends and reject reuse after their recovery generation changes.
///
[TestCase(false, false)]
[TestCase(false, true)]
[TestCase(true, false)]
[TestCase(true, true)]
public async Task RecoveryViewsCancelInFlightSendsAndRejectPreviousGenerationsAsync(
bool legacy,
bool ignoresTransportCancellation)
{
using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(10));
var policy = new ExponentialBackoffChannelReconnectPolicy
{
MinDelay = TimeSpan.Zero,
MaxDelay = TimeSpan.Zero,
MaxAttempts = 1,
ParticipantTimeout = Timeout.InfiniteTimeSpan
};
(ClientChannelManager manager, Certificate server, Mock transport) =
CreateMockedSut(reconnectPolicy: policy);
var blocked = new TaskCompletionSource(
TaskCreationOptions.RunContinuationsAsynchronously);
CancellationToken wireToken = default;
transport.Setup(channel => channel.SendRequestAsync(
It.IsAny(), It.IsAny()))
.Returns((IServiceRequest _, CancellationToken ct) =>
{
wireToken = ct;
return new ValueTask(
ignoresTransportCancellation ? blocked.Task : blocked.Task.WaitAsync(ct));
});
ITransportChannel? previous = null;
var participant = new Mock();
participant.SetupGet(value => value.Id).Returns("cancelled-recovery-send");
participant.SetupGet(value => value.Endpoint).Returns(GetTestEndpoint(server));
if (legacy)
{
participant.Setup(value => value.OnReconnectAsync(
It.IsAny(), 0, It.IsAny()))
.Returns((IManagedTransportChannel view, int _, CancellationToken _) => RecoverAsync(view));
}
else
{
participant.As().Setup(value => value.OnReconnectAsync(
It.IsAny(), It.IsAny(),
0, It.IsAny()))
.Returns((IManagedTransportChannel _, ITransportChannel view, int _, CancellationToken _) =>
RecoverAsync(view));
}
try
{
using IManagedTransportChannel lease = await manager.GetAsync(participant.Object, timeout.Token)
.ConfigureAwait(false);
await manager.ReconnectAsync(lease, timeout.Token).ConfigureAwait(false);
await manager.ReconnectAsync(lease, timeout.Token).ConfigureAwait(false);
Assert.That(lease.State, Is.EqualTo(ChannelState.Ready));
Assert.That(manager.GetChannelDiagnostics().Single().Refcount, Is.EqualTo(1));
transport.Verify(channel => channel.SendRequestAsync(
It.IsAny(), It.IsAny()), Times.Exactly(2));
}
finally
{
blocked.TrySetResult(new ActivateSessionResponse());
await manager.DisposeAsync().ConfigureAwait(false);
server.Dispose();
}
async ValueTask RecoverAsync(ITransportChannel view)
{
if (previous != null)
{
ServiceResultException stale = await AssertThrowsAsync(
previous.SendRequestAsync(new ActivateSessionRequest(), timeout.Token).AsTask(),
s_completionTimeout).ConfigureAwait(false);
Assert.That(stale.StatusCode, Is.EqualTo(StatusCodes.BadInvalidState));
}
previous = view;
using var cancellation = new CancellationTokenSource();
Task send = view.SendRequestAsync(
new ActivateSessionRequest(), cancellation.Token).AsTask();
Assert.That(send.IsCompleted, Is.False);
cancellation.Cancel();
await Assert.ThatAsync(() => send, Throws.InstanceOf())
.ConfigureAwait(false);
Assert.That(wireToken.IsCancellationRequested, Is.True);
await Assert.ThatAsync(
async () => await view.ReconnectAsync(ct: timeout.Token).ConfigureAwait(false),
Throws.TypeOf().With.Property(nameof(ServiceResultException.StatusCode))
.EqualTo(StatusCodes.BadInvalidState)).ConfigureAwait(false);
return ParticipantReconnectResult.Reactivated;
}
}
[Test]
public async Task RecoverySendViewExpiresWithoutReleasingLeaseOrAdmittingOrdinaryCallsAsync()
{
using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(10));
(ClientChannelManager manager, Certificate server, Mock transport) = CreateMockedSut();
var entered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
var ordinaryStarted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
ITransportChannel? captured = null;
Task? ordinary = null;
var response = new ReadResponse();
transport.Setup(value => value.SendRequestAsync(
It.IsAny(), It.IsAny()))
.Returns(new ValueTask(response));
var participant = new Mock();
participant.SetupGet(value => value.Id).Returns("scoped-recovery");
participant.SetupGet(value => value.Endpoint).Returns(GetTestEndpoint(server));
participant.Setup(value => value.OnReconnectAsync(
It.IsAny(), It.IsAny(),
It.IsAny(), It.IsAny()))
.Returns(async (IManagedTransportChannel lease, ITransportChannel view, int _, CancellationToken ct) =>
{
captured = view;
IServiceResponse result = await view.SendRequestAsync(new ReadRequest(), ct).ConfigureAwait(false);
Assert.That(result, Is.SameAs(response));
ordinary = Task.Run(() =>
{
Task request = lease.SendRequestAsync(
new ReadRequest(), timeout.Token).AsTask();
ordinaryStarted.TrySetResult(true);
return request;
});
await ordinaryStarted.Task.WaitAsync(ct).ConfigureAwait(false);
entered.TrySetResult(true);
await release.Task.WaitAsync(ct).ConfigureAwait(false);
return ParticipantReconnectResult.Reactivated;
});
try
{
using IManagedTransportChannel lease = await manager.GetAsync(participant.Object, timeout.Token)
.ConfigureAwait(false);
Task reconnect = manager.ReconnectAsync(lease, timeout.Token).AsTask();
await entered.Task.WaitAsync(timeout.Token).ConfigureAwait(false);
Assert.That(ordinary!.IsCompleted, Is.False);
Assert.That(manager.GetChannelDiagnostics().Single().Refcount, Is.EqualTo(1));
release.TrySetResult(true);
await reconnect.WaitAsync(timeout.Token).ConfigureAwait(false);
Assert.That(await ordinary.ConfigureAwait(false), Is.SameAs(response));
ServiceResultException expired = Assert.ThrowsAsync(async () =>
await captured!.SendRequestAsync(new ReadRequest(), timeout.Token).ConfigureAwait(false))!;
Assert.That(expired.StatusCode, Is.EqualTo(StatusCodes.BadInvalidState));
await captured!.CloseAsync(timeout.Token).ConfigureAwait(false);
captured.Dispose();
Assert.That(manager.GetChannelDiagnostics().Single().Refcount, Is.EqualTo(1));
Assert.That(await lease.SendRequestAsync(new ReadRequest(), timeout.Token).ConfigureAwait(false),
Is.SameAs(response));
participant.Verify(value => value.CompleteRecoveryAsync(It.IsAny()), Times.Once);
}
finally
{
release.TrySetResult(true);
await manager.DisposeAsync().ConfigureAwait(false);
server.Dispose();
}
}
///
/// The injected clock bounds recreation and invalidates both scoped and legacy recovery views.
///
[TestCase(false)]
[TestCase(true)]
public async Task RecreateParticipantTimeoutUsesInjectedClockAsync(bool scoped)
{
var timeProvider = new ObservableFakeTimeProvider();
var participantTimeout = TimeSpan.FromMilliseconds(200);
var policy = new ExponentialBackoffChannelReconnectPolicy
{
MinDelay = TimeSpan.Zero,
MaxDelay = TimeSpan.Zero,
MaxAttempts = 1,
ParticipantTimeout = participantTimeout
};
var recreateStarted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
var completion = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
ITransportChannel? capturedView = null;
CancellationToken callbackToken = default;
(ClientChannelManager manager, Certificate server, _) =
CreateMockedSut(reconnectPolicy: policy, timeProvider: timeProvider);
try
{
var participant = new Mock();
participant.SetupGet(value => value.Id).Returns("bounded-recreate");
participant.SetupGet(value => value.Endpoint).Returns(GetTestEndpoint(server));
participant.Setup(value => value.OnReconnectAsync(
It.IsAny(), 0, It.IsAny()))
.Returns((IManagedTransportChannel channel, int _, CancellationToken _) =>
{
capturedView = channel;
return new ValueTask(
ParticipantReconnectResult.RequiresSessionRecreate);
});
participant.Setup(value => value.RecreateAsync(It.IsAny()))
.Returns((CancellationToken ct) =>
{
callbackToken = ct;
recreateStarted.TrySetResult(true);
return new ValueTask(completion.Task);
});
if (scoped)
{
Mock recovery = participant.As();
recovery.Setup(value => value.OnReconnectAsync(
It.IsAny(), It.IsAny(),
0, It.IsAny()))
.Returns(new ValueTask(
ParticipantReconnectResult.RequiresSessionRecreate));
recovery.Setup(value => value.RecreateAsync(
It.IsAny(), It.IsAny(),
It.IsAny()))
.Returns((IManagedTransportChannel _, ITransportChannel view, CancellationToken ct) =>
{
capturedView = view;
callbackToken = ct;
recreateStarted.TrySetResult(true);
return new ValueTask(completion.Task.WaitAsync(ct));
});
}
using IManagedTransportChannel lease = await manager.GetAsync(participant.Object)
.ConfigureAwait(false);
Task timerCreated = timeProvider.WaitForTimersCreatedAsync(dueTime: participantTimeout);
Task reconnect = manager.ReconnectAsync(
lease, new RetryBudget(TimeSpan.FromSeconds(5), timeProvider), default).AsTask();
await recreateStarted.Task.WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false);
await timerCreated.WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false);
timeProvider.Advance(participantTimeout - TimeSpan.FromTicks(1));
Assert.That(reconnect.IsCompleted, Is.False);
Assert.That(lease.State, Is.EqualTo(ChannelState.TransportConnectedSessionReactivating));
timeProvider.Advance(TimeSpan.FromTicks(1));
ServiceResultException exception = await AssertThrowsAsync(
reconnect, s_completionTimeout).ConfigureAwait(false);
Assert.That(exception.StatusCode, Is.EqualTo(StatusCodes.BadSecureChannelClosed));
Assert.That(lease.State, Is.EqualTo(ChannelState.Faulted));
Assert.That(completion.Task.IsCompleted, Is.False);
Assert.That(callbackToken.IsCancellationRequested, Is.True);
ServiceResultException expired = await AssertThrowsAsync(
capturedView!.SendRequestAsync(new ReadRequest()).AsTask(),
s_completionTimeout).ConfigureAwait(false);
Assert.That(expired.StatusCode, Is.EqualTo(StatusCodes.BadInvalidState));
capturedView.Dispose();
Assert.That(manager.GetChannelDiagnostics().Single().Refcount, Is.EqualTo(1));
if (scoped)
{
participant.As().Verify(value => value.RecreateAsync(
It.IsAny(), It.IsAny(),
It.IsAny()), Times.Once);
}
else
{
participant.Verify(value => value.RecreateAsync(It.IsAny()), Times.Once);
}
}
finally
{
completion.TrySetResult(true);
await manager.DisposeAsync().ConfigureAwait(false);
server.Dispose();
}
}
///
/// Budget expiry interrupts transport or participant work at the exact recovery deadline.
///
[TestCase(true)]
[TestCase(false)]
public async Task ReconnectBudgetExpiresDuringInFlightWorkAsync(bool transport)
{
var time = new ObservableFakeTimeProvider();
var entered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
var cancelled = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
var notified = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
var policy = new ExponentialBackoffChannelReconnectPolicy
{
MinDelay = TimeSpan.Zero,
MaxDelay = TimeSpan.Zero,
MaxAttempts = 1,
ParticipantTimeout = Timeout.InfiniteTimeSpan
};
(ClientChannelManager manager, Certificate certificate, Mock channel) =
CreateMockedSut(reconnectPolicy: policy, timeProvider: time);
Task? reconnect = null;
try
{
var participant = new Mock();
participant.SetupGet(value => value.Id).Returns("deadline-participant");
participant.SetupGet(value => value.Endpoint).Returns(GetTestEndpoint(certificate));
participant.Setup(value => value.OnReconnectAsync(
It.IsAny(), -1, It.IsAny