Merge lane E5: alarm comment audit table (#748)

Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
This commit is contained in:
Joseph Doherty
2026-10-22 17:06:37 -04:00
9 changed files with 1360 additions and 123 deletions
@@ -1,11 +1,13 @@
using System.Globalization;
using System.Text.Json;
using Microsoft.Data.Sqlite;
namespace ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian;
/// <summary>
/// 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.
/// </summary>
/// <remarks>
/// <para>
@@ -34,14 +36,58 @@ namespace ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian;
/// the last transition. Timestamps are round-trip ("O") TEXT so lexicographic order is
/// chronological; nullable timestamps and audit fields are stored as NULL.
/// </para>
/// <para>
/// <b>The comment trail is NOT on the condition row.</b> It used to be serialized whole into
/// <c>comments_json</c>, 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
/// <see cref="CommentsTable"/>, one row per entry keyed by <c>(scripted_alarm_id, seq)</c>,
/// where <c>seq</c> 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. <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"/> 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
/// 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 <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. 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
{
/// <summary>Table holding one condition-state row per scripted-alarm identity.</summary>
public const string StateTable = "alarm_condition_state";
/// <summary>Append-only table holding one row per comment-trail entry, keyed by alarm id + ordinal.</summary>
public const string CommentsTable = "alarm_condition_comment";
/// <summary>What <c>comments_json</c> holds once the trail lives in <see cref="CommentsTable"/>.</summary>
public const string EmptyInlineComments = "[]";
/// <summary>
/// Creates the state table if it does not already exist. Idempotent.
/// The one append statement for <see cref="CommentsTable"/>, shared by the migration and the
/// store. <c>OR IGNORE</c> is what makes the trail append-only: an ordinal already held is
/// never rewritten. Parameters: <c>@Id @Seq @Ts @User @Kind @Text</c>.
/// </summary>
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);
/// <summary>
/// Creates the state table and the comment table if they do not already exist. Idempotent.
/// </summary>
/// <param name="connection">
/// An already-open connection. <c>ILocalDb.CreateConnection()</c> hands out open,
@@ -55,10 +101,11 @@ public static class AlarmConditionStateSchema
// The four *_state columns are always written to a mapped, non-null string (Enabled/Disabled,
// Acknowledged/Unacknowledged, Confirmed/Unconfirmed, Unshelved/OneShotShelved/TimedShelved),
// so they are NOT NULL. comments_json is never null either — an empty trail is the literal
// "[]". The audit user/comment/utc columns and shelving_expires_utc are NULL when the alarm
// has no such history yet. updated_at_utc carries LastTransitionUtc (there is no dedicated
// transition column).
// so they are NOT NULL. comments_json is never null either — it is the literal "[]" since the
// trail moved to alarm_condition_comment (an old-shape row may still carry an inline trail
// until the alarm's next save; MigrateInlineComments copies it without touching this row).
// The audit user/comment/utc columns and shelving_expires_utc are NULL when the alarm has no such history yet. updated_at_utc
// carries LastTransitionUtc (there is no dedicated transition column).
cmd.CommandText = """
CREATE TABLE IF NOT EXISTS alarm_condition_state (
scripted_alarm_id TEXT NOT NULL PRIMARY KEY,
@@ -76,7 +123,161 @@ public static class AlarmConditionStateSchema
comments_json TEXT NOT NULL,
updated_at_utc TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS alarm_condition_comment (
scripted_alarm_id TEXT NOT NULL,
seq INTEGER NOT NULL,
timestamp_utc TEXT NOT NULL,
user_name TEXT NOT NULL,
kind TEXT NOT NULL,
text TEXT NOT NULL,
PRIMARY KEY (scripted_alarm_id, seq)
);
""";
cmd.ExecuteNonQuery();
}
/// <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>
/// 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>
/// <b>Run it AFTER the tables are registered for replication</b> — it writes rows, and a
/// row written before the capture triggers exist never reaches the peer (the same
/// 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) 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>What was copied, and which rows could not be read.</returns>
public static InlineCommentMigrationResult MigrateInlineComments(SqliteConnection connection)
{
ArgumentNullException.ThrowIfNull(connection);
var pending = new List<(string AlarmId, string Json)>();
using (var select = connection.CreateCommand())
{
select.CommandText =
"SELECT scripted_alarm_id, comments_json FROM alarm_condition_state WHERE comments_json <> @Empty";
select.Parameters.AddWithValue("@Empty", EmptyInlineComments);
using var reader = select.ExecuteReader();
while (reader.Read())
pending.Add((reader.GetString(0), reader.GetString(1)));
}
var migrated = 0;
var unparseable = new List<string>();
foreach (var (alarmId, json) in pending)
{
IReadOnlyList<InlineComment> entries;
try
{
entries = ParseInlineComments(json);
}
catch (JsonException)
{
unparseable.Add(alarmId);
continue;
}
var inserted = 0;
using var tx = connection.BeginTransaction();
for (var seq = 0; seq < entries.Count; seq++)
{
using var insert = connection.CreateCommand();
insert.Transaction = tx;
insert.CommandText = InsertCommentSql;
insert.Parameters.AddWithValue("@Id", alarmId);
insert.Parameters.AddWithValue("@Seq", seq);
insert.Parameters.AddWithValue("@Ts", FormatTimestamp(entries[seq].TimestampUtc));
insert.Parameters.AddWithValue("@User", entries[seq].User);
insert.Parameters.AddWithValue("@Kind", entries[seq].Kind);
insert.Parameters.AddWithValue("@Text", entries[seq].Text);
inserted += insert.ExecuteNonQuery();
}
tx.Commit();
if (inserted > 0)
migrated++;
}
return new InlineCommentMigrationResult(migrated, unparseable);
}
/// <summary>Formats a comment timestamp for <see cref="CommentsTable"/> (round-trip "O" text).</summary>
/// <param name="value">The timestamp.</param>
/// <returns>The round-trip text form.</returns>
public static string FormatTimestamp(DateTime value) => value.ToString("O", CultureInfo.InvariantCulture);
/// <summary>Parses a <see cref="CommentsTable"/> timestamp, preserving its <see cref="DateTimeKind"/>.</summary>
/// <param name="value">The round-trip text form.</param>
/// <returns>The timestamp.</returns>
public static DateTime ParseTimestamp(string value) =>
DateTime.Parse(value, CultureInfo.InvariantCulture, DateTimeStyles.RoundtripKind);
/// <summary>
/// Parses a legacy inline <c>comments_json</c> trail exactly as the pre-split store did
/// (<see cref="JsonSerializerDefaults.Web"/>; null strings become empty).
/// </summary>
/// <param name="json">The column value.</param>
/// <returns>The entries in trail order; empty for a blank or <c>"[]"</c> column.</returns>
/// <exception cref="JsonException">The column is not a valid trail.</exception>
public static IReadOnlyList<InlineComment> ParseInlineComments(string? json)
{
if (string.IsNullOrWhiteSpace(json)) return [];
var dtos = JsonSerializer.Deserialize<List<InlineCommentDto>>(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();
}
/// <summary>One comment-trail entry as stored — the persistence-side mirror of the engine's <c>AlarmComment</c>.</summary>
/// <param name="TimestampUtc">When the action happened.</param>
/// <param name="User">Identity of the actor.</param>
/// <param name="Kind">Classification (Acknowledge, Confirm, AddComment, …).</param>
/// <param name="Text">Operator-supplied or engine-generated text.</param>
public sealed record InlineComment(DateTime TimestampUtc, string User, string Kind, string Text);
/// <summary>The pre-split on-disk shape of one entry in <c>comments_json</c>.</summary>
private sealed class InlineCommentDto
{
/// <summary>When the comment was recorded.</summary>
public DateTime TimestampUtc { get; set; }
/// <summary>Identity of the actor.</summary>
public string? User { get; set; }
/// <summary>Classification of the entry.</summary>
public string? Kind { get; set; }
/// <summary>Entry text.</summary>
public string? Text { get; set; }
}
}
@@ -7,7 +7,7 @@ namespace ZB.MOM.WW.OtOpcUa.Host.Configuration;
/// <summary>
/// The <c>onReady</c> callback handed to <c>AddZbLocalDb</c>: 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.
/// </summary>
/// <remarks>
@@ -21,7 +21,7 @@ namespace ZB.MOM.WW.OtOpcUa.Host.Configuration;
public static class LocalDbSetup
{
/// <summary>
/// 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.
/// </summary>
/// <remarks>
@@ -36,6 +36,7 @@ public static class LocalDbSetup
DeploymentCacheSchema.PointerTable,
AlarmSfSchema.EventsTable,
AlarmConditionStateSchema.StateTable,
AlarmConditionStateSchema.CommentsTable,
];
/// <summary>
@@ -47,8 +48,9 @@ public static class LocalDbSetup
/// <c>RegisterReplicated</c> 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. <see cref="AlarmSfLegacyMigrator"/> therefore runs <b>last</b>, after every
/// registration — it writes rows, and they must be captured like any other write.
/// history. <see cref="AlarmConditionStateSchema.MigrateInlineComments"/> and
/// <see cref="AlarmSfLegacyMigrator"/> therefore run <b>last</b>, after every
/// registration — they write rows, and those must be captured like any other write.
/// </para>
/// <para>
/// <b>Registration is CONDITIONAL on <see cref="ReplicationIsConfigured"/>.</b> It used
@@ -68,7 +70,7 @@ public static class LocalDbSetup
/// </para>
/// <para>
/// <b>A failure part-way through either loop fails host start, and that is deliberate.</b>
/// 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 <c>AddZbLocalDb</c> singleton factory — a node that came up
/// anyway would be capturing (or not) on an arbitrary subset, which is silently partial
/// replication with nothing in the log to say so. Both library calls are idempotent, so
@@ -116,7 +118,36 @@ public static class LocalDbSetup
else
RemoveStaleCapture(db, logger ?? Serilog.Log.Logger);
// LAST, and only here. This is the one call in OnReady that writes rows.
// LAST, and only here. These are the only calls in OnReady that write rows.
// Copies any pre-split inline comment trail (alarm_condition_state.comments_json) into the
// append-only alarm_condition_comment table (review R05:4). Idempotent: an ordinal already
// held is never rewritten, and the condition row itself is never touched (the alarm's next
// save clears the column). An unreadable trail is reported, not skipped silently.
AlarmConditionStateSchema.InlineCommentMigrationResult commentMigration;
using (var connection = db.CreateConnection())
{
commentMigration = AlarmConditionStateSchema.MigrateInlineComments(connection);
}
var migrationLog = logger ?? Serilog.Log.Logger;
if (commentMigration.RowsMoved > 0)
{
migrationLog.Information(
"LocalDb: copied the inline comment trail of {AlarmCount} scripted alarm(s) into {Table}.",
commentMigration.RowsMoved,
AlarmConditionStateSchema.CommentsTable);
}
foreach (var alarmId in commentMigration.UnparseableAlarmIds)
{
migrationLog.Warning(
"LocalDb: scripted alarm {AlarmId} has an unreadable inline comment trail "
+ "(alarm_condition_state.comments_json is not valid JSON); its history was NOT moved to {Table} "
+ "and the alarm's state will not load until the row is repaired.",
alarmId,
AlarmConditionStateSchema.CommentsTable);
}
AlarmSfLegacyMigrator.Migrate(db, configuration);
}
@@ -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;
/// <summary>
/// Node-local <see cref="IAlarmStateStore"/> backed by the replicated
/// <see cref="AlarmConditionStateSchema.StateTable"/> table in the node's consolidated LocalDb.
/// Maps the full Part 9 <see cref="AlarmConditionState"/> — Enabled / Acked / Confirmed /
/// Shelving + the ack/confirm audit trail + operator comments.
/// <see cref="AlarmConditionStateSchema.StateTable"/> and
/// <see cref="AlarmConditionStateSchema.CommentsTable"/> tables in the node's consolidated
/// LocalDb. Maps the full Part 9 <see cref="AlarmConditionState"/> — Enabled / Acked / Confirmed /
/// Shelving + the ack/confirm audit fields on the condition row, and the append-only comment
/// trail one row per entry.
/// </summary>
/// <remarks>
/// <para>
@@ -24,44 +24,75 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.ScriptedAlarms;
/// state then mirrors to its redundant pair peer instead of depending on central SQL.
/// </para>
/// <para>
/// <b>The round-trip is byte-identical to the EF store</b> — 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:
/// </para>
/// <para>
/// <b>ActiveState is NOT persisted</b> — there is no column; on <see cref="LoadAsync"/> it is
/// restored as <see cref="AlarmActiveState.Inactive"/> and the engine re-derives it from the
/// live predicate. <b>LastTransitionUtc ↔ updated_at_utc</b>: no dedicated transition column,
/// so the last transition rides the row-write timestamp. <b>LastActiveUtc / LastClearedUtc</b>
/// have no columns and default to <c>null</c> on load. <see cref="AlarmConditionState.Comments"/>
/// serializes to/from <c>comments_json</c>; an empty list round-trips as <c>"[]"</c>.
/// have no columns and default to <c>null</c> on load.
/// </para>
/// <para>
/// <b>Save is a single unconditional PK upsert.</b> 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).
/// <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
/// <c>comments_json</c> trail (written by the pre-split store and not yet moved by
/// <see cref="AlarmConditionStateSchema.MigrateInlineComments"/>) 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.
/// </para>
/// <para>
/// <b>Save is one transaction:</b> 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.
/// </para>
/// </remarks>
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";
/// <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;
/// <summary>Initializes a new instance of the <see cref="LocalDbAlarmConditionStateStore"/> class.</summary>
/// <param name="db">
/// The node's local database. Its <see cref="AlarmConditionStateSchema.StateTable"/> table
/// must already exist and be registered for replication — the host does both in
/// <c>LocalDbSetup.OnReady</c>, 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 <c>LocalDbSetup.OnReady</c>, before
/// any consumer can resolve this store.
/// </param>
/// <param name="logger">The logger instance.</param>
public LocalDbAlarmConditionStateStore(ILocalDb db, ILogger<LocalDbAlarmConditionStateStore> logger)
@@ -79,7 +110,15 @@ public sealed class LocalDbAlarmConditionStateStore : IAlarmStateStore
MapRow,
new { Id = alarmId },
ct).ConfigureAwait(false);
return rows.Count > 0 ? rows[0] : null;
if (rows.Count == 0) return null;
var comments = await _db.QueryAsync(
$"SELECT {SelectCommentColumns} FROM alarm_condition_comment WHERE scripted_alarm_id = @Id ORDER BY seq",
MapComment,
new { Id = alarmId },
ct).ConfigureAwait(false);
return WithComments(rows[0], comments);
}
/// <inheritdoc />
@@ -90,7 +129,19 @@ public sealed class LocalDbAlarmConditionStateStore : IAlarmStateStore
MapRow,
parameters: null,
ct).ConfigureAwait(false);
return rows;
var comments = await _db.QueryAsync(
$"SELECT {SelectCommentColumns} FROM alarm_condition_comment ORDER BY scripted_alarm_id, seq",
MapComment,
parameters: null,
ct).ConfigureAwait(false);
var byAlarm = comments
.GroupBy(c => c.AlarmId, StringComparer.Ordinal)
.ToDictionary(g => g.Key, g => (IReadOnlyList<StoredComment>)g.ToList(), StringComparer.Ordinal);
return rows
.Select(r => WithComments(r, byAlarm.TryGetValue(r.State.AlarmId, out var c) ? c : []))
.ToList();
}
/// <inheritdoc />
@@ -98,10 +149,15 @@ public sealed class LocalDbAlarmConditionStateStore : IAlarmStateStore
{
ArgumentNullException.ThrowIfNull(state);
await using var tx = await _db.BeginTransactionAsync(ct).ConfigureAwait(false);
await AppendNewCommentsAsync(tx, state, ct).ConfigureAwait(false);
// A single unconditional upsert keyed on the alarm identity. Convergence across the pair is
// last-writer-wins over the primary key, handled by the LocalDb replication HLC — no
// app-level LWW here (see the class remarks).
await _db.ExecuteAsync(
// app-level LWW here (see the class remarks). comments_json is always "[]": the trail lives
// in alarm_condition_comment, which is what keeps this row a fixed size.
await tx.ExecuteAsync(
"""
INSERT INTO alarm_condition_state
(scripted_alarm_id, enabled_state, acked_state, confirmed_state, shelving_state,
@@ -140,60 +196,243 @@ public sealed class LocalDbAlarmConditionStateStore : IAlarmStateStore
LastConfirmUser = state.LastConfirmUser,
LastConfirmComment = state.LastConfirmComment,
LastConfirmUtc = FormatNullable(state.LastConfirmUtc),
CommentsJson = SerializeComments(state.Comments),
CommentsJson = AlarmConditionStateSchema.EmptyInlineComments,
// No dedicated transition column — persist LastTransitionUtc into updated_at_utc.
UpdatedAtUtc = Format(state.LastTransitionUtc),
},
ct).ConfigureAwait(false);
await tx.CommitAsync(ct).ConfigureAwait(false);
_logger.LogTrace("Persisted alarm-condition state for {AlarmId}", state.AlarmId);
}
/// <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)
{
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(
/// <summary>A condition row as read, before its trail is attached.</summary>
private sealed record ConditionRow(AlarmConditionState State, string InlineCommentsJson);
/// <summary>One <c>alarm_condition_comment</c> row.</summary>
private sealed record StoredComment(string AlarmId, long Seq, AlarmComment Comment);
/// <summary>
/// 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.
/// </summary>
private static AlarmConditionState WithComments(ConditionRow row, IReadOnlyList<StoredComment> 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<long, AlarmComment>();
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<AlarmComment>.Empty),
InlineCommentsJson: r.GetString(12));
// Timestamps are stored as round-trip ("O") TEXT so lexicographic order stays chronological and
// the DateTimeKind.Utc round-trips exactly.
private static string Format(DateTime value) => value.ToString("O", CultureInfo.InvariantCulture);
private static string Format(DateTime value) => AlarmConditionStateSchema.FormatTimestamp(value);
private static string? FormatNullable(DateTime? value) => value?.ToString("O", CultureInfo.InvariantCulture);
private static string? FormatNullable(DateTime? value) => value is { } v ? Format(v) : null;
private static DateTime Parse(string value) =>
DateTime.Parse(value, CultureInfo.InvariantCulture, DateTimeStyles.RoundtripKind);
private static DateTime Parse(string value) => AlarmConditionStateSchema.ParseTimestamp(value);
private static DateTime? ParseNullable(SqliteDataReader r, int ordinal) =>
r.IsDBNull(ordinal) ? null : Parse(r.GetString(ordinal));
@@ -220,45 +459,4 @@ public sealed class LocalDbAlarmConditionStateStore : IAlarmStateStore
"TimedShelved" => ShelvingKind.Timed,
_ => ShelvingKind.Unshelved, // unknown string → Unshelved (safe default)
};
private static string SerializeComments(ImmutableList<AlarmComment> 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<AlarmComment> DeserializeComments(string? json)
{
if (string.IsNullOrWhiteSpace(json)) return ImmutableList<AlarmComment>.Empty;
var dtos = JsonSerializer.Deserialize<List<CommentDto>>(json, JsonOptions);
if (dtos is null || dtos.Count == 0) return ImmutableList<AlarmComment>.Empty;
return dtos
.Select(d => new AlarmComment(d.TimestampUtc, d.User ?? string.Empty, d.Kind ?? string.Empty, d.Text ?? string.Empty))
.ToImmutableList();
}
/// <summary>Stable on-disk shape for a persisted <see cref="AlarmComment"/> in <c>comments_json</c>.</summary>
private sealed class CommentDto
{
/// <summary>When the comment was recorded (UTC).</summary>
public DateTime TimestampUtc { get; set; }
/// <summary>Identity of the actor that wrote the comment.</summary>
public string? User { get; set; }
/// <summary>Human-readable classification of the comment (Acknowledge, Confirm, …).</summary>
public string? Kind { get; set; }
/// <summary>Operator-supplied or engine-generated comment text.</summary>
public string? Text { get; set; }
}
}