diff --git a/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/Convergence/BidirectionalConvergenceTests.cs b/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/Convergence/BidirectionalConvergenceTests.cs new file mode 100644 index 0000000..8661bec --- /dev/null +++ b/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/Convergence/BidirectionalConvergenceTests.cs @@ -0,0 +1,188 @@ +namespace ZB.MOM.WW.LocalDb.Tests.Convergence; + +/// +/// End-to-end bidirectional convergence over a REAL gRPC TestHost (see ): +/// concurrent writes, an update/delete race, offline accumulation, restart-mid-stream, and a pruned-horizon +/// snapshot resync — each driven through the normal ILocalDb API while both sync sessions run. +/// +public sealed class BidirectionalConvergenceTests +{ + [Fact] + public async Task ConcurrentWrites_BothSides_Converge() + { + await using var fx = new ConvergenceFixture(); + await fx.StartAsync(); + + // Interleave writes on both nodes while the sessions run; disjoint key ranges converge to + // the union. + for (var i = 1; i <= 25; i++) + { + await ConvergenceFixture.UpsertAsync(fx.A, i, "A", i); + await ConvergenceFixture.UpsertAsync(fx.B, 1000 + i, "B", i); + } + + // Shared key: A writes it first and lets it converge, THEN B overwrites it. Because B's HLC + // clock observed A's replicated stamp, B's write is provably the higher HLC -> the LWW winner + // is deterministically B on both nodes (avoids an HLC-tie decided by random node-id order). + await ConvergenceFixture.UpsertAsync(fx.A, 100, "A_FIRST", 1); + await fx.AssertConvergedAsync(); + await ConvergenceFixture.UpsertAsync(fx.B, 100, "B_WINS", 2); + + await fx.AssertConvergedAsync(); + + var a = await ReadRow(fx.A, 100); + var b = await ReadRow(fx.B, 100); + Assert.Equal("B_WINS", a.Sku); + Assert.Equal(a, b); + Assert.Equal(51, await ConvergenceFixture.CountOrdersAsync(fx.A)); + Assert.Equal(51, await ConvergenceFixture.CountOrdersAsync(fx.B)); + } + + [Fact] + public async Task UpdateDeleteRace_Converges() + { + await using var fx = new ConvergenceFixture(); + + // Seed the contested key on both nodes and let it converge first. + await ConvergenceFixture.UpsertAsync(fx.A, 7, "SEED", 1); + await fx.StartAsync(); + await fx.AssertConvergedAsync(); + + // Race: A updates the same pk while B deletes it, inside one window. Whichever HLC wins, both + // nodes must end identical (either the updated row on both, or absent on both). + await ConvergenceFixture.UpsertAsync(fx.A, 7, "A_UPDATED", 99); + await ConvergenceFixture.DeleteAsync(fx.B, 7); + + await fx.AssertConvergedAsync(); + + var a = await ConvergenceFixture.CountOrdersAsync(fx.A); + var b = await ConvergenceFixture.CountOrdersAsync(fx.B); + Assert.Equal(a, b); // identical outcome; AssertConverged already proved row-level identity + } + + [Fact] + public async Task OfflineAccumulation_Reconnect_Converges() + { + await using var fx = new ConvergenceFixture(); + await fx.StartAsync(); + await fx.AssertConvergedAsync(); + + await fx.KillTransportAsync(); + + // Both nodes keep taking local writes while the transport is down. + for (var i = 1; i <= 20; i++) + { + await ConvergenceFixture.UpsertAsync(fx.A, i, "A_OFFLINE", i); + await ConvergenceFixture.UpsertAsync(fx.B, 500 + i, "B_OFFLINE", i); + } + + await fx.RestartTransportAsync(); + await fx.AssertConvergedAsync(); + + Assert.Equal(40, await ConvergenceFixture.CountOrdersAsync(fx.A)); + Assert.Equal(40, await ConvergenceFixture.CountOrdersAsync(fx.B)); + } + + [Fact] + public async Task RestartMidStream_NoLoss_NoDuplication() + { + const int rows = 400; + // Small batches so a large transfer is genuinely in-flight when the transport dies mid-stream. + await using var fx = new ConvergenceFixture( + tuneInitiator: d => d["LocalDb:Replication:MaxBatchSize"] = "10"); + + for (var i = 1; i <= rows; i++) + await ConvergenceFixture.UpsertAsync(fx.A, i, "R", i); + + await fx.StartAsync(); + + // Sever the transport once the transfer is under way (first rows landed on B). In-process + // gRPC is fast, so the kill may land anywhere in the transfer — the point is that a + // kill+restart around a large multi-batch transfer neither loses nor duplicates rows. + await WaitUntilAsync(async () => await ConvergenceFixture.CountOrdersAsync(fx.B) >= 1, + TimeSpan.FromSeconds(10)); + await fx.KillTransportAsync(); + + await fx.RestartTransportAsync(); + await fx.AssertConvergedAsync(); + + // Exact count on both sides: no dupes (INSERT OR REPLACE + LWW idempotency) and no loss. + Assert.Equal(rows, await ConvergenceFixture.CountOrdersAsync(fx.A)); + Assert.Equal(rows, await ConvergenceFixture.CountOrdersAsync(fx.B)); + } + + [Fact] + public async Task PruneDuringOutage_SnapshotRecovers_E2E() + { + // Tiny backlog cap on A so its oplog prunes below B's watermark during the outage, forcing a + // full-snapshot resync on reconnect rather than incremental deltas. + await using var fx = new ConvergenceFixture( + tuneInitiator: d => d["LocalDb:Replication:MaxOplogRows"] = "3"); + await fx.StartAsync(); + await fx.AssertConvergedAsync(); + + await fx.KillTransportAsync(); + + // Accumulate well past the cap while B is offline. + for (var i = 1; i <= 20; i++) + await ConvergenceFixture.UpsertAsync(fx.A, i, "PRUNED", i); + Assert.True(await fx.EnforceCapsAAsync(), "expected a prune + needs_snapshot flag"); + + await fx.RestartTransportAsync(); + // Snapshot delivers all 20 rows; A's snapshot-covered tail oplog is not delta-acked yet, so + // assert DATA convergence here (not oplog-empty). + await fx.AssertDataConvergedAsync(); + Assert.Equal(20, await ConvergenceFixture.CountOrdersAsync(fx.B)); + + // Deltas resume after the snapshot: a fresh write flows incrementally, its ack prunes the + // snapshot-covered tail, and the oplog fully drains -> full convergence. + await ConvergenceFixture.UpsertAsync(fx.A, 21, "POST_SNAP", 21); + await fx.AssertConvergedAsync(); + Assert.Equal(21, await ConvergenceFixture.CountOrdersAsync(fx.B)); + } + + // TODO(product bug): `INSERT ... ON CONFLICT(id) DO UPDATE` on a replicated table crashes the + // AFTER-UPDATE capture trigger `__localdb__au` with + // SQLite Error 19 (1555 SQLITE_CONSTRAINT_PRIMARYKEY): + // 'UNIQUE constraint failed: __localdb_row_version.table_name, __localdb_row_version.pk_json'. + // Minimal repro: on a fresh replicated `orders` table, upsert the SAME pk twice via ON CONFLICT + // DO UPDATE (first = insert, second = the DO-UPDATE branch) — no replication, no concurrency. + // A plain `UPDATE orders SET ...` firing the same au trigger does NOT fail; the defect is + // specific to the upsert form perturbing last_insert_rowid() inside the au trigger's + // `INSERT OR REPLACE INTO __localdb_row_version ... WHERE seq = last_insert_rowid()` capture. + // The convergence tests intentionally avoid the upsert form (see ConvergenceFixture.UpsertAsync). + // Un-skip once the capture trigger supports UPSERT. + [Fact(Skip = "Product bug: ON CONFLICT DO UPDATE crashes the au capture trigger (UNIQUE on __localdb_row_version). See TODO above.")] + public async Task UpsertOnConflictDoUpdate_CrashesCaptureTrigger_ProductBug() + { + await using var fx = new ConvergenceFixture(); + await fx.A.ExecuteAsync( + "INSERT INTO orders (id, sku, qty) VALUES (7, 'FIRST', 1) " + + "ON CONFLICT(id) DO UPDATE SET sku = excluded.sku, qty = excluded.qty"); + // Second upsert takes the DO UPDATE branch and throws the UNIQUE-constraint SqliteException. + await fx.A.ExecuteAsync( + "INSERT INTO orders (id, sku, qty) VALUES (7, 'SECOND', 2) " + + "ON CONFLICT(id) DO UPDATE SET sku = excluded.sku, qty = excluded.qty"); + } + + private static async Task<(long Id, string? Sku, long? Qty)> ReadRow(ILocalDb db, long id) + { + var rows = await db.QueryAsync( + "SELECT id, sku, qty FROM orders WHERE id = @id", + static r => (r.GetInt64(0), r.IsDBNull(1) ? null : r.GetString(1), (long?)(r.IsDBNull(2) ? null : r.GetInt64(2))), + new { id }); + return rows.Single(); + } + + private static async Task WaitUntilAsync(Func> predicate, TimeSpan timeout) + { + var deadline = DateTime.UtcNow + timeout; + while (DateTime.UtcNow < deadline) + { + if (await predicate()) + return; + await Task.Delay(10); + } + throw new TimeoutException("Condition not reached within " + timeout); + } +} diff --git a/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/Convergence/ConvergenceFixture.cs b/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/Convergence/ConvergenceFixture.cs new file mode 100644 index 0000000..d29c3f1 --- /dev/null +++ b/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/Convergence/ConvergenceFixture.cs @@ -0,0 +1,340 @@ +using System.Net; +using Grpc.Net.Client; +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.Options; +using ZB.MOM.WW.LocalDb.Internal; +using ZB.MOM.WW.LocalDb.Replication; +using ZB.MOM.WW.LocalDb.Replication.Internal; + +namespace ZB.MOM.WW.LocalDb.Tests.Convergence; + +/// +/// Two FULL replication stacks over a REAL loopback gRPC transport (Kestrel h2c on 127.0.0.1), +/// proving end-to-end bidirectional convergence. Node A is the initiator (client: +/// dials the peer through an injected channel factory); node B +/// is passive (server: hosts MapZbLocalDbSync). Both build the identical +/// stack (OplogStore + LwwApplier + snapshot hooks) via +/// . +/// +/// A real loopback socket (not an in-memory TestServer) is deliberate: KillTransport disposes the +/// server and CLOSES the socket, which faults the initiator's active stream PROMPTLY — an in-memory +/// TestServer does not model a connection drop and leaves the client hanging for tens of seconds. +/// +/// The two instances are OWNED BY THE FIXTURE and registered into the +/// DI containers as pre-constructed singletons — MS.DI does not dispose externally-`new`ed instances, +/// so a container/host tear-down (KillTransport) leaves the databases intact and writable. That is +/// what lets a test accumulate writes on both nodes while the transport is down. +/// +public sealed class ConvergenceFixture : IAsyncDisposable +{ + private static readonly TimeSpan DefaultConvergeTimeout = TimeSpan.FromSeconds(30); + + private readonly IConfiguration _initiatorConfig; + private readonly IConfiguration _passiveConfig; + private readonly string _pathA; + private readonly string _pathB; + + private readonly SqliteLocalDb _dbA; + private readonly SqliteLocalDb _dbB; + + private ServiceProvider? _clientProvider; + private SyncBackgroundService? _bg; + private IHost? _serverHost; + + // Read fresh by the client's channel factory on every (re)connect: null => transport is down and + // the factory throws, driving the initiator's reconnect backoff until RestartTransport refreshes it. + private volatile string? _serverAddress; + + static ConvergenceFixture() => + // Grpc.Net.Client dials the loopback server over HTTP/2 cleartext (h2c). + AppContext.SetSwitch("System.Net.Http.SocketsHttpHandler.Http2UnencryptedSupport", true); + + private ReplicationOptions InitiatorOptions => + _clientProvider!.GetRequiredService>().Value; + + /// Mutates node A's LocalDb:Replication:* config (e.g. MaxBatchSize, MaxOplogRows). + /// Mutates node B's config. + public ConvergenceFixture( + Action>? tuneInitiator = null, + Action>? tunePassive = null) + { + _pathA = Path.Combine(Path.GetTempPath(), "convA-" + Guid.NewGuid() + ".db"); + _pathB = Path.Combine(Path.GetTempPath(), "convB-" + Guid.NewGuid() + ".db"); + _initiatorConfig = BuildConfig(initiator: true, tuneInitiator); + _passiveConfig = BuildConfig(initiator: false, tunePassive); + + _dbA = NewDb(_pathA); + _dbB = NewDb(_pathB); + } + + /// The initiator (client) node's database. + public ILocalDb A => _dbA; + + /// The passive (server) node's database. + public ILocalDb B => _dbB; + + /// Initiator connection attempts so far (1 = initial dial; each reconnect increments it). + public int InitiatorConnectionAttempts => _bg?.ConnectionAttempts ?? 0; + + // ---- lifecycle controls ----------------------------------------------------------------- + + /// Brings up the passive server, wires the initiator's channel factory to it, and starts the initiator pump. + public async Task StartAsync() + { + (_serverHost, _serverAddress) = await BuildServerHostAsync(); + + var services = new ServiceCollection(); + services.AddLogging(); + // Pre-constructed instance: the container will NOT dispose _dbA (verified MS.DI behavior). + services.AddSingleton(_dbA); + services.AddZbLocalDbReplication(_initiatorConfig); + _clientProvider = services.BuildServiceProvider(); + + _bg = _clientProvider.GetServices().OfType().Single(); + _bg.ChannelFactory = () => + { + var address = _serverAddress + ?? throw new InvalidOperationException("transport down (server disposed)"); + return GrpcChannel.ForAddress(address); + }; + await _bg.StartAsync(CancellationToken.None); + } + + /// Graceful stop of both the initiator pump and the passive server. + public async Task StopAsync() + { + if (_bg is not null) + try { await _bg.StopAsync(CancellationToken.None); } catch { /* teardown */ } + await DisposeServerHostAsync(); + } + + /// Closes the server socket so the initiator's in-flight stream faults (transport loss). The databases survive. + public async Task KillTransportAsync() + { + _serverAddress = null; + await DisposeServerHostAsync(); + } + + /// Stands up a fresh server on a new loopback port over the SAME databases; the initiator reconnects on its next backoff tick. + public async Task RestartTransportAsync() => + (_serverHost, _serverAddress) = await BuildServerHostAsync(); + + // ---- convergence assertion -------------------------------------------------------------- + + /// + /// Polls until both oplogs are fully acked/pruned (empty) AND both nodes' orders + + /// __localdb_row_version dumps (ordered by pk) match; on timeout, dumps both states. + /// + /// Row-version comparison excludes tombstone_utc: that column is a LOCAL retention clock + /// (the delete trigger stamps the originating wall-clock; stamps the + /// receiving node's apply-time), so it diverges by design and is not part of the converged + /// logical state. The convergence-defining columns (hlc, node_id, is_tombstone) are compared. + /// + public Task AssertConvergedAsync(TimeSpan? timeout = null) => + PollUntilConvergedAsync(requireEmptyOplog: true, timeout); + + /// + /// Polls until only the DATA has converged (orders + row_version identical), tolerating a + /// non-empty oplog. Use this to observe convergence immediately after a snapshot resync: the + /// sender's snapshot-covered tail oplog rows are delivered by the snapshot, not delta-acked, so + /// they are not pruned until a SUBSEQUENT delta carries the peer ack past them. + /// + public Task AssertDataConvergedAsync(TimeSpan? timeout = null) => + PollUntilConvergedAsync(requireEmptyOplog: false, timeout); + + private async Task PollUntilConvergedAsync(bool requireEmptyOplog, TimeSpan? timeout) + { + var deadline = DateTime.UtcNow + (timeout ?? DefaultConvergeTimeout); + while (DateTime.UtcNow < deadline) + { + if (await IsConvergedAsync(requireEmptyOplog)) + return; + await Task.Delay(25); + } + + if (await IsConvergedAsync(requireEmptyOplog)) + return; + + Assert.Fail(await DumpStateAsync()); + } + + private async Task IsConvergedAsync(bool requireEmptyOplog) + { + if (requireEmptyOplog && (await OplogCountAsync(_dbA) != 0 || await OplogCountAsync(_dbB) != 0)) + return false; + if (await DumpOrdersAsync(_dbA) != await DumpOrdersAsync(_dbB)) + return false; + return await DumpRowVersionAsync(_dbA) == await DumpRowVersionAsync(_dbB); + } + + private async Task DumpStateAsync() + { + var sb = new System.Text.StringBuilder(); + sb.AppendLine("Convergence NOT reached within timeout."); + sb.AppendLine($"--- A oplog rows: {await OplogCountAsync(_dbA)} B oplog rows: {await OplogCountAsync(_dbB)}"); + sb.AppendLine("--- A.orders ---").AppendLine(await DumpOrdersAsync(_dbA)); + sb.AppendLine("--- B.orders ---").AppendLine(await DumpOrdersAsync(_dbB)); + sb.AppendLine("--- A.row_version ---").AppendLine(await DumpRowVersionAsync(_dbA)); + sb.AppendLine("--- B.row_version ---").AppendLine(await DumpRowVersionAsync(_dbB)); + return sb.ToString(); + } + + // ---- test helpers ----------------------------------------------------------------------- + + /// + /// Inserts or updates one row through the normal consumer API (fires the ai/au capture triggers). + /// Uses a plain UPDATE-else-INSERT inside one transaction — NOT INSERT ... ON CONFLICT DO + /// UPDATE: that upsert form crashes the AFTER-UPDATE capture trigger with + /// "UNIQUE constraint failed: __localdb_row_version" (see the Skipped regression test in + /// BidirectionalConvergenceTests). The transaction makes the update-or-insert atomic against a + /// concurrent replicated apply of the same pk (no INSERT/UNIQUE race on orders). + /// + public static async Task UpsertAsync(ILocalDb db, long id, string sku, long qty) + { + await using var tx = await db.BeginTransactionAsync(); + var updated = await tx.ExecuteAsync( + "UPDATE orders SET sku = @sku, qty = @qty WHERE id = @id", new { id, sku, qty }); + if (updated == 0) + await tx.ExecuteAsync( + "INSERT INTO orders (id, sku, qty) VALUES (@id, @sku, @qty)", new { id, sku, qty }); + await tx.CommitAsync(); + } + + /// Deletes one row through the normal consumer API (fires the delete capture trigger). + public static Task DeleteAsync(ILocalDb db, long id) => + db.ExecuteAsync("DELETE FROM orders WHERE id = @id", new { id }); + + public static async Task CountOrdersAsync(ILocalDb db) + { + var rows = await db.QueryAsync("SELECT COUNT(*) FROM orders", static r => r.GetInt64(0)); + return (int)rows[0]; + } + + /// + /// Enforces the oplog backlog caps on node A (prunes + flags needs_snapshot when over the cap), + /// mirroring SnapshotResyncTests. The running engine has no automatic caps enforcer, so a + /// test that needs a pruned horizon triggers it here on a store over the same database. + /// + public Task EnforceCapsAAsync() => + new OplogStore(_dbA, InitiatorOptions).EnforceCapsAsync(); + + // ---- internals -------------------------------------------------------------------------- + + private static SqliteLocalDb NewDb(string path) + { + var db = new SqliteLocalDb(new LocalDbOptions { Path = path }); + using (var conn = db.CreateConnection()) + using (var cmd = conn.CreateCommand()) + { + cmd.CommandText = "CREATE TABLE IF NOT EXISTS orders (id INTEGER PRIMARY KEY, sku TEXT, qty INTEGER)"; + cmd.ExecuteNonQuery(); + } + db.RegisterReplicated("orders"); + return db; + } + + private async Task<(IHost Host, string Address)> BuildServerHostAsync() + { + var host = await new HostBuilder() + .ConfigureWebHost(web => + { + web.UseKestrel(o => + // Ephemeral loopback port, h2c (HTTP/2 cleartext) so gRPC works without TLS. + o.Listen(IPAddress.Loopback, 0, listen => listen.Protocols = HttpProtocols.Http2)); + web.ConfigureServices(services => + { + services.AddRouting(); + services.AddGrpc(); + services.AddLogging(); + // Pre-constructed instance: host tear-down (KillTransport) will NOT dispose _dbB. + services.AddSingleton(_dbB); + services.AddZbLocalDbReplication(_passiveConfig); + }); + web.Configure(app => + { + app.UseRouting(); + app.UseEndpoints(e => e.MapZbLocalDbSync()); + }); + }) + .StartAsync(); + + var address = host.Services.GetRequiredService() + .Features.Get()!.Addresses.Single(); + return (host, address); + } + + private async Task DisposeServerHostAsync() + { + var host = _serverHost; + _serverHost = null; + if (host is null) + return; + try { await host.StopAsync(TimeSpan.FromSeconds(5)); } catch { /* teardown */ } + host.Dispose(); + } + + private static async Task OplogCountAsync(ILocalDb db) + { + var rows = await db.QueryAsync("SELECT COUNT(*) FROM __localdb_oplog", static r => r.GetInt64(0)); + return rows[0]; + } + + private static async Task DumpOrdersAsync(ILocalDb db) + { + var rows = await db.QueryAsync( + "SELECT id, sku, qty FROM orders ORDER BY id", + static r => $"{r.GetInt64(0)}|{(r.IsDBNull(1) ? "" : r.GetString(1))}|{(r.IsDBNull(2) ? "" : r.GetInt64(2).ToString())}"); + return string.Join("\n", rows); + } + + private static async Task DumpRowVersionAsync(ILocalDb db) + { + var rows = await db.QueryAsync( + "SELECT table_name, pk_json, hlc, node_id, is_tombstone FROM __localdb_row_version " + + "ORDER BY table_name, pk_json", + static r => $"{r.GetString(0)}|{r.GetString(1)}|{r.GetInt64(2)}|{r.GetString(3)}|{r.GetInt64(4)}"); + return string.Join("\n", rows); + } + + private static IConfiguration BuildConfig(bool initiator, Action>? tune) + { + var dict = new Dictionary + { + ["LocalDb:Replication:FlushInterval"] = "00:00:00.050", + // Bound the initiator's reconnect backoff so recovery after a KillTransport/RestartTransport + // cycle is deterministic (the default 60s max lets the doubling backoff grow enough to + // occasionally overrun the 30s converge poll). + ["LocalDb:Replication:ReconnectBackoffMax"] = "00:00:02", + }; + if (initiator) + dict["LocalDb:Replication:PeerAddress"] = "http://localhost"; + tune?.Invoke(dict); + return new ConfigurationBuilder().AddInMemoryCollection(dict).Build(); + } + + public async ValueTask DisposeAsync() + { + if (_bg is not null) + try { await _bg.StopAsync(CancellationToken.None); } catch { /* teardown */ } + await DisposeServerHostAsync(); + if (_clientProvider is not null) + await _clientProvider.DisposeAsync(); + + _dbA.Dispose(); + _dbB.Dispose(); + Microsoft.Data.Sqlite.SqliteConnection.ClearAllPools(); + foreach (var p in new[] { _pathA, _pathB }) + { + if (File.Exists(p)) File.Delete(p); + if (File.Exists(p + "-wal")) File.Delete(p + "-wal"); + if (File.Exists(p + "-shm")) File.Delete(p + "-shm"); + } + } +} diff --git a/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/Convergence/RandomOpsConvergenceTests.cs b/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/Convergence/RandomOpsConvergenceTests.cs new file mode 100644 index 0000000..308cc6b --- /dev/null +++ b/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/Convergence/RandomOpsConvergenceTests.cs @@ -0,0 +1,102 @@ +namespace ZB.MOM.WW.LocalDb.Tests.Convergence; + +/// +/// Randomized property test: a large stream of mixed insert/update/delete operations on both nodes +/// over a small shared key space, interleaved with genuine transport disconnect/reconnect cycles, +/// must always converge to byte-identical state. Seeded Random only (no time-based seeds) so every +/// failure is reproducible from the logged seed. +/// +public sealed class RandomOpsConvergenceTests +{ + private const int OpsPerNode = 500; + private const int KeySpace = 40; + + [Theory] + [InlineData(1701)] + [InlineData(42)] + [InlineData(7)] + public async Task RandomInterleavedOps_Converge(int seed) + { + var rng = new Random(seed); + + // Segment plan derived up front from the seeded RNG (so the whole run is deterministic): + // steady, OUTAGE-1, steady, OUTAGE-2, steady — summing to OpsPerNode ops on each node. + var seg1 = rng.Next(80, 150); + var outage1 = rng.Next(20, 50); + var seg2 = rng.Next(80, 150); + var outage2 = rng.Next(20, 50); + var tail = OpsPerNode - (seg1 + outage1 + seg2 + outage2); + + await using var fx = new ConvergenceFixture(); + await fx.StartAsync(); + + try + { + await RunOpsAsync(fx, rng, seg1); + await OutageAsync(fx, rng, outage1); + await RunOpsAsync(fx, rng, seg2); + await OutageAsync(fx, rng, outage2); + await RunOpsAsync(fx, rng, tail); + + await fx.AssertConvergedAsync(); + } + catch (Exception ex) when (ex is not Xunit.Sdk.XunitException) + { + throw new Xunit.Sdk.XunitException($"seed {seed}: run faulted.\n{ex}"); + } + catch (Xunit.Sdk.XunitException ex) + { + throw new Xunit.Sdk.XunitException($"seed {seed}: convergence failed.\n{ex.Message}"); + } + + // Both outage windows actually severed and re-dialed the transport (initial dial + 2 + // reconnects) — proof the run exercised real offline accumulation + resync. + Assert.True(fx.InitiatorConnectionAttempts >= 3, + $"seed {seed}: expected >=3 connection attempts (2 reconnects), got {fx.InitiatorConnectionAttempts}"); + } + + // Severs the transport, accumulates offline writes on BOTH nodes, waits until the initiator has + // actually observed the loss (a reconnect attempt started), then restores the transport. + private static async Task OutageAsync(ConvergenceFixture fx, Random rng, int ops) + { + var baseline = fx.InitiatorConnectionAttempts; + await fx.KillTransportAsync(); + await RunOpsAsync(fx, rng, ops); + await WaitForAsync(() => Task.FromResult(fx.InitiatorConnectionAttempts > baseline), + TimeSpan.FromSeconds(10)); + await fx.RestartTransportAsync(); + } + + private static async Task RunOpsAsync(ConvergenceFixture fx, Random rng, int count) + { + // One op per node per iteration; no sleeps — the sessions' pumps drain concurrently. + for (var i = 0; i < count; i++) + { + await ApplyRandomOpAsync(fx.A, rng); + await ApplyRandomOpAsync(fx.B, rng); + } + } + + private static Task ApplyRandomOpAsync(ILocalDb db, Random rng) + { + var id = rng.Next(1, KeySpace + 1); + return rng.Next(3) switch + { + // insert / update both resolve through the supported upsert helper (plain UPDATE-else-INSERT). + 0 or 1 => ConvergenceFixture.UpsertAsync(db, id, "v" + rng.Next(1000), rng.Next(10_000)), + _ => ConvergenceFixture.DeleteAsync(db, id), + }; + } + + private static async Task WaitForAsync(Func> predicate, TimeSpan timeout) + { + var deadline = DateTime.UtcNow + timeout; + while (DateTime.UtcNow < deadline) + { + if (await predicate()) + return; + await Task.Delay(20); + } + throw new TimeoutException("Condition not reached within " + timeout); + } +}