From 6669fae7e567f97cd24fa2248f0a5954d8f5743c Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 06:29:10 -0400 Subject: [PATCH] feat(historian): ServerHistorian:MaxRawValuesPerRead caps raw pages and backend reads Review R08:1.1: an uncapped HistoryRead-Raw (NumValuesPerNode = 0) read a tag's whole window into memory on every backend (SQL TOP(int.MaxValue), QuestDB LIMIT, Gateway max_values = int.MaxValue + a second snapshot copy). New ServerHistorian:MaxRawValuesPerRead (default 10 000; <= 0 unbounded with a Validate() warning; 1 raised to 2). Each backend data source takes it, reports it as IHistorianDataSource.MaxRawValuesPerRead (the Gateway: min with a strict gateway's learned ceiling), so the node manager's existing RawPageLimit paging serves N = 0 and N > cap as pages of at most the cap with a continuation point. Independently, a backend raw read asking for 0 or more than the cap reads cap + 1 rows and is REFUSED (HistoryReadLimitExceededException => BadTooManyOperations) when the window holds more - never answered truncated, since the server's paging reads a short answer as the last page. Bound through the Host registry, the SqlServer/QuestDb providers and GatewayHistorian.CreateDataSource. Tests: options (default/binding/warnings), provider binding, Gateway unit + through a real SDK node manager (uncapped -> 999/998/998/605 + CPs), and a conformance case for the SQL backends. Refs #736 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT --- .../GatewayHistorianDataSource.cs | 61 +++- ...wayHistorianServiceCollectionExtensions.cs | 4 +- .../QuestDbHistorianDataSource.cs | 38 ++- .../QuestDbHistorianProvider.cs | 8 +- .../SqlServerHistorianDataSource.cs | 40 ++- .../SqlServerHistorianProvider.cs | 8 +- .../Configuration/HistorianBackendRegistry.cs | 4 +- .../Historian/ServerHistorianOptions.cs | 36 +++ .../IConformanceBackend.cs | 11 +- .../RawReadConformance.cs | 37 +++ .../GatewayRawCapTests.cs | 305 ++++++++++++++++++ .../QuestDbConformanceBackend.cs | 5 +- .../QuestDbHistorianProviderTests.cs | 15 + .../ReadTestDb.cs | 5 +- .../SqlServerConformanceBackend.cs | 3 +- .../SqlServerHistorianProviderTests.cs | 14 + .../Historian/ServerHistorianOptionsTests.cs | 60 ++++ 17 files changed, 621 insertions(+), 33 deletions(-) create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayRawCapTests.cs 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 6f3dfd71..af0c2195 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 @@ -67,6 +67,12 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis /// The tie-cluster limit the crowding re-reads honour — exposed so the factory wiring is testable. internal int MaxPreWindowRows => _maxPreWindowRows; + /// + /// The server raw cap (ServerHistorian:MaxRawValuesPerRead, R08:1.1): the most samples one raw read may + /// return; 0 = none. See and . + /// + private readonly int _maxRawValuesPerRead; + /// A strict gateway's row ceiling, learned from its first refusal. private readonly GatewayRowCeiling _ceiling; private readonly GatewayTagPresenceCache _tags; @@ -96,11 +102,19 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis /// The most rows before a raw window a crowding re-read makes room for (ServerHistorian:MaxTieClusterOverfetch, /// values below 1 count as 1). More ⇒ (BadTooManyOperations). /// + /// + /// The server raw cap (ServerHistorian:MaxRawValuesPerRead, R08:1.1): the most samples one + /// may return — reported as (bounded further by a strict + /// gateway's learned ceiling), so the server pages at it. A read asking for more (or for everything) asks the gateway + /// for cap + 1 rows and is refused when the window holds more, never truncated. <= 0 = unbounded (the pre-#736 + /// behaviour: an uncapped read is sent as ). + /// public GatewayHistorianDataSource(IHistorianGatewayClient client, ILogger logger, int maxRawRows = TimeWeightedAggregator.DefaultMaxWindowRows, TimeProvider? timeProvider = null, - bool atTimeFromRaw = true, int maxPreWindowRows = DefaultMaxPreWindowRows) + bool atTimeFromRaw = true, int maxPreWindowRows = DefaultMaxPreWindowRows, int maxRawValuesPerRead = 0) { _maxPreWindowRows = Math.Max(1, maxPreWindowRows); + _maxRawValuesPerRead = Math.Max(0, maxRawValuesPerRead); _atTimeFromRaw = atTimeFromRaw; ArgumentNullException.ThrowIfNull(client); ArgumentNullException.ThrowIfNull(logger); @@ -203,6 +217,13 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis /// a cancelled gateway stream evicts its pooled historian session). ReturnBounds is not plumbed to the /// data source on any backend, so bounding values are never synthesised here. /// + /// + /// The server raw cap (R08:1.1, #736). With a cap, a read asking for everything (0) or for more than + /// the cap runs as a capped read of cap + 1 — one more than it may return — so an over-cap window is DETECTED and + /// refused () rather than answered truncated (the server's paging + /// reads a short answer as the last page) and never sent as . The server itself never asks + /// for more than the cap (it pages at ); this bounds a caller that bypasses it. + /// /// private async Task> ReadRawWindowAsync( string fullReference, DateTime startUtc, DateTime endUtc, uint maxValuesPerNode, @@ -210,6 +231,9 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis { // The gateway seam caps with an int; OPC UA hands us a uint, so clamp to int range. var cap = (int)Math.Min(maxValuesPerNode, int.MaxValue); + var serverCap = _maxRawValuesPerRead; + var probe = serverCap > 0 && (cap == 0 || cap > serverCap); + if (probe) cap = (int)Math.Min(int.MaxValue, (long)serverCap + 1); if (cap == 0) { var all = await PassAsync(int.MaxValue).ConfigureAwait(false); @@ -225,7 +249,7 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis var asked = (int)Math.Min(int.MaxValue, (long)cap + RawAggregateReader.PreWindowAllowance); var pass = await PassAsync(asked).ConfigureAwait(false); - if (pass.Result.Complete) return pass.Result.Samples; + if (pass.Result.Complete) return Bounded(pass.Result.Samples); // A strict gateway's ceiling already bounded that read: a larger budget would only be refused again. The // refusal carries the pre-window rows it saw: the server re-reads the page that many values smaller. @@ -256,7 +280,7 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis } if (budget <= pass.Budget) break; // no more room, never retry the same full read pass = await PassAsync(budget).ConfigureAwait(false); - if (pass.Result.Complete) return pass.Result.Samples; + if (pass.Result.Complete) return Bounded(pass.Result.Samples); preWindow = Math.Max(preWindow, pass.Result.PreWindowRows); if (pass.Budget < budget) break; // bounded by a ceiling learned meanwhile if (room >= _maxPreWindowRows) break; // the limit's worth of room did not suffice @@ -271,6 +295,16 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis // Pre-window rows the allowance did not cover are what crowded the read; fewer are not the cause. static int Crowding(int preWindowRows) => preWindowRows > RawAggregateReader.PreWindowAllowance ? preWindowRows : 0; + // The cap + 1 probe came back with one sample too many: the window holds more than the server raw cap. + List Bounded(List samples) + { + if (probe && samples.Count > serverCap) + throw new HistoryReadLimitExceededException(serverCap, + $"The raw read of '{fullReference}' holds more than ServerHistorian:MaxRawValuesPerRead ({serverCap}) " + + "samples; set NumValuesPerNode at or below it to page, or narrow the window."); + return samples; + } + // One drained pass; a strict gateway's refusal of the budget is retried once at the ceiling it names. Task<((List Samples, int PreWindowRows, bool Complete) Result, int Budget)> PassAsync(int budget) => _ceiling.ReadAsync(budget, @@ -531,13 +565,20 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis public bool SupportsReverseRaw => false; /// - // 0 until a strict gateway (AllowStreamingBeyondCeiling=false) has refused a budget and - // named its ceiling (GatewayRowCeiling); then that ceiling less the - // RawAggregateReader.PreWindowAllowance a capped read adds for AVEVA's boundary row — so a page of this - // size asks for exactly the ceiling. Back to 0 when a re-probe finds the gateway streaming again. - public int MaxRawValuesPerRead => _ceiling.Learned is var c and > 0 - ? Math.Max(1, c - RawAggregateReader.PreWindowAllowance) - : 0; + // The server raw cap (ServerHistorian:MaxRawValuesPerRead; 0 = none), bounded further by a strict gateway's + // (AllowStreamingBeyondCeiling=false) learned ceiling once a refusal has named it (GatewayRowCeiling) — that + // ceiling less the RawAggregateReader.PreWindowAllowance a capped read adds for AVEVA's boundary row, so a page of + // this size asks for exactly the ceiling. The learned part is forgotten when a re-probe finds the gateway streaming + // again; the cap stays. 0 only with no cap and no learned ceiling. + public int MaxRawValuesPerRead + { + get + { + var learned = _ceiling.Learned is var c and > 0 ? Math.Max(1, c - RawAggregateReader.PreWindowAllowance) : 0; + var cap = _maxRawValuesPerRead; + return cap > 0 && learned > 0 ? Math.Min(cap, learned) : cap > 0 ? cap : learned; + } + } /// // false: the gateway's ReadEvents streams ascending only. The server never sends it a diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianServiceCollectionExtensions.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianServiceCollectionExtensions.cs index 496ba830..a2a9b0d2 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianServiceCollectionExtensions.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayHistorianServiceCollectionExtensions.cs @@ -44,7 +44,9 @@ public static class GatewayHistorian // AtTime from raw reads + the shared evaluator; false = AVEVA's own interpolation (kill switch). atTimeFromRaw: options.GatewayAtTimeFromRaw, // A crowding re-read never makes room for more pre-window rows than the tie-cluster limit. - maxPreWindowRows: options.MaxTieClusterOverfetch); + maxPreWindowRows: options.MaxTieClusterOverfetch, + // The server raw cap (R08:1.1): the page limit the server reads at, and the bound on every raw read. + maxRawValuesPerRead: options.EffectiveMaxRawValuesPerRead); } /// diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataSource.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataSource.cs index a0d9871a..49b814d6 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataSource.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataSource.cs @@ -66,6 +66,7 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource private readonly TimeProvider _time; private readonly HistorianHealthCounters _health = new(); private readonly int _maxTimeWeightedRawRows; + private readonly int _maxRawValuesPerRead; /// Creates the data source over the provider's shared connection, bootstrapper and tag catalog. /// Connection helper. @@ -74,11 +75,19 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource /// Optional logger. /// Clock for "now" (AtTime beyond the last sample, server timestamps); system by default. /// Raw in-window row cap of one time-weighted read (<= 0 = unbounded). + /// + /// The server raw cap (ServerHistorian:MaxRawValuesPerRead, R08:1.1): the most samples one + /// may return — reported as , so the server pages at it. + /// A read asking for more (or for everything) is refused when the window holds more, never truncated. + /// <= 0 = unbounded (the pre-#736 behaviour). + /// internal QuestDbHistorianDataSource( QuestDbConnection conn, QuestDbSchemaBootstrapper bootstrap, QuestDbTagCatalog catalog, ILogger? log = null, - TimeProvider? timeProvider = null, int maxTimeWeightedRawRows = TimeWeightedAggregator.DefaultMaxWindowRows) + TimeProvider? timeProvider = null, int maxTimeWeightedRawRows = TimeWeightedAggregator.DefaultMaxWindowRows, + int maxRawValuesPerRead = 0) { _maxTimeWeightedRawRows = maxTimeWeightedRawRows; + _maxRawValuesPerRead = Math.Max(0, maxRawValuesPerRead); _conn = conn ?? throw new ArgumentNullException(nameof(conn)); _bootstrap = bootstrap ?? throw new ArgumentNullException(nameof(bootstrap)); _catalog = catalog ?? throw new ArgumentNullException(nameof(catalog)); @@ -93,12 +102,21 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource // (int.MaxValue); larger values clamp. At most top samples, ContinuationPoint always null: the node // manager pages by re-reading from the last returned timestamp inclusive, which the (ts, tag) dedup // key makes exact. + // The server raw cap (R08:1.1, #736): with a cap, a read asking for everything or for more than the cap reads + // cap + 1 rows — one more than it may return — so an over-cap window is DETECTED and refused + // (HistoryReadLimitExceededException ⇒ BadTooManyOperations) rather than answered truncated: the server's paging + // reads a short answer as the last page, so a silent LIMIT would lose data. The server itself never asks for more + // than the cap (it pages at MaxRawValuesPerRead); this bounds a caller that bypasses it. /// public async Task ReadRawAsync( string fullReference, DateTime startUtc, DateTime endUtc, uint maxValuesPerNode, CancellationToken cancellationToken) { ArgumentNullException.ThrowIfNull(fullReference); - var top = maxValuesPerNode == 0 ? int.MaxValue : (int)Math.Min(maxValuesPerNode, int.MaxValue); + var cap = _maxRawValuesPerRead; + var probe = cap > 0 && (maxValuesPerNode == 0 || maxValuesPerNode > (uint)cap); + var top = probe + ? (int)Math.Min(int.MaxValue, (long)cap + 1) + : maxValuesPerNode == 0 ? int.MaxValue : (int)Math.Min(maxValuesPerNode, int.MaxValue); try { var samples = new List(); @@ -129,9 +147,19 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource } } + if (probe && samples.Count > cap) + throw new HistoryReadLimitExceededException(cap, + $"The raw read of '{fullReference}' holds more than ServerHistorian:MaxRawValuesPerRead ({cap}) samples; " + + "set NumValuesPerNode at or below it to page, or narrow the window."); + RecordSuccess(); return new HistoryReadResult(samples, ContinuationPoint: null); } + catch (HistoryReadLimitExceededException) + { + // A client-request limit (the server raw cap), not a backend fault: not counted as a failure. + throw; + } catch (Exception ex) { RecordFailure(ex); @@ -360,9 +388,11 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource /// public bool SupportsReverseRaw => true; - // 0: a capped raw read is one bounded statement - no backend row ceiling. + // The server raw cap (ServerHistorian:MaxRawValuesPerRead; 0 = none): the server pages every raw read at it, so a + // client's NumValuesPerNode = 0 is served as pages of at most this many values. QuestDB itself has no row ceiling - + // a capped raw read is one bounded statement. /// - public int MaxRawValuesPerRead => 0; + public int MaxRawValuesPerRead => _maxRawValuesPerRead; // Reverse events reads are served natively (ORDER BY ... DESC). /// diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianProvider.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianProvider.cs index 36579dfb..34ca413e 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianProvider.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianProvider.cs @@ -35,14 +35,17 @@ public sealed class QuestDbHistorianProvider : IHistorianBackendProvider, IAsync /// Creates the provider over the ServerHistorian:QuestDb options (validated elsewhere). /// The ServerHistorian:QuestDb options. /// The shared ServerHistorian:MaxTimeWeightedRawRows cap (<= 0 = unbounded). + /// The shared ServerHistorian:MaxRawValuesPerRead raw cap (<= 0 = unbounded; R08:1.1). public QuestDbHistorianProvider(QuestDbHistorianOptions options, - int maxTimeWeightedRawRows = TimeWeightedAggregator.DefaultMaxWindowRows) + int maxTimeWeightedRawRows = TimeWeightedAggregator.DefaultMaxWindowRows, int maxRawValuesPerRead = 0) { _options = options ?? throw new ArgumentNullException(nameof(options)); _maxTimeWeightedRawRows = maxTimeWeightedRawRows; + _maxRawValuesPerRead = maxRawValuesPerRead; } private readonly int _maxTimeWeightedRawRows; + private readonly int _maxRawValuesPerRead; /// public string Name => "QuestDb"; @@ -75,7 +78,8 @@ public sealed class QuestDbHistorianProvider : IHistorianBackendProvider, IAsync { var s = Get(sp); return new QuestDbHistorianDataSource(s.Conn, s.Bootstrap, s.Catalog, - Loggers(sp).CreateLogger(), maxTimeWeightedRawRows: _maxTimeWeightedRawRows); + Loggers(sp).CreateLogger(), maxTimeWeightedRawRows: _maxTimeWeightedRawRows, + maxRawValuesPerRead: _maxRawValuesPerRead); } /// diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianDataSource.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianDataSource.cs index 07d6887e..6525a173 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianDataSource.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianDataSource.cs @@ -61,6 +61,7 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource private readonly HistorianHealthCounters _health = new(); private readonly int _maxTimeWeightedRawRows; + private readonly int _maxRawValuesPerRead; /// Creates the data source over the provider's shared connection, bootstrapper and tag catalog. /// Connection helper. @@ -69,11 +70,19 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource /// Optional logger. /// Clock for "now" (AtTime beyond the last sample, server timestamps); system by default. /// Raw in-window row cap of one time-weighted read (<= 0 = unbounded). + /// + /// The server raw cap (ServerHistorian:MaxRawValuesPerRead, R08:1.1): the most samples one + /// may return — reported as , so the server pages at it. + /// A read asking for more (or for everything) is refused when the window holds more, never truncated. + /// <= 0 = unbounded (the pre-#736 behaviour). + /// internal SqlServerHistorianDataSource( SqlHistorianConnection conn, SqlServerSchemaBootstrapper bootstrap, SqlServerTagCatalog catalog, ILogger? log = null, - TimeProvider? timeProvider = null, int maxTimeWeightedRawRows = TimeWeightedAggregator.DefaultMaxWindowRows) + TimeProvider? timeProvider = null, int maxTimeWeightedRawRows = TimeWeightedAggregator.DefaultMaxWindowRows, + int maxRawValuesPerRead = 0) { _maxTimeWeightedRawRows = maxTimeWeightedRawRows; + _maxRawValuesPerRead = Math.Max(0, maxRawValuesPerRead); _conn = conn ?? throw new ArgumentNullException(nameof(conn)); _bootstrap = bootstrap ?? throw new ArgumentNullException(nameof(bootstrap)); _catalog = catalog ?? throw new ArgumentNullException(nameof(catalog)); @@ -90,13 +99,22 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource // 0 is the node manager's "unbounded" sentinel; values above clamp. // The node manager pages by re-reading from the last returned timestamp inclusive and trimming it; the // (TagId, SourceTimeUtc) primary key rules out tie clusters, so that is exact. + // The server raw cap (R08:1.1, #736): with a cap, a read asking for everything or for more than the cap reads + // cap + 1 rows — one more than it may return — so an over-cap window is DETECTED and refused + // (HistoryReadLimitExceededException ⇒ BadTooManyOperations) rather than answered truncated: the server's paging + // reads a short answer as the last page, so a silent TOP would lose data. The server itself never asks for more than + // the cap (it pages at MaxRawValuesPerRead); this bounds a caller that bypasses it. // /// public async Task ReadRawAsync( string fullReference, DateTime startUtc, DateTime endUtc, uint maxValuesPerNode, CancellationToken cancellationToken) { ArgumentNullException.ThrowIfNull(fullReference); - var top = maxValuesPerNode == 0 ? int.MaxValue : (int)Math.Min(maxValuesPerNode, int.MaxValue); + var cap = _maxRawValuesPerRead; + var probe = cap > 0 && (maxValuesPerNode == 0 || maxValuesPerNode > (uint)cap); + var top = probe + ? (int)Math.Min(int.MaxValue, (long)cap + 1) + : maxValuesPerNode == 0 ? int.MaxValue : (int)Math.Min(maxValuesPerNode, int.MaxValue); try { var tagId = await ResolveAsync(fullReference, cancellationToken).ConfigureAwait(false); @@ -126,9 +144,19 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource } } + if (probe && samples.Count > cap) + throw new HistoryReadLimitExceededException(cap, + $"The raw read of '{fullReference}' holds more than ServerHistorian:MaxRawValuesPerRead ({cap}) samples; " + + "set NumValuesPerNode at or below it to page, or narrow the window."); + RecordSuccess(); return new HistoryReadResult(samples, ContinuationPoint: null); } + catch (HistoryReadLimitExceededException) + { + // A client-request limit (the server raw cap), not a backend fault: not counted as a failure. + throw; + } catch (Exception ex) { RecordFailure(ex); @@ -319,9 +347,13 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource /// public bool SupportsReverseRaw => true; - // 0: a capped raw read is one bounded statement — no backend row ceiling. + // + // The server raw cap (ServerHistorian:MaxRawValuesPerRead; 0 = none): the server pages every raw read at it, so a + // client's NumValuesPerNode = 0 is served as pages of at most this many values. SQL Server itself has no row + // ceiling — a capped raw read is one bounded statement. + // /// - public int MaxRawValuesPerRead => 0; + public int MaxRawValuesPerRead => _maxRawValuesPerRead; // Reverse events reads are served natively (ORDER BY … DESC). /// diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianProvider.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianProvider.cs index 80cfe919..52884a95 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianProvider.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianProvider.cs @@ -31,14 +31,17 @@ public sealed class SqlServerHistorianProvider : IHistorianBackendProvider /// Creates the provider over the ServerHistorian:SqlServer options (validated elsewhere). /// The ServerHistorian:SqlServer options. /// The shared ServerHistorian:MaxTimeWeightedRawRows cap (<= 0 = unbounded). + /// The shared ServerHistorian:MaxRawValuesPerRead raw cap (<= 0 = unbounded; R08:1.1). public SqlServerHistorianProvider(SqlServerHistorianOptions options, - int maxTimeWeightedRawRows = TimeWeightedAggregator.DefaultMaxWindowRows) + int maxTimeWeightedRawRows = TimeWeightedAggregator.DefaultMaxWindowRows, int maxRawValuesPerRead = 0) { _options = options ?? throw new ArgumentNullException(nameof(options)); _maxTimeWeightedRawRows = maxTimeWeightedRawRows; + _maxRawValuesPerRead = maxRawValuesPerRead; } private readonly int _maxTimeWeightedRawRows; + private readonly int _maxRawValuesPerRead; /// public string Name => "SqlServer"; @@ -69,7 +72,8 @@ public sealed class SqlServerHistorianProvider : IHistorianBackendProvider { var s = Get(sp); return new SqlServerHistorianDataSource(s.Conn, s.Bootstrap, s.Catalog, - Loggers(sp).CreateLogger(), maxTimeWeightedRawRows: _maxTimeWeightedRawRows); + Loggers(sp).CreateLogger(), maxTimeWeightedRawRows: _maxTimeWeightedRawRows, + maxRawValuesPerRead: _maxRawValuesPerRead); } /// diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/HistorianBackendRegistry.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/HistorianBackendRegistry.cs index 221180d2..0fdba329 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/HistorianBackendRegistry.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/HistorianBackendRegistry.cs @@ -45,8 +45,8 @@ public static class HistorianBackendRegistry new Dictionary>(StringComparer.OrdinalIgnoreCase) { [ServerHistorianOptions.GatewayBackendName] = (o, _) => new GatewayHistorianProvider(o), - [SqlServerBackendName] = (o, c) => new SqlServerHistorianProvider(BindSqlServer(c), o.MaxTimeWeightedRawRows), - [QuestDbBackendName] = (o, c) => new QuestDbHistorianProvider(BindQuestDb(c), o.MaxTimeWeightedRawRows), + [SqlServerBackendName] = (o, c) => new SqlServerHistorianProvider(BindSqlServer(c), o.MaxTimeWeightedRawRows, o.EffectiveMaxRawValuesPerRead), + [QuestDbBackendName] = (o, c) => new QuestDbHistorianProvider(BindQuestDb(c), o.MaxTimeWeightedRawRows, o.EffectiveMaxRawValuesPerRead), }; /// The registered backend names, for validation messages. diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ServerHistorianOptions.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ServerHistorianOptions.cs index d19dbe38..1835c5ef 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ServerHistorianOptions.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ServerHistorianOptions.cs @@ -154,6 +154,38 @@ public sealed class ServerHistorianOptions /// public int MaxTimeWeightedRawRows { get; init; } = 1_000_000; + /// The default of : 10 000 values per raw page / backend read. + public const int DefaultMaxRawValuesPerRead = 10_000; + + /// + /// The smallest usable : a read must carry a page of at least one value AND the + /// lookahead row that tells the server more data follows, so a cap of 1 is raised to 2 (). + /// + public const int MinMaxRawValuesPerRead = 2; + + /// + /// Every backend: the most raw values the server returns on ONE HistoryRead-Raw page, and the most rows one + /// backend raw read may materialise — the server's own row ceiling (review R08:1.1, #736). A client's + /// NumValuesPerNode = 0 ("all values") no longer reads a whole window into memory: it is served as pages of at + /// most this many values, each with a continuation point (Part 11 allows fewer values than requested with one), through + /// the same tie-safe paging a capped read uses (RawPageLimit: page + boundary ties + lookahead ≤ cap, so a first + /// page holds cap − 1 values). A NumValuesPerNode above it is paged at it too. Advertised as + /// HistoryServerCapabilities/MaxReturnDataValues. The backends' own TOP / LIMIT / max_values + /// are bounded by it independently: a raw read asking for more than this (or for everything) reads at most cap + 1 rows + /// and is REFUSED (BadTooManyOperations) when the window holds more — never answered truncated — so a caller that + /// bypasses the node manager's paging cannot bypass the bound. On the Gateway backend a strict gateway's learned row + /// ceiling (less its boundary-row allowance) still applies when lower. The tie-cluster bound of a resumed page is also + /// at most cap − 1 (as under a gateway ceiling; above that is inert). Default + /// 10 000. Non-positive = unbounded, the pre-#736 behaviour (a warning); 1 is raised to 2. + /// + public int MaxRawValuesPerRead { get; init; } = DefaultMaxRawValuesPerRead; + + /// + /// as the backends and the node manager apply it: 0 (unbounded) when + /// non-positive, otherwise at least . The one place the rule lives. + /// + public int EffectiveMaxRawValuesPerRead => MaxRawValuesPerRead <= 0 ? 0 : Math.Max(MinMaxRawValuesPerRead, MaxRawValuesPerRead); + /// /// how far into the future an OPC UA HistoryUpdate (Insert / Replace / Update) item's /// SourceTimestamp may lie, measured from the server clock. An item beyond now + skew is refused @@ -208,6 +240,10 @@ public sealed class ServerHistorianOptions warnings.Add($"ServerHistorian:MaxConcurrentHistoryReadBatches is {MaxConcurrentHistoryReadBatches} — must be > 0; the process-wide HistoryRead batch limiter is disabled."); if (MaxTimeWeightedRawRows <= 0) warnings.Add($"ServerHistorian:MaxTimeWeightedRawRows is {MaxTimeWeightedRawRows} — must be > 0; time-weighted aggregate reads on the SqlServer/QuestDb backends stream an unbounded number of raw samples."); + if (MaxRawValuesPerRead <= 0) + warnings.Add($"ServerHistorian:MaxRawValuesPerRead is {MaxRawValuesPerRead} — must be > 0; HistoryRead-Raw pages and backend raw reads are unbounded, so a client asking for a whole window (NumValuesPerNode = 0) materialises every sample in memory."); + else if (MaxRawValuesPerRead < MinMaxRawValuesPerRead) + warnings.Add($"ServerHistorian:MaxRawValuesPerRead is {MaxRawValuesPerRead} — must be at least {MinMaxRawValuesPerRead} (a page plus its lookahead row); treated as {MinMaxRawValuesPerRead}."); if (EventReadHorizon < TimeSpan.Zero) warnings.Add($"ServerHistorian:EventReadHorizon is {EventReadHorizon} — negative is treated as no horizon (00:00:00); HistoryReadEvents reads every window in full."); if (HistoryReadDeadline <= TimeSpan.Zero) diff --git a/tests/Drivers/Historian.Conformance/IConformanceBackend.cs b/tests/Drivers/Historian.Conformance/IConformanceBackend.cs index afc9f69f..970e7c77 100644 --- a/tests/Drivers/Historian.Conformance/IConformanceBackend.cs +++ b/tests/Drivers/Historian.Conformance/IConformanceBackend.cs @@ -22,9 +22,14 @@ public interface IConformanceBackend : IAsyncDisposable /// Its "now"; null means the system clock. IHistorianDataSource CreateDataSource(TimeProvider? clock = null); - /// A fresh data source whose time-weighted raw-row cap (ServerHistorian:MaxTimeWeightedRawRows) is . - /// The cap (<= 0 = unbounded). - IHistorianDataSource CreateDataSource(int maxTimeWeightedRawRows); + /// + /// A fresh data source whose time-weighted raw-row cap (ServerHistorian:MaxTimeWeightedRawRows) is + /// and whose server raw cap (ServerHistorian:MaxRawValuesPerRead, + /// R08:1.1 / #736) is . + /// + /// The time-weighted cap (<= 0 = unbounded). + /// The raw cap (<= 0 = unbounded, the default). + IHistorianDataSource CreateDataSource(int maxTimeWeightedRawRows, int maxRawValuesPerRead = 0); /// A fresh live-value writer over this backend's store. /// The clock used to stamp values that carry no timestamp; null means the system clock. diff --git a/tests/Drivers/Historian.Conformance/RawReadConformance.cs b/tests/Drivers/Historian.Conformance/RawReadConformance.cs index 8a04adc6..39c1d79f 100644 --- a/tests/Drivers/Historian.Conformance/RawReadConformance.cs +++ b/tests/Drivers/Historian.Conformance/RawReadConformance.cs @@ -25,6 +25,43 @@ public abstract class RawReadConformance : ConformanceBase r.ContinuationPoint.ShouldBeNull(); } + /// + /// R08:1.1 (#736): the server raw cap bounds the backend's own TOP / LIMIT. A read asking for everything + /// (or for more than the cap) reads at most cap + 1 rows and is refused when the window holds more — never answered + /// truncated (a short answer is what the server's paging reads as "the last page") — and is served in full when it fits. + /// A read at or under the cap is unchanged. The refusal is a client-request limit, not a backend fault. + /// + [Fact] + public async Task The_server_raw_cap_bounds_every_read_and_refuses_an_over_cap_window_rather_than_truncating() + { + await ConformanceData.SeedAAsync(B); // five samples at +0, +10, +20, +30, +40 + var src = B.CreateDataSource(maxTimeWeightedRawRows: 1_000_000, maxRawValuesPerRead: 3); + src.MaxRawValuesPerRead.ShouldBe(3); + + // Five rows under a cap of three: everything / above the cap ⇒ refused; at the cap ⇒ the first three; under ⇒ as asked. + var refused = await Should.ThrowAsync(() => src.ReadRawAsync("A", At(-1), At(100), 0, Ct)); + refused.Limit.ShouldBe(3); + await Should.ThrowAsync(() => src.ReadRawAsync("A", At(-1), At(100), 4, Ct)); + await Should.ThrowAsync(() => src.ReadRawAsync("A", At(-1), At(100), uint.MaxValue, Ct)); + Values(await src.ReadRawAsync("A", At(-1), At(100), 3, Ct)).ShouldBe([10d, 20d, 30d]); + Values(await src.ReadRawAsync("A", At(-1), At(100), 2, Ct)).ShouldBe([10d, 20d]); + + // Three rows fit: everything / above the cap are served in full. + Values(await src.ReadRawAsync("A", At(10), At(40), 0, Ct)).ShouldBe([20d, 30d, 40d]); + Values(await src.ReadRawAsync("A", At(10), At(40), 4, Ct)).ShouldBe([20d, 30d, 40d]); + + // Reverse reads are bounded the same way, keeping the LATEST rows. + await Should.ThrowAsync(() => src.ReadRawAsync("A", At(100), At(-1), 0, Ct)); + Values(await src.ReadRawAsync("A", At(100), At(-1), 3, Ct)).ShouldBe([50d, 40d, 30d]); + Values(await src.ReadRawAsync("A", At(40), At(10), 0, Ct)).ShouldBe([50d, 40d, 30d]); + + // A point read and an empty window are unaffected. + Values(await src.ReadRawAsync("A", At(10), At(10), 0, Ct)).ShouldBe([20d]); + (await src.ReadRawAsync("A", At(50), At(60), 0, Ct)).Samples.ShouldBeEmpty(); + + src.GetHealthSnapshot().TotalFailures.ShouldBe(0, "a refused over-cap read is a client-request limit, not a backend fault"); + } + [Fact] public async Task Forward_read_honours_the_limit() { diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayRawCapTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayRawCapTests.cs new file mode 100644 index 00000000..c1c5da7f --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayRawCapTests.cs @@ -0,0 +1,305 @@ +using Google.Protobuf.WellKnownTypes; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging.Abstractions; +using Opc.Ua; +using Opc.Ua.Server; +using Shouldly; +using Xunit; +using ZB.MOM.WW.HistorianGateway.Contracts.Grpc; +using ZB.MOM.WW.OtOpcUa.Commons.OpcUa; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.OpcUaServer; +using ZB.MOM.WW.OtOpcUa.Runtime.Historian; +using SdkHistoryReadResult = Opc.Ua.HistoryReadResult; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests; + +/// +/// Review R08:1.1 (#736): the server's raw cap (ServerHistorian:MaxRawValuesPerRead) bounds the Gateway +/// backend's max_values independently of the node manager. An uncapped raw read (NumValuesPerNode = 0) +/// used to be sent as and materialise the whole window twice (the protobuf list and the +/// snapshot list); a read above the cap now asks for cap + 1 rows and is REFUSED (, +/// BadTooManyOperations) when the window holds more — never answered truncated, because a short answer is what the +/// server's paging reads as "the last page". The data source reports the cap as its +/// (or a strict gateway's lower learned ceiling), so the node +/// manager's existing ceiling paging serves NumValuesPerNode = 0 as pages of at most the cap, each with a +/// continuation point. +/// +public sealed class GatewayRawCapTests : IDisposable +{ + private static CancellationToken Ct => TestContext.Current.CancellationToken; + + private static readonly DateTime T0 = new(2026, 10, 22, 10, 0, 0, DateTimeKind.Utc); + + private const int Cap = 1000; + + private readonly string _pkiRoot = Path.Combine(Path.GetTempPath(), $"otopcua-gwrawcap-{Guid.NewGuid():N}"); + + private static DateTime At(int seconds) => T0.AddSeconds(seconds); + + private static HistorianSample Row(DateTime t, double v, uint quality = 0) => + new() { Tag = "T", Timestamp = Timestamp.FromDateTime(t), NumericValue = v, OpcQuality = 192, Quality = quality }; + + /// AVEVA-shaped: a boundary row at the query start, [start, end] inclusive, capped at max. + private static FakeHistorianGatewayClient Store(int count, int? ceiling = null) + { + var stored = Enumerable.Range(0, count).Select(i => (T: At(i), V: (double)i)).ToList(); + return new FakeHistorianGatewayClient + { + MaxValuesCeiling = ceiling, + RawSource = (start, end, max) => + { + if (max <= 0) throw new ArgumentOutOfRangeException(nameof(max), "max_values must be positive"); // the client package + var rows = new List(); + if (stored.All(s => s.T != start)) + { + var prior = stored.LastOrDefault(s => s.T < start); + rows.Add(prior.T == default + ? new HistorianSample { Tag = "T", Timestamp = Timestamp.FromDateTime(start), NumericValue = 0, OpcQuality = 0, Quality = 1 } + : Row(start, prior.V, 133)); + } + rows.AddRange(stored.Where(s => s.T >= start && s.T <= end).Select(s => Row(s.T, s.V))); + return rows.Take(max); + }, + }; + } + + private static GatewayHistorianDataSource Source(FakeHistorianGatewayClient fake, int cap = Cap) => + new(fake, NullLogger.Instance, maxRawValuesPerRead: cap); + + private static int Probe => Cap + 1 + RawAggregateReader.PreWindowAllowance; // cap + 1 rows + the boundary-row allowance + + // ---- the data source alone ---------------------------------------------------------------------------------------- + + [Fact] + public void Without_a_cap_the_source_is_unchanged() + { + new GatewayHistorianDataSource(Store(10), NullLogger.Instance).MaxRawValuesPerRead.ShouldBe(0); + Source(Store(10), cap: 0).MaxRawValuesPerRead.ShouldBe(0); + } + + [Fact] + public async Task The_cap_is_the_page_limit_until_a_lower_gateway_ceiling_is_learned() + { + Source(Store(3600)).MaxRawValuesPerRead.ShouldBe(Cap); + + // A strict gateway whose ceiling is ABOVE the cap: the cap wins, nothing is refused by the gateway. + var higher = Store(3600, ceiling: 2000); + var ds = Source(higher); + await Should.ThrowAsync(() => ds.ReadRawAsync("T", At(0), At(3600), 0, Ct)); + higher.RawCalls.ShouldAllBe(c => c.MaxValues <= 2000); + ds.MaxRawValuesPerRead.ShouldBe(Cap); + + // A strict gateway whose ceiling is BELOW the cap: the learned ceiling (less the allowance) wins once known. + var lower = Store(3600, ceiling: 500); + var ds2 = Source(lower); + ds2.MaxRawValuesPerRead.ShouldBe(Cap); + await Should.ThrowAsync(() => ds2.ReadRawAsync("T", At(0), At(3600), 0, Ct)); + ds2.MaxRawValuesPerRead.ShouldBe(500 - RawAggregateReader.PreWindowAllowance); + } + + [Fact] + public async Task An_uncapped_read_over_the_cap_is_refused_never_truncated_and_never_asks_for_everything() + { + var fake = Store(3600); + var ds = Source(fake); + + var ex = await Should.ThrowAsync(() => ds.ReadRawAsync("T", At(0), At(3600), 0, Ct)); + + ex.Limit.ShouldBe(Cap); + ex.Message.ShouldContain("MaxRawValuesPerRead"); + fake.RawCalls.Select(c => c.MaxValues).ShouldBe([Probe]); // never int.MaxValue + fake.RawStreamsCompleted.ShouldBe(1, "the bounded stream is drained, never abandoned"); + ds.GetHealthSnapshot().TotalFailures.ShouldBe(0, "a client-request limit is not a backend fault"); + } + + [Theory] + [InlineData(10)] + [InlineData(Cap - 1)] + [InlineData(Cap)] + public async Task An_uncapped_read_that_fits_the_cap_is_served_in_full(int count) + { + var fake = Store(count); + var r = await Source(fake).ReadRawAsync("T", At(0), At(3600), 0, Ct); + + r.Samples.Select(s => s.Value).ShouldBe(Enumerable.Range(0, count).Select(i => (object?)(double)i)); + fake.RawCalls.Select(c => c.MaxValues).ShouldBe([Probe]); + } + + [Theory] + [InlineData(Cap + 1u)] + [InlineData(5000u)] + [InlineData(uint.MaxValue)] + public async Task A_client_cap_above_the_server_cap_is_bounded_by_it(uint n) + { + await Should.ThrowAsync(() => Source(Store(3600)).ReadRawAsync("T", At(0), At(3600), n, Ct)); + + var fake = Store(900); + var r = await Source(fake).ReadRawAsync("T", At(0), At(3600), n, Ct); + r.Samples.Count.ShouldBe(900); + fake.RawCalls.ShouldAllBe(c => c.MaxValues == Probe); + } + + [Theory] + [InlineData(Cap)] + [InlineData(Cap - 1)] + [InlineData(1)] + public async Task A_client_cap_at_or_under_the_server_cap_reads_as_before(int n) + { + var fake = Store(3600); + var r = await Source(fake).ReadRawAsync("T", At(0), At(3600), (uint)n, Ct); + + r.Samples.Count.ShouldBe(n); + fake.RawCalls.Select(c => c.MaxValues).ShouldBe([n + RawAggregateReader.PreWindowAllowance]); + } + + [Fact] + public async Task A_point_read_and_an_empty_window_are_unaffected() + { + var fake = Store(3600); + var ds = Source(fake); + (await ds.ReadRawAsync("T", At(5), At(5), 0, Ct)).Samples.Single().Value.ShouldBe(5d); + (await ds.ReadRawAsync("T", At(5000), At(6000), 0, Ct)).Samples.ShouldBeEmpty(); + fake.RawCalls.ShouldAllBe(c => c.MaxValues == Probe); + } + + // ---- the factory binding -------------------------------------------------------------------------------------------- + + [Theory] + [InlineData(777, 777)] + [InlineData(0, 0)] + [InlineData(-1, 0)] + [InlineData(1, ServerHistorianOptions.MinMaxRawValuesPerRead)] + public void The_factory_binds_the_effective_option(int configured, int expected) + { + var options = new ServerHistorianOptions + { + Endpoint = "http://localhost:5222", UseTls = false, ApiKey = "histgw_test_secret", MaxRawValuesPerRead = configured, + }; + using var ds = GatewayHistorian.CreateDataSource(options, new ServiceCollection().BuildServiceProvider()); + ds.ShouldBeOfType().MaxRawValuesPerRead.ShouldBe(expected); + } + + [Fact] + public void The_default_option_caps_the_source_at_ten_thousand() + { + var options = new ServerHistorianOptions { Endpoint = "http://localhost:5222", UseTls = false, ApiKey = "histgw_test_secret" }; + using var ds = new GatewayHistorianProvider(options).CreateDataSource(new ServiceCollection().BuildServiceProvider()); + ds.MaxRawValuesPerRead.ShouldBe(ServerHistorianOptions.DefaultMaxRawValuesPerRead); + } + + // ---- through the node manager --------------------------------------------------------------------------------------- + + [Theory] + [InlineData(0u, false)] + [InlineData(0u, true)] + [InlineData(5000u, false)] + [InlineData(uint.MaxValue, false)] + public async Task An_uncapped_HistoryRead_pages_at_the_cap_with_continuation_points(uint n, bool reverse) + { + var fake = Store(3600); + var (host, nm, nodeId) = await BootAsync(fake); + try + { + var (values, pages, status) = reverse + ? DrainFrom(nm, nodeId, At(3600), At(-1), n, null) + : DrainFrom(nm, nodeId, At(0), At(3600), n, null); + + status.Code.ShouldBe(StatusCodes.Good); + var expected = Enumerable.Range(0, 3600).Select(i => (double)i); + values.ShouldBe(reverse ? expected.Reverse() : expected); + pages.Count.ShouldBeGreaterThan(1, "a year-sized window at N = 0 is paged, not one shot"); + pages.ShouldAllBe(p => p > 0 && p <= Cap); + pages[0].ShouldBe(Cap - 1); // page + lookahead fills a read + fake.RawCalls.ShouldAllBe(c => c.MaxValues > 0 && c.MaxValues <= Cap + RawAggregateReader.PreWindowAllowance); + } + finally { await host.DisposeAsync(); } + } + + [Fact] + public async Task A_bounded_uncapped_HistoryRead_pages_within_the_cap() + { + var fake = Store(3600); + var (host, nm, nodeId) = await BootAsync(fake); + try + { + var (values, pages, status) = DrainFrom(nm, nodeId, At(10).AddMilliseconds(500), At(3000).AddMilliseconds(500), 0, null, returnBounds: true); + status.Code.ShouldBe(StatusCodes.Good); + values.ShouldBe(Enumerable.Range(10, 2992).Select(i => (double)i)); // start bound +10, data +11..+3000, end bound +3001 + pages.ShouldAllBe(p => p <= Cap); + fake.RawCalls.ShouldAllBe(c => c.MaxValues > 0 && c.MaxValues <= Cap + RawAggregateReader.PreWindowAllowance); + } + finally { await host.DisposeAsync(); } + } + + // ---- helpers --------------------------------------------------------------------------------------------------------- + + private async Task<(OpcUaApplicationHost Host, OtOpcUaNodeManager Nm, NodeId NodeId)> BootAsync(FakeHistorianGatewayClient fake) + { + var started = await OpcUaHostStart.StartAsync( + port => new OpcUaApplicationHost( + new OpcUaApplicationHostOptions + { + ApplicationName = "OtOpcUa.GatewayRawCapTest", + ApplicationUri = $"urn:OtOpcUa.GatewayRawCapTest:{Guid.NewGuid():N}", + OpcUaPort = port, + PublicHostname = "localhost", + PkiStoreRoot = _pkiRoot, + }, + NullLogger.Instance), + () => new OtOpcUaSdkServer(), Ct); + var nm = started.Server.NodeManager!; + nm.HistoryContinuationStore = new InMemoryHistoryContinuationStore(); + nm.HistorianDataSource = Source(fake); + nm.EnsureVariable("eq-1/pv", parentFolderNodeId: null, displayName: "PV", dataType: "Double", + writable: false, historianTagname: "T", realm: AddressSpaceRealm.Uns); + return (started.Host, nm, nm.TryGetVariable("eq-1/pv")!.NodeId); + } + + private static (List Values, List Pages, StatusCode Status) DrainFrom( + OtOpcUaNodeManager nm, NodeId nodeId, DateTime start, DateTime end, uint max, byte[]? cp, int maxPages = 20, + bool returnBounds = false) + { + var values = new List(); + var pages = new List(); + ServiceResult error; + do + { + (var page, error, cp) = ReadRaw(nm, nodeId, start, end, max, cp, returnBounds); + values.AddRange(page); + pages.Add(page.Count); + pages.Count.ShouldBeLessThanOrEqualTo(maxPages, "paging must terminate"); + } + while (cp is not null && StatusCode.IsGood(error.StatusCode)); + return (values, pages, error.StatusCode); + } + + private static (IReadOnlyList Values, ServiceResult Error, byte[]? Cp) ReadRaw( + OtOpcUaNodeManager nm, NodeId nodeId, DateTime start, DateTime end, uint max, byte[]? inboundCp, bool returnBounds = false) + { + var context = new OperationContext(new RequestHeader(), secureChannelContext: null, RequestType.HistoryRead, identity: null); + var details = new ReadRawModifiedDetails + { + ReturnBounds = returnBounds, IsReadModified = false, StartTime = start, EndTime = end, NumValuesPerNode = max, + }; + var nodesToRead = new List { new() { NodeId = nodeId, ContinuationPoint = inboundCp } }; + var results = new List { null! }; + var errors = new List { null! }; + nm.HistoryRead(context, details, TimestampsToReturn.Both, releaseContinuationPoints: false, nodesToRead, results, errors); + + var values = new List(); + if (results[0]?.HistoryData is { } ho && !ExtensionObject.IsNull(ho)) + foreach (var dv in ((HistoryData)ExtensionObject.ToEncodeable(ho)).DataValues) + values.Add(Convert.ToDouble(dv.Value)); + return (values, errors[0], results[0]?.ContinuationPoint); + } + + public void Dispose() + { + if (Directory.Exists(_pkiRoot)) + { + try { Directory.Delete(_pkiRoot, recursive: true); } + catch { /* best-effort */ } + } + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/QuestDbConformanceBackend.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/QuestDbConformanceBackend.cs index 5c90931d..7c8857f0 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/QuestDbConformanceBackend.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/QuestDbConformanceBackend.cs @@ -110,8 +110,9 @@ internal sealed class QuestDbConformanceBackend : IConformanceBackend new QuestDbHistorianDataSource(Conn, Boot, NewCatalog(), timeProvider: clock); /// - public IHistorianDataSource CreateDataSource(int maxTimeWeightedRawRows) => - new QuestDbHistorianDataSource(Conn, Boot, NewCatalog(), maxTimeWeightedRawRows: maxTimeWeightedRawRows); + public IHistorianDataSource CreateDataSource(int maxTimeWeightedRawRows, int maxRawValuesPerRead = 0) => + new QuestDbHistorianDataSource(Conn, Boot, NewCatalog(), maxTimeWeightedRawRows: maxTimeWeightedRawRows, + maxRawValuesPerRead: maxRawValuesPerRead); /// /// The T4 live-value writer over a fresh catalog (fresh positive cache) and this instance's WAL wait. diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests/QuestDbHistorianProviderTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests/QuestDbHistorianProviderTests.cs index c67a9b03..cf950e99 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests/QuestDbHistorianProviderTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests/QuestDbHistorianProviderTests.cs @@ -47,6 +47,21 @@ public sealed class QuestDbHistorianProviderTests p.SupportedKinds.Count.ShouldBe(6); } + /// R08:1.1 (#736): the server raw cap the Host binds reaches the data source; without one it is unbounded as before. + [Fact] + public void The_server_raw_cap_reaches_the_data_source() + { + var sp = new NullSp(); + using var plain = Make(); + plain.CreateDataSource(sp).MaxRawValuesPerRead.ShouldBe(0); + + using var capped = new QuestDbHistorianProvider(new QuestDbHistorianOptions + { + ConnectionString = "Host=localhost;Port=18812;Username=admin;Password=quest;Database=qdb;ServerCompatibilityMode=NoTypeLoading", + }, maxRawValuesPerRead: 2500); + capped.CreateDataSource(sp).MaxRawValuesPerRead.ShouldBe(2500); + } + [Fact] public void Creates_all_five_seams_without_touching_questdb() { diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/ReadTestDb.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/ReadTestDb.cs index 0fd95144..ce1cefd9 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/ReadTestDb.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/ReadTestDb.cs @@ -45,8 +45,9 @@ internal sealed class ReadTestDb /// A fresh data source (fresh catalog cache) over this schema; is its "now". public SqlServerHistorianDataSource Source(TimeProvider? clock = null, - int maxTimeWeightedRawRows = TimeWeightedAggregator.DefaultMaxWindowRows) => - new(Conn, Boot, new SqlServerTagCatalog(Conn, Boot), timeProvider: clock, maxTimeWeightedRawRows: maxTimeWeightedRawRows); + int maxTimeWeightedRawRows = TimeWeightedAggregator.DefaultMaxWindowRows, int maxRawValuesPerRead = 0) => + new(Conn, Boot, new SqlServerTagCatalog(Conn, Boot), timeProvider: clock, maxTimeWeightedRawRows: maxTimeWeightedRawRows, + maxRawValuesPerRead: maxRawValuesPerRead); /// Inserts a tag row and the given samples. ServerTimeUtc = source time + 1 s (so tests can tell them apart). public async Task SeedAsync(string tag, params (DateTime T, HistorianValue V, uint Status)[] samples) diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/SqlServerConformanceBackend.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/SqlServerConformanceBackend.cs index 69afa55b..8c5ecc94 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/SqlServerConformanceBackend.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/SqlServerConformanceBackend.cs @@ -26,7 +26,8 @@ internal sealed class SqlServerConformanceBackend : IConformanceBackend public IHistorianDataSource CreateDataSource(TimeProvider? clock = null) => _db.Source(clock); /// - public IHistorianDataSource CreateDataSource(int maxTimeWeightedRawRows) => _db.Source(null, maxTimeWeightedRawRows); + public IHistorianDataSource CreateDataSource(int maxTimeWeightedRawRows, int maxRawValuesPerRead = 0) => + _db.Source(null, maxTimeWeightedRawRows, maxRawValuesPerRead); /// public IHistorianValueWriter CreateValueWriter(Func? utcNow = null) => diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests/SqlServerHistorianProviderTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests/SqlServerHistorianProviderTests.cs index f3ab5176..668c76f7 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests/SqlServerHistorianProviderTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests/SqlServerHistorianProviderTests.cs @@ -47,6 +47,20 @@ public sealed class SqlServerHistorianProviderTests p.SupportedKinds.Count.ShouldBe(6); } + /// R08:1.1 (#736): the server raw cap the Host binds reaches the data source; without one it is unbounded as before. + [Fact] + public void The_server_raw_cap_reaches_the_data_source() + { + var sp = new NullSp(); + Make().CreateDataSource(sp).MaxRawValuesPerRead.ShouldBe(0); + + var capped = new SqlServerHistorianProvider(new SqlServerHistorianOptions + { + ConnectionString = "Server=localhost;User Id=sa;Password=x;TrustServerCertificate=True", + }, maxRawValuesPerRead: 2500); + capped.CreateDataSource(sp).MaxRawValuesPerRead.ShouldBe(2500); + } + [Fact] public void Creates_all_five_seams_without_touching_sql() { diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/ServerHistorianOptionsTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/ServerHistorianOptionsTests.cs index 8e310f30..8ad76302 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/ServerHistorianOptionsTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/ServerHistorianOptionsTests.cs @@ -168,4 +168,64 @@ public sealed class ServerHistorianOptionsTests }.Validate(); Assert.Empty(w); } + + // ── MaxRawValuesPerRead (R08:1.1, #736) ───────────────────────────────────────────────────── + + [Fact] + public void MaxRawValuesPerRead_defaults_to_ten_thousand_and_binds() + { + var o = new ServerHistorianOptions(); + Assert.Equal(10_000, o.MaxRawValuesPerRead); + Assert.Equal(10_000, ServerHistorianOptions.DefaultMaxRawValuesPerRead); + Assert.Equal(10_000, o.EffectiveMaxRawValuesPerRead); + + var bound = new ConfigurationBuilder() + .AddInMemoryCollection(new Dictionary { ["ServerHistorian:MaxRawValuesPerRead"] = "2500" }) + .Build() + .GetSection(ServerHistorianOptions.SectionName) + .Get()!; + Assert.Equal(2500, bound.MaxRawValuesPerRead); + Assert.Equal(2500, bound.EffectiveMaxRawValuesPerRead); + Assert.Empty(new ServerHistorianOptions + { + Enabled = true, Endpoint = "https://h:5222", ApiKey = "histgw_x_y", MaxRawValuesPerRead = 2500, + }.Validate()); + } + + [Theory] + [InlineData(0)] + [InlineData(-5)] + public void Nonpositive_MaxRawValuesPerRead_is_unbounded_and_warns_when_enabled(int value) + { + var o = new ServerHistorianOptions + { + Enabled = true, Endpoint = "https://h:5222", ApiKey = "histgw_x_y", MaxRawValuesPerRead = value, + }; + Assert.Equal(0, o.EffectiveMaxRawValuesPerRead); + var w = o.Validate(); + var m = Assert.Single(w); + Assert.Contains("MaxRawValuesPerRead", m); + Assert.Contains("unbounded", m); + } + + [Fact] + public void A_MaxRawValuesPerRead_of_one_is_raised_to_two_and_warns_when_enabled() + { + // One row per read cannot carry a page AND the lookahead row that says "more follows" — a fresh page would be refused. + var o = new ServerHistorianOptions + { + Enabled = true, Endpoint = "https://h:5222", ApiKey = "histgw_x_y", MaxRawValuesPerRead = 1, + }; + Assert.Equal(2, o.EffectiveMaxRawValuesPerRead); + Assert.Equal(ServerHistorianOptions.MinMaxRawValuesPerRead, o.EffectiveMaxRawValuesPerRead); + var m = Assert.Single(o.Validate()); + Assert.Contains("MaxRawValuesPerRead", m); + } + + [Fact] + public void Disabled_with_an_odd_MaxRawValuesPerRead_is_silent() + { + Assert.Empty(new ServerHistorianOptions { Enabled = false, MaxRawValuesPerRead = 0 }.Validate()); + Assert.Empty(new ServerHistorianOptions { Enabled = false, MaxRawValuesPerRead = 1 }.Validate()); + } }