dohertj2/scadaproj
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

fix(localdb): snapshot delivers FK-linked rows in any order (scadaproj#37)

Browse Source 
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
This commit is contained in:
Joseph Doherty
2026-11-09 14:52:35 -05:00
parent 48813bba6f
commit 4af0ea8152
7 changed files with 351 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/LwwApplier.cs
+34 -5
View File
@@ -6,7 +6,15 @@ using ZB.MOM.WW.LocalDb.Registration;
namespace ZB.MOM.WW.LocalDb.Replication.Internal;
/// <summary>Outcome of applying one inbound batch. <see cref="AppliedThruSeq"/> is the batch's highest seq regardless of per-entry disposition, so the watermark advances even over discarded/dead-lettered entries.</summary>
internal sealed record ApplyResult(long AppliedThruSeq, int Applied, int Discarded, int DeadLettered);
/// <remarks>
/// <see cref="Deferred"/> 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.
/// </remarks>
internal sealed record ApplyResult(long AppliedThruSeq, int Applied, int Discarded, int DeadLettered)
{
public IReadOnlyList<OplogEntryRecord> Deferred { get; init; } = [];
}
/// <summary>
/// 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<DateTimeOffset>? utcNow
/// </summary>
internal int LastBatchStatementCount { get; private set; }
public async Task<ApplyResult> ApplyBatchAsync(IReadOnlyList<OplogEntryRecord> entries, CancellationToken ct)
/// <param name="entries">The batch, in apply order.</param>
/// <param name="ct">Cancellation.</param>
/// <param name="deferForeignKeyViolations">
/// 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 <see cref="ApplyResult.Deferred"/> 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.
/// </param>
public async Task<ApplyResult> ApplyBatchAsync(
IReadOnlyList<OplogEntryRecord> 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<DateTimeOffset>? utcNow
var applied = 0;
var discarded = 0;
var deadLettered = 0;
List<OplogEntryRecord>? 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<DateTimeOffset>? 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<DateTimeOffset>? utcNow
ex, recoveryFailure);
}
deadLettered++;
if (defer)
(deferred ??= []).Add(entry);
else
deadLettered++;
}
}
@@ -153,9 +179,12 @@ internal sealed class LwwApplier(SqliteLocalDb db, Func<DateTimeOffset>? 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 ?? [] };
}
/// <summary>SQLITE_CONSTRAINT_FOREIGNKEY (787): the statement failed its immediate FK check.</summary>
private static bool IsForeignKeyViolation(SqliteException ex) => ex.SqliteExtendedErrorCode == 787;
/// <summary>
/// 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
ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/SnapshotStreamer.cs
+86 -4
View File
@@ -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;
}
/// <summary>
/// Orders the tables so every table is streamed after the tables it references by foreign key
/// (LocalDb-002): name order put <c>a_child</c> before <c>z_parent</c>, 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.
/// </summary>
private async Task<IReadOnlyList<ReplicatedTable>> ParentsFirstAsync(List<ReplicatedTable> tables, CancellationToken ct)
{
var byName = tables.ToDictionary(t => t.Name, StringComparer.Ordinal);
var parents = new Dictionary<string, HashSet<string>>(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<ReplicatedTable>(tables.Count);
var remaining = new SortedSet<string>(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;
}
/// <summary>
/// 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<OplogEntryRecord> _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);
}
/// <summary>
/// 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.
/// </summary>
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.