Compare commits

..

1 Commits

Author SHA1 Message Date
Joseph Doherty 41bdc178c2 fix(localdb): refuse cascade-delete FK parents; make one entry's row+ledger writes atomic (0.3.3, ScadaBridge arch-review P4-7)
Two halves of the same INSERT OR REPLACE hazard in the LWW apply path. Wire-compatible
with every earlier peer: no proto change, no lib_schema_version bump, no on-disk schema
change.

1. RegisterReplicated now REFUSES a table that another table references with a
   row-mutating ON DELETE action (CASCADE / SET NULL / SET DEFAULT). Applying a remote row
   uses INSERT OR REPLACE, whose implicit DELETE fires that referential action — while the
   apply path holds the capture guard raised, precisely so its own writes are not
   re-captured. The child rows it deletes or nulls therefore produce no oplog entry: the
   peer keeps them and the pair diverges with nothing reporting a fault. Registration is
   the only point at which the problem is still visible, so it is refused there with an
   actionable LocalDbRegistrationException naming the referencing table and its action.
   NO ACTION / RESTRICT stay allowed: they mutate no child row, so the implicit delete just
   fails and the applier dead-letters the entry — loud, not silent. Registration-time only;
   a cascading child table created LATER is not caught (SQLite offers no DDL hook).

   No current consumer has a foreign key on a replicated table (ScadaBridge's ten tables
   have none), so this is a guard against a future foot-gun, not a live fix.

2. One entry's row write and its ledger write are now atomic. They were two statements
   under the poison catch, so a failure landing between them committed a deleted/replaced
   row carrying a STALE LIVE row-version entry — exactly the shape SnapshotStreamer warns
   about and skips, so that key never travels again and the pair is permanently divergent
   for it. A per-entry SAVEPOINT makes the pair all-or-nothing while keeping the catch's
   per-ENTRY granularity: entries applied earlier in the same batch survive the rollback,
   the poison entry is still dead-lettered, and the batch still continues. The RELEASE runs
   outside the catch on purpose, so a failure there rolls the whole batch back rather than
   being swallowed as poison. Measured cost on a 100k-row backlog drain: ~7%
   (178.6k rows/s -> 166.1k rows/s).

Tests: both hazards proven against the unfixed code first (FK parent registered silently;
an injected ledger-write abort left the row deleted with a live ledger entry). Adds FK
guard coverage (parent refused, child allowed, SET NULL / SET DEFAULT refused, NO ACTION /
RESTRICT allowed, unrelated cascades ignored) and two applier atomicity tests. The
statement-count lock-in moves 7 -> 9 for the two new savepoint statements.

Version 0.3.2 -> 0.3.3; README limitations + release notes document the constraint.
2026-08-19 16:22:44 -04:00
10 changed files with 248 additions and 387 deletions
+1 -1
View File
File diff suppressed because one or more lines are too long
+1 -1
View File
@@ -5,7 +5,7 @@
<Nullable>enable</Nullable>
<ImplicitUsings>enable</ImplicitUsings>
<LangVersion>latest</LangVersion>
<Version>0.3.4</Version>
<Version>0.3.3</Version>
<ManagePackageVersionsCentrally>true</ManagePackageVersionsCentrally>
<TreatWarningsAsErrors>true</TreatWarningsAsErrors>
<!-- NU1510 (package-pruning advice): the Replication and Tests projects carry a
+32 -45
View File
@@ -10,51 +10,35 @@ Change capture uses SQLite triggers writing a single HLC-stamped oplog inside th
consumer's own write transaction (crash-consistent, non-bypassable). Convergence is
last-writer-wins over a hybrid logical clock. All timestamps are UTC.
## Packages (all `0.3.4`)
## Packages (all `0.3.3`)
| Package | Contents |
|---|---|
| `ZB.MOM.WW.LocalDb` | Embedded DB host: `ILocalDb` SQL surface (`ExecuteAsync` / `QueryAsync` / transactions), `RegisterReplicated(table[, baselineExistingRows])` (PK + BLOB + foreign-key validation, trigger install, oplog schema), `DeregisterReplicated(table)` (trigger removal + oplog/ledger prune), `HybridLogicalClock`, `AddZbLocalDb()` DI. |
| `ZB.MOM.WW.LocalDb` | Embedded DB host: `ILocalDb` SQL surface (`ExecuteAsync` / `QueryAsync` / transactions), `RegisterReplicated(table[, baselineExistingRows])` (PK validation + trigger install + oplog schema), `DeregisterReplicated(table)` (trigger removal + oplog/ledger prune), `HybridLogicalClock`, `AddZbLocalDb()` DI. |
| `ZB.MOM.WW.LocalDb.Contracts` | `localdb_sync.v1` proto + generated gRPC types (packable; a future peer could be non-.NET). |
| `ZB.MOM.WW.LocalDb.Replication` | gRPC sync engine: passive `LocalDbSyncService`, initiating `SyncBackgroundService`, `MaintenanceBackgroundService`, LWW apply, watermark acks + pruning, snapshot resync, `ISyncStatus`, `LocalDbMetrics`, `AddZbLocalDbReplication()` / `MapZbLocalDbSync()`. |
> Status: **published to the Gitea feed; adopted by ScadaBridge (Phases 1+2) and OtOpcUa (Phase 1).**
>
> - `0.3.4` (2026-08-20) — **a converged pair no longer re-streams a full snapshot on every
> reconnect once the sender's oplog has been ack-pruned to empty.** Found on the ScadaBridge
> docker rig (its register row 106): a rebuilt node (watermark 0) was healed by a snapshot
> stamped `as_of_seq = MAX(seq)` — which is 0 on an empty oplog — so its MAX-guarded watermark
> stayed 0, and the healthy side's empty-oplog horizon rule (`0.1.2`: streamable from
> `last_acked_seq + 1`) saw a gap on every later handshake and re-sent the whole database
> (127k rows per restart on the rig) until it happened to write a new row. The as-of is now
> `max(MAX(seq), last_acked_seq)`: every delta pruned below `last_acked_seq` is already folded
> into the rows the snapshot carries, so that is the floor it may claim. Sender-side only,
> wire-compatible, no schema change; pinned by
> `WipedPeer_SnapshotLiftsItsWatermark_SoTheNextReconnectSendsNoSecondSnapshot`.
> - `0.3.3` (2026-08-20) — two latent apply-path hazards from the ScadaBridge adversarial
> architecture review (P4-7). Neither had a live trigger in any current consumer; both are the
> kind that surface as silent divergence, not as an error. Wire-compatible: no proto change, no
> - `0.3.3` (2026-08-19) — one Low finding from the ScadaBridge adversarial architecture review
> (P4-7), in two halves. Wire-compatible with every earlier peer: no proto change, no
> `lib_schema_version` bump, no on-disk schema change.
> - **A replicated table may no longer be the parent of a row-mutating `ON DELETE` foreign key.**
> The apply path writes remote rows with `INSERT OR REPLACE`, and REPLACE resolves a PK conflict
> by DELETING the existing row first. Connections run `PRAGMA foreign_keys=ON`, so that implicit
> delete fires the child's `ON DELETE CASCADE` / `SET NULL` / `SET DEFAULT` — while the capture
> guard is raised. The child rows change on the applying node ONLY, produce no oplog entry, and
> the pair diverges with nothing able to detect it (LWW compares only rows it was told about).
> `RegisterReplicated` now inspects every foreign key in the database and refuses such a table
> with a `LocalDbRegistrationException` naming the offending child column and action.
> `RESTRICT` / `NO ACTION` remain fine — they abort the statement instead of mutating rows
> behind capture's back, which the apply path already handles as per-entry poison. Checked at
> registration; SQLite cannot add a foreign key to an existing table, so a child table created
> later is caught at the next startup's registration rather than at the DDL.
> - **An entry's row write and its ledger write are now atomic with each other.** They are two
> statements, and SQLite aborts only the failing one — so a statement-level fault between them
> (the case the per-entry poison catch exists for) committed a half-applied entry: a tombstone
> whose table row was gone while `__localdb_row_version` still called it live, which the snapshot
> streamer can only warn about and skip, so the peer never learned the row died; or an upsert
> whose new row carried the OLD version and therefore lost the next comparison it should have
> won. The pair is wrapped in a per-entry `SAVEPOINT` that is rolled back before dead-lettering,
> so a poison entry leaves both writes or neither and the rest of the batch is untouched.
> - **`RegisterReplicated` now refuses a table that another table cascades a DELETE onto.**
> Applying a remote row uses `INSERT OR REPLACE`, whose implicit DELETE fires `ON DELETE
> CASCADE` / `SET NULL` / `SET DEFAULT` on every child — while the apply path holds the capture
> guard raised, so the child rows it touches produce no oplog entry, the peer keeps them, and the
> pair diverges with nothing reporting a fault. Registration is the only place that is visible,
> so it is refused there with an actionable `LocalDbRegistrationException` naming the referencing
> table and its action. `NO ACTION` / `RESTRICT` are still allowed: they mutate no child row (the
> implicit delete simply fails, which the applier dead-letters — loud, not silent). No current
> consumer has a foreign key on a replicated table, so this is a guard against a future foot-gun
> rather than a live fix.
> - **Applying one entry is atomic across its row write and its ledger write.** They were two
> statements under the poison catch, so a failure landing between them committed a
> deleted/replaced row carrying a STALE LIVE ledger version — the shape `SnapshotStreamer` warns
> about and skips, meaning that key never travels again. A per-entry `SAVEPOINT` makes the pair
> all-or-nothing while keeping the catch's per-ENTRY granularity: entries applied earlier in the
> same batch survive, and the poison entry is still dead-lettered and the batch still continues.
> - `0.3.2` (2026-08-19) — two Medium findings from the ScadaBridge adversarial architecture review
> (P4-1, P4-3). Wire-compatible with every earlier peer: no proto change, no `lib_schema_version`
> bump, no on-disk schema change — the keepalive is transport-level and the rest is local policy.
@@ -295,15 +279,18 @@ All options are validated at startup (`ZB.MOM.WW.Configuration` validators).
`RegisterReplicated`, because `json_object` cannot capture it). A *typeless* column can
still hold a BLOB value at runtime, which then surfaces as a write-time error, not a
registration error.
- **A replicated table may not be the parent of a row-mutating `ON DELETE` foreign key**
(`CASCADE` / `SET NULL` / `SET DEFAULT` from any other table, or from itself). Applying a
remote row uses `INSERT OR REPLACE`, whose implicit delete fires that action with change
capture suppressed — the child rows would change on the applying node only and never
replicate, diverging the pair undetectably. `RegisterReplicated` refuses such a table
(`0.3.3`); use `ON DELETE NO ACTION`/`RESTRICT` and delete children explicitly, or leave
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`.
- **Replicated tables must not be cascade-delete foreign-key parents.** If another table
references the one you are registering with `ON DELETE CASCADE` (or `SET NULL` /
`SET DEFAULT`), `RegisterReplicated` throws. Applying a remote row uses `INSERT OR
REPLACE`, whose implicit DELETE fires that referential action while change capture is
suppressed — so the child rows it deletes or nulls are never captured, the peer keeps
them, and the pair diverges with nothing reporting a fault. Model the parent/child
relationship with `ON DELETE NO ACTION` (or `RESTRICT`) and delete children explicitly, or
leave the parent unreplicated. `NO ACTION` / `RESTRICT` are allowed because they mutate no
child row: the implicit delete just fails, and the applier dead-letters the entry — loud,
not silent. The check runs at **registration time**, so a cascading child table created
*later* is not caught (SQLite offers no DDL hook); register after schema setup, which is
where startup does it anyway.
- **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
@@ -12,9 +12,10 @@ internal sealed record ApplyResult(long AppliedThruSeq, int Applied, int Discard
/// Applies a peer's oplog batch under last-writer-wins in one transaction: the applying guard is
/// raised so capture triggers skip these writes, per-entry LWW resolves against the local
/// row-version (hlc, then node_id ordinal tie-break), poison rows are dead-lettered while the
/// batch continues — each entry's row write and ledger write sharing a savepoint so a dead-lettered
/// entry leaves neither — and the remote watermark advances in the SAME transaction as the data so a
/// batch continues, and the remote watermark advances in the SAME transaction as the data so a
/// crash-replay of the batch is a no-op (LWW's strict-greater compare loses on equal versions).
/// Each entry's row write and its ledger write are additionally scoped by a per-entry SAVEPOINT, so
/// the pair is all-or-nothing even when the poison catch keeps the batch alive around it.
/// This never writes <c>__localdb_oplog</c>: the 2-node topology does not re-forward applied deltas.
/// </summary>
/// <remarks>
@@ -89,60 +90,50 @@ internal sealed class LwwApplier(SqliteLocalDb db, Func<DateTimeOffset>? utcNow
continue;
}
var table = db.ReplicatedTables[entry.TableName];
// An entry is TWO statements — the row change and its ledger row — and SQLite aborts
// only the failing statement, so a fault between them used to commit a half-applied
// entry: a tombstone whose table row is gone while __localdb_row_version still calls it
// live (which SnapshotStreamer can only warn about and skip, so the peer never learns
// the row died), or an upsert whose new row carries the OLD version and therefore loses
// the next LWW comparison it should win. A per-entry savepoint makes the pair
// all-or-nothing without widening the poison catch to the whole batch.
//
// Opened OUTSIDE the try on purpose: a SAVEPOINT that failed opened nothing, so the
// handler's ROLLBACK TO would name a savepoint that does not exist and fault in turn.
await sql.EnterEntryAsync(ct);
// The row mutation and its ledger entry are TWO statements and must land together or
// not at all: a failure between them commits a deleted/replaced row carrying a stale
// ledger version, which SnapshotStreamer then warns about and skips — so that key never
// travels again and the pair is silently, permanently divergent for it. A per-entry
// savepoint makes the pair atomic without giving up the poison catch's per-ENTRY
// granularity (the surrounding transaction, and every entry already applied in it,
// survives a rollback to the savepoint). Cost measured on the path that matters — a
// 100 000-row backlog drain — is ~7% (178.6k rows/s → 166.1k rows/s): the two extra
// statements are prepared once like the rest and neither one touches the disk, since
// an inner savepoint's release just merges into the enclosing transaction.
var pairApplied = false;
await sql.SavepointAsync(ct);
try
{
var table = db.ReplicatedTables[entry.TableName];
if (entry.IsTombstone)
await sql.DeleteAsync(table, entry.PkJson, ct);
else
await sql.UpsertAsync(table, entry.RowJson!, ct);
await sql.WriteRowVersionAsync(entry, utcNow, ct);
await sql.ReleaseEntryAsync(ct);
applied++;
pairApplied = true;
}
catch (SqliteException ex)
{
// Deliberately narrow: only statement-level SQLite errors (e.g. json_extract on
// malformed row_json) are treated as per-entry poison — SQLite aborts just the
// 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).
try
{
await sql.RollbackEntryAsync(ct);
await sql.DeadLetterAsync(entry, ex.Message, utcNow, ct);
}
catch (Exception recoveryFailure) when (recoveryFailure is not OperationCanceledException)
{
// Recovering from the poison entry itself failed — a wedged connection, an I/O
// error, a dead-letter INSERT that faults. The outcome is unchanged and still
// fail-closed (nothing commits; disposing the batch transaction unwinds it),
// but reporting only the recovery failure would throw away the ORIGINAL cause,
// which is the one that says WHY this entry is poison. Both are carried out:
// the original as InnerException (AggregateException surfaces the first inner
// there), the recovery failure alongside it with its own stack intact.
throw new AggregateException(
$"Failed to dead-letter a poison oplog entry for table '{entry.TableName}' " +
$"(seq {entry.Seq}): {recoveryFailure.Message}. The entry's own fault is the " +
"inner exception. The batch is not committed.",
ex, recoveryFailure);
}
// Any other exception propagates and rolls back the whole batch (fail-closed).
// The rollback comes FIRST so the dead-letter row it records is not discarded with
// the half-applied pair.
await sql.RollbackToSavepointAsync(ct);
await sql.DeadLetterAsync(entry, ex.Message, utcNow, ct);
deadLettered++;
}
// Released outside the catch on purpose: a RELEASE that itself fails means the
// transaction is no longer sound, and there is no savepoint state left to reason about,
// so it propagates and rolls the whole batch back rather than being swallowed as poison.
if (pairApplied)
{
await sql.ReleaseSavepointAsync(ct);
applied++;
}
}
await sql.AdvancePeerStateAsync(maxSeq, maxHlc, utcNow, ct);
@@ -173,21 +164,47 @@ internal sealed class LwwApplier(SqliteLocalDb db, Func<DateTimeOffset>? utcNow
private readonly Dictionary<string, SqliteCommand> _deletes = new(StringComparer.Ordinal);
private readonly Dictionary<string, SqliteCommand> _upserts = new(StringComparer.Ordinal);
/// <summary>Name of the per-entry savepoint. Fixed: only one is ever open at a time.</summary>
private const string EntrySavepoint = "zb_lww_entry";
private SqliteCommand? _applying;
private SqliteCommand? _enterEntry;
private SqliteCommand? _releaseEntry;
private SqliteCommand? _rollbackEntry;
private SqliteCommand? _lookup;
private SqliteCommand? _rowVersion;
private SqliteCommand? _deadLetter;
private SqliteCommand? _peerState;
private SqliteCommand? _savepoint;
private SqliteCommand? _releaseSavepoint;
private SqliteCommand? _rollbackToSavepoint;
/// <summary>The per-entry savepoint's name. Reused for every entry — one is open at a time.</summary>
private const string SavepointName = "__localdb_apply_entry";
/// <summary>How many distinct statements this set has prepared.</summary>
public int StatementCount => _owned.Count;
/// <summary>Opens the per-entry savepoint that scopes one entry's row + ledger writes.</summary>
public Task SavepointAsync(CancellationToken ct)
{
var cmd = _savepoint ??= Create("SAVEPOINT " + SavepointName);
return cmd.ExecuteNonQueryAsync(ct);
}
/// <summary>Commits the entry's savepoint into the surrounding batch transaction.</summary>
public Task ReleaseSavepointAsync(CancellationToken ct)
{
var cmd = _releaseSavepoint ??= Create("RELEASE " + SavepointName);
return cmd.ExecuteNonQueryAsync(ct);
}
/// <summary>
/// Undoes everything the entry did and closes its savepoint. ROLLBACK TO leaves the savepoint
/// OPEN, so the RELEASE is required — without it the names would nest one deeper per poison
/// entry and the next rollback would unwind more than one entry's work.
/// </summary>
public async Task RollbackToSavepointAsync(CancellationToken ct)
{
var cmd = _rollbackToSavepoint ??= Create("ROLLBACK TO " + SavepointName);
await cmd.ExecuteNonQueryAsync(ct);
await ReleaseSavepointAsync(ct);
}
public Task SetApplyingAsync(int value, CancellationToken ct)
{
var cmd = _applying ??= Create(
@@ -196,32 +213,6 @@ internal sealed class LwwApplier(SqliteLocalDb db, Func<DateTimeOffset>? utcNow
return cmd.ExecuteNonQueryAsync(ct);
}
/// <summary>Opens the per-entry savepoint inside the batch transaction.</summary>
public Task EnterEntryAsync(CancellationToken ct)
{
var cmd = _enterEntry ??= Create($"SAVEPOINT {EntrySavepoint}");
return cmd.ExecuteNonQueryAsync(ct);
}
/// <summary>Commits the entry's work into the batch transaction (pops the savepoint).</summary>
public Task ReleaseEntryAsync(CancellationToken ct)
{
var cmd = _releaseEntry ??= Create($"RELEASE {EntrySavepoint}");
return cmd.ExecuteNonQueryAsync(ct);
}
/// <summary>
/// Undoes everything the entry did and pops the savepoint, leaving the batch transaction
/// open and usable. Two statements, because ROLLBACK TO rewinds to the savepoint but leaves
/// it ON the stack — without the RELEASE, every dead-lettered entry would nest another one.
/// </summary>
public async Task RollbackEntryAsync(CancellationToken ct)
{
var cmd = _rollbackEntry ??= Create($"ROLLBACK TO {EntrySavepoint}");
await cmd.ExecuteNonQueryAsync(ct);
await ReleaseEntryAsync(ct);
}
public async Task<(long Hlc, string NodeId)?> ReadLocalVersionAsync(
string tableName, string pkJson, CancellationToken ct)
{
@@ -61,16 +61,7 @@ internal sealed class SnapshotStreamer(SqliteLocalDb db, OplogStore store, Repli
/// </summary>
public async Task<long> SendAsync(SnapshotScope scope, Func<SyncMessage, CancellationToken, Task> send, CancellationToken ct)
{
// The as-of is the highest seq this snapshot can stand in for. MAX(seq) alone is wrong once
// ack-pruning has emptied the oplog: it reads 0, and a receiver whose watermark genuinely IS
// 0 (a rebuilt node) is then never lifted by the snapshot — its MAX-guarded watermark stays
// 0, the empty-oplog horizon rule (streamable from last_acked_seq + 1) sees a gap on the
// next handshake, and the whole database is re-streamed on every reconnect until this node
// happens to write again. Every delta pruned below last_acked_seq is already folded into the
// rows this snapshot carries, so last_acked_seq is the floor the as-of must carry.
var asOfSeq = Math.Max(
await store.GetMaxSeqAsync(ct),
(await store.GetPeerStateAsync(ct)).LastAckedSeq);
var asOfSeq = await store.GetMaxSeqAsync(ct);
var begin = new SnapshotBegin { AsOfSeq = asOfSeq };
begin.Tables.AddRange(scope.Tables);
await send(new SyncMessage { SnapshotBegin = begin }, ct);
@@ -38,19 +38,17 @@ public interface ILocalDb
/// schema digest, and installs the three AFTER triggers that capture every change into the
/// oplog + row-version tables within the same transaction as the consumer's write. Idempotent
/// (re-registering regenerates the triggers). Throws <see cref="LocalDbRegistrationException"/>
/// if the name is unsafe, the table is missing, it has no primary key, or it is the parent of a
/// foreign key whose <c>ON DELETE</c> action mutates the child (<c>CASCADE</c> / <c>SET NULL</c> /
/// <c>SET DEFAULT</c>) — see the remarks for why that last one cannot be replicated.
/// if the name is unsafe, the table is missing, it has no primary key, it has a BLOB column, or
/// another table references it with a row-mutating <c>ON DELETE</c> action.
/// </summary>
/// <remarks>
/// <b>Replicated tables may not be the parent of a cascading foreign key.</b> The apply path
/// writes remote rows with <c>INSERT OR REPLACE</c>; REPLACE deletes the conflicting row first,
/// and with <c>PRAGMA foreign_keys=ON</c> that delete fires the child's <c>ON DELETE</c> action
/// while change capture is suppressed. The child rows would be mutated on the applying node only,
/// never captured and never shipped, diverging the pair with nothing able to detect it. The check
/// runs at registration and inspects the whole database's foreign keys; SQLite cannot add a
/// foreign key to an existing table, so a later-created child table is caught on the next
/// registration (i.e. the next startup), not at the moment of the DDL.
/// <b>Replicated tables must not be cascade-delete foreign-key parents.</b> Applying a remote row
/// uses <c>INSERT OR REPLACE</c>, whose implicit DELETE fires <c>ON DELETE CASCADE</c> /
/// <c>SET NULL</c> / <c>SET DEFAULT</c> on referencing tables while change capture is suppressed
/// — the child rows it touches are never captured, so the peer keeps them and the pair diverges
/// silently. Such a table is refused here, because registration is the only point at which the
/// problem is still visible. <c>NO ACTION</c> / <c>RESTRICT</c> are allowed: they mutate no child
/// row, so the implicit delete simply fails and the applier dead-letters the entry.
/// </remarks>
ReplicatedTable RegisterReplicated(string tableName);
@@ -199,7 +199,7 @@ internal sealed class SqliteLocalDb : ILocalDb, IDisposable
throw new LocalDbRegistrationException(
$"Cannot register table '{tableName}': replication requires an explicit primary key.");
EnsureNoInboundActionOnDelete(tableName);
EnsureNoRowMutatingDeleteAction(tableName);
var allColumns = columns.Select(c => (c.Name, c.Type)).ToList();
var digest = ReplicatedTable.ComputeDigest(tableName, pkColumns, allColumns);
@@ -296,65 +296,86 @@ internal sealed class SqliteLocalDb : ILocalDb, IDisposable
}
/// <summary>
/// Refuses to register a table that is the PARENT of a foreign key whose <c>ON DELETE</c> action
/// mutates the child (<c>CASCADE</c>, <c>SET NULL</c>, <c>SET DEFAULT</c>).
/// Refuses to register <paramref name="tableName"/> when another table references it with an
/// <c>ON DELETE</c> action that MUTATES the referencing rows (<c>CASCADE</c>, <c>SET NULL</c>,
/// <c>SET DEFAULT</c>).
/// </summary>
/// <remarks>
/// <para>
/// <b>Why this is a replication hazard, not a schema opinion.</b> The apply path writes a remote
/// row with <c>INSERT OR REPLACE</c>, and REPLACE resolves a primary-key conflict by DELETING the
/// existing row before inserting the new one. Connections run <c>PRAGMA foreign_keys=ON</c>, and
/// that implicit delete fires foreign-key actions (verified behaviour, not inference). It runs
/// with the capture guard raised, so the resulting child-row deletes/nulls are invisible to the
/// oplog: they happen on this node and are never shipped. The peer, applying the same parent row,
/// keeps its children — and the pair has silently diverged with nothing to detect it, since LWW
/// only ever compares rows it was told about.
/// <b>Why this is a replication correctness rule and not a schema opinion.</b> Applying a peer's
/// row runs <c>INSERT OR REPLACE</c>, and REPLACE resolves a primary-key conflict by DELETING the
/// existing row first. Connections run with <c>PRAGMA foreign_keys=ON</c>, so that implicit delete
/// fires the referential action on every child table — while the apply path holds the capture
/// guard raised, precisely so its own writes are not re-captured. The child rows are therefore
/// deleted (or nulled) locally and produce NO oplog entry: the peer never hears about them, keeps
/// them, and the pair is durably divergent with nothing anywhere reporting a fault.
/// </para>
/// <para>
/// <b>Checked at registration only.</b> SQLite cannot add a foreign key to an existing table, so
/// a new offender can only arrive as a child table created (or a parent table rebuilt) after this
/// call. Consumers register their tables at startup, after schema creation/migration, which is
/// where the check belongs; a schema change that adds a cascading child to an already-registered
/// parent is caught on the next boot, not at the moment of the DDL.
/// Refusing at registration is the only place the problem is visible. At apply time the damage is
/// already done and indistinguishable from a normal write, and no runtime signal exists to detect
/// it. <b>NO ACTION / RESTRICT are deliberately allowed</b>: they mutate no child row — the
/// implicit delete simply fails, which the applier surfaces as a dead-lettered entry. Loud, not
/// silent, so there is nothing to guard.
/// </para>
/// <para>
/// <c>RESTRICT</c> / <c>NO ACTION</c> are deliberately allowed: they abort the offending statement
/// with a constraint error rather than mutating rows behind capture's back, which the apply path
/// already handles as per-entry poison.
/// The check is registration-time, so a cascading child table CREATED LATER is not caught (SQLite
/// has no DDL hook to catch it on). Registration normally runs at startup after schema setup,
/// which is where the constraint belongs.
/// </para>
/// </remarks>
private void EnsureNoInboundActionOnDelete(string tableName)
private void EnsureNoRowMutatingDeleteAction(string tableName)
{
var offenders = new List<string>();
var candidates = new List<string>();
using (var cmd = _master.CreateCommand())
{
// pragma_foreign_key_list is the table-valued form of PRAGMA foreign_key_list, so the
// whole database is one query: fk."table" is the PARENT this child points at.
cmd.CommandText =
"SELECT m.name, fk.\"from\", fk.\"on_delete\" " +
"FROM sqlite_master m JOIN pragma_foreign_key_list(m.name) fk " +
"WHERE m.type = 'table' AND fk.\"table\" = $t COLLATE NOCASE " +
"AND UPPER(fk.\"on_delete\") IN ('CASCADE', 'SET NULL', 'SET DEFAULT') " +
"ORDER BY m.name, fk.\"from\"";
cmd.Parameters.AddWithValue("$t", tableName);
// Every user table is a candidate CHILD; the internal sqlite_* tables carry no
// foreign keys. The registered table itself is included: a self-referencing cascade
// is the same hazard.
cmd.CommandText = "SELECT name FROM sqlite_master WHERE type = 'table' AND name NOT LIKE 'sqlite_%'";
using var reader = cmd.ExecuteReader();
while (reader.Read())
offenders.Add(
$"{reader.GetString(0)}.{reader.GetString(1)} ON DELETE {reader.GetString(2).ToUpperInvariant()}");
candidates.Add(reader.GetString(0));
}
if (offenders.Count == 0)
return;
foreach (var child in candidates)
{
using var cmd = _master.CreateCommand();
cmd.CommandText = $"PRAGMA foreign_key_list({TriggerSqlGenerator.Ident(child)})";
using var reader = cmd.ExecuteReader();
while (reader.Read())
{
// foreign_key_list cols: 0=id, 1=seq, 2=table (the PARENT), 3=from, 4=to,
// 5=on_update, 6=on_delete, 7=match.
var parent = reader.IsDBNull(2) ? "" : reader.GetString(2);
// SQLite identifiers are case-insensitive, and the REFERENCES clause need not
// match the CREATE TABLE spelling.
if (!string.Equals(parent, tableName, StringComparison.OrdinalIgnoreCase))
continue;
throw new LocalDbRegistrationException(
$"Cannot register table '{tableName}': it is the parent of a foreign key with a row-mutating " +
$"ON DELETE action ({string.Join(", ", offenders)}). Replication applies remote rows with " +
"INSERT OR REPLACE, whose implicit delete fires that action with change capture suppressed — " +
"the child rows would be changed on this node only and never replicated, diverging the pair " +
"undetectably. Use ON DELETE NO ACTION/RESTRICT and delete children explicitly, or leave this " +
"table unreplicated.");
var onDelete = reader.IsDBNull(6) ? "" : reader.GetString(6);
if (!RowMutatingDeleteActions.Contains(onDelete))
continue;
throw new LocalDbRegistrationException(
$"Cannot register table '{tableName}': table '{child}' has a foreign key referencing it " +
$"with ON DELETE {onDelete}. Replicated tables must not be cascade-delete foreign-key " +
"parents: applying a remote row uses INSERT OR REPLACE, whose implicit DELETE fires the " +
"referential action while change capture is suppressed — the child rows it touches are " +
"never captured, so the peer never learns of them and the two nodes diverge silently. " +
$"Drop the referential action on '{child}' (ON DELETE NO ACTION, removing children " +
"explicitly instead), or do not replicate this table.");
}
}
}
/// <summary>
/// The <c>ON DELETE</c> actions that mutate referencing rows, as SQLite spells them in
/// <c>PRAGMA foreign_key_list</c>. NO ACTION / RESTRICT are absent on purpose — see
/// <see cref="EnsureNoRowMutatingDeleteAction"/>.
/// </summary>
private static readonly HashSet<string> RowMutatingDeleteActions =
new(["CASCADE", "SET NULL", "SET DEFAULT"], StringComparer.OrdinalIgnoreCase);
/// <summary>
/// Seeds <c>__localdb_row_version</c> at the baseline floor for every row of
/// <paramref name="table"/> that has no ledger entry, and flags a snapshot resync if it seeded
@@ -428,9 +428,10 @@ public sealed class LwwApplierTests : IDisposable
Assert.Equal(202, result.Applied);
// Six table-independent statements (applying guard, version lookup, row-version write,
// peer state, and the per-entry savepoint's enter/release), plus an upsert per table
// touched and a delete for the one table tombstoned. The savepoint's rollback statement is
// built on first use like the rest, and nothing in this batch is poison.
// peer state, plus the per-entry savepoint's SAVEPOINT and RELEASE), an upsert per table
// touched, and a delete for the one table tombstoned. The savepoint's ROLLBACK TO and the
// dead-letter insert are absent because this batch has no poison entry — statements are
// built on first use, so the count tracks the statement KINDS the batch actually ran.
Assert.Equal(9, applier.LastBatchStatementCount);
// ...and the batch is still correct, statement reuse notwithstanding.
@@ -464,118 +465,70 @@ public sealed class LwwApplierTests : IDisposable
Assert.Equal(("LAST", (long?)5), await ReadOrder(db, 5));
}
/// <summary>
/// Arms a BEFORE INSERT trigger on the row-version ledger that aborts the write for one key —
/// a genuine statement-level SQLite fault (SQLITE_CONSTRAINT), landing exactly between an
/// entry's table write and its ledger write, with no production test seam.
/// </summary>
private static Task PoisonRowVersionWriteFor(SqliteLocalDb db, string pkJson) =>
db.ExecuteAsync(
"CREATE TRIGGER poison_rv BEFORE INSERT ON __localdb_row_version " +
$"WHEN NEW.pk_json = '{pkJson}' BEGIN SELECT RAISE(ABORT, 'ledger write failed'); END");
[Fact]
public async Task Apply_TombstoneWhoseLedgerWriteFails_LeavesRowAndLedgerUntouched()
public async Task Apply_LedgerWriteFails_LeavesNeitherTheDeleteNorTheLedgerEntry()
{
// The hazard: DELETE and the ledger write are two statements, and SQLite aborts only the
// failing one. Without a per-entry savepoint the row is gone while __localdb_row_version
// still calls it live — SnapshotStreamer can then only warn and skip it, so the peer never
// learns the row died and the pair diverges. Both or neither.
// The hazard (P4-7): applying a tombstone is TWO statements — delete the row, then write the
// ledger's tombstone version — and the poison catch commits whatever the first one already
// did. A failure landing between them therefore committed a DELETED ROW carrying a STALE LIVE
// ledger entry, which is exactly the shape SnapshotStreamer warns about and skips: the key
// then never travels again and the pair is divergent for that row, permanently and silently.
//
// The failure is injected with a trigger that aborts the ledger write (in production the
// realistic causes are disk-full / IO error / corruption — none of them reproducible here),
// and RAISE(ABORT) aborts only the statement, so the applier's narrow SqliteException catch
// treats it as per-entry poison exactly as it would a real one.
using var db = await NewOrdersDb();
await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'LOCAL', 5)");
var localHlc = await RowVersionHlc(db, Pk1);
await PoisonRowVersionWriteFor(db, Pk1);
await db.ExecuteAsync(
"CREATE TRIGGER poison_ledger BEFORE INSERT ON __localdb_row_version " +
"WHEN NEW.is_tombstone = 1 BEGIN SELECT RAISE(ABORT, 'injected ledger failure'); END");
var result = await new LwwApplier(db).ApplyBatchAsync(
[Entry(90, Pk1, null, localHlc + 1000, "remote-node", tombstone: true)], default);
Assert.Equal(0, result.Applied);
// The entry is poison, so it is dead-lettered and the batch continues — unchanged semantics.
Assert.Equal(1, result.DeadLettered);
Assert.Equal(0, result.Applied);
// BOTH or NEITHER. Neither, here: the row must still be present, because its ledger entry
// still says "live at the local version".
Assert.Equal(("LOCAL", (long?)5), await ReadOrder(db, 1));
var rv = await db.QueryAsync(
"SELECT hlc, is_tombstone FROM __localdb_row_version WHERE table_name = 'orders' AND pk_json = @pk",
x => (Hlc: x.GetInt64(0), Tomb: x.GetInt64(1)), new { pk = Pk1 });
Assert.Single(rv);
Assert.Equal(localHlc, rv[0].Hlc);
Assert.Equal(0, rv[0].Tomb);
}
[Fact]
public async Task Apply_UpsertWhoseLedgerWriteFails_LeavesRowAndLedgerUntouched()
public async Task Apply_LedgerWriteFails_MidBatch_LeavesTheOtherEntriesApplied()
{
// Same split, other direction: the new row would sit in the table carrying the OLD version,
// so it would lose the very next LWW comparison it should win.
// The atomicity must be per-ENTRY, not per-batch: rolling back to the savepoint may not
// discard the rows applied before the poison one, nor block the ones after it.
using var db = await NewOrdersDb();
await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'LOCAL', 5)");
var localHlc = await RowVersionHlc(db, Pk1);
await PoisonRowVersionWriteFor(db, Pk1);
await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (2, 'DOOMED', 2)");
var doomedHlc = await RowVersionHlc(db, "{\"id\":2}");
var result = await new LwwApplier(db).ApplyBatchAsync(
[Entry(91, Pk1, "{\"id\":1,\"sku\":\"REMOTE\",\"qty\":99}", localHlc + 1000, "remote-node")], default);
Assert.Equal(0, result.Applied);
Assert.Equal(1, result.DeadLettered);
Assert.Equal(("LOCAL", (long?)5), await ReadOrder(db, 1));
Assert.Equal(localHlc, await RowVersionHlc(db, Pk1));
}
[Fact]
public async Task Apply_LedgerWriteFailureMidBatch_DoesNotUndoTheRestOfTheBatch()
{
// The savepoint must rewind ONE entry, not the batch: entries either side of the poison
// still apply, and the batch still commits.
using var db = await NewOrdersDb();
await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'LOCAL', 5)");
var localHlc = await RowVersionHlc(db, Pk1);
await PoisonRowVersionWriteFor(db, Pk1);
await db.ExecuteAsync(
"CREATE TRIGGER poison_ledger BEFORE INSERT ON __localdb_row_version " +
"WHEN NEW.is_tombstone = 1 BEGIN SELECT RAISE(ABORT, 'injected ledger failure'); END");
var result = await new LwwApplier(db).ApplyBatchAsync(
[
Entry(92, "{\"id\":2}", "{\"id\":2,\"sku\":\"BEFORE\",\"qty\":2}", localHlc + 1000, "remote-node"),
Entry(93, Pk1, null, localHlc + 1001, "remote-node", tombstone: true),
Entry(94, "{\"id\":3}", "{\"id\":3,\"sku\":\"AFTER\",\"qty\":3}", localHlc + 1002, "remote-node"),
Entry(91, Pk1, "{\"id\":1,\"sku\":\"BEFORE\",\"qty\":1}", doomedHlc + 1, "remote-node"),
Entry(92, "{\"id\":2}", null, doomedHlc + 1000, "remote-node", tombstone: true),
Entry(93, "{\"id\":3}", "{\"id\":3,\"sku\":\"AFTER\",\"qty\":3}", doomedHlc + 3, "remote-node"),
], default);
Assert.Equal(2, result.Applied);
Assert.Equal(1, result.DeadLettered);
Assert.Equal(94, result.AppliedThruSeq);
Assert.Equal(("BEFORE", (long?)2), await ReadOrder(db, 2));
Assert.Equal(("BEFORE", (long?)1), await ReadOrder(db, 1));
Assert.Equal(("DOOMED", (long?)2), await ReadOrder(db, 2));
Assert.Equal(("AFTER", (long?)3), await ReadOrder(db, 3));
Assert.Equal(("LOCAL", (long?)5), await ReadOrder(db, 1));
}
[Fact]
public async Task Apply_DeadLetterFailure_ReportsBothCausesAndCommitsNothing()
{
// Two faults stacked: the entry is poison (its ledger write aborts) AND recovering from it
// fails (the dead-letter INSERT aborts too). Reporting only the second would discard the
// one that says WHY the entry was poison, leaving an operator with "dead-letter failed"
// and no cause.
using var db = await NewOrdersDb();
await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'LOCAL', 5)");
var localHlc = await RowVersionHlc(db, Pk1);
await PoisonRowVersionWriteFor(db, Pk1);
await db.ExecuteAsync(
"CREATE TRIGGER poison_dl BEFORE INSERT ON __localdb_dead_letter " +
"BEGIN SELECT RAISE(ABORT, 'dead-letter write failed'); END");
var ex = await Assert.ThrowsAsync<AggregateException>(() => new LwwApplier(db).ApplyBatchAsync(
[
Entry(95, "{\"id\":2}", "{\"id\":2,\"sku\":\"BEFORE\",\"qty\":2}", localHlc + 1000, "remote-node"),
Entry(96, Pk1, null, localHlc + 1001, "remote-node", tombstone: true),
], default));
// The entry's own fault first (InnerException), the recovery fault carried alongside it.
Assert.Contains("ledger write failed", ex.InnerException!.Message);
Assert.Contains("dead-letter write failed", ex.InnerExceptions[1].Message);
Assert.Contains("orders", ex.Message);
Assert.Contains("96", ex.Message);
// Fail-closed: the batch never commits, so even the entry that applied cleanly is gone.
Assert.Null(await ReadOrder(db, 2));
Assert.Equal(("LOCAL", (long?)5), await ReadOrder(db, 1));
var watermark = await db.QueryAsync(
"SELECT last_applied_remote_seq FROM __localdb_peer_state WHERE id = 1", x => x.GetInt64(0));
Assert.Equal(0, watermark[0]);
}
[Fact]
@@ -70,71 +70,75 @@ public sealed class RegistrationTests : IDisposable
Assert.Contains("BLOB", ex.Message);
}
// ---- Cascading-delete foreign-key parents (P4-7) --------------------------------------
[Fact]
public async Task Register_CascadeDeleteParent_Throws()
public async Task Register_CascadeDeleteFkParent_Throws()
{
// The apply path writes remote rows with INSERT OR REPLACE, whose implicit delete fires
// ON DELETE CASCADE with capture suppressed — the child rows would die on this node only.
// The hazard: applying a remote row uses INSERT OR REPLACE, whose implicit DELETE fires
// ON DELETE CASCADE while the capture guard is RAISED — so the cascaded child rows are
// deleted locally and never captured, and the peer keeps them forever.
using var db = Create();
await CreateOrders(db);
await db.ExecuteAsync(
"CREATE TABLE order_lines (id INTEGER PRIMARY KEY, " +
"order_id INTEGER REFERENCES orders(id) ON DELETE CASCADE, sku TEXT)");
"CREATE TABLE order_lines (id INTEGER PRIMARY KEY, order_id INTEGER " +
"REFERENCES orders(id) ON DELETE CASCADE)");
var ex = Assert.Throws<LocalDbRegistrationException>(() => db.RegisterReplicated("orders"));
Assert.Contains("orders", ex.Message);
Assert.Contains("order_lines.order_id", ex.Message);
Assert.Contains("order_lines", ex.Message);
Assert.Contains("ON DELETE CASCADE", ex.Message);
// Refused, not half-registered.
// Refused means refused: no triggers installed, nothing in the registry.
Assert.Empty(await Triggers(db));
Assert.False(db.ReplicatedTables.ContainsKey("orders"));
}
[Fact]
public async Task Register_CascadeDeleteParent_IsWhatWouldActuallyDiverge()
public async Task Register_CascadeDeleteFkChild_IsAllowed()
{
// Proves the guard is guarding something real: with the parent NOT registered, applying a
// replace to it silently wipes child rows that produce no oplog entry.
// Only the PARENT is refused. The child side is safe: replacing a child row cascades
// nothing, and the child's own deletes are captured normally.
using var db = Create();
await CreateOrders(db);
await db.ExecuteAsync(
"CREATE TABLE order_lines (id INTEGER PRIMARY KEY, " +
"order_id INTEGER REFERENCES orders(id) ON DELETE CASCADE, sku TEXT)");
await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'A', 1)");
await db.ExecuteAsync("INSERT INTO order_lines (id, order_id, sku) VALUES (10, 1, 'L')");
"CREATE TABLE order_lines (id INTEGER PRIMARY KEY, order_id INTEGER " +
"REFERENCES orders(id) ON DELETE CASCADE)");
await db.ExecuteAsync("INSERT OR REPLACE INTO orders (id, sku, qty) VALUES (1, 'B', 2)");
db.RegisterReplicated("order_lines");
var lines = await db.QueryAsync("SELECT COUNT(*) FROM order_lines", r => r.GetInt64(0));
Assert.Equal(0, lines[0]);
Assert.True(db.ReplicatedTables.ContainsKey("order_lines"));
}
[Theory]
[InlineData("SET NULL")]
[InlineData("SET DEFAULT")]
public async Task Register_RowMutatingOnDeleteParent_Throws(string action)
[InlineData("ON DELETE SET NULL", "ON DELETE SET NULL")]
[InlineData("ON DELETE SET DEFAULT", "ON DELETE SET DEFAULT")]
public async Task Register_OtherRowMutatingDeleteActions_AlsoThrow(string clause, string expected)
{
// SET NULL / SET DEFAULT mutate child rows on the same implicit DELETE and are captured
// just as little — same divergence, so the same refusal.
using var db = Create();
await CreateOrders(db);
await db.ExecuteAsync(
"CREATE TABLE order_lines (id INTEGER PRIMARY KEY, " +
$"order_id INTEGER DEFAULT 0 REFERENCES orders(id) ON DELETE {action}, sku TEXT)");
"CREATE TABLE order_lines (id INTEGER PRIMARY KEY, order_id INTEGER DEFAULT 0 " +
$"REFERENCES orders(id) {clause})");
var ex = Assert.Throws<LocalDbRegistrationException>(() => db.RegisterReplicated("orders"));
Assert.Contains($"ON DELETE {action}", ex.Message);
Assert.Contains(expected, ex.Message);
}
[Fact]
public async Task Register_NonMutatingOnDeleteParent_IsAllowed()
public async Task Register_NonCascadingFkParent_IsAllowed()
{
// RESTRICT/NO ACTION abort the statement instead of changing rows behind capture's back,
// which the apply path already handles as per-entry poison.
// The default (NO ACTION / RESTRICT) mutates no child row: the implicit DELETE simply
// fails the statement, which the applier already surfaces as a dead-lettered entry. Noisy,
// not silent — so there is nothing to refuse.
using var db = Create();
await CreateOrders(db);
await db.ExecuteAsync(
"CREATE TABLE order_lines (id INTEGER PRIMARY KEY, " +
"order_id INTEGER REFERENCES orders(id) ON DELETE RESTRICT, sku TEXT)");
"CREATE TABLE order_lines (id INTEGER PRIMARY KEY, order_id INTEGER REFERENCES orders(id))");
await db.ExecuteAsync("CREATE TABLE restricted (id INTEGER PRIMARY KEY, " +
"order_id INTEGER REFERENCES orders(id) ON DELETE RESTRICT)");
db.RegisterReplicated("orders");
@@ -142,35 +146,19 @@ public sealed class RegistrationTests : IDisposable
}
[Fact]
public async Task Register_SelfReferencingCascade_Throws()
public async Task Register_UnrelatedTablesWithCascades_DoNotBlockRegistration()
{
// A table can be its own cascading parent: replacing a row would delete that row's
// children in the SAME table, uncaptured. The guard reads every foreign key in the
// database, so it does not need a special case — but the README promises this, so it is
// pinned.
using var db = Create();
await db.ExecuteAsync(
"CREATE TABLE nodes (id INTEGER PRIMARY KEY, " +
"parent_id INTEGER REFERENCES nodes(id) ON DELETE CASCADE, name TEXT)");
var ex = Assert.Throws<LocalDbRegistrationException>(() => db.RegisterReplicated("nodes"));
Assert.Contains("nodes.parent_id", ex.Message);
Assert.Contains("ON DELETE CASCADE", ex.Message);
}
[Fact]
public async Task Register_CascadeDeleteChild_IsAllowed()
{
// Only the PARENT side is hazardous: replacing a child row deletes no other table's rows.
// The scan must key on the table being registered, not on "the database contains a cascade".
using var db = Create();
await CreateOrders(db);
await db.ExecuteAsync("CREATE TABLE widgets (id INTEGER PRIMARY KEY, name TEXT)");
await db.ExecuteAsync(
"CREATE TABLE order_lines (id INTEGER PRIMARY KEY, " +
"order_id INTEGER REFERENCES orders(id) ON DELETE CASCADE, sku TEXT)");
"CREATE TABLE widget_parts (id INTEGER PRIMARY KEY, widget_id INTEGER " +
"REFERENCES widgets(id) ON DELETE CASCADE)");
var table = db.RegisterReplicated("order_lines");
db.RegisterReplicated("orders");
Assert.Equal("order_lines", table.Name);
Assert.True(db.ReplicatedTables.ContainsKey("orders"));
}
[Fact]
@@ -52,14 +52,14 @@ public sealed class SnapshotResyncTests : IDisposable
// A fresh session stack over an EXISTING database. Reconnecting after a session ends needs a new
// session (RunAsync is single-use), so any scenario with two connection episodes over one node's
// data — a peer being replaced between them — goes through here.
private static Side NewSessionOver(SqliteLocalDb db, ReplicationOptions? options = null, ILogger? streamerLogger = null)
private static Side NewSessionOver(SqliteLocalDb db, ReplicationOptions? options = null)
{
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);
var streamer = new SnapshotStreamer(db, store, options, streamerLogger ?? NullLogger.Instance);
var streamer = new SnapshotStreamer(db, store, options, NullLogger.Instance);
var snapshotApplier = new SnapshotApplier(db, applier, store, NullLogger.Instance);
session.SnapshotSender = streamer.SendAsync;
session.SnapshotApplier = snapshotApplier;
@@ -291,74 +291,6 @@ public sealed class SnapshotResyncTests : IDisposable
await SwallowAsync(runB);
}
[Fact]
public async Task WipedPeer_SnapshotLiftsItsWatermark_SoTheNextReconnectSendsNoSecondSnapshot()
{
// The follow-on to the wiped-peer gap above, found on the ScadaBridge docker rig 2026-08-20
// (its deferred-work register row 106): the snapshot DID heal the wiped node, but it was
// stamped as_of_seq = MAX(seq) of an oplog that ack-pruning had emptied — i.e. 0 — so the
// receiver's (correctly MAX-guarded) watermark stayed at 0. On the next handshake the
// healthy side's empty-oplog horizon rule (streamable from last_acked_seq + 1) saw
// 0 < last_acked_seq as a gap again and re-streamed the whole database — on EVERY reconnect,
// until the healthy side happened to write a new row (127k rows per restart on the rig).
// A full snapshot taken now covers every delta ever pruned below last_acked_seq, so that is
// the floor the as-of must carry.
var a = await NewSideAsync();
for (var i = 1; i <= 3; i++)
await a.Db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (@id, 'A', @id)", new { id = i });
await a.Store.RecordPeerAckAsync(await a.Store.GetMaxSeqAsync());
Assert.Equal(0, await OplogCount(a.Db));
var b = await NewSideAsync();
// Episode 1: the wiped peer is healed by a full snapshot.
var (d1A, d1B) = DuplexPair();
using (var cts1 = new CancellationTokenSource(RunTimeout))
{
var run1A = a.Session.RunAsync(d1A, cts1.Token);
var run1B = b.Session.RunAsync(d1B, cts1.Token);
await WaitForAsync(async () => (await ReadOrders(b.Db)).Count == 3, RunTimeout);
cts1.Cancel();
await SwallowAsync(run1A);
await SwallowAsync(run1B);
}
// The snapshot must have lifted B's watermark to everything A ever produced (seq 3), not 0.
Assert.Equal(3, await LastAppliedRemoteSeq(b.Db));
// Episode 2: a routine reconnect. A writes one more row so the delta path gives us a
// barrier; the snapshot streamer's logger proves whether a snapshot was sent at all.
var streamerLog = new CapturingLogger();
var a2 = NewSessionOver(a.Db, streamerLogger: streamerLog);
var b2 = NewSessionOver(b.Db);
var (d2A, d2B) = DuplexPair();
using var cts2 = new CancellationTokenSource(RunTimeout);
var run2A = a2.Session.RunAsync(d2A, cts2.Token);
var run2B = b2.Session.RunAsync(d2B, cts2.Token);
await WaitForAsync(() => Task.FromResult(b2.Session.PeerNodeId is not null), RunTimeout);
await a.Db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (4, 'DELTA', 4)");
await WaitForAsync(async () => (await ReadOrders(b.Db)).Count == 4, RunTimeout);
Assert.DoesNotContain(streamerLog.Messages, m => m.StartsWith("Snapshot sent", StringComparison.Ordinal));
Assert.Equal(4, await LastAppliedRemoteSeq(b.Db));
cts2.Cancel();
await SwallowAsync(run2A);
await SwallowAsync(run2B);
}
private sealed class CapturingLogger : ILogger
{
private readonly List<string> _messages = [];
public IReadOnlyList<string> Messages { get { lock (_messages) return _messages.ToList(); } }
public IDisposable? BeginScope<TState>(TState state) where TState : notnull => null;
public bool IsEnabled(LogLevel logLevel) => true;
public void Log<TState>(LogLevel logLevel, EventId eventId, TState state, Exception? exception, Func<TState, Exception?, string> formatter)
{
lock (_messages) _messages.Add(formatter(state, exception));
}
}
[Fact]
public async Task RebuiltPeer_OwnWritesReachTheHealthyNode_DespiteTheOldPeersWatermark()
{