Merge qdb/T7: QuestDb provider + Host wiring; conformance green on both backends

Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
This commit is contained in:
Joseph Doherty
2026-10-11 03:04:43 -04:00
29 changed files with 968 additions and 117 deletions
+6
View File
@@ -441,6 +441,9 @@ services:
# docker compose -f docker-dev/docker-compose.yml up -d central-1 central-2 # docker compose -f docker-dev/docker-compose.yml up -d central-1 central-2
ServerHistorian__Backend: "${OTOPCUA_HISTORIAN_BACKEND:-Gateway}" ServerHistorian__Backend: "${OTOPCUA_HISTORIAN_BACKEND:-Gateway}"
ServerHistorian__SqlServer__ConnectionString: "${OTOPCUA_HISTORIAN_SQL:-Server=sql,1433;Database=OtOpcUaHistorian;User Id=sa;Password=OtOpcUa!Dev123;TrustServerCertificate=True;Encrypt=False}" ServerHistorian__SqlServer__ConnectionString: "${OTOPCUA_HISTORIAN_SQL:-Server=sql,1433;Database=OtOpcUaHistorian;User Id=sa;Password=OtOpcUa!Dev123;TrustServerCertificate=True;Encrypt=False}"
# QuestDb = the QuestDB historian; the `questdb` service comes from the docker-dev/docker-compose.questdb.yml overlay
# (add `-f docker-dev/docker-compose.questdb.yml` and OTOPCUA_HISTORIAN_BACKEND=QuestDb).
ServerHistorian__QuestDb__ConnectionString: "${OTOPCUA_HISTORIAN_QUESTDB:-Host=questdb;Port=8812;Username=admin;Password=quest;Database=qdb;ServerCompatibilityMode=NoTypeLoading}"
# Continuous historization of driver tag values (Gitea #491's live gate). Gated on BOTH # Continuous historization of driver tag values (Gitea #491's live gate). Gated on BOTH
# ContinuousHistorization:Enabled AND ServerHistorian:Enabled — with either off no recorder is # ContinuousHistorization:Enabled AND ServerHistorian:Enabled — with either off no recorder is
# spawned, so this stays inert on the default rig. OutboxPath is the FasterLog crash boundary and # spawned, so this stays inert on the default rig. OutboxPath is the FasterLog crash boundary and
@@ -574,6 +577,9 @@ services:
# docker compose -f docker-dev/docker-compose.yml up -d central-1 central-2 # docker compose -f docker-dev/docker-compose.yml up -d central-1 central-2
ServerHistorian__Backend: "${OTOPCUA_HISTORIAN_BACKEND:-Gateway}" ServerHistorian__Backend: "${OTOPCUA_HISTORIAN_BACKEND:-Gateway}"
ServerHistorian__SqlServer__ConnectionString: "${OTOPCUA_HISTORIAN_SQL:-Server=sql,1433;Database=OtOpcUaHistorian;User Id=sa;Password=OtOpcUa!Dev123;TrustServerCertificate=True;Encrypt=False}" ServerHistorian__SqlServer__ConnectionString: "${OTOPCUA_HISTORIAN_SQL:-Server=sql,1433;Database=OtOpcUaHistorian;User Id=sa;Password=OtOpcUa!Dev123;TrustServerCertificate=True;Encrypt=False}"
# QuestDb = the QuestDB historian; the `questdb` service comes from the docker-dev/docker-compose.questdb.yml overlay
# (add `-f docker-dev/docker-compose.questdb.yml` and OTOPCUA_HISTORIAN_BACKEND=QuestDb).
ServerHistorian__QuestDb__ConnectionString: "${OTOPCUA_HISTORIAN_QUESTDB:-Host=questdb;Port=8812;Username=admin;Password=quest;Database=qdb;ServerCompatibilityMode=NoTypeLoading}"
# Continuous historization — see central-1 for the full note + the gate recipe. Both centrals # Continuous historization — see central-1 for the full note + the gate recipe. Both centrals
# carry it so the pair is symmetric; :9200 round-robins them, and the recorder must exist on # carry it so the pair is symmetric; :9200 round-robins them, and the recorder must exist on
# whichever node holds the drivers. # whichever node holds the drivers.
@@ -107,4 +107,19 @@ public interface IHistorianDataSource : IDisposable
/// </summary> /// </summary>
/// <returns>The current health snapshot for the historian backend.</returns> /// <returns>The current health snapshot for the historian backend.</returns>
HistorianHealthSnapshot GetHealthSnapshot(); HistorianHealthSnapshot GetHealthSnapshot();
/// <summary>
/// The backend's storage timestamp resolution, in <see cref="DateTime"/> ticks (100 ns): the smallest step
/// between two distinct stored timestamps. <c>1</c> (the default) means full tick precision; QuestDB stores
/// microseconds and reports <c>10</c>.
/// </summary>
/// <remarks>
/// The server's HistoryRead-Raw paging steps OFF a drained timestamp by this amount (forward
/// <c>T + resolution</c>, reverse <c>T − resolution</c>). Stepping by less than the resolution is not
/// progress on a backend that truncates its inputs: <c>T + 1 tick</c> is truncated back to <c>T</c>, the
/// same sample comes back, and a one-value-per-page forward read loops forever. Stepping by the resolution is
/// still lossless, because no stored timestamp can lie strictly between <c>T</c> and <c>T ± resolution</c>.
/// Same-timestamp ties are unaffected (they are paged within the timestamp, never stepped over).
/// </remarks>
long TimestampResolutionTicks => 1;
} }
@@ -19,8 +19,10 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb;
/// <para> /// <para>
/// <b>Timestamps (R6).</b> Every inbound time is normalised with <see cref="QuestDbConnection.Round"/> and every /// <b>Timestamps (R6).</b> Every inbound time is normalised with <see cref="QuestDbConnection.Round"/> and every
/// <see cref="DateTime"/> is bound only through <see cref="QuestDbConnection.BindTimestamp"/>. Read bounds are not /// <see cref="DateTime"/> is bound only through <see cref="QuestDbConnection.BindTimestamp"/>. Read bounds are not
/// clamped to the epoch (a pre-1970 bound is valid); only <c>MinValue</c>/<c>MaxValue</c> are nudged. Results /// clamped to the epoch (a pre-1970 bound is valid); only <c>MinValue</c>/<c>MaxValue</c> are nudged. Raw and
/// carry the rounded times, so an AtTime result is stamped at the rounded request time. /// Processed results carry stored (microsecond) times. An AtTime result is evaluated at the rounded request time
/// but stamped with the client's ORIGINAL requested time (as the SQL Server backend does), so a client can match
/// each answer to its request.
/// </para> /// </para>
/// <para> /// <para>
/// <b>Reads never create tags (R7).</b> A tag resolves only through <see cref="QuestDbTagCatalog.ExistsAsync"/>; /// <b>Reads never create tags (R7).</b> A tag resolves only through <see cref="QuestDbTagCatalog.ExistsAsync"/>;
@@ -44,6 +46,10 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb;
/// </remarks> /// </remarks>
public sealed class QuestDbHistorianDataSource : IHistorianDataSource public sealed class QuestDbHistorianDataSource : IHistorianDataSource
{ {
/// <inheritdoc />
/// <remarks>QuestDB stores microseconds: 10 ticks. HistoryRead paging steps off a drained timestamp by this much.</remarks>
public long TimestampResolutionTicks => 10;
/// <summary>Most AtTime request times per statement (two <c>UNION ALL</c> branches each; QuestDB 10.0.1 refuses more than 126 branches).</summary> /// <summary>Most AtTime request times per statement (two <c>UNION ALL</c> branches each; QuestDB 10.0.1 refuses more than 126 branches).</summary>
public const int AtTimeChunkSize = 50; public const int AtTimeChunkSize = 50;
@@ -139,8 +145,9 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource
/// <c>LIMIT 1</c> seeks, <c>UNION ALL</c>-ed, at most <see cref="AtTimeChunkSize"/> times per statement (spec §3, /// <c>LIMIT 1</c> seeks, <c>UNION ALL</c>-ed, at most <see cref="AtTimeChunkSize"/> times per statement (spec §3,
/// T0 Q9). <see cref="AtTimeEvaluator.Evaluate"/> then decides each value exactly as for SQL Server (exact hit, /// T0 Q9). <see cref="AtTimeEvaluator.Evaluate"/> then decides each value exactly as for SQL Server (exact hit,
/// no previous ⇒ <c>BadNoData</c>, inside an outage, held before an outage, linear, stepped, after now). The /// no previous ⇒ <c>BadNoData</c>, inside an outage, held before an outage, linear, stepped, after now). The
/// result matches the request in length and order (duplicates included), each entry stamped at its rounded /// result matches the request in length and order (duplicates included). Each entry is evaluated at its rounded
/// time. Unknown tag ⇒ every entry <c>BadNoData</c>. /// time but stamped with the requested time itself (UTC-normalised as <see cref="QuestDbConnection.Round"/> does,
/// not truncated), matching the SQL Server backend. Unknown tag ⇒ every entry <c>BadNoData</c>.
/// </remarks> /// </remarks>
public async Task<HistoryReadResult> ReadAtTimeAsync( public async Task<HistoryReadResult> ReadAtTimeAsync(
string fullReference, IReadOnlyList<DateTime> timestampsUtc, CancellationToken cancellationToken) string fullReference, IReadOnlyList<DateTime> timestampsUtc, CancellationToken cancellationToken)
@@ -197,10 +204,11 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource
var result = new DataValueSnapshot[rounded.Length]; var result = new DataValueSnapshot[rounded.Length];
for (var i = 0; i < result.Length; i++) for (var i = 0; i < result.Length; i++)
{ {
var t = rounded[i]; // Seek/evaluate at the rounded time; stamp the client's own requested time (M3).
result[i] = byTicks.TryGetValue(t.Ticks, out var v) var requested = AsUtc(timestampsUtc[i]);
? v result[i] = byTicks.TryGetValue(rounded[i].Ticks, out var v)
: new DataValueSnapshot(null, HistorianStatusCodes.BadNoData, t, now); ? v with { SourceTimestampUtc = requested }
: new DataValueSnapshot(null, HistorianStatusCodes.BadNoData, requested, now);
} }
RecordSuccess(); RecordSuccess();
@@ -213,6 +221,14 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource
} }
} }
/// <summary>The UTC normalisation of <see cref="QuestDbConnection.Round"/> without its truncation or nudging.</summary>
private static DateTime AsUtc(DateTime t) => t.Kind switch
{
DateTimeKind.Local => t.ToUniversalTime(),
DateTimeKind.Utc => t,
_ => DateTime.SpecifyKind(t, DateTimeKind.Utc),
};
/// <summary> /// <summary>
/// The T0 Q9 statement for <paramref name="count"/> request times: <c>$1</c> = tag, <c>$(k+2)</c> = time <c>k</c>. /// The T0 Q9 statement for <paramref name="count"/> request times: <c>$1</c> = tag, <c>$(k+2)</c> = time <c>k</c>.
/// Columns: <c>req, dir</c> (0 = previous-or-equal, 1 = next), then <c>ts, kind, num_value, int_value, str_value, /// Columns: <c>req, dir</c> (0 = previous-or-equal, 1 = next), then <c>ts, kind, num_value, int_value, str_value,
@@ -31,9 +31,6 @@ internal sealed class QuestDbHistorianDataUpdater : IHistorianDataUpdater
/// <summary>Rows per INSERT and per DeleteRaw page (T0 Q2: 9,000 parameters, far under the 32,767 limit).</summary> /// <summary>Rows per INSERT and per DeleteRaw page (T0 Q2: 9,000 parameters, far under the 32,767 limit).</summary>
internal const int BatchSize = 1000; internal const int BatchSize = 1000;
/// <summary>Request times further apart than this are read with separate range queries (I-2).</summary>
internal static readonly TimeSpan ExistenceClusterGap = TimeSpan.FromMinutes(1);
private const long ResolutionTicks = 10; private const long ResolutionTicks = 10;
private const int LockStripes = 64; private const int LockStripes = 64;
private readonly SemaphoreSlim[] _locks = Enumerable.Range(0, LockStripes).Select(_ => new SemaphoreSlim(1, 1)).ToArray(); private readonly SemaphoreSlim[] _locks = Enumerable.Range(0, LockStripes).Select(_ => new SemaphoreSlim(1, 1)).ToArray();
@@ -60,26 +57,6 @@ internal sealed class QuestDbHistorianDataUpdater : IHistorianDataUpdater
/// <summary>Rows returned by existence reads so far (test/diagnostic counter).</summary> /// <summary>Rows returned by existence reads so far (test/diagnostic counter).</summary>
internal long LiveRowsRead => Interlocked.Read(ref _liveRowsRead); internal long LiveRowsRead => Interlocked.Read(ref _liveRowsRead);
/// <summary>Splits times into clusters wherever consecutive sorted times are further apart than <paramref name="gap"/>.</summary>
/// <param name="times">Request times (any order).</param>
/// <param name="gap">Largest gap kept inside one cluster.</param>
/// <returns>Inclusive [Lo, Hi] ranges, ascending.</returns>
internal static List<(DateTime Lo, DateTime Hi)> Cluster(IReadOnlyList<DateTime> times, TimeSpan gap)
{
var result = new List<(DateTime, DateTime)>();
if (times.Count == 0) return result;
var sorted = times.OrderBy(t => t).ToList();
var lo = sorted[0];
var prev = lo;
foreach (var t in sorted.Skip(1))
{
if (t - prev > gap) { result.Add((lo, prev)); lo = t; }
prev = t;
}
result.Add((lo, prev));
return result;
}
private async Task<IDisposable> LockAsync(string tag, CancellationToken ct) private async Task<IDisposable> LockAsync(string tag, CancellationToken ct)
{ {
var sem = _locks[(int)((uint)StringComparer.Ordinal.GetHashCode(tag) % LockStripes)]; var sem = _locks[(int)((uint)StringComparer.Ordinal.GetHashCode(tag) % LockStripes)];
@@ -231,7 +208,7 @@ internal sealed class QuestDbHistorianDataUpdater : IHistorianDataUpdater
{ {
var live = new HashSet<long>(); var live = new HashSet<long>();
await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false); await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false);
foreach (var (lo, hi) in Cluster(times, ExistenceClusterGap)) foreach (var (lo, hi) in QuestDbTimeClusters.Cluster(times, QuestDbTimeClusters.ExistenceGap))
{ {
await using var cmd = _conn.CreateCommand(conn, await using var cmd = _conn.CreateCommand(conn,
$"SELECT ts FROM {_conn.SampleTable} WHERE tag = $1 AND ts >= $2 AND ts <= $3 AND NOT deleted"); $"SELECT ts FROM {_conn.SampleTable} WHERE tag = $1 AND ts >= $2 AND ts <= $3 AND NOT deleted");
@@ -0,0 +1,106 @@
using System.Collections.Frozen;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian;
using ZB.MOM.WW.OtOpcUa.Runtime.Historian;
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb;
/// <summary>
/// <see cref="IHistorianBackendProvider"/> for the QuestDB historian. Supports every
/// <see cref="HistorianValueKind"/> and HistoryUpdate (soft delete). All seams share ONE connection helper, ONE
/// schema bootstrapper, ONE tag catalog (R7: the only "known tag" authority) and ONE WAL visibility wait, so the
/// schema is ensured once and known tags are cached once.
/// </summary>
/// <remarks>
/// The shared state is built on the first <c>Create*</c> call, not in the constructor: the constructor does no
/// validation and no I/O, so resolving the provider for a backend nobody consumes never fails on its options
/// (they are validated, consumer-gated, by the host's <c>ServerHistorianOptionsValidator</c>). The first
/// <c>Create*</c> call's service provider supplies the logger factory the shared pieces log through.
/// </remarks>
public sealed class QuestDbHistorianProvider : IHistorianBackendProvider
{
private static readonly FrozenSet<HistorianValueKind> Kinds =
Enum.GetValues<HistorianValueKind>().ToFrozenSet();
private readonly QuestDbHistorianOptions _options;
private readonly object _initLock = new();
private Shared? _shared;
/// <summary>Creates the provider over the <c>ServerHistorian:QuestDb</c> options (validated elsewhere).</summary>
/// <param name="options">The <c>ServerHistorian:QuestDb</c> options.</param>
public QuestDbHistorianProvider(QuestDbHistorianOptions options)
{
_options = options ?? throw new ArgumentNullException(nameof(options));
}
/// <inheritdoc />
public string Name => "QuestDb";
/// <inheritdoc />
public IReadOnlySet<HistorianValueKind> SupportedKinds => Kinds;
private static ILoggerFactory Loggers(IServiceProvider sp) =>
(sp.GetService(typeof(ILoggerFactory)) as ILoggerFactory) ?? NullLoggerFactory.Instance;
private Shared Get(IServiceProvider sp)
{
if (Volatile.Read(ref _shared) is { } built) return built;
lock (_initLock)
{
if (_shared is { } raced) return raced;
var loggers = Loggers(sp);
var conn = new QuestDbConnection(_options);
var bootstrap = new QuestDbSchemaBootstrapper(conn, _options, loggers.CreateLogger<QuestDbSchemaBootstrapper>());
var wal = new QuestDbWalVisibility(conn, _options.VisibilityTimeout, loggers.CreateLogger<QuestDbWalVisibility>());
var shared = new Shared(conn, bootstrap, new QuestDbTagCatalog(conn, bootstrap, wal), wal);
Volatile.Write(ref _shared, shared);
return shared;
}
}
/// <inheritdoc />
public IHistorianDataSource CreateDataSource(IServiceProvider sp)
{
var s = Get(sp);
return new QuestDbHistorianDataSource(s.Conn, s.Bootstrap, s.Catalog,
Loggers(sp).CreateLogger<QuestDbHistorianDataSource>());
}
/// <inheritdoc />
public IHistorianValueWriter CreateValueWriter(IServiceProvider sp)
{
var s = Get(sp);
return new QuestDbHistorianValueWriter(s.Conn, s.Catalog, s.Wal,
Loggers(sp).CreateLogger<QuestDbHistorianValueWriter>());
}
/// <inheritdoc />
public IAlarmHistorianWriter CreateAlarmWriter(IServiceProvider sp)
{
var s = Get(sp);
return new QuestDbAlarmHistorianWriter(s.Conn, s.Bootstrap, s.Wal,
Loggers(sp).CreateLogger<QuestDbAlarmHistorianWriter>());
}
/// <inheritdoc />
public IHistorianProvisioning CreateProvisioning(IServiceProvider sp)
{
var s = Get(sp);
return new QuestDbTagProvisioner(s.Conn, s.Bootstrap, s.Catalog, s.Wal,
Loggers(sp).CreateLogger<QuestDbTagProvisioner>());
}
/// <inheritdoc />
/// <remarks>The soft-delete updater: deletes write tombstones (INSERT only, never QuestDB UPDATE/DELETE).</remarks>
public IHistorianDataUpdater? CreateUpdater(IServiceProvider sp)
{
var s = Get(sp);
return new QuestDbHistorianDataUpdater(s.Conn, s.Catalog, s.Wal,
Loggers(sp).CreateLogger<QuestDbHistorianDataUpdater>());
}
private sealed record Shared(
QuestDbConnection Conn, QuestDbSchemaBootstrapper Bootstrap, QuestDbTagCatalog Catalog, QuestDbWalVisibility Wal);
}
@@ -98,49 +98,13 @@ internal sealed class QuestDbHistorianValueWriter : IHistorianValueWriter
} }
} }
/// <summary>A gap between consecutive batch times larger than this starts a new existence-query range.</summary>
internal static readonly TimeSpan ExistenceClusterGap = TimeSpan.FromMinutes(1);
/// <summary>
/// Splits ascending <paramref name="sortedTimes"/> into inclusive <c>[Lo, Hi]</c> ranges: a new range starts where the gap to the
/// previous time exceeds <paramref name="gap"/> or the range already holds <paramref name="maxPerCluster"/> times. One outlier
/// time therefore costs its own tiny query instead of widening one range over every stored row in between.
/// </summary>
/// <param name="sortedTimes">Times in ascending order.</param>
/// <param name="gap">Largest gap kept inside one range.</param>
/// <param name="maxPerCluster">Most times in one range (&gt;= 1).</param>
/// <returns>The ranges, ascending.</returns>
internal static IReadOnlyList<(DateTime Lo, DateTime Hi)> Cluster(IReadOnlyList<DateTime> sortedTimes, TimeSpan gap, int maxPerCluster)
{
ArgumentOutOfRangeException.ThrowIfLessThan(maxPerCluster, 1);
var result = new List<(DateTime, DateTime)>();
if (sortedTimes.Count == 0) return result;
var lo = sortedTimes[0];
var prev = lo;
var n = 1;
for (var i = 1; i < sortedTimes.Count; i++)
{
var t = sortedTimes[i];
if (t - prev > gap || n >= maxPerCluster)
{
result.Add((lo, prev));
lo = t;
n = 0;
}
prev = t;
n++;
}
result.Add((lo, prev));
return result;
}
// T0 Q11 range form, one query per cluster: tombstones are NOT filtered out (a deleted row must not be resurrected). // T0 Q11 range form, one query per cluster: tombstones are NOT filtered out (a deleted row must not be resurrected).
private async Task<HashSet<long>> ExistingTicksAsync(string tag, List<HistorianSampleRow> rows, CancellationToken ct) private async Task<HashSet<long>> ExistingTicksAsync(string tag, List<HistorianSampleRow> rows, CancellationToken ct)
{ {
var times = rows.Select(r => QuestDbConnection.Round(r.TimeUtc)).OrderBy(t => t).ToList(); var times = rows.Select(r => QuestDbConnection.Round(r.TimeUtc)).OrderBy(t => t).ToList();
var set = new HashSet<long>(); var set = new HashSet<long>();
await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false); await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false);
foreach (var (lo, hi) in Cluster(times, ExistenceClusterGap, BatchRows)) foreach (var (lo, hi) in QuestDbTimeClusters.Cluster(times, QuestDbTimeClusters.ExistenceGap, BatchRows))
{ {
await using var cmd = _conn.CreateCommand(conn, $"SELECT ts FROM {_conn.SampleTable} WHERE tag = $1 AND ts >= $2 AND ts <= $3"); await using var cmd = _conn.CreateCommand(conn, $"SELECT ts FROM {_conn.SampleTable} WHERE tag = $1 AND ts >= $2 AND ts <= $3");
QuestDbConnection.BindText(cmd, tag); QuestDbConnection.BindText(cmd, tag);
@@ -0,0 +1,49 @@
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb;
/// <summary>
/// Groups the times of one write or HistoryUpdate request into inclusive <c>[Lo, Hi]</c> ranges so that the
/// "which of these times already exist" check reads one range query per cluster instead of one query spanning
/// <c>[min, max]</c> of the whole request. One far-away outlier time then costs its own tiny query instead of
/// widening a single range over every stored row in between. Shared by the live-value writer and the updater.
/// </summary>
internal static class QuestDbTimeClusters
{
/// <summary>Request times further apart than this start a new existence-query range.</summary>
internal static readonly TimeSpan ExistenceGap = TimeSpan.FromMinutes(1);
/// <summary>
/// Splits <paramref name="times"/> (any order; sorted here) into inclusive ranges, ascending: a new range starts
/// where the gap to the previous time exceeds <paramref name="gap"/> (exactly <paramref name="gap"/> stays
/// together), or where the current range already holds <paramref name="maxPerCluster"/> times.
/// </summary>
/// <param name="times">Request times, in any order.</param>
/// <param name="gap">Largest gap kept inside one range.</param>
/// <param name="maxPerCluster">Most times in one range (&gt;= 1); unbounded by default.</param>
/// <returns>The ranges, ascending; empty for no times.</returns>
internal static IReadOnlyList<(DateTime Lo, DateTime Hi)> Cluster(
IReadOnlyList<DateTime> times, TimeSpan gap, int maxPerCluster = int.MaxValue)
{
ArgumentNullException.ThrowIfNull(times);
ArgumentOutOfRangeException.ThrowIfLessThan(maxPerCluster, 1);
var result = new List<(DateTime, DateTime)>();
if (times.Count == 0) return result;
var sorted = times.Order().ToList();
var lo = sorted[0];
var prev = lo;
var n = 1;
for (var i = 1; i < sorted.Count; i++)
{
var t = sorted[i];
if (t - prev > gap || n >= maxPerCluster)
{
result.Add((lo, prev));
lo = t;
n = 0;
}
prev = t;
n++;
}
result.Add((lo, prev));
return result;
}
}
@@ -1,6 +1,7 @@
using Microsoft.Extensions.Configuration; using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.Options; using Microsoft.Extensions.Options;
using ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway; using ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway;
using ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb;
using ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer; using ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer;
using ZB.MOM.WW.OtOpcUa.Runtime.Historian; using ZB.MOM.WW.OtOpcUa.Runtime.Historian;
@@ -25,11 +26,23 @@ public static class HistorianBackendRegistry
return o; return o;
} }
/// <summary>Backend name for the QuestDB historian.</summary>
public const string QuestDbBackendName = "QuestDb";
/// <summary>Binds the <c>ServerHistorian:QuestDb</c> section (shared by the registry and the validator).</summary>
public static QuestDbHistorianOptions BindQuestDb(IConfiguration configuration)
{
var o = new QuestDbHistorianOptions();
configuration.GetSection($"{ServerHistorianOptions.SectionName}:QuestDb").Bind(o);
return o;
}
private static readonly IReadOnlyDictionary<string, Func<ServerHistorianOptions, IConfiguration, IHistorianBackendProvider>> Factories = private static readonly IReadOnlyDictionary<string, Func<ServerHistorianOptions, IConfiguration, IHistorianBackendProvider>> Factories =
new Dictionary<string, Func<ServerHistorianOptions, IConfiguration, IHistorianBackendProvider>>(StringComparer.OrdinalIgnoreCase) new Dictionary<string, Func<ServerHistorianOptions, IConfiguration, IHistorianBackendProvider>>(StringComparer.OrdinalIgnoreCase)
{ {
[ServerHistorianOptions.GatewayBackendName] = (o, _) => new GatewayHistorianProvider(o), [ServerHistorianOptions.GatewayBackendName] = (o, _) => new GatewayHistorianProvider(o),
[SqlServerBackendName] = (_, c) => new SqlServerHistorianProvider(BindSqlServer(c)), [SqlServerBackendName] = (_, c) => new SqlServerHistorianProvider(BindSqlServer(c)),
[QuestDbBackendName] = (_, c) => new QuestDbHistorianProvider(BindQuestDb(c)),
}; };
/// <summary>The registered backend names, for validation messages.</summary> /// <summary>The registered backend names, for validation messages.</summary>
@@ -78,6 +78,13 @@ public sealed class ServerHistorianOptionsValidator : OptionsValidatorBase<Serve
return; return;
} }
if (string.Equals(options.Backend?.Trim(), HistorianBackendRegistry.QuestDbBackendName, StringComparison.OrdinalIgnoreCase))
{
foreach (var error in HistorianBackendRegistry.BindQuestDb(_configuration).Validate())
builder.RequireThat(false, error);
return;
}
if (!options.IsGatewayBackend) return; if (!options.IsGatewayBackend) return;
var endpointValid = Uri.TryCreate(options.Endpoint, UriKind.Absolute, out var uri) var endpointValid = Uri.TryCreate(options.Endpoint, UriKind.Absolute, out var uri)
@@ -81,6 +81,7 @@
<ProjectReference Include="..\..\Drivers\ZB.MOM.WW.OtOpcUa.Driver.Galaxy\ZB.MOM.WW.OtOpcUa.Driver.Galaxy.csproj"/> <ProjectReference Include="..\..\Drivers\ZB.MOM.WW.OtOpcUa.Driver.Galaxy\ZB.MOM.WW.OtOpcUa.Driver.Galaxy.csproj"/>
<ProjectReference Include="..\..\Drivers\ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway\ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.csproj"/> <ProjectReference Include="..\..\Drivers\ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway\ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.csproj"/>
<ProjectReference Include="..\..\Drivers\ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer\ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.csproj"/> <ProjectReference Include="..\..\Drivers\ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer\ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.csproj"/>
<ProjectReference Include="..\..\Drivers\ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb\ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.csproj"/>
<ProjectReference Include="..\..\Drivers\ZB.MOM.WW.OtOpcUa.Driver.Modbus\ZB.MOM.WW.OtOpcUa.Driver.Modbus.csproj"/> <ProjectReference Include="..\..\Drivers\ZB.MOM.WW.OtOpcUa.Driver.Modbus\ZB.MOM.WW.OtOpcUa.Driver.Modbus.csproj"/>
<ProjectReference Include="..\..\Drivers\ZB.MOM.WW.OtOpcUa.Driver.Mqtt\ZB.MOM.WW.OtOpcUa.Driver.Mqtt.csproj"/> <ProjectReference Include="..\..\Drivers\ZB.MOM.WW.OtOpcUa.Driver.Mqtt\ZB.MOM.WW.OtOpcUa.Driver.Mqtt.csproj"/>
<ProjectReference Include="..\..\Drivers\ZB.MOM.WW.OtOpcUa.Driver.MTConnect\ZB.MOM.WW.OtOpcUa.Driver.MTConnect.csproj"/> <ProjectReference Include="..\..\Drivers\ZB.MOM.WW.OtOpcUa.Driver.MTConnect\ZB.MOM.WW.OtOpcUa.Driver.MTConnect.csproj"/>
@@ -212,25 +212,34 @@ internal static class HistoryPaging
/// after <paramref name="endUtc"/> (the reverse window excludes <c>end</c>; a resume at exactly <c>end</c> /// after <paramref name="endUtc"/> (the reverse window excludes <c>end</c>; a resume at exactly <c>end</c>
/// would turn into a <c>start == end</c> point read and return the excluded sample). Forward is unchanged. /// would turn into a <c>start == end</c> point read and return the excluded sample). Forward is unchanged.
/// </param> /// </param>
/// <param name="resolutionTicks">
/// The backend's storage timestamp resolution (<c>IHistorianDataSource.TimestampResolutionTicks</c>, &gt;= 1). A
/// drained cluster steps off <paramref name="boundaryT"/> by this many ticks rather than by one: a backend that
/// truncates its inputs to its resolution (QuestDB, 10 ticks) would otherwise truncate <c>T + 1 tick</c> back
/// to <c>T</c> and return the drained sample again, forever. Lossless for the same reason one tick is at tick
/// precision: no stored timestamp lies strictly between <c>T</c> and <c>T ± resolution</c>. The default
/// <c>1</c> keeps the tick-precision behaviour exactly.
/// </param>
public static void SliceTieCluster( public static void SliceTieCluster(
int clusterCount, int skip, uint cap, DateTime boundaryT, DateTime endUtc, int clusterCount, int skip, uint cap, DateTime boundaryT, DateTime endUtc,
out int sliceStart, out int sliceCount, out DateTime? nextStartUtc, out int nextSkip, out int sliceStart, out int sliceCount, out DateTime? nextStartUtc, out int nextSkip,
bool reverse = false) bool reverse = false, long resolutionTicks = 1)
{ {
ArgumentOutOfRangeException.ThrowIfLessThan(resolutionTicks, 1L);
sliceStart = Math.Min(skip, clusterCount); sliceStart = Math.Min(skip, clusterCount);
sliceCount = Math.Min((int)cap, clusterCount - sliceStart); sliceCount = Math.Min((int)cap, clusterCount - sliceStart);
var emitted = sliceStart + sliceCount; var emitted = sliceStart + sliceCount;
if (emitted < clusterCount) { nextStartUtc = boundaryT; nextSkip = emitted; } if (emitted < clusterCount) { nextStartUtc = boundaryT; nextSkip = emitted; }
else if (reverse) else if (reverse)
{ {
// DateTime.MinValue cannot step down; nothing can lie below it anyway. // Nothing can lie below DateTime.MinValue, so a boundary within one resolution step of it cannot step down.
var more = boundaryT > DateTime.MinValue && boundaryT.AddTicks(-1) > endUtc; var more = boundaryT.Ticks >= resolutionTicks && boundaryT.AddTicks(-resolutionTicks) > endUtc;
nextStartUtc = more ? boundaryT.AddTicks(-1) : null; nextStartUtc = more ? boundaryT.AddTicks(-resolutionTicks) : null;
nextSkip = 0; nextSkip = 0;
} }
else else
{ {
var next = boundaryT.AddTicks(1); var next = boundaryT.AddTicks(resolutionTicks);
if (next <= endUtc) { nextStartUtc = next; nextSkip = 0; } if (next <= endUtc) { nextStartUtc = next; nextSkip = 0; }
else { nextStartUtc = null; nextSkip = 0; } else { nextStartUtc = null; nextSkip = 0; }
} }
@@ -5931,7 +5931,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
HistoryPaging.SliceTieCluster( HistoryPaging.SliceTieCluster(
cluster.Count, boundarySkip, numValuesPerNode, startUtc, endUtc, cluster.Count, boundarySkip, numValuesPerNode, startUtc, endUtc,
out var sliceStart, out var sliceCount, out var nextStartUtc, out var nextSkip, out var sliceStart, out var sliceCount, out var nextStartUtc, out var nextSkip,
reverse: reverse); reverse: reverse, resolutionTicks: Math.Max(1L, HistorianDataSource.TimestampResolutionTicks));
var slice = new List<DataValueSnapshot>(sliceCount); var slice = new List<DataValueSnapshot>(sliceCount);
for (var i = sliceStart; i < sliceStart + sliceCount; i++) slice.Add(cluster[i]); for (var i = sliceStart; i < sliceStart + sliceCount; i++) slice.Add(cluster[i]);
@@ -35,7 +35,7 @@ public abstract class AtTimeConformance : ConformanceBase
var s = (await B.CreateDataSource().ReadAtTimeAsync("A", [At(10), At(20)], Ct)).Samples; var s = (await B.CreateDataSource().ReadAtTimeAsync("A", [At(10), At(20)], Ct)).Samples;
s[0].Value.ShouldBe(20d); s[0].Value.ShouldBe(20d);
s[0].StatusCode.ShouldBe(0u); s[0].StatusCode.ShouldBe(0u);
s[0].SourceTimestampUtc.ShouldBe(B.Round(At(10))); s[0].SourceTimestampUtc.ShouldBe(At(10)); // the requested time itself, never rounded (M3)
s[0].ServerTimestampUtc.ShouldBe(B.Round(At(11))); // the stored ServerTimeUtc s[0].ServerTimestampUtc.ShouldBe(B.Round(At(11))); // the stored ServerTimeUtc
s[1].Value.ShouldBe(30d); // a stored Bad sample comes back as stored s[1].Value.ShouldBe(30d); // a stored Bad sample comes back as stored
s[1].StatusCode.ShouldBe(0x80000000u); s[1].StatusCode.ShouldBe(0x80000000u);
@@ -50,7 +50,7 @@ public abstract class AtTimeConformance : ConformanceBase
s[1].Value.ShouldBe(12.5d); // 10 + 10 * 2.5/10 s[1].Value.ShouldBe(12.5d); // 10 + 10 * 2.5/10
s[2].Value.ShouldBe(42.5d); // 40 + 10 * 2.5/10 s[2].Value.ShouldBe(42.5d); // 40 + 10 * 2.5/10
s.ShouldAllBe(x => x.StatusCode == GoodInterpolated); s.ShouldAllBe(x => x.StatusCode == GoodInterpolated);
s[0].SourceTimestampUtc.ShouldBe(B.Round(At(5))); // stamped at the requested (sub-ms) time s[0].SourceTimestampUtc.ShouldBe(At(5)); // stamped at the requested (sub-µs) time itself, never rounded (M3)
} }
[Fact] [Fact]
@@ -61,7 +61,7 @@ public abstract class AtTimeConformance : ConformanceBase
var s = (await B.CreateDataSource().ReadAtTimeAsync("A", [before], Ct)).Samples.Single(); var s = (await B.CreateDataSource().ReadAtTimeAsync("A", [before], Ct)).Samples.Single();
s.Value.ShouldBeNull(); s.Value.ShouldBeNull();
s.StatusCode.ShouldBe(BadNoData); s.StatusCode.ShouldBe(BadNoData);
s.SourceTimestampUtc.ShouldBe(B.Round(before)); s.SourceTimestampUtc.ShouldBe(before);
} }
[Fact] [Fact]
@@ -80,7 +80,7 @@ public abstract class AtTimeConformance : ConformanceBase
var s = (await B.CreateDataSource().ReadAtTimeAsync("A", [At(25)], Ct)).Samples.Single(); var s = (await B.CreateDataSource().ReadAtTimeAsync("A", [At(25)], Ct)).Samples.Single();
s.Value.ShouldBeNull(); // NOT 35 (30→40) and NOT 30 (a Bad value) s.Value.ShouldBeNull(); // NOT 35 (30→40) and NOT 30 (a Bad value)
s.StatusCode.ShouldBe(0x80000000u); // the previous sample's Bad code — never upgraded to Uncertain s.StatusCode.ShouldBe(0x80000000u); // the previous sample's Bad code — never upgraded to Uncertain
s.SourceTimestampUtc.ShouldBe(B.Round(At(25))); s.SourceTimestampUtc.ShouldBe(At(25));
} }
[Fact] [Fact]
@@ -173,7 +173,7 @@ public abstract class AtTimeConformance : ConformanceBase
var s = (await B.CreateDataSource().ReadAtTimeAsync("A", [At(5), At(-1), At(5), At(10)], Ct)).Samples; var s = (await B.CreateDataSource().ReadAtTimeAsync("A", [At(5), At(-1), At(5), At(10)], Ct)).Samples;
s.Count.ShouldBe(4); s.Count.ShouldBe(4);
s.Select(x => x.Value).ShouldBe([15d, null, 15d, 20d]); s.Select(x => x.Value).ShouldBe([15d, null, 15d, 20d]);
s.Select(x => x.SourceTimestampUtc).ShouldBe([B.Round(At(5)), B.Round(At(-1)), B.Round(At(5)), B.Round(At(10))]); s.Select(x => x.SourceTimestampUtc).ShouldBe([At(5), At(-1), At(5), At(10)]);
} }
[Fact] [Fact]
@@ -0,0 +1,85 @@
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Driver.Historian.Conformance;
using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common;
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests;
/// <summary>
/// QuestDB AtTime regressions found in the T5 review (M1, M3), and the data source's reported timestamp
/// resolution (the HistoryRead paging step, C1).
/// </summary>
[Collection(QuestDbCollection.Name)]
public sealed class AtTimeRegressionTests(QuestDbFixture fx) : IAsyncLifetime
{
private static readonly CancellationToken Ct = CancellationToken.None;
private const uint GoodInterpolated = 0x00000402u;
private QuestDbConformanceBackend _b = null!;
private static DateTime At(double s) => ConformanceData.At(s);
public async ValueTask InitializeAsync() => _b = await QuestDbConformanceBackend.CreateConcreteAsync(fx);
public async ValueTask DisposeAsync()
{
if (_b is not null) await _b.DisposeAsync();
}
/// <summary>
/// M3: a request time with a sub-microsecond remainder is answered stamped with THAT time (as SQL Server does),
/// not with the rounded time the seek used. The value is still the one at the rounded time.
/// </summary>
[Fact]
public async Task AtTime_result_is_stamped_with_the_original_sub_microsecond_request_time()
{
await ConformanceData.SeedAAsync(_b); // +0=10, +10=20, ...
var requested = _b.Round(At(5)).AddTicks(3); // 0.3 µs past a microsecond (T0 itself is not µs-aligned)
var unspecified = DateTime.SpecifyKind(_b.Round(At(7)).AddTicks(7), DateTimeKind.Unspecified);
var s = (await _b.CreateDataSource().ReadAtTimeAsync("A", [requested, unspecified], Ct)).Samples;
s[0].SourceTimestampUtc.ShouldBe(requested);
s[0].SourceTimestampUtc!.Value.Kind.ShouldBe(DateTimeKind.Utc);
s[0].Value.ShouldBe(15d); // evaluated at the rounded time Round(At(5))
s[0].StatusCode.ShouldBe(GoodInterpolated);
s[1].SourceTimestampUtc.ShouldBe(DateTime.SpecifyKind(unspecified, DateTimeKind.Utc));
s[1].SourceTimestampUtc!.Value.Ticks.ShouldBe(unspecified.Ticks);
}
/// <summary>M3 on the unknown-tag path: every BadNoData entry carries its own requested time.</summary>
[Fact]
public async Task AtTime_unknown_tag_entries_are_stamped_with_the_original_request_time()
{
var requested = _b.Round(At(5)).AddTicks(3);
var s = (await _b.CreateDataSource().ReadAtTimeAsync("nope", [requested], Ct)).Samples;
s.Single().StatusCode.ShouldBe(HistorianStatusCodes.BadNoData);
s.Single().SourceTimestampUtc.ShouldBe(requested);
}
/// <summary>
/// M1: the NEXT seek skips a tombstone and finds the next live row. With +10 deleted, At(5) lies between the
/// live +0 (10) and the live +20 (30), so it interpolates to 15, Good. Seeing the tombstone as "next" would
/// instead hold 10 as Uncertain (an Empty BadNoData neighbour is not usable).
/// </summary>
[Fact]
public async Task AtTime_next_seek_skips_a_tombstone_and_finds_the_next_live_row()
{
await _b.SeedAsync("M",
(At(0), HistorianValue.FromDouble(10), ConformanceData.Good),
(At(10), HistorianValue.FromDouble(20), ConformanceData.Good),
(At(20), HistorianValue.FromDouble(30), ConformanceData.Good));
await _b.InsertRowsAsync("M", [(At(10), HistorianValue.Empty, HistorianStatusCodes.BadNoData, true)]);
var s = (await _b.CreateDataSource().ReadAtTimeAsync("M", [At(5)], Ct)).Samples.Single();
s.Value.ShouldBe(15d); // 10 + (30 - 10) * 5/20
s.StatusCode.ShouldBe(GoodInterpolated);
}
/// <summary>C1: the data source reports QuestDB's microsecond storage resolution to the paging code.</summary>
[Fact]
public void Data_source_reports_a_ten_tick_timestamp_resolution() =>
((IHistorianDataSource)_b.CreateDataSource()).TimestampResolutionTicks.ShouldBe(10);
}
@@ -0,0 +1,118 @@
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Driver.Historian.Conformance;
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests;
/// <summary>
/// C1 (T7 review I-1): the facts the server's HistoryRead-Raw paging relies on, pinned against the REAL QuestDB data
/// source (the node-manager walk in <c>NodeManagerHistoryReadPagingTests</c> runs on a fake that models them).
/// <c>HistoryPaging</c> is internal to OpcUaServer and not visible from this project, so the end-to-end walk below
/// re-states the node manager's cursor arithmetic (resume inclusive + boundary trim; stalled ⇒ point-read the
/// cluster and step off by <see cref="IHistorianDataSource.TimestampResolutionTicks"/>) over the real reads.
/// </summary>
[Collection(QuestDbCollection.Name)]
public sealed class PagingRegressionTests(QuestDbFixture fx) : IAsyncLifetime
{
private static readonly CancellationToken Ct = CancellationToken.None;
private static readonly DateTime T0 = new(2026, 3, 1, 12, 0, 0, DateTimeKind.Utc); // µs-aligned
private QuestDbConformanceBackend _b = null!;
/// <summary>Sample i at T0 + i µs (adjacent microseconds: a one-tick step lands strictly between two of them).</summary>
private static DateTime S(int i) => T0.AddTicks(10L * i);
public async ValueTask InitializeAsync()
{
_b = await QuestDbConformanceBackend.CreateConcreteAsync(fx);
await _b.SeedAsync("P", Enumerable.Range(0, 6)
.Select(i => (S(i), HistorianValue.FromDouble(i), ConformanceData.Good)).ToArray());
}
public async ValueTask DisposeAsync()
{
if (_b is not null) await _b.DisposeAsync();
}
private async Task<DateTime?[]> RawAsync(IHistorianDataSource ds, DateTime start, DateTime end, uint max) =>
(await ds.ReadRawAsync("P", start, end, max, Ct)).Samples.Select(s => s.SourceTimestampUtc).ToArray();
[Fact]
public async Task Forward_raw_reads_truncate_one_tick_but_advance_by_one_resolution_step()
{
var ds = _b.CreateDataSource();
var end = T0.AddHours(1);
var res = ds.TimestampResolutionTicks;
(await RawAsync(ds, S(2).AddTicks(1), end, 1)).ShouldBe([S(2)]); // T + 1 tick ⇒ back to T (the C1 loop)
(await RawAsync(ds, S(2).AddTicks(res), end, 1)).ShouldBe([S(3)]); // T + resolution ⇒ the next sample
(await RawAsync(ds, S(2), S(2), 10)).ShouldBe([S(2)]); // point read at T ⇒ T
}
[Fact]
public async Task Reverse_raw_reads_step_down_by_one_resolution_step()
{
var ds = _b.CreateDataSource();
var end = T0.AddHours(-1);
var res = ds.TimestampResolutionTicks;
(await RawAsync(ds, S(3).AddTicks(-1), end, 1)).ShouldBe([S(2)]); // T − 1 tick truncates DOWN ⇒ previous
(await RawAsync(ds, S(3).AddTicks(-res), end, 1)).ShouldBe([S(2)]); // T − resolution ⇒ previous
(await RawAsync(ds, S(3), end, 1)).ShouldBe([S(3)]); // reverse window (end, start] includes start
}
[Theory]
[InlineData(false)]
[InlineData(true)]
public async Task One_value_per_page_walk_returns_every_sample_once_and_terminates(bool reverse)
{
var ds = _b.CreateDataSource();
var (start, end) = reverse ? (T0.AddHours(1), T0.AddHours(-1)) : (T0.AddHours(-1), T0.AddHours(1));
var collected = await WalkAsync(ds, start, end);
var expected = Enumerable.Range(0, 6).Select(i => (DateTime?)S(i));
collected.ShouldBe(reverse ? expected.Reverse() : expected);
}
/// <summary>The node manager's Raw paging loop at <c>NumValuesPerNode = 1</c> (see <c>ServeRawPagedAsync</c>).</summary>
private static async Task<List<DateTime?>> WalkAsync(IHistorianDataSource ds, DateTime start, DateTime end)
{
const uint cap = 1;
var res = ds.TimestampResolutionTicks;
var reverse = start > end;
var collected = new List<DateTime?>();
DateTime? cursor = start;
var skip = 0;
var resumed = false;
for (var pages = 0; cursor is { } from; pages++)
{
pages.ShouldBeLessThan(60, "paging must terminate, not loop");
var raw = (await ds.ReadRawAsync("P", from, end, cap, Ct)).Samples;
var full = raw.Count >= cap;
var page = raw.ToList();
if (resumed)
for (var dropped = 0; dropped < skip && page.Count > 0 && page[0].SourceTimestampUtc == from; dropped++)
page.RemoveAt(0);
if (resumed && full && (page.Count == 0 || page[^1].SourceTimestampUtc == from))
{
// Stalled on the boundary "cluster": point-read it, emit what is left, step off by the resolution.
var cluster = (await ds.ReadRawAsync("P", from, from, 1000, Ct)).Samples;
var sliceStart = Math.Min(skip, cluster.Count);
var slice = cluster.Skip(sliceStart).Take((int)cap).ToList();
collected.AddRange(slice.Select(s => s.SourceTimestampUtc));
if (sliceStart + slice.Count < cluster.Count) { skip = sliceStart + slice.Count; continue; }
var next = from.AddTicks(reverse ? -res : res);
cursor = reverse ? (next > end ? next : null) : (next <= end ? next : null);
skip = 0;
continue;
}
collected.AddRange(page.Select(s => s.SourceTimestampUtc));
if (!full || page.Count == 0) break;
cursor = page[^1].SourceTimestampUtc;
skip = page.Count(s => s.SourceTimestampUtc == cursor);
resumed = true;
}
return collected;
}
}
@@ -0,0 +1,55 @@
using ZB.MOM.WW.OtOpcUa.Driver.Historian.Conformance;
using Xunit;
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests;
// One concrete subclass per conformance class: the suite's tests run here against QuestDB.
[Collection(QuestDbCollection.Name)]
public sealed class QuestDbRawReadConformance(QuestDbFixture fx) : RawReadConformance
{
/// <inheritdoc/>
protected override Task<IConformanceBackend> CreateBackendAsync() => QuestDbConformanceBackend.CreateAsync(fx);
}
[Collection(QuestDbCollection.Name)]
public sealed class QuestDbAtTimeConformance(QuestDbFixture fx) : AtTimeConformance
{
/// <inheritdoc/>
protected override Task<IConformanceBackend> CreateBackendAsync() => QuestDbConformanceBackend.CreateAsync(fx);
}
[Collection(QuestDbCollection.Name)]
public sealed class QuestDbProcessedConformance(QuestDbFixture fx) : ProcessedConformance
{
/// <inheritdoc/>
protected override Task<IConformanceBackend> CreateBackendAsync() => QuestDbConformanceBackend.CreateAsync(fx);
}
[Collection(QuestDbCollection.Name)]
public sealed class QuestDbQualityContractConformance(QuestDbFixture fx) : QualityContractConformance
{
/// <inheritdoc/>
protected override Task<IConformanceBackend> CreateBackendAsync() => QuestDbConformanceBackend.CreateAsync(fx);
}
[Collection(QuestDbCollection.Name)]
public sealed class QuestDbLiveWriteConformance(QuestDbFixture fx) : LiveWriteConformance
{
/// <inheritdoc/>
protected override Task<IConformanceBackend> CreateBackendAsync() => QuestDbConformanceBackend.CreateAsync(fx);
}
[Collection(QuestDbCollection.Name)]
public sealed class QuestDbUpdateConformance(QuestDbFixture fx) : UpdateConformance
{
/// <inheritdoc/>
protected override Task<IConformanceBackend> CreateBackendAsync() => QuestDbConformanceBackend.CreateAsync(fx);
}
[Collection(QuestDbCollection.Name)]
public sealed class QuestDbTagNameConformance(QuestDbFixture fx) : TagNameConformance
{
/// <inheritdoc/>
protected override Task<IConformanceBackend> CreateBackendAsync() => QuestDbConformanceBackend.CreateAsync(fx);
}
@@ -76,14 +76,13 @@ internal sealed class QuestDbConformanceBackend : IConformanceBackend
new QuestDbHistorianDataSource(Conn, Boot, NewCatalog(), timeProvider: clock); new QuestDbHistorianDataSource(Conn, Boot, NewCatalog(), timeProvider: clock);
/// <inheritdoc/> /// <inheritdoc/>
/// <remarks>TODO(T7): build the T4 <c>QuestDbHistorianValueWriter</c> here once T4 is merged.</remarks> /// <remarks>The T4 live-value writer over a fresh catalog (fresh positive cache) and this instance's WAL wait.</remarks>
public IHistorianValueWriter CreateValueWriter(Func<DateTime>? utcNow = null) => public IHistorianValueWriter CreateValueWriter(Func<DateTime>? utcNow = null) =>
throw new NotSupportedException("TODO(T7): the QuestDB value writer (T4) is not merged into this branch yet."); new QuestDbHistorianValueWriter(Conn, NewCatalog(), Wal, utcNow: utcNow);
/// <inheritdoc/> /// <inheritdoc/>
/// <remarks>TODO(T7): build the T6 <c>QuestDbHistorianDataUpdater</c> here once T6 is merged.</remarks> /// <remarks>The T6 soft-delete updater over a fresh catalog and this instance's WAL wait.</remarks>
public IHistorianDataUpdater CreateUpdater() => public IHistorianDataUpdater CreateUpdater() => new QuestDbHistorianDataUpdater(Conn, NewCatalog(), Wal);
throw new NotSupportedException("TODO(T7): the QuestDB history updater (T6) is not merged into this branch yet.");
/// <inheritdoc/> /// <inheritdoc/>
/// <remarks> /// <remarks>
@@ -43,9 +43,19 @@ public sealed class QuestDbFixture : IAsyncLifetime
/// <param name="retentionDays">Optional TTL.</param> /// <param name="retentionDays">Optional TTL.</param>
/// <param name="autoCreate">AutoCreateSchema.</param> /// <param name="autoCreate">AutoCreateSchema.</param>
/// <returns>Valid options.</returns> /// <returns>Valid options.</returns>
/// <remarks>
/// Pooled connections idle for more than a second are closed (<c>Connection Idle Lifetime</c> /
/// <c>Connection Pruning Interval</c>). Many tests build their own <see cref="QuestDbConnection"/> (one Npgsql
/// data source each) and never dispose it; with Npgsql's default 300 s idle lifetime their pooled connections
/// piled up until QuestDB's PG-wire limit (64) was reached and later tests' opens timed out after 15 s.
/// </remarks>
public QuestDbHistorianOptions Options(string prefix, int? retentionDays = null, bool autoCreate = true) => new() public QuestDbHistorianOptions Options(string prefix, int? retentionDays = null, bool autoCreate = true) => new()
{ {
ConnectionString = ConnectionString, ConnectionString = new NpgsqlConnectionStringBuilder(ConnectionString)
{
ConnectionIdleLifetime = 1,
ConnectionPruningInterval = 1,
}.ConnectionString,
TablePrefix = prefix, TablePrefix = prefix,
RetentionDays = retentionDays, RetentionDays = retentionDays,
AutoCreateSchema = autoCreate, AutoCreateSchema = autoCreate,
@@ -184,7 +184,7 @@ public sealed class ReadTests(QuestDbFixture fx) : IAsyncLifetime
times.Add(times[0]); times.Add(times[0]);
var s = (await _b.CreateDataSource().ReadAtTimeAsync("A", times, Ct)).Samples; var s = (await _b.CreateDataSource().ReadAtTimeAsync("A", times, Ct)).Samples;
s.Count.ShouldBe(times.Count); s.Count.ShouldBe(times.Count);
s.Select(x => x.SourceTimestampUtc).ShouldBe(times.Select(t => (DateTime?)_b.Round(t))); s.Select(x => x.SourceTimestampUtc).ShouldBe(times.Select(t => (DateTime?)t)); // the requested times (M3)
s[0].Value.ShouldBe(49d); // +39: 40 + 10 × 9/10 s[0].Value.ShouldBe(49d); // +39: 40 + 10 × 9/10
s[0].StatusCode.ShouldBe(GoodInterpolated); s[0].StatusCode.ShouldBe(GoodInterpolated);
s[^1].Value.ShouldBe(49d); // the duplicate s[^1].Value.ShouldBe(49d); // the duplicate
@@ -0,0 +1,38 @@
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common;
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests;
public sealed partial class TombstoneTests
{
/// <summary>
/// The live writer's outbox redelivers batches (at-least-once). A batch replayed after an operator's
/// DeleteAtTime must not bring the deleted sample back: the writer's existence check counts a tombstone as
/// present (it deliberately does not filter <c>NOT deleted</c>), so the redelivered row is skipped.
/// </summary>
[Fact]
public async Task Tombstone_is_not_resurrected_by_a_writer_redelivery()
{
fx.SkipIfUnavailable();
var rig = UpdateTests.MakeRig(fx);
var writer = new QuestDbHistorianValueWriter(rig.Conn, rig.Tags,
new QuestDbWalVisibility(rig.Conn, TimeSpan.FromSeconds(5)));
HistorizationValue[] batch =
[
new(T0, HistorianValue.FromDouble(1), HistorianStatusCodes.Good),
new(T0.AddSeconds(1), HistorianValue.FromDouble(2), HistorianStatusCodes.Good),
];
(await writer.WriteLiveValuesAsync("r", batch, default)).ShouldBeTrue();
(await rig.Updater.DeleteAtTimeAsync("r", [T0], default)).ShouldBe([HistorianStatusCodes.Good]);
(await writer.WriteLiveValuesAsync("r", batch, default)).ShouldBeTrue(); // the redelivery
(await UpdateTests.RowsAsync(rig, "r", liveOnly: true)).Select(r => r.Num).ShouldBe([2d]);
var all = await UpdateTests.RowsAsync(rig, "r");
all.Count.ShouldBe(2);
all[0].Deleted.ShouldBeTrue();
all[0].Num.ShouldBeNull();
}
}
@@ -7,7 +7,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests;
/// <summary>Soft-delete behaviour. The writer-redelivery case lives in T7 (<c>TombstoneTests.Redelivery.cs</c>).</summary> /// <summary>Soft-delete behaviour. The writer-redelivery case lives in T7 (<c>TombstoneTests.Redelivery.cs</c>).</summary>
[Collection(QuestDbCollection.Name)] [Collection(QuestDbCollection.Name)]
public sealed class TombstoneTests(QuestDbFixture fx) public sealed partial class TombstoneTests(QuestDbFixture fx)
{ {
private static readonly DateTime T0 = new(2026, 3, 1, 12, 0, 0, DateTimeKind.Utc); private static readonly DateTime T0 = new(2026, 3, 1, 12, 0, 0, DateTimeKind.Utc);
@@ -249,16 +249,6 @@ public sealed class UpdateTests(QuestDbFixture fx)
} }
} }
[Fact]
public void Cluster_splits_where_the_gap_exceeds_the_limit()
{
var gap = QuestDbHistorianDataUpdater.ExistenceClusterGap;
var c = QuestDbHistorianDataUpdater.Cluster(
[T0.AddHours(2), T0, T0.AddSeconds(30), T0.AddHours(1)], gap);
c.ShouldBe([(T0, T0.AddSeconds(30)), (T0.AddHours(1), T0.AddHours(1)), (T0.AddHours(2), T0.AddHours(2))]);
QuestDbHistorianDataUpdater.Cluster([], gap).ShouldBeEmpty();
}
[Fact] [Fact]
public async Task Existence_reads_are_bounded_when_request_times_are_far_apart() public async Task Existence_reads_are_bounded_when_request_times_are_far_apart()
{ {
@@ -274,23 +274,4 @@ public sealed class WriteTests(QuestDbFixture fx)
(await w.WriteLiveValuesAsync("tag.far", vals, CancellationToken.None)).ShouldBeTrue(); (await w.WriteLiveValuesAsync("tag.far", vals, CancellationToken.None)).ShouldBeTrue();
(await RowsAsync(c, "tag.far")).Count.ShouldBe(2); (await RowsAsync(c, "tag.far")).Count.ShouldBe(2);
} }
[Fact]
public void Cluster_splits_on_a_gap_over_the_limit_and_at_the_per_cluster_cap()
{
var g = TimeSpan.FromMinutes(1);
QuestDbHistorianValueWriter.Cluster([], g, 10).ShouldBeEmpty();
QuestDbHistorianValueWriter.Cluster([T0], g, 10).ShouldBe([(T0, T0)]);
// exactly the gap stays together; one tick over splits
QuestDbHistorianValueWriter.Cluster([T0, T0 + g], g, 10).ShouldBe([(T0, T0 + g)]);
QuestDbHistorianValueWriter.Cluster([T0, T0 + g + TimeSpan.FromTicks(10)], g, 10)
.ShouldBe([(T0, T0), (T0 + g + TimeSpan.FromTicks(10), T0 + g + TimeSpan.FromTicks(10))]);
// outlier gets its own range
var far = T0.AddYears(1);
QuestDbHistorianValueWriter.Cluster([T0, T0.AddSeconds(1), far], g, 10).ShouldBe([(T0, T0.AddSeconds(1)), (far, far)]);
// cap: 5 close times, max 2 per cluster
var five = Enumerable.Range(0, 5).Select(i => T0.AddSeconds(i)).ToList();
QuestDbHistorianValueWriter.Cluster(five, g, 2)
.ShouldBe([(five[0], five[1]), (five[2], five[3]), (five[4], five[4])]);
}
} }
@@ -0,0 +1,90 @@
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests;
public sealed class QuestDbHistorianProviderTests
{
private static QuestDbHistorianProvider Make() => new(new QuestDbHistorianOptions
{
ConnectionString = "Host=localhost;Port=18812;Username=admin;Password=quest;Database=qdb;ServerCompatibilityMode=NoTypeLoading",
});
[Fact]
public void Name_and_all_six_kinds()
{
var p = Make();
p.Name.ShouldBe("QuestDb");
p.SupportedKinds.ShouldBe(Enum.GetValues<HistorianValueKind>(), ignoreOrder: true);
p.SupportedKinds.Count.ShouldBe(6);
}
[Fact]
public void Creates_all_five_seams_without_touching_questdb()
{
var sp = new NullSp();
var p = Make();
p.CreateDataSource(sp).ShouldBeOfType<QuestDbHistorianDataSource>();
p.CreateValueWriter(sp).ShouldBeOfType<QuestDbHistorianValueWriter>();
p.CreateAlarmWriter(sp).ShouldBeOfType<QuestDbAlarmHistorianWriter>();
p.CreateProvisioning(sp).ShouldBeOfType<QuestDbTagProvisioner>();
p.CreateUpdater(sp).ShouldBeOfType<QuestDbHistorianDataUpdater>();
}
[Fact]
public void Supported_kinds_are_a_frozen_set() =>
Make().SupportedKinds.ShouldBeAssignableTo<System.Collections.Frozen.FrozenSet<HistorianValueKind>>();
/// <summary>
/// An invalid table prefix must not fail provider construction (the host resolves the provider even when no
/// consumer is enabled); it surfaces only when a seam is actually built.
/// </summary>
[Fact]
public void An_invalid_prefix_fails_only_when_a_seam_is_built()
{
var p = new QuestDbHistorianProvider(new QuestDbHistorianOptions { ConnectionString = "Host=x", TablePrefix = "bad;prefix" });
p.Name.ShouldBe("QuestDb");
Should.Throw<ArgumentException>(() => p.CreateDataSource(new NullSp()));
}
[Fact]
public void Every_seam_shares_one_connection_bootstrapper_catalog_and_wal_wait()
{
var sp = new NullSp();
var p = Make();
var a = (QuestDbHistorianDataSource)p.CreateDataSource(sp);
var b = (QuestDbHistorianDataSource)p.CreateDataSource(sp);
var w = p.CreateValueWriter(sp);
var u = p.CreateUpdater(sp)!;
var prov = p.CreateProvisioning(sp);
var alarm = p.CreateAlarmWriter(sp);
Field(a, "_conn").ShouldBeSameAs(Field(b, "_conn"));
Field(a, "_conn").ShouldBeSameAs(Field(w, "_conn"));
Field(a, "_conn").ShouldBeSameAs(Field(u, "_conn"));
Field(a, "_conn").ShouldBeSameAs(Field(alarm, "_conn"));
Field(a, "_bootstrap").ShouldBeSameAs(Field(b, "_bootstrap"));
Field(a, "_bootstrap").ShouldBeSameAs(Field(prov, "_bootstrap"));
Field(a, "_bootstrap").ShouldBeSameAs(Field(alarm, "_bootstrap"));
Field(a, "_catalog").ShouldBeSameAs(Field(b, "_catalog"));
Field(a, "_catalog").ShouldBeSameAs(Field(w, "_catalog"));
Field(a, "_catalog").ShouldBeSameAs(Field(u, "_tags"));
Field(a, "_catalog").ShouldBeSameAs(Field(prov, "_catalog"));
Field(w, "_wal").ShouldBeSameAs(Field(u, "_wal"));
Field(w, "_wal").ShouldBeSameAs(Field(prov, "_wal"));
Field(w, "_wal").ShouldBeSameAs(Field(alarm, "_wal"));
}
[Fact]
public void The_constructor_rejects_null_options() =>
Should.Throw<ArgumentNullException>(() => new QuestDbHistorianProvider(null!));
private static object? Field(object o, string name) =>
o.GetType().GetField(name, System.Reflection.BindingFlags.Instance | System.Reflection.BindingFlags.NonPublic)!.GetValue(o);
private sealed class NullSp : IServiceProvider
{
public object? GetService(Type serviceType) => null;
}
}
@@ -0,0 +1,62 @@
using Shouldly;
using Xunit;
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests;
/// <summary>
/// The shared existence-query clustering (moved here, names unchanged, from the writer's <c>WriteTests</c> and the
/// updater's <c>UpdateTests</c> when the two private copies were consolidated into <see cref="QuestDbTimeClusters"/>).
/// </summary>
public sealed class QuestDbTimeClustersTests
{
private static readonly DateTime T0 = new(2026, 3, 1, 12, 0, 0, DateTimeKind.Utc);
[Fact]
public void Cluster_splits_on_a_gap_over_the_limit_and_at_the_per_cluster_cap()
{
var g = TimeSpan.FromMinutes(1);
QuestDbTimeClusters.Cluster([], g, 10).ShouldBeEmpty();
QuestDbTimeClusters.Cluster([T0], g, 10).ShouldBe([(T0, T0)]);
// exactly the gap stays together; one tick over splits
QuestDbTimeClusters.Cluster([T0, T0 + g], g, 10).ShouldBe([(T0, T0 + g)]);
QuestDbTimeClusters.Cluster([T0, T0 + g + TimeSpan.FromTicks(10)], g, 10)
.ShouldBe([(T0, T0), (T0 + g + TimeSpan.FromTicks(10), T0 + g + TimeSpan.FromTicks(10))]);
// outlier gets its own range
var far = T0.AddYears(1);
QuestDbTimeClusters.Cluster([T0, T0.AddSeconds(1), far], g, 10).ShouldBe([(T0, T0.AddSeconds(1)), (far, far)]);
// cap: 5 close times, max 2 per cluster
var five = Enumerable.Range(0, 5).Select(i => T0.AddSeconds(i)).ToList();
QuestDbTimeClusters.Cluster(five, g, 2)
.ShouldBe([(five[0], five[1]), (five[2], five[3]), (five[4], five[4])]);
}
[Fact]
public void Cluster_splits_where_the_gap_exceeds_the_limit()
{
var gap = QuestDbTimeClusters.ExistenceGap;
var c = QuestDbTimeClusters.Cluster(
[T0.AddHours(2), T0, T0.AddSeconds(30), T0.AddHours(1)], gap);
c.ShouldBe([(T0, T0.AddSeconds(30)), (T0.AddHours(1), T0.AddHours(1)), (T0.AddHours(2), T0.AddHours(2))]);
QuestDbTimeClusters.Cluster([], gap).ShouldBeEmpty();
}
[Fact]
public void The_existence_gap_is_one_minute_and_the_default_cap_is_unbounded()
{
QuestDbTimeClusters.ExistenceGap.ShouldBe(TimeSpan.FromMinutes(1));
var many = Enumerable.Range(0, 5000).Select(i => T0.AddMilliseconds(i)).ToList();
QuestDbTimeClusters.Cluster(many, QuestDbTimeClusters.ExistenceGap).ShouldBe([(many[0], many[^1])]);
}
[Fact]
public void Unsorted_input_is_sorted_before_capping()
{
var times = new[] { T0.AddSeconds(3), T0, T0.AddSeconds(2), T0.AddSeconds(1) };
QuestDbTimeClusters.Cluster(times, TimeSpan.FromMinutes(1), 2)
.ShouldBe([(T0, T0.AddSeconds(1)), (T0.AddSeconds(2), T0.AddSeconds(3))]);
}
[Fact]
public void A_cap_below_one_is_rejected() =>
Should.Throw<ArgumentOutOfRangeException>(() => QuestDbTimeClusters.Cluster([T0], TimeSpan.FromMinutes(1), 0));
}
@@ -375,4 +375,89 @@ public sealed class ServerHistorianOptionsValidatorTests
result.Failed.ShouldBeTrue(); result.Failed.ShouldBeTrue();
result.Failures.ShouldContain(f => f.Contains("ServerHistorian:SqlServer:Schema")); result.Failures.ShouldContain(f => f.Contains("ServerHistorian:SqlServer:Schema"));
} }
private static ServerHistorianOptionsValidator QuestDbSut(string? connectionString, string? prefix = null, bool alarmHistorianEnabled = false)
{
var values = new Dictionary<string, string?> { ["AlarmHistorian:Enabled"] = alarmHistorianEnabled ? "true" : "false" };
if (connectionString is not null) values["ServerHistorian:QuestDb:ConnectionString"] = connectionString;
if (prefix is not null) values["ServerHistorian:QuestDb:TablePrefix"] = prefix;
return new ServerHistorianOptionsValidator(new ConfigurationBuilder().AddInMemoryCollection(values).Build());
}
/// <summary>QuestDb with a connection string passes; the gateway Endpoint/ApiKey/UseTls checks do not fire.</summary>
[Fact]
public void QuestDb_backend_with_connection_string_passes_without_gateway_settings()
{
var result = QuestDbSut("Host=questdb;Port=8812").Validate(null,
new ServerHistorianOptions { Backend = "QuestDb", Enabled = true });
result.Failed.ShouldBeFalse(string.Join("; ", result.Failures ?? []));
}
/// <summary>QuestDb with no connection string fails, naming the key, and never asks for gateway settings.</summary>
[Fact]
public void QuestDb_backend_without_connection_string_fails_naming_the_key()
{
var result = QuestDbSut(null).Validate(null, new ServerHistorianOptions { Backend = "questdb", Enabled = true });
result.Failed.ShouldBeTrue();
result.Failures.ShouldContain(f => f.Contains("ServerHistorian:QuestDb:ConnectionString"));
result.Failures.ShouldNotContain(f => f.Contains("ServerHistorian:Endpoint"));
result.Failures.ShouldNotContain(f => f.Contains("ServerHistorian:ApiKey"));
}
/// <summary>Every <c>QuestDbHistorianOptions.Validate()</c> error is reported, each naming its key.</summary>
[Fact]
public void QuestDb_backend_reports_every_options_error_naming_the_key()
{
var configuration = new ConfigurationBuilder().AddInMemoryCollection(new Dictionary<string, string?>
{
["AlarmHistorian:Enabled"] = "false",
["ServerHistorian:QuestDb:TablePrefix"] = "bad;prefix",
["ServerHistorian:QuestDb:RetentionDays"] = "0",
["ServerHistorian:QuestDb:CommandTimeout"] = "00:00:00",
["ServerHistorian:QuestDb:VisibilityTimeout"] = "-00:00:01",
}).Build();
var result = new ServerHistorianOptionsValidator(configuration)
.Validate(null, new ServerHistorianOptions { Backend = "QuestDb", Enabled = true });
result.Failed.ShouldBeTrue();
foreach (var key in new[] { "ConnectionString", "TablePrefix", "RetentionDays", "CommandTimeout", "VisibilityTimeout" })
result.Failures.ShouldContain(f => f.Contains($"ServerHistorian:QuestDb:{key}"), key);
}
/// <summary>Unconsumed QuestDb section (nothing enabled) is not validated.</summary>
[Fact]
public void QuestDb_backend_not_consumed_passes_without_connection_string() =>
QuestDbSut(null).Validate(null, new ServerHistorianOptions { Backend = "QuestDb" }).Failed.ShouldBeFalse();
/// <summary>
/// An invalid <c>ServerHistorian:QuestDb:TablePrefix</c> must not fail the host when nothing consumes the
/// section: <see cref="HistorianBackendRegistry.Resolve"/> does not throw and the validator does not check it.
/// </summary>
[Fact]
public void QuestDb_invalid_prefix_not_consumed_neither_resolve_nor_validation_fails()
{
var configuration = new ConfigurationBuilder().AddInMemoryCollection(new Dictionary<string, string?>
{
["AlarmHistorian:Enabled"] = "false",
["ServerHistorian:QuestDb:ConnectionString"] = "Host=questdb;Port=8812",
["ServerHistorian:QuestDb:TablePrefix"] = "bad;prefix",
}).Build();
var options = new ServerHistorianOptions { Backend = "QuestDb" };
Should.NotThrow(() => HistorianBackendRegistry.Resolve(options, configuration));
new ServerHistorianOptionsValidator(configuration).Validate(null, options).Failed.ShouldBeFalse();
}
/// <summary>Consumed (here by AlarmHistorian alone), the same invalid prefix is a named validator error.</summary>
[Fact]
public void QuestDb_invalid_prefix_consumed_is_a_validator_error_naming_the_key()
{
var options = new ServerHistorianOptions { Backend = "QuestDb" };
var result = QuestDbSut("Host=questdb;Port=8812", "bad;prefix", alarmHistorianEnabled: true).Validate(null, options);
result.Failed.ShouldBeTrue();
result.Failures.ShouldContain(f => f.Contains("ServerHistorian:QuestDb:TablePrefix"));
}
} }
@@ -6,6 +6,7 @@ using Xunit;
using ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway; using ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway;
using ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Recorder; using ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Recorder;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions; using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb;
using ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer; using ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer;
using ZB.MOM.WW.OtOpcUa.Host.Configuration; using ZB.MOM.WW.OtOpcUa.Host.Configuration;
using ZB.MOM.WW.OtOpcUa.Runtime.Historian; using ZB.MOM.WW.OtOpcUa.Runtime.Historian;
@@ -86,4 +87,46 @@ public sealed class HistorianBackendRegistryTests
holder.Updater.ShouldNotBeNull(); holder.Updater.ShouldNotBeNull();
holder.Updater!.GetType().Name.ShouldBe("SqlServerHistorianDataUpdater"); holder.Updater!.GetType().Name.ShouldBe("SqlServerHistorianDataUpdater");
} }
private static IConfiguration QuestDbCfg() => new ConfigurationBuilder().AddInMemoryCollection(new Dictionary<string, string?>
{
["ServerHistorian:QuestDb:ConnectionString"] = "Host=localhost;Port=18812;Username=admin;Password=quest;Database=qdb",
["ServerHistorian:QuestDb:TablePrefix"] = "rig_",
}).Build();
[Theory]
[InlineData("questdb")]
[InlineData("QUESTDB")]
[InlineData("QuestDb")]
public void QuestDb_backend_resolves_case_insensitively(string name) =>
HistorianBackendRegistry.Resolve(new ServerHistorianOptions { Backend = name, Enabled = true }, QuestDbCfg())
.ShouldBeOfType<QuestDbHistorianProvider>().Name.ShouldBe("QuestDb");
[Fact]
public void QuestDb_backend_resolves_and_builds_the_questdb_seams()
{
var sp = new ServiceCollection().BuildServiceProvider();
var p = HistorianBackendRegistry.Resolve(new ServerHistorianOptions { Backend = "questdb", Enabled = true }, QuestDbCfg());
p.ShouldBeOfType<QuestDbHistorianProvider>();
HistorianBackendRegistry.KnownBackends.ShouldContain("QuestDb");
p.SupportedKinds.Count.ShouldBe(6);
p.SupportedKinds.ShouldContain(HistorianValueKind.String);
p.SupportedKinds.ShouldContain(HistorianValueKind.Empty);
p.CreateDataSource(sp).ShouldBeOfType<QuestDbHistorianDataSource>();
p.CreateValueWriter(sp).GetType().Name.ShouldBe("QuestDbHistorianValueWriter");
p.CreateAlarmWriter(sp).GetType().Name.ShouldBe("QuestDbAlarmHistorianWriter");
p.CreateProvisioning(sp).GetType().Name.ShouldBe("QuestDbTagProvisioner");
var holder = new HistorianUpdaterHolder(p.CreateUpdater(sp));
holder.Updater.ShouldNotBeNull();
holder.Updater!.GetType().Name.ShouldBe("QuestDbHistorianDataUpdater");
}
[Fact]
public void BindQuestDb_reads_the_QuestDb_section()
{
var o = HistorianBackendRegistry.BindQuestDb(QuestDbCfg());
o.ConnectionString.ShouldContain("Port=18812");
o.TablePrefix.ShouldBe("rig_");
}
} }
@@ -345,6 +345,51 @@ public sealed class HistoryPagingTests
store.TryTake(null, cp3).ShouldNotBeNull(); // retained store.TryTake(null, cp3).ShouldNotBeNull(); // retained
} }
[Fact]
public void SliceTieCluster_forward_drain_steps_by_the_backend_resolution()
{
// A microsecond backend (QuestDB, 10 ticks) truncates T + 1 tick back to T, so the drained cluster must be
// stepped off by the resolution, not by one tick, or the same sample comes back forever.
var t = new DateTime(2026, 1, 1, 0, 0, 5, DateTimeKind.Utc);
HistoryPaging.SliceTieCluster(
clusterCount: 1, skip: 1, cap: 1, boundaryT: t, endUtc: t.AddHours(1),
out _, out var sliceCount, out var nextStartUtc, out var nextSkip, resolutionTicks: 10);
sliceCount.ShouldBe(0);
nextStartUtc.ShouldBe(t.AddTicks(10));
nextSkip.ShouldBe(0);
}
[Fact]
public void SliceTieCluster_reverse_drain_steps_down_by_the_backend_resolution()
{
var t = new DateTime(2026, 1, 1, 0, 0, 5, DateTimeKind.Utc);
HistoryPaging.SliceTieCluster(
clusterCount: 1, skip: 1, cap: 1, boundaryT: t, endUtc: t.AddHours(-1),
out _, out _, out var nextStartUtc, out _, reverse: true, resolutionTicks: 10);
nextStartUtc.ShouldBe(t.AddTicks(-10));
}
[Fact]
public void SliceTieCluster_forward_drain_terminates_when_one_resolution_step_passes_the_end()
{
var t = new DateTime(2026, 1, 1, 0, 0, 5, DateTimeKind.Utc);
HistoryPaging.SliceTieCluster(
clusterCount: 1, skip: 1, cap: 1, boundaryT: t, endUtc: t.AddTicks(9),
out _, out _, out var nextStartUtc, out _, resolutionTicks: 10);
nextStartUtc.ShouldBeNull();
}
[Fact]
public void SliceTieCluster_rejects_a_resolution_below_one() =>
Should.Throw<ArgumentOutOfRangeException>(() => HistoryPaging.SliceTieCluster(
1, 1, 1, DateTime.UtcNow, DateTime.UtcNow.AddHours(1), out _, out _, out _, out _, resolutionTicks: 0));
private static HistoryContinuationState RawState() => new( private static HistoryContinuationState RawState() => new(
"WW.Tag", DateTime.UtcNow, DateTime.UtcNow.AddHours(1), "WW.Tag", DateTime.UtcNow, DateTime.UtcNow.AddHours(1),
BoundarySkipCount: 1, NumValuesPerNode: 50); BoundarySkipCount: 1, NumValuesPerNode: 50);
@@ -357,6 +357,56 @@ public sealed class NodeManagerHistoryReadPagingTests : IDisposable
await host.DisposeAsync(); await host.DisposeAsync();
} }
/// <summary>
/// Forward paging at <c>NumValuesPerNode = 1</c> over a MICROSECOND backend (QuestDB: it truncates every inbound
/// time to 10 ticks and reports <c>TimestampResolutionTicks = 10</c>) returns every sample exactly once and
/// terminates. Before the fix the drained one-sample "cluster" resumed at <c>T + 1 tick</c>, which the backend
/// truncated back to <c>T</c>: the same sample came back with no boundary skip, and the read looped forever.
/// </summary>
[Fact]
public async Task Raw_forward_read_with_one_value_per_page_on_a_microsecond_backend_returns_every_sample_once()
{
var (host, server) = await BootAsync();
var nm = server.NodeManager!;
nm.HistoryContinuationStore = new InMemoryHistoryContinuationStore();
// Microsecond-spaced samples, so a one-tick step would land strictly between two of them.
var series = Enumerable.Range(0, 6)
.Select(i => { var t = Epoch.AddTicks(10L * i); return new DataValueSnapshot((double)i, StatusCodes.Good, t, t); })
.ToArray();
nm.HistorianDataSource = new MicrosecondHistorianDataSource(series);
nm.EnsureVariable("eq-1/us", parentFolderNodeId: null, displayName: "Us", dataType: "Double",
writable: false, historianTagname: "WW.Us", realm: AddressSpaceRealm.Uns);
var nodeId = nm.TryGetVariable("eq-1/us")!.NodeId;
var collected = await DrainAsync(nm, nodeId, start: Epoch, end: Epoch.AddHours(1), max: 1);
collected.ShouldBe([0d, 1d, 2d, 3d, 4d, 5d]);
await host.DisposeAsync();
}
/// <summary>The reverse mirror on the same microsecond backend: newest first, each sample once.</summary>
[Fact]
public async Task Raw_reverse_read_with_one_value_per_page_on_a_microsecond_backend_returns_every_sample_once()
{
var (host, server) = await BootAsync();
var nm = server.NodeManager!;
nm.HistoryContinuationStore = new InMemoryHistoryContinuationStore();
var series = Enumerable.Range(0, 6)
.Select(i => { var t = Epoch.AddTicks(10L * (i + 1)); return new DataValueSnapshot((double)i, StatusCodes.Good, t, t); })
.ToArray();
nm.HistorianDataSource = new MicrosecondHistorianDataSource(series);
nm.EnsureVariable("eq-1/usrev", parentFolderNodeId: null, displayName: "UsRev", dataType: "Double",
writable: false, historianTagname: "WW.UsRev", realm: AddressSpaceRealm.Uns);
var nodeId = nm.TryGetVariable("eq-1/usrev")!.NodeId;
var collected = await DrainAsync(nm, nodeId, start: Epoch.AddHours(1), end: Epoch, max: 1);
collected.ShouldBe([5d, 4d, 3d, 2d, 1d, 0d]);
await host.DisposeAsync();
}
/// <summary>Pages a Raw read to completion and returns the union of the pages in arrival order.</summary> /// <summary>Pages a Raw read to completion and returns the union of the pages in arrival order.</summary>
private static Task<List<double>> DrainAsync(OtOpcUaNodeManager nm, NodeId nodeId, DateTime start, DateTime end, uint max) private static Task<List<double>> DrainAsync(OtOpcUaNodeManager nm, NodeId nodeId, DateTime start, DateTime end, uint max)
{ {
@@ -520,6 +570,43 @@ public sealed class NodeManagerHistoryReadPagingTests : IDisposable
public void Dispose() { } public void Dispose() { }
} }
/// <summary>
/// A microsecond-resolution backend, shaped like QuestDB: every inbound window bound is TRUNCATED to 10 ticks
/// (as <c>QuestDbConnection.Round</c> does) before the series is filtered, and the source reports
/// <see cref="TimestampResolutionTicks"/> = 10.
/// </summary>
private sealed class MicrosecondHistorianDataSource(IReadOnlyList<DataValueSnapshot> series) : IHistorianDataSource
{
private readonly SeriesHistorianDataSource _inner = new(series);
public long TimestampResolutionTicks => 10;
private static DateTime Truncate(DateTime t) => new(t.Ticks - t.Ticks % 10, DateTimeKind.Utc);
public Task<HistorianRead> ReadRawAsync(
string fullReference, DateTime startUtc, DateTime endUtc, uint maxValuesPerNode,
CancellationToken cancellationToken) =>
_inner.ReadRawAsync(fullReference, Truncate(startUtc), Truncate(endUtc), maxValuesPerNode, cancellationToken);
public Task<HistorianRead> ReadProcessedAsync(
string fullReference, DateTime startUtc, DateTime endUtc, TimeSpan interval,
HistoryAggregateType aggregate, CancellationToken cancellationToken) =>
_inner.ReadProcessedAsync(fullReference, startUtc, endUtc, interval, aggregate, cancellationToken);
public Task<HistorianRead> ReadAtTimeAsync(
string fullReference, IReadOnlyList<DateTime> timestampsUtc, CancellationToken cancellationToken) =>
_inner.ReadAtTimeAsync(fullReference, timestampsUtc, cancellationToken);
public Task<HistoricalEventsResult> ReadEventsAsync(
string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents,
CancellationToken cancellationToken) =>
_inner.ReadEventsAsync(sourceName, startUtc, endUtc, maxEvents, cancellationToken);
public HistorianHealthSnapshot GetHealthSnapshot() => _inner.GetHealthSnapshot();
public void Dispose() { }
}
private async Task<(OpcUaApplicationHost Host, OtOpcUaSdkServer Server)> BootAsync() private async Task<(OpcUaApplicationHost Host, OtOpcUaSdkServer Server)> BootAsync()
{ {
var applicationUri = $"urn:OtOpcUa.HistoryPagingTest:{Guid.NewGuid():N}"; var applicationUri = $"urn:OtOpcUa.HistoryPagingTest:{Guid.NewGuid():N}";