feat(localdb): deployment-cache schema + OnReady registration
This commit is contained in:
Joseph Doherty
@@ -0,0 +1,48 @@
|
||||
using ZB.MOM.WW.LocalDb;
|
||||
using ZB.MOM.WW.OtOpcUa.Runtime.Deployment;
|
||||
|
||||
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.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// Public rather than internal only so <c>LocalDbSetupTests</c> 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 <b>this</b> method.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
public static class LocalDbSetup
|
||||
{
|
||||
/// <summary>
|
||||
/// Initialises the local database: DDL first, then replication registration.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// <b>THE ORDER IS LOAD-BEARING: DDL → RegisterReplicated → writes.</b>
|
||||
/// <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.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
/// <param name="db">The freshly constructed local database.</param>
|
||||
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);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,70 @@
|
||||
using Microsoft.Data.Sqlite;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Runtime.Deployment;
|
||||
|
||||
/// <summary>
|
||||
/// DDL for the node-local deployment-artifact cache: the chunked artifact table plus the
|
||||
/// per-cluster current-deployment pointer.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <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.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Why the artifact is chunked base64 TEXT and not one BLOB row.</b> Two independent
|
||||
/// constraints force it. <c>RegisterReplicated</c> rejects BLOB columns outright; and the
|
||||
/// replication engine batches by <i>row count</i> 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.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Why no autoincrement PKs.</b> 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.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
public static class DeploymentCacheSchema
|
||||
{
|
||||
/// <summary>Table holding the artifact bytes, split into base64 chunks.</summary>
|
||||
public const string ArtifactsTable = "deployment_artifacts";
|
||||
|
||||
/// <summary>Table holding one current-deployment pointer per cluster.</summary>
|
||||
public const string PointerTable = "deployment_pointer";
|
||||
|
||||
/// <summary>
|
||||
/// Creates both cache tables if they do 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();
|
||||
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();
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
|
||||
/// <summary>
|
||||
/// LocalDb Phase 1 (Task 2) — pins the deployment-cache schema and the load-bearing
|
||||
/// <c>DDL → RegisterReplicated → writes</c> ordering inside <see cref="LocalDbSetup.OnReady"/>.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// The ordering assertion is the point of this file. <c>RegisterReplicated</c> installs the
|
||||
/// capture triggers; rows written <b>before</b> it are never captured into the oplog and
|
||||
/// therefore never reach the peer — silently, and forever. A reordered <c>OnReady</c> passes
|
||||
/// every schema-shape test while shipping a replication path that quietly moves nothing.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// Built against a real temp-file <see cref="ILocalDb"/> 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.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
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<string, string?> { ["LocalDb:Path"] = _dbPath })
|
||||
.Build();
|
||||
|
||||
_provider = new ServiceCollection()
|
||||
.AddZbLocalDb(configuration, LocalDbSetup.OnReady)
|
||||
.BuildServiceProvider();
|
||||
|
||||
return _provider.GetRequiredService<ILocalDb>();
|
||||
}
|
||||
|
||||
[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);
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user