fix(client): failover runs under the subscription gate; ConnectAsync refuses when connected

R08 4.5: every subscription operation that awaits I/O and the failover's teardown + reconnect + replay share one async gate, and every read/write of the subscription fields (failover's null-outs included) is under _subscriptionLock. A subscribe in flight when a failover starts is recorded first and then replayed onto the new session instead of landing on the dying session's subscription. The dead session is disposed before the gate is taken so a blocked call fails fast. A DisconnectAsync during a failover is no longer undone by it. ConnectAsync on a connected or reconnecting client throws instead of overwriting (and leaking) the live session. Refs #747 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
2026-10-22 16:30:35 -04:00
parent 540b1632c5
commit 2d8bb037f5
3 changed files with 292 additions and 96 deletions
@@ -15,16 +15,17 @@ public sealed class OpcUaClientService : IOpcUaClientService
 {
    private static readonly ILogger Logger = Log.ForContext<OpcUaClientService>();

-    // Guards all access to the subscription-bookkeeping state below
-    // (_activeDataSubscriptions and _activeAlarmSubscription). The dictionary
-    // and tuple are mutated from the caller thread, the keep-alive failover
-    // path, and DisconnectAsync, so every read/write must be inside this lock.
+    // Guards all access to the subscription state below (_activeDataSubscriptions, _activeAlarmSubscription,
+    // _dataSubscription, _alarmSubscription). They are touched from the caller thread, the keep-alive failover path
+    // and DisconnectAsync, so every read/write is inside this lock — including failover's null-outs.
    private readonly object _subscriptionLock = new();

-    // Serialises SubscribeAlarmsAsync / UnsubscribeAlarmsAsync so concurrent callers
-    // cannot both pass the _alarmSubscription == null check and create duplicate event
-    // subscriptions. Capacity 1 = at most one waiter; async-friendly (no thread-block).
-    private readonly SemaphoreSlim _alarmSubscribeSemaphore = new(1, 1);
+    // Serialises every subscription operation that awaits I/O (Subscribe/Unsubscribe, SubscribeAlarms/
+    // UnsubscribeAlarms) AND the failover's teardown + reconnect + replay. A subscribe is therefore either
+    // wholly before a failover (recorded, then replayed onto the new session) or wholly after it (on the new
+    // session) — never half on the dying session. Async-friendly (no thread block). Deliberately never disposed:
+    // async-only waits hold no kernel handle, and a failover still running at Dispose releases it afterwards.
+    private readonly SemaphoreSlim _subscriptionGate = new(1, 1);

    // Track active data subscriptions for replay after failover
    private readonly Dictionary<string, (NodeId NodeId, int IntervalMs, uint Handle)> _activeDataSubscriptions = new();
@@ -34,6 +35,9 @@ public sealed class OpcUaClientService : IOpcUaClientService
    // run at a time. 0 = idle, 1 = failover in progress (Interlocked-managed).
    private int _failoverInProgress;

+    /// <summary>The most recent keep-alive-triggered failover task (test seam: lets a test await it).</summary>
+    internal Task? LastFailover { get; private set; }
+
    private readonly IApplicationConfigurationFactory _configFactory;
    private readonly IEndpointDiscovery _endpointDiscovery;

@@ -99,6 +103,12 @@ public sealed class OpcUaClientService : IOpcUaClientService
        ThrowIfDisposed();
        settings.Validate();

+        // Overwriting a live (or reconnecting) session would leak it and orphan its subscriptions and keep-alive
+        // handler. The caller must disconnect first.
+        if (_state != ConnectionState.Disconnected || _session != null)
+            throw new InvalidOperationException(
+                $"Already {_state.ToString().ToLowerInvariant()}; call DisconnectAsync before connecting again.");
+
        _settings = settings;
        _allEndpointUrls = FailoverUrlParser.Parse(settings.EndpointUrl, settings.FailoverUrls);
        _currentEndpointIndex = 0;
@@ -112,7 +122,7 @@ public sealed class OpcUaClientService : IOpcUaClientService

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

            CurrentConnectionInfo = BuildConnectionInfo(session);
@@ -248,20 +258,39 @@ public sealed class OpcUaClientService : IOpcUaClientService
        ThrowIfNotConnected();

        var nodeIdStr = nodeId.ToString();
-        lock (_subscriptionLock)
+        await _subscriptionGate.WaitAsync(ct);
+        try
        {
-            if (_activeDataSubscriptions.ContainsKey(nodeIdStr))
-                return; // Already subscribed
+            // A failover may have run (and failed) while this call waited for the gate.
+            ThrowIfNotConnected();
+
+            ISubscriptionAdapter? subscription;
+            ISessionAdapter session;
+            lock (_subscriptionLock)
+            {
+                if (_activeDataSubscriptions.ContainsKey(nodeIdStr))
+                    return; // Already subscribed
+                subscription = _dataSubscription;
+                session = _session!;
+            }
+
+            if (subscription == null)
+            {
+                subscription = await session.CreateSubscriptionAsync(intervalMs, ct);
+                lock (_subscriptionLock) _dataSubscription = subscription;
+            }
+
+            var handle = await subscription.AddDataChangeMonitoredItemAsync(
+                nodeId, intervalMs, OnDataChangeNotification, ct);
+
+            lock (_subscriptionLock)
+            {
+                _activeDataSubscriptions[nodeIdStr] = (nodeId, intervalMs, handle);
+            }
        }
-
-        if (_dataSubscription == null) _dataSubscription = await _session!.CreateSubscriptionAsync(intervalMs, ct);
-
-        var handle = await _dataSubscription.AddDataChangeMonitoredItemAsync(
-            nodeId, intervalMs, OnDataChangeNotification, ct);
-
-        lock (_subscriptionLock)
+        finally
        {
-            _activeDataSubscriptions[nodeIdStr] = (nodeId, intervalMs, handle);
+            _subscriptionGate.Release();
        }

        Logger.Debug("Subscribed to data changes on {NodeId}", nodeId);
@@ -273,18 +302,28 @@ public sealed class OpcUaClientService : IOpcUaClientService
        ThrowIfDisposed();

        var nodeIdStr = nodeId.ToString();
-        (NodeId NodeId, int IntervalMs, uint Handle) sub;
-        lock (_subscriptionLock)
+        await _subscriptionGate.WaitAsync(ct);
+        try
        {
-            if (!_activeDataSubscriptions.TryGetValue(nodeIdStr, out sub))
-                return; // Not subscribed, safe to ignore
+            (NodeId NodeId, int IntervalMs, uint Handle) sub;
+            ISubscriptionAdapter? subscription;
+            lock (_subscriptionLock)
+            {
+                if (!_activeDataSubscriptions.TryGetValue(nodeIdStr, out sub))
+                    return; // Not subscribed, safe to ignore
+                subscription = _dataSubscription;
+            }
+
+            if (subscription != null) await subscription.RemoveMonitoredItemAsync(sub.Handle, ct);
+
+            lock (_subscriptionLock)
+            {
+                _activeDataSubscriptions.Remove(nodeIdStr);
+            }
        }
-
-        if (_dataSubscription != null) await _dataSubscription.RemoveMonitoredItemAsync(sub.Handle, ct);
-
-        lock (_subscriptionLock)
+        finally
        {
-            _activeDataSubscriptions.Remove(nodeIdStr);
+            _subscriptionGate.Release();
        }

        Logger.Debug("Unsubscribed from data changes on {NodeId}", nodeId);
@@ -300,17 +339,25 @@ public sealed class OpcUaClientService : IOpcUaClientService
        // Serialise check-and-create under the semaphore: concurrent
        // callers cannot both pass the null-check and create duplicate alarm subscriptions.
        // The semaphore is async-friendly; the subscription I/O runs inside the critical section.
-        await _alarmSubscribeSemaphore.WaitAsync(ct);
+        await _subscriptionGate.WaitAsync(ct);
        try
        {
-            if (_alarmSubscription != null)
-                return; // Already subscribed to alarms
+            ThrowIfNotConnected();
+
+            ISessionAdapter session;
+            lock (_subscriptionLock)
+            {
+                if (_alarmSubscription != null)
+                    return; // Already subscribed to alarms
+                session = _session!;
+            }

            var monitorNode = sourceNodeId ?? ObjectIds.Server;
-            _alarmSubscription = await _session!.CreateSubscriptionAsync(intervalMs, ct);
+            var subscription = await session.CreateSubscriptionAsync(intervalMs, ct);
+            lock (_subscriptionLock) _alarmSubscription = subscription;

            var filter = CreateAlarmEventFilter();
-            await _alarmSubscription.AddEventMonitoredItemAsync(
+            await subscription.AddEventMonitoredItemAsync(
                monitorNode, intervalMs, filter, OnAlarmEventNotification, ct);

            lock (_subscriptionLock)
@@ -322,7 +369,7 @@ public sealed class OpcUaClientService : IOpcUaClientService
        }
        finally
        {
-            _alarmSubscribeSemaphore.Release();
+            _subscriptionGate.Release();
        }
    }

@@ -331,15 +378,27 @@ public sealed class OpcUaClientService : IOpcUaClientService
    {
        ThrowIfDisposed();

-        if (_alarmSubscription == null)
-            return;
-
-        await _alarmSubscription.DeleteAsync(ct);
-        _alarmSubscription = null;
-
-        lock (_subscriptionLock)
+        await _subscriptionGate.WaitAsync(ct);
+        try
        {
-            _activeAlarmSubscription = null;
+            ISubscriptionAdapter? subscription;
+            lock (_subscriptionLock)
+            {
+                subscription = _alarmSubscription;
+                if (subscription == null) return;
+            }
+
+            await subscription.DeleteAsync(ct);
+
+            lock (_subscriptionLock)
+            {
+                _alarmSubscription = null;
+                _activeAlarmSubscription = null;
+            }
+        }
+        finally
+        {
+            _subscriptionGate.Release();
        }

        Logger.Debug("Unsubscribed from alarm events");
@@ -351,10 +410,12 @@ public sealed class OpcUaClientService : IOpcUaClientService
        ThrowIfDisposed();
        ThrowIfNotConnected();

-        if (_alarmSubscription == null)
+        ISubscriptionAdapter? subscription;
+        lock (_subscriptionLock) subscription = _alarmSubscription;
+        if (subscription == null)
            throw new InvalidOperationException("No alarm subscription is active.");

-        await _alarmSubscription.ConditionRefreshAsync(ct);
+        await subscription.ConditionRefreshAsync(ct);
        Logger.Debug("Condition refresh requested");
    }

@@ -643,7 +704,6 @@ public sealed class OpcUaClientService : IOpcUaClientService
        _dataSubscription?.Dispose();
        _alarmSubscription?.Dispose();
        _session?.Dispose();
-        _alarmSubscribeSemaphore.Dispose();

        lock (_subscriptionLock)
        {
@@ -710,69 +770,93 @@ public sealed class OpcUaClientService : IOpcUaClientService
        if (_state == ConnectionState.Reconnecting || _state == ConnectionState.Disconnected)
            return;

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

-        // Close old session
-        if (_session != null)
+        // Dispose the dead session BEFORE taking the gate: an in-flight subscribe holding the gate is blocked on
+        // that session, and disposing it is what makes the call fail fast and release the gate.
+        if (oldSession != null)
        {
            try
            {
-                _session.Dispose();
+                oldSession.Dispose();
            }
            catch
            {
            }
-
-            _session = null;
        }

-        _dataSubscription = null;
-        _alarmSubscription = null;
-
-        if (_settings == null || _allEndpointUrls == null)
+        // The rest runs under the subscription gate (and its null-outs under _subscriptionLock): no subscribe can
+        // interleave with the teardown, the reconnect or the replay.
+        await _subscriptionGate.WaitAsync();
+        try
        {
-            TransitionState(ConnectionState.Disconnected, oldEndpoint);
-            return;
-        }
-
-        // Try each endpoint
-        for (var attempt = 0; attempt < _allEndpointUrls.Length; attempt++)
-        {
-            _currentEndpointIndex = (_currentEndpointIndex + 1) % _allEndpointUrls.Length;
-            var url = _allEndpointUrls[_currentEndpointIndex];
-
-            try
+            lock (_subscriptionLock)
            {
-                Logger.Information("Failover attempt to {EndpointUrl}", url);
-                var session = await ConnectToEndpointAsync(_settings, url, CancellationToken.None);
-                _session = session;
+                if (ReferenceEquals(_session, oldSession)) _session = null;
+                _dataSubscription = null;
+                _alarmSubscription = null;
+            }

-                session.RegisterKeepAliveHandler(isGood =>
-                {
-                    if (!isGood) _ = HandleKeepAliveFailureAsync();
-                });
-
-                CurrentConnectionInfo = BuildConnectionInfo(session);
-                TransitionState(ConnectionState.Connected, url);
-                Logger.Information("Failover succeeded to {EndpointUrl}", url);
-
-                // Replay subscriptions
-                await ReplaySubscriptionsAsync();
+            if (_settings == null || _allEndpointUrls == null)
+            {
+                TransitionState(ConnectionState.Disconnected, oldEndpoint);
                return;
            }
-            catch (Exception ex)
-            {
-                Logger.Warning(ex, "Failover to {EndpointUrl} failed", url);
-            }
-        }

-        Logger.Error("All failover endpoints unreachable");
-        TransitionState(ConnectionState.Disconnected, oldEndpoint);
+            // Try each endpoint
+            for (var attempt = 0; attempt < _allEndpointUrls.Length; attempt++)
+            {
+                _currentEndpointIndex = (_currentEndpointIndex + 1) % _allEndpointUrls.Length;
+                var url = _allEndpointUrls[_currentEndpointIndex];
+
+                try
+                {
+                    Logger.Information("Failover attempt to {EndpointUrl}", url);
+                    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)
+                    {
+                        session.Dispose();
+                        return;
+                    }
+
+                    _session = session;
+
+                    session.RegisterKeepAliveHandler(isGood =>
+                    {
+                        if (!isGood) LastFailover = HandleKeepAliveFailureAsync();
+                    });
+
+                    CurrentConnectionInfo = BuildConnectionInfo(session);
+                    TransitionState(ConnectionState.Connected, url);
+                    Logger.Information("Failover succeeded to {EndpointUrl}", url);
+
+                    // Replay subscriptions
+                    await ReplaySubscriptionsAsync(session);
+                    return;
+                }
+                catch (Exception ex)
+                {
+                    Logger.Warning(ex, "Failover to {EndpointUrl} failed", url);
+                }
+            }
+
+            Logger.Error("All failover endpoints unreachable");
+            TransitionState(ConnectionState.Disconnected, oldEndpoint);
+        }
+        finally
+        {
+            _subscriptionGate.Release();
+        }
    }

-    private async Task ReplaySubscriptionsAsync()
+    /// <summary>Re-creates the recorded subscriptions on <paramref name="session"/>. Runs under the subscription gate.</summary>
+    private async Task ReplaySubscriptionsAsync(ISessionAdapter session)
    {
        // Snapshot the bookkeeping state under the lock, then clear it so the
        // replayed handles can be recorded fresh as each monitored item is
@@ -788,13 +872,17 @@ public sealed class OpcUaClientService : IOpcUaClientService
        }

        // Replay data subscriptions
+        ISubscriptionAdapter? dataSubscription = null;
        foreach (var (nodeIdStr, (nodeId, intervalMs, _)) in subscriptions)
            try
            {
-                if (_dataSubscription == null)
-                    _dataSubscription = await _session!.CreateSubscriptionAsync(intervalMs, CancellationToken.None);
+                if (dataSubscription == null)
+                {
+                    dataSubscription = await session.CreateSubscriptionAsync(intervalMs, CancellationToken.None);
+                    lock (_subscriptionLock) _dataSubscription = dataSubscription;
+                }

-                var handle = await _dataSubscription.AddDataChangeMonitoredItemAsync(
+                var handle = await dataSubscription.AddDataChangeMonitoredItemAsync(
                    nodeId, intervalMs, OnDataChangeNotification, CancellationToken.None);

                lock (_subscriptionLock)
@@ -814,9 +902,10 @@ public sealed class OpcUaClientService : IOpcUaClientService
            try
            {
                var monitorNode = sourceNodeId ?? ObjectIds.Server;
-                _alarmSubscription = await _session!.CreateSubscriptionAsync(intervalMs, CancellationToken.None);
+                var replayedAlarms = await session.CreateSubscriptionAsync(intervalMs, CancellationToken.None);
+                lock (_subscriptionLock) _alarmSubscription = replayedAlarms;
                var filter = CreateAlarmEventFilter();
-                await _alarmSubscription.AddEventMonitoredItemAsync(
+                await replayedAlarms.AddEventMonitoredItemAsync(
                    monitorNode, intervalMs, filter, OnAlarmEventNotification, CancellationToken.None);

                lock (_subscriptionLock)
@@ -167,15 +167,31 @@ internal sealed class FakeSessionAdapter : ISessionAdapter
        return Task.FromResult<IReadOnlyList<DataValue>>(HistoryReadAggregateResponse);
    }

+    /// <summary>
+    /// Optional gate that, when set, holds <see cref="CreateSubscriptionAsync" /> open until completed, so a test can keep
+    /// a subscribe in flight while something else (a failover) happens. Signalled via <see cref="CreateSubscriptionEntered" />.
+    /// </summary>
+    public TaskCompletionSource? CreateSubscriptionGate { get; set; }
+
+    /// <summary>Completed when <see cref="CreateSubscriptionAsync" /> is entered while a gate is set.</summary>
+    public TaskCompletionSource CreateSubscriptionEntered { get; } =
+        new(TaskCreationOptions.RunContinuationsAsynchronously);
+
    /// <inheritdoc />
-    public Task<ISubscriptionAdapter> CreateSubscriptionAsync(int publishingIntervalMs, CancellationToken ct)
+    public async Task<ISubscriptionAdapter> CreateSubscriptionAsync(int publishingIntervalMs, CancellationToken ct)
    {
+        if (CreateSubscriptionGate is { } gate)
+        {
+            CreateSubscriptionEntered.TrySetResult();
+            await gate.Task;
+        }
+
        lock (_createdSubscriptions)
        {
            var sub = NextSubscription ?? new FakeSubscriptionAdapter();
            NextSubscription = null;
            _createdSubscriptions.Add(sub);
-            return Task.FromResult<ISubscriptionAdapter>(sub);
+            return sub;
        }
    }

@@ -1446,6 +1446,97 @@ public class OpcUaClientServiceTests : IDisposable
        _service.CurrentConnectionInfo!.EndpointUrl.ShouldBe("opc.tcp://backup:4840");
    }

+    /// <summary>
+    /// R08 4.5: a ConnectAsync on an already-connected client is refused instead of overwriting (and leaking) the live
+    /// session. The positive control: after a disconnect the same client connects again.
+    /// </summary>
+    [Fact]
+    public async Task ConnectAsync_WhenAlreadyConnected_IsRefused_AndKeepsTheLiveSession()
+    {
+        var session1 = new FakeSessionAdapter();
+        _sessionFactory.EnqueueSession(session1);
+        await _service.ConnectAsync(ValidSettings());
+
+        await Should.ThrowAsync<InvalidOperationException>(() => _service.ConnectAsync(ValidSettings()));
+
+        _sessionFactory.CreateCallCount.ShouldBe(1);
+        session1.Disposed.ShouldBeFalse();
+        _service.IsConnected.ShouldBeTrue();
+
+        await _service.DisconnectAsync();
+        await _service.ConnectAsync(ValidSettings());
+        _sessionFactory.CreateCallCount.ShouldBe(2);
+        _service.IsConnected.ShouldBeTrue();
+    }
+
+    /// <summary>
+    /// R08 4.5: a SubscribeAsync in flight when a keep-alive failover starts must not be lost. Failover runs under the
+    /// subscription gate, so it waits for the in-flight subscribe to record its item and then replays it onto the new
+    /// session. Before the fix, failover nulled the subscription state outside the lock and finished first; the subscribe
+    /// then completed onto the dying session's subscription and was never replayed.
+    /// </summary>
+    [Fact]
+    public async Task SubscribeAsync_in_flight_during_failover_is_replayed_onto_the_new_session()
+    {
+        var session1 = new FakeSessionAdapter { EndpointUrl = "opc.tcp://primary:4840" };
+        var session2 = new FakeSessionAdapter { EndpointUrl = "opc.tcp://backup:4840" };
+        _sessionFactory.EnqueueSession(session1);
+        _sessionFactory.EnqueueSession(session2);
+        var settings = ValidSettings("opc.tcp://primary:4840");
+        settings.FailoverUrls = ["opc.tcp://backup:4840"];
+        await _service.ConnectAsync(settings);
+
+        var gate = new TaskCompletionSource();
+        session1.CreateSubscriptionGate = gate;
+        var subscribe = _service.SubscribeAsync(new NodeId("ns=2;s=Racer"), 1000);
+        await session1.CreateSubscriptionEntered.Task.WaitAsync(TimeSpan.FromSeconds(10));
+
+        session1.SimulateKeepAlive(false);
+        var failover = _service.LastFailover.ShouldNotBeNull();
+
+        gate.SetResult();
+        await subscribe.WaitAsync(TimeSpan.FromSeconds(10));
+        await failover.WaitAsync(TimeSpan.FromSeconds(10));
+
+        _service.CurrentConnectionInfo!.EndpointUrl.ShouldBe("opc.tcp://backup:4840");
+        var replayed = session2.CreatedSubscriptions.ShouldHaveSingleItem();
+        replayed.AddDataChangeCount.ShouldBe(1);
+
+        // And it is the live subscription: unsubscribing removes it from session2's subscription, not session1's.
+        await _service.UnsubscribeAsync(new NodeId("ns=2;s=Racer"));
+        replayed.RemoveCount.ShouldBe(1);
+        replayed.ActiveHandles.ShouldBeEmpty();
+    }
+
+    /// <summary>
+    /// A DisconnectAsync issued while a failover is still connecting wins: the failover disposes the session it just
+    /// opened instead of resurrecting the connection. The positive control is the plain failover test above, where the
+    /// same gated attempt ends Connected.
+    /// </summary>
+    [Fact]
+    public async Task DisconnectAsync_during_failover_is_not_undone_by_the_failover()
+    {
+        var session1 = new FakeSessionAdapter { EndpointUrl = "opc.tcp://primary:4840" };
+        var session2 = new FakeSessionAdapter { EndpointUrl = "opc.tcp://backup:4840" };
+        _sessionFactory.EnqueueSession(session1);
+        _sessionFactory.EnqueueSession(session2);
+        var settings = ValidSettings("opc.tcp://primary:4840");
+        settings.FailoverUrls = ["opc.tcp://backup:4840"];
+        await _service.ConnectAsync(settings);
+
+        var gate = new TaskCompletionSource();
+        _sessionFactory.CreateGate = gate;
+        session1.SimulateKeepAlive(false);
+        var failover = _service.LastFailover.ShouldNotBeNull();
+
+        await _service.DisconnectAsync();
+        gate.SetResult();
+        await failover.WaitAsync(TimeSpan.FromSeconds(10));
+
+        _service.IsConnected.ShouldBeFalse();
+        session2.Disposed.ShouldBeTrue();
+    }
+
    /// <summary>
    /// Regression for Client.Shared-005: concurrent subscribe/unsubscribe calls mutating the
    /// active-subscription bookkeeping must not corrupt the dictionary or throw.