dohertj2/lmxopcua
1
0
Fork 0
Code Issues 32 Pull Requests Actions Packages Projects Releases Wiki Activity

fix(opcua): reverse HistoryRead-Raw paging no longer stops after the first page

Browse Source 
A reverse read (start > end) with a small NumValuesPerNode resumed through
the tie-cluster path, which drained the cluster and stepped FORWARD to
T + 1 tick - outside a reverse window - so the read ended with GoodNoData
after one sample. SliceTieCluster now steps down to T - 1 tick for a
reverse read (terminal at the excluded end), and the node manager slices
the over-fetched cluster in the read's own direction so the skip count is
not applied from the wrong end. Forward paging is unchanged.

Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
This commit is contained in:
Joseph Doherty
2026-10-09 14:20:00 -04:00
parent 821763a09d
commit 8d51380ff1
4 changed files with 134 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/HistoryPaging.cs
+16 -1
View File
@@ -205,14 +205,29 @@ internal static class HistoryPaging
/// remains, or <c>null</c> when the window is exhausted (terminal — no continuation point).</param>
/// <param name="nextSkip">The next page's boundary skip count: the running emitted-tie total while the
/// cluster drains, else <c>0</c> after advancing past it.</param>
/// <param name="reverse">
/// <c>true</c> for a reverse read (Part 11: <c>start &gt; end</c>, samples in descending time, window
/// <c>(end, start]</c>). A drained cluster then steps DOWN to <c>T − 1 tick</c>, which is lossless for the
/// same reason <c>T + 1 tick</c> is going forward, and the read continues only while that is still strictly
/// after <paramref name="endUtc"/> (the reverse window excludes <c>end</c>; a resume at exactly <c>end</c>
/// would turn into a <c>start == end</c> point read and return the excluded sample). Forward is unchanged.
/// </param>
public static void SliceTieCluster(
int clusterCount, int skip, uint cap, DateTime boundaryT, DateTime endUtc,
out int sliceStart, out int sliceCount, out DateTime? nextStartUtc, out int nextSkip)
out int sliceStart, out int sliceCount, out DateTime? nextStartUtc, out int nextSkip,
bool reverse = false)
{
sliceStart = Math.Min(skip, clusterCount);
sliceCount = Math.Min((int)cap, clusterCount - sliceStart);
var emitted = sliceStart + sliceCount;
if (emitted < clusterCount) { nextStartUtc = boundaryT; nextSkip = emitted; }
else if (reverse)
{
// DateTime.MinValue cannot step down; nothing can lie below it anyway.
var more = boundaryT > DateTime.MinValue && boundaryT.AddTicks(-1) > endUtc;
nextStartUtc = more ? boundaryT.AddTicks(-1) : null;
nextSkip = 0;
}
else
{
var next = boundaryT.AddTicks(1);
src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs
+11 -2
View File
@@ -5898,10 +5898,16 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
// Compute in uint so +1 can never wrap: Math.Max(1,…) also guards against a zero/negative
// config value slipping through (Validate() warns but does not block startup).
var overfetchCap = (uint)Math.Max(1, MaxTieClusterOverfetch) + 1u;
var cluster = (await HistorianDataSource
IReadOnlyList<DataValueSnapshot> cluster = (await HistorianDataSource
.ReadRawAsync(tagname, startUtc, startUtc, overfetchCap, ct)
.ConfigureAwait(false)).Samples;
// A start == end point read returns the ties in the backend's FORWARD order, but a reverse read
// emitted them (and counted boundarySkip) in descending order — the mirror of forward. Slice the
// cluster in the read's own direction, or the skip would index from the wrong end and re-emit ties.
var reverse = startUtc > endUtc;
if (reverse) cluster = cluster.Reverse().ToList();
// Absurd burst: more ties than we're willing to buffer in memory. Preserve today's loud-fail
// for that node rather than over-fetch an unbounded cluster; the operator's remedy is a
// larger ServerHistorian:MaxTieClusterOverfetch (or NumValuesPerNode).
@@ -5916,9 +5922,12 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
return;
}
// Direction comes from the window itself (Part 11: start > end is a reverse read). A resumed reverse
// read keeps start > end, because a reverse resume point is always strictly after endUtc.
HistoryPaging.SliceTieCluster(
cluster.Count, boundarySkip, numValuesPerNode, startUtc, endUtc,
out var sliceStart, out var sliceCount, out var nextStartUtc, out var nextSkip);
out var sliceStart, out var sliceCount, out var nextStartUtc, out var nextSkip,
reverse: reverse);
var slice = new List<DataValueSnapshot>(sliceCount);
for (var i = sliceStart; i < sliceStart + sliceCount; i++) slice.Add(cluster[i]);
tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/HistoryPagingTests.cs
+31
View File
@@ -218,6 +218,37 @@ public sealed class HistoryPagingTests
nextSkip.ShouldBe(0);
}
[Fact]
public void SliceTieCluster_reverse_drain_steps_down_one_tick_while_the_window_remains()
{
// Reverse read (start > end): a drained cluster at T resumes at T-1tick, not T+1tick (which lies outside
// a reverse window and used to end the read after its first page).
var t = new DateTime(2026, 1, 1, 0, 0, 5, DateTimeKind.Utc);
HistoryPaging.SliceTieCluster(
clusterCount: 1, skip: 1, cap: 1, boundaryT: t, endUtc: t.AddHours(-1),
out _, out var sliceCount, out var nextStartUtc, out var nextSkip, reverse: true);
sliceCount.ShouldBe(0);
nextStartUtc.ShouldBe(t.AddTicks(-1));
nextSkip.ShouldBe(0);
}
[Fact]
public void SliceTieCluster_reverse_drain_terminates_when_the_next_tick_is_the_excluded_end()
{
// The reverse window is (end, start]; resuming at exactly end would become a point read of the excluded
// sample, so T-1tick == end is terminal.
var t = new DateTime(2026, 1, 1, 0, 0, 5, DateTimeKind.Utc);
HistoryPaging.SliceTieCluster(
clusterCount: 2, skip: 0, cap: 5, boundaryT: t, endUtc: t.AddTicks(-1),
out _, out var sliceCount, out var nextStartUtc, out _, reverse: true);
sliceCount.ShouldBe(2);
nextStartUtc.ShouldBeNull();
}
[Fact]
public void SliceTieCluster_self_heals_when_skip_exceeds_cluster()
{
tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadPagingTests.cs
+76 -4
View File
@@ -312,6 +312,70 @@ public sealed class NodeManagerHistoryReadPagingTests : IDisposable
await host.DisposeAsync();
}
/// <summary>
/// A reverse read (start &gt; end) with <c>NumValuesPerNode = 1</c> over three distinct-timestamp samples must
/// page through every sample, newest first. Before the fix the second page's resume read re-returned the
/// boundary sample, the trim emptied it, the tie-cluster path drained the one-sample "cluster" and stepped
/// FORWARD to <c>T + 1 tick</c> — past a reverse window's end — so paging stopped after the first sample.
/// </summary>
[Fact]
public async Task Raw_reverse_read_with_one_value_per_page_returns_every_sample_descending()
{
var (host, server) = await BootAsync();
var nm = server.NodeManager!;
nm.HistoryContinuationStore = new InMemoryHistoryContinuationStore();
nm.HistorianDataSource = new SeriesHistorianDataSource(MakeSeries(count: 3, stepSeconds: 1));
nm.EnsureVariable("eq-1/rev", parentFolderNodeId: null, displayName: "Rev", dataType: "Double",
writable: false, historianTagname: "WW.Rev", realm: AddressSpaceRealm.Uns);
var nodeId = nm.TryGetVariable("eq-1/rev")!.NodeId;
var collected = await DrainAsync(nm, nodeId, start: Epoch.AddHours(1), end: Epoch.AddSeconds(-1), max: 1);
collected.ShouldBe([2d, 1d, 0d]);
await host.DisposeAsync();
}
/// <summary>The reverse mirror of <see cref="Raw_oversized_tie_cluster_pages_within_the_timestamp"/>: a 5-way tie
/// cluster between distinct samples, cap 2, read newest-first — every value once, in descending order.</summary>
[Fact]
public async Task Raw_reverse_read_pages_through_an_oversized_tie_cluster()
{
var (host, server) = await BootAsync();
var nm = server.NodeManager!;
nm.HistoryContinuationStore = new InMemoryHistoryContinuationStore();
nm.HistorianDataSource = new SeriesHistorianDataSource(
MakeSeriesWithTieCluster(distinctBefore: 3, tieCount: 5, distinctAfter: 2));
nm.EnsureVariable("eq-1/revburst", parentFolderNodeId: null, displayName: "RevBurst", dataType: "Double",
writable: false, historianTagname: "WW.RevBurst", realm: AddressSpaceRealm.Uns);
var nodeId = nm.TryGetVariable("eq-1/revburst")!.NodeId;
var collected = await DrainAsync(nm, nodeId, start: Epoch.AddHours(1), end: Epoch.AddSeconds(-1), max: 2);
collected.ShouldBe(Enumerable.Range(0, 10).Reverse().Select(i => (double)i));
await host.DisposeAsync();
}
/// <summary>Pages a Raw read to completion and returns the union of the pages in arrival order.</summary>
private static Task<List<double>> DrainAsync(OtOpcUaNodeManager nm, NodeId nodeId, DateTime start, DateTime end, uint max)
{
var collected = new List<double>();
byte[]? cp = null;
var pageCount = 0;
do
{
var (values, error, nextCp) = ReadRaw(nm, nodeId, start, end, max, inboundCp: cp);
error.StatusCode.Code.ShouldBe(StatusCodes.Good);
collected.AddRange(values);
cp = nextCp;
pageCount++;
pageCount.ShouldBeLessThan(40, "paging must terminate, not loop");
}
while (cp is not null);
return Task.FromResult(collected);
}
// --- helpers --------------------------------------------------------------------------------
/// <summary>Issue a single-node Raw HistoryRead and return that node's decoded sample values, error,
@@ -420,10 +484,18 @@ public sealed class NodeManagerHistoryReadPagingTests : IDisposable
CancellationToken cancellationToken)
{
Interlocked.Increment(ref _readCount);
var window = series
.Where(s => (s.SourceTimestampUtc ?? DateTime.MinValue) >= startUtc
&& (s.SourceTimestampUtc ?? DateTime.MinValue) <= endUtc)
.ToList();
// Reverse (Part 11, start > end): (end, start] in DESCENDING time — the SQL backend's contract.
// Ties keep series order reversed, so a reverse page of a cluster is deterministic.
var window = startUtc > endUtc
? series
.Where(s => (s.SourceTimestampUtc ?? DateTime.MinValue) > endUtc
&& (s.SourceTimestampUtc ?? DateTime.MinValue) <= startUtc)
.Reverse()
.ToList()
: series
.Where(s => (s.SourceTimestampUtc ?? DateTime.MinValue) >= startUtc
&& (s.SourceTimestampUtc ?? DateTime.MinValue) <= endUtc)
.ToList();
// maxValuesPerNode == 0 means "no limit".
var page = maxValuesPerNode == 0 ? window : window.Take((int)maxValuesPerNode).ToList();
return Task.FromResult(new HistorianRead(page, null));