fix(alarms): move scripted-alarm comment trail to its own append-only table (R05:4)

Every Part 9 transition used to re-serialize the alarm's whole comment trail into alarm_condition_state.comments_json and replicate it to the pair peer: O(n) per transition, O(n^2) over an alarm's life (measured: one ack at 10k comments shipped 1,080,553 oplog bytes; now ~1.6 KB, the same as at 10). - New replicated table alarm_condition_comment, PK (scripted_alarm_id, seq), seq = the entry's ordinal in the trail. Ordinal (not content) keying makes the two pair nodes' copies of one logical entry (each stamped by its own clock, since both nodes apply every command) collapse to one row under LWW. - LocalDbAlarmConditionStateStore.SaveAsync appends only entries past the stored MAX(seq) (INSERT OR IGNORE, append-only) and upserts a fixed-size condition row with comments_json = "[]", in one transaction. Remove drops the trail with the row. Load merges the table with any inline trail. - comments_json column kept (always "[]") so the condition table's shape and replication digest are unchanged and old-shape rows still load exactly. - AlarmConditionStateSchema.MigrateInlineComments: idempotent per-row move of inline trails, run by LocalDbSetup.OnReady after RegisterReplicated so the moved entries replicate. Comment table joins ReplicatedTables (now five). Tests: constant transition payload at 10 vs 10k comments (+ instrument positive control), old-shape back-compat load, migration idempotency / partial re-run / lazy move, OnReady boot migration captured in the oplog, and two loopback pair tests (A->B and B->A trail convergence; the shared entry converges to one copy, with a distinct-alarm positive control). Refs #748 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
2026-10-22 16:36:33 -04:00
parent 7feffc963a
commit 846c8d6335
9 changed files with 952 additions and 123 deletions
@@ -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,15 @@ public static class LocalDbSetup
        else
            RemoveStaleCapture(db, logger ?? Serilog.Log.Logger);

-        // LAST, and only here. This is the one call in OnReady that writes rows.
+        // LAST, and only here. These are the only calls in OnReady that write rows.
+        // Moves any pre-split inline comment trail (alarm_condition_state.comments_json) into the
+        // append-only alarm_condition_comment table (review R05:4). Idempotent: a boot after the
+        // first finds no inline trail and writes nothing.
+        using (var connection = db.CreateConnection())
+        {
+            AlarmConditionStateSchema.MigrateInlineComments(connection);
+        }
+
        AlarmSfLegacyMigrator.Migrate(db, configuration);
    }

@@ -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,52 @@ 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 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).
+///     </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";
+
    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 +87,15 @@ public sealed class LocalDbAlarmConditionStateStore : IAlarmStateStore
            MapRow,
            new { Id = alarmId },
            ct).ConfigureAwait(false);
-        return rows.Count > 0 ? rows[0] : null;
+        if (rows.Count == 0) return null;
+
+        var comments = await _db.QueryAsync(
+            $"SELECT {SelectCommentColumns} FROM alarm_condition_comment WHERE scripted_alarm_id = @Id ORDER BY seq",
+            MapComment,
+            new { Id = alarmId },
+            ct).ConfigureAwait(false);
+
+        return WithComments(rows[0], comments);
    }

    /// <inheritdoc />
@@ -90,7 +106,19 @@ public sealed class LocalDbAlarmConditionStateStore : IAlarmStateStore
            MapRow,
            parameters: null,
            ct).ConfigureAwait(false);
-        return rows;
+
+        var comments = await _db.QueryAsync(
+            $"SELECT {SelectCommentColumns} FROM alarm_condition_comment ORDER BY scripted_alarm_id, seq",
+            MapComment,
+            parameters: null,
+            ct).ConfigureAwait(false);
+        var byAlarm = comments
+            .GroupBy(c => c.AlarmId, StringComparer.Ordinal)
+            .ToDictionary(g => g.Key, g => (IReadOnlyList<StoredComment>)g.ToList(), StringComparer.Ordinal);
+
+        return rows
+            .Select(r => WithComments(r, byAlarm.TryGetValue(r.State.AlarmId, out var c) ? c : []))
+            .ToList();
    }

    /// <inheritdoc />
@@ -98,10 +126,40 @@ public sealed class LocalDbAlarmConditionStateStore : IAlarmStateStore
    {
        ArgumentNullException.ThrowIfNull(state);

+        await using var tx = await _db.BeginTransactionAsync(ct).ConfigureAwait(false);
+
+        // Append only what lies past the highest ordinal already stored. An indexed MAX over the
+        // (scripted_alarm_id, seq) primary key — O(log n), and a read, so it costs no replication.
+        var maxSeq = (await tx.QueryAsync(
+            "SELECT COALESCE(MAX(seq), -1) FROM alarm_condition_comment WHERE scripted_alarm_id = @Id",
+            static r => r.GetInt64(0),
+            new { Id = state.AlarmId },
+            ct).ConfigureAwait(false))[0];
+
+        for (var seq = (int)Math.Max(maxSeq + 1, 0); seq < state.Comments.Count; seq++)
+        {
+            var c = state.Comments[seq];
+            await tx.ExecuteAsync(
+                AlarmConditionStateSchema.InsertCommentSql,
+                new
+                {
+                    Id = state.AlarmId,
+                    Seq = seq,
+                    // AlarmComment.TimestampUtc must be DateTimeKind.Utc for a Utc round-trip; the
+                    // engine always creates comments with Utc kind.
+                    Ts = AlarmConditionStateSchema.FormatTimestamp(c.TimestampUtc),
+                    User = c.User,
+                    Kind = c.Kind,
+                    Text = c.Text,
+                },
+                ct).ConfigureAwait(false);
+        }
+
        // A single unconditional upsert keyed on the alarm identity. Convergence across the pair is
        // last-writer-wins over the primary key, handled by the LocalDb replication HLC — no
-        // app-level LWW here (see the class remarks).
-        await _db.ExecuteAsync(
+        // app-level LWW here (see the class remarks). comments_json is always "[]": the trail lives
+        // in alarm_condition_comment, which is what keeps this row a fixed size.
+        await tx.ExecuteAsync(
            """
            INSERT INTO alarm_condition_state
                (scripted_alarm_id, enabled_state, acked_state, confirmed_state, shelving_state,
@@ -140,12 +198,14 @@ public sealed class LocalDbAlarmConditionStateStore : IAlarmStateStore
                LastConfirmUser = state.LastConfirmUser,
                LastConfirmComment = state.LastConfirmComment,
                LastConfirmUtc = FormatNullable(state.LastConfirmUtc),
-                CommentsJson = SerializeComments(state.Comments),
+                CommentsJson = AlarmConditionStateSchema.EmptyInlineComments,
                // No dedicated transition column — persist LastTransitionUtc into updated_at_utc.
                UpdatedAtUtc = Format(state.LastTransitionUtc),
            },
            ct).ConfigureAwait(false);

+        await tx.CommitAsync(ct).ConfigureAwait(false);
+
        _logger.LogTrace("Persisted alarm-condition state for {AlarmId}", state.AlarmId);
    }

@@ -153,47 +213,93 @@ public sealed class LocalDbAlarmConditionStateStore : IAlarmStateStore
    public async Task RemoveAsync(string alarmId, CancellationToken ct)
    {
        ArgumentNullException.ThrowIfNull(alarmId);
-        await _db.ExecuteAsync(
+
+        // The trail goes with the condition, as it did when it was a column of the row.
+        await using var tx = await _db.BeginTransactionAsync(ct).ConfigureAwait(false);
+        await tx.ExecuteAsync(
+            "DELETE FROM alarm_condition_comment WHERE scripted_alarm_id = @Id",
+            new { Id = alarmId },
+            ct).ConfigureAwait(false);
+        await tx.ExecuteAsync(
            "DELETE FROM alarm_condition_state WHERE scripted_alarm_id = @Id",
            new { Id = alarmId },
            ct).ConfigureAwait(false);
+        await tx.CommitAsync(ct).ConfigureAwait(false);
    }

-    private static AlarmConditionState MapRow(SqliteDataReader r) => new(
+    /// <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 +326,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; }
-    }
 }