From 44a03bedc029d1af1b0f23cf5fa34df9528d1dd7 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Mon, 12 Oct 2026 14:31:31 -0400 Subject: [PATCH 1/5] =?UTF-8?q?fix(historian):=20#685=20Gateway=20native?= =?UTF-8?q?=20TimeAverage/Total=20=E2=80=94=20drop=20AVEVA's=20pre-window?= =?UTF-8?q?=20row,=20keep=20Uncertain=20values?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit AVEVA answers one extra row for the interval before the window; it was mapped as a result stamped before StartTime (N+1 values). Rows whose interval lies outside the requested span (either direction) are now dropped after the drained stream. An Uncertain (opc 64-191) interval kept no value (opc >= 192 was the only data test); it now carries its value as UncertainDataSubNormal | Calculated. Docs: AVEVA's TWA excludes NULL-gap time while the SQL backends bridge it; end-to-end per-interval values also need historiangw#27. --- CLAUDE.md | 2 +- docs/Historian.md | 33 +++- .../GatewayHistorianDataSource.cs | 6 +- .../Mapping/SampleMapper.cs | 74 ++++++-- .../GatewayNativeAggregateRowsTests.cs | 158 ++++++++++++++++++ 5 files changed, 249 insertions(+), 24 deletions(-) create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayNativeAggregateRowsTests.cs diff --git a/CLAUDE.md b/CLAUDE.md index eb39bbc5..eda1de8e 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -1365,7 +1365,7 @@ historian tagname, which is the tag's driver `FullName`. The server dispatches all OPC UA HistoryRead to the registered `IHistorianDataSource` — the `GatewayHistorianDataSource` read client when enabled, else the `NullHistorianDataSource` default (historized nodes return `GoodNoData` for Raw/AtTime — Processed is `BadAggregateNotSupported` since #682). Supported variants: Raw, Processed -(backend-dependent: SqlServer/QuestDb serve seventeen; Gateway twelve = `GatewayHistorianDataSource.ServedAggregates` — ⚠️ #681, client-visible: the ten per-sample ones (Average, Count, Min/Max, Range, Start/End/Delta, Min/MaxActualTime) are evaluated from RAW gateway reads by the shared `Sql.Common` evaluator — identical to the SQL backends, Calculated bits, interval-start stamps, #677 reverse; Min/Max are no longer AVEVA modes, the read moves every raw sample of the window (a month of 1 Hz ⇒ `BadTooManyOperations` at the default `MaxTimeWeightedRawRows` cap), and every gateway stream is consumed to completion (an abandoned one makes the gateway evict its shared historian session) — plus TimeAverage/Total via AVEVA modes (now interval-start stamped — pending the live leg — + Calculated); the other time-weighted five stay `BadAggregateNotSupported` because the gateway has no prior-sample read for Part 13 bounds; `docs/Historian.md` §Aggregate set), and AtTime over historized variable nodes; Events over +(backend-dependent: SqlServer/QuestDb serve seventeen; Gateway twelve = `GatewayHistorianDataSource.ServedAggregates` — ⚠️ #681, client-visible: the ten per-sample ones (Average, Count, Min/Max, Range, Start/End/Delta, Min/MaxActualTime) are evaluated from RAW gateway reads by the shared `Sql.Common` evaluator — identical to the SQL backends, Calculated bits, interval-start stamps, #677 reverse; Min/Max are no longer AVEVA modes, the read moves every raw sample of the window (a month of 1 Hz ⇒ `BadTooManyOperations` at the default `MaxTimeWeightedRawRows` cap), and every gateway stream is consumed to completion (an abandoned one makes the gateway evict its shared historian session) — plus TimeAverage/Total via AVEVA modes (interval-start stamped + Calculated; ⚠️ #685: AVEVA's extra pre-window row is dropped ⇒ one row per interval in `[Start, End)`, and an Uncertain (opc 64) interval keeps its value as `UncertainDataSubNormal` — AVEVA's TWA excludes NULL-gap time where the SQL backends bridge it; per-interval values still need gateway fix historiangw#27); the other time-weighted five stay `BadAggregateNotSupported` because the gateway has no prior-sample read for Part 13 bounds; `docs/Historian.md` §Aggregate set), and AtTime over historized variable nodes; Events over alarm-owning equipment-folder event-notifier nodes. Reads are ungated (served from any redundancy node); authorization uses the standard `AccessLevels.HistoryRead` bit set at materialization. diff --git a/docs/Historian.md b/docs/Historian.md index c780be2e..dfe74f86 100644 --- a/docs/Historian.md +++ b/docs/Historian.md @@ -795,17 +795,40 @@ The OPC UA `Total` aggregate is **supported** over the HistorianGateway backend. native **`Integral`** retrieval mode, so `Total` maps straight to it (`HistoryAggregateType.Total → RetrievalMode.Integral`) — no client-side scaling. (This replaces the retired Wonderware path, which had no `Total` column and derived it client-side as time-weighted `Average × interval-seconds`.) Values pass through unchanged; a Good -bucket is `Good | Calculated` (`0x00000401`) and an empty / null bucket `BadNoData | Calculated` (#681), stamped with its interval -start (see below). +bucket is `Good | Calculated` (`0x00000401`), an Uncertain one `UncertainDataSubNormal | Calculated` (`0x40A40401`) **with its +value** (#685), and an empty / null bucket `BadNoData | Calculated` (#681), stamped with its interval start (see below). + +**Native `TimeAverage` / `Total` rows (#685 — client-visible).** Two mapping defects found by the live AVEVA gate (2026-10-12) are +fixed: + +- **One result per interval in `[Start, End)`.** AVEVA always answers one extra row for the interval *before* the window + (`[start − interval, start)` — a real value when data exists there, a no-data row otherwise). It used to be passed through, so a + client got N + 1 values, the first stamped before its StartTime. Rows whose interval start lies outside the requested span are now + dropped (a row with no start time is placed by its end time, `(lo, hi]`); the span is taken in either order, so the filter does + not depend on the read direction. The stream is still drained in full and filtered afterwards — never broken out of. +- **An Uncertain interval keeps its value.** AVEVA marks an interval that holds a NULL/Bad sample `opc_quality = 64` (Uncertain) and + still supplies its value. It used to be mapped `BadNoData` with the value discarded (`opc_quality >= 192` was the only "has data" + test); every Uncertain-family quality (64–191, low byte) now returns the value as `UncertainDataSubNormal | Calculated` — the + sub-code the SQL backends give an interval that includes non-good data. Only a Bad-family quality (< 64) is `BadNoData`, null. +- ⚠️ **AVEVA's semantics differ from the SQL backends' across a NULL gap.** AVEVA's `TimeWeightedAverage` holds the previous value + up to the NULL sample and **excludes the gap time from both numerator and denominator** (live: an interval with 10 good seconds + of (5 → 1) answers 3, `Integral` 30); the `SqlServer`/`QuestDb` evaluator bridges across the outage (Part 13 interpolated + bounding — 6.08 Uncertain for the same samples, see [Time-weighted aggregates](#time-weighted-aggregates-part-13-671)). + Extrapolation past the last sample is `Good` on AVEVA and `Uncertain` on the SQL backends. On all-Good spans the magnitudes + match exactly (both sloped/trapezoidal). +- ⚠️ **End-to-end correctness also needs the gateway fix historiangw#27:** the gateway sends the aggregate interval to AVEVA in + ticks where AVEVA reads milliseconds (10,000× too large), so today a native read returns one row for the whole window (plus the + pre-window row, now dropped). OtOpcUa does not work around it; per-interval values return once the gateway is fixed. **Integral unit (#677).** AVEVA's integral is `∫ value dt` (seconds) ÷ the tag's `IntegralDivisor`. The proto3 `double` default for `HistorianTagDefinition.IntegralDivisor` is `0`, so `GatewayTagProvisioner` now sends **`IntegralDivisor = 1`** explicitly on every `EnsureTags` — the value·seconds unit the SQL backends' `Total` uses. A tag provisioned earlier keeps its divisor until its next -`EnsureTags` (an upsert). The live magnitude check against AVEVA (a constant 1 integrated over 1 s reads 1) is still open — it needs the -VPN. Reverse processed reads of the AVEVA-native `TimeAverage`/`Total` are not re-anchored by #677 (the gateway refuses `end < start`); +`EnsureTags` (an upsert). The live magnitude check against AVEVA (a constant 1 integrated over 1 s reads 1) passed on 2026-10-12 at the AVEVA +level; end to end it waits on historiangw#27 (above). Reverse processed reads of the AVEVA-native `TimeAverage`/`Total` are not re-anchored by #677 (the gateway refuses `end < start`); the raw-evaluated per-sample aggregates (#681) do follow #677. Since #681 every native bucket is stamped with its interval **start** (Part 13; it was the interval end) and carries the Calculated info bits (`Good` ⇒ `0x00000401`, no data ⇒ `BadNoData | 0x401`). -⚠️ **The interval-start stamp is pending the live leg:** it reads the aggregate row's `start_time`, which histsdk decodes from a row +The live AVEVA gate (2026-10-12) confirmed `start_time` is the interval start (and the `value·s` unit). Previously noted as +pending: it reads the aggregate row's `start_time`, which histsdk decodes from a row field (`tail + 43`) its own reverse-engineering notes still mark unverified (HistorianGateway `docs/plans/2026-07-13-archreview2-D-histsdk-residue-design.md`); if that field proves wrong on live AVEVA the stamp is wrong with it. 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 f84df290..e62567b5 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 @@ -136,7 +136,8 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis /// /// /// TimeAverage / Total stay native AVEVA retrieval modes (), stamped with - /// the interval start and the Calculated bit. Everything else is + /// the interval start and the Calculated bit; AVEVA's extra row for the interval before the window is dropped and + /// an Uncertain interval keeps its value (#685, ). Everything else is /// (see ). /// /// @@ -165,7 +166,8 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis buckets.Add(bucket); } - var snapshots = SampleMapper.ToAggregateSnapshots(buckets); + // Filtered while mapping, after the stream was drained (#685: AVEVA's extra pre-window row). + var snapshots = SampleMapper.ToAggregateSnapshots(buckets, startUtc, endUtc); RecordOutcome(success: true, error: null); return new HistoryReadResult(snapshots, ContinuationPoint: null); } diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/Mapping/SampleMapper.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/Mapping/SampleMapper.cs index 232b453b..1e090edd 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/Mapping/SampleMapper.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/Mapping/SampleMapper.cs @@ -17,6 +17,9 @@ internal static class SampleMapper /// OPC DA "Good" family floor — a quality byte at/above this carries usable data. private const byte GoodQualityFloor = 192; + /// OPC DA "Uncertain" family floor (64–191). + private const byte UncertainQualityFloor = 64; + /// Maps a single raw sample to a value snapshot. /// The gateway raw sample. /// The driver-agnostic snapshot. @@ -66,38 +69,77 @@ internal static class SampleMapper /// The driver-agnostic snapshot. /// /// - /// Unlike the legacy Wonderware DTO (a nullable double?), the gateway proto carries a - /// non-optional double value, so an unavailable (no-data) bucket cannot be signalled by a - /// null value. Instead it is signalled by a non-Good opc_quality: a Good bucket - /// (opc_quality >= 192) yields its value with a Good status, anything else maps to - /// BadNoData with a null value — preserving the Wonderware aggregate contract (binary - /// Good-with-value / BadNoData-null). + /// The gateway proto carries a non-optional double value, so an unavailable (no-data) bucket cannot be signalled + /// by a null value; it is signalled by the OPC DA quality family of opc_quality (its low byte, as the raw mapping + /// reads it): /// + /// + /// Good (≥ 192) ⇒ the value, Good. + /// Uncertain (64–191) ⇒ the value, UncertainDataSubNormal (#685): AVEVA marks an interval that holds a + /// NULL/Bad sample opc_quality = 64 and still answers its value (computed over the interval's good time only — + /// see docs/Historian.md). It used to map to BadNoData with the value discarded. The sub-code is the one the SQL + /// backends' aggregates give an interval that includes non-good data. + /// Bad (< 64) ⇒ BadNoData, null value (the wire's 0 is not a value). + /// /// - /// #681: both statuses carry the Calculated info bits (0x401) like every other backend's processed value, and + /// #681: every status carries the Calculated info bits (0x401) like every other backend's processed value, and /// the bucket is stamped with its interval START (Part 13; it was the interval end), falling back to the end only when - /// the start is absent. Pending the live leg: histsdk decodes start_time from a row field (tail + 43) its - /// reverse-engineering notes still mark unverified. + /// the start is absent (verified on live AVEVA 2026-10-12: start_time is the interval start). /// /// public static DataValueSnapshot ToAggregateSnapshot(HistorianAggregateSample aggregate) { - bool hasData = aggregate.OpcQuality >= GoodQualityFloor; + var q = (byte)aggregate.OpcQuality; + var status = q >= GoodQualityFloor ? StatusGood + : q >= UncertainQualityFloor ? Sql.Common.HistorianStatusCodes.UncertainDataSubNormal + : StatusBadNoData; return new DataValueSnapshot( - Value: hasData ? aggregate.Value : null, // boxes as System.Double when present - StatusCode: HistorianInfoBits.WithCalculated(hasData ? StatusGood : StatusBadNoData), + Value: status == StatusBadNoData ? null : aggregate.Value, // boxes as System.Double when present + StatusCode: HistorianInfoBits.WithCalculated(status), SourceTimestampUtc: (aggregate.StartTime ?? aggregate.EndTime)?.ToDateTime(), // bucket timestamp = interval start ServerTimestampUtc: DateTime.UtcNow); } - /// Maps a batch of aggregate buckets to value snapshots, in order. + /// + /// True when the aggregate row's interval lies in the requested window — the span between + /// and in either order (a reverse request covers the same span). AVEVA answers one extra row + /// for the interval BEFORE the window ([start − interval, start), #685), which Part 11/13 has no place for: an + /// interval belongs to the window when its start lies in [lo, hi); a row without a start time (the mapper then + /// stamps it with its end) belongs when its end lies in (lo, hi]. + /// + /// The gateway aggregate sample. + /// The request's start time. + /// The request's end time. + /// Whether the row is one of the window's intervals. + public static bool IsInWindow(HistorianAggregateSample aggregate, DateTime startUtc, DateTime endUtc) + { + var lo = startUtc <= endUtc ? startUtc : endUtc; + var hi = startUtc <= endUtc ? endUtc : startUtc; + if (aggregate.StartTime is { } s) + { + var t = s.ToDateTime(); + return t >= lo && t < hi; + } + if (aggregate.EndTime is { } e) + { + var t = e.ToDateTime(); + return t > lo && t <= hi; + } + return false; + } + + /// Maps the aggregate buckets of a window to value snapshots, in order, dropping rows outside it (#685). /// The gateway aggregate samples. - /// The driver-agnostic snapshots. - public static IReadOnlyList ToAggregateSnapshots(IEnumerable aggregates) + /// The request's start time. + /// The request's end time. + /// The driver-agnostic snapshots, one per in-window interval row. + public static IReadOnlyList ToAggregateSnapshots(IEnumerable aggregates, + DateTime startUtc, DateTime endUtc) { var result = new List(); foreach (var aggregate in aggregates) - result.Add(ToAggregateSnapshot(aggregate)); + if (IsInWindow(aggregate, startUtc, endUtc)) + result.Add(ToAggregateSnapshot(aggregate)); return result; } } diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayNativeAggregateRowsTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayNativeAggregateRowsTests.cs new file mode 100644 index 00000000..34abda80 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayNativeAggregateRowsTests.cs @@ -0,0 +1,158 @@ +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; + +/// +/// #685: the AVEVA-native TimeAverage / Total rows. The fake emits the REAL wire shape captured by the live +/// AVEVA gate (2026-10-12): AVEVA always adds one row for the interval BEFORE the window ([start − interval, start), +/// carrying a real value when data exists there, a no-data row otherwise — and, under historiangw#27, starting days +/// early), and marks an interval holding a NULL/Bad sample opc_quality = 64 while still supplying its value. +/// Every row carries a value on the wire (proto3 non-optional double; a no-data row is 0 + a Bad OPC quality). +/// +public sealed class GatewayNativeAggregateRowsTests +{ + private static readonly CancellationToken Ct = CancellationToken.None; + + /// 2026-10-12 18:06:00 UTC — the live gate's tag HistGW.LiveTest.Gate681. + private static readonly DateTime T0 = new(2026, 10, 12, 18, 6, 0, DateTimeKind.Utc); + + private static readonly TimeSpan I20 = TimeSpan.FromSeconds(20); + + private const uint GoodCalc = 0x00000401u; + private const uint UncertainCalc = 0x40A40401u; // UncertainDataSubNormal + Calculated, as the SQL backends' aggregates + private const uint NoDataCalc = 0x809B0401u; + + private static HistorianAggregateSample Row(DateTime? start, DateTime? end, double value, uint opc) => new() + { + Tag = "T", + StartTime = start is { } s ? Timestamp.FromDateTime(s) : null, + EndTime = end is { } e ? Timestamp.FromDateTime(e) : null, + Value = value, + OpcQuality = opc, + }; + + private static async Task> Read(IReadOnlyList wire, + DateTime start, DateTime end, HistoryAggregateType aggregate = HistoryAggregateType.TimeAverage) + { + var fake = new FakeHistorianGatewayClient { AggregateSamples = wire }; + var ds = new GatewayHistorianDataSource(fake, NullLogger.Instance); + return (await ds.ReadProcessedAsync("T", start, end, I20, aggregate, Ct)).Samples; + } + + // ---- D1: rows outside the requested window ----------------------------------------------------------------------- + + [Fact] + public async Task The_leading_pre_window_row_carrying_a_real_value_is_dropped() + { + // Live: TimeAverage over [18:06:20, 18:07:20) returned "18:06:00 22.5 Good" first — the interval before the window. + var start = T0.AddSeconds(20); + var wire = new[] + { + Row(T0, start, 22.5, 192), + Row(start, start + I20, 30, 192), + Row(start + I20, start + 2 * I20, 3, 64), + Row(start + 2 * I20, start + 3 * I20, 1, 192), + }; + var s = await Read(wire, start, start + 3 * I20); + s.Select(x => x.SourceTimestampUtc).ShouldBe(new DateTime?[] { start, start + I20, start + 2 * I20 }); + s[0].Value.ShouldBe(30d); + } + + [Fact] + public async Task A_no_data_pre_window_row_is_dropped_rather_than_answered_BadNoData() + { + var wire = new[] { Row(T0 - I20, T0, 0, 0), Row(T0, T0 + I20, 22.5, 192) }; + var s = await Read(wire, T0, T0 + I20, HistoryAggregateType.Total); + s.Count.ShouldBe(1); + s[0].SourceTimestampUtc.ShouldBe(T0); + s[0].StatusCode.ShouldBe(GoodCalc); + } + + [Fact] + public async Task A_pre_window_row_starting_days_early_is_dropped() + { + // historiangw#27: the interval reaches AVEVA 10,000× too large, so the leading row starts days before the window. + var wire = new[] { Row(T0.AddSeconds(-200_000), T0, 0, 0), Row(T0, T0.AddMinutes(2), 11.727, 64) }; + var s = await Read(wire, T0, T0.AddMinutes(2)); + s.Count.ShouldBe(1); + s[0].SourceTimestampUtc.ShouldBe(T0); + } + + [Fact] + public async Task A_row_whose_interval_starts_at_or_after_the_window_end_is_dropped() + { + var wire = new[] { Row(T0, T0 + I20, 1, 192), Row(T0 + I20, T0 + 2 * I20, 2, 192) }; + var s = await Read(wire, T0, T0 + I20); + s.Count.ShouldBe(1); + s[0].Value.ShouldBe(1d); + } + + [Fact] + public async Task Without_a_start_time_a_row_is_placed_by_its_end_time() + { + // Mapper fallback: no StartTime ⇒ stamped with its EndTime. The pre-window row then ENDS at the window start (out); + // an in-window row ends inside (lo, hi] (in). + var wire = new[] { Row(null, T0, 5, 192), Row(null, T0 + I20, 6, 192), Row(null, T0 + 2 * I20, 7, 192) }; + var s = await Read(wire, T0, T0 + 2 * I20); + s.Select(x => x.Value).ShouldBe(new object?[] { 6d, 7d }); + } + + [Fact] + public async Task The_window_filter_is_direction_agnostic() + { + // A reverse request (start > end) covers the same span (end, start]; a row for the interval before the span is out. + var wire = new[] { Row(T0 - I20, T0, 9, 192), Row(T0, T0 + I20, 1, 192), Row(T0 + I20, T0 + 2 * I20, 2, 192) }; + var s = await Read(wire, T0 + 2 * I20, T0); + s.Select(x => x.Value).ShouldBe(new object?[] { 1d, 2d }); + } + + // ---- D2: an Uncertain row keeps its value ------------------------------------------------------------------------ + + [Fact] + public async Task An_Uncertain_row_keeps_its_value_with_an_Uncertain_Calculated_status() + { + // Live: interval 18:06:40 holds the NULL sample ⇒ AVEVA TWA 3 / Integral 30 with opc_quality 64. + var at = T0.AddSeconds(40); + var s = await Read([Row(at, at + I20, 3, 64)], at, at + I20); + s.Single().Value.ShouldBe(3d); + s.Single().StatusCode.ShouldBe(UncertainCalc); + } + + [Theory] + [InlineData(64u)] + [InlineData(68u)] + [InlineData(191u)] + [InlineData(0x1_0040u)] // only the low byte is the OPC DA quality (as the raw mapping reads it) + public async Task Every_Uncertain_family_quality_keeps_its_value(uint opc) + { + var s = await Read([Row(T0, T0 + I20, 4.5, opc)], T0, T0 + I20); + s.Single().Value.ShouldBe(4.5d); + s.Single().StatusCode.ShouldBe(UncertainCalc); + } + + [Theory] + [InlineData(0u)] + [InlineData(1u)] + [InlineData(63u)] + public async Task A_Bad_family_row_is_BadNoData_with_no_value_although_the_wire_carries_one(uint opc) + { + var s = await Read([Row(T0, T0 + I20, 0, opc)], T0, T0 + I20); + s.Single().Value.ShouldBeNull(); + s.Single().StatusCode.ShouldBe(NoDataCalc); + } + + [Theory] + [InlineData(192u)] + [InlineData(216u)] + public async Task A_Good_family_row_is_Good_Calculated(uint opc) + { + var s = await Read([Row(T0, T0 + I20, 2.5, opc)], T0, T0 + I20); + s.Single().Value.ShouldBe(2.5d); + s.Single().StatusCode.ShouldBe(GoodCalc); + } +} From 9e47532283ed3a29bc4011c86d46cb6b0a112e47 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Mon, 12 Oct 2026 14:35:20 -0400 Subject: [PATCH 2/5] fix(historian): #687 Gateway raw reads drop AVEVA's synthetic start-boundary row MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An AVEVA Full raw read leads with a row stamped at the query start (the value in effect, Quality=133, or a no-data NULL row); it was returned as a stored sample. Raw reads now use the #681 classification (RawAggregateReader.RequestStartFor / IsWindowSample — one authority, by timestamp, since a stored NULL looks the same on the wire): request from 1 ms early with NumValuesPerNode + 16 rows, keep rows at or after StartTime, drain the whole stream. A full page still holds exactly the cap, so #669 paging/resume/tie-cluster reads are unchanged; a margin that swallows the allowance fails the node rather than ending paging early. ReturnBounds is not plumbed on any backend (documented). --- CLAUDE.md | 2 +- docs/Historian.md | 28 +++ .../GatewayHistorianDataSource.cs | 75 ++++++- .../RawAggregateReader.cs | 27 ++- .../GatewayHistorianDataSourceTests.cs | 6 +- .../GatewayRawBoundaryRowTests.cs | 209 ++++++++++++++++++ 6 files changed, 332 insertions(+), 15 deletions(-) create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayRawBoundaryRowTests.cs diff --git a/CLAUDE.md b/CLAUDE.md index eda1de8e..96dafeef 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -1364,7 +1364,7 @@ historian tagname, which is the tag's driver `FullName`. The server dispatches all OPC UA HistoryRead to the registered `IHistorianDataSource` — the `GatewayHistorianDataSource` read client when enabled, else the `NullHistorianDataSource` default -(historized nodes return `GoodNoData` for Raw/AtTime — Processed is `BadAggregateNotSupported` since #682). Supported variants: Raw, Processed +(historized nodes return `GoodNoData` for Raw/AtTime — Processed is `BadAggregateNotSupported` since #682). Supported variants: Raw (⚠️ #687, client-visible: Gateway raw reads no longer return AVEVA's synthetic start-boundary row — asked from 1 ms early with `NumValuesPerNode + 16` rows and filtered by timestamp, the #681 classification — so an empty window is `GoodNoData`, not a fabricated value at StartTime; `ReturnBounds` is ignored on every backend), Processed (backend-dependent: SqlServer/QuestDb serve seventeen; Gateway twelve = `GatewayHistorianDataSource.ServedAggregates` — ⚠️ #681, client-visible: the ten per-sample ones (Average, Count, Min/Max, Range, Start/End/Delta, Min/MaxActualTime) are evaluated from RAW gateway reads by the shared `Sql.Common` evaluator — identical to the SQL backends, Calculated bits, interval-start stamps, #677 reverse; Min/Max are no longer AVEVA modes, the read moves every raw sample of the window (a month of 1 Hz ⇒ `BadTooManyOperations` at the default `MaxTimeWeightedRawRows` cap), and every gateway stream is consumed to completion (an abandoned one makes the gateway evict its shared historian session) — plus TimeAverage/Total via AVEVA modes (interval-start stamped + Calculated; ⚠️ #685: AVEVA's extra pre-window row is dropped ⇒ one row per interval in `[Start, End)`, and an Uncertain (opc 64) interval keeps its value as `UncertainDataSubNormal` — AVEVA's TWA excludes NULL-gap time where the SQL backends bridge it; per-interval values still need gateway fix historiangw#27); the other time-weighted five stay `BadAggregateNotSupported` because the gateway has no prior-sample read for Part 13 bounds; `docs/Historian.md` §Aggregate set), and AtTime over historized variable nodes; Events over alarm-owning equipment-folder event-notifier nodes. Reads are ungated (served from any redundancy node); authorization uses the standard `AccessLevels.HistoryRead` bit set at materialization. diff --git a/docs/Historian.md b/docs/Historian.md index dfe74f86..6353e9ce 100644 --- a/docs/Historian.md +++ b/docs/Historian.md @@ -673,6 +673,32 @@ are disposed when the session closes). Resuming an unknown / evicted / released `BadContinuationPointInvalid`. `releaseContinuationPoints` drops the stored cursors without reading data. +### Gateway raw reads return stored samples only (#687) + +> ⚠️ **Client-visible (#687).** An AVEVA `Full` raw read starts with a synthetic row stamped exactly at the **query start**: the +> value in effect there (AVEVA `Quality = 133`, the previous sample's value and quality — a value never stored at that time) or, +> before the tag's first sample, a NULL row (value 0, Bad, `Quality = 1`). The `Gateway` backend used to return that row as a stored +> sample: a window with no stored sample in it answered one fabricated value stamped at StartTime (live: `[18:06:21, 18:06:39]` ⇒ +> `18:06:21 v=30 Good`), a window before the tag's history answered `v=0 Bad` instead of `GoodNoData`, and a start-only read led +> with the fabricated row. It is now dropped; such windows read `GoodNoData`. + +- **How.** The same classification the raw-evaluated aggregates use (#681, `RawAggregateReader.RequestStartFor` / + `IsWindowSample` — one authority): the gateway is asked from **1 ms before** the window, so the boundary row lands before it, and + only rows stamped at or after StartTime are returned. The decision is by **timestamp**, not by value or quality — a stored NULL + sample looks exactly like the no-data boundary row on the wire (value 0, Bad), and a stored NULL inside the window is still + returned (Bad), as it is a stored sample. +- **Paging (#669) is unaffected.** The gateway is asked for `NumValuesPerNode + 16` rows (`NumValuesPerNode = 0` stays unlimited), + so the boundary row cannot shorten a page: a full page still holds exactly `NumValuesPerNode` stored samples and carries a + continuation point, and a resumed page (which re-reads from the last returned timestamp inclusive, and the tie-cluster over-fetch + `start == end` read) no longer gets a fabricated row at its resume point. If more than 16 rows lie in the 1 ms before the window + and exhaust that allowance, the page cannot be read completely and the node fails (`HistoryReadLimitExceededException`, not + counted as a backend fault) rather than answering a short page that would end the client's paging early. Every gateway stream is + still drained to completion. +- **`ReturnBounds` is not honoured** — on any backend: the server does not pass it to the data source, so no bounding values are + returned (see [Known limitations](#known-limitations)). The AVEVA boundary row is not used as a bound either: it is stamped at the + query start rather than carrying the time of the real prior sample, so it could only ever be a Part 11 *interpolated* bound, which + this backend does not compute. + ### Continuation-point paging (Events) (#669) A capped event read (`NumValuesPerNode > 0`) pages exactly like Raw, with one difference: the trigger is the **source's @@ -964,6 +990,8 @@ whatever the SDK sets — #682 touches only the i=11201 folder (follow-up candid sample per requested timestamp, so the single-shot backend returns the complete result in one read and there is no "full page ⇒ maybe more" signal to page on. Returning the full result with no continuation point is spec-conformant. +- **`ReturnBounds` is ignored** (every backend): Raw reads return stored values only; no bounding values are added at the window + edges. The node manager does not pass the flag to `IHistorianDataSource`. - **No modified-value history on the `Gateway` backend.** Modified reads (`IsReadModified = true`) are served by `SqlServer` and `QuestDb` ([Modified-value history](#modified-value-history-part-11), #668); the HistorianGateway exposes no modified/edited-history surface, so on `Gateway` (and with no historian) they return `BadHistoryOperationUnsupported`. 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 e62567b5..828fc113 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 @@ -80,6 +80,8 @@ 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. + /// Stored samples only (#687 — Part 11 Raw): AVEVA's synthetic start-boundary row is never returned, see + /// . /// public async Task ReadRawAsync( string fullReference, DateTime startUtc, DateTime endUtc, uint maxValuesPerNode, @@ -90,17 +92,8 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis try { - // 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) - .ConfigureAwait(false)) - { - samples.Add(sample); - } - + var samples = await ReadRawWindowAsync(fullReference, startUtc, endUtc, maxValuesPerNode, cancellationToken) + .ConfigureAwait(false); var snapshots = SampleMapper.ToSnapshots(samples); RecordOutcome(success: true, error: null); return new HistoryReadResult(snapshots, ContinuationPoint: null); @@ -117,6 +110,11 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis $"The HistorianGateway refused the row budget of the raw read of '{fullReference}' (its " + "Historian:ReadLimits:MaxValuesCeiling); set NumValuesPerNode or narrow the window."); } + catch (HistoryReadLimitExceededException) + { + // A client-request limit (#687: the pre-window margin swallowed the row budget), not a backend fault. + throw; + } catch (Exception ex) { RecordReadFailure(ex); @@ -124,6 +122,61 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis } } + /// + /// The stored samples of [startUtc, endUtc], oldest first, at most + /// (0 = no cap) — without AVEVA's synthetic rows (#687). + /// + /// + /// + /// An AVEVA Full raw read starts with a row stamped exactly at the query start: the value in effect there + /// (Quality = 133, the prior sample's value — never stored at that time) or, before the tag's first sample, + /// a NULL row (value 0, Bad). Part 11 Raw with ReturnBounds = false wants stored values only, and the + /// server's paging (#669) counts every returned row toward the page and resumes at the last returned timestamp, + /// so a fabricated row would also be counted as a sample and re-fabricated at every resume point. The read + /// therefore uses the same classification as the raw-evaluated aggregates (#681): it is requested from + /// (1 ms early) and only rows + /// are kept — the decision is by TIMESTAMP, because a stored NULL + /// sample and the no-data boundary row look the same on the wire (both carry a value 0 with a Bad quality). + /// + /// + /// The gateway is asked for extra rows so the boundary row + /// (and any stored sample inside the margin) cannot shorten a page: a full page still holds exactly the cap of + /// stored samples, so the server's "full page ⇒ continuation point" rule is unchanged. If the stream delivers its + /// whole budget yet fewer than the cap belong to the window, the page cannot be read completely; answering a short + /// page would end the client's paging early, so the read fails with + /// instead. The stream is always consumed to completion (never + /// broken out of — a cancelled gateway stream evicts its pooled historian session). ReturnBounds is not + /// plumbed to the data source on any backend, so bounding values are never synthesised here. + /// + /// + private async Task> ReadRawWindowAsync( + string fullReference, DateTime startUtc, DateTime endUtc, uint maxValuesPerNode, + CancellationToken cancellationToken) + { + // The gateway seam caps with an int; OPC UA hands us a uint, so clamp to int range. 0 ("all values") is sent as + // int.MaxValue: the client package and the gateway reject max_values <= 0 (#686 review). + var cap = (int)Math.Min(maxValuesPerNode, int.MaxValue); + var budget = cap == 0 ? int.MaxValue : (int)Math.Min(int.MaxValue, (long)cap + RawAggregateReader.PreWindowAllowance); + var samples = new List(); + long received = 0; + await foreach (var sample in _client + .ReadRawAsync(fullReference, RawAggregateReader.RequestStartFor(startUtc), endUtc, budget, cancellationToken) + .ConfigureAwait(false)) + { + received++; + // Filter while draining — the rest of the (bounded) stream is consumed, never abandoned. + if (RawAggregateReader.IsWindowSample(sample, startUtc) && (cap == 0 || samples.Count < cap)) + samples.Add(sample); + } + + if (cap > 0 && received >= budget && samples.Count < cap) + throw new HistoryReadLimitExceededException(cap, + $"The raw read of '{fullReference}' could not be read completely: more than " + + $"{RawAggregateReader.PreWindowAllowance} rows lie in the {RawAggregateReader.BoundaryMargin.TotalMilliseconds} ms " + + "before the window and exhausted the row budget."); + return samples; + } + /// /// /// diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/RawAggregateReader.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/RawAggregateReader.cs index 64738bca..ae75f362 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/RawAggregateReader.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/RawAggregateReader.cs @@ -74,6 +74,31 @@ internal static class RawAggregateReader /// Rows the earlier-data probe asks for (a possible boundary row plus the first stored sample). internal const int ProbeRows = 2; + /// + /// The ReadRaw start for a window that begins at : + /// earlier, so the AVEVA start-boundary row — stamped exactly at the QUERY start, carrying the value in effect there + /// (Quality = 133) or, with no earlier data, a NULL row (value 0, Bad OPC quality) — lands BEFORE the window + /// (). The one place both the raw-evaluated aggregates (#681) and a raw HistoryRead + /// (#687) decide what the request start is. No margin when the window starts within one margin of + /// . + /// + /// The window's (inclusive) start. + /// The start to request. + public static DateTime RequestStartFor(DateTime windowStartUtc) => + windowStartUtc.Ticks > BoundaryMargin.Ticks ? windowStartUtc - BoundaryMargin : windowStartUtc; + + /// + /// True when a row of a stream requested from is a stored sample of the window that + /// begins at : it is stamped at or after the window start. A row before it is the + /// AVEVA boundary row (whatever its value or quality — a stored NULL looks the same on the wire, so the timestamp is + /// the evidence) or a stored sample inside the margin; a row without a timestamp is never a sample. + /// + /// The gateway row. + /// The window's (inclusive) start. + /// Whether the row belongs to the window. + public static bool IsWindowSample(HistorianSample raw, DateTime windowStartUtc) => + raw.Timestamp is { } ts && ts.ToDateTime() >= windowStartUtc; + /// True for the aggregates this reader serves: the per-sample families of . /// Requested aggregate. /// Whether the aggregate is evaluated from raw samples. @@ -113,7 +138,7 @@ internal static class RawAggregateReader { if (plan.BucketCount == 0) return []; var needsRange = AggregateEvaluator.NeedsDataRange(aggregate, plan); - var requestStart = plan.StartUtc.Ticks > BoundaryMargin.Ticks ? plan.StartUtc - BoundaryMargin : plan.StartUtc; + var requestStart = RequestStartFor(plan.StartUtc); var maxValues = MaxValuesFor(maxWindowRows); var buckets = new RawBucketAccumulator(aggregate == HistoryAggregateType.MaximumActualTime); 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 c5e13159..8149ed0d 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 @@ -20,11 +20,13 @@ public sealed class GatewayHistorianDataSourceTests }, }; var ds = new GatewayHistorianDataSource(fake, NullLogger.Instance); - var r = await ds.ReadRawAsync("T", DateTime.UtcNow.AddMinutes(-5), DateTime.UtcNow, 100, TestContext.Current.CancellationToken); + var start = new DateTime(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc); + var r = await ds.ReadRawAsync("T", start, start.AddMinutes(5), 100, TestContext.Current.CancellationToken); Assert.Equal(2, r.Samples.Count); Assert.Equal(0x80000000u, r.Samples[1].StatusCode); // Bad from quality 0 Assert.Equal("T", fake.LastReadRawTag); - Assert.Equal(100, fake.LastReadRawMaxValues); + // #687: asked from 1 ms before the window with room for AVEVA's boundary row (GatewayRawBoundaryRowTests). + Assert.Equal(100 + RawAggregateReader.PreWindowAllowance, fake.LastReadRawMaxValues); } [Fact] diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayRawBoundaryRowTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayRawBoundaryRowTests.cs new file mode 100644 index 00000000..4a95d364 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayRawBoundaryRowTests.cs @@ -0,0 +1,209 @@ +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; + +/// +/// #687: a raw HistoryRead must return stored samples only (Part 11, ReturnBounds = false). An AVEVA Full +/// raw read starts with a synthetic row stamped exactly at the QUERY start — the value in effect there +/// (Quality = 133, the prior sample's value and OPC quality) or, with no earlier data, a NULL row. The fake emits +/// the REAL wire shape the live gate captured (2026-10-12): every row carries a value (histsdk puts NumericValue +/// on every numeric row), so a NULL row is NumericValue = 0 + OPC quality 0 / Quality = 1 — the evidence +/// that a row is synthetic is its timestamp (the query start), never "it has no value". A stored NULL/Bad sample looks +/// the same on the wire (q=1 opc=0) and IS a stored sample. +/// +public sealed class GatewayRawBoundaryRowTests +{ + private static readonly CancellationToken Ct = CancellationToken.None; + + /// The live gate's tag HistGW.LiveTest.Gate681 (18:06:40 is a stored NULL). + private static readonly DateTime T0 = new(2026, 10, 12, 18, 6, 0, DateTimeKind.Utc); + + private static readonly (double Sec, double Value, bool Null)[] Gate681 = + [ + (0, 10, false), (5, 20, false), (20, 30, false), (40, 0, true), (50, 5, false), (60, 1, false), (90, 1, false), + (100, 7, false), + ]; + + private static DateTime At(double seconds) => T0.AddTicks((long)(seconds * TimeSpan.TicksPerSecond)); + + private static HistorianSample Stored(DateTime t, double v, bool isNull) => new() + { + Tag = "T", + Timestamp = Timestamp.FromDateTime(t), + NumericValue = isNull ? 0 : v, + OpcQuality = isNull ? 0u : 192u, + Quality = isNull ? 1u : 0u, + }; + + /// + /// An AVEVA-shaped gateway over : ReadRaw over [start, end] ascending, capped at + /// max, led by the boundary row at the query start when nothing is stored exactly there. + /// + private static FakeHistorianGatewayClient Aveva(IReadOnlyList<(DateTime T, double V, bool Null)> data) + { + var sorted = data.OrderBy(d => d.T).ToList(); + return new FakeHistorianGatewayClient + { + RawSource = (start, end, max) => + { + var rows = new List(); + if (sorted.All(d => d.T != start)) + { + var prior = sorted.Where(d => d.T < start).Select(d => ((DateTime T, double V, bool Null)?)d).LastOrDefault(); + HistorianSample boundary; + if (prior is { } p) + { + boundary = Stored(start, p.V, p.Null); + if (!p.Null) boundary.Quality = 133; // copies the prior's quality; a Bad prior stays q=1 opc=0 + } + else + { + boundary = new HistorianSample { Tag = "T", Timestamp = Timestamp.FromDateTime(start), NumericValue = 0, OpcQuality = 0, Quality = 1 }; + } + rows.Add(boundary); + } + rows.AddRange(sorted.Where(d => d.T >= start && d.T <= end).Select(d => Stored(d.T, d.V, d.Null))); + return max > 0 ? rows.Take(max) : rows; + }, + }; + } + + private static FakeHistorianGatewayClient Gate() => Aveva(Gate681.Select(x => (At(x.Sec), x.Value, x.Null)).ToList()); + + private static GatewayHistorianDataSource Source(FakeHistorianGatewayClient fake) => + new(fake, NullLogger.Instance); + + [Fact] + public async Task A_window_holding_no_stored_sample_returns_nothing_not_the_value_in_effect() + { + // Live: Raw [18:06:21, 18:06:39] returned "18:06:21 v=30 Good" — a value never stored at that time. + var r = await Source(Gate()).ReadRawAsync("T", At(21), At(39), 0, Ct); + r.Samples.ShouldBeEmpty(); + } + + [Fact] + public async Task A_window_before_the_first_sample_returns_nothing_not_a_Bad_zero() + { + // Live: Raw [18:05:00, 18:05:30] returned "18:05:00 v=0 Bad" instead of GoodNoData. + var r = await Source(Gate()).ReadRawAsync("T", At(-60), At(-30), 10, Ct); + r.Samples.ShouldBeEmpty(); + } + + [Fact] + public async Task A_start_only_read_starts_at_the_first_stored_sample() + { + var r = await Source(Gate()).ReadRawAsync("T", At(30), new DateTime(DateTime.MaxValue.Ticks, DateTimeKind.Utc), 3, Ct); + r.Samples.Select(s => s.SourceTimestampUtc).ShouldBe(new DateTime?[] { At(40), At(50), At(60) }); + } + + [Fact] + public async Task A_stored_NULL_sample_in_the_window_is_kept_as_the_Bad_sample_it_is() + { + var r = await Source(Gate()).ReadRawAsync("T", At(30), At(55), 0, Ct); + r.Samples.Select(s => s.SourceTimestampUtc).ShouldBe(new DateTime?[] { At(40), At(50) }); + r.Samples[0].StatusCode.ShouldBe(0x80000000u); + } + + [Fact] + public async Task A_stored_sample_exactly_at_the_window_start_is_kept() + { + var r = await Source(Gate()).ReadRawAsync("T", At(20), At(45), 0, Ct); + r.Samples.Select(s => s.SourceTimestampUtc).ShouldBe(new DateTime?[] { At(20), At(40) }); + r.Samples[0].Value.ShouldBe(30d); + } + + [Fact] + public async Task The_raw_request_starts_one_millisecond_early_and_asks_for_the_boundary_allowance() + { + var fake = Gate(); + await Source(fake).ReadRawAsync("T", At(30), At(55), 5, Ct); + fake.RawCalls.Single().ShouldBe((At(30) - RawAggregateReader.BoundaryMargin, At(55), 5 + RawAggregateReader.PreWindowAllowance)); + } + + [Fact] + public async Task An_unlimited_read_stays_unlimited() + { + var fake = Gate(); + await Source(fake).ReadRawAsync("T", At(30), At(55), 0, Ct); + fake.RawCalls.Single().MaxValues.ShouldBe(0); + } + + [Fact] + public async Task A_full_page_holds_exactly_the_cap_and_the_stream_is_drained_not_abandoned() + { + var fake = Gate(); + var r = await Source(fake).ReadRawAsync("T", At(0), At(200), 3, Ct); + r.Samples.Select(s => s.SourceTimestampUtc).ShouldBe(new DateTime?[] { At(0), At(5), At(20) }); + fake.RawStreamsCompleted.ShouldBe(1); + } + + [Theory] + [InlineData(1u)] + [InlineData(2u)] + [InlineData(3u)] + public async Task Server_style_paging_returns_every_stored_sample_once_and_nothing_else(uint cap) + { + // Ties at +60 (three samples) and a stored NULL at +40, over the AVEVA-shaped wire. + var data = Gate681.Select(x => (At(x.Sec), x.Value, x.Null)) + .Concat([(At(60), 2d, false), (At(60), 3d, false)]).ToList(); + var fake = Aveva(data); + var ds = Source(fake); + var start = At(-10); + var end = At(200); + var got = new List(); + var skip = 0; + var cursor = start; + var resume = false; + for (var guard = 0; guard < 100; guard++) + { + // The node manager's paging (OtOpcUaNodeManager.ServeRawPagedAsync + HistoryPaging): a full page ⇒ resume at + // the last timestamp INCLUSIVE dropping the ties already returned; a short page is the last; a resume that + // cannot advance (a tie cluster larger than the cap) over-fetches the cluster with a start == end read. + var page = (await ds.ReadRawAsync("T", cursor, end, cap, Ct)).Samples; + var full = page.Count >= cap; + var samples = resume + ? page.SkipWhile((x, i) => i < skip && x.SourceTimestampUtc == cursor).ToList() + : page.ToList(); + if (resume && full && (samples.Count == 0 || samples[^1].SourceTimestampUtc == cursor)) + { + var cluster = (await ds.ReadRawAsync("T", cursor, cursor, 1000, Ct)).Samples; + var sliceStart = Math.Min(skip, cluster.Count); + var sliceCount = Math.Min((int)cap, cluster.Count - sliceStart); + got.AddRange(cluster.Skip(sliceStart).Take(sliceCount)); + if (sliceStart + sliceCount < cluster.Count) + { + skip = sliceStart + sliceCount; + continue; + } + cursor = cursor.AddTicks(1); + skip = 0; + if (cursor > end) break; + continue; + } + got.AddRange(samples); + if (!full || samples.Count == 0) break; + cursor = samples[^1].SourceTimestampUtc!.Value; + skip = samples.AsEnumerable().Reverse().TakeWhile(x => x.SourceTimestampUtc == cursor).Count(); + resume = true; + } + got.Select(s => s.SourceTimestampUtc).ShouldBe(data.Select(d => (DateTime?)d.Item1).OrderBy(t => t).ToArray()); + } + + [Fact] + public async Task A_margin_that_swallows_the_row_budget_fails_the_node_rather_than_answering_a_short_page() + { + // More stored rows inside the 1 ms margin than the allowance: the page could not be read, and a short page would + // end the client's paging early — refuse it (a request limit, not a backend fault). + var margin = Enumerable.Range(0, 40).Select(i => (At(30).AddTicks(-1 - i), (double)i, false)); + var fake = Aveva(margin.Append((At(35), 1d, false)).ToList()); + var ds = Source(fake); + await Should.ThrowAsync(() => ds.ReadRawAsync("T", At(30), At(40), 5, Ct)); + ds.GetHealthSnapshot().TotalFailures.ShouldBe(0); + fake.RawStreamsCompleted.ShouldBe(1); + } +} From 6796ba412640603b03a52b54b04d64a56d79510d Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Mon, 12 Oct 2026 14:37:35 -0400 Subject: [PATCH 3/5] fix(historian): #688 clamp the Gateway per-sample row budget at the gateway's default ceiling The default MaxTimeWeightedRawRows (1,000,000) + 18 exceeded the gateway's default MaxValuesCeiling (1,000,000), so with AllowStreamingBeyondCeiling=false every Gateway per-sample aggregate read was refused. The allowance is now clamped at 1,000,000 while the cap is at or under it (RawAggregateReader. ClampToGatewayCeiling); a cap above it keeps its allowance. Raw reads are not clamped (rebased onto #686). --- CLAUDE.md | 2 +- docs/Historian.md | 18 +++++++---- .../RawAggregateReader.cs | 29 +++++++++++++++-- .../GatewayRawAggregateTests.cs | 32 ++++++++++++++++++- 4 files changed, 69 insertions(+), 12 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 96dafeef..c9c902df 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -1383,7 +1383,7 @@ authorization uses the standard `AccessLevels.HistoryRead` bit set at materializ | `MaxTieClusterOverfetch` | `65536` | Bounded over-fetch the HistoryRead-Raw paging uses to page within an oversized same-timestamp tie cluster (retained from the prior backend) | | `HistoryReadBatchParallelism` | `4` | Max nodes served concurrently within one HistoryRead batch (03/S3); `⌈N/P⌉ × CallTimeout` worst case instead of sequential | | `MaxConcurrentHistoryReadBatches` | `16` | Process-wide concurrent HistoryRead batch cap (03/S3); flood degrades to `BadTooManyOperations`. Total in-flight reads capped at 16 × 4 = 64 | -| `MaxTimeWeightedRawRows` | `1000000` | `SqlServer`/`QuestDb` (#671): raw in-window rows one time-weighted aggregate read (incl. `Total` and, since #680, PercentGood/PercentBad) may stream into the C# evaluator — and, since #681, every `Gateway` per-sample read (raw-evaluated); window query capped at this + 1, past it ⇒ `BadTooManyOperations` for that node. ≤ 0 = unbounded | +| `MaxTimeWeightedRawRows` | `1000000` | `SqlServer`/`QuestDb` (#671): raw in-window rows one time-weighted aggregate read (incl. `Total` and, since #680, PercentGood/PercentBad) may stream into the C# evaluator — and, since #681, every `Gateway` per-sample read (raw-evaluated); window query capped at this + 1, past it ⇒ `BadTooManyOperations` for that node. ≤ 0 = unbounded. Gateway (#688): the `cap + 18` budget is clamped at the gateway's default `MaxValuesCeiling` (1,000,000) while the cap is ≤ it, so the default no longer trips a strict (`AllowStreamingBeyondCeiling=false`) gateway | | `HistoryReadDeadline` | `00:01:00` | Per-request HistoryRead wall-clock deadline (03/S3); a node still in flight surfaces `BadTimeout`. Non-positive = unbounded | ### Alarm-history path (`AlarmHistorian` section) diff --git a/docs/Historian.md b/docs/Historian.md index 6353e9ce..cdbc450b 100644 --- a/docs/Historian.md +++ b/docs/Historian.md @@ -146,7 +146,7 @@ Processed read is `BadAggregateNotSupported` and no history aggregate functions | `MaxTieClusterOverfetch` | int | `65536` | (Node-manager paging; all backends — a `SqlServer` PK rules out tie clusters, so it is moot there.) Maximum samples the server will fetch in one shot to page through a tie cluster (multiple samples sharing one `SourceTimestamp`). A cluster larger than this ceiling fails `BadTooManyOperations` (#686; was `BadHistoryOperationUnsupported`). Also the per-read row budget of a Gateway reverse raw read (walked back from forward reads, #686): one timestamp holding more samples is refused `BadTooManyOperations`. Values below 1 count as 1. Raise to handle abnormally large tie clusters; the default covers all normal-data cases. | | `HistoryReadBatchParallelism` | int | `4` | Max historized nodes served **concurrently** within one HistoryRead batch (arch-review 03/S3). The SDK's `HistoryRead*` override surface is synchronous, so each arm bridges once at its boundary; this turns per-node serving from sequential (`N × CallTimeout` worst case) into `⌈N / P⌉ × CallTimeout`. Also caps the gateway-load multiplication a single client can command. `≤ 1` serves sequentially. | | `MaxConcurrentHistoryReadBatches` | int | `16` | Process-wide limit on HistoryRead (and HistoryUpdate) batches served **concurrently** (arch-review 03/S3). A flood of history clients degrades to `BadTooManyOperations` rather than exhausting the SDK request-thread pool. With `HistoryReadBatchParallelism` this caps total in-flight gateway reads at `MaxConcurrentHistoryReadBatches × HistoryReadBatchParallelism` (default 16 × 4 = 64). A batch that cannot get a slot within `min(HistoryReadDeadline, 5 s)` fails every handle `BadTooManyOperations`. | -| `MaxTimeWeightedRawRows` | int | `1000000` | `SqlServer` / `QuestDb` (#671), and the `Gateway` per-sample aggregates (#681 — every per-sample read there streams raw samples, so the cap applies to the window's raw rows; see [Aggregate set](#aggregate-set-and-badaggregatenotsupported-285)). The most raw in-window samples one [time-weighted aggregate](#time-weighted-aggregates-part-13-671) read (TimeAverage, TimeAverage2, **Total**, Total2, Interpolative, and — since #680 — **PercentGood/PercentBad**) may stream into the server-side evaluator — that family is **not** pushed down to SQL. The window query is capped at this + 1 rows; one row past the cap fails **that node** with `BadTooManyOperations` (not counted as a backend fault) instead of holding a pooled connection until `HistoryReadDeadline` and returning `BadTimeout`. Bound rows outside the window do not count; per-sample aggregates are not capped on the SQL backends (pushed down). 1 000 000 ≈ 11.5 days of a 1 Hz tag. Non-positive = unbounded (a `Validate()` warning). | +| `MaxTimeWeightedRawRows` | int | `1000000` | `SqlServer` / `QuestDb` (#671), and the `Gateway` per-sample aggregates (#681 — every per-sample read there streams raw samples, so the cap applies to the window's raw rows; see [Aggregate set](#aggregate-set-and-badaggregatenotsupported-285)). The most raw in-window samples one [time-weighted aggregate](#time-weighted-aggregates-part-13-671) read (TimeAverage, TimeAverage2, **Total**, Total2, Interpolative, and — since #680 — **PercentGood/PercentBad**) may stream into the server-side evaluator — that family is **not** pushed down to SQL. The window query is capped at this + 1 rows; one row past the cap fails **that node** with `BadTooManyOperations` (not counted as a backend fault) instead of holding a pooled connection until `HistoryReadDeadline` and returning `BadTimeout`. Bound rows outside the window do not count; per-sample aggregates are not capped on the SQL backends (pushed down). 1 000 000 ≈ 11.5 days of a 1 Hz tag. Non-positive = unbounded (a `Validate()` warning). `Gateway` (#688): the row budget it asks the gateway for (`cap + 18`) is clamped at the gateway's default `MaxValuesCeiling` (1,000,000) while the cap is at or under it, so the default works against a gateway with `AllowStreamingBeyondCeiling=false`; against a gateway with a *lower* ceiling, set this to `ceiling − 18`. | | `HistoryReadDeadline` | TimeSpan | `00:01:00` | Per-request wall-clock deadline for a HistoryRead (and HistoryUpdate) batch (arch-review 03/S3). A node still in flight at the deadline surfaces `BadTimeout`, so one request can never hold an SDK request thread longer than this regardless of node count. Non-positive = unbounded (a `Validate()` warning). | | `HistoryUpdateMaxFutureSkew` | TimeSpan | `1.00:00:00` | #674. How far past the server clock a HistoryUpdate (Insert / Replace / Update) item's `SourceTimestamp` may lie; an item beyond `now + skew` is refused **`BadOutOfRange`** by the node manager before any backend sees it (every backend). Guards against a far-future typo (a 2062 timestamp). The live recorder is not bounded. Non-positive disables the bound (no warning — a deliberate choice). See [HistoryUpdate](#historyupdate). | @@ -890,12 +890,16 @@ and evicts its shared historian session (see the HistorianGateway issue text in - **Look-ahead.** The end-of-data rules need to know whether data exists after the window: a sample exactly at the window end (forward) comes from the window read; otherwise a 2-row read `[end, now + 1 day]`. In both small reads a row stamped exactly at the query start (a boundary row) or the query end (a possible terminal row) is not a sample, whatever its value. -- **Bound.** In-window rows are capped by `MaxTimeWeightedRawRows`; the gateway is asked for `cap + 18` rows. Past the cap the rest - of the (bounded) stream is drained, then the node fails `BadTooManyOperations`; a stream that delivers the whole budget cannot be - proven complete and fails the same way. The gateway's own refusal of that budget (`InvalidArgument` "max_values exceeds the - server ceiling" — `Historian:ReadLimits:MaxValuesCeiling`, default 1,000,000, with `AllowStreamingBeyondCeiling=false`; the - default `1,000,000 + 18` exceeds it) is also `BadTooManyOperations`, not a backend fault — lower `MaxTimeWeightedRawRows` to - `ceiling − 18` on such a gateway. A caller cancellation (client cancel, `HistoryReadDeadline`) is not counted as a backend fault. +- **Bound.** In-window rows are capped by `MaxTimeWeightedRawRows`; the gateway is asked for `cap + 18` rows, **clamped at the + gateway's default ceiling of 1,000,000 when the cap itself is at or under it** (#688 — the default `1,000,000 + 18` used to exceed + that ceiling, so a gateway with `AllowStreamingBeyondCeiling=false` refused **every** per-sample read, even a 9-row window, until + the cap was lowered; the default config now works against the default gateway). A cap above 1,000,000 is a deliberate choice for a + streaming gateway and keeps its `+ 18`. Past the cap the rest of the (bounded) stream is drained, then the node fails + `BadTooManyOperations`; a stream that delivers the whole budget cannot be proven complete and fails the same way (so with the + default cap a window of more than ~999,982 rows fails, as it would one row later). The gateway's own refusal of the budget + (`InvalidArgument` "max_values exceeds the server ceiling" — `Historian:ReadLimits:MaxValuesCeiling` lowered below the default, + with `AllowStreamingBeyondCeiling=false`) is also `BadTooManyOperations`, not a backend fault — lower `MaxTimeWeightedRawRows` to + `ceiling − 18` on such a gateway. A raw read's budget is **not** clamped (see [#687](#gateway-raw-reads-return-stored-samples-only-687)). A caller cancellation (client cancel, `HistoryReadDeadline`) is not counted as a backend fault. - **Bucket statistics** are `RawBucketAccumulator`, the C# statement of the SQL backends' per-sample columns — incl. #684: Average / Minimum / Maximum / Range / *ActualTime over **Good** samples only, the Uncertain extremes feeding only the Min/Max status rule, and an Uncertain String/DateTime sample not refusing the Good-value aggregates. A change to those SQL columns diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/RawAggregateReader.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/RawAggregateReader.cs index ae75f362..484b56c4 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/RawAggregateReader.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/RawAggregateReader.cs @@ -47,8 +47,9 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway; /// SQL backends' time-weighted reads use): past it the node fails with /// (⇒ BadTooManyOperations) once the stream — bounded by the /// requested row budget — has been drained. The gateway is asked for at most cap + 2 + -/// rows; a stream that delivers all of them cannot be proven complete and fails the -/// same way rather than answering from a truncated window. +/// rows, clamped at while the cap is at +/// or under it (#688, ); a stream that delivers all of them cannot be proven +/// complete and fails the same way rather than answering from a truncated window. /// /// /// Values and status. A sample's value is its numeric value (Double), else its string value, else Empty; its @@ -109,7 +110,29 @@ internal static class RawAggregateReader /// The configured cap (<= 0 = unbounded). /// The row limit to request. public static int MaxValuesFor(int maxWindowRows) => - maxWindowRows <= 0 ? int.MaxValue : (int)Math.Min(int.MaxValue, (long)maxWindowRows + 2 + PreWindowAllowance); + maxWindowRows <= 0 ? int.MaxValue : ClampToGatewayCeiling((long)maxWindowRows + 2 + PreWindowAllowance, maxWindowRows); + + /// + /// The HistorianGateway's default Historian:ReadLimits:MaxValuesCeiling. With + /// AllowStreamingBeyondCeiling=false the gateway refuses any max_values above its ceiling. + /// + internal const int GatewayDefaultMaxValuesCeiling = 1_000_000; + + /// + /// #688: a ReadRaw row budget of plus this backend's own allowance + /// () clamped at — but never below + /// itself. The allowance (the boundary row, the margin, the completeness row) must not push a + /// budget that fits the gateway's default ceiling over it: the default MaxTimeWeightedRawRows (1,000,000) + 18 + /// used to exceed it, so a gateway with AllowStreamingBeyondCeiling=false refused EVERY per-sample read. A + /// count above the ceiling is the operator's (or client's) deliberate choice for a streaming gateway and keeps its + /// allowance. The cost of clamping is that a window holding within the allowance of the ceiling cannot be proven + /// complete and fails as over the cap — the same outcome it would get one row later. + /// + /// The budget including the allowance. + /// The caller's own row count (the cap, or NumValuesPerNode), > 0. + /// The max_values to request. + internal static int ClampToGatewayCeiling(long requested, long rows) => + (int)Math.Min(int.MaxValue, rows > GatewayDefaultMaxValuesCeiling ? requested : Math.Min(requested, GatewayDefaultMaxValuesCeiling)); /// Maps one gateway raw sample to a stored sample (value + mapped status, Empty clamped Bad). /// The gateway sample; its timestamp must be set. diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayRawAggregateTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayRawAggregateTests.cs index 5ce178df..b99f2b57 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayRawAggregateTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayRawAggregateTests.cs @@ -369,7 +369,7 @@ public sealed class GatewayRawAggregateTests await Source(count).ReadProcessedAsync("A", At(0), At(80), TimeSpan.FromSeconds(20), HistoryAggregateType.Count, Ct); // The window read ends AT the window end; probe + look-ahead are separate 2-row reads, all consumed to completion. count.RawCalls.ShouldBe([ - (At(0) - TimeSpan.FromMilliseconds(1), At(80), 1_000_000 + 2 + 16), + (At(0) - TimeSpan.FromMilliseconds(1), At(80), 1_000_000), // cap + 18, clamped at the gateway default ceiling (#688) (DateTime.UnixEpoch, At(0).AddTicks(-1), 2), (At(80), Now + TimeSpan.FromDays(1), 2), ]); @@ -473,6 +473,36 @@ public sealed class GatewayRawAggregateTests ds2.GetHealthSnapshot().TotalFailures.ShouldBe(1); } + [Fact] // #688: the default cap's budget fits the gateway's default ceiling, so a strict gateway serves the default config + public async Task The_default_caps_row_budget_does_not_exceed_the_gateways_default_ceiling() + { + var fake = Store([D(-5, 1), D(5, 2)], avevaBoundary: true); + await Source(fake).ReadProcessedAsync("T", At(0), At(10), TimeSpan.Zero, HistoryAggregateType.Average, Ct); + TimeWeightedAggregator.DefaultMaxWindowRows.ShouldBe(1_000_000); + fake.RawCalls[0].MaxValues.ShouldBe(RawAggregateReader.GatewayDefaultMaxValuesCeiling); + } + + [Theory] // #688: the + 18 allowance is clamped at the gateway's default ceiling, never below the cap itself + [InlineData(10, 28)] + [InlineData(999_982, 1_000_000)] + [InlineData(999_990, 1_000_000)] + [InlineData(1_000_000, 1_000_000)] + [InlineData(2_000_000, 2_000_018)] // a cap above the default ceiling is a deliberate choice for a streaming gateway + [InlineData(0, int.MaxValue)] + [InlineData(-1, int.MaxValue)] + public void The_window_row_budget_is_clamped_at_the_gateways_default_ceiling(int cap, int expected) => + RawAggregateReader.MaxValuesFor(cap).ShouldBe(expected); + + [Fact] // #688: near the ceiling the window is still served when it is not full + public async Task A_cap_at_the_ceiling_still_serves_a_window_below_it() + { + var data = Enumerable.Range(0, 10).Select(i => D(i, i)).ToArray(); + var fake = Store(data, avevaBoundary: true); + var r = await Source(fake, maxRawRows: 1_000_000) + .ReadProcessedAsync("T", At(0), At(10), TimeSpan.Zero, HistoryAggregateType.Average, Ct); + r.Samples.Single().Value.ShouldBe(4.5); + } + [Fact] public async Task A_caller_cancellation_is_not_counted_as_a_backend_failure() { From 8114f13a28cc41683350c96d0dfd11b446a667de Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Mon, 12 Oct 2026 15:19:34 -0400 Subject: [PATCH 4/5] =?UTF-8?q?fix(historian):=20#687=20review=20=E2=80=94?= =?UTF-8?q?=20one=20bounded=20re-read=20instead=20of=20failing,=20raw=20bu?= =?UTF-8?q?dget=20unclamped,=20exclusive=20forward=20end?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - M1/M2: when the 1 ms pre-window margin (boundary row + stored samples, e.g. a >16-sample tie cluster at start-1ms / lastTs-1ms) uses up the cap + 16 budget, re-read ONCE asking for cap + m + 16 (m = pre-window rows seen); fail BadTooManyOperations only if the re-read is still short because of them. The raw budget is no longer clamped at the gateway default ceiling (AVEVA always emits the boundary row, so page sizes 999,985-1,000,000 always came back short); the #688 clamp stays on the per-sample aggregate path only. - Merge trap: NumValuesPerNode 0 => int.MaxValue, never through the clamp. - L3: the forward end is exclusive ([start, end), the IHistorianDataSource contract); a point read (start == end, the tie-cluster over-fetch) inclusive. - RequestStartFor never goes before the Unix epoch (#686's walk-back floor). - L4 (docs): an Uncertain native row for an all-NULL interval is unverified live. --- CLAUDE.md | 2 +- docs/Historian.md | 37 ++++-- .../GatewayHistorianDataSource.cs | 79 ++++++++---- .../RawAggregateReader.cs | 25 ++-- .../GatewayRawBoundaryRowTests.cs | 122 +++++++++++++++--- 5 files changed, 202 insertions(+), 63 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index c9c902df..9672b8ce 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -1364,7 +1364,7 @@ historian tagname, which is the tag's driver `FullName`. The server dispatches all OPC UA HistoryRead to the registered `IHistorianDataSource` — the `GatewayHistorianDataSource` read client when enabled, else the `NullHistorianDataSource` default -(historized nodes return `GoodNoData` for Raw/AtTime — Processed is `BadAggregateNotSupported` since #682). Supported variants: Raw (⚠️ #687, client-visible: Gateway raw reads no longer return AVEVA's synthetic start-boundary row — asked from 1 ms early with `NumValuesPerNode + 16` rows and filtered by timestamp, the #681 classification — so an empty window is `GoodNoData`, not a fabricated value at StartTime; `ReturnBounds` is ignored on every backend), Processed +(historized nodes return `GoodNoData` for Raw/AtTime — Processed is `BadAggregateNotSupported` since #682). Supported variants: Raw (⚠️ #687, client-visible: Gateway raw reads no longer return AVEVA's synthetic start-boundary row — asked from 1 ms early (epoch-floored) with `NumValuesPerNode + 16` rows (unclamped; 0 ⇒ `int.MaxValue`) and filtered by timestamp — so an empty window is `GoodNoData`, not a fabricated value at StartTime; the forward end is now exclusive (a point read inclusive); pre-window rows that crowd the budget trigger ONE re-read with room for them, else `BadTooManyOperations`; `ReturnBounds` is ignored on every backend), Processed (backend-dependent: SqlServer/QuestDb serve seventeen; Gateway twelve = `GatewayHistorianDataSource.ServedAggregates` — ⚠️ #681, client-visible: the ten per-sample ones (Average, Count, Min/Max, Range, Start/End/Delta, Min/MaxActualTime) are evaluated from RAW gateway reads by the shared `Sql.Common` evaluator — identical to the SQL backends, Calculated bits, interval-start stamps, #677 reverse; Min/Max are no longer AVEVA modes, the read moves every raw sample of the window (a month of 1 Hz ⇒ `BadTooManyOperations` at the default `MaxTimeWeightedRawRows` cap), and every gateway stream is consumed to completion (an abandoned one makes the gateway evict its shared historian session) — plus TimeAverage/Total via AVEVA modes (interval-start stamped + Calculated; ⚠️ #685: AVEVA's extra pre-window row is dropped ⇒ one row per interval in `[Start, End)`, and an Uncertain (opc 64) interval keeps its value as `UncertainDataSubNormal` — AVEVA's TWA excludes NULL-gap time where the SQL backends bridge it; per-interval values still need gateway fix historiangw#27); the other time-weighted five stay `BadAggregateNotSupported` because the gateway has no prior-sample read for Part 13 bounds; `docs/Historian.md` §Aggregate set), and AtTime over historized variable nodes; Events over alarm-owning equipment-folder event-notifier nodes. Reads are ungated (served from any redundancy node); authorization uses the standard `AccessLevels.HistoryRead` bit set at materialization. diff --git a/docs/Historian.md b/docs/Historian.md index cdbc450b..64363ba9 100644 --- a/docs/Historian.md +++ b/docs/Historian.md @@ -682,18 +682,28 @@ data. > `18:06:21 v=30 Good`), a window before the tag's history answered `v=0 Bad` instead of `GoodNoData`, and a start-only read led > with the fabricated row. It is now dropped; such windows read `GoodNoData`. -- **How.** The same classification the raw-evaluated aggregates use (#681, `RawAggregateReader.RequestStartFor` / - `IsWindowSample` — one authority): the gateway is asked from **1 ms before** the window, so the boundary row lands before it, and - only rows stamped at or after StartTime are returned. The decision is by **timestamp**, not by value or quality — a stored NULL - sample looks exactly like the no-data boundary row on the wire (value 0, Bad), and a stored NULL inside the window is still - returned (Bad), as it is a stored sample. -- **Paging (#669) is unaffected.** The gateway is asked for `NumValuesPerNode + 16` rows (`NumValuesPerNode = 0` stays unlimited), - so the boundary row cannot shorten a page: a full page still holds exactly `NumValuesPerNode` stored samples and carries a - continuation point, and a resumed page (which re-reads from the last returned timestamp inclusive, and the tie-cluster over-fetch - `start == end` read) no longer gets a fabricated row at its resume point. If more than 16 rows lie in the 1 ms before the window - and exhaust that allowance, the page cannot be read completely and the node fails (`HistoryReadLimitExceededException`, not - counted as a backend fault) rather than answering a short page that would end the client's paging early. Every gateway stream is - still drained to completion. +- **How.** The same request start the raw-evaluated aggregates use (#681, `RawAggregateReader.RequestStartFor` — one authority): + the gateway is asked from **1 ms before** the window (never before the Unix epoch, the start of the historian's time encoding), + so the boundary row lands before it, and only rows `RawAggregateReader.IsWindowSample` are returned. The decision is by + **timestamp**, not by value or quality — a stored NULL sample looks exactly like the no-data boundary row on the wire (value 0, + Bad), and a stored NULL inside the window is still returned (Bad), as it is a stored sample. +- **The forward end is exclusive** (`[StartTime, EndTime)`, the `IHistorianDataSource` contract and what the SQL backends return; + the gateway's own `ReadRaw` end is inclusive, so a sample stored exactly at EndTime used to be returned). A point read + (`StartTime == EndTime` — the server's tie-cluster over-fetch) still returns the samples at that instant. The #686 reverse + walk-back is exact either way. +- **Row budget.** `NumValuesPerNode = 0` ("all values") asks the gateway for an unbounded `max_values` (`int.MaxValue` — the client + package and the gateway reject 0, #686). A cap asks for `NumValuesPerNode + 16` rows, **not** clamped at the gateway's default + ceiling (#688 clamps only the per-sample aggregate budget): AVEVA always emits the boundary row, so a clamped budget would make + every page size from 999,985 to 1,000,000 come back short. A gateway with `AllowStreamingBeyondCeiling=false` refuses an + over-ceiling budget instead ⇒ `BadTooManyOperations` (not a backend fault); a default gateway serves it. +- **Paging (#669) is unaffected.** A full page still holds exactly `NumValuesPerNode` stored samples and carries a continuation + point, and a resumed page (which re-reads from the last returned timestamp inclusive, and the tie-cluster over-fetch + `start == end` read) no longer gets a fabricated row at its resume point. A read is complete when the stream came back short, + filled the page, or reached EndTime. If rows **before** the window — the boundary row plus stored samples inside the 1 ms margin, + e.g. a tie cluster at `StartTime − 1 ms` or at a resume point `− 1 ms` — used up the budget, the window is re-read **once** + asking for `NumValuesPerNode + m + 16` (m = the pre-window rows the first read saw, unclamped). Only if that re-read is still + short because of pre-window rows does the node fail `BadTooManyOperations` (not a backend fault) rather than answer a short page + that would end the client's paging early. Every gateway stream is still drained to completion. - **`ReturnBounds` is not honoured** — on any backend: the server does not pass it to the data source, so no bounding values are returned (see [Known limitations](#known-limitations)). The AVEVA boundary row is not used as a bound either: it is stamped at the query start rather than carrying the time of the real prior sample, so it could only ever be a Part 11 *interpolated* bound, which @@ -845,6 +855,9 @@ fixed: - ⚠️ **End-to-end correctness also needs the gateway fix historiangw#27:** the gateway sends the aggregate interval to AVEVA in ticks where AVEVA reads milliseconds (10,000× too large), so today a native read returns one row for the whole window (plus the pre-window row, now dropped). OtOpcUa does not work around it; per-interval values return once the gateway is fixed. +- **Not verified live:** what AVEVA answers for an interval holding **only** NULL samples (no good time at all). If it marks such a + row Uncertain (`opc_quality` 64–191) with a placeholder value, that value is now passed through as Uncertain; the gate only + observed Uncertain rows that also held good time. Re-check on the VPN once historiangw#27 is fixed. **Integral unit (#677).** AVEVA's integral is `∫ value dt` (seconds) ÷ the tag's `IntegralDivisor`. The proto3 `double` default for `HistorianTagDefinition.IntegralDivisor` is `0`, so `GatewayTagProvisioner` now sends **`IntegralDivisor = 1`** explicitly on every 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 828fc113..79986db2 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 @@ -123,8 +123,8 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis } /// - /// The stored samples of [startUtc, endUtc], oldest first, at most - /// (0 = no cap) — without AVEVA's synthetic rows (#687). + /// The stored samples of [startUtc, endUtc) (a point read start == end: of that instant), oldest first, + /// at most (0 = all) — without AVEVA's synthetic rows (#687). /// /// /// @@ -133,48 +133,81 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis /// a NULL row (value 0, Bad). Part 11 Raw with ReturnBounds = false wants stored values only, and the /// server's paging (#669) counts every returned row toward the page and resumes at the last returned timestamp, /// so a fabricated row would also be counted as a sample and re-fabricated at every resume point. The read - /// therefore uses the same classification as the raw-evaluated aggregates (#681): it is requested from + /// therefore uses the same request start as the raw-evaluated aggregates (#681): it is requested from /// (1 ms early) and only rows /// are kept — the decision is by TIMESTAMP, because a stored NULL - /// sample and the no-data boundary row look the same on the wire (both carry a value 0 with a Bad quality). + /// sample and the no-data boundary row look the same on the wire (both carry a value 0 with a Bad quality). The + /// forward end is exclusive (the interface contract; the gateway's own end is inclusive), a point read inclusive. /// /// - /// The gateway is asked for extra rows so the boundary row - /// (and any stored sample inside the margin) cannot shorten a page: a full page still holds exactly the cap of - /// stored samples, so the server's "full page ⇒ continuation point" rule is unchanged. If the stream delivers its - /// whole budget yet fewer than the cap belong to the window, the page cannot be read completely; answering a short - /// page would end the client's paging early, so the read fails with - /// instead. The stream is always consumed to completion (never - /// broken out of — a cancelled gateway stream evicts its pooled historian session). ReturnBounds is not - /// plumbed to the data source on any backend, so bounding values are never synthesised here. + /// Row budget. 0 ("all values") asks for (the client package and the + /// gateway reject max_values <= 0, #686). A cap asks for cap + + /// rows — NOT clamped at the gateway's default ceiling (#688 + /// clamps only the per-sample aggregate budget): AVEVA always emits the boundary row, so a clamped budget would make + /// every page size within the allowance of the ceiling come back short. A strict gateway refuses an over-ceiling + /// budget instead (⇒ , BadTooManyOperations). + /// + /// + /// Completeness. A full page must hold exactly the cap of stored samples — a short page ends the client's + /// paging. A read is complete when the stream came back short, filled the cap, or reached the window end. If rows + /// BEFORE the window (the boundary row and stored samples inside the 1 ms margin, e.g. a tie cluster at + /// start − 1 ms) used up the budget, the window is read ONCE more asking for cap + m + + /// (m = the pre-window rows the first read saw); only if that + /// re-read is still short because of pre-window rows does the read fail with + /// . Every stream is consumed to completion (never broken out of — + /// a cancelled gateway stream evicts its pooled historian session). ReturnBounds is not plumbed to the + /// data source on any backend, so bounding values are never synthesised here. /// /// private async Task> ReadRawWindowAsync( string fullReference, DateTime startUtc, DateTime endUtc, uint maxValuesPerNode, CancellationToken cancellationToken) { - // The gateway seam caps with an int; OPC UA hands us a uint, so clamp to int range. 0 ("all values") is sent as - // int.MaxValue: the client package and the gateway reject max_values <= 0 (#686 review). + // The gateway seam caps with an int; OPC UA hands us a uint, so clamp to int range. var cap = (int)Math.Min(maxValuesPerNode, int.MaxValue); - var budget = cap == 0 ? int.MaxValue : (int)Math.Min(int.MaxValue, (long)cap + RawAggregateReader.PreWindowAllowance); + if (cap == 0) + return (await ReadRawPassAsync(fullReference, startUtc, endUtc, cap, int.MaxValue, cancellationToken) + .ConfigureAwait(false)).Samples; + + var first = await ReadRawPassAsync(fullReference, startUtc, endUtc, cap, + (int)Math.Min(int.MaxValue, (long)cap + RawAggregateReader.PreWindowAllowance), cancellationToken).ConfigureAwait(false); + if (first.Complete) return first.Samples; + + // The margin crowded the window out: one bounded re-read with room for the pre-window rows already seen. + var second = await ReadRawPassAsync(fullReference, startUtc, endUtc, cap, + (int)Math.Min(int.MaxValue, (long)cap + first.PreWindowRows + RawAggregateReader.PreWindowAllowance), + cancellationToken).ConfigureAwait(false); + if (second.Complete) return second.Samples; + + throw new HistoryReadLimitExceededException(cap, + $"The raw read of '{fullReference}' could not be read completely: more than {second.PreWindowRows} rows lie in " + + $"the {RawAggregateReader.BoundaryMargin.TotalMilliseconds} ms before the window and exhausted the row budget."); + } + + /// One forward raw read of a window, drained, with its pre-window row count and whether it is complete. + private async Task<(List Samples, int PreWindowRows, bool Complete)> ReadRawPassAsync( + string fullReference, DateTime startUtc, DateTime endUtc, int cap, int budget, CancellationToken cancellationToken) + { var samples = new List(); long received = 0; + var preWindow = 0; + var reachedEnd = false; await foreach (var sample in _client .ReadRawAsync(fullReference, RawAggregateReader.RequestStartFor(startUtc), endUtc, budget, cancellationToken) .ConfigureAwait(false)) { received++; // Filter while draining — the rest of the (bounded) stream is consumed, never abandoned. - if (RawAggregateReader.IsWindowSample(sample, startUtc) && (cap == 0 || samples.Count < cap)) - samples.Add(sample); + if (RawAggregateReader.IsWindowSample(sample, startUtc, endUtc)) + { + if (cap == 0 || samples.Count < cap) samples.Add(sample); + } + else if (sample.Timestamp is { } ts && ts.ToDateTime() < startUtc) preWindow++; + else if (sample.Timestamp is not null) reachedEnd = true; } - if (cap > 0 && received >= budget && samples.Count < cap) - throw new HistoryReadLimitExceededException(cap, - $"The raw read of '{fullReference}' could not be read completely: more than " + - $"{RawAggregateReader.PreWindowAllowance} rows lie in the {RawAggregateReader.BoundaryMargin.TotalMilliseconds} ms " + - "before the window and exhausted the row budget."); - return samples; + var complete = received < budget || samples.Count >= cap || reachedEnd || cap == 0; + return (samples, preWindow, complete); } /// diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/RawAggregateReader.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/RawAggregateReader.cs index 484b56c4..1723cb1d 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/RawAggregateReader.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/RawAggregateReader.cs @@ -80,25 +80,32 @@ internal static class RawAggregateReader /// earlier, so the AVEVA start-boundary row — stamped exactly at the QUERY start, carrying the value in effect there /// (Quality = 133) or, with no earlier data, a NULL row (value 0, Bad OPC quality) — lands BEFORE the window /// (). The one place both the raw-evaluated aggregates (#681) and a raw HistoryRead - /// (#687) decide what the request start is. No margin when the window starts within one margin of - /// . + /// (#687) decide what the request start is. Never before (the Unix epoch — the start of + /// the historian's time encoding, the same floor the server's reverse walk-back uses, #686): a window starting less + /// than one margin after it is requested from the epoch itself. /// /// The window's (inclusive) start. /// The start to request. public static DateTime RequestStartFor(DateTime windowStartUtc) => - windowStartUtc.Ticks > BoundaryMargin.Ticks ? windowStartUtc - BoundaryMargin : windowStartUtc; + windowStartUtc - ProbeFloor > BoundaryMargin ? windowStartUtc - BoundaryMargin + : windowStartUtc > ProbeFloor ? ProbeFloor + : windowStartUtc; /// - /// True when a row of a stream requested from is a stored sample of the window that - /// begins at : it is stamped at or after the window start. A row before it is the - /// AVEVA boundary row (whatever its value or quality — a stored NULL looks the same on the wire, so the timestamp is - /// the evidence) or a stored sample inside the margin; a row without a timestamp is never a sample. + /// True when a row of a forward raw stream requested from is a stored sample of the raw + /// window [windowStartUtc, windowEndUtc) — the contract + /// (exclusive end, as on the SQL backends; the gateway's own ReadRaw end is inclusive) — or, for a point read + /// (start == end, the server's tie-cluster over-fetch), of that one instant. A row before the start is the AVEVA + /// boundary row (whatever its value or quality — a stored NULL looks the same on the wire, so the timestamp is the + /// evidence) or a stored sample inside the margin; a row without a timestamp is never a sample. /// /// The gateway row. /// The window's (inclusive) start. + /// The window's end: exclusive, or inclusive when it equals the start. /// Whether the row belongs to the window. - public static bool IsWindowSample(HistorianSample raw, DateTime windowStartUtc) => - raw.Timestamp is { } ts && ts.ToDateTime() >= windowStartUtc; + public static bool IsWindowSample(HistorianSample raw, DateTime windowStartUtc, DateTime windowEndUtc) => + raw.Timestamp is { } ts && ts.ToDateTime() is var t && t >= windowStartUtc + && (t < windowEndUtc || (t == windowEndUtc && windowStartUtc == windowEndUtc)); /// True for the aggregates this reader serves: the per-sample families of . /// Requested aggregate. diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayRawBoundaryRowTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayRawBoundaryRowTests.cs index 4a95d364..11d62de9 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayRawBoundaryRowTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayRawBoundaryRowTests.cs @@ -125,14 +125,6 @@ public sealed class GatewayRawBoundaryRowTests fake.RawCalls.Single().ShouldBe((At(30) - RawAggregateReader.BoundaryMargin, At(55), 5 + RawAggregateReader.PreWindowAllowance)); } - [Fact] - public async Task An_unlimited_read_stays_unlimited() - { - var fake = Gate(); - await Source(fake).ReadRawAsync("T", At(30), At(55), 0, Ct); - fake.RawCalls.Single().MaxValues.ShouldBe(0); - } - [Fact] public async Task A_full_page_holds_exactly_the_cap_and_the_stream_is_drained_not_abandoned() { @@ -148,13 +140,19 @@ public sealed class GatewayRawBoundaryRowTests [InlineData(3u)] public async Task Server_style_paging_returns_every_stored_sample_once_and_nothing_else(uint cap) { - // Ties at +60 (three samples) and a stored NULL at +40, over the AVEVA-shaped wire. - var data = Gate681.Select(x => (At(x.Sec), x.Value, x.Null)) - .Concat([(At(60), 2d, false), (At(60), 3d, false)]).ToList(); - var fake = Aveva(data); - var ds = Source(fake); + // Ties at +60 (three samples), a stored NULL at +40, a 20-sample tie cluster 1 ms before the +60 resume point + // (crowds that resume read's margin: more than the 16-row allowance) and one 1 ms before the window start (outside + // the window; crowds the first read's margin), over the AVEVA-shaped wire. var start = At(-10); var end = At(200); + var inside = Gate681.Select(x => (At(x.Sec), x.Value, x.Null)) + .Concat([(At(60), 2d, false), (At(60), 3d, false)]) + .Concat(Enumerable.Range(0, 20).Select(i => (At(60) - TimeSpan.FromMilliseconds(1), 100d + i, false))) + .ToList(); + var outside = Enumerable.Range(0, 20).Select(i => (start - TimeSpan.FromMilliseconds(1), 200d + i, false)); + var data = inside; + var fake = Aveva(inside.Concat(outside).ToList()); + var ds = Source(fake); var got = new List(); var skip = 0; var cursor = start; @@ -195,15 +193,103 @@ public sealed class GatewayRawBoundaryRowTests } [Fact] - public async Task A_margin_that_swallows_the_row_budget_fails_the_node_rather_than_answering_a_short_page() + public async Task A_margin_that_swallows_the_row_budget_is_re_read_once_with_room_for_it() { - // More stored rows inside the 1 ms margin than the allowance: the page could not be read, and a short page would - // end the client's paging early — refuse it (a request limit, not a backend fault). - var margin = Enumerable.Range(0, 40).Select(i => (At(30).AddTicks(-1 - i), (double)i, false)); + // 30 stored rows inside the 1 ms margin (plus AVEVA's boundary row) eat the whole first budget (5 + 16): the page is + // re-read ONCE asking for cap + m + 16 (m = the pre-window rows the first read saw), which fits, and the full page + // is answered — a short page would end the client's paging early. + var margin = Enumerable.Range(0, 30).Select(i => (At(30).AddTicks(-1 - i), (double)i, false)); + var window = Enumerable.Range(0, 10).Select(i => (At(31 + i), 1000d + i, false)); + var fake = Aveva(margin.Concat(window).ToList()); + var ds = Source(fake); + + var r = await ds.ReadRawAsync("T", At(30), At(50), 5, Ct); + + r.Samples.Select(s => (double)s.Value!).ShouldBe([1000d, 1001d, 1002d, 1003d, 1004d]); + fake.RawCalls.Select(c => c.MaxValues).ShouldBe([5 + 16, 5 + 21 + 16]); + fake.RawCalls.ShouldAllBe(c => c.Start == At(30) - RawAggregateReader.BoundaryMargin); + fake.RawStreamsCompleted.ShouldBe(2); + ds.GetHealthSnapshot().TotalFailures.ShouldBe(0); + } + + [Fact] + public async Task A_margin_still_crowding_the_re_read_fails_the_node_as_a_request_limit() + { + // 200 rows in the margin: the single bounded re-read (5 + 21 + 16) is still short of the cap because of them. + var margin = Enumerable.Range(0, 200).Select(i => (At(30).AddTicks(-1 - i), (double)i, false)); var fake = Aveva(margin.Append((At(35), 1d, false)).ToList()); var ds = Source(fake); + await Should.ThrowAsync(() => ds.ReadRawAsync("T", At(30), At(40), 5, Ct)); + + fake.ReadRawCallCount.ShouldBe(2); // one bounded re-read, never more + fake.RawStreamsCompleted.ShouldBe(2); ds.GetHealthSnapshot().TotalFailures.ShouldBe(0); - fake.RawStreamsCompleted.ShouldBe(1); + } + + [Fact] + public async Task A_short_read_is_complete_and_is_not_re_read() + { + var fake = Gate(); + await Source(fake).ReadRawAsync("T", At(30), At(55), 5, Ct); + fake.ReadRawCallCount.ShouldBe(1); + } + + [Fact] + public async Task A_read_that_reached_past_the_window_end_is_complete_and_is_not_re_read() + { + // Budget 3 + 16 = 19 = boundary + 16 margin rows + the one window row + the row stored at the (exclusive) end: + // the stream is used up and the page is short, but it reached the end — every window row was seen. + var margin = Enumerable.Range(0, 16).Select(i => (At(30).AddTicks(-1 - i), (double)i, false)); + var fake = Aveva(margin.Append((At(35), 1d, false)).Append((At(40), 2d, false)).ToList()); + var ds = Source(fake); + + var r = await ds.ReadRawAsync("T", At(30), At(40), 3, Ct); + + r.Samples.Select(s => (double)s.Value!).ShouldBe([1d]); + fake.RawYieldedCount.ShouldBe(19); + fake.ReadRawCallCount.ShouldBe(1); + } + + [Theory] // M1: the raw budget is NOT clamped at the gateway's default ceiling — AVEVA always emits a boundary row, so a + // page size within 16 of the ceiling would always come back short (a strict gateway refuses instead ⇒ + // BadTooManyOperations, a default gateway serves it). + [InlineData(999_984u, 1_000_000)] + [InlineData(999_990u, 1_000_006)] + [InlineData(1_000_000u, 1_000_016)] + public async Task The_raw_row_budget_is_not_clamped_at_the_gateways_default_ceiling(uint cap, int expected) + { + var fake = Gate(); + await Source(fake).ReadRawAsync("T", At(30), At(55), cap, Ct); + fake.RawCalls.Single().MaxValues.ShouldBe(expected); + } + + // ---- L3: the forward end bound is exclusive; a point read is inclusive ---------------------------------------- + + [Fact] + public async Task A_stored_sample_exactly_at_the_window_end_is_not_returned() + { + var r = await Source(Gate()).ReadRawAsync("T", At(0), At(20), 0, Ct); + r.Samples.Select(s => s.SourceTimestampUtc).ShouldBe(new DateTime?[] { At(0), At(5) }); + } + + [Fact] + public async Task A_point_read_returns_the_samples_stored_at_that_instant() + { + var data = new List<(DateTime, double, bool)> { (At(10), 1, false), (At(20), 2, false), (At(20), 3, false), (At(30), 4, false) }; + var r = await Source(Aveva(data)).ReadRawAsync("T", At(20), At(20), 1000, Ct); + r.Samples.Select(s => (double)s.Value!).ShouldBe([2d, 3d]); + } + + // ---- the Unix-epoch floor ---------------------------------------------------------------------------------------- + + [Theory] // the gateway's time encoding starts at the Unix epoch: the 1 ms margin never reaches before it + [InlineData(0L)] + [InlineData(5_000L)] + public void The_request_start_never_precedes_the_Unix_epoch(long ticksAfterEpoch) + { + var start = DateTime.UnixEpoch.AddTicks(ticksAfterEpoch); + RawAggregateReader.RequestStartFor(start).ShouldBe(DateTime.UnixEpoch); + RawAggregateReader.RequestStartFor(At(0)).ShouldBe(At(0) - RawAggregateReader.BoundaryMargin); } } From a3ee917fb4ee4100537dd5e47deb80634a45f4c3 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Mon, 12 Oct 2026 15:47:39 -0400 Subject: [PATCH 5/5] =?UTF-8?q?fix(historian):=20#687=20re-review=20?= =?UTF-8?q?=E2=80=94=20a=20row=20at/before=20the=20Unix=20epoch=20is=20nev?= =?UTF-8?q?er=20a=20stored=20sample?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit With StartTime at or before the epoch, RequestStartFor has no margin left, so AVEVA's synthetic no-data/boundary row is stamped exactly at StartTime and was kept: raw reads returned a fabricated Bad sample at 1970-01-01 and the per-sample aggregates put it in bucket 0 (Start/End fabricated, Min/Max status Uncertain). IsWindowSample and the aggregate loop now drop any row with t <= ProbeFloor (RawAggregateReader.IsAfterFloor), matching the #686 walk-back's assumption. --- docs/Historian.md | 4 +++- .../RawAggregateReader.cs | 12 +++++++++++- .../GatewayRawAggregateTests.cs | 17 +++++++++++++++++ .../GatewayRawBoundaryRowTests.cs | 12 ++++++++++++ 4 files changed, 43 insertions(+), 2 deletions(-) diff --git a/docs/Historian.md b/docs/Historian.md index 64363ba9..39a894af 100644 --- a/docs/Historian.md +++ b/docs/Historian.md @@ -686,7 +686,9 @@ data. the gateway is asked from **1 ms before** the window (never before the Unix epoch, the start of the historian's time encoding), so the boundary row lands before it, and only rows `RawAggregateReader.IsWindowSample` are returned. The decision is by **timestamp**, not by value or quality — a stored NULL sample looks exactly like the no-data boundary row on the wire (value 0, - Bad), and a stored NULL inside the window is still returned (Bad), as it is a stored sample. + Bad), and a stored NULL inside the window is still returned (Bad), as it is a stored sample. A row stamped **at or before the + epoch** is never a stored sample (AVEVA stores nothing there — the #686 walk-back's floor too): a window starting at or before + the epoch gets no margin, so AVEVA's no-data row lands exactly at its start; raw reads and the per-sample aggregates both drop it. - **The forward end is exclusive** (`[StartTime, EndTime)`, the `IHistorianDataSource` contract and what the SQL backends return; the gateway's own `ReadRaw` end is inclusive, so a sample stored exactly at EndTime used to be returned). A point read (`StartTime == EndTime` — the server's tie-cluster over-fetch) still returns the samples at that instant. The #686 reverse diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/RawAggregateReader.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/RawAggregateReader.cs index 1723cb1d..a9f95808 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/RawAggregateReader.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/RawAggregateReader.cs @@ -104,9 +104,18 @@ internal static class RawAggregateReader /// The window's end: exclusive, or inclusive when it equals the start. /// Whether the row belongs to the window. public static bool IsWindowSample(HistorianSample raw, DateTime windowStartUtc, DateTime windowEndUtc) => - raw.Timestamp is { } ts && ts.ToDateTime() is var t && t >= windowStartUtc + raw.Timestamp is { } ts && ts.ToDateTime() is var t && IsAfterFloor(t) && t >= windowStartUtc && (t < windowEndUtc || (t == windowEndUtc && windowStartUtc == windowEndUtc)); + /// + /// A row stamped at or before (the Unix epoch) is never a stored sample: AVEVA stores nothing + /// there (the server's #686 walk-back assumes the same), and a window starting at or before the epoch gets no 1 ms + /// margin (), so AVEVA's synthetic boundary / no-data row lands exactly at its start. + /// + /// The row's timestamp. + /// Whether the row can be a stored sample. + internal static bool IsAfterFloor(DateTime t) => t > ProbeFloor; + /// True for the aggregates this reader serves: the per-sample families of . /// Requested aggregate. /// Whether the aggregate is evaluated from raw samples. @@ -184,6 +193,7 @@ internal static class RawAggregateReader if (overCap || raw.Timestamp is null) continue; var s = ToStored(raw); var t = s.TimeUtc; + if (!IsAfterFloor(t)) continue; // the boundary row of a window starting at/before the epoch, never a sample if (plan.IndexOf(t) is { } bucket) { if (maxWindowRows > 0 && ++windowRows > maxWindowRows) diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayRawAggregateTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayRawAggregateTests.cs index b99f2b57..d1759e11 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayRawAggregateTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayRawAggregateTests.cs @@ -473,6 +473,23 @@ public sealed class GatewayRawAggregateTests ds2.GetHealthSnapshot().TotalFailures.ShouldBe(1); } + [Theory] // a window starting AT the epoch gets no 1 ms margin, so AVEVA's no-data row is stamped at the window start: + // it must not become a stored Bad sample of bucket 0 + [InlineData(HistoryAggregateType.Start)] + [InlineData(HistoryAggregateType.Count)] + [InlineData(HistoryAggregateType.Minimum)] + public async Task A_window_starting_at_the_epoch_does_not_evaluate_the_fabricated_row(HistoryAggregateType agg) + { + var epoch = DateTime.UnixEpoch; + var data = new[] { new StoredSample(epoch.AddSeconds(5), HistorianValue.FromDouble(7), Good) }; + var fake = Store(data, avevaBoundary: true); + var s = (await Source(fake).ReadProcessedAsync("T", epoch, epoch.AddSeconds(20), TimeSpan.FromSeconds(20), agg, Ct)) + .Samples.Single(); + var expected = SqlBackendAnswer(agg, BucketPlan.Create(epoch, epoch.AddSeconds(20), TimeSpan.FromSeconds(20)), data).Single(); + (s.Value, s.StatusCode).ShouldBe((expected.Value, expected.StatusCode)); + fake.RawCalls[0].Start.ShouldBe(epoch); // the request start cannot go before the epoch + } + [Fact] // #688: the default cap's budget fits the gateway's default ceiling, so a strict gateway serves the default config public async Task The_default_caps_row_budget_does_not_exceed_the_gateways_default_ceiling() { diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayRawBoundaryRowTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayRawBoundaryRowTests.cs index 11d62de9..2aec396e 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayRawBoundaryRowTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayRawBoundaryRowTests.cs @@ -292,4 +292,16 @@ public sealed class GatewayRawBoundaryRowTests RawAggregateReader.RequestStartFor(start).ShouldBe(DateTime.UnixEpoch); RawAggregateReader.RequestStartFor(At(0)).ShouldBe(At(0) - RawAggregateReader.BoundaryMargin); } + + [Theory] // with no margin left before the epoch, AVEVA's no-data row lands AT StartTime: a row at/before the epoch is + // never a stored sample (AVEVA stores nothing there — the #686 walk-back's floor too) + [InlineData(0L)] + [InlineData(-10_000_000L)] + public async Task A_read_starting_at_or_before_the_epoch_does_not_return_the_fabricated_row(long startTicksFromEpoch) + { + var epoch = DateTime.UnixEpoch; + var fake = Aveva([(epoch.AddSeconds(10), 7d, false)]); + var r = await Source(fake).ReadRawAsync("T", epoch.AddTicks(startTicksFromEpoch), epoch.AddSeconds(60), 5, Ct); + r.Samples.Select(s => s.SourceTimestampUtc).ShouldBe(new DateTime?[] { epoch.AddSeconds(10) }); + } }