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
This commit is contained in:
@@ -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).
|
**`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.
|
- **#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)
|
## Index history (migrated from the scadaproj root CLAUDE.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
|
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,
|
foreign keys, so a child table created *after* registration is caught at the next startup,
|
||||||
not at the moment of the `CREATE TABLE`.
|
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
|
- **CDC semantics** — rows that existed **before** `RegisterReplicated` are not captured
|
||||||
and are not snapshotted; only changes after registration are replicated. **Opt out with
|
and are not snapshotted; only changes after registration are replicated. **Opt out with
|
||||||
`RegisterReplicated(table, baselineExistingRows: true)`**, which seeds the row-version
|
`RegisterReplicated(table, baselineExistingRows: true)`**, which seeds the row-version
|
||||||
|
|||||||
@@ -6,7 +6,15 @@ using ZB.MOM.WW.LocalDb.Registration;
|
|||||||
namespace ZB.MOM.WW.LocalDb.Replication.Internal;
|
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>
|
/// <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>
|
/// <summary>
|
||||||
/// Applies a peer's oplog batch under last-writer-wins in one transaction: the applying guard is
|
/// 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>
|
/// </summary>
|
||||||
internal int LastBatchStatementCount { get; private set; }
|
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)
|
if (entries.Count == 0)
|
||||||
return new ApplyResult(0, 0, 0, 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 applied = 0;
|
||||||
var discarded = 0;
|
var discarded = 0;
|
||||||
var deadLettered = 0;
|
var deadLettered = 0;
|
||||||
|
List<OplogEntryRecord>? deferred = null;
|
||||||
|
|
||||||
await using var tx = await db.BeginPreparedTransactionAsync(ct);
|
await using var tx = await db.BeginPreparedTransactionAsync(ct);
|
||||||
using var sql = new PreparedStatements(tx.Connection, tx.Sqlite);
|
using var sql = new PreparedStatements(tx.Connection, tx.Sqlite);
|
||||||
@@ -120,9 +141,11 @@ internal sealed class LwwApplier(SqliteLocalDb db, Func<DateTimeOffset>? utcNow
|
|||||||
// statement, not the transaction, so the batch continues after dead-lettering.
|
// statement, not the transaction, so the batch continues after dead-lettering.
|
||||||
// Any other exception propagates with the savepoint still open, and disposing the
|
// Any other exception propagates with the savepoint still open, and disposing the
|
||||||
// outer transaction unwinds the whole batch (fail-closed).
|
// outer transaction unwinds the whole batch (fail-closed).
|
||||||
|
var defer = deferForeignKeyViolations && IsForeignKeyViolation(ex);
|
||||||
try
|
try
|
||||||
{
|
{
|
||||||
await sql.RollbackEntryAsync(ct);
|
await sql.RollbackEntryAsync(ct);
|
||||||
|
if (!defer)
|
||||||
await sql.DeadLetterAsync(entry, ex.Message, utcNow, ct);
|
await sql.DeadLetterAsync(entry, ex.Message, utcNow, ct);
|
||||||
}
|
}
|
||||||
catch (Exception recoveryFailure) when (recoveryFailure is not OperationCanceledException)
|
catch (Exception recoveryFailure) when (recoveryFailure is not OperationCanceledException)
|
||||||
@@ -141,6 +164,9 @@ internal sealed class LwwApplier(SqliteLocalDb db, Func<DateTimeOffset>? utcNow
|
|||||||
ex, recoveryFailure);
|
ex, recoveryFailure);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if (defer)
|
||||||
|
(deferred ??= []).Add(entry);
|
||||||
|
else
|
||||||
deadLettered++;
|
deadLettered++;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -153,9 +179,12 @@ internal sealed class LwwApplier(SqliteLocalDb db, Func<DateTimeOffset>? utcNow
|
|||||||
await tx.CommitAsync(ct);
|
await tx.CommitAsync(ct);
|
||||||
|
|
||||||
db.Clock.Observe(maxHlc);
|
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>
|
/// <summary>
|
||||||
/// The statement set one apply-batch runs on: each distinct statement is built and prepared on
|
/// 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
|
/// first use, then rebound and re-executed per row. Statements whose SQL depends on the table
|
||||||
|
|||||||
@@ -76,9 +76,8 @@ internal sealed class SnapshotStreamer(SqliteLocalDb db, OplogStore store, Repli
|
|||||||
await send(new SyncMessage { SnapshotBegin = begin }, ct);
|
await send(new SyncMessage { SnapshotBegin = begin }, ct);
|
||||||
|
|
||||||
var totalRows = 0L;
|
var totalRows = 0L;
|
||||||
foreach (var table in db.ReplicatedTables.Values
|
foreach (var table in await ParentsFirstAsync(
|
||||||
.Where(t => scope.Includes(t.Name))
|
db.ReplicatedTables.Values.Where(t => scope.Includes(t.Name)).ToList(), ct))
|
||||||
.OrderBy(t => t.Name, StringComparer.Ordinal))
|
|
||||||
{
|
{
|
||||||
var sql = BuildPageSql(table);
|
var sql = BuildPageSql(table);
|
||||||
// Keyset cursor (pk_json > last seen), not LIMIT/OFFSET: concurrent inserts/prunes
|
// 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;
|
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>
|
/// <summary>
|
||||||
/// Serialized bytes one row contributes to its batch message: the row's own encoding plus the
|
/// 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.
|
/// 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 long _maxHlc;
|
||||||
private bool _partial;
|
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
|
// 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.
|
// 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)
|
public SnapshotApplier(SqliteLocalDb db, LwwApplier applier, OplogStore store, ILogger logger)
|
||||||
@@ -236,6 +275,7 @@ internal sealed class SnapshotApplier : ISnapshotApplier
|
|||||||
_asOfSeq = begin.AsOfSeq;
|
_asOfSeq = begin.AsOfSeq;
|
||||||
_rows = 0;
|
_rows = 0;
|
||||||
_maxHlc = 0;
|
_maxHlc = 0;
|
||||||
|
_pending = [];
|
||||||
// A non-empty table list scopes the snapshot; empty (which is all a pre-0.3.0 sender can
|
// 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
|
// 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.
|
// 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(
|
entries.Add(new OplogEntryRecord(
|
||||||
0, batch.TableName, row.PkJson, row.HasRowJson ? row.RowJson : null, row.Hlc, row.NodeId, row.IsTombstone));
|
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;
|
_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)
|
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
|
// 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
|
// 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.
|
// 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;
|
var seq = _partial ? 0 : complete.AsOfSeq;
|
||||||
// SetLastAppliedRemoteSeqAsync is MAX-guarded, so a delta applied mid-snapshot with a seq
|
// 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.
|
// above the snapshot's as-of (or an HLC above what the snapshot carried) is never regressed.
|
||||||
|
|||||||
@@ -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;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// 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 <c>pk_json</c> 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.
|
||||||
|
/// </summary>
|
||||||
|
public sealed class SnapshotForeignKeyTests : IDisposable
|
||||||
|
{
|
||||||
|
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 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<Side> 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<SyncMessage>();
|
||||||
|
var bToA = Channel.CreateUnbounded<SyncMessage>();
|
||||||
|
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<long> Count(SqliteLocalDb db, string table) =>
|
||||||
|
(await db.QueryAsync($"SELECT COUNT(*) FROM {table}", r => r.GetInt64(0)))[0];
|
||||||
|
|
||||||
|
private static async Task<List<string>> 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<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) { }
|
||||||
|
}
|
||||||
|
|
||||||
|
[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<string>();
|
||||||
|
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"));
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -10,7 +10,7 @@
|
|||||||
| Last reviewed | 2026-11-09 |
|
| Last reviewed | 2026-11-09 |
|
||||||
| Reviewer | Claude |
|
| Reviewer | Claude |
|
||||||
| Commit reviewed | `91fb2b4` |
|
| Commit reviewed | `91fb2b4` |
|
||||||
| Open findings | 6 |
|
| Open findings | 5 |
|
||||||
|
|
||||||
## Summary
|
## Summary
|
||||||
|
|
||||||
@@ -150,7 +150,7 @@ non-displacement cases fail.
|
|||||||
| Severity | Medium |
|
| Severity | Medium |
|
||||||
| Priority | P2 |
|
| Priority | P2 |
|
||||||
| Category | Correctness & logic bugs |
|
| Category | Correctness & logic bugs |
|
||||||
| Status | Open |
|
| Status | Resolved |
|
||||||
| Location | `ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/SnapshotStreamer.cs:79` |
|
| Location | `ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/SnapshotStreamer.cs:79` |
|
||||||
|
|
||||||
**Description**
|
**Description**
|
||||||
@@ -203,7 +203,26 @@ Also add a snapshot test with a non-cascading FK, and consider not dead-letterin
|
|||||||
|
|
||||||
**Resolution**
|
**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
|
### LocalDb-003 — `INSERT OR REPLACE` apply silently deletes a different row on a secondary `UNIQUE` conflict, so the pair diverges
|
||||||
|
|
||||||
|
|||||||
@@ -40,7 +40,7 @@ code-reviews/
|
|||||||
|
|
||||||
## Summary
|
## 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,
|
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
|
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 |
|
| Critical | 0 |
|
||||||
| High | 0 |
|
| High | 0 |
|
||||||
| Medium | 3 |
|
| Medium | 2 |
|
||||||
| Low | 19 |
|
| Low | 19 |
|
||||||
| **Total** | **22** |
|
| **Total** | **21** |
|
||||||
|
|
||||||
## Library Status
|
## 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 |
|
| [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 |
|
| [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 |
|
| [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 |
|
| [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 |
|
| [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 |
|
| [Telemetry](Telemetry/findings.md) | 2026-11-09 | `91fb2b4` | 0/0/0/0 | 0 | 11 |
|
||||||
@@ -84,11 +84,10 @@ _None open._
|
|||||||
|
|
||||||
_None open._
|
_None open._
|
||||||
|
|
||||||
### Medium (3)
|
### Medium (2)
|
||||||
|
|
||||||
| ID | Library | Title |
|
| 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-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 |
|
| 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 |
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user