fix(historian): #691.3 review — a ceiling-bound per-sample read names the gateway ceiling, not the cap

2026-10-12 16:47:57 -04:00
parent 0e47e887de
commit ab5030174a
2 changed files with 43 additions and 3 deletions
@@ -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
    /// </summary>
    private static bool IsNotBadOnTheWire(HistorianSample raw) => (byte)raw.OpcQuality >= 64; // low byte, as SampleMapper.ToStatus reads it

+    /// <summary>
+    ///     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.
+    /// </summary>
+    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 " +
@@ -182,4 +182,23 @@ public sealed class GatewayStrictCeilingTests
        await Should.ThrowAsync<HistoryReadLimitExceededException>(() => 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<HistoryReadLimitExceededException>(() => 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<HistoryReadLimitExceededException>(() => 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");
+    }
 }