diff --git a/CLAUDE.md b/CLAUDE.md index 2cf73e4a..cb59b809 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -650,7 +650,9 @@ deliberately unresolvable gateway endpoint, so the buffer is always a live, non- `.Replication`), bumped from `0.1.3` on 2026-08-15. It was a **pure pin bump — no code change**: `RegisterReplicated(string)` is unchanged, everything 0.2.x added is additive, the wire protocol is untouched (0.1.x peers still sync), and the on-disk bookkeeping schema v1 → v2 upgrades in place on -open, so no rig or volume action was needed. Three things about it are worth knowing: +open, so no rig or volume action was needed. **Its two new APIs — `DeregisterReplicated` and +`RegisterReplicated(table, baselineExistingRows:)` — are now both used** (2026-08-18); the bullets +below are the shape that landed, not the bump itself. Two things are worth knowing: - **`MaxBatchSize` no longer means what the rig's `16` was chosen to mean.** 0.2.0 made batching **byte-budgeted** — `LocalDb:Replication:MaxBatchBytes` (2 MB default, left unset here, sized @@ -660,21 +662,46 @@ open, so no rig or volume action was needed. Three things about it are worth kno `16` — it now bounds that transient allocation (≈2.7 MB at ≈171 KB artifact chunk rows), not the message size. Every comment and doc row that called it "row-count-only against gRPC's 4 MB cap" was corrected with the bump; do not restore that rationale. -- **`DeregisterReplicated` exists and this repo deliberately does not call it.** `LocalDbSetup.OnReady` - registers all four tables **unconditionally**, on every driver node, replicating or not — a - replication-configured node and a default-OFF one (central, site-b) carry identical CDC triggers. - ScadaBridge went the other way (conditional registration + a boot-time deregister self-heal), so - do not read that pattern across as settled here: it would be a behaviour change, and - `LocalDbSetupTests` pins the registered-table set and the DDL→register ordering exactly. The - cost of the current shape is three unused triggers per table on an unreplicated node; the - benefit is that turning replication on is a config change that needs no re-registration pass. -- **Consequently, `baselineExistingRows` is not used either.** Rows written on a node *before* - replication is enabled are not in `__localdb_row_version`, so enabling replication on a - long-running default-OFF node (central, site-b) converges on subsequent **writes**, not on the - rows already in the file. Passing `RegisterReplicated(table, baselineExistingRows: true)` is the - library's answer, but it flags a snapshot resync and is only correct when gated on replication - actually being configured — which the unconditional registration above cannot express. Enable - replication on a fresh pair, or accept the gap and let the next deploy re-cache. +- **Registration FOLLOWS the replication config — it is no longer unconditional (2026-08-18).** + `LocalDbSetup.OnReady` reads one predicate, `LocalDbSetup.ReplicationIsConfigured`: + **`LocalDb:Replication:PeerAddress` OR `LocalDb:Replication:ApiKey` is non-blank.** The **OR is + not belt-and-braces** — only the dialling half of a pair sets `PeerAddress`, so keying on it + alone would strip capture from every *passive* node (site-a-2 carries `ApiKey` + + `LocalDb:SyncListenPort` and nothing else) and the pair would converge in one direction only, + silently. ⚠️ **`SyncListenPort` deliberately does NOT join the OR:** the passive endpoint is + fail-closed on the key, so a port without an `ApiKey` rejects every inbound stream and has + nothing to capture for. + - **Configured ⇒ `RegisterReplicated(table, baselineExistingRows: true)`**, per table, in the + same load-bearing DDL→register order. Baselining is what makes enabling replication on a node + that has been running *without* it actually converge: capture is CDC and the snapshot streamer + pages from `__localdb_row_version`, so pre-registration rows were in neither and a late opt-in + used to replicate only writes made after the restart. It seeds the ledger at the LWW floor + (HLC `0`, this node's id) and flags a resync — a baselined row loses to any genuine remote + write of the same key and wins only where the peer has no version at all. It is idempotent, so + it is free on every boot after the first. ⚠️ On the pinned **0.2.1** the snapshot flag is + per-**database**, not per-table, so baselining anything streams *every* registered table in + full, in both directions at once — trim `alarm_sf_events` before enabling replication on a node + that has been buffering alarm history for months, and expect one slow first sync. + - **Not configured ⇒ `DeregisterReplicated(table)` per table**, an idempotent boot-time + self-heal: it drops the capture triggers an earlier (unconditional) build left in the file and + prunes that table's oplog + row-version rows. Merely declining to register would leave every + pre-2026-08-18 file capturing forever — two INSERTs plus a `json_object` of the whole row + inside every write transaction — on precisely the node that has no peer to capture for. The + call returns `true` when it removed something, so the Information line fires **once, and only + when it actually cleaned**; a default-OFF node's ordinary boot stays silent. The logger is + Serilog's static `Log.Logger` (this runs inside the `AddZbLocalDb` singleton factory, where + resolving one from DI would nest a resolution inside a singleton construction; `Program.cs` + assigns it before anything resolves `ILocalDb`), with an optional parameter so tests observe + the line rather than infer it. + - ⚠️ **PAIR SYMMETRY: flipping registration is a BOTH-NODES-TOGETHER change.** The sync handshake + compares registered-table digests **fail-closed**, so a node that deregisters a table its peer + still replicates stops syncing with a schema mismatch. Enabling replication on an established + pair means setting the config and **restarting both nodes** — do it on both, or on neither. + - **Rollout note:** the only replicating pair in the fleet is docker-dev **site-a**, and it was + enabled from a fresh pair, so the impact of the old gap was low — central and site-b are + default-OFF and simply shed their unused triggers on their next boot. `LocalDbSetupTests` + pins both branches (registered-table set on a configured node, empty set + zero triggers + + the HLC-0 baseline ledger entry on the other) and the DDL→register ordering, unchanged. See `docs/Configuration.md` (`LocalDb` section), `docs/Redundancy.md` (pair-local config cache), `docs/AlarmHistorian.md`, and the runbook diff --git a/docs/Configuration.md b/docs/Configuration.md index d6b03b42..2bccc843 100644 --- a/docs/Configuration.md +++ b/docs/Configuration.md @@ -418,6 +418,18 @@ replication stays inert until the sync port + peer are set. | `LocalDb:Replication:MaxBatchSize` | `500` (library) | **Secondary** row-count cap on a batch — *and* the DB read-page size. `16` on the replicating pair. Since LocalDb 0.2.0 the wire message is bounded by `MaxBatchBytes`, not by this, so it is no longer the hand-tuned byte proxy the 4 MB cap once forced. It still matters because `OplogStore.ReadBatchAboveAsync`/`SnapshotStreamer` materialise a whole read page **before** the byte-budget split runs — `16` bounds that transient allocation at ≈2.7 MB for ≈171 KB artifact chunk rows. | | `LocalDb:Replication:MaxBatchBytes` | `2 MB` (library) | **Primary** cap on one outbound delta/snapshot message: the summed serialized size of its entries, sized under gRPC's 4 MB receive limit. Left unset across the fleet. A single row larger than the budget is sent **alone** rather than stalling the stream. Added in LocalDb 0.2.0; needs no wire change, so a node on this version interoperates with any earlier peer in both directions. | +**These two keys also decide whether the node captures at all.** `LocalDbSetup.ReplicationIsConfigured` is +`PeerAddress` **OR** `ApiKey` non-blank — the OR because only the *dialling* half sets `PeerAddress`, so +keying on it alone would strip change capture from every passive node and leave the pair converging in one +direction only. `SyncListenPort` is deliberately **not** part of it (a port with no `ApiKey` is fail-closed +and syncs with nobody). A configured node registers all four tables with `baselineExistingRows: true`, so +turning replication on converges the rows **already in the file** and not merely subsequent writes; an +unconfigured node **deregisters** them at boot, dropping capture triggers an earlier build installed. ⚠️ The +sync handshake compares registered-table digests **fail-closed**, so flipping either key is a +both-nodes-together change: set it on both and restart both. On LocalDb 0.2.1 baselining flags a +**per-database** snapshot resync (every registered table, in full, both directions), so trim a long-running +`alarm_sf_events` buffer before enabling replication late. + Replication is **default-OFF** across the fleet; it is enabled per-pair as an opt-in. See [`docs/operations/2026-07-20-localdb-pair-replication.md`](operations/2026-07-20-localdb-pair-replication.md) for the enablement runbook (ApiKey handling, stop/start-together rule, tombstone-retention window, and the diff --git a/docs/operations/2026-07-20-localdb-pair-replication.md b/docs/operations/2026-07-20-localdb-pair-replication.md index 2afe8ca6..51c401c8 100644 --- a/docs/operations/2026-07-20-localdb-pair-replication.md +++ b/docs/operations/2026-07-20-localdb-pair-replication.md @@ -82,8 +82,28 @@ Replication has two knobs, both under `LocalDb`: > wire protocol is unchanged (0.1.x peers still sync), and the on-disk bookkeeping schema v1 → v2 > is upgraded **in place on open**, so an existing node's database file needs no operator action. > Two behaviours changed that an operator should know: batching is now byte-budgeted (the two rows -> above), and `DeregisterReplicated` exists — this repo never calls it, so the deregistration -> symmetry rule and the HLC-anchor fix 0.2.1 shipped for it do not apply to an OtOpcUa node. +> above), and **both of 0.2.x's lifecycle APIs are now used** (2026-08-18) — see the box below. + +> **⚠️ Registration follows this config, and flipping it is a BOTH-NODES change.** Since +> 2026-08-18 `LocalDbSetup.OnReady` installs the change-capture triggers only when the node is +> configured to replicate — the predicate is **`PeerAddress` OR `ApiKey` non-blank** (the OR because +> the passive node has no `PeerAddress`; `SyncListenPort` is deliberately not part of it, since a +> port with no key is fail-closed and syncs with nobody). Consequences at enablement time: +> +> - A configured node registers with **`baselineExistingRows: true`**, so the rows **already in the +> file** are seeded into the row-version ledger at the LWW floor (HLC `0`) and actually travel. +> Before this, enabling replication on a node that had been running converged on *subsequent +> writes only*. ⚠️ On 0.2.1 the snapshot flag is **per-database**: baselining streams *every* +> registered table in full, in both directions at once. **Trim a long-standing `alarm_sf_events` +> backlog before enabling replication late**, and expect one slow first sync. +> - An unconfigured node **deregisters** at boot — the capture triggers a pre-2026-08-18 build left +> behind are dropped and that table's oplog/row-version rows pruned. It logs one Information line +> naming the table count, and stays silent when there was nothing to clean. +> - The sync handshake compares **registered-table digests, fail-closed**, so a node that +> deregisters a table its peer still replicates stops syncing with a schema mismatch. **Set the +> keys on both nodes and restart both** — never one. (0.2.1's HLC-anchor fix now matters here: +> the clock is flushed inside the deregistration transaction, so a crash right after a +> deregister boot cannot bring the clock below stamps the peer already holds.) Example (site-a-1 initiator, site-a-2 passive): 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 cd70e38c..87c307cc 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/LocalDbSetup.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/LocalDbSetup.cs @@ -7,8 +7,8 @@ namespace ZB.MOM.WW.OtOpcUa.Host.Configuration; /// /// The onReady callback handed to AddZbLocalDb: creates the deployment-cache, -/// alarm store-and-forward, and scripted-alarm condition-state tables and opts them into -/// replication. +/// alarm store-and-forward, and scripted-alarm condition-state tables and — when this node +/// actually replicates — opts them into change capture. /// /// /// @@ -21,7 +21,25 @@ namespace ZB.MOM.WW.OtOpcUa.Host.Configuration; public static class LocalDbSetup { /// - /// Initialises the local database: DDL first, then replication registration. + /// The four tables that participate in replication — registered when this node is configured + /// to replicate, and deregistered (stale capture triggers dropped) when it is not. + /// + /// + /// One list shared by both branches on purpose. A table added to only the registration + /// branch would replicate on a configured node and then never be cleaned off an + /// unconfigured one; a table added to only the deregistration branch would be cleaned on a + /// node that never registered it. + /// + private static readonly string[] ReplicatedTables = + [ + DeploymentCacheSchema.ArtifactsTable, + DeploymentCacheSchema.PointerTable, + AlarmSfSchema.EventsTable, + AlarmConditionStateSchema.StateTable, + ]; + + /// + /// Initialises the local database: DDL first, then replication registration (or its removal). /// /// /// @@ -33,20 +51,45 @@ public static class LocalDbSetup /// registration — it writes rows, and they must be captured like any other write. /// /// + /// Registration is CONDITIONAL on . It used + /// to be unconditional, on every driver node, replicating or not. That was not merely + /// wasteful — an unreplicated node paid two INSERTs plus a json_object of the + /// whole row inside every write transaction, for an oplog nothing ever drained — it also + /// made baselineExistingRows unusable, because passing it on a node with no peer + /// would flag a snapshot resync for a peer that does not exist. The ordering rule above + /// is unaffected: when registration is skipped there is no capture to be too late for. + /// + /// + /// The unconfigured branch actively cleans up rather than merely abstaining. + /// DeregisterReplicated drops any capture triggers an earlier build left in the + /// file and prunes that table's oplog and row-version rows. Declining to register alone + /// would leave every file written by a pre-2026-08-18 build capturing forever — on + /// precisely the node that has no peer to capture for. + /// + /// /// 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. + /// node has AlarmHistorian:Enabled set, and regardless of replication. Creating + /// one 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. /// - /// Application configuration, read by the legacy migrator for the pre-consolidation queue's - /// path. Required rather than optional: an overload that silently skipped the migration - /// would be one wiring mistake away from discarding a node's undelivered alarm history. + /// Application configuration, read for the replication predicate and by the legacy migrator + /// for the pre-consolidation queue's path. Required rather than optional: an overload that + /// silently skipped the migration would be one wiring mistake away from discarding a node's + /// undelivered alarm history. /// - public static void OnReady(ILocalDb db, IConfiguration configuration) + /// + /// Where the deregistration self-heal reports what it cleaned. Defaults to Serilog's static + /// logger, which is what production uses: this runs inside the AddZbLocalDb singleton + /// factory, where resolving an ILogger would nest a service resolution inside a + /// singleton construction. Program.cs assigns Serilog.Log.Logger from the DI + /// root logger before anything resolves ILocalDb. The parameter exists so tests can + /// observe the line rather than infer it. + /// + public static void OnReady(ILocalDb db, IConfiguration configuration, Serilog.ILogger? logger = null) { ArgumentNullException.ThrowIfNull(db); ArgumentNullException.ThrowIfNull(configuration); @@ -60,12 +103,138 @@ public static class LocalDbSetup AlarmConditionStateSchema.Apply(connection); } - db.RegisterReplicated(DeploymentCacheSchema.ArtifactsTable); - db.RegisterReplicated(DeploymentCacheSchema.PointerTable); - db.RegisterReplicated(AlarmSfSchema.EventsTable); - db.RegisterReplicated(AlarmConditionStateSchema.StateTable); + if (ReplicationIsConfigured(configuration)) + RegisterForCapture(db); + else + RemoveStaleCapture(db, logger ?? Serilog.Log.Logger); // LAST, and only here. This is the one call in OnReady that writes rows. AlarmSfLegacyMigrator.Migrate(db, configuration); } + + /// + /// Whether this node participates in LocalDb replication, and therefore needs the capture + /// triggers at all. + /// + /// + /// + /// Either key counts, and the OR is not belt-and-braces — it is the only predicate + /// that is true on BOTH halves of a replicating pair. Replication is one + /// bidirectional stream that exactly one side dials, so only the initiator sets + /// LocalDb:Replication:PeerAddress; the passive node carries the shared + /// ApiKey and a LocalDb:SyncListenPort and nothing else. (The docker-dev + /// rig is the reference: site-a-1 has PeerAddress + ApiKey, site-a-2 has + /// ApiKey + SyncListenPort, and central / site-b carry no Replication section at + /// all.) Keying on PeerAddress alone would strip capture from every passive node — its + /// local writes would stop reaching the initiator and the pair would converge in one + /// direction only, silently. + /// + /// + /// LocalDb:SyncListenPort deliberately does NOT join the OR. The passive + /// endpoint is fail-closed on the key: with a port but no ApiKey, + /// LocalDbSyncAuthInterceptor refuses every inbound stream, so such a node + /// replicates nothing and has nothing to capture for. Adding the port to the predicate + /// would install capture triggers on a node whose sync endpoint rejects all callers. + /// LocalDbReplicationHealthCheck deliberately asks a different question — + /// "was this node MEANT to replicate?", which is PeerAddress OR a bound listen + /// port — so a port-without-key node correctly reports Degraded there while correctly + /// capturing nothing here. The two predicates diverge only on that misconfiguration, + /// and each is right about its own question. + /// + /// + /// The keys are read straight from configuration rather than through + /// IOptions<ReplicationOptions> because this runs inside the + /// AddZbLocalDb factory, where resolving another option would nest a service + /// resolution inside a singleton construction. The section path is + /// — the same constant + /// AddZbLocalDbReplication binds — so the predicate and the engine cannot + /// disagree about which node replicates. + /// + /// + /// Both directions of a change to this predicate are handled at boot, and both + /// require the PAIR to be changed together. The sync handshake compares + /// registered-table digests fail-closed, so a node that drops a table its peer still + /// replicates stops syncing with a schema mismatch. Enable (or disable) replication on + /// both nodes and restart both — never one. + /// + /// + /// The application configuration. + /// true when this node is configured to replicate. + public static bool ReplicationIsConfigured(IConfiguration configuration) + { + ArgumentNullException.ThrowIfNull(configuration); + + var section = configuration.GetSection(LocalDbRegistration.ReplicationSectionPath); + + return !string.IsNullOrWhiteSpace(section["PeerAddress"]) + || !string.IsNullOrWhiteSpace(section["ApiKey"]); + } + + /// Opts every replicated table into change capture, baselining what is already there. + /// + /// + /// baselineExistingRows: true is what makes turning replication ON for a node that + /// has been running WITHOUT it actually converge. Capture is change-data-capture and the + /// snapshot streamer pages from __localdb_row_version, so rows written before the + /// first registration are in neither: a late opt-in used to replicate only writes made + /// after the restart, silently and forever. Baselining seeds the ledger at the LWW floor + /// (HLC 0, this node's id) and flags a snapshot resync, so a baselined row loses to any + /// genuine remote write of the same key and wins only where the peer has no version at + /// all. It is idempotent, and therefore free on every boot after the first — which is + /// why it is unconditional here rather than behind a one-shot flag. + /// + /// + /// ⚠️ On the pinned LocalDb 0.2.1 the snapshot flag is per-DATABASE, not per-table. + /// Baselining anything flags a resync that streams every registered table in full, + /// and on a pair enabling replication late both nodes do that to each other at once — the + /// largest transfer the library performs. Bounded, but expect one slow first sync, and + /// trim alarm_sf_events before enabling replication on a node that has been + /// buffering alarm history for a long time. + /// + /// + /// The per-table loop is kept rather than collapsed into a hypothetical bulk call: the + /// library's contract is per-table (PK validation, digest, and — from LocalDb 0.3.0 — the + /// snapshot flag itself), so a bulk API would only be able to give back the coarse + /// behaviour. + /// + /// + private static void RegisterForCapture(ILocalDb db) + { + foreach (var table in ReplicatedTables) + db.RegisterReplicated(table, baselineExistingRows: true); + } + + /// Drops capture this node must not pay for, and says so once if it found any. + /// + /// DeregisterReplicated is idempotent and deliberately usable on a table this process + /// never registered, so a never-replicated file simply reports nothing to clean and stays + /// silent. Symmetry is not a concern on this branch: this node has no peer, so there is no + /// handshake digest for a one-sided deregistration to fail — and re-enabling later is safe + /// despite the ledger prune, because baselines. + /// + private static void RemoveStaleCapture(ILocalDb db, Serilog.ILogger logger) + { + var cleaned = 0; + + foreach (var table in ReplicatedTables) + { + if (db.DeregisterReplicated(table)) + cleaned++; + } + + // Only when something was actually removed. Every default-OFF node in the fleet takes this + // branch on every boot; logging unconditionally would be a daily non-event, which is how the + // one boot that DID silently stop capturing gets overlooked. + if (cleaned > 0) + { + logger.Information( + "LocalDb: removed stale change-capture from {TableCount} table(s) — this node has no " + + "replication configured ({PeerAddressKey}/{ApiKeyKey} are both unset), so the capture " + + "triggers an earlier build installed have been dropped and their oplog/row-version " + + "rows pruned.", + cleaned, + $"{LocalDbRegistration.ReplicationSectionPath}:PeerAddress", + $"{LocalDbRegistration.ReplicationSectionPath}:ApiKey"); + } + } } diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDb/AlarmSfLegacyMigratorTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDb/AlarmSfLegacyMigratorTests.cs index 0a581633..30660edd 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDb/AlarmSfLegacyMigratorTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDb/AlarmSfLegacyMigratorTests.cs @@ -194,8 +194,17 @@ public sealed class AlarmSfLegacyMigratorTests : IDisposable private ILocalDb BuildDb() { + // The sync ApiKey is what makes OnReady register the tables for capture at all — since + // 2026-08-18 registration follows the replication config rather than being unconditional + // (see LocalDbSetup.ReplicationIsConfigured). Without it these databases would have no + // capture triggers and Migrated_rows_enter_the_oplog would pin nothing. A passive node + // carries exactly this and no PeerAddress, so it is also the realistic shape. var configuration = new ConfigurationBuilder() - .AddInMemoryCollection(new Dictionary { ["LocalDb:Path"] = _dbPath }) + .AddInMemoryCollection(new Dictionary + { + ["LocalDb:Path"] = _dbPath, + ["LocalDb:Replication:ApiKey"] = "test-sync-key", + }) .Build(); _provider = new ServiceCollection() diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDb/LocalDbPairHarness.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDb/LocalDbPairHarness.cs index abfd7e75..0c4a7884 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDb/LocalDbPairHarness.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDb/LocalDbPairHarness.cs @@ -91,8 +91,8 @@ public sealed class LocalDbPairHarness : IAsyncDisposable _apiKeyA = apiKeyA ?? DefaultApiKey; _apiKeyB = apiKeyB ?? DefaultApiKey; - _dbProviderA = BuildDatabaseProvider(_pathA); - _dbProviderB = BuildDatabaseProvider(_pathB); + _dbProviderA = BuildDatabaseProvider(_pathA, _apiKeyA); + _dbProviderB = BuildDatabaseProvider(_pathB, _apiKeyB); } /// Node A — the initiator, which dials the peer. @@ -148,7 +148,7 @@ public sealed class LocalDbPairHarness : IAsyncDisposable DeleteDatabaseFiles(_pathB); _pathB = Path.Combine(Path.GetTempPath(), $"otopcua-pairB-{Guid.NewGuid():N}.db"); - _dbProviderB = BuildDatabaseProvider(_pathB); + _dbProviderB = BuildDatabaseProvider(_pathB, _apiKeyB); _ = B; } @@ -258,10 +258,20 @@ public sealed class LocalDbPairHarness : IAsyncDisposable /// A provider owning one deployment-cache database, initialised through the host's own /// — same schema, same registration order. /// - private static ServiceProvider BuildDatabaseProvider(string path) + private static ServiceProvider BuildDatabaseProvider(string path, string apiKey) { + // The ApiKey is not decoration here. Since 2026-08-18 OnReady registers the tables for + // change capture only when this node is CONFIGURED to replicate — PeerAddress or ApiKey + // (LocalDbSetup.ReplicationIsConfigured) — so a database built from `LocalDb:Path` alone + // would come up with no capture triggers and every convergence scenario below would time + // out with nothing to send. Production reads one configuration; this harness happens to + // build the database and the replication engine from two, and they must agree. var config = new ConfigurationBuilder() - .AddInMemoryCollection(new Dictionary { ["LocalDb:Path"] = path }) + .AddInMemoryCollection(new Dictionary + { + ["LocalDb:Path"] = path, + ["LocalDb:Replication:ApiKey"] = apiKey, + }) .Build(); return new ServiceCollection() 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 df45f06a..d749d476 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDbSetupTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDbSetupTests.cs @@ -1,6 +1,8 @@ using Microsoft.Data.Sqlite; using Microsoft.Extensions.Configuration; using Microsoft.Extensions.DependencyInjection; +using Serilog.Core; +using Serilog.Events; using Shouldly; using Xunit; using ZB.MOM.WW.LocalDb; @@ -20,49 +22,153 @@ namespace ZB.MOM.WW.OtOpcUa.Host.IntegrationTests; /// every schema-shape test while shipping a replication path that quietly moves nothing. /// /// +/// BEHAVIOUR CHANGE (2026-08-18): registration is CONDITIONAL on this node actually +/// being configured to replicate. Until now OnReady registered all four tables +/// unconditionally, so every test here could build a database from LocalDb:Path +/// alone and still find them registered. They now pass a replication section, because that +/// is what makes registration happen at all — see +/// +/// for the other branch. The registered-table-set pin is therefore now a pin on +/// configured nodes; the unconfigured node's pin is the deregistration self-heal. +/// +/// /// Built against a real temp-file rather than a hand-written schema: /// the library has no in-memory mode, and asserting against a schema the test itself wrote -/// would only prove the test agrees with itself. +/// would only prove the test agrees with itself. That is doubly true here — the baseline and +/// deregistration behaviours live entirely inside the library, so only a real file can show +/// them. /// /// public sealed class LocalDbSetupTests : IDisposable { + /// + /// The initiating half of a replicating pair: it dials, so it carries PeerAddress. + /// + private static readonly KeyValuePair[] InitiatorReplication = + [ + new("LocalDb:Replication:PeerAddress", "http://peer:9001"), + new("LocalDb:Replication:ApiKey", "test-sync-key"), + ]; + + /// + /// The passive half of the same pair. It never dials, so it has no PeerAddress + /// — only the shared key (plus a listen port, which is Program.cs's concern). Keying the + /// predicate on PeerAddress alone would strip capture from this node and leave the + /// pair converging in one direction only. + /// + private static readonly KeyValuePair[] PassiveReplication = + [ + new("LocalDb:Replication:ApiKey", "test-sync-key"), + ]; + private readonly string _dbPath = Path.Combine(Path.GetTempPath(), $"otopcua-localdb-setup-{Guid.NewGuid():N}.db"); - private ServiceProvider? _provider; + private readonly List _providers = []; - private ILocalDb BuildDb() + /// + /// Boots over the test's database file. + /// + /// + /// Successive calls reopen the same file, which is how the self-heal and late-opt-in + /// cases are expressed: boot once under one replication configuration, write rows, then boot + /// again under another. That is exactly the sequence an operator performs by editing config + /// and restarting the node. + /// + private ILocalDb Boot( + KeyValuePair[]? replication = null, + Serilog.ILogger? logger = null) { + // A previous boot's provider still holds the file open; the library has no second-handle + // mode. Closing it (and the pool) is what makes a reopen behave like a process restart. + DisposeProviders(); + + var settings = new List> + { + new("LocalDb:Path", _dbPath), + }; + settings.AddRange(replication ?? []); + var configuration = new ConfigurationBuilder() - .AddInMemoryCollection(new Dictionary { ["LocalDb:Path"] = _dbPath }) + .AddInMemoryCollection(settings) .Build(); - _provider = new ServiceCollection() - .AddZbLocalDb(configuration, db => LocalDbSetup.OnReady(db, configuration)) + var provider = new ServiceCollection() + .AddZbLocalDb(configuration, db => LocalDbSetup.OnReady(db, configuration, logger)) .BuildServiceProvider(); - return _provider.GetRequiredService(); + _providers.Add(provider); + + return provider.GetRequiredService(); } + private static async Task ScalarAsync(ILocalDb db, string sql) => + (await db.QueryAsync(sql, r => r.GetInt32(0), parameters: null, TestContext.Current.CancellationToken))[0]; + + private static Task InsertPointerAsync(ILocalDb db, string clusterId) => + db.ExecuteAsync( + """ + INSERT INTO deployment_pointer + (cluster_id, deployment_id, revision_hash, artifact_sha256, applied_at_utc) + VALUES (@ClusterId, @DeploymentId, @RevisionHash, @Sha, @AppliedAtUtc) + """, + new + { + ClusterId = clusterId, + DeploymentId = "0123456789abcdef0123456789abcdef", + RevisionHash = new string('a', 64), + Sha = new string('b', 64), + AppliedAtUtc = "2026-07-20T00:00:00.0000000Z", + }, + TestContext.Current.CancellationToken); + [Fact] - public void OnReady_RegistersExactlyTheFourReplicatedTables() + public void OnReady_WhenReplicationIsConfigured_RegistersExactlyTheFourReplicatedTables() { // Both directions are load-bearing. "No fewer" catches a dropped RegisterReplicated call // (that table then never replicates). "No more" catches an accidental registration — // every replicated table costs three triggers and oplog volume on every write. - var db = BuildDb(); + var db = Boot(InitiatorReplication); db.ReplicatedTables.Keys.OrderBy(k => k, StringComparer.Ordinal) .ShouldBe(["alarm_condition_state", "alarm_sf_events", "deployment_artifacts", "deployment_pointer"]); } + [Fact] + public void OnReady_WhenOnlyTheSyncApiKeyIsSet_StillRegistersTheFourTables() + { + // The passive half of a pair carries the key and nothing else. If the predicate were + // "PeerAddress is set", this node would install no capture triggers and its OWN writes + // would never reach the initiator — a pair that replicates in one direction only, with + // nothing logged anywhere. This test is the falsifier for that narrower predicate. + var db = Boot(PassiveReplication); + + db.ReplicatedTables.Keys.OrderBy(k => k, StringComparer.Ordinal) + .ShouldBe(["alarm_condition_state", "alarm_sf_events", "deployment_artifacts", "deployment_pointer"]); + } + + [Fact] + public async Task OnReady_WhenReplicationIsNotConfigured_RegistersNothingAndInstallsNoTriggers() + { + // The default-OFF node. It has no peer, so capture buys it nothing and costs it two INSERTs + // plus a json_object of the whole row inside every write transaction — for an oplog that is + // never drained. Asserting the trigger count as well as ReplicatedTables matters: the + // dictionary is in-process bookkeeping, the triggers are what actually run. + var db = Boot(); + + db.ReplicatedTables.ShouldBeEmpty(); + + // None of the three application schemas defines a trigger of its own, so every trigger in + // this file would be a capture trigger. + (await ScalarAsync(db, "SELECT COUNT(*) FROM sqlite_master WHERE type = 'trigger'")).ShouldBe(0); + } + [Fact] public void DeploymentArtifacts_PkIsDeploymentIdPlusChunkIndex() { // The composite PK is what makes an artifact chunkable. A single-column PK here would // make every chunk of a deployment collide onto one row under LWW. - var db = BuildDb(); + var db = Boot(InitiatorReplication); db.ReplicatedTables["deployment_artifacts"].PkColumns .ShouldBe(["deployment_id", "chunk_index"]); @@ -73,7 +179,7 @@ public sealed class LocalDbSetupTests : IDisposable { // One current-deployment pointer per cluster: the pair's two nodes converge on the same // row rather than each keeping its own. - var db = BuildDb(); + var db = Boot(InitiatorReplication); db.ReplicatedTables["deployment_pointer"].PkColumns.ShouldBe(["cluster_id"]); } @@ -84,7 +190,7 @@ public sealed class LocalDbSetupTests : IDisposable // 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(); + var db = Boot(InitiatorReplication); db.ReplicatedTables["alarm_sf_events"].PkColumns.ShouldBe(["id"]); } @@ -94,7 +200,7 @@ public sealed class LocalDbSetupTests : IDisposable { // One condition-state row per alarm identity. The pair's two nodes converge on the same row // under LWW rather than each keeping its own, and Save is an unconditional upsert onto it. - var db = BuildDb(); + var db = Boot(InitiatorReplication); db.ReplicatedTables["alarm_condition_state"].PkColumns.ShouldBe(["scripted_alarm_id"]); } @@ -106,7 +212,7 @@ public sealed class LocalDbSetupTests : IDisposable // 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(); + var db = Boot(InitiatorReplication); await db.ExecuteAsync( """ @@ -123,13 +229,8 @@ public sealed class LocalDbSetupTests : IDisposable }, 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); + (await ScalarAsync(db, "SELECT COUNT(*) FROM __localdb_oplog WHERE table_name = 'alarm_sf_events'")) + .ShouldBe(1); } [Fact] @@ -138,40 +239,114 @@ public sealed class LocalDbSetupTests : IDisposable // THE ordering assertion. If DDL and RegisterReplicated were swapped — or a write were // added to OnReady ahead of registration — the triggers would not exist at write time and // this count would stay 0 while every other test in this file still passed. - var db = BuildDb(); + var db = Boot(InitiatorReplication); - await db.ExecuteAsync( - """ - INSERT INTO deployment_pointer - (cluster_id, deployment_id, revision_hash, artifact_sha256, applied_at_utc) - VALUES (@ClusterId, @DeploymentId, @RevisionHash, @Sha, @AppliedAtUtc) - """, - new - { - ClusterId = "SITE-A", - DeploymentId = "0123456789abcdef0123456789abcdef", - RevisionHash = new string('a', 64), - Sha = new string('b', 64), - AppliedAtUtc = "2026-07-20T00:00:00.0000000Z", - }, - TestContext.Current.CancellationToken); + await InsertPointerAsync(db, "SITE-A"); - var oplogRows = await db.QueryAsync( - "SELECT COUNT(*) FROM __localdb_oplog", - r => r.GetInt32(0), - parameters: null, - TestContext.Current.CancellationToken); + (await ScalarAsync(db, "SELECT COUNT(*) FROM __localdb_oplog")).ShouldBeGreaterThanOrEqualTo(1); + } - oplogRows[0].ShouldBeGreaterThanOrEqualTo(1); + [Fact] + public async Task EnablingReplicationLater_BaselinesTheRowsAlreadyInTheFile() + { + // The defect this whole change exists for. Capture is change-data-capture and the snapshot + // streamer pages from __localdb_row_version, so rows written while replication was off are + // in neither — a late opt-in used to converge on subsequent writes only, silently and + // forever. RegisterReplicated(table, baselineExistingRows: true) seeds the ledger at the LWW + // floor and flags a resync so they actually travel. + // + // The assertion is on the ledger entry's HLC, not merely on its existence. A row captured by + // a trigger carries a genuine (large, non-zero) stamp; a BASELINED row is seeded at the LWW + // floor of HLC 0, which is what makes it lose to any real remote write of the same key and + // win only where the peer holds no version at all. Asserting existence alone would pass + // against the old unconditional registration, where the trigger simply captured the write — + // i.e. it would prove nothing about the flag. + var unreplicated = Boot(); + await InsertPointerAsync(unreplicated, "SITE-A"); + + var replicated = Boot(InitiatorReplication); + + (await ScalarAsync( + replicated, + """ + SELECT COUNT(*) FROM __localdb_row_version + WHERE table_name = 'deployment_pointer' AND hlc = 0 + """)) + .ShouldBe(1); + } + + [Fact] + public async Task DisablingReplication_DropsTheCaptureTriggersAnEarlierBootInstalled() + { + // The self-heal half. Merely declining to register would leave a file registered by an + // earlier build capturing forever — on precisely the node that has no peer to capture for. + // DeregisterReplicated drops the triggers and prunes that table's oplog and row-version + // rows. + var replicated = Boot(InitiatorReplication); + await InsertPointerAsync(replicated, "SITE-A"); + (await ScalarAsync(replicated, "SELECT COUNT(*) FROM sqlite_master WHERE type = 'trigger'")) + .ShouldBe(12); + + var unreplicated = Boot(); + + unreplicated.ReplicatedTables.ShouldBeEmpty(); + (await ScalarAsync(unreplicated, "SELECT COUNT(*) FROM sqlite_master WHERE type = 'trigger'")) + .ShouldBe(0); + } + + [Fact] + public async Task DisablingReplication_LogsOnceWhenItActuallyCleanedSomething() + { + // An operator-visible record that a node just stopped capturing, and how much it dropped. + // Silent trigger removal on a node someone MEANT to replicate is the failure this line + // exists to make diagnosable. + var replicated = Boot(InitiatorReplication); + await InsertPointerAsync(replicated, "SITE-A"); + + var sink = new CollectingSink(); + Boot(replication: null, logger: BuildLogger(sink)); + + sink.Events.Count(e => e.Level == LogEventLevel.Information).ShouldBe(1); + } + + [Fact] + public void DisablingReplication_SaysNothingWhenThereWasNothingToClean() + { + // The steady state of every default-OFF node in the fleet: it boots unreplicated every day + // and has nothing to deregister. A log line here would be a daily false alarm, which is how + // the real one gets ignored. + var sink = new CollectingSink(); + Boot(replication: null, logger: BuildLogger(sink)); + + sink.Events.ShouldBeEmpty(); + } + + private static Serilog.ILogger BuildLogger(ILogEventSink sink) => + new Serilog.LoggerConfiguration().MinimumLevel.Verbose().WriteTo.Sink(sink).CreateLogger(); + + /// Captures whatever logs. + private sealed class CollectingSink : ILogEventSink + { + public List Events { get; } = []; + + public void Emit(LogEvent logEvent) => Events.Add(logEvent); + } + + private void DisposeProviders() + { + foreach (var provider in _providers) + provider.Dispose(); + + _providers.Clear(); + + // No in-memory mode, so these are real files. Pooled connections keep a handle open past + // dispose; clearing the pools first is what lets a reopen (or the delete below) succeed. + SqliteConnection.ClearAllPools(); } public void Dispose() { - _provider?.Dispose(); - - // No in-memory mode, so these are real files. Pooled connections keep a handle open past - // dispose; clearing the pools first is what makes the delete actually succeed. - SqliteConnection.ClearAllPools(); + DisposeProviders(); foreach (var path in new[] { _dbPath, $"{_dbPath}-wal", $"{_dbPath}-shm" }) { diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDbWiringTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDbWiringTests.cs index 4dc9bd56..6798faf5 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDbWiringTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDbWiringTests.cs @@ -45,16 +45,27 @@ public sealed class LocalDbWiringTests : IDisposable /// , health, and observability methods, /// over a temp-file LocalDb:Path. /// - private IServiceProvider BuildDriverGraph() + /// + /// Whether this node carries a replication configuration. Since 2026-08-18 that is what + /// decides whether LocalDbSetup.OnReady registers the tables for change capture at + /// all, so the two graphs are genuinely different nodes and both are pinned below. Most of + /// the fleet is the default-OFF one. + /// + private IServiceProvider BuildDriverGraph(bool replicating = false) { var dbPath = Path.Combine(Path.GetTempPath(), $"otopcua-wiring-{Guid.NewGuid():N}.db"); _dbPaths.Add(dbPath); - var builder = WebApplication.CreateBuilder(new WebApplicationOptions { Args = [] }); - builder.Configuration.AddInMemoryCollection(new Dictionary + var settings = new Dictionary { ["LocalDb:Path"] = dbPath, - }); + }; + + if (replicating) + settings["LocalDb:Replication:ApiKey"] = "wiring-test-sync-key"; + + var builder = WebApplication.CreateBuilder(new WebApplicationOptions { Args = [] }); + builder.Configuration.AddInMemoryCollection(settings); // The exact LocalDb composition Program.cs runs under hasDriver. builder.Services.AddOtOpcUaLocalDb(builder.Configuration); @@ -86,9 +97,9 @@ public sealed class LocalDbWiringTests : IDisposable } [Fact] - public void DriverGraph_LocalDbResolvesAsASingletonWithEveryReplicatedTableRegistered() + public void ReplicatingDriverGraph_LocalDbResolvesAsASingletonWithEveryReplicatedTableRegistered() { - var sp = BuildDriverGraph(); + var sp = BuildDriverGraph(replicating: true); var first = sp.GetService(); first.ShouldNotBeNull(); @@ -100,6 +111,22 @@ public sealed class LocalDbWiringTests : IDisposable .ShouldBe(["alarm_condition_state", "alarm_sf_events", "deployment_artifacts", "deployment_pointer"]); } + [Fact] + public void DefaultOffDriverGraph_LocalDbStillResolvesButRegistersNoTables() + { + // BEHAVIOUR CHANGE (2026-08-18): registration follows the replication config. This graph is + // most of the fleet — central and site-b are default-OFF — and it must still get a working + // local database (the artifact cache and the alarm buffer both depend on it) while paying + // for none of the change-capture machinery it has no peer to send to. + var sp = BuildDriverGraph(); + + var db = sp.GetService(); + db.ShouldNotBeNull(); + sp.GetService().ShouldNotBeNull(); + + db.ReplicatedTables.ShouldBeEmpty(); + } + [Fact] public void DriverGraph_ArtifactCacheResolvesToTheLocalDbBackedImplementation() {