feat(historian): SQL Server value writer, tag provisioner, alarm writer (T9)

Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
2026-10-09 13:08:43 -04:00
parent 65e6352dad
commit bef47f89da
8 changed files with 864 additions and 0 deletions
@@ -0,0 +1,168 @@
+using System.Data;
+using System.Security.Cryptography;
+using System.Text;
+using Microsoft.Data.SqlClient;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Logging.Abstractions;
+using ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian;
+using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer;
+
+/// <summary>
+///     <see cref="IAlarmHistorianWriter"/> over the SQL <c>Event</c> table. Per event: insert-if-absent keyed by a
+///     deterministic <see cref="ComputeEventId"/>, so the store-and-forward drain's at-least-once redelivery (and both
+///     pair nodes writing the same event) converge on one row. Outcomes: Ack on insert or already-present;
+///     RetryPlease on transient SQL / schema-bootstrap / unclassified SQL errors and on cancellation; PermanentFail only
+///     for data errors (truncation, conversion, null, range) and non-SQL exceptions. Never throws.
+/// </summary>
+internal sealed class SqlServerAlarmHistorianWriter : IAlarmHistorianWriter
+{
+    // 2628/8152 string truncation, 515 null into NOT NULL, 245/8114 conversion, 220/8115 overflow, 242/241 datetime range.
+    private static readonly int[] DataErrorNumbers = [2628, 8152, 515, 245, 8114, 220, 8115, 242, 241];
+
+    private readonly SqlHistorianConnection _conn;
+    private readonly SqlServerSchemaBootstrapper _bootstrap;
+    private readonly ILogger _log;
+
+    /// <summary>Creates the writer.</summary>
+    /// <param name="conn">Connection helper.</param>
+    /// <param name="bootstrap">Schema bootstrapper.</param>
+    /// <param name="log">Optional logger.</param>
+    public SqlServerAlarmHistorianWriter(SqlHistorianConnection conn, SqlServerSchemaBootstrapper bootstrap, ILogger? log = null)
+    {
+        _conn = conn;
+        _bootstrap = bootstrap;
+        _log = log ?? NullLogger.Instance;
+    }
+
+    /// <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("SQL alarm historian bootstrap failed ({Exception}); will retry.", ex.GetType().Name);
+            Array.Fill(outcomes, HistorianWriteOutcome.RetryPlease);
+            return outcomes;
+        }
+
+        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);
+        }
+
+        return outcomes;
+    }
+
+    private async Task<HistorianWriteOutcome> WriteOneAsync(AlarmHistorianEvent e, CancellationToken ct)
+    {
+        try
+        {
+            // String parameters are NVARCHAR(MAX) on purpose: a sized parameter silently truncates client-side, which would
+            // hide an over-long value; the server-side 8152 is what classifies it PermanentFail.
+            var sql =
+                $"INSERT INTO {_conn.Q("Event")} (EventId, SourceName, AlarmId, AlarmName, AlarmTypeName, Severity, EventKind, Message, UserName, Comment, EventTimeUtc, ReceivedTimeUtc) " +
+                "SELECT @id, @src, @aid, @an, @at, @sev, @ek, @msg, @usr, @cmt, @t, SYSUTCDATETIME() " +
+                $"WHERE NOT EXISTS (SELECT 1 FROM {_conn.Q("Event")} x WITH (UPDLOCK, HOLDLOCK) WHERE x.EventId = @id)";
+            {
+                try
+                {
+                    await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false);
+                    await using var cmd = _conn.CreateCommand(conn, sql);
+                    cmd.Parameters.Add(new SqlParameter("@id", SqlDbType.NVarChar, 64) { Value = ComputeEventId(e) });
+                    cmd.Parameters.Add(new SqlParameter("@src", SqlDbType.NVarChar, -1) { Value = e.EquipmentPath ?? "" });
+                    cmd.Parameters.Add(new SqlParameter("@aid", SqlDbType.NVarChar, -1) { Value = e.AlarmId ?? "" });
+                    cmd.Parameters.Add(new SqlParameter("@an", SqlDbType.NVarChar, -1) { Value = e.AlarmName ?? "" });
+                    cmd.Parameters.Add(new SqlParameter("@at", SqlDbType.NVarChar, -1) { Value = e.AlarmTypeName ?? "" });
+                    cmd.Parameters.Add(new SqlParameter("@sev", SqlDbType.Int) { Value = ToNumericSeverity(e.Severity) });
+                    cmd.Parameters.Add(new SqlParameter("@ek", SqlDbType.NVarChar, -1) { Value = e.EventKind ?? "" });
+                    cmd.Parameters.Add(new SqlParameter("@msg", SqlDbType.NVarChar, -1) { Value = (object?)e.Message ?? DBNull.Value });
+                    cmd.Parameters.Add(new SqlParameter("@usr", SqlDbType.NVarChar, -1) { Value = (object?)e.User ?? DBNull.Value });
+                    cmd.Parameters.Add(new SqlParameter("@cmt", SqlDbType.NVarChar, -1) { Value = (object?)e.Comment ?? DBNull.Value });
+                    SqlHistorianConnection.AddDateTime2(cmd, "@t", e.TimestampUtc);
+                    await cmd.ExecuteNonQueryAsync(ct).ConfigureAwait(false);
+                    return HistorianWriteOutcome.Ack;
+                }
+                // Single-row statement: a duplicate-key error means the row IS there (a concurrent writer won).
+                catch (SqlException ex) when (SqlHistorianConnection.IsDuplicateKey(ex))
+                {
+                    return HistorianWriteOutcome.Ack;
+                }
+            }
+        }
+        catch (OperationCanceledException) when (ct.IsCancellationRequested)
+        {
+            return HistorianWriteOutcome.RetryPlease;
+        }
+        catch (Exception ex)
+        {
+            var outcome = Classify(ex);
+            _log.LogWarning("SQL alarm event write failed ({Exception}); outcome {Outcome}.", ex.GetType().Name, outcome);
+            return outcome;
+        }
+    }
+
+    private static HistorianWriteOutcome Classify(Exception ex)
+    {
+        if (SqlHistorianConnection.IsTransient(ex)) return HistorianWriteOutcome.RetryPlease;
+        if (ex is SqlException sql)
+            return sql.Errors.Cast<SqlError>().Any(x => Array.IndexOf(DataErrorNumbers, x.Number) >= 0)
+                ? HistorianWriteOutcome.PermanentFail
+                : HistorianWriteOutcome.RetryPlease;   // config/auth/availability problems must not dead-letter data
+        return HistorianWriteOutcome.PermanentFail;
+    }
+
+    // Same numeric Part 9 priorities the gateway path sends (AlarmEventMapper.ToNumericSeverity).
+    private static int ToNumericSeverity(AlarmSeverity s) => s switch
+    {
+        AlarmSeverity.Low => 200,
+        AlarmSeverity.Medium => 500,
+        AlarmSeverity.High => 700,
+        AlarmSeverity.Critical => 900,
+        _ => 500,
+    };
+
+    /// <summary>
+    ///     Lowercase-hex SHA-256 over length-prefixed UTF-8 fields (little-endian int32 byte length, -1 for null):
+    ///     EquipmentPath, AlarmId, EventKind, TimestampUtc.Ticks (decimal), User, Comment, Message. Length-prefixing keeps
+    ///     field boundaries unambiguous (<c>"a|b"+"c"</c> ≠ <c>"a"+"b|c"</c>).
+    /// </summary>
+    /// <param name="e">The event.</param>
+    /// <returns>64 hex characters.</returns>
+    internal static string ComputeEventId(AlarmHistorianEvent e)
+    {
+        using var ms = new MemoryStream();
+        Field(ms, e.EquipmentPath);
+        Field(ms, e.AlarmId);
+        Field(ms, e.EventKind);
+        Field(ms, e.TimestampUtc.Ticks.ToString(System.Globalization.CultureInfo.InvariantCulture));
+        Field(ms, e.User);
+        Field(ms, e.Comment);
+        Field(ms, e.Message);
+        return Convert.ToHexStringLower(SHA256.HashData(ms.ToArray()));
+    }
+
+    private static void Field(MemoryStream ms, string? value)
+    {
+        Span<byte> len = stackalloc byte[4];
+        if (value is null)
+        {
+            System.Buffers.Binary.BinaryPrimitives.WriteInt32LittleEndian(len, -1);
+            ms.Write(len);
+            return;
+        }
+
+        var bytes = Encoding.UTF8.GetBytes(value);
+        System.Buffers.Binary.BinaryPrimitives.WriteInt32LittleEndian(len, bytes.Length);
+        ms.Write(len);
+        ms.Write(bytes);
+    }
+}
@@ -0,0 +1,134 @@
+using Microsoft.Data.SqlClient;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Logging.Abstractions;
+using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer;
+
+/// <summary>
+///     The one place that maps value kinds to <c>Tag.DataType</c> (the schema stores a <see cref="DriverDataType"/> byte).
+///     Used only when the writer has to create a tag row on a miss; provisioning later corrects it.
+/// </summary>
+internal static class SqlHistorianDataTypes
+{
+    /// <summary>Data type used when a tag is created from a batch that holds no typed value.</summary>
+    public const DriverDataType DefaultForEmpty = DriverDataType.Float64;
+
+    /// <summary>Maps a value kind to the stored <c>Tag.DataType</c> byte (Empty ⇒ <see cref="DefaultForEmpty"/>).</summary>
+    /// <param name="kind">Value kind.</param>
+    /// <returns>The <see cref="DriverDataType"/> as a byte.</returns>
+    public static byte FromKind(HistorianValueKind kind) => (byte)(kind switch
+    {
+        HistorianValueKind.Double => DriverDataType.Float64,
+        HistorianValueKind.Int64 => DriverDataType.Int64,
+        HistorianValueKind.Boolean => DriverDataType.Boolean,
+        HistorianValueKind.String => DriverDataType.String,
+        HistorianValueKind.DateTime => DriverDataType.DateTime,
+        _ => DefaultForEmpty,
+    });
+}
+
+/// <summary>
+///     <see cref="IHistorianValueWriter"/> over the SQL <c>Sample</c> table. Redelivery-idempotent: rows are
+///     inserted only where <c>(TagId, SourceTimeUtc)</c> is absent, so the recorder's at-least-once redelivery
+///     leaves the first-stored sample untouched. Never throws; returns false on any failure so the recorder keeps
+///     the entries and its #538 per-tag quarantine keeps working.
+/// </summary>
+/// <remarks>
+///     <b>Non-finite doubles.</b> NaN / ±Infinity cannot live in a SQL <c>float</c> (the TVP throws). They are stored
+///     as <see cref="HistorianValueKind.Empty"/> keeping the original status code, so the sample's presence and
+///     status survive and the tag is not quarantined for one bad reading.
+///     <b>Duplicate timestamps within a batch</b> collapse to the first (matching insert-if-absent across batches).
+/// </remarks>
+internal sealed class SqlServerHistorianValueWriter : IHistorianValueWriter
+{
+    private readonly SqlHistorianConnection _conn;
+    private readonly SqlServerSchemaBootstrapper _bootstrap;
+    private readonly SqlServerTagCatalog _catalog;
+    private readonly ILogger _log;
+    private readonly Func<DateTime> _utcNow;
+
+    /// <summary>Creates the writer.</summary>
+    /// <param name="conn">Connection helper.</param>
+    /// <param name="bootstrap">Schema bootstrapper.</param>
+    /// <param name="catalog">Tag catalog.</param>
+    /// <param name="log">Optional logger.</param>
+    /// <param name="utcNow">Clock for null timestamps (tests).</param>
+    public SqlServerHistorianValueWriter(SqlHistorianConnection conn, SqlServerSchemaBootstrapper bootstrap,
+        SqlServerTagCatalog catalog, ILogger? log = null, Func<DateTime>? utcNow = null)
+    {
+        _conn = conn;
+        _bootstrap = bootstrap;
+        _catalog = catalog;
+        _log = log ?? NullLogger.Instance;
+        _utcNow = utcNow ?? (() => DateTime.UtcNow);
+    }
+
+    /// <inheritdoc/>
+    public async Task<bool> WriteLiveValuesAsync(string tag, IReadOnlyList<HistorizationValue> values, CancellationToken ct)
+    {
+        try
+        {
+            await _bootstrap.EnsureAsync(ct).ConfigureAwait(false);
+            if (values is null || values.Count == 0) return true;
+
+            var rows = Prepare(values, _utcNow);
+            var dataType = SqlHistorianDataTypes.FromKind(rows.Select(r => r.Value.Kind).FirstOrDefault(k => k != HistorianValueKind.Empty));
+            var tagId = await _catalog.GetOrCreateTagIdAsync(tag, dataType, ct).ConfigureAwait(false);
+
+            var sql =
+                $"INSERT INTO {_conn.Q("Sample")} (TagId, SourceTimeUtc, ValueKind, NumValue, IntValue, StrValue, StatusCode, ServerTimeUtc) " +
+                "SELECT @tagId, r.SourceTimeUtc, r.ValueKind, r.NumValue, r.IntValue, r.StrValue, r.StatusCode, SYSUTCDATETIME() " +
+                "FROM @rows r WHERE NOT EXISTS (SELECT 1 FROM " + _conn.Q("Sample") +
+                " s WITH (UPDLOCK, HOLDLOCK) WHERE s.TagId = @tagId AND s.SourceTimeUtc = r.SourceTimeUtc)";
+
+            // A duplicate-key error rolls the WHOLE statement back, so "someone else wrote a row" is NOT proof our rows are
+            // present. The retry re-evaluates NOT EXISTS against what is now there; only its success counts.
+            for (var attempt = 0; ; attempt++)
+            {
+                try
+                {
+                    await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false);
+                    await using var cmd = _conn.CreateCommand(conn, sql);
+                    cmd.Parameters.Add(new SqlParameter("@tagId", System.Data.SqlDbType.Int) { Value = tagId });
+                    _conn.AddStructured(cmd, "@rows", "SampleRow", SampleRowMapper.ToSampleRecords(rows));
+                    await cmd.ExecuteNonQueryAsync(ct).ConfigureAwait(false);
+                    return true;
+                }
+                catch (SqlException ex) when (attempt < 2 && SqlHistorianConnection.IsDuplicateKey(ex))
+                {
+                }
+            }
+        }
+        catch (Exception ex)
+        {
+            _log.LogWarning("SQL historian live-value write failed ({Exception}); will retry.", ex.GetType().Name);
+            return false;
+        }
+    }
+
+    /// <summary>
+    ///     Normalises a batch for the TVP: null timestamp ⇒ now, non-finite doubles ⇒ Empty (status kept), and
+    ///     duplicate timestamps collapsed to the first.
+    /// </summary>
+    /// <param name="values">Input samples, in append order.</param>
+    /// <param name="utcNow">Clock for null timestamps.</param>
+    /// <returns>Time-unique rows.</returns>
+    internal static List<SqlSampleRow> Prepare(IReadOnlyList<HistorizationValue> values, Func<DateTime> utcNow)
+    {
+        var seen = new HashSet<long>();
+        var rows = new List<SqlSampleRow>(values.Count);
+        foreach (var v in values)
+        {
+            var t = v.TimestampUtc ?? utcNow();
+            t = t.Kind == DateTimeKind.Local ? t.ToUniversalTime() : DateTime.SpecifyKind(t, DateTimeKind.Utc);
+            if (!seen.Add(t.Ticks)) continue;
+            var value = v.Value;
+            if (value.Kind == HistorianValueKind.Double && !double.IsFinite(value.Number))
+                value = HistorianValue.Empty;
+            rows.Add(new SqlSampleRow(t, value, v.StatusCode));
+        }
+
+        return rows;
+    }
+}
@@ -0,0 +1,120 @@
+using System.Data;
+using System.Text;
+using Microsoft.Data.SqlClient;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Logging.Abstractions;
+using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer;
+
+/// <summary>
+///     <see cref="IHistorianProvisioning"/> for the SQL backend: one <c>MERGE … WITH (HOLDLOCK)</c> upsert per chunk of
+///     tags (chunked only to stay under SQL's 2100-parameter limit). Every data type is provisionable (Skipped is
+///     always 0). A null engineering unit / description in a request keeps the stored value rather than wiping it.
+///     Never throws.
+/// </summary>
+internal sealed class SqlServerTagProvisioner : IHistorianProvisioning
+{
+    private const int ChunkSize = 400;          // 4 parameters per tag
+    private const int MaxNameLength = 400;
+    private const int MaxUnitLength = 64;
+    private const int MaxDescriptionLength = 400;
+    private const int MaxAttempts = 4;
+
+    private readonly SqlHistorianConnection _conn;
+    private readonly SqlServerSchemaBootstrapper _bootstrap;
+    private readonly ILogger _log;
+
+    /// <summary>Creates the provisioner.</summary>
+    /// <param name="conn">Connection helper.</param>
+    /// <param name="bootstrap">Schema bootstrapper.</param>
+    /// <param name="log">Optional logger.</param>
+    public SqlServerTagProvisioner(SqlHistorianConnection conn, SqlServerSchemaBootstrapper bootstrap, ILogger? log = null)
+    {
+        _conn = conn;
+        _bootstrap = bootstrap;
+        _log = log ?? NullLogger.Instance;
+    }
+
+    /// <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 (a MERGE source must not repeat a key); last request wins. Over-long/empty names fail alone.
+            var byName = new Dictionary<string, (HistorianTagProvisionRequest Req, int Count)>(StringComparer.Ordinal);
+            foreach (var r in requests!)
+            {
+                if (r is null || string.IsNullOrWhiteSpace(r.TagName) || r.TagName.Length > MaxNameLength) { 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 TryMergeAsync(chunk.Select(c => c.Req).ToList(), ct).ConfigureAwait(false)) ensured += weight;
+                else failed += weight;
+            }
+        }
+        catch (Exception ex)
+        {
+            _log.LogWarning("SQL historian provisioning failed ({Exception}).", ex.GetType().Name);
+            failed = requested - ensured;
+        }
+
+        return new HistorianProvisionResult(requested, ensured, 0, failed);
+    }
+
+    private async Task<bool> TryMergeAsync(List<HistorianTagProvisionRequest> chunk, CancellationToken ct)
+    {
+        var sql = new StringBuilder();
+        sql.Append($"MERGE {_conn.Q("Tag")} WITH (HOLDLOCK) AS t USING (VALUES ");
+        for (var i = 0; i < chunk.Count; i++)
+            sql.Append(i == 0 ? "" : ",").Append($"(@n{i},@d{i},@e{i},@c{i})");
+        sql.Append(") AS s (Name, DataType, EngineeringUnit, Description) ON t.Name = s.Name " +
+                   "WHEN MATCHED THEN UPDATE SET DataType = s.DataType, EngineeringUnit = COALESCE(s.EngineeringUnit, t.EngineeringUnit), " +
+                   "Description = COALESCE(s.Description, t.Description), UpdatedUtc = SYSUTCDATETIME() " +
+                   "WHEN NOT MATCHED THEN INSERT (Name, DataType, EngineeringUnit, Description, CreatedUtc, UpdatedUtc) " +
+                   "VALUES (s.Name, s.DataType, s.EngineeringUnit, s.Description, SYSUTCDATETIME(), SYSUTCDATETIME());");
+
+        for (var attempt = 1; ; attempt++)
+        {
+            try
+            {
+                await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false);
+                await using var cmd = _conn.CreateCommand(conn, sql.ToString());
+                for (var i = 0; i < chunk.Count; i++)
+                {
+                    var r = chunk[i];
+                    cmd.Parameters.Add(new SqlParameter($"@n{i}", SqlDbType.NVarChar, MaxNameLength) { Value = r.TagName });
+                    cmd.Parameters.Add(new SqlParameter($"@d{i}", SqlDbType.TinyInt) { Value = (byte)r.DataType });
+                    cmd.Parameters.Add(new SqlParameter($"@e{i}", SqlDbType.NVarChar, MaxUnitLength) { Value = Clip(r.EngineeringUnit, MaxUnitLength) });
+                    cmd.Parameters.Add(new SqlParameter($"@c{i}", SqlDbType.NVarChar, MaxDescriptionLength) { Value = Clip(r.Description, MaxDescriptionLength) });
+                }
+
+                await cmd.ExecuteNonQueryAsync(ct).ConfigureAwait(false);
+                return true;
+            }
+            catch (SqlException ex) when (attempt < MaxAttempts &&
+                (SqlHistorianConnection.IsDuplicateKey(ex) || SqlHistorianConnection.IsTransient(ex)))
+            {
+                await Task.Delay(20 * attempt, ct).ConfigureAwait(false);
+            }
+            catch (Exception ex)
+            {
+                _log.LogWarning("SQL historian tag MERGE failed ({Exception}).", ex.GetType().Name);
+                return false;
+            }
+        }
+    }
+
+    private static object Clip(string? s, int max) => s is null ? DBNull.Value : s.Length <= max ? s : s[..max];
+}