Merge lane H1: HistoryRead raw cap (#736)

Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
This commit is contained in:
Joseph Doherty
2026-10-22 10:41:54 -04:00
22 changed files with 1212 additions and 62 deletions
@@ -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. &lt;= 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,30 +231,39 @@ 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);
if (cap == 0)
var serverCap = _maxRawValuesPerRead;
// Under the server raw cap, a read asking for everything or for more than the cap runs as a capped read of
// cap + 1 (budgetCap); `cap` stays what the client asked for, and a refusal names the cap that applied (limit).
var probe = serverCap > 0 && (cap == 0 || cap > serverCap);
var budgetCap = probe ? (int)Math.Min(int.MaxValue, (long)serverCap + 1) : cap;
var limit = probe ? serverCap : cap;
var advice = probe
? $"set NumValuesPerNode at or below ServerHistorian:MaxRawValuesPerRead ({serverCap}) to page"
: cap == 0 ? "set NumValuesPerNode to page" : "lower NumValuesPerNode";
if (budgetCap == 0)
{
var all = await PassAsync(int.MaxValue).ConfigureAwait(false);
if (all.Result.Complete) return all.Result.Samples;
// Only reachable on a strict gateway: the ceiling re-read delivered the whole ceiling without reaching
// the window end — the window may hold more, and "all values" is never answered from a truncated read.
throw new HistoryReadLimitExceededException(cap,
throw new HistoryReadLimitExceededException(limit,
$"The raw read of '{fullReference}' holds more than the HistorianGateway's row ceiling ({all.Budget}, its " +
"Historian:ReadLimits:MaxValuesCeiling with AllowStreamingBeyondCeiling=false); set NumValuesPerNode to page, " +
"or narrow the window.")
{ PreWindowRows = Crowding(all.Result.PreWindowRows) };
}
var asked = (int)Math.Min(int.MaxValue, (long)cap + RawAggregateReader.PreWindowAllowance);
var asked = (int)Math.Min(int.MaxValue, (long)budgetCap + 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.
if (pass.Budget < asked)
throw new HistoryReadLimitExceededException(cap,
throw new HistoryReadLimitExceededException(limit,
$"The raw read of '{fullReference}' could not be read completely within the HistorianGateway's row ceiling " +
$"({pass.Budget}, its Historian:ReadLimits:MaxValuesCeiling with AllowStreamingBeyondCeiling=false); lower " +
"NumValuesPerNode or narrow the window.")
$"({pass.Budget}, its Historian:ReadLimits:MaxValuesCeiling with AllowStreamingBeyondCeiling=false); {advice}, " +
"or narrow the window.")
{ PreWindowRows = Crowding(pass.Result.PreWindowRows) };
// The margin crowded the window out: re-read with room for the pre-window rows. A read that was itself cut short saw
@@ -245,25 +275,25 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
var room = Math.Min(Math.Max(1, preWindow), _maxPreWindowRows);
for (var attempt = 0; attempt < MaxCrowdedReReads && preWindow <= _maxPreWindowRows; attempt++)
{
var budget = (int)Math.Min(int.MaxValue, (long)cap + room + RawAggregateReader.PreWindowAllowance);
var budget = (int)Math.Min(int.MaxValue, (long)budgetCap + room + RawAggregateReader.PreWindowAllowance);
// doubling can leap over the ceiling while the cluster still fits. Spend the remaining room once.
if (_ceiling.Learned is var known and > 0 && budget > known)
{
var available = known - cap - RawAggregateReader.PreWindowAllowance;
var available = known - budgetCap - RawAggregateReader.PreWindowAllowance;
if (available <= 0) break;
room = Math.Min(room, available);
budget = cap + room + RawAggregateReader.PreWindowAllowance;
budget = budgetCap + room + RawAggregateReader.PreWindowAllowance;
}
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
room = (int)Math.Min(_maxPreWindowRows, Math.Max(preWindow, 2L * room));
}
throw new HistoryReadLimitExceededException(cap,
throw new HistoryReadLimitExceededException(limit,
$"The raw read of '{fullReference}' could not be read completely: more than {pass.Result.PreWindowRows} rows lie in " +
$"the {RawAggregateReader.BoundaryMargin.TotalMilliseconds} ms before the window and exhausted the row budget.")
{ PreWindowRows = preWindow };
@@ -271,10 +301,20 @@ 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,
b => ReadRawPassAsync(fullReference, startUtc, endUtc, cap, b, cancellationToken),
b => ReadRawPassAsync(fullReference, startUtc, endUtc, budgetCap, b, cancellationToken),
(r, _) => !r.Complete);
}
@@ -531,13 +571,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
@@ -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>
@@ -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 (&lt;= 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.
/// &lt;= 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 (&lt;= 0 = unbounded).</param>
/// <param name="maxRawValuesPerRead">The shared <c>ServerHistorian:MaxRawValuesPerRead</c> raw cap (&lt;= 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 />
@@ -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 (&lt;= 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.
/// &lt;= 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 />
@@ -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 (&lt;= 0 = unbounded).</param>
/// <param name="maxRawValuesPerRead">The shared <c>ServerHistorian:MaxRawValuesPerRead</c> raw cap (&lt;= 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>
@@ -1,3 +1,5 @@
using Opc.Ua;
using Opc.Ua.Server;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
namespace ZB.MOM.WW.OtOpcUa.OpcUaServer;
@@ -110,24 +112,62 @@ internal enum HistoryCursorKind
/// <c>NumValuesPerNode</c> samples with a null continuation point — so paging is synthesised here,
/// time-based:
/// <list type="bullet">
/// <item>A page that returns EXACTLY the requested cap (<c>NumValuesPerNode &gt; 0</c>) MAY have
/// more behind it ⇒ emit a continuation point.</item>
/// <item>A short page (fewer than the cap) is the last page ⇒ no continuation point.</item>
/// <item><c>NumValuesPerNode == 0</c> means "all values, no limit" (OPC UA Part 11) ⇒ never page;
/// return everything in one shot.</item>
/// <item>A page whose read returned the lookahead row past it MAY have more behind it ⇒ emit a continuation
/// point (<see cref="IsFullPage"/> judges a read against what was asked for).</item>
/// <item>A short read (no lookahead row) is the last page ⇒ no continuation point.</item>
/// <item><c>NumValuesPerNode == 0</c> means "all values" (OPC UA Part 11) — but not "in one shot": the page is
/// sized to the backend's <see cref="IHistorianDataSource.MaxRawValuesPerRead"/> by <see cref="RawPageLimit"/>
/// (the server raw cap <c>ServerHistorian:MaxRawValuesPerRead</c>, default 10 000, and/or a strict gateway's
/// learned ceiling; review R08:1.1, #736), and a full page carries a continuation point like any other. Only
/// with no limit at all (<c>0</c>, the opt-out) is everything returned in one shot.</item>
/// </list>
/// All methods are static + pure so they unit-test without a server, a session, or the SDK.
/// All paging methods are static + pure so they unit-test without a server, a session, or the SDK;
/// <see cref="AdvertiseMaxReturnDataValues"/> is the one SDK-facing member.
/// </summary>
internal static class HistoryPaging
{
/// <summary>
/// Decide whether a just-returned page is "full" and therefore MAY be followed by more data —
/// the signal to emit a continuation point. A page is full when the client asked for a finite
/// cap (<paramref name="numValuesPerNode"/> &gt; 0) and the backend returned exactly that many
/// samples. A short page (or an unlimited <c>0</c> request) is terminal.
/// Advertise the server raw cap as <c>HistoryServerCapabilities/MaxReturnDataValues</c> (i=11273; Part 11 §5.4.2:
/// "the maximum number of values that can be returned by the Server for each Node", 0 = no limit). The cap is
/// the most values one HistoryRead-Raw page returns (<c>ServerHistorian:MaxRawValuesPerRead</c>), so a client can
/// size its <c>NumValuesPerNode</c> and expect a continuation point past it. Idempotent; a re-sync with a
/// different cap overwrites. Never throws: a missing diagnostics manager or node, or any SDK failure writing the
/// property, leaves the SDK default (0) and returns <c>false</c> for the caller to log — HistoryRead itself still
/// enforces the cap.
/// </summary>
/// <param name="returnedCount">The number of samples the backend returned for this page.</param>
/// <param name="numValuesPerNode">The client's per-page cap; <c>0</c> means "no limit".</param>
/// <param name="server">The running server.</param>
/// <param name="maxRawValuesPerRead">The cap (&lt;= 0 advertises 0 = no limit).</param>
/// <returns><c>true</c> when the property was written.</returns>
public static bool AdvertiseMaxReturnDataValues(IServerInternal? server, int maxRawValuesPerRead)
{
if (server?.DiagnosticsNodeManager is not DiagnosticsNodeManager diagnostics) return false;
try
{
lock (diagnostics.Lock)
{
var capabilities = diagnostics.GetDefaultHistoryCapabilities();
if (capabilities?.MaxReturnDataValues is not { } property) return false;
property.Value = (uint)Math.Max(0, maxRawValuesPerRead);
property.ClearChangeMasks(diagnostics.SystemContext, false);
return true;
}
}
catch (Exception)
{
return false; // advisory only: the caller logs; the cap is enforced by HistoryRead regardless
}
}
/// <summary>
/// Decide whether a just-returned read is "full" and therefore MAY be followed by more data —
/// the signal to emit a continuation point. A read is full when it was issued with a finite request
/// cap (<paramref name="numValuesPerNode"/> &gt; 0 — the <c>RawPagePlan.RequestCap</c>: page + boundary ties +
/// lookahead, which under a backend limit or the server raw cap is always finite, uncapped client reads included)
/// and the backend returned at least that many samples. A short read is terminal; so is a read issued with a
/// request cap of <c>0</c>, which only happens with no limit at all (one shot of the whole window).
/// </summary>
/// <param name="returnedCount">The number of samples the backend returned for this read.</param>
/// <param name="numValuesPerNode">The request cap the read was issued with; <c>0</c> means "no limit".</param>
/// <returns><c>true</c> when a continuation point should be emitted; otherwise <c>false</c>.</returns>
public static bool IsFullPage(int returnedCount, uint numValuesPerNode) =>
numValuesPerNode > 0 && returnedCount >= numValuesPerNode;
@@ -11,17 +11,22 @@ internal readonly record struct RawPagePlan(int Limit, uint PageCap, uint Reques
/// <summary>
/// sizes a HistoryRead-Raw page to what the backend can return in one read
/// (<see cref="IHistorianDataSource.MaxRawValuesPerRead"/> = L — a strict HistorianGateway's learned row ceiling, less
/// its boundary-row allowance). With a limit the read — page P + boundary ties k + the lookahead row — is at most L,
/// so P = min(want, L − k − 1) (all values: P = L − k − 1); a full page carries a continuation point like any other, as
/// Part 11 allows fewer values than <c>NumValuesPerNode</c> with one. Previously a cap above a strict gateway's ceiling
/// (or an uncapped read of a window holding more) failed <c>BadTooManyOperations</c>.
/// (<see cref="IHistorianDataSource.MaxRawValuesPerRead"/> = L — the server raw cap
/// <c>ServerHistorian:MaxRawValuesPerRead</c>, default 10 000, which every backend reports (review R08:1.1, #736),
/// and/or a strict HistorianGateway's learned row ceiling less its boundary-row allowance, whichever is lower). With
/// a limit the read — page P + boundary ties k + the lookahead row — is at most L, so P = min(want, L − k − 1) (all
/// values: P = L − k − 1, i.e. a first page of 9 999 at the default cap); a full page carries a continuation point like
/// any other, as Part 11 allows fewer values than <c>NumValuesPerNode</c> with one. Previously a cap above a strict
/// gateway's ceiling (or an uncapped read of a window holding more) failed <c>BadTooManyOperations</c>, and before
/// #736 an uncapped read on a backend without a ceiling read the whole window in one shot.
/// </summary>
/// <remarks>
/// <para>
/// <b>Never a silent truncation.</b> The backend still answers a read in full or refuses it; only the server's page is
/// smaller, and a page is judged full against what was asked for. A read the backend cannot serve at all (a tie
/// cluster larger than one read) stays <c>BadTooManyOperations</c>.
/// cluster larger than one read) stays <c>BadTooManyOperations</c>. The backends enforce the same cap themselves
/// (a read asking for more than L, or for everything, is refused when the window holds more than L rows), so this
/// planner is what keeps the server's own reads inside it: with L &gt; 0 no read it plans ever asks for 0 or more than L.
/// </para>
/// <para>
/// <b>One re-read, planned again.</b> A refused read (<see cref="HistoryReadLimitExceededException"/>) is re-planned and
@@ -154,6 +154,42 @@ 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 resumed page's read must carry at least one value, the
/// boundary tie it re-reads and trims, AND the lookahead row that tells the server more data follows. At 2 a resumed
/// page has no room at all and every value costs an empty page with a continuation point (the #692 defect, re-created
/// by configuration), so 1 and 2 are raised to 3 (<see cref="EffectiveMaxRawValuesPerRead"/>).
/// </summary>
public const int MinMaxRawValuesPerRead = 3;
/// <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), and a
/// page whose boundary tie cluster is that large steps past it with one empty page. Default 10 000. Non-positive =
/// unbounded, the pre-#736 behaviour (a <see cref="Validate"/> warning); below <see cref="MinMaxRawValuesPerRead"/>
/// (3) is raised to it.
/// </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 +244,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 value, its boundary tie and the 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 (&lt;= 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 (&lt;= 0 = unbounded).</param>
/// <param name="maxRawValuesPerRead">The raw cap (&lt;= 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 */ }
}
}
}
@@ -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>
@@ -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()
{
@@ -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)
@@ -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) =>
@@ -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()
{
@@ -0,0 +1,58 @@
using Opc.Ua;
using Opc.Ua.Server;
using Shouldly;
using Xunit;
namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests;
/// <summary>
/// R08:1.1 (#736): the server raw cap (<c>ServerHistorian:MaxRawValuesPerRead</c>) is advertised as
/// <c>HistoryServerCapabilities/MaxReturnDataValues</c> (i=11273) through <see cref="HistoryPaging.AdvertiseMaxReturnDataValues"/>,
/// read back the way a client would — through the SDK master node manager — over a real server.
/// </summary>
public sealed class HistoryServerCapabilitiesAdvertisementTests(BoundsServerFixture fx) : IClassFixture<BoundsServerFixture>
{
private async Task<uint> ReadAdvertisedAsync()
{
var context = new OperationContext(new RequestHeader(), secureChannelContext: null, RequestType.Read, identity: null);
var nodesToRead = new ReadValueIdCollection
{
new ReadValueId { NodeId = VariableIds.HistoryServerCapabilities_MaxReturnDataValues, AttributeId = Attributes.Value },
};
var (values, _) = await fx.Nm.Server.NodeManager.ReadAsync(context, 0, TimestampsToReturn.Neither, nodesToRead, TestContext.Current.CancellationToken);
values.Count.ShouldBe(1);
StatusCode.IsGood(values[0].StatusCode).ShouldBeTrue(values[0].StatusCode.ToString());
return (uint)values[0].Value;
}
[Fact]
public async Task The_cap_is_advertised_as_MaxReturnDataValues_and_a_re_sync_overwrites_it()
{
await fx.Gate.WaitAsync(TestContext.Current.CancellationToken);
try
{
HistoryPaging.AdvertiseMaxReturnDataValues(fx.Nm.Server, 10_000).ShouldBeTrue();
(await ReadAdvertisedAsync()).ShouldBe(10_000u);
HistoryPaging.AdvertiseMaxReturnDataValues(fx.Nm.Server, 2_500).ShouldBeTrue();
(await ReadAdvertisedAsync()).ShouldBe(2_500u);
// Unbounded (the opt-out) advertises Part 11's "no limit".
HistoryPaging.AdvertiseMaxReturnDataValues(fx.Nm.Server, 0).ShouldBeTrue();
(await ReadAdvertisedAsync()).ShouldBe(0u);
HistoryPaging.AdvertiseMaxReturnDataValues(fx.Nm.Server, -5).ShouldBeTrue();
(await ReadAdvertisedAsync()).ShouldBe(0u);
}
finally
{
HistoryPaging.AdvertiseMaxReturnDataValues(fx.Nm.Server, 0);
fx.Gate.Release();
}
}
[Fact]
public void No_server_is_a_quiet_false()
{
HistoryPaging.AdvertiseMaxReturnDataValues(null, 10_000).ShouldBeFalse();
}
}
@@ -0,0 +1,436 @@
using Opc.Ua;
using Opc.Ua.Server;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using HistorianRead = ZB.MOM.WW.OtOpcUa.Core.Abstractions.HistoryReadResult;
using SdkHistoryReadResult = Opc.Ua.HistoryReadResult;
namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests;
/// <summary>
/// R08:1.1 (#736): the paged-vs-unpaged fuzz for the server raw cap. Every paging change in this repo has had boundary
/// bugs that the hand-written cases missed (silent truncation, an endless cursor, an empty page between every value), so
/// the cap is proved the only way that has worked: random series (bursts, gaps, sub-millisecond stamps, tie clusters),
/// random windows (empty, on a tie, one step past a tie, between samples, forward / reverse / end-only, with and without
/// bounds) and random caps — <c>NumValuesPerNode</c> 0 / 1 / 2 / small / huge, the server cap 3 … 1000 or off, the
/// tie-cluster budget 3 / 8 / 65536 — paged through a REAL SDK node manager against a wire-faithful fake of each backend
/// (SqlServer: tick precision, native reverse, <c>TOP</c>; QuestDB: µs truncation of bounds AND stamps, native reverse,
/// <c>LIMIT</c>; Gateway: forward only, <c>[start, end)</c> after its own boundary-row filtering, walked back by the
/// server) that enforces the backend half of the cap exactly as the real sources do (a read asking for 0 or more than
/// the cap reads cap + 1 rows and is refused when the window holds more).
/// <para>
/// The oracle is the same node manager's one-shot answer with the cap OFF (<c>NumValuesPerNode = 0</c>, no limit),
/// which for a plain read is also checked against the series itself. The concatenation of the pages must equal it
/// exactly; the read must terminate within the number of values plus two pages; no page may exceed the cap; the fake
/// must never see a request for 0 or more than the cap (the "bounded by the same cap" invariant); and a
/// <c>BadTooManyOperations</c> is accepted only when a tie cluster (or, on the Gateway shape, one millisecond) is
/// larger than the budget that the cap and <c>MaxTieClusterOverfetch</c> leave — after an exact prefix. The run is
/// non-vacuous by construction: it asserts that empty windows, mid-tie windows, reverse reads, <c>NumValuesPerNode</c>
/// of 1 and 2, the smallest server cap (3) and every shape occurred, and that most iterations were exact.
/// </para>
/// <para>
/// Iterations: <c>OTOPCUA_PAGING_FUZZ_ITERATIONS</c> (default 10 000, the lane gate). Seeds are stable, so a failure
/// names a reproducible seed.
/// </para>
/// </summary>
public sealed class NodeManagerHistoryReadRawCapFuzzTests(BoundsServerFixture fx) : IClassFixture<BoundsServerFixture>
{
private static readonly DateTime T0 = new(2026, 10, 22, 12, 0, 0, DateTimeKind.Utc);
/// <summary>The backend the fake is shaped like.</summary>
public enum Shape { SqlServer, QuestDb, Gateway }
private enum Direction { Forward, Reverse, EndOnly }
private static int Iterations =>
int.TryParse(Environment.GetEnvironmentVariable("OTOPCUA_PAGING_FUZZ_ITERATIONS"), out var n) && n > 0 ? n : 10_000;
[Fact]
public async Task Paged_equals_unpaged_on_random_series_windows_and_caps_and_always_terminates()
{
var iterations = Iterations;
var counts = new Dictionary<string, int>();
void Count(string key) => counts[key] = counts.GetValueOrDefault(key) + 1;
await fx.Gate.WaitAsync(TestContext.Current.CancellationToken);
try
{
for (var seed = 0; seed < iterations; seed++)
{
var rng = new Random(seed);
var shape = (Shape)(seed % 3);
var resolution = shape == Shape.QuestDb ? 10L : 1L;
var series = RandomSeries(rng, resolution);
var cap = rng.Next(10) switch { 0 => 0, 1 or 2 => 3, 3 => 4, 4 => 5, 5 => 7, 6 => 10, 7 => 16, 8 => 33, _ => 1000 };
var n = rng.Next(9) switch { 0 or 1 => 0u, 2 or 3 => 1u, 4 or 5 => 2u, 6 => (uint)rng.Next(3, 20), 7 => (uint)rng.Next(20, 200), _ => 100_000u };
var tieBudget = rng.Next(4) switch { 0 => 3, 1 => 8, _ => 65536 };
var direction = rng.Next(5) switch { 0 => Direction.Reverse, 1 => Direction.EndOnly, _ => Direction.Forward };
if (direction == Direction.EndOnly && n == 0) direction = Direction.Forward; // Part 11: one bound only with a cap
var bounds = direction != Direction.EndOnly && rng.Next(4) == 0;
var a = PickBoundary(rng, series, resolution);
var b = PickBoundary(rng, series, resolution);
var (lo, hi) = a <= b ? (a, b) : (b, a);
var (start, end) = direction switch
{
Direction.Forward => (lo, hi),
Direction.Reverse => (hi, lo),
_ => (DateTime.MinValue, hi),
};
var tag = $"seed {seed} {shape} cap {cap} N {n} tie {tieBudget} {direction} bounds {bounds} [{start:O} .. {end:O}]";
// The walk-back (Gateway shape only: reverse / end-only reads, and the bound lookups of any bounded read) refuses
// a millisecond holding more samples than its budget — computed up front, over the whole series, because the
// walk and the lookups read around the window.
var walks = shape == Shape.Gateway && (direction != Direction.Forward || bounds);
var msMax = walks ? MillisecondMax(series) : 0;
// The oracle: the one-shot answer with the cap off (and, for a plain read, the series itself).
var truthSource = new CappedFake(series, shape, cap: 0);
fx.Nm.HistoricalSourceForTest(truthSource, tieBudget);
// An end-only read is paged backward from EndTime (the node manager flips it to a reverse read down to the
// beginning of time); its oracle is the explicit reverse read down to the Unix epoch — the walk-back's floor,
// and Part 11 (the SDK checks it) needs both bounds set when NumValuesPerNode = 0.
var (truth, truthError, truthCp) = direction == Direction.EndOnly
? Read(hi, DateTime.UnixEpoch, 0, bounds: false, cp: null)
: Read(start, end, 0, bounds, cp: null);
if (truthError.StatusCode.Code == StatusCodes.BadTooManyOperations)
{
// Only the walk-back can refuse an uncapped one-shot read, and only past its budget (max(1, M) with no cap).
(walks && msMax > Math.Max(1, tieBudget)).ShouldBeTrue($"the oracle read was refused: {tag} (ms max {msMax})");
Count("oracle refused");
continue;
}
truthError.StatusCode.Code.ShouldBe(StatusCodes.Good, $"oracle: {tag}");
truthCp.ShouldBeNull($"oracle (no cap, N = 0) is one page: {tag}");
if (!bounds)
Render(truth).ShouldBe(Render(Oracle(series, start, end, direction, resolution)), $"oracle vs series: {tag}");
truthSource.BadRequests.ShouldBe(0, $"a reverse window reached a forward-only backend on the oracle read: {tag}");
var windowTies = LargestTieCluster(truth);
if (truth.Count == 0) Count("empty");
if (windowTies >= 2) Count("ties in window");
if (EndsOnTie(series, start, end, direction, resolution)) Count("window ends on a tie");
if (direction == Direction.Reverse) Count("reverse");
if (direction == Direction.EndOnly) Count("end-only");
if (n == 1) Count("N=1");
if (n == 2) Count("N=2");
if (n == 0) Count("N=0");
if (cap == 3) Count("cap=3");
if (bounds) Count("bounds");
Count(shape.ToString());
// A boundary tie cluster as large as the cap leaves a resumed page no room: the server steps past it with one
// empty page carrying a continuation point (the documented #694 ceiling behaviour) — the one empty page allowed.
var oversizedClusters = cap > 0
? truth.Where(v => !(v.Value is null && v.StatusCode.Code == StatusCodes.BadBoundNotFound))
.GroupBy(v => v.SourceTimestamp.Ticks).Count(g => g.Count() >= Math.Max(2, cap - 1))
: 0;
// The read under test.
var source = new CappedFake(series, shape, cap);
fx.Nm.HistoricalSourceForTest(source, tieBudget);
var pages = new List<IReadOnlyList<DataValue>>();
var collected = new List<DataValue>();
byte[]? cp = null;
StatusCode status;
do
{
var (values, error, next) = Read(start, end, n, bounds, cp);
status = error.StatusCode;
if (!StatusCode.IsGood(status)) break;
pages.Add(values);
collected.AddRange(values);
cp = next;
pages.Count.ShouldBeLessThanOrEqualTo(truth.Count + 2 + oversizedClusters, $"paging must terminate: {tag} pages {Sizes(pages)}");
}
while (cp is not null);
source.OverCapRequests.ShouldBe(0, $"the server asked the backend for 0 or more than the cap: {tag}");
source.BadRequests.ShouldBe(0, $"a reverse window reached a forward-only backend: {tag}");
if (status.Code == StatusCodes.BadTooManyOperations)
{
// Legitimate only past the budget the cap leaves: a tie cluster larger than min(M, cap − 1), or on the
// Gateway shape a millisecond holding more than min(M, cap − 2) samples (the walk-back's budget).
var overfetchBound = cap > 0 ? Math.Min(Math.Max(1, tieBudget), Math.Max(1, cap - 1)) : Math.Max(1, tieBudget);
var walkBudget = cap > 0 ? Math.Max(1, Math.Min(tieBudget, cap - 2)) : Math.Max(1, tieBudget);
var legit = windowTies > overfetchBound || (walks && msMax > walkBudget);
legit.ShouldBeTrue($"refused with no tie cluster over {overfetchBound} (max {windowTies}; ms max {msMax} vs {walkBudget}): {tag}");
Render(collected).ShouldBe(Render(truth.Take(collected.Count)), $"the prefix before the refusal: {tag}");
Count("refused");
continue;
}
status.Code.ShouldBe(StatusCodes.Good, $"{tag}");
Render(collected).ShouldBe(Render(truth), $"paged ≠ unpaged: {tag} pages {Sizes(pages)}");
if (cap > 0)
{
// A bounded read whose data fits one page carries its start bound, the data and its end bound: the node
// manager fits the end bound against the client's N but not against the server cap (follow-up F2), so that
// one page may hold cap + 1 values; every other page is at most the cap.
var bothBoundsOnOnePage = bounds && pages.Count == 1;
pages.ShouldAllBe(p => p.Count <= cap + (bothBoundsOnOnePage ? 1 : 0), $"a page over the cap: {tag} pages {Sizes(pages)}");
if (bothBoundsOnOnePage && pages[0].Count == cap + 1) Count("bounded single page at cap+1 (F2)");
}
if (n > 0) pages.ShouldAllBe(p => p.Count <= n, $"a page over N: {tag} pages {Sizes(pages)}");
for (var i = 0; i < pages.Count - 1; i++)
{
if (pages[i].Count > 0) continue;
// Allowed only as the step past a boundary tie cluster the cap cannot page within.
var previousLast = i > 0 && pages[i - 1].Count > 0 ? pages[i - 1][^1].SourceTimestamp.Ticks : (long?)null;
var clusterAtBoundary = previousLast is { } t ? truth.Count(v => v.SourceTimestamp.Ticks == t) : 0;
(cap > 0 && clusterAtBoundary >= Math.Max(2, cap - 1)).ShouldBeTrue(
$"an empty page with a continuation point (page {i}, boundary cluster {clusterAtBoundary}): {tag} pages {Sizes(pages)}");
Count("empty step over an oversized tie cluster");
}
if (pages.Count > 1) Count("paged");
if (pages.Count > 1 && n == 0) Count("paged at N=0");
// A cost guard against a read explosion, not a budget. A native shape reads once per page plus the bound
// lookups. The Gateway shape's walk-back has its own cost profile — ~30 reads a page on a dense tag
// (#691/#693), far more on a sparse series under a tiny tie budget (seed 320: 82 reads with NO cap, seed 6428:
// 120+) — so it is judged RELATIVE to the same request with the cap off (the pre-#736 behaviour), per page:
// the cap must not make the walk explode compared to before. The baseline asks for the page size the cap
// imposes (N = cap − 1): an uncapped N = 0 reverse read used to be ONE int.MaxValue forward read — the very
// memory vector the cap removes — so "what N = 0 cost before" is not the comparison.
var allowedReads = 8 * pages.Count + 8;
if (shape == Shape.Gateway)
{
var baseline = new CappedFake(series, shape, cap: 0);
fx.Nm.HistoricalSourceForTest(baseline, tieBudget);
var baseN = cap > 0 && (n == 0 || n > (uint)cap) ? (uint)(cap - 1) : n;
var basePages = 0;
byte[]? baseCp = null;
do
{
var (_, baseError, next) = Read(start, end, baseN, bounds, baseCp);
if (!StatusCode.IsGood(baseError.StatusCode)) break;
basePages++;
baseCp = next;
}
while (baseCp is not null && basePages <= truth.Count + 2 + oversizedClusters);
allowedReads = 2 * baseline.Reads * Math.Max(1, pages.Count) / Math.Max(1, basePages) + 32 * pages.Count + 16;
}
source.Reads.ShouldBeLessThanOrEqualTo(allowedReads,
$"backend reads exploded: {source.Reads} for {pages.Count} pages (allowed {allowedReads}): {tag}");
Count("exact");
}
}
finally
{
fx.Nm.HistoricalSourceForTest(NullHistorianDataSource.Instance, 65536);
fx.Gate.Release();
}
// Non-vacuous: every category the brief names occurred, and most runs were exact.
var summary = string.Join(", ", counts.OrderBy(kv => kv.Key).Select(kv => $"{kv.Key}={kv.Value}"));
foreach (var required in new[] { "empty", "ties in window", "window ends on a tie", "reverse", "end-only", "N=1", "N=2", "N=0", "cap=3", "bounds", "paged", "paged at N=0", "SqlServer", "QuestDb", "Gateway" })
counts.GetValueOrDefault(required).ShouldBeGreaterThan(iterations / 200, $"{required} under-represented: {summary}");
counts.GetValueOrDefault("exact").ShouldBeGreaterThan(iterations * 6 / 10, summary);
TestContext.Current.TestOutputHelper?.WriteLine($"raw-cap fuzz: {iterations} iterations — {summary}");
}
// --- generators -----------------------------------------------------------------------------
private static DateTime Ts(DataValueSnapshot s) => s.SourceTimestampUtc ?? DateTime.MinValue;
private static DateTime Trunc(DateTime t, long resolution) => new(t.Ticks - t.Ticks % resolution, DateTimeKind.Utc);
/// <summary>Random ascending series: bursts, gaps, sub-millisecond stamps and tie clusters; stamps stored at the resolution.</summary>
private static List<DataValueSnapshot> RandomSeries(Random rng, long resolution)
{
var list = new List<DataValueSnapshot>();
var t = T0.Ticks - rng.Next(0, 3) * TimeSpan.TicksPerHour;
var count = rng.Next(1, 120);
var v = 0;
while (list.Count < count)
{
t += rng.Next(6) switch
{
0 => rng.Next(1, 10), // sub-millisecond (sub-µs on QuestDB: collapses onto a tie)
1 => rng.Next(1, 5) * TimeSpan.TicksPerMillisecond, // dense
2 => rng.Next(1, 3) * TimeSpan.TicksPerSecond,
3 => rng.Next(1, 30) * TimeSpan.TicksPerMinute, // a gap
_ => rng.Next(1, 1000) * TimeSpan.TicksPerMillisecond,
};
var ties = rng.Next(7) == 0 ? rng.Next(2, 9) : 1;
for (var k = 0; k < ties; k++)
{
var ts = Trunc(new DateTime(t, DateTimeKind.Utc), resolution);
var status = rng.Next(12) == 0 ? StatusCodes.Bad : StatusCodes.Good;
list.Add(new DataValueSnapshot(status == StatusCodes.Bad ? null : (double)v, status, ts, ts));
v++;
}
}
return list;
}
/// <summary>A window bound: before / after the data, exactly on a sample (a tie, maybe), one step past one, a tick before one, or between two.</summary>
private static DateTime PickBoundary(Random rng, List<DataValueSnapshot> series, long resolution)
{
var first = Ts(series[0]);
var last = Ts(series[^1]);
var i = rng.Next(series.Count);
var t = Ts(series[i]);
switch (rng.Next(9))
{
case 0: return first.AddSeconds(-rng.Next(1, 100));
case 1: return last.AddSeconds(rng.Next(1, 100));
case 2:
case 3: return t;
case 4: return t.AddTicks(resolution);
case 5: return t.AddTicks(-1); // sub-resolution: a microsecond backend truncates it back ONTO the sample
case 6: return t.AddMilliseconds(1);
default:
var next = i + 1 < series.Count ? Ts(series[i + 1]) : t.AddSeconds(10);
return new DateTime(t.Ticks + (long)((next.Ticks - t.Ticks) * rng.NextDouble()), DateTimeKind.Utc);
}
}
/// <summary>The series' own answer for a plain (no bounds) read: forward <c>[start, end)</c>, point, reverse / end-only <c>(end, start]</c> newest first.</summary>
private static List<DataValue> Oracle(List<DataValueSnapshot> series, DateTime start, DateTime end, Direction direction, long resolution)
{
var s = Trunc(start, resolution);
var e = Trunc(end, resolution);
IEnumerable<DataValueSnapshot> window = direction switch
{
// start == end is Part 11's point read whichever way it was generated: the node manager reads it forward.
Direction.Forward or Direction.Reverse when s == e => series.Where(x => Ts(x) == s),
Direction.Forward => series.Where(x => Ts(x) >= s && Ts(x) < e),
Direction.Reverse => series.Where(x => Ts(x) > e && Ts(x) <= s).Reverse(),
_ => series.Where(x => Ts(x) <= e).Reverse(),
};
return window.Select(x => new DataValue(new Variant(x.Value), new StatusCode(x.StatusCode), Ts(x), Ts(x))).ToList();
}
/// <summary>The largest tie cluster among the returned stored values (a bound placeholder is not a stored value).</summary>
private static int LargestTieCluster(IReadOnlyList<DataValue> truth) =>
truth.Where(v => !(v.Value is null && v.StatusCode.Code == StatusCodes.BadBoundNotFound))
.GroupBy(v => v.SourceTimestamp.Ticks).Select(g => g.Count()).DefaultIfEmpty(0).Max();
/// <summary>The most samples within one millisecond anywhere in the series (the walk-back's refusal rule).</summary>
private static int MillisecondMax(List<DataValueSnapshot> series)
{
var msMax = 0;
for (int i = 0, j = 0; i < series.Count; i++)
{
while (Ts(series[i]).Ticks - Ts(series[j]).Ticks >= TimeSpan.TicksPerMillisecond) j++;
msMax = Math.Max(msMax, i - j + 1);
}
return msMax;
}
private static bool EndsOnTie(List<DataValueSnapshot> series, DateTime start, DateTime end, Direction direction, long resolution)
{
var edge = Trunc(direction == Direction.Forward ? end : direction == Direction.Reverse ? start : end, resolution);
return series.Count(x => Ts(x) == edge) >= 2;
}
// --- helpers ----------------------------------------------------------------------------------
private static string Sizes(IEnumerable<IReadOnlyList<DataValue>> pages) => "[" + string.Join(",", pages.Select(p => p.Count)) + "]";
private static string Render(IEnumerable<DataValue> values) =>
string.Join(" | ", values.Select(v => $"{v.SourceTimestamp.Ticks}={v.Value ?? "null"}/{v.StatusCode.Code:X8}"));
private (IReadOnlyList<DataValue> Values, ServiceResult Error, byte[]? Cp) Read(DateTime start, DateTime end, uint max, bool bounds, byte[]? cp)
{
var context = new OperationContext(new RequestHeader(), secureChannelContext: null, RequestType.HistoryRead, identity: null);
var details = new ReadRawModifiedDetails
{
StartTime = start, EndTime = end, NumValuesPerNode = max, ReturnBounds = bounds, IsReadModified = false,
};
var nodesToRead = new List<HistoryReadValueId> { new() { NodeId = fx.NodeId, ContinuationPoint = cp } };
var results = new List<SdkHistoryReadResult> { null! };
var errors = new List<ServiceResult> { null! };
fx.Nm.HistoryRead(context, details, TimestampsToReturn.Both, releaseContinuationPoints: false, nodesToRead, results, errors);
var values = new List<DataValue>();
if (results[0]?.HistoryData is { } ho && !ExtensionObject.IsNull(ho))
values.AddRange(((HistoryData)ExtensionObject.ToEncodeable(ho)).DataValues);
return (values, errors[0], results[0]?.ContinuationPoint);
}
/// <summary>
/// A series-backed historian shaped like one backend, enforcing the backend half of the server raw cap exactly as
/// the real data sources do: it reports the cap as <see cref="MaxRawValuesPerRead"/>; a read asking for 0 or more than
/// the cap reads cap + 1 rows and throws <see cref="HistoryReadLimitExceededException"/> when the window holds more
/// (never a truncated answer). It also records every such request (<see cref="OverCapRequests"/>) — the server must
/// never issue one — and every reverse window sent to the forward-only shape (<see cref="BadRequests"/>).
/// </summary>
private sealed class CappedFake(IReadOnlyList<DataValueSnapshot> series, Shape shape, int cap) : IHistorianDataSource
{
private readonly Lock _gate = new();
public int Reads { get; private set; }
public int OverCapRequests { get; private set; }
public int BadRequests { get; private set; }
public bool SupportsReverseRaw => shape != Shape.Gateway;
public long TimestampResolutionTicks => shape == Shape.QuestDb ? 10 : 1;
public int MaxRawValuesPerRead => cap;
private DateTime Trunc(DateTime t) => NodeManagerHistoryReadRawCapFuzzTests.Trunc(t, TimestampResolutionTicks);
public Task<HistorianRead> ReadRawAsync(
string fullReference, DateTime startUtc, DateTime endUtc, uint maxValuesPerNode, CancellationToken cancellationToken)
{
var probe = cap > 0 && (maxValuesPerNode == 0 || maxValuesPerNode > (uint)cap);
lock (_gate)
{
Reads++;
if (probe) OverCapRequests++;
if (shape == Shape.Gateway && startUtc > endUtc) BadRequests++;
}
if (shape == Shape.Gateway && startUtc > endUtc)
throw new InvalidOperationException("start_time must be less than or equal to end_time. (InvalidArgument)");
startUtc = Trunc(startUtc);
endUtc = Trunc(endUtc);
List<DataValueSnapshot> window;
if (startUtc > endUtc) window = series.Where(s => Ts(s) > endUtc && Ts(s) <= startUtc).Reverse().ToList();
else if (startUtc == endUtc) window = series.Where(s => Ts(s) == startUtc).ToList();
else window = series.Where(s => Ts(s) >= startUtc && Ts(s) < endUtc).ToList();
var top = probe ? cap + 1 : maxValuesPerNode == 0 ? int.MaxValue : (int)Math.Min(maxValuesPerNode, int.MaxValue);
var page = window.Take(top).ToList();
if (probe && page.Count > cap)
throw new HistoryReadLimitExceededException(cap, $"more than MaxRawValuesPerRead ({cap}) samples");
return Task.FromResult(new HistorianRead(page, null));
}
public Task<HistorianRead> ReadProcessedAsync(
string fullReference, DateTime startUtc, DateTime endUtc, TimeSpan interval,
HistoryAggregateType aggregate, CancellationToken cancellationToken) =>
Task.FromResult(new HistorianRead(Array.Empty<DataValueSnapshot>(), null));
public Task<HistorianRead> ReadAtTimeAsync(
string fullReference, IReadOnlyList<DateTime> timestampsUtc, CancellationToken cancellationToken) =>
Task.FromResult(new HistorianRead(Array.Empty<DataValueSnapshot>(), null));
public Task<HistoricalEventsResult> ReadEventsAsync(
string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents, CancellationToken cancellationToken) =>
Task.FromResult(new HistoricalEventsResult(Array.Empty<HistoricalEvent>(), null));
public HistorianHealthSnapshot GetHealthSnapshot() => NullHistorianDataSource.Instance.GetHealthSnapshot();
public void Dispose() { }
}
}
/// <summary>Test-only wiring shared by the paging fuzz: swap the source and the tie budget, with a fresh continuation store.</summary>
internal static class NodeManagerFuzzExtensions
{
public static void HistoricalSourceForTest(this OtOpcUaNodeManager nm, IHistorianDataSource source, int tieBudget)
{
nm.HistoryContinuationStore = new InMemoryHistoryContinuationStore();
nm.MaxTieClusterOverfetch = tieBudget;
nm.HistorianDataSource = source;
}
}
@@ -24,6 +24,15 @@ public sealed class RawPageLimitTests
[InlineData(5000u, 983, 984, 0u, 0u, true)] // the ties alone fill a read: never a 0 (uncapped) request
[InlineData(5000u, 5000, 984, 0u, 0u, true)] // never an underflow
[InlineData(5000u, 0, 1, 0u, 0u, true)]
// R08:1.1 (#736): the server raw cap (default 10 000) is the limit every backend reports — an uncapped read pages at it.
[InlineData(0u, 0, 10_000, 9_999u, 10_000u, false)] // NumValuesPerNode = 0: a first page of cap − 1 + the lookahead row
[InlineData(0u, 1, 10_000, 9_998u, 10_000u, false)] // a resumed page re-reads its one boundary tie
[InlineData(100_000u, 0, 10_000, 9_999u, 10_000u, false)] // a client cap above the server cap is bounded by it
[InlineData(10_000u, 0, 10_000, 9_999u, 10_000u, false)] // exactly the cap: still cap − 1 (the lookahead row needs room)
[InlineData(500u, 0, 10_000, 500u, 501u, false)] // under the cap: unchanged
[InlineData(0u, 0, 3, 2u, 3u, false)] // the smallest usable cap (MinMaxRawValuesPerRead = 3): a first page of two
[InlineData(0u, 1, 3, 1u, 3u, false)] // … and a resumed page still carries one value beside its boundary tie
[InlineData(0u, 1, 2, 0u, 0u, true)] // at 2 a resumed page has NO room — every value would cost an empty page (why 3 is the floor)
public void Plan_fits_page_ties_and_lookahead_into_the_limit(uint want, int k, int limit, uint page, uint request, bool noRoom)
{
var plan = RawPageLimit.Plan(want, k, limit);
@@ -168,4 +168,68 @@ 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);
}
[Theory]
[InlineData(1)]
[InlineData(2)]
public void A_MaxRawValuesPerRead_below_three_is_raised_to_three_and_warns_when_enabled(int value)
{
// A resumed page's read must carry a value, the boundary tie it re-reads and the lookahead row that says "more
// follows": at 2 it has no room and every value costs an empty page with a continuation point (the #692 defect).
var o = new ServerHistorianOptions
{
Enabled = true, Endpoint = "https://h:5222", ApiKey = "histgw_x_y", MaxRawValuesPerRead = value,
};
Assert.Equal(3, o.EffectiveMaxRawValuesPerRead);
Assert.Equal(ServerHistorianOptions.MinMaxRawValuesPerRead, o.EffectiveMaxRawValuesPerRead);
var m = Assert.Single(o.Validate());
Assert.Contains("MaxRawValuesPerRead", m);
Assert.Equal(3, new ServerHistorianOptions { MaxRawValuesPerRead = 3 }.EffectiveMaxRawValuesPerRead);
}
[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());
}
}