GithubStar/UA-.NETStandard
1
0
Fork 0
mirror of https://github.com/OPCFoundation/UA-.NETStandard.git synced 2026-10-16 21:02:51 +08:00
Code Issues Packages Projects Releases Wiki Activity

ManagedSession: never skip an outstanding explicit reconnect; gate the skip tests deterministically

Browse Source 
Review feedback on #4781:

- An automatic trigger after an explicit ReconnectAsync could overwrite the
  trigger mark, and a later channel-manager reactivation then let
  HandleReconnectAsync skip the explicit reconnect. Track explicit requests
  and the number a completed reconnect has served instead; while one is
  outstanding the outer reconnect always runs, whatever triggers happen
  meanwhile. Adds ExplicitManagedReconnectRunsAfterChannelManagerReactivationAsync,
  which reproduced the skip before this change.
- The skip test relied on the manual recovery finishing inside a 3 s
  wall-clock backoff. A gated IReconnectPolicy now holds the outer reconnect in
  its backoff until the test has completed the manager recovery.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Roman Ettlinger
2026-10-14 07:40:52 +02:00
parent 3b464627a1
commit 004b5b83c3
2 changed files with 166 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Opc.Ua.Client/Session/ManagedSession.cs
+37 -7
View File
@@ -876,10 +876,12 @@ namespace Opc.Ua.Client
? null
: (_, token) => HandleManualReconnectAsync(connection, channel, ct, token);
// An explicit reconnect always runs, even after a channel recovery.
Interlocked.Exchange(ref m_reconnectTriggerMark, NoReconnectTrigger);
// An explicit reconnect always runs, even after a channel recovery, until a
// reconnect that started after this request completed.
Interlocked.Increment(ref m_explicitReconnectRequests);
if (!StateMachine.RequestReconnect(operation))
{
Interlocked.Decrement(ref m_explicitReconnectRequests);
throw new ServiceResultException(
StatusCodes.BadInvalidState, "The managed session cannot start the requested reconnect.");
}
@@ -1614,6 +1616,7 @@ namespace Opc.Ua.Client
CancellationToken workerToken)
{
using var linked = CancellationTokenSource.CreateLinkedTokenSource(callerToken, workerToken);
long explicitRequests = Interlocked.Read(ref m_explicitReconnectRequests);
try
{
using IDisposable recovery = InnerSession.DeferSubscriptionRecovery();
@@ -1636,6 +1639,7 @@ namespace Opc.Ua.Client
}
}
await RefreshRedundancyInfoBestEffortAsync(linked.Token).ConfigureAwait(false);
CompleteReconnect(explicitRequests);
return ServiceResult.Good;
}
catch (Exception exception)
@@ -1660,6 +1664,8 @@ namespace Opc.Ua.Client
return await HandleConnectAsync(ct).ConfigureAwait(false);
}
// Explicit requests made after this point are served by a later attempt.
long explicitRequests = Interlocked.Read(ref m_explicitReconnectRequests);
try
{
m_logger.ManagedSessionReconnecting();
@@ -1671,8 +1677,10 @@ namespace Opc.Ua.Client
// The channel manager may have recovered the channel while this
// reconnect waited out its backoff, e.g. when the faulted lease
// was reconnected directly. Reconnecting again would only tear
// down the freshly reactivated channel.
if (ChannelReactivatedSinceReconnectTrigger(session))
// down the freshly reactivated channel. An explicit reconnect
// that is still outstanding always runs.
if (explicitRequests <= Interlocked.Read(ref m_explicitReconnectsServed) &&
ChannelReactivatedSinceReconnectTrigger(session))
{
Interlocked.Exchange(ref m_reconnectTriggerMark, NoReconnectTrigger);
m_reconnectPolicy.Reset();
@@ -1756,7 +1764,7 @@ namespace Opc.Ua.Client
await RefreshRedundancyInfoBestEffortAsync(ct).ConfigureAwait(false);
Interlocked.Exchange(ref m_reconnectTriggerMark, NoReconnectTrigger);
CompleteReconnect(explicitRequests);
m_reconnectPolicy.Reset();
m_logger.ManagedSessionReconnected();
@@ -2439,6 +2447,20 @@ namespace Opc.Ua.Client
}
}
/// <summary>
/// Records a completed reconnect: the pending trigger is consumed and every
/// explicit request made before the reconnect started is served.
/// </summary>
private void CompleteReconnect(long explicitRequests)
{
Interlocked.Exchange(ref m_reconnectTriggerMark, NoReconnectTrigger);
if (explicitRequests > Interlocked.Read(ref m_explicitReconnectsServed))
{
// Only the state machine worker completes reconnects.
Interlocked.Exchange(ref m_explicitReconnectsServed, explicitRequests);
}
}
/// <summary>
/// Whether the channel manager reactivated the session after the pending
/// outer reconnect was requested and the channel is still ready, in which
@@ -2996,11 +3018,19 @@ namespace Opc.Ua.Client
/// <summary>
/// <see cref="m_channelSessionReactivations"/> when a channel fault or a
/// keep-alive failure last requested an outer reconnect, or
/// <see cref="NoReconnectTrigger"/> when the pending reconnect must run
/// regardless (an explicit request, or none recorded).
/// <see cref="NoReconnectTrigger"/> when none is recorded.
/// </summary>
private long m_reconnectTriggerMark = NoReconnectTrigger;
private const long NoReconnectTrigger = -1;
/// <summary>
/// Explicit <see cref="ReconnectAsync(ITransportWaitingConnection?, ITransportChannel?, CancellationToken)"/>
/// requests, and how many of them a completed reconnect has served. While a
/// request is outstanding the outer reconnect is never skipped, whatever
/// automatic triggers happen meanwhile.
/// </summary>
private long m_explicitReconnectRequests;
private long m_explicitReconnectsServed;
private ServerRedundancyInfo? m_redundancyInfo;
private Task<ServerRedundancyInfo>? m_redundancyRefreshTask;
private Task<ServerRedundancyInfo>? m_redundancyEndpointRefreshTask;
tests/Opc.Ua.Client.Tests/Session/SessionBindingLiveTests.Transport.cs
+129 -19
View File
@@ -218,36 +218,22 @@ namespace Opc.Ua.Client.Tests
using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(30));
var channels = new BarrierChannelBindings();
// The outer backoff outlasts the manual recovery below (about one second).
var policy = new ReconnectPolicy
{
InitialDelay = TimeSpan.FromSeconds(3),
JitterFactor = 0
};
using var policy = new GatedReconnectPolicy();
await WithSessionAsync(async session =>
{
var managed = (Client.ManagedSession)session;
var lease = (IManagedTransportChannel)session.TransportChannel;
try
{
channels.FailConnect = true;
var budget = new RetryBudget(TimeSpan.FromSeconds(10), TimeProvider.System);
var manager = (ClientChannelManager)lease.Manager;
await Assert.ThatAsync(async () => await manager.ReconnectAsync(lease, budget, timeout.Token)
.ConfigureAwait(false),
Throws.TypeOf<ServiceResultException>()).ConfigureAwait(false);
Assert.That(lease.State, Is.EqualTo(ChannelState.Faulted));
while (!managed.Reconnecting)
{
await Task.Delay(10, timeout.Token).ConfigureAwait(false);
}
await FaultLeaseAsync(lease, channels, timeout.Token).ConfigureAwait(false);
channels.FailConnect = false;
// The outer reconnect is held in its backoff until the manager recovered.
await policy.Entered.WaitAsync(timeout.Token).ConfigureAwait(false);
await lease.Manager.ReconnectAsync(lease, timeout.Token).ConfigureAwait(false);
Assert.That(managed.Reconnecting, Is.True, "The outer reconnect is still waiting.");
int connections = channels.Connections;
policy.Release();
await managed.StateMachine.WaitForConnectedAsync(timeout.Token).ConfigureAwait(false);
Assert.That(lease.State, Is.EqualTo(ChannelState.Ready));
Assert.That(channels.Connections, Is.EqualTo(connections),
@@ -259,10 +245,134 @@ namespace Opc.Ua.Client.Tests
finally
{
channels.FailConnect = false;
policy.Release();
}
}, timeout.Token, channels, policy).ConfigureAwait(false);
}
/// <summary>
/// Verifies that an explicit reconnect still reconnects when it joins a pending
/// outer reconnect, a later fault triggers it again, and the channel manager
/// reactivates the session meanwhile.
/// </summary>
[Test]
public async Task ExplicitManagedReconnectRunsAfterChannelManagerReactivationAsync()
{
if (!m_managed)
{
Assert.Ignore("Only a managed session runs an outer reconnect.");
}
using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(30));
var channels = new BarrierChannelBindings();
using var policy = new GatedReconnectPolicy();
await WithSessionAsync(async session =>
{
var managed = (Client.ManagedSession)session;
var lease = (IManagedTransportChannel)session.TransportChannel;
try
{
await FaultLeaseAsync(lease, channels, timeout.Token).ConfigureAwait(false);
await policy.Entered.WaitAsync(timeout.Token).ConfigureAwait(false);
Task reconnect = managed.ReconnectAsync(null, null, timeout.Token);
// An automatic trigger (another fault) after the explicit request,
// followed by a channel manager recovery, must not turn the explicit
// reconnect into a skippable one.
await lease.Manager.ReconnectAsync(lease, timeout.Token).ConfigureAwait(false);
await FaultLeaseAsync(lease, channels, timeout.Token).ConfigureAwait(false);
await lease.Manager.ReconnectAsync(lease, timeout.Token).ConfigureAwait(false);
Assert.That(reconnect.IsCompleted, Is.False, "The outer reconnect is still waiting.");
int connections = channels.Connections;
policy.Release();
await reconnect.ConfigureAwait(false);
Assert.That(lease.State, Is.EqualTo(ChannelState.Ready));
Assert.That(channels.Connections, Is.GreaterThan(connections),
"An explicit reconnect must not be skipped.");
DataValue value = await session.ReadValueAsync(
VariableIds.Server_ServerStatus_State, timeout.Token).ConfigureAwait(false);
Assert.That(value.StatusCode, Is.EqualTo(StatusCodes.Good));
}
finally
{
channels.FailConnect = false;
policy.Release();
}
}, timeout.Token, channels, policy).ConfigureAwait(false);
}
/// <summary>
/// Exhausts the channel manager's reconnect policy so the lease faults and the
/// managed session's outer reconnect takes over, then lets connects succeed again.
/// </summary>
private static async Task FaultLeaseAsync(
IManagedTransportChannel lease,
BarrierChannelBindings channels,
CancellationToken ct)
{
channels.FailConnect = true;
try
{
var budget = new RetryBudget(TimeSpan.FromSeconds(10), TimeProvider.System);
var manager = (ClientChannelManager)lease.Manager;
await Assert.ThatAsync(async () => await manager.ReconnectAsync(lease, budget, ct)
.ConfigureAwait(false),
Throws.TypeOf<ServiceResultException>()).ConfigureAwait(false);
Assert.That(lease.State, Is.EqualTo(ChannelState.Faulted));
}
finally
{
channels.FailConnect = false;
}
}
/// <summary>
/// Holds the first outer reconnect in its backoff until released, so a test can
/// order the channel manager's recovery before it without depending on timing.
/// </summary>
private sealed class GatedReconnectPolicy : IReconnectPolicy, IDisposable
{
public Task Entered => m_entered.Task;
public void Release()
{
m_gate.Set();
}
public TimeSpan? GetNextDelay(int attempt, CancellationToken ct = default)
{
m_entered.TrySetResult(true);
m_gate.Wait(ct);
return TimeSpan.Zero;
}
public bool TryGetNextDelay(
int attempt,
StatusCode lastStatus,
TimeSpan? serverRetryAfter,
out TimeSpan? delay,
CancellationToken ct = default)
{
delay = null;
return false;
}
public void Reset()
{
}
public void Dispose()
{
m_gate.Dispose();
}
private readonly TaskCompletionSource<bool> m_entered =
new(TaskCreationOptions.RunContinuationsAsynchronously);
private readonly ManualResetEventSlim m_gate = new();
}
private sealed class BarrierChannelBindings : ITransportChannelBindings, IUaSCByteTransportFactory
{
/// <summary>