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()
{