feat(localdb): snapshot resync (LWW-merging, tombstone-carrying, delta resume)

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

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

@@ -0,0 +1,171 @@
+using Microsoft.Extensions.Logging;
+using ZB.MOM.WW.LocalDb.Contracts;
+using ZB.MOM.WW.LocalDb.Internal;
+using ZB.MOM.WW.LocalDb.Registration;
+
+namespace ZB.MOM.WW.LocalDb.Replication.Internal;
+
+/// <summary>
+/// Consumes an inbound full-snapshot the receive loop routes to it: SnapshotBegin opens a receive,
+/// each SnapshotBatch merges through the SAME last-writer-wins path as deltas (never truncating —
+/// the receiver may hold newer local writes), and SnapshotComplete advances the durable watermark
+/// so incremental deltas resume above it.
+/// </summary>
+internal interface ISnapshotApplier
+{
+    Task OnBeginAsync(SnapshotBegin begin, CancellationToken ct);
+    Task OnBatchAsync(SnapshotBatch batch, CancellationToken ct);
+    Task OnCompleteAsync(SnapshotComplete complete, CancellationToken ct);
+}
+
+/// <summary>
+/// Streams a full snapshot of every registered table to a peer that has fallen behind the pruned
+/// oplog horizon. Reads the authoritative per-row version from <c>__localdb_row_version</c> joined
+/// to each live table, pages it into <see cref="SnapshotBatch"/> messages through the gated data
+/// lane, and brackets them with SnapshotBegin/SnapshotComplete carrying the max oplog seq as-of.
+/// </summary>
+internal sealed class SnapshotStreamer(SqliteLocalDb db, OplogStore store, ReplicationOptions options, ILogger logger)
+{
+    public async Task SendAsync(Func<SyncMessage, CancellationToken, Task> send, CancellationToken ct)
+    {
+        var asOfSeq = await store.GetMaxSeqAsync(ct);
+        await send(new SyncMessage { SnapshotBegin = new SnapshotBegin { AsOfSeq = asOfSeq } }, ct);
+
+        var totalRows = 0L;
+        foreach (var table in db.ReplicatedTables.Values.OrderBy(t => t.Name, StringComparer.Ordinal))
+        {
+            var sql = BuildPageSql(table);
+            var offset = 0;
+            while (true)
+            {
+                var rows = await db.QueryAsync(
+                    sql,
+                    // Column 4 is NULL for tombstones; 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);
+                if (rows.Count == 0)
+                    break;
+
+                var batch = new SnapshotBatch { TableName = table.Name };
+                foreach (var row in rows)
+                {
+                    var snapshotRow = new SnapshotRow
+                    {
+                        PkJson = row.PkJson,
+                        Hlc = row.Hlc,
+                        NodeId = row.NodeId,
+                        IsTombstone = row.IsTombstone,
+                    };
+                    if (row.RowJson is not null)
+                        snapshotRow.RowJson = row.RowJson;
+                    batch.Rows.Add(snapshotRow);
+                }
+                await send(new SyncMessage { SnapshotBatch = batch }, ct);
+                totalRows += rows.Count;
+
+                offset += rows.Count;
+                if (rows.Count < options.MaxBatchSize)
+                    break;
+            }
+        }
+
+        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.
+        await store.SetNeedsSnapshotAsync(false, ct);
+        logger.LogInformation("Snapshot sent (as_of_seq {AsOfSeq}, {Rows} rows).", asOfSeq, totalRows);
+    }
+
+    // Pages one table's rows by joining the row-version ledger to the live table on the PK columns
+    // extracted from pk_json. Rows present in the live table but absent from __localdb_row_version
+    // (data written BEFORE the table was registered for replication) are intentionally NOT snapshotted:
+    // they were never captured, consistent with CDC semantics — a full re-registration is required to
+    // baseline them.
+    private static string BuildPageSql(ReplicatedTable table)
+    {
+        var name = Ident(table.Name);
+        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)}")) + ")";
+        return
+            "SELECT rv.pk_json, rv.hlc, rv.node_id, rv.is_tombstone, " +
+            $"CASE WHEN rv.is_tombstone = 1 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";
+    }
+
+    private static string Ident(string name) => "\"" + name.Replace("\"", "\"\"") + "\"";
+
+    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
+    // still resolves: $."col".
+    private static string JsonPath(string column) => "'$.\"" + column.Replace("\"", "\\\"") + "\"'";
+}
+
+/// <summary>
+/// The inbound-snapshot consumer wired into <see cref="SyncSession"/>. Each batch merges through the
+/// shared <see cref="LwwApplier"/> (seq 0 — snapshot rows carry no oplog seq — so the applier's
+/// MAX-guarded watermark update is a no-op, and per-row LWW protects any newer local write). On
+/// complete, the remote watermark is set to the snapshot's as-of seq so deltas resume above it.
+/// </summary>
+internal sealed class SnapshotApplier : ISnapshotApplier
+{
+    private readonly LwwApplier _applier;
+    private readonly OplogStore _store;
+    private readonly ILogger _logger;
+    private long _asOfSeq;
+    private long _rows;
+    private long _maxHlc;
+
+    // db is part of the collaborator's canonical shape (mirrors the streamer) even though the merge
+    // reaches the database through the applier and store; kept for construction symmetry in the factory.
+    public SnapshotApplier(SqliteLocalDb db, LwwApplier applier, OplogStore store, ILogger logger)
+    {
+        _ = db;
+        _applier = applier;
+        _store = store;
+        _logger = logger;
+    }
+
+    public Task OnBeginAsync(SnapshotBegin begin, CancellationToken ct)
+    {
+        _asOfSeq = begin.AsOfSeq;
+        _rows = 0;
+        _maxHlc = 0;
+        _logger.LogInformation("Snapshot receive begun (as_of_seq {AsOfSeq}).", _asOfSeq);
+        return Task.CompletedTask;
+    }
+
+    public async Task OnBatchAsync(SnapshotBatch batch, CancellationToken ct)
+    {
+        if (batch.Rows.Count == 0)
+            return;
+
+        var entries = new List<OplogEntryRecord>(batch.Rows.Count);
+        foreach (var row in batch.Rows)
+        {
+            if (row.Hlc > _maxHlc) _maxHlc = row.Hlc;
+            entries.Add(new OplogEntryRecord(
+                0, batch.TableName, row.PkJson, row.HasRowJson ? row.RowJson : null, row.Hlc, row.NodeId, row.IsTombstone));
+        }
+        await _applier.ApplyBatchAsync(entries, ct);
+        _rows += batch.Rows.Count;
+    }
+
+    public async Task OnCompleteAsync(SnapshotComplete complete, CancellationToken ct)
+    {
+        // Preserve last_seen_hlc against a concurrently-applied delta that observed a higher HLC:
+        // SetLastAppliedRemoteSeqAsync overwrites the column, so pass the max of what we saw and
+        // what is already recorded.
+        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

@@ -39,6 +39,10 @@ internal sealed class SyncSession
     private string? _peerNodeId;
     private bool _driftWarned;
 
+    // Set true between an inbound SnapshotBegin and its SnapshotComplete. Only ever touched on the
+    // single receive-loop task, so no synchronization is needed.
+    private bool _snapshotReceiving;
+
     public SyncSession(
         SqliteLocalDb db, OplogStore store, LwwApplier applier, ReplicationOptions options, ILogger logger,
         Func<DateTimeOffset>? utcNow = null)
@@ -52,16 +56,18 @@ internal sealed class SyncSession
     }
 
     /// <summary>
-    /// Invoked when the peer's HandshakeAck requests a snapshot before deltas. Receives a gated
+    /// Invoked when THIS node owes the peer a snapshot (its own snapshot-required computation, not
-    /// send delegate that enqueues on the bounded data lane (never the raw transport). Task 12 fills this in.
+    /// 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"/>.
     /// </summary>
     public Func<Func<SyncMessage, CancellationToken, Task>, CancellationToken, Task>? SnapshotSender { get; set; }
 
     /// <summary>
-    /// Invoked on an inbound SnapshotBegin, with the triggering message and a gated data-lane send
+    /// Consumes an inbound snapshot inline on the receive loop: SnapshotBegin opens the receive,
-    /// delegate. Task 12 fills this in.
+    /// SnapshotBatch merges through LWW, SnapshotComplete advances the watermark. Wired to
+    /// <see cref="Internal.SnapshotApplier"/>.
     /// </summary>
-    public Func<SnapshotBegin, Func<SyncMessage, CancellationToken, Task>, CancellationToken, Task>? SnapshotReceiver { get; set; }
+    public ISnapshotApplier? SnapshotApplier { get; set; }
 
     /// <summary>Highest seq the peer has acknowledged applying this session.</summary>
     public long PeerAckedSeq => Interlocked.Read(ref _peerAckedSeq);
@@ -86,7 +92,7 @@ internal sealed class SyncSession
             await HandshakeAsync(duplex, control.Writer, GatedDataSend, linked.Token);
 
             var pump = PumpLoopAsync(data.Writer, linked.Token);
-            var receive = ReceiveLoopAsync(duplex, control.Writer, GatedDataSend, linked.Token);
+            var receive = ReceiveLoopAsync(duplex, control.Writer, linked.Token);
             var loops = new[] { pump, receive, writer };
 
             // The first COMPLETED task is the primary outcome. Cancel the siblings, observe every
@@ -142,7 +148,10 @@ internal sealed class SyncSession
             throw new InvalidOperationException(
                 $"Replication protocol error: expected a HandshakeAck, got {peerAckMsg.MsgCase}.");
 
-        if (peerAckMsg.HandshakeAck.SnapshotRequired)
+        // 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.
+        if (snapshotRequired)
         {
             if (SnapshotSender is null)
                 throw new NotSupportedException("snapshot required but no snapshot sender configured");
@@ -213,8 +222,7 @@ internal sealed class SyncSession
     }
 
     private async Task ReceiveLoopAsync(
-        SyncDuplex duplex, ChannelWriter<SyncMessage> controlLane,
+        SyncDuplex duplex, ChannelWriter<SyncMessage> controlLane, CancellationToken ct)
-        Func<SyncMessage, CancellationToken, Task> dataSend, CancellationToken ct)
     {
         while (await duplex.Inbox.WaitToReadAsync(ct))
         {
@@ -223,6 +231,7 @@ internal sealed class SyncSession
                 switch (msg.MsgCase)
                 {
                     case SyncMessage.MsgOneofCase.DeltaBatch:
+                        // Deltas apply normally even mid-snapshot: LWW makes the interleave safe.
                         await HandleDeltaBatchAsync(controlLane, msg.DeltaBatch, ct);
                         break;
                     case SyncMessage.MsgOneofCase.DeltaAck:
@@ -231,9 +240,23 @@ internal sealed class SyncSession
                         await _store.RecordPeerAckAsync(acked, ct);
                         break;
                     case SyncMessage.MsgOneofCase.SnapshotBegin:
-                        if (SnapshotReceiver is null)
+                        if (SnapshotApplier is null)
-                            throw new NotSupportedException("snapshot received but no snapshot receiver configured");
+                            throw new NotSupportedException("snapshot received but no snapshot applier configured");
-                        await SnapshotReceiver(msg.SnapshotBegin, dataSend, ct);
+                        _snapshotReceiving = true;
+                        await SnapshotApplier.OnBeginAsync(msg.SnapshotBegin, ct);
+                        break;
+                    case SyncMessage.MsgOneofCase.SnapshotBatch:
+                        if (!_snapshotReceiving)
+                            throw new InvalidOperationException(
+                                "Replication protocol error: SnapshotBatch with no active snapshot (missing SnapshotBegin).");
+                        await SnapshotApplier!.OnBatchAsync(msg.SnapshotBatch, ct);
+                        break;
+                    case SyncMessage.MsgOneofCase.SnapshotComplete:
+                        if (!_snapshotReceiving)
+                            throw new InvalidOperationException(
+                                "Replication protocol error: SnapshotComplete with no active snapshot (missing SnapshotBegin).");
+                        await SnapshotApplier!.OnCompleteAsync(msg.SnapshotComplete, ct);
+                        _snapshotReceiving = false;
                         break;
                     default:
                         throw new InvalidOperationException(

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

@@ -25,6 +25,11 @@ internal static class SyncSessionFactory
         var store = new OplogStore(db, options);
         var applier = new LwwApplier(db);
         var logger = loggerFactory.CreateLogger<SyncSession>();
-        return new SyncSession(db, store, applier, options, logger);
+        var session = new SyncSession(db, store, applier, options, logger);
+
+        var streamer = new SnapshotStreamer(db, store, options, loggerFactory.CreateLogger<SnapshotStreamer>());
+        session.SnapshotSender = streamer.SendAsync;
+        session.SnapshotApplier = new SnapshotApplier(db, applier, store, loggerFactory.CreateLogger<SnapshotApplier>());
+        return session;
     }
 }

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

@@ -0,0 +1,388 @@
+using System.Threading.Channels;
+using Microsoft.Extensions.Logging;
+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>
+/// Full-stack snapshot-resync coverage: a node whose peer fell behind the pruned oplog horizon
+/// streams a full snapshot of its registered tables; the receiver MERGES it through LWW (never
+/// truncates), tombstones are carried, and incremental deltas resume afterwards.
+/// </summary>
+public sealed class SnapshotResyncTests : IDisposable
+{
+    private const string OrdersSql = "CREATE TABLE orders (id INTEGER PRIMARY KEY, sku TEXT, qty INTEGER)";
+    private static readonly TimeSpan RunTimeout = TimeSpan.FromSeconds(15);
+
+    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, LwwApplier Applier, SyncSession Session);
+
+    // Builds a full replication stack with the snapshot hooks wired exactly as SyncSessionFactory does.
+    private async Task<Side> NewSideAsync(ReplicationOptions? options = null)
+    {
+        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(OrdersSql);
+        db.RegisterReplicated("orders");
+
+        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);
+
+        var streamer = new SnapshotStreamer(db, store, options, NullLogger.Instance);
+        var snapshotApplier = new SnapshotApplier(db, applier, store, NullLogger.Instance);
+        session.SnapshotSender = streamer.SendAsync;
+        session.SnapshotApplier = snapshotApplier;
+
+        return new Side(db, store, applier, session);
+    }
+
+    private static (SyncDuplex A, SyncDuplex B) DuplexPair()
+    {
+        var aToB = Channel.CreateUnbounded<SyncMessage>();
+        var bToA = Channel.CreateUnbounded<SyncMessage>();
+        var a = new SyncDuplex { Send = (m, ct) => aToB.Writer.WriteAsync(m, ct).AsTask(), Inbox = bToA.Reader };
+        var b = new SyncDuplex { Send = (m, ct) => bToA.Writer.WriteAsync(m, ct).AsTask(), Inbox = aToB.Reader };
+        return (a, b);
+    }
+
+    private static (SyncDuplex Duplex, ChannelWriter<SyncMessage> ToSession, ChannelReader<SyncMessage> FromSession) ScriptedPeer()
+    {
+        var toSession = Channel.CreateUnbounded<SyncMessage>();
+        var fromSession = Channel.CreateUnbounded<SyncMessage>();
+        var duplex = new SyncDuplex { Send = (m, ct) => fromSession.Writer.WriteAsync(m, ct).AsTask(), Inbox = toSession.Reader };
+        return (duplex, toSession.Writer, fromSession.Reader);
+    }
+
+    private static Handshake MatchingHandshake(SqliteLocalDb sessionDb, string peerNodeId, long lastAppliedRemoteSeq = 0)
+    {
+        var h = new Handshake { NodeId = peerNodeId, LibSchemaVersion = 1, LastAppliedRemoteSeq = lastAppliedRemoteSeq };
+        foreach (var t in sessionDb.ReplicatedTables.Values.OrderBy(t => t.Name, StringComparer.Ordinal))
+            h.Tables.Add(new TableDigest { TableName = t.Name, Digest = t.Digest });
+        return h;
+    }
+
+    private static async Task<SyncMessage> NextAsync(ChannelReader<SyncMessage> reader, CancellationToken ct)
+    {
+        await reader.WaitToReadAsync(ct);
+        reader.TryRead(out var msg);
+        return msg!;
+    }
+
+    private static async Task<SyncMessage> NextOfCaseAsync(
+        ChannelReader<SyncMessage> reader, SyncMessage.MsgOneofCase msgCase, CancellationToken ct)
+    {
+        SyncMessage msg;
+        do { msg = await NextAsync(reader, ct); } while (msg.MsgCase != msgCase);
+        return msg;
+    }
+
+    // Completes the two-phase handshake against a real session: the peer never requests a snapshot.
+    private static async Task DriveHandshakeAsync(
+        SqliteLocalDb sessionDb, ChannelWriter<SyncMessage> toSession, ChannelReader<SyncMessage> fromSession, CancellationToken ct)
+    {
+        await NextOfCaseAsync(fromSession, SyncMessage.MsgOneofCase.Handshake, ct);
+        await toSession.WriteAsync(new SyncMessage { Handshake = MatchingHandshake(sessionDb, "peer") }, ct);
+        await NextOfCaseAsync(fromSession, SyncMessage.MsgOneofCase.HandshakeAck, ct);
+        await toSession.WriteAsync(new SyncMessage { HandshakeAck = new HandshakeAck { NodeId = "peer", SnapshotRequired = false } }, ct);
+    }
+
+    private static async Task SwallowAsync(Task task)
+    {
+        try { await task; }
+        catch (OperationCanceledException) { }
+    }
+
+    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<List<(long Id, string? Sku, long? Qty)>> ReadOrders(SqliteLocalDb db)
+    {
+        var rows = await db.QueryAsync(
+            "SELECT id, sku, qty FROM orders ORDER BY id",
+            x => (x.GetInt64(0), x.IsDBNull(1) ? null : x.GetString(1), (long?)(x.IsDBNull(2) ? null : x.GetInt64(2))));
+        return rows.ToList();
+    }
+
+    private static async Task<long> LastAppliedRemoteSeq(SqliteLocalDb db)
+    {
+        var r = await db.QueryAsync("SELECT last_applied_remote_seq FROM __localdb_peer_state WHERE id = 1", x => x.GetInt64(0));
+        return r[0];
+    }
+
+    private static async Task<long> NeedsSnapshot(SqliteLocalDb db)
+    {
+        var r = await db.QueryAsync("SELECT needs_snapshot FROM __localdb_peer_state WHERE id = 1", x => x.GetInt64(0));
+        return r[0];
+    }
+
+    private static SnapshotRow SnapRow(long id, string sku, long hlc, string nodeId = "peer") =>
+        new()
+        {
+            PkJson = $"{{\"id\":{id}}}",
+            RowJson = $"{{\"id\":{id},\"sku\":\"{sku}\",\"qty\":1}}",
+            Hlc = hlc,
+            NodeId = nodeId,
+            IsTombstone = false,
+        };
+
+    [Fact]
+    public async Task PrunedPeer_GetsSnapshot_ThenConverges()
+    {
+        var a = await NewSideAsync(new ReplicationOptions { FlushInterval = TimeSpan.FromMilliseconds(20), MaxOplogRows = 2 });
+        var b = await NewSideAsync();
+        for (var i = 1; i <= 6; i++)
+            await a.Db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (@id, 'A', @id)", new { id = i });
+
+        // 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 (da, db) = DuplexPair();
+        using var cts = new CancellationTokenSource(RunTimeout);
+        var runA = a.Session.RunAsync(da, cts.Token);
+        var runB = b.Session.RunAsync(db, cts.Token);
+
+        await WaitForAsync(async () => (await ReadOrders(b.Db)).Count == 6, RunTimeout);
+        Assert.Equal(await ReadOrders(a.Db), await ReadOrders(b.Db));
+
+        // 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)");
+        await WaitForAsync(async () => (await ReadOrders(b.Db)).Any(r => r.Id == 7), RunTimeout);
+
+        cts.Cancel();
+        await SwallowAsync(runA);
+        await SwallowAsync(runB);
+    }
+
+    [Fact]
+    public async Task Snapshot_MergesNotTruncates()
+    {
+        var a = await NewSideAsync(new ReplicationOptions { FlushInterval = TimeSpan.FromMilliseconds(20), MaxOplogRows = 2 });
+        var b = await NewSideAsync();
+
+        await a.Db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'A_OLD', 1)");
+        await Task.Delay(5);
+        // B's row for the same pk is written later -> strictly higher HLC, so it must survive the merge.
+        await b.Db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'B_NEW', 9)");
+        // A owes B a snapshot; a single-row oplog would not trip the cap, so flag it directly.
+        await a.Store.SetNeedsSnapshotAsync(true);
+
+        var (da, db) = DuplexPair();
+        using var cts = new CancellationTokenSource(RunTimeout);
+        var runA = a.Session.RunAsync(da, cts.Token);
+        var runB = b.Session.RunAsync(db, cts.Token);
+
+        async Task<bool> Converged()
+        {
+            var ra = await ReadOrders(a.Db);
+            var rb = await ReadOrders(b.Db);
+            return ra.Count == 1 && rb.Count == 1 && ra[0].Sku == "B_NEW" && rb[0].Sku == "B_NEW";
+        }
+
+        await WaitForAsync(Converged, RunTimeout);
+
+        cts.Cancel();
+        await SwallowAsync(runA);
+        await SwallowAsync(runB);
+    }
+
+    [Fact]
+    public async Task Snapshot_CarriesTombstones()
+    {
+        var a = await NewSideAsync(new ReplicationOptions { FlushInterval = TimeSpan.FromMilliseconds(20), MaxOplogRows = 2 });
+        var b = await NewSideAsync();
+
+        await a.Db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'A1', 1), (2, 'A2', 2)");
+        await b.Db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'B1', 1), (2, 'B2', 2)");
+        await Task.Delay(5);
+        // A deletes id=1 last (newest HLC), so the snapshot's tombstone must win over B's live id=1.
+        await a.Db.ExecuteAsync("DELETE FROM orders WHERE id = 1");
+        Assert.True(await a.Store.EnforceCapsAsync());
+
+        var (da, db) = DuplexPair();
+        using var cts = new CancellationTokenSource(RunTimeout);
+        var runA = a.Session.RunAsync(da, cts.Token);
+        var runB = b.Session.RunAsync(db, cts.Token);
+
+        await WaitForAsync(async () =>
+        {
+            var rb = await ReadOrders(b.Db);
+            return rb.Count == 1 && rb[0].Id == 2;
+        }, RunTimeout);
+
+        Assert.DoesNotContain(await ReadOrders(b.Db), r => r.Id == 1);
+        Assert.DoesNotContain(await ReadOrders(a.Db), r => r.Id == 1);
+
+        cts.Cancel();
+        await SwallowAsync(runA);
+        await SwallowAsync(runB);
+    }
+
+    [Fact]
+    public async Task Snapshot_WatermarkNoOp_WhenSeqZero()
+    {
+        // Snapshot rows carry seq 0; applying them must not regress an already-advanced watermark.
+        var a = await NewSideAsync();
+        await a.Store.SetLastAppliedRemoteSeqAsync(50, 12345);
+
+        var entries = new List<OplogEntryRecord>
+        {
+            new(0, "orders", "{\"id\":1}", "{\"id\":1,\"sku\":\"S\",\"qty\":1}", 6_000_000, "peer", false),
+        };
+        await a.Applier.ApplyBatchAsync(entries, default);
+
+        Assert.Equal(50, await LastAppliedRemoteSeq(a.Db)); // MAX(50, 0) -> unchanged
+        Assert.Single(await ReadOrders(a.Db));
+    }
+
+    [Fact]
+    public async Task SnapshotComplete_AdvancesWatermark_DeltasResume()
+    {
+        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 { SnapshotBegin = new SnapshotBegin { AsOfSeq = 100 } }, 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 = 100 } }, cts.Token);
+
+        await WaitForAsync(async () => await LastAppliedRemoteSeq(a.Db) == 100, RunTimeout);
+        Assert.Single(await ReadOrders(a.Db));
+
+        // A normal delta above the snapshot horizon resumes incrementally.
+        await toSession.WriteAsync(new SyncMessage
+        {
+            DeltaBatch = new DeltaBatch { Entries = { new OplogEntry
+            {
+                Seq = 101, TableName = "orders", PkJson = "{\"id\":2}",
+                RowJson = "{\"id\":2,\"sku\":\"D\",\"qty\":1}", Hlc = 6_000_001, NodeId = "peer", IsTombstone = false,
+            } } },
+        }, cts.Token);
+        var ack = await NextOfCaseAsync(fromSession, SyncMessage.MsgOneofCase.DeltaAck, cts.Token);
+        Assert.Equal(101, ack.DeltaAck.AppliedThruSeq);
+        Assert.Equal(101, await LastAppliedRemoteSeq(a.Db));
+        Assert.Equal(2, (await ReadOrders(a.Db)).Count);
+
+        cts.Cancel();
+        await SwallowAsync(run);
+    }
+
+    [Fact]
+    public async Task SnapshotBatch_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
+        {
+            SnapshotBatch = new SnapshotBatch { TableName = "orders", Rows = { SnapRow(1, "ORPHAN", 6_000_000) } },
+        }, cts.Token);
+
+        await Assert.ThrowsAsync<InvalidOperationException>(() => run);
+    }
+
+    [Fact]
+    public async Task DeltaDuringSnapshot_AppliesSafely()
+    {
+        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 { SnapshotBegin = new SnapshotBegin { AsOfSeq = 200 } }, cts.Token);
+        await toSession.WriteAsync(new SyncMessage
+        {
+            SnapshotBatch = new SnapshotBatch { TableName = "orders", Rows = { SnapRow(1, "SNAP1", 6_000_000) } },
+        }, cts.Token);
+        // A normal delta interleaved MID-snapshot must still apply (LWW makes interleaving safe).
+        await toSession.WriteAsync(new SyncMessage
+        {
+            DeltaBatch = new DeltaBatch { Entries = { new OplogEntry
+            {
+                Seq = 205, TableName = "orders", PkJson = "{\"id\":2}",
+                RowJson = "{\"id\":2,\"sku\":\"DELTA\",\"qty\":1}", Hlc = 6_000_010, NodeId = "peer", IsTombstone = false,
+            } } },
+        }, cts.Token);
+        await NextOfCaseAsync(fromSession, SyncMessage.MsgOneofCase.DeltaAck, cts.Token);
+        await toSession.WriteAsync(new SyncMessage
+        {
+            SnapshotBatch = new SnapshotBatch { TableName = "orders", Rows = { SnapRow(3, "SNAP3", 6_000_020) } },
+        }, cts.Token);
+        await toSession.WriteAsync(new SyncMessage { SnapshotComplete = new SnapshotComplete { AsOfSeq = 200 } }, cts.Token);
+
+        await WaitForAsync(async () => (await ReadOrders(a.Db)).Count == 3, RunTimeout);
+        var rows = await ReadOrders(a.Db);
+        Assert.Equal([1L, 2L, 3L], rows.Select(r => r.Id));
+
+        cts.Cancel();
+        await SwallowAsync(run);
+    }
+
+    [Fact]
+    public async Task Sender_ClearsNeedsSnapshotFlag()
+    {
+        var a = await NewSideAsync(new ReplicationOptions { FlushInterval = TimeSpan.FromMilliseconds(20), MaxOplogRows = 2 });
+        var b = await NewSideAsync();
+        for (var i = 1; i <= 5; i++)
+            await a.Db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (@id, 'A', @id)", new { id = i });
+        Assert.True(await a.Store.EnforceCapsAsync());
+        Assert.Equal(1, await NeedsSnapshot(a.Db));
+
+        var (da, db) = DuplexPair();
+        using var cts = new CancellationTokenSource(RunTimeout);
+        var runA = a.Session.RunAsync(da, cts.Token);
+        var runB = b.Session.RunAsync(db, cts.Token);
+
+        await WaitForAsync(async () => (await ReadOrders(b.Db)).Count == 5, RunTimeout);
+        await WaitForAsync(async () => await NeedsSnapshot(a.Db) == 0, RunTimeout);
+
+        cts.Cancel();
+        await SwallowAsync(runA);
+        await SwallowAsync(runB);
+    }
+}

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

@@ -546,10 +546,11 @@ public sealed class SyncSessionTests : IDisposable
     [Fact]
     public async Task PeerNeedsSnapshot_InvokesSnapshotHooks()
     {
-        // (a) peer's HandshakeAck.snapshot_required = true -> our SnapshotSender fires, and its
+        // (a) THIS node owes the peer a snapshot (its own needs_snapshot) -> our SnapshotSender fires,
-        // gated send delegate routes messages through the writer onto the wire.
+        // and its gated send delegate routes messages through the writer onto the wire.
         {
             var a = await NewSideAsync();
+            await a.Store.SetNeedsSnapshotAsync(true, default);
             var (duplex, toSession, fromSession) = ScriptedPeer();
             var sent = new TaskCompletionSource();
             a.Session.SnapshotSender = (send, token) =>
@@ -564,7 +565,7 @@ public sealed class SyncSessionTests : IDisposable
             await NextAsync(fromSession, cts.Token); // session Handshake
             await toSession.WriteAsync(new SyncMessage { Handshake = MatchingHandshake(a.Db, "peer") }, cts.Token);
             await NextAsync(fromSession, cts.Token); // session HandshakeAck
-            await toSession.WriteAsync(new SyncMessage { HandshakeAck = new HandshakeAck { NodeId = "peer", SnapshotRequired = true } }, cts.Token);
+            await toSession.WriteAsync(new SyncMessage { HandshakeAck = new HandshakeAck { NodeId = "peer", SnapshotRequired = false } }, cts.Token);
 
             await sent.Task.WaitAsync(RunTimeout);
             var complete = await NextOfCaseAsync(fromSession, SyncMessage.MsgOneofCase.SnapshotComplete, cts.Token);
@@ -574,16 +575,12 @@ public sealed class SyncSessionTests : IDisposable
             await SwallowAsync(run);
         }
 
-        // (b) inbound SnapshotBegin -> our SnapshotReceiver fires with the triggering message.
+        // (b) inbound SnapshotBegin -> our ISnapshotApplier.OnBeginAsync fires with the triggering message.
         {
             var a = await NewSideAsync();
             var (duplex, toSession, fromSession) = ScriptedPeer();
-            var received = new TaskCompletionSource<long>();
+            var applier = new CapturingApplier();
-            a.Session.SnapshotReceiver = (begin, _, _) =>
+            a.Session.SnapshotApplier = applier;
-            {
-                received.TrySetResult(begin.AsOfSeq);
-                return Task.CompletedTask;
-            };
 
             using var cts = new CancellationTokenSource(RunTimeout);
             var run = a.Session.RunAsync(duplex, cts.Token);
@@ -591,14 +588,16 @@ public sealed class SyncSessionTests : IDisposable
             await DriveHandshakeAsync(a.Db, toSession, fromSession, cts.Token);
             await toSession.WriteAsync(new SyncMessage { SnapshotBegin = new SnapshotBegin { AsOfSeq = 42 } }, cts.Token);
 
-            Assert.Equal(42, await received.Task.WaitAsync(RunTimeout));
+            Assert.Equal(42, await applier.BeganAsOf.Task.WaitAsync(RunTimeout));
             cts.Cancel();
             await SwallowAsync(run);
         }
 
-        // (c) snapshot required but no sender hook -> NotSupportedException.
+        // (c) this node owes a snapshot but no sender hook -> NotSupportedException.
         {
             var a = await NewSideAsync();
+            await a.Store.SetNeedsSnapshotAsync(true, default);
+            a.Session.SnapshotSender = null;
             var (duplex, toSession, fromSession) = ScriptedPeer();
 
             using var cts = new CancellationTokenSource(RunTimeout);
@@ -607,14 +606,15 @@ public sealed class SyncSessionTests : IDisposable
             await NextAsync(fromSession, cts.Token);
             await toSession.WriteAsync(new SyncMessage { Handshake = MatchingHandshake(a.Db, "peer") }, cts.Token);
             await NextAsync(fromSession, cts.Token);
-            await toSession.WriteAsync(new SyncMessage { HandshakeAck = new HandshakeAck { NodeId = "peer", SnapshotRequired = true } }, cts.Token);
+            await toSession.WriteAsync(new SyncMessage { HandshakeAck = new HandshakeAck { NodeId = "peer", SnapshotRequired = false } }, cts.Token);
 
             await Assert.ThrowsAsync<NotSupportedException>(() => run);
         }
 
-        // (d) inbound SnapshotBegin but no receiver hook -> NotSupportedException.
+        // (d) inbound SnapshotBegin but no applier hook -> NotSupportedException.
         {
             var a = await NewSideAsync();
+            a.Session.SnapshotApplier = null;
             var (duplex, toSession, fromSession) = ScriptedPeer();
 
             using var cts = new CancellationTokenSource(RunTimeout);
@@ -627,6 +627,20 @@ public sealed class SyncSessionTests : IDisposable
         }
     }
 
+    private sealed class CapturingApplier : ISnapshotApplier
+    {
+        public TaskCompletionSource<long> BeganAsOf { get; } = new();
+
+        public Task OnBeginAsync(SnapshotBegin begin, CancellationToken ct)
+        {
+            BeganAsOf.TrySetResult(begin.AsOfSeq);
+            return Task.CompletedTask;
+        }
+
+        public Task OnBatchAsync(SnapshotBatch batch, CancellationToken ct) => Task.CompletedTask;
+        public Task OnCompleteAsync(SnapshotComplete complete, CancellationToken ct) => Task.CompletedTask;
+    }
+
     [Fact]
     public async Task SecondHandshake_MidStream_IsProtocolError()
     {