From 75c18975226a986b3c07a894737104de07180bbf Mon Sep 17 00:00:00 2001
From: Joseph Doherty <dohejw01@gmail.com>
Date: Thu, 20 Aug 2026 09:52:19 -0400
Subject: [PATCH] =?UTF-8?q?fix(localdb):=20snapshot=20as=5Fof=5Fseq=20floo?=
 =?UTF-8?q?rs=20at=20last=5Facked=5Fseq=20so=20a=20rebuilt=20peer's=20wate?=
 =?UTF-8?q?rmark=20is=20lifted=20=E2=80=94=20ends=20the=20full-snapshot-on?=
 =?UTF-8?q?-every-reconnect=20loop=20(0.3.4,=20ScadaBridge=20register=20ro?=
 =?UTF-8?q?w=20106)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

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.
---
 ZB.MOM.WW.LocalDb/Directory.Build.props       |  2 +-
 ZB.MOM.WW.LocalDb/README.md                   | 13 +++-
 .../Internal/SnapshotStreamer.cs              | 11 ++-
 .../SnapshotResyncTests.cs                    | 72 ++++++++++++++++++-
 4 files changed, 93 insertions(+), 5 deletions(-)

diff --git a/ZB.MOM.WW.LocalDb/Directory.Build.props b/ZB.MOM.WW.LocalDb/Directory.Build.props
index 4c7a35b..66165f7 100644
--- a/ZB.MOM.WW.LocalDb/Directory.Build.props
+++ b/ZB.MOM.WW.LocalDb/Directory.Build.props
@@ -5,7 +5,7 @@
     <Nullable>enable</Nullable>
     <ImplicitUsings>enable</ImplicitUsings>
     <LangVersion>latest</LangVersion>
-    <Version>0.3.3</Version>
+    <Version>0.3.4</Version>
     <ManagePackageVersionsCentrally>true</ManagePackageVersionsCentrally>
     <TreatWarningsAsErrors>true</TreatWarningsAsErrors>
     <!-- NU1510 (package-pruning advice): the Replication and Tests projects carry a
diff --git a/ZB.MOM.WW.LocalDb/README.md b/ZB.MOM.WW.LocalDb/README.md
index 27d46be..69e43bb 100644
--- a/ZB.MOM.WW.LocalDb/README.md
+++ b/ZB.MOM.WW.LocalDb/README.md
@@ -10,7 +10,7 @@ Change capture uses SQLite triggers writing a single HLC-stamped oplog inside th
 consumer's own write transaction (crash-consistent, non-bypassable). Convergence is
 last-writer-wins over a hybrid logical clock. All timestamps are UTC.
 
-## Packages (all `0.3.3`)
+## Packages (all `0.3.4`)
 
 | Package | Contents |
 |---|---|
@@ -20,6 +20,17 @@ last-writer-wins over a hybrid logical clock. All timestamps are UTC.
 
 > Status: **published to the Gitea feed; adopted by ScadaBridge (Phases 1+2) and OtOpcUa (Phase 1).**
 >
+> - `0.3.4` (2026-08-20) — **a converged pair no longer re-streams a full snapshot on every
+>   reconnect once the sender's oplog has been ack-pruned to empty.** Found on the ScadaBridge
+>   docker rig (its register row 106): a rebuilt node (watermark 0) was healed by a snapshot
+>   stamped `as_of_seq = MAX(seq)` — which is 0 on an empty oplog — so its MAX-guarded watermark
+>   stayed 0, and the healthy side's empty-oplog horizon rule (`0.1.2`: streamable from
+>   `last_acked_seq + 1`) saw a gap on every later handshake and re-sent the whole database
+>   (127k rows per restart on the rig) until it happened to write a new row. The as-of is now
+>   `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. Sender-side only,
+>   wire-compatible, no schema change; pinned by
+>   `WipedPeer_SnapshotLiftsItsWatermark_SoTheNextReconnectSendsNoSecondSnapshot`.
 > - `0.3.3` (2026-08-20) — two latent apply-path hazards from the ScadaBridge adversarial
 >   architecture review (P4-7). Neither had a live trigger in any current consumer; both are the
 >   kind that surface as silent divergence, not as an error. Wire-compatible: no proto change, no
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 40ab3fc..e876401 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
@@ -61,7 +61,16 @@ internal sealed class SnapshotStreamer(SqliteLocalDb db, OplogStore store, Repli
     /// </summary>
     public async Task<long> SendAsync(SnapshotScope scope, Func<SyncMessage, CancellationToken, Task> send, CancellationToken ct)
     {
-        var asOfSeq = await store.GetMaxSeqAsync(ct);
+        // The as-of is the highest seq this snapshot can stand in for. MAX(seq) alone is wrong once
+        // ack-pruning has emptied the oplog: it reads 0, and a receiver whose watermark genuinely IS
+        // 0 (a rebuilt node) is then never lifted by the snapshot — its MAX-guarded watermark stays
+        // 0, the empty-oplog horizon rule (streamable from last_acked_seq + 1) sees a gap on the
+        // next handshake, and the whole database is re-streamed on every reconnect until this node
+        // happens to write again. Every delta pruned below last_acked_seq is already folded into the
+        // rows this snapshot carries, so last_acked_seq is the floor the as-of must carry.
+        var asOfSeq = Math.Max(
+            await store.GetMaxSeqAsync(ct),
+            (await store.GetPeerStateAsync(ct)).LastAckedSeq);
         var begin = new SnapshotBegin { AsOfSeq = asOfSeq };
         begin.Tables.AddRange(scope.Tables);
         await send(new SyncMessage { SnapshotBegin = begin }, ct);
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 6fff574..5b1332e 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
@@ -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()
     {