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

test(localdb): 2-node convergence harness + scenarios

Browse Source 
Two driver nodes replicate the deployment-artifact cache over a real loopback
h2c transport through the real fail-closed LocalDbSyncAuthInterceptor; both DBs
init via the production LocalDbSetup.OnReady. Scenarios: byte-identical
convergence with matching pointer HLC+origin, retention prune reaching B as
tombstones, writes-while-transport-down surviving rejoin, and wrong-key
non-convergence with a matching-key positive control.

DoD positive control: commenting out both RegisterReplicated calls turns all
four scenarios red (verified locally, restored before commit).
This commit is contained in:
Joseph Doherty
2026-07-20 22:02:03 -04:00
parent 5a21be0a62
commit afa5be713a
2 changed files with 502 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDb/LocalDbPairConvergenceTests.cs
+178
View File
@@ -0,0 +1,178 @@
using Xunit;
using ZB.MOM.WW.LocalDb;
using ZB.MOM.WW.OtOpcUa.Runtime.DeploymentCache;
namespace ZB.MOM.WW.OtOpcUa.Host.IntegrationTests.LocalDb;
/// <summary>
/// LocalDb Phase 1 (Task 12) — two driver nodes replicating the deployment-artifact cache over a
/// real loopback h2c transport, through the real fail-closed interceptor.
/// </summary>
/// <remarks>
/// This is the test the whole phase exists to pass: does a redundant driver pair actually share
/// the cached configuration a node boots from when central SQL is down? Every upstream piece —
/// the schema, the DI wiring, the interceptor, the chunked cache — can be individually green
/// while the pair still fails to converge.
/// </remarks>
[Collection("LocalDbPairConvergence")]
public sealed class LocalDbPairConvergenceTests
{
private const string Cluster = "cluster-1";
/// <summary>
/// A deterministic artifact large enough to force multiple 128 KiB chunks (≈ 3 here), so the
/// tests exercise the chunk-split/reassemble path rather than a single-row shortcut.
/// </summary>
private static byte[] MultiChunkArtifact(int seed, int size = 300 * 1024)
{
var bytes = new byte[size];
new Random(seed).NextBytes(bytes);
return bytes;
}
private static async Task<byte[]?> ReadArtifactAsync(IDeploymentArtifactCache cache, string cluster)
{
var cached = await cache.GetCurrentAsync(cluster, TestContext.Current.CancellationToken);
return cached?.Artifact;
}
[Fact]
public async Task ArtifactStoredOnA_ConvergesToB_ByteIdentical()
{
await using var harness = new LocalDbPairHarness();
await harness.StartAsync();
var deploymentId = Guid.NewGuid().ToString("N");
var artifact = MultiChunkArtifact(seed: 1);
await harness.CacheA.StoreAsync(Cluster, deploymentId, "rev-1", artifact,
TestContext.Current.CancellationToken);
// B's cache reassembles byte-for-byte what A stored — proving the chunk rows, chunk_count,
// and pointer all replicated intact through the real transport + interceptor.
await LocalDbPairHarness.WaitUntilAsync(
async () =>
{
var onB = await ReadArtifactAsync(harness.CacheB, Cluster);
return onB is not null && onB.AsSpan().SequenceEqual(artifact);
},
"the artifact stored on node A to be byte-identical on node B");
// The pointer row on B carries A's origin HLC + node id, not a locally re-derived stamp: the
// row_version dumps for the pointer table are identical on both nodes. This is what
// distinguishes "B replicated A's row" from "both nodes happened to write equal content".
await LocalDbPairHarness.WaitUntilAsync(
async () =>
await LocalDbPairHarness.DumpRowVersionAsync(harness.A, DeploymentCacheSchema.PointerTable)
== await LocalDbPairHarness.DumpRowVersionAsync(harness.B, DeploymentCacheSchema.PointerTable),
"both nodes' pointer row_version (hlc + origin node id) to match");
}
[Fact]
public async Task RetentionPruneOnA_TombstonesReachB()
{
await using var harness = new LocalDbPairHarness();
await harness.StartAsync();
// Three deployments for one cluster. The cache retains the newest two, so the first is pruned
// on A — and that prune must replicate to B as tombstones, not linger as live chunks.
var dep1 = Guid.NewGuid().ToString("N");
var dep2 = Guid.NewGuid().ToString("N");
var dep3 = Guid.NewGuid().ToString("N");
await harness.CacheA.StoreAsync(Cluster, dep1, "rev-1", MultiChunkArtifact(1),
TestContext.Current.CancellationToken);
await harness.CacheA.StoreAsync(Cluster, dep2, "rev-2", MultiChunkArtifact(2),
TestContext.Current.CancellationToken);
// dep1's chunks are still present until the third store prunes them — confirm they reached B
// first, so the later disappearance is a real replicated prune rather than never-arrived.
await LocalDbPairHarness.WaitUntilAsync(
async () => await LocalDbPairHarness.ChunkCountAsync(harness.B, dep1) > 0,
"dep1's chunks to reach node B before the prune");
await harness.CacheA.StoreAsync(Cluster, dep3, "rev-3", MultiChunkArtifact(3),
TestContext.Current.CancellationToken);
// A pruned dep1 locally; B must follow via replicated deletes: dep1 gone, tombstones present,
// and the surviving newest (dep3) intact.
await LocalDbPairHarness.WaitUntilAsync(
async () =>
await LocalDbPairHarness.ChunkCountAsync(harness.B, dep1) == 0
&& await LocalDbPairHarness.TombstoneCountAsync(harness.B, DeploymentCacheSchema.ArtifactsTable) > 0
&& await LocalDbPairHarness.ChunkCountAsync(harness.B, dep3) > 0,
"node B to prune dep1 (with tombstones) while keeping dep3");
}
[Fact]
public async Task WritesWhileTransportDown_SurviveRejoin()
{
await using var harness = new LocalDbPairHarness();
await harness.StartAsync();
var dep1 = Guid.NewGuid().ToString("N");
var first = MultiChunkArtifact(seed: 10);
await harness.CacheA.StoreAsync(Cluster, dep1, "rev-1", first,
TestContext.Current.CancellationToken);
await LocalDbPairHarness.WaitUntilAsync(
async () => await ReadArtifactAsync(harness.CacheB, Cluster) is { } b && b.AsSpan().SequenceEqual(first),
"the first artifact to converge before the outage");
// B goes offline; A keeps deploying. B's database survives the host teardown (pre-constructed
// instance), so the rejoin is genuine rather than a fresh node.
await harness.StopPassiveAsync();
var dep2 = Guid.NewGuid().ToString("N");
var second = MultiChunkArtifact(seed: 11);
await harness.CacheA.StoreAsync(Cluster, dep2, "rev-2", second,
TestContext.Current.CancellationToken);
await harness.RestartPairAsync();
// Everything written during the outage catches up: B now serves the newer deployment.
await LocalDbPairHarness.WaitUntilAsync(
async () =>
{
var cached = await harness.CacheB.GetCurrentAsync(Cluster, TestContext.Current.CancellationToken);
return cached is not null
&& cached.DeploymentId == dep2
&& cached.Artifact.AsSpan().SequenceEqual(second);
},
"node B to catch up on the deployment written while it was offline");
}
[Fact]
public async Task WrongApiKey_NeverConverges_WithMatchingKeyPositiveControl()
{
// Negative half: A and B hold different keys, so B's interceptor fail-closes on A's stream.
await using (var mismatched = new LocalDbPairHarness(
apiKeyA: LocalDbPairHarness.DefaultApiKey, apiKeyB: "a-different-key-entirely"))
{
await mismatched.StartAsync();
var deploymentId = Guid.NewGuid().ToString("N");
await mismatched.CacheA.StoreAsync(Cluster, deploymentId, "rev-1", MultiChunkArtifact(20),
TestContext.Current.CancellationToken);
var converged = await LocalDbPairHarness.HeldWithinAsync(
async () => await ReadArtifactAsync(mismatched.CacheB, Cluster) is not null,
TimeSpan.FromSeconds(5));
Assert.False(converged,
"a key mismatch must stop the pair converging — the interceptor is fail-closed");
}
// Positive control: the SAME scenario with matching keys DOES converge. Without this, the
// negative assertion above would pass even if convergence were broken for an unrelated reason.
await using var matched = new LocalDbPairHarness();
await matched.StartAsync();
var controlId = Guid.NewGuid().ToString("N");
var controlArtifact = MultiChunkArtifact(seed: 21);
await matched.CacheA.StoreAsync(Cluster, controlId, "rev-1", controlArtifact,
TestContext.Current.CancellationToken);
await LocalDbPairHarness.WaitUntilAsync(
async () => await ReadArtifactAsync(matched.CacheB, Cluster) is { } b && b.AsSpan().SequenceEqual(controlArtifact),
"the matching-key control pair to converge");
}
}
tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDb/LocalDbPairHarness.cs
+324
View File
@@ -0,0 +1,324 @@
using System.Net;
using Microsoft.AspNetCore.Builder;
using Microsoft.AspNetCore.Hosting;
using Microsoft.AspNetCore.Hosting.Server;
using Microsoft.AspNetCore.Hosting.Server.Features;
using Microsoft.AspNetCore.Server.Kestrel.Core;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging.Abstractions;
using Xunit;
using ZB.MOM.WW.LocalDb;
using ZB.MOM.WW.LocalDb.Replication;
using ZB.MOM.WW.OtOpcUa.Host.Configuration;
using ZB.MOM.WW.OtOpcUa.Runtime.DeploymentCache;
namespace ZB.MOM.WW.OtOpcUa.Host.IntegrationTests.LocalDb;
/// <summary>
/// Serializes the two-node convergence tests against one another: each stands up a real Kestrel
/// h2c listener plus two SQLite files, and running them concurrently under CI contention is a
/// flakiness risk. The rest of the assembly parallelizes normally.
/// </summary>
[CollectionDefinition("LocalDbPairConvergence")]
public sealed class LocalDbPairConvergenceCollection;
/// <summary>
/// Two driver-role OtOpcUa nodes replicating the deployment-artifact cache over a REAL loopback
/// gRPC transport (Kestrel h2c on 127.0.0.1), through the REAL fail-closed
/// <see cref="LocalDbSyncAuthInterceptor"/>. Node A is the initiator (it dials the peer); node B
/// is passive (it hosts <c>MapZbLocalDbSync</c>).
/// </summary>
/// <remarks>
/// <para>
/// Both databases are initialised through the production <see cref="LocalDbSetup.OnReady"/> —
/// a hand-written schema here would prove only that the test agrees with itself. The tables,
/// their primary keys, and the DDL→<c>RegisterReplicated</c> ordering under test are the ones
/// the host actually runs.
/// </para>
/// <para>
/// The two <see cref="ILocalDb"/> instances are owned by the harness and registered into the
/// hosts as pre-constructed singletons. MS.DI does not dispose instances it did not create,
/// so tearing a host down (the transport-loss scenario) leaves the databases intact and
/// writable — which is exactly what lets node A accumulate writes while B is down.
/// </para>
/// <para>
/// A real loopback socket (not an in-memory TestServer) is deliberate: killing the passive
/// host disposes the server and closes the socket, faulting the initiator's active stream
/// promptly. An in-memory TestServer does not model a connection drop and leaves the client
/// hanging.
/// </para>
/// <para>
/// The API keys are per-node so the wrong-key scenario can hand A and B different keys and
/// assert non-convergence — with a matching-key positive control proving the same harness
/// does converge when the keys agree (an absence assertion without a positive control passed
/// vacuously in ScadaBridge).
/// </para>
/// <para>Offline: no docker, no external services.</para>
/// </remarks>
public sealed class LocalDbPairHarness : IAsyncDisposable
{
/// <summary>The key both nodes share unless a test overrides one of them.</summary>
public const string DefaultApiKey = "otopcua-localdb-pair-convergence-key";
/// <summary>How long a scenario waits for the pair to agree (or to stay diverged) before deciding.</summary>
public static readonly TimeSpan ConvergeTimeout = TimeSpan.FromSeconds(30);
private readonly string _apiKeyA;
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");
private readonly ServiceProvider _dbProviderA;
private readonly ServiceProvider _dbProviderB;
private IHost? _serverHost; // node B — passive
private IHost? _initiatorHost; // node A — dials the peer
static LocalDbPairHarness() =>
// Grpc.Net.Client dials the loopback server over HTTP/2 cleartext (h2c).
AppContext.SetSwitch("System.Net.Http.SocketsHttpHandler.Http2UnencryptedSupport", true);
/// <param name="apiKeyA">Node A's replication key. Defaults to <see cref="DefaultApiKey"/>.</param>
/// <param name="apiKeyB">Node B's replication key. Defaults to <see cref="DefaultApiKey"/>.</param>
public LocalDbPairHarness(string? apiKeyA = null, string? apiKeyB = null)
{
_apiKeyA = apiKeyA ?? DefaultApiKey;
_apiKeyB = apiKeyB ?? DefaultApiKey;
_dbProviderA = BuildDatabaseProvider(_pathA);
_dbProviderB = BuildDatabaseProvider(_pathB);
}
/// <summary>Node A — the initiator, which dials the peer.</summary>
public ILocalDb A => _dbProviderA.GetRequiredService<ILocalDb>();
/// <summary>Node B — the passive node, which listens.</summary>
public ILocalDb B => _dbProviderB.GetRequiredService<ILocalDb>();
/// <summary>The production artifact cache over node A's database.</summary>
public IDeploymentArtifactCache CacheA =>
new LocalDbDeploymentArtifactCache(A, NullLogger<LocalDbDeploymentArtifactCache>.Instance);
/// <summary>The production artifact cache over node B's database.</summary>
public IDeploymentArtifactCache CacheB =>
new LocalDbDeploymentArtifactCache(B, NullLogger<LocalDbDeploymentArtifactCache>.Instance);
// ---- lifecycle --------------------------------------------------------------------------
/// <summary>Forces both databases to construct (running OnReady) then brings the pair online.</summary>
public async Task StartAsync()
{
_ = A;
_ = B;
await StartPassiveAsync();
await StartInitiatorAsync();
}
/// <summary>Takes node B's listener down, leaving its database intact and writable.</summary>
public async Task StopPassiveAsync()
{
await StopHostAsync(_serverHost);
_serverHost = null;
}
/// <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.
/// </summary>
public async Task RestartPairAsync()
{
await StartPassiveAsync();
await StopHostAsync(_initiatorHost);
_initiatorHost = null;
await StartInitiatorAsync();
}
// ---- convergence helpers ----------------------------------------------------------------
/// <summary>Polls <paramref name="condition"/> until true or the deadline passes; fails otherwise.</summary>
public static async Task WaitUntilAsync(Func<Task<bool>> condition, string because)
{
var deadline = DateTime.UtcNow + ConvergeTimeout;
while (DateTime.UtcNow < deadline)
{
if (await condition())
return;
await Task.Delay(50);
}
Assert.Fail($"Timed out after {ConvergeTimeout.TotalSeconds:0}s waiting for: {because}");
}
/// <summary>Waits out a bounded window and returns whether <paramref name="condition"/> ever held.</summary>
/// <remarks>
/// For the negative half of the wrong-key scenario: it must NOT converge. A short, fixed
/// window keeps the test quick while still giving a matching-key control ample time to
/// converge (the control uses <see cref="WaitUntilAsync"/> with the full timeout).
/// </remarks>
public static async Task<bool> HeldWithinAsync(Func<Task<bool>> condition, TimeSpan window)
{
var deadline = DateTime.UtcNow + window;
while (DateTime.UtcNow < deadline)
{
if (await condition())
return true;
await Task.Delay(50);
}
return false;
}
/// <summary>
/// Dumps the <c>__localdb_row_version</c> rows for one table, ordered by pk, as
/// <c>pk_json|hlc|node_id|is_tombstone</c>. Equal dumps on both nodes prove B holds A's
/// origin-stamped row (same HLC + node id), not a locally re-derived one.
/// </summary>
public static async Task<string> DumpRowVersionAsync(ILocalDb db, string table)
{
var rows = await db.QueryAsync(
"""
SELECT pk_json, hlc, node_id, is_tombstone
FROM __localdb_row_version
WHERE table_name = @Table
ORDER BY pk_json
""",
static r => $"{r.GetString(0)}|{r.GetInt64(1)}|{r.GetString(2)}|{r.GetInt64(3)}",
new { Table = table });
return string.Join("\n", rows);
}
/// <summary>Counts <c>deployment_artifacts</c> chunk rows for one deployment on a node.</summary>
public static async Task<long> ChunkCountAsync(ILocalDb db, string deploymentId)
{
var rows = await db.QueryAsync(
"SELECT COUNT(*) FROM deployment_artifacts WHERE deployment_id = @DeploymentId",
static r => r.GetInt64(0),
new { DeploymentId = deploymentId });
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)
{
var rows = await db.QueryAsync(
"""
SELECT COUNT(*) FROM __localdb_row_version
WHERE table_name = @Table AND is_tombstone = 1
""",
static r => r.GetInt64(0),
new { Table = table });
return rows[0];
}
// ---- fixture internals ------------------------------------------------------------------
/// <summary>
/// A provider owning one deployment-cache database, initialised through the host's own
/// <see cref="LocalDbSetup.OnReady"/> — same schema, same registration order.
/// </summary>
private static ServiceProvider BuildDatabaseProvider(string path)
{
var config = new ConfigurationBuilder()
.AddInMemoryCollection(new Dictionary<string, string?> { ["LocalDb:Path"] = path })
.Build();
return new ServiceCollection()
.AddZbLocalDb(config, LocalDbSetup.OnReady)
.BuildServiceProvider();
}
private IConfiguration ReplicationConfig(string apiKey, string? peerAddress)
{
var values = new Dictionary<string, string?>
{
// Tight flush + bounded reconnect backoff so convergence is observable well inside the
// poll deadline. The 60 s production default would let the doubling backoff overrun it
// after a peer outage.
["LocalDb:Replication:FlushInterval"] = "00:00:00.050",
["LocalDb:Replication:ReconnectBackoffMax"] = "00:00:02",
["LocalDb:Replication:ApiKey"] = apiKey,
};
if (peerAddress is not null)
values["LocalDb:Replication:PeerAddress"] = peerAddress;
return new ConfigurationBuilder().AddInMemoryCollection(values).Build();
}
/// <summary>Starts node B, the passive listener, behind the real auth interceptor.</summary>
private async Task StartPassiveAsync()
{
var config = ReplicationConfig(_apiKeyB, peerAddress: null);
_serverHost = await new HostBuilder()
.ConfigureWebHost(web =>
{
web.UseKestrel(o =>
o.Listen(IPAddress.Loopback, 0, listen => listen.Protocols = HttpProtocols.Http2));
web.ConfigureServices(services =>
{
services.AddLogging();
services.AddRouting();
// The REAL interceptor, not a stand-in. If it rejected legitimate peer traffic,
// every matching-key scenario would fail — which is the point.
services.AddGrpc(o => o.Interceptors.Add<LocalDbSyncAuthInterceptor>());
services.AddSingleton(B);
services.AddZbLocalDbReplication(config);
})
.Configure(app =>
{
app.UseRouting();
app.UseEndpoints(e => e.MapZbLocalDbSync());
});
})
.StartAsync();
}
/// <summary>Starts node A, which dials the passive node.</summary>
private async Task StartInitiatorAsync()
{
var config = ReplicationConfig(_apiKeyA, PassiveAddress());
_initiatorHost = await new HostBuilder()
.ConfigureServices(services =>
{
services.AddLogging();
services.AddSingleton(A);
services.AddZbLocalDbReplication(config);
})
.StartAsync();
}
private string PassiveAddress()
=> _serverHost!.Services.GetRequiredService<IServer>()
.Features.Get<IServerAddressesFeature>()!.Addresses.Single();
private static async Task StopHostAsync(IHost? host)
{
if (host is null)
return;
try { await host.StopAsync(TimeSpan.FromSeconds(5)); } catch { /* teardown */ }
host.Dispose();
}
public async ValueTask DisposeAsync()
{
await StopHostAsync(_initiatorHost);
await StopHostAsync(_serverHost);
await _dbProviderA.DisposeAsync();
await _dbProviderB.DisposeAsync();
Microsoft.Data.Sqlite.SqliteConnection.ClearAllPools();
foreach (var path in new[] { _pathA, _pathB })
{
foreach (var suffix in new[] { "", "-wal", "-shm" })
{
try { File.Delete(path + suffix); } catch { /* best effort */ }
}
}
}
}