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
This commit is contained in:
+51
-10
@@ -67,6 +67,12 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
|
||||
/// <summary>The tie-cluster limit the crowding re-reads honour — exposed so the factory wiring is testable.</summary>
|
||||
internal int MaxPreWindowRows => _maxPreWindowRows;
|
||||
|
||||
/// <summary>
|
||||
/// The server raw cap (<c>ServerHistorian:MaxRawValuesPerRead</c>, R08:1.1): the most samples one raw read may
|
||||
/// return; 0 = none. See <see cref="MaxRawValuesPerRead"/> and <see cref="ReadRawWindowAsync"/>.
|
||||
/// </summary>
|
||||
private readonly int _maxRawValuesPerRead;
|
||||
|
||||
/// <summary>A strict gateway's row ceiling, learned from its first refusal.</summary>
|
||||
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 (<c>ServerHistorian:MaxTieClusterOverfetch</c>,
|
||||
/// values below 1 count as 1). More ⇒ <see cref="HistoryReadLimitExceededException"/> (<c>BadTooManyOperations</c>).
|
||||
/// </param>
|
||||
/// <param name="maxRawValuesPerRead">
|
||||
/// The server raw cap (<c>ServerHistorian:MaxRawValuesPerRead</c>, R08:1.1): the most samples one
|
||||
/// <see cref="ReadRawAsync"/> may return — reported as <see cref="MaxRawValuesPerRead"/> (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 <see cref="int.MaxValue"/>).
|
||||
/// </param>
|
||||
public GatewayHistorianDataSource(IHistorianGatewayClient client, ILogger<GatewayHistorianDataSource> 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). <c>ReturnBounds</c> is not plumbed to the
|
||||
/// data source on any backend, so bounding values are never synthesised here.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>The server raw cap</b> (R08:1.1, #736). With a cap, a read asking for everything (<c>0</c>) 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 (<see cref="HistoryReadLimitExceededException"/>) rather than answered truncated (the server's paging
|
||||
/// reads a short answer as the last page) and never sent as <see cref="int.MaxValue"/>. The server itself never asks
|
||||
/// for more than the cap (it pages at <see cref="MaxRawValuesPerRead"/>); this bounds a caller that bypasses it.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
private async Task<List<HistorianSample>> 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<HistorianSample> Bounded(List<HistorianSample> 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<HistorianSample> 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;
|
||||
|
||||
/// <inheritdoc />
|
||||
// 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;
|
||||
}
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
// false: the gateway's ReadEvents streams ascending only. The server never sends it a
|
||||
|
||||
+3
-1
@@ -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);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
|
||||
+34
-4
@@ -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;
|
||||
|
||||
/// <summary>Creates the data source over the provider's shared connection, bootstrapper and tag catalog.</summary>
|
||||
/// <param name="conn">Connection helper.</param>
|
||||
@@ -74,11 +75,19 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource
|
||||
/// <param name="log">Optional logger.</param>
|
||||
/// <param name="timeProvider">Clock for "now" (AtTime beyond the last sample, server timestamps); system by default.</param>
|
||||
/// <param name="maxTimeWeightedRawRows">Raw in-window row cap of one time-weighted read (<= 0 = unbounded).</param>
|
||||
/// <param name="maxRawValuesPerRead">
|
||||
/// The server raw cap (<c>ServerHistorian:MaxRawValuesPerRead</c>, R08:1.1): the most samples one
|
||||
/// <see cref="ReadRawAsync"/> may return — reported as <see cref="MaxRawValuesPerRead"/>, 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).
|
||||
/// </param>
|
||||
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.
|
||||
/// <inheritdoc />
|
||||
public async Task<HistoryReadResult> 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<DataValueSnapshot>();
|
||||
@@ -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
|
||||
/// <inheritdoc />
|
||||
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.
|
||||
/// <inheritdoc />
|
||||
public int MaxRawValuesPerRead => 0;
|
||||
public int MaxRawValuesPerRead => _maxRawValuesPerRead;
|
||||
|
||||
// Reverse events reads are served natively (ORDER BY ... DESC).
|
||||
/// <inheritdoc />
|
||||
|
||||
@@ -35,14 +35,17 @@ public sealed class QuestDbHistorianProvider : IHistorianBackendProvider, IAsync
|
||||
/// <summary>Creates the provider over the <c>ServerHistorian:QuestDb</c> options (validated elsewhere).</summary>
|
||||
/// <param name="options">The <c>ServerHistorian:QuestDb</c> options.</param>
|
||||
/// <param name="maxTimeWeightedRawRows">The shared <c>ServerHistorian:MaxTimeWeightedRawRows</c> cap (<= 0 = unbounded).</param>
|
||||
/// <param name="maxRawValuesPerRead">The shared <c>ServerHistorian:MaxRawValuesPerRead</c> raw cap (<= 0 = unbounded; R08:1.1).</param>
|
||||
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;
|
||||
|
||||
/// <inheritdoc />
|
||||
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<QuestDbHistorianDataSource>(), maxTimeWeightedRawRows: _maxTimeWeightedRawRows);
|
||||
Loggers(sp).CreateLogger<QuestDbHistorianDataSource>(), maxTimeWeightedRawRows: _maxTimeWeightedRawRows,
|
||||
maxRawValuesPerRead: _maxRawValuesPerRead);
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
|
||||
+36
-4
@@ -61,6 +61,7 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource
|
||||
|
||||
private readonly HistorianHealthCounters _health = new();
|
||||
private readonly int _maxTimeWeightedRawRows;
|
||||
private readonly int _maxRawValuesPerRead;
|
||||
|
||||
/// <summary>Creates the data source over the provider's shared connection, bootstrapper and tag catalog.</summary>
|
||||
/// <param name="conn">Connection helper.</param>
|
||||
@@ -69,11 +70,19 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource
|
||||
/// <param name="log">Optional logger.</param>
|
||||
/// <param name="timeProvider">Clock for "now" (AtTime beyond the last sample, server timestamps); system by default.</param>
|
||||
/// <param name="maxTimeWeightedRawRows">Raw in-window row cap of one time-weighted read (<= 0 = unbounded).</param>
|
||||
/// <param name="maxRawValuesPerRead">
|
||||
/// The server raw cap (<c>ServerHistorian:MaxRawValuesPerRead</c>, R08:1.1): the most samples one
|
||||
/// <see cref="ReadRawAsync"/> may return — reported as <see cref="MaxRawValuesPerRead"/>, 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).
|
||||
/// </param>
|
||||
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
|
||||
// <c>0</c> is the node manager's "unbounded" sentinel; values above <see cref="int.MaxValue"/> clamp.
|
||||
// The node manager pages by re-reading from the last returned timestamp inclusive and trimming it; the
|
||||
// <c>(TagId, SourceTimeUtc)</c> 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.
|
||||
// </remarks>
|
||||
/// <inheritdoc />
|
||||
public async Task<HistoryReadResult> 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
|
||||
/// <inheritdoc />
|
||||
public bool SupportsReverseRaw => true;
|
||||
|
||||
// <remarks><c>0</c>: a capped raw read is one bounded statement — no backend row ceiling.</remarks>
|
||||
// <remarks>
|
||||
// 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.
|
||||
// </remarks>
|
||||
/// <inheritdoc />
|
||||
public int MaxRawValuesPerRead => 0;
|
||||
public int MaxRawValuesPerRead => _maxRawValuesPerRead;
|
||||
|
||||
// <remarks>Reverse events reads are served natively (<c>ORDER BY … DESC</c>).</remarks>
|
||||
/// <inheritdoc />
|
||||
|
||||
+6
-2
@@ -31,14 +31,17 @@ public sealed class SqlServerHistorianProvider : IHistorianBackendProvider
|
||||
/// <summary>Creates the provider over the <c>ServerHistorian:SqlServer</c> options (validated elsewhere).</summary>
|
||||
/// <param name="options">The <c>ServerHistorian:SqlServer</c> options.</param>
|
||||
/// <param name="maxTimeWeightedRawRows">The shared <c>ServerHistorian:MaxTimeWeightedRawRows</c> cap (<= 0 = unbounded).</param>
|
||||
/// <param name="maxRawValuesPerRead">The shared <c>ServerHistorian:MaxRawValuesPerRead</c> raw cap (<= 0 = unbounded; R08:1.1).</param>
|
||||
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;
|
||||
|
||||
/// <inheritdoc />
|
||||
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<SqlServerHistorianDataSource>(), maxTimeWeightedRawRows: _maxTimeWeightedRawRows);
|
||||
Loggers(sp).CreateLogger<SqlServerHistorianDataSource>(), maxTimeWeightedRawRows: _maxTimeWeightedRawRows,
|
||||
maxRawValuesPerRead: _maxRawValuesPerRead);
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
|
||||
@@ -45,8 +45,8 @@ public static class HistorianBackendRegistry
|
||||
new Dictionary<string, Func<ServerHistorianOptions, IConfiguration, IHistorianBackendProvider>>(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),
|
||||
};
|
||||
|
||||
/// <summary>The registered backend names, for validation messages.</summary>
|
||||
|
||||
@@ -154,6 +154,38 @@ public sealed class ServerHistorianOptions
|
||||
/// </summary>
|
||||
public int MaxTimeWeightedRawRows { get; init; } = 1_000_000;
|
||||
|
||||
/// <summary>The default of <see cref="MaxRawValuesPerRead"/>: 10 000 values per raw page / backend read.</summary>
|
||||
public const int DefaultMaxRawValuesPerRead = 10_000;
|
||||
|
||||
/// <summary>
|
||||
/// The smallest usable <see cref="MaxRawValuesPerRead"/>: 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 (<see cref="EffectiveMaxRawValuesPerRead"/>).
|
||||
/// </summary>
|
||||
public const int MinMaxRawValuesPerRead = 2;
|
||||
|
||||
/// <summary>
|
||||
/// 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
|
||||
/// <c>NumValuesPerNode = 0</c> ("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 (<c>RawPageLimit</c>: page + boundary ties + lookahead ≤ cap, so a first
|
||||
/// page holds cap − 1 values). A <c>NumValuesPerNode</c> above it is paged at it too. Advertised as
|
||||
/// <c>HistoryServerCapabilities/MaxReturnDataValues</c>. The backends' own <c>TOP</c> / <c>LIMIT</c> / <c>max_values</c>
|
||||
/// 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 (<c>BadTooManyOperations</c>) 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; <see cref="MaxTieClusterOverfetch"/> above that is inert). Default
|
||||
/// 10 000. Non-positive = unbounded, the pre-#736 behaviour (a <see cref="Validate"/> warning); 1 is raised to 2.
|
||||
/// </summary>
|
||||
public int MaxRawValuesPerRead { get; init; } = DefaultMaxRawValuesPerRead;
|
||||
|
||||
/// <summary>
|
||||
/// <see cref="MaxRawValuesPerRead"/> as the backends and the node manager apply it: <c>0</c> (unbounded) when
|
||||
/// non-positive, otherwise at least <see cref="MinMaxRawValuesPerRead"/>. The one place the rule lives.
|
||||
/// </summary>
|
||||
public int EffectiveMaxRawValuesPerRead => MaxRawValuesPerRead <= 0 ? 0 : Math.Max(MinMaxRawValuesPerRead, MaxRawValuesPerRead);
|
||||
|
||||
/// <summary>
|
||||
/// how far into the future an OPC UA HistoryUpdate (Insert / Replace / Update) item's
|
||||
/// <c>SourceTimestamp</c> may lie, measured from the server clock. An item beyond <c>now + skew</c> 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)
|
||||
|
||||
@@ -22,9 +22,14 @@ public interface IConformanceBackend : IAsyncDisposable
|
||||
/// <param name="clock">Its "now"; null means the system clock.</param>
|
||||
IHistorianDataSource CreateDataSource(TimeProvider? clock = null);
|
||||
|
||||
/// <summary>A fresh data source whose time-weighted raw-row cap (<c>ServerHistorian:MaxTimeWeightedRawRows</c>) is <paramref name="maxTimeWeightedRawRows"/>.</summary>
|
||||
/// <param name="maxTimeWeightedRawRows">The cap (<= 0 = unbounded).</param>
|
||||
IHistorianDataSource CreateDataSource(int maxTimeWeightedRawRows);
|
||||
/// <summary>
|
||||
/// A fresh data source whose time-weighted raw-row cap (<c>ServerHistorian:MaxTimeWeightedRawRows</c>) is
|
||||
/// <paramref name="maxTimeWeightedRawRows"/> and whose server raw cap (<c>ServerHistorian:MaxRawValuesPerRead</c>,
|
||||
/// R08:1.1 / #736) is <paramref name="maxRawValuesPerRead"/>.
|
||||
/// </summary>
|
||||
/// <param name="maxTimeWeightedRawRows">The time-weighted cap (<= 0 = unbounded).</param>
|
||||
/// <param name="maxRawValuesPerRead">The raw cap (<= 0 = unbounded, the default).</param>
|
||||
IHistorianDataSource CreateDataSource(int maxTimeWeightedRawRows, int maxRawValuesPerRead = 0);
|
||||
|
||||
/// <summary>A fresh live-value writer over this backend's store.</summary>
|
||||
/// <param name="utcNow">The clock used to stamp values that carry no timestamp; null means the system clock.</param>
|
||||
|
||||
@@ -25,6 +25,43 @@ public abstract class RawReadConformance : ConformanceBase
|
||||
r.ContinuationPoint.ShouldBeNull();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// R08:1.1 (#736): the server raw cap bounds the backend's own <c>TOP</c> / <c>LIMIT</c>. 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.
|
||||
/// </summary>
|
||||
[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<HistoryReadLimitExceededException>(() => src.ReadRawAsync("A", At(-1), At(100), 0, Ct));
|
||||
refused.Limit.ShouldBe(3);
|
||||
await Should.ThrowAsync<HistoryReadLimitExceededException>(() => src.ReadRawAsync("A", At(-1), At(100), 4, Ct));
|
||||
await Should.ThrowAsync<HistoryReadLimitExceededException>(() => 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<HistoryReadLimitExceededException>(() => 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()
|
||||
{
|
||||
|
||||
@@ -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;
|
||||
|
||||
/// <summary>
|
||||
/// Review R08:1.1 (#736): the server's raw cap (<c>ServerHistorian:MaxRawValuesPerRead</c>) bounds the Gateway
|
||||
/// backend's <c>max_values</c> independently of the node manager. An uncapped raw read (<c>NumValuesPerNode = 0</c>)
|
||||
/// used to be sent as <see cref="int.MaxValue"/> 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 (<see cref="HistoryReadLimitExceededException"/>,
|
||||
/// <c>BadTooManyOperations</c>) 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
|
||||
/// <see cref="IHistorianDataSource.MaxRawValuesPerRead"/> (or a strict gateway's lower learned ceiling), so the node
|
||||
/// manager's existing ceiling paging serves <c>NumValuesPerNode = 0</c> as pages of at most the cap, each with a
|
||||
/// continuation point.
|
||||
/// </summary>
|
||||
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 };
|
||||
|
||||
/// <summary>AVEVA-shaped: a boundary row at the query start, <c>[start, end]</c> inclusive, capped at <c>max</c>.</summary>
|
||||
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<HistorianSample>();
|
||||
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<GatewayHistorianDataSource>.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<GatewayHistorianDataSource>.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<HistoryReadLimitExceededException>(() => 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<HistoryReadLimitExceededException>(() => 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<HistoryReadLimitExceededException>(() => 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<HistoryReadLimitExceededException>(() => 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<GatewayHistorianDataSource>().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<OpcUaApplicationHost>.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<double> Values, List<int> 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<double>();
|
||||
var pages = new List<int>();
|
||||
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<double> 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<HistoryReadValueId> { new() { NodeId = nodeId, ContinuationPoint = inboundCp } };
|
||||
var results = new List<SdkHistoryReadResult> { null! };
|
||||
var errors = new List<ServiceResult> { null! };
|
||||
nm.HistoryRead(context, details, TimestampsToReturn.Both, releaseContinuationPoints: false, nodesToRead, results, errors);
|
||||
|
||||
var values = new List<double>();
|
||||
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 */ }
|
||||
}
|
||||
}
|
||||
}
|
||||
+3
-2
@@ -110,8 +110,9 @@ internal sealed class QuestDbConformanceBackend : IConformanceBackend
|
||||
new QuestDbHistorianDataSource(Conn, Boot, NewCatalog(), timeProvider: clock);
|
||||
|
||||
/// <inheritdoc/>
|
||||
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);
|
||||
|
||||
/// <inheritdoc/>
|
||||
/// <remarks>The T4 live-value writer over a fresh catalog (fresh positive cache) and this instance's WAL wait.</remarks>
|
||||
|
||||
+15
@@ -47,6 +47,21 @@ public sealed class QuestDbHistorianProviderTests
|
||||
p.SupportedKinds.Count.ShouldBe(6);
|
||||
}
|
||||
|
||||
/// <summary>R08:1.1 (#736): the server raw cap the Host binds reaches the data source; without one it is unbounded as before.</summary>
|
||||
[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()
|
||||
{
|
||||
|
||||
+3
-2
@@ -45,8 +45,9 @@ internal sealed class ReadTestDb
|
||||
|
||||
/// <summary>A fresh data source (fresh catalog cache) over this schema; <paramref name="clock"/> is its "now".</summary>
|
||||
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);
|
||||
|
||||
/// <summary>Inserts a tag row and the given samples. ServerTimeUtc = source time + 1 s (so tests can tell them apart).</summary>
|
||||
public async Task<int> SeedAsync(string tag, params (DateTime T, HistorianValue V, uint Status)[] samples)
|
||||
|
||||
+2
-1
@@ -26,7 +26,8 @@ internal sealed class SqlServerConformanceBackend : IConformanceBackend
|
||||
public IHistorianDataSource CreateDataSource(TimeProvider? clock = null) => _db.Source(clock);
|
||||
|
||||
/// <inheritdoc/>
|
||||
public IHistorianDataSource CreateDataSource(int maxTimeWeightedRawRows) => _db.Source(null, maxTimeWeightedRawRows);
|
||||
public IHistorianDataSource CreateDataSource(int maxTimeWeightedRawRows, int maxRawValuesPerRead = 0) =>
|
||||
_db.Source(null, maxTimeWeightedRawRows, maxRawValuesPerRead);
|
||||
|
||||
/// <inheritdoc/>
|
||||
public IHistorianValueWriter CreateValueWriter(Func<DateTime>? utcNow = null) =>
|
||||
|
||||
+14
@@ -47,6 +47,20 @@ public sealed class SqlServerHistorianProviderTests
|
||||
p.SupportedKinds.Count.ShouldBe(6);
|
||||
}
|
||||
|
||||
/// <summary>R08:1.1 (#736): the server raw cap the Host binds reaches the data source; without one it is unbounded as before.</summary>
|
||||
[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()
|
||||
{
|
||||
|
||||
@@ -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<string, string?> { ["ServerHistorian:MaxRawValuesPerRead"] = "2500" })
|
||||
.Build()
|
||||
.GetSection(ServerHistorianOptions.SectionName)
|
||||
.Get<ServerHistorianOptions>()!;
|
||||
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());
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user