fix(version): persist the retention boundary across reopen

The boundary a GC compaction or `clear` established lived only in the
in-memory version history: `SuperVersions::new` seeded the recovered
version at seqno 0, so after a restart every historical snapshot was
served from the surviving version, silently answering with data the
snapshot never saw (the exact outcome `SnapshotBelowRetention` exists to
refuse).

- `Version::retention_floor`: the highest snapshot seqno the version can
  no longer serve, monotone across versions; persisted as the manifest
  `retention_floor` section and an appended edit-log field (both
  optional, so older manifests recover as floor 0)
- `RetentionEffect` names what an install does to older snapshots and is
  passed to `upgrade_version`: a GC compaction (`GcBelow(w)`) raises the
  floor to `w - 1`, a `clear` / table drop (`DropsData`) to its own
  install seqno, additive installs (`Keep`) leave it alone; the floor
  rides in the same version edit as the data loss it records
- a reopened history is seeded at the floor, version seqnos are clamped
  non-decreasing (a counter reset below the floor cannot slip a version
  under it) and `get_version_for_snapshot` checks the front explicitly
- `Config::repair_retention_floor` (default 0): a rebuilt manifest cannot
  derive the floor (a GC compaction zeroes the settled rows' seqnos) and
  must not guess it (the external-WAL reconciliation reads intermediate
  snapshots back), so the deployment that ran the compactions supplies it
- `BlobTree::multi_get` validates the snapshot before the empty-batch
  return, matching `Tree::multi_get`
- the retention fixture pins `first + 1 < oldest_retained_seqno()` so the
  strict-below probes stay strict
- docs: INVARIANTS (durable boundary), manifest-recovery (retention
  floor), external-wal (repair floor)

Regression tests cover reopen after GC compaction / clear / drop_range /
FIFO eviction / leveled merge, additive installs, manifest rotation, a
reopen chain, a reset counter, checkpoints and repair, on both tree
types; edit / diff / recovery / history unit tests cover the codec and
the boundary check.

Part of #616

Author: polaz

docs/INVARIANTS.md

@@ -145,6 +145,21 @@ matching entry (and add one for a new subsystem).
   `src/version/super_version.rs` (`get_version_for_snapshot`) and surfaced by
   every read path that resolves a snapshot (`src/tree`, `src/blob_tree`).
 
+- **The retention boundary is durable.** An install that discards what older
+  snapshots saw raises the version's *retention floor* in the same version
+  edit (`Version::retention_floor`, the `retention_floor` manifest section and
+  the appended edit-log field): a GC compaction with watermark `w` sets it to
+  `w - 1`, a `clear` or a table drop to its own install seqno; a flush,
+  ingest, move or relocation leaves it alone. A reopened history is seeded at
+  the floor, so the snapshots the live tree refused stay refused after a
+  restart instead of being answered from the surviving version. Version seqnos
+  are non-decreasing along the history (`upgrade_version_with_seqno` clamps),
+  so a counter reset below the floor cannot slip a version under it. A
+  manifest rebuilt by `Config::repair` seeds the floor from
+  `Config::repair_retention_floor` (default `0`): the tables cannot record it
+  and the engine must not guess it (see
+  [manifest-recovery.md](manifest-recovery.md#retention-floor)).
+
 - **Retention is versioned, so its storage cost scales with write + compaction
   volume.** History is served from whole retained `SuperVersion`s: every table a
   compaction consumed stays on disk until the GC watermark passes that

docs/external-wal.md

@@ -240,6 +240,16 @@ up to `b` for `LostUpTo(b)`, unbounded for `FullHistory`.
 Run the replay BEFORE publishing your visible watermark, so readers never
 observe the repaired-but-not-yet-reconciled state.
 
+**Snapshots below your GC watermark stay refused.** A live tree refuses a
+snapshot read below the history a compaction's GC watermark collected
+(`Error::SnapshotBelowRetention`), and the manifest carries that boundary
+across a normal reopen. A repair rebuilds the manifest from the tables, which
+do not record it, so pass the highest `seqno_threshold` you ever applied,
+minus one, as `Config::repair_retention_floor` on the repairing `Config`: the
+repaired tree then refuses exactly the snapshots the source refused, and the
+reconciled replay above the floor stays readable. Left at `0`, the repaired
+tree serves every snapshot, including ones a past compaction collected.
+
 ## Executable companion
 
 This recipe is not only specified here; it is executed and self-verified in the

docs/manifest-recovery.md

@@ -312,6 +312,31 @@ claiming success would describe a tree that cannot open. Quarantine is not
 a fallback here — it preserves damaged *data*, and this file holds none. A
 retry after the filesystem is fixed completes the drop.
 
+## Retention floor
+
+A live tree refuses a snapshot read below the oldest version its history
+retains (`Error::SnapshotBelowRetention`), and the manifest records that
+boundary as the version's *retention floor* so the refusal holds across a
+reopen: a GC compaction raises it to its watermark minus one, a `clear` or a
+table drop to the install's own seqno. The lost manifest was the floor's only
+durable copy, and the tables cannot stand in for it: a GC compaction zeroes
+the seqnos of the rows it settles, so a table's highest seqno reads as `0`
+precisely on the trees where history was collected.
+
+Nor may the rebuilt manifest guess the floor high (say, at the highest
+persisted seqno): the external-WAL reconciliation that follows a repair
+([external-wal.md](external-wal.md), section 4) restores intermediate
+snapshots and reads them back, so a guessed floor would refuse history the
+caller has just made whole. The floor is therefore **supplied by the caller**,
+the only party that knows the GC watermarks it passed:
+`Config::repair_retention_floor(floor)` seeds the rebuilt manifest with it,
+and the reopened tree refuses every snapshot at or below `floor` exactly as it
+did before the manifest was lost. Pass the highest compaction watermark ever
+applied, minus one. Left at the default `0`, a repaired tree serves every
+snapshot, which is correct only if history was never collected. No
+resurrection question arises: the floor discards nothing, it only declines to
+answer what cannot be answered honestly.
+
 ## Blob frame validation and salvage
 
 Before a blob file's digest is recorded, its live frame range is walked frame

src/abstract_tree.rs

@@ -997,6 +997,17 @@ pub trait AbstractTree: sealed::Sealed {
     /// here before reading and report "history collected" instead of an
     /// unexpected error mid-scan. A fresh tree reports `0`.
     ///
+    /// The boundary survives a reopen. The install that discards what older
+    /// snapshots saw records it in the same version edit: after a GC
+    /// compaction with watermark `w` the reopened boundary is `w - 1` (the
+    /// retained pre-compaction version that served reads between the live
+    /// front and `w` does not survive a restart), after a `clear` or a table
+    /// drop it is that install's seqno. A manifest rebuilt by
+    /// [`Config::repair`](crate::Config::repair) cannot know what the lost
+    /// manifest recorded and seeds the boundary from
+    /// [`Config::repair_retention_floor`](crate::Config::repair_retention_floor)
+    /// (default `0`: every snapshot served).
+    ///
     /// # Examples
     ///
     /// ```

src/blob_tree/ingest.rs

@@ -299,6 +299,9 @@ impl<'a> BlobIngestion<'a> {
             &*self.tree.index.config.fs,
             self.tree.index.0.runtime_config.load_full(),
             self.tree.index.0.config.encryption.clone(),
+            // Ingestion only adds tables and blob files: older snapshots keep
+            // everything.
+            crate::version::RetentionEffect::Keep,
         )?;
 
         // Perform maintenance on the version history (e.g., clean up old versions).

src/blob_tree/mod.rs

@@ -951,6 +951,9 @@ impl AbstractTree for BlobTree {
             &*config.fs,
             self.index.0.runtime_config.load_full(),
             self.index.0.config.encryption.clone(),
+            // Every table and blob file goes: no snapshot up to this install
+            // is servable after a reopen.
+            crate::version::RetentionEffect::DropsData,
         )?;
 
         // Same MVCC-safe reclaim as the standard tree, plus the blob files:
@@ -1422,13 +1425,17 @@ impl AbstractTree for BlobTree {
         keys: impl IntoIterator<Item = K>,
         seqno: SeqNo,
     ) -> crate::Result<Vec<Option<crate::UserValue>>> {
+        // The snapshot is validated BEFORE the empty-batch return: the
+        // retention contract is per read, not per key, and the standard tree
+        // refuses an empty batch below retention the same way.
+        let super_version = self.index.snapshot_for_read(seqno)?;
+
         let keys: Vec<_> = keys.into_iter().collect();
         let n = keys.len();
         if n == 0 {
             return Ok(Vec::new());
         }
 
-        let super_version = self.index.snapshot_for_read(seqno)?;
         let comparator = self.index.config.comparator.as_ref();
 
         // For small batches, use the simple per-key path

src/compaction/flavour.rs

@@ -478,6 +478,9 @@ pub(super) fn install_merge(
             &*opts.config.fs,
             opts.runtime_config.load_full(),
             opts.encryption.clone(),
+            // The merge stream GC'd below the watermark: versions a snapshot
+            // below it depended on are gone from the outputs.
+            crate::version::RetentionEffect::GcBelow(opts.mvcc_gc_watermark),
         )
         .inspect_err(|_| {
             for table in &created_tables {

src/compaction/worker.rs

@@ -1330,6 +1330,9 @@ fn run_tight_space_compaction(
                 &*opts.config.fs,
                 opts.runtime_config.load_full(),
                 opts.encryption.clone(),
+                // A slice is a merge with GC below the watermark, plus the
+                // punched input prefix: older snapshots lose both.
+                crate::version::RetentionEffect::GcBelow(opts.mvcc_gc_watermark),
             );
             if let Err(e) = install {
                 // The install did not commit, so no sidecar was written (the mark
@@ -1849,6 +1852,8 @@ fn move_tables(
         &*opts.config.fs,
         opts.runtime_config.load_full(),
         opts.encryption.clone(),
+        // A trivial move rewrites nothing: every version survives as-is.
+        crate::version::RetentionEffect::Keep,
     )?;
 
     if let Err(e) = version_history_lock.maintenance(
@@ -2848,6 +2853,10 @@ fn drop_tables(
         &*opts.config.fs,
         opts.runtime_config.load_full(),
         opts.encryption.clone(),
+        // Whole tables go (drop-range, FIFO / TTL eviction): the rows every
+        // older snapshot saw in them are gone, so no snapshot up to this
+        // install is servable after a reopen.
+        crate::version::RetentionEffect::DropsData,
     )?;
 
     if let Err(e) = version_history_lock.maintenance(

src/config/mod.rs

@@ -611,6 +611,16 @@ pub struct Config {
     /// on every upgrade, degenerating to the full-rewrite-per-version behaviour.
     pub(crate) manifest_log_rotate_bytes: u64,
 
+    /// Retention floor a manifest REPAIR seeds the rebuilt version with: the
+    /// highest snapshot seqno the repaired tree refuses to serve. The lost
+    /// manifest was the only record of the floor a past GC compaction or
+    /// `clear` established, and the tables cannot stand in for it, so the
+    /// deployment that ran those compactions (and knows the watermark it
+    /// passed) supplies it here. Defaults to `0`: a repaired tree serves
+    /// every snapshot, which is right only if history was never collected.
+    /// Set via [`Config::repair_retention_floor`].
+    pub(crate) repair_retention_floor: crate::SeqNo,
+
     /// Compaction I/O rate limit in bytes per second.
     ///
     /// Caps the rate at which the compaction worker is allowed to issue
@@ -792,6 +802,7 @@ impl Default for Config {
             #[cfg(feature = "std")]
             recovery_progress: None,
             manifest_log_rotate_bytes: 1024 * 1024,
+            repair_retention_floor: 0,
             compaction_rate_limit: 0,
 
             #[cfg(feature = "std")]
@@ -1633,6 +1644,28 @@ impl Config {
         self
     }
 
+    /// Sets the retention floor a manifest repair seeds the rebuilt tree with
+    /// (default `0`): after [`repair`](Self::repair) /
+    /// [`open_or_repair`](Self::open_or_repair) every snapshot at or below
+    /// `floor` is refused with
+    /// [`Error::SnapshotBelowRetention`](crate::Error::SnapshotBelowRetention),
+    /// exactly as the tree refused it before the manifest was lost.
+    ///
+    /// A normal open needs no help: the manifest carries the floor a GC
+    /// compaction or `clear` established (see
+    /// [`AbstractTree::oldest_retained_seqno`](crate::AbstractTree::oldest_retained_seqno)).
+    /// A repair rebuilds the manifest from the tables, which do not record it
+    /// (a GC compaction zeroes the seqnos of the rows it settles), so only the
+    /// deployment that ran those compactions knows it: pass the highest GC
+    /// watermark (`seqno_threshold`) ever applied, minus one. Left at `0`, a
+    /// repaired tree serves every snapshot, which is correct only if history
+    /// was never collected. Has no effect on an open that finds a manifest.
+    #[must_use]
+    pub fn repair_retention_floor(mut self, floor: crate::SeqNo) -> Self {
+        self.repair_retention_floor = floor;
+        self
+    }
+
     /// Sets the compaction I/O rate limit in bytes per second.
     ///
     /// Caps how fast the compaction worker may issue I/O so background

src/error.rs

@@ -463,7 +463,8 @@ pub enum Error {
     /// from. Serving it from the oldest retained version instead would
     /// silently answer with data the snapshot never saw, so the read is
     /// refused. Snapshot `0` is the exception: it sees no entry from any
-    /// version and is always served (empty).
+    /// version and is always served (empty). The boundary is persisted with
+    /// the manifest, so the refusal holds across a reopen as well.
     ///
     /// Point reads return it directly; iterators yield it as their first and
     /// only item. [`AbstractTree::oldest_retained_seqno`](crate::AbstractTree::oldest_retained_seqno)