From cf1af557207016b71f92094c369a94f5c32cc0df Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Mon, 12 Oct 2026 15:11:21 -0400 Subject: [PATCH] =?UTF-8?q?fix(historian):=20#686=20review=20=E2=80=94=20b?= =?UTF-8?q?ounded,=20density-aware=20reverse=20raw=20walk;=20uncapped=20Ga?= =?UTF-8?q?teway=20raw=20reads?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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). --- .../GatewayHistorianDataSource.cs | 17 +- .../HistoryPaging.cs | 7 +- .../OtOpcUaNodeManager.cs | 41 ++-- .../ReverseRawReader.cs | 226 ++++++++++++------ .../ZB.MOM.WW.OtOpcUa.OpcUaServer.csproj | 2 + .../Historian/ServerHistorianOptions.cs | 9 +- .../GatewayRawDirectionTests.cs | 148 +++++++++++- ...NodeManagerHistoryReadEventsPagingTests.cs | 4 +- .../NodeManagerHistoryReadPagingTests.cs | 209 +++++++++++++++- 9 files changed, 554 insertions(+), 109 deletions(-) diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianDataSource.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianDataSource.cs index 5fcdf0c5..f84df290 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianDataSource.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianDataSource.cs @@ -90,8 +90,9 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis try { - // The gateway seam caps with an int; OPC UA hands us a uint, so clamp to int range. - var maxValues = (int)Math.Min(maxValuesPerNode, int.MaxValue); + // The gateway seam caps with an int; OPC UA hands us a uint, so clamp to int range. 0 ("all values") is sent + // 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(); await foreach (var sample in _client .ReadRawAsync(fullReference, startUtc, endUtc, maxValues, cancellationToken) @@ -104,6 +105,18 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis RecordOutcome(success: true, error: 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) { RecordReadFailure(ex); diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/HistoryPaging.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/HistoryPaging.cs index 667d16a1..fe825f0e 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/HistoryPaging.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/HistoryPaging.cs @@ -51,6 +51,10 @@ namespace ZB.MOM.WW.OtOpcUa.OpcUaServer; /// (EndUtc, NextStartUtc]. A resumed events read takes its direction from here, never from the resume /// request's details. Raw keeps deriving it from the window (NextStartUtc > EndUtc). /// +/// +/// 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. +/// internal sealed record HistoryContinuationState( string Tagname, DateTime NextStartUtc, @@ -60,7 +64,8 @@ internal sealed record HistoryContinuationState( HistoryCursorKind Kind = HistoryCursorKind.Raw, long LastModifiedId = 0, DateTime WindowStartUtc = default, - bool Reverse = false); + bool Reverse = false, + long SegmentWidthTicks = 0); /// The HistoryRead kind that issued a . internal enum HistoryCursorKind diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs index 77712df5..09b486f8 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs @@ -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 /// paging over-fetches up to MaxTieClusterOverfetch + 1 ties at that single timestamp (a /// start == end read) and slices through them via . - /// A cluster strictly larger than this still surfaces BadHistoryOperationUnsupported for that - /// node (the absurd-burst backstop). Mirrors the configured + /// A cluster strictly larger than this still fails that node BadTooManyOperations (the absurd-burst + /// backstop; BadHistoryOperationUnsupported before #686). Also the per-read row budget of a reverse raw read + /// walked back from forward reads (). Mirrors the configured /// ServerHistorianOptions.MaxTieClusterOverfetch; the Host sets it at StartAsync. The /// default (65536) survives when the historian section is absent. /// @@ -5359,8 +5360,8 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 "OtOpcUaNodeManager: HistoryReadEvents tie cluster at {StartUtc:O} for source '{Source}' has {Count} events, " + "exceeding MaxTieClusterOverfetch={Max}; cannot page within it. Increase MaxTieClusterOverfetch.", startUtc, sourceName, cluster.Count, MaxTieClusterOverfetch); - errors[handle.Index] = StatusCodes.BadHistoryOperationUnsupported; - results[handle.Index] = new SdkHistoryReadResult { StatusCode = StatusCodes.BadHistoryOperationUnsupported }; + errors[handle.Index] = StatusCodes.BadTooManyOperations; + results[handle.Index] = new SdkHistoryReadResult { StatusCode = StatusCodes.BadTooManyOperations }; return; } @@ -6183,6 +6184,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 { DateTime startUtc; var boundarySkip = 0; + long segmentWidth = 0; // #686: the walk-back's width, carried page to page string tagname; if (inboundCp is { Length: > 0 }) @@ -6203,6 +6205,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 boundarySkip = state.BoundarySkipCount; endUtc = state.EndUtc; numValuesPerNode = state.NumValuesPerNode; + segmentWidth = state.SegmentWidthTicks; } 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 // start > end, which failed the node and counted as a backend fault). var source = HistorianDataSource; - var sourceResult = startUtc > endUtc && !source.SupportsReverseRaw - ? new HistorianRead( - await ReverseRawReader.ReadAsync( - source, tagname, startUtc, endUtc, numValuesPerNode, - (int)Math.Min(int.MaxValue, (long)Math.Max(1, MaxTieClusterOverfetch) + 1), ct) - .ConfigureAwait(false), - ContinuationPoint: null) - : await source + HistorianRead sourceResult; + if (startUtc > endUtc && !source.SupportsReverseRaw) + { + var page = await ReverseRawReader.ReadAsync( + source, tagname, startUtc, endUtc, numValuesPerNode, MaxTieClusterOverfetch, + Clock.GetUtcNow().UtcDateTime, segmentWidth, ct) + .ConfigureAwait(false); + segmentWidth = page.WidthHintTicks; + sourceResult = new HistorianRead(page.Samples, ContinuationPoint: null); + } + else + { + sourceResult = await source .ReadRawAsync(tagname, startUtc, endUtc, numValuesPerNode, ct) .ConfigureAwait(false); + } 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, " + "exceeding MaxTieClusterOverfetch={Max}; cannot page within it. Increase MaxTieClusterOverfetch.", startUtc, tagname, cluster.Count, MaxTieClusterOverfetch); - errors[handle.Index] = StatusCodes.BadHistoryOperationUnsupported; - results[handle.Index] = new SdkHistoryReadResult { StatusCode = StatusCodes.BadHistoryOperationUnsupported }; + errors[handle.Index] = StatusCodes.BadTooManyOperations; + results[handle.Index] = new SdkHistoryReadResult { StatusCode = StatusCodes.BadTooManyOperations }; return; } @@ -6315,7 +6324,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 if (nextStartUtc is { } resumeAt) { var clusterState = new HistoryContinuationState( - tagname, resumeAt, endUtc, nextSkip, numValuesPerNode); + tagname, resumeAt, endUtc, nextSkip, numValuesPerNode, SegmentWidthTicks: segmentWidth); clusterCp = _historyContinuationStore.Save(session, clusterState); } @@ -6338,7 +6347,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 { HistoryPaging.ComputeResumeCursor(samples, out var nextStart, out var skip); 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. // Built AFTER historyData so a failure projecting samples can never orphan a stored cursor. outboundCp = _historyContinuationStore.Save(session, nextState); diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ReverseRawReader.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ReverseRawReader.cs index feadf0e0..3ebea102 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ReverseRawReader.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ReverseRawReader.cs @@ -2,6 +2,11 @@ using ZB.MOM.WW.OtOpcUa.Core.Abstractions; namespace ZB.MOM.WW.OtOpcUa.OpcUaServer; +/// One walked-back reverse raw page (). +/// At most the requested cap, newest first. +/// The segment width the walk ended on — the next page's starting width (0 = none). +internal readonly record struct ReverseRawPage(IReadOnlyList Samples, long WidthHintTicks); + /// /// Gitea #686: serves a Part 11 reverse raw read — (lo, hi] newest-first, capped — on a historian backend that can /// only read forward ( = false: the Gateway, whose @@ -9,112 +14,191 @@ namespace ZB.MOM.WW.OtOpcUa.OpcUaServer; /// /// /// -/// Walk back in segments. Starting at hi, read one segment (segLo, segHi] FORWARD, keep it -/// newest-first, and step to the next older segment until the cap is met or lo is reached. A segment starts -/// at 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. +/// Uncapped (maxValues = 0): ONE forward read of the whole window, filtered and reversed in memory — the +/// same rows, round trips and memory as an uncapped forward read (and as a native backend's uncapped reverse read). /// /// -/// Never answer from a truncated read. Each segment read asks for at most segmentCap rows (the server's -/// MaxTieClusterOverfetch + 1 over-fetch bound). A read that comes back FULL may have been cut short — its -/// newest rows are exactly the ones a reverse read needs — so it is discarded and the segment re-read at a quarter of -/// the width. A full read of a segment already one timestamp-resolution wide (a single-timestamp tie cluster larger -/// than the bound) cannot be served and throws +/// Capped: walk back in segments. Starting at hi, read one segment (segLo, segHi] FORWARD, keep +/// it newest-first, and step to the next older segment until the cap is met or lo is reached. The first +/// segment is wide, or the width the previous page of the same read ended on (carried in +/// the continuation point). After each read the next width is re-aimed at about half the row budget from the density +/// 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. +/// +/// +/// Never answer from a truncated read. The row budget is M = max(1, MaxTieClusterOverfetch); a segment +/// read asks for M + 2 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 segHi. 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 2M + 4 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 M samples — a segment already +/// one timestamp-resolution wide that still does not fit — is refused: /// (⇒ BadTooManyOperations, a request limit, never a backend fault). /// /// -/// Edges, for any forward contract. A segment is requested as [segLo, segHi + resolution] and filtered to -/// (segLo, segHi] on the samples' own timestamps, so it is exact whether the backend's forward end is exclusive -/// (the interface contract) or inclusive (the gateway), and whether it truncates its bounds (QuestDB-style). The -/// filter also drops the rows an AVEVA Full read fabricates at the query start (the boundary / no-data row, -/// #687) and at the query end. Adjacent segments share no timestamp, so nothing is returned twice. +/// Edges. A segment is requested as [⌊segLo⌋, ⌈segHi + 1 tick⌉], both rounded outward to the millisecond +/// (the gateway's / AVEVA's resolution), and filtered to (segLo, segHi] on the samples' own timestamps. That is +/// exact whether the backend's forward end is exclusive (the interface contract) or inclusive (the gateway) and +/// whether it truncates its bounds (QuestDB-style); and because the request start is already on a millisecond, a row +/// 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. /// /// /// Ties 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. /// /// -/// Floor. The walk stops at (the Unix epoch; its earliest request bound is one tick -/// before it): the AVEVA historian holds nothing earlier, and an end-only read's open lower bound (year 1) would -/// otherwise cost dozens of empty reads and send a bound outside the historian's time encoding. Samples before the -/// floor are not returned by a walked-back read; a window entirely before it is empty. +/// Bounds. The walk never asks for data after now + (one day — the historian stores +/// nothing further ahead; a StartTime = DateTime.MaxValue read would otherwise send centuries-ahead query +/// starts), nor before (the Unix epoch — AVEVA holds nothing earlier, and an end-only read's year-1 +/// 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. /// /// internal static class ReverseRawReader { - /// The first segment's width. + /// The first segment's width when no hint is carried. internal static readonly TimeSpan InitialSegment = TimeSpan.FromMinutes(1); - /// The earliest instant a walk reads (see the remarks). + /// The walk's exclusive lower limit (see the remarks). internal static readonly DateTime Floor = DateTime.UnixEpoch; + /// How far past "now" the walk starts at most (the same horizon as the Gateway's raw-aggregate look-ahead). + internal static readonly TimeSpan LookAhead = TimeSpan.FromDays(1); + + /// Request bounds are rounded outward to this (the gateway / AVEVA timestamp resolution). + private const long Alignment = TimeSpan.TicksPerMillisecond; + /// Reads (lo, hi] newest-first from forward reads. /// A backend that reads forward. /// Historian tag. /// The later (inclusive) bound — the request's StartTime or a reverse resume point. /// The earlier (exclusive) bound. - /// Cap; 0 = every sample in the window. - /// Rows one segment read may return before it is treated as truncated (≥ 2). + /// Cap; 0 = every sample in the window (one forward read). + /// Row budget per segment read (MaxTieClusterOverfetch; values below 1 count as 1). + /// Now, for the look-ahead clamp. + /// Starting segment width (the previous page's ), 0 = none. /// Cancellation. - /// At most samples, newest first. - /// A single-timestamp tie cluster exceeds the segment cap. - public static async Task> ReadAsync( - IHistorianDataSource source, string tagname, DateTime hi, DateTime lo, uint maxValues, int segmentCap, - CancellationToken ct) + /// The page, newest first, and the width to start the next page with. + /// One timestamp holds more than the row budget. + public static async Task ReadAsync( + IHistorianDataSource source, string tagname, DateTime hi, DateTime lo, uint maxValues, int maxTieCluster, + DateTime nowUtc, long widthHintTicks, CancellationToken ct) { 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(); - 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 - // in the window (and the row a backend fabricates at that query start is still filtered out). - var stopTicks = lo.Ticks < Floor.Ticks ? Floor.Ticks - 1 : lo.Ticks; + var horizon = nowUtc.Ticks > DateTime.MaxValue.Ticks - LookAhead.Ticks ? DateTime.MaxValue.Ticks : nowUtc.Ticks + LookAhead.Ticks; + var hiTicks = Math.Min(hi.Ticks, horizon); + 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(); - var segLoTicks = segHiTicks - stopTicks <= width ? stopTicks : segHiTicks - width; - var requestEndTicks = segHiTicks > DateTime.MaxValue.Ticks - resolution - ? DateTime.MaxValue.Ticks - : 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) + // One forward read of the whole window, reversed — the uncapped forward read's own cost and bound. + var all = (await source.ReadRawAsync(tagname, Utc(AlignDown(stopTicks, granularity)), + Utc(AlignUp(hiTicks + 1, granularity)), 0, ct).ConfigureAwait(false)).Samples; + for (var i = all.Count - 1; i >= 0; i--) { - // Possibly truncated: its newest rows — the ones needed — may be missing. Narrow and re-read. - if (width <= resolution) - 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 ts = Ticks(all[i]); + if (ts > stopTicks && ts <= hiTicks) result.Add(all[i]); } - var inSegment = 0; - 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 new ReverseRawPage(result, widthHintTicks); } - 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 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; } } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ZB.MOM.WW.OtOpcUa.OpcUaServer.csproj b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ZB.MOM.WW.OtOpcUa.OpcUaServer.csproj index 52dc5e0d..8d8ae722 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ZB.MOM.WW.OtOpcUa.OpcUaServer.csproj +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ZB.MOM.WW.OtOpcUa.OpcUaServer.csproj @@ -35,6 +35,8 @@ + + diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ServerHistorianOptions.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ServerHistorianOptions.cs index eea6046f..15b08763 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ServerHistorianOptions.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ServerHistorianOptions.cs @@ -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 /// MaxTieClusterOverfetch + 1 ties at that single timestamp (a start == end read) and /// pages through them; this bounds how large a single-timestamp burst the server will buffer in - /// memory before it gives up and surfaces BadHistoryOperationUnsupported for that node. The - /// default (65536) comfortably covers any realistic same-millisecond burst. + /// memory before it gives up and surfaces BadTooManyOperations for that node (#686; was + /// BadHistoryOperationUnsupported). 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 BadTooManyOperations. The default (65536) comfortably covers + /// any realistic same-millisecond burst; a value below 1 is treated as 1. /// 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, // so a zero/negative value is harmless (and noise-free) when the historian is disabled. 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 // (these gate the per-node fan-out / batch limiter that only run when a real data source is wired). if (HistoryReadBatchParallelism <= 0) diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayRawDirectionTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayRawDirectionTests.cs index 515ec866..d8f0b4eb 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayRawDirectionTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayRawDirectionTests.cs @@ -1,7 +1,12 @@ +using Google.Protobuf.WellKnownTypes; +using Grpc.Core; using Microsoft.Extensions.Logging.Abstractions; using Shouldly; 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.OpcUaServer; 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 ReadRaw streams ascending only and rejects start > end with /// InvalidArgument, 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 -/// 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. /// public sealed class GatewayRawDirectionTests { 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.Instance); + [Fact] public void The_gateway_source_declares_it_cannot_read_raw_newest_first() { - IHistorianDataSource ds = new GatewayHistorianDataSource( - new FakeHistorianGatewayClient(), NullLogger.Instance); + IHistorianDataSource ds = Source(new FakeHistorianGatewayClient()); 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() { var fake = new FakeHistorianGatewayClient(); - IHistorianDataSource ds = new GatewayHistorianDataSource(fake, NullLogger.Instance); + IHistorianDataSource ds = Source(fake); var end = new DateTime(2026, 10, 12, 18, 7, 45, DateTimeKind.Utc); await Should.ThrowAsync(() => @@ -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); fake.ReadRawCallCount.ShouldBe(1); } + + /// + /// An uncapped raw read (NumValuesPerNode = 0, "all values") is sent as an unbounded max_values. The + /// client package and the gateway both reject max_values <= 0, so forwarding 0 failed every uncapped raw + /// read — forward, and the one-shot read an uncapped reverse read is served from (#686 review). + /// + [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(() => 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(() => 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 }; + + /// + /// A gateway over stored samples (value i at T0 + i s, i = 0..299) shaped like AVEVA: ReadRaw rejects + /// start > end and max_values <= 0, returns [start, end] ascending capped at + /// max_values, and first emits a row stamped at the query start carrying the value in effect there + /// (Quality = 133), or a NULL row (OpcQuality = 0, value 0) when there is none. + /// + 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(); + 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); + } } diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadEventsPagingTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadEventsPagingTests.cs index b0e69ec3..17c446ac 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadEventsPagingTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadEventsPagingTests.cs @@ -69,7 +69,7 @@ public sealed class NodeManagerHistoryReadEventsPagingTests : IDisposable var p1 = ReadEvents(nm, node, max: 3, inboundCp: null); p1.Cp.ShouldNotBeNull(); 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(); } @@ -283,7 +283,7 @@ public sealed class NodeManagerHistoryReadEventsPagingTests : IDisposable var p1 = ReadEvents(nm, node, max: 3, inboundCp: null, end: Early, start: Late); p1.Cp.ShouldNotBeNull(); ReadEvents(nm, node, max: 3, inboundCp: p1.Cp, end: Early, start: Late).Error.StatusCode.Code - .ShouldBe(StatusCodes.BadHistoryOperationUnsupported); + .ShouldBe(StatusCodes.BadTooManyOperations); await host.DisposeAsync(); } diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadPagingTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadPagingTests.cs index 8ed393ff..05eaa14b 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadPagingTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadPagingTests.cs @@ -253,8 +253,9 @@ public sealed class NodeManagerHistoryReadPagingTests : IDisposable } /// The absurd-burst backstop is preserved: a tie cluster STRICTLY larger than the configured - /// still surfaces - /// BadHistoryOperationUnsupported for that node rather than buffering an unbounded burst. With + /// still fails that node rather than buffering an unbounded + /// burst — BadTooManyOperations, the request-limit code the reverse walk-back uses too (#686; was + /// BadHistoryOperationUnsupported). With /// the bound set to 3 and a 5-way tie cluster, the resume read that hits the cluster fails loudly. [Fact] 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 // silent GoodNoData that would drop the un-emitted ties. 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(); cp2.ShouldBeNull(); @@ -531,6 +532,7 @@ public sealed class NodeManagerHistoryReadPagingTests : IDisposable values.ShouldBe(Enumerable.Range(10, 191).Reverse().Select(i => (double)i)); cp.ShouldBeNull(); 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(); } @@ -564,7 +566,7 @@ public sealed class NodeManagerHistoryReadPagingTests : IDisposable var collected = await DrainAsync(nm, nodeId, DateTime.MinValue, Epoch.AddHours(1), max: 100); 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); await host.DisposeAsync(); } @@ -587,7 +589,8 @@ public sealed class NodeManagerHistoryReadPagingTests : IDisposable error.StatusCode.Code.ShouldBe(StatusCodes.Good); values.ShouldBe([2d, 1d]); 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); await host.DisposeAsync(); } @@ -609,6 +612,159 @@ public sealed class NodeManagerHistoryReadPagingTests : IDisposable await host.DisposeAsync(); } + /// #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. + [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(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(); + } + + /// #686 review: a reverse read from StartTime = DateTime.MaxValue walks back from now + 1 day, not from + /// year 9999. + [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(); + } + + /// #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. + [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(); + } + + /// #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. + [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(); + } + + /// #686 review: a tie cluster of EXACTLY MaxTieClusterOverfetch is served — the row the backend + /// fabricates at the query start does not consume the budget. + [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(); + } + + /// #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. + [Fact] + public async Task Raw_reverse_read_on_a_forward_only_backend_serves_adjacent_tie_clusters_within_the_budget() + { + var series = new List(); + 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(); + } + + /// #686 review: a MaxTieClusterOverfetch of 1 (or a misconfigured 0) does not make every reverse read on + /// a forward-only backend refuse — distinct-timestamp data is still served. + [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(); + } + /// A backend that DECLARES native reverse support is still sent the reverse window — one read. [Fact] 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 } /// Pages a Raw read to completion and returns the union of the pages in arrival order. - private static Task> DrainAsync(OtOpcUaNodeManager nm, NodeId nodeId, DateTime start, DateTime end, uint max) + private static Task> DrainAsync(OtOpcUaNodeManager nm, NodeId nodeId, DateTime start, DateTime end, uint max, + int maxPages = 40) { var collected = new List(); byte[]? cp = null; @@ -642,7 +799,7 @@ public sealed class NodeManagerHistoryReadPagingTests : IDisposable collected.AddRange(values); cp = nextCp; pageCount++; - pageCount.ShouldBeLessThan(40, "paging must terminate, not loop"); + pageCount.ShouldBeLessThan(maxPages, "paging must terminate, not loop"); } while (cp is not null); 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. /// private sealed class ForwardOnlyHistorianDataSource( - IReadOnlyList series, bool emitBoundaryRow = true) : IHistorianDataSource + IReadOnlyList series, bool emitBoundaryRow = true, bool boundaryAtCeilingMs = false) + : IHistorianDataSource { + private readonly Lock _gate = new(); private int _readCount; private int _reverseWindowCount; private int _maxRowsReturned; @@ -812,6 +971,19 @@ public sealed class NodeManagerHistoryReadPagingTests : IDisposable public int ReverseWindowCount => _reverseWindowCount; public int MaxRowsReturned => _maxRowsReturned; + /// The earliest query start and latest query end this backend was sent. + public DateTime MinRequestedStart { get; private set; } = DateTime.MaxValue; + public DateTime MaxRequestedEnd { get; private set; } = DateTime.MinValue; + + /// Reads whose row count equalled the cap (a full, possibly truncated, read). + public int FullReads { get; private set; } + + public void ResetCounts() + { + lock (_gate) FullReads = 0; + Interlocked.Exchange(ref _readCount, 0); + } + public bool SupportsReverseRaw => false; public Task 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)"); } + lock (_gate) + { + if (startUtc < MinRequestedStart) MinRequestedStart = startUtc; + if (endUtc > MaxRequestedEnd) MaxRequestedEnd = endUtc; + } + static DateTime Ts(DataValueSnapshot s) => s.SourceTimestampUtc ?? DateTime.MinValue; 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(); if (emitBoundaryRow) rows.Add(prior is null - ? new DataValueSnapshot(0d, StatusCodes.BadNoData, startUtc, startUtc) - : new DataValueSnapshot(prior.Value, prior.StatusCode, startUtc, startUtc)); + ? new DataValueSnapshot(0d, StatusCodes.BadNoData, stamp, stamp) + : new DataValueSnapshot(prior.Value, prior.StatusCode, stamp, stamp)); rows.AddRange(series.Where(s => Ts(s) >= startUtc && Ts(s) <= endUtc)); var page = maxValuesPerNode == 0 ? rows : rows.Take((int)maxValuesPerNode).ToList(); + if (maxValuesPerNode > 0 && page.Count == maxValuesPerNode) + lock (_gate) FullReads++; int seen; while (page.Count > (seen = Volatile.Read(ref _maxRowsReturned)) && Interlocked.CompareExchange(ref _maxRowsReturned, page.Count, seen) != seen) { } @@ -898,6 +1082,11 @@ public sealed class NodeManagerHistoryReadPagingTests : IDisposable public void Dispose() { } } + private sealed class FixedClock(DateTime now) : TimeProvider + { + public override DateTimeOffset GetUtcNow() => new(now); + } + private async Task<(OpcUaApplicationHost Host, OtOpcUaSdkServer Server)> BootAsync() { var applicationUri = $"urn:OtOpcUa.HistoryPagingTest:{Guid.NewGuid():N}";