perf(historian): batch alarm-history writes; pipeline gateway SendEvent (lmxopcua#746)
All three alarm-history writers inserted one event per round trip, serially (SqlServer also opened a pooled connection per event), so a 10000-event alarm storm drained at 10000 x RTT while the backlog sat in LocalDb. - Sql.Common.AlarmBatchWrite: one statement per chunk; a chunk that fails PermanentFail-class is re-sent event by event so only the poison event dead-letters; a transient failure retries the chunk; a cancellation never dead-letters. - SqlServer: multi-row insert-if-absent over a VALUES set (150 events per statement, under the 2100-parameter limit), duplicate ids in a batch written once; a duplicate-key race isolates per event, where it is Ack. No new table type, so no schema migration (the review suggested a TVP; the VALUES form gets the same one-round-trip-per-chunk without one). - QuestDb: multi-row INSERTs of 500 events; pre-1970 events still dead-letter unsent. - Gateway: no batch RPC exists, so sends are pipelined, at most 8 in flight, outcomes kept in input order. Rebased from the unpushed remed/lane-H2 branch (originally authored 2026-10-22 in session_01Ub9hGmYLQvn9XETzw7CMsT) onto current master. Claude-Session: https://claude.ai/code/session_01GT67AEHd2Acv1YPwGfXEoV
This commit is contained in:
+40
-13
@@ -19,8 +19,9 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway;
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// Each event is sent individually so one poison event cannot fail the whole batch: a permanent
|
||||
/// failure on event N is dead-lettered while its siblings continue. Outcomes are returned in
|
||||
/// input order, one per event; an empty batch yields an empty list with no gateway call.
|
||||
/// failure on event N is dead-lettered while its siblings continue. The sends are pipelined — at most
|
||||
/// <see cref="MaxSendsInFlight"/> in flight — because the gateway has no batch <c>SendEvent</c> RPC yet.
|
||||
/// Outcomes are returned in input order, one per event; an empty batch yields an empty list with no gateway call.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Outcome mapping.</b> Success (or store-forward-queued) ack ⇒ <see cref="HistorianWriteOutcome.Ack"/>.
|
||||
@@ -63,23 +64,49 @@ public sealed class GatewayAlarmHistorianWriter : IAlarmHistorianWriter, IDispos
|
||||
|
||||
var outcomes = new HistorianWriteOutcome[batch.Count];
|
||||
|
||||
// The gateway has no batch SendEvent RPC, so the sends are PIPELINED: up to MaxSendsInFlight at once, each still
|
||||
// its own RPC with its own outcome (so one poison event cannot fail the batch), outcomes kept in input order. A
|
||||
// sequential drain cost one round trip per event — an alarm storm over a WAN took minutes.
|
||||
using var window = new SemaphoreSlim(MaxSendsInFlight);
|
||||
var sends = new Task[batch.Count];
|
||||
for (var i = 0; i < batch.Count; i++)
|
||||
{
|
||||
if (cancellationToken.IsCancellationRequested)
|
||||
{
|
||||
// Shutdown mid-drain: short-circuit the remaining events to RetryPlease rather than
|
||||
// calling the gateway with a cancelled token. They stay queued for retry next startup
|
||||
// — a cancellation must NEVER dead-letter an in-flight event (silent data loss).
|
||||
outcomes[i] = HistorianWriteOutcome.RetryPlease;
|
||||
continue;
|
||||
}
|
||||
|
||||
outcomes[i] = await SendOneAsync(batch[i], cancellationToken).ConfigureAwait(false);
|
||||
var index = i;
|
||||
sends[i] = SendInWindowAsync(index);
|
||||
}
|
||||
|
||||
// Every task completes (none throws), so the window is never disposed under a waiter.
|
||||
await Task.WhenAll(sends).ConfigureAwait(false);
|
||||
return outcomes;
|
||||
|
||||
async Task SendInWindowAsync(int index)
|
||||
{
|
||||
try
|
||||
{
|
||||
await window.WaitAsync(cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
// Shutdown mid-drain: an event that never started is RetryPlease (stays queued for next startup) — a
|
||||
// cancellation must NEVER dead-letter an in-flight event (silent data loss).
|
||||
outcomes[index] = HistorianWriteOutcome.RetryPlease;
|
||||
return;
|
||||
}
|
||||
try
|
||||
{
|
||||
outcomes[index] = cancellationToken.IsCancellationRequested
|
||||
? HistorianWriteOutcome.RetryPlease
|
||||
: await SendOneAsync(batch[index], cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
finally
|
||||
{
|
||||
window.Release();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>The most <c>SendEvent</c> RPCs one batch has in flight at once (bounded pipelining until a batch RPC exists).</summary>
|
||||
public const int MaxSendsInFlight = 8;
|
||||
|
||||
private async Task<HistorianWriteOutcome> SendOneAsync(AlarmHistorianEvent evt, CancellationToken cancellationToken)
|
||||
{
|
||||
try
|
||||
|
||||
+44
-33
@@ -12,7 +12,8 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb;
|
||||
/// <see cref="AlarmEventIds.ComputeEventId"/> and is inserted through the table's <c>DEDUP UPSERT KEYS(ts, event_id)</c>
|
||||
/// (T0 Q3 confirmed VARCHAR works as a key), so the store-and-forward drain's at-least-once redelivery, and both
|
||||
/// pair nodes writing the same event, converge on one row (the redelivered row is identical except
|
||||
/// <c>received_utc</c>). Outcomes: Ack on insert; RetryPlease on transient / unclassified errors and on
|
||||
/// <c>received_utc</c>). A batch goes in multi-row INSERTs of <see cref="ChunkRows"/> events (<see cref="AlarmBatchWrite"/>:
|
||||
/// one round trip per chunk, a poison event isolated by re-sending its chunk event by event). Outcomes: Ack on insert; RetryPlease on transient / unclassified errors and on
|
||||
/// cancellation; PermanentFail only for data the server will always reject (a time before 1970, which is not a
|
||||
/// storable designated timestamp, or a server message naming an inconvertible / out-of-range value). Never throws.
|
||||
/// </summary>
|
||||
@@ -60,36 +61,55 @@ internal sealed class QuestDbAlarmHistorianWriter : IAlarmHistorianWriter
|
||||
return outcomes;
|
||||
}
|
||||
|
||||
var wrote = false;
|
||||
// A time before 1970 is not a storable designated timestamp: dead-lettered without being sent (one such row would
|
||||
// fail its whole statement). The rest go in multi-row INSERTs of ChunkRows events (AlarmBatchWrite): one round trip
|
||||
// per chunk; a chunk's data error is re-sent event by event so only the poison event dead-letters.
|
||||
var slots = new HistorianWriteOutcome?[batch.Count];
|
||||
for (var i = 0; i < batch.Count; i++)
|
||||
{
|
||||
if (cancellationToken.IsCancellationRequested) { outcomes[i] = HistorianWriteOutcome.RetryPlease; continue; }
|
||||
outcomes[i] = await WriteOneAsync(batch[i], cancellationToken).ConfigureAwait(false);
|
||||
wrote |= outcomes[i] == HistorianWriteOutcome.Ack;
|
||||
if (QuestDbConnection.IsStorable(batch[i].TimestampUtc)) continue;
|
||||
_log.LogWarning("QuestDB alarm event dropped: its time is before 1970-01-01 and cannot be stored.");
|
||||
slots[i] = HistorianWriteOutcome.PermanentFail;
|
||||
}
|
||||
|
||||
if (wrote)
|
||||
var outcomes2 = await AlarmBatchWrite.RunAsync(slots, ChunkRows,
|
||||
(indexes, ct) => InsertChunkAsync(batch, indexes, ct),
|
||||
Classify,
|
||||
(ex, outcome) => _log.LogWarning("QuestDB alarm event write failed ({Exception}); outcome {Outcome}.", ex.GetType().Name, outcome),
|
||||
cancellationToken).ConfigureAwait(false);
|
||||
|
||||
if (outcomes2.Any(o => o == HistorianWriteOutcome.Ack))
|
||||
{
|
||||
try { await _wal.WaitAsync(_conn.EventTable, cancellationToken).ConfigureAwait(false); }
|
||||
catch (Exception ex) { _log.LogDebug("QuestDB alarm visibility wait failed ({Exception}).", ex.GetType().Name); }
|
||||
}
|
||||
return outcomes;
|
||||
return outcomes2;
|
||||
}
|
||||
|
||||
private async Task<HistorianWriteOutcome> WriteOneAsync(AlarmHistorianEvent e, CancellationToken ct)
|
||||
{
|
||||
try
|
||||
{
|
||||
if (!QuestDbConnection.IsStorable(e.TimestampUtc))
|
||||
{
|
||||
_log.LogWarning("QuestDB alarm event dropped: its time is before 1970-01-01 and cannot be stored.");
|
||||
return HistorianWriteOutcome.PermanentFail;
|
||||
}
|
||||
/// <summary>Events per INSERT: 12 bound parameters each.</summary>
|
||||
internal const int ChunkRows = 500;
|
||||
|
||||
await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false);
|
||||
await using var cmd = _conn.CreateCommand(conn,
|
||||
$"INSERT INTO {_conn.EventTable} (ts, event_id, source, alarm_id, alarm_name, alarm_type_name, severity, event_kind, message, user_name, comment, received_utc) " +
|
||||
"VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12)");
|
||||
private long _statements;
|
||||
|
||||
/// <summary>INSERT statements sent so far (lifetime) — test/diagnostic.</summary>
|
||||
internal long Statements => Interlocked.Read(ref _statements);
|
||||
|
||||
private async Task InsertChunkAsync(IReadOnlyList<AlarmHistorianEvent> batch, IReadOnlyList<int> indexes, CancellationToken ct)
|
||||
{
|
||||
var sql = new System.Text.StringBuilder(
|
||||
"INSERT INTO " + _conn.EventTable + " (ts, event_id, source, alarm_id, alarm_name, alarm_type_name, severity, event_kind, message, user_name, comment, received_utc) VALUES ");
|
||||
await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false);
|
||||
await using var cmd = _conn.CreateCommand(conn, "");
|
||||
var received = _clock.GetUtcNow().UtcDateTime;
|
||||
var n = 1;
|
||||
for (var k = 0; k < indexes.Count; k++)
|
||||
{
|
||||
if (k > 0) sql.Append(',');
|
||||
sql.Append('(');
|
||||
for (var p = 0; p < 12; p++) sql.Append(p == 0 ? "$" : ",$").Append(n + p);
|
||||
sql.Append(')');
|
||||
n += 12;
|
||||
var e = batch[indexes[k]];
|
||||
QuestDbConnection.BindTimestamp(cmd, e.TimestampUtc);
|
||||
QuestDbConnection.BindText(cmd, AlarmEventIds.ComputeEventId(e));
|
||||
QuestDbConnection.BindText(cmd, e.EquipmentPath ?? "");
|
||||
@@ -101,20 +121,11 @@ internal sealed class QuestDbAlarmHistorianWriter : IAlarmHistorianWriter
|
||||
QuestDbConnection.BindText(cmd, e.Message);
|
||||
QuestDbConnection.BindText(cmd, e.User);
|
||||
QuestDbConnection.BindText(cmd, e.Comment);
|
||||
QuestDbConnection.BindTimestamp(cmd, _clock.GetUtcNow().UtcDateTime);
|
||||
await cmd.ExecuteNonQueryAsync(ct).ConfigureAwait(false);
|
||||
return HistorianWriteOutcome.Ack;
|
||||
}
|
||||
catch (OperationCanceledException) when (ct.IsCancellationRequested)
|
||||
{
|
||||
return HistorianWriteOutcome.RetryPlease;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
var outcome = Classify(ex);
|
||||
_log.LogWarning("QuestDB alarm event write failed ({Exception}); outcome {Outcome}.", ex.GetType().Name, outcome);
|
||||
return outcome;
|
||||
QuestDbConnection.BindTimestamp(cmd, received);
|
||||
}
|
||||
cmd.CommandText = sql.ToString();
|
||||
Interlocked.Increment(ref _statements);
|
||||
await cmd.ExecuteNonQueryAsync(ct).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
/// <summary>Classifies a write exception as retryable or a permanent dead-letter failure.</summary>
|
||||
|
||||
@@ -0,0 +1,87 @@
|
||||
using ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common;
|
||||
|
||||
/// <summary>
|
||||
/// The batched alarm-history write both SQL backends share: events go to the database in chunks of one statement each
|
||||
/// (one round trip per chunk, not per event), and the per-event outcomes the store-and-forward drain needs are kept:
|
||||
/// <list type="bullet">
|
||||
/// <item>a chunk that succeeds is Ack for every event in it (each was inserted or already present — the writes are
|
||||
/// insert-if-absent / deduplicated by event id);</item>
|
||||
/// <item>a chunk failing <see cref="HistorianWriteOutcome.RetryPlease"/>-class (transient: the database is down or
|
||||
/// busy) is RetryPlease for all its events — re-sending them one by one during an outage would only multiply
|
||||
/// the round trips;</item>
|
||||
/// <item>a chunk failing <see cref="HistorianWriteOutcome.PermanentFail"/>-class (a data error, or anything the
|
||||
/// caller's classifier wants isolated) is re-sent ONE EVENT AT A TIME, so only the poison event dead-letters and
|
||||
/// its siblings still Ack;</item>
|
||||
/// <item>a cancellation of the caller's token makes every event not yet decided RetryPlease (never a dead letter).</item>
|
||||
/// </list>
|
||||
/// </summary>
|
||||
public static class AlarmBatchWrite
|
||||
{
|
||||
/// <summary>Writes the undecided events of a batch and returns one outcome per event.</summary>
|
||||
/// <param name="outcomes">
|
||||
/// One slot per batch event; a non-null slot is already decided (e.g. a pre-1970 time on QuestDB) and is not sent.
|
||||
/// Filled in place; every slot is non-null on return.
|
||||
/// </param>
|
||||
/// <param name="chunkSize">Events per statement (at least 1).</param>
|
||||
/// <param name="writeChunk">Writes the events at the given batch indexes in ONE statement; throws on failure.</param>
|
||||
/// <param name="classify">Maps a write failure to the outcome it implies (see the summary for multi-event chunks).</param>
|
||||
/// <param name="onFailure">Optional diagnostic hook per failure (the exception and its classified outcome).</param>
|
||||
/// <param name="ct">The drain's cancellation token.</param>
|
||||
/// <returns>The filled outcomes.</returns>
|
||||
public static async Task<HistorianWriteOutcome[]> RunAsync(
|
||||
HistorianWriteOutcome?[] outcomes, int chunkSize,
|
||||
Func<IReadOnlyList<int>, CancellationToken, Task> writeChunk,
|
||||
Func<Exception, HistorianWriteOutcome> classify,
|
||||
Action<Exception, HistorianWriteOutcome>? onFailure,
|
||||
CancellationToken ct)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(outcomes);
|
||||
ArgumentNullException.ThrowIfNull(writeChunk);
|
||||
ArgumentNullException.ThrowIfNull(classify);
|
||||
chunkSize = Math.Max(1, chunkSize);
|
||||
|
||||
var pending = new List<int>(outcomes.Length);
|
||||
for (var i = 0; i < outcomes.Length; i++)
|
||||
if (outcomes[i] is null) pending.Add(i);
|
||||
|
||||
for (var start = 0; start < pending.Count; start += chunkSize)
|
||||
{
|
||||
var chunk = pending.GetRange(start, Math.Min(chunkSize, pending.Count - start));
|
||||
var outcome = await TryAsync(chunk).ConfigureAwait(false);
|
||||
if (outcome == HistorianWriteOutcome.PermanentFail && chunk.Count > 1)
|
||||
{
|
||||
// Isolate the poison event: one statement per event, so the siblings still land.
|
||||
foreach (var index in chunk)
|
||||
outcomes[index] = await TryAsync([index]).ConfigureAwait(false);
|
||||
continue;
|
||||
}
|
||||
foreach (var index in chunk) outcomes[index] ??= outcome;
|
||||
}
|
||||
|
||||
var result = new HistorianWriteOutcome[outcomes.Length];
|
||||
for (var i = 0; i < outcomes.Length; i++) result[i] = outcomes[i] ?? HistorianWriteOutcome.RetryPlease;
|
||||
return result;
|
||||
|
||||
async Task<HistorianWriteOutcome> TryAsync(IReadOnlyList<int> indexes)
|
||||
{
|
||||
if (ct.IsCancellationRequested) return HistorianWriteOutcome.RetryPlease;
|
||||
try
|
||||
{
|
||||
await writeChunk(indexes, ct).ConfigureAwait(false);
|
||||
return HistorianWriteOutcome.Ack;
|
||||
}
|
||||
catch (OperationCanceledException) when (ct.IsCancellationRequested)
|
||||
{
|
||||
return HistorianWriteOutcome.RetryPlease; // shutdown: never a dead letter
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
var o = classify(ex);
|
||||
onFailure?.Invoke(ex, o);
|
||||
return o;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
+70
-41
@@ -9,9 +9,11 @@ using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common;
|
||||
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="AlarmEventIds.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;
|
||||
/// <see cref="IAlarmHistorianWriter"/> over the SQL <c>Event</c> table. Insert-if-absent keyed by a deterministic
|
||||
/// <see cref="AlarmEventIds.ComputeEventId"/>, so the store-and-forward drain's at-least-once redelivery (and both
|
||||
/// pair nodes writing the same event) converge on one row. A batch is written in multi-row statements of
|
||||
/// <see cref="ChunkRows"/> events (<see cref="AlarmBatchWrite"/> — one round trip per chunk, a poison event isolated by
|
||||
/// re-sending its chunk event by event). 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, constraint); unknown/non-SQL exceptions retry. Never throws.
|
||||
/// </summary>
|
||||
@@ -52,58 +54,85 @@ internal sealed class SqlServerAlarmHistorianWriter : IAlarmHistorianWriter
|
||||
return outcomes;
|
||||
}
|
||||
|
||||
// One statement per chunk (AlarmBatchWrite): a multi-row insert-if-absent over a VALUES set, so a drain batch costs
|
||||
// ceil(n / ChunkRows) round trips instead of n. A chunk's data error (or a duplicate-key race with the peer node) is
|
||||
// re-sent event by event, where a duplicate key means "already there" (Ack) and a data error dead-letters only
|
||||
// its own event. Duplicate event ids WITHIN the batch are written once and share that write's outcome (two rows
|
||||
// with one id in one statement would violate the key).
|
||||
var representative = new Dictionary<string, int>(StringComparer.Ordinal);
|
||||
var ids = new string[batch.Count];
|
||||
var slots = new HistorianWriteOutcome?[batch.Count];
|
||||
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);
|
||||
ids[i] = AlarmEventIds.ComputeEventId(batch[i]);
|
||||
if (!representative.TryAdd(ids[i], i)) slots[i] = HistorianWriteOutcome.Ack; // not sent; takes its twin's outcome below
|
||||
}
|
||||
|
||||
var written = await AlarmBatchWrite.RunAsync(slots, ChunkRows,
|
||||
(indexes, ct) => WriteChunkAsync(batch, ids, indexes, ct),
|
||||
ClassifyChunk,
|
||||
(ex, outcome) => _log.LogWarning("SQL alarm event write failed ({Exception}); outcome {Outcome}.", ex.GetType().Name, outcome),
|
||||
cancellationToken).ConfigureAwait(false);
|
||||
for (var i = 0; i < batch.Count; i++) outcomes[i] = written[representative[ids[i]]];
|
||||
return outcomes;
|
||||
}
|
||||
|
||||
private async Task<HistorianWriteOutcome> WriteOneAsync(AlarmHistorianEvent e, CancellationToken ct)
|
||||
/// <summary>Events per statement: 11 parameters each, under SQL Server's 2,100-parameter limit.</summary>
|
||||
internal const int ChunkRows = 150;
|
||||
|
||||
private long _statements;
|
||||
|
||||
/// <summary>Statements sent so far (lifetime) — test/diagnostic.</summary>
|
||||
internal long Statements => Interlocked.Read(ref _statements);
|
||||
|
||||
// A duplicate key on a MULTI-row statement is a race with another writer for one of its events: isolate (re-send
|
||||
// per event, where it is an Ack). A single row never reaches here with one (WriteChunkAsync treats it as present).
|
||||
private static HistorianWriteOutcome ClassifyChunk(Exception ex) =>
|
||||
SqlHistorianConnection.IsDuplicateKey(ex) ? HistorianWriteOutcome.PermanentFail : Classify(ex);
|
||||
|
||||
private async Task WriteChunkAsync(IReadOnlyList<AlarmHistorianEvent> batch, string[] ids, IReadOnlyList<int> indexes,
|
||||
CancellationToken ct)
|
||||
{
|
||||
// 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 values = new System.Text.StringBuilder();
|
||||
for (var k = 0; k < indexes.Count; k++)
|
||||
{
|
||||
if (k > 0) values.Append(',');
|
||||
values.Append("(@id").Append(k).Append(",@src").Append(k).Append(",@aid").Append(k).Append(",@an").Append(k)
|
||||
.Append(",@at").Append(k).Append(",@sev").Append(k).Append(",@ek").Append(k).Append(",@msg").Append(k)
|
||||
.Append(",@usr").Append(k).Append(",@cmt").Append(k).Append(",@t").Append(k).Append(')');
|
||||
}
|
||||
var sql =
|
||||
"INSERT INTO " + _conn.Q("Event") + " (EventId, SourceName, AlarmId, AlarmName, AlarmTypeName, Severity, EventKind, Message, UserName, Comment, EventTimeUtc, ReceivedTimeUtc) " +
|
||||
"SELECT v.EventId, v.SourceName, v.AlarmId, v.AlarmName, v.AlarmTypeName, v.Severity, v.EventKind, v.Message, v.UserName, v.Comment, v.EventTimeUtc, SYSUTCDATETIME() " +
|
||||
"FROM (VALUES " + values + ") v (EventId, SourceName, AlarmId, AlarmName, AlarmTypeName, Severity, EventKind, Message, UserName, Comment, EventTimeUtc) " +
|
||||
"WHERE NOT EXISTS (SELECT 1 FROM " + _conn.Q("Event") + " x WITH (UPDLOCK, HOLDLOCK) WHERE x.EventId = v.EventId)";
|
||||
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);
|
||||
for (var k = 0; k < indexes.Count; k++)
|
||||
{
|
||||
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 = AlarmEventIds.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 = AlarmEventIds.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;
|
||||
var e = batch[indexes[k]];
|
||||
cmd.Parameters.Add(new SqlParameter("@id" + k, SqlDbType.NVarChar, 64) { Value = ids[indexes[k]] });
|
||||
cmd.Parameters.Add(new SqlParameter("@src" + k, SqlDbType.NVarChar, -1) { Value = e.EquipmentPath ?? "" });
|
||||
cmd.Parameters.Add(new SqlParameter("@aid" + k, SqlDbType.NVarChar, -1) { Value = e.AlarmId ?? "" });
|
||||
cmd.Parameters.Add(new SqlParameter("@an" + k, SqlDbType.NVarChar, -1) { Value = e.AlarmName ?? "" });
|
||||
cmd.Parameters.Add(new SqlParameter("@at" + k, SqlDbType.NVarChar, -1) { Value = e.AlarmTypeName ?? "" });
|
||||
cmd.Parameters.Add(new SqlParameter("@sev" + k, SqlDbType.Int) { Value = AlarmEventIds.ToNumericSeverity(e.Severity) });
|
||||
cmd.Parameters.Add(new SqlParameter("@ek" + k, SqlDbType.NVarChar, -1) { Value = e.EventKind ?? "" });
|
||||
cmd.Parameters.Add(new SqlParameter("@msg" + k, SqlDbType.NVarChar, -1) { Value = (object?)e.Message ?? DBNull.Value });
|
||||
cmd.Parameters.Add(new SqlParameter("@usr" + k, SqlDbType.NVarChar, -1) { Value = (object?)e.User ?? DBNull.Value });
|
||||
cmd.Parameters.Add(new SqlParameter("@cmt" + k, SqlDbType.NVarChar, -1) { Value = (object?)e.Comment ?? DBNull.Value });
|
||||
SqlHistorianConnection.AddDateTime2(cmd, "@t" + k, e.TimestampUtc);
|
||||
}
|
||||
Interlocked.Increment(ref _statements);
|
||||
await cmd.ExecuteNonQueryAsync(ct).ConfigureAwait(false);
|
||||
}
|
||||
catch (OperationCanceledException) when (ct.IsCancellationRequested)
|
||||
// Single-row statement: a duplicate-key error means the row IS there (a concurrent writer won).
|
||||
catch (SqlException ex) when (indexes.Count == 1 && SqlHistorianConnection.IsDuplicateKey(ex))
|
||||
{
|
||||
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;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user