diff --git a/docs/Historian.md b/docs/Historian.md index 78a29031..55813c60 100644 --- a/docs/Historian.md +++ b/docs/Historian.md @@ -112,6 +112,7 @@ Processed read is `BadAggregateNotSupported` and no history aggregate functions "CallTimeout": "00:00:30", "MaxTieClusterOverfetch": 65536, "HonorReturnBounds": true, + "GatewayAtTimeFromRaw": true, "HistoryReadBatchParallelism": 4, "MaxConcurrentHistoryReadBatches": 16, "HistoryReadDeadline": "00:01:00", @@ -146,6 +147,7 @@ Processed read is `BadAggregateNotSupported` and no history aggregate functions | `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 `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. | | `HonorReturnBounds` | bool | `true` | (#691, all backends.) Whether HistoryRead-Raw honours `ReturnBounds` (Part 11 bounding values, [Bounding values](#bounding-values-returnbounds-691)). `false` is the kill switch back to the pre-#691 behaviour: the flag is ignored (stored samples only) and a modified read carrying it is not refused `Bad_InvalidArgument`. Note the .NET SDK defaults both `ReturnBounds` and `IsReadModified` to `true`. | +| `GatewayAtTimeFromRaw` | bool | `true` | (#693, **Gateway backend only.**) Whether HistoryRead-AtTime is served from raw gateway reads evaluated with the SQL backends' AtTime rules ([Gateway AtTime](#gateway-attime-is-evaluated-from-raw-reads-693)). `false` is the kill switch back to the pre-#693 answer: AVEVA's own interpolation (the gateway's `ReadAtTime`) passed through unchanged. | | `HistoryReadBatchParallelism` | int | `4` | Max historized nodes served **concurrently** within one HistoryRead batch (arch-review 03/S3). The SDK's `HistoryRead*` override surface is synchronous, so each arm bridges once at its boundary; this turns per-node serving from sequential (`N × CallTimeout` worst case) into `⌈N / P⌉ × CallTimeout`. Also caps the gateway-load multiplication a single client can command. `≤ 1` serves sequentially. | | `MaxConcurrentHistoryReadBatches` | int | `16` | Process-wide limit on HistoryRead (and HistoryUpdate) batches served **concurrently** (arch-review 03/S3). A flood of history clients degrades to `BadTooManyOperations` rather than exhausting the SDK request-thread pool. With `HistoryReadBatchParallelism` this caps total in-flight gateway reads at `MaxConcurrentHistoryReadBatches × HistoryReadBatchParallelism` (default 16 × 4 = 64). A batch that cannot get a slot within `min(HistoryReadDeadline, 5 s)` fails every handle `BadTooManyOperations`. | | `MaxTimeWeightedRawRows` | int | `1000000` | `SqlServer` / `QuestDb` (#671), and the `Gateway` per-sample aggregates (#681 — every per-sample read there streams raw samples, so the cap applies to the window's raw rows; see [Aggregate set](#aggregate-set-and-badaggregatenotsupported-285)). The most raw in-window samples one [time-weighted aggregate](#time-weighted-aggregates-part-13-671) read (TimeAverage, TimeAverage2, **Total**, Total2, Interpolative, and — since #680 — **PercentGood/PercentBad**) may stream into the server-side evaluator — that family is **not** pushed down to SQL. The window query is capped at this + 1 rows; one row past the cap fails **that node** with `BadTooManyOperations` (not counted as a backend fault) instead of holding a pooled connection until `HistoryReadDeadline` and returning `BadTimeout`. Bound rows outside the window do not count; per-sample aggregates are not capped on the SQL backends (pushed down). 1 000 000 ≈ 11.5 days of a 1 Hz tag. Non-positive = unbounded (a `Validate()` warning). `Gateway` (#688): the row budget it asks the gateway for (`cap + 18`) is clamped at the gateway's default `MaxValuesCeiling` (1,000,000) while the cap is at or under it, so the default works against a gateway with `AllowStreamingBeyondCeiling=false`; against a gateway with a *lower* ceiling the refused budget is retried once at the ceiling the gateway names (#691.3), so no change is needed (`ceiling − 18` only saves that round-trip). | @@ -780,6 +782,71 @@ data. it could only ever be a Part 11 *interpolated* bound. The start bound on the Gateway is the real prior sample, found by the #686 walk-back. +### Gateway AtTime is evaluated from raw reads (#693) + +> ⚠️ **Client-visible (#693).** Gateway AtTime used to pass AVEVA's own interpolated answer (the gateway's `ReadAtTime`) through, +> which broke the [AtTime rules](#attime-reads) every other backend follows (Part 11 §6.5.4). Live (`HistGW.LiveTest.Gate685`: +> 100 @ 19:00:00, 107 @ 19:00:02, **NULL** @ 19:00:03, 108 @ 19:00:05) it returned **107 Good at the NULL's own time**, **null + Bad +> at 19:00:05** (a stored Good 108 right after the NULL), **99.99999999944 for the stored 100**, and no Interpolated bit on any +> interpolated value. It is now answered from **raw** reads with the SQL backends' shared `AtTimeEvaluator` (as #681 did for the +> per-sample aggregates), so the same stored samples give the same answer on every backend: the exact stored value (no +> Interpolated bit) at a stored time; null + the outage's Bad code at or after a stored NULL; the previous value held, +> `UncertainDataSubNormal` + Interpolated, before a Bad neighbour; linear + Interpolated between usable neighbours; `BadNoData` +> before the first sample; stepped after the last (`UncertainDataSubNormal` once later than now). Kill switch: +> `ServerHistorian:GatewayAtTimeFromRaw = false`. + +- **Read plan (`GatewayAtTimeReader`).** Each requested time needs its stored sample at-or-before (`prev`) and after (`next`; read only + when the evaluator uses it — a usable `prev` that is not an exact hit). The distinct times are served by **one** forward window read + `[first − 1 ms, last + 1 ms]` of at most 1,024 rows, plus a **walk-back** for the sample before the first time and one small forward + **probe** (18 rows) for the sample after the last. Measured on the fake: ten times one second apart on a 1 Hz tag cost **1 read** + (all exact hits) or **3** (interpolated); ten times an hour apart on a 1 Hz tag **12** (≈ 9.5k rows, never the 32k between); one + time 400 days after the only samples **5**; a tag with no history **4**. A window read that fills its budget resolves the times before + its last row; the next time either continues from there (if the rows in between should fit one read) or **jumps** to it with a + small read (64 rows) that starts a density-sized margin earlier (so its `prev` is normally in the same read) — 1,000 times + spread over ten years of a 1 Hz tag cost ≈ 1,000 reads of ≤ 64 rows (≈ 66k rows; it used to be 1,024-row reads). +- **Walk-back** (the gateway reads forward only), in two phases. **Gallop:** segments stepping back from the time, the first sized + from the window read's density (else 1 minute), growing ×4 when empty; after two empty segments the whole remainder down to the + epoch is asked once — a complete answer ends the walk, a **full** one seeds the bisection with its last row. **Bisect** (once a + stored row is known below — a full segment's, or an earlier window read's last row): read `(mid, hi]`; empty ⇒ `hi = mid`; full ⇒ + the known row moves up to its last row; complete with rows ⇒ done. Right after a full read one segment is aimed at about a quarter + of a read at the density it showed (until an aimed segment comes back empty — a gap — then plain halving); below 2 ms the segment + is the whole stretch from the known row (never a 100 ns step). Measured (review harness, 5 years of history, time = now − 1 min): + a tag stopped 1 h / 1 d / 7 d / 30 d / 365 d ago costs 6–18 reads at 1/min, 21–25 at 1 Hz and 25–31 at 10 Hz; a 1 kHz tag behind + a 10-minute gap 6 reads (< 10k rows). (Before the review fix a full segment reset the width and a full whole-remainder read was + discarded, so a dense tag that stopped weeks ago gave up after 65 reads.) More than 1,024 stored samples within one millisecond + fail the node `BadTooManyOperations` (not counted as a backend failure). +- **Bounded.** One request takes at most **64 + 16 × distinct times** raw reads; past it the node fails `BadTooManyOperations` + ("needed more than N raw reads", naming a small strict-gateway ceiling when one bounded the reads). It is a runaway guard, not a + budget (`HistoryReadDeadline` bounds the wall clock): a time inside a gap of a bursty tag legitimately costs a jump read plus a + bisection back across the gap, ≈ 6 reads — a 1 h/day 1 Hz tag asked at noon on 30 / 100 / 300 consecutive days takes 186 / 606 / + 1,806 reads, which the first cut's 64 + 4 per time refused although AVEVA's own AtTime served it. A strict gateway whose ceiling is + below 4 rows is refused up front with a message naming the ceiling; a small one (e.g. 10) is served by paging within the cap. +- **Wire shape.** Every request bound is a whole millisecond and a row counts only **strictly inside** its request, so AVEVA's synthetic + boundary row at the query start (#687) — and any row at the query end — is never taken for a stored sample. A Bad row is value-less + (#690). Nothing is asked past now + 1 day (the gateway's AVEVA may refuse a far-future end; a time beyond it is stepped from the last + sample). Every stream is drained. A strict gateway's ceiling (#691.3) is learned on its first refusal and honoured; a full read is + ordinary paging here, so it is not re-probed. +- **Values follow the node's type (type hint).** The gateway wire carries no data type — every historizable type is stored as an + AVEVA analog (a Boolean as a 0/1 `Int2`) — so the node manager passes each node's scalar DataType with the AtTime call + (`IHistorianDataSource.ReadAtTimeAsync(…, TypeCode? valueType, …)`, a default-interface overload: SqlServer / QuestDB / Null + ignore it; Raw and UNS NodeIds give the same hint, arrays and other types none). The Gateway converts each raw row to that type + **before** the shared evaluator runs, so its own type rules apply exactly as on the SQL backends: a **Boolean** node + (value ≠ 0) is **stepped** — never 0.5; an **integer** node is interpolated, rounded half away from zero and returned as the + node's own integer type (`Int32` ⇒ `int`); Float / Double interpolate as Double (as the SQL backends return them). With no + hint (an unknown node) every numeric tag interpolates as Double. A **DateTime** node is stepped, but its value stays the + gateway's Double: the Gateway never historizes a DateTime node (`HistorianTypeMapper` skips it at provisioning and the recorder drops + non-numeric values), so such a node only has history if something else wrote that AVEVA tag, and the numeric encoding is not known + here — the SQL backends store DateTime as its own kind and return a `DateTime`. Under an integer hint a NaN, ±∞ or + out-of-Int64-range stored value is no integer: it is treated as unusable (as `BadNoData`), so a neighbour like that is never + interpolated or saturated into a Good value. +- **What differs between backends (by design, not changed by #693).** Only the **Gateway** converts an AtTime result to the node's + integer type (`Int32` ⇒ `int`): the SQL backends store and return integer nodes as `Int64` (and Boolean as Boolean). Gateway + **Raw** reads still return a Boolean node's samples as the stored Double 0/1 (the hint is passed only on AtTime). +- ⚠️ **Shared-evaluator fix (all backends that use it):** `AtTimeEvaluator` returned every interpolated `Int64` boxed as a + **Double** (a C# `cond ? (long)x : double` conditional widens the long); its "keeping the Int64 type" tests passed because + Shouldly's `ShouldBe` compares numbers across types. SqlServer / QuestDB interpolated `Int64` AtTime values are now `Int64`, + as documented in [AtTime reads](#attime-reads). + ### Bounding values (`ReturnBounds`, #691) ⚠️ **Client-visible — a breaking change for .NET clients that rely on SDK defaults.** Until #691 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 309a0b63..7b0cc72c 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 @@ -81,6 +81,31 @@ public interface IHistorianDataSource : IDisposable IReadOnlyList timestampsUtc, CancellationToken cancellationToken); + /// + /// Gitea #693: with the value type of + /// the node being read, for a backend whose stored values do not carry their type (the Gateway: AVEVA stores every + /// historized type as a numeric analog, so a Boolean comes back as a 0/1 Double). Such a backend uses the hint to apply + /// the AtTime rules exactly as the typed backends do — a Boolean (or String / DateTime) node is stepped, an integer node + /// is interpolated, rounded half away from zero and returned as the node's integer type, a floating-point node is + /// interpolated. A backend that stores typed values ignores it: the default forwards to the hint-less overload, so + /// SqlServer, QuestDB and the Null source need no change. + /// + /// The full reference of the tag to read. + /// The list of timestamps to read values at. + /// + /// The scalar value type of the node (, the integer codes, , + /// , , ), or null when + /// unknown, not scalar, or not one of those — the backend's own policy then applies. + /// + /// A cancellation token that can be used to cancel the operation. + /// One sample per requested timestamp, in the same order as . + Task ReadAtTimeAsync( + string fullReference, + IReadOnlyList timestampsUtc, + TypeCode? valueType, + CancellationToken cancellationToken) => + ReadAtTimeAsync(fullReference, timestampsUtc, cancellationToken); + /// /// Read historical alarm / event records — OPC UA HistoryReadEvents service. /// Distinct from any live event stream; sources here come from the historian's @@ -95,7 +120,7 @@ public interface IHistorianDataSource : IDisposable /// uint maxValuesPerNode because their OPC UA HistoryRead surface has no /// equivalent "use default" sentinel. /// - /// This surface declares and + /// This surface declares and /// as required members — a server-side historian owns the full read surface. /// /// The source name to filter events, or null to return events from all sources. diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayAtTimeReader.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayAtTimeReader.cs new file mode 100644 index 00000000..f6d9f107 --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayAtTimeReader.cs @@ -0,0 +1,494 @@ +using ZB.MOM.WW.HistorianGateway.Contracts.Grpc; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Mapping; +using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway; + +/// +/// Gitea #693: HistoryRead AtTime on the Gateway backend, served from RAW gateway reads and evaluated with the SQL backends' +/// shared (Part 11 §6.5.4) — as #681 did for the per-sample aggregates — so the Gateway +/// answers exactly what SqlServer and QuestDB answer for the same stored samples. AVEVA's own interpolated answer (the +/// gateway's ReadAtTime) is not used: live, it returned the previous value as Good at a stored NULL's own time, +/// null + Bad at a Good sample right after a NULL, a drifted value (99.99999999944) at an exact stored 100, and no +/// Interpolated bit. +/// +/// +/// +/// What each requested time needs. prev = the latest stored sample at or before it, next = the +/// first stored sample after it. next is only read when the evaluator uses it (a usable prev that is not +/// an exact hit). +/// +/// +/// Read plan — O(1) reads for a dense batch, never two round-trips per timestamp by default. The distinct +/// requested times are walked in ascending order with ONE forward window read [t − 1 ms, max + 1 ms] of at most +/// rows, which yields every in-window neighbour. The sample before the first time comes from a +/// walk-back (below), the sample after the last time from one small forward probe. A window read that fills its budget +/// (dense data, widely spread times) resolves the times before its last row; the next unresolved time either +/// continues from that row (when the rows in between are expected to fit one read) or jumps there with a read that +/// starts a density-sized margin before it (so its prev is normally in the same read). Ten times one second +/// apart on a 1 Hz tag cost 1 read (all exact hits) or 3 (interpolated); ten times an hour apart about one read each. +/// +/// +/// Walk-back (the gateway reads forward only), in two phases. Gallop: segments (lo, hi] stepping +/// back from hi, each one forward read, the first sized from the density the window read showed (else +/// ), an empty one growing the next ×4; after two empty segments the whole remainder down +/// to the epoch is asked once — a complete answer ends the walk, a FULL one seeds the bisection with its last row. +/// Bisect (once a stored row is known below hi — a full segment's last row, or the last row of an earlier +/// window read): read (mid, hi]; empty ⇒ hi = mid; full ⇒ the known row moves up to its last row; +/// complete with rows ⇒ its last row is the answer. Right after a full read the next segment is aimed at about a +/// quarter of a read at the density it showed (one try — an empty aimed segment falls back to halving), and below +/// 2 ms the segment is the whole stretch from the known row (never a 100 ns step). +/// +/// +/// Bounded. One request may take at most 64 + 16 × distinct times raw reads (, +/// ); past it — or with a strict-gateway ceiling below rows — the +/// node fails BadTooManyOperations with a message naming the limit. A jump read (to a time far past the previous +/// read) asks for only rows, so a thousand times spread over ten years cost about a thousand +/// small reads, not a thousand 1,024-row reads. +/// +/// +/// Wire shape. Every request bound is a whole millisecond (AVEVA stores milliseconds and truncates request +/// times); a row counts only strictly INSIDE its request — AVEVA's synthetic boundary row at the query start (the value +/// in effect there, Quality = 133, or a NULL row) can never be taken for a stored sample, nor a possible row at +/// the query end; the epoch and before hold nothing. A Bad row (a stored NULL arrives as value 0, q=1 opc=0, +/// #690) is value-less. Every stream is drained, never abandoned (a cancelled gateway stream evicts its pooled +/// historian session). Nothing is asked past now + (the +/// historian stores nothing further ahead, and the gateway's AVEVA may refuse a far-future end); a strict gateway's +/// row ceiling (#691.3) is learned and honoured — a short read is ordinary paging here, never re-probed. +/// +/// +/// Values. A gateway value is a Double or a String whatever the node's data type (AVEVA stores every historized +/// type as a numeric analog — a Boolean as a 0/1 Int2). The node's type hint converts each row BEFORE the +/// shared evaluator runs, so its own type rules apply exactly as on the SQL backends: Boolean (value ≠ 0) and String +/// are stepped; an integer type is an Int64 — interpolated, rounded half away from zero — and the result is returned +/// as the node's integer type; Float / Double (and no hint) interpolate linearly as Double. +/// +/// +internal static class GatewayAtTimeReader +{ + /// The row budget of a window read (the first, or one continuing from a full one) and a walk-back segment. + internal const int ReadRows = 1024; + + /// The row budget of a jump read (a window read restarting at a time far past the previous read's last row). + internal const int JumpRows = 64; + + /// The row budget of the forward probe for the sample after the last time (boundary row + ties + one). + internal const int ProbeRows = 2 + RawAggregateReader.PreWindowAllowance; + + /// The smallest row budget a read can work with (the boundary row, a tie, one sample beyond). + internal const int MinBudget = 4; + + /// Raw reads one request may always take … + internal const int BaseReads = 64; + + /// + /// … plus this many per distinct requested time. A runaway guard, not a budget (HistoryReadDeadline bounds the + /// wall clock): a time inside a gap of a bursty tag legitimately costs a jump read plus a bisection back across the gap + /// (≈ 6 reads — a 1 h/day tag asked at noon on consecutive days), so 4 refused requests AVEVA's own AtTime served. + /// + internal const int ReadsPerTime = 16; + + /// The first walk-back segment's width when the window read showed no density. + internal static readonly TimeSpan InitialWalkWidth = TimeSpan.FromMinutes(1); + + /// The widest walk-back segment sized from density, before ×4 growth. + private static readonly TimeSpan MaxSeededWidth = TimeSpan.FromDays(1); + + private const long Ms = TimeSpan.TicksPerMillisecond; + + private static readonly TimeSpan OneMs = TimeSpan.FromMilliseconds(1); + + private static readonly DateTime Floor = RawAggregateReader.ProbeFloor; + + /// One drained read: the rows strictly inside the request (ascending) and whether the stream ended short. + private readonly record struct Read(List Rows, bool Complete); + + /// Evaluates AtTime for (same length and order, duplicates included). + /// The gateway client. + /// The data source's learned gateway row ceiling (#691.3). + /// Historian tag name. + /// Requested times (any kind; treated as UTC, as on the SQL backends). + /// Server timestamp, the beyond-now rule and the look-ahead horizon. + /// Cancellation. + /// The node's scalar value type (the hinted IHistorianDataSource.ReadAtTimeAsync), or null. + /// The request's raw-read cap; null = + per distinct time. + /// One value per requested time, stamped with it. + /// More than rows within one millisecond, the + /// read cap, or a strict gateway ceiling below . + public static async Task> ReadAsync(IHistorianGatewayClient client, GatewayRowCeiling ceiling, + string tag, IReadOnlyList timestampsUtc, DateTime now, CancellationToken ct, TypeCode? valueType = null, + int? maxReads = null) + { + var requested = timestampsUtc.Select(t => DateTime.SpecifyKind(t, DateTimeKind.Utc)).ToArray(); + if (requested.Length == 0) return []; + var times = requested.Distinct().Order().ToArray(); + var ctx = new Ctx(client, ceiling, tag, ct, valueType, maxReads ?? BaseReads + ReadsPerTime * times.Length); + var neighbours = await ResolveAsync(ctx, times, now).ConfigureAwait(false); + var result = new DataValueSnapshot[requested.Length]; + for (var i = 0; i < requested.Length; i++) + { + var t = requested[i]; + var (prev, next) = neighbours[t.Ticks]; + // A DateTime node is stepped (as on the SQL backends): a usable non-numeric next sends the evaluator down its + // stepped branch, with the next's own status. + if (valueType == TypeCode.DateTime && next is { IsUsable: true } n) + next = n with { Value = HistorianValue.FromString(string.Empty) }; + result[i] = ToNodeType(AtTimeEvaluator.Evaluate(t, prev, next, now, now), valueType); + } + return result; + } + + /// One request's read context: the client, the type hint, and the read count against its cap. + private sealed class Ctx(IHistorianGatewayClient client, GatewayRowCeiling ceiling, string tag, CancellationToken ct, + TypeCode? valueType, int maxReads) + { + public IHistorianGatewayClient Client { get; } = client; + public GatewayRowCeiling Ceiling { get; } = ceiling; + public string Tag { get; } = tag; + public CancellationToken Ct { get; } = ct; + public TypeCode? ValueType { get; } = valueType; + public int MaxReads { get; } = maxReads; + public int Reads { get; set; } + + /// The smallest budget a read was served with (a strict gateway's ceiling shows here). + public int SmallestBudget { get; set; } = int.MaxValue; + } + + /// True for the integer type codes (evaluated as Int64, returned as the node's own type). + private static bool IsInteger(TypeCode? t) => + t is TypeCode.SByte or TypeCode.Byte or TypeCode.Int16 or TypeCode.UInt16 or TypeCode.Int32 or TypeCode.UInt32 + or TypeCode.Int64 or TypeCode.UInt64; + + /// + /// An evaluated Int64 (integer hint) returned as the node's integer type. A value the type cannot hold (only possible + /// for a stored value already outside the node's range) stays the evaluator's Int64. + /// + private static DataValueSnapshot ToNodeType(DataValueSnapshot v, TypeCode? valueType) + { + if (v.Value is not long l || !IsInteger(valueType) || valueType == TypeCode.Int64) return v; + try + { + return v with { Value = Convert.ChangeType(l, valueType!.Value, System.Globalization.CultureInfo.InvariantCulture) }; + } + catch (OverflowException) + { + return v; + } + } + + /// prev / next for every time of (distinct, ascending). + private static async Task> ResolveAsync( + Ctx ctx, DateTime[] times, DateTime now) + { + var result = new Dictionary(times.Length); + var horizon = MsFloor(DateTime.MaxValue - now > RawAggregateReader.LookAheadHorizon + ? DateTime.SpecifyKind(now + RawAggregateReader.LookAheadHorizon, DateTimeKind.Utc) + : DateTime.MaxValue.AddTicks(-Ms)); + + // Nothing is stored at or before the epoch. + var i = 0; + for (; i < times.Length && times[i] <= Floor; i++) result[times[i].Ticks] = (null, null); + if (i == times.Length) return result; + + // Reads are planned on times clamped to the horizon (nothing is stored beyond it, and no read ends past it). + var qe = Min(MsFloor(Min(times[^1], horizon)) + OneMs, horizon); + var qs = RequestStart(Min(times[i], horizon)); + var budget = ReadRows; + StoredSample? before = null; // the latest stored sample at or before qs … + var beforeKnown = false; // … once known + DateTime? walkKnown = null; // a stored sample exists here (at or before qs): the walk-back bisects above it + var walkWidth = InitialWalkWidth; + var probeFrom = -1; // first index whose next lies at or after qe (complete last read) + + while (true) + { + var read = await ReadTrustedAsync(ctx, qs, qe, budget).ConfigureAwait(false); + var rows = read.Rows; + if (!read.Complete && rows.Count == 0) + throw TooDense(ctx, qs); + if (rows.Count >= 2 && !beforeKnown && walkKnown is null) + walkWidth = Clamp(TimeSpan.FromTicks(4 * (rows[^1].TimeUtc - rows[0].TimeUtc).Ticks / (rows.Count - 1))); + + // A short stream saw every row up to qe; a full one every row before its last row's time x. + var x = read.Complete ? qe : rows[^1].TimeUtc; + var j = -1; // index of the last row at or before the current time + var k = i; + for (; k < times.Length; k++) + { + var t = times[k]; + var tr = Min(t, horizon); + if (!read.Complete && tr >= x) break; + while (j + 1 < rows.Count && rows[j + 1].TimeUtc <= tr) j++; + + StoredSample? prev; + if (j >= 0) prev = rows[j]; + else + { + if (!beforeKnown) + { + before = await WalkBackAsync(ctx, qs, walkKnown, walkWidth).ConfigureAwait(false); + beforeKnown = true; + } + prev = before; + } + + StoredSample? next = null; + if (NeedsNext(t, prev) && t < horizon) + { + if (j + 1 < rows.Count) next = rows[j + 1]; + else if (probeFrom < 0) probeFrom = k; // after every row of a complete read: the probe answers it + } + result[t.Ticks] = (prev, next); + } + + if (k == times.Length) break; + + // Truncated before times[k]: continue from the last row, or jump to times[k] with a density-sized margin and a + // small budget (it needs that time's neighbours, and whatever later times follow closely). + var tn = Min(times[k], horizon); + var spanTicks = Math.Max(1, (x - rows[0].TimeUtc).Ticks); + var density = (double)rows.Count / spanTicks; // rows per tick + var gapRows = density * (tn - x).Ticks; + var margin = Clamp(TimeSpan.FromTicks((long)Math.Min(TimeSpan.MaxValue.Ticks / 2.0, JumpRows / 4.0 / Math.Max(density, 1e-18)))); + var resume = MsFloor(x) - OneMs; + var jump = gapRows >= ReadRows / 2.0; + var nextQs = jump ? Max(RequestStart(tn - Min(margin, tn - Floor)), resume) : resume; + if (nextQs <= qs) throw TooDense(ctx, x); + + if (nextQs < x) + { + // Every row before x was seen: the latest at or before nextQs is known. + var lastIndex = rows.FindLastIndex(r => r.TimeUtc <= nextQs); + if (lastIndex >= 0) + { + before = rows[lastIndex]; + beforeKnown = true; + } + walkKnown = null; + } + else + { + beforeKnown = false; + before = null; + walkKnown = x; + walkWidth = margin; + } + budget = jump ? JumpRows : ReadRows; + qs = nextQs; + i = k; + } + + if (probeFrom >= 0 && qe < horizon) + { + var after = await FirstAfterAsync(ctx, qe - OneMs, horizon, Min(times[^1], horizon)).ConfigureAwait(false); + if (after is not null) + for (var k = probeFrom; k < times.Length; k++) + { + var t = times[k]; + var (prev, next) = result[t.Ticks]; + if (next is null && t < horizon && NeedsNext(t, prev) && after.Value.TimeUtc > t) + result[t.Ticks] = (prev, after); + } + } + return result; + } + + /// The evaluator reads next only for a usable prev that is not an exact hit. + private static bool NeedsNext(DateTime t, StoredSample? prev) => prev is { IsUsable: true } p && p.TimeUtc != t; + + /// One walk-back segment (lo, hi]: its oldest and newest rows, and whether the read was complete. + private static async Task<(StoredSample? First, StoredSample? Last, bool Complete)> SegmentAsync( + Ctx ctx, DateTime lo, DateTime hi) + { + var read = await ReadTrustedAsync(ctx, MsFloor(lo), MsFloor(hi) + OneMs, ReadRows).ConfigureAwait(false); + StoredSample? first = null, last = null; + foreach (var r in read.Rows) + if (r.TimeUtc > lo && r.TimeUtc <= hi) + { + first ??= r; + last = r; + } + return (first, last, read.Complete); + } + + /// + /// The latest stored sample with time in (Floor, hi] (see the type remarks: gallop, then bisect). + /// : a stored sample exists at that time (≤ ) — the walk then bisects + /// above it straight away, its first segment aimed at . + /// + private static async Task WalkBackAsync(Ctx ctx, DateTime hi, DateTime? known, TimeSpan width) + { + TimeSpan? aim = known is null ? null : width; + + // Gallop: no stored row known below hi yet. + var empties = 0; + while (known is null) + { + if (hi <= Floor) return null; + var rest = empties >= 2; + var lo = rest || (hi - Floor).Ticks <= width.Ticks ? Floor : hi - width; + var seg = await SegmentAsync(ctx, lo, hi).ConfigureAwait(false); + if (seg.Complete) + { + if (seg.Last is { } found) return found; + if (lo <= Floor) return null; + hi = lo; + empties++; + width = Grow(width); + continue; + } + // Full — a whole-remainder read of deep history too: its last row is a stored row below hi. Bisect above it. + if (seg.Last is not { } seen) throw TooDense(ctx, lo); + known = seen.TimeUtc; + aim = AimFrom(seg.First!.Value.TimeUtc, seen.TimeUtc); + } + + // Bisect: a stored row exists at `known`, nothing newer than `hi` matters; the answer is the newest row in [known, hi]. + var aimMissed = false; + while (true) + { + var k = known.Value; + DateTime lo; + var aimed = false; + if (hi - k < TimeSpan.FromMilliseconds(2)) + lo = k.AddTicks(-1); // the whole stretch from the known row: never a sub-millisecond step + else + { + lo = k + TimeSpan.FromTicks((hi - k).Ticks / 2); + if (aim is { } a && hi - a > lo) + { + lo = hi - a; // an aimed try right after a full read — until one comes back empty (a gap: halve from then on) + aimed = true; + } + } + aim = null; + + var seg = await SegmentAsync(ctx, lo, hi).ConfigureAwait(false); + if (seg.Complete) + { + if (seg.Last is { } found) return found; + if (lo < k) return null; // the known row is gone (deleted meanwhile): nothing is left to find + hi = lo; + aimMissed |= aimed; + continue; + } + if (seg.Last is not { } seen || seen.TimeUtc <= k) throw TooDense(ctx, k); + known = seen.TimeUtc; + if (!aimMissed) aim = AimFrom(seg.First!.Value.TimeUtc, seen.TimeUtc); + } + } + + /// About a quarter of a read at the density a full read showed (its rows from to ). + private static TimeSpan AimFrom(DateTime first, DateTime last) => + TimeSpan.FromTicks(Math.Max(Ms, (last - first).Ticks / 4)); + + /// The first stored sample after in (qs, end), from a small forward read. + private static async Task FirstAfterAsync(Ctx ctx, DateTime qs, DateTime end, DateTime after) + { + foreach (var budget in new[] { ProbeRows, ReadRows }) + { + var read = await ReadTrustedAsync(ctx, qs, end, budget).ConfigureAwait(false); + foreach (var r in read.Rows) + if (r.TimeUtc > after) return r; + if (read.Complete) return null; + } + throw TooDense(ctx, qs); + } + + /// + /// One forward read [qs, qe] (whole milliseconds), drained, keeping only rows strictly inside it and after the + /// epoch — counted against the request's read cap. A strict gateway's ceiling is learned on refusal and then asked for + /// directly; a full read is not re-probed (the caller pages on, it does not need the rest of this read). + /// + private static async Task ReadTrustedAsync(Ctx ctx, DateTime qs, DateTime qe, int budget) + { + if (++ctx.Reads > ctx.MaxReads) + throw new HistoryReadLimitExceededException(ctx.MaxReads, + $"The AtTime read of '{ctx.Tag}' needed more than {ctx.MaxReads} raw reads ({BaseReads} + {ReadsPerTime} per " + + "requested time)" + CeilingNote(ctx) + "; the Gateway backend serves AtTime from raw reads — request fewer times."); + var (read, used) = await ctx.Ceiling.ReadAsync(budget, b => PassAsync(ctx, qs, qe, b), (_, _) => false) + .ConfigureAwait(false); + if (used < ctx.SmallestBudget) ctx.SmallestBudget = used; + if (used < MinBudget) + throw new HistoryReadLimitExceededException(used, + $"The HistorianGateway's row ceiling ({used}, its Historian:ReadLimits:MaxValuesCeiling with " + + $"AllowStreamingBeyondCeiling=false) is below the {MinBudget} rows an AtTime raw read of '{ctx.Tag}' needs."); + return read; + } + + private static string CeilingNote(Ctx ctx) => + ctx.SmallestBudget < JumpRows + ? $", with reads bounded by the HistorianGateway's row ceiling of {ctx.SmallestBudget} (Historian:ReadLimits:MaxValuesCeiling)" + : string.Empty; + + private static async Task PassAsync(Ctx ctx, DateTime qs, DateTime qe, int budget) + { + var rows = new List(); + var received = 0; + await foreach (var raw in ctx.Client.ReadRawAsync(ctx.Tag, qs, qe, budget, ctx.Ct).ConfigureAwait(false)) + { + received++; // drained to completion, never broken out of + if (raw.Timestamp is null) continue; + var t = raw.Timestamp.ToDateTime(); + if (t > qs && t < qe && t > Floor) rows.Add(ToStored(raw, t, ctx.ValueType)); + } + return new Read(rows, received < budget); + } + + /// + /// A gateway row as a stored sample of the node's type (see the type remarks); a Bad row carries no value (#690: a + /// stored NULL arrives as value 0). + /// + private static StoredSample ToStored(HistorianSample raw, DateTime t, TypeCode? valueType) + { + var status = SampleMapper.ToStatus(raw); + var value = StatusBits.IsBad(status) ? HistorianValue.Empty + : raw.HasNumericValue ? Typed(raw.NumericValue, valueType) + : raw.HasStringValue ? HistorianValue.FromString(raw.StringValue) + : HistorianValue.Empty; + return StoredSample.FromStored(t, value, status); + } + + /// A gateway number as the hinted kind, so the shared evaluator's own type branch decides stepped vs linear. + private static HistorianValue Typed(double v, TypeCode? valueType) => valueType switch + { + TypeCode.Boolean => HistorianValue.FromBoolean(v != 0), + TypeCode.String => HistorianValue.FromString(v.ToString("R", System.Globalization.CultureInfo.InvariantCulture)), + // In Int64 range only (2^63 is exactly representable); NaN, ±∞ or out of range is no integer at all — Empty, which + // StoredSample.FromStored makes BadNoData (unusable), never a saturated Good value. + _ when IsInteger(valueType) => v >= -9.2233720368547758E18 && v < 9.2233720368547758E18 + ? HistorianValue.FromInt64((long)Math.Round(v, MidpointRounding.AwayFromZero)) + : HistorianValue.Empty, + _ => HistorianValue.FromDouble(v), + }; + + /// The window read's start for a time: a millisecond before its millisecond (never before the epoch). + private static DateTime RequestStart(DateTime t) + { + var s = MsFloor(t) - TimeSpan.FromMilliseconds(1); + return s < Floor ? Floor : s; + } + + private static DateTime MsFloor(DateTime t) => new(t.Ticks - t.Ticks % Ms, DateTimeKind.Utc); + + private static DateTime Min(DateTime a, DateTime b) => a <= b ? a : b; + + private static DateTime Max(DateTime a, DateTime b) => a >= b ? a : b; + + private static TimeSpan Min(TimeSpan a, TimeSpan b) => a <= b ? a : b; + + private static TimeSpan Clamp(TimeSpan w) => + w < TimeSpan.FromMilliseconds(1) ? TimeSpan.FromMilliseconds(1) : w > MaxSeededWidth ? MaxSeededWidth : w; + + private static TimeSpan Grow(TimeSpan w) => w.Ticks > TimeSpan.MaxValue.Ticks / 4 ? TimeSpan.MaxValue : w * 4; + + private static HistoryReadLimitExceededException TooDense(Ctx ctx, DateTime at) + { + var rows = Math.Min(ReadRows, ctx.SmallestBudget); + return new(rows, + $"The AtTime read of '{ctx.Tag}' met more than {rows} stored samples within about one millisecond of {at:O}" + + (rows < ReadRows ? $" (reads are bounded by the HistorianGateway's row ceiling of {rows})" : string.Empty) + + "; the Gateway backend serves AtTime from raw reads."); + } +} 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 a5041334..1d250222 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 @@ -46,6 +46,10 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis private readonly ILogger _logger; private readonly int _maxRawRows; private readonly TimeProvider _time; + private readonly bool _atTimeFromRaw; + + /// #693: AtTime from raw reads (true) or AVEVA's own interpolation (the kill switch). + internal bool AtTimeFromRaw => _atTimeFromRaw; /// A strict gateway's row ceiling, learned from its first refusal (#691.3). private readonly GatewayRowCeiling _ceiling = new(); @@ -67,9 +71,15 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis /// <= 0 = unbounded). Past it the node fails BadTooManyOperations. /// /// Clock for result server timestamps and the raw look-ahead horizon; null = system. + /// + /// #693: serve AtTime from raw reads with the shared AtTime rules (; default); + /// false = AVEVA's own interpolation passed through (ServerHistorian:GatewayAtTimeFromRaw kill switch). + /// public GatewayHistorianDataSource(IHistorianGatewayClient client, ILogger logger, - int maxRawRows = TimeWeightedAggregator.DefaultMaxWindowRows, TimeProvider? timeProvider = null) + int maxRawRows = TimeWeightedAggregator.DefaultMaxWindowRows, TimeProvider? timeProvider = null, + bool atTimeFromRaw = true) { + _atTimeFromRaw = atTimeFromRaw; ArgumentNullException.ThrowIfNull(client); ArgumentNullException.ThrowIfNull(logger); _client = client; @@ -364,7 +374,59 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis internal static bool IsMaxValuesCeilingRejection(Exception ex) => GatewayReadLimits.IsCeilingRejection(ex); /// + /// + /// #693: served from RAW gateway reads — each requested time's stored sample at-or-before and after it — evaluated with + /// the SQL backends' shared (Part 11 §6.5.4), so the Gateway answers what SqlServer and + /// QuestDB answer for the same stored samples; see for the read plan and its cost. + /// AVEVA's own interpolated answer (the gateway's ReadAtTime) returned the previous value as Good at a stored + /// NULL's time, null + Bad at a Good sample right after a NULL, a drifted value at an exact stored sample, and no + /// Interpolated bit; it is only used when ServerHistorian:GatewayAtTimeFromRaw = false (kill switch). + /// A gateway value is a Double (or a String) whatever the node's type (AVEVA stores a Boolean as a 0/1 Int2), so + /// without a type hint every numeric tag is interpolated linearly; the node manager passes the node's type through + /// . + /// + public Task ReadAtTimeAsync( + string fullReference, IReadOnlyList timestampsUtc, CancellationToken cancellationToken) => + ReadAtTimeAsync(fullReference, timestampsUtc, valueType: null, cancellationToken); + + /// + /// + /// #693: the hint makes the AtTime rules type-exact, as on the SQL backends — the gateway's Double rows are converted to + /// the node's type BEFORE the shared evaluator runs (Boolean = value ≠ 0 ⇒ stepped; an integer type ⇒ Int64 ⇒ + /// interpolated and rounded half away from zero, then returned as the node's integer type; String ⇒ stepped; Float / + /// Double / no hint ⇒ linear Double). Ignored with the kill switch (AVEVA's own answer). + /// public async Task ReadAtTimeAsync( + string fullReference, IReadOnlyList timestampsUtc, TypeCode? valueType, CancellationToken cancellationToken) + { + ArgumentNullException.ThrowIfNull(timestampsUtc); + if (!_atTimeFromRaw) return await ReadAtTimeNativeAsync(fullReference, timestampsUtc, cancellationToken).ConfigureAwait(false); + try + { + var now = _time.GetUtcNow().UtcDateTime; + var values = await GatewayAtTimeReader + .ReadAsync(_client, _ceiling, fullReference, timestampsUtc, now, cancellationToken, valueType) + .ConfigureAwait(false); + RecordOutcome(success: true, error: null); + return new HistoryReadResult(values, ContinuationPoint: null); + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + throw; // caller-requested, not a backend fault + } + catch (HistoryReadLimitExceededException) + { + throw; // a client-request limit (a pathological tie cluster), not a backend fault + } + catch (Exception ex) + { + RecordReadFailure(ex); + throw; + } + } + + /// The pre-#693 AtTime: AVEVA's own interpolation, aligned to the request (kill switch only). + private async Task ReadAtTimeNativeAsync( string fullReference, IReadOnlyList timestampsUtc, CancellationToken cancellationToken) { try @@ -494,7 +556,7 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis /// /// Reconciles a gateway at-time reply against the requested timestamps to honour the - /// contract: exactly one snapshot per + /// contract: exactly one snapshot per /// requested timestamp, in request order. Returned samples are indexed by timestamp ticks; /// any requested timestamp the gateway did not return is filled with a Bad-quality /// (0x80000000) snapshot stamped at the requested time rather than positionally diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianServiceCollectionExtensions.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianServiceCollectionExtensions.cs index 7927e71d..67e91ad6 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianServiceCollectionExtensions.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianServiceCollectionExtensions.cs @@ -40,7 +40,9 @@ public static class GatewayHistorian HistorianGatewayClientAdapter.Create(options, loggerFactory), logger, // #681: the raw-evaluated per-sample aggregates share the SQL backends' raw-row cap. - maxRawRows: options.MaxTimeWeightedRawRows); + maxRawRows: options.MaxTimeWeightedRawRows, + // #693: AtTime from raw reads + the shared evaluator; false = AVEVA's own interpolation (kill switch). + atTimeFromRaw: options.GatewayAtTimeFromRaw); } /// diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common/AtTimeEvaluator.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common/AtTimeEvaluator.cs index 9440b843..c324680c 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common/AtTimeEvaluator.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common/AtTimeEvaluator.cs @@ -49,8 +49,10 @@ public static class AtTimeEvaluator { var fraction = (double)(t.Ticks - p.TimeUtc.Ticks) / (n.TimeUtc.Ticks - p.TimeUtc.Ticks); var v = p.Value.Number + (nv - p.Value.Number) * fraction; + // Each arm boxed on its own: `cond ? (long)x : v` is a DOUBLE expression in C# (the long widens), which returned + // every interpolated Int64 as a Double until #693. object value = p.Value.Kind == HistorianValueKind.Int64 - ? (long)Math.Round(v, MidpointRounding.AwayFromZero) + ? (object)(long)Math.Round(v, MidpointRounding.AwayFromZero) : v; var status = p.IsGood && n.IsGood ? HistorianStatusCodes.Good : HistorianStatusCodes.UncertainDataSubNormal; return new DataValueSnapshot(value, HistorianInfoBits.WithInterpolated(status), t, serverTime); diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs index 735effdf..c58d144d 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs @@ -3399,6 +3399,36 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 return NodeId.Null; } + /// + /// Gitea #693: the AtTime value-type hint of a historized variable — its DataType's built-in type as a + /// when the node is a scalar of a type the historian evaluates (Boolean, the integers, Float, + /// Double, String, DateTime); null for an array, an unknown node, or any other type. A UNS mirror carries its + /// backing raw tag's DataType, so either NodeId gives the same hint. + /// + /// The HistoryRead request's node id. + /// The hint, or null. + internal TypeCode? AtTimeValueType(NodeId nodeId) + { + if (!_variables.TryGetValue(MapKey(nodeId), out var variable) || variable.ValueRank != ValueRanks.Scalar) return null; + return TypeInfo.GetBuiltInType(variable.DataType) switch + { + BuiltInType.Boolean => TypeCode.Boolean, + BuiltInType.SByte => TypeCode.SByte, + BuiltInType.Byte => TypeCode.Byte, + BuiltInType.Int16 => TypeCode.Int16, + BuiltInType.UInt16 => TypeCode.UInt16, + BuiltInType.Int32 => TypeCode.Int32, + BuiltInType.UInt32 => TypeCode.UInt32, + BuiltInType.Int64 => TypeCode.Int64, + BuiltInType.UInt64 => TypeCode.UInt64, + BuiltInType.Float => TypeCode.Single, + BuiltInType.Double => TypeCode.Double, + BuiltInType.String => TypeCode.String, + BuiltInType.DateTime => TypeCode.DateTime, + _ => null, + }; + } + /// Map a Tag.DataType string ("Boolean", "Int32", "Float", "Double", "String", /// "DateTime") to the OPC UA built-in NodeId. Unknown names fall back to BaseDataType /// (matches CreateVariable's default for lazy-created nodes). @@ -5177,9 +5207,12 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 var work = new List>(nodesToProcess.Count); foreach (var handle in nodesToProcess) { + // #693: the node's scalar value type, for a backend whose stored values carry none (the Gateway). + var valueType = AtTimeValueType(handle.NodeId); work.Add(() => ServeNodeAsync(handle, results, errors, (source, tagname, token) => source.ReadAtTimeAsync( tagname, timestamps, + valueType, token), ct)); } 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 37c04091..70c65680 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ServerHistorianOptions.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ServerHistorianOptions.cs @@ -97,6 +97,15 @@ public sealed class ServerHistorianOptions /// public bool HonorReturnBounds { get; init; } = true; + /// + /// Gitea #693, Gateway backend only: whether HistoryRead-AtTime is served from RAW gateway reads evaluated with + /// the SQL backends' shared AtTime rules (exact stored value at a stored time, linear between usable neighbours with + /// the Interpolated bit, held before a Bad neighbour as Uncertain, null + the outage's Bad code inside an outage, + /// BadNoData before the first sample). Default true. false is the kill switch back to the + /// pre-#693 answer: AVEVA's own interpolation (the gateway's ReadAtTime), passed through unchanged. + /// + public bool GatewayAtTimeFromRaw { get; init; } = true; + /// /// Maximum number of historized nodes served CONCURRENTLY within one HistoryRead batch (arch-review /// 03/S3). The SDK's HistoryRead* override surface is synchronous, so each arm block-bridges diff --git a/tests/Drivers/Historian.Conformance/AtTimeConformance.cs b/tests/Drivers/Historian.Conformance/AtTimeConformance.cs index 2f8a9341..0379f484 100644 --- a/tests/Drivers/Historian.Conformance/AtTimeConformance.cs +++ b/tests/Drivers/Historian.Conformance/AtTimeConformance.cs @@ -139,6 +139,7 @@ public abstract class AtTimeConformance : ConformanceBase (At(10), HistorianValue.FromInt64(15), Good)); var s = (await B.CreateDataSource().ReadAtTimeAsync("I", [At(5), At(2)], Ct)).Samples; s[0].Value.ShouldBe(13L); // 12.5 ⇒ 13 (away from zero) + s[0].Value.ShouldBeOfType(); // ShouldBe compares numbers across types — the type needs its own check (#693) s[1].Value.ShouldBe(11L); // 11.0 s[0].StatusCode.ShouldBe(GoodInterpolated); } diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayAtTimeTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayAtTimeTests.cs new file mode 100644 index 00000000..11e47fa2 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayAtTimeTests.cs @@ -0,0 +1,717 @@ +using Google.Protobuf.WellKnownTypes; +using Microsoft.Extensions.Logging.Abstractions; +using Shouldly; +using Xunit; +using ZB.MOM.WW.HistorianGateway.Contracts.Grpc; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests; + +/// +/// Gitea #693: Gateway HistoryRead AtTime is served from RAW reads and evaluated with the SQL backends' shared +/// AtTimeEvaluator (Part 11 §6.5.4) — never AVEVA's own interpolated answer, which the live gate caught returning +/// the previous value as Good at a stored NULL's own time, null + Bad at a Good sample right after a NULL, a drifted value +/// at an exact stored sample, and no Interpolated bit. The fake is AVEVA-shaped (the real wire shape): ReadRaw over +/// [start, end] ascending, capped, led by the synthetic boundary row at the query start (the value in effect there, +/// Quality = 133, or a NULL row), NULL rows as NumericValue = 0 + q=1 opc=0, millisecond timestamps; +/// and it REJECTS what the real gateway rejects (max_values <= 0, end < start). +/// +public sealed class GatewayAtTimeTests +{ + private static CancellationToken Ct => TestContext.Current.CancellationToken; + + private const uint Good = 0u; + private const uint Bad = 0x80000000u; + private const uint BadNoData = 0x809B0000u; + private const uint GoodInterpolated = 0x00000402u; + private const uint UncertainInterpolated = 0x40A40402u; // UncertainDataSubNormal | InfoType DataValue | Interpolated + + /// The live gate's tag HistGW.LiveTest.Gate685: 19:00:03 is a stored NULL. + private static readonly DateTime T0 = new(2026, 10, 12, 19, 0, 0, DateTimeKind.Utc); + + private static readonly DateTime Now = T0.AddHours(1); + + private static DateTime At(double seconds) => T0.AddTicks((long)(seconds * TimeSpan.TicksPerSecond)); + + private static List<(DateTime T, double V, bool Null)> Gate685() => + [(At(0), 100, false), (At(2), 107, false), (At(3), 0, true), (At(5), 108, false)]; + + /// An AVEVA-shaped store over sorted rows, counting reads and rejecting what the real gateway rejects. + private sealed class AvevaStore + { + private readonly (DateTime T, double V, bool Null)[] _rows; + + public AvevaStore(IEnumerable<(DateTime T, double V, bool Null)> rows) => _rows = rows.OrderBy(r => r.T).ToArray(); + + /// Also emit a row at the exact query END carrying the value in effect there (unverified AVEVA behaviour). + public bool TerminalRow { get; init; } + + /// Total rows streamed (cost). + public long RowsStreamed; + + public FakeHistorianGatewayClient Client(int? ceiling = null) => new() + { + RawSource = Serve, + MaxValuesCeiling = ceiling, + }; + + private static HistorianSample Row(DateTime t, double v, bool isNull, uint quality = 0) => new() + { + Tag = "T", + Timestamp = Timestamp.FromDateTime(t), + NumericValue = isNull ? 0 : v, + OpcQuality = isNull ? 0u : 192u, + Quality = isNull ? 1u : quality, + }; + + private int FirstAtOrAfter(DateTime t) + { + int lo = 0, hi = _rows.Length; + while (lo < hi) + { + var mid = (lo + hi) / 2; + if (_rows[mid].T < t) lo = mid + 1; else hi = mid; + } + return lo; + } + + private HistorianSample InEffect(DateTime t) + { + var i = FirstAtOrAfter(t) - 1; + if (i < 0) return new HistorianSample { Tag = "T", Timestamp = Timestamp.FromDateTime(t), NumericValue = 0, OpcQuality = 0, Quality = 1 }; + var p = _rows[i]; + return Row(t, p.V, p.Null, p.Null ? 1u : 133u); + } + + private IEnumerable Serve(DateTime start, DateTime end, int max) + { + if (max <= 0) + throw new Grpc.Core.RpcException(new Grpc.Core.Status(Grpc.Core.StatusCode.InvalidArgument, "max_values must be greater than zero.")); + if (end < start) + throw new Grpc.Core.RpcException(new Grpc.Core.Status(Grpc.Core.StatusCode.InvalidArgument, + "start_time must be less than or equal to end_time.")); + if (end > Now.AddDays(2)) + throw new Grpc.Core.RpcException(new Grpc.Core.Status(Grpc.Core.StatusCode.InvalidArgument, + "end_time is outside the historian's time range")); + if (start.Ticks % TimeSpan.TicksPerMillisecond != 0 || end.Ticks % TimeSpan.TicksPerMillisecond != 0) + throw new InvalidOperationException("AVEVA truncates request times to the millisecond; send millisecond bounds."); + + var rows = new List(); + var i = FirstAtOrAfter(start); + if (i >= _rows.Length || _rows[i].T != start) rows.Add(InEffect(start)); + for (; i < _rows.Length && _rows[i].T <= end && rows.Count < max; i++) + rows.Add(Row(_rows[i].T, _rows[i].V, _rows[i].Null)); + if (TerminalRow && rows.Count < max && (i >= _rows.Length || _rows[i].T > end) && _rows.All(r => r.T != end)) + rows.Add(InEffect(end)); + var served = rows.Take(max).ToList(); + RowsStreamed += served.Count; + return served; + } + } + + private static GatewayHistorianDataSource Source(FakeHistorianGatewayClient fake) => + new(fake, NullLogger.Instance, timeProvider: new FixedTime(Now)); + + private sealed class FixedTime(DateTime now) : TimeProvider + { + public override DateTimeOffset GetUtcNow() => new(now); + } + + private static async Task> AtTime(FakeHistorianGatewayClient fake, params DateTime[] times) + { + var r = await Source(fake).ReadAtTimeAsync("T", times, Ct); + r.Samples.Count.ShouldBe(times.Length); + for (var i = 0; i < times.Length; i++) r.Samples[i].SourceTimestampUtc.ShouldBe(times[i]); + var refused = fake.MaxValuesCeiling is { } c ? fake.RawCalls.Count(x => x.MaxValues > c) : 0; + fake.RawStreamsCompleted.ShouldBe(fake.ReadRawCallCount - refused); // never abandons a gateway stream + fake.ReadAtTimeCallCount.ShouldBe(0); // never AVEVA's own interpolation + return r.Samples; + } + + // ---- the four live findings ------------------------------------------------------------------------------------ + + [Fact] // live: 19:00:03 (a stored NULL) ⇒ 107 Good + public async Task At_a_stored_NULLs_own_time_is_null_and_Bad() + { + var s = (await AtTime(new AvevaStore(Gate685()).Client(), At(3))).Single(); + s.Value.ShouldBeNull(); + s.StatusCode.ShouldBe(Bad); + } + + [Fact] // live: 19:00:05 (stored 108, right after the NULL) ⇒ null + Bad + public async Task At_a_Good_sample_right_after_a_NULL_is_the_stored_value() + { + var s = (await AtTime(new AvevaStore(Gate685()).Client(), At(5))).Single(); + s.Value.ShouldBe(108d); + s.StatusCode.ShouldBe(Good); + } + + [Fact] // live: 99.99999999944 for a stored 100 + public async Task At_an_exact_stored_sample_is_the_exact_stored_value_without_the_Interpolated_bit() + { + var s = await AtTime(new AvevaStore(Gate685()).Client(), At(0), At(2)); + s[0].Value.ShouldBe(100d); + s[0].StatusCode.ShouldBe(Good); + s[1].Value.ShouldBe(107d); + s[1].StatusCode.ShouldBe(Good); + } + + [Fact] // live: no Interpolated bit + public async Task Between_two_Good_samples_is_linear_with_the_Interpolated_bit() + { + var s = (await AtTime(new AvevaStore(Gate685()).Client(), At(1))).Single(); + s.Value.ShouldBe(103.5d); + s.StatusCode.ShouldBe(GoodInterpolated); + new Opc.Ua.StatusCode(s.StatusCode).AggregateBits.ShouldBe(Opc.Ua.AggregateBits.Interpolated); + } + + // ---- the rest of the AtTime rules -------------------------------------------------------------------------------- + + [Fact] + public async Task Before_a_Bad_neighbour_the_previous_value_is_held_Uncertain_and_Interpolated() + { + var s = (await AtTime(new AvevaStore(Gate685()).Client(), At(2.5))).Single(); + s.Value.ShouldBe(107d); + s.StatusCode.ShouldBe(UncertainInterpolated); + } + + [Fact] + public async Task Inside_an_outage_is_null_with_the_outages_Bad_code() + { + var s = (await AtTime(new AvevaStore(Gate685()).Client(), At(4))).Single(); + s.Value.ShouldBeNull(); + s.StatusCode.ShouldBe(Bad); + } + + [Fact] + public async Task Before_the_first_sample_is_BadNoData_not_the_boundary_rows_zero() + { + var s = await AtTime(new AvevaStore(Gate685()).Client(), At(-1), At(-0.001), At(-3600)); + s.ShouldAllBe(x => x.StatusCode == BadNoData && x.Value == null); + } + + [Fact] + public async Task After_the_last_sample_is_stepped_Good_until_now_then_Uncertain() + { + var s = await AtTime(new AvevaStore(Gate685()).Client(), At(10), Now.AddMinutes(5), Now.AddDays(3)); + s[0].Value.ShouldBe(108d); + s[0].StatusCode.ShouldBe(GoodInterpolated); + s[1].Value.ShouldBe(108d); + s[1].StatusCode.ShouldBe(UncertainInterpolated); // later than now: not known to still hold + s[2].Value.ShouldBe(108d); + s[2].StatusCode.ShouldBe(UncertainInterpolated); // beyond the horizon: never sent to the gateway (it refuses) + } + + [Fact] + public async Task A_mixed_request_keeps_order_and_duplicates() + { + var s = await AtTime(new AvevaStore(Gate685()).Client(), At(5), At(1), At(-1), At(3), At(1), At(2.5)); + s.Select(x => x.Value).ShouldBe(new object?[] { 108d, 103.5d, null, null, 103.5d, 107d }); + s.Select(x => x.StatusCode).ShouldBe(new[] { Good, GoodInterpolated, BadNoData, Bad, GoodInterpolated, UncertainInterpolated }); + } + + [Fact] + public async Task An_empty_request_reads_nothing() + { + var fake = new AvevaStore(Gate685()).Client(); + (await Source(fake).ReadAtTimeAsync("T", [], Ct)).Samples.ShouldBeEmpty(); + fake.ReadRawCallCount.ShouldBe(0); + } + + [Fact] + public async Task A_sub_millisecond_request_time_interpolates_between_the_millisecond_samples() + { + var s = (await AtTime(new AvevaStore(Gate685()).Client(), At(0.0005))).Single(); + ((double)s.Value!).ShouldBe(100.00175d, 1e-9); + s.StatusCode.ShouldBe(GoodInterpolated); + } + + [Fact] + public async Task A_row_at_the_query_end_never_counts_as_a_stored_sample() + { + var store = new AvevaStore(Gate685()) { TerminalRow = true }; + var s = await AtTime(store.Client(), At(1), At(2.5), At(4), At(10)); + s.Select(x => x.Value).ShouldBe(new object?[] { 103.5d, 107d, null, 108d }); + s.Select(x => x.StatusCode).ShouldBe(new[] { GoodInterpolated, UncertainInterpolated, Bad, GoodInterpolated }); + } + + [Fact] + public async Task A_string_tag_is_stepped() + { + var fake = new FakeHistorianGatewayClient + { + RawSource = (start, end, max) => + { + var rows = new List + { + new() { Tag = "T", Timestamp = Timestamp.FromDateTime(start), StringValue = "", OpcQuality = 0, Quality = 1 }, + }; + foreach (var (t, v) in new[] { (At(0), "a"), (At(10), "b") }) + if (t >= start && t <= end) + rows.Add(new HistorianSample { Tag = "T", Timestamp = Timestamp.FromDateTime(t), StringValue = v, OpcQuality = 192 }); + return rows.Take(max); + }, + }; + var s = (await AtTime(fake, At(5))).Single(); + s.Value.ShouldBe("a"); + s.StatusCode.ShouldBe(GoodInterpolated); + } + + // ---- the node's value type (the node manager's hint; AVEVA stores a Boolean as a 0/1 Int2) ------------------------ + + /// Reads through the INTERFACE's hinted overload, as the node manager does. + private static async Task> AtTimeTyped( + FakeHistorianGatewayClient fake, TypeCode? valueType, params DateTime[] times) + { + IHistorianDataSource source = Source(fake); + var r = await source.ReadAtTimeAsync("T", times, valueType, Ct); + r.Samples.Count.ShouldBe(times.Length); + fake.ReadAtTimeCallCount.ShouldBe(0); + return r.Samples; + } + + [Fact] + public async Task A_Boolean_node_is_stepped_never_a_fraction() + { + var fake = new AvevaStore([(At(0), 0, false), (At(10), 1, false), (At(20), 0, false)]).Client(); + var s = await AtTimeTyped(fake, TypeCode.Boolean, At(5), At(10), At(15), At(25)); + s.Select(x => x.Value).ShouldBe(new object?[] { false, true, true, false }); + s.ShouldAllBe(x => x.Value is bool); + s.Select(x => x.StatusCode).ShouldBe(new[] { GoodInterpolated, Good, GoodInterpolated, GoodInterpolated }); + } + + [Fact] + public async Task Without_a_hint_a_Boolean_tag_interpolates_as_before() + { + var fake = new AvevaStore([(At(0), 0, false), (At(10), 1, false)]).Client(); + (await AtTimeTyped(fake, null, At(5))).Single().Value.ShouldBeOfType().ShouldBe(0.5d); + } + + [Fact] + public async Task An_Int32_node_interpolates_rounds_half_away_from_zero_and_keeps_its_type() + { + var s = await AtTimeTyped(new AvevaStore(Gate685()).Client(), TypeCode.Int32, At(0), At(0.5), At(1), At(2.5), At(3)); + s[0].Value.ShouldBe(100); // exact hit, as stored, Int32 + s[0].StatusCode.ShouldBe(Good); + s[1].Value.ShouldBe(102); // 101.75 ⇒ 102 + s[2].Value.ShouldBe(104); // 103.5 ⇒ 104 (half away from zero) + s[1].StatusCode.ShouldBe(GoodInterpolated); + s[3].Value.ShouldBe(107); // held before the NULL, Uncertain + s[3].StatusCode.ShouldBe(UncertainInterpolated); + s[4].Value.ShouldBeNull(); // the NULL itself + s[4].StatusCode.ShouldBe(Bad); + s.Take(4).Select(x => x.Value + ":" + x.Value!.GetType().Name).ShouldBe(["100:Int32", "102:Int32", "104:Int32", "107:Int32"]); + } + + [Fact] + public async Task A_negative_midpoint_rounds_away_from_zero() + { + var fake = new AvevaStore([(At(0), -100, false), (At(2), -107, false)]).Client(); + (await AtTimeTyped(fake, TypeCode.Int16, At(1))).Single().Value.ShouldBeOfType().ShouldBe((short)-104); + } + + [Fact] + public async Task Other_integer_nodes_keep_their_own_type() + { + var fake = new AvevaStore(Gate685()).Client(); + (await AtTimeTyped(fake, TypeCode.UInt16, At(1))).Single().Value.ShouldBeOfType().ShouldBe((ushort)104); + (await AtTimeTyped(fake, TypeCode.Int64, At(1))).Single().Value.ShouldBeOfType().ShouldBe(104L); + (await AtTimeTyped(fake, TypeCode.Byte, At(1))).Single().Value.ShouldBeOfType().ShouldBe((byte)104); + } + + [Fact] + public async Task Float_and_Double_nodes_interpolate_unchanged() + { + var fake = new AvevaStore(Gate685()).Client(); + (await AtTimeTyped(fake, TypeCode.Double, At(1))).Single().Value.ShouldBeOfType().ShouldBe(103.5d); + (await AtTimeTyped(fake, TypeCode.Single, At(1))).Single().Value.ShouldBeOfType().ShouldBe(103.5d); // as the SQL backends (Double kind) + } + + // ---- cost ------------------------------------------------------------------------------------------------------ + + private static IEnumerable<(DateTime T, double V, bool Null)> OneHertz(DateTime from, int seconds) => + Enumerable.Range(0, seconds).Select(i => (from.AddSeconds(i), (double)i, false)); + + [Fact] + public async Task Ten_exact_hits_on_dense_data_are_one_read() + { + var fake = new AvevaStore(OneHertz(T0.AddHours(-1), 3600)).Client(); + var times = Enumerable.Range(0, 10).Select(i => T0.AddMinutes(-30).AddSeconds(i)).ToArray(); + var s = await AtTime(fake, times); + s.Select(x => x.Value).ShouldBe(Enumerable.Range(1800, 10).Select(i => (object?)(double)i)); + s.ShouldAllBe(x => x.StatusCode == Good); + fake.ReadRawCallCount.ShouldBe(1); + } + + [Fact] + public async Task Ten_interpolated_times_on_dense_data_are_three_reads() + { + var fake = new AvevaStore(OneHertz(T0.AddHours(-1), 3600)).Client(); + var times = Enumerable.Range(0, 10).Select(i => T0.AddMinutes(-30).AddSeconds(i + 0.5)).ToArray(); + var s = await AtTime(fake, times); + s.Select(x => x.Value).ShouldBe(Enumerable.Range(1800, 10).Select(i => (object?)(i + 0.5))); + s.ShouldAllBe(x => x.StatusCode == GoodInterpolated); + fake.ReadRawCallCount.ShouldBe(3); // the window, the sample before the first time, the sample after the last + } + + [Fact] + public async Task Ten_times_an_hour_apart_on_dense_data_cost_about_one_read_each_not_the_rows_between() + { + var store = new AvevaStore(OneHertz(T0.AddHours(-12), 12 * 3600)); + var fake = store.Client(); + var times = Enumerable.Range(0, 10).Select(i => T0.AddHours(-11).AddHours(i).AddSeconds(0.5)).ToArray(); + var s = await AtTime(fake, times); + s.Select(x => x.Value).ShouldBe(Enumerable.Range(0, 10).Select(i => (object?)(3600.0 * (i + 1) + 0.5))); + s.ShouldAllBe(x => x.StatusCode == GoodInterpolated); + fake.ReadRawCallCount.ShouldBe(12); // 10 window reads (each finds its own prev) + 1 walk-back + 1 probe + store.RowsStreamed.ShouldBeLessThan(10 * 1100); // never the ~32,400 rows between the first and last time + } + + [Fact] + public async Task Sparse_history_far_back_is_found_in_a_bounded_number_of_reads() + { + var fake = new AvevaStore([(T0.AddDays(-400), 42, false), (T0.AddDays(-400).AddSeconds(10), 43, false)]).Client(); + var s = (await AtTime(fake, T0)).Single(); + s.Value.ShouldBe(43d); + s.StatusCode.ShouldBe(GoodInterpolated); + fake.ReadRawCallCount.ShouldBe(5); // window, 1-min + 4-min segments, the whole remainder, the probe after + } + + [Fact] + public async Task A_tag_with_no_history_is_BadNoData_in_a_bounded_number_of_reads() + { + var fake = new AvevaStore([]).Client(); + var s = await AtTime(fake, At(1), At(2)); + s.ShouldAllBe(x => x.StatusCode == BadNoData); + fake.ReadRawCallCount.ShouldBe(4); // window, two segments, the whole remainder (no probe: nothing to interpolate) + } + + [Fact] + public async Task Dense_data_far_denser_than_one_read_still_finds_the_neighbours() + { + // 1 kHz for 10 minutes: a 1-minute walk-back segment holds 60,000 rows (truncated) — it must narrow, not fail. + var rows = Enumerable.Range(0, 600_000).Select(i => (T0.AddMinutes(-20).AddMilliseconds(i), (double)i, false)); + var fake = new AvevaStore(rows).Client(); + var t = T0.AddMinutes(-15).AddTicks(5000); // 0.5 ms past sample 300,000 + var s = (await AtTime(fake, t)).Single(); + s.Value.ShouldBe(300_000.5d); + s.StatusCode.ShouldBe(GoodInterpolated); + fake.ReadRawCallCount.ShouldBe(2); // the window (holds the prev) + the probe + } + + [Fact] + public async Task A_walk_back_into_data_far_denser_than_one_read_narrows_instead_of_paging() + { + // 1 kHz for 10 minutes, then a 10-minute gap: the walk-back crosses the gap and lands in 600,000 rows. + var rows = Enumerable.Range(0, 600_000).Select(i => (T0.AddMinutes(-20).AddMilliseconds(i), (double)i, false)); + var store = new AvevaStore(rows); + var fake = store.Client(); + var s = (await AtTime(fake, T0.AddMinutes(-5))).Single(); + s.Value.ShouldBe(599_999d); + s.StatusCode.ShouldBe(GoodInterpolated); // stepped from the last sample (before now) + fake.ReadRawCallCount.ShouldBe(6); // window, 1m, 4m, remainder (full: seeds the bisection), one aimed segment, probe + store.RowsStreamed.ShouldBeLessThan(10_000); + } + + [Fact] + public async Task A_strict_gateway_with_a_small_ceiling_is_still_served() + { + var store = new AvevaStore(OneHertz(T0.AddHours(-1), 3600)); + var fake = store.Client(ceiling: 100); + var times = Enumerable.Range(0, 10).Select(i => T0.AddMinutes(-30).AddSeconds(i * 30 + 0.5)).ToArray(); + var s = await AtTime(fake, times); + s.Select(x => x.Value).ShouldBe(Enumerable.Range(0, 10).Select(i => (object?)(1800 + i * 30 + 0.5))); + s.ShouldAllBe(x => x.StatusCode == GoodInterpolated); + fake.RawCalls.Count(c => c.MaxValues > 100).ShouldBe(1); // learned once, then remembered — a short page is not an error + } + + [Fact] + public async Task Success_is_recorded_once() + { + var fake = new AvevaStore(Gate685()).Client(); + var ds = Source(fake); + await ds.ReadAtTimeAsync("T", [At(1), At(4)], Ct); + var h = ds.GetHealthSnapshot(); + h.TotalQueries.ShouldBe(1); + h.TotalSuccesses.ShouldBe(1); + } + + [Fact] + public async Task A_gateway_fault_is_a_recorded_failure() + { + var fake = new AvevaStore(Gate685()).Client(); + fake.ThrowOnRead = new Grpc.Core.RpcException(new Grpc.Core.Status(Grpc.Core.StatusCode.Unavailable, "down")); + var ds = Source(fake); + await Should.ThrowAsync(() => ds.ReadAtTimeAsync("T", [At(1)], Ct)); + ds.GetHealthSnapshot().TotalFailures.ShouldBe(1); + } + + // ---- long histories (review H1/M1/L1): a generated store, too large to materialise ----------------------------- + + /// + /// A regular series (value = sample index) over [from, to] every , optionally with a + /// hole (gapFrom, gapTo), served in the AVEVA wire shape (boundary row at the query start, ms bounds only, + /// max_values > 0, start <= end, nothing past now + 2 days). + /// + private sealed class SeriesStore(DateTime from, DateTime to, TimeSpan step, DateTime? gapFrom = null, DateTime? gapTo = null) + { + public long RowsStreamed; + + private bool InGap(DateTime t) => gapFrom is { } a && gapTo is { } b && t > a && t < b; + + private DateTime? FirstAtOrAfter(DateTime t) + { + if (t > to) return null; + var k = t <= from ? 0 : ((t - from).Ticks + step.Ticks - 1) / step.Ticks; + var r = from.AddTicks(k * step.Ticks); + if (InGap(r)) return FirstAtOrAfter(gapTo!.Value); + return r <= to ? r : null; + } + + private DateTime? LastBefore(DateTime t) + { + if (t <= from) return null; + var k = ((Min(t, to.AddTicks(1)) - from).Ticks - 1) / step.Ticks; + var r = from.AddTicks(k * step.Ticks); + return InGap(r) ? LastBefore(gapFrom!.Value.AddTicks(1)) : r; + } + + private static DateTime Min(DateTime a, DateTime b) => a <= b ? a : b; + + /// The stored row at , stamped (the boundary row) when given. + private HistorianSample Row(DateTime t, uint quality = 0, DateTime? at = null) => new() + { + Tag = "T", Timestamp = Timestamp.FromDateTime(at ?? t), NumericValue = (t - from).Ticks / step.Ticks, OpcQuality = 192, + Quality = quality, + }; + + public FakeHistorianGatewayClient Client(int? ceiling = null) => new() { RawSource = Serve, MaxValuesCeiling = ceiling }; + + private IEnumerable Serve(DateTime start, DateTime end, int max) + { + if (max <= 0 || end < start || end > Now.AddDays(2) + || start.Ticks % TimeSpan.TicksPerMillisecond != 0 || end.Ticks % TimeSpan.TicksPerMillisecond != 0) + throw new Grpc.Core.RpcException(new Grpc.Core.Status(Grpc.Core.StatusCode.InvalidArgument, "refused")); + var rows = new List(); + if (FirstAtOrAfter(start) != start) + rows.Add(LastBefore(start) is { } p + ? Row(p, 133, start) + : new HistorianSample { Tag = "T", Timestamp = Timestamp.FromDateTime(start), OpcQuality = 0, Quality = 1, NumericValue = 0 }); + for (var t = FirstAtOrAfter(start); t is { } u && u <= end && rows.Count < max; t = FirstAtOrAfter(u.AddTicks(1))) + rows.Add(Row(u)); + RowsStreamed += rows.Count; + return rows; + } + } + + [Theory] // H1: a dense tag that stopped reporting — the walk crosses the silence by bisection, not by regrowing segments + [InlineData(1.0, 30)] + [InlineData(1.0, 365)] + [InlineData(10.0, 30)] + [InlineData(10.0, 365)] + public async Task A_dense_tag_that_stopped_long_ago_is_found_in_a_bounded_number_of_reads(double hz, int stoppedDaysAgo) + { + var step = TimeSpan.FromTicks((long)(TimeSpan.TicksPerSecond / hz)); + var end = Now.AddDays(-stoppedDaysAgo); + var fake = new SeriesStore(end.AddYears(-5), end, step).Client(); + var s = (await AtTime(fake, Now.AddMinutes(-1))).Single(); + s.Value.ShouldBe((double)((end - end.AddYears(-5)).Ticks / step.Ticks)); // the last sample, held + s.StatusCode.ShouldBe(GoodInterpolated); + fake.ReadRawCallCount.ShouldBeLessThan(40); + } + + [Fact] // H1: a time inside a month-long hole in the middle of a dense history + public async Task A_time_inside_a_mid_history_gap_interpolates_across_it() + { + var from = Now.AddYears(-2); + var gapFrom = Now.AddDays(-200); + var gapTo = gapFrom.AddDays(30); + var fake = new SeriesStore(from, Now, TimeSpan.FromSeconds(1), gapFrom, gapTo).Client(); + var t = gapFrom.AddDays(15); + var s = (await AtTime(fake, t)).Single(); + var before = (gapFrom - from).Ticks / TimeSpan.TicksPerSecond; + var after = (gapTo - from).Ticks / TimeSpan.TicksPerSecond; + ((double)s.Value!).ShouldBe(before + (after - before) * 0.5, 1e-6); + s.StatusCode.ShouldBe(GoodInterpolated); + fake.ReadRawCallCount.ShouldBeLessThan(40); + } + + [Fact] // H1: a jump to a time in a gap — the walk starts from the earlier read's last row, far below + public async Task A_jump_into_a_gap_bisects_up_from_the_earlier_reads_last_row() + { + var from = Now.AddDays(-30); + var gapFrom = from.AddDays(2); + var gapTo = gapFrom.AddDays(10); + var fake = new SeriesStore(from, Now, TimeSpan.FromSeconds(1), gapFrom, gapTo).Client(); + var t0 = from.AddSeconds(10.5); + var t1 = gapFrom.AddDays(5); + var s = await AtTime(fake, t0, t1); + s[0].Value.ShouldBe(10.5d); + var before = (gapFrom - from).Ticks / TimeSpan.TicksPerSecond; + var after = (gapTo - from).Ticks / TimeSpan.TicksPerSecond; + ((double)s[1].Value!).ShouldBe(before + (after - before) * 0.5, 1e-6); + fake.ReadRawCallCount.ShouldBeLessThan(40); + } + + [Fact] // M1: widely spread times cost one SMALL read each, not one 1,024-row read each + public async Task A_thousand_times_over_ten_years_cost_one_small_read_each() + { + var store = new SeriesStore(Now.AddYears(-10), Now, TimeSpan.FromSeconds(1)); + var fake = store.Client(); + var rnd = new Random(1); + var times = Enumerable.Range(0, 1000).Select(_ => Now.AddYears(-9).AddTicks((long)(rnd.NextDouble() * TimeSpan.TicksPerDay * 3000))).ToArray(); + var s = await AtTime(fake, times); + s.ShouldAllBe(x => x.StatusCode == GoodInterpolated); + fake.ReadRawCallCount.ShouldBeLessThanOrEqualTo(1010); + store.RowsStreamed.ShouldBeLessThan(80_000); // ~64 rows a read (was ~1,000,000) + } + + [Fact] // L1: a strict-gateway ceiling too small to work with is refused with a message naming it + public async Task A_ceiling_below_a_usable_read_is_refused_naming_the_ceiling() + { + var fake = new SeriesStore(Now.AddDays(-1), Now, TimeSpan.FromSeconds(1)).Client(ceiling: 2); + var ex = await Should.ThrowAsync( + () => Source(fake).ReadAtTimeAsync("T", [Now.AddHours(-1)], Ct)); + ex.Message.ShouldContain("row ceiling (2"); + } + + /// A bursty tag: 1 Hz for the first hour of every day (value = seconds since ). + private sealed class BurstStore(DateTime from, int days) + { + public long RowsStreamed; + private const long Burst = 3600 * TimeSpan.TicksPerSecond; + + private DateTime? FirstAtOrAfter(DateTime t) + { + if (t <= from) return from; + var d = (t - from).Ticks / TimeSpan.TicksPerDay; + var inDay = (t - from).Ticks % TimeSpan.TicksPerDay; + var r = inDay >= Burst + ? from.AddDays(d + 1) + : from.AddTicks(d * TimeSpan.TicksPerDay + (inDay + TimeSpan.TicksPerSecond - 1) / TimeSpan.TicksPerSecond * TimeSpan.TicksPerSecond); + if ((r - from).Ticks % TimeSpan.TicksPerDay >= Burst) r = from.AddDays(d + 1); + return r < from.AddDays(days) ? r : null; + } + + private DateTime? LastBefore(DateTime t) + { + if (t <= from) return null; + var x = (t - from).Ticks - 1; + var d = x / TimeSpan.TicksPerDay; + var inDay = x % TimeSpan.TicksPerDay; + var r = inDay >= Burst - TimeSpan.TicksPerSecond + ? from.AddTicks(d * TimeSpan.TicksPerDay + Burst - TimeSpan.TicksPerSecond) + : from.AddTicks(d * TimeSpan.TicksPerDay + inDay / TimeSpan.TicksPerSecond * TimeSpan.TicksPerSecond); + var last = from.AddDays(days - 1).AddTicks(Burst - TimeSpan.TicksPerSecond); + return r > last ? last : r; + } + + private HistorianSample Row(DateTime t, uint quality = 0, DateTime? at = null) => new() + { + Tag = "T", Timestamp = Timestamp.FromDateTime(at ?? t), NumericValue = (t - from).Ticks / TimeSpan.TicksPerSecond, + OpcQuality = 192, Quality = quality, + }; + + public FakeHistorianGatewayClient Client() => new() { RawSource = Serve }; + + private IEnumerable Serve(DateTime start, DateTime end, int max) + { + if (max <= 0 || end < start || end > Now.AddDays(2) + || start.Ticks % TimeSpan.TicksPerMillisecond != 0 || end.Ticks % TimeSpan.TicksPerMillisecond != 0) + throw new Grpc.Core.RpcException(new Grpc.Core.Status(Grpc.Core.StatusCode.InvalidArgument, "refused")); + var rows = new List(); + if (FirstAtOrAfter(start) != start) + rows.Add(LastBefore(start) is { } p + ? Row(p, 133, start) + : new HistorianSample { Tag = "T", Timestamp = Timestamp.FromDateTime(start), OpcQuality = 0, Quality = 1 }); + for (var t = FirstAtOrAfter(start); t is { } u && u <= end && rows.Count < max; t = FirstAtOrAfter(u.AddTicks(1))) + rows.Add(Row(u)); + RowsStreamed += rows.Count; + return rows; + } + } + + [Theory] // re-review: every time in a gap of a bursty tag costs a jump read + a bisection — the cap must allow that + [InlineData(30)] + [InlineData(100)] + [InlineData(300)] + public async Task A_bursty_tag_asked_at_noon_on_consecutive_days_is_served(int days) + { + var from = Now.Date.AddDays(-days - 1); + var fake = new BurstStore(from, days + 1).Client(); + var times = Enumerable.Range(0, days).Select(d => from.AddDays(d).AddHours(12)).ToArray(); + var s = await AtTime(fake, times); + for (var d = 0; d < days; d++) + { + double prev = d * 86400 + 3599, next = (d + 1) * 86400; + ((double)s[d].Value!).ShouldBe(prev + (next - prev) * (12 * 3600 - 3599.0) / (86400 - 3599.0), 1e-6); + s[d].StatusCode.ShouldBe(GoodInterpolated); + } + fake.ReadRawCallCount.ShouldBeLessThanOrEqualTo(7 * days); // ≈ 6 a time; the cap is 64 + 16 a time + } + + [Fact] // L1: a small ceiling is served (paging), within the read cap + public async Task A_small_ceiling_is_served_within_the_read_cap() + { + var fake = new SeriesStore(Now.AddYears(-1), Now, TimeSpan.FromSeconds(1)).Client(ceiling: 10); + var times = Enumerable.Range(0, 50).Select(i => Now.AddDays(-300 + i * 5).AddSeconds(0.5)).ToArray(); + var s = await AtTime(fake, times); + s.ShouldAllBe(x => x.StatusCode == GoodInterpolated); + fake.ReadRawCallCount.ShouldBeLessThanOrEqualTo(GatewayAtTimeReader.BaseReads + GatewayAtTimeReader.ReadsPerTime * 50); + } + + [Fact] // L1: the request's read cap stops a walk with a clear error, never a hang (here a cap of 3 vs. a 5-read walk) + public async Task The_read_cap_stops_a_request_with_a_clear_error() + { + var fake = new AvevaStore([(T0.AddDays(-400), 42, false), (T0.AddDays(-400).AddSeconds(10), 43, false)]).Client(); + var ex = await Should.ThrowAsync(() => GatewayAtTimeReader.ReadAsync( + fake, new GatewayRowCeiling(), "T", [T0], Now, Ct, maxReads: 3)); + ex.Message.ShouldContain("needed more than 3 raw reads"); + fake.ReadRawCallCount.ShouldBe(3); + fake.RawStreamsCompleted.ShouldBe(3); + } + + [Fact] // the default cap scales with the request: 64 + 16 per distinct time (a runaway guard; a bursty tag needs ≈ 6) + public async Task The_default_read_cap_is_64_plus_4_per_distinct_time() + { + (GatewayAtTimeReader.BaseReads, GatewayAtTimeReader.ReadsPerTime).ShouldBe((64, 16)); + await Task.CompletedTask; + } + + // ---- type hint edge cases (review L2/L3) ------------------------------------------------------------------------ + + [Fact] + public async Task A_DateTime_node_is_stepped() + { + var fake = new AvevaStore([(At(0), 0, false), (At(10), 10, false)]).Client(); + var s = (await AtTimeTyped(fake, TypeCode.DateTime, At(5))).Single(); + s.Value.ShouldBe(0d); + s.StatusCode.ShouldBe(GoodInterpolated); + } + + [Theory] + [InlineData(double.PositiveInfinity)] + [InlineData(double.NaN)] + [InlineData(1e30)] + public async Task Under_an_integer_hint_a_non_integer_next_value_is_unusable_not_saturated(double next) + { + var fake = new AvevaStore([(At(0), 100, false), (At(10), next, false)]).Client(); + var s = (await AtTimeTyped(fake, TypeCode.Int32, At(5))).Single(); + s.Value.ShouldBeOfType().ShouldBe(100); // held before an unusable neighbour + s.StatusCode.ShouldBe(UncertainInterpolated); + } + + // ---- kill switch ----------------------------------------------------------------------------------------------- + + [Fact] + public async Task The_kill_switch_restores_AVEVAs_own_answer() + { + var fake = new AvevaStore(Gate685()).Client(); + fake.AtTimeSamples = [new HistorianSample { NumericValue = 107, OpcQuality = 192, Timestamp = Timestamp.FromDateTime(At(3)) }]; + var ds = new GatewayHistorianDataSource(fake, NullLogger.Instance, atTimeFromRaw: false); + var s = (await ds.ReadAtTimeAsync("T", [At(3)], Ct)).Samples.Single(); + s.Value.ShouldBe(107d); + fake.ReadAtTimeCallCount.ShouldBe(1); + fake.ReadRawCallCount.ShouldBe(0); + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayHistorianDataSourceTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayHistorianDataSourceTests.cs index 0920a0d4..e5406dc1 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayHistorianDataSourceTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayHistorianDataSourceTests.cs @@ -106,7 +106,7 @@ public sealed class GatewayHistorianDataSourceTests var t0 = new DateTime(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc); var t1 = t0.AddSeconds(1); fake.AtTimeSamples = new[] { new HistorianSample { NumericValue = 5.0, OpcQuality = 192, Timestamp = Timestamp.FromDateTime(t0) } }; - var ds = new GatewayHistorianDataSource(fake, NullLogger.Instance); + var ds = new GatewayHistorianDataSource(fake, NullLogger.Instance, atTimeFromRaw: false); // #693 kill switch var r = await ds.ReadAtTimeAsync("T", new[] { t0, t1 }, TestContext.Current.CancellationToken); Assert.Equal(2, r.Samples.Count); // exactly one per requested ts, in order Assert.Equal(5.0, r.Samples[0].Value); @@ -122,7 +122,7 @@ public sealed class GatewayHistorianDataSourceTests { new HistorianSample { NumericValue = 0, OpcQuality = 0, Quality = 1, Timestamp = Timestamp.FromDateTime(t0) }, }; - var ds = new GatewayHistorianDataSource(fake, NullLogger.Instance); + var ds = new GatewayHistorianDataSource(fake, NullLogger.Instance, atTimeFromRaw: false); // #693 kill switch var r = await ds.ReadAtTimeAsync("T", new[] { t0 }, TestContext.Current.CancellationToken); Assert.Null(r.Samples[0].Value); Assert.Equal(0x80000000u, r.Samples[0].StatusCode); diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/HistorianGatewayClientAdapterTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/HistorianGatewayClientAdapterTests.cs index 55719278..7c8bdd60 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/HistorianGatewayClientAdapterTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/HistorianGatewayClientAdapterTests.cs @@ -117,4 +117,18 @@ public sealed class HistorianGatewayClientAdapterTests var dataSource = GatewayHistorian.CreateDataSource(opts, services); Assert.IsType(dataSource); } + + [Theory] // #693: the ServerHistorian:GatewayAtTimeFromRaw kill switch reaches the data source (default on) + [InlineData(null, true)] + [InlineData(true, true)] + [InlineData(false, false)] + public void Factory_passes_the_AtTime_kill_switch(bool? configured, bool expected) + { + var opts = configured is { } c + ? new ServerHistorianOptions { Enabled = true, Endpoint = "https://localhost:5222", ApiKey = "histgw_x_y", GatewayAtTimeFromRaw = c } + : new ServerHistorianOptions { Enabled = true, Endpoint = "https://localhost:5222", ApiKey = "histgw_x_y" }; + using var services = new ServiceCollection().BuildServiceProvider(); + var dataSource = Assert.IsType(GatewayHistorian.CreateDataSource(opts, services)); + dataSource.AtTimeFromRaw.ShouldBe(expected); + } } diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests/AtTimeEvaluatorTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests/AtTimeEvaluatorTests.cs index 70dc5a96..deaaceca 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests/AtTimeEvaluatorTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests/AtTimeEvaluatorTests.cs @@ -83,6 +83,11 @@ public sealed class AtTimeEvaluatorTests var n = S(10, HistorianValue.FromInt64(15), Good); AtTimeEvaluator.Evaluate(At(5), p, n, Now, Now).Value.ShouldBe(13L); // 12.5 ⇒ 13 AtTimeEvaluator.Evaluate(At(2), p, n, Now, Now).Value.ShouldBe(11L); + // #693: ShouldBe compares numbers across types, so the type needs its own assertion — a C# conditional + // `long : double` boxed a Double here, and the line above still passed. + AtTimeEvaluator.Evaluate(At(5), p, n, Now, Now).Value.ShouldBeOfType(); + AtTimeEvaluator.Evaluate(At(5), S(0, HistorianValue.FromDouble(10), Good), S(10, HistorianValue.FromDouble(15), Good), Now, Now) + .Value.ShouldBeOfType(); } [Fact] diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadTests.cs index 0ecffbe1..0d314f6d 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadTests.cs @@ -586,6 +586,78 @@ public sealed class NodeManagerHistoryReadTests : IDisposable await host.DisposeAsync(); } + /// + /// #693: AtTime passes each node's scalar value type to the source (the Gateway needs it — its stored values carry no + /// type), through a Raw NodeId and a UNS NodeId alike; an array node passes none. (The fake above implements only the + /// hint-less overload, and proves the interface default forwards.) + /// + [Fact] + public async Task AtTime_passes_the_nodes_value_type_for_Raw_and_UNS_nodes() + { + var (host, server) = await BootAsync(); + var nm = server.NodeManager!; + var fake = new HintRecordingSource(); + nm.HistorianDataSource = fake; + + nm.EnsureVariable("Plant/Modbus/dev1/Flag", parentFolderNodeId: null, displayName: "Flag", dataType: "Boolean", + writable: false, historianTagname: "R.Flag", realm: AddressSpaceRealm.Raw); + nm.EnsureVariable("Area/Line/Eq/Flag", parentFolderNodeId: null, displayName: "Flag", dataType: "Boolean", + writable: false, historianTagname: "U.Flag", realm: AddressSpaceRealm.Uns); + nm.EnsureVariable("Plant/Modbus/dev1/Count", parentFolderNodeId: null, displayName: "Count", dataType: "Int32", + writable: false, historianTagname: "R.Count", realm: AddressSpaceRealm.Raw); + nm.EnsureVariable("Area/Line/Eq/Flow", parentFolderNodeId: null, displayName: "Flow", dataType: "Double", + writable: false, historianTagname: "U.Flow", realm: AddressSpaceRealm.Uns); + nm.EnsureVariable("Plant/Modbus/dev1/Arr", parentFolderNodeId: null, displayName: "Arr", dataType: "Int16", + writable: false, historianTagname: "R.Arr", realm: AddressSpaceRealm.Raw, isArray: true, arrayLength: 4); + + string[] paths = ["Plant/Modbus/dev1/Flag", "Area/Line/Eq/Flag", "Plant/Modbus/dev1/Count", "Area/Line/Eq/Flow", "Plant/Modbus/dev1/Arr"]; + AddressSpaceRealm[] realms = [AddressSpaceRealm.Raw, AddressSpaceRealm.Uns, AddressSpaceRealm.Raw, AddressSpaceRealm.Uns, AddressSpaceRealm.Raw]; + var nodeIds = paths.Select((p, i) => nm.TryGetVariable(p, realms[i])!.NodeId).ToArray(); + + var (_, errors) = InvokeHistoryRead(server, nm, + new ReadAtTimeDetails { ReqTimes = new DateTimeCollection { DateTime.UtcNow.AddMinutes(-1) } }, nodeIds); + + errors.ShouldAllBe(e => StatusCode.IsGood(e.StatusCode)); + fake.Hints["R.Flag"].ShouldBe(TypeCode.Boolean); + fake.Hints["U.Flag"].ShouldBe(TypeCode.Boolean); + fake.Hints["R.Count"].ShouldBe(TypeCode.Int32); + fake.Hints["U.Flow"].ShouldBe(TypeCode.Double); + fake.Hints["R.Arr"].ShouldBeNull(); + + await host.DisposeAsync(); + } + + /// A source recording the AtTime value-type hint per tagname (the hinted overload only). + private sealed class HintRecordingSource : IHistorianDataSource + { + public System.Collections.Concurrent.ConcurrentDictionary Hints { get; } = new(); + + public Task ReadAtTimeAsync( + string fullReference, IReadOnlyList timestampsUtc, TypeCode? valueType, CancellationToken cancellationToken) + { + Hints[fullReference] = valueType; + return Task.FromResult(Empty()); + } + + public Task ReadAtTimeAsync( + string fullReference, IReadOnlyList timestampsUtc, CancellationToken cancellationToken) => + throw new InvalidOperationException("the node manager must call the hinted overload"); + + public Task ReadRawAsync(string fullReference, DateTime startUtc, DateTime endUtc, uint maxValuesPerNode, + CancellationToken cancellationToken) => Task.FromResult(Empty()); + + public Task ReadProcessedAsync(string fullReference, DateTime startUtc, DateTime endUtc, TimeSpan interval, + HistoryAggregateType aggregate, CancellationToken cancellationToken) => Task.FromResult(Empty()); + + 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() { } + } + /// /// Two nodes in a single Raw HistoryRead batch — one backend succeeds, one throws. Proves that /// per-node error isolation is enforced across handle indices: the good node's slot is