perf(localdb): dereg API, O(1) backlog depth, byte-budget batching, prepare-once apply (0.2.0)

WP3.3 of the ScadaBridge architectural-review remediation. Six changes, all 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 when a file is opened. DeregisterReplicated(table). There was no way to stop capturing once a file had been registered, so a node that does not replicate kept paying three triggers inside every write transaction, appending to an oplog nothing would ever drain. Drops the triggers and prunes the table's oplog + row-version rows in one transaction; idempotent, and deliberately usable on a table this process never registered, since stale triggers left by an EARLIER build are the primary case. Deregistration must be symmetric across a pair, and already is enforced: the handshake compares registered-table digests fail-closed, so a node that drops a table its peer still replicates stops syncing loudly rather than diverging. RegisterReplicated(table, baselineExistingRows: true). Closes the late-opt-in hole: the snapshot streamer pages from __localdb_row_version, which only triggers populate, so a node that started replicating LATE never sent its pre-existing rows — silently, forever. Seeding alone would not have fixed it, because a node whose oplog starts at seq 1 presents no gap at the handshake and is told it needs no snapshot; the seed therefore also flags needs_snapshot, and the existing resync machinery carries the rows. The floor is HLC 0 stamped with this node's id: every genuine HLC is a UTC millisecond shifted left 16 bits and so is strictly greater, which makes a baselined row lose every comparison against a real remote write and win only where the peer has no version of that key at all. Two nodes baselining the same key both hold 0 and resolve by the node-id tie-break — convergent, and nothing better is available for two rows of unordered antiquity. Opt-in, so the documented CDC contract is unchanged unless asked. Pruning the ledger on deregister is safe precisely because re-registration can re-baseline; the two APIs are a pair. O(1) backlog depth. GetOplogDepth ran COUNT(*) over the entire unacked backlog, once a minute from maintenance AND on every metrics scrape — a full scan of a million rows exactly when a node is already behind. Now MAX(seq) - MAX(last_acked_seq, MIN(seq) - 1), floored at 0: three b-tree seeks, measured at 58 VM steps against 100,020 for the count on a 50k backlog. Exact, because seq is AUTOINCREMENT and every prune path removes a prefix (ack pruning by watermark, the row cap by count, the age cap by an HLC cutoff that rises with seq). The MIN clamp is what keeps it exact after a cap prune has moved the horizon above a stalled watermark — without it the caps would re-flag on every tick forever. Byte-budget batching. MaxBatchSize was the only cap on an outbound message, which made it a proxy for a byte budget that consumers had to tune against their widest row: ScadaBridge pinned it to 16 because 70 KB config_json times the 500 default is ~35 MB against a 4 MB gRPC limit. MaxBatchBytes (default 2 MB) now bounds delta and snapshot messages by summed serialized size, with the row count demoted to a secondary cap. A row larger than the budget is sent alone rather than stalling the stream on it. Since more, smaller messages would otherwise mean more apply transactions and more acks, inbound delta batches and acks ALREADY queued are coalesced — one transaction and one ack per burst, bounded by the sender's own caps, and a no-op under a trickle. LwwApplier prepares each statement once. It went through the generic transaction façade, which builds a fresh SqliteCommand per call: three SQL parses per applied row, on the path that drains a reconnect backlog. Statements are now held for the batch and rebound. Dropped __localdb_oplog_hlc. 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. 184/184 tests pass (was 149). New coverage: schema upgrade drops the index and preserves identity/watermarks/data, deregistration idempotency and trigger absence and cross-process cleanup, baseline seeding at the floor with LWW safety proved through the real applier, depth asserted equal to the counting query across every prune scenario plus a constant-time assertion over a 200k backlog, byte-budget chunking including the one-giant-row edge on both the delta and snapshot paths, prepare-once statement counting, and coalescing that must not swallow the message that ends its run. Version 0.2.0.
2026-08-14 22:08:43 -04:00
parent 92246c9734
commit 9377fa1678
17 changed files with 1782 additions and 123 deletions
@@ -5,7 +5,7 @@
    <Nullable>enable</Nullable>
    <ImplicitUsings>enable</ImplicitUsings>
    <LangVersion>latest</LangVersion>
-    <Version>0.1.3</Version>
+    <Version>0.2.0</Version>
    <ManagePackageVersionsCentrally>true</ManagePackageVersionsCentrally>
    <TreatWarningsAsErrors>true</TreatWarningsAsErrors>
    <!-- NU1510 (package-pruning advice): the Replication and Tests projects carry a
@@ -10,16 +10,38 @@ 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.0`)
 | 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.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 +129,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. |
@@ -161,8 +184,13 @@ 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
+  and are not snapshotted; only changes after registration are replicated. **Opt out with
-  pre-existing rows requires re-writing them.
+  `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.
 - **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).
@@ -198,6 +226,16 @@ 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.
 - **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 +246,14 @@ 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
+184 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
+computation, schema-version upgrade, deregistration idempotency, baseline-floor LWW
-(bidirectional convergence, delete/update races, offline-accumulate → reconnect,
+safety, O(1) depth vs a seeded backlog, byte-budget chunking), integration over two real
-prune → snapshot resync), plus seeded randomized property tests (random interleaved op
+SQLite DBs on an in-memory (and real) gRPC pipe (bidirectional convergence, delete/update
-sequences → both DBs identical after quiesce). Family conventions apply
+races, offline-accumulate → reconnect, prune → snapshot resync), plus seeded randomized
-(`TreatWarningsAsErrors`).
+property tests (random interleaved op sequences → both DBs identical after quiesce).
 Family conventions apply (`TreatWarningsAsErrors`).
 Build / test from this directory:
@@ -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 using var tx = await db.BeginPreparedTransactionAsync(ct);
-        await tx.ExecuteAsync("UPDATE __localdb_applying SET applying = 1 WHERE id = 1", null, 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(
+            var local = await sql.ReadLocalVersionAsync(entry.TableName, entry.PkJson, ct);
                "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 remoteWins = local.Count == 0
+            var remoteWins = local is null
-                || entry.Hlc > local[0].Hlc
+                || entry.Hlc > local.Value.Hlc
-                || (entry.Hlc == local[0].Hlc && string.CompareOrdinal(entry.NodeId, local[0].NodeId) > 0);
+                || (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(
+                await sql.WriteRowVersionAsync(entry, utcNow, ct);
                    "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);
                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(
+        await sql.AdvancePeerStateAsync(maxSeq, maxHlc, utcNow, ct);
-            "UPDATE __localdb_peer_state SET " +
+        await sql.SetApplyingAsync(0, ct);
-            "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",
+        LastBatchStatementCount = sql.StatementCount;
            new { maxSeq, maxHlc, utc = utcNow }, ct);
        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(
+    /// <summary>
-        ILocalDbTransaction tx, OplogEntryRecord entry, string error, string utcNow, CancellationToken ct) =>
+    /// The statement set one apply-batch runs on: each distinct statement is built and prepared on
-        tx.ExecuteAsync(
+    /// first use, then rebound and re-executed per row. Statements whose SQL depends on the table
-            "INSERT INTO __localdb_dead_letter " +
+    /// (delete / upsert) are cached per table; the rest are shared across all of them.
-            "(received_utc, table_name, pk_json, row_json, hlc, node_id, error) " +
+    /// </summary>
-            "VALUES (@utc, @t, @pk, @row, @hlc, @node, @err)",
+    /// <remarks>
-            new
+    /// 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
-                utc = utcNow,
+    /// and only recompiles when the text changes. <see cref="SqliteCommand.Prepare"/> is called
-                t = entry.TableName,
+    /// explicitly anyway, so the intent is stated rather than inferred from an implementation detail.
-                pk = entry.PkJson,
+    /// </remarks>
-                row = entry.RowJson,
+    private sealed class PreparedStatements(SqliteConnection connection, SqliteTransaction transaction) : IDisposable
                hlc = entry.Hlc,
                node = entry.NodeId,
                err = error,
            }, ct);
    private static (string Delete, string Upsert) BuildTableSql(ReplicatedTable table)
    {
-        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)})"));
+        private SqliteCommand? _applying;
-        var delete = $"DELETE FROM {name} WHERE {pkMatch}";
+        private SqliteCommand? _lookup;
        private SqliteCommand? _rowVersion;
        private SqliteCommand? _deadLetter;
        private SqliteCommand? _peerState;
-        var columns = string.Join(", ", table.Columns.Select(c => Ident(c.Name)));
+        /// <summary>How many distinct statements this set has prepared.</summary>
-        var values = string.Join(", ", table.Columns.Select(c => $"json_extract(@row, {JsonPath(c.Name)})"));
+        public int StatementCount => _owned.Count;
        var upsert = $"INSERT OR REPLACE INTO {name} ({columns}) VALUES ({values})";
-        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("\"", "\\\"") + "\"'";
 }
@@ -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(
+        var rows = await db.QueryAsync(DepthSql, static r => r.GetInt64(0), null, ct);
            "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);
        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 =
+        cmd.CommandText = DepthSql;
            "SELECT COUNT(*) FROM __localdb_oplog WHERE seq > (SELECT last_acked_seq FROM __localdb_peer_state WHERE id = 1)";
        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(
+        // Same O(1) expression as GetOplogDepthAsync — the caps check is the per-minute reader that
-            "SELECT COUNT(*) FROM __localdb_oplog WHERE seq > (SELECT last_acked_seq FROM __localdb_peer_state WHERE id = 1)",
+        // made the old COUNT(*) expensive in the first place.
-            static r => r.GetInt64(0), null, ct))[0];
+        var depth = (await tx.QueryAsync(DepthSql, static r => r.GetInt64(0), null, ct))[0];
        var overRows = depth > options.MaxOplogRows;
        var oldest = await tx.QueryAsync(
@@ -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:
+    // (data written BEFORE the table was registered for replication) are NOT snapshotted: they were
-    // they were never captured, consistent with CDC semantics — a full re-registration is required to
+    // never captured, consistent with CDC semantics.
-    // baseline them.
+    //
    // 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);
@@ -246,6 +246,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 +265,30 @@ internal sealed class SyncSession
            var batch = await _store.ReadBatchAboveAsync(_sentThruSeq, _options.MaxBatchSize, ct);
            if (batch.Count > 0)
            {
-                var delta = new DeltaBatch();
+                var index = 0;
-                foreach (var entry in batch)
+                while (index < batch.Count)
-                    delta.Entries.Add(ToProto(entry));
+                {
-                // Blocking here on a full data lane IS the pump's backpressure.
+                    var delta = new DeltaBatch();
-                await dataLane.WriteAsync(new SyncMessage { DeltaBatch = delta }, ct);
+                    var bytes = 0L;
-                _sentThruSeq = batch[^1].Seq;
+                    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 +296,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;
+                        lookahead = await HandleCoalescedAcksAsync(duplex, msg.DeltaAck, ct);
                        InterlockedMax(ref _peerAckedSeq, acked);
                        await _store.RecordPeerAckAsync(acked, ct);
                        break;
                    case SyncMessage.MsgOneofCase.SnapshotBegin:
                        if (SnapshotApplier is null)
@@ -314,19 +363,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);
@@ -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;
@@ -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)
@@ -42,6 +42,83 @@ 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>
    /// </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>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; }
 }
@@ -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();
@@ -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,105 @@ 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();
            }
            // 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>
    /// </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 +443,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();
@@ -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(BuildUninstallScript(table.Name));
            sb.Append("DROP TRIGGER IF EXISTS ").Append(Ident(TriggerName(table.Name, suffix))).Append(";\n");
        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("'", "''") + "'";
 }
@@ -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);
    }
 }
@@ -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]
@@ -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()
    {
@@ -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()
    {
@@ -128,4 +128,302 @@ 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));
    }
    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;
 }