fix(historian): #686 review — bounded, density-aware reverse raw walk; uncapped Gateway raw reads

1. Uncapped reverse raw read on a forward-only backend = ONE forward read of the
   window, filtered and reversed in memory (was a segment-by-segment walk).
2. Segment width is aimed at ~half the row budget from the density just observed
   (overflow: from the truncated read's own sample span) and carried page to page
   in HistoryContinuationState.SegmentWidthTicks.
3. The walk starts no later than now + 1 day and never reads before the epoch.
4. Request bounds are rounded outward to the millisecond, so a row AVEVA
   fabricates at a rounded-up query start cannot land inside the segment.
5. Budget M = max(1, MaxTieClusterOverfetch); reads ask M + 2 rows, completeness is
   'short OR last row past the segment', and a read crowded by rows before the
   segment is re-read once (<= 2M + 4). A cluster of exactly M, neighbouring
   clusters each within M, and M <= 1 are all served; only one timestamp over M
   is refused. ServerHistorianOptions text updated.
6. The raw and events tie-path oversized-cluster refusal is BadTooManyOperations
   (was BadHistoryOperationUnsupported), consistent with the walk.
7. GatewayHistorianDataSource.ReadRawAsync: a caller cancellation is not a backend
   failure; NumValuesPerNode = 0 is sent as int.MaxValue (the client package and
   the gateway reject max_values <= 0, so every uncapped Gateway raw read failed);
   a MaxValuesCeiling refusal is a HistoryReadLimitExceededException.
8. Tests: upper-edge filter, epoch floor, look-ahead, ms alignment, exact/adjacent
   clusters, minimal budget, carried density, and the walk over the real Gateway
   source + FakeHistorianGatewayClient (OpcUaServer internals visible to it).
This commit is contained in:
Joseph Doherty
2026-10-12 15:11:21 -04:00
parent be2f0cc14d
commit cf1af55720
9 changed files with 554 additions and 109 deletions
@@ -90,8 +90,9 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
try try
{ {
// The gateway seam caps with an int; OPC UA hands us a uint, so clamp to int range. // The gateway seam caps with an int; OPC UA hands us a uint, so clamp to int range. 0 ("all values") is sent
var maxValues = (int)Math.Min(maxValuesPerNode, int.MaxValue); // unbounded: the client package and the gateway reject max_values <= 0 (#686 review).
var maxValues = maxValuesPerNode == 0 ? int.MaxValue : (int)Math.Min(maxValuesPerNode, int.MaxValue);
var samples = new List<HistorianSample>(); var samples = new List<HistorianSample>();
await foreach (var sample in _client await foreach (var sample in _client
.ReadRawAsync(fullReference, startUtc, endUtc, maxValues, cancellationToken) .ReadRawAsync(fullReference, startUtc, endUtc, maxValues, cancellationToken)
@@ -104,6 +105,18 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
RecordOutcome(success: true, error: null); RecordOutcome(success: true, error: null);
return new HistoryReadResult(snapshots, ContinuationPoint: null); return new HistoryReadResult(snapshots, ContinuationPoint: null);
} }
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
{
throw; // the caller (client cancel / HistoryReadDeadline) asked for this — not a backend fault
}
catch (Exception ex) when (IsMaxValuesCeilingRejection(ex))
{
// An unbounded read over Historian:ReadLimits:MaxValuesCeiling with AllowStreamingBeyondCeiling=false: a
// client-request limit (BadTooManyOperations), not a backend fault.
throw new HistoryReadLimitExceededException(maxValuesPerNode,
$"The HistorianGateway refused the row budget of the raw read of '{fullReference}' (its " +
"Historian:ReadLimits:MaxValuesCeiling); set NumValuesPerNode or narrow the window.");
}
catch (Exception ex) catch (Exception ex)
{ {
RecordReadFailure(ex); RecordReadFailure(ex);
@@ -51,6 +51,10 @@ namespace ZB.MOM.WW.OtOpcUa.OpcUaServer;
/// <c>(EndUtc, NextStartUtc]</c>. A resumed events read takes its direction from here, never from the resume /// <c>(EndUtc, NextStartUtc]</c>. A resumed events read takes its direction from here, never from the resume
/// request's details. Raw keeps deriving it from the window (<c>NextStartUtc &gt; EndUtc</c>). /// request's details. Raw keeps deriving it from the window (<c>NextStartUtc &gt; EndUtc</c>).
/// </param> /// </param>
/// <param name="SegmentWidthTicks">
/// <see cref="HistoryCursorKind.Raw"/> reverse reads on a forward-only backend only (Gitea #686): the segment width the
/// walk-back ended on, so the next page starts at the observed density instead of re-discovering it. 0 = none.
/// </param>
internal sealed record HistoryContinuationState( internal sealed record HistoryContinuationState(
string Tagname, string Tagname,
DateTime NextStartUtc, DateTime NextStartUtc,
@@ -60,7 +64,8 @@ internal sealed record HistoryContinuationState(
HistoryCursorKind Kind = HistoryCursorKind.Raw, HistoryCursorKind Kind = HistoryCursorKind.Raw,
long LastModifiedId = 0, long LastModifiedId = 0,
DateTime WindowStartUtc = default, DateTime WindowStartUtc = default,
bool Reverse = false); bool Reverse = false,
long SegmentWidthTicks = 0);
/// <summary>The HistoryRead kind that issued a <see cref="HistoryContinuationState"/>.</summary> /// <summary>The HistoryRead kind that issued a <see cref="HistoryContinuationState"/>.</summary>
internal enum HistoryCursorKind internal enum HistoryCursorKind
@@ -536,8 +536,9 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
/// 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 (the boundary-tie trim empties the page), the
/// paging over-fetches up to <c>MaxTieClusterOverfetch + 1</c> ties at that single timestamp (a /// 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"/>. /// <c>start == end</c> read) and slices through them via <see cref="HistoryPaging.SliceTieCluster"/>.
/// A cluster strictly larger than this still surfaces <c>BadHistoryOperationUnsupported</c> for that /// A cluster strictly larger than this still fails that node <c>BadTooManyOperations</c> (the absurd-burst
/// node (the absurd-burst backstop). Mirrors the configured /// backstop; <c>BadHistoryOperationUnsupported</c> before #686). Also the per-read row budget of a reverse raw read
/// walked back from forward reads (<see cref="ReverseRawReader"/>). Mirrors the configured
/// <c>ServerHistorianOptions.MaxTieClusterOverfetch</c>; the Host sets it at <c>StartAsync</c>. The /// <c>ServerHistorianOptions.MaxTieClusterOverfetch</c>; the Host sets it at <c>StartAsync</c>. The
/// default (65536) survives when the historian section is absent. /// default (65536) survives when the historian section is absent.
/// </summary> /// </summary>
@@ -5359,8 +5360,8 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
"OtOpcUaNodeManager: HistoryReadEvents tie cluster at {StartUtc:O} for source '{Source}' has {Count} events, " + "OtOpcUaNodeManager: HistoryReadEvents tie cluster at {StartUtc:O} for source '{Source}' has {Count} events, " +
"exceeding MaxTieClusterOverfetch={Max}; cannot page within it. Increase MaxTieClusterOverfetch.", "exceeding MaxTieClusterOverfetch={Max}; cannot page within it. Increase MaxTieClusterOverfetch.",
startUtc, sourceName, cluster.Count, MaxTieClusterOverfetch); startUtc, sourceName, cluster.Count, MaxTieClusterOverfetch);
errors[handle.Index] = StatusCodes.BadHistoryOperationUnsupported; errors[handle.Index] = StatusCodes.BadTooManyOperations;
results[handle.Index] = new SdkHistoryReadResult { StatusCode = StatusCodes.BadHistoryOperationUnsupported }; results[handle.Index] = new SdkHistoryReadResult { StatusCode = StatusCodes.BadTooManyOperations };
return; return;
} }
@@ -6183,6 +6184,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
{ {
DateTime startUtc; DateTime startUtc;
var boundarySkip = 0; var boundarySkip = 0;
long segmentWidth = 0; // #686: the walk-back's width, carried page to page
string tagname; string tagname;
if (inboundCp is { Length: > 0 }) if (inboundCp is { Length: > 0 })
@@ -6203,6 +6205,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
boundarySkip = state.BoundarySkipCount; boundarySkip = state.BoundarySkipCount;
endUtc = state.EndUtc; endUtc = state.EndUtc;
numValuesPerNode = state.NumValuesPerNode; numValuesPerNode = state.NumValuesPerNode;
segmentWidth = state.SegmentWidthTicks;
} }
else else
{ {
@@ -6235,16 +6238,22 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
// can read newest-first; otherwise it is walked back from bounded forward reads (#686 — the Gateway rejects // 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). // start > end, which failed the node and counted as a backend fault).
var source = HistorianDataSource; var source = HistorianDataSource;
var sourceResult = startUtc > endUtc && !source.SupportsReverseRaw HistorianRead sourceResult;
? new HistorianRead( if (startUtc > endUtc && !source.SupportsReverseRaw)
await ReverseRawReader.ReadAsync( {
source, tagname, startUtc, endUtc, numValuesPerNode, var page = await ReverseRawReader.ReadAsync(
(int)Math.Min(int.MaxValue, (long)Math.Max(1, MaxTieClusterOverfetch) + 1), ct) source, tagname, startUtc, endUtc, numValuesPerNode, MaxTieClusterOverfetch,
.ConfigureAwait(false), Clock.GetUtcNow().UtcDateTime, segmentWidth, ct)
ContinuationPoint: null) .ConfigureAwait(false);
: await source segmentWidth = page.WidthHintTicks;
sourceResult = new HistorianRead(page.Samples, ContinuationPoint: null);
}
else
{
sourceResult = await source
.ReadRawAsync(tagname, startUtc, endUtc, numValuesPerNode, ct) .ReadRawAsync(tagname, startUtc, endUtc, numValuesPerNode, ct)
.ConfigureAwait(false); .ConfigureAwait(false);
}
var backendFull = HistoryPaging.IsFullPage(sourceResult.Samples.Count, numValuesPerNode); var backendFull = HistoryPaging.IsFullPage(sourceResult.Samples.Count, numValuesPerNode);
@@ -6293,8 +6302,8 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
"OtOpcUaNodeManager: HistoryReadRaw tie cluster at {StartUtc:O} for tag '{Tagname}' has {Count} samples, " + "OtOpcUaNodeManager: HistoryReadRaw tie cluster at {StartUtc:O} for tag '{Tagname}' has {Count} samples, " +
"exceeding MaxTieClusterOverfetch={Max}; cannot page within it. Increase MaxTieClusterOverfetch.", "exceeding MaxTieClusterOverfetch={Max}; cannot page within it. Increase MaxTieClusterOverfetch.",
startUtc, tagname, cluster.Count, MaxTieClusterOverfetch); startUtc, tagname, cluster.Count, MaxTieClusterOverfetch);
errors[handle.Index] = StatusCodes.BadHistoryOperationUnsupported; errors[handle.Index] = StatusCodes.BadTooManyOperations;
results[handle.Index] = new SdkHistoryReadResult { StatusCode = StatusCodes.BadHistoryOperationUnsupported }; results[handle.Index] = new SdkHistoryReadResult { StatusCode = StatusCodes.BadTooManyOperations };
return; return;
} }
@@ -6315,7 +6324,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
if (nextStartUtc is { } resumeAt) if (nextStartUtc is { } resumeAt)
{ {
var clusterState = new HistoryContinuationState( var clusterState = new HistoryContinuationState(
tagname, resumeAt, endUtc, nextSkip, numValuesPerNode); tagname, resumeAt, endUtc, nextSkip, numValuesPerNode, SegmentWidthTicks: segmentWidth);
clusterCp = _historyContinuationStore.Save(session, clusterState); clusterCp = _historyContinuationStore.Save(session, clusterState);
} }
@@ -6338,7 +6347,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
{ {
HistoryPaging.ComputeResumeCursor(samples, out var nextStart, out var skip); HistoryPaging.ComputeResumeCursor(samples, out var nextStart, out var skip);
var nextState = new HistoryContinuationState( var nextState = new HistoryContinuationState(
tagname, nextStart, endUtc, skip, numValuesPerNode); tagname, nextStart, endUtc, skip, numValuesPerNode, SegmentWidthTicks: segmentWidth);
// Save may return null (no session on this request) ⇒ degrade to single-shot for this node. // 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. // Built AFTER historyData so a failure projecting samples can never orphan a stored cursor.
outboundCp = _historyContinuationStore.Save(session, nextState); outboundCp = _historyContinuationStore.Save(session, nextState);
@@ -2,6 +2,11 @@ using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
namespace ZB.MOM.WW.OtOpcUa.OpcUaServer; namespace ZB.MOM.WW.OtOpcUa.OpcUaServer;
/// <summary>One walked-back reverse raw page (<see cref="ReverseRawReader"/>).</summary>
/// <param name="Samples">At most the requested cap, newest first.</param>
/// <param name="WidthHintTicks">The segment width the walk ended on — the next page's starting width (0 = none).</param>
internal readonly record struct ReverseRawPage(IReadOnlyList<DataValueSnapshot> Samples, long WidthHintTicks);
/// <summary> /// <summary>
/// Gitea #686: serves a Part 11 reverse raw read — <c>(lo, hi]</c> newest-first, capped — on a historian backend that can /// Gitea #686: serves a Part 11 reverse raw read — <c>(lo, hi]</c> newest-first, capped — on a historian backend that can
/// only read forward (<see cref="IHistorianDataSource.SupportsReverseRaw"/> = <c>false</c>: the Gateway, whose /// only read forward (<see cref="IHistorianDataSource.SupportsReverseRaw"/> = <c>false</c>: the Gateway, whose
@@ -9,112 +14,191 @@ namespace ZB.MOM.WW.OtOpcUa.OpcUaServer;
/// </summary> /// </summary>
/// <remarks> /// <remarks>
/// <para> /// <para>
/// <b>Walk back in segments.</b> Starting at <c>hi</c>, read one segment <c>(segLo, segHi]</c> FORWARD, keep it /// <b>Uncapped</b> (<c>maxValues = 0</c>): ONE forward read of the whole window, filtered and reversed in memory — the
/// newest-first, and step to the next older segment until the cap is met or <c>lo</c> is reached. A segment starts /// same rows, round trips and memory as an uncapped forward read (and as a native backend's uncapped reverse read).
/// at <see cref="InitialSegment"/> and grows ×4 while segments come back sparse, so "the latest N" of a live tag is
/// usually one read, and sparse history decades back is found in a few dozen tiny reads rather than one read of
/// all history.
/// </para> /// </para>
/// <para> /// <para>
/// <b>Never answer from a truncated read.</b> Each segment read asks for at most <c>segmentCap</c> rows (the server's /// <b>Capped: walk back in segments.</b> Starting at <c>hi</c>, read one segment <c>(segLo, segHi]</c> FORWARD, keep
/// <c>MaxTieClusterOverfetch + 1</c> over-fetch bound). A read that comes back FULL may have been cut short — its /// it newest-first, and step to the next older segment until the cap is met or <c>lo</c> is reached. The first
/// newest rows are exactly the ones a reverse read needs — so it is discarded and the segment re-read at a quarter of /// segment is <see cref="InitialSegment"/> wide, or the width the previous page of the same read ended on (carried in
/// the width. A full read of a segment already one timestamp-resolution wide (a single-timestamp tie cluster larger /// the continuation point). After each read the next width is re-aimed at about half the row budget from the density
/// than the bound) cannot be served and throws <see cref="HistoryReadLimitExceededException"/> /// just observed, growing at most ×4 per step — so a live tag's "latest N" is usually one read, a dense tag does not
/// throw away a full read per page, and sparse history is found in a few dozen tiny reads.
/// </para>
/// <para>
/// <b>Never answer from a truncated read.</b> The row budget is <c>M = max(1, MaxTieClusterOverfetch)</c>; a segment
/// read asks for <c>M + 2</c> rows (room for the row an AVEVA read fabricates at its query start, and one more to see
/// past the segment). Rows stream ascending, so a read is complete for its segment when it came back short OR its last
/// row is already past <c>segHi</c>. An incomplete read is discarded: when rows BEFORE the segment (a neighbouring
/// tie cluster inside the alignment slack, the fabricated row) crowded it out it is re-read once with room for them
/// (at most <c>2M + 4</c> rows); otherwise the segment is narrowed (to about half the budget at the density the
/// truncated read showed) and re-read. Only a single timestamp holding more than <c>M</c> samples — a segment already
/// one timestamp-resolution wide that still does not fit — is refused: <see cref="HistoryReadLimitExceededException"/>
/// (⇒ <c>BadTooManyOperations</c>, a request limit, never a backend fault). /// (⇒ <c>BadTooManyOperations</c>, a request limit, never a backend fault).
/// </para> /// </para>
/// <para> /// <para>
/// <b>Edges, for any forward contract.</b> A segment is requested as <c>[segLo, segHi + resolution]</c> and filtered to /// <b>Edges.</b> A segment is requested as <c>[⌊segLo⌋, ⌈segHi + 1 tick⌉]</c>, both rounded outward to the millisecond
/// <c>(segLo, segHi]</c> on the samples' own timestamps, so it is exact whether the backend's forward end is exclusive /// (the gateway's / AVEVA's resolution), and filtered to <c>(segLo, segHi]</c> on the samples' own timestamps. That is
/// (the interface contract) or inclusive (the gateway), and whether it truncates its bounds (QuestDB-style). The /// exact whether the backend's forward end is exclusive (the interface contract) or inclusive (the gateway) and
/// filter also drops the rows an AVEVA <c>Full</c> read fabricates at the query start (the boundary / no-data row, /// whether it truncates its bounds (QuestDB-style); and because the request start is already on a millisecond, a row
/// #687) and at the query end. Adjacent segments share no timestamp, so nothing is returned twice. /// an AVEVA read fabricates at its (possibly rounded-up) query start can never land inside the segment (#687).
/// Adjacent segments share no timestamp, so nothing is returned twice.
/// </para> /// </para>
/// <para> /// <para>
/// <b>Ties</b> come back in the backend's forward order reversed — the same order the node manager's tie-cluster /// <b>Ties</b> come back in the backend's forward order reversed — the same order the node manager's tie-cluster
/// over-fetch uses for a reverse read — so a resumed page's boundary-tie skip lines up. /// over-fetch uses for a reverse read — so a resumed page's boundary-tie skip lines up.
/// </para> /// </para>
/// <para> /// <para>
/// <b>Floor.</b> The walk stops at <see cref="Floor"/> (the Unix epoch; its earliest request bound is one tick /// <b>Bounds.</b> The walk never asks for data after now + <see cref="LookAhead"/> (one day — the historian stores
/// before it): the AVEVA historian holds nothing earlier, and an end-only read's open lower bound (year 1) would /// nothing further ahead; a <c>StartTime = DateTime.MaxValue</c> read would otherwise send centuries-ahead query
/// otherwise cost dozens of empty reads and send a bound outside the historian's time encoding. Samples before the /// starts), nor before <see cref="Floor"/> (the Unix epoch — AVEVA holds nothing earlier, and an end-only read's year-1
/// floor are not returned by a walked-back read; a window entirely before it is empty. /// lower bound is outside the historian's time encoding). A walked-back read therefore returns only samples strictly
/// after the epoch and no later than now + one day.
/// </para> /// </para>
/// </remarks> /// </remarks>
internal static class ReverseRawReader internal static class ReverseRawReader
{ {
/// <summary>The first segment's width.</summary> /// <summary>The first segment's width when no hint is carried.</summary>
internal static readonly TimeSpan InitialSegment = TimeSpan.FromMinutes(1); internal static readonly TimeSpan InitialSegment = TimeSpan.FromMinutes(1);
/// <summary>The earliest instant a walk reads (see the remarks).</summary> /// <summary>The walk's exclusive lower limit (see the remarks).</summary>
internal static readonly DateTime Floor = DateTime.UnixEpoch; internal static readonly DateTime Floor = DateTime.UnixEpoch;
/// <summary>How far past "now" the walk starts at most (the same horizon as the Gateway's raw-aggregate look-ahead).</summary>
internal static readonly TimeSpan LookAhead = TimeSpan.FromDays(1);
/// <summary>Request bounds are rounded outward to this (the gateway / AVEVA timestamp resolution).</summary>
private const long Alignment = TimeSpan.TicksPerMillisecond;
/// <summary>Reads <c>(lo, hi]</c> newest-first from forward reads.</summary> /// <summary>Reads <c>(lo, hi]</c> newest-first from forward reads.</summary>
/// <param name="source">A backend that reads forward.</param> /// <param name="source">A backend that reads forward.</param>
/// <param name="tagname">Historian tag.</param> /// <param name="tagname">Historian tag.</param>
/// <param name="hi">The later (inclusive) bound — the request's StartTime or a reverse resume point.</param> /// <param name="hi">The later (inclusive) bound — the request's StartTime or a reverse resume point.</param>
/// <param name="lo">The earlier (exclusive) bound.</param> /// <param name="lo">The earlier (exclusive) bound.</param>
/// <param name="maxValues">Cap; <c>0</c> = every sample in the window.</param> /// <param name="maxValues">Cap; <c>0</c> = every sample in the window (one forward read).</param>
/// <param name="segmentCap">Rows one segment read may return before it is treated as truncated (≥ 2).</param> /// <param name="maxTieCluster">Row budget per segment read (<c>MaxTieClusterOverfetch</c>; values below 1 count as 1).</param>
/// <param name="nowUtc">Now, for the look-ahead clamp.</param>
/// <param name="widthHintTicks">Starting segment width (the previous page's <see cref="ReverseRawPage.WidthHintTicks"/>), 0 = none.</param>
/// <param name="ct">Cancellation.</param> /// <param name="ct">Cancellation.</param>
/// <returns>At most <paramref name="maxValues"/> samples, newest first.</returns> /// <returns>The page, newest first, and the width to start the next page with.</returns>
/// <exception cref="HistoryReadLimitExceededException">A single-timestamp tie cluster exceeds the segment cap.</exception> /// <exception cref="HistoryReadLimitExceededException">One timestamp holds more than the row budget.</exception>
public static async Task<IReadOnlyList<DataValueSnapshot>> ReadAsync( public static async Task<ReverseRawPage> ReadAsync(
IHistorianDataSource source, string tagname, DateTime hi, DateTime lo, uint maxValues, int segmentCap, IHistorianDataSource source, string tagname, DateTime hi, DateTime lo, uint maxValues, int maxTieCluster,
CancellationToken ct) DateTime nowUtc, long widthHintTicks, CancellationToken ct)
{ {
var resolution = Math.Max(1L, source.TimestampResolutionTicks); var resolution = Math.Max(1L, source.TimestampResolutionTicks);
var cap = (uint)Math.Max(2, segmentCap); var granularity = Math.Max(Alignment, resolution);
var budget = Math.Max(1, maxTieCluster);
var result = new List<DataValueSnapshot>(); var result = new List<DataValueSnapshot>();
var width = Math.Max(resolution, InitialSegment.Ticks);
var segHiTicks = hi.Ticks;
// The walk's exclusive lower edge. Clamped to one tick before the floor, so a sample exactly at the floor is still var horizon = nowUtc.Ticks > DateTime.MaxValue.Ticks - LookAhead.Ticks ? DateTime.MaxValue.Ticks : nowUtc.Ticks + LookAhead.Ticks;
// in the window (and the row a backend fabricates at that query start is still filtered out). var hiTicks = Math.Min(hi.Ticks, horizon);
var stopTicks = lo.Ticks < Floor.Ticks ? Floor.Ticks - 1 : lo.Ticks; var stopTicks = Math.Max(lo.Ticks, Floor.Ticks);
if (hiTicks <= stopTicks) return new ReverseRawPage(result, widthHintTicks);
while (segHiTicks > stopTicks && (maxValues == 0 || result.Count < maxValues)) if (maxValues == 0)
{ {
ct.ThrowIfCancellationRequested(); // One forward read of the whole window, reversed — the uncapped forward read's own cost and bound.
var segLoTicks = segHiTicks - stopTicks <= width ? stopTicks : segHiTicks - width; var all = (await source.ReadRawAsync(tagname, Utc(AlignDown(stopTicks, granularity)),
var requestEndTicks = segHiTicks > DateTime.MaxValue.Ticks - resolution Utc(AlignUp(hiTicks + 1, granularity)), 0, ct).ConfigureAwait(false)).Samples;
? DateTime.MaxValue.Ticks for (var i = all.Count - 1; i >= 0; i--)
: segHiTicks + resolution;
var rows = (await source.ReadRawAsync(
tagname,
new DateTime(segLoTicks, DateTimeKind.Utc),
new DateTime(requestEndTicks, DateTimeKind.Utc),
cap, ct)
.ConfigureAwait(false)).Samples;
if (rows.Count >= cap)
{ {
// Possibly truncated: its newest rows — the ones needed — may be missing. Narrow and re-read. var ts = Ticks(all[i]);
if (width <= resolution) if (ts > stopTicks && ts <= hiTicks) result.Add(all[i]);
throw new HistoryReadLimitExceededException(cap - 1,
$"A reverse raw read of '{tagname}' at {new DateTime(segHiTicks, DateTimeKind.Utc):O} found more than " +
$"{cap - 1} samples at one timestamp; the backend reads forward only, so the latest of them cannot be " +
"selected. Increase ServerHistorian:MaxTieClusterOverfetch.");
width = Math.Max(resolution, width / 4);
continue;
} }
var inSegment = 0; return new ReverseRawPage(result, widthHintTicks);
for (var i = rows.Count - 1; i >= 0; i--)
{
var ts = (rows[i].SourceTimestampUtc ?? DateTime.MinValue).Ticks;
if (ts <= segLoTicks || ts > segHiTicks) continue;
inSegment++;
if (maxValues == 0 || result.Count < maxValues) result.Add(rows[i]);
}
// Sparse ⇒ widen; otherwise keep the width that fit.
if (inSegment <= cap / 8)
width = width > DateTime.MaxValue.Ticks / 4 ? DateTime.MaxValue.Ticks : width * 4;
segHiTicks = segLoTicks;
} }
return result; var maxWidth = DateTime.MaxValue.Ticks;
var width = widthHintTicks > 0 ? Math.Clamp(widthHintTicks, resolution, maxWidth) : Math.Max(resolution, InitialSegment.Ticks);
var baseCap = budget + 2;
var maxCap = 2 * budget + 4;
var segHi = hiTicks;
while (segHi > stopTicks && result.Count < maxValues)
{
var segLo = segHi - stopTicks <= width ? stopTicks : segHi - width;
var start = Utc(AlignDown(segLo, granularity));
var end = Utc(AlignUp(segHi + 1, granularity));
var cap = baseCap;
IReadOnlyList<DataValueSnapshot> rows;
var narrowed = false;
while (true)
{
ct.ThrowIfCancellationRequested();
rows = (await source.ReadRawAsync(tagname, start, end, (uint)cap, ct).ConfigureAwait(false)).Samples;
if (rows.Count < cap || Ticks(rows[^1]) > segHi) break; // complete for (segLo, segHi]
var leading = 0;
var inner = 0;
var firstInner = 0L;
foreach (var r in rows)
{
var ts = Ticks(r);
if (ts <= segLo) leading++;
else if (ts <= segHi && inner++ == 0) firstInner = ts;
}
// Crowded out by rows before the segment, with the segment itself possibly within budget: re-read once
// with room for them.
var roomier = leading + budget + 2;
if (leading > 0 && inner <= budget && roomier > cap && roomier <= maxCap)
{
cap = roomier;
continue;
}
if (width <= resolution)
throw new HistoryReadLimitExceededException(budget,
$"A reverse raw read of '{tagname}' at {Utc(segHi):O} found more than {budget} samples at one " +
"timestamp; the backend reads forward only, so the latest of them cannot be selected. Increase " +
"ServerHistorian:MaxTieClusterOverfetch.");
// Narrow to about half the budget at the density the truncated read's own samples showed (their span, not
// the segment's — the segment may open on an empty stretch), at least halving.
var lastTs = Ticks(rows[^1]);
var aimed = inner > 1 && lastTs > firstInner
? (long)Math.Min((double)(lastTs - firstInner) * (budget / 2.0) / (inner - 1), width / 2.0)
: width / 4;
width = Math.Max(resolution, aimed);
narrowed = true;
break;
}
if (narrowed) continue; // re-read the same segHi, narrower
var count = 0;
for (var i = rows.Count - 1; i >= 0; i--)
{
var ts = Ticks(rows[i]);
if (ts <= segLo || ts > segHi) continue;
count++;
if (result.Count < maxValues) result.Add(rows[i]);
}
// Re-aim at about half the budget from the density just seen; grow at most x4.
var grown = width > maxWidth / 4 ? maxWidth : width * 4;
width = count == 0
? grown
: (long)Math.Clamp((double)width * Math.Max(1.0, budget / 2.0) / count, resolution, grown);
segHi = segLo;
}
return new ReverseRawPage(result, width);
}
private static long Ticks(DataValueSnapshot s) => (s.SourceTimestampUtc ?? DateTime.MinValue).Ticks;
private static DateTime Utc(long ticks) => new(ticks, DateTimeKind.Utc);
private static long AlignDown(long ticks, long step) => ticks - ticks % step;
private static long AlignUp(long ticks, long step)
{
var down = AlignDown(ticks, step);
if (down == ticks) return ticks;
return down > DateTime.MaxValue.Ticks - step ? DateTime.MaxValue.Ticks : down + step;
} }
} }
@@ -35,6 +35,8 @@
<!-- Gitea #562: the over-the-wire event-revalidation test asserts the server-side fan-out, because a <!-- Gitea #562: the over-the-wire event-revalidation test asserts the server-side fan-out, because a
deleted item resurrected onto a notifier is invisible to the client that deleted it. --> deleted item resurrected onto a notifier is invisible to the client that deleted it. -->
<InternalsVisibleTo Include="ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests"/> <InternalsVisibleTo Include="ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests"/>
<!-- Gitea #686: the reverse-raw walk-back is driven over the real Gateway source + its fake gateway client. -->
<InternalsVisibleTo Include="ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests"/>
</ItemGroup> </ItemGroup>
</Project> </Project>
@@ -81,8 +81,11 @@ public sealed class ServerHistorianOptions
/// cap). When a resume read stalls on such a cluster, the node manager over-fetches up to /// cap). When a resume read stalls on such a cluster, the node manager over-fetches up to
/// <c>MaxTieClusterOverfetch + 1</c> ties at that single timestamp (a <c>start == end</c> read) and /// <c>MaxTieClusterOverfetch + 1</c> ties at that single timestamp (a <c>start == end</c> read) and
/// pages through them; this bounds how large a single-timestamp burst the server will buffer in /// pages through them; this bounds how large a single-timestamp burst the server will buffer in
/// memory before it gives up and surfaces <c>BadHistoryOperationUnsupported</c> for that node. The /// memory before it gives up and surfaces <c>BadTooManyOperations</c> for that node (#686; was
/// default (65536) comfortably covers any realistic same-millisecond burst. /// <c>BadHistoryOperationUnsupported</c>). It is also the per-read row budget of a reverse raw read walked
/// back from forward reads on a backend that cannot read newest-first (the Gateway): one timestamp holding
/// more samples than this is refused <c>BadTooManyOperations</c>. The default (65536) comfortably covers
/// any realistic same-millisecond burst; a value below 1 is treated as 1.
/// </summary> /// </summary>
public int MaxTieClusterOverfetch { get; init; } = 65536; public int MaxTieClusterOverfetch { get; init; } = 65536;
@@ -167,7 +170,7 @@ public sealed class ServerHistorianOptions
// the over-fetch code path only runs when a real IHistorianDataSource is wired in, // the over-fetch code path only runs when a real IHistorianDataSource is wired in,
// so a zero/negative value is harmless (and noise-free) when the historian is disabled. // so a zero/negative value is harmless (and noise-free) when the historian is disabled.
if (MaxTieClusterOverfetch <= 0) if (MaxTieClusterOverfetch <= 0)
warnings.Add($"ServerHistorian:MaxTieClusterOverfetch is {MaxTieClusterOverfetch} — must be > 0; HistoryRead-Raw cannot page within an oversized tie cluster and will surface BadHistoryOperationUnsupported for those reads."); warnings.Add($"ServerHistorian:MaxTieClusterOverfetch is {MaxTieClusterOverfetch} — must be > 0 (treated as 1); HistoryRead-Raw and -Events cannot page within a tie cluster larger than that, and a Gateway reverse raw read cannot serve one timestamp holding more samples — those reads surface BadTooManyOperations.");
// S3 HistoryRead concurrency knobs — a non-positive value disables the corresponding bound, so warn // S3 HistoryRead concurrency knobs — a non-positive value disables the corresponding bound, so warn
// (these gate the per-node fan-out / batch limiter that only run when a real data source is wired). // (these gate the per-node fan-out / batch limiter that only run when a real data source is wired).
if (HistoryReadBatchParallelism <= 0) if (HistoryReadBatchParallelism <= 0)
@@ -1,7 +1,12 @@
using Google.Protobuf.WellKnownTypes;
using Grpc.Core;
using Microsoft.Extensions.Logging.Abstractions; using Microsoft.Extensions.Logging.Abstractions;
using Shouldly; using Shouldly;
using Xunit; using Xunit;
using ZB.MOM.WW.HistorianGateway.Client;
using ZB.MOM.WW.HistorianGateway.Contracts.Grpc;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions; using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.OpcUaServer;
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests; namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests;
@@ -9,17 +14,21 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests;
/// Gitea #686: the gateway's <c>ReadRaw</c> streams ascending only and rejects <c>start &gt; end</c> with /// Gitea #686: the gateway's <c>ReadRaw</c> streams ascending only and rejects <c>start &gt; end</c> with
/// <c>InvalidArgument</c>, which used to be recorded as a backend health failure. The source declares it cannot read /// <c>InvalidArgument</c>, which used to be recorded as a backend health failure. The source declares it cannot read
/// newest-first (so the server walks back with forward reads), and refuses a reverse window itself without sending /// newest-first (so the server walks back with forward reads), and refuses a reverse window itself without sending
/// anything or counting a failure. /// anything or counting a failure. The walk itself is driven here over the REAL source and an AVEVA-shaped fake.
/// </summary> /// </summary>
public sealed class GatewayRawDirectionTests public sealed class GatewayRawDirectionTests
{ {
private static CancellationToken Ct => TestContext.Current.CancellationToken; private static CancellationToken Ct => TestContext.Current.CancellationToken;
private static readonly DateTime T0 = new(2026, 10, 12, 18, 6, 0, DateTimeKind.Utc);
private static GatewayHistorianDataSource Source(FakeHistorianGatewayClient fake) =>
new(fake, NullLogger<GatewayHistorianDataSource>.Instance);
[Fact] [Fact]
public void The_gateway_source_declares_it_cannot_read_raw_newest_first() public void The_gateway_source_declares_it_cannot_read_raw_newest_first()
{ {
IHistorianDataSource ds = new GatewayHistorianDataSource( IHistorianDataSource ds = Source(new FakeHistorianGatewayClient());
new FakeHistorianGatewayClient(), NullLogger<GatewayHistorianDataSource>.Instance);
ds.SupportsReverseRaw.ShouldBeFalse(); ds.SupportsReverseRaw.ShouldBeFalse();
} }
@@ -27,7 +36,7 @@ public sealed class GatewayRawDirectionTests
public async Task A_reverse_raw_window_is_refused_unsent_and_is_not_a_backend_failure() public async Task A_reverse_raw_window_is_refused_unsent_and_is_not_a_backend_failure()
{ {
var fake = new FakeHistorianGatewayClient(); var fake = new FakeHistorianGatewayClient();
IHistorianDataSource ds = new GatewayHistorianDataSource(fake, NullLogger<GatewayHistorianDataSource>.Instance); IHistorianDataSource ds = Source(fake);
var end = new DateTime(2026, 10, 12, 18, 7, 45, DateTimeKind.Utc); var end = new DateTime(2026, 10, 12, 18, 7, 45, DateTimeKind.Utc);
await Should.ThrowAsync<NotSupportedException>(() => await Should.ThrowAsync<NotSupportedException>(() =>
@@ -42,4 +51,135 @@ public sealed class GatewayRawDirectionTests
await ds.ReadRawAsync("Gate681", new DateTime(2026, 10, 12, 18, 0, 0, DateTimeKind.Utc), end, 3, Ct); await ds.ReadRawAsync("Gate681", new DateTime(2026, 10, 12, 18, 0, 0, DateTimeKind.Utc), end, 3, Ct);
fake.ReadRawCallCount.ShouldBe(1); fake.ReadRawCallCount.ShouldBe(1);
} }
/// <summary>
/// An uncapped raw read (<c>NumValuesPerNode = 0</c>, "all values") is sent as an unbounded <c>max_values</c>. The
/// client package and the gateway both reject <c>max_values &lt;= 0</c>, so forwarding 0 failed every uncapped raw
/// read — forward, and the one-shot read an uncapped reverse read is served from (#686 review).
/// </summary>
[Fact]
public async Task An_uncapped_raw_read_asks_the_gateway_for_an_unbounded_max_values()
{
var fake = new FakeHistorianGatewayClient();
await Source(fake).ReadRawAsync("T", T0, T0.AddHours(1), 0, Ct);
fake.LastReadRawMaxValues.ShouldBe(int.MaxValue);
}
[Fact]
public async Task A_deadline_cancellation_is_not_a_backend_failure()
{
var fake = new FakeHistorianGatewayClient { RawSamples = [Wire(T0, 1)] };
var ds = Source(fake);
using var cts = new CancellationTokenSource();
await cts.CancelAsync();
await Should.ThrowAsync<OperationCanceledException>(() => ds.ReadRawAsync("T", T0, T0.AddHours(1), 10, cts.Token));
ds.GetHealthSnapshot().TotalFailures.ShouldBe(0);
}
[Fact]
public async Task The_gateways_max_values_ceiling_rejection_of_a_raw_read_is_a_request_limit_not_a_backend_fault()
{
var rpc = new RpcException(new Status(StatusCode.InvalidArgument,
"max_values exceeds the server ceiling of 1000000; read large ranges by time-slicing ..."));
var ds = Source(new FakeHistorianGatewayClient { ThrowOnRead = new HistorianGatewayException(rpc.Status.Detail, rpc) });
await Should.ThrowAsync<HistoryReadLimitExceededException>(() => ds.ReadRawAsync("T", T0, T0.AddHours(1), 0, Ct));
ds.GetHealthSnapshot().TotalFailures.ShouldBe(0);
}
// ---- the walk-back over the real source ---------------------------------------------------------------------------
private static HistorianSample Wire(DateTime t, double v, uint opc = 192) =>
new() { Tag = "T", Timestamp = Timestamp.FromDateTime(t), NumericValue = v, OpcQuality = opc };
/// <summary>
/// A gateway over stored samples (value i at T0 + i s, i = 0..299) shaped like AVEVA: ReadRaw rejects
/// <c>start &gt; end</c> and <c>max_values &lt;= 0</c>, returns <c>[start, end]</c> ascending capped at
/// <c>max_values</c>, and first emits a row stamped at the query start carrying the value in effect there
/// (<c>Quality = 133</c>), or a NULL row (<c>OpcQuality = 0</c>, value 0) when there is none.
/// </summary>
private static FakeHistorianGatewayClient AvevaStore()
{
var stored = Enumerable.Range(0, 300).Select(i => (Time: T0.AddSeconds(i), Value: (double)i)).ToList();
return new FakeHistorianGatewayClient
{
RawSource = (start, end, max) =>
{
if (start > end) throw new RpcException(new Status(StatusCode.InvalidArgument, "start_time must be less than or equal to end_time."));
if (max <= 0) throw new ArgumentOutOfRangeException(nameof(max));
var rows = new List<HistorianSample>();
if (stored.All(s => s.Time != start))
{
var prior = stored.LastOrDefault(s => s.Time < start);
var boundary = prior.Time == default
? new HistorianSample { Tag = "T", NumericValue = 0, OpcQuality = 0, Quality = 1 }
: Wire(prior.Time, prior.Value);
boundary.Timestamp = Timestamp.FromDateTime(start);
if (prior.Time != default) boundary.Quality = 133;
rows.Add(boundary);
}
rows.AddRange(stored.Where(s => s.Time >= start && s.Time <= end).Select(s => Wire(s.Time, s.Value)));
return rows.Take(max);
},
};
}
private static void ShouldOnlyHaveSentValidForwardReads(FakeHistorianGatewayClient fake)
{
fake.RawCalls.ShouldNotBeEmpty();
foreach (var (start, end, max) in fake.RawCalls)
{
start.ShouldBeLessThanOrEqualTo(end);
max.ShouldBeGreaterThan(0);
start.ShouldBeGreaterThanOrEqualTo(DateTime.UnixEpoch);
}
fake.RawStreamsCompleted.ShouldBe(fake.ReadRawCallCount); // no gateway stream abandoned (historiangw#26)
}
[Fact]
public async Task A_capped_reverse_read_walked_back_over_the_gateway_returns_the_latest_values_newest_first()
{
var fake = AvevaStore();
var ds = Source(fake);
var page = await ReverseRawReader.ReadAsync(ds, "T", hi: T0.AddSeconds(200), lo: T0.AddSeconds(50), maxValues: 10,
maxTieCluster: 65536, nowUtc: T0.AddHours(1), widthHintTicks: 0, Ct);
page.Samples.Select(s => (double)s.Value!).ShouldBe(Enumerable.Range(191, 10).Reverse().Select(i => (double)i));
page.Samples.ShouldAllBe(s => s.StatusCode == 0u);
ShouldOnlyHaveSentValidForwardReads(fake);
ds.GetHealthSnapshot().TotalFailures.ShouldBe(0);
}
[Fact]
public async Task An_end_only_read_walked_back_over_the_gateway_reaches_the_first_sample_and_no_fabricated_row()
{
var fake = AvevaStore();
var ds = Source(fake);
var page = await ReverseRawReader.ReadAsync(ds, "T", hi: T0.AddSeconds(2), lo: new DateTime(0, DateTimeKind.Utc),
maxValues: 10, maxTieCluster: 65536, nowUtc: T0.AddHours(1), widthHintTicks: 0, Ct);
// 2, 1, 0 — and not the NULL row AVEVA fabricates at each query start before the first sample.
page.Samples.Select(s => (double)s.Value!).ShouldBe([2d, 1d, 0d]);
ShouldOnlyHaveSentValidForwardReads(fake);
}
[Fact]
public async Task An_uncapped_reverse_read_over_the_gateway_is_one_forward_read()
{
var fake = AvevaStore();
var ds = Source(fake);
var page = await ReverseRawReader.ReadAsync(ds, "T", hi: T0.AddSeconds(200), lo: T0.AddSeconds(50), maxValues: 0,
maxTieCluster: 65536, nowUtc: T0.AddHours(1), widthHintTicks: 0, Ct);
page.Samples.Select(s => (double)s.Value!).ShouldBe(Enumerable.Range(51, 150).Reverse().Select(i => (double)i));
fake.ReadRawCallCount.ShouldBe(1);
ShouldOnlyHaveSentValidForwardReads(fake);
}
} }
@@ -69,7 +69,7 @@ public sealed class NodeManagerHistoryReadEventsPagingTests : IDisposable
var p1 = ReadEvents(nm, node, max: 3, inboundCp: null); var p1 = ReadEvents(nm, node, max: 3, inboundCp: null);
p1.Cp.ShouldNotBeNull(); p1.Cp.ShouldNotBeNull();
var p2 = ReadEvents(nm, node, max: 3, inboundCp: p1.Cp); var p2 = ReadEvents(nm, node, max: 3, inboundCp: p1.Cp);
p2.Error.StatusCode.Code.ShouldBe(StatusCodes.BadHistoryOperationUnsupported); p2.Error.StatusCode.Code.ShouldBe(StatusCodes.BadTooManyOperations);
await host.DisposeAsync(); await host.DisposeAsync();
} }
@@ -283,7 +283,7 @@ public sealed class NodeManagerHistoryReadEventsPagingTests : IDisposable
var p1 = ReadEvents(nm, node, max: 3, inboundCp: null, end: Early, start: Late); var p1 = ReadEvents(nm, node, max: 3, inboundCp: null, end: Early, start: Late);
p1.Cp.ShouldNotBeNull(); p1.Cp.ShouldNotBeNull();
ReadEvents(nm, node, max: 3, inboundCp: p1.Cp, end: Early, start: Late).Error.StatusCode.Code ReadEvents(nm, node, max: 3, inboundCp: p1.Cp, end: Early, start: Late).Error.StatusCode.Code
.ShouldBe(StatusCodes.BadHistoryOperationUnsupported); .ShouldBe(StatusCodes.BadTooManyOperations);
await host.DisposeAsync(); await host.DisposeAsync();
} }
@@ -253,8 +253,9 @@ public sealed class NodeManagerHistoryReadPagingTests : IDisposable
} }
/// <summary>The absurd-burst backstop is preserved: a tie cluster STRICTLY larger than the configured /// <summary>The absurd-burst backstop is preserved: a tie cluster STRICTLY larger than the configured
/// <see cref="OtOpcUaNodeManager.MaxTieClusterOverfetch"/> still surfaces /// <see cref="OtOpcUaNodeManager.MaxTieClusterOverfetch"/> still fails that node rather than buffering an unbounded
/// <c>BadHistoryOperationUnsupported</c> for that node rather than buffering an unbounded burst. With /// burst — <c>BadTooManyOperations</c>, the request-limit code the reverse walk-back uses too (#686; was
/// <c>BadHistoryOperationUnsupported</c>). With
/// the bound set to 3 and a 5-way tie cluster, the resume read that hits the cluster fails loudly.</summary> /// the bound set to 3 and a 5-way tie cluster, the resume read that hits the cluster fails loudly.</summary>
[Fact] [Fact]
public async Task Raw_tie_cluster_beyond_overfetch_bound_fails_loudly() public async Task Raw_tie_cluster_beyond_overfetch_bound_fails_loudly()
@@ -283,7 +284,7 @@ public sealed class NodeManagerHistoryReadPagingTests : IDisposable
// Page 2: the resume read hits the cluster; over-fetch finds 5 > bound 3 ⇒ a clear error, NOT a // Page 2: the resume read hits the cluster; over-fetch finds 5 > bound 3 ⇒ a clear error, NOT a
// silent GoodNoData that would drop the un-emitted ties. // silent GoodNoData that would drop the un-emitted ties.
var (r2, e2, cp2) = ReadRaw(nm, nodeId, Epoch, Epoch.AddHours(1), max: 2, inboundCp: cp1); var (r2, e2, cp2) = ReadRaw(nm, nodeId, Epoch, Epoch.AddHours(1), max: 2, inboundCp: cp1);
e2.StatusCode.Code.ShouldBe(StatusCodes.BadHistoryOperationUnsupported); e2.StatusCode.Code.ShouldBe(StatusCodes.BadTooManyOperations);
r2.ShouldBeEmpty(); r2.ShouldBeEmpty();
cp2.ShouldBeNull(); cp2.ShouldBeNull();
@@ -531,6 +532,7 @@ public sealed class NodeManagerHistoryReadPagingTests : IDisposable
values.ShouldBe(Enumerable.Range(10, 191).Reverse().Select(i => (double)i)); values.ShouldBe(Enumerable.Range(10, 191).Reverse().Select(i => (double)i));
cp.ShouldBeNull(); cp.ShouldBeNull();
source.ReverseWindowCount.ShouldBe(0); source.ReverseWindowCount.ShouldBe(0);
source.ReadCount.ShouldBe(1, "an uncapped reverse read is ONE forward read, reversed in memory — not a walk");
await host.DisposeAsync(); await host.DisposeAsync();
} }
@@ -564,7 +566,7 @@ public sealed class NodeManagerHistoryReadPagingTests : IDisposable
var collected = await DrainAsync(nm, nodeId, DateTime.MinValue, Epoch.AddHours(1), max: 100); var collected = await DrainAsync(nm, nodeId, DateTime.MinValue, Epoch.AddHours(1), max: 100);
collected.ShouldBe(Enumerable.Range(0, 250).Reverse().Select(i => (double)i)); collected.ShouldBe(Enumerable.Range(0, 250).Reverse().Select(i => (double)i));
source.MaxRowsReturned.ShouldBeLessThanOrEqualTo(17); source.MaxRowsReturned.ShouldBeLessThanOrEqualTo(2 * 16 + 4);
source.ReverseWindowCount.ShouldBe(0); source.ReverseWindowCount.ShouldBe(0);
await host.DisposeAsync(); await host.DisposeAsync();
} }
@@ -587,7 +589,8 @@ public sealed class NodeManagerHistoryReadPagingTests : IDisposable
error.StatusCode.Code.ShouldBe(StatusCodes.Good); error.StatusCode.Code.ShouldBe(StatusCodes.Good);
values.ShouldBe([2d, 1d]); values.ShouldBe([2d, 1d]);
cp.ShouldBeNull(); cp.ShouldBeNull();
source.ReadCount.ShouldBeLessThan(40); source.ReadCount.ShouldBeLessThanOrEqualTo(20);
source.MinRequestedStart.ShouldBeGreaterThanOrEqualTo(DateTime.UnixEpoch, "the walk never asks for data before 1970");
source.ReverseWindowCount.ShouldBe(0); source.ReverseWindowCount.ShouldBe(0);
await host.DisposeAsync(); await host.DisposeAsync();
} }
@@ -609,6 +612,159 @@ public sealed class NodeManagerHistoryReadPagingTests : IDisposable
await host.DisposeAsync(); await host.DisposeAsync();
} }
/// <summary>#686 review: a dense tag is not re-discovered every page — the width the walk ended on rides the
/// continuation point, so after the first page each page is about one read, with no full (discarded) reads.</summary>
[Fact]
public async Task Raw_reverse_paging_on_a_forward_only_backend_carries_the_observed_density_between_pages()
{
// 2000 samples 10 ms apart: a 1-minute first segment holds all of them, far over a budget of 64.
var series = Enumerable.Range(0, 2000)
.Select(i => { var t = Epoch.AddMilliseconds(10L * i); return new DataValueSnapshot((double)i, StatusCodes.Good, t, t); })
.ToArray();
var source = new ForwardOnlyHistorianDataSource(series);
var (host, nm, nodeId) = await BootForwardOnlyAsync(source, "FwdDensity");
nm.MaxTieClusterOverfetch = 64;
// Page 1 finds the data (walking down an empty hour) and its density; measure the pages after it.
var (first, _, cp) = ReadRaw(nm, nodeId, DateTime.MinValue, Epoch.AddSeconds(20), max: 20, inboundCp: null);
first.ShouldBe(Enumerable.Range(1980, 20).Reverse().Select(i => (double)i));
source.ResetCounts();
var collected = new List<double>(first);
var pages = 0;
while (cp is not null)
{
var (values, error, next) = ReadRaw(nm, nodeId, DateTime.MinValue, Epoch.AddSeconds(20), max: 20, inboundCp: cp);
error.StatusCode.Code.ShouldBe(StatusCodes.Good);
collected.AddRange(values);
cp = next;
(++pages).ShouldBeLessThan(200);
}
collected.ShouldBe(Enumerable.Range(0, 2000).Reverse().Select(i => (double)i));
// The discriminating count: reads that came back full and were thrown away. Without the carried width every page
// starts on a 1-minute segment holding all 2000 samples and discards full reads first (~2 per page).
source.FullReads.ShouldBeLessThanOrEqualTo(2);
// About one read per page, plus the final pages' walks from the first sample down to the 1970 floor (~16 each).
source.ReadCount.ShouldBeLessThanOrEqualTo(pages + 40);
await host.DisposeAsync();
}
/// <summary>#686 review: a reverse read from <c>StartTime = DateTime.MaxValue</c> walks back from now + 1 day, not from
/// year 9999.</summary>
[Fact]
public async Task Raw_reverse_read_from_max_value_on_a_forward_only_backend_starts_at_now_plus_one_day()
{
var source = new ForwardOnlyHistorianDataSource(MakeSeries(count: 250, stepSeconds: 1));
var (host, nm, nodeId) = await BootForwardOnlyAsync(source, "FwdMax");
var now = Epoch.AddHours(1);
nm.Clock = new FixedClock(now);
var (values, error, _) = ReadRaw(nm, nodeId, DateTime.MaxValue, Epoch.AddSeconds(-1), max: 3, inboundCp: null);
error.StatusCode.Code.ShouldBe(StatusCodes.Good);
values.ShouldBe([249d, 248d, 247d]);
source.MaxRequestedEnd.ShouldBeLessThanOrEqualTo(now.AddDays(1).AddMilliseconds(1));
source.ReadCount.ShouldBeLessThanOrEqualTo(10);
await host.DisposeAsync();
}
/// <summary>#686 review: AVEVA may stamp the row it fabricates at a query start rounded UP to the millisecond. The walk
/// requests millisecond-aligned starts, so that row can never land inside (EndTime, …] when EndTime is not on a
/// millisecond.</summary>
[Fact]
public async Task Raw_reverse_read_on_a_forward_only_backend_never_returns_a_boundary_row_rounded_up_into_the_window()
{
var source = new ForwardOnlyHistorianDataSource(MakeSeries(count: 250, stepSeconds: 1), boundaryAtCeilingMs: true);
var (host, nm, nodeId) = await BootForwardOnlyAsync(source, "FwdCeil");
var (values, error, _) = ReadRaw(nm, nodeId, Epoch.AddSeconds(10), Epoch.AddTicks(5_000), max: 100, inboundCp: null);
error.StatusCode.Code.ShouldBe(StatusCodes.Good);
values.ShouldBe(Enumerable.Range(1, 10).Reverse().Select(i => (double)i)); // (Epoch+0.5ms, Epoch+10s]: no 0
await host.DisposeAsync();
}
/// <summary>#686 review: the upper edge is filtered too — the request end is rounded UP to the millisecond, so a
/// sample one tick after StartTime comes back from the backend and must not be returned.</summary>
[Fact]
public async Task Raw_reverse_read_on_a_forward_only_backend_excludes_a_sample_one_tick_after_the_start()
{
DataValueSnapshot At(long ticks, double v)
{
var t = Epoch.AddTicks(ticks);
return new DataValueSnapshot(v, StatusCodes.Good, t, t);
}
var t0 = TimeSpan.TicksPerSecond;
var source = new ForwardOnlyHistorianDataSource([At(0, 0), At(t0, 1), At(t0 + 1, 99)]);
var (host, nm, nodeId) = await BootForwardOnlyAsync(source, "FwdUpper");
var (values, error, _) = ReadRaw(nm, nodeId, Epoch.AddTicks(t0), Epoch.AddSeconds(-1), max: 10, inboundCp: null);
error.StatusCode.Code.ShouldBe(StatusCodes.Good);
values.ShouldBe([1d, 0d]);
await host.DisposeAsync();
}
/// <summary>#686 review: a tie cluster of EXACTLY <c>MaxTieClusterOverfetch</c> is served — the row the backend
/// fabricates at the query start does not consume the budget.</summary>
[Fact]
public async Task Raw_reverse_read_on_a_forward_only_backend_serves_a_tie_cluster_of_exactly_the_budget()
{
var source = new ForwardOnlyHistorianDataSource(MakeSeriesWithTieCluster(distinctBefore: 2, tieCount: 4, distinctAfter: 2));
var (host, nm, nodeId) = await BootForwardOnlyAsync(source, "FwdExact");
nm.MaxTieClusterOverfetch = 4;
var (values, error, _) = ReadRaw(nm, nodeId, Epoch.AddHours(1), Epoch.AddSeconds(-1), max: 100, inboundCp: null);
error.StatusCode.Code.ShouldBe(StatusCodes.Good);
values.ShouldBe(Enumerable.Range(0, 8).Reverse().Select(i => (double)i));
await host.DisposeAsync();
}
/// <summary>#686 review: adjacent tie clusters one millisecond apart, none over the budget, are served — a neighbour
/// inside the request's alignment slack does not get the segment refused.</summary>
[Fact]
public async Task Raw_reverse_read_on_a_forward_only_backend_serves_adjacent_tie_clusters_within_the_budget()
{
var series = new List<DataValueSnapshot>();
var v = 0;
for (var c = 0; c < 3; c++)
{
var t = Epoch.AddSeconds(5).AddMilliseconds(c);
for (var i = 0; i < 4; i++) series.Add(new DataValueSnapshot((double)v++, StatusCodes.Good, t, t));
}
var source = new ForwardOnlyHistorianDataSource(series);
var (host, nm, nodeId) = await BootForwardOnlyAsync(source, "FwdAdjacent");
nm.MaxTieClusterOverfetch = 4;
var (values, error, _) = ReadRaw(nm, nodeId, Epoch.AddHours(1), Epoch.AddSeconds(-1), max: 100, inboundCp: null);
error.StatusCode.Code.ShouldBe(StatusCodes.Good);
values.ShouldBe(Enumerable.Range(0, 12).Reverse().Select(i => (double)i));
await host.DisposeAsync();
}
/// <summary>#686 review: a <c>MaxTieClusterOverfetch</c> of 1 (or a misconfigured 0) does not make every reverse read on
/// a forward-only backend refuse — distinct-timestamp data is still served.</summary>
[Theory]
[InlineData(1)]
[InlineData(0)]
public async Task Raw_reverse_read_on_a_forward_only_backend_works_with_a_minimal_budget(int budget)
{
var source = new ForwardOnlyHistorianDataSource(MakeSeries(count: 250, stepSeconds: 1));
var (host, nm, nodeId) = await BootForwardOnlyAsync(source, $"FwdMin{budget}");
nm.MaxTieClusterOverfetch = budget;
var (values, error, _) = ReadRaw(nm, nodeId, DateTime.MinValue, Epoch.AddSeconds(100), max: 3, inboundCp: null);
error.StatusCode.Code.ShouldBe(StatusCodes.Good);
values.ShouldBe([100d, 99d, 98d]);
await host.DisposeAsync();
}
/// <summary>A backend that DECLARES native reverse support is still sent the reverse window — one read.</summary> /// <summary>A backend that DECLARES native reverse support is still sent the reverse window — one read.</summary>
[Fact] [Fact]
public async Task Raw_end_only_read_on_a_reverse_capable_backend_is_one_native_reverse_read() public async Task Raw_end_only_read_on_a_reverse_capable_backend_is_one_native_reverse_read()
@@ -630,7 +786,8 @@ public sealed class NodeManagerHistoryReadPagingTests : IDisposable
} }
/// <summary>Pages a Raw read to completion and returns the union of the pages in arrival order.</summary> /// <summary>Pages a Raw read to completion and returns the union of the pages in arrival order.</summary>
private static Task<List<double>> DrainAsync(OtOpcUaNodeManager nm, NodeId nodeId, DateTime start, DateTime end, uint max) private static Task<List<double>> DrainAsync(OtOpcUaNodeManager nm, NodeId nodeId, DateTime start, DateTime end, uint max,
int maxPages = 40)
{ {
var collected = new List<double>(); var collected = new List<double>();
byte[]? cp = null; byte[]? cp = null;
@@ -642,7 +799,7 @@ public sealed class NodeManagerHistoryReadPagingTests : IDisposable
collected.AddRange(values); collected.AddRange(values);
cp = nextCp; cp = nextCp;
pageCount++; pageCount++;
pageCount.ShouldBeLessThan(40, "paging must terminate, not loop"); pageCount.ShouldBeLessThan(maxPages, "paging must terminate, not loop");
} }
while (cp is not null); while (cp is not null);
return Task.FromResult(collected); return Task.FromResult(collected);
@@ -802,8 +959,10 @@ public sealed class NodeManagerHistoryReadPagingTests : IDisposable
/// the query start carrying the value in effect there (Bad 0 when there is none) — counted against the cap. /// the query start carrying the value in effect there (Bad 0 when there is none) — counted against the cap.
/// </summary> /// </summary>
private sealed class ForwardOnlyHistorianDataSource( private sealed class ForwardOnlyHistorianDataSource(
IReadOnlyList<DataValueSnapshot> series, bool emitBoundaryRow = true) : IHistorianDataSource IReadOnlyList<DataValueSnapshot> series, bool emitBoundaryRow = true, bool boundaryAtCeilingMs = false)
: IHistorianDataSource
{ {
private readonly Lock _gate = new();
private int _readCount; private int _readCount;
private int _reverseWindowCount; private int _reverseWindowCount;
private int _maxRowsReturned; private int _maxRowsReturned;
@@ -812,6 +971,19 @@ public sealed class NodeManagerHistoryReadPagingTests : IDisposable
public int ReverseWindowCount => _reverseWindowCount; public int ReverseWindowCount => _reverseWindowCount;
public int MaxRowsReturned => _maxRowsReturned; public int MaxRowsReturned => _maxRowsReturned;
/// <summary>The earliest query start and latest query end this backend was sent.</summary>
public DateTime MinRequestedStart { get; private set; } = DateTime.MaxValue;
public DateTime MaxRequestedEnd { get; private set; } = DateTime.MinValue;
/// <summary>Reads whose row count equalled the cap (a full, possibly truncated, read).</summary>
public int FullReads { get; private set; }
public void ResetCounts()
{
lock (_gate) FullReads = 0;
Interlocked.Exchange(ref _readCount, 0);
}
public bool SupportsReverseRaw => false; public bool SupportsReverseRaw => false;
public Task<HistorianRead> ReadRawAsync( public Task<HistorianRead> ReadRawAsync(
@@ -825,15 +997,27 @@ public sealed class NodeManagerHistoryReadPagingTests : IDisposable
throw new InvalidOperationException("start_time must be less than or equal to end_time. (InvalidArgument)"); throw new InvalidOperationException("start_time must be less than or equal to end_time. (InvalidArgument)");
} }
lock (_gate)
{
if (startUtc < MinRequestedStart) MinRequestedStart = startUtc;
if (endUtc > MaxRequestedEnd) MaxRequestedEnd = endUtc;
}
static DateTime Ts(DataValueSnapshot s) => s.SourceTimestampUtc ?? DateTime.MinValue; static DateTime Ts(DataValueSnapshot s) => s.SourceTimestampUtc ?? DateTime.MinValue;
var prior = series.LastOrDefault(s => Ts(s) <= startUtc); var prior = series.LastOrDefault(s => Ts(s) <= startUtc);
// AVEVA may stamp its fabricated row at the query start rounded UP to its millisecond resolution.
var stamp = boundaryAtCeilingMs && startUtc.Ticks % TimeSpan.TicksPerMillisecond != 0
? new DateTime(startUtc.Ticks - startUtc.Ticks % TimeSpan.TicksPerMillisecond + TimeSpan.TicksPerMillisecond, DateTimeKind.Utc)
: startUtc;
var rows = new List<DataValueSnapshot>(); var rows = new List<DataValueSnapshot>();
if (emitBoundaryRow) if (emitBoundaryRow)
rows.Add(prior is null rows.Add(prior is null
? new DataValueSnapshot(0d, StatusCodes.BadNoData, startUtc, startUtc) ? new DataValueSnapshot(0d, StatusCodes.BadNoData, stamp, stamp)
: new DataValueSnapshot(prior.Value, prior.StatusCode, startUtc, startUtc)); : new DataValueSnapshot(prior.Value, prior.StatusCode, stamp, stamp));
rows.AddRange(series.Where(s => Ts(s) >= startUtc && Ts(s) <= endUtc)); rows.AddRange(series.Where(s => Ts(s) >= startUtc && Ts(s) <= endUtc));
var page = maxValuesPerNode == 0 ? rows : rows.Take((int)maxValuesPerNode).ToList(); var page = maxValuesPerNode == 0 ? rows : rows.Take((int)maxValuesPerNode).ToList();
if (maxValuesPerNode > 0 && page.Count == maxValuesPerNode)
lock (_gate) FullReads++;
int seen; int seen;
while (page.Count > (seen = Volatile.Read(ref _maxRowsReturned)) while (page.Count > (seen = Volatile.Read(ref _maxRowsReturned))
&& Interlocked.CompareExchange(ref _maxRowsReturned, page.Count, seen) != seen) { } && Interlocked.CompareExchange(ref _maxRowsReturned, page.Count, seen) != seen) { }
@@ -898,6 +1082,11 @@ public sealed class NodeManagerHistoryReadPagingTests : IDisposable
public void Dispose() { } public void Dispose() { }
} }
private sealed class FixedClock(DateTime now) : TimeProvider
{
public override DateTimeOffset GetUtcNow() => new(now);
}
private async Task<(OpcUaApplicationHost Host, OtOpcUaSdkServer Server)> BootAsync() private async Task<(OpcUaApplicationHost Host, OtOpcUaSdkServer Server)> BootAsync()
{ {
var applicationUri = $"urn:OtOpcUa.HistoryPagingTest:{Guid.NewGuid():N}"; var applicationUri = $"urn:OtOpcUa.HistoryPagingTest:{Guid.NewGuid():N}";