using System.Globalization; using System.Text.RegularExpressions; using Grpc.Core; namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway; /// /// The HistorianGateway's raw-read row ceiling (Historian:ReadLimits:MaxValuesCeiling, default 1,000,000) as seen /// from this side of the wire. With AllowStreamingBeyondCeiling=false (a "strict" gateway) the gateway refuses /// any ReadRaw whose max_values exceeds the ceiling with InvalidArgument — BEFORE it streams a row /// — and names the ceiling in the refusal ("max_values exceeds the server ceiling of {n}; …"). The gateway has /// no RPC that reports the ceiling, so the refusal is how it is learned (#691.3). /// /// /// 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 (); the caller treats a stream that delivered the whole /// ceiling as possibly truncated (never answered from). /// internal static class GatewayReadLimits { private const string CeilingRejectionText = "max_values exceeds the server ceiling"; private static readonly Regex CeilingNumber = new(@"max_values exceeds the server ceiling of (\d+)", RegexOptions.CultureInvariant | RegexOptions.Compiled); /// /// True when (or an inner exception — the client package wraps the ) /// is the gateway's InvalidArgument "max_values exceeds the server ceiling" rejection. /// /// The exception. /// Whether it is the ceiling rejection. public static bool IsCeilingRejection(Exception ex) => Rejection(ex) is not null; /// /// The ceiling a ceiling rejection names, or — when is a ceiling rejection that does not /// carry a parseable number — the gateway's default (). /// /// The exception. /// The ceiling (> 0) when this returns true. /// Whether is a ceiling rejection. public static bool TryGetCeiling(Exception ex, out int ceiling) { ceiling = 0; if (Rejection(ex) is not { } rpc) return false; var m = CeilingNumber.Match(rpc.Status.Detail); ceiling = m.Success && int.TryParse(m.Groups[1].Value, NumberStyles.None, CultureInfo.InvariantCulture, out var n) && n > 0 ? n : RawAggregateReader.GatewayDefaultMaxValuesCeiling; return true; } private static RpcException? Rejection(Exception ex) { for (var e = ex; e is not null; e = e.InnerException) if (e is RpcException { StatusCode: StatusCode.InvalidArgument } rpc && rpc.Status.Detail.Contains(CeilingRejectionText, StringComparison.Ordinal)) return rpc; return null; } } /// /// 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 , read/written atomically; a /// race costs at most one extra refusal or re-probe, never a wrong answer). /// /// /// /// Rules. (1) Unknown ceiling: ask for the full budget; a refusal naming N < budget learns N and /// the read is re-issued once at N. (2) Learned ceiling C below the budget: ask for C — a refusal /// there (the gateway was LOWERED) re-learns its N as in (1). (3) If that C-bounded read came back /// full but not provably complete, the gateway may have been RAISED (or switched to streaming) since C was /// learned, and answering BadTooManyOperations 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 N ⇒ N is learned and /// the read is re-issued at N; refused naming C again ⇒ the C-bounded result stands (no repeat). /// /// /// Why no TTL. A raise only matters to a read that needs more than 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 BadTooManyOperations whatever is done. /// /// internal sealed class GatewayRowCeiling { private int _learned; // 0 = unknown / not strict /// The learned ceiling (0 = none). public int Learned => Volatile.Read(ref _learned); /// Runs a drained read under the rules above. /// The pass result. /// The max_values the read wants. /// One drained read with the given budget. /// True when a pass with the given budget delivered it all without proving completeness. /// The result and the budget it was read with. public async Task<(T Result, int Budget)> ReadAsync(int budget, Func> pass, Func 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(int budget, Func> 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); } }