fix(version): refuse a snapshot read below the retained history (#617)

## Summary

`SuperVersions::get_version_for_snapshot` panicked with `should always
find a SuperVersion` when a read asked for a snapshot at or below the
oldest retained version's seqno. Compaction maintenance (with the
caller's GC watermark) and `clear` prune the version history past such
snapshots, so every read API (`get`, `multi_get`, range / prefix
iterators, seekable and batch scans, columnar scan, range estimates)
could take the tree down on a valid argument.

The read is now refused with a typed error instead of served from a
newer version (which would silently return data the snapshot never saw),
and the boundary is durable:

- `Error::SnapshotBelowRetention { requested, oldest_retained }`: point
reads return it directly; iterators yield it as their first and only
item (also via `peek_key` on the seekable iterator, whose seeks become
no-ops). Standard and KV-separated trees alike, including an empty
`multi_get` batch.
- `AbstractTree::oldest_retained_seqno()`: the read boundary, so a
caller can validate a long-lived snapshot before reading. A snapshot is
servable iff it is `0` or strictly above that seqno.
- Snapshot `0` stays served from the oldest retained version: nothing is
visible at `0` from any version, so the choice is immaterial and probing
an empty tree keeps working after pruning.
- **The boundary survives a reopen.** `Version::retention_floor`
(persisted as a manifest section plus an appended edit-log field, both
optional so older manifests recover as `0`) records the highest snapshot
an install made unservable: a GC compaction with watermark `w` sets it
to `w - 1`, capped at the compaction's own install seqno (so a
`SeqNo::MAX` watermark cannot refuse every later snapshot); a `clear` /
`drop_range` / FIFO eviction / a compaction whose user filter removed or
rewrote rows sets it to its own install seqno; additive installs (flush,
ingest, trivial move) and an empty drop leave it alone. The install
passes a `RetentionEffect` to `upgrade_version`, so 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, and the
front is checked explicitly, so a counter reset below the floor cannot
slip a version under it.
- `AbstractTree::retention_floor()`: the persisted boundary (what a
reopen enforces), distinct from the live `oldest_retained_seqno()`, and
the value a deployment records for a later repair.
- `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 after a repair), so the deployment supplies
the `retention_floor()` it last recorded.
- `Tree::create_{iter,range,prefix,seekable_range_bounds}` (doc-hidden)
return `Result` and raise the error before any I/O; the happy path gains
no per-item branch, and the lock-free `snapshot_for_read` fast path is
unchanged.
- Docs: `INVARIANTS.md` (retention boundary, its durability, and the
storage cost of a retention window), `manifest-recovery.md` (retention
floor after a repair), `external-wal.md` (repair floor), rustdoc on the
new API.

## Testing

- `tests/snapshot_below_retention.rs` (31 tests): every read surface on
both tree types, the `clear` path, snapshot `0`, the exact boundary
(`oldest` fails, `oldest + 1` succeeds), the unpruned case; the boundary
after a reopen following a GC compaction, `clear`, `drop_range`, FIFO
eviction, a leveled merge and a filtering compaction at watermark 0, a
`SeqNo::MAX` watermark capped at the install, an empty `drop_range`
leaving it alone, additive installs leaving it at `0`, manifest rotation
(floor read from the snapshot section), a reopen chain with a rising
floor, a reset counter, the persisted-vs-live boundary, checkpoints, and
repair with and without the configured floor.
- Unit tests: edit codec round-trips (floor with / without blob
frontiers, pre-floor payloads), `diff` emits the floor only when
changed, recovery applies it, the history seeds at the floor and checks
the front before searching.
- Gates: `cargo fmt --check`, `cargo clippy --workspace --all-targets`
(default and `--all-features`) with `-D warnings`, `cargo doc --no-deps`
(default and `--all-features`) 0 warnings, no-std check 0 errors, `cargo
nextest run --workspace --all-features` 3288/3288, `cargo test --doc
--all-features` 80/80.

Closes #616

Author: polaz

docs/INVARIANTS.md

@@ -132,6 +132,45 @@ matching entry (and add one for a new subsystem).
   visible at read seqno `s + 1`. Enforced in the read path (`src/tree`, `src/mvcc_stream.rs`) and the
   seqno ordering (`src/seqno.rs`, `src/value.rs`).
 
+- **A snapshot is served from a retained version, or refused; never clamped.**
+  A read at snapshot `R` resolves to the newest `SuperVersion` installed below
+  `R`. Compaction maintenance keeps only the newest version below the caller's
+  GC watermark (`major_compact`'s `seqno_threshold`) and releases the older
+  ones, and `clear` drains the history to the new empty version; a read at
+  `0 < R <= oldest_retained_seqno()` then has no version to be served from and
+  fails with `Error::SnapshotBelowRetention` (point reads directly, iterators as
+  their first and only item) rather than being served from a newer version,
+  which would return data the snapshot never saw. Snapshot `0` sees nothing
+  from any version and is always served empty. Enforced in
+  `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` (capped at its own install seqno), a `clear`, a table drop or a
+  compaction whose filter removed or rewrote rows to its own install seqno;
+  a flush, ingest, move, relocation or an empty drop leaves it alone.
+  `AbstractTree::retention_floor` exposes the persisted value. 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
+  compaction's install seqno, whether or not the snapshot window still needs any
+  key version inside it. A wide window therefore costs the disk of every flush
+  and compaction output produced while it was open, not just the disk of the
+  superseded key versions; keep `seqno_threshold` as close to the oldest live
+  snapshot as the caller can prove.
+
 - **Re-applying a put / delete at its original seqno is idempotent; a merge
   operand is NOT.** For a put or delete, the same (key, value, seqno) reproduces
   the same MVCC version (an overwrite), which is what makes external-WAL replay

docs/external-wal.md

@@ -240,6 +240,20 @@ 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 the retention boundary stay refused.** A live tree refuses
+a snapshot read below the history it no longer holds
+(`Error::SnapshotBelowRetention`): a compaction's GC watermark, a `clear`, a
+table drop (`drop_range`, FIFO eviction) or a filtering compaction each raise
+that boundary, and the manifest carries it across a normal reopen. A repair
+rebuilds the manifest from the tables, which do not record it, so record
+`retention_floor()` in your own durable state whenever you record the trim
+watermark `W` (it already folds in every operation that raised the boundary,
+not only the GC watermark) and pass the last recorded value 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 or drop 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,34 @@ 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 (capped at the
+compaction's own install seqno), a `clear`, a table drop or a compaction
+whose filter transformed rows 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**:
+`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. Record `AbstractTree::retention_floor()` in
+the deployment's own durable state (it is the maximum over every operation
+that raised the boundary: GC compactions, `clear`, table drops, filtering
+compactions) and pass the last recorded value; deriving it from GC watermarks
+alone would miss the drops. 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

@@ -565,6 +565,11 @@ pub trait AbstractTree: sealed::Sealed {
     /// Returns an iterator that scans through the entire tree.
     ///
     /// Avoid using this function, or limit it as otherwise it may scan a lot of items.
+    ///
+    /// A snapshot the history no longer retains (see
+    /// [`oldest_retained_seqno`](Self::oldest_retained_seqno)) yields
+    /// [`Error::SnapshotBelowRetention`](crate::Error::SnapshotBelowRetention)
+    /// as the iterator's first and only item.
     fn iter(
         &self,
         seqno: SeqNo,
@@ -576,6 +581,11 @@ pub trait AbstractTree: sealed::Sealed {
     /// Returns an iterator over a prefixed set of items.
     ///
     /// Avoid using an empty prefix as it may scan a lot of items (unless limited).
+    ///
+    /// A snapshot the history no longer retains (see
+    /// [`oldest_retained_seqno`](Self::oldest_retained_seqno)) yields
+    /// [`Error::SnapshotBelowRetention`](crate::Error::SnapshotBelowRetention)
+    /// as the iterator's first and only item.
     fn prefix<K: AsRef<[u8]>>(
         &self,
         prefix: K,
@@ -586,6 +596,11 @@ pub trait AbstractTree: sealed::Sealed {
     /// Returns an iterator over a range of items.
     ///
     /// Avoid using full or unbounded ranges as they may scan a lot of items (unless limited).
+    ///
+    /// A snapshot the history no longer retains (see
+    /// [`oldest_retained_seqno`](Self::oldest_retained_seqno)) yields
+    /// [`Error::SnapshotBelowRetention`](crate::Error::SnapshotBelowRetention)
+    /// as the iterator's first and only item.
     fn range<K: AsRef<[u8]>, R: RangeBounds<K>>(
         &self,
         range: R,
@@ -601,6 +616,12 @@ pub trait AbstractTree: sealed::Sealed {
     /// so a consumer can jump a live iterator to any key (`RocksDB` `Seek` /
     /// `SeekForPrev`) — enabling data-dependent scans (joins, skip-scan) without
     /// reopening per-SST readers per jump.
+    ///
+    /// A snapshot the history no longer retains (see
+    /// [`oldest_retained_seqno`](Self::oldest_retained_seqno)) yields
+    /// [`Error::SnapshotBelowRetention`](crate::Error::SnapshotBelowRetention)
+    /// as the iterator's first and only item (also through `peek_key`); seeks
+    /// on such an iterator are no-ops.
     fn range_seekable<K: AsRef<[u8]>, R: RangeBounds<K>>(
         &self,
         range: R,
@@ -619,6 +640,11 @@ pub trait AbstractTree: sealed::Sealed {
     ///
     /// The interval source is pulled lazily, so intervals may be produced on
     /// demand (e.g. computed from rows already returned).
+    ///
+    /// A snapshot the history no longer retains (see
+    /// [`oldest_retained_seqno`](Self::oldest_retained_seqno)) yields
+    /// [`Error::SnapshotBelowRetention`](crate::Error::SnapshotBelowRetention)
+    /// as the iterator's first and only item.
     fn batch_range_scan<K: AsRef<[u8]>, R: RangeBounds<K> + 'static, I: IntoIterator<Item = R>>(
         &self,
         intervals: I,
@@ -686,6 +712,16 @@ pub trait AbstractTree: sealed::Sealed {
     ///   or GC happens** — `major_compact(target, 0)` only restructures tables and
     ///   leaves a merge-only key's full operand chain intact.
     ///
+    /// The same watermark prunes the version history: every version older
+    /// than the newest one installed below `seqno_threshold` is released
+    /// (and the tables only those versions referenced become deletable).
+    /// Afterwards a snapshot at or below the oldest retained version's seqno
+    /// (see [`oldest_retained_seqno`](Self::oldest_retained_seqno)) can no
+    /// longer be read and fails with
+    /// [`Error::SnapshotBelowRetention`](crate::Error::SnapshotBelowRetention),
+    /// which is why the watermark must not exceed the oldest snapshot still in
+    /// use.
+    ///
     /// # Errors
     ///
     /// Will return `Err` if an IO error occurs.
@@ -946,6 +982,81 @@ pub trait AbstractTree: sealed::Sealed {
     /// Returns the highest sequence number that is flushed to disk.
     fn get_highest_persisted_seqno(&self) -> Option<SeqNo>;
 
+    /// Returns the seqno of the oldest version the history still retains:
+    /// the lower bound of the readable snapshot window.
+    ///
+    /// A read at snapshot `seqno` is served from the newest retained version
+    /// installed below it, so a snapshot is servable iff it is `0` (sees
+    /// nothing from any version) or strictly above this seqno; a read at
+    /// `0 < seqno <= oldest_retained_seqno()` fails with
+    /// [`Error::SnapshotBelowRetention`](crate::Error::SnapshotBelowRetention).
+    /// The boundary advances when [`major_compact`](Self::major_compact)
+    /// prunes the history up to its `seqno_threshold` and when
+    /// [`clear`](Self::clear) drains it, so a caller holding a long-lived
+    /// snapshot (a lagging consumer, a point-in-time query) can validate it
+    /// 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`, capped
+    /// at the compaction's own install seqno (the retained pre-compaction
+    /// version that served reads between the live front and `w` does not
+    /// survive a restart), after a `clear`, a table drop or a compaction
+    /// whose filter transformed rows 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
+    ///
+    /// ```
+    /// # let folder = tempfile::tempdir()?;
+    /// use lsm_tree::{AbstractTree, Config, Error, SequenceNumberCounter};
+    ///
+    /// let seqno = SequenceNumberCounter::default();
+    /// let tree = Config::new(folder, seqno.clone(), Default::default()).open()?;
+    /// assert_eq!(tree.oldest_retained_seqno(), 0);
+    ///
+    /// tree.insert("a", "v1", seqno.next());
+    /// tree.flush_active_memtable(0)?;
+    /// let stale = seqno.get();
+    /// tree.insert("a", "v2", seqno.next());
+    /// tree.flush_active_memtable(0)?;
+    ///
+    /// // A watermark above every live snapshot lets compaction prune the
+    /// // history; the boundary moves past the stale snapshot.
+    /// tree.major_compact(u64::MAX, seqno.get())?;
+    /// let oldest = tree.oldest_retained_seqno();
+    /// assert!(stale <= oldest);
+    /// assert!(matches!(
+    ///     tree.get("a", stale),
+    ///     Err(Error::SnapshotBelowRetention { .. })
+    /// ));
+    /// assert_eq!(tree.get("a", oldest + 1)?.as_deref(), Some(b"v2".as_slice()));
+    /// #
+    /// # Ok::<(), lsm_tree::Error>(())
+    /// ```
+    fn oldest_retained_seqno(&self) -> SeqNo;
+
+    /// Returns the PERSISTED retention boundary: the highest snapshot seqno
+    /// the tree will refuse after a reopen.
+    ///
+    /// [`oldest_retained_seqno`](Self::oldest_retained_seqno) is the LIVE
+    /// boundary, which the in-memory version history may hold below this
+    /// one: a table drop with no GC watermark keeps its pre-drop version
+    /// retained (and serving) until the restart, while the manifest already
+    /// records the drop's install seqno here. The two agree after a reopen.
+    ///
+    /// This is the value a deployment records for a later manifest repair
+    /// ([`Config::repair_retention_floor`](crate::Config::repair_retention_floor)):
+    /// it already folds in every operation that advanced the boundary (GC
+    /// compactions, `clear`, table drops, filtering compactions), so no
+    /// caller-side bookkeeping of watermarks is needed. `0` until the first
+    /// such install.
+    fn retention_floor(&self) -> SeqNo;
+
     /// Scans the entire tree, returning the number of items.
     ///
     /// ###### Caution
@@ -1127,7 +1238,17 @@ pub trait AbstractTree: sealed::Sealed {
     ///
     /// # Errors
     ///
-    /// Will return `Err` if an IO error occurs.
+    /// Will return `Err` if an IO error occurs, or
+    /// [`Error::SnapshotBelowRetention`](crate::Error::SnapshotBelowRetention)
+    /// when the history no longer retains a version for `seqno` (see
+    /// [`oldest_retained_seqno`](Self::oldest_retained_seqno)). The same applies
+    /// to every read that resolves a snapshot: [`get_pinned`](Self::get_pinned),
+    /// [`contains_key`](Self::contains_key), [`size_of`](Self::size_of),
+    /// [`multi_get`](Self::multi_get), [`len`](Self::len),
+    /// [`is_empty`](Self::is_empty), [`first_key_value`](Self::first_key_value),
+    /// [`last_key_value`](Self::last_key_value),
+    /// [`approximate_range_stats`](Self::approximate_range_stats) and
+    /// [`approximate_range_cardinality`](Self::approximate_range_cardinality).
     fn get<K: AsRef<[u8]>>(&self, key: K, seqno: SeqNo) -> crate::Result<Option<UserValue>>;
 
     /// Retrieves an item from the tree as a [`PinnableSlice`](crate::PinnableSlice).

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).