feat(historian): QuestDB backend skeleton — options, connection, schema bootstrap with TTL

Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
2026-10-11 01:52:57 -04:00
parent ca2c20f852
commit 11b703708c
18 changed files with 1457 additions and 0 deletions
@@ -0,0 +1,143 @@
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Logging.Abstractions;
+using Npgsql;
+using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb;
+
+/// <summary>
+///     Lazily creates the three tables once per instance (<c>CREATE TABLE IF NOT EXISTS</c>), then reconciles the TTL
+///     of the sample and event tables with <see cref="QuestDbHistorianOptions.RetentionDays"/>.
+///     <list type="bullet">
+///     <item><b>Concurrency:</b> two nodes starting together are safe. If a CREATE still fails with a server error
+///     whose message says <c>already exists</c>, <c>tables()</c> is re-checked and the error is swallowed only when the
+///     table is really there; anything else is rethrown (SQLSTATE is always <c>00000</c>, so it cannot classify).</item>
+///     <item><b>TTL (RC3):</b> <c>tables()</c> normalises units (7 days reads back as <c>1 WEEK</c>) so values are compared
+///     in hours, and an ALTER becomes visible only after the WAL applies it, so the bootstrapper never re-reads to verify
+///     its own ALTER. A boot whose TTL already matches issues no ALTER.</item>
+///     <item><b>Failure:</b> not latched. Within <see cref="BootstrapFailureCache.DefaultDuration"/> the last failure is
+///     re-thrown without touching QuestDB.</item>
+///     <item><c>AutoCreateSchema=false</c> only verifies the tables exist (no TTL changes) and throws
+///     <see cref="HistorianSchemaMissingException"/> otherwise.</item>
+///     </list>
+/// </summary>
+internal sealed class QuestDbSchemaBootstrapper
+{
+    private readonly QuestDbConnection _conn;
+    private readonly QuestDbHistorianOptions _options;
+    private readonly ILogger _log;
+    private readonly SemaphoreSlim _gate = new(1, 1);
+    private readonly BootstrapFailureCache _failure; // guarded by _gate
+    private volatile bool _ready;
+
+    /// <summary>Creates the bootstrapper.</summary>
+    /// <param name="conn">Connection helper.</param>
+    /// <param name="options">Validated options.</param>
+    /// <param name="log">Optional logger.</param>
+    /// <param name="timeProvider">Clock for the failure cache; system by default.</param>
+    public QuestDbSchemaBootstrapper(QuestDbConnection conn, QuestDbHistorianOptions options, ILogger? log = null,
+        TimeProvider? timeProvider = null)
+    {
+        _conn = conn;
+        _options = options;
+        _log = log ?? NullLogger.Instance;
+        _failure = new BootstrapFailureCache(timeProvider ?? TimeProvider.System, BootstrapFailureCache.DefaultDuration);
+    }
+
+    /// <summary>Ensures the schema is ready; no-op after the first success.</summary>
+    /// <param name="ct">Cancellation.</param>
+    /// <returns>A task completing when the tables are usable.</returns>
+    /// <exception cref="HistorianSchemaMissingException">AutoCreateSchema is false and a table is absent.</exception>
+    public async Task EnsureAsync(CancellationToken ct)
+    {
+        if (_ready) return;
+        await _gate.WaitAsync(ct).ConfigureAwait(false);
+        try
+        {
+            if (_ready) return;
+            _failure.ThrowIfCached();
+            try
+            {
+                await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false);
+                if (_options.AutoCreateSchema)
+                {
+                    await CreateTablesAsync(conn, ct).ConfigureAwait(false);
+                    await ReconcileTtlAsync(conn, ct).ConfigureAwait(false);
+                }
+                else
+                {
+                    await VerifyAsync(conn, ct).ConfigureAwait(false);
+                }
+                _ready = true;
+                _failure.Clear();
+            }
+            catch (Exception ex) when (ex is not OperationCanceledException || !ct.IsCancellationRequested)
+            {
+                _failure.Record(ex);
+                _log.LogWarning("QuestDB historian schema bootstrap (prefix '{Prefix}') failed ({Exception}: {Message}); " +
+                    "historian calls fail fast for {Seconds} s before retrying.",
+                    _conn.TablePrefix, ex.GetType().Name, ex.Message, BootstrapFailureCache.DefaultDuration.TotalSeconds);
+                throw;
+            }
+        }
+        finally
+        {
+            _gate.Release();
+        }
+    }
+
+    private async Task VerifyAsync(NpgsqlConnection conn, CancellationToken ct)
+    {
+        foreach (var table in QuestDbTables.All(_conn.TablePrefix))
+        {
+            if (!await _conn.TableExistsAsync(conn, table, ct).ConfigureAwait(false))
+                throw new HistorianSchemaMissingException(
+                    $"QuestDB historian table '{table}' is missing and AutoCreateSchema is false. " +
+                    "Create the tables (see docs/Historian.md) or enable AutoCreateSchema.");
+        }
+    }
+
+    private async Task CreateTablesAsync(NpgsqlConnection conn, CancellationToken ct)
+    {
+        var statements = QuestDbTables.CreateAll(_conn.TablePrefix, _options.RetentionDays);
+        var tables = QuestDbTables.All(_conn.TablePrefix);
+        for (var i = 0; i < statements.Length; i++)
+        {
+            try
+            {
+                await using var cmd = _conn.CreateCommand(conn, statements[i]);
+                await cmd.ExecuteNonQueryAsync(ct).ConfigureAwait(false);
+            }
+            catch (PostgresException ex) when (QuestDbConnection.IsAlreadyExists(ex))
+            {
+                // A concurrent creator won the race - swallow ONLY if the table is really there.
+                if (!await _conn.TableExistsAsync(conn, tables[i], ct).ConfigureAwait(false)) throw;
+                _log.LogDebug("QuestDB table {Table} was created concurrently.", tables[i]);
+            }
+        }
+    }
+
+    private async Task ReconcileTtlAsync(NpgsqlConnection conn, CancellationToken ct)
+    {
+        var wantHours = _options.RetentionDays is { } d ? d * 24L : 0L;
+        foreach (var table in QuestDbTables.Expiring(_conn.TablePrefix))
+        {
+            long haveHours;
+            await using (var cmd = _conn.CreateCommand(conn, "SELECT ttlValue, ttlUnit FROM tables() WHERE table_name = $1"))
+            {
+                QuestDbConnection.BindText(cmd, table);
+                await using var r = await cmd.ExecuteReaderAsync(ct).ConfigureAwait(false);
+                if (!await r.ReadAsync(ct).ConfigureAwait(false)) continue; // vanished (dropped between create and here)
+                var value = r.IsDBNull(0) ? 0L : Convert.ToInt64(r.GetValue(0));
+                var unit = r.IsDBNull(1) ? null : r.GetString(1);
+                haveHours = QuestDbTables.TtlHours(value, unit);
+            }
+            if (haveHours == wantHours) continue;
+
+            // The ALTER is a WAL transaction: tables() keeps the old value until it applies, so do NOT re-read to verify.
+            await using (var alter = _conn.CreateCommand(conn, QuestDbTables.AlterTtl(table, _options.RetentionDays)))
+                await alter.ExecuteNonQueryAsync(ct).ConfigureAwait(false);
+            _log.LogInformation("Applied TTL {Days} day(s) to QuestDB table {Table}.", _options.RetentionDays, table);
+        }
+    }
+}