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()
{