From 9e47532283ed3a29bc4011c86d46cb6b0a112e47 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Mon, 12 Oct 2026 14:35:20 -0400 Subject: [PATCH] fix(historian): #687 Gateway raw reads drop AVEVA's synthetic start-boundary row MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An AVEVA Full raw read leads with a row stamped at the query start (the value in effect, Quality=133, or a no-data NULL row); it was returned as a stored sample. Raw reads now use the #681 classification (RawAggregateReader.RequestStartFor / IsWindowSample — one authority, by timestamp, since a stored NULL looks the same on the wire): request from 1 ms early with NumValuesPerNode + 16 rows, keep rows at or after StartTime, drain the whole stream. A full page still holds exactly the cap, so #669 paging/resume/tie-cluster reads are unchanged; a margin that swallows the allowance fails the node rather than ending paging early. ReturnBounds is not plumbed on any backend (documented). --- CLAUDE.md | 2 +- docs/Historian.md | 28 +++ .../GatewayHistorianDataSource.cs | 75 ++++++- .../RawAggregateReader.cs | 27 ++- .../GatewayHistorianDataSourceTests.cs | 6 +- .../GatewayRawBoundaryRowTests.cs | 209 ++++++++++++++++++ 6 files changed, 332 insertions(+), 15 deletions(-) create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayRawBoundaryRowTests.cs diff --git a/CLAUDE.md b/CLAUDE.md index eda1de8e..96dafeef 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -1364,7 +1364,7 @@ historian tagname, which is the tag's driver `FullName`. The server dispatches all OPC UA HistoryRead to the registered `IHistorianDataSource` — the `GatewayHistorianDataSource` read client when enabled, else the `NullHistorianDataSource` default -(historized nodes return `GoodNoData` for Raw/AtTime — Processed is `BadAggregateNotSupported` since #682). Supported variants: Raw, Processed +(historized nodes return `GoodNoData` for Raw/AtTime — Processed is `BadAggregateNotSupported` since #682). Supported variants: Raw (⚠️ #687, client-visible: Gateway raw reads no longer return AVEVA's synthetic start-boundary row — asked from 1 ms early with `NumValuesPerNode + 16` rows and filtered by timestamp, the #681 classification — so an empty window is `GoodNoData`, not a fabricated value at StartTime; `ReturnBounds` is ignored on every backend), Processed (backend-dependent: SqlServer/QuestDb serve seventeen; Gateway twelve = `GatewayHistorianDataSource.ServedAggregates` — ⚠️ #681, client-visible: the ten per-sample ones (Average, Count, Min/Max, Range, Start/End/Delta, Min/MaxActualTime) are evaluated from RAW gateway reads by the shared `Sql.Common` evaluator — identical to the SQL backends, Calculated bits, interval-start stamps, #677 reverse; Min/Max are no longer AVEVA modes, the read moves every raw sample of the window (a month of 1 Hz ⇒ `BadTooManyOperations` at the default `MaxTimeWeightedRawRows` cap), and every gateway stream is consumed to completion (an abandoned one makes the gateway evict its shared historian session) — plus TimeAverage/Total via AVEVA modes (interval-start stamped + Calculated; ⚠️ #685: AVEVA's extra pre-window row is dropped ⇒ one row per interval in `[Start, End)`, and an Uncertain (opc 64) interval keeps its value as `UncertainDataSubNormal` — AVEVA's TWA excludes NULL-gap time where the SQL backends bridge it; per-interval values still need gateway fix historiangw#27); the other time-weighted five stay `BadAggregateNotSupported` because the gateway has no prior-sample read for Part 13 bounds; `docs/Historian.md` §Aggregate set), and AtTime over historized variable nodes; Events over alarm-owning equipment-folder event-notifier nodes. Reads are ungated (served from any redundancy node); authorization uses the standard `AccessLevels.HistoryRead` bit set at materialization. diff --git a/docs/Historian.md b/docs/Historian.md index dfe74f86..6353e9ce 100644 --- a/docs/Historian.md +++ b/docs/Historian.md @@ -673,6 +673,32 @@ are disposed when the session closes). Resuming an unknown / evicted / released `BadContinuationPointInvalid`. `releaseContinuationPoints` drops the stored cursors without reading data. +### Gateway raw reads return stored samples only (#687) + +> ⚠️ **Client-visible (#687).** An AVEVA `Full` raw read starts with a synthetic row stamped exactly at the **query start**: the +> value in effect there (AVEVA `Quality = 133`, the previous sample's value and quality — a value never stored at that time) or, +> before the tag's first sample, a NULL row (value 0, Bad, `Quality = 1`). The `Gateway` backend used to return that row as a stored +> sample: a window with no stored sample in it answered one fabricated value stamped at StartTime (live: `[18:06:21, 18:06:39]` ⇒ +> `18:06:21 v=30 Good`), a window before the tag's history answered `v=0 Bad` instead of `GoodNoData`, and a start-only read led +> with the fabricated row. It is now dropped; such windows read `GoodNoData`. + +- **How.** The same classification the raw-evaluated aggregates use (#681, `RawAggregateReader.RequestStartFor` / + `IsWindowSample` — one authority): the gateway is asked from **1 ms before** the window, so the boundary row lands before it, and + only rows stamped at or after StartTime are returned. The decision is by **timestamp**, not by value or quality — a stored NULL + sample looks exactly like the no-data boundary row on the wire (value 0, Bad), and a stored NULL inside the window is still + returned (Bad), as it is a stored sample. +- **Paging (#669) is unaffected.** The gateway is asked for `NumValuesPerNode + 16` rows (`NumValuesPerNode = 0` stays unlimited), + so the boundary row cannot shorten a page: a full page still holds exactly `NumValuesPerNode` stored samples and carries a + continuation point, and a resumed page (which re-reads from the last returned timestamp inclusive, and the tie-cluster over-fetch + `start == end` read) no longer gets a fabricated row at its resume point. If more than 16 rows lie in the 1 ms before the window + and exhaust that allowance, the page cannot be read completely and the node fails (`HistoryReadLimitExceededException`, not + counted as a backend fault) rather than answering a short page that would end the client's paging early. Every gateway stream is + still drained to completion. +- **`ReturnBounds` is not honoured** — on any backend: the server does not pass it to the data source, so no bounding values are + returned (see [Known limitations](#known-limitations)). The AVEVA boundary row is not used as a bound either: it is stamped at the + query start rather than carrying the time of the real prior sample, so it could only ever be a Part 11 *interpolated* bound, which + this backend does not compute. + ### Continuation-point paging (Events) (#669) A capped event read (`NumValuesPerNode > 0`) pages exactly like Raw, with one difference: the trigger is the **source's @@ -964,6 +990,8 @@ whatever the SDK sets — #682 touches only the i=11201 folder (follow-up candid sample per requested timestamp, so the single-shot backend returns the complete result in one read and there is no "full page ⇒ maybe more" signal to page on. Returning the full result with no continuation point is spec-conformant. +- **`ReturnBounds` is ignored** (every backend): Raw reads return stored values only; no bounding values are added at the window + edges. The node manager does not pass the flag to `IHistorianDataSource`. - **No modified-value history on the `Gateway` backend.** Modified reads (`IsReadModified = true`) are served by `SqlServer` and `QuestDb` ([Modified-value history](#modified-value-history-part-11), #668); the HistorianGateway exposes no modified/edited-history surface, so on `Gateway` (and with no historian) they return `BadHistoryOperationUnsupported`. 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 e62567b5..828fc113 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 @@ -80,6 +80,8 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis /// A reverse window (start > end) is refused with before anything is sent /// and is not recorded as a health failure (Gitea #686): the gateway rejects it InvalidArgument, and the server /// never asks — is false, so it walks a reverse read back with forward reads. + /// Stored samples only (#687 — Part 11 Raw): AVEVA's synthetic start-boundary row is never returned, see + /// . /// public async Task ReadRawAsync( string fullReference, DateTime startUtc, DateTime endUtc, uint maxValuesPerNode, @@ -90,17 +92,8 @@ 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. 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) - .ConfigureAwait(false)) - { - samples.Add(sample); - } - + var samples = await ReadRawWindowAsync(fullReference, startUtc, endUtc, maxValuesPerNode, cancellationToken) + .ConfigureAwait(false); var snapshots = SampleMapper.ToSnapshots(samples); RecordOutcome(success: true, error: null); return new HistoryReadResult(snapshots, ContinuationPoint: null); @@ -117,6 +110,11 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis $"The HistorianGateway refused the row budget of the raw read of '{fullReference}' (its " + "Historian:ReadLimits:MaxValuesCeiling); set NumValuesPerNode or narrow the window."); } + catch (HistoryReadLimitExceededException) + { + // A client-request limit (#687: the pre-window margin swallowed the row budget), not a backend fault. + throw; + } catch (Exception ex) { RecordReadFailure(ex); @@ -124,6 +122,61 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis } } + /// + /// The stored samples of [startUtc, endUtc], oldest first, at most + /// (0 = no cap) — without AVEVA's synthetic rows (#687). + /// + /// + /// + /// An AVEVA Full raw read starts with a row stamped exactly at the query start: the value in effect there + /// (Quality = 133, the prior sample's value — never stored at that time) or, before the tag's first sample, + /// a NULL row (value 0, Bad). Part 11 Raw with ReturnBounds = false wants stored values only, and the + /// server's paging (#669) counts every returned row toward the page and resumes at the last returned timestamp, + /// so a fabricated row would also be counted as a sample and re-fabricated at every resume point. The read + /// therefore uses the same classification as the raw-evaluated aggregates (#681): it is requested from + /// (1 ms early) and only rows + /// are kept — the decision is by TIMESTAMP, because a stored NULL + /// sample and the no-data boundary row look the same on the wire (both carry a value 0 with a Bad quality). + /// + /// + /// The gateway is asked for extra rows so the boundary row + /// (and any stored sample inside the margin) cannot shorten a page: a full page still holds exactly the cap of + /// stored samples, so the server's "full page ⇒ continuation point" rule is unchanged. If the stream delivers its + /// whole budget yet fewer than the cap belong to the window, the page cannot be read completely; answering a short + /// page would end the client's paging early, so the read fails with + /// instead. The stream is always consumed to completion (never + /// broken out of — a cancelled gateway stream evicts its pooled historian session). ReturnBounds is not + /// plumbed to the data source on any backend, so bounding values are never synthesised here. + /// + /// + private async Task> ReadRawWindowAsync( + string fullReference, DateTime startUtc, DateTime endUtc, uint maxValuesPerNode, + CancellationToken cancellationToken) + { + // The gateway seam caps with an int; OPC UA hands us a uint, so clamp to int range. 0 ("all values") is sent as + // int.MaxValue: the client package and the gateway reject max_values <= 0 (#686 review). + var cap = (int)Math.Min(maxValuesPerNode, int.MaxValue); + var budget = cap == 0 ? int.MaxValue : (int)Math.Min(int.MaxValue, (long)cap + RawAggregateReader.PreWindowAllowance); + var samples = new List(); + long received = 0; + await foreach (var sample in _client + .ReadRawAsync(fullReference, RawAggregateReader.RequestStartFor(startUtc), endUtc, budget, cancellationToken) + .ConfigureAwait(false)) + { + received++; + // Filter while draining — the rest of the (bounded) stream is consumed, never abandoned. + if (RawAggregateReader.IsWindowSample(sample, startUtc) && (cap == 0 || samples.Count < cap)) + samples.Add(sample); + } + + if (cap > 0 && received >= budget && samples.Count < cap) + throw new HistoryReadLimitExceededException(cap, + $"The raw read of '{fullReference}' could not be read completely: more than " + + $"{RawAggregateReader.PreWindowAllowance} rows lie in the {RawAggregateReader.BoundaryMargin.TotalMilliseconds} ms " + + "before the window and exhausted the row budget."); + return samples; + } + /// /// /// diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/RawAggregateReader.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/RawAggregateReader.cs index 64738bca..ae75f362 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/RawAggregateReader.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/RawAggregateReader.cs @@ -74,6 +74,31 @@ internal static class RawAggregateReader /// Rows the earlier-data probe asks for (a possible boundary row plus the first stored sample). internal const int ProbeRows = 2; + /// + /// The ReadRaw start for a window that begins at : + /// earlier, so the AVEVA start-boundary row — stamped exactly at the QUERY start, carrying the value in effect there + /// (Quality = 133) or, with no earlier data, a NULL row (value 0, Bad OPC quality) — lands BEFORE the window + /// (). The one place both the raw-evaluated aggregates (#681) and a raw HistoryRead + /// (#687) decide what the request start is. No margin when the window starts within one margin of + /// . + /// + /// The window's (inclusive) start. + /// The start to request. + public static DateTime RequestStartFor(DateTime windowStartUtc) => + windowStartUtc.Ticks > BoundaryMargin.Ticks ? windowStartUtc - BoundaryMargin : windowStartUtc; + + /// + /// True when a row of a stream requested from is a stored sample of the window that + /// begins at : it is stamped at or after the window start. A row before it is the + /// AVEVA boundary row (whatever its value or quality — a stored NULL looks the same on the wire, so the timestamp is + /// the evidence) or a stored sample inside the margin; a row without a timestamp is never a sample. + /// + /// The gateway row. + /// The window's (inclusive) start. + /// Whether the row belongs to the window. + public static bool IsWindowSample(HistorianSample raw, DateTime windowStartUtc) => + raw.Timestamp is { } ts && ts.ToDateTime() >= windowStartUtc; + /// True for the aggregates this reader serves: the per-sample families of . /// Requested aggregate. /// Whether the aggregate is evaluated from raw samples. @@ -113,7 +138,7 @@ internal static class RawAggregateReader { if (plan.BucketCount == 0) return []; var needsRange = AggregateEvaluator.NeedsDataRange(aggregate, plan); - var requestStart = plan.StartUtc.Ticks > BoundaryMargin.Ticks ? plan.StartUtc - BoundaryMargin : plan.StartUtc; + var requestStart = RequestStartFor(plan.StartUtc); var maxValues = MaxValuesFor(maxWindowRows); var buckets = new RawBucketAccumulator(aggregate == HistoryAggregateType.MaximumActualTime); diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayHistorianDataSourceTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayHistorianDataSourceTests.cs index c5e13159..8149ed0d 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayHistorianDataSourceTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayHistorianDataSourceTests.cs @@ -20,11 +20,13 @@ public sealed class GatewayHistorianDataSourceTests }, }; var ds = new GatewayHistorianDataSource(fake, NullLogger.Instance); - var r = await ds.ReadRawAsync("T", DateTime.UtcNow.AddMinutes(-5), DateTime.UtcNow, 100, TestContext.Current.CancellationToken); + var start = new DateTime(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc); + var r = await ds.ReadRawAsync("T", start, start.AddMinutes(5), 100, TestContext.Current.CancellationToken); Assert.Equal(2, r.Samples.Count); Assert.Equal(0x80000000u, r.Samples[1].StatusCode); // Bad from quality 0 Assert.Equal("T", fake.LastReadRawTag); - Assert.Equal(100, fake.LastReadRawMaxValues); + // #687: asked from 1 ms before the window with room for AVEVA's boundary row (GatewayRawBoundaryRowTests). + Assert.Equal(100 + RawAggregateReader.PreWindowAllowance, fake.LastReadRawMaxValues); } [Fact] diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayRawBoundaryRowTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayRawBoundaryRowTests.cs new file mode 100644 index 00000000..4a95d364 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayRawBoundaryRowTests.cs @@ -0,0 +1,209 @@ +using Google.Protobuf.WellKnownTypes; +using Microsoft.Extensions.Logging.Abstractions; +using Shouldly; +using Xunit; +using ZB.MOM.WW.HistorianGateway.Contracts.Grpc; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests; + +/// +/// #687: a raw HistoryRead must return stored samples only (Part 11, ReturnBounds = false). An AVEVA Full +/// raw read starts with a synthetic row stamped exactly at the QUERY start — the value in effect there +/// (Quality = 133, the prior sample's value and OPC quality) or, with no earlier data, a NULL row. The fake emits +/// the REAL wire shape the live gate captured (2026-10-12): every row carries a value (histsdk puts NumericValue +/// on every numeric row), so a NULL row is NumericValue = 0 + OPC quality 0 / Quality = 1 — the evidence +/// that a row is synthetic is its timestamp (the query start), never "it has no value". A stored NULL/Bad sample looks +/// the same on the wire (q=1 opc=0) and IS a stored sample. +/// +public sealed class GatewayRawBoundaryRowTests +{ + private static readonly CancellationToken Ct = CancellationToken.None; + + /// The live gate's tag HistGW.LiveTest.Gate681 (18:06:40 is a stored NULL). + private static readonly DateTime T0 = new(2026, 10, 12, 18, 6, 0, DateTimeKind.Utc); + + private static readonly (double Sec, double Value, bool Null)[] Gate681 = + [ + (0, 10, false), (5, 20, false), (20, 30, false), (40, 0, true), (50, 5, false), (60, 1, false), (90, 1, false), + (100, 7, false), + ]; + + private static DateTime At(double seconds) => T0.AddTicks((long)(seconds * TimeSpan.TicksPerSecond)); + + private static HistorianSample Stored(DateTime t, double v, bool isNull) => new() + { + Tag = "T", + Timestamp = Timestamp.FromDateTime(t), + NumericValue = isNull ? 0 : v, + OpcQuality = isNull ? 0u : 192u, + Quality = isNull ? 1u : 0u, + }; + + /// + /// An AVEVA-shaped gateway over : ReadRaw over [start, end] ascending, capped at + /// max, led by the boundary row at the query start when nothing is stored exactly there. + /// + private static FakeHistorianGatewayClient Aveva(IReadOnlyList<(DateTime T, double V, bool Null)> data) + { + var sorted = data.OrderBy(d => d.T).ToList(); + return new FakeHistorianGatewayClient + { + RawSource = (start, end, max) => + { + var rows = new List(); + if (sorted.All(d => d.T != start)) + { + var prior = sorted.Where(d => d.T < start).Select(d => ((DateTime T, double V, bool Null)?)d).LastOrDefault(); + HistorianSample boundary; + if (prior is { } p) + { + boundary = Stored(start, p.V, p.Null); + if (!p.Null) boundary.Quality = 133; // copies the prior's quality; a Bad prior stays q=1 opc=0 + } + else + { + boundary = new HistorianSample { Tag = "T", Timestamp = Timestamp.FromDateTime(start), NumericValue = 0, OpcQuality = 0, Quality = 1 }; + } + rows.Add(boundary); + } + rows.AddRange(sorted.Where(d => d.T >= start && d.T <= end).Select(d => Stored(d.T, d.V, d.Null))); + return max > 0 ? rows.Take(max) : rows; + }, + }; + } + + private static FakeHistorianGatewayClient Gate() => Aveva(Gate681.Select(x => (At(x.Sec), x.Value, x.Null)).ToList()); + + private static GatewayHistorianDataSource Source(FakeHistorianGatewayClient fake) => + new(fake, NullLogger.Instance); + + [Fact] + public async Task A_window_holding_no_stored_sample_returns_nothing_not_the_value_in_effect() + { + // Live: Raw [18:06:21, 18:06:39] returned "18:06:21 v=30 Good" — a value never stored at that time. + var r = await Source(Gate()).ReadRawAsync("T", At(21), At(39), 0, Ct); + r.Samples.ShouldBeEmpty(); + } + + [Fact] + public async Task A_window_before_the_first_sample_returns_nothing_not_a_Bad_zero() + { + // Live: Raw [18:05:00, 18:05:30] returned "18:05:00 v=0 Bad" instead of GoodNoData. + var r = await Source(Gate()).ReadRawAsync("T", At(-60), At(-30), 10, Ct); + r.Samples.ShouldBeEmpty(); + } + + [Fact] + public async Task A_start_only_read_starts_at_the_first_stored_sample() + { + var r = await Source(Gate()).ReadRawAsync("T", At(30), new DateTime(DateTime.MaxValue.Ticks, DateTimeKind.Utc), 3, Ct); + r.Samples.Select(s => s.SourceTimestampUtc).ShouldBe(new DateTime?[] { At(40), At(50), At(60) }); + } + + [Fact] + public async Task A_stored_NULL_sample_in_the_window_is_kept_as_the_Bad_sample_it_is() + { + var r = await Source(Gate()).ReadRawAsync("T", At(30), At(55), 0, Ct); + r.Samples.Select(s => s.SourceTimestampUtc).ShouldBe(new DateTime?[] { At(40), At(50) }); + r.Samples[0].StatusCode.ShouldBe(0x80000000u); + } + + [Fact] + public async Task A_stored_sample_exactly_at_the_window_start_is_kept() + { + var r = await Source(Gate()).ReadRawAsync("T", At(20), At(45), 0, Ct); + r.Samples.Select(s => s.SourceTimestampUtc).ShouldBe(new DateTime?[] { At(20), At(40) }); + r.Samples[0].Value.ShouldBe(30d); + } + + [Fact] + public async Task The_raw_request_starts_one_millisecond_early_and_asks_for_the_boundary_allowance() + { + var fake = Gate(); + await Source(fake).ReadRawAsync("T", At(30), At(55), 5, Ct); + fake.RawCalls.Single().ShouldBe((At(30) - RawAggregateReader.BoundaryMargin, At(55), 5 + RawAggregateReader.PreWindowAllowance)); + } + + [Fact] + public async Task An_unlimited_read_stays_unlimited() + { + var fake = Gate(); + await Source(fake).ReadRawAsync("T", At(30), At(55), 0, Ct); + fake.RawCalls.Single().MaxValues.ShouldBe(0); + } + + [Fact] + public async Task A_full_page_holds_exactly_the_cap_and_the_stream_is_drained_not_abandoned() + { + var fake = Gate(); + var r = await Source(fake).ReadRawAsync("T", At(0), At(200), 3, Ct); + r.Samples.Select(s => s.SourceTimestampUtc).ShouldBe(new DateTime?[] { At(0), At(5), At(20) }); + fake.RawStreamsCompleted.ShouldBe(1); + } + + [Theory] + [InlineData(1u)] + [InlineData(2u)] + [InlineData(3u)] + public async Task Server_style_paging_returns_every_stored_sample_once_and_nothing_else(uint cap) + { + // Ties at +60 (three samples) and a stored NULL at +40, over the AVEVA-shaped wire. + var data = Gate681.Select(x => (At(x.Sec), x.Value, x.Null)) + .Concat([(At(60), 2d, false), (At(60), 3d, false)]).ToList(); + var fake = Aveva(data); + var ds = Source(fake); + var start = At(-10); + var end = At(200); + var got = new List(); + var skip = 0; + var cursor = start; + var resume = false; + for (var guard = 0; guard < 100; guard++) + { + // The node manager's paging (OtOpcUaNodeManager.ServeRawPagedAsync + HistoryPaging): a full page ⇒ resume at + // the last timestamp INCLUSIVE dropping the ties already returned; a short page is the last; a resume that + // cannot advance (a tie cluster larger than the cap) over-fetches the cluster with a start == end read. + var page = (await ds.ReadRawAsync("T", cursor, end, cap, Ct)).Samples; + var full = page.Count >= cap; + var samples = resume + ? page.SkipWhile((x, i) => i < skip && x.SourceTimestampUtc == cursor).ToList() + : page.ToList(); + if (resume && full && (samples.Count == 0 || samples[^1].SourceTimestampUtc == cursor)) + { + var cluster = (await ds.ReadRawAsync("T", cursor, cursor, 1000, Ct)).Samples; + var sliceStart = Math.Min(skip, cluster.Count); + var sliceCount = Math.Min((int)cap, cluster.Count - sliceStart); + got.AddRange(cluster.Skip(sliceStart).Take(sliceCount)); + if (sliceStart + sliceCount < cluster.Count) + { + skip = sliceStart + sliceCount; + continue; + } + cursor = cursor.AddTicks(1); + skip = 0; + if (cursor > end) break; + continue; + } + got.AddRange(samples); + if (!full || samples.Count == 0) break; + cursor = samples[^1].SourceTimestampUtc!.Value; + skip = samples.AsEnumerable().Reverse().TakeWhile(x => x.SourceTimestampUtc == cursor).Count(); + resume = true; + } + got.Select(s => s.SourceTimestampUtc).ShouldBe(data.Select(d => (DateTime?)d.Item1).OrderBy(t => t).ToArray()); + } + + [Fact] + public async Task A_margin_that_swallows_the_row_budget_fails_the_node_rather_than_answering_a_short_page() + { + // More stored rows inside the 1 ms margin than the allowance: the page could not be read, and a short page would + // end the client's paging early — refuse it (a request limit, not a backend fault). + var margin = Enumerable.Range(0, 40).Select(i => (At(30).AddTicks(-1 - i), (double)i, false)); + var fake = Aveva(margin.Append((At(35), 1d, false)).ToList()); + var ds = Source(fake); + await Should.ThrowAsync(() => ds.ReadRawAsync("T", At(30), At(40), 5, Ct)); + ds.GetHealthSnapshot().TotalFailures.ShouldBe(0); + fake.RawStreamsCompleted.ShouldBe(1); + } +}