fix(alarms): never drop a comment-trail entry on pair divergence (E5 review r1)

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
This commit is contained in:
Joseph Doherty
2026-10-22 16:58:46 -04:00
parent 13f1357ad2
commit dad28d37a2
6 changed files with 490 additions and 82 deletions
@@ -46,7 +46,8 @@ namespace ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian;
/// O(1) however long the history is. <c>comments_json</c> is kept, always written as
/// <c>"[]"</c>, 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;
/// <see cref="MigrateInlineComments"/> moves any inline trail across.
/// <see cref="MigrateInlineComments"/> copies any inline trail across (the alarm's next
/// save clears the column).
/// </para>
/// <para>
/// <b>Why the ordinal, and not a content hash, is the key.</b> 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 <i>n</i> 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 <c>INSERT OR IGNORE</c> (<see cref="InsertCommentSql"/>), 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).
/// </para>
/// </remarks>
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();
}
/// <summary>What one <see cref="MigrateInlineComments"/> run did.</summary>
/// <param name="RowsMoved">Condition rows for which at least one inline entry was copied across.</param>
/// <param name="UnparseableAlarmIds">
/// 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.
/// </param>
public sealed record InlineCommentMigrationResult(int RowsMoved, IReadOnlyList<string> UnparseableAlarmIds);
/// <summary>
/// Moves every condition row's inline <c>comments_json</c> trail into
/// <see cref="CommentsTable"/> and resets the column to <c>"[]"</c>. Idempotent.
/// Copies every condition row's inline <c>comments_json</c> trail into
/// <see cref="CommentsTable"/>. Idempotent; it never writes the condition row itself.
/// </summary>
/// <remarks>
/// <para>
@@ -145,18 +156,26 @@ public static class AlarmConditionStateSchema
/// DDL → register → writes rule <c>AlarmSfLegacyMigrator</c> follows).
/// </para>
/// <para>
/// <b>Copy, don't clear.</b> Resetting <c>comments_json</c> 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 <c>"[]"</c> with the alarm's real state.
/// </para>
/// <para>
/// Each entry lands at its ordinal in the inline array via <see cref="InsertCommentSql"/>,
/// 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 <see cref="InlineCommentMigrationResult.UnparseableAlarmIds"/> rather
/// than failing host start.
/// </para>
/// </remarks>
/// <param name="connection">An already-open connection from <c>ILocalDb.CreateConnection()</c>.</param>
/// <returns>The number of condition rows whose inline trail was moved.</returns>
public static int MigrateInlineComments(SqliteConnection connection)
/// <returns>What was copied, and which rows could not be read.</returns>
public static InlineCommentMigrationResult MigrateInlineComments(SqliteConnection connection)
{
ArgumentNullException.ThrowIfNull(connection);
@@ -172,6 +191,7 @@ public static class AlarmConditionStateSchema
}
var migrated = 0;
var unparseable = new List<string>();
foreach (var (alarmId, json) in pending)
{
IReadOnlyList<InlineComment> 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);
}
/// <summary>Formats a comment timestamp for <see cref="CommentsTable"/> (round-trip "O" text).</summary>
@@ -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);
@@ -31,13 +31,22 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.ScriptedAlarms;
/// have no columns and default to <c>null</c> on load.
/// </para>
/// <para>
/// <b>The comment trail is written incrementally (review R05:4).</b> 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 <c>comments_json</c> is always written as <c>"[]"</c>. 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 <see cref="AlarmConditionStateSchema"/> for why the ordinal is the key).
/// <b>The comment trail is written incrementally (review R05:4).</b> 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
/// <c>comments_json</c> is always written as <c>"[]"</c>. 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
/// <c>AppendNewCommentsAsync</c> and, for why the ordinal is the key,
/// <see cref="AlarmConditionStateSchema"/>.
/// </para>
/// <para>
/// <b>Remove vs. a lagging peer.</b> <see cref="RemoveAsync"/> 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).
/// </para>
/// <para>
/// <b>Old-shape rows load exactly as before.</b> 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";
/// <summary>
/// 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.
/// </summary>
private const int MatchWindow = 16;
/// <summary>
/// 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).
/// </summary>
private static readonly TimeSpan SameCommandSkew = TimeSpan.FromMinutes(5);
private readonly ILocalDb _db;
private readonly ILogger<LocalDbAlarmConditionStateStore> _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);
}
/// <summary>
/// Persists the trail entries <paramref name="state"/> holds that the table does not, writing
/// a bounded number of rows however long the trail is.
/// </summary>
/// <remarks>
/// <para>
/// <b>Normal case.</b> The in-memory trail and the stored trail agree position by
/// position; the only new entries are past <c>MAX(seq)</c> and land at their own ordinal.
/// </para>
/// <para>
/// <b>Divergence (review I-1).</b> 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
/// <c>MAX(seq)</c>. 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 <see cref="MatchWindow"/> entries) is matched against the stored rows from
/// that position on, by content: same Kind, User and Text, timestamps within
/// <see cref="SameCommandSkew"/> (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 (<c>MAX(seq)+1</c>) with a Warning — kept, never
/// dropped, and never silent.
/// </para>
/// <para>
/// <b>Why this cannot double-count.</b> 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).
/// </para>
/// <para>
/// ⚠️ <b>The in-memory index is the durable ordinal.</b> Trimming
/// <see cref="AlarmConditionState.Comments"/> 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.
/// </para>
/// </remarks>
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<int> 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);
/// <inheritdoc />
public async Task RemoveAsync(string alarmId, CancellationToken ct)
{