Joseph Doherty
41bdc178c2
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.
552 lines
24 KiB
C#
552 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, 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.
|
|
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));
|
|
}
|
|
|
|
[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]
|
|
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));
|
|
}
|
|
}
|