diff --git a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/LwwApplier.cs b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/LwwApplier.cs new file mode 100644 index 0000000..6c37e81 --- /dev/null +++ b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/LwwApplier.cs @@ -0,0 +1,151 @@ +using System.Globalization; +using Microsoft.Data.Sqlite; +using ZB.MOM.WW.LocalDb.Internal; +using ZB.MOM.WW.LocalDb.Registration; + +namespace ZB.MOM.WW.LocalDb.Replication.Internal; + +/// Outcome of applying one inbound batch. is the batch's highest seq regardless of per-entry disposition, so the watermark advances even over discarded/dead-lettered entries. +internal sealed record ApplyResult(long AppliedThruSeq, int Applied, int Discarded, int DeadLettered); + +/// +/// Applies a peer's oplog batch under last-writer-wins in one transaction: the applying guard is +/// raised so capture triggers skip these writes, per-entry LWW resolves against the local +/// row-version (hlc, then node_id ordinal tie-break), poison rows are dead-lettered while the +/// batch continues, and the remote watermark advances in the SAME transaction as the data so a +/// crash-replay of the batch is a no-op (LWW's strict-greater compare loses on equal versions). +/// This never writes __localdb_oplog: the 2-node topology does not re-forward applied deltas. +/// +internal sealed class LwwApplier(SqliteLocalDb db) +{ + public async Task ApplyBatchAsync(IReadOnlyList entries, CancellationToken ct) + { + if (entries.Count == 0) + return new ApplyResult(0, 0, 0, 0); + + // Validated before opening the transaction: an unknown table is a schema disagreement + // between the peers (a protocol violation), not a per-row poison — reject the whole batch + // so nothing is half-applied. + foreach (var entry in entries) + if (!db.ReplicatedTables.ContainsKey(entry.TableName)) + throw new LocalDbSchemaMismatchException( + $"Received an oplog entry for table '{entry.TableName}', which is not registered for replication on this node."); + + var utcNow = DateTimeOffset.UtcNow.UtcDateTime.ToString(OplogStore.TombstoneUtcFormat, CultureInfo.InvariantCulture); + var maxSeq = 0L; + var maxHlc = 0L; + foreach (var entry in entries) + { + if (entry.Seq > maxSeq) maxSeq = entry.Seq; + if (entry.Hlc > maxHlc) maxHlc = entry.Hlc; + } + + var sqlByTable = new Dictionary(StringComparer.Ordinal); + var applied = 0; + var discarded = 0; + var deadLettered = 0; + + await using var tx = await db.BeginTransactionAsync(ct); + await tx.ExecuteAsync("UPDATE __localdb_applying SET applying = 1 WHERE id = 1", null, ct); + + foreach (var entry in entries) + { + var local = await tx.QueryAsync( + "SELECT hlc, node_id FROM __localdb_row_version WHERE table_name = @t AND pk_json = @pk", + static r => (Hlc: r.GetInt64(0), NodeId: r.GetString(1)), + new { t = entry.TableName, pk = entry.PkJson }, ct); + + var remoteWins = local.Count == 0 + || entry.Hlc > local[0].Hlc + || (entry.Hlc == local[0].Hlc && string.CompareOrdinal(entry.NodeId, local[0].NodeId) > 0); + + if (!remoteWins) + { + discarded++; + continue; + } + + if (!sqlByTable.TryGetValue(entry.TableName, out var sql)) + { + sql = BuildTableSql(db.ReplicatedTables[entry.TableName]); + sqlByTable[entry.TableName] = sql; + } + + try + { + if (entry.IsTombstone) + await tx.ExecuteAsync(sql.Delete, new { pk = entry.PkJson }, ct); + else + await tx.ExecuteAsync(sql.Upsert, new { row = entry.RowJson }, ct); + + await tx.ExecuteAsync( + "INSERT OR REPLACE INTO __localdb_row_version " + + "(table_name, pk_json, hlc, node_id, is_tombstone, tombstone_utc) " + + "VALUES (@t, @pk, @hlc, @node, @tomb, @tut)", + new + { + t = entry.TableName, + pk = entry.PkJson, + hlc = entry.Hlc, + node = entry.NodeId, + tomb = entry.IsTombstone ? 1 : 0, + tut = entry.IsTombstone ? utcNow : null, + }, ct); + applied++; + } + catch (SqliteException ex) + { + // A statement-level SQLite error (e.g. json_extract on malformed row_json) aborts only + // that statement, not the transaction, so the batch continues after dead-lettering. + await tx.ExecuteAsync( + "INSERT INTO __localdb_dead_letter " + + "(received_utc, table_name, pk_json, row_json, hlc, node_id, error) " + + "VALUES (@utc, @t, @pk, @row, @hlc, @node, @err)", + new + { + utc = utcNow, + t = entry.TableName, + pk = entry.PkJson, + row = entry.RowJson, + hlc = entry.Hlc, + node = entry.NodeId, + err = ex.Message, + }, ct); + deadLettered++; + } + } + + await tx.ExecuteAsync( + "UPDATE __localdb_peer_state SET " + + "last_applied_remote_seq = MAX(last_applied_remote_seq, @maxSeq), " + + "last_seen_hlc = MAX(last_seen_hlc, @maxHlc), last_sync_utc = @utc WHERE id = 1", + new { maxSeq, maxHlc, utc = utcNow }, ct); + + await tx.ExecuteAsync("UPDATE __localdb_applying SET applying = 0 WHERE id = 1", null, ct); + await tx.CommitAsync(ct); + + db.Clock.Observe(maxHlc); + return new ApplyResult(maxSeq, applied, discarded, deadLettered); + } + + private static (string Delete, string Upsert) BuildTableSql(ReplicatedTable table) + { + var name = Ident(table.Name); + + var pkMatch = string.Join(" AND ", table.PkColumns.Select(p => $"{Ident(p)} = json_extract(@pk, {JsonPath(p)})")); + var delete = $"DELETE FROM {name} WHERE {pkMatch}"; + + var columns = string.Join(", ", table.Columns.Select(c => Ident(c.Name))); + var values = string.Join(", ", table.Columns.Select(c => $"json_extract(@row, {JsonPath(c.Name)})")); + var upsert = $"INSERT OR REPLACE INTO {name} ({columns}) VALUES ({values})"; + + return (delete, upsert); + } + + // Local copy of TriggerSqlGenerator's convention (that helper lives in the core assembly). + private static string Ident(string name) => "\"" + name.Replace("\"", "\"\"") + "\""; + + // A single-quoted JSON path with a quoted key member, so a column name with reserved characters + // still resolves: $."col". The key's double quotes are backslash-escaped per JSON path grammar. + private static string JsonPath(string column) => "'$.\"" + column.Replace("\"", "\\\"") + "\"'"; +} diff --git a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/OplogStore.cs b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/OplogStore.cs index ae844c3..078784d 100644 --- a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/OplogStore.cs +++ b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/OplogStore.cs @@ -18,7 +18,7 @@ internal sealed record PeerState( internal sealed class OplogStore(ILocalDb db, ReplicationOptions options, Func? utcNow = null) { // Must match the delete trigger's strftime('%Y-%m-%dT%H:%M:%fZ') output so string comparison is chronological. - private const string TombstoneUtcFormat = "yyyy-MM-ddTHH:mm:ss.fffZ"; + internal const string TombstoneUtcFormat = "yyyy-MM-ddTHH:mm:ss.fffZ"; private readonly Func _utcNow = utcNow ?? (() => DateTimeOffset.UtcNow); diff --git a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/LocalDbSchemaMismatchException.cs b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/LocalDbSchemaMismatchException.cs new file mode 100644 index 0000000..8d45664 --- /dev/null +++ b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/LocalDbSchemaMismatchException.cs @@ -0,0 +1,15 @@ +namespace ZB.MOM.WW.LocalDb.Replication; + +/// +/// Thrown when a peer sends replication data whose schema this node cannot honor: an oplog entry +/// for a table not registered for replication here. Unlike a poison entry (dead-lettered per-row), +/// this is a protocol violation between the two nodes' registered schemas, so the whole batch is +/// rejected before any of it is applied. Task 10 reuses it to fail the handshake closed. +/// +public sealed class LocalDbSchemaMismatchException : Exception +{ + public LocalDbSchemaMismatchException(string message) : base(message) { } + + public LocalDbSchemaMismatchException(string message, Exception innerException) + : base(message, innerException) { } +} diff --git a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/ZB.MOM.WW.LocalDb.csproj b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/ZB.MOM.WW.LocalDb.csproj index a16d144..d9eca86 100644 --- a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/ZB.MOM.WW.LocalDb.csproj +++ b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/ZB.MOM.WW.LocalDb.csproj @@ -26,6 +26,7 @@ + diff --git a/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/LwwApplierTests.cs b/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/LwwApplierTests.cs new file mode 100644 index 0000000..a3cf59b --- /dev/null +++ b/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/LwwApplierTests.cs @@ -0,0 +1,308 @@ +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"); + + public void Dispose() + { + SqliteConnection.ClearAllPools(); + if (File.Exists(_path)) + File.Delete(_path); + } + + 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); + } +}