fix(localdb): snapshot review fixes — MAX-guarded complete watermark, keyset pagination, phantom-row guard

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

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

@@ -146,10 +146,15 @@ internal sealed class OplogStore(ILocalDb db, ReplicationOptions options, Func<D
     public Task SetNeedsSnapshotAsync(bool value, CancellationToken ct = default) =>
         db.ExecuteAsync("UPDATE __localdb_peer_state SET needs_snapshot = @v WHERE id = 1", new { v = value ? 1 : 0 }, ct);
 
-    /// <summary>Records the highest remote seq applied locally plus the peer HLC observed with it; also stamps <c>last_sync_utc</c>.</summary>
+    /// <summary>
+    /// Records the highest remote seq applied locally plus the peer HLC observed with it; also stamps
+    /// <c>last_sync_utc</c>. Both watermarks are MAX-guarded (monotonic): a snapshot completing with a
+    /// lower as-of seq than a concurrently-applied delta must never regress the durable watermark.
+    /// </summary>
     public Task SetLastAppliedRemoteSeqAsync(long seq, long lastSeenHlc, CancellationToken ct = default) =>
         db.ExecuteAsync(
-            "UPDATE __localdb_peer_state SET last_applied_remote_seq = @seq, last_seen_hlc = @hlc, last_sync_utc = @now WHERE id = 1",
+            "UPDATE __localdb_peer_state SET last_applied_remote_seq = MAX(last_applied_remote_seq, @seq), " +
+            "last_seen_hlc = MAX(last_seen_hlc, @hlc), last_sync_utc = @now WHERE id = 1",
             new { seq, hlc = lastSeenHlc, now = FormatUtc(_utcNow()) }, ct);
 
     private static string FormatUtc(DateTimeOffset value) =>

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

@@ -26,7 +26,8 @@ internal interface ISnapshotApplier
 /// </summary>
 internal sealed class SnapshotStreamer(SqliteLocalDb db, OplogStore store, ReplicationOptions options, ILogger logger)
 {
-    public async Task SendAsync(Func<SyncMessage, CancellationToken, Task> send, CancellationToken ct)
+    /// <summary>Streams the full snapshot; returns the as-of oplog seq it covers, so the session can skip already-covered tail deltas.</summary>
+    public async Task<long> SendAsync(Func<SyncMessage, CancellationToken, Task> send, CancellationToken ct)
     {
         var asOfSeq = await store.GetMaxSeqAsync(ct);
         await send(new SyncMessage { SnapshotBegin = new SnapshotBegin { AsOfSeq = asOfSeq } }, ct);
@@ -35,26 +36,40 @@ internal sealed class SnapshotStreamer(SqliteLocalDb db, OplogStore store, Repli
         foreach (var table in db.ReplicatedTables.Values.OrderBy(t => t.Name, StringComparer.Ordinal))
         {
             var sql = BuildPageSql(table);
-            var offset = 0;
+            // Keyset cursor (pk_json > last seen), not LIMIT/OFFSET: concurrent inserts/prunes
+            // between pages cannot shift the window, so no ledger row is skipped or re-sent.
+            var lastPk = "";
             while (true)
             {
                 var rows = await db.QueryAsync(
                     sql,
-                    // Column 4 is NULL for tombstones; a proto optional string cannot be assigned null,
+                    // Column 4 is NULL for tombstones AND for live rows whose joined table row is
-                    // so carry it as a nullable and only set row_json when present.
+                    // missing; a proto optional string cannot be assigned null, so carry it as a
+                    // nullable and only set row_json when present.
                     static r => (
                         PkJson: r.GetString(0),
                         Hlc: r.GetInt64(1),
                         NodeId: r.GetString(2),
                         IsTombstone: r.GetInt64(3) != 0,
                         RowJson: r.IsDBNull(4) ? null : r.GetString(4)),
-                    new { t = table.Name, page = options.MaxBatchSize, off = offset }, ct);
+                    new { t = table.Name, page = options.MaxBatchSize, lastPk }, ct);
                 if (rows.Count == 0)
                     break;
+                lastPk = rows[^1].PkJson;
 
                 var batch = new SnapshotBatch { TableName = table.Name };
                 foreach (var row in rows)
                 {
+                    if (!row.IsTombstone && row.RowJson is null)
+                    {
+                        // Ledger says live but the LEFT JOIN found no table row: ledger/table
+                        // divergence (e.g. an unreplicated manual delete). Emitting {"col":null,...}
+                        // would fabricate a phantom row on the peer, so skip it defensively.
+                        logger.LogWarning(
+                            "Snapshot skipping live-ledger row {Table} {PkJson}: no matching table row (ledger/table divergence).",
+                            batch.TableName, row.PkJson);
+                        continue;
+                    }
                     var snapshotRow = new SnapshotRow
                     {
                         PkJson = row.PkJson,
@@ -66,10 +81,10 @@ internal sealed class SnapshotStreamer(SqliteLocalDb db, OplogStore store, Repli
                         snapshotRow.RowJson = row.RowJson;
                     batch.Rows.Add(snapshotRow);
                 }
-                await send(new SyncMessage { SnapshotBatch = batch }, ct);
+                if (batch.Rows.Count > 0)
-                totalRows += rows.Count;
+                    await send(new SyncMessage { SnapshotBatch = batch }, ct);
+                totalRows += batch.Rows.Count;
 
-                offset += rows.Count;
                 if (rows.Count < options.MaxBatchSize)
                     break;
             }
@@ -77,9 +92,16 @@ internal sealed class SnapshotStreamer(SqliteLocalDb db, OplogStore store, Repli
 
         await send(new SyncMessage { SnapshotComplete = new SnapshotComplete { AsOfSeq = asOfSeq } }, ct);
         // The sender owed this snapshot (its own needs_snapshot); clearing it here — after the whole
-        // snapshot is on the wire — stops it re-streaming on the next reconnect.
+        // snapshot is on the wire — stops it re-streaming on the next reconnect. Narrow window: the
+        // messages are enqueued on the data lane, not yet transport-delivered, so a crash/drop after
+        // this clear could lose the snapshot with the flag already down. The reconnect handshake's
+        // gap check (peer watermark below the pruned horizon) re-detects that in every case where the
+        // oplog is non-empty; the sole undetectable corner is an EMPTY oplog (nothing to gap-check),
+        // where the peer stays behind until new local writes recreate a horizon. Accepted: closing it
+        // would need transport-level delivery acks for snapshot frames.
         await store.SetNeedsSnapshotAsync(false, ct);
         logger.LogInformation("Snapshot sent (as_of_seq {AsOfSeq}, {Rows} rows).", asOfSeq, totalRows);
+        return asOfSeq;
     }
 
     // Pages one table's rows by joining the row-version ledger to the live table on the PK columns
@@ -93,11 +115,15 @@ internal sealed class SnapshotStreamer(SqliteLocalDb db, OplogStore store, Repli
         var join = string.Join(" AND ", table.PkColumns.Select(p => $"t.{Ident(p)} = json_extract(rv.pk_json, {JsonPath(p)})"));
         var jsonObject = "json_object(" +
             string.Join(", ", table.Columns.Select(c => $"{StringLiteral(c.Name)}, t.{Ident(c.Name)}")) + ")";
+        // A live ledger row whose join found no table row (first PK column NULL — PK columns are
+        // never NULL on a real join hit) also yields NULL row json, so the caller can detect the
+        // divergence instead of shipping a fabricated all-NULL row.
+        var joinMiss = $"t.{Ident(table.PkColumns[0])} IS NULL";
         return
             "SELECT rv.pk_json, rv.hlc, rv.node_id, rv.is_tombstone, " +
-            $"CASE WHEN rv.is_tombstone = 1 THEN NULL ELSE {jsonObject} END " +
+            $"CASE WHEN rv.is_tombstone = 1 OR {joinMiss} THEN NULL ELSE {jsonObject} END " +
             $"FROM __localdb_row_version rv LEFT JOIN {name} t ON {join} " +
-            "WHERE rv.table_name = @t ORDER BY rv.pk_json LIMIT @page OFFSET @off";
+            "WHERE rv.table_name = @t AND rv.pk_json > @lastPk ORDER BY rv.pk_json LIMIT @page";
     }
 
     private static string Ident(string name) => "\"" + name.Replace("\"", "\"\"") + "\"";
@@ -161,11 +187,9 @@ internal sealed class SnapshotApplier : ISnapshotApplier
 
     public async Task OnCompleteAsync(SnapshotComplete complete, CancellationToken ct)
     {
-        // Preserve last_seen_hlc against a concurrently-applied delta that observed a higher HLC:
+        // SetLastAppliedRemoteSeqAsync is MAX-guarded, so a delta applied mid-snapshot with a seq
-        // SetLastAppliedRemoteSeqAsync overwrites the column, so pass the max of what we saw and
+        // above the snapshot's as-of (or an HLC above what the snapshot carried) is never regressed.
-        // what is already recorded.
+        await _store.SetLastAppliedRemoteSeqAsync(complete.AsOfSeq, _maxHlc, ct);
-        var lastSeenHlc = Math.Max(_maxHlc, (await _store.GetPeerStateAsync(ct)).LastSeenHlc);
-        await _store.SetLastAppliedRemoteSeqAsync(complete.AsOfSeq, lastSeenHlc, ct);
         _logger.LogInformation("Snapshot receive complete (as_of_seq {AsOfSeq}, {Rows} rows).", complete.AsOfSeq, _rows);
     }
 }

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

@@ -58,9 +58,10 @@ internal sealed class SyncSession
     /// <summary>
     /// Invoked when THIS node owes the peer a snapshot (its own snapshot-required computation, not
     /// the peer's) before deltas resume. Receives a gated send delegate that enqueues on the bounded
-    /// data lane (never the raw transport). Wired to <see cref="SnapshotStreamer.SendAsync"/>.
+    /// data lane (never the raw transport); returns the as-of oplog seq the snapshot covers so the
+    /// pump can skip already-covered tail deltas. Wired to <see cref="SnapshotStreamer.SendAsync"/>.
     /// </summary>
-    public Func<Func<SyncMessage, CancellationToken, Task>, CancellationToken, Task>? SnapshotSender { get; set; }
+    public Func<Func<SyncMessage, CancellationToken, Task>, CancellationToken, Task<long>>? SnapshotSender { get; set; }
 
     /// <summary>
     /// Consumes an inbound snapshot inline on the receive loop: SnapshotBegin opens the receive,
@@ -72,6 +73,9 @@ internal sealed class SyncSession
     /// <summary>Highest seq the peer has acknowledged applying this session.</summary>
     public long PeerAckedSeq => Interlocked.Read(ref _peerAckedSeq);
 
+    /// <summary>Highest oplog seq handed to the wire (or covered by a sent snapshot). Test/metric observation only.</summary>
+    internal long SentThruSeq => Interlocked.Read(ref _sentThruSeq);
+
     /// <summary>The peer node's identity, once the handshake has completed. Volatile: written on the session task, read from test/metric threads.</summary>
     public string? PeerNodeId
     {
@@ -151,11 +155,19 @@ internal sealed class SyncSession
         // We stream the snapshot when WE owe the peer one (computed from our pruned horizon /
         // needs_snapshot), then it lands on the peer's receive loop. The peer's own flag drives ITS
         // send direction symmetrically; both are safe to run because LWW merges either way.
+        // Memory profile of a MUTUAL snapshot: while we send here (receive loop not yet running),
+        // the peer's inbound frames — potentially its entire snapshot — buffer in the unbounded
+        // duplex inbox until our send completes. Worst case one full peer snapshot held in memory;
+        // acceptable for the small LocalDb datasets this library targets.
         if (snapshotRequired)
         {
             if (SnapshotSender is null)
                 throw new NotSupportedException("snapshot required but no snapshot sender configured");
-            await SnapshotSender(dataSend, ct);
+            var snapshotAsOfSeq = await SnapshotSender(dataSend, ct);
+            // The snapshot already carries every row state up to as-of; skipping the pump past it
+            // avoids re-sending tail deltas the peer would only discard via LWW.
+            if (snapshotAsOfSeq > _sentThruSeq)
+                _sentThruSeq = snapshotAsOfSeq;
         }
     }
 

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

@@ -167,6 +167,7 @@ public sealed class SnapshotResyncTests : IDisposable
         // Prune A's oplog below B's (empty) watermark and flag a resync: incremental deltas alone
         // can no longer carry rows 1..4 to B.
         Assert.True(await a.Store.EnforceCapsAsync());
+        var asOfSeq = await a.Store.GetMaxSeqAsync();
 
         var (da, db) = DuplexPair();
         using var cts = new CancellationTokenSource(RunTimeout);
@@ -175,6 +176,9 @@ public sealed class SnapshotResyncTests : IDisposable
 
         await WaitForAsync(async () => (await ReadOrders(b.Db)).Count == 6, RunTimeout);
         Assert.Equal(await ReadOrders(a.Db), await ReadOrders(b.Db));
+        // The snapshot covered everything through asOfSeq, so the pump starts above it (the surviving
+        // tail oplog rows are not re-sent as deltas).
+        Assert.True(a.Session.SentThruSeq >= asOfSeq, $"SentThruSeq {a.Session.SentThruSeq} < asOfSeq {asOfSeq}");
 
         // Deltas resume after the snapshot: a post-snapshot write on A reaches B incrementally.
         await a.Db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (7, 'POST', 7)");
@@ -323,6 +327,59 @@ public sealed class SnapshotResyncTests : IDisposable
         await Assert.ThrowsAsync<InvalidOperationException>(() => run);
     }
 
+    [Fact]
+    public async Task SnapshotComplete_WithoutBegin_IsProtocolError()
+    {
+        var a = await NewSideAsync();
+        var (duplex, toSession, fromSession) = ScriptedPeer();
+
+        using var cts = new CancellationTokenSource(RunTimeout);
+        var run = a.Session.RunAsync(duplex, cts.Token);
+
+        await DriveHandshakeAsync(a.Db, toSession, fromSession, cts.Token);
+        await toSession.WriteAsync(new SyncMessage { SnapshotComplete = new SnapshotComplete { AsOfSeq = 9 } }, cts.Token);
+
+        await Assert.ThrowsAsync<InvalidOperationException>(() => run);
+    }
+
+    [Fact]
+    public async Task SnapshotComplete_EmptyOplogAsOfZero_DoesNotRegress()
+    {
+        // A sender with an empty oplog streams as_of_seq 0; a receiver already at watermark 500 must
+        // keep 500 (MAX-guarded complete), or it would re-request everything on the next handshake.
+        var a = await NewSideAsync();
+        await a.Store.SetLastAppliedRemoteSeqAsync(500, 0);
+        var (duplex, toSession, fromSession) = ScriptedPeer();
+
+        using var cts = new CancellationTokenSource(RunTimeout);
+        var run = a.Session.RunAsync(duplex, cts.Token);
+
+        await DriveHandshakeAsync(a.Db, toSession, fromSession, cts.Token);
+        await toSession.WriteAsync(new SyncMessage { SnapshotBegin = new SnapshotBegin { AsOfSeq = 0 } }, cts.Token);
+        await toSession.WriteAsync(new SyncMessage
+        {
+            SnapshotBatch = new SnapshotBatch { TableName = "orders", Rows = { SnapRow(1, "SNAP", 6_000_000) } },
+        }, cts.Token);
+        await toSession.WriteAsync(new SyncMessage { SnapshotComplete = new SnapshotComplete { AsOfSeq = 0 } }, cts.Token);
+
+        // Barrier (seq 0 does not move the watermark): its ack proves Complete{0} was processed.
+        await toSession.WriteAsync(new SyncMessage
+        {
+            DeltaBatch = new DeltaBatch { Entries = { new OplogEntry
+            {
+                Seq = 0, TableName = "orders", PkJson = "{\"id\":2}",
+                RowJson = "{\"id\":2,\"sku\":\"BARRIER\",\"qty\":1}", Hlc = 6_000_001, NodeId = "peer", IsTombstone = false,
+            } } },
+        }, cts.Token);
+        await NextOfCaseAsync(fromSession, SyncMessage.MsgOneofCase.DeltaAck, cts.Token);
+
+        Assert.Equal(500, await LastAppliedRemoteSeq(a.Db));
+        Assert.Equal(2, (await ReadOrders(a.Db)).Count); // the snapshot row still merged
+
+        cts.Cancel();
+        await SwallowAsync(run);
+    }
+
     [Fact]
     public async Task DeltaDuringSnapshot_AppliesSafely()
     {
@@ -359,6 +416,19 @@ public sealed class SnapshotResyncTests : IDisposable
         var rows = await ReadOrders(a.Db);
         Assert.Equal([1L, 2L, 3L], rows.Select(r => r.Id));
 
+        // Barrier (seq 0 does not move the watermark): its ack proves Complete{200} was processed;
+        // the MAX-guarded complete must not have regressed the delta's 205.
+        await toSession.WriteAsync(new SyncMessage
+        {
+            DeltaBatch = new DeltaBatch { Entries = { new OplogEntry
+            {
+                Seq = 0, TableName = "orders", PkJson = "{\"id\":4}",
+                RowJson = "{\"id\":4,\"sku\":\"BARRIER\",\"qty\":1}", Hlc = 6_000_030, NodeId = "peer", IsTombstone = false,
+            } } },
+        }, cts.Token);
+        await NextOfCaseAsync(fromSession, SyncMessage.MsgOneofCase.DeltaAck, cts.Token);
+        Assert.Equal(205, await LastAppliedRemoteSeq(a.Db));
+
         cts.Cancel();
         await SwallowAsync(run);
     }

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

@@ -553,10 +553,11 @@ public sealed class SyncSessionTests : IDisposable
             await a.Store.SetNeedsSnapshotAsync(true, default);
             var (duplex, toSession, fromSession) = ScriptedPeer();
             var sent = new TaskCompletionSource();
-            a.Session.SnapshotSender = (send, token) =>
+            a.Session.SnapshotSender = async (send, token) =>
             {
                 sent.TrySetResult();
-                return send(new SyncMessage { SnapshotComplete = new SnapshotComplete { AsOfSeq = 7 } }, token);
+                await send(new SyncMessage { SnapshotComplete = new SnapshotComplete { AsOfSeq = 7 } }, token);
+                return 7;
             };
 
             using var cts = new CancellationTokenSource(RunTimeout);