diff --git a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/SnapshotStreamer.cs b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/SnapshotStreamer.cs new file mode 100644 index 0000000..0811d96 --- /dev/null +++ b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/SnapshotStreamer.cs @@ -0,0 +1,171 @@ +using Microsoft.Extensions.Logging; +using ZB.MOM.WW.LocalDb.Contracts; +using ZB.MOM.WW.LocalDb.Internal; +using ZB.MOM.WW.LocalDb.Registration; + +namespace ZB.MOM.WW.LocalDb.Replication.Internal; + +/// +/// Consumes an inbound full-snapshot the receive loop routes to it: SnapshotBegin opens a receive, +/// each SnapshotBatch merges through the SAME last-writer-wins path as deltas (never truncating — +/// the receiver may hold newer local writes), and SnapshotComplete advances the durable watermark +/// so incremental deltas resume above it. +/// +internal interface ISnapshotApplier +{ + Task OnBeginAsync(SnapshotBegin begin, CancellationToken ct); + Task OnBatchAsync(SnapshotBatch batch, CancellationToken ct); + Task OnCompleteAsync(SnapshotComplete complete, CancellationToken ct); +} + +/// +/// Streams a full snapshot of every registered table to a peer that has fallen behind the pruned +/// oplog horizon. Reads the authoritative per-row version from __localdb_row_version joined +/// to each live table, pages it into messages through the gated data +/// lane, and brackets them with SnapshotBegin/SnapshotComplete carrying the max oplog seq as-of. +/// +internal sealed class SnapshotStreamer(SqliteLocalDb db, OplogStore store, ReplicationOptions options, ILogger logger) +{ + public async Task SendAsync(Func send, CancellationToken ct) + { + var asOfSeq = await store.GetMaxSeqAsync(ct); + await send(new SyncMessage { SnapshotBegin = new SnapshotBegin { AsOfSeq = asOfSeq } }, ct); + + var totalRows = 0L; + foreach (var table in db.ReplicatedTables.Values.OrderBy(t => t.Name, StringComparer.Ordinal)) + { + var sql = BuildPageSql(table); + var offset = 0; + while (true) + { + var rows = await db.QueryAsync( + sql, + // Column 4 is NULL for tombstones; a proto optional string cannot be assigned null, + // so carry it as a nullable and only set row_json when present. + static r => ( + PkJson: r.GetString(0), + Hlc: r.GetInt64(1), + NodeId: r.GetString(2), + IsTombstone: r.GetInt64(3) != 0, + RowJson: r.IsDBNull(4) ? null : r.GetString(4)), + new { t = table.Name, page = options.MaxBatchSize, off = offset }, ct); + if (rows.Count == 0) + break; + + var batch = new SnapshotBatch { TableName = table.Name }; + foreach (var row in rows) + { + var snapshotRow = new SnapshotRow + { + PkJson = row.PkJson, + Hlc = row.Hlc, + NodeId = row.NodeId, + IsTombstone = row.IsTombstone, + }; + if (row.RowJson is not null) + snapshotRow.RowJson = row.RowJson; + batch.Rows.Add(snapshotRow); + } + await send(new SyncMessage { SnapshotBatch = batch }, ct); + totalRows += rows.Count; + + offset += rows.Count; + if (rows.Count < options.MaxBatchSize) + break; + } + } + + await send(new SyncMessage { SnapshotComplete = new SnapshotComplete { AsOfSeq = asOfSeq } }, ct); + // The sender owed this snapshot (its own needs_snapshot); clearing it here — after the whole + // snapshot is on the wire — stops it re-streaming on the next reconnect. + await store.SetNeedsSnapshotAsync(false, ct); + logger.LogInformation("Snapshot sent (as_of_seq {AsOfSeq}, {Rows} rows).", asOfSeq, totalRows); + } + + // Pages one table's rows by joining the row-version ledger to the live table on the PK columns + // extracted from pk_json. Rows present in the live table but absent from __localdb_row_version + // (data written BEFORE the table was registered for replication) are intentionally NOT snapshotted: + // they were never captured, consistent with CDC semantics — a full re-registration is required to + // baseline them. + private static string BuildPageSql(ReplicatedTable table) + { + var name = Ident(table.Name); + var join = string.Join(" AND ", table.PkColumns.Select(p => $"t.{Ident(p)} = json_extract(rv.pk_json, {JsonPath(p)})")); + var jsonObject = "json_object(" + + string.Join(", ", table.Columns.Select(c => $"{StringLiteral(c.Name)}, t.{Ident(c.Name)}")) + ")"; + return + "SELECT rv.pk_json, rv.hlc, rv.node_id, rv.is_tombstone, " + + $"CASE WHEN rv.is_tombstone = 1 THEN NULL ELSE {jsonObject} END " + + $"FROM __localdb_row_version rv LEFT JOIN {name} t ON {join} " + + "WHERE rv.table_name = @t ORDER BY rv.pk_json LIMIT @page OFFSET @off"; + } + + private static string Ident(string name) => "\"" + name.Replace("\"", "\"\"") + "\""; + + private static string StringLiteral(string value) => "'" + value.Replace("'", "''") + "'"; + + // A single-quoted JSON path with a quoted key member so a column name with reserved characters + // still resolves: $."col". + private static string JsonPath(string column) => "'$.\"" + column.Replace("\"", "\\\"") + "\"'"; +} + +/// +/// The inbound-snapshot consumer wired into . Each batch merges through the +/// shared (seq 0 — snapshot rows carry no oplog seq — so the applier's +/// MAX-guarded watermark update is a no-op, and per-row LWW protects any newer local write). On +/// complete, the remote watermark is set to the snapshot's as-of seq so deltas resume above it. +/// +internal sealed class SnapshotApplier : ISnapshotApplier +{ + private readonly LwwApplier _applier; + private readonly OplogStore _store; + private readonly ILogger _logger; + private long _asOfSeq; + private long _rows; + private long _maxHlc; + + // db is part of the collaborator's canonical shape (mirrors the streamer) even though the merge + // reaches the database through the applier and store; kept for construction symmetry in the factory. + public SnapshotApplier(SqliteLocalDb db, LwwApplier applier, OplogStore store, ILogger logger) + { + _ = db; + _applier = applier; + _store = store; + _logger = logger; + } + + public Task OnBeginAsync(SnapshotBegin begin, CancellationToken ct) + { + _asOfSeq = begin.AsOfSeq; + _rows = 0; + _maxHlc = 0; + _logger.LogInformation("Snapshot receive begun (as_of_seq {AsOfSeq}).", _asOfSeq); + return Task.CompletedTask; + } + + public async Task OnBatchAsync(SnapshotBatch batch, CancellationToken ct) + { + if (batch.Rows.Count == 0) + return; + + var entries = new List(batch.Rows.Count); + foreach (var row in batch.Rows) + { + if (row.Hlc > _maxHlc) _maxHlc = row.Hlc; + entries.Add(new OplogEntryRecord( + 0, batch.TableName, row.PkJson, row.HasRowJson ? row.RowJson : null, row.Hlc, row.NodeId, row.IsTombstone)); + } + await _applier.ApplyBatchAsync(entries, ct); + _rows += batch.Rows.Count; + } + + public async Task OnCompleteAsync(SnapshotComplete complete, CancellationToken ct) + { + // Preserve last_seen_hlc against a concurrently-applied delta that observed a higher HLC: + // SetLastAppliedRemoteSeqAsync overwrites the column, so pass the max of what we saw and + // what is already recorded. + var lastSeenHlc = Math.Max(_maxHlc, (await _store.GetPeerStateAsync(ct)).LastSeenHlc); + await _store.SetLastAppliedRemoteSeqAsync(complete.AsOfSeq, lastSeenHlc, ct); + _logger.LogInformation("Snapshot receive complete (as_of_seq {AsOfSeq}, {Rows} rows).", complete.AsOfSeq, _rows); + } +} 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 index fd45e3e..94ffe0f 100644 --- 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 @@ -39,6 +39,10 @@ internal sealed class SyncSession private string? _peerNodeId; private bool _driftWarned; + // Set true between an inbound SnapshotBegin and its SnapshotComplete. Only ever touched on the + // single receive-loop task, so no synchronization is needed. + private bool _snapshotReceiving; + public SyncSession( SqliteLocalDb db, OplogStore store, LwwApplier applier, ReplicationOptions options, ILogger logger, Func? utcNow = null) @@ -52,16 +56,18 @@ internal sealed class SyncSession } /// - /// Invoked when the peer's HandshakeAck requests a snapshot before deltas. Receives a gated - /// send delegate that enqueues on the bounded data lane (never the raw transport). Task 12 fills this in. + /// Invoked when THIS node owes the peer a snapshot (its own snapshot-required computation, not + /// the peer's) before deltas resume. Receives a gated send delegate that enqueues on the bounded + /// data lane (never the raw transport). Wired to . /// public Func, CancellationToken, Task>? SnapshotSender { get; set; } /// - /// Invoked on an inbound SnapshotBegin, with the triggering message and a gated data-lane send - /// delegate. Task 12 fills this in. + /// Consumes an inbound snapshot inline on the receive loop: SnapshotBegin opens the receive, + /// SnapshotBatch merges through LWW, SnapshotComplete advances the watermark. Wired to + /// . /// - public Func, CancellationToken, Task>? SnapshotReceiver { get; set; } + public ISnapshotApplier? SnapshotApplier { get; set; } /// Highest seq the peer has acknowledged applying this session. public long PeerAckedSeq => Interlocked.Read(ref _peerAckedSeq); @@ -86,7 +92,7 @@ internal sealed class SyncSession await HandshakeAsync(duplex, control.Writer, GatedDataSend, linked.Token); var pump = PumpLoopAsync(data.Writer, linked.Token); - var receive = ReceiveLoopAsync(duplex, control.Writer, GatedDataSend, linked.Token); + var receive = ReceiveLoopAsync(duplex, control.Writer, linked.Token); var loops = new[] { pump, receive, writer }; // The first COMPLETED task is the primary outcome. Cancel the siblings, observe every @@ -142,7 +148,10 @@ internal sealed class SyncSession throw new InvalidOperationException( $"Replication protocol error: expected a HandshakeAck, got {peerAckMsg.MsgCase}."); - if (peerAckMsg.HandshakeAck.SnapshotRequired) + // We stream the snapshot when WE owe the peer one (computed from our pruned horizon / + // needs_snapshot), then it lands on the peer's receive loop. The peer's own flag drives ITS + // send direction symmetrically; both are safe to run because LWW merges either way. + if (snapshotRequired) { if (SnapshotSender is null) throw new NotSupportedException("snapshot required but no snapshot sender configured"); @@ -213,8 +222,7 @@ internal sealed class SyncSession } private async Task ReceiveLoopAsync( - SyncDuplex duplex, ChannelWriter controlLane, - Func dataSend, CancellationToken ct) + SyncDuplex duplex, ChannelWriter controlLane, CancellationToken ct) { while (await duplex.Inbox.WaitToReadAsync(ct)) { @@ -223,6 +231,7 @@ internal sealed class SyncSession switch (msg.MsgCase) { case SyncMessage.MsgOneofCase.DeltaBatch: + // Deltas apply normally even mid-snapshot: LWW makes the interleave safe. await HandleDeltaBatchAsync(controlLane, msg.DeltaBatch, ct); break; case SyncMessage.MsgOneofCase.DeltaAck: @@ -231,9 +240,23 @@ internal sealed class SyncSession await _store.RecordPeerAckAsync(acked, ct); break; case SyncMessage.MsgOneofCase.SnapshotBegin: - if (SnapshotReceiver is null) - throw new NotSupportedException("snapshot received but no snapshot receiver configured"); - await SnapshotReceiver(msg.SnapshotBegin, dataSend, ct); + if (SnapshotApplier is null) + throw new NotSupportedException("snapshot received but no snapshot applier configured"); + _snapshotReceiving = true; + await SnapshotApplier.OnBeginAsync(msg.SnapshotBegin, ct); + break; + case SyncMessage.MsgOneofCase.SnapshotBatch: + if (!_snapshotReceiving) + throw new InvalidOperationException( + "Replication protocol error: SnapshotBatch with no active snapshot (missing SnapshotBegin)."); + await SnapshotApplier!.OnBatchAsync(msg.SnapshotBatch, ct); + break; + case SyncMessage.MsgOneofCase.SnapshotComplete: + if (!_snapshotReceiving) + throw new InvalidOperationException( + "Replication protocol error: SnapshotComplete with no active snapshot (missing SnapshotBegin)."); + await SnapshotApplier!.OnCompleteAsync(msg.SnapshotComplete, ct); + _snapshotReceiving = false; break; default: throw new InvalidOperationException( diff --git a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/SyncSessionFactory.cs b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/SyncSessionFactory.cs index 7c46bf3..6ef7c55 100644 --- a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/SyncSessionFactory.cs +++ b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/SyncSessionFactory.cs @@ -25,6 +25,11 @@ internal static class SyncSessionFactory var store = new OplogStore(db, options); var applier = new LwwApplier(db); var logger = loggerFactory.CreateLogger(); - return new SyncSession(db, store, applier, options, logger); + var session = new SyncSession(db, store, applier, options, logger); + + var streamer = new SnapshotStreamer(db, store, options, loggerFactory.CreateLogger()); + session.SnapshotSender = streamer.SendAsync; + session.SnapshotApplier = new SnapshotApplier(db, applier, store, loggerFactory.CreateLogger()); + return session; } } diff --git a/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/SnapshotResyncTests.cs b/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/SnapshotResyncTests.cs new file mode 100644 index 0000000..1e97319 --- /dev/null +++ b/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/SnapshotResyncTests.cs @@ -0,0 +1,388 @@ +using System.Threading.Channels; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; +using ZB.MOM.WW.LocalDb.Contracts; +using ZB.MOM.WW.LocalDb.Internal; +using ZB.MOM.WW.LocalDb.Replication; +using ZB.MOM.WW.LocalDb.Replication.Internal; + +namespace ZB.MOM.WW.LocalDb.Tests; + +/// +/// 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. +/// +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 _paths = []; + private readonly List _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 NewSideAsync(ReplicationOptions? options = null) + { + var path = Path.Combine(Path.GetTempPath(), Guid.NewGuid() + ".db"); + _paths.Add(path); + var db = new SqliteLocalDb(new LocalDbOptions { Path = path }); + _disposables.Add(db); + await db.ExecuteAsync(OrdersSql); + db.RegisterReplicated("orders"); + + options ??= new ReplicationOptions { FlushInterval = TimeSpan.FromMilliseconds(20) }; + var store = new OplogStore(db, options); + var applier = new LwwApplier(db); + var session = new SyncSession(db, store, applier, options, NullLogger.Instance); + + var streamer = new SnapshotStreamer(db, store, options, NullLogger.Instance); + var snapshotApplier = new SnapshotApplier(db, applier, store, NullLogger.Instance); + session.SnapshotSender = streamer.SendAsync; + session.SnapshotApplier = snapshotApplier; + + return new Side(db, store, applier, session); + } + + private static (SyncDuplex A, SyncDuplex B) DuplexPair() + { + var aToB = Channel.CreateUnbounded(); + 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, long lastAppliedRemoteSeq = 0) + { + var h = new Handshake { NodeId = peerNodeId, LibSchemaVersion = 1, LastAppliedRemoteSeq = lastAppliedRemoteSeq }; + foreach (var t in sessionDb.ReplicatedTables.Values.OrderBy(t => t.Name, StringComparer.Ordinal)) + h.Tables.Add(new TableDigest { TableName = t.Name, Digest = t.Digest }); + return h; + } + + private static async Task NextAsync(ChannelReader reader, CancellationToken ct) + { + await reader.WaitToReadAsync(ct); + reader.TryRead(out var msg); + return msg!; + } + + private static async Task NextOfCaseAsync( + ChannelReader 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 toSession, ChannelReader fromSession, CancellationToken ct) + { + await NextOfCaseAsync(fromSession, SyncMessage.MsgOneofCase.Handshake, ct); + await toSession.WriteAsync(new SyncMessage { Handshake = MatchingHandshake(sessionDb, "peer") }, ct); + await NextOfCaseAsync(fromSession, SyncMessage.MsgOneofCase.HandshakeAck, ct); + await toSession.WriteAsync(new SyncMessage { HandshakeAck = new HandshakeAck { NodeId = "peer", SnapshotRequired = false } }, ct); + } + + private static async Task SwallowAsync(Task task) + { + try { await task; } + catch (OperationCanceledException) { } + } + + private static async Task WaitForAsync(Func> 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 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 NeedsSnapshot(SqliteLocalDb db) + { + var r = await db.QueryAsync("SELECT needs_snapshot FROM __localdb_peer_state WHERE id = 1", x => x.GetInt64(0)); + return r[0]; + } + + private static SnapshotRow SnapRow(long id, string sku, long hlc, string nodeId = "peer") => + new() + { + PkJson = $"{{\"id\":{id}}}", + RowJson = $"{{\"id\":{id},\"sku\":\"{sku}\",\"qty\":1}}", + Hlc = hlc, + NodeId = nodeId, + IsTombstone = false, + }; + + [Fact] + public async Task PrunedPeer_GetsSnapshot_ThenConverges() + { + var a = await NewSideAsync(new ReplicationOptions { FlushInterval = TimeSpan.FromMilliseconds(20), MaxOplogRows = 2 }); + var b = await NewSideAsync(); + for (var i = 1; i <= 6; i++) + await a.Db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (@id, 'A', @id)", new { id = i }); + + // Prune A's oplog below B's (empty) watermark and flag a resync: incremental deltas alone + // can no longer carry rows 1..4 to B. + Assert.True(await a.Store.EnforceCapsAsync()); + + var (da, db) = DuplexPair(); + using var cts = new CancellationTokenSource(RunTimeout); + var runA = a.Session.RunAsync(da, cts.Token); + var runB = b.Session.RunAsync(db, cts.Token); + + await WaitForAsync(async () => (await ReadOrders(b.Db)).Count == 6, RunTimeout); + Assert.Equal(await ReadOrders(a.Db), await ReadOrders(b.Db)); + + // Deltas resume after the snapshot: a post-snapshot write on A reaches B incrementally. + await a.Db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (7, 'POST', 7)"); + await WaitForAsync(async () => (await ReadOrders(b.Db)).Any(r => r.Id == 7), RunTimeout); + + cts.Cancel(); + await SwallowAsync(runA); + await SwallowAsync(runB); + } + + [Fact] + public async Task Snapshot_MergesNotTruncates() + { + var a = await NewSideAsync(new ReplicationOptions { FlushInterval = TimeSpan.FromMilliseconds(20), MaxOplogRows = 2 }); + var b = await NewSideAsync(); + + await a.Db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'A_OLD', 1)"); + await Task.Delay(5); + // B's row for the same pk is written later -> strictly higher HLC, so it must survive the merge. + await b.Db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'B_NEW', 9)"); + // A owes B a snapshot; a single-row oplog would not trip the cap, so flag it directly. + await a.Store.SetNeedsSnapshotAsync(true); + + var (da, db) = DuplexPair(); + using var cts = new CancellationTokenSource(RunTimeout); + var runA = a.Session.RunAsync(da, cts.Token); + var runB = b.Session.RunAsync(db, cts.Token); + + async Task 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 + { + 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(() => run); + } + + [Fact] + public async Task DeltaDuringSnapshot_AppliesSafely() + { + var a = await NewSideAsync(); + var (duplex, toSession, fromSession) = ScriptedPeer(); + + using var cts = new CancellationTokenSource(RunTimeout); + var run = a.Session.RunAsync(duplex, cts.Token); + + await DriveHandshakeAsync(a.Db, toSession, fromSession, cts.Token); + + await toSession.WriteAsync(new SyncMessage { SnapshotBegin = new SnapshotBegin { AsOfSeq = 200 } }, cts.Token); + await toSession.WriteAsync(new SyncMessage + { + SnapshotBatch = new SnapshotBatch { TableName = "orders", Rows = { SnapRow(1, "SNAP1", 6_000_000) } }, + }, cts.Token); + // A normal delta interleaved MID-snapshot must still apply (LWW makes interleaving safe). + await toSession.WriteAsync(new SyncMessage + { + DeltaBatch = new DeltaBatch { Entries = { new OplogEntry + { + Seq = 205, TableName = "orders", PkJson = "{\"id\":2}", + RowJson = "{\"id\":2,\"sku\":\"DELTA\",\"qty\":1}", Hlc = 6_000_010, NodeId = "peer", IsTombstone = false, + } } }, + }, cts.Token); + await NextOfCaseAsync(fromSession, SyncMessage.MsgOneofCase.DeltaAck, cts.Token); + await toSession.WriteAsync(new SyncMessage + { + SnapshotBatch = new SnapshotBatch { TableName = "orders", Rows = { SnapRow(3, "SNAP3", 6_000_020) } }, + }, cts.Token); + await toSession.WriteAsync(new SyncMessage { SnapshotComplete = new SnapshotComplete { AsOfSeq = 200 } }, cts.Token); + + await WaitForAsync(async () => (await ReadOrders(a.Db)).Count == 3, RunTimeout); + var rows = await ReadOrders(a.Db); + Assert.Equal([1L, 2L, 3L], rows.Select(r => r.Id)); + + cts.Cancel(); + await SwallowAsync(run); + } + + [Fact] + public async Task Sender_ClearsNeedsSnapshotFlag() + { + var a = await NewSideAsync(new ReplicationOptions { FlushInterval = TimeSpan.FromMilliseconds(20), MaxOplogRows = 2 }); + var b = await NewSideAsync(); + for (var i = 1; i <= 5; i++) + await a.Db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (@id, 'A', @id)", new { id = i }); + Assert.True(await a.Store.EnforceCapsAsync()); + Assert.Equal(1, await NeedsSnapshot(a.Db)); + + var (da, db) = DuplexPair(); + using var cts = new CancellationTokenSource(RunTimeout); + var runA = a.Session.RunAsync(da, cts.Token); + var runB = b.Session.RunAsync(db, cts.Token); + + await WaitForAsync(async () => (await ReadOrders(b.Db)).Count == 5, RunTimeout); + await WaitForAsync(async () => await NeedsSnapshot(a.Db) == 0, RunTimeout); + + cts.Cancel(); + await SwallowAsync(runA); + await SwallowAsync(runB); + } +} 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 index 7334d72..06b81b7 100644 --- 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 @@ -546,10 +546,11 @@ public sealed class SyncSessionTests : IDisposable [Fact] public async Task PeerNeedsSnapshot_InvokesSnapshotHooks() { - // (a) peer's HandshakeAck.snapshot_required = true -> our SnapshotSender fires, and its - // gated send delegate routes messages through the writer onto the wire. + // (a) THIS node owes the peer a snapshot (its own needs_snapshot) -> our SnapshotSender fires, + // and its gated send delegate routes messages through the writer onto the wire. { var a = await NewSideAsync(); + await a.Store.SetNeedsSnapshotAsync(true, default); var (duplex, toSession, fromSession) = ScriptedPeer(); var sent = new TaskCompletionSource(); a.Session.SnapshotSender = (send, token) => @@ -564,7 +565,7 @@ public sealed class SyncSessionTests : IDisposable await NextAsync(fromSession, cts.Token); // session Handshake await toSession.WriteAsync(new SyncMessage { Handshake = MatchingHandshake(a.Db, "peer") }, cts.Token); await NextAsync(fromSession, cts.Token); // session HandshakeAck - await toSession.WriteAsync(new SyncMessage { HandshakeAck = new HandshakeAck { NodeId = "peer", SnapshotRequired = true } }, cts.Token); + await toSession.WriteAsync(new SyncMessage { HandshakeAck = new HandshakeAck { NodeId = "peer", SnapshotRequired = false } }, cts.Token); await sent.Task.WaitAsync(RunTimeout); var complete = await NextOfCaseAsync(fromSession, SyncMessage.MsgOneofCase.SnapshotComplete, cts.Token); @@ -574,16 +575,12 @@ public sealed class SyncSessionTests : IDisposable await SwallowAsync(run); } - // (b) inbound SnapshotBegin -> our SnapshotReceiver fires with the triggering message. + // (b) inbound SnapshotBegin -> our ISnapshotApplier.OnBeginAsync fires with the triggering message. { var a = await NewSideAsync(); var (duplex, toSession, fromSession) = ScriptedPeer(); - var received = new TaskCompletionSource(); - a.Session.SnapshotReceiver = (begin, _, _) => - { - received.TrySetResult(begin.AsOfSeq); - return Task.CompletedTask; - }; + var applier = new CapturingApplier(); + a.Session.SnapshotApplier = applier; using var cts = new CancellationTokenSource(RunTimeout); var run = a.Session.RunAsync(duplex, cts.Token); @@ -591,14 +588,16 @@ public sealed class SyncSessionTests : IDisposable await DriveHandshakeAsync(a.Db, toSession, fromSession, cts.Token); await toSession.WriteAsync(new SyncMessage { SnapshotBegin = new SnapshotBegin { AsOfSeq = 42 } }, cts.Token); - Assert.Equal(42, await received.Task.WaitAsync(RunTimeout)); + Assert.Equal(42, await applier.BeganAsOf.Task.WaitAsync(RunTimeout)); cts.Cancel(); await SwallowAsync(run); } - // (c) snapshot required but no sender hook -> NotSupportedException. + // (c) this node owes a snapshot but no sender hook -> NotSupportedException. { var a = await NewSideAsync(); + await a.Store.SetNeedsSnapshotAsync(true, default); + a.Session.SnapshotSender = null; var (duplex, toSession, fromSession) = ScriptedPeer(); using var cts = new CancellationTokenSource(RunTimeout); @@ -607,14 +606,15 @@ public sealed class SyncSessionTests : IDisposable await NextAsync(fromSession, cts.Token); await toSession.WriteAsync(new SyncMessage { Handshake = MatchingHandshake(a.Db, "peer") }, cts.Token); await NextAsync(fromSession, cts.Token); - await toSession.WriteAsync(new SyncMessage { HandshakeAck = new HandshakeAck { NodeId = "peer", SnapshotRequired = true } }, cts.Token); + await toSession.WriteAsync(new SyncMessage { HandshakeAck = new HandshakeAck { NodeId = "peer", SnapshotRequired = false } }, cts.Token); await Assert.ThrowsAsync(() => run); } - // (d) inbound SnapshotBegin but no receiver hook -> NotSupportedException. + // (d) inbound SnapshotBegin but no applier hook -> NotSupportedException. { var a = await NewSideAsync(); + a.Session.SnapshotApplier = null; var (duplex, toSession, fromSession) = ScriptedPeer(); using var cts = new CancellationTokenSource(RunTimeout); @@ -627,6 +627,20 @@ public sealed class SyncSessionTests : IDisposable } } + private sealed class CapturingApplier : ISnapshotApplier + { + public TaskCompletionSource BeganAsOf { get; } = new(); + + public Task OnBeginAsync(SnapshotBegin begin, CancellationToken ct) + { + BeganAsOf.TrySetResult(begin.AsOfSeq); + return Task.CompletedTask; + } + + public Task OnBatchAsync(SnapshotBatch batch, CancellationToken ct) => Task.CompletedTask; + public Task OnCompleteAsync(SnapshotComplete complete, CancellationToken ct) => Task.CompletedTask; + } + [Fact] public async Task SecondHandshake_MidStream_IsProtocolError() {