From 4af0ea8152dd77b8eec2cbd1f83d88398d7addb5 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Mon, 9 Nov 2026 14:52:35 -0500 Subject: [PATCH] fix(localdb): snapshot delivers FK-linked rows in any order (scadaproj#37) The snapshot streamed tables by name and rows by pk_json text, so a child (or a self-reference whose parent's key sorts later) reached the receiver first, failed its immediate FK check and was dead-lettered - permanently, since the snapshot's watermark and ack covered it (LocalDb-002). Sender: stream referenced tables first (Kahn order, name tie-break). Receiver: SnapshotApplier defers rows that fail only SQLITE_CONSTRAINT_FOREIGNKEY, retries them as rows land, and dead-letters true orphans at SnapshotComplete. Delta path unchanged; wire unchanged. Claude-Session: https://claude.ai/code/session_01GT67AEHd2Acv1YPwGfXEoV --- ZB.MOM.WW.LocalDb/CLAUDE.md | 1 + ZB.MOM.WW.LocalDb/README.md | 9 + .../Internal/LwwApplier.cs | 39 +++- .../Internal/SnapshotStreamer.cs | 90 +++++++- .../SnapshotForeignKeyTests.cs | 194 ++++++++++++++++++ code-reviews/LocalDb/findings.md | 25 ++- code-reviews/README.md | 11 +- 7 files changed, 351 insertions(+), 18 deletions(-) create mode 100644 ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/SnapshotForeignKeyTests.cs diff --git a/ZB.MOM.WW.LocalDb/CLAUDE.md b/ZB.MOM.WW.LocalDb/CLAUDE.md index d1e1903..f15750a 100644 --- a/ZB.MOM.WW.LocalDb/CLAUDE.md +++ b/ZB.MOM.WW.LocalDb/CLAUDE.md @@ -8,6 +8,7 @@ Shared embedded SQLite cache + 2-node LWW sync library for the **ZB.MOM.WW SCADA **`0.3.6` (unpublished, scadaproj#36–#39):** code-review fixes, all wire-compatible (no proto, `lib_schema_version` or on-disk schema change). - **#36 LocalDb-001:** the passive `LocalDbSyncService` now enforces `LocalDb:Replication:ApiKey` itself (Bearer, constant-time, checked before the handshake and before newest-wins takeover; `Unauthenticated` otherwise). Blank key = unchanged behaviour plus a startup warning. ScadaBridge's and OtOpcUa's own `LocalDbSyncAuthInterceptor`s use the same key and header, so they keep working; they are now redundant except for their fail-closed-when-blank rule. +- **#37 LocalDb-002:** snapshots deliver FK-linked rows in any order. `SnapshotStreamer` sends referenced tables first (Kahn order, name tie-break); `SnapshotApplier` defers rows that fail only SQLITE_CONSTRAINT_FOREIGNKEY (`LwwApplier.ApplyBatchAsync(..., deferForeignKeyViolations: true)`), retries them as rows land, and dead-letters true orphans at `SnapshotComplete`. Before, a child that arrived first was dead-lettered and acked over permanently. ## Index history (migrated from the scadaproj root CLAUDE.md) diff --git a/ZB.MOM.WW.LocalDb/README.md b/ZB.MOM.WW.LocalDb/README.md index 14d484c..f3ae9a0 100644 --- a/ZB.MOM.WW.LocalDb/README.md +++ b/ZB.MOM.WW.LocalDb/README.md @@ -304,6 +304,15 @@ All options are validated at startup (`ZB.MOM.WW.Configuration` validators). the parent unreplicated. The check runs at registration and reads the whole database's foreign keys, so a child table created *after* registration is caught at the next startup, not at the moment of the `CREATE TABLE`. +- **Foreign keys between replicated tables (`NO ACTION`/`RESTRICT`) survive a snapshot as of + `0.3.6`.** The sender streams referenced tables before the tables that reference them, and + the receiver defers a snapshot row that fails only its FK check, retrying it as more rows + land; one whose parent never arrives is dead-lettered at `SnapshotComplete`. Before `0.3.6` + a child that arrived before its parent (name order, or a self-reference whose parent's key + 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. - **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 diff --git a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/LwwApplier.cs b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/LwwApplier.cs index a384b77..0894b96 100644 --- a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/LwwApplier.cs +++ b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/LwwApplier.cs @@ -6,7 +6,15 @@ using ZB.MOM.WW.LocalDb.Registration; namespace ZB.MOM.WW.LocalDb.Replication.Internal; /// Outcome of applying one inbound batch. is the batch's highest seq regardless of per-entry disposition, so the watermark advances even over discarded/dead-lettered entries. -internal sealed record ApplyResult(long AppliedThruSeq, int Applied, int Discarded, int DeadLettered); +/// +/// is only ever non-empty when the caller asked for foreign-key violations to +/// be deferred (the snapshot path): those entries were rolled back, NOT dead-lettered, and the caller +/// owns retrying them. +/// +internal sealed record ApplyResult(long AppliedThruSeq, int Applied, int Discarded, int DeadLettered) +{ + public IReadOnlyList Deferred { get; init; } = []; +} /// /// Applies a peer's oplog batch under last-writer-wins in one transaction: the applying guard is @@ -35,7 +43,19 @@ internal sealed class LwwApplier(SqliteLocalDb db, Func? utcNow /// internal int LastBatchStatementCount { get; private set; } - public async Task ApplyBatchAsync(IReadOnlyList entries, CancellationToken ct) + /// The batch, in apply order. + /// Cancellation. + /// + /// Snapshot path only (LocalDb-002). A snapshot can deliver a child row before its parent — a + /// table that sorts earlier, a self-reference whose parent's key sorts later, or any order an + /// older sender chose — and the delta path's rule (an SQLite error is per-entry poison) would + /// dead-letter it permanently. With this set, an entry that fails ONLY its foreign-key check is + /// rolled back and returned in for the caller to retry once + /// more of the snapshot has landed. Off for deltas: they arrive in commit order, so a parent + /// always precedes its child there. + /// + public async Task ApplyBatchAsync( + IReadOnlyList entries, CancellationToken ct, bool deferForeignKeyViolations = false) { if (entries.Count == 0) return new ApplyResult(0, 0, 0, 0); @@ -60,6 +80,7 @@ internal sealed class LwwApplier(SqliteLocalDb db, Func? utcNow var applied = 0; var discarded = 0; var deadLettered = 0; + List? deferred = null; await using var tx = await db.BeginPreparedTransactionAsync(ct); using var sql = new PreparedStatements(tx.Connection, tx.Sqlite); @@ -120,10 +141,12 @@ internal sealed class LwwApplier(SqliteLocalDb db, Func? utcNow // statement, not the transaction, so the batch continues after dead-lettering. // Any other exception propagates with the savepoint still open, and disposing the // outer transaction unwinds the whole batch (fail-closed). + var defer = deferForeignKeyViolations && IsForeignKeyViolation(ex); try { await sql.RollbackEntryAsync(ct); - await sql.DeadLetterAsync(entry, ex.Message, utcNow, ct); + if (!defer) + await sql.DeadLetterAsync(entry, ex.Message, utcNow, ct); } catch (Exception recoveryFailure) when (recoveryFailure is not OperationCanceledException) { @@ -141,7 +164,10 @@ internal sealed class LwwApplier(SqliteLocalDb db, Func? utcNow ex, recoveryFailure); } - deadLettered++; + if (defer) + (deferred ??= []).Add(entry); + else + deadLettered++; } } @@ -153,9 +179,12 @@ internal sealed class LwwApplier(SqliteLocalDb db, Func? utcNow await tx.CommitAsync(ct); db.Clock.Observe(maxHlc); - return new ApplyResult(maxSeq, applied, discarded, deadLettered); + return new ApplyResult(maxSeq, applied, discarded, deadLettered) { Deferred = deferred ?? [] }; } + /// SQLITE_CONSTRAINT_FOREIGNKEY (787): the statement failed its immediate FK check. + private static bool IsForeignKeyViolation(SqliteException ex) => ex.SqliteExtendedErrorCode == 787; + /// /// The statement set one apply-batch runs on: each distinct statement is built and prepared on /// first use, then rebound and re-executed per row. Statements whose SQL depends on the table 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 e876401..d43153d 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 @@ -76,9 +76,8 @@ internal sealed class SnapshotStreamer(SqliteLocalDb db, OplogStore store, Repli await send(new SyncMessage { SnapshotBegin = begin }, ct); var totalRows = 0L; - foreach (var table in db.ReplicatedTables.Values - .Where(t => scope.Includes(t.Name)) - .OrderBy(t => t.Name, StringComparer.Ordinal)) + foreach (var table in await ParentsFirstAsync( + db.ReplicatedTables.Values.Where(t => scope.Includes(t.Name)).ToList(), ct)) { var sql = BuildPageSql(table); // Keyset cursor (pk_json > last seen), not LIMIT/OFFSET: concurrent inserts/prunes @@ -162,6 +161,40 @@ internal sealed class SnapshotStreamer(SqliteLocalDb db, OplogStore store, Repli return asOfSeq; } + /// + /// Orders the tables so every table is streamed after the tables it references by foreign key + /// (LocalDb-002): name order put a_child before z_parent, and a receiver with + /// immediate FK checks rejected the child. Kahn's algorithm with name order as the tie-break, so + /// unrelated tables keep their old relative order; self-references are ignored (the receiver + /// defers those rows), and a reference cycle falls back to name order for its members. Purely a + /// sender-side choice — the wire is unchanged and any order is still correct at a 0.3.6+ + /// receiver, which defers FK-failed snapshot rows instead of dead-lettering them. + /// + private async Task> ParentsFirstAsync(List tables, CancellationToken ct) + { + var byName = tables.ToDictionary(t => t.Name, StringComparer.Ordinal); + var parents = new Dictionary>(StringComparer.Ordinal); + foreach (var table in tables) + { + var referenced = await db.QueryAsync( + "SELECT DISTINCT \"table\" FROM pragma_foreign_key_list(@t)", r => r.GetString(0), new { t = table.Name }, ct); + parents[table.Name] = referenced + .Where(p => byName.ContainsKey(p) && !string.Equals(p, table.Name, StringComparison.Ordinal)) + .ToHashSet(StringComparer.Ordinal); + } + + var ordered = new List(tables.Count); + var remaining = new SortedSet(byName.Keys, StringComparer.Ordinal); + while (remaining.Count > 0) + { + var next = remaining.FirstOrDefault(n => !parents[n].Overlaps(remaining)) + ?? remaining.Min!; // a cycle: break it in name order + ordered.Add(byName[next]); + remaining.Remove(next); + } + return ordered; + } + /// /// Serialized bytes one row contributes to its batch message: the row's own encoding plus the /// repeated field's tag and length prefix, as protobuf itself computes them. @@ -221,6 +254,12 @@ internal sealed class SnapshotApplier : ISnapshotApplier private long _maxHlc; private bool _partial; + // Snapshot rows whose foreign-key parent had not arrived yet (LocalDb-002). Held for the life of + // one snapshot receive: retried as more rows land, and dead-lettered at SnapshotComplete only if + // their parent never came. A session that ends mid-snapshot drops them, which is safe — the + // sender's debt clears only on the SnapshotAck, so the whole snapshot is re-streamed. + private List _pending = []; + // db is part of the collaborator's canonical shape (mirrors the streamer) even though the merge // reaches the database through the applier and store; kept for construction symmetry in the factory. public SnapshotApplier(SqliteLocalDb db, LwwApplier applier, OplogStore store, ILogger logger) @@ -236,6 +275,7 @@ internal sealed class SnapshotApplier : ISnapshotApplier _asOfSeq = begin.AsOfSeq; _rows = 0; _maxHlc = 0; + _pending = []; // A non-empty table list scopes the snapshot; empty (which is all a pre-0.3.0 sender can // emit) means every registered table. Read here rather than on complete because // SnapshotComplete does not repeat it — the receive loop guarantees a Begin precedes both. @@ -258,8 +298,31 @@ internal sealed class SnapshotApplier : ISnapshotApplier entries.Add(new OplogEntryRecord( 0, batch.TableName, row.PkJson, row.HasRowJson ? row.RowJson : null, row.Hlc, row.NodeId, row.IsTombstone)); } - await _applier.ApplyBatchAsync(entries, ct); + var result = await _applier.ApplyBatchAsync(entries, ct, deferForeignKeyViolations: true); _rows += batch.Rows.Count; + _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); + } + + /// + /// Re-applies the rows deferred on a foreign-key violation until a pass makes no progress. A pass + /// that applies anything may unblock another deferred row (a chain of self-references, a parent + /// tombstone waiting for its child's), hence the loop; it ends because every pass that continues + /// shrinks the list. + /// + private async Task RetryPendingAsync(CancellationToken ct) + { + while (_pending.Count > 0) + { + var result = await _applier.ApplyBatchAsync(_pending, ct, deferForeignKeyViolations: true); + var progressed = result.Deferred.Count < _pending.Count; + _pending = result.Deferred.ToList(); + if (!progressed) + return; + } } public async Task OnCompleteAsync(SnapshotComplete complete, CancellationToken ct) @@ -272,6 +335,25 @@ internal sealed class SnapshotApplier : ISnapshotApplier // has stamped. The sender makes the mirror-image concession: it does not advance its own // sent-thru watermark past a partial snapshot, so those deltas are still pumped normally // and merely re-deliver rows the snapshot already carried, which LWW absorbs. + // Rows still deferred have no parent anywhere in the snapshot: true orphans. One last pass + // with deferral off applies anything that became valid and dead-letters the rest, so they + // are visible in __localdb_dead_letter instead of silently missing. + if (_pending.Count > 0) + { + await RetryPendingAsync(ct); + if (_pending.Count > 0) + { + var final = await _applier.ApplyBatchAsync(_pending, ct); + if (final.DeadLettered > 0) + { + _logger.LogWarning( + "Snapshot receive dead-lettered {Count} row(s) whose foreign-key parent never arrived.", + final.DeadLettered); + } + _pending = []; + } + } + var seq = _partial ? 0 : complete.AsOfSeq; // SetLastAppliedRemoteSeqAsync is MAX-guarded, so a delta applied mid-snapshot with a seq // above the snapshot's as-of (or an HLC above what the snapshot carried) is never regressed. diff --git a/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/SnapshotForeignKeyTests.cs b/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/SnapshotForeignKeyTests.cs new file mode 100644 index 0000000..d8065f6 --- /dev/null +++ b/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/SnapshotForeignKeyTests.cs @@ -0,0 +1,194 @@ +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; + +/// +/// LocalDb-002 (scadaproj#37): a snapshot must deliver FK-linked rows whatever order they arrive in. +/// The streamer used to send tables by name and rows by pk_json text, so a child (or a +/// self-referencing row whose parent sorts later) reached the receiver first, failed its immediate +/// FK check, and was dead-lettered — permanently, because the snapshot's watermark and ack covered it. +/// +public sealed class SnapshotForeignKeyTests : IDisposable +{ + 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 sealed record Side(SqliteLocalDb Db, OplogStore Store, SyncSession Session); + + // Parent sorts AFTER child by name ("z_" vs "a_"), and the self-referencing table's child row + // (pk {"id":10}) sorts BEFORE its parent ({"id":2}) as pk_json text — both orders the old + // streamer produced. + private async Task NewSideAsync(ReplicationOptions? options = null) + { + var path = Path.Combine(Path.GetTempPath(), Guid.NewGuid() + ".db"); + _paths.Add(path); + var db = new SqliteLocalDb(new LocalDbOptions { Path = path }); + _disposables.Add(db); + await db.ExecuteAsync("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))"); + await db.ExecuteAsync("CREATE TABLE node (id INTEGER PRIMARY KEY, parent_id INTEGER REFERENCES node(id))"); + db.RegisterReplicated("z_parent"); + db.RegisterReplicated("a_child"); + db.RegisterReplicated("node"); + + options ??= new ReplicationOptions { FlushInterval = TimeSpan.FromMilliseconds(20) }; + var store = new OplogStore(db, options); + var applier = new LwwApplier(db); + var session = new SyncSession(db, store, applier, options, NullLogger.Instance); + session.SnapshotSender = new SnapshotStreamer(db, store, options, NullLogger.Instance).SendAsync; + session.SnapshotApplier = new SnapshotApplier(db, applier, store, NullLogger.Instance); + return new Side(db, store, session); + } + + private static (SyncDuplex A, SyncDuplex B) DuplexPair() + { + var aToB = Channel.CreateUnbounded(); + var bToA = Channel.CreateUnbounded(); + return ( + new SyncDuplex { Send = (m, ct) => aToB.Writer.WriteAsync(m, ct).AsTask(), Inbox = bToA.Reader }, + new SyncDuplex { Send = (m, ct) => bToA.Writer.WriteAsync(m, ct).AsTask(), Inbox = aToB.Reader }); + } + + private static async Task Count(SqliteLocalDb db, string table) => + (await db.QueryAsync($"SELECT COUNT(*) FROM {table}", r => r.GetInt64(0)))[0]; + + private static async Task> DeadLetters(SqliteLocalDb db) => + (await db.QueryAsync("SELECT table_name || ' ' || pk_json || ': ' || error FROM __localdb_dead_letter", r => r.GetString(0))).ToList(); + + 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) { } + } + + [Fact] + public async Task FullSnapshot_ToAWipedPeer_DeliversChildrenAndSelfReferences() + { + // MaxBatchSize 1 puts every row in its own batch (own transaction), so nothing can hide an + // ordering problem by landing parent and child in one apply. + var options = new ReplicationOptions { FlushInterval = TimeSpan.FromMilliseconds(20), MaxBatchSize = 1 }; + var a = await NewSideAsync(options); + await a.Db.ExecuteAsync("INSERT INTO z_parent (id, name) VALUES (1, 'p')"); + await a.Db.ExecuteAsync("INSERT INTO a_child (id, parent_id) VALUES (10, 1)"); + await a.Db.ExecuteAsync("INSERT INTO node (id, parent_id) VALUES (2, NULL)"); + await a.Db.ExecuteAsync("INSERT INTO node (id, parent_id) VALUES (10, 2)"); + // Everything acked and pruned: a wiped peer can only be healed by a snapshot. + await a.Store.RecordPeerAckAsync(await a.Store.GetMaxSeqAsync()); + + var b = await NewSideAsync(options); + var (da, dbx) = DuplexPair(); + using var cts = new CancellationTokenSource(RunTimeout); + var runA = a.Session.RunAsync(da, cts.Token); + var runB = b.Session.RunAsync(dbx, cts.Token); + + await WaitForAsync(async () => + (await b.Db.QueryAsync("SELECT last_applied_remote_seq FROM __localdb_peer_state", r => r.GetInt64(0)))[0] > 0, + RunTimeout); + + Assert.Empty(await DeadLetters(b.Db)); + Assert.Equal(1, await Count(b.Db, "z_parent")); + Assert.Equal(1, await Count(b.Db, "a_child")); + Assert.Equal(2, await Count(b.Db, "node")); + + cts.Cancel(); + await SwallowAsync(runA); + await SwallowAsync(runB); + } + + [Fact] + public async Task Streamer_SendsParentTablesBeforeTheirChildren() + { + var a = await NewSideAsync(); + await a.Db.ExecuteAsync("INSERT INTO z_parent (id, name) VALUES (1, 'p')"); + await a.Db.ExecuteAsync("INSERT INTO a_child (id, parent_id) VALUES (10, 1)"); + var streamer = new SnapshotStreamer(a.Db, a.Store, new ReplicationOptions(), NullLogger.Instance); + + var tables = new List(); + await streamer.SendAsync(SnapshotScope.Full, (m, _) => + { + if (m.MsgCase == SyncMessage.MsgOneofCase.SnapshotBatch) + tables.Add(m.SnapshotBatch.TableName); + return Task.CompletedTask; + }, CancellationToken.None); + + Assert.Equal(new[] { "z_parent", "a_child" }, tables); + } + + // ---- receiver side: an older sender still streams children first -------------------------- + + private static SnapshotBatch Batch(string table, params (long Id, string Json)[] rows) + { + var batch = new SnapshotBatch { TableName = table }; + foreach (var (id, json) in rows) + batch.Rows.Add(new SnapshotRow { PkJson = $"{{\"id\":{id}}}", RowJson = json, Hlc = 1000 + id, NodeId = "peer" }); + return batch; + } + + [Fact] + public async Task Receiver_AppliesAChildThatArrivesBeforeItsParent() + { + var b = await NewSideAsync(); + var store = new OplogStore(b.Db, new ReplicationOptions()); + var applier = new SnapshotApplier(b.Db, new LwwApplier(b.Db), store, NullLogger.Instance); + + await applier.OnBeginAsync(new SnapshotBegin { AsOfSeq = 5 }, default); + await applier.OnBatchAsync(Batch("a_child", (10, "{\"id\":10,\"parent_id\":1}")), default); + await applier.OnBatchAsync(Batch("node", (10, "{\"id\":10,\"parent_id\":2}")), default); + await applier.OnBatchAsync(Batch("node", (2, "{\"id\":2,\"parent_id\":null}")), default); + await applier.OnBatchAsync(Batch("z_parent", (1, "{\"id\":1,\"name\":\"p\"}")), default); + await applier.OnCompleteAsync(new SnapshotComplete { AsOfSeq = 5 }, default); + + Assert.Empty(await DeadLetters(b.Db)); + Assert.Equal(1, await Count(b.Db, "a_child")); + Assert.Equal(2, await Count(b.Db, "node")); + } + + [Fact] + public async Task Receiver_DeadLettersATrueOrphan_AtSnapshotComplete() + { + var b = await NewSideAsync(); + var store = new OplogStore(b.Db, new ReplicationOptions()); + var applier = new SnapshotApplier(b.Db, new LwwApplier(b.Db), store, NullLogger.Instance); + + await applier.OnBeginAsync(new SnapshotBegin { AsOfSeq = 5 }, default); + await applier.OnBatchAsync(Batch("a_child", (10, "{\"id\":10,\"parent_id\":99}")), default); + // Still pending, not yet dead-lettered: its parent could be in a later batch. + Assert.Empty(await DeadLetters(b.Db)); + await applier.OnCompleteAsync(new SnapshotComplete { AsOfSeq = 5 }, default); + + var dl = Assert.Single(await DeadLetters(b.Db)); + Assert.Contains("FOREIGN KEY", dl, StringComparison.OrdinalIgnoreCase); + Assert.Equal(0, await Count(b.Db, "a_child")); + } +} diff --git a/code-reviews/LocalDb/findings.md b/code-reviews/LocalDb/findings.md index f841d3f..c625582 100644 --- a/code-reviews/LocalDb/findings.md +++ b/code-reviews/LocalDb/findings.md @@ -10,7 +10,7 @@ | Last reviewed | 2026-11-09 | | Reviewer | Claude | | Commit reviewed | `91fb2b4` | -| Open findings | 6 | +| Open findings | 5 | ## Summary @@ -150,7 +150,7 @@ non-displacement cases fail. | Severity | Medium | | Priority | P2 | | Category | Correctness & logic bugs | -| Status | Open | +| Status | Resolved | | Location | `ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/SnapshotStreamer.cs:79` | **Description** @@ -203,7 +203,26 @@ Also add a snapshot test with a non-cascading FK, and consider not dead-letterin **Resolution** -_Unresolved._ +Resolved 2026-11-09 (scadaproj#37, LocalDb `0.3.6`, unpublished). Re-verified with a 2-node test: a wiped peer +received `a_child` before `z_parent` and `node {"id":10}` before `node {"id":2}`. Both children were dead-lettered +(`SQLite Error 19: 'FOREIGN KEY constraint failed'`) and the watermark advanced past them. Fixed on both sides, and +the wire is unchanged: +- **Sender.** `SnapshotStreamer.ParentsFirstAsync` orders the tables in scope topologically by + `pragma_foreign_key_list` (Kahn's algorithm, ordinal name tie-break). Self-references are ignored, and a cycle + falls back to name order. +- **Receiver.** `LwwApplier.ApplyBatchAsync` gained `deferForeignKeyViolations`, used only by `SnapshotApplier`. An + entry that fails with SQLITE_CONSTRAINT_FOREIGNKEY (787) is rolled back to its savepoint and returned as deferred, + not dead-lettered. The applier retries deferred rows after every batch that applied something, until a pass makes + no progress. At `SnapshotComplete` it makes one final pass with deferral off, which dead-letters the true orphans + (with a warning). The delta path is unchanged, because commit order already puts parents first. + +Tests (`SnapshotForeignKeyTests`, all 4 fail pre-fix): +- 2-node wiped-peer snapshot with `MaxBatchSize = 1`, cross-table and self-referencing; +- streamer table order; +- a receiver fed child-first in the old sender order; +- a true orphan dead-lettered only at complete. + +README §Limitations documents the mixed-version caveat: a pre-0.3.6 receiver still loses child-first self-references. ### LocalDb-003 — `INSERT OR REPLACE` apply silently deletes a different row on a secondary `UNIQUE` conflict, so the pair diverges diff --git a/code-reviews/README.md b/code-reviews/README.md index 72abbe1..8c1b377 100644 --- a/code-reviews/README.md +++ b/code-reviews/README.md @@ -40,7 +40,7 @@ code-reviews/ ## Summary -10 of 10 libraries reviewed. 22 pending findings across all libraries. +10 of 10 libraries reviewed. 21 pending findings across all libraries. Coverage: Secrets, LocalDb, GalaxyRepository and Overview first reviewed, and Auth, Health, Theme, Audit, Configuration and Telemetry re-reviewed, on 2026-11-09 (#15). Each finding also @@ -50,9 +50,9 @@ carries a P1/P2/P3 priority (P1 = Critical/High, P2 = Medium, P3 = Low). |----------|---------------| | Critical | 0 | | High | 0 | -| Medium | 3 | +| Medium | 2 | | Low | 19 | -| **Total** | **22** | +| **Total** | **21** | ## Library Status @@ -63,7 +63,7 @@ carries a P1/P2/P3 priority (P1 = Critical/High, P2 = Medium, P3 = Low). | [Configuration](Configuration/findings.md) | 2026-11-09 | `91fb2b4` | 0/0/0/0 | 0 | 4 | | [GalaxyRepository](GalaxyRepository/findings.md) | 2026-11-09 | `91fb2b4` | 0/0/0/0 | 0 | 8 | | [Health](Health/findings.md) | 2026-11-09 | `91fb2b4` | 0/0/0/0 | 0 | 7 | -| [LocalDb](LocalDb/findings.md) | 2026-11-09 | `91fb2b4` | 0/0/3/3 | 6 | 7 | +| [LocalDb](LocalDb/findings.md) | 2026-11-09 | `91fb2b4` | 0/0/2/3 | 5 | 7 | | [Overview](Overview/findings.md) | 2026-11-09 | `91fb2b4` | 0/0/0/0 | 0 | 5 | | [Secrets](Secrets/findings.md) | 2026-11-09 | `91fb2b4` | 0/0/0/13 | 13 | 19 | | [Telemetry](Telemetry/findings.md) | 2026-11-09 | `91fb2b4` | 0/0/0/0 | 0 | 11 | @@ -84,11 +84,10 @@ _None open._ _None open._ -### Medium (3) +### Medium (2) | ID | Library | Title | |----|---------|-------| -| LocalDb-002 | [LocalDb](LocalDb/findings.md) | Snapshot streams tables by name and rows by `pk_json` text, so FK children land before parents and are dead-lettered for good | | LocalDb-003 | [LocalDb](LocalDb/findings.md) | `INSERT OR REPLACE` apply silently deletes a different row on a secondary `UNIQUE` conflict, so the pair diverges | | LocalDb-004 | [LocalDb](LocalDb/findings.md) | `FailClosedOnDrift` is bypassed on the snapshot path, so a future-stamped snapshot row moves the local HLC forward permanently |