fix(localdb): refuse cascade-delete FK parents; make one entry's row+ledger writes atomic (0.3.3, ScadaBridge arch-review P4-7)

Two halves of the same INSERT OR REPLACE hazard in the LWW apply path. Wire-compatible
with every earlier peer: no proto change, no lib_schema_version bump, no on-disk schema
change.

1. RegisterReplicated now REFUSES a table that another table references with a
   row-mutating ON DELETE action (CASCADE / SET NULL / SET DEFAULT). Applying a remote row
   uses INSERT OR REPLACE, whose implicit DELETE fires that referential action — while the
   apply path holds the capture guard raised, precisely so its own writes are not
   re-captured. The child rows it deletes or nulls therefore produce no oplog entry: the
   peer keeps them and the pair diverges with nothing reporting a fault. Registration is
   the only point at which the problem is still visible, so it is refused there with an
   actionable LocalDbRegistrationException naming the referencing table and its action.
   NO ACTION / RESTRICT stay allowed: they mutate no child row, so the implicit delete just
   fails and the applier dead-letters the entry — loud, not silent. Registration-time only;
   a cascading child table created LATER is not caught (SQLite offers no DDL hook).

   No current consumer has a foreign key on a replicated table (ScadaBridge's ten tables
   have none), so this is a guard against a future foot-gun, not a live fix.

2. One entry's row write and its ledger write are now atomic. They were two statements
   under the poison catch, so a failure landing between them committed a deleted/replaced
   row carrying a STALE LIVE row-version entry — exactly the shape SnapshotStreamer warns
   about and skips, so that key never travels again and the pair is permanently divergent
   for it. A per-entry SAVEPOINT makes the pair all-or-nothing while keeping the catch's
   per-ENTRY granularity: entries applied earlier in the same batch survive the rollback,
   the poison entry is still dead-lettered, and the batch still continues. The RELEASE runs
   outside the catch on purpose, so a failure there rolls the whole batch back rather than
   being swallowed as poison. Measured cost on a 100k-row backlog drain: ~7%
   (178.6k rows/s -> 166.1k rows/s).

Tests: both hazards proven against the unfixed code first (FK parent registered silently;
an injected ledger-write abort left the row deleted with a live ledger entry). Adds FK
guard coverage (parent refused, child allowed, SET NULL / SET DEFAULT refused, NO ACTION /
RESTRICT allowed, unrelated cascades ignored) and two applier atomicity tests. The
statement-count lock-in moves 7 -> 9 for the two new savepoint statements.

Version 0.3.2 -> 0.3.3; README limitations + release notes document the constraint.

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

@@ -5,7 +5,7 @@
     <Nullable>enable</Nullable>
     <ImplicitUsings>enable</ImplicitUsings>
     <LangVersion>latest</LangVersion>
-    <Version>0.3.1</Version>
+    <Version>0.3.3</Version>
     <ManagePackageVersionsCentrally>true</ManagePackageVersionsCentrally>
     <TreatWarningsAsErrors>true</TreatWarningsAsErrors>
     <!-- NU1510 (package-pruning advice): the Replication and Tests projects carry a

ZB.MOM.WW.LocalDb/README.md

@@ -10,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.3.0`)
+## Packages (all `0.3.3`)
 
 | Package | Contents |
 |---|---|
@@ -20,6 +20,50 @@ 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.3` (2026-08-19) — one Low finding from the ScadaBridge adversarial architecture review
+>   (P4-7), in two halves. Wire-compatible with every earlier peer: no proto change, no
+>   `lib_schema_version` bump, no on-disk schema change.
+>   - **`RegisterReplicated` now refuses a table that another table cascades a DELETE onto.**
+>     Applying a remote row uses `INSERT OR REPLACE`, whose implicit DELETE fires `ON DELETE
+>     CASCADE` / `SET NULL` / `SET DEFAULT` on every child — while the apply path holds the capture
+>     guard raised, so the child rows it touches produce no oplog entry, the peer keeps them, and the
+>     pair diverges with nothing reporting a fault. Registration is the only place that is visible,
+>     so it is refused there with an actionable `LocalDbRegistrationException` naming the referencing
+>     table and its action. `NO ACTION` / `RESTRICT` are still allowed: they mutate no child row (the
+>     implicit delete simply fails, which the applier dead-letters — loud, not silent). No current
+>     consumer has a foreign key on a replicated table, so this is a guard against a future foot-gun
+>     rather than a live fix.
+>   - **Applying one entry is atomic across its row write and its ledger write.** They were two
+>     statements under the poison catch, so a failure landing between them committed a
+>     deleted/replaced row carrying a STALE LIVE ledger version — the shape `SnapshotStreamer` warns
+>     about and skips, meaning that key never travels again. A per-entry `SAVEPOINT` makes the pair
+>     all-or-nothing while keeping the catch's per-ENTRY granularity: entries applied earlier in the
+>     same batch survive, and the poison entry is still dead-lettered and the batch still continues.
+> - `0.3.2` (2026-08-19) — two Medium findings from the ScadaBridge adversarial architecture review
+>   (P4-1, P4-3). Wire-compatible with every earlier peer: no proto change, no `lib_schema_version`
+>   bump, no on-disk schema change — the keepalive is transport-level and the rest is local policy.
+>   - **A live session now ends itself when a cap prune eats deltas it has not sent.** The pump reads
+>     `seq > sentThruSeq`, which resumes SILENTLY above a pruned hole; the peer acks the later seqs
+>     and the pair is durably divergent for the whole life of that session. `needs_snapshot` (set in
+>     the prune's own transaction) only heals a RECONNECT, and nothing forced one — so on a stable
+>     LAN with a slow-but-alive peer the divergence was unbounded. Every pump iteration now compares
+>     the oplog horizon (`MIN(seq)`, one b-tree seek) against its sent watermark and throws
+>     `LocalDbOplogGapException` when deltas it still owes are gone. The horizon, not the flag,
+>     because the flag is also raised for reasons that do not invalidate the running session.
+>   - **The sync stream has HTTP/2 keepalive, and the server's single-stream guard is newest-wins.**
+>     Neither end ever timed out on silence (a converged pair is legitimately quiet), so a hard crash
+>     or partition of the dialling node left the passive node's handler parked in `MoveNext` on a
+>     half-open stream — holding the guard, reporting itself connected, and rejecting the rebooted
+>     peer's reconnects with `AlreadyExists`, indefinitely on an idle pair. The initiator now dials
+>     with `KeepAlivePingDelay` / `KeepAlivePingTimeout` (new options, 60s/20s, `Always` policy so
+>     idle connections ping too), and a fresh authenticated `Sync` cancels the incumbent and waits
+>     for it to unwind before starting — the guard serializes sessions, it was never a fence, and an
+>     authenticated same-key dialler is by definition the peer. An incumbent that will not yield
+>     within the takeover timeout still gets `AlreadyExists`, because two live sessions over one
+>     database is worse than a refused dial.
+> - `0.3.1` (2026-08-18) — a faulting session flushes its control lane before cancelling its writer,
+>   so a locally-detected handshake mismatch actually reaches the peer instead of leaving it to
+>   reconnect-loop against a generic disconnect.
 > - `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
@@ -235,6 +279,18 @@ All options are validated at startup (`ZB.MOM.WW.Configuration` validators).
   `RegisterReplicated`, because `json_object` cannot capture it). A *typeless* column can
   still hold a BLOB value at runtime, which then surfaces as a write-time error, not a
   registration error.
+- **Replicated tables must not be cascade-delete foreign-key parents.** If another table
+  references the one you are registering with `ON DELETE CASCADE` (or `SET NULL` /
+  `SET DEFAULT`), `RegisterReplicated` throws. Applying a remote row uses `INSERT OR
+  REPLACE`, whose implicit DELETE fires that referential action while change capture is
+  suppressed — so the child rows it deletes or nulls are never captured, the peer keeps
+  them, and the pair diverges with nothing reporting a fault. Model the parent/child
+  relationship with `ON DELETE NO ACTION` (or `RESTRICT`) and delete children explicitly, or
+  leave the parent unreplicated. `NO ACTION` / `RESTRICT` are allowed because they mutate no
+  child row: the implicit delete just fails, and the applier dead-letters the entry — loud,
+  not silent. The check runs at **registration time**, so a cascading child table created
+  *later* is not caught (SQLite offers no DDL hook); register after schema setup, which is
+  where startup does it anyway.
 - **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

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

@@ -14,6 +14,8 @@ internal sealed record ApplyResult(long AppliedThruSeq, int Applied, int Discard
 /// row-version (hlc, then node_id ordinal tie-break), poison rows are dead-lettered while the
 /// batch continues, and the remote watermark advances in the SAME transaction as the data so a
 /// crash-replay of the batch is a no-op (LWW's strict-greater compare loses on equal versions).
+/// Each entry's row write and its ledger write are additionally scoped by a per-entry SAVEPOINT, so
+/// the pair is all-or-nothing even when the poison catch keeps the batch alive around it.
 /// This never writes <c>__localdb_oplog</c>: the 2-node topology does not re-forward applied deltas.
 /// </summary>
 /// <remarks>
@@ -88,6 +90,18 @@ internal sealed class LwwApplier(SqliteLocalDb db, Func<DateTimeOffset>? utcNow
                 continue;
             }
 
+            // The row mutation and its ledger entry are TWO statements and must land together or
+            // not at all: a failure between them commits a deleted/replaced row carrying a stale
+            // ledger version, which SnapshotStreamer then warns about and skips — so that key never
+            // travels again and the pair is silently, permanently divergent for it. A per-entry
+            // savepoint makes the pair atomic without giving up the poison catch's per-ENTRY
+            // granularity (the surrounding transaction, and every entry already applied in it,
+            // survives a rollback to the savepoint). Cost measured on the path that matters — a
+            // 100 000-row backlog drain — is ~7% (178.6k rows/s → 166.1k rows/s): the two extra
+            // statements are prepared once like the rest and neither one touches the disk, since
+            // an inner savepoint's release just merges into the enclosing transaction.
+            var pairApplied = false;
+            await sql.SavepointAsync(ct);
             try
             {
                 var table = db.ReplicatedTables[entry.TableName];
@@ -97,7 +111,7 @@ internal sealed class LwwApplier(SqliteLocalDb db, Func<DateTimeOffset>? utcNow
                     await sql.UpsertAsync(table, entry.RowJson!, ct);
 
                 await sql.WriteRowVersionAsync(entry, utcNow, ct);
-                applied++;
+                pairApplied = true;
             }
             catch (SqliteException ex)
             {
@@ -105,9 +119,21 @@ 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).
+                // The rollback comes FIRST so the dead-letter row it records is not discarded with
+                // the half-applied pair.
+                await sql.RollbackToSavepointAsync(ct);
                 await sql.DeadLetterAsync(entry, ex.Message, utcNow, ct);
                 deadLettered++;
             }
+
+            // Released outside the catch on purpose: a RELEASE that itself fails means the
+            // transaction is no longer sound, and there is no savepoint state left to reason about,
+            // so it propagates and rolls the whole batch back rather than being swallowed as poison.
+            if (pairApplied)
+            {
+                await sql.ReleaseSavepointAsync(ct);
+                applied++;
+            }
         }
 
         await sql.AdvancePeerStateAsync(maxSeq, maxHlc, utcNow, ct);
@@ -143,10 +169,42 @@ internal sealed class LwwApplier(SqliteLocalDb db, Func<DateTimeOffset>? utcNow
         private SqliteCommand? _rowVersion;
         private SqliteCommand? _deadLetter;
         private SqliteCommand? _peerState;
+        private SqliteCommand? _savepoint;
+        private SqliteCommand? _releaseSavepoint;
+        private SqliteCommand? _rollbackToSavepoint;
+
+        /// <summary>The per-entry savepoint's name. Reused for every entry — one is open at a time.</summary>
+        private const string SavepointName = "__localdb_apply_entry";
 
         /// <summary>How many distinct statements this set has prepared.</summary>
         public int StatementCount => _owned.Count;
 
+        /// <summary>Opens the per-entry savepoint that scopes one entry's row + ledger writes.</summary>
+        public Task SavepointAsync(CancellationToken ct)
+        {
+            var cmd = _savepoint ??= Create("SAVEPOINT " + SavepointName);
+            return cmd.ExecuteNonQueryAsync(ct);
+        }
+
+        /// <summary>Commits the entry's savepoint into the surrounding batch transaction.</summary>
+        public Task ReleaseSavepointAsync(CancellationToken ct)
+        {
+            var cmd = _releaseSavepoint ??= Create("RELEASE " + SavepointName);
+            return cmd.ExecuteNonQueryAsync(ct);
+        }
+
+        /// <summary>
+        /// Undoes everything the entry did and closes its savepoint. ROLLBACK TO leaves the savepoint
+        /// OPEN, so the RELEASE is required — without it the names would nest one deeper per poison
+        /// entry and the next rollback would unwind more than one entry's work.
+        /// </summary>
+        public async Task RollbackToSavepointAsync(CancellationToken ct)
+        {
+            var cmd = _rollbackToSavepoint ??= Create("ROLLBACK TO " + SavepointName);
+            await cmd.ExecuteNonQueryAsync(ct);
+            await ReleaseSavepointAsync(ct);
+        }
+
         public Task SetApplyingAsync(int value, CancellationToken ct)
         {
             var cmd = _applying ??= Create(

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

@@ -24,8 +24,9 @@ internal sealed class OplogStore(ILocalDb db, ReplicationOptions options, Func<D
 
     /// <summary>
     /// Reads up to <paramref name="maxBatch"/> oplog entries with <c>seq &gt; afterSeq</c> in ascending
-    /// seq order. Caller contract: check <c>needs_snapshot</c> before trusting incremental reads —
-    /// after cap-pruning, reads silently resume past a gap.
+    /// seq order. Caller contract: this read resumes SILENTLY past a cap-pruned gap, so a caller
+    /// streaming incrementally must first confirm its position is still below the horizon — see
+    /// <see cref="GetMinSeqAsync"/>, which <c>SyncSession</c>'s pump checks on every iteration.
     /// </summary>
     public Task<IReadOnlyList<OplogEntryRecord>> ReadBatchAboveAsync(long afterSeq, int maxBatch, CancellationToken ct = default) =>
         db.QueryAsync(
@@ -148,6 +149,24 @@ internal sealed class OplogStore(ILocalDb db, ReplicationOptions options, Func<D
         return rows[0];
     }
 
+    /// <summary>
+    /// The oldest oplog seq still present — the delta horizon — or 0 when the oplog is empty.
+    /// </summary>
+    /// <remarks>
+    /// One b-tree seek (<c>seq</c> is the rowid and SQLite optimizes MIN into a seek), which is what
+    /// makes it affordable for the pump to check on every flush interval. Compared against a
+    /// session's sent watermark it answers the only question the pump has: is everything I have not
+    /// yet sent still here? Ack-pruning deletes <c>seq &lt;= last_acked_seq</c> and the peer can only
+    /// ack what it received, so in a healthy session the horizon never climbs past
+    /// <c>sentThruSeq + 1</c>; a higher one means a cap prune ate unsent deltas.
+    /// </remarks>
+    public async Task<long> GetMinSeqAsync(CancellationToken ct = default)
+    {
+        var rows = await db.QueryAsync(
+            "SELECT COALESCE(MIN(seq), 0) FROM __localdb_oplog", static r => r.GetInt64(0), null, ct);
+        return rows[0];
+    }
+
     /// <summary>The highest oplog seq present, or 0 when the oplog is empty.</summary>
     public async Task<long> GetMaxSeqAsync(CancellationToken ct = default)
     {

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

@@ -356,9 +356,14 @@ internal sealed class SyncSession
         while (true)
         {
             ct.ThrowIfCancellationRequested();
+            // BEFORE the read, because the read itself cannot tell a pruned gap from an idle oplog.
+            ThrowIfPruneGap(await _store.GetMinSeqAsync(ct));
             var batch = await _store.ReadBatchAboveAsync(_sentThruSeq, _options.MaxBatchSize, ct);
             if (batch.Count > 0)
             {
+                // The exact guard on the rows about to go out, for a prune that landed between the
+                // probe and the read. Free (the seq is already in hand) and strictly tighter.
+                ThrowIfPruneGap(batch[0].Seq);
                 var index = 0;
                 while (index < batch.Count)
                 {
@@ -390,6 +395,40 @@ internal sealed class SyncSession
         }
     }
 
+    /// <summary>
+    /// Faults the session when the oplog horizon has climbed above this session's read position —
+    /// i.e. a cap prune deleted deltas between <see cref="_sentThruSeq"/> and
+    /// <paramref name="oldestAvailableSeq"/>. An empty oplog (0) is not a gap.
+    /// </summary>
+    /// <remarks>
+    /// <para>
+    /// <b>Why a live session has to die for this.</b> <c>EnforceCapsAsync</c> sets
+    /// <c>needs_snapshot</c> in the same transaction as the prune, so a RECONNECT heals — the
+    /// handshake serves a full snapshot instead of deltas. Nothing, however, made the live session
+    /// notice: the pump reads <c>seq &gt; _sentThruSeq</c>, which silently resumes above the hole, and
+    /// the peer acks those later seqs as if the pair were converged. On a stable LAN with a slow but
+    /// alive peer that session can run for days, and the pair is durably divergent for all of it —
+    /// the flag that would fix it is only ever read at handshake time. Tearing down is what turns a
+    /// persisted "needs snapshot" into an actual snapshot.
+    /// </para>
+    /// <para>
+    /// <b>Why the horizon, not the flag.</b> <c>needs_snapshot</c> is also set for reasons that do
+    /// not invalidate THIS session's position (and a full snapshot this session just streamed clears
+    /// it only when the peer's <c>SnapshotAck</c> lands, which is after the pump has started).
+    /// Comparing the horizon against the sent watermark asks the precise question instead, so a
+    /// session is faulted when — and only when — rows it still owes the peer are gone.
+    /// </para>
+    /// </remarks>
+    private void ThrowIfPruneGap(long oldestAvailableSeq)
+    {
+        if (oldestAvailableSeq == 0 || oldestAvailableSeq <= _sentThruSeq + 1)
+            return;
+        throw new LocalDbOplogGapException(
+            $"Oplog deltas above the sent watermark {_sentThruSeq} were pruned (oldest surviving seq " +
+            $"{oldestAvailableSeq}): this session can no longer stream incrementally without skipping " +
+            "them. Ending the session so the reconnect handshake serves a snapshot resync.");
+    }
+
     /// <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.

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

@@ -0,0 +1,23 @@
+namespace ZB.MOM.WW.LocalDb.Replication;
+
+/// <summary>
+/// Thrown by a live sync session when the oplog horizon has moved ABOVE the position that session
+/// is streaming from: a cap prune deleted deltas it had not yet sent, so every later delta it could
+/// ship would land on the far side of a hole the peer will never receive.
+/// </summary>
+/// <remarks>
+/// This is not a transport failure and not a data error — it is the session discovering that its own
+/// incremental position has become unsound. Nothing can repair it in-session (the deleted rows are
+/// gone), so the session tears itself down; the same prune set <c>needs_snapshot</c>, which makes the
+/// reconnect handshake serve a snapshot instead of deltas. Catching this and continuing would
+/// reintroduce exactly the silent divergence it exists to stop.
+/// </remarks>
+public sealed class LocalDbOplogGapException : Exception
+{
+    /// <summary>Creates the exception with a message describing the gap.</summary>
+    public LocalDbOplogGapException(string message) : base(message) { }
+
+    /// <summary>Creates the exception with a message and the underlying cause.</summary>
+    public LocalDbOplogGapException(string message, Exception innerException)
+        : base(message, innerException) { }
+}

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

@@ -21,8 +21,26 @@ public sealed class LocalDbSyncService : LocalDbSync.LocalDbSyncBase
     private readonly LocalDbMetrics? _metrics;
     private readonly SyncStatus? _status;
 
-    // 0 = idle, 1 = a stream is active. Interlocked-guarded across concurrent calls (singleton service).
-    private int _active;
+    // The stream currently holding the guard, or null when idle. Guarded by _gate (singleton service).
+    private readonly object _gate = new();
+    private Registration? _active;
+
+    /// <summary>
+    /// How long a displacing call waits for the incumbent it evicted to actually finish before
+    /// giving up and refusing itself. Test seam; the default is generous because eviction only has
+    /// to outrun the operator, not the network.
+    /// </summary>
+    internal TimeSpan TakeoverTimeout { get; set; } = TimeSpan.FromSeconds(10);
+
+    /// <summary>One in-flight <c>Sync</c> handler: its cancellation handle plus a completion signal
+    /// a displacing handler can await, so two sessions never run against the database at once.</summary>
+    private sealed class Registration(CancellationTokenSource cts)
+    {
+        public CancellationTokenSource Cts { get; } = cts;
+
+        public TaskCompletionSource Finished { get; } =
+            new(TaskCreationOptions.RunContinuationsAsynchronously);
+    }
 
     /// <summary>
     /// Creates the passive sync service. <paramref name="status"/> is taken as the public
@@ -41,21 +59,47 @@ public sealed class LocalDbSyncService : LocalDbSync.LocalDbSyncBase
         _status = status as SyncStatus;
     }
 
+    /// <summary>
+    /// Accepts one inbound sync stream, evicting any stream already holding the guard.
+    /// </summary>
+    /// <remarks>
+    /// <para>
+    /// <b>Newest wins.</b> The guard exists to SERIALIZE sessions — a node syncs with exactly one
+    /// peer, and two sessions against one database would fight over the same watermarks — not to
+    /// fence out an impostor; authentication is the interceptor's job, and an authenticated caller
+    /// on this key IS the peer by definition. Refusing the newcomer therefore protected nothing and
+    /// cost everything in the case that actually happens: the dialling node crashes or is
+    /// partitioned, this handler stays parked forever inside <c>MoveNext</c> on a half-open stream
+    /// that will never produce another byte, and every reconnect from the rebooted peer is rejected
+    /// <c>AlreadyExists</c> — potentially for as long as the pair stays idle, while this side keeps
+    /// reporting itself connected. The newest dialler is the one with a live link, so it takes over.
+    /// </para>
+    /// <para>
+    /// Eviction is cancel-then-await, not cancel-and-go: the incumbent's cancellation unwinds its
+    /// session and reader pump, and the newcomer waits for that to complete before starting its own,
+    /// so the "one session at a time" property the guard was built for still holds. An incumbent
+    /// that cannot be evicted within <see cref="TakeoverTimeout"/> is the one case where the old
+    /// answer is still right — the newcomer refuses itself rather than double-running.
+    /// </para>
+    /// </remarks>
     public override async Task Sync(
         IAsyncStreamReader<SyncMessage> requestStream,
         IServerStreamWriter<SyncMessage> responseStream,
         ServerCallContext context)
     {
-        if (Interlocked.CompareExchange(ref _active, 1, 0) != 0)
-            throw new RpcException(new Status(StatusCode.AlreadyExists, "a sync stream is already active"));
+        // The linked source exists so the reader pump is deterministically torn down on EVERY exit
+        // path — including a session fault for a local (non-transport) reason, where the call token
+        // never fires and the pump would otherwise keep reading a stream whose handler has exited.
+        // It is also the handle a later call cancels to displace this one.
+        using var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(context.CancellationToken);
+        var mine = new Registration(linkedCts);
 
         try
         {
-            // The linked source exists so the reader pump is deterministically torn down on EVERY
-            // exit path — including a session fault for a local (non-transport) reason, where the
-            // call token never fires and the pump would otherwise keep reading a stream whose
-            // handler has exited.
-            using var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(context.CancellationToken);
+            // Inside the try so the finally below always releases the guard — including when the
+            // takeover itself is abandoned because THIS caller gave up mid-handover.
+            await TakeOverAsync(mine, context.CancellationToken);
+
             var (duplex, readerTask) = GrpcSyncDuplex.Create(requestStream, responseStream, linkedCts.Token);
             var session = SyncSessionFactory.Create(_db, _options, _loggerFactory, _metrics, _status);
             _status?.SessionStarted();
@@ -81,13 +125,66 @@ public sealed class LocalDbSyncService : LocalDbSync.LocalDbSyncBase
         {
             // Client disconnect / server shutdown: the normal end of a long-lived streaming RPC.
         }
+        catch (OperationCanceledException)
+        {
+            // Displaced by a newer stream: an orderly handover, not a failure of this node.
+            _logger.LogInformation("Sync stream ended: displaced by a newer stream from the peer.");
+        }
         catch (LocalDbSchemaMismatchException ex)
         {
             throw new RpcException(new Status(StatusCode.FailedPrecondition, ex.Message));
         }
         finally
         {
-            Interlocked.Exchange(ref _active, 0);
+            lock (_gate)
+            {
+                if (ReferenceEquals(_active, mine))
+                    _active = null;
+            }
+            mine.Finished.TrySetResult();
+        }
+    }
+
+    /// <summary>
+    /// Claims the single-stream guard for <paramref name="mine"/>, cancelling and awaiting whatever
+    /// held it. Throws <c>AlreadyExists</c> only when the incumbent refuses to end.
+    /// </summary>
+    private async Task TakeOverAsync(Registration mine, CancellationToken ct)
+    {
+        Registration? displaced;
+        lock (_gate)
+        {
+            displaced = _active;
+            _active = mine;
+        }
+        if (displaced is null)
+            return;
+
+        _logger.LogWarning(
+            "A new sync stream arrived while one was already active; displacing the incumbent (newest wins).");
+        try
+        {
+            // The incumbent may already be disposing its own source between its finally and here.
+            displaced.Cts.Cancel();
+        }
+        catch (ObjectDisposedException) { }
+
+        try
+        {
+            await displaced.Finished.Task.WaitAsync(TakeoverTimeout, ct);
+        }
+        catch (TimeoutException)
+        {
+            // Could not evict it: two live sessions over one database is worse than a refused dial,
+            // so fall back to the historical answer — and hand the guard back to the incumbent.
+            lock (_gate)
+            {
+                if (ReferenceEquals(_active, mine))
+                    _active = displaced;
+            }
+            throw new RpcException(new Status(
+                StatusCode.AlreadyExists,
+                "a sync stream is already active and did not yield within the takeover timeout"));
         }
     }
 }

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

@@ -54,7 +54,7 @@ public sealed class MaintenanceBackgroundService : BackgroundService
                 {
                     _snapshotFlagWarned = true;
                     _logger.LogWarning(
-                        "Oplog backlog/age caps exceeded: pruned to the ceiling and flagged needs_snapshot — the peer must snapshot-resync. Not re-logged until a clean tick.");
+                        "Oplog backlog/age caps exceeded: pruned to the ceiling and flagged needs_snapshot — the peer must snapshot-resync. A session streaming from below the new horizon ends itself on its next pump tick so the reconnect serves that snapshot. Not re-logged until a clean tick.");
                 }
                 else if (!snapshotFlagged)
                 {

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

@@ -64,6 +64,34 @@ public sealed class ReplicationOptions
     /// <summary>Upper bound on the exponential reconnect backoff.</summary>
     public TimeSpan ReconnectBackoffMax { get; set; } = TimeSpan.FromSeconds(60);
 
+    /// <summary>
+    /// How long the initiator's HTTP/2 connection may sit without a keepalive PING before one is
+    /// sent. Default 60s.
+    /// </summary>
+    /// <remarks>
+    /// <para>
+    /// A sync stream is long-lived and, on a converged pair, almost entirely silent — which is
+    /// exactly the shape a half-open TCP connection hides behind. Without pings, a hard crash or a
+    /// partition of one node leaves the other holding a stream that will never produce another byte
+    /// and never fail either: the initiator sits in a session it believes is healthy, and the
+    /// passive node's handler sits inside its single-stream guard, refusing the reconnect that would
+    /// heal the pair. Pings turn that into a detected failure on both ends.
+    /// </para>
+    /// <para>
+    /// Sent with <c>KeepAlivePingPolicy.Always</c>, so they flow on an idle connection too — an
+    /// active-streams-only policy would not fire in precisely the silent-pair case above. Safe over
+    /// h2c: a PING is an HTTP/2 frame, nothing to do with TLS.
+    /// </para>
+    /// </remarks>
+    public TimeSpan KeepAlivePingDelay { get; set; } = TimeSpan.FromSeconds(60);
+
+    /// <summary>
+    /// How long the initiator waits for a keepalive PING ack before tearing the connection down.
+    /// Default 20s. Detection therefore lands within roughly
+    /// <see cref="KeepAlivePingDelay"/> + this.
+    /// </summary>
+    public TimeSpan KeepAlivePingTimeout { get; set; } = TimeSpan.FromSeconds(20);
+
     /// <summary>How often the maintenance service enforces the oplog caps and prunes expired tombstones.</summary>
     public TimeSpan MaintenanceInterval { get; set; } = TimeSpan.FromSeconds(60);
 

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

@@ -40,6 +40,10 @@ public sealed class ReplicationOptionsValidator : IValidateOptions<ReplicationOp
             failures.Add($"LocalDb:Replication:ReconnectBackoffMax must be greater than zero; got {options.ReconnectBackoffMax}.");
         else if (options.ReconnectBackoffMax > ReconnectBackoffCeiling)
             failures.Add($"LocalDb:Replication:ReconnectBackoffMax must be at most {ReconnectBackoffCeiling}; got {options.ReconnectBackoffMax}.");
+        if (options.KeepAlivePingDelay <= TimeSpan.Zero)
+            failures.Add($"LocalDb:Replication:KeepAlivePingDelay must be greater than zero; got {options.KeepAlivePingDelay}.");
+        if (options.KeepAlivePingTimeout <= TimeSpan.Zero)
+            failures.Add($"LocalDb:Replication:KeepAlivePingTimeout must be greater than zero; got {options.KeepAlivePingTimeout}.");
         if (options.MaintenanceInterval <= TimeSpan.Zero)
             failures.Add($"LocalDb:Replication:MaintenanceInterval must be greater than zero; got {options.MaintenanceInterval}.");
         if (options.MaxHlcDriftAhead <= TimeSpan.Zero)

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

@@ -1,3 +1,4 @@
+using System.Net.Http;
 using Grpc.Core;
 using Grpc.Net.Client;
 using Microsoft.Extensions.Hosting;
@@ -76,7 +77,7 @@ public sealed class SyncBackgroundService : BackgroundService
             {
                 var channel = ChannelFactory is { } factory
                     ? factory()
-                    : ownedChannel = GrpcChannel.ForAddress(_options.PeerAddress);
+                    : ownedChannel = DialPeer();
 
                 var client = new LocalDbSync.LocalDbSyncClient(channel);
                 var headers = new Metadata();
@@ -152,6 +153,39 @@ public sealed class SyncBackgroundService : BackgroundService
         }
     }
 
+    /// <summary>
+    /// Dials the peer over a channel that sends HTTP/2 keepalive PINGs, so a half-open link is
+    /// detected instead of being waited on forever.
+    /// </summary>
+    /// <remarks>
+    /// <para>
+    /// The bare <c>GrpcChannel.ForAddress(address)</c> this replaces had NO keepalive of any kind,
+    /// and neither end of a sync stream ever times out on silence by design (a converged pair is
+    /// legitimately quiet). A hard crash or partition of the peer therefore left this node inside a
+    /// session that could never make progress and could never fail — and its own half of that stream
+    /// pinned on the peer, where the single-stream guard turns it into a refusal of the very
+    /// reconnect that heals the pair.
+    /// </para>
+    /// <para>
+    /// <c>DisposeHttpClient</c> hands the handler's lifetime to the channel, which this loop already
+    /// disposes on every exit path.
+    /// </para>
+    /// </remarks>
+    private GrpcChannel DialPeer() =>
+        GrpcChannel.ForAddress(_options.PeerAddress, new GrpcChannelOptions
+        {
+            HttpHandler = new SocketsHttpHandler
+            {
+                KeepAlivePingDelay = _options.KeepAlivePingDelay,
+                KeepAlivePingTimeout = _options.KeepAlivePingTimeout,
+                // Always, not WithActiveRequests: the case being closed is a SILENT stream on a dead
+                // link, and an active-requests policy stops pinging once the peer stops responding.
+                KeepAlivePingPolicy = HttpKeepAlivePingPolicy.Always,
+                EnableMultipleHttp2Connections = true,
+            },
+            DisposeHttpClient = true,
+        });
+
     private static bool IsSchemaMismatch(Exception? ex) =>
         ex is LocalDbSchemaMismatchException
         || (ex is RpcException rpc && rpc.StatusCode == StatusCode.FailedPrecondition);

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

@@ -38,8 +38,18 @@ public interface ILocalDb
     /// schema digest, and installs the three AFTER triggers that capture every change into the
     /// oplog + row-version tables within the same transaction as the consumer's write. Idempotent
     /// (re-registering regenerates the triggers). Throws <see cref="LocalDbRegistrationException"/>
-    /// if the name is unsafe, the table is missing, or it has no primary key.
+    /// if the name is unsafe, the table is missing, it has no primary key, it has a BLOB column, or
+    /// another table references it with a row-mutating <c>ON DELETE</c> action.
     /// </summary>
+    /// <remarks>
+    /// <b>Replicated tables must not be cascade-delete foreign-key parents.</b> Applying a remote row
+    /// uses <c>INSERT OR REPLACE</c>, whose implicit DELETE fires <c>ON DELETE CASCADE</c> /
+    /// <c>SET NULL</c> / <c>SET DEFAULT</c> on referencing tables while change capture is suppressed
+    /// — the child rows it touches are never captured, so the peer keeps them and the pair diverges
+    /// silently. Such a table is refused here, because registration is the only point at which the
+    /// problem is still visible. <c>NO ACTION</c> / <c>RESTRICT</c> are allowed: they mutate no child
+    /// row, so the implicit delete simply fails and the applier dead-letters the entry.
+    /// </remarks>
     ReplicatedTable RegisterReplicated(string tableName);
 
     /// <summary>

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

@@ -199,6 +199,8 @@ internal sealed class SqliteLocalDb : ILocalDb, IDisposable
                 throw new LocalDbRegistrationException(
                     $"Cannot register table '{tableName}': replication requires an explicit primary key.");
 
+            EnsureNoRowMutatingDeleteAction(tableName);
+
             var allColumns = columns.Select(c => (c.Name, c.Type)).ToList();
             var digest = ReplicatedTable.ComputeDigest(tableName, pkColumns, allColumns);
             var table = new ReplicatedTable(tableName, pkColumns, allColumns, digest);
@@ -293,6 +295,87 @@ internal sealed class SqliteLocalDb : ILocalDb, IDisposable
         }
     }
 
+    /// <summary>
+    /// Refuses to register <paramref name="tableName"/> when another table references it with an
+    /// <c>ON DELETE</c> action that MUTATES the referencing rows (<c>CASCADE</c>, <c>SET NULL</c>,
+    /// <c>SET DEFAULT</c>).
+    /// </summary>
+    /// <remarks>
+    /// <para>
+    /// <b>Why this is a replication correctness rule and not a schema opinion.</b> Applying a peer's
+    /// row runs <c>INSERT OR REPLACE</c>, and REPLACE resolves a primary-key conflict by DELETING the
+    /// existing row first. Connections run with <c>PRAGMA foreign_keys=ON</c>, so that implicit delete
+    /// fires the referential action on every child table — while the apply path holds the capture
+    /// guard raised, precisely so its own writes are not re-captured. The child rows are therefore
+    /// deleted (or nulled) locally and produce NO oplog entry: the peer never hears about them, keeps
+    /// them, and the pair is durably divergent with nothing anywhere reporting a fault.
+    /// </para>
+    /// <para>
+    /// Refusing at registration is the only place the problem is visible. At apply time the damage is
+    /// already done and indistinguishable from a normal write, and no runtime signal exists to detect
+    /// it. <b>NO ACTION / RESTRICT are deliberately allowed</b>: they mutate no child row — the
+    /// implicit delete simply fails, which the applier surfaces as a dead-lettered entry. Loud, not
+    /// silent, so there is nothing to guard.
+    /// </para>
+    /// <para>
+    /// The check is registration-time, so a cascading child table CREATED LATER is not caught (SQLite
+    /// has no DDL hook to catch it on). Registration normally runs at startup after schema setup,
+    /// which is where the constraint belongs.
+    /// </para>
+    /// </remarks>
+    private void EnsureNoRowMutatingDeleteAction(string tableName)
+    {
+        var candidates = new List<string>();
+        using (var cmd = _master.CreateCommand())
+        {
+            // Every user table is a candidate CHILD; the internal sqlite_* tables carry no
+            // foreign keys. The registered table itself is included: a self-referencing cascade
+            // is the same hazard.
+            cmd.CommandText = "SELECT name FROM sqlite_master WHERE type = 'table' AND name NOT LIKE 'sqlite_%'";
+            using var reader = cmd.ExecuteReader();
+            while (reader.Read())
+                candidates.Add(reader.GetString(0));
+        }
+
+        foreach (var child in candidates)
+        {
+            using var cmd = _master.CreateCommand();
+            cmd.CommandText = $"PRAGMA foreign_key_list({TriggerSqlGenerator.Ident(child)})";
+            using var reader = cmd.ExecuteReader();
+            while (reader.Read())
+            {
+                // foreign_key_list cols: 0=id, 1=seq, 2=table (the PARENT), 3=from, 4=to,
+                // 5=on_update, 6=on_delete, 7=match.
+                var parent = reader.IsDBNull(2) ? "" : reader.GetString(2);
+                // SQLite identifiers are case-insensitive, and the REFERENCES clause need not
+                // match the CREATE TABLE spelling.
+                if (!string.Equals(parent, tableName, StringComparison.OrdinalIgnoreCase))
+                    continue;
+
+                var onDelete = reader.IsDBNull(6) ? "" : reader.GetString(6);
+                if (!RowMutatingDeleteActions.Contains(onDelete))
+                    continue;
+
+                throw new LocalDbRegistrationException(
+                    $"Cannot register table '{tableName}': table '{child}' has a foreign key referencing it " +
+                    $"with ON DELETE {onDelete}. Replicated tables must not be cascade-delete foreign-key " +
+                    "parents: applying a remote row uses INSERT OR REPLACE, whose implicit DELETE fires the " +
+                    "referential action while change capture is suppressed — the child rows it touches are " +
+                    "never captured, so the peer never learns of them and the two nodes diverge silently. " +
+                    $"Drop the referential action on '{child}' (ON DELETE NO ACTION, removing children " +
+                    "explicitly instead), or do not replicate this table.");
+            }
+        }
+    }
+
+    /// <summary>
+    /// The <c>ON DELETE</c> actions that mutate referencing rows, as SQLite spells them in
+    /// <c>PRAGMA foreign_key_list</c>. NO ACTION / RESTRICT are absent on purpose — see
+    /// <see cref="EnsureNoRowMutatingDeleteAction"/>.
+    /// </summary>
+    private static readonly HashSet<string> RowMutatingDeleteActions =
+        new(["CASCADE", "SET NULL", "SET DEFAULT"], StringComparer.OrdinalIgnoreCase);
+
     /// <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

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

@@ -70,8 +70,12 @@ public sealed class GrpcAdapterTests : IAsyncLifetime
     }
 
     [Fact]
-    public async Task SecondConcurrentStream_Rejected()
+    public async Task SecondConcurrentStream_DisplacesTheIncumbent()
     {
+        // Newest wins (0.3.2). The guard serializes sessions; it was never a fence — the auth
+        // interceptor is, and an authenticated same-key dialler IS the peer. Refusing the newcomer
+        // bought nothing and cost the pair its only route back from a half-open link, where the
+        // incumbent handler can hold the guard indefinitely on a dead stream.
         var serverHost = await BuildServerAsync(NewDbPath(), apiKey: null, sink: null);
         var serverDb = serverHost.Services.GetRequiredService<ILocalDb>();
         var handler = serverHost.GetTestServer().CreateHandler();
@@ -85,11 +89,15 @@ public sealed class GrpcAdapterTests : IAsyncLifetime
 
         // Convergence proves the first stream's server handler is active and holding the single-stream guard.
         await WaitForAsync(async () => (await ReadOrders(serverDb)).Count == 1, RunTimeout);
+        var attemptsBefore = bg.ConnectionAttempts;
 
         var client = new LocalDbSync.LocalDbSyncClient(ChannelOver(handler));
         using var call2 = client.Sync(cancellationToken: cts.Token);
-        var ex = await Assert.ThrowsAsync<RpcException>(() => call2.ResponseStream.MoveNext(cts.Token));
-        Assert.Equal(StatusCode.AlreadyExists, ex.StatusCode);
+        Assert.True(await call2.ResponseStream.MoveNext(cts.Token));
+        Assert.Equal(SyncMessage.MsgOneofCase.Handshake, call2.ResponseStream.Current.MsgCase);
+
+        // The incumbent really ended: the displaced client notices and dials again.
+        await WaitForAsync(() => Task.FromResult(bg.ConnectionAttempts > attemptsBefore), RunTimeout);
 
         await bg.StopAsync(CancellationToken.None);
     }
@@ -262,9 +270,13 @@ public sealed class GrpcAdapterTests : IAsyncLifetime
             TombstoneRetention = TimeSpan.Zero,
             ReconnectBackoffMax = TimeSpan.Zero,
             MaxHlcDriftAhead = TimeSpan.Zero,
+            KeepAlivePingDelay = TimeSpan.Zero,
+            KeepAlivePingTimeout = TimeSpan.Zero,
         });
 
         Assert.True(result.Failed);
+        Assert.Contains(result.Failures, f => f.Contains("KeepAlivePingDelay", StringComparison.Ordinal));
+        Assert.Contains(result.Failures, f => f.Contains("KeepAlivePingTimeout", StringComparison.Ordinal));
         Assert.Contains(result.Failures, f => f.Contains("FlushInterval", StringComparison.Ordinal));
         Assert.Contains(result.Failures, f => f.Contains("MaxBatchSize", StringComparison.Ordinal));
         Assert.Contains(result.Failures, f => f.Contains("MaxOplogRows", StringComparison.Ordinal));

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

@@ -427,9 +427,12 @@ public sealed class LwwApplierTests : IDisposable
         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);
+        // Six table-independent statements (applying guard, version lookup, row-version write,
+        // peer state, plus the per-entry savepoint's SAVEPOINT and RELEASE), an upsert per table
+        // touched, and a delete for the one table tombstoned. The savepoint's ROLLBACK TO and the
+        // dead-letter insert are absent because this batch has no poison entry — statements are
+        // built on first use, so the count tracks the statement KINDS the batch actually ran.
+        Assert.Equal(9, applier.LastBatchStatementCount);
 
         // ...and the batch is still correct, statement reuse notwithstanding.
         Assert.Null(await ReadOrder(db, 1));
@@ -462,6 +465,72 @@ public sealed class LwwApplierTests : IDisposable
         Assert.Equal(("LAST", (long?)5), await ReadOrder(db, 5));
     }
 
+    [Fact]
+    public async Task Apply_LedgerWriteFails_LeavesNeitherTheDeleteNorTheLedgerEntry()
+    {
+        // The hazard (P4-7): applying a tombstone is TWO statements — delete the row, then write the
+        // ledger's tombstone version — and the poison catch commits whatever the first one already
+        // did. A failure landing between them therefore committed a DELETED ROW carrying a STALE LIVE
+        // ledger entry, which is exactly the shape SnapshotStreamer warns about and skips: the key
+        // then never travels again and the pair is divergent for that row, permanently and silently.
+        //
+        // The failure is injected with a trigger that aborts the ledger write (in production the
+        // realistic causes are disk-full / IO error / corruption — none of them reproducible here),
+        // and RAISE(ABORT) aborts only the statement, so the applier's narrow SqliteException catch
+        // treats it as per-entry poison exactly as it would a real one.
+        using var db = await NewOrdersDb();
+        await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'LOCAL', 5)");
+        var localHlc = await RowVersionHlc(db, Pk1);
+
+        await db.ExecuteAsync(
+            "CREATE TRIGGER poison_ledger BEFORE INSERT ON __localdb_row_version " +
+            "WHEN NEW.is_tombstone = 1 BEGIN SELECT RAISE(ABORT, 'injected ledger failure'); END");
+
+        var result = await new LwwApplier(db).ApplyBatchAsync(
+            [Entry(90, Pk1, null, localHlc + 1000, "remote-node", tombstone: true)], default);
+
+        // The entry is poison, so it is dead-lettered and the batch continues — unchanged semantics.
+        Assert.Equal(1, result.DeadLettered);
+        Assert.Equal(0, result.Applied);
+
+        // BOTH or NEITHER. Neither, here: the row must still be present, because its ledger entry
+        // still says "live at the local version".
+        Assert.Equal(("LOCAL", (long?)5), await ReadOrder(db, 1));
+        var rv = await db.QueryAsync(
+            "SELECT hlc, is_tombstone FROM __localdb_row_version WHERE table_name = 'orders' AND pk_json = @pk",
+            x => (Hlc: x.GetInt64(0), Tomb: x.GetInt64(1)), new { pk = Pk1 });
+        Assert.Single(rv);
+        Assert.Equal(localHlc, rv[0].Hlc);
+        Assert.Equal(0, rv[0].Tomb);
+    }
+
+    [Fact]
+    public async Task Apply_LedgerWriteFails_MidBatch_LeavesTheOtherEntriesApplied()
+    {
+        // The atomicity must be per-ENTRY, not per-batch: rolling back to the savepoint may not
+        // discard the rows applied before the poison one, nor block the ones after it.
+        using var db = await NewOrdersDb();
+        await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (2, 'DOOMED', 2)");
+        var doomedHlc = await RowVersionHlc(db, "{\"id\":2}");
+
+        await db.ExecuteAsync(
+            "CREATE TRIGGER poison_ledger BEFORE INSERT ON __localdb_row_version " +
+            "WHEN NEW.is_tombstone = 1 BEGIN SELECT RAISE(ABORT, 'injected ledger failure'); END");
+
+        var result = await new LwwApplier(db).ApplyBatchAsync(
+        [
+            Entry(91, Pk1, "{\"id\":1,\"sku\":\"BEFORE\",\"qty\":1}", doomedHlc + 1, "remote-node"),
+            Entry(92, "{\"id\":2}", null, doomedHlc + 1000, "remote-node", tombstone: true),
+            Entry(93, "{\"id\":3}", "{\"id\":3,\"sku\":\"AFTER\",\"qty\":3}", doomedHlc + 3, "remote-node"),
+        ], default);
+
+        Assert.Equal(2, result.Applied);
+        Assert.Equal(1, result.DeadLettered);
+        Assert.Equal(("BEFORE", (long?)1), await ReadOrder(db, 1));
+        Assert.Equal(("DOOMED", (long?)2), await ReadOrder(db, 2));
+        Assert.Equal(("AFTER", (long?)3), await ReadOrder(db, 3));
+    }
+
     [Fact]
     public async Task Apply_ReplaceExistingRow_DoesNotFireCaptureTriggers()
     {

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

@@ -70,6 +70,97 @@ public sealed class RegistrationTests : IDisposable
         Assert.Contains("BLOB", ex.Message);
     }
 
+    // ---- Cascading-delete foreign-key parents (P4-7) --------------------------------------
+
+    [Fact]
+    public async Task Register_CascadeDeleteFkParent_Throws()
+    {
+        // The hazard: applying a remote row uses INSERT OR REPLACE, whose implicit DELETE fires
+        // ON DELETE CASCADE while the capture guard is RAISED — so the cascaded child rows are
+        // deleted locally and never captured, and the peer keeps them forever.
+        using var db = Create();
+        await CreateOrders(db);
+        await db.ExecuteAsync(
+            "CREATE TABLE order_lines (id INTEGER PRIMARY KEY, order_id INTEGER " +
+            "REFERENCES orders(id) ON DELETE CASCADE)");
+
+        var ex = Assert.Throws<LocalDbRegistrationException>(() => db.RegisterReplicated("orders"));
+        Assert.Contains("orders", ex.Message);
+        Assert.Contains("order_lines", ex.Message);
+        Assert.Contains("ON DELETE CASCADE", ex.Message);
+
+        // Refused means refused: no triggers installed, nothing in the registry.
+        Assert.Empty(await Triggers(db));
+        Assert.False(db.ReplicatedTables.ContainsKey("orders"));
+    }
+
+    [Fact]
+    public async Task Register_CascadeDeleteFkChild_IsAllowed()
+    {
+        // Only the PARENT is refused. The child side is safe: replacing a child row cascades
+        // nothing, and the child's own deletes are captured normally.
+        using var db = Create();
+        await CreateOrders(db);
+        await db.ExecuteAsync(
+            "CREATE TABLE order_lines (id INTEGER PRIMARY KEY, order_id INTEGER " +
+            "REFERENCES orders(id) ON DELETE CASCADE)");
+
+        db.RegisterReplicated("order_lines");
+
+        Assert.True(db.ReplicatedTables.ContainsKey("order_lines"));
+    }
+
+    [Theory]
+    [InlineData("ON DELETE SET NULL", "ON DELETE SET NULL")]
+    [InlineData("ON DELETE SET DEFAULT", "ON DELETE SET DEFAULT")]
+    public async Task Register_OtherRowMutatingDeleteActions_AlsoThrow(string clause, string expected)
+    {
+        // SET NULL / SET DEFAULT mutate child rows on the same implicit DELETE and are captured
+        // just as little — same divergence, so the same refusal.
+        using var db = Create();
+        await CreateOrders(db);
+        await db.ExecuteAsync(
+            "CREATE TABLE order_lines (id INTEGER PRIMARY KEY, order_id INTEGER DEFAULT 0 " +
+            $"REFERENCES orders(id) {clause})");
+
+        var ex = Assert.Throws<LocalDbRegistrationException>(() => db.RegisterReplicated("orders"));
+        Assert.Contains(expected, ex.Message);
+    }
+
+    [Fact]
+    public async Task Register_NonCascadingFkParent_IsAllowed()
+    {
+        // The default (NO ACTION / RESTRICT) mutates no child row: the implicit DELETE simply
+        // fails the statement, which the applier already surfaces as a dead-lettered entry. Noisy,
+        // not silent — so there is nothing to refuse.
+        using var db = Create();
+        await CreateOrders(db);
+        await db.ExecuteAsync(
+            "CREATE TABLE order_lines (id INTEGER PRIMARY KEY, order_id INTEGER REFERENCES orders(id))");
+        await db.ExecuteAsync("CREATE TABLE restricted (id INTEGER PRIMARY KEY, " +
+            "order_id INTEGER REFERENCES orders(id) ON DELETE RESTRICT)");
+
+        db.RegisterReplicated("orders");
+
+        Assert.Equal(3, (await Triggers(db)).Count);
+    }
+
+    [Fact]
+    public async Task Register_UnrelatedTablesWithCascades_DoNotBlockRegistration()
+    {
+        // The scan must key on the table being registered, not on "the database contains a cascade".
+        using var db = Create();
+        await CreateOrders(db);
+        await db.ExecuteAsync("CREATE TABLE widgets (id INTEGER PRIMARY KEY, name TEXT)");
+        await db.ExecuteAsync(
+            "CREATE TABLE widget_parts (id INTEGER PRIMARY KEY, widget_id INTEGER " +
+            "REFERENCES widgets(id) ON DELETE CASCADE)");
+
+        db.RegisterReplicated("orders");
+
+        Assert.True(db.ReplicatedTables.ContainsKey("orders"));
+    }
+
     [Fact]
     public async Task Register_IsIdempotent_RegeneratesTriggers()
     {

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

@@ -125,11 +125,15 @@ 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, bool supportsPartialSnapshot = false)
+        CancellationToken ct, bool supportsPartialSnapshot = false, long lastAppliedRemoteSeq = 0)
     {
         await NextOfCaseAsync(fromSession, SyncMessage.MsgOneofCase.Handshake, ct);
         await toSession.WriteAsync(
-            new SyncMessage { Handshake = MatchingHandshake(sessionDb, "peer", supportsPartialSnapshot: supportsPartialSnapshot) }, ct);
+            new SyncMessage
+            {
+                Handshake = MatchingHandshake(
+                    sessionDb, "peer", lastAppliedRemoteSeq, supportsPartialSnapshot),
+            }, ct);
         await NextOfCaseAsync(fromSession, SyncMessage.MsgOneofCase.HandshakeAck, ct);
         await toSession.WriteAsync(new SyncMessage { HandshakeAck = new HandshakeAck { NodeId = "peer", SnapshotRequired = false } }, ct);
     }
@@ -373,6 +377,93 @@ public sealed class SnapshotResyncTests : IDisposable
         await SwallowAsync(runB);
     }
 
+    [Fact]
+    public async Task CapPrune_IntoALiveSessionsUnsentRange_FaultsIt_AndTheReconnectSnapshots()
+    {
+        // P4-1. The pump reads `seq > sentThruSeq`, which resumes SILENTLY above a hole a cap prune
+        // punched — so it keeps shipping later deltas, the peer acks them, and the pair is durably
+        // divergent. `needs_snapshot` (set in the prune's own transaction) heals only a RECONNECT,
+        // and nothing forced one: on a stable LAN with a slow-but-alive peer the session, and the
+        // divergence, could last for days. Faulting is what turns the persisted flag into a snapshot.
+        var a = await NewSideAsync();
+        for (var i = 1; i <= 6; i++)
+            await a.Db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (@id, 'A', @id)", new { id = i });
+
+        var (duplex, toSession, fromSession) = ScriptedPeer();
+        using var cts = new CancellationTokenSource(RunTimeout);
+        var run = a.Session.RunAsync(duplex, cts.Token);
+
+        // The peer holds everything so far: no snapshot is owed, and the pump idles at seq 6.
+        await DriveHandshakeAsync(a.Db, toSession, fromSession, cts.Token, lastAppliedRemoteSeq: 6);
+
+        // Write three more rows and prune all but the newest, in ONE transaction — which is what a
+        // cap prune is, and what makes this deterministic: the pump's reader can only ever see the
+        // before state (nothing above seq 6) or the after state (seqs 7 and 8 already gone). It can
+        // never catch rows 7/8 in between and legitimately advance past them.
+        await using (var tx = await a.Db.BeginTransactionAsync(cts.Token))
+        {
+            await tx.ExecuteAsync(
+                "INSERT INTO orders (id, sku, qty) VALUES (7,'PRUNED',7),(8,'PRUNED',8),(9,'KEPT',9)", null, cts.Token);
+            await tx.ExecuteAsync("UPDATE __localdb_peer_state SET needs_snapshot = 1 WHERE id = 1", null, cts.Token);
+            await tx.ExecuteAsync(
+                "DELETE FROM __localdb_oplog WHERE seq < (SELECT MAX(seq) FROM __localdb_oplog)", null, cts.Token);
+            await tx.CommitAsync(cts.Token);
+        }
+
+        var fault = await Assert.ThrowsAsync<LocalDbOplogGapException>(() => run);
+        Assert.Contains("6", fault.Message, StringComparison.Ordinal);
+
+        // And it faulted BEFORE shipping anything from above the hole: a delta for seq 9 would have
+        // moved the peer's watermark past rows 7 and 8, which is the divergence itself.
+        while (fromSession.TryRead(out var stray))
+            Assert.NotEqual(SyncMessage.MsgOneofCase.DeltaBatch, stray.MsgCase);
+
+        // The reconnect is what heals: the prune's needs_snapshot makes the handshake owe a full
+        // snapshot, which carries the pruned rows' current state to a real peer.
+        var a2 = NewSessionOver(a.Db);
+        var b = await NewSideAsync();
+        var (da, db) = DuplexPair();
+        var run2A = a2.Session.RunAsync(da, cts.Token);
+        var run2B = b.Session.RunAsync(db, cts.Token);
+
+        await WaitForAsync(async () => (await ReadOrders(b.Db)).Count == 9, RunTimeout);
+        Assert.Equal(await ReadOrders(a.Db), await ReadOrders(b.Db));
+
+        cts.Cancel();
+        await SwallowAsync(run2A);
+        await SwallowAsync(run2B);
+    }
+
+    [Fact]
+    public async Task AckPruningToEmpty_DoesNotFaultALiveSession()
+    {
+        // The other half of the horizon check: ordinary ack-pruning deletes seq <= last_acked_seq,
+        // which by construction sits at or below the sent watermark, so a converged pair that has
+        // pruned its whole oplog must keep streaming. A gap test written against `needs_snapshot`,
+        // or one that treated any risen horizon as a fault, would kill healthy sessions here.
+        var a = await NewSideAsync();
+        var b = await NewSideAsync();
+        for (var i = 1; i <= 4; i++)
+            await a.Db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (@id, 'A', @id)", new { id = i });
+
+        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);
+
+        await WaitForAsync(async () => (await ReadOrders(b.Db)).Count == 4, RunTimeout);
+        await WaitForAsync(async () => await OplogCount(a.Db) == 0, RunTimeout);
+
+        // Still live, still incremental: a post-prune write reaches B as an ordinary delta.
+        await a.Db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (5, 'POST', 5)");
+        await WaitForAsync(async () => (await ReadOrders(b.Db)).Any(r => r.Id == 5), RunTimeout);
+        Assert.False(runA.IsFaulted, runA.Exception?.ToString());
+
+        cts.Cancel();
+        await SwallowAsync(runA);
+        await SwallowAsync(runB);
+    }
+
     [Fact]
     public async Task Snapshot_MergesNotTruncates()
     {

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

@@ -0,0 +1,181 @@
+using System.Threading.Channels;
+using Grpc.Core;
+using Microsoft.Extensions.Logging.Abstractions;
+using Microsoft.Extensions.Options;
+using ZB.MOM.WW.LocalDb.Contracts;
+using ZB.MOM.WW.LocalDb.Replication;
+
+namespace ZB.MOM.WW.LocalDb.Tests;
+
+/// <summary>
+/// The passive service's single-stream guard under a HALF-OPEN link: the dialling node is gone, but
+/// nothing on this side knows it, so the handler sits inside <c>MoveNext</c> forever holding the
+/// guard. Through 0.3.1 that made the pair unrecoverable without a restart — every reconnect from
+/// the rebooted peer was refused <c>AlreadyExists</c>, indefinitely on an idle pair, while this node
+/// went on reporting itself connected. The guard is there to serialize sessions, not to fence out a
+/// caller the auth interceptor already vouched for, so the newest dialler wins.
+/// </summary>
+public sealed class SyncStreamGuardTests : IDisposable
+{
+    private static readonly TimeSpan Timeout = TimeSpan.FromSeconds(10);
+
+    private readonly List<string> _paths = [];
+    private readonly List<IDisposable> _disposables = [];
+
+    public void Dispose()
+    {
+        foreach (var d in _disposables)
+            d.Dispose();
+        Microsoft.Data.Sqlite.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 LocalDbSyncService NewService()
+    {
+        var path = Path.Combine(Path.GetTempPath(), Guid.NewGuid() + ".db");
+        _paths.Add(path);
+        var db = new Internal.SqliteLocalDb(new LocalDbOptions { Path = path });
+        _disposables.Add(db);
+        db.ExecuteAsync("CREATE TABLE orders (id INTEGER PRIMARY KEY, sku TEXT)").GetAwaiter().GetResult();
+        db.RegisterReplicated("orders");
+
+        return new LocalDbSyncService(
+            db,
+            Options.Create(new ReplicationOptions { FlushInterval = TimeSpan.FromMilliseconds(20) }),
+            NullLoggerFactory.Instance);
+    }
+
+    private static ServerCallContext ContextFor(CancellationToken ct) => new FakeCallContext(ct);
+
+    [Fact]
+    public async Task WedgedIncumbent_IsDisplacedByANewStream()
+    {
+        var service = NewService();
+        // Eviction is normally generous; the test only needs it to not be the thing under test.
+        service.TakeoverTimeout = Timeout;
+
+        // The incumbent's peer crashed: its stream will never deliver another message, and neither
+        // end times out on silence — a converged pair is legitimately quiet. The handler parks in
+        // the handshake's first read, holding the guard.
+        using var incumbentCts = new CancellationTokenSource();
+        var incumbentOut = new RecordingWriter();
+        var incumbent = service.Sync(new SilentReader(), incumbentOut, ContextFor(incumbentCts.Token));
+
+        // It really is running and really did claim the guard (its handshake is on the wire).
+        await incumbentOut.NextAsync(Timeout);
+        Assert.False(incumbent.IsCompleted);
+
+        // The rebooted peer dials again. Pre-fix this threw AlreadyExists and the pair stayed wedged.
+        using var newcomerCts = new CancellationTokenSource();
+        var newcomerOut = new RecordingWriter();
+        var newcomer = service.Sync(new SilentReader(), newcomerOut, ContextFor(newcomerCts.Token));
+
+        // The incumbent is torn down — gracefully, not as a fault of this node.
+        await incumbent.WaitAsync(Timeout);
+        // ...and the new stream is genuinely served: its own handshake goes out.
+        var handshake = await newcomerOut.NextAsync(Timeout);
+        Assert.Equal(SyncMessage.MsgOneofCase.Handshake, handshake.MsgCase);
+        Assert.False(newcomer.IsCompleted);
+
+        newcomerCts.Cancel();
+        await SwallowAsync(newcomer);
+    }
+
+    [Fact]
+    public async Task DisplacedIncumbent_ReleasesTheGuard_SoAThirdStreamIsAlsoServed()
+    {
+        // Takeover must leave the guard in a state that can be taken over AGAIN — a handover that
+        // leaked the slot would simply move the wedge one reconnect down the line.
+        var service = NewService();
+        service.TakeoverTimeout = Timeout;
+
+        using var firstCts = new CancellationTokenSource();
+        var firstOut = new RecordingWriter();
+        var first = service.Sync(new SilentReader(), firstOut, ContextFor(firstCts.Token));
+        await firstOut.NextAsync(Timeout);
+
+        using var secondCts = new CancellationTokenSource();
+        var secondOut = new RecordingWriter();
+        var second = service.Sync(new SilentReader(), secondOut, ContextFor(secondCts.Token));
+        await first.WaitAsync(Timeout);
+        await secondOut.NextAsync(Timeout);
+
+        using var thirdCts = new CancellationTokenSource();
+        var thirdOut = new RecordingWriter();
+        var third = service.Sync(new SilentReader(), thirdOut, ContextFor(thirdCts.Token));
+        await second.WaitAsync(Timeout);
+        Assert.Equal(SyncMessage.MsgOneofCase.Handshake, (await thirdOut.NextAsync(Timeout)).MsgCase);
+
+        thirdCts.Cancel();
+        await SwallowAsync(third);
+    }
+
+    private static async Task SwallowAsync(Task task)
+    {
+        try { await task.WaitAsync(Timeout); }
+        catch (OperationCanceledException) { }
+        catch (RpcException) { }
+    }
+
+    /// <summary>
+    /// The minimum <see cref="ServerCallContext"/> the handler actually reads: its cancellation
+    /// token. (Grpc.Core.Api ships no test double, and the ASP.NET Core one is not reachable outside
+    /// a live call.)
+    /// </summary>
+    private sealed class FakeCallContext(CancellationToken ct) : ServerCallContext
+    {
+        protected override string MethodCore => "/localdb_sync.v1.LocalDbSync/Sync";
+        protected override string HostCore => "test";
+        protected override string PeerCore => "test";
+        protected override DateTime DeadlineCore => DateTime.MaxValue;
+        protected override Metadata RequestHeadersCore => [];
+        protected override CancellationToken CancellationTokenCore => ct;
+        protected override Metadata ResponseTrailersCore { get; } = [];
+        protected override Status StatusCore { get; set; }
+        protected override WriteOptions? WriteOptionsCore { get; set; }
+        protected override AuthContext AuthContextCore => new(null, new Dictionary<string, List<AuthProperty>>());
+
+        protected override ContextPropagationToken CreatePropagationTokenCore(ContextPropagationOptions? options) =>
+            throw new NotSupportedException();
+
+        protected override Task WriteResponseHeadersAsyncCore(Metadata responseHeaders) => Task.CompletedTask;
+    }
+
+    /// <summary>A stream that never delivers anything and never ends — a half-open link.</summary>
+    private sealed class SilentReader : IAsyncStreamReader<SyncMessage>
+    {
+        public SyncMessage Current => throw new InvalidOperationException("no message");
+
+        public async Task<bool> MoveNext(CancellationToken cancellationToken)
+        {
+            await Task.Delay(System.Threading.Timeout.Infinite, cancellationToken);
+            return false;
+        }
+    }
+
+    private sealed class RecordingWriter : IServerStreamWriter<SyncMessage>
+    {
+        private readonly Channel<SyncMessage> _written = Channel.CreateUnbounded<SyncMessage>();
+
+        public WriteOptions? WriteOptions { get; set; }
+
+        public Task WriteAsync(SyncMessage message) => WriteAsync(message, CancellationToken.None);
+
+        public Task WriteAsync(SyncMessage message, CancellationToken cancellationToken)
+        {
+            _written.Writer.TryWrite(message);
+            return Task.CompletedTask;
+        }
+
+        public async Task<SyncMessage> NextAsync(TimeSpan timeout)
+        {
+            using var cts = new CancellationTokenSource(timeout);
+            return await _written.Reader.ReadAsync(cts.Token);
+        }
+    }
+}