fix(localdb): applier review fixes — null-row_json guard, clock recovery via last_seen_hlc, fidelity + guard tests

Claude-Session: https://claude.ai/code/session_01BL2Vu1ESDQ9SCN4gVKkdts

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 <c>__localdb_oplog</c>: the 2-node topology does not re-forward applied deltas.
 /// </summary>
-internal sealed class LwwApplier(SqliteLocalDb db)
+internal sealed class LwwApplier(SqliteLocalDb db, Func<DateTimeOffset>? utcNow = null)
 {
+    private readonly Func<DateTimeOffset> _utcNow = utcNow ?? (() => DateTimeOffset.UtcNow);
+
     public async Task<ApplyResult> ApplyBatchAsync(IReadOnlyList<OplogEntryRecord> 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);

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<D
             "SELECT MAX((SELECT COALESCE(MAX(seq), 0) FROM __localdb_oplog), " +
             "(SELECT last_acked_seq FROM __localdb_peer_state WHERE id = 1))",
             static r => 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(