diff --git a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/SyncDuplex.cs b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/SyncDuplex.cs new file mode 100644 index 0000000..a2a9043 --- /dev/null +++ b/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; + +/// +/// The transport seam a 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 +/// pairs crosswise. 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. +/// +internal sealed class SyncDuplex +{ + public required Func Send { get; init; } + + public required ChannelReader Inbox { get; init; } +} diff --git a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/SyncSession.cs b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/SyncSession.cs new file mode 100644 index 0000000..0aa796d --- /dev/null +++ b/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; + +/// +/// The symmetric 2-node sync protocol as one state machine: both the gRPC service and the gRPC +/// client run this identical logic over a . 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. +/// +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 _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? utcNow = null) + { + _db = db; + _store = store; + _applier = applier; + _options = options; + _logger = logger; + _utcNow = utcNow ?? (() => DateTimeOffset.UtcNow); + } + + /// Invoked when the peer's HandshakeAck requests a snapshot before deltas. Task 12 fills this in. + public Func? SnapshotSender { get; set; } + + /// Invoked on an inbound SnapshotBegin. Task 12 fills this in. + public Func? SnapshotReceiver { get; set; } + + /// Highest seq the peer has acknowledged applying this session. + public long PeerAckedSeq => Interlocked.Read(ref _peerAckedSeq); + + /// The peer node's identity, once the handshake has completed. + 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(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(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(); + 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 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 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); + } +} diff --git a/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/SyncSessionTests.cs b/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/SyncSessionTests.cs new file mode 100644 index 0000000..5ed8dd1 --- /dev/null +++ b/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 _paths = []; + private readonly List _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 NewSideAsync( + string createSql = OrdersSql, + string table = "orders", + ReplicationOptions? options = null, + ILogger? logger = null, + Func? 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(); + var bToA = Channel.CreateUnbounded(); + 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 ToSession, ChannelReader FromSession) ScriptedPeer() + { + var toSession = Channel.CreateUnbounded(); + var fromSession = Channel.CreateUnbounded(); + 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 NextAsync(ChannelReader 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> 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> 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 OplogCount(SqliteLocalDb db) + { + var r = await db.QueryAsync("SELECT COUNT(*) FROM __localdb_oplog", x => x.GetInt64(0)); + return r[0]; + } + + private static async Task 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(() => runA); + var exB = await Assert.ThrowsAsync(() => 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(() => 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 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(() => 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(() => 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(() => 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(() => 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 toSession, ChannelReader 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 state) where TState : notnull => NullScope.Instance; + public bool IsEnabled(LogLevel logLevel) => true; + + public void Log( + LogLevel logLevel, EventId eventId, TState state, Exception? exception, Func formatter) + { + lock (Entries) + Entries.Add((logLevel, formatter(state, exception))); + } + + private sealed class NullScope : IDisposable + { + public static readonly NullScope Instance = new(); + public void Dispose() { } + } + } +}