From b2adddbc8ec9c80f0e88d1e07f153b224d51323b Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Mon, 9 Nov 2026 15:12:41 -0500 Subject: [PATCH] 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 | 14 +- .../Internal/SnapshotStreamer.cs | 44 ++- .../Internal/SyncSessionFactory.cs | 2 +- .../LocalDbMetrics.cs | 7 +- .../ReplicationOptions.cs | 22 ++ .../ReplicationOptionsValidator.cs | 2 + .../SnapshotDeadLetterTests.cs | 292 ++++++++++++++++++ 7 files changed, 376 insertions(+), 7 deletions(-) create mode 100644 ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/SnapshotDeadLetterTests.cs diff --git a/ZB.MOM.WW.LocalDb/README.md b/ZB.MOM.WW.LocalDb/README.md index 5b853a0..d62a293 100644 --- a/ZB.MOM.WW.LocalDb/README.md +++ b/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() 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 - rejects. + entries), over-drift rejects (when `FailClosedOnDrift`), snapshot rows whose FK parent never + 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 ` 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 diff --git a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/SnapshotStreamer.cs b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/SnapshotStreamer.cs index e0aaa9c..b3d88f2 100644 --- a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/SnapshotStreamer.cs +++ b/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 _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 _pending = []; + /// Rows currently held back for a missing FK parent. Test observation. + internal int DeferredCount => _pending.Count; + /// The receiving database (dead-letter target for over-drift rows). /// The shared LWW applier. /// The oplog/peer-state store. @@ -274,10 +278,15 @@ internal sealed class SnapshotApplier : ISnapshotApplier /// warn, never dead-letter. /// /// Test seam for the drift clock. + /// + /// Optional; when set, every row this snapshot dead-letters (poison, drift reject, FK orphan, + /// cap spill) is added to localdb.sync.dead_lettered, as the delta path already does. + /// public SnapshotApplier( SqliteLocalDb db, LwwApplier applier, OplogStore store, ILogger logger, - ReplicationOptions? options = null, Func? utcNow = null) + ReplicationOptions? options = null, Func? 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); + } + + /// + /// Bounds the held-back rows at + /// (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 SnapshotComplete. Without the cap a child table that an older, + /// name-ordered sender streams ahead of its parent would sit in memory whole. + /// + 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); + } } /// @@ -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( diff --git a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/SyncSessionFactory.cs b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/SyncSessionFactory.cs index 4c45972..7894a93 100644 --- a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/SyncSessionFactory.cs +++ b/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()); session.SnapshotSender = streamer.SendAsync; - session.SnapshotApplier = new SnapshotApplier(db, applier, store, loggerFactory.CreateLogger(), options); + session.SnapshotApplier = new SnapshotApplier(db, applier, store, loggerFactory.CreateLogger(), options, utcNow: null, metrics); return session; } } diff --git a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/LocalDbMetrics.cs b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/LocalDbMetrics.cs index 143dbde..1ef1704 100644 --- a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/LocalDbMetrics.cs +++ b/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); } - /// Adds poison / over-drift rows routed to the dead-letter table to localdb.sync.dead_lettered (no-op when not positive). + /// + /// Adds rows routed to the dead-letter table to localdb.sync.dead_lettered + /// (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 + /// . + /// public void RecordDeadLettered(long count) { if (count > 0) _deadLettered.Add(count); diff --git a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/ReplicationOptions.cs b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/ReplicationOptions.cs index 666bd59..fcbd9e2 100644 --- a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/ReplicationOptions.cs +++ b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/ReplicationOptions.cs @@ -103,6 +103,28 @@ public sealed class ReplicationOptions /// How often the maintenance service enforces the oplog caps and prunes expired tombstones. public TimeSpan MaintenanceInterval { get; set; } = TimeSpan.FromSeconds(60); + /// + /// 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. + /// + /// + /// + /// 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. + /// + /// + /// A spilled row is lost to this snapshot, exactly like an orphan dead-lettered at + /// SnapshotComplete: the sender's debt clears on the ack, so it is not re-sent. It is + /// visible in __localdb_dead_letter and in localdb.sync.dead_lettered. Upgrade the + /// sender (parents first) or raise this cap if spills appear. + /// + /// + public int MaxSnapshotDeferredRows { get; set; } = 100_000; + /// Maximum tolerated HLC drift where a peer's clock runs ahead of ours. public TimeSpan MaxHlcDriftAhead { get; set; } = TimeSpan.FromMinutes(5); diff --git a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/ReplicationOptionsValidator.cs b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/ReplicationOptionsValidator.cs index 625c745..73e2be0 100644 --- a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/ReplicationOptionsValidator.cs +++ b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/ReplicationOptionsValidator.cs @@ -46,6 +46,8 @@ public sealed class ReplicationOptionsValidator : IValidateOptions +/// scadaproj#50, two follow-ups to the 0.3.6 snapshot fixes: +/// +/// rows the SNAPSHOT path dead-letters (FK orphans from #37, drift rejects from #39, poison) +/// were never counted in localdb.sync.dead_lettered — only the delta path recorded it; +/// 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 +/// and spilled to the dead-letter table +/// past it. +/// +/// +public sealed class SnapshotDeadLetterTests : IDisposable +{ + private const string DeadLettered = "localdb.sync.dead_lettered"; + private static readonly TimeSpan RunTimeout = TimeSpan.FromSeconds(15); + + private readonly List _paths = []; + private readonly List _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 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 Scalar(SqliteLocalDb db, string sql) => + (await db.QueryAsync(sql, r => r.GetInt64(0)))[0]; + + private static Task DeadLetterRows(SqliteLocalDb db) => Scalar(db, "SELECT COUNT(*) FROM __localdb_dead_letter"); + + private static async Task WaitForAsync(Func> 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(); + var bToA = Channel.CreateUnbounded(); + 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 _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((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(); + } +}