From 8dfa165110edad3df1d1b7fa8b5fcccac97ec3b3 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Tue, 13 Oct 2026 00:42:13 -0400 Subject: [PATCH 1/3] =?UTF-8?q?fix(historian):=20#692=20every=20HistoryRea?= =?UTF-8?q?d-Raw=20page=20is=20full=20=E2=80=94=20no=20empty=20page=20betw?= =?UTF-8?q?een=20values?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A resumed raw page asked the backend for only N rows from the boundary timestamp (inclusive), then trimmed the boundary ties already emitted: it netted N - skip values, and at N = 1 nothing — the empty "full" page was taken for a stalled tie cluster and a MaxTieClusterOverfetch + 1 (65,537-row) over-fetch stepped one tick, returning an EMPTY page with a continuation point between every value (8 values = 17 pages / 25 reads), forward and reverse, with or without ReturnBounds, on every backend. Each page's read now asks for N + boundarySkip + 1 rows: the ties it trims, the page and one lookahead row. The lookahead row is the 'more data' signal, so an exact multiple ends on its last full page (no trailing empty page). The over-fetch fires only when every row past the trimmed ties is still on the boundary (a genuine cluster larger than the page); a cursor left inside a cluster is flagged WithinTieCluster and over-fetches directly. A page that never leaves the boundary adds the prior skip to its cursor. Low items: BoundedReadHasAnythingAsync carries its walk hint and, when there is no data, the end bound it found (HistoryEndBound.Resolved + DataExhausted), so the next page reads nothing; HistoryBounds.IsPlaceholder is an explicit marker (ConditionalWeakTable over the NotFound instances) instead of status+null shape, so a stored Bad_BoundNotFound row is data. Tests: NodeManagerHistoryReadPageFillTests (N=1 every shape/direction/bounds, tie clusters straddling boundaries at N=1/2/3, exact-multiple, carried end bound + walk hint, stored placeholder-shaped row). Existing paging tests updated where they encoded the short pages. --- .../HistoryBounds.cs | 36 +- .../HistoryPaging.cs | 8 +- .../OtOpcUaNodeManager.cs | 191 +++++---- .../NodeManagerHistoryReadPageFillTests.cs | 376 ++++++++++++++++++ .../NodeManagerHistoryReadPagingTests.cs | 32 +- .../NodeManagerHistoryReadTests.cs | 2 +- 6 files changed, 554 insertions(+), 91 deletions(-) create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadPageFillTests.cs diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/HistoryBounds.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/HistoryBounds.cs index 2dbf03a7..25bf3a16 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/HistoryBounds.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/HistoryBounds.cs @@ -17,8 +17,13 @@ namespace ZB.MOM.WW.OtOpcUa.OpcUaServer; /// The end time was not given (a one-sided read's sentinel): there is no end bound to look up. /// The data is drained and the end bound did not fit on the previous page. /// The last value's time returned on an earlier page (stamps an open bound's placeholder). +/// +/// Gitea #692: the end bound an earlier page already looked up (a stored sample — never a placeholder), so the page that +/// returns it does not read the backend again. null = not looked up yet. +/// internal sealed record HistoryEndBound( - DateTime AtUtc, bool Reverse, bool Inclusive, bool Open, bool DataExhausted = false, DateTime? LastReturnedUtc = null); + DateTime AtUtc, bool Reverse, bool Inclusive, bool Open, bool DataExhausted = false, DateTime? LastReturnedUtc = null, + DataValueSnapshot? Resolved = null); /// /// Gitea #691: OPC UA Part 11 bounding values for HistoryRead-Raw (ReadRawModifiedDetails.ReturnBounds), composed @@ -59,6 +64,15 @@ internal static class HistoryBounds { private static readonly TimeSpan OpenBoundStep = TimeSpan.FromSeconds(1); + /// + /// Gitea #692: the placeholders this server synthesised (), by reference. A placeholder is known by + /// this marker, never by its shape: a stored row can legitimately carry Bad_BoundNotFound with a null value (a + /// client may write one through HistoryUpdate), and that row is data. + /// + private static readonly System.Runtime.CompilerServices.ConditionalWeakTable Placeholders = new(); + + private static readonly object PlaceholderMark = new(); + /// The end bound owed by a fresh read over the normalised window. /// The (normalised) start: the later time of a reverse read. /// The (normalised) end; a one-sided read's sentinel (tick 0 / MaxValue) is an open bound. @@ -66,11 +80,13 @@ internal static class HistoryBounds public static HistoryEndBound For(DateTime startUtc, DateTime endUtc) => new(endUtc, Reverse: startUtc > endUtc, Inclusive: startUtc != endUtc, Open: IsOpen(endUtc)); - /// Whether a returned value is a placeholder rather than a stored sample. + /// + /// Whether a returned value is a placeholder this server synthesised rather than a stored sample + /// — by an explicit marker (#692), so a stored row whose status happens to be Bad_BoundNotFound is not taken for one. + /// /// The value. - /// true for a Bad_BoundNotFound placeholder. - public static bool IsPlaceholder(DataValueSnapshot value) => - value.StatusCode == StatusCodes.BadBoundNotFound && value.Value is null; + /// true only for an instance produced. + public static bool IsPlaceholder(DataValueSnapshot value) => Placeholders.TryGetValue(value, out _); /// The pending end bound to store behind a continuation point, remembering the page's last returned time. /// The pending end bound (null = bounds not requested). @@ -84,8 +100,12 @@ internal static class HistoryBounds /// The Bad_BoundNotFound placeholder (null value) stamped at . /// The bound's time. /// The placeholder. - public static DataValueSnapshot NotFound(DateTime atUtc) => - new(null, StatusCodes.BadBoundNotFound, atUtc, atUtc); + public static DataValueSnapshot NotFound(DateTime atUtc) + { + var placeholder = new DataValueSnapshot(null, StatusCodes.BadBoundNotFound, atUtc, atUtc); + Placeholders.AddOrUpdate(placeholder, PlaceholderMark); + return placeholder; + } /// /// The start bound to return ahead of the data, or null when a stored sample sits on StartTime (that sample is @@ -123,6 +143,8 @@ internal static class HistoryBounds return NotFound(new DateTime(ticks, DateTimeKind.Utc)); } + if (bound.Resolved is { } resolved) return resolved; // #692: an earlier page already looked it up + DataValueSnapshot? found; if (bound.Reverse) found = await LatestAtOrBeforeAsync(source, tagname, bound.AtUtc, maxTieCluster, nowUtc, ct).ConfigureAwait(false); diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/HistoryPaging.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/HistoryPaging.cs index b618dbf9..28082027 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/HistoryPaging.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/HistoryPaging.cs @@ -66,6 +66,11 @@ namespace ZB.MOM.WW.OtOpcUa.OpcUaServer; /// with ReturnBounds = true only (Gitea #691): the end bound still owed — /// Part 11 returns it with the last page. null = bounds not requested. /// +/// +/// only (Gitea #692): the previous page was a slice of an oversized tie cluster at +/// that still has un-emitted ties — the next page over-fetches the cluster directly +/// instead of first re-reading N + skip + 1 rows that are known to be ties. +/// internal sealed record HistoryContinuationState( string Tagname, DateTime NextStartUtc, @@ -79,7 +84,8 @@ internal sealed record HistoryContinuationState( long SegmentWidthTicks = 0, long EmptyFromTicks = 0, long EmptyToTicks = 0, - HistoryEndBound? PendingEndBound = null) + HistoryEndBound? PendingEndBound = null, + bool WithinTieCluster = false) { /// The walk-back hint this cursor carries (#686/#689/#691). public ReverseWalkHint WalkHint => new(SegmentWidthTicks, EmptyFromTicks, EmptyToTicks); diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs index eb0c81d8..735effdf 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs @@ -533,7 +533,8 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 /// /// The upper bound on the bounded over-fetch uses to page WITHIN an /// oversized "tie cluster" — more raw samples sharing one SourceTimestamp than the client's per-page - /// cap. When a resume read stalls on such a cluster (the boundary-tie trim empties the page), the + /// cap. When a resume read stalls on such a cluster (every row of its N + skip + 1 read, past the + /// trimmed ties, still sits on the boundary timestamp — #692: only a genuine cluster larger than the page), the /// paging over-fetches up to MaxTieClusterOverfetch + 1 ties at that single timestamp (a /// start == end read) and slices through them via . /// A cluster strictly larger than this still fails that node BadTooManyOperations (the absurd-burst @@ -6190,11 +6191,16 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 /// /// Fresh read (no inbound continuation point): read the window from /// details.StartTime to capped at - /// . If the page comes back FULL (exactly the cap, and the - /// cap is > 0), store a resume cursor and emit a continuation point. + /// + 1 — one lookahead row (#692). If the lookahead row came + /// back, return the first N, store a resume cursor and emit a continuation point; otherwise the page is + /// the last and carries none (so a total that is an exact multiple of N ends on a full page, not on an + /// empty trailing one). /// Resume read (inbound continuation point present): take the stored cursor, read - /// the next page from the boundary forward, trim already-emitted boundary ties, and emit a - /// FRESH continuation point only if THIS page is also full — else null (done). + /// N + boundarySkip + 1 rows from the boundary (the ties about to be trimmed, the page, and the + /// lookahead row), trim the already-emitted boundary ties, and continue exactly as a fresh read — so every + /// page carries N values. (Asking for only N used to net N − skip per resumed page, and at N = 1 an EMPTY + /// page between every value: the trimmed page looked like a stalled tie cluster and paid a + /// + 1 over-fetch to step one tick — Gitea #692.) /// /// The resume cursor is tie-safe (see / /// ): the next page resumes from the boundary @@ -6238,6 +6244,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 var boundarySkip = 0; var walkHint = default(ReverseWalkHint); // #686/#689/#691: the walk-back's width + proven-empty stretch, page to page HistoryEndBound? endBound = null; // #691: the end bound still owed (null = bounds not requested) + var withinTieCluster = false; // #692: the previous page sliced an oversized tie cluster that still has ties string tagname; if (inboundCp is { Length: > 0 }) @@ -6260,6 +6267,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 numValuesPerNode = state.NumValuesPerNode; walkHint = state.WalkHint; endBound = state.PendingEndBound; + withinTieCluster = state.WithinTieCluster; } else { @@ -6314,18 +6322,29 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 } // #691 review: whether a bounded read whose start bound was NOT found has any data or a found end bound — only - // asked when a one-value page would otherwise be the start-bound placeholder alone. - async Task BoundedReadHasAnythingAsync() + // asked when a one-value page would otherwise be the start-bound placeholder alone. #692: what it learns rides the + // cursor — the walk hint its walk-back computed, and, when there is no data, the end bound it found (so the next + // page is that bound with no backend read at all). + async Task<(bool Any, HistoryEndBound Pending)> BoundedReadHasAnythingAsync() { var src = HistorianDataSource; - IReadOnlyList first = startUtc > endUtc && !src.SupportsReverseRaw - ? (await ReverseRawReader.ReadAsync(src, tagname, startUtc, endUtc, 1, MaxTieClusterOverfetch, nowUtc, 0, ct) - .ConfigureAwait(false)).Samples - : (await src.ReadRawAsync(tagname, startUtc, endUtc, 1, ct).ConfigureAwait(false)).Samples; - if (first.Any(x => HistoryBounds.InDataWindow(x, endBound!, resolution))) return true; + IReadOnlyList first; + if (startUtc > endUtc && !src.SupportsReverseRaw) + { + var walked = await ReverseRawReader.ReadAsync( + src, tagname, startUtc, endUtc, 1, MaxTieClusterOverfetch, nowUtc, walkHint, ct) + .ConfigureAwait(false); + walkHint = walked.Hint; + first = walked.Samples; + } + else first = (await src.ReadRawAsync(tagname, startUtc, endUtc, 1, ct).ConfigureAwait(false)).Samples; + + if (first.Any(x => HistoryBounds.InDataWindow(x, endBound!, resolution))) return (true, endBound!); var end = await HistoryBounds.EndBoundAsync( src, tagname, endBound!, startUtc, MaxTieClusterOverfetch, nowUtc, ct).ConfigureAwait(false); - return !HistoryBounds.IsPlaceholder(end); + return HistoryBounds.IsPlaceholder(end) + ? (false, endBound!) + : (true, endBound! with { DataExhausted = true, Resolved = end }); } // #691: a resumed bounded read whose data is drained (the end bound did not fit on the previous page, or the @@ -6354,14 +6373,18 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 startBound = await HistoryBounds.StartBoundAsync( HistorianDataSource, tagname, startUtc, endBound.Reverse, MaxTieClusterOverfetch, nowUtc, ct) .ConfigureAwait(false); - if (startBound is not null && numValuesPerNode == 1 - && (!HistoryBounds.IsPlaceholder(startBound) || await BoundedReadHasAnythingAsync().ConfigureAwait(false))) + var (anything, pending) = startBound is not null && numValuesPerNode == 1 + ? HistoryBounds.IsPlaceholder(startBound) + ? await BoundedReadHasAnythingAsync().ConfigureAwait(false) + : (true, endBound) + : (false, endBound); + if (startBound is not null && numValuesPerNode == 1 && anything) { // The page is the start bound alone; the data starts at StartTime on the next page. var boundOnlyCp = _historyContinuationStore.Save(session, new HistoryContinuationState( tagname, startUtc, endUtc, 0, numValuesPerNode, SegmentWidthTicks: walkHint.WidthTicks, EmptyFromTicks: walkHint.EmptyFromTicks, EmptyToTicks: walkHint.EmptyToTicks, - PendingEndBound: HistoryBounds.Carry(endBound, [startBound]))); + PendingEndBound: HistoryBounds.Carry(pending, [startBound]))); results[handle.Index] = new SdkHistoryReadResult { StatusCode = StatusCodes.Good, @@ -6388,60 +6411,78 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 if (startBound is not null && numValuesPerNode > 0) readCap = numValuesPerNode - 1; } + // #692: the page's read asks for the boundary ties it is about to trim (a resumed page re-reads them, because the + // cursor resumes AT the boundary timestamp, inclusive) plus ONE lookahead row past the page. Without them a resumed + // page netted N − skip values, and at N = 1 nothing at all: the lone boundary sample came back, was trimmed, the + // "full" empty page was taken for a stalled tie cluster, and a MaxTieClusterOverfetch + 1 over-fetch stepped one + // tick to return an EMPTY page with a continuation point — between every value. The lookahead row is what says + // "more data follows": a page whose read found no row past it is the last, so the read no longer ends with an + // empty trailing page either. + var resumed = inboundCp is { Length: > 0 }; + var requestCap = readCap == 0 + ? 0u + : (uint)Math.Min(uint.MaxValue, (long)readCap + (resumed ? Math.Max(0, boundarySkip) : 0) + 1L); + // HistoryRead is NOT under the node-manager Lock — awaiting the async source is safe. // A reverse window (explicit, resumed, or the end-only read above) goes to the backend only when it declares it // can read newest-first; otherwise it is walked back from bounded forward reads (#686 — the Gateway rejects // start > end, which failed the node and counted as a backend fault). + // A cursor left inside an oversized tie cluster (#692) skips this read: every row it could return is a tie. var source = HistorianDataSource; - HistorianRead sourceResult; - if (startUtc > endUtc && !source.SupportsReverseRaw) + IReadOnlyList samples = []; + var backendFull = false; + if (!withinTieCluster) { - var page = await ReverseRawReader.ReadAsync( - source, tagname, startUtc, endUtc, readCap, MaxTieClusterOverfetch, - nowUtc, walkHint, ct) - .ConfigureAwait(false); - walkHint = page.Hint; - sourceResult = new HistorianRead(page.Samples, ContinuationPoint: null); - } - else - { - sourceResult = await source - .ReadRawAsync(tagname, startUtc, endUtc, readCap, ct) - .ConfigureAwait(false); - } - - var backendFull = HistoryPaging.IsFullPage(sourceResult.Samples.Count, readCap); - - // On a resume read, drop the boundary ties already returned on the prior page. - var samples = inboundCp is { Length: > 0 } - ? HistoryPaging.TrimBoundaryDuplicates(sourceResult.Samples, startUtc, boundarySkip) - : sourceResult.Samples; - - // #691: a bounded read's data window excludes the end time (a sample there is the end bound). A backend whose - // forward end is inclusive (the gateway's own) returns it as data; dropping it also means the data is drained. - if (endBound is not null) - { - var inside = samples.Where(x => HistoryBounds.InDataWindow(x, endBound, resolution)).ToList(); - if (inside.Count != samples.Count) + HistorianRead sourceResult; + if (startUtc > endUtc && !source.SupportsReverseRaw) { - samples = inside; - backendFull = false; + var page = await ReverseRawReader.ReadAsync( + source, tagname, startUtc, endUtc, requestCap, MaxTieClusterOverfetch, + nowUtc, walkHint, ct) + .ConfigureAwait(false); + walkHint = page.Hint; + sourceResult = new HistorianRead(page.Samples, ContinuationPoint: null); + } + else + { + sourceResult = await source + .ReadRawAsync(tagname, startUtc, endUtc, requestCap, ct) + .ConfigureAwait(false); + } + + backendFull = HistoryPaging.IsFullPage(sourceResult.Samples.Count, requestCap); + + // On a resume read, drop the boundary ties already returned on the prior page. + samples = resumed + ? HistoryPaging.TrimBoundaryDuplicates(sourceResult.Samples, startUtc, boundarySkip) + : sourceResult.Samples; + + // #691: a bounded read's data window excludes the end time (a sample there is the end bound). A backend whose + // forward end is inclusive (the gateway's own) returns it as data; dropping it also means the data is drained. + if (endBound is not null) + { + var inside = samples.Where(x => HistoryBounds.InDataWindow(x, endBound, resolution)).ToList(); + if (inside.Count != samples.Count) + { + samples = inside; + backendFull = false; + } } } - // Oversized tie cluster: a resume read whose FULL backend page does not advance past the resume - // boundary timestamp `startUtc`. That happens when more samples share `startUtc` than the page - // cap — a tie cluster larger than NumValuesPerNode. The fixed-(start,end,cap) backend can only - // ever return the first `cap` of those ties, so the (timestamp, skip) cursor alone can never page - // past (or even reliably within) the cluster: either the boundary-tie trim empties the page, or - // it re-emits the same head ties forever. Detect BOTH stall shapes — an empty trimmed page, or a - // non-empty page whose LAST sample is still at `startUtc` (no forward progress) — and, rather - // than fail the read, over-fetch the WHOLE cluster with an explicit, BOUNDED cap (a start==end - // read at the boundary timestamp) and page WITHIN the timestamp via SliceTieCluster. - var stalledOnTieCluster = inboundCp is { Length: > 0 } - && backendFull - && (samples.Count == 0 - || (samples[^1].SourceTimestampUtc ?? DateTime.MinValue) == startUtc); + // Oversized tie cluster: a resume read whose FULL backend read does not advance past the resume boundary + // timestamp `startUtc` — every row it returned past the trimmed ties, the lookahead row included, still sits on + // `startUtc`. That happens only when more samples share `startUtc` than the page plus the already-emitted ties + // plus one: a genuine tie cluster larger than the page. The fixed-(start,end,cap) backend can only ever return + // the first `cap` of those ties, so growing the read page by page would re-read the cluster quadratically. + // Instead over-fetch the WHOLE cluster with an explicit, BOUNDED cap (a start==end read at the boundary + // timestamp) and page WITHIN the timestamp via SliceTieCluster. (#692: before the read asked for the trimmed ties + // and a lookahead row, a resumed N = 1 page tripped this on every single sample.) + var stalledOnTieCluster = resumed + && (withinTieCluster + || (backendFull + && (samples.Count == 0 + || (samples[^1].SourceTimestampUtc ?? DateTime.MinValue) == startUtc))); if (stalledOnTieCluster) { // The over-fetch cap MUST be explicit and non-zero: a cap of 0 falls back to the backend's @@ -6493,7 +6534,8 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 var clusterState = new HistoryContinuationState( tagname, resumeAt, endUtc, nextSkip, numValuesPerNode, SegmentWidthTicks: walkHint.WidthTicks, EmptyFromTicks: walkHint.EmptyFromTicks, EmptyToTicks: walkHint.EmptyToTicks, - PendingEndBound: HistoryBounds.Carry(endBound, slice)); + PendingEndBound: HistoryBounds.Carry(endBound, slice), + WithinTieCluster: resumeAt == startUtc); clusterCp = _historyContinuationStore.Save(session, clusterState); } else if (endBound is not null) @@ -6512,6 +6554,24 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 return; } + // #692: more than the page came back (the lookahead row, after trimming) ⇒ there is more data: return the page and + // resume after it. Otherwise the data is drained and this is the last data page. + var more = readCap > 0 && samples.Count > readCap; + if (more) samples = samples.Take((int)readCap).ToList(); + + HistoryContinuationState? nextState = null; + if (more) + { + // A page that never left the boundary timestamp (ties continuing a resumed page) adds to the ties already + // emitted there, or the next page would re-emit the earlier ones. + HistoryPaging.ComputeResumeCursor(samples, out var nextStart, out var skip); + if (resumed && nextStart == startUtc) skip += boundarySkip; + nextState = new HistoryContinuationState( + tagname, nextStart, endUtc, skip, numValuesPerNode, SegmentWidthTicks: walkHint.WidthTicks, + EmptyFromTicks: walkHint.EmptyFromTicks, EmptyToTicks: walkHint.EmptyToTicks, + PendingEndBound: HistoryBounds.Carry(endBound, samples)); + } + // #691: the start bound (first page only) leads the data. if (startBound is not null) { @@ -6520,18 +6580,11 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 samples = led; } - // The "full page" test is against the RAW backend count (before trimming): the backend honoured - // the cap, so a full backend page ⇒ there may be more even if we trimmed some boundary ties. var historyData = ToHistoryDataFromSamples(samples); byte[]? outboundCp = null; - if (backendFull && samples.Count > 0) + if (nextState is not null) { - HistoryPaging.ComputeResumeCursor(samples, out var nextStart, out var skip); - var nextState = new HistoryContinuationState( - tagname, nextStart, endUtc, skip, numValuesPerNode, SegmentWidthTicks: walkHint.WidthTicks, - EmptyFromTicks: walkHint.EmptyFromTicks, EmptyToTicks: walkHint.EmptyToTicks, - PendingEndBound: HistoryBounds.Carry(endBound, samples)); // Save may return null (no session on this request) ⇒ degrade to single-shot for this node. // Built AFTER historyData so a failure projecting samples can never orphan a stored cursor. outboundCp = _historyContinuationStore.Save(session, nextState); diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadPageFillTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadPageFillTests.cs new file mode 100644 index 00000000..36c7cf3e --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadPageFillTests.cs @@ -0,0 +1,376 @@ +using Opc.Ua; +using Opc.Ua.Server; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using HistorianRead = ZB.MOM.WW.OtOpcUa.Core.Abstractions.HistoryReadResult; +using SdkHistoryReadResult = Opc.Ua.HistoryReadResult; + +namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests; + +/// +/// Gitea #692: a paged HistoryRead-Raw fills every page. With NumValuesPerNode = 1 a resumed page used to re-read +/// only the boundary sample, trim it, take the empty page for a stalled tie cluster and over-fetch +/// MaxTieClusterOverfetch + 1 rows to step one tick — an EMPTY page (GoodNoData + continuation point) between every +/// value (8 values = 17 pages / 25 backend reads), on every backend, forward and reverse, with or without ReturnBounds. +/// Larger pages lost one value per resumed page the same way. Every page must now carry exactly N values (bounds count), +/// the over-fetch must fire only for a genuine tie cluster, and the read must not end with an empty trailing page when the +/// last data page already knew it was the last. +/// +public sealed class NodeManagerHistoryReadPageFillTests(BoundsServerFixture fx) : IClassFixture +{ + private static readonly DateTime T0 = new(2026, 10, 12, 19, 0, 0, DateTimeKind.Utc); + + public enum Shape { Native, ForwardOnly, Microsecond } + + public enum Direction { Forward, Reverse, EndOnly } + + /// The live-gate repro (#692): 8 values one second apart, read over [19:00:00.1, 19:00:10.5). + private static DataValueSnapshot[] EightValues() => + Enumerable.Range(1, 8).Select(i => Sample(i, T0.AddSeconds(i))).ToArray(); + + private static DataValueSnapshot Sample(double v, DateTime t) => new(v, StatusCodes.Good, t, t); + + private static (DateTime Start, DateTime End) Window(Direction direction) => direction switch + { + Direction.Forward => (T0.AddMilliseconds(100), T0.AddMilliseconds(10_500)), + Direction.Reverse => (T0.AddMilliseconds(10_500), T0.AddMilliseconds(100)), + _ => (DateTime.MinValue, T0.AddMilliseconds(10_500)), + }; + + public static TheoryData AllShapes() + { + var data = new TheoryData(); + foreach (var s in Enum.GetValues()) + foreach (var d in Enum.GetValues()) + foreach (var b in new[] { false, true }) + data.Add(s, d, b); + return data; + } + + /// + /// N = 1 paged to completion: every page carries exactly one value, the pages concatenate to the one-page answer, the + /// read never asks the backend for a tie-cluster over-fetch, and the backend read count stays proportional to the + /// pages (native backends: one data read per page plus the bound lookups). + /// + [Theory] + [MemberData(nameof(AllShapes))] + public async Task One_value_per_page_returns_one_value_on_every_page(Shape shape, Direction direction, bool bounds) + { + var source = Source(shape, EightValues()); + var (start, end) = Window(direction); + + var expected = await SinglePageAsync(source, start, end, bounds); + expected.Count.ShouldBeGreaterThanOrEqualTo(8); + source.ResetCounts(); + + var pages = await DrainPagesAsync(source, start, end, 1, bounds); + + Render(pages.SelectMany(p => p)).ShouldBe(Render(expected)); + pages.Select(p => p.Count).ShouldAllBe(c => c == 1, $"pages: {string.Join(",", pages.Select(p => p.Count))}"); + pages.Count.ShouldBe(expected.Count); + source.MaxCapRequested.ShouldBeLessThanOrEqualTo((uint)(ReverseWalk.DiscoveryRows + 2), + "no tie-cluster over-fetch for a series without ties"); + if (shape != Shape.ForwardOnly) + source.Reads.ShouldBeLessThanOrEqualTo(pages.Count + 4, $"{source.Reads} backend reads for {pages.Count} pages"); + } + + /// The live-gate numbers, pinned: 8 values at N = 1 on a native backend is 8 pages and 8 backend reads (was 17 / 25). + [Fact] + public async Task Eight_values_at_one_per_page_cost_eight_pages_and_eight_reads() + { + var source = Source(Shape.Native, EightValues()); + var (start, end) = Window(Direction.Forward); + + var pages = await DrainPagesAsync(source, start, end, 1, bounds: false); + + pages.Count.ShouldBe(8, $"{source.Reads} reads, {source.RequestedRows} rows requested"); + source.Reads.ShouldBe(8); + source.RequestedRows.ShouldBeLessThanOrEqualTo(8 * 3); + } + + /// A series whose tie clusters straddle the page boundaries at every page size. + private static DataValueSnapshot[] TiedValues() + { + var seconds = new[] { 1, 2, 2, 3, 3, 3, 4, 5, 5, 6, 7, 7, 7, 8 }; + return seconds.Select((s, i) => Sample(i, T0.AddSeconds(s))).ToArray(); + } + + public static TheoryData TiedShapes() + { + var data = new TheoryData(); + foreach (var s in Enum.GetValues()) + foreach (var d in Enum.GetValues()) + foreach (var b in new[] { false, true }) + foreach (var n in new uint[] { 1, 2, 3 }) + data.Add(s, d, b, n); + return data; + } + + /// + /// #669 tie semantics at the page boundary are kept: ties that straddle a page boundary come back once each, in the + /// read's order, at N = 1/2/3 — and every page that has a continuation point is full, no page is empty. + /// + [Theory] + [MemberData(nameof(TiedShapes))] + public async Task Tie_clusters_straddling_page_boundaries_page_without_loss_duplication_or_short_pages( + Shape shape, Direction direction, bool bounds, uint n) + { + var source = Source(shape, TiedValues()); + var (start, end) = Window(direction); + var expected = await SinglePageAsync(source, start, end, bounds); + + var pages = await DrainPagesAsync(source, start, end, n, bounds); + + Render(pages.SelectMany(p => p)).ShouldBe(Render(expected)); + pages.ShouldAllBe(p => p.Count > 0, "no empty page"); + for (var i = 0; i < pages.Count - 1; i++) + pages[i].Count.ShouldBe((int)n, $"page {i} of {pages.Count}: {string.Join(",", pages.Select(p => p.Count))}"); + } + + /// + /// A read whose total is an exact multiple of the page size ends on its last full page: the lookahead row tells the + /// server there is nothing more, so there is no trailing 0-value page (the #692 gate's reverse N = 2 observation). + /// + [Theory] + [InlineData(Shape.Native, Direction.Forward)] + [InlineData(Shape.Native, Direction.Reverse)] + [InlineData(Shape.ForwardOnly, Direction.Forward)] + [InlineData(Shape.ForwardOnly, Direction.Reverse)] + [InlineData(Shape.Microsecond, Direction.Reverse)] + public async Task An_exact_multiple_ends_on_its_last_full_page(Shape shape, Direction direction) + { + var (start, end) = Window(direction); + var pages = await DrainPagesAsync(Source(shape, EightValues()), start, end, 2, bounds: false); + + pages.Select(p => p.Count).ShouldBe([2, 2, 2, 2]); + } + + /// + /// #692 (low): a one-value bounded read whose start bound is not found asks whether the read holds anything. When the + /// answer is "only the end bound", the bound it just found rides the continuation point — the next page is served + /// without reading the backend again. + /// + [Theory] + [InlineData(Shape.Native, Direction.Forward)] + [InlineData(Shape.Native, Direction.Reverse)] + [InlineData(Shape.ForwardOnly, Direction.Forward)] + [InlineData(Shape.ForwardOnly, Direction.Reverse)] + [InlineData(Shape.Microsecond, Direction.Forward)] + public async Task A_start_bound_only_page_carries_the_end_bound_it_found(Shape shape, Direction direction) + { + // Forward [5s, 7s): nothing at/before 5s, nothing inside, the end bound at 9s. Reverse (5s ← 7s]: mirrored. + var series = direction == Direction.Forward ? new[] { Sample(9, T0.AddSeconds(9)) } : [Sample(3, T0.AddSeconds(3))]; + var source = Source(shape, series); + var (start, end) = direction == Direction.Forward ? (T0.AddSeconds(5), T0.AddSeconds(7)) : (T0.AddSeconds(7), T0.AddSeconds(5)); + + var (first, _, cp) = await ReadAsync(source, start, end, 1, bounds: true, cp: null); + first.Single().StatusCode.Code.ShouldBe(StatusCodes.BadBoundNotFound); + cp.ShouldNotBeNull(); + + source.ResetCounts(); + var (second, _, cp2) = await ReadAsync(source, start, end, 1, bounds: true, cp: cp); + second.Single().Value.ShouldBe(direction == Direction.Forward ? 9d : 3d); + cp2.ShouldBeNull(); + source.Reads.ShouldBe(0, "the end bound was found by the first page and carried"); + } + + /// + /// #692 (low): a one-value bounded reverse read on a forward-only backend walks back to answer "is there any data"; + /// the walk hint that walk computed rides the continuation point (it used to be dropped). + /// + [Fact] + public async Task A_start_bound_only_page_carries_the_walk_hint_it_computed() + { + var source = Source(Shape.ForwardOnly, EightValues()); + var store = new RecordingStore(); + await fx.Gate.WaitAsync(TestContext.Current.CancellationToken); + try + { + fx.Nm.HistoryContinuationStore = store; + fx.Nm.HistorianDataSource = source; + // Reverse (0.1s ← 30s]: no sample at or after 30s ⇒ the start bound is a placeholder; the data is below. + var (values, _, cp) = Read(T0.AddSeconds(30), T0.AddMilliseconds(100), 1, bounds: true, cp: null); + values.Single().StatusCode.Code.ShouldBe(StatusCodes.BadBoundNotFound); + cp.ShouldNotBeNull(); + store.Saved.Last().WalkHint.WidthTicks.ShouldBeGreaterThan(0); + } + finally + { + fx.Nm.HistoryContinuationStore = new InMemoryHistoryContinuationStore(); + fx.Gate.Release(); + } + } + + /// + /// #692 (low): only a SERVER-synthesised bound is a placeholder. A stored row that happens to carry + /// Bad_BoundNotFound with a null value (a client can write one through HistoryUpdate) is data — it used to make + /// the read look "all placeholders" and come back GoodNoData with no values. + /// + [Fact] + public async Task A_stored_row_with_the_bound_not_found_status_is_data_not_a_placeholder() + { + var stored = new DataValueSnapshot(null, StatusCodes.BadBoundNotFound, T0.AddSeconds(6), T0.AddSeconds(6)); + HistoryBounds.IsPlaceholder(stored).ShouldBeFalse(); + HistoryBounds.IsPlaceholder(HistoryBounds.NotFound(T0)).ShouldBeTrue(); + + var (values, _, cp) = await ReadAsync(Source(Shape.Native, [stored]), T0.AddSeconds(5), T0.AddSeconds(7), 0, bounds: true, cp: null); + cp.ShouldBeNull(); + values.Count.ShouldBe(3, "start placeholder, the stored row, end placeholder"); + values[1].SourceTimestamp.ShouldBe(T0.AddSeconds(6)); + } + + // --- helpers -------------------------------------------------------------------------------- + + private static string Render(IEnumerable values) => + string.Join(" | ", values.Select(v => $"{v.SourceTimestamp:HH:mm:ss.fffffff}={v.Value ?? "null"}/{v.StatusCode.Code:X8}")); + + private async Task> SinglePageAsync(CountingSource source, DateTime start, DateTime end, bool bounds) + { + var (values, error, cp) = await ReadAsync(source, start, end, 10_000, bounds, cp: null); + error.StatusCode.Code.ShouldBe(StatusCodes.Good); + cp.ShouldBeNull(); + return values.ToList(); + } + + private async Task>> DrainPagesAsync( + CountingSource source, DateTime start, DateTime end, uint max, bool bounds) + { + var pages = new List>(); + byte[]? cp = null; + do + { + var (values, error, next) = await ReadAsync(source, start, end, max, bounds, cp); + error.StatusCode.Code.ShouldBe(StatusCodes.Good); + pages.Add(values); + cp = next; + pages.Count.ShouldBeLessThan(80, "paging must terminate"); + } + while (cp is not null); + + return pages; + } + + private async Task<(IReadOnlyList Values, ServiceResult Error, byte[]? Cp)> ReadAsync( + IHistorianDataSource source, DateTime start, DateTime end, uint max, bool bounds, byte[]? cp) + { + await fx.Gate.WaitAsync(TestContext.Current.CancellationToken); + try + { + fx.Nm.HistorianDataSource = source; + return Read(start, end, max, bounds, cp); + } + finally { fx.Gate.Release(); } + } + + private (IReadOnlyList Values, ServiceResult Error, byte[]? Cp) Read( + DateTime start, DateTime end, uint max, bool bounds, byte[]? cp) + { + var context = new OperationContext(new RequestHeader(), secureChannelContext: null, RequestType.HistoryRead, identity: null); + var details = new ReadRawModifiedDetails + { + StartTime = start, EndTime = end, NumValuesPerNode = max, ReturnBounds = bounds, IsReadModified = false, + }; + var nodesToRead = new List { new() { NodeId = fx.NodeId, ContinuationPoint = cp } }; + var results = new List { null! }; + var errors = new List { null! }; + fx.Nm.HistoryRead(context, details, TimestampsToReturn.Both, releaseContinuationPoints: false, nodesToRead, results, errors); + + var values = new List(); + if (results[0]?.HistoryData is { } ho && !ExtensionObject.IsNull(ho)) + values.AddRange(((HistoryData)ExtensionObject.ToEncodeable(ho)).DataValues); + return (values, errors[0], results[0]?.ContinuationPoint); + } + + private static CountingSource Source(Shape shape, IReadOnlyList series) => shape switch + { + Shape.Native => new CountingSource(series, reverse: true, resolution: 1), + Shape.ForwardOnly => new CountingSource(series, reverse: false, resolution: 1), + _ => new CountingSource(series, reverse: true, resolution: 10), + }; + + /// Records every saved cursor (to inspect what rides a continuation point). + private sealed class RecordingStore : IHistoryContinuationStore + { + private readonly InMemoryHistoryContinuationStore _inner = new(); + public List Saved { get; } = []; + + public byte[]? Save(ISession? session, HistoryContinuationState state) + { + Saved.Add(state); + return _inner.Save(session, state); + } + + public HistoryContinuationState? TryTake(ISession? session, byte[] continuationPoint) => _inner.TryTake(session, continuationPoint); + + public void Release(ISession? session, byte[] continuationPoint) => _inner.Release(session, continuationPoint); + } + + /// + /// A series-backed historian that counts what it is asked for. reverse: serves start > end newest-first + /// (SqlServer/QuestDB); otherwise refuses it (the Gateway — the server walks back with forward reads). Forward reads + /// are [start, end), a point read the instant; resolution > 1 truncates the window bounds (QuestDB µs). + /// + private sealed class CountingSource(IReadOnlyList series, bool reverse, long resolution) : IHistorianDataSource + { + private readonly Lock _gate = new(); + + public int Reads { get; private set; } + public long RequestedRows { get; private set; } + public uint MaxCapRequested { get; private set; } + + public void ResetCounts() + { + lock (_gate) { Reads = 0; RequestedRows = 0; MaxCapRequested = 0; } + } + + public bool SupportsReverseRaw => reverse; + + public long TimestampResolutionTicks => resolution; + + private DateTime Trunc(DateTime t) => new(t.Ticks - t.Ticks % resolution, DateTimeKind.Utc); + + public Task ReadRawAsync( + string fullReference, DateTime startUtc, DateTime endUtc, uint maxValuesPerNode, CancellationToken cancellationToken) + { + lock (_gate) + { + Reads++; + RequestedRows += maxValuesPerNode; + if (maxValuesPerNode > MaxCapRequested) MaxCapRequested = maxValuesPerNode; + } + + static DateTime Ts(DataValueSnapshot s) => s.SourceTimestampUtc ?? DateTime.MinValue; + startUtc = Trunc(startUtc); + endUtc = Trunc(endUtc); + List window; + if (startUtc > endUtc) + { + if (!reverse) throw new InvalidOperationException("start_time must be less than or equal to end_time. (InvalidArgument)"); + window = series.Where(s => Ts(s) > endUtc && Ts(s) <= startUtc).Reverse().ToList(); + } + else if (startUtc == endUtc) window = series.Where(s => Ts(s) == startUtc).ToList(); + else window = series.Where(s => Ts(s) >= startUtc && Ts(s) < endUtc).ToList(); + + var page = maxValuesPerNode == 0 ? window : window.Take((int)maxValuesPerNode).ToList(); + return Task.FromResult(new HistorianRead(page, null)); + } + + public Task ReadProcessedAsync( + string fullReference, DateTime startUtc, DateTime endUtc, TimeSpan interval, + HistoryAggregateType aggregate, CancellationToken cancellationToken) => + Task.FromResult(new HistorianRead(Array.Empty(), null)); + + public Task ReadAtTimeAsync( + string fullReference, IReadOnlyList timestampsUtc, CancellationToken cancellationToken) => + Task.FromResult(new HistorianRead(Array.Empty(), null)); + + public Task ReadEventsAsync( + string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents, CancellationToken cancellationToken) => + Task.FromResult(new HistoricalEventsResult(Array.Empty(), null)); + + public HistorianHealthSnapshot GetHealthSnapshot() => NullHistorianDataSource.Instance.GetHealthSnapshot(); + + public void Dispose() { } + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadPagingTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadPagingTests.cs index b881d737..4a471f56 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadPagingTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadPagingTests.cs @@ -29,8 +29,8 @@ public sealed class NodeManagerHistoryReadPagingTests : IDisposable /// A full first page emits a continuation point; resuming returns subsequent pages; the /// final short page clears the continuation point — and the union of pages is the complete series with - /// no duplicate or skipped sample. (Resumed pages net slightly fewer than the cap because each resume - /// re-reads + trims the boundary sample to stay tie-safe — see the design; the union is what matters.) + /// no duplicate or skipped sample. (#692: every page is full — a resume re-reads the boundary ties it trims plus one + /// lookahead row, so 250 at 100 is exactly three pages of 100, 100 and 50.) [Fact] public async Task Raw_pages_full_series_across_continuation_points() { @@ -64,7 +64,7 @@ public sealed class NodeManagerHistoryReadPagingTests : IDisposable while (cp is not null); sawCp.ShouldBeTrue("a 250-sample series capped at 100 must page"); - pageCount.ShouldBeGreaterThan(1); + pageCount.ShouldBe(3, "100 + 100 + 50 — every page full (#692)"); // The union is the exact series — no duplicates, no skips, in order. collected.Count.ShouldBe(250); @@ -132,11 +132,11 @@ public sealed class NodeManagerHistoryReadPagingTests : IDisposable await host.DisposeAsync(); } - /// An exactly-full series whose total is a multiple of the cap pages to a trailing - /// EMPTY page (GoodNoData, no CP) — the server can't know the prior page was the last until it reads - /// past the end. + /// An exactly-full series whose total is a multiple of the cap ends on its last full page: each read asks for + /// one lookahead row past the page (#692), so the server knows the page is the last and emits no continuation point — + /// there is no trailing EMPTY page any more. A stale point resumed anyway still terminates with GoodNoData. [Fact] - public async Task Raw_exact_multiple_terminates_with_empty_final_page() + public async Task Raw_exact_multiple_ends_on_its_last_full_page() { var (host, server) = await BootAsync(); var nm = server.NodeManager!; @@ -149,12 +149,17 @@ public sealed class NodeManagerHistoryReadPagingTests : IDisposable var (r1, _, cp1) = ReadRaw(nm, nodeId, Epoch, Epoch.AddHours(1), max: 100, inboundCp: null); r1.Count.ShouldBe(100); - cp1.ShouldNotBeNull("a full page emits a CP even when it happens to be the last"); + cp1.ShouldBeNull("the lookahead row found nothing past the page — it is the last"); - var (r2, e2, cp2) = ReadRaw(nm, nodeId, Epoch, Epoch.AddHours(1), max: 100, inboundCp: cp1); - r2.Count.ShouldBe(0); + // Positive control: one more sample past the page ⇒ the same full page DOES carry a continuation point. + nm.HistorianDataSource = new SeriesHistorianDataSource(MakeSeries(count: 101, stepSeconds: 1)); + var (r2, e2, cp2) = ReadRaw(nm, nodeId, Epoch, Epoch.AddHours(1), max: 100, inboundCp: null); + r2.Count.ShouldBe(100); e2.StatusCode.Code.ShouldBe(StatusCodes.Good); - cp2.ShouldBeNull("the empty trailing page terminates paging"); + cp2.ShouldNotBeNull(); + var (r3, _, cp3) = ReadRaw(nm, nodeId, Epoch, Epoch.AddHours(1), max: 100, inboundCp: cp2); + r3.ShouldBe([100d]); + cp3.ShouldBeNull(); await host.DisposeAsync(); } @@ -675,8 +680,9 @@ public sealed class NodeManagerHistoryReadPagingTests : IDisposable var (values, error, next) = ReadRaw(nm, nodeId, DateTime.MinValue, end, max: 3, inboundCp: cp); error.StatusCode.Code.ShouldBe(StatusCodes.Good); values.ShouldNotBeEmpty(); - values[^1].ShouldBe((double)(expected - 2), $"page {page} continues exactly"); - expected = (int)values[^1]; + // #692: every page is full (N = 3) — a resumed page used to net N − 1 after trimming its boundary sample. + values.ShouldBe([expected, expected - 1d, expected - 2d], $"page {page} continues exactly"); + expected -= 3; next.ShouldNotBeNull(); cp = next; diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadTests.cs index a3c8a266..0ecffbe1 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadTests.cs @@ -66,7 +66,7 @@ public sealed class NodeManagerHistoryReadTests : IDisposable fake.LastTagname.ShouldBe("WW.Temp"); fake.LastStart.ShouldBe(start); fake.LastEnd.ShouldBe(end); - fake.LastMaxValues.ShouldBe(100u); + fake.LastMaxValues.ShouldBe(101u, "the page plus one lookahead row (#692) — it tells the server whether more follows"); errors[0].StatusCode.Code.ShouldBe(StatusCodes.Good); results[0].StatusCode.Code.ShouldBe(StatusCodes.Good); From 3d7eea6280a5ac452e716ae58edd624927b24fa6 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Tue, 13 Oct 2026 00:42:13 -0400 Subject: [PATCH 2/3] docs(historian): #692 raw paging fills every page; lookahead row; marked placeholders --- docs/Historian.md | 32 +++++++++++++++++++++++++------- 1 file changed, 25 insertions(+), 7 deletions(-) diff --git a/docs/Historian.md b/docs/Historian.md index dbebd16e..97667113 100644 --- a/docs/Historian.md +++ b/docs/Historian.md @@ -582,23 +582,35 @@ gateway is provisioned, without producing error spikes in connected clients. `NumValuesPerNode` samples with no continuation point of its own), so the server synthesises paging time-based: -- A page that returns **exactly** `NumValuesPerNode` samples (with `NumValuesPerNode > 0`) MAY - have more behind it, so the server stores a resume cursor and returns an opaque continuation +- Every page's backend read asks for **one lookahead row** past the page (`NumValuesPerNode + 1`, plus the boundary + ties a resumed page is about to trim — see below). When the lookahead row comes back there is more data: the page + returns exactly `NumValuesPerNode` values, and the server stores a resume cursor and returns an opaque continuation point (16 bytes). The client hands it back to fetch the next page. -- A short page (fewer than the cap) is the last page — no continuation point. +- When it does not, the page is the **last** — no continuation point, even when it is exactly full. (#692: a total that + is an exact multiple of the cap used to end with an extra empty `GoodNoData` page.) - `NumValuesPerNode == 0` ("all values, no limit") is never paged; the whole window returns in one shot. - The resume cursor is **tie-safe**: the next page resumes from the last returned sample's SourceTimestamp *inclusive* and drops the boundary samples already emitted, so samples sharing - the boundary timestamp are neither duplicated nor skipped. + the boundary timestamp are neither duplicated nor skipped. Because it re-reads them, a resumed page asks for + `NumValuesPerNode + skip + 1` rows, so it still carries a full page. +- ⚠️ **#692 (client-visible): every page is full.** A resumed page used to ask for only `NumValuesPerNode` rows, so after + the trim it netted `N − skip` values — and at `NumValuesPerNode = 1` **nothing**: the lone re-read boundary sample was + trimmed, the empty "full" page was taken for a stalled tie cluster, and a `MaxTieClusterOverfetch + 1` (65,537-row) + over-fetch stepped one tick to return an **empty page with a continuation point between every value** (live, AVEVA: + 8 values = 17 pages / 25 backend reads; on every backend, forward and reverse, with or without `ReturnBounds`). Now 8 + values at N = 1 is 8 pages and 8 reads. Clients that drained to the end were unaffected in content, only in cost; a + client that stopped at the first empty page lost data. > **Oversized tie clusters — within-timestamp paging.** When more samples share one > `SourceTimestamp` than the current page cap, the server detects that the cursor has stalled on -> a tie cluster (the last returned timestamp equals the resume timestamp). It then **over-fetches +> a tie cluster (every row of the resumed `N + skip + 1` read past the trimmed ties — the lookahead row included — still +> sits on the resume timestamp; since #692 nothing else trips it). It then **over-fetches > the entire cluster** at that single timestamp up to a bounded ceiling controlled by > `ServerHistorian:MaxTieClusterOverfetch` (default **65 536**), then serves the cluster > `NumValuesPerNode` samples at a time across successive pages, advancing the cursor one tick past -> the timestamp once the cluster is fully drained. +> the timestamp once the cluster is fully drained. A cursor left inside the cluster is marked so (`WithinTieCluster`), and +> its next page over-fetches the cluster directly instead of first re-reading rows it knows are ties. > > **Reverse reads** (`start > end`, newest first) page the same way in the opposite direction: a drained > cluster steps the cursor one tick **down**, the read ends when that would reach the (excluded) `end`, and @@ -792,7 +804,13 @@ switch `ServerHistorian:HonorReturnBounds = false` restores the pre-#691 behavio - **Counting and paging.** Bounds **count toward `NumValuesPerNode`** (§6.5.3.2: "any Bounding Values returned are included in the numValuesPerNode count"; `NumValuesPerNode = 1` returns the start bound alone). The start bound leads the **first** page and the end bound ends the **last**: the pending end bound rides the continuation point, and when a page is exactly full the end bound is - the whole next page. Paging a bounded read at any page size concatenates to the unpaged answer. + the whole next page. Paging a bounded read at any page size concatenates to the unpaged answer. At `NumValuesPerNode = 1` + with no start bound found, the server first checks the read holds anything (else `GoodNoData`); since #692 what that check + learns rides the continuation point — the walk-back hint and, when there is no data, the end bound it found — so the next + page reads nothing it already read. +- **Placeholders are marked, not recognised by shape (#692).** Only a `Bad_BoundNotFound` value the server synthesised is a + placeholder; a stored row that carries `Bad_BoundNotFound` with a null value (writable through HistoryUpdate) is data, and + never turns a read into the "no data and no bound" `GoodNoData` answer. - **How it is served — one authority, `HistoryBounds` in the OPC UA server.** "At or before T" is a newest-first read of `(0, T]` capped at one — native on SqlServer/QuestDB, the #686 walk-back of bounded forward reads on the Gateway (a tag with no sample before StartTime is proven empty down to 1970 by the #689 walk — after two empty segments it asks once for the whole remainder — From e04e1bf127e147103cba68bd6dd3c84213790c33 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Tue, 13 Oct 2026 01:02:12 -0400 Subject: [PATCH 3/3] =?UTF-8?q?fix(historian):=20#692=20review=20=E2=80=94?= =?UTF-8?q?=20a=20drained=20tie=20cluster=20never=20resumes=20onto=20the?= =?UTF-8?q?=20exclusive=20EndTime?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit F1 (pre-existing): SliceTieCluster continued a forward read while next <= endUtc, but the forward window is end-exclusive; an oversized cluster one step below EndTime resumed AT EndTime, the resume became a start == end point read and returned the EndTime samples as data (microsecond: EndTime 8.1 us truncated to 8.0). Continue only while next is strictly before EndTime truncated to the backend resolution. Tests: A_tie_cluster_just_below_the_end_time_never_resumes_onto_the_end_time (native + microsecond; both fail on the old rule) and A_cursor_inside_a_tie_cluster_over_fetches_without_a_normal_read_first (pins the WithinTieCluster shortcut by per-page read count; fails with the shortcut disabled). Docs: the oversized-cluster exception to 'every page full / no lookahead => last' (F2), and the N + skip + 1 strict-ceiling edge note referencing #694 (F4). --- docs/Historian.md | 17 +++-- .../HistoryPaging.cs | 15 ++--- .../NodeManagerHistoryReadPageFillTests.cs | 62 +++++++++++++++++++ 3 files changed, 83 insertions(+), 11 deletions(-) diff --git a/docs/Historian.md b/docs/Historian.md index 97667113..78a29031 100644 --- a/docs/Historian.md +++ b/docs/Historian.md @@ -587,14 +587,20 @@ paging time-based: returns exactly `NumValuesPerNode` values, and the server stores a resume cursor and returns an opaque continuation point (16 bytes). The client hands it back to fetch the next page. - When it does not, the page is the **last** — no continuation point, even when it is exactly full. (#692: a total that - is an exact multiple of the cap used to end with an extra empty `GoodNoData` page.) + is an exact multiple of the cap used to end with an extra empty `GoodNoData` page.) The exception is an oversized tie + cluster (below): its pages are sliced from the over-fetched cluster, its last slice is `min(N, remaining)` values and + still carries a continuation point when the window goes on — so stepping past a cluster at the very end of the data can + yield one empty final page. - `NumValuesPerNode == 0` ("all values, no limit") is never paged; the whole window returns in one shot. - The resume cursor is **tie-safe**: the next page resumes from the last returned sample's SourceTimestamp *inclusive* and drops the boundary samples already emitted, so samples sharing the boundary timestamp are neither duplicated nor skipped. Because it re-reads them, a resumed page asks for - `NumValuesPerNode + skip + 1` rows, so it still carries a full page. -- ⚠️ **#692 (client-visible): every page is full.** A resumed page used to ask for only `NumValuesPerNode` rows, so after + `NumValuesPerNode + skip + 1` rows, so it still carries a full page. On a strict Gateway + (`AllowStreamingBeyondCeiling=false`) that moves the row-ceiling edge by `skip + 1`: a `NumValuesPerNode` just under the + learned ceiling may now cost one learning refusal (#691.3 re-asks at the ceiling). A page clamp to the learned ceiling is + being added separately (#694). +- ⚠️ **#692 (client-visible): every page is full** (outside an oversized tie cluster — see above). A resumed page used to ask for only `NumValuesPerNode` rows, so after the trim it netted `N − skip` values — and at `NumValuesPerNode = 1` **nothing**: the lone re-read boundary sample was trimmed, the empty "full" page was taken for a stalled tie cluster, and a `MaxTieClusterOverfetch + 1` (65,537-row) over-fetch stepped one tick to return an **empty page with a continuation point between every value** (live, AVEVA: @@ -610,7 +616,10 @@ paging time-based: > `ServerHistorian:MaxTieClusterOverfetch` (default **65 536**), then serves the cluster > `NumValuesPerNode` samples at a time across successive pages, advancing the cursor one tick past > the timestamp once the cluster is fully drained. A cursor left inside the cluster is marked so (`WithinTieCluster`), and -> its next page over-fetches the cluster directly instead of first re-reading rows it knows are ties. +> its next page over-fetches the cluster directly instead of first re-reading rows it knows are ties. A drained cluster +> resumes only while the next step is strictly before `EndTime` (truncated to the backend's resolution) — the forward +> window is end-exclusive; resuming AT `EndTime` used to make the next read a point read that returned the samples at +> `EndTime` as data (#692 review). > > **Reverse reads** (`start > end`, newest first) page the same way in the opposite direction: a drained > cluster steps the cursor one tick **down**, the read ends when that would reach the (excluded) `end`, and diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/HistoryPaging.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/HistoryPaging.cs index 28082027..c16ba97b 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/HistoryPaging.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/HistoryPaging.cs @@ -258,12 +258,11 @@ internal static class HistoryPaging /// The client's per-page cap (NumValuesPerNode); must be > 0 to make progress. /// The single SourceTimestamp every tie in the cluster shares. /// The (inclusive) upper bound of the read window; unchanged across pages. - /// OpcUaServer-003, accepted by design: the backend contract - /// (IHistorianDataSource.ReadRawAsync) treats endUtc as EXCLUSIVE, so the - /// next <= endUtc test here is deliberately the conservative side — at worst one extra - /// empty resume page ([endUtc, endUtc) → GoodNoData, no continuation point), never a - /// dropped or duplicated sample. Tightening to < would be correct only if the live - /// backend's boundary is confirmed exclusive; it is not worth the opposite off-by-one risk. + /// The forward window is end-EXCLUSIVE (Part 11; the IHistorianDataSource.ReadRawAsync contract), so a drained + /// cluster continues only while next is strictly before truncated to + /// . (OpcUaServer-003 had kept next <= endUtc as "the conservative side", + /// but it was not: resuming AT EndTime makes the next read a start == end point read, which returns the samples + /// at EndTime as data — #692 review F1; on a microsecond backend EndTime is truncated first.) /// The index into the cluster the emitted slice starts at (clamped to the count). /// How many ties the emitted slice contains (may be 0 on a stale-skip self-heal). /// The next page's inclusive start: while the @@ -306,7 +305,9 @@ internal static class HistoryPaging else { var next = boundaryT.AddTicks(resolutionTicks); - if (next <= endUtc) { nextStartUtc = next; nextSkip = 0; } + // The forward window is end-EXCLUSIVE, at the backend's resolution: resuming AT EndTime would turn the next read + // into a start == end point read, which returns the samples at EndTime as data (#692 review F1). + if (next.Ticks < endUtc.Ticks - endUtc.Ticks % resolutionTicks) { nextStartUtc = next; nextSkip = 0; } else { nextStartUtc = null; nextSkip = 0; } } } diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadPageFillTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadPageFillTests.cs index 36c7cf3e..87ae0cde 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadPageFillTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadPageFillTests.cs @@ -146,6 +146,68 @@ public sealed class NodeManagerHistoryReadPageFillTests(BoundsServerFixture fx) pages.Select(p => p.Count).ShouldBe([2, 2, 2, 2]); } + /// + /// #692 review F1: an oversized tie cluster one step below EndTime used to resume AT EndTime (the drain check was + /// next <= end, but the forward window is end-exclusive), so the next read was a start == end point + /// read that returned the samples AT EndTime as data. Paged must equal unpaged — neither holds the EndTime samples. + /// The microsecond shape truncates EndTime (8.1 µs ⇒ 8.0 µs) before the same thing happens. + /// + [Theory] + [InlineData(Shape.Native, 1L, 8L)] + [InlineData(Shape.Microsecond, 10L, 81L)] + public async Task A_tie_cluster_just_below_the_end_time_never_resumes_onto_the_end_time(Shape shape, long unit, long endTicks) + { + var series = new List(); + var v = 0; + foreach (var (at, count) in new[] { (5L, 5), (6L, 2), (7L, 3), (8L, 3) }) + for (var i = 0; i < count; i++) + series.Add(Sample(v++, T0.AddTicks(at * unit))); + var source = Source(shape, series); + var (start, end) = (T0.AddSeconds(-3), T0.AddTicks(endTicks)); + + var unpaged = await SinglePageAsync(source, start, end, bounds: false); + unpaged.Count.ShouldBe(10, "the samples at the (exclusive) end time are not data"); + + var pages = await DrainPagesAsync(source, start, end, 1, bounds: false); + Render(pages.SelectMany(p => p)).ShouldBe(Render(unpaged)); + } + + /// + /// #692: a cursor left inside an oversized tie cluster is flagged, and its next page goes straight to the cluster + /// over-fetch — one backend read — instead of first re-reading N + skip + 1 rows it knows are ties. + /// + [Fact] + public async Task A_cursor_inside_a_tie_cluster_over_fetches_without_a_normal_read_first() + { + // a @1s, five ties @2s, b @3s; N = 1. + var series = new List { Sample(0, T0.AddSeconds(1)) }; + for (var i = 1; i <= 5; i++) series.Add(Sample(i, T0.AddSeconds(2))); + series.Add(Sample(6, T0.AddSeconds(3))); + var source = Source(Shape.Native, series); + var (start, end) = (T0, T0.AddSeconds(10)); + + var readsPerPage = new List(); + var all = new List(); + byte[]? cp = null; + do + { + source.ResetCounts(); + var (values, error, next) = await ReadAsync(source, start, end, 1, bounds: false, cp); + error.StatusCode.Code.ShouldBe(StatusCodes.Good); + values.Count.ShouldBe(1); + all.AddRange(values); + readsPerPage.Add(source.Reads); + cp = next; + readsPerPage.Count.ShouldBeLessThan(20); + } + while (cp is not null); + + all.Select(x => (double)x.Value).ShouldBe([0d, 1d, 2d, 3d, 4d, 5d, 6d]); + // Page 1 [a], page 2 [t1]: one read each. Page 3 [t2]: the read stalls on the cluster, then the over-fetch (2). + // Pages 4 [t3], 5 [t4], 6 [t5]: the flagged cursor over-fetches directly (1 each). Page 7 [b]: one read. + readsPerPage.ShouldBe([1, 1, 2, 1, 1, 1, 1]); + } + /// /// #692 (low): a one-value bounded read whose start bound is not found asks whether the read holds anything. When the /// answer is "only the end bound", the bound it just found rides the continuation point — the next page is served