Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 41bdc178c2 |
@@ -5,7 +5,7 @@
|
|||||||
<Nullable>enable</Nullable>
|
<Nullable>enable</Nullable>
|
||||||
<ImplicitUsings>enable</ImplicitUsings>
|
<ImplicitUsings>enable</ImplicitUsings>
|
||||||
<LangVersion>latest</LangVersion>
|
<LangVersion>latest</LangVersion>
|
||||||
<Version>0.3.2</Version>
|
<Version>0.3.3</Version>
|
||||||
<ManagePackageVersionsCentrally>true</ManagePackageVersionsCentrally>
|
<ManagePackageVersionsCentrally>true</ManagePackageVersionsCentrally>
|
||||||
<TreatWarningsAsErrors>true</TreatWarningsAsErrors>
|
<TreatWarningsAsErrors>true</TreatWarningsAsErrors>
|
||||||
<!-- NU1510 (package-pruning advice): the Replication and Tests projects carry a
|
<!-- NU1510 (package-pruning advice): the Replication and Tests projects carry a
|
||||||
|
|||||||
@@ -10,7 +10,7 @@ Change capture uses SQLite triggers writing a single HLC-stamped oplog inside th
|
|||||||
consumer's own write transaction (crash-consistent, non-bypassable). Convergence is
|
consumer's own write transaction (crash-consistent, non-bypassable). Convergence is
|
||||||
last-writer-wins over a hybrid logical clock. All timestamps are UTC.
|
last-writer-wins over a hybrid logical clock. All timestamps are UTC.
|
||||||
|
|
||||||
## Packages (all `0.3.2`)
|
## Packages (all `0.3.3`)
|
||||||
|
|
||||||
| Package | Contents |
|
| Package | Contents |
|
||||||
|---|---|
|
|---|---|
|
||||||
@@ -20,6 +20,25 @@ last-writer-wins over a hybrid logical clock. All timestamps are UTC.
|
|||||||
|
|
||||||
> Status: **published to the Gitea feed; adopted by ScadaBridge (Phases 1+2) and OtOpcUa (Phase 1).**
|
> Status: **published to the Gitea feed; adopted by ScadaBridge (Phases 1+2) and OtOpcUa (Phase 1).**
|
||||||
>
|
>
|
||||||
|
> - `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.
|
||||||
|
> - **`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
|
> - `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`
|
> (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.
|
> bump, no on-disk schema change — the keepalive is transport-level and the rest is local policy.
|
||||||
@@ -260,6 +279,18 @@ All options are validated at startup (`ZB.MOM.WW.Configuration` validators).
|
|||||||
`RegisterReplicated`, because `json_object` cannot capture it). A *typeless* column can
|
`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
|
still hold a BLOB value at runtime, which then surfaces as a write-time error, not a
|
||||||
registration error.
|
registration error.
|
||||||
|
- **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
|
- **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
|
||||||
|
|||||||
@@ -14,6 +14,8 @@ internal sealed record ApplyResult(long AppliedThruSeq, int Applied, int Discard
|
|||||||
/// row-version (hlc, then node_id ordinal tie-break), poison rows are dead-lettered while the
|
/// 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, 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).
|
/// 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.
|
/// This never writes <c>__localdb_oplog</c>: the 2-node topology does not re-forward applied deltas.
|
||||||
/// </summary>
|
/// </summary>
|
||||||
/// <remarks>
|
/// <remarks>
|
||||||
@@ -88,6 +90,18 @@ internal sealed class LwwApplier(SqliteLocalDb db, Func<DateTimeOffset>? utcNow
|
|||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// 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
|
try
|
||||||
{
|
{
|
||||||
var table = db.ReplicatedTables[entry.TableName];
|
var table = db.ReplicatedTables[entry.TableName];
|
||||||
@@ -97,7 +111,7 @@ internal sealed class LwwApplier(SqliteLocalDb db, Func<DateTimeOffset>? utcNow
|
|||||||
await sql.UpsertAsync(table, entry.RowJson!, ct);
|
await sql.UpsertAsync(table, entry.RowJson!, ct);
|
||||||
|
|
||||||
await sql.WriteRowVersionAsync(entry, utcNow, ct);
|
await sql.WriteRowVersionAsync(entry, utcNow, ct);
|
||||||
applied++;
|
pairApplied = true;
|
||||||
}
|
}
|
||||||
catch (SqliteException ex)
|
catch (SqliteException ex)
|
||||||
{
|
{
|
||||||
@@ -105,9 +119,21 @@ internal sealed class LwwApplier(SqliteLocalDb db, Func<DateTimeOffset>? utcNow
|
|||||||
// malformed row_json) are treated as per-entry poison — SQLite aborts just the
|
// malformed row_json) are treated as per-entry poison — SQLite aborts just the
|
||||||
// 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 and rolls back the whole batch (fail-closed).
|
// 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);
|
await sql.DeadLetterAsync(entry, ex.Message, utcNow, ct);
|
||||||
deadLettered++;
|
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);
|
await sql.AdvancePeerStateAsync(maxSeq, maxHlc, utcNow, ct);
|
||||||
@@ -143,10 +169,42 @@ internal sealed class LwwApplier(SqliteLocalDb db, Func<DateTimeOffset>? utcNow
|
|||||||
private SqliteCommand? _rowVersion;
|
private SqliteCommand? _rowVersion;
|
||||||
private SqliteCommand? _deadLetter;
|
private SqliteCommand? _deadLetter;
|
||||||
private SqliteCommand? _peerState;
|
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>
|
/// <summary>How many distinct statements this set has prepared.</summary>
|
||||||
public int StatementCount => _owned.Count;
|
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)
|
public Task SetApplyingAsync(int value, CancellationToken ct)
|
||||||
{
|
{
|
||||||
var cmd = _applying ??= Create(
|
var cmd = _applying ??= Create(
|
||||||
|
|||||||
@@ -38,8 +38,18 @@ public interface ILocalDb
|
|||||||
/// schema digest, and installs the three AFTER triggers that capture every change into the
|
/// 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
|
/// oplog + row-version tables within the same transaction as the consumer's write. Idempotent
|
||||||
/// (re-registering regenerates the triggers). Throws <see cref="LocalDbRegistrationException"/>
|
/// (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, it has a BLOB column, or
|
||||||
|
/// another table references it with a row-mutating <c>ON DELETE</c> action.
|
||||||
/// </summary>
|
/// </summary>
|
||||||
|
/// <remarks>
|
||||||
|
/// <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);
|
ReplicatedTable RegisterReplicated(string tableName);
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
|
|||||||
@@ -199,6 +199,8 @@ internal sealed class SqliteLocalDb : ILocalDb, IDisposable
|
|||||||
throw new LocalDbRegistrationException(
|
throw new LocalDbRegistrationException(
|
||||||
$"Cannot register table '{tableName}': replication requires an explicit primary key.");
|
$"Cannot register table '{tableName}': replication requires an explicit primary key.");
|
||||||
|
|
||||||
|
EnsureNoRowMutatingDeleteAction(tableName);
|
||||||
|
|
||||||
var allColumns = columns.Select(c => (c.Name, c.Type)).ToList();
|
var allColumns = columns.Select(c => (c.Name, c.Type)).ToList();
|
||||||
var digest = ReplicatedTable.ComputeDigest(tableName, pkColumns, allColumns);
|
var digest = ReplicatedTable.ComputeDigest(tableName, pkColumns, allColumns);
|
||||||
var table = new ReplicatedTable(tableName, pkColumns, allColumns, digest);
|
var table = new ReplicatedTable(tableName, pkColumns, allColumns, digest);
|
||||||
@@ -293,6 +295,87 @@ internal sealed class SqliteLocalDb : ILocalDb, IDisposable
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// 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 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>
|
||||||
|
/// 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>
|
||||||
|
/// 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 EnsureNoRowMutatingDeleteAction(string tableName)
|
||||||
|
{
|
||||||
|
var candidates = new List<string>();
|
||||||
|
using (var cmd = _master.CreateCommand())
|
||||||
|
{
|
||||||
|
// 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())
|
||||||
|
candidates.Add(reader.GetString(0));
|
||||||
|
}
|
||||||
|
|
||||||
|
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;
|
||||||
|
|
||||||
|
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>
|
/// <summary>
|
||||||
/// Seeds <c>__localdb_row_version</c> at the baseline floor for every row of
|
/// 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
|
/// <paramref name="table"/> that has no ledger entry, and flags a snapshot resync if it seeded
|
||||||
|
|||||||
@@ -427,9 +427,12 @@ public sealed class LwwApplierTests : IDisposable
|
|||||||
var result = await applier.ApplyBatchAsync(entries, default);
|
var result = await applier.ApplyBatchAsync(entries, default);
|
||||||
|
|
||||||
Assert.Equal(202, result.Applied);
|
Assert.Equal(202, result.Applied);
|
||||||
// Four table-independent statements (applying guard, version lookup, row-version write,
|
// Six 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.
|
// peer state, plus the per-entry savepoint's SAVEPOINT and RELEASE), an upsert per table
|
||||||
Assert.Equal(7, applier.LastBatchStatementCount);
|
// 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.
|
// ...and the batch is still correct, statement reuse notwithstanding.
|
||||||
Assert.Null(await ReadOrder(db, 1));
|
Assert.Null(await ReadOrder(db, 1));
|
||||||
@@ -462,6 +465,72 @@ public sealed class LwwApplierTests : IDisposable
|
|||||||
Assert.Equal(("LAST", (long?)5), await ReadOrder(db, 5));
|
Assert.Equal(("LAST", (long?)5), await ReadOrder(db, 5));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task Apply_LedgerWriteFails_LeavesNeitherTheDeleteNorTheLedgerEntry()
|
||||||
|
{
|
||||||
|
// 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 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);
|
||||||
|
|
||||||
|
// 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_LedgerWriteFails_MidBatch_LeavesTheOtherEntriesApplied()
|
||||||
|
{
|
||||||
|
// 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 (2, 'DOOMED', 2)");
|
||||||
|
var doomedHlc = await RowVersionHlc(db, "{\"id\":2}");
|
||||||
|
|
||||||
|
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(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(("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));
|
||||||
|
}
|
||||||
|
|
||||||
[Fact]
|
[Fact]
|
||||||
public async Task Apply_ReplaceExistingRow_DoesNotFireCaptureTriggers()
|
public async Task Apply_ReplaceExistingRow_DoesNotFireCaptureTriggers()
|
||||||
{
|
{
|
||||||
|
|||||||
@@ -70,6 +70,97 @@ public sealed class RegistrationTests : IDisposable
|
|||||||
Assert.Contains("BLOB", ex.Message);
|
Assert.Contains("BLOB", ex.Message);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ---- Cascading-delete foreign-key parents (P4-7) --------------------------------------
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task Register_CascadeDeleteFkParent_Throws()
|
||||||
|
{
|
||||||
|
// 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)");
|
||||||
|
|
||||||
|
var ex = Assert.Throws<LocalDbRegistrationException>(() => db.RegisterReplicated("orders"));
|
||||||
|
Assert.Contains("orders", ex.Message);
|
||||||
|
Assert.Contains("order_lines", ex.Message);
|
||||||
|
Assert.Contains("ON DELETE CASCADE", ex.Message);
|
||||||
|
|
||||||
|
// 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_CascadeDeleteFkChild_IsAllowed()
|
||||||
|
{
|
||||||
|
// 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)");
|
||||||
|
|
||||||
|
db.RegisterReplicated("order_lines");
|
||||||
|
|
||||||
|
Assert.True(db.ReplicatedTables.ContainsKey("order_lines"));
|
||||||
|
}
|
||||||
|
|
||||||
|
[Theory]
|
||||||
|
[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) {clause})");
|
||||||
|
|
||||||
|
var ex = Assert.Throws<LocalDbRegistrationException>(() => db.RegisterReplicated("orders"));
|
||||||
|
Assert.Contains(expected, ex.Message);
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task Register_NonCascadingFkParent_IsAllowed()
|
||||||
|
{
|
||||||
|
// 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))");
|
||||||
|
await db.ExecuteAsync("CREATE TABLE restricted (id INTEGER PRIMARY KEY, " +
|
||||||
|
"order_id INTEGER REFERENCES orders(id) ON DELETE RESTRICT)");
|
||||||
|
|
||||||
|
db.RegisterReplicated("orders");
|
||||||
|
|
||||||
|
Assert.Equal(3, (await Triggers(db)).Count);
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task Register_UnrelatedTablesWithCascades_DoNotBlockRegistration()
|
||||||
|
{
|
||||||
|
// 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 widget_parts (id INTEGER PRIMARY KEY, widget_id INTEGER " +
|
||||||
|
"REFERENCES widgets(id) ON DELETE CASCADE)");
|
||||||
|
|
||||||
|
db.RegisterReplicated("orders");
|
||||||
|
|
||||||
|
Assert.True(db.ReplicatedTables.ContainsKey("orders"));
|
||||||
|
}
|
||||||
|
|
||||||
[Fact]
|
[Fact]
|
||||||
public async Task Register_IsIdempotent_RegeneratesTriggers()
|
public async Task Register_IsIdempotent_RegeneratesTriggers()
|
||||||
{
|
{
|
||||||
|
|||||||
Reference in New Issue
Block a user