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
This commit is contained in:
Joseph Doherty
2026-10-11 03:35:35 -04:00
parent e177a50046
commit 8d82d049e9
11 changed files with 309 additions and 30 deletions
@@ -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 <see cref="IsTransient"/> / <see cref="IsAlreadyExists"/>. Every QuestDB server error carries
/// SQLSTATE <c>00000</c>, so classification is by message text, never by SQLSTATE.</para>
/// </summary>
internal sealed class QuestDbConnection : IAsyncDisposable
internal sealed class QuestDbConnection : IAsyncDisposable, IDisposable
{
/// <summary>
/// Npgsql pool cap applied when the connection string does not set <c>Maximum Pool Size</c> 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.
/// </summary>
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
/// <summary>Creates the connection helper; options must already be valid.</summary>
/// <param name="options">Validated options.</param>
public QuestDbConnection(QuestDbHistorianOptions options)
/// <param name="log">Optional logger; the effective pool size is logged once at Information.</param>
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();
}
/// <summary>The Npgsql <c>Maximum Pool Size</c> in effect (the connection string's own value, else <see cref="DefaultMaxPoolSize"/>).</summary>
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;
}
/// <summary>The validated table prefix.</summary>
@@ -70,6 +105,9 @@ internal sealed class QuestDbConnection : IAsyncDisposable
/// <inheritdoc/>
public ValueTask DisposeAsync() => _dataSource.DisposeAsync();
/// <inheritdoc/>
public void Dispose() => _dataSource.Dispose();
// ---- timestamps (R6) -------------------------------------------------------------------------------------------
/// <summary>
@@ -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 <c>ServerHistorianOptionsValidator</c>). The first
/// <c>Create*</c> call's service provider supplies the logger factory the shared pieces log through.
/// <para>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.</para>
/// </remarks>
public sealed class QuestDbHistorianProvider : IHistorianBackendProvider
public sealed class QuestDbHistorianProvider : IHistorianBackendProvider, IAsyncDisposable, IDisposable
{
private static readonly FrozenSet<HistorianValueKind> Kinds =
Enum.GetValues<HistorianValueKind>().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;
/// <summary>Creates the provider over the <c>ServerHistorian:QuestDb</c> options (validated elsewhere).</summary>
/// <param name="options">The <c>ServerHistorian:QuestDb</c> options.</param>
@@ -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<QuestDbConnection>());
var bootstrap = new QuestDbSchemaBootstrapper(conn, _options, loggers.CreateLogger<QuestDbSchemaBootstrapper>());
var wal = new QuestDbWalVisibility(conn, _options.VisibilityTimeout, loggers.CreateLogger<QuestDbWalVisibility>());
var shared = new Shared(conn, bootstrap, new QuestDbTagCatalog(conn, bootstrap, wal), wal);
@@ -101,6 +105,28 @@ public sealed class QuestDbHistorianProvider : IHistorianBackendProvider
Loggers(sp).CreateLogger<QuestDbHistorianDataUpdater>());
}
/// <summary>Disposes the shared connection pool, when one was built. Idempotent.</summary>
/// <returns>A task completing when the pool is disposed.</returns>
public async ValueTask DisposeAsync()
{
if (TakeForDispose() is { } conn) await conn.DisposeAsync().ConfigureAwait(false);
}
/// <summary>Disposes the shared connection pool, when one was built. Idempotent.</summary>
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);
}
@@ -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 <see cref="LiveWritePreparer.Prepare"/> at µs resolution (non-finite doubles become Empty
/// + Bad, an in-batch µs collision keeps the first). <b>RC1:</b> 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 <see cref="DroppedPre1970"/>) and the call still succeeds for the rest.
/// (Warning at most once per tag per <see cref="Pre1970WarningInterval"/>; every one counted in <see cref="DroppedPre1970"/>) and the call still succeeds for the rest.
/// <b>R5:</b> after inserting, the writer waits for WAL visibility so an immediate redelivery sees the rows.
/// Inserts are chunked at <see cref="BatchRows"/> rows (8 bound parameters per row, far under the 32,767 limit).
/// </remarks>
@@ -28,12 +29,19 @@ internal sealed class QuestDbHistorianValueWriter : IHistorianValueWriter
/// <summary>Rows per multi-row INSERT (T0 Q2: 1,000 rows is reliable).</summary>
internal const int BatchRows = 1000;
/// <summary>
/// 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; <see cref="DroppedPre1970"/> still counts every dropped sample.
/// </summary>
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<DateTime> _utcNow;
private long _droppedPre1970;
private readonly ConcurrentDictionary<string, DateTime> _pre1970WarnedUtc = new(StringComparer.Ordinal);
/// <summary>Creates the writer.</summary>
/// <param name="conn">Connection helper.</param>
@@ -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<HashSet<long>> ExistingTicksAsync(string tag, List<HistorianSampleRow> rows, CancellationToken ct)
{
@@ -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;
}
/// <summary>Registers the tag when absent (no-op when present) and waits for the row to become visible.</summary>
private async Task<bool> 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;
}
/// <summary>
/// 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).
/// </summary>
/// <param name="name">Historian tag name.</param>
/// <param name="dataType">Initial <c>data_type</c> (only written on create; provisioning corrects it later).</param>
/// <param name="ct">Cancellation.</param>
@@ -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,
+10 -6
View File
@@ -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<IHistorianBackendProvider>(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<IHistorianBackendProvider>(_ => 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<IHistorianBackendProvider>().CreateUpdater(sp)));
builder.Services.AddAlarmHistorian(
builder.Configuration,
(_, sp) => historianProvider.CreateAlarmWriter(sp));
(_, sp) => sp.GetRequiredService<IHistorianBackendProvider>().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<IHistorianBackendProvider>().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<IHistorianBackendProvider>().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<IHistorianValueWriter>(sp => historianProvider.CreateValueWriter(sp));
builder.Services.AddSingleton<IHistorianValueWriter>(sp => sp.GetRequiredService<IHistorianBackendProvider>().CreateValueWriter(sp));
}
// Bind every cross-platform driver factory before AddAkka resolves IDriverFactory — replaces