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

fix(historian): #687 review — one bounded re-read instead of failing, raw budget unclamped, exclusive forward end

Browse Source 
- M1/M2: when the 1 ms pre-window margin (boundary row + stored samples, e.g. a
  >16-sample tie cluster at start-1ms / lastTs-1ms) uses up the cap + 16 budget,
  re-read ONCE asking for cap + m + 16 (m = pre-window rows seen); fail
  BadTooManyOperations only if the re-read is still short because of them. The
  raw budget is no longer clamped at the gateway default ceiling (AVEVA always
  emits the boundary row, so page sizes 999,985-1,000,000 always came back
  short); the #688 clamp stays on the per-sample aggregate path only.
- Merge trap: NumValuesPerNode 0 => int.MaxValue, never through the clamp.
- L3: the forward end is exclusive ([start, end), the IHistorianDataSource
  contract); a point read (start == end, the tie-cluster over-fetch) inclusive.
- RequestStartFor never goes before the Unix epoch (#686's walk-back floor).
- L4 (docs): an Uncertain native row for an all-NULL interval is unverified live.
This commit is contained in:
Joseph Doherty
2026-10-12 15:19:34 -04:00
parent 6796ba4126
commit 8114f13a28
5 changed files with 202 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianDataSource.cs
+56 -23
View File
@@ -123,8 +123,8 @@ 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).
/// The stored samples of <c>[startUtc, endUtc)</c> (a point read <c>start == end</c>: of that instant), oldest first,
/// at most <paramref name="maxValuesPerNode"/> (<c>0</c> = all) — without AVEVA's synthetic rows (#687).
/// </summary>
/// <remarks>
/// <para>
@@ -133,48 +133,81 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
/// 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
/// therefore uses the same request start 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).
/// sample and the no-data boundary row look the same on the wire (both carry a value 0 with a Bad quality). The
/// forward end is exclusive (the interface contract; the gateway's own end is inclusive), a point read inclusive.
/// </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.
/// <b>Row budget.</b> <c>0</c> ("all values") asks for <see cref="int.MaxValue"/> (the client package and the
/// gateway reject <c>max_values &lt;= 0</c>, #686). A cap asks for <c>cap + </c>
/// <see cref="RawAggregateReader.PreWindowAllowance"/> rows — NOT clamped at the gateway's default ceiling (#688
/// clamps only the per-sample aggregate budget): AVEVA always emits the boundary row, so a clamped budget would make
/// every page size within the allowance of the ceiling come back short. A strict gateway refuses an over-ceiling
/// budget instead (⇒ <see cref="HistoryReadLimitExceededException"/>, <c>BadTooManyOperations</c>).
/// </para>
/// <para>
/// <b>Completeness.</b> A full page must hold exactly the cap of stored samples — a short page ends the client's
/// paging. A read is complete when the stream came back short, filled the cap, or reached the window end. If rows
/// BEFORE the window (the boundary row and stored samples inside the 1 ms margin, e.g. a tie cluster at
/// <c>start − 1 ms</c>) used up the budget, the window is read ONCE more asking for <c>cap + m + </c>
/// <see cref="RawAggregateReader.PreWindowAllowance"/> (m = the pre-window rows the first read saw); only if that
/// re-read is still short because of pre-window rows does the read fail with
/// <see cref="HistoryReadLimitExceededException"/>. Every stream is 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).
// The gateway seam caps with an int; OPC UA hands us a uint, so clamp to int range.
var cap = (int)Math.Min(maxValuesPerNode, int.MaxValue);
var budget = cap == 0 ? int.MaxValue : (int)Math.Min(int.MaxValue, (long)cap + RawAggregateReader.PreWindowAllowance);
if (cap == 0)
return (await ReadRawPassAsync(fullReference, startUtc, endUtc, cap, int.MaxValue, cancellationToken)
.ConfigureAwait(false)).Samples;
var first = await ReadRawPassAsync(fullReference, startUtc, endUtc, cap,
(int)Math.Min(int.MaxValue, (long)cap + RawAggregateReader.PreWindowAllowance), cancellationToken).ConfigureAwait(false);
if (first.Complete) return first.Samples;
// The margin crowded the window out: one bounded re-read with room for the pre-window rows already seen.
var second = await ReadRawPassAsync(fullReference, startUtc, endUtc, cap,
(int)Math.Min(int.MaxValue, (long)cap + first.PreWindowRows + RawAggregateReader.PreWindowAllowance),
cancellationToken).ConfigureAwait(false);
if (second.Complete) return second.Samples;
throw new HistoryReadLimitExceededException(cap,
$"The raw read of '{fullReference}' could not be read completely: more than {second.PreWindowRows} rows lie in " +
$"the {RawAggregateReader.BoundaryMargin.TotalMilliseconds} ms before the window and exhausted the row budget.");
}
/// <summary>One forward raw read of a window, drained, with its pre-window row count and whether it is complete.</summary>
private async Task<(List<HistorianSample> Samples, int PreWindowRows, bool Complete)> ReadRawPassAsync(
string fullReference, DateTime startUtc, DateTime endUtc, int cap, int budget, CancellationToken cancellationToken)
{
var samples = new List<HistorianSample>();
long received = 0;
var preWindow = 0;
var reachedEnd = false;
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 (RawAggregateReader.IsWindowSample(sample, startUtc, endUtc))
{
if (cap == 0 || samples.Count < cap) samples.Add(sample);
}
else if (sample.Timestamp is { } ts && ts.ToDateTime() < startUtc) preWindow++;
else if (sample.Timestamp is not null) reachedEnd = true;
}
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;
var complete = received < budget || samples.Count >= cap || reachedEnd || cap == 0;
return (samples, preWindow, complete);
}
/// <inheritdoc />
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/RawAggregateReader.cs
+16 -9
View File
@@ -80,25 +80,32 @@ internal static class RawAggregateReader
/// 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"/>.
/// (#687) decide what the request start is. Never before <see cref="ProbeFloor"/> (the Unix epoch — the start of
/// the historian's time encoding, the same floor the server's reverse walk-back uses, #686): a window starting less
/// than one margin after it is requested from the epoch itself.
/// </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;
windowStartUtc - ProbeFloor > BoundaryMargin ? windowStartUtc - BoundaryMargin
: windowStartUtc > ProbeFloor ? ProbeFloor
: 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.
/// True when a row of a forward raw stream requested from <see cref="RequestStartFor"/> is a stored sample of the raw
/// window <c>[windowStartUtc, windowEndUtc)</c> — the <see cref="IHistorianDataSource.ReadRawAsync"/> contract
/// (exclusive end, as on the SQL backends; the gateway's own <c>ReadRaw</c> end is inclusive) — or, for a point read
/// (<c>start == end</c>, the server's tie-cluster over-fetch), of that one instant. A row before the start 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>
/// <param name="windowEndUtc">The window's end: exclusive, or inclusive when it equals the 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;
public static bool IsWindowSample(HistorianSample raw, DateTime windowStartUtc, DateTime windowEndUtc) =>
raw.Timestamp is { } ts && ts.ToDateTime() is var t && t >= windowStartUtc
&& (t < windowEndUtc || (t == windowEndUtc && windowStartUtc == windowEndUtc));
/// <summary>True for the aggregates this reader serves: the per-sample families of <see cref="AggregateFamilies"/>.</summary>
/// <param name="aggregate">Requested aggregate.</param>