using Microsoft.Data.Sqlite; 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; public sealed class LwwApplierTests : IDisposable { private const string Pk1 = "{\"id\":1}"; private readonly string _path = Path.Combine(Path.GetTempPath(), Guid.NewGuid() + ".db"); private readonly string _path2 = Path.Combine(Path.GetTempPath(), Guid.NewGuid() + ".db"); public void Dispose() { SqliteConnection.ClearAllPools(); if (File.Exists(_path)) File.Delete(_path); if (File.Exists(_path2)) File.Delete(_path2); } private async Task NewOrdersDb() { var db = new SqliteLocalDb(new LocalDbOptions { Path = _path }); await db.ExecuteAsync("CREATE TABLE orders (id INTEGER PRIMARY KEY, sku TEXT, qty INTEGER)"); db.RegisterReplicated("orders"); return db; } private static async Task RowVersionHlc(SqliteLocalDb db, string pkJson) { var r = await db.QueryAsync( "SELECT hlc FROM __localdb_row_version WHERE table_name = 'orders' AND pk_json = @pk", x => x.GetInt64(0), new { pk = pkJson }); return r[0]; } private static async Task<(string? Sku, long? Qty)?> ReadOrder(SqliteLocalDb db, long id) { var r = await db.QueryAsync( "SELECT sku, qty FROM orders WHERE id = @id", x => (x.IsDBNull(0) ? null : x.GetString(0), (long?)(x.IsDBNull(1) ? null : x.GetInt64(1))), new { id }); return r.Count == 0 ? null : r[0]; } private static async Task OplogCount(SqliteLocalDb db) { var r = await db.QueryAsync("SELECT COUNT(*) FROM __localdb_oplog", x => x.GetInt64(0)); return r[0]; } private static OplogEntryRecord Entry( long seq, string pkJson, string? rowJson, long hlc, string nodeId, bool tombstone = false) => new(seq, "orders", pkJson, rowJson, hlc, nodeId, tombstone); [Fact] public async Task Apply_NewerRemote_Wins() { using var db = await NewOrdersDb(); await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'LOCAL', 5)"); var localHlc = await RowVersionHlc(db, Pk1); var result = await new LwwApplier(db).ApplyBatchAsync( [Entry(1, Pk1, "{\"id\":1,\"sku\":\"REMOTE\",\"qty\":99}", localHlc + 1000, "remote-node")], default); Assert.Equal(1, result.Applied); var row = await ReadOrder(db, 1); Assert.Equal(("REMOTE", (long?)99), row); var rv = await db.QueryAsync( "SELECT hlc, node_id FROM __localdb_row_version WHERE pk_json = @pk", x => (Hlc: x.GetInt64(0), Node: x.GetString(1)), new { pk = Pk1 }); Assert.Equal(localHlc + 1000, rv[0].Hlc); Assert.Equal("remote-node", rv[0].Node); } [Fact] public async Task Apply_OlderRemote_Discarded_StillCounted() { using var db = await NewOrdersDb(); await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'LOCAL', 5)"); var localHlc = await RowVersionHlc(db, Pk1); var result = await new LwwApplier(db).ApplyBatchAsync( [Entry(7, Pk1, "{\"id\":1,\"sku\":\"OLD\",\"qty\":1}", localHlc - 1000, "remote-node")], default); Assert.Equal(0, result.Applied); Assert.Equal(1, result.Discarded); Assert.Equal(7, result.AppliedThruSeq); Assert.Equal(("LOCAL", (long?)5), await ReadOrder(db, 1)); var peer = await db.QueryAsync( "SELECT last_applied_remote_seq FROM __localdb_peer_state WHERE id = 1", x => x.GetInt64(0)); Assert.Equal(7, peer[0]); } [Fact] public async Task Apply_EqualHlc_NodeIdTieBreak_Deterministic() { // Greater remote node_id wins the tie. using (var db = await NewOrdersDb()) { await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'LOCAL', 5)"); var localHlc = await RowVersionHlc(db, Pk1); var greater = db.NodeId + "z"; Assert.True(string.CompareOrdinal(greater, db.NodeId) > 0); var result = await new LwwApplier(db).ApplyBatchAsync( [Entry(1, Pk1, "{\"id\":1,\"sku\":\"WINS\",\"qty\":9}", localHlc, greater)], default); Assert.Equal(1, result.Applied); Assert.Equal(("WINS", (long?)9), await ReadOrder(db, 1)); } SqliteConnection.ClearAllPools(); File.Delete(_path); // Smaller remote node_id loses the tie. using (var db = await NewOrdersDb()) { await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'LOCAL', 5)"); var localHlc = await RowVersionHlc(db, Pk1); var smaller = "!" + db.NodeId; Assert.True(string.CompareOrdinal(smaller, db.NodeId) < 0); var result = await new LwwApplier(db).ApplyBatchAsync( [Entry(1, Pk1, "{\"id\":1,\"sku\":\"LOSES\",\"qty\":9}", localHlc, smaller)], default); Assert.Equal(0, result.Applied); Assert.Equal(1, result.Discarded); Assert.Equal(("LOCAL", (long?)5), await ReadOrder(db, 1)); } } [Fact] public async Task Apply_NewRow_Inserts() { using var db = await NewOrdersDb(); var result = await new LwwApplier(db).ApplyBatchAsync( [Entry(3, Pk1, "{\"id\":1,\"sku\":\"NEW\",\"qty\":42}", 5_000_000, "remote-node")], default); Assert.Equal(1, result.Applied); Assert.Equal(("NEW", (long?)42), await ReadOrder(db, 1)); } [Fact] public async Task Apply_Tombstone_DeletesRow() { using var db = await NewOrdersDb(); await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'LOCAL', 5)"); var localHlc = await RowVersionHlc(db, Pk1); var result = await new LwwApplier(db).ApplyBatchAsync( [Entry(4, Pk1, null, localHlc + 1000, "remote-node", tombstone: true)], default); Assert.Equal(1, result.Applied); Assert.Null(await ReadOrder(db, 1)); var rv = await db.QueryAsync( "SELECT is_tombstone, hlc FROM __localdb_row_version WHERE pk_json = @pk", x => (Tomb: x.GetInt64(0), Hlc: x.GetInt64(1)), new { pk = Pk1 }); Assert.Equal(1, rv[0].Tomb); Assert.Equal(localHlc + 1000, rv[0].Hlc); } [Fact] public async Task Apply_DoesNotFireCaptureTriggers() { using var db = await NewOrdersDb(); await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'LOCAL', 5)"); var before = await OplogCount(db); await new LwwApplier(db).ApplyBatchAsync( [ Entry(1, Pk1, "{\"id\":1,\"sku\":\"R\",\"qty\":9}", 9_000_000, "remote-node"), Entry(2, "{\"id\":2}", "{\"id\":2,\"sku\":\"R2\",\"qty\":7}", 9_000_001, "remote-node"), ], default); Assert.Equal(before, await OplogCount(db)); } [Fact] public async Task Apply_Batch_Atomic_WatermarkSameTxn() { using var db = await NewOrdersDb(); var result = await new LwwApplier(db).ApplyBatchAsync( [ Entry(10, Pk1, "{\"id\":1,\"sku\":\"A\",\"qty\":1}", 1_000_000, "remote-node"), Entry(11, "{\"id\":2}", "{\"id\":2,\"sku\":\"B\",\"qty\":2}", 1_000_001, "remote-node"), ], default); Assert.Equal(2, result.Applied); Assert.Equal(11, result.AppliedThruSeq); // Data and the watermark are both visible => they committed in one transaction. Assert.Equal(("A", (long?)1), await ReadOrder(db, 1)); Assert.Equal(("B", (long?)2), await ReadOrder(db, 2)); var peer = await db.QueryAsync( "SELECT last_applied_remote_seq, last_seen_hlc FROM __localdb_peer_state WHERE id = 1", x => (Seq: x.GetInt64(0), Hlc: x.GetInt64(1))); Assert.Equal(11, peer[0].Seq); Assert.Equal(1_000_001, peer[0].Hlc); } [Fact] public async Task Apply_UnregisteredTable_Throws_NothingApplied() { using var db = await NewOrdersDb(); var batch = new OplogEntryRecord[] { Entry(20, Pk1, "{\"id\":1,\"sku\":\"A\",\"qty\":1}", 1_000_000, "remote-node"), new(21, "widgets", "{\"id\":9}", "{\"id\":9}", 1_000_001, "remote-node", false), }; await Assert.ThrowsAsync( () => new LwwApplier(db).ApplyBatchAsync(batch, default)); // The valid entry that preceded the violation must not have been applied. Assert.Null(await ReadOrder(db, 1)); var peer = await db.QueryAsync( "SELECT last_applied_remote_seq FROM __localdb_peer_state WHERE id = 1", x => x.GetInt64(0)); Assert.Equal(0, peer[0]); } [Fact] public async Task Apply_PoisonEntry_DeadLettered_BatchContinues() { using var db = await NewOrdersDb(); var result = await new LwwApplier(db).ApplyBatchAsync( [ Entry(30, Pk1, "NOT VALID JSON", 1_000_000, "remote-node"), Entry(31, "{\"id\":2}", "{\"id\":2,\"sku\":\"OK\",\"qty\":8}", 1_000_001, "remote-node"), ], default); Assert.Equal(1, result.DeadLettered); Assert.Equal(1, result.Applied); Assert.Equal(("OK", (long?)8), await ReadOrder(db, 2)); Assert.Null(await ReadOrder(db, 1)); var dead = await db.QueryAsync( "SELECT pk_json, error FROM __localdb_dead_letter", x => (Pk: x.GetString(0), Err: x.GetString(1))); Assert.Single(dead); Assert.Equal(Pk1, dead[0].Pk); Assert.NotEmpty(dead[0].Err); } [Fact] public async Task Apply_IsIdempotent() { using var db = await NewOrdersDb(); var batch = new OplogEntryRecord[] { Entry(40, Pk1, "{\"id\":1,\"sku\":\"X\",\"qty\":3}", 7_000_000, "remote-node"), new(41, "orders", "{\"id\":2}", "{\"id\":2,\"sku\":\"Y\",\"qty\":4}", 7_000_001, "remote-node", false), }; var applier = new LwwApplier(db); var first = await applier.ApplyBatchAsync(batch, default); Assert.Equal(2, first.Applied); var second = await applier.ApplyBatchAsync(batch, default); Assert.Equal(0, second.Applied); Assert.Equal(2, second.Discarded); Assert.Equal(("X", (long?)3), await ReadOrder(db, 1)); Assert.Equal(("Y", (long?)4), await ReadOrder(db, 2)); var rows = await db.QueryAsync("SELECT COUNT(*) FROM orders", x => x.GetInt64(0)); Assert.Equal(2, rows[0]); } [Fact] public async Task Apply_ObservesRemoteHlc() { using var db = await NewOrdersDb(); var maxHlc = (DateTimeOffset.UtcNow.AddDays(365).ToUnixTimeMilliseconds() << 16); await new LwwApplier(db).ApplyBatchAsync( [Entry(1, Pk1, "{\"id\":1,\"sku\":\"R\",\"qty\":1}", maxHlc, "remote-node")], default); Assert.True(db.Clock.Next() > maxHlc); } [Fact] public async Task Apply_AdvancesPeerState() { using var db = await NewOrdersDb(); await new LwwApplier(db).ApplyBatchAsync( [ Entry(50, Pk1, "{\"id\":1,\"sku\":\"A\",\"qty\":1}", 3_000_000, "remote-node"), Entry(52, "{\"id\":2}", "{\"id\":2,\"sku\":\"B\",\"qty\":2}", 3_000_050, "remote-node"), ], default); var peer = await db.QueryAsync( "SELECT last_applied_remote_seq, last_seen_hlc, last_sync_utc FROM __localdb_peer_state WHERE id = 1", x => (Seq: x.GetInt64(0), Hlc: x.GetInt64(1), Utc: x.IsDBNull(2) ? null : x.GetString(2))); Assert.Equal(52, peer[0].Seq); Assert.Equal(3_000_050, peer[0].Hlc); Assert.NotNull(peer[0].Utc); Assert.EndsWith("Z", peer[0].Utc); } [Fact] public async Task Apply_NullRowJsonNonTombstone_DeadLettered_NoPhantomRow() { using var db = await NewOrdersDb(); await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'LOCAL', 5)"); var result = await new LwwApplier(db).ApplyBatchAsync( [Entry(60, "{\"id\":2}", null, 9_000_000_000_000_000, "remote-node", tombstone: false)], default); Assert.Equal(1, result.DeadLettered); Assert.Equal(0, result.Applied); var dead = await db.QueryAsync( "SELECT pk_json, error FROM __localdb_dead_letter", x => (Pk: x.GetString(0), Err: x.GetString(1))); Assert.Single(dead); Assert.Equal("{\"id\":2}", dead[0].Pk); Assert.Contains("null row_json", dead[0].Err); // No phantom auto-rowid row: the table still holds exactly the one pre-existing row. var rows = await db.QueryAsync("SELECT COUNT(*) FROM orders", x => x.GetInt64(0)); Assert.Equal(1, rows[0]); Assert.Null(await ReadOrder(db, 2)); } [Fact] public async Task Apply_RoundTrip_PreservesRealAndBigIntegerExactly() { const double real17 = 1.2345678901234567; // 17 significant digits const long bigInt = 9_007_199_254_740_993; // 2^53 + 1: not representable as a double using var source = new SqliteLocalDb(new LocalDbOptions { Path = _path }); await source.ExecuteAsync("CREATE TABLE meas (id INTEGER PRIMARY KEY, r REAL, big INTEGER)"); source.RegisterReplicated("meas"); await source.ExecuteAsync( "INSERT INTO meas (id, r, big) VALUES (1, @r, @big)", new { r = real17, big = bigInt }); var captured = await source.QueryAsync( "SELECT seq, table_name, pk_json, row_json, hlc, node_id, is_tombstone FROM __localdb_oplog", r => new OplogEntryRecord( r.GetInt64(0), r.GetString(1), r.GetString(2), r.IsDBNull(3) ? null : r.GetString(3), r.GetInt64(4), r.GetString(5), r.GetInt64(6) != 0)); Assert.Single(captured); using var target = new SqliteLocalDb(new LocalDbOptions { Path = _path2 }); await target.ExecuteAsync("CREATE TABLE meas (id INTEGER PRIMARY KEY, r REAL, big INTEGER)"); target.RegisterReplicated("meas"); var result = await new LwwApplier(target).ApplyBatchAsync(captured, default); Assert.Equal(1, result.Applied); var applied = await target.QueryAsync( "SELECT r, big FROM meas WHERE id = 1", x => (R: x.GetDouble(0), Big: x.GetInt64(1))); Assert.Equal(BitConverter.DoubleToInt64Bits(real17), BitConverter.DoubleToInt64Bits(applied[0].R)); Assert.Equal(bigInt, applied[0].Big); } [Fact] public async Task Apply_GuardAlwaysLoweredAfterBatch() { // The abort/rollback path needs no separate coverage: applying = 1 is set INSIDE the // transaction, so any rollback restores it to 0 by SQLite transaction semantics. using var db = await NewOrdersDb(); await new LwwApplier(db).ApplyBatchAsync( [Entry(1, Pk1, "{\"id\":1,\"sku\":\"R\",\"qty\":1}", 5_000_000, "remote-node")], default); var guard = await db.QueryAsync( "SELECT applying FROM __localdb_applying WHERE id = 1", x => x.GetInt64(0)); Assert.Equal(0, guard[0]); // A subsequent consumer write must capture normally — proof the guard is really down. var before = await OplogCount(db); await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (2, 'LOCAL', 2)"); Assert.Equal(before + 1, await OplogCount(db)); } [Fact] public async Task Apply_SameTable_TombstoneAndUpsert_OneBatch() { using var db = await NewOrdersDb(); await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'LOCAL', 5)"); var localHlc = await RowVersionHlc(db, Pk1); var result = await new LwwApplier(db).ApplyBatchAsync( [ Entry(70, Pk1, null, localHlc + 1000, "remote-node", tombstone: true), Entry(71, "{\"id\":2}", "{\"id\":2,\"sku\":\"NEW\",\"qty\":7}", localHlc + 1001, "remote-node"), ], default); Assert.Equal(2, result.Applied); Assert.Null(await ReadOrder(db, 1)); Assert.Equal(("NEW", (long?)7), await ReadOrder(db, 2)); } [Fact] public async Task Apply_ReplaceExistingRow_DoesNotFireCaptureTriggers() { // INSERT OR REPLACE over an existing row also exercises the delete side of REPLACE; // neither half may reach the capture triggers. using var db = await NewOrdersDb(); await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'LOCAL', 5)"); var localHlc = await RowVersionHlc(db, Pk1); var before = await OplogCount(db); var result = await new LwwApplier(db).ApplyBatchAsync( [Entry(80, Pk1, "{\"id\":1,\"sku\":\"REPL\",\"qty\":6}", localHlc + 1000, "remote-node")], default); Assert.Equal(1, result.Applied); Assert.Equal(("REPL", (long?)6), await ReadOrder(db, 1)); Assert.Equal(before, await OplogCount(db)); } }