diff --git a/CLAUDE.md b/CLAUDE.md index 18a614fe..d732c194 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 (⚠️ #687, client-visible: Gateway raw reads no longer return AVEVA's synthetic start-boundary row — asked from 1 ms early (epoch-floored) with `NumValuesPerNode + 16` rows (unclamped; 0 ⇒ `int.MaxValue`) and filtered by timestamp — so an empty window is `GoodNoData`, not a fabricated value at StartTime; the forward end is now exclusive (a point read inclusive); pre-window rows that crowd the budget trigger ONE re-read with room for them, else `BadTooManyOperations`; `ReturnBounds` is ignored on every backend; ⚠️ #690: a Gateway Bad row — a stored NULL, which histsdk sends as value 0 — now reads value **null** + Bad on Raw and AtTime, as on the SQL backends), 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 (epoch-floored) with `NumValuesPerNode + 16` rows (unclamped; 0 ⇒ `int.MaxValue`) and filtered by timestamp — so an empty window is `GoodNoData`, not a fabricated value at StartTime; the forward end is now exclusive (a point read inclusive); pre-window rows that crowd the budget trigger ONE re-read with room for them, else `BadTooManyOperations`; `ReturnBounds` is ignored on every backend; ⚠️ #690: a Gateway Bad row — a stored NULL, which histsdk sends as value 0 — now reads value **null** + Bad on Raw and AtTime, as on the SQL backends; ⚠️ #691.3: a strict gateway (`AllowStreamingBeyondCeiling=false`) that refuses a raw budget is re-asked ONCE at the ceiling its refusal names — raw and per-sample reads alike — so an uncapped (`NumValuesPerNode = 0`) read of a small window is served; a ceiling-bounded stream that is not provably complete is `BadTooManyOperations`, never a truncated answer), 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 f15f4a0e..78fce238 100644 --- a/docs/Historian.md +++ b/docs/Historian.md @@ -146,7 +146,7 @@ Processed read is `BadAggregateNotSupported` and no history aggregate functions | `MaxTieClusterOverfetch` | int | `65536` | (Node-manager paging; all backends — a `SqlServer` PK rules out tie clusters, so it is moot there.) Maximum samples the server will fetch in one shot to page through a tie cluster (multiple samples sharing one `SourceTimestamp`). A cluster larger than this ceiling fails `BadTooManyOperations` (#686; was `BadHistoryOperationUnsupported`). Also the per-read row budget of a Gateway reverse raw read (walked back from forward reads, #686): one timestamp holding more samples is refused `BadTooManyOperations`. Values below 1 count as 1. Raise to handle abnormally large tie clusters; the default covers all normal-data cases. | | `HistoryReadBatchParallelism` | int | `4` | Max historized nodes served **concurrently** within one HistoryRead batch (arch-review 03/S3). The SDK's `HistoryRead*` override surface is synchronous, so each arm bridges once at its boundary; this turns per-node serving from sequential (`N × CallTimeout` worst case) into `⌈N / P⌉ × CallTimeout`. Also caps the gateway-load multiplication a single client can command. `≤ 1` serves sequentially. | | `MaxConcurrentHistoryReadBatches` | int | `16` | Process-wide limit on HistoryRead (and HistoryUpdate) batches served **concurrently** (arch-review 03/S3). A flood of history clients degrades to `BadTooManyOperations` rather than exhausting the SDK request-thread pool. With `HistoryReadBatchParallelism` this caps total in-flight gateway reads at `MaxConcurrentHistoryReadBatches × HistoryReadBatchParallelism` (default 16 × 4 = 64). A batch that cannot get a slot within `min(HistoryReadDeadline, 5 s)` fails every handle `BadTooManyOperations`. | -| `MaxTimeWeightedRawRows` | int | `1000000` | `SqlServer` / `QuestDb` (#671), and the `Gateway` per-sample aggregates (#681 — every per-sample read there streams raw samples, so the cap applies to the window's raw rows; see [Aggregate set](#aggregate-set-and-badaggregatenotsupported-285)). The most raw in-window samples one [time-weighted aggregate](#time-weighted-aggregates-part-13-671) read (TimeAverage, TimeAverage2, **Total**, Total2, Interpolative, and — since #680 — **PercentGood/PercentBad**) may stream into the server-side evaluator — that family is **not** pushed down to SQL. The window query is capped at this + 1 rows; one row past the cap fails **that node** with `BadTooManyOperations` (not counted as a backend fault) instead of holding a pooled connection until `HistoryReadDeadline` and returning `BadTimeout`. Bound rows outside the window do not count; per-sample aggregates are not capped on the SQL backends (pushed down). 1 000 000 ≈ 11.5 days of a 1 Hz tag. Non-positive = unbounded (a `Validate()` warning). `Gateway` (#688): the row budget it asks the gateway for (`cap + 18`) is clamped at the gateway's default `MaxValuesCeiling` (1,000,000) while the cap is at or under it, so the default works against a gateway with `AllowStreamingBeyondCeiling=false`; against a gateway with a *lower* ceiling, set this to `ceiling − 18`. | +| `MaxTimeWeightedRawRows` | int | `1000000` | `SqlServer` / `QuestDb` (#671), and the `Gateway` per-sample aggregates (#681 — every per-sample read there streams raw samples, so the cap applies to the window's raw rows; see [Aggregate set](#aggregate-set-and-badaggregatenotsupported-285)). The most raw in-window samples one [time-weighted aggregate](#time-weighted-aggregates-part-13-671) read (TimeAverage, TimeAverage2, **Total**, Total2, Interpolative, and — since #680 — **PercentGood/PercentBad**) may stream into the server-side evaluator — that family is **not** pushed down to SQL. The window query is capped at this + 1 rows; one row past the cap fails **that node** with `BadTooManyOperations` (not counted as a backend fault) instead of holding a pooled connection until `HistoryReadDeadline` and returning `BadTimeout`. Bound rows outside the window do not count; per-sample aggregates are not capped on the SQL backends (pushed down). 1 000 000 ≈ 11.5 days of a 1 Hz tag. Non-positive = unbounded (a `Validate()` warning). `Gateway` (#688): the row budget it asks the gateway for (`cap + 18`) is clamped at the gateway's default `MaxValuesCeiling` (1,000,000) while the cap is at or under it, so the default works against a gateway with `AllowStreamingBeyondCeiling=false`; against a gateway with a *lower* ceiling the refused budget is retried once at the ceiling the gateway names (#691.3), so no change is needed (`ceiling − 18` only saves that round-trip). | | `HistoryReadDeadline` | TimeSpan | `00:01:00` | Per-request wall-clock deadline for a HistoryRead (and HistoryUpdate) batch (arch-review 03/S3). A node still in flight at the deadline surfaces `BadTimeout`, so one request can never hold an SDK request thread longer than this regardless of node count. Non-positive = unbounded (a `Validate()` warning). | | `HistoryUpdateMaxFutureSkew` | TimeSpan | `1.00:00:00` | #674. How far past the server clock a HistoryUpdate (Insert / Replace / Update) item's `SourceTimestamp` may lie; an item beyond `now + skew` is refused **`BadOutOfRange`** by the node manager before any backend sees it (every backend). Guards against a far-future typo (a 2062 timestamp). The live recorder is not bounded. Non-positive disables the bound (no warning — a deliberate choice). See [HistoryUpdate](#historyupdate). | @@ -664,8 +664,9 @@ backend: > ⚠️ **Also fixed with #686: uncapped raw reads on Gateway.** `NumValuesPerNode = 0` ("all values") used to be forwarded as > `max_values = 0`, which the client package and the gateway reject — every uncapped raw read (forward too) failed the node > and counted as a backend failure. It is now sent unbounded (`int.MaxValue`); a gateway configured with -> `AllowStreamingBeyondCeiling=false` refuses that over its `MaxValuesCeiling`, which is answered **`BadTooManyOperations`** -> (not a backend fault). A cancelled read (client cancel / `HistoryReadDeadline`) is no longer counted as a backend failure. +> `AllowStreamingBeyondCeiling=false` refuses that over its `MaxValuesCeiling`; since #691.3 the read is retried once at the +> ceiling the refusal names and answered when provably complete, else **`BadTooManyOperations`** (not a backend fault) — see +> [#687](#gateway-raw-reads-return-stored-samples-only-687) "Strict gateway". A cancelled read (client cancel / `HistoryReadDeadline`) is no longer counted as a backend failure. Continuation points are bound to the OPC UA session (the SDK's `ServerConfiguration.MaxHistoryContinuationPoints` cap, default 100, with oldest-eviction; points @@ -701,8 +702,17 @@ data. - **Row budget.** `NumValuesPerNode = 0` ("all values") asks the gateway for an unbounded `max_values` (`int.MaxValue` — the client package and the gateway reject 0, #686). A cap asks for `NumValuesPerNode + 16` 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 from 999,985 to 1,000,000 come back short. A gateway with `AllowStreamingBeyondCeiling=false` refuses an - over-ceiling budget instead ⇒ `BadTooManyOperations` (not a backend fault); a default gateway serves it. + every page size from 999,985 to 1,000,000 come back short. A default gateway serves any budget. +- ⚠️ **Strict gateway (#691.3, client-visible).** A gateway with `AllowStreamingBeyondCeiling=false` refuses a budget over its + `MaxValuesCeiling` (before streaming) and names the ceiling in the refusal; the gateway has no RPC that reports it. The read is + then **re-issued once with the ceiling as the budget** — nothing is cached, so a reconfigured gateway is honoured on the next + read. So an uncapped read (`NumValuesPerNode = 0`, which used to be refused outright on a strict gateway, even for a 10-row + window) and a page size within 16 of the ceiling are now served when the window fits. A ceiling-bounded read is answered only + when it is **provably complete** — the stream came back short of the ceiling, or reached `EndTime` (or, capped, filled the page); + a stream that delivered the whole ceiling short of `EndTime` may be truncated and fails **`BadTooManyOperations`** (not a backend + fault) — set `NumValuesPerNode` to page, or narrow the window. A refusal whose text carries no number falls back to the gateway's + default ceiling (1,000,000). Every stream is still drained. The #686 reverse walk-back (segment budget `MaxTieClusterOverfetch + 2`) + inherits this: against a ceiling below that budget a dense segment fails `BadTooManyOperations` instead of being narrowed. - **Paging (#669) is unaffected.** 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. A read is complete when the stream came back short, @@ -917,9 +927,11 @@ and evicts its shared historian session (see the HistorianGateway issue text in streaming gateway and keeps its `+ 18`. Past the cap the rest of the (bounded) stream is drained, then the node fails `BadTooManyOperations`; a stream that delivers the whole budget cannot be proven complete and fails the same way (so with the default cap a window of more than ~999,982 rows fails, as it would one row later). The gateway's own refusal of the budget - (`InvalidArgument` "max_values exceeds the server ceiling" — `Historian:ReadLimits:MaxValuesCeiling` lowered below the default, - with `AllowStreamingBeyondCeiling=false`) is also `BadTooManyOperations`, not a backend fault — lower `MaxTimeWeightedRawRows` to - `ceiling − 18` on such a gateway. A raw read's budget is **not** clamped (see [#687](#gateway-raw-reads-return-stored-samples-only-687)). A caller cancellation (client cancel, `HistoryReadDeadline`) is not counted as a backend fault. + (`InvalidArgument` "max_values exceeds the server ceiling of N" — `Historian:ReadLimits:MaxValuesCeiling` lowered below the + default, or an unbounded `MaxTimeWeightedRawRows ≤ 0`, with `AllowStreamingBeyondCeiling=false`) is **retried once with N as the + budget** (#691.3): a window that fits is served, one that delivers all N rows fails `BadTooManyOperations` as above (not a backend + fault). Lowering `MaxTimeWeightedRawRows` to `ceiling − 18` on such a gateway is no longer needed — it only saves the refused + round-trip. A raw read's budget is **not** clamped (see [#687](#gateway-raw-reads-return-stored-samples-only-687)). A caller cancellation (client cancel, `HistoryReadDeadline`) is not counted as a backend fault. - **Bucket statistics** are `RawBucketAccumulator`, the C# statement of the SQL backends' per-sample columns — incl. #684: Average / Minimum / Maximum / Range / *ActualTime over **Good** samples only, the Uncertain extremes feeding only the Min/Max status rule, and an Uncertain String/DateTime sample not refusing the Good-value aggregates. A change to those SQL columns 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 79986db2..09001a6e 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 @@ -144,8 +144,12 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis /// gateway reject max_values <= 0, #686). A cap asks for cap + /// 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 (⇒ , BadTooManyOperations). + /// every page size within the allowance of the ceiling come back short. A strict gateway + /// (AllowStreamingBeyondCeiling=false) refuses an over-ceiling budget and names its ceiling; the read is + /// then re-issued ONCE with the ceiling as budget (#691.3, ) — so an uncapped read + /// of a small window is served — and a ceiling-bounded read that is not provably complete (it delivered the + /// whole ceiling without reaching the window end, or a page short of its cap) fails + /// (BadTooManyOperations), never a truncated answer. /// /// /// Completeness. A full page must hold exactly the cap of stored samples — a short page ends the client's @@ -166,22 +170,42 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis // 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); if (cap == 0) - return (await ReadRawPassAsync(fullReference, startUtc, endUtc, cap, int.MaxValue, cancellationToken) - .ConfigureAwait(false)).Samples; + { + var all = await PassAsync(int.MaxValue).ConfigureAwait(false); + if (all.Result.Complete) return all.Result.Samples; + // Only reachable on a strict gateway (#691.3): the ceiling re-read delivered the whole ceiling without reaching + // the window end — the window may hold more, and "all values" is never answered from a truncated read. + throw new HistoryReadLimitExceededException(cap, + $"The raw read of '{fullReference}' holds more than the HistorianGateway's row ceiling ({all.Budget}, its " + + "Historian:ReadLimits:MaxValuesCeiling with AllowStreamingBeyondCeiling=false); set NumValuesPerNode to page, " + + "or narrow the window."); + } - 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; + var asked = (int)Math.Min(int.MaxValue, (long)cap + RawAggregateReader.PreWindowAllowance); + var first = await PassAsync(asked).ConfigureAwait(false); + if (first.Result.Complete) return first.Result.Samples; + + // A strict gateway's ceiling already bounded that read (#691.3): a larger budget would only be refused again. + if (first.Budget < asked) + throw new HistoryReadLimitExceededException(cap, + $"The raw read of '{fullReference}' could not be read completely within the HistorianGateway's row ceiling " + + $"({first.Budget}, its Historian:ReadLimits:MaxValuesCeiling with AllowStreamingBeyondCeiling=false); lower " + + "NumValuesPerNode or narrow the window."); // 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; + var second = await PassAsync( + (int)Math.Min(int.MaxValue, (long)cap + first.Result.PreWindowRows + RawAggregateReader.PreWindowAllowance)) + .ConfigureAwait(false); + if (second.Result.Complete) return second.Result.Samples; throw new HistoryReadLimitExceededException(cap, - $"The raw read of '{fullReference}' could not be read completely: more than {second.PreWindowRows} rows lie in " + + $"The raw read of '{fullReference}' could not be read completely: more than {second.Result.PreWindowRows} rows lie in " + $"the {RawAggregateReader.BoundaryMargin.TotalMilliseconds} ms before the window and exhausted the row budget."); + + // One drained pass; a strict gateway's refusal of the budget is retried once at the ceiling it names (#691.3). + Task<((List Samples, int PreWindowRows, bool Complete) Result, int Budget)> PassAsync(int budget) => + GatewayReadLimits.ReadWithCeilingRetryAsync(budget, + b => ReadRawPassAsync(fullReference, startUtc, endUtc, cap, b, cancellationToken)); } /// One forward raw read of a window, drained, with its pre-window row count and whether it is complete. @@ -206,7 +230,9 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis else if (sample.Timestamp is not null) reachedEnd = true; } - var complete = received < budget || samples.Count >= cap || reachedEnd || cap == 0; + // cap == 0 ("all values") is complete only when the stream proves it: short of the budget (always, unless a strict + // gateway's ceiling became the budget — #691.3) or past the window end. + var complete = received < budget || reachedEnd || (cap > 0 && samples.Count >= cap); return (samples, preWindow, complete); } @@ -315,14 +341,7 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis /// True when (or an inner exception — the client package wraps the ) /// is the gateway's InvalidArgument "max_values exceeds the server ceiling" rejection. /// - internal static bool IsMaxValuesCeilingRejection(Exception ex) - { - for (var e = ex; e is not null; e = e.InnerException) - if (e is RpcException { StatusCode: StatusCode.InvalidArgument } rpc - && rpc.Status.Detail.Contains("max_values exceeds the server ceiling", StringComparison.Ordinal)) - return true; - return false; - } + internal static bool IsMaxValuesCeilingRejection(Exception ex) => GatewayReadLimits.IsCeilingRejection(ex); /// public async Task ReadAtTimeAsync( diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayReadLimits.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayReadLimits.cs new file mode 100644 index 00000000..5f7dc1f1 --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayReadLimits.cs @@ -0,0 +1,84 @@ +using System.Globalization; +using System.Text.RegularExpressions; +using Grpc.Core; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway; + +/// +/// The HistorianGateway's raw-read row ceiling (Historian:ReadLimits:MaxValuesCeiling, default 1,000,000) as seen +/// from this side of the wire. With AllowStreamingBeyondCeiling=false (a "strict" gateway) the gateway refuses +/// any ReadRaw whose max_values exceeds the ceiling with InvalidArgument — BEFORE it streams a row +/// — and names the ceiling in the refusal ("max_values exceeds the server ceiling of {n}; …"). The gateway has +/// no RPC that reports the ceiling, so the refusal is how it is learned (#691.3). +/// +/// +/// A read whose budget the gateway refuses is re-issued ONCE with the ceiling as its budget +/// (); the caller then treats a stream that delivered the whole ceiling as +/// possibly truncated (never answered from). Nothing is cached: a re-configured gateway is honoured on the next read, +/// and a refusal costs one round-trip with no historian work behind it (it is validated before a session is acquired). +/// +internal static class GatewayReadLimits +{ + private const string CeilingRejectionText = "max_values exceeds the server ceiling"; + + private static readonly Regex CeilingNumber = new(@"max_values exceeds the server ceiling of (\d+)", + RegexOptions.CultureInvariant | RegexOptions.Compiled); + + /// + /// True when (or an inner exception — the client package wraps the ) + /// is the gateway's InvalidArgument "max_values exceeds the server ceiling" rejection. + /// + /// The exception. + /// Whether it is the ceiling rejection. + public static bool IsCeilingRejection(Exception ex) => Rejection(ex) is not null; + + /// + /// The ceiling a ceiling rejection names, or — when is a ceiling rejection that does not + /// carry a parseable number — the gateway's default (). + /// + /// The exception. + /// The ceiling (> 0) when this returns true. + /// Whether is a ceiling rejection. + public static bool TryGetCeiling(Exception ex, out int ceiling) + { + ceiling = 0; + if (Rejection(ex) is not { } rpc) return false; + var m = CeilingNumber.Match(rpc.Status.Detail); + ceiling = m.Success && int.TryParse(m.Groups[1].Value, NumberStyles.None, CultureInfo.InvariantCulture, out var n) && n > 0 + ? n + : RawAggregateReader.GatewayDefaultMaxValuesCeiling; + return true; + } + + /// + /// Runs with ; if a strict gateway refuses that budget, runs it once + /// more with the ceiling it named — only when that ceiling is smaller than the refused budget (otherwise the refusal + /// is rethrown). The refusal arrives before any row, so the first pass has consumed nothing. + /// + /// The pass result. + /// The max_values to ask for first. + /// One drained read with the given budget. + /// The pass result and the budget it actually used. + public static async Task<(T Result, int Budget)> ReadWithCeilingRetryAsync(int budget, Func> pass) + { + int ceiling; + try + { + return (await pass(budget).ConfigureAwait(false), budget); + } + catch (Exception ex) when (TryGetCeiling(ex, out ceiling) && ceiling < budget) + { + // fall through to the retry (outside the catch, so a second refusal is not nested in this one) + } + return (await pass(ceiling).ConfigureAwait(false), ceiling); + } + + private static RpcException? Rejection(Exception ex) + { + for (var e = ex; e is not null; e = e.InnerException) + if (e is RpcException { StatusCode: StatusCode.InvalidArgument } rpc + && rpc.Status.Detail.Contains(CeilingRejectionText, StringComparison.Ordinal)) + return rpc; + return null; + } +} 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 a9f95808..c4ebff5c 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 @@ -49,7 +49,10 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway; /// requested row budget — has been drained. The gateway is asked for at most cap + 2 + /// rows, clamped at while the cap is at /// or under it (#688, ); a stream that delivers all of them cannot be proven -/// complete and fails the same way rather than answering from a truncated window. +/// complete and fails the same way rather than answering from a truncated window. A strict gateway whose ceiling is +/// below that budget (a ceiling under 1,000,000, or an unbounded cap) refuses it naming its ceiling; the window is +/// then re-read once with the ceiling as budget (#691.3, ), and the same +/// "delivered the whole budget ⇒ cannot be proven complete" rule applies. /// /// /// Values and status. A sample's value is its numeric value (Double), else its string value, else Empty; its @@ -178,8 +181,43 @@ internal static class RawAggregateReader if (plan.BucketCount == 0) return []; var needsRange = AggregateEvaluator.NeedsDataRange(aggregate, plan); var requestStart = RequestStartFor(plan.StartUtc); - var maxValues = MaxValuesFor(maxWindowRows); + // A strict gateway that refuses the budget is re-asked once at the ceiling it names (#691.3; e.g. a ceiling below + // the 1,000,000 #688 clamps at, or an unbounded cap) — and a stream that fills that ceiling is never evaluated. + var (pass, maxValues) = await GatewayReadLimits.ReadWithCeilingRetryAsync(MaxValuesFor(maxWindowRows), + budget => ReadWindowAsync(client, tag, aggregate, plan, requestStart, budget, maxWindowRows, ct)).ConfigureAwait(false); + var (buckets, earliest, latest, received, overCap) = pass; + if (overCap || received >= maxValues) + throw LimitExceeded(tag, maxWindowRows); + + if (needsRange) + { + // Earlier data: no non-Bad pre-window row ⇒ ask explicitly (normally only when the tag's history really starts + // inside the window). + var earlierSeen = earliest is { } e && (e < plan.StartUtc || (plan.Reversed && e == plan.StartUtc)); + if (!earlierSeen && plan.StartUtc.AddTicks(-1) > ProbeFloor) + if (await FirstSampleAsync(client, tag, ProbeFloor, plan.StartUtc.AddTicks(-1), ct).ConfigureAwait(false) is { } before) + earliest = before; + + // Later data: the main read covers a sample exactly at the window end (forward: that is already later data); + // otherwise a small separate read from the end to now + horizon. + var laterSeen = latest is { } l && (plan.Reversed ? l > plan.EndUtc : l >= plan.EndUtc); + var horizon = DateTime.MaxValue - now > LookAheadHorizon + ? DateTime.SpecifyKind(now + LookAheadHorizon, DateTimeKind.Utc) + : DateTime.MaxValue; + if (!laterSeen && horizon > plan.EndUtc) + if (await FirstSampleAsync(client, tag, plan.EndUtc, horizon, ct).ConfigureAwait(false) is { } after) + latest = after; + } + + return AggregateEvaluator.EvaluateAll(aggregate, plan, buckets.Rows(), earliest, latest, now); + } + + /// The window read of with one row budget, drained. + private static async Task<(RawBucketAccumulator Buckets, DateTime? Earliest, DateTime? Latest, long Received, bool OverCap)> + ReadWindowAsync(IHistorianGatewayClient client, string tag, HistoryAggregateType aggregate, BucketPlan plan, + DateTime requestStart, int maxValues, int maxWindowRows, CancellationToken ct) + { var buckets = new RawBucketAccumulator(aggregate == HistoryAggregateType.MaximumActualTime); DateTime? earliest = null, latest = null; long windowRows = 0, received = 0; @@ -214,30 +252,7 @@ internal static class RawAggregateReader if (t >= plan.StartUtc && (latest is null || t > latest)) latest = t; } } - if (overCap || (maxWindowRows > 0 && received >= maxValues)) - throw LimitExceeded(tag, maxWindowRows); - - if (needsRange) - { - // Earlier data: no non-Bad pre-window row ⇒ ask explicitly (normally only when the tag's history really starts - // inside the window). - var earlierSeen = earliest is { } e && (e < plan.StartUtc || (plan.Reversed && e == plan.StartUtc)); - if (!earlierSeen && plan.StartUtc.AddTicks(-1) > ProbeFloor) - if (await FirstSampleAsync(client, tag, ProbeFloor, plan.StartUtc.AddTicks(-1), ct).ConfigureAwait(false) is { } before) - earliest = before; - - // Later data: the main read covers a sample exactly at the window end (forward: that is already later data); - // otherwise a small separate read from the end to now + horizon. - var laterSeen = latest is { } l && (plan.Reversed ? l > plan.EndUtc : l >= plan.EndUtc); - var horizon = DateTime.MaxValue - now > LookAheadHorizon - ? DateTime.SpecifyKind(now + LookAheadHorizon, DateTimeKind.Utc) - : DateTime.MaxValue; - if (!laterSeen && horizon > plan.EndUtc) - if (await FirstSampleAsync(client, tag, plan.EndUtc, horizon, ct).ConfigureAwait(false) is { } after) - latest = after; - } - - return AggregateEvaluator.EvaluateAll(aggregate, plan, buckets.Rows(), earliest, latest, now); + return (buckets, earliest, latest, received, overCap); } /// diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/FakeHistorianGatewayClient.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/FakeHistorianGatewayClient.cs index 2d91b1af..fd354a62 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/FakeHistorianGatewayClient.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/FakeHistorianGatewayClient.cs @@ -102,10 +102,30 @@ public sealed class FakeHistorianGatewayClient : IHistorianGatewayClient LastReadRawMaxValues = maxValues; ReadRawCallCount++; RawCalls.Add((startUtc, endUtc, maxValues)); + if (MaxValuesCeiling is { } ceiling && maxValues > ceiling) + { + // The real gateway's refusal (HistorianReadService.ReadRaw), raised on the first MoveNext before any row, as + // the client package surfaces it: a HistorianGatewayException wrapping the RpcException. + var rpc = new Grpc.Core.RpcException(new Grpc.Core.Status(Grpc.Core.StatusCode.InvalidArgument, + CeilingRejectionDetail ?? $"max_values exceeds the server ceiling of {ceiling}; read large ranges by time-slicing " + + "(split the range and read the slices sequentially), or enable " + + "Historian:ReadLimits:AllowStreamingBeyondCeiling to page through server-side.")); + return ToAsyncStream(Array.Empty(), + new ZB.MOM.WW.HistorianGateway.Client.HistorianGatewayException(rpc.Status.Detail, rpc), ct); + } var items = RawSource is null ? RawSamples : RawSource(startUtc, endUtc, maxValues).ToList(); return CountingStream(ToAsyncStream(items, ThrowOnRead, ct)); } + /// + /// A strict gateway (Historian:ReadLimits:AllowStreamingBeyondCeiling=false) with this + /// MaxValuesCeiling: a ReadRaw asking for more is refused InvalidArgument (#691.3). + /// + public int? MaxValuesCeiling; + + /// Overrides the refusal's detail text (an older/other gateway wording). + public string? CeilingRejectionDetail; + public IAsyncEnumerable ReadAggregateAsync( string tag, DateTime startUtc, diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayStrictCeilingTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayStrictCeilingTests.cs new file mode 100644 index 00000000..825b2c77 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayStrictCeilingTests.cs @@ -0,0 +1,185 @@ +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; + +/// +/// #691.3: a gateway with Historian:ReadLimits:AllowStreamingBeyondCeiling=false refuses any ReadRaw whose +/// max_values exceeds its MaxValuesCeiling (before streaming, naming the ceiling). An uncapped raw read +/// (NumValuesPerNode = 0, sent as ) used to be refused outright — even a 10-row window. +/// The read is now re-issued once with the ceiling the refusal named, and a stream that delivered the whole ceiling +/// without reaching the window end is never answered from (BadTooManyOperations). The fake is AVEVA-shaped: +/// a boundary row at the query start, [start, end] inclusive, capped at max. +/// +public sealed class GatewayStrictCeilingTests +{ + private static readonly CancellationToken Ct = CancellationToken.None; + private static readonly DateTime T0 = new(2026, 10, 12, 19, 0, 0, DateTimeKind.Utc); + private static readonly DateTime Now = T0.AddDays(1); + + private static DateTime At(int seconds) => T0.AddSeconds(seconds); + + private static HistorianSample Row(DateTime t, double v, uint quality = 0) => + new() { Tag = "T", Timestamp = Timestamp.FromDateTime(t), NumericValue = v, OpcQuality = 192, Quality = quality }; + + /// One stored sample per second, value = seconds, at +0 … +(count − 1); strict ceiling when given. + private static FakeHistorianGatewayClient Store(int count, int? ceiling) + { + var stored = Enumerable.Range(0, count).Select(i => (T: At(i), V: (double)i)).ToList(); + return new FakeHistorianGatewayClient + { + MaxValuesCeiling = ceiling, + RawSource = (start, end, max) => + { + var rows = new List(); + if (stored.All(s => s.T != start)) + { + var prior = stored.LastOrDefault(s => s.T < start); + rows.Add(prior.T == default + ? new HistorianSample { Tag = "T", Timestamp = Timestamp.FromDateTime(start), NumericValue = 0, OpcQuality = 0, Quality = 1 } + : Row(start, prior.V, 133)); + } + rows.AddRange(stored.Where(s => s.T >= start && s.T <= end).Select(s => Row(s.T, s.V))); + return rows.Take(max); + }, + }; + } + + private static GatewayHistorianDataSource Source(FakeHistorianGatewayClient fake, int maxRawRows = 1_000_000) => + new(fake, NullLogger.Instance, maxRawRows, new FixedClock(Now)); + + private sealed class FixedClock(DateTime now) : TimeProvider + { + public override DateTimeOffset GetUtcNow() => new(now); + } + + // ---- raw, uncapped --------------------------------------------------------------------------------------------------- + + [Fact] + public async Task An_uncapped_read_of_a_small_window_on_a_strict_gateway_is_served_with_the_ceiling_as_budget() + { + var fake = Store(10, ceiling: 1000); + var ds = Source(fake); + var r = await ds.ReadRawAsync("T", At(0), At(60), 0, Ct); + + r.Samples.Select(s => s.Value).ShouldBe(Enumerable.Range(0, 10).Select(i => (object?)(double)i)); + fake.RawCalls.Select(c => c.MaxValues).ShouldBe([int.MaxValue, 1000]); // refused, then the ceiling it named + fake.RawStreamsCompleted.ShouldBe(1); + ds.GetHealthSnapshot().TotalFailures.ShouldBe(0); + } + + [Fact] + public async Task An_uncapped_read_on_a_default_gateway_still_sends_one_unbounded_request() + { + var fake = Store(10, ceiling: null); + await Source(fake).ReadRawAsync("T", At(0), At(60), 0, Ct); + fake.RawCalls.Select(c => c.MaxValues).ShouldBe([int.MaxValue]); + } + + [Fact] + public async Task An_uncapped_window_holding_more_than_the_ceiling_is_a_request_limit_never_a_truncated_answer() + { + var fake = Store(100, ceiling: 50); + var ds = Source(fake); + await Should.ThrowAsync(() => ds.ReadRawAsync("T", At(0), At(200), 0, Ct)); + fake.RawStreamsCompleted.ShouldBe(1); // drained, never abandoned + ds.GetHealthSnapshot().TotalFailures.ShouldBe(0); + } + + [Fact] + public async Task A_ceiling_full_stream_that_reached_the_window_end_is_complete() + { + // 50 rows: the boundary row + +0 … +48; the window end is +48 (the gateway's end is inclusive), so the stream + // provably ended at the window end — complete although it delivered the whole ceiling. + var fake = Store(100, ceiling: 50); + var r = await Source(fake).ReadRawAsync("T", At(0), At(48), 0, Ct); + r.Samples.Count.ShouldBe(48); // [+0, +48): the end is exclusive + } + + [Fact] + public async Task A_refusal_without_a_parseable_ceiling_retries_at_the_gateways_default_ceiling() + { + var fake = Store(10, ceiling: 1_000_000); + fake.CeilingRejectionDetail = "max_values exceeds the server ceiling; read large ranges by time-slicing."; + var r = await Source(fake).ReadRawAsync("T", At(0), At(60), 0, Ct); + r.Samples.Count.ShouldBe(10); + fake.RawCalls.Select(c => c.MaxValues).ShouldBe([int.MaxValue, RawAggregateReader.GatewayDefaultMaxValuesCeiling]); + } + + // ---- raw, capped ----------------------------------------------------------------------------------------------------- + + [Fact] + public async Task A_cap_whose_allowance_exceeds_the_ceiling_is_served_with_the_ceiling_as_budget() + { + // cap 995 + 16 = 1011 > 1000: refused, re-asked with 1000 — a small window is served in full. + var fake = Store(10, ceiling: 1000); + var r = await Source(fake).ReadRawAsync("T", At(0), At(60), 995, Ct); + r.Samples.Count.ShouldBe(10); + fake.RawCalls.Select(c => c.MaxValues).ShouldBe([995 + RawAggregateReader.PreWindowAllowance, 1000]); + } + + [Fact] + public async Task A_cap_above_the_ceiling_over_a_dense_window_is_a_request_limit_not_a_short_page() + { + // A short page would end the client's paging early (silent truncation), so it must fail instead. + var fake = Store(100, ceiling: 50); + var ds = Source(fake); + await Should.ThrowAsync(() => ds.ReadRawAsync("T", At(0), At(200), 80, Ct)); + fake.RawCalls.Count.ShouldBe(2); // the ceiling re-read is final: no third read at the same ceiling + ds.GetHealthSnapshot().TotalFailures.ShouldBe(0); + } + + [Fact] + public async Task A_cap_that_fits_the_ceiling_is_unaffected() + { + var fake = Store(100, ceiling: 50); + var r = await Source(fake).ReadRawAsync("T", At(0), At(200), 10, Ct); + r.Samples.Count.ShouldBe(10); + fake.RawCalls.Select(c => c.MaxValues).ShouldBe([10 + RawAggregateReader.PreWindowAllowance]); + } + + // ---- per-sample aggregates (#688 clamp vs a ceiling below 1,000,000) --------------------------------------------------- + + [Fact] + public async Task A_per_sample_aggregate_on_a_gateway_with_a_lower_ceiling_is_served_when_the_window_fits() + { + var fake = Store(30, ceiling: 1000); // default cap 1,000,000 ⇒ budget 1,000,000 > 1000 + var s = (await Source(fake).ReadProcessedAsync("T", At(0), At(30), TimeSpan.FromSeconds(10), + HistoryAggregateType.Count, Ct)).Samples; + s.Select(x => x.Value).ShouldBe([10, 10, 10]); + fake.RawCalls[0].MaxValues.ShouldBe(RawAggregateReader.GatewayDefaultMaxValuesCeiling); + fake.RawCalls[1].MaxValues.ShouldBe(1000); + } + + [Fact] + public async Task A_per_sample_aggregate_whose_window_fills_the_lower_ceiling_is_a_request_limit() + { + var fake = Store(100, ceiling: 50); + var ds = Source(fake); + await Should.ThrowAsync(() => ds.ReadProcessedAsync("T", At(0), At(100), + TimeSpan.FromSeconds(10), HistoryAggregateType.Count, Ct)); + ds.GetHealthSnapshot().TotalFailures.ShouldBe(0); + } + + [Fact] + public async Task An_unbounded_per_sample_aggregate_on_a_strict_gateway_is_served_with_the_ceiling() + { + var fake = Store(30, ceiling: 1000); + var s = (await Source(fake, maxRawRows: 0).ReadProcessedAsync("T", At(0), At(30), TimeSpan.FromSeconds(10), + HistoryAggregateType.Count, Ct)).Samples; + s.Select(x => x.Value).ShouldBe([10, 10, 10]); + fake.RawCalls.Take(2).Select(c => c.MaxValues).ShouldBe([int.MaxValue, 1000]); + } + + [Fact] + public async Task An_unbounded_per_sample_aggregate_filling_the_ceiling_is_a_request_limit() + { + var fake = Store(100, ceiling: 50); + await Should.ThrowAsync(() => Source(fake, maxRawRows: 0).ReadProcessedAsync("T", + At(0), At(100), TimeSpan.FromSeconds(10), HistoryAggregateType.Count, Ct)); + } +}