diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDb/LocalDbPairConvergenceTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDb/LocalDbPairConvergenceTests.cs new file mode 100644 index 00000000..bfa317d3 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDb/LocalDbPairConvergenceTests.cs @@ -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; + +/// +/// 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. +/// +/// +/// 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. +/// +[Collection("LocalDbPairConvergence")] +public sealed class LocalDbPairConvergenceTests +{ + private const string Cluster = "cluster-1"; + + /// + /// 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. + /// + 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 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"); + } +} 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 new file mode 100644 index 00000000..d2815590 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDb/LocalDbPairHarness.cs @@ -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; + +/// +/// 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. +/// +[CollectionDefinition("LocalDbPairConvergence")] +public sealed class LocalDbPairConvergenceCollection; + +/// +/// 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 +/// . Node A is the initiator (it dials the peer); node B +/// is passive (it hosts MapZbLocalDbSync). +/// +/// +/// +/// Both databases are initialised through the production — +/// a hand-written schema here would prove only that the test agrees with itself. The tables, +/// their primary keys, and the DDL→RegisterReplicated ordering under test are the ones +/// the host actually runs. +/// +/// +/// The two 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. +/// +/// +/// 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. +/// +/// +/// 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). +/// +/// Offline: no docker, no external services. +/// +public sealed class LocalDbPairHarness : IAsyncDisposable +{ + /// The key both nodes share unless a test overrides one of them. + public const string DefaultApiKey = "otopcua-localdb-pair-convergence-key"; + + /// How long a scenario waits for the pair to agree (or to stay diverged) before deciding. + 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); + + /// Node A's replication key. Defaults to . + /// Node B's replication key. Defaults to . + public LocalDbPairHarness(string? apiKeyA = null, string? apiKeyB = null) + { + _apiKeyA = apiKeyA ?? DefaultApiKey; + _apiKeyB = apiKeyB ?? DefaultApiKey; + + _dbProviderA = BuildDatabaseProvider(_pathA); + _dbProviderB = BuildDatabaseProvider(_pathB); + } + + /// Node A — the initiator, which dials the peer. + public ILocalDb A => _dbProviderA.GetRequiredService(); + + /// Node B — the passive node, which listens. + public ILocalDb B => _dbProviderB.GetRequiredService(); + + /// The production artifact cache over node A's database. + public IDeploymentArtifactCache CacheA => + new LocalDbDeploymentArtifactCache(A, NullLogger.Instance); + + /// The production artifact cache over node B's database. + public IDeploymentArtifactCache CacheB => + new LocalDbDeploymentArtifactCache(B, NullLogger.Instance); + + // ---- lifecycle -------------------------------------------------------------------------- + + /// Forces both databases to construct (running OnReady) then brings the pair online. + public async Task StartAsync() + { + _ = A; + _ = B; + + await StartPassiveAsync(); + await StartInitiatorAsync(); + } + + /// Takes node B's listener down, leaving its database intact and writable. + public async Task StopPassiveAsync() + { + await StopHostAsync(_serverHost); + _serverHost = null; + } + + /// + /// 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. + /// + public async Task RestartPairAsync() + { + await StartPassiveAsync(); + await StopHostAsync(_initiatorHost); + _initiatorHost = null; + await StartInitiatorAsync(); + } + + // ---- convergence helpers ---------------------------------------------------------------- + + /// Polls until true or the deadline passes; fails otherwise. + public static async Task WaitUntilAsync(Func> 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}"); + } + + /// Waits out a bounded window and returns whether ever held. + /// + /// 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 with the full timeout). + /// + public static async Task HeldWithinAsync(Func> condition, TimeSpan window) + { + var deadline = DateTime.UtcNow + window; + while (DateTime.UtcNow < deadline) + { + if (await condition()) + return true; + await Task.Delay(50); + } + + return false; + } + + /// + /// Dumps the __localdb_row_version rows for one table, ordered by pk, as + /// pk_json|hlc|node_id|is_tombstone. Equal dumps on both nodes prove B holds A's + /// origin-stamped row (same HLC + node id), not a locally re-derived one. + /// + public static async Task 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); + } + + /// Counts deployment_artifacts chunk rows for one deployment on a node. + public static async Task 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]; + } + + /// Counts tombstone rows in __localdb_row_version for one table on a node. + public static async Task 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 ------------------------------------------------------------------ + + /// + /// A provider owning one deployment-cache database, initialised through the host's own + /// — same schema, same registration order. + /// + private static ServiceProvider BuildDatabaseProvider(string path) + { + var config = new ConfigurationBuilder() + .AddInMemoryCollection(new Dictionary { ["LocalDb:Path"] = path }) + .Build(); + + return new ServiceCollection() + .AddZbLocalDb(config, LocalDbSetup.OnReady) + .BuildServiceProvider(); + } + + private IConfiguration ReplicationConfig(string apiKey, string? peerAddress) + { + var values = new Dictionary + { + // 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(); + } + + /// Starts node B, the passive listener, behind the real auth interceptor. + 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()); + services.AddSingleton(B); + services.AddZbLocalDbReplication(config); + }) + .Configure(app => + { + app.UseRouting(); + app.UseEndpoints(e => e.MapZbLocalDbSync()); + }); + }) + .StartAsync(); + } + + /// Starts node A, which dials the passive node. + 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() + .Features.Get()!.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 */ } + } + } + } +}