fix(historian): SQL quality contract, long-name refusal, info-bit masking, AtTime-after-now, bootstrap hardening

- I-1: an Empty sample is always Bad. The writer stores a non-Bad NaN/Inf as
  Empty+BadOutOfRange and any other non-Bad Empty as BadNoData; the updater
  refuses a non-Bad Empty with BadInvalidArgument; reads clamp a legacy
  non-Bad Empty row to BadNoData (Raw and AtTime).
- Tag/source names over 400 chars are refused, never truncated: writer
  per-tag failure, updater BadInvalidArgument, reads not-found, events none.
- HistorianInfoBits masks the historian bits before setting one.
- AtTime past now, stepped from the last sample => UncertainDataSubNormal.
- Bootstrapper: logger, 5 s negative cache, ClearPool + Warning when
  sp_releaseapplock fails.
- Provider builds its shared state lazily (an invalid Schema no longer
  throws at Resolve); Kinds is a FrozenSet; Gateway reuses the recorder's
  DefaultSupportedKinds.
- Cleanup: SqlHistorianDataTypes in its own file (updater dup removed),
  dead Invalidate and unused status consts deleted, named severity masks,
  alarm writer brace.

Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
This commit is contained in:
Joseph Doherty
2026-10-09 14:16:31 -04:00
parent fc57eefca6
commit 821763a09d
20 changed files with 611 additions and 126 deletions
@@ -14,13 +14,6 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway;
/// </summary>
public sealed class GatewayHistorianProvider : IHistorianBackendProvider
{
private static readonly IReadOnlySet<HistorianValueKind> Kinds = new HashSet<HistorianValueKind>
{
HistorianValueKind.Double,
HistorianValueKind.Int64,
HistorianValueKind.Boolean,
};
private readonly ServerHistorianOptions _options;
/// <summary>Creates the provider over the bound <c>ServerHistorian</c> options.</summary>
@@ -34,7 +27,11 @@ public sealed class GatewayHistorianProvider : IHistorianBackendProvider
public string Name => "Gateway";
/// <inheritdoc />
public IReadOnlySet<HistorianValueKind> SupportedKinds => Kinds;
/// <remarks>
/// The recorder's own default set (<see cref="ContinuousHistorizationRecorder.DefaultSupportedKinds"/>), so the
/// gateway's kinds and the recorder's fallback can never drift apart.
/// </remarks>
public IReadOnlySet<HistorianValueKind> SupportedKinds => ContinuousHistorizationRecorder.DefaultSupportedKinds;
/// <inheritdoc />
public IHistorianDataSource CreateDataSource(IServiceProvider sp) => GatewayHistorian.CreateDataSource(_options, sp);
@@ -0,0 +1,27 @@
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 by the value writer and the updater only when they have 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,
});
}
@@ -7,17 +7,46 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer;
internal static class SqlHistorianStatusCodes
{
public const uint Good = 0x00000000u;
public const uint GoodNoData = 0x00A50000u;
public const uint GoodEntryInserted = 0x00A20000u;
public const uint GoodEntryReplaced = 0x00A30000u;
public const uint BadNoData = 0x809B0000u;
public const uint BadEntryExists = 0x809F0000u;
public const uint BadNoEntryExists = 0x80A00000u;
public const uint BadInvalidTimestamp = 0x80230000u;
public const uint BadTypeMismatch = 0x80740000u;
public const uint BadAggregateInvalidInputs = 0x80D60000u;
public const uint BadHistoryOperationUnsupported = 0x80720000u;
public const uint UncertainDataSubNormal = 0x40A40000u;
/// <summary>Stored for a non-finite Double (NaN / ±Infinity) that arrived with a non-Bad status.</summary>
public const uint BadOutOfRange = 0x803C0000u;
}
/// <summary>
/// Bit masks over a 32-bit OPC UA status code (Part 4 §7.39). These are masks, not emitted codes, so their
/// names deliberately avoid the Good/Bad/Uncertain prefix the status-code parity scan keys on.
/// </summary>
internal static class StatusBits
{
/// <summary>Severity + sub-code (top 16 bits) of a status, i.e. the code without info bits.</summary>
public const uint CodeBitsMask = 0xFFFF0000u;
/// <summary>The two severity bits (Good = 00, Uncertain = 01, Bad = 10).</summary>
public const uint SeverityMask = 0xC0000000u;
/// <summary>The severity bit set on every Bad code.</summary>
public const uint SeverityBadBit = 0x80000000u;
/// <summary>The two historian info bits (Calculated, Interpolated) in the low bits of the info field.</summary>
public const uint HistorianBitsMask = 0x3u;
/// <summary>True when <paramref name="status"/> has Bad severity.</summary>
/// <param name="status">A status code.</param>
/// <returns>Whether the Bad severity bit is set.</returns>
public static bool IsBad(uint status) => (status & SeverityBadBit) != 0;
/// <summary>True when <paramref name="status"/> has Good severity.</summary>
/// <param name="status">A status code.</param>
/// <returns>Whether both severity bits are clear.</returns>
public static bool IsGood(uint status) => (status & SeverityMask) == 0;
}
/// <summary>Part 4 §7.39 historian info bits OR-ed onto a status code.</summary>
@@ -32,13 +61,21 @@ internal static class HistorianInfoBits
/// <summary>The value was interpolated.</summary>
public const uint HistorianInterpolated = 0x2u;
/// <summary>Marks <paramref name="status"/> as a Calculated data value.</summary>
/// <summary>
/// Marks <paramref name="status"/> as a Calculated data value. Any historian bits the status already carries
/// (e.g. a stored Interpolated sample) are cleared first, so the result is never both.
/// </summary>
/// <param name="status">Base status code.</param>
/// <returns>Status with InfoType and Calculated bits set.</returns>
public static uint WithCalculated(uint status) => status | InfoTypeDataValue | HistorianCalculated;
public static uint WithCalculated(uint status) =>
(status & ~StatusBits.HistorianBitsMask) | InfoTypeDataValue | HistorianCalculated;
/// <summary>Marks <paramref name="status"/> as an Interpolated data value.</summary>
/// <summary>
/// Marks <paramref name="status"/> as an Interpolated data value. Any historian bits the status already
/// carries (e.g. a stored Calculated sample) are cleared first, so the result is never both.
/// </summary>
/// <param name="status">Base status code.</param>
/// <returns>Status with InfoType and Interpolated bits set.</returns>
public static uint WithInterpolated(uint status) => status | InfoTypeDataValue | HistorianInterpolated;
public static uint WithInterpolated(uint status) =>
(status & ~StatusBits.HistorianBitsMask) | InfoTypeDataValue | HistorianInterpolated;
}
@@ -95,7 +95,7 @@ internal sealed class SqlServerAlarmHistorianWriter : IAlarmHistorianWriter
{
return HistorianWriteOutcome.Ack;
}
}
}
catch (OperationCanceledException) when (ct.IsCancellationRequested)
{
return HistorianWriteOutcome.RetryPlease;
@@ -34,6 +34,14 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer;
/// markers included. Aggregates exclude Bad samples from the math, report the interval Uncertain when
/// any non-Good sample is present, and count them in PercentBad. AtTime never interpolates across a Bad
/// neighbour and reports a time inside an outage with the outage's Bad code (see <see cref="ReadAtTimeAsync"/>).
/// <b>An Empty sample is always served Bad:</b> a row with no value whose stored status is not Bad (only
/// possible for rows written before the writer forced Empty samples Bad) is read back as <c>BadNoData</c>,
/// on Raw and AtTime alike.
/// </para>
/// <para>
/// <b>Long names.</b> A tag name or event source name longer than <see cref="SqlServerTagCatalog.MaxNameLength"/>
/// can never be stored, so it is answered as not found (Raw: empty; AtTime / Processed: <c>BadNoData</c>;
/// Events: none) without being truncated onto another series.
/// </para>
/// </remarks>
public sealed class SqlServerHistorianDataSource : IHistorianDataSource
@@ -41,15 +49,13 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource
/// <summary>Spec §9.7: a processed read covering more intervals than this is refused.</summary>
public const int MaxProcessedBuckets = 100_000;
/// <summary>Severity + sub-code (top 16 bits) of a status, i.e. the code without info bits.</summary>
private const uint StatusCodeBitsMask = 0xFFFF0000u;
private const string SampleColumns = "SourceTimeUtc, ValueKind, NumValue, IntValue, StrValue, StatusCode, ServerTimeUtc";
private readonly SqlHistorianConnection _conn;
private readonly SqlServerSchemaBootstrapper _bootstrap;
private readonly SqlServerTagCatalog _catalog;
private readonly ILogger _log;
private readonly TimeProvider _time;
private readonly object _healthLock = new();
private DateTime? _lastSuccessUtc;
@@ -65,13 +71,16 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource
/// <param name="bootstrap">Schema bootstrapper (shared with the other SQL seams).</param>
/// <param name="catalog">Tag catalog (shared; reads use only its read-only lookup).</param>
/// <param name="log">Optional logger.</param>
/// <param name="timeProvider">Clock for "now" (AtTime beyond the last sample, server timestamps); system by default.</param>
internal SqlServerHistorianDataSource(
SqlHistorianConnection conn, SqlServerSchemaBootstrapper bootstrap, SqlServerTagCatalog catalog, ILogger? log = null)
SqlHistorianConnection conn, SqlServerSchemaBootstrapper bootstrap, SqlServerTagCatalog catalog, ILogger? log = null,
TimeProvider? timeProvider = null)
{
_conn = conn ?? throw new ArgumentNullException(nameof(conn));
_bootstrap = bootstrap ?? throw new ArgumentNullException(nameof(bootstrap));
_catalog = catalog ?? throw new ArgumentNullException(nameof(catalog));
_log = log ?? NullLogger.Instance;
_time = timeProvider ?? TimeProvider.System;
}
// ------------------------------------------------------------------------------------------ Raw
@@ -149,7 +158,8 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource
/// neighbours are Good, else <c>UncertainDataSubNormal</c> + Interpolated (never better than the
/// inputs).</item>
/// <item>Boolean/String/DateTime, or no <c>n</c> (after the last sample) ⇒ stepped: <c>p</c>'s value with
/// <c>p</c>'s status code + Interpolated bits.</item>
/// <c>p</c>'s status code + Interpolated bits — except that a time later than now that is stepped from the
/// last sample is <c>UncertainDataSubNormal</c> + Interpolated (the value is not known to still hold).</item>
/// </list>
/// Unknown tag ⇒ every entry <c>BadNoData</c>.
/// </remarks>
@@ -160,7 +170,7 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource
ArgumentNullException.ThrowIfNull(timestampsUtc);
try
{
var now = DateTime.UtcNow;
var now = _time.GetUtcNow().UtcDateTime;
var byTicks = new Dictionary<long, DataValueSnapshot>(timestampsUtc.Count);
var tagId = timestampsUtc.Count == 0 ? null : await ResolveAsync(fullReference, cancellationToken).ConfigureAwait(false);
if (tagId is { } id)
@@ -182,7 +192,7 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource
StoredSample? p = reader.IsDBNull(1) ? null : ReadStored(reader, 1);
StoredSample? n = reader.IsDBNull(8) ? null : ReadStored(reader, 8);
var serverTime = p is { } ps && ps.TimeUtc == t ? SqlHistorianConnection.GetUtc(reader, 7) : now;
byTicks[t.Ticks] = AtTimeValue(t, p, n, serverTime);
byTicks[t.Ticks] = AtTimeValue(t, p, n, serverTime, now);
}
}
@@ -206,7 +216,7 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource
}
/// <summary>Applies the AtTime rules (see <see cref="ReadAtTimeAsync"/>) to one requested time.</summary>
private static DataValueSnapshot AtTimeValue(DateTime t, StoredSample? prev, StoredSample? next, DateTime serverTime)
private static DataValueSnapshot AtTimeValue(DateTime t, StoredSample? prev, StoredSample? next, DateTime serverTime, DateTime now)
{
if (prev is not { } p)
return new DataValueSnapshot(null, SqlHistorianStatusCodes.BadNoData, t, serverTime);
@@ -215,7 +225,7 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource
// Inside an outage: report the outage's own Bad code, never an upgraded Uncertain.
if (!p.IsUsable)
return new DataValueSnapshot(null, p.Status & StatusCodeBitsMask, t, serverTime);
return new DataValueSnapshot(null, p.Status & StatusBits.CodeBitsMask, t, serverTime);
// Usable previous, Bad next: hold the previous value up to the outage; do not interpolate into it.
if (next is { IsUsable: false })
@@ -234,6 +244,11 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource
return new DataValueSnapshot(value, HistorianInfoBits.WithInterpolated(status), t, serverTime);
}
// Beyond the last sample AND beyond now: the value is held into the future, so it is not known to be Good.
if (next is null && t > now)
return new DataValueSnapshot(p.Value.ToObject(),
HistorianInfoBits.WithInterpolated(SqlHistorianStatusCodes.UncertainDataSubNormal), t, serverTime);
// Stepped (non-numeric kind, or after the last sample): hold the previous value and its status.
return new DataValueSnapshot(p.Value.ToObject(), HistorianInfoBits.WithInterpolated(p.Status), t, serverTime);
}
@@ -252,6 +267,10 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource
public async Task<HistoricalEventsResult> ReadEventsAsync(
string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents, CancellationToken cancellationToken)
{
// No stored event can carry a source longer than the column (the alarm writer dead-letters it rather than
// truncating), so an over-long filter matches nothing — never a truncated prefix of someone else's source.
if (sourceName is not null && sourceName.Length > SqlServerTagCatalog.MaxNameLength)
return new HistoricalEventsResult([], null);
try
{
await _bootstrap.EnsureAsync(cancellationToken).ConfigureAwait(false);
@@ -269,7 +288,7 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource
SqlHistorianConnection.AddDateTime2(cmd, "@lo", lo);
SqlHistorianConnection.AddDateTime2(cmd, "@hi", hi);
if (sourceName is not null)
cmd.Parameters.Add(new SqlParameter("@source", SqlDbType.NVarChar, 400) { Value = sourceName });
cmd.Parameters.Add(new SqlParameter("@source", SqlDbType.NVarChar, SqlServerTagCatalog.MaxNameLength) { Value = sourceName });
var events = new List<HistoricalEvent>();
byte[]? continuation = null;
@@ -349,7 +368,7 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource
try
{
var now = DateTime.UtcNow;
var now = _time.GetUtcNow().UtcDateTime;
var rows = new Dictionary<long, BucketRow>();
var tagId = bucketCount == 0 ? null : await ResolveAsync(fullReference, cancellationToken).ConfigureAwait(false);
if (tagId is { } id)
@@ -492,21 +511,29 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource
private readonly record struct StoredSample(DateTime TimeUtc, HistorianValue Value, uint Status)
{
/// <summary>Severity Good or Uncertain, and a value present (an Empty sample is an outage marker).</summary>
public bool IsUsable => (Status & 0x80000000u) == 0 && Value.Kind != HistorianValueKind.Empty;
public bool IsUsable => !StatusBits.IsBad(Status) && Value.Kind != HistorianValueKind.Empty;
/// <summary>Usable with severity Good.</summary>
public bool IsGood => IsUsable && (Status & 0xC0000000u) == 0;
public bool IsGood => IsUsable && StatusBits.IsGood(Status);
}
/// <summary>Reads <c>SourceTimeUtc, ValueKind, NumValue, IntValue, StrValue, StatusCode</c> starting at <paramref name="o"/>.</summary>
private static StoredSample ReadStored(SqlDataReader r, int o) => new(
SqlHistorianConnection.GetUtc(r, o),
SampleRowMapper.FromColumns(
/// <summary>
/// Reads <c>SourceTimeUtc, ValueKind, NumValue, IntValue, StrValue, StatusCode</c> starting at <paramref name="o"/>.
/// An Empty (no-value) row whose stored status is not Bad is clamped to <c>BadNoData</c> — defence in depth for
/// rows written before the writer forced every Empty sample Bad.
/// </summary>
private static StoredSample ReadStored(SqlDataReader r, int o)
{
var value = SampleRowMapper.FromColumns(
r.GetByte(o + 1),
r.IsDBNull(o + 2) ? null : r.GetDouble(o + 2),
r.IsDBNull(o + 3) ? null : r.GetInt64(o + 3),
r.IsDBNull(o + 4) ? null : r.GetString(o + 4)),
SampleRowMapper.FromDbStatus(r.GetInt32(o + 5)));
r.IsDBNull(o + 4) ? null : r.GetString(o + 4));
var status = SampleRowMapper.FromDbStatus(r.GetInt32(o + 5));
if (value.Kind == HistorianValueKind.Empty && !StatusBits.IsBad(status))
status = SqlHistorianStatusCodes.BadNoData;
return new StoredSample(SqlHistorianConnection.GetUtc(r, o), value, status);
}
/// <summary>Read-only tag lookup (never creates a row); runs the schema bootstrap on a cache miss.</summary>
private Task<int?> ResolveAsync(string name, CancellationToken ct) => _catalog.TryGetTagIdAsync(name, ct);
@@ -6,7 +6,10 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer;
/// <summary>Status codes owned by the updater (named consts so <c>StatusCodeParityTests</c> sees them).</summary>
internal static class SqlHistorianUpdateStatusCodes
{
/// <summary>A value that cannot be stored (Double NaN / ±Infinity).</summary>
/// <summary>
/// A value that cannot be stored: a Double NaN / ±Infinity, an Empty (null) value with a non-Bad status, or a
/// tag name longer than <see cref="SqlServerTagCatalog.MaxNameLength"/>.
/// </summary>
public const uint BadInvalidArgument = 0x80AB0000u;
}
@@ -19,7 +22,13 @@ internal static class SqlHistorianUpdateStatusCodes
/// DeleteAtTime — the first occurrence reports the real outcome, later ones <c>BadNoEntryExists</c>.</para>
/// <para><b>Invalid items.</b> A null or <see cref="DateTime.MinValue"/> timestamp is
/// <c>BadInvalidTimestamp</c>. A Double NaN/±Infinity cannot live in the SQL <c>float</c> column and is
/// rejected with <c>BadInvalidArgument</c> (not silently stored as Empty). Other items still proceed.</para>
/// rejected with <c>BadInvalidArgument</c> (not silently stored as Empty). An Empty (null) value is only
/// accepted with a Bad status (an operator-authored outage marker); with a Good or Uncertain status it is
/// <c>BadInvalidArgument</c>, because a stored sample is never served better than "no value". Other items
/// still proceed.</para>
/// <para><b>Tag names over <see cref="SqlServerTagCatalog.MaxNameLength"/> characters</b> are refused, never
/// truncated: every item of the call (and DeleteRaw's single result) reports <c>BadInvalidArgument</c> and nothing
/// is written.</para>
/// <para><b>Unknown tag.</b> Insert/Update create it via <see cref="SqlServerTagCatalog"/>; Replace and the
/// deletes report <c>BadNoEntryExists</c> (DeleteRaw: <c>BadNoData</c>) without creating anything.</para>
/// </summary>
@@ -44,6 +53,11 @@ internal sealed class SqlServerHistorianDataUpdater : IHistorianDataUpdater
ArgumentNullException.ThrowIfNull(tag);
ArgumentNullException.ThrowIfNull(values);
var results = new uint[values.Count];
if (!SqlServerTagCatalog.IsValidName(tag))
{
Array.Fill(results, SqlHistorianUpdateStatusCodes.BadInvalidArgument);
return results;
}
var winner = new Dictionary<long, int>(); // tick -> index of the item actually sent
for (var i = 0; i < values.Count; i++)
{
@@ -53,7 +67,8 @@ internal sealed class SqlServerHistorianDataUpdater : IHistorianDataUpdater
results[i] = SqlHistorianStatusCodes.BadInvalidTimestamp;
continue;
}
if (v.Value.Kind == HistorianValueKind.Double && !double.IsFinite(v.Value.Number))
if ((v.Value.Kind == HistorianValueKind.Double && !double.IsFinite(v.Value.Number))
|| (v.Value.Kind == HistorianValueKind.Empty && !StatusBits.IsBad(v.StatusCode)))
{
results[i] = SqlHistorianUpdateStatusCodes.BadInvalidArgument;
continue;
@@ -71,7 +86,7 @@ internal sealed class SqlServerHistorianDataUpdater : IHistorianDataUpdater
var tagId = mode == HistoryUpdateMode.Replace
? await _tags.TryGetTagIdAsync(tag, ct).ConfigureAwait(false)
: await _tags.GetOrCreateTagIdAsync(tag, DataTypeFor(values[send[0]].Value.Kind), ct).ConfigureAwait(false);
: await _tags.GetOrCreateTagIdAsync(tag, SqlHistorianDataTypes.FromKind(values[send[0]].Value.Kind), ct).ConfigureAwait(false);
var outcome = tagId is null
? new Dictionary<long, uint>()
@@ -98,6 +113,7 @@ internal sealed class SqlServerHistorianDataUpdater : IHistorianDataUpdater
public async Task<uint> DeleteRawAsync(string tag, DateTime startUtc, DateTime endUtc, CancellationToken ct)
{
ArgumentNullException.ThrowIfNull(tag);
if (!SqlServerTagCatalog.IsValidName(tag)) return SqlHistorianUpdateStatusCodes.BadInvalidArgument;
var tagId = await _tags.TryGetTagIdAsync(tag, ct).ConfigureAwait(false);
if (tagId is null) return SqlHistorianStatusCodes.BadNoData;
await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false);
@@ -121,6 +137,11 @@ internal sealed class SqlServerHistorianDataUpdater : IHistorianDataUpdater
ArgumentNullException.ThrowIfNull(tag);
ArgumentNullException.ThrowIfNull(timesUtc);
var results = new uint[timesUtc.Count];
if (!SqlServerTagCatalog.IsValidName(tag))
{
Array.Fill(results, SqlHistorianUpdateStatusCodes.BadInvalidArgument);
return results;
}
var first = new Dictionary<long, int>();
for (var i = 0; i < timesUtc.Count; i++)
{
@@ -177,14 +198,4 @@ internal sealed class SqlServerHistorianDataUpdater : IHistorianDataUpdater
await tx.CommitAsync(ct).ConfigureAwait(false);
return outcome;
}
// T9 owns the canonical HistorianValueKind -> Tag.DataType mapping; a later fix wave will dedupe the two.
private static byte DataTypeFor(HistorianValueKind kind) => (byte)(kind switch
{
HistorianValueKind.Int64 => DriverDataType.Int64,
HistorianValueKind.Boolean => DriverDataType.Boolean,
HistorianValueKind.String => DriverDataType.String,
HistorianValueKind.DateTime => DriverDataType.DateTime,
_ => DriverDataType.Float64,
});
}
@@ -1,3 +1,4 @@
using System.Collections.Frozen;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
@@ -11,23 +12,26 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer;
/// <see cref="HistorianValueKind"/> and HistoryUpdate. All seams share ONE connection helper, ONE schema
/// bootstrapper and ONE tag catalog, so the schema is ensured once and tag ids are cached once.
/// </summary>
/// <remarks>
/// The shared state is built on the first <c>Create*</c> call, not in the constructor: the constructor does no
/// 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 bootstrapper logs through.
/// </remarks>
public sealed class SqlServerHistorianProvider : IHistorianBackendProvider
{
private static readonly IReadOnlySet<HistorianValueKind> Kinds =
new HashSet<HistorianValueKind>(Enum.GetValues<HistorianValueKind>());
private static readonly FrozenSet<HistorianValueKind> Kinds =
Enum.GetValues<HistorianValueKind>().ToFrozenSet();
private readonly SqlHistorianConnection _conn;
private readonly SqlServerSchemaBootstrapper _bootstrap;
private readonly SqlServerTagCatalog _catalog;
private readonly SqlServerHistorianOptions _options;
private readonly object _initLock = new();
private Shared? _shared;
/// <summary>Creates the provider over validated options.</summary>
/// <summary>Creates the provider over the <c>ServerHistorian:SqlServer</c> options (validated elsewhere).</summary>
/// <param name="options">The <c>ServerHistorian:SqlServer</c> options.</param>
public SqlServerHistorianProvider(SqlServerHistorianOptions options)
{
ArgumentNullException.ThrowIfNull(options);
_conn = new SqlHistorianConnection(options);
_bootstrap = new SqlServerSchemaBootstrapper(_conn, options);
_catalog = new SqlServerTagCatalog(_conn, _bootstrap);
_options = options ?? throw new ArgumentNullException(nameof(options));
}
/// <inheritdoc />
@@ -36,26 +40,61 @@ public sealed class SqlServerHistorianProvider : IHistorianBackendProvider
/// <inheritdoc />
public IReadOnlySet<HistorianValueKind> SupportedKinds => Kinds;
private static ILogger Log<T>(IServiceProvider sp) =>
((sp.GetService(typeof(ILoggerFactory)) as ILoggerFactory) ?? NullLoggerFactory.Instance).CreateLogger<T>();
private static ILoggerFactory Loggers(IServiceProvider sp) =>
(sp.GetService(typeof(ILoggerFactory)) as ILoggerFactory) ?? NullLoggerFactory.Instance;
private Shared Get(IServiceProvider sp)
{
if (Volatile.Read(ref _shared) is { } built) return built;
lock (_initLock)
{
if (_shared is { } raced) return raced;
var conn = new SqlHistorianConnection(_options);
var bootstrap = new SqlServerSchemaBootstrapper(conn, _options,
Loggers(sp).CreateLogger<SqlServerSchemaBootstrapper>());
var shared = new Shared(conn, bootstrap, new SqlServerTagCatalog(conn, bootstrap));
Volatile.Write(ref _shared, shared);
return shared;
}
}
/// <inheritdoc />
public IHistorianDataSource CreateDataSource(IServiceProvider sp) =>
new SqlServerHistorianDataSource(_conn, _bootstrap, _catalog, Log<SqlServerHistorianDataSource>(sp));
public IHistorianDataSource CreateDataSource(IServiceProvider sp)
{
var s = Get(sp);
return new SqlServerHistorianDataSource(s.Conn, s.Bootstrap, s.Catalog,
Loggers(sp).CreateLogger<SqlServerHistorianDataSource>());
}
/// <inheritdoc />
public IHistorianValueWriter CreateValueWriter(IServiceProvider sp) =>
new SqlServerHistorianValueWriter(_conn, _bootstrap, _catalog, Log<SqlServerHistorianValueWriter>(sp));
public IHistorianValueWriter CreateValueWriter(IServiceProvider sp)
{
var s = Get(sp);
return new SqlServerHistorianValueWriter(s.Conn, s.Bootstrap, s.Catalog,
Loggers(sp).CreateLogger<SqlServerHistorianValueWriter>());
}
/// <inheritdoc />
public IAlarmHistorianWriter CreateAlarmWriter(IServiceProvider sp) =>
new SqlServerAlarmHistorianWriter(_conn, _bootstrap, Log<SqlServerAlarmHistorianWriter>(sp));
public IAlarmHistorianWriter CreateAlarmWriter(IServiceProvider sp)
{
var s = Get(sp);
return new SqlServerAlarmHistorianWriter(s.Conn, s.Bootstrap,
Loggers(sp).CreateLogger<SqlServerAlarmHistorianWriter>());
}
/// <inheritdoc />
public IHistorianProvisioning CreateProvisioning(IServiceProvider sp) =>
new SqlServerTagProvisioner(_conn, _bootstrap, Log<SqlServerTagProvisioner>(sp));
public IHistorianProvisioning CreateProvisioning(IServiceProvider sp)
{
var s = Get(sp);
return new SqlServerTagProvisioner(s.Conn, s.Bootstrap, Loggers(sp).CreateLogger<SqlServerTagProvisioner>());
}
/// <inheritdoc />
public IHistorianDataUpdater? CreateUpdater(IServiceProvider sp) =>
new SqlServerHistorianDataUpdater(_conn, _catalog);
public IHistorianDataUpdater? CreateUpdater(IServiceProvider sp)
{
var s = Get(sp);
return new SqlServerHistorianDataUpdater(s.Conn, s.Catalog);
}
private sealed record Shared(SqlHistorianConnection Conn, SqlServerSchemaBootstrapper Bootstrap, SqlServerTagCatalog Catalog);
}
@@ -5,29 +5,6 @@ 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
@@ -36,8 +13,13 @@ internal static class SqlHistorianDataTypes
/// </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.
/// as <see cref="HistorianValueKind.Empty"/> so the sample's presence survives and the tag is not quarantined
/// for one bad reading. <b>An Empty sample is always Bad</b> (the quality contract: a stored sample is never
/// served better than "no value"): a non-finite Double whose status is not already Bad is stored with
/// <c>BadOutOfRange</c>; any other Empty with a non-Bad status is stored with <c>BadNoData</c>. An existing
/// Bad code (e.g. the comms-loss <c>BadNoCommunication</c> marker) is kept as it is.
/// <b>Tag names over <see cref="SqlServerTagCatalog.MaxNameLength"/> characters</b> are refused (the write
/// returns false, a per-tag failure for the recorder's #538 quarantine), never truncated onto another tag.
/// <b>Duplicate timestamps within a batch</b> collapse to the first (matching insert-if-absent across batches).
/// </remarks>
internal sealed class SqlServerHistorianValueWriter : IHistorianValueWriter
@@ -69,6 +51,12 @@ internal sealed class SqlServerHistorianValueWriter : IHistorianValueWriter
{
try
{
if (!SqlServerTagCatalog.IsValidName(tag))
{
_log.LogWarning("SQL historian live-value write refused: tag name is missing or longer than {Max} characters.",
SqlServerTagCatalog.MaxNameLength);
return false;
}
await _bootstrap.EnsureAsync(ct).ConfigureAwait(false);
if (values is null || values.Count == 0) return true;
@@ -108,8 +96,9 @@ internal sealed class SqlServerHistorianValueWriter : IHistorianValueWriter
}
/// <summary>
/// Normalises a batch for the TVP: null timestamp ⇒ now, non-finite doubles ⇒ Empty (status kept), and
/// duplicate timestamps collapsed to the first.
/// Normalises a batch for the TVP: null timestamp ⇒ now, non-finite doubles ⇒ Empty, every Empty forced Bad
/// (non-finite ⇒ <c>BadOutOfRange</c>, otherwise <c>BadNoData</c>, an existing Bad code 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>
@@ -124,9 +113,17 @@ internal sealed class SqlServerHistorianValueWriter : IHistorianValueWriter
t = t.Kind == DateTimeKind.Local ? t.ToUniversalTime() : DateTime.SpecifyKind(t, DateTimeKind.Utc);
if (!seen.Add(t.Ticks)) continue;
var value = v.Value;
var status = v.StatusCode;
if (value.Kind == HistorianValueKind.Double && !double.IsFinite(value.Number))
{
value = HistorianValue.Empty;
rows.Add(new SqlSampleRow(t, value, v.StatusCode));
if (!StatusBits.IsBad(status)) status = SqlHistorianStatusCodes.BadOutOfRange;
}
else if (value.Kind == HistorianValueKind.Empty && !StatusBits.IsBad(status))
{
status = SqlHistorianStatusCodes.BadNoData;
}
rows.Add(new SqlSampleRow(t, value, status));
}
return rows;
@@ -1,5 +1,6 @@
using System.Data;
using System.Reflection;
using System.Runtime.ExceptionServices;
using System.Text.RegularExpressions;
using Microsoft.Data.SqlClient;
using Microsoft.Extensions.Logging;
@@ -9,7 +10,10 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer;
/// <summary>
/// Lazily applies the embedded, idempotent schema scripts once per instance, under a SQL application lock
/// so concurrent nodes/instances serialise. A failure is NOT latched: the next <see cref="EnsureAsync"/> retries.
/// so concurrent nodes/instances serialise. A failure is NOT latched: an <see cref="EnsureAsync"/> after
/// <see cref="FailureCacheDuration"/> retries. Within that window the last failure is re-thrown without touching
/// SQL, so a missing schema (or a SQL outage) does not serialise every write, read and alarm event behind one
/// semaphore and one round-trip each.
/// With <c>AutoCreateSchema=false</c> it only verifies <c>MAX(Version) &gt;= 1</c>.
/// </summary>
internal sealed partial class SqlServerSchemaBootstrapper
@@ -20,21 +24,30 @@ internal sealed partial class SqlServerSchemaBootstrapper
private const string LockResource = "OtOpcUaHistorianSchema";
private const int LockTimeoutMs = 30_000;
/// <summary>How long the last bootstrap failure is re-thrown before SQL is tried again.</summary>
public static readonly TimeSpan FailureCacheDuration = TimeSpan.FromSeconds(5);
private readonly SqlHistorianConnection _conn;
private readonly bool _autoCreate;
private readonly ILogger _log;
private readonly TimeProvider _time;
private readonly SemaphoreSlim _gate = new(1, 1);
private volatile bool _ready;
private ExceptionDispatchInfo? _lastFailure; // guarded by _gate
private DateTimeOffset _retryAfter; // guarded by _gate
/// <summary>Creates the bootstrapper.</summary>
/// <param name="conn">Connection helper.</param>
/// <param name="options">Validated options.</param>
/// <param name="log">Optional logger.</param>
public SqlServerSchemaBootstrapper(SqlHistorianConnection conn, SqlServerHistorianOptions options, ILogger? log = null)
/// <param name="timeProvider">Clock for the failure cache; system by default.</param>
public SqlServerSchemaBootstrapper(SqlHistorianConnection conn, SqlServerHistorianOptions options, ILogger? log = null,
TimeProvider? timeProvider = null)
{
_conn = conn;
_autoCreate = options.AutoCreateSchema;
_log = log ?? NullLogger.Instance;
_time = timeProvider ?? TimeProvider.System;
}
/// <summary>Ensures the schema is ready; no-op after the first success.</summary>
@@ -48,10 +61,25 @@ internal sealed partial class SqlServerSchemaBootstrapper
try
{
if (_ready) return;
await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false);
if (_autoCreate) await ApplyAsync(conn, ct).ConfigureAwait(false);
else await VerifyAsync(conn, ct).ConfigureAwait(false);
_ready = true;
if (_lastFailure is { } cached && _time.GetUtcNow() < _retryAfter) cached.Throw();
try
{
await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false);
if (_autoCreate) await ApplyAsync(conn, ct).ConfigureAwait(false);
else await VerifyAsync(conn, ct).ConfigureAwait(false);
_ready = true;
_lastFailure = null;
}
catch (Exception ex) when (ex is not OperationCanceledException || !ct.IsCancellationRequested)
{
// A caller's own cancellation is not a backend failure and is not cached.
_lastFailure = ExceptionDispatchInfo.Capture(ex);
_retryAfter = _time.GetUtcNow() + FailureCacheDuration;
_log.LogWarning("Historian schema bootstrap for [{Schema}] failed ({Exception}: {Message}); " +
"historian calls fail fast for {Seconds} s before retrying.",
_conn.Schema, ex.GetType().Name, ex.Message, FailureCacheDuration.TotalSeconds);
throw;
}
}
finally
{
@@ -127,8 +155,13 @@ internal sealed partial class SqlServerSchemaBootstrapper
}
catch (Exception ex)
{
// Closing the session releases a Session-owned lock anyway.
_log.LogDebug(ex, "sp_releaseapplock failed; the lock dies with the connection");
// A pooled connection keeps its server session (sp_reset_connection does not release a Session-owned
// applock), so disposing this connection would return it to the pool STILL HOLDING the lock and block
// every other node's bootstrap until that pooled session dies. Clearing the pool marks it to be
// physically closed on dispose instead, which ends the session and releases the lock.
_log.LogWarning("sp_releaseapplock failed ({Exception}); discarding the connection pool so the " +
"session holding the historian schema applock is closed rather than pooled.", ex.GetType().Name);
SqlConnection.ClearPool(conn);
}
}
@@ -6,9 +6,16 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer;
/// <summary>
/// Tag name → <c>TagId</c> resolution shared by writes, reads, provisioning and updates. Caches positive
/// results only. Both methods run the schema bootstrap first, so callers need not.
/// <para><b>Names longer than <see cref="MaxNameLength"/></b> (the <c>Tag.Name</c> column width) are never
/// truncated: a sized <c>NVarChar(400)</c> parameter would silently cut them and resolve them onto whatever tag
/// owns the first 400 characters. <see cref="TryGetTagIdAsync"/> reports such a name as not found and
/// <see cref="GetOrCreateTagIdAsync"/> throws.</para>
/// </summary>
internal sealed class SqlServerTagCatalog
{
/// <summary>Width of the <c>Tag.Name</c> column; longer names are refused.</summary>
public const int MaxNameLength = 400;
private readonly SqlHistorianConnection _conn;
private readonly SqlServerSchemaBootstrapper _bootstrap;
private readonly ConcurrentDictionary<string, int> _cache = new(StringComparer.Ordinal);
@@ -22,12 +29,18 @@ internal sealed class SqlServerTagCatalog
_bootstrap = bootstrap;
}
/// <summary>Read-only lookup; never creates a row.</summary>
/// <summary>True when <paramref name="name"/> fits the <c>Tag.Name</c> column without truncation.</summary>
/// <param name="name">Historian tag name.</param>
/// <returns>Whether the name is non-null and at most <see cref="MaxNameLength"/> characters.</returns>
public static bool IsValidName(string? name) => name is not null && name.Length <= MaxNameLength;
/// <summary>Read-only lookup; never creates a row. A name over <see cref="MaxNameLength"/> is not found.</summary>
/// <param name="name">Historian tag name.</param>
/// <param name="ct">Cancellation.</param>
/// <returns>The TagId, or null when the tag does not exist.</returns>
public async Task<int?> TryGetTagIdAsync(string name, CancellationToken ct)
{
if (!IsValidName(name)) return null;
if (_cache.TryGetValue(name, out var cached)) return cached;
await _bootstrap.EnsureAsync(ct).ConfigureAwait(false);
await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false);
@@ -41,8 +54,11 @@ internal sealed class SqlServerTagCatalog
/// <param name="dataType">Initial <c>Tag.DataType</c> (only used on create; provisioning corrects it later).</param>
/// <param name="ct">Cancellation.</param>
/// <returns>The TagId.</returns>
/// <exception cref="ArgumentException">The name is longer than <see cref="MaxNameLength"/>.</exception>
public async Task<int> GetOrCreateTagIdAsync(string name, byte dataType, CancellationToken ct)
{
if (!IsValidName(name))
throw new ArgumentException($"Historian tag name exceeds {MaxNameLength} characters.", nameof(name));
if (_cache.TryGetValue(name, out var cached)) return cached;
await _bootstrap.EnsureAsync(ct).ConfigureAwait(false);
await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false);
@@ -52,7 +68,7 @@ internal sealed class SqlServerTagCatalog
try
{
await using var cmd = _conn.CreateCommand(conn, sql);
cmd.Parameters.Add(new SqlParameter("@name", System.Data.SqlDbType.NVarChar, 400) { Value = name });
cmd.Parameters.Add(new SqlParameter("@name", System.Data.SqlDbType.NVarChar, MaxNameLength) { Value = name });
cmd.Parameters.Add(new SqlParameter("@dt", System.Data.SqlDbType.TinyInt) { Value = dataType });
await cmd.ExecuteNonQueryAsync(ct).ConfigureAwait(false);
}
@@ -66,14 +82,10 @@ internal sealed class SqlServerTagCatalog
return id;
}
/// <summary>Drops a cached id (e.g. after a tag row is deleted).</summary>
/// <param name="name">Historian tag name.</param>
public void Invalidate(string name) => _cache.TryRemove(name, out _);
private async Task<int?> SelectIdAsync(SqlConnection conn, string name, CancellationToken ct)
{
await using var cmd = _conn.CreateCommand(conn, $"SELECT TagId FROM {_conn.Q("Tag")} WHERE Name = @name");
cmd.Parameters.Add(new SqlParameter("@name", System.Data.SqlDbType.NVarChar, 400) { Value = name });
cmd.Parameters.Add(new SqlParameter("@name", System.Data.SqlDbType.NVarChar, MaxNameLength) { Value = name });
var o = await cmd.ExecuteScalarAsync(ct).ConfigureAwait(false);
return o is null or DBNull ? null : Convert.ToInt32(o);
}
@@ -116,6 +116,19 @@ public sealed class AtTimeReadTests(SqlHistorianFixture fx)
s.StatusCode.ShouldBe(GoodInterpolated);
}
[Fact]
public async Task A_time_after_now_stepped_from_the_last_sample_is_uncertain()
{
var db = await ReadTestDb.CreateAsync(fx);
await db.SeedAAsync(); // last sample +40 = 50, Good
var src = db.Source(new FixedClock(new DateTimeOffset(At(42))));
var s = (await src.ReadAtTimeAsync("A", [At(41), At(45)], Ct)).Samples;
s[0].Value.ShouldBe(50d);
s[0].StatusCode.ShouldBe(GoodInterpolated); // after the last sample but not after now: held Good
s[1].Value.ShouldBe(50d);
s[1].StatusCode.ShouldBe(UncertainInterpolated); // after now: the value is not known to still hold
}
[Fact]
public async Task An_uncertain_neighbour_makes_the_interpolation_uncertain()
{
@@ -0,0 +1,9 @@
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests;
/// <summary>A settable clock for the data source's "now" and the bootstrapper's failure cache.</summary>
internal sealed class FixedClock(DateTimeOffset now) : TimeProvider
{
public DateTimeOffset Now { get; set; } = now;
public override DateTimeOffset GetUtcNow() => Now;
}
@@ -0,0 +1,87 @@
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests;
/// <summary>
/// Names longer than the 400-character <c>Tag.Name</c> / <c>Event.SourceName</c> columns are refused, never
/// truncated. Each test seeds a series under the 400-character PREFIX of the long name: a sized
/// <c>NVarChar(400)</c> parameter would silently cut the long name to that prefix and hit the wrong series.
/// </summary>
[Collection(SqlHistorianCollection.Name)]
public sealed class LongNameTests(SqlHistorianFixture fx)
{
private static readonly CancellationToken Ct = CancellationToken.None;
private static DateTime At(double s) => ReadTestDb.At(s);
private const uint BadInvalidArgument = 0x80AB0000u;
private const uint BadNoData = 0x809B0000u;
private static readonly string Prefix = new('p', 400);
private static readonly string Long = Prefix + "-x";
[Fact]
public async Task The_writer_refuses_a_long_name_as_a_per_tag_failure_and_writes_nothing()
{
var db = await ReadTestDb.CreateAsync(fx);
await db.SeedAsync(Prefix, (At(0), HistorianValue.FromDouble(1), ReadTestDb.Good));
var w = new SqlServerHistorianValueWriter(db.Conn, db.Boot, new SqlServerTagCatalog(db.Conn, db.Boot));
(await w.WriteLiveValuesAsync(Long, [new(At(5), HistorianValue.FromDouble(2), 0u)], Ct)).ShouldBeFalse();
(await db.ScalarAsync($"SELECT COUNT(*) FROM {db.Conn.Q("Sample")}")).ShouldBe(1);
(await db.ScalarAsync($"SELECT COUNT(*) FROM {db.Conn.Q("Tag")}")).ShouldBe(1);
}
[Fact]
public async Task The_updater_answers_BadInvalidArgument_for_every_item_of_a_long_name()
{
var db = await ReadTestDb.CreateAsync(fx);
await db.SeedAsync(Prefix, (At(0), HistorianValue.FromDouble(1), ReadTestDb.Good));
var u = new SqlServerHistorianDataUpdater(db.Conn, new SqlServerTagCatalog(db.Conn, db.Boot));
(await u.UpdateDataAsync(Long, HistoryUpdateMode.Update,
[new(At(0), HistorianValue.FromDouble(9), 0u), new(At(1), HistorianValue.FromDouble(9), 0u)], Ct))
.ShouldBe([BadInvalidArgument, BadInvalidArgument]);
(await u.DeleteRawAsync(Long, At(-1), At(10), Ct)).ShouldBe(BadInvalidArgument);
(await u.DeleteAtTimeAsync(Long, [At(0)], Ct)).ShouldBe([BadInvalidArgument]);
// The prefix series is untouched: not replaced, not deleted.
var raw = (await db.Source().ReadRawAsync(Prefix, At(-1), At(10), 0, Ct)).Samples;
raw.Select(s => s.Value).ShouldBe([(object)1d]);
}
[Fact]
public async Task Reads_of_a_long_name_are_not_found_rather_than_the_prefix_series()
{
var db = await ReadTestDb.CreateAsync(fx);
await db.SeedAsync(Prefix, (At(0), HistorianValue.FromDouble(1), ReadTestDb.Good));
var src = db.Source();
(await src.ReadRawAsync(Long, At(-1), At(10), 0, Ct)).Samples.ShouldBeEmpty();
(await src.ReadAtTimeAsync(Long, [At(0)], Ct)).Samples.Single().StatusCode.ShouldBe(BadNoData);
(await src.ReadProcessedAsync(Long, At(-1), At(10), TimeSpan.Zero, HistoryAggregateType.Count, Ct))
.Samples.Single().StatusCode.ShouldBe(HistorianInfoBits.WithCalculated(BadNoData));
src.GetHealthSnapshot().TotalFailures.ShouldBe(0);
}
[Fact]
public async Task An_event_read_filtered_by_a_long_source_matches_nothing()
{
var db = await ReadTestDb.CreateAsync(fx);
await using (var c = await db.Conn.OpenAsync(Ct))
await using (var cmd = db.Conn.CreateCommand(c,
$"INSERT {db.Conn.Q("Event")} (EventId, SourceName, AlarmId, AlarmName, AlarmTypeName, Severity, EventKind, Message, " +
"UserName, Comment, EventTimeUtc, ReceivedTimeUtc) VALUES ('e1', @src, 'a', 'n', 'T', 500, 'Activated', NULL, NULL, NULL, @t, @t)"))
{
cmd.Parameters.AddWithValue("@src", Prefix);
SqlHistorianConnection.AddDateTime2(cmd, "@t", At(0));
await cmd.ExecuteNonQueryAsync(Ct);
}
var src = db.Source();
(await src.ReadEventsAsync(Prefix, At(-1), At(10), 0, Ct)).Events.Count.ShouldBe(1); // control
(await src.ReadEventsAsync(Long, At(-1), At(10), 0, Ct)).Events.ShouldBeEmpty();
}
}
@@ -0,0 +1,104 @@
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests;
/// <summary>
/// Final-review I-1: a stored sample is never served with a better quality than "no value". An Empty (null-value)
/// sample is always Bad — forced Bad by the value writer, refused non-Bad by the updater, and clamped to
/// <c>BadNoData</c> on read for rows written before either rule existed.
/// </summary>
[Collection(SqlHistorianCollection.Name)]
public sealed class QualityContractTests(SqlHistorianFixture fx)
{
private static readonly CancellationToken Ct = CancellationToken.None;
private static DateTime At(double s) => ReadTestDb.At(s);
private const uint BadOutOfRange = 0x803C0000u;
private const uint BadNoData = 0x809B0000u;
private const uint BadInvalidArgument = 0x80AB0000u;
private const uint BadNoCommunication = 0x80310000u;
private static bool IsBad(uint status) => (status & 0x80000000u) != 0;
private static SqlServerHistorianValueWriter Writer(ReadTestDb db) =>
new(db.Conn, db.Boot, new SqlServerTagCatalog(db.Conn, db.Boot));
private static SqlServerHistorianDataUpdater Updater(ReadTestDb db) =>
new(db.Conn, new SqlServerTagCatalog(db.Conn, db.Boot));
[Fact]
public async Task A_good_NaN_written_through_the_real_writer_reads_back_Bad_on_Raw_and_AtTime()
{
var db = await ReadTestDb.CreateAsync(fx);
(await Writer(db).WriteLiveValuesAsync("nan", [
new(At(0), HistorianValue.FromDouble(10), 0u),
new(At(10), HistorianValue.FromDouble(double.NaN), 0u),
new(At(20), HistorianValue.FromDouble(double.PositiveInfinity), 0x40000000u),
new(At(30), HistorianValue.FromDouble(double.NegativeInfinity), BadNoCommunication),
], Ct)).ShouldBeTrue();
var raw = (await db.Source().ReadRawAsync("nan", At(0), At(100), 0, Ct)).Samples;
raw.Count.ShouldBe(4);
raw[1].Value.ShouldBeNull();
raw[1].StatusCode.ShouldBe(BadOutOfRange); // Good NaN ⇒ stored Bad
raw[2].StatusCode.ShouldBe(BadOutOfRange); // Uncertain +Inf ⇒ stored Bad
raw[3].StatusCode.ShouldBe(BadNoCommunication); // an existing Bad code is kept
var at = (await db.Source().ReadAtTimeAsync("nan", [At(10), At(15)], Ct)).Samples;
at.ShouldAllBe(s => s.Value == null);
at.ShouldAllBe(s => IsBad(s.StatusCode));
at[1].StatusCode.ShouldBe(BadOutOfRange); // inside the "outage" the NaN opened
}
[Fact]
public async Task A_non_bad_Empty_written_through_the_real_writer_is_stored_BadNoData()
{
var db = await ReadTestDb.CreateAsync(fx);
(await Writer(db).WriteLiveValuesAsync("empty", [
new(At(0), HistorianValue.Empty, 0u),
new(At(1), HistorianValue.Empty, 0x40000000u),
new(At(2), HistorianValue.Empty, BadNoCommunication),
], Ct)).ShouldBeTrue();
// Read the stored column directly so the read-side clamp cannot mask a writer that forgot to force Bad.
(await db.ScalarAsync($"SELECT COUNT(*) FROM {db.Conn.Q("Sample")} WHERE StatusCode >= 0")).ShouldBe(0);
var raw = (await db.Source().ReadRawAsync("empty", At(0), At(10), 0, Ct)).Samples;
raw.Select(s => s.StatusCode).ShouldBe([BadNoData, BadNoData, BadNoCommunication]);
}
[Fact]
public async Task A_legacy_good_Empty_row_is_clamped_to_BadNoData_on_Raw_and_AtTime()
{
var db = await ReadTestDb.CreateAsync(fx);
// Seeded with plain INSERTs, bypassing the writer: what a pre-fix build could have stored.
await db.SeedAsync("legacy",
(At(0), HistorianValue.FromDouble(10), ReadTestDb.Good),
(At(10), HistorianValue.Empty, ReadTestDb.Good),
(At(20), HistorianValue.Empty, ReadTestDb.Uncertain),
(At(30), HistorianValue.FromDouble(40), ReadTestDb.Good));
var raw = (await db.Source().ReadRawAsync("legacy", At(0), At(100), 0, Ct)).Samples;
raw.Select(s => s.StatusCode).ShouldBe([ReadTestDb.Good, BadNoData, BadNoData, ReadTestDb.Good]);
var at = (await db.Source().ReadAtTimeAsync("legacy", [At(10), At(15), At(25)], Ct)).Samples;
at.ShouldAllBe(s => s.Value == null && s.StatusCode == BadNoData);
}
[Fact]
public async Task The_updater_refuses_a_non_bad_Empty_and_accepts_a_bad_one()
{
var db = await ReadTestDb.CreateAsync(fx);
var r = await Updater(db).UpdateDataAsync("u", HistoryUpdateMode.Insert, [
new(At(0), HistorianValue.Empty, 0u),
new(At(1), HistorianValue.Empty, 0x40000000u),
new(At(2), HistorianValue.Empty, BadNoCommunication),
new(At(3), HistorianValue.FromDouble(1), 0u),
], Ct);
r.ShouldBe([BadInvalidArgument, BadInvalidArgument, SqlHistorianStatusCodes.GoodEntryInserted,
SqlHistorianStatusCodes.GoodEntryInserted]);
(await db.Source().ReadRawAsync("u", At(0), At(10), 0, Ct)).Samples
.Select(s => s.StatusCode).ShouldBe([BadNoCommunication, 0u]);
}
}
@@ -42,8 +42,9 @@ internal sealed class ReadTestDb
return new ReadTestDb(o, conn, boot);
}
/// <summary>A fresh data source (fresh catalog cache) over this schema.</summary>
public SqlServerHistorianDataSource Source() => new(Conn, Boot, new SqlServerTagCatalog(Conn, Boot));
/// <summary>A fresh data source (fresh catalog cache) over this schema; <paramref name="clock"/> is its "now".</summary>
public SqlServerHistorianDataSource Source(TimeProvider? clock = null) =>
new(Conn, Boot, new SqlServerTagCatalog(Conn, Boot), timeProvider: clock);
/// <summary>Inserts a tag row and the given samples. ServerTimeUtc = source time + 1 s (so tests can tell them apart).</summary>
public async Task<int> SeedAsync(string tag, params (DateTime T, HistorianValue V, uint Status)[] samples)
@@ -237,7 +238,8 @@ public sealed class RawReadTests(SqlHistorianFixture fx)
ConnectionString = fx.ConnectionString, Schema = SqlHistorianFixture.UniqueSchema(), AutoCreateSchema = false,
};
var conn = new SqlHistorianConnection(o);
var boot = new SqlServerSchemaBootstrapper(conn, o);
var clock = new FixedClock(DateTimeOffset.UtcNow);
var boot = new SqlServerSchemaBootstrapper(conn, o, timeProvider: clock);
var src = new SqlServerHistorianDataSource(conn, boot, new SqlServerTagCatalog(conn, boot));
await Should.ThrowAsync<HistorianSchemaMissingException>(() => src.ReadRawAsync("A", At(0), At(1), 0, Ct));
@@ -251,6 +253,7 @@ public sealed class RawReadTests(SqlHistorianFixture fx)
// Another node creates the schema; the same data source then succeeds (bootstrap failure is not latched).
var creator = new SqlServerHistorianOptions { ConnectionString = o.ConnectionString, Schema = o.Schema };
await new SqlServerSchemaBootstrapper(new SqlHistorianConnection(creator), creator).EnsureAsync(Ct);
clock.Now += SqlServerSchemaBootstrapper.FailureCacheDuration; // past the 5 s failure cache (M-2)
(await src.ReadRawAsync("A", At(0), At(1), 0, Ct)).Samples.ShouldBeEmpty();
src.GetHealthSnapshot().ConsecutiveFailures.ShouldBe(0);
}
@@ -77,10 +77,37 @@ public sealed class SchemaBootstrapIntegrationTests(SqlHistorianFixture fx)
{
fx.SkipIfUnavailable();
var s = SqlHistorianFixture.UniqueSchema();
var checker = Boot(Opts(s, auto: false));
var clock = new FixedClock(DateTimeOffset.UtcNow);
var o = Opts(s, auto: false);
var checker = new SqlServerSchemaBootstrapper(new SqlHistorianConnection(o), o, timeProvider: clock);
await Should.ThrowAsync<HistorianSchemaMissingException>(() => checker.EnsureAsync(CancellationToken.None));
await Boot(Opts(s)).EnsureAsync(CancellationToken.None);
await checker.EnsureAsync(CancellationToken.None); // same instance retries
clock.Now += SqlServerSchemaBootstrapper.FailureCacheDuration;
await checker.EnsureAsync(CancellationToken.None); // same instance retries once the failure cache expires
}
/// <summary>
/// M-2: within <see cref="SqlServerSchemaBootstrapper.FailureCacheDuration"/> of a failure the bootstrapper
/// re-throws it without touching SQL — proven by creating the schema in between: a call that went to SQL
/// would now pass.
/// </summary>
[Fact]
public async Task A_bootstrap_failure_is_cached_for_five_seconds_then_retried()
{
fx.SkipIfUnavailable();
var s = SqlHistorianFixture.UniqueSchema();
var clock = new FixedClock(DateTimeOffset.UtcNow);
var o = Opts(s, auto: false);
var checker = new SqlServerSchemaBootstrapper(new SqlHistorianConnection(o), o, timeProvider: clock);
await Should.ThrowAsync<HistorianSchemaMissingException>(() => checker.EnsureAsync(CancellationToken.None));
await Boot(Opts(s)).EnsureAsync(CancellationToken.None); // the schema now exists
clock.Now += SqlServerSchemaBootstrapper.FailureCacheDuration - TimeSpan.FromTicks(1);
await Should.ThrowAsync<HistorianSchemaMissingException>(() => checker.EnsureAsync(CancellationToken.None));
clock.Now += TimeSpan.FromTicks(1);
await checker.EnsureAsync(CancellationToken.None);
}
[Fact]
@@ -89,7 +89,7 @@ public sealed class WriteTests(SqlHistorianFixture fx)
}
[Fact]
public async Task Non_finite_doubles_store_as_Empty_with_their_status_and_do_not_fail_the_write()
public async Task Non_finite_doubles_store_as_Empty_forced_Bad_and_do_not_fail_the_write()
{
fx.SkipIfUnavailable();
var (w, s) = Make();
@@ -101,7 +101,8 @@ public sealed class WriteTests(SqlHistorianFixture fx)
rows.Count.ShouldBe(3);
rows[0].Kind.ShouldBe((byte)HistorianValueKind.Empty);
rows[0].N.ShouldBeNull();
rows[0].Status.ShouldBe(unchecked((int)0x40000000u));
rows[0].Status.ShouldBe(unchecked((int)SqlHistorianStatusCodes.BadOutOfRange)); // Uncertain NaN ⇒ Bad (I-1)
rows[1].Status.ShouldBe(unchecked((int)SqlHistorianStatusCodes.BadOutOfRange)); // Good +Inf ⇒ Bad (I-1)
rows[2].N.ShouldBe(7);
}
@@ -32,6 +32,36 @@ public sealed class SqlServerHistorianProviderTests
p.CreateUpdater(sp).ShouldBeOfType<SqlServerHistorianDataUpdater>();
}
[Fact]
public void Supported_kinds_are_a_frozen_set() =>
Make().SupportedKinds.ShouldBeAssignableTo<System.Collections.Frozen.FrozenSet<HistorianValueKind>>();
/// <summary>
/// An invalid schema must not fail provider construction (the host resolves the provider even when no
/// consumer is enabled); it surfaces only when a seam is actually built.
/// </summary>
[Fact]
public void An_invalid_schema_fails_only_when_a_seam_is_built()
{
var p = new SqlServerHistorianProvider(new SqlServerHistorianOptions { ConnectionString = "x", Schema = "bad;schema" });
p.Name.ShouldBe("SqlServer");
Should.Throw<ArgumentException>(() => p.CreateDataSource(new NullSp()));
}
[Fact]
public void Every_seam_shares_one_connection_bootstrapper_and_catalog()
{
var sp = new NullSp();
var p = Make();
var a = (SqlServerHistorianDataSource)p.CreateDataSource(sp);
var b = (SqlServerHistorianDataSource)p.CreateDataSource(sp);
Field(a, "_bootstrap").ShouldBeSameAs(Field(b, "_bootstrap"));
Field(a, "_catalog").ShouldBeSameAs(Field(b, "_catalog"));
}
private static object? Field(object o, string name) =>
o.GetType().GetField(name, System.Reflection.BindingFlags.Instance | System.Reflection.BindingFlags.NonPublic)!.GetValue(o);
private sealed class NullSp : IServiceProvider
{
public object? GetService(Type serviceType) => null;
@@ -22,6 +22,23 @@ public sealed class StatusCodeOracleTests
((uint)Opc.Ua.AggregateBits.Interpolated).ShouldBe(HistorianInfoBits.HistorianInterpolated);
}
/// <summary>
/// A stored status that already carries a historian bit (a sample written via HistoryUpdate with the Calculated
/// bit, say) must come back with ONE historian bit, never both: the existing pair is masked off first. The SDK
/// agrees — <c>SetAggregateBits</c> replaces, it does not OR.
/// </summary>
[Fact]
public void Setting_one_historian_bit_clears_the_other()
{
var calculatedUncertain = SqlHistorianStatusCodes.UncertainDataSubNormal | 0x400u | HistorianInfoBits.HistorianCalculated;
var interpolated = HistorianInfoBits.WithInterpolated(calculatedUncertain);
interpolated.ShouldBe(SqlHistorianStatusCodes.UncertainDataSubNormal | 0x400u | HistorianInfoBits.HistorianInterpolated);
new Opc.Ua.StatusCode(interpolated).AggregateBits.ShouldBe(Opc.Ua.AggregateBits.Interpolated);
var interpolatedGood = 0x400u | HistorianInfoBits.HistorianInterpolated;
HistorianInfoBits.WithCalculated(interpolatedGood).ShouldBe(0x400u | HistorianInfoBits.HistorianCalculated);
}
[Fact]
public void Info_type_is_data_value() =>
HistorianInfoBits.InfoTypeDataValue.ShouldBe(0x400u);
@@ -9,18 +9,32 @@ public sealed class ValueWriterPrepareTests
private static readonly DateTime T = new(2026, 10, 9, 1, 2, 3, DateTimeKind.Utc);
[Fact]
public void Non_finite_doubles_become_Empty_keeping_status()
public void Non_finite_doubles_become_Empty_and_are_forced_Bad()
{
const uint status = 0x40000000u;
const uint uncertain = 0x40000000u;
const uint badNoCommunication = 0x80310000u;
var rows = SqlServerHistorianValueWriter.Prepare(
[
new(T, HistorianValue.FromDouble(double.NaN), status),
new(T, HistorianValue.FromDouble(double.NaN), uncertain),
new(T.AddTicks(1), HistorianValue.FromDouble(double.PositiveInfinity), 0),
new(T.AddTicks(2), HistorianValue.FromDouble(double.NegativeInfinity), 0),
new(T.AddTicks(2), HistorianValue.FromDouble(double.NegativeInfinity), badNoCommunication),
new(T.AddTicks(3), HistorianValue.FromDouble(1.5), 0),
], () => T);
rows.Select(r => r.Value.Kind).ShouldBe([HistorianValueKind.Empty, HistorianValueKind.Empty, HistorianValueKind.Empty, HistorianValueKind.Double]);
rows[0].StatusCode.ShouldBe(status);
rows.Select(r => r.StatusCode).ShouldBe([SqlHistorianStatusCodes.BadOutOfRange, SqlHistorianStatusCodes.BadOutOfRange,
badNoCommunication, 0u]);
}
[Fact]
public void A_non_bad_Empty_is_stored_BadNoData_and_a_bad_Empty_keeps_its_code()
{
var rows = SqlServerHistorianValueWriter.Prepare(
[
new(T, HistorianValue.Empty, 0),
new(T.AddTicks(1), HistorianValue.Empty, 0x40A40000u),
new(T.AddTicks(2), HistorianValue.Empty, 0x80310000u),
], () => T);
rows.Select(r => r.StatusCode).ShouldBe([SqlHistorianStatusCodes.BadNoData, SqlHistorianStatusCodes.BadNoData, 0x80310000u]);
}
[Fact]