diff --git a/src/Client/ZB.MOM.WW.OtOpcUa.Client.Shared/OpcUaClientService.cs b/src/Client/ZB.MOM.WW.OtOpcUa.Client.Shared/OpcUaClientService.cs index fd3f1971..9d78961e 100644 --- a/src/Client/ZB.MOM.WW.OtOpcUa.Client.Shared/OpcUaClientService.cs +++ b/src/Client/ZB.MOM.WW.OtOpcUa.Client.Shared/OpcUaClientService.cs @@ -15,16 +15,17 @@ public sealed class OpcUaClientService : IOpcUaClientService { private static readonly ILogger Logger = Log.ForContext(); - // 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 _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; + /// The most recent keep-alive-triggered failover task (test seam: lets a test await it). + 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() + /// Re-creates the recorded subscriptions on . Runs under the subscription gate. + 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) diff --git a/tests/Client/ZB.MOM.WW.OtOpcUa.Client.Shared.Tests/Fakes/FakeSessionAdapter.cs b/tests/Client/ZB.MOM.WW.OtOpcUa.Client.Shared.Tests/Fakes/FakeSessionAdapter.cs index 14f1df52..27504765 100644 --- a/tests/Client/ZB.MOM.WW.OtOpcUa.Client.Shared.Tests/Fakes/FakeSessionAdapter.cs +++ b/tests/Client/ZB.MOM.WW.OtOpcUa.Client.Shared.Tests/Fakes/FakeSessionAdapter.cs @@ -167,15 +167,31 @@ internal sealed class FakeSessionAdapter : ISessionAdapter return Task.FromResult>(HistoryReadAggregateResponse); } + /// + /// Optional gate that, when set, holds open until completed, so a test can keep + /// a subscribe in flight while something else (a failover) happens. Signalled via . + /// + public TaskCompletionSource? CreateSubscriptionGate { get; set; } + + /// Completed when is entered while a gate is set. + public TaskCompletionSource CreateSubscriptionEntered { get; } = + new(TaskCreationOptions.RunContinuationsAsynchronously); + /// - public Task CreateSubscriptionAsync(int publishingIntervalMs, CancellationToken ct) + public async Task 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(sub); + return sub; } } diff --git a/tests/Client/ZB.MOM.WW.OtOpcUa.Client.Shared.Tests/OpcUaClientServiceTests.cs b/tests/Client/ZB.MOM.WW.OtOpcUa.Client.Shared.Tests/OpcUaClientServiceTests.cs index f1035075..a0b89bb9 100644 --- a/tests/Client/ZB.MOM.WW.OtOpcUa.Client.Shared.Tests/OpcUaClientServiceTests.cs +++ b/tests/Client/ZB.MOM.WW.OtOpcUa.Client.Shared.Tests/OpcUaClientServiceTests.cs @@ -1446,6 +1446,97 @@ public class OpcUaClientServiceTests : IDisposable _service.CurrentConnectionInfo!.EndpointUrl.ShouldBe("opc.tcp://backup:4840"); } + /// + /// 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. + /// + [Fact] + public async Task ConnectAsync_WhenAlreadyConnected_IsRefused_AndKeepsTheLiveSession() + { + var session1 = new FakeSessionAdapter(); + _sessionFactory.EnqueueSession(session1); + await _service.ConnectAsync(ValidSettings()); + + await Should.ThrowAsync(() => _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(); + } + + /// + /// 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. + /// + [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(); + } + + /// + /// 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. + /// + [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(); + } + /// /// Regression for Client.Shared-005: concurrent subscribe/unsubscribe calls mutating the /// active-subscription bookkeeping must not corrupt the dictionary or throw.