diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian/AlarmConditionStateSchema.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian/AlarmConditionStateSchema.cs index 38b3acf5..464cb90d 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian/AlarmConditionStateSchema.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian/AlarmConditionStateSchema.cs @@ -1,11 +1,13 @@ +using System.Globalization; +using System.Text.Json; using Microsoft.Data.Sqlite; namespace ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian; /// -/// DDL for the Part 9 scripted-alarm condition state: one row per alarm identity holding the -/// operator-supplied state (Enabled / Acked / Confirmed / Shelving) plus the ack/confirm audit -/// trail and comment history. +/// DDL for the Part 9 scripted-alarm condition state: one fixed-size row per alarm identity +/// holding the operator-supplied state (Enabled / Acked / Confirmed / Shelving) plus the +/// ack/confirm audit fields, and a separate append-only table holding each alarm's comment trail. /// /// /// @@ -34,14 +36,58 @@ namespace ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian; /// the last transition. Timestamps are round-trip ("O") TEXT so lexicographic order is /// chronological; nullable timestamps and audit fields are stored as NULL. /// +/// +/// The comment trail is NOT on the condition row. It used to be serialized whole into +/// comments_json, so every transition rewrote — and replicated — the alarm's entire +/// history: O(n) per transition, O(n²) over the alarm's life (review R05:4). It now lives in +/// , one row per entry keyed by (scripted_alarm_id, seq), +/// where seq is the entry's ordinal in the alarm's trail. A transition inserts only the +/// entries it appended, so the condition row stays a fixed size and a transition's write is +/// O(1) however long the history is. comments_json is kept, always written as +/// "[]", so the condition table's shape — and with it the pair's replication digest +/// for that table — does not change and an old-shape row stays readable; +/// copies any inline trail across (the alarm's next +/// save clears the column). +/// +/// +/// Why the ordinal, and not a content hash, is the key. Both halves of a redundant pair +/// run the engine and both apply every operator command, each stamping the entry with its own +/// clock. A content-derived id would keep TWO rows for every logical entry once the pair +/// converged. Keyed by ordinal, the two nodes' copies of entry n share one primary key +/// and LWW collapses them to one row — the convergence the whole-row LWW gave before, now per +/// entry. Entries go in with INSERT OR IGNORE (), so an +/// entry already held — written locally, or arrived from the peer — is never rewritten. A +/// node that MISSED a peer command keeps its own colliding entry by appending it after the +/// stored trail, with a Warning (see the store's divergence handling). +/// /// public static class AlarmConditionStateSchema { /// Table holding one condition-state row per scripted-alarm identity. public const string StateTable = "alarm_condition_state"; + /// Append-only table holding one row per comment-trail entry, keyed by alarm id + ordinal. + public const string CommentsTable = "alarm_condition_comment"; + + /// What comments_json holds once the trail lives in . + public const string EmptyInlineComments = "[]"; + /// - /// Creates the state table if it does not already exist. Idempotent. + /// The one append statement for , shared by the migration and the + /// store. OR IGNORE is what makes the trail append-only: an ordinal already held is + /// never rewritten. Parameters: @Id @Seq @Ts @User @Kind @Text. + /// + public const string InsertCommentSql = + """ + INSERT OR IGNORE INTO alarm_condition_comment + (scripted_alarm_id, seq, timestamp_utc, user_name, kind, text) + VALUES (@Id, @Seq, @Ts, @User, @Kind, @Text) + """; + + private static readonly JsonSerializerOptions InlineJsonOptions = new(JsonSerializerDefaults.Web); + + /// + /// Creates the state table and the comment table if they do not already exist. Idempotent. /// /// /// An already-open connection. ILocalDb.CreateConnection() hands out open, @@ -55,10 +101,11 @@ public static class AlarmConditionStateSchema // The four *_state columns are always written to a mapped, non-null string (Enabled/Disabled, // Acknowledged/Unacknowledged, Confirmed/Unconfirmed, Unshelved/OneShotShelved/TimedShelved), - // so they are NOT NULL. comments_json is never null either — an empty trail is the literal - // "[]". The audit user/comment/utc columns and shelving_expires_utc are NULL when the alarm - // has no such history yet. updated_at_utc carries LastTransitionUtc (there is no dedicated - // transition column). + // so they are NOT NULL. comments_json is never null either — it is the literal "[]" since the + // trail moved to alarm_condition_comment (an old-shape row may still carry an inline trail + // until the alarm's next save; MigrateInlineComments copies it without touching this row). + // The audit user/comment/utc columns and shelving_expires_utc are NULL when the alarm has no such history yet. updated_at_utc + // carries LastTransitionUtc (there is no dedicated transition column). cmd.CommandText = """ CREATE TABLE IF NOT EXISTS alarm_condition_state ( scripted_alarm_id TEXT NOT NULL PRIMARY KEY, @@ -76,7 +123,161 @@ public static class AlarmConditionStateSchema comments_json TEXT NOT NULL, updated_at_utc TEXT NOT NULL ); + + CREATE TABLE IF NOT EXISTS alarm_condition_comment ( + scripted_alarm_id TEXT NOT NULL, + seq INTEGER NOT NULL, + timestamp_utc TEXT NOT NULL, + user_name TEXT NOT NULL, + kind TEXT NOT NULL, + text TEXT NOT NULL, + PRIMARY KEY (scripted_alarm_id, seq) + ); """; cmd.ExecuteNonQuery(); } + + /// What one run did. + /// Condition rows for which at least one inline entry was copied across. + /// + /// Alarms whose inline trail is not valid JSON and was left where it is — the caller logs these, + /// so an unmigratable history is visible rather than skipped silently on every boot. + /// + public sealed record InlineCommentMigrationResult(int RowsMoved, IReadOnlyList UnparseableAlarmIds); + + /// + /// Copies every condition row's inline comments_json trail into + /// . Idempotent; it never writes the condition row itself. + /// + /// + /// + /// Run it AFTER the tables are registered for replication — it writes rows, and a + /// row written before the capture triggers exist never reaches the peer (the same + /// DDL → register → writes rule AlarmSfLegacyMigrator follows). + /// + /// + /// Copy, don't clear. Resetting comments_json here would be a full-row UPDATE + /// that re-stamps the condition row with a fresh HLC — during a pair's rolling upgrade + /// that stale-but-newly-stamped copy would beat a newer peer state (an ack made on the + /// already-upgraded node) under LWW (review m1). The column is left as it is: the store's + /// load merges an inline trail with the table by ordinal (the table wins), and the + /// engine's next save of that alarm writes "[]" with the alarm's real state. + /// + /// + /// Each entry lands at its ordinal in the inline array via , + /// so a re-run — or a row whose trail was already (partly) copied — neither duplicates nor + /// reorders anything and writes nothing for an ordinal already held. Each row copies in its + /// own transaction, so a crash leaves it fully copied or untouched. A row whose inline JSON + /// does not parse is left exactly as it is (the pre-split store could not load it either) + /// and reported in rather + /// than failing host start. + /// + /// + /// An already-open connection from ILocalDb.CreateConnection(). + /// What was copied, and which rows could not be read. + public static InlineCommentMigrationResult MigrateInlineComments(SqliteConnection connection) + { + ArgumentNullException.ThrowIfNull(connection); + + var pending = new List<(string AlarmId, string Json)>(); + using (var select = connection.CreateCommand()) + { + select.CommandText = + "SELECT scripted_alarm_id, comments_json FROM alarm_condition_state WHERE comments_json <> @Empty"; + select.Parameters.AddWithValue("@Empty", EmptyInlineComments); + using var reader = select.ExecuteReader(); + while (reader.Read()) + pending.Add((reader.GetString(0), reader.GetString(1))); + } + + var migrated = 0; + var unparseable = new List(); + foreach (var (alarmId, json) in pending) + { + IReadOnlyList entries; + try + { + entries = ParseInlineComments(json); + } + catch (JsonException) + { + unparseable.Add(alarmId); + continue; + } + + var inserted = 0; + using var tx = connection.BeginTransaction(); + for (var seq = 0; seq < entries.Count; seq++) + { + using var insert = connection.CreateCommand(); + insert.Transaction = tx; + insert.CommandText = InsertCommentSql; + insert.Parameters.AddWithValue("@Id", alarmId); + insert.Parameters.AddWithValue("@Seq", seq); + insert.Parameters.AddWithValue("@Ts", FormatTimestamp(entries[seq].TimestampUtc)); + insert.Parameters.AddWithValue("@User", entries[seq].User); + insert.Parameters.AddWithValue("@Kind", entries[seq].Kind); + insert.Parameters.AddWithValue("@Text", entries[seq].Text); + inserted += insert.ExecuteNonQuery(); + } + + tx.Commit(); + if (inserted > 0) + migrated++; + } + + return new InlineCommentMigrationResult(migrated, unparseable); + } + + /// Formats a comment timestamp for (round-trip "O" text). + /// The timestamp. + /// The round-trip text form. + public static string FormatTimestamp(DateTime value) => value.ToString("O", CultureInfo.InvariantCulture); + + /// Parses a timestamp, preserving its . + /// The round-trip text form. + /// The timestamp. + public static DateTime ParseTimestamp(string value) => + DateTime.Parse(value, CultureInfo.InvariantCulture, DateTimeStyles.RoundtripKind); + + /// + /// Parses a legacy inline comments_json trail exactly as the pre-split store did + /// (; null strings become empty). + /// + /// The column value. + /// The entries in trail order; empty for a blank or "[]" column. + /// The column is not a valid trail. + public static IReadOnlyList ParseInlineComments(string? json) + { + if (string.IsNullOrWhiteSpace(json)) return []; + var dtos = JsonSerializer.Deserialize>(json, InlineJsonOptions); + if (dtos is null || dtos.Count == 0) return []; + return dtos + .Select(d => new InlineComment( + d.TimestampUtc, d.User ?? string.Empty, d.Kind ?? string.Empty, d.Text ?? string.Empty)) + .ToList(); + } + + /// One comment-trail entry as stored — the persistence-side mirror of the engine's AlarmComment. + /// When the action happened. + /// Identity of the actor. + /// Classification (Acknowledge, Confirm, AddComment, …). + /// Operator-supplied or engine-generated text. + public sealed record InlineComment(DateTime TimestampUtc, string User, string Kind, string Text); + + /// The pre-split on-disk shape of one entry in comments_json. + private sealed class InlineCommentDto + { + /// When the comment was recorded. + public DateTime TimestampUtc { get; set; } + + /// Identity of the actor. + public string? User { get; set; } + + /// Classification of the entry. + public string? Kind { get; set; } + + /// Entry text. + public string? Text { get; set; } + } } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/LocalDbSetup.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/LocalDbSetup.cs index 3a4129a2..d3f89979 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/LocalDbSetup.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/LocalDbSetup.cs @@ -7,7 +7,7 @@ namespace ZB.MOM.WW.OtOpcUa.Host.Configuration; /// /// The onReady callback handed to AddZbLocalDb: creates the deployment-cache, -/// alarm store-and-forward, and scripted-alarm condition-state tables and — when this node +/// alarm store-and-forward, and scripted-alarm condition-state + comment tables and — when this node /// actually replicates — opts them into change capture. /// /// @@ -21,7 +21,7 @@ namespace ZB.MOM.WW.OtOpcUa.Host.Configuration; public static class LocalDbSetup { /// - /// The four tables that participate in replication — registered when this node is configured + /// The five tables that participate in replication — registered when this node is configured /// to replicate, and deregistered (stale capture triggers dropped) when it is not. /// /// @@ -36,6 +36,7 @@ public static class LocalDbSetup DeploymentCacheSchema.PointerTable, AlarmSfSchema.EventsTable, AlarmConditionStateSchema.StateTable, + AlarmConditionStateSchema.CommentsTable, ]; /// @@ -47,8 +48,9 @@ public static class LocalDbSetup /// RegisterReplicated is what installs the three AFTER triggers that capture /// changes into the oplog. Any row written before that call is never captured, so it /// never reaches the peer — silently, and permanently, because nothing ever revisits - /// history. therefore runs last, after every - /// registration — it writes rows, and they must be captured like any other write. + /// history. and + /// therefore run last, after every + /// registration — they write rows, and those must be captured like any other write. /// /// /// Registration is CONDITIONAL on . It used @@ -68,7 +70,7 @@ public static class LocalDbSetup /// /// /// A failure part-way through either loop fails host start, and that is deliberate. - /// Neither branch catches, so a throw on the third of four tables leaves two flipped and + /// Neither branch catches, so a throw on the third of five tables leaves two flipped and /// propagates out of the AddZbLocalDb singleton factory — a node that came up /// anyway would be capturing (or not) on an arbitrary subset, which is silently partial /// replication with nothing in the log to say so. Both library calls are idempotent, so @@ -116,7 +118,36 @@ public static class LocalDbSetup else RemoveStaleCapture(db, logger ?? Serilog.Log.Logger); - // LAST, and only here. This is the one call in OnReady that writes rows. + // LAST, and only here. These are the only calls in OnReady that write rows. + // Copies any pre-split inline comment trail (alarm_condition_state.comments_json) into the + // append-only alarm_condition_comment table (review R05:4). Idempotent: an ordinal already + // held is never rewritten, and the condition row itself is never touched (the alarm's next + // save clears the column). An unreadable trail is reported, not skipped silently. + AlarmConditionStateSchema.InlineCommentMigrationResult commentMigration; + using (var connection = db.CreateConnection()) + { + commentMigration = AlarmConditionStateSchema.MigrateInlineComments(connection); + } + + var migrationLog = logger ?? Serilog.Log.Logger; + if (commentMigration.RowsMoved > 0) + { + migrationLog.Information( + "LocalDb: copied the inline comment trail of {AlarmCount} scripted alarm(s) into {Table}.", + commentMigration.RowsMoved, + AlarmConditionStateSchema.CommentsTable); + } + + foreach (var alarmId in commentMigration.UnparseableAlarmIds) + { + migrationLog.Warning( + "LocalDb: scripted alarm {AlarmId} has an unreadable inline comment trail " + + "(alarm_condition_state.comments_json is not valid JSON); its history was NOT moved to {Table} " + + "and the alarm's state will not load until the row is repaired.", + alarmId, + AlarmConditionStateSchema.CommentsTable); + } + AlarmSfLegacyMigrator.Migrate(db, configuration); } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ScriptedAlarms/LocalDbAlarmConditionStateStore.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ScriptedAlarms/LocalDbAlarmConditionStateStore.cs index d540f52c..514b7ea9 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ScriptedAlarms/LocalDbAlarmConditionStateStore.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ScriptedAlarms/LocalDbAlarmConditionStateStore.cs @@ -1,6 +1,4 @@ using System.Collections.Immutable; -using System.Globalization; -using System.Text.Json; using Microsoft.Data.Sqlite; using Microsoft.Extensions.Logging; using ZB.MOM.WW.LocalDb; @@ -11,9 +9,11 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.ScriptedAlarms; /// /// Node-local backed by the replicated -/// table in the node's consolidated LocalDb. -/// Maps the full Part 9 — Enabled / Acked / Confirmed / -/// Shelving + the ack/confirm audit trail + operator comments. +/// and +/// tables in the node's consolidated +/// LocalDb. Maps the full Part 9 — Enabled / Acked / Confirmed / +/// Shelving + the ack/confirm audit fields on the condition row, and the append-only comment +/// trail one row per entry. /// /// /// @@ -24,44 +24,75 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.ScriptedAlarms; /// state then mirrors to its redundant pair peer instead of depending on central SQL. /// /// -/// The round-trip is byte-identical to the EF store — the enum↔string mappers and the -/// comment serialization below are duplicated verbatim from it (it is deleted in a later -/// task; the ported round-trip tests prove parity). The persistence rules carry over -/// unchanged: -/// -/// /// ActiveState is NOT persisted — there is no column; on it is /// restored as and the engine re-derives it from the /// live predicate. LastTransitionUtc ↔ updated_at_utc: no dedicated transition column, /// so the last transition rides the row-write timestamp. LastActiveUtc / LastClearedUtc -/// have no columns and default to null on load. -/// serializes to/from comments_json; an empty list round-trips as "[]". +/// have no columns and default to null on load. /// /// -/// Save is a single unconditional PK upsert. No application-level last-write-wins or -/// HLC comparison is layered on top — the LocalDb replication HLC handles convergence, and an -/// unconditional upsert keyed by the primary key is HLC-safe (the same discipline the -/// alarm-sf sink and the deployment pointer follow). +/// The comment trail is written incrementally (review R05:4). Each save writes only +/// the entries the table does not yet hold, so a transition's write — and its replication +/// payload — is constant however long the history is; the condition row's +/// comments_json is always written as "[]". What is "already stored" is read +/// from the table inside the save's transaction (a bounded read of the trail's tail) rather +/// than cached, so the store holds no per-alarm state. The pair peer's copy of the same +/// command is recognised and not duplicated; a local entry that collides with a peer entry +/// this node never applied is appended after it with a Warning, never dropped — see +/// AppendNewCommentsAsync and, for why the ordinal is the key, +/// . +/// +/// +/// Remove vs. a lagging peer. deletes the whole trail; if the +/// pair peer has not yet applied the same deploy, its next save finds no stored trail and +/// re-inserts every entry, which replicates back. The next deploy removes them again — the +/// same resurrection class the single-row store had, with n+1 rows instead of one (review m3). +/// +/// +/// Old-shape rows load exactly as before. A row still carrying an inline +/// comments_json trail (written by the pre-split store and not yet moved by +/// ) is read through the same +/// parser the old store used, merged by ordinal with whatever the comment table holds (the +/// table wins an ordinal both hold), and moved to the table on that alarm's next save. +/// +/// +/// Save is one transaction: the comment append and an unconditional PK upsert of the +/// condition row. No application-level last-write-wins or HLC comparison is layered on top — +/// the LocalDb replication HLC handles convergence. /// /// public sealed class LocalDbAlarmConditionStateStore : IAlarmStateStore { - private static readonly JsonSerializerOptions JsonOptions = new(JsonSerializerDefaults.Web); - // Fixed column order shared by every read — the reader map below indexes by these ordinals. private const string SelectColumns = "scripted_alarm_id, enabled_state, acked_state, confirmed_state, shelving_state, " + "shelving_expires_utc, last_ack_user, last_ack_comment, last_ack_utc, " + "last_confirm_user, last_confirm_comment, last_confirm_utc, comments_json, updated_at_utc"; + private const string SelectCommentColumns = "scripted_alarm_id, seq, timestamp_utc, user_name, kind, text"; + + /// + /// How many trailing in-memory entries each save reconciles against the stored trail by + /// content. The engine appends at most one entry per save, so this comfortably covers a few + /// missed commands while keeping each save's read bounded. + /// + private const int MatchWindow = 16; + + /// + /// How far apart two nodes' timestamps for the SAME command may be and still count as one + /// entry: command fan-out latency plus inter-node clock skew. Beyond it both copies are kept + /// (the safe direction — a duplicate, never a loss). + /// + private static readonly TimeSpan SameCommandSkew = TimeSpan.FromMinutes(5); + private readonly ILocalDb _db; private readonly ILogger _logger; /// Initializes a new instance of the class. /// - /// The node's local database. Its table - /// must already exist and be registered for replication — the host does both in - /// LocalDbSetup.OnReady, before any consumer can resolve this store. + /// The node's local database. Its condition-state and comment tables must already exist and + /// be registered for replication — the host does both in LocalDbSetup.OnReady, before + /// any consumer can resolve this store. /// /// The logger instance. public LocalDbAlarmConditionStateStore(ILocalDb db, ILogger logger) @@ -79,7 +110,15 @@ public sealed class LocalDbAlarmConditionStateStore : IAlarmStateStore MapRow, new { Id = alarmId }, ct).ConfigureAwait(false); - return rows.Count > 0 ? rows[0] : null; + if (rows.Count == 0) return null; + + var comments = await _db.QueryAsync( + $"SELECT {SelectCommentColumns} FROM alarm_condition_comment WHERE scripted_alarm_id = @Id ORDER BY seq", + MapComment, + new { Id = alarmId }, + ct).ConfigureAwait(false); + + return WithComments(rows[0], comments); } /// @@ -90,7 +129,19 @@ public sealed class LocalDbAlarmConditionStateStore : IAlarmStateStore MapRow, parameters: null, ct).ConfigureAwait(false); - return rows; + + var comments = await _db.QueryAsync( + $"SELECT {SelectCommentColumns} FROM alarm_condition_comment ORDER BY scripted_alarm_id, seq", + MapComment, + parameters: null, + ct).ConfigureAwait(false); + var byAlarm = comments + .GroupBy(c => c.AlarmId, StringComparer.Ordinal) + .ToDictionary(g => g.Key, g => (IReadOnlyList)g.ToList(), StringComparer.Ordinal); + + return rows + .Select(r => WithComments(r, byAlarm.TryGetValue(r.State.AlarmId, out var c) ? c : [])) + .ToList(); } /// @@ -98,10 +149,15 @@ public sealed class LocalDbAlarmConditionStateStore : IAlarmStateStore { ArgumentNullException.ThrowIfNull(state); + await using var tx = await _db.BeginTransactionAsync(ct).ConfigureAwait(false); + + await AppendNewCommentsAsync(tx, state, ct).ConfigureAwait(false); + // A single unconditional upsert keyed on the alarm identity. Convergence across the pair is // last-writer-wins over the primary key, handled by the LocalDb replication HLC — no - // app-level LWW here (see the class remarks). - await _db.ExecuteAsync( + // app-level LWW here (see the class remarks). comments_json is always "[]": the trail lives + // in alarm_condition_comment, which is what keeps this row a fixed size. + await tx.ExecuteAsync( """ INSERT INTO alarm_condition_state (scripted_alarm_id, enabled_state, acked_state, confirmed_state, shelving_state, @@ -140,60 +196,243 @@ public sealed class LocalDbAlarmConditionStateStore : IAlarmStateStore LastConfirmUser = state.LastConfirmUser, LastConfirmComment = state.LastConfirmComment, LastConfirmUtc = FormatNullable(state.LastConfirmUtc), - CommentsJson = SerializeComments(state.Comments), + CommentsJson = AlarmConditionStateSchema.EmptyInlineComments, // No dedicated transition column — persist LastTransitionUtc into updated_at_utc. UpdatedAtUtc = Format(state.LastTransitionUtc), }, ct).ConfigureAwait(false); + await tx.CommitAsync(ct).ConfigureAwait(false); + _logger.LogTrace("Persisted alarm-condition state for {AlarmId}", state.AlarmId); } + /// + /// Persists the trail entries holds that the table does not, writing + /// a bounded number of rows however long the trail is. + /// + /// + /// + /// Normal case. The in-memory trail and the stored trail agree position by + /// position; the only new entries are past MAX(seq) and land at their own ordinal. + /// + /// + /// Divergence (review I-1). If this node missed a command its peer applied, the + /// peer's entry arrives in this table at an ordinal this node's in-memory trail does not + /// know about, and this node's NEXT entry has an in-memory index at or below + /// MAX(seq). Positional "already stored" would silently drop it — the survivor's + /// newest operator action, in the failover case. So the tail of the in-memory trail + /// (the last entries) is matched against the stored rows from + /// that position on, by content: same Kind, User and Text, timestamps within + /// (each node stamps the same command with its own clock). + /// A matched stored row is consumed, so repeated identical entries are counted, not + /// merged. An entry with no match is new: at its own ordinal when that is past the stored + /// trail, otherwise appended after it (MAX(seq)+1) with a Warning — kept, never + /// dropped, and never silent. + /// + /// + /// Why this cannot double-count. Both pair nodes apply every command; the second + /// node to save finds the first's copy among the stored rows (same Kind/User/Text, close + /// timestamp) and consumes it instead of appending. A node's own earlier divergence-append + /// is found the same way, so a re-save writes nothing. What it cannot tell apart: a peer + /// entry this node missed and a byte-identical entry this node later records within the + /// skew window — that pair is kept once. A clock skew between the nodes beyond the window + /// errs the other way (both copies kept), never towards loss. True union semantics need an + /// originator-assigned entry id on the alarm command (follow-up F3). + /// + /// + /// ⚠️ The in-memory index is the durable ordinal. Trimming + /// in the engine without carrying a persisted + /// base offset would make every later save look "already stored" — do not bound that list + /// without changing this method. + /// + /// + private async Task AppendNewCommentsAsync(ILocalDbTransaction tx, AlarmConditionState state, CancellationToken ct) + { + var comments = state.Comments; + var count = comments.Count; + + // An indexed MAX over the (scripted_alarm_id, seq) primary key — a read, so no replication cost. + var max = (await tx.QueryAsync( + "SELECT COALESCE(MAX(seq), -1) FROM alarm_condition_comment WHERE scripted_alarm_id = @Id", + static r => r.GetInt64(0), + new { Id = state.AlarmId }, + ct).ConfigureAwait(false))[0]; + + var windowStart = Math.Max(0, count - MatchWindow); + + // Entries below the window that lie past the stored trail (a first save of a long trail, an + // old-shape row's lazy move): plain appends at their own ordinal — nothing is stored there. + for (var seq = (int)Math.Max(max + 1, 0); seq < windowStart; seq++) + { + await InsertCommentAsync(tx, state.AlarmId, seq, comments[seq], ct).ConfigureAwait(false); + max = seq; + } + + if (windowStart >= count) return; + + // The stored rows the window can match: ordinal >= windowStart. Bounded by the window plus the + // number of commands this node missed. + var stored = max >= windowStart + ? await tx.QueryAsync( + $"SELECT {SelectCommentColumns} FROM alarm_condition_comment WHERE scripted_alarm_id = @Id AND seq >= @Lo ORDER BY seq", + MapComment, + new { Id = state.AlarmId, Lo = windowStart }, + ct).ConfigureAwait(false) + : []; + var consumed = new bool[stored.Count]; + + for (var i = windowStart; i < count; i++) + { + var entry = comments[i]; + var match = -1; + for (var j = 0; j < stored.Count; j++) + { + if (!consumed[j] && IsSameEntry(stored[j].Comment, entry)) + { + match = j; + break; + } + } + + if (match >= 0) + { + consumed[match] = true; + continue; + } + + if (i > max) + { + await InsertCommentAsync(tx, state.AlarmId, i, entry, ct).ConfigureAwait(false); + max = i; + continue; + } + + var target = max + 1; + await InsertCommentAsync(tx, state.AlarmId, target, entry, ct).ConfigureAwait(false); + max = target; + _logger.LogWarning( + "Alarm {AlarmId}: comment trail diverged from the pair peer — the {Kind} entry by {User} at in-memory " + + "position {Position} collides with a stored entry this node never applied (a missed command). " + + "Kept: appended at ordinal {Ordinal}.", + state.AlarmId, entry.Kind, entry.User, i, target); + } + } + + private static bool IsSameEntry(AlarmComment stored, AlarmComment local) => + string.Equals(stored.Kind, local.Kind, StringComparison.Ordinal) + && string.Equals(stored.User, local.User, StringComparison.Ordinal) + && string.Equals(stored.Text, local.Text, StringComparison.Ordinal) + && (stored.TimestampUtc - local.TimestampUtc).Duration() <= SameCommandSkew; + + private static Task InsertCommentAsync( + ILocalDbTransaction tx, string alarmId, long seq, AlarmComment c, CancellationToken ct) => + tx.ExecuteAsync( + AlarmConditionStateSchema.InsertCommentSql, + new + { + Id = alarmId, + Seq = seq, + // AlarmComment.TimestampUtc must be DateTimeKind.Utc for a Utc round-trip; the engine + // always creates comments with Utc kind. + Ts = AlarmConditionStateSchema.FormatTimestamp(c.TimestampUtc), + User = c.User, + Kind = c.Kind, + Text = c.Text, + }, + ct); + /// public async Task RemoveAsync(string alarmId, CancellationToken ct) { ArgumentNullException.ThrowIfNull(alarmId); - await _db.ExecuteAsync( + + // The trail goes with the condition, as it did when it was a column of the row. + await using var tx = await _db.BeginTransactionAsync(ct).ConfigureAwait(false); + await tx.ExecuteAsync( + "DELETE FROM alarm_condition_comment WHERE scripted_alarm_id = @Id", + new { Id = alarmId }, + ct).ConfigureAwait(false); + await tx.ExecuteAsync( "DELETE FROM alarm_condition_state WHERE scripted_alarm_id = @Id", new { Id = alarmId }, ct).ConfigureAwait(false); + await tx.CommitAsync(ct).ConfigureAwait(false); } - private static AlarmConditionState MapRow(SqliteDataReader r) => new( + /// A condition row as read, before its trail is attached. + private sealed record ConditionRow(AlarmConditionState State, string InlineCommentsJson); + + /// One alarm_condition_comment row. + private sealed record StoredComment(string AlarmId, long Seq, AlarmComment Comment); + + /// + /// Attaches the trail: the comment table's entries by ordinal, plus — for an old-shape row not + /// yet migrated — any inline entry at an ordinal the table does not hold. + /// + private static AlarmConditionState WithComments(ConditionRow row, IReadOnlyList stored) + { + var inline = AlarmConditionStateSchema.ParseInlineComments(row.InlineCommentsJson); + if (inline.Count == 0) + return row.State with { Comments = stored.Select(s => s.Comment).ToImmutableList() }; + + var merged = new SortedDictionary(); + foreach (var s in stored) + merged[s.Seq] = s.Comment; + for (var i = 0; i < inline.Count; i++) + { + var c = inline[i]; + merged.TryAdd(i, new AlarmComment(c.TimestampUtc, c.User, c.Kind, c.Text)); + } + + return row.State with { Comments = merged.Values.ToImmutableList() }; + } + + private static StoredComment MapComment(SqliteDataReader r) => new( AlarmId: r.GetString(0), - Enabled: string.Equals(r.GetString(1), "Disabled", StringComparison.Ordinal) - ? AlarmEnabledState.Disabled - : AlarmEnabledState.Enabled, // unknown string → Enabled (safe default) - // Active is not persisted — the engine re-derives it from the predicate at startup. - Active: AlarmActiveState.Inactive, - Acked: string.Equals(r.GetString(2), "Acknowledged", StringComparison.Ordinal) - ? AlarmAckedState.Acknowledged - : AlarmAckedState.Unacknowledged, // unknown string → Unacknowledged (safe default) - Confirmed: string.Equals(r.GetString(3), "Confirmed", StringComparison.Ordinal) - ? AlarmConfirmedState.Confirmed - : AlarmConfirmedState.Unconfirmed, // unknown string → Unconfirmed (safe default) - Shelving: new ShelvingState(MapShelvingFromColumn(r.GetString(4)), ParseNullable(r, 5)), - // No transition column — updated_at_utc carries the last transition timestamp. - LastTransitionUtc: Parse(r.GetString(13)), - // LastActiveUtc / LastClearedUtc have no columns — they re-derive with Active, so null on load. - LastActiveUtc: null, - LastClearedUtc: null, - LastAckUtc: ParseNullable(r, 8), - LastAckUser: r.IsDBNull(6) ? null : r.GetString(6), - LastAckComment: r.IsDBNull(7) ? null : r.GetString(7), - LastConfirmUtc: ParseNullable(r, 11), - LastConfirmUser: r.IsDBNull(9) ? null : r.GetString(9), - LastConfirmComment: r.IsDBNull(10) ? null : r.GetString(10), - Comments: DeserializeComments(r.GetString(12))); + Seq: r.GetInt64(1), + Comment: new AlarmComment( + AlarmConditionStateSchema.ParseTimestamp(r.GetString(2)), + r.GetString(3), + r.GetString(4), + r.GetString(5))); + + private static ConditionRow MapRow(SqliteDataReader r) => new( + new AlarmConditionState( + AlarmId: r.GetString(0), + Enabled: string.Equals(r.GetString(1), "Disabled", StringComparison.Ordinal) + ? AlarmEnabledState.Disabled + : AlarmEnabledState.Enabled, // unknown string → Enabled (safe default) + // Active is not persisted — the engine re-derives it from the predicate at startup. + Active: AlarmActiveState.Inactive, + Acked: string.Equals(r.GetString(2), "Acknowledged", StringComparison.Ordinal) + ? AlarmAckedState.Acknowledged + : AlarmAckedState.Unacknowledged, // unknown string → Unacknowledged (safe default) + Confirmed: string.Equals(r.GetString(3), "Confirmed", StringComparison.Ordinal) + ? AlarmConfirmedState.Confirmed + : AlarmConfirmedState.Unconfirmed, // unknown string → Unconfirmed (safe default) + Shelving: new ShelvingState(MapShelvingFromColumn(r.GetString(4)), ParseNullable(r, 5)), + // No transition column — updated_at_utc carries the last transition timestamp. + LastTransitionUtc: Parse(r.GetString(13)), + // LastActiveUtc / LastClearedUtc have no columns — they re-derive with Active, so null on load. + LastActiveUtc: null, + LastClearedUtc: null, + LastAckUtc: ParseNullable(r, 8), + LastAckUser: r.IsDBNull(6) ? null : r.GetString(6), + LastAckComment: r.IsDBNull(7) ? null : r.GetString(7), + LastConfirmUtc: ParseNullable(r, 11), + LastConfirmUser: r.IsDBNull(9) ? null : r.GetString(9), + LastConfirmComment: r.IsDBNull(10) ? null : r.GetString(10), + Comments: ImmutableList.Empty), + InlineCommentsJson: r.GetString(12)); // Timestamps are stored as round-trip ("O") TEXT so lexicographic order stays chronological and // the DateTimeKind.Utc round-trips exactly. - private static string Format(DateTime value) => value.ToString("O", CultureInfo.InvariantCulture); + private static string Format(DateTime value) => AlarmConditionStateSchema.FormatTimestamp(value); - private static string? FormatNullable(DateTime? value) => value?.ToString("O", CultureInfo.InvariantCulture); + private static string? FormatNullable(DateTime? value) => value is { } v ? Format(v) : null; - private static DateTime Parse(string value) => - DateTime.Parse(value, CultureInfo.InvariantCulture, DateTimeStyles.RoundtripKind); + private static DateTime Parse(string value) => AlarmConditionStateSchema.ParseTimestamp(value); private static DateTime? ParseNullable(SqliteDataReader r, int ordinal) => r.IsDBNull(ordinal) ? null : Parse(r.GetString(ordinal)); @@ -220,45 +459,4 @@ public sealed class LocalDbAlarmConditionStateStore : IAlarmStateStore "TimedShelved" => ShelvingKind.Timed, _ => ShelvingKind.Unshelved, // unknown string → Unshelved (safe default) }; - - private static string SerializeComments(ImmutableList comments) - { - if (comments.IsEmpty) return "[]"; - var dtos = comments.Select(c => new CommentDto - { - // AlarmComment.TimestampUtc must be DateTimeKind.Utc for correct ISO-8601 round-trip; - // the engine always creates AlarmComment instances with Utc kind. - TimestampUtc = c.TimestampUtc, - User = c.User, - Kind = c.Kind, - Text = c.Text, - }); - return JsonSerializer.Serialize(dtos, JsonOptions); - } - - private static ImmutableList DeserializeComments(string? json) - { - if (string.IsNullOrWhiteSpace(json)) return ImmutableList.Empty; - var dtos = JsonSerializer.Deserialize>(json, JsonOptions); - if (dtos is null || dtos.Count == 0) return ImmutableList.Empty; - return dtos - .Select(d => new AlarmComment(d.TimestampUtc, d.User ?? string.Empty, d.Kind ?? string.Empty, d.Text ?? string.Empty)) - .ToImmutableList(); - } - - /// Stable on-disk shape for a persisted in comments_json. - private sealed class CommentDto - { - /// When the comment was recorded (UTC). - public DateTime TimestampUtc { get; set; } - - /// Identity of the actor that wrote the comment. - public string? User { get; set; } - - /// Human-readable classification of the comment (Acknowledge, Confirm, …). - public string? Kind { get; set; } - - /// Operator-supplied or engine-generated comment text. - public string? Text { get; set; } - } } diff --git a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian.Tests/AlarmConditionStateSchemaTests.cs b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian.Tests/AlarmConditionStateSchemaTests.cs index 028bc49f..8e9cf432 100644 --- a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian.Tests/AlarmConditionStateSchemaTests.cs +++ b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian.Tests/AlarmConditionStateSchemaTests.cs @@ -34,6 +34,24 @@ public sealed class AlarmConditionStateSchemaTests ((long)cmd.ExecuteScalar()!).ShouldBe(1); } + /// Apply also creates the append-only comment table, keyed by alarm id + ordinal. + [Fact] + public void Apply_creates_the_comment_table_with_a_composite_key() + { + using var connection = OpenInMemory(); + + AlarmConditionStateSchema.Apply(connection); + + using var cmd = connection.CreateCommand(); + cmd.CommandText = "SELECT name FROM pragma_table_info(@Name) WHERE pk > 0 ORDER BY pk"; + cmd.Parameters.AddWithValue("@Name", AlarmConditionStateSchema.CommentsTable); + var pk = new List(); + using (var reader = cmd.ExecuteReader()) + while (reader.Read()) pk.Add(reader.GetString(0)); + + pk.ShouldBe(["scripted_alarm_id", "seq"]); + } + /// Applying twice on the same connection does not throw (CREATE TABLE IF NOT EXISTS). [Fact] public void Apply_is_idempotent() diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDb/AlarmConditionCommentConvergenceTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDb/AlarmConditionCommentConvergenceTests.cs new file mode 100644 index 00000000..8a6a7e62 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDb/AlarmConditionCommentConvergenceTests.cs @@ -0,0 +1,176 @@ +using System.Collections.Immutable; +using Microsoft.Extensions.Logging.Abstractions; +using Shouldly; +using Xunit; +using ZB.MOM.WW.LocalDb; +using ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms; +using ZB.MOM.WW.OtOpcUa.Runtime.ScriptedAlarms; + +namespace ZB.MOM.WW.OtOpcUa.Host.IntegrationTests.LocalDb; + +/// +/// Review R05:4 — the scripted-alarm comment trail moved off the condition row into its own +/// append-only alarm_condition_comment table. This proves the move kept the property that +/// mattered: the trail still reaches the redundant pair peer, over a real loopback h2c transport +/// through the real fail-closed interceptor, with both nodes initialised by the production +/// LocalDbSetup.OnReady. +/// +/// +/// State is written and read through the production +/// on each node, so the ordinal keying, the incremental append and the load-side merge under +/// test are the ones production runs. +/// +[Collection("LocalDbPairConvergence")] +public sealed class AlarmConditionCommentConvergenceTests +{ + private static readonly DateTime T0 = new(2026, 06, 10, 12, 00, 00, DateTimeKind.Utc); + + private static LocalDbAlarmConditionStateStore Store(ILocalDb db) => + new(db, NullLogger.Instance); + + private static CancellationToken Ct => TestContext.Current.CancellationToken; + + private static async Task CommentCountAsync(ILocalDb db, string alarmId) => + (await Store(db).LoadAsync(alarmId, Ct))?.Comments.Count ?? -1; + + [Fact] + public async Task CommentsWrittenOnA_ArriveOnB_AndBsAppendArrivesBackOnA() + { + await using var pair = new LocalDbPairHarness(); + await pair.StartAsync(); + + var storeA = Store(pair.A); + var state = AlarmConditionState.Fresh("eq/alarm-1", T0) with + { + Active = AlarmActiveState.Active, + Acked = AlarmAckedState.Unacknowledged, + }; + state = Part9StateMachine.ApplyAddComment(state, "jane", "first look", T0.AddMinutes(1)).State; + state = Part9StateMachine.ApplyAcknowledge(state, "bob", "on it", T0.AddMinutes(2)).State; + await storeA.SaveAsync(state, Ct); + + await LocalDbPairHarness.WaitUntilAsync( + async () => await CommentCountAsync(pair.B, "eq/alarm-1") == 2, + "node B to hold node A's two comment-trail entries"); + + var onB = await Store(pair.B).LoadAsync("eq/alarm-1", Ct); + onB.ShouldNotBeNull(); + onB.Comments.ShouldBe(state.Comments); + onB.Acked.ShouldBe(AlarmAckedState.Acknowledged); + + // The reverse direction, as an incremental append: B adds one entry to the trail it loaded, + // and only that entry has to travel for A to see the full trail. + var appended = Part9StateMachine.ApplyAddComment(onB, "amy", "from B", T0.AddMinutes(3)).State; + await Store(pair.B).SaveAsync(appended, Ct); + + await LocalDbPairHarness.WaitUntilAsync( + async () => await CommentCountAsync(pair.A, "eq/alarm-1") == 3, + "node A to receive the entry node B appended"); + + (await storeA.LoadAsync("eq/alarm-1", Ct))!.Comments.ShouldBe(appended.Comments); + } + + [Fact] + public async Task Failover_AfterAMissedCommand_BothTheMissedEntryAndTheSurvivorsNextEntry_ReachBothPeers() + { + // Review I-1, end to end over the real pair: + // 1. a shared trail on both nodes; + // 2. an AddComment reaches only node A (B's command delivery failed) — A's entry replicates + // into B's table, but B's engine never applied it, so B's IN-MEMORY trail is one short; + // 3. A goes away (link down) and the operator acks on the survivor B — B's in-memory Ack + // collides with A's entry at the same ordinal; + // 4. the pair rejoins. + // Both entries must survive on both peers. Before the fix B's Ack was silently dropped. + await using var pair = new LocalDbPairHarness(); + await pair.StartAsync(); + + var shared = AlarmConditionState.Fresh("eq/alarm-fo", T0) with + { + Active = AlarmActiveState.Active, + Acked = AlarmAckedState.Unacknowledged, + }; + for (var i = 0; i < 3; i++) + shared = Part9StateMachine.ApplyAddComment(shared, "jane", $"note {i}", T0.AddMinutes(i)).State; + await Store(pair.A).SaveAsync(shared, Ct); + await LocalDbPairHarness.WaitUntilAsync( + async () => await CommentCountAsync(pair.B, "eq/alarm-fo") == 3, + "node B to hold the shared trail"); + + var onlyOnA = Part9StateMachine.ApplyAddComment(shared, "jane", "seen on A only", T0.AddHours(1)).State; + await Store(pair.A).SaveAsync(onlyOnA, Ct); + await LocalDbPairHarness.WaitUntilAsync( + async () => await CommentCountAsync(pair.B, "eq/alarm-fo") == 4, + "node A's entry to replicate into node B's table"); + + await pair.StopPassiveAsync(); + + // B's engine still holds the 3-entry trail it applied; the operator acks there. + var ackOnB = Part9StateMachine.ApplyAcknowledge(shared, "bob", "acked on B", T0.AddHours(2)).State; + await Store(pair.B).SaveAsync(ackOnB, Ct); + + await pair.RestartPairAsync(); + + await LocalDbPairHarness.WaitUntilAsync( + async () => await CommentCountAsync(pair.A, "eq/alarm-fo") == 5 + && await CommentCountAsync(pair.B, "eq/alarm-fo") == 5, + "both nodes to hold the shared trail + A's missed-by-B entry + B's ack"); + + var onA = (await Store(pair.A).LoadAsync("eq/alarm-fo", Ct))!; + var onB = (await Store(pair.B).LoadAsync("eq/alarm-fo", Ct))!; + onA.Comments.ShouldBe(onB.Comments); + onA.Comments[3].ShouldBe(onlyOnA.Comments[3]); + onA.Comments[4].ShouldBe(ackOnB.Comments[3]); + + // Held, not merely reached: nothing late duplicates either entry. + (await LocalDbPairHarness.HeldWithinAsync( + async () => await CommentCountAsync(pair.A, "eq/alarm-fo") != 5 + || await CommentCountAsync(pair.B, "eq/alarm-fo") != 5, + TimeSpan.FromSeconds(2))) + .ShouldBeFalse(); + } + + [Fact] + public async Task BothNodesRecordingTheSameEntry_ConvergeToOneEntry_NotTwo() + { + // Both pair nodes run the engine and both apply every operator command, each stamping the + // entry with its own clock. Ordinal keying is what collapses the two copies to one entry; + // a content-derived key would leave the trail doubled on both nodes. + await using var pair = new LocalDbPairHarness(); + await pair.StartAsync(); + + var fresh = AlarmConditionState.Fresh("eq/alarm-2", T0); + var onA = Part9StateMachine.ApplyAddComment(fresh, "jane", "checked", T0.AddSeconds(1)).State; + var onB = Part9StateMachine.ApplyAddComment(fresh, "jane", "checked", T0.AddSeconds(1).AddMilliseconds(7)).State; + await Store(pair.A).SaveAsync(onA, Ct); + await Store(pair.B).SaveAsync(onB, Ct); + + // Positive control: a DISTINCT second alarm written only on B. Without it "one entry on + // both" would also hold for two databases that never replicated at all. + var onlyOnB = Part9StateMachine.ApplyAddComment( + AlarmConditionState.Fresh("eq/only-on-b", T0), "amy", "b only", T0.AddSeconds(2)).State; + await Store(pair.B).SaveAsync(onlyOnB, Ct); + + await LocalDbPairHarness.WaitUntilAsync( + async () => await CommentCountAsync(pair.A, "eq/only-on-b") == 1, + "node A to receive node B's distinct alarm"); + + // LWW picks ONE node's copy and both nodes end up holding that same copy. + await LocalDbPairHarness.WaitUntilAsync( + async () => + { + var a = await Store(pair.A).LoadAsync("eq/alarm-2", Ct); + var b = await Store(pair.B).LoadAsync("eq/alarm-2", Ct); + return a is not null && b is not null + && a.Comments.Count == 1 && b.Comments.Count == 1 + && a.Comments[0] == b.Comments[0]; + }, + "both nodes to agree on a single copy of the shared entry"); + + // Held, not merely reached: nothing late may double the trail on either node. + (await LocalDbPairHarness.HeldWithinAsync( + async () => await CommentCountAsync(pair.A, "eq/alarm-2") != 1 + || await CommentCountAsync(pair.B, "eq/alarm-2") != 1, + TimeSpan.FromSeconds(2))) + .ShouldBeFalse("the shared entry must stay one entry on both nodes, never two"); + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDbSetupTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDbSetupTests.cs index 197439df..06351576 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDbSetupTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDbSetupTests.cs @@ -134,7 +134,7 @@ public sealed class LocalDbSetupTests : IDisposable TestContext.Current.CancellationToken); [Fact] - public void OnReady_WhenReplicationIsConfigured_RegistersExactlyTheFourReplicatedTables() + public void OnReady_WhenReplicationIsConfigured_RegistersExactlyTheFiveReplicatedTables() { // Both directions are load-bearing. "No fewer" catches a dropped RegisterReplicated call // (that table then never replicates). "No more" catches an accidental registration — @@ -142,11 +142,11 @@ public sealed class LocalDbSetupTests : IDisposable var db = Boot(InitiatorReplication); db.ReplicatedTables.Keys.OrderBy(k => k, StringComparer.Ordinal) - .ShouldBe(["alarm_condition_state", "alarm_sf_events", "deployment_artifacts", "deployment_pointer"]); + .ShouldBe(["alarm_condition_comment", "alarm_condition_state", "alarm_sf_events", "deployment_artifacts", "deployment_pointer"]); } [Fact] - public void OnReady_WhenOnlyTheSyncApiKeyIsSet_StillRegistersTheFourTables() + public void OnReady_WhenOnlyTheSyncApiKeyIsSet_StillRegistersTheFiveTables() { // The passive half of a pair carries the key and nothing else. If the predicate were // "PeerAddress is set", this node would install no capture triggers and its OWN writes @@ -155,7 +155,7 @@ public sealed class LocalDbSetupTests : IDisposable var db = Boot(PassiveReplication); db.ReplicatedTables.Keys.OrderBy(k => k, StringComparer.Ordinal) - .ShouldBe(["alarm_condition_state", "alarm_sf_events", "deployment_artifacts", "deployment_pointer"]); + .ShouldBe(["alarm_condition_comment", "alarm_condition_state", "alarm_sf_events", "deployment_artifacts", "deployment_pointer"]); } [Fact] @@ -216,6 +216,56 @@ public sealed class LocalDbSetupTests : IDisposable db.ReplicatedTables["alarm_condition_state"].PkColumns.ShouldBe(["scripted_alarm_id"]); } + [Fact] + public void AlarmConditionComment_PkIsTheAlarmIdPlusTheTrailOrdinal() + { + // One row per trail entry. Keyed by ordinal (not content) so the pair's two copies of the + // same logical entry — each stamped by its own node's clock — collapse to one row under LWW. + var db = Boot(InitiatorReplication); + + db.ReplicatedTables["alarm_condition_comment"].PkColumns.ShouldBe(["scripted_alarm_id", "seq"]); + } + + [Fact] + public async Task OnReady_MovesAnOldShapeInlineCommentTrail_AfterRegistration_SoTheMoveReplicates() + { + // A file written by the pre-split store: the trail is inline on the condition row. + var first = Boot(InitiatorReplication); + await first.ExecuteAsync( + """ + INSERT INTO alarm_condition_state + (scripted_alarm_id, enabled_state, acked_state, confirmed_state, shelving_state, + comments_json, updated_at_utc) + VALUES ('eq/alarm-legacy', 'Enabled', 'Acknowledged', 'Confirmed', 'Unshelved', + '[{"timestampUtc":"2026-06-10T10:00:00Z","user":"jane","kind":"Acknowledge","text":"a"},{"timestampUtc":"2026-06-10T10:05:00Z","user":"bob","kind":"Confirm","text":"b"}]', + '2026-06-10T10:05:00.0000000Z') + """, + parameters: null, + TestContext.Current.CancellationToken); + // Positive control: before the restart the trail is inline and the comment table is empty. + (await ScalarAsync(first, "SELECT COUNT(*) FROM alarm_condition_comment")).ShouldBe(0); + + // The upgrade restart. + var db = Boot(InitiatorReplication); + + (await ScalarAsync(db, "SELECT COUNT(*) FROM alarm_condition_comment WHERE scripted_alarm_id = 'eq/alarm-legacy'")) + .ShouldBe(2); + // The condition row itself is NOT rewritten (review m1): a migration UPDATE would re-stamp it + // with a fresh HLC and could beat a newer peer state under LWW. Only the seeding INSERT is in + // the oplog for it; the alarm's next save clears the inline column. + (await ScalarAsync(db, "SELECT COUNT(*) FROM __localdb_oplog WHERE table_name = 'alarm_condition_state'")) + .ShouldBe(1); + // Captured: the migration ran after RegisterReplicated, so the moved entries reach the peer. + (await ScalarAsync(db, "SELECT COUNT(*) FROM __localdb_oplog WHERE table_name = 'alarm_condition_comment'")) + .ShouldBe(2); + + // Idempotent: a further boot writes nothing more. + var third = Boot(InitiatorReplication); + (await ScalarAsync(third, "SELECT COUNT(*) FROM __localdb_oplog WHERE table_name = 'alarm_condition_comment'")) + .ShouldBe(2); + (await ScalarAsync(third, "SELECT COUNT(*) FROM alarm_condition_comment")).ShouldBe(2); + } + [Fact] public async Task AlarmSfEventRows_EnterTheOplog() { @@ -296,7 +346,7 @@ public sealed class LocalDbSetupTests : IDisposable var replicated = Boot(InitiatorReplication); await InsertPointerAsync(replicated, "SITE-A"); (await ScalarAsync(replicated, "SELECT COUNT(*) FROM sqlite_master WHERE type = 'trigger'")) - .ShouldBe(12); + .ShouldBe(15); var unreplicated = Boot(); @@ -350,6 +400,45 @@ public sealed class LocalDbSetupTests : IDisposable sink.Events.Count(e => e.Level == LogEventLevel.Information).ShouldBe(1); } + [Fact] + public async Task OnReady_WarnsOncePerAlarmWhoseInlineCommentTrailIsUnreadable() + { + // Review m2: an unmigratable history must be visible, not skipped silently on every boot. + var first = Boot(); + await first.ExecuteAsync( + """ + INSERT INTO alarm_condition_state + (scripted_alarm_id, enabled_state, acked_state, confirmed_state, shelving_state, + comments_json, updated_at_utc) + VALUES (@Id, 'Enabled', 'Acknowledged', 'Confirmed', 'Unshelved', @Json, '2026-06-10T10:05:00.0000000Z') + """, + new { Id = "eq/alarm-bad", Json = "{not json" }, + TestContext.Current.CancellationToken); + await first.ExecuteAsync( + """ + INSERT INTO alarm_condition_state + (scripted_alarm_id, enabled_state, acked_state, confirmed_state, shelving_state, + comments_json, updated_at_utc) + VALUES (@Id, 'Enabled', 'Acknowledged', 'Confirmed', 'Unshelved', @Json, '2026-06-10T10:05:00.0000000Z') + """, + new + { + Id = "eq/alarm-good", + Json = """[{"timestampUtc":"2026-06-10T10:00:00Z","user":"jane","kind":"AddComment","text":"a"}]""", + }, + TestContext.Current.CancellationToken); + + var sink = new CollectingSink(); + Boot(replication: null, logger: BuildLogger(sink)); + + var warnings = sink.Events.Where(e => e.Level == LogEventLevel.Warning).ToList(); + warnings.Count.ShouldBe(1); + warnings[0].RenderMessage().ShouldContain("eq/alarm-bad"); + // Positive control: the readable row DID move, and said so. + sink.Events.Count(e => e.Level == LogEventLevel.Information && e.RenderMessage().Contains("copied")) + .ShouldBe(1); + } + [Fact] public void DisablingReplication_SaysNothingWhenThereWasNothingToClean() { @@ -365,13 +454,13 @@ public sealed class LocalDbSetupTests : IDisposable [Fact] public void AFailureMidwayThroughRegistration_FailsBootLoudly_AndTheRetriedBootConverges() { - // OnReady flips four tables in a loop with no try/catch, so a failure on the third leaves - // two registered and two not. That is the RIGHT behaviour and this pins both halves of why: + // OnReady flips five tables in a loop with no try/catch, so a failure on the third leaves + // two registered and three not. That is the RIGHT behaviour and this pins both halves of why: // the throw propagates out of the AddZbLocalDb singleton factory and fails host start // (a node half-capturing is a node whose replication is silently partial), and because both // library APIs are idempotent the retried boot completes the flip rather than compounding it. // - // Swallowing the throw would leave the node up, serving, and replicating two of four tables + // Swallowing the throw would leave the node up, serving, and replicating two of five tables // with nothing in the log to say so. var real = Boot(InitiatorReplication); @@ -387,11 +476,11 @@ public sealed class LocalDbSetupTests : IDisposable // pinning some unrelated throw, and the "partial flip" it claims to model never happened. faulty.Calls.ShouldBe(3); - // The retry: same file, no wrapper. Idempotent re-registration converges on all four. + // The retry: same file, no wrapper. Idempotent re-registration converges on all five. LocalDbSetup.OnReady(real, configuration); real.ReplicatedTables.Keys.OrderBy(k => k, StringComparer.Ordinal) - .ShouldBe(["alarm_condition_state", "alarm_sf_events", "deployment_artifacts", "deployment_pointer"]); + .ShouldBe(["alarm_condition_comment", "alarm_condition_state", "alarm_sf_events", "deployment_artifacts", "deployment_pointer"]); } /// diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDbWiringTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDbWiringTests.cs index 6798faf5..5477afea 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDbWiringTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDbWiringTests.cs @@ -108,7 +108,7 @@ public sealed class LocalDbWiringTests : IDisposable // Exact set, both directions — a dropped registration replicates nothing; an extra one costs // three triggers and oplog volume on every write. first.ReplicatedTables.Keys.OrderBy(k => k, StringComparer.Ordinal) - .ShouldBe(["alarm_condition_state", "alarm_sf_events", "deployment_artifacts", "deployment_pointer"]); + .ShouldBe(["alarm_condition_comment", "alarm_condition_state", "alarm_sf_events", "deployment_artifacts", "deployment_pointer"]); } [Fact] diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/ScriptedAlarms/LocalDbAlarmCommentAuditTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/ScriptedAlarms/LocalDbAlarmCommentAuditTests.cs new file mode 100644 index 00000000..e6c422e9 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/ScriptedAlarms/LocalDbAlarmCommentAuditTests.cs @@ -0,0 +1,523 @@ +using System.Collections.Immutable; +using Microsoft.Data.Sqlite; +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging.Abstractions; +using Shouldly; +using Xunit; +using ZB.MOM.WW.LocalDb; +using ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian; +using ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms; +using ZB.MOM.WW.OtOpcUa.Runtime.ScriptedAlarms; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.ScriptedAlarms; + +/// +/// The alarm comment audit trail lives in its own append-only table +/// () keyed by alarm id, so the condition +/// row stays a fixed size and a transition's write — and therefore its replication payload — +/// does not grow with the alarm's history. +/// +/// +/// The payload is measured where it costs: the bytes the LocalDb capture triggers write into +/// __localdb_oplog for one save, which is exactly what ships to the pair peer. Both +/// tables are registered for capture here, as they are on a replicating node. +/// +public sealed class LocalDbAlarmCommentAuditTests : IDisposable +{ + private static readonly DateTime T0 = new(2026, 06, 10, 12, 00, 00, DateTimeKind.Utc); + + private readonly string _dbPath = + Path.Combine(Path.GetTempPath(), $"otopcua-alarm-comments-{Guid.NewGuid():N}.db"); + + private readonly ServiceProvider _provider; + private readonly ILocalDb _db; + + public LocalDbAlarmCommentAuditTests() + { + var configuration = new ConfigurationBuilder() + .AddInMemoryCollection(new Dictionary { ["LocalDb:Path"] = _dbPath }) + .Build(); + + _provider = new ServiceCollection() + .AddZbLocalDb(configuration, db => + { + using (var connection = db.CreateConnection()) + { + AlarmConditionStateSchema.Apply(connection); + } + + db.RegisterReplicated(AlarmConditionStateSchema.StateTable); + db.RegisterReplicated(AlarmConditionStateSchema.CommentsTable); + }) + .BuildServiceProvider(); + + _db = _provider.GetRequiredService(); + } + + public void Dispose() + { + _provider.Dispose(); + SqliteConnection.ClearAllPools(); + foreach (var path in new[] { _dbPath, $"{_dbPath}-wal", $"{_dbPath}-shm" }) + { + try { if (File.Exists(path)) File.Delete(path); } catch { /* best effort */ } + } + } + + private LocalDbAlarmConditionStateStore NewStore() + => new(_db, NullLogger.Instance); + + private static CancellationToken Ct => CancellationToken.None; + + private static ImmutableList Comments(int count, string prefix = "c") + { + var builder = ImmutableList.CreateBuilder(); + for (var i = 0; i < count; i++) + builder.Add(new AlarmComment(T0.AddSeconds(i), "jane", "AddComment", $"{prefix}-{i:D6}")); + return builder.ToImmutable(); + } + + /// An unacknowledged state carrying , ready to be acked. + private static AlarmConditionState Unacked(string alarmId, ImmutableList comments) => + AlarmConditionState.Fresh(alarmId, T0) with + { + Active = AlarmActiveState.Active, + Acked = AlarmAckedState.Unacknowledged, + Comments = comments, + }; + + private async Task OplogHighWaterAsync() => + (await _db.QueryAsync("SELECT COALESCE(MAX(seq), 0) FROM __localdb_oplog", r => r.GetInt64(0), null, Ct))[0]; + + /// Bytes captured into the oplog above — the replication payload. + private async Task OplogBytesAboveAsync(long afterSeq) => + (await _db.QueryAsync( + "SELECT COALESCE(SUM(LENGTH(pk_json) + COALESCE(LENGTH(row_json), 0)), 0) FROM __localdb_oplog WHERE seq > @S", + r => r.GetInt64(0), + new { S = afterSeq }, + Ct))[0]; + + private async Task OplogEntriesAboveAsync(long afterSeq, string table) => + (await _db.QueryAsync( + "SELECT COUNT(*) FROM __localdb_oplog WHERE seq > @S AND table_name = @T", + r => r.GetInt64(0), + new { S = afterSeq, T = table }, + Ct))[0]; + + private async Task ScalarAsync(string sql) => + (await _db.QueryAsync(sql, r => r.GetInt64(0), null, Ct))[0]; + + /// + /// Seeds with comments, then + /// returns the oplog bytes one acknowledge transition (one new comment) costs. + /// + private async Task AckTransitionBytesAsync(string alarmId, int historyLength, string ackComment) + { + var store = NewStore(); + var seeded = Unacked(alarmId, Comments(historyLength)); + await store.SaveAsync(seeded, Ct); + + var before = await OplogHighWaterAsync(); + var acked = Part9StateMachine.ApplyAcknowledge(seeded, "bob", ackComment, T0.AddDays(1)).State; + await store.SaveAsync(acked, Ct); + return await OplogBytesAboveAsync(before); + } + + [Fact] + public async Task TransitionPayload_IsConstant_WhetherTheAlarmCarries10OrTenThousandComments() + { + var small = await AckTransitionBytesAsync("alarm-small", 10, "ack"); + var large = await AckTransitionBytesAsync("alarm-large", 10_000, "ack"); + + // Constant: the only difference allowed is the width of the new comment's ordinal ("10" vs + // "10000", in pk_json and row_json) — a handful of bytes, not O(n). The alarm ids are the + // same length on purpose. + small.ShouldBeGreaterThan(0); + Math.Abs(large - small).ShouldBeLessThanOrEqualTo(16, + $"one ack transition cost {small} bytes at 10 comments but {large} at 10,000"); + large.ShouldBeLessThan(2_048); + + // Positive control on the instrument: the seeded history really is large and really was + // captured — otherwise a "constant" result could just mean capture is off. + (await ScalarAsync("SELECT COUNT(*) FROM alarm_condition_comment WHERE scripted_alarm_id = 'alarm-large'")) + .ShouldBe(10_001); + var historyBytes = await ScalarAsync( + "SELECT SUM(LENGTH(row_json)) FROM __localdb_oplog WHERE table_name = 'alarm_condition_comment'"); + historyBytes.ShouldBeGreaterThan(1_000_000); + } + + [Fact] + public async Task TransitionPayload_PositiveControl_TheMeasurementSeesTheNewCommentsBytes() + { + // Proves the measurement above is not vacuous: the oplog delta of a transition does carry + // the appended comment, so a 4 KB comment costs at least 4 KB. + var plain = await AckTransitionBytesAsync("alarm-plain", 10, "ack"); + var heavy = await AckTransitionBytesAsync("alarm-heavy", 10, new string('x', 4_096)); + + (heavy - plain).ShouldBeGreaterThanOrEqualTo(4_096 * 2 - 16, + "the comment text rides both the comment row and last_ack_comment on the condition row"); + } + + [Fact] + public async Task Transition_WritesOneConditionRowAndOnlyTheAppendedCommentRow() + { + var store = NewStore(); + var seeded = Unacked("alarm-append", Comments(50)); + await store.SaveAsync(seeded, Ct); + + var before = await OplogHighWaterAsync(); + var acked = Part9StateMachine.ApplyAcknowledge(seeded, "bob", "ack", T0.AddDays(1)).State; + await store.SaveAsync(acked, Ct); + + (await OplogEntriesAboveAsync(before, AlarmConditionStateSchema.StateTable)).ShouldBe(1); + (await OplogEntriesAboveAsync(before, AlarmConditionStateSchema.CommentsTable)).ShouldBe(1); + + var loaded = await store.LoadAsync("alarm-append", Ct); + loaded.ShouldNotBeNull(); + loaded.Comments.ShouldBe(acked.Comments); + } + + [Fact] + public async Task ConditionRow_StaysFixedSize_InlineColumnIsAlwaysEmpty() + { + var store = NewStore(); + await store.SaveAsync(Unacked("alarm-fixed", Comments(500)), Ct); + + (await _db.QueryAsync( + "SELECT comments_json FROM alarm_condition_state WHERE scripted_alarm_id = 'alarm-fixed'", + r => r.GetString(0), null, Ct))[0] + .ShouldBe("[]"); + (await store.LoadAsync("alarm-fixed", Ct))!.Comments.Count.ShouldBe(500); + } + + [Fact] + public async Task UnchangedSave_WritesNoCommentRows() + { + var store = NewStore(); + var state = Unacked("alarm-idle", Comments(20)); + await store.SaveAsync(state, Ct); + + var before = await OplogHighWaterAsync(); + await store.SaveAsync(state with { LastTransitionUtc = T0.AddMinutes(1) }, Ct); + + (await OplogEntriesAboveAsync(before, AlarmConditionStateSchema.CommentsTable)).ShouldBe(0); + (await OplogEntriesAboveAsync(before, AlarmConditionStateSchema.StateTable)).ShouldBe(1, + "positive control: the save itself was captured"); + } + + [Fact] + public async Task LoadAll_ReturnsEachAlarmsOwnTrailInOrder() + { + var store = NewStore(); + var a = Unacked("alarm-a", Comments(3, "a")); + var b = Unacked("alarm-b", Comments(5, "b")); + await store.SaveAsync(a, Ct); + await store.SaveAsync(b, Ct); + + var all = (await store.LoadAllAsync(Ct)).ToDictionary(s => s.AlarmId); + + all["alarm-a"].Comments.ShouldBe(a.Comments); + all["alarm-b"].Comments.ShouldBe(b.Comments); + } + + [Fact] + public async Task Remove_DeletesTheAlarmsTrail_AndAReCreatedAlarmStartsEmpty() + { + var store = NewStore(); + await store.SaveAsync(Unacked("alarm-gone", Comments(4)), Ct); + await store.SaveAsync(Unacked("alarm-kept", Comments(2)), Ct); + + await store.RemoveAsync("alarm-gone", Ct); + + (await ScalarAsync("SELECT COUNT(*) FROM alarm_condition_comment WHERE scripted_alarm_id = 'alarm-gone'")) + .ShouldBe(0); + (await store.LoadAsync("alarm-kept", Ct))!.Comments.Count.ShouldBe(2, "positive control"); + + var recreated = AlarmConditionState.Fresh("alarm-gone", T0) + with { Comments = ImmutableList.Create(new AlarmComment(T0, "amy", "AddComment", "new life")) }; + await store.SaveAsync(recreated, Ct); + (await store.LoadAsync("alarm-gone", Ct))!.Comments.ShouldBe(recreated.Comments); + } + + [Fact] + public async Task CommentsAlreadyHeldAtAnOrdinal_AreNeverRewritten() + { + // Append-only: a save whose list disagrees with what is stored at an existing ordinal (the + // peer's copy of the same logical entry already arrived) leaves the stored entry alone. + var store = NewStore(); + var original = Unacked("alarm-ao", Comments(3)); + await store.SaveAsync(original, Ct); + + var rewritten = original with { Comments = original.Comments.SetItem(1, new AlarmComment(T0, "eve", "AddComment", "forged")) }; + await store.SaveAsync(rewritten, Ct); + + // The stored entry at ordinal 1 is untouched; the unrecognised local entry is kept by + // appending it (never dropped), not by overwriting history. + var loaded = (await store.LoadAsync("alarm-ao", Ct))!.Comments; + loaded.Take(3).ShouldBe(original.Comments); + loaded.Count.ShouldBe(4); + loaded[3].Text.ShouldBe("forged"); + } + + // ---- back-compat + migration ----------------------------------------------------------- + + /// + /// The exact shape the pre-split store wrote into comments_json + /// (JsonSerializerDefaults.Web: camelCase, Utc timestamps with a trailing Z). + /// + private const string OldShapeCommentsJson = + """[{"timestampUtc":"2026-06-10T10:00:00Z","user":"jane","kind":"Acknowledge","text":"checking pump"},{"timestampUtc":"2026-06-10T11:30:00.1234567Z","user":"bob","kind":"Confirm","text":"all clear"},{"timestampUtc":"2026-06-10T11:30:00.1234567Z","user":"system","kind":"AutoUnshelve","text":""}]"""; + + private static readonly ImmutableList OldShapeComments = ImmutableList.Create( + new AlarmComment(new DateTime(2026, 06, 10, 10, 00, 00, DateTimeKind.Utc), "jane", "Acknowledge", "checking pump"), + new AlarmComment(new DateTime(2026, 06, 10, 11, 30, 00, DateTimeKind.Utc).AddTicks(1_234_567), "bob", "Confirm", "all clear"), + new AlarmComment(new DateTime(2026, 06, 10, 11, 30, 00, DateTimeKind.Utc).AddTicks(1_234_567), "system", "AutoUnshelve", "")); + + private async Task InsertOldShapeRowAsync(string alarmId, string commentsJson = OldShapeCommentsJson) => + await _db.ExecuteAsync( + """ + INSERT INTO alarm_condition_state + (scripted_alarm_id, enabled_state, acked_state, confirmed_state, shelving_state, + shelving_expires_utc, last_ack_user, last_ack_comment, last_ack_utc, + last_confirm_user, last_confirm_comment, last_confirm_utc, comments_json, updated_at_utc) + VALUES + (@Id, 'Enabled', 'Acknowledged', 'Confirmed', 'Unshelved', + NULL, 'jane', 'checking pump', '2026-06-10T10:00:00.0000000Z', + 'bob', 'all clear', '2026-06-10T11:30:00.1234567Z', @Json, '2026-06-10T11:30:00.1234567Z') + """, + new { Id = alarmId, Json = commentsJson }, + Ct); + + private static void ShouldBeTheOldShapeState(AlarmConditionState? loaded, string alarmId) + { + loaded.ShouldNotBeNull(); + loaded.AlarmId.ShouldBe(alarmId); + loaded.Acked.ShouldBe(AlarmAckedState.Acknowledged); + loaded.Confirmed.ShouldBe(AlarmConfirmedState.Confirmed); + loaded.LastAckUser.ShouldBe("jane"); + loaded.LastConfirmComment.ShouldBe("all clear"); + loaded.Comments.ShouldBe(OldShapeComments); + loaded.Comments.ShouldAllBe(c => c.TimestampUtc.Kind == DateTimeKind.Utc); + } + + [Fact] + public async Task OldShapeRow_WithInlineComments_LoadsExactlyAsBefore() + { + await InsertOldShapeRowAsync("alarm-legacy"); + + ShouldBeTheOldShapeState(await NewStore().LoadAsync("alarm-legacy", Ct), "alarm-legacy"); + ShouldBeTheOldShapeState((await NewStore().LoadAllAsync(Ct)).Single(), "alarm-legacy"); + } + + [Fact] + public async Task Migration_CopiesInlineCommentsToTheAuditTable_AndNeverRewritesTheConditionRow() + { + await InsertOldShapeRowAsync("alarm-m1"); + await InsertOldShapeRowAsync("alarm-m2"); + await InsertOldShapeRowAsync("alarm-empty", "[]"); + + var before = await OplogHighWaterAsync(); + AlarmConditionStateSchema.InlineCommentMigrationResult result; + using (var connection = _db.CreateConnection()) + result = AlarmConditionStateSchema.MigrateInlineComments(connection); + + result.RowsMoved.ShouldBe(2); + result.UnparseableAlarmIds.ShouldBeEmpty(); + (await ScalarAsync("SELECT COUNT(*) FROM alarm_condition_comment")).ShouldBe(6); + + // Review m1: the migration must not touch the condition row. A full-row UPDATE re-stamps it + // with a fresh HLC and can revert a newer peer state (e.g. an ack) under LWW. The next save + // (the engine's boot save) resets the column instead. + (await OplogEntriesAboveAsync(before, AlarmConditionStateSchema.StateTable)).ShouldBe(0); + (await OplogEntriesAboveAsync(before, AlarmConditionStateSchema.CommentsTable)).ShouldBe(6, + "positive control: the copied entries were captured"); + + ShouldBeTheOldShapeState(await NewStore().LoadAsync("alarm-m1", Ct), "alarm-m1"); + ShouldBeTheOldShapeState(await NewStore().LoadAsync("alarm-m2", Ct), "alarm-m2"); + (await NewStore().LoadAsync("alarm-empty", Ct))!.Comments.ShouldBeEmpty(); + } + + [Fact] + public async Task Migration_IsIdempotent_ASecondRunChangesNothing() + { + await InsertOldShapeRowAsync("alarm-idem"); + using (var connection = _db.CreateConnection()) + AlarmConditionStateSchema.MigrateInlineComments(connection).RowsMoved.ShouldBe(1, "positive control"); + + var before = await OplogHighWaterAsync(); + using (var connection = _db.CreateConnection()) + AlarmConditionStateSchema.MigrateInlineComments(connection).RowsMoved.ShouldBe(0); + + (await OplogHighWaterAsync()).ShouldBe(before, "a re-run must write nothing"); + (await ScalarAsync("SELECT COUNT(*) FROM alarm_condition_comment")).ShouldBe(3); + ShouldBeTheOldShapeState(await NewStore().LoadAsync("alarm-idem", Ct), "alarm-idem"); + } + + [Fact] + public async Task Migration_OverAPartiallyMigratedRow_FillsOnlyTheMissingOrdinals() + { + // Only ordinal 0 made it across (e.g. a restored backup): the run fills 1 and 2, keeps 0, + // duplicates nothing. + await InsertOldShapeRowAsync("alarm-partial"); + await _db.ExecuteAsync( + AlarmConditionStateSchema.InsertCommentSql, + new + { + Id = "alarm-partial", Seq = 0, + Ts = AlarmConditionStateSchema.FormatTimestamp(OldShapeComments[0].TimestampUtc), + User = OldShapeComments[0].User, Kind = OldShapeComments[0].Kind, Text = OldShapeComments[0].Text, + }, + Ct); + + using (var connection = _db.CreateConnection()) + AlarmConditionStateSchema.MigrateInlineComments(connection).RowsMoved.ShouldBe(1); + + (await ScalarAsync("SELECT COUNT(*) FROM alarm_condition_comment")).ShouldBe(3); + ShouldBeTheOldShapeState(await NewStore().LoadAsync("alarm-partial", Ct), "alarm-partial"); + } + + [Fact] + public async Task Migration_ReportsAnUnparseableInlineTrail_AndLeavesItAlone() + { + // Review m2: an unreadable trail must surface (OnReady logs a Warning per id), not be + // skipped silently on every boot. + await InsertOldShapeRowAsync("alarm-bad", "{not json"); + await InsertOldShapeRowAsync("alarm-good"); + + AlarmConditionStateSchema.InlineCommentMigrationResult result; + using (var connection = _db.CreateConnection()) + result = AlarmConditionStateSchema.MigrateInlineComments(connection); + + result.UnparseableAlarmIds.ShouldBe(["alarm-bad"]); + result.RowsMoved.ShouldBe(1, "positive control: the good row still moved"); + (await _db.QueryAsync( + "SELECT comments_json FROM alarm_condition_state WHERE scripted_alarm_id = 'alarm-bad'", + r => r.GetString(0), null, Ct))[0] + .ShouldBe("{not json"); + } + + // ---- divergence / failover (review I-1) ------------------------------------------------- + + /// Captures log entries so a divergence Warning can be asserted, not inferred. + private sealed class CapturingLogger : Microsoft.Extensions.Logging.ILogger + { + public List<(Microsoft.Extensions.Logging.LogLevel Level, string Message)> Entries { get; } = []; + + public IDisposable? BeginScope(TState state) where TState : notnull => null; + + public bool IsEnabled(Microsoft.Extensions.Logging.LogLevel logLevel) => true; + + public void Log( + Microsoft.Extensions.Logging.LogLevel logLevel, Microsoft.Extensions.Logging.EventId eventId, + TState state, Exception? exception, Func formatter) + { + lock (Entries) Entries.Add((logLevel, formatter(state, exception))); + } + + public int Warnings => Entries.Count(e => e.Level == Microsoft.Extensions.Logging.LogLevel.Warning); + } + + [Fact] + public async Task Failover_TheSurvivorsNextEntry_IsPersisted_AfterItMissedACommandThePeerApplied() + { + // The I-1 sequence on node B's database: + // 1. a shared trail of 5 entries; + // 2. an AddComment reaches only node A (B's command delivery failed once) and A's row at + // ordinal 5 replicates into B's table — B's IN-MEMORY trail still has 5 entries; + // 3. A dies and the operator acks on B: B's in-memory Ack sits at index 5, colliding with + // A's entry at ordinal 5. + // The Ack must be persisted (and say so), not silently treated as "already stored". + var shared = Unacked("alarm-fo", Comments(5)); + var a = NewStore(); + await a.SaveAsync(shared, Ct); + var onlyOnA = Part9StateMachine.ApplyAddComment(shared, "jane", "seen on A only", T0.AddHours(1)).State; + await a.SaveAsync(onlyOnA, Ct); + + var log = new CapturingLogger(); + var b = new LocalDbAlarmConditionStateStore(_db, log); + var ackOnB = Part9StateMachine.ApplyAcknowledge(shared, "bob", "acked on B", T0.AddHours(2)).State; + await b.SaveAsync(ackOnB, Ct); + + var loaded = (await NewStore().LoadAsync("alarm-fo", Ct))!; + loaded.Comments.Count.ShouldBe(7); + loaded.Comments[5].ShouldBe(onlyOnA.Comments[5]); + loaded.Comments[6].ShouldBe(ackOnB.Comments[5]); + log.Warnings.ShouldBe(1, "a divergence is never silent"); + + // Re-saving with no new entry must not append the Ack a second time (its own earlier + // divergence-append shifted it off its in-memory index). + var before = await OplogHighWaterAsync(); + await b.SaveAsync(ackOnB with { LastTransitionUtc = T0.AddHours(3) }, Ct); + (await OplogEntriesAboveAsync(before, AlarmConditionStateSchema.CommentsTable)).ShouldBe(0); + (await OplogEntriesAboveAsync(before, AlarmConditionStateSchema.StateTable)).ShouldBe(1, + "positive control: the re-save itself was captured"); + + // And B's following entry (still one behind in memory) is persisted too, exactly once. + var confirmOnB = Part9StateMachine.ApplyAddComment(ackOnB, "bob", "follow-up on B", T0.AddHours(4)).State; + await b.SaveAsync(confirmOnB, Ct); + await b.SaveAsync(confirmOnB, Ct); + var after = (await NewStore().LoadAsync("alarm-fo", Ct))!; + after.Comments.Count.ShouldBe(8); + after.Comments[7].ShouldBe(confirmOnB.Comments[6]); + } + + [Fact] + public async Task PeersCopyOfTheSameCommand_IsNotAppendedAgain_AndLogsNothing() + { + // The positive control for the divergence path: in normal operation both nodes apply every + // command, each with its own clock. When the peer's copy is already at the ordinal, the + // local copy is the same entry and must NOT be appended — otherwise the fix would double + // every entry on a healthy pair. + var shared = Unacked("alarm-dual", Comments(5)); + await NewStore().SaveAsync(shared, Ct); + var ackOnA = Part9StateMachine.ApplyAcknowledge(shared, "bob", "on it", T0.AddHours(1)).State; + await NewStore().SaveAsync(ackOnA, Ct); + + var log = new CapturingLogger(); + var b = new LocalDbAlarmConditionStateStore(_db, log); + var ackOnB = Part9StateMachine.ApplyAcknowledge(shared, "bob", "on it", T0.AddHours(1).AddMilliseconds(40)).State; + var before = await OplogHighWaterAsync(); + await b.SaveAsync(ackOnB, Ct); + + (await OplogEntriesAboveAsync(before, AlarmConditionStateSchema.CommentsTable)).ShouldBe(0); + (await NewStore().LoadAsync("alarm-dual", Ct))!.Comments.Count.ShouldBe(6); + log.Warnings.ShouldBe(0); + } + + [Fact] + public async Task ARepeatedIdenticalComment_IsKept_AsItsOwnEntry() + { + // Matching ignores the timestamp only within the clock-skew window; an operator repeating the + // same text later is a new entry, not a duplicate of the earlier one. + var store = NewStore(); + var s = Unacked("alarm-repeat", Comments(2)); + s = Part9StateMachine.ApplyAddComment(s, "amy", "still high", T0.AddHours(1)).State; + await store.SaveAsync(s, Ct); + s = Part9StateMachine.ApplyAddComment(s, "amy", "still high", T0.AddHours(2)).State; + await store.SaveAsync(s, Ct); + + (await NewStore().LoadAsync("alarm-repeat", Ct))!.Comments.ShouldBe(s.Comments); + } + + [Fact] + public async Task SavingALoadedOldShapeState_MovesItsCommentsAndAppendsTheNewOne() + { + // The lazy path: an old-shape row the boot migration never saw (it was written before the + // first save on this build) still converges on the first save. + await InsertOldShapeRowAsync("alarm-lazy"); + var store = NewStore(); + var loaded = await store.LoadAsync("alarm-lazy", Ct); + + var next = Part9StateMachine.ApplyAddComment(loaded!, "amy", "follow-up", T0.AddDays(2)).State; + await store.SaveAsync(next, Ct); + + (await ScalarAsync("SELECT COUNT(*) FROM alarm_condition_comment WHERE scripted_alarm_id = 'alarm-lazy'")) + .ShouldBe(4); + (await _db.QueryAsync( + "SELECT comments_json FROM alarm_condition_state WHERE scripted_alarm_id = 'alarm-lazy'", + r => r.GetString(0), null, Ct))[0] + .ShouldBe("[]"); + (await NewStore().LoadAsync("alarm-lazy", Ct))!.Comments.ShouldBe(next.Comments); + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/ScriptedAlarms/LocalDbAlarmConditionStateStoreTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/ScriptedAlarms/LocalDbAlarmConditionStateStoreTests.cs index fc0d9525..20a4085e 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/ScriptedAlarms/LocalDbAlarmConditionStateStoreTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/ScriptedAlarms/LocalDbAlarmConditionStateStoreTests.cs @@ -48,6 +48,7 @@ public sealed class LocalDbAlarmConditionStateStoreTests : IDisposable } db.RegisterReplicated(AlarmConditionStateSchema.StateTable); + db.RegisterReplicated(AlarmConditionStateSchema.CommentsTable); }) .BuildServiceProvider();