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.
docs(index): LocalDb 0.3.2 published — cap-prune session fault, sync keepalives, newest-wins stream guard (ScadaBridge arch-review P4-1 + P4-3); ScadaBridge pins 0.3.2
2026-08-19 16:22:44 -04:00 · 2026-08-19 15:06:47 -04:00 · 2026-08-19 15:02:04 -04:00 · 2026-08-18 07:30:59 -04:00 · 2026-08-18 07:25:18 -04:00 · 2026-08-18 07:11:14 -04:00
28 changed files with 1076 additions and 44 deletions
@@ -1,4 +1,10 @@
 #!/usr/bin/env bash
 # pack.sh — produce the ZB.MOM.WW.Audit NuGet package into ./artifacts.
 set -euo pipefail
-dotnet pack -c Release -o ./artifacts
+# cd to the repo root and name the solution + an absolute output path: `dotnet pack` with
+# neither a project argument nor a fixed working directory packs whatever solution happens
+# to sit in the CALLER's directory (the bug behind the LocalDb feed incident, 0038d46).
+REPO_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+cd "$REPO_ROOT"
+
+dotnet pack ZB.MOM.WW.Audit.slnx -c Release -o "$REPO_ROOT/artifacts"
@@ -16,9 +16,20 @@ set -euo pipefail
 : "${GITEA_NUGET_SOURCE:?set GITEA_NUGET_SOURCE to your Gitea NuGet feed URL}"
 : "${GITEA_NUGET_KEY:?set GITEA_NUGET_KEY to your Gitea access token}"

-dotnet pack -c Release -o ./artifacts
+# Named solution, named output, and a cd to the repo root, all for one reason: `dotnet pack`
+# with neither a project argument nor a fixed working directory packs whatever solution
+# happens to sit in the CALLER's directory — and then pushes it to a live feed under this
+# script's name. Invoked by absolute path from another repo's checkout, the relative form
+# packed and published 25 unrelated packages (LocalDb's copy of this script, fixed in
+# 0038d46; recovered with `dotnet nuget delete`, but only because the token happened to
+# carry delete rights). A publish script must be unambiguous about WHAT it publishes no
+# matter where it is run.
+REPO_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+cd "$REPO_ROOT"

-dotnet nuget push "./artifacts/*.nupkg" \
+dotnet pack ZB.MOM.WW.Audit.slnx -c Release -o "$REPO_ROOT/artifacts"
+
+dotnet nuget push "$REPO_ROOT/artifacts/*.nupkg" \
  --source "$GITEA_NUGET_SOURCE" \
  --api-key "$GITEA_NUGET_KEY" \
  --skip-duplicate
@@ -6,4 +6,10 @@

 set -euo pipefail

-dotnet pack -c Release -o ./artifacts
+# cd to the repo root and name the solution + an absolute output path: `dotnet pack` with
+# neither a project argument nor a fixed working directory packs whatever solution happens
+# to sit in the CALLER's directory (the bug behind the LocalDb feed incident, 0038d46).
+REPO_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+cd "$REPO_ROOT"
+
+dotnet pack ZB.MOM.WW.Auth.slnx -c Release -o "$REPO_ROOT/artifacts"
@@ -16,9 +16,20 @@ set -euo pipefail
 : "${GITEA_NUGET_SOURCE:?set GITEA_NUGET_SOURCE to your Gitea NuGet feed URL}"
 : "${GITEA_NUGET_KEY:?set GITEA_NUGET_KEY to your Gitea access token}"

-dotnet pack -c Release -o ./artifacts
+# Named solution, named output, and a cd to the repo root, all for one reason: `dotnet pack`
+# with neither a project argument nor a fixed working directory packs whatever solution
+# happens to sit in the CALLER's directory — and then pushes it to a live feed under this
+# script's name. Invoked by absolute path from another repo's checkout, the relative form
+# packed and published 25 unrelated packages (LocalDb's copy of this script, fixed in
+# 0038d46; recovered with `dotnet nuget delete`, but only because the token happened to
+# carry delete rights). A publish script must be unambiguous about WHAT it publishes no
+# matter where it is run.
+REPO_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+cd "$REPO_ROOT"

-dotnet nuget push "./artifacts/*.nupkg" \
+dotnet pack ZB.MOM.WW.Auth.slnx -c Release -o "$REPO_ROOT/artifacts"
+
+dotnet nuget push "$REPO_ROOT/artifacts/*.nupkg" \
  --source "$GITEA_NUGET_SOURCE" \
  --api-key "$GITEA_NUGET_KEY" \
  --skip-duplicate
@@ -5,7 +5,7 @@
    <Nullable>enable</Nullable>
    <ImplicitUsings>enable</ImplicitUsings>
    <LangVersion>latest</LangVersion>
-    <Version>0.3.0</Version>
+    <Version>0.3.3</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.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
@@ -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(
@@ -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)
    {
@@ -27,6 +27,10 @@ internal sealed class SyncSession
    // inbound batch rate (the peer's flow control), so its lane can safely be unbounded.
    private const int DataLaneCapacity = 16;

+    // How long a faulting session's teardown lets the writer keep flushing already-enqueued
+    // messages before the backstop cancel; only ever reached when the transport itself stalls.
+    private static readonly TimeSpan FaultFlushGrace = TimeSpan.FromSeconds(2);
+
    private readonly SqliteLocalDb _db;
    private readonly OplogStore _store;
    private readonly LwwApplier _applier;
@@ -140,7 +144,20 @@ internal sealed class SyncSession
        {
            // Handshake-phase faults (and the primary rethrow above) land here: the writer may
            // still be running, so stop and observe it before the primary exception surfaces.
-            linked.Cancel();
+            // It must be allowed to DRAIN, not be cancelled outright: the Handshake enqueued at
+            // the top of HandshakeAsync can still be sitting in the control lane (the enqueue →
+            // receive → validate → fault path has no yields, while the writer's wakeup waits in
+            // the thread-pool queue), and that message is the only thing that lets the peer
+            // compute its own fail-closed verdict — mismatch detection is symmetric, so this
+            // node's locally-thrown verdict reaches the peer only as the Handshake it transmits.
+            // Cancelling first killed it almost every time the peer's handshake was already
+            // buffered, leaving the healthy peer to misread a schema mismatch as a generic
+            // disconnect and reconnect-loop at the fast backoff. Completing the lanes lets the
+            // writer send what is already enqueued and exit on its own; the deferred cancel is
+            // the backstop that keeps a wedged transport from holding teardown hostage.
+            control.Writer.TryComplete();
+            data.Writer.TryComplete();
+            linked.CancelAfter(FaultFlushGrace);
            await DrainSecondaryAsync(writer);
            throw;
        }
@@ -339,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)
                {
@@ -373,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.
@@ -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) { }
+}
@@ -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"));
        }
    }
 }
@@ -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)
                {
@@ -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);

@@ -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)
@@ -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);
@@ -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>
@@ -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
@@ -70,6 +70,20 @@ public sealed class BatchBudgetTests : IDisposable
        return msg!;
    }

+    // Reads the next scripted-peer message while also observing the session task: a session that
+    // faults would otherwise leave this wait to run out the whole test timeout, reporting an
+    // opaque cancellation instead of the session's real exception.
+    private static async Task<SyncMessage> NextAsync(Task run, ChannelReader<SyncMessage> reader, CancellationToken ct)
+    {
+        var next = NextAsync(reader, ct);
+        if (ReferenceEquals(await Task.WhenAny(run, next), run))
+        {
+            await run;
+            throw new InvalidOperationException("The sync session ended before the expected message arrived.");
+        }
+        return await next;
+    }
+
    private static async Task SwallowAsync(Task task)
    {
        try { await task; }
@@ -140,7 +154,7 @@ public sealed class BatchBudgetTests : IDisposable
        var seen = new List<long>();
        while (seen.Count < 20)
        {
-            var msg = await NextAsync(fromSession, cts.Token);
+            var msg = await NextAsync(run, fromSession, cts.Token);
            if (msg.MsgCase != SyncMessage.MsgOneofCase.DeltaBatch)
                continue;
            Assert.True(BatchBytes(msg.DeltaBatch) <= budget,
@@ -180,7 +194,7 @@ public sealed class BatchBudgetTests : IDisposable
        var batches = new List<DeltaBatch>();
        while (batches.Sum(b => b.Entries.Count) < 2)
        {
-            var msg = await NextAsync(fromSession, cts.Token);
+            var msg = await NextAsync(run, fromSession, cts.Token);
            if (msg.MsgCase == SyncMessage.MsgOneofCase.DeltaBatch)
                batches.Add(msg.DeltaBatch);
        }
@@ -220,7 +234,7 @@ public sealed class BatchBudgetTests : IDisposable
        var seen = 0;
        while (seen < 12)
        {
-            var msg = await NextAsync(fromSession, cts.Token);
+            var msg = await NextAsync(run, fromSession, cts.Token);
            if (msg.MsgCase != SyncMessage.MsgOneofCase.DeltaBatch)
                continue;
            Assert.True(msg.DeltaBatch.Entries.Count <= 5);
@@ -312,7 +326,7 @@ public sealed class BatchBudgetTests : IDisposable
        }

        SyncMessage ack;
-        do { ack = await NextAsync(fromSession, cts.Token); }
+        do { ack = await NextAsync(run, fromSession, cts.Token); }
        while (ack.MsgCase != SyncMessage.MsgOneofCase.DeltaAck);

        // The discriminator: uncoalesced, the FIRST ack could only cover seq 1.
@@ -348,7 +362,7 @@ public sealed class BatchBudgetTests : IDisposable
            new SyncMessage { SnapshotBegin = new SnapshotBegin { AsOfSeq = 77 } }, cts.Token);

        SyncMessage ack;
-        do { ack = await NextAsync(fromSession, cts.Token); }
+        do { ack = await NextAsync(run, fromSession, cts.Token); }
        while (ack.MsgCase != SyncMessage.MsgOneofCase.DeltaAck);
        Assert.Equal(1, ack.DeltaAck.AppliedThruSeq);

@@ -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));
@@ -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()
    {
@@ -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()
    {
@@ -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()
    {
@@ -208,6 +208,50 @@ public sealed class SyncSessionTests : IDisposable
        Assert.Empty(await ReadOrders(b.Db));
    }

+    [Fact]
+    public async Task Handshake_FaultTeardown_StillTransmitsTheLocalHandshake()
+    {
+        // Schema-mismatch detection is symmetric: each side computes its verdict FROM the other's
+        // Handshake. So a session that faults on the peer's handshake must still flush its own —
+        // otherwise the healthy peer learns nothing and waits out its receive cap (tests) or
+        // misclassifies the mismatch as a generic disconnect and reconnect-loops (production).
+        //
+        // The race being pinned: with the peer's handshake already buffered, the path from
+        // enqueueing our Handshake on the control lane to the validation throw has no yields,
+        // while the writer's wakeup sits in the thread-pool queue — a teardown that cancels the
+        // writer can therefore kill the Handshake before it ever transmits. Preloading makes the
+        // sprint as tight as possible, and running the scenarios concurrently starves the pool
+        // the way a busy test suite does.
+        const int Iterations = 40;
+        var sides = new List<Side>(Iterations);
+        for (var i = 0; i < Iterations; i++)
+            sides.Add(await NewSideAsync());
+
+        using var cts = new CancellationTokenSource(RunTimeout);
+        var runs = new List<(Task Run, ChannelReader<SyncMessage> FromSession)>(Iterations);
+        foreach (var side in sides)
+        {
+            var (duplex, toSession, fromSession) = ScriptedPeer();
+            var mismatched = MatchingHandshake(side.Db, "peer");
+            mismatched.Tables[0].Digest = "not-the-local-digest";
+            await toSession.WriteAsync(new SyncMessage { Handshake = mismatched }, cts.Token);
+            runs.Add((side.Session.RunAsync(duplex, cts.Token), fromSession));
+        }
+
+        var lost = 0;
+        foreach (var (run, fromSession) in runs)
+        {
+            await Assert.ThrowsAsync<LocalDbSchemaMismatchException>(() => run);
+            var sent = false;
+            while (fromSession.TryRead(out var msg))
+                sent |= msg.MsgCase == SyncMessage.MsgOneofCase.Handshake;
+            if (!sent)
+                lost++;
+        }
+
+        Assert.True(lost == 0, $"{lost}/{Iterations} sessions tore down without transmitting their own Handshake");
+    }
+
    [Fact]
    public async Task Handshake_SchemaVersionMismatch_FailsClosed()
    {
@@ -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);
+        }
+    }
+}
@@ -16,9 +16,20 @@ set -euo pipefail
 : "${GITEA_NUGET_SOURCE:?set GITEA_NUGET_SOURCE to your Gitea NuGet feed URL}"
 : "${GITEA_NUGET_KEY:?set GITEA_NUGET_KEY to your Gitea access token}"

-dotnet pack -c Release -o ./artifacts
+# Named solution, named output, and a cd to the repo root, all for one reason: `dotnet pack`
+# with neither a project argument nor a fixed working directory packs whatever solution
+# happens to sit in the CALLER's directory — and then pushes it to a live feed under this
+# script's name. Invoked by absolute path from another repo's checkout, the relative form
+# packed and published 25 unrelated packages (LocalDb's copy of this script, fixed in
+# 0038d46; recovered with `dotnet nuget delete`, but only because the token happened to
+# carry delete rights). A publish script must be unambiguous about WHAT it publishes no
+# matter where it is run.
+REPO_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+cd "$REPO_ROOT"

-dotnet nuget push "./artifacts/*.nupkg" \
+dotnet pack ZB.MOM.WW.Secrets.slnx -c Release -o "$REPO_ROOT/artifacts"
+
+dotnet nuget push "$REPO_ROOT/artifacts/*.nupkg" \
  --source "$GITEA_NUGET_SOURCE" \
  --api-key "$GITEA_NUGET_KEY" \
  --skip-duplicate
@@ -6,4 +6,10 @@

 set -euo pipefail

-dotnet pack -c Release -o ./artifacts
+# cd to the repo root and name the solution + an absolute output path: `dotnet pack` with
+# neither a project argument nor a fixed working directory packs whatever solution happens
+# to sit in the CALLER's directory (the bug behind the LocalDb feed incident, 0038d46).
+REPO_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+cd "$REPO_ROOT"
+
+dotnet pack ZB.MOM.WW.Theme.slnx -c Release -o "$REPO_ROOT/artifacts"
@@ -16,9 +16,20 @@ set -euo pipefail
 : "${GITEA_NUGET_SOURCE:?set GITEA_NUGET_SOURCE to your Gitea NuGet feed URL}"
 : "${GITEA_NUGET_KEY:?set GITEA_NUGET_KEY to your Gitea access token}"

-dotnet pack -c Release -o ./artifacts
+# Named solution, named output, and a cd to the repo root, all for one reason: `dotnet pack`
+# with neither a project argument nor a fixed working directory packs whatever solution
+# happens to sit in the CALLER's directory — and then pushes it to a live feed under this
+# script's name. Invoked by absolute path from another repo's checkout, the relative form
+# packed and published 25 unrelated packages (LocalDb's copy of this script, fixed in
+# 0038d46; recovered with `dotnet nuget delete`, but only because the token happened to
+# carry delete rights). A publish script must be unambiguous about WHAT it publishes no
+# matter where it is run.
+REPO_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+cd "$REPO_ROOT"

-dotnet nuget push "./artifacts/*.nupkg" \
+dotnet pack ZB.MOM.WW.Theme.slnx -c Release -o "$REPO_ROOT/artifacts"
+
+dotnet nuget push "$REPO_ROOT/artifacts/*.nupkg" \
  --source "$GITEA_NUGET_SOURCE" \
  --api-key "$GITEA_NUGET_KEY" \
  --skip-duplicate
Author	SHA1	Message	Date
Joseph Doherty	41bdc178c2	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.	2026-08-19 16:22:44 -04:00
Joseph Doherty	e9dc13689a	docs(index): LocalDb 0.3.2 published — cap-prune session fault, sync keepalives, newest-wins stream guard (ScadaBridge arch-review P4-1 + P4-3); ScadaBridge pins 0.3.2	2026-08-19 15:06:47 -04:00
Joseph Doherty	f62b017cc7	fix(localdb): fault live sync session on cap-prune; keepalive + newest-wins sync guard (0.3.2, ScadaBridge arch-review P4-1 + P4-3)	2026-08-19 15:02:04 -04:00
Joseph Doherty	dbcbc5bd41	docs(index): LocalDb 0.3.1 published — handshake-fault control-lane flush (register row 41)	2026-08-18 07:30:59 -04:00
Joseph Doherty	1145df682e	docs(index): OtOpcUa now registers LocalDb tables conditionally — baseline on enable, deregister self-heal on disable (row 39 closed)	2026-08-18 07:25:18 -04:00
Joseph Doherty	3e5a5fde81	fix(localdb): flush the control lane before a faulting session cancels its writer (0.3.1) A session that faulted during the handshake cancelled its writer loop immediately, which could kill its OWN Handshake while it still sat in the control lane: the enqueue -> receive -> validate -> fault path has no yields, while the writer's wakeup waits in the thread-pool queue, so with the peer's handshake already buffered the cancel won the race almost every time (39/40 in the new pinning test). Mismatch detection is symmetric — the peer computes its fail-closed verdict FROM our Handshake — so the healthy peer never learned anything: in tests it waited out the full 15 s receive cap (the register-row-41 flake on SyncSessionTests.Handshake_DigestMismatch_FailsClosed), and in production it would misclassify a permanent schema mismatch as a generic disconnect and reconnect-loop at the fast backoff instead of taking the classified FailedPrecondition -> LogError + max-backoff path. RunAsync's fault teardown now completes both lanes so the writer drains what is already enqueued and exits on its own, with a 2 s deferred cancel as the backstop for a wedged transport. Caller-initiated cancellation is unchanged (the linked token is already cancelled, so shutdown is not delayed), and nothing on the wire changes — a 0.3.0 peer simply receives the Handshake it was always owed. New SyncSessionTests.Handshake_FaultTeardown_StillTransmitsTheLocalHandshake pins the race (40 concurrent preloaded-mismatch sessions; 39/40 red pre-fix, green post-fix). BatchBudgetTests hardened separately: the register's other flaky test (Pump_RowLargerThanTheBudget_IsSentAloneRatherThanStallingTheStream) awaits the scripted-peer channel without ever observing the session task, so ANY session fault surfaced as an opaque 15 s timeout. A run-observing NextAsync overload now races the session task at every read site in the class, so the next flake reports the session's real exception instead. (The handshake-race root cause does not structurally apply to this test — no handshake fault occurs there — so its intermittent timeout is attributed to the fault-hiding wait, now removed.) Suite 200/200 green. Closes ScadaBridge deferred-work register row 41 (cross-repo).	2026-08-18 07:11:14 -04:00
Joseph Doherty	b62981ff80	fix(build): anchor pack/push scripts to their own repo root (`0038d46` pattern, all seven) Secrets/push.sh, Theme/pack.sh+push.sh, Audit/pack.sh+push.sh and Auth/pack.sh+push.sh all carried the CWD-pack bug LocalDb's push.sh was cured of in `0038d46`: a bare 'dotnet pack -c Release -o ./artifacts' packs whatever solution sits in the CALLER's working directory, and the push scripts then shove the caller's ./artifacts/*.nupkg onto the live Gitea feed (the incident that briefly published 25 unrelated packages). Each script now cds to its own repo root, names its .slnx explicitly, writes to an absolute artifacts path, and (push.sh) pushes an anchored glob. Verified by bash -n on all seven plus a positive control: Audit's pack.sh run from a foreign CWD packed only ZB.MOM.WW.Audit into its own artifacts dir and left the caller's directory untouched. Push scripts deliberately NOT executed (they publish); pack half verified only. Closes ScadaBridge deferred-work register row 42 (cross-repo).	2026-08-18 07:00:42 -04:00