From de8bc7a12cdf1fa6135ab92a2ebaff3415d95a68 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Tue, 13 Oct 2026 01:53:14 -0400 Subject: [PATCH] fix(historian): #694 R2 cap the crowding re-read room at MaxTieClusterOverfetch The #687/F4 crowding re-read doubled its room up to 8 re-reads with no cap, so a pathological single-ms cluster on a streaming gateway was streamed several times over for one page. The room is now capped at ServerHistorian:MaxTieClusterOverfetch (plumbed into GatewayHistorianDataSource as maxPreWindowRows, default 65536); past it, or once a read has already seen more, the read is BadTooManyOperations with PreWindowRows set. --- docs/Historian.md | 7 ++-- .../GatewayHistorianDataSource.cs | 29 +++++++++++++--- ...wayHistorianServiceCollectionExtensions.cs | 4 ++- .../GatewayRawBoundaryRowTests.cs | 34 +++++++++++++++++++ .../HistorianGatewayClientAdapterTests.cs | 13 +++++++ 5 files changed, 80 insertions(+), 7 deletions(-) diff --git a/docs/Historian.md b/docs/Historian.md index a43378a0..6c774fa9 100644 --- a/docs/Historian.md +++ b/docs/Historian.md @@ -146,7 +146,7 @@ Processed read is `BadAggregateNotSupported` and no history aggregate functions | `AllowUntrustedServerCertificate` | bool | `false` | **Gateway backend only.** Accept a self-signed / untrusted server certificate (dev / on-prem only). | | `CaCertificatePath` | string\|null | `null` | **Gateway backend only.** PEM CA file pinning the gateway's TLS chain. Null/empty uses the OS trust store. | | `CallTimeout` | TimeSpan | `00:00:30` | **Gateway backend only.** Per-call deadline applied to each unary gateway read. A read for a tag the historian does not yet know about — e.g. a freshly-historized tag that has **not** been provisioned (see [Tag auto-provisioning](#tag-auto-provisioning-ensuretags)) — can run to this **full** deadline before the gateway errors, so an unprovisioned-tag HistoryRead can block for ~30 s. Lower this to fail faster, at the cost of truncating legitimately long reads. | -| `MaxTieClusterOverfetch` | int | `65536` | (Node-manager paging; all backends — a `SqlServer` PK rules out tie clusters, so it is moot there.) Maximum samples the server will fetch in one shot to page through a tie cluster (multiple samples sharing one `SourceTimestamp`). A cluster larger than this ceiling fails `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. | +| `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`. Also (Gateway, #694 R2) the most rows inside the 1 ms look-back before a raw window that a crowding re-read makes room for. Values below 1 count as 1. Raise to handle abnormally large tie clusters; the default covers all normal-data cases. | | `HonorReturnBounds` | bool | `true` | (#691, all backends.) Whether HistoryRead-Raw honours `ReturnBounds` (Part 11 bounding values, [Bounding values](#bounding-values-returnbounds-691)). `false` is the kill switch back to the pre-#691 behaviour: the flag is ignored (stored samples only) and a modified read carrying it is not refused `Bad_InvalidArgument`. Note the .NET SDK defaults both `ReturnBounds` and `IsReadModified` to `true`. | | `GatewayAtTimeFromRaw` | bool | `true` | (#693, **Gateway backend only.**) Whether HistoryRead-AtTime is served from raw gateway reads evaluated with the SQL backends' AtTime rules ([Gateway AtTime](#gateway-attime-is-evaluated-from-raw-reads-693)). `false` is the kill switch back to the pre-#693 answer: AVEVA's own interpolation (the gateway's `ReadAtTime`) passed through unchanged. | | `EventReadHorizon` | TimeSpan | `00:00:00` | (#694, all backends, **opt-in, lossy**.) `00:00:00` = no horizon: every HistoryReadEvents read is exact to its own window. A positive span raises the earlier bound of every fresh events window to `now − EventReadHorizon` (the floor instant itself included), in **both** directions — a window wholly older is `GoodNoData` and is not sent; a page-through keeps the first page's floor. It exists for the Gateway, where a walked reverse / end-only read proves "nothing older" down to the Unix epoch (~3.8 s live; an unfiltered one can hit the gateway's 30 s SQL timeout) and a forward read from a far-past `StartTime` scans the same range. ⚠️ While set, events older than the horizon are **invisible** to HistoryReadEvents. Raw/Processed/AtTime are unaffected. Negative ⇒ treated as none (a `Validate()` warning). See [Continuation-point paging (Events)](#continuation-point-paging-events-669). | @@ -812,7 +812,10 @@ data. 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 with room for them: `cap + room + 16`, room = the pre-window rows the first read saw, at least **doubling** per re-read while a read that was itself cut short keeps finding more (up to 8 re-reads, never past a known strict ceiling — review F4: a single re-read sized from - a truncated read's count failed a 300–1,700-row cluster just before a resume point even on a default gateway). Only if the + a truncated read's count failed a 300–1,700-row cluster just before a resume point even on a default gateway). The room is + **capped at `MaxTieClusterOverfetch`** (#694 R2, the limit every other tie path uses): more rows than that inside the look-back ⇒ + `BadTooManyOperations` after a bounded number of bounded reads — uncapped, a pathological million-row single-millisecond cluster + on a streaming gateway would have been streamed several times over for one page. Only if the re-reads are still crowded 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 served by the server, not by this row** (#691, [Bounding values](#bounding-values-returnbounds-691)). The AVEVA 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 19a2b24e..f4cc345d 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 @@ -54,6 +54,19 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis /// The most #687 crowding re-reads one raw read makes (#694 F4): each at least doubles the pre-window room. private const int MaxCrowdedReReads = 8; + /// The default of maxPreWindowRows: ServerHistorian:MaxTieClusterOverfetch's default. + public const int DefaultMaxPreWindowRows = 65536; + + /// + /// The most rows before a raw window (inside the 1 ms look-back) a crowding re-read makes room for (#694 R2) — the + /// server's tie-cluster limit (ServerHistorian:MaxTieClusterOverfetch), which every other tie path treats as + /// the limit. More ⇒ with PreWindowRows set. + /// + private readonly int _maxPreWindowRows; + + /// The tie-cluster limit the crowding re-reads honour (#694 R2) — exposed so the factory wiring is testable. + internal int MaxPreWindowRows => _maxPreWindowRows; + /// A strict gateway's row ceiling, learned from its first refusal (#691.3). private readonly GatewayRowCeiling _ceiling = new(); @@ -78,10 +91,15 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis /// #693: serve AtTime from raw reads with the shared AtTime rules (; default); /// false = AVEVA's own interpolation passed through (ServerHistorian:GatewayAtTimeFromRaw kill switch). /// + /// + /// The most rows before a raw window a crowding re-read makes room for (#694 R2; ServerHistorian:MaxTieClusterOverfetch, + /// values below 1 count as 1). More ⇒ (BadTooManyOperations). + /// public GatewayHistorianDataSource(IHistorianGatewayClient client, ILogger logger, int maxRawRows = TimeWeightedAggregator.DefaultMaxWindowRows, TimeProvider? timeProvider = null, - bool atTimeFromRaw = true) + bool atTimeFromRaw = true, int maxPreWindowRows = DefaultMaxPreWindowRows) { + _maxPreWindowRows = Math.Max(1, maxPreWindowRows); _atTimeFromRaw = atTimeFromRaw; ArgumentNullException.ThrowIfNull(client); ArgumentNullException.ThrowIfNull(logger); @@ -214,9 +232,11 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis // The margin crowded the window out: re-read with room for the pre-window rows. A read that was itself cut short saw // only some of them (#694 F4: a tie cluster of hundreds just before a resume point used to fail even with no // ceiling), so the room at least doubles per re-read — bounded by MaxCrowdedReReads, and never past a known ceiling. + // #694 R2: the room is capped at the tie-cluster limit (MaxTieClusterOverfetch) — a pathological single-millisecond + // cluster on a streaming gateway would otherwise be streamed several times over for one page. var preWindow = pass.Result.PreWindowRows; - var room = Math.Max(1, preWindow); - for (var attempt = 0; attempt < MaxCrowdedReReads; attempt++) + var room = Math.Min(Math.Max(1, preWindow), _maxPreWindowRows); + for (var attempt = 0; attempt < MaxCrowdedReReads && preWindow <= _maxPreWindowRows; attempt++) { var budget = (int)Math.Min(int.MaxValue, (long)cap + room + RawAggregateReader.PreWindowAllowance); if (_ceiling.Learned is var known and > 0 && budget > known) break; // over a strict ceiling: refused anyway @@ -224,7 +244,8 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis if (pass.Result.Complete) return pass.Result.Samples; preWindow = Math.Max(preWindow, pass.Result.PreWindowRows); if (pass.Budget < budget) break; // bounded by a ceiling learned meanwhile - room = (int)Math.Min(int.MaxValue / 2, Math.Max(preWindow, 2L * room)); + if (room >= _maxPreWindowRows) break; // the limit's worth of room did not suffice + room = (int)Math.Min(_maxPreWindowRows, Math.Max(preWindow, 2L * room)); } throw new HistoryReadLimitExceededException(cap, diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianServiceCollectionExtensions.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianServiceCollectionExtensions.cs index 67e91ad6..dc74b10b 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianServiceCollectionExtensions.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianServiceCollectionExtensions.cs @@ -42,7 +42,9 @@ public static class GatewayHistorian // #681: the raw-evaluated per-sample aggregates share the SQL backends' raw-row cap. maxRawRows: options.MaxTimeWeightedRawRows, // #693: AtTime from raw reads + the shared evaluator; false = AVEVA's own interpolation (kill switch). - atTimeFromRaw: options.GatewayAtTimeFromRaw); + atTimeFromRaw: options.GatewayAtTimeFromRaw, + // #694 R2: a crowding re-read never makes room for more pre-window rows than the tie-cluster limit. + maxPreWindowRows: options.MaxTieClusterOverfetch); } /// 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 97da1a91..2913114f 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 @@ -255,6 +255,40 @@ public sealed class GatewayRawBoundaryRowTests ds.GetHealthSnapshot().TotalFailures.ShouldBe(0); } + [Fact] + public async Task The_crowding_re_reads_never_make_room_for_more_than_the_tie_cluster_limit() + { + // #694 R2: the room used to double up to 8 re-reads with no cap — on a streaming gateway a pathological 1M-row + // single-millisecond cluster streamed ~4M rows for one page where it used to fail at once. The room is capped at + // MaxTieClusterOverfetch (here 100); past it the read is a request limit, after a bounded number of bounded reads. + var margin = Enumerable.Range(0, 150).Select(i => (At(30).AddTicks(-1 - i), (double)i, false)); + var fake = Aveva(margin.Append((At(35), 1d, false)).ToList()); + var ds = new GatewayHistorianDataSource(fake, NullLogger.Instance, maxPreWindowRows: 100); + + var ex = await Should.ThrowAsync(() => ds.ReadRawAsync("T", At(30), At(40), 5, Ct)); + + ex.PreWindowRows.ShouldBeGreaterThan(100); + // Room 21, 42, 84: that read already saw more than 100 rows before the window, so it stops — never asking past the cap. + fake.ReadRawCallCount.ShouldBe(4); + fake.RawCalls.Max(c => c.MaxValues).ShouldBeLessThanOrEqualTo(5 + 100 + RawAggregateReader.PreWindowAllowance); + fake.RawYieldedCount.ShouldBeLessThanOrEqualTo(21 + 42 + 63 + 105); + fake.RawStreamsCompleted.ShouldBe(4); + ds.GetHealthSnapshot().TotalFailures.ShouldBe(0); + } + + [Fact] + public async Task Crowding_rows_just_under_the_tie_cluster_limit_are_read_around() + { + var margin = Enumerable.Range(0, 90).Select(i => (At(30).AddTicks(-1 - i), (double)i, false)); + var fake = Aveva(margin.Append((At(35), 1d, false)).ToList()); + var ds = new GatewayHistorianDataSource(fake, NullLogger.Instance, maxPreWindowRows: 100); + + var r = await ds.ReadRawAsync("T", At(30), At(40), 5, Ct); + + r.Samples.Select(s => s.Value).ShouldBe([1d]); + fake.RawCalls.Max(c => c.MaxValues).ShouldBeLessThanOrEqualTo(5 + 100 + RawAggregateReader.PreWindowAllowance); + } + [Fact] public async Task A_short_read_is_complete_and_is_not_re_read() { diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/HistorianGatewayClientAdapterTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/HistorianGatewayClientAdapterTests.cs index 7c8bdd60..ff441a47 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/HistorianGatewayClientAdapterTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/HistorianGatewayClientAdapterTests.cs @@ -131,4 +131,17 @@ public sealed class HistorianGatewayClientAdapterTests var dataSource = Assert.IsType(GatewayHistorian.CreateDataSource(opts, services)); dataSource.AtTimeFromRaw.ShouldBe(expected); } + + [Theory] // #694 R2: ServerHistorian:MaxTieClusterOverfetch bounds the crowding re-reads (default 65536) + [InlineData(null, 65536)] + [InlineData(1000, 1000)] + public void Factory_passes_the_tie_cluster_limit_to_the_crowding_re_reads(int? configured, int expected) + { + var opts = configured is { } c + ? new ServerHistorianOptions { Enabled = true, Endpoint = "https://localhost:5222", ApiKey = "histgw_x_y", MaxTieClusterOverfetch = c } + : new ServerHistorianOptions { Enabled = true, Endpoint = "https://localhost:5222", ApiKey = "histgw_x_y" }; + using var services = new ServiceCollection().BuildServiceProvider(); + var dataSource = Assert.IsType(GatewayHistorian.CreateDataSource(opts, services)); + dataSource.MaxPreWindowRows.ShouldBe(expected); + } }