feat(localdb): alarm_sf_events replicated table
Adds the alarm store-and-forward buffer to the consolidated LocalDb file and registers it for replication, so a node that dies with undelivered alarm history no longer takes it to the grave. The table keeps the legacy queue's shape rather than the status column the plan sketched. An acknowledged row is deleted, not marked delivered: that needs no second sweeper to keep the table bounded, leaves the capacity semantics untouched, and the replication engine carries the delete as a tombstone so the peer drops its copy anyway -- which is what the status column was for. last_error is retained because it is the only operator-facing record of why a row was dead-lettered. The primary key is app-minted TEXT. The legacy AUTOINCREMENT RowId cannot replicate under last-writer-wins: two nodes would independently allocate rowid 7 to different alarms and silently overwrite each other. The drain index therefore orders by enqueued_at_utc rather than insertion order, with id as a tiebreak so the ordering is total. Tables are created unconditionally, independent of AlarmHistorian:Enabled. An empty registered table costs three triggers; creating it lazily would mean a node that enables the historian later writes rows before its capture triggers exist, which is exactly the silent-loss shape OnReady's ordering comment warns about. Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW
This commit is contained in:
Joseph Doherty
@@ -10,10 +10,11 @@
|
||||
{
|
||||
"id": 1,
|
||||
"subject": "Task 1: alarm_sf_events schema + registration (+ exact-set pin update)",
|
||||
"status": "pending",
|
||||
"status": "completed",
|
||||
"blockedBy": [
|
||||
0
|
||||
]
|
||||
],
|
||||
"note": "alarm_sf_events created in AlarmSfSchema (Core.AlarmHistorian) + registered third in LocalDbSetup.OnReady. Exact-set pin updated to 3 tables. DEVIATION D-5: kept the legacy delete-on-ack + dead_lettered flag + last_error rather than the plan's status column (no sweeper for 'delivered' rows; last_error is the only record of why a row died). Drain ORDER BY moves to (enqueued_at_utc, id)."
|
||||
},
|
||||
{
|
||||
"id": 2,
|
||||
|
||||
@@ -242,6 +242,27 @@ the rewritten sink, so the sink cannot compile in its final shape without it, an
|
||||
commit with a rewired-but-ungated sink would be a commit in which a replicated table is drained by
|
||||
both nodes. Not a state worth having in history.
|
||||
|
||||
### D-5 — The table keeps the legacy delete-on-ack shape, not a `status` column
|
||||
|
||||
**The plan proposes** `status TEXT NOT NULL DEFAULT 'pending'` with values `pending | delivered |
|
||||
dead`, and drops the legacy `LastError` column.
|
||||
|
||||
**Two problems.** A `delivered` status means acknowledged rows accumulate in the table forever —
|
||||
the plan specifies no sweeper for them, and they would also have to be excluded from the capacity
|
||||
count, changing a semantic §3 says to preserve. And `LastError` is not dead weight: `DeadLetterRow`
|
||||
writes the failure reason into it and `RetryDeadLettered` clears it, so it is the only operator-
|
||||
facing record of *why* a row was dead-lettered.
|
||||
|
||||
**The decision.** Keep the legacy shape: DELETE on acknowledgement, `dead_lettered` as the same 0/1
|
||||
flag, and retain `last_error`. Deleting keeps the table bounded with no second sweeper, and the
|
||||
replication engine carries the delete as a tombstone (pruned after `TombstoneRetention`, 7 d), so
|
||||
the peer drops its copy too — which is the property the plan wanted the `delivered` status for.
|
||||
Columns then map 1:1 onto the legacy table, making the Task-4 migrator a straight copy.
|
||||
|
||||
One genuine change: the drain's `ORDER BY` moves from `RowId ASC` to `enqueued_at_utc, id`, because
|
||||
a hashed TEXT id carries no insertion order. Round-trip ("O") timestamps sort lexicographically in
|
||||
chronological order, and `id` makes the ordering total; the index covers both.
|
||||
|
||||
### D-4 — The live gate needs `MaxAttempts` raised on the rig
|
||||
|
||||
Not actionable until Task 8, recorded now so it is not rediscovered there. The rig has no
|
||||
|
||||
@@ -0,0 +1,77 @@
|
||||
using Microsoft.Data.Sqlite;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian;
|
||||
|
||||
/// <summary>
|
||||
/// DDL for the alarm store-and-forward buffer: one row per alarm event awaiting delivery to
|
||||
/// the historian gateway.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// Replaces the standalone <c>alarm-historian.db</c> the sink used to own outright. Living
|
||||
/// in the consolidated LocalDb file is what lets the buffer replicate to the redundant pair
|
||||
/// peer, so a node that dies with undelivered alarm history no longer takes it to the grave.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// Deliberately depends on nothing but <see cref="SqliteConnection"/> so it can be applied
|
||||
/// to any connection — the host's <c>LocalDbSetup.OnReady</c> in production, and a bare
|
||||
/// connection in a test — without dragging the DI graph along. Mirrors
|
||||
/// <c>DeploymentCacheSchema</c>.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Why <see cref="IdColumn"/> is TEXT and app-minted.</b> The legacy queue keyed on
|
||||
/// <c>RowId INTEGER PRIMARY KEY AUTOINCREMENT</c>. Convergence is last-writer-wins over the
|
||||
/// primary key, so two nodes independently allocating rowid 7 for different alarms would
|
||||
/// silently overwrite one another. The sink mints the id from a hash of the payload, which
|
||||
/// additionally makes the same event converge to one row when both nodes of a pair enqueue
|
||||
/// it — as they legitimately do in the window before the first redundancy snapshot arrives.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
public static class AlarmSfSchema
|
||||
{
|
||||
/// <summary>Table holding queued alarm events awaiting delivery.</summary>
|
||||
public const string EventsTable = "alarm_sf_events";
|
||||
|
||||
/// <summary>The single primary-key column.</summary>
|
||||
public const string IdColumn = "id";
|
||||
|
||||
/// <summary>
|
||||
/// Creates the buffer table if it does not already exist. Idempotent.
|
||||
/// </summary>
|
||||
/// <param name="connection">
|
||||
/// An already-open connection. <c>ILocalDb.CreateConnection()</c> hands out open,
|
||||
/// pragma-configured connections — do not call <c>Open()</c> on one.
|
||||
/// </param>
|
||||
public static void Apply(SqliteConnection connection)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(connection);
|
||||
|
||||
using var cmd = connection.CreateCommand();
|
||||
|
||||
// Columns map 1:1 onto the legacy Queue table so the one-time migrator is a straight copy
|
||||
// and the drain keeps its existing semantics: dead_lettered is the same 0/1 flag, and an
|
||||
// acknowledged row is DELETEd rather than marked. Deleting keeps the table bounded without
|
||||
// a second sweeper, and the replication engine carries the delete as a tombstone, so the
|
||||
// peer drops its copy too.
|
||||
//
|
||||
// The drain index orders by enqueued_at_utc because a hashed TEXT id carries no insertion
|
||||
// order the way the legacy AUTOINCREMENT RowId did. Timestamps are round-trip ("O") format,
|
||||
// so lexicographic ordering is chronological; id breaks ties so the order is total and the
|
||||
// index covers the drain's read.
|
||||
cmd.CommandText = """
|
||||
CREATE TABLE IF NOT EXISTS alarm_sf_events (
|
||||
id TEXT NOT NULL PRIMARY KEY,
|
||||
alarm_id TEXT NOT NULL,
|
||||
enqueued_at_utc TEXT NOT NULL,
|
||||
payload_json TEXT NOT NULL,
|
||||
attempt_count INTEGER NOT NULL DEFAULT 0,
|
||||
last_attempt_utc TEXT NULL,
|
||||
last_error TEXT NULL,
|
||||
dead_lettered INTEGER NOT NULL DEFAULT 0
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS ix_alarm_sf_events_drain
|
||||
ON alarm_sf_events (dead_lettered, enqueued_at_utc, id);
|
||||
""";
|
||||
cmd.ExecuteNonQuery();
|
||||
}
|
||||
}
|
||||
@@ -1,11 +1,12 @@
|
||||
using ZB.MOM.WW.LocalDb;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian;
|
||||
using ZB.MOM.WW.OtOpcUa.Runtime.DeploymentCache;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Host.Configuration;
|
||||
|
||||
/// <summary>
|
||||
/// The <c>onReady</c> callback handed to <c>AddZbLocalDb</c>: creates the deployment-cache
|
||||
/// tables and opts them into replication.
|
||||
/// The <c>onReady</c> callback handed to <c>AddZbLocalDb</c>: creates the deployment-cache and
|
||||
/// alarm store-and-forward tables and opts them into replication.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
@@ -26,8 +27,15 @@ 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. Phase 1 writes nothing here; when Phase 2 adds its store-and-forward
|
||||
/// migrator, the migrator must run <i>after</i> both registrations for the same reason.
|
||||
/// history. The Phase-2 legacy alarm migrator therefore runs <i>after</i> every
|
||||
/// registration, not alongside the DDL.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// The alarm buffer's tables are created unconditionally, regardless of whether this
|
||||
/// node has <c>AlarmHistorian:Enabled</c> set. An empty registered table costs three
|
||||
/// triggers and nothing else, whereas creating it lazily when the sink first appears
|
||||
/// would mean a node that enables the historian later writes rows before its triggers
|
||||
/// exist — which is precisely the silent-loss shape above.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
/// <param name="db">The freshly constructed local database.</param>
|
||||
@@ -40,9 +48,11 @@ public static class LocalDbSetup
|
||||
using (var connection = db.CreateConnection())
|
||||
{
|
||||
DeploymentCacheSchema.Apply(connection);
|
||||
AlarmSfSchema.Apply(connection);
|
||||
}
|
||||
|
||||
db.RegisterReplicated(DeploymentCacheSchema.ArtifactsTable);
|
||||
db.RegisterReplicated(DeploymentCacheSchema.PointerTable);
|
||||
db.RegisterReplicated(AlarmSfSchema.EventsTable);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -46,7 +46,7 @@ public sealed class LocalDbSetupTests : IDisposable
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void OnReady_RegistersExactlyTheTwoDeploymentTables()
|
||||
public void OnReady_RegistersExactlyTheThreeReplicatedTables()
|
||||
{
|
||||
// Both directions are load-bearing. "No fewer" catches a dropped RegisterReplicated call
|
||||
// (that table then never replicates). "No more" catches an accidental registration —
|
||||
@@ -54,7 +54,7 @@ public sealed class LocalDbSetupTests : IDisposable
|
||||
var db = BuildDb();
|
||||
|
||||
db.ReplicatedTables.Keys.OrderBy(k => k, StringComparer.Ordinal)
|
||||
.ShouldBe(["deployment_artifacts", "deployment_pointer"]);
|
||||
.ShouldBe(["alarm_sf_events", "deployment_artifacts", "deployment_pointer"]);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
@@ -78,6 +78,50 @@ public sealed class LocalDbSetupTests : IDisposable
|
||||
db.ReplicatedTables["deployment_pointer"].PkColumns.ShouldBe(["cluster_id"]);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void AlarmSfEvents_PkIsTheAppMintedId()
|
||||
{
|
||||
// A single TEXT PK the application mints. The legacy queue this table replaces keyed on
|
||||
// `RowId INTEGER PRIMARY KEY AUTOINCREMENT`, which LWW cannot replicate: two nodes would
|
||||
// independently allocate rowid 7 for different alarms and silently overwrite each other.
|
||||
var db = BuildDb();
|
||||
|
||||
db.ReplicatedTables["alarm_sf_events"].PkColumns.ShouldBe(["id"]);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task AlarmSfEventRows_EnterTheOplog()
|
||||
{
|
||||
// Same ordering assertion as the deployment-pointer test below, for the table Phase 2 adds.
|
||||
// Worth its own case because the alarm table is registered by a different call site, and a
|
||||
// registration added ahead of its DDL would fail loudly while one added after the migrator
|
||||
// would fail silently.
|
||||
var db = BuildDb();
|
||||
|
||||
await db.ExecuteAsync(
|
||||
"""
|
||||
INSERT INTO alarm_sf_events
|
||||
(id, alarm_id, enqueued_at_utc, payload_json, attempt_count)
|
||||
VALUES (@Id, @AlarmId, @EnqueuedAtUtc, @PayloadJson, 0)
|
||||
""",
|
||||
new
|
||||
{
|
||||
Id = new string('c', 64),
|
||||
AlarmId = "equip/alarm-1",
|
||||
EnqueuedAtUtc = "2026-07-21T00:00:00.0000000Z",
|
||||
PayloadJson = """{"AlarmId":"equip/alarm-1"}""",
|
||||
},
|
||||
TestContext.Current.CancellationToken);
|
||||
|
||||
var oplogRows = await db.QueryAsync(
|
||||
"SELECT COUNT(*) FROM __localdb_oplog WHERE table_name = 'alarm_sf_events'",
|
||||
r => r.GetInt32(0),
|
||||
parameters: null,
|
||||
TestContext.Current.CancellationToken);
|
||||
|
||||
oplogRows[0].ShouldBe(1);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task RowsWrittenAfterOnReady_EnterTheOplog()
|
||||
{
|
||||
|
||||
Reference in New Issue
Block a user