From 74afc3f90a83e125bb507385b140b1177c1b8bf9 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 16:54:37 -0400 Subject: [PATCH] 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 --- docs/Historian.md | 8 +- .../GatewayAlarmHistorianWriter.cs | 53 +++++++-- .../QuestDbAlarmHistorianWriter.cs | 77 ++++++------ .../AlarmBatchWrite.cs | 87 ++++++++++++++ .../SqlServerAlarmHistorianWriter.cs | 111 +++++++++++------- .../FakeHistorianGatewayClient.cs | 8 ++ .../GatewayAlarmWriterPipeliningTests.cs | 101 ++++++++++++++++ .../AlarmWriterTests.cs | 20 ++++ .../AlarmBatchWriteTests.cs | 102 ++++++++++++++++ .../AlarmWriterTests.cs | 25 ++++ 10 files changed, 503 insertions(+), 89 deletions(-) create mode 100644 src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common/AlarmBatchWrite.cs create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayAlarmWriterPipeliningTests.cs create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests/AlarmBatchWriteTests.cs diff --git a/docs/Historian.md b/docs/Historian.md index 29823488a..8878bffe5 100644 --- a/docs/Historian.md +++ b/docs/Historian.md @@ -80,7 +80,7 @@ consumes the historian. The existing enable gates (`ServerHistorian:Enabled`, `A | Processed aggregates | Twelve: the ten per-sample ones (Average, Count, Minimum, Maximum, Range, Start, End, Delta, Min/MaxActualTime) evaluated from raw reads with the SQL backends' evaluator — identical results (#681) — plus TimeAverage and Total from AVEVA's native modes | The seventeen in the [aggregate table](#aggregates-processed-reads) (incl. Part 13 time-weighted, #671) | The same seventeen, identical results | | HistoryUpdate | No (`BadHistoryOperationUnsupported`) | Yes (Insert / Replace / Update / DeleteRaw / DeleteAtTime) | Yes — soft delete (tombstones), see [below](#historyupdate-soft-delete) | | Modified-value history (`IsReadModified`, #668) | No (`BadHistoryOperationUnsupported`) | Yes — `SampleModified`, atomic with the change | Yes — `{prefix}sample_modified`, written before the change (orphan-on-crash) | -| Alarm history | gateway `SendEvent` | `Event` table (insert-if-absent) | `{prefix}event` table (dedup upsert on `(ts, event_id)`) | +| Alarm history | gateway `SendEvent`, one RPC per event, up to 8 in flight (no batch RPC yet) | `Event` table (insert-if-absent), multi-row statements of 150 events | `{prefix}event` table (dedup upsert on `(ts, event_id)`), multi-row INSERTs of 500 events | | Provisioning (`EnsureTags`) | numeric types only; others skipped | `MERGE` upsert of every request, all types | Upsert into `{prefix}tag`, all types | | Connection | `Endpoint` + `ApiKey` + TLS keys (top-level `ServerHistorian:*`) | `ServerHistorian:SqlServer:*` | `ServerHistorian:QuestDb:*` | @@ -1411,7 +1411,11 @@ with the first HistoryUpdate after it. `INSERT … SELECT … WHERE NOT EXISTS` through the `SampleRow` TVP. A null timestamp is stamped now; `ServerTimeUtc = SYSUTCDATETIME()`. It returns `false` (never throws) on a SQL failure so the recorder's retry/quarantine owns it. Semantics: [delivery semantics](#outbox-format-v2-and-rollback). -- **Alarm events** (`IAlarmHistorianWriter`): per event, insert-if-absent keyed by a deterministic `EventId` = lowercase-hex +- **Alarm events** (`IAlarmHistorianWriter`): a drain batch is written in **multi-row statements of 150 events** (one round + trip per chunk — it used to be a fresh connection and a single-row statement per event, so a 10 000-event alarm storm cost + 10 000 round trips). A chunk that fails with a data error (or a duplicate-key race with the peer node) is re-sent **event by + event**, so only the poison event dead-letters and its siblings still `Ack`; a transient failure retries the whole chunk. The + shared orchestration is `Sql.Common.AlarmBatchWrite` (QuestDB uses it with 500-event INSERTs). Each event: insert-if-absent keyed by a deterministic `EventId` = lowercase-hex SHA-256 over the length-prefixed fields `EquipmentPath, AlarmId, EventKind, TimestampUtc.Ticks, User, Comment, Message`. Outcomes: `Ack` on insert or already-present (including a concurrent writer winning a 2627/2601 race); `RetryPlease` on transient SQL errors (connection, timeout, deadlock 1205 …), bootstrap failures, cancellation and any unclassified error; diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayAlarmHistorianWriter.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayAlarmHistorianWriter.cs index 1817c8a34..a3e14e7ec 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayAlarmHistorianWriter.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayAlarmHistorianWriter.cs @@ -19,8 +19,9 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway; /// /// /// 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 +/// in flight — because the gateway has no batch SendEvent RPC yet. +/// Outcomes are returned in input order, one per event; an empty batch yields an empty list with no gateway call. /// /// /// Outcome mapping. Success (or store-forward-queued) 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(); + } + } } + /// The most SendEvent RPCs one batch has in flight at once (bounded pipelining until a batch RPC exists). + public const int MaxSendsInFlight = 8; + private async Task SendOneAsync(AlarmHistorianEvent evt, CancellationToken cancellationToken) { try diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbAlarmHistorianWriter.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbAlarmHistorianWriter.cs index fe3505236..e61eb1346 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbAlarmHistorianWriter.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbAlarmHistorianWriter.cs @@ -12,7 +12,8 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb; /// and is inserted through the table's DEDUP UPSERT KEYS(ts, event_id) /// (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 -/// received_utc). Outcomes: Ack on insert; RetryPlease on transient / unclassified errors and on +/// received_utc). A batch goes in multi-row INSERTs of events (: +/// 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. /// @@ -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 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; - } + /// Events per INSERT: 12 bound parameters each. + 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; + + /// INSERT statements sent so far (lifetime) — test/diagnostic. + internal long Statements => Interlocked.Read(ref _statements); + + private async Task InsertChunkAsync(IReadOnlyList batch, IReadOnlyList 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); } /// Classifies a write exception as retryable or a permanent dead-letter failure. diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common/AlarmBatchWrite.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common/AlarmBatchWrite.cs new file mode 100644 index 000000000..448caf183 --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common/AlarmBatchWrite.cs @@ -0,0 +1,87 @@ +using ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common; + +/// +/// 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: +/// +/// 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); +/// a chunk failing -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; +/// a chunk failing -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; +/// a cancellation of the caller's token makes every event not yet decided RetryPlease (never a dead letter). +/// +/// +public static class AlarmBatchWrite +{ + /// Writes the undecided events of a batch and returns one outcome per event. + /// + /// 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. + /// + /// Events per statement (at least 1). + /// Writes the events at the given batch indexes in ONE statement; throws on failure. + /// Maps a write failure to the outcome it implies (see the summary for multi-event chunks). + /// Optional diagnostic hook per failure (the exception and its classified outcome). + /// The drain's cancellation token. + /// The filled outcomes. + public static async Task RunAsync( + HistorianWriteOutcome?[] outcomes, int chunkSize, + Func, CancellationToken, Task> writeChunk, + Func classify, + Action? onFailure, + CancellationToken ct) + { + ArgumentNullException.ThrowIfNull(outcomes); + ArgumentNullException.ThrowIfNull(writeChunk); + ArgumentNullException.ThrowIfNull(classify); + chunkSize = Math.Max(1, chunkSize); + + var pending = new List(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 TryAsync(IReadOnlyList 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; + } + } + } +} 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 5ed366934..583194388 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 @@ -9,9 +9,11 @@ using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common; namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer; /// -/// over the SQL Event table. Per event: insert-if-absent keyed by a -/// 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; +/// over the SQL Event table. Insert-if-absent keyed by a deterministic +/// , 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 +/// events ( — 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. /// @@ -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(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 WriteOneAsync(AlarmHistorianEvent e, CancellationToken ct) + /// Events per statement: 11 parameters each, under SQL Server's 2,100-parameter limit. + internal const int ChunkRows = 150; + + private long _statements; + + /// Statements sent so far (lifetime) — test/diagnostic. + 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 batch, string[] ids, IReadOnlyList 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; } } diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/FakeHistorianGatewayClient.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/FakeHistorianGatewayClient.cs index 744da36d2..57fb1ef80 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/FakeHistorianGatewayClient.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/FakeHistorianGatewayClient.cs @@ -227,8 +227,16 @@ public sealed class FakeHistorianGatewayClient : IHistorianGatewayClient : Task.FromResult(WriteLiveValuesResult); } + /// When set, SendEvent is served by it (an asynchronous, observable gateway — concurrency tests). + public Func>? SendEventSource; + public Task SendEventAsync(HistorianEvent evt, CancellationToken ct) { + if (SendEventSource is { } source) + { + Interlocked.Increment(ref SendEventCallCount); + return source(evt, ct); + } LastSendEvent = evt; SendEventCallCount++; return SendEventThrows is not null diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayAlarmWriterPipeliningTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayAlarmWriterPipeliningTests.cs new file mode 100644 index 000000000..c1ffcf683 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayAlarmWriterPipeliningTests.cs @@ -0,0 +1,101 @@ +using Grpc.Core; +using Microsoft.Extensions.Logging.Abstractions; +using Shouldly; +using Xunit; +using ZB.MOM.WW.HistorianGateway.Contracts.Grpc; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests; + +/// +/// The gateway has no batch SendEvent RPC, so the alarm writer pipelines its per-event sends — at most +/// at once — instead of one round trip after another. Every +/// event still gets its own outcome, in input order. +/// +public sealed class GatewayAlarmWriterPipeliningTests +{ + private static CancellationToken Ct => TestContext.Current.CancellationToken; + + private static AlarmHistorianEvent Evt(int i) => new( + $"alarm-{i}", "Area/Pump", "N", "LimitAlarm", AlarmSeverity.High, "Activated", $"m{i}", "u", null, + new DateTime(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc).AddSeconds(i)); + + [Fact] + public async Task Sends_overlap_up_to_the_bound_and_outcomes_keep_input_order() + { + var inFlight = 0; + var maxInFlight = 0; + var fake = new FakeHistorianGatewayClient + { + SendEventSource = async (e, ct) => + { + var now = Interlocked.Increment(ref inFlight); + InterlockedMax(ref maxInFlight, now); + await Task.Delay(20, ct); + Interlocked.Decrement(ref inFlight); + // Odd messages are soft failures: their outcome must land on their own slot. + return e.Properties.TryGetValue("Message", out var m) && int.Parse(m[1..]) % 2 == 1 + ? new WriteAck { Success = false } + : new WriteAck { Success = true }; + }, + }; + using var writer = new GatewayAlarmHistorianWriter(fake, NullLogger.Instance); + + var outcomes = await writer.WriteBatchAsync(Enumerable.Range(0, 40).Select(Evt).ToList(), Ct); + + outcomes.Count.ShouldBe(40); + for (var i = 0; i < 40; i++) + outcomes[i].ShouldBe(i % 2 == 1 ? HistorianWriteOutcome.RetryPlease : HistorianWriteOutcome.Ack, $"event {i}"); + maxInFlight.ShouldBeGreaterThan(1, "the sends must overlap — a sequential drain is 40 round trips"); + maxInFlight.ShouldBeLessThanOrEqualTo(GatewayAlarmHistorianWriter.MaxSendsInFlight); + fake.SendEventCallCount.ShouldBe(40); + } + + [Fact] + public async Task A_cancellation_mid_batch_sends_nothing_more_and_dead_letters_nothing() + { + using var cts = new CancellationTokenSource(); + var sent = 0; + var fake = new FakeHistorianGatewayClient + { + SendEventSource = async (_, ct) => + { + if (Interlocked.Increment(ref sent) == 3) await cts.CancelAsync(); + await Task.Delay(10, ct); + return new WriteAck { Success = true }; + }, + }; + using var writer = new GatewayAlarmHistorianWriter(fake, NullLogger.Instance); + + var outcomes = await writer.WriteBatchAsync(Enumerable.Range(0, 40).Select(Evt).ToList(), cts.Token); + + outcomes.Count.ShouldBe(40); + outcomes.ShouldNotContain(HistorianWriteOutcome.PermanentFail); + outcomes.Count(o => o == HistorianWriteOutcome.RetryPlease).ShouldBeGreaterThan(0); + sent.ShouldBeLessThanOrEqualTo(GatewayAlarmHistorianWriter.MaxSendsInFlight + 2, "no new send starts after the cancel"); + } + + [Fact] + public async Task A_poison_event_dead_letters_alone() + { + var fake = new FakeHistorianGatewayClient + { + SendEventSource = (e, _) => e.Properties.TryGetValue("Message", out var m) && m == "m5" + ? Task.FromException(new RpcException(new Status(StatusCode.InvalidArgument, "bad"))) + : Task.FromResult(new WriteAck { Success = true }), + }; + using var writer = new GatewayAlarmHistorianWriter(fake, NullLogger.Instance); + + var outcomes = await writer.WriteBatchAsync(Enumerable.Range(0, 10).Select(Evt).ToList(), Ct); + + outcomes[5].ShouldBe(HistorianWriteOutcome.PermanentFail); + outcomes.Where((_, i) => i != 5).ShouldAllBe(o => o == HistorianWriteOutcome.Ack); + } + + private static void InterlockedMax(ref int target, int value) + { + int current; + while ((current = Volatile.Read(ref target)) < value && Interlocked.CompareExchange(ref target, value, current) != current) { } + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/AlarmWriterTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/AlarmWriterTests.cs index 032ffa869..7299b2c87 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/AlarmWriterTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/AlarmWriterTests.cs @@ -32,6 +32,26 @@ public sealed class AlarmWriterTests(QuestDbFixture fx) return Convert.ToInt64(await cmd.ExecuteScalarAsync()); } + [Fact] + public async Task A_batch_is_one_insert_per_chunk_and_a_redelivery_converges() + { + fx.SkipIfUnavailable(); + var (w, c) = Make(); + var events = Enumerable.Range(0, 1200).Select(i => Ev(kind: $"k{i}", at: T.AddSeconds(i))).ToList(); + events.Insert(3, new AlarmHistorianEvent("alarm-1", "Plant/Line/Eq", "HiTemp", "LimitAlarm", AlarmSeverity.High, + "old", "msg", "user", null, new DateTime(1969, 12, 31, 0, 0, 0, DateTimeKind.Utc))); // pre-1970: never sent + + var r = await w.WriteBatchAsync(events, CancellationToken.None); + r[3].ShouldBe(HistorianWriteOutcome.PermanentFail); + r.Where((_, i) => i != 3).ShouldAllBe(o => o == HistorianWriteOutcome.Ack); + w.Statements.ShouldBe(3, "1,200 events = 500 + 500 + 200, not 1,200 round trips"); + (await CountAsync(c)).ShouldBe(1200); + + (await w.WriteBatchAsync(events.Where((_, i) => i != 3).ToList(), CancellationToken.None)) + .ShouldAllBe(o => o == HistorianWriteOutcome.Ack); + (await CountAsync(c)).ShouldBe(1200, "DEDUP UPSERT KEYS(ts, event_id) converges a redelivered batch"); + } + [Fact] public async Task Event_round_trips_and_a_repeat_is_one_row_with_every_write_Ack() { diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests/AlarmBatchWriteTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests/AlarmBatchWriteTests.cs new file mode 100644 index 000000000..cd6c18c93 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests/AlarmBatchWriteTests.cs @@ -0,0 +1,102 @@ +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests; + +/// +/// The shared batched alarm write: one statement per chunk instead of one round trip per event, with the per-event +/// outcomes the store-and-forward drain needs kept intact (a poison event is isolated, an outage retries everything, +/// a cancellation never dead-letters). +/// +public sealed class AlarmBatchWriteTests +{ + private sealed class PoisonException : Exception; + private sealed class OutageException : Exception; + + private static HistorianWriteOutcome Classify(Exception ex) => + ex is PoisonException ? HistorianWriteOutcome.PermanentFail : HistorianWriteOutcome.RetryPlease; + + private static HistorianWriteOutcome?[] Undecided(int n) => new HistorianWriteOutcome?[n]; + + [Fact] + public async Task A_clean_batch_is_one_statement_per_chunk_and_all_Ack() + { + var statements = new List(); + var r = await AlarmBatchWrite.RunAsync(Undecided(25), 10, + (ix, _) => { statements.Add(ix.ToArray()); return Task.CompletedTask; }, Classify, null, CancellationToken.None); + + r.ShouldAllBe(o => o == HistorianWriteOutcome.Ack); + statements.Select(s => s.Length).ShouldBe([10, 10, 5]); + statements.SelectMany(s => s).ShouldBe(Enumerable.Range(0, 25)); + } + + [Fact] + public async Task A_poison_event_is_isolated_and_its_siblings_still_Ack() + { + var statements = 0; + var r = await AlarmBatchWrite.RunAsync(Undecided(6), 10, (ix, _) => + { + statements++; + if (ix.Contains(3)) throw new PoisonException(); + return Task.CompletedTask; + }, Classify, null, CancellationToken.None); + + r.ShouldBe([HistorianWriteOutcome.Ack, HistorianWriteOutcome.Ack, HistorianWriteOutcome.Ack, + HistorianWriteOutcome.PermanentFail, HistorianWriteOutcome.Ack, HistorianWriteOutcome.Ack]); + statements.ShouldBe(1 + 6, "the failed chunk, then one statement per event"); + } + + [Fact] + public async Task An_outage_retries_the_whole_chunk_without_re_sending_event_by_event() + { + var statements = 0; + var r = await AlarmBatchWrite.RunAsync(Undecided(6), 10, + (_, _) => { statements++; throw new OutageException(); }, Classify, null, CancellationToken.None); + + r.ShouldAllBe(o => o == HistorianWriteOutcome.RetryPlease); + statements.ShouldBe(1); + } + + [Fact] + public async Task Pre_decided_slots_are_kept_and_never_sent() + { + var sent = new List(); + var outcomes = Undecided(4); + outcomes[1] = HistorianWriteOutcome.PermanentFail; // e.g. a pre-1970 time on QuestDB + var r = await AlarmBatchWrite.RunAsync(outcomes, 10, + (ix, _) => { sent.AddRange(ix); return Task.CompletedTask; }, Classify, null, CancellationToken.None); + + r.ShouldBe([HistorianWriteOutcome.Ack, HistorianWriteOutcome.PermanentFail, HistorianWriteOutcome.Ack, HistorianWriteOutcome.Ack]); + sent.ShouldBe([0, 2, 3]); + } + + [Fact] + public async Task A_cancellation_mid_batch_retries_everything_undecided_and_dead_letters_nothing() + { + using var cts = new CancellationTokenSource(); + var r = await AlarmBatchWrite.RunAsync(Undecided(5), 2, (ix, ct) => + { + if (ix.Contains(2)) { cts.Cancel(); ct.ThrowIfCancellationRequested(); } + return Task.CompletedTask; + }, Classify, null, cts.Token); + + r.ShouldBe([HistorianWriteOutcome.Ack, HistorianWriteOutcome.Ack, + HistorianWriteOutcome.RetryPlease, HistorianWriteOutcome.RetryPlease, HistorianWriteOutcome.RetryPlease]); + } + + [Fact] + public async Task A_cancellation_during_poison_isolation_retries_the_rest() + { + using var cts = new CancellationTokenSource(); + var r = await AlarmBatchWrite.RunAsync(Undecided(4), 10, (ix, ct) => + { + if (ix.Count > 1) throw new PoisonException(); + if (ix[0] == 1) { cts.Cancel(); ct.ThrowIfCancellationRequested(); } + return Task.CompletedTask; + }, Classify, null, cts.Token); + + r.ShouldBe([HistorianWriteOutcome.Ack, HistorianWriteOutcome.RetryPlease, HistorianWriteOutcome.RetryPlease, + HistorianWriteOutcome.RetryPlease]); + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/AlarmWriterTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/AlarmWriterTests.cs index a32e0cb81..0ecf22281 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/AlarmWriterTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/AlarmWriterTests.cs @@ -63,6 +63,31 @@ public sealed class AlarmWriterTests(SqlHistorianFixture fx) r.ShouldBe([HistorianWriteOutcome.Ack, HistorianWriteOutcome.PermanentFail, HistorianWriteOutcome.Ack]); } + [Fact] + public async Task A_batch_is_one_statement_per_chunk_and_a_redelivery_converges() + { + fx.SkipIfUnavailable(); + var (w, s) = Make(); + var events = Enumerable.Range(0, 400) + .Select(i => new AlarmHistorianEvent("alarm-1", "Plant/Line/Eq", "HiTemp", "LimitAlarm", AlarmSeverity.High, + "Activated", $"m{i}", "user", null, T.AddSeconds(i))) + .ToList(); + + (await w.WriteBatchAsync(events, CancellationToken.None)).ShouldAllBe(o => o == HistorianWriteOutcome.Ack); + w.Statements.ShouldBe(3, "400 events = 150 + 150 + 100, not 400 round trips"); + + // Redelivered (the at-least-once drain), with an in-batch duplicate: every event Acks, no row is added. + var again = events.Append(events[0]).ToList(); + var r = await w.WriteBatchAsync(again, CancellationToken.None); + r.Count.ShouldBe(401); + r.ShouldAllBe(o => o == HistorianWriteOutcome.Ack); + await using var c = new SqlConnection(fx.ConnectionString); + await c.OpenAsync(); + await using var cmd = c.CreateCommand(); + cmd.CommandText = $"SELECT COUNT(*) FROM [{s}].Event"; + Convert.ToInt32(await cmd.ExecuteScalarAsync()).ShouldBe(400); + } + [Fact] public async Task Concurrent_writers_of_the_same_event_all_Ack_with_one_row() {