Compare commits
3 Commits
low/T5
...
5e02c1c38f
| Author | SHA1 | Date | |
|---|---|---|---|
| 5e02c1c38f | |||
| 9663279594 | |||
| 122940c0e4 |
@@ -5,7 +5,7 @@
|
||||
<Nullable>enable</Nullable>
|
||||
<ImplicitUsings>enable</ImplicitUsings>
|
||||
<LangVersion>latest</LangVersion>
|
||||
<Version>0.3.2</Version>
|
||||
<Version>0.3.3</Version>
|
||||
<ManagePackageVersionsCentrally>true</ManagePackageVersionsCentrally>
|
||||
<TreatWarningsAsErrors>true</TreatWarningsAsErrors>
|
||||
<!-- NU1510 (package-pruning advice): the Replication and Tests projects carry a
|
||||
|
||||
@@ -10,16 +10,40 @@ 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.2`)
|
||||
## Packages (all `0.3.3`)
|
||||
|
||||
| Package | Contents |
|
||||
|---|---|
|
||||
| `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` | 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.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.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
|
||||
> `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.
|
||||
> - `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.
|
||||
@@ -260,6 +284,15 @@ 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`.
|
||||
- **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,7 +12,8 @@ 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, and the remote watermark advances in the SAME transaction as the data so a
|
||||
/// 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
|
||||
/// crash-replay of the batch is a no-op (LWW's strict-greater compare loses on equal versions).
|
||||
/// This never writes <c>__localdb_oplog</c>: the 2-node topology does not re-forward applied deltas.
|
||||
/// </summary>
|
||||
@@ -88,15 +89,28 @@ 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);
|
||||
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++;
|
||||
}
|
||||
catch (SqliteException ex)
|
||||
@@ -104,8 +118,29 @@ internal sealed class LwwApplier(SqliteLocalDb db, Func<DateTimeOffset>? utcNow
|
||||
// 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 and rolls back the whole batch (fail-closed).
|
||||
await sql.DeadLetterAsync(entry, ex.Message, utcNow, ct);
|
||||
// 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);
|
||||
}
|
||||
|
||||
deadLettered++;
|
||||
}
|
||||
}
|
||||
@@ -138,7 +173,13 @@ 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;
|
||||
@@ -155,6 +196,32 @@ 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)
|
||||
{
|
||||
|
||||
@@ -38,8 +38,20 @@ 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, or it has no primary key.
|
||||
/// 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.
|
||||
/// </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.
|
||||
/// </remarks>
|
||||
ReplicatedTable RegisterReplicated(string tableName);
|
||||
|
||||
/// <summary>
|
||||
|
||||
@@ -199,6 +199,8 @@ internal sealed class SqliteLocalDb : ILocalDb, IDisposable
|
||||
throw new LocalDbRegistrationException(
|
||||
$"Cannot register table '{tableName}': replication requires an explicit primary key.");
|
||||
|
||||
EnsureNoInboundActionOnDelete(tableName);
|
||||
|
||||
var allColumns = columns.Select(c => (c.Name, c.Type)).ToList();
|
||||
var digest = ReplicatedTable.ComputeDigest(tableName, pkColumns, allColumns);
|
||||
var table = new ReplicatedTable(tableName, pkColumns, allColumns, digest);
|
||||
@@ -293,6 +295,66 @@ 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>).
|
||||
/// </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.
|
||||
/// </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.
|
||||
/// </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.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
private void EnsureNoInboundActionOnDelete(string tableName)
|
||||
{
|
||||
var offenders = 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);
|
||||
using var reader = cmd.ExecuteReader();
|
||||
while (reader.Read())
|
||||
offenders.Add(
|
||||
$"{reader.GetString(0)}.{reader.GetString(1)} ON DELETE {reader.GetString(2).ToUpperInvariant()}");
|
||||
}
|
||||
|
||||
if (offenders.Count == 0)
|
||||
return;
|
||||
|
||||
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.");
|
||||
}
|
||||
|
||||
/// <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
|
||||
|
||||
@@ -427,9 +427,11 @@ public sealed class LwwApplierTests : IDisposable
|
||||
var result = await applier.ApplyBatchAsync(entries, default);
|
||||
|
||||
Assert.Equal(202, result.Applied);
|
||||
// Four table-independent statements (applying guard, version lookup, row-version write,
|
||||
// peer state), plus an upsert per table touched and a delete for the one table tombstoned.
|
||||
Assert.Equal(7, applier.LastBatchStatementCount);
|
||||
// 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.
|
||||
Assert.Equal(9, applier.LastBatchStatementCount);
|
||||
|
||||
// ...and the batch is still correct, statement reuse notwithstanding.
|
||||
Assert.Null(await ReadOrder(db, 1));
|
||||
@@ -462,6 +464,120 @@ 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()
|
||||
{
|
||||
// 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.
|
||||
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);
|
||||
|
||||
var result = await new LwwApplier(db).ApplyBatchAsync(
|
||||
[Entry(90, Pk1, null, localHlc + 1000, "remote-node", tombstone: true)], default);
|
||||
|
||||
Assert.Equal(0, result.Applied);
|
||||
Assert.Equal(1, result.DeadLettered);
|
||||
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.Equal(localHlc, rv[0].Hlc);
|
||||
Assert.Equal(0, rv[0].Tomb);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Apply_UpsertWhoseLedgerWriteFails_LeavesRowAndLedgerUntouched()
|
||||
{
|
||||
// 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.
|
||||
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);
|
||||
|
||||
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);
|
||||
|
||||
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"),
|
||||
], 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(("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]
|
||||
public async Task Apply_ReplaceExistingRow_DoesNotFireCaptureTriggers()
|
||||
{
|
||||
|
||||
@@ -70,6 +70,109 @@ public sealed class RegistrationTests : IDisposable
|
||||
Assert.Contains("BLOB", ex.Message);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Register_CascadeDeleteParent_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.
|
||||
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)");
|
||||
|
||||
var ex = Assert.Throws<LocalDbRegistrationException>(() => db.RegisterReplicated("orders"));
|
||||
Assert.Contains("orders", ex.Message);
|
||||
Assert.Contains("order_lines.order_id", ex.Message);
|
||||
Assert.Contains("ON DELETE CASCADE", ex.Message);
|
||||
|
||||
// Refused, not half-registered.
|
||||
Assert.Empty(await Triggers(db));
|
||||
Assert.False(db.ReplicatedTables.ContainsKey("orders"));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Register_CascadeDeleteParent_IsWhatWouldActuallyDiverge()
|
||||
{
|
||||
// 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.
|
||||
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')");
|
||||
|
||||
await db.ExecuteAsync("INSERT OR REPLACE INTO orders (id, sku, qty) VALUES (1, 'B', 2)");
|
||||
|
||||
var lines = await db.QueryAsync("SELECT COUNT(*) FROM order_lines", r => r.GetInt64(0));
|
||||
Assert.Equal(0, lines[0]);
|
||||
}
|
||||
|
||||
[Theory]
|
||||
[InlineData("SET NULL")]
|
||||
[InlineData("SET DEFAULT")]
|
||||
public async Task Register_RowMutatingOnDeleteParent_Throws(string action)
|
||||
{
|
||||
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)");
|
||||
|
||||
var ex = Assert.Throws<LocalDbRegistrationException>(() => db.RegisterReplicated("orders"));
|
||||
Assert.Contains($"ON DELETE {action}", ex.Message);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Register_NonMutatingOnDeleteParent_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.
|
||||
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)");
|
||||
|
||||
db.RegisterReplicated("orders");
|
||||
|
||||
Assert.Equal(3, (await Triggers(db)).Count);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Register_SelfReferencingCascade_Throws()
|
||||
{
|
||||
// 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.
|
||||
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)");
|
||||
|
||||
var table = db.RegisterReplicated("order_lines");
|
||||
|
||||
Assert.Equal("order_lines", table.Name);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Register_IsIdempotent_RegeneratesTriggers()
|
||||
{
|
||||
|
||||
Reference in New Issue
Block a user