diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerAlarmHistorianWriter.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerAlarmHistorianWriter.cs index 4d7bdb1f..a4f63652 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerAlarmHistorianWriter.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerAlarmHistorianWriter.cs @@ -14,12 +14,12 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer; /// deterministic , 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. /// 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().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().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). diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerTagProvisioner.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerTagProvisioner.cs index c0582b88..4ca9f501 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerTagProvisioner.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerTagProvisioner.cs @@ -24,6 +24,7 @@ internal sealed class SqlServerTagProvisioner : IHistorianProvisioning private readonly SqlHistorianConnection _conn; private readonly SqlServerSchemaBootstrapper _bootstrap; private readonly ILogger _log; + private long _dupKeyRetries; /// Creates the provisioner. /// Connection helper. @@ -36,6 +37,10 @@ internal sealed class SqlServerTagProvisioner : IHistorianProvisioning _log = log ?? NullLogger.Instance; } + /// 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. + internal long DupKeyRetryCount => Interlocked.Read(ref _dupKeyRetries); + /// public async Task EnsureTagsAsync(IReadOnlyList 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) diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/ProvisioningTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/ProvisioningTests.cs index 5ddff410..904457ba 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/ProvisioningTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/ProvisioningTests.cs @@ -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] diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/WriteTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/WriteTests.cs index 946ac321..5a2aa6ad 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/WriteTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/WriteTests.cs @@ -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); } diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests/AlarmEventIdTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests/AlarmEventIdTests.cs index 69d67dc0..03b12e75 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests/AlarmEventIdTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests/AlarmEventIdTests.cs @@ -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); }