Merge qdb/final: final-review fix wave (pool cap, catalog metadata, warn rate-limit, disposal, docs)

Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
2026-10-11 03:37:55 -04:00
parent e177a50046 8d82d049e9
commit 2ebe0a4226
11 changed files with 309 additions and 30 deletions
@@ -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)

@@ -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`;
@@ -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,
@@ -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
@@ -92,6 +92,39 @@ public sealed class ProvisioningTests(QuestDbFixture fx)
        (await TagAsync(c, "k")).ShouldBe(((byte)DriverDataType.Float64, "kg", "mass"));
    }

+    /// <summary>
+    ///     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.
+    /// </summary>
+    [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()
    {
@@ -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<string> Messages { get; } = [];
+        public IDisposable? BeginScope<TState>(TState state) where TState : notnull => null;
+        public bool IsEnabled(Microsoft.Extensions.Logging.LogLevel logLevel) => true;
+        public void Log<TState>(Microsoft.Extensions.Logging.LogLevel logLevel, Microsoft.Extensions.Logging.EventId eventId, TState state,
+            Exception? exception, Func<TState, Exception?, string> formatter)
+        {
+            if (logLevel == Microsoft.Extensions.Logging.LogLevel.Information) Messages.Add(formatter(state, exception));
+        }
+    }
 }
@@ -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<ObjectDisposedException>(() => pool.CreateConnection().Open());
+        await p.DisposeAsync();
+        p.Dispose();
+        Should.Throw<ObjectDisposedException>(() => 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<ArgumentNullException>(() => new QuestDbHistorianProvider(null!));
@@ -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;
+
+/// <summary>
+///     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.
+/// </summary>
+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<TimeSpan> 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>(TState state) where TState : notnull => null;
+        public bool IsEnabled(LogLevel logLevel) => true;
+        public void Log<TState>(LogLevel logLevel, EventId eventId, TState state, Exception? exception, Func<TState, Exception?, string> formatter)
+        {
+            if (logLevel == LogLevel.Warning) Interlocked.Increment(ref _warnings);
+        }
+    }
+}