From ab5030174a74aaff03bd28e08c726533fc8c6948 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Mon, 12 Oct 2026 16:47:57 -0400 Subject: [PATCH] =?UTF-8?q?fix(historian):=20#691.3=20review=20=E2=80=94?= =?UTF-8?q?=20a=20ceiling-bound=20per-sample=20read=20names=20the=20gatewa?= =?UTF-8?q?y=20ceiling,=20not=20the=20cap?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../RawAggregateReader.cs | 27 ++++++++++++++++--- .../GatewayStrictCeilingTests.cs | 19 +++++++++++++ 2 files changed, 43 insertions(+), 3 deletions(-) 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 c4ebff5c..cdc25858 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 @@ -184,11 +184,12 @@ internal static class RawAggregateReader // A strict gateway that refuses the budget is re-asked once at the ceiling it names (#691.3; e.g. a ceiling below // the 1,000,000 #688 clamps at, or an unbounded cap) — and a stream that fills that ceiling is never evaluated. - var (pass, maxValues) = await GatewayReadLimits.ReadWithCeilingRetryAsync(MaxValuesFor(maxWindowRows), + var requested = MaxValuesFor(maxWindowRows); + var (pass, maxValues) = await GatewayReadLimits.ReadWithCeilingRetryAsync(requested, budget => ReadWindowAsync(client, tag, aggregate, plan, requestStart, budget, maxWindowRows, ct)).ConfigureAwait(false); var (buckets, earliest, latest, received, overCap) = pass; - if (overCap || received >= maxValues) - throw LimitExceeded(tag, maxWindowRows); + if (overCap) throw LimitExceeded(tag, maxWindowRows); + if (received >= maxValues) throw BudgetExhausted(tag, maxWindowRows, requested, maxValues); if (needsRange) { @@ -281,6 +282,26 @@ internal static class RawAggregateReader /// private static bool IsNotBadOnTheWire(HistorianSample raw) => (byte)raw.OpcQuality >= 64; // low byte, as SampleMapper.ToStatus reads it + /// + /// The stream delivered its whole row budget, so the window cannot be proven complete. Names the limit that actually + /// bound the budget: the gateway's ceiling (a strict gateway refused the requested budget and named it, #691.3), the + /// #688 clamp at the gateway's default ceiling, or the configured cap. + /// + private static HistoryReadLimitExceededException BudgetExhausted(string tag, int maxWindowRows, int requested, int budget) + { + if (budget < requested) + return new HistoryReadLimitExceededException(budget, + $"The processed read of '{tag}' covers more raw samples than the HistorianGateway's row ceiling ({budget}, its " + + "Historian:ReadLimits:MaxValuesCeiling with AllowStreamingBeyondCeiling=false); the Gateway backend evaluates " + + "per-sample aggregates from raw reads — narrow the window."); + if (maxWindowRows > 0 && budget < (long)maxWindowRows + 2 + PreWindowAllowance) + return new HistoryReadLimitExceededException(budget, + $"The processed read of '{tag}' covers more raw samples than the HistorianGateway's default row ceiling " + + $"({budget}) the request is clamped at; the Gateway backend evaluates per-sample aggregates from raw reads — " + + "narrow the window."); + return LimitExceeded(tag, maxWindowRows); + } + private static HistoryReadLimitExceededException LimitExceeded(string tag, int maxWindowRows) => new(maxWindowRows, $"The processed read of '{tag}' covers more than {maxWindowRows} raw samples " + diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayStrictCeilingTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayStrictCeilingTests.cs index 825b2c77..1b20ed63 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayStrictCeilingTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayStrictCeilingTests.cs @@ -182,4 +182,23 @@ public sealed class GatewayStrictCeilingTests await Should.ThrowAsync(() => Source(fake, maxRawRows: 0).ReadProcessedAsync("T", At(0), At(100), TimeSpan.FromSeconds(10), HistoryAggregateType.Count, Ct)); } + + [Theory] // review: the message names the limit that actually bound the read — the gateway's ceiling, not the cap + [InlineData(0)] // unbounded cap: used to say "more than 0 raw samples (MaxTimeWeightedRawRows)" + [InlineData(1_000_000)] // default cap: used to cite 1,000,000 + public async Task A_ceiling_bound_per_sample_read_names_the_gateway_ceiling(int maxRawRows) + { + var ex = await Should.ThrowAsync(() => Source(Store(100, ceiling: 50), maxRawRows) + .ReadProcessedAsync("T", At(0), At(100), TimeSpan.FromSeconds(10), HistoryAggregateType.Count, Ct)); + ex.Message.ShouldContain("row ceiling (50,"); + ex.Message.ShouldNotContain("MaxTimeWeightedRawRows); the Gateway"); + } + + [Fact] + public async Task A_cap_bound_per_sample_read_still_names_the_cap() + { + var ex = await Should.ThrowAsync(() => Source(Store(100, ceiling: null), 20) + .ReadProcessedAsync("T", At(0), At(100), TimeSpan.FromSeconds(10), HistoryAggregateType.Count, Ct)); + ex.Message.ShouldContain("more than 20 raw samples"); + } }