diff --git a/src/Client/ZB.MOM.WW.OtOpcUa.Client.Shared/Adapters/DefaultSessionAdapter.cs b/src/Client/ZB.MOM.WW.OtOpcUa.Client.Shared/Adapters/DefaultSessionAdapter.cs index dd871594..c414f1c1 100644 --- a/src/Client/ZB.MOM.WW.OtOpcUa.Client.Shared/Adapters/DefaultSessionAdapter.cs +++ b/src/Client/ZB.MOM.WW.OtOpcUa.Client.Shared/Adapters/DefaultSessionAdapter.cs @@ -164,46 +164,16 @@ internal sealed class DefaultSessionAdapter : ISessionAdapter ReturnBounds = false }; - var nodesToRead = new HistoryReadValueIdCollection - { - new HistoryReadValueId { NodeId = nodeId } - }; - - var allValues = new List(); - byte[]? continuationPoint = null; - - do - { - if (continuationPoint != null) - nodesToRead[0].ContinuationPoint = continuationPoint; - - // Use the async overload so this method is genuinely asynchronous, - // honors the cancellation token, and does not block the caller's thread - // (which would block the UI dispatcher for client.ui consumers). - var response = await _session.HistoryReadAsync( - null, - new ExtensionObject(details), - TimestampsToReturn.Source, - continuationPoint != null, - nodesToRead, - ct).ConfigureAwait(false); - - var results = response.Results; - if (results == null || results.Count == 0) - break; - - var result = results[0]; - // A refused read (BadHistoryOperationUnsupported, BadUserAccessDenied, ...) must surface; only - // BadNoData is an empty result. Swallowing it printed "0 values returned" for a refusal. - HistoryReadStatus.ThrowIfRefused(result.StatusCode, nodeId); - - if (result.HistoryData is ExtensionObject ext && ext.Body is HistoryData historyData) - allValues.AddRange(historyData.DataValues); - - continuationPoint = result.ContinuationPoint; - } while (continuationPoint != null && continuationPoint.Length > 0 && allValues.Count < maxValues); - - return allValues; + // Continuation pages are READ, never released, until the server finishes or the cap is reached; a read stopped + // by the cap releases the server's cursor (HistoryReadPaging). This path used to pass + // `releaseContinuationPoints: continuationPoint != null`, which RELEASED the second page instead of reading it, + // silently truncating every raw read to its first server page. + return await HistoryReadPaging.ReadAllAsync( + (cp, release, token) => ReadHistoryPageAsync(details, nodeId, cp, release, TimestampsToReturn.Source, token), + result => result.HistoryData is ExtensionObject { Body: HistoryData historyData } + ? historyData.DataValues + : [], + maxValues, nodeId, ct).ConfigureAwait(false); } /// @@ -219,59 +189,41 @@ internal sealed class DefaultSessionAdapter : ISessionAdapter ReturnBounds = false }; + return await HistoryReadPaging.ReadAllAsync( + (cp, release, token) => ReadHistoryPageAsync(details, nodeId, cp, release, TimestampsToReturn.Both, token), + ToModifiedEntries, maxValues, nodeId, ct).ConfigureAwait(false); + } + + private static IEnumerable ToModifiedEntries(HistoryReadResult result) + { + if (result.HistoryData is not ExtensionObject { Body: HistoryModifiedData modified }) yield break; + for (var i = 0; i < modified.DataValues.Count; i++) + { + var info = i < modified.ModificationInfos.Count ? modified.ModificationInfos[i] : null; + yield return new Models.HistoryModifiedEntry( + modified.DataValues[i], + info?.ModificationTime ?? DateTime.MinValue, + info?.UpdateType ?? default, + info?.UserName); + } + } + + /// Issues one single-node HistoryRead (or a release of its continuation point). + private async Task ReadHistoryPageAsync( + ReadRawModifiedDetails details, NodeId nodeId, byte[]? continuationPoint, bool release, + TimestampsToReturn timestamps, CancellationToken ct) + { var nodesToRead = new HistoryReadValueIdCollection { - new HistoryReadValueId { NodeId = nodeId } + new HistoryReadValueId { NodeId = nodeId, ContinuationPoint = continuationPoint } }; - var all = new List(); - byte[]? continuationPoint = null; + // The async overload honours the token and does not block the caller's thread (the UI dispatcher for + // client.ui consumers). + var response = await _session.HistoryReadAsync( + null, new ExtensionObject(details), timestamps, release, nodesToRead, ct).ConfigureAwait(false); - do - { - if (continuationPoint != null) - nodesToRead[0].ContinuationPoint = continuationPoint; - - var response = await _session.HistoryReadAsync( - null, - new ExtensionObject(details), - TimestampsToReturn.Both, - false, - nodesToRead, - ct).ConfigureAwait(false); - - var results = response.Results; - if (results == null || results.Count == 0) - break; - - var result = results[0]; - HistoryReadStatus.ThrowIfRefused(result.StatusCode, nodeId); - - if (result.HistoryData is ExtensionObject ext && ext.Body is HistoryModifiedData modified) - { - for (var i = 0; i < modified.DataValues.Count; i++) - { - var info = i < modified.ModificationInfos.Count ? modified.ModificationInfos[i] : null; - all.Add(new Models.HistoryModifiedEntry( - modified.DataValues[i], - info?.ModificationTime ?? DateTime.MinValue, - info?.UpdateType ?? default, - info?.UserName)); - } - } - - continuationPoint = result.ContinuationPoint; - } while (continuationPoint != null && continuationPoint.Length > 0 && all.Count < maxValues); - - // A read stopped by the client's own cap leaves a server cursor behind; release it. - if (continuationPoint is { Length: > 0 }) - { - nodesToRead[0].ContinuationPoint = continuationPoint; - await _session.HistoryReadAsync(null, new ExtensionObject(details), TimestampsToReturn.Both, true, - nodesToRead, ct).ConfigureAwait(false); - } - - return all; + return response.Results is { Count: > 0 } results ? results[0] : null; } /// @@ -327,9 +279,22 @@ internal sealed class DefaultSessionAdapter : ISessionAdapter }; _session.AddSubscription(subscription); - await subscription.CreateAsync(ct); + try + { + await subscription.CreateAsync(ct); + } + catch + { + // A subscription the server never created must not linger in Session.Subscriptions either. + try { await _session.RemoveSubscriptionAsync(subscription, CancellationToken.None); } + catch (Exception ex) { Logger.Warning(ex, "Error removing a subscription that failed to create"); } + throw; + } - return new DefaultSubscriptionAdapter(subscription); + // Deleting the adapter also removes the subscription from the SDK session, so a re-targeted alarm/data + // subscription does not accumulate dead Subscription objects in Session.Subscriptions. + return new DefaultSubscriptionAdapter(subscription, + async (s, token) => await _session.RemoveSubscriptionAsync(s, token).ConfigureAwait(false)); } /// 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 deb15dd5..927e0094 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 @@ -12,14 +12,23 @@ internal sealed class DefaultSubscriptionAdapter : ISubscriptionAdapter private static readonly ILogger Logger = Log.ForContext(); private readonly Dictionary _monitoredItems = new(); private readonly Subscription _subscription; + private readonly Func? _detach; + private int _detached; /// /// Wraps a live OPC UA subscription so client code can manage monitored items through a testable abstraction. /// /// The underlying OPC UA subscription that owns monitored items for this client workflow. - public DefaultSubscriptionAdapter(Subscription subscription) + /// + /// Removes the subscription from its owning SDK Session once it is deleted. Without it a deleted + /// subscription stays in Session.Subscriptions until the session is disposed, so a UI that re-targets + /// alarms repeatedly accumulates dead subscriptions the SDK still iterates (transfer / recreate on reconnect). + /// + public DefaultSubscriptionAdapter(Subscription subscription, + Func? detach = null) { _subscription = subscription; + _detach = detach; } /// @@ -113,9 +122,24 @@ internal sealed class DefaultSubscriptionAdapter : ISubscriptionAdapter Logger.Warning(ex, "Error deleting subscription"); } + // Detach whether or not the server-side delete succeeded: the adapter is finished with it either way. + await DetachAsync(ct); _monitoredItems.Clear(); } + private async Task DetachAsync(CancellationToken ct) + { + if (_detach == null || Interlocked.Exchange(ref _detached, 1) != 0) return; + try + { + await _detach(_subscription, ct); + } + catch (Exception ex) + { + Logger.Warning(ex, "Error removing subscription from its session"); + } + } + /// /// Releases the wrapped OPC UA subscription and clears tracked monitored items held by the adapter. /// @@ -129,6 +153,14 @@ internal sealed class DefaultSubscriptionAdapter : ISubscriptionAdapter { } + try + { + DisposeBridge.Run(DetachAsync); + } + catch + { + } + _monitoredItems.Clear(); } } 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 new file mode 100644 index 00000000..2ec33e93 --- /dev/null +++ b/src/Client/ZB.MOM.WW.OtOpcUa.Client.Shared/Helpers/HistoryReadPaging.cs @@ -0,0 +1,66 @@ +using Opc.Ua; +using Serilog; + +namespace ZB.MOM.WW.OtOpcUa.Client.Shared.Helpers; + +/// +/// The client-side HistoryRead paging loop shared by the raw and modified reads. Continuation pages are +/// READ (releaseContinuationPoints = false) while the caller still wants values; a read stopped by the +/// caller's own cap while the server still holds a cursor releases that cursor so it does not occupy one of the +/// session's continuation-point slots until eviction. +/// +internal static class HistoryReadPaging +{ + private static readonly ILogger Logger = Log.ForContext(typeof(HistoryReadPaging)); + + /// Reads (or, with , releases) one page. + /// The cursor to resume, or for the first page. + /// to release the cursor instead of reading. + /// The caller's cancellation token. + /// The per-node result, or when the response carried none. + public delegate Task PageReader( + byte[]? continuationPoint, bool releaseContinuationPoints, CancellationToken ct); + + /// 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 node being read, for refusal messages. + /// The caller's cancellation token. + /// Every element read. + /// A page was refused (see ). + public static async Task> ReadAllAsync( + PageReader readPage, Func> extract, int maxValues, NodeId nodeId, + CancellationToken ct) + { + 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); + } + catch (Exception ex) when (ex is not OperationCanceledException) + { + Logger.Warning(ex, "Releasing the HistoryRead continuation point for {NodeId} failed", nodeId); + } + + return all; + } +} diff --git a/tests/Client/ZB.MOM.WW.OtOpcUa.Client.Shared.Tests/Adapters/DefaultSubscriptionAdapterTests.cs b/tests/Client/ZB.MOM.WW.OtOpcUa.Client.Shared.Tests/Adapters/DefaultSubscriptionAdapterTests.cs new file mode 100644 index 00000000..07980b36 --- /dev/null +++ b/tests/Client/ZB.MOM.WW.OtOpcUa.Client.Shared.Tests/Adapters/DefaultSubscriptionAdapterTests.cs @@ -0,0 +1,57 @@ +using Opc.Ua.Client; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Client.Shared.Adapters; + +namespace ZB.MOM.WW.OtOpcUa.Client.Shared.Tests.Adapters; + +/// +/// A deleted subscription must also leave the SDK session's Subscriptions list, or every alarm re-target in the UI +/// leaves a dead behind until the session is disposed. +/// +public class DefaultSubscriptionAdapterTests +{ + /// DeleteAsync detaches the wrapped subscription exactly once, even when Dispose follows. + [Fact] + public async Task DeleteAsync_detaches_the_subscription_from_its_session_once() + { + var subscription = new Subscription(); + var detached = new List(); + var adapter = new DefaultSubscriptionAdapter(subscription, (s, _) => + { + lock (detached) detached.Add(s); + return Task.CompletedTask; + }); + + await adapter.DeleteAsync(TestContext.Current.CancellationToken); + adapter.Dispose(); + + detached.ShouldHaveSingleItem().ShouldBeSameAs(subscription); + } + + /// Dispose alone (no prior DeleteAsync) detaches too. + [Fact] + public void Dispose_detaches_the_subscription_from_its_session() + { + var detached = 0; + var adapter = new DefaultSubscriptionAdapter(new Subscription(), (_, _) => + { + Interlocked.Increment(ref detached); + return Task.CompletedTask; + }); + + adapter.Dispose(); + + detached.ShouldBe(1); + } + + /// A failing detach is logged, not thrown: teardown stays best effort. + [Fact] + public async Task A_failing_detach_does_not_throw_from_DeleteAsync() + { + var adapter = new DefaultSubscriptionAdapter(new Subscription(), + (_, _) => throw new InvalidOperationException("boom")); + + await Should.NotThrowAsync(() => adapter.DeleteAsync(TestContext.Current.CancellationToken)); + } +} 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 new file mode 100644 index 00000000..0f9240b0 --- /dev/null +++ b/tests/Client/ZB.MOM.WW.OtOpcUa.Client.Shared.Tests/Helpers/HistoryReadPagingTests.cs @@ -0,0 +1,144 @@ +using Opc.Ua; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Client.Shared.Helpers; + +namespace ZB.MOM.WW.OtOpcUa.Client.Shared.Tests.Helpers; + +/// +/// Pins the client-side HistoryRead paging loop shared by the raw and modified reads: continuation pages are read +/// (never released) while the caller wants more, and a read stopped by the caller's own cap releases the server cursor. +/// +public class HistoryReadPagingTests +{ + private static readonly NodeId Node = new("t", 2); + + /// One recorded call to the page reader. + private sealed record Call(byte[]? ContinuationPoint, bool Release); + + /// A scripted server: each read returns the next page; a release returns an empty Good result. + private sealed class ScriptedServer + { + private readonly Queue<(int Count, byte[]? Cp)> _pages; + public List Calls { get; } = []; + + public ScriptedServer(params (int Count, byte[]? Cp)[] pages) => _pages = new Queue<(int, byte[]?)>(pages); + + public Task ReadPageAsync(byte[]? cp, bool release, CancellationToken ct) + { + Calls.Add(new Call(cp, release)); + if (release) return Task.FromResult(new HistoryReadResult { StatusCode = StatusCodes.Good }); + var (count, next) = _pages.Dequeue(); + var data = new HistoryData(); + for (var i = 0; i < count; i++) data.DataValues.Add(new DataValue(new Variant(i))); + return Task.FromResult(new HistoryReadResult + { + StatusCode = StatusCodes.Good, + ContinuationPoint = next, + HistoryData = new ExtensionObject(data) + }); + } + } + + private static IEnumerable Values(HistoryReadResult r) => + r.HistoryData?.Body is HistoryData d ? d.DataValues : []; + + /// + /// A server page limit below the caller's cap hands back a continuation point: the follow-up must READ the next + /// page. Releasing it instead (the old raw path passed release = cp != null) silently truncated every raw read + /// to its first server page. + /// + [Fact] + public async Task Continuation_pages_are_read_not_released_until_the_server_finishes() + { + var server = new ScriptedServer((3, [1]), (3, [2]), (2, null)); + + var values = await HistoryReadPaging.ReadAllAsync(server.ReadPageAsync, Values, maxValues: 100, Node, + CancellationToken.None); + + values.Count.ShouldBe(8); + server.Calls.Select(c => c.Release).ShouldAllBe(r => r == false); + server.Calls.Select(c => c.ContinuationPoint?.FirstOrDefault()).ShouldBe(new byte?[] { null, 1, 2 }); + } + + /// A read stopped by the caller's cap while the server still holds a cursor releases that cursor. + [Fact] + public async Task A_read_stopped_by_the_cap_releases_the_outstanding_continuation_point() + { + var server = new ScriptedServer((5, [7])); + + var values = await HistoryReadPaging.ReadAllAsync(server.ReadPageAsync, Values, maxValues: 5, Node, + CancellationToken.None); + + values.Count.ShouldBe(5); + server.Calls.Count.ShouldBe(2); + server.Calls[1].Release.ShouldBeTrue(); + server.Calls[1].ContinuationPoint.ShouldBe(new byte[] { 7 }); + } + + /// + /// A read the server finished (no continuation point) has nothing to release — and the positive control above + /// proves a release would have been recorded had one been owed. + /// + [Fact] + public async Task A_read_the_server_finished_issues_no_release() + { + var server = new ScriptedServer((5, null)); + + var values = await HistoryReadPaging.ReadAllAsync(server.ReadPageAsync, Values, maxValues: 5, Node, + CancellationToken.None); + + values.Count.ShouldBe(5); + 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() + { + var server = new ScriptedServer((2, [1]), (2, [2]), (1, null)); + + var values = await HistoryReadPaging.ReadAllAsync(server.ReadPageAsync, Values, maxValues: 0, Node, + CancellationToken.None); + + values.Count.ShouldBe(5); + server.Calls.ShouldAllBe(c => !c.Release); + } + + /// A failed release is best effort: the values already read are still returned. + [Fact] + public async Task A_failed_release_does_not_lose_the_values_read() + { + var calls = 0; + Task Read(byte[]? cp, bool release, CancellationToken ct) + { + calls++; + if (release) throw new ServiceResultException(StatusCodes.BadSessionClosed); + var data = new HistoryData(); + data.DataValues.Add(new DataValue(new Variant(1))); + return Task.FromResult(new HistoryReadResult + { + StatusCode = StatusCodes.Good, ContinuationPoint = [9], HistoryData = new ExtensionObject(data) + }); + } + + var values = await HistoryReadPaging.ReadAllAsync(Read, Values, maxValues: 1, Node, CancellationToken.None); + + values.Count.ShouldBe(1); + calls.ShouldBe(2); + } + + /// A refused page still surfaces as an exception, not an empty read. + [Fact] + public async Task A_refused_page_throws() + { + Task Read(byte[]? cp, bool release, CancellationToken ct) => + Task.FromResult(new HistoryReadResult + { + StatusCode = StatusCodes.BadHistoryOperationUnsupported + }); + + await Should.ThrowAsync(() => + HistoryReadPaging.ReadAllAsync(Read, Values, maxValues: 10, Node, CancellationToken.None)); + } +}