fix(historian): #691.3 review — remember a strict gateway's ceiling per data source; re-probe once when a bounded read comes back full

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianDataSource.cs

@@ -47,6 +47,9 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
     private readonly int _maxRawRows;
     private readonly TimeProvider _time;
 
+    /// <summary>A strict gateway's row ceiling, learned from its first refusal (#691.3).</summary>
+    private readonly GatewayRowCeiling _ceiling = new();
+
     private readonly object _healthLock = new();
     private DateTime? _lastSuccessUtc;
     private DateTime? _lastFailureUtc;
@@ -204,8 +207,9 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
 
         // One drained pass; a strict gateway's refusal of the budget is retried once at the ceiling it names (#691.3).
         Task<((List<HistorianSample> Samples, int PreWindowRows, bool Complete) Result, int Budget)> PassAsync(int budget) =>
-            GatewayReadLimits.ReadWithCeilingRetryAsync(budget,
-                b => ReadRawPassAsync(fullReference, startUtc, endUtc, cap, b, cancellationToken));
+            _ceiling.ReadAsync(budget,
+                b => ReadRawPassAsync(fullReference, startUtc, endUtc, cap, b, cancellationToken),
+                (r, _) => !r.Complete);
     }
 
     /// <summary>One forward raw read of a window, drained, with its pre-window row count and whether it is complete.</summary>
@@ -349,7 +353,7 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
         {
             var now = _time.GetUtcNow().UtcDateTime;
             var values = await RawAggregateReader
-                .ReadAsync(_client, fullReference, aggregate, plan, _maxRawRows, now, cancellationToken)
+                .ReadAsync(_client, _ceiling, fullReference, aggregate, plan, _maxRawRows, now, cancellationToken)
                 .ConfigureAwait(false);
             RecordOutcome(success: true, error: null);
             return new HistoryReadResult(values, ContinuationPoint: null);

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayReadLimits.cs

@@ -12,10 +12,9 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway;
 ///     no RPC that reports the ceiling, so the refusal is how it is learned (#691.3).
 /// </summary>
 /// <remarks>
-///     A read whose budget the gateway refuses is re-issued ONCE with the ceiling as its budget
-///     (<see cref="ReadWithCeilingRetryAsync{T}"/>); the caller then treats a stream that delivered the whole ceiling as
-///     possibly truncated (never answered from). Nothing is cached: a re-configured gateway is honoured on the next read,
-///     and a refusal costs one round-trip with no historian work behind it (it is validated before a session is acquired).
+///     A read whose budget the gateway refuses is re-issued ONCE with the ceiling as its budget, and the ceiling is
+///     remembered per data source (<see cref="GatewayRowCeiling"/>); the caller treats a stream that delivered the whole
+///     ceiling as possibly truncated (never answered from).
 /// </remarks>
 internal static class GatewayReadLimits
 {
@@ -50,29 +49,6 @@ internal static class GatewayReadLimits
         return true;
     }
 
-    /// <summary>
-    ///     Runs <paramref name="pass"/> with <paramref name="budget"/>; if a strict gateway refuses that budget, runs it once
-    ///     more with the ceiling it named — only when that ceiling is smaller than the refused budget (otherwise the refusal
-    ///     is rethrown). The refusal arrives before any row, so the first pass has consumed nothing.
-    /// </summary>
-    /// <typeparam name="T">The pass result.</typeparam>
-    /// <param name="budget">The <c>max_values</c> to ask for first.</param>
-    /// <param name="pass">One drained read with the given budget.</param>
-    /// <returns>The pass result and the budget it actually used.</returns>
-    public static async Task<(T Result, int Budget)> ReadWithCeilingRetryAsync<T>(int budget, Func<int, Task<T>> pass)
-    {
-        int ceiling;
-        try
-        {
-            return (await pass(budget).ConfigureAwait(false), budget);
-        }
-        catch (Exception ex) when (TryGetCeiling(ex, out ceiling) && ceiling < budget)
-        {
-            // fall through to the retry (outside the catch, so a second refusal is not nested in this one)
-        }
-        return (await pass(ceiling).ConfigureAwait(false), ceiling);
-    }
-
     private static RpcException? Rejection(Exception ex)
     {
         for (var e = ex; e is not null; e = e.InnerException)
@@ -82,3 +58,78 @@ internal static class GatewayReadLimits
         return null;
     }
 }
+
+/// <summary>
+///     A strict gateway's row ceiling as learned by ONE data source (#691.3 review): every refused attempt counts as an
+///     error in the gateway's own outcome metrics, so the ceiling a refusal names is remembered and later reads ask for at
+///     most it — no refusal per read in the steady state. Thread-safe (one <see cref="int"/>, read/written atomically; a
+///     race costs at most one extra refusal or re-probe, never a wrong answer).
+/// </summary>
+/// <remarks>
+///     <para>
+///         <b>Rules.</b> (1) Unknown ceiling: ask for the full budget; a refusal naming <c>N</c> &lt; budget learns <c>N</c> and
+///         the read is re-issued once at <c>N</c>. (2) Learned ceiling <c>C</c> below the budget: ask for <c>C</c> — a refusal
+///         there (the gateway was LOWERED) re-learns its <c>N</c> as in (1). (3) If that <c>C</c>-bounded read came back
+///         full but not provably complete, the gateway may have been RAISED (or switched to streaming) since <c>C</c> was
+///         learned, and answering <c>BadTooManyOperations</c> from a stale ceiling would be wrong: the read is re-probed ONCE
+///         at the full budget — served ⇒ the ceiling is forgotten; refused naming a larger <c>N</c> ⇒ <c>N</c> is learned and
+///         the read is re-issued at <c>N</c>; refused naming <c>C</c> again ⇒ the <c>C</c>-bounded result stands (no repeat).
+///     </para>
+///     <para>
+///         <b>Why no TTL.</b> A raise only matters to a read that needs more than <c>C</c>, and such a read re-probes anyway
+///         (rule 3), so a timer would only add refusals for reads that fit. The cost of the rule is one refused re-probe per
+///         genuinely over-ceiling read — a read that fails <c>BadTooManyOperations</c> whatever is done.
+///     </para>
+/// </remarks>
+internal sealed class GatewayRowCeiling
+{
+    private int _learned; // 0 = unknown / not strict
+
+    /// <summary>The learned ceiling (0 = none).</summary>
+    public int Learned => Volatile.Read(ref _learned);
+
+    /// <summary>Runs a drained read under the rules above.</summary>
+    /// <typeparam name="T">The pass result.</typeparam>
+    /// <param name="budget">The <c>max_values</c> the read wants.</param>
+    /// <param name="pass">One drained read with the given budget.</param>
+    /// <param name="needsMore">True when a pass with the given budget delivered it all without proving completeness.</param>
+    /// <returns>The result and the budget it was read with.</returns>
+    public async Task<(T Result, int Budget)> ReadAsync<T>(int budget, Func<int, Task<T>> pass, Func<T, int, bool> needsMore)
+    {
+        var learned = Learned;
+        if (learned <= 0 || learned >= budget) return await AttemptAsync(budget, pass).ConfigureAwait(false);
+
+        var bounded = await AttemptAsync(learned, pass).ConfigureAwait(false);
+        if (bounded.Budget < learned || !needsMore(bounded.Result, bounded.Budget)) return bounded;
+
+        // Rule 3: full at a possibly stale ceiling — re-probe once at the full budget.
+        int fresh;
+        try
+        {
+            var full = await pass(budget).ConfigureAwait(false);
+            Interlocked.CompareExchange(ref _learned, 0, learned);
+            return (full, budget);
+        }
+        catch (Exception ex) when (GatewayReadLimits.TryGetCeiling(ex, out fresh) && fresh < budget)
+        {
+            // fall through (outside the catch)
+        }
+        Volatile.Write(ref _learned, fresh);
+        return fresh > learned ? (await pass(fresh).ConfigureAwait(false), fresh) : bounded;
+    }
+
+    private async Task<(T Result, int Budget)> AttemptAsync<T>(int budget, Func<int, Task<T>> pass)
+    {
+        int ceiling;
+        try
+        {
+            return (await pass(budget).ConfigureAwait(false), budget);
+        }
+        catch (Exception ex) when (GatewayReadLimits.TryGetCeiling(ex, out ceiling) && ceiling < budget)
+        {
+            // fall through to the retry (outside the catch, so a second refusal is not nested in this one)
+        }
+        Volatile.Write(ref _learned, ceiling);
+        return (await pass(ceiling).ConfigureAwait(false), ceiling);
+    }
+}

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/RawAggregateReader.cs

@@ -166,6 +166,7 @@ internal static class RawAggregateReader
 
     /// <summary>Reads and evaluates one per-sample processed read.</summary>
     /// <param name="client">The gateway client.</param>
+    /// <param name="ceiling">The data source's learned gateway row ceiling (#691.3).</param>
     /// <param name="tag">Historian tag name.</param>
     /// <param name="aggregate">A per-sample aggregate (<see cref="Serves"/>).</param>
     /// <param name="plan">The bucket plan.</param>
@@ -175,7 +176,7 @@ internal static class RawAggregateReader
     /// <returns>One value per interval, in the plan's order.</returns>
     /// <exception cref="HistoryReadLimitExceededException">The window holds more than <paramref name="maxWindowRows"/> rows,
     ///     or the stream could not be read to completion within the requested row budget.</exception>
-    public static async Task<IReadOnlyList<DataValueSnapshot>> ReadAsync(IHistorianGatewayClient client, string tag,
+    public static async Task<IReadOnlyList<DataValueSnapshot>> ReadAsync(IHistorianGatewayClient client, GatewayRowCeiling ceiling, string tag,
         HistoryAggregateType aggregate, BucketPlan plan, int maxWindowRows, DateTime now, CancellationToken ct)
     {
         if (plan.BucketCount == 0) return [];
@@ -185,8 +186,9 @@ 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 requested = MaxValuesFor(maxWindowRows);
-        var (pass, maxValues) = await GatewayReadLimits.ReadWithCeilingRetryAsync(requested,
-            budget => ReadWindowAsync(client, tag, aggregate, plan, requestStart, budget, maxWindowRows, ct)).ConfigureAwait(false);
+        var (pass, maxValues) = await ceiling.ReadAsync(requested,
+            budget => ReadWindowAsync(client, tag, aggregate, plan, requestStart, budget, maxWindowRows, ct),
+            (r, budget) => !r.OverCap && r.Received >= budget).ConfigureAwait(false);
         var (buckets, earliest, latest, received, overCap) = pass;
         if (overCap) throw LimitExceeded(tag, maxWindowRows);
         if (received >= maxValues) throw BudgetExhausted(tag, maxWindowRows, requested, maxValues);