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

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");

        return NewSessionOver(db, options);
    }

    // A fresh session stack over an EXISTING database. Reconnecting after a session ends needs a new
    // session (RunAsync is single-use), so any scenario with two connection episodes over one node's
    // data — a peer being replaced between them — goes through here.
    private static Side NewSessionOver(SqliteLocalDb db, ReplicationOptions? options = null)
    {
        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);
    }

    // capacity models a transport that applies backpressure (which the gRPC duplex now is, via its
    // bounded inbox); null is the historical unbounded pipe.
    private static (SyncDuplex A, SyncDuplex B) DuplexPair(int? capacity = null)
    {
        Channel<SyncMessage> New() => capacity is int c
            ? Channel.CreateBounded<SyncMessage>(c)
            : Channel.CreateUnbounded<SyncMessage>();
        var aToB = New();
        var bToA = New();
        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 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 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 MutualSnapshot_OverABoundedTransport_ConvergesWithoutDeadlock()
    {
        // BOTH sides owe a snapshot — the ordinary outcome of enabling replication late on a pair,
        // since baselining flags needs_snapshot on whichever node baselined and both nodes do.
        //
        // The snapshot used to be streamed from INSIDE the handshake, before the receive loop
        // existed, so neither side drained its inbox while sending. That exchange completed only
        // because the inbox was unbounded and could hold the peer's ENTIRE snapshot in memory —
        // which is the memory profile this design gave up on. Over a transport that pushes back
        // (which the gRPC duplex now is, with its bounded inbox) the old shape deadlocks outright.
        // Streaming the snapshot alongside an already-running receive loop is what makes both the
        // bound and this test possible.
        var options = new ReplicationOptions { MaxBatchSize = 2, FlushInterval = TimeSpan.FromMilliseconds(20) };
        var a = await NewSideAsync(options);
        var b = await NewSideAsync(options);

        const int rowsPerSide = 60;
        for (var i = 1; i <= rowsPerSide; i++)
        {
            await a.Db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (@id, 'A', @id)", new { id = i });
            await b.Db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (@id, 'B', @id)", new { id = 100 + i });
        }
        await a.Store.SetNeedsSnapshotAsync(true, default);
        await b.Store.SetNeedsSnapshotAsync(true, default);

        // Capacity 1 each way: the tightest possible bound, so a side that stops draining its inbox
        // to send stalls at once instead of getting away with it on buffer slack.
        var (da, db) = DuplexPair(capacity: 1);
        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(a.Db)).Count == 2 * rowsPerSide
                && (await ReadOrders(b.Db)).Count == 2 * rowsPerSide,
            RunTimeout);
        Assert.Equal(await ReadOrders(a.Db), await ReadOrders(b.Db));

        cts.Cancel();
        await SwallowAsync(runA);
        await SwallowAsync(runB);
    }

    [Fact]
    public async Task WipedPeer_AfterEverythingWasAckedAndPruned_StillGetsSnapshot()
    {
        // The gap the OtOpcUa Phase 1 live gate found (check 4). A converged pair prunes its oplog
        // on ack, so the steady state is "no oplog at all". If the peer is then wiped — a rebuilt
        // node, a lost volume — it comes back with an empty database and a zero watermark, and the
        // healthy side has no delta left to send it. Snapshot detection compared the peer's
        // watermark against the oldest oplog row, which does not exist, so it concluded no snapshot
        // was needed and the wiped node stayed empty indefinitely: the one case where replication
        // most obviously should heal a node was the case it could not.
        var a = await NewSideAsync();
        for (var i = 1; i <= 3; i++)
            await a.Db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (@id, 'A', @id)", new { id = i });

        await a.Store.RecordPeerAckAsync(await a.Store.GetMaxSeqAsync());
        Assert.Equal(0, await OplogCount(a.Db));
        // Not the already-covered cap-exceeded path: nothing here flagged a resync.
        Assert.Equal(0, await NeedsSnapshot(a.Db));

        // A brand-new database is exactly what a wiped node is: no rows, watermark zero.
        var b = await NewSideAsync();
        Assert.Empty(await ReadOrders(b.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 == 3, RunTimeout);
        Assert.Equal(await ReadOrders(a.Db), await ReadOrders(b.Db));

        cts.Cancel();
        await SwallowAsync(runA);
        await SwallowAsync(runB);
    }

    [Fact]
    public async Task RebuiltPeer_OwnWritesReachTheHealthyNode_DespiteTheOldPeersWatermark()
    {
        // The mirror image of the wiped-peer snapshot gap, found on the OtOpcUa docker-dev rig once
        // back-fill started working and the rebuilt node could be observed writing again.
        //
        // last_applied_remote_seq is a watermark in the PEER's seq space. A rebuilt peer is a new
        // database, so its seqs restart at 1 — but the healthy node kept the old peer's watermark
        // and discarded everything at or below it. The rebuilt node's own writes were silently
        // dropped until its counter climbed past a number that has nothing to do with it. Silent is
        // the operative word: both nodes cache after every apply, so the pair looks converged right
        // up until the moment only the rebuilt node applies something.
        var a = await NewSideAsync();
        var bOld = await NewSideAsync();

        // Phase 1: converge with the ORIGINAL peer, so A's inbound watermark advances well past the
        // seq numbers a rebuilt peer would start from.
        for (var i = 1; i <= 5; i++)
            await bOld.Db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (@id, 'OLD', @id)", new { id = i });

        var (d1A, d1B) = DuplexPair();
        using (var cts1 = new CancellationTokenSource(RunTimeout))
        {
            var run1A = a.Session.RunAsync(d1A, cts1.Token);
            var run1B = bOld.Session.RunAsync(d1B, cts1.Token);
            await WaitForAsync(async () => (await ReadOrders(a.Db)).Count == 5, RunTimeout);
            cts1.Cancel();
            await SwallowAsync(run1A);
            await SwallowAsync(run1B);
        }

        Assert.True(await LastAppliedRemoteSeq(a.Db) >= 5);

        // Phase 2: B is rebuilt — a new database, a new node id, and an oplog numbering from 1. Its
        // write must still reach A.
        var bNew = await NewSideAsync();
        await bNew.Db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (99, 'REBUILT', 99)");

        var a2 = NewSessionOver(a.Db);
        var (d2A, d2B) = DuplexPair();
        using var cts2 = new CancellationTokenSource(RunTimeout);
        var run2A = a2.Session.RunAsync(d2A, cts2.Token);
        var run2B = bNew.Session.RunAsync(d2B, cts2.Token);

        await WaitForAsync(
            async () => (await ReadOrders(a.Db)).Any(r => r.Id == 99 && r.Sku == "REBUILT"),
            RunTimeout);

        cts2.Cancel();
        await SwallowAsync(run2A);
        await SwallowAsync(run2B);
    }

    [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 asOfSeq = await a.Store.GetMaxSeqAsync();

        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));
        // 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)");
        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 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()
    {
        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));

        // 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);
    }

    [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);
    }
}