diff --git a/CLAUDE.md b/CLAUDE.md index c5e0b0f4..5f302a95 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -1364,7 +1364,7 @@ historian tagname, which is the tag's driver `FullName`. The server dispatches all OPC UA HistoryRead to the registered `IHistorianDataSource` — the `GatewayHistorianDataSource` read client when enabled, else the `NullHistorianDataSource` default -(historized nodes return `GoodNoData` for Raw/AtTime — Processed is `BadAggregateNotSupported` since #682). Supported variants: Raw (⚠️ #687, client-visible: Gateway raw reads no longer return AVEVA's synthetic start-boundary row — asked from 1 ms early (epoch-floored) with `NumValuesPerNode + 16` rows (unclamped; 0 ⇒ `int.MaxValue`) and filtered by timestamp — so an empty window is `GoodNoData`, not a fabricated value at StartTime; the forward end is now exclusive (a point read inclusive); pre-window rows that crowd the budget trigger ONE re-read with room for them, else `BadTooManyOperations`; `ReturnBounds` is ignored on every backend; ⚠️ #690: a Gateway Bad row — a stored NULL, which histsdk sends as value 0 — now reads value **null** + Bad on Raw and AtTime, as on the SQL backends; ⚠️ #691.3: a strict gateway (`AllowStreamingBeyondCeiling=false`) that refuses a raw budget is re-asked ONCE at the ceiling its refusal names — raw and per-sample reads alike — so an uncapped (`NumValuesPerNode = 0`) read of a small window is served; a ceiling-bounded stream that is not provably complete is `BadTooManyOperations`, never a truncated answer), Processed +(historized nodes return `GoodNoData` for Raw/AtTime — Processed is `BadAggregateNotSupported` since #682). Supported variants: Raw (⚠️ #687, client-visible: Gateway raw reads no longer return AVEVA's synthetic start-boundary row — asked from 1 ms early (epoch-floored) with `NumValuesPerNode + 16` rows (unclamped; 0 ⇒ `int.MaxValue`) and filtered by timestamp — so an empty window is `GoodNoData`, not a fabricated value at StartTime; the forward end is now exclusive (a point read inclusive); pre-window rows that crowd the budget trigger ONE re-read with room for them, else `BadTooManyOperations`; `ReturnBounds` is ignored on every backend; ⚠️ #690: a Gateway Bad row — a stored NULL, which histsdk sends as value 0 — now reads value **null** + Bad on Raw and AtTime, as on the SQL backends; ⚠️ #691.3: a strict gateway (`AllowStreamingBeyondCeiling=false`) that refuses a raw budget is re-asked ONCE at the ceiling its refusal names — raw and per-sample reads alike; the ceiling is then remembered per data source (`GatewayRowCeiling`: refusals count as gateway errors), re-learned from a refusal when lowered, and re-probed once at the full budget when a ceiling-bounded read comes back full (raised / streaming) — so an uncapped (`NumValuesPerNode = 0`) read of a small window is served; a ceiling-bounded stream that is not provably complete is `BadTooManyOperations`, never a truncated answer), Processed (backend-dependent: SqlServer/QuestDb serve seventeen; Gateway twelve = `GatewayHistorianDataSource.ServedAggregates` — ⚠️ #681, client-visible: the ten per-sample ones (Average, Count, Min/Max, Range, Start/End/Delta, Min/MaxActualTime) are evaluated from RAW gateway reads by the shared `Sql.Common` evaluator — identical to the SQL backends, Calculated bits, interval-start stamps, #677 reverse; Min/Max are no longer AVEVA modes, the read moves every raw sample of the window (a month of 1 Hz ⇒ `BadTooManyOperations` at the default `MaxTimeWeightedRawRows` cap), and every gateway stream is consumed to completion (an abandoned one makes the gateway evict its shared historian session) — plus TimeAverage/Total via AVEVA modes (interval-start stamped + Calculated; ⚠️ #685: AVEVA's extra pre-window row is dropped ⇒ one row per interval in `[Start, End)`, and an Uncertain (opc 64) interval keeps its value as `UncertainDataSubNormal` — AVEVA's TWA excludes NULL-gap time where the SQL backends bridge it; per-interval values still need gateway fix historiangw#27; ⚠️ #691.2: a REVERSE native read — refused by the gateway before — is now served by forward native requests on the later-anchored #677 grid (whole intervals in one read, the clipped earliest interval as its own exact-span read, Partial unless Bad), newest first, later-edge stamped); the other time-weighted five stay `BadAggregateNotSupported` because the gateway has no prior-sample read for Part 13 bounds; `docs/Historian.md` §Aggregate set), and AtTime over historized variable nodes; Events over alarm-owning equipment-folder event-notifier nodes. Reads are ungated (served from any redundancy node); authorization uses the standard `AccessLevels.HistoryRead` bit set at materialization. diff --git a/docs/Historian.md b/docs/Historian.md index 44325087..35d4c5c3 100644 --- a/docs/Historian.md +++ b/docs/Historian.md @@ -705,8 +705,14 @@ data. every page size from 999,985 to 1,000,000 come back short. A default gateway serves any budget. - ⚠️ **Strict gateway (#691.3, client-visible).** A gateway with `AllowStreamingBeyondCeiling=false` refuses a budget over its `MaxValuesCeiling` (before streaming) and names the ceiling in the refusal; the gateway has no RPC that reports it. The read is - then **re-issued once with the ceiling as the budget** — nothing is cached, so a reconfigured gateway is honoured on the next - read. So an uncapped read (`NumValuesPerNode = 0`, which used to be refused outright on a strict gateway, even for a 10-row + then **re-issued once with the ceiling as the budget**, and the ceiling is **remembered per data source** (`GatewayRowCeiling`, + shared by raw and per-sample reads): every refused attempt counts as an error in the gateway's own outcome metrics, so after the + first refusal later reads ask for at most the learned ceiling and are not refused. A **lowered** ceiling is re-learned from the + refusal of the remembered one. A **raised** ceiling (or a switch to streaming) is noticed by the only reads it matters to: when a + read bounded by the remembered ceiling comes back full without proving completeness, it is re-probed **once** at its full budget + — served ⇒ the ceiling is forgotten; refused naming a larger ceiling ⇒ that one is learned and the read re-issued at it; refused + naming the same ceiling ⇒ the bounded result stands. No timer: a TTL would only add refusals for reads that already fit; the cost + of the rule is one refused re-probe per genuinely over-ceiling read (which fails either way). So an uncapped read (`NumValuesPerNode = 0`, which used to be refused outright on a strict gateway, even for a 10-row window) and a page size within 16 of the ceiling are now served when the window fits. A ceiling-bounded read is answered only when it is **provably complete** — the stream came back short of the ceiling, or reached `EndTime` (or, capped, filled the page); a stream that delivered the whole ceiling short of `EndTime` may be truncated and fails **`BadTooManyOperations`** (not a backend 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 3600a70f..4d182f2e 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 @@ -47,6 +47,9 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis private readonly int _maxRawRows; private readonly TimeProvider _time; + /// A strict gateway's row ceiling, learned from its first refusal (#691.3). + 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 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); } /// One forward raw read of a window, drained, with its pre-window row count and whether it is complete. @@ -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); diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayReadLimits.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayReadLimits.cs index 5f7dc1f1..a4467136 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayReadLimits.cs +++ b/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). /// /// -/// A read whose budget the gateway refuses is re-issued ONCE with the ceiling as its budget -/// (); 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 (); the caller treats a stream that delivered the whole +/// ceiling as possibly truncated (never answered from). /// internal static class GatewayReadLimits { @@ -50,29 +49,6 @@ internal static class GatewayReadLimits return true; } - /// - /// Runs with ; 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. - /// - /// The pass result. - /// The max_values to ask for first. - /// One drained read with the given budget. - /// The pass result and the budget it actually used. - public static async Task<(T Result, int Budget)> ReadWithCeilingRetryAsync(int budget, Func> 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; } } + +/// +/// 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); + } +} 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 cdc25858..6f144c77 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 @@ -166,6 +166,7 @@ internal static class RawAggregateReader /// Reads and evaluates one per-sample processed read. /// The gateway client. + /// The data source's learned gateway row ceiling (#691.3). /// Historian tag name. /// A per-sample aggregate (). /// The bucket plan. @@ -175,7 +176,7 @@ internal static class RawAggregateReader /// One value per interval, in the plan's order. /// The window holds more than rows, /// or the stream could not be read to completion within the requested row budget. - public static async Task> ReadAsync(IHistorianGatewayClient client, string tag, + public static async Task> 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); diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/FakeHistorianGatewayClient.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/FakeHistorianGatewayClient.cs index e1694c9c..a9888666 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/FakeHistorianGatewayClient.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/FakeHistorianGatewayClient.cs @@ -100,8 +100,11 @@ public sealed class FakeHistorianGatewayClient : IHistorianGatewayClient LastReadRawStartUtc = startUtc; LastReadRawEndUtc = endUtc; LastReadRawMaxValues = maxValues; - ReadRawCallCount++; - RawCalls.Add((startUtc, endUtc, maxValues)); + lock (RawCalls) + { + ReadRawCallCount++; + RawCalls.Add((startUtc, endUtc, maxValues)); + } if (MaxValuesCeiling is { } ceiling && maxValues > ceiling) { // The real gateway's refusal (HistorianReadService.ReadRaw), raised on the first MoveNext before any row, as 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 1b20ed63..2dbd50ae 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 @@ -110,6 +110,92 @@ public sealed class GatewayStrictCeilingTests fake.RawCalls.Select(c => c.MaxValues).ShouldBe([int.MaxValue, RawAggregateReader.GatewayDefaultMaxValuesCeiling]); } + // ---- the learned ceiling (review: a refused attempt counts as an error in the gateway's own metrics) ----------------- + + private static int Refusals(FakeHistorianGatewayClient fake) => + fake.RawCalls.Count(c => fake.MaxValuesCeiling is { } n && c.MaxValues > n); + + [Fact] + public async Task The_ceiling_is_learned_once_per_data_source_so_later_reads_are_not_refused() + { + var fake = Store(10, ceiling: 1000); + var ds = Source(fake); + await ds.ReadRawAsync("T", At(0), At(60), 0, Ct); + await ds.ReadRawAsync("T", At(0), At(60), 0, Ct); + await ds.ReadRawAsync("T", At(0), At(60), 995, Ct); + await ds.ReadProcessedAsync("T", At(0), At(10), TimeSpan.FromSeconds(10), HistoryAggregateType.Count, Ct); + + fake.RawCalls.Select(c => c.MaxValues).Take(4).ShouldBe([int.MaxValue, 1000, 1000, 1000]); + fake.RawCalls[4].MaxValues.ShouldBe(1000); // the per-sample read shares it (its own budget 1,000,000) + Refusals(fake).ShouldBe(1); + } + + [Fact] + public async Task A_gateway_that_now_streams_is_re_probed_when_a_learned_ceiling_read_comes_back_full() + { + var fake = Store(100, ceiling: 50); + var ds = Source(fake); + await ds.ReadRawAsync("T", At(0), At(5), 0, Ct); // learns 50 + fake.MaxValuesCeiling = null; // the operator enabled AllowStreamingBeyondCeiling + + var r = await ds.ReadRawAsync("T", At(0), At(200), 0, Ct); + r.Samples.Count.ShouldBe(100); // not BadTooManyOperations from a stale ceiling + fake.RawCalls.Skip(2).Select(c => c.MaxValues).ShouldBe([50, int.MaxValue]); + + await ds.ReadRawAsync("T", At(0), At(5), 0, Ct); + fake.RawCalls[^1].MaxValues.ShouldBe(int.MaxValue); // forgotten + } + + [Fact] + public async Task A_raised_but_still_strict_ceiling_is_re_learned_from_the_re_probe() + { + var fake = Store(100, ceiling: 50); + var ds = Source(fake); + await ds.ReadRawAsync("T", At(0), At(5), 0, Ct); + fake.MaxValuesCeiling = 500; + + (await ds.ReadRawAsync("T", At(0), At(200), 0, Ct)).Samples.Count.ShouldBe(100); + fake.RawCalls.Skip(2).Select(c => c.MaxValues).ShouldBe([50, int.MaxValue, 500]); + + await ds.ReadRawAsync("T", At(0), At(5), 0, Ct); + fake.RawCalls[^1].MaxValues.ShouldBe(500); + } + + [Fact] + public async Task A_genuinely_over_ceiling_read_costs_one_re_probe_and_no_repeat_at_the_same_ceiling() + { + var fake = Store(100, ceiling: 50); + var ds = Source(fake); + await ds.ReadRawAsync("T", At(0), At(5), 0, Ct); + await Should.ThrowAsync(() => ds.ReadRawAsync("T", At(0), At(200), 0, Ct)); + fake.RawCalls.Skip(2).Select(c => c.MaxValues).ShouldBe([50, int.MaxValue]); + ds.GetHealthSnapshot().TotalFailures.ShouldBe(0); + } + + [Fact] + public async Task A_lowered_ceiling_is_re_learned_from_the_refusal_of_the_learned_one() + { + var fake = Store(10, ceiling: 1000); + var ds = Source(fake); + await ds.ReadRawAsync("T", At(0), At(60), 0, Ct); + fake.MaxValuesCeiling = 50; + + (await ds.ReadRawAsync("T", At(0), At(60), 0, Ct)).Samples.Count.ShouldBe(10); + await ds.ReadRawAsync("T", At(0), At(60), 0, Ct); + fake.RawCalls.Skip(2).Select(c => c.MaxValues).ShouldBe([1000, 50, 50]); + } + + [Fact] + public async Task Concurrent_reads_share_the_learned_ceiling_safely() + { + var fake = Store(10, ceiling: 1000); + var ds = Source(fake); + await ds.ReadRawAsync("T", At(0), At(60), 0, Ct); + var results = await Task.WhenAll(Enumerable.Range(0, 32).Select(_ => Task.Run(() => ds.ReadRawAsync("T", At(0), At(60), 0, Ct)))); + results.ShouldAllBe(r => r.Samples.Count == 10); + Refusals(fake).ShouldBe(1); + } + // ---- raw, capped ----------------------------------------------------------------------------------------------------- [Fact]