diff --git a/docker-dev/docker-compose.yml b/docker-dev/docker-compose.yml index 3d410597..d6e439b8 100644 --- a/docker-dev/docker-compose.yml +++ b/docker-dev/docker-compose.yml @@ -441,6 +441,9 @@ services: # docker compose -f docker-dev/docker-compose.yml up -d central-1 central-2 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}" + # 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 # 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 @@ -574,6 +577,9 @@ services: # docker compose -f docker-dev/docker-compose.yml up -d central-1 central-2 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}" + # 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 # carry it so the pair is symmetric; :9200 round-robins them, and the recorder must exist on # whichever node holds the drivers. diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/IHistorianDataSource.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/IHistorianDataSource.cs index 9942954e..beb0dea2 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/IHistorianDataSource.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/IHistorianDataSource.cs @@ -107,4 +107,19 @@ public interface IHistorianDataSource : IDisposable /// /// The current health snapshot for the historian backend. HistorianHealthSnapshot GetHealthSnapshot(); + + /// + /// The backend's storage timestamp resolution, in ticks (100 ns): the smallest step + /// between two distinct stored timestamps. 1 (the default) means full tick precision; QuestDB stores + /// microseconds and reports 10. + /// + /// + /// The server's HistoryRead-Raw paging steps OFF a drained timestamp by this amount (forward + /// T + resolution, reverse T − resolution). Stepping by less than the resolution is not + /// progress on a backend that truncates its inputs: T + 1 tick is truncated back to T, 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 T and T ± resolution. + /// Same-timestamp ties are unaffected (they are paged within the timestamp, never stepped over). + /// + long TimestampResolutionTicks => 1; } diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataSource.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataSource.cs index f4b97539..b1d92d01 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataSource.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataSource.cs @@ -19,8 +19,10 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb; /// /// Timestamps (R6). Every inbound time is normalised with and every /// is bound only through . Read bounds are not -/// clamped to the epoch (a pre-1970 bound is valid); only MinValue/MaxValue are nudged. Results -/// carry the rounded times, so an AtTime result is stamped at the rounded request time. +/// clamped to the epoch (a pre-1970 bound is valid); only MinValue/MaxValue are nudged. Raw and +/// 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. /// /// /// Reads never create tags (R7). A tag resolves only through ; @@ -44,6 +46,10 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb; /// public sealed class QuestDbHistorianDataSource : IHistorianDataSource { + /// + /// QuestDB stores microseconds: 10 ticks. HistoryRead paging steps off a drained timestamp by this much. + public long TimestampResolutionTicks => 10; + /// Most AtTime request times per statement (two UNION ALL branches each; QuestDB 10.0.1 refuses more than 126 branches). public const int AtTimeChunkSize = 50; @@ -139,8 +145,9 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource /// LIMIT 1 seeks, UNION ALL-ed, at most times per statement (spec §3, /// T0 Q9). then decides each value exactly as for SQL Server (exact hit, /// no previous ⇒ BadNoData, 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 - /// time. Unknown tag ⇒ every entry BadNoData. + /// result matches the request in length and order (duplicates included). Each entry is evaluated at its rounded + /// time but stamped with the requested time itself (UTC-normalised as does, + /// not truncated), matching the SQL Server backend. Unknown tag ⇒ every entry BadNoData. /// public async Task ReadAtTimeAsync( string fullReference, IReadOnlyList timestampsUtc, CancellationToken cancellationToken) @@ -197,10 +204,11 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource var result = new DataValueSnapshot[rounded.Length]; for (var i = 0; i < result.Length; i++) { - var t = rounded[i]; - result[i] = byTicks.TryGetValue(t.Ticks, out var v) - ? v - : new DataValueSnapshot(null, HistorianStatusCodes.BadNoData, t, now); + // Seek/evaluate at the rounded time; stamp the client's own requested time (M3). + var requested = AsUtc(timestampsUtc[i]); + result[i] = byTicks.TryGetValue(rounded[i].Ticks, out var v) + ? v with { SourceTimestampUtc = requested } + : new DataValueSnapshot(null, HistorianStatusCodes.BadNoData, requested, now); } RecordSuccess(); @@ -213,6 +221,14 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource } } + /// The UTC normalisation of without its truncation or nudging. + private static DateTime AsUtc(DateTime t) => t.Kind switch + { + DateTimeKind.Local => t.ToUniversalTime(), + DateTimeKind.Utc => t, + _ => DateTime.SpecifyKind(t, DateTimeKind.Utc), + }; + /// /// The T0 Q9 statement for request times: $1 = tag, $(k+2) = time k. /// Columns: req, dir (0 = previous-or-equal, 1 = next), then ts, kind, num_value, int_value, str_value, diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataUpdater.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataUpdater.cs index 27e4f416..f4e00820 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataUpdater.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataUpdater.cs @@ -31,9 +31,6 @@ internal sealed class QuestDbHistorianDataUpdater : IHistorianDataUpdater /// Rows per INSERT and per DeleteRaw page (T0 Q2: 9,000 parameters, far under the 32,767 limit). internal const int BatchSize = 1000; - /// Request times further apart than this are read with separate range queries (I-2). - internal static readonly TimeSpan ExistenceClusterGap = TimeSpan.FromMinutes(1); - private const long ResolutionTicks = 10; private const int LockStripes = 64; private readonly SemaphoreSlim[] _locks = Enumerable.Range(0, LockStripes).Select(_ => new SemaphoreSlim(1, 1)).ToArray(); @@ -60,26 +57,6 @@ internal sealed class QuestDbHistorianDataUpdater : IHistorianDataUpdater /// Rows returned by existence reads so far (test/diagnostic counter). internal long LiveRowsRead => Interlocked.Read(ref _liveRowsRead); - /// Splits times into clusters wherever consecutive sorted times are further apart than . - /// Request times (any order). - /// Largest gap kept inside one cluster. - /// Inclusive [Lo, Hi] ranges, ascending. - internal static List<(DateTime Lo, DateTime Hi)> Cluster(IReadOnlyList 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 LockAsync(string tag, CancellationToken ct) { var sem = _locks[(int)((uint)StringComparer.Ordinal.GetHashCode(tag) % LockStripes)]; @@ -231,7 +208,7 @@ internal sealed class QuestDbHistorianDataUpdater : IHistorianDataUpdater { var live = new HashSet(); 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, $"SELECT ts FROM {_conn.SampleTable} WHERE tag = $1 AND ts >= $2 AND ts <= $3 AND NOT deleted"); diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianProvider.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianProvider.cs new file mode 100644 index 00000000..39af8413 --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianProvider.cs @@ -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; + +/// +/// for the QuestDB historian. Supports every +/// 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. +/// +/// +/// The shared state is built on the first Create* 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 ServerHistorianOptionsValidator). The first +/// Create* call's service provider supplies the logger factory the shared pieces log through. +/// +public sealed class QuestDbHistorianProvider : IHistorianBackendProvider +{ + private static readonly FrozenSet Kinds = + Enum.GetValues().ToFrozenSet(); + + private readonly QuestDbHistorianOptions _options; + private readonly object _initLock = new(); + private Shared? _shared; + + /// Creates the provider over the ServerHistorian:QuestDb options (validated elsewhere). + /// The ServerHistorian:QuestDb options. + public QuestDbHistorianProvider(QuestDbHistorianOptions options) + { + _options = options ?? throw new ArgumentNullException(nameof(options)); + } + + /// + public string Name => "QuestDb"; + + /// + public IReadOnlySet 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()); + var wal = new QuestDbWalVisibility(conn, _options.VisibilityTimeout, loggers.CreateLogger()); + var shared = new Shared(conn, bootstrap, new QuestDbTagCatalog(conn, bootstrap, wal), wal); + Volatile.Write(ref _shared, shared); + return shared; + } + } + + /// + public IHistorianDataSource CreateDataSource(IServiceProvider sp) + { + var s = Get(sp); + return new QuestDbHistorianDataSource(s.Conn, s.Bootstrap, s.Catalog, + Loggers(sp).CreateLogger()); + } + + /// + public IHistorianValueWriter CreateValueWriter(IServiceProvider sp) + { + var s = Get(sp); + return new QuestDbHistorianValueWriter(s.Conn, s.Catalog, s.Wal, + Loggers(sp).CreateLogger()); + } + + /// + public IAlarmHistorianWriter CreateAlarmWriter(IServiceProvider sp) + { + var s = Get(sp); + return new QuestDbAlarmHistorianWriter(s.Conn, s.Bootstrap, s.Wal, + Loggers(sp).CreateLogger()); + } + + /// + public IHistorianProvisioning CreateProvisioning(IServiceProvider sp) + { + var s = Get(sp); + return new QuestDbTagProvisioner(s.Conn, s.Bootstrap, s.Catalog, s.Wal, + Loggers(sp).CreateLogger()); + } + + /// + /// The soft-delete updater: deletes write tombstones (INSERT only, never QuestDB UPDATE/DELETE). + public IHistorianDataUpdater? CreateUpdater(IServiceProvider sp) + { + var s = Get(sp); + return new QuestDbHistorianDataUpdater(s.Conn, s.Catalog, s.Wal, + Loggers(sp).CreateLogger()); + } + + private sealed record Shared( + QuestDbConnection Conn, QuestDbSchemaBootstrapper Bootstrap, QuestDbTagCatalog Catalog, QuestDbWalVisibility Wal); +} diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianValueWriter.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianValueWriter.cs index 791ef010..a7fba836 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianValueWriter.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianValueWriter.cs @@ -98,49 +98,13 @@ internal sealed class QuestDbHistorianValueWriter : IHistorianValueWriter } } - /// A gap between consecutive batch times larger than this starts a new existence-query range. - internal static readonly TimeSpan ExistenceClusterGap = TimeSpan.FromMinutes(1); - - /// - /// Splits ascending into inclusive [Lo, Hi] ranges: a new range starts where the gap to the - /// previous time exceeds or the range already holds times. One outlier - /// time therefore costs its own tiny query instead of widening one range over every stored row in between. - /// - /// Times in ascending order. - /// Largest gap kept inside one range. - /// Most times in one range (>= 1). - /// The ranges, ascending. - internal static IReadOnlyList<(DateTime Lo, DateTime Hi)> Cluster(IReadOnlyList 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). private async Task> ExistingTicksAsync(string tag, List rows, CancellationToken ct) { var times = rows.Select(r => QuestDbConnection.Round(r.TimeUtc)).OrderBy(t => t).ToList(); var set = new HashSet(); 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"); QuestDbConnection.BindText(cmd, tag); diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbTimeClusters.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbTimeClusters.cs new file mode 100644 index 00000000..442e7b34 --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbTimeClusters.cs @@ -0,0 +1,49 @@ +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb; + +/// +/// Groups the times of one write or HistoryUpdate request into inclusive [Lo, Hi] ranges so that the +/// "which of these times already exist" check reads one range query per cluster instead of one query spanning +/// [min, max] 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. +/// +internal static class QuestDbTimeClusters +{ + /// Request times further apart than this start a new existence-query range. + internal static readonly TimeSpan ExistenceGap = TimeSpan.FromMinutes(1); + + /// + /// Splits (any order; sorted here) into inclusive ranges, ascending: a new range starts + /// where the gap to the previous time exceeds (exactly stays + /// together), or where the current range already holds times. + /// + /// Request times, in any order. + /// Largest gap kept inside one range. + /// Most times in one range (>= 1); unbounded by default. + /// The ranges, ascending; empty for no times. + internal static IReadOnlyList<(DateTime Lo, DateTime Hi)> Cluster( + IReadOnlyList 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; + } +} diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/HistorianBackendRegistry.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/HistorianBackendRegistry.cs index ce68db91..a14e5bd6 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/HistorianBackendRegistry.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/HistorianBackendRegistry.cs @@ -1,6 +1,7 @@ using Microsoft.Extensions.Configuration; using Microsoft.Extensions.Options; 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.Runtime.Historian; @@ -25,11 +26,23 @@ public static class HistorianBackendRegistry return o; } + /// Backend name for the QuestDB historian. + public const string QuestDbBackendName = "QuestDb"; + + /// Binds the ServerHistorian:QuestDb section (shared by the registry and the validator). + public static QuestDbHistorianOptions BindQuestDb(IConfiguration configuration) + { + var o = new QuestDbHistorianOptions(); + configuration.GetSection($"{ServerHistorianOptions.SectionName}:QuestDb").Bind(o); + return o; + } + private static readonly IReadOnlyDictionary> Factories = new Dictionary>(StringComparer.OrdinalIgnoreCase) { [ServerHistorianOptions.GatewayBackendName] = (o, _) => new GatewayHistorianProvider(o), [SqlServerBackendName] = (_, c) => new SqlServerHistorianProvider(BindSqlServer(c)), + [QuestDbBackendName] = (_, c) => new QuestDbHistorianProvider(BindQuestDb(c)), }; /// The registered backend names, for validation messages. diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/ServerHistorianOptionsValidator.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/ServerHistorianOptionsValidator.cs index e421747f..ac51be90 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/ServerHistorianOptionsValidator.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/ServerHistorianOptionsValidator.cs @@ -78,6 +78,13 @@ public sealed class ServerHistorianOptionsValidator : OptionsValidatorBase + diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/HistoryPaging.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/HistoryPaging.cs index d1ffae37..6fa8c12b 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/HistoryPaging.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/HistoryPaging.cs @@ -212,25 +212,34 @@ internal static class HistoryPaging /// after (the reverse window excludes end; a resume at exactly end /// would turn into a start == end point read and return the excluded sample). Forward is unchanged. /// + /// + /// The backend's storage timestamp resolution (IHistorianDataSource.TimestampResolutionTicks, >= 1). A + /// drained cluster steps off by this many ticks rather than by one: a backend that + /// truncates its inputs to its resolution (QuestDB, 10 ticks) would otherwise truncate T + 1 tick back + /// to T and return the drained sample again, forever. Lossless for the same reason one tick is at tick + /// precision: no stored timestamp lies strictly between T and T ± resolution. The default + /// 1 keeps the tick-precision behaviour exactly. + /// public static void SliceTieCluster( int clusterCount, int skip, uint cap, DateTime boundaryT, DateTime endUtc, 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); sliceCount = Math.Min((int)cap, clusterCount - sliceStart); var emitted = sliceStart + sliceCount; if (emitted < clusterCount) { nextStartUtc = boundaryT; nextSkip = emitted; } else if (reverse) { - // DateTime.MinValue cannot step down; nothing can lie below it anyway. - var more = boundaryT > DateTime.MinValue && boundaryT.AddTicks(-1) > endUtc; - nextStartUtc = more ? boundaryT.AddTicks(-1) : null; + // Nothing can lie below DateTime.MinValue, so a boundary within one resolution step of it cannot step down. + var more = boundaryT.Ticks >= resolutionTicks && boundaryT.AddTicks(-resolutionTicks) > endUtc; + nextStartUtc = more ? boundaryT.AddTicks(-resolutionTicks) : null; nextSkip = 0; } else { - var next = boundaryT.AddTicks(1); + var next = boundaryT.AddTicks(resolutionTicks); if (next <= endUtc) { nextStartUtc = next; nextSkip = 0; } else { nextStartUtc = null; nextSkip = 0; } } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs index 57f249c2..859b69b2 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs @@ -5931,7 +5931,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 HistoryPaging.SliceTieCluster( cluster.Count, boundarySkip, numValuesPerNode, startUtc, endUtc, 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(sliceCount); for (var i = sliceStart; i < sliceStart + sliceCount; i++) slice.Add(cluster[i]); diff --git a/tests/Drivers/Historian.Conformance/AtTimeConformance.cs b/tests/Drivers/Historian.Conformance/AtTimeConformance.cs index 8a2acea7..2f8a9341 100644 --- a/tests/Drivers/Historian.Conformance/AtTimeConformance.cs +++ b/tests/Drivers/Historian.Conformance/AtTimeConformance.cs @@ -35,7 +35,7 @@ public abstract class AtTimeConformance : ConformanceBase var s = (await B.CreateDataSource().ReadAtTimeAsync("A", [At(10), At(20)], Ct)).Samples; s[0].Value.ShouldBe(20d); 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[1].Value.ShouldBe(30d); // a stored Bad sample comes back as stored 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[2].Value.ShouldBe(42.5d); // 40 + 10 * 2.5/10 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] @@ -61,7 +61,7 @@ public abstract class AtTimeConformance : ConformanceBase var s = (await B.CreateDataSource().ReadAtTimeAsync("A", [before], Ct)).Samples.Single(); s.Value.ShouldBeNull(); s.StatusCode.ShouldBe(BadNoData); - s.SourceTimestampUtc.ShouldBe(B.Round(before)); + s.SourceTimestampUtc.ShouldBe(before); } [Fact] @@ -80,7 +80,7 @@ public abstract class AtTimeConformance : ConformanceBase 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.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] @@ -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; s.Count.ShouldBe(4); 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] diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/AtTimeRegressionTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/AtTimeRegressionTests.cs new file mode 100644 index 00000000..6f5a40fa --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/AtTimeRegressionTests.cs @@ -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; + +/// +/// QuestDB AtTime regressions found in the T5 review (M1, M3), and the data source's reported timestamp +/// resolution (the HistoryRead paging step, C1). +/// +[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(); + } + + /// + /// 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. + /// + [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); + } + + /// M3 on the unknown-tag path: every BadNoData entry carries its own requested time. + [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); + } + + /// + /// 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). + /// + [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); + } + + /// C1: the data source reports QuestDB's microsecond storage resolution to the paging code. + [Fact] + public void Data_source_reports_a_ten_tick_timestamp_resolution() => + ((IHistorianDataSource)_b.CreateDataSource()).TimestampResolutionTicks.ShouldBe(10); +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/PagingRegressionTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/PagingRegressionTests.cs new file mode 100644 index 00000000..642c6b8f --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/PagingRegressionTests.cs @@ -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; + +/// +/// 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 NodeManagerHistoryReadPagingTests runs on a fake that models them). +/// HistoryPaging 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 ) over the real reads. +/// +[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!; + + /// Sample i at T0 + i µs (adjacent microseconds: a one-tick step lands strictly between two of them). + 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 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); + } + + /// The node manager's Raw paging loop at NumValuesPerNode = 1 (see ServeRawPagedAsync). + private static async Task> WalkAsync(IHistorianDataSource ds, DateTime start, DateTime end) + { + const uint cap = 1; + var res = ds.TimestampResolutionTicks; + var reverse = start > end; + var collected = new List(); + 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; + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/QuestDbConformance.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/QuestDbConformance.cs new file mode 100644 index 00000000..63477891 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/QuestDbConformance.cs @@ -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 +{ + /// + protected override Task CreateBackendAsync() => QuestDbConformanceBackend.CreateAsync(fx); +} + +[Collection(QuestDbCollection.Name)] +public sealed class QuestDbAtTimeConformance(QuestDbFixture fx) : AtTimeConformance +{ + /// + protected override Task CreateBackendAsync() => QuestDbConformanceBackend.CreateAsync(fx); +} + +[Collection(QuestDbCollection.Name)] +public sealed class QuestDbProcessedConformance(QuestDbFixture fx) : ProcessedConformance +{ + /// + protected override Task CreateBackendAsync() => QuestDbConformanceBackend.CreateAsync(fx); +} + +[Collection(QuestDbCollection.Name)] +public sealed class QuestDbQualityContractConformance(QuestDbFixture fx) : QualityContractConformance +{ + /// + protected override Task CreateBackendAsync() => QuestDbConformanceBackend.CreateAsync(fx); +} + +[Collection(QuestDbCollection.Name)] +public sealed class QuestDbLiveWriteConformance(QuestDbFixture fx) : LiveWriteConformance +{ + /// + protected override Task CreateBackendAsync() => QuestDbConformanceBackend.CreateAsync(fx); +} + +[Collection(QuestDbCollection.Name)] +public sealed class QuestDbUpdateConformance(QuestDbFixture fx) : UpdateConformance +{ + /// + protected override Task CreateBackendAsync() => QuestDbConformanceBackend.CreateAsync(fx); +} + +[Collection(QuestDbCollection.Name)] +public sealed class QuestDbTagNameConformance(QuestDbFixture fx) : TagNameConformance +{ + /// + protected override Task CreateBackendAsync() => QuestDbConformanceBackend.CreateAsync(fx); +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/QuestDbConformanceBackend.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/QuestDbConformanceBackend.cs index 5dfc6f4c..bf863363 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/QuestDbConformanceBackend.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/QuestDbConformanceBackend.cs @@ -76,14 +76,13 @@ internal sealed class QuestDbConformanceBackend : IConformanceBackend new QuestDbHistorianDataSource(Conn, Boot, NewCatalog(), timeProvider: clock); /// - /// TODO(T7): build the T4 QuestDbHistorianValueWriter here once T4 is merged. + /// The T4 live-value writer over a fresh catalog (fresh positive cache) and this instance's WAL wait. public IHistorianValueWriter CreateValueWriter(Func? 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); /// - /// TODO(T7): build the T6 QuestDbHistorianDataUpdater here once T6 is merged. - public IHistorianDataUpdater CreateUpdater() => - throw new NotSupportedException("TODO(T7): the QuestDB history updater (T6) is not merged into this branch yet."); + /// The T6 soft-delete updater over a fresh catalog and this instance's WAL wait. + public IHistorianDataUpdater CreateUpdater() => new QuestDbHistorianDataUpdater(Conn, NewCatalog(), Wal); /// /// diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/QuestDbFixture.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/QuestDbFixture.cs index dcaba660..4d9d0877 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/QuestDbFixture.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/QuestDbFixture.cs @@ -43,9 +43,19 @@ public sealed class QuestDbFixture : IAsyncLifetime /// Optional TTL. /// AutoCreateSchema. /// Valid options. + /// + /// Pooled connections idle for more than a second are closed (Connection Idle Lifetime / + /// Connection Pruning Interval). Many tests build their own (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. + /// 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, RetentionDays = retentionDays, AutoCreateSchema = autoCreate, diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/ReadTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/ReadTests.cs index d0f29d5a..6345613b 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/ReadTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/ReadTests.cs @@ -184,7 +184,7 @@ public sealed class ReadTests(QuestDbFixture fx) : IAsyncLifetime times.Add(times[0]); var s = (await _b.CreateDataSource().ReadAtTimeAsync("A", times, Ct)).Samples; 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].StatusCode.ShouldBe(GoodInterpolated); s[^1].Value.ShouldBe(49d); // the duplicate diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/TombstoneTests.Redelivery.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/TombstoneTests.Redelivery.cs new file mode 100644 index 00000000..c3b85295 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/TombstoneTests.Redelivery.cs @@ -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 +{ + /// + /// 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 NOT deleted), so the redelivered row is skipped. + /// + [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(); + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/TombstoneTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/TombstoneTests.cs index 2f3f53e3..841b73b7 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/TombstoneTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/TombstoneTests.cs @@ -7,7 +7,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests; /// Soft-delete behaviour. The writer-redelivery case lives in T7 (TombstoneTests.Redelivery.cs). [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); diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/UpdateTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/UpdateTests.cs index 0f721065..4d4e52a5 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/UpdateTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/UpdateTests.cs @@ -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] public async Task Existence_reads_are_bounded_when_request_times_are_far_apart() { diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/WriteTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/WriteTests.cs index 39e75e1b..a090346f 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/WriteTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/WriteTests.cs @@ -274,23 +274,4 @@ public sealed class WriteTests(QuestDbFixture fx) (await w.WriteLiveValuesAsync("tag.far", vals, CancellationToken.None)).ShouldBeTrue(); (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])]); - } } diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests/QuestDbHistorianProviderTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests/QuestDbHistorianProviderTests.cs new file mode 100644 index 00000000..b5df992b --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests/QuestDbHistorianProviderTests.cs @@ -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(), 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(); + p.CreateValueWriter(sp).ShouldBeOfType(); + p.CreateAlarmWriter(sp).ShouldBeOfType(); + p.CreateProvisioning(sp).ShouldBeOfType(); + p.CreateUpdater(sp).ShouldBeOfType(); + } + + [Fact] + public void Supported_kinds_are_a_frozen_set() => + Make().SupportedKinds.ShouldBeAssignableTo>(); + + /// + /// 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. + /// + [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(() => 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(() => 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; + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests/QuestDbTimeClustersTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests/QuestDbTimeClustersTests.cs new file mode 100644 index 00000000..f68f9521 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests/QuestDbTimeClustersTests.cs @@ -0,0 +1,62 @@ +using Shouldly; +using Xunit; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests; + +/// +/// The shared existence-query clustering (moved here, names unchanged, from the writer's WriteTests and the +/// updater's UpdateTests when the two private copies were consolidated into ). +/// +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(() => QuestDbTimeClusters.Cluster([T0], TimeSpan.FromMinutes(1), 0)); +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/ServerHistorianOptionsValidatorTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/ServerHistorianOptionsValidatorTests.cs index 00945b11..9b378e4a 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/ServerHistorianOptionsValidatorTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/ServerHistorianOptionsValidatorTests.cs @@ -375,4 +375,89 @@ public sealed class ServerHistorianOptionsValidatorTests result.Failed.ShouldBeTrue(); 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 { ["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()); + } + + /// QuestDb with a connection string passes; the gateway Endpoint/ApiKey/UseTls checks do not fire. + [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 ?? [])); + } + + /// QuestDb with no connection string fails, naming the key, and never asks for gateway settings. + [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")); + } + + /// Every QuestDbHistorianOptions.Validate() error is reported, each naming its key. + [Fact] + public void QuestDb_backend_reports_every_options_error_naming_the_key() + { + var configuration = new ConfigurationBuilder().AddInMemoryCollection(new Dictionary + { + ["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); + } + + /// Unconsumed QuestDb section (nothing enabled) is not validated. + [Fact] + public void QuestDb_backend_not_consumed_passes_without_connection_string() => + QuestDbSut(null).Validate(null, new ServerHistorianOptions { Backend = "QuestDb" }).Failed.ShouldBeFalse(); + + /// + /// An invalid ServerHistorian:QuestDb:TablePrefix must not fail the host when nothing consumes the + /// section: does not throw and the validator does not check it. + /// + [Fact] + public void QuestDb_invalid_prefix_not_consumed_neither_resolve_nor_validation_fails() + { + var configuration = new ConfigurationBuilder().AddInMemoryCollection(new Dictionary + { + ["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(); + } + + /// Consumed (here by AlarmHistorian alone), the same invalid prefix is a named validator error. + [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")); + } } diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Configuration/HistorianBackendRegistryTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Configuration/HistorianBackendRegistryTests.cs index 9251df6a..fe66467c 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Configuration/HistorianBackendRegistryTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Configuration/HistorianBackendRegistryTests.cs @@ -6,6 +6,7 @@ using Xunit; using ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway; using ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Recorder; 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.Host.Configuration; using ZB.MOM.WW.OtOpcUa.Runtime.Historian; @@ -86,4 +87,46 @@ public sealed class HistorianBackendRegistryTests holder.Updater.ShouldNotBeNull(); holder.Updater!.GetType().Name.ShouldBe("SqlServerHistorianDataUpdater"); } + + private static IConfiguration QuestDbCfg() => new ConfigurationBuilder().AddInMemoryCollection(new Dictionary + { + ["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().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(); + HistorianBackendRegistry.KnownBackends.ShouldContain("QuestDb"); + p.SupportedKinds.Count.ShouldBe(6); + p.SupportedKinds.ShouldContain(HistorianValueKind.String); + p.SupportedKinds.ShouldContain(HistorianValueKind.Empty); + p.CreateDataSource(sp).ShouldBeOfType(); + 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_"); + } } diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/HistoryPagingTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/HistoryPagingTests.cs index b97d859a..aeab3c5e 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/HistoryPagingTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/HistoryPagingTests.cs @@ -345,6 +345,51 @@ public sealed class HistoryPagingTests 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(() => HistoryPaging.SliceTieCluster( + 1, 1, 1, DateTime.UtcNow, DateTime.UtcNow.AddHours(1), out _, out _, out _, out _, resolutionTicks: 0)); + private static HistoryContinuationState RawState() => new( "WW.Tag", DateTime.UtcNow, DateTime.UtcNow.AddHours(1), BoundarySkipCount: 1, NumValuesPerNode: 50); diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadPagingTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadPagingTests.cs index b9388b55..80785796 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadPagingTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadPagingTests.cs @@ -357,6 +357,56 @@ public sealed class NodeManagerHistoryReadPagingTests : IDisposable await host.DisposeAsync(); } + /// + /// Forward paging at NumValuesPerNode = 1 over a MICROSECOND backend (QuestDB: it truncates every inbound + /// time to 10 ticks and reports TimestampResolutionTicks = 10) returns every sample exactly once and + /// terminates. Before the fix the drained one-sample "cluster" resumed at T + 1 tick, which the backend + /// truncated back to T: the same sample came back with no boundary skip, and the read looped forever. + /// + [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(); + } + + /// The reverse mirror on the same microsecond backend: newest first, each sample once. + [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(); + } + /// Pages a Raw read to completion and returns the union of the pages in arrival order. private static Task> DrainAsync(OtOpcUaNodeManager nm, NodeId nodeId, DateTime start, DateTime end, uint max) { @@ -520,6 +570,43 @@ public sealed class NodeManagerHistoryReadPagingTests : IDisposable public void Dispose() { } } + /// + /// A microsecond-resolution backend, shaped like QuestDB: every inbound window bound is TRUNCATED to 10 ticks + /// (as QuestDbConnection.Round does) before the series is filtered, and the source reports + /// = 10. + /// + private sealed class MicrosecondHistorianDataSource(IReadOnlyList 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 ReadRawAsync( + string fullReference, DateTime startUtc, DateTime endUtc, uint maxValuesPerNode, + CancellationToken cancellationToken) => + _inner.ReadRawAsync(fullReference, Truncate(startUtc), Truncate(endUtc), maxValuesPerNode, cancellationToken); + + public Task ReadProcessedAsync( + string fullReference, DateTime startUtc, DateTime endUtc, TimeSpan interval, + HistoryAggregateType aggregate, CancellationToken cancellationToken) => + _inner.ReadProcessedAsync(fullReference, startUtc, endUtc, interval, aggregate, cancellationToken); + + public Task ReadAtTimeAsync( + string fullReference, IReadOnlyList timestampsUtc, CancellationToken cancellationToken) => + _inner.ReadAtTimeAsync(fullReference, timestampsUtc, cancellationToken); + + public Task 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() { var applicationUri = $"urn:OtOpcUa.HistoryPagingTest:{Guid.NewGuid():N}";