diff --git a/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/LocalDbSitePairConvergenceTests.cs b/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/LocalDbSitePairConvergenceTests.cs new file mode 100644 index 00000000..b9c13ced --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/LocalDbSitePairConvergenceTests.cs @@ -0,0 +1,348 @@ +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 ZB.MOM.WW.LocalDb; +using ZB.MOM.WW.LocalDb.Replication; +using ZB.MOM.WW.ScadaBridge.Host; + +namespace ZB.MOM.WW.ScadaBridge.IntegrationTests; + +/// +/// Serializes the site-pair convergence tests against each other: each one stands up a real +/// Kestrel listener plus two SQLite files, and running them concurrently under CI +/// contention is a flakiness risk. +/// +[CollectionDefinition("LocalDbSitePairConvergence")] +public sealed class LocalDbSitePairConvergenceCollection; + +/// +/// Two ScadaBridge site nodes replicating the consolidated site database over a REAL +/// loopback gRPC transport, through the REAL fail-closed auth interceptor. +/// +/// +/// +/// This is the test that answers the question Phase 1 exists to answer: does a site node +/// pair actually stop losing operation-tracking and site-event state? Everything upstream +/// of it — schema helpers, DI wiring, the interceptor — can be individually green while the +/// pair still fails to converge. +/// +/// +/// It uses , not a hand-written schema, so the tables, +/// their primary keys, and the registration ORDER under test are the ones the host actually +/// runs. A separate schema here would prove only that the test agrees with itself. +/// +/// +/// Offline: no docker, no external services. Loopback Kestrel with h2c. +/// +/// +[Collection("LocalDbSitePairConvergence")] +public sealed class LocalDbSitePairConvergenceTests : IAsyncLifetime +{ + private const string SharedApiKey = "site-pair-convergence-key"; + private static readonly TimeSpan ConvergeTimeout = TimeSpan.FromSeconds(30); + + private readonly string _pathA = Path.Combine(Path.GetTempPath(), $"sitepairA-{Guid.NewGuid():N}.db"); + private readonly string _pathB = Path.Combine(Path.GetTempPath(), $"sitepairB-{Guid.NewGuid():N}.db"); + + // The databases are owned by the fixture, in their own providers, and registered into the + // hosts as pre-constructed instances. MS.DI does not dispose instances it did not create, + // so tearing a host down (the offline-peer scenario) leaves the databases intact and + // writable — which is exactly what lets node A accumulate writes while B is down. + private ServiceProvider _dbProviderA = null!; + private ServiceProvider _dbProviderB = null!; + + private IHost? _serverHost; // node B — passive + private IHost? _initiatorHost; // node A — dials the peer + + static LocalDbSitePairConvergenceTests() => + // Grpc.Net.Client dials the loopback server over HTTP/2 cleartext. + AppContext.SetSwitch("System.Net.Http.SocketsHttpHandler.Http2UnencryptedSupport", true); + + private ILocalDb A => _dbProviderA.GetRequiredService(); + private ILocalDb B => _dbProviderB.GetRequiredService(); + + public async Task InitializeAsync() + { + _dbProviderA = BuildDatabaseProvider(_pathA, "node-a"); + _dbProviderB = BuildDatabaseProvider(_pathB, "node-b"); + + // Force construction (and therefore OnReady) before anything replicates. + _ = A; + _ = B; + + await StartPassiveAsync(); + await StartInitiatorAsync(); + } + + public async Task 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 */ } + } + } + } + + // ---- fixture internals ------------------------------------------------------------ + + /// + /// A provider owning one consolidated site database, initialized through the host's own + /// — same schema, same registration order. + /// + private static ServiceProvider BuildDatabaseProvider(string path, string nodeName) + { + var config = new ConfigurationBuilder() + .AddInMemoryCollection(new Dictionary + { + ["LocalDb:Path"] = path, + ["ScadaBridge:Node:NodeName"] = nodeName, + // Point the legacy migrator at paths that do not exist, so it no-ops rather + // than picking up stray files from the test working directory. + ["ScadaBridge:OperationTracking:ConnectionString"] = + $"Data Source={Path.Combine(Path.GetTempPath(), $"absent-{Guid.NewGuid():N}.db")}", + ["ScadaBridge:SiteEventLog:DatabasePath"] = + Path.Combine(Path.GetTempPath(), $"absent-{Guid.NewGuid():N}.db"), + }) + .Build(); + + return new ServiceCollection() + .AddZbLocalDb(config, db => SiteLocalDbSetup.OnReady(db, config)) + .BuildServiceProvider(); + } + + private static IConfiguration ReplicationConfig(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", + // Both nodes share one key — the interceptor is fail-closed, so a mismatch here + // turns every scenario below red (verified by deliberately breaking it). + ["LocalDb:Replication:ApiKey"] = SharedApiKey, + }; + + if (peerAddress is not null) + values["LocalDb:Replication:PeerAddress"] = peerAddress; + + return new ConfigurationBuilder().AddInMemoryCollection(values).Build(); + } + + private async Task StartPassiveAsync() + { + var config = ReplicationConfig(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 scenario below would fail — which is the point. + services.AddGrpc(o => o.Interceptors.Add()); + services.AddSingleton(B); + services.AddZbLocalDbReplication(config); + }); + web.Configure(app => + { + app.UseRouting(); + app.UseEndpoints(e => e.MapZbLocalDbSync()); + }); + }) + .StartAsync(); + } + + private async Task StartInitiatorAsync() + { + var config = ReplicationConfig(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(); + } + + // ---- data helpers ----------------------------------------------------------------- + + private static Task WriteTrackingAsync(ILocalDb db, string id, string status, string target) + => db.ExecuteAsync( + """ + INSERT INTO OperationTracking ( + TrackedOperationId, Kind, TargetSummary, Status, RetryCount, + CreatedAtUtc, UpdatedAtUtc) + VALUES (@id, 'ApiCallCached', @target, @status, 0, @now, @now) + ON CONFLICT(TrackedOperationId) DO UPDATE SET + Status = excluded.Status, + TargetSummary = excluded.TargetSummary, + UpdatedAtUtc = excluded.UpdatedAtUtc; + """, + new { id, status, target, now = DateTime.UtcNow.ToString("o") }); + + private static Task WriteEventAsync(ILocalDb db, string id, string message) + => db.ExecuteAsync( + """ + INSERT INTO site_events (id, timestamp, event_type, severity, source, message) + VALUES (@id, @ts, 'script', 'Info', 'convergence-test', @message); + """, + new { id, ts = DateTimeOffset.UtcNow.ToString("o"), message }); + + private static async Task ReadTrackingStatusAsync(ILocalDb db, string id) + { + var rows = await db.QueryAsync( + "SELECT Status FROM OperationTracking WHERE TrackedOperationId = @id", + static r => r.GetString(0), new { id }); + return rows.Count == 0 ? null : rows[0]; + } + + private static Task> ReadEventIdsAsync(ILocalDb db) + => db.QueryAsync("SELECT id FROM site_events ORDER BY id", static r => r.GetString(0)); + + /// Polls until true or the deadline passes. + private 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}"); + } + + // ---- scenarios -------------------------------------------------------------------- + + [Fact] + public async Task TrackingRow_WrittenOnA_BecomesReadableOnB() + { + await WriteTrackingAsync(A, "op-a-1", "Submitted", "ERP.GetOrder"); + + await WaitUntilAsync( + async () => await ReadTrackingStatusAsync(B, "op-a-1") == "Submitted", + "the tracking row written on node A to appear on node B"); + } + + [Fact] + public async Task TrackingRow_WrittenOnB_BecomesReadableOnA() + { + // Replication is bidirectional even though only A dials: proving the passive node's + // writes flow back is what makes a failover in EITHER direction safe. + await WriteTrackingAsync(B, "op-b-1", "Submitted", "ERP.GetOrder"); + + await WaitUntilAsync( + async () => await ReadTrackingStatusAsync(A, "op-b-1") == "Submitted", + "the tracking row written on node B to appear on node A"); + } + + [Fact] + public async Task SameOperation_UpdatedOnBothNodes_ConvergesToOneWinner() + { + // Last-writer-wins on the primary key. The specific winner is not asserted — that is + // the HLC's business — but the two nodes MUST agree, and must agree on a value one of + // them actually wrote rather than a merge of both. + await WriteTrackingAsync(A, "op-conflict", "Submitted", "ERP.GetOrder"); + await WaitUntilAsync( + async () => await ReadTrackingStatusAsync(B, "op-conflict") is not null, + "the conflicting row to exist on both nodes before it is updated"); + + await WriteTrackingAsync(A, "op-conflict", "Delivered", "ERP.GetOrder"); + await WriteTrackingAsync(B, "op-conflict", "Parked", "ERP.GetOrder"); + + await WaitUntilAsync( + async () => + { + var a = await ReadTrackingStatusAsync(A, "op-conflict"); + var b = await ReadTrackingStatusAsync(B, "op-conflict"); + return a is not null && a == b; + }, + "both nodes to converge on one status for the contended operation"); + + var winner = await ReadTrackingStatusAsync(A, "op-conflict"); + Assert.Contains(winner, new[] { "Delivered", "Parked" }); + } + + [Fact] + public async Task EventsLoggedOnBothNodes_ConvergeToTheUnion_WithNoIdCollisions() + { + // The reason site_events moved to GUID ids. Under the old autoincrement scheme both + // nodes would independently mint id=1, id=2, ... and last-writer-wins on the primary + // key would silently DESTROY one node's events instead of merging them. The union + // count is the assertion that catches that. + var idsFromA = Enumerable.Range(0, 5).Select(_ => Guid.NewGuid().ToString("N")).ToList(); + var idsFromB = Enumerable.Range(0, 5).Select(_ => Guid.NewGuid().ToString("N")).ToList(); + + foreach (var id in idsFromA) await WriteEventAsync(A, id, "from A"); + foreach (var id in idsFromB) await WriteEventAsync(B, id, "from B"); + + var expected = idsFromA.Concat(idsFromB).OrderBy(x => x, StringComparer.Ordinal).ToList(); + + await WaitUntilAsync( + async () => (await ReadEventIdsAsync(A)).SequenceEqual(expected) + && (await ReadEventIdsAsync(B)).SequenceEqual(expected), + "both nodes to hold the union of all 10 events"); + } + + [Fact] + public async Task PeerOffline_ThenRejoins_CatchesUpOnEverythingItMissed() + { + // The failover case that motivates Phase 1: one node is down while the other keeps + // working, and nothing written during the outage may be lost. + await StopHostAsync(_serverHost); + _serverHost = null; + + var idsDuringOutage = Enumerable.Range(0, 5).Select(_ => Guid.NewGuid().ToString("N")).ToList(); + foreach (var id in idsDuringOutage) await WriteEventAsync(A, id, "written while B was down"); + await WriteTrackingAsync(A, "op-during-outage", "Delivered", "ERP.GetOrder"); + + // Node B's database survived the host teardown (pre-constructed instance), so this is + // a genuine rejoin rather than a fresh node. It comes back on a NEW loopback port; + // the initiator's channel factory re-reads the peer address on each reconnect. + await StartPassiveAsync(); + await StopHostAsync(_initiatorHost); + await StartInitiatorAsync(); + + await WaitUntilAsync( + async () => + { + var events = await ReadEventIdsAsync(B); + return idsDuringOutage.All(events.Contains) + && await ReadTrackingStatusAsync(B, "op-during-outage") == "Delivered"; + }, + "node B to catch up on everything written while it was offline"); + } +}