From 3b7489a7b0da9c2bebd6aaca0425bbde71f585c3 Mon Sep 17 00:00:00 2001
From: Joseph Doherty <dohejw01@gmail.com>
Date: Tue, 21 Jul 2026 02:06:03 -0400
Subject: [PATCH] fix(localdb): a rebuilt peer's own writes were silently
 dropped
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The other half of the rebuilt-peer story, and only reachable once 0.1.2 made
back-fill work: with the wiped node repopulated and writing again, its writes
never reached its peer.

last_applied_remote_seq is a watermark in the PEER's seq space, and a rebuilt
peer is a new database whose oplog numbers from 1. Two things then went wrong,
and both had to be fixed — the first alone leaves data stuck:

- The healthy node kept advertising the OLD peer's watermark, and the sending
  side seeds its pump from exactly that number, so the rebuilt node skipped its
  entire oplog. The watermark (and the observed peer clock) is now reset when
  the peer's node id changes. last_acked_seq is deliberately NOT reset: it
  describes our own oplog's pruned horizon and is what tells a rebuilt peer it
  needs a snapshot — clearing it would turn 0.1.2's back-fill back off.

- A peer claiming to have applied more of our stream than we have ever produced
  can only be remembering a previous incarnation of us. That claim is now
  rejected in favour of starting from the beginning; LWW makes the re-send
  harmless. This mirrors the clamp RecordPeerAckAsync already applies to acks.

Found on the OtOpcUa docker-dev rig, where the healthy node held watermark 10
while the rebuilt peer's oplog was seqs 1-3 and its last_acked stayed 0 — the
peer never accepted a single one. Covered by a test that converges with one
peer, replaces it with a fresh database, and requires a write on the new peer
to arrive: RED before this fix, green after. 149/149 pass.

Version 0.1.3.

Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW
---
 ZB.MOM.WW.LocalDb/Directory.Build.props       |  2 +-
 ZB.MOM.WW.LocalDb/README.md                   | 10 ++-
 .../Internal/OplogStore.cs                    | 36 +++++++++++
 .../Internal/SyncSession.cs                   | 31 +++++++++-
 .../SnapshotResyncTests.cs                    | 61 +++++++++++++++++++
 5 files changed, 136 insertions(+), 4 deletions(-)

diff --git a/ZB.MOM.WW.LocalDb/Directory.Build.props b/ZB.MOM.WW.LocalDb/Directory.Build.props
index 9fdd454..d18578d 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.1.2</Version>
+    <Version>0.1.3</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 f8b3558..8c9f970 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.1.2`)
+## Packages (all `0.1.3`)
 
 | Package | Contents |
 |---|---|
@@ -20,6 +20,14 @@ 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.1.3` (2026-07-21) — fixes the other half of the rebuilt-peer story: a rebuilt peer's OWN
+>   writes were silently dropped. `last_applied_remote_seq` is a watermark in the *peer's* seq
+>   space, and a rebuilt peer numbers from 1 again — so the healthy node advertised the old peer's
+>   watermark, the rebuilt node started its pump above it, and its first N writes were never sent.
+>   Now the watermark is reset when the peer's node id changes, and a peer claiming to have applied
+>   more of our stream than we ever produced is treated as evidence that *we* were rebuilt (pump
+>   restarts from the beginning; LWW makes re-sending harmless). Found on the OtOpcUa docker-dev rig
+>   — it only became reachable once `0.1.2` made back-fill work.
 > - `0.1.2` (2026-07-21) — fixes snapshot resync for a **wiped peer**: a converged pair prunes its
 >   oplog to empty on ack, and the handshake read an empty oplog as "no gap possible", so a peer
 >   that came back with an empty database and a zero watermark was never snapshotted. The gap is
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 424f383..b9d76a1 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
@@ -97,6 +97,20 @@ internal sealed class OplogStore(ILocalDb db, ReplicationOptions options, Func<D
         return (long)cmd.ExecuteScalar()!;
     }
 
+    /// <summary>
+    /// The highest seq this node has ever produced: the oplog's max, or — once acked rows have been
+    /// pruned away — the ack watermark that pruned them. Nothing we emit can exceed it, so a peer
+    /// claiming to have applied more than this is not talking about the stream we are producing now.
+    /// </summary>
+    public async Task<long> GetKnownMaxSeqAsync(CancellationToken ct = default)
+    {
+        var rows = await db.QueryAsync(
+            "SELECT MAX((SELECT COALESCE(MAX(seq), 0) FROM __localdb_oplog), " +
+            "(SELECT last_acked_seq FROM __localdb_peer_state WHERE id = 1))",
+            static r => r.GetInt64(0), null, ct);
+        return rows[0];
+    }
+
     /// <summary>The highest oplog seq present, or 0 when the oplog is empty.</summary>
     public async Task<long> GetMaxSeqAsync(CancellationToken ct = default)
     {
@@ -160,6 +174,28 @@ internal sealed class OplogStore(ILocalDb db, ReplicationOptions options, Func<D
     public Task SetPeerNodeIdAsync(string peerNodeId, CancellationToken ct = default) =>
         db.ExecuteAsync("UPDATE __localdb_peer_state SET peer_node_id = @v WHERE id = 1", new { v = peerNodeId }, ct);
 
+    /// <summary>
+    /// Forgets everything we knew about the previous peer's stream: the inbound seq watermark and
+    /// its observed clock. Called when the peer's node id changes.
+    /// </summary>
+    /// <remarks>
+    /// <para>
+    /// <c>last_applied_remote_seq</c> is a watermark in the PEER's seq space, and a rebuilt peer is
+    /// a new database whose oplog numbers from 1. Carrying the old peer's watermark across the
+    /// swap makes the new peer's early writes look already-applied, so they are dropped — silently,
+    /// and for as many writes as the old watermark was high.
+    /// </para>
+    /// <para>
+    /// <c>last_acked_seq</c> is deliberately NOT reset: it describes OUR oplog (what has been
+    /// pruned), not the peer's, and it is what tells a rebuilt peer's handshake that it needs a
+    /// snapshot. Clearing it would make an already-pruned node look like it had never written
+    /// anything and turn the back-fill back off.
+    /// </para>
+    /// </remarks>
+    public Task ResetInboundWatermarkAsync(CancellationToken ct = default) =>
+        db.ExecuteAsync(
+            "UPDATE __localdb_peer_state SET last_applied_remote_seq = 0, last_seen_hlc = 0 WHERE id = 1", null, ct);
+
     /// <summary>Sets or clears the snapshot-resync-required flag.</summary>
     public Task SetNeedsSnapshotAsync(bool value, CancellationToken ct = default) =>
         db.ExecuteAsync("UPDATE __localdb_peer_state SET needs_snapshot = @v WHERE id = 1", new { v = value ? 1 : 0 }, ct);
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 bf8068c..e9556bc 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
@@ -142,11 +142,38 @@ internal sealed class SyncSession
 
         ValidateHandshake(localHandshake, peerHandshake);
 
+        // A different node id on the far end means the peer was replaced, not merely restarted — a
+        // rebuilt host, a restored-from-nothing volume. Its oplog numbers from 1 again, so the
+        // watermark we hold (in the OLD peer's seq space) would silently swallow its first N
+        // writes. Forget it before anything reads it.
+        if (peerState.PeerNodeId is not null && peerState.PeerNodeId != peerHandshake.NodeId)
+        {
+            _logger.LogWarning(
+                "Replication peer identity changed ({OldPeer} -> {NewPeer}): the peer was rebuilt, so its seq space restarted. Resetting the inbound watermark.",
+                peerState.PeerNodeId, peerHandshake.NodeId);
+            await _store.ResetInboundWatermarkAsync(ct);
+            peerState = peerState with { LastAppliedRemoteSeq = 0, LastSeenHlc = 0 };
+        }
+
         await _store.SetPeerNodeIdAsync(peerHandshake.NodeId, ct);
         PeerNodeId = peerHandshake.NodeId;
         _status?.SetPeerNodeId(peerHandshake.NodeId);
-        _sentThruSeq = peerHandshake.LastAppliedRemoteSeq;
-        Interlocked.Exchange(ref _peerAckedSeq, peerHandshake.LastAppliedRemoteSeq);
+        // The peer's watermark decides where our pump starts, so an impossible claim silently
+        // skips real data. It can only exceed what we have ever produced if the peer is remembering
+        // a previous incarnation of THIS node — i.e. WE were rebuilt and our seqs restarted at 1 —
+        // in which case its number says nothing about the stream we hold now, and starting from it
+        // would skip our entire oplog. Fall back to the beginning; LWW makes re-sending harmless.
+        var knownMaxSeq = await _store.GetKnownMaxSeqAsync(ct);
+        var peerAppliedThruSeq = peerHandshake.LastAppliedRemoteSeq <= knownMaxSeq
+            ? peerHandshake.LastAppliedRemoteSeq
+            : 0;
+        if (peerAppliedThruSeq != peerHandshake.LastAppliedRemoteSeq)
+            _logger.LogWarning(
+                "Peer claims to have applied our seq {PeerClaim}, beyond the {KnownMax} we have ever produced — this node was rebuilt. Re-sending our oplog from the start.",
+                peerHandshake.LastAppliedRemoteSeq, knownMaxSeq);
+
+        _sentThruSeq = peerAppliedThruSeq;
+        Interlocked.Exchange(ref _peerAckedSeq, peerAppliedThruSeq);
 
         var snapshotRequired = await ComputeSnapshotRequiredAsync(peerHandshake, peerState, ct);
         await controlLane.WriteAsync(
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 3a78c9c..dbaa0d7 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
@@ -46,6 +46,14 @@ public sealed class SnapshotResyncTests : IDisposable
         await db.ExecuteAsync(OrdersSql);
         db.RegisterReplicated("orders");
 
+        return NewSessionOver(db, options);
+    }
+
+    // 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)
+    {
         options ??= new ReplicationOptions { FlushInterval = TimeSpan.FromMilliseconds(20) };
         var store = new OplogStore(db, options);
         var applier = new LwwApplier(db);
@@ -198,6 +206,59 @@ public sealed class SnapshotResyncTests : IDisposable
         await SwallowAsync(runB);
     }
 
+    [Fact]
+    public async Task RebuiltPeer_OwnWritesReachTheHealthyNode_DespiteTheOldPeersWatermark()
+    {
+        // The mirror image of the wiped-peer snapshot gap, found on the OtOpcUa docker-dev rig once
+        // back-fill started working and the rebuilt node could be observed writing again.
+        //
+        // last_applied_remote_seq is a watermark in the PEER's seq space. A rebuilt peer is a new
+        // database, so its seqs restart at 1 — but the healthy node kept the old peer's watermark
+        // and discarded everything at or below it. The rebuilt node's own writes were silently
+        // dropped until its counter climbed past a number that has nothing to do with it. Silent is
+        // the operative word: both nodes cache after every apply, so the pair looks converged right
+        // up until the moment only the rebuilt node applies something.
+        var a = await NewSideAsync();
+        var bOld = await NewSideAsync();
+
+        // Phase 1: converge with the ORIGINAL peer, so A's inbound watermark advances well past the
+        // seq numbers a rebuilt peer would start from.
+        for (var i = 1; i <= 5; i++)
+            await bOld.Db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (@id, 'OLD', @id)", new { id = i });
+
+        var (d1A, d1B) = DuplexPair();
+        using (var cts1 = new CancellationTokenSource(RunTimeout))
+        {
+            var run1A = a.Session.RunAsync(d1A, cts1.Token);
+            var run1B = bOld.Session.RunAsync(d1B, cts1.Token);
+            await WaitForAsync(async () => (await ReadOrders(a.Db)).Count == 5, RunTimeout);
+            cts1.Cancel();
+            await SwallowAsync(run1A);
+            await SwallowAsync(run1B);
+        }
+
+        Assert.True(await LastAppliedRemoteSeq(a.Db) >= 5);
+
+        // Phase 2: B is rebuilt — a new database, a new node id, and an oplog numbering from 1. Its
+        // write must still reach A.
+        var bNew = await NewSideAsync();
+        await bNew.Db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (99, 'REBUILT', 99)");
+
+        var a2 = NewSessionOver(a.Db);
+        var (d2A, d2B) = DuplexPair();
+        using var cts2 = new CancellationTokenSource(RunTimeout);
+        var run2A = a2.Session.RunAsync(d2A, cts2.Token);
+        var run2B = bNew.Session.RunAsync(d2B, cts2.Token);
+
+        await WaitForAsync(
+            async () => (await ReadOrders(a.Db)).Any(r => r.Id == 99 && r.Sku == "REBUILT"),
+            RunTimeout);
+
+        cts2.Cancel();
+        await SwallowAsync(run2A);
+        await SwallowAsync(run2B);
+    }
+
     [Fact]
     public async Task PrunedPeer_GetsSnapshot_ThenConverges()
     {