From 846c8d6335fe3e2a2628c1efe27534669d427526 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 16:36:33 -0400 Subject: [PATCH 1/2] fix(alarms): move scripted-alarm comment trail to its own append-only table (R05:4) Every Part 9 transition used to re-serialize the alarm's whole comment trail into alarm_condition_state.comments_json and replicate it to the pair peer: O(n) per transition, O(n^2) over an alarm's life (measured: one ack at 10k comments shipped 1,080,553 oplog bytes; now ~1.6 KB, the same as at 10). - New replicated table alarm_condition_comment, PK (scripted_alarm_id, seq), seq = the entry's ordinal in the trail. Ordinal (not content) keying makes the two pair nodes' copies of one logical entry (each stamped by its own clock, since both nodes apply every command) collapse to one row under LWW. - LocalDbAlarmConditionStateStore.SaveAsync appends only entries past the stored MAX(seq) (INSERT OR IGNORE, append-only) and upserts a fixed-size condition row with comments_json = "[]", in one transaction. Remove drops the trail with the row. Load merges the table with any inline trail. - comments_json column kept (always "[]") so the condition table's shape and replication digest are unchanged and old-shape rows still load exactly. - AlarmConditionStateSchema.MigrateInlineComments: idempotent per-row move of inline trails, run by LocalDbSetup.OnReady after RegisterReplicated so the moved entries replicate. Comment table joins ReplicatedTables (now five). Tests: constant transition payload at 10 vs 10k comments (+ instrument positive control), old-shape back-compat load, migration idempotency / partial re-run / lazy move, OnReady boot migration captured in the oplog, and two loopback pair tests (A->B and B->A trail convergence; the shared entry converges to one copy, with a distinct-alarm positive control). Refs #748 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT --- .../AlarmConditionStateSchema.cs | 204 +++++++++- .../Configuration/LocalDbSetup.cs | 22 +- .../LocalDbAlarmConditionStateStore.cs | 261 +++++++----- .../AlarmConditionStateSchemaTests.cs | 18 + .../AlarmConditionCommentConvergenceTests.cs | 117 ++++++ .../LocalDbSetupTests.cs | 67 ++- .../LocalDbWiringTests.cs | 2 +- .../LocalDbAlarmCommentAuditTests.cs | 383 ++++++++++++++++++ .../LocalDbAlarmConditionStateStoreTests.cs | 1 + 9 files changed, 952 insertions(+), 123 deletions(-) create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDb/AlarmConditionCommentConvergenceTests.cs create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/ScriptedAlarms/LocalDbAlarmCommentAuditTests.cs 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..1ec6b043 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,55 @@ 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; +/// moves any inline trail across. +/// +/// +/// 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. +/// /// 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 +98,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 MigrateInlineComments or the next save moves it). 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 +120,151 @@ 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(); } + + /// + /// Moves every condition row's inline comments_json trail into + /// and resets the column to "[]". Idempotent. + /// + /// + /// + /// 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). + /// + /// + /// Each entry lands at its ordinal in the inline array via , + /// so a re-run — or a row whose trail was already (partly) moved — neither duplicates nor + /// reorders anything, and a second run finds no non-empty inline column and writes + /// nothing. Each row moves in its own transaction together with its column reset, so a + /// crash leaves every row either fully moved 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) + /// rather than failing host start. + /// + /// + /// An already-open connection from ILocalDb.CreateConnection(). + /// The number of condition rows whose inline trail was moved. + public static int 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; + foreach (var (alarmId, json) in pending) + { + IReadOnlyList entries; + try + { + entries = ParseInlineComments(json); + } + catch (JsonException) + { + continue; + } + + 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); + insert.ExecuteNonQuery(); + } + + using (var clear = connection.CreateCommand()) + { + clear.Transaction = tx; + clear.CommandText = + "UPDATE alarm_condition_state SET comments_json = @Empty WHERE scripted_alarm_id = @Id"; + clear.Parameters.AddWithValue("@Empty", EmptyInlineComments); + clear.Parameters.AddWithValue("@Id", alarmId); + clear.ExecuteNonQuery(); + } + + tx.Commit(); + migrated++; + } + + return migrated; + } + + /// 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..c841db21 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,15 @@ 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. + // Moves any pre-split inline comment trail (alarm_condition_state.comments_json) into the + // append-only alarm_condition_comment table (review R05:4). Idempotent: a boot after the + // first finds no inline trail and writes nothing. + using (var connection = db.CreateConnection()) + { + AlarmConditionStateSchema.MigrateInlineComments(connection); + } + 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..8437574d 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,52 @@ 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 appends only +/// the entries past the highest ordinal already stored for the alarm, 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 "[]". The "already stored" boundary +/// is read from the table inside the save's transaction rather than cached, so the store +/// holds no per-alarm state and an entry the pair peer already wrote at an ordinal is kept, +/// not duplicated (see for why the ordinal is the key). +/// +/// +/// 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"; + 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 +87,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 +106,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 +126,40 @@ public sealed class LocalDbAlarmConditionStateStore : IAlarmStateStore { ArgumentNullException.ThrowIfNull(state); + await using var tx = await _db.BeginTransactionAsync(ct).ConfigureAwait(false); + + // Append only what lies past the highest ordinal already stored. An indexed MAX over the + // (scripted_alarm_id, seq) primary key — O(log n), and a read, so it costs no replication. + var maxSeq = (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]; + + for (var seq = (int)Math.Max(maxSeq + 1, 0); seq < state.Comments.Count; seq++) + { + var c = state.Comments[seq]; + await tx.ExecuteAsync( + AlarmConditionStateSchema.InsertCommentSql, + new + { + Id = state.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).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,12 +198,14 @@ 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); } @@ -153,47 +213,93 @@ public sealed class LocalDbAlarmConditionStateStore : IAlarmStateStore 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 +326,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..26b1fe14 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDb/AlarmConditionCommentConvergenceTests.cs @@ -0,0 +1,117 @@ +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 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..280ac35a 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,53 @@ 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); + (await ScalarAsync(db, "SELECT COUNT(*) FROM alarm_condition_state WHERE comments_json <> '[]'")) + .ShouldBe(0); + // 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 +343,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(); @@ -365,13 +412,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 +434,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..47438ceb --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/ScriptedAlarms/LocalDbAlarmCommentAuditTests.cs @@ -0,0 +1,383 @@ +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); + + (await store.LoadAsync("alarm-ao", Ct))!.Comments.ShouldBe(original.Comments); + } + + // ---- 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_MovesInlineCommentsToTheAuditTable_AndClearsTheInlineColumn() + { + await InsertOldShapeRowAsync("alarm-m1"); + await InsertOldShapeRowAsync("alarm-m2"); + await InsertOldShapeRowAsync("alarm-empty", "[]"); + + int migrated; + using (var connection = _db.CreateConnection()) + migrated = AlarmConditionStateSchema.MigrateInlineComments(connection); + + migrated.ShouldBe(2); + (await ScalarAsync("SELECT COUNT(*) FROM alarm_condition_comment")).ShouldBe(6); + (await ScalarAsync("SELECT COUNT(*) FROM alarm_condition_state WHERE comments_json <> '[]'")).ShouldBe(0); + + 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).ShouldBe(1, "positive control"); + + var before = await OplogHighWaterAsync(); + using (var connection = _db.CreateConnection()) + AlarmConditionStateSchema.MigrateInlineComments(connection).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_DoesNotDuplicateOrReorder() + { + // An inline row reappearing after its comments were already moved (e.g. a restored backup): + // the ordinals already held are kept, nothing is duplicated. + await InsertOldShapeRowAsync("alarm-partial"); + using (var connection = _db.CreateConnection()) + AlarmConditionStateSchema.MigrateInlineComments(connection); + await _db.ExecuteAsync( + "UPDATE alarm_condition_state SET comments_json = @Json WHERE scripted_alarm_id = 'alarm-partial'", + new { Json = OldShapeCommentsJson }, Ct); + + using (var connection = _db.CreateConnection()) + AlarmConditionStateSchema.MigrateInlineComments(connection).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 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(); From dad28d37a2f16ae96702df72aae37126d764de47 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 16:58:46 -0400 Subject: [PATCH 2/2] fix(alarms): never drop a comment-trail entry on pair divergence (E5 review r1) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit I-1: a node whose in-memory trail is one short of the stored trail (it missed a command the peer applied, whose entry then replicated in) silently dropped its own next entry — in failover, the survivor's newest ack — because the positional "past MAX(seq)" rule treated it as already stored. SaveAsync now reconciles the last 16 in-memory entries against the stored rows from that position on, by content (Kind/User/Text, timestamps within a 5-minute skew window; matched rows are consumed so repeats count). The peer's copy of the same command is recognised and not re-appended; an unmatched entry is inserted at its own ordinal when free, otherwise appended at MAX(seq)+1 with a Warning — kept, never silent. Bounded read, constant write. m1: MigrateInlineComments copies only — it no longer UPDATEs the condition row (a fresh-HLC re-stamp could beat a newer peer ack under LWW); the next save clears comments_json. m2: it returns unparseable alarm ids and OnReady logs a Warning per id (+ an Information tally). m3/m4: documented in the store remarks (Remove vs lagging peer; in-memory index == durable ordinal). Tests: store-level failover sequence (missed command -> survivor's ack and next entry persisted once, re-save writes 0 rows, one Warning) with the dual-write positive control (peer copy present -> 0 rows, no Warning) and a repeated-identical-comment case; a loopback pair failover test (link down, survivor acks, rejoin -> both entries on both peers, held); migration no-row-rewrite, partial fill and unparseable reporting; OnReady Warning. Both failover tests fail against the round-0 store. Refs #748 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT --- .../AlarmConditionStateSchema.cs | 67 +++--- .../Configuration/LocalDbSetup.cs | 29 ++- .../LocalDbAlarmConditionStateStore.cs | 199 +++++++++++++++--- .../AlarmConditionCommentConvergenceTests.cs | 59 ++++++ .../LocalDbSetupTests.cs | 46 +++- .../LocalDbAlarmCommentAuditTests.cs | 172 +++++++++++++-- 6 files changed, 490 insertions(+), 82 deletions(-) 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 1ec6b043..464cb90d 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian/AlarmConditionStateSchema.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian/AlarmConditionStateSchema.cs @@ -46,7 +46,8 @@ namespace ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian; /// 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; -/// moves any inline trail across. +/// 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 @@ -55,7 +56,9 @@ namespace ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian; /// 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. +/// 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 @@ -100,8 +103,8 @@ public static class AlarmConditionStateSchema // Acknowledged/Unacknowledged, Confirmed/Unconfirmed, Unshelved/OneShotShelved/TimedShelved), // 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 MigrateInlineComments or the next save moves it). The audit user/comment/utc columns - // and shelving_expires_utc are NULL when the alarm has no such history yet. updated_at_utc + // 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 ( @@ -134,9 +137,17 @@ public static class AlarmConditionStateSchema 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); + /// - /// Moves every condition row's inline comments_json trail into - /// and resets the column to "[]". Idempotent. + /// Copies every condition row's inline comments_json trail into + /// . Idempotent; it never writes the condition row itself. /// /// /// @@ -145,18 +156,26 @@ public static class AlarmConditionStateSchema /// 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) moved — neither duplicates nor - /// reorders anything, and a second run finds no non-empty inline column and writes - /// nothing. Each row moves in its own transaction together with its column reset, so a - /// crash leaves every row either fully moved 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) - /// rather than failing host start. + /// 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(). - /// The number of condition rows whose inline trail was moved. - public static int MigrateInlineComments(SqliteConnection connection) + /// What was copied, and which rows could not be read. + public static InlineCommentMigrationResult MigrateInlineComments(SqliteConnection connection) { ArgumentNullException.ThrowIfNull(connection); @@ -172,6 +191,7 @@ public static class AlarmConditionStateSchema } var migrated = 0; + var unparseable = new List(); foreach (var (alarmId, json) in pending) { IReadOnlyList entries; @@ -181,9 +201,11 @@ public static class AlarmConditionStateSchema } catch (JsonException) { + unparseable.Add(alarmId); continue; } + var inserted = 0; using var tx = connection.BeginTransaction(); for (var seq = 0; seq < entries.Count; seq++) { @@ -196,24 +218,15 @@ public static class AlarmConditionStateSchema insert.Parameters.AddWithValue("@User", entries[seq].User); insert.Parameters.AddWithValue("@Kind", entries[seq].Kind); insert.Parameters.AddWithValue("@Text", entries[seq].Text); - insert.ExecuteNonQuery(); - } - - using (var clear = connection.CreateCommand()) - { - clear.Transaction = tx; - clear.CommandText = - "UPDATE alarm_condition_state SET comments_json = @Empty WHERE scripted_alarm_id = @Id"; - clear.Parameters.AddWithValue("@Empty", EmptyInlineComments); - clear.Parameters.AddWithValue("@Id", alarmId); - clear.ExecuteNonQuery(); + inserted += insert.ExecuteNonQuery(); } tx.Commit(); - migrated++; + if (inserted > 0) + migrated++; } - return migrated; + return new InlineCommentMigrationResult(migrated, unparseable); } /// Formats a comment timestamp for (round-trip "O" text). 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 c841db21..d3f89979 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/LocalDbSetup.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/LocalDbSetup.cs @@ -119,12 +119,33 @@ public static class LocalDbSetup RemoveStaleCapture(db, logger ?? Serilog.Log.Logger); // LAST, and only here. These are the only calls in OnReady that write rows. - // Moves any pre-split inline comment trail (alarm_condition_state.comments_json) into the - // append-only alarm_condition_comment table (review R05:4). Idempotent: a boot after the - // first finds no inline trail and writes nothing. + // 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()) { - AlarmConditionStateSchema.MigrateInlineComments(connection); + 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 8437574d..514b7ea9 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ScriptedAlarms/LocalDbAlarmConditionStateStore.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ScriptedAlarms/LocalDbAlarmConditionStateStore.cs @@ -31,13 +31,22 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.ScriptedAlarms; /// have no columns and default to null on load. /// /// -/// The comment trail is written incrementally (review R05:4). Each save appends only -/// the entries past the highest ordinal already stored for the alarm, 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 "[]". The "already stored" boundary -/// is read from the table inside the save's transaction rather than cached, so the store -/// holds no per-alarm state and an entry the pair peer already wrote at an ordinal is kept, -/// not duplicated (see for why the ordinal is the key). +/// 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 @@ -62,6 +71,20 @@ public sealed class LocalDbAlarmConditionStateStore : IAlarmStateStore 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; @@ -128,32 +151,7 @@ public sealed class LocalDbAlarmConditionStateStore : IAlarmStateStore await using var tx = await _db.BeginTransactionAsync(ct).ConfigureAwait(false); - // Append only what lies past the highest ordinal already stored. An indexed MAX over the - // (scripted_alarm_id, seq) primary key — O(log n), and a read, so it costs no replication. - var maxSeq = (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]; - - for (var seq = (int)Math.Max(maxSeq + 1, 0); seq < state.Comments.Count; seq++) - { - var c = state.Comments[seq]; - await tx.ExecuteAsync( - AlarmConditionStateSchema.InsertCommentSql, - new - { - Id = state.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).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 @@ -209,6 +207,141 @@ public sealed class LocalDbAlarmConditionStateStore : IAlarmStateStore _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) { 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 index 26b1fe14..8a6a7e62 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDb/AlarmConditionCommentConvergenceTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDb/AlarmConditionCommentConvergenceTests.cs @@ -70,6 +70,65 @@ public sealed class AlarmConditionCommentConvergenceTests (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() { 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 280ac35a..06351576 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDbSetupTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDbSetupTests.cs @@ -250,8 +250,11 @@ public sealed class LocalDbSetupTests : IDisposable (await ScalarAsync(db, "SELECT COUNT(*) FROM alarm_condition_comment WHERE scripted_alarm_id = 'eq/alarm-legacy'")) .ShouldBe(2); - (await ScalarAsync(db, "SELECT COUNT(*) FROM alarm_condition_state WHERE comments_json <> '[]'")) - .ShouldBe(0); + // 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); @@ -397,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() { 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 index 47438ceb..e6c422e9 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/ScriptedAlarms/LocalDbAlarmCommentAuditTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/ScriptedAlarms/LocalDbAlarmCommentAuditTests.cs @@ -252,7 +252,12 @@ public sealed class LocalDbAlarmCommentAuditTests : IDisposable var rewritten = original with { Comments = original.Comments.SetItem(1, new AlarmComment(T0, "eve", "AddComment", "forged")) }; await store.SaveAsync(rewritten, Ct); - (await store.LoadAsync("alarm-ao", Ct))!.Comments.ShouldBe(original.Comments); + // 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 ----------------------------------------------------------- @@ -306,19 +311,27 @@ public sealed class LocalDbAlarmCommentAuditTests : IDisposable } [Fact] - public async Task Migration_MovesInlineCommentsToTheAuditTable_AndClearsTheInlineColumn() + public async Task Migration_CopiesInlineCommentsToTheAuditTable_AndNeverRewritesTheConditionRow() { await InsertOldShapeRowAsync("alarm-m1"); await InsertOldShapeRowAsync("alarm-m2"); await InsertOldShapeRowAsync("alarm-empty", "[]"); - int migrated; + var before = await OplogHighWaterAsync(); + AlarmConditionStateSchema.InlineCommentMigrationResult result; using (var connection = _db.CreateConnection()) - migrated = AlarmConditionStateSchema.MigrateInlineComments(connection); + result = AlarmConditionStateSchema.MigrateInlineComments(connection); - migrated.ShouldBe(2); + result.RowsMoved.ShouldBe(2); + result.UnparseableAlarmIds.ShouldBeEmpty(); (await ScalarAsync("SELECT COUNT(*) FROM alarm_condition_comment")).ShouldBe(6); - (await ScalarAsync("SELECT COUNT(*) FROM alarm_condition_state WHERE comments_json <> '[]'")).ShouldBe(0); + + // 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"); @@ -330,11 +343,11 @@ public sealed class LocalDbAlarmCommentAuditTests : IDisposable { await InsertOldShapeRowAsync("alarm-idem"); using (var connection = _db.CreateConnection()) - AlarmConditionStateSchema.MigrateInlineComments(connection).ShouldBe(1, "positive control"); + AlarmConditionStateSchema.MigrateInlineComments(connection).RowsMoved.ShouldBe(1, "positive control"); var before = await OplogHighWaterAsync(); using (var connection = _db.CreateConnection()) - AlarmConditionStateSchema.MigrateInlineComments(connection).ShouldBe(0); + 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); @@ -342,24 +355,151 @@ public sealed class LocalDbAlarmCommentAuditTests : IDisposable } [Fact] - public async Task Migration_OverAPartiallyMigratedRow_DoesNotDuplicateOrReorder() + public async Task Migration_OverAPartiallyMigratedRow_FillsOnlyTheMissingOrdinals() { - // An inline row reappearing after its comments were already moved (e.g. a restored backup): - // the ordinals already held are kept, nothing is duplicated. + // 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"); - using (var connection = _db.CreateConnection()) - AlarmConditionStateSchema.MigrateInlineComments(connection); await _db.ExecuteAsync( - "UPDATE alarm_condition_state SET comments_json = @Json WHERE scripted_alarm_id = 'alarm-partial'", - new { Json = OldShapeCommentsJson }, Ct); + 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).ShouldBe(1); + 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() {