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

Merge fix/692-n1-paging: every HistoryRead-Raw page is full; N=1 no longer alternates empty pages (#692)

Browse Source
This commit is contained in:
Joseph Doherty
2026-10-13 01:02:31 -04:00
parent 15ccd9b722 e04e1bf127
commit aa923cea83
7 changed files with 658 additions and 105 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/Historian.md
+34 -7
View File
@@ -582,23 +582,44 @@ 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.) 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.
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. 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:
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. 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
@@ -792,7 +813,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 —
src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/HistoryBounds.cs
+29 -7
View File
@@ -17,8 +17,13 @@ namespace ZB.MOM.WW.OtOpcUa.OpcUaServer;
/// <param name="Open">The end time was not given (a one-sided read's sentinel): there is no end bound to look up.</param>
/// <param name="DataExhausted">The data is drained and the end bound did not fit on the previous page.</param>
/// <param name="LastReturnedUtc">The last value's time returned on an earlier page (stamps an open bound's placeholder).</param>
/// <param name="Resolved">
/// 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. <c>null</c> = not looked up yet.
/// </param>
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);
/// <summary>
/// Gitea #691: OPC UA Part 11 bounding values for HistoryRead-Raw (<c>ReadRawModifiedDetails.ReturnBounds</c>), composed
@@ -59,6 +64,15 @@ internal static class HistoryBounds
{
private static readonly TimeSpan OpenBoundStep = TimeSpan.FromSeconds(1);
/// <summary>
/// Gitea #692: the placeholders this server synthesised (<see cref="NotFound"/>), by reference. A placeholder is known by
/// this marker, never by its shape: a stored row can legitimately carry <c>Bad_BoundNotFound</c> with a null value (a
/// client may write one through HistoryUpdate), and that row is data.
/// </summary>
private static readonly System.Runtime.CompilerServices.ConditionalWeakTable<DataValueSnapshot, object> Placeholders = new();
private static readonly object PlaceholderMark = new();
/// <summary>The end bound owed by a fresh read over the normalised window.</summary>
/// <param name="startUtc">The (normalised) start: the later time of a reverse read.</param>
/// <param name="endUtc">The (normalised) end; a one-sided read's sentinel (tick 0 / MaxValue) is an open bound.</param>
@@ -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));
/// <summary>Whether a returned value is a <see cref="NotFound"/> placeholder rather than a stored sample.</summary>
/// <summary>
/// Whether a returned value is a <see cref="NotFound"/> placeholder this server synthesised rather than a stored sample
/// — by an explicit marker (#692), so a stored row whose status happens to be <c>Bad_BoundNotFound</c> is not taken for one.
/// </summary>
/// <param name="value">The value.</param>
/// <returns><c>true</c> for a <c>Bad_BoundNotFound</c> placeholder.</returns>
public static bool IsPlaceholder(DataValueSnapshot value) =>
value.StatusCode == StatusCodes.BadBoundNotFound && value.Value is null;
/// <returns><c>true</c> only for an instance <see cref="NotFound"/> produced.</returns>
public static bool IsPlaceholder(DataValueSnapshot value) => Placeholders.TryGetValue(value, out _);
/// <summary>The pending end bound to store behind a continuation point, remembering the page's last returned time.</summary>
/// <param name="bound">The pending end bound (null = bounds not requested).</param>
@@ -84,8 +100,12 @@ internal static class HistoryBounds
/// <summary>The <c>Bad_BoundNotFound</c> placeholder (null value) stamped at <paramref name="atUtc"/>.</summary>
/// <param name="atUtc">The bound's time.</param>
/// <returns>The placeholder.</returns>
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;
}
/// <summary>
/// The start bound to return ahead of the data, or <c>null</c> 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);
src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/HistoryPaging.cs
+15 -8
View File
@@ -66,6 +66,11 @@ namespace ZB.MOM.WW.OtOpcUa.OpcUaServer;
/// <see cref="HistoryCursorKind.Raw"/> with <c>ReturnBounds = true</c> only (Gitea #691): the end bound still owed —
/// Part 11 returns it with the last page. <c>null</c> = bounds not requested.
/// </param>
/// <param name="WithinTieCluster">
/// <see cref="HistoryCursorKind.Raw"/> only (Gitea #692): the previous page was a slice of an oversized tie cluster at
/// <paramref name="NextStartUtc"/> that still has un-emitted ties — the next page over-fetches the cluster directly
/// instead of first re-reading <c>N + skip + 1</c> rows that are known to be ties.
/// </param>
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)
{
/// <summary>The walk-back hint this cursor carries (#686/#689/#691).</summary>
public ReverseWalkHint WalkHint => new(SegmentWidthTicks, EmptyFromTicks, EmptyToTicks);
@@ -252,12 +258,11 @@ internal static class HistoryPaging
/// <param name="cap">The client's per-page cap (<c>NumValuesPerNode</c>); must be &gt; 0 to make progress.</param>
/// <param name="boundaryT">The single SourceTimestamp every tie in the cluster shares.</param>
/// <param name="endUtc">The (inclusive) upper bound of the read window; unchanged across pages.
/// OpcUaServer-003, accepted by design: the backend contract
/// (<c>IHistorianDataSource.ReadRawAsync</c>) treats <c>endUtc</c> as EXCLUSIVE, so the
/// <c>next &lt;= endUtc</c> test here is deliberately the conservative side — at worst one extra
/// empty resume page (<c>[endUtc, endUtc)</c> → GoodNoData, no continuation point), never a
/// dropped or duplicated sample. Tightening to <c>&lt;</c> would be correct only if the live
/// backend's boundary is confirmed exclusive; it is not worth the opposite off-by-one risk.</param>
/// The forward window is end-EXCLUSIVE (Part 11; the <c>IHistorianDataSource.ReadRawAsync</c> contract), so a drained
/// cluster continues only while <c>next</c> is strictly before <paramref name="endUtc"/> truncated to
/// <paramref name="resolutionTicks"/>. (OpcUaServer-003 had kept <c>next &lt;= endUtc</c> as "the conservative side",
/// but it was not: resuming AT EndTime makes the next read a <c>start == end</c> point read, which returns the samples
/// at EndTime as data — #692 review F1; on a microsecond backend EndTime is truncated first.)</param>
/// <param name="sliceStart">The index into the cluster the emitted slice starts at (clamped to the count).</param>
/// <param name="sliceCount">How many ties the emitted slice contains (may be <c>0</c> on a stale-skip self-heal).</param>
/// <param name="nextStartUtc">The next page's inclusive start: <paramref name="boundaryT"/> while the
@@ -300,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; }
}
}
src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs
+122 -69
View File
@@ -533,7 +533,8 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
/// <summary>
/// The upper bound on the bounded over-fetch <see cref="ServeRawPaged"/> 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 <c>N + skip + 1</c> 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 <c>MaxTieClusterOverfetch + 1</c> ties at that single timestamp (a
/// <c>start == end</c> read) and slices through them via <see cref="HistoryPaging.SliceTieCluster"/>.
/// A cluster strictly larger than this still fails that node <c>BadTooManyOperations</c> (the absurd-burst
@@ -6190,11 +6191,16 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
/// <list type="bullet">
/// <item><b>Fresh read</b> (no inbound continuation point): read the window from
/// <c>details.StartTime</c> to <paramref name="endUtc"/> capped at
/// <paramref name="numValuesPerNode"/>. If the page comes back FULL (exactly the cap, and the
/// cap is &gt; 0), store a resume cursor and emit a continuation point.</item>
/// <paramref name="numValuesPerNode"/> <b>+ 1</b> — 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).</item>
/// <item><b>Resume read</b> (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).</item>
/// <c>N + boundarySkip + 1</c> 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
/// <see cref="MaxTieClusterOverfetch"/> + 1 over-fetch to step one tick — Gitea #692.)</item>
/// </list>
/// The resume cursor is tie-safe (see <see cref="HistoryPaging.ComputeResumeCursor"/> /
/// <see cref="HistoryPaging.TrimBoundaryDuplicates"/>): 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<bool> 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<DataValueSnapshot> 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<DataValueSnapshot> 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<DataValueSnapshot> 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);
tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadPageFillTests.cs
+438
View File
@@ -0,0 +1,438 @@
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;
/// <summary>
/// Gitea #692: a paged HistoryRead-Raw fills every page. With <c>NumValuesPerNode = 1</c> 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
/// <c>MaxTieClusterOverfetch + 1</c> 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.
/// </summary>
public sealed class NodeManagerHistoryReadPageFillTests(BoundsServerFixture fx) : IClassFixture<BoundsServerFixture>
{
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 }
/// <summary>The live-gate repro (#692): 8 values one second apart, read over [19:00:00.1, 19:00:10.5).</summary>
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<Shape, Direction, bool> AllShapes()
{
var data = new TheoryData<Shape, Direction, bool>();
foreach (var s in Enum.GetValues<Shape>())
foreach (var d in Enum.GetValues<Direction>())
foreach (var b in new[] { false, true })
data.Add(s, d, b);
return data;
}
/// <summary>
/// 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).
/// </summary>
[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");
}
/// <summary>The live-gate numbers, pinned: 8 values at N = 1 on a native backend is 8 pages and 8 backend reads (was 17 / 25).</summary>
[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);
}
/// <summary>A series whose tie clusters straddle the page boundaries at every page size.</summary>
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<Shape, Direction, bool, uint> TiedShapes()
{
var data = new TheoryData<Shape, Direction, bool, uint>();
foreach (var s in Enum.GetValues<Shape>())
foreach (var d in Enum.GetValues<Direction>())
foreach (var b in new[] { false, true })
foreach (var n in new uint[] { 1, 2, 3 })
data.Add(s, d, b, n);
return data;
}
/// <summary>
/// #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.
/// </summary>
[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))}");
}
/// <summary>
/// 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).
/// </summary>
[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]);
}
/// <summary>
/// #692 review F1: an oversized tie cluster one step below EndTime used to resume AT EndTime (the drain check was
/// <c>next &lt;= end</c>, but the forward window is end-exclusive), so the next read was a <c>start == end</c> 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.
/// </summary>
[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<DataValueSnapshot>();
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));
}
/// <summary>
/// #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 <c>N + skip + 1</c> rows it knows are ties.
/// </summary>
[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<DataValueSnapshot> { 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<int>();
var all = new List<DataValue>();
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]);
}
/// <summary>
/// #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.
/// </summary>
[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");
}
/// <summary>
/// #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).
/// </summary>
[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();
}
}
/// <summary>
/// #692 (low): only a SERVER-synthesised bound is a placeholder. A stored row that happens to carry
/// <c>Bad_BoundNotFound</c> 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.
/// </summary>
[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<DataValue> values) =>
string.Join(" | ", values.Select(v => $"{v.SourceTimestamp:HH:mm:ss.fffffff}={v.Value ?? "null"}/{v.StatusCode.Code:X8}"));
private async Task<List<DataValue>> 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<List<IReadOnlyList<DataValue>>> DrainPagesAsync(
CountingSource source, DateTime start, DateTime end, uint max, bool bounds)
{
var pages = new List<IReadOnlyList<DataValue>>();
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<DataValue> 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<DataValue> 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<HistoryReadValueId> { new() { NodeId = fx.NodeId, ContinuationPoint = cp } };
var results = new List<SdkHistoryReadResult> { null! };
var errors = new List<ServiceResult> { null! };
fx.Nm.HistoryRead(context, details, TimestampsToReturn.Both, releaseContinuationPoints: false, nodesToRead, results, errors);
var values = new List<DataValue>();
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<DataValueSnapshot> 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),
};
/// <summary>Records every saved cursor (to inspect what rides a continuation point).</summary>
private sealed class RecordingStore : IHistoryContinuationStore
{
private readonly InMemoryHistoryContinuationStore _inner = new();
public List<HistoryContinuationState> 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);
}
/// <summary>
/// A series-backed historian that counts what it is asked for. <c>reverse</c>: serves <c>start &gt; end</c> newest-first
/// (SqlServer/QuestDB); otherwise refuses it (the Gateway — the server walks back with forward reads). Forward reads
/// are <c>[start, end)</c>, a point read the instant; <c>resolution</c> &gt; 1 truncates the window bounds (QuestDB µs).
/// </summary>
private sealed class CountingSource(IReadOnlyList<DataValueSnapshot> 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<HistorianRead> 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<DataValueSnapshot> 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<HistorianRead> ReadProcessedAsync(
string fullReference, DateTime startUtc, DateTime endUtc, TimeSpan interval,
HistoryAggregateType aggregate, CancellationToken cancellationToken) =>
Task.FromResult(new HistorianRead(Array.Empty<DataValueSnapshot>(), null));
public Task<HistorianRead> ReadAtTimeAsync(
string fullReference, IReadOnlyList<DateTime> timestampsUtc, CancellationToken cancellationToken) =>
Task.FromResult(new HistorianRead(Array.Empty<DataValueSnapshot>(), null));
public Task<HistoricalEventsResult> ReadEventsAsync(
string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents, CancellationToken cancellationToken) =>
Task.FromResult(new HistoricalEventsResult(Array.Empty<HistoricalEvent>(), null));
public HistorianHealthSnapshot GetHealthSnapshot() => NullHistorianDataSource.Instance.GetHealthSnapshot();
public void Dispose() { }
}
}
tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadPagingTests.cs
+19 -13
View File
@@ -29,8 +29,8 @@ public sealed class NodeManagerHistoryReadPagingTests : IDisposable
/// <summary>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.)</summary>
/// 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.)</summary>
[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();
}
/// <summary>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.</summary>
/// <summary>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.</summary>
[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;
tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadTests.cs
+1 -1
View File
@@ -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);