feat(localdb): per-table needs_snapshot (0.3.0, on-disk schema v3)

Baselining one table used to set a single per-DATABASE needs_snapshot bit, and a snapshot streams every registered table in full — so opting one table into replication late re-sent all of them, in both directions at once on a pair enabling replication late. On a database holding a high-volume table (an event log at its retention cap) that is the largest transfer this library performs. 0.2.1 documented the cost and deferred the fix as needing an on-disk change. Schema v3 adds __localdb_snapshot_state (table_name PK, needs_snapshot), created by the shared DDL with IF NOT EXISTS so an older file gains it on open; the v2->v3 transition is a version stamp only. RegisterReplicated(..., baselineExistingRows: true) now flags that table alone. The database-wide flag survives unchanged for the paths that genuinely reach every table — a cap prune or an impossible peer ack loses delta continuity for the oplog as a whole. DeregisterReplicated drops a table's pending flag with its ledger rows, since nothing would ever stream (and so clear) it afterwards. Wire compatibility is additive and negotiated by CAPABILITY, not version: lib_schema_version stays 1 because ValidateHandshake compares it fail-closed for equality, so bumping it would stop 0.1.x/0.2.x pairs syncing outright. Handshake gains supports_partial_snapshot (field 5) and SnapshotBegin gains a tables scope (field 2) where EMPTY means all tables — precisely what a pre-0.3.0 sender emits and what a pre-0.3.0 receiver assumes. A peer that does not advertise the capability is served a full snapshot: a scoped one would make it advance last_applied_remote_seq to the snapshot's as-of and silently discard every delta below it for the tables left out. For the same reason a partial snapshot moves no watermark on either side. The receiver keeps last_applied_remote_seq (carrying only the HLC across, which is MAX-guarded), and the sender keeps _sentThruSeq, so the deltas the snapshot overlaps are still pumped and merged by LWW. Crash safety: snapshot debt now clears on the receiver's new SnapshotAck rather than on enqueue, so an ungraceful exit part-way through re-streams the same scope on reconnect. The ack rides the unbounded CONTROL lane like DeltaAck, so it composes with 0.2.1's bounded 64-message inbox and its out-of-handshake snapshot. Against a pre-0.3.0 peer the 0.2.x clear-on-enqueue behaviour is kept — there is no ack to wait for — with the handshake's horizon gap check as the net. Tests 199/199 (was 188): single-table baseline streams only that table, mixed- version interop (peer without the capability gets a full snapshot), debt standing until SnapshotAck, partial snapshot not advancing the receiver's watermark, no SnapshotAck emitted to an old peer, end-to-end over two real databases, plus v2->v3 upgrade and per-table flag isolation. README, ILocalDb remarks (which named this as the follow-up) and the umbrella CLAUDE.md index updated.
2026-08-15 03:34:54 -04:00
parent f627cee9c0
commit cea3930b94
16 changed files with 791 additions and 96 deletions
@@ -5,7 +5,7 @@
    <Nullable>enable</Nullable>
    <ImplicitUsings>enable</ImplicitUsings>
    <LangVersion>latest</LangVersion>
-    <Version>0.2.1</Version>
+    <Version>0.3.0</Version>
    <ManagePackageVersionsCentrally>true</ManagePackageVersionsCentrally>
    <TreatWarningsAsErrors>true</TreatWarningsAsErrors>
    <!-- NU1510 (package-pruning advice): the Replication and Tests projects carry a
@@ -10,7 +10,7 @@ Change capture uses SQLite triggers writing a single HLC-stamped oplog inside th
 consumer's own write transaction (crash-consistent, non-bypassable). Convergence is
 last-writer-wins over a hybrid logical clock. All timestamps are UTC.
-## Packages (all `0.2.1`)
+## Packages (all `0.3.0`)
 | Package | Contents |
 |---|---|
@@ -20,6 +20,30 @@ last-writer-wins over a hybrid logical clock. All timestamps are UTC.
 > Status: **published to the Gitea feed; adopted by ScadaBridge (Phases 1+2) and OtOpcUa (Phase 1).**
 >
 > - `0.3.0` (2026-08-15) — **the snapshot-resync flag is per-table.** On-disk bookkeeping schema
 >   goes to **v3** (new `__localdb_snapshot_state`, upgraded in place on open); wire-compatible with
 >   `0.1.x`/`0.2.x` peers, and `lib_schema_version` deliberately stays `1` because it is compared
 >   fail-closed for equality.
 >   - **`RegisterReplicated(table, baselineExistingRows: true)` now flags a resync for THAT TABLE.**
 >     Through `0.2.x` the flag was one bit for the whole database, so baselining one table
 >     re-streamed every registered table in full — and on a pair enabling replication late, both
 >     nodes did that to each other at once. A snapshot now streams only the flagged tables.
 >   - **The database-wide flag survives, unchanged, for the cases that genuinely need it**: an oplog
 >     cap prune or an impossible peer ack loses deltas for the oplog as a whole, so nothing short of
 >     every table is sound. What moved is only where a *baseline* records its debt.
 >   - **Additive wire extension, negotiated by capability, not by version.** `Handshake` gains
 >     `supports_partial_snapshot`; `SnapshotBegin` gains a `tables` scope where **empty means "all
 >     tables"** — which is precisely what a pre-`0.3.0` sender emits and what a pre-`0.3.0` receiver
 >     assumes. A peer that does not advertise the capability is served a **full** snapshot, because a
 >     scoped one would make it advance `last_applied_remote_seq` past deltas it never received.
 >   - **A partial snapshot moves no watermark on either side.** The receiver keeps
 >     `last_applied_remote_seq` (it is one number for the whole peer oplog, and the untouched tables'
 >     deltas still sit below the snapshot's as-of); the sender keeps pumping from where it was. The
 >     overlap re-delivers rows the snapshot already carried, which LWW absorbs.
 >   - **Snapshot debt now clears on acknowledgement, not on enqueue** (new `SnapshotAck`, exchanged
 >     only between `0.3.0` peers). A crash part-way through a snapshot re-streams the same scope on
 >     reconnect instead of losing it. Against a pre-`0.3.0` peer the `0.2.x` clear-on-enqueue
 >     behaviour is kept — there is no ack to wait for.
 > - `0.2.1` (2026-08-14) — two adversarial-review findings against `0.2.0`. Wire-compatible with
 >   `0.1.x`/`0.2.0` peers; no proto change, no `lib_schema_version` bump, no on-disk schema change.
 >   - **`DeregisterReplicated` no longer leaves the HLC without a crash anchor.** Clock recovery on
@@ -177,10 +201,15 @@ All options are validated at startup (`ZB.MOM.WW.Configuration` validators).
  delta/snapshot batches), so receiving never blocks on the ability to ack.
 - **Resync** — when a peer's watermark falls below the pruned oplog horizon, it is sent a
  full snapshot of the registered tables (streamed, keyset-paged) and deltas resume above
-  the snapshot's as-of seq. **Each node decides from its OWN state** whether it owes the
+  the snapshot's as-of seq. When instead a single table was *baselined*, only that table is
-  peer a snapshot. The snapshot is streamed *after* the handshake, alongside a running
+  streamed and **no watermark moves on either side** — a partial snapshot speaks for nothing
-  receive loop, so a node sending one still applies what the peer sends it — a mutual
+  outside its own tables, so the deltas it overlaps are still pumped and merged by LWW.
-  snapshot exchanges under flow control instead of being buffered whole in memory.
+  **Each node decides from its OWN state** whether it owes the peer a snapshot, and a peer
  that does not advertise `supports_partial_snapshot` is sent a full one. The snapshot is
  streamed *after* the handshake, alongside a running receive loop, so a node sending one
  still applies what the peer sends it — a mutual snapshot exchanges under flow control
  instead of being buffered whole in memory. Between `0.3.0` peers the sender's flags are
  cleared by the receiver's `SnapshotAck`, so a crash mid-snapshot re-streams it.
 - **Maintenance** — a background service on **both** roles enforces the oplog row/age caps
  and prunes expired tombstones every `MaintenanceInterval`.
@@ -209,17 +238,18 @@ All options are validated at startup (`ZB.MOM.WW.Configuration` validators).
 - **CDC semantics** — rows that existed **before** `RegisterReplicated` are not captured
  and are not snapshotted; only changes after registration are replicated. **Opt out with
  `RegisterReplicated(table, baselineExistingRows: true)`**, which seeds the row-version
-  ledger for existing rows at the baseline floor (HLC `0`) and flags a snapshot resync so
+  ledger for existing rows at the baseline floor (HLC `0`) and flags a snapshot resync
-  they reach the peer. The floor is below every genuine stamp, so a baselined row loses
+  **for that table** so they reach the peer. The floor is below every genuine stamp, so a baselined row loses
  every comparison against a real remote write and wins only where the peer has no version
  of that key at all. Two nodes that baseline the same key both hold HLC `0` and resolve by
  the node-id tie-break — convergent, but arbitrary as to which content survives.
- **The snapshot flag is per-database, not per-table.** Baselining one table flags a resync
+- **A snapshot still ships whole tables.** The flag is per-table as of `0.3.0`, so baselining
-  that streams **every** registered table in full, and on a pair enabling replication late
+  one table streams that table and no other — but *that* table travels in full, and two
-  both nodes do that to each other simultaneously — the largest transfer this library
+  cases still stream **everything**: an oplog cap prune / impossible peer ack (delta
-  performs. Bounded (batch budget out, bounded inbox in), but **purge or trim high-volume
+  continuity is lost for the oplog as a whole), and any peer older than `0.3.0` (which
-  tables such as event logs before enabling replication late**, and expect one slow first
+  cannot be told a snapshot is partial). So **purge or trim high-volume tables such as event
-  sync. A per-table flag needs an on-disk schema change and is not in `0.2.x`.
+  logs before baselining them**, and expect one slow first sync. Bounded either way (batch
  budget out, bounded inbox in).
 - **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).
@@ -256,8 +286,15 @@ provider is wired or the poll threw, so a DB failure never reads as "0 backlog")
 ## Ops notes
- When the oplog row/age caps are exceeded, maintenance prunes to the ceiling and flags
+- When the oplog row/age caps are exceeded, maintenance prunes to the ceiling and flags the
-  `needs_snapshot` (disk safety over delta continuity); the peer then snapshot-resyncs.
+  **database-wide** `needs_snapshot` (disk safety over delta continuity); the peer then gets a
  full snapshot. Per-table debt lives separately in `__localdb_snapshot_state` and is what
  baselining writes.
 - **Upgrading a pair to `0.3.0`** needs no coordination: the two nodes negotiate scoped snapshots
  by capability, so a `0.3.0` node paired with a `0.2.x` one simply keeps sending full snapshots
  until both are upgraded. Rolling the file forward is one-way — a `0.2.x` build reopening a v3
  file leaves `__localdb_snapshot_state` in place and ignores it, so any per-table debt recorded
  by `0.3.0` is invisible to it (roll forward again, or re-baseline).
 - **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
@@ -281,14 +318,16 @@ provider is wired or the poll threw, so a DB failure never reads as "0 backlog")
 ## Testing
-188 tests, all offline / no live dependencies — unit (HLC ordering + restart
+199 tests, all offline / no live dependencies — unit (HLC ordering + restart
 persistence, deregistration's HLC anchor surviving an ungraceful reopen, trigger SQL
 generation, LWW decision table, watermark/pruning, digest computation, schema-version
 upgrade, deregistration idempotency, baseline-floor LWW safety, O(1) depth vs a seeded
 backlog, byte-budget chunking, bounded-inbox backpressure under a slow reader), integration
 over two real SQLite DBs on an in-memory (and real) gRPC pipe (bidirectional convergence,
 delete/update races, offline-accumulate → reconnect, prune → snapshot resync, mutual
-snapshot over a capacity-1 transport), plus seeded randomized property tests (random interleaved op sequences → both DBs identical after quiesce).
+snapshot over a capacity-1 transport, single-table baseline streaming only that table,
 mixed-version interop where a peer without `supports_partial_snapshot` gets a full one,
 snapshot debt surviving until the peer's `SnapshotAck`), plus seeded randomized property tests (random interleaved op sequences → both DBs identical after quiesce).
 Family conventions apply (`TreatWarningsAsErrors`).
 Build / test from this directory:
@@ -16,6 +16,7 @@ message SyncMessage {
    SnapshotBegin snapshot_begin = 5; // sender switches to snapshot mode
    SnapshotBatch snapshot_batch = 6;
    SnapshotComplete snapshot_complete = 7;
    SnapshotAck snapshot_ack = 8;     // 0.3.0+; only sent when the peer advertised support
  }
 }
@@ -24,6 +25,13 @@ message Handshake {
  uint32 lib_schema_version = 2;
  repeated TableDigest tables = 3;    // fail-closed on any mismatch
  int64 last_applied_remote_seq = 4;  // highest of YOUR oplog seqs I have applied
  // 0.3.0+ capability, additive: I understand a snapshot scoped to a SUBSET of the registered
  // tables (SnapshotBegin.tables) and I will answer a completed snapshot with a SnapshotAck.
  // A pre-0.3.0 peer leaves this false — proto3 default — and a 0.3.0 sender then promotes any
  // scoped snapshot to a full one and clears its flags on enqueue, exactly as 0.2.x did. This is
  // why lib_schema_version stays 1: it is compared fail-closed for EQUALITY, so bumping it would
  // stop 0.1.x/0.2.x pairs syncing outright.
  bool supports_partial_snapshot = 5;
 }
 message TableDigest { string table_name = 1; string digest = 2; }
 message HandshakeAck { string node_id = 1; bool snapshot_required = 2; }
@@ -40,7 +48,12 @@ message OplogEntry {
 message DeltaBatch { repeated OplogEntry entries = 1; }
 message DeltaAck { int64 applied_thru_seq = 1; }
-message SnapshotBegin { int64 as_of_seq = 1; }
+// tables: the SCOPE of this snapshot (0.3.0+). EMPTY means every registered table — a full
 // snapshot, which is the only thing a pre-0.3.0 sender ever emits and the only thing a
 // pre-0.3.0 receiver can safely assume. A non-empty list means the snapshot is PARTIAL, so the
 // receiver must NOT advance last_applied_remote_seq to as_of_seq (deltas below it for the tables
 // NOT listed have not been delivered); it merges the rows through LWW and leaves the watermark.
 message SnapshotBegin { int64 as_of_seq = 1; repeated string tables = 2; }
 message SnapshotBatch { string table_name = 1; repeated SnapshotRow rows = 2; }
 message SnapshotRow {
  string pk_json = 1;
@@ -50,3 +63,8 @@ message SnapshotRow {
  bool is_tombstone = 5;
 }
 message SnapshotComplete { int64 as_of_seq = 1; }
 // Sent by the RECEIVER once a snapshot has been durably applied, so the sender clears the
 // per-table snapshot flags only after the peer actually holds the rows (a crash mid-snapshot then
 // re-streams the same scope on reconnect). Only exchanged when both peers advertised
 // supports_partial_snapshot; a pre-0.3.0 pair keeps the 0.2.x clear-on-enqueue behaviour.
 message SnapshotAck { int64 as_of_seq = 1; }
@@ -233,10 +233,61 @@ internal sealed class OplogStore(ILocalDb db, ReplicationOptions options, Func<D
        db.ExecuteAsync(
            "UPDATE __localdb_peer_state SET last_applied_remote_seq = 0, last_seen_hlc = 0 WHERE id = 1", null, ct);
-    /// <summary>Sets or clears the snapshot-resync-required flag.</summary>
+    /// <summary>
    /// Sets or clears the DATABASE-WIDE snapshot-resync flag — the one that means "every registered
    /// table". Set by the oplog-horizon paths (cap prune, impossible peer ack), which invalidate
    /// delta continuity for all tables at once. Per-table flags live in
    /// <c>__localdb_snapshot_state</c>; see <see cref="GetTablesNeedingSnapshotAsync"/>.
    /// </summary>
    public Task SetNeedsSnapshotAsync(bool value, CancellationToken ct = default) =>
        db.ExecuteAsync("UPDATE __localdb_peer_state SET needs_snapshot = @v WHERE id = 1", new { v = value ? 1 : 0 }, ct);
    /// <summary>
    /// The tables individually flagged as owing the peer a snapshot (schema v3), in ordinal name
    /// order. Independent of the database-wide flag: baselining one table flags only that table.
    /// </summary>
    public Task<IReadOnlyList<string>> GetTablesNeedingSnapshotAsync(CancellationToken ct = default) =>
        db.QueryAsync(
            "SELECT table_name FROM __localdb_snapshot_state WHERE needs_snapshot <> 0 ORDER BY table_name",
            static r => r.GetString(0), null, ct);
    /// <summary>
    /// Clears the snapshot debt a completed snapshot discharged: the named tables when
    /// <paramref name="tables"/> is non-empty, otherwise <b>everything</b> — the database-wide flag
    /// and every per-table row, because a full snapshot subsumes each of them.
    /// </summary>
    /// <remarks>
    /// <para>
    /// Rows are deleted rather than zeroed: absence is the resting state, and keeping a row per
    /// table that has ever been baselined would accumulate bookkeeping that means nothing. Runs in
    /// one transaction so a full snapshot's two clears cannot half-apply and leave the pair
    /// re-snapshotting on every reconnect.
    /// </para>
    /// <para>
    /// A scoped clear deliberately does NOT touch the database-wide flag. If the cap enforcer
    /// flagged a whole-database resync while this snapshot was in flight, that debt is real and
    /// unrelated — the snapshot only carried some of the tables — so it must survive. The full
    /// clear may take it down because a full snapshot carries every row state as of its as-of seq,
    /// which is exactly what the pruned deltas would have delivered.
    /// </para>
    /// </remarks>
    public async Task ClearSnapshotDebtAsync(IReadOnlyList<string> tables, CancellationToken ct = default)
    {
        await using var tx = await db.BeginTransactionAsync(ct);
        if (tables.Count == 0)
        {
            await tx.ExecuteAsync("UPDATE __localdb_peer_state SET needs_snapshot = 0 WHERE id = 1", null, ct);
            await tx.ExecuteAsync("DELETE FROM __localdb_snapshot_state", null, ct);
        }
        else
        {
            foreach (var table in tables)
                await tx.ExecuteAsync(
                    "DELETE FROM __localdb_snapshot_state WHERE table_name = @t", new { t = table }, ct);
        }
        await tx.CommitAsync(ct);
    }
    /// <summary>
    /// Records the highest remote seq applied locally plus the peer HLC observed with it; also stamps
    /// <c>last_sync_utc</c>. Both watermarks are MAX-guarded (monotonic): a snapshot completing with a
@@ -19,21 +19,57 @@ internal interface ISnapshotApplier
 }
 /// <summary>
-/// Streams a full snapshot of every registered table to a peer that has fallen behind the pruned
+/// Which tables a snapshot covers. An EMPTY <see cref="Tables"/> is the wire's own encoding of
-/// oplog horizon. Reads the authoritative per-row version from <c>__localdb_row_version</c> joined
+/// "all registered tables" (see <c>SnapshotBegin.tables</c>) and is what every pre-0.3.0 peer both
 /// sends and expects, so it is deliberately the same value here rather than a separate flag.
 /// </summary>
 /// <remarks>
 /// The distinction is not cosmetic: a FULL snapshot carries every row state as of its
 /// <c>as_of_seq</c>, so the receiver may jump its watermark there and the sender may skip the
 /// deltas it covered. A PARTIAL one carries only the listed tables, so neither side may move a
 /// watermark that also speaks for the tables it left out.
 /// </remarks>
 internal sealed record SnapshotScope(IReadOnlyList<string> Tables)
 {
    /// <summary>Every registered table — the only scope a pre-0.3.0 peer can be sent.</summary>
    public static readonly SnapshotScope Full = new([]);
    public bool IsPartial => Tables.Count > 0;
    public bool Includes(string tableName) =>
        Tables.Count == 0 || Tables.Contains(tableName, StringComparer.Ordinal);
 }
 /// <summary>
 /// Streams a snapshot of the tables in the requested <see cref="SnapshotScope"/> — every registered
 /// table for a peer that has fallen behind the pruned oplog horizon, or just the baselined ones for
 /// a scoped resync. Reads the authoritative per-row version from <c>__localdb_row_version</c> joined
 /// to each live table, pages it into <see cref="SnapshotBatch"/> messages through the gated data
 /// lane, and brackets them with SnapshotBegin/SnapshotComplete carrying the max oplog seq as-of.
 /// </summary>
 /// <remarks>
 /// Clearing the snapshot debt is deliberately NOT done here: whether the flags may be cleared at all
 /// depends on the peer's capability (a 0.3.0 peer acknowledges the snapshot, a pre-0.3.0 one cannot),
 /// which is session state. <see cref="SyncSession"/> owns that decision; this type only streams.
 /// </remarks>
 internal sealed class SnapshotStreamer(SqliteLocalDb db, OplogStore store, ReplicationOptions options, ILogger logger)
 {
-    /// <summary>Streams the full snapshot; returns the as-of oplog seq it covers, so the session can skip already-covered tail deltas.</summary>
+    /// <summary>
-    public async Task<long> SendAsync(Func<SyncMessage, CancellationToken, Task> send, CancellationToken ct)
+    /// Streams the snapshot for <paramref name="scope"/>; returns the as-of oplog seq it covers, so
    /// the session can skip already-covered tail deltas (a FULL snapshot only — a partial one speaks
    /// for nothing outside its tables).
    /// </summary>
    public async Task<long> SendAsync(SnapshotScope scope, Func<SyncMessage, CancellationToken, Task> send, CancellationToken ct)
    {
        var asOfSeq = await store.GetMaxSeqAsync(ct);
-        await send(new SyncMessage { SnapshotBegin = new SnapshotBegin { AsOfSeq = asOfSeq } }, ct);
+        var begin = new SnapshotBegin { AsOfSeq = asOfSeq };
        begin.Tables.AddRange(scope.Tables);
        await send(new SyncMessage { SnapshotBegin = begin }, ct);
        var totalRows = 0L;
-        foreach (var table in db.ReplicatedTables.Values.OrderBy(t => t.Name, StringComparer.Ordinal))
+        foreach (var table in db.ReplicatedTables.Values
                     .Where(t => scope.Includes(t.Name))
                     .OrderBy(t => t.Name, StringComparer.Ordinal))
        {
            var sql = BuildPageSql(table);
            // Keyset cursor (pk_json > last seen), not LIMIT/OFFSET: concurrent inserts/prunes
@@ -111,16 +147,9 @@ internal sealed class SnapshotStreamer(SqliteLocalDb db, OplogStore store, Repli
        }
        await send(new SyncMessage { SnapshotComplete = new SnapshotComplete { AsOfSeq = asOfSeq } }, ct);
-        // The sender owed this snapshot (its own needs_snapshot); clearing it here — after the whole
+        logger.LogInformation(
-        // snapshot is on the wire — stops it re-streaming on the next reconnect. Narrow window: the
+            "Snapshot sent (as_of_seq {AsOfSeq}, {Rows} rows, scope {Scope}).",
-        // messages are enqueued on the data lane, not yet transport-delivered, so a crash/drop after
+            asOfSeq, totalRows, scope.IsPartial ? string.Join(",", scope.Tables) : "all");
        // this clear could lose the snapshot with the flag already down. The reconnect handshake's
        // gap check (peer watermark below the pruned horizon) re-detects that in every case where the
        // oplog is non-empty; the sole undetectable corner is an EMPTY oplog (nothing to gap-check),
        // where the peer stays behind until new local writes recreate a horizon. Accepted: closing it
        // would need transport-level delivery acks for snapshot frames.
        await store.SetNeedsSnapshotAsync(false, ct);
        logger.LogInformation("Snapshot sent (as_of_seq {AsOfSeq}, {Rows} rows).", asOfSeq, totalRows);
        return asOfSeq;
    }
@@ -181,6 +210,7 @@ internal sealed class SnapshotApplier : ISnapshotApplier
    private long _asOfSeq;
    private long _rows;
    private long _maxHlc;
    private bool _partial;
    // db is part of the collaborator's canonical shape (mirrors the streamer) even though the merge
    // reaches the database through the applier and store; kept for construction symmetry in the factory.
@@ -197,7 +227,13 @@ internal sealed class SnapshotApplier : ISnapshotApplier
        _asOfSeq = begin.AsOfSeq;
        _rows = 0;
        _maxHlc = 0;
-        _logger.LogInformation("Snapshot receive begun (as_of_seq {AsOfSeq}).", _asOfSeq);
+        // A non-empty table list scopes the snapshot; empty (which is all a pre-0.3.0 sender can
        // emit) means every registered table. Read here rather than on complete because
        // SnapshotComplete does not repeat it — the receive loop guarantees a Begin precedes both.
        _partial = begin.Tables.Count > 0;
        _logger.LogInformation(
            "Snapshot receive begun (as_of_seq {AsOfSeq}, scope {Scope}).",
            _asOfSeq, _partial ? string.Join(",", begin.Tables) : "all");
        return Task.CompletedTask;
    }
@@ -219,9 +255,20 @@ internal sealed class SnapshotApplier : ISnapshotApplier
    public async Task OnCompleteAsync(SnapshotComplete complete, CancellationToken ct)
    {
        // A PARTIAL snapshot may not move the inbound seq watermark. That watermark is one number
        // for the whole peer oplog, and jumping it to as_of_seq would declare every table applied
        // through that point — including the ones this snapshot deliberately left out, whose deltas
        // below as_of_seq the sender would then never resend. Only the HLC is carried across (seq 0
        // is a no-op against the MAX guard), so the clock still cannot regress below what the peer
        // has stamped. The sender makes the mirror-image concession: it does not advance its own
        // sent-thru watermark past a partial snapshot, so those deltas are still pumped normally
        // and merely re-deliver rows the snapshot already carried, which LWW absorbs.
        var seq = _partial ? 0 : complete.AsOfSeq;
        // SetLastAppliedRemoteSeqAsync is MAX-guarded, so a delta applied mid-snapshot with a seq
        // above the snapshot's as-of (or an HLC above what the snapshot carried) is never regressed.
-        await _store.SetLastAppliedRemoteSeqAsync(complete.AsOfSeq, _maxHlc, ct);
+        await _store.SetLastAppliedRemoteSeqAsync(seq, _maxHlc, ct);
-        _logger.LogInformation("Snapshot receive complete (as_of_seq {AsOfSeq}, {Rows} rows).", complete.AsOfSeq, _rows);
+        _logger.LogInformation(
            "Snapshot receive complete (as_of_seq {AsOfSeq}, {Rows} rows, partial {Partial}).",
            complete.AsOfSeq, _rows, _partial);
    }
 }
@@ -47,6 +47,17 @@ internal sealed class SyncSession
    // single receive-loop task, so no synchronization is needed.
    private bool _snapshotReceiving;
    // Whether the peer speaks the 0.3.0 snapshot extensions: scoped snapshots
    // (SnapshotBegin.tables) and the SnapshotAck that answers a completed one. Written once during
    // the handshake — which completes before either loop starts — then read-only.
    private bool _peerSupportsPartialSnapshot;
    // The scope of the snapshot this node is streaming, held until the peer's SnapshotAck discharges
    // it; null on every session that owes none. Assigned on the pump task strictly BEFORE the
    // snapshot's first message is enqueued and read on the receive loop when the ack lands, so the
    // ordering is causal — volatile only to make the cross-thread publication explicit.
    private SnapshotScope? _sentSnapshotScope;
    public SyncSession(
        SqliteLocalDb db, OplogStore store, LwwApplier applier, ReplicationOptions options, ILogger logger,
        Func<DateTimeOffset>? utcNow = null, LocalDbMetrics? metrics = null, SyncStatus? status = null)
@@ -63,11 +74,12 @@ internal sealed class SyncSession
    /// <summary>
    /// Invoked when THIS node owes the peer a snapshot (its own snapshot-required computation, not
-    /// the peer's) before deltas resume. Receives a gated send delegate that enqueues on the bounded
+    /// the peer's) before deltas resume. Receives the negotiated <see cref="SnapshotScope"/> and a
-    /// data lane (never the raw transport); returns the as-of oplog seq the snapshot covers so the
+    /// gated send delegate that enqueues on the bounded data lane (never the raw transport); returns
-    /// pump can skip already-covered tail deltas. Wired to <see cref="SnapshotStreamer.SendAsync"/>.
+    /// the as-of oplog seq the snapshot covers so the pump can skip already-covered tail deltas.
    /// Wired to <see cref="SnapshotStreamer.SendAsync"/>.
    /// </summary>
-    public Func<Func<SyncMessage, CancellationToken, Task>, CancellationToken, Task<long>>? SnapshotSender { get; set; }
+    public Func<SnapshotScope, Func<SyncMessage, CancellationToken, Task>, CancellationToken, Task<long>>? SnapshotSender { get; set; }
    /// <summary>
    /// Consumes an inbound snapshot inline on the receive loop: SnapshotBegin opens the receive,
@@ -99,7 +111,7 @@ internal sealed class SyncSession
        var writer = WriterLoopAsync(duplex, control.Reader, data.Reader, linked.Token);
        try
        {
-            var snapshotRequired = await HandshakeAsync(duplex, control.Writer, linked.Token);
+            var snapshotScope = await HandshakeAsync(duplex, control.Writer, linked.Token);
            // The receive loop starts BEFORE any snapshot this node owes is streamed. That ordering
            // is load-bearing, not incidental: a snapshot sent from inside the handshake left the
@@ -111,7 +123,7 @@ internal sealed class SyncSession
            // ordinary flow control. Interleaving is already safe by design: snapshot rows and
            // deltas both merge through LWW, and the snapshot's watermark write is MAX-guarded.
            var receive = ReceiveLoopAsync(duplex, control.Writer, linked.Token);
-            var pump = SnapshotThenPumpLoopAsync(snapshotRequired, GatedDataSend, data.Writer, linked.Token);
+            var pump = SnapshotThenPumpLoopAsync(snapshotScope, GatedDataSend, data.Writer, linked.Token);
            var loops = new[] { pump, receive, writer };
            // The first COMPLETED task is the primary outcome. Cancel the siblings, observe every
@@ -135,12 +147,12 @@ internal sealed class SyncSession
    }
    /// <summary>
-    /// Runs the two-phase fail-closed handshake and returns whether THIS node owes the peer a full
+    /// Runs the two-phase fail-closed handshake and returns the snapshot THIS node owes the peer —
-    /// snapshot (its own computation from its own state — the peer's answer drives the peer's send
+    /// null for none, otherwise the <see cref="SnapshotScope"/> to stream (its own computation from
-    /// direction, symmetrically). Sending that snapshot is deliberately NOT part of the handshake;
+    /// its own state; the peer's answer drives the peer's send direction, symmetrically). Sending
-    /// see <see cref="RunAsync"/>.
+    /// that snapshot is deliberately NOT part of the handshake; see <see cref="RunAsync"/>.
    /// </summary>
-    private async Task<bool> HandshakeAsync(
+    private async Task<SnapshotScope?> HandshakeAsync(
        SyncDuplex duplex, ChannelWriter<SyncMessage> controlLane, CancellationToken ct)
    {
        var peerState = await _store.GetPeerStateAsync(ct);
@@ -158,6 +170,11 @@ internal sealed class SyncSession
        ValidateHandshake(localHandshake, peerHandshake);
        // Read before anything can act on it. A pre-0.3.0 peer leaves the field at its proto3
        // default, so "false" covers both "explicitly unsupported" and "older build" without a
        // version check — which is the point of advertising a capability rather than a version.
        _peerSupportsPartialSnapshot = peerHandshake.SupportsPartialSnapshot;
        // A different node id on the far end means the peer was replaced, not merely restarted — a
        // rebuilt host, a restored-from-nothing volume. Its oplog numbers from 1 again, so the
        // watermark we hold (in the OLD peer's seq space) would silently swallow its first N
@@ -191,16 +208,16 @@ internal sealed class SyncSession
        _sentThruSeq = peerAppliedThruSeq;
        Interlocked.Exchange(ref _peerAckedSeq, peerAppliedThruSeq);
-        var snapshotRequired = await ComputeSnapshotRequiredAsync(peerHandshake, peerState, ct);
+        var snapshotScope = await ComputeSnapshotScopeAsync(peerHandshake, peerState, ct);
        await controlLane.WriteAsync(
-            new SyncMessage { HandshakeAck = new HandshakeAck { NodeId = _db.NodeId, SnapshotRequired = snapshotRequired } }, ct);
+            new SyncMessage { HandshakeAck = new HandshakeAck { NodeId = _db.NodeId, SnapshotRequired = snapshotScope is not null } }, ct);
        var peerAckMsg = await ReceiveOneAsync(duplex, ct);
        if (peerAckMsg.MsgCase != SyncMessage.MsgOneofCase.HandshakeAck)
            throw new InvalidOperationException(
                $"Replication protocol error: expected a HandshakeAck, got {peerAckMsg.MsgCase}.");
-        return snapshotRequired;
+        return snapshotScope;
    }
    /// <summary>
@@ -216,25 +233,55 @@ internal sealed class SyncSession
    /// than racing the transport.
    /// </remarks>
    private async Task SnapshotThenPumpLoopAsync(
-        bool snapshotRequired, Func<SyncMessage, CancellationToken, Task> dataSend,
+        SnapshotScope? snapshotScope, Func<SyncMessage, CancellationToken, Task> dataSend,
        ChannelWriter<SyncMessage> dataLane, CancellationToken ct)
    {
-        if (snapshotRequired)
+        if (snapshotScope is not null)
        {
            if (SnapshotSender is null)
                throw new NotSupportedException("snapshot required but no snapshot sender configured");
-            var snapshotAsOfSeq = await SnapshotSender(dataSend, ct);
+            // Published BEFORE the first frame is enqueued, so the peer's SnapshotAck can never
            // arrive on the receive loop while this is still null.
            Volatile.Write(ref _sentSnapshotScope, snapshotScope);
            var snapshotAsOfSeq = await SnapshotSender(snapshotScope, dataSend, ct);
            _metrics?.RecordSnapshotSent();
            _status?.SetLastSyncUtc(_utcNow());
-            // The snapshot already carries every row state up to as-of; skipping the pump past it
+            // A FULL snapshot carries every row state up to as-of, so skipping the pump past it
-            // avoids re-sending tail deltas the peer would only discard via LWW.
+            // avoids re-sending tail deltas the peer would only discard via LWW. A PARTIAL one
-            if (snapshotAsOfSeq > _sentThruSeq)
+            // speaks only for its own tables, so the pump must still deliver every delta below
            // as-of — the peer has not moved its watermark either (see SnapshotApplier.OnComplete).
            if (!snapshotScope.IsPartial && snapshotAsOfSeq > _sentThruSeq)
                _sentThruSeq = snapshotAsOfSeq;
            // A peer that cannot acknowledge the snapshot leaves only the 0.2.x option: clear once
            // the frames are on the data lane. The window that opens (enqueued, not yet delivered —
            // a crash there loses the snapshot with the debt already discharged) is why 0.3.0 peers
            // wait for the ack instead. The reconnect handshake's gap check re-detects it whenever
            // the oplog is non-empty; an EMPTY oplog remains the undetectable corner.
            if (!_peerSupportsPartialSnapshot)
                await ClearSentSnapshotDebtAsync(ct);
        }
        await PumpLoopAsync(dataLane, ct);
    }
    /// <summary>
    /// Discharges the snapshot debt this session's outbound snapshot covered, exactly once.
    /// </summary>
    /// <remarks>
    /// Taking the scope out of the field before clearing makes this idempotent, which matters
    /// because two paths reach it — the pre-0.3.0 clear-on-enqueue above and the peer's SnapshotAck
    /// — and a peer that acknowledges anyway (a newer build than its handshake claimed) must not
    /// clear a second time on top of flags a later baseline may already have re-raised.
    /// </remarks>
    private async Task ClearSentSnapshotDebtAsync(CancellationToken ct)
    {
        var scope = Interlocked.Exchange(ref _sentSnapshotScope, null);
        if (scope is null)
            return;
        await _store.ClearSnapshotDebtAsync(scope.Tables, ct);
    }
    // The ONLY caller of duplex.Send. Control (handshake + acks) always preempts data so the peer's
    // inbound processing is never starved of its acks behind a backlog of DeltaBatches.
    private static async Task WriterLoopAsync(
@@ -384,6 +431,20 @@ internal sealed class SyncSession
                        _snapshotReceiving = false;
                        _metrics?.RecordSnapshotReceived();
                        _status?.SetLastSyncUtc(_utcNow());
                        // Acknowledge only to a peer that asked for it. Sent AFTER the applier's
                        // durable watermark write, so the ack means "I hold these rows", which is
                        // what lets the sender drop its per-table flags. On the unbounded control
                        // lane, like DeltaAck: an ack must never queue behind a data backlog.
                        if (_peerSupportsPartialSnapshot)
                            await controlLane.WriteAsync(
                                new SyncMessage { SnapshotAck = new SnapshotAck { AsOfSeq = msg.SnapshotComplete.AsOfSeq } }, ct);
                        break;
                    case SyncMessage.MsgOneofCase.SnapshotAck:
                        // The peer has durably applied the snapshot THIS node streamed, so the
                        // tables it covered no longer owe one. Nothing to reconcile against the
                        // ack's as-of seq: only one snapshot is ever sent per session, so the scope
                        // held in _sentSnapshotScope is unambiguously the one being acknowledged.
                        await ClearSentSnapshotDebtAsync(ct);
                        break;
                    default:
                        throw new InvalidOperationException(
@@ -565,6 +626,9 @@ internal sealed class SyncSession
            NodeId = _db.NodeId,
            LibSchemaVersion = LibSchemaVersion,
            LastAppliedRemoteSeq = peerState.LastAppliedRemoteSeq,
            // Unconditional: this build always understands a scoped snapshot and always answers a
            // completed one. It is a statement about the code, not about configuration or state.
            SupportsPartialSnapshot = true,
        };
        foreach (var table in _db.ReplicatedTables.Values.OrderBy(t => t.Name, StringComparer.Ordinal))
            handshake.Tables.Add(new TableDigest { TableName = table.Name, Digest = table.Digest });
@@ -595,11 +659,63 @@ internal sealed class SyncSession
            "Replication handshake rejected (fail-closed): " + string.Join("; ", parts) + ".");
    }
-    private async Task<bool> ComputeSnapshotRequiredAsync(Handshake peerHandshake, PeerState peerState, CancellationToken ct)
+    /// <summary>
    /// Decides what snapshot this node owes the peer: null for none, <see cref="SnapshotScope.Full"/>
    /// for every registered table, or a scoped snapshot naming just the tables that need one.
    /// </summary>
    /// <remarks>
    /// <para>
    /// The two sources of snapshot debt are genuinely different in reach, which is why they are
    /// stored separately. An <b>oplog-horizon</b> gap — the peer's watermark below the oldest delta
    /// still streamable, which is what a cap prune or an impossible peer ack leaves behind — means
    /// deltas have been lost for the oplog as a whole, so nothing short of every table is sound. A
    /// <b>baseline</b> means one table's pre-existing rows were never captured, and says nothing at
    /// all about the others.
    /// </para>
    /// <para>
    /// A scope that happens to name every registered table is returned as Full rather than as a
    /// list: it covers exactly what a full snapshot covers, so it should also get a full snapshot's
    /// watermark advance on both sides, and it needs no promotion for a pre-0.3.0 peer.
    /// </para>
    /// <para>
    /// Flags naming tables that are not registered are ignored (they can only stream nothing).
    /// <c>DeregisterReplicated</c> drops them with the table's ledger rows, so the surviving way to
    /// hold one is to drop the SQL table without deregistering — which the handshake's fail-closed
    /// digest comparison rejects for its own reasons.
    /// </para>
    /// </remarks>
    private async Task<SnapshotScope?> ComputeSnapshotScopeAsync(Handshake peerHandshake, PeerState peerState, CancellationToken ct)
    {
-        if (peerState.NeedsSnapshot)
+        if (peerState.NeedsSnapshot || await HasHorizonGapAsync(peerHandshake, peerState, ct))
-            return true;
+            return SnapshotScope.Full;
        var flagged = await _store.GetTablesNeedingSnapshotAsync(ct);
        if (flagged.Count == 0)
            return null;
        var scoped = flagged.Where(_db.ReplicatedTables.ContainsKey).ToList();
        if (scoped.Count == 0)
            return null;
        if (scoped.Count >= _db.ReplicatedTables.Count)
            return SnapshotScope.Full;
        // A peer that would read SnapshotBegin.tables as "all tables" and then advance its watermark
        // past deltas this snapshot did not carry gets the full snapshot instead. Wider than asked
        // for, but it is the only shape the older protocol can express correctly — and it is what
        // 0.2.x did for every baseline.
        if (!_peerSupportsPartialSnapshot)
        {
            _logger.LogInformation(
                "Peer {PeerNodeId} does not support scoped snapshots; promoting the snapshot of {Tables} to a full one.",
                PeerNodeId, string.Join(",", scoped));
            return SnapshotScope.Full;
        }
        return new SnapshotScope(scoped);
    }
    private async Task<bool> HasHorizonGapAsync(Handshake peerHandshake, PeerState peerState, CancellationToken ct)
    {
        var minRows = await _db.QueryAsync(
            "SELECT COALESCE(MIN(seq), 0) FROM __localdb_oplog", static r => r.GetInt64(0), null, ct);
        var minSeq = minRows[0];
@@ -6,7 +6,7 @@ namespace ZB.MOM.WW.LocalDb.Replication.Internal;
 /// <summary>
 /// Shared construction of a <see cref="SyncSession"/> and its dependencies (OplogStore + LwwApplier)
 /// from a <see cref="SqliteLocalDb"/> and the bound options. Both the service and client adapters use
-/// this so they build an identical session; snapshot hooks are left null (Task 12 fills them in).
+/// this so they build an identical session.
 /// </summary>
 internal static class SyncSessionFactory
 {
@@ -50,7 +50,8 @@ public interface ILocalDb
    /// <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.
+    /// anything was seeded, a snapshot resync is flagged <b>for this table</b> so the peer actually
    /// receives them.
    /// </param>
    /// <remarks>
    /// <para>
@@ -80,16 +81,29 @@ public interface ILocalDb
    /// and no snapshot is flagged. Passing true on every boot is therefore free after the first.
    /// </para>
    /// <para>
-    /// <b>Cost, and why it is worth planning for.</b> The snapshot flag is a single per-DATABASE
+    /// <b>Cost, and what bounds it (0.3.0).</b> The snapshot flag is <b>per-table</b>
-    /// flag, and a snapshot streams EVERY registered table in full — so baselining one table
+    /// (<c>__localdb_snapshot_state</c>, on-disk schema v3), so baselining one table asks the peer
-    /// re-sends all of them, and a pair that enables replication late has both nodes doing that to
+    /// for that table alone — the tables around it are not re-streamed. Through 0.2.x the flag was a
-    /// each other at once. On a database holding a high-volume table (an event log at its retention
+    /// single per-database bit and baselining anything re-sent everything, in both directions at
-    /// cap, say) that is the largest transfer this library ever performs, in both directions
+    /// once on a pair enabling replication late.
-    /// simultaneously. It is bounded work, not runaway: outbound messages respect
+    /// </para>
-    /// <c>MaxBatchBytes</c>, and the receive side is bounded by the transport's inbox. But the
+    /// <para>
-    /// operational advice is to <b>purge or trim high-volume tables before enabling replication
+    /// Two cases still stream every registered table, by design. A <b>pre-0.3.0 peer</b> cannot be
-    /// late</b>, and to expect one slow first sync rather than a slow steady state. A per-table
+    /// told a snapshot is partial (it would advance its watermark past deltas it never received),
-    /// snapshot flag would narrow this; it needs an on-disk schema change and is not in 0.2.x.
+    /// so a scoped snapshot is promoted to a full one for it. And the <b>oplog-horizon</b> paths —
    /// a cap prune, an impossible peer ack — invalidate delta continuity for every table at once
    /// and still set the database-wide flag. So the operational advice survives, narrowed: baseline
    /// a high-volume table (an event log at its retention cap, say) deliberately, and <b>purge or
    /// trim it first</b>, because that ONE table's rows still travel in full. The work is bounded
    /// either way — outbound messages respect <c>MaxBatchBytes</c>, the receive side is bounded by
    /// the transport's inbox.
    /// </para>
    /// <para>
    /// <b>Crash safety.</b> Between 0.3.0 peers a table's flag is cleared only when the receiver
    /// acknowledges the completed snapshot, so an ungraceful exit part-way through re-streams the
    /// same scope on reconnect rather than losing it. Against a pre-0.3.0 peer (which sends no such
    /// acknowledgement) the 0.2.x behaviour is kept: the flag clears once the snapshot is on the
    /// wire, and the handshake's horizon gap check is the net.
    /// </para>
    /// </remarks>
    ReplicatedTable RegisterReplicated(string tableName, bool baselineExistingRows);
@@ -27,9 +27,20 @@ internal static class LocalDbSchema
    /// <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>
    /// <item><description>
    /// <b>3</b> — adds <c>__localdb_snapshot_state</c>, the PER-TABLE snapshot-resync flag. Until
    /// v3 the only flag was <c>__localdb_peer_state.needs_snapshot</c>, one bit for the whole
    /// database, so baselining a single table re-streamed EVERY registered table in full — in both
    /// directions at once on a pair enabling replication late. The database-wide bit survives with
    /// its original meaning (the oplog-horizon and cap-prune paths genuinely invalidate every
    /// table); the new table narrows the baseline case to the tables that actually need it. The
    /// transition itself is a version stamp only — <see cref="Ddl"/> creates the table with
    /// <c>IF NOT EXISTS</c> on every open, so an older file gains it whatever its stamp says, and
    /// an empty flag table means exactly what it meant before: nothing per-table is owed.
    /// </description></item>
    /// </list>
    /// </remarks>
-    internal const long CurrentSchemaVersion = 2;
+    internal const long CurrentSchemaVersion = 3;
    private const string Ddl = """
        CREATE TABLE IF NOT EXISTS __localdb_meta (
@@ -73,6 +84,14 @@ internal static class LocalDbSchema
          id INTEGER PRIMARY KEY CHECK (id = 1),
          applying INTEGER NOT NULL DEFAULT 0
        );
        -- Per-table snapshot-resync flags (schema v3). A row exists only while that table owes the
        -- peer a snapshot; the whole-database __localdb_peer_state.needs_snapshot bit still means
        -- "every registered table" and is what the cap/horizon paths set.
        -- WITHOUT ROWID: table_name IS the only access path (read as a set, cleared by name).
        CREATE TABLE IF NOT EXISTS __localdb_snapshot_state (
          table_name TEXT NOT NULL PRIMARY KEY,
          needs_snapshot INTEGER NOT NULL DEFAULT 0
        ) WITHOUT ROWID;
        CREATE TABLE IF NOT EXISTS __localdb_dead_letter (
          id INTEGER PRIMARY KEY AUTOINCREMENT,
          received_utc TEXT NOT NULL,
@@ -143,6 +162,12 @@ internal static class LocalDbSchema
        if (current < 2)
            steps.Add("DROP INDEX IF EXISTS __localdb_oplog_hlc;");
        // 2 -> 3: __localdb_snapshot_state. No step: the shared Ddl above already ran
        // CREATE TABLE IF NOT EXISTS for it in this same transaction, and an empty flag table is
        // the correct v3 state for a file that has never baselined a table. Repeating the DDL here
        // would be a second source of truth for the same layout. The stamp still moves, because
        // schema_version records what the file's layout IS, not which statements produced it.
        if (steps.Count > 0)
        {
            using var ddl = connection.CreateCommand();
@@ -276,7 +276,12 @@ internal sealed class SqliteLocalDb : ILocalDb, IDisposable
                prune.Transaction = tx;
                prune.CommandText =
                    "DELETE FROM __localdb_oplog WHERE table_name = $t;" +
-                    "DELETE FROM __localdb_row_version WHERE table_name = $t;";
+                    "DELETE FROM __localdb_row_version WHERE table_name = $t;" +
                    // A pending per-table snapshot flag for a table nothing streams any more would
                    // never clear — the streamer skips unregistered tables, so the flag would sit
                    // in the file forever, and re-registering later would find a stale "already
                    // owed" it did not ask for. Dropped with the ledger it refers to.
                    "DELETE FROM __localdb_snapshot_state WHERE table_name = $t;";
                prune.Parameters.AddWithValue("$t", tableName);
                prunedRows = prune.ExecuteNonQuery();
            }
@@ -308,12 +313,12 @@ internal sealed class SqliteLocalDb : ILocalDb, IDisposable
    /// the same snapshot anyway, via a cap-prune.
    /// </para>
    /// <para>
-    /// <b>The flag is per-DATABASE, not per-table</b>, and a snapshot streams every registered table
+    /// <b>The flag is PER-TABLE</b> (<c>__localdb_snapshot_state</c>, schema v3): baselining one
-    /// in full — so baselining one table re-sends all of them, and both nodes of a late-enabled pair
+    /// table asks for a snapshot of that table alone, not of every registered table. The
-    /// do it to each other at once. Bounded (batch budget outbound, bounded inbox inbound) but
+    /// database-wide <c>__localdb_peer_state.needs_snapshot</c> bit is untouched here and keeps its
-    /// potentially the largest transfer this library performs; see
+    /// original meaning — the oplog-horizon and cap-prune paths invalidate every table's delta
-    /// <see cref="ILocalDb.RegisterReplicated(string, bool)"/> for the operational advice. Narrowing
+    /// continuity at once, and only those paths set it. A peer too old to understand a scoped
-    /// it to a per-table flag needs an on-disk schema change and is deferred.
+    /// snapshot is served a full one instead; the negotiation lives in the replication package.
    /// </para>
    /// </remarks>
    private void BaselineExistingRows(SqliteTransaction tx, ReplicatedTable table)
@@ -339,7 +344,10 @@ internal sealed class SqliteLocalDb : ILocalDb, IDisposable
        using var flag = _master.CreateCommand();
        flag.Transaction = tx;
-        flag.CommandText = "UPDATE __localdb_peer_state SET needs_snapshot = 1 WHERE id = 1";
+        flag.CommandText =
            "INSERT INTO __localdb_snapshot_state (table_name, needs_snapshot) VALUES ($t, 1) " +
            "ON CONFLICT(table_name) DO UPDATE SET needs_snapshot = 1";
        flag.Parameters.AddWithValue("$t", table.Name);
        flag.ExecuteNonQuery();
    }
@@ -250,7 +250,7 @@ public sealed class BatchBudgetTests : IDisposable
        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);
+        await streamer.SendAsync(SnapshotScope.Full, (m, _) => { sent.Add(m); return Task.CompletedTask; }, default);
        var batches = sent.Where(m => m.MsgCase == SyncMessage.MsgOneofCase.SnapshotBatch)
            .Select(m => m.SnapshotBatch).ToList();
@@ -279,7 +279,7 @@ public sealed class BatchBudgetTests : IDisposable
        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);
+        await streamer.SendAsync(SnapshotScope.Full, (m, _) => { sent.Add(m); return Task.CompletedTask; }, default);
        var batches = sent.Where(m => m.MsgCase == SyncMessage.MsgOneofCase.SnapshotBatch)
            .Select(m => m.SnapshotBatch).ToList();
@@ -51,7 +51,8 @@ public sealed class LocalDbSchemaTests : IDisposable
        foreach (var expected in new[]
                 {
                     "__localdb_meta", "__localdb_oplog", "__localdb_peer_state",
-                     "__localdb_row_version", "__localdb_applying", "__localdb_dead_letter"
+                     "__localdb_row_version", "__localdb_applying", "__localdb_dead_letter",
                     "__localdb_snapshot_state",
                 })
            Assert.Contains(expected, tables);
@@ -85,6 +86,44 @@ public sealed class LocalDbSchemaTests : IDisposable
            ScalarLong(conn, "SELECT schema_version FROM __localdb_meta WHERE id=1"));
    }
    [Fact]
    public void EnsureCreated_UpgradesV2File_AddsPerTableSnapshotState()
    {
        using var conn = Open();
        // A file exactly as schema v2 left it: everything except __localdb_snapshot_state, stamped 2.
        LocalDbSchema.EnsureCreated(conn);
        Exec(conn, "DROP TABLE __localdb_snapshot_state;" +
                   "UPDATE __localdb_meta SET schema_version = 2 WHERE id = 1;");
        LocalDbSchema.EnsureCreated(conn);
        Assert.Equal(1, ScalarLong(conn,
            "SELECT count(*) FROM sqlite_master WHERE type='table' AND name='__localdb_snapshot_state'"));
        // Empty is the right v3 state for a file that never baselined anything: no table owes a
        // snapshot, and the database-wide flag it was carrying keeps whatever value it had.
        Assert.Equal(0, ScalarLong(conn, "SELECT COUNT(*) FROM __localdb_snapshot_state"));
        Assert.Equal(LocalDbSchema.CurrentSchemaVersion,
            ScalarLong(conn, "SELECT schema_version FROM __localdb_meta WHERE id=1"));
    }
    [Fact]
    public void EnsureCreated_Upgrade_PreservesTheDatabaseWideSnapshotFlag()
    {
        // A v2 file mid-resync (cap prune flagged a full snapshot, the peer was down) must still owe
        // that snapshot after the upgrade. v3 narrows where NEW debt is recorded; it does not
        // reinterpret debt already on disk.
        using var conn = Open();
        LocalDbSchema.EnsureCreated(conn);
        Exec(conn, "DROP TABLE __localdb_snapshot_state;" +
                   "UPDATE __localdb_meta SET schema_version = 2 WHERE id = 1;" +
                   "UPDATE __localdb_peer_state SET needs_snapshot = 1 WHERE id = 1;");
        LocalDbSchema.EnsureCreated(conn);
        Assert.Equal(1, ScalarLong(conn, "SELECT needs_snapshot FROM __localdb_peer_state WHERE id=1"));
    }
    [Fact]
    public void EnsureCreated_Upgrade_PreservesNodeIdentityAndOplogRows()
    {
@@ -291,6 +291,45 @@ public sealed class RegistrationTests : IDisposable
        // 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));
        // ...and it is flagged for THIS TABLE ONLY. Through 0.2.x this set the one database-wide
        // bit, so baselining `orders` re-streamed every other registered table with it.
        Assert.False(await NeedsFullSnapshot(db));
    }
    [Fact]
    public async Task Register_WithBaseline_FlagsOnlyTheBaselinedTable()
    {
        // The per-table flag's reason to exist: opting one table into replication late must not
        // schedule a full re-stream of its unrelated neighbours.
        using var db = Create();
        await CreateOrders(db);
        await db.ExecuteAsync("CREATE TABLE widgets (id INTEGER PRIMARY KEY, name TEXT)");
        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.RegisterReplicated("widgets");
        db.RegisterReplicated("orders", baselineExistingRows: true);
        Assert.True(await NeedsSnapshot(db, "orders"));
        Assert.False(await NeedsSnapshot(db, "widgets"));
        Assert.False(await NeedsFullSnapshot(db));
    }
    [Fact]
    public async Task Deregister_DropsThePendingSnapshotFlag()
    {
        // A flag naming a table nothing streams any more can never clear itself — the streamer skips
        // unregistered tables — so it would sit in the file and re-appear as unasked-for debt if the
        // table were registered again.
        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);
        Assert.True(await NeedsSnapshot(db));
        db.DeregisterReplicated("orders");
        Assert.False(await NeedsSnapshot(db));
    }
    [Fact]
@@ -301,7 +340,8 @@ public sealed class RegistrationTests : IDisposable
        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");
+        // Stand in for the snapshot having been delivered and acknowledged.
        await db.ExecuteAsync("DELETE FROM __localdb_snapshot_state");
        db.RegisterReplicated("orders", baselineExistingRows: true);
@@ -480,7 +520,17 @@ public sealed class RegistrationTests : IDisposable
    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) =>
+    /// <summary>Whether <paramref name="table"/> individually owes the peer a snapshot (schema v3).</summary>
    private static async Task<bool> NeedsSnapshot(SqliteLocalDb db, string table = "orders") =>
        (await db.QueryAsync(
            "SELECT COUNT(*) FROM __localdb_snapshot_state WHERE table_name = @t AND needs_snapshot <> 0",
            r => r.GetInt64(0), new { t = table }))[0] != 0;
    /// <summary>
    /// The DATABASE-WIDE flag — "every registered table". Baselining must never set it: that is the
    /// whole point of the per-table flags, and setting it would restore the 0.2.x behaviour silently.
    /// </summary>
    private static async Task<bool> NeedsFullSnapshot(SqliteLocalDb db) =>
        (await db.QueryAsync(
            "SELECT needs_snapshot FROM __localdb_peer_state WHERE id = 1", r => r.GetInt64(0)))[0] != 0;
 }
@@ -89,9 +89,19 @@ public sealed class SnapshotResyncTests : IDisposable
        return (duplex, toSession.Writer, fromSession.Reader);
    }
-    private static Handshake MatchingHandshake(SqliteLocalDb sessionDb, string peerNodeId, long lastAppliedRemoteSeq = 0)
+    // supportsPartialSnapshot defaults to FALSE, which is exactly what a 0.1.x/0.2.x peer puts on
    // the wire (the field does not exist in its proto, so proto3 decodes the default). Scripted
    // peers are therefore old peers unless a test says otherwise.
    private static Handshake MatchingHandshake(
        SqliteLocalDb sessionDb, string peerNodeId, long lastAppliedRemoteSeq = 0, bool supportsPartialSnapshot = false)
    {
-        var h = new Handshake { NodeId = peerNodeId, LibSchemaVersion = 1, LastAppliedRemoteSeq = lastAppliedRemoteSeq };
+        var h = new Handshake
        {
            NodeId = peerNodeId,
            LibSchemaVersion = 1,
            LastAppliedRemoteSeq = lastAppliedRemoteSeq,
            SupportsPartialSnapshot = supportsPartialSnapshot,
        };
        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;
@@ -114,14 +124,30 @@ public sealed class SnapshotResyncTests : IDisposable
    // Completes the two-phase handshake against a real session: the peer never requests a snapshot.
    private static async Task DriveHandshakeAsync(
-        SqliteLocalDb sessionDb, ChannelWriter<SyncMessage> toSession, ChannelReader<SyncMessage> fromSession, CancellationToken ct)
+        SqliteLocalDb sessionDb, ChannelWriter<SyncMessage> toSession, ChannelReader<SyncMessage> fromSession,
        CancellationToken ct, bool supportsPartialSnapshot = false)
    {
        await NextOfCaseAsync(fromSession, SyncMessage.MsgOneofCase.Handshake, ct);
-        await toSession.WriteAsync(new SyncMessage { Handshake = MatchingHandshake(sessionDb, "peer") }, ct);
+        await toSession.WriteAsync(
            new SyncMessage { Handshake = MatchingHandshake(sessionDb, "peer", supportsPartialSnapshot: supportsPartialSnapshot) }, ct);
        await NextOfCaseAsync(fromSession, SyncMessage.MsgOneofCase.HandshakeAck, ct);
        await toSession.WriteAsync(new SyncMessage { HandshakeAck = new HandshakeAck { NodeId = "peer", SnapshotRequired = false } }, ct);
    }
    // Reads outbound messages until (and including) the first of the given kind.
    private static async Task<List<SyncMessage>> DrainThroughAsync(
        ChannelReader<SyncMessage> reader, SyncMessage.MsgOneofCase last, CancellationToken ct)
    {
        var all = new List<SyncMessage>();
        while (true)
        {
            var msg = await NextAsync(reader, ct);
            all.Add(msg);
            if (msg.MsgCase == last)
                return all;
        }
    }
    private static async Task SwallowAsync(Task task)
    {
        try { await task; }
@@ -159,6 +185,12 @@ public sealed class SnapshotResyncTests : IDisposable
        return r[0];
    }
    /// <summary>The tables individually flagged as owing the peer a snapshot (schema v3).</summary>
    private static async Task<List<string>> PendingSnapshotTables(SqliteLocalDb db) =>
        (await db.QueryAsync(
            "SELECT table_name FROM __localdb_snapshot_state WHERE needs_snapshot <> 0 ORDER BY table_name",
            x => x.GetString(0))).ToList();
    private static async Task<long> OplogCount(SqliteLocalDb db)
    {
        var r = await db.QueryAsync("SELECT COUNT(*) FROM __localdb_oplog", x => x.GetInt64(0));
@@ -585,6 +617,261 @@ public sealed class SnapshotResyncTests : IDisposable
        await SwallowAsync(run);
    }
    // ---- Per-table snapshot scoping (0.3.0) ----------------------------------------------
    private const string WidgetsSql = "CREATE TABLE widgets (id INTEGER PRIMARY KEY, name TEXT)";
    /// <summary>
    /// A node with two registered tables where exactly one — <c>orders</c> — was baselined.
    /// <c>widgets</c> is registered BEFORE its rows are written, so ordinary capture ledgers them;
    /// <c>orders</c> is registered after, so only baselining can. That asymmetry is what makes the
    /// scope observable: a full snapshot carries both tables' ledgers, a scoped one carries orders
    /// alone — and both tables hold rows either way, so an empty widgets stream cannot be mistaken
    /// for "there was nothing to send".
    /// </summary>
    private async Task<Side> NewBaselinedSideAsync(ReplicationOptions? options = null)
    {
        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(OrdersSql);
        await db.ExecuteAsync(WidgetsSql);
        // Predates registration: invisible to CDC, so only baselining puts it in the ledger.
        await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1,'PRE',1),(2,'PRE',2)");
        db.RegisterReplicated("widgets");
        await db.ExecuteAsync("INSERT INTO widgets (id, name) VALUES (1,'W1'),(2,'W2')");
        db.RegisterReplicated("orders", baselineExistingRows: true);
        return NewSessionOver(db, options);
    }
    /// <summary>The peer of <see cref="NewBaselinedSideAsync"/>: the same two tables, empty, owing nothing.</summary>
    private async Task<Side> NewTwoTableSideAsync(ReplicationOptions? options = null)
    {
        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(OrdersSql);
        await db.ExecuteAsync(WidgetsSql);
        db.RegisterReplicated("orders");
        db.RegisterReplicated("widgets");
        return NewSessionOver(db, options);
    }
    private static List<string> SnapshotTablesIn(IEnumerable<SyncMessage> messages) =>
        messages.Where(m => m.MsgCase == SyncMessage.MsgOneofCase.SnapshotBatch)
            .Select(m => m.SnapshotBatch.TableName)
            .Distinct(StringComparer.Ordinal)
            .Order(StringComparer.Ordinal)
            .ToList();
    [Fact]
    public async Task Baselining_OneTable_StreamsOnlyThatTable()
    {
        // The whole point of the per-table flag. Through 0.2.x the flag was one bit for the whole
        // database, so baselining `orders` re-streamed `widgets` (and everything else) in full — in
        // both directions at once on a pair enabling replication late, which on a database holding
        // an event log at its retention cap is the largest transfer this library performs.
        var a = await NewBaselinedSideAsync();
        Assert.Equal(["orders"], await PendingSnapshotTables(a.Db));
        Assert.Equal(0, await NeedsSnapshot(a.Db)); // NOT the database-wide flag
        var (duplex, toSession, fromSession) = ScriptedPeer();
        using var cts = new CancellationTokenSource(RunTimeout);
        var run = a.Session.RunAsync(duplex, cts.Token);
        await DriveHandshakeAsync(a.Db, toSession, fromSession, cts.Token, supportsPartialSnapshot: true);
        var frames = await DrainThroughAsync(fromSession, SyncMessage.MsgOneofCase.SnapshotComplete, cts.Token);
        var begin = frames.Single(m => m.MsgCase == SyncMessage.MsgOneofCase.SnapshotBegin).SnapshotBegin;
        Assert.Equal(["orders"], begin.Tables);
        Assert.Equal(["orders"], SnapshotTablesIn(frames));
        // The baselined rows really are in there — the scope narrowed the snapshot, it did not empty it.
        var pks = frames.Where(m => m.MsgCase == SyncMessage.MsgOneofCase.SnapshotBatch)
            .SelectMany(m => m.SnapshotBatch.Rows).Select(r => r.PkJson).Order(StringComparer.Ordinal);
        Assert.Equal(["{\"id\":1}", "{\"id\":2}"], pks);
        cts.Cancel();
        await SwallowAsync(run);
    }
    [Fact]
    public async Task Baselining_AgainstAPeerWithoutPartialSupport_StreamsEveryTable()
    {
        // Mixed-version interop, the hard constraint. A pre-0.3.0 peer ignores SnapshotBegin.tables
        // and would advance last_applied_remote_seq to as_of_seq on complete — which for a SCOPED
        // snapshot silently discards every delta below as_of for the tables it left out. There is no
        // way to express "partial" to that peer, so it is served the 0.2.x answer: everything.
        var a = await NewBaselinedSideAsync();
        var (duplex, toSession, fromSession) = ScriptedPeer();
        using var cts = new CancellationTokenSource(RunTimeout);
        var run = a.Session.RunAsync(duplex, cts.Token);
        // supportsPartialSnapshot omitted == what a 0.1.x/0.2.x handshake decodes to.
        await DriveHandshakeAsync(a.Db, toSession, fromSession, cts.Token);
        var frames = await DrainThroughAsync(fromSession, SyncMessage.MsgOneofCase.SnapshotComplete, cts.Token);
        var begin = frames.Single(m => m.MsgCase == SyncMessage.MsgOneofCase.SnapshotBegin).SnapshotBegin;
        Assert.Empty(begin.Tables); // empty == full, the only scope the old wire can express
        Assert.Equal(["orders", "widgets"], SnapshotTablesIn(frames));
        cts.Cancel();
        await SwallowAsync(run);
    }
    [Fact]
    public async Task PartialSnapshot_DebtClearsOnlyWhenThePeerAcknowledges()
    {
        // Crash safety: an ungraceful exit part-way through a snapshot must re-stream it, so the
        // flag cannot come down when the frames are merely enqueued. The peer's SnapshotAck is
        // written after its own durable watermark write, so it means "I hold these rows".
        var a = await NewBaselinedSideAsync();
        var (duplex, toSession, fromSession) = ScriptedPeer();
        using var cts = new CancellationTokenSource(RunTimeout);
        var run = a.Session.RunAsync(duplex, cts.Token);
        await DriveHandshakeAsync(a.Db, toSession, fromSession, cts.Token, supportsPartialSnapshot: true);
        var frames = await DrainThroughAsync(fromSession, SyncMessage.MsgOneofCase.SnapshotComplete, cts.Token);
        var complete = frames[^1].SnapshotComplete;
        // The entire snapshot is on the wire and unacknowledged: the debt must still stand.
        Assert.Equal(["orders"], await PendingSnapshotTables(a.Db));
        await toSession.WriteAsync(new SyncMessage { SnapshotAck = new SnapshotAck { AsOfSeq = complete.AsOfSeq } }, cts.Token);
        await WaitForAsync(async () => (await PendingSnapshotTables(a.Db)).Count == 0, RunTimeout);
        cts.Cancel();
        await SwallowAsync(run);
    }
    [Fact]
    public async Task Snapshot_ToAPeerWithoutPartialSupport_ClearsOnEnqueue()
    {
        // The 0.2.x behaviour, kept deliberately: a peer that sends no SnapshotAck would otherwise
        // leave the debt standing forever and re-stream the snapshot on every reconnect.
        var a = await NewBaselinedSideAsync();
        var (duplex, toSession, fromSession) = ScriptedPeer();
        using var cts = new CancellationTokenSource(RunTimeout);
        var run = a.Session.RunAsync(duplex, cts.Token);
        await DriveHandshakeAsync(a.Db, toSession, fromSession, cts.Token);
        await DrainThroughAsync(fromSession, SyncMessage.MsgOneofCase.SnapshotComplete, cts.Token);
        await WaitForAsync(async () => (await PendingSnapshotTables(a.Db)).Count == 0, RunTimeout);
        cts.Cancel();
        await SwallowAsync(run);
    }
    [Fact]
    public async Task PartialSnapshot_DoesNotAdvanceTheReceiversWatermark()
    {
        // The receiver's half of the same invariant. last_applied_remote_seq is ONE number for the
        // whole peer oplog; moving it to a partial snapshot's as_of would declare the untouched
        // tables applied through that point, and their deltas below it would never be resent.
        var a = await NewSideAsync();
        await a.Store.SetLastAppliedRemoteSeqAsync(10, 0);
        var (duplex, toSession, fromSession) = ScriptedPeer();
        using var cts = new CancellationTokenSource(RunTimeout);
        var run = a.Session.RunAsync(duplex, cts.Token);
        await DriveHandshakeAsync(a.Db, toSession, fromSession, cts.Token, supportsPartialSnapshot: true);
        var begin = new SnapshotBegin { AsOfSeq = 900 };
        begin.Tables.Add("orders");
        await toSession.WriteAsync(new SyncMessage { SnapshotBegin = begin }, cts.Token);
        await toSession.WriteAsync(new SyncMessage
        {
            SnapshotBatch = new SnapshotBatch { TableName = "orders", Rows = { SnapRow(1, "SNAP", 6_000_000) } },
        }, cts.Token);
        await toSession.WriteAsync(new SyncMessage { SnapshotComplete = new SnapshotComplete { AsOfSeq = 900 } }, cts.Token);
        // The ack proves the complete was processed — and that a 0.3.0 receiver answers one at all.
        var ack = await NextOfCaseAsync(fromSession, SyncMessage.MsgOneofCase.SnapshotAck, cts.Token);
        Assert.Equal(900, ack.SnapshotAck.AsOfSeq);
        Assert.Equal(10, await LastAppliedRemoteSeq(a.Db)); // NOT 900
        Assert.Single(await ReadOrders(a.Db));              // the rows still merged
        cts.Cancel();
        await SwallowAsync(run);
    }
    [Fact]
    public async Task FullSnapshot_ToAnOldPeer_DrawsNoSnapshotAck()
    {
        // Symmetry check on the negotiation: a 0.3.0 receiver must not put a message a pre-0.3.0
        // sender has never heard of on the wire back to it.
        var a = await NewSideAsync();
        var (duplex, toSession, fromSession) = ScriptedPeer();
        using var cts = new CancellationTokenSource(RunTimeout);
        var run = a.Session.RunAsync(duplex, cts.Token);
        await DriveHandshakeAsync(a.Db, toSession, fromSession, cts.Token);
        await toSession.WriteAsync(new SyncMessage { SnapshotBegin = new SnapshotBegin { AsOfSeq = 300 } }, cts.Token);
        await toSession.WriteAsync(new SyncMessage
        {
            SnapshotBatch = new SnapshotBatch { TableName = "orders", Rows = { SnapRow(1, "SNAP", 6_000_000) } },
        }, cts.Token);
        await toSession.WriteAsync(new SyncMessage { SnapshotComplete = new SnapshotComplete { AsOfSeq = 300 } }, cts.Token);
        // A full snapshot from an old peer still advances the watermark — that is the 0.2.x contract.
        await WaitForAsync(async () => await LastAppliedRemoteSeq(a.Db) == 300, RunTimeout);
        // Barrier: an ack for this delta proves everything before it was processed, so an absent
        // SnapshotAck is absent rather than merely late.
        await toSession.WriteAsync(new SyncMessage
        {
            DeltaBatch = new DeltaBatch { Entries = { new OplogEntry
            {
                Seq = 301, TableName = "orders", PkJson = "{\"id\":2}",
                RowJson = "{\"id\":2,\"sku\":\"BARRIER\",\"qty\":1}", Hlc = 6_000_001, NodeId = "peer", IsTombstone = false,
            } } },
        }, cts.Token);
        await NextOfCaseAsync(fromSession, SyncMessage.MsgOneofCase.DeltaAck, cts.Token);
        Assert.False(fromSession.TryPeek(out var stray) && stray.MsgCase == SyncMessage.MsgOneofCase.SnapshotAck);
        cts.Cancel();
        await SwallowAsync(run);
    }
    [Fact]
    public async Task BaselinedTable_ReachesARealPeer_WithoutRestreamingTheOthers()
    {
        // End to end over two real databases: the baselined table's pre-existing rows arrive via the
        // scoped snapshot, the untouched table's rows arrive as ordinary deltas, and the debt clears.
        var options = new ReplicationOptions { FlushInterval = TimeSpan.FromMilliseconds(20) };
        var a = await NewBaselinedSideAsync(options);
        var b = await NewTwoTableSideAsync(options);
        var (da, db) = DuplexPair();
        using var cts = new CancellationTokenSource(RunTimeout);
        var runA = a.Session.RunAsync(da, cts.Token);
        var runB = b.Session.RunAsync(db, cts.Token);
        async Task<bool> Converged() =>
            (await ReadOrders(b.Db)).Count == 2
            && (await b.Db.QueryAsync("SELECT COUNT(*) FROM widgets", x => x.GetInt64(0)))[0] == 2;
        await WaitForAsync(Converged, RunTimeout);
        Assert.Equal(await ReadOrders(a.Db), await ReadOrders(b.Db));
        await WaitForAsync(async () => (await PendingSnapshotTables(a.Db)).Count == 0, RunTimeout);
        // A partial snapshot speaks for nothing outside its tables, so neither side may skip the
        // deltas it covered: B's watermark tracks the oplog, not the snapshot's as-of.
        Assert.Equal(0, await NeedsSnapshot(a.Db));
        cts.Cancel();
        await SwallowAsync(runA);
        await SwallowAsync(runB);
    }
    [Fact]
    public async Task Sender_ClearsNeedsSnapshotFlag()
    {
@@ -553,8 +553,9 @@ public sealed class SyncSessionTests : IDisposable
            await a.Store.SetNeedsSnapshotAsync(true, default);
            var (duplex, toSession, fromSession) = ScriptedPeer();
            var sent = new TaskCompletionSource();
-            a.Session.SnapshotSender = async (send, token) =>
+            a.Session.SnapshotSender = async (scope, send, token) =>
            {
                Assert.False(scope.IsPartial); // the database-wide flag always means every table
                sent.TrySetResult();
                await send(new SyncMessage { SnapshotComplete = new SnapshotComplete { AsOfSeq = 7 } }, token);
                return 7;