fix(localdb): snapshot as_of_seq floors at last_acked_seq so a rebuilt peer's watermark is lifted — ends the full-snapshot-on-every-reconnect loop (0.3.4, ScadaBridge register row 106)

SnapshotStreamer.SendAsync stamped as_of_seq = MAX(seq), which reads 0 once ack-pruning has emptied a converged sender's oplog. A receiver whose watermark genuinely is 0 (a rebuilt node — ScadaBridge's site-a-a since its 2026-08-15 rebuild) was healed by the snapshot but never lifted by it (the receiver's watermark update is correctly MAX-guarded), so the 0.1.2 empty-oplog horizon rule saw 0 < last_acked_seq as a gap on every subsequent handshake and re-streamed the whole database on every reconnect until the sender wrote again. Observed live: the same 127k-row full snapshot on two consecutive restarts of the pair. Fix: as_of_seq = max(MAX(seq), last_acked_seq). Every delta pruned below last_acked_seq is already folded into the rows the snapshot carries, so that is the floor it may claim; the receiver's MAX guard then lifts the watermark and the next handshake finds no gap. Sender-side only; wire-compatible; no schema change. Test WipedPeer_SnapshotLiftsItsWatermark_SoTheNextReconnectSendsNoSecondSnapshot fails on the unfixed code exactly as the rig does (watermark 0, expected 3) and passes with the fix: the second episode exchanges only a delta, and a capturing logger on the snapshot streamer proves no snapshot was sent. Suite 216/216 on 8 of 9 runs; one unnamed single-test failure on the first run did not recur in eight further runs and could not be attributed.
2026-08-20 09:52:19 -04:00
parent 5e02c1c38f
commit 75c1897522
4 changed files with 93 additions and 5 deletions
@@ -52,14 +52,14 @@ public sealed class SnapshotResyncTests : IDisposable
    // A fresh session stack over an EXISTING database. Reconnecting after a session ends needs a new
    // session (RunAsync is single-use), so any scenario with two connection episodes over one node's
    // data — a peer being replaced between them — goes through here.
-    private static Side NewSessionOver(SqliteLocalDb db, ReplicationOptions? options = null)
+    private static Side NewSessionOver(SqliteLocalDb db, ReplicationOptions? options = null, ILogger? streamerLogger = null)
    {
        options ??= new ReplicationOptions { FlushInterval = TimeSpan.FromMilliseconds(20) };
        var store = new OplogStore(db, options);
        var applier = new LwwApplier(db);
        var session = new SyncSession(db, store, applier, options, NullLogger.Instance);

-        var streamer = new SnapshotStreamer(db, store, options, NullLogger.Instance);
+        var streamer = new SnapshotStreamer(db, store, options, streamerLogger ?? NullLogger.Instance);
        var snapshotApplier = new SnapshotApplier(db, applier, store, NullLogger.Instance);
        session.SnapshotSender = streamer.SendAsync;
        session.SnapshotApplier = snapshotApplier;
@@ -291,6 +291,74 @@ public sealed class SnapshotResyncTests : IDisposable
        await SwallowAsync(runB);
    }

+    [Fact]
+    public async Task WipedPeer_SnapshotLiftsItsWatermark_SoTheNextReconnectSendsNoSecondSnapshot()
+    {
+        // The follow-on to the wiped-peer gap above, found on the ScadaBridge docker rig 2026-08-20
+        // (its deferred-work register row 106): the snapshot DID heal the wiped node, but it was
+        // stamped as_of_seq = MAX(seq) of an oplog that ack-pruning had emptied — i.e. 0 — so the
+        // receiver's (correctly MAX-guarded) watermark stayed at 0. On the next handshake the
+        // healthy side's empty-oplog horizon rule (streamable from last_acked_seq + 1) saw
+        // 0 < last_acked_seq as a gap again and re-streamed the whole database — on EVERY reconnect,
+        // until the healthy side happened to write a new row (127k rows per restart on the rig).
+        // A full snapshot taken now covers every delta ever pruned below last_acked_seq, so that is
+        // the floor the as-of must carry.
+        var a = await NewSideAsync();
+        for (var i = 1; i <= 3; i++)
+            await a.Db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (@id, 'A', @id)", new { id = i });
+        await a.Store.RecordPeerAckAsync(await a.Store.GetMaxSeqAsync());
+        Assert.Equal(0, await OplogCount(a.Db));
+
+        var b = await NewSideAsync();
+
+        // Episode 1: the wiped peer is healed by a full snapshot.
+        var (d1A, d1B) = DuplexPair();
+        using (var cts1 = new CancellationTokenSource(RunTimeout))
+        {
+            var run1A = a.Session.RunAsync(d1A, cts1.Token);
+            var run1B = b.Session.RunAsync(d1B, cts1.Token);
+            await WaitForAsync(async () => (await ReadOrders(b.Db)).Count == 3, RunTimeout);
+            cts1.Cancel();
+            await SwallowAsync(run1A);
+            await SwallowAsync(run1B);
+        }
+
+        // The snapshot must have lifted B's watermark to everything A ever produced (seq 3), not 0.
+        Assert.Equal(3, await LastAppliedRemoteSeq(b.Db));
+
+        // Episode 2: a routine reconnect. A writes one more row so the delta path gives us a
+        // barrier; the snapshot streamer's logger proves whether a snapshot was sent at all.
+        var streamerLog = new CapturingLogger();
+        var a2 = NewSessionOver(a.Db, streamerLogger: streamerLog);
+        var b2 = NewSessionOver(b.Db);
+        var (d2A, d2B) = DuplexPair();
+        using var cts2 = new CancellationTokenSource(RunTimeout);
+        var run2A = a2.Session.RunAsync(d2A, cts2.Token);
+        var run2B = b2.Session.RunAsync(d2B, cts2.Token);
+        await WaitForAsync(() => Task.FromResult(b2.Session.PeerNodeId is not null), RunTimeout);
+        await a.Db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (4, 'DELTA', 4)");
+        await WaitForAsync(async () => (await ReadOrders(b.Db)).Count == 4, RunTimeout);
+
+        Assert.DoesNotContain(streamerLog.Messages, m => m.StartsWith("Snapshot sent", StringComparison.Ordinal));
+        Assert.Equal(4, await LastAppliedRemoteSeq(b.Db));
+
+        cts2.Cancel();
+        await SwallowAsync(run2A);
+        await SwallowAsync(run2B);
+    }
+
+    private sealed class CapturingLogger : ILogger
+    {
+        private readonly List<string> _messages = [];
+        public IReadOnlyList<string> Messages { get { lock (_messages) return _messages.ToList(); } }
+        public IDisposable? BeginScope<TState>(TState state) where TState : notnull => null;
+        public bool IsEnabled(LogLevel logLevel) => true;
+        public void Log<TState>(LogLevel logLevel, EventId eventId, TState state, Exception? exception, Func<TState, Exception?, string> formatter)
+        {
+            lock (_messages) _messages.Add(formatter(state, exception));
+        }
+    }
+
    [Fact]
    public async Task RebuiltPeer_OwnWritesReachTheHealthyNode_DespiteTheOldPeersWatermark()
    {