From b223507070079d7d8a301730ff19161b05b05103 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Fri, 17 Jul 2026 22:15:51 -0400 Subject: [PATCH] =?UTF-8?q?fix(localdb):=20applier=20review=20fixes=20?= =?UTF-8?q?=E2=80=94=20null-row=5Fjson=20guard,=20clock=20recovery=20via?= =?UTF-8?q?=20last=5Fseen=5Fhlc,=20fidelity=20+=20guard=20tests?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Claude-Session: https://claude.ai/code/session_01BL2Vu1ESDQ9SCN4gVKkdts --- .../Internal/LwwApplier.cs | 53 +++++--- .../Internal/OplogStore.cs | 2 + .../Internal/SqliteLocalDb.cs | 6 +- .../LwwApplierTests.cs | 116 ++++++++++++++++++ .../OplogStoreTests.cs | 21 ++++ 5 files changed, 179 insertions(+), 19 deletions(-) diff --git a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/LwwApplier.cs b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/LwwApplier.cs index 6c37e81..e090cff 100644 --- a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/LwwApplier.cs +++ b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/LwwApplier.cs @@ -16,8 +16,10 @@ internal sealed record ApplyResult(long AppliedThruSeq, int Applied, int Discard /// crash-replay of the batch is a no-op (LWW's strict-greater compare loses on equal versions). /// This never writes __localdb_oplog: the 2-node topology does not re-forward applied deltas. /// -internal sealed class LwwApplier(SqliteLocalDb db) +internal sealed class LwwApplier(SqliteLocalDb db, Func? utcNow = null) { + private readonly Func _utcNow = utcNow ?? (() => DateTimeOffset.UtcNow); + public async Task ApplyBatchAsync(IReadOnlyList entries, CancellationToken ct) { if (entries.Count == 0) @@ -31,7 +33,7 @@ internal sealed class LwwApplier(SqliteLocalDb db) throw new LocalDbSchemaMismatchException( $"Received an oplog entry for table '{entry.TableName}', which is not registered for replication on this node."); - var utcNow = DateTimeOffset.UtcNow.UtcDateTime.ToString(OplogStore.TombstoneUtcFormat, CultureInfo.InvariantCulture); + var utcNow = _utcNow().UtcDateTime.ToString(OplogStore.TombstoneUtcFormat, CultureInfo.InvariantCulture); var maxSeq = 0L; var maxHlc = 0L; foreach (var entry in entries) @@ -50,6 +52,15 @@ internal sealed class LwwApplier(SqliteLocalDb db) foreach (var entry in entries) { + // Without row data the upsert would bind NULL to every column — on an INTEGER PRIMARY + // KEY table that silently inserts a phantom auto-rowid row — so it is poison, not a win. + if (!entry.IsTombstone && entry.RowJson is null) + { + await DeadLetterAsync(tx, entry, "non-tombstone entry with null row_json", utcNow, ct); + deadLettered++; + continue; + } + var local = await tx.QueryAsync( "SELECT hlc, node_id FROM __localdb_row_version WHERE table_name = @t AND pk_json = @pk", static r => (Hlc: r.GetInt64(0), NodeId: r.GetString(1)), @@ -95,22 +106,11 @@ internal sealed class LwwApplier(SqliteLocalDb db) } catch (SqliteException ex) { - // A statement-level SQLite error (e.g. json_extract on malformed row_json) aborts only - // that statement, not the transaction, so the batch continues after dead-lettering. - await tx.ExecuteAsync( - "INSERT INTO __localdb_dead_letter " + - "(received_utc, table_name, pk_json, row_json, hlc, node_id, error) " + - "VALUES (@utc, @t, @pk, @row, @hlc, @node, @err)", - new - { - utc = utcNow, - t = entry.TableName, - pk = entry.PkJson, - row = entry.RowJson, - hlc = entry.Hlc, - node = entry.NodeId, - err = ex.Message, - }, ct); + // 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 DeadLetterAsync(tx, entry, ex.Message, utcNow, ct); deadLettered++; } } @@ -128,6 +128,23 @@ internal sealed class LwwApplier(SqliteLocalDb db) return new ApplyResult(maxSeq, applied, discarded, deadLettered); } + private static Task DeadLetterAsync( + ILocalDbTransaction tx, OplogEntryRecord entry, string error, string utcNow, CancellationToken ct) => + tx.ExecuteAsync( + "INSERT INTO __localdb_dead_letter " + + "(received_utc, table_name, pk_json, row_json, hlc, node_id, error) " + + "VALUES (@utc, @t, @pk, @row, @hlc, @node, @err)", + new + { + utc = utcNow, + t = entry.TableName, + pk = entry.PkJson, + row = entry.RowJson, + hlc = entry.Hlc, + node = entry.NodeId, + err = error, + }, ct); + private static (string Delete, string Upsert) BuildTableSql(ReplicatedTable table) { var name = Ident(table.Name); diff --git a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/OplogStore.cs b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/OplogStore.cs index 078784d..ee28a65 100644 --- a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/OplogStore.cs +++ b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/OplogStore.cs @@ -51,6 +51,8 @@ internal sealed class OplogStore(ILocalDb db, ReplicationOptions options, Func r.GetInt64(0), null, ct))[0]; + // Pruning up to the clamped ack may delete rows the peer never actually received; that is + // safe only because needs_snapshot = 1 commits in the SAME transaction, forcing a resync. if (ackedSeq > knownMax) await tx.ExecuteAsync("UPDATE __localdb_peer_state SET needs_snapshot = 1 WHERE id = 1", null, ct); await tx.ExecuteAsync( diff --git a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/Internal/SqliteLocalDb.cs b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/Internal/SqliteLocalDb.cs index 122171e..abfd1e5 100644 --- a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/Internal/SqliteLocalDb.cs +++ b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/Internal/SqliteLocalDb.cs @@ -61,11 +61,15 @@ internal sealed class SqliteLocalDb : ILocalDb, IDisposable LocalDbSchema.EnsureCreated(_master); + // last_seen_hlc covers remote HLCs whose only local trace was a dead-lettered entry: after + // an ungraceful crash neither oplog nor row_version carries them, and recovering below a + // peer HLC already observed would let the clock issue regressed stamps. var initial = ScalarLong(_master, "SELECT MAX(" + "(SELECT hlc_high_water FROM __localdb_meta WHERE id=1)," + "COALESCE((SELECT MAX(hlc) FROM __localdb_oplog),0)," + - "COALESCE((SELECT MAX(hlc) FROM __localdb_row_version),0))"); + "COALESCE((SELECT MAX(hlc) FROM __localdb_row_version),0)," + + "COALESCE((SELECT last_seen_hlc FROM __localdb_peer_state WHERE id=1),0))"); _clock = new HybridLogicalClock(initial); _master.CreateFunction("zb_hlc_next", () => _clock.Next()); diff --git a/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/LwwApplierTests.cs b/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/LwwApplierTests.cs index a3cf59b..ccd8896 100644 --- a/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/LwwApplierTests.cs +++ b/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/LwwApplierTests.cs @@ -10,12 +10,15 @@ 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 NewOrdersDb() @@ -305,4 +308,117 @@ public sealed class LwwApplierTests : IDisposable 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_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)); + } } diff --git a/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/OplogStoreTests.cs b/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/OplogStoreTests.cs index becfd3e..d88a6c3 100644 --- a/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/OplogStoreTests.cs +++ b/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/OplogStoreTests.cs @@ -191,6 +191,27 @@ public sealed class OplogStoreTests : IDisposable Assert.Equal(0, count[0]); } + [Fact] + public async Task CapExceeded_ByAge_MixedFixture_FreshRowSurvives() + { + using var db = await NewOrdersDb(); + await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1,'A',1),(2,'B',2)"); + // Backdate only the first entry's HLC past the 7d MaxOplogAge; the second stays fresh. + var expiredHlc = DateTimeOffset.UtcNow.AddDays(-30).ToUnixTimeMilliseconds() << 16; + await db.ExecuteAsync( + "UPDATE __localdb_oplog SET hlc = @h WHERE seq = (SELECT MIN(seq) FROM __localdb_oplog)", + new { h = expiredHlc }); + var store = new OplogStore(db, new ReplicationOptions()); + + var flagged = await store.EnforceCapsAsync(default); + + Assert.True(flagged); + Assert.True((await store.GetPeerStateAsync(default)).NeedsSnapshot); + var remaining = await store.ReadBatchAboveAsync(0, 100, default); + Assert.Single(remaining); + Assert.Equal("{\"id\":2}", remaining[0].PkJson); + } + [Fact] public async Task PeerState_RoundTrips() {