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, ILogger? streamerLogger = 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, streamerLogger ?? 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);
    }

    // supportsPartialSnapshot defaults to FALSE, which is exactly what a 0.1.x/0.2.x peer puts on
    // the wire (the field does not exist in its proto, so proto3 decodes the default). Scripted
    // peers are therefore old peers unless a test says otherwise.
    private static Handshake MatchingHandshake(
        SqliteLocalDb sessionDb, string peerNodeId, long lastAppliedRemoteSeq = 0, bool supportsPartialSnapshot = false)
    {
        var h = new Handshake
        {
            NodeId = peerNodeId,
            LibSchemaVersion = 1,
            LastAppliedRemoteSeq = lastAppliedRemoteSeq,
            SupportsPartialSnapshot = supportsPartialSnapshot,
        };
        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, bool supportsPartialSnapshot = false, long lastAppliedRemoteSeq = 0)
    {
        await NextOfCaseAsync(fromSession, SyncMessage.MsgOneofCase.Handshake, ct);
        await toSession.WriteAsync(
            new SyncMessage
            {
                Handshake = MatchingHandshake(
                    sessionDb, "peer", lastAppliedRemoteSeq, supportsPartialSnapshot),
            }, ct);
        await NextOfCaseAsync(fromSession, SyncMessage.MsgOneofCase.HandshakeAck, ct);
        await toSession.WriteAsync(new SyncMessage { HandshakeAck = new HandshakeAck { NodeId = "peer", SnapshotRequired = false } }, ct);
    }

    // Reads outbound messages until (and including) the first of the given kind.
    private static async Task<List<SyncMessage>> DrainThroughAsync(
        ChannelReader<SyncMessage> reader, SyncMessage.MsgOneofCase last, CancellationToken ct)
    {
        var all = new List<SyncMessage>();
        while (true)
        {
            var msg = await NextAsync(reader, ct);
            all.Add(msg);
            if (msg.MsgCase == last)
                return all;
        }
    }

    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];
    }

    /// <summary>The tables individually flagged as owing the peer a snapshot (schema v3).</summary>
    private static async Task<List<string>> PendingSnapshotTables(SqliteLocalDb db) =>
        (await db.QueryAsync(
            "SELECT table_name FROM __localdb_snapshot_state WHERE needs_snapshot <> 0 ORDER BY table_name",
            x => x.GetString(0))).ToList();

    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 WipedPeer_SnapshotLiftsItsWatermark_SoTheNextReconnectSendsNoSecondSnapshot()
    {
        // The follow-on to the wiped-peer gap above, found on the ScadaBridge docker rig 2026-08-20
        // (its deferred-work register row 106): the snapshot DID heal the wiped node, but it was
        // stamped as_of_seq = MAX(seq) of an oplog that ack-pruning had emptied — i.e. 0 — so the
        // receiver's (correctly MAX-guarded) watermark stayed at 0. On the next handshake the
        // healthy side's empty-oplog horizon rule (streamable from last_acked_seq + 1) saw
        // 0 < last_acked_seq as a gap again and re-streamed the whole database — on EVERY reconnect,
        // until the healthy side happened to write a new row (127k rows per restart on the rig).
        // A full snapshot taken now covers every delta ever pruned below last_acked_seq, so that is
        // the floor the as-of must carry.
        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));

        var b = await NewSideAsync();

        // Episode 1: the wiped peer is healed by a full snapshot.
        var (d1A, d1B) = DuplexPair();
        using (var cts1 = new CancellationTokenSource(RunTimeout))
        {
            var run1A = a.Session.RunAsync(d1A, cts1.Token);
            var run1B = b.Session.RunAsync(d1B, cts1.Token);
            await WaitForAsync(async () => (await ReadOrders(b.Db)).Count == 3, RunTimeout);
            cts1.Cancel();
            await SwallowAsync(run1A);
            await SwallowAsync(run1B);
        }

        // The snapshot must have lifted B's watermark to everything A ever produced (seq 3), not 0.
        Assert.Equal(3, await LastAppliedRemoteSeq(b.Db));

        // Episode 2: a routine reconnect. A writes one more row so the delta path gives us a
        // barrier; the snapshot streamer's logger proves whether a snapshot was sent at all.
        var streamerLog = new CapturingLogger();
        var a2 = NewSessionOver(a.Db, streamerLogger: streamerLog);
        var b2 = NewSessionOver(b.Db);
        var (d2A, d2B) = DuplexPair();
        using var cts2 = new CancellationTokenSource(RunTimeout);
        var run2A = a2.Session.RunAsync(d2A, cts2.Token);
        var run2B = b2.Session.RunAsync(d2B, cts2.Token);
        await WaitForAsync(() => Task.FromResult(b2.Session.PeerNodeId is not null), RunTimeout);
        await a.Db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (4, 'DELTA', 4)");
        await WaitForAsync(async () => (await ReadOrders(b.Db)).Count == 4, RunTimeout);

        Assert.DoesNotContain(streamerLog.Messages, m => m.StartsWith("Snapshot sent", StringComparison.Ordinal));
        Assert.Equal(4, await LastAppliedRemoteSeq(b.Db));

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

    private sealed class CapturingLogger : ILogger
    {
        private readonly List<string> _messages = [];
        public IReadOnlyList<string> Messages { get { lock (_messages) return _messages.ToList(); } }
        public IDisposable? BeginScope<TState>(TState state) where TState : notnull => null;
        public bool IsEnabled(LogLevel logLevel) => true;
        public void Log<TState>(LogLevel logLevel, EventId eventId, TState state, Exception? exception, Func<TState, Exception?, string> formatter)
        {
            lock (_messages) _messages.Add(formatter(state, exception));
        }
    }

    [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 CapPrune_IntoALiveSessionsUnsentRange_FaultsIt_AndTheReconnectSnapshots()
    {
        // P4-1. The pump reads `seq > sentThruSeq`, which resumes SILENTLY above a hole a cap prune
        // punched — so it keeps shipping later deltas, the peer acks them, and the pair is durably
        // divergent. `needs_snapshot` (set in the prune's own transaction) heals only a RECONNECT,
        // and nothing forced one: on a stable LAN with a slow-but-alive peer the session, and the
        // divergence, could last for days. Faulting is what turns the persisted flag into a snapshot.
        var a = 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 });

        var (duplex, toSession, fromSession) = ScriptedPeer();
        using var cts = new CancellationTokenSource(RunTimeout);
        var run = a.Session.RunAsync(duplex, cts.Token);

        // The peer holds everything so far: no snapshot is owed, and the pump idles at seq 6.
        await DriveHandshakeAsync(a.Db, toSession, fromSession, cts.Token, lastAppliedRemoteSeq: 6);

        // Write three more rows and prune all but the newest, in ONE transaction — which is what a
        // cap prune is, and what makes this deterministic: the pump's reader can only ever see the
        // before state (nothing above seq 6) or the after state (seqs 7 and 8 already gone). It can
        // never catch rows 7/8 in between and legitimately advance past them.
        await using (var tx = await a.Db.BeginTransactionAsync(cts.Token))
        {
            await tx.ExecuteAsync(
                "INSERT INTO orders (id, sku, qty) VALUES (7,'PRUNED',7),(8,'PRUNED',8),(9,'KEPT',9)", null, cts.Token);
            await tx.ExecuteAsync("UPDATE __localdb_peer_state SET needs_snapshot = 1 WHERE id = 1", null, cts.Token);
            await tx.ExecuteAsync(
                "DELETE FROM __localdb_oplog WHERE seq < (SELECT MAX(seq) FROM __localdb_oplog)", null, cts.Token);
            await tx.CommitAsync(cts.Token);
        }

        var fault = await Assert.ThrowsAsync<LocalDbOplogGapException>(() => run);
        Assert.Contains("6", fault.Message, StringComparison.Ordinal);

        // And it faulted BEFORE shipping anything from above the hole: a delta for seq 9 would have
        // moved the peer's watermark past rows 7 and 8, which is the divergence itself.
        while (fromSession.TryRead(out var stray))
            Assert.NotEqual(SyncMessage.MsgOneofCase.DeltaBatch, stray.MsgCase);

        // The reconnect is what heals: the prune's needs_snapshot makes the handshake owe a full
        // snapshot, which carries the pruned rows' current state to a real peer.
        var a2 = NewSessionOver(a.Db);
        var b = await NewSideAsync();
        var (da, db) = DuplexPair();
        var run2A = a2.Session.RunAsync(da, cts.Token);
        var run2B = b.Session.RunAsync(db, cts.Token);

        await WaitForAsync(async () => (await ReadOrders(b.Db)).Count == 9, RunTimeout);
        Assert.Equal(await ReadOrders(a.Db), await ReadOrders(b.Db));

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

    [Fact]
    public async Task AckPruningToEmpty_DoesNotFaultALiveSession()
    {
        // The other half of the horizon check: ordinary ack-pruning deletes seq <= last_acked_seq,
        // which by construction sits at or below the sent watermark, so a converged pair that has
        // pruned its whole oplog must keep streaming. A gap test written against `needs_snapshot`,
        // or one that treated any risen horizon as a fault, would kill healthy sessions here.
        var a = await NewSideAsync();
        var b = await NewSideAsync();
        for (var i = 1; i <= 4; i++)
            await a.Db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (@id, 'A', @id)", new { id = i });

        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 == 4, RunTimeout);
        await WaitForAsync(async () => await OplogCount(a.Db) == 0, RunTimeout);

        // Still live, still incremental: a post-prune write reaches B as an ordinary delta.
        await a.Db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (5, 'POST', 5)");
        await WaitForAsync(async () => (await ReadOrders(b.Db)).Any(r => r.Id == 5), RunTimeout);
        Assert.False(runA.IsFaulted, runA.Exception?.ToString());

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

    // ---- Per-table snapshot scoping (0.3.0) ----------------------------------------------

    private const string WidgetsSql = "CREATE TABLE widgets (id INTEGER PRIMARY KEY, name TEXT)";

    /// <summary>
    /// A node with two registered tables where exactly one — <c>orders</c> — was baselined.
    /// <c>widgets</c> is registered BEFORE its rows are written, so ordinary capture ledgers them;
    /// <c>orders</c> is registered after, so only baselining can. That asymmetry is what makes the
    /// scope observable: a full snapshot carries both tables' ledgers, a scoped one carries orders
    /// alone — and both tables hold rows either way, so an empty widgets stream cannot be mistaken
    /// for "there was nothing to send".
    /// </summary>
    private async Task<Side> NewBaselinedSideAsync(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);
        await db.ExecuteAsync(WidgetsSql);

        // Predates registration: invisible to CDC, so only baselining puts it in the ledger.
        await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1,'PRE',1),(2,'PRE',2)");

        db.RegisterReplicated("widgets");
        await db.ExecuteAsync("INSERT INTO widgets (id, name) VALUES (1,'W1'),(2,'W2')");

        db.RegisterReplicated("orders", baselineExistingRows: true);

        return NewSessionOver(db, options);
    }

    /// <summary>The peer of <see cref="NewBaselinedSideAsync"/>: the same two tables, empty, owing nothing.</summary>
    private async Task<Side> NewTwoTableSideAsync(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);
        await db.ExecuteAsync(WidgetsSql);
        db.RegisterReplicated("orders");
        db.RegisterReplicated("widgets");
        return NewSessionOver(db, options);
    }

    private static List<string> SnapshotTablesIn(IEnumerable<SyncMessage> messages) =>
        messages.Where(m => m.MsgCase == SyncMessage.MsgOneofCase.SnapshotBatch)
            .Select(m => m.SnapshotBatch.TableName)
            .Distinct(StringComparer.Ordinal)
            .Order(StringComparer.Ordinal)
            .ToList();

    [Fact]
    public async Task Baselining_OneTable_StreamsOnlyThatTable()
    {
        // The whole point of the per-table flag. Through 0.2.x the flag was one bit for the whole
        // database, so baselining `orders` re-streamed `widgets` (and everything else) in full — in
        // both directions at once on a pair enabling replication late, which on a database holding
        // an event log at its retention cap is the largest transfer this library performs.
        var a = await NewBaselinedSideAsync();
        Assert.Equal(["orders"], await PendingSnapshotTables(a.Db));
        Assert.Equal(0, await NeedsSnapshot(a.Db)); // NOT the database-wide flag

        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, supportsPartialSnapshot: true);
        var frames = await DrainThroughAsync(fromSession, SyncMessage.MsgOneofCase.SnapshotComplete, cts.Token);

        var begin = frames.Single(m => m.MsgCase == SyncMessage.MsgOneofCase.SnapshotBegin).SnapshotBegin;
        Assert.Equal(["orders"], begin.Tables);
        Assert.Equal(["orders"], SnapshotTablesIn(frames));
        // The baselined rows really are in there — the scope narrowed the snapshot, it did not empty it.
        var pks = frames.Where(m => m.MsgCase == SyncMessage.MsgOneofCase.SnapshotBatch)
            .SelectMany(m => m.SnapshotBatch.Rows).Select(r => r.PkJson).Order(StringComparer.Ordinal);
        Assert.Equal(["{\"id\":1}", "{\"id\":2}"], pks);

        cts.Cancel();
        await SwallowAsync(run);
    }

    [Fact]
    public async Task Baselining_AgainstAPeerWithoutPartialSupport_StreamsEveryTable()
    {
        // Mixed-version interop, the hard constraint. A pre-0.3.0 peer ignores SnapshotBegin.tables
        // and would advance last_applied_remote_seq to as_of_seq on complete — which for a SCOPED
        // snapshot silently discards every delta below as_of for the tables it left out. There is no
        // way to express "partial" to that peer, so it is served the 0.2.x answer: everything.
        var a = await NewBaselinedSideAsync();
        var (duplex, toSession, fromSession) = ScriptedPeer();
        using var cts = new CancellationTokenSource(RunTimeout);
        var run = a.Session.RunAsync(duplex, cts.Token);

        // supportsPartialSnapshot omitted == what a 0.1.x/0.2.x handshake decodes to.
        await DriveHandshakeAsync(a.Db, toSession, fromSession, cts.Token);
        var frames = await DrainThroughAsync(fromSession, SyncMessage.MsgOneofCase.SnapshotComplete, cts.Token);

        var begin = frames.Single(m => m.MsgCase == SyncMessage.MsgOneofCase.SnapshotBegin).SnapshotBegin;
        Assert.Empty(begin.Tables); // empty == full, the only scope the old wire can express
        Assert.Equal(["orders", "widgets"], SnapshotTablesIn(frames));

        cts.Cancel();
        await SwallowAsync(run);
    }

    [Fact]
    public async Task PartialSnapshot_DebtClearsOnlyWhenThePeerAcknowledges()
    {
        // Crash safety: an ungraceful exit part-way through a snapshot must re-stream it, so the
        // flag cannot come down when the frames are merely enqueued. The peer's SnapshotAck is
        // written after its own durable watermark write, so it means "I hold these rows".
        var a = await NewBaselinedSideAsync();
        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, supportsPartialSnapshot: true);
        var frames = await DrainThroughAsync(fromSession, SyncMessage.MsgOneofCase.SnapshotComplete, cts.Token);
        var complete = frames[^1].SnapshotComplete;

        // The entire snapshot is on the wire and unacknowledged: the debt must still stand.
        Assert.Equal(["orders"], await PendingSnapshotTables(a.Db));

        await toSession.WriteAsync(new SyncMessage { SnapshotAck = new SnapshotAck { AsOfSeq = complete.AsOfSeq } }, cts.Token);
        await WaitForAsync(async () => (await PendingSnapshotTables(a.Db)).Count == 0, RunTimeout);

        cts.Cancel();
        await SwallowAsync(run);
    }

    [Fact]
    public async Task Snapshot_ToAPeerWithoutPartialSupport_ClearsOnEnqueue()
    {
        // The 0.2.x behaviour, kept deliberately: a peer that sends no SnapshotAck would otherwise
        // leave the debt standing forever and re-stream the snapshot on every reconnect.
        var a = await NewBaselinedSideAsync();
        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 DrainThroughAsync(fromSession, SyncMessage.MsgOneofCase.SnapshotComplete, cts.Token);

        await WaitForAsync(async () => (await PendingSnapshotTables(a.Db)).Count == 0, RunTimeout);

        cts.Cancel();
        await SwallowAsync(run);
    }

    [Fact]
    public async Task PartialSnapshot_DoesNotAdvanceTheReceiversWatermark()
    {
        // The receiver's half of the same invariant. last_applied_remote_seq is ONE number for the
        // whole peer oplog; moving it to a partial snapshot's as_of would declare the untouched
        // tables applied through that point, and their deltas below it would never be resent.
        var a = await NewSideAsync();
        await a.Store.SetLastAppliedRemoteSeqAsync(10, 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, supportsPartialSnapshot: true);

        var begin = new SnapshotBegin { AsOfSeq = 900 };
        begin.Tables.Add("orders");
        await toSession.WriteAsync(new SyncMessage { SnapshotBegin = begin }, 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 = 900 } }, cts.Token);

        // The ack proves the complete was processed — and that a 0.3.0 receiver answers one at all.
        var ack = await NextOfCaseAsync(fromSession, SyncMessage.MsgOneofCase.SnapshotAck, cts.Token);
        Assert.Equal(900, ack.SnapshotAck.AsOfSeq);

        Assert.Equal(10, await LastAppliedRemoteSeq(a.Db)); // NOT 900
        Assert.Single(await ReadOrders(a.Db));              // the rows still merged

        cts.Cancel();
        await SwallowAsync(run);
    }

    [Fact]
    public async Task FullSnapshot_ToAnOldPeer_DrawsNoSnapshotAck()
    {
        // Symmetry check on the negotiation: a 0.3.0 receiver must not put a message a pre-0.3.0
        // sender has never heard of on the wire back to it.
        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 = 300 } }, 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 = 300 } }, cts.Token);

        // A full snapshot from an old peer still advances the watermark — that is the 0.2.x contract.
        await WaitForAsync(async () => await LastAppliedRemoteSeq(a.Db) == 300, RunTimeout);
        // Barrier: an ack for this delta proves everything before it was processed, so an absent
        // SnapshotAck is absent rather than merely late.
        await toSession.WriteAsync(new SyncMessage
        {
            DeltaBatch = new DeltaBatch { Entries = { new OplogEntry
            {
                Seq = 301, 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.False(fromSession.TryPeek(out var stray) && stray.MsgCase == SyncMessage.MsgOneofCase.SnapshotAck);

        cts.Cancel();
        await SwallowAsync(run);
    }

    [Fact]
    public async Task BaselinedTable_ReachesARealPeer_WithoutRestreamingTheOthers()
    {
        // End to end over two real databases: the baselined table's pre-existing rows arrive via the
        // scoped snapshot, the untouched table's rows arrive as ordinary deltas, and the debt clears.
        var options = new ReplicationOptions { FlushInterval = TimeSpan.FromMilliseconds(20) };
        var a = await NewBaselinedSideAsync(options);
        var b = await NewTwoTableSideAsync(options);

        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() =>
            (await ReadOrders(b.Db)).Count == 2
            && (await b.Db.QueryAsync("SELECT COUNT(*) FROM widgets", x => x.GetInt64(0)))[0] == 2;

        await WaitForAsync(Converged, RunTimeout);
        Assert.Equal(await ReadOrders(a.Db), await ReadOrders(b.Db));
        await WaitForAsync(async () => (await PendingSnapshotTables(a.Db)).Count == 0, RunTimeout);

        // A partial snapshot speaks for nothing outside its tables, so neither side may skip the
        // deltas it covered: B's watermark tracks the oplog, not the snapshot's as-of.
        Assert.Equal(0, await NeedsSnapshot(a.Db));

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

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