Merge branch 'localdb-perf-remediation'

The umbrella LocalDb row and the pack/test quick-reference now name 0.2.1 (188 tests,
3 nupkgs on the feed). Records both findings, the snapshot-out-of-handshake change that
bounding the inbox required, and the deferred per-table needs_snapshot. Consumer pins are
unchanged: ScadaBridge still pins 0.2.0, OtOpcUa still pins 0.1.3.

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

@@ -5,7 +5,7 @@
     <Nullable>enable</Nullable>
     <ImplicitUsings>enable</ImplicitUsings>
     <LangVersion>latest</LangVersion>
-    <Version>0.1.3</Version>
+    <Version>0.2.1</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,16 +10,59 @@ 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.3`)
+## Packages (all `0.2.1`)
 
 | Package | Contents |
 |---|---|
-| `ZB.MOM.WW.LocalDb` | Embedded DB host: `ILocalDb` SQL surface (`ExecuteAsync` / `QueryAsync` / transactions), `RegisterReplicated(table)` (PK validation + trigger install + oplog schema), `HybridLogicalClock`, `AddZbLocalDb()` DI. |
+| `ZB.MOM.WW.LocalDb` | Embedded DB host: `ILocalDb` SQL surface (`ExecuteAsync` / `QueryAsync` / transactions), `RegisterReplicated(table[, baselineExistingRows])` (PK validation + trigger install + oplog schema), `DeregisterReplicated(table)` (trigger removal + oplog/ledger prune), `HybridLogicalClock`, `AddZbLocalDb()` DI. |
 | `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: **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.
+>   - **`DeregisterReplicated(table)`** — drops a table's capture triggers and prunes its oplog and
+>     row-version rows. A node that does not replicate should not be paying for capture on every
+>     write, and until now there was no way to stop once a file had been registered. Idempotent, and
+>     usable on triggers left behind by an earlier build.
+>   - **`RegisterReplicated(table, baselineExistingRows: true)`** — closes the late-opt-in hole: rows
+>     that predate registration were invisible to the peer forever, because the snapshot streamer
+>     pages from the row-version ledger and CDC never wrote them there. Opting in seeds the ledger at
+>     an LWW-safe floor (HLC `0`, this node's id) and flags a snapshot resync so they actually
+>     travel. Default is still off — the documented CDC contract is unchanged unless you ask.
+>   - **O(1) backlog depth** — `MAX(seq) - MAX(last_acked_seq, MIN(seq) - 1)` replaces a
+>     `COUNT(*)` over the whole unacked backlog that ran once a minute AND on every metrics scrape
+>     (measured 58 VM steps vs 100,020 on a 50k backlog).
+>   - **Byte-budget batching** — `MaxBatchBytes` (default 2 MB) bounds delta and snapshot messages by
+>     summed serialized size, with `MaxBatchSize` demoted to a secondary row cap. Consumers no longer
+>     have to tune the row count against their widest row to stay under the gRPC message limit.
+>     Inbound delta batches and acks already queued are coalesced into one transaction / one ack.
+>   - **`LwwApplier` prepares once per statement** instead of three SQL parses per applied row.
+>   - **Dropped `__localdb_oplog_hlc`** — an index no query ever used, costing a b-tree insert inside
+>     every captured consumer write. Existing files lose it on the next open.
 > - `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
@@ -107,7 +150,8 @@ node still pushes its own deltas back. Only one side needs `PeerAddress`.
 | `PeerAddress` | `""` | gRPC address to dial. Empty ⇒ passive (initiator idles). |
 | `ApiKey` | `null` | Bearer token the initiator presents on its channel; `null` ⇒ no key header. |
 | `FlushInterval` | `250 ms` | How often the change pump flushes pending oplog deltas. |
-| `MaxBatchSize` | `500` | Max oplog entries per outbound batch (also the snapshot page size). |
+| `MaxBatchBytes` | `2 MB` | **Primary** batch cap: summed serialized bytes per outbound delta/snapshot message. Sized under the 4 MB gRPC default; a single row above the budget is still sent, alone. |
+| `MaxBatchSize` | `500` | **Secondary** row-count cap per outbound batch (also the snapshot DB page size). |
 | `MaxOplogRows` | `1_000_000` | Backlog ceiling; exceeding it flags a snapshot resync and prunes oldest rows. |
 | `MaxOplogAge` | `7 days` | Age ceiling; an unpruned entry older than this flags a snapshot resync. |
 | `TombstoneRetention` | `7 days` | How long deleted-row tombstones are kept before pruning. |
@@ -134,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`.
 
@@ -161,13 +207,27 @@ All options are validated at startup (`ZB.MOM.WW.Configuration` validators).
   still hold a BLOB value at runtime, which then surfaces as a write-time error, not a
   registration error.
 - **CDC semantics** — rows that existed **before** `RegisterReplicated` are not captured
-  and are not snapshotted; only changes after registration are replicated. Re-baselining
-  pre-existing rows requires re-writing them.
+  and are not snapshotted; only changes after registration are replicated. **Opt out with
+  `RegisterReplicated(table, baselineExistingRows: true)`**, which seeds the row-version
+  ledger for existing rows at the baseline floor (HLC `0`) and flags a snapshot resync so
+  they reach the peer. The floor is below every genuine stamp, so a baselined row loses
+  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
 
@@ -198,6 +258,19 @@ provider is wired or the poll threw, so a DB failure never reads as "0 backlog")
 
 - When the oplog row/age caps are exceeded, maintenance prunes to the ceiling and flags
   `needs_snapshot` (disk safety over delta continuity); the peer then snapshot-resyncs.
+- **Turning a table off** — `DeregisterReplicated(table)` drops its capture triggers and prunes
+  its oplog + ledger rows. It must be applied **symmetrically**: the handshake compares
+  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 — 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
+  keeps syncing.
 - The **`__localdb_dead_letter`** table holds poison rows (bad JSON, null-row non-tombstone
   entries) and over-drift rejects (when `FailClosedOnDrift`). Inspect it to diagnose
   rejects.
@@ -208,13 +281,15 @@ provider is wired or the poll threw, so a DB failure never reads as "0 backlog")
 
 ## Testing
 
-144 tests, all offline / no live dependencies — unit (HLC ordering + restart
-persistence, trigger SQL generation, LWW decision table, watermark/pruning, digest
-computation), 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). Family conventions apply
-(`TreatWarningsAsErrors`).
+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:
 

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

ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/LwwApplier.cs

@@ -16,10 +16,24 @@ internal sealed record ApplyResult(long AppliedThruSeq, int Applied, int Discard
 /// crash-replay of the batch is a no-op (LWW's strict-greater compare loses on equal versions).
 /// This never writes <c>__localdb_oplog</c>: the 2-node topology does not re-forward applied deltas.
 /// </summary>
+/// <remarks>
+/// Every statement is issued through a <see cref="PreparedStatements"/> set held for the life of the
+/// batch, so each distinct statement is prepared ONCE and then rebound per row. The previous shape
+/// went through the generic <c>ILocalDbTransaction</c> façade, which builds a fresh
+/// <see cref="SqliteCommand"/> per call — three SQL parses per applied row (version lookup, the
+/// table's delete/upsert, the row-version write), on the path that drains a reconnect backlog.
+/// </remarks>
 internal sealed class LwwApplier(SqliteLocalDb db, Func<DateTimeOffset>? utcNow = null)
 {
     private readonly Func<DateTimeOffset> _utcNow = utcNow ?? (() => DateTimeOffset.UtcNow);
 
+    /// <summary>
+    /// Distinct statements prepared by the most recent <see cref="ApplyBatchAsync"/>. Test-only
+    /// observation: it is the direct evidence that statement count tracks the number of distinct
+    /// STATEMENTS rather than the number of rows.
+    /// </summary>
+    internal int LastBatchStatementCount { get; private set; }
+
     public async Task<ApplyResult> ApplyBatchAsync(IReadOnlyList<OplogEntryRecord> entries, CancellationToken ct)
     {
         if (entries.Count == 0)
@@ -42,13 +56,14 @@ internal sealed class LwwApplier(SqliteLocalDb db, Func<DateTimeOffset>? utcNow
             if (entry.Hlc > maxHlc) maxHlc = entry.Hlc;
         }
 
-        var sqlByTable = new Dictionary<string, (string Delete, string Upsert)>(StringComparer.Ordinal);
         var applied = 0;
         var discarded = 0;
         var deadLettered = 0;
 
-        await using var tx = await db.BeginTransactionAsync(ct);
-        await tx.ExecuteAsync("UPDATE __localdb_applying SET applying = 1 WHERE id = 1", null, ct);
+        await using var tx = await db.BeginPreparedTransactionAsync(ct);
+        using var sql = new PreparedStatements(tx.Connection, tx.Sqlite);
+
+        await sql.SetApplyingAsync(1, ct);
 
         foreach (var entry in entries)
         {
@@ -56,19 +71,16 @@ internal sealed class LwwApplier(SqliteLocalDb db, Func<DateTimeOffset>? utcNow
             // KEY table that silently inserts a phantom auto-rowid row — so it is poison, not a win.
             if (!entry.IsTombstone && entry.RowJson is null)
             {
-                await DeadLetterAsync(tx, entry, "non-tombstone entry with null row_json", utcNow, ct);
+                await sql.DeadLetterAsync(entry, "non-tombstone entry with null row_json", utcNow, ct);
                 deadLettered++;
                 continue;
             }
 
-            var local = await tx.QueryAsync(
-                "SELECT hlc, node_id FROM __localdb_row_version WHERE table_name = @t AND pk_json = @pk",
-                static r => (Hlc: r.GetInt64(0), NodeId: r.GetString(1)),
-                new { t = entry.TableName, pk = entry.PkJson }, ct);
+            var local = await sql.ReadLocalVersionAsync(entry.TableName, entry.PkJson, ct);
 
-            var remoteWins = local.Count == 0
-                || entry.Hlc > local[0].Hlc
-                || (entry.Hlc == local[0].Hlc && string.CompareOrdinal(entry.NodeId, local[0].NodeId) > 0);
+            var remoteWins = local is null
+                || entry.Hlc > local.Value.Hlc
+                || (entry.Hlc == local.Value.Hlc && string.CompareOrdinal(entry.NodeId, local.Value.NodeId) > 0);
 
             if (!remoteWins)
             {
@@ -76,32 +88,15 @@ internal sealed class LwwApplier(SqliteLocalDb db, Func<DateTimeOffset>? utcNow
                 continue;
             }
 
-            if (!sqlByTable.TryGetValue(entry.TableName, out var sql))
-            {
-                sql = BuildTableSql(db.ReplicatedTables[entry.TableName]);
-                sqlByTable[entry.TableName] = sql;
-            }
-
             try
             {
+                var table = db.ReplicatedTables[entry.TableName];
                 if (entry.IsTombstone)
-                    await tx.ExecuteAsync(sql.Delete, new { pk = entry.PkJson }, ct);
+                    await sql.DeleteAsync(table, entry.PkJson, ct);
                 else
-                    await tx.ExecuteAsync(sql.Upsert, new { row = entry.RowJson }, ct);
+                    await sql.UpsertAsync(table, entry.RowJson!, ct);
 
-                await tx.ExecuteAsync(
-                    "INSERT OR REPLACE INTO __localdb_row_version " +
-                    "(table_name, pk_json, hlc, node_id, is_tombstone, tombstone_utc) " +
-                    "VALUES (@t, @pk, @hlc, @node, @tomb, @tut)",
-                    new
-                    {
-                        t = entry.TableName,
-                        pk = entry.PkJson,
-                        hlc = entry.Hlc,
-                        node = entry.NodeId,
-                        tomb = entry.IsTombstone ? 1 : 0,
-                        tut = entry.IsTombstone ? utcNow : null,
-                    }, ct);
+                await sql.WriteRowVersionAsync(entry, utcNow, ct);
                 applied++;
             }
             catch (SqliteException ex)
@@ -110,59 +105,180 @@ internal sealed class LwwApplier(SqliteLocalDb db, Func<DateTimeOffset>? utcNow
                 // malformed row_json) are treated as per-entry poison — SQLite aborts just the
                 // statement, not the transaction, so the batch continues after dead-lettering.
                 // Any other exception propagates and rolls back the whole batch (fail-closed).
-                await DeadLetterAsync(tx, entry, ex.Message, utcNow, ct);
+                await sql.DeadLetterAsync(entry, ex.Message, utcNow, ct);
                 deadLettered++;
             }
         }
 
-        await tx.ExecuteAsync(
-            "UPDATE __localdb_peer_state SET " +
-            "last_applied_remote_seq = MAX(last_applied_remote_seq, @maxSeq), " +
-            "last_seen_hlc = MAX(last_seen_hlc, @maxHlc), last_sync_utc = @utc WHERE id = 1",
-            new { maxSeq, maxHlc, utc = utcNow }, ct);
+        await sql.AdvancePeerStateAsync(maxSeq, maxHlc, utcNow, ct);
+        await sql.SetApplyingAsync(0, ct);
+
+        LastBatchStatementCount = sql.StatementCount;
 
-        await tx.ExecuteAsync("UPDATE __localdb_applying SET applying = 0 WHERE id = 1", null, ct);
         await tx.CommitAsync(ct);
 
         db.Clock.Observe(maxHlc);
         return new ApplyResult(maxSeq, applied, discarded, deadLettered);
     }
 
-    private static Task DeadLetterAsync(
-        ILocalDbTransaction tx, OplogEntryRecord entry, string error, string utcNow, CancellationToken ct) =>
-        tx.ExecuteAsync(
-            "INSERT INTO __localdb_dead_letter " +
-            "(received_utc, table_name, pk_json, row_json, hlc, node_id, error) " +
-            "VALUES (@utc, @t, @pk, @row, @hlc, @node, @err)",
-            new
-            {
-                utc = utcNow,
-                t = entry.TableName,
-                pk = entry.PkJson,
-                row = entry.RowJson,
-                hlc = entry.Hlc,
-                node = entry.NodeId,
-                err = error,
-            }, ct);
-
-    private static (string Delete, string Upsert) BuildTableSql(ReplicatedTable table)
+    /// <summary>
+    /// The statement set one apply-batch runs on: each distinct statement is built and prepared on
+    /// first use, then rebound and re-executed per row. Statements whose SQL depends on the table
+    /// (delete / upsert) are cached per table; the rest are shared across all of them.
+    /// </summary>
+    /// <remarks>
+    /// Reusing the same <see cref="SqliteCommand"/> with unchanged <c>CommandText</c> is what makes
+    /// this prepare-once: Microsoft.Data.Sqlite holds the compiled <c>sqlite3_stmt</c> on the command
+    /// and only recompiles when the text changes. <see cref="SqliteCommand.Prepare"/> is called
+    /// explicitly anyway, so the intent is stated rather than inferred from an implementation detail.
+    /// </remarks>
+    private sealed class PreparedStatements(SqliteConnection connection, SqliteTransaction transaction) : IDisposable
     {
-        var name = Ident(table.Name);
+        private readonly List<SqliteCommand> _owned = [];
+        private readonly Dictionary<string, SqliteCommand> _deletes = new(StringComparer.Ordinal);
+        private readonly Dictionary<string, SqliteCommand> _upserts = new(StringComparer.Ordinal);
 
-        var pkMatch = string.Join(" AND ", table.PkColumns.Select(p => $"{Ident(p)} = json_extract(@pk, {JsonPath(p)})"));
-        var delete = $"DELETE FROM {name} WHERE {pkMatch}";
+        private SqliteCommand? _applying;
+        private SqliteCommand? _lookup;
+        private SqliteCommand? _rowVersion;
+        private SqliteCommand? _deadLetter;
+        private SqliteCommand? _peerState;
 
-        var columns = string.Join(", ", table.Columns.Select(c => Ident(c.Name)));
-        var values = string.Join(", ", table.Columns.Select(c => $"json_extract(@row, {JsonPath(c.Name)})"));
-        var upsert = $"INSERT OR REPLACE INTO {name} ({columns}) VALUES ({values})";
+        /// <summary>How many distinct statements this set has prepared.</summary>
+        public int StatementCount => _owned.Count;
 
-        return (delete, upsert);
+        public Task SetApplyingAsync(int value, CancellationToken ct)
+        {
+            var cmd = _applying ??= Create(
+                "UPDATE __localdb_applying SET applying = @applying WHERE id = 1", "@applying");
+            cmd.Parameters["@applying"].Value = value;
+            return cmd.ExecuteNonQueryAsync(ct);
+        }
+
+        public async Task<(long Hlc, string NodeId)?> ReadLocalVersionAsync(
+            string tableName, string pkJson, CancellationToken ct)
+        {
+            var cmd = _lookup ??= Create(
+                "SELECT hlc, node_id FROM __localdb_row_version WHERE table_name = @t AND pk_json = @pk",
+                "@t", "@pk");
+            cmd.Parameters["@t"].Value = tableName;
+            cmd.Parameters["@pk"].Value = pkJson;
+
+            await using var reader = await cmd.ExecuteReaderAsync(ct);
+            if (!await reader.ReadAsync(ct))
+                return null;
+            return (reader.GetInt64(0), reader.GetString(1));
+        }
+
+        public Task DeleteAsync(ReplicatedTable table, string pkJson, CancellationToken ct)
+        {
+            if (!_deletes.TryGetValue(table.Name, out var cmd))
+                _deletes[table.Name] = cmd = Create(BuildDelete(table), "@pk");
+            cmd.Parameters["@pk"].Value = pkJson;
+            return cmd.ExecuteNonQueryAsync(ct);
+        }
+
+        public Task UpsertAsync(ReplicatedTable table, string rowJson, CancellationToken ct)
+        {
+            if (!_upserts.TryGetValue(table.Name, out var cmd))
+                _upserts[table.Name] = cmd = Create(BuildUpsert(table), "@row");
+            cmd.Parameters["@row"].Value = rowJson;
+            return cmd.ExecuteNonQueryAsync(ct);
+        }
+
+        public Task WriteRowVersionAsync(OplogEntryRecord entry, string utcNow, CancellationToken ct)
+        {
+            var cmd = _rowVersion ??= Create(
+                "INSERT OR REPLACE INTO __localdb_row_version " +
+                "(table_name, pk_json, hlc, node_id, is_tombstone, tombstone_utc) " +
+                "VALUES (@t, @pk, @hlc, @node, @tomb, @tut)",
+                "@t", "@pk", "@hlc", "@node", "@tomb", "@tut");
+            cmd.Parameters["@t"].Value = entry.TableName;
+            cmd.Parameters["@pk"].Value = entry.PkJson;
+            cmd.Parameters["@hlc"].Value = entry.Hlc;
+            cmd.Parameters["@node"].Value = entry.NodeId;
+            cmd.Parameters["@tomb"].Value = entry.IsTombstone ? 1 : 0;
+            cmd.Parameters["@tut"].Value = entry.IsTombstone ? utcNow : DBNull.Value;
+            return cmd.ExecuteNonQueryAsync(ct);
+        }
+
+        public Task DeadLetterAsync(OplogEntryRecord entry, string error, string utcNow, CancellationToken ct)
+        {
+            var cmd = _deadLetter ??= Create(
+                "INSERT INTO __localdb_dead_letter " +
+                "(received_utc, table_name, pk_json, row_json, hlc, node_id, error) " +
+                "VALUES (@utc, @t, @pk, @row, @hlc, @node, @err)",
+                "@utc", "@t", "@pk", "@row", "@hlc", "@node", "@err");
+            cmd.Parameters["@utc"].Value = utcNow;
+            cmd.Parameters["@t"].Value = entry.TableName;
+            cmd.Parameters["@pk"].Value = entry.PkJson;
+            cmd.Parameters["@row"].Value = (object?)entry.RowJson ?? DBNull.Value;
+            cmd.Parameters["@hlc"].Value = entry.Hlc;
+            cmd.Parameters["@node"].Value = entry.NodeId;
+            cmd.Parameters["@err"].Value = error;
+            return cmd.ExecuteNonQueryAsync(ct);
+        }
+
+        public Task AdvancePeerStateAsync(long maxSeq, long maxHlc, string utcNow, CancellationToken ct)
+        {
+            var cmd = _peerState ??= Create(
+                "UPDATE __localdb_peer_state SET " +
+                "last_applied_remote_seq = MAX(last_applied_remote_seq, @maxSeq), " +
+                "last_seen_hlc = MAX(last_seen_hlc, @maxHlc), last_sync_utc = @utc WHERE id = 1",
+                "@maxSeq", "@maxHlc", "@utc");
+            cmd.Parameters["@maxSeq"].Value = maxSeq;
+            cmd.Parameters["@maxHlc"].Value = maxHlc;
+            cmd.Parameters["@utc"].Value = utcNow;
+            return cmd.ExecuteNonQueryAsync(ct);
+        }
+
+        private SqliteCommand Create(string sql, params string[] parameterNames)
+        {
+            var cmd = connection.CreateCommand();
+            try
+            {
+                cmd.Transaction = transaction;
+                cmd.CommandText = sql;
+                foreach (var name in parameterNames)
+                    cmd.Parameters.Add(new SqliteParameter(name, DBNull.Value));
+                cmd.Prepare();
+                _owned.Add(cmd);
+                return cmd;
+            }
+            catch
+            {
+                cmd.Dispose();
+                throw;
+            }
+        }
+
+        public void Dispose()
+        {
+            foreach (var cmd in _owned)
+                cmd.Dispose();
+            _owned.Clear();
+        }
+
+        private static string BuildDelete(ReplicatedTable table)
+        {
+            var pkMatch = string.Join(
+                " AND ", table.PkColumns.Select(p => $"{Ident(p)} = json_extract(@pk, {JsonPath(p)})"));
+            return $"DELETE FROM {Ident(table.Name)} WHERE {pkMatch}";
+        }
+
+        private static string BuildUpsert(ReplicatedTable table)
+        {
+            var columns = string.Join(", ", table.Columns.Select(c => Ident(c.Name)));
+            var values = string.Join(", ", table.Columns.Select(c => $"json_extract(@row, {JsonPath(c.Name)})"));
+            return $"INSERT OR REPLACE INTO {Ident(table.Name)} ({columns}) VALUES ({values})";
+        }
+
+        // Local copy of TriggerSqlGenerator's convention (that helper lives in the core assembly).
+        private static string Ident(string name) => "\"" + name.Replace("\"", "\"\"") + "\"";
+
+        // A single-quoted JSON path with a quoted key member, so a column name with reserved
+        // characters still resolves: $."col". The key's double quotes are backslash-escaped per
+        // JSON path grammar.
+        private static string JsonPath(string column) => "'$.\"" + column.Replace("\"", "\\\"") + "\"'";
     }
-
-    // Local copy of TriggerSqlGenerator's convention (that helper lives in the core assembly).
-    private static string Ident(string name) => "\"" + name.Replace("\"", "\"\"") + "\"";
-
-    // A single-quoted JSON path with a quoted key member, so a column name with reserved characters
-    // still resolves: $."col". The key's double quotes are backslash-escaped per JSON path grammar.
-    private static string JsonPath(string column) => "'$.\"" + column.Replace("\"", "\\\"") + "\"'";
 }

ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/OplogStore.cs

@@ -70,12 +70,42 @@ internal sealed class OplogStore(ILocalDb db, ReplicationOptions options, Func<D
             "DELETE FROM __localdb_row_version WHERE is_tombstone = 1 AND tombstone_utc < @cutoff",
             new { cutoff = FormatUtc(_utcNow() - options.TombstoneRetention) }, ct);
 
-    /// <summary>Counts the unacked backlog: oplog entries above the peer's last-acked watermark.</summary>
+    /// <summary>
+    /// The unacked backlog — oplog entries above the peer's last-acked watermark — in O(1).
+    /// </summary>
+    /// <remarks>
+    /// <para>
+    /// <b>Why not <c>COUNT(*)</c>.</b> The obvious query walks every unacked row, and this value is
+    /// read once a minute by maintenance AND on every metrics scrape. A node whose peer has been
+    /// down for a day therefore pays a full scan of a million-row backlog, repeatedly, precisely
+    /// when it is already under stress.
+    /// </para>
+    /// <para>
+    /// <b>The formula.</b> <c>MAX(seq) - MAX(last_acked_seq, MIN(seq) - 1)</c>, floored at 0. Both
+    /// MIN and MAX over <c>seq</c> are single b-tree seeks (it is the rowid, and SQLite's min/max
+    /// optimization turns each into one seek), and the peer-state read is a one-row lookup — so the
+    /// whole thing is three seeks regardless of backlog size.
+    /// </para>
+    /// <para>
+    /// <b>Why it is exact.</b> <c>seq</c> is AUTOINCREMENT, so retained rows are a contiguous run
+    /// wherever nothing has punched a hole in the middle — and no ordinary path punches holes: ack
+    /// pruning deletes <c>seq &lt;= last_acked_seq</c> (a prefix), the row cap deletes below
+    /// <c>MAX(seq) - keep + 1</c> (a prefix), and the age cap deletes below an HLC cutoff, which is
+    /// a prefix because HLCs are stamped in seq order. Clamping the floor to <c>MIN(seq) - 1</c> is
+    /// what keeps the answer exact after a prune has moved the horizon above the ack watermark;
+    /// without it a cap-pruned oplog would report its whole seq range forever and re-flag the cap on
+    /// every tick.
+    /// </para>
+    /// <para>
+    /// <b>The one inexact case</b> is <c>ILocalDb.DeregisterReplicated</c>, which deletes one
+    /// table's rows from wherever they sit and so can leave holes. The estimate then over-reports by
+    /// at most the number of holes, until the watermark advances past them. Over-reporting is the
+    /// safe direction: it can only trigger a snapshot resync earlier, never suppress one.
+    /// </para>
+    /// </remarks>
     public async Task<long> GetOplogDepthAsync(CancellationToken ct = default)
     {
-        var rows = await db.QueryAsync(
-            "SELECT COUNT(*) FROM __localdb_oplog WHERE seq > (SELECT last_acked_seq FROM __localdb_peer_state WHERE id = 1)",
-            static r => r.GetInt64(0), null, ct);
+        var rows = await db.QueryAsync(DepthSql, static r => r.GetInt64(0), null, ct);
         return rows[0];
     }
 
@@ -84,7 +114,7 @@ internal sealed class OplogStore(ILocalDb db, ReplicationOptions options, Func<D
     /// metric-observation callbacks (gauge scrape / status poll); mirrors
     /// <see cref="GetOplogDepthAsync"/>'s query. Deliberate tradeoff: synchronous I/O on the
     /// metrics-collection thread is acceptable here because WAL-mode readers never block (or are
-    /// blocked by) writers, the COUNT is indexed-cheap, and the worst case is bounded by
+    /// blocked by) writers, the query is three b-tree seeks, and the worst case is bounded by
     /// <c>BusyTimeoutMs</c> — all at scrape frequency, not on any data path. May throw if the db
     /// is disposing — callers on the observation path swallow and drop the sample.
     /// </summary>
@@ -92,11 +122,18 @@ internal sealed class OplogStore(ILocalDb db, ReplicationOptions options, Func<D
     {
         using var conn = db.CreateConnection();
         using var cmd = conn.CreateCommand();
-        cmd.CommandText =
-            "SELECT COUNT(*) FROM __localdb_oplog WHERE seq > (SELECT last_acked_seq FROM __localdb_peer_state WHERE id = 1)";
+        cmd.CommandText = DepthSql;
         return (long)cmd.ExecuteScalar()!;
     }
 
+    // Both depth accessors run exactly this; keeping it in one place is what makes "mirrors the
+    // async query" a fact rather than a comment that can rot.
+    internal const string DepthSql =
+        "SELECT MAX(0, " +
+        "COALESCE((SELECT MAX(seq) FROM __localdb_oplog), 0) - " +
+        "MAX((SELECT last_acked_seq FROM __localdb_peer_state WHERE id = 1), " +
+        "COALESCE((SELECT MIN(seq) FROM __localdb_oplog), 1) - 1))";
+
     /// <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
@@ -133,9 +170,9 @@ internal sealed class OplogStore(ILocalDb db, ReplicationOptions options, Func<D
         // Condition reads and prunes share the transaction so the decision cannot go stale (TOCTOU).
         await using var tx = await db.BeginTransactionAsync(ct);
 
-        var depth = (await tx.QueryAsync(
-            "SELECT COUNT(*) FROM __localdb_oplog WHERE seq > (SELECT last_acked_seq FROM __localdb_peer_state WHERE id = 1)",
-            static r => r.GetInt64(0), null, ct))[0];
+        // Same O(1) expression as GetOplogDepthAsync — the caps check is the per-minute reader that
+        // made the old COUNT(*) expensive in the first place.
+        var depth = (await tx.QueryAsync(DepthSql, static r => r.GetInt64(0), null, ct))[0];
         var overRows = depth > options.MaxOplogRows;
 
         var oldest = await tx.QueryAsync(

ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/SnapshotStreamer.cs

@@ -57,7 +57,13 @@ internal sealed class SnapshotStreamer(SqliteLocalDb db, OplogStore store, Repli
                     break;
                 lastPk = rows[^1].PkJson;
 
+                // One DB page may span several messages: the page is bounded by ROW COUNT, but a
+                // message is bounded by BYTES (ReplicationOptions.MaxBatchBytes) — the same budget
+                // the delta pump honours, and for the same reason. Wide rows are exactly why a
+                // snapshot is the easiest way to exceed the transport's message limit: it ships
+                // whole rows, all of them, back to back.
                 var batch = new SnapshotBatch { TableName = table.Name };
+                var bytes = 0L;
                 foreach (var row in rows)
                 {
                     if (!row.IsTombstone && row.RowJson is null)
@@ -79,11 +85,25 @@ internal sealed class SnapshotStreamer(SqliteLocalDb db, OplogStore store, Repli
                     };
                     if (row.RowJson is not null)
                         snapshotRow.RowJson = row.RowJson;
+
+                    var size = RowSize(snapshotRow);
+                    // Flush before adding, never after: a message keeps at least one row however
+                    // large, so an oversized row goes out alone rather than stalling the snapshot.
+                    if (batch.Rows.Count > 0 && bytes + size > options.MaxBatchBytes)
+                    {
+                        await send(new SyncMessage { SnapshotBatch = batch }, ct);
+                        totalRows += batch.Rows.Count;
+                        batch = new SnapshotBatch { TableName = table.Name };
+                        bytes = 0;
+                    }
                     batch.Rows.Add(snapshotRow);
+                    bytes += size;
                 }
                 if (batch.Rows.Count > 0)
+                {
                     await send(new SyncMessage { SnapshotBatch = batch }, ct);
-                totalRows += batch.Rows.Count;
+                    totalRows += batch.Rows.Count;
+                }
 
                 if (rows.Count < options.MaxBatchSize)
                     break;
@@ -104,11 +124,23 @@ internal sealed class SnapshotStreamer(SqliteLocalDb db, OplogStore store, Repli
         return asOfSeq;
     }
 
+    /// <summary>
+    /// Serialized bytes one row contributes to its batch message: the row's own encoding plus the
+    /// repeated field's tag and length prefix, as protobuf itself computes them.
+    /// </summary>
+    private static long RowSize(SnapshotRow row) =>
+        1 + Google.Protobuf.CodedOutputStream.ComputeMessageSize(row);
+
     // Pages one table's rows by joining the row-version ledger to the live table on the PK columns
     // extracted from pk_json. Rows present in the live table but absent from __localdb_row_version
-    // (data written BEFORE the table was registered for replication) are intentionally NOT snapshotted:
-    // they were never captured, consistent with CDC semantics — a full re-registration is required to
-    // baseline them.
+    // (data written BEFORE the table was registered for replication) are NOT snapshotted: they were
+    // never captured, consistent with CDC semantics.
+    //
+    // That is a real hole for a node that starts replicating LATE — its whole pre-existing dataset is
+    // invisible to the peer, silently and permanently — and the supported way to close it is to
+    // register with `ILocalDb.RegisterReplicated(table, baselineExistingRows: true)`, which seeds the
+    // ledger at an LWW-safe floor and flags the resync that carries the rows here. Nothing changes on
+    // this side: baselined rows are ordinary ledger rows and page like any other.
     private static string BuildPageSql(ReplicatedTable table)
     {
         var name = Ident(table.Name);

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
@@ -246,6 +276,17 @@ internal sealed class SyncSession
         }
     }
 
+    /// <summary>
+    /// Reads oplog deltas above the sent watermark and ships them, splitting each read into as many
+    /// messages as <see cref="ReplicationOptions.MaxBatchBytes"/> requires.
+    /// </summary>
+    /// <remarks>
+    /// Row count alone cannot bound a message: rows are arbitrarily wide, so any row-count cap is
+    /// really a guess about the widest row, and guessing wrong produces a message the transport
+    /// rejects — replication then stops converging with a size error, not a data error. Budgeting by
+    /// summed serialized bytes bounds the thing that is actually limited, and lets the row cap go
+    /// back to meaning what it says.
+    /// </remarks>
     private async Task PumpLoopAsync(ChannelWriter<SyncMessage> dataLane, CancellationToken ct)
     {
         while (true)
@@ -254,12 +295,30 @@ internal sealed class SyncSession
             var batch = await _store.ReadBatchAboveAsync(_sentThruSeq, _options.MaxBatchSize, ct);
             if (batch.Count > 0)
             {
-                var delta = new DeltaBatch();
-                foreach (var entry in batch)
-                    delta.Entries.Add(ToProto(entry));
-                // Blocking here on a full data lane IS the pump's backpressure.
-                await dataLane.WriteAsync(new SyncMessage { DeltaBatch = delta }, ct);
-                _sentThruSeq = batch[^1].Seq;
+                var index = 0;
+                while (index < batch.Count)
+                {
+                    var delta = new DeltaBatch();
+                    var bytes = 0L;
+                    while (index < batch.Count)
+                    {
+                        var proto = ToProto(batch[index]);
+                        var size = EntrySize(proto);
+                        // The first entry of a message is taken however large it is: refusing an
+                        // oversized row would stall the stream on it permanently, whereas sending it
+                        // alone either succeeds or fails loudly at the transport with its real size.
+                        if (delta.Entries.Count > 0 && bytes + size > _options.MaxBatchBytes)
+                            break;
+                        delta.Entries.Add(proto);
+                        bytes += size;
+                        index++;
+                    }
+                    // Blocking here on a full data lane IS the pump's backpressure.
+                    await dataLane.WriteAsync(new SyncMessage { DeltaBatch = delta }, ct);
+                    // Advanced per MESSAGE, not per read: a fault mid-split must not re-send what
+                    // already went out, and must not skip what did not.
+                    _sentThruSeq = batch[index - 1].Seq;
+                }
                 continue;
             }
 
@@ -267,23 +326,43 @@ internal sealed class SyncSession
         }
     }
 
+    /// <summary>
+    /// Serialized bytes one entry contributes to its batch message: the entry's own encoding plus
+    /// the repeated field's tag and length prefix, as protobuf itself computes them.
+    /// </summary>
+    private static long EntrySize(OplogEntry entry) =>
+        1 + Google.Protobuf.CodedOutputStream.ComputeMessageSize(entry);
+
     private async Task ReceiveLoopAsync(
         SyncDuplex duplex, ChannelWriter<SyncMessage> controlLane, CancellationToken ct)
     {
-        while (await duplex.Inbox.WaitToReadAsync(ct))
+        // One-message lookahead. Coalescing means reading ahead until a message turns out NOT to be
+        // of the kind being coalesced, and a Channel cannot un-read — so that message is parked here
+        // and processed first on the next pass. Its presence also keeps the outer loop from waiting
+        // on the inbox when there is already work in hand.
+        SyncMessage? lookahead = null;
+
+        while (lookahead is not null || await duplex.Inbox.WaitToReadAsync(ct))
         {
-            while (duplex.Inbox.TryRead(out var msg))
+            while (true)
             {
+                var msg = lookahead;
+                lookahead = null;
+                if (msg is null)
+                {
+                    if (!duplex.Inbox.TryRead(out var read))
+                        break;
+                    msg = read;
+                }
+
                 switch (msg.MsgCase)
                 {
                     case SyncMessage.MsgOneofCase.DeltaBatch:
                         // Deltas apply normally even mid-snapshot: LWW makes the interleave safe.
-                        await HandleDeltaBatchAsync(controlLane, msg.DeltaBatch, ct);
+                        lookahead = await HandleCoalescedDeltasAsync(duplex, controlLane, msg.DeltaBatch, ct);
                         break;
                     case SyncMessage.MsgOneofCase.DeltaAck:
-                        var acked = msg.DeltaAck.AppliedThruSeq;
-                        InterlockedMax(ref _peerAckedSeq, acked);
-                        await _store.RecordPeerAckAsync(acked, ct);
+                        lookahead = await HandleCoalescedAcksAsync(duplex, msg.DeltaAck, ct);
                         break;
                     case SyncMessage.MsgOneofCase.SnapshotBegin:
                         if (SnapshotApplier is null)
@@ -314,19 +393,98 @@ internal sealed class SyncSession
         }
     }
 
-    private async Task HandleDeltaBatchAsync(ChannelWriter<SyncMessage> controlLane, DeltaBatch batch, CancellationToken ct)
+    /// <summary>
+    /// Applies the given delta batch together with any delta batches ALREADY waiting in the inbox,
+    /// as one transaction answered by one ack; returns the non-delta message that ended the run (to
+    /// be processed next), or null if the inbox merely ran dry.
+    /// </summary>
+    /// <remarks>
+    /// <para>
+    /// Byte-budgeting the sender produces more, smaller messages, and one apply transaction plus one
+    /// ack round-trip per message is the cost that would otherwise scale with that. Coalescing pays
+    /// it once per burst instead — and a burst is exactly what a reconnect after an outage delivers.
+    /// </para>
+    /// <para>
+    /// Only messages already queued are absorbed (<c>TryRead</c>, never a wait), so this adds no
+    /// latency: a trickle of one batch at a time coalesces nothing and behaves exactly as before.
+    /// The run stops at the sender's own caps, so the extra memory held is bounded by roughly one
+    /// sender batch beyond the budget — the overshoot of finishing whichever batch crosses it.
+    /// </para>
+    /// </remarks>
+    private async Task<SyncMessage?> HandleCoalescedDeltasAsync(
+        SyncDuplex duplex, ChannelWriter<SyncMessage> controlLane, DeltaBatch first, CancellationToken ct)
     {
-        if (batch.Entries.Count == 0)
+        var entries = new List<OplogEntry>(first.Entries);
+        var bytes = 0L;
+        foreach (var entry in first.Entries)
+            bytes += EntrySize(entry);
+
+        SyncMessage? lookahead = null;
+        while (entries.Count < _options.MaxBatchSize && bytes < _options.MaxBatchBytes
+               && duplex.Inbox.TryRead(out var next))
+        {
+            if (next.MsgCase != SyncMessage.MsgOneofCase.DeltaBatch)
+            {
+                lookahead = next;
+                break;
+            }
+            foreach (var entry in next.DeltaBatch.Entries)
+            {
+                entries.Add(entry);
+                bytes += EntrySize(entry);
+            }
+        }
+
+        await ApplyDeltaEntriesAsync(controlLane, entries, ct);
+        return lookahead;
+    }
+
+    /// <summary>
+    /// Records the given ack together with any acks ALREADY waiting in the inbox, as a single
+    /// watermark write; returns the non-ack message that ended the run, or null if the inbox ran dry.
+    /// </summary>
+    /// <remarks>
+    /// Exactly equivalent to recording each in turn, and strictly cheaper: <c>applied_thru_seq</c> is
+    /// monotonic and <c>RecordPeerAckAsync</c> is MAX-guarded, so every superseded ack would have
+    /// written a watermark that the final one overwrites and run a DELETE below its own horizon.
+    /// Unbounded coalescing is safe here — acks collapse into one number, accumulating nothing.
+    /// </remarks>
+    private async Task<SyncMessage?> HandleCoalescedAcksAsync(
+        SyncDuplex duplex, DeltaAck first, CancellationToken ct)
+    {
+        var acked = first.AppliedThruSeq;
+
+        SyncMessage? lookahead = null;
+        while (duplex.Inbox.TryRead(out var next))
+        {
+            if (next.MsgCase != SyncMessage.MsgOneofCase.DeltaAck)
+            {
+                lookahead = next;
+                break;
+            }
+            if (next.DeltaAck.AppliedThruSeq > acked)
+                acked = next.DeltaAck.AppliedThruSeq;
+        }
+
+        InterlockedMax(ref _peerAckedSeq, acked);
+        await _store.RecordPeerAckAsync(acked, ct);
+        return lookahead;
+    }
+
+    private async Task ApplyDeltaEntriesAsync(
+        ChannelWriter<SyncMessage> controlLane, IReadOnlyList<OplogEntry> batchEntries, CancellationToken ct)
+    {
+        if (batchEntries.Count == 0)
             return;
 
         var now = _utcNow();
         var nowMs = now.ToUnixTimeMilliseconds();
         var driftLimitMs = nowMs + (long)_options.MaxHlcDriftAhead.TotalMilliseconds;
-        var toApply = new List<OplogEntryRecord>(batch.Entries.Count);
+        var toApply = new List<OplogEntryRecord>(batchEntries.Count);
         var maxSeq = 0L;
         var driftDeadLettered = 0;
 
-        foreach (var proto in batch.Entries)
+        foreach (var proto in batchEntries)
         {
             if (proto.Seq > maxSeq) maxSeq = proto.Seq;
             var entry = FromProto(proto);

ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/ReplicationOptions.cs

@@ -14,9 +14,44 @@ public sealed class ReplicationOptions
     /// <summary>How often the change pump flushes pending oplog deltas to the peer.</summary>
     public TimeSpan FlushInterval { get; set; } = TimeSpan.FromMilliseconds(250);
 
-    /// <summary>Maximum oplog entries sent per outbound batch.</summary>
+    /// <summary>
+    /// Secondary, row-count cap on an outbound batch (and the snapshot streamer's DB page size).
+    /// <see cref="MaxBatchBytes"/> is the primary cap.
+    /// </summary>
+    /// <remarks>
+    /// Historically this was the ONLY cap, which made it a proxy for a byte budget that consumers
+    /// had to tune by hand against their widest row — ScadaBridge pinned it to 16 because a 70 KB
+    /// <c>config_json</c> times the 500 default is ~35 MB against a 4 MB gRPC limit. With
+    /// <see cref="MaxBatchBytes"/> doing that job directly, this is back to being what its name
+    /// says: a bound on how many rows one message may carry.
+    /// </remarks>
     public int MaxBatchSize { get; set; } = 500;
 
+    /// <summary>
+    /// Primary cap on an outbound batch: the summed serialized size of the entries in one
+    /// <c>DeltaBatch</c> / <c>SnapshotBatch</c> message, in bytes. Default 2 MB.
+    /// </summary>
+    /// <remarks>
+    /// <para>
+    /// Sized at half the default 4 MB gRPC receive limit, so the message framing, the oneof wrapper
+    /// and any per-entry overhead the estimate does not model all fit inside the margin.
+    /// </para>
+    /// <para>
+    /// <b>A single row larger than the budget is still sent, alone.</b> Skipping it would stall the
+    /// stream permanently on one oversized row; sending it gives the transport the chance to accept
+    /// it (gRPC limits are configurable) and, if it refuses, a loud transport error naming the size
+    /// rather than silent non-convergence.
+    /// </para>
+    /// <para>
+    /// <b>Wire compatibility:</b> none required. Batch size is a sender-side choice the protocol
+    /// already leaves free — <c>DeltaBatch</c>/<c>SnapshotBatch</c> carry a repeated field with no
+    /// declared count — so a node on this version interoperates with a peer on any earlier one in
+    /// both directions, with no proto change and no <c>lib_schema_version</c> bump. (ScadaBridge
+    /// site pairs stop and start together regardless; this simply does not depend on it.)
+    /// </para>
+    /// </remarks>
+    public int MaxBatchBytes { get; set; } = 2 * 1024 * 1024;
+
     /// <summary>Backlog ceiling: exceeding it flags a snapshot resync and prunes the oldest oplog rows.</summary>
     public long MaxOplogRows { get; set; } = 1_000_000;
 

ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/ReplicationOptionsValidator.cs

@@ -28,6 +28,8 @@ public sealed class ReplicationOptionsValidator : IValidateOptions<ReplicationOp
             failures.Add($"LocalDb:Replication:FlushInterval must be greater than zero; got {options.FlushInterval}.");
         if (options.MaxBatchSize <= 0)
             failures.Add($"LocalDb:Replication:MaxBatchSize must be greater than 0; got {options.MaxBatchSize}.");
+        if (options.MaxBatchBytes <= 0)
+            failures.Add($"LocalDb:Replication:MaxBatchBytes must be greater than 0; got {options.MaxBatchBytes}.");
         if (options.MaxOplogRows <= 0)
             failures.Add($"LocalDb:Replication:MaxOplogRows must be greater than 0; got {options.MaxOplogRows}.");
         if (options.MaxOplogAge <= TimeSpan.Zero)

ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/ILocalDb.cs

@@ -42,6 +42,105 @@ public interface ILocalDb
     /// </summary>
     ReplicatedTable RegisterReplicated(string tableName);
 
+    /// <summary>
+    /// As <see cref="RegisterReplicated(string)"/>, additionally baselining rows that already exist
+    /// in the table when <paramref name="baselineExistingRows"/> is true.
+    /// </summary>
+    /// <param name="tableName">The table to opt into replication.</param>
+    /// <param name="baselineExistingRows">
+    /// When true, every row already in the table that has no <c>__localdb_row_version</c> entry is
+    /// given one at the <b>baseline floor</b> — HLC <c>0</c>, stamped with this node's id — and, if
+    /// anything was seeded, a snapshot resync is flagged so the peer actually receives them.
+    /// </param>
+    /// <remarks>
+    /// <para>
+    /// <b>The problem this solves.</b> Capture is change-data-capture: without baselining, a row
+    /// written before registration exists in neither the oplog nor the row-version ledger, and the
+    /// snapshot streamer pages from the LEDGER — so a node that starts replicating LATE never sends
+    /// its pre-existing rows, silently and forever. Opting in seeds the ledger so a snapshot can
+    /// carry them.
+    /// </para>
+    /// <para>
+    /// <b>Why the floor is HLC 0, and why that is LWW-safe.</b> A seeded version must never beat a
+    /// genuine write. Every genuine HLC is a UTC millisecond shifted left 16 bits, so every genuine
+    /// HLC is greater than 0: a baselined row therefore LOSES every comparison against a real remote
+    /// write of the same key, and wins only where the peer has no version of that key at all
+    /// (which is precisely the case baselining exists to fix). It cannot perturb the local clock
+    /// either — clock recovery takes a MAX over existing HLCs, and 0 raises nothing.
+    /// </para>
+    /// <para>
+    /// <b>The one genuinely ambiguous case</b> is both nodes baselining the SAME key (e.g. both
+    /// restored the same legacy file). Both sides hold HLC 0, so the LWW tie-break falls to the
+    /// node-id ordinal: deterministic and convergent, but arbitrary as to content. Nothing can do
+    /// better — two rows of unknown, unordered antiquity carry no information about which is newer.
+    /// </para>
+    /// <para>
+    /// <b>Idempotent.</b> Seeding uses <c>ON CONFLICT DO NOTHING</c>, so a row that already has a
+    /// 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);
+
+    /// <summary>
+    /// Removes a table from replication: drops its three capture triggers and prunes its rows from
+    /// <c>__localdb_oplog</c> and <c>__localdb_row_version</c>, in one transaction. Idempotent, and
+    /// deliberately usable on a table this process never registered — cleaning up triggers left
+    /// behind by an EARLIER build is the primary use.
+    /// </summary>
+    /// <param name="tableName">The table to stop capturing. The table's own data is never touched.</param>
+    /// <returns>True if anything was removed (a trigger dropped or a row pruned); false if there was nothing to clean up.</returns>
+    /// <remarks>
+    /// <para>
+    /// <b>Intended caller:</b> an application that has decided this node no longer replicates (or no
+    /// longer replicates this table). Capture is not free — the three triggers run inside every
+    /// consumer write transaction and append to an oplog nothing will ever drain — so a node with no
+    /// peer should not be paying for it.
+    /// </para>
+    /// <para>
+    /// <b>Pruning the row-version ledger is safe with respect to re-registration</b>, because
+    /// <see cref="RegisterReplicated(string, bool)"/> can re-baseline. Without baselining, a
+    /// deregister-then-register cycle would leave the previously-known rows unstreamable — the same
+    /// late-opt-in hole — so the two APIs are designed as a pair.
+    /// </para>
+    /// <para>
+    /// <b>Deregistration must be symmetric across a replicating pair, and the library enforces it
+    /// loudly.</b> The sync handshake compares the two nodes' registered-table digests fail-closed,
+    /// so a node that drops a table its peer still replicates simply stops syncing — with a
+    /// 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
+    /// trigger an earlier snapshot resync, never suppress one — and on the node this API is for
+    /// (no peer at all) depth is moot.
+    /// </para>
+    /// </remarks>
+    bool DeregisterReplicated(string tableName);
+
     /// <summary>The tables currently opted into replication, keyed by table name.</summary>
     IReadOnlyDictionary<string, ReplicatedTable> ReplicatedTables { get; }
 }

ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/Internal/LocalDbSchema.cs

@@ -1,3 +1,4 @@
+using System.Globalization;
 using Microsoft.Data.Sqlite;
 
 namespace ZB.MOM.WW.LocalDb.Internal;
@@ -7,6 +8,29 @@ namespace ZB.MOM.WW.LocalDb.Internal;
 /// </summary>
 internal static class LocalDbSchema
 {
+    /// <summary>
+    /// The on-disk bookkeeping-schema version this build creates and upgrades to. Bumping it means
+    /// adding a case to <see cref="Upgrade"/>; existing files are upgraded in place on open.
+    /// </summary>
+    /// <remarks>
+    /// Deliberately NOT the same number as <c>SyncSession.LibSchemaVersion</c>, which is the WIRE
+    /// protocol version and is compared fail-closed against the peer's. This one is local-file-only:
+    /// two nodes may briefly sit on different file-schema versions during an upgrade and still sync,
+    /// because nothing about the file layout crosses the wire.
+    /// <list type="bullet">
+    /// <item><description>
+    /// <b>1</b> — the original layout, which also created <c>__localdb_oplog_hlc</c>.
+    /// </description></item>
+    /// <item><description>
+    /// <b>2</b> — drops <c>__localdb_oplog_hlc</c>. That index (<c>table_name, pk_json, hlc</c>) was
+    /// never read by any query: every oplog access path is by <c>seq</c> (the rowid) or a full
+    /// <c>hlc</c>-predicate delete. It cost a b-tree insert on the CAPTURE path — i.e. inside every
+    /// consumer write transaction — to serve nothing.
+    /// </description></item>
+    /// </list>
+    /// </remarks>
+    internal const long CurrentSchemaVersion = 2;
+
     private const string Ddl = """
         CREATE TABLE IF NOT EXISTS __localdb_meta (
           id INTEGER PRIMARY KEY CHECK (id = 1),
@@ -25,7 +49,6 @@ internal static class LocalDbSchema
           node_id TEXT NOT NULL,
           is_tombstone INTEGER NOT NULL DEFAULT 0
         );
-        CREATE INDEX IF NOT EXISTS __localdb_oplog_hlc ON __localdb_oplog(table_name, pk_json, hlc);
         CREATE TABLE IF NOT EXISTS __localdb_peer_state (
           id INTEGER PRIMARY KEY CHECK (id = 1),
           peer_node_id TEXT NULL,
@@ -80,14 +103,61 @@ internal static class LocalDbSchema
             meta.Transaction = transaction;
             meta.CommandText =
                 "INSERT INTO __localdb_meta (id, node_id, schema_version) " +
-                "SELECT 1, $nodeId, 1 WHERE NOT EXISTS (SELECT 1 FROM __localdb_meta)";
+                "SELECT 1, $nodeId, $version WHERE NOT EXISTS (SELECT 1 FROM __localdb_meta)";
             meta.Parameters.AddWithValue("$nodeId", Guid.NewGuid().ToString("D"));
+            meta.Parameters.AddWithValue("$version", CurrentSchemaVersion);
             meta.ExecuteNonQuery();
         }
 
+        Upgrade(connection, transaction);
+
         transaction.Commit();
     }
 
+    /// <summary>
+    /// Brings a file created by an OLDER build up to <see cref="CurrentSchemaVersion"/>, in the
+    /// caller's transaction. A file already at the current version is untouched.
+    /// </summary>
+    /// <remarks>
+    /// Each step is written to be safe to re-run (IF EXISTS / IF NOT EXISTS), so a crash between the
+    /// DDL and the version stamp costs nothing but a repeat on the next open. Steps run in ascending
+    /// order and the stamp is written once at the end, inside the same transaction as the DDL — a
+    /// file is therefore never observed at the new version with an unapplied step.
+    /// </remarks>
+    private static void Upgrade(SqliteConnection connection, SqliteTransaction transaction)
+    {
+        long current;
+        using (var read = connection.CreateCommand())
+        {
+            read.Transaction = transaction;
+            read.CommandText = "SELECT schema_version FROM __localdb_meta WHERE id = 1";
+            current = Convert.ToInt64(read.ExecuteScalar(), CultureInfo.InvariantCulture);
+        }
+
+        if (current >= CurrentSchemaVersion)
+            return;
+
+        var steps = new List<string>();
+
+        // 1 -> 2: drop the write-amplifying, never-read oplog index (see CurrentSchemaVersion).
+        if (current < 2)
+            steps.Add("DROP INDEX IF EXISTS __localdb_oplog_hlc;");
+
+        if (steps.Count > 0)
+        {
+            using var ddl = connection.CreateCommand();
+            ddl.Transaction = transaction;
+            ddl.CommandText = string.Concat(steps);
+            ddl.ExecuteNonQuery();
+        }
+
+        using var stamp = connection.CreateCommand();
+        stamp.Transaction = transaction;
+        stamp.CommandText = "UPDATE __localdb_meta SET schema_version = $version WHERE id = 1";
+        stamp.Parameters.AddWithValue("$version", CurrentSchemaVersion);
+        stamp.ExecuteNonQuery();
+    }
+
     internal static (string NodeId, long HlcHighWater) ReadMeta(SqliteConnection connection)
     {
         using var cmd = connection.CreateCommand();
@@ -99,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();
     }

ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/Internal/SqliteLocalDb.cs

@@ -138,7 +138,15 @@ internal sealed class SqliteLocalDb : ILocalDb, IDisposable
         return await ReadAllAsync(cmd, map, ct);
     }
 
-    public async Task<ILocalDbTransaction> BeginTransactionAsync(CancellationToken ct = default)
+    public async Task<ILocalDbTransaction> BeginTransactionAsync(CancellationToken ct = default) =>
+        await BeginPreparedTransactionAsync(ct);
+
+    /// <summary>
+    /// As <see cref="BeginTransactionAsync"/>, but typed so internal callers reach the underlying
+    /// connection and transaction — which is what lets a hot loop hold <see cref="SqliteCommand"/>s
+    /// across many rows and rebind them, instead of re-preparing statements per statement per row.
+    /// </summary>
+    internal async Task<Transaction> BeginPreparedTransactionAsync(CancellationToken ct = default)
     {
         var conn = await CreateConnectionAsync(ct);
         try
@@ -155,8 +163,11 @@ internal sealed class SqliteLocalDb : ILocalDb, IDisposable
 
     public IReadOnlyDictionary<string, ReplicatedTable> ReplicatedTables => _replicatedTables;
 
+    public ReplicatedTable RegisterReplicated(string tableName) =>
+        RegisterReplicated(tableName, baselineExistingRows: false);
+
     // Thread-safe: all master-connection work runs under _masterLock.
-    public ReplicatedTable RegisterReplicated(string tableName)
+    public ReplicatedTable RegisterReplicated(string tableName, bool baselineExistingRows)
     {
         if (string.IsNullOrWhiteSpace(tableName) || !TableNameRegex.IsMatch(tableName))
             throw new LocalDbRegistrationException(
@@ -201,6 +212,10 @@ internal sealed class SqliteLocalDb : ILocalDb, IDisposable
                     cmd.CommandText = script;
                     cmd.ExecuteNonQuery();
                 }
+
+                if (baselineExistingRows)
+                    BaselineExistingRows(tx, table);
+
                 tx.Commit();
             }
 
@@ -209,6 +224,125 @@ internal sealed class SqliteLocalDb : ILocalDb, IDisposable
         }
     }
 
+    // Thread-safe: all master-connection work runs under _masterLock.
+    public bool DeregisterReplicated(string tableName)
+    {
+        if (string.IsNullOrWhiteSpace(tableName) || !TableNameRegex.IsMatch(tableName))
+            throw new LocalDbRegistrationException(
+                $"Cannot deregister table '{tableName}': name must be a plain identifier matching {TableNameRegex}.");
+
+        lock (_masterLock)
+        {
+            using var tx = _master.BeginTransaction();
+
+            // Counted BEFORE the drop: DROP TRIGGER reports no row count, and "did anything change?"
+            // is the caller's cue to log a one-time cleanup rather than a no-op every boot.
+            long triggersPresent;
+            using (var probe = _master.CreateCommand())
+            {
+                probe.Transaction = tx;
+                probe.CommandText =
+                    "SELECT COUNT(*) FROM sqlite_master WHERE type = 'trigger' AND name IN ($ai, $au, $ad)";
+                foreach (var suffix in TriggerSqlGenerator.Suffixes)
+                    probe.Parameters.AddWithValue("$" + suffix, TriggerSqlGenerator.TriggerName(tableName, suffix));
+                triggersPresent = Convert.ToInt64(probe.ExecuteScalar(), CultureInfo.InvariantCulture);
+            }
+
+            using (var drop = _master.CreateCommand())
+            {
+                drop.Transaction = tx;
+                drop.CommandText = TriggerSqlGenerator.BuildUninstallScript(tableName);
+                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.
+            int prunedRows;
+            using (var prune = _master.CreateCommand())
+            {
+                prune.Transaction = tx;
+                prune.CommandText =
+                    "DELETE FROM __localdb_oplog WHERE table_name = $t;" +
+                    "DELETE FROM __localdb_row_version WHERE table_name = $t;";
+                prune.Parameters.AddWithValue("$t", tableName);
+                prunedRows = prune.ExecuteNonQuery();
+            }
+
+            tx.Commit();
+
+            _replicatedTables.TryRemove(tableName, out _);
+            return triggersPresent > 0 || prunedRows > 0;
+        }
+    }
+
+    /// <summary>
+    /// Seeds <c>__localdb_row_version</c> at the baseline floor for every row of
+    /// <paramref name="table"/> that has no ledger entry, and flags a snapshot resync if it seeded
+    /// anything. Runs in the registration transaction so triggers and baseline commit together.
+    /// </summary>
+    /// <remarks>
+    /// <para>
+    /// The floor is HLC <c>0</c> — see <see cref="ILocalDb.RegisterReplicated(string, bool)"/> for
+    /// why that is the only LWW-safe choice.
+    /// </para>
+    /// <para>
+    /// <b>The <c>needs_snapshot</c> flag is what makes this actually deliver.</b> Seeding the ledger
+    /// alone changes nothing on the wire: a node whose oplog starts at seq 1 presents no gap at the
+    /// handshake, so the peer is told it needs no snapshot and only ever receives post-registration
+    /// deltas. Flagging the resync is what turns the seeded ledger into rows the peer receives.
+    /// Deliberately NOT done by writing oplog entries instead: that would dump the entire table into
+    /// 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)
+    {
+        int seeded;
+        using (var cmd = _master.CreateCommand())
+        {
+            cmd.Transaction = tx;
+            // The trailing WHERE 1 disambiguates the upsert parse on a SELECT source — the same
+            // reason TriggerSqlGenerator's row-version statements carry one.
+            cmd.CommandText =
+                "INSERT INTO __localdb_row_version (table_name, pk_json, hlc, node_id, is_tombstone, tombstone_utc) " +
+                $"SELECT $t, {TriggerSqlGenerator.JsonObject("t", table.PkColumns)}, 0, $node, 0, NULL " +
+                $"FROM {TriggerSqlGenerator.Ident(table.Name)} t WHERE 1 " +
+                "ON CONFLICT(table_name, pk_json) DO NOTHING";
+            cmd.Parameters.AddWithValue("$t", table.Name);
+            cmd.Parameters.AddWithValue("$node", _nodeId);
+            seeded = cmd.ExecuteNonQuery();
+        }
+
+        if (seeded == 0)
+            return;
+
+        using var flag = _master.CreateCommand();
+        flag.Transaction = tx;
+        flag.CommandText = "UPDATE __localdb_peer_state SET needs_snapshot = 1 WHERE id = 1";
+        flag.ExecuteNonQuery();
+    }
+
     public void Dispose()
     {
         if (_disposed)
@@ -329,8 +463,14 @@ internal sealed class SqliteLocalDb : ILocalDb, IDisposable
         return Convert.ToInt64(cmd.ExecuteScalar(), CultureInfo.InvariantCulture);
     }
 
-    private sealed class Transaction(SqliteConnection connection, SqliteTransaction transaction) : ILocalDbTransaction
+    internal sealed class Transaction(SqliteConnection connection, SqliteTransaction transaction) : ILocalDbTransaction
     {
+        /// <summary>The pinned connection, for internal callers holding their own commands.</summary>
+        internal SqliteConnection Connection => connection;
+
+        /// <summary>The underlying transaction, to assign to internally-held commands.</summary>
+        internal SqliteTransaction Sqlite => transaction;
+
         public async Task<int> ExecuteAsync(string sql, object? parameters = null, CancellationToken ct = default)
         {
             await using var cmd = connection.CreateCommand();

ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/Registration/TriggerSqlGenerator.cs

@@ -27,14 +27,16 @@ internal static class TriggerSqlGenerator
         "ON CONFLICT(table_name, pk_json) DO UPDATE SET hlc = excluded.hlc, node_id = excluded.node_id, " +
         "is_tombstone = excluded.is_tombstone, tombstone_utc = excluded.tombstone_utc";
 
+    /// <summary>The three trigger suffixes, in the order the install/uninstall scripts emit them.</summary>
+    internal static readonly string[] Suffixes = ["ai", "au", "ad"];
+
     public static string TriggerName(string table, string suffix) => $"__localdb_{table}_{suffix}";
 
     /// <summary>DROP IF EXISTS + CREATE for all three triggers, ready to run in one transaction.</summary>
     public static string BuildInstallScript(ReplicatedTable table)
     {
         var sb = new StringBuilder();
-        foreach (var suffix in new[] { "ai", "au", "ad" })
-            sb.Append("DROP TRIGGER IF EXISTS ").Append(Ident(TriggerName(table.Name, suffix))).Append(";\n");
+        sb.Append(BuildUninstallScript(table.Name));
 
         sb.Append(BuildInsert(table)).Append('\n');
         sb.Append(BuildUpdate(table)).Append('\n');
@@ -42,6 +44,20 @@ internal static class TriggerSqlGenerator
         return sb.ToString();
     }
 
+    /// <summary>
+    /// DROP IF EXISTS for all three triggers of <paramref name="tableName"/>. Takes the bare name
+    /// rather than a <see cref="ReplicatedTable"/> because uninstall must work for a table this
+    /// process never registered — the whole point of the cleanup API is stale triggers left by an
+    /// EARLIER build, whose column list (and therefore digest) is unavailable and irrelevant.
+    /// </summary>
+    public static string BuildUninstallScript(string tableName)
+    {
+        var sb = new StringBuilder();
+        foreach (var suffix in Suffixes)
+            sb.Append("DROP TRIGGER IF EXISTS ").Append(Ident(TriggerName(tableName, suffix))).Append(";\n");
+        return sb.ToString();
+    }
+
     private static string BuildInsert(ReplicatedTable t) =>
         Wrap(t, "ai", "INSERT", NewRowCapture(t));
 
@@ -112,13 +128,16 @@ internal static class TriggerSqlGenerator
     }
 
     // json_object('col', ALIAS."col", ...) — natively handles NULLs and SQLite scalar types.
-    private static string JsonObject(string alias, IEnumerable<string> columns)
+    // Internal rather than private: baseline seeding must build a pk_json identical to what the
+    // capture triggers emit, character for character, or a seeded ledger key would never match the
+    // key a later capture writes for the same row.
+    internal static string JsonObject(string alias, IEnumerable<string> columns)
     {
         var args = columns.Select(c => $"{Literal(c)}, {alias}.{Ident(c)}");
         return $"json_object({string.Join(", ", args)})";
     }
 
-    private static string Ident(string name) => "\"" + name.Replace("\"", "\"\"") + "\"";
+    internal static string Ident(string name) => "\"" + name.Replace("\"", "\"\"") + "\"";
 
     private static string Literal(string value) => "'" + value.Replace("'", "''") + "'";
 }

ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/BatchBudgetTests.cs

@@ -0,0 +1,430 @@
+using System.Threading.Channels;
+using Microsoft.Data.Sqlite;
+using Microsoft.Extensions.Logging.Abstractions;
+using ZB.MOM.WW.LocalDb.Contracts;
+using ZB.MOM.WW.LocalDb.Internal;
+using ZB.MOM.WW.LocalDb.Replication;
+using ZB.MOM.WW.LocalDb.Replication.Internal;
+
+namespace ZB.MOM.WW.LocalDb.Tests;
+
+/// <summary>
+/// Outbound batching by summed serialized BYTES (with the row count kept as a secondary cap), and
+/// the inbound coalescing that keeps one transaction + one ack per burst rather than per message.
+/// </summary>
+public sealed class BatchBudgetTests : IDisposable
+{
+    private static readonly TimeSpan RunTimeout = TimeSpan.FromSeconds(15);
+
+    private readonly List<string> _paths = [];
+    private readonly List<IDisposable> _disposables = [];
+
+    public void Dispose()
+    {
+        foreach (var d in _disposables)
+            d.Dispose();
+        SqliteConnection.ClearAllPools();
+        foreach (var p in _paths)
+        {
+            if (File.Exists(p)) File.Delete(p);
+            if (File.Exists(p + "-wal")) File.Delete(p + "-wal");
+            if (File.Exists(p + "-shm")) File.Delete(p + "-shm");
+        }
+    }
+
+    private async Task<SqliteLocalDb> NewOrdersDbAsync()
+    {
+        var path = Path.Combine(Path.GetTempPath(), Guid.NewGuid() + ".db");
+        _paths.Add(path);
+        var db = new SqliteLocalDb(new LocalDbOptions { Path = path });
+        _disposables.Add(db);
+        await db.ExecuteAsync("CREATE TABLE orders (id INTEGER PRIMARY KEY, sku TEXT, qty INTEGER)");
+        db.RegisterReplicated("orders");
+        return db;
+    }
+
+    private static (SyncDuplex Duplex, ChannelWriter<SyncMessage> ToSession, ChannelReader<SyncMessage> FromSession) ScriptedPeer()
+    {
+        var toSession = Channel.CreateUnbounded<SyncMessage>();
+        var fromSession = Channel.CreateUnbounded<SyncMessage>();
+        var duplex = new SyncDuplex
+        {
+            Send = (m, ct) => fromSession.Writer.WriteAsync(m, ct).AsTask(),
+            Inbox = toSession.Reader,
+        };
+        return (duplex, toSession.Writer, fromSession.Reader);
+    }
+
+    private static Handshake MatchingHandshake(SqliteLocalDb sessionDb, string peerNodeId)
+    {
+        var h = new Handshake { NodeId = peerNodeId, LibSchemaVersion = 1, LastAppliedRemoteSeq = 0 };
+        foreach (var t in sessionDb.ReplicatedTables.Values.OrderBy(t => t.Name, StringComparer.Ordinal))
+            h.Tables.Add(new TableDigest { TableName = t.Name, Digest = t.Digest });
+        return h;
+    }
+
+    private static async Task<SyncMessage> NextAsync(ChannelReader<SyncMessage> reader, CancellationToken ct)
+    {
+        await reader.WaitToReadAsync(ct);
+        reader.TryRead(out var msg);
+        return msg!;
+    }
+
+    private static async Task SwallowAsync(Task task)
+    {
+        try { await task; }
+        catch (OperationCanceledException) { }
+    }
+
+    // Queues a complete peer handshake AHEAD of the session's own, so anything queued after it is
+    // guaranteed to be sitting in the inbox by the time the receive loop starts. That is what makes
+    // the coalescing assertions deterministic rather than a race with the reader.
+    private static async Task PreloadHandshakeAsync(
+        SqliteLocalDb db, ChannelWriter<SyncMessage> toSession, CancellationToken ct)
+    {
+        await toSession.WriteAsync(new SyncMessage { Handshake = MatchingHandshake(db, "peer") }, ct);
+        await toSession.WriteAsync(
+            new SyncMessage { HandshakeAck = new HandshakeAck { NodeId = "peer", SnapshotRequired = false } }, ct);
+    }
+
+    private static long BatchBytes(DeltaBatch batch) =>
+        batch.Entries.Sum(e => 1L + Google.Protobuf.CodedOutputStream.ComputeMessageSize(e));
+
+    private static long BatchBytes(SnapshotBatch batch) =>
+        batch.Rows.Sum(r => 1L + Google.Protobuf.CodedOutputStream.ComputeMessageSize(r));
+
+    private static SyncSession NewSession(SqliteLocalDb db, ReplicationOptions options)
+    {
+        var store = new OplogStore(db, options);
+        var applier = new LwwApplier(db);
+        return new SyncSession(db, store, applier, options, NullLogger.Instance);
+    }
+
+    private static OplogEntry ProtoEntry(long seq, long id, long hlc) =>
+        new()
+        {
+            Seq = seq,
+            TableName = "orders",
+            PkJson = $"{{\"id\":{id}}}",
+            RowJson = $"{{\"id\":{id},\"sku\":\"S\",\"qty\":1}}",
+            Hlc = hlc,
+            NodeId = "peer",
+            IsTombstone = false,
+        };
+
+    // ---- Outbound: delta pump --------------------------------------------------------------
+
+    [Fact]
+    public async Task Pump_SplitsOneReadIntoSeveralMessages_EachWithinTheByteBudget()
+    {
+        const int budget = 4_000;
+        var db = await NewOrdersDbAsync();
+        var wide = new string('x', 900);
+        for (var i = 1; i <= 20; i++)
+            await db.ExecuteAsync(
+                "INSERT INTO orders (id, sku, qty) VALUES (@id, @sku, @id)", new { id = i, sku = wide });
+
+        // Row count alone would put all 20 wide rows (~18 KB) in ONE message; the byte budget is
+        // what actually bounds the message the transport has to carry.
+        var options = new ReplicationOptions
+        {
+            FlushInterval = TimeSpan.FromMilliseconds(20), MaxBatchSize = 500, MaxBatchBytes = budget,
+        };
+        var session = NewSession(db, options);
+        var (duplex, toSession, fromSession) = ScriptedPeer();
+
+        using var cts = new CancellationTokenSource(RunTimeout);
+        var run = session.RunAsync(duplex, cts.Token);
+        await PreloadHandshakeAsync(db, toSession, cts.Token);
+
+        var seen = new List<long>();
+        while (seen.Count < 20)
+        {
+            var msg = await NextAsync(fromSession, cts.Token);
+            if (msg.MsgCase != SyncMessage.MsgOneofCase.DeltaBatch)
+                continue;
+            Assert.True(BatchBytes(msg.DeltaBatch) <= budget,
+                $"a {BatchBytes(msg.DeltaBatch)}-byte message exceeded the {budget}-byte budget");
+            Assert.True(msg.DeltaBatch.Entries.Count > 1, "the budget should still fit several of these rows");
+            seen.AddRange(msg.DeltaBatch.Entries.Select(e => e.Seq));
+        }
+
+        // Split, not dropped or duplicated: every captured seq travelled exactly once, in order.
+        Assert.Equal(20, seen.Count);
+        Assert.Equal(seen.OrderBy(x => x).ToList(), seen);
+        Assert.Equal(20, seen.Distinct().Count());
+
+        cts.Cancel();
+        await SwallowAsync(run);
+    }
+
+    [Fact]
+    public async Task Pump_RowLargerThanTheBudget_IsSentAloneRatherThanStallingTheStream()
+    {
+        var db = await NewOrdersDbAsync();
+        await db.ExecuteAsync(
+            "INSERT INTO orders (id, sku, qty) VALUES (1, @sku, 1)", new { sku = new string('x', 5_000) });
+        await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (2, 'small', 2)");
+
+        var options = new ReplicationOptions
+        {
+            FlushInterval = TimeSpan.FromMilliseconds(20), MaxBatchSize = 500, MaxBatchBytes = 100,
+        };
+        var session = NewSession(db, options);
+        var (duplex, toSession, fromSession) = ScriptedPeer();
+
+        using var cts = new CancellationTokenSource(RunTimeout);
+        var run = session.RunAsync(duplex, cts.Token);
+        await PreloadHandshakeAsync(db, toSession, cts.Token);
+
+        var batches = new List<DeltaBatch>();
+        while (batches.Sum(b => b.Entries.Count) < 2)
+        {
+            var msg = await NextAsync(fromSession, cts.Token);
+            if (msg.MsgCase == SyncMessage.MsgOneofCase.DeltaBatch)
+                batches.Add(msg.DeltaBatch);
+        }
+
+        // The oversized row goes out on its own — refusing it would wedge the stream on one row
+        // forever, and everything behind it with it.
+        Assert.Equal(2, batches.Count);
+        Assert.Single(batches[0].Entries);
+        Assert.Equal(1, batches[0].Entries[0].Seq);
+        Assert.True(BatchBytes(batches[0]) > options.MaxBatchBytes);
+        Assert.Single(batches[1].Entries);
+        Assert.Equal(2, batches[1].Entries[0].Seq);
+
+        cts.Cancel();
+        await SwallowAsync(run);
+    }
+
+    [Fact]
+    public async Task Pump_RowCountRemainsASecondaryCap()
+    {
+        var db = await NewOrdersDbAsync();
+        for (var i = 1; i <= 12; i++)
+            await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (@id, 'S', @id)", new { id = i });
+
+        // Tiny rows, enormous byte budget: only the row cap can be doing the splitting here.
+        var options = new ReplicationOptions
+        {
+            FlushInterval = TimeSpan.FromMilliseconds(20), MaxBatchSize = 5, MaxBatchBytes = 8 * 1024 * 1024,
+        };
+        var session = NewSession(db, options);
+        var (duplex, toSession, fromSession) = ScriptedPeer();
+
+        using var cts = new CancellationTokenSource(RunTimeout);
+        var run = session.RunAsync(duplex, cts.Token);
+        await PreloadHandshakeAsync(db, toSession, cts.Token);
+
+        var seen = 0;
+        while (seen < 12)
+        {
+            var msg = await NextAsync(fromSession, cts.Token);
+            if (msg.MsgCase != SyncMessage.MsgOneofCase.DeltaBatch)
+                continue;
+            Assert.True(msg.DeltaBatch.Entries.Count <= 5);
+            seen += msg.DeltaBatch.Entries.Count;
+        }
+
+        cts.Cancel();
+        await SwallowAsync(run);
+    }
+
+    // ---- Outbound: snapshot --------------------------------------------------------------
+
+    [Fact]
+    public async Task Snapshot_SplitsPagesIntoMessagesWithinTheByteBudget()
+    {
+        const int budget = 4_000;
+        var db = await NewOrdersDbAsync();
+        var wide = new string('x', 900);
+        for (var i = 1; i <= 20; i++)
+            await db.ExecuteAsync(
+                "INSERT INTO orders (id, sku, qty) VALUES (@id, @sku, @id)", new { id = i, sku = wide });
+
+        // A snapshot is the easiest way to blow a message limit: it ships whole rows, all of them,
+        // back to back — so its page size has to be bounded in bytes too, not just in rows.
+        var options = new ReplicationOptions { MaxBatchSize = 500, MaxBatchBytes = budget };
+        var store = new OplogStore(db, options);
+        var streamer = new SnapshotStreamer(db, store, options, NullLogger.Instance);
+
+        var sent = new List<SyncMessage>();
+        await streamer.SendAsync((m, _) => { sent.Add(m); return Task.CompletedTask; }, default);
+
+        var batches = sent.Where(m => m.MsgCase == SyncMessage.MsgOneofCase.SnapshotBatch)
+            .Select(m => m.SnapshotBatch).ToList();
+        Assert.True(batches.Count > 1, "20 wide rows must not have travelled as one message");
+        Assert.All(batches, b => Assert.True(BatchBytes(b) <= budget,
+            $"a {BatchBytes(b)}-byte snapshot message exceeded the {budget}-byte budget"));
+
+        // Bracketed and complete: every row, exactly once.
+        Assert.Equal(SyncMessage.MsgOneofCase.SnapshotBegin, sent[0].MsgCase);
+        Assert.Equal(SyncMessage.MsgOneofCase.SnapshotComplete, sent[^1].MsgCase);
+        var pks = batches.SelectMany(b => b.Rows).Select(r => r.PkJson).ToList();
+        Assert.Equal(20, pks.Count);
+        Assert.Equal(20, pks.Distinct().Count());
+    }
+
+    [Fact]
+    public async Task Snapshot_RowLargerThanTheBudget_IsSentAlone()
+    {
+        var db = await NewOrdersDbAsync();
+        await db.ExecuteAsync(
+            "INSERT INTO orders (id, sku, qty) VALUES (1, @sku, 1)", new { sku = new string('x', 5_000) });
+        await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (2, 'small', 2)");
+
+        var options = new ReplicationOptions { MaxBatchSize = 500, MaxBatchBytes = 100 };
+        var store = new OplogStore(db, options);
+        var streamer = new SnapshotStreamer(db, store, options, NullLogger.Instance);
+
+        var sent = new List<SyncMessage>();
+        await streamer.SendAsync((m, _) => { sent.Add(m); return Task.CompletedTask; }, default);
+
+        var batches = sent.Where(m => m.MsgCase == SyncMessage.MsgOneofCase.SnapshotBatch)
+            .Select(m => m.SnapshotBatch).ToList();
+        Assert.Equal(2, batches.Count);
+        Assert.All(batches, b => Assert.Single(b.Rows));
+        Assert.Equal("{\"id\":1}", batches[0].Rows[0].PkJson);
+        Assert.Equal("{\"id\":2}", batches[1].Rows[0].PkJson);
+    }
+
+    // ---- Inbound: coalescing -------------------------------------------------------------
+
+    [Fact]
+    public async Task Receive_CoalescesQueuedDeltaBatches_IntoOneAck()
+    {
+        var db = await NewOrdersDbAsync();
+        var options = new ReplicationOptions { FlushInterval = TimeSpan.FromSeconds(30) };
+        var session = NewSession(db, options);
+        var (duplex, toSession, fromSession) = ScriptedPeer();
+
+        using var cts = new CancellationTokenSource(RunTimeout);
+        var run = session.RunAsync(duplex, cts.Token);
+
+        // Queued ahead of the receive loop, so all three are demonstrably in the inbox together.
+        await PreloadHandshakeAsync(db, toSession, cts.Token);
+        for (var i = 1; i <= 3; i++)
+        {
+            var batch = new DeltaBatch();
+            batch.Entries.Add(ProtoEntry(i, i, 1_000_000 + i));
+            await toSession.WriteAsync(new SyncMessage { DeltaBatch = batch }, cts.Token);
+        }
+
+        SyncMessage ack;
+        do { ack = await NextAsync(fromSession, cts.Token); }
+        while (ack.MsgCase != SyncMessage.MsgOneofCase.DeltaAck);
+
+        // The discriminator: uncoalesced, the FIRST ack could only cover seq 1.
+        Assert.Equal(3, ack.DeltaAck.AppliedThruSeq);
+
+        var rows = await db.QueryAsync("SELECT COUNT(*) FROM orders", r => r.GetInt64(0));
+        Assert.Equal(3, rows[0]);
+
+        cts.Cancel();
+        await SwallowAsync(run);
+    }
+
+    [Fact]
+    public async Task Receive_DeltaCoalescing_DoesNotSwallowTheMessageThatEndsTheRun()
+    {
+        // The risk the one-message lookahead exists to handle: coalescing reads ahead, and whatever
+        // it reads that is NOT a delta must still be processed, not dropped on the floor.
+        var db = await NewOrdersDbAsync();
+        var options = new ReplicationOptions { FlushInterval = TimeSpan.FromSeconds(30) };
+        var session = NewSession(db, options);
+        var captured = new CapturingSnapshotApplier();
+        session.SnapshotApplier = captured;
+        var (duplex, toSession, fromSession) = ScriptedPeer();
+
+        using var cts = new CancellationTokenSource(RunTimeout);
+        var run = session.RunAsync(duplex, cts.Token);
+
+        await PreloadHandshakeAsync(db, toSession, cts.Token);
+        var batch = new DeltaBatch();
+        batch.Entries.Add(ProtoEntry(1, 1, 1_000_001));
+        await toSession.WriteAsync(new SyncMessage { DeltaBatch = batch }, cts.Token);
+        await toSession.WriteAsync(
+            new SyncMessage { SnapshotBegin = new SnapshotBegin { AsOfSeq = 77 } }, cts.Token);
+
+        SyncMessage ack;
+        do { ack = await NextAsync(fromSession, cts.Token); }
+        while (ack.MsgCase != SyncMessage.MsgOneofCase.DeltaAck);
+        Assert.Equal(1, ack.DeltaAck.AppliedThruSeq);
+
+        // The SnapshotBegin that terminated the coalescing run still reached its handler.
+        await WaitForAsync(() => captured.BeginAsOfSeq == 77, RunTimeout);
+
+        cts.Cancel();
+        await SwallowAsync(run);
+    }
+
+    [Fact]
+    public async Task Receive_CoalescesQueuedAcks_ToTheHighestWatermark_WithoutLosingTheNextMessage()
+    {
+        var db = await NewOrdersDbAsync();
+        for (var i = 1; i <= 9; i++)
+            await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (@id, 'S', @id)", new { id = i });
+
+        var options = new ReplicationOptions { FlushInterval = TimeSpan.FromSeconds(30) };
+        var session = NewSession(db, options);
+        var captured = new CapturingSnapshotApplier();
+        session.SnapshotApplier = captured;
+        var (duplex, toSession, fromSession) = ScriptedPeer();
+
+        using var cts = new CancellationTokenSource(RunTimeout);
+        var run = session.RunAsync(duplex, cts.Token);
+
+        await PreloadHandshakeAsync(db, toSession, cts.Token);
+        // Out of order on purpose: collapsing acks must take the MAX, never the last.
+        foreach (var seq in new long[] { 4, 9, 2 })
+            await toSession.WriteAsync(
+                new SyncMessage { DeltaAck = new DeltaAck { AppliedThruSeq = seq } }, cts.Token);
+        await toSession.WriteAsync(
+            new SyncMessage { SnapshotBegin = new SnapshotBegin { AsOfSeq = 55 } }, cts.Token);
+
+        await WaitForAsync(async () =>
+        {
+            var acked = await db.QueryAsync(
+                "SELECT last_acked_seq FROM __localdb_peer_state WHERE id = 1", r => r.GetInt64(0));
+            return acked[0] == 9;
+        }, RunTimeout);
+
+        Assert.Equal(9, session.PeerAckedSeq);
+        await WaitForAsync(() => captured.BeginAsOfSeq == 55, RunTimeout);
+
+        cts.Cancel();
+        await SwallowAsync(run);
+    }
+
+    private sealed class CapturingSnapshotApplier : ISnapshotApplier
+    {
+        private long _beginAsOfSeq = -1;
+
+        public long BeginAsOfSeq => Interlocked.Read(ref _beginAsOfSeq);
+
+        public Task OnBeginAsync(SnapshotBegin begin, CancellationToken ct)
+        {
+            Interlocked.Exchange(ref _beginAsOfSeq, begin.AsOfSeq);
+            return Task.CompletedTask;
+        }
+
+        public Task OnBatchAsync(SnapshotBatch batch, CancellationToken ct) => Task.CompletedTask;
+
+        public Task OnCompleteAsync(SnapshotComplete complete, CancellationToken ct) => Task.CompletedTask;
+    }
+
+    private static Task WaitForAsync(Func<bool> predicate, TimeSpan timeout) =>
+        WaitForAsync(() => Task.FromResult(predicate()), timeout);
+
+    private static async Task WaitForAsync(Func<Task<bool>> predicate, TimeSpan timeout)
+    {
+        var deadline = DateTime.UtcNow + timeout;
+        while (DateTime.UtcNow < deadline)
+        {
+            if (await predicate()) return;
+            await Task.Delay(20);
+        }
+        throw new TimeoutException("Condition not reached within " + timeout);
+    }
+}

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]

ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/LocalDbSchemaTests.cs

@@ -55,8 +55,75 @@ public sealed class LocalDbSchemaTests : IDisposable
                  })
             Assert.Contains(expected, tables);
 
+        // __localdb_oplog_hlc is deliberately NOT created: schema v2 removed it. It indexed
+        // (table_name, pk_json, hlc), which no query has ever used — every oplog access path is by
+        // seq or a full hlc-predicate delete — while costing a b-tree insert inside every captured
+        // consumer write.
+        Assert.Equal(0, ScalarLong(conn,
+            "SELECT count(*) FROM sqlite_master WHERE type='index' AND name='__localdb_oplog_hlc'"));
+        Assert.Equal(LocalDbSchema.CurrentSchemaVersion,
+            ScalarLong(conn, "SELECT schema_version FROM __localdb_meta WHERE id=1"));
+    }
+
+    [Fact]
+    public void EnsureCreated_UpgradesLegacyFile_DropsOplogHlcIndex()
+    {
+        using var conn = Open();
+
+        // A file exactly as schema v1 left it: the tables, the index, and a v1 stamp.
+        LocalDbSchema.EnsureCreated(conn);
+        Exec(conn, "CREATE INDEX __localdb_oplog_hlc ON __localdb_oplog(table_name, pk_json, hlc);" +
+                   "UPDATE __localdb_meta SET schema_version = 1 WHERE id = 1;");
         Assert.Equal(1, ScalarLong(conn,
             "SELECT count(*) FROM sqlite_master WHERE type='index' AND name='__localdb_oplog_hlc'"));
+
+        LocalDbSchema.EnsureCreated(conn);
+
+        Assert.Equal(0, ScalarLong(conn,
+            "SELECT count(*) FROM sqlite_master WHERE type='index' AND name='__localdb_oplog_hlc'"));
+        Assert.Equal(LocalDbSchema.CurrentSchemaVersion,
+            ScalarLong(conn, "SELECT schema_version FROM __localdb_meta WHERE id=1"));
+    }
+
+    [Fact]
+    public void EnsureCreated_Upgrade_PreservesNodeIdentityAndOplogRows()
+    {
+        using var conn = Open();
+        LocalDbSchema.EnsureCreated(conn);
+        var nodeId = ScalarString(conn, "SELECT node_id FROM __localdb_meta WHERE id=1");
+        Exec(conn,
+            "CREATE INDEX __localdb_oplog_hlc ON __localdb_oplog(table_name, pk_json, hlc);" +
+            "UPDATE __localdb_meta SET schema_version = 1 WHERE id = 1;" +
+            "INSERT INTO __localdb_oplog (table_name, pk_json, row_json, hlc, node_id, is_tombstone) " +
+            "VALUES ('orders', '{\"id\":1}', '{\"id\":1}', 42, 'n', 0);" +
+            "UPDATE __localdb_peer_state SET last_acked_seq = 7 WHERE id = 1;");
+
+        LocalDbSchema.EnsureCreated(conn);
+
+        // The upgrade is DDL only — identity, replication watermarks and captured data survive it.
+        Assert.Equal(nodeId, ScalarString(conn, "SELECT node_id FROM __localdb_meta WHERE id=1"));
+        Assert.Equal(1, ScalarLong(conn, "SELECT COUNT(*) FROM __localdb_oplog"));
+        Assert.Equal(42, ScalarLong(conn, "SELECT hlc FROM __localdb_oplog"));
+        Assert.Equal(7, ScalarLong(conn, "SELECT last_acked_seq FROM __localdb_peer_state WHERE id=1"));
+    }
+
+    [Fact]
+    public void EnsureCreated_AlreadyCurrent_LeavesVersionAlone()
+    {
+        using var conn = Open();
+        LocalDbSchema.EnsureCreated(conn);
+        LocalDbSchema.EnsureCreated(conn);
+        LocalDbSchema.EnsureCreated(conn);
+
+        Assert.Equal(LocalDbSchema.CurrentSchemaVersion,
+            ScalarLong(conn, "SELECT schema_version FROM __localdb_meta WHERE id=1"));
+    }
+
+    private static void Exec(SqliteConnection conn, string sql)
+    {
+        using var cmd = conn.CreateCommand();
+        cmd.CommandText = sql;
+        cmd.ExecuteNonQuery();
     }
 
     [Fact]

ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/LwwApplierTests.cs

@@ -404,6 +404,64 @@ public sealed class LwwApplierTests : IDisposable
         Assert.Equal(("NEW", (long?)7), await ReadOrder(db, 2));
     }
 
+    [Fact]
+    public async Task Apply_PreparesEachStatementOnce_NotOncePerRow()
+    {
+        // The regression this guards: the old shape went through the generic transaction façade,
+        // which builds a fresh SqliteCommand — and therefore re-parses the SQL — per call. Three
+        // statements per applied row means a 200-row batch parsed 600 statements to run 6 distinct
+        // ones. Statement count must track distinct SQL, not rows.
+        using var db = await NewOrdersDb();
+        await db.ExecuteAsync("CREATE TABLE widgets (id INTEGER PRIMARY KEY, name TEXT)");
+        db.RegisterReplicated("widgets");
+
+        var entries = new List<OplogEntryRecord>();
+        for (var i = 1; i <= 200; i++)
+            entries.Add(Entry(i, $"{{\"id\":{i}}}", $"{{\"id\":{i},\"sku\":\"S\",\"qty\":{i}}}", 1_000_000 + i, "remote"));
+        // A second table and a tombstone, so every statement kind in the set is exercised.
+        entries.Add(new OplogEntryRecord(
+            201, "widgets", "{\"id\":1}", "{\"id\":1,\"name\":\"W\"}", 1_000_300, "remote", false));
+        entries.Add(Entry(202, "{\"id\":1}", null, 1_000_400, "remote", tombstone: true));
+
+        var applier = new LwwApplier(db);
+        var result = await applier.ApplyBatchAsync(entries, default);
+
+        Assert.Equal(202, result.Applied);
+        // Four table-independent statements (applying guard, version lookup, row-version write,
+        // peer state), plus an upsert per table touched and a delete for the one table tombstoned.
+        Assert.Equal(7, applier.LastBatchStatementCount);
+
+        // ...and the batch is still correct, statement reuse notwithstanding.
+        Assert.Null(await ReadOrder(db, 1));
+        Assert.Equal(("S", (long?)200), await ReadOrder(db, 200));
+        var widget = await db.QueryAsync("SELECT name FROM widgets WHERE id = 1", x => x.GetString(0));
+        Assert.Equal("W", widget[0]);
+    }
+
+    [Fact]
+    public async Task Apply_ReusedStatements_SurviveAPoisonRowMidBatch()
+    {
+        // A statement that faults is reset and reused for the following rows; if it were not, one
+        // poison row would silently corrupt every row after it in the batch.
+        using var db = await NewOrdersDb();
+
+        var applier = new LwwApplier(db);
+        var result = await applier.ApplyBatchAsync(
+        [
+            Entry(1, "{\"id\":1}", "{\"id\":1,\"sku\":\"BEFORE\",\"qty\":1}", 1_000_000, "remote"),
+            Entry(2, "{\"id\":2}", "NOT VALID JSON", 1_000_001, "remote"),
+            Entry(3, "{\"id\":3}", "{\"id\":3,\"sku\":\"AFTER\",\"qty\":3}", 1_000_002, "remote"),
+            Entry(4, "{\"id\":4}", "ALSO NOT JSON", 1_000_003, "remote"),
+            Entry(5, "{\"id\":5}", "{\"id\":5,\"sku\":\"LAST\",\"qty\":5}", 1_000_004, "remote"),
+        ], default);
+
+        Assert.Equal(3, result.Applied);
+        Assert.Equal(2, result.DeadLettered);
+        Assert.Equal(("BEFORE", (long?)1), await ReadOrder(db, 1));
+        Assert.Equal(("AFTER", (long?)3), await ReadOrder(db, 3));
+        Assert.Equal(("LAST", (long?)5), await ReadOrder(db, 5));
+    }
+
     [Fact]
     public async Task Apply_ReplaceExistingRow_DoesNotFireCaptureTriggers()
     {

ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/OplogStoreTests.cs

@@ -212,6 +212,137 @@ public sealed class OplogStoreTests : IDisposable
         Assert.Equal("{\"id\":2}", remaining[0].PkJson);
     }
 
+    // ---- O(1) backlog depth --------------------------------------------------------------
+
+    // The definition the O(1) formula has to reproduce. Every scenario below is asserted against
+    // this, not against a hand-computed number, so the two can never drift apart.
+    private static async Task<long> CountingDepth(SqliteLocalDb db) =>
+        (await db.QueryAsync(
+            "SELECT COUNT(*) FROM __localdb_oplog WHERE seq > (SELECT last_acked_seq FROM __localdb_peer_state WHERE id = 1)",
+            r => r.GetInt64(0)))[0];
+
+    private static async Task AssertDepthMatchesCount(SqliteLocalDb db, OplogStore store)
+    {
+        var expected = await CountingDepth(db);
+        Assert.Equal(expected, await store.GetOplogDepthAsync(default));
+        Assert.Equal(expected, store.GetOplogDepthSync());
+    }
+
+    [Fact]
+    public async Task OplogDepth_Empty_IsZero()
+    {
+        using var db = await NewOrdersDb();
+        var store = new OplogStore(db, new ReplicationOptions());
+
+        await AssertDepthMatchesCount(db, store);
+        Assert.Equal(0, await store.GetOplogDepthAsync(default));
+    }
+
+    [Fact]
+    public async Task OplogDepth_AllAcked_EmptiesToZero_NotToTheSeqRange()
+    {
+        using var db = await NewOrdersDb();
+        await WriteRows(db, 5);
+        var store = new OplogStore(db, new ReplicationOptions());
+        await store.RecordPeerAckAsync(await store.GetMaxSeqAsync(default), default);
+
+        await AssertDepthMatchesCount(db, store);
+        Assert.Equal(0, await store.GetOplogDepthAsync(default));
+    }
+
+    [Fact]
+    public async Task OplogDepth_AfterCapPrune_ReportsRetainedRows_NotTheWholeSeqRange()
+    {
+        // The reason the formula clamps its floor to MIN(seq) - 1. A cap prune leaves the retained
+        // rows far above an ack watermark that never moved, so a naive MAX(seq) - last_acked would
+        // report the entire seq range forever and re-flag the cap on every maintenance tick.
+        using var db = await NewOrdersDb();
+        await WriteRows(db, 40);
+        var store = new OplogStore(db, new ReplicationOptions { MaxOplogRows = 5 });
+
+        Assert.True(await store.EnforceCapsAsync(default));
+
+        await AssertDepthMatchesCount(db, store);
+        Assert.Equal(5, await store.GetOplogDepthAsync(default));
+        // And the next tick is clean: depth is not over cap any more.
+        Assert.False(await store.EnforceCapsAsync(default));
+    }
+
+    [Fact]
+    public async Task OplogDepth_AfterAgePrune_MatchesCount()
+    {
+        using var db = await NewOrdersDb();
+        await WriteRows(db, 6);
+        // Backdate the oldest three past the age cap; HLCs rise with seq, so this is a prefix.
+        var expired = DateTimeOffset.UtcNow.AddDays(-30).ToUnixTimeMilliseconds() << 16;
+        await db.ExecuteAsync(
+            "UPDATE __localdb_oplog SET hlc = @h WHERE seq IN (SELECT seq FROM __localdb_oplog ORDER BY seq LIMIT 3)",
+            new { h = expired });
+        var store = new OplogStore(db, new ReplicationOptions());
+
+        Assert.True(await store.EnforceCapsAsync(default));
+
+        await AssertDepthMatchesCount(db, store);
+        Assert.Equal(3, await store.GetOplogDepthAsync(default));
+    }
+
+    [Fact]
+    public async Task OplogDepth_PartialAck_MatchesCount()
+    {
+        using var db = await NewOrdersDb();
+        await WriteRows(db, 9);
+        var store = new OplogStore(db, new ReplicationOptions());
+        var batch = await store.ReadBatchAboveAsync(0, 100, default);
+
+        await store.RecordPeerAckAsync(batch[3].Seq, default);
+
+        await AssertDepthMatchesCount(db, store);
+        Assert.Equal(5, await store.GetOplogDepthAsync(default));
+    }
+
+    [Fact]
+    public async Task OplogDepth_AckPruneThenMoreWrites_MatchesCount()
+    {
+        using var db = await NewOrdersDb();
+        await WriteRows(db, 4);
+        var store = new OplogStore(db, new ReplicationOptions());
+        await store.RecordPeerAckAsync(await store.GetMaxSeqAsync(default), default);
+
+        // The steady state of a converged pair: oplog pruned empty, watermark high, new writes above it.
+        await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (91,'S',1),(92,'S',2)");
+
+        await AssertDepthMatchesCount(db, store);
+        Assert.Equal(2, await store.GetOplogDepthAsync(default));
+    }
+
+    [Fact]
+    public async Task OplogDepth_LargeBacklog_IsConstantTime()
+    {
+        // 200k rows: a COUNT(*) over the backlog runs ~200k VM steps per call (measured: 100,020 at
+        // 50k rows), the seek-based formula runs 58 regardless of size. The loop below costs tens of
+        // milliseconds at seek shape and seconds at scan shape, so the budget separates them by a
+        // wide margin in both directions.
+        using var db = await NewOrdersDb();
+        await db.ExecuteAsync(
+            "INSERT INTO __localdb_oplog (table_name, pk_json, row_json, hlc, node_id, is_tombstone) " +
+            "WITH RECURSIVE c(x) AS (SELECT 1 UNION ALL SELECT x + 1 FROM c WHERE x < 200000) " +
+            "SELECT 'orders', '{\"id\":' || x || '}', '{\"id\":' || x || '}', x << 16, 'n', 0 FROM c");
+        var store = new OplogStore(db, new ReplicationOptions());
+
+        Assert.Equal(200_000, await store.GetOplogDepthAsync(default));
+
+        using var conn = db.CreateConnection();
+        using var cmd = conn.CreateCommand();
+        cmd.CommandText = OplogStore.DepthSql;
+        var sw = System.Diagnostics.Stopwatch.StartNew();
+        for (var i = 0; i < 2_000; i++)
+            Assert.Equal(200_000L, (long)cmd.ExecuteScalar()!);
+        sw.Stop();
+
+        Assert.True(sw.ElapsedMilliseconds < 1_500,
+            $"2000 depth reads over a 200k backlog took {sw.ElapsedMilliseconds} ms — that is scan-shaped, not seek-shaped.");
+    }
+
     [Fact]
     public async Task PeerState_RoundTrips()
     {

ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/RegistrationTests.cs

@@ -128,4 +128,359 @@ public sealed class RegistrationTests : IDisposable
         Assert.Equal(new[] { "id" }, table.PkColumns);
         Assert.Equal(new[] { "id", "sku", "qty" }, table.Columns.Select(c => c.Name).ToArray());
     }
+
+    // ---- Deregistration ------------------------------------------------------------------
+
+    [Fact]
+    public async Task Deregister_DropsTriggers_PrunesOplogAndLedger_LeavesDataAlone()
+    {
+        using var db = Create();
+        await CreateOrders(db);
+        db.RegisterReplicated("orders");
+        await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1,'A',1),(2,'B',2)");
+
+        var removed = db.DeregisterReplicated("orders");
+
+        Assert.True(removed);
+        Assert.Empty(await Triggers(db));
+        Assert.Equal(0, await CountAsync(db, "SELECT COUNT(*) FROM __localdb_oplog WHERE table_name='orders'"));
+        Assert.Equal(0, await CountAsync(db, "SELECT COUNT(*) FROM __localdb_row_version WHERE table_name='orders'"));
+        Assert.False(db.ReplicatedTables.ContainsKey("orders"));
+
+        // The consumer's own data is never touched — this removes CAPTURE, not rows.
+        Assert.Equal(2, await CountAsync(db, "SELECT COUNT(*) FROM orders"));
+    }
+
+    [Fact]
+    public async Task Deregister_StopsCapture()
+    {
+        using var db = Create();
+        await CreateOrders(db);
+        db.RegisterReplicated("orders");
+        db.DeregisterReplicated("orders");
+
+        await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1,'A',1)");
+        await db.ExecuteAsync("UPDATE orders SET qty = 9 WHERE id = 1");
+        await db.ExecuteAsync("DELETE FROM orders WHERE id = 1");
+
+        Assert.Equal(0, await CountAsync(db, "SELECT COUNT(*) FROM __localdb_oplog"));
+        Assert.Equal(0, await CountAsync(db, "SELECT COUNT(*) FROM __localdb_row_version"));
+    }
+
+    [Fact]
+    public async Task Deregister_IsIdempotent_SecondCallReportsNothingToDo()
+    {
+        using var db = Create();
+        await CreateOrders(db);
+        db.RegisterReplicated("orders");
+        await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1,'A',1)");
+
+        Assert.True(db.DeregisterReplicated("orders"));
+        Assert.False(db.DeregisterReplicated("orders"));
+        Assert.False(db.DeregisterReplicated("orders"));
+        Assert.Empty(await Triggers(db));
+    }
+
+    [Fact]
+    public async Task Deregister_NeverRegisteredTable_IsANoOp()
+    {
+        using var db = Create();
+        await CreateOrders(db);
+
+        Assert.False(db.DeregisterReplicated("orders"));
+        Assert.False(db.DeregisterReplicated("ghost"));
+    }
+
+    [Fact]
+    public async Task Deregister_ClearsTriggersLeftByAnEarlierProcess()
+    {
+        // The primary use: a database file registered by an OLDER build, opened by one that has
+        // decided this node does not replicate. Nothing in memory says the table was ever
+        // registered — the evidence is entirely on disk.
+        using (var first = Create())
+        {
+            await CreateOrders(first);
+            first.RegisterReplicated("orders");
+            await first.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1,'A',1)");
+        }
+        SqliteConnection.ClearAllPools();
+
+        using var second = Create();
+        Assert.Equal(3, (await Triggers(second)).Count);
+        Assert.False(second.ReplicatedTables.ContainsKey("orders"));
+
+        Assert.True(second.DeregisterReplicated("orders"));
+
+        Assert.Empty(await Triggers(second));
+        Assert.Equal(0, await CountAsync(second, "SELECT COUNT(*) FROM __localdb_oplog"));
+    }
+
+    [Fact]
+    public async Task Deregister_LeavesOtherTablesCapturing()
+    {
+        using var db = Create();
+        await CreateOrders(db);
+        await db.ExecuteAsync("CREATE TABLE widgets (id INTEGER PRIMARY KEY, name TEXT)");
+        db.RegisterReplicated("orders");
+        db.RegisterReplicated("widgets");
+        await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1,'A',1)");
+        await db.ExecuteAsync("INSERT INTO widgets (id, name) VALUES (1,'W')");
+
+        db.DeregisterReplicated("orders");
+        await db.ExecuteAsync("INSERT INTO widgets (id, name) VALUES (2,'W2')");
+
+        Assert.Equal(0, await CountAsync(db, "SELECT COUNT(*) FROM __localdb_oplog WHERE table_name='orders'"));
+        Assert.Equal(2, await CountAsync(db, "SELECT COUNT(*) FROM __localdb_oplog WHERE table_name='widgets'"));
+        Assert.True(db.ReplicatedTables.ContainsKey("widgets"));
+    }
+
+    [Fact]
+    public async Task Deregister_ThenRegisterWithBaseline_RestoresTheLedger()
+    {
+        // Pruning the ledger is only safe because re-registration can re-baseline; this is that
+        // round trip.
+        using var db = Create();
+        await CreateOrders(db);
+        db.RegisterReplicated("orders");
+        await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1,'A',1),(2,'B',2)");
+
+        db.DeregisterReplicated("orders");
+        Assert.Equal(0, await CountAsync(db, "SELECT COUNT(*) FROM __localdb_row_version"));
+
+        db.RegisterReplicated("orders", baselineExistingRows: true);
+
+        Assert.Equal(2, await CountAsync(db, "SELECT COUNT(*) FROM __localdb_row_version WHERE table_name='orders'"));
+    }
+
+    // ---- Baselining ----------------------------------------------------------------------
+
+    [Fact]
+    public async Task Register_WithoutBaseline_LeavesPreexistingRowsUnledgered()
+    {
+        // The documented CDC contract, unchanged: opting in is what changes behaviour, not upgrading.
+        using var db = Create();
+        await CreateOrders(db);
+        await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1,'A',1)");
+
+        db.RegisterReplicated("orders");
+
+        Assert.Equal(0, await CountAsync(db, "SELECT COUNT(*) FROM __localdb_row_version"));
+        Assert.False(await NeedsSnapshot(db));
+    }
+
+    [Fact]
+    public async Task Register_WithBaseline_SeedsAtHlcZero_AndFlagsSnapshot()
+    {
+        using var db = Create();
+        await CreateOrders(db);
+        await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1,'A',1),(2,'B',2)");
+
+        db.RegisterReplicated("orders", baselineExistingRows: true);
+
+        var seeded = await db.QueryAsync(
+            "SELECT pk_json, hlc, node_id, is_tombstone FROM __localdb_row_version ORDER BY pk_json",
+            r => (Pk: r.GetString(0), Hlc: r.GetInt64(1), Node: r.GetString(2), Tomb: r.GetInt64(3)));
+
+        Assert.Equal(2, seeded.Count);
+        Assert.Equal(["{\"id\":1}", "{\"id\":2}"], seeded.Select(s => s.Pk));
+        // HLC 0 is the floor: below every genuine stamp (physical ms << 16), so it can never beat one.
+        Assert.All(seeded, s => Assert.Equal(0, s.Hlc));
+        Assert.All(seeded, s => Assert.Equal(db.NodeId, s.Node));
+        Assert.All(seeded, s => Assert.Equal(0, s.Tomb));
+
+        // Without the flag the seeding would be inert: no gap exists at the handshake, so the peer
+        // would be told it needs no snapshot and the baselined rows would never travel.
+        Assert.True(await NeedsSnapshot(db));
+    }
+
+    [Fact]
+    public async Task Register_WithBaseline_IsIdempotent_AndDoesNotReflagSnapshot()
+    {
+        using var db = Create();
+        await CreateOrders(db);
+        await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1,'A',1)");
+
+        db.RegisterReplicated("orders", baselineExistingRows: true);
+        await db.ExecuteAsync("UPDATE __localdb_peer_state SET needs_snapshot = 0 WHERE id = 1");
+
+        db.RegisterReplicated("orders", baselineExistingRows: true);
+
+        Assert.Equal(1, await CountAsync(db, "SELECT COUNT(*) FROM __localdb_row_version"));
+        // Nothing new was seeded, so no resync is owed: passing true every boot is free after the first.
+        Assert.False(await NeedsSnapshot(db));
+    }
+
+    [Fact]
+    public async Task Register_WithBaseline_NeverOverwritesACapturedVersion()
+    {
+        // The LWW-safety core: a row with a real, captured version must keep it. A floor that
+        // clobbered genuine versions would make every baselined node look newer than its peer.
+        using var db = Create();
+        await CreateOrders(db);
+        db.RegisterReplicated("orders");
+        await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1,'A',1)");
+        var captured = (await db.QueryAsync(
+            "SELECT hlc FROM __localdb_row_version WHERE pk_json='{\"id\":1}'", r => r.GetInt64(0)))[0];
+        Assert.True(captured > 0);
+
+        // A second, pre-existing row that capture never saw (inserted with triggers removed).
+        db.DeregisterReplicated("orders");
+        await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (2,'B',2)");
+        await db.ExecuteAsync(
+            "INSERT INTO __localdb_row_version (table_name, pk_json, hlc, node_id, is_tombstone) " +
+            "VALUES ('orders', '{\"id\":1}', @h, 'genuine', 0)", new { h = captured });
+
+        db.RegisterReplicated("orders", baselineExistingRows: true);
+
+        var versions = await db.QueryAsync(
+            "SELECT pk_json, hlc, node_id FROM __localdb_row_version ORDER BY pk_json",
+            r => (Pk: r.GetString(0), Hlc: r.GetInt64(1), Node: r.GetString(2)));
+        Assert.Equal(2, versions.Count);
+        Assert.Equal(captured, versions[0].Hlc);
+        Assert.Equal("genuine", versions[0].Node);
+        Assert.Equal(0, versions[1].Hlc);
+    }
+
+    [Fact]
+    public async Task Register_WithBaseline_SeedsPkJsonIdenticalToWhatCaptureWrites()
+    {
+        // A seeded key that differs by so much as a space from the capture triggers' json_object
+        // output would be a DIFFERENT ledger row, and the baseline would silently never merge.
+        using var db = Create();
+        await db.ExecuteAsync("CREATE TABLE kv (a INTEGER, b TEXT, v TEXT, PRIMARY KEY (a, b))");
+        await db.ExecuteAsync("INSERT INTO kv (a, b, v) VALUES (1, 'x', 'seeded')");
+
+        db.RegisterReplicated("kv", baselineExistingRows: true);
+        await db.ExecuteAsync("UPDATE kv SET v = 'captured' WHERE a = 1 AND b = 'x'");
+
+        // One ledger row, not two: the capture upsert landed on the seeded key.
+        var rows = await db.QueryAsync(
+            "SELECT pk_json, hlc FROM __localdb_row_version", r => (Pk: r.GetString(0), Hlc: r.GetInt64(1)));
+        Assert.Single(rows);
+        Assert.Equal("{\"a\":1,\"b\":\"x\"}", rows[0].Pk);
+        Assert.True(rows[0].Hlc > 0, "the genuine capture must have replaced the floor");
+    }
+
+    [Fact]
+    public async Task Register_WithBaseline_FloorLosesToEveryGenuineRemoteWrite()
+    {
+        // Proved through the real LWW path rather than by reasoning about the comparison: an
+        // absurdly small remote HLC of 1 still beats the baseline floor of 0.
+        using var db = Create();
+        await CreateOrders(db);
+        await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1,'BASELINE',1)");
+        db.RegisterReplicated("orders", baselineExistingRows: true);
+
+        var result = await new Replication.Internal.LwwApplier(db).ApplyBatchAsync(
+            [new Replication.Internal.OplogEntryRecord(
+                1, "orders", "{\"id\":1}", "{\"id\":1,\"sku\":\"REMOTE\",\"qty\":9}", 1, "peer", false)],
+            default);
+
+        Assert.Equal(1, result.Applied);
+        var sku = await db.QueryAsync("SELECT sku FROM orders WHERE id = 1", r => r.GetString(0));
+        Assert.Equal("REMOTE", sku[0]);
+    }
+
+    [Fact]
+    public async Task Register_WithBaseline_TieBetweenTwoBaselinedNodes_ResolvesByNodeIdOrdinal()
+    {
+        // Two nodes that baselined the same key both hold HLC 0 — genuinely unordered history. The
+        // node-id tie-break is what keeps that convergent instead of oscillating.
+        using var db = Create();
+        await CreateOrders(db);
+        await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1,'MINE',1)");
+        db.RegisterReplicated("orders", baselineExistingRows: true);
+
+        var applier = new Replication.Internal.LwwApplier(db);
+        var lower = await applier.ApplyBatchAsync(
+            [new Replication.Internal.OplogEntryRecord(
+                1, "orders", "{\"id\":1}", "{\"id\":1,\"sku\":\"LOWER\",\"qty\":9}", 0, "!" + db.NodeId, false)],
+            default);
+        Assert.Equal(1, lower.Discarded);
+
+        var higher = await applier.ApplyBatchAsync(
+            [new Replication.Internal.OplogEntryRecord(
+                2, "orders", "{\"id\":1}", "{\"id\":1,\"sku\":\"HIGHER\",\"qty\":9}", 0, db.NodeId + "z", false)],
+            default);
+        Assert.Equal(1, higher.Applied);
+
+        var sku = await db.QueryAsync("SELECT sku FROM orders WHERE id = 1", r => r.GetString(0));
+        Assert.Equal("HIGHER", sku[0]);
+    }
+
+    [Fact]
+    public async Task Register_WithBaseline_EmptyTable_SeedsNothing_FlagsNothing()
+    {
+        using var db = Create();
+        await CreateOrders(db);
+
+        db.RegisterReplicated("orders", baselineExistingRows: true);
+
+        Assert.Equal(0, await CountAsync(db, "SELECT COUNT(*) FROM __localdb_row_version"));
+        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];
+
+    private static async Task<bool> NeedsSnapshot(SqliteLocalDb db) =>
+        (await db.QueryAsync(
+            "SELECT needs_snapshot FROM __localdb_peer_state WHERE id = 1", r => r.GetInt64(0)))[0] != 0;
 }

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