diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/LocalDbSetup.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/LocalDbSetup.cs new file mode 100644 index 00000000..e2838d63 --- /dev/null +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/LocalDbSetup.cs @@ -0,0 +1,48 @@ +using ZB.MOM.WW.LocalDb; +using ZB.MOM.WW.OtOpcUa.Runtime.Deployment; + +namespace ZB.MOM.WW.OtOpcUa.Host.Configuration; + +/// +/// The onReady callback handed to AddZbLocalDb: creates the deployment-cache +/// tables and opts them into replication. +/// +/// +/// +/// Public rather than internal only so LocalDbSetupTests can drive the production +/// callback directly. Initialising a test database from a hand-written copy of this schema +/// would prove only that the test agrees with itself — the whole value of those tests is +/// that they exercise this method. +/// +/// +public static class LocalDbSetup +{ + /// + /// Initialises the local database: DDL first, then replication registration. + /// + /// + /// + /// THE ORDER IS LOAD-BEARING: DDL → RegisterReplicated → writes. + /// 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. + /// + /// + /// The freshly constructed local database. + public static void OnReady(ILocalDb db) + { + ArgumentNullException.ThrowIfNull(db); + + // CreateConnection() hands back an already-open, pragma-configured connection carrying the + // zb_hlc_next() UDF the capture triggers need. Calling Open() on it would throw. + using (var connection = db.CreateConnection()) + { + DeploymentCacheSchema.Apply(connection); + } + + db.RegisterReplicated(DeploymentCacheSchema.ArtifactsTable); + db.RegisterReplicated(DeploymentCacheSchema.PointerTable); + } +} diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Deployment/DeploymentCacheSchema.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Deployment/DeploymentCacheSchema.cs new file mode 100644 index 00000000..28f21165 --- /dev/null +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Deployment/DeploymentCacheSchema.cs @@ -0,0 +1,70 @@ +using Microsoft.Data.Sqlite; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Deployment; + +/// +/// DDL for the node-local deployment-artifact cache: the chunked artifact table plus the +/// per-cluster current-deployment pointer. +/// +/// +/// +/// 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. +/// +/// +/// Why the artifact is chunked base64 TEXT and not one BLOB row. Two independent +/// constraints force it. RegisterReplicated rejects BLOB columns outright; and the +/// replication engine batches by row count against gRPC's 4 MB default message cap, +/// so a single multi-megabyte artifact row would simply never be deliverable — at any +/// batch size. A 128 KiB raw chunk is ≈ 171 KB once base64-encoded, which keeps a +/// worst-case batch comfortably inside the cap. +/// +/// +/// Why no autoincrement PKs. Convergence is last-writer-wins keyed on the primary +/// key. Two nodes independently allocating rowid 7 for different rows would silently +/// overwrite one another. Every key here is TEXT or a composite of caller-supplied values. +/// +/// +public static class DeploymentCacheSchema +{ + /// Table holding the artifact bytes, split into base64 chunks. + public const string ArtifactsTable = "deployment_artifacts"; + + /// Table holding one current-deployment pointer per cluster. + public const string PointerTable = "deployment_pointer"; + + /// + /// Creates both cache tables if they do 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(); + cmd.CommandText = """ + CREATE TABLE IF NOT EXISTS deployment_artifacts ( + deployment_id TEXT NOT NULL, + chunk_index INTEGER NOT NULL, + cluster_id TEXT NOT NULL, + revision_hash TEXT NOT NULL, + chunk_count INTEGER NOT NULL, + chunk_base64 TEXT NOT NULL, + cached_at_utc TEXT NOT NULL, + PRIMARY KEY (deployment_id, chunk_index) + ); + CREATE TABLE IF NOT EXISTS deployment_pointer ( + cluster_id TEXT NOT NULL PRIMARY KEY, + deployment_id TEXT NOT NULL, + revision_hash TEXT NOT NULL, + artifact_sha256 TEXT NOT NULL, + applied_at_utc TEXT NOT NULL + ); + """; + cmd.ExecuteNonQuery(); + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDbSetupTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDbSetupTests.cs new file mode 100644 index 00000000..2db014c2 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDbSetupTests.cs @@ -0,0 +1,128 @@ +using Microsoft.Data.Sqlite; +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.DependencyInjection; +using Shouldly; +using Xunit; +using ZB.MOM.WW.LocalDb; +using ZB.MOM.WW.OtOpcUa.Host.Configuration; + +namespace ZB.MOM.WW.OtOpcUa.Host.IntegrationTests; + +/// +/// LocalDb Phase 1 (Task 2) — pins the deployment-cache schema and the load-bearing +/// DDL → RegisterReplicated → writes ordering inside . +/// +/// +/// +/// The ordering assertion is the point of this file. RegisterReplicated installs the +/// capture triggers; rows written before it are never captured into the oplog and +/// therefore never reach the peer — silently, and forever. A reordered OnReady passes +/// every schema-shape test while shipping a replication path that quietly moves nothing. +/// +/// +/// 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. +/// +/// +public sealed class LocalDbSetupTests : IDisposable +{ + private readonly string _dbPath = + Path.Combine(Path.GetTempPath(), $"otopcua-localdb-setup-{Guid.NewGuid():N}.db"); + + private ServiceProvider? _provider; + + private ILocalDb BuildDb() + { + var configuration = new ConfigurationBuilder() + .AddInMemoryCollection(new Dictionary { ["LocalDb:Path"] = _dbPath }) + .Build(); + + _provider = new ServiceCollection() + .AddZbLocalDb(configuration, LocalDbSetup.OnReady) + .BuildServiceProvider(); + + return _provider.GetRequiredService(); + } + + [Fact] + public void OnReady_RegistersExactlyTheTwoDeploymentTables() + { + // 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(); + + db.ReplicatedTables.Keys.OrderBy(k => k, StringComparer.Ordinal) + .ShouldBe(["deployment_artifacts", "deployment_pointer"]); + } + + [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(); + + db.ReplicatedTables["deployment_artifacts"].PkColumns + .ShouldBe(["deployment_id", "chunk_index"]); + } + + [Fact] + public void DeploymentPointer_PkIsClusterId() + { + // 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(); + + db.ReplicatedTables["deployment_pointer"].PkColumns.ShouldBe(["cluster_id"]); + } + + [Fact] + public async Task RowsWrittenAfterOnReady_EnterTheOplog() + { + // 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(); + + 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); + + var oplogRows = await db.QueryAsync( + "SELECT COUNT(*) FROM __localdb_oplog", + r => r.GetInt32(0), + parameters: null, + TestContext.Current.CancellationToken); + + oplogRows[0].ShouldBeGreaterThanOrEqualTo(1); + } + + 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(); + + foreach (var path in new[] { _dbPath, $"{_dbPath}-wal", $"{_dbPath}-shm" }) + { + if (File.Exists(path)) + File.Delete(path); + } + } +}