Files
lmxopcua/tests/Client/ZB.MOM.WW.OtOpcUa.Client.Shared.Tests/Helpers/HistoryReadPagingTests.cs
T
Joseph Doherty 0c9e4f4e3d 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
2026-10-22 17:18:26 -04:00

220 lines
9.4 KiB
C#

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>
/// 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 <c>(uint)-1</c> on the wire asked the server for everything.
/// </summary>
/// <param name="max">The cap.</param>
[Theory]
[InlineData(0)]
[InlineData(-1)]
public async Task A_non_positive_cap_is_refused_before_any_read(int max)
{
var server = new ScriptedServer();
await Should.ThrowAsync<ArgumentOutOfRangeException>(() =>
HistoryReadPaging.ReadAllAsync(server.ReadPageAsync, Values, max, Node, CancellationToken.None));
server.Calls.ShouldBeEmpty();
}
/// <summary>Review m2: server pages that overshoot the cap are trimmed to it (60 + 60 with a cap of 100 is 100).</summary>
[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(100);
server.Calls.Last().Release.ShouldBeTrue(); // the outstanding cursor is still released
}
/// <summary>
/// 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.
/// </summary>
[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<HistoryReadResult?> Read(byte[]? cp, bool release, CancellationToken ct)
{
calls.Add((cp, release, ct.IsCancellationRequested));
if (release) return Task.FromResult<HistoryReadResult?>(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<HistoryReadResult?>(new HistoryReadResult
{
StatusCode = StatusCodes.Good, ContinuationPoint = [5], HistoryData = new ExtensionObject(data)
});
}
await Should.ThrowAsync<OperationCanceledException>(() =>
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();
}
/// <summary>A cancellation that lands while the cap release is in flight does not lose the values read.</summary>
[Fact]
public async Task The_cap_release_does_not_ride_the_callers_token()
{
using var cts = new CancellationTokenSource();
var releaseSawCancelled = (bool?)null;
Task<HistoryReadResult?> Read(byte[]? cp, bool release, CancellationToken ct)
{
if (release)
{
releaseSawCancelled = ct.IsCancellationRequested;
return Task.FromResult<HistoryReadResult?>(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<HistoryReadResult?>(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);
}
/// <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));
}
}