From 522cfbfda677a2b07a3e57bdc436933ac31ccf41 Mon Sep 17 00:00:00 2001
From: Joseph Doherty <dohejw01@gmail.com>
Date: Fri, 14 Aug 2026 23:34:55 -0400
Subject: [PATCH] fix(localdb): flush the HLC anchor on deregister, bound the
 sync inbox (0.2.1)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Two adversarial-review findings against 0.2.0. Wire-compatible with 0.1.x/0.2.0 peers:
no proto change, no lib_schema_version bump, no on-disk schema change.

F4 — DeregisterReplicated destroyed the HLC clock's only durable crash anchor. Clock
recovery on open is MAX(hlc_high_water, MAX(oplog.hlc), MAX(row_version.hlc),
last_seen_hlc); hlc_high_water is written only by Dispose, and deregistration deletes the
table's rows from BOTH ledgers. On a file whose registered tables have all been
deregistered, an ungraceful exit (crash, SIGKILL, power loss) therefore left nothing above
zero to recover from: the reopened clock restarts from wall-clock time and can re-issue
stamps below HLCs peers already hold, which LWW discards as older. The current clock is now
flushed inside the deregistration transaction, before the prune, as a separate command so
it cannot inflate the "did anything change?" answer. FlushHlcHighWater takes an optional
transaction (SQLite rejects a command with no transaction on a connection that has one) and
is now MAX-guarded — every writer of that column is flushing a monotonic clock, so a lower
value can only be stale.

F3 — the sync inbox was Channel.CreateUnbounded, making inbound memory a function of what
the PEER sends. The bad case is not hypothetical: a snapshot streams every row of every
registered table, needs_snapshot is per-database, and both sides of a pair that enables
replication late baseline unconditionally — so a mutual full snapshot of a large dataset
(an event log near its storage cap) was buffered whole, in memory, on both nodes at once.
The inbox is now bounded at 64 messages with wait-mode backpressure, so the ceiling is
~64 x MaxBatchBytes — a configured number rather than a peer-controlled one — and a full
inbox stops draining the gRPC stream, which closes the peer's flow-control window. The
pump writes with WriteAsync, not TryWrite: on a full bounded channel TryWrite drops the
message, and every inbound message is protocol state.

Bounding alone would have deadlocked the very case it was aimed at. The snapshot a node
owes was streamed from INSIDE the handshake, before the receive loop existed, so neither
side drained its inbox while sending; a mutual snapshot converged only because the inbox
was unbounded. The snapshot now runs as the first phase of the outbound task, started
alongside the receive loop, so the inbox is drained for the whole life of the session.
Interleaving needed no new safety: snapshot rows and deltas both merge through LWW and the
snapshot's watermark write is MAX-guarded. Ordering within the outbound lane is preserved
(the pump still starts only after the snapshot, from its as-of seq).

Per-table needs_snapshot (the finding's optional third part) is NOT done: the flag lives in
__localdb_peer_state and narrowing it needs an on-disk schema change, which this fix
deliberately avoids. Documented as a follow-up, and the cost is now written down where it
is decided — RegisterReplicated(baselineExistingRows) remarks and the README both say that
baselining one table re-streams every registered table in both directions, and that
high-volume tables should be trimmed before enabling replication late.

188/188 tests pass (was 184). New coverage: the deregistered clock anchor surviving an
ungraceful reopen (a second instance over the same file with no Dispose — fails without the
flush, recovering at 0), the flush not counting as work done, bounded-inbox backpressure
under a slow reader (the pump stalls at the bound and loses nothing, proven by contiguous
in-order delivery), and a mutual snapshot over a capacity-1 transport converging — that
last one deadlocks and times out against the old handshake-phase snapshot ordering.

Version 0.2.1.
---
 ZB.MOM.WW.LocalDb/Directory.Build.props       |  2 +-
 ZB.MOM.WW.LocalDb/README.md                   | 60 +++++++++++++----
 .../Internal/GrpcSyncDuplex.cs                | 58 ++++++++++++++---
 .../Internal/SyncSession.cs                   | 64 +++++++++++++-----
 .../src/ZB.MOM.WW.LocalDb/ILocalDb.cs         | 22 +++++++
 .../Internal/LocalDbSchema.cs                 | 16 ++++-
 .../Internal/SqliteLocalDb.cs                 | 20 ++++++
 .../GrpcAdapterTests.cs                       | 65 +++++++++++++++++++
 .../RegistrationTests.cs                      | 57 ++++++++++++++++
 .../SnapshotResyncTests.cs                    | 55 +++++++++++++++-
 10 files changed, 376 insertions(+), 43 deletions(-)

diff --git a/ZB.MOM.WW.LocalDb/Directory.Build.props b/ZB.MOM.WW.LocalDb/Directory.Build.props
index c41fe2f..c8eccfa 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.2.0</Version>
+    <Version>0.2.1</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 7f0f12c..3fbb6bb 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.2.0`)
+## Packages (all `0.2.1`)
 
 | Package | Contents |
 |---|---|
@@ -20,6 +20,27 @@ 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.2.1` (2026-08-14) — two adversarial-review findings against `0.2.0`. Wire-compatible with
+>   `0.1.x`/`0.2.0` peers; no proto change, no `lib_schema_version` bump, no on-disk schema change.
+>   - **`DeregisterReplicated` no longer leaves the HLC without a crash anchor.** Clock recovery on
+>     open is `MAX(hlc_high_water, oplog HLCs, ledger HLCs, last_seen_hlc)`, and `hlc_high_water` was
+>     written only by graceful disposal — while deregistration deletes the table's oplog AND ledger
+>     rows. A crash after a deregister boot could therefore recover the clock below stamps the peer
+>     already holds, and LWW discards those as older. The current clock is now flushed inside the
+>     deregistration transaction, before the prune (and the flush is MAX-guarded, so it can never
+>     regress the anchor).
+>   - **The sync inbox is bounded (64 messages) instead of unbounded**, so inbound memory is capped
+>     at roughly `64 x MaxBatchBytes` rather than "whatever the peer sends" — the bad case being a
+>     mutual full snapshot of a large dataset, buffered whole on both nodes at once. A full inbox now
+>     stops draining the gRPC stream, which closes the peer's flow-control window.
+>   - **A snapshot this node owes is streamed alongside a running receive loop**, not from inside the
+>     handshake. That is what makes bounding safe: previously neither side drained its inbox while
+>     sending, so a mutual snapshot converged only because the inbox was unbounded. Interleaving was
+>     already safe (snapshot rows and deltas both merge through LWW, the watermark write is
+>     MAX-guarded) — only the ordering changed.
+>   - **Documented the cost of late baselining**: the snapshot flag is per-database, so baselining
+>     one table re-streams every registered table, in both directions on a late-enabled pair. Trim
+>     high-volume tables (event logs) before turning replication on late.
 > - `0.2.0` (2026-08-14) — performance + lifecycle work from the ScadaBridge architectural review.
 >   All of it is wire-compatible with `0.1.x` peers (no proto change, no `lib_schema_version` bump);
 >   the on-disk bookkeeping schema goes to **v2** and is upgraded in place on open.
@@ -157,7 +178,9 @@ All options are validated at startup (`ZB.MOM.WW.Configuration` validators).
 - **Resync** — when a peer's watermark falls below the pruned oplog horizon, it is sent a
   full snapshot of the registered tables (streamed, keyset-paged) and deltas resume above
   the snapshot's as-of seq. **Each node decides from its OWN state** whether it owes the
-  peer a snapshot.
+  peer a snapshot. The snapshot is streamed *after* the handshake, alongside a running
+  receive loop, so a node sending one still applies what the peer sends it — a mutual
+  snapshot exchanges under flow control instead of being buffered whole in memory.
 - **Maintenance** — a background service on **both** roles enforces the oplog row/age caps
   and prunes expired tombstones every `MaintenanceInterval`.
 
@@ -191,11 +214,20 @@ All options are validated at startup (`ZB.MOM.WW.Configuration` validators).
   every comparison against a real remote write and wins only where the peer has no version
   of that key at all. Two nodes that baseline the same key both hold HLC `0` and resolve by
   the node-id tie-break — convergent, but arbitrary as to which content survives.
+- **The snapshot flag is per-database, not per-table.** Baselining one table flags a resync
+  that streams **every** registered table in full, and on a pair enabling replication late
+  both nodes do that to each other simultaneously — the largest transfer this library
+  performs. Bounded (batch budget out, bounded inbox in), but **purge or trim high-volume
+  tables such as event logs before enabling replication late**, and expect one slow first
+  sync. A per-table flag needs an on-disk schema change and is not in `0.2.x`.
 - **REAL ±Infinity** is carried in SQLite's JSON dialect on the wire (fine for the
   SQLite↔SQLite peers this library targets).
 - `AddZbLocalDb` / `AddZbLocalDbReplication` are **first-registration-wins** (idempotent).
-- A **mutual simultaneous snapshot** buffers the peer's inbound snapshot in memory while
-  this node is still sending its own (acceptable for the small datasets this lib targets).
+- **Inbound buffering is bounded, not zero.** The duplex inbox holds up to 64 messages, so
+  peak inbound memory is roughly `64 x MaxBatchBytes` (~128 MB at the 2 MB default, far
+  less in practice — messages are only that large when the sender has that much backlog).
+  Lower `MaxBatchBytes` to lower the ceiling. Beyond the bound the stream stops being
+  drained and the peer is slowed by flow control.
 
 ## Observability
 
@@ -231,7 +263,10 @@ provider is wired or the poll threw, so a DB failure never reads as "0 backlog")
   registered-table digests fail-closed, so a node that drops a table its peer still
   replicates stops syncing with a schema-mismatch error. Deregister on both nodes, or on
   neither. Re-enabling later should pass `baselineExistingRows: true`, or the rows that
-  accumulated while capture was off stay invisible to the peer.
+  accumulated while capture was off stay invisible to the peer — but trim high-volume
+  tables first (see Limitations: baselining flags a full, all-tables snapshot). The HLC's
+  durable anchor is flushed in the same transaction, so a crash after a deregister boot
+  cannot bring the clock back below stamps peers already hold.
 - **On-disk schema upgrades** are applied in place when the file is opened, in one
   transaction, and every step is safe to re-run. `schema_version` in `__localdb_meta` is
   local-file state only — it is NOT the wire `lib_schema_version`, so a pair mid-upgrade
@@ -246,13 +281,14 @@ provider is wired or the poll threw, so a DB failure never reads as "0 backlog")
 
 ## Testing
 
-184 tests, all offline / no live dependencies — unit (HLC ordering + restart
-persistence, trigger SQL generation, LWW decision table, watermark/pruning, digest
-computation, schema-version upgrade, deregistration idempotency, baseline-floor LWW
-safety, O(1) depth vs a seeded backlog, byte-budget chunking), integration over two real
-SQLite DBs on an in-memory (and real) gRPC pipe (bidirectional convergence, delete/update
-races, offline-accumulate → reconnect, prune → snapshot resync), plus seeded randomized
-property tests (random interleaved op sequences → both DBs identical after quiesce).
+188 tests, all offline / no live dependencies — unit (HLC ordering + restart
+persistence, deregistration's HLC anchor surviving an ungraceful reopen, trigger SQL
+generation, LWW decision table, watermark/pruning, digest computation, schema-version
+upgrade, deregistration idempotency, baseline-floor LWW safety, O(1) depth vs a seeded
+backlog, byte-budget chunking, bounded-inbox backpressure under a slow reader), integration
+over two real SQLite DBs on an in-memory (and real) gRPC pipe (bidirectional convergence,
+delete/update races, offline-accumulate → reconnect, prune → snapshot resync, mutual
+snapshot over a capacity-1 transport), plus seeded randomized property tests (random interleaved op sequences → both DBs identical after quiesce).
 Family conventions apply (`TreatWarningsAsErrors`).
 
 Build / test from this directory:
diff --git a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/GrpcSyncDuplex.cs b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/GrpcSyncDuplex.cs
index 8aa64a3..0a20aa6 100644
--- a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/GrpcSyncDuplex.cs
+++ b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/GrpcSyncDuplex.cs
@@ -12,19 +12,58 @@ namespace ZB.MOM.WW.LocalDb.Replication.Internal;
 internal static class GrpcSyncDuplex
 {
     /// <summary>
-    /// Wires <paramref name="reader"/> into an unbounded inbox channel drained by a background loop,
-    /// and exposes <paramref name="writer"/> as the duplex send. Returns the duplex plus the reader
-    /// loop's task so the caller can observe it. On clean stream end the inbox channel is completed;
-    /// on a reader fault it is completed WITH the exception — the session's handshake and receive
-    /// phases both await the inbox, so terminating it is what unblocks them when the call dies (they
-    /// would otherwise hang on a dead stream).
+    /// Inbound messages buffered between the transport and the session's receive loop.
+    /// </summary>
+    /// <remarks>
+    /// <para>
+    /// <b>Why bounded at all.</b> An unbounded inbox makes this process's memory a function of what
+    /// the PEER decides to send. The worst case is not hypothetical: a full snapshot streams every
+    /// row of every registered table, and both sides of a pair that enabled replication late
+    /// snapshot each other simultaneously — so an unbounded inbox holds an entire peer dataset (an
+    /// event log at its storage cap, say) in memory on both nodes at once. Bounding it converts that
+    /// into backpressure: the pump stops draining the gRPC stream, the HTTP/2 flow-control window
+    /// closes, and the SENDER slows down.
+    /// </para>
+    /// <para>
+    /// <b>Why 64.</b> Memory here is bounded by roughly capacity x
+    /// <see cref="ReplicationOptions.MaxBatchBytes"/> (the sender's own per-message budget), so the
+    /// ceiling is a configured number rather than a peer-controlled one. 64 is four times the
+    /// sender's outbound data-lane depth, so a healthy stream never reaches the bound and pays
+    /// nothing; it engages only when this node is genuinely slower than the peer for a sustained
+    /// burst, which is exactly when backpressure is wanted. Lower <c>MaxBatchBytes</c> to lower the
+    /// ceiling.
+    /// </para>
+    /// <para>
+    /// <b>What makes bounding safe.</b> Backpressure on the inbox deadlocks any design where both
+    /// peers can stop draining their inbox at the same time — mutual snapshots were exactly that
+    /// case while the send ran inside the handshake. <see cref="SyncSession"/> therefore starts its
+    /// receive loop BEFORE streaming a snapshot it owes, so the inbox is drained continuously for
+    /// the whole life of the session; see the remarks on its snapshot phase.
+    /// </para>
+    /// </remarks>
+    internal const int InboxCapacity = 64;
+
+    /// <summary>
+    /// Wires <paramref name="reader"/> into a bounded inbox channel (see
+    /// <see cref="InboxCapacity"/>) drained by a background loop, and exposes
+    /// <paramref name="writer"/> as the duplex send. Returns the duplex plus the reader loop's task
+    /// so the caller can observe it. On clean stream end the inbox channel is completed; on a reader
+    /// fault it is completed WITH the exception — the session's handshake and receive phases both
+    /// await the inbox, so terminating it is what unblocks them when the call dies (they would
+    /// otherwise hang on a dead stream).
     /// </summary>
     public static (SyncDuplex Duplex, Task ReaderTask) Create(
         IAsyncStreamReader<SyncMessage> reader,
         IAsyncStreamWriter<SyncMessage> writer,
         CancellationToken ct)
     {
-        var inbox = Channel.CreateUnbounded<SyncMessage>(new UnboundedChannelOptions { SingleWriter = true });
+        var inbox = Channel.CreateBounded<SyncMessage>(new BoundedChannelOptions(InboxCapacity)
+        {
+            // Wait, never Drop: every inbound message is protocol state (a delta the peer will not
+            // resend in-session, a snapshot frame, an ack). Dropping one is silent divergence.
+            FullMode = BoundedChannelFullMode.Wait,
+            SingleWriter = true,
+        });
         var readerTask = PumpInboundAsync(reader, inbox.Writer, ct);
         var duplex = new SyncDuplex
         {
@@ -44,7 +83,10 @@ internal static class GrpcSyncDuplex
         try
         {
             while (await reader.MoveNext(ct))
-                inbox.TryWrite(reader.Current);
+                // WriteAsync, not TryWrite: on a FULL bounded channel TryWrite returns false and the
+                // message is silently lost. Awaiting here is the backpressure — it stops draining the
+                // gRPC stream, which closes the peer's flow-control window.
+                await inbox.WriteAsync(reader.Current, ct);
         }
         catch (Exception ex)
         {
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 419cd23..5f474b9 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
@@ -10,11 +10,13 @@ namespace ZB.MOM.WW.LocalDb.Replication.Internal;
 /// <summary>
 /// The symmetric 2-node sync protocol as one state machine: both the gRPC service and the gRPC
 /// client run this identical logic over a <see cref="SyncDuplex"/>. It performs the fail-closed
-/// handshake, then runs a bidirectional steady state — an outbound change pump and an inbound
-/// apply/ack loop — until the token is cancelled or the stream ends. All outbound traffic flows
-/// through a single writer task fed by two lanes: an unbounded CONTROL lane (handshake messages,
-/// acks) and a bounded DATA lane (delta/snapshot batches). Receiving therefore never blocks on the
-/// ability to write — acks always fit — while the bounded data lane is the pump's backpressure.
+/// handshake, then runs a bidirectional steady state until the token is cancelled or the stream
+/// ends: an outbound phase (the full snapshot this node owes, if any, then the change pump) and an
+/// inbound apply/ack loop, STARTED TOGETHER so the inbox keeps draining even while a snapshot is
+/// being sent. All outbound traffic flows through a single writer task fed by two lanes: an
+/// unbounded CONTROL lane (handshake messages, acks) and a bounded DATA lane (delta/snapshot
+/// batches). Receiving therefore never blocks on the ability to write — acks always fit — while the
+/// bounded data lane is the pump's backpressure.
 /// Local writes never block on the peer; the first fault tears down all loops and propagates.
 /// </summary>
 internal sealed class SyncSession
@@ -97,10 +99,19 @@ internal sealed class SyncSession
         var writer = WriterLoopAsync(duplex, control.Reader, data.Reader, linked.Token);
         try
         {
-            await HandshakeAsync(duplex, control.Writer, GatedDataSend, linked.Token);
+            var snapshotRequired = await HandshakeAsync(duplex, control.Writer, linked.Token);
 
-            var pump = PumpLoopAsync(data.Writer, linked.Token);
+            // The receive loop starts BEFORE any snapshot this node owes is streamed. That ordering
+            // is load-bearing, not incidental: a snapshot sent from inside the handshake left the
+            // inbox undrained for its whole duration, so a MUTUAL snapshot (both sides flagged —
+            // e.g. both baselined an existing dataset) required an unbounded inbox to avoid
+            // deadlock, and therefore buffered a whole peer dataset in memory. With the receive loop
+            // running throughout, the peer's snapshot is applied as it arrives, the inbox can be
+            // bounded (see GrpcSyncDuplex.InboxCapacity), and mutual snapshots exchange under
+            // ordinary flow control. Interleaving is already safe by design: snapshot rows and
+            // deltas both merge through LWW, and the snapshot's watermark write is MAX-guarded.
             var receive = ReceiveLoopAsync(duplex, control.Writer, linked.Token);
+            var pump = SnapshotThenPumpLoopAsync(snapshotRequired, GatedDataSend, data.Writer, linked.Token);
             var loops = new[] { pump, receive, writer };
 
             // The first COMPLETED task is the primary outcome. Cancel the siblings, observe every
@@ -123,9 +134,14 @@ internal sealed class SyncSession
         }
     }
 
-    private async Task HandshakeAsync(
-        SyncDuplex duplex, ChannelWriter<SyncMessage> controlLane,
-        Func<SyncMessage, CancellationToken, Task> dataSend, CancellationToken ct)
+    /// <summary>
+    /// Runs the two-phase fail-closed handshake and returns whether THIS node owes the peer a full
+    /// snapshot (its own computation from its own state — the peer's answer drives the peer's send
+    /// direction, symmetrically). Sending that snapshot is deliberately NOT part of the handshake;
+    /// see <see cref="RunAsync"/>.
+    /// </summary>
+    private async Task<bool> HandshakeAsync(
+        SyncDuplex duplex, ChannelWriter<SyncMessage> controlLane, CancellationToken ct)
     {
         var peerState = await _store.GetPeerStateAsync(ct);
         var localHandshake = BuildHandshake(peerState);
@@ -184,13 +200,25 @@ internal sealed class SyncSession
             throw new InvalidOperationException(
                 $"Replication protocol error: expected a HandshakeAck, got {peerAckMsg.MsgCase}.");
 
-        // 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.
+        return snapshotRequired;
+    }
+
+    /// <summary>
+    /// The outbound steady state: stream the full snapshot this node owes the peer (if any), then
+    /// run the incremental delta pump. One task, because the ordering between them is a protocol
+    /// requirement — the pump must not put deltas on the data lane ahead of the snapshot, and it
+    /// starts from the as-of seq the snapshot covered.
+    /// </summary>
+    /// <remarks>
+    /// Split out of the handshake so the receive loop is already draining the inbox while this
+    /// streams (see <see cref="RunAsync"/>). Nothing else about the snapshot changes: it goes
+    /// through the same gated data lane, so it is still bounded by the writer's flow control rather
+    /// than racing the transport.
+    /// </remarks>
+    private async Task SnapshotThenPumpLoopAsync(
+        bool snapshotRequired, Func<SyncMessage, CancellationToken, Task> dataSend,
+        ChannelWriter<SyncMessage> dataLane, CancellationToken ct)
+    {
         if (snapshotRequired)
         {
             if (SnapshotSender is null)
@@ -203,6 +231,8 @@ internal sealed class SyncSession
             if (snapshotAsOfSeq > _sentThruSeq)
                 _sentThruSeq = snapshotAsOfSeq;
         }
+
+        await PumpLoopAsync(dataLane, ct);
     }
 
     // The ONLY caller of duplex.Send. Control (handshake + acks) always preempts data so the peer's
diff --git a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/ILocalDb.cs b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/ILocalDb.cs
index 0e07b77..c0e1125 100644
--- a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/ILocalDb.cs
+++ b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/ILocalDb.cs
@@ -79,6 +79,18 @@ public interface ILocalDb
     /// ledger entry — captured normally, or seeded on an earlier boot — keeps its genuine version
     /// and no snapshot is flagged. Passing true on every boot is therefore free after the first.
     /// </para>
+    /// <para>
+    /// <b>Cost, and why it is worth planning for.</b> The snapshot flag is a single per-DATABASE
+    /// flag, and a snapshot streams EVERY registered table in full — so baselining one table
+    /// re-sends all of them, and a pair that enables replication late has both nodes doing that to
+    /// each other at once. On a database holding a high-volume table (an event log at its retention
+    /// cap, say) that is the largest transfer this library ever performs, in both directions
+    /// simultaneously. It is bounded work, not runaway: outbound messages respect
+    /// <c>MaxBatchBytes</c>, and the receive side is bounded by the transport's inbox. But the
+    /// operational advice is to <b>purge or trim high-volume tables before enabling replication
+    /// late</b>, and to expect one slow first sync rather than a slow steady state. A per-table
+    /// snapshot flag would narrow this; it needs an on-disk schema change and is not in 0.2.x.
+    /// </para>
     /// </remarks>
     ReplicatedTable RegisterReplicated(string tableName, bool baselineExistingRows);
 
@@ -110,6 +122,16 @@ public interface ILocalDb
     /// schema-mismatch error, not silent divergence. Deregister on both nodes, or on neither.
     /// </para>
     /// <para>
+    /// <b>The HLC's durable crash anchor is flushed as part of the same transaction</b>, before the
+    /// prune. The clock recovers on open as the MAX of <c>hlc_high_water</c>, the oplog's HLCs, the
+    /// row-version ledger's HLCs and the last-seen peer HLC — and <c>hlc_high_water</c> is otherwise
+    /// written only on graceful disposal. Since this API deletes the deregistered table's oplog and
+    /// ledger rows, a file with no other registered table would, after an UNGRACEFUL exit (crash,
+    /// SIGKILL, power loss), have nothing left to recover from: the clock would restart from
+    /// wall-clock time and could re-issue stamps below HLCs peers already hold, which LWW then
+    /// discards as older. Flushing here removes that window.
+    /// </para>
+    /// <para>
     /// <b>Backlog-depth residual.</b> Pruning one table's rows can leave holes in the oplog's seq
     /// range, and the O(1) depth estimate assumes the retained range is contiguous, so it can
     /// over-report until the peer acks past the holes. Over-reporting is conservative — it can only
diff --git a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/Internal/LocalDbSchema.cs b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/Internal/LocalDbSchema.cs
index 2a55a20..8be02d4 100644
--- a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/Internal/LocalDbSchema.cs
+++ b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/Internal/LocalDbSchema.cs
@@ -169,10 +169,22 @@ internal static class LocalDbSchema
         return (reader.GetString(0), reader.GetInt64(1));
     }
 
-    internal static void FlushHlcHighWater(SqliteConnection connection, long value)
+    /// <summary>
+    /// Writes the clock's durable crash anchor. <paramref name="transaction"/> must be supplied when
+    /// the connection has one open (SQLite commands are rejected otherwise), which is what lets a
+    /// caller flush the anchor atomically with a change that would otherwise erase the clock's other
+    /// recovery evidence — see <see cref="ILocalDb.DeregisterReplicated"/>.
+    /// </summary>
+    /// <remarks>
+    /// MAX-guarded rather than an assignment: every writer of this column is flushing a monotonic
+    /// clock, so a lower value can only be a stale one, and the anchor exists precisely to never
+    /// regress.
+    /// </remarks>
+    internal static void FlushHlcHighWater(SqliteConnection connection, long value, SqliteTransaction? transaction = null)
     {
         using var cmd = connection.CreateCommand();
-        cmd.CommandText = "UPDATE __localdb_meta SET hlc_high_water = $value WHERE id = 1";
+        cmd.Transaction = transaction;
+        cmd.CommandText = "UPDATE __localdb_meta SET hlc_high_water = MAX(hlc_high_water, $value) WHERE id = 1";
         cmd.Parameters.AddWithValue("$value", value);
         cmd.ExecuteNonQuery();
     }
diff --git a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/Internal/SqliteLocalDb.cs b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/Internal/SqliteLocalDb.cs
index 9a3c893..9419039 100644
--- a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/Internal/SqliteLocalDb.cs
+++ b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/Internal/SqliteLocalDb.cs
@@ -255,6 +255,18 @@ internal sealed class SqliteLocalDb : ILocalDb, IDisposable
                 drop.ExecuteNonQuery();
             }
 
+            // Flushed BEFORE the prune below, in the SAME transaction, because that prune destroys
+            // the clock's crash evidence. Recovery on open takes a MAX over hlc_high_water, the
+            // oplog's HLCs, the row-version ledger's HLCs and last_seen_hlc — and hlc_high_water is
+            // otherwise written only by Dispose. Deregistering a table deletes that table's oplog
+            // AND ledger rows, so on a file whose registered tables are all deregistered, an
+            // UNGRACEFUL exit (crash, SIGKILL, power loss — no Dispose) afterwards leaves nothing
+            // above zero to recover from: the reopened clock would restart at wall-clock time and
+            // could re-issue HLCs below stamps peers have already observed, which LWW reads as
+            // "older" and discards. One row-level UPDATE on an administrative one-shot buys the
+            // anchor back; it is a separate command from the prune so it cannot inflate prunedRows.
+            LocalDbSchema.FlushHlcHighWater(_master, _clock.Current, tx);
+
             // Full scan of the oplog by table_name — the index that would have served it is exactly
             // the one schema v2 drops, and paying a write on every capture to speed up a one-shot
             // administrative prune is the wrong trade.
@@ -295,6 +307,14 @@ internal sealed class SqliteLocalDb : ILocalDb, IDisposable
     /// the oplog on the first boot after opting in, likely blowing <c>MaxOplogRows</c> and forcing
     /// the same snapshot anyway, via a cap-prune.
     /// </para>
+    /// <para>
+    /// <b>The flag is per-DATABASE, not per-table</b>, and a snapshot streams every registered table
+    /// in full — so baselining one table re-sends all of them, and both nodes of a late-enabled pair
+    /// do it to each other at once. Bounded (batch budget outbound, bounded inbox inbound) but
+    /// potentially the largest transfer this library performs; see
+    /// <see cref="ILocalDb.RegisterReplicated(string, bool)"/> for the operational advice. Narrowing
+    /// it to a per-table flag needs an on-disk schema change and is deferred.
+    /// </para>
     /// </remarks>
     private void BaselineExistingRows(SqliteTransaction tx, ReplicatedTable table)
     {
diff --git a/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/GrpcAdapterTests.cs b/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/GrpcAdapterTests.cs
index 9d8e182..41381d7 100644
--- a/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/GrpcAdapterTests.cs
+++ b/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/GrpcAdapterTests.cs
@@ -173,6 +173,71 @@ public sealed class GrpcAdapterTests : IAsyncLifetime
         Assert.True(sw.Elapsed < TimeSpan.FromSeconds(3), $"teardown took {sw.Elapsed}");
     }
 
+    // ---- inbox backpressure ------------------------------------------------------------------
+
+    [Fact]
+    public async Task Inbox_SlowReader_StopsDrainingTheStream_AndLosesNothing()
+    {
+        // The pump must STOP pulling from the gRPC stream once the inbox is full — that stall is
+        // what closes the peer's flow-control window and bounds this process's memory at
+        // InboxCapacity x MaxBatchBytes instead of "whatever the peer decides to send" (a full
+        // snapshot of a large dataset, on both nodes at once, when a pair baselines late).
+        const int capacity = Replication.Internal.GrpcSyncDuplex.InboxCapacity;
+        using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(15));
+        var reader = new CountingReader();
+        var (duplex, readerTask) = Replication.Internal.GrpcSyncDuplex.Create(
+            reader, new HangingWriter(), cts.Token);
+
+        // Nothing reads the inbox: the pump fills it and blocks.
+        await WaitForAsync(() => Task.FromResult(reader.Produced >= capacity), TimeSpan.FromSeconds(5));
+        await Task.Delay(200, cts.Token);
+        // capacity in the channel + at most one in the pump's hand, blocked on WriteAsync.
+        Assert.True(reader.Produced <= capacity + 1,
+            $"the reader pump kept draining an already-full inbox: {reader.Produced} messages read for a {capacity}-message bound");
+
+        // Draining releases it, and every message arrives exactly once in stream order — TryWrite
+        // on a full bounded channel would have dropped them silently instead.
+        for (long expected = 1; expected <= capacity + 1; expected++)
+        {
+            var msg = await ReadInboxAsync(duplex, cts.Token);
+            Assert.Equal(expected, msg.DeltaAck.AppliedThruSeq);
+        }
+        await WaitForAsync(() => Task.FromResult(reader.Produced > capacity + 1), TimeSpan.FromSeconds(5));
+
+        cts.Cancel();
+        try { await readerTask.WaitAsync(TimeSpan.FromSeconds(5)); }
+        catch (OperationCanceledException) { }
+    }
+
+    private static async Task<SyncMessage> ReadInboxAsync(
+        Replication.Internal.SyncDuplex duplex, CancellationToken ct)
+    {
+        await duplex.Inbox.WaitToReadAsync(ct);
+        duplex.Inbox.TryRead(out var message);
+        return message!;
+    }
+
+    // Produces messages as fast as the inbox will take them, numbering each one so the consumer can
+    // prove nothing was dropped or reordered.
+    private sealed class CountingReader : IAsyncStreamReader<SyncMessage>
+    {
+        private long _produced;
+
+        public long Produced => Interlocked.Read(ref _produced);
+
+        public SyncMessage Current { get; private set; } = new();
+
+        public Task<bool> MoveNext(CancellationToken cancellationToken)
+        {
+            cancellationToken.ThrowIfCancellationRequested();
+            Current = new SyncMessage
+            {
+                DeltaAck = new DeltaAck { AppliedThruSeq = Interlocked.Increment(ref _produced) },
+            };
+            return Task.FromResult(true);
+        }
+    }
+
     // ---- validator -------------------------------------------------------------------------
 
     [Fact]
diff --git a/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/RegistrationTests.cs b/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/RegistrationTests.cs
index f09c3c7..1c6005a 100644
--- a/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/RegistrationTests.cs
+++ b/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/RegistrationTests.cs
@@ -420,6 +420,63 @@ public sealed class RegistrationTests : IDisposable
         Assert.False(await NeedsSnapshot(db));
     }
 
+    // ---- Deregistration and the HLC crash anchor ------------------------------------------
+
+    [Fact]
+    public async Task Deregister_FlushesHlcAnchor_SoAnUngracefulReopenCannotRegress()
+    {
+        // The clock recovers as MAX(hlc_high_water, oplog HLCs, ledger HLCs, last_seen_hlc), and
+        // deregistering the only registered table empties both ledgers. Without the in-transaction
+        // flush, an ungraceful exit right afterwards leaves hlc_high_water at 0 and the reopened
+        // clock restarts from wall-clock time — below stamps the peer already holds.
+        var crashed = Create();
+        try
+        {
+            await CreateOrders(crashed);
+            crashed.RegisterReplicated("orders");
+            await crashed.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1,'A',1)");
+
+            // A peer HLC an hour ahead, merged as the applier would merge it: this is what makes the
+            // clock outrun wall-clock time, and therefore what a wall-clock restart would regress.
+            var peerHlc = DateTimeOffset.UtcNow.AddHours(1).ToUnixTimeMilliseconds() << 16;
+            crashed.Clock.Observe(peerHlc);
+            var preDereg = crashed.Clock.Current;
+            Assert.True(preDereg >= peerHlc);
+
+            crashed.DeregisterReplicated("orders");
+
+            Assert.Equal(preDereg, await CountAsync(crashed, "SELECT hlc_high_water FROM __localdb_meta WHERE id = 1"));
+
+            // The evidence the clock would otherwise have recovered from is gone.
+            Assert.Equal(0, await CountAsync(crashed, "SELECT COUNT(*) FROM __localdb_oplog"));
+            Assert.Equal(0, await CountAsync(crashed, "SELECT COUNT(*) FROM __localdb_row_version"));
+            Assert.Equal(0, await CountAsync(crashed, "SELECT last_seen_hlc FROM __localdb_peer_state WHERE id = 1"));
+
+            // Ungraceful reopen: a second instance over the same file while the first was NEVER
+            // disposed, so nothing but the deregistration itself can have written the anchor.
+            using var reopened = Create();
+            Assert.True(reopened.Clock.Current >= preDereg,
+                $"recovered clock {reopened.Clock.Current} regressed below the pre-deregistration stamp {preDereg}");
+            Assert.True(reopened.Clock.Next() > preDereg);
+        }
+        finally
+        {
+            crashed.Dispose();
+        }
+    }
+
+    [Fact]
+    public async Task Deregister_HlcFlush_DoesNotCountAsWorkDone()
+    {
+        // The flush is an UPDATE on every call, including the ones that clean up nothing; it must
+        // not turn the "was there anything to do?" answer into a permanent true.
+        using var db = Create();
+        await CreateOrders(db);
+
+        Assert.False(db.DeregisterReplicated("orders"));
+        Assert.False(db.DeregisterReplicated("ghost"));
+    }
+
     private static async Task<long> CountAsync(SqliteLocalDb db, string sql) =>
         (await db.QueryAsync(sql, r => r.GetInt64(0)))[0];
 
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 dbaa0d7..ef4e93a 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
@@ -67,10 +67,15 @@ public sealed class SnapshotResyncTests : IDisposable
         return new Side(db, store, applier, session);
     }
 
-    private static (SyncDuplex A, SyncDuplex B) DuplexPair()
+    // capacity models a transport that applies backpressure (which the gRPC duplex now is, via its
+    // bounded inbox); null is the historical unbounded pipe.
+    private static (SyncDuplex A, SyncDuplex B) DuplexPair(int? capacity = null)
     {
-        var aToB = Channel.CreateUnbounded<SyncMessage>();
-        var bToA = Channel.CreateUnbounded<SyncMessage>();
+        Channel<SyncMessage> New() => capacity is int c
+            ? Channel.CreateBounded<SyncMessage>(c)
+            : Channel.CreateUnbounded<SyncMessage>();
+        var aToB = New();
+        var bToA = New();
         var a = new SyncDuplex { Send = (m, ct) => aToB.Writer.WriteAsync(m, ct).AsTask(), Inbox = bToA.Reader };
         var b = new SyncDuplex { Send = (m, ct) => bToA.Writer.WriteAsync(m, ct).AsTask(), Inbox = aToB.Reader };
         return (a, b);
@@ -170,6 +175,50 @@ public sealed class SnapshotResyncTests : IDisposable
             IsTombstone = false,
         };
 
+    [Fact]
+    public async Task MutualSnapshot_OverABoundedTransport_ConvergesWithoutDeadlock()
+    {
+        // BOTH sides owe a snapshot — the ordinary outcome of enabling replication late on a pair,
+        // since baselining flags needs_snapshot on whichever node baselined and both nodes do.
+        //
+        // The snapshot used to be streamed from INSIDE the handshake, before the receive loop
+        // existed, so neither side drained its inbox while sending. That exchange completed only
+        // because the inbox was unbounded and could hold the peer's ENTIRE snapshot in memory —
+        // which is the memory profile this design gave up on. Over a transport that pushes back
+        // (which the gRPC duplex now is, with its bounded inbox) the old shape deadlocks outright.
+        // Streaming the snapshot alongside an already-running receive loop is what makes both the
+        // bound and this test possible.
+        var options = new ReplicationOptions { MaxBatchSize = 2, FlushInterval = TimeSpan.FromMilliseconds(20) };
+        var a = await NewSideAsync(options);
+        var b = await NewSideAsync(options);
+
+        const int rowsPerSide = 60;
+        for (var i = 1; i <= rowsPerSide; i++)
+        {
+            await a.Db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (@id, 'A', @id)", new { id = i });
+            await b.Db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (@id, 'B', @id)", new { id = 100 + i });
+        }
+        await a.Store.SetNeedsSnapshotAsync(true, default);
+        await b.Store.SetNeedsSnapshotAsync(true, default);
+
+        // Capacity 1 each way: the tightest possible bound, so a side that stops draining its inbox
+        // to send stalls at once instead of getting away with it on buffer slack.
+        var (da, db) = DuplexPair(capacity: 1);
+        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(a.Db)).Count == 2 * rowsPerSide
+                && (await ReadOrders(b.Db)).Count == 2 * rowsPerSide,
+            RunTimeout);
+        Assert.Equal(await ReadOrders(a.Db), await ReadOrders(b.Db));
+
+        cts.Cancel();
+        await SwallowAsync(runA);
+        await SwallowAsync(runB);
+    }
+
     [Fact]
     public async Task WipedPeer_AfterEverythingWasAckedAndPruned_StillGetsSnapshot()
     {