diff --git a/CLAUDE.md b/CLAUDE.md index 9447335c..eb39bbc5 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -1353,6 +1353,7 @@ historian tagname, which is the tag's driver `FullName`. - **SQL Server** (`Driver.Historian.SqlServer`): `hist.Sample` PK `(TagId, SourceTimeUtc)` — writes are insert-if-absent, so outbox redelivery is idempotent. Stores every value kind with its **real status**; per-sample aggregates (10, simple definitions) are pushed down to T-SQL with bucket math in **100 ns ticks**; the Part 13 time-weighted seven (TimeAverage/TimeAverage2/Total/Total2/Interpolative, #671 — ⚠️ `Total` is now the time-weighted integral, no longer `SUM` — plus the **duration-based** PercentGood/PercentBad, #680) stream window + 4 bound rows into the shared `TimeWeightedAggregator` (both SQL backends, identical results; reverse `Interpolative` anchored at the later time; window capped by `ServerHistorian:MaxTimeWeightedRawRows` ⇒ `BadTooManyOperations`, percentages included). ⚠️ **#680 status rules (client-visible):** TimeAverage/Total are Uncertain if any part of the interval uses an Uncertain value, is bridged across a Bad sample, is extrapolated past the last sample (⇒ a live tag's latest interval is Uncertain) or is not covered; TimeAverage2/Total2 keep the SDK time-based rule (Uncertain time counts as Good) but never extrapolate past the last sample — that interval is Partial, later ones `BadNoData` (a report-by-exception tag with nothing after its last sample reads `BadNoData` until its next sample arrives). `docs/Historian.md` §"Time-weighted aggregates". ⚠️ **#680 per-sample (client-visible):** `Count` counts **Good** samples only (value-based status; an all-Bad interval is generic Bad, value null) and an empty interval **inside the tag's data range** is `0`/Good (outside it still `BadNoData`); `Start`/`End` return the **raw** first/last sample of any quality — its own timestamp and status, value null when Bad; `Delta` is Uncertain only when the interval's first or last sample is unusable; ⚠️ #684: `Average`/`Minimum`/`Maximum`/`Range`/`*ActualTime` use **Good samples only** (was Good + Uncertain; an Uncertain-only interval is `BadNoData`), Min/Max-family status Uncertain on a Bad sample, an Uncertain value beyond the Good extreme (Part 13 — the SDK ignores that) or an Uncertain non-numeric sample, and only a **Good** String/DateTime sample refuses them (`BadAggregateInvalidInputs`; Delta still refuses any usable one); and every per-sample aggregate but `Average` sets **end-of-data Partial** on the interval holding the last sample (forward incl. its earlier edge, reverse strictly inside), never on a Bad result. The data-range lookup is a key seek on SQL Server and a `tag` symbol-index seek on QuestDB (#683 — `{p}sample.tag` is `SYMBOL INDEX`; the bootstrapper ADDs it to a pre-#683 table, idempotent), read BEFORE the bucket statement so a racing append can never yield a Good `Count` of 0 (a concurrent HistoryUpdate delete is the mirror case and still can, transiently). `docs/Historian.md` §"Aggregates". Docker-dev recipe, schema and aggregate semantics: `docs/Historian.md` §"SQL Server backend". Integration tests need `HISTORIAN_SQL_TEST_CONNECTION`. **Retention (#667):** opt-in `SqlServer:RetentionDays` (+ `PurgeInterval` 1h, `PurgeBatchSize` 4000) -> `HistorianRetentionService` purges old `hist.Sample` (per tag) / `hist.Event` in `DELETE TOP(n)` batches, Primary-gated on `IHistorizationRoleView`, one purger per DB via `sp_getapplock`; counter `otopcua.historian.retention.purged{table}`. Optional `IHistorianBackendProvider.CreateRetention` (default null). `docs/Historian.md` §Retention. - **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 and carried page to page in the continuation point, ≤ `MaxTieClusterOverfetch + 2` rows per read — a full read is re-read, never answered truncated; only one timestamp holding more than max(1, `MaxTieClusterOverfetch`) samples ⇒ `BadTooManyOperations`; walks no later than now + 1 day and never before the Unix epoch). 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 still differ (end-only forward / explicit reverse refused on Gateway, #675). `docs/Historian.md` §"Open-ended and reverse raw reads". - **Advertised aggregates (#682):** `IHistorianDataSource.SupportedAggregates` (Gateway 4 via `GatewayHistorianDataSource.ServedAggregates`, SqlServer/QuestDB 17 via `AggregateFamilies.Served`, 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. - **Real status fix:** the recorder used to stamp every sample Good (`192`); it now carries the driver's status. ⚠️ On the **Gateway** backend this changes nothing stored: the gateway discards quality (C-002), so AVEVA still stores Good. - **Comms loss is historized as an outage marker** (`Empty` + `BadNoCommunication`, one per tag per outage) on SqlServer; dropped (`DroppedUnsupportedKind`) on Gateway. Outbox records are **format v2** — drain or delete `ContinuousHistorization:OutboxPath` before downgrading a node. diff --git a/docs/Historian.md b/docs/Historian.md index c1b85d3e..c780be2e 100644 --- a/docs/Historian.md +++ b/docs/Historian.md @@ -143,7 +143,7 @@ Processed read is `BadAggregateNotSupported` and no history aggregate functions | `AllowUntrustedServerCertificate` | bool | `false` | **Gateway backend only.** Accept a self-signed / untrusted server certificate (dev / on-prem only). | | `CaCertificatePath` | string\|null | `null` | **Gateway backend only.** PEM CA file pinning the gateway's TLS chain. Null/empty uses the OS trust store. | | `CallTimeout` | TimeSpan | `00:00:30` | **Gateway backend only.** Per-call deadline applied to each unary gateway read. A read for a tag the historian does not yet know about — e.g. a freshly-historized tag that has **not** been provisioned (see [Tag auto-provisioning](#tag-auto-provisioning-ensuretags)) — can run to this **full** deadline before the gateway errors, so an unprovisioned-tag HistoryRead can block for ~30 s. Lower this to fail faster, at the cost of truncating legitimately long reads. | -| `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 `BadHistoryOperationUnsupported`. Raise to handle abnormally large tie clusters; the default covers all normal-data cases. | +| `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). | @@ -612,7 +612,8 @@ paging time-based: > > **Cluster larger than `MaxTieClusterOverfetch`.** If the over-fetch itself reaches the ceiling > without spanning the full cluster, the node fails **loudly** with -> `BadHistoryOperationUnsupported` and the tag + timestamp + ceiling are logged. Remedies: raise +> `BadTooManyOperations` (a request limit; `BadHistoryOperationUnsupported` before #686 — events reads changed the same +> way) and the tag + timestamp + ceiling are logged. Remedies: raise > `MaxTieClusterOverfetch` (or `NumValuesPerNode`) to cover the full cluster, or investigate the > data anomaly (raw samples normally carry strictly increasing distinct timestamps). > @@ -620,6 +621,52 @@ paging time-based: > anomaly. The ceiling exists to bound server-side memory on pathological data, not to cap normal > operation. +**Open-ended and reverse raw reads (#669, #686).** With `NumValuesPerNode > 0` one bound may be unset (Part 11): a +start-only read pages **forward** to an open upper bound; an end-only read pages **backward** from `EndTime` — the reverse +window `(MinValue, EndTime]`, newest first, so the first page is the `NumValuesPerNode` values at or before `EndTime`. An +explicit `StartTime > EndTime` is the reverse window `(EndTime, StartTime]`, newest first. These semantics hold on **every** +backend: + +- **SqlServer / QuestDB** report `IHistorianDataSource.SupportsReverseRaw = true` and are sent the reverse window directly + (`… > lo AND … <= hi ORDER BY time DESC`). +- **Gateway** (the default backend) reports `false` — the gateway's `ReadRaw` streams ascending only and rejects + `start > end` with `InvalidArgument`. The server never sends it one (and the source refuses one unsent, not counted as a + health failure); it serves the reverse window from forward reads (`ReverseRawReader`): + - **Uncapped** (`NumValuesPerNode = 0`): **one** forward read of the whole window, filtered and reversed in memory — the + cost and memory of an uncapped forward read. + - **Capped: a bounded walk back.** It reads one segment `(segLo, segHi]` forward, keeps it newest-first, and steps to + the next older segment until the page is full or the window is exhausted. A segment is requested as + `[⌊segLo⌋, ⌈segHi + 1 tick⌉]`, both rounded **outward to the millisecond**, and filtered on the samples' own + timestamps — exact for an inclusive or exclusive forward end, and the row AVEVA fabricates at a query's (possibly + rounded-up) start can never land inside the segment. The first segment is **1 minute** (or the width the previous + page ended on — it rides the continuation point); each next width is aimed at about half the row budget from the + density just observed, growing at most ×4. So "the last N values" of a live tag is normally one read, a dense tag + costs about one read per page, and sparse history is found in a few dozen tiny ones. + - **Never answered from a truncated read.** The row budget is `M = max(1, MaxTieClusterOverfetch)`; a segment read asks + for `M + 2` rows (room for the fabricated row and one to see past the segment). Rows stream ascending, so a read is + complete when it came back short or its last row is past the segment; otherwise it is discarded — re-read once with + room for the rows before the segment that crowded it (a neighbouring tie cluster, at most `2M + 4` rows), or narrowed + to the density it showed. Only **one timestamp holding more than `M` samples** is refused, **`BadTooManyOperations`** + (a request limit, not a backend fault); a cluster of exactly `M`, or neighbouring clusters each within `M`, is served. + - **Bounds.** The walk starts no later than **now + 1 day** (a `StartTime = DateTime.MaxValue` read does not send + year-9999 query starts) and never reads before the **Unix epoch** (AVEVA holds nothing earlier): a walked-back read + returns only samples strictly after 1970-01-01T00:00:00Z and no later than now + 1 day. + + Paging is unchanged — a reverse cursor resumes as a fresh walk from its resume point. +- **Null** reports `true` (empty in either direction). + +> ⚠️ **Client-visible (#686).** Since #669 an end-only capped raw read, and any explicit reverse raw read, on the Gateway +> backend failed the node (`BadHistoryOperationUnsupported`) and was counted as a historian backend failure (the gateway +> refused `start > end`). Before #669 an end-only read on Gateway returned the **oldest** N values ascending. Both now return +> the Part 11 answer — the latest values at or before `EndTime`, newest first. **Events differ:** an end-only events read on +> Gateway still keeps its forward meaning and an explicit reverse events read is refused (see below). +> +> ⚠️ **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. + Continuation points are bound to the OPC UA session (the SDK's `ServerConfiguration.MaxHistoryContinuationPoints` cap, default 100, with oldest-eviction; points are disposed when the session closes). Resuming an unknown / evicted / released point returns diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/IHistorianDataSource.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/IHistorianDataSource.cs index 2cba63ed..926881c1 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/IHistorianDataSource.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/IHistorianDataSource.cs @@ -17,6 +17,12 @@ public interface IHistorianDataSource : IDisposable /// /// Read raw historical samples for a single tag over a time range. /// + /// + /// Forward (startUtc < endUtc) is [start, end) ascending; startUtc == endUtc is a point read. + /// startUtc > endUtc is the Part 11 reverse read (end, start] newest-first — but only a backend that + /// declares is ever sent one (Gitea #686); the server serves a reverse read on any + /// other backend from bounded forward reads. + /// /// The full reference of the tag to read. /// The start of the time range in UTC. /// The end of the time range in UTC. @@ -30,6 +36,16 @@ public interface IHistorianDataSource : IDisposable uint maxValuesPerNode, CancellationToken cancellationToken); + /// + /// Whether this backend can serve a reverse (newest-first) raw read itself — with + /// startUtc > endUtc (Gitea #686). The server checks it before every raw read. When false it never + /// sends start > end: it serves the reverse read — an explicit one, a resumed reverse page, and the + /// end-only capped read Part 11 reads backward from EndTime — from a series of bounded FORWARD reads walking back + /// from the later bound (ReverseRawReader in the OPC UA server). Default false, the safe direction: a + /// source that only reads forward is still answered correctly. A backend declares it itself (a guard test pins that). + /// + bool SupportsReverseRaw => false; + /// /// Read processed (interval-bucketed) samples — average / min / max / count / etc. /// A bucket with no source data returns a sample whose diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/NullHistorianDataSource.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/NullHistorianDataSource.cs index d0553988..769f1d4e 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/NullHistorianDataSource.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/NullHistorianDataSource.cs @@ -55,6 +55,11 @@ public sealed class NullHistorianDataSource : IHistorianDataSource int maxEvents, CancellationToken cancellationToken) => Task.FromResult(EmptyEvents); + /// + /// true (Gitea #686): there is no history, so the empty result is the correct answer in either + /// direction — no need for the server to walk back through forward reads to find nothing. + public bool SupportsReverseRaw => true; + /// /// false (Gitea #675): there is no history to read in either direction, and an explicit reverse read /// is answered BadHistoryOperationUnsupported like any backend that cannot read newest-first. 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 aa87cbef..f84df290 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 @@ -76,14 +76,23 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis } /// + /// + /// A reverse window (start > end) is refused with before anything is sent + /// and is not recorded as a health failure (Gitea #686): the gateway rejects it InvalidArgument, and the server + /// never asks — is false, so it walks a reverse read back with forward reads. + /// public async Task ReadRawAsync( string fullReference, DateTime startUtc, DateTime endUtc, uint maxValuesPerNode, CancellationToken cancellationToken) { + if (startUtc > endUtc) + throw new NotSupportedException("The HistorianGateway reads raw history oldest-first only (start > end)."); + try { - // The gateway seam caps with an int; OPC UA hands us a uint, so clamp to int range. - var maxValues = (int)Math.Min(maxValuesPerNode, int.MaxValue); + // The gateway seam caps with an int; OPC UA hands us a uint, so clamp to int range. 0 ("all values") is sent + // unbounded: the client package and the gateway reject max_values <= 0 (#686 review). + var maxValues = maxValuesPerNode == 0 ? int.MaxValue : (int)Math.Min(maxValuesPerNode, int.MaxValue); var samples = new List(); await foreach (var sample in _client .ReadRawAsync(fullReference, startUtc, endUtc, maxValues, cancellationToken) @@ -96,6 +105,18 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis RecordOutcome(success: true, error: null); return new HistoryReadResult(snapshots, ContinuationPoint: null); } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + throw; // the caller (client cancel / HistoryReadDeadline) asked for this — not a backend fault + } + catch (Exception ex) when (IsMaxValuesCeilingRejection(ex)) + { + // An unbounded read over Historian:ReadLimits:MaxValuesCeiling with AllowStreamingBeyondCeiling=false: a + // client-request limit (BadTooManyOperations), not a backend fault. + throw new HistoryReadLimitExceededException(maxValuesPerNode, + $"The HistorianGateway refused the row budget of the raw read of '{fullReference}' (its " + + "Historian:ReadLimits:MaxValuesCeiling); set NumValuesPerNode or narrow the window."); + } catch (Exception ex) { RecordReadFailure(ex); @@ -235,6 +256,14 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis } } + /// + /// + /// false (Gitea #686): the gateway's ReadRaw streams ascending only and rejects end < start + /// (InvalidArgument). The server serves reverse and end-only capped raw reads from bounded forward reads + /// instead, and never sends this source a reverse window ( refuses one unsent). + /// + public bool SupportsReverseRaw => false; + /// /// /// false: the gateway's ReadEvents streams ascending only (Gitea #675). The server therefore refuses an diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataSource.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataSource.cs index 5bc333ea..07256c45 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataSource.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataSource.cs @@ -355,6 +355,10 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource // ------------------------------------------------------------------------------------------ Events + /// + /// Reverse raw reads are served natively ((end, start], ORDER BY … DESC; #669/#686). + public bool SupportsReverseRaw => true; + /// /// Reverse events reads are served natively (ORDER BY … DESC, Gitea #675). public bool SupportsReverseEvents => true; diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianDataSource.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianDataSource.cs index 4c7f1a44..27aeaf4c 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianDataSource.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianDataSource.cs @@ -312,6 +312,10 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource // ------------------------------------------------------------------------------------------ Events + /// + /// Reverse raw reads are served natively ((end, start], ORDER BY … DESC; #669/#686). + public bool SupportsReverseRaw => true; + /// /// Reverse events reads are served natively (ORDER BY … DESC, Gitea #675). public bool SupportsReverseEvents => true; diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/HistoryPaging.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/HistoryPaging.cs index 667d16a1..fe825f0e 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/HistoryPaging.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/HistoryPaging.cs @@ -51,6 +51,10 @@ namespace ZB.MOM.WW.OtOpcUa.OpcUaServer; /// (EndUtc, NextStartUtc]. A resumed events read takes its direction from here, never from the resume /// request's details. Raw keeps deriving it from the window (NextStartUtc > EndUtc). /// +/// +/// reverse reads on a forward-only backend only (Gitea #686): the segment width the +/// walk-back ended on, so the next page starts at the observed density instead of re-discovering it. 0 = none. +/// internal sealed record HistoryContinuationState( string Tagname, DateTime NextStartUtc, @@ -60,7 +64,8 @@ internal sealed record HistoryContinuationState( HistoryCursorKind Kind = HistoryCursorKind.Raw, long LastModifiedId = 0, DateTime WindowStartUtc = default, - bool Reverse = false); + bool Reverse = false, + long SegmentWidthTicks = 0); /// The HistoryRead kind that issued a . internal enum HistoryCursorKind diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs index 3ae1fe4e..09b486f8 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs @@ -536,8 +536,9 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 /// cap. When a resume read stalls on such a cluster (the boundary-tie trim empties the page), the /// paging over-fetches up to MaxTieClusterOverfetch + 1 ties at that single timestamp (a /// start == end read) and slices through them via . - /// A cluster strictly larger than this still surfaces BadHistoryOperationUnsupported for that - /// node (the absurd-burst backstop). Mirrors the configured + /// A cluster strictly larger than this still fails that node BadTooManyOperations (the absurd-burst + /// backstop; BadHistoryOperationUnsupported before #686). Also the per-read row budget of a reverse raw read + /// walked back from forward reads (). Mirrors the configured /// ServerHistorianOptions.MaxTieClusterOverfetch; the Host sets it at StartAsync. The /// default (65536) survives when the historian section is absent. /// @@ -5359,8 +5360,8 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 "OtOpcUaNodeManager: HistoryReadEvents tie cluster at {StartUtc:O} for source '{Source}' has {Count} events, " + "exceeding MaxTieClusterOverfetch={Max}; cannot page within it. Increase MaxTieClusterOverfetch.", startUtc, sourceName, cluster.Count, MaxTieClusterOverfetch); - errors[handle.Index] = StatusCodes.BadHistoryOperationUnsupported; - results[handle.Index] = new SdkHistoryReadResult { StatusCode = StatusCodes.BadHistoryOperationUnsupported }; + errors[handle.Index] = StatusCodes.BadTooManyOperations; + results[handle.Index] = new SdkHistoryReadResult { StatusCode = StatusCodes.BadTooManyOperations }; return; } @@ -6183,6 +6184,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 { DateTime startUtc; var boundarySkip = 0; + long segmentWidth = 0; // #686: the walk-back's width, carried page to page string tagname; if (inboundCp is { Length: > 0 }) @@ -6203,6 +6205,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 boundarySkip = state.BoundarySkipCount; endUtc = state.EndUtc; numValuesPerNode = state.NumValuesPerNode; + segmentWidth = state.SegmentWidthTicks; } else { @@ -6219,6 +6222,8 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 // Part 11: with NumValuesPerNode > 0 one bound may be left unset. Only a StartTime ⇒ page FORWARD from // it to an open upper bound; only an EndTime ⇒ page BACKWARD from it (a reverse read, start > end). // Without this an unset EndTime (MinValue) looked like a reverse read to a lower bound of "now". + // Every backend serves the backward read (#686): natively when it declares SupportsReverseRaw, otherwise + // walked back from bounded forward reads below — unlike the events path, which keeps it forward there. if (numValuesPerNode > 0) { if (endUtc == DateTime.MinValue && startUtc != DateTime.MinValue) @@ -6229,9 +6234,26 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 } // HistoryRead is NOT under the node-manager Lock — awaiting the async source is safe. - var sourceResult = await HistorianDataSource - .ReadRawAsync(tagname, startUtc, endUtc, numValuesPerNode, ct) - .ConfigureAwait(false); + // A reverse window (explicit, resumed, or the end-only read above) goes to the backend only when it declares it + // can read newest-first; otherwise it is walked back from bounded forward reads (#686 — the Gateway rejects + // start > end, which failed the node and counted as a backend fault). + var source = HistorianDataSource; + HistorianRead sourceResult; + if (startUtc > endUtc && !source.SupportsReverseRaw) + { + var page = await ReverseRawReader.ReadAsync( + source, tagname, startUtc, endUtc, numValuesPerNode, MaxTieClusterOverfetch, + Clock.GetUtcNow().UtcDateTime, segmentWidth, ct) + .ConfigureAwait(false); + segmentWidth = page.WidthHintTicks; + sourceResult = new HistorianRead(page.Samples, ContinuationPoint: null); + } + else + { + sourceResult = await source + .ReadRawAsync(tagname, startUtc, endUtc, numValuesPerNode, ct) + .ConfigureAwait(false); + } var backendFull = HistoryPaging.IsFullPage(sourceResult.Samples.Count, numValuesPerNode); @@ -6280,8 +6302,8 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 "OtOpcUaNodeManager: HistoryReadRaw tie cluster at {StartUtc:O} for tag '{Tagname}' has {Count} samples, " + "exceeding MaxTieClusterOverfetch={Max}; cannot page within it. Increase MaxTieClusterOverfetch.", startUtc, tagname, cluster.Count, MaxTieClusterOverfetch); - errors[handle.Index] = StatusCodes.BadHistoryOperationUnsupported; - results[handle.Index] = new SdkHistoryReadResult { StatusCode = StatusCodes.BadHistoryOperationUnsupported }; + errors[handle.Index] = StatusCodes.BadTooManyOperations; + results[handle.Index] = new SdkHistoryReadResult { StatusCode = StatusCodes.BadTooManyOperations }; return; } @@ -6302,7 +6324,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 if (nextStartUtc is { } resumeAt) { var clusterState = new HistoryContinuationState( - tagname, resumeAt, endUtc, nextSkip, numValuesPerNode); + tagname, resumeAt, endUtc, nextSkip, numValuesPerNode, SegmentWidthTicks: segmentWidth); clusterCp = _historyContinuationStore.Save(session, clusterState); } @@ -6325,7 +6347,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 { HistoryPaging.ComputeResumeCursor(samples, out var nextStart, out var skip); var nextState = new HistoryContinuationState( - tagname, nextStart, endUtc, skip, numValuesPerNode); + tagname, nextStart, endUtc, skip, numValuesPerNode, SegmentWidthTicks: segmentWidth); // Save may return null (no session on this request) ⇒ degrade to single-shot for this node. // Built AFTER historyData so a failure projecting samples can never orphan a stored cursor. outboundCp = _historyContinuationStore.Save(session, nextState); @@ -6348,6 +6370,14 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 errors[handle.Index] = StatusCodes.BadTimeout; results[handle.Index] = new SdkHistoryReadResult { StatusCode = StatusCodes.BadTimeout }; } + catch (HistoryReadLimitExceededException ex) + { + // A request limit (#686: a reverse read walked back from forward reads met a tie cluster over the row + // budget) — BadTooManyOperations for THIS node, as on the processed path; not a backend fault. + _logger.LogWarning("OtOpcUaNodeManager: HistoryReadRaw limit exceeded for node {NodeId}: {Message}", handle.NodeId, ex.Message); + errors[handle.Index] = StatusCodes.BadTooManyOperations; + results[handle.Index] = new SdkHistoryReadResult { StatusCode = StatusCodes.BadTooManyOperations }; + } catch (Exception ex) { // One node's backend failure must not throw out of the batch — Bad for THIS node only. diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ReverseRawReader.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ReverseRawReader.cs new file mode 100644 index 00000000..3ebea102 --- /dev/null +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ReverseRawReader.cs @@ -0,0 +1,204 @@ +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.OpcUaServer; + +/// One walked-back reverse raw page (). +/// At most the requested cap, newest first. +/// The segment width the walk ended on — the next page's starting width (0 = none). +internal readonly record struct ReverseRawPage(IReadOnlyList Samples, long WidthHintTicks); + +/// +/// Gitea #686: serves a Part 11 reverse raw read — (lo, hi] newest-first, capped — on a historian backend that can +/// only read forward ( = false: the Gateway, whose +/// ReadRaw streams ascending and rejects start > end). +/// +/// +/// +/// Uncapped (maxValues = 0): ONE forward read of the whole window, filtered and reversed in memory — the +/// same rows, round trips and memory as an uncapped forward read (and as a native backend's uncapped reverse read). +/// +/// +/// Capped: walk back in segments. Starting at hi, read one segment (segLo, segHi] FORWARD, keep +/// it newest-first, and step to the next older segment until the cap is met or lo is reached. The first +/// segment is wide, or the width the previous page of the same read ended on (carried in +/// the continuation point). After each read the next width is re-aimed at about half the row budget from the density +/// just observed, growing at most ×4 per step — so a live tag's "latest N" is usually one read, a dense tag does not +/// throw away a full read per page, and sparse history is found in a few dozen tiny reads. +/// +/// +/// Never answer from a truncated read. The row budget is M = max(1, MaxTieClusterOverfetch); a segment +/// read asks for M + 2 rows (room for the row an AVEVA read fabricates at its query start, and one more to see +/// past the segment). Rows stream ascending, so a read is complete for its segment when it came back short OR its last +/// row is already past segHi. An incomplete read is discarded: when rows BEFORE the segment (a neighbouring +/// tie cluster inside the alignment slack, the fabricated row) crowded it out it is re-read once with room for them +/// (at most 2M + 4 rows); otherwise the segment is narrowed (to about half the budget at the density the +/// truncated read showed) and re-read. Only a single timestamp holding more than M samples — a segment already +/// one timestamp-resolution wide that still does not fit — is refused: +/// (⇒ BadTooManyOperations, a request limit, never a backend fault). +/// +/// +/// Edges. A segment is requested as [⌊segLo⌋, ⌈segHi + 1 tick⌉], both rounded outward to the millisecond +/// (the gateway's / AVEVA's resolution), and filtered to (segLo, segHi] on the samples' own timestamps. That is +/// exact whether the backend's forward end is exclusive (the interface contract) or inclusive (the gateway) and +/// whether it truncates its bounds (QuestDB-style); and because the request start is already on a millisecond, a row +/// an AVEVA read fabricates at its (possibly rounded-up) query start can never land inside the segment (#687). +/// Adjacent segments share no timestamp, so nothing is returned twice. +/// +/// +/// Ties come back in the backend's forward order reversed — the same order the node manager's tie-cluster +/// over-fetch uses for a reverse read — so a resumed page's boundary-tie skip lines up. +/// +/// +/// Bounds. The walk never asks for data after now + (one day — the historian stores +/// nothing further ahead; a StartTime = DateTime.MaxValue read would otherwise send centuries-ahead query +/// starts), nor before (the Unix epoch — AVEVA holds nothing earlier, and an end-only read's year-1 +/// lower bound is outside the historian's time encoding). A walked-back read therefore returns only samples strictly +/// after the epoch and no later than now + one day. +/// +/// +internal static class ReverseRawReader +{ + /// The first segment's width when no hint is carried. + internal static readonly TimeSpan InitialSegment = TimeSpan.FromMinutes(1); + + /// The walk's exclusive lower limit (see the remarks). + internal static readonly DateTime Floor = DateTime.UnixEpoch; + + /// How far past "now" the walk starts at most (the same horizon as the Gateway's raw-aggregate look-ahead). + internal static readonly TimeSpan LookAhead = TimeSpan.FromDays(1); + + /// Request bounds are rounded outward to this (the gateway / AVEVA timestamp resolution). + private const long Alignment = TimeSpan.TicksPerMillisecond; + + /// Reads (lo, hi] newest-first from forward reads. + /// A backend that reads forward. + /// Historian tag. + /// The later (inclusive) bound — the request's StartTime or a reverse resume point. + /// The earlier (exclusive) bound. + /// Cap; 0 = every sample in the window (one forward read). + /// Row budget per segment read (MaxTieClusterOverfetch; values below 1 count as 1). + /// Now, for the look-ahead clamp. + /// Starting segment width (the previous page's ), 0 = none. + /// Cancellation. + /// The page, newest first, and the width to start the next page with. + /// One timestamp holds more than the row budget. + public static async Task ReadAsync( + IHistorianDataSource source, string tagname, DateTime hi, DateTime lo, uint maxValues, int maxTieCluster, + DateTime nowUtc, long widthHintTicks, CancellationToken ct) + { + var resolution = Math.Max(1L, source.TimestampResolutionTicks); + var granularity = Math.Max(Alignment, resolution); + var budget = Math.Max(1, maxTieCluster); + var result = new List(); + + var horizon = nowUtc.Ticks > DateTime.MaxValue.Ticks - LookAhead.Ticks ? DateTime.MaxValue.Ticks : nowUtc.Ticks + LookAhead.Ticks; + var hiTicks = Math.Min(hi.Ticks, horizon); + var stopTicks = Math.Max(lo.Ticks, Floor.Ticks); + if (hiTicks <= stopTicks) return new ReverseRawPage(result, widthHintTicks); + + if (maxValues == 0) + { + // One forward read of the whole window, reversed — the uncapped forward read's own cost and bound. + var all = (await source.ReadRawAsync(tagname, Utc(AlignDown(stopTicks, granularity)), + Utc(AlignUp(hiTicks + 1, granularity)), 0, ct).ConfigureAwait(false)).Samples; + for (var i = all.Count - 1; i >= 0; i--) + { + var ts = Ticks(all[i]); + if (ts > stopTicks && ts <= hiTicks) result.Add(all[i]); + } + + return new ReverseRawPage(result, widthHintTicks); + } + + var maxWidth = DateTime.MaxValue.Ticks; + var width = widthHintTicks > 0 ? Math.Clamp(widthHintTicks, resolution, maxWidth) : Math.Max(resolution, InitialSegment.Ticks); + var baseCap = budget + 2; + var maxCap = 2 * budget + 4; + var segHi = hiTicks; + + while (segHi > stopTicks && result.Count < maxValues) + { + var segLo = segHi - stopTicks <= width ? stopTicks : segHi - width; + var start = Utc(AlignDown(segLo, granularity)); + var end = Utc(AlignUp(segHi + 1, granularity)); + var cap = baseCap; + IReadOnlyList rows; + var narrowed = false; + + while (true) + { + ct.ThrowIfCancellationRequested(); + rows = (await source.ReadRawAsync(tagname, start, end, (uint)cap, ct).ConfigureAwait(false)).Samples; + if (rows.Count < cap || Ticks(rows[^1]) > segHi) break; // complete for (segLo, segHi] + + var leading = 0; + var inner = 0; + var firstInner = 0L; + foreach (var r in rows) + { + var ts = Ticks(r); + if (ts <= segLo) leading++; + else if (ts <= segHi && inner++ == 0) firstInner = ts; + } + + // Crowded out by rows before the segment, with the segment itself possibly within budget: re-read once + // with room for them. + var roomier = leading + budget + 2; + if (leading > 0 && inner <= budget && roomier > cap && roomier <= maxCap) + { + cap = roomier; + continue; + } + + if (width <= resolution) + throw new HistoryReadLimitExceededException(budget, + $"A reverse raw read of '{tagname}' at {Utc(segHi):O} found more than {budget} samples at one " + + "timestamp; the backend reads forward only, so the latest of them cannot be selected. Increase " + + "ServerHistorian:MaxTieClusterOverfetch."); + + // Narrow to about half the budget at the density the truncated read's own samples showed (their span, not + // the segment's — the segment may open on an empty stretch), at least halving. + var lastTs = Ticks(rows[^1]); + var aimed = inner > 1 && lastTs > firstInner + ? (long)Math.Min((double)(lastTs - firstInner) * (budget / 2.0) / (inner - 1), width / 2.0) + : width / 4; + width = Math.Max(resolution, aimed); + narrowed = true; + break; + } + + if (narrowed) continue; // re-read the same segHi, narrower + + var count = 0; + for (var i = rows.Count - 1; i >= 0; i--) + { + var ts = Ticks(rows[i]); + if (ts <= segLo || ts > segHi) continue; + count++; + if (result.Count < maxValues) result.Add(rows[i]); + } + + // Re-aim at about half the budget from the density just seen; grow at most x4. + var grown = width > maxWidth / 4 ? maxWidth : width * 4; + width = count == 0 + ? grown + : (long)Math.Clamp((double)width * Math.Max(1.0, budget / 2.0) / count, resolution, grown); + segHi = segLo; + } + + return new ReverseRawPage(result, width); + } + + private static long Ticks(DataValueSnapshot s) => (s.SourceTimestampUtc ?? DateTime.MinValue).Ticks; + + private static DateTime Utc(long ticks) => new(ticks, DateTimeKind.Utc); + + private static long AlignDown(long ticks, long step) => ticks - ticks % step; + + private static long AlignUp(long ticks, long step) + { + var down = AlignDown(ticks, step); + if (down == ticks) return ticks; + return down > DateTime.MaxValue.Ticks - step ? DateTime.MaxValue.Ticks : down + step; + } +} diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ZB.MOM.WW.OtOpcUa.OpcUaServer.csproj b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ZB.MOM.WW.OtOpcUa.OpcUaServer.csproj index 52dc5e0d..8d8ae722 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ZB.MOM.WW.OtOpcUa.OpcUaServer.csproj +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ZB.MOM.WW.OtOpcUa.OpcUaServer.csproj @@ -35,6 +35,8 @@ + + diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ServerHistorianOptions.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ServerHistorianOptions.cs index eea6046f..15b08763 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ServerHistorianOptions.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ServerHistorianOptions.cs @@ -81,8 +81,11 @@ public sealed class ServerHistorianOptions /// cap). When a resume read stalls on such a cluster, the node manager over-fetches up to /// MaxTieClusterOverfetch + 1 ties at that single timestamp (a start == end read) and /// pages through them; this bounds how large a single-timestamp burst the server will buffer in - /// memory before it gives up and surfaces BadHistoryOperationUnsupported for that node. The - /// default (65536) comfortably covers any realistic same-millisecond burst. + /// memory before it gives up and surfaces BadTooManyOperations for that node (#686; was + /// BadHistoryOperationUnsupported). It is also the per-read row budget of a reverse raw read walked + /// back from forward reads on a backend that cannot read newest-first (the Gateway): one timestamp holding + /// more samples than this is refused BadTooManyOperations. The default (65536) comfortably covers + /// any realistic same-millisecond burst; a value below 1 is treated as 1. /// public int MaxTieClusterOverfetch { get; init; } = 65536; @@ -167,7 +170,7 @@ public sealed class ServerHistorianOptions // the over-fetch code path only runs when a real IHistorianDataSource is wired in, // so a zero/negative value is harmless (and noise-free) when the historian is disabled. if (MaxTieClusterOverfetch <= 0) - warnings.Add($"ServerHistorian:MaxTieClusterOverfetch is {MaxTieClusterOverfetch} — must be > 0; HistoryRead-Raw cannot page within an oversized tie cluster and will surface BadHistoryOperationUnsupported for those reads."); + warnings.Add($"ServerHistorian:MaxTieClusterOverfetch is {MaxTieClusterOverfetch} — must be > 0 (treated as 1); HistoryRead-Raw and -Events cannot page within a tie cluster larger than that, and a Gateway reverse raw read cannot serve one timestamp holding more samples — those reads surface BadTooManyOperations."); // S3 HistoryRead concurrency knobs — a non-positive value disables the corresponding bound, so warn // (these gate the per-node fan-out / batch limiter that only run when a real data source is wired). if (HistoryReadBatchParallelism <= 0) diff --git a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions.Tests/Historian/HistorianDataSourceReverseEventsGuardTests.cs b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions.Tests/Historian/HistorianDataSourceReverseEventsGuardTests.cs index fdc4d365..107f72ee 100644 --- a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions.Tests/Historian/HistorianDataSourceReverseEventsGuardTests.cs +++ b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions.Tests/Historian/HistorianDataSourceReverseEventsGuardTests.cs @@ -64,6 +64,22 @@ public sealed class HistorianDataSourceReverseEventsGuardTests } } + [Fact] + public void Every_implementation_declares_its_reverse_raw_capability_itself() + { + // Gitea #686: the default (false) is safe — the server walks back with forward reads — but a backend that serves + // reverse natively and inherited it would silently lose its native path, and a wrapper of the Gateway that claimed + // true would send start > end to a gateway that rejects it. Declare it, both ways. + foreach (var type in ProductionImplementations()) + { + var map = type.GetInterfaceMap(typeof(IHistorianDataSource)); + var getter = map.TargetMethods[Array.FindIndex(map.InterfaceMethods, + m => m.Name == "get_" + nameof(IHistorianDataSource.SupportsReverseRaw))]; + getter.DeclaringType.ShouldBe(type, + $"{type.FullName} must declare SupportsReverseRaw itself, not inherit the interface default"); + } + } + [Fact] public void Every_implementation_declares_its_supported_aggregates_itself() { diff --git a/tests/Drivers/Historian.Conformance/RawReadConformance.cs b/tests/Drivers/Historian.Conformance/RawReadConformance.cs index 41282291..8a04adc6 100644 --- a/tests/Drivers/Historian.Conformance/RawReadConformance.cs +++ b/tests/Drivers/Historian.Conformance/RawReadConformance.cs @@ -56,6 +56,14 @@ public abstract class RawReadConformance : ConformanceBase Values(await B.CreateDataSource().ReadRawAsync("A", At(10), max, 0, Ct)).ShouldBe([20d, 30d, 40d, 50d]); } + [Fact] + public void The_backend_reports_reverse_raw_support() + { + // #686: the server only sends start > end to a backend that declares it; these backends serve it natively. + using var src = B.CreateDataSource(); + src.SupportsReverseRaw.ShouldBeTrue(); + } + [Fact] public async Task Reverse_read_is_descending_over_end_exclusive_start_inclusive() { diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayRawDirectionTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayRawDirectionTests.cs new file mode 100644 index 00000000..d8f0b4eb --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayRawDirectionTests.cs @@ -0,0 +1,185 @@ +using Google.Protobuf.WellKnownTypes; +using Grpc.Core; +using Microsoft.Extensions.Logging.Abstractions; +using Shouldly; +using Xunit; +using ZB.MOM.WW.HistorianGateway.Client; +using ZB.MOM.WW.HistorianGateway.Contracts.Grpc; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.OpcUaServer; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests; + +/// +/// Gitea #686: the gateway's ReadRaw streams ascending only and rejects start > end with +/// InvalidArgument, which used to be recorded as a backend health failure. The source declares it cannot read +/// newest-first (so the server walks back with forward reads), and refuses a reverse window itself without sending +/// anything or counting a failure. The walk itself is driven here over the REAL source and an AVEVA-shaped fake. +/// +public sealed class GatewayRawDirectionTests +{ + private static CancellationToken Ct => TestContext.Current.CancellationToken; + + private static readonly DateTime T0 = new(2026, 10, 12, 18, 6, 0, DateTimeKind.Utc); + + private static GatewayHistorianDataSource Source(FakeHistorianGatewayClient fake) => + new(fake, NullLogger.Instance); + + [Fact] + public void The_gateway_source_declares_it_cannot_read_raw_newest_first() + { + IHistorianDataSource ds = Source(new FakeHistorianGatewayClient()); + ds.SupportsReverseRaw.ShouldBeFalse(); + } + + [Fact] + public async Task A_reverse_raw_window_is_refused_unsent_and_is_not_a_backend_failure() + { + var fake = new FakeHistorianGatewayClient(); + IHistorianDataSource ds = Source(fake); + var end = new DateTime(2026, 10, 12, 18, 7, 45, DateTimeKind.Utc); + + await Should.ThrowAsync(() => + ds.ReadRawAsync("Gate681", end, new DateTime(0, DateTimeKind.Utc), 3, Ct)); + + fake.ReadRawCallCount.ShouldBe(0); + var health = ds.GetHealthSnapshot(); + health.TotalFailures.ShouldBe(0); + health.ConsecutiveFailures.ShouldBe(0); + + // Positive control: a forward read of the same window is sent. + await ds.ReadRawAsync("Gate681", new DateTime(2026, 10, 12, 18, 0, 0, DateTimeKind.Utc), end, 3, Ct); + fake.ReadRawCallCount.ShouldBe(1); + } + + /// + /// An uncapped raw read (NumValuesPerNode = 0, "all values") is sent as an unbounded max_values. The + /// client package and the gateway both reject max_values <= 0, so forwarding 0 failed every uncapped raw + /// read — forward, and the one-shot read an uncapped reverse read is served from (#686 review). + /// + [Fact] + public async Task An_uncapped_raw_read_asks_the_gateway_for_an_unbounded_max_values() + { + var fake = new FakeHistorianGatewayClient(); + await Source(fake).ReadRawAsync("T", T0, T0.AddHours(1), 0, Ct); + fake.LastReadRawMaxValues.ShouldBe(int.MaxValue); + } + + [Fact] + public async Task A_deadline_cancellation_is_not_a_backend_failure() + { + var fake = new FakeHistorianGatewayClient { RawSamples = [Wire(T0, 1)] }; + var ds = Source(fake); + using var cts = new CancellationTokenSource(); + await cts.CancelAsync(); + + await Should.ThrowAsync(() => ds.ReadRawAsync("T", T0, T0.AddHours(1), 10, cts.Token)); + + ds.GetHealthSnapshot().TotalFailures.ShouldBe(0); + } + + [Fact] + public async Task The_gateways_max_values_ceiling_rejection_of_a_raw_read_is_a_request_limit_not_a_backend_fault() + { + var rpc = new RpcException(new Status(StatusCode.InvalidArgument, + "max_values exceeds the server ceiling of 1000000; read large ranges by time-slicing ...")); + var ds = Source(new FakeHistorianGatewayClient { ThrowOnRead = new HistorianGatewayException(rpc.Status.Detail, rpc) }); + + await Should.ThrowAsync(() => ds.ReadRawAsync("T", T0, T0.AddHours(1), 0, Ct)); + ds.GetHealthSnapshot().TotalFailures.ShouldBe(0); + } + + // ---- the walk-back over the real source --------------------------------------------------------------------------- + + private static HistorianSample Wire(DateTime t, double v, uint opc = 192) => + new() { Tag = "T", Timestamp = Timestamp.FromDateTime(t), NumericValue = v, OpcQuality = opc }; + + /// + /// A gateway over stored samples (value i at T0 + i s, i = 0..299) shaped like AVEVA: ReadRaw rejects + /// start > end and max_values <= 0, returns [start, end] ascending capped at + /// max_values, and first emits a row stamped at the query start carrying the value in effect there + /// (Quality = 133), or a NULL row (OpcQuality = 0, value 0) when there is none. + /// + private static FakeHistorianGatewayClient AvevaStore() + { + var stored = Enumerable.Range(0, 300).Select(i => (Time: T0.AddSeconds(i), Value: (double)i)).ToList(); + return new FakeHistorianGatewayClient + { + RawSource = (start, end, max) => + { + if (start > end) throw new RpcException(new Status(StatusCode.InvalidArgument, "start_time must be less than or equal to end_time.")); + if (max <= 0) throw new ArgumentOutOfRangeException(nameof(max)); + var rows = new List(); + if (stored.All(s => s.Time != start)) + { + var prior = stored.LastOrDefault(s => s.Time < start); + var boundary = prior.Time == default + ? new HistorianSample { Tag = "T", NumericValue = 0, OpcQuality = 0, Quality = 1 } + : Wire(prior.Time, prior.Value); + boundary.Timestamp = Timestamp.FromDateTime(start); + if (prior.Time != default) boundary.Quality = 133; + rows.Add(boundary); + } + + rows.AddRange(stored.Where(s => s.Time >= start && s.Time <= end).Select(s => Wire(s.Time, s.Value))); + return rows.Take(max); + }, + }; + } + + private static void ShouldOnlyHaveSentValidForwardReads(FakeHistorianGatewayClient fake) + { + fake.RawCalls.ShouldNotBeEmpty(); + foreach (var (start, end, max) in fake.RawCalls) + { + start.ShouldBeLessThanOrEqualTo(end); + max.ShouldBeGreaterThan(0); + start.ShouldBeGreaterThanOrEqualTo(DateTime.UnixEpoch); + } + + fake.RawStreamsCompleted.ShouldBe(fake.ReadRawCallCount); // no gateway stream abandoned (historiangw#26) + } + + [Fact] + public async Task A_capped_reverse_read_walked_back_over_the_gateway_returns_the_latest_values_newest_first() + { + var fake = AvevaStore(); + var ds = Source(fake); + + var page = await ReverseRawReader.ReadAsync(ds, "T", hi: T0.AddSeconds(200), lo: T0.AddSeconds(50), maxValues: 10, + maxTieCluster: 65536, nowUtc: T0.AddHours(1), widthHintTicks: 0, Ct); + + page.Samples.Select(s => (double)s.Value!).ShouldBe(Enumerable.Range(191, 10).Reverse().Select(i => (double)i)); + page.Samples.ShouldAllBe(s => s.StatusCode == 0u); + ShouldOnlyHaveSentValidForwardReads(fake); + ds.GetHealthSnapshot().TotalFailures.ShouldBe(0); + } + + [Fact] + public async Task An_end_only_read_walked_back_over_the_gateway_reaches_the_first_sample_and_no_fabricated_row() + { + var fake = AvevaStore(); + var ds = Source(fake); + + var page = await ReverseRawReader.ReadAsync(ds, "T", hi: T0.AddSeconds(2), lo: new DateTime(0, DateTimeKind.Utc), + maxValues: 10, maxTieCluster: 65536, nowUtc: T0.AddHours(1), widthHintTicks: 0, Ct); + + // 2, 1, 0 — and not the NULL row AVEVA fabricates at each query start before the first sample. + page.Samples.Select(s => (double)s.Value!).ShouldBe([2d, 1d, 0d]); + ShouldOnlyHaveSentValidForwardReads(fake); + } + + [Fact] + public async Task An_uncapped_reverse_read_over_the_gateway_is_one_forward_read() + { + var fake = AvevaStore(); + var ds = Source(fake); + + var page = await ReverseRawReader.ReadAsync(ds, "T", hi: T0.AddSeconds(200), lo: T0.AddSeconds(50), maxValues: 0, + maxTieCluster: 65536, nowUtc: T0.AddHours(1), widthHintTicks: 0, Ct); + + page.Samples.Select(s => (double)s.Value!).ShouldBe(Enumerable.Range(51, 150).Reverse().Select(i => (double)i)); + fake.ReadRawCallCount.ShouldBe(1); + ShouldOnlyHaveSentValidForwardReads(fake); + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadEventsPagingTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadEventsPagingTests.cs index b0e69ec3..17c446ac 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadEventsPagingTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadEventsPagingTests.cs @@ -69,7 +69,7 @@ public sealed class NodeManagerHistoryReadEventsPagingTests : IDisposable var p1 = ReadEvents(nm, node, max: 3, inboundCp: null); p1.Cp.ShouldNotBeNull(); var p2 = ReadEvents(nm, node, max: 3, inboundCp: p1.Cp); - p2.Error.StatusCode.Code.ShouldBe(StatusCodes.BadHistoryOperationUnsupported); + p2.Error.StatusCode.Code.ShouldBe(StatusCodes.BadTooManyOperations); await host.DisposeAsync(); } @@ -283,7 +283,7 @@ public sealed class NodeManagerHistoryReadEventsPagingTests : IDisposable var p1 = ReadEvents(nm, node, max: 3, inboundCp: null, end: Early, start: Late); p1.Cp.ShouldNotBeNull(); ReadEvents(nm, node, max: 3, inboundCp: p1.Cp, end: Early, start: Late).Error.StatusCode.Code - .ShouldBe(StatusCodes.BadHistoryOperationUnsupported); + .ShouldBe(StatusCodes.BadTooManyOperations); await host.DisposeAsync(); } diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadPagingTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadPagingTests.cs index 43ba469f..05eaa14b 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadPagingTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadPagingTests.cs @@ -253,8 +253,9 @@ public sealed class NodeManagerHistoryReadPagingTests : IDisposable } /// The absurd-burst backstop is preserved: a tie cluster STRICTLY larger than the configured - /// still surfaces - /// BadHistoryOperationUnsupported for that node rather than buffering an unbounded burst. With + /// still fails that node rather than buffering an unbounded + /// burst — BadTooManyOperations, the request-limit code the reverse walk-back uses too (#686; was + /// BadHistoryOperationUnsupported). With /// the bound set to 3 and a 5-way tie cluster, the resume read that hits the cluster fails loudly. [Fact] public async Task Raw_tie_cluster_beyond_overfetch_bound_fails_loudly() @@ -283,7 +284,7 @@ public sealed class NodeManagerHistoryReadPagingTests : IDisposable // Page 2: the resume read hits the cluster; over-fetch finds 5 > bound 3 ⇒ a clear error, NOT a // silent GoodNoData that would drop the un-emitted ties. var (r2, e2, cp2) = ReadRaw(nm, nodeId, Epoch, Epoch.AddHours(1), max: 2, inboundCp: cp1); - e2.StatusCode.Code.ShouldBe(StatusCodes.BadHistoryOperationUnsupported); + e2.StatusCode.Code.ShouldBe(StatusCodes.BadTooManyOperations); r2.ShouldBeEmpty(); cp2.ShouldBeNull(); @@ -466,8 +467,327 @@ public sealed class NodeManagerHistoryReadPagingTests : IDisposable await host.DisposeAsync(); } + // --- #686: reverse / end-only raw reads on a backend that only reads forward (the Gateway) ------------------------- + + /// Boots a node manager over a and one historized variable. + private async Task<(OpcUaApplicationHost Host, OtOpcUaNodeManager Nm, NodeId NodeId)> BootForwardOnlyAsync( + ForwardOnlyHistorianDataSource source, string name) + { + var (host, server) = await BootAsync(); + var nm = server.NodeManager!; + nm.HistoryContinuationStore = new InMemoryHistoryContinuationStore(); + nm.HistorianDataSource = source; + nm.EnsureVariable($"eq-1/{name}", parentFolderNodeId: null, displayName: name, dataType: "Double", + writable: false, historianTagname: $"WW.{name}", realm: AddressSpaceRealm.Uns); + return (host, nm, nm.TryGetVariable($"eq-1/{name}")!.NodeId); + } + + /// + /// #686: an end-only capped read (Part 11: page BACKWARD from EndTime) on a backend that cannot read newest-first + /// pages every sample newest-first — and never sends that backend start > end (the gateway rejects it + /// InvalidArgument, which used to fail the node and count as a backend fault). + /// + [Fact] + public async Task Raw_end_only_read_on_a_forward_only_backend_pages_descending_without_a_reverse_window() + { + var source = new ForwardOnlyHistorianDataSource(MakeSeries(count: 250, stepSeconds: 1)); + var (host, nm, nodeId) = await BootForwardOnlyAsync(source, "FwdEndOnly"); + + var collected = await DrainAsync(nm, nodeId, DateTime.MinValue, Epoch.AddHours(1), max: 100); + + collected.ShouldBe(Enumerable.Range(0, 250).Reverse().Select(i => (double)i)); + source.ReverseWindowCount.ShouldBe(0); + await host.DisposeAsync(); + } + + /// #686: the first page of an end-only capped read is the N values AT OR BEFORE EndTime, newest first — + /// not the oldest N in history (the pre-#669 forward meaning), and not the synthetic boundary row the AVEVA-shaped + /// fake emits at each query start. + [Fact] + public async Task Raw_end_only_capped_read_on_a_forward_only_backend_returns_the_latest_values_before_end_time() + { + var source = new ForwardOnlyHistorianDataSource(MakeSeries(count: 250, stepSeconds: 1)); + var (host, nm, nodeId) = await BootForwardOnlyAsync(source, "FwdLatest"); + + var (values, error, cp) = ReadRaw(nm, nodeId, DateTime.MinValue, Epoch.AddSeconds(100), max: 3, inboundCp: null); + + error.StatusCode.Code.ShouldBe(StatusCodes.Good); + values.ShouldBe([100d, 99d, 98d]); + cp.ShouldNotBeNull(); + source.ReverseWindowCount.ShouldBe(0); + await host.DisposeAsync(); + } + + /// #686: an explicit reverse read with no cap returns the whole window newest-first in one page. + [Fact] + public async Task Raw_unbounded_explicit_reverse_read_on_a_forward_only_backend_returns_the_window_descending() + { + var source = new ForwardOnlyHistorianDataSource(MakeSeries(count: 250, stepSeconds: 1)); + var (host, nm, nodeId) = await BootForwardOnlyAsync(source, "FwdRevAll"); + + // (end, start] = (Epoch+9s, Epoch+200s] — both edges exercised. + var (values, error, cp) = ReadRaw(nm, nodeId, Epoch.AddSeconds(200), Epoch.AddSeconds(9), max: 0, inboundCp: null); + + error.StatusCode.Code.ShouldBe(StatusCodes.Good); + values.ShouldBe(Enumerable.Range(10, 191).Reverse().Select(i => (double)i)); + cp.ShouldBeNull(); + source.ReverseWindowCount.ShouldBe(0); + source.ReadCount.ShouldBe(1, "an uncapped reverse read is ONE forward read, reversed in memory — not a walk"); + await host.DisposeAsync(); + } + + /// #686: a reverse read paging through a tie cluster on a forward-only backend — every value once, newest first. + [Fact] + public async Task Raw_reverse_read_on_a_forward_only_backend_pages_through_a_tie_cluster() + { + // No AVEVA boundary row here: paging WITHIN the cluster re-reads it with the node manager's start == end point + // over-fetch, and a row fabricated at a point read's start is indistinguishable from a tie at that instant — + // dropping it is the Gateway source's job (#687), not the walk's. + var source = new ForwardOnlyHistorianDataSource( + MakeSeriesWithTieCluster(distinctBefore: 3, tieCount: 5, distinctAfter: 2), emitBoundaryRow: false); + var (host, nm, nodeId) = await BootForwardOnlyAsync(source, "FwdRevBurst"); + + var collected = await DrainAsync(nm, nodeId, start: Epoch.AddHours(1), end: Epoch.AddSeconds(-1), max: 2); + + collected.ShouldBe(Enumerable.Range(0, 10).Reverse().Select(i => (double)i)); + source.ReverseWindowCount.ShouldBe(0); + await host.DisposeAsync(); + } + + /// #686: a segment denser than the per-read row budget (MaxTieClusterOverfetch) is narrowed and re-read, + /// never answered from a truncated forward read. + [Fact] + public async Task Raw_reverse_read_on_a_forward_only_backend_narrows_a_segment_over_the_row_budget() + { + var source = new ForwardOnlyHistorianDataSource(MakeSeries(count: 250, stepSeconds: 1)); + var (host, nm, nodeId) = await BootForwardOnlyAsync(source, "FwdDense"); + nm.MaxTieClusterOverfetch = 16; // a 1-minute segment holds 60 samples + + var collected = await DrainAsync(nm, nodeId, DateTime.MinValue, Epoch.AddHours(1), max: 100); + + collected.ShouldBe(Enumerable.Range(0, 250).Reverse().Select(i => (double)i)); + source.MaxRowsReturned.ShouldBeLessThanOrEqualTo(2 * 16 + 4); + source.ReverseWindowCount.ShouldBe(0); + await host.DisposeAsync(); + } + + /// #686: sparse history far before EndTime is found, in a bounded number of forward reads. + [Fact] + public async Task Raw_end_only_read_on_a_forward_only_backend_finds_sparse_history_in_bounded_reads() + { + static DataValueSnapshot At(int year, double v) + { + var t = new DateTime(year, 1, 1, 0, 0, 0, DateTimeKind.Utc); + return new DataValueSnapshot(v, StatusCodes.Good, t, t); + } + + var source = new ForwardOnlyHistorianDataSource([At(1990, 1), At(2000, 2)]); + var (host, nm, nodeId) = await BootForwardOnlyAsync(source, "FwdSparse"); + + var (values, error, cp) = ReadRaw(nm, nodeId, DateTime.MinValue, Epoch, max: 5, inboundCp: null); + + error.StatusCode.Code.ShouldBe(StatusCodes.Good); + values.ShouldBe([2d, 1d]); + cp.ShouldBeNull(); + source.ReadCount.ShouldBeLessThanOrEqualTo(20); + source.MinRequestedStart.ShouldBeGreaterThanOrEqualTo(DateTime.UnixEpoch, "the walk never asks for data before 1970"); + source.ReverseWindowCount.ShouldBe(0); + await host.DisposeAsync(); + } + + /// #686: a single-timestamp tie cluster larger than the row budget cannot be read newest-first from + /// forward reads — the node is refused BadTooManyOperations (a request limit, not a backend fault). + [Fact] + public async Task Raw_reverse_read_on_a_forward_only_backend_refuses_a_tie_cluster_over_the_row_budget() + { + var source = new ForwardOnlyHistorianDataSource(MakeSeriesWithTieCluster(distinctBefore: 1, tieCount: 10, distinctAfter: 0)); + var (host, nm, nodeId) = await BootForwardOnlyAsync(source, "FwdHugeTie"); + nm.MaxTieClusterOverfetch = 4; + + var (_, error, cp) = ReadRaw(nm, nodeId, Epoch.AddHours(1), Epoch.AddSeconds(-1), max: 2, inboundCp: null); + + error.StatusCode.Code.ShouldBe(StatusCodes.BadTooManyOperations); + cp.ShouldBeNull(); + source.ReverseWindowCount.ShouldBe(0); + await host.DisposeAsync(); + } + + /// #686 review: a dense tag is not re-discovered every page — the width the walk ended on rides the + /// continuation point, so after the first page each page is about one read, with no full (discarded) reads. + [Fact] + public async Task Raw_reverse_paging_on_a_forward_only_backend_carries_the_observed_density_between_pages() + { + // 2000 samples 10 ms apart: a 1-minute first segment holds all of them, far over a budget of 64. + var series = Enumerable.Range(0, 2000) + .Select(i => { var t = Epoch.AddMilliseconds(10L * i); return new DataValueSnapshot((double)i, StatusCodes.Good, t, t); }) + .ToArray(); + var source = new ForwardOnlyHistorianDataSource(series); + var (host, nm, nodeId) = await BootForwardOnlyAsync(source, "FwdDensity"); + nm.MaxTieClusterOverfetch = 64; + + // Page 1 finds the data (walking down an empty hour) and its density; measure the pages after it. + var (first, _, cp) = ReadRaw(nm, nodeId, DateTime.MinValue, Epoch.AddSeconds(20), max: 20, inboundCp: null); + first.ShouldBe(Enumerable.Range(1980, 20).Reverse().Select(i => (double)i)); + source.ResetCounts(); + + var collected = new List(first); + var pages = 0; + while (cp is not null) + { + var (values, error, next) = ReadRaw(nm, nodeId, DateTime.MinValue, Epoch.AddSeconds(20), max: 20, inboundCp: cp); + error.StatusCode.Code.ShouldBe(StatusCodes.Good); + collected.AddRange(values); + cp = next; + (++pages).ShouldBeLessThan(200); + } + + collected.ShouldBe(Enumerable.Range(0, 2000).Reverse().Select(i => (double)i)); + // The discriminating count: reads that came back full and were thrown away. Without the carried width every page + // starts on a 1-minute segment holding all 2000 samples and discards full reads first (~2 per page). + source.FullReads.ShouldBeLessThanOrEqualTo(2); + // About one read per page, plus the final pages' walks from the first sample down to the 1970 floor (~16 each). + source.ReadCount.ShouldBeLessThanOrEqualTo(pages + 40); + await host.DisposeAsync(); + } + + /// #686 review: a reverse read from StartTime = DateTime.MaxValue walks back from now + 1 day, not from + /// year 9999. + [Fact] + public async Task Raw_reverse_read_from_max_value_on_a_forward_only_backend_starts_at_now_plus_one_day() + { + var source = new ForwardOnlyHistorianDataSource(MakeSeries(count: 250, stepSeconds: 1)); + var (host, nm, nodeId) = await BootForwardOnlyAsync(source, "FwdMax"); + var now = Epoch.AddHours(1); + nm.Clock = new FixedClock(now); + + var (values, error, _) = ReadRaw(nm, nodeId, DateTime.MaxValue, Epoch.AddSeconds(-1), max: 3, inboundCp: null); + + error.StatusCode.Code.ShouldBe(StatusCodes.Good); + values.ShouldBe([249d, 248d, 247d]); + source.MaxRequestedEnd.ShouldBeLessThanOrEqualTo(now.AddDays(1).AddMilliseconds(1)); + source.ReadCount.ShouldBeLessThanOrEqualTo(10); + await host.DisposeAsync(); + } + + /// #686 review: AVEVA may stamp the row it fabricates at a query start rounded UP to the millisecond. The walk + /// requests millisecond-aligned starts, so that row can never land inside (EndTime, …] when EndTime is not on a + /// millisecond. + [Fact] + public async Task Raw_reverse_read_on_a_forward_only_backend_never_returns_a_boundary_row_rounded_up_into_the_window() + { + var source = new ForwardOnlyHistorianDataSource(MakeSeries(count: 250, stepSeconds: 1), boundaryAtCeilingMs: true); + var (host, nm, nodeId) = await BootForwardOnlyAsync(source, "FwdCeil"); + + var (values, error, _) = ReadRaw(nm, nodeId, Epoch.AddSeconds(10), Epoch.AddTicks(5_000), max: 100, inboundCp: null); + + error.StatusCode.Code.ShouldBe(StatusCodes.Good); + values.ShouldBe(Enumerable.Range(1, 10).Reverse().Select(i => (double)i)); // (Epoch+0.5ms, Epoch+10s]: no 0 + await host.DisposeAsync(); + } + + /// #686 review: the upper edge is filtered too — the request end is rounded UP to the millisecond, so a + /// sample one tick after StartTime comes back from the backend and must not be returned. + [Fact] + public async Task Raw_reverse_read_on_a_forward_only_backend_excludes_a_sample_one_tick_after_the_start() + { + DataValueSnapshot At(long ticks, double v) + { + var t = Epoch.AddTicks(ticks); + return new DataValueSnapshot(v, StatusCodes.Good, t, t); + } + + var t0 = TimeSpan.TicksPerSecond; + var source = new ForwardOnlyHistorianDataSource([At(0, 0), At(t0, 1), At(t0 + 1, 99)]); + var (host, nm, nodeId) = await BootForwardOnlyAsync(source, "FwdUpper"); + + var (values, error, _) = ReadRaw(nm, nodeId, Epoch.AddTicks(t0), Epoch.AddSeconds(-1), max: 10, inboundCp: null); + + error.StatusCode.Code.ShouldBe(StatusCodes.Good); + values.ShouldBe([1d, 0d]); + await host.DisposeAsync(); + } + + /// #686 review: a tie cluster of EXACTLY MaxTieClusterOverfetch is served — the row the backend + /// fabricates at the query start does not consume the budget. + [Fact] + public async Task Raw_reverse_read_on_a_forward_only_backend_serves_a_tie_cluster_of_exactly_the_budget() + { + var source = new ForwardOnlyHistorianDataSource(MakeSeriesWithTieCluster(distinctBefore: 2, tieCount: 4, distinctAfter: 2)); + var (host, nm, nodeId) = await BootForwardOnlyAsync(source, "FwdExact"); + nm.MaxTieClusterOverfetch = 4; + + var (values, error, _) = ReadRaw(nm, nodeId, Epoch.AddHours(1), Epoch.AddSeconds(-1), max: 100, inboundCp: null); + + error.StatusCode.Code.ShouldBe(StatusCodes.Good); + values.ShouldBe(Enumerable.Range(0, 8).Reverse().Select(i => (double)i)); + await host.DisposeAsync(); + } + + /// #686 review: adjacent tie clusters one millisecond apart, none over the budget, are served — a neighbour + /// inside the request's alignment slack does not get the segment refused. + [Fact] + public async Task Raw_reverse_read_on_a_forward_only_backend_serves_adjacent_tie_clusters_within_the_budget() + { + var series = new List(); + var v = 0; + for (var c = 0; c < 3; c++) + { + var t = Epoch.AddSeconds(5).AddMilliseconds(c); + for (var i = 0; i < 4; i++) series.Add(new DataValueSnapshot((double)v++, StatusCodes.Good, t, t)); + } + + var source = new ForwardOnlyHistorianDataSource(series); + var (host, nm, nodeId) = await BootForwardOnlyAsync(source, "FwdAdjacent"); + nm.MaxTieClusterOverfetch = 4; + + var (values, error, _) = ReadRaw(nm, nodeId, Epoch.AddHours(1), Epoch.AddSeconds(-1), max: 100, inboundCp: null); + + error.StatusCode.Code.ShouldBe(StatusCodes.Good); + values.ShouldBe(Enumerable.Range(0, 12).Reverse().Select(i => (double)i)); + await host.DisposeAsync(); + } + + /// #686 review: a MaxTieClusterOverfetch of 1 (or a misconfigured 0) does not make every reverse read on + /// a forward-only backend refuse — distinct-timestamp data is still served. + [Theory] + [InlineData(1)] + [InlineData(0)] + public async Task Raw_reverse_read_on_a_forward_only_backend_works_with_a_minimal_budget(int budget) + { + var source = new ForwardOnlyHistorianDataSource(MakeSeries(count: 250, stepSeconds: 1)); + var (host, nm, nodeId) = await BootForwardOnlyAsync(source, $"FwdMin{budget}"); + nm.MaxTieClusterOverfetch = budget; + + var (values, error, _) = ReadRaw(nm, nodeId, DateTime.MinValue, Epoch.AddSeconds(100), max: 3, inboundCp: null); + + error.StatusCode.Code.ShouldBe(StatusCodes.Good); + values.ShouldBe([100d, 99d, 98d]); + await host.DisposeAsync(); + } + + /// A backend that DECLARES native reverse support is still sent the reverse window — one read. + [Fact] + public async Task Raw_end_only_read_on_a_reverse_capable_backend_is_one_native_reverse_read() + { + var (host, server) = await BootAsync(); + var nm = server.NodeManager!; + nm.HistoryContinuationStore = new InMemoryHistoryContinuationStore(); + var source = new SeriesHistorianDataSource(MakeSeries(count: 250, stepSeconds: 1)); + nm.HistorianDataSource = source; + nm.EnsureVariable("eq-1/native", parentFolderNodeId: null, displayName: "Native", dataType: "Double", + writable: false, historianTagname: "WW.Native", realm: AddressSpaceRealm.Uns); + var nodeId = nm.TryGetVariable("eq-1/native")!.NodeId; + + var (values, _, _) = ReadRaw(nm, nodeId, DateTime.MinValue, Epoch.AddSeconds(100), max: 3, inboundCp: null); + + values.ShouldBe([100d, 99d, 98d]); + source.ReadCount.ShouldBe(1); + await host.DisposeAsync(); + } + /// Pages a Raw read to completion and returns the union of the pages in arrival order. - private static Task> DrainAsync(OtOpcUaNodeManager nm, NodeId nodeId, DateTime start, DateTime end, uint max) + private static Task> DrainAsync(OtOpcUaNodeManager nm, NodeId nodeId, DateTime start, DateTime end, uint max, + int maxPages = 40) { var collected = new List(); byte[]? cp = null; @@ -479,7 +799,7 @@ public sealed class NodeManagerHistoryReadPagingTests : IDisposable collected.AddRange(values); cp = nextCp; pageCount++; - pageCount.ShouldBeLessThan(40, "paging must terminate, not loop"); + pageCount.ShouldBeLessThan(maxPages, "paging must terminate, not loop"); } while (cp is not null); return Task.FromResult(collected); @@ -584,6 +904,9 @@ public sealed class NodeManagerHistoryReadPagingTests : IDisposable private sealed class SeriesHistorianDataSource(IReadOnlyList series) : IHistorianDataSource { private int _readCount; + + /// Serves start > end natively — the SQL backends' contract (#686: declared, like they do). + public bool SupportsReverseRaw => true; public int ReadCount => _readCount; public void ResetReadCount() => _readCount = 0; @@ -629,6 +952,97 @@ public sealed class NodeManagerHistoryReadPagingTests : IDisposable public void Dispose() { } } + /// + /// #686: a Gateway-shaped backend that reads FORWARD only. is false; a reverse window + /// (start > end) is counted and rejected as the gateway does (InvalidArgument). A forward read is + /// inclusive [start, end] and, like an AVEVA Full read, first emits a synthetic boundary row stamped at + /// the query start carrying the value in effect there (Bad 0 when there is none) — counted against the cap. + /// + private sealed class ForwardOnlyHistorianDataSource( + IReadOnlyList series, bool emitBoundaryRow = true, bool boundaryAtCeilingMs = false) + : IHistorianDataSource + { + private readonly Lock _gate = new(); + private int _readCount; + private int _reverseWindowCount; + private int _maxRowsReturned; + + public int ReadCount => _readCount; + public int ReverseWindowCount => _reverseWindowCount; + public int MaxRowsReturned => _maxRowsReturned; + + /// The earliest query start and latest query end this backend was sent. + public DateTime MinRequestedStart { get; private set; } = DateTime.MaxValue; + public DateTime MaxRequestedEnd { get; private set; } = DateTime.MinValue; + + /// Reads whose row count equalled the cap (a full, possibly truncated, read). + public int FullReads { get; private set; } + + public void ResetCounts() + { + lock (_gate) FullReads = 0; + Interlocked.Exchange(ref _readCount, 0); + } + + public bool SupportsReverseRaw => false; + + public Task ReadRawAsync( + string fullReference, DateTime startUtc, DateTime endUtc, uint maxValuesPerNode, + CancellationToken cancellationToken) + { + Interlocked.Increment(ref _readCount); + if (startUtc > endUtc) + { + Interlocked.Increment(ref _reverseWindowCount); + throw new InvalidOperationException("start_time must be less than or equal to end_time. (InvalidArgument)"); + } + + lock (_gate) + { + if (startUtc < MinRequestedStart) MinRequestedStart = startUtc; + if (endUtc > MaxRequestedEnd) MaxRequestedEnd = endUtc; + } + + static DateTime Ts(DataValueSnapshot s) => s.SourceTimestampUtc ?? DateTime.MinValue; + var prior = series.LastOrDefault(s => Ts(s) <= startUtc); + // AVEVA may stamp its fabricated row at the query start rounded UP to its millisecond resolution. + var stamp = boundaryAtCeilingMs && startUtc.Ticks % TimeSpan.TicksPerMillisecond != 0 + ? new DateTime(startUtc.Ticks - startUtc.Ticks % TimeSpan.TicksPerMillisecond + TimeSpan.TicksPerMillisecond, DateTimeKind.Utc) + : startUtc; + var rows = new List(); + if (emitBoundaryRow) + rows.Add(prior is null + ? new DataValueSnapshot(0d, StatusCodes.BadNoData, stamp, stamp) + : new DataValueSnapshot(prior.Value, prior.StatusCode, stamp, stamp)); + rows.AddRange(series.Where(s => Ts(s) >= startUtc && Ts(s) <= endUtc)); + var page = maxValuesPerNode == 0 ? rows : rows.Take((int)maxValuesPerNode).ToList(); + if (maxValuesPerNode > 0 && page.Count == maxValuesPerNode) + lock (_gate) FullReads++; + int seen; + while (page.Count > (seen = Volatile.Read(ref _maxRowsReturned)) + && Interlocked.CompareExchange(ref _maxRowsReturned, page.Count, seen) != seen) { } + return Task.FromResult(new HistorianRead(page, null)); + } + + public Task ReadProcessedAsync( + string fullReference, DateTime startUtc, DateTime endUtc, TimeSpan interval, + HistoryAggregateType aggregate, CancellationToken cancellationToken) => + Task.FromResult(new HistorianRead(Array.Empty(), null)); + + public Task ReadAtTimeAsync( + string fullReference, IReadOnlyList timestampsUtc, CancellationToken cancellationToken) => + Task.FromResult(new HistorianRead(Array.Empty(), null)); + + public Task ReadEventsAsync( + string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents, + CancellationToken cancellationToken) => + Task.FromResult(new HistoricalEventsResult(Array.Empty(), null)); + + public HistorianHealthSnapshot GetHealthSnapshot() => NullHistorianDataSource.Instance.GetHealthSnapshot(); + + public void Dispose() { } + } + /// /// A microsecond-resolution backend, shaped like QuestDB: every inbound window bound is TRUNCATED to 10 ticks /// (as QuestDbConnection.Round does) before the series is filtered, and the source reports @@ -640,6 +1054,8 @@ public sealed class NodeManagerHistoryReadPagingTests : IDisposable public long TimestampResolutionTicks => 10; + public bool SupportsReverseRaw => true; // QuestDB serves reverse natively + private static DateTime Truncate(DateTime t) => new(t.Ticks - t.Ticks % 10, DateTimeKind.Utc); public Task ReadRawAsync( @@ -666,6 +1082,11 @@ public sealed class NodeManagerHistoryReadPagingTests : IDisposable public void Dispose() { } } + private sealed class FixedClock(DateTime now) : TimeProvider + { + public override DateTimeOffset GetUtcNow() => new(now); + } + private async Task<(OpcUaApplicationHost Host, OtOpcUaSdkServer Server)> BootAsync() { var applicationUri = $"urn:OtOpcUa.HistoryPagingTest:{Guid.NewGuid():N}";