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
This commit is contained in:
Joseph Doherty
2026-11-09 15:12:41 -05:00
parent fd46999946
commit b2adddbc8e
7 changed files with 376 additions and 7 deletions
@@ -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(
@@ -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;
}
}
@@ -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);
@@ -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);
@@ -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}.");