From 0c9e4f4e3d1fd0cb8fd74a874d51338904e93743 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 17:18:26 -0400 Subject: [PATCH] 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 --- .../Adapters/DefaultSubscriptionAdapter.cs | 17 +- .../Helpers/HistoryReadPaging.cs | 72 ++++--- .../OpcUaClientService.cs | 143 ++++++++++---- .../FailoverEpochTests.cs | 177 ++++++++++++++++++ .../Fakes/FakeSessionFactory.cs | 13 +- .../Fakes/FakeSubscriptionAdapter.cs | 5 + .../Helpers/HistoryReadPagingTests.cs | 89 ++++++++- 7 files changed, 442 insertions(+), 74 deletions(-) create mode 100644 tests/Client/ZB.MOM.WW.OtOpcUa.Client.Shared.Tests/FailoverEpochTests.cs diff --git a/src/Client/ZB.MOM.WW.OtOpcUa.Client.Shared/Adapters/DefaultSubscriptionAdapter.cs b/src/Client/ZB.MOM.WW.OtOpcUa.Client.Shared/Adapters/DefaultSubscriptionAdapter.cs index 927e0094..6473abf6 100644 --- a/src/Client/ZB.MOM.WW.OtOpcUa.Client.Shared/Adapters/DefaultSubscriptionAdapter.cs +++ b/src/Client/ZB.MOM.WW.OtOpcUa.Client.Shared/Adapters/DefaultSubscriptionAdapter.cs @@ -145,17 +145,16 @@ internal sealed class DefaultSubscriptionAdapter : ISubscriptionAdapter /// 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 { diff --git a/src/Client/ZB.MOM.WW.OtOpcUa.Client.Shared/Helpers/HistoryReadPaging.cs b/src/Client/ZB.MOM.WW.OtOpcUa.Client.Shared/Helpers/HistoryReadPaging.cs index 2ec33e93..16bb72b6 100644 --- a/src/Client/ZB.MOM.WW.OtOpcUa.Client.Shared/Helpers/HistoryReadPaging.cs +++ b/src/Client/ZB.MOM.WW.OtOpcUa.Client.Shared/Helpers/HistoryReadPaging.cs @@ -21,46 +21,76 @@ internal static class HistoryReadPaging public delegate Task PageReader( byte[]? continuationPoint, bool releaseContinuationPoints, CancellationToken ct); + /// The bound on a best-effort release that does not ride the caller's (possibly cancelled) token. + internal static readonly TimeSpan ReleaseTimeout = TimeSpan.FromSeconds(5); + /// Reads pages until the server finishes or the caller's cap is reached. /// The element type extracted from each page. /// Issues one HistoryRead for the single node. /// Extracts the page's elements from a result. - /// The caller's cap; <= 0 means no cap (follow every continuation point). + /// The caller's cap; must be positive. The result is trimmed to it. /// The node being read, for refusal messages. /// The caller's cancellation token. - /// Every element read. + /// At most elements. + /// is not positive: an uncapped client + /// read would materialise a tag's whole window (the client side of review 08 1.1). /// A page was refused (see ). + /// was cancelled mid-read; the outstanding + /// cursor is released (best effort) first. public static async Task> ReadAllAsync( PageReader readPage, Func> extract, int maxValues, NodeId nodeId, CancellationToken ct) { + ArgumentOutOfRangeException.ThrowIfNegativeOrZero(maxValues); + var all = new List(); 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 Trim(List 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; } } 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 9d78961e..272e3695 100644 --- a/src/Client/ZB.MOM.WW.OtOpcUa.Client.Shared/OpcUaClientService.cs +++ b/src/Client/ZB.MOM.WW.OtOpcUa.Client.Shared/OpcUaClientService.cs @@ -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; + /// The most recent keep-alive-triggered failover task (test seam: lets a test await it). 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 /// 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() + /// A bad keep-alive from the session of connection . + 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; + } + + /// + /// A state transition made only while is still current (checked and applied under + /// , the event raised outside it). + /// + /// when the epoch has moved on (the caller must abandon). + 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)); } diff --git a/tests/Client/ZB.MOM.WW.OtOpcUa.Client.Shared.Tests/FailoverEpochTests.cs b/tests/Client/ZB.MOM.WW.OtOpcUa.Client.Shared.Tests/FailoverEpochTests.cs new file mode 100644 index 00000000..362bc456 --- /dev/null +++ b/tests/Client/ZB.MOM.WW.OtOpcUa.Client.Shared.Tests/FailoverEpochTests.cs @@ -0,0 +1,177 @@ +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Client.Shared.Models; +using ZB.MOM.WW.OtOpcUa.Client.Shared.Tests.Fakes; + +namespace ZB.MOM.WW.OtOpcUa.Client.Shared.Tests; + +/// +/// Lane K review I1: a keep-alive failover belongs to the connection it started under. A user Disconnect + Connect +/// while it is still connecting starts a new connection epoch; the stale failover must abandon itself — never +/// overwrite the user's new session, never flip its state. Every step is gate-driven (no sleeps). +/// +public sealed class FailoverEpochTests : IDisposable +{ + private const string Primary = "opc.tcp://primary:4840"; + private const string Backup = "opc.tcp://backup:4840"; + private static readonly TimeSpan Wait = TimeSpan.FromSeconds(10); + + private readonly FakeSessionFactory _factory = new(); + private readonly OpcUaClientService _service; + private readonly List _states = []; + + public FailoverEpochTests() + { + _service = new OpcUaClientService(new FakeApplicationConfigurationFactory(), new FakeEndpointDiscovery(), _factory); + _service.ConnectionStateChanged += (_, e) => { lock (_states) _states.Add(e.NewState); }; + } + + public void Dispose() => _service.Dispose(); + + private static ConnectionSettings Settings() => new() + { + EndpointUrl = Primary, FailoverUrls = [Backup], SessionTimeoutSeconds = 60 + }; + + /// Connects S1, then holds a keep-alive failover inside its connect attempt. + private async Task<(FakeSessionAdapter S1, Task Failover, TaskCompletionSource Gate)> StartHeldFailoverAsync() + { + var s1 = new FakeSessionAdapter(); + _factory.EnqueueSession(s1); + await _service.ConnectAsync(Settings()); + + var gate = new TaskCompletionSource(); + _factory.CreateGate = gate; + s1.SimulateKeepAlive(false); + var failover = _service.LastFailover.ShouldNotBeNull(); + await _factory.CreateEntered.Task.WaitAsync(Wait); + return (s1, failover, gate); + } + + /// The stale failover's attempt SUCCEEDS after the user reconnected: its session is disposed, the user's stays. + [Fact] + public async Task A_stale_failover_that_connects_does_not_replace_the_users_new_session() + { + var (_, failover, gate) = await StartHeldFailoverAsync(); + var userSession = new FakeSessionAdapter(); + var failoverSession = new FakeSessionAdapter(); + _factory.EnqueueSession(userSession); // dequeued by the user's Connect (the failover is still held) + _factory.EnqueueSession(failoverSession); // dequeued by the failover once released + + await _service.DisconnectAsync(); + await _service.ConnectAsync(Settings()); + lock (_states) _states.Clear(); + + gate.SetResult(); + await failover.WaitAsync(Wait); + + failoverSession.Disposed.ShouldBeTrue(); + userSession.Disposed.ShouldBeFalse(); + _service.IsConnected.ShouldBeTrue(); + _service.CurrentConnectionInfo!.EndpointUrl.ShouldBe(Primary); + lock (_states) _states.ShouldBeEmpty(); // the stale failover made no state transition at all + } + + /// The stale failover's attempts all FAIL after the user reconnected: the user's connection stays Connected. + [Fact] + public async Task A_stale_failover_that_fails_does_not_flip_the_users_connection_to_Disconnected() + { + var (_, failover, gate) = await StartHeldFailoverAsync(); + _factory.EnqueueSession(new FakeSessionAdapter()); + + await _service.DisconnectAsync(); + await _service.ConnectAsync(Settings()); + lock (_states) _states.Clear(); + var createsBefore = _factory.CreateCallCount; + + _factory.ThrowOnCreate = true; + gate.SetResult(); + await failover.WaitAsync(Wait); + + _service.IsConnected.ShouldBeTrue(); + lock (_states) _states.ShouldBeEmpty(); + // It stopped at the abandoned attempt instead of trying every remaining endpoint (review m5). + _factory.CreateCallCount.ShouldBe(createsBefore); + // And the user can still disconnect / reconnect normally afterwards. + _factory.ThrowOnCreate = false; + await _service.DisconnectAsync(); + await _service.ConnectAsync(Settings()); + _service.IsConnected.ShouldBeTrue(); + } + + /// Positive control: an undisturbed failover still reconnects to the backup. + [Fact] + public async Task A_genuine_failover_still_reconnects() + { + var (_, failover, gate) = await StartHeldFailoverAsync(); + _factory.EnqueueSession(new FakeSessionAdapter()); + + gate.SetResult(); + await failover.WaitAsync(Wait); + + _service.IsConnected.ShouldBeTrue(); + _service.CurrentConnectionInfo!.EndpointUrl.ShouldBe(Backup); + lock (_states) _states.ShouldContain(ConnectionState.Reconnecting); + } + + /// + /// A late bad keep-alive from the PREVIOUS connection's session must not start a failover against the new one. + /// Positive control: the same signal from the current session does (the genuine-failover test above). + /// + [Fact] + public async Task A_keep_alive_from_a_previous_connection_does_not_fail_over_the_current_one() + { + var s1 = new FakeSessionAdapter(); + _factory.EnqueueSession(s1); + await _service.ConnectAsync(Settings()); + await _service.DisconnectAsync(); + _factory.EnqueueSession(new FakeSessionAdapter()); + await _service.ConnectAsync(Settings()); + var creates = _factory.CreateCallCount; + lock (_states) _states.Clear(); + + s1.SimulateKeepAlive(false); + if (_service.LastFailover is { } t) await t.WaitAsync(Wait); + + _factory.CreateCallCount.ShouldBe(creates); + lock (_states) _states.ShouldBeEmpty(); + _service.IsConnected.ShouldBeTrue(); + } + + /// Review m8: re-entrant bad keep-alives do not overwrite the seam with an already-completed no-op. + [Fact] + public async Task Reentrant_keep_alives_leave_the_seam_on_the_running_failover() + { + var (s1, failover, gate) = await StartHeldFailoverAsync(); + s1.SimulateKeepAlive(false); + s1.SimulateKeepAlive(false); + + _service.LastFailover.ShouldBeSameAs(failover); + failover.IsCompleted.ShouldBeFalse(); + _factory.EnqueueSession(new FakeSessionAdapter()); + gate.SetResult(); + await failover.WaitAsync(Wait); + } + + /// + /// Review m5: an Unsubscribe whose server-side remove fails (the session died under it) still drops the + /// bookkeeping, so a failover does not replay an item the caller asked to stop. Proven by a re-subscribe of the + /// same node creating a fresh monitored item instead of returning "already subscribed". + /// + [Fact] + public async Task A_failed_unsubscribe_still_forgets_the_item() + { + var sub = new FakeSubscriptionAdapter(); + _factory.EnqueueSession(new FakeSessionAdapter { NextSubscription = sub }); + await _service.ConnectAsync(Settings()); + var node = new Opc.Ua.NodeId("ns=2;s=X"); + await _service.SubscribeAsync(node); + + sub.ThrowOnRemove = true; + await Should.ThrowAsync(() => _service.UnsubscribeAsync(node)); + sub.ThrowOnRemove = false; + + await _service.SubscribeAsync(node); + sub.AddDataChangeCount.ShouldBe(2); + } +} diff --git a/tests/Client/ZB.MOM.WW.OtOpcUa.Client.Shared.Tests/Fakes/FakeSessionFactory.cs b/tests/Client/ZB.MOM.WW.OtOpcUa.Client.Shared.Tests/Fakes/FakeSessionFactory.cs index c1189a78..24cbd13a 100644 --- a/tests/Client/ZB.MOM.WW.OtOpcUa.Client.Shared.Tests/Fakes/FakeSessionFactory.cs +++ b/tests/Client/ZB.MOM.WW.OtOpcUa.Client.Shared.Tests/Fakes/FakeSessionFactory.cs @@ -21,6 +21,9 @@ internal sealed class FakeSessionFactory : ISessionFactory /// public TaskCompletionSource? CreateGate { get; set; } + /// Completed when a call observes . + public TaskCompletionSource CreateEntered { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); + /// Gets the list of sessions created via CreateSessionAsync. public IReadOnlyList CreatedSessions => _createdSessions; @@ -39,8 +42,14 @@ internal sealed class FakeSessionFactory : ISessionFactory CreateCallCount++; LastEndpointUrl = endpoint.EndpointUrl; - if (CreateGate != null) - await CreateGate.Task; + // One-shot: the gate holds only the call that observed it, so a later call (e.g. a user's Connect while a + // failover is held) runs freely. + if (CreateGate is { } gate) + { + CreateGate = null; + CreateEntered.TrySetResult(); + await gate.Task; + } if (ThrowOnCreate) throw new InvalidOperationException("FakeSessionFactory configured to fail."); diff --git a/tests/Client/ZB.MOM.WW.OtOpcUa.Client.Shared.Tests/Fakes/FakeSubscriptionAdapter.cs b/tests/Client/ZB.MOM.WW.OtOpcUa.Client.Shared.Tests/Fakes/FakeSubscriptionAdapter.cs index 0c7c4233..1e442575 100644 --- a/tests/Client/ZB.MOM.WW.OtOpcUa.Client.Shared.Tests/Fakes/FakeSubscriptionAdapter.cs +++ b/tests/Client/ZB.MOM.WW.OtOpcUa.Client.Shared.Tests/Fakes/FakeSubscriptionAdapter.cs @@ -37,6 +37,9 @@ internal sealed class FakeSubscriptionAdapter : ISubscriptionAdapter /// Gets the number of times AddEventMonitoredItemAsync was called. public int AddEventCount { get; private set; } + /// Gets or sets a value indicating whether RemoveMonitoredItemAsync throws (a session that died). + public bool ThrowOnRemove { get; set; } + /// Gets the number of times RemoveMonitoredItemAsync was called. public int RemoveCount { get; private set; } @@ -70,6 +73,8 @@ internal sealed class FakeSubscriptionAdapter : ISubscriptionAdapter /// public Task RemoveMonitoredItemAsync(uint clientHandle, CancellationToken ct) { + if (ThrowOnRemove) + throw new ServiceResultException(StatusCodes.BadSessionClosed, "Session closed"); lock (_itemsLock) { RemoveCount++; diff --git a/tests/Client/ZB.MOM.WW.OtOpcUa.Client.Shared.Tests/Helpers/HistoryReadPagingTests.cs b/tests/Client/ZB.MOM.WW.OtOpcUa.Client.Shared.Tests/Helpers/HistoryReadPagingTests.cs index 0f9240b0..bd8c3d1c 100644 --- a/tests/Client/ZB.MOM.WW.OtOpcUa.Client.Shared.Tests/Helpers/HistoryReadPagingTests.cs +++ b/tests/Client/ZB.MOM.WW.OtOpcUa.Client.Shared.Tests/Helpers/HistoryReadPagingTests.cs @@ -92,17 +92,92 @@ public class HistoryReadPagingTests server.Calls.ShouldHaveSingleItem().Release.ShouldBeFalse(); } - /// An uncapped read (maxValues <= 0) follows every continuation point to the end. - [Fact] - public async Task An_uncapped_read_follows_every_page() + /// + /// Review m3: a non-positive cap is refused. An uncapped client read would materialise a tag's whole window + /// (the client side of R08 1.1), and (uint)-1 on the wire asked the server for everything. + /// + /// The cap. + [Theory] + [InlineData(0)] + [InlineData(-1)] + public async Task A_non_positive_cap_is_refused_before_any_read(int max) { - var server = new ScriptedServer((2, [1]), (2, [2]), (1, null)); + var server = new ScriptedServer(); - var values = await HistoryReadPaging.ReadAllAsync(server.ReadPageAsync, Values, maxValues: 0, Node, + await Should.ThrowAsync(() => + HistoryReadPaging.ReadAllAsync(server.ReadPageAsync, Values, max, Node, CancellationToken.None)); + server.Calls.ShouldBeEmpty(); + } + + /// Review m2: server pages that overshoot the cap are trimmed to it (60 + 60 with a cap of 100 is 100). + [Fact] + public async Task The_result_is_trimmed_to_the_cap() + { + var server = new ScriptedServer((60, [1]), (60, [2])); + + var values = await HistoryReadPaging.ReadAllAsync(server.ReadPageAsync, Values, maxValues: 100, Node, CancellationToken.None); - values.Count.ShouldBe(5); - server.Calls.ShouldAllBe(c => !c.Release); + values.Count.ShouldBe(100); + server.Calls.Last().Release.ShouldBeTrue(); // the outstanding cursor is still released + } + + /// + /// Review m1: a read cancelled between pages releases the outstanding cursor (best effort, not on the caller's + /// cancelled token) and still surfaces the cancellation. Positive control: the release IS recorded. + /// + [Fact] + public async Task A_read_cancelled_mid_paging_releases_the_cursor_and_rethrows() + { + using var cts = new CancellationTokenSource(); + var calls = new List<(byte[]? Cp, bool Release, bool TokenCancelled)>(); + Task Read(byte[]? cp, bool release, CancellationToken ct) + { + calls.Add((cp, release, ct.IsCancellationRequested)); + if (release) return Task.FromResult(new HistoryReadResult { StatusCode = StatusCodes.Good }); + if (cp != null) { cts.Cancel(); ct.ThrowIfCancellationRequested(); } + var data = new HistoryData(); + data.DataValues.Add(new DataValue(new Variant(1))); + return Task.FromResult(new HistoryReadResult + { + StatusCode = StatusCodes.Good, ContinuationPoint = [5], HistoryData = new ExtensionObject(data) + }); + } + + await Should.ThrowAsync(() => + HistoryReadPaging.ReadAllAsync(Read, Values, maxValues: 10, Node, cts.Token)); + + var release = calls.Where(c => c.Release).ShouldHaveSingleItem(); + release.Cp.ShouldBe(new byte[] { 5 }); + release.TokenCancelled.ShouldBeFalse(); + } + + /// A cancellation that lands while the cap release is in flight does not lose the values read. + [Fact] + public async Task The_cap_release_does_not_ride_the_callers_token() + { + using var cts = new CancellationTokenSource(); + var releaseSawCancelled = (bool?)null; + Task Read(byte[]? cp, bool release, CancellationToken ct) + { + if (release) + { + releaseSawCancelled = ct.IsCancellationRequested; + return Task.FromResult(new HistoryReadResult { StatusCode = StatusCodes.Good }); + } + var data = new HistoryData(); + data.DataValues.Add(new DataValue(new Variant(1))); + cts.Cancel(); // the caller gives up right as the capped page arrives + return Task.FromResult(new HistoryReadResult + { + StatusCode = StatusCodes.Good, ContinuationPoint = [5], HistoryData = new ExtensionObject(data) + }); + } + + var values = await HistoryReadPaging.ReadAllAsync(Read, Values, maxValues: 1, Node, cts.Token); + + values.Count.ShouldBe(1); + releaseSawCancelled.ShouldBe(false); } /// A failed release is best effort: the values already read are still returned.