Joseph Doherty
122940c0e4
Two latent apply-path hazards from the ScadaBridge adversarial architecture review. Neither has a live trigger in any current consumer — ScadaBridge's ten replicated tables carry no foreign keys at all (verified against the running rig's site-localdb.db) and OtOpcUa's LocalDb tables carry none either — but both are the kind that surface as silent divergence rather than as an error, so they are closed before a consumer's schema grows into them. Wire-compatible: no proto change, no lib_schema_version bump, no on-disk schema change. No consumer code change. - RegisterReplicated now refuses a table that is the PARENT of a foreign key with a row-mutating ON DELETE action (CASCADE / SET NULL / SET DEFAULT), naming the offending child column and action. The apply path writes remote rows with INSERT OR REPLACE, whose PK-conflict resolution DELETES the existing row first; connections run PRAGMA foreign_keys=ON, so that implicit delete fires the child's ON DELETE action while the capture guard is raised. The child rows would change on the applying node only, produce no oplog entry, and diverge the pair with nothing able to detect it — LWW compares only rows it was told about. RESTRICT / NO ACTION stay allowed: they abort the statement instead of mutating rows behind capture's back, which the apply path already handles as per-entry poison. The check inspects every foreign key in the database and runs 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, not at the DDL. Documented on ILocalDb.RegisterReplicated and in the README's limitations list. - 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 (SnapshotStreamer can only warn and skip that, so the peer never learns 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 now runs inside a per-entry SAVEPOINT, rolled back (and released, so nothing nests) before dead-lettering, leaving both writes or neither while the rest of the batch proceeds and commits. Every new test was run against the unfixed code first and fails there with exactly the divergence shape described: the tombstone case reads back null instead of the live row, the upsert case reads back the remote row against an unchanged ledger, and the registration cases throw nothing. Suite: 213/213 green, build 0 warnings / 0 errors.
564 lines
24 KiB
C#
564 lines
24 KiB
C#
using Microsoft.Data.Sqlite;
|
|
using ZB.MOM.WW.LocalDb.Internal;
|
|
using ZB.MOM.WW.LocalDb.Replication;
|
|
using ZB.MOM.WW.LocalDb.Replication.Internal;
|
|
|
|
namespace ZB.MOM.WW.LocalDb.Tests;
|
|
|
|
public sealed class LwwApplierTests : IDisposable
|
|
{
|
|
private const string Pk1 = "{\"id\":1}";
|
|
|
|
private readonly string _path = Path.Combine(Path.GetTempPath(), Guid.NewGuid() + ".db");
|
|
private readonly string _path2 = Path.Combine(Path.GetTempPath(), Guid.NewGuid() + ".db");
|
|
|
|
public void Dispose()
|
|
{
|
|
SqliteConnection.ClearAllPools();
|
|
if (File.Exists(_path))
|
|
File.Delete(_path);
|
|
if (File.Exists(_path2))
|
|
File.Delete(_path2);
|
|
}
|
|
|
|
private async Task<SqliteLocalDb> NewOrdersDb()
|
|
{
|
|
var db = new SqliteLocalDb(new LocalDbOptions { Path = _path });
|
|
await db.ExecuteAsync("CREATE TABLE orders (id INTEGER PRIMARY KEY, sku TEXT, qty INTEGER)");
|
|
db.RegisterReplicated("orders");
|
|
return db;
|
|
}
|
|
|
|
private static async Task<long> RowVersionHlc(SqliteLocalDb db, string pkJson)
|
|
{
|
|
var r = await db.QueryAsync(
|
|
"SELECT hlc FROM __localdb_row_version WHERE table_name = 'orders' AND pk_json = @pk",
|
|
x => x.GetInt64(0), new { pk = pkJson });
|
|
return r[0];
|
|
}
|
|
|
|
private static async Task<(string? Sku, long? Qty)?> ReadOrder(SqliteLocalDb db, long id)
|
|
{
|
|
var r = await db.QueryAsync(
|
|
"SELECT sku, qty FROM orders WHERE id = @id",
|
|
x => (x.IsDBNull(0) ? null : x.GetString(0), (long?)(x.IsDBNull(1) ? null : x.GetInt64(1))),
|
|
new { id });
|
|
return r.Count == 0 ? null : r[0];
|
|
}
|
|
|
|
private static async Task<long> OplogCount(SqliteLocalDb db)
|
|
{
|
|
var r = await db.QueryAsync("SELECT COUNT(*) FROM __localdb_oplog", x => x.GetInt64(0));
|
|
return r[0];
|
|
}
|
|
|
|
private static OplogEntryRecord Entry(
|
|
long seq, string pkJson, string? rowJson, long hlc, string nodeId, bool tombstone = false) =>
|
|
new(seq, "orders", pkJson, rowJson, hlc, nodeId, tombstone);
|
|
|
|
[Fact]
|
|
public async Task Apply_NewerRemote_Wins()
|
|
{
|
|
using var db = await NewOrdersDb();
|
|
await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'LOCAL', 5)");
|
|
var localHlc = await RowVersionHlc(db, Pk1);
|
|
|
|
var result = await new LwwApplier(db).ApplyBatchAsync(
|
|
[Entry(1, Pk1, "{\"id\":1,\"sku\":\"REMOTE\",\"qty\":99}", localHlc + 1000, "remote-node")], default);
|
|
|
|
Assert.Equal(1, result.Applied);
|
|
var row = await ReadOrder(db, 1);
|
|
Assert.Equal(("REMOTE", (long?)99), row);
|
|
|
|
var rv = await db.QueryAsync(
|
|
"SELECT hlc, node_id FROM __localdb_row_version WHERE pk_json = @pk",
|
|
x => (Hlc: x.GetInt64(0), Node: x.GetString(1)), new { pk = Pk1 });
|
|
Assert.Equal(localHlc + 1000, rv[0].Hlc);
|
|
Assert.Equal("remote-node", rv[0].Node);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Apply_OlderRemote_Discarded_StillCounted()
|
|
{
|
|
using var db = await NewOrdersDb();
|
|
await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'LOCAL', 5)");
|
|
var localHlc = await RowVersionHlc(db, Pk1);
|
|
|
|
var result = await new LwwApplier(db).ApplyBatchAsync(
|
|
[Entry(7, Pk1, "{\"id\":1,\"sku\":\"OLD\",\"qty\":1}", localHlc - 1000, "remote-node")], default);
|
|
|
|
Assert.Equal(0, result.Applied);
|
|
Assert.Equal(1, result.Discarded);
|
|
Assert.Equal(7, result.AppliedThruSeq);
|
|
Assert.Equal(("LOCAL", (long?)5), await ReadOrder(db, 1));
|
|
|
|
var peer = await db.QueryAsync(
|
|
"SELECT last_applied_remote_seq FROM __localdb_peer_state WHERE id = 1", x => x.GetInt64(0));
|
|
Assert.Equal(7, peer[0]);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Apply_EqualHlc_NodeIdTieBreak_Deterministic()
|
|
{
|
|
// Greater remote node_id wins the tie.
|
|
using (var db = await NewOrdersDb())
|
|
{
|
|
await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'LOCAL', 5)");
|
|
var localHlc = await RowVersionHlc(db, Pk1);
|
|
var greater = db.NodeId + "z";
|
|
Assert.True(string.CompareOrdinal(greater, db.NodeId) > 0);
|
|
|
|
var result = await new LwwApplier(db).ApplyBatchAsync(
|
|
[Entry(1, Pk1, "{\"id\":1,\"sku\":\"WINS\",\"qty\":9}", localHlc, greater)], default);
|
|
|
|
Assert.Equal(1, result.Applied);
|
|
Assert.Equal(("WINS", (long?)9), await ReadOrder(db, 1));
|
|
}
|
|
|
|
SqliteConnection.ClearAllPools();
|
|
File.Delete(_path);
|
|
|
|
// Smaller remote node_id loses the tie.
|
|
using (var db = await NewOrdersDb())
|
|
{
|
|
await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'LOCAL', 5)");
|
|
var localHlc = await RowVersionHlc(db, Pk1);
|
|
var smaller = "!" + db.NodeId;
|
|
Assert.True(string.CompareOrdinal(smaller, db.NodeId) < 0);
|
|
|
|
var result = await new LwwApplier(db).ApplyBatchAsync(
|
|
[Entry(1, Pk1, "{\"id\":1,\"sku\":\"LOSES\",\"qty\":9}", localHlc, smaller)], default);
|
|
|
|
Assert.Equal(0, result.Applied);
|
|
Assert.Equal(1, result.Discarded);
|
|
Assert.Equal(("LOCAL", (long?)5), await ReadOrder(db, 1));
|
|
}
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Apply_NewRow_Inserts()
|
|
{
|
|
using var db = await NewOrdersDb();
|
|
|
|
var result = await new LwwApplier(db).ApplyBatchAsync(
|
|
[Entry(3, Pk1, "{\"id\":1,\"sku\":\"NEW\",\"qty\":42}", 5_000_000, "remote-node")], default);
|
|
|
|
Assert.Equal(1, result.Applied);
|
|
Assert.Equal(("NEW", (long?)42), await ReadOrder(db, 1));
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Apply_Tombstone_DeletesRow()
|
|
{
|
|
using var db = await NewOrdersDb();
|
|
await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'LOCAL', 5)");
|
|
var localHlc = await RowVersionHlc(db, Pk1);
|
|
|
|
var result = await new LwwApplier(db).ApplyBatchAsync(
|
|
[Entry(4, Pk1, null, localHlc + 1000, "remote-node", tombstone: true)], default);
|
|
|
|
Assert.Equal(1, result.Applied);
|
|
Assert.Null(await ReadOrder(db, 1));
|
|
|
|
var rv = await db.QueryAsync(
|
|
"SELECT is_tombstone, hlc FROM __localdb_row_version WHERE pk_json = @pk",
|
|
x => (Tomb: x.GetInt64(0), Hlc: x.GetInt64(1)), new { pk = Pk1 });
|
|
Assert.Equal(1, rv[0].Tomb);
|
|
Assert.Equal(localHlc + 1000, rv[0].Hlc);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Apply_DoesNotFireCaptureTriggers()
|
|
{
|
|
using var db = await NewOrdersDb();
|
|
await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'LOCAL', 5)");
|
|
var before = await OplogCount(db);
|
|
|
|
await new LwwApplier(db).ApplyBatchAsync(
|
|
[
|
|
Entry(1, Pk1, "{\"id\":1,\"sku\":\"R\",\"qty\":9}", 9_000_000, "remote-node"),
|
|
Entry(2, "{\"id\":2}", "{\"id\":2,\"sku\":\"R2\",\"qty\":7}", 9_000_001, "remote-node"),
|
|
], default);
|
|
|
|
Assert.Equal(before, await OplogCount(db));
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Apply_Batch_Atomic_WatermarkSameTxn()
|
|
{
|
|
using var db = await NewOrdersDb();
|
|
|
|
var result = await new LwwApplier(db).ApplyBatchAsync(
|
|
[
|
|
Entry(10, Pk1, "{\"id\":1,\"sku\":\"A\",\"qty\":1}", 1_000_000, "remote-node"),
|
|
Entry(11, "{\"id\":2}", "{\"id\":2,\"sku\":\"B\",\"qty\":2}", 1_000_001, "remote-node"),
|
|
], default);
|
|
|
|
Assert.Equal(2, result.Applied);
|
|
Assert.Equal(11, result.AppliedThruSeq);
|
|
|
|
// Data and the watermark are both visible => they committed in one transaction.
|
|
Assert.Equal(("A", (long?)1), await ReadOrder(db, 1));
|
|
Assert.Equal(("B", (long?)2), await ReadOrder(db, 2));
|
|
var peer = await db.QueryAsync(
|
|
"SELECT last_applied_remote_seq, last_seen_hlc FROM __localdb_peer_state WHERE id = 1",
|
|
x => (Seq: x.GetInt64(0), Hlc: x.GetInt64(1)));
|
|
Assert.Equal(11, peer[0].Seq);
|
|
Assert.Equal(1_000_001, peer[0].Hlc);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Apply_UnregisteredTable_Throws_NothingApplied()
|
|
{
|
|
using var db = await NewOrdersDb();
|
|
|
|
var batch = new OplogEntryRecord[]
|
|
{
|
|
Entry(20, Pk1, "{\"id\":1,\"sku\":\"A\",\"qty\":1}", 1_000_000, "remote-node"),
|
|
new(21, "widgets", "{\"id\":9}", "{\"id\":9}", 1_000_001, "remote-node", false),
|
|
};
|
|
|
|
await Assert.ThrowsAsync<LocalDbSchemaMismatchException>(
|
|
() => new LwwApplier(db).ApplyBatchAsync(batch, default));
|
|
|
|
// The valid entry that preceded the violation must not have been applied.
|
|
Assert.Null(await ReadOrder(db, 1));
|
|
var peer = await db.QueryAsync(
|
|
"SELECT last_applied_remote_seq FROM __localdb_peer_state WHERE id = 1", x => x.GetInt64(0));
|
|
Assert.Equal(0, peer[0]);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Apply_PoisonEntry_DeadLettered_BatchContinues()
|
|
{
|
|
using var db = await NewOrdersDb();
|
|
|
|
var result = await new LwwApplier(db).ApplyBatchAsync(
|
|
[
|
|
Entry(30, Pk1, "NOT VALID JSON", 1_000_000, "remote-node"),
|
|
Entry(31, "{\"id\":2}", "{\"id\":2,\"sku\":\"OK\",\"qty\":8}", 1_000_001, "remote-node"),
|
|
], default);
|
|
|
|
Assert.Equal(1, result.DeadLettered);
|
|
Assert.Equal(1, result.Applied);
|
|
Assert.Equal(("OK", (long?)8), await ReadOrder(db, 2));
|
|
Assert.Null(await ReadOrder(db, 1));
|
|
|
|
var dead = await db.QueryAsync(
|
|
"SELECT pk_json, error FROM __localdb_dead_letter",
|
|
x => (Pk: x.GetString(0), Err: x.GetString(1)));
|
|
Assert.Single(dead);
|
|
Assert.Equal(Pk1, dead[0].Pk);
|
|
Assert.NotEmpty(dead[0].Err);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Apply_IsIdempotent()
|
|
{
|
|
using var db = await NewOrdersDb();
|
|
var batch = new OplogEntryRecord[]
|
|
{
|
|
Entry(40, Pk1, "{\"id\":1,\"sku\":\"X\",\"qty\":3}", 7_000_000, "remote-node"),
|
|
new(41, "orders", "{\"id\":2}", "{\"id\":2,\"sku\":\"Y\",\"qty\":4}", 7_000_001, "remote-node", false),
|
|
};
|
|
var applier = new LwwApplier(db);
|
|
|
|
var first = await applier.ApplyBatchAsync(batch, default);
|
|
Assert.Equal(2, first.Applied);
|
|
|
|
var second = await applier.ApplyBatchAsync(batch, default);
|
|
Assert.Equal(0, second.Applied);
|
|
Assert.Equal(2, second.Discarded);
|
|
|
|
Assert.Equal(("X", (long?)3), await ReadOrder(db, 1));
|
|
Assert.Equal(("Y", (long?)4), await ReadOrder(db, 2));
|
|
var rows = await db.QueryAsync("SELECT COUNT(*) FROM orders", x => x.GetInt64(0));
|
|
Assert.Equal(2, rows[0]);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Apply_ObservesRemoteHlc()
|
|
{
|
|
using var db = await NewOrdersDb();
|
|
var maxHlc = (DateTimeOffset.UtcNow.AddDays(365).ToUnixTimeMilliseconds() << 16);
|
|
|
|
await new LwwApplier(db).ApplyBatchAsync(
|
|
[Entry(1, Pk1, "{\"id\":1,\"sku\":\"R\",\"qty\":1}", maxHlc, "remote-node")], default);
|
|
|
|
Assert.True(db.Clock.Next() > maxHlc);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Apply_AdvancesPeerState()
|
|
{
|
|
using var db = await NewOrdersDb();
|
|
|
|
await new LwwApplier(db).ApplyBatchAsync(
|
|
[
|
|
Entry(50, Pk1, "{\"id\":1,\"sku\":\"A\",\"qty\":1}", 3_000_000, "remote-node"),
|
|
Entry(52, "{\"id\":2}", "{\"id\":2,\"sku\":\"B\",\"qty\":2}", 3_000_050, "remote-node"),
|
|
], default);
|
|
|
|
var peer = await db.QueryAsync(
|
|
"SELECT last_applied_remote_seq, last_seen_hlc, last_sync_utc FROM __localdb_peer_state WHERE id = 1",
|
|
x => (Seq: x.GetInt64(0), Hlc: x.GetInt64(1), Utc: x.IsDBNull(2) ? null : x.GetString(2)));
|
|
|
|
Assert.Equal(52, peer[0].Seq);
|
|
Assert.Equal(3_000_050, peer[0].Hlc);
|
|
Assert.NotNull(peer[0].Utc);
|
|
Assert.EndsWith("Z", peer[0].Utc);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Apply_NullRowJsonNonTombstone_DeadLettered_NoPhantomRow()
|
|
{
|
|
using var db = await NewOrdersDb();
|
|
await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'LOCAL', 5)");
|
|
|
|
var result = await new LwwApplier(db).ApplyBatchAsync(
|
|
[Entry(60, "{\"id\":2}", null, 9_000_000_000_000_000, "remote-node", tombstone: false)], default);
|
|
|
|
Assert.Equal(1, result.DeadLettered);
|
|
Assert.Equal(0, result.Applied);
|
|
|
|
var dead = await db.QueryAsync(
|
|
"SELECT pk_json, error FROM __localdb_dead_letter",
|
|
x => (Pk: x.GetString(0), Err: x.GetString(1)));
|
|
Assert.Single(dead);
|
|
Assert.Equal("{\"id\":2}", dead[0].Pk);
|
|
Assert.Contains("null row_json", dead[0].Err);
|
|
|
|
// No phantom auto-rowid row: the table still holds exactly the one pre-existing row.
|
|
var rows = await db.QueryAsync("SELECT COUNT(*) FROM orders", x => x.GetInt64(0));
|
|
Assert.Equal(1, rows[0]);
|
|
Assert.Null(await ReadOrder(db, 2));
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Apply_RoundTrip_PreservesRealAndBigIntegerExactly()
|
|
{
|
|
const double real17 = 1.2345678901234567; // 17 significant digits
|
|
const long bigInt = 9_007_199_254_740_993; // 2^53 + 1: not representable as a double
|
|
|
|
using var source = new SqliteLocalDb(new LocalDbOptions { Path = _path });
|
|
await source.ExecuteAsync("CREATE TABLE meas (id INTEGER PRIMARY KEY, r REAL, big INTEGER)");
|
|
source.RegisterReplicated("meas");
|
|
await source.ExecuteAsync(
|
|
"INSERT INTO meas (id, r, big) VALUES (1, @r, @big)", new { r = real17, big = bigInt });
|
|
|
|
var captured = await source.QueryAsync(
|
|
"SELECT seq, table_name, pk_json, row_json, hlc, node_id, is_tombstone FROM __localdb_oplog",
|
|
r => new OplogEntryRecord(
|
|
r.GetInt64(0), r.GetString(1), r.GetString(2), r.IsDBNull(3) ? null : r.GetString(3),
|
|
r.GetInt64(4), r.GetString(5), r.GetInt64(6) != 0));
|
|
Assert.Single(captured);
|
|
|
|
using var target = new SqliteLocalDb(new LocalDbOptions { Path = _path2 });
|
|
await target.ExecuteAsync("CREATE TABLE meas (id INTEGER PRIMARY KEY, r REAL, big INTEGER)");
|
|
target.RegisterReplicated("meas");
|
|
|
|
var result = await new LwwApplier(target).ApplyBatchAsync(captured, default);
|
|
Assert.Equal(1, result.Applied);
|
|
|
|
var applied = await target.QueryAsync(
|
|
"SELECT r, big FROM meas WHERE id = 1", x => (R: x.GetDouble(0), Big: x.GetInt64(1)));
|
|
Assert.Equal(BitConverter.DoubleToInt64Bits(real17), BitConverter.DoubleToInt64Bits(applied[0].R));
|
|
Assert.Equal(bigInt, applied[0].Big);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Apply_GuardAlwaysLoweredAfterBatch()
|
|
{
|
|
// The abort/rollback path needs no separate coverage: applying = 1 is set INSIDE the
|
|
// transaction, so any rollback restores it to 0 by SQLite transaction semantics.
|
|
using var db = await NewOrdersDb();
|
|
|
|
await new LwwApplier(db).ApplyBatchAsync(
|
|
[Entry(1, Pk1, "{\"id\":1,\"sku\":\"R\",\"qty\":1}", 5_000_000, "remote-node")], default);
|
|
|
|
var guard = await db.QueryAsync(
|
|
"SELECT applying FROM __localdb_applying WHERE id = 1", x => x.GetInt64(0));
|
|
Assert.Equal(0, guard[0]);
|
|
|
|
// A subsequent consumer write must capture normally — proof the guard is really down.
|
|
var before = await OplogCount(db);
|
|
await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (2, 'LOCAL', 2)");
|
|
Assert.Equal(before + 1, await OplogCount(db));
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Apply_SameTable_TombstoneAndUpsert_OneBatch()
|
|
{
|
|
using var db = await NewOrdersDb();
|
|
await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'LOCAL', 5)");
|
|
var localHlc = await RowVersionHlc(db, Pk1);
|
|
|
|
var result = await new LwwApplier(db).ApplyBatchAsync(
|
|
[
|
|
Entry(70, Pk1, null, localHlc + 1000, "remote-node", tombstone: true),
|
|
Entry(71, "{\"id\":2}", "{\"id\":2,\"sku\":\"NEW\",\"qty\":7}", localHlc + 1001, "remote-node"),
|
|
], default);
|
|
|
|
Assert.Equal(2, result.Applied);
|
|
Assert.Null(await ReadOrder(db, 1));
|
|
Assert.Equal(("NEW", (long?)7), await ReadOrder(db, 2));
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Apply_PreparesEachStatementOnce_NotOncePerRow()
|
|
{
|
|
// The regression this guards: the old shape went through the generic transaction façade,
|
|
// which builds a fresh SqliteCommand — and therefore re-parses the SQL — per call. Three
|
|
// statements per applied row means a 200-row batch parsed 600 statements to run 6 distinct
|
|
// ones. Statement count must track distinct SQL, not rows.
|
|
using var db = await NewOrdersDb();
|
|
await db.ExecuteAsync("CREATE TABLE widgets (id INTEGER PRIMARY KEY, name TEXT)");
|
|
db.RegisterReplicated("widgets");
|
|
|
|
var entries = new List<OplogEntryRecord>();
|
|
for (var i = 1; i <= 200; i++)
|
|
entries.Add(Entry(i, $"{{\"id\":{i}}}", $"{{\"id\":{i},\"sku\":\"S\",\"qty\":{i}}}", 1_000_000 + i, "remote"));
|
|
// A second table and a tombstone, so every statement kind in the set is exercised.
|
|
entries.Add(new OplogEntryRecord(
|
|
201, "widgets", "{\"id\":1}", "{\"id\":1,\"name\":\"W\"}", 1_000_300, "remote", false));
|
|
entries.Add(Entry(202, "{\"id\":1}", null, 1_000_400, "remote", tombstone: true));
|
|
|
|
var applier = new LwwApplier(db);
|
|
var result = await applier.ApplyBatchAsync(entries, default);
|
|
|
|
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.
|
|
Assert.Equal(9, applier.LastBatchStatementCount);
|
|
|
|
// ...and the batch is still correct, statement reuse notwithstanding.
|
|
Assert.Null(await ReadOrder(db, 1));
|
|
Assert.Equal(("S", (long?)200), await ReadOrder(db, 200));
|
|
var widget = await db.QueryAsync("SELECT name FROM widgets WHERE id = 1", x => x.GetString(0));
|
|
Assert.Equal("W", widget[0]);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Apply_ReusedStatements_SurviveAPoisonRowMidBatch()
|
|
{
|
|
// A statement that faults is reset and reused for the following rows; if it were not, one
|
|
// poison row would silently corrupt every row after it in the batch.
|
|
using var db = await NewOrdersDb();
|
|
|
|
var applier = new LwwApplier(db);
|
|
var result = await applier.ApplyBatchAsync(
|
|
[
|
|
Entry(1, "{\"id\":1}", "{\"id\":1,\"sku\":\"BEFORE\",\"qty\":1}", 1_000_000, "remote"),
|
|
Entry(2, "{\"id\":2}", "NOT VALID JSON", 1_000_001, "remote"),
|
|
Entry(3, "{\"id\":3}", "{\"id\":3,\"sku\":\"AFTER\",\"qty\":3}", 1_000_002, "remote"),
|
|
Entry(4, "{\"id\":4}", "ALSO NOT JSON", 1_000_003, "remote"),
|
|
Entry(5, "{\"id\":5}", "{\"id\":5,\"sku\":\"LAST\",\"qty\":5}", 1_000_004, "remote"),
|
|
], default);
|
|
|
|
Assert.Equal(3, result.Applied);
|
|
Assert.Equal(2, result.DeadLettered);
|
|
Assert.Equal(("BEFORE", (long?)1), await ReadOrder(db, 1));
|
|
Assert.Equal(("AFTER", (long?)3), await ReadOrder(db, 3));
|
|
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_ReplaceExistingRow_DoesNotFireCaptureTriggers()
|
|
{
|
|
// INSERT OR REPLACE over an existing row also exercises the delete side of REPLACE;
|
|
// neither half may reach the capture triggers.
|
|
using var db = await NewOrdersDb();
|
|
await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'LOCAL', 5)");
|
|
var localHlc = await RowVersionHlc(db, Pk1);
|
|
var before = await OplogCount(db);
|
|
|
|
var result = await new LwwApplier(db).ApplyBatchAsync(
|
|
[Entry(80, Pk1, "{\"id\":1,\"sku\":\"REPL\",\"qty\":6}", localHlc + 1000, "remote-node")], default);
|
|
|
|
Assert.Equal(1, result.Applied);
|
|
Assert.Equal(("REPL", (long?)6), await ReadOrder(db, 1));
|
|
Assert.Equal(before, await OplogCount(db));
|
|
}
|
|
}
|