Merge fix/localdb-wiped-peer-snapshot: LocalDb 0.1.3 rebuilt-peer replication fixes

Two defects about a node whose LocalDb file is lost, the second only reachable
once the first was fixed:

- 0.1.2 — a converged pair prunes its oplog to empty on ack, and snapshot
  detection read an empty oplog as 'no gap possible'. The steady state of a
  healthy pair was the one state from which a rebuilt node could never be
  back-filled.
- 0.1.3 — with back-fill working, the rebuilt node'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, so the healthy node's stale watermark made the
  sending side skip its whole oplog.

Both found by the OtOpcUa LocalDb Phase 1 live gate on the docker-dev rig and
reproduced offline as RED tests first. 149/149 pass.

Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW

ZB.MOM.WW.LocalDb/Directory.Build.props

@@ -5,7 +5,7 @@
     <Nullable>enable</Nullable>
     <ImplicitUsings>enable</ImplicitUsings>
     <LangVersion>latest</LangVersion>
-    <Version>0.1.1</Version>
+    <Version>0.1.3</Version>
     <ManagePackageVersionsCentrally>true</ManagePackageVersionsCentrally>
     <TreatWarningsAsErrors>true</TreatWarningsAsErrors>
     <!-- NU1510 (package-pruning advice): the Replication and Tests projects carry a

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.0`)
+## Packages (all `0.1.3`)
 
 | Package | Contents |
 |---|---|
@@ -18,7 +18,22 @@ last-writer-wins over a hybrid logical clock. All timestamps are UTC.
 | `ZB.MOM.WW.LocalDb.Contracts` | `localdb_sync.v1` proto + generated gRPC types (packable; a future peer could be non-.NET). |
 | `ZB.MOM.WW.LocalDb.Replication` | gRPC sync engine: passive `LocalDbSyncService`, initiating `SyncBackgroundService`, `MaintenanceBackgroundService`, LWW apply, watermark acks + pruning, snapshot resync, `ISyncStatus`, `LocalDbMetrics`, `AddZbLocalDbReplication()` / `MapZbLocalDbSync()`. |
 
-> Status: **built and published to the Gitea feed at `0.1.0` (2026-07-18); no app has adopted it yet.**
+> 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
+>   now measured against `last_acked_seq` when the oplog is empty.
+> - `0.1.1` (2026-07-20) — creates the parent directory of `LocalDb:Path` (a missing directory was
+>   a hard boot failure, `SQLite Error 14`).
 
 ## Quickstart
 

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);

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(
@@ -418,9 +445,20 @@ internal sealed class SyncSession
         var minRows = await _db.QueryAsync(
             "SELECT COALESCE(MIN(seq), 0) FROM __localdb_oplog", static r => r.GetInt64(0), null, ct);
         var minSeq = minRows[0];
-        // minSeq 0 => oplog empty. A gap exists when the peer's applied watermark falls below the
-        // oldest seq we can still stream (pruning discarded everything at or below its horizon).
-        return minSeq > 0 && peerHandshake.LastAppliedRemoteSeq < minSeq - 1;
+
+        // The oldest seq we could still stream as a delta. With rows in the oplog that is simply the
+        // oldest one; with an EMPTY oplog it is the next seq we will ever write — one past
+        // last_acked_seq, because ack-pruning is the only thing that empties the oplog without
+        // flagging needs_snapshot, and it deletes exactly seq <= last_acked_seq.
+        //
+        // Treating an empty oplog as "no gap possible" is what let a wiped peer starve: a converged
+        // pair prunes to empty on ack, so the healthy node held no delta at all, and a peer that
+        // came back with a zero watermark and an empty database was told it needed nothing. The
+        // steady state of a healthy pair was precisely the state that could not heal one.
+        var streamableFrom = minSeq > 0 ? minSeq : peerState.LastAckedSeq + 1;
+
+        // A gap exists when the peer's applied watermark falls below what we can still stream.
+        return peerHandshake.LastAppliedRemoteSeq < streamableFrom - 1;
     }
 
     private static async Task<SyncMessage> ReceiveOneAsync(SyncDuplex duplex, CancellationToken ct)

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);
@@ -146,6 +154,12 @@ public sealed class SnapshotResyncTests : IDisposable
         return r[0];
     }
 
+    private static async Task<long> OplogCount(SqliteLocalDb db)
+    {
+        var r = await db.QueryAsync("SELECT COUNT(*) FROM __localdb_oplog", x => x.GetInt64(0));
+        return r[0];
+    }
+
     private static SnapshotRow SnapRow(long id, string sku, long hlc, string nodeId = "peer") =>
         new()
         {
@@ -156,6 +170,95 @@ public sealed class SnapshotResyncTests : IDisposable
             IsTombstone = false,
         };
 
+    [Fact]
+    public async Task WipedPeer_AfterEverythingWasAckedAndPruned_StillGetsSnapshot()
+    {
+        // The gap the OtOpcUa Phase 1 live gate found (check 4). A converged pair prunes its oplog
+        // on ack, so the steady state is "no oplog at all". If the peer is then wiped — a rebuilt
+        // node, a lost volume — it comes back with an empty database and a zero watermark, and the
+        // healthy side has no delta left to send it. Snapshot detection compared the peer's
+        // watermark against the oldest oplog row, which does not exist, so it concluded no snapshot
+        // was needed and the wiped node stayed empty indefinitely: the one case where replication
+        // most obviously should heal a node was the case it could not.
+        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));
+        // Not the already-covered cap-exceeded path: nothing here flagged a resync.
+        Assert.Equal(0, await NeedsSnapshot(a.Db));
+
+        // A brand-new database is exactly what a wiped node is: no rows, watermark zero.
+        var b = await NewSideAsync();
+        Assert.Empty(await ReadOrders(b.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 == 3, RunTimeout);
+        Assert.Equal(await ReadOrders(a.Db), await ReadOrders(b.Db));
+
+        cts.Cancel();
+        await SwallowAsync(runA);
+        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()
     {