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()
     {