From 8befebd47645e5943a1f4673fee862997c03a2da Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Fri, 17 Jul 2026 23:50:42 -0400 Subject: [PATCH] =?UTF-8?q?fix(localdb):=20snapshot=20review=20fixes=20?= =?UTF-8?q?=E2=80=94=20MAX-guarded=20complete=20watermark,=20keyset=20pagi?= =?UTF-8?q?nation,=20phantom-row=20guard?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Claude-Session: https://claude.ai/code/session_01BL2Vu1ESDQ9SCN4gVKkdts --- .../Internal/OplogStore.cs | 9 ++- .../Internal/SnapshotStreamer.cs | 56 ++++++++++----- .../Internal/SyncSession.cs | 18 ++++- .../SnapshotResyncTests.cs | 70 +++++++++++++++++++ .../SyncSessionTests.cs | 5 +- 5 files changed, 135 insertions(+), 23 deletions(-) diff --git a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/OplogStore.cs b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/OplogStore.cs index ee28a65..88a090f 100644 --- a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/OplogStore.cs +++ b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/OplogStore.cs @@ -146,10 +146,15 @@ internal sealed class OplogStore(ILocalDb db, ReplicationOptions options, Func db.ExecuteAsync("UPDATE __localdb_peer_state SET needs_snapshot = @v WHERE id = 1", new { v = value ? 1 : 0 }, ct); - /// Records the highest remote seq applied locally plus the peer HLC observed with it; also stamps last_sync_utc. + /// + /// Records the highest remote seq applied locally plus the peer HLC observed with it; also stamps + /// last_sync_utc. Both watermarks are MAX-guarded (monotonic): a snapshot completing with a + /// lower as-of seq than a concurrently-applied delta must never regress the durable watermark. + /// public Task SetLastAppliedRemoteSeqAsync(long seq, long lastSeenHlc, CancellationToken ct = default) => db.ExecuteAsync( - "UPDATE __localdb_peer_state SET last_applied_remote_seq = @seq, last_seen_hlc = @hlc, last_sync_utc = @now WHERE id = 1", + "UPDATE __localdb_peer_state SET last_applied_remote_seq = MAX(last_applied_remote_seq, @seq), " + + "last_seen_hlc = MAX(last_seen_hlc, @hlc), last_sync_utc = @now WHERE id = 1", new { seq, hlc = lastSeenHlc, now = FormatUtc(_utcNow()) }, ct); private static string FormatUtc(DateTimeOffset value) => 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 index 0811d96..efb974b 100644 --- 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 @@ -26,7 +26,8 @@ internal interface ISnapshotApplier /// internal sealed class SnapshotStreamer(SqliteLocalDb db, OplogStore store, ReplicationOptions options, ILogger logger) { - public async Task SendAsync(Func send, CancellationToken ct) + /// Streams the full snapshot; returns the as-of oplog seq it covers, so the session can skip already-covered tail deltas. + public async Task SendAsync(Func send, CancellationToken ct) { var asOfSeq = await store.GetMaxSeqAsync(ct); await send(new SyncMessage { SnapshotBegin = new SnapshotBegin { AsOfSeq = asOfSeq } }, ct); @@ -35,26 +36,40 @@ internal sealed class SnapshotStreamer(SqliteLocalDb db, OplogStore store, Repli foreach (var table in db.ReplicatedTables.Values.OrderBy(t => t.Name, StringComparer.Ordinal)) { var sql = BuildPageSql(table); - var offset = 0; + // Keyset cursor (pk_json > last seen), not LIMIT/OFFSET: concurrent inserts/prunes + // between pages cannot shift the window, so no ledger row is skipped or re-sent. + var lastPk = ""; 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. + // Column 4 is NULL for tombstones AND for live rows whose joined table row is + // missing; 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); + new { t = table.Name, page = options.MaxBatchSize, lastPk }, ct); if (rows.Count == 0) break; + lastPk = rows[^1].PkJson; var batch = new SnapshotBatch { TableName = table.Name }; foreach (var row in rows) { + if (!row.IsTombstone && row.RowJson is null) + { + // Ledger says live but the LEFT JOIN found no table row: ledger/table + // divergence (e.g. an unreplicated manual delete). Emitting {"col":null,...} + // would fabricate a phantom row on the peer, so skip it defensively. + logger.LogWarning( + "Snapshot skipping live-ledger row {Table} {PkJson}: no matching table row (ledger/table divergence).", + batch.TableName, row.PkJson); + continue; + } var snapshotRow = new SnapshotRow { PkJson = row.PkJson, @@ -66,10 +81,10 @@ internal sealed class SnapshotStreamer(SqliteLocalDb db, OplogStore store, Repli snapshotRow.RowJson = row.RowJson; batch.Rows.Add(snapshotRow); } - await send(new SyncMessage { SnapshotBatch = batch }, ct); - totalRows += rows.Count; + if (batch.Rows.Count > 0) + await send(new SyncMessage { SnapshotBatch = batch }, ct); + totalRows += batch.Rows.Count; - offset += rows.Count; if (rows.Count < options.MaxBatchSize) break; } @@ -77,9 +92,16 @@ internal sealed class SnapshotStreamer(SqliteLocalDb db, OplogStore store, Repli 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. + // snapshot is on the wire — stops it re-streaming on the next reconnect. Narrow window: the + // messages are enqueued on the data lane, not yet transport-delivered, so a crash/drop after + // this clear could lose the snapshot with the flag already down. The reconnect handshake's + // gap check (peer watermark below the pruned horizon) re-detects that in every case where the + // oplog is non-empty; the sole undetectable corner is an EMPTY oplog (nothing to gap-check), + // where the peer stays behind until new local writes recreate a horizon. Accepted: closing it + // would need transport-level delivery acks for snapshot frames. await store.SetNeedsSnapshotAsync(false, ct); logger.LogInformation("Snapshot sent (as_of_seq {AsOfSeq}, {Rows} rows).", asOfSeq, totalRows); + return asOfSeq; } // Pages one table's rows by joining the row-version ledger to the live table on the PK columns @@ -93,11 +115,15 @@ internal sealed class SnapshotStreamer(SqliteLocalDb db, OplogStore store, Repli 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)}")) + ")"; + // A live ledger row whose join found no table row (first PK column NULL — PK columns are + // never NULL on a real join hit) also yields NULL row json, so the caller can detect the + // divergence instead of shipping a fabricated all-NULL row. + var joinMiss = $"t.{Ident(table.PkColumns[0])} IS NULL"; return "SELECT rv.pk_json, rv.hlc, rv.node_id, rv.is_tombstone, " + - $"CASE WHEN rv.is_tombstone = 1 THEN NULL ELSE {jsonObject} END " + + $"CASE WHEN rv.is_tombstone = 1 OR {joinMiss} 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"; + "WHERE rv.table_name = @t AND rv.pk_json > @lastPk ORDER BY rv.pk_json LIMIT @page"; } private static string Ident(string name) => "\"" + name.Replace("\"", "\"\"") + "\""; @@ -161,11 +187,9 @@ internal sealed class SnapshotApplier : ISnapshotApplier 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); + // SetLastAppliedRemoteSeqAsync is MAX-guarded, so a delta applied mid-snapshot with a seq + // above the snapshot's as-of (or an HLC above what the snapshot carried) is never regressed. + await _store.SetLastAppliedRemoteSeqAsync(complete.AsOfSeq, _maxHlc, 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 94ffe0f..ce9b91b 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 @@ -58,9 +58,10 @@ internal sealed class SyncSession /// /// 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 . + /// data lane (never the raw transport); returns the as-of oplog seq the snapshot covers so the + /// pump can skip already-covered tail deltas. Wired to . /// - public Func, CancellationToken, Task>? SnapshotSender { get; set; } + public Func, CancellationToken, Task>? SnapshotSender { get; set; } /// /// Consumes an inbound snapshot inline on the receive loop: SnapshotBegin opens the receive, @@ -72,6 +73,9 @@ internal sealed class SyncSession /// Highest seq the peer has acknowledged applying this session. public long PeerAckedSeq => Interlocked.Read(ref _peerAckedSeq); + /// Highest oplog seq handed to the wire (or covered by a sent snapshot). Test/metric observation only. + internal long SentThruSeq => Interlocked.Read(ref _sentThruSeq); + /// The peer node's identity, once the handshake has completed. Volatile: written on the session task, read from test/metric threads. public string? PeerNodeId { @@ -151,11 +155,19 @@ internal sealed class SyncSession // 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. + // Memory profile of a MUTUAL snapshot: while we send here (receive loop not yet running), + // the peer's inbound frames — potentially its entire snapshot — buffer in the unbounded + // duplex inbox until our send completes. Worst case one full peer snapshot held in memory; + // acceptable for the small LocalDb datasets this library targets. if (snapshotRequired) { if (SnapshotSender is null) throw new NotSupportedException("snapshot required but no snapshot sender configured"); - await SnapshotSender(dataSend, ct); + var snapshotAsOfSeq = await SnapshotSender(dataSend, ct); + // The snapshot already carries every row state up to as-of; skipping the pump past it + // avoids re-sending tail deltas the peer would only discard via LWW. + if (snapshotAsOfSeq > _sentThruSeq) + _sentThruSeq = snapshotAsOfSeq; } } 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 index 1e97319..3025964 100644 --- 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 @@ -167,6 +167,7 @@ public sealed class SnapshotResyncTests : IDisposable // Prune A's oplog below B's (empty) watermark and flag a resync: incremental deltas alone // can no longer carry rows 1..4 to B. Assert.True(await a.Store.EnforceCapsAsync()); + var asOfSeq = await a.Store.GetMaxSeqAsync(); var (da, db) = DuplexPair(); using var cts = new CancellationTokenSource(RunTimeout); @@ -175,6 +176,9 @@ public sealed class SnapshotResyncTests : IDisposable await WaitForAsync(async () => (await ReadOrders(b.Db)).Count == 6, RunTimeout); Assert.Equal(await ReadOrders(a.Db), await ReadOrders(b.Db)); + // The snapshot covered everything through asOfSeq, so the pump starts above it (the surviving + // tail oplog rows are not re-sent as deltas). + Assert.True(a.Session.SentThruSeq >= asOfSeq, $"SentThruSeq {a.Session.SentThruSeq} < asOfSeq {asOfSeq}"); // Deltas resume after the snapshot: a post-snapshot write on A reaches B incrementally. await a.Db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (7, 'POST', 7)"); @@ -323,6 +327,59 @@ public sealed class SnapshotResyncTests : IDisposable await Assert.ThrowsAsync(() => run); } + [Fact] + public async Task SnapshotComplete_WithoutBegin_IsProtocolError() + { + var a = await NewSideAsync(); + var (duplex, toSession, fromSession) = ScriptedPeer(); + + using var cts = new CancellationTokenSource(RunTimeout); + var run = a.Session.RunAsync(duplex, cts.Token); + + await DriveHandshakeAsync(a.Db, toSession, fromSession, cts.Token); + await toSession.WriteAsync(new SyncMessage { SnapshotComplete = new SnapshotComplete { AsOfSeq = 9 } }, cts.Token); + + await Assert.ThrowsAsync(() => run); + } + + [Fact] + public async Task SnapshotComplete_EmptyOplogAsOfZero_DoesNotRegress() + { + // A sender with an empty oplog streams as_of_seq 0; a receiver already at watermark 500 must + // keep 500 (MAX-guarded complete), or it would re-request everything on the next handshake. + var a = await NewSideAsync(); + await a.Store.SetLastAppliedRemoteSeqAsync(500, 0); + var (duplex, toSession, fromSession) = ScriptedPeer(); + + using var cts = new CancellationTokenSource(RunTimeout); + var run = a.Session.RunAsync(duplex, cts.Token); + + await DriveHandshakeAsync(a.Db, toSession, fromSession, cts.Token); + await toSession.WriteAsync(new SyncMessage { SnapshotBegin = new SnapshotBegin { AsOfSeq = 0 } }, cts.Token); + await toSession.WriteAsync(new SyncMessage + { + SnapshotBatch = new SnapshotBatch { TableName = "orders", Rows = { SnapRow(1, "SNAP", 6_000_000) } }, + }, cts.Token); + await toSession.WriteAsync(new SyncMessage { SnapshotComplete = new SnapshotComplete { AsOfSeq = 0 } }, cts.Token); + + // Barrier (seq 0 does not move the watermark): its ack proves Complete{0} was processed. + await toSession.WriteAsync(new SyncMessage + { + DeltaBatch = new DeltaBatch { Entries = { new OplogEntry + { + Seq = 0, TableName = "orders", PkJson = "{\"id\":2}", + RowJson = "{\"id\":2,\"sku\":\"BARRIER\",\"qty\":1}", Hlc = 6_000_001, NodeId = "peer", IsTombstone = false, + } } }, + }, cts.Token); + await NextOfCaseAsync(fromSession, SyncMessage.MsgOneofCase.DeltaAck, cts.Token); + + Assert.Equal(500, await LastAppliedRemoteSeq(a.Db)); + Assert.Equal(2, (await ReadOrders(a.Db)).Count); // the snapshot row still merged + + cts.Cancel(); + await SwallowAsync(run); + } + [Fact] public async Task DeltaDuringSnapshot_AppliesSafely() { @@ -359,6 +416,19 @@ public sealed class SnapshotResyncTests : IDisposable var rows = await ReadOrders(a.Db); Assert.Equal([1L, 2L, 3L], rows.Select(r => r.Id)); + // Barrier (seq 0 does not move the watermark): its ack proves Complete{200} was processed; + // the MAX-guarded complete must not have regressed the delta's 205. + await toSession.WriteAsync(new SyncMessage + { + DeltaBatch = new DeltaBatch { Entries = { new OplogEntry + { + Seq = 0, TableName = "orders", PkJson = "{\"id\":4}", + RowJson = "{\"id\":4,\"sku\":\"BARRIER\",\"qty\":1}", Hlc = 6_000_030, NodeId = "peer", IsTombstone = false, + } } }, + }, cts.Token); + await NextOfCaseAsync(fromSession, SyncMessage.MsgOneofCase.DeltaAck, cts.Token); + Assert.Equal(205, await LastAppliedRemoteSeq(a.Db)); + cts.Cancel(); await SwallowAsync(run); } 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 06b81b7..19aaa01 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 @@ -553,10 +553,11 @@ public sealed class SyncSessionTests : IDisposable await a.Store.SetNeedsSnapshotAsync(true, default); var (duplex, toSession, fromSession) = ScriptedPeer(); var sent = new TaskCompletionSource(); - a.Session.SnapshotSender = (send, token) => + a.Session.SnapshotSender = async (send, token) => { sent.TrySetResult(); - return send(new SyncMessage { SnapshotComplete = new SnapshotComplete { AsOfSeq = 7 } }, token); + await send(new SyncMessage { SnapshotComplete = new SnapshotComplete { AsOfSeq = 7 } }, token); + return 7; }; using var cts = new CancellationTokenSource(RunTimeout);