fix(historian): #687 Gateway raw reads drop AVEVA's synthetic start-boundary row

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).
2026-10-12 14:35:20 -04:00
parent 44a03bedc0
commit 9e47532283
6 changed files with 332 additions and 15 deletions
@@ -80,6 +80,8 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
    ///     A reverse window (<c>start &gt; end</c>) is refused with <see cref="NotSupportedException"/> before anything is sent
    ///     and is not recorded as a health failure (Gitea #686): the gateway rejects it <c>InvalidArgument</c>, and the server
    ///     never asks — <see cref="SupportsReverseRaw"/> is <c>false</c>, 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
+    ///     <see cref="ReadRawWindowAsync"/>.
    /// </remarks>
    public async Task<HistoryReadResult> 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<HistorianSample>();
-            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
        }
    }

+    /// <summary>
+    ///     The stored samples of <c>[startUtc, endUtc]</c>, oldest first, at most <paramref name="maxValuesPerNode"/>
+    ///     (<c>0</c> = no cap) — without AVEVA's synthetic rows (#687).
+    /// </summary>
+    /// <remarks>
+    ///     <para>
+    ///         An AVEVA <c>Full</c> raw read starts with a row stamped exactly at the query start: the value in effect there
+    ///         (<c>Quality = 133</c>, 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 <c>ReturnBounds = false</c> 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
+    ///         <see cref="RawAggregateReader.RequestStartFor"/> (1 ms early) and only rows
+    ///         <see cref="RawAggregateReader.IsWindowSample"/> 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).
+    ///     </para>
+    ///     <para>
+    ///         The gateway is asked for <see cref="RawAggregateReader.PreWindowAllowance"/> 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
+    ///         <see cref="HistoryReadLimitExceededException"/> instead. The stream is always consumed to completion (never
+    ///         broken out of — a cancelled gateway stream evicts its pooled historian session). <c>ReturnBounds</c> is not
+    ///         plumbed to the data source on any backend, so bounding values are never synthesised here.
+    ///     </para>
+    /// </remarks>
+    private async Task<List<HistorianSample>> 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<HistorianSample>();
+        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;
+    }
+
    /// <inheritdoc />
    /// <remarks>
    ///     <para>
@@ -74,6 +74,31 @@ internal static class RawAggregateReader
    /// <summary>Rows the earlier-data probe asks for (a possible boundary row plus the first stored sample).</summary>
    internal const int ProbeRows = 2;

+    /// <summary>
+    ///     The <c>ReadRaw</c> start for a window that begins at <paramref name="windowStartUtc"/>: <see cref="BoundaryMargin"/>
+    ///     earlier, so the AVEVA start-boundary row — stamped exactly at the QUERY start, carrying the value in effect there
+    ///     (<c>Quality = 133</c>) or, with no earlier data, a NULL row (value 0, Bad OPC quality) — lands BEFORE the window
+    ///     (<see cref="IsWindowSample"/>). 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
+    ///     <see cref="DateTime.MinValue"/>.
+    /// </summary>
+    /// <param name="windowStartUtc">The window's (inclusive) start.</param>
+    /// <returns>The start to request.</returns>
+    public static DateTime RequestStartFor(DateTime windowStartUtc) =>
+        windowStartUtc.Ticks > BoundaryMargin.Ticks ? windowStartUtc - BoundaryMargin : windowStartUtc;
+
+    /// <summary>
+    ///     True when a row of a stream requested from <see cref="RequestStartFor"/> is a stored sample of the window that
+    ///     begins at <paramref name="windowStartUtc"/>: 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.
+    /// </summary>
+    /// <param name="raw">The gateway row.</param>
+    /// <param name="windowStartUtc">The window's (inclusive) start.</param>
+    /// <returns>Whether the row belongs to the window.</returns>
+    public static bool IsWindowSample(HistorianSample raw, DateTime windowStartUtc) =>
+        raw.Timestamp is { } ts && ts.ToDateTime() >= windowStartUtc;
+
    /// <summary>True for the aggregates this reader serves: the per-sample families of <see cref="AggregateFamilies"/>.</summary>
    /// <param name="aggregate">Requested aggregate.</param>
    /// <returns>Whether the aggregate is evaluated from raw samples.</returns>
@@ -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);