From 8d82d049e955ddd2186873e8a0840bddc121a944 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Sun, 11 Oct 2026 03:35:35 -0400 Subject: [PATCH] fix(historian): QuestDB final-review fix wave (I-1, M-1, M-3, M-5, M-8, M-9) - I-1: QuestDbConnection caps Npgsql Maximum Pool Size at 24 (DefaultMaxPoolSize) when the connection string does not set it; an explicit value is kept. The effective value is logged once at Information. - M-1: QuestDbTagCatalog.EnsureAsync waits for the tag table's sequenced commits and re-checks before inserting, so it no longer overwrites a provisioner row (unit/description) that is committed but not yet applied. Integration test reproduces the clobber against real QuestDB (falsified). - M-3: the pre-1970 drop Warning is rate-limited per tag (once per 10 min); DroppedPre1970 still counts every sample. - M-5: QuestDbHistorianProvider is IAsyncDisposable/IDisposable and disposes the shared Npgsql pool; Program.cs registers the provider through a factory and every seam resolves it from the container, so DI disposes it at shutdown. - M-8, M-9: docs fixes (HistoryUpdate transaction wording; folded duplicate HistoryUpdate paragraph). Pool-size docs updated for the new default cap. Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe --- docs/Configuration.md | 12 ++-- docs/Historian.md | 13 ++-- .../QuestDbConnection.cs | 44 +++++++++++- .../QuestDbHistorianProvider.cs | 30 +++++++- .../QuestDbHistorianValueWriter.cs | 30 +++++++- .../QuestDbTagCatalog.cs | 29 ++++++-- src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs | 16 +++-- .../ProvisioningTests.cs | 33 +++++++++ .../QuestDbConnectionTests.cs | 44 ++++++++++++ .../QuestDbHistorianProviderTests.cs | 17 +++++ .../QuestDbHistorianValueWriterTests.cs | 71 +++++++++++++++++++ 11 files changed, 309 insertions(+), 30 deletions(-) create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests/QuestDbHistorianValueWriterTests.cs diff --git a/docs/Configuration.md b/docs/Configuration.md index 597fa8cc..4b23cf3c 100644 --- a/docs/Configuration.md +++ b/docs/Configuration.md @@ -419,19 +419,17 @@ when `ServerHistorian:Enabled` **or** `AlarmHistorian:Enabled` is true (`QuestDb | Key | Default | Validation / meaning | |---|---|---| -| `ServerHistorian:QuestDb:ConnectionString` | `""` | **Required.** Npgsql connection string for QuestDB's PG-wire port, e.g. `Host=questdb;Port=8812;Username=admin;Password=quest;Database=qdb;ServerCompatibilityMode=NoTypeLoading`. Env `ServerHistorian__QuestDb__ConnectionString` / `${secret:}` — never commit a real one. Set Npgsql `Maximum Pool Size` below QuestDB's 64-connection default limit (Npgsql's own default is 100), per node. | +| `ServerHistorian:QuestDb:ConnectionString` | `""` | **Required.** Npgsql connection string for QuestDB's PG-wire port, e.g. `Host=questdb;Port=8812;Username=admin;Password=quest;Database=qdb;ServerCompatibilityMode=NoTypeLoading`. Env `ServerHistorian__QuestDb__ConnectionString` / `${secret:}` — never commit a real one. When the string does not set Npgsql `Maximum Pool Size`, the backend caps it at **24** (Npgsql's own default of 100 exceeds QuestDB's 64-connection default limit); an explicit value is kept as written, and the effective value is logged at Information. Keep the per-node total of a pair under QuestDB's limit. | | `ServerHistorian:QuestDb:TablePrefix` | `hist_` | Must match `^[A-Za-z0-9_]+$` (spliced into DDL/DML). Tables are `{prefix}sample`, `{prefix}event`, `{prefix}tag`. | | `ServerHistorian:QuestDb:RetentionDays` | unset | `>= 1` when set. Applied as table TTL to the sample and event tables (enforced on write commits only — a quiet table never drops). Unset keeps data forever. | | `ServerHistorian:QuestDb:CommandTimeout` | `00:00:30` | Must be `> 0`; rounded up to whole seconds. | | `ServerHistorian:QuestDb:AutoCreateSchema` | `true` | `true` creates the tables and reconciles the TTL on first use; `false` only verifies the tables exist and fails every historian call (health `LastError`, no crash) if one is missing. | | `ServerHistorian:QuestDb:VisibilityTimeout` | `00:00:05` | Must be `>= 0`. How long a write waits for its rows to become visible to queries (WAL apply); `0` disables the wait. A timeout only logs a Warning. | -HistoryUpdate (soft delete) needs `ServerHistorian:Enabled=true` and a `HistoryUpdate` grant, as for `SqlServer`. Semantics, the µs timestamp rules, -pre-1970 limits, TTL, and the docker-dev overlay recipe: [`docs/Historian.md`](Historian.md#questdb-backend). - -HistoryUpdate (SqlServer and QuestDb only) additionally needs `ServerHistorian:Enabled=true` and a key whose principal holds the -`HistoryUpdate` permission — see [`docs/security.md`](security.md#historyupdate-historian-writes). Full semantics, schema, aggregates and the -docker-dev recipe: [`docs/Historian.md`](Historian.md#sql-server-backend). +HistoryUpdate (SqlServer and QuestDb only; QuestDb deletes are soft — tombstones) additionally needs `ServerHistorian:Enabled=true` and a key +whose principal holds the `HistoryUpdate` permission — see [`docs/security.md`](security.md#historyupdate-historian-writes). Full semantics, +schema, aggregates and the docker-dev recipe: [`docs/Historian.md`](Historian.md#sql-server-backend) (SqlServer) and +[`docs/Historian.md`](Historian.md#questdb-backend) (QuestDb: the µs timestamp rules, pre-1970 limits, TTL and the docker-dev overlay). #### `ContinuousHistorization` (recorder + outbox knobs) diff --git a/docs/Historian.md b/docs/Historian.md index e18e8ae1..38d9bd37 100644 --- a/docs/Historian.md +++ b/docs/Historian.md @@ -840,9 +840,11 @@ Read only when `Backend=QuestDb`; bound by `HistorianBackendRegistry.BindQuestDb | `QuestDb:VisibilityTimeout` | `00:00:05` | `>= 0` | How long a write waits for its rows to become visible to queries ([WAL visibility](#wal-visibility-and-the-read-your-writes-wait)). `0` disables the wait. | ⚠️ **Connection pool size.** QuestDB's default limit is **64** concurrent PG-wire connections (`pg.net.connection.limit`), while Npgsql's -default `Maximum Pool Size` is **100**. A busy node (HistoryRead batches, the recorder, the updater and alarm writes all share one pool) -can therefore be refused by QuestDB before Npgsql's own pool is full. Set `Maximum Pool Size` in the connection string below the server -limit — and below it **per node**, since a redundant pair shares one QuestDB: for example `…;Maximum Pool Size=24` on each of two nodes. +own default `Maximum Pool Size` is **100**. A busy node (HistoryRead batches, the recorder, the updater and alarm writes all share one pool) +could therefore be refused by QuestDB before Npgsql's pool is full. **When the connection string does not set `Maximum Pool Size` +(or `MaxPoolSize`), the backend now caps it at 24**, so the two nodes of a redundant pair sharing one QuestDB stay under the server limit +together (2 × 24 = 48). An explicit value in the connection string is always kept as written — including one above 64, since QuestDB's +limit is server-configurable. The effective value is logged once at Information when the backend first builds its pool. ### Tables @@ -876,7 +878,7 @@ through `BindTimestamp`. Consequences: timestamp by that amount (forward `T + resolution`, reverse `T − resolution`). Stepping by 1 tick would be truncated back to `T` and a one-value-per-page forward read would loop forever. Same-timestamp ties (legal on the Gateway backend) are still paged within the timestamp. - **Pre-1970 timestamps cannot be stored** (QuestDB: `designated timestamp before 1970-01-01 is not allowed`; the whole INSERT fails). So: - - the **live writer** drops such samples with one Warning per batch (counted in `DroppedPre1970`) — a retry could never succeed, and sending + - the **live writer** drops such samples with at most one Warning per tag per 10 minutes (a device with an unset clock would otherwise warn every drain pass; every dropped sample is still counted in `DroppedPre1970`) — a retry could never succeed, and sending them would push the tag into the #538 quarantine; - **HistoryUpdate** Insert / Replace / Update of such an item returns `BadOutOfRange` (nothing is sent); `DeleteAtTime` returns `BadNoEntryExists`; `DeleteRaw` finds nothing; @@ -1008,7 +1010,8 @@ batch limiter and `HistoryReadDeadline`. | `DeleteRawModifiedDetails` (`IsDeleteModified=false`) | `HistoryDeleteRawModified` | one node-level code: `Good`; `BadNoData` if `[start, end)` held nothing | | `DeleteAtTimeDetails` | `HistoryDeleteAtTime` | per time: `Good`; `BadNoEntryExists` | -Each backend call is one transaction (`MERGE … OUTPUT $action` against the TVP). The node-level `StatusCode` stays `Good` when item results are +Each backend call is applied as a unit (SqlServer: one transaction, `MERGE … OUTPUT $action` against the TVP; QuestDb: an INSERT per +chunk plus the WAL-visibility wait, no transaction — see [QuestDB backend](#questdb-backend)). The node-level `StatusCode` stays `Good` when item results are mixed (Part 11 reports item outcomes in `OperationResults`); it is Bad only for whole-node failures. **Per-item rejections** (answered without touching SQL; other items still proceed): a null item ⇒ `BadInvalidArgument`; diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbConnection.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbConnection.cs index 1b003425..5d965006 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbConnection.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbConnection.cs @@ -1,4 +1,6 @@ using System.Net.Sockets; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; using Npgsql; using NpgsqlTypes; @@ -12,8 +14,15 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb; /// failures with / . Every QuestDB server error carries /// SQLSTATE 00000, so classification is by message text, never by SQLSTATE. /// -internal sealed class QuestDbConnection : IAsyncDisposable +internal sealed class QuestDbConnection : IAsyncDisposable, IDisposable { + /// + /// Npgsql pool cap applied when the connection string does not set Maximum Pool Size itself. Npgsql's own + /// default (100) exceeds QuestDB's default PG-wire connection limit (64), and the server's default read + /// concurrency alone can reach 64; an explicit value in the connection string is always kept. + /// + internal const int DefaultMaxPoolSize = 24; + private static readonly long MinSafeTicks = 10; // DateTime.MinValue itself binds as -infinity private static readonly long MaxSafeTicks = DateTime.MaxValue.Ticks - DateTime.MaxValue.Ticks % 10; // MaxValue itself binds as +infinity @@ -21,14 +30,40 @@ internal sealed class QuestDbConnection : IAsyncDisposable /// Creates the connection helper; options must already be valid. /// Validated options. - public QuestDbConnection(QuestDbHistorianOptions options) + /// Optional logger; the effective pool size is logged once at Information. + public QuestDbConnection(QuestDbHistorianOptions options, ILogger? log = null) { ArgumentNullException.ThrowIfNull(options); if (!QuestDbHistorianOptions.IsValidPrefix(options.TablePrefix)) throw new ArgumentException("Invalid table prefix.", nameof(options)); TablePrefix = options.TablePrefix; CommandTimeoutSeconds = Math.Max(1, (int)Math.Ceiling(options.CommandTimeout.TotalSeconds)); - _dataSource = new NpgsqlDataSourceBuilder(options.ConnectionString).Build(); + var csb = new NpgsqlConnectionStringBuilder(options.ConnectionString); + var explicitPool = HasExplicitMaxPoolSize(options.ConnectionString); + if (!explicitPool) csb.MaxPoolSize = DefaultMaxPoolSize; + EffectiveMaxPoolSize = csb.MaxPoolSize; + (log ?? NullLogger.Instance).LogInformation( + "QuestDB historian connection pool: Maximum Pool Size = {MaxPoolSize} ({Source}).", + EffectiveMaxPoolSize, explicitPool ? "from the connection string" : "default cap"); + _dataSource = new NpgsqlDataSourceBuilder(csb.ConnectionString).Build(); + } + + /// The Npgsql Maximum Pool Size in effect (the connection string's own value, else ). + internal int EffectiveMaxPoolSize { get; } + + // Npgsql accepts "Maximum Pool Size" and "MaxPoolSize" (case-insensitive); a generic parse keeps the caller's own + // keywords so "was it set" is answered by the string itself, not by the builder's default. + private static bool HasExplicitMaxPoolSize(string connectionString) + { + var generic = new System.Data.Common.DbConnectionStringBuilder { ConnectionString = connectionString }; + foreach (string key in generic.Keys) + { + var k = key.Replace(" ", "", StringComparison.Ordinal); + if (string.Equals(k, "MaxPoolSize", StringComparison.OrdinalIgnoreCase) + || string.Equals(k, "MaximumPoolSize", StringComparison.OrdinalIgnoreCase)) + return true; + } + return false; } /// The validated table prefix. @@ -70,6 +105,9 @@ internal sealed class QuestDbConnection : IAsyncDisposable /// public ValueTask DisposeAsync() => _dataSource.DisposeAsync(); + /// + public void Dispose() => _dataSource.Dispose(); + // ---- timestamps (R6) ------------------------------------------------------------------------------------------- /// diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianProvider.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianProvider.cs index 39af8413..f56d2205 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianProvider.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianProvider.cs @@ -18,8 +18,10 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb; /// 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. +/// The provider owns the shared connection pool and disposes it (IAsyncDisposable/IDisposable); the host +/// registers the provider through a DI factory so the container disposes it at shutdown. /// -public sealed class QuestDbHistorianProvider : IHistorianBackendProvider +public sealed class QuestDbHistorianProvider : IHistorianBackendProvider, IAsyncDisposable, IDisposable { private static readonly FrozenSet Kinds = Enum.GetValues().ToFrozenSet(); @@ -27,6 +29,7 @@ public sealed class QuestDbHistorianProvider : IHistorianBackendProvider private readonly QuestDbHistorianOptions _options; private readonly object _initLock = new(); private Shared? _shared; + private bool _disposed; /// Creates the provider over the ServerHistorian:QuestDb options (validated elsewhere). /// The ServerHistorian:QuestDb options. @@ -51,7 +54,8 @@ public sealed class QuestDbHistorianProvider : IHistorianBackendProvider { if (_shared is { } raced) return raced; var loggers = Loggers(sp); - var conn = new QuestDbConnection(_options); + ObjectDisposedException.ThrowIf(_disposed, this); + var conn = new QuestDbConnection(_options, loggers.CreateLogger()); 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); @@ -101,6 +105,28 @@ public sealed class QuestDbHistorianProvider : IHistorianBackendProvider Loggers(sp).CreateLogger()); } + /// Disposes the shared connection pool, when one was built. Idempotent. + /// A task completing when the pool is disposed. + public async ValueTask DisposeAsync() + { + if (TakeForDispose() is { } conn) await conn.DisposeAsync().ConfigureAwait(false); + } + + /// Disposes the shared connection pool, when one was built. Idempotent. + public void Dispose() => TakeForDispose()?.Dispose(); + + private QuestDbConnection? TakeForDispose() + { + lock (_initLock) + { + if (_disposed) return null; + _disposed = true; + var conn = _shared?.Conn; + _shared = null; + return conn; + } + } + 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 a7fba836..c9eb37d0 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 @@ -1,3 +1,4 @@ +using System.Collections.Concurrent; using System.Text; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging.Abstractions; @@ -19,7 +20,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb; /// Batch normalisation is at µs resolution (non-finite doubles become Empty /// + Bad, an in-batch µs collision keeps the first). RC1: a sample whose time is before 1970 cannot be a /// QuestDB designated timestamp and one such row would fail the whole multi-row INSERT, so those samples are dropped -/// (Warning once per batch, counted in ) and the call still succeeds for the rest. +/// (Warning at most once per tag per ; every one counted in ) and the call still succeeds for the rest. /// R5: after inserting, the writer waits for WAL visibility so an immediate redelivery sees the rows. /// Inserts are chunked at rows (8 bound parameters per row, far under the 32,767 limit). /// @@ -28,12 +29,19 @@ internal sealed class QuestDbHistorianValueWriter : IHistorianValueWriter /// Rows per multi-row INSERT (T0 Q2: 1,000 rows is reliable). internal const int BatchRows = 1000; + /// + /// Minimum time between two pre-1970 drop Warnings for the same tag. A device with an unset clock sends such + /// times on every drain pass; still counts every dropped sample. + /// + internal static readonly TimeSpan Pre1970WarningInterval = TimeSpan.FromMinutes(10); + private readonly QuestDbConnection _conn; private readonly QuestDbTagCatalog _catalog; private readonly QuestDbWalVisibility _wal; private readonly ILogger _log; private readonly Func _utcNow; private long _droppedPre1970; + private readonly ConcurrentDictionary _pre1970WarnedUtc = new(StringComparer.Ordinal); /// Creates the writer. /// Connection helper. @@ -73,7 +81,9 @@ internal sealed class QuestDbHistorianValueWriter : IHistorianValueWriter if (dropped > 0) { Interlocked.Add(ref _droppedPre1970, dropped); - _log.LogWarning("QuestDB historian dropped {Count} sample(s) for a tag because their time is before 1970-01-01 (not storable).", dropped); + if (ShouldWarnPre1970(tag)) + _log.LogWarning("QuestDB historian dropped {Count} sample(s) for a tag because their time is before 1970-01-01 (not storable); " + + "further drops for this tag are not logged for {Minutes} minutes.", dropped, Pre1970WarningInterval.TotalMinutes); } if (rows.Count == 0) return true; @@ -98,6 +108,22 @@ internal sealed class QuestDbHistorianValueWriter : IHistorianValueWriter } } + // Once per tag per Pre1970WarningInterval, measured on the writer's clock. + private bool ShouldWarnPre1970(string tag) + { + var now = _utcNow(); + var warn = false; + _pre1970WarnedUtc.AddOrUpdate(tag, + _ => { warn = true; return now; }, + (_, last) => + { + if (now - last < Pre1970WarningInterval) { warn = false; return last; } + warn = true; + return now; + }); + return warn; + } + // 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) { diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbTagCatalog.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbTagCatalog.cs index 341ba89d..8076eded 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbTagCatalog.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbTagCatalog.cs @@ -42,15 +42,24 @@ internal sealed class QuestDbTagCatalog if (!HistorianTagNames.IsValid(name)) return false; if (_known.ContainsKey(name)) return true; await _bootstrap.EnsureAsync(ct).ConfigureAwait(false); - await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false); - await using var cmd = _conn.CreateCommand(conn, $"SELECT count() FROM {_conn.TagTable} WHERE name = $1"); - QuestDbConnection.BindText(cmd, name); - var found = Convert.ToInt64(await cmd.ExecuteScalarAsync(ct).ConfigureAwait(false)) > 0; + var found = await StoredAsync(name, ct).ConfigureAwait(false); if (found) _known[name] = 0; return found; } - /// Registers the tag when absent (no-op when present) and waits for the row to become visible. + private async Task StoredAsync(string name, CancellationToken ct) + { + await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false); + await using var cmd = _conn.CreateCommand(conn, $"SELECT count() FROM {_conn.TagTable} WHERE name = $1"); + QuestDbConnection.BindText(cmd, name); + return Convert.ToInt64(await cmd.ExecuteScalarAsync(ct).ConfigureAwait(false)) > 0; + } + + /// + /// Registers the tag when absent (no-op when present) and waits for the row to become visible. On a miss it first + /// waits for the tag table's already-sequenced commits and re-checks, so a row the provisioner has committed but + /// QuestDB has not yet applied is found rather than overwritten (its unit/description kept). + /// /// Historian tag name. /// Initial data_type (only written on create; provisioning corrects it later). /// Cancellation. @@ -62,6 +71,16 @@ internal sealed class QuestDbTagCatalog throw new ArgumentException($"Historian tag name exceeds {HistorianTagNames.MaxLength} characters.", nameof(name)); if (await ExistsAsync(name, ct).ConfigureAwait(false)) return; + // The miss may only mean another writer's row (the provisioner's, carrying eng_unit/description) is committed + // but not yet applied. Inserting now would land AFTER it and the dedup key would replace it with a row that has + // no metadata, so first wait for everything already sequenced on the tag table and look again. + await _wal.WaitAsync(_conn.TagTable, ct).ConfigureAwait(false); + if (await StoredAsync(name, ct).ConfigureAwait(false)) + { + _known[name] = 0; + return; + } + await using (var conn = await _conn.OpenAsync(ct).ConfigureAwait(false)) { await using var cmd = _conn.CreateCommand(conn, diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs index 6fa61706..416052f2 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs @@ -235,15 +235,19 @@ if (hasDriver) // Backend selection: resolve the provider once (unknown name => OptionsValidationException naming the // allowed values) and build every historian seam through it. var historianProvider = HistorianBackendRegistry.Resolve(serverHistorianOptions, builder.Configuration); - builder.Services.AddSingleton(historianProvider); + // Registered through a factory (not as an instance) so the container owns it and disposes it at shutdown + // (the QuestDb provider's shared Npgsql pool); every seam below resolves the provider from the container + // rather than capturing the local, so whichever seam builds the backend's shared state also makes the + // container track the provider for disposal. + builder.Services.AddSingleton(_ => historianProvider); // HistoryUpdate is gated like HistoryRead: None unconditionally, the real holder only when enabled. builder.Services.AddSingleton(HistorianUpdaterHolder.None); if (serverHistorianOptions.Enabled) - builder.Services.AddSingleton(sp => new HistorianUpdaterHolder(historianProvider.CreateUpdater(sp))); + builder.Services.AddSingleton(sp => new HistorianUpdaterHolder(sp.GetRequiredService().CreateUpdater(sp))); builder.Services.AddAlarmHistorian( builder.Configuration, - (_, sp) => historianProvider.CreateAlarmWriter(sp)); + (_, sp) => sp.GetRequiredService().CreateAlarmWriter(sp)); // Node-local LocalDb: caches the deployed-configuration artifact so this node can boot from // its last-known-good config when central SQL is unreachable, and optionally replicates that @@ -269,7 +273,7 @@ if (hasDriver) // node manager's HistoryRead overrides block-bridge to. Runtime owns the gating. builder.Services.AddServerHistorian( builder.Configuration, - (_, sp) => historianProvider.CreateDataSource(sp)); + (_, sp) => sp.GetRequiredService().CreateDataSource(sp)); // Config-gated historian tag provisioning. When the ServerHistorian section is enabled this registers the // selected backend's provisioner (GatewayTagProvisioner -> the gateway's EnsureTags, or the SqlServer Tag-row @@ -280,7 +284,7 @@ if (hasDriver) // the SAME ServerHistorian:Enabled flag as the read path — both use the one resolved backend. builder.Services.AddHistorianProvisioning( builder.Configuration, - (_, sp) => historianProvider.CreateProvisioning(sp)); + (_, sp) => sp.GetRequiredService().CreateProvisioning(sp)); // Continuous historization of driver (non-Galaxy) tag values. Gated on ContinuousHistorization:Enabled // AND ServerHistorian:Enabled: the recorder drains driver-tag live values through the selected backend's @@ -319,7 +323,7 @@ if (hasDriver) // The backend's live-value writer. On Gateway it opens its OWN gRPC channel to the same gateway (a second // channel to a co-located sidecar is cheap — the gateway pools the historian sessions server-side); on // SqlServer it shares the provider's connection helper, schema bootstrapper and tag catalog. - builder.Services.AddSingleton(sp => historianProvider.CreateValueWriter(sp)); + builder.Services.AddSingleton(sp => sp.GetRequiredService().CreateValueWriter(sp)); } // Bind every cross-platform driver factory before AddAkka resolves IDriverFactory — replaces diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/ProvisioningTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/ProvisioningTests.cs index e1b8605e..68ce7c68 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/ProvisioningTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/ProvisioningTests.cs @@ -92,6 +92,39 @@ public sealed class ProvisioningTests(QuestDbFixture fx) (await TagAsync(c, "k")).ShouldBe(((byte)DriverDataType.Float64, "kg", "mass")); } + /// + /// Final review M-1: the catalog's first sighting of a tag can run while the provisioner's row is committed but + /// not yet applied (its WAL-visibility window). An unconditional catalog INSERT then lands after it and the + /// dedup key replaces the provisioner's row with one carrying no unit/description. The provisioner here does + /// not wait (VisibilityTimeout 0) and has its own catalog, so every catalog Ensure runs inside that window. + /// + [Fact] + public async Task A_catalog_Ensure_inside_the_provisioners_visibility_window_keeps_its_metadata() + { + fx.SkipIfUnavailable(); + var o = fx.Options(fx.UniquePrefix()); + await using var c = new QuestDbConnection(o); + var b = new QuestDbSchemaBootstrapper(c, o); + await b.EnsureAsync(CancellationToken.None); + var noWait = new QuestDbWalVisibility(c, TimeSpan.Zero); + var provisioner = new QuestDbTagProvisioner(c, b, new QuestDbTagCatalog(c, b, noWait), noWait); + var catalog = new QuestDbTagCatalog(c, b, new QuestDbWalVisibility(c, o.VisibilityTimeout)); + + const int Tags = 25; + for (var i = 0; i < Tags; i++) + { + var name = $"race{i}"; + (await provisioner.EnsureTagsAsync([new(name, DriverDataType.Float64, "kg", $"mass {i}")], CancellationToken.None)) + .Ensured.ShouldBe(1); + await catalog.EnsureAsync(name, 1, CancellationToken.None); // the recorder's first value for the tag + } + + await new QuestDbWalVisibility(c, o.VisibilityTimeout).WaitAsync(c.TagTable, CancellationToken.None); + for (var i = 0; i < Tags; i++) + (await TagAsync(c, $"race{i}")).ShouldBe(((byte)DriverDataType.Float64, "kg", $"mass {i}"), $"race{i}"); + (await CountAsync(c)).ShouldBe(Tags); + } + [Fact] public async Task A_connection_failure_is_a_Failed_tally_not_an_exception() { diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests/QuestDbConnectionTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests/QuestDbConnectionTests.cs index 0f39bf6e..afc09155 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests/QuestDbConnectionTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests/QuestDbConnectionTests.cs @@ -137,4 +137,48 @@ public sealed class QuestDbConnectionTests c.EventTable.ShouldBe("p_event"); c.TagTable.ShouldBe("p_tag"); } + + // ---- pool size (final review I-1) ------------------------------------------------------------------------------- + + private static int PoolOf(string cs) + { + using var c = new QuestDbConnection(new QuestDbHistorianOptions { ConnectionString = cs }); + return c.EffectiveMaxPoolSize; + } + + [Fact] + public void An_unset_Maximum_Pool_Size_is_capped_at_24() => + PoolOf("Host=127.0.0.1;Port=1").ShouldBe(24); + + [Fact] + public void The_default_cap_constant_is_24() => QuestDbConnection.DefaultMaxPoolSize.ShouldBe(24); + + [Theory] + [InlineData("Host=127.0.0.1;Port=1;Maximum Pool Size=10", 10)] + [InlineData("Host=127.0.0.1;Port=1;Maximum Pool Size=80", 80)] + [InlineData("Host=127.0.0.1;Port=1;maximum pool size=30", 30)] + [InlineData("Host=127.0.0.1;Port=1;MaxPoolSize=80", 80)] + public void An_explicit_Maximum_Pool_Size_is_preserved(string cs, int expected) => + PoolOf(cs).ShouldBe(expected); + + [Fact] + public void The_effective_pool_size_is_logged_once_at_Information() + { + var log = new InfoLogger(); + using var c = new QuestDbConnection(new QuestDbHistorianOptions { ConnectionString = "Host=127.0.0.1;Port=1" }, log); + log.Messages.Count.ShouldBe(1); + log.Messages[0].ShouldContain("24"); + } + + private sealed class InfoLogger : Microsoft.Extensions.Logging.ILogger + { + public List Messages { get; } = []; + public IDisposable? BeginScope(TState state) where TState : notnull => null; + public bool IsEnabled(Microsoft.Extensions.Logging.LogLevel logLevel) => true; + public void Log(Microsoft.Extensions.Logging.LogLevel logLevel, Microsoft.Extensions.Logging.EventId eventId, TState state, + Exception? exception, Func formatter) + { + if (logLevel == Microsoft.Extensions.Logging.LogLevel.Information) Messages.Add(formatter(state, exception)); + } + } } diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests/QuestDbHistorianProviderTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests/QuestDbHistorianProviderTests.cs index b5df992b..8f3b4d5d 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests/QuestDbHistorianProviderTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests/QuestDbHistorianProviderTests.cs @@ -76,6 +76,23 @@ public sealed class QuestDbHistorianProviderTests Field(w, "_wal").ShouldBeSameAs(Field(alarm, "_wal")); } + [Fact] + public async Task Disposing_the_provider_disposes_the_shared_pool_and_is_idempotent() + { + var sp = new NullSp(); + var p = Make(); + var conn = (QuestDbConnection)Field(p.CreateDataSource(sp), "_conn")!; + var pool = (Npgsql.NpgsqlDataSource)Field(conn, "_dataSource")!; + await p.DisposeAsync(); + Should.Throw(() => pool.CreateConnection().Open()); + await p.DisposeAsync(); + p.Dispose(); + Should.Throw(() => p.CreateDataSource(sp)); + } + + [Fact] + public async Task Disposing_an_unused_provider_is_a_no_op() => await Make().DisposeAsync(); + [Fact] public void The_constructor_rejects_null_options() => Should.Throw(() => new QuestDbHistorianProvider(null!)); diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests/QuestDbHistorianValueWriterTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests/QuestDbHistorianValueWriterTests.cs new file mode 100644 index 00000000..6650188e --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests/QuestDbHistorianValueWriterTests.cs @@ -0,0 +1,71 @@ +using Microsoft.Extensions.Logging; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests; + +/// +/// The pre-1970 drop Warning is rate-limited per tag (final review M-3) while the counter still counts every dropped +/// sample. All-pre-1970 batches return before any QuestDB call, so the connection points at a closed port. +/// +public sealed class QuestDbHistorianValueWriterTests +{ + private static readonly DateTime FileTimeZero = new(1601, 1, 1, 0, 0, 0, DateTimeKind.Utc); + + private static (QuestDbHistorianValueWriter W, ListLogger Log, Action Advance) Make() + { + var opts = new QuestDbHistorianOptions { ConnectionString = "Host=127.0.0.1;Port=1;Timeout=1" }; + var conn = new QuestDbConnection(opts); + var boot = new QuestDbSchemaBootstrapper(conn, opts); + var wal = new QuestDbWalVisibility(conn, opts.VisibilityTimeout); + var now = new DateTime(2026, 10, 11, 12, 0, 0, DateTimeKind.Utc); + var log = new ListLogger(); + var w = new QuestDbHistorianValueWriter(conn, new QuestDbTagCatalog(conn, boot, wal), wal, log, () => now); + return (w, log, d => now += d); + } + + private static HistorizationValue[] Pre1970(int n) => + Enumerable.Range(0, n).Select(i => new HistorizationValue(FileTimeZero.AddSeconds(i), HistorianValue.FromDouble(i), 0)).ToArray(); + + [Fact] + public async Task The_pre_1970_warning_is_logged_once_per_tag_per_interval_but_every_drop_is_counted() + { + var (w, log, advance) = Make(); + + (await w.WriteLiveValuesAsync("a", Pre1970(2), CancellationToken.None)).ShouldBeTrue(); + advance(TimeSpan.FromSeconds(2)); + (await w.WriteLiveValuesAsync("a", Pre1970(3), CancellationToken.None)).ShouldBeTrue(); + advance(QuestDbHistorianValueWriter.Pre1970WarningInterval - TimeSpan.FromSeconds(3)); + (await w.WriteLiveValuesAsync("a", Pre1970(1), CancellationToken.None)).ShouldBeTrue(); + + log.Warnings.ShouldBe(1); + w.DroppedPre1970.ShouldBe(6); + + // A different tag has its own budget. + (await w.WriteLiveValuesAsync("b", Pre1970(1), CancellationToken.None)).ShouldBeTrue(); + log.Warnings.ShouldBe(2); + + // Once the interval has passed since the last Warning for "a", it warns again. + advance(TimeSpan.FromSeconds(1)); + (await w.WriteLiveValuesAsync("a", Pre1970(1), CancellationToken.None)).ShouldBeTrue(); + log.Warnings.ShouldBe(3); + w.DroppedPre1970.ShouldBe(8); + } + + [Fact] + public void The_interval_is_ten_minutes() => + QuestDbHistorianValueWriter.Pre1970WarningInterval.ShouldBe(TimeSpan.FromMinutes(10)); + + private sealed class ListLogger : ILogger + { + private int _warnings; + public int Warnings => Volatile.Read(ref _warnings); + public IDisposable? BeginScope(TState state) where TState : notnull => null; + public bool IsEnabled(LogLevel logLevel) => true; + public void Log(LogLevel logLevel, EventId eventId, TState state, Exception? exception, Func formatter) + { + if (logLevel == LogLevel.Warning) Interlocked.Increment(ref _warnings); + } + } +}