fix(localdb): escape apostrophes in apply/snapshot JSON paths (LocalDb-006)

A column name containing ' was captured fine but ended the JSON-path SQL
literal early on the apply and snapshot paths, dead-lettering every row of the
table on the peer and faulting the snapshot page query for a quoted PK column.
Both paths now share JsonPathSql.For, which doubles ' after the JSON-path
escape. 2-node delta and wiped-peer snapshot tests fail pre-fix.

(scadaproj#46)

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

ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/JsonPathSql.cs

@@ -0,0 +1,24 @@
+namespace ZB.MOM.WW.LocalDb.Replication.Internal;
+
+/// <summary>
+/// Builds the SQL string literal for a JSON path that selects one column's member out of a captured
+/// <c>row_json</c> / <c>pk_json</c> object. Shared by <see cref="LwwApplier"/> and
+/// <see cref="SnapshotStreamer"/> so the two cannot drift apart again.
+/// </summary>
+internal static class JsonPathSql
+{
+    /// <summary>
+    /// <c>'$."col"'</c>: a quoted key member, so a column name with reserved characters still
+    /// resolves. Two escaping layers, applied in order: the column's <c>"</c> is backslash-escaped
+    /// for the JSON path grammar, then every <c>'</c> in the resulting path is doubled for the SQL
+    /// string literal around it. Column names come from <c>PRAGMA table_info</c> and are not
+    /// regex-gated the way table names are, and capture (<c>TriggerSqlGenerator</c>) already accepts
+    /// any name — so a column called <c>it's</c> used to be captured, then end this literal early
+    /// and dead-letter every row of its table on the peer (LocalDb-006).
+    /// </summary>
+    public static string For(string column)
+    {
+        var path = "$.\"" + column.Replace("\"", "\\\"") + "\"";
+        return "'" + path.Replace("'", "''") + "'";
+    }
+}

ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/LwwApplier.cs

@@ -388,9 +388,6 @@ internal sealed class LwwApplier(SqliteLocalDb db, Func<DateTimeOffset>? utcNow
         // 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
+        private static string JsonPath(string column) => JsonPathSql.For(column);
-        // 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/SnapshotStreamer.cs

@@ -234,9 +234,7 @@ internal sealed class SnapshotStreamer(SqliteLocalDb db, OplogStore store, Repli
 
     private static string StringLiteral(string value) => "'" + value.Replace("'", "''") + "'";
 
-    // A single-quoted JSON path with a quoted key member so a column name with reserved characters
+    private static string JsonPath(string column) => JsonPathSql.For(column);
-    // still resolves: $."col".
-    private static string JsonPath(string column) => "'$.\"" + column.Replace("\"", "\\\"") + "\"'";
 }
 
 /// <summary>

ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/QuotedColumnNameTests.cs

@@ -0,0 +1,160 @@
+using System.Threading.Channels;
+using Microsoft.Extensions.Logging.Abstractions;
+using ZB.MOM.WW.LocalDb.Contracts;
+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;
+
+/// <summary>
+/// LocalDb-006 (scadaproj#46): capture escapes both quote styles, so a column called <c>it's</c> was
+/// captured fine — but the apply and snapshot SQL embedded the column in a JSON-path string literal
+/// that escaped <c>"</c> and not <c>'</c>. The apostrophe ended the literal early, every row of the
+/// table was dead-lettered on the receiver (<c>near "s": syntax error</c>), and a quoted PK column
+/// faulted the snapshot page query on every reconnect.
+/// </summary>
+public sealed class QuotedColumnNameTests : IDisposable
+{
+    private static readonly TimeSpan RunTimeout = TimeSpan.FromSeconds(15);
+
+    // Both quote characters, in a PK column and in a non-PK column.
+    private const string Ddl = "CREATE TABLE q (\"o'key\" TEXT PRIMARY KEY, \"it's\" TEXT, \"say \"\"hi\"\"\" TEXT)";
+
+    private readonly List<string> _paths = [];
+    private readonly List<IDisposable> _disposables = [];
+
+    public void Dispose()
+    {
+        foreach (var d in _disposables)
+            d.Dispose();
+        Microsoft.Data.Sqlite.SqliteConnection.ClearAllPools();
+        foreach (var p in _paths)
+        {
+            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");
+        }
+    }
+
+    private sealed record Side(SqliteLocalDb Db, OplogStore Store, SyncSession Session);
+
+    private async Task<Side> NewSideAsync()
+    {
+        var path = Path.Combine(Path.GetTempPath(), Guid.NewGuid() + ".db");
+        _paths.Add(path);
+        var db = new SqliteLocalDb(new LocalDbOptions { Path = path });
+        _disposables.Add(db);
+        await db.ExecuteAsync(Ddl);
+        db.RegisterReplicated("q");
+
+        var options = new ReplicationOptions { FlushInterval = TimeSpan.FromMilliseconds(20) };
+        var store = new OplogStore(db, options);
+        var applier = new LwwApplier(db);
+        var session = new SyncSession(db, store, applier, options, NullLogger.Instance);
+        session.SnapshotSender = new SnapshotStreamer(db, store, options, NullLogger.Instance).SendAsync;
+        session.SnapshotApplier = new SnapshotApplier(db, applier, store, NullLogger.Instance);
+        return new Side(db, store, session);
+    }
+
+    private static (SyncDuplex A, SyncDuplex B) DuplexPair()
+    {
+        var aToB = Channel.CreateUnbounded<SyncMessage>();
+        var bToA = Channel.CreateUnbounded<SyncMessage>();
+        return (
+            new SyncDuplex { Send = (m, ct) => aToB.Writer.WriteAsync(m, ct).AsTask(), Inbox = bToA.Reader },
+            new SyncDuplex { Send = (m, ct) => bToA.Writer.WriteAsync(m, ct).AsTask(), Inbox = aToB.Reader });
+    }
+
+    private static async Task<List<string>> Rows(SqliteLocalDb db) =>
+        (await db.QueryAsync(
+            "SELECT \"o'key\" || '|' || \"it's\" || '|' || \"say \"\"hi\"\"\" FROM q ORDER BY \"o'key\"",
+            r => r.GetString(0))).ToList();
+
+    private static async Task<List<string>> DeadLetters(SqliteLocalDb db) =>
+        (await db.QueryAsync("SELECT error FROM __localdb_dead_letter", r => r.GetString(0))).ToList();
+
+    private static async Task WaitForAsync(Func<Task<bool>> 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);
+    }
+
+    private static async Task SwallowAsync(Task task)
+    {
+        try { await task; }
+        catch (OperationCanceledException) { }
+    }
+
+    [Fact]
+    public async Task Deltas_WithQuotesInColumnNames_ReplicateBothWays()
+    {
+        var a = await NewSideAsync();
+        var b = await NewSideAsync();
+        await a.Db.ExecuteAsync("INSERT INTO q VALUES ('k1', 'from A', 'a')");
+        await b.Db.ExecuteAsync("INSERT INTO q VALUES ('k2', 'from B', 'b')");
+
+        var (da, dbx) = DuplexPair();
+        using var cts = new CancellationTokenSource(RunTimeout);
+        var runA = a.Session.RunAsync(da, cts.Token);
+        var runB = b.Session.RunAsync(dbx, cts.Token);
+
+        string[] expected = ["k1|from A|a", "k2|from B|b"];
+        try
+        {
+            await WaitForAsync(async () =>
+                (await Rows(a.Db)).SequenceEqual(expected) && (await Rows(b.Db)).SequenceEqual(expected), RunTimeout);
+        }
+        catch (TimeoutException)
+        {
+            Assert.Fail($"Not converged. A dead-letters: [{string.Join("; ", await DeadLetters(a.Db))}], " +
+                        $"B dead-letters: [{string.Join("; ", await DeadLetters(b.Db))}]");
+        }
+
+        // An update and a delete exercise the PK-keyed json_extract paths too.
+        await a.Db.ExecuteAsync("UPDATE q SET \"it's\" = 'changed' WHERE \"o'key\" = 'k2'");
+        await b.Db.ExecuteAsync("DELETE FROM q WHERE \"o'key\" = 'k1'");
+        await WaitForAsync(async () =>
+            (await Rows(a.Db)).SequenceEqual(["k2|changed|b"]) && (await Rows(b.Db)).SequenceEqual(["k2|changed|b"]),
+            RunTimeout);
+
+        Assert.Empty(await DeadLetters(a.Db));
+        Assert.Empty(await DeadLetters(b.Db));
+
+        cts.Cancel();
+        await SwallowAsync(runA);
+        await SwallowAsync(runB);
+    }
+
+    [Fact]
+    public async Task Snapshot_WithAQuotedPrimaryKeyColumn_ReachesAWipedPeer()
+    {
+        var a = await NewSideAsync();
+        await a.Db.ExecuteAsync("INSERT INTO q VALUES ('k1', 'one', 'x')");
+        await a.Db.ExecuteAsync("INSERT INTO q VALUES ('k2', 'two', 'y')");
+        // Acked and pruned: a wiped peer can only be healed by a snapshot, whose page query joins
+        // the ledger to the table through the quoted PK column.
+        await a.Store.RecordPeerAckAsync(await a.Store.GetMaxSeqAsync());
+
+        var b = await NewSideAsync();
+        var (da, dbx) = DuplexPair();
+        using var cts = new CancellationTokenSource(RunTimeout);
+        var runA = a.Session.RunAsync(da, cts.Token);
+        var runB = b.Session.RunAsync(dbx, cts.Token);
+
+        await WaitForAsync(async () => (await Rows(b.Db)).Count == 2 || runA.IsCompleted, RunTimeout);
+
+        Assert.False(runA.IsFaulted, runA.Exception?.ToString());
+        Assert.Equal(new[] { "k1|one|x", "k2|two|y" }, await Rows(b.Db));
+        Assert.Empty(await DeadLetters(b.Db));
+
+        cts.Cancel();
+        await SwallowAsync(runA);
+        await SwallowAsync(runB);
+    }
+}

code-reviews/LocalDb/findings.md

@@ -10,7 +10,7 @@
 | Last reviewed | 2026-11-09 |
 | Reviewer | Claude |
 | Commit reviewed | `91fb2b4` |
-| Open findings | 3 |
+| Open findings | 2 |
 
 ## Summary
 
@@ -402,7 +402,7 @@ _Unresolved._
 | Severity | Low |
 | Priority | P3 |
 | Category | Correctness & logic bugs |
-| Status | Open |
+| Status | Resolved |
 | Location | `ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/LwwApplier.cs:349` |
 
 **Description**
@@ -432,7 +432,18 @@ test.
 
 **Resolution**
 
-_Unresolved._
+Resolved 2026-11-09 (scadaproj#46, LocalDb `0.3.6`, unpublished). Re-verified with a 2-node test: on a table
+`q("o'key" TEXT PRIMARY KEY, "it's" TEXT, "say ""hi""" TEXT)` both nodes dead-lettered every peer row with
+`SQLite Error 1: 'near "key": syntax error'`, and a wiped peer's snapshot faulted the sender's page query. The
+apply and snapshot paths now build their JSON paths through one shared helper, `JsonPathSql.For`: the column's `"`
+is backslash-escaped for the JSON path grammar, then every `'` in the path is doubled for the SQL string literal.
+`LwwApplier` and `SnapshotStreamer` no longer carry their own copies. Column names are not newly restricted at
+registration, so no existing schema is refused; the wire format is unchanged (the `row_json`/`pk_json` keys come from
+capture, which already escaped both quote styles).
+
+Tests (`QuotedColumnNameTests`): a 2-node delta round trip (insert both ways, then an update and a delete keyed on
+the quoted PK) and a wiped-peer snapshot through a quoted PK column. Both fail pre-fix and pass after. The table also carries a
+`"` column, which the old escape already handled.
 
 ### LocalDb-007 — README is stale on version, adoption, test count, `Synchronous` values and baseline scope