dohertj2/lmxopcua
1
0
Fork 0
Code Issues 8 Pull Requests Actions Packages Projects Releases Wiki Activity

fix(localdb): close both Phase 1 live-gate follow-ups

Browse Source 
1. Wiped-node back-fill (live gate check 4). Fixed upstream in
   ZB.MOM.WW.LocalDb 0.1.2 and pinned here. The gate's diagnosis was slightly
   off: the oplog cap was not the gate. Snapshot detection measured the peer's
   gap against the oldest surviving oplog row and read an empty oplog as "no
   gap possible" — and an empty oplog is the steady state of a converged pair,
   since ack-pruning deletes everything the peer confirmed. The healthy state
   was the one state that could not heal a wiped peer. Now measured against
   last_acked_seq when the oplog is empty.

   Pinned at this level too, not just the library's: the pair harness grows a
   WipePassiveAsync (a NEW database — a rebuilt node comes back with a new node
   id and a zero watermark), and the new scenario asserts the emptied oplog as
   its precondition before wiping, then requires the deployment artifact to
   come back byte-identical with no new deploy. Verified RED against the pinned
   0.1.1 (times out waiting for back-fill) and green on 0.1.2.

2. Oplog growth on default-OFF nodes (live gate check 8), fixed at the source:
   StoreAsync now skips entirely when the pointer already names this
   deployment/revision, the SHA matches the bytes, and the expected chunk count
   is present. Re-caching an artifact the node already holds — every restart's
   boot-from-cache, every RestoreApplied — writes nothing and mints no oplog
   rows, where before it cost a delete plus an insert per chunk plus a pointer
   update, all identical to what was already there.

   Identity is over the bytes and the chunks are counted, both deliberately: a
   re-composed artifact can carry the same ids with different bytes, and a
   pointer can name a deployment whose chunks are missing, where skipping would
   make an unreadable cache permanent. Both guards verified RED against a naive
   pointer-only skip.

   The gate's stated bound was also wrong and is corrected in the doc: growth
   was never headed for the 1M row cap. AddZbLocalDbReplication is registered
   unconditionally, so MaintenanceBackgroundService runs on default-OFF nodes
   and the 7-day MaxOplogAge cap prunes.

Runtime.Tests 412/0/31, Host.IntegrationTests LocalDb 45/45. Pre-existing and
unrelated: AbCip_Green_AgainstSim (needs the AB CIP docker fixture, red on a
clean master too) and a flaky Roslyn race test (green 2/2 in isolation).

Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW
This commit is contained in:
Joseph Doherty
2026-07-21 01:27:02 -04:00
parent c957db52e2
commit 232bff5df7
7 changed files with 295 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDb/LocalDbPairConvergenceTests.cs
+48
View File
@@ -103,6 +103,54 @@ public sealed class LocalDbPairConvergenceTests
"node B to prune dep1 (with tombstones) while keeping dep3");
}
[Fact]
public async Task WipedNode_IsBackFilledByItsPeer_WithoutAnyNewDeploy()
{
// The Phase 1 live gate's check 4, as an offline scenario. A node whose LocalDb volume is
// lost used to come back permanently empty: the pair had already converged, so every oplog
// row was acked and pruned, and the healthy node had no delta left to send. Nothing short of
// a fresh deploy could repopulate it — which is the worst possible moment to depend on
// central being reachable, since the cache exists precisely for when it is not.
//
// Fixed in ZB.MOM.WW.LocalDb 0.1.2 (snapshot resync now measures the peer's gap against
// last_acked_seq when the oplog is empty). Pin the payoff at this level, not just the
// library's: what matters here is that the deployment artifact itself comes back.
await using var harness = new LocalDbPairHarness();
await harness.StartAsync();
var deploymentId = Guid.NewGuid().ToString("N");
var artifact = MultiChunkArtifact(seed: 20);
await harness.CacheA.StoreAsync(Cluster, deploymentId, "rev-1", artifact,
TestContext.Current.CancellationToken);
await LocalDbPairHarness.WaitUntilAsync(
async () => await ReadArtifactAsync(harness.CacheB, Cluster) is { } b && b.AsSpan().SequenceEqual(artifact),
"the artifact to converge before the wipe");
// Convergence alone is not the precondition — an emptied oplog on A is. That is the state
// that used to make back-fill impossible, so assert it rather than assume it.
await LocalDbPairHarness.WaitUntilAsync(
async () => await LocalDbPairHarness.OplogDepthAsync(harness.A) == 0,
"node A's oplog to be fully acked and pruned");
await harness.WipePassiveAsync();
Assert.Null(await ReadArtifactAsync(harness.CacheB, Cluster));
await harness.RestartPairAsync();
// No new deploy, no central: the rebuilt node gets its cached configuration back from its
// peer alone, byte-identical, and can boot from cache again.
await LocalDbPairHarness.WaitUntilAsync(
async () =>
{
var cached = await harness.CacheB.GetCurrentAsync(Cluster, TestContext.Current.CancellationToken);
return cached is not null
&& cached.DeploymentId == deploymentId
&& cached.Artifact.AsSpan().SequenceEqual(artifact);
},
"the wiped node to be back-filled from its peer");
}
[Fact]
public async Task WritesWhileTransportDown_SurviveRejoin()
{
tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDb/LocalDbPairHarness.cs
+49 -7
View File
@@ -69,10 +69,13 @@ public sealed class LocalDbPairHarness : IAsyncDisposable
private readonly string _apiKeyB;
private readonly string _pathA = Path.Combine(Path.GetTempPath(), $"otopcua-pairA-{Guid.NewGuid():N}.db");
private readonly string _pathB = Path.Combine(Path.GetTempPath(), $"otopcua-pairB-{Guid.NewGuid():N}.db");
// Not readonly: WipePassiveAsync replaces node B's database wholesale, which is what a rebuilt
// node with a lost volume actually is — a new file, and with it a new node id.
private string _pathB = Path.Combine(Path.GetTempPath(), $"otopcua-pairB-{Guid.NewGuid():N}.db");
private readonly ServiceProvider _dbProviderA;
private readonly ServiceProvider _dbProviderB;
private ServiceProvider _dbProviderB;
private IHost? _serverHost; // node B — passive
private IHost? _initiatorHost; // node A — dials the peer
@@ -125,6 +128,30 @@ public sealed class LocalDbPairHarness : IAsyncDisposable
_serverHost = null;
}
/// <summary>
/// Destroys node B's database and replaces it with an empty one, leaving the pairing config
/// untouched — the rebuilt-node case: a lost volume, a re-imaged host, a fresh container.
/// </summary>
/// <remarks>
/// A new file, not a truncated one, because that is what the real thing is: the node id lives
/// in the database, so a rebuilt node comes back with a new identity and a zero watermark. The
/// caller follows with <see cref="RestartPairAsync"/> to bring the wiped node back online.
/// </remarks>
public async Task WipePassiveAsync()
{
await StopPassiveAsync();
await _dbProviderB.DisposeAsync();
// Pooled connections outlive the provider; without this the delete silently no-ops on
// Windows and the "wiped" node would still hold its rows.
Microsoft.Data.Sqlite.SqliteConnection.ClearAllPools();
DeleteDatabaseFiles(_pathB);
_pathB = Path.Combine(Path.GetTempPath(), $"otopcua-pairB-{Guid.NewGuid():N}.db");
_dbProviderB = BuildDatabaseProvider(_pathB);
_ = B;
}
/// <summary>
/// Brings node B back on a NEW loopback port and re-dials from A. The initiator re-reads the
/// peer address on each reconnect, so this is a genuine rejoin over the same databases.
@@ -201,6 +228,17 @@ public sealed class LocalDbPairHarness : IAsyncDisposable
return rows[0];
}
/// <summary>
/// Counts oplog rows on a node. Zero means every local change has been acked by the peer and
/// pruned — the steady state of a converged pair, and the state from which a wiped peer can
/// only be healed by a snapshot.
/// </summary>
public static async Task<long> OplogDepthAsync(ILocalDb db)
{
var rows = await db.QueryAsync("SELECT COUNT(*) FROM __localdb_oplog", static r => r.GetInt64(0));
return rows[0];
}
/// <summary>Counts tombstone rows in <c>__localdb_row_version</c> for one table on a node.</summary>
public static async Task<long> TombstoneCountAsync(ILocalDb db, string table)
{
@@ -313,12 +351,16 @@ public sealed class LocalDbPairHarness : IAsyncDisposable
await _dbProviderB.DisposeAsync();
Microsoft.Data.Sqlite.SqliteConnection.ClearAllPools();
foreach (var path in new[] { _pathA, _pathB })
DeleteDatabaseFiles(_pathA);
DeleteDatabaseFiles(_pathB);
}
/// <summary>Deletes a SQLite database and its WAL/shm sidecars, best effort.</summary>
private static void DeleteDatabaseFiles(string path)
{
foreach (var suffix in new[] { "", "-wal", "-shm" })
{
foreach (var suffix in new[] { "", "-wal", "-shm" })
{
try { File.Delete(path + suffix); } catch { /* best effort */ }
}
try { File.Delete(path + suffix); } catch { /* best effort */ }
}
}
}
tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Deployment/LocalDbDeploymentArtifactCacheTests.cs
+88
View File
@@ -90,6 +90,23 @@ public sealed class LocalDbDeploymentArtifactCacheTests : IDisposable
"SELECT DISTINCT deployment_id FROM deployment_artifacts ORDER BY deployment_id",
r => r.GetString(0));
/// <summary>
/// Total capture-trigger rows in the replication oplog — the direct measure of how much
/// replication traffic (and, on an unpaired node, how much permanent oplog growth) a store
/// costs.
/// </summary>
private async Task<long> OplogDepthAsync()
=> (await _db.QueryAsync("SELECT COUNT(*) FROM __localdb_oplog", r => r.GetInt64(0)))[0];
/// <summary>
/// The LWW version stamps of every replicated row, as a stable ordered projection. Comparing
/// it across a call proves whether that call touched replicated state at all.
/// </summary>
private async Task<IReadOnlyList<string>> RowVersionsAsync()
=> await _db.QueryAsync(
"SELECT table_name, pk_json, hlc FROM __localdb_row_version ORDER BY table_name, pk_json",
r => $"{r.GetString(0)}|{r.GetString(1)}|{r.GetInt64(2)}");
[Fact]
public async Task StoreThenGetCurrent_RoundTripsAMultiChunkArtifactByteForByte()
{
@@ -221,6 +238,77 @@ public sealed class LocalDbDeploymentArtifactCacheTests : IDisposable
cached.Artifact.ShouldBe(second);
}
[Fact]
public async Task Store_WritesNothingWhenTheCachedArtifactIsAlreadyIdentical()
{
// Phase 1 live gate, check 8: every store rewrote the chunks and the pointer unconditionally,
// so re-caching an artifact the node already holds — which is what boot-from-cache and
// RestoreApplied both do on every restart — minted fresh oplog rows for a row nobody changed.
// On a replicating pair that is pure churn; on a default-OFF node there is no peer to ack
// them, so they sit in the oplog until the age cap eventually prunes them.
var cache = CreateCache();
var artifact = RandomArtifact(300 * 1024);
await cache.StoreAsync(ClusterA, "dep-1", "rev-1", artifact);
var oplogDepth = await OplogDepthAsync();
var rowVersions = await RowVersionsAsync();
await cache.StoreAsync(ClusterA, "dep-1", "rev-1", artifact);
(await OplogDepthAsync()).ShouldBe(oplogDepth);
(await RowVersionsAsync()).ShouldBe(rowVersions);
// Skipping must not cost readability: the artifact still reads back intact.
var cached = await cache.GetCurrentAsync(ClusterA);
cached.ShouldNotBeNull();
cached.DeploymentId.ShouldBe("dep-1");
cached.Artifact.ShouldBe(artifact);
}
[Fact]
public async Task Store_RewritesWhenTheCachedChunksAreIncompleteDespiteAMatchingPointer()
{
// The dangerous half of skip-if-unchanged. A pointer can name a deployment whose chunks are
// gone or partial — a half-replicated artifact, or a prune that raced the pointer. Trusting
// the pointer alone would turn a repairable cache into a permanently unreadable one, because
// the re-store that would have fixed it is exactly the call being skipped.
var cache = CreateCache();
var artifact = RandomArtifact(300 * 1024);
await cache.StoreAsync(ClusterA, "dep-1", "rev-1", artifact);
await _db.ExecuteAsync(
"DELETE FROM deployment_artifacts WHERE deployment_id = @DeploymentId AND chunk_index = 2",
new { DeploymentId = "dep-1" });
(await cache.GetCurrentAsync(ClusterA)).ShouldBeNull();
await cache.StoreAsync(ClusterA, "dep-1", "rev-1", artifact);
(await CountChunksAsync("dep-1")).ShouldBe(3);
var cached = await cache.GetCurrentAsync(ClusterA);
cached.ShouldNotBeNull();
cached.Artifact.ShouldBe(artifact);
}
[Fact]
public async Task Store_RewritesWhenTheArtifactBytesChangedUnderAnUnchangedRevision()
{
// The identity check must be over the bytes, not just the ids. A rebuilt artifact carrying
// the same deployment id and revision hash is a real case (the same revision re-composed),
// and silently keeping the stale bytes would serve a boot-from-cache node the wrong address
// space with every id-shaped signal saying it was correct.
var cache = CreateCache();
await cache.StoreAsync(ClusterA, "dep-1", "rev-1", RandomArtifact(1024));
var rebuilt = RandomArtifact(1024);
await cache.StoreAsync(ClusterA, "dep-1", "rev-1", rebuilt);
var cached = await cache.GetCurrentAsync(ClusterA);
cached.ShouldNotBeNull();
cached.Artifact.ShouldBe(rebuilt);
}
[Fact]
public async Task GetCurrentUnkeyed_ReturnsTheOnlyCachedArtifact()
{