feat(localdb): SyncSession protocol state machine (handshake fail-closed, bidirectional delta pump)

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

@@ -0,0 +1,18 @@
+using System.Threading.Channels;
+using ZB.MOM.WW.LocalDb.Contracts;
+
+namespace ZB.MOM.WW.LocalDb.Replication.Internal;
+
+/// <summary>
+/// The transport seam a <see cref="SyncSession"/> runs over: an outbound send delegate plus an
+/// inbound message reader. Task 11 adapts a gRPC duplex stream to this; tests wire two in-memory
+/// <see cref="Channel{T}"/> pairs crosswise. <see cref="Send"/> may be called concurrently by the
+/// session's pump and receive loops — the session serializes those calls, so an adapter's writer
+/// only ever sees one send at a time.
+/// </summary>
+internal sealed class SyncDuplex
+{
+    public required Func<SyncMessage, CancellationToken, Task> Send { get; init; }
+
+    public required ChannelReader<SyncMessage> Inbox { get; init; }
+}

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

@@ -0,0 +1,286 @@
+using Microsoft.Extensions.Logging;
+using ZB.MOM.WW.LocalDb.Contracts;
+using ZB.MOM.WW.LocalDb.Hlc;
+using ZB.MOM.WW.LocalDb.Internal;
+
+namespace ZB.MOM.WW.LocalDb.Replication.Internal;
+
+/// <summary>
+/// The symmetric 2-node sync protocol as one state machine: both the gRPC service and the gRPC
+/// client run this identical logic over a <see cref="SyncDuplex"/>. It performs the fail-closed
+/// handshake, then runs a bidirectional steady state — an outbound change pump and an inbound
+/// apply/ack loop — until the token is cancelled or the stream ends. Local writes never block on
+/// the peer; every failure is a typed exception that tears down both loops and propagates.
+/// </summary>
+internal sealed class SyncSession
+{
+    private const uint LibSchemaVersion = 1;
+
+    private readonly SqliteLocalDb _db;
+    private readonly OplogStore _store;
+    private readonly LwwApplier _applier;
+    private readonly ReplicationOptions _options;
+    private readonly ILogger _logger;
+    private readonly Func<DateTimeOffset> _utcNow;
+
+    // gRPC stream writers are not safe for concurrent writes; the pump and receive loops both send,
+    // so every outbound message goes through this gate to serialize them into one writer.
+    private readonly SemaphoreSlim _sendGate = new(1, 1);
+
+    // Highest oplog seq handed to the wire this session; the pump reads strictly above it so a batch
+    // is never re-sent before its ack (the reliable ordered stream delivers exactly once in-session).
+    private long _sentThruSeq;
+    private long _peerAckedSeq;
+
+    public SyncSession(
+        SqliteLocalDb db, OplogStore store, LwwApplier applier, ReplicationOptions options, ILogger logger,
+        Func<DateTimeOffset>? utcNow = null)
+    {
+        _db = db;
+        _store = store;
+        _applier = applier;
+        _options = options;
+        _logger = logger;
+        _utcNow = utcNow ?? (() => DateTimeOffset.UtcNow);
+    }
+
+    /// <summary>Invoked when the peer's HandshakeAck requests a snapshot before deltas. Task 12 fills this in.</summary>
+    public Func<SyncDuplex, CancellationToken, Task>? SnapshotSender { get; set; }
+
+    /// <summary>Invoked on an inbound SnapshotBegin. Task 12 fills this in.</summary>
+    public Func<SnapshotBegin, SyncDuplex, CancellationToken, Task>? SnapshotReceiver { get; set; }
+
+    /// <summary>Highest seq the peer has acknowledged applying this session.</summary>
+    public long PeerAckedSeq => Interlocked.Read(ref _peerAckedSeq);
+
+    /// <summary>The peer node's identity, once the handshake has completed.</summary>
+    public string? PeerNodeId { get; private set; }
+
+    public async Task RunAsync(SyncDuplex duplex, CancellationToken ct)
+    {
+        var peerState = await _store.GetPeerStateAsync(ct);
+
+        var localHandshake = BuildHandshake(peerState);
+        await SendAsync(duplex, new SyncMessage { Handshake = localHandshake }, ct);
+
+        var firstInbound = await ReceiveOneAsync(duplex, ct);
+        if (firstInbound.MsgCase != SyncMessage.MsgOneofCase.Handshake)
+            throw new InvalidOperationException(
+                $"Replication protocol error: expected a Handshake as the first message, got {firstInbound.MsgCase}.");
+        var peerHandshake = firstInbound.Handshake;
+
+        ValidateHandshake(localHandshake, peerHandshake);
+
+        await _store.SetPeerNodeIdAsync(peerHandshake.NodeId, ct);
+        PeerNodeId = peerHandshake.NodeId;
+        _sentThruSeq = peerHandshake.LastAppliedRemoteSeq;
+        Interlocked.Exchange(ref _peerAckedSeq, peerHandshake.LastAppliedRemoteSeq);
+
+        var snapshotRequired = await ComputeSnapshotRequiredAsync(peerHandshake, peerState, ct);
+        await SendAsync(duplex, new SyncMessage { HandshakeAck = new HandshakeAck { NodeId = _db.NodeId, SnapshotRequired = snapshotRequired } }, ct);
+
+        var peerAckMsg = await ReceiveOneAsync(duplex, ct);
+        if (peerAckMsg.MsgCase != SyncMessage.MsgOneofCase.HandshakeAck)
+            throw new InvalidOperationException(
+                $"Replication protocol error: expected a HandshakeAck, got {peerAckMsg.MsgCase}.");
+
+        if (peerAckMsg.HandshakeAck.SnapshotRequired)
+        {
+            if (SnapshotSender is null)
+                throw new NotSupportedException("snapshot required but no snapshot sender configured");
+            await SnapshotSender(duplex, ct);
+        }
+
+        await RunSteadyStateAsync(duplex, ct);
+    }
+
+    private async Task RunSteadyStateAsync(SyncDuplex duplex, CancellationToken ct)
+    {
+        using var linked = CancellationTokenSource.CreateLinkedTokenSource(ct);
+        var pump = PumpLoopAsync(duplex, linked.Token);
+        var receive = ReceiveLoopAsync(duplex, linked.Token);
+
+        // Whichever loop completes first (fault or stream-end) drives teardown: cancel the sibling,
+        // drain it (swallowing only the cancellation we caused), then surface the primary outcome.
+        var finished = await Task.WhenAny(pump, receive);
+        linked.Cancel();
+        var other = finished == pump ? receive : pump;
+        try { await other; }
+        catch (OperationCanceledException) { }
+        await finished;
+    }
+
+    private async Task PumpLoopAsync(SyncDuplex duplex, CancellationToken ct)
+    {
+        while (!ct.IsCancellationRequested)
+        {
+            var batch = await _store.ReadBatchAboveAsync(_sentThruSeq, _options.MaxBatchSize, ct);
+            if (batch.Count > 0)
+            {
+                var delta = new DeltaBatch();
+                foreach (var entry in batch)
+                    delta.Entries.Add(ToProto(entry));
+                await SendAsync(duplex, new SyncMessage { DeltaBatch = delta }, ct);
+                _sentThruSeq = batch[^1].Seq;
+                continue;
+            }
+
+            await Task.Delay(_options.FlushInterval, ct);
+        }
+    }
+
+    private async Task ReceiveLoopAsync(SyncDuplex duplex, CancellationToken ct)
+    {
+        while (await duplex.Inbox.WaitToReadAsync(ct))
+        {
+            while (duplex.Inbox.TryRead(out var msg))
+            {
+                switch (msg.MsgCase)
+                {
+                    case SyncMessage.MsgOneofCase.DeltaBatch:
+                        await HandleDeltaBatchAsync(duplex, msg.DeltaBatch, ct);
+                        break;
+                    case SyncMessage.MsgOneofCase.DeltaAck:
+                        var acked = msg.DeltaAck.AppliedThruSeq;
+                        InterlockedMax(ref _peerAckedSeq, acked);
+                        await _store.RecordPeerAckAsync(acked, ct);
+                        break;
+                    case SyncMessage.MsgOneofCase.SnapshotBegin:
+                        if (SnapshotReceiver is null)
+                            throw new NotSupportedException("snapshot received but no snapshot receiver configured");
+                        await SnapshotReceiver(msg.SnapshotBegin, duplex, ct);
+                        break;
+                    default:
+                        throw new InvalidOperationException(
+                            $"Replication protocol error: unexpected {msg.MsgCase} after the handshake.");
+                }
+            }
+        }
+    }
+
+    private async Task HandleDeltaBatchAsync(SyncDuplex duplex, DeltaBatch batch, CancellationToken ct)
+    {
+        var entries = new List<OplogEntryRecord>(batch.Entries.Count);
+        var maxHlc = 0L;
+        foreach (var e in batch.Entries)
+        {
+            entries.Add(FromProto(e));
+            if (e.Hlc > maxHlc) maxHlc = e.Hlc;
+        }
+
+        if (entries.Count > 0)
+        {
+            var maxHlcMs = HybridLogicalClock.PhysicalMs(maxHlc);
+            var driftLimitMs = _utcNow().Add(_options.MaxHlcDriftAhead).ToUnixTimeMilliseconds();
+            if (maxHlcMs > driftLimitMs)
+            {
+                _logger.LogWarning(
+                    "Inbound delta HLC physical time {HlcMs} ms exceeds local clock + MaxHlcDriftAhead ({LimitMs} ms).",
+                    maxHlcMs, driftLimitMs);
+                if (_options.FailClosedOnDrift)
+                    throw new InvalidOperationException(
+                        $"Inbound HLC physical time {maxHlcMs} ms exceeds MaxHlcDriftAhead limit {driftLimitMs} ms; rejecting batch (fail-closed).");
+            }
+        }
+
+        var result = await _applier.ApplyBatchAsync(entries, ct);
+        await SendAsync(duplex, new SyncMessage { DeltaAck = new DeltaAck { AppliedThruSeq = result.AppliedThruSeq } }, ct);
+    }
+
+    private Handshake BuildHandshake(PeerState peerState)
+    {
+        var handshake = new Handshake
+        {
+            NodeId = _db.NodeId,
+            LibSchemaVersion = LibSchemaVersion,
+            LastAppliedRemoteSeq = peerState.LastAppliedRemoteSeq,
+        };
+        foreach (var table in _db.ReplicatedTables.Values.OrderBy(t => t.Name, StringComparer.Ordinal))
+            handshake.Tables.Add(new TableDigest { TableName = table.Name, Digest = table.Digest });
+        return handshake;
+    }
+
+    private static void ValidateHandshake(Handshake local, Handshake peer)
+    {
+        var versionMismatch = local.LibSchemaVersion != peer.LibSchemaVersion;
+
+        var localByName = local.Tables.ToDictionary(t => t.TableName, t => t.Digest, StringComparer.Ordinal);
+        var peerByName = peer.Tables.ToDictionary(t => t.TableName, t => t.Digest, StringComparer.Ordinal);
+        var mismatched = new SortedSet<string>(StringComparer.Ordinal);
+        foreach (var name in localByName.Keys.Union(peerByName.Keys, StringComparer.Ordinal))
+            if (!localByName.TryGetValue(name, out var ld) || !peerByName.TryGetValue(name, out var pd) || ld != pd)
+                mismatched.Add(name);
+
+        if (!versionMismatch && mismatched.Count == 0)
+            return;
+
+        var parts = new List<string>();
+        if (versionMismatch)
+            parts.Add($"lib_schema_version {local.LibSchemaVersion} != peer {peer.LibSchemaVersion}");
+        if (mismatched.Count > 0)
+            parts.Add($"table schema mismatch on: {string.Join(", ", mismatched)}");
+
+        throw new LocalDbSchemaMismatchException(
+            "Replication handshake rejected (fail-closed): " + string.Join("; ", parts) + ".");
+    }
+
+    private async Task<bool> ComputeSnapshotRequiredAsync(Handshake peerHandshake, PeerState peerState, CancellationToken ct)
+    {
+        if (peerState.NeedsSnapshot)
+            return true;
+
+        var minRows = await _db.QueryAsync(
+            "SELECT COALESCE(MIN(seq), 0) FROM __localdb_oplog", static r => r.GetInt64(0), null, ct);
+        var minSeq = minRows[0];
+        // minSeq 0 => oplog empty. A gap exists when the peer's applied watermark falls below the
+        // oldest seq we can still stream (pruning discarded everything at or below its horizon).
+        return minSeq > 0 && peerHandshake.LastAppliedRemoteSeq < minSeq - 1;
+    }
+
+    private async Task SendAsync(SyncDuplex duplex, SyncMessage message, CancellationToken ct)
+    {
+        await _sendGate.WaitAsync(ct);
+        try { await duplex.Send(message, ct); }
+        finally { _sendGate.Release(); }
+    }
+
+    private static async Task<SyncMessage> ReceiveOneAsync(SyncDuplex duplex, CancellationToken ct)
+    {
+        if (await duplex.Inbox.WaitToReadAsync(ct) && duplex.Inbox.TryRead(out var message))
+            return message;
+        throw new InvalidOperationException(
+            "Replication protocol error: the peer closed the stream before completing the handshake.");
+    }
+
+    private static OplogEntry ToProto(OplogEntryRecord entry)
+    {
+        var proto = new OplogEntry
+        {
+            Seq = entry.Seq,
+            TableName = entry.TableName,
+            PkJson = entry.PkJson,
+            Hlc = entry.Hlc,
+            NodeId = entry.NodeId,
+            IsTombstone = entry.IsTombstone,
+        };
+        if (entry.RowJson is not null)
+            proto.RowJson = entry.RowJson;
+        return proto;
+    }
+
+    private static OplogEntryRecord FromProto(OplogEntry entry) =>
+        new(entry.Seq, entry.TableName, entry.PkJson, entry.HasRowJson ? entry.RowJson : null,
+            entry.Hlc, entry.NodeId, entry.IsTombstone);
+
+    private static void InterlockedMax(ref long target, long value)
+    {
+        long current;
+        do
+        {
+            current = Interlocked.Read(ref target);
+            if (value <= current)
+                return;
+        }
+        while (Interlocked.CompareExchange(ref target, value, current) != current);
+    }
+}

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

@@ -0,0 +1,446 @@
+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;
+
+public sealed class SyncSessionTests : 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();
+        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 static void SqliteConnection_ClearAllPools() => Microsoft.Data.Sqlite.SqliteConnection.ClearAllPools();
+
+    private sealed record Side(
+        SqliteLocalDb Db, OplogStore Store, LwwApplier Applier, SyncSession Session);
+
+    private async Task<Side> NewSideAsync(
+        string createSql = OrdersSql,
+        string table = "orders",
+        ReplicationOptions? options = null,
+        ILogger? logger = null,
+        Func<DateTimeOffset>? utcNow = 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(createSql);
+        db.RegisterReplicated(table);
+
+        options ??= new ReplicationOptions { FlushInterval = TimeSpan.FromMilliseconds(20) };
+        var store = new OplogStore(db, options, utcNow);
+        var applier = new LwwApplier(db, utcNow);
+        var session = new SyncSession(db, store, applier, options, logger ?? NullLogger.Instance, utcNow);
+        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, uint version = 1, long lastAppliedRemoteSeq = 0)
+    {
+        var h = new Handshake { NodeId = peerNodeId, LibSchemaVersion = version, 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 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> OplogCount(SqliteLocalDb db)
+    {
+        var r = await db.QueryAsync("SELECT COUNT(*) FROM __localdb_oplog", x => x.GetInt64(0));
+        return r[0];
+    }
+
+    private static async Task<long> LastAckedSeq(SqliteLocalDb db)
+    {
+        var r = await db.QueryAsync("SELECT last_acked_seq FROM __localdb_peer_state WHERE id = 1", x => x.GetInt64(0));
+        return r[0];
+    }
+
+    [Fact]
+    public async Task Handshake_Exchanged_BothProceed()
+    {
+        var a = await NewSideAsync();
+        var b = await NewSideAsync();
+        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(() => Task.FromResult(a.Session.PeerNodeId is not null && b.Session.PeerNodeId is not null), RunTimeout);
+
+        Assert.Equal(b.Db.NodeId, a.Session.PeerNodeId);
+        Assert.Equal(a.Db.NodeId, b.Session.PeerNodeId);
+
+        cts.Cancel();
+        await SwallowAsync(runA);
+        await SwallowAsync(runB);
+    }
+
+    [Fact]
+    public async Task Handshake_DigestMismatch_FailsClosed()
+    {
+        var a = await NewSideAsync();
+        var b = await NewSideAsync("CREATE TABLE orders (id INTEGER PRIMARY KEY, sku TEXT, qty INTEGER, extra TEXT)");
+        await a.Db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'A', 5)");
+        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);
+
+        var exA = await Assert.ThrowsAsync<LocalDbSchemaMismatchException>(() => runA);
+        var exB = await Assert.ThrowsAsync<LocalDbSchemaMismatchException>(() => runB);
+        Assert.Contains("orders", exA.Message);
+        Assert.Contains("orders", exB.Message);
+
+        // No delta crossed the handshake: B never received A's row.
+        Assert.Empty(await ReadOrders(b.Db));
+    }
+
+    [Fact]
+    public async Task Handshake_SchemaVersionMismatch_FailsClosed()
+    {
+        var a = await NewSideAsync();
+        var (duplex, toSession, _) = ScriptedPeer();
+
+        using var cts = new CancellationTokenSource(RunTimeout);
+        var run = a.Session.RunAsync(duplex, cts.Token);
+
+        await toSession.WriteAsync(new SyncMessage { Handshake = MatchingHandshake(a.Db, "peer", version: 2) }, cts.Token);
+
+        await Assert.ThrowsAsync<LocalDbSchemaMismatchException>(() => run);
+    }
+
+    [Fact]
+    public async Task Deltas_FlowBothDirections_AcksAdvanceWatermarks()
+    {
+        var a = await NewSideAsync();
+        var b = await NewSideAsync();
+        await a.Db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'FROM_A', 10)");
+        await b.Db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (2, 'FROM_B', 20)");
+        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 == 2 && rb.Count == 2
+                && await OplogCount(a.Db) == 0 && await OplogCount(b.Db) == 0;
+        }
+
+        await WaitForAsync(Converged, RunTimeout);
+
+        var rowsA = await ReadOrders(a.Db);
+        var rowsB = await ReadOrders(b.Db);
+        Assert.Equal(rowsA, rowsB);
+        Assert.Equal([(1L, "FROM_A", (long?)10), (2L, "FROM_B", (long?)20)], rowsA);
+        Assert.True(await LastAckedSeq(a.Db) >= 1);
+        Assert.True(await LastAckedSeq(b.Db) >= 1);
+
+        cts.Cancel();
+        await SwallowAsync(runA);
+        await SwallowAsync(runB);
+    }
+
+    [Fact]
+    public async Task Backlog_DrainsWithoutWaitingFlushInterval()
+    {
+        var options = new ReplicationOptions { MaxBatchSize = 5, FlushInterval = TimeSpan.FromSeconds(30) };
+        var a = await NewSideAsync(options: options);
+        var b = await NewSideAsync(options: options);
+
+        const int rows = 15; // 3 x MaxBatchSize
+        for (var i = 1; i <= rows; i++)
+            await a.Db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (@id, 'R', @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);
+
+        // FlushInterval is 30 s: reaching convergence in a couple seconds proves the pump loops
+        // immediately on backlog instead of waiting a flush tick between batches.
+        var sw = System.Diagnostics.Stopwatch.StartNew();
+        await WaitForAsync(async () => (await ReadOrders(b.Db)).Count == rows, TimeSpan.FromSeconds(5));
+        sw.Stop();
+        Assert.True(sw.Elapsed < TimeSpan.FromSeconds(5), $"drain took {sw.Elapsed}");
+
+        cts.Cancel();
+        await SwallowAsync(runA);
+        await SwallowAsync(runB);
+    }
+
+    [Fact]
+    public async Task HlcDriftAhead_Warns()
+    {
+        var logger = new ListLogger();
+        var a = await NewSideAsync(
+            options: new ReplicationOptions { FlushInterval = TimeSpan.FromMilliseconds(20), FailClosedOnDrift = false },
+            logger: logger);
+        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);
+
+        var aheadHlc = (DateTimeOffset.UtcNow.AddMinutes(10).ToUnixTimeMilliseconds() << 16);
+        await toSession.WriteAsync(new SyncMessage
+        {
+            DeltaBatch = new DeltaBatch
+            {
+                Entries = { new OplogEntry
+                {
+                    Seq = 1, TableName = "orders", PkJson = "{\"id\":1}",
+                    RowJson = "{\"id\":1,\"sku\":\"AHEAD\",\"qty\":9}", Hlc = aheadHlc, NodeId = "peer", IsTombstone = false,
+                } },
+            },
+        }, cts.Token);
+
+        // Drain the DeltaAck the session emits after applying.
+        SyncMessage ack;
+        do { ack = await NextAsync(fromSession, cts.Token); } while (ack.MsgCase != SyncMessage.MsgOneofCase.DeltaAck);
+        Assert.Equal(1, ack.DeltaAck.AppliedThruSeq);
+
+        Assert.Single(await ReadOrders(a.Db)); // applied despite drift
+        Assert.Contains(logger.Entries, e => e.Level == LogLevel.Warning);
+
+        cts.Cancel();
+        await SwallowAsync(run);
+    }
+
+    [Fact]
+    public async Task HlcDriftAhead_FailClosed_Throws()
+    {
+        var a = await NewSideAsync(
+            options: new ReplicationOptions { FlushInterval = TimeSpan.FromMilliseconds(20), FailClosedOnDrift = true });
+        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);
+
+        var aheadHlc = (DateTimeOffset.UtcNow.AddMinutes(10).ToUnixTimeMilliseconds() << 16);
+        await toSession.WriteAsync(new SyncMessage
+        {
+            DeltaBatch = new DeltaBatch
+            {
+                Entries = { new OplogEntry
+                {
+                    Seq = 1, TableName = "orders", PkJson = "{\"id\":1}",
+                    RowJson = "{\"id\":1,\"sku\":\"AHEAD\",\"qty\":9}", Hlc = aheadHlc, NodeId = "peer", IsTombstone = false,
+                } },
+            },
+        }, cts.Token);
+
+        await Assert.ThrowsAsync<InvalidOperationException>(() => run);
+        Assert.Empty(await ReadOrders(a.Db)); // rejected, not applied
+    }
+
+    [Fact]
+    public async Task PeerNeedsSnapshot_InvokesSnapshotHooks()
+    {
+        // (a) peer's HandshakeAck.snapshot_required = true -> our SnapshotSender fires.
+        {
+            var a = await NewSideAsync();
+            var (duplex, toSession, fromSession) = ScriptedPeer();
+            var sent = new TaskCompletionSource();
+            a.Session.SnapshotSender = (_, _) => { sent.TrySetResult(); return Task.CompletedTask; };
+
+            using var cts = new CancellationTokenSource(RunTimeout);
+            var run = a.Session.RunAsync(duplex, cts.Token);
+
+            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 sent.Task.WaitAsync(RunTimeout);
+            cts.Cancel();
+            await SwallowAsync(run);
+        }
+
+        // (b) inbound SnapshotBegin -> our SnapshotReceiver fires.
+        {
+            var a = await NewSideAsync();
+            var (duplex, toSession, fromSession) = ScriptedPeer();
+            var received = new TaskCompletionSource();
+            a.Session.SnapshotReceiver = (_, _, _) => { received.TrySetResult(); return Task.CompletedTask; };
+
+            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 received.Task.WaitAsync(RunTimeout);
+            cts.Cancel();
+            await SwallowAsync(run);
+        }
+
+        // (c) snapshot required but no sender hook -> NotSupportedException.
+        {
+            var a = await NewSideAsync();
+            var (duplex, toSession, fromSession) = ScriptedPeer();
+
+            using var cts = new CancellationTokenSource(RunTimeout);
+            var run = a.Session.RunAsync(duplex, cts.Token);
+
+            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 Assert.ThrowsAsync<NotSupportedException>(() => run);
+        }
+
+        // (d) inbound SnapshotBegin but no receiver hook -> NotSupportedException.
+        {
+            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 = 0 } }, cts.Token);
+
+            await Assert.ThrowsAsync<NotSupportedException>(() => run);
+        }
+    }
+
+    [Fact]
+    public async Task SecondHandshake_MidStream_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 { Handshake = MatchingHandshake(a.Db, "peer") }, cts.Token);
+
+        await Assert.ThrowsAsync<InvalidOperationException>(() => run);
+    }
+
+    // Completes the two-phase handshake against a real session: consumes its Handshake + HandshakeAck
+    // and replies with a matching Handshake + a no-snapshot HandshakeAck, leaving the session in steady state.
+    private static async Task DriveHandshakeAsync(
+        SqliteLocalDb sessionDb, ChannelWriter<SyncMessage> toSession, ChannelReader<SyncMessage> fromSession, CancellationToken ct)
+    {
+        var hs = await NextAsync(fromSession, ct);
+        Assert.Equal(SyncMessage.MsgOneofCase.Handshake, hs.MsgCase);
+        await toSession.WriteAsync(new SyncMessage { Handshake = MatchingHandshake(sessionDb, "peer") }, ct);
+        var ack = await NextAsync(fromSession, ct);
+        Assert.Equal(SyncMessage.MsgOneofCase.HandshakeAck, ack.MsgCase);
+        await toSession.WriteAsync(new SyncMessage { HandshakeAck = new HandshakeAck { NodeId = "peer", SnapshotRequired = false } }, ct);
+    }
+
+    private sealed class ListLogger : ILogger
+    {
+        public List<(LogLevel Level, string Message)> Entries { get; } = [];
+
+        public IDisposable BeginScope<TState>(TState state) where TState : notnull => NullScope.Instance;
+        public bool IsEnabled(LogLevel logLevel) => true;
+
+        public void Log<TState>(
+            LogLevel logLevel, EventId eventId, TState state, Exception? exception, Func<TState, Exception?, string> formatter)
+        {
+            lock (Entries)
+                Entries.Add((logLevel, formatter(state, exception)));
+        }
+
+        private sealed class NullScope : IDisposable
+        {
+            public static readonly NullScope Instance = new();
+            public void Dispose() { }
+        }
+    }
+}