fix(client): release a capped HistoryRead cursor, read (not release) follow-up pages, detach deleted subscriptions

R08 1.4: raw and modified history reads share HistoryReadPaging. A read stopped by the caller's cap releases the server continuation point. The raw path also passed releaseContinuationPoints = (cp != null), which released page two instead of reading it, so every raw read was silently truncated to its first server page. R08 1.5: DefaultSubscriptionAdapter removes its Subscription from the SDK Session on delete/dispose (and a failed CreateAsync removes it too), so re-targeted alarm subscriptions no longer accumulate in Session.Subscriptions. Refs #747 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
2026-10-22 16:21:22 -04:00
parent 7feffc963a
commit 57ae69a05e
5 changed files with 355 additions and 91 deletions
@@ -164,46 +164,16 @@ internal sealed class DefaultSessionAdapter : ISessionAdapter
            ReturnBounds = false
        };

-        var nodesToRead = new HistoryReadValueIdCollection
-        {
-            new HistoryReadValueId { NodeId = nodeId }
-        };
-
-        var allValues = new List<DataValue>();
-        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);
    }

    /// <inheritdoc />
@@ -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<Models.HistoryModifiedEntry> 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);
+        }
+    }
+
+    /// <summary>Issues one single-node HistoryRead (or a release of its continuation point).</summary>
+    private async Task<HistoryReadResult?> 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<Models.HistoryModifiedEntry>();
-        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;
    }

    /// <inheritdoc />
@@ -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));
    }

    /// <inheritdoc />
@@ -12,14 +12,23 @@ internal sealed class DefaultSubscriptionAdapter : ISubscriptionAdapter
    private static readonly ILogger Logger = Log.ForContext<DefaultSubscriptionAdapter>();
    private readonly Dictionary<uint, MonitoredItem> _monitoredItems = new();
    private readonly Subscription _subscription;
+    private readonly Func<Subscription, CancellationToken, Task>? _detach;
+    private int _detached;

    /// <summary>
    /// Wraps a live OPC UA subscription so client code can manage monitored items through a testable abstraction.
    /// </summary>
    /// <param name="subscription">The underlying OPC UA subscription that owns monitored items for this client workflow.</param>
-    public DefaultSubscriptionAdapter(Subscription subscription)
+    /// <param name="detach">
+    ///     Removes the subscription from its owning SDK <c>Session</c> once it is deleted. Without it a deleted
+    ///     subscription stays in <c>Session.Subscriptions</c> until the session is disposed, so a UI that re-targets
+    ///     alarms repeatedly accumulates dead subscriptions the SDK still iterates (transfer / recreate on reconnect).
+    /// </param>
+    public DefaultSubscriptionAdapter(Subscription subscription,
+        Func<Subscription, CancellationToken, Task>? detach = null)
    {
        _subscription = subscription;
+        _detach = detach;
    }

    /// <inheritdoc />
@@ -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");
+        }
+    }
+
    /// <summary>
    /// Releases the wrapped OPC UA subscription and clears tracked monitored items held by the adapter.
    /// </summary>
@@ -129,6 +153,14 @@ internal sealed class DefaultSubscriptionAdapter : ISubscriptionAdapter
        {
        }

+        try
+        {
+            DisposeBridge.Run(DetachAsync);
+        }
+        catch
+        {
+        }
+
        _monitoredItems.Clear();
    }
 }
@@ -0,0 +1,66 @@
+using Opc.Ua;
+using Serilog;
+
+namespace ZB.MOM.WW.OtOpcUa.Client.Shared.Helpers;
+
+/// <summary>
+///     The client-side HistoryRead paging loop shared by the raw and modified reads. Continuation pages are
+///     READ (<c>releaseContinuationPoints = false</c>) 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.
+/// </summary>
+internal static class HistoryReadPaging
+{
+    private static readonly ILogger Logger = Log.ForContext(typeof(HistoryReadPaging));
+
+    /// <summary>Reads (or, with <paramref name="releaseContinuationPoints"/>, releases) one page.</summary>
+    /// <param name="continuationPoint">The cursor to resume, or <see langword="null"/> for the first page.</param>
+    /// <param name="releaseContinuationPoints"><see langword="true"/> to release the cursor instead of reading.</param>
+    /// <param name="ct">The caller's cancellation token.</param>
+    /// <returns>The per-node result, or <see langword="null"/> when the response carried none.</returns>
+    public delegate Task<HistoryReadResult?> PageReader(
+        byte[]? continuationPoint, bool releaseContinuationPoints, CancellationToken ct);
+
+    /// <summary>Reads pages until the server finishes or the caller's cap is reached.</summary>
+    /// <typeparam name="T">The element type extracted from each page.</typeparam>
+    /// <param name="readPage">Issues one HistoryRead for the single node.</param>
+    /// <param name="extract">Extracts the page's elements from a result.</param>
+    /// <param name="maxValues">The caller's cap; <c>&lt;= 0</c> means no cap (follow every continuation point).</param>
+    /// <param name="nodeId">The node being read, for refusal messages.</param>
+    /// <param name="ct">The caller's cancellation token.</param>
+    /// <returns>Every element read.</returns>
+    /// <exception cref="ServiceResultException">A page was refused (see <see cref="HistoryReadStatus"/>).</exception>
+    public static async Task<List<T>> ReadAllAsync<T>(
+        PageReader readPage, Func<HistoryReadResult, IEnumerable<T>> extract, int maxValues, NodeId nodeId,
+        CancellationToken ct)
+    {
+        var all = new List<T>();
+        byte[]? continuationPoint = null;
+
+        while (true)
+        {
+            var result = await readPage(continuationPoint, false, ct).ConfigureAwait(false);
+            if (result == null) return all;
+
+            HistoryReadStatus.ThrowIfRefused(result.StatusCode, nodeId);
+            all.AddRange(extract(result));
+
+            continuationPoint = result.ContinuationPoint;
+            if (continuationPoint is not { Length: > 0 }) return all;
+            if (maxValues > 0 && all.Count >= maxValues) break;
+        }
+
+        // Stopped by the client's own cap: the server still holds a cursor for this session. Release it; failing to
+        // is not worth failing a read whose values are already in hand.
+        try
+        {
+            await readPage(continuationPoint, true, ct).ConfigureAwait(false);
+        }
+        catch (Exception ex) when (ex is not OperationCanceledException)
+        {
+            Logger.Warning(ex, "Releasing the HistoryRead continuation point for {NodeId} failed", nodeId);
+        }
+
+        return all;
+    }
+}
@@ -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;
+
+/// <summary>
+/// A deleted subscription must also leave the SDK session's <c>Subscriptions</c> list, or every alarm re-target in the UI
+/// leaves a dead <see cref="Subscription"/> behind until the session is disposed.
+/// </summary>
+public class DefaultSubscriptionAdapterTests
+{
+    /// <summary>DeleteAsync detaches the wrapped subscription exactly once, even when Dispose follows.</summary>
+    [Fact]
+    public async Task DeleteAsync_detaches_the_subscription_from_its_session_once()
+    {
+        var subscription = new Subscription();
+        var detached = new List<Subscription>();
+        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);
+    }
+
+    /// <summary>Dispose alone (no prior DeleteAsync) detaches too.</summary>
+    [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);
+    }
+
+    /// <summary>A failing detach is logged, not thrown: teardown stays best effort.</summary>
+    [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));
+    }
+}
@@ -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;
+
+/// <summary>
+/// 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.
+/// </summary>
+public class HistoryReadPagingTests
+{
+    private static readonly NodeId Node = new("t", 2);
+
+    /// <summary>One recorded call to the page reader.</summary>
+    private sealed record Call(byte[]? ContinuationPoint, bool Release);
+
+    /// <summary>A scripted server: each read returns the next page; a release returns an empty Good result.</summary>
+    private sealed class ScriptedServer
+    {
+        private readonly Queue<(int Count, byte[]? Cp)> _pages;
+        public List<Call> Calls { get; } = [];
+
+        public ScriptedServer(params (int Count, byte[]? Cp)[] pages) => _pages = new Queue<(int, byte[]?)>(pages);
+
+        public Task<HistoryReadResult?> ReadPageAsync(byte[]? cp, bool release, CancellationToken ct)
+        {
+            Calls.Add(new Call(cp, release));
+            if (release) return Task.FromResult<HistoryReadResult?>(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<HistoryReadResult?>(new HistoryReadResult
+            {
+                StatusCode = StatusCodes.Good,
+                ContinuationPoint = next,
+                HistoryData = new ExtensionObject(data)
+            });
+        }
+    }
+
+    private static IEnumerable<DataValue> Values(HistoryReadResult r) =>
+        r.HistoryData?.Body is HistoryData d ? d.DataValues : [];
+
+    /// <summary>
+    /// 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 <c>release = cp != null</c>) silently truncated every raw read
+    /// to its first server page.
+    /// </summary>
+    [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 });
+    }
+
+    /// <summary>A read stopped by the caller's cap while the server still holds a cursor releases that cursor.</summary>
+    [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 });
+    }
+
+    /// <summary>
+    /// 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.
+    /// </summary>
+    [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();
+    }
+
+    /// <summary>An uncapped read (<c>maxValues &lt;= 0</c>) follows every continuation point to the end.</summary>
+    [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);
+    }
+
+    /// <summary>A failed release is best effort: the values already read are still returned.</summary>
+    [Fact]
+    public async Task A_failed_release_does_not_lose_the_values_read()
+    {
+        var calls = 0;
+        Task<HistoryReadResult?> 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<HistoryReadResult?>(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);
+    }
+
+    /// <summary>A refused page still surfaces as an exception, not an empty read.</summary>
+    [Fact]
+    public async Task A_refused_page_throws()
+    {
+        Task<HistoryReadResult?> Read(byte[]? cp, bool release, CancellationToken ct) =>
+            Task.FromResult<HistoryReadResult?>(new HistoryReadResult
+            {
+                StatusCode = StatusCodes.BadHistoryOperationUnsupported
+            });
+
+        await Should.ThrowAsync<ServiceResultException>(() =>
+            HistoryReadPaging.ReadAllAsync(Read, Values, maxValues: 10, Node, CancellationToken.None));
+    }
+}