feat(historian): QuestDB live writes, provisioning and alarm events

Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
2026-10-11 02:08:04 -04:00
parent ecc0f914cc
commit 4c053cc4d0
6 changed files with 854 additions and 0 deletions
@@ -0,0 +1,130 @@
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Logging.Abstractions;
+using Npgsql;
+using NpgsqlTypes;
+using ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian;
+using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb;
+
+/// <summary>
+///     <see cref="IAlarmHistorianWriter"/> over the QuestDB event table. Each event carries the deterministic
+///     <see cref="AlarmEventIds.ComputeEventId"/> and is inserted through the table's <c>DEDUP UPSERT KEYS(ts, event_id)</c>
+///     (T0 Q3 confirmed VARCHAR works as a key), so the store-and-forward drain's at-least-once redelivery, and both
+///     pair nodes writing the same event, converge on one row (the redelivered row is identical except
+///     <c>received_utc</c>). Outcomes: Ack on insert; RetryPlease on transient / unclassified errors and on
+///     cancellation; PermanentFail only for data the server will always reject (a time before 1970, which is not a
+///     storable designated timestamp, or a server message naming an inconvertible / out-of-range value). Never throws.
+/// </summary>
+internal sealed class QuestDbAlarmHistorianWriter : IAlarmHistorianWriter
+{
+    private static readonly string[] PermanentMessageFragments =
+        ["designated timestamp", "inconvertible", "out of range", "too long", "cannot be null"];
+
+    private readonly QuestDbConnection _conn;
+    private readonly QuestDbSchemaBootstrapper _bootstrap;
+    private readonly QuestDbWalVisibility _wal;
+    private readonly ILogger _log;
+    private readonly TimeProvider _clock;
+
+    /// <summary>Creates the writer.</summary>
+    /// <param name="conn">Connection helper.</param>
+    /// <param name="bootstrap">Schema bootstrapper.</param>
+    /// <param name="wal">WAL visibility wait (once per batch, so a following read sees the events).</param>
+    /// <param name="log">Optional logger.</param>
+    /// <param name="clock">Clock for <c>received_utc</c>.</param>
+    public QuestDbAlarmHistorianWriter(QuestDbConnection conn, QuestDbSchemaBootstrapper bootstrap, QuestDbWalVisibility wal,
+        ILogger? log = null, TimeProvider? clock = null)
+    {
+        _conn = conn;
+        _bootstrap = bootstrap;
+        _wal = wal;
+        _log = log ?? NullLogger.Instance;
+        _clock = clock ?? TimeProvider.System;
+    }
+
+    /// <inheritdoc/>
+    public async Task<IReadOnlyList<HistorianWriteOutcome>> WriteBatchAsync(IReadOnlyList<AlarmHistorianEvent> batch, CancellationToken cancellationToken)
+    {
+        ArgumentNullException.ThrowIfNull(batch);
+        if (batch.Count == 0) return [];
+        var outcomes = new HistorianWriteOutcome[batch.Count];
+        try
+        {
+            await _bootstrap.EnsureAsync(cancellationToken).ConfigureAwait(false);
+        }
+        catch (Exception ex)
+        {
+            _log.LogWarning("QuestDB alarm historian bootstrap failed ({Exception}); will retry.", ex.GetType().Name);
+            Array.Fill(outcomes, HistorianWriteOutcome.RetryPlease);
+            return outcomes;
+        }
+
+        var wrote = false;
+        for (var i = 0; i < batch.Count; i++)
+        {
+            if (cancellationToken.IsCancellationRequested) { outcomes[i] = HistorianWriteOutcome.RetryPlease; continue; }
+            outcomes[i] = await WriteOneAsync(batch[i], cancellationToken).ConfigureAwait(false);
+            wrote |= outcomes[i] == HistorianWriteOutcome.Ack;
+        }
+
+        if (wrote)
+        {
+            try { await _wal.WaitAsync(_conn.EventTable, cancellationToken).ConfigureAwait(false); }
+            catch (Exception ex) { _log.LogDebug("QuestDB alarm visibility wait failed ({Exception}).", ex.GetType().Name); }
+        }
+        return outcomes;
+    }
+
+    private async Task<HistorianWriteOutcome> WriteOneAsync(AlarmHistorianEvent e, CancellationToken ct)
+    {
+        try
+        {
+            if (!QuestDbConnection.IsStorable(e.TimestampUtc))
+            {
+                _log.LogWarning("QuestDB alarm event dropped: its time is before 1970-01-01 and cannot be stored.");
+                return HistorianWriteOutcome.PermanentFail;
+            }
+
+            await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false);
+            await using var cmd = _conn.CreateCommand(conn,
+                $"INSERT INTO {_conn.EventTable} (ts, event_id, source, alarm_id, alarm_name, alarm_type_name, severity, event_kind, message, user_name, comment, received_utc) " +
+                "VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12)");
+            QuestDbConnection.BindTimestamp(cmd, e.TimestampUtc);
+            QuestDbConnection.BindText(cmd, AlarmEventIds.ComputeEventId(e));
+            QuestDbConnection.BindText(cmd, e.EquipmentPath ?? "");
+            QuestDbConnection.BindText(cmd, e.AlarmId ?? "");
+            QuestDbConnection.BindText(cmd, e.AlarmName ?? "");
+            QuestDbConnection.BindText(cmd, e.AlarmTypeName ?? "");
+            cmd.Parameters.Add(new NpgsqlParameter { NpgsqlDbType = NpgsqlDbType.Integer, Value = AlarmEventIds.ToNumericSeverity(e.Severity) });
+            QuestDbConnection.BindText(cmd, e.EventKind ?? "");
+            QuestDbConnection.BindText(cmd, e.Message);
+            QuestDbConnection.BindText(cmd, e.User);
+            QuestDbConnection.BindText(cmd, e.Comment);
+            QuestDbConnection.BindTimestamp(cmd, _clock.GetUtcNow().UtcDateTime);
+            await cmd.ExecuteNonQueryAsync(ct).ConfigureAwait(false);
+            return HistorianWriteOutcome.Ack;
+        }
+        catch (OperationCanceledException) when (ct.IsCancellationRequested)
+        {
+            return HistorianWriteOutcome.RetryPlease;
+        }
+        catch (Exception ex)
+        {
+            var outcome = Classify(ex);
+            _log.LogWarning("QuestDB alarm event write failed ({Exception}); outcome {Outcome}.", ex.GetType().Name, outcome);
+            return outcome;
+        }
+    }
+
+    // Only a data error QuestDB will reject forever dead-letters; anything unknown or non-database retries (never lose alarm
+    // history). SQLSTATE is always 00000 on QuestDB, so a server rejection is classified by message text.
+    internal static HistorianWriteOutcome Classify(Exception ex)
+    {
+        if (QuestDbConnection.IsTransient(ex)) return HistorianWriteOutcome.RetryPlease;
+        if (ex is PostgresException pg &&
+            PermanentMessageFragments.Any(f => pg.MessageText.Contains(f, StringComparison.OrdinalIgnoreCase)))
+            return HistorianWriteOutcome.PermanentFail;
+        return HistorianWriteOutcome.RetryPlease;
+    }
+}
@@ -0,0 +1,142 @@
+using System.Text;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Logging.Abstractions;
+using Npgsql;
+using NpgsqlTypes;
+using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb;
+
+/// <summary>
+///     <see cref="IHistorianValueWriter"/> over the QuestDB sample table. Redelivery-idempotent with first-wins
+///     semantics: QuestDB's <c>DEDUP UPSERT KEYS(ts, tag)</c> would let a re-sent sample overwrite the stored one (and a
+///     re-sent live sample resurrect a tombstone), so the writer reads which timestamps already exist for the tag in
+///     the batch's range (tombstones count as present, T0 Q11 range form) and inserts only the absent rows.
+///     Never throws; returns false on any failure so the recorder keeps its entries.
+/// </summary>
+/// <remarks>
+///     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.
+///     <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>
+internal sealed class QuestDbHistorianValueWriter : IHistorianValueWriter
+{
+    /// <summary>Rows per multi-row INSERT (T0 Q2: 1,000 rows is reliable).</summary>
+    internal const int BatchRows = 1000;
+
+    private readonly QuestDbConnection _conn;
+    private readonly QuestDbTagCatalog _catalog;
+    private readonly QuestDbWalVisibility _wal;
+    private readonly ILogger _log;
+    private readonly Func<DateTime> _utcNow;
+    private long _droppedPre1970;
+
+    /// <summary>Creates the writer.</summary>
+    /// <param name="conn">Connection helper.</param>
+    /// <param name="catalog">Tag catalog (the only tag authority).</param>
+    /// <param name="wal">WAL visibility wait.</param>
+    /// <param name="log">Optional logger.</param>
+    /// <param name="utcNow">Clock for null timestamps and <c>server_ts</c> (tests).</param>
+    public QuestDbHistorianValueWriter(QuestDbConnection conn, QuestDbTagCatalog catalog, QuestDbWalVisibility wal,
+        ILogger? log = null, Func<DateTime>? utcNow = null)
+    {
+        _conn = conn;
+        _catalog = catalog;
+        _wal = wal;
+        _log = log ?? NullLogger.Instance;
+        _utcNow = utcNow ?? (() => DateTime.UtcNow);
+    }
+
+    /// <summary>Samples dropped so far because their time is before 1970 (RC1).</summary>
+    internal long DroppedPre1970 => Interlocked.Read(ref _droppedPre1970);
+
+    /// <inheritdoc/>
+    public async Task<bool> WriteLiveValuesAsync(string tag, IReadOnlyList<HistorizationValue> values, CancellationToken ct)
+    {
+        try
+        {
+            if (!HistorianTagNames.IsValid(tag))
+            {
+                _log.LogWarning("QuestDB historian live-value write refused: tag name is missing or longer than {Max} characters.",
+                    HistorianTagNames.MaxLength);
+                return false;
+            }
+            if (values is null || values.Count == 0) return true;
+
+            var prepared = LiveWritePreparer.Prepare(values, _utcNow, resolutionTicks: 10);
+            var rows = prepared.Where(r => QuestDbConnection.IsStorable(r.TimeUtc)).ToList();
+            var dropped = prepared.Count - rows.Count;
+            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 (rows.Count == 0) return true;
+
+            var dataType = HistorianDataTypes.FromKind(rows.Select(r => r.Value.Kind).FirstOrDefault(k => k != HistorianValueKind.Empty));
+            await _catalog.EnsureAsync(tag, dataType, ct).ConfigureAwait(false);
+
+            var present = await ExistingTicksAsync(tag, rows, ct).ConfigureAwait(false);
+            var fresh = rows.Where(r => !present.Contains(QuestDbConnection.Round(r.TimeUtc).Ticks)).ToList();
+            if (fresh.Count == 0) return true;
+
+            var serverTs = _utcNow();
+            for (var i = 0; i < fresh.Count; i += BatchRows)
+                await InsertAsync(tag, fresh.GetRange(i, Math.Min(BatchRows, fresh.Count - i)), serverTs, ct).ConfigureAwait(false);
+
+            await _wal.WaitAsync(_conn.SampleTable, ct).ConfigureAwait(false); // timeout/suspension does not change the result (R5)
+            return true;
+        }
+        catch (Exception ex)
+        {
+            _log.LogWarning("QuestDB historian live-value write failed ({Exception}); will retry.", ex.GetType().Name);
+            return false;
+        }
+    }
+
+    // T0 Q11 range form: 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)
+    {
+        var min = rows.Min(r => QuestDbConnection.Round(r.TimeUtc));
+        var max = rows.Max(r => QuestDbConnection.Round(r.TimeUtc));
+        var set = new HashSet<long>();
+        await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false);
+        await using var cmd = _conn.CreateCommand(conn, $"SELECT ts FROM {_conn.SampleTable} WHERE tag = $1 AND ts >= $2 AND ts <= $3");
+        QuestDbConnection.BindText(cmd, tag);
+        QuestDbConnection.BindTimestamp(cmd, min);
+        QuestDbConnection.BindTimestamp(cmd, max);
+        await using var r = await cmd.ExecuteReaderAsync(ct).ConfigureAwait(false);
+        while (await r.ReadAsync(ct).ConfigureAwait(false)) set.Add(QuestDbConnection.ReadUtc(r, 0).Ticks);
+        return set;
+    }
+
+    private async Task InsertAsync(string tag, List<HistorianSampleRow> chunk, DateTime serverTs, CancellationToken ct)
+    {
+        var sql = new StringBuilder($"INSERT INTO {_conn.SampleTable} (ts, tag, kind, num_value, int_value, str_value, status, deleted, server_ts) VALUES ");
+        await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false);
+        await using var cmd = _conn.CreateCommand(conn, "");
+        var n = 1;
+        for (var i = 0; i < chunk.Count; i++)
+        {
+            if (i > 0) sql.Append(',');
+            sql.Append($"(${n},${n + 1},${n + 2},${n + 3},${n + 4},${n + 5},${n + 6},false,${n + 7})");
+            n += 8;
+            var row = chunk[i];
+            var c = SampleColumnMapper.ToColumns(row.Value);
+            QuestDbConnection.BindTimestamp(cmd, row.TimeUtc);
+            QuestDbConnection.BindText(cmd, tag);
+            QuestDbConnection.BindByte(cmd, c.ValueKind);
+            cmd.Parameters.Add(new NpgsqlParameter { NpgsqlDbType = NpgsqlDbType.Double, Value = c.NumValue is { } d ? d : DBNull.Value });
+            cmd.Parameters.Add(new NpgsqlParameter { NpgsqlDbType = NpgsqlDbType.Bigint, Value = c.IntValue is { } l ? l : DBNull.Value });
+            QuestDbConnection.BindText(cmd, c.StrValue);
+            QuestDbConnection.BindLong(cmd, SampleColumnMapper.ToLongStatus(row.StatusCode));
+            QuestDbConnection.BindTimestamp(cmd, serverTs);
+        }
+        cmd.CommandText = sql.ToString();
+        await cmd.ExecuteNonQueryAsync(ct).ConfigureAwait(false);
+    }
+}
@@ -0,0 +1,141 @@
+using System.Text;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Logging.Abstractions;
+using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb;
+
+/// <summary>
+///     <see cref="IHistorianProvisioning"/> for the QuestDB backend: upserts <c>tag</c> rows (fixed designated timestamp
+///     1970-01-01, so the dedup key makes it an upsert by name). QuestDB replaces the WHOLE row on a key hit, so a null
+///     engineering unit / description in a request would wipe the stored one; per chunk the existing rows are read
+///     (<c>WHERE name IN (…)</c>) and <c>COALESCE(request, stored)</c> is applied client-side before inserting.
+///     Every data type is provisionable (Skipped is always 0). Never throws. After each chunk it waits for WAL
+///     visibility and tells the catalog the names are known (R7).
+/// </summary>
+internal sealed class QuestDbTagProvisioner : IHistorianProvisioning
+{
+    private const int ChunkSize = 400;          // 6 parameters per tag on insert, 400 on the IN list
+    private const int MaxUnitLength = 64;
+    private const int MaxDescriptionLength = 400;
+
+    private readonly QuestDbConnection _conn;
+    private readonly QuestDbSchemaBootstrapper _bootstrap;
+    private readonly QuestDbTagCatalog _catalog;
+    private readonly QuestDbWalVisibility _wal;
+    private readonly ILogger _log;
+    private readonly TimeProvider _clock;
+
+    /// <summary>Creates the provisioner.</summary>
+    /// <param name="conn">Connection helper.</param>
+    /// <param name="bootstrap">Schema bootstrapper.</param>
+    /// <param name="catalog">Tag catalog (told which names are now known).</param>
+    /// <param name="wal">WAL visibility wait.</param>
+    /// <param name="log">Optional logger.</param>
+    /// <param name="clock">Clock for <c>updated_utc</c>.</param>
+    public QuestDbTagProvisioner(QuestDbConnection conn, QuestDbSchemaBootstrapper bootstrap, QuestDbTagCatalog catalog,
+        QuestDbWalVisibility wal, ILogger? log = null, TimeProvider? clock = null)
+    {
+        _conn = conn;
+        _bootstrap = bootstrap;
+        _catalog = catalog;
+        _wal = wal;
+        _log = log ?? NullLogger.Instance;
+        _clock = clock ?? TimeProvider.System;
+    }
+
+    /// <inheritdoc/>
+    public async Task<HistorianProvisionResult> EnsureTagsAsync(IReadOnlyList<HistorianTagProvisionRequest> requests, CancellationToken ct)
+    {
+        var requested = requests?.Count ?? 0;
+        if (requested == 0) return new HistorianProvisionResult(0, 0, 0, 0);
+        var failed = 0;
+        var ensured = 0;
+        try
+        {
+            await _bootstrap.EnsureAsync(ct).ConfigureAwait(false);
+
+            // Distinct names (last request wins); over-long/empty names fail alone. Weight = requests folded into the name.
+            var byName = new Dictionary<string, (HistorianTagProvisionRequest Req, int Count)>(StringComparer.Ordinal);
+            foreach (var r in requests!)
+            {
+                if (r is null || string.IsNullOrWhiteSpace(r.TagName) || !HistorianTagNames.IsValid(r.TagName)) { failed++; continue; }
+                byName[r.TagName] = (r, byName.TryGetValue(r.TagName, out var e) ? e.Count + 1 : 1);
+            }
+
+            var ordered = byName.OrderBy(kv => kv.Key, StringComparer.Ordinal).Select(kv => kv.Value).ToList();
+            for (var i = 0; i < ordered.Count; i += ChunkSize)
+            {
+                var chunk = ordered.Skip(i).Take(ChunkSize).ToList();
+                var weight = chunk.Sum(c => c.Count);
+                if (await TryUpsertAsync(chunk.Select(c => c.Req).ToList(), ct).ConfigureAwait(false)) ensured += weight;
+                else failed += weight;
+            }
+        }
+        catch (Exception ex)
+        {
+            _log.LogWarning("QuestDB historian provisioning failed ({Exception}).", ex.GetType().Name);
+            failed = requested - ensured;
+        }
+
+        return new HistorianProvisionResult(requested, ensured, 0, failed);
+    }
+
+    private async Task<bool> TryUpsertAsync(List<HistorianTagProvisionRequest> chunk, CancellationToken ct)
+    {
+        try
+        {
+            var stored = new Dictionary<string, (string? Unit, string? Desc)>(StringComparer.Ordinal);
+            await using (var conn = await _conn.OpenAsync(ct).ConfigureAwait(false))
+            {
+                var sel = new StringBuilder($"SELECT name, eng_unit, description FROM {_conn.TagTable} WHERE name IN (");
+                await using var cmd = _conn.CreateCommand(conn, "");
+                for (var i = 0; i < chunk.Count; i++)
+                {
+                    sel.Append(i == 0 ? "$" : ",$").Append(i + 1);
+                    QuestDbConnection.BindText(cmd, chunk[i].TagName);
+                }
+                cmd.CommandText = sel.Append(')').ToString();
+                await using var r = await cmd.ExecuteReaderAsync(ct).ConfigureAwait(false);
+                while (await r.ReadAsync(ct).ConfigureAwait(false))
+                    stored[r.GetString(0)] = (r.IsDBNull(1) ? null : r.GetString(1), r.IsDBNull(2) ? null : r.GetString(2));
+            }
+
+            var now = _clock.GetUtcNow().UtcDateTime;
+            await using (var conn = await _conn.OpenAsync(ct).ConfigureAwait(false))
+            {
+                var ins = new StringBuilder($"INSERT INTO {_conn.TagTable} (ts, name, data_type, eng_unit, description, updated_utc) VALUES ");
+                await using var cmd = _conn.CreateCommand(conn, "");
+                var n = 1;
+                for (var i = 0; i < chunk.Count; i++)
+                {
+                    var req = chunk[i];
+                    stored.TryGetValue(req.TagName, out var old);
+                    if (i > 0) ins.Append(',');
+                    ins.Append($"(${n},${n + 1},${n + 2},${n + 3},${n + 4},${n + 5})");
+                    n += 6;
+                    QuestDbConnection.BindTimestamp(cmd, QuestDbTables.TagEpoch);
+                    QuestDbConnection.BindText(cmd, req.TagName);
+                    QuestDbConnection.BindByte(cmd, (byte)req.DataType);
+                    QuestDbConnection.BindText(cmd, Clip(req.EngineeringUnit ?? old.Unit, MaxUnitLength));
+                    QuestDbConnection.BindText(cmd, Clip(req.Description ?? old.Desc, MaxDescriptionLength));
+                    QuestDbConnection.BindTimestamp(cmd, now);
+                }
+                cmd.CommandText = ins.ToString();
+                await cmd.ExecuteNonQueryAsync(ct).ConfigureAwait(false);
+            }
+
+            await _wal.WaitAsync(_conn.TagTable, ct).ConfigureAwait(false);
+            foreach (var c in chunk) _catalog.MarkKnown(c.TagName);
+            return true;
+        }
+        catch (Exception ex)
+        {
+            _log.LogWarning("QuestDB historian tag upsert failed ({Exception}).", ex.GetType().Name);
+            return false;
+        }
+    }
+
+    private static string? Clip(string? s, int max) => s is null ? null : s.Length <= max ? s : s[..max];
+}