From b36cacbeac23e658b28381ddb5db89a72b18d73e Mon Sep 17 00:00:00 2001 From: romanett Date: Wed, 14 Oct 2026 10:02:52 +0200 Subject: [PATCH] ManagedSession: skip redundant outer reconnect after channel-manager recovery (fixes BoundClientCannotRetryAcrossFaultedManagedEntryReplacementAsync flake) (#4781) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Problem `SessionBindingLiveTests.BoundClientCannotRetryAcrossFaultedManagedEntryReplacementAsync` (added in #4752) failed once on CI (`test-windows-net10.0 (2/15)` of #4775, run 38508438526): ``` Assert.That(lease.State, Is.EqualTo(ChannelState.Ready)) Expected: Ready But was: TransportReconnecting ``` ## Root cause Only the `ManagedSession` fixture variant (`[TestFixture(true)]`) fails; the plain-`Session` run in the same job passed. From the failing run's log: - 28.476: the budgeted reconnect is exhausted and the channel goes `Faulted`. `ManagedSession.OnManagedChannelStateChanged` calls `TriggerReconnect`, so the outer `ConnectionStateMachine` takes over by design: `Reconnect attempt 0, delay 1025 ms`. - 28.976 – 29.502: the test's manual `ReconnectAsync` swaps in a fresh entry and runs its reconnect cycle, which reactivates the session. `Reconnect completed ... Outcome=success` is logged and the state is `Ready`. - 29.502: the outer backoff expires (`ManagedSession: Reconnecting.`). `HandleReconnectAsync` runs a full transport reconnect on the channel that was just recovered, and the state goes `Ready -> TransportReconnecting` just before the test reads `lease.State`. `ClientChannelManager.ReconnectAsync` behaves correctly: it returns only after its own swap and cycle have completed. The defect is the redundant outer reconnect. It disconnects a freshly reactivated channel and costs an extra transport connection (which could also break the test's "no retry" connection-count assertion). ## Fix **`ManagedSession`** - Count channel-driven session reactivations: `Ready` transitions out of `TransportConnectedSessionReactivating` on this session's lease. A participant that failed fatally is removed from the entry before `Ready`, so it is never counted. - When a channel fault (`Faulted`/`Closed`) or a keep-alive failure requests the outer reconnect, record the current count as the trigger mark. - `HandleReconnectAsync` (under the recovery writer lock) skips the reconnect and reports success if a reactivation happened after the mark, the channel is still `Ready` and no channel recovery is in progress. It logs `ManagedSession: Reconnect skipped; the channel manager already reactivated the session.` (`ManagedSession + 30`). - Explicit `ManagedSession.ReconnectAsync` requests are counted separately (requested vs. served by a completed reconnect that started after them). While one is outstanding, the outer reconnect is never skipped, whatever automatic triggers happen meanwhile. A trigger the state machine rejects clears the mark, and overwriting the mark with a later trigger only makes skipping less likely. **Tests** - New `ManagedSessionSkipsOuterReconnectAfterChannelManagerReactivationAsync` (managed variant only). A gated `IReconnectPolicy` holds the outer reconnect in its backoff while the test faults the lease, recovers it through the manager and takes a connection snapshot; it then releases the gate, waits for `Connected`, and asserts no extra transport connection and a working session. No wall-clock dependency. Without the skip it fails with one extra connection. - New `ExplicitManagedReconnectRunsAfterChannelManagerReactivationAsync`: an explicit `ReconnectAsync` joins the gated outer reconnect, a later fault re-triggers it, and the manager recovers the channel. The explicit reconnect must still open a new connection. It failed (`Expected: greater than 9, But was: 9`) before explicit requests were tracked separately. - `BoundClientCannotRetryAcrossFaultedManagedEntryReplacementAsync`: in the managed variant, wait for the outer recovery to start before the manual reconnect, and await `StateMachine.WaitForConnectedAsync` before asserting `Ready` and taking the connection-count snapshot. - `WithSessionAsync` takes an optional `IReconnectPolicy`; new helpers `FaultLeaseAsync` and `GatedReconnectPolicy`. ## Verification - Injected `await Task.Delay(200)` after the reconnect cycle in `ClientChannelManager.ReconnectLeaseAsync` (not committed). The old test then fails every time with the exact CI assertion, and the updated test passes. - `CustomTestTarget=net10.0 dotnet test -c Release -f net10.0 --filter FullyQualifiedName~SessionBindingLiveTests`: 5 sequential runs, 94/94 passed each time. These runs had the test change only; the `ManagedSession` skip was not yet in place. - With the skip: the new test and the original test pass; with the skip disabled, the new test fails as described above. - `--filter "FullyQualifiedName~ManagedSession|FullyQualifiedName~SessionBindingLive|FullyQualifiedName~Reconnect"` on `Opc.Ua.Client.Tests` (net10.0, Release) with the final code: 514 passed, 2 skipped (the managed-only tests in the plain-`Session` fixture). Full `Opc.Ua.Client.Tests` / `Opc.Ua.Sessions.Tests` runs and the net48 build are left to CI. 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 --- src/Opc.Ua.Client/Session/ManagedSession.cs | 108 ++++++++++- .../SessionBindingLiveTests.Transport.cs | 183 ++++++++++++++++++ .../Session/SessionBindingLiveTests.cs | 5 +- 3 files changed, 291 insertions(+), 5 deletions(-) diff --git a/src/Opc.Ua.Client/Session/ManagedSession.cs b/src/Opc.Ua.Client/Session/ManagedSession.cs index 0904e14b23..b816f07b6b 100644 --- a/src/Opc.Ua.Client/Session/ManagedSession.cs +++ b/src/Opc.Ua.Client/Session/ManagedSession.cs @@ -875,8 +875,13 @@ namespace Opc.Ua.Client ConnectionStateBudgetOperation? operation = connection == null && channel == null ? null : (_, token) => HandleManualReconnectAsync(connection, channel, ct, token); + + // 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."); } @@ -1611,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(); @@ -1633,6 +1639,7 @@ namespace Opc.Ua.Client } } await RefreshRedundancyInfoBestEffortAsync(linked.Token).ConfigureAwait(false); + CompleteReconnect(explicitRequests); return ServiceResult.Good; } catch (Exception exception) @@ -1657,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(); @@ -1665,6 +1674,20 @@ namespace Opc.Ua.Client using (await RecoveryWriterLockAsync(ct) .ConfigureAwait(false)) { + // 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. 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(); + m_logger.ManagedSessionReconnectSkippedChannelRecovered(); + return ServiceResult.Good; + } + try { ITransportWaitingConnection? connection = await WaitForRecoveryConnectionAsync( @@ -1741,6 +1764,7 @@ namespace Opc.Ua.Client await RefreshRedundancyInfoBestEffortAsync(ct).ConfigureAwait(false); + CompleteReconnect(explicitRequests); m_reconnectPolicy.Reset(); m_logger.ManagedSessionReconnected(); @@ -2343,7 +2367,7 @@ namespace Opc.Ua.Client } else { - StateMachine.TriggerReconnect(); + TriggerRecoveryReconnect(); } } @@ -2376,6 +2400,10 @@ namespace Opc.Ua.Client } break; case ChannelState.Ready: + if (change.PreviousState == ChannelState.TransportConnectedSessionReactivating) + { + Interlocked.Increment(ref m_channelSessionReactivations); + } Interlocked.Exchange(ref m_channelReconnectInProgress, 0); if (m_session?.ChannelRecoveryInProgress != true) { @@ -2388,7 +2416,7 @@ namespace Opc.Ua.Client // Channel-mgr gave up. Surface to outer state // machine so the IReconnectPolicy / failover path // can run. - StateMachine.TriggerReconnect(change); + TriggerRecoveryReconnect(change); break; case ChannelState.Closed: Interlocked.Exchange(ref m_channelReconnectInProgress, 0); @@ -2396,11 +2424,57 @@ namespace Opc.Ua.Client // Channel-mgr closed. Surface to outer state // machine so the IReconnectPolicy / failover path // can run. - StateMachine.TriggerReconnect(); + TriggerRecoveryReconnect(); break; } } + /// + /// Requests an outer reconnect for a channel fault or keep-alive failure + /// and remembers how many channel reactivations had happened by then, so + /// the reconnect can be skipped if the channel manager reactivates the + /// session again before it runs. + /// + private void TriggerRecoveryReconnect(ChannelStateChange? change = null) + { + // Recorded before the request so the worker never reads an older mark. + Interlocked.Exchange( + ref m_reconnectTriggerMark, + Interlocked.Read(ref m_channelSessionReactivations)); + if (!StateMachine.RequestReconnect(underlyingChannelState: change)) + { + Interlocked.Exchange(ref m_reconnectTriggerMark, NoReconnectTrigger); + } + } + + /// + /// Records a completed reconnect: the pending trigger is consumed and every + /// explicit request made before the reconnect started is served. + /// + 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); + } + } + + /// + /// Whether the channel manager reactivated the session after the pending + /// outer reconnect was requested and the channel is still ready, in which + /// case another reconnect would only tear down the recovered channel. + /// + private bool ChannelReactivatedSinceReconnectTrigger(Session session) + { + long mark = Interlocked.Read(ref m_reconnectTriggerMark); + return mark != NoReconnectTrigger && + Interlocked.Read(ref m_channelSessionReactivations) > mark && + !session.ChannelRecoveryInProgress && + session.ManagedChannel?.State == ChannelState.Ready; + } + private void RaiseChannelStateChanged(ChannelStateChange change) { try @@ -2933,6 +3007,30 @@ namespace Opc.Ua.Client /// suppressed by a stuck reconnect reports how long it has been suppressed. /// private long m_channelReconnectStartedAt; + + /// + /// Counts channel-manager reconnect cycles that reactivated this session + /// (a Ready transition out of + /// ). + /// + private long m_channelSessionReactivations; + + /// + /// when a channel fault or a + /// keep-alive failure last requested an outer reconnect, or + /// when none is recorded. + /// + private long m_reconnectTriggerMark = NoReconnectTrigger; + private const long NoReconnectTrigger = -1; + + /// + /// Explicit + /// 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. + /// + private long m_explicitReconnectRequests; + private long m_explicitReconnectsServed; private ServerRedundancyInfo? m_redundancyInfo; private Task? m_redundancyRefreshTask; private Task? m_redundancyEndpointRefreshTask; @@ -3069,5 +3167,9 @@ namespace Opc.Ua.Client [LoggerMessage(EventId = ClientEventIds.ManagedSession + 29, Level = LogLevel.Error, Message = "ManagedSession: Asynchronous teardown initiated by Dispose failed.")] public static partial void ManagedSessionDisposalFailed(this ILogger logger, Exception? exception); + + [LoggerMessage(EventId = ClientEventIds.ManagedSession + 30, Level = LogLevel.Information, + Message = "ManagedSession: Reconnect skipped; the channel manager already reactivated the session.")] + public static partial void ManagedSessionReconnectSkippedChannelRecovered(this ILogger logger); } } diff --git a/tests/Opc.Ua.Client.Tests/Session/SessionBindingLiveTests.Transport.cs b/tests/Opc.Ua.Client.Tests/Session/SessionBindingLiveTests.Transport.cs index 47511124d2..3eb3bfd14b 100644 --- a/tests/Opc.Ua.Client.Tests/Session/SessionBindingLiveTests.Transport.cs +++ b/tests/Opc.Ua.Client.Tests/Session/SessionBindingLiveTests.Transport.cs @@ -169,8 +169,22 @@ namespace Opc.Ua.Client.Tests .ConfigureAwait(false), Throws.TypeOf()).ConfigureAwait(false); Assert.That(lease.State, Is.EqualTo(ChannelState.Faulted)); + + // Once the channel manager gives up, a managed session's state machine + // takes over and reconnects after its own backoff. Let that recovery + // start before the manual one and settle after it, so it can neither + // overlap the state assertion nor count as a retry of the bound request. + var managed = session as Client.ManagedSession; + while (managed != null && !managed.Reconnecting) + { + await Task.Delay(10, timeout.Token).ConfigureAwait(false); + } channels.FailConnect = false; await lease.Manager.ReconnectAsync(lease, timeout.Token).ConfigureAwait(false); + if (managed != null) + { + await managed.StateMachine.WaitForConnectedAsync(timeout.Token).ConfigureAwait(false); + } Assert.That(lease.State, Is.EqualTo(ChannelState.Ready)); int connections = channels.Connections; await Assert.ThatAsync(() => binding.ReadValueAsync( @@ -190,6 +204,175 @@ namespace Opc.Ua.Client.Tests }, timeout.Token, channels).ConfigureAwait(false); } + /// + /// Verifies that a managed session does not reconnect a channel again once the + /// channel manager has reactivated the session while the outer reconnect waited. + /// + [Test] + public async Task ManagedSessionSkipsOuterReconnectAfterChannelManagerReactivationAsync() + { + 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); + + // 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), + "The outer reconnect must not reconnect a channel the manager already recovered."); + 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); + } + + /// + /// 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. + /// + [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); + } + + /// + /// 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. + /// + 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()).ConfigureAwait(false); + Assert.That(lease.State, Is.EqualTo(ChannelState.Faulted)); + } + finally + { + channels.FailConnect = false; + } + } + + /// + /// 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. + /// + 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 m_entered = + new(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly ManualResetEventSlim m_gate = new(); + } + private sealed class BarrierChannelBindings : ITransportChannelBindings, IUaSCByteTransportFactory { /// diff --git a/tests/Opc.Ua.Client.Tests/Session/SessionBindingLiveTests.cs b/tests/Opc.Ua.Client.Tests/Session/SessionBindingLiveTests.cs index f5e39b3dfa..47b8f7f6da 100644 --- a/tests/Opc.Ua.Client.Tests/Session/SessionBindingLiveTests.cs +++ b/tests/Opc.Ua.Client.Tests/Session/SessionBindingLiveTests.cs @@ -340,7 +340,8 @@ namespace Opc.Ua.Client.Tests private async Task WithSessionAsync( Func action, CancellationToken ct, - ITransportChannelBindings? bindings = null) + ITransportChannelBindings? bindings = null, + IReconnectPolicy? reconnectPolicy = null) { ITelemetryContext telemetry = NUnitTelemetryContext.Create(); string root = Path.Combine(Path.GetTempPath(), "wsb-" + Guid.NewGuid().ToString("N")[..8]); @@ -375,7 +376,7 @@ namespace Opc.Ua.Client.Tests { await using Client.ManagedSession managed = await Client.ManagedSession.CreateAsync( connection.Config, connection.Endpoint, connection.SessionFactory, - telemetry: telemetry, ct: ct).ConfigureAwait(false); + reconnectPolicy: reconnectPolicy, telemetry: telemetry, ct: ct).ConfigureAwait(false); await action(managed).ConfigureAwait(false); } else