fix(localdb): keep a tombstone until the peer holds it — no resurrection past TombstoneRetention (scadaproj#16)

0.3.7 (unpublished), owner-chosen option A. A tombstone is now pruned only when
it is past TombstoneRetention AND the peer is known to hold it:
- held while database-wide or per-table snapshot debt is outstanding (the
  snapshot is built from the row-version ledger, so a pruned tombstone is a
  delete it can no longer carry — the issue's second path);
- a local tombstone held while its delete is still in the unacked oplog.
Delivery (delta ack covering the delete, or snapshot debt discharged) restamps
tombstone_utc so retention runs from delivery. New optional
LocalDb:Replication:TombstoneHardCap (default off, >= TombstoneRetention)
bounds an undelivered tombstone's life. No proto, lib_schema_version or
on-disk schema change.

Regression test TombstoneResurrectionTests (two real nodes over loopback gRPC):
offline update on B, newer delete on A, A's clock +8 days (oplog age cap +
tombstone prune) -> reconnect. Fails on 0.3.6 (row resurrected), passes now;
plus a no-prune positive control and a pruned-after-delivery control, and five
controlled-clock OplogStore tests. Each guard clause mutation-killed. 259/259.

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

ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/OplogStore.cs

@@ -56,20 +56,93 @@ internal sealed class OplogStore(ILocalDb db, ReplicationOptions options, Func<D
         // 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);
+        var now = FormatUtc(_utcNow());
+        var acked = Math.Min(ackedSeq, knownMax);
+        // scadaproj#16: the peer now holds every local tombstone in the newly acked range, so their
+        // retention clock starts HERE rather than at the delete. Must run before the prune below
+        // deletes the oplog rows that identify them, and before last_acked_seq moves (it bounds the
+        // range from below).
+        await tx.ExecuteAsync(RestampDeliveredTombstonesSql, new { acked, now }, ct);
         await tx.ExecuteAsync(
             "UPDATE __localdb_peer_state SET last_acked_seq = MAX(last_acked_seq, @acked), last_sync_utc = @now WHERE id = 1",
-            new { acked = Math.Min(ackedSeq, knownMax), now = FormatUtc(_utcNow()) }, ct);
+            new { acked, now }, ct);
         await tx.ExecuteAsync(
             "DELETE FROM __localdb_oplog WHERE seq <= (SELECT last_acked_seq FROM __localdb_peer_state WHERE id = 1)",
             null, ct);
         await tx.CommitAsync(ct);
     }
 
-    /// <summary>Deletes row-version tombstones older than <see cref="ReplicationOptions.TombstoneRetention"/>; live rows and fresh tombstones are kept.</summary>
-    public Task PruneTombstonesAsync(CancellationToken ct = default) =>
-        db.ExecuteAsync(
-            "DELETE FROM __localdb_row_version WHERE is_tombstone = 1 AND tombstone_utc < @cutoff",
-            new { cutoff = FormatUtc(_utcNow() - options.TombstoneRetention) }, ct);
+    // Local tombstones whose delete is in the acked range (old last_acked_seq, @acked]. A local
+    // tombstone's ledger row and its oplog row share (table_name, pk_json); a later re-delete of the
+    // same key that is still unacked is restamped too, which is harmless — the prune below still
+    // holds it on the unacked-oplog floor. Never moves the stamp backwards.
+    private const string RestampDeliveredTombstonesSql =
+        "UPDATE __localdb_row_version SET tombstone_utc = @now " +
+        "WHERE is_tombstone = 1 AND (tombstone_utc IS NULL OR tombstone_utc < @now) " +
+        "AND (table_name, pk_json) IN (" +
+        "SELECT table_name, pk_json FROM __localdb_oplog WHERE is_tombstone = 1 " +
+        "AND seq > (SELECT last_acked_seq FROM __localdb_peer_state WHERE id = 1) AND seq <= @acked)";
+
+    // A tombstone the peer may not hold yet. Each clause is evaluated once per statement (they are
+    // uncorrelated scalar subqueries), so the prune stays one pass over the tombstones.
+    //  - Database-wide snapshot debt: a cap prune (or an impossible ack) dropped oplog rows the peer
+    //    never received, and the snapshot that replaces them is built from this ledger — a pruned
+    //    tombstone is a delete the snapshot can no longer carry.
+    //  - Per-table snapshot debt: the same, scoped to the baselined table.
+    //  - A LOCAL tombstone at or above the unacked-oplog floor: its delete is still waiting in the
+    //    oplog. The ledger row carries the same HLC as its oplog row and local HLCs rise with seq, so
+    //    "hlc >= HLC of the first unacked oplog row" is exactly "not yet acked" (equal is held, the
+    //    conservative side; no unacked row at all ⇒ the COALESCE ceiling ⇒ nothing held). Remote-origin
+    //    tombstones came FROM the peer and are never in our oplog.
+    private const string HeldTombstonePredicate =
+        "(SELECT needs_snapshot FROM __localdb_peer_state WHERE id = 1) <> 0 " +
+        "OR table_name IN (SELECT table_name FROM __localdb_snapshot_state WHERE needs_snapshot <> 0) " +
+        "OR (node_id = (SELECT node_id FROM __localdb_meta WHERE id = 1) " +
+        "AND hlc >= COALESCE((SELECT hlc FROM __localdb_oplog " +
+        "WHERE seq > (SELECT last_acked_seq FROM __localdb_peer_state WHERE id = 1) ORDER BY seq LIMIT 1), " +
+        "9223372036854775807))";
+
+    /// <summary>
+    /// Deletes row-version tombstones that are past <see cref="ReplicationOptions.TombstoneRetention"/>
+    /// <b>and</b> known to be held by the peer; returns how many were removed.
+    /// </summary>
+    /// <remarks>
+    /// <para>
+    /// <b>Why retention alone is not enough (scadaproj#16).</b> A tombstone is the only thing that
+    /// lets LWW reject an OLDER write of a deleted row. Up to 0.3.6 it was pruned on age alone, so a
+    /// peer offline longer than the retention came back to a node that had forgotten the delete: the
+    /// delete's oplog row had been cap-pruned too, the snapshot (built from this ledger) no longer
+    /// carried it, and the peer's stale update re-created the row on both nodes.
+    /// </para>
+    /// <para>
+    /// <b>The rule.</b> A tombstone is kept while the peer might not hold it — see
+    /// <see cref="HeldTombstonePredicate"/>: outstanding snapshot debt, or a local delete still in the
+    /// unacked oplog. Once the peer does hold it (delta ack, or snapshot debt discharged) its
+    /// <c>tombstone_utc</c> is restamped to that moment, so retention is measured from DELIVERY: the
+    /// peer's own stale writes, queued before it received the delete, have a full retention to drain
+    /// before the tombstone stops being able to reject them. Remote-origin tombstones need no
+    /// delivery — the peer wrote them — and age from the moment they were applied here.
+    /// </para>
+    /// <para>
+    /// <b>Cost.</b> While the peer is down nothing local is delivered, so tombstones accumulate in the
+    /// ledger for the whole outage. <see cref="ReplicationOptions.TombstoneHardCap"/> bounds that when
+    /// set, at the price of reintroducing the resurrection risk for an outage longer than the cap.
+    /// </para>
+    /// </remarks>
+    public Task<int> PruneTombstonesAsync(CancellationToken ct = default)
+    {
+        var now = _utcNow();
+        var cutoff = FormatUtc(now - options.TombstoneRetention);
+        if (options.TombstoneHardCap is { } hardCap)
+            return db.ExecuteAsync(
+                "DELETE FROM __localdb_row_version WHERE is_tombstone = 1 AND (tombstone_utc < @hardCutoff " +
+                $"OR (tombstone_utc < @cutoff AND NOT ({HeldTombstonePredicate})))",
+                new { cutoff, hardCutoff = FormatUtc(now - hardCap) }, ct);
+        return db.ExecuteAsync(
+            "DELETE FROM __localdb_row_version WHERE is_tombstone = 1 " +
+            $"AND tombstone_utc < @cutoff AND NOT ({HeldTombstonePredicate})",
+            new { cutoff }, ct);
+    }
 
     /// <summary>
     /// The unacked backlog — oplog entries above the peer's last-acked watermark — in O(1).
@@ -293,16 +366,28 @@ internal sealed class OplogStore(ILocalDb db, ReplicationOptions options, Func<D
     public async Task ClearSnapshotDebtAsync(IReadOnlyList<string> tables, CancellationToken ct = default)
     {
         await using var tx = await db.BeginTransactionAsync(ct);
+        var now = FormatUtc(_utcNow());
         if (tables.Count == 0)
         {
             await tx.ExecuteAsync("UPDATE __localdb_peer_state SET needs_snapshot = 0 WHERE id = 1", null, ct);
             await tx.ExecuteAsync("DELETE FROM __localdb_snapshot_state", null, ct);
+            // scadaproj#16: the snapshot carried every tombstone; retention restarts at delivery.
+            await tx.ExecuteAsync(
+                "UPDATE __localdb_row_version SET tombstone_utc = @now " +
+                "WHERE is_tombstone = 1 AND (tombstone_utc IS NULL OR tombstone_utc < @now)",
+                new { now }, ct);
         }
         else
         {
             foreach (var table in tables)
+            {
                 await tx.ExecuteAsync(
                     "DELETE FROM __localdb_snapshot_state WHERE table_name = @t", new { t = table }, ct);
+                await tx.ExecuteAsync(
+                    "UPDATE __localdb_row_version SET tombstone_utc = @now WHERE table_name = @t " +
+                    "AND is_tombstone = 1 AND (tombstone_utc IS NULL OR tombstone_utc < @now)",
+                    new { t = table, now }, ct);
+            }
         }
         await tx.CommitAsync(ct);
     }

ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/MaintenanceBackgroundService.cs

@@ -8,7 +8,7 @@ namespace ZB.MOM.WW.LocalDb.Replication;
 /// <summary>
 /// Periodic oplog maintenance: every <see cref="ReplicationOptions.MaintenanceInterval"/> it
 /// enforces the backlog/age caps (which flag <c>needs_snapshot</c> when exceeded) and prunes
-/// expired tombstone row-versions. Runs on BOTH roles — a passive node with a long-dead peer is
+/// expired tombstone row-versions the peer is known to hold (see <c>OplogStore.PruneTombstonesAsync</c>). Runs on BOTH roles — a passive node with a long-dead peer is
 /// exactly the case where the oplog would otherwise grow unbounded. A failing tick is logged and
 /// skipped; maintenance never dies.
 /// </summary>
@@ -48,7 +48,7 @@ public sealed class MaintenanceBackgroundService : BackgroundService
             try
             {
                 var snapshotFlagged = await _store.EnforceCapsAsync(stoppingToken);
-                await _store.PruneTombstonesAsync(stoppingToken);
+                var pruned = await _store.PruneTombstonesAsync(stoppingToken);
 
                 if (snapshotFlagged && !_snapshotFlagWarned)
                 {
@@ -61,7 +61,9 @@ public sealed class MaintenanceBackgroundService : BackgroundService
                     _snapshotFlagWarned = false;
                 }
 
-                _logger.LogDebug("Maintenance tick complete (snapshotFlagged={SnapshotFlagged}).", snapshotFlagged);
+                _logger.LogDebug(
+                    "Maintenance tick complete (snapshotFlagged={SnapshotFlagged}, tombstonesPruned={TombstonesPruned}).",
+                    snapshotFlagged, pruned);
             }
             catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
             {

ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/ReplicationOptions.cs

@@ -66,9 +66,32 @@ public sealed class ReplicationOptions
     /// <summary>Age ceiling: an unpruned oplog entry older than this flags a snapshot resync.</summary>
     public TimeSpan MaxOplogAge { get; set; } = TimeSpan.FromDays(7);
 
-    /// <summary>How long deleted-row tombstones are retained before pruning.</summary>
+    /// <summary>
+    /// How long a deleted-row tombstone is retained once the peer holds it, before pruning.
+    /// </summary>
+    /// <remarks>
+    /// Since 0.3.7 (scadaproj#16) a tombstone is pruned only when it is past this retention AND the
+    /// peer has received it (its delete acked, or the snapshot that carries it acknowledged), and the
+    /// retention clock restarts at that delivery. Up to 0.3.6 it was pruned on age alone, which let a
+    /// peer offline longer than the retention resurrect the deleted row. A tombstone the peer has
+    /// not received is kept for as long as the outage lasts — bounded only by
+    /// <see cref="TombstoneHardCap"/>.
+    /// </remarks>
     public TimeSpan TombstoneRetention { get; set; } = TimeSpan.FromDays(7);
 
+    /// <summary>
+    /// Optional absolute ceiling on how long a tombstone the peer has NOT received is kept. Default
+    /// <c>null</c> (off): an undelivered tombstone is kept until the peer receives it, however long
+    /// that takes. Must be at least <see cref="TombstoneRetention"/> when set.
+    /// </summary>
+    /// <remarks>
+    /// Bounds ledger growth on a node whose peer is gone for good (high-churn tables such as a
+    /// store-and-forward queue add one tombstone per delete for the whole outage). The trade is
+    /// explicit: a peer that comes back after more than this can resurrect rows deleted while it was
+    /// away, exactly as before 0.3.7 — rebuild such a peer instead of reconnecting it.
+    /// </remarks>
+    public TimeSpan? TombstoneHardCap { get; set; }
+
     /// <summary>Upper bound on the exponential reconnect backoff.</summary>
     public TimeSpan ReconnectBackoffMax { get; set; } = TimeSpan.FromSeconds(60);
 

ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/ReplicationOptionsValidator.cs

@@ -36,6 +36,9 @@ public sealed class ReplicationOptionsValidator : IValidateOptions<ReplicationOp
             failures.Add($"LocalDb:Replication:MaxOplogAge must be greater than zero; got {options.MaxOplogAge}.");
         if (options.TombstoneRetention <= TimeSpan.Zero)
             failures.Add($"LocalDb:Replication:TombstoneRetention must be greater than zero; got {options.TombstoneRetention}.");
+        if (options.TombstoneHardCap is { } hardCap && hardCap < options.TombstoneRetention)
+            failures.Add(
+                $"LocalDb:Replication:TombstoneHardCap must be at least TombstoneRetention ({options.TombstoneRetention}) when set; got {hardCap}.");
         if (options.ReconnectBackoffMax <= TimeSpan.Zero)
             failures.Add($"LocalDb:Replication:ReconnectBackoffMax must be greater than zero; got {options.ReconnectBackoffMax}.");
         else if (options.ReconnectBackoffMax > ReconnectBackoffCeiling)