diff --git a/docs/plans/2026-07-20-localdb-adoption-phase2.md.tasks.json b/docs/plans/2026-07-20-localdb-adoption-phase2.md.tasks.json index 2f31a474..db3f2c94 100644 --- a/docs/plans/2026-07-20-localdb-adoption-phase2.md.tasks.json +++ b/docs/plans/2026-07-20-localdb-adoption-phase2.md.tasks.json @@ -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, diff --git a/docs/plans/2026-07-20-localdb-phase2-recon.md b/docs/plans/2026-07-20-localdb-phase2-recon.md index 97fad21b..f5714fba 100644 --- a/docs/plans/2026-07-20-localdb-phase2-recon.md +++ b/docs/plans/2026-07-20-localdb-phase2-recon.md @@ -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 diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian/AlarmSfSchema.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian/AlarmSfSchema.cs new file mode 100644 index 00000000..eaf4553c --- /dev/null +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian/AlarmSfSchema.cs @@ -0,0 +1,77 @@ +using Microsoft.Data.Sqlite; + +namespace ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian; + +/// +/// DDL for the alarm store-and-forward buffer: one row per alarm event awaiting delivery to +/// the historian gateway. +/// +/// +/// +/// Replaces the standalone alarm-historian.db 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. +/// +/// +/// Deliberately depends on nothing but so it can be applied +/// to any connection — the host's LocalDbSetup.OnReady in production, and a bare +/// connection in a test — without dragging the DI graph along. Mirrors +/// DeploymentCacheSchema. +/// +/// +/// Why is TEXT and app-minted. The legacy queue keyed on +/// RowId INTEGER PRIMARY KEY AUTOINCREMENT. 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. +/// +/// +public static class AlarmSfSchema +{ + /// Table holding queued alarm events awaiting delivery. + public const string EventsTable = "alarm_sf_events"; + + /// The single primary-key column. + public const string IdColumn = "id"; + + /// + /// Creates the buffer table if it does not already exist. Idempotent. + /// + /// + /// An already-open connection. ILocalDb.CreateConnection() hands out open, + /// pragma-configured connections — do not call Open() on one. + /// + 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(); + } +} diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/LocalDbSetup.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/LocalDbSetup.cs index 0ce2d3be..2819ea17 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/LocalDbSetup.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/LocalDbSetup.cs @@ -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; /// -/// The onReady callback handed to AddZbLocalDb: creates the deployment-cache -/// tables and opts them into replication. +/// The onReady callback handed to AddZbLocalDb: creates the deployment-cache and +/// alarm store-and-forward tables and opts them into replication. /// /// /// @@ -26,8 +27,15 @@ public static class LocalDbSetup /// RegisterReplicated 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 after both registrations for the same reason. + /// history. The Phase-2 legacy alarm migrator therefore runs after every + /// registration, not alongside the DDL. + /// + /// + /// The alarm buffer's tables are created unconditionally, regardless of whether this + /// node has AlarmHistorian:Enabled 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. /// /// /// The freshly constructed local database. @@ -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); } } diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDbSetupTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDbSetupTests.cs index 2db014c2..77481b9b 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDbSetupTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDbSetupTests.cs @@ -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() {