fix(historian): refuse aggregates not valid for the node's DataType up front (#897)
v2-ci / unit-tests (push) Has been cancelled
v2-ci / unit-tests (push) Has been cancelled
TimeAverage / Total on a Boolean node over the Gateway backend (AVEVA discrete tag, live SysPulse) went to the gateway since native TimeAverage/Total became default-on (#694). AVEVA refuses both modes at StartQuery; the gateway reports that only as Internal / HistorianServerOperationException (the same answer as a historian storage fault), so the node answered BadHistoryOperationUnsupported and the read counted as a backend failure. HistoryRead-Processed now checks the node's scalar DataType before calling any backend and answers BadAggregateNotSupported (Part 11 6.5.4: "If the Aggregate is not valid for the Node then the StatusCode shall be Bad_AggregateNotSupported"). Nothing is sent, nothing is counted. - Shared, every backend: HistoryAggregateDataTypes (Core.Abstractions) applies Part 13 5.4.2.3 Table 50: the numeric-only aggregates (Average, TimeAverage(2), Total(2), Min/Max, ActualTime, Range, Delta, DeltaBounds, Interpolative, stddev/variance) on a String or DateTime node are refused. Client-visible: the SQL backends used to answer every interval BadAggregateInvalidInputs. - Per backend: IHistorianDataSource.SupportsAggregateFor(aggregate, TypeCode?) (default true; every production backend declares it, guard-tested). Gateway: false for native TimeAverage/Total on a Boolean node. SqlServer/QuestDb: true, since they serve Boolean as 0/1 (stepped), so Boolean stays served there, as do the Gateway's raw-evaluated aggregates. The residual case (a node declared numeric whose AVEVA tag is discrete) cannot be told apart from a historian fault on the wire and still counts; that needs the gateway to surface AVEVA's native error code as a deterministic status. Tests: HistoryAggregateDataTypesTests, guard test, node-manager String/DateTime and per-type refusals, Gateway capability matrix, DiscreteTagTimeWeightedWireTests (production Gateway backend over an AVEVA-shaped fake: unsent, 0 health failures, positive controls). Each mutation (no check, no shared rule, no backend hint, Gateway hint true, shared rule always true, SQL inheriting the default) fails at least one. Live gate re-run read-only against AVEVA 2023 R2 via a private HistorianGateway fa9cf31 container: SysPulse TimeAverage on the wire = BadAggregateNotSupported, health failures unchanged; #801 gate 244 cases pass. Claude-Session: https://claude.ai/code/session_01GT67ACmpMR7b9LGRPLijMi
This commit is contained in:
Joseph Doherty
@@ -36,7 +36,7 @@ historian tagname, which is the tag's driver `FullName`.
|
||||
- **QuestDB** (`Driver.Historian.QuestDb`, `ServerHistorian:QuestDb:*`, Npgsql over PG wire; docker-dev overlay `docker-compose.questdb.yml`): same kinds/status/aggregates/HistoryUpdate as SqlServer, soft delete by tombstone (INSERT only — never `UPDATE`/`DELETE`; every read filters `NOT deleted`). ⚠️ **Timestamps are µs** (`QuestDbConnection.Round`/`BindTimestamp`; `IHistorianDataSource.TimestampResolutionTicks` = 10 steps HistoryRead paging), **pre-1970 cannot be stored** (writer drops, HistoryUpdate `BadOutOfRange`), every write waits for WAL visibility, TTL is enforced on write commits only, set Npgsql `Maximum Pool Size` < QuestDB's 64. `docs/Historian.md` §"QuestDB backend".
|
||||
- ⚠️ **Reverse processed reads are Part 13/SDK-anchored (#677, client-visible):** every aggregate on SqlServer/QuestDB now anchors a reverse read at its later time — window `(end, start]`, partial interval earliest, each stamped with its later edge (was: earlier-anchored, stamped earlier edge); on forward reads Partial also marks the interval the data starts inside (not Average — the SDK's rule, forward-only); a non-default client `AggregateConfiguration` ⇒ `BadAggregateConfigurationRejected`; Gateway provisioning sends `IntegralDivisor = 1`. SDK-parity tests: `SdkAggregateParityTests`. `docs/Historian.md` §Aggregates.
|
||||
- ⚠️ **Reverse / end-only RAW reads work on Gateway (#686, client-visible):** `IHistorianDataSource.SupportsReverseRaw` (SqlServer/QuestDB/Null `true`, Gateway `false`, interface default `false`, guard-tested) — the node manager sends `start > end` only to a backend that declares it; otherwise `ReverseRawReader` serves it from FORWARD reads: uncapped ⇒ ONE forward read reversed in memory; capped ⇒ a bounded walk back in segments (ms-aligned request bounds, width aimed from observed density at ≈ N + 8 rows and carried page to page in the continuation point — #689: it used to aim at half `MaxTieClusterOverfetch`, ~32k rows per page on a dense tag; a read asks for `2N + 18` rows (a discovery segment ≤ 1,026), `MaxTieClusterOverfetch` is only the tie-cluster ceiling — a full read is re-read, never answered truncated; only more than max(1, `MaxTieClusterOverfetch`) samples within ONE MILLISECOND ⇒ `BadTooManyOperations` (per ms, not per timestamp — requests are ms-rounded, and letting a segment narrow below 1 ms stepped a page through one ms a tick per read, ~10k reads); walks no later than now + 1 day and never before the Unix epoch; #691: after two empty segments it asks once for the whole remainder, and the stretch a page proved empty rides the continuation point — the last pages of a deep page-through no longer each walk ~14 reads to 1970). The raw/events tie-path oversized-cluster refusal is now `BadTooManyOperations` too (was `BadHistoryOperationUnsupported`). Uncapped Gateway raw reads (forward too) used to send `max_values = 0`, which the client/gateway reject — now unbounded (`int.MaxValue`; a ceiling refusal ⇒ `BadTooManyOperations`), and a cancelled read is no longer a backend failure. Since #669 such reads failed the node on Gateway and counted as a backend fault (the gateway rejects start > end — `GatewayHistorianDataSource.ReadRawAsync` now refuses one unsent); before #669 an end-only Gateway read returned the OLDEST N ascending — now the latest N at/before EndTime, newest first. Events follow since #691.4: a reverse/end-only capped events read on Gateway (`SupportsReverseEvents` false; Null now reports `true` and answers empty) is walked back the same way (ties sorted by EventId — the gateway orders by time only; one extra event found ⇒ continuation point) — it used to be refused `BadHistoryOperationUnsupported` (explicit reverse) or answered with the OLDEST N ascending (end-only); needs the gateway's `RuntimeDb:EventReadMaxRows` 0/unlimited (it truncates silently). Uncapped Gateway events reads (forward too) used to throw "Array dimensions exceeded" (result list pre-sized to `int.MaxValue`) — fixed. Above a truncated read the walk gallops up from the newest row seen, then bisects (never below 1 ms): a burst that ended 100 days ago ≈ 18 reads, a dense tag that stopped days ago ≈ 26. `docs/Historian.md` §"Open-ended and reverse raw reads".
|
||||
- **Advertised aggregates (#682):** `IHistorianDataSource.SupportedAggregates` (Gateway 4 via `GatewayHistorianDataSource.ServedAggregates`, SqlServer/QuestDB 19 via `AggregateFamilies.Served` — DurationGood/DurationBad added by #820, the PercentGood/PercentBad region times in ms, Null **none**; a null set ⇒ all) is the one authority for `HistoryServerCapabilities/AggregateFunctions` (i=11201 — only its folder references are trimmed) and the up-front `BadAggregateNotSupported` in HistoryRead-Processed. ⚠️ `ServerCapabilities/AggregateFunctions` (i=2997) is deliberately NOT trimmed: it keeps the SDK's 37 because live `AggregateFilter` subscriptions still compute them all. ⚠️ Historian disabled ⇒ Processed is now `BadAggregateNotSupported`, not `GoodNoData`. `docs/Historian.md` §Advertised aggregates.
|
||||
- **Advertised aggregates (#682):** `IHistorianDataSource.SupportedAggregates` (Gateway 4 via `GatewayHistorianDataSource.ServedAggregates`, SqlServer/QuestDB 19 via `AggregateFamilies.Served` — DurationGood/DurationBad added by #820, the PercentGood/PercentBad region times in ms, Null **none**; a null set ⇒ all) is the one authority for `HistoryServerCapabilities/AggregateFunctions` (i=11201 — only its folder references are trimmed) and the up-front `BadAggregateNotSupported` in HistoryRead-Processed. ⚠️ `ServerCapabilities/AggregateFunctions` (i=2997) is deliberately NOT trimmed: it keeps the SDK's 37 because live `AggregateFilter` subscriptions still compute them all. ⚠️ Historian disabled ⇒ Processed is now `BadAggregateNotSupported`, not `GoodNoData`. `docs/Historian.md` §Advertised aggregates. ⚠️ **#897 (client-visible): per-node DataType check before any backend call** ⇒ `BadAggregateNotSupported` (Part 11 §6.5.4), unsent, no backend failure: shared Part 13 Table 50 rule `HistoryAggregateDataTypes` (numeric-only aggregates on a String/DateTime node, every backend — SQL used to answer `BadAggregateInvalidInputs` per interval), then `IHistorianDataSource.SupportsAggregateFor(aggregate, TypeCode?)` (every production backend declares it, guard-tested; only Gateway narrows: native TimeAverage/Total on a Boolean node — AVEVA refuses them on a discrete tag, which the gateway reports only as `Internal`/`HistorianServerOperationException`). Boolean stays served as 0/1 on SQL and by Gateway's raw-evaluated aggregates. `docs/Historian.md` §Aggregates by node DataType.
|
||||
- **Raw `ReturnBounds` is honoured on every backend (#691, client-visible):** Part 11 bounding values (§4.6 Table 1 is the test oracle) are composed in the server by `HistoryBounds` from each backend's raw reads — "at or before" = capped reverse read (native on SQL backends, the #686 `ReverseRawReader` walk-back on Gateway, never AVEVA's synthetic boundary row), "at or after" = capped forward read. Bounds are stored samples of any quality, count toward `NumValuesPerNode`, start bound on the first page / end bound on the last (pending end bound rides the continuation point); missing ⇒ `Bad_BoundNotFound` (null, at the bound time; a one-sided read's open side ±1 s from the previous value). ⚠️ **Breaking for SDK-default clients:** the .NET SDK's `new ReadRawModifiedDetails()` defaults BOTH `ReturnBounds = true` AND `IsReadModified = true` — a client that cleared only `IsReadModified` now gets bounds, and one that set neither is a modified read with bounds ⇒ `Bad_InvalidArgument` (§6.5.3.3); set both explicitly (Client.Shared does; tests building details must too). No data AND no bound found ⇒ `GoodNoData`, no values (the pre-bounds answer). On Gateway the at-or-after lookup ends at now + 1 day (never a year-9999 end to AVEVA). Kill switch `ServerHistorian:HonorReturnBounds` (default `true`; `false` = pre-#691). `docs/Historian.md` §Bounding values.
|
||||
- ⚠️ **Raw pages are full (#692, client-visible, every backend):** each raw page asks the backend for `N + skip + 1` rows (the boundary ties it will trim + one lookahead row) — a resumed page used to ask only N, trim the ties and come back short; at N=1 that was an EMPTY page + continuation point that then fell into the `MaxTieClusterOverfetch + 1` over-fetch (live: 8 values = 17 pages / 25 reads / 524k rows ⇒ now 8/8/≤24). No lookahead row ⇒ last page, no CP (an exact multiple no longer ends on an empty page). Exception: an oversized tie cluster — its last slice is `min(N, remaining)` with a CP, and stepping past one at the end of data can give one empty final page. The over-fetch fires only when every row past the trimmed ties sits on the boundary timestamp; a cursor inside such a cluster is marked `WithinTieCluster` (next page goes straight to the over-fetch). `SliceTieCluster` resumes only strictly before EndTime (truncated to the backend resolution) — it used to resume AT the exclusive EndTime and return samples there. Bound placeholders carry an explicit marker (`HistoryBounds.IsPlaceholder` no longer matches a stored `Bad_BoundNotFound`/null row); the N=1 bounds probe carries its walk hint + resolved end bound into the cursor. Reviewer fuzz harnesses (paged vs unpaged over random tie series) are the right test for any paging change — repo tests passed while master mismatched 1-in-160.
|
||||
- ⚠️ **Strict-ceiling paging (#694):** `IHistorianDataSource.MaxRawValuesPerRead` = L = learned gateway ceiling − 16 (0 on SQL backends/Null, guard-tested; 0 until a refusal names the ceiling). `RawPageLimit.Plan(want, k, L)` sizes every raw page so page + boundary ties + lookahead ≤ L (room = L − k − 1, signed; first page L−1, resumed L−k−1; uncapped `N = 0` reads page too) — a capped read above the ceiling used to be `BadTooManyOperations`, now a ceiling-sized page + CP. One re-plan when the read taught a new limit or was refused for crowding (`HistoryReadLimitExceededException.PreWindowRows`, set only above the 16-row allowance; the page shrinks by it). No room on a resumed page ⇒ the tie-cluster over-fetch, bounded min(M, L−1); the slice size is never 0. The page path passes `retryOnLearnedLimit: false` to `ReverseRawReader` (the walk re-running at the old N cost 200–3,000 reads cold); bounds lookups keep it. ⚠️ A Gateway window starting at/before the Unix epoch counts AVEVA's synthetic row as pre-window — it used to set `reachedEnd`, so a ceiling-truncated read came back Good with no CP (forward-from-1970 = first page only; end-only past the data = the OLDEST rows as "latest"; pre-existing). The #687 crowding re-read now doubles its room up to 8 re-reads, never past a known ceiling, and its room is capped at `MaxTieClusterOverfetch` (passed to the Gateway source by the factory; past it ⇒ `BadTooManyOperations` with `PreWindowRows`). A raised ceiling is not re-probed by raw paging (pages stay small until another read re-probes or restart).
|
||||
|
||||
+32
-2
@@ -200,7 +200,7 @@ Processed read is `BadAggregateNotSupported` and no history aggregate functions
|
||||
|
||||
| Key | Type | Default | Description |
|
||||
|---|---|---|---|
|
||||
| `Gateway:EnableNativeTimeWeighted` | bool | `true` | Serve HistoryRead-Processed `TimeAverage` / `Total` through AVEVA's native retrieval modes. ⚠️ **On by default since #694 (client-visible: both are advertised and served again).** The gateway interval fix historiangw#27 (gateway server `9f711c1`+) is live-verified: forward and reverse per-interval values, clipped, sub-millisecond clipped and fractional-millisecond intervals against a step-held reference, and all-NULL intervals ⇒ `BadNoData` ([evidence](plans/evidence/2026-11-12-historian-694-700-746/aveva-gate.md)). `false` is the kill switch for a gateway older than the fix (it answers one Good-status whole-span value on the earliest interval): both are then `BadAggregateNotSupported` and **not advertised**. On a discrete (Boolean) AVEVA tag AVEVA refuses both modes and the node answers `BadHistoryOperationUnsupported` (counted as a backend failure; #897). |
|
||||
| `Gateway:EnableNativeTimeWeighted` | bool | `true` | Serve HistoryRead-Processed `TimeAverage` / `Total` through AVEVA's native retrieval modes. ⚠️ **On by default since #694 (client-visible: both are advertised and served again).** The gateway interval fix historiangw#27 (gateway server `9f711c1`+) is live-verified: forward and reverse per-interval values, clipped, sub-millisecond clipped and fractional-millisecond intervals against a step-held reference, and all-NULL intervals ⇒ `BadNoData` ([evidence](plans/evidence/2026-11-12-historian-694-700-746/aveva-gate.md)). `false` is the kill switch for a gateway older than the fix (it answers one Good-status whole-span value on the earliest interval): both are then `BadAggregateNotSupported` and **not advertised**. On a **Boolean** node both are refused `BadAggregateNotSupported` up front (AVEVA refuses both modes on a discrete tag; #897 — before, the read went to the gateway, answered `BadHistoryOperationUnsupported` and counted as a backend failure). See [Aggregates by node DataType](#aggregates-by-node-datatype-897). |
|
||||
| `Gateway:ProbeInterval` | TimeSpan | `00:00:30` | How often the read client calls the gateway's `ProbeAsync` for the historian health snapshot (`ProcessConnectionOpen` / `EventConnectionOpen`, shown on `/hosts`). The first probe runs at start-up, so a node that has served no HistoryRead yet — or only writes (continuous historization) — reports a healthy gateway as open; it used to read closed until some client happened to read. The latest observation (probe or read outcome) wins; probes never count as queries or failures. Zero/negative = no probe (read outcomes only, the previous behaviour). |
|
||||
| `Gateway:EventDrainBudget` | int | `10000` | Since #746 a capped HistoryReadEvents read sends cap + 1 as the server-side `max_events` and the gateway completes the stream there, so this budget applies only to the uncapped fallback read (an ambiguous server-capped answer — foreign rows filtered out of a stream the server cut — or a gateway refusing the cap) and to a stream that runs past the `max_events` it was sent. It is the number of stream rows such a read consumes (and discards) past its page cap before it abandons the gateway stream; an **abandoned** stream cancels the RPC and the gateway evicts its pooled historian session (the next read reconnects; concurrent reads on that session fail). The usual small overshoot is drained to the stream's end; past the budget the stream is abandoned, so a page over a window far larger than the page does not stream the whole window. Rows of another source never count before the "more events" signal is seen, so an abandoned stream never answers a page as complete. `0` = abandon at the cap (the previous behaviour); negative counts as `0` (a `Validate()` warning). |
|
||||
|
||||
@@ -1508,7 +1508,37 @@ two things:
|
||||
- **What HistoryRead-Processed refuses up front.** An aggregate outside the set is `BadAggregateNotSupported` for that node
|
||||
without a backend call (the backend still throws `HistoryAggregateNotSupportedException` for one as defence in depth —
|
||||
derived from the same table, never a second list). A source whose `SupportedAggregates` is `null` (out-of-tree, against
|
||||
the contract) is treated as the interface default — **all** — by both consumers.
|
||||
the contract) is treated as the interface default — **all** — by both consumers. An aggregate in the set that is not
|
||||
valid for the node's DataType is refused the same way — see the next section.
|
||||
|
||||
### Aggregates by node DataType (#897)
|
||||
|
||||
⚠️ **Client-visible.** HistoryRead-Processed checks the node's scalar DataType before calling any backend, and an aggregate
|
||||
that does not apply to it is **`BadAggregateNotSupported`** for that node — Part 11 §6.5.4: *"If the Aggregate is not valid
|
||||
for the Node then the StatusCode shall be Bad_AggregateNotSupported."* Nothing is sent and no backend failure is counted.
|
||||
Two layers, in order:
|
||||
|
||||
1. **Shared, every backend** (`HistoryAggregateDataTypes`, Core.Abstractions): Part 13 §5.4.2.3 Table 50 lists Average,
|
||||
TimeAverage, TimeAverage2, Total, Total2, Minimum, Maximum, the ActualTime pair, Range, Delta, DeltaBounds, Interpolative and
|
||||
the standard-deviation / variance family as **numeric only** ("dates, strings, arrays, etc. are not supported"). A
|
||||
**String** or **DateTime** node asked for one of them is refused. Count, Start, End, DurationGood/Bad and PercentGood/Bad
|
||||
apply to any type. (The SQL backends used to answer each interval of such a read `BadAggregateInvalidInputs`; that is
|
||||
still the answer for a usable String/DateTime sample inside a numeric node's interval.)
|
||||
2. **Per backend** (`IHistorianDataSource.SupportsAggregateFor(aggregate, valueType)`, declared by every production backend,
|
||||
guard-tested): only the **Gateway** narrows it — native **TimeAverage / Total on a Boolean node** are refused, because AVEVA
|
||||
refuses its `TimeWeightedAverage` / `Integral` modes on a discrete tag at `StartQuery`. The gateway reports that only as
|
||||
`Internal` / `HistorianServerOperationException`, the same answer as a historian storage fault, so it cannot be told apart
|
||||
on the wire; before #897 such a read (live: `SysPulse`) answered `BadHistoryOperationUnsupported` and counted as a backend
|
||||
failure in the historian health snapshot.
|
||||
|
||||
Boolean is deliberately **not** refused by the shared rule although Table 50 lists it apart from numeric: the SQL backends
|
||||
serve a Boolean node's numeric aggregates as 0/1 (stepped for the time-weighted ones — TimeAverage is the fraction of time
|
||||
true), and the Gateway's raw-evaluated Average/Min/Max/… do the same. So on a Boolean node TimeAverage / Total are served by
|
||||
`SqlServer` / `QuestDb` and refused by `Gateway`. An unknown node type (BaseDataType, an array) is not refused.
|
||||
|
||||
The residual case — a node declared numeric whose AVEVA tag is discrete — still reaches AVEVA and answers
|
||||
`BadHistoryOperationUnsupported`, counted as a failure, until the gateway surfaces AVEVA's native error code as a
|
||||
deterministic status.
|
||||
|
||||
⚠️ **`Server/ServerCapabilities/AggregateFunctions` (`i=2997`) is deliberately NOT trimmed** — it keeps the SDK's full 37 in
|
||||
every configuration, including with no historian. Part 5 defines it as every aggregate the *server* supports, and the server
|
||||
|
||||
BIN
Binary file not shown.
@@ -84,6 +84,13 @@ deterministic `GatewayAlarmWriterBatchTests`.
|
||||
gateway answers `Internal`, and the node answers **`BadHistoryOperationUnsupported`**. That is a Bad status, never a value,
|
||||
but the read counts as a backend failure in the health snapshot. Before `4c8e977e` the pair was refused up front with
|
||||
`BadAggregateNotSupported`. Follow-up: [#897](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/897).
|
||||
**Fixed by #897:** a Boolean node's TimeAverage / Total are now refused `BadAggregateNotSupported` up front on the
|
||||
Gateway backend (nothing sent, no failure counted); see docs/Historian.md §Aggregates by node DataType.
|
||||
Live re-run 2026-11-13 (followups gate + #801 gate, both passed, strict fixture not run) against a private
|
||||
HistorianGateway `fa9cf31` container (auth disabled, live writes off, removed afterwards): the wire answers
|
||||
`BadAggregateNotSupported` for `SysPulse` TimeAverage with the health failure count unchanged, while the gate's own
|
||||
direct backend call still gets `HistorianServerOperationException`, so AVEVA still refuses the mode
|
||||
([report](aveva-gate-followups-fix-897.json.gz), 65 cases, sha256 of the JSON `6b7d136029ea…`).
|
||||
4. AVEVA integrates an integer tag (`SysTimeSec`) as a step (TimeAverage of 0..19 over 20 s = 9.5). Float tags are sloped
|
||||
(see docs/Historian.md). The step reference above is therefore exact for `SysTimeSec`.
|
||||
|
||||
|
||||
@@ -67,6 +67,64 @@ public static class HistoryAggregateTypes
|
||||
System.Collections.Frozen.FrozenSet<HistoryAggregateType>.Empty;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Which <see cref="HistoryAggregateType"/>s apply to a node's DataType — OPC UA Part 13 §5.4.2.3 Table 50, the shared
|
||||
/// rule every backend's HistoryRead-Processed is checked against before the backend is called (lmxopcua#897).
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// Table 50 puts the averaging / extremum / integral aggregates (Average, TimeAverage, TimeAverage2, Total, Total2,
|
||||
/// Minimum, Maximum, the ActualTime pair, Range, Delta, DeltaBounds, Interpolative, the standard deviation /
|
||||
/// variance family) under "numeric only — dates, strings, arrays, etc. are not supported", and Part 11 §6.5.4 says
|
||||
/// that "if the Aggregate is not valid for the Node then the StatusCode shall be <c>Bad_AggregateNotSupported</c>".
|
||||
/// So a String or DateTime node asked for one of them is refused for the node, unsent, on every backend.
|
||||
/// (A SQL backend used to answer every interval <c>Bad_AggregateInvalidInputs</c> instead — the per-value status
|
||||
/// Part 13 keeps for data that cannot be aggregated, which is still what a usable String/DateTime sample inside a
|
||||
/// numeric node's interval gets.)
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// Boolean is deliberately NOT refused here, although Table 50 lists it apart from numeric: the SQL backends
|
||||
/// serve a Boolean node's numeric aggregates as 0/1 (stepped for the time-weighted ones; the fraction of time true
|
||||
/// is a common reading of TimeAverage) and have since #671. A backend that cannot aggregate a Boolean says so per
|
||||
/// aggregate through <see cref="IHistorianDataSource.SupportsAggregateFor"/> (the Gateway: AVEVA refuses
|
||||
/// TimeAverage / Total on a discrete tag).
|
||||
/// </para>
|
||||
/// <para>An unknown node type (<c>null</c>: BaseDataType, an array, a type with no CLR mapping) is not refused.</para>
|
||||
/// </remarks>
|
||||
public static class HistoryAggregateDataTypes
|
||||
{
|
||||
/// <summary>The aggregates Part 13 Table 50 defines for numeric data only.</summary>
|
||||
public static readonly IReadOnlySet<HistoryAggregateType> NumericOnly =
|
||||
System.Collections.Frozen.FrozenSet.ToFrozenSet(
|
||||
[
|
||||
HistoryAggregateType.Average,
|
||||
HistoryAggregateType.Minimum,
|
||||
HistoryAggregateType.Maximum,
|
||||
HistoryAggregateType.Total,
|
||||
HistoryAggregateType.Interpolative,
|
||||
HistoryAggregateType.TimeAverage,
|
||||
HistoryAggregateType.Delta,
|
||||
HistoryAggregateType.Range,
|
||||
HistoryAggregateType.MinimumActualTime,
|
||||
HistoryAggregateType.MaximumActualTime,
|
||||
HistoryAggregateType.DeltaBounds,
|
||||
HistoryAggregateType.StandardDeviationPopulation,
|
||||
HistoryAggregateType.VariancePopulation,
|
||||
HistoryAggregateType.TimeAverage2,
|
||||
HistoryAggregateType.Total2,
|
||||
]);
|
||||
|
||||
/// <summary>
|
||||
/// Whether <paramref name="aggregate"/> is valid for a node whose scalar value type is <paramref name="valueType"/>:
|
||||
/// false only for a <see cref="NumericOnly"/> aggregate on a String or DateTime node.
|
||||
/// </summary>
|
||||
/// <param name="aggregate">The requested aggregate.</param>
|
||||
/// <param name="valueType">The node's scalar value type, or <c>null</c> when unknown (never refused).</param>
|
||||
/// <returns><see langword="false"/> when the server must answer <c>Bad_AggregateNotSupported</c> for the node.</returns>
|
||||
public static bool AppliesTo(HistoryAggregateType aggregate, TypeCode? valueType) =>
|
||||
valueType is not (TypeCode.String or TypeCode.DateTime) || !NumericOnly.Contains(aggregate);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Thrown by a history backend when it cannot serve the requested <see cref="HistoryAggregateType"/>.
|
||||
/// The Server HistoryRead path translates it to <c>BadAggregateNotSupported</c> for that node only.
|
||||
|
||||
@@ -250,6 +250,26 @@ public interface IHistorianDataSource : IDisposable
|
||||
/// </remarks>
|
||||
IReadOnlySet<HistoryAggregateType> SupportedAggregates => HistoryAggregateTypes.All;
|
||||
|
||||
/// <summary>
|
||||
/// Whether this backend can serve <paramref name="aggregate"/> (a member of <see cref="SupportedAggregates"/>) on a
|
||||
/// node whose scalar value type is <paramref name="valueType"/> — a per-type refinement of
|
||||
/// <see cref="SupportedAggregates"/>. <see langword="false"/> ⇒ HistoryRead-Processed answers
|
||||
/// <c>BadAggregateNotSupported</c> for that node up front (Part 11 §6.5.4), without calling the backend and without
|
||||
/// counting a backend failure (lmxopcua#897).
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// The Part 13 Table 50 data-type rule shared by every backend (no numeric aggregate on a String / DateTime node) is
|
||||
/// <see cref="HistoryAggregateDataTypes.AppliesTo"/>, checked by the server before this; this member is only for
|
||||
/// what one backend cannot do on top of it — the Gateway's native TimeAverage / Total on a Boolean node, which AVEVA
|
||||
/// refuses for a discrete tag. <paramref name="valueType"/> is <c>null</c> when the node's type is unknown. The
|
||||
/// default (<see langword="true"/>) keeps a test fake or out-of-tree source unchanged; every production backend
|
||||
/// declares it itself (a reflection guard pins that). Called on every processed read: it must be cheap and pure.
|
||||
/// </remarks>
|
||||
/// <param name="aggregate">The requested aggregate.</param>
|
||||
/// <param name="valueType">The node's scalar value type, or <c>null</c> when unknown.</param>
|
||||
/// <returns><see langword="true"/> when the backend can serve the aggregate on such a node.</returns>
|
||||
bool SupportsAggregateFor(HistoryAggregateType aggregate, TypeCode? valueType) => true;
|
||||
|
||||
/// <summary>
|
||||
/// What this backend serves for HistoryRead — the authority for the <c>AccessHistoryDataCapability</c>,
|
||||
/// <c>AccessHistoryEventsCapability</c> and <c>ServerTimestampSupported</c> flags the server advertises under
|
||||
|
||||
@@ -87,6 +87,10 @@ public sealed class NullHistorianDataSource : IHistorianDataSource
|
||||
// historian misled capability discovery. Raw / AtTime / Events reads still return the empty GoodNoData result.
|
||||
public IReadOnlySet<HistoryAggregateType> SupportedAggregates => HistoryAggregateTypes.None;
|
||||
|
||||
/// <inheritdoc />
|
||||
// true: no type narrows an empty SupportedAggregates further (every processed read is refused by it already).
|
||||
public bool SupportsAggregateFor(HistoryAggregateType aggregate, TypeCode? valueType) => true;
|
||||
|
||||
/// <inheritdoc />
|
||||
// None: no history is served, so HistoryServerCapabilities advertises no data / event access (lmxopcua#881) and no
|
||||
// folder is promoted to an event-history notifier.
|
||||
|
||||
@@ -722,6 +722,15 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
|
||||
public IReadOnlySet<HistoryAggregateType> SupportedAggregates =>
|
||||
_enableNativeTimeWeighted ? ServedAggregates : RawEvaluatedAggregates;
|
||||
|
||||
/// <inheritdoc />
|
||||
// false for AVEVA's native TimeAverage / Total on a Boolean node (lmxopcua#897): AVEVA refuses both retrieval modes on a
|
||||
// discrete tag at StartQuery, and the gateway answers that only as Internal / HistorianServerOperationException — the
|
||||
// same answer as a historian storage fault — so it read as BadHistoryOperationUnsupported and a backend failure. The
|
||||
// server now answers BadAggregateNotSupported up front and nothing is sent. The raw-evaluated aggregates serve a
|
||||
// Boolean as 0/1 (as the SQL backends do); a String / DateTime node is refused by the shared HistoryAggregateDataTypes rule.
|
||||
public bool SupportsAggregateFor(HistoryAggregateType aggregate, TypeCode? valueType) =>
|
||||
!(valueType == TypeCode.Boolean && AggregateModeMapper.NativeAggregates.Contains(aggregate));
|
||||
|
||||
/// <inheritdoc />
|
||||
// Data + events, but NOT ServerTimestamps: AVEVA keeps no per-sample server time, so SampleMapper stamps the read
|
||||
// time — advertising ServerTimestampSupported would promise a stored value we do not have (lmxopcua#881).
|
||||
|
||||
@@ -438,6 +438,11 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource
|
||||
/// <inheritdoc />
|
||||
public IReadOnlySet<HistoryAggregateType> SupportedAggregates => AggregateFamilies.Served;
|
||||
|
||||
/// <inheritdoc />
|
||||
// true for every type: a Boolean node's numeric aggregates are served as 0/1 (stepped for the time-weighted ones), and a
|
||||
// String / DateTime node's numeric ones are refused by the server's shared HistoryAggregateDataTypes rule before this.
|
||||
public bool SupportsAggregateFor(HistoryAggregateType aggregate, TypeCode? valueType) => true;
|
||||
|
||||
/// <inheritdoc />
|
||||
// Data + events; every sample row stores its server time and a raw read returns it (lmxopcua#881).
|
||||
public HistorianReadCapabilities ReadCapabilities =>
|
||||
|
||||
+5
@@ -363,6 +363,11 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource
|
||||
/// <inheritdoc />
|
||||
public IReadOnlySet<HistoryAggregateType> SupportedAggregates => AggregateFamilies.Served;
|
||||
|
||||
/// <inheritdoc />
|
||||
// true for every type: a Boolean node's numeric aggregates are served as 0/1 (stepped for the time-weighted ones), and a
|
||||
// String / DateTime node's numeric ones are refused by the server's shared HistoryAggregateDataTypes rule before this.
|
||||
public bool SupportsAggregateFor(HistoryAggregateType aggregate, TypeCode? valueType) => true;
|
||||
|
||||
/// <inheritdoc />
|
||||
// Data + events; every sample row stores ServerTimeUtc (written by the recorder / HistoryUpdate) and a raw read
|
||||
// returns it (lmxopcua#881).
|
||||
|
||||
@@ -7627,6 +7627,18 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
|
||||
continue;
|
||||
}
|
||||
|
||||
// An aggregate not valid for THIS node's DataType (lmxopcua#897) — Part 11 §6.5.4: "If the Aggregate is not
|
||||
// valid for the Node then the StatusCode shall be Bad_AggregateNotSupported". The shared Part 13 Table 50
|
||||
// rule (no numeric aggregate on a String / DateTime node, every backend), then the backend's own per-type
|
||||
// limit (the Gateway's native TimeAverage / Total on a Boolean node). Unsent, so never a backend failure.
|
||||
var nodeValueType = AtTimeValueType(handle.NodeId);
|
||||
if (!HistoryAggregateDataTypes.AppliesTo(aggregate.Value, nodeValueType)
|
||||
|| !HistorianDataSource.SupportsAggregateFor(aggregate.Value, nodeValueType))
|
||||
{
|
||||
errors[handle.Index] = StatusCodes.BadAggregateNotSupported;
|
||||
continue;
|
||||
}
|
||||
|
||||
if (!configurationServable)
|
||||
{
|
||||
errors[handle.Index] = StatusCodes.BadAggregateConfigurationRejected;
|
||||
|
||||
+16
@@ -111,6 +111,22 @@ public sealed class HistorianDataSourceReverseEventsGuardTests
|
||||
}
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Every_implementation_declares_its_per_type_aggregate_support_itself()
|
||||
{
|
||||
// lmxopcua#897: the default (true) would let a Gateway wrapper send AVEVA native TimeAverage / Total for a Boolean
|
||||
// node, which AVEVA refuses at StartQuery — BadHistoryOperationUnsupported and a backend failure, not the up-front
|
||||
// BadAggregateNotSupported.
|
||||
foreach (var type in ProductionImplementations())
|
||||
{
|
||||
var map = type.GetInterfaceMap(typeof(IHistorianDataSource));
|
||||
var method = map.TargetMethods[Array.FindIndex(map.InterfaceMethods,
|
||||
m => m.Name == nameof(IHistorianDataSource.SupportsAggregateFor))];
|
||||
method.DeclaringType.ShouldBe(type,
|
||||
$"{type.FullName} must declare SupportsAggregateFor itself, not inherit the interface default");
|
||||
}
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Every_implementation_declares_its_raw_page_limit_itself()
|
||||
{
|
||||
|
||||
+62
@@ -0,0 +1,62 @@
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Core.Abstractions.Tests.Historian;
|
||||
|
||||
/// <summary>
|
||||
/// lmxopcua#897: the shared Part 13 §5.4.2.3 Table 50 rule — which aggregates apply to a node's DataType — that every
|
||||
/// backend's HistoryRead-Processed is checked against before the backend is called.
|
||||
/// </summary>
|
||||
public sealed class HistoryAggregateDataTypesTests
|
||||
{
|
||||
public static TheoryData<HistoryAggregateType> NumericOnly() => [.. HistoryAggregateDataTypes.NumericOnly];
|
||||
|
||||
public static TheoryData<HistoryAggregateType> AnyType() =>
|
||||
[.. Enum.GetValues<HistoryAggregateType>().Where(a => !HistoryAggregateDataTypes.NumericOnly.Contains(a))];
|
||||
|
||||
[Fact]
|
||||
public void The_numeric_only_set_is_Table_50s()
|
||||
{
|
||||
HistoryAggregateDataTypes.NumericOnly.ShouldBe(
|
||||
[
|
||||
HistoryAggregateType.Average, HistoryAggregateType.Minimum, HistoryAggregateType.Maximum,
|
||||
HistoryAggregateType.Total, HistoryAggregateType.Interpolative, HistoryAggregateType.TimeAverage,
|
||||
HistoryAggregateType.Delta, HistoryAggregateType.Range, HistoryAggregateType.MinimumActualTime,
|
||||
HistoryAggregateType.MaximumActualTime, HistoryAggregateType.DeltaBounds,
|
||||
HistoryAggregateType.StandardDeviationPopulation, HistoryAggregateType.VariancePopulation,
|
||||
HistoryAggregateType.TimeAverage2, HistoryAggregateType.Total2,
|
||||
],
|
||||
ignoreOrder: true);
|
||||
// Table 50 "all types" / "numeric or Boolean" members stay out of it.
|
||||
foreach (var a in new[]
|
||||
{
|
||||
HistoryAggregateType.Count, HistoryAggregateType.Start, HistoryAggregateType.End,
|
||||
HistoryAggregateType.StartBound, HistoryAggregateType.EndBound, HistoryAggregateType.DurationGood,
|
||||
HistoryAggregateType.DurationBad, HistoryAggregateType.PercentGood, HistoryAggregateType.PercentBad,
|
||||
HistoryAggregateType.WorstQuality, HistoryAggregateType.AnnotationCount,
|
||||
HistoryAggregateType.NumberOfTransitions,
|
||||
})
|
||||
HistoryAggregateDataTypes.NumericOnly.ShouldNotContain(a);
|
||||
}
|
||||
|
||||
[Theory]
|
||||
[MemberData(nameof(NumericOnly))]
|
||||
public void A_numeric_only_aggregate_does_not_apply_to_a_String_or_DateTime_node(HistoryAggregateType a)
|
||||
{
|
||||
HistoryAggregateDataTypes.AppliesTo(a, TypeCode.String).ShouldBeFalse();
|
||||
HistoryAggregateDataTypes.AppliesTo(a, TypeCode.DateTime).ShouldBeFalse();
|
||||
|
||||
// ...but to every numeric node, to Boolean (served as 0/1 where a backend can) and to an unknown type.
|
||||
foreach (var t in new TypeCode?[] { TypeCode.Double, TypeCode.Single, TypeCode.Int32, TypeCode.UInt16, TypeCode.Boolean, null })
|
||||
HistoryAggregateDataTypes.AppliesTo(a, t).ShouldBeTrue($"{a} on {t?.ToString() ?? "an unknown type"}");
|
||||
}
|
||||
|
||||
[Theory]
|
||||
[MemberData(nameof(AnyType))]
|
||||
public void Every_other_aggregate_applies_to_any_node(HistoryAggregateType a)
|
||||
{
|
||||
foreach (var t in new TypeCode?[] { TypeCode.String, TypeCode.DateTime, TypeCode.Boolean, TypeCode.Double, null })
|
||||
HistoryAggregateDataTypes.AppliesTo(a, t).ShouldBeTrue($"{a} on {t?.ToString() ?? "an unknown type"}");
|
||||
}
|
||||
}
|
||||
+18
@@ -53,6 +53,24 @@ public sealed class GatewayNativeTimeWeightedGateTests
|
||||
fake.ReadAggregateCallCount.ShouldBeGreaterThan(0);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// lmxopcua#897: AVEVA refuses its native TimeAverage / Total retrieval modes on a discrete tag (live: SysPulse,
|
||||
/// StartQuery UnknownException 85 ⇒ gateway Internal), so the source declares both unsupported on a Boolean node and
|
||||
/// the server refuses them up front. Every other type keeps them, and the raw-evaluated aggregates keep Boolean.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void The_native_pair_is_declared_unsupported_on_a_Boolean_node_only()
|
||||
{
|
||||
var ds = new GatewayHistorianDataSource(new FakeHistorianGatewayClient(), NullLogger<GatewayHistorianDataSource>.Instance);
|
||||
foreach (var a in GatewayHistorianDataSource.ServedAggregates)
|
||||
{
|
||||
var native = a is HistoryAggregateType.TimeAverage or HistoryAggregateType.Total;
|
||||
ds.SupportsAggregateFor(a, TypeCode.Boolean).ShouldBe(!native, $"{a} on a Boolean node");
|
||||
foreach (var t in new TypeCode?[] { TypeCode.Double, TypeCode.Single, TypeCode.Int32, TypeCode.UInt16, TypeCode.String, null })
|
||||
ds.SupportsAggregateFor(a, t).ShouldBeTrue($"{a} on {t?.ToString() ?? "an unknown type"}");
|
||||
}
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void The_default_set_is_all_twelve_and_the_kill_switch_leaves_the_ten_raw_evaluated()
|
||||
{
|
||||
|
||||
+7
-3
@@ -270,7 +270,9 @@ public sealed class AvevaGatewayHistorianFollowupsLiveGateTests
|
||||
})
|
||||
{
|
||||
var intervalMs = (end - start).TotalMilliseconds / 2;
|
||||
var failuresBefore = source.GetHealthSnapshot().TotalFailures;
|
||||
var p = await ReadProcessedAsync(session, Node(node), start, end, intervalMs, aggId, ct);
|
||||
var failuresAfter = source.GetHealthSnapshot().TotalFailures;
|
||||
ZB.MOM.WW.OtOpcUa.Core.Abstractions.HistoryReadResult? d = null;
|
||||
string? directError = null;
|
||||
try { d = await source.ReadProcessedAsync(tag, start, end, TimeSpan.FromMilliseconds(intervalMs), aggType, ct); }
|
||||
@@ -279,10 +281,12 @@ public sealed class AvevaGatewayHistorianFollowupsLiveGateTests
|
||||
$"{Types(p.Rows)} direct={directError ?? "ok"}", p.Rows));
|
||||
if (label == "boolean" && aggType == HistoryAggregateType.TimeAverage)
|
||||
{
|
||||
// AVEVA refuses its native time-weighted modes on a discrete tag (StartQuery fails): recorded; the node
|
||||
// must answer Bad, never a value.
|
||||
Check(failures, p.Status.StartsWith("Bad", StringComparison.Ordinal) && p.Rows.Count == 0,
|
||||
// AVEVA refuses its native time-weighted modes on a discrete tag (StartQuery fails — the direct read
|
||||
// records it). Since #897 the node is refused BadAggregateNotSupported up front: nothing sent, no failure.
|
||||
Check(failures, p.Status == "BadAggregateNotSupported" && p.Rows.Count == 0,
|
||||
$"700.existing.processed.TimeAverage.boolean: {p.Status} rows={p.Rows.Count} (direct: {directError ?? "served"})");
|
||||
Check(failures, failuresAfter == failuresBefore,
|
||||
$"700.existing.processed.TimeAverage.boolean: the wire read counted {failuresAfter - failuresBefore} backend failure(s)");
|
||||
continue;
|
||||
}
|
||||
if (d is null) { failures.Add($"700.existing.processed.{aggName}.{label}: direct read failed: {directError}"); continue; }
|
||||
|
||||
+146
@@ -0,0 +1,146 @@
|
||||
using System.Runtime.CompilerServices;
|
||||
using Google.Protobuf.WellKnownTypes;
|
||||
using RpcException = Grpc.Core.RpcException;
|
||||
using RpcStatus = Grpc.Core.Status;
|
||||
using RpcStatusCode = Grpc.Core.StatusCode;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using Opc.Ua;
|
||||
using Opc.Ua.Client;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.HistorianGateway.Contracts.Grpc;
|
||||
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
|
||||
using ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway;
|
||||
using static ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests.ServerEventRig;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests;
|
||||
|
||||
/// <summary>
|
||||
/// lmxopcua#897 over a real SDK session, with the production Gateway backend over a deterministic AVEVA-shaped gateway:
|
||||
/// TimeAverage / Total on a Boolean node (AVEVA discrete tag, live <c>SysPulse</c>) are refused
|
||||
/// <c>BadAggregateNotSupported</c> up front — nothing is sent to the gateway and the historian health snapshot counts no
|
||||
/// failure — where AVEVA's refusal (gateway <c>Internal</c> / <c>HistorianServerOperationException</c>, the same
|
||||
/// answer as a historian storage fault) used to come back <c>BadHistoryOperationUnsupported</c> and count as a backend
|
||||
/// failure. A numeric node keeps the native pair, and the Boolean node keeps the raw-evaluated Average.
|
||||
/// </summary>
|
||||
public sealed class DiscreteTagTimeWeightedWireTests
|
||||
{
|
||||
private const string Discrete = "SysPulse";
|
||||
private const string Analog = "SysTimeSec";
|
||||
private static readonly DateTime S0 = new(2026, 11, 11, 14, 0, 0, DateTimeKind.Utc);
|
||||
|
||||
[Fact]
|
||||
public async Task TimeAverage_and_Total_on_a_Boolean_node_are_refused_up_front_and_never_count_as_a_backend_failure()
|
||||
{
|
||||
var ct = TestContext.Current.CancellationToken;
|
||||
await using var rig = await StartAsync("DiscreteTimeWeighted", encrypted: false, ct);
|
||||
var gateway = new AvevaShapedGateway();
|
||||
var source = new GatewayHistorianDataSource(gateway, NullLogger<GatewayHistorianDataSource>.Instance);
|
||||
rig.Nm.HistorianDataSource = source;
|
||||
rig.Enforce(false, [Key()]);
|
||||
var boolNode = rig.EnsureRawVariable("HistBool", "Boolean", writable: false, Discrete);
|
||||
var intNode = rig.EnsureRawVariable("HistInt", "Int32", writable: false, Analog);
|
||||
|
||||
using var session = await rig.ConnectAsync(anonymous: true, ct);
|
||||
var ns = (ushort)session.NamespaceUris.GetIndex(V3NodeIds.RawNamespaceUri);
|
||||
|
||||
foreach (var aggregate in new[] { ObjectIds.AggregateFunction_TimeAverage, ObjectIds.AggregateFunction_Total })
|
||||
{
|
||||
var (status, _) = await ReadProcessedAsync(session, new NodeId(boolNode, ns), aggregate, ct);
|
||||
status.Code.ShouldBe(StatusCodes.BadAggregateNotSupported.Code, $"{aggregate} on a Boolean node");
|
||||
}
|
||||
gateway.AggregateTags.ShouldNotContain(Discrete, "the refusal is decided before the gateway is called");
|
||||
var health = source.GetHealthSnapshot();
|
||||
health.TotalFailures.ShouldBe(0, "a refused aggregate is a client request, not a backend failure");
|
||||
health.ConsecutiveFailures.ShouldBe(0);
|
||||
health.LastError.ShouldBeNull();
|
||||
|
||||
// Positive controls: the native pair on a numeric node reaches the gateway and is served; the Boolean node's
|
||||
// raw-evaluated Average is served (0/1, as on the SQL backends).
|
||||
var (intStatus, intRows) = await ReadProcessedAsync(session, new NodeId(intNode, ns), ObjectIds.AggregateFunction_TimeAverage, ct);
|
||||
intStatus.Code.ShouldBe(StatusCodes.Good.Code);
|
||||
intRows.ShouldBeGreaterThan(0);
|
||||
gateway.AggregateTags.ShouldContain(Analog);
|
||||
var (avgStatus, avgRows) = await ReadProcessedAsync(session, new NodeId(boolNode, ns), ObjectIds.AggregateFunction_Average, ct);
|
||||
avgStatus.Code.ShouldBe(StatusCodes.Good.Code);
|
||||
avgRows.ShouldBe(2);
|
||||
source.GetHealthSnapshot().TotalFailures.ShouldBe(0);
|
||||
}
|
||||
|
||||
private static async Task<(StatusCode Status, int Rows)> ReadProcessedAsync(ISession session, NodeId node, NodeId aggregate,
|
||||
CancellationToken ct)
|
||||
{
|
||||
var details = new ReadProcessedDetails
|
||||
{
|
||||
StartTime = S0, EndTime = S0.AddMinutes(4), ProcessingInterval = 120_000,
|
||||
AggregateType = new List<NodeId> { aggregate },
|
||||
AggregateConfiguration = new AggregateConfiguration { UseServerCapabilitiesDefaults = true },
|
||||
};
|
||||
var response = await session.HistoryReadAsync(null, new ExtensionObject(details), TimestampsToReturn.Both, false,
|
||||
new HistoryReadValueId[] { new() { NodeId = node } }, ct);
|
||||
var result = response.Results.ToList().ShouldHaveSingleItem();
|
||||
var rows = !result.HistoryData.IsNull && ExtensionObject.ToEncodeable(result.HistoryData) is HistoryData data
|
||||
? data.DataValues.Count
|
||||
: 0;
|
||||
return (result.StatusCode, rows);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// AVEVA's behaviour as the live #694 gate saw it: the native TimeWeightedAverage / Integral modes on a discrete tag
|
||||
/// fail StartQuery, which the gateway answers <c>Internal</c> with the redacted detail
|
||||
/// <c>HistorianServerOperationException</c>; an analog tag answers one row per interval. Raw reads answer a 0/1
|
||||
/// toggle every 30 s for the discrete tag.
|
||||
/// </summary>
|
||||
private sealed class AvevaShapedGateway : IHistorianGatewayClient
|
||||
{
|
||||
private readonly List<string> _aggregateTags = [];
|
||||
|
||||
public IReadOnlyList<string> AggregateTags { get { lock (_aggregateTags) return _aggregateTags.ToList(); } }
|
||||
|
||||
public Task<bool> TagExistsAsync(string tag, CancellationToken ct) => Task.FromResult(true);
|
||||
|
||||
public async IAsyncEnumerable<HistorianAggregateSample> ReadAggregateAsync(string tag, DateTime startUtc, DateTime endUtc,
|
||||
RetrievalMode mode, TimeSpan interval, [EnumeratorCancellation] CancellationToken ct)
|
||||
{
|
||||
lock (_aggregateTags) _aggregateTags.Add(tag);
|
||||
await Task.Yield();
|
||||
if (tag == Discrete)
|
||||
throw new RpcException(new RpcStatus(RpcStatusCode.Internal, "HistorianServerOperationException"));
|
||||
for (var t = startUtc; t < endUtc; t += interval)
|
||||
yield return new HistorianAggregateSample
|
||||
{
|
||||
Value = 29.5, OpcQuality = 192, StartTime = Timestamp.FromDateTime(t), EndTime = Timestamp.FromDateTime(t + interval),
|
||||
};
|
||||
}
|
||||
|
||||
public async IAsyncEnumerable<HistorianSample> ReadRawAsync(string tag, DateTime startUtc, DateTime endUtc, int maxValues,
|
||||
[EnumeratorCancellation] CancellationToken ct)
|
||||
{
|
||||
await Task.Yield();
|
||||
var n = 0;
|
||||
for (var t = S0.AddMinutes(-10); t < S0.AddMinutes(10) && n < maxValues; t = t.AddSeconds(30))
|
||||
{
|
||||
if (t < startUtc || t > endUtc) continue;
|
||||
n++;
|
||||
yield return new HistorianSample
|
||||
{
|
||||
Tag = tag, Timestamp = Timestamp.FromDateTime(t), NumericValue = (t - S0).TotalSeconds / 30 % 2 == 0 ? 0 : 1,
|
||||
OpcQuality = 192, Quality = 0,
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
public Task<IReadOnlyList<HistorianSample>> ReadAtTimeAsync(string tag, IReadOnlyList<DateTime> timestampsUtc, CancellationToken ct) =>
|
||||
throw new NotSupportedException();
|
||||
public IAsyncEnumerable<HistorianEvent> ReadEventsAsync(string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents,
|
||||
CancellationToken ct) => throw new NotSupportedException();
|
||||
public Task<WriteAck> WriteLiveValuesAsync(string tag, IReadOnlyList<HistorianLiveValue> values, CancellationToken ct) =>
|
||||
throw new NotSupportedException();
|
||||
public Task<WriteAck> SendEventAsync(HistorianEvent evt, CancellationToken ct) => throw new NotSupportedException();
|
||||
public Task<TagOperationResults> EnsureTagsAsync(IReadOnlyList<HistorianTagDefinition> definitions, CancellationToken ct) =>
|
||||
throw new NotSupportedException();
|
||||
public Task<bool> ProbeAsync(CancellationToken ct) => Task.FromResult(true);
|
||||
public Task<ConnectionStatus> GetConnectionStatusAsync(CancellationToken ct) => Task.FromResult(new ConnectionStatus());
|
||||
public ValueTask DisposeAsync() => ValueTask.CompletedTask;
|
||||
}
|
||||
}
|
||||
@@ -979,6 +979,103 @@ public sealed class NodeManagerHistoryReadTests : IDisposable
|
||||
finally { await host.DisposeAsync(); }
|
||||
}
|
||||
|
||||
private static ReadProcessedDetails ProcessedDetails(NodeId aggregate) => new()
|
||||
{
|
||||
StartTime = DateTime.UtcNow.AddHours(-1),
|
||||
EndTime = DateTime.UtcNow,
|
||||
ProcessingInterval = 10_000.0,
|
||||
AggregateConfiguration = new AggregateConfiguration { UseServerCapabilitiesDefaults = true },
|
||||
AggregateType = new List<NodeId> { aggregate },
|
||||
};
|
||||
|
||||
/// <summary>
|
||||
/// lmxopcua#897 — Part 11 §6.5.4 ("If the Aggregate is not valid for the Node then the StatusCode shall be
|
||||
/// Bad_AggregateNotSupported") with Part 13 Table 50 (TimeAverage, Total, Average, … are numeric only): a String or
|
||||
/// DateTime node asked for a numeric aggregate is refused up front on EVERY backend, without a backend call, while
|
||||
/// an all-types aggregate (Count) on the same node and the numeric one on a numeric node still reach the backend.
|
||||
/// </summary>
|
||||
[Theory]
|
||||
[InlineData("String", "TimeAverage")]
|
||||
[InlineData("String", "Total")]
|
||||
[InlineData("String", "Average")]
|
||||
[InlineData("String", "Maximum")]
|
||||
[InlineData("DateTime", "TimeAverage")]
|
||||
[InlineData("DateTime", "Delta")]
|
||||
public async Task A_numeric_aggregate_on_a_non_numeric_node_is_refused_unsent(string dataType, string aggregate)
|
||||
{
|
||||
var (host, server) = await BootAsync();
|
||||
try
|
||||
{
|
||||
var nm = server.NodeManager!;
|
||||
var fake = new RecordingHistorianDataSource { ProcessedResult = Empty() };
|
||||
nm.HistorianDataSource = fake;
|
||||
nm.EnsureVariable("eq/text", null, "Text", dataType, false, historianTagname: "T.Text", realm: AddressSpaceRealm.Uns);
|
||||
nm.EnsureVariable("eq/num", null, "Num", "Double", false, historianTagname: "T.Num", realm: AddressSpaceRealm.Uns);
|
||||
var aggregateId = aggregate switch
|
||||
{
|
||||
"TimeAverage" => ObjectIds.AggregateFunction_TimeAverage,
|
||||
"Total" => ObjectIds.AggregateFunction_Total,
|
||||
"Average" => ObjectIds.AggregateFunction_Average,
|
||||
"Maximum" => ObjectIds.AggregateFunction_Maximum,
|
||||
_ => ObjectIds.AggregateFunction_Delta,
|
||||
};
|
||||
|
||||
var (_, errors) = InvokeHistoryRead(server, nm, ProcessedDetails(aggregateId), nm.TryGetVariable("eq/text")!.NodeId);
|
||||
errors[0].StatusCode.Code.ShouldBe(StatusCodes.BadAggregateNotSupported.Code);
|
||||
fake.LastCall.ShouldBeNull("the refusal is decided before any backend read");
|
||||
|
||||
// Positive controls: an all-types aggregate on the same node, and the same aggregate on a numeric node.
|
||||
InvokeHistoryRead(server, nm, ProcessedDetails(ObjectIds.AggregateFunction_Count), nm.TryGetVariable("eq/text")!.NodeId)
|
||||
.Item2[0].StatusCode.Code.ShouldBe(StatusCodes.Good.Code);
|
||||
fake.LastCall.ShouldBe("Processed");
|
||||
fake.LastAggregate.ShouldBe(HistoryAggregateType.Count);
|
||||
InvokeHistoryRead(server, nm, ProcessedDetails(aggregateId), nm.TryGetVariable("eq/num")!.NodeId)
|
||||
.Item2[0].StatusCode.Code.ShouldBe(StatusCodes.Good.Code);
|
||||
fake.LastTagname.ShouldBe("T.Num");
|
||||
}
|
||||
finally { await host.DisposeAsync(); }
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// lmxopcua#897: a backend's own per-type limit (<see cref="IHistorianDataSource.SupportsAggregateFor"/> — the
|
||||
/// Gateway refuses native TimeAverage / Total on a Boolean node) is consulted with the node's scalar type and answers
|
||||
/// <c>BadAggregateNotSupported</c> without a backend read; the same aggregate on another type, and another aggregate
|
||||
/// on the Boolean node, are served.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task A_backend_per_type_refusal_answers_BadAggregateNotSupported_unsent()
|
||||
{
|
||||
var (host, server) = await BootAsync();
|
||||
try
|
||||
{
|
||||
var nm = server.NodeManager!;
|
||||
var fake = new RecordingHistorianDataSource
|
||||
{
|
||||
ProcessedResult = Empty(),
|
||||
SupportsFor = (a, t) => !(t == TypeCode.Boolean && a == HistoryAggregateType.TimeAverage),
|
||||
};
|
||||
nm.HistorianDataSource = fake;
|
||||
nm.EnsureVariable("eq/pulse", null, "Pulse", "Boolean", false, historianTagname: "SysPulse", realm: AddressSpaceRealm.Uns);
|
||||
nm.EnsureVariable("eq/sec", null, "Sec", "Int32", false, historianTagname: "SysTimeSec", realm: AddressSpaceRealm.Uns);
|
||||
var pulse = nm.TryGetVariable("eq/pulse")!.NodeId;
|
||||
|
||||
var (_, errors) = InvokeHistoryRead(server, nm, ProcessedDetails(ObjectIds.AggregateFunction_TimeAverage), pulse);
|
||||
errors[0].StatusCode.Code.ShouldBe(StatusCodes.BadAggregateNotSupported.Code);
|
||||
fake.LastCall.ShouldBeNull();
|
||||
fake.SupportsForCalls.ShouldContain((HistoryAggregateType.TimeAverage, (TypeCode?)TypeCode.Boolean),
|
||||
"the node's scalar type is what the backend is asked about");
|
||||
|
||||
InvokeHistoryRead(server, nm, ProcessedDetails(ObjectIds.AggregateFunction_Average), pulse)
|
||||
.Item2[0].StatusCode.Code.ShouldBe(StatusCodes.Good.Code);
|
||||
fake.LastAggregate.ShouldBe(HistoryAggregateType.Average);
|
||||
InvokeHistoryRead(server, nm, ProcessedDetails(ObjectIds.AggregateFunction_TimeAverage), nm.TryGetVariable("eq/sec")!.NodeId)
|
||||
.Item2[0].StatusCode.Code.ShouldBe(StatusCodes.Good.Code);
|
||||
fake.LastTagname.ShouldBe("SysTimeSec");
|
||||
fake.LastAggregate.ShouldBe(HistoryAggregateType.TimeAverage);
|
||||
}
|
||||
finally { await host.DisposeAsync(); }
|
||||
}
|
||||
|
||||
private static HistorianRead Empty() => new(Array.Empty<DataValueSnapshot>(), null);
|
||||
|
||||
/// <summary>A recording fake historian source — captures the last call's kind + arguments and returns
|
||||
@@ -992,6 +1089,13 @@ public sealed class NodeManagerHistoryReadTests : IDisposable
|
||||
public bool LimitExceeded { get; init; }
|
||||
public IReadOnlySet<HistoryAggregateType>? Supported { get; init; } = HistoryAggregateTypes.All;
|
||||
public IReadOnlySet<HistoryAggregateType> SupportedAggregates => Supported!;
|
||||
public Func<HistoryAggregateType, TypeCode?, bool>? SupportsFor { get; init; }
|
||||
public List<(HistoryAggregateType Aggregate, TypeCode? Type)> SupportsForCalls { get; } = [];
|
||||
public bool SupportsAggregateFor(HistoryAggregateType aggregate, TypeCode? valueType)
|
||||
{
|
||||
lock (SupportsForCalls) SupportsForCalls.Add((aggregate, valueType));
|
||||
return SupportsFor?.Invoke(aggregate, valueType) ?? true;
|
||||
}
|
||||
public HistorianRead RawResult { get; set; } = new(Array.Empty<DataValueSnapshot>(), null);
|
||||
public HistorianRead ProcessedResult { get; set; } = new(Array.Empty<DataValueSnapshot>(), null);
|
||||
public HistorianRead AtTimeResult { get; set; } = new(Array.Empty<DataValueSnapshot>(), null);
|
||||
|
||||
Reference in New Issue
Block a user