fix(localdb): count snapshot dead-letters and cap held-back snapshot rows

- SnapshotApplier now takes LocalDbMetrics (wired by SyncSessionFactory) and
  adds every row it dead-letters - poison, drift reject, FK orphan, retry
  failure, cap spill - to localdb.sync.dead_lettered. Before, only the delta
  path recorded the metric.
- New ReplicationOptions.MaxSnapshotDeferredRows (default 100,000, validated
  > 0) bounds the rows held for a missing FK parent; past it the oldest get
  one more apply with deferral off and are dead-lettered if still orphaned.
  An older name-ordered sender could otherwise defer a whole child table.

Unit + 2-node (SyncSessionFactory) tests fail pre-fix. No wire change.

(scadaproj#50)

Claude-Session: https://claude.ai/code/session_01GT67AEHd2Acv1YPwGfXEoV

ZB.MOM.WW.LocalDb/README.md

@@ -243,6 +243,7 @@ node still pushes its own deltas back. Only one side needs `PeerAddress`.
 | `TombstoneRetention` | `7 days` | How long deleted-row tombstones are kept before pruning. |
 | `ReconnectBackoffMax` | `60 s` | Upper bound on exponential reconnect backoff (validator caps ≤ 1 day). |
 | `MaintenanceInterval` | `60 s` | How often maintenance enforces caps + prunes expired tombstones. |
+| `MaxSnapshotDeferredRows` | `100_000` | Ceiling on snapshot rows held in memory while their FK parent has not arrived (`0.3.6`). Past it the oldest are dead-lettered (and counted). |
 | `MaxHlcDriftAhead` | `5 min` | Max tolerated HLC drift where the peer's clock runs ahead of ours. |
 | `FailClosedOnDrift` | `false` | When true, an over-drift entry is dead-lettered instead of applied. |
 
@@ -327,7 +328,10 @@ All options are validated at startup (`ZB.MOM.WW.Configuration` validators).
   sorts later as `pk_json` text) was dead-lettered for good. **Mixed pairs:** a pre-`0.3.6`
   receiver still loses self-referencing rows that arrive child-first, and a pre-`0.3.6`
   sender's name order relies on the `0.3.6` receiver's deferral — so upgrade both nodes before
-  adding FK-linked tables. Deferred rows are held in memory until the snapshot completes.
+  adding FK-linked tables. Deferred rows are held in memory until the snapshot completes, up
+  to `MaxSnapshotDeferredRows` (default 100,000); past that the oldest are dead-lettered rather
+  than buffered without bound — an older, name-ordered sender whose child table sorts first can
+  otherwise defer a whole table.
 - **Secondary `UNIQUE` keys must be unique across both nodes.** Remote rows are applied with
   `INSERT … ON CONFLICT(<pk>) DO UPDATE` (`0.3.6`), so only the row with the same primary key
   is touched. If the same natural key is created under different primary keys on the two nodes
@@ -370,7 +374,7 @@ null-guarded — replication runs identically whether or not metrics are collect
 | `localdb.sync.lag.seconds` | ObservableGauge | Seconds since the last successful sync (no measurement when never synced). |
 | `localdb.sync.applied` | Counter | Rows applied from inbound **delta** batches. |
 | `localdb.sync.conflicts_discarded` | Counter | Inbound rows that lost LWW. |
-| `localdb.sync.dead_lettered` | Counter | Poison / over-drift rows routed to the dead-letter table. |
+| `localdb.sync.dead_lettered` | Counter | Rows routed to the dead-letter table — poison / over-drift rows from deltas **and** (since `0.3.6`) snapshots, including FK orphans and rows spilled past `MaxSnapshotDeferredRows`. |
 | `localdb.sync.reconnects` | Counter | Reconnect attempts (initiator). |
 | `localdb.sync.snapshots` | Counter | Snapshots, tagged `direction=sent`/`received` (one increment per snapshot). |
 
@@ -409,8 +413,10 @@ provider is wired or the poll threw, so a DB failure never reads as "0 backlog")
   local-file state only — it is NOT the wire `lib_schema_version`, so a pair mid-upgrade
   keeps syncing.
 - The **`__localdb_dead_letter`** table holds poison rows (bad JSON, null-row non-tombstone
-  entries) and over-drift rejects (when `FailClosedOnDrift`). Inspect it to diagnose
+  entries), over-drift rejects (when `FailClosedOnDrift`), snapshot rows whose FK parent never
-  rejects.
+  arrived, and secondary-`UNIQUE` clashes. Every row written there, on the delta and the
+  snapshot path, is counted in `localdb.sync.dead_lettered` (snapshot-path rows were not
+  counted before `0.3.6`). Inspect it to diagnose rejects.
 - **Peer auth** (0.3.6+): with `ApiKey` set, the initiator sends `Authorization: Bearer <key>`
   and the passive `LocalDbSyncService` verifies it (constant-time) **before** the handshake and
   before it displaces the current stream — a caller without the key gets `Unauthenticated`, sees

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

@@ -251,6 +251,7 @@ internal sealed class SnapshotApplier : ISnapshotApplier
     private readonly ILogger _logger;
     private readonly ReplicationOptions _options;
     private readonly Func<DateTimeOffset> _utcNow;
+    private readonly LocalDbMetrics? _metrics;
     private bool _driftWarned;
     private long _asOfSeq;
     private long _rows;
@@ -263,6 +264,9 @@ internal sealed class SnapshotApplier : ISnapshotApplier
     // sender's debt clears only on the SnapshotAck, so the whole snapshot is re-streamed.
     private List<OplogEntryRecord> _pending = [];
 
+    /// <summary>Rows currently held back for a missing FK parent. Test observation.</summary>
+    internal int DeferredCount => _pending.Count;
+
     /// <param name="db">The receiving database (dead-letter target for over-drift rows).</param>
     /// <param name="applier">The shared LWW applier.</param>
     /// <param name="store">The oplog/peer-state store.</param>
@@ -274,10 +278,15 @@ internal sealed class SnapshotApplier : ISnapshotApplier
     /// warn, never dead-letter.
     /// </param>
     /// <param name="utcNow">Test seam for the drift clock.</param>
+    /// <param name="metrics">
+    /// Optional; when set, every row this snapshot dead-letters (poison, drift reject, FK orphan,
+    /// cap spill) is added to <c>localdb.sync.dead_lettered</c>, as the delta path already does.
+    /// </param>
     public SnapshotApplier(
         SqliteLocalDb db, LwwApplier applier, OplogStore store, ILogger logger,
-        ReplicationOptions? options = null, Func<DateTimeOffset>? utcNow = null)
+        ReplicationOptions? options = null, Func<DateTimeOffset>? utcNow = null, LocalDbMetrics? metrics = null)
     {
+        _metrics = metrics;
         _db = db;
         _applier = applier;
         _store = store;
@@ -335,6 +344,7 @@ internal sealed class SnapshotApplier : ISnapshotApplier
                 {
                     // Skipped rows contribute nothing to _maxHlc, so last_seen_hlc stays honest.
                     await HlcDriftGuard.DeadLetterAsync(_db, entry, physicalMs, nowMs, now, _options.MaxHlcDriftAhead, ct);
+                    _metrics?.RecordDeadLettered(1);
                     continue;
                 }
             }
@@ -348,11 +358,39 @@ internal sealed class SnapshotApplier : ISnapshotApplier
         }
         var result = await _applier.ApplyBatchAsync(entries, ct, deferForeignKeyViolations: true);
         _rows += batch.Rows.Count;
+        _metrics?.RecordDeadLettered(result.DeadLettered);
         _pending.AddRange(result.Deferred);
 
         // Something landed, so a deferred child may now have its parent.
         if (result.Applied > 0 && _pending.Count > 0)
             await RetryPendingAsync(ct);
+
+        if (_pending.Count > _options.MaxSnapshotDeferredRows)
+            await SpillPendingAsync(ct);
+    }
+
+    /// <summary>
+    /// Bounds the held-back rows at <see cref="ReplicationOptions.MaxSnapshotDeferredRows"/>
+    /// (scadaproj#50). The oldest rows past the cap get one more apply with deferral off — so any whose
+    /// parent has landed by now (a local write, say) still apply — and the rest are dead-lettered,
+    /// exactly as an orphan is at <c>SnapshotComplete</c>. Without the cap a child table that an older,
+    /// name-ordered sender streams ahead of its parent would sit in memory whole.
+    /// </summary>
+    private async Task SpillPendingAsync(CancellationToken ct)
+    {
+        var excess = _pending.Count - _options.MaxSnapshotDeferredRows;
+        var spill = _pending.GetRange(0, excess);
+        _pending.RemoveRange(0, excess);
+
+        var result = await _applier.ApplyBatchAsync(spill, ct);
+        _metrics?.RecordDeadLettered(result.DeadLettered);
+        if (result.DeadLettered > 0)
+        {
+            _logger.LogWarning(
+                "Snapshot receive is holding more than MaxSnapshotDeferredRows ({Cap}) rows whose foreign-key parent has not arrived; " +
+                "dead-lettered the {Count} oldest. Upgrade the sending node (0.3.6+ sends parent tables first) or raise the cap.",
+                _options.MaxSnapshotDeferredRows, result.DeadLettered);
+        }
     }
 
     /// <summary>
@@ -366,6 +404,9 @@ internal sealed class SnapshotApplier : ISnapshotApplier
         while (_pending.Count > 0)
         {
             var result = await _applier.ApplyBatchAsync(_pending, ct, deferForeignKeyViolations: true);
+            // A retried row can now fail on something other than its FK (a secondary UNIQUE clash with
+            // a row that landed meanwhile) and be dead-lettered here.
+            _metrics?.RecordDeadLettered(result.DeadLettered);
             var progressed = result.Deferred.Count < _pending.Count;
             _pending = result.Deferred.ToList();
             if (!progressed)
@@ -392,6 +433,7 @@ internal sealed class SnapshotApplier : ISnapshotApplier
             if (_pending.Count > 0)
             {
                 var final = await _applier.ApplyBatchAsync(_pending, ct);
+                _metrics?.RecordDeadLettered(final.DeadLettered);
                 if (final.DeadLettered > 0)
                 {
                     _logger.LogWarning(

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

@@ -31,7 +31,7 @@ internal static class SyncSessionFactory
 
         var streamer = new SnapshotStreamer(db, store, options, loggerFactory.CreateLogger<SnapshotStreamer>());
         session.SnapshotSender = streamer.SendAsync;
-        session.SnapshotApplier = new SnapshotApplier(db, applier, store, loggerFactory.CreateLogger<SnapshotApplier>(), options);
+        session.SnapshotApplier = new SnapshotApplier(db, applier, store, loggerFactory.CreateLogger<SnapshotApplier>(), options, utcNow: null, metrics);
         return session;
     }
 }

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

@@ -62,7 +62,12 @@ public sealed class LocalDbMetrics : IDisposable
         if (count > 0) _conflictsDiscarded.Add(count);
     }
 
-    /// <summary>Adds <paramref name="count"/> poison / over-drift rows routed to the dead-letter table to <c>localdb.sync.dead_lettered</c> (no-op when not positive).</summary>
+    /// <summary>
+    /// Adds <paramref name="count"/> rows routed to the dead-letter table to <c>localdb.sync.dead_lettered</c>
+    /// (no-op when not positive): poison and over-drift rows from delta batches AND, since 0.3.6, from
+    /// snapshots — including snapshot rows whose foreign-key parent never arrived and rows spilled past
+    /// <see cref="ReplicationOptions.MaxSnapshotDeferredRows"/>.
+    /// </summary>
     public void RecordDeadLettered(long count)
     {
         if (count > 0) _deadLettered.Add(count);

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

@@ -103,6 +103,28 @@ public sealed class ReplicationOptions
     /// <summary>How often the maintenance service enforces the oplog caps and prunes expired tombstones.</summary>
     public TimeSpan MaintenanceInterval { get; set; } = TimeSpan.FromSeconds(60);
 
+    /// <summary>
+    /// Ceiling on the snapshot rows a receiver holds in memory while their foreign-key parent has not
+    /// arrived yet. Default 100,000. Past it the OLDEST held rows are applied once more and, if their
+    /// parent is still missing, dead-lettered.
+    /// </summary>
+    /// <remarks>
+    /// <para>
+    /// Since 0.3.6 a snapshot row that fails only its FK check is deferred and retried as the rest of
+    /// the snapshot lands, instead of being dead-lettered. A 0.3.6 sender streams parent tables first,
+    /// so in practice only self-referencing rows whose parent sorts later wait; but an OLDER sender
+    /// streams tables by name, and a child table that sorts before its parent then defers every one of
+    /// its rows. This bounds that memory.
+    /// </para>
+    /// <para>
+    /// <b>A spilled row is lost to this snapshot</b>, exactly like an orphan dead-lettered at
+    /// <c>SnapshotComplete</c>: the sender's debt clears on the ack, so it is not re-sent. It is
+    /// visible in <c>__localdb_dead_letter</c> and in <c>localdb.sync.dead_lettered</c>. Upgrade the
+    /// sender (parents first) or raise this cap if spills appear.
+    /// </para>
+    /// </remarks>
+    public int MaxSnapshotDeferredRows { get; set; } = 100_000;
+
     /// <summary>Maximum tolerated HLC drift where a peer's clock runs ahead of ours.</summary>
     public TimeSpan MaxHlcDriftAhead { get; set; } = TimeSpan.FromMinutes(5);
 

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

@@ -46,6 +46,8 @@ public sealed class ReplicationOptionsValidator : IValidateOptions<ReplicationOp
             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.MaxSnapshotDeferredRows <= 0)
+            failures.Add($"LocalDb:Replication:MaxSnapshotDeferredRows must be greater than 0; got {options.MaxSnapshotDeferredRows}.");
         if (options.MaxHlcDriftAhead <= TimeSpan.Zero)
             failures.Add($"LocalDb:Replication:MaxHlcDriftAhead must be greater than zero; got {options.MaxHlcDriftAhead}.");
 

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

@@ -0,0 +1,292 @@
+using System.Diagnostics.Metrics;
+using System.Threading.Channels;
+using Microsoft.Extensions.Logging.Abstractions;
+using ZB.MOM.WW.LocalDb.Contracts;
+using ZB.MOM.WW.LocalDb.Internal;
+using ZB.MOM.WW.LocalDb.Replication;
+using ZB.MOM.WW.LocalDb.Replication.Internal;
+
+namespace ZB.MOM.WW.LocalDb.Tests;
+
+/// <summary>
+/// scadaproj#50, two follow-ups to the 0.3.6 snapshot fixes:
+/// <list type="bullet">
+///   <item>rows the SNAPSHOT path dead-letters (FK orphans from #37, drift rejects from #39, poison)
+///   were never counted in <c>localdb.sync.dead_lettered</c> — only the delta path recorded it;</item>
+///   <item>snapshot rows held back for a missing FK parent were kept in memory without bound, which
+///   against an older child-first sender could be a whole table. They are now capped by
+///   <see cref="ReplicationOptions.MaxSnapshotDeferredRows"/> and spilled to the dead-letter table
+///   past it.</item>
+/// </list>
+/// </summary>
+public sealed class SnapshotDeadLetterTests : IDisposable
+{
+    private const string DeadLettered = "localdb.sync.dead_lettered";
+    private static readonly TimeSpan RunTimeout = TimeSpan.FromSeconds(15);
+
+    private readonly List<string> _paths = [];
+    private readonly List<IDisposable> _disposables = [];
+
+    public void Dispose()
+    {
+        foreach (var d in _disposables)
+            d.Dispose();
+        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 async Task<SqliteLocalDb> NewDbAsync()
+    {
+        var path = Path.Combine(Path.GetTempPath(), Guid.NewGuid() + ".db");
+        _paths.Add(path);
+        var db = new SqliteLocalDb(new LocalDbOptions { Path = path });
+        _disposables.Add(db);
+        await db.ExecuteAsync("CREATE TABLE z_parent (id INTEGER PRIMARY KEY, name TEXT)");
+        await db.ExecuteAsync("CREATE TABLE a_child (id INTEGER PRIMARY KEY, parent_id INTEGER REFERENCES z_parent(id))");
+        db.RegisterReplicated("z_parent");
+        db.RegisterReplicated("a_child");
+        return db;
+    }
+
+    private static SnapshotApplier ApplierFor(SqliteLocalDb db, ReplicationOptions options, LocalDbMetrics metrics) =>
+        new(db, new LwwApplier(db), new OplogStore(db, options), NullLogger.Instance, options, utcNow: null, metrics);
+
+    private static SnapshotBatch Children(params (long Id, long ParentId)[] rows)
+    {
+        var batch = new SnapshotBatch { TableName = "a_child" };
+        foreach (var (id, parent) in rows)
+            batch.Rows.Add(new SnapshotRow
+            {
+                PkJson = $"{{\"id\":{id}}}", RowJson = $"{{\"id\":{id},\"parent_id\":{parent}}}", Hlc = 1000 + id, NodeId = "peer",
+            });
+        return batch;
+    }
+
+    private static SnapshotBatch Parent(long id)
+    {
+        var batch = new SnapshotBatch { TableName = "z_parent" };
+        batch.Rows.Add(new SnapshotRow { PkJson = $"{{\"id\":{id}}}", RowJson = $"{{\"id\":{id},\"name\":\"p\"}}", Hlc = 1000 + id, NodeId = "peer" });
+        return batch;
+    }
+
+    private static async Task<long> Scalar(SqliteLocalDb db, string sql) =>
+        (await db.QueryAsync(sql, r => r.GetInt64(0)))[0];
+
+    private static Task<long> DeadLetterRows(SqliteLocalDb db) => Scalar(db, "SELECT COUNT(*) FROM __localdb_dead_letter");
+
+    private static async Task WaitForAsync(Func<Task<bool>> predicate, TimeSpan timeout)
+    {
+        var deadline = DateTime.UtcNow + timeout;
+        while (DateTime.UtcNow < deadline)
+        {
+            if (await predicate()) return;
+            await Task.Delay(20);
+        }
+        throw new TimeoutException("Condition not reached within " + timeout);
+    }
+
+    private static async Task SwallowAsync(Task task)
+    {
+        try { await task; }
+        catch (OperationCanceledException) { }
+    }
+
+    // ---- metric ---------------------------------------------------------------------------------
+
+    [Fact]
+    public async Task SnapshotOrphan_IsCountedInTheDeadLetterMetric()
+    {
+        using var metrics = new LocalDbMetrics();
+        using var collector = new Collector(metrics.Meter);
+        var db = await NewDbAsync();
+        var applier = ApplierFor(db, new ReplicationOptions(), metrics);
+
+        await applier.OnBeginAsync(new SnapshotBegin { AsOfSeq = 5 }, default);
+        await applier.OnBatchAsync(Children((10, 99)), default);
+        await applier.OnCompleteAsync(new SnapshotComplete { AsOfSeq = 5 }, default);
+
+        Assert.Equal(1, await DeadLetterRows(db));
+        Assert.Equal(1, collector.Sum(DeadLettered));
+    }
+
+    [Fact]
+    public async Task SnapshotDriftReject_IsCountedInTheDeadLetterMetric()
+    {
+        using var metrics = new LocalDbMetrics();
+        using var collector = new Collector(metrics.Meter);
+        var db = await NewDbAsync();
+        var applier = ApplierFor(db, new ReplicationOptions { FailClosedOnDrift = true }, metrics);
+
+        var batch = Parent(1);
+        batch.Rows[0].Hlc = DateTimeOffset.UtcNow.AddDays(365).ToUnixTimeMilliseconds() << 16;
+        await applier.OnBeginAsync(new SnapshotBegin { AsOfSeq = 5 }, default);
+        await applier.OnBatchAsync(batch, default);
+        await applier.OnCompleteAsync(new SnapshotComplete { AsOfSeq = 5 }, default);
+
+        Assert.Equal(1, await DeadLetterRows(db));
+        Assert.Equal(1, collector.Sum(DeadLettered));
+    }
+
+    [Fact]
+    public async Task SnapshotPoisonRow_IsCountedInTheDeadLetterMetric()
+    {
+        using var metrics = new LocalDbMetrics();
+        using var collector = new Collector(metrics.Meter);
+        var db = await NewDbAsync();
+        var applier = ApplierFor(db, new ReplicationOptions(), metrics);
+
+        // A live row with no row data is poison on any path.
+        var batch = new SnapshotBatch { TableName = "z_parent" };
+        batch.Rows.Add(new SnapshotRow { PkJson = "{\"id\":1}", Hlc = 1001, NodeId = "peer" });
+        await applier.OnBeginAsync(new SnapshotBegin { AsOfSeq = 5 }, default);
+        await applier.OnBatchAsync(batch, default);
+        await applier.OnCompleteAsync(new SnapshotComplete { AsOfSeq = 5 }, default);
+
+        Assert.Equal(1, await DeadLetterRows(db));
+        Assert.Equal(1, collector.Sum(DeadLettered));
+    }
+
+    [Fact]
+    public async Task TwoNodes_SnapshotOrphan_IsCountedByTheProductionWiring()
+    {
+        // The sender holds an orphan child (written with FK enforcement off on that connection), so a
+        // wiped peer receives a row whose parent never arrives. Both sessions are built by
+        // SyncSessionFactory, exactly as the hosts build them, so this also proves the factory hands
+        // the metrics to the snapshot applier.
+        var options = new ReplicationOptions { FlushInterval = TimeSpan.FromMilliseconds(20) };
+        var dbA = await NewDbAsync();
+        await dbA.ExecuteAsync("INSERT INTO z_parent (id, name) VALUES (1, 'p')");
+        await dbA.ExecuteAsync("INSERT INTO a_child (id, parent_id) VALUES (10, 1)");
+        using (var conn = dbA.CreateConnection())
+        {
+            using var cmd = conn.CreateCommand();
+            cmd.CommandText = "PRAGMA foreign_keys=OFF; INSERT INTO a_child (id, parent_id) VALUES (11, 99);";
+            cmd.ExecuteNonQuery();
+        }
+        var storeA = new OplogStore(dbA, options);
+        await storeA.RecordPeerAckAsync(await storeA.GetMaxSeqAsync());
+
+        var dbB = await NewDbAsync();
+        using var metricsB = new LocalDbMetrics();
+        using var collector = new Collector(metricsB.Meter);
+
+        var a = SyncSessionFactory.Create(dbA, options, NullLoggerFactory.Instance);
+        var b = SyncSessionFactory.Create(dbB, options, NullLoggerFactory.Instance, metricsB);
+        var aToB = Channel.CreateUnbounded<SyncMessage>();
+        var bToA = Channel.CreateUnbounded<SyncMessage>();
+        var da = new SyncDuplex { Send = (m, ct) => aToB.Writer.WriteAsync(m, ct).AsTask(), Inbox = bToA.Reader };
+        var dbx = new SyncDuplex { Send = (m, ct) => bToA.Writer.WriteAsync(m, ct).AsTask(), Inbox = aToB.Reader };
+        using var cts = new CancellationTokenSource(RunTimeout);
+        var runA = a.RunAsync(da, cts.Token);
+        var runB = b.RunAsync(dbx, cts.Token);
+
+        await WaitForAsync(async () =>
+            await Scalar(dbB, "SELECT last_applied_remote_seq FROM __localdb_peer_state") > 0, RunTimeout);
+
+        Assert.Equal(1, await Scalar(dbB, "SELECT COUNT(*) FROM a_child"));
+        Assert.Equal(1, await DeadLetterRows(dbB));
+        Assert.Equal(1, collector.Sum(DeadLettered));
+
+        cts.Cancel();
+        await SwallowAsync(runA);
+        await SwallowAsync(runB);
+    }
+
+    // ---- cap on held-back rows -------------------------------------------------------------------
+
+    [Fact]
+    public async Task DeferredRows_PastTheCap_AreSpilledToDeadLetter_BeforeComplete()
+    {
+        using var metrics = new LocalDbMetrics();
+        using var collector = new Collector(metrics.Meter);
+        var db = await NewDbAsync();
+        var applier = ApplierFor(db, new ReplicationOptions { MaxSnapshotDeferredRows = 2 }, metrics);
+
+        await applier.OnBeginAsync(new SnapshotBegin { AsOfSeq = 5 }, default);
+        await applier.OnBatchAsync(Children((10, 1), (11, 1)), default);
+        // At the cap, nothing is spilled yet.
+        Assert.Equal(0, await DeadLetterRows(db));
+        Assert.Equal(2, applier.DeferredCount);
+
+        // One more held-back row: the OLDEST (10) is spilled, so memory stays at the cap.
+        await applier.OnBatchAsync(Children((12, 1)), default);
+        Assert.Equal(2, applier.DeferredCount);
+        Assert.Equal(1, await DeadLetterRows(db));
+        Assert.Equal(1, await Scalar(db, "SELECT COUNT(*) FROM __localdb_dead_letter WHERE pk_json = '{\"id\":10}'"));
+        Assert.Equal(1, collector.Sum(DeadLettered));
+
+        // The rows still held land when their parent arrives; the spilled one stays dead-lettered.
+        await applier.OnBatchAsync(Parent(1), default);
+        await applier.OnCompleteAsync(new SnapshotComplete { AsOfSeq = 5 }, default);
+
+        Assert.Equal(0, applier.DeferredCount);
+        Assert.Equal(new long[] { 11, 12 },
+            (await db.QueryAsync("SELECT id FROM a_child ORDER BY id", r => r.GetInt64(0))).ToArray());
+        Assert.Equal(1, await DeadLetterRows(db));
+        Assert.Equal(1, collector.Sum(DeadLettered));
+    }
+
+    [Fact]
+    public async Task DeferredRows_PastTheCap_ThatHaveBecomeValid_AreAppliedNotDeadLettered()
+    {
+        // The spill re-applies with deferral off first, so a row whose parent HAS landed by then is
+        // applied rather than dead-lettered.
+        var db = await NewDbAsync();
+        using var metrics = new LocalDbMetrics();
+        var applier = ApplierFor(db, new ReplicationOptions { MaxSnapshotDeferredRows = 1 }, metrics);
+
+        await applier.OnBeginAsync(new SnapshotBegin { AsOfSeq = 5 }, default);
+        await applier.OnBatchAsync(Children((10, 1)), default);
+        // Insert the parent locally outside the snapshot, so nothing in the applier retries row 10.
+        await db.ExecuteAsync("INSERT INTO z_parent (id, name) VALUES (1, 'local')");
+        await applier.OnBatchAsync(Children((11, 2)), default);
+
+        Assert.Equal(0, await DeadLetterRows(db));
+        Assert.Equal(1, await Scalar(db, "SELECT COUNT(*) FROM a_child WHERE id = 10"));
+        Assert.Equal(1, applier.DeferredCount);
+    }
+
+    [Fact]
+    public void Validator_RejectsANonPositiveCap()
+    {
+        var result = new ReplicationOptionsValidator().Validate(null, new ReplicationOptions { MaxSnapshotDeferredRows = 0 });
+        Assert.True(result.Failed);
+        Assert.Contains(result.Failures!, f => f.Contains("MaxSnapshotDeferredRows", StringComparison.Ordinal));
+    }
+
+    private sealed class Collector : IDisposable
+    {
+        private readonly MeterListener _listener;
+        private readonly Dictionary<string, long> _sums = new(StringComparer.Ordinal);
+        private readonly Lock _lock = new();
+
+        public Collector(Meter meter)
+        {
+            _listener = new MeterListener
+            {
+                InstrumentPublished = (inst, l) =>
+                {
+                    if (ReferenceEquals(inst.Meter, meter))
+                        l.EnableMeasurementEvents(inst);
+                },
+            };
+            _listener.SetMeasurementEventCallback<long>((inst, val, _, _) =>
+            {
+                lock (_lock) _sums[inst.Name] = _sums.GetValueOrDefault(inst.Name) + val;
+            });
+            _listener.Start();
+        }
+
+        public long Sum(string name)
+        {
+            lock (_lock) return _sums.GetValueOrDefault(name);
+        }
+
+        public void Dispose() => _listener.Dispose();
+    }
+}