feat(localdb): LWW applier (hlc+node_id tie-break, applying guard, dead-letter, atomic watermark)

Claude-Session: https://claude.ai/code/session_01BL2Vu1ESDQ9SCN4gVKkdts

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;
+
+/// <summary>Outcome of applying one inbound batch. <see cref="AppliedThruSeq"/> is the batch's highest seq regardless of per-entry disposition, so the watermark advances even over discarded/dead-lettered entries.</summary>
+internal sealed record ApplyResult(long AppliedThruSeq, int Applied, int Discarded, int DeadLettered);
+
+/// <summary>
+/// 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 <c>__localdb_oplog</c>: the 2-node topology does not re-forward applied deltas.
+/// </summary>
+internal sealed class LwwApplier(SqliteLocalDb db)
+{
+    public async Task<ApplyResult> ApplyBatchAsync(IReadOnlyList<OplogEntryRecord> 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<string, (string Delete, string Upsert)>(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("\"", "\\\"") + "\"'";
+}

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<DateTimeOffset>? 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<DateTimeOffset> _utcNow = utcNow ?? (() => DateTimeOffset.UtcNow);
 

ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/LocalDbSchemaMismatchException.cs

@@ -0,0 +1,15 @@
+namespace ZB.MOM.WW.LocalDb.Replication;
+
+/// <summary>
+/// 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.
+/// </summary>
+public sealed class LocalDbSchemaMismatchException : Exception
+{
+    public LocalDbSchemaMismatchException(string message) : base(message) { }
+
+    public LocalDbSchemaMismatchException(string message, Exception innerException)
+        : base(message, innerException) { }
+}

ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/ZB.MOM.WW.LocalDb.csproj

@@ -26,6 +26,7 @@
   </ItemGroup>
 
   <ItemGroup>
+    <InternalsVisibleTo Include="ZB.MOM.WW.LocalDb.Replication" />
     <InternalsVisibleTo Include="ZB.MOM.WW.LocalDb.Tests" />
   </ItemGroup>
 

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<SqliteLocalDb> 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<long> 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<long> 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<LocalDbSchemaMismatchException>(
+            () => 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);
+    }
+}