ManagedSession: skip redundant outer reconnect after channel-manager recovery (fixes BoundClientCannotRetryAcrossFaultedManagedEntryReplacementAsync flake) (#4781)

## 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 <noreply@anthropic.com>
2026-10-16 21:02:51 +08:00 · 2026-10-14 10:02:52 +02:00
parent 27c17fb410
commit b36cacbeac
3 changed files with 291 additions and 5 deletions
@@ -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;
            }
        }

+        /// <summary>
+        /// 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.
+        /// </summary>
+        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);
+            }
+        }
+
+        /// <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
+        /// case another reconnect would only tear down the recovered channel.
+        /// </summary>
+        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.
        /// </summary>
        private long m_channelReconnectStartedAt;
+
+        /// <summary>
+        /// Counts channel-manager reconnect cycles that reactivated this session
+        /// (a Ready transition out of
+        /// <see cref="ChannelState.TransportConnectedSessionReactivating"/>).
+        /// </summary>
+        private long m_channelSessionReactivations;
+
+        /// <summary>
+        /// <see cref="m_channelSessionReactivations"/> when a channel fault or a
+        /// keep-alive failure last requested an outer reconnect, or
+        /// <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;
@@ -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);
    }
 }
@@ -169,8 +169,22 @@ namespace Opc.Ua.Client.Tests
                            .ConfigureAwait(false),
                        Throws.TypeOf<ServiceResultException>()).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);
        }

+        /// <summary>
+        /// Verifies that a managed session does not reconnect a channel again once the
+        /// channel manager has reactivated the session while the outer reconnect waited.
+        /// </summary>
+        [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);
+        }
+
+        /// <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>
@@ -340,7 +340,8 @@ namespace Opc.Ua.Client.Tests
        private async Task WithSessionAsync(
            Func<ISession, Task> 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