mirror of
https://github.com/OPCFoundation/UA-.NETStandard.git
synced 2026-10-18 14:43:08 +08:00
romanett
0716603543
# Description Sessions created concurrently through one `ClientChannelManager` against the same endpoint (`Session.CreateAsync(IClientChannelManager, ...)`) failed with `BadSecureChannelClosed 'The managed transport is closed.'`, and the log showed `Background operation OpenChannel started by ClientChannelManager failed. Channel is Closed.` Creating the same sessions one after another worked. ## Root cause A caller that finds an existing shared entry gets a lease right away, even while the entry's creator is still opening the transport: 1. That session's `OpenAsync` called `lease.Entry.SnapshotClientCertificate()` before any transport was installed and threw *"The managed transport is closed."* 2. The failed session disposed its lease. The refcount dropped to 0 and the entry was torn down. 3. The creator's in-flight `OpenInitialAsync` then found the entry closing and threw *"Channel is Closed."* Every session attached to that entry failed. ## Changes - **`Session.OpenAsync`** waits until the shared entry has an installed transport (new `ChannelEntry.WaitForTransportAsync`) before it reads the certificate. The signal fires when the initial open installs a transport, or when a reconnect that started during the initial open does. It does not wait for `Ready`: `OpenCoreAsync` also runs for session recreate inside a reconnect cycle, and the cycle installs the transport before participants reactivate. - **`ClientChannelManager.GetCoreAsync`**: the creator holds no lease while it opens the entry, so other participants' releases can close the entry under the open. The creator now retries on a fresh entry instead of failing. When the initial open fails on its own, the entry records it (`InitialOpenFaulted`, set under the entry lock at failure time, because the later lifecycle state cannot tell the two cases apart). A faulted open, or one closed by manager disposal, still fails the creator. ## Tests - `SharedChannelConcurrentSessionTests.SessionAttachedWhileChannelOpensWaitsForTransportAsync` (Client.Tests): holds the creator's transport through gated channel bindings until a second session has attached to the opening entry, then checks that both sessions open and share the channel. Fails deterministically without the fix. - `ClientChannelManagerManagedTests.WaitForTransportWaitsForReconnectStartedDuringInitialOpenAsync` and `CreatorDoesNotRetryInitialOpenThatFaultedAsync` (Core.Tests): cover the reconnect-during-open wait and the no-retry-on-fault path. - `SharedChannelConcurrentSessionTests.ConcurrentlyCreatedSessionsShareOneChannelAsync` (Client.Tests): creates 20 sessions concurrently through one manager against the reference server. It checks that they share one channel with refcount 20 and that every session can read. Fails without the fix. - `ClientChannelManagerManagedTests.CreatorRetriesWhenLeaseReleasedDuringInitialOpenAsync` (Core.Tests): holds the initial open with a mock transport, then attaches and releases a second lease during it. It checks that the creator gets a lease on a fresh entry. Fails without the fix with "Channel is Closed." Verification: - **Local runs (net10.0, sequential):** Opc.Ua.Client.Tests 2778 passed; Opc.Ua.Sessions.Tests 851 passed, 379 skipped; Opc.Ua.Core.Tests 6241 passed, 88 skipped. - **net48:** Client.Tests and Core.Tests compile cleanly. - **ServerLoadHarness repro:** `--client shared --sessions 20 --connect-parallel 4/20` now connects every run. The same build without the fix still fails. Note: when the creator's open is cut short this way, the `OpenChannel` background failure is still logged at `fail` level even though the creator now recovers. ## Related Issues - None filed yet. Found while load-testing the shared-channel client with the ServerLoadHarness. ## 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. - [ ] I fixed **all** failing and flaky tests in the CI pipelines and **all** CodeQL warnings. - [ ] I have addressed **all** PR feedback received. 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
3766 lines
179 KiB
C#
3766 lines
179 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.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
|
|
{
|
|
/// <summary>
|
|
/// Unit tests for the managed (sharing, refcount, coalesced
|
|
/// reconnect, participant notification) behavior of
|
|
/// <see cref="ClientChannelManager"/> implementing
|
|
/// <see cref="IClientChannelManager"/>.
|
|
/// </summary>
|
|
[TestFixture]
|
|
[Category("Client")]
|
|
[SetCulture("en-us")]
|
|
[SetUICulture("en-us")]
|
|
[Parallelizable]
|
|
public sealed class ClientChannelManagerManagedTests
|
|
{
|
|
private static readonly ICertificateFactory s_factory = DefaultCertificateFactory.Instance;
|
|
|
|
/// <summary>
|
|
/// How long a test waits for an asynchronous operation it has already
|
|
/// unblocked to be observed as complete.
|
|
/// </summary>
|
|
/// <remarks>
|
|
/// 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.
|
|
/// </remarks>
|
|
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));
|
|
}
|
|
|
|
/// <summary>
|
|
/// 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.
|
|
/// </summary>
|
|
[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));
|
|
}
|
|
|
|
/// <summary>
|
|
/// Two endpoints that name the same transport profile still share a key,
|
|
/// so the profile was added to the identity rather than replacing it.
|
|
/// </summary>
|
|
[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()));
|
|
}
|
|
|
|
/// <summary>
|
|
/// 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.
|
|
/// </summary>
|
|
[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<IChannel> 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<Uri>(),
|
|
It.IsAny<TransportChannelSettings>(),
|
|
It.IsAny<CancellationToken>()),
|
|
Times.Exactly(2));
|
|
|
|
channel.Dispose();
|
|
}
|
|
finally
|
|
{
|
|
serverCert.Dispose();
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task TwoParticipantsSameEndpointShareUnderlyingChannelAsync()
|
|
{
|
|
(ClientChannelManager sut, Certificate serverCert, Mock<IChannel> 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<Uri>(),
|
|
It.IsAny<TransportChannelSettings>(),
|
|
It.IsAny<CancellationToken>()),
|
|
Times.Once);
|
|
|
|
ch1.Dispose();
|
|
ch2.Dispose();
|
|
}
|
|
finally
|
|
{
|
|
serverCert.Dispose();
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task ReleasingOneLeaseKeepsChannelAliveForOtherAsync()
|
|
{
|
|
(ClientChannelManager sut, Certificate serverCert, Mock<IChannel> 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<CancellationToken>()), 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<IChannel> 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<CancellationToken>()), 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<CancellationToken>()), 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<IChannel> 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<CancellationToken>()), Times.Never);
|
|
|
|
lease.Dispose();
|
|
await WaitForMockInvocationAsync(
|
|
() => chMock.Verify(c => c.CloseAsync(It.IsAny<CancellationToken>()), 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<IChannel> 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<OperationCanceledException>()
|
|
.And.Not.InstanceOf<TimeoutException>(),
|
|
"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<OperationCanceledException>()
|
|
.And.Not.InstanceOf<TimeoutException>(),
|
|
"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<IChannel> 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<ServiceResultException>(
|
|
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<Uri>(),
|
|
It.IsAny<TransportChannelSettings>(),
|
|
It.IsAny<CancellationToken>()),
|
|
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<IChannel> 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<IManagedTransportChannel>());
|
|
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<Uri>(),
|
|
It.IsAny<TransportChannelSettings>(),
|
|
It.IsAny<CancellationToken>()),
|
|
Times.Once);
|
|
|
|
discoveryClient.Dispose();
|
|
discoveryClient = null;
|
|
|
|
chMock.Verify(c => c.CloseAsync(It.IsAny<CancellationToken>()), 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<CancellationToken>()), Times.Once))
|
|
.ConfigureAwait(false);
|
|
}
|
|
finally
|
|
{
|
|
serverCert.Dispose();
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task RegistrationClientCreateAsyncUsesManagedLeaseAndReleasesItAsync()
|
|
{
|
|
(ClientChannelManager sut, Certificate serverCert, Mock<IChannel> 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<IManagedTransportChannel>());
|
|
chMock.Verify(c => c.OpenAsync(
|
|
It.IsAny<Uri>(),
|
|
It.IsAny<TransportChannelSettings>(),
|
|
It.IsAny<CancellationToken>()),
|
|
Times.Once);
|
|
|
|
registrationClient.Dispose();
|
|
|
|
chMock.Verify(c => c.CloseAsync(It.IsAny<CancellationToken>()), Times.Once);
|
|
}
|
|
finally
|
|
{
|
|
serverCert.Dispose();
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task ReconnectAsyncNotifiesAttachedParticipantsAsync()
|
|
{
|
|
(ClientChannelManager sut, Certificate serverCert, Mock<IChannel> 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<ITransportWaitingConnection?>(),
|
|
It.IsAny<CancellationToken>()))
|
|
.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<IChannel> 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<ITransportWaitingConnection?>(),
|
|
It.IsAny<CancellationToken>()))
|
|
.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<bool>(
|
|
TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var releaseInitialOpen = new TaskCompletionSource<bool>(
|
|
TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var reconnectOpenStarted = new TaskCompletionSource<bool>(
|
|
TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var initialChannel = new Mock<IChannel>();
|
|
var reconnectChannel = new Mock<IChannel>();
|
|
|
|
initialChannel
|
|
.Setup(channel => channel.OpenAsync(
|
|
It.IsAny<Uri>(),
|
|
It.IsAny<TransportChannelSettings>(),
|
|
It.IsAny<CancellationToken>()))
|
|
.Callback(() => initialOpenStarted.TrySetResult(true))
|
|
.Returns(() => new ValueTask(releaseInitialOpen.Task));
|
|
reconnectChannel
|
|
.Setup(channel => channel.OpenAsync(
|
|
It.IsAny<Uri>(),
|
|
It.IsAny<TransportChannelSettings>(),
|
|
It.IsAny<CancellationToken>()))
|
|
.Callback(() => reconnectOpenStarted.TrySetResult(true))
|
|
.Returns(new ValueTask());
|
|
initialChannel
|
|
.Setup(channel => channel.CloseAsync(It.IsAny<CancellationToken>()))
|
|
.Returns(new ValueTask());
|
|
reconnectChannel
|
|
.Setup(channel => channel.CloseAsync(It.IsAny<CancellationToken>()))
|
|
.Returns(new ValueTask());
|
|
|
|
int createCount = 0;
|
|
var bindings = new Mock<ITransportChannelBindings>();
|
|
bindings
|
|
.Setup(factory => factory.Create(
|
|
It.IsAny<string>(),
|
|
It.IsAny<ITelemetryContext>()))
|
|
.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<bool> 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<CancellationToken>()),
|
|
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);
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// 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.
|
|
/// </summary>
|
|
[Test]
|
|
public async Task CreatorRetriesWhenLeaseReleasedDuringInitialOpenAsync()
|
|
{
|
|
var initialOpenStarted = new TaskCompletionSource<bool>(
|
|
TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var releaseInitialOpen = new TaskCompletionSource<bool>(
|
|
TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var initialChannel = new Mock<IChannel>();
|
|
var retryChannel = new Mock<IChannel>();
|
|
initialChannel
|
|
.Setup(channel => channel.OpenAsync(
|
|
It.IsAny<Uri>(),
|
|
It.IsAny<TransportChannelSettings>(),
|
|
It.IsAny<CancellationToken>()))
|
|
.Callback(() => initialOpenStarted.TrySetResult(true))
|
|
.Returns(() => new ValueTask(releaseInitialOpen.Task));
|
|
retryChannel
|
|
.Setup(channel => channel.OpenAsync(
|
|
It.IsAny<Uri>(),
|
|
It.IsAny<TransportChannelSettings>(),
|
|
It.IsAny<CancellationToken>()))
|
|
.Returns(new ValueTask());
|
|
initialChannel
|
|
.Setup(channel => channel.CloseAsync(It.IsAny<CancellationToken>()))
|
|
.Returns(new ValueTask());
|
|
retryChannel
|
|
.Setup(channel => channel.CloseAsync(It.IsAny<CancellationToken>()))
|
|
.Returns(new ValueTask());
|
|
|
|
int createCount = 0;
|
|
var bindings = new Mock<ITransportChannelBindings>();
|
|
bindings
|
|
.Setup(factory => factory.Create(
|
|
It.IsAny<string>(),
|
|
It.IsAny<ITelemetryContext>()))
|
|
.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<IManagedTransportChannel> 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<CancellationToken>()),
|
|
Times.Once);
|
|
}
|
|
finally
|
|
{
|
|
releaseInitialOpen.TrySetResult(true);
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// 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.
|
|
/// </summary>
|
|
[Test]
|
|
public async Task WaitForTransportWaitsForReconnectStartedDuringInitialOpenAsync()
|
|
{
|
|
var releaseInitialOpen = new TaskCompletionSource<bool>(
|
|
TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var initialOpenStarted = new TaskCompletionSource<bool>(
|
|
TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var reconnectOpenStarted = new TaskCompletionSource<bool>(
|
|
TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var releaseReconnectOpen = new TaskCompletionSource<bool>(
|
|
TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var initialChannel = new Mock<IChannel>();
|
|
var reconnectChannel = new Mock<IChannel>();
|
|
initialChannel
|
|
.Setup(channel => channel.OpenAsync(
|
|
It.IsAny<Uri>(),
|
|
It.IsAny<TransportChannelSettings>(),
|
|
It.IsAny<CancellationToken>()))
|
|
.Callback(() => initialOpenStarted.TrySetResult(true))
|
|
.Returns(() => new ValueTask(releaseInitialOpen.Task));
|
|
reconnectChannel
|
|
.Setup(channel => channel.OpenAsync(
|
|
It.IsAny<Uri>(),
|
|
It.IsAny<TransportChannelSettings>(),
|
|
It.IsAny<CancellationToken>()))
|
|
.Callback(() => reconnectOpenStarted.TrySetResult(true))
|
|
.Returns(() => new ValueTask(releaseReconnectOpen.Task));
|
|
initialChannel
|
|
.Setup(channel => channel.CloseAsync(It.IsAny<CancellationToken>()))
|
|
.Returns(new ValueTask());
|
|
reconnectChannel
|
|
.Setup(channel => channel.CloseAsync(It.IsAny<CancellationToken>()))
|
|
.Returns(new ValueTask());
|
|
|
|
int createCount = 0;
|
|
var bindings = new Mock<ITransportChannelBindings>();
|
|
bindings
|
|
.Setup(factory => factory.Create(
|
|
It.IsAny<string>(),
|
|
It.IsAny<ITelemetryContext>()))
|
|
.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<bool> 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);
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// 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.
|
|
/// </summary>
|
|
[Test]
|
|
public async Task CreatorDoesNotRetryInitialOpenThatFaultedAsync()
|
|
{
|
|
var channel = new Mock<IChannel>();
|
|
channel
|
|
.Setup(c => c.OpenAsync(
|
|
It.IsAny<Uri>(),
|
|
It.IsAny<TransportChannelSettings>(),
|
|
It.IsAny<CancellationToken>()))
|
|
.Returns(() => new ValueTask(Task.FromException(
|
|
new ServiceResultException(StatusCodes.BadSecureChannelClosed))));
|
|
channel
|
|
.Setup(c => c.CloseAsync(It.IsAny<CancellationToken>()))
|
|
.Returns(new ValueTask());
|
|
|
|
int createCount = 0;
|
|
var bindings = new Mock<ITransportChannelBindings>();
|
|
bindings
|
|
.Setup(factory => factory.Create(
|
|
It.IsAny<string>(),
|
|
It.IsAny<ITelemetryContext>()))
|
|
.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<ServiceResultException>());
|
|
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<ServiceResultException>()
|
|
.With.Property(nameof(ServiceResultException.StatusCode))
|
|
.EqualTo(StatusCodes.BadSecureChannelClosed));
|
|
});
|
|
|
|
createCount = 0;
|
|
ServiceResultException? sre = Assert.ThrowsAsync<ServiceResultException>(
|
|
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));
|
|
}
|
|
|
|
/// <summary>
|
|
/// Fatal participant verdicts fault the channel without starting another cycle for a bounded caller.
|
|
/// </summary>
|
|
[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<IChannel> 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<ITransportWaitingConnection?>(),
|
|
It.IsAny<CancellationToken>()))
|
|
.Returns(new ValueTask());
|
|
|
|
if (boundedCaller)
|
|
{
|
|
await AssertThrowsAsync<ServiceResultException>(
|
|
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<ITransportWaitingConnection?>(), It.IsAny<CancellationToken>()), Times.Once);
|
|
ch.Dispose();
|
|
}
|
|
finally
|
|
{
|
|
await sut.DisposeAsync().ConfigureAwait(false);
|
|
serverCert.Dispose();
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task FatalForParticipantDetachesOnlyThatParticipantAsync()
|
|
{
|
|
(ClientChannelManager sut, Certificate serverCert, Mock<IChannel> 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<ITransportWaitingConnection?>(),
|
|
It.IsAny<CancellationToken>()))
|
|
.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<IChannel> 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<ITransportWaitingConnection?>(),
|
|
It.IsAny<CancellationToken>()))
|
|
.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<IChannel> 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<bool>(
|
|
TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var allowReconnect = new TaskCompletionSource<bool>(
|
|
TaskCreationOptions.RunContinuationsAsynchronously);
|
|
chMock.Setup(c => c.SupportedFeatures).Returns(TransportChannelFeatures.Reconnect);
|
|
chMock.Setup(c => c.ReconnectAsync(
|
|
It.IsAny<ITransportWaitingConnection?>(),
|
|
It.IsAny<CancellationToken>()))
|
|
.Returns(() => new ValueTask(WaitForReconnectReleaseAsync(
|
|
reconnectEntered,
|
|
allowReconnect)));
|
|
chMock.Setup(c => c.SendRequestAsync(
|
|
It.IsAny<IServiceRequest>(),
|
|
It.IsAny<CancellationToken>()))
|
|
.Returns(new ValueTask<IServiceResponse>(new ReadResponse
|
|
{
|
|
ResponseHeader = new ResponseHeader { ServiceResult = StatusCodes.Good }
|
|
}));
|
|
|
|
Task reconnectTask = sut.ReconnectAsync(ch, default).AsTask();
|
|
await reconnectEntered.Task.WaitAsync(s_completionTimeout).ConfigureAwait(false);
|
|
Task<IServiceResponse> 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<string, object?>[] tags)
|
|
{
|
|
return WaitForMeasurementAsync(metrics, instrumentName, null, tags);
|
|
}
|
|
|
|
private static async Task WaitForMeasurementAsync(
|
|
ChannelMetricListener metrics,
|
|
string instrumentName,
|
|
double? value,
|
|
params KeyValuePair<string, object?>[] 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<bool> 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<IChannel> 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<ITransportWaitingConnection?>(),
|
|
It.IsAny<CancellationToken>()))
|
|
.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<IChannel> 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<ITransportWaitingConnection?>(),
|
|
It.IsAny<CancellationToken>()))
|
|
.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<ManagedChannelDiagnostic> 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();
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Exhausting the shared retry budget stops further reconnect attempts and faults the channel.
|
|
/// </summary>
|
|
[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<IChannel> 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<ServiceResultException>(
|
|
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<ITransportWaitingConnection?>(),
|
|
It.IsAny<CancellationToken>()),
|
|
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<bool>(
|
|
TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var reconnecting = new TaskCompletionSource<bool>(
|
|
TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var ready = new TaskCompletionSource<bool>(
|
|
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<ServiceResultException>(
|
|
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<bool> 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<bool> 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<IChannel> 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<bool>(
|
|
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<bool> 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<ServiceResultException>(
|
|
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<ITransportWaitingConnection?>(),
|
|
It.IsAny<CancellationToken>()),
|
|
Times.Never);
|
|
ch.Dispose();
|
|
}
|
|
finally
|
|
{
|
|
await sut.DisposeAsync().ConfigureAwait(false);
|
|
serverCert.Dispose();
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Verifies the injected clock expires a stalled reactivation at its timeout
|
|
/// and issues one final notification.
|
|
/// </summary>
|
|
[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<ParticipantReconnectResult>(
|
|
TaskCreationOptions.RunContinuationsAsynchronously);
|
|
(ClientChannelManager sut, Certificate serverCert, _) =
|
|
CreateMockedSut(reconnectPolicy: reconnectPolicy, timeProvider: timeProvider);
|
|
try
|
|
{
|
|
var participant = new Mock<IReconnectParticipant>();
|
|
participant.SetupGet(p => p.Id).Returns("bounded-participant");
|
|
participant.SetupGet(p => p.Endpoint).Returns(GetTestEndpoint(serverCert));
|
|
participant.Setup(p => p.OnReconnectAsync(
|
|
It.IsAny<IManagedTransportChannel>(),
|
|
0,
|
|
It.IsAny<CancellationToken>()))
|
|
.Returns(() => new ValueTask<ParticipantReconnectResult>(participantCompletion.Task));
|
|
participant.Setup(p => p.OnReconnectAsync(
|
|
It.IsAny<IManagedTransportChannel>(),
|
|
-1,
|
|
It.IsAny<CancellationToken>()))
|
|
.Returns(new ValueTask<ParticipantReconnectResult>(ParticipantReconnectResult.Reactivated));
|
|
using IManagedTransportChannel channel = await sut.GetAsync(participant.Object, default)
|
|
.ConfigureAwait(false);
|
|
Task<bool> 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<ServiceResultException>(
|
|
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<IManagedTransportChannel>(),
|
|
0,
|
|
It.IsAny<CancellationToken>()), Times.Once);
|
|
participant.Verify(p => p.OnReconnectAsync(
|
|
It.IsAny<IManagedTransportChannel>(),
|
|
-1,
|
|
It.IsAny<CancellationToken>()), 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<IChannel> transport) =
|
|
CreateMockedSut(reconnectPolicy: policy);
|
|
var entered = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var release = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
IManagedTransportChannel? captured = null;
|
|
var response = new ActivateSessionResponse();
|
|
transport.Setup(value => value.SendRequestAsync(
|
|
It.IsAny<IServiceRequest>(), It.IsAny<CancellationToken>()))
|
|
.Returns(new ValueTask<IServiceResponse>(response));
|
|
var participant = new Mock<IReconnectParticipant>();
|
|
participant.SetupGet(value => value.Id).Returns("legacy-recovery");
|
|
participant.SetupGet(value => value.Endpoint).Returns(GetTestEndpoint(server));
|
|
participant.Setup(value => value.OnReconnectAsync(
|
|
It.IsAny<IManagedTransportChannel>(), 0, It.IsAny<CancellationToken>()))
|
|
.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<IRecreateAwareReconnectParticipant>()
|
|
.Setup(value => value.RecreateAsync(It.IsAny<CancellationToken>()))
|
|
.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<IServiceResponse> 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<ServiceResultException>(
|
|
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<ActivateSessionRequest>(), It.IsAny<CancellationToken>()), Times.Once);
|
|
transport.Verify(value => value.SendRequestAsync(
|
|
It.IsAny<CreateSessionRequest>(), It.IsAny<CancellationToken>()),
|
|
recreate ? Times.Once() : Times.Never());
|
|
}
|
|
finally
|
|
{
|
|
release.TrySetResult(true);
|
|
await manager.DisposeAsync().ConfigureAwait(false);
|
|
server.Dispose();
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Recovery views cancel pending sends and reject reuse after their recovery generation changes.
|
|
/// </summary>
|
|
[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<IChannel> transport) =
|
|
CreateMockedSut(reconnectPolicy: policy);
|
|
var blocked = new TaskCompletionSource<IServiceResponse>(
|
|
TaskCreationOptions.RunContinuationsAsynchronously);
|
|
CancellationToken wireToken = default;
|
|
transport.Setup(channel => channel.SendRequestAsync(
|
|
It.IsAny<ActivateSessionRequest>(), It.IsAny<CancellationToken>()))
|
|
.Returns((IServiceRequest _, CancellationToken ct) =>
|
|
{
|
|
wireToken = ct;
|
|
return new ValueTask<IServiceResponse>(
|
|
ignoresTransportCancellation ? blocked.Task : blocked.Task.WaitAsync(ct));
|
|
});
|
|
ITransportChannel? previous = null;
|
|
var participant = new Mock<IReconnectParticipant>();
|
|
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<IManagedTransportChannel>(), 0, It.IsAny<CancellationToken>()))
|
|
.Returns((IManagedTransportChannel view, int _, CancellationToken _) => RecoverAsync(view));
|
|
}
|
|
else
|
|
{
|
|
participant.As<IChannelRecoveryParticipant>().Setup(value => value.OnReconnectAsync(
|
|
It.IsAny<IManagedTransportChannel>(), It.IsAny<ITransportChannel>(),
|
|
0, It.IsAny<CancellationToken>()))
|
|
.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<ActivateSessionRequest>(), It.IsAny<CancellationToken>()), Times.Exactly(2));
|
|
}
|
|
finally
|
|
{
|
|
blocked.TrySetResult(new ActivateSessionResponse());
|
|
await manager.DisposeAsync().ConfigureAwait(false);
|
|
server.Dispose();
|
|
}
|
|
|
|
async ValueTask<ParticipantReconnectResult> RecoverAsync(ITransportChannel view)
|
|
{
|
|
if (previous != null)
|
|
{
|
|
ServiceResultException stale = await AssertThrowsAsync<ServiceResultException>(
|
|
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<IServiceResponse> send = view.SendRequestAsync(
|
|
new ActivateSessionRequest(), cancellation.Token).AsTask();
|
|
Assert.That(send.IsCompleted, Is.False);
|
|
cancellation.Cancel();
|
|
await Assert.ThatAsync(() => send, Throws.InstanceOf<OperationCanceledException>())
|
|
.ConfigureAwait(false);
|
|
Assert.That(wireToken.IsCancellationRequested, Is.True);
|
|
await Assert.ThatAsync(
|
|
async () => await view.ReconnectAsync(ct: timeout.Token).ConfigureAwait(false),
|
|
Throws.TypeOf<ServiceResultException>().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<IChannel> transport) = CreateMockedSut();
|
|
var entered = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var release = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var ordinaryStarted = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
ITransportChannel? captured = null;
|
|
Task<IServiceResponse>? ordinary = null;
|
|
var response = new ReadResponse();
|
|
transport.Setup(value => value.SendRequestAsync(
|
|
It.IsAny<IServiceRequest>(), It.IsAny<CancellationToken>()))
|
|
.Returns(new ValueTask<IServiceResponse>(response));
|
|
var participant = new Mock<IChannelRecoveryParticipant>();
|
|
participant.SetupGet(value => value.Id).Returns("scoped-recovery");
|
|
participant.SetupGet(value => value.Endpoint).Returns(GetTestEndpoint(server));
|
|
participant.Setup(value => value.OnReconnectAsync(
|
|
It.IsAny<IManagedTransportChannel>(), It.IsAny<ITransportChannel>(),
|
|
It.IsAny<int>(), It.IsAny<CancellationToken>()))
|
|
.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<IServiceResponse> 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<ServiceResultException>(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<CancellationToken>()), Times.Once);
|
|
}
|
|
finally
|
|
{
|
|
release.TrySetResult(true);
|
|
await manager.DisposeAsync().ConfigureAwait(false);
|
|
server.Dispose();
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// The injected clock bounds recreation and invalidates both scoped and legacy recovery views.
|
|
/// </summary>
|
|
[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<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var completion = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
ITransportChannel? capturedView = null;
|
|
CancellationToken callbackToken = default;
|
|
(ClientChannelManager manager, Certificate server, _) =
|
|
CreateMockedSut(reconnectPolicy: policy, timeProvider: timeProvider);
|
|
try
|
|
{
|
|
var participant = new Mock<IRecreateAwareReconnectParticipant>();
|
|
participant.SetupGet(value => value.Id).Returns("bounded-recreate");
|
|
participant.SetupGet(value => value.Endpoint).Returns(GetTestEndpoint(server));
|
|
participant.Setup(value => value.OnReconnectAsync(
|
|
It.IsAny<IManagedTransportChannel>(), 0, It.IsAny<CancellationToken>()))
|
|
.Returns((IManagedTransportChannel channel, int _, CancellationToken _) =>
|
|
{
|
|
capturedView = channel;
|
|
return new ValueTask<ParticipantReconnectResult>(
|
|
ParticipantReconnectResult.RequiresSessionRecreate);
|
|
});
|
|
participant.Setup(value => value.RecreateAsync(It.IsAny<CancellationToken>()))
|
|
.Returns((CancellationToken ct) =>
|
|
{
|
|
callbackToken = ct;
|
|
recreateStarted.TrySetResult(true);
|
|
return new ValueTask(completion.Task);
|
|
});
|
|
if (scoped)
|
|
{
|
|
Mock<IChannelRecoveryParticipant> recovery = participant.As<IChannelRecoveryParticipant>();
|
|
recovery.Setup(value => value.OnReconnectAsync(
|
|
It.IsAny<IManagedTransportChannel>(), It.IsAny<ITransportChannel>(),
|
|
0, It.IsAny<CancellationToken>()))
|
|
.Returns(new ValueTask<ParticipantReconnectResult>(
|
|
ParticipantReconnectResult.RequiresSessionRecreate));
|
|
recovery.Setup(value => value.RecreateAsync(
|
|
It.IsAny<IManagedTransportChannel>(), It.IsAny<ITransportChannel>(),
|
|
It.IsAny<CancellationToken>()))
|
|
.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<bool> 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<ServiceResultException>(
|
|
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<ServiceResultException>(
|
|
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<IChannelRecoveryParticipant>().Verify(value => value.RecreateAsync(
|
|
It.IsAny<IManagedTransportChannel>(), It.IsAny<ITransportChannel>(),
|
|
It.IsAny<CancellationToken>()), Times.Once);
|
|
}
|
|
else
|
|
{
|
|
participant.Verify(value => value.RecreateAsync(It.IsAny<CancellationToken>()), Times.Once);
|
|
}
|
|
}
|
|
finally
|
|
{
|
|
completion.TrySetResult(true);
|
|
await manager.DisposeAsync().ConfigureAwait(false);
|
|
server.Dispose();
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Budget expiry interrupts transport or participant work at the exact recovery deadline.
|
|
/// </summary>
|
|
[TestCase(true)]
|
|
[TestCase(false)]
|
|
public async Task ReconnectBudgetExpiresDuringInFlightWorkAsync(bool transport)
|
|
{
|
|
var time = new ObservableFakeTimeProvider();
|
|
var entered = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var release = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var cancelled = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var notified = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var policy = new ExponentialBackoffChannelReconnectPolicy
|
|
{
|
|
MinDelay = TimeSpan.Zero,
|
|
MaxDelay = TimeSpan.Zero,
|
|
MaxAttempts = 1,
|
|
ParticipantTimeout = Timeout.InfiniteTimeSpan
|
|
};
|
|
(ClientChannelManager manager, Certificate certificate, Mock<IChannel> channel) =
|
|
CreateMockedSut(reconnectPolicy: policy, timeProvider: time);
|
|
Task? reconnect = null;
|
|
try
|
|
{
|
|
var participant = new Mock<IReconnectParticipant>();
|
|
participant.SetupGet(value => value.Id).Returns("deadline-participant");
|
|
participant.SetupGet(value => value.Endpoint).Returns(GetTestEndpoint(certificate));
|
|
participant.Setup(value => value.OnReconnectAsync(
|
|
It.IsAny<IManagedTransportChannel>(), -1, It.IsAny<CancellationToken>()))
|
|
.Callback(() => notified.TrySetResult(true))
|
|
.Returns(new ValueTask<ParticipantReconnectResult>(ParticipantReconnectResult.Reactivated));
|
|
participant.Setup(value => value.OnReconnectAsync(
|
|
It.IsAny<IManagedTransportChannel>(), 0, It.IsAny<CancellationToken>()))
|
|
.Returns(async (IManagedTransportChannel _, int _, CancellationToken ct) =>
|
|
{
|
|
if (!transport)
|
|
{
|
|
await WaitForReleaseAsync(ct).ConfigureAwait(false);
|
|
}
|
|
return ParticipantReconnectResult.Reactivated;
|
|
});
|
|
channel.SetupGet(value => value.SupportedFeatures).Returns(TransportChannelFeatures.Reconnect);
|
|
channel.Setup(value => value.ReconnectAsync(
|
|
It.IsAny<ITransportWaitingConnection?>(), It.IsAny<CancellationToken>()))
|
|
.Returns(async (ITransportWaitingConnection? _, CancellationToken ct) =>
|
|
{
|
|
if (transport)
|
|
{
|
|
await WaitForReleaseAsync(ct).ConfigureAwait(false);
|
|
}
|
|
});
|
|
using IManagedTransportChannel lease = await manager.GetAsync(participant.Object).ConfigureAwait(false);
|
|
var duration = TimeSpan.FromSeconds(1);
|
|
reconnect = manager.ReconnectAsync(lease, new RetryBudget(duration, time)).AsTask();
|
|
await entered.Task.WaitAsync(TimeSpan.FromSeconds(10)).ConfigureAwait(false);
|
|
|
|
time.Advance(duration - TimeSpan.FromTicks(1));
|
|
Assert.That(reconnect.IsCompleted, Is.False);
|
|
Assert.That(cancelled.Task.IsCompleted, Is.False);
|
|
|
|
time.Advance(TimeSpan.FromTicks(1));
|
|
await cancelled.Task.WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false);
|
|
ServiceResultException error = await AssertThrowsAsync<ServiceResultException>(
|
|
reconnect, TimeSpan.FromSeconds(5)).ConfigureAwait(false);
|
|
|
|
Assert.That(error.StatusCode, Is.EqualTo(StatusCodes.BadSecureChannelClosed));
|
|
Assert.That(lease.State, Is.EqualTo(ChannelState.Faulted));
|
|
Assert.That(GetInternalIntProperty(lease, "SwapCount"), Is.Zero);
|
|
await notified.Task.WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false);
|
|
participant.Verify(value => value.OnReconnectAsync(
|
|
It.IsAny<IManagedTransportChannel>(), -1, It.IsAny<CancellationToken>()), Times.Once);
|
|
}
|
|
finally
|
|
{
|
|
release.TrySetResult(true);
|
|
if (reconnect != null)
|
|
{
|
|
await ObserveCompletionAsync(reconnect).ConfigureAwait(false);
|
|
}
|
|
await manager.DisposeAsync().ConfigureAwait(false);
|
|
certificate.Dispose();
|
|
}
|
|
|
|
async Task WaitForReleaseAsync(CancellationToken ct)
|
|
{
|
|
entered.TrySetResult(true);
|
|
try
|
|
{
|
|
await release.Task.WaitAsync(ct).ConfigureAwait(false);
|
|
}
|
|
catch (OperationCanceledException) when (ct.IsCancellationRequested)
|
|
{
|
|
cancelled.TrySetResult(true);
|
|
throw;
|
|
}
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// A coalesced caller's tighter budget cancels work that is already in flight.
|
|
/// </summary>
|
|
[Test]
|
|
public async Task ReconnectBudgetTighteningCancelsAlreadyRunningWorkAsync()
|
|
{
|
|
var time = new ObservableFakeTimeProvider();
|
|
var entered = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var release = new TaskCompletionSource<ParticipantReconnectResult>(
|
|
TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var policy = new ExponentialBackoffChannelReconnectPolicy
|
|
{
|
|
MinDelay = TimeSpan.Zero,
|
|
MaxDelay = TimeSpan.Zero,
|
|
MaxAttempts = 1,
|
|
ParticipantTimeout = Timeout.InfiniteTimeSpan
|
|
};
|
|
(ClientChannelManager manager, Certificate certificate, _) =
|
|
CreateMockedSut(reconnectPolicy: policy, timeProvider: time);
|
|
Task? reconnect = null;
|
|
Task? joiner = null;
|
|
try
|
|
{
|
|
var participant = new Mock<IReconnectParticipant>();
|
|
participant.SetupGet(value => value.Id).Returns("tightened-participant");
|
|
participant.SetupGet(value => value.Endpoint).Returns(GetTestEndpoint(certificate));
|
|
participant.Setup(value => value.OnReconnectAsync(
|
|
It.IsAny<IManagedTransportChannel>(), -1, It.IsAny<CancellationToken>()))
|
|
.Returns(new ValueTask<ParticipantReconnectResult>(ParticipantReconnectResult.Reactivated));
|
|
participant.Setup(value => value.OnReconnectAsync(
|
|
It.IsAny<IManagedTransportChannel>(), 0, It.IsAny<CancellationToken>()))
|
|
.Returns((IManagedTransportChannel _, int _, CancellationToken ct) =>
|
|
{
|
|
entered.TrySetResult(true);
|
|
return new ValueTask<ParticipantReconnectResult>(release.Task.WaitAsync(ct));
|
|
});
|
|
using IManagedTransportChannel lease = await manager.GetAsync(participant.Object).ConfigureAwait(false);
|
|
reconnect = manager.ReconnectAsync(lease, new RetryBudget(TimeSpan.FromMinutes(1), time)).AsTask();
|
|
await entered.Task.WaitAsync(TimeSpan.FromSeconds(10)).ConfigureAwait(false);
|
|
time.Advance(TimeSpan.FromSeconds(10));
|
|
|
|
joiner = manager.ReconnectAsync(lease, new RetryBudget(TimeSpan.FromSeconds(1), time)).AsTask();
|
|
time.Advance(TimeSpan.FromSeconds(1));
|
|
|
|
await AssertThrowsAsync<ServiceResultException>(reconnect, TimeSpan.FromSeconds(5))
|
|
.ConfigureAwait(false);
|
|
await AssertThrowsAsync<ServiceResultException>(joiner, TimeSpan.FromSeconds(5))
|
|
.ConfigureAwait(false);
|
|
Assert.That(lease.State, Is.EqualTo(ChannelState.Faulted));
|
|
participant.Verify(value => value.OnReconnectAsync(
|
|
It.IsAny<IManagedTransportChannel>(), 0, It.IsAny<CancellationToken>()), Times.Once);
|
|
}
|
|
finally
|
|
{
|
|
release.TrySetResult(ParticipantReconnectResult.Reactivated);
|
|
if (reconnect != null)
|
|
{
|
|
await ObserveCompletionAsync(reconnect).ConfigureAwait(false);
|
|
}
|
|
if (joiner != null)
|
|
{
|
|
await ObserveCompletionAsync(joiner).ConfigureAwait(false);
|
|
}
|
|
await manager.DisposeAsync().ConfigureAwait(false);
|
|
certificate.Dispose();
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// The base participant contract supplies the earliest shared deadline without a caller override.
|
|
/// </summary>
|
|
[Test]
|
|
public async Task ParticipantBudgetsBoundSharedRecoveryWithoutACallerOverrideAsync()
|
|
{
|
|
var time = new FakeTimeProvider();
|
|
var entered = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var release = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var policy = new ExponentialBackoffChannelReconnectPolicy
|
|
{
|
|
MinDelay = TimeSpan.Zero,
|
|
MaxDelay = TimeSpan.Zero,
|
|
ParticipantTimeout = Timeout.InfiniteTimeSpan
|
|
};
|
|
(ClientChannelManager manager, Certificate certificate, Mock<IChannel> transport) =
|
|
CreateMockedSut(reconnectPolicy: policy, timeProvider: time);
|
|
try
|
|
{
|
|
Mock<IReconnectParticipant> shortParticipant =
|
|
CreateParticipant("short", TimeSpan.FromSeconds(1));
|
|
Mock<IReconnectParticipant> longParticipant =
|
|
CreateParticipant("long", TimeSpan.FromSeconds(10));
|
|
transport.SetupGet(value => value.SupportedFeatures).Returns(TransportChannelFeatures.Reconnect);
|
|
transport.Setup(value => value.ReconnectAsync(
|
|
It.IsAny<ITransportWaitingConnection?>(), It.IsAny<CancellationToken>()))
|
|
.Returns(async (ITransportWaitingConnection? _, CancellationToken ct) =>
|
|
{
|
|
entered.TrySetResult(true);
|
|
await release.Task.WaitAsync(ct).ConfigureAwait(false);
|
|
});
|
|
using IManagedTransportChannel first = await manager.GetAsync(shortParticipant.Object)
|
|
.ConfigureAwait(false);
|
|
using IManagedTransportChannel second = await manager.GetAsync(longParticipant.Object)
|
|
.ConfigureAwait(false);
|
|
|
|
Task reconnect = manager.ReconnectAsync(second).AsTask();
|
|
await entered.Task.WaitAsync(s_completionTimeout).ConfigureAwait(false);
|
|
Task joiner = manager.ReconnectAsync(first).AsTask();
|
|
time.Advance(TimeSpan.FromSeconds(1));
|
|
await Task.WhenAll(reconnect, joiner).WaitAsync(s_completionTimeout).ConfigureAwait(false);
|
|
|
|
Assert.That(first.State, Is.EqualTo(ChannelState.Faulted));
|
|
Assert.That(second.State, Is.EqualTo(ChannelState.Faulted));
|
|
Assert.That(GetInternalIntProperty(first, "SwapCount"), Is.Zero);
|
|
Assert.That(GetInternalIntProperty(second, "SwapCount"), Is.Zero);
|
|
shortParticipant.Verify(value => value.CreateReconnectBudget(time), Times.Once);
|
|
longParticipant.Verify(value => value.CreateReconnectBudget(time), Times.Once);
|
|
transport.Verify(value => value.ReconnectAsync(
|
|
It.IsAny<ITransportWaitingConnection?>(), It.IsAny<CancellationToken>()), Times.Once);
|
|
}
|
|
finally
|
|
{
|
|
release.TrySetResult(true);
|
|
await manager.DisposeAsync().ConfigureAwait(false);
|
|
certificate.Dispose();
|
|
}
|
|
|
|
Mock<IReconnectParticipant> CreateParticipant(string id, TimeSpan duration)
|
|
{
|
|
var participant = new Mock<IReconnectParticipant>();
|
|
participant.SetupGet(value => value.Id).Returns(id);
|
|
participant.SetupGet(value => value.Endpoint).Returns(GetTestEndpoint(certificate));
|
|
participant.Setup(value => value.CreateReconnectBudget(time))
|
|
.Returns(() => new RetryBudget(duration, time));
|
|
return participant;
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Provisional Ready does not complete recovery before restoration succeeds or fully unwinds on expiry.
|
|
/// </summary>
|
|
[TestCase(false, false)]
|
|
[TestCase(false, true)]
|
|
[TestCase(true, false)]
|
|
[TestCase(true, true)]
|
|
public async Task ReconnectDeadlineIncludesRestorationAfterProvisionalReadyAsync(
|
|
bool succeeds,
|
|
bool stalledTransport)
|
|
{
|
|
var time = new FakeTimeProvider();
|
|
var entered = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var release = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var policy = new ExponentialBackoffChannelReconnectPolicy
|
|
{
|
|
MinDelay = TimeSpan.Zero,
|
|
MaxDelay = TimeSpan.Zero,
|
|
ParticipantTimeout = Timeout.InfiniteTimeSpan
|
|
};
|
|
(ClientChannelManager manager, Certificate certificate, Mock<IChannel> transport) =
|
|
CreateMockedSut(reconnectPolicy: policy, timeProvider: time);
|
|
bool unwound = false;
|
|
bool unwoundBeforeFault = false;
|
|
int faults = 0;
|
|
IManagedTransportChannel? owningChannel = null;
|
|
try
|
|
{
|
|
transport.Setup(value => value.SendRequestAsync(
|
|
It.IsAny<IServiceRequest>(), It.IsAny<CancellationToken>()))
|
|
.Returns(async () =>
|
|
{
|
|
entered.TrySetResult(true);
|
|
await release.Task.ConfigureAwait(false);
|
|
return new CreateSubscriptionResponse();
|
|
});
|
|
var participant = new Mock<IChannelRecoveryParticipant>();
|
|
participant.SetupGet(value => value.Id).Returns("restoring");
|
|
participant.SetupGet(value => value.Endpoint).Returns(GetTestEndpoint(certificate));
|
|
participant.Setup(value => value.OnReconnectAsync(
|
|
It.IsAny<IManagedTransportChannel>(), It.IsAny<ITransportChannel>(),
|
|
0, It.IsAny<CancellationToken>()))
|
|
.Returns((IManagedTransportChannel owner, ITransportChannel _, int _, CancellationToken _) =>
|
|
{
|
|
owningChannel = owner;
|
|
return new ValueTask<ParticipantReconnectResult>(ParticipantReconnectResult.Reactivated);
|
|
});
|
|
participant.Setup(value => value.CompleteRecoveryAsync(It.IsAny<CancellationToken>()))
|
|
.Returns(async (CancellationToken ct) =>
|
|
{
|
|
try
|
|
{
|
|
if (stalledTransport)
|
|
{
|
|
await owningChannel!.SendRequestAsync(new CreateSubscriptionRequest(), ct)
|
|
.ConfigureAwait(false);
|
|
}
|
|
else
|
|
{
|
|
entered.TrySetResult(true);
|
|
await release.Task.WaitAsync(ct).ConfigureAwait(false);
|
|
}
|
|
}
|
|
finally
|
|
{
|
|
unwound = true;
|
|
}
|
|
});
|
|
using IManagedTransportChannel lease = await manager.GetAsync(participant.Object).ConfigureAwait(false);
|
|
lease.StateChanged += (_, change) =>
|
|
{
|
|
if (change.NewState == ChannelState.Faulted)
|
|
{
|
|
unwoundBeforeFault = unwound;
|
|
faults++;
|
|
}
|
|
};
|
|
Task reconnect = manager.ReconnectAsync(lease, new RetryBudget(TimeSpan.FromSeconds(1), time)).AsTask();
|
|
await entered.Task.WaitAsync(s_completionTimeout).ConfigureAwait(false);
|
|
Assert.That(lease.State, Is.EqualTo(ChannelState.Ready));
|
|
time.Advance(TimeSpan.FromSeconds(1) - TimeSpan.FromTicks(1));
|
|
Assert.That(reconnect.IsCompleted, Is.False);
|
|
Assert.That(unwound, Is.False);
|
|
|
|
if (succeeds)
|
|
{
|
|
release.TrySetResult(true);
|
|
await reconnect.WaitAsync(s_completionTimeout).ConfigureAwait(false);
|
|
Assert.That(lease.State, Is.EqualTo(ChannelState.Ready));
|
|
}
|
|
else
|
|
{
|
|
time.Advance(TimeSpan.FromTicks(1));
|
|
await AssertThrowsAsync<ServiceResultException>(reconnect, s_completionTimeout)
|
|
.ConfigureAwait(false);
|
|
Assert.That(lease.State, Is.EqualTo(ChannelState.Faulted));
|
|
Assert.That(unwoundBeforeFault, Is.True);
|
|
}
|
|
|
|
time.Advance(TimeSpan.FromSeconds(10));
|
|
Assert.That(unwound, Is.True);
|
|
Assert.That(faults, Is.EqualTo(succeeds ? 0 : 1));
|
|
}
|
|
finally
|
|
{
|
|
release.TrySetResult(true);
|
|
await manager.DisposeAsync().ConfigureAwait(false);
|
|
certificate.Dispose();
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Per-callback timeouts and retries consume the same overall recovery window.
|
|
/// </summary>
|
|
[Test]
|
|
public async Task ParticipantTimeoutRetriesConsumeOneRecoveryWindowAsync()
|
|
{
|
|
var time = new ObservableFakeTimeProvider();
|
|
var completion = new TaskCompletionSource<ParticipantReconnectResult>(
|
|
TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var callbackTimeout = TimeSpan.FromMilliseconds(200);
|
|
var policy = new ExponentialBackoffChannelReconnectPolicy
|
|
{
|
|
MinDelay = TimeSpan.Zero,
|
|
MaxDelay = TimeSpan.Zero,
|
|
ParticipantTimeout = callbackTimeout
|
|
};
|
|
(ClientChannelManager manager, Certificate certificate, _) =
|
|
CreateMockedSut(reconnectPolicy: policy, timeProvider: time);
|
|
try
|
|
{
|
|
var participant = new Mock<IReconnectParticipant>();
|
|
participant.SetupGet(value => value.Id).Returns("partial-progress");
|
|
participant.SetupGet(value => value.Endpoint).Returns(GetTestEndpoint(certificate));
|
|
participant.Setup(value => value.OnReconnectAsync(
|
|
It.IsAny<IManagedTransportChannel>(), It.Is<int>(attempt => attempt >= 0),
|
|
It.IsAny<CancellationToken>()))
|
|
.Returns(() => new ValueTask<ParticipantReconnectResult>(completion.Task));
|
|
using IManagedTransportChannel lease = await manager.GetAsync(participant.Object).ConfigureAwait(false);
|
|
Task timer = time.WaitForTimersCreatedAsync(dueTime: callbackTimeout);
|
|
Task reconnect = manager.ReconnectAsync(
|
|
lease, new RetryBudget(TimeSpan.FromMilliseconds(500), time)).AsTask();
|
|
await timer.WaitAsync(s_completionTimeout).ConfigureAwait(false);
|
|
|
|
for (int i = 0; i < 2; i++)
|
|
{
|
|
timer = time.WaitForTimersCreatedAsync(dueTime: callbackTimeout);
|
|
time.Advance(callbackTimeout);
|
|
await timer.WaitAsync(s_completionTimeout).ConfigureAwait(false);
|
|
}
|
|
time.Advance(TimeSpan.FromMilliseconds(99));
|
|
Assert.That(reconnect.IsCompleted, Is.False);
|
|
time.Advance(TimeSpan.FromMilliseconds(1));
|
|
await AssertThrowsAsync<ServiceResultException>(reconnect, s_completionTimeout).ConfigureAwait(false);
|
|
|
|
Assert.That(lease.State, Is.EqualTo(ChannelState.Faulted));
|
|
participant.Verify(value => value.OnReconnectAsync(
|
|
It.IsAny<IManagedTransportChannel>(), It.Is<int>(attempt => attempt >= 0),
|
|
It.IsAny<CancellationToken>()), Times.Exactly(3));
|
|
}
|
|
finally
|
|
{
|
|
completion.TrySetResult(ParticipantReconnectResult.Reactivated);
|
|
await manager.DisposeAsync().ConfigureAwait(false);
|
|
certificate.Dispose();
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Cancelling one caller neither starts an unwanted cycle nor cancels another caller's shared recovery.
|
|
/// </summary>
|
|
[TestCase(false)]
|
|
[TestCase(true)]
|
|
public async Task CallerCancellationDoesNotStartOrCancelSharedRecoveryAsync(bool alreadyCancelled)
|
|
{
|
|
using var cancellation = new CancellationTokenSource();
|
|
var entered = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var release = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var policy = new ExponentialBackoffChannelReconnectPolicy
|
|
{
|
|
MinDelay = TimeSpan.Zero,
|
|
MaxDelay = TimeSpan.Zero,
|
|
ParticipantTimeout = Timeout.InfiniteTimeSpan
|
|
};
|
|
(ClientChannelManager manager, Certificate certificate, Mock<IChannel> transport) =
|
|
CreateMockedSut(reconnectPolicy: policy);
|
|
try
|
|
{
|
|
transport.SetupGet(value => value.SupportedFeatures).Returns(TransportChannelFeatures.Reconnect);
|
|
transport.Setup(value => value.ReconnectAsync(
|
|
It.IsAny<ITransportWaitingConnection?>(), It.IsAny<CancellationToken>()))
|
|
.Returns(async (ITransportWaitingConnection? _, CancellationToken ct) =>
|
|
{
|
|
entered.TrySetResult(true);
|
|
await release.Task.WaitAsync(ct).ConfigureAwait(false);
|
|
});
|
|
using IManagedTransportChannel lease = await manager.GetAsync(
|
|
new TestParticipant("cancelled-caller", GetTestEndpoint(certificate))).ConfigureAwait(false);
|
|
if (alreadyCancelled)
|
|
{
|
|
await cancellation.CancelAsync().ConfigureAwait(false);
|
|
}
|
|
Task reconnect = manager.ReconnectAsync(lease, cancellation.Token).AsTask();
|
|
if (!alreadyCancelled)
|
|
{
|
|
await entered.Task.WaitAsync(s_completionTimeout).ConfigureAwait(false);
|
|
await cancellation.CancelAsync().ConfigureAwait(false);
|
|
Assert.That(lease.State, Is.EqualTo(ChannelState.TransportReconnecting));
|
|
}
|
|
await Assert.ThatAsync(
|
|
() => reconnect.WaitAsync(s_completionTimeout),
|
|
Throws.InstanceOf<OperationCanceledException>()).ConfigureAwait(false);
|
|
if (!alreadyCancelled)
|
|
{
|
|
Task joiner = manager.ReconnectAsync(lease).AsTask();
|
|
release.TrySetResult(true);
|
|
await joiner.WaitAsync(s_completionTimeout).ConfigureAwait(false);
|
|
}
|
|
|
|
Assert.That(lease.State, Is.EqualTo(ChannelState.Ready));
|
|
transport.Verify(value => value.ReconnectAsync(
|
|
It.IsAny<ITransportWaitingConnection?>(), It.IsAny<CancellationToken>()),
|
|
alreadyCancelled ? Times.Never() : Times.Once());
|
|
}
|
|
finally
|
|
{
|
|
release.TrySetResult(true);
|
|
await manager.DisposeAsync().ConfigureAwait(false);
|
|
certificate.Dispose();
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// A participant budget failure faults the shared channel and notifies every participant
|
|
/// without starting transport recovery or replacing the failed budget with an unlimited one.
|
|
/// </summary>
|
|
[Test]
|
|
[Combinatorial]
|
|
public async Task ParticipantBudgetFailuresFaultSharedRecoveryAndNotifyAllParticipantsAsync(
|
|
[Values] bool boundedCaller,
|
|
[Values("factory", "consume", "exhaustion", "expiry")] string failureStage,
|
|
[Values] bool cancellationException)
|
|
{
|
|
var time = new FakeTimeProvider();
|
|
var policy = new ExponentialBackoffChannelReconnectPolicy
|
|
{
|
|
MinDelay = TimeSpan.Zero,
|
|
MaxDelay = TimeSpan.Zero
|
|
};
|
|
(ClientChannelManager manager, Certificate certificate, Mock<IChannel> transport) =
|
|
CreateMockedSut(reconnectPolicy: policy, timeProvider: time);
|
|
try
|
|
{
|
|
Exception budgetError = cancellationException
|
|
? new OperationCanceledException("Participant budget failed without cancellation.")
|
|
: new InvalidOperationException("Participant budget failed.");
|
|
var notifications = new ConcurrentQueue<(string Participant, int Attempt, bool Cancelled)>();
|
|
Mock<IReconnectParticipant> faulty = CreateParticipant("faulty");
|
|
Mock<IReconnectParticipant> healthy = CreateParticipant("healthy");
|
|
healthy.Setup(value => value.CreateReconnectBudget(time))
|
|
.Returns(() => new RetryBudget(TimeSpan.FromMinutes(1), time));
|
|
if (failureStage == "factory")
|
|
{
|
|
faulty.Setup(value => value.CreateReconnectBudget(time)).Throws(budgetError);
|
|
}
|
|
else
|
|
{
|
|
var budget = new Mock<IRetryBudget>();
|
|
var remaining = TimeSpan.FromSeconds(1);
|
|
if (failureStage == "consume")
|
|
{
|
|
budget.Setup(value => value.TryConsume(out remaining)).Throws(budgetError);
|
|
}
|
|
else
|
|
{
|
|
budget.Setup(value => value.TryConsume(out remaining)).Returns(true);
|
|
budget.SetupGet(value => value.IsExhausted)
|
|
.Callback(() =>
|
|
{
|
|
if (failureStage == "expiry")
|
|
{
|
|
time.Advance(remaining);
|
|
}
|
|
})
|
|
.Throws(budgetError);
|
|
}
|
|
faulty.Setup(value => value.CreateReconnectBudget(time)).Returns(budget.Object);
|
|
}
|
|
transport.SetupGet(value => value.SupportedFeatures).Returns(TransportChannelFeatures.Reconnect);
|
|
transport.Setup(value => value.ReconnectAsync(
|
|
It.IsAny<ITransportWaitingConnection?>(), It.IsAny<CancellationToken>()))
|
|
.Returns(new ValueTask());
|
|
using IManagedTransportChannel first = await manager.GetAsync(faulty.Object).ConfigureAwait(false);
|
|
using IManagedTransportChannel second = await manager.GetAsync(healthy.Object).ConfigureAwait(false);
|
|
var faults = new ConcurrentQueue<int>();
|
|
first.StateChanged += (_, change) =>
|
|
{
|
|
if (change.NewState == ChannelState.Faulted)
|
|
{
|
|
faults.Enqueue(notifications.Count);
|
|
}
|
|
};
|
|
Task reconnect = boundedCaller
|
|
? manager.ReconnectAsync(second, new RetryBudget(TimeSpan.FromMinutes(1), time)).AsTask()
|
|
: manager.ReconnectAsync(second).AsTask();
|
|
Exception? reconnectError = null;
|
|
try
|
|
{
|
|
await reconnect.WaitAsync(s_completionTimeout).ConfigureAwait(false);
|
|
}
|
|
catch (Exception error) when (
|
|
error is ServiceResultException or InvalidOperationException or OperationCanceledException)
|
|
{
|
|
reconnectError = error;
|
|
}
|
|
|
|
Assert.Multiple(() =>
|
|
{
|
|
Assert.That(first.State, Is.EqualTo(ChannelState.Faulted));
|
|
Assert.That(second.State, Is.EqualTo(ChannelState.Faulted));
|
|
Assert.That(notifications, Is.EquivalentTo(
|
|
[
|
|
("faulty", -1, false),
|
|
("healthy", -1, false)
|
|
]));
|
|
Assert.That(faults, Has.Count.EqualTo(1).And.All.EqualTo(2),
|
|
"The Faulted event must follow both final notifications.");
|
|
Assert.That(GetInternalIntProperty(first, "SwapCount"), Is.Zero);
|
|
Assert.That(GetInternalIntProperty(second, "SwapCount"), Is.Zero);
|
|
Assert.That(manager.GetChannelDiagnostics(), Has.Count.EqualTo(1));
|
|
});
|
|
ServiceResult? lastError = manager.GetChannelDiagnostics().Single().LastError;
|
|
Assert.That(lastError, Is.Not.Null);
|
|
Assert.That(lastError!.StatusCode, Is.EqualTo(StatusCodes.BadSecureChannelClosed));
|
|
string expectedError = failureStage == "expiry" && cancellationException
|
|
? "Channel recovery deadline expired"
|
|
: budgetError.Message;
|
|
Assert.That(lastError.ToLongString(), Does.Contain(expectedError));
|
|
if (boundedCaller)
|
|
{
|
|
Assert.That(reconnectError, Is.TypeOf<ServiceResultException>());
|
|
Assert.That(((ServiceResultException)reconnectError!).StatusCode,
|
|
Is.EqualTo(StatusCodes.BadSecureChannelClosed));
|
|
}
|
|
else
|
|
{
|
|
Assert.That(reconnectError, Is.Null);
|
|
}
|
|
faulty.Verify(value => value.CreateReconnectBudget(time), Times.Once);
|
|
transport.Verify(value => value.ReconnectAsync(
|
|
It.IsAny<ITransportWaitingConnection?>(), It.IsAny<CancellationToken>()), Times.Never);
|
|
|
|
time.Advance(TimeSpan.FromHours(1));
|
|
Assert.That(notifications, Has.Count.EqualTo(2));
|
|
Assert.That(faults, Has.Count.EqualTo(1));
|
|
|
|
Mock<IReconnectParticipant> CreateParticipant(string id)
|
|
{
|
|
var participant = new Mock<IReconnectParticipant>();
|
|
participant.SetupGet(value => value.Id).Returns(id);
|
|
participant.SetupGet(value => value.Endpoint).Returns(GetTestEndpoint(certificate));
|
|
participant.Setup(value => value.OnReconnectAsync(
|
|
It.IsAny<IManagedTransportChannel>(), It.IsAny<int>(), It.IsAny<CancellationToken>()))
|
|
.Returns((IManagedTransportChannel _, int attempt, CancellationToken ct) =>
|
|
{
|
|
notifications.Enqueue((id, attempt, ct.IsCancellationRequested));
|
|
return new ValueTask<ParticipantReconnectResult>(ParticipantReconnectResult.Reactivated);
|
|
});
|
|
return participant;
|
|
}
|
|
}
|
|
finally
|
|
{
|
|
await manager.DisposeAsync().ConfigureAwait(false);
|
|
certificate.Dispose();
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// A throwing custom budget releases recovery ownership so a later valid request can proceed.
|
|
/// </summary>
|
|
[Test]
|
|
public async Task InvalidCustomBudgetDoesNotRetainRecoveryOwnershipAsync()
|
|
{
|
|
var policy = new ExponentialBackoffChannelReconnectPolicy
|
|
{
|
|
MinDelay = TimeSpan.Zero,
|
|
MaxDelay = TimeSpan.Zero
|
|
};
|
|
(ClientChannelManager manager, Certificate certificate, _) = CreateMockedSut(reconnectPolicy: policy);
|
|
try
|
|
{
|
|
var budget = new Mock<IRetryBudget>();
|
|
TimeSpan remaining = default;
|
|
budget.Setup(value => value.TryConsume(out remaining)).Throws<InvalidOperationException>();
|
|
using IManagedTransportChannel lease = await manager.GetAsync(
|
|
new TestParticipant("invalid-budget", GetTestEndpoint(certificate))).ConfigureAwait(false);
|
|
await AssertThrowsAsync<InvalidOperationException>(
|
|
manager.ReconnectAsync(lease, budget.Object).AsTask(), s_completionTimeout).ConfigureAwait(false);
|
|
|
|
await manager.ReconnectAsync(lease).AsTask().WaitAsync(s_completionTimeout).ConfigureAwait(false);
|
|
Assert.That(lease.State, Is.EqualTo(ChannelState.Ready));
|
|
Assert.That(GetInternalIntProperty(lease, "SwapCount"), Is.Zero);
|
|
Assert.That(manager.GetChannelDiagnostics().Single().Refcount, Is.EqualTo(1));
|
|
}
|
|
finally
|
|
{
|
|
await manager.DisposeAsync().ConfigureAwait(false);
|
|
certificate.Dispose();
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Releasing the final lease cancels an otherwise unlimited recovery cycle.
|
|
/// </summary>
|
|
[Test]
|
|
public async Task ReleasingTheLastLeaseCancelsAnUnboundedRecoveryAsync()
|
|
{
|
|
var entered = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var release = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var policy = new ExponentialBackoffChannelReconnectPolicy
|
|
{
|
|
MinDelay = TimeSpan.Zero,
|
|
MaxDelay = TimeSpan.Zero,
|
|
ParticipantTimeout = Timeout.InfiniteTimeSpan
|
|
};
|
|
(ClientChannelManager manager, Certificate certificate, Mock<IChannel> transport) =
|
|
CreateMockedSut(reconnectPolicy: policy);
|
|
try
|
|
{
|
|
var participant = new TestParticipant("disposed", GetTestEndpoint(certificate));
|
|
transport.SetupGet(value => value.SupportedFeatures).Returns(TransportChannelFeatures.Reconnect);
|
|
transport.Setup(value => value.ReconnectAsync(
|
|
It.IsAny<ITransportWaitingConnection?>(), It.IsAny<CancellationToken>()))
|
|
.Returns(async (ITransportWaitingConnection? _, CancellationToken ct) =>
|
|
{
|
|
entered.TrySetResult(true);
|
|
await release.Task.WaitAsync(ct).ConfigureAwait(false);
|
|
});
|
|
using IManagedTransportChannel lease = await manager.GetAsync(participant).ConfigureAwait(false);
|
|
Task reconnect = manager.ReconnectAsync(lease).AsTask();
|
|
await entered.Task.WaitAsync(s_completionTimeout).ConfigureAwait(false);
|
|
|
|
await lease.CloseAsync().ConfigureAwait(false);
|
|
await Assert.ThatAsync(
|
|
() => reconnect.WaitAsync(s_completionTimeout),
|
|
Throws.InstanceOf<OperationCanceledException>()).ConfigureAwait(false);
|
|
Assert.That(lease.State, Is.EqualTo(ChannelState.Closed));
|
|
Assert.That(manager.GetChannelDiagnostics(), Is.Empty);
|
|
Assert.That(participant.NotificationCount, Is.Zero);
|
|
}
|
|
finally
|
|
{
|
|
release.TrySetResult(true);
|
|
await manager.DisposeAsync().ConfigureAwait(false);
|
|
certificate.Dispose();
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// A stalled final notification cannot retain the recovery owner after a terminal outcome.
|
|
/// </summary>
|
|
[Test]
|
|
[Combinatorial]
|
|
public async Task TerminalNotificationsCannotHoldTheRecoveryOwnerAsync(
|
|
[Values("policy", "expired", "elapsed", "budget-failure")] string outcome,
|
|
[Values] bool cooperative)
|
|
{
|
|
var time = new ObservableFakeTimeProvider();
|
|
var entered = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var transportEntered = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var transportRelease = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var release = new TaskCompletionSource<ParticipantReconnectResult>(
|
|
TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var completed = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var policy = new ExponentialBackoffChannelReconnectPolicy
|
|
{
|
|
MinDelay = TimeSpan.Zero,
|
|
MaxDelay = TimeSpan.Zero,
|
|
MaxAttempts = outcome == "elapsed" ? 1 : 0,
|
|
ParticipantTimeout = Timeout.InfiniteTimeSpan
|
|
};
|
|
(ClientChannelManager manager, Certificate certificate, Mock<IChannel> transport) =
|
|
CreateMockedSut(reconnectPolicy: policy, timeProvider: time);
|
|
CancellationToken notificationToken = default;
|
|
bool? completedAtFault = null;
|
|
try
|
|
{
|
|
if (outcome == "elapsed")
|
|
{
|
|
transport.SetupGet(value => value.SupportedFeatures).Returns(TransportChannelFeatures.Reconnect);
|
|
transport.Setup(value => value.ReconnectAsync(
|
|
It.IsAny<ITransportWaitingConnection?>(), It.IsAny<CancellationToken>()))
|
|
.Returns((ITransportWaitingConnection? _, CancellationToken ct) =>
|
|
{
|
|
transportEntered.TrySetResult(true);
|
|
return new ValueTask(transportRelease.Task.WaitAsync(ct));
|
|
});
|
|
}
|
|
var participant = new Mock<IReconnectParticipant>();
|
|
participant.SetupGet(value => value.Id).Returns("stalled-final-notification");
|
|
participant.SetupGet(value => value.Endpoint).Returns(GetTestEndpoint(certificate));
|
|
if (outcome == "budget-failure")
|
|
{
|
|
participant.Setup(value => value.CreateReconnectBudget(time))
|
|
.Throws<InvalidOperationException>();
|
|
}
|
|
participant.Setup(value => value.OnReconnectAsync(
|
|
It.IsAny<IManagedTransportChannel>(), -1, It.IsAny<CancellationToken>()))
|
|
.Returns(async (IManagedTransportChannel _, int _, CancellationToken ct) =>
|
|
{
|
|
notificationToken = ct;
|
|
entered.TrySetResult(true);
|
|
ParticipantReconnectResult result = cooperative
|
|
? await release.Task.WaitAsync(ct).ConfigureAwait(false)
|
|
: await release.Task.ConfigureAwait(false);
|
|
completed.TrySetResult(true);
|
|
return result;
|
|
});
|
|
using IManagedTransportChannel lease = await manager.GetAsync(participant.Object).ConfigureAwait(false);
|
|
lease.StateChanged += (_, change) =>
|
|
{
|
|
if (change.NewState == ChannelState.Faulted)
|
|
{
|
|
completedAtFault = completed.Task.IsCompleted;
|
|
}
|
|
};
|
|
Task timer = time.WaitForTimersCreatedAsync(dueTime: TimeSpan.FromSeconds(5));
|
|
TimeSpan budget = outcome switch
|
|
{
|
|
"expired" => TimeSpan.Zero,
|
|
"elapsed" => TimeSpan.FromSeconds(1),
|
|
_ => Timeout.InfiniteTimeSpan
|
|
};
|
|
Task reconnect = manager.ReconnectAsync(lease, new RetryBudget(budget, time)).AsTask();
|
|
if (outcome == "elapsed")
|
|
{
|
|
await transportEntered.Task.WaitAsync(s_completionTimeout).ConfigureAwait(false);
|
|
time.Advance(budget);
|
|
}
|
|
await entered.Task.WaitAsync(s_completionTimeout).ConfigureAwait(false);
|
|
Assert.That(notificationToken.CanBeCanceled, Is.True);
|
|
Assert.That(notificationToken.IsCancellationRequested, Is.False,
|
|
"Final notification must not inherit the already-expired recovery deadline.");
|
|
await timer.WaitAsync(s_completionTimeout).ConfigureAwait(false);
|
|
Assert.That(reconnect.IsCompleted, Is.False);
|
|
Assert.That(lease.State, Is.Not.EqualTo(ChannelState.Faulted));
|
|
if (cooperative)
|
|
{
|
|
release.TrySetResult(ParticipantReconnectResult.Reactivated);
|
|
}
|
|
else
|
|
{
|
|
time.Advance(TimeSpan.FromSeconds(5) - TimeSpan.FromTicks(1));
|
|
Assert.That(reconnect.IsCompleted, Is.False);
|
|
Assert.That(lease.State, Is.Not.EqualTo(ChannelState.Faulted));
|
|
time.Advance(TimeSpan.FromTicks(1));
|
|
}
|
|
await AssertThrowsAsync<ServiceResultException>(reconnect, s_completionTimeout).ConfigureAwait(false);
|
|
|
|
Assert.That(lease.State, Is.EqualTo(ChannelState.Faulted));
|
|
Assert.That(completedAtFault, Is.EqualTo(cooperative),
|
|
"The Faulted event must observe the completed cooperative callback.");
|
|
Assert.That(completed.Task.IsCompleted, Is.EqualTo(cooperative));
|
|
Assert.That(notificationToken.IsCancellationRequested, Is.True);
|
|
Assert.That(GetInternalIntProperty(lease, "SwapCount"), Is.Zero);
|
|
participant.Verify(value => value.OnReconnectAsync(
|
|
It.IsAny<IManagedTransportChannel>(), -1, It.IsAny<CancellationToken>()), Times.Once);
|
|
}
|
|
finally
|
|
{
|
|
transportRelease.TrySetResult(true);
|
|
release.TrySetResult(ParticipantReconnectResult.Reactivated);
|
|
await manager.DisposeAsync().ConfigureAwait(false);
|
|
certificate.Dispose();
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task TerminalNotificationTokenRemainsLinkedToManagerShutdownAsync()
|
|
{
|
|
var time = new ObservableFakeTimeProvider();
|
|
CancellationToken notificationToken = default;
|
|
var entered = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var cancelled = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var release = new TaskCompletionSource<ParticipantReconnectResult>(
|
|
TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var policy = new ExponentialBackoffChannelReconnectPolicy
|
|
{
|
|
MinDelay = TimeSpan.Zero,
|
|
MaxDelay = TimeSpan.Zero,
|
|
MaxAttempts = 0,
|
|
ParticipantTimeout = Timeout.InfiniteTimeSpan
|
|
};
|
|
(ClientChannelManager manager, Certificate certificate, _) =
|
|
CreateMockedSut(reconnectPolicy: policy, timeProvider: time);
|
|
try
|
|
{
|
|
var participant = new Mock<IReconnectParticipant>();
|
|
participant.SetupGet(value => value.Id).Returns("shutdown-final-notification");
|
|
participant.SetupGet(value => value.Endpoint).Returns(GetTestEndpoint(certificate));
|
|
participant.Setup(value => value.OnReconnectAsync(
|
|
It.IsAny<IManagedTransportChannel>(), -1, It.IsAny<CancellationToken>()))
|
|
.Returns(async (IManagedTransportChannel _, int _, CancellationToken ct) =>
|
|
{
|
|
notificationToken = ct;
|
|
entered.TrySetResult(true);
|
|
try
|
|
{
|
|
return await release.Task.WaitAsync(ct).ConfigureAwait(false);
|
|
}
|
|
finally
|
|
{
|
|
cancelled.TrySetResult(ct.IsCancellationRequested);
|
|
}
|
|
});
|
|
using IManagedTransportChannel lease = await manager.GetAsync(participant.Object).ConfigureAwait(false);
|
|
Task reconnect = manager.ReconnectAsync(
|
|
lease, new RetryBudget(Timeout.InfiniteTimeSpan, time)).AsTask();
|
|
await entered.Task.WaitAsync(s_completionTimeout).ConfigureAwait(false);
|
|
Assert.That(notificationToken.CanBeCanceled, Is.True);
|
|
Assert.That(notificationToken.IsCancellationRequested, Is.False);
|
|
|
|
await manager.DisposeAsync().AsTask().WaitAsync(s_completionTimeout).ConfigureAwait(false);
|
|
Assert.That(notificationToken.IsCancellationRequested, Is.True,
|
|
"Manager shutdown must cancel a final notification after its policy has been exhausted.");
|
|
Assert.That(await cancelled.Task.WaitAsync(s_completionTimeout).ConfigureAwait(false), Is.True);
|
|
ServiceResultException error = await AssertThrowsAsync<ServiceResultException>(
|
|
reconnect, s_completionTimeout).ConfigureAwait(false);
|
|
|
|
Assert.That(error.StatusCode, Is.EqualTo(StatusCodes.BadSecureChannelClosed));
|
|
Assert.That(lease.State, Is.EqualTo(ChannelState.Closed));
|
|
Assert.That(manager.GetChannelDiagnostics(), Is.Empty);
|
|
Assert.That(release.Task.IsCompleted, Is.False);
|
|
}
|
|
finally
|
|
{
|
|
release.TrySetResult(ParticipantReconnectResult.Reactivated);
|
|
await manager.DisposeAsync().ConfigureAwait(false);
|
|
certificate.Dispose();
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// A transport that returns after cancellation is closed instead of reviving the expired channel.
|
|
/// </summary>
|
|
[Test]
|
|
public async Task CancellationIgnoringTransportIsClosedAfterLateReconnectAsync()
|
|
{
|
|
var time = new FakeTimeProvider();
|
|
var entered = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var release = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var closed = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var policy = new ExponentialBackoffChannelReconnectPolicy
|
|
{
|
|
MinDelay = TimeSpan.Zero,
|
|
MaxDelay = TimeSpan.Zero
|
|
};
|
|
(ClientChannelManager manager, Certificate certificate, Mock<IChannel> transport) =
|
|
CreateMockedSut(reconnectPolicy: policy, timeProvider: time);
|
|
try
|
|
{
|
|
transport.SetupGet(value => value.SupportedFeatures).Returns(TransportChannelFeatures.Reconnect);
|
|
transport.Setup(value => value.ReconnectAsync(
|
|
It.IsAny<ITransportWaitingConnection?>(), It.IsAny<CancellationToken>()))
|
|
.Returns(() =>
|
|
{
|
|
entered.TrySetResult(true);
|
|
return new ValueTask(release.Task);
|
|
});
|
|
transport.Setup(value => value.CloseAsync(It.IsAny<CancellationToken>()))
|
|
.Callback(() => closed.TrySetResult(true))
|
|
.Returns(new ValueTask());
|
|
using IManagedTransportChannel lease = await manager.GetAsync(
|
|
new TestParticipant("late-transport", GetTestEndpoint(certificate))).ConfigureAwait(false);
|
|
Task reconnect = manager.ReconnectAsync(
|
|
lease, new RetryBudget(TimeSpan.FromSeconds(1), time)).AsTask();
|
|
await entered.Task.WaitAsync(s_completionTimeout).ConfigureAwait(false);
|
|
time.Advance(TimeSpan.FromSeconds(1));
|
|
await AssertThrowsAsync<ServiceResultException>(reconnect, s_completionTimeout).ConfigureAwait(false);
|
|
|
|
Assert.That(closed.Task.IsCompleted, Is.False);
|
|
release.TrySetResult(true);
|
|
await closed.Task.WaitAsync(s_completionTimeout).ConfigureAwait(false);
|
|
Assert.That(lease.State, Is.EqualTo(ChannelState.Faulted));
|
|
}
|
|
finally
|
|
{
|
|
release.TrySetResult(true);
|
|
await manager.DisposeAsync().ConfigureAwait(false);
|
|
certificate.Dispose();
|
|
}
|
|
}
|
|
|
|
private static async Task ObserveCompletionAsync(Task task)
|
|
{
|
|
try
|
|
{
|
|
await task.WaitAsync(TimeSpan.FromSeconds(10)).ConfigureAwait(false);
|
|
}
|
|
catch (ServiceResultException)
|
|
{
|
|
// A terminal reconnect fault is the expected outcome of the deadline cases.
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task ReconnectAsyncWithExhaustedPolicyDoesNotSwapTheEntryAsync()
|
|
{
|
|
var timeProvider = new ObservableFakeTimeProvider();
|
|
var reconnectPolicy = new ExponentialBackoffChannelReconnectPolicy
|
|
{
|
|
MinDelay = TimeSpan.FromMilliseconds(10),
|
|
MaxDelay = TimeSpan.FromMilliseconds(10),
|
|
MaxAttempts = 0
|
|
};
|
|
(ClientChannelManager sut, Certificate serverCert, Mock<IChannel> 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);
|
|
|
|
// Deliberately generous. The policy, not the budget, is what ends this
|
|
// cycle: GetDelay returns the infinite sentinel as soon as the attempt
|
|
// count reaches MaxAttempts, before the budget is ever consulted. A
|
|
// race check that only asked whether the budget still had room would
|
|
// see plenty here, mistake the deliberate stop for a lost race against
|
|
// a concurrent close, and swap the entry to run a second, unbudgeted
|
|
// reconnect cycle behind the swap back-off.
|
|
var budget = new RetryBudget(TimeSpan.FromMinutes(1), timeProvider);
|
|
|
|
ServiceResultException ex = await AssertThrowsAsync<ServiceResultException>(
|
|
sut.ReconnectAsync(ch, budget, default).AsTask(),
|
|
s_completionTimeout).ConfigureAwait(false);
|
|
|
|
Assert.That(ex.StatusCode, Is.EqualTo(StatusCodes.BadSecureChannelClosed));
|
|
Assert.That(ch.State, Is.EqualTo(ChannelState.Faulted));
|
|
Assert.That(GetInternalIntProperty(ch, "SwapCount"), Is.Zero);
|
|
chMock.Verify(c => c.ReconnectAsync(
|
|
It.IsAny<ITransportWaitingConnection?>(),
|
|
It.IsAny<CancellationToken>()),
|
|
Times.Never);
|
|
ch.Dispose();
|
|
}
|
|
finally
|
|
{
|
|
await sut.DisposeAsync().ConfigureAwait(false);
|
|
serverCert.Dispose();
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Awaits <paramref name="task"/> and returns the exception it faulted
|
|
/// with, failing the test if it succeeded or threw something else.
|
|
/// </summary>
|
|
/// <typeparam name="TException">The expected exception type.</typeparam>
|
|
/// <remarks>
|
|
/// Deliberately not <c>Assert.ThrowsAsync</c>. That blocks the calling
|
|
/// thread until the task completes - sync over async - so on a
|
|
/// constrained CI agent it can starve the very continuation it is
|
|
/// waiting for, and it has no timeout, so the block is unbounded. These
|
|
/// tests drive a fake clock from the test thread, which makes both
|
|
/// failure modes fatal: the test cannot advance the clock while it is
|
|
/// blocked, NUnit's runner thread never returns, and the whole test host
|
|
/// hangs until the blame collector kills it.
|
|
/// </remarks>
|
|
/// <exception cref="InvalidOperationException"></exception>
|
|
private static async Task<TException> AssertThrowsAsync<TException>(
|
|
Task task,
|
|
TimeSpan timeout)
|
|
where TException : Exception
|
|
{
|
|
try
|
|
{
|
|
await task.WaitAsync(timeout).ConfigureAwait(false);
|
|
}
|
|
catch (TException expected)
|
|
{
|
|
return expected;
|
|
}
|
|
catch (Exception other)
|
|
{
|
|
Assert.Fail(
|
|
$"Expected {typeof(TException).Name} but got " +
|
|
$"{other.GetType().Name}: {other}");
|
|
throw;
|
|
}
|
|
|
|
Assert.Fail(
|
|
$"Expected {typeof(TException).Name} but the operation completed successfully.");
|
|
throw new InvalidOperationException("unreachable");
|
|
}
|
|
|
|
private static (ClientChannelManager sut, Certificate serverCert, Mock<IChannel> chMock) CreateMockedSut(
|
|
ITelemetryContext? telemetry = null,
|
|
IChannelReconnectPolicy? reconnectPolicy = null,
|
|
TimeProvider? timeProvider = null)
|
|
{
|
|
telemetry ??= NUnitTelemetryContext.Create();
|
|
Certificate serverCert = s_factory.CreateCertificate("CN=server").CreateForRSA();
|
|
|
|
var chMock = new Mock<IChannel>();
|
|
chMock.Setup(c => c.OpenAsync(
|
|
It.IsAny<Uri>(),
|
|
It.IsAny<TransportChannelSettings>(),
|
|
It.IsAny<CancellationToken>()))
|
|
.Callback<Uri, TransportChannelSettings, CancellationToken>(
|
|
(_, settings, _) => DisposeMockOwnedCertificates(settings))
|
|
.Returns(new ValueTask());
|
|
chMock.Setup(c => c.OpenAsync(
|
|
It.IsAny<ITransportWaitingConnection>(),
|
|
It.IsAny<TransportChannelSettings>(),
|
|
It.IsAny<CancellationToken>()))
|
|
.Callback<ITransportWaitingConnection, TransportChannelSettings, CancellationToken>(
|
|
(_, settings, _) => DisposeMockOwnedCertificates(settings))
|
|
.Returns(new ValueTask());
|
|
chMock.Setup(c => c.CloseAsync(It.IsAny<CancellationToken>()))
|
|
.Returns(new ValueTask());
|
|
chMock.Setup(c => c.SupportedFeatures).Returns(TransportChannelFeatures.None);
|
|
|
|
var bindings = new Mock<ITransportChannelBindings>();
|
|
bindings.Setup(b => b.Create(It.IsAny<string>(), It.IsAny<ITelemetryContext>()))
|
|
.Returns(chMock.Object);
|
|
|
|
var configuration = new ApplicationConfiguration(telemetry);
|
|
var sut = new ClientChannelManager(
|
|
configuration,
|
|
telemetry,
|
|
bindings.Object,
|
|
reconnectPolicy,
|
|
timeProvider);
|
|
return (sut, serverCert, chMock);
|
|
}
|
|
|
|
private static void DisposeMockOwnedCertificates(TransportChannelSettings settings)
|
|
{
|
|
settings.ServerCertificate?.Dispose();
|
|
settings.ServerCertificate = null;
|
|
}
|
|
|
|
private static ConfiguredEndpoint GetTestEndpoint(Certificate serverCert)
|
|
{
|
|
var endpoint = new ConfiguredEndpoint
|
|
{
|
|
Configuration = new EndpointConfiguration
|
|
{
|
|
OperationTimeout = 6000
|
|
}
|
|
};
|
|
endpoint.Description.EndpointUrl = "opc.tcp://localhost:4840";
|
|
endpoint.Description.SecurityMode = MessageSecurityMode.SignAndEncrypt;
|
|
endpoint.Description.SecurityPolicyUri = SecurityPolicies.Basic256Sha256;
|
|
endpoint.Description.ServerCertificate = serverCert.RawData.ToByteString();
|
|
return endpoint;
|
|
}
|
|
|
|
private static ConfiguredEndpoint GetNoneSecurityEndpoint(EndpointConfiguration endpointConfiguration)
|
|
{
|
|
var description = new EndpointDescription
|
|
{
|
|
EndpointUrl = "opc.tcp://localhost:4840",
|
|
SecurityMode = MessageSecurityMode.None,
|
|
SecurityPolicyUri = SecurityPolicies.None
|
|
};
|
|
description.Server.ApplicationUri = description.EndpointUrl;
|
|
description.Server.ApplicationType = ApplicationType.DiscoveryServer;
|
|
|
|
return new ConfiguredEndpoint(null, description, endpointConfiguration)
|
|
{
|
|
UpdateBeforeConnect = false
|
|
};
|
|
}
|
|
|
|
private static KeyValuePair<string, object?> Tag(string key, object? value)
|
|
{
|
|
return new KeyValuePair<string, object?>(key, value);
|
|
}
|
|
|
|
private static async Task WaitForReconnectReleaseAsync(
|
|
TaskCompletionSource<bool> reconnectEntered,
|
|
TaskCompletionSource<bool> allowReconnect)
|
|
{
|
|
reconnectEntered.TrySetResult(true);
|
|
await allowReconnect.Task.ConfigureAwait(false);
|
|
}
|
|
|
|
private static object GetLeaseEntry(IManagedTransportChannel channel)
|
|
{
|
|
return GetInternalPropertyValue(channel, "Entry");
|
|
}
|
|
|
|
private static ChannelState GetEntryState(object entry)
|
|
{
|
|
return (ChannelState)GetInternalPropertyValue(entry, "State");
|
|
}
|
|
|
|
private static int GetInternalIntProperty(object target, string propertyName)
|
|
{
|
|
return (int)GetInternalPropertyValue(target, propertyName);
|
|
}
|
|
|
|
private static object GetInternalPropertyValue(object target, string propertyName)
|
|
{
|
|
const BindingFlags flags = BindingFlags.Instance | BindingFlags.Public | BindingFlags.NonPublic;
|
|
PropertyInfo? property = target.GetType().GetProperty(propertyName, flags);
|
|
Assert.That(property, Is.Not.Null, $"Expected property {propertyName} on {target.GetType()}.");
|
|
object? value = property!.GetValue(target);
|
|
Assert.That(value, Is.Not.Null, $"Expected property {propertyName} to return a value.");
|
|
return value!;
|
|
}
|
|
|
|
private static void SetPrivateField(object target, string fieldName, object value)
|
|
{
|
|
FieldInfo? field = target.GetType().GetField(
|
|
fieldName,
|
|
BindingFlags.Instance | BindingFlags.NonPublic);
|
|
Assert.That(field, Is.Not.Null, $"Expected field {fieldName} on {target.GetType()}.");
|
|
field!.SetValue(target, value);
|
|
}
|
|
|
|
private static ManagedChannelDiagnostic GetDiagnostic(
|
|
ClientChannelManager sut,
|
|
ManagedChannelKey key)
|
|
{
|
|
return sut.GetChannelDiagnostics().Single(d => d.Key == key);
|
|
}
|
|
|
|
private static void AssertChannelDiagnostic(
|
|
ManagedChannelDiagnostic diagnostic,
|
|
ManagedChannelKey key)
|
|
{
|
|
Assert.That(diagnostic.Key, Is.EqualTo(key));
|
|
Assert.That(diagnostic.State, Is.EqualTo(ChannelState.Ready));
|
|
Assert.That(diagnostic.Refcount, Is.EqualTo(1));
|
|
Assert.That(diagnostic.ParticipantCount, Is.EqualTo(1));
|
|
Assert.That(diagnostic.OpenedAt, Is.Not.Default);
|
|
Assert.That(diagnostic.LastStateChange, Is.Not.Default);
|
|
Assert.That(diagnostic.LastReconnectAttempt, Is.Zero);
|
|
Assert.That(diagnostic.LastError, Is.Null);
|
|
}
|
|
|
|
public interface IChannel : ITransportChannel, ISecureChannel;
|
|
|
|
private sealed class ChannelActivityListener : IDisposable
|
|
{
|
|
public ChannelActivityListener()
|
|
{
|
|
m_listener = new ActivityListener
|
|
{
|
|
ShouldListenTo = source => source.Name == "Opc.Ua.ChannelManager",
|
|
Sample = (ref _) =>
|
|
ActivitySamplingResult.AllDataAndRecorded,
|
|
ActivityStopped = activity =>
|
|
{
|
|
StoppedActivities.Add(activity);
|
|
_ = m_stoppedActivity.TrySetResult(activity);
|
|
}
|
|
};
|
|
ActivitySource.AddActivityListener(m_listener);
|
|
}
|
|
|
|
public List<Activity> StoppedActivities { get; } = [];
|
|
|
|
public async Task<Activity> WaitForStoppedActivityAsync(string operationName)
|
|
{
|
|
Activity activity = await m_stoppedActivity.Task
|
|
.WaitAsync(s_completionTimeout)
|
|
.ConfigureAwait(false);
|
|
|
|
Assert.That(activity.OperationName, Is.EqualTo(operationName));
|
|
return activity;
|
|
}
|
|
|
|
public void Dispose()
|
|
{
|
|
m_listener.Dispose();
|
|
}
|
|
|
|
private readonly ActivityListener m_listener;
|
|
|
|
private readonly TaskCompletionSource<Activity> m_stoppedActivity = new(
|
|
TaskCreationOptions.RunContinuationsAsynchronously);
|
|
}
|
|
|
|
private sealed class ChannelMetricListener : IDisposable
|
|
{
|
|
public ChannelMetricListener()
|
|
{
|
|
m_listener = new MeterListener
|
|
{
|
|
InstrumentPublished = (instrument, listener) =>
|
|
{
|
|
if (instrument.Name.StartsWith("opc.ua.channel.", StringComparison.Ordinal))
|
|
{
|
|
listener.EnableMeasurementEvents(instrument);
|
|
}
|
|
}
|
|
};
|
|
m_listener.SetMeasurementEventCallback<long>(OnLongMeasurementRecorded);
|
|
m_listener.SetMeasurementEventCallback<double>(OnDoubleMeasurementRecorded);
|
|
m_listener.Start();
|
|
}
|
|
|
|
public ConcurrentQueue<MeasurementRecord> Measurements { get; } = new();
|
|
|
|
public void RecordObservableInstruments()
|
|
{
|
|
m_listener.RecordObservableInstruments();
|
|
}
|
|
|
|
public bool HasMeasurement(
|
|
string instrumentName,
|
|
params KeyValuePair<string, object?>[] tags)
|
|
{
|
|
return HasMeasurement(instrumentName, null, tags);
|
|
}
|
|
|
|
public bool HasMeasurement(
|
|
string instrumentName,
|
|
double? value,
|
|
params KeyValuePair<string, object?>[] tags)
|
|
{
|
|
// Snapshot under enumeration to avoid races with the
|
|
// metric callbacks that fire concurrently from the
|
|
// channel manager's threadpool teardown work.
|
|
foreach (MeasurementRecord m in Measurements)
|
|
{
|
|
if (m.InstrumentName == instrumentName &&
|
|
(value == null || m.Value == value.Value) &&
|
|
tags.All(tag => m.Tags.TryGetValue(tag.Key, out object? actual) &&
|
|
Equals(actual, tag.Value)))
|
|
{
|
|
return true;
|
|
}
|
|
}
|
|
return false;
|
|
}
|
|
|
|
public string FormatMeasurements()
|
|
{
|
|
var builder = new StringBuilder();
|
|
// ConcurrentQueue enumeration is snapshot-stable.
|
|
foreach (MeasurementRecord measurement in Measurements)
|
|
{
|
|
string tags = string.Join(
|
|
", ",
|
|
measurement.Tags.Select(t => $"{t.Key}={t.Value}"));
|
|
builder
|
|
.Append(measurement.InstrumentName)
|
|
.Append('=')
|
|
.Append(measurement.Value.ToString(CultureInfo.InvariantCulture))
|
|
.Append(" {")
|
|
.Append(tags)
|
|
.AppendLine("}");
|
|
}
|
|
return builder.ToString();
|
|
}
|
|
|
|
public void Dispose()
|
|
{
|
|
m_listener.Dispose();
|
|
}
|
|
|
|
private void OnLongMeasurementRecorded(
|
|
Instrument instrument,
|
|
long measurement,
|
|
ReadOnlySpan<KeyValuePair<string, object?>> tags,
|
|
object? state)
|
|
{
|
|
Measurements.Enqueue(new MeasurementRecord(instrument.Name, measurement, tags.ToArray()));
|
|
}
|
|
|
|
private void OnDoubleMeasurementRecorded(
|
|
Instrument instrument,
|
|
double measurement,
|
|
ReadOnlySpan<KeyValuePair<string, object?>> tags,
|
|
object? state)
|
|
{
|
|
Measurements.Enqueue(new MeasurementRecord(instrument.Name, measurement, tags.ToArray()));
|
|
}
|
|
|
|
private readonly MeterListener m_listener;
|
|
}
|
|
|
|
private sealed class MeasurementRecord
|
|
{
|
|
public MeasurementRecord(
|
|
string instrumentName,
|
|
double value,
|
|
KeyValuePair<string, object?>[] tags)
|
|
{
|
|
InstrumentName = instrumentName;
|
|
Value = value;
|
|
Tags = tags.ToDictionary(t => t.Key, t => t.Value);
|
|
}
|
|
|
|
public string InstrumentName { get; }
|
|
|
|
public double Value { get; }
|
|
|
|
public Dictionary<string, object?> Tags { get; }
|
|
}
|
|
|
|
private sealed class TestParticipant : IReconnectParticipant
|
|
{
|
|
private readonly Func<IManagedTransportChannel, int, CancellationToken,
|
|
ParticipantReconnectResult>? m_onReconnect;
|
|
|
|
private int m_notificationCount;
|
|
|
|
public TestParticipant(
|
|
string id,
|
|
ConfiguredEndpoint endpoint,
|
|
Func<IManagedTransportChannel, int, CancellationToken,
|
|
ParticipantReconnectResult>? onReconnect = null)
|
|
{
|
|
Id = id;
|
|
Endpoint = endpoint;
|
|
m_onReconnect = onReconnect;
|
|
}
|
|
|
|
public string Id { get; }
|
|
public ConfiguredEndpoint Endpoint { get; }
|
|
public int NotificationCount => Volatile.Read(ref m_notificationCount);
|
|
|
|
/// <inheritdoc/>
|
|
public IRetryBudget? CreateReconnectBudget(TimeProvider timeProvider)
|
|
{
|
|
return null;
|
|
}
|
|
|
|
public ValueTask<ParticipantReconnectResult> OnReconnectAsync(
|
|
IManagedTransportChannel channel,
|
|
int reconnectAttempt,
|
|
CancellationToken ct)
|
|
{
|
|
Interlocked.Increment(ref m_notificationCount);
|
|
ParticipantReconnectResult result = m_onReconnect?.Invoke(channel, reconnectAttempt, ct)
|
|
?? ParticipantReconnectResult.Reactivated;
|
|
return new ValueTask<ParticipantReconnectResult>(result);
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// A <see cref="FakeTimeProvider"/> that lets a test wait until the code
|
|
/// under test has actually registered its back-off timers.
|
|
/// </summary>
|
|
/// <remarks>
|
|
/// Waiting is deliberately <b>relative</b>: a waiter is armed for "N
|
|
/// more timers from now" rather than for "the Nth timer of the run".
|
|
/// Absolute numbering is a race - anything else that happens to create a
|
|
/// timer on this provider first (an earlier reconnect in the same test,
|
|
/// or manager housekeeping) consumes the low numbers, the waiter then
|
|
/// completes before the timer the test cares about exists, and the
|
|
/// subsequent Advance fires nothing. The reconnect is left parked on a
|
|
/// fake clock that nobody will move again, which hangs the test - and,
|
|
/// because NUnit blocks the runner thread, the whole test host.
|
|
/// </remarks>
|
|
private sealed class ObservableFakeTimeProvider : FakeTimeProvider
|
|
{
|
|
/// <inheritdoc/>
|
|
public override ITimer CreateTimer(
|
|
TimerCallback callback,
|
|
object? state,
|
|
TimeSpan dueTime,
|
|
TimeSpan period)
|
|
{
|
|
ITimer timer = base.CreateTimer(callback, state, dueTime, period);
|
|
|
|
List<TaskCompletionSource<bool>>? ready = null;
|
|
lock (m_lock)
|
|
{
|
|
for (int i = m_waiters.Count - 1; i >= 0; i--)
|
|
{
|
|
(int remaining, TimeSpan? expectedTime, TaskCompletionSource<bool> completion) = m_waiters[i];
|
|
if (expectedTime.HasValue && expectedTime.Value != dueTime)
|
|
{
|
|
continue;
|
|
}
|
|
if (--remaining == 0)
|
|
{
|
|
(ready ??= []).Add(completion);
|
|
m_waiters.RemoveAt(i);
|
|
}
|
|
else
|
|
{
|
|
m_waiters[i] = (remaining, expectedTime, completion);
|
|
}
|
|
}
|
|
}
|
|
|
|
// Completed outside the lock: the continuations run
|
|
// asynchronously, but there is no reason to hold the lock while
|
|
// handing them off.
|
|
if (ready != null)
|
|
{
|
|
foreach (TaskCompletionSource<bool> completion in ready)
|
|
{
|
|
completion.TrySetResult(true);
|
|
}
|
|
}
|
|
|
|
return timer;
|
|
}
|
|
|
|
/// <summary>
|
|
/// Returns a task that completes once <paramref name="count"/>
|
|
/// further timers have been created, counted from this call. Arm it
|
|
/// <b>before</b> starting the operation whose timers are awaited.
|
|
/// </summary>
|
|
/// <exception cref="ArgumentOutOfRangeException"></exception>
|
|
public Task<bool> WaitForTimersCreatedAsync(int count = 1, TimeSpan? dueTime = null)
|
|
{
|
|
if (count < 1)
|
|
{
|
|
throw new ArgumentOutOfRangeException(nameof(count));
|
|
}
|
|
|
|
var completion = new TaskCompletionSource<bool>(
|
|
TaskCreationOptions.RunContinuationsAsynchronously);
|
|
lock (m_lock)
|
|
{
|
|
m_waiters.Add((count, dueTime, completion));
|
|
}
|
|
return completion.Task;
|
|
}
|
|
|
|
private readonly Lock m_lock = new();
|
|
|
|
private readonly List<(int Remaining, TimeSpan? DueTime, TaskCompletionSource<bool> Completion)> m_waiters =
|
|
[];
|
|
}
|
|
}
|
|
}
|