dohertj2/lmxopcua
1
0
Fork 0
Code Issues 14 Pull Requests Actions Packages Projects Releases Wiki Activity

fix(historian): T9 review round 1 - dup-key retry counter, alarm Classify retries unknown, test hardening

Browse Source 
Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
This commit is contained in:
Joseph Doherty
2026-10-09 13:30:42 -04:00
parent bef47f89da
commit c2ab68ceb5
5 changed files with 71 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerAlarmHistorianWriter.cs
+30 -33
View File
@@ -14,12 +14,12 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer;
/// 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.
/// for data errors (truncation, conversion, null, range, constraint); unknown/non-SQL exceptions retry. 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];
// 2628/8152 string truncation, 515 null into NOT NULL, 547 constraint (non-dup-key), 245/8114 conversion, 220/8115 overflow, 242/241 datetime range.
private static readonly int[] DataErrorNumbers = [2628, 8152, 515, 547, 245, 8114, 220, 8115, 242, 241];
private readonly SqlHistorianConnection _conn;
private readonly SqlServerSchemaBootstrapper _bootstrap;
@@ -72,32 +72,30 @@ internal sealed class SqlServerAlarmHistorianWriter : IAlarmHistorianWriter
$"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
{
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;
}
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;
@@ -110,14 +108,13 @@ internal sealed class SqlServerAlarmHistorianWriter : IAlarmHistorianWriter
}
}
private static HistorianWriteOutcome Classify(Exception ex)
// Only a data error SQL Server will reject forever dead-letters; anything unknown or non-SQL retries (never lose alarm history).
internal 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;
if (ex is SqlException sql && sql.Errors.Cast<SqlError>().Any(x => Array.IndexOf(DataErrorNumbers, x.Number) >= 0))
return HistorianWriteOutcome.PermanentFail;
return HistorianWriteOutcome.RetryPlease;
}
// Same numeric Part 9 priorities the gateway path sends (AlarmEventMapper.ToNumericSeverity).
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerTagProvisioner.cs
+6
View File
@@ -24,6 +24,7 @@ internal sealed class SqlServerTagProvisioner : IHistorianProvisioning
private readonly SqlHistorianConnection _conn;
private readonly SqlServerSchemaBootstrapper _bootstrap;
private readonly ILogger _log;
private long _dupKeyRetries;
/// <summary>Creates the provisioner.</summary>
/// <param name="conn">Connection helper.</param>
@@ -36,6 +37,10 @@ internal sealed class SqlServerTagProvisioner : IHistorianProvisioning
_log = log ?? NullLogger.Instance;
}
/// <summary>Number of duplicate-key (2627/2601) retries so far. A correct HOLDLOCK MERGE never hits one under contention; tests assert 0.
/// Transient retries (deadlock victims etc.) are legitimate and deliberately not counted.</summary>
internal long DupKeyRetryCount => Interlocked.Read(ref _dupKeyRetries);
/// <inheritdoc/>
public async Task<HistorianProvisionResult> EnsureTagsAsync(IReadOnlyList<HistorianTagProvisionRequest> requests, CancellationToken ct)
{
@@ -106,6 +111,7 @@ internal sealed class SqlServerTagProvisioner : IHistorianProvisioning
catch (SqlException ex) when (attempt < MaxAttempts &&
(SqlHistorianConnection.IsDuplicateKey(ex) || SqlHistorianConnection.IsTransient(ex)))
{
if (SqlHistorianConnection.IsDuplicateKey(ex)) Interlocked.Increment(ref _dupKeyRetries);
await Task.Delay(20 * attempt, ct).ConfigureAwait(false);
}
catch (Exception ex)
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/ProvisioningTests.cs
+19 -5
View File
@@ -68,18 +68,32 @@ public sealed class ProvisioningTests(SqlHistorianFixture fx)
var s = SqlHistorianFixture.UniqueSchema();
var (p0, _) = Make(s);
await p0.EnsureTagsAsync([new("warm", DriverDataType.Int32, null, null)], CancellationToken.None); // bootstrap once
for (var round = 0; round < 10; round++)
for (var round = 0; round < 20; round++)
{
var reqs = Enumerable.Range(0, 60).Select(i => new HistorianTagProvisionRequest($"r{round}.t{i}", DriverDataType.Float64, "u", null)).ToList();
var rs = await Task.WhenAll(Enumerable.Range(0, 4).Select(_ => Task.Run(() => Make(s).P.EnsureTagsAsync(reqs, CancellationToken.None))));
foreach (var r in rs) r.ShouldBe(new HistorianProvisionResult(60, 60, 0, 0));
var ps = Enumerable.Range(0, 16).Select(_ => Make(s).P).ToList();
// Warm each instance (its own bootstrap-once) and release all together, else bootstrap serialisation staggers the race away.
foreach (var p in ps) await p.EnsureTagsAsync([new("warm", DriverDataType.Int32, null, null)], CancellationToken.None);
var gate = new TaskCompletionSource();
// Staggered windows over one pool: every worker's FIRST name differs, so a MERGE without HOLDLOCK cannot hide behind
// a lock convoy on a shared first key and the overlap really races (union per round = 6 names; the single/few-row shape is the one T8 proved races).
var tasks = ps.Select((p, k) => Task.Run(async () =>
{
var reqs = Enumerable.Range(k % 4, 3).Select(i => new HistorianTagProvisionRequest($"r{round}.t{i}", DriverDataType.Float64, "u", null)).ToList();
await gate.Task;
return await p.EnsureTagsAsync(reqs, CancellationToken.None);
})).ToList();
gate.SetResult();
var rs = await Task.WhenAll(tasks);
foreach (var r in rs) r.ShouldBe(new HistorianProvisionResult(3, 3, 0, 0));
// The production dup-key retry would mask a missing HOLDLOCK; HOLDLOCK must mean it never fires.
ps.Sum(p => p.DupKeyRetryCount).ShouldBe(0);
}
await using var c = new SqlConnection(fx.ConnectionString);
await c.OpenAsync();
await using var cmd = c.CreateCommand();
cmd.CommandText = $"SELECT COUNT(*) FROM [{s}].Tag";
Convert.ToInt32(await cmd.ExecuteScalarAsync()).ShouldBe(601);
Convert.ToInt32(await cmd.ExecuteScalarAsync()).ShouldBe(20 * 6 + 1);
}
[Fact]
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/WriteTests.cs
+8 -1
View File
@@ -57,6 +57,7 @@ public sealed class WriteTests(SqlHistorianFixture fx)
for (var i = 0; i < 6; i++) rows[i].T.Ticks.ShouldBe(vals[i].TimestampUtc!.Value.Ticks);
var back = rows.Select(r => SampleRowMapper.FromColumns(r.Kind, r.N, r.I, r.S)).ToList();
back.ShouldBe(vals.Select(v => v.Value).ToList());
(rows[5].N, rows[5].I, rows[5].S).ShouldBe((null, null, null));
rows[1].Status.ShouldBe(unchecked((int)0x40000000u));
SampleRowMapper.FromDbStatus(rows[5].Status).ShouldBe(0x80310000u);
}
@@ -148,7 +149,13 @@ public sealed class WriteTests(SqlHistorianFixture fx)
fx.SkipIfUnavailable();
var (w, s) = Make();
var vals = Enumerable.Range(0, 200).Select(i => new HistorizationValue(T0.AddMilliseconds(i), HistorianValue.FromDouble(i), 0)).ToArray();
var results = await Task.WhenAll(Enumerable.Range(0, 8).Select(_ => Task.Run(() => w.WriteLiveValuesAsync("tag.race", vals, CancellationToken.None))));
// A deadlock victim (1205) legitimately returns false and the recorder retries; model that, and require convergence.
var results = await Task.WhenAll(Enumerable.Range(0, 8).Select(_ => Task.Run(async () =>
{
for (var i = 0; i < 5; i++)
if (await w.WriteLiveValuesAsync("tag.race", vals, CancellationToken.None)) return true;
return false;
})));
results.ShouldAllBe(r => r);
(await RowsAsync(s, "tag.race")).Count.ShouldBe(200);
}
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests/AlarmEventIdTests.cs
+8
View File
@@ -44,4 +44,12 @@ public sealed class AlarmEventIdTests
})
SqlServerAlarmHistorianWriter.ComputeEventId(other).ShouldNotBe(basis);
}
[Theory]
[InlineData(typeof(InvalidOperationException))]
[InlineData(typeof(ArgumentException))]
[InlineData(typeof(FormatException))]
public void Non_SQL_exceptions_retry_rather_than_dead_letter(Type type) =>
SqlServerAlarmHistorianWriter.Classify((Exception)Activator.CreateInstance(type)!)
.ShouldBe(HistorianWriteOutcome.RetryPlease);
}