lsm_tree/

abstract_tree.rs

// SPDX-License-Identifier: Apache-2.0
// Copyright (c) 2024-present, fjall-rs
// Copyright (c) 2026-present, Dmitry Prudnikov

use crate::tree::inner::{FlushGuard, VersionsWriteGuard};
use crate::{
    Config, Guard, InternalValue, KvPair, Memtable, SeqNo, TableId, UserKey, UserValue,
    iter_guard::{IterGuardImpl, SeekableGuardIter},
    table::Table,
    version::Version,
    vlog::BlobFile,
};
use alloc::sync::Arc;
#[cfg(not(feature = "std"))]
use alloc::{boxed::Box, vec::Vec};
use core::ops::RangeBounds;

pub(crate) type RangeItem = crate::Result<KvPair>;

pub(crate) type FlushToTablesResult = (
    Vec<Table>,
    Option<Vec<BlobFile>>,
    crate::runtime_config::WritePin,
);

/// Summary of a checkpoint produced by
/// [`AbstractTree::create_checkpoint`].
///
/// All byte counts are *logical* file sizes — hard links share the
/// underlying inode storage, so a checkpoint's marginal disk usage is
/// typically zero until the original files are compacted away.
#[derive(Debug, Clone, Copy)]
pub struct CheckpointInfo {
    /// Number of SST files captured.
    pub sst_files: usize,
    /// Number of blob (value-log) files captured. Always `0` for a
    /// standard [`Tree`](crate::Tree).
    pub blob_files: usize,
    /// Sum of the logical file sizes of every file the checkpoint captured:
    /// the SSTs, the blob files, and the `.restrict-bound` sidecar that belongs
    /// to a tight-space-restricted SST (per-table state, written beside its
    /// table, that a restore needs to recover the exact bound). Tree-level
    /// metadata (the manifest snapshot and its edit log, the config, and the
    /// compression dictionaries under `dicts/`) is NOT counted: this figure
    /// describes the captured data, not the whole directory. A checkpoint of a
    /// dictionary-compressed tree therefore holds a little more on disk than
    /// this reports, by the size of the dictionaries it carries (order 100 KiB
    /// each, and shared by every table written against them rather than
    /// per-table like the data this counts).
    ///
    /// It measures the same SET of files as
    /// [`StorageStats::used_bytes`](crate::StorageStats::used_bytes), which
    /// reports them PHYSICALLY, so the two differ by exactly the bytes a
    /// tight-space reclaim punched out and agree everywhere else.
    pub total_bytes: u64,
    /// The version ID of the manifest snapshot the checkpoint installs.
    pub version_id: u64,
    /// Lower-bound visible-seqno watermark for the snapshot.
    ///
    /// Captured from the tree's `visible_seqno` generator BEFORE
    /// [`AbstractTree::current_version`]. Following the standard
    /// "lowest-excluded" watermark convention, `info.seqno = N` means
    /// every record with `seqno < N` was committed at sample time and
    /// is therefore guaranteed to be present in the snapshot. Records
    /// with `seqno == N` may or may not be included (writers can hold
    /// a record in the memtable for an instant before publishing the
    /// next watermark); records with `seqno > N` may also be present
    /// (writers can advance the counter between sample and version
    /// snapshot, and those keys still land in the captured memtable).
    ///
    /// PITR consumers MUST use `seqno < info.seqno` as the inclusion
    /// gate. Using `<=` (treating this as a max-included ceiling)
    /// could move a recovery cutoff past data still needed from WAL
    /// or replication; the field is a strict lower-exclusive watermark,
    /// not a max-included ceiling.
    pub seqno: SeqNo,
}

// Sealed on purpose: this trait is still public as a consumer-side bound
// (`&impl AbstractTree`), but external implementations are no longer part of
// the supported extension surface. Internal flush/version hooks keep evolving
// with crate-owned tree types and must not create downstream semver traps.
//
// `sealed` is crate-visible so sibling modules in this crate can write
// `crate::abstract_tree::sealed::Sealed` in their impls; downstream crates
// cannot name or implement this trait.
pub(crate) mod sealed {
    pub trait Sealed {}
}

/// Outcome of [`AbstractTree::refresh_table_checksum`].
#[cfg(feature = "std")]
#[doc(hidden)]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ChecksumRefreshOutcome {
    /// The digest was installed into the manifest.
    Refreshed,
    /// Legitimate no-op: the table was compacted away, or its restriction no
    /// longer matches the one the digest was computed for; the current view
    /// carries its own (compaction-installed) digest for the next patrol to
    /// reconcile.
    Stale,
    /// The install lock was held by a concurrent compaction (blocking on it
    /// would invert the heal-lock / compaction-state order and deadlock). The
    /// manifest digest stays stale against durable healed bytes: the caller
    /// must surface this as a finding, keep the attestation, and let a later
    /// patrol retry.
    Contended,
}

/// Generic Tree API
pub trait AbstractTree: sealed::Sealed {
    /// Debug method for tracing the MVCC history of a key.
    #[doc(hidden)]
    fn print_trace(&self, key: &[u8]) -> crate::Result<()>;

    /// Returns the number of cached table file descriptors.
    fn table_file_cache_size(&self) -> usize;

    #[doc(hidden)]
    fn next_table_id(&self) -> TableId;

    #[doc(hidden)]
    fn id(&self) -> crate::TreeId;

    /// Like [`AbstractTree::get`], but returns the actual internal entry, not just the user value.
    ///
    /// Used in tests.
    #[doc(hidden)]
    fn get_internal_entry(&self, key: &[u8], seqno: SeqNo) -> crate::Result<Option<InternalValue>>;

    #[doc(hidden)]
    fn current_version(&self) -> Version;

    /// Records that the table with `table_id` now has the digest `checksum`,
    /// persisting it to the manifest through a version upgrade.
    ///
    /// Called after an in-place heal rewrites a table's bytes: the healed
    /// file's digest no longer matches the one captured at recovery, and
    /// without the refresh a later [`verify`](crate::verify) pass (or a
    /// repair) would flag the healed file as corrupted against the stale
    /// manifest digest. A no-op when the table is no longer part of the
    /// current version (compacted away while the heal ran — the old file is
    /// on its way out).
    ///
    /// `expected_restriction` is the tight-space restriction bound the CALLER
    /// computed `checksum` for (its captured view's [`restrict_lower_bound`]).
    /// This method holds the install lock, so it re-checks it against the CURRENT
    /// view under that lock and refuses the install (a no-op) if a compaction
    /// swapped the view to a different restriction meanwhile: a whole-file digest
    /// installed into a suffix-only restricted manifest (or vice versa) could
    /// never match the punched file.
    ///
    /// Returns [`ChecksumRefreshOutcome::Refreshed`] when the digest was
    /// installed, [`ChecksumRefreshOutcome::Stale`] on a legitimate no-op (the
    /// table was compacted away, or its restriction no longer matches
    /// `expected_restriction` — the current view carries its own digest), and
    /// [`ChecksumRefreshOutcome::Contended`] when the install lock was held by
    /// a concurrent compaction. The caller uses this to decide whether to clear
    /// the heal attestation (only `Refreshed` may) and whether the pass stayed
    /// clean: a contended skip leaves the manifest digest stale against durable
    /// healed bytes, which must surface as a finding rather than a clean
    /// report.
    ///
    /// [`restrict_lower_bound`]: crate::table::Table::restrict_lower_bound
    #[cfg(feature = "std")]
    #[doc(hidden)]
    fn refresh_table_checksum(
        &self,
        table_id: TableId,
        checksum: crate::checksum::Checksum,
        expected_restriction: Option<&crate::UserKey>,
    ) -> crate::Result<ChecksumRefreshOutcome>;

    /// The tree's configured durability mode
    /// ([`Config::sync_mode`](crate::config::Config::sync_mode)). Maintenance
    /// paths that write outside the flush pipeline (the in-place heal) read
    /// it here so their syncs honor the same durability the tree's own
    /// writes use.
    #[doc(hidden)]
    fn sync_mode(&self) -> crate::fs::SyncMode;

    /// Waits for the table file pools to finish the refills running off the
    /// writing threads: a test that counts the syncs an operation causes
    /// waits for them before it counts.
    #[doc(hidden)]
    fn wait_for_table_file_refills(&self);

    /// The tree's configured prefix extractor, or `None` when it indexes no
    /// prefixes. The patrol scrub's filter cross-check reads it here so it
    /// can verify a rebuilt full filter carries the source's prefix hashes,
    /// not just its complete-key hashes.
    #[doc(hidden)]
    fn prefix_extractor(&self) -> Option<alloc::sync::Arc<dyn crate::prefix::PrefixExtractor>>;

    /// Returns a read-only snapshot of the tree's on-disk storage footprint:
    /// total used bytes, entry count, the average shape of a stored entry
    /// (average key / value bytes), and an estimate of how many more
    /// average-shaped entries fit in a byte budget (see
    /// [`StorageStats::estimated_remaining_entries`](crate::StorageStats::estimated_remaining_entries)).
    ///
    /// Computed from the live version's table + blob metadata plus one
    /// size-stat per live file; it never reads a data block. The default
    /// implementation reports [`StorageStatus::Healthy`](crate::StorageStatus::Healthy);
    /// the standard tree overrides it to report
    /// [`StorageStatus::CompactionInProgress`](crate::StorageStatus::CompactionInProgress) while a
    /// compaction runs.
    ///
    /// # Examples
    ///
    /// ```
    /// # use lsm_tree::Error as TreeError;
    /// use lsm_tree::{AbstractTree, Config};
    ///
    /// let folder = tempfile::tempdir()?;
    /// let tree = Config::new(&folder, Default::default(), Default::default()).open()?;
    ///
    /// for i in 0..100u64 {
    ///     tree.insert(format!("key{i:04}"), "value", i);
    /// }
    /// tree.flush_active_memtable(0)?;
    ///
    /// let stats = tree.storage_stats()?;
    /// assert_eq!(stats.item_count, 100);
    /// // Roughly how many more average-shaped entries fit in another 1 MiB.
    /// let _headroom = stats.estimated_remaining_entries(1024 * 1024);
    /// #
    /// # Ok::<(), TreeError>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns an error if a live file's size cannot be stat-ed, or a table's
    /// blob-link section cannot be read or parsed.
    fn storage_stats(&self) -> crate::Result<crate::StorageStats> {
        crate::storage_stats::compute_storage_stats(&self.current_version(), false, true)
    }

    /// Per-LSM-level and per-segment size + entry-count stats, for tiering and
    /// erasure-coding placement decisions (which level / segment is large enough
    /// to demote, EC-encode, or migrate).
    ///
    /// Cheap: derived from the live version's metadata plus one file-size stat
    /// per segment (no data-block scan); in a tree that separates values, each
    /// segment's blob-link section is also read once, then kept. The per-level
    /// totals reconcile with
    /// [`storage_stats`](Self::storage_stats): summed across levels they equal
    /// the SST portion of [`StorageStats::used_bytes`](crate::StorageStats::used_bytes)
    /// and [`StorageStats::item_count`](crate::StorageStats::item_count).
    ///
    /// # Examples
    ///
    /// ```
    /// # use lsm_tree::Error as TreeError;
    /// use lsm_tree::{AbstractTree, Config};
    ///
    /// let folder = tempfile::tempdir()?;
    /// let tree = Config::new(&folder, Default::default(), Default::default()).open()?;
    /// for i in 0..100u32 {
    ///     tree.insert(format!("k{i:04}"), "v", 0);
    /// }
    /// tree.flush_active_memtable(0)?;
    ///
    /// let levels = tree.level_segment_stats()?;
    /// let total: u64 = levels.iter().map(|l| l.item_count).sum();
    /// assert_eq!(total, tree.storage_stats()?.item_count);
    /// #
    /// # Ok::<(), TreeError>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns an error if a segment's file size cannot be stat-ed, or its
    /// blob-link section cannot be read or parsed.
    fn level_segment_stats(&self) -> crate::Result<Vec<crate::LevelStats>> {
        crate::storage_stats::compute_level_segment_stats(&self.current_version())
    }

    /// Estimated bytes pending compaction under `strategy`: on-disk data above
    /// its level's target that must eventually be rewritten downward (a `RocksDB`
    /// `estimate-pending-compaction-bytes` analog), a compaction-debt signal for a
    /// scheduler / tiering consumer.
    ///
    /// The strategy is supplied by the caller because the engine does not own a
    /// configured compaction strategy (it is injected per compaction run); a
    /// `&dyn` keeps this object-safe. Returns `0` for strategies without a
    /// size-target notion of debt (FIFO), or when the tree is at or
    /// below its target shape. See
    /// [`CompactionStrategy::pending_compaction_bytes`](crate::compaction::CompactionStrategy::pending_compaction_bytes).
    fn compaction_debt(&self, strategy: &dyn crate::compaction::CompactionStrategy) -> u64 {
        strategy.pending_compaction_bytes(&self.current_version())
    }

    /// Computed write-backpressure verdict from the live L0 run count, the
    /// strategy's pending-compaction bytes and the unflushed memtable bytes,
    /// against the configured
    /// [`RuntimeConfig::backpressure`](crate::runtime_config::RuntimeConfig)
    /// thresholds.
    ///
    /// Advisory, mirroring [`write_admission`](Self::write_admission): the caller
    /// consults it and throttles in its own write loop (sleep `suggested_delay`
    /// at [`Slowdown`](crate::Backpressure::Slowdown), pause at
    /// [`Stop`](crate::Backpressure::Stop)). The engine never blocks on it,
    /// because it does not own the compaction that drains the debt — an internal
    /// stall could deadlock the very thread that would compact.
    ///
    /// The strategy is supplied by the caller for the same reason as
    /// [`compaction_debt`](Self::compaction_debt). Returns
    /// [`Backpressure::None`](crate::Backpressure::None) by default (no
    /// thresholds configured).
    fn write_backpressure(
        &self,
        _strategy: &dyn crate::compaction::CompactionStrategy,
    ) -> crate::Backpressure {
        crate::Backpressure::None
    }

    /// The verdict for one write of `bytes`: [`write_backpressure`](Self::write_backpressure),
    /// except that a slowdown on a tree with a write rate limit (see
    /// [`Config::write_rate_limit`](crate::Config::write_rate_limit)) debits
    /// `bytes` from the tree's write limiter and suggests waiting until they,
    /// and every byte debited before them, are paid off at its rate. Writes
    /// honouring it are then bounded by the rate in bytes per second, large
    /// and small alike, however many writers share the limiter. Outside the
    /// slowdown tier nothing is debited.
    ///
    /// # Examples
    ///
    /// ```
    /// use lsm_tree::{AbstractTree, Backpressure, Config, compaction::Leveled};
    ///
    /// let folder = tempfile::tempdir()?;
    /// let tree = Config::new(&folder, Default::default(), Default::default())
    ///     .write_rate_limit(1_024 * 1_024)
    ///     .open()?;
    /// // No threshold set: full speed, nothing debited.
    /// assert_eq!(tree.write_backpressure_for(&Leveled::default(), 4_096), Backpressure::None);
    /// # Ok::<(), lsm_tree::Error>(())
    /// ```
    fn write_backpressure_for(
        &self,
        strategy: &dyn crate::compaction::CompactionStrategy,
        bytes: u64,
    ) -> crate::Backpressure {
        let verdict = self.write_backpressure(strategy);
        if !matches!(verdict, crate::Backpressure::Slowdown { .. }) {
            return verdict;
        }
        let limiter = self.write_rate_limiter();
        if limiter.rate() == 0 {
            return verdict;
        }
        #[cfg(feature = "std")]
        {
            crate::Backpressure::Slowdown {
                suggested_delay: limiter.reserve(bytes),
            }
        }
        // no-std: acquire_wait with a caller-provided monotonic clock; without
        // one there is no rate to pace against and the slowdown keeps its ramp.
        #[cfg(not(feature = "std"))]
        {
            let _ = bytes;
            verdict
        }
    }

    /// The limiter writes are paced with under backpressure slowdown: shared
    /// with other trees if one was supplied through
    /// [`Config::write_rate_limiter`](crate::Config::write_rate_limiter), else
    /// the tree's own. Retune it live with
    /// [`RateLimiter::set_rate`](crate::rate_limiter::RateLimiter::set_rate);
    /// `0` turns byte pacing off.
    fn write_rate_limiter(&self) -> alloc::sync::Arc<crate::rate_limiter::RateLimiter>;

    /// Storage admission gate: `Ok(())` if a write may proceed, or
    /// [`Error::StorageFull`](crate::Error::StorageFull) if the tree is
    /// over budget and should be treated as read-only.
    ///
    /// Opt-in: returns `Ok(())` unless
    /// [`storage_admission_check`](crate::runtime_config::RuntimeConfig::storage_admission_check)
    /// is enabled. The predicate is computed (not latched), so once space is
    /// freed — the budget raised, a compaction reclaiming space, or disk freed —
    /// the next call admits again with no restart.
    ///
    /// Intended as a cheap pre-check the caller consults before applying a
    /// write batch. The footprint is cached per version, so the check is a
    /// constant-time read on the common path and only re-measures when a new
    /// version is installed (flush / compaction). Internal flush / compaction
    /// are never gated, so the engine can always reclaim.
    ///
    /// # Errors
    ///
    /// [`Error::StorageFull`](crate::Error::StorageFull) when the live footprint
    /// plus reserved headroom exceeds the effective budget.
    fn write_admission(&self) -> crate::Result<()> {
        Ok(())
    }

    /// `true` when the admission gate is currently closed (see
    /// [`write_admission`](Self::write_admission)). Convenience for callers that
    /// want a boolean rather than a `Result`. Always `false` unless admission
    /// control is enabled and the tree is over budget.
    ///
    /// Only [`Error::StorageFull`](crate::Error::StorageFull) counts as
    /// read-only: an unrelated admission error (e.g. an I/O failure while
    /// measuring the footprint) is NOT an out-of-space condition and must not be
    /// reported as one.
    fn is_read_only(&self) -> bool {
        matches!(
            self.write_admission(),
            Err(crate::Error::StorageFull { .. })
        )
    }

    /// Proactively verifies every block's XXH3 checksum across every SST in
    /// the tree's current version — a scrubber for catching bit rot before it
    /// surfaces as a user-visible read failure (cron / scrub jobs).
    ///
    /// Reports at block granularity and never aborts early. The returned
    /// [`BlockVerifyReport`](crate::verify::BlockVerifyReport) records
    /// block-corruption findings with `(file, offset)`, while file-level errors
    /// (e.g. [`BlockVerifyError::SstFileUnreadable`](crate::verify::BlockVerifyError::SstFileUnreadable))
    /// carry the file only (no offset). It does not surface per-entry indices or
    /// ECC-correction counts (when ECC-at-rest is enabled a within-budget corrupt
    /// block may still be healed on read as a side effect of the scan, but the
    /// report does not tally corrections).
    ///
    /// Filesystems with native per-block integrity (ZFS, Btrfs, `ReFS`, S3 —
    /// see [`Fs::capabilities`](crate::fs::Fs::capabilities)) already detect
    /// corruption on read; this scrub is the portable check for the rest.
    ///
    /// Use [`Self::verify_checksum_with`] for parallelism / throttle control.
    #[cfg(feature = "std")]
    fn verify_checksum(&self) -> crate::verify::BlockVerifyReport
    where
        Self: Sized,
    {
        crate::verify::verify_block_checksums(self)
    }

    /// Like [`Self::verify_checksum`] but with configurable parallelism and
    /// I/O throttle (see [`VerifyOptions`](crate::verify::VerifyOptions)).
    #[cfg(feature = "std")]
    fn verify_checksum_with(
        &self,
        options: &crate::verify::VerifyOptions,
    ) -> crate::verify::BlockVerifyReport
    where
        Self: Sized,
    {
        crate::verify::verify_block_checksums_with(self, options)
    }

    #[doc(hidden)]
    fn get_version_history_lock(&self) -> VersionsWriteGuard<'_>;

    /// Creates a hard-linked checkpoint of the tree's on-disk state in
    /// `target_path` for point-in-time recovery (PITR) backup.
    ///
    /// The checkpoint is a fully functional tree that can be opened
    /// independently via [`Config::open`](crate::Config::open). For the
    /// common single-filesystem case all SST files (and blob files, for
    /// [`BlobTree`](crate::BlobTree)) are hard-linked rather than copied, so the operation
    /// is O(1) per file and consumes zero additional disk space until the
    /// original files are compacted away — at which point the inode is
    /// kept alive by the checkpoint link.
    ///
    /// # Cross-filesystem / cross-backend fall-back
    ///
    /// When a source file lives on a different filesystem than the
    /// checkpoint target — e.g. an SST routed to a hot tier via
    /// [`level_routes`](crate::Config::level_routes) on a separate volume,
    /// or a backup directory on a foreign mount — the hard link cannot
    /// be created (Unix `EXDEV`). In that case the checkpoint silently
    /// falls back to a streamed byte copy, which:
    ///
    /// - takes time linear in the file size instead of O(1), and
    /// - consumes disk space equal to the copied bytes on the target
    ///   volume (no inode sharing across filesystems).
    ///
    /// Each fall-back call emits one [`log::debug`] line (deliberately not
    /// `warn`: a misconfigured tier could trigger this path once per SST
    /// and per blob — thousands of times per snapshot — and per-file
    /// warnings would drown real signal). Operators wanting hard-visibility
    /// of unexpected full copies should enable debug logging on the `fs`
    /// module or watch the `CheckpointInfo.total_bytes` figure (≫ inode
    /// link cost means the fallback fired). The same `debug` policy applies
    /// when source and target use entirely different [`Fs`](crate::fs::Fs)
    /// backends (e.g. [`MemFs`](crate::fs::MemFs) → [`StdFs`](crate::fs::StdFs)
    /// in tests).
    ///
    /// # Durability
    ///
    /// `Ok` means the checkpoint's files and directories are persisted under
    /// [`Config::sync_mode`](crate::Config::sync_mode): durable under
    /// [`SyncMode::Full`](crate::fs::SyncMode::Full); under
    /// [`SyncMode::Barrier`](crate::fs::SyncMode::Barrier) synced in order, one
    /// barrier per file, and durable once a later
    /// [`Fs::sync_device`](crate::fs::Fs::sync_device) of `target_path` returns
    /// `Ok`.
    ///
    /// # Concurrency
    ///
    /// While the checkpoint is being built, compaction continues normally
    /// but the physical removal of obsolete files is deferred until the
    /// checkpoint hard-links are in place. This is implemented by an
    /// internal reference-counted deletion gate; callers do not have to
    /// pause compaction themselves.
    ///
    /// # Errors
    ///
    /// Returns an error if:
    /// - the active memtable could not be flushed,
    /// - `target_path` already exists (to prevent accidental overwrites),
    /// - a hard link / copy fall-back could not be created, or
    /// - the manifest files could not be replicated.
    ///
    /// On error any partial checkpoint files are removed automatically
    /// (best-effort) so callers can safely retry against the same path.
    // std-only: checkpoint creation hard-links / copies files via std::fs.
    #[cfg(feature = "std")]
    fn create_checkpoint(&self, target_path: &crate::path::Path) -> crate::Result<CheckpointInfo>;

    /// Pins the tree's current state for a point-in-time copy that costs no
    /// file I/O: seals the active memtable and returns a
    /// [`PinnedVersion`](crate::PinnedVersion) holding the sealed memtables,
    /// tables, blob files and dictionaries as they are now.
    ///
    /// Nothing is copied, linked, synced or written. The pinned files stay on
    /// disk with unchanged bytes while flushes and compactions supersede them,
    /// until the last pin listing them drops; a consumer streams them byte for
    /// byte and rebuilds the tree elsewhere with
    /// [`Config::restore_pinned`](crate::Config::restore_pinned). A restart
    /// drops every pin.
    ///
    /// Waits for an in-place heal or a tight-space hole punch already under
    /// way to finish; new ones on a pinned file wait for the pin.
    ///
    /// # Examples
    ///
    /// ```
    /// use lsm_tree::{AbstractTree, Config, SequenceNumberCounter};
    ///
    /// let folder = tempfile::tempdir()?;
    /// let seqno = SequenceNumberCounter::default();
    /// let visible = SequenceNumberCounter::default();
    /// let tree = Config::new(&folder, seqno.clone(), visible.clone()).open()?;
    /// tree.insert("a", "1", seqno.next());
    /// tree.flush_active_memtable(0)?;
    /// tree.insert("b", "2", seqno.next());
    /// // The writes are published: the pin holds every write below this.
    /// visible.set(seqno.get());
    ///
    /// let pin = tree.pin();
    /// assert_eq!(pin.tables().len(), 1);
    /// assert_eq!(pin.memtable_entries().count(), 1);
    /// # Ok::<(), lsm_tree::Error>(())
    /// ```
    #[cfg(feature = "std")]
    fn pin(&self) -> crate::PinnedVersion;

    /// Stops the tree owing the repair report with this
    /// [`RepairReport::id`](crate::RepairReport::id), installing a version
    /// that no longer carries it: no later
    /// [`Config::open_or_repair`](crate::Config::open_or_repair) hands it out.
    ///
    /// A repair's report stays in the tree until this call, so a process that
    /// ends before the caller acted on it gets it again. Call it once the
    /// report's replay obligation is met. Returns `false`, changing nothing,
    /// when the tree owes no report or owes one with another id.
    ///
    /// # Errors
    ///
    /// Propagates the install's failure; the report is then still owed, or,
    /// on [`Error::InstallOutcomeUnknown`](crate::Error::InstallOutcomeUnknown),
    /// may be after a power loss.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use lsm_tree::{AbstractTree, Config, RepairPolicy, SequenceNumberCounter};
    ///
    /// let config = Config::new(
    ///     "/var/lib/mydb",
    ///     SequenceNumberCounter::default(),
    ///     SequenceNumberCounter::default(),
    /// );
    /// let (tree, repaired) = config.open_or_repair(RepairPolicy::default())?;
    /// if let Some(report) = repaired {
    ///     // Replay the external log as `report.wal_replay_scope()` requires,
    ///     // then let the tree forget the report.
    ///     tree.acknowledge_repair_report(report.id)?;
    /// }
    /// # Ok::<_, lsm_tree::Error>(())
    /// ```
    #[cfg(feature = "std")]
    fn acknowledge_repair_report(&self, id: u128) -> crate::Result<bool>;

    /// Seals the active memtable and flushes it, with every memtable sealed
    /// before it, to tables: the durability barrier for a caller that keeps
    /// its own journal of the writes.
    ///
    /// `Ok` means the tables, their directories and the manifest edits that
    /// name them are persisted under [`Config::sync_mode`](crate::Config::sync_mode)
    /// for the memtable active at the call and for every memtable sealed before
    /// it, in sealing order, including one another thread began flushing: the
    /// call waits for that flush and, should it fail, flushes the memtable
    /// itself. A caller may release the journal entries those memtables hold
    /// once the call returns `Ok`; under
    /// [`SyncMode::Barrier`](crate::fs::SyncMode::Barrier) the writes are only
    /// ordered then, and the entries may go once
    /// [`sync_devices`](Self::sync_devices) has returned `Ok` too.
    ///
    /// `gc_watermark` has the meaning documented at
    /// [`major_compact`](Self::major_compact); pass `0` to collect nothing.
    ///
    /// # Errors
    ///
    /// Returns the error of [`flush`](Self::flush); the memtables it could not
    /// persist stay sealed, and the next call flushes them.
    ///
    /// # Examples
    ///
    /// ```
    /// use lsm_tree::{AbstractTree, Config, SequenceNumberCounter};
    ///
    /// let folder = tempfile::tempdir()?;
    /// let tree = Config::new(
    ///     &folder,
    ///     SequenceNumberCounter::default(),
    ///     SequenceNumberCounter::default(),
    /// )
    /// .open()?;
    /// tree.insert("key", "value", 0);
    /// // The write is on disk now (the default sync mode is not `Barrier`): a
    /// // journal that holds it can drop it.
    /// tree.flush_active_memtable(0)?;
    /// assert_eq!(tree.table_count(), 1);
    /// # Ok::<(), lsm_tree::Error>(())
    /// ```
    fn flush_active_memtable(&self, gc_watermark: SeqNo) -> crate::Result<()> {
        let lock = self.get_flush_lock();
        self.rotate_memtable();
        self.flush(&lock, gc_watermark)?;
        Ok(())
    }

    /// Makes every flush installed before this call durable, under
    /// [`FlushDurability::Deferred`](crate::config::FlushDurability::Deferred):
    /// the tables and blob files those flushes wrote are synced, then their
    /// folders, then the manifest is written. Once it returns `Ok` a caller
    /// may drop its own log up to the writes those flushes held; under
    /// [`SyncMode::Barrier`](crate::fs::SyncMode::Barrier) after
    /// [`sync_devices`](Self::sync_devices) as well. A no-op when nothing is
    /// pending, and under [`FlushDurability::Synced`](crate::config::FlushDurability::Synced)
    /// every flush is durable when it returns.
    ///
    /// Every other install (a compaction, an ingestion, a drop) persists the
    /// pending flushes with it, and so do a checkpoint and a dropped tree.
    ///
    /// # Examples
    ///
    /// ```
    /// # let folder = tempfile::tempdir()?;
    /// use lsm_tree::{AbstractTree, Config, SequenceNumberCounter, config::FlushDurability};
    ///
    /// let tree = Config::new(
    ///     folder.path(),
    ///     SequenceNumberCounter::default(),
    ///     SequenceNumberCounter::default(),
    /// )
    /// .flush_durability(FlushDurability::Deferred)
    /// .open()?;
    /// for i in 0..3_u64 {
    ///     tree.insert(i.to_be_bytes(), "v", i);
    ///     tree.flush_active_memtable(0)?;
    /// }
    /// // One barrier for the three flushes.
    /// tree.persist_flushes()?;
    /// # Ok::<(), lsm_tree::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns a sync or manifest write error; the flushes stay installed and
    /// readable, and the next persist retries them.
    fn persist_flushes(&self) -> crate::Result<()>;

    /// Makes durable everything this tree synced on each device it lies on:
    /// the tree folder's, that of every level's tables, that of the
    /// dictionaries folder and, with KV separation, that of the blobs folder.
    ///
    /// Under [`SyncMode::Barrier`](crate::fs::SyncMode::Barrier) a
    /// [`flush_active_memtable`](Self::flush_active_memtable) that returns `Ok`
    /// has its writes ordered, and they are durable once this returns `Ok`; a
    /// caller releases its journal after both. Under any other mode the flush
    /// alone is durable and this only flushes the device caches.
    ///
    /// # Errors
    ///
    /// Returns the error of a device flush that failed; the devices not yet
    /// flushed are not.
    fn sync_devices(&self) -> crate::Result<()> {
        let config = self.tree_config();
        let mut flushed: Vec<(Arc<dyn crate::fs::Fs>, crate::path::PathBuf)> = Vec::new();
        let levels = (0..config.level_count).map(|level| {
            let (folder, fs) = config.tables_folder_for_level(level);
            (fs, folder)
        });
        // The blobs folder may be a mount of its own under the tree folder.
        let blobs = config.kv_separation_opts.is_some().then(|| {
            (
                Arc::clone(&config.fs),
                config.path.join(crate::file::BLOBS_FOLDER),
            )
        });
        // So may the dictionaries folder, which the tables written against a
        // dictionary need to be read.
        #[cfg(zstd_any)]
        let dicts = Some((
            Arc::clone(&config.fs),
            config.path.join(crate::file::DICTS_FOLDER),
        ));
        #[cfg(not(zstd_any))]
        let dicts: Option<(Arc<dyn crate::fs::Fs>, crate::path::PathBuf)> = None;
        for (fs, path) in core::iter::once((Arc::clone(&config.fs), config.path.clone()))
            .chain(levels)
            .chain(blobs)
            .chain(dicts)
        {
            // One flush per device: a folder on a device already flushed
            // gains nothing from another.
            let volume = fs.volume_id(&path);
            if flushed.iter().any(|(done, at)| {
                (Arc::ptr_eq(done, &fs) && *at == path)
                    || volume.is_some_and(|volume| done.volume_id(at) == Some(volume))
            }) {
                continue;
            }
            match fs.sync_device(&path) {
                Ok(()) => flushed.push((fs, path)),
                // A folder never made holds nothing to flush.
                Err(e) if e.kind() == crate::io::ErrorKind::NotFound => {}
                Err(e) => return Err(e.into()),
            }
        }
        Ok(())
    }

    /// Synchronously flushes pending sealed memtables to tables.
    ///
    /// Returns the sum of flushed memtable sizes that were flushed.
    ///
    /// The function may not return a result, if nothing was flushed.
    ///
    /// `gc_watermark` has the meaning documented at
    /// [`major_compact`](Self::major_compact): the flush folds away versions
    /// below it exactly as a compaction does, and when it does, the install
    /// raises the retention floor ([`retention_floor`](Self::retention_floor))
    /// to match. Pass `0` to collect nothing.
    ///
    /// # Errors
    ///
    /// Returns `Err` on an I/O error, or on a memtable residence-verification
    /// failure under [`KvChecksumComputePoint::AtInsert`](crate::runtime_config::KvChecksumComputePoint::AtInsert):
    /// a sealed memtable whose insert-time per-KV digests do not verify before
    /// flush surfaces [`crate::Error::MemtableKvChecksumMismatch`],
    /// [`crate::Error::MemtableKvChecksumCorruptAlgorithm`], or
    /// [`crate::Error::InvalidTag`] (corrupt `value_type`).
    fn flush(&self, _lock: &FlushGuard<'_>, gc_watermark: SeqNo) -> crate::Result<Option<u64>> {
        use crate::{
            compaction::stream::CompactionStream, merge::Merger, range_tombstone::RangeTombstone,
        };

        let version_history = self.get_version_history_lock();
        let latest = version_history.latest_version();

        if latest.sealed_memtables.len() == 0 {
            return Ok(None);
        }

        let sealed_ids = latest
            .sealed_memtables
            .iter()
            .map(|mt| mt.id)
            .collect::<Vec<_>>();

        let flushed_size = latest.sealed_memtables.iter().map(|mt| mt.size()).sum();
        // The keys the flush's filters cover and the hashes they hold, which
        // the filter plan keeps room for: counted only when a filter budget
        // plans them, over the merge the flush writes, so a key in several
        // memtables counts once.
        let flushed = if self.tree_config().filter_advisor.is_some() {
            // A partitioned filter holds no prefix hashes.
            let prefixes = if self.tree_config().filter_block_partitioning_policy.get(0) {
                None
            } else {
                self.tree_config().prefix_extractor.as_deref()
            };
            let merged = Merger::new(
                latest
                    .sealed_memtables
                    .iter()
                    .map(|mt| mt.iter().map(Ok))
                    .collect::<Vec<_>>(),
                self.tree_config().comparator.clone(),
            );
            // Memtable entries read without I/O: the merge yields no error.
            crate::memtable::filter_count(
                merged.filter_map(|item| item.ok().map(|item| item.key.user_key)),
                prefixes,
            )
        } else {
            crate::filter_budget::FilterCount::default()
        };

        // AtInsert residence check: verify each sealed memtable's insert-time
        // per-KV digests against a recompute over the entries' current bytes
        // before writing them out. A divergence means an entry was corrupted
        // (a RAM bit-flip) while it sat in the memtable. Memtables with no
        // insert digests (the default) return immediately without walking.
        //
        // This is a SEPARATE pass over the as-inserted memtable entries, not
        // fused into the writer's per-KV footer encode, and deliberately so:
        // the footer digest is computed over POST-merge / post-seqno-filter
        // bytes (the CompactionStream below applies the merge operator), so a
        // merge operator's combined value differs from any single inserted
        // value. Comparing the carried insert digest against the footer digest
        // would false-positive on every legitimate merge. Residence corruption
        // is a property of what was inserted (pre-merge); it must be checked
        // here, against the raw memtable entries.
        for mt in latest.sealed_memtables.iter() {
            mt.verify_kv_residence()?;
        }

        // Collect range tombstones from sealed memtables
        let mut range_tombstones: Vec<RangeTombstone> = Vec::new();
        for mt in latest.sealed_memtables.iter() {
            range_tombstones.extend(mt.range_tombstones_sorted());
        }
        range_tombstones
            .sort_by(|a, b| a.cmp_with_comparator(b, self.tree_config().comparator.as_ref()));
        range_tombstones.dedup();

        let merger = Merger::new(
            latest
                .sealed_memtables
                .iter()
                .map(|mt| mt.iter().map(Ok))
                .collect::<Vec<_>>(),
            self.tree_config().comparator.clone(),
        );
        // Observed under the stream, which drops the versions that show a key
        // was overwritten; only a blob tree that groups values by lifetime
        // asks.
        let overwritten = crate::blob_tree::overwritten::OverwrittenKeys::default();
        let marks = self
            .tree_config()
            .kv_separation_opts
            .as_ref()
            .is_some_and(|kv| kv.lifetime_classifier().is_grouping())
            .then_some(&overwritten);
        let merger = crate::blob_tree::overwritten::MarkOverwritten::new(merger, marks);
        // RT suppression is not needed here: flush writes both entries and RTs
        // to the output tables. Suppression happens at read time, not write time.
        let stream = CompactionStream::new(merger, gc_watermark)
            .with_merge_operator(self.tree_config().merge_operator.clone());
        // Versions that go in and do not come out, so the install can tell a
        // flush that collected history from one that collected none and only
        // raise the retention floor for the former. Read once the stream is
        // drained, below.
        let gc_balance = stream.gc_balance();

        drop(version_history);

        let has_range_tombstones = !range_tombstones.is_empty();
        if let Some((tables, blob_files, write_pin)) = self.flush_to_tables_with_rt(
            stream,
            range_tombstones,
            flushed.keys,
            flushed.hashes,
            marks,
        )? {
            // A writer given range tombstones and no key writes a table around
            // them, so a flush that acknowledges a deletion has put it on disk.
            debug_assert!(
                !tables.is_empty() || !has_range_tombstones,
                "a flush of range tombstones wrote no table"
            );

            #[cfg(all(test, feature = "std"))]
            self.tree_config().fire_before_output_install();

            self.register_tables(
                &tables,
                blob_files.as_deref(),
                None,
                &sealed_ids,
                gc_watermark,
                gc_balance.load(core::sync::atomic::Ordering::Relaxed) > 0,
            )?;
            // The installed files name their dictionaries from here on.
            drop(write_pin);
        }

        Ok(Some(flushed_size))
    }

    /// 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 at or below the retention floor (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,
        index: Option<(Arc<Memtable>, SeqNo)>,
    ) -> Box<dyn DoubleEndedIterator<Item = IterGuardImpl> + Send + 'static> {
        self.range::<&[u8], _>(.., seqno, index)
    }

    /// 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 at or below the retention floor (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,
        seqno: SeqNo,
        index: Option<(Arc<Memtable>, SeqNo)>,
    ) -> Box<dyn DoubleEndedIterator<Item = IterGuardImpl> + Send + 'static>;

    /// 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 at or below the retention floor (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,
        seqno: SeqNo,
        index: Option<(Arc<Memtable>, SeqNo)>,
    ) -> Box<dyn DoubleEndedIterator<Item = IterGuardImpl> + Send + 'static>;

    /// Returns a range iterator that can reposition (seek) in place without
    /// reopening its per-SST readers.
    ///
    /// Unlike [`range`](Self::range)'s boxed `DoubleEndedIterator`, the returned
    /// [`SeekableGuardIter`] additionally exposes `seek_to` / `seek_to_for_prev`,
    /// 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 at or below the retention floor (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,
        seqno: SeqNo,
        index: Option<(Arc<Memtable>, SeqNo)>,
    ) -> Box<dyn SeekableGuardIter + 'static>;

    /// Scans a sequence of disjoint, ascending key sub-intervals, reusing one set
    /// of per-SST readers across all of them.
    ///
    /// The per-SST setup is paid once (when the underlying seekable iterator is
    /// opened); each interval is served by repositioning that iterator. This
    /// amortizes the setup across `N` intervals instead of paying it per
    /// interval, so multi-interval scan throughput scales with the total rows
    /// returned rather than the interval count.
    ///
    /// The interval source is pulled lazily, so intervals may be produced on
    /// demand (e.g. computed from rows already returned).
    ///
    /// A snapshot at or below the retention floor (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,
        seqno: SeqNo,
        index: Option<(Arc<Memtable>, SeqNo)>,
    ) -> Box<dyn Iterator<Item = IterGuardImpl> + Send + 'static>
    where
        I::IntoIter: Send + 'static;

    /// Returns the approximate number of tombstones in the tree.
    fn tombstone_count(&self) -> u64;

    /// Returns the approximate number of weak tombstones (single deletes) in the tree.
    fn weak_tombstone_count(&self) -> u64;

    /// Returns the approximate number of values reclaimable once weak tombstones can be GC'd.
    fn weak_tombstone_reclaimable_count(&self) -> u64;

    /// Drops all tables and clears all memtables atomically.
    ///
    /// # Errors
    ///
    /// Will return `Err` only if an IO error occurs.
    fn clear(&self) -> crate::Result<()>;

    /// Performs major compaction, blocking the caller until it's done.
    ///
    /// Returns a [`crate::compaction::CompactionResult`] describing what action was taken.
    ///
    /// # GC watermark (`gc_watermark`)
    ///
    /// `gc_watermark` is the one number that decides what history the engine's
    /// own GC may collect: every snapshot at or above it that is still servable
    /// (above the current retention floor) stays readable, and a version only
    /// snapshots below it could see may be dropped. It never lowers the floor:
    /// a snapshot an earlier run already refused stays refused. Concretely,
    /// only entries whose seqno is `< gc_watermark` are eligible for:
    ///
    /// - dropping shadowed versions / GC-ing tombstones, and
    /// - **folding merge operands** via the [`crate::MergeOperator`]: a key written
    ///   only through [`Self::merge`] (no base value) accumulates one operand per
    ///   call, and reads re-apply the whole chain (`O(operands)` per read) until
    ///   compaction folds it. Folding a chain into a single value is only
    ///   MVCC-safe when no live snapshot reads *between* the operands, which is
    ///   exactly what `gc_watermark` certifies.
    ///
    /// The engine does **not** track snapshots (unlike a `RocksDB`-style
    /// snapshot list); the caller owns snapshot lifecycle and must supply this
    /// watermark:
    ///
    /// - To fold/GC everything (no active snapshots), pass a value **above every
    ///   live seqno** (e.g. the next value from the [`crate::SequenceNumberCounter`]).
    /// - With outstanding snapshots, pass the **oldest** snapshot's seqno so their
    ///   reads stay correct.
    /// - `gc_watermark == 0` certifies nothing as collapsible, so **no folding
    ///   or GC happens**: `major_compact(target, 0)` only restructures tables
    ///   and leaves a merge-only key's full operand chain intact.
    ///
    /// A run that collects anything raises the retention floor to
    /// `gc_watermark - 1` (capped at the install's own seqno) unless it is
    /// already higher. When the opt-in tight-space mode takes over a run
    /// (`RuntimeConfig::tight_space_compaction`), each slice raises it that
    /// far whether or not it collected. The floor is the
    /// only read boundary: from then on a new read at or below it (see
    /// [`oldest_retained_seqno`](Self::oldest_retained_seqno)) fails with
    /// [`Error::SnapshotBelowRetention`](crate::Error::SnapshotBelowRetention),
    /// which is why the watermark must not exceed the oldest snapshot still in
    /// use. A reader opened before the install keeps its own version, and the
    /// tables it references, until it is dropped.
    ///
    /// A [compaction filter](crate::compaction::filter) is the exception: it
    /// acts on whatever version it is shown, regardless of `gc_watermark`
    /// (`0` included). A run whose filter removed or rewrote anything raises
    /// the floor to the install's own seqno, so every snapshot up to it is
    /// refused from then on, however low the watermark was.
    ///
    /// # Errors
    ///
    /// Will return `Err` if an IO error occurs.
    fn major_compact(
        &self,
        target_size: u64,
        gc_watermark: SeqNo,
    ) -> crate::Result<crate::compaction::CompactionResult>;

    /// Returns the disk space used by stale blobs.
    fn stale_blob_bytes(&self) -> u64 {
        0
    }

    /// Gets the space usage of all filters in the tree.
    ///
    /// May not correspond to the actual memory size because filter blocks may be paged out.
    fn filter_size(&self) -> u64;

    /// Gets the memory usage of all pinned filters in the tree.
    fn pinned_filter_size(&self) -> usize;

    /// The serialised and the resident bytes of the tree's filters, and how
    /// they stand against a [`FilterAdvisor`](crate::config::FilterAdvisor)
    /// budget.
    ///
    /// Looks up each live table's filter blocks in the block cache without
    /// counting a hit.
    ///
    /// # Examples
    ///
    /// ```
    /// use lsm_tree::{AbstractTree, Config, config::FilterAdvisor};
    ///
    /// # let folder = tempfile::tempdir()?;
    /// let tree = Config::new(&folder, Default::default(), Default::default())
    ///     .filter_advisor(Some(FilterAdvisor::new(64 * 1_024)))
    ///     .open()?;
    /// tree.insert("a", "value", 0);
    /// tree.flush_active_memtable(0)?;
    ///
    /// let memory = tree.filter_memory();
    /// assert_eq!(memory.serialised_bytes, tree.filter_size());
    /// assert_eq!(memory.budget_bytes, Some(64 * 1_024));
    /// assert!(!memory.over_budget);
    /// # Ok::<(), lsm_tree::Error>(())
    /// ```
    fn filter_memory(&self) -> crate::FilterMemory;

    /// Gets the memory usage of all pinned index blocks in the tree.
    fn pinned_block_index_size(&self) -> usize;

    /// Returns the metrics structure.
    #[cfg(feature = "metrics")]
    fn metrics(&self) -> &Arc<crate::Metrics>;

    /// A point-in-time [`CacheStats`](crate::CacheStats) snapshot of block-cache
    /// effectiveness (cumulative hit / miss counts and rate) and occupancy
    /// (current size against capacity).
    ///
    /// The stable, owned observability view over the block cache, so a consumer
    /// can read cache health without holding the mutable
    /// [`metrics`](Self::metrics) handle. Counts are cumulative since process
    /// start; derive a rate over an interval from the delta between two polls.
    #[cfg(feature = "metrics")]
    fn cache_stats(&self) -> crate::CacheStats;

    /// Acquires the flush lock which is required to call [`Tree::flush`](crate::Tree::flush).
    fn get_flush_lock(&self) -> FlushGuard<'_>;

    /// Synchronously flushes a memtable to a table.
    ///
    /// This method will not make the table immediately available,
    /// use [`AbstractTree::register_tables`] for that.
    ///
    /// Alongside the files comes a [`WritePin`](crate::runtime_config::WritePin)
    /// on the compression dictionaries they were written against. Keep it until
    /// `register_tables` returns: until then no installed file names those
    /// dictionaries, and a collection in between would take them.
    ///
    /// # Errors
    ///
    /// Will return `Err` if an IO error occurs.
    fn flush_to_tables(
        &self,
        stream: impl Iterator<Item = crate::Result<InternalValue>>,
    ) -> crate::Result<Option<FlushToTablesResult>> {
        // The stream's upper bound, when it tells one, is the most entries it
        // holds; the hashes over them are learnt from the first filter.
        let keys = stream
            .size_hint()
            .1
            .map_or(0, |upper| u64::try_from(upper).unwrap_or(u64::MAX));
        self.flush_to_tables_with_rt(stream, Vec::new(), keys, keys, None)
    }

    /// Like [`AbstractTree::flush_to_tables`], but also writes range tombstones.
    /// `keys` bounds the distinct keys of `stream` from above and `hashes` the
    /// filter hashes over them, zero when unknown: a filter advisor keeps room
    /// for every table the flush writes by them. `overwritten`, when given,
    /// names the keys of `stream` its source held more than one version of.
    ///
    /// This is an internal extension hook on the crate's sealed tree types and
    /// is hidden from generated documentation.
    ///
    /// # Errors
    ///
    /// Will return `Err` if an IO error occurs.
    #[doc(hidden)]
    fn flush_to_tables_with_rt(
        &self,
        stream: impl Iterator<Item = crate::Result<InternalValue>>,
        range_tombstones: Vec<crate::range_tombstone::RangeTombstone>,
        keys: u64,
        hashes: u64,
        overwritten: Option<&crate::blob_tree::overwritten::OverwrittenKeys>,
    ) -> crate::Result<Option<FlushToTablesResult>>;

    /// Atomically registers flushed tables into the tree, removing their associated sealed memtables.
    ///
    /// `gc_watermark` must be the same threshold the stream that produced
    /// `tables` ran with, and `collected_below_watermark` whether that stream
    /// actually dropped a version because of it. Together they are the
    /// install's retention effect: understating either admits reads whose
    /// answer the flush already collected, overstating `collected_below_watermark`
    /// refuses reads whose data is still on disk.
    ///
    /// # Errors
    ///
    /// Will return `Err` if an IO error occurs.
    fn register_tables(
        &self,
        tables: &[Table],
        blob_files: Option<&[BlobFile]>,
        frag_map: Option<crate::blob_tree::FragmentationMap>,
        sealed_memtables_to_delete: &[crate::tree::inner::MemtableId],
        gc_watermark: SeqNo,
        collected_below_watermark: bool,
    ) -> crate::Result<()>;

    /// Clears the active memtable atomically.
    fn clear_active_memtable(&self);

    /// Returns the number of sealed memtables.
    fn sealed_memtable_count(&self) -> usize;

    /// Performs compaction on the tree's levels, blocking the caller until it's done.
    ///
    /// Returns a [`crate::compaction::CompactionResult`] describing what action was taken.
    ///
    /// `gc_watermark` has the meaning documented at
    /// [`major_compact`](Self::major_compact): the engine's own GC collects
    /// only versions below it, and a run that collects one raises the
    /// retention floor. Before the strategy chooses, the run unlinks every
    /// table a flushed range tombstone below the watermark spans whole and
    /// is newer than (see [`remove_range`](Self::remove_range)).
    ///
    /// Two things act regardless of the watermark: a compaction filter, and a
    /// strategy that drops whole tables (FIFO eviction, a custom strategy
    /// choosing to drop). Either removes data at any watermark, `0` included,
    /// and raises the floor to the install's own seqno, so a low watermark
    /// does not protect a snapshot from them.
    ///
    /// # Errors
    ///
    /// Will return `Err` if an IO error occurs.
    fn compact(
        &self,
        strategy: Arc<dyn crate::compaction::CompactionStrategy>,
        gc_watermark: SeqNo,
    ) -> crate::Result<crate::compaction::CompactionResult>;

    /// Returns the next table's ID.
    fn get_next_table_id(&self) -> TableId;

    /// Returns the tree config.
    fn tree_config(&self) -> &Config;

    /// Returns the highest sequence number.
    fn get_highest_seqno(&self) -> Option<SeqNo> {
        let memtable_seqno = self.get_highest_memtable_seqno();
        let table_seqno = self.get_highest_persisted_seqno();
        memtable_seqno.max(table_seqno)
    }

    /// Returns the active memtable.
    fn active_memtable(&self) -> Arc<Memtable>;

    /// Whether the active memtable holds nothing, so that the next write makes
    /// one more flush owed. Sealed memtables do not count.
    ///
    /// Under `std` it takes no lock, unlike [`active_memtable`](Self::active_memtable),
    /// so it may be called from inside a version install, as a metered file
    /// system's admission is. A concurrent write can change the answer as it
    /// returns.
    ///
    /// # Examples
    ///
    /// ```
    /// use lsm_tree::{AbstractTree, Config, SequenceNumberCounter};
    /// # let folder = tempfile::tempdir()?;
    /// let tree = Config::new(
    ///     &folder,
    ///     SequenceNumberCounter::default(),
    ///     SequenceNumberCounter::default(),
    /// )
    /// .open()?;
    /// assert!(tree.active_memtable_is_empty());
    ///
    /// tree.insert("key", "value", 0);
    /// assert!(!tree.active_memtable_is_empty());
    ///
    /// tree.rotate_memtable();
    /// assert!(tree.active_memtable_is_empty());
    /// # Ok::<(), lsm_tree::Error>(())
    /// ```
    fn active_memtable_is_empty(&self) -> bool;

    /// Returns the tree type.
    fn tree_type(&self) -> crate::TreeType {
        if self.tree_config().kv_separation_opts.is_some() {
            crate::TreeType::Blob
        } else {
            crate::TreeType::Standard
        }
    }

    /// Seals the active memtable.
    fn rotate_memtable(&self) -> Option<Arc<Memtable>>;

    /// Returns the number of tables currently in the tree.
    fn table_count(&self) -> usize;

    /// Returns the number of tables in `levels[idx]`.
    ///
    /// Returns `None` if the level does not exist (if idx >= 7).
    fn level_table_count(&self, idx: usize) -> Option<usize>;

    /// Returns the number of disjoint runs in L0.
    ///
    /// Can be used to determine whether to write stall.
    fn l0_run_count(&self) -> usize;

    /// Returns the number of blob files currently in the tree.
    fn blob_file_count(&self) -> usize;

    /// Approximates the number of items in the tree.
    fn approximate_len(&self) -> usize;

    /// Returns the disk space usage.
    fn disk_space(&self) -> u64;

    /// Estimates the on-disk bytes and entry count contained in `range` at
    /// `seqno`, WITHOUT reading any data block.
    ///
    /// The estimate interpolates each overlapping SST's data-block offsets at
    /// the range boundaries (block granularity) and adds the active + sealed
    /// memtables' in-range share. For a KV-separated tree the per-SST blob
    /// bytes are apportioned by the same in-range fraction (blob files are not
    /// key-indexed, so a finer estimate is impossible without reading data).
    /// Intended for query planning (split-point selection, cost-based join
    /// ordering), not exact accounting; accuracy is typically within ~10-15%
    /// on roughly-uniform data.
    ///
    /// # Examples
    ///
    /// ```
    /// # use lsm_tree::Error as TreeError;
    /// use lsm_tree::{AbstractTree, Config};
    ///
    /// let folder = tempfile::tempdir()?;
    /// let tree = Config::new(&folder, Default::default(), Default::default()).open()?;
    /// for i in 0..100u32 {
    ///     tree.insert(format!("k{i:04}"), "value", 0);
    /// }
    /// tree.flush_active_memtable(0)?;
    ///
    /// // The full range covers every entry exactly once it is all flushed —
    /// // estimated without reading a single data block.
    /// let all = tree.approximate_range_stats::<&str, _>(.., 1)?;
    /// assert_eq!(all.key_count, tree.approximate_len() as u64);
    /// assert!(all.bytes > 0);
    ///
    /// // A range past every key is empty.
    /// let none = tree.approximate_range_stats("zzzz".."zzzzz", 1)?;
    /// assert_eq!(none.key_count, 0);
    /// assert_eq!(none.bytes, 0);
    /// #
    /// # Ok::<(), TreeError>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns an error if a block index or table metadata read fails.
    fn approximate_range_stats<K: AsRef<[u8]>, R: RangeBounds<K>>(
        &self,
        range: R,
        seqno: SeqNo,
    ) -> crate::Result<crate::ApproximateRangeStats>;

    /// Estimates the row cardinality and selectivity of `range` at `seqno`,
    /// WITHOUT reading any data block.
    ///
    /// Uses the per-data-block zone map (per-block row counts + key ranges) for a
    /// block-granularity row count: every data block whose key range overlaps the
    /// query contributes its recorded row count, plus the active + sealed
    /// memtables' in-range counts. When a table has no zone map the row count
    /// falls back to the byte-fraction estimate of [`approximate_range_stats`].
    /// Selectivity is `rows / total_rows`, monotonic in predicate tightness, for
    /// cost-based planning (join ordering, scan-vs-seek). Not exact accounting.
    ///
    /// [`approximate_range_stats`]: AbstractTree::approximate_range_stats
    ///
    /// # Examples
    ///
    /// ```
    /// # use lsm_tree::Error as TreeError;
    /// use lsm_tree::{AbstractTree, Config};
    ///
    /// let folder = tempfile::tempdir()?;
    /// let tree = Config::new(&folder, Default::default(), Default::default()).open()?;
    /// for i in 0..100u32 {
    ///     tree.insert(format!("k{i:04}"), "v", 0);
    /// }
    /// tree.flush_active_memtable(0)?;
    ///
    /// // The full range covers every row; selectivity is 1.0.
    /// let all = tree.approximate_range_cardinality::<&str, _>(.., 1)?;
    /// assert_eq!(all.rows, tree.approximate_len() as u64);
    /// assert!((all.selectivity - 1.0).abs() < 1e-9);
    ///
    /// // A tighter range selects fewer rows.
    /// let part = tree.approximate_range_cardinality("k0000".."k0050", 1)?;
    /// assert!(part.selectivity <= all.selectivity);
    /// #
    /// # Ok::<(), TreeError>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns an error if a block index, zone map, or table metadata read fails.
    fn approximate_range_cardinality<K: AsRef<[u8]>, R: RangeBounds<K>>(
        &self,
        range: R,
        seqno: SeqNo,
    ) -> crate::Result<crate::RangeCardinality>;

    /// Returns the highest sequence number of the active memtable.
    fn get_highest_memtable_seqno(&self) -> Option<SeqNo>;

    /// Returns the highest sequence number that is flushed to disk.
    ///
    /// Under [`FlushDurability::Deferred`](crate::config::FlushDurability::Deferred)
    /// only the flushes a persist made durable count: a flush that is
    /// installed but not yet persisted would be lost by a power loss.
    fn get_highest_persisted_seqno(&self) -> Option<SeqNo>;

    /// Returns the retention floor: the lower bound of the readable snapshot
    /// window.
    ///
    /// A snapshot is servable iff it is `0` (sees nothing) or strictly above
    /// this seqno; a new read at `0 < seqno <= oldest_retained_seqno()` fails
    /// with
    /// [`Error::SnapshotBelowRetention`](crate::Error::SnapshotBelowRetention).
    /// 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 reader
    /// opened before the floor rose is not affected: it keeps the version it
    /// resolved. A fresh tree reports `0`.
    ///
    /// The install that discards what older snapshots saw raises the floor in
    /// the same version edit, so the boundary is the same live and after a
    /// reopen: after a GC with watermark `w` it is `w - 1`, capped at the
    /// install's own seqno; 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 collect the
    /// // old version; 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 floor: the highest snapshot seqno the
    /// tree refuses, now and after a reopen. It always equals
    /// [`oldest_retained_seqno`](Self::oldest_retained_seqno).
    ///
    /// 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 moved the boundary (GC
    /// compactions, `clear`, table drops, filtering compactions, and
    /// [`lower_retention_floor`](Self::lower_retention_floor)), so no
    /// caller-side bookkeeping of watermarks is needed. `0` until the first
    /// such install.
    fn retention_floor(&self) -> SeqNo;

    /// Lowers the persisted retention floor to `floor`, so snapshots above it
    /// are served again, now and after a reopen.
    ///
    /// For a rebuild from a source that kept history above its own floor:
    /// [`clear`](Self::clear) the tree, reinstall the source's history at its
    /// original seqnos, then lower the floor to the source's
    /// [`retention_floor`](Self::retention_floor). Until this call the clear's
    /// floor refuses every snapshot below it, so a reader never sees the
    /// history half reinstalled, and a crash before the call leaves that
    /// refusal in place.
    ///
    /// The tree cannot check the claim this call makes: that it holds every
    /// version above `floor` a snapshot above `floor` could see. That includes
    /// not collecting any of the reinstalled history with a GC watermark above
    /// `floor + 1` before the call.
    ///
    /// # Errors
    ///
    /// [`Error::RetentionFloorRaise`](crate::Error::RetentionFloorRaise) when
    /// `floor` is above the current floor, which only GC and the installs that
    /// drop data raise; nothing changes.
    /// [`Error::RetentionFloorAboveSeqno`](crate::Error::RetentionFloorAboveSeqno)
    /// when `floor` is above the seqno counter (one reset on reopen can stand
    /// below the current floor); nothing changes. An I/O error persisting the
    /// new floor.
    ///
    /// # Examples
    ///
    /// ```
    /// # let folder = tempfile::tempdir()?;
    /// use lsm_tree::{AbstractTree, Config, SequenceNumberCounter};
    ///
    /// let tree = Config::new(folder, SequenceNumberCounter::new(100), Default::default()).open()?;
    /// tree.clear()?;
    /// // Reinstall a source's history, which it kept above seqno 5.
    /// tree.insert("k", "v", 10);
    /// tree.lower_retention_floor(5)?;
    /// assert_eq!(tree.get("k", 11)?.as_deref(), Some(b"v".as_slice()));
    /// assert!(tree.get("k", 5).is_err());
    /// #
    /// # Ok::<(), lsm_tree::Error>(())
    /// ```
    fn lower_retention_floor(&self, floor: SeqNo) -> crate::Result<()>;

    /// Returns the limiter that throttles this tree's compactions: the one
    /// supplied through
    /// [`Config::compaction_rate_limiter`](crate::Config::compaction_rate_limiter),
    /// which other trees may share, or the tree's own built from
    /// [`Config::compaction_rate_limit`](crate::Config::compaction_rate_limit).
    ///
    /// Change the rate while the tree runs, with no restart, through
    /// [`RateLimiter::set_rate`](crate::rate_limiter::RateLimiter::set_rate):
    /// every tree holding the limiter is bounded by the new rate from its next
    /// request.
    ///
    /// # Examples
    ///
    /// ```
    /// use lsm_tree::{AbstractTree, Config};
    ///
    /// let folder = tempfile::tempdir()?;
    /// let tree = Config::new(&folder, Default::default(), Default::default())
    ///     .compaction_rate_limit(16 * 1_024 * 1_024)
    ///     .open()?;
    /// tree.compaction_rate_limiter().set_rate(64 * 1_024 * 1_024);
    /// assert_eq!(tree.compaction_rate_limiter().rate(), 64 * 1_024 * 1_024);
    /// #
    /// # Ok::<(), lsm_tree::Error>(())
    /// ```
    fn compaction_rate_limiter(&self) -> alloc::sync::Arc<crate::rate_limiter::RateLimiter>;

    /// Scans the entire tree, returning the number of items.
    ///
    /// ###### Caution
    ///
    /// This operation scans the entire tree: O(n) complexity!
    ///
    /// Never, under any circumstances, use .`len()` == 0 to check
    /// if the tree is empty, use [`Tree::is_empty`](crate::Tree::is_empty) instead.
    ///
    /// # Examples
    ///
    /// ```
    /// # use lsm_tree::Error as TreeError;
    /// use lsm_tree::{AbstractTree, Config, Tree};
    ///
    /// let folder = tempfile::tempdir()?;
    /// let tree = Config::new(&folder, Default::default(), Default::default()).open()?;
    ///
    /// assert_eq!(tree.len(0, None)?, 0);
    /// tree.insert("1", "abc", 0);
    /// tree.insert("3", "abc", 1);
    /// tree.insert("5", "abc", 2);
    /// assert_eq!(tree.len(3, None)?, 3);
    /// #
    /// # Ok::<(), TreeError>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Will return `Err` if an IO error occurs.
    fn len(&self, seqno: SeqNo, index: Option<(Arc<Memtable>, SeqNo)>) -> crate::Result<usize> {
        let mut count = 0;

        for item in self.iter(seqno, index) {
            let _ = item.key()?;
            count += 1;
        }

        Ok(count)
    }

    /// Returns `true` if the tree is empty.
    ///
    /// This operation has O(log N) complexity.
    ///
    /// # Examples
    ///
    /// ```
    /// # let folder = tempfile::tempdir()?;
    /// use lsm_tree::{AbstractTree, Config, Tree};
    ///
    /// let tree = Config::new(folder, Default::default(), Default::default()).open()?;
    /// assert!(tree.is_empty(0, None)?);
    ///
    /// tree.insert("a", "abc", 0);
    /// assert!(!tree.is_empty(1, None)?);
    /// #
    /// # Ok::<(), lsm_tree::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Will return `Err` if an IO error occurs.
    fn is_empty(&self, seqno: SeqNo, index: Option<(Arc<Memtable>, SeqNo)>) -> crate::Result<bool> {
        Ok(self
            .first_key_value(seqno, index)
            .map(crate::Guard::key)
            .transpose()?
            .is_none())
    }

    /// Returns the first key-value pair in the tree.
    /// The key in this pair is the minimum key in the tree.
    ///
    /// # Examples
    ///
    /// ```
    /// # use lsm_tree::Error as TreeError;
    /// # use lsm_tree::{AbstractTree, Config, Tree, Guard};
    /// #
    /// # let folder = tempfile::tempdir()?;
    /// let tree = Config::new(folder, Default::default(), Default::default()).open()?;
    ///
    /// tree.insert("1", "abc", 0);
    /// tree.insert("3", "abc", 1);
    /// tree.insert("5", "abc", 2);
    ///
    /// let key = tree.first_key_value(3, None).expect("item should exist").key()?;
    /// assert_eq!(&*key, "1".as_bytes());
    /// #
    /// # Ok::<(), TreeError>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Will return `Err` if an IO error occurs.
    fn first_key_value(
        &self,
        seqno: SeqNo,
        index: Option<(Arc<Memtable>, SeqNo)>,
    ) -> Option<IterGuardImpl> {
        self.iter(seqno, index).next()
    }

    /// Returns the last key-value pair in the tree.
    /// The key in this pair is the maximum key in the tree.
    ///
    /// # Examples
    ///
    /// ```
    /// # use lsm_tree::Error as TreeError;
    /// # use lsm_tree::{AbstractTree, Config, Tree, Guard};
    /// #
    /// # let folder = tempfile::tempdir()?;
    /// # let tree = Config::new(folder, Default::default(), Default::default()).open()?;
    /// #
    /// tree.insert("1", "abc", 0);
    /// tree.insert("3", "abc", 1);
    /// tree.insert("5", "abc", 2);
    ///
    /// let key = tree.last_key_value(3, None).expect("item should exist").key()?;
    /// assert_eq!(&*key, "5".as_bytes());
    /// #
    /// # Ok::<(), TreeError>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Will return `Err` if an IO error occurs.
    fn last_key_value(
        &self,
        seqno: SeqNo,
        index: Option<(Arc<Memtable>, SeqNo)>,
    ) -> Option<IterGuardImpl> {
        self.iter(seqno, index).next_back()
    }

    /// Returns the size of a value if it exists.
    ///
    /// # Examples
    ///
    /// ```
    /// # let folder = tempfile::tempdir()?;
    /// use lsm_tree::{AbstractTree, Config, Tree};
    ///
    /// let tree = Config::new(folder, Default::default(), Default::default()).open()?;
    /// tree.insert("a", "my_value", 0);
    ///
    /// let size = tree.size_of("a", 1)?.unwrap_or_default();
    /// assert_eq!("my_value".len() as u32, size);
    ///
    /// let size = tree.size_of("b", 1)?.unwrap_or_default();
    /// assert_eq!(0, size);
    /// #
    /// # Ok::<(), lsm_tree::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Will return `Err` if an IO error occurs.
    fn size_of<K: AsRef<[u8]>>(&self, key: K, seqno: SeqNo) -> crate::Result<Option<u32>>;

    /// Retrieves an item from the tree.
    ///
    /// # Examples
    ///
    /// ```
    /// # let folder = tempfile::tempdir()?;
    /// use lsm_tree::{AbstractTree, Config, Tree};
    ///
    /// let tree = Config::new(folder, Default::default(), Default::default()).open()?;
    /// tree.insert("a", "my_value", 0);
    ///
    /// let item = tree.get("a", 1)?;
    /// assert_eq!(Some("my_value".as_bytes().into()), item);
    /// #
    /// # Ok::<(), lsm_tree::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Will return `Err` if an IO error occurs, or
    /// [`Error::SnapshotBelowRetention`](crate::Error::SnapshotBelowRetention)
    /// when `seqno` is at or below the retention floor (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).
    ///
    /// When the value is backed by an on-disk data block, implementations
    /// may return [`PinnableSlice::Pinned`](crate::PinnableSlice::Pinned) holding a reference to that block's
    /// decompressed buffer (avoiding a data copy). Memtable and blob-resolved
    /// values use [`PinnableSlice::Owned`](crate::PinnableSlice::Owned). The default implementation always
    /// returns `Owned`; only [`Tree`](crate::Tree) overrides with the pinned path.
    ///
    /// The existing [`AbstractTree::get`] method is unaffected.
    ///
    /// # Examples
    ///
    /// ```
    /// # let folder = tempfile::tempdir()?;
    /// use lsm_tree::{AbstractTree, Config, Tree};
    ///
    /// let tree = Config::new(&folder, Default::default(), Default::default()).open()?;
    /// tree.insert("a", "my_value", 0);
    ///
    /// let item = tree.get_pinned("a", 1)?;
    /// assert_eq!(item.as_ref().map(|v| v.as_ref()), Some("my_value".as_bytes()));
    /// #
    /// # Ok::<(), lsm_tree::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Will return `Err` if an IO error occurs.
    fn get_pinned<K: AsRef<[u8]>>(
        &self,
        key: K,
        seqno: SeqNo,
    ) -> crate::Result<Option<crate::PinnableSlice>> {
        // Default: delegate to get() and wrap as Owned
        self.get(key, seqno)
            .map(|opt| opt.map(crate::PinnableSlice::owned))
    }

    /// Returns `true` if the tree contains the specified key.
    ///
    /// # Examples
    ///
    /// ```
    /// # let folder = tempfile::tempdir()?;
    /// # use lsm_tree::{AbstractTree, Config, Tree};
    /// #
    /// let tree = Config::new(folder, Default::default(), Default::default()).open()?;
    /// assert!(!tree.contains_key("a", 0)?);
    ///
    /// tree.insert("a", "abc", 0);
    /// assert!(tree.contains_key("a", 1)?);
    /// #
    /// # Ok::<(), lsm_tree::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Will return `Err` if an IO error occurs.
    fn contains_key<K: AsRef<[u8]>>(&self, key: K, seqno: SeqNo) -> crate::Result<bool> {
        self.get(key, seqno).map(|x| x.is_some())
    }

    /// Returns `true` if the tree contains any key with the given prefix.
    ///
    /// This is a convenience method that checks whether the corresponding
    /// prefix iterator yields at least one item, while surfacing any IO
    /// errors via the `Result` return type. Implementations may override
    /// this method to provide a more efficient prefix-existence check.
    ///
    /// # Examples
    ///
    /// ```
    /// # let folder = tempfile::tempdir()?;
    /// use lsm_tree::{AbstractTree, Config, Tree};
    ///
    /// let tree = Config::new(folder, Default::default(), Default::default()).open()?;
    /// assert!(!tree.contains_prefix("abc", 0, None)?);
    ///
    /// tree.insert("abc:1", "value", 0);
    /// assert!(tree.contains_prefix("abc", 1, None)?);
    /// assert!(!tree.contains_prefix("xyz", 1, None)?);
    /// #
    /// # Ok::<(), lsm_tree::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Will return `Err` if an IO error occurs.
    fn contains_prefix<K: AsRef<[u8]>>(
        &self,
        prefix: K,
        seqno: SeqNo,
        index: Option<(Arc<Memtable>, SeqNo)>,
    ) -> crate::Result<bool> {
        match self.prefix(prefix, seqno, index).next() {
            Some(guard) => guard.key().map(|_| true),
            None => Ok(false),
        }
    }

    /// Reads multiple keys from the tree.
    ///
    /// Implementations may choose to perform all lookups against a single
    /// version snapshot and acquire the version lock only once, which can be
    /// more efficient than calling [`AbstractTree::get`] in a loop. The
    /// default trait implementation, however, is a convenience wrapper that
    /// simply calls [`AbstractTree::get`] for each key and therefore does not
    /// guarantee a single-snapshot or single-lock acquisition. Optimized
    /// implementations (such as [`Tree`](crate::Tree) and
    /// [`BlobTree`](crate::BlobTree)) provide the
    /// single-snapshot/one-lock behavior.
    ///
    /// # Examples
    ///
    /// ```
    /// # let folder = tempfile::tempdir()?;
    /// use lsm_tree::{AbstractTree, Config, Tree};
    ///
    /// let tree = Config::new(folder, Default::default(), Default::default()).open()?;
    /// tree.insert("a", "value_a", 0);
    /// tree.insert("b", "value_b", 1);
    ///
    /// let results = tree.multi_get(["a", "b", "c"], 2)?;
    /// assert_eq!(results[0], Some("value_a".as_bytes().into()));
    /// assert_eq!(results[1], Some("value_b".as_bytes().into()));
    /// assert_eq!(results[2], None);
    /// #
    /// # Ok::<(), lsm_tree::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Will return `Err` if an IO error occurs.
    fn multi_get<K: AsRef<[u8]>>(
        &self,
        keys: impl IntoIterator<Item = K>,
        seqno: SeqNo,
    ) -> crate::Result<Vec<Option<UserValue>>> {
        keys.into_iter().map(|key| self.get(key, seqno)).collect()
    }

    /// Applies a [`WriteBatch`](crate::WriteBatch) with the given sequence number.
    ///
    /// All entries share a single seqno. This is more efficient than individual
    /// writes because the version-history lock and memtable size accounting
    /// are performed only once for the entire batch.
    ///
    /// **Visibility:** entries become individually visible to concurrent readers
    /// as they are inserted. For atomic batch visibility, the caller must
    /// publish `seqno` (via `visible_seqno.fetch_max(seqno + 1)`) only
    /// **after** this method returns.
    ///
    /// Returns the total bytes added and new size of the memtable.
    ///
    /// # Examples
    ///
    /// ```
    /// # let folder = tempfile::tempdir()?;
    /// use lsm_tree::{AbstractTree, Config, Tree, WriteBatch};
    ///
    /// let tree = Config::new(&folder, Default::default(), Default::default()).open()?;
    ///
    /// let mut batch = WriteBatch::new();
    /// batch.insert("key1", "value1");
    /// batch.insert("key2", "value2");
    /// batch.remove("key3");
    ///
    /// let (bytes_added, memtable_size) = tree.apply_batch(batch, 0)?;
    /// assert!(bytes_added > 0);
    /// #
    /// # Ok::<(), lsm_tree::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`Error::MixedOperationBatch`](crate::Error::MixedOperationBatch)
    /// if the batch contains mixed operation types for the same user key.
    fn apply_batch(&self, batch: crate::WriteBatch, seqno: SeqNo) -> crate::Result<(u64, u64)>;

    /// Inserts a key-value pair into the tree.
    ///
    /// If the key already exists, the item will be overwritten.
    ///
    /// Returns the added item's size and new size of the memtable.
    ///
    /// # Examples
    ///
    /// ```
    /// # let folder = tempfile::tempdir()?;
    /// use lsm_tree::{AbstractTree, Config, Tree};
    ///
    /// let tree = Config::new(folder, Default::default(), Default::default()).open()?;
    /// tree.insert("a", "abc", 0);
    /// #
    /// # Ok::<(), lsm_tree::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Will return `Err` if an IO error occurs.
    fn insert<K: Into<UserKey>, V: Into<UserValue>>(
        &self,
        key: K,
        value: V,
        seqno: SeqNo,
    ) -> (u64, u64);

    /// Admission-gated [`insert`](Self::insert): consults
    /// [`write_admission`](Self::write_admission) first and declines with
    /// [`Error::StorageFull`](crate::Error::StorageFull) when the tree is over
    /// budget, otherwise inserts and returns the same `(added_bytes,
    /// memtable_size)` tuple. This is the write entry a space-aware caller uses
    /// so over-budget writes are refused up front rather than failing a flush
    /// later; bare [`insert`](Self::insert) stays infallible for callers that
    /// do not opt into admission control.
    ///
    /// # Errors
    ///
    /// [`Error::StorageFull`](crate::Error::StorageFull) when the admission gate
    /// is closed.
    fn try_insert<K: Into<UserKey>, V: Into<UserValue>>(
        &self,
        key: K,
        value: V,
        seqno: SeqNo,
    ) -> crate::Result<(u64, u64)> {
        self.write_admission()?;
        Ok(self.insert(key, value, seqno))
    }

    /// Admission-gated [`merge`](Self::merge). See
    /// [`try_insert`](Self::try_insert).
    ///
    /// # Errors
    ///
    /// [`Error::StorageFull`](crate::Error::StorageFull) when the admission gate
    /// is closed.
    fn try_merge<K: Into<UserKey>, V: Into<UserValue>>(
        &self,
        key: K,
        operand: V,
        seqno: SeqNo,
    ) -> crate::Result<(u64, u64)> {
        self.write_admission()?;
        Ok(self.merge(key, operand, seqno))
    }

    /// Admission-gated [`remove`](Self::remove). See
    /// [`try_insert`](Self::try_insert).
    ///
    /// Note a tombstone is itself a write that consumes space; an over-budget
    /// tree refuses it too. Reclaim space via compaction (never gated) rather
    /// than relying on deletes when already read-only.
    ///
    /// # Errors
    ///
    /// [`Error::StorageFull`](crate::Error::StorageFull) when the admission gate
    /// is closed.
    fn try_remove<K: Into<UserKey>>(&self, key: K, seqno: SeqNo) -> crate::Result<(u64, u64)> {
        self.write_admission()?;
        Ok(self.remove(key, seqno))
    }

    /// Admission-gated [`remove_weak`](Self::remove_weak). See
    /// [`try_insert`](Self::try_insert).
    ///
    /// # Errors
    ///
    /// [`Error::StorageFull`](crate::Error::StorageFull) when the admission gate
    /// is closed.
    fn try_remove_weak<K: Into<UserKey>>(&self, key: K, seqno: SeqNo) -> crate::Result<(u64, u64)> {
        self.write_admission()?;
        Ok(self.remove_weak(key, seqno))
    }

    /// Admission-gated [`remove_range`](Self::remove_range). See
    /// [`try_insert`](Self::try_insert).
    ///
    /// # Errors
    ///
    /// [`Error::StorageFull`](crate::Error::StorageFull) when the admission gate
    /// is closed.
    fn try_remove_range<K: Into<UserKey>>(
        &self,
        start: K,
        end: K,
        seqno: SeqNo,
    ) -> crate::Result<u64> {
        self.write_admission()?;
        Ok(self.remove_range(start, end, seqno))
    }

    /// Removes an item from the tree.
    ///
    /// Returns the added item's size and new size of the memtable.
    ///
    /// # Examples
    ///
    /// ```
    /// # let folder = tempfile::tempdir()?;
    /// # use lsm_tree::{AbstractTree, Config, Tree};
    /// #
    /// # let tree = Config::new(folder, Default::default(), Default::default()).open()?;
    /// tree.insert("a", "abc", 0);
    ///
    /// let item = tree.get("a", 1)?.expect("should have item");
    /// assert_eq!("abc".as_bytes(), &*item);
    ///
    /// tree.remove("a", 1);
    ///
    /// let item = tree.get("a", 2)?;
    /// assert_eq!(None, item);
    /// #
    /// # Ok::<(), lsm_tree::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Will return `Err` if an IO error occurs.
    fn remove<K: Into<UserKey>>(&self, key: K, seqno: SeqNo) -> (u64, u64);

    /// Writes a merge operand for a key.
    ///
    /// The operand is stored as a partial update that will be combined with
    /// other operands and/or a base value via the configured [`crate::MergeOperator`]
    /// during reads and compaction.
    ///
    /// Returns the added item's size and new size of the memtable.
    ///
    /// # Examples
    ///
    /// ```
    /// # let folder = tempfile::tempdir()?;
    /// # use lsm_tree::{AbstractTree, Config, MergeOperator, UserValue};
    /// # use std::sync::Arc;
    /// # struct SumMerge;
    /// # impl MergeOperator for SumMerge {
    /// #     fn merge(&self, _key: &[u8], base: Option<&[u8]>, operands: &[&[u8]]) -> lsm_tree::Result<UserValue> {
    /// #         let mut sum: i64 = base.map_or(0, |b| i64::from_le_bytes(b.try_into().unwrap()));
    /// #         for op in operands { sum += i64::from_le_bytes((*op).try_into().unwrap()); }
    /// #         Ok(sum.to_le_bytes().to_vec().into())
    /// #     }
    /// # }
    /// # let tree = Config::new(folder, Default::default(), Default::default())
    /// #     .with_merge_operator(Some(Arc::new(SumMerge)))
    /// #     .open()?;
    /// tree.merge("counter", 1_i64.to_le_bytes(), 0);
    /// # Ok::<(), lsm_tree::Error>(())
    /// ```
    fn merge<K: Into<UserKey>, V: Into<UserValue>>(
        &self,
        key: K,
        operand: V,
        seqno: SeqNo,
    ) -> (u64, u64);

    /// Removes an item from the tree.
    ///
    /// The tombstone marker of this delete operation will vanish when it
    /// collides with its corresponding insertion.
    /// This may cause older versions of the value to be resurrected, so it should
    /// only be used and preferred in scenarios where a key is only ever written once.
    ///
    /// Returns the added item's size and new size of the memtable.
    ///
    /// # Examples
    ///
    /// ```
    /// # let folder = tempfile::tempdir()?;
    /// # use lsm_tree::{AbstractTree, Config, Tree};
    /// #
    /// # let tree = Config::new(folder, Default::default(), Default::default()).open()?;
    /// tree.insert("a", "abc", 0);
    ///
    /// let item = tree.get("a", 1)?.expect("should have item");
    /// assert_eq!("abc".as_bytes(), &*item);
    ///
    /// tree.remove_weak("a", 1);
    ///
    /// let item = tree.get("a", 2)?;
    /// assert_eq!(None, item);
    /// #
    /// # Ok::<(), lsm_tree::Error>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Will return `Err` if an IO error occurs.
    #[doc(hidden)]
    fn remove_weak<K: Into<UserKey>>(&self, key: K, seqno: SeqNo) -> (u64, u64);

    /// Deletes all keys in the range `[start, end)` by inserting a range tombstone.
    ///
    /// This is much more efficient than deleting keys individually when
    /// removing a contiguous range of keys. The deletion is exact at once,
    /// whatever tables hold the keys; a snapshot below `seqno` still reads them.
    ///
    /// Space comes back without rewriting: once the tombstone is flushed, every
    /// compaction whose `gc_watermark` is above `seqno` first unlinks each
    /// table the tombstone spans whole and is newer than, before its strategy
    /// chooses. Tables that reach past `[start, end)` keep their bytes until a
    /// merge rewrites them. Dropping tables raises the retention floor, at most
    /// to `seqno`: snapshots at or below the floor report
    /// [`Error::SnapshotBelowRetention`](crate::Error::SnapshotBelowRetention).
    ///
    /// Nothing is rewritten only where the range ends at table boundaries,
    /// which holds when keys arrive in the order of the range (a time series
    /// expiring its oldest days); a range cutting through tables leaves their
    /// survivors to the merge that rewrites them. The deletion covers this
    /// tree only: a caller indexing these keys elsewhere removes them there.
    ///
    /// To empty the whole tree, [`clear`](Self::clear) does it without a
    /// tombstone.
    ///
    /// Returns the approximate size added to the memtable.
    /// Returns 0 if `start >= end` (invalid interval is silently ignored).
    ///
    /// This is a required method on the crate's sealed tree types.
    fn remove_range<K: Into<UserKey>>(&self, start: K, end: K, seqno: SeqNo) -> u64;

    /// Deletes all keys with the given prefix by inserting a range tombstone.
    ///
    /// This is sugar over [`AbstractTree::remove_range`] using prefix bounds.
    ///
    /// Returns the approximate size added to the memtable.
    /// Returns 0 for empty prefixes or all-`0xFF` prefixes (cannot form valid half-open range).
    fn remove_prefix<K: AsRef<[u8]>>(&self, prefix: K, seqno: SeqNo) -> u64 {
        use crate::range::prefix_to_range;
        use core::ops::Bound;

        let (lo, hi) = prefix_to_range(prefix.as_ref());

        let Bound::Included(start) = lo else { return 0 };

        // Bound::Unbounded means the prefix is all 0xFF — no representable
        // exclusive upper bound exists, so we cannot form a valid range tombstone.
        let Bound::Excluded(end) = hi else { return 0 };

        self.remove_range(start, end, seqno)
    }
}