fix(client): a failover abandons itself once the user disconnects or reconnects; paging and teardown minors

Review I1: ConnectAsync, DisconnectAsync and Dispose each start a new connection epoch under _subscriptionLock. A keep-alive failover runs under the epoch it started in and abandons itself the moment it moves: it installs its session only if the epoch still matches (checked and assigned under the lock, otherwise the session is disposed), its state transitions are epoch-guarded, it stops trying endpoints, and it never flips a user's new connection to Disconnected. A late keep-alive from a previous connection's session starts no failover. Gate-driven tests cover the success and all-fail arms, the stale keep-alive, with a genuine failover as the positive control (4 of them red before). Minors: m8 only the CompareExchange winner sets LastFailover; m5 the endpoint loop stops early, and a failed Unsubscribe still drops its bookkeeping; m1 a read cancelled mid-paging releases its cursor (bounded token) and rethrows, and the cap release no longer rides the caller's token; m2 results are trimmed to the cap; m3 a non-positive cap is refused (CLI and UI already bound it to >= 1); m6 the subscription adapter's Dispose uses one bounded bridge for delete + detach. Refs #747 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
2026-10-22 17:18:26 -04:00
parent 31605e547d
commit 0c9e4f4e3d
7 changed files with 442 additions and 74 deletions
@@ -145,17 +145,16 @@ internal sealed class DefaultSubscriptionAdapter : ISubscriptionAdapter
    /// </summary>
    public void Dispose()
    {
+        // One bounded bridge (5 s) for delete + detach, so a dispose never blocks longer than before (review m6).
+        // DetachAsync swallows its own failures; a failed delete must not skip the detach.
        try
        {
-            DisposeBridge.Run(ct => _subscription.DeleteAsync(true, ct));
-        }
-        catch
-        {
-        }
-
-        try
-        {
-            DisposeBridge.Run(DetachAsync);
+            DisposeBridge.Run(async ct =>
+            {
+                try { await _subscription.DeleteAsync(true, ct); }
+                catch (Exception ex) { Logger.Warning(ex, "Error deleting subscription"); }
+                await DetachAsync(ct);
+            });
        }
        catch
        {
@@ -21,46 +21,76 @@ internal static class HistoryReadPaging
    public delegate Task<HistoryReadResult?> PageReader(
        byte[]? continuationPoint, bool releaseContinuationPoints, CancellationToken ct);

+    /// <summary>The bound on a best-effort release that does not ride the caller's (possibly cancelled) token.</summary>
+    internal static readonly TimeSpan ReleaseTimeout = TimeSpan.FromSeconds(5);
+
    /// <summary>Reads pages until the server finishes or the caller's cap is reached.</summary>
    /// <typeparam name="T">The element type extracted from each page.</typeparam>
    /// <param name="readPage">Issues one HistoryRead for the single node.</param>
    /// <param name="extract">Extracts the page's elements from a result.</param>
-    /// <param name="maxValues">The caller's cap; <c>&lt;= 0</c> means no cap (follow every continuation point).</param>
+    /// <param name="maxValues">The caller's cap; must be positive. The result is trimmed to it.</param>
    /// <param name="nodeId">The node being read, for refusal messages.</param>
    /// <param name="ct">The caller's cancellation token.</param>
-    /// <returns>Every element read.</returns>
+    /// <returns>At most <paramref name="maxValues"/> elements.</returns>
+    /// <exception cref="ArgumentOutOfRangeException"><paramref name="maxValues"/> is not positive: an uncapped client
+    /// read would materialise a tag's whole window (the client side of review 08 1.1).</exception>
    /// <exception cref="ServiceResultException">A page was refused (see <see cref="HistoryReadStatus"/>).</exception>
+    /// <exception cref="OperationCanceledException"><paramref name="ct"/> was cancelled mid-read; the outstanding
+    /// cursor is released (best effort) first.</exception>
    public static async Task<List<T>> ReadAllAsync<T>(
        PageReader readPage, Func<HistoryReadResult, IEnumerable<T>> extract, int maxValues, NodeId nodeId,
        CancellationToken ct)
    {
+        ArgumentOutOfRangeException.ThrowIfNegativeOrZero(maxValues);
+
        var all = new List<T>();
        byte[]? continuationPoint = null;

-        while (true)
-        {
-            var result = await readPage(continuationPoint, false, ct).ConfigureAwait(false);
-            if (result == null) return all;
-
-            HistoryReadStatus.ThrowIfRefused(result.StatusCode, nodeId);
-            all.AddRange(extract(result));
-
-            continuationPoint = result.ContinuationPoint;
-            if (continuationPoint is not { Length: > 0 }) return all;
-            if (maxValues > 0 && all.Count >= maxValues) break;
-        }
-
-        // Stopped by the client's own cap: the server still holds a cursor for this session. Release it; failing to
-        // is not worth failing a read whose values are already in hand.
        try
        {
-            await readPage(continuationPoint, true, ct).ConfigureAwait(false);
+            while (true)
+            {
+                var result = await readPage(continuationPoint, false, ct).ConfigureAwait(false);
+                if (result == null) return Trim(all, maxValues);
+
+                HistoryReadStatus.ThrowIfRefused(result.StatusCode, nodeId);
+                all.AddRange(extract(result));
+
+                continuationPoint = result.ContinuationPoint;
+                if (continuationPoint is not { Length: > 0 }) return Trim(all, maxValues);
+                if (all.Count >= maxValues) break;
+            }
        }
-        catch (Exception ex) when (ex is not OperationCanceledException)
+        catch (OperationCanceledException)
+        {
+            // Cancelled between pages (or during page N+1) with a cursor outstanding: release it rather than leave it
+            // occupying one of the session's continuation-point slots, then surface the cancellation.
+            if (continuationPoint is { Length: > 0 }) await ReleaseAsync(readPage, continuationPoint, nodeId);
+            throw;
+        }
+
+        // Stopped by the client's own cap: the server still holds a cursor for this session. Release it — on its own
+        // bounded token, so a cancellation racing the release cannot lose the values already in hand.
+        await ReleaseAsync(readPage, continuationPoint, nodeId);
+        return Trim(all, maxValues);
+    }
+
+    private static List<T> Trim<T>(List<T> all, int maxValues)
+    {
+        if (all.Count > maxValues) all.RemoveRange(maxValues, all.Count - maxValues);
+        return all;
+    }
+
+    private static async Task ReleaseAsync(PageReader readPage, byte[] continuationPoint, NodeId nodeId)
+    {
+        try
+        {
+            using var cts = new CancellationTokenSource(ReleaseTimeout);
+            await readPage(continuationPoint, true, cts.Token).ConfigureAwait(false);
+        }
+        catch (Exception ex)
        {
            Logger.Warning(ex, "Releasing the HistoryRead continuation point for {NodeId} failed", nodeId);
        }
-
-        return all;
    }
 }
@@ -35,6 +35,12 @@ public sealed class OpcUaClientService : IOpcUaClientService
    // run at a time. 0 = idle, 1 = failover in progress (Interlocked-managed).
    private int _failoverInProgress;

+    // The connection epoch (review I1): ConnectAsync and DisconnectAsync each start a new one (under
+    // _subscriptionLock). A failover runs under the epoch it started in and abandons itself — disposing any session it
+    // opened, making no state transition — the moment the epoch moves, so a user Disconnect + Connect during a
+    // failover is never undone by it. A keep-alive from a previous connection's session cannot start one either.
+    private long _connectionEpoch;
+
    /// <summary>The most recent keep-alive-triggered failover task (test seam: lets a test await it).</summary>
    internal Task? LastFailover { get; private set; }

@@ -109,6 +115,9 @@ public sealed class OpcUaClientService : IOpcUaClientService
            throw new InvalidOperationException(
                $"Already {_state.ToString().ToLowerInvariant()}; call DisconnectAsync before connecting again.");

+        long epoch;
+        lock (_subscriptionLock) epoch = ++_connectionEpoch;
+
        _settings = settings;
        _allEndpointUrls = FailoverUrlParser.Parse(settings.EndpointUrl, settings.FailoverUrls);
        _currentEndpointIndex = 0;
@@ -118,11 +127,11 @@ public sealed class OpcUaClientService : IOpcUaClientService
        try
        {
            var session = await ConnectToEndpointAsync(settings, _allEndpointUrls[0], ct);
-            _session = session;
+            lock (_subscriptionLock) _session = session;

            session.RegisterKeepAliveHandler(isGood =>
            {
-                if (!isGood) LastFailover = HandleKeepAliveFailureAsync();
+                if (!isGood) OnKeepAliveFailed(epoch);
            });

            CurrentConnectionInfo = BuildConnectionInfo(session);
@@ -143,6 +152,9 @@ public sealed class OpcUaClientService : IOpcUaClientService
        if (_state == ConnectionState.Disconnected)
            return;

+        // A new epoch first: a failover still connecting abandons itself instead of resurrecting the connection.
+        lock (_subscriptionLock) _connectionEpoch++;
+
        var endpointUrl = _session?.EndpointUrl ?? _settings?.EndpointUrl ?? string.Empty;

        // Snapshot the subscription adapter references under the lock so a concurrent
@@ -314,11 +326,18 @@ public sealed class OpcUaClientService : IOpcUaClientService
                subscription = _dataSubscription;
            }

-            if (subscription != null) await subscription.RemoveMonitoredItemAsync(sub.Handle, ct);
-
-            lock (_subscriptionLock)
+            try
            {
-                _activeDataSubscriptions.Remove(nodeIdStr);
+                if (subscription != null) await subscription.RemoveMonitoredItemAsync(sub.Handle, ct);
+            }
+            finally
+            {
+                // The caller asked to stop: drop the bookkeeping even when the server-side remove failed (a session
+                // that died under us), or a later failover would replay the item onto the new session (review m5).
+                lock (_subscriptionLock)
+                {
+                    _activeDataSubscriptions.Remove(nodeIdStr);
+                }
            }
        }
        finally
@@ -698,8 +717,12 @@ public sealed class OpcUaClientService : IOpcUaClientService
    /// </summary>
    public void Dispose()
    {
-        if (_disposed) return;
-        _disposed = true;
+        lock (_subscriptionLock)
+        {
+            if (_disposed) return;
+            _disposed = true;
+            _connectionEpoch++;   // a failover still running abandons itself
+        }

        _dataSubscription?.Dispose();
        _alarmSubscription?.Dispose();
@@ -745,19 +768,27 @@ public sealed class OpcUaClientService : IOpcUaClientService
            ct);
    }

-    private async Task HandleKeepAliveFailureAsync()
+    /// <summary>A bad keep-alive from the session of connection <paramref name="epoch"/>.</summary>
+    private void OnKeepAliveFailed(long epoch)
    {
-        // Serialize failover: the OPC UA stack raises KeepAlive repeatedly
-        // while a session is down, so multiple bad keep-alives can fire before
-        // the first failover loop finishes. CompareExchange atomically claims
-        // the failover slot; a re-entrant call sees 1 and returns immediately,
-        // guaranteeing exactly one failover loop runs at a time.
+        // A late keep-alive from a previous connection's session is not about the current one.
+        if (!IsCurrentEpoch(epoch)) return;
+
+        // Serialize failover: the OPC UA stack raises KeepAlive repeatedly while a session is down, so several bad
+        // keep-alives can fire before the first failover loop finishes. CompareExchange claims the failover slot; a
+        // re-entrant call sees 1 and returns, so exactly one loop runs — and only the winner sets the test seam
+        // (review m8: a re-entrant call used to overwrite it with an already-completed no-op).
        if (Interlocked.CompareExchange(ref _failoverInProgress, 1, 0) != 0)
            return;

+        LastFailover = HandleKeepAliveFailureAsync(epoch);
+    }
+
+    private async Task HandleKeepAliveFailureAsync(long epoch)
+    {
        try
        {
-            await RunFailoverAsync();
+            await RunFailoverAsync(epoch);
        }
        finally
        {
@@ -765,14 +796,39 @@ public sealed class OpcUaClientService : IOpcUaClientService
        }
    }

-    private async Task RunFailoverAsync()
+    private bool IsCurrentEpoch(long epoch)
+    {
+        lock (_subscriptionLock) return _connectionEpoch == epoch && !_disposed;
+    }
+
+    /// <summary>
+    ///     A state transition made only while <paramref name="epoch"/> is still current (checked and applied under
+    ///     <see cref="_subscriptionLock"/>, the event raised outside it).
+    /// </summary>
+    /// <returns><see langword="false"/> when the epoch has moved on (the caller must abandon).</returns>
+    private bool TransitionStateIfCurrent(long epoch, ConnectionState newState, string endpointUrl)
+    {
+        ConnectionState oldState;
+        lock (_subscriptionLock)
+        {
+            if (_connectionEpoch != epoch || _disposed) return false;
+            oldState = _state;
+            if (oldState == newState) return true;
+            _state = newState;
+        }
+
+        ConnectionStateChanged?.Invoke(this, new ConnectionStateChangedEventArgs(oldState, newState, endpointUrl));
+        return true;
+    }
+
+    private async Task RunFailoverAsync(long epoch)
    {
        if (_state == ConnectionState.Reconnecting || _state == ConnectionState.Disconnected)
            return;

        var oldSession = _session;
        var oldEndpoint = oldSession?.EndpointUrl ?? string.Empty;
-        TransitionState(ConnectionState.Reconnecting, oldEndpoint);
+        if (!TransitionStateIfCurrent(epoch, ConnectionState.Reconnecting, oldEndpoint)) return;
        Logger.Warning("Session lost on {EndpointUrl}. Attempting failover...", oldEndpoint);

        // Dispose the dead session BEFORE taking the gate: an in-flight subscribe holding the gate is blocked on
@@ -793,47 +849,60 @@ public sealed class OpcUaClientService : IOpcUaClientService
        await _subscriptionGate.WaitAsync();
        try
        {
+            // The user's settings / endpoint list as of THIS epoch: a Connect during the failover replaces the fields.
+            ConnectionSettings? settings;
+            string[]? endpoints;
            lock (_subscriptionLock)
            {
+                if (_connectionEpoch != epoch || _disposed) return;   // the user disconnected / reconnected meanwhile
                if (ReferenceEquals(_session, oldSession)) _session = null;
                _dataSubscription = null;
                _alarmSubscription = null;
+                settings = _settings;
+                endpoints = _allEndpointUrls;
            }

-            if (_settings == null || _allEndpointUrls == null)
+            if (settings == null || endpoints == null)
            {
-                TransitionState(ConnectionState.Disconnected, oldEndpoint);
+                TransitionStateIfCurrent(epoch, ConnectionState.Disconnected, oldEndpoint);
                return;
            }

            // Try each endpoint
-            for (var attempt = 0; attempt < _allEndpointUrls.Length; attempt++)
+            for (var attempt = 0; attempt < endpoints.Length; attempt++)
            {
-                _currentEndpointIndex = (_currentEndpointIndex + 1) % _allEndpointUrls.Length;
-                var url = _allEndpointUrls[_currentEndpointIndex];
+                if (!IsCurrentEpoch(epoch)) return;               // stop trying once the user has moved on (m5)
+                _currentEndpointIndex = (_currentEndpointIndex + 1) % endpoints.Length;
+                var url = endpoints[_currentEndpointIndex];

                try
                {
                    Logger.Information("Failover attempt to {EndpointUrl}", url);
-                    var session = await ConnectToEndpointAsync(_settings, url, CancellationToken.None);
+                    var session = await ConnectToEndpointAsync(settings, url, CancellationToken.None);

-                    // A DisconnectAsync / Dispose that ran while this attempt was connecting wins: do not resurrect
-                    // the connection behind the caller's back.
-                    if (_disposed || _state == ConnectionState.Disconnected)
+                    // A Disconnect / Connect / Dispose that ran while this attempt was connecting wins: never
+                    // resurrect the old connection or overwrite the user's new session. Checked and installed
+                    // atomically against the epoch.
+                    bool installed;
+                    lock (_subscriptionLock)
+                    {
+                        installed = _connectionEpoch == epoch && !_disposed;
+                        if (installed) _session = session;
+                    }
+
+                    if (!installed)
                    {
                        session.Dispose();
                        return;
                    }

-                    _session = session;
-
                    session.RegisterKeepAliveHandler(isGood =>
                    {
-                        if (!isGood) LastFailover = HandleKeepAliveFailureAsync();
+                        if (!isGood) OnKeepAliveFailed(epoch);
                    });

                    CurrentConnectionInfo = BuildConnectionInfo(session);
-                    TransitionState(ConnectionState.Connected, url);
+                    if (!TransitionStateIfCurrent(epoch, ConnectionState.Connected, url)) return;
                    Logger.Information("Failover succeeded to {EndpointUrl}", url);

                    // Replay subscriptions
@@ -846,8 +915,8 @@ public sealed class OpcUaClientService : IOpcUaClientService
                }
            }

-            Logger.Error("All failover endpoints unreachable");
-            TransitionState(ConnectionState.Disconnected, oldEndpoint);
+            if (TransitionStateIfCurrent(epoch, ConnectionState.Disconnected, oldEndpoint))
+                Logger.Error("All failover endpoints unreachable");
        }
        finally
        {
@@ -1023,9 +1092,13 @@ public sealed class OpcUaClientService : IOpcUaClientService

    private void TransitionState(ConnectionState newState, string endpointUrl)
    {
-        var oldState = _state;
-        if (oldState == newState) return;
-        _state = newState;
+        ConnectionState oldState;
+        lock (_subscriptionLock)
+        {
+            oldState = _state;
+            if (oldState == newState) return;
+            _state = newState;
+        }
        ConnectionStateChanged?.Invoke(this, new ConnectionStateChangedEventArgs(oldState, newState, endpointUrl));
    }