coordinode-lsm-tree 7.6.0

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

pub(crate) mod blob_lifetime;
mod block_size;
mod column_encoding;
mod compression;
mod delete_strategy;
mod filter;
mod flush_durability;
mod hash_ratio;
mod locator;
mod pinning;
mod restart_interval;

pub use blob_lifetime::{LifetimeGroups, LifetimeHint, LifetimeHintFn};
pub use block_size::{BlockSizePolicy, MAX_BLOCK_SIZE};
pub use column_encoding::{ColumnEncoding, ColumnEncodingPolicy};
pub use compression::CompressionPolicy;
pub use delete_strategy::{DeleteStrategy, DeleteStrategyPolicy};
pub use filter::{BloomConstructionPolicy, FilterAdvisor, FilterPolicy, FilterPolicyEntry};
pub use flush_durability::FlushDurability;
pub use hash_ratio::HashRatioPolicy;
pub use locator::{LocatorPolicy, LocatorPolicyEntry, LocatorPrecision};
pub use pinning::PinningPolicy;
pub use restart_interval::RestartIntervalPolicy;

/// Partitioning policy for indexes and filters
pub type PartitioningPolicy = PinningPolicy;

use crate::path::PathBuf;
use crate::{
    AnyTree, BlobTree, Cache, CompressionType, DescriptorTable, SharedSequenceNumberGenerator,
    Tree,
    compaction::filter::Factory,
    comparator::SharedComparator,
    encryption::EncryptionProvider,
    file::TABLES_FOLDER,
    fs::{Fs, SyncMode},
    merge_operator::MergeOperator,
    path::absolute_path,
    prefix::PrefixExtractor,
};
// std-only: used solely by the std-gated `Config::default` / `Config::new`
// constructors (the no_std path builds `Config` field-by-field).
#[cfg(feature = "std")]
use crate::{SequenceNumberCounter, comparator, path::Path, version::DEFAULT_LEVEL_COUNT};
use alloc::sync::Arc;
#[cfg(not(feature = "std"))]
use alloc::vec::Vec;
use core::ops::Range;

/// The size a columnar row group is cut at unless
/// [`Config::columnar_row_group_size_policy`] says otherwise.
///
/// Four row pages of [`DEFAULT_COLUMNAR_PAGE_SIZE`]: the smallest group that
/// reads fewer bytes than one-page groups on a near-full scan and a point
/// read without reading more on a sparse scan. The measured grid it was chosen
/// from is in `docs/columnar-page-format.md`.
pub const DEFAULT_COLUMNAR_ROW_GROUP_SIZE: u32 = 16 * 1_024;

/// The size a columnar row group's rows are cut into row pages at unless
/// [`Config::columnar_page_size_policy`] says otherwise.
pub const DEFAULT_COLUMNAR_PAGE_SIZE: u32 = 4_096;

/// How a columnar read fetches the pages it wants: how much one request may
/// ask for, and how many requests it keeps in flight at once.
///
/// A property of the reader, not of the file: the same table reads under any
/// budget and returns the same rows, and a larger budget takes the pages in
/// fewer requests, because a column's pages lie next to each other and a run
/// of them is one request up to the I/O buffer. Set per tree with
/// [`Config::columnar_read_budget`].
///
/// # Examples
///
/// ```
/// use lsm_tree::config::ReadBudget;
///
/// // 256 KiB per request, eight requests in flight at once.
/// let budget = ReadBudget::new(256 * 1_024, 8);
/// assert_eq!(budget.io_buffer(), 256 * 1_024);
/// assert_eq!(budget.in_flight(), 8);
/// ```
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct ReadBudget {
    io_buffer: u32,
    in_flight: u16,
}

impl ReadBudget {
    /// The least a request may ask for: a block header's worth, below which
    /// no request holds even the header of the block it reads.
    pub const MIN_IO_BUFFER: u32 = 64;

    /// A budget of `io_buffer` bytes per request and `in_flight` requests at
    /// once. A budget below [`Self::MIN_IO_BUFFER`] is raised to it, and a
    /// zero `in_flight` to one: a read needs at least one request of at least
    /// a header to make progress.
    ///
    /// A request never splits a block: a directory, zone block or page larger
    /// than the I/O buffer is still one request, and blocks are asked for
    /// together only while they fit it.
    #[must_use]
    pub const fn new(io_buffer: u32, in_flight: u16) -> Self {
        Self {
            io_buffer: if io_buffer < Self::MIN_IO_BUFFER {
                Self::MIN_IO_BUFFER
            } else {
                io_buffer
            },
            in_flight: if in_flight == 0 { 1 } else { in_flight },
        }
    }

    /// Bytes one request may ask for.
    #[must_use]
    pub const fn io_buffer(&self) -> u32 {
        self.io_buffer
    }

    /// Requests submitted together, which a backend with batched I/O keeps
    /// in flight at once.
    #[must_use]
    pub const fn in_flight(&self) -> u16 {
        self.in_flight
    }
}

impl Default for ReadBudget {
    /// 1 MiB per request, the I/O buffer Vortex reads a column with, and 16
    /// requests in flight.
    fn default() -> Self {
        Self::new(1 << 20, 16)
    }
}

/// The page bytes a projected columnar scan holds at once unless
/// [`Config::columnar_scan_budget`] says otherwise: what one read request of
/// the default [`ReadBudget`] fetches.
pub const DEFAULT_COLUMNAR_SCAN_BUDGET: u64 = 1 << 20;

/// The default of [`Config::multi_get_metadata_budget`].
///
/// Room for the filters of a few large tables in flight together, which is
/// all the overlap the reads of a level gain from.
pub const DEFAULT_MULTI_GET_METADATA_BUDGET: u64 = 8 << 20;

/// The default of [`Config::table_target_size`]: 64 MiB, the size the leveled
/// and tiered strategies cut compaction output at.
pub const DEFAULT_TABLE_TARGET_SIZE: u64 = 64 * 1_024 * 1_024;

/// Per-level filesystem routing entry for tiered storage.
///
/// Maps a range of LSM levels to a base directory and filesystem backend.
/// Tables at these levels are stored under `path/tables/`.
///
/// # Example
///
/// ```
/// use lsm_tree::config::LevelRoute;
/// use lsm_tree::fs::StdFs;
/// use std::sync::Arc;
///
/// // Hot tier: L0-L1 on NVMe
/// let hot = LevelRoute {
///     levels: 0..2,
///     path: "/mnt/nvme/db".into(),
///     fs: Arc::new(StdFs),
/// };
///
/// // Cold tier: L4-L6 on HDD
/// let cold = LevelRoute {
///     levels: 4..7,
///     path: "/mnt/hdd/db".into(),
///     fs: Arc::new(StdFs),
/// };
/// ```
#[derive(Clone)]
pub struct LevelRoute {
    /// LSM levels this route covers (e.g., `0..2` for L0–L1).
    pub levels: Range<u8>,

    /// Base data directory for tables at these levels.
    pub path: PathBuf,

    /// Filesystem backend for I/O at these levels.
    pub fs: Arc<dyn Fs>,
}

impl core::fmt::Debug for LevelRoute {
    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
        f.debug_struct("LevelRoute")
            .field("levels", &self.levels)
            .field("path", &self.path)
            .finish_non_exhaustive()
    }
}

/// Policy governing what `Tree::open` does when the manifest's edit log ends
/// inside a record.
///
/// The manifest is a snapshot plus an append-only edit log. The snapshot's
/// sections are checksummed blocks its checksummed footer indexes, so they
/// are read whole or not at all and no mode relaxes them. The only damage an
/// append can leave that is not already covered is a torn last record: power
/// was lost mid-append, so the operation it records was never acknowledged.
/// This is the one thing a mode chooses.
///
/// Corruption of anything already committed fails the open in every mode.
/// Rolling a committed edit back would roll back every committed edit after
/// it, and the open then deletes the tables those edits added as orphans,
/// losing acknowledged data for good. `RocksDB` and Pebble refuse the same
/// way; the remedy is [`Config::repair`], which rebuilds the manifest from
/// the table files themselves.
#[derive(Copy, Clone, Debug, PartialEq, Eq, Default)]
pub enum ManifestRecoveryMode {
    /// Any defect fails the open, a torn last edit included, so an operator
    /// decides how to resolve it ([`Config::repair`] recovers every table on
    /// disk). The default.
    #[default]
    AbsoluteConsistency,

    /// A torn last edit is rolled back and the tree opens at the durable
    /// prefix, with a `warn!`. The torn edit was never acknowledged, so
    /// nothing a caller was told is durable is lost. Any other defect still
    /// fails the open.
    TolerateCorruptedTailRecords,
}

/// LSM-tree type
#[derive(Copy, Clone, Debug, PartialEq, Eq)]
pub enum TreeType {
    /// Standard LSM-tree, see [`Tree`]
    Standard,

    /// Key-value separated LSM-tree, see [`BlobTree`]
    Blob,
}

impl From<TreeType> for u8 {
    fn from(val: TreeType) -> Self {
        match val {
            TreeType::Standard => 0,
            TreeType::Blob => 1,
        }
    }
}

impl TryFrom<u8> for TreeType {
    type Error = ();

    fn try_from(value: u8) -> Result<Self, Self::Error> {
        match value {
            0 => Ok(Self::Standard),
            1 => Ok(Self::Blob),
            _ => Err(()),
        }
    }
}

#[cfg_attr(
    not(feature = "std"),
    allow(
        dead_code,
        reason = "default data-folder path used only on the std-gated default-config path"
    )
)]
const DEFAULT_FILE_FOLDER: &str = ".lsm.data";

/// Options for key-value separation
///
/// Blob compression is not among them: it changes on a live tree, so it lives in
/// [`RuntimeConfig::blob_compression`](crate::runtime_config::RuntimeConfig::blob_compression),
/// seeded by [`Config::blob_compression`].
#[derive(Clone, Debug, PartialEq)]
pub struct KvSeparationOptions {
    /// Blob file target size in bytes
    #[doc(hidden)]
    pub file_target_size: u64,

    /// Key-value separation threshold in bytes
    #[doc(hidden)]
    pub separation_threshold: u32,

    /// Separation thresholds of the fields of rows written as cells, as
    /// `(column, bytes)` pairs; a column without one uses
    /// `separation_threshold`. See [`Self::cell_separation_threshold`].
    #[doc(hidden)]
    pub cell_separation_thresholds: alloc::vec::Vec<(u16, u32)>,

    #[doc(hidden)]
    pub staleness_threshold: f32,

    #[doc(hidden)]
    pub age_cutoff: f32,

    /// How many upcoming values a scan reads ahead in one coalesced batch.
    ///
    /// `0` disables read-ahead. See
    /// [`scan_prefetch`](KvSeparationOptions::scan_prefetch).
    #[doc(hidden)]
    pub scan_prefetch: u16,

    /// Relocation of blob files for scan locality, off when `None`. See
    /// [`Self::relocate_for_locality`].
    #[doc(hidden)]
    pub locality_relocation: Option<BlobLocalityRelocation>,

    /// How many lifetime groups values are split into. See
    /// [`Self::lifetime_groups`].
    #[doc(hidden)]
    pub lifetime_groups: LifetimeGroups,

    /// The caller's lifetime class per key, over the observed one. See
    /// [`Self::lifetime_hint`].
    #[doc(hidden)]
    pub lifetime_hint: Option<LifetimeHint>,

    /// A zstd dictionary to register with the tree at open, for the blob
    /// compression to name. See [`Self::dict`].
    #[cfg(zstd_any)]
    #[doc(hidden)]
    pub zstd_dictionary: Option<alloc::sync::Arc<crate::compression::ZstdDictionary>>,
}

/// Whether a value read from a blob file goes into the shared cache.
///
/// Lookups are the same either way: a value already cached is served from
/// it. Set on a live tree through
/// [`RuntimeConfig::blob_cache_fill`](crate::runtime_config::RuntimeConfig::blob_cache_fill),
/// seeded by [`Config::blob_cache_fill`].
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub enum BlobCacheFill {
    /// Every value read from a blob file is cached, and a scan reads its
    /// values ahead into the cache.
    Always,
    /// No value read from a blob file is cached: values large enough to be
    /// separated are mostly read once, and cached they would evict the index,
    /// filter and data blocks of every tree sharing the cache. A scan still
    /// reads its values ahead, into a store of its own holding at most one
    /// coalesced read, from which it hands them out.
    #[default]
    Never,
}

impl BlobCacheFill {
    /// Whether a value read now goes into the cache.
    #[must_use]
    pub const fn fills(self) -> bool {
        matches!(self, Self::Always)
    }
}

/// When and how much a compaction relocates blob files to restore scan
/// locality. Set through [`KvSeparationOptions::relocate_for_locality`].
#[derive(Copy, Clone, Debug, PartialEq)]
pub struct BlobLocalityRelocation {
    /// The blob reference depth a compaction's inputs may reach before it
    /// relocates for locality. Relocation fires only above it.
    pub max_depth: core::num::NonZeroU64,

    /// The extra I/O relocation may add, as a fraction of the bytes the
    /// compaction writes anyway.
    pub budget: f32,
}

/// The threshold `thresholds` gives `column`, else `default`.
pub(crate) fn cell_threshold(thresholds: &[(u16, u32)], default: u32, column: u16) -> u32 {
    thresholds
        .iter()
        .find(|(set, _)| *set == column)
        .map_or(default, |(_, bytes)| *bytes)
}

impl Default for KvSeparationOptions {
    fn default() -> Self {
        Self {
            file_target_size: /* 64 MiB */ 64 * 1_024 * 1_024,
            separation_threshold: /* 1 KiB */ 1_024,
            cell_separation_thresholds: alloc::vec::Vec::new(),

            staleness_threshold: 0.25,
            age_cutoff: 0.25,

            scan_prefetch: 64,

            locality_relocation: None,

            lifetime_groups: LifetimeGroups::MAX,
            lifetime_hint: None,

            #[cfg(zstd_any)]
            zstd_dictionary: None,
        }
    }
}

impl KvSeparationOptions {
    /// Sets the target size of blob files.
    ///
    /// Smaller blob files allow more granular garbage collection
    /// which allows lower space amp for lower write I/O cost.
    ///
    /// Larger blob files decrease the number of files on disk and maintenance
    /// overhead.
    ///
    /// Defaults to 64 MiB.
    #[must_use]
    pub fn file_target_size(mut self, bytes: u64) -> Self {
        self.file_target_size = bytes;
        self
    }

    /// Sets the key-value separation threshold in bytes.
    ///
    /// Smaller value will reduce compaction overhead and thus write amplification,
    /// at the cost of lower read performance.
    ///
    /// Defaults to 1 KiB.
    #[must_use]
    pub fn separation_threshold(mut self, bytes: u32) -> Self {
        self.separation_threshold = bytes;
        self
    }

    /// Sets the separation threshold of `column` in rows written as cells
    /// (see [`BlobTree::insert_cells`](crate::BlobTree::insert_cells)): a
    /// value field of that column goes to a blob file when it is at least
    /// `bytes` long. Setting a column again replaces its threshold; a column
    /// never set uses [`Self::separation_threshold`].
    ///
    /// A field that is always read with the row's compact attributes can stay
    /// inline whatever its size (`u32::MAX`), and a field that is rarely read
    /// can go to a blob file however small (`0`).
    ///
    /// # Examples
    ///
    /// ```
    /// use lsm_tree::KvSeparationOptions;
    ///
    /// // Column 3 (status) never separates; column 4 (body) separates from 256 bytes.
    /// let opts = KvSeparationOptions::default()
    ///     .cell_separation_threshold(3, u32::MAX)
    ///     .cell_separation_threshold(4, 256);
    /// ```
    #[must_use]
    pub fn cell_separation_threshold(mut self, column: u16, bytes: u32) -> Self {
        match self
            .cell_separation_thresholds
            .iter_mut()
            .find(|(set, _)| *set == column)
        {
            Some(entry) => entry.1 = bytes,
            None => self.cell_separation_thresholds.push((column, bytes)),
        }
        self
    }

    /// What picks the lifetime class of a value written now.
    #[must_use]
    pub(crate) fn lifetime_classifier(&self) -> blob_lifetime::LifetimeClassifier<'_> {
        blob_lifetime::LifetimeClassifier {
            groups: self.lifetime_groups,
            hint: self.lifetime_hint.as_ref(),
        }
    }

    /// The separation threshold of the fields of `column`.
    #[must_use]
    pub(crate) fn cell_threshold(&self, column: u16) -> u32 {
        cell_threshold(
            &self.cell_separation_thresholds,
            self.separation_threshold,
            column,
        )
    }

    /// Sets how many upcoming values a scan reads ahead in one batch.
    ///
    /// A scan resolves separated values one at a time, and each is its own read
    /// of a few hundred bytes. Values sit in the blob file in the order the
    /// flush wrote them, which is key order, so a scan's next values are
    /// usually its immediate on-disk neighbours: reading a window of them at
    /// once collapses that stream of small reads into a few large ones.
    ///
    /// Read-ahead starts only once the scan's caller resolves a value
    /// unconditionally, so a scan that reads keys alone never pays for it, and
    /// neither does one that decides per key whether the value is worth
    /// reading. Larger windows amortize better on long scans; smaller ones
    /// waste less on a scan that stops early. `0` disables it.
    ///
    /// Applies to the sequential scans: `iter`, `range`, `prefix` and
    /// `batch_range_scan`. NOT to the seekable iterator, where a seek would
    /// throw away whatever was read ahead, and which exists for jumping around
    /// rather than walking a run.
    ///
    /// Defaults to 64.
    #[must_use]
    pub fn scan_prefetch(mut self, items: u16) -> Self {
        self.scan_prefetch = items;
        self
    }

    /// Lets compactions relocate blob files to restore scan locality.
    ///
    /// Values keep the order they were flushed in, so keys written in
    /// separate batches end up interleaved across many blob files, and a range
    /// scan alternates between all of them. A compaction whose input tables
    /// reference files interleaved deeper than `max_depth` (see
    /// [`BlobReferenceStats::depth`](crate::BlobReferenceStats::depth)) rewrites
    /// the deepest ones in key order, so each output file holds one run of
    /// adjacent keys. A table a strategy would only move to another level is
    /// merged instead when this applies to it, so the interleaving is not
    /// carried down unchanged. Files written in consecutive key runs have a
    /// depth of one and are never relocated for this, however many there are.
    ///
    /// Relocation copies values as they are stored, so files written under
    /// different blob codecs are rewritten into one file per codec and do not
    /// merge further; a file is relocated only together with another of its
    /// codec that it overlaps.
    ///
    /// Relocation is bounded three ways:
    /// - only whole files are rewritten, and only those no table outside the
    ///   compaction references;
    /// - reading and writing them together may cost at most `budget` times the
    ///   bytes the compaction writes anyway (its tables, plus the stale blob
    ///   files it already relocates, which are always served first);
    /// - nothing is relocated for locality when the output would not leave the
    ///   reserved free space intact.
    ///
    /// A `budget` of zero, negative or NaN relocates nothing. Off by default:
    /// compactions then behave and write exactly as without this option.
    ///
    /// # Examples
    ///
    /// ```
    /// use core::num::NonZeroU64;
    /// use lsm_tree::KvSeparationOptions;
    ///
    /// let depth = NonZeroU64::new(4).expect("non-zero");
    /// let opts = KvSeparationOptions::default().relocate_for_locality(depth, 0.5);
    /// assert!(opts.locality_relocation.is_some());
    /// ```
    #[must_use]
    pub fn relocate_for_locality(mut self, max_depth: core::num::NonZeroU64, budget: f32) -> Self {
        self.locality_relocation = Some(BlobLocalityRelocation { max_depth, budget });
        self
    }

    /// Splits blob values into `groups` by how long they are expected to
    /// live, each group written to its own blob files.
    ///
    /// A blob file is reclaimed as a whole, so a value that lives long and
    /// shares a file with values that are overwritten quickly is copied
    /// forward every time that file is collected. Grouping keeps the two apart:
    ///
    /// - group `0` takes values a flush finds overwritten within the memtable
    ///   it writes, which are likely to be overwritten again soon;
    /// - group `1` takes every other value written for the first time;
    /// - a value relocated out of a file moves one group up, having outlived
    ///   it, up to the last group.
    ///
    /// [`Self::lifetime_hint`] overrides the observed group per key. On Linux
    /// the device is also told each file's expected lifetime
    /// (`F_SET_RW_HINT`), which a flash device uses to keep data that dies
    /// together in the same erase blocks.
    ///
    /// The group only decides where a value is written. Which values are
    /// visible, and when a blob file may be removed, follows references and
    /// retention exactly as without grouping. Separating values by lifetime
    /// spreads keys that a range scan reads together over more files, so it
    /// trades scan locality for less relocation work.
    ///
    /// Defaults to [`LifetimeGroups::MAX`], four groups. [`LifetimeGroups::ONE`]
    /// turns grouping off: every value then shares the same files and no
    /// device hint is given. Any count from one to [`LifetimeGroups::MAX`] is
    /// accepted ([`LifetimeGroups::new`] refuses the rest), and a flush or a
    /// relocation keeps at most that many files open per blob codec.
    ///
    /// The setting applies to the files written from now on and may change
    /// between opens: a file keeps the class it was written with, and a class
    /// above the configured groups counts as the last group.
    ///
    /// # Examples
    ///
    /// ```
    /// use lsm_tree::{KvSeparationOptions, config::LifetimeGroups};
    ///
    /// // Grouped by default.
    /// assert_eq!(
    ///     KvSeparationOptions::default().lifetime_groups,
    ///     LifetimeGroups::MAX,
    /// );
    ///
    /// // Fewer groups, or none at all.
    /// let groups = LifetimeGroups::new(2).expect("within the bound");
    /// let two = KvSeparationOptions::default().lifetime_groups(groups);
    /// assert_eq!(two.lifetime_groups.get(), 2);
    /// let off = KvSeparationOptions::default().lifetime_groups(LifetimeGroups::ONE);
    /// assert_eq!(off.lifetime_groups.get(), 1);
    /// ```
    #[must_use]
    pub fn lifetime_groups(mut self, groups: LifetimeGroups) -> Self {
        self.lifetime_groups = groups;
        self
    }

    /// Sets the caller's estimate of each key's lifetime class, used over
    /// the observed one wherever it answers (see [`Self::lifetime_groups`]).
    ///
    /// `hint` returns the class for a key, `0` for the shortest-lived; a class
    /// past the last group goes to the last group, and `None` leaves the key
    /// to the observed estimate. The hint is advisory: a wrong one costs
    /// relocation work and never changes what a read returns. It has no effect
    /// with a single group.
    ///
    /// # Examples
    ///
    /// ```
    /// use lsm_tree::{
    ///     KvSeparationOptions,
    ///     config::{LifetimeGroups, LifetimeHint},
    /// };
    ///
    /// // Session payloads churn, archived objects stay.
    /// let hint = LifetimeHint::new(|key: &[u8]| {
    ///     if key.starts_with(b"session/") {
    ///         Some(0)
    ///     } else if key.starts_with(b"archive/") {
    ///         Some(3)
    ///     } else {
    ///         None
    ///     }
    /// });
    /// let opts = KvSeparationOptions::default()
    ///     .lifetime_groups(LifetimeGroups::MAX)
    ///     .lifetime_hint(hint);
    /// assert!(opts.lifetime_hint.is_some());
    /// ```
    #[must_use]
    pub fn lifetime_hint(mut self, hint: LifetimeHint) -> Self {
        self.lifetime_hint = Some(hint);
        self
    }

    /// Sets the staleness threshold percentage.
    ///
    /// The staleness percentage determines how much a blob file needs to be fragmented to be
    /// picked up by the garbage collection.
    ///
    /// Defaults to 33%.
    #[must_use]
    pub fn staleness_threshold(mut self, ratio: f32) -> Self {
        self.staleness_threshold = ratio;
        self
    }

    /// Sets the age cutoff threshold.
    ///
    /// Defaults to 20%.
    #[must_use]
    pub fn age_cutoff(mut self, ratio: f32) -> Self {
        self.age_cutoff = ratio;
        self
    }

    /// Supplies a zstd dictionary for the blob compression to name.
    ///
    /// The open registers it with the tree, which stores it, so a later open
    /// needs nothing supplied; this is also how the bytes reach a tree the
    /// first time. Which dictionary blob files are written against is decided
    /// by [`Config::blob_compression`]: a [`CompressionType::ZstdDict`] there
    /// names one by id, and [`Config::open`] refuses one the tree does not hold
    /// with [`Error::ZstdDictMismatch`](crate::Error::ZstdDictMismatch).
    #[cfg(zstd_any)]
    #[must_use]
    pub fn dict(
        mut self,
        dictionary: alloc::sync::Arc<crate::compression::ZstdDictionary>,
    ) -> Self {
        self.zstd_dictionary = Some(dictionary);
        self
    }
}

/// Where a compaction ends its outputs.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) enum OutputCuts {
    /// On the boundaries between the tables of the level the outputs are
    /// merged into next, within half the target size and twice it.
    Aligned,
    /// At the target size alone.
    TargetSize,
}

/// Tree configuration builder
// Clone: every shared handle is `Arc`-backed, so a clone is a cheap second
// reference to the same backends — which is what lets `open_or_repair` retry
// an `open(self)` after a repair.
#[derive(Clone)]
pub struct Config {
    /// Folder path
    #[doc(hidden)]
    pub path: PathBuf,

    /// Default filesystem backend for levels without an explicit route.
    ///
    /// Defaults to `IoUringFs` where the kernel offers it, else
    /// [`StdFs`](crate::fs::StdFs). Use [`Config::with_fs`] to plug in an
    /// alternative backend such as [`MemFs`](crate::fs::MemFs).
    ///
    /// Both fresh tree creation and reopening (recovery) are supported
    /// for any backend that implements [`Fs`].
    #[doc(hidden)]
    pub fs: Arc<dyn Fs>,

    /// Per-level filesystem routing for tiered storage.
    ///
    /// When set, tables at different LSM levels can be stored on different
    /// storage devices (e.g., NVMe for L0–L1, SSD for L2–L4, HDD for L5–L6).
    /// Each entry maps a range of levels to a base directory and filesystem
    /// backend. Uncovered levels fall back to the primary `path` and `fs`.
    ///
    /// Zero additional overhead when `None` — only a single branch check;
    /// path construction allocations are unchanged.
    #[doc(hidden)]
    pub level_routes: Option<Vec<LevelRoute>>,

    /// Block cache to use
    #[doc(hidden)]
    pub cache: Arc<Cache>,

    /// Descriptor table to use
    #[doc(hidden)]
    pub descriptor_table: Option<Arc<DescriptorTable>>,

    /// Number of levels of the LSM tree (depth of tree)
    ///
    /// Once set, the level count is fixed (in the manifest snapshot `v<id>`)
    pub level_count: u8,

    /// Restart interval inside data blocks
    pub data_block_restart_interval_policy: RestartIntervalPolicy,

    /// Restart interval inside index blocks
    pub index_block_restart_interval_policy: RestartIntervalPolicy,

    /// Block size of data blocks
    pub data_block_size_policy: BlockSizePolicy,

    /// Uncompressed size a columnar table's row groups are cut at, per level:
    /// the columnar counterpart of [`Self::data_block_size_policy`], which a
    /// columnar table does not use for its data. A row group is split into
    /// independently readable column pages, so it is sized for what a read of
    /// one page is worth rather than for what a read of the whole group costs.
    pub columnar_row_group_size_policy: BlockSizePolicy,

    /// Uncompressed bytes of row data a columnar row page holds, per level.
    /// A row group's rows are cut into row pages shared by every column, so a
    /// read of a few rows decodes only the pages that hold them.
    pub columnar_page_size_policy: BlockSizePolicy,

    /// How a columnar table's column pages store their values, per level;
    /// see [`ColumnEncoding`].
    pub column_encoding_policy: ColumnEncodingPolicy,

    /// How columnar reads of this tree fetch their pages; see [`ReadBudget`].
    pub columnar_read_budget: ReadBudget,

    /// The page bytes one projected columnar scan holds at once, shared by
    /// the segments it merges; see [`Self::columnar_scan_budget`].
    pub columnar_scan_budget: u64,

    /// The filter and index bytes a multi-get holds at once across the tables
    /// of a level it reads stage by stage; see
    /// [`Self::multi_get_metadata_budget`].
    pub multi_get_metadata_budget: u64,

    /// Whether to pin index blocks
    pub index_block_pinning_policy: PinningPolicy,

    /// Whether to pin filter blocks
    pub filter_block_pinning_policy: PinningPolicy,

    /// Whether to pin top level index of partitioned index
    pub top_level_index_block_pinning_policy: PinningPolicy,

    /// Whether to pin top level index of partitioned filter
    pub top_level_filter_block_pinning_policy: PinningPolicy,

    /// Data block hash ratio
    pub data_block_hash_ratio_policy: HashRatioPolicy,

    /// Whether to partition index blocks
    pub index_block_partitioning_policy: PartitioningPolicy,

    /// Whether to partition filter blocks
    pub filter_block_partitioning_policy: PartitioningPolicy,

    /// Partition size when using partitioned indexes, per level; see
    /// [`Self::index_block_partition_size_policy`].
    pub index_block_partition_size_policy: BlockSizePolicy,

    /// Partition size when using partitioned filters, per level; see
    /// [`Self::filter_block_partition_size_policy`].
    pub filter_block_partition_size_policy: BlockSizePolicy,

    /// If `true`, the last level will not build filters, reducing the filter size of a database
    /// by ~90% typically
    pub(crate) expect_point_read_hits: bool,

    /// Where a compaction ends its outputs; see
    /// [`Self::compaction_output_alignment`].
    pub(crate) compaction_output_cuts: OutputCuts,

    /// Per-block Page ECC. When `true`, every block on disk carries a parity
    /// trailer; on read, if the block's XXH3 disagrees with the on-disk bytes,
    /// the reader attempts recovery from the trailer before surfacing the
    /// corruption. The correction scheme is selected at runtime
    /// (`update_runtime_config`): per-word SEC-DED (the default), single XOR
    /// parity, or Reed-Solomon. Requires the `page_ecc` cargo feature — opening a
    /// tree with `page_ecc = true` on a build without the feature returns
    /// [`crate::Error::PageEccUnsupported`].
    ///
    /// Off by default. `RocksDB` ships per-block ECC as an operator-
    /// chosen knob (typically off on RAID-protected media, on on
    /// single-drive) and the cost is non-trivial on the write path,
    /// so the default keeps the existing behaviour.
    pub(crate) page_ecc: bool,

    /// Initial [`crate::runtime_config::RuntimeConfig`] snapshot
    /// the tree starts with. Seeds both the first
    /// `persist_version` call and the Tree's
    /// `RuntimeConfigHandle`, so a non-default value supplied via
    /// [`Config::with_runtime_config`] is honoured from byte zero
    /// of the manifest. Defaults to `RuntimeConfig::default()` —
    /// matches the pre-existing implicit behaviour.
    #[expect(
        clippy::struct_field_names,
        reason = "name mirrors the type for grep-ability across the persist + Tree handle init wiring"
    )]
    pub(crate) initial_runtime_config: crate::runtime_config::RuntimeConfig,

    /// Filter construction policy
    pub filter_policy: FilterPolicy,

    /// Allocates filter memory by measured negative-probe load; `None` (the
    /// default) leaves every filter to [`Self::filter_policy`].
    pub filter_advisor: Option<FilterAdvisor>,

    /// Retrieval-ribbon locator policy (per level). Defaults to
    /// [`LocatorPolicy::block_level`]: written SSTs carry an optional `locator`
    /// section mapping each key to its data block for O(1) point reads (skipping
    /// the index-block binary search). Set [`LocatorPolicy::disabled`] to opt
    /// out — disabled levels produce byte-identical SSTs (no section).
    pub locator_policy: LocatorPolicy,

    /// Compaction filter factory
    pub compaction_filter_factory: Option<Arc<dyn Factory>>,

    /// Prefix extractor for prefix bloom filters.
    ///
    /// When set, the bloom filter indexes extracted prefixes in addition to
    /// full keys, allowing prefix scans to skip segments that contain no
    /// matching prefixes.
    pub prefix_extractor: Option<Arc<dyn PrefixExtractor>>,

    /// Key domains no table spans (see [`Config::key_domains`]).
    pub key_domains: Option<Arc<dyn crate::KeyDomains>>,

    /// Merge operator for commutative operations
    ///
    /// When set, enables `merge()` operations that store partial updates
    /// which are lazily combined during reads and compaction.
    pub merge_operator: Option<Arc<dyn MergeOperator>>,

    #[doc(hidden)]
    pub kv_separation_opts: Option<KvSeparationOptions>,

    /// Custom user key comparator.
    ///
    /// When set, all key comparisons use this comparator instead of the
    /// default lexicographic byte ordering. Once a tree is opened with a
    /// comparator, it must always be re-opened with the same comparator.
    // Not `pub` — use `Config::comparator()` builder method as the public API.
    #[doc(hidden)]
    pub(crate) comparator: SharedComparator,

    /// Block-level encryption provider for encryption at rest.
    ///
    /// When set, all blocks (data, index, filter, meta) are encrypted
    /// using this provider after compression and before checksumming.
    pub(crate) encryption: Option<Arc<dyn EncryptionProvider>>,

    /// The wall clock of this tree; `None` reads the process clock (see
    /// [`Config::with_clock`]).
    pub(crate) clock: Option<Arc<dyn crate::Clock>>,

    /// Policy governing what `Tree::open` does when the on-disk
    /// MANIFEST contains corrupt records. Defaults to
    /// [`ManifestRecoveryMode::AbsoluteConsistency`], the only
    /// production-safe choice — any corruption aborts the open. The only
    /// other mode, [`ManifestRecoveryMode::TolerateCorruptedTailRecords`],
    /// additionally drops a torn edit-log tail (an append the writer never
    /// finished); damage to anything committed fails the open in both.
    pub(crate) manifest_recovery_mode: ManifestRecoveryMode,

    /// Durability level for every fsync the tree issues (SST writes,
    /// manifest, version persist, directory syncs).
    ///
    /// Defaults to [`SyncMode::Normal`] (plain `fsync`), matching the
    /// out-of-the-box durability of `RocksDB` and `SQLite`. Only observable on
    /// macOS, where [`SyncMode::Full`] opts into the much slower
    /// `F_FULLFSYNC` barrier and [`SyncMode::Barrier`] into ordered syncs;
    /// on other platforms all modes are plain `fsync`. Set via
    /// [`Config::sync_mode`].
    pub(crate) sync_mode: SyncMode,

    /// When a flush's tables become durable. Set via
    /// [`Config::flush_durability`].
    pub(crate) flush_durability: FlushDurability,

    /// Bytes a table or blob file gathers between the writebacks its writer
    /// starts. Set via [`Config::writeback_bytes`].
    pub(crate) writeback_bytes: u64,

    /// Size a flush or an ingestion cuts its output tables at. Set via
    /// [`Config::table_target_size`].
    pub(crate) table_target_size: u64,

    /// When `true` (the default), [`Config::open`] and [`Config::repair`]
    /// acquire an exclusive cross-process lock on a `LOCK` file in the tree
    /// directory (an advisory OS file lock) and hold it for the lifetime of the
    /// [`Tree`] (open) or the duration of the call (repair). A
    /// second process attempting to open / repair the same directory fails fast
    /// with [`Error::Locked`](crate::Error::Locked) instead of racing on the
    /// manifest. Set `false` via [`Config::with_directory_lock`] only when the
    /// embedder already enforces exclusive directory ownership at a higher layer
    /// (e.g. a keyspace / journal manager). Best-effort per `Fs` backend: real
    /// on-disk backends enforce it, in-memory backends are single-process and
    /// satisfy it vacuously.
    pub(crate) directory_lock: bool,

    /// Shared live-progress counters the repair / salvage paths tick while
    /// they run, or `None` (the default) to skip publishing. Set via
    /// [`Config::with_recovery_progress`]; observed by polling
    /// [`RecoveryProgress::snapshot`](crate::RecoveryProgress::snapshot) from
    /// another thread.
    #[cfg(feature = "std")]
    pub(crate) recovery_progress: Option<Arc<crate::RecoveryProgress>>,

    /// Edit-log size (bytes) past which the next manifest persist rotates: it
    /// writes a fresh full snapshot and starts an empty log instead of appending
    /// another [`VersionEdit`](crate::version::edit::VersionEdit). Bounds both
    /// recovery replay time (edits to re-apply) and log disk use, while keeping
    /// the common per-flush path a tiny `O(changed-levels)` append rather than an
    /// `O(all-SSTs)` full manifest rewrite.
    ///
    /// Defaults to 1 MiB (≈ tens of thousands of edits). Set via
    /// [`Config::manifest_log_rotate_bytes`]. A smaller value rotates more
    /// often (shorter recovery, more frequent full-snapshot writes); `0` rotates
    /// on every upgrade, degenerating to the full-rewrite-per-version behaviour.
    pub(crate) manifest_log_rotate_bytes: u64,

    /// Empty table files each tables folder creates at a time, ahead of the
    /// tables written into them, with one folder sync for the batch. Set via
    /// [`Config::table_file_pool`].
    pub(crate) table_file_pool: usize,

    /// Where a table file pool refills: off the claiming write's thread, or
    /// on it, as it does without `std`. Set via
    /// [`Config::table_file_refill_inline`].
    pub(crate) table_file_refill: crate::table::file_pool::RefillOn,

    /// Retention floor a manifest REPAIR seeds the rebuilt version with: the
    /// highest snapshot seqno the repaired tree refuses to serve. The lost
    /// manifest was the only record of the floor a past GC compaction or
    /// `clear` established, and the tables cannot stand in for it, so the
    /// deployment that ran those compactions (and knows the watermark it
    /// passed) supplies it here. Defaults to `0`: a repaired tree serves
    /// every snapshot, which is right only if history was never collected.
    /// Set via [`Config::repair_retention_floor`].
    pub(crate) repair_retention_floor: crate::SeqNo,

    /// Compaction I/O rate limit in bytes per second.
    ///
    /// Caps the rate at which the compaction worker is allowed to issue
    /// I/O, so background compaction cannot saturate the device and starve
    /// user point reads / range scans (P99 stability). `0` (the default)
    /// means unlimited — no throttling, no behaviour change. Flush and
    /// user reads are never throttled, only compaction. Set via
    /// [`Config::compaction_rate_limit`].
    pub(crate) compaction_rate_limit: u64,

    /// Optional shared compaction rate limiter. `None` (default) = the tree
    /// builds its own from [`Self::compaction_rate_limit`]. `Some` = every
    /// tree holding this `Arc` is bounded by it together, at the rate it
    /// holds. Set via [`Config::compaction_rate_limiter`].
    pub(crate) compaction_rate_limiter: Option<Arc<crate::rate_limiter::RateLimiter>>,

    /// The limiter the open tree throttles its compactions with, set at open
    /// for a strategy to read the live rate. Apart from
    /// [`Self::compaction_rate_limiter`], so a copy of the tree's
    /// configuration opened as another tree does not share it.
    pub(crate) live_compaction_rate_limiter: Option<Arc<crate::rate_limiter::RateLimiter>>,

    /// Optional pool compactions reserve their room in, shared with other
    /// trees. `None` (default) = the tree keeps a pool of its own. Set via
    /// [`Config::compaction_capacity`].
    pub(crate) compaction_capacity: Option<Arc<crate::CapacityPool>>,

    /// Bytes per second writes are paced to while the backpressure verdict is
    /// in its slowdown tier; `0` (the default) leaves the slowdown delay to
    /// the ramp. Set via [`Config::write_rate_limit`].
    pub(crate) write_rate_limit: u64,

    /// Optional write rate limiter shared with other trees, used in place of
    /// the per-tree one built from [`Self::write_rate_limit`]. Set via
    /// [`Config::write_rate_limiter`].
    pub(crate) write_rate_limiter: Option<Arc<crate::rate_limiter::RateLimiter>>,

    /// Optional write-buffer budget shared with other trees, which the
    /// memtable axis of the backpressure verdict compares in place of this
    /// tree's own unflushed bytes. Set via [`Config::write_buffer_budget`].
    // no-std: the budget reads members through their std-only arc-swap mirror
    #[cfg(feature = "std")]
    pub(crate) write_buffer_budget: Option<Arc<crate::backpressure::WriteBufferBudget>>,

    /// Worker-thread count for compaction parallelism (`std` only), used two
    /// ways: it sizes the per-tree pool built at open when
    /// [`Self::compaction_pool`] is `None`, which prepares data blocks and
    /// builds partitioned-filter partitions, and it caps how many range-parallel
    /// sub-compactions a single compaction is split into. Default
    /// `max(1, available_parallelism / 2)` — leaves half the cores for
    /// application work. `1` forces the serial path for both. Without the
    /// `parallel` feature there is no built-in pool, so block compression and
    /// sub-compaction ranges run serially even for a value > 1. Set via
    /// [`Config::compaction_threads`].
    #[cfg(feature = "std")] // no-std: parallel compaction unavailable (no threads)
    pub(crate) compaction_threads: usize,

    /// Optional shared compaction thread pool. `None` (default) = a per-tree
    /// pool is built at [`crate::Tree::open`] sized by [`Self::compaction_threads`]
    /// (predictable, matches the per-DB pattern). `Some` = caller-supplied
    /// executor shared across every tree holding this `Arc`, bounding total
    /// threads regardless of tree count. Set via [`Config::compaction_pool`].
    #[cfg(feature = "std")]
    pub(crate) compaction_pool: Option<Arc<dyn crate::table::writer::CompactionSpawner>>,

    /// Payload size (bytes) below which a block of a parallel-compressed table
    /// is prepared on the writer thread instead of a worker: its transform
    /// costs less than handing it to a worker and back. `None` (default)
    /// derives it from the table's block codec. Set via
    /// [`Config::parallel_compression_inline_below`].
    #[cfg(feature = "std")]
    pub(crate) parallel_compression_inline_below: Option<u32>,

    /// Minimum total input size (bytes) for a compaction to be split into
    /// parallel sub-compactions. Below it the compaction stays single-threaded
    /// (per-thread setup + extra output tables outweigh the parallelism on small
    /// compactions). Default
    /// [`SUBCOMPACTION_MIN_INPUT_BYTES`](crate::compaction::worker::SUBCOMPACTION_MIN_INPUT_BYTES)
    /// (8 MiB). Set via [`Config::subcompaction_min_bytes`].
    #[cfg(feature = "std")]
    pub(crate) subcompaction_min_bytes: u64,

    /// Test-only failpoint: when armed, the first parallel sub-compaction range
    /// that observes it returns an error and disarms it, so the crash-safety
    /// rollback paths (sibling output rollback, input restore) can be exercised
    /// deterministically. Behind `cfg(test)`, never compiled into release builds.
    #[cfg(all(test, feature = "std"))]
    pub(crate) fail_one_subcompaction: Arc<core::sync::atomic::AtomicBool>,

    /// Test-only failpoint: when armed, the serial merge raises the stop signal
    /// after writing its first item and disarms itself, so the interrupted-merge
    /// path is reachable without racing a tree drop against an in-flight
    /// compaction. Behind `cfg(test)`, never compiled into release builds.
    #[cfg(all(test, feature = "std"))]
    pub(crate) stop_serial_merge_after_first_item: Arc<core::sync::atomic::AtomicBool>,

    /// Test-only failpoint: when armed, a tight-space compaction returns an error
    /// immediately after durably installing (and punching) its FIRST slice, so
    /// the crash-mid-loop recovery path (reopen a tree whose manifest carries a
    /// persisted input restriction) can be exercised deterministically. Behind
    /// `cfg(test)`, never compiled into release builds.
    #[cfg(all(test, feature = "std"))]
    pub(crate) fail_tight_after_first_slice: Arc<core::sync::atomic::AtomicBool>,

    /// Test-only: bytes another tree sharing the capacity pool takes on the
    /// destination volume right before a tight-space pass reserves its first
    /// slice, after the pass measured the room. Behind `cfg(test)`.
    #[cfg(all(test, feature = "std"))]
    pub(crate) tight_slice_squeeze: Arc<core::sync::atomic::AtomicU64>,

    /// Test-only failpoint: when armed, a tight-space relocation fails at the
    /// restricted-blob reopen step of its current slice — after the slice's
    /// outputs were finalized but before the install — so the pre-install
    /// rollback (retract the finalized-but-unreferenced outputs) can be
    /// exercised deterministically. Behind `cfg(test)`, never compiled into
    /// release builds.
    #[cfg(all(test, feature = "std"))]
    pub(crate) fail_tight_blob_reopen: Arc<core::sync::atomic::AtomicBool>,

    /// Test-only failpoint: run once, then disarmed, between the moment a write
    /// has finished its output files and the moment it installs them (a flush,
    /// an ingestion, a parallel or tight-space compaction), so a test can put a
    /// concurrent operation into that window deterministically. Behind
    /// `cfg(test)`, never compiled into release builds.
    #[cfg(all(test, feature = "std"))]
    pub(crate) before_output_install: Arc<std::sync::Mutex<Option<OutputInstallHook>>>,

    /// Test-only failpoint: run once, then disarmed, while a compaction
    /// finishes its outputs, after their last write and before their reopen.
    #[cfg(all(test, feature = "std"))]
    pub(crate) while_outputs_finish: Arc<std::sync::Mutex<Option<OutputInstallHook>>>,

    /// A dictionary supplied for the data block policy to name. The open
    /// registers it with the tree, so it survives the reopen without being
    /// supplied again.
    ///
    /// Supply only: which dictionary new blocks are written against is the
    /// policy's to say, by id, and a writer resolves that id from
    /// [`Self::zstd_dictionaries`] when it starts. A slot filled at open could
    /// not follow a policy that changes on a live tree.
    #[cfg(zstd_any)]
    pub(crate) zstd_dictionary: Option<Arc<crate::compression::ZstdDictionary>>,

    /// Every dictionary the tree can decompress against: the ones it has
    /// registered, plus [`Self::zstd_dictionary`] when one was supplied.
    ///
    /// Populated at open from the tree's own `dicts/` folder, so a reopen needs
    /// no dictionary from the caller at all. A table resolves the id it
    /// recorded against this set rather than being compared to a single
    /// configured dictionary, which is what lets one tree hold data written
    /// under more than one.
    ///
    /// Shared and swappable, because registration happens on a LIVE tree: a
    /// plain field would leave a freshly registered dictionary unresolvable
    /// until the next open, so the very tables the registration was made for
    /// would fail to read.
    #[cfg(zstd_any)]
    pub(crate) zstd_dictionaries: Arc<arc_swap::ArcSwap<crate::compression::ZstdDictionaries>>,

    /// The global sequence number generator.
    ///
    /// Should be shared between multiple trees of a database.
    pub(crate) seqno: SharedSequenceNumberGenerator,

    /// Sequence number watermark that is visible to readers.
    ///
    /// Used for MVCC snapshots and to control which updates are
    /// observable in a given view of the database.
    pub(crate) visible_seqno: SharedSequenceNumberGenerator,
}

// TODO: remove default?
// std-only: the default backends are std-bound and the default path is resolved
// via std::path::absolute. no_std callers construct `Config` explicitly with a
// caller-provided `Fs`.
#[cfg(feature = "std")]
impl Default for Config {
    fn default() -> Self {
        Self {
            path: absolute_path(Path::new(DEFAULT_FILE_FOLDER)),
            fs: crate::fs::default_fs(),
            level_routes: None,
            descriptor_table: Some(Arc::new(DescriptorTable::new(256))),
            seqno: SharedSequenceNumberGenerator::from(SequenceNumberCounter::default()),
            visible_seqno: SharedSequenceNumberGenerator::from(SequenceNumberCounter::default()),

            cache: Arc::new(Cache::with_capacity_bytes(
                /* 16 MiB */ 16 * 1_024 * 1_024,
            )),

            data_block_restart_interval_policy: RestartIntervalPolicy::all(16),
            index_block_restart_interval_policy: RestartIntervalPolicy::all(1),

            level_count: DEFAULT_LEVEL_COUNT,

            data_block_size_policy: BlockSizePolicy::all(4_096),

            columnar_row_group_size_policy: BlockSizePolicy::all(DEFAULT_COLUMNAR_ROW_GROUP_SIZE),

            columnar_page_size_policy: BlockSizePolicy::all(DEFAULT_COLUMNAR_PAGE_SIZE),

            column_encoding_policy: ColumnEncodingPolicy::default(),

            columnar_read_budget: ReadBudget::default(),
            columnar_scan_budget: DEFAULT_COLUMNAR_SCAN_BUDGET,
            multi_get_metadata_budget: DEFAULT_MULTI_GET_METADATA_BUDGET,

            index_block_pinning_policy: PinningPolicy::new([true, true, false]),
            filter_block_pinning_policy: PinningPolicy::new([true, false]),

            top_level_index_block_pinning_policy: PinningPolicy::all(true), // TODO: implement
            top_level_filter_block_pinning_policy: PinningPolicy::all(true), // TODO: implement

            // Partitioned at every level so a bit-flip inside one
            // sub-index block only takes out the keys covered by that
            // partition, not the entire SST. A full-index SST has no
            // within-block redundancy: one corrupt byte in the single
            // index block makes every data block in the table
            // unreachable. See tests/partitioned_index_blast_radius.rs
            // for the isolation property this default relies on.
            index_block_partitioning_policy: PinningPolicy::all(true),
            // Filter-block default intentionally left at the pre-#329
            // shape (L3+ only). A corrupt filter block can produce a
            // false negative (filter says "not present" → read short-
            // circuits → caller misses an existing key), which is a
            // correctness hazard distinct from index corruption (where
            // the read fails loudly). Flipping this default is tracked
            // as a separate decision pending a filter blast-radius /
            // false-negative analysis; symmetry with index is not
            // sufficient justification on its own.
            filter_block_partitioning_policy: PinningPolicy::new([false, false, false, true]),

            index_block_partition_size_policy: BlockSizePolicy::all(4_096),
            filter_block_partition_size_policy: BlockSizePolicy::all(4_096),

            data_block_hash_ratio_policy: HashRatioPolicy::all(0.0),

            locator_policy: LocatorPolicy::block_level(),
            filter_policy: FilterPolicy::all(FilterPolicyEntry::Bloom(
                BloomConstructionPolicy::BitsPerKey(10.0),
            )),
            filter_advisor: None,

            compaction_filter_factory: None,
            merge_operator: None,

            prefix_extractor: None,
            key_domains: None,

            expect_point_read_hits: false,

            compaction_output_cuts: OutputCuts::Aligned,

            page_ecc: false,

            initial_runtime_config: crate::runtime_config::RuntimeConfig::default(),

            kv_separation_opts: None,

            #[cfg(zstd_any)]
            zstd_dictionary: None,
            #[cfg(zstd_any)]
            zstd_dictionaries: Arc::new(arc_swap::ArcSwap::from_pointee(
                crate::compression::ZstdDictionaries::new(),
            )),

            comparator: comparator::default_comparator(),
            encryption: None,
            clock: None,
            manifest_recovery_mode: ManifestRecoveryMode::AbsoluteConsistency,
            sync_mode: SyncMode::Normal,
            flush_durability: FlushDurability::Synced,
            writeback_bytes: 1_024 * 1_024,
            table_target_size: DEFAULT_TABLE_TARGET_SIZE,
            directory_lock: true,
            #[cfg(feature = "std")]
            recovery_progress: None,
            manifest_log_rotate_bytes: 1024 * 1024,
            table_file_pool: crate::table::file_pool::DEFAULT_TABLE_FILE_POOL,
            table_file_refill: crate::table::file_pool::RefillOn::Background,
            repair_retention_floor: 0,
            compaction_rate_limit: 0,
            compaction_rate_limiter: None,
            live_compaction_rate_limiter: None,
            compaction_capacity: None,
            write_rate_limit: 0,
            write_rate_limiter: None,
            #[cfg(feature = "std")]
            write_buffer_budget: None,

            #[cfg(feature = "std")]
            compaction_threads: std::thread::available_parallelism()
                .map_or(1, |n| (n.get() / 2).max(1)),
            #[cfg(feature = "std")]
            compaction_pool: None,
            #[cfg(feature = "std")]
            parallel_compression_inline_below: None,
            #[cfg(feature = "std")]
            subcompaction_min_bytes: crate::compaction::worker::SUBCOMPACTION_MIN_INPUT_BYTES,
            #[cfg(all(test, feature = "std"))]
            fail_one_subcompaction: Arc::new(core::sync::atomic::AtomicBool::new(false)),
            #[cfg(all(test, feature = "std"))]
            stop_serial_merge_after_first_item: Arc::new(core::sync::atomic::AtomicBool::new(
                false,
            )),
            #[cfg(all(test, feature = "std"))]
            fail_tight_after_first_slice: Arc::new(core::sync::atomic::AtomicBool::new(false)),
            #[cfg(all(test, feature = "std"))]
            tight_slice_squeeze: Arc::new(core::sync::atomic::AtomicU64::new(0)),
            #[cfg(all(test, feature = "std"))]
            fail_tight_blob_reopen: Arc::new(core::sync::atomic::AtomicBool::new(false)),
            #[cfg(all(test, feature = "std"))]
            before_output_install: Arc::new(std::sync::Mutex::new(None)),
            #[cfg(all(test, feature = "std"))]
            while_outputs_finish: Arc::new(std::sync::Mutex::new(None)),
        }
    }
}

/// The callback the [`Config::before_output_install`] failpoint runs.
#[cfg(all(test, feature = "std"))]
pub(crate) type OutputInstallHook = Box<dyn FnOnce() + Send>;

#[cfg(all(test, feature = "std"))]
impl Config {
    /// Arms the [`Self::before_output_install`] failpoint with `hook`.
    pub(crate) fn arm_before_output_install(&self, hook: impl FnOnce() + Send + 'static) {
        #[expect(clippy::unwrap_used, reason = "test-only seam")]
        self.before_output_install
            .lock()
            .unwrap()
            .replace(Box::new(hook));
    }

    /// Runs the [`Self::before_output_install`] failpoint once, if armed.
    pub(crate) fn fire_before_output_install(&self) {
        #[expect(clippy::unwrap_used, reason = "test-only seam")]
        let hook = self.before_output_install.lock().unwrap().take();
        if let Some(hook) = hook {
            hook();
        }
    }

    /// Arms the [`Self::while_outputs_finish`] failpoint with `hook`.
    pub(crate) fn arm_while_outputs_finish(&self, hook: impl FnOnce() + Send + 'static) {
        #[expect(clippy::unwrap_used, reason = "test-only seam")]
        self.while_outputs_finish
            .lock()
            .unwrap()
            .replace(Box::new(hook));
    }

    /// Runs the [`Self::while_outputs_finish`] failpoint once, if armed.
    pub(crate) fn fire_while_outputs_finish(&self) {
        #[expect(clippy::unwrap_used, reason = "test-only seam")]
        let hook = self.while_outputs_finish.lock().unwrap().take();
        if let Some(hook) = hook {
            hook();
        }
    }
}

/// Name of the lock file created in a tree directory for the cross-process
/// exclusive directory lock.
#[cfg_attr(
    not(feature = "std"),
    allow(
        dead_code,
        reason = "directory-lock filename used only by the std-gated lock-acquisition path"
    )
)]
pub(crate) const DIRECTORY_LOCK_FILE: &str = "LOCK";

/// How an open comes by the directory lock.
#[cfg(feature = "std")]
pub(crate) enum DirectoryLock {
    /// The open takes it.
    Acquire,
    /// The caller already holds it (`None` when the config disables the lock),
    /// having written the tree's files under it.
    Held(Option<Box<dyn crate::fs::FsFile>>),
}

/// Acquires the cross-process exclusive directory lock when `enabled`.
///
/// Opens (creating if absent) a `LOCK` file under `dir` and takes a
/// non-blocking exclusive advisory lock on it through the `Fs` backend. Returns
/// the locked handle to hold for as long as exclusivity is required; dropping it
/// releases the lock (the OS frees an advisory lock when the fd / handle
/// closes). `Ok(None)` when `enabled` is false. Fails with
/// [`Error::Locked`](crate::Error::Locked) when another live instance holds the
/// lock. The directory must already exist (the caller creates it for a fresh
/// tree before acquiring).
#[cfg(feature = "std")]
pub(crate) fn acquire_directory_lock(
    fs: &dyn Fs,
    dir: &Path,
    enabled: bool,
) -> crate::Result<Option<Box<dyn crate::fs::FsFile>>> {
    if !enabled {
        return Ok(None);
    }
    let lock_path = dir.join(DIRECTORY_LOCK_FILE);
    let file = fs.open(
        &lock_path,
        &crate::fs::FsOpenOptions::new()
            .read(true)
            .write(true)
            .create(true)
            .write_origin(crate::fs::WriteOrigin::Metadata),
    )?;
    if file.try_lock_exclusive()? {
        Ok(Some(file))
    } else {
        Err(crate::Error::Locked(dir.display().to_string()))
    }
}

impl Config {
    /// Initializes a new config
    // std-only: seeds the remaining fields from `Config::default`, whose
    // default `Fs` is `StdFs`. no_std callers build `Config` field-by-field
    // with a caller-provided `Fs`.
    #[cfg(feature = "std")]
    pub fn new<P: AsRef<Path>>(
        path: P,
        seqno: SequenceNumberCounter,
        visible_seqno: SequenceNumberCounter,
    ) -> Self {
        Self {
            path: absolute_path(path.as_ref()),
            seqno: Arc::new(seqno),
            visible_seqno: Arc::new(visible_seqno),
            ..Default::default()
        }
    }

    /// Sets the default filesystem backend used for levels without an explicit route.
    ///
    /// Defaults to `IoUringFs` where the kernel offers it, else
    /// [`StdFs`](crate::fs::StdFs). Use [`MemFs`](crate::fs::MemFs) for
    /// in-memory trees (testing, ephemeral indexes).
    ///
    /// # Example
    ///
    /// ```
    /// # fn main() -> lsm_tree::Result<()> {
    /// use lsm_tree::{Config, SequenceNumberCounter};
    /// use lsm_tree::fs::MemFs;
    ///
    /// let tree = Config::new(
    ///     "/virtual/tree",
    ///     SequenceNumberCounter::default(),
    ///     SequenceNumberCounter::default(),
    /// )
    /// .with_fs(MemFs::new())
    /// .open()?;
    /// # Ok(())
    /// # }
    /// ```
    #[must_use]
    pub fn with_fs<F: Fs>(mut self, fs: F) -> Self {
        self.fs = Arc::new(fs);
        self
    }

    /// Sets the default filesystem backend from an existing shared handle.
    ///
    /// Useful when multiple configs should reuse the same backend
    /// instance, including trait objects and backends that are not `Clone`.
    ///
    #[must_use]
    pub fn with_shared_fs(mut self, fs: Arc<dyn Fs>) -> Self {
        self.fs = fs;
        self
    }

    /// Opens a tree using the config.
    ///
    /// # Errors
    ///
    /// Will return `Err` if an IO error occurs.
    /// Returns [`Error::ZstdDictMismatch`](crate::Error::ZstdDictMismatch) if
    /// the compression policy references a `dict_id` that doesn't match the
    /// configured dictionary.
    pub fn open(self) -> crate::Result<AnyTree> {
        self.open_with(
            #[cfg(feature = "std")]
            DirectoryLock::Acquire,
            crate::tree::NewFolderNames::Synced,
        )
        .map(|(tree, _)| tree)
    }

    /// Opens a tree for each config, making each one that does not exist
    /// yet, and makes the new trees' folder names durable with one sync of
    /// each folder that holds them, after all of them are made, rather than
    /// one sync of it per tree. A consumer that makes many trees in one data
    /// directory at once pays one device flush for that directory.
    ///
    /// The trees come back in the order of `configs`. They are durable once
    /// this returns: before that a power loss may take back any new tree of
    /// the batch, with everything in it. A batch that fails takes back every
    /// tree it made and closes the ones it opened, so a retry finds what was
    /// there before it.
    ///
    /// # Errors
    ///
    /// As [`Self::open`], for the first config whose tree fails to open, or a
    /// failed sync of a folder holding a new tree's name.
    ///
    /// # Examples
    ///
    /// ```
    /// use lsm_tree::fs::MemFs;
    /// use lsm_tree::{AbstractTree, Config, SequenceNumberCounter};
    /// use std::sync::Arc;
    ///
    /// let fs = Arc::new(MemFs::new());
    /// let config = |name: &str| {
    ///     Config::new(
    ///         format!("/data/{name}"),
    ///         SequenceNumberCounter::default(),
    ///         SequenceNumberCounter::default(),
    ///     )
    ///     .with_shared_fs(fs.clone())
    /// };
    /// let trees = Config::open_many([config("nodes"), config("edges")])?;
    /// trees[0].insert("a", "node", 0);
    /// assert!(trees[1].is_empty(lsm_tree::SeqNo::MAX, None)?);
    /// # Ok::<(), lsm_tree::Error>(())
    /// ```
    pub fn open_many(configs: impl IntoIterator<Item = Self>) -> crate::Result<Vec<AnyTree>> {
        crate::tree::open_batch(configs)
    }

    /// [`Self::open`], coming by the directory lock as `lock` says, and
    /// leaving the new tree folder's name to its caller to sync as `names`
    /// says; with the tree, when this open made it, what takes it back.
    pub(crate) fn open_with(
        self,
        #[cfg(feature = "std")] lock: DirectoryLock,
        names: crate::tree::NewFolderNames,
    ) -> crate::Result<(AnyTree, Option<crate::tree::Creation>)> {
        // The dictionary registry is loaded and the compression policy validated
        // inside `Tree::open`, under the directory lock — NOT here. Both read the
        // tree's `dicts/` folder, and the load SWEEPS unpublished temps; doing
        // that before the lock lets an opener that is about to fail with
        // `Locked` delete a live owner's in-flight registration.

        // On a zstd build the live block path seals encrypted blocks through
        // the AAD-bound envelope, so the configured provider MUST implement it.
        // Reject an opaque-only provider here, at open time, instead of letting
        // it fail on the first encrypted read/write.
        #[cfg(zstd_any)]
        if self
            .encryption
            .as_ref()
            .is_some_and(|enc| !enc.supports_aad_block_path())
        {
            return Err(crate::Error::Encrypt(
                "encryption provider does not implement the AAD-bound block path \
                 (encrypt_block_aad / decrypt_block_aad) required for encrypted \
                 blocks on a zstd build",
            ));
        }

        Ok(if self.kv_separation_opts.is_some() {
            let (tree, creation) = BlobTree::open(
                self,
                #[cfg(feature = "std")]
                lock,
                names,
            )?;
            (AnyTree::Blob(tree), creation)
        } else {
            let (tree, creation) = Tree::open(
                self,
                #[cfg(feature = "std")]
                lock,
                names,
            )?;
            (AnyTree::Standard(tree), creation)
        })
    }

    /// Builds a tree at [`Self::path`] from a [`PinnedVersion`](crate::PinnedVersion)
    /// taken elsewhere, and opens it.
    ///
    /// Before the call, the pin's files lie in the tree's folders under their
    /// ids: each of [`tables`](crate::PinnedVersion::tables) as `tables/{id}`
    /// (with its [`heal_attestation`](crate::PinnedFile::heal_attestation),
    /// when it has one, as `tables/{id}.heal-attest`), each of
    /// [`blob_files`](crate::PinnedVersion::blob_files) as `blobs/{id}`, each of
    /// [`dictionaries`](crate::PinnedVersion::dictionaries) as `dicts/{id}`.
    /// `layout` is the pin's [`layout`](crate::PinnedVersion::layout), and
    /// `entries` and `range_tombstones` are its
    /// [`memtable_entries`](crate::PinnedVersion::memtable_entries) and
    /// [`memtable_range_tombstones`](crate::PinnedVersion::memtable_range_tombstones).
    ///
    /// The directory lock is taken first and held into the open, so no other
    /// opener sees the folder half restored. Until the restored memtable
    /// contents are durable a `restoring` marker lies in the folder: an
    /// ordinary [`Self::open`] refuses it, and calling this again with the
    /// same pin finishes an interrupted restore. The files are synced and the
    /// layout published as the tree's manifest, then the memtable contents are
    /// written with their sequence numbers and flushed. The returned tree reads
    /// as the source did at the pin, for every key and every snapshot the
    /// source served then. The config's sequence number counters are raised to
    /// the pin's seqno (and the seqno counter past every restored version), so
    /// later writes land above the restored data. The config must match the
    /// source's where it shapes the files: the tree type, comparator,
    /// encryption provider and dictionaries.
    ///
    /// # Errors
    ///
    /// [`Error::InvalidHeader`](crate::Error::InvalidHeader) for a layout too
    /// short to hold its header; [`Error::Locked`](crate::Error::Locked) when
    /// another instance holds the folder; an `AlreadyExists` I/O error, before
    /// anything is written, when the folder already holds a tree (a manifest
    /// snapshot or edit log) that no interrupted restore left; otherwise
    /// the failures of the syncs, the
    /// manifest write, [`Self::open`] and the flush.
    #[cfg(feature = "std")]
    pub fn restore_pinned(
        self,
        layout: &[u8],
        entries: impl IntoIterator<Item = crate::PinnedEntry>,
        range_tombstones: impl IntoIterator<Item = crate::PinnedRangeTombstone>,
    ) -> crate::Result<AnyTree> {
        crate::pin::restore(self, layout, entries, range_tombstones)
    }

    /// A snapshot of the dictionaries the tree can currently decompress
    /// against.
    ///
    /// One load per table open, which is where the set is needed; the returned
    /// value is a handle (one `Arc`), not a copy of the dictionaries.
    #[cfg(zstd_any)]
    pub(crate) fn current_zstd_dictionaries(&self) -> crate::compression::ZstdDictionaries {
        self.zstd_dictionaries.load().as_ref().clone()
    }

    /// Gives this config a registry of its OWN, holding every dictionary the
    /// tree at [`Self::path`] stores plus the ones supplied through
    /// [`Self::zstd_dictionary`] and the KV-separation options.
    ///
    /// Every path that opens a table has to call this first: a table resolves
    /// the dictionary id it recorded against this set, so an unloaded one fails
    /// every dictionary-compressed table in the tree.
    ///
    /// The registry is REPLACED rather than stored into, because `Config` is
    /// `Clone` and a keyspace derives one config per partition from a base. The
    /// clones would otherwise share a single set, and the second tree's load
    /// would take over the first tree's registry — after which the first tree
    /// stops resolving the tables it wrote itself.
    ///
    /// # Errors
    ///
    /// Propagates the folder scan and the store's integrity check: a dictionary
    /// whose bytes no longer hash to its name fails the open rather than
    /// dropping silently out of the set.
    #[cfg(zstd_any)]
    pub(crate) fn install_own_zstd_dictionaries(&mut self) -> crate::Result<()> {
        let folder = self.path.join(crate::file::DICTS_FOLDER);
        // A `.tmp` is an unpublished registration referenced by no version, so a
        // crashed one is swept rather than left to linger.
        crate::dicts::sweep_temps(&*self.fs, &folder)?;

        let dicts = crate::dicts::read_all(&*self.fs, &folder, self.encryption.as_deref())?;
        self.install_zstd_dictionary_set(dicts);
        Ok(())
    }

    /// [`Self::install_own_zstd_dictionaries`] for a repair: a dictionary whose
    /// bytes no longer hash to its name is left out of the set instead of
    /// failing the load, and its id is returned. Nothing is changed on disk.
    ///
    /// # Errors
    ///
    /// Propagates the folder scan and every read failure other than the
    /// integrity check.
    #[cfg(zstd_any)]
    pub(crate) fn install_own_zstd_dictionaries_skipping_damaged(
        &mut self,
    ) -> crate::Result<alloc::vec::Vec<crate::file::DictId>> {
        let folder = self.path.join(crate::file::DICTS_FOLDER);
        crate::dicts::sweep_temps(&*self.fs, &folder)?;

        let (dicts, damaged) = crate::dicts::read_all_skipping_damaged(
            &*self.fs,
            &folder,
            self.encryption.as_deref(),
        )?;
        self.install_zstd_dictionary_set(dicts);
        Ok(damaged)
    }

    /// Joins the supplied dictionaries to `dicts` and installs it as this
    /// config's own registry.
    ///
    /// The supplied ones join the SAME set the tree's stored ones are in, and
    /// every read and write resolves out of it by id: a table or blob file by
    /// the id it recorded, a writer by the id its policy names.
    #[cfg(zstd_any)]
    fn install_zstd_dictionary_set(&mut self, mut dicts: crate::compression::ZstdDictionaries) {
        if let Some(supplied) = self.zstd_dictionary.clone() {
            dicts = dicts.with(supplied);
        }
        if let Some(supplied) = self
            .kv_separation_opts
            .as_ref()
            .and_then(|o| o.zstd_dictionary.clone())
        {
            dicts = dicts.with(supplied);
        }
        self.zstd_dictionaries = Arc::new(arc_swap::ArcSwap::from_pointee(dicts));
    }

    /// Like [`Config::new`], but accepts pre-built shared generators.
    ///
    /// This is useful when the caller already has
    /// [`SharedSequenceNumberGenerator`] instances (e.g., from a higher-level
    /// database that shares generators across multiple trees).
    // std-only: see [`Config::new`] — seeds via `Config::default` (`StdFs`).
    #[cfg(feature = "std")]
    pub fn new_with_generators<P: AsRef<Path>>(
        path: P,
        seqno: SharedSequenceNumberGenerator,
        visible_seqno: SharedSequenceNumberGenerator,
    ) -> Self {
        Self {
            path: absolute_path(path.as_ref()),
            seqno,
            visible_seqno,
            ..Default::default()
        }
    }
}

#[cfg(all(test, zstd_any))]
mod tests;

impl Config {
    /// The [`level_routes`](Self::level_routes) entry covering `level`, if any.
    ///
    /// The single place the level→route lookup lives, so the path a level's
    /// tables are written to and the backend its reads are submitted to cannot
    /// disagree about which route covers it.
    fn route_for_level(&self, level: u8) -> Option<&LevelRoute> {
        self.level_routes
            .as_ref()?
            .iter()
            .find(|route| route.levels.contains(&level))
    }

    /// Returns the tables folder path and [`Fs`] backend for the given level.
    ///
    /// If [`level_routes`](Self::level_routes) has an entry covering this
    /// level, uses that entry's path and `Fs`. Otherwise falls back to the
    /// primary [`path`](Self::path) and [`fs`](Self::fs).
    #[must_use]
    pub fn tables_folder_for_level(&self, level: u8) -> (PathBuf, Arc<dyn Fs>) {
        match self.route_for_level(level) {
            Some(route) => (route.path.join(TABLES_FOLDER), route.fs.clone()),
            None => (self.path.join(TABLES_FOLDER), self.fs.clone()),
        }
    }

    /// Best-effort minimum free space (bytes) across every filesystem this tree
    /// writes to: the primary [`path`](Self::path) plus each
    /// [`level_routes`](Self::level_routes) volume.
    ///
    /// The tightest volume bounds storage admission and compaction space gating,
    /// since a full routed (cold-tier) volume fails a flush / compaction
    /// targeting it even while the primary still has room. A backend that cannot
    /// report free space (or an I/O hiccup) contributes `u64::MAX`, so a probe
    /// failure never fabricates disk pressure.
    #[must_use]
    pub(crate) fn min_available_space(&self) -> u64 {
        let mut free = self.fs.available_space(&self.path).unwrap_or(u64::MAX);
        if let Some(routes) = &self.level_routes {
            for route in routes {
                free = free.min(route.fs.available_space(&route.path).unwrap_or(u64::MAX));
            }
        }
        free
    }

    /// Returns all unique tables folders that need to be scanned during
    /// recovery: the primary folder plus every [`LevelRoute`] folder.
    #[must_use]
    pub fn all_tables_folders(&self) -> Vec<(PathBuf, Arc<dyn Fs>)> {
        let primary_fs: Arc<dyn Fs> = self.fs.clone();
        let mut folders: Vec<(PathBuf, Arc<dyn Fs>)> =
            vec![(self.path.join(TABLES_FOLDER), primary_fs)];

        if let Some(routes) = &self.level_routes {
            for route in routes {
                let folder = route.path.join(TABLES_FOLDER);
                // Dedup by path: scanning the same directory twice would cause
                // already-recovered tables to be classified as orphans and
                // deleted. Routing the same path through different Fs backends
                // is a configuration error (level_routes validation in
                // Config::level_routes rejects overlapping ranges).
                if !folders.iter().any(|(p, _)| *p == folder) {
                    folders.push((folder, route.fs.clone()));
                }
            }
        }

        folders
    }

    /// Configures per-level filesystem routing for tiered storage.
    ///
    /// Each [`LevelRoute`] maps a range of LSM levels to a base directory
    /// and filesystem backend. Levels not covered by any route fall back to
    /// the primary `path` and `fs`.
    ///
    /// # Reopen contract
    ///
    /// The route configuration is **not persisted** in the manifest.
    /// On reopen, the [`Config`] must specify `level_routes` such that
    /// [`all_tables_folders`](Self::all_tables_folders) includes every
    /// directory and filesystem pair that may contain existing SST files
    /// for this tree.
    ///
    /// Changing the mapping from levels to paths is allowed as long as
    /// the previously used folders remain covered. If old folders are
    /// omitted, recovery may fail with
    /// [`RouteMismatch`](crate::Error::RouteMismatch) (when all missing
    /// tables are on uncovered levels) or
    /// [`Unrecoverable`](crate::Error::Unrecoverable) (when some missing
    /// tables are on levels that are still covered).
    ///
    /// # Panics
    ///
    /// Panics if any route has an empty range or if any two routes have
    /// overlapping level ranges.
    #[must_use]
    pub fn level_routes(mut self, routes: Vec<LevelRoute>) -> Self {
        // Validate no empty/inverted ranges
        for route in &routes {
            assert!(
                route.levels.start < route.levels.end,
                "empty or inverted level route range: {:?}",
                route.levels,
            );
        }

        // Validate no overlapping ranges
        for (i, a) in routes.iter().enumerate() {
            for b in routes.iter().skip(i + 1) {
                assert!(
                    a.levels.end <= b.levels.start || b.levels.end <= a.levels.start,
                    "overlapping level routes: {:?} and {:?}",
                    a.levels,
                    b.levels,
                );
            }
        }
        self.level_routes = if routes.is_empty() {
            None
        } else {
            // Normalize paths the same way Config::new normalizes self.path
            Some(
                routes
                    .into_iter()
                    .map(|mut r| {
                        r.path = absolute_path(&r.path);
                        r
                    })
                    .collect(),
            )
        };
        self
    }

    /// Overrides the sequence number generator.
    ///
    /// By default, [`SequenceNumberCounter`] is used. This allows plugging in
    /// a custom generator (e.g., HLC for distributed databases).
    #[must_use]
    pub fn seqno_generator(mut self, generator: SharedSequenceNumberGenerator) -> Self {
        self.seqno = generator;
        self
    }

    /// Overrides the visible sequence number generator.
    #[must_use]
    pub fn visible_seqno_generator(mut self, generator: SharedSequenceNumberGenerator) -> Self {
        self.visible_seqno = generator;
        self
    }

    /// Sets the global cache.
    ///
    /// You can create a global [`Cache`] and share it between multiple
    /// trees to cap global cache memory usage.
    ///
    /// Defaults to a cache with 16 MiB of capacity *per tree*.
    #[must_use]
    pub fn use_cache(mut self, cache: Arc<Cache>) -> Self {
        self.cache = cache;
        self
    }

    /// Sets the file descriptor cache.
    ///
    /// Can be shared across trees.
    #[must_use]
    pub fn use_descriptor_table(mut self, descriptor_table: Option<Arc<DescriptorTable>>) -> Self {
        self.descriptor_table = descriptor_table;
        self
    }

    /// If `true`, the last level will not build filters, reducing the filter size of a database
    /// by ~90% typically.
    ///
    /// **Enable this only if you know that point reads generally are expected to find a key-value pair.**
    #[must_use]
    pub fn expect_point_read_hits(mut self, b: bool) -> Self {
        self.expect_point_read_hits = b;
        self
    }

    /// Whether a compaction ends its outputs on the boundaries between the
    /// tables of the level they are merged into next.
    ///
    /// An output straddling such a boundary drags both tables into that next
    /// merge, which reads and rewrites the extra one. Aligned, an output ends
    /// at the first boundary past half the target size; each boundary it
    /// passes without ending raises that share by five percent, up to ninety;
    /// it ends at twice the target whatever the boundaries, and at the target
    /// when no boundary is left ahead. A user key's versions stay in one output
    /// above every size bound. Off, outputs end at the target size alone.
    ///
    /// Default = `true`.
    ///
    /// # Examples
    ///
    /// ```
    /// # let folder = tempfile::tempdir()?;
    /// use lsm_tree::{Config, SequenceNumberCounter};
    ///
    /// let config = Config::new(
    ///     folder,
    ///     SequenceNumberCounter::default(),
    ///     SequenceNumberCounter::default(),
    /// )
    /// .compaction_output_alignment(false);
    /// # let _ = config;
    /// # Ok::<(), Box<dyn std::error::Error>>(())
    /// ```
    #[must_use]
    pub fn compaction_output_alignment(mut self, enabled: bool) -> Self {
        self.compaction_output_cuts = if enabled {
            OutputCuts::Aligned
        } else {
            OutputCuts::TargetSize
        };
        self
    }

    /// Enables per-block Page ECC.
    ///
    /// When enabled, every block written by this tree carries a parity
    /// trailer; on read, if the block's XXH3 disagrees with the on-disk
    /// bytes, the reader attempts recovery from the trailer before surfacing
    /// the corruption. The correction scheme defaults to per-word SEC-DED and
    /// is selectable at runtime (`update_runtime_config`): per-word SEC-DED,
    /// single XOR parity, or Reed-Solomon.
    ///
    /// Opening a tree with `page_ecc = true` on a build that does not
    /// have the `page_ecc` cargo feature enabled returns
    /// [`crate::Error::PageEccUnsupported`] at `Tree::open` — the
    /// reader has no way to honour the parity trailer without the
    /// codec, so silently downgrading integrity is not an option.
    ///
    /// Wired into the on-disk write path via `MultiWriter::use_page_ecc`
    /// at every `Tree::open` / `Tree::ingestion` / compaction-worker
    /// `MultiWriter` construction site. With this flag set, every
    /// `Block::write_into` call those writers make upgrades its
    /// `BlockTransform` to the matching `*Ecc` variant — emitting the
    /// configured scheme's parity trailer and setting the `ECC_PARITY` flag
    /// in each block header (the trailer length is derived from
    /// `data_length`, not stored).
    #[must_use]
    pub fn page_ecc(mut self, enabled: bool) -> Self {
        self.page_ecc = enabled;
        self
    }

    /// Enables or disables the cross-process directory lock (default: enabled).
    ///
    /// When enabled, [`Config::open`] and [`Config::repair`] acquire an
    /// exclusive advisory lock on a `LOCK` file in the tree directory, so a
    /// second process opening / repairing the same directory fails fast with
    /// [`Error::Locked`](crate::Error::Locked) rather than corrupting the shared
    /// manifest. Disable ONLY when exclusive directory ownership is already
    /// guaranteed at a higher layer (e.g. an embedding keyspace / journal
    /// manager that opens each directory at most once per host).
    #[must_use]
    pub fn with_directory_lock(mut self, enabled: bool) -> Self {
        self.directory_lock = enabled;
        self
    }

    /// Wires shared live-progress counters into the repair / salvage paths.
    ///
    /// A repair over a large store streams every SST and blob file and can run
    /// for a long time; the handle set here is ticked as files are discovered
    /// and blocks / rows are recovered, so another thread can poll
    /// [`RecoveryProgress::snapshot`](crate::RecoveryProgress::snapshot) while
    /// [`Config::repair`] (or a salvage it triggers) runs. Without it, repair
    /// publishes no progress (zero overhead).
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use lsm_tree::{Config, RecoveryProgress, SequenceNumberCounter};
    /// use std::sync::Arc;
    ///
    /// let progress = Arc::new(RecoveryProgress::default());
    /// let config = Config::new(
    ///     "my-tree",
    ///     SequenceNumberCounter::default(),
    ///     SequenceNumberCounter::default(),
    /// )
    /// .with_recovery_progress(progress.clone());
    /// // spawn the repair, then poll `progress.snapshot()` elsewhere
    /// let report = config.repair()?;
    /// # Ok::<_, lsm_tree::Error>(())
    /// ```
    #[cfg(feature = "std")]
    #[must_use]
    pub fn with_recovery_progress(mut self, progress: Arc<crate::RecoveryProgress>) -> Self {
        self.recovery_progress = Some(progress);
        self
    }

    /// Sets the Page ECC scheme used when [`Self::page_ecc`] is enabled.
    ///
    /// ECC is off until `page_ecc(true)`. When on, this picks the
    /// algorithm:
    /// [`EccScheme::Secded`](crate::runtime_config::EccScheme::Secded)
    /// (per-word single-bit correct / double-bit detect, the default, supported
    /// at Block granularity),
    /// [`EccScheme::Xor`](crate::runtime_config::EccScheme::Xor) (RAID-5
    /// single-parity), or
    /// [`EccScheme::ReedSolomon`](crate::runtime_config::EccScheme::ReedSolomon).
    /// There is no implicit RS(4,2) default.
    #[must_use]
    pub fn ecc_scheme(mut self, scheme: crate::runtime_config::EccScheme) -> Self {
        self.initial_runtime_config.ecc_scheme = scheme;
        self
    }

    /// Sets whether the writer clears per-file copy-on-write on newly created
    /// SST / blob files when the backing filesystem is copy-on-write (Btrfs).
    ///
    /// Default `true`: write-once SSTs gain no benefit from `CoW` but suffer a
    /// fragmentation penalty (~20% write throughput on Btrfs), so clearing it
    /// recovers the ext4-equivalent baseline. A no-op on non-`CoW` filesystems.
    /// Set `false` to preserve `CoW` (e.g. Btrfs subvolume snapshots that depend
    /// on it). See [`crate::runtime_config::RuntimeConfig::disable_cow_on_sst_files`].
    #[must_use]
    pub fn disable_cow_on_sst_files(mut self, enabled: bool) -> Self {
        self.initial_runtime_config.disable_cow_on_sst_files = enabled;
        self
    }

    /// Sets whether [`crate::AbstractTree::create_checkpoint`] clones files via
    /// reflink (`FICLONE` / `clonefile`) when the filesystem supports it,
    /// falling back to a hard link otherwise.
    ///
    /// Default `true`: a reflinked checkpoint has an independent inode (no
    /// max-links constraint, modifications never touch the original) at O(1)
    /// cost via copy-on-write block sharing. A no-op (hard-link path) on
    /// filesystems without reflink. See
    /// [`crate::runtime_config::RuntimeConfig::use_reflink_for_checkpoint`].
    #[must_use]
    pub fn use_reflink_for_checkpoint(mut self, enabled: bool) -> Self {
        self.initial_runtime_config.use_reflink_for_checkpoint = enabled;
        self
    }

    /// Sets the initial [`crate::runtime_config::RuntimeConfig`]
    /// snapshot the tree will start with.
    ///
    /// Seeds both the first manifest write and the live
    /// `RuntimeConfigHandle` exposed via
    /// [`crate::Tree::runtime_config`].
    ///
    /// Replaces the snapshot, including what the builders that seed it set
    /// earlier ([`Self::ecc_scheme`] and the other `RuntimeConfig`-backed
    /// builders), EXCEPT the three compression policies: those keep what
    /// [`Self::data_block_compression_policy`],
    /// [`Self::index_block_compression_policy`] and [`Self::blob_compression`]
    /// set, in whichever order they are called, and the policies in `runtime`
    /// are ignored. A caller that set its compression through the builders and
    /// its runtime toggles through this does not lose the former to the
    /// latter's defaults. On a live tree, all three change through
    /// [`crate::Tree::update_runtime_config`].
    ///
    /// **Manifest-hardening toggles** in the supplied snapshot
    /// that are currently wired through the writer
    /// (`manifest_footer_mirror`, `page_ecc` *as consumed by
    /// `manifest_blocks::writer` when picking the `BlockTransform`
    /// variant*) take effect from byte zero of the on-disk
    /// manifest rather than waiting for a post-open
    /// [`crate::Tree::update_runtime_config`] call. Subsequent
    /// updates still flow through the live handle and apply to
    /// the next manifest write.
    ///
    /// `manifest_kv_checksums` is plumbed in the snapshot but the
    /// writer does NOT yet consult or persist it (per-entry
    /// framing + footer-flag slot land in a follow-up). Setting
    /// it here today has no on-disk effect; it is exposed for
    /// forward-compat with no behaviour break.
    ///
    /// **Note on data-block ECC:** `RuntimeConfig::page_ecc`
    /// currently affects manifest Blocks only — data-block ECC is
    /// still gated by [`Config::page_ecc`] at tree-open time. The
    /// SST writer path consumes the tree-static config, not the
    /// runtime handle. Wiring through SST emission is a follow-up.
    #[must_use]
    pub fn with_runtime_config(mut self, runtime: crate::runtime_config::RuntimeConfig) -> Self {
        let kept = core::mem::replace(&mut self.initial_runtime_config, runtime);
        self.initial_runtime_config.data_block_compression_policy =
            kept.data_block_compression_policy;
        self.initial_runtime_config.index_block_compression_policy =
            kept.index_block_compression_policy;
        self.initial_runtime_config.blob_compression = kept.blob_compression;
        self
    }

    /// Sets the partitioning policy for filter blocks.
    #[must_use]
    pub fn filter_block_partitioning_policy(mut self, policy: PinningPolicy) -> Self {
        self.filter_block_partitioning_policy = policy;
        self
    }

    /// Sets the partitioning policy for index blocks.
    #[must_use]
    pub fn index_block_partitioning_policy(mut self, policy: PinningPolicy) -> Self {
        self.index_block_partitioning_policy = policy;
        self
    }

    /// Sets the pinning policy for filter blocks.
    #[must_use]
    pub fn filter_block_pinning_policy(mut self, policy: PinningPolicy) -> Self {
        self.filter_block_pinning_policy = policy;
        self
    }

    /// Sets the pinning policy for index blocks.
    #[must_use]
    pub fn index_block_pinning_policy(mut self, policy: PinningPolicy) -> Self {
        self.index_block_pinning_policy = policy;
        self
    }

    /// Sets the restart interval inside data blocks.
    ///
    /// A higher restart interval saves space while increasing lookup times
    /// inside data blocks.
    ///
    /// Default = 16
    ///
    /// # Panics
    ///
    /// Panics if any restart interval in `policy` is zero.
    #[must_use]
    pub fn data_block_restart_interval_policy(mut self, policy: RestartIntervalPolicy) -> Self {
        assert!(
            policy.iter().all(|interval| *interval > 0),
            "data block restart interval must be greater than zero",
        );
        self.data_block_restart_interval_policy = policy;
        self
    }

    /// Sets the restart interval inside index blocks.
    ///
    /// A higher restart interval saves space while increasing lookup times
    /// inside index blocks.
    ///
    /// Default = 1
    ///
    /// # Panics
    ///
    /// Panics if any restart interval in `policy` is zero.
    #[must_use]
    pub fn index_block_restart_interval_policy(mut self, policy: RestartIntervalPolicy) -> Self {
        assert!(
            policy.iter().all(|interval| *interval > 0),
            "index block restart interval must be greater than zero",
        );
        self.index_block_restart_interval_policy = policy;
        self
    }

    /// Sets the filter construction policy.
    #[must_use]
    pub fn filter_policy(mut self, policy: FilterPolicy) -> Self {
        self.filter_policy = policy;
        self
    }

    /// Allocates filter memory by measured negative-probe load; see
    /// [`FilterAdvisor`]. `None` (the default) leaves every filter to the
    /// [`FilterPolicy`] and counts nothing on the read path, so the tables it
    /// writes are the ones a tree without the advisor writes.
    ///
    /// With an advisor, each flush and compaction chooses the width of the
    /// filters it writes; [`AbstractTree::filter_memory`](crate::AbstractTree::filter_memory)
    /// reports the serialised and resident filter bytes against the budget.
    ///
    /// # Examples
    ///
    /// ```
    /// use lsm_tree::{Config, config::FilterAdvisor};
    ///
    /// # let folder = tempfile::tempdir()?;
    /// // 64 MiB of filters, widths from 8 to 16 bits per key.
    /// let advisor = FilterAdvisor::new(64 << 20).with_bits_per_key([8, 10, 12, 14, 16]);
    /// let tree = Config::new(&folder, Default::default(), Default::default())
    ///     .filter_advisor(Some(advisor))
    ///     .open()?;
    /// # Ok::<(), lsm_tree::Error>(())
    /// ```
    #[must_use]
    pub fn filter_advisor(mut self, advisor: Option<FilterAdvisor>) -> Self {
        self.filter_advisor = advisor;
        self
    }

    /// Sets the retrieval-ribbon locator policy.
    ///
    /// On by default at [`LocatorPrecision::Block`] (see
    /// [`LocatorPolicy::block_level`]). When enabled for a level, written SSTs on
    /// that level carry an optional `locator` section mapping each key to its
    /// data block (and, at finer precisions, its slot), letting point reads skip
    /// the index-block binary search. Set [`LocatorPolicy::disabled`] to opt out;
    /// disabled levels emit byte-identical SSTs.
    #[must_use]
    pub fn locator_policy(mut self, policy: LocatorPolicy) -> Self {
        self.locator_policy = policy;
        self
    }

    /// Sets the compression the tree starts with for data blocks.
    ///
    /// The initial value of
    /// [`RuntimeConfig::data_block_compression_policy`](crate::runtime_config::RuntimeConfig::data_block_compression_policy),
    /// which a live tree changes with [`crate::Tree::update_runtime_config`].
    #[must_use]
    pub fn data_block_compression_policy(mut self, policy: CompressionPolicy) -> Self {
        self.initial_runtime_config.data_block_compression_policy = policy;
        self
    }

    /// Sets the compression the tree starts with for index blocks.
    ///
    /// The initial value of
    /// [`RuntimeConfig::index_block_compression_policy`](crate::runtime_config::RuntimeConfig::index_block_compression_policy),
    /// which a live tree changes with [`crate::Tree::update_runtime_config`].
    #[must_use]
    pub fn index_block_compression_policy(mut self, policy: CompressionPolicy) -> Self {
        self.initial_runtime_config.index_block_compression_policy = policy;
        self
    }

    /// Sets the compression a KV-separated tree starts with for blob files.
    ///
    /// The initial value of
    /// [`RuntimeConfig::blob_compression`](crate::runtime_config::RuntimeConfig::blob_compression),
    /// which a live tree changes with [`crate::BlobTree::update_runtime_config`].
    /// A standard tree ignores it.
    #[must_use]
    pub fn blob_compression(mut self, compression: CompressionType) -> Self {
        self.initial_runtime_config.blob_compression = compression;
        self
    }

    /// Sets whether a KV-separated tree starts caching the values it reads
    /// from blob files. It does not by default.
    ///
    /// The initial value of
    /// [`RuntimeConfig::blob_cache_fill`](crate::runtime_config::RuntimeConfig::blob_cache_fill),
    /// which a live tree changes with [`crate::BlobTree::update_runtime_config`].
    /// A standard tree ignores it.
    ///
    /// # Examples
    ///
    /// ```
    /// use lsm_tree::{Config, KvSeparationOptions, SequenceNumberCounter, config::BlobCacheFill};
    ///
    /// # let folder = tempfile::tempdir()?;
    /// let config = Config::new(&folder, SequenceNumberCounter::default(), SequenceNumberCounter::default())
    ///     .with_kv_separation(Some(KvSeparationOptions::default()))
    ///     .blob_cache_fill(BlobCacheFill::Always);
    /// # Ok::<(), Box<dyn std::error::Error>>(())
    /// ```
    #[must_use]
    pub fn blob_cache_fill(mut self, fill: BlobCacheFill) -> Self {
        self.initial_runtime_config.blob_cache_fill = fill;
        self
    }

    // TODO: level count is fixed to 7 right now
    // /// Sets the number of levels of the LSM tree (depth of tree).
    // ///
    // /// Defaults to 7, like `LevelDB` and `RocksDB`.
    // ///
    // /// Cannot be changed once set.
    // ///
    // /// # Panics
    // ///
    // /// Panics if `n` is 0.
    // #[must_use]
    // pub fn level_count(mut self, n: u8) -> Self {
    //     assert!(n > 0);

    //     self.level_count = n;
    //     self
    // }

    /// Sets the data block size policy.
    #[must_use]
    pub fn data_block_size_policy(mut self, policy: BlockSizePolicy) -> Self {
        self.data_block_size_policy = policy;
        self
    }

    /// Sets the size a partitioned block index cuts its partitions at, per
    /// level. Applies where [`Self::index_block_partitioning_policy`]
    /// partitions the index. The default is 4 KiB at every level.
    ///
    /// # Examples
    ///
    /// ```
    /// use lsm_tree::{Config, SequenceNumberCounter, config::BlockSizePolicy};
    /// # let folder = tempfile::tempdir()?;
    /// let config = Config::new(
    ///     folder.path(),
    ///     SequenceNumberCounter::default(),
    ///     SequenceNumberCounter::default(),
    /// )
    /// .index_block_partition_size_policy(BlockSizePolicy::all(16 * 1_024));
    /// # Ok::<(), Box<dyn std::error::Error>>(())
    /// ```
    #[must_use]
    pub fn index_block_partition_size_policy(mut self, policy: BlockSizePolicy) -> Self {
        self.index_block_partition_size_policy = policy;
        self
    }

    /// Sets the size a partitioned filter cuts its partitions at, per level,
    /// in the bytes a partition takes on disk. Applies where
    /// [`Self::filter_block_partitioning_policy`] partitions the filter. A
    /// point read that misses the block cache loads one partition, so a
    /// smaller partition costs less per read and more top-level index
    /// entries. The default is 4 KiB at every level.
    ///
    /// # Examples
    ///
    /// ```
    /// use lsm_tree::{Config, SequenceNumberCounter, config::BlockSizePolicy};
    /// # let folder = tempfile::tempdir()?;
    /// let config = Config::new(
    ///     folder.path(),
    ///     SequenceNumberCounter::default(),
    ///     SequenceNumberCounter::default(),
    /// )
    /// .filter_block_partition_size_policy(BlockSizePolicy::all(16 * 1_024));
    /// # Ok::<(), Box<dyn std::error::Error>>(())
    /// ```
    #[must_use]
    pub fn filter_block_partition_size_policy(mut self, policy: BlockSizePolicy) -> Self {
        self.filter_block_partition_size_policy = policy;
        self
    }

    /// Sets the size a columnar table's row groups are cut at, per level.
    ///
    /// Applies to the tables written while the runtime `columnar` switch is
    /// on; a row-major table cuts its blocks at
    /// [`Self::data_block_size_policy`]. The default is
    /// [`DEFAULT_COLUMNAR_ROW_GROUP_SIZE`] at every level.
    ///
    /// # Examples
    ///
    /// ```
    /// use lsm_tree::{Config, SequenceNumberCounter, config::BlockSizePolicy};
    /// # let folder = tempfile::tempdir()?;
    /// let config = Config::new(
    ///     folder.path(),
    ///     SequenceNumberCounter::default(),
    ///     SequenceNumberCounter::default(),
    /// )
    /// .columnar_row_group_size_policy(BlockSizePolicy::all(256 * 1_024));
    /// # Ok::<(), Box<dyn std::error::Error>>(())
    /// ```
    #[must_use]
    pub fn columnar_row_group_size_policy(mut self, policy: BlockSizePolicy) -> Self {
        self.columnar_row_group_size_policy = policy;
        self
    }

    /// Sets the size a columnar row group's rows are cut into row pages at,
    /// per level.
    ///
    /// A row page closes once its rows reach this many uncompressed bytes, so
    /// a page holds at least one row and a group at least one page. A size at
    /// or above the row group size writes one row page per group the writer
    /// closes at its size; a group an ingested batch took past it is still cut
    /// into pages of this size. The default is [`DEFAULT_COLUMNAR_PAGE_SIZE`]
    /// at every level, four row pages to a group of the default size.
    ///
    /// # Examples
    ///
    /// ```
    /// use lsm_tree::{Config, SequenceNumberCounter, config::BlockSizePolicy};
    /// # let folder = tempfile::tempdir()?;
    /// let config = Config::new(
    ///     folder.path(),
    ///     SequenceNumberCounter::default(),
    ///     SequenceNumberCounter::default(),
    /// )
    /// .columnar_row_group_size_policy(BlockSizePolicy::all(256 * 1_024))
    /// .columnar_page_size_policy(BlockSizePolicy::all(8 * 1_024));
    /// # Ok::<(), Box<dyn std::error::Error>>(())
    /// ```
    #[must_use]
    pub fn columnar_page_size_policy(mut self, policy: BlockSizePolicy) -> Self {
        self.columnar_page_size_policy = policy;
        self
    }

    /// Sets how a columnar table's column pages store their values, per
    /// level.
    ///
    /// The default is [`ColumnEncoding::Plain`] at every level: a read serves
    /// a plain page as a view of itself, while an encoded one is built back
    /// into its layout by each read that wants the column whole. Choose
    /// [`ColumnEncoding::Auto`] for the levels whose tables are mostly stored
    /// rather than read, as a stronger compression is chosen for them. A table
    /// reads the same under either; only what its pages hold differs.
    ///
    /// # Examples
    ///
    /// ```
    /// use lsm_tree::{
    ///     Config, SequenceNumberCounter,
    ///     config::{ColumnEncoding, ColumnEncodingPolicy},
    /// };
    /// # let folder = tempfile::tempdir()?;
    /// // Plain in the first three levels, encoded below them.
    /// let config = Config::new(
    ///     folder.path(),
    ///     SequenceNumberCounter::default(),
    ///     SequenceNumberCounter::default(),
    /// )
    /// .column_encoding_policy(ColumnEncodingPolicy::new([
    ///     ColumnEncoding::Plain,
    ///     ColumnEncoding::Plain,
    ///     ColumnEncoding::Plain,
    ///     ColumnEncoding::Auto,
    /// ]));
    /// # Ok::<(), Box<dyn std::error::Error>>(())
    /// ```
    #[must_use]
    pub fn column_encoding_policy(mut self, policy: ColumnEncodingPolicy) -> Self {
        self.column_encoding_policy = policy;
        self
    }

    /// Sets how columnar reads of this tree fetch their pages: the bytes one
    /// request may ask for and the requests kept in flight at once. A reader
    /// setting, not a format one: a tree reopened under another budget reads
    /// the same tables and returns the same rows. The default is
    /// [`ReadBudget::default`].
    ///
    /// # Examples
    ///
    /// ```
    /// use lsm_tree::{Config, SequenceNumberCounter, config::ReadBudget};
    /// # let folder = tempfile::tempdir()?;
    /// let config = Config::new(
    ///     folder.path(),
    ///     SequenceNumberCounter::default(),
    ///     SequenceNumberCounter::default(),
    /// )
    /// .columnar_read_budget(ReadBudget::new(64 * 1_024, 4));
    /// # Ok::<(), Box<dyn std::error::Error>>(())
    /// ```
    #[must_use]
    pub fn columnar_read_budget(mut self, budget: ReadBudget) -> Self {
        self.columnar_read_budget = budget;
        self
    }

    /// Sets the page bytes a projected columnar scan of this tree holds at
    /// once. A scan over overlapping segments shares it between them, so a
    /// scan that merges more of them reads each in smaller runs of row pages
    /// rather than holding proportionally more. A row page is the least a
    /// read takes, so one larger than a segment's share is read on its own
    /// and counted, not refused; so is a row group a pushed-down predicate
    /// reads whole to select its row pages. A reader setting: the rows returned are the
    /// same under any budget. The default is [`DEFAULT_COLUMNAR_SCAN_BUDGET`].
    ///
    /// # Examples
    ///
    /// ```
    /// use lsm_tree::{Config, SequenceNumberCounter};
    /// # let folder = tempfile::tempdir()?;
    /// let config = Config::new(
    ///     folder.path(),
    ///     SequenceNumberCounter::default(),
    ///     SequenceNumberCounter::default(),
    /// )
    /// .columnar_scan_budget(256 * 1_024);
    /// # Ok::<(), Box<dyn std::error::Error>>(())
    /// ```
    #[must_use]
    pub fn columnar_scan_budget(mut self, bytes: u64) -> Self {
        self.columnar_scan_budget = bytes;
        self
    }

    /// Sets the filter and index bytes a multi-get, blocking or resumable,
    /// holds at once, in flight or read, across the tables of a level it
    /// reads stage by stage. A block
    /// in flight counts at the size it is read as, a block read at its decoded
    /// size. A table whose blocks would pass it waits until a table before it
    /// is planned and lets its blocks go; one that alone passes it is read
    /// when no other is in flight. A stage is let in at its on-disk size, so
    /// with compressed index blocks the read can go past the budget by what
    /// the stages let in together grow when decoded; no further stage is let
    /// in until it is back under. Independent of the block cache, so a small
    /// cache does not make the tables of a level wait for each other. A reader
    /// setting: the values returned are the same under any budget. The
    /// default is [`DEFAULT_MULTI_GET_METADATA_BUDGET`].
    ///
    /// # Examples
    ///
    /// ```
    /// use lsm_tree::{Config, SequenceNumberCounter};
    /// # let folder = tempfile::tempdir()?;
    /// let config = Config::new(
    ///     folder.path(),
    ///     SequenceNumberCounter::default(),
    ///     SequenceNumberCounter::default(),
    /// )
    /// .multi_get_metadata_budget(8 * 1_024 * 1_024);
    /// # Ok::<(), Box<dyn std::error::Error>>(())
    /// ```
    #[must_use]
    pub fn multi_get_metadata_budget(mut self, bytes: u64) -> Self {
        self.multi_get_metadata_budget = bytes;
        self
    }

    /// Sets the hash ratio policy for data blocks.
    ///
    /// If greater than 0.0, a hash index is embedded into data blocks that can speed up reads
    /// inside the data block.
    #[must_use]
    pub fn data_block_hash_ratio_policy(mut self, policy: HashRatioPolicy) -> Self {
        self.data_block_hash_ratio_policy = policy;
        self
    }

    /// Toggles key-value separation.
    #[must_use]
    pub fn with_kv_separation(mut self, opts: Option<KvSeparationOptions>) -> Self {
        self.kv_separation_opts = opts;
        self
    }

    /// Installs a custom compaction filter.
    #[must_use]
    pub fn with_compaction_filter_factory(mut self, factory: Option<Arc<dyn Factory>>) -> Self {
        self.compaction_filter_factory = factory;
        self
    }

    /// Sets the prefix extractor for prefix bloom filters.
    ///
    /// When configured, bloom filters will index key prefixes returned by
    /// the extractor. Prefix scans can then skip segments whose bloom
    /// filter reports no match for the scan prefix.
    ///
    /// The extractor may be added or replaced on reopen. A table records the
    /// [`name`](PrefixExtractor::name) of the extractor its filter indexes,
    /// and a prefix scan skips by a table's filter only when that is the
    /// configured extractor: tables written without it, or under another,
    /// are read instead, until compaction rewrites them.
    #[must_use]
    pub fn prefix_extractor(mut self, extractor: Arc<dyn PrefixExtractor>) -> Self {
        self.prefix_extractor = Some(extractor);
        self
    }

    /// Ends every table that flush, ingest and compaction write before the
    /// first key of the next of `domains`, even below the target size, so no
    /// new table holds keys of two domains.
    ///
    /// The policy may be added or replaced on reopen: tables written before it
    /// keep their keys until compaction rewrites them.
    #[must_use]
    pub fn key_domains(mut self, domains: Arc<dyn crate::KeyDomains>) -> Self {
        self.key_domains = Some(domains);
        self
    }

    /// Installs a merge operator for commutative operations.
    ///
    /// When set, enables [`crate::AbstractTree::merge`] which stores partial updates
    /// (operands) that are lazily combined during reads and compaction.
    #[must_use]
    pub fn with_merge_operator(mut self, op: Option<Arc<dyn MergeOperator>>) -> Self {
        self.merge_operator = op;
        self
    }

    /// Sets a custom user key comparator.
    ///
    /// When configured, all key ordering (memtable, block index, merge,
    /// range scans) uses this comparator instead of the default lexicographic
    /// byte ordering.
    ///
    /// # Important
    ///
    /// The comparator's [`crate::UserComparator::name`] is persisted when a tree is
    /// first created. On subsequent opens the stored name is compared against
    /// the supplied comparator's name — a mismatch causes the open to fail
    /// with [`Error::ComparatorMismatch`](crate::Error::ComparatorMismatch).
    #[must_use]
    pub fn comparator(mut self, comparator: SharedComparator) -> Self {
        self.comparator = comparator;
        self
    }

    /// Sets the block-level encryption provider for encryption at rest.
    ///
    /// When set, all blocks written to SST files are encrypted after
    /// compression and before checksumming, using the provided
    /// [`EncryptionProvider`]. The compression dictionaries the tree stores
    /// are sealed with it too, since a dictionary keeps literal stretches of
    /// the records it was trained on.
    ///
    /// The caller is responsible for key management and rotation.
    /// See `crate::Aes256GcmProvider` (behind the `encryption` feature)
    /// for a ready-to-use AES-256-GCM implementation.
    ///
    /// **Important constraints:**
    /// - Encryption state is NOT recorded in SST metadata. Opening an
    ///   encrypted tree without the correct provider (or vice versa) will
    ///   cause block validation errors, not silent corruption.
    /// - Blob files (KV-separated large values) are NOT covered by
    ///   block-level encryption. Large values stored via KV separation
    ///   remain in plaintext on disk.
    #[must_use]
    pub fn with_encryption(mut self, encryption: Option<Arc<dyn EncryptionProvider>>) -> Self {
        self.encryption = encryption;
        self
    }

    /// Sets the wall clock this tree reads: the creation time stamped on its
    /// tables and blob files, those a repair salvages included, and the time
    /// FIFO compaction measures TTL against.
    ///
    /// Without one the tree reads the process clock: [`SystemClock`](crate::SystemClock)
    /// under `std`, or the clock registered with `set_clock` under `no_std`.
    ///
    /// A fixed clock makes the files a tree writes reproducible: two trees
    /// given the same fixed clock, the same writes in the same order and the
    /// same configuration write byte-identical tables and blob files, wherever
    /// their folders are. Encryption breaks that, since every sealed block
    /// takes a fresh nonce.
    ///
    /// # Examples
    ///
    /// ```
    /// # use lsm_tree::{Clock, Config, SequenceNumberCounter};
    /// # use std::{sync::Arc, time::Duration};
    /// struct Epoch;
    /// impl Clock for Epoch {
    ///     fn unix_time(&self) -> Duration {
    ///         Duration::from_secs(1_700_000_000)
    ///     }
    /// }
    /// # let folder = tempfile::tempdir()?;
    /// let config = Config::new(
    ///     folder.path(),
    ///     SequenceNumberCounter::default(),
    ///     SequenceNumberCounter::default(),
    /// )
    /// .with_clock(Arc::new(Epoch));
    /// # Ok::<(), Box<dyn std::error::Error>>(())
    /// ```
    #[must_use]
    pub fn with_clock(mut self, clock: Arc<dyn crate::Clock>) -> Self {
        self.clock = Some(clock);
        self
    }

    /// Sets what `Tree::open` does with a torn tail of the manifest edit log.
    ///
    /// The only difference between the modes is an edit-log record the writer
    /// never finished appending (power loss mid-append): the default
    /// ([`ManifestRecoveryMode::AbsoluteConsistency`]) fails the open,
    /// [`ManifestRecoveryMode::TolerateCorruptedTailRecords`] drops it and
    /// opens at the last complete edit. That record was never acknowledged, so
    /// dropping it loses no committed state. Damage to anything committed (a
    /// snapshot section, a fully written edit) fails the open in either mode;
    /// [`Config::repair`] rebuilds the manifest from the tables on disk.
    ///
    /// See the [`ManifestRecoveryMode`] doc for per-variant semantics.
    #[must_use]
    pub fn manifest_recovery_mode(mut self, mode: ManifestRecoveryMode) -> Self {
        self.manifest_recovery_mode = mode;
        self
    }

    /// Sets the durability level for every fsync the tree issues.
    ///
    /// Defaults to [`SyncMode::Normal`] (plain `fsync`, matching `RocksDB` /
    /// `SQLite` defaults). Pass [`SyncMode::Full`] to force `F_FULLFSYNC` on
    /// macOS for power-loss durability without an external journal — at a
    /// large per-flush cost. Pass [`SyncMode::Barrier`] to keep writes in
    /// order at a barrier's cost and make them durable with
    /// [`AbstractTree::sync_devices`](crate::AbstractTree::sync_devices), once
    /// per batch of flushes. On non-macOS platforms all three are plain
    /// `fsync`.
    #[must_use]
    pub fn sync_mode(mut self, mode: SyncMode) -> Self {
        self.sync_mode = mode;
        self
    }

    /// Sets when the tables a flush writes become durable.
    ///
    /// Defaults to [`FlushDurability::Synced`]: a flush that returns `Ok` is
    /// on disk. [`FlushDurability::Deferred`] leaves a flush unsynced until
    /// [`AbstractTree::persist_flushes`](crate::AbstractTree::persist_flushes),
    /// for a caller whose own write-ahead log already makes its writes
    /// durable.
    ///
    /// # 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()?;
    /// tree.insert("a", "1", 0);
    /// // Written and readable, not yet durable: the caller's log still holds it.
    /// tree.flush_active_memtable(0)?;
    /// // Durable from here: the caller may drop its log up to this point.
    /// tree.persist_flushes()?;
    /// # Ok::<(), lsm_tree::Error>(())
    /// ```
    #[must_use]
    pub fn flush_durability(mut self, durability: FlushDurability) -> Self {
        self.flush_durability = durability;
        self
    }

    /// Sets how many bytes a table or blob file gathers before its writer
    /// starts writing them back, without waiting (default 1 MiB; `0` starts
    /// none).
    ///
    /// The final sync of a large file otherwise writes all its dirty pages at
    /// once, and on a device shared with the caller's journal that burst
    /// delays the journal's own syncs. Writeback makes nothing durable; the
    /// final sync still does. Only Linux starts one
    /// (`sync_file_range(SYNC_FILE_RANGE_WRITE)`); elsewhere this is unused.
    ///
    /// # Examples
    ///
    /// ```
    /// use lsm_tree::{Config, SequenceNumberCounter};
    ///
    /// let folder = tempfile::tempdir()?;
    /// let tree = Config::new(
    ///     &folder,
    ///     SequenceNumberCounter::default(),
    ///     SequenceNumberCounter::default(),
    /// )
    /// .writeback_bytes(4 * 1_024 * 1_024)
    /// .open()?;
    /// # Ok::<(), lsm_tree::Error>(())
    /// ```
    #[must_use]
    pub fn writeback_bytes(mut self, bytes: u64) -> Self {
        self.writeback_bytes = bytes;
        self
    }

    /// Sets the size a flush or an ingestion cuts its output tables at
    /// (default [`DEFAULT_TABLE_TARGET_SIZE`], 64 MiB).
    ///
    /// A memtable larger than this flushes into several tables of one run;
    /// an ingestion cuts its stream the same way, at whatever level it writes
    /// to. Compaction output is sized by its strategy, not by this value.
    ///
    /// The leveled L0 trigger counts runs, so the cut does not change how
    /// often L0 merges into L1; it bounds how much per-key state a flush's
    /// writer holds. A flush of fewer, larger tables writes less in the
    /// merges that follow, so with a memtable much larger than the default a
    /// matching target is the cheaper shape, as far as the writer's memory
    /// allows.
    ///
    /// # Examples
    ///
    /// ```
    /// use lsm_tree::{Config, SequenceNumberCounter};
    ///
    /// let folder = tempfile::tempdir()?;
    /// let tree = Config::new(
    ///     &folder,
    ///     SequenceNumberCounter::default(),
    ///     SequenceNumberCounter::default(),
    /// )
    /// .table_target_size(8 * 1_024 * 1_024)
    /// .open()?;
    /// # Ok::<(), lsm_tree::Error>(())
    /// ```
    #[must_use]
    pub fn table_target_size(mut self, bytes: u64) -> Self {
        self.table_target_size = bytes;
        self
    }

    /// Sets the edit-log rotation threshold in bytes (default 1 MiB).
    ///
    /// Once the manifest edit log exceeds this size, the next version upgrade
    /// writes a fresh full snapshot and starts an empty log instead of appending
    /// another edit. Lower it to shorten recovery replay and cap log size at the
    /// cost of more frequent full-snapshot writes; `0` rotates on every upgrade.
    #[must_use]
    pub fn manifest_log_rotate_bytes(mut self, bytes: u64) -> Self {
        self.manifest_log_rotate_bytes = bytes;
        self
    }

    /// Sets how many empty table files each tables folder creates at a time,
    /// ahead of the tables written into them (default 16).
    ///
    /// A new table's name has to be durable before the manifest edit that
    /// names it, which costs a sync of its folder. A pooled file's name was
    /// made durable with its batch, one folder sync for the whole batch, so a
    /// flush or a compaction into one syncs its tables and the manifest edit
    /// only. The next batch is made off the writing thread once a claim
    /// leaves half a batch or less. An unused pooled file is empty and takes
    /// no space; the next open
    /// uses it again. `0` creates each table file as its table is written and
    /// syncs the folder at every install that names one.
    ///
    /// # Examples
    ///
    /// ```
    /// use lsm_tree::{Config, SequenceNumberCounter};
    ///
    /// let folder = tempfile::tempdir()?;
    /// let tree = Config::new(
    ///     &folder,
    ///     SequenceNumberCounter::default(),
    ///     SequenceNumberCounter::default(),
    /// )
    /// .table_file_pool(64)
    /// .open()?;
    /// # Ok::<(), lsm_tree::Error>(())
    /// ```
    #[must_use]
    pub fn table_file_pool(mut self, files: usize) -> Self {
        self.table_file_pool = files;
        self
    }

    /// Refills a table file pool on the thread of the write that claims from
    /// it, rather than off it (default `false`). A write then waits on the
    /// batch it starts, and every sync the pool issues falls inside one: a
    /// test that counts an operation's syncs, or fails one of them, sees the
    /// same ones on every run.
    #[doc(hidden)]
    #[must_use]
    pub fn table_file_refill_inline(mut self, inline: bool) -> Self {
        use crate::table::file_pool::RefillOn;
        self.table_file_refill = if inline {
            RefillOn::ClaimingThread
        } else {
            RefillOn::Background
        };
        self
    }

    /// Sets the retention floor a manifest repair seeds the rebuilt tree with
    /// (default `0`): after [`repair`](Self::repair) /
    /// [`open_or_repair`](Self::open_or_repair) every snapshot at or below
    /// `floor` is refused with
    /// [`Error::SnapshotBelowRetention`](crate::Error::SnapshotBelowRetention),
    /// exactly as the tree refused it before the manifest was lost.
    ///
    /// A normal open needs no help: the manifest carries the floor every
    /// retention-advancing install established (a GC compaction, `clear`, a
    /// table drop, a filtering compaction; see
    /// [`AbstractTree::retention_floor`](crate::AbstractTree::retention_floor)).
    /// A repair rebuilds the manifest from the tables, which do not record it
    /// (a GC compaction zeroes the seqnos of the rows it settles), so only the
    /// deployment knows it: record
    /// [`retention_floor()`](crate::AbstractTree::retention_floor) in your own
    /// durable state (it already folds in every operation that raised the
    /// boundary, so no per-operation bookkeeping is needed) and pass the last
    /// recorded value here. Left at `0`, a repaired tree serves every
    /// snapshot, which is correct only if history was never collected. Has no
    /// effect on an open that finds a manifest.
    #[must_use]
    pub fn repair_retention_floor(mut self, floor: crate::SeqNo) -> Self {
        self.repair_retention_floor = floor;
        self
    }

    /// Sets the compaction I/O rate limit in bytes per second.
    ///
    /// Caps how fast the compaction worker may issue I/O so background
    /// compaction does not saturate the device and spike user read P99.
    /// `0` (the default) disables throttling. Only compaction is limited;
    /// flush and user reads always pass through.
    ///
    /// The tree builds its own limiter from this figure at open; change it
    /// later through [`AbstractTree::compaction_rate_limiter`](crate::AbstractTree::compaction_rate_limiter).
    /// Ignored when a shared limiter is supplied through
    /// [`Self::compaction_rate_limiter`].
    ///
    /// The figure is the share of the device the deployment gives compaction,
    /// and only the deployment knows it. Below what the write rate times the
    /// strategy's write amplification needs, compaction falls behind: runs
    /// pile up and every read visits more of them. The
    /// [`Planned`](crate::compaction::Planned) strategy reads the limit and
    /// plans a tree it can keep (see its section on the compaction budget);
    /// [`Leveled`](crate::compaction::Leveled) keeps its shape and lags.
    #[must_use]
    pub fn compaction_rate_limit(mut self, bytes_per_sec: u64) -> Self {
        self.compaction_rate_limit = bytes_per_sec;
        self
    }

    /// Supplies a compaction rate limiter shared with other trees, used in
    /// place of the per-tree one built from [`Self::compaction_rate_limit`].
    ///
    /// Hand the same `Arc` to every tree on one device: their compactions are
    /// then bounded by it together, at the rate it holds, whatever each
    /// tree's configured [`Self::compaction_rate_limit`] says. A retune through
    /// [`RateLimiter::set_rate`](crate::rate_limiter::RateLimiter::set_rate)
    /// on any holder applies to all of them on their next request.
    ///
    /// # Examples
    ///
    /// ```
    /// use lsm_tree::rate_limiter::RateLimiter;
    /// use lsm_tree::Config;
    /// use std::sync::Arc;
    ///
    /// # let a = tempfile::tempdir()?;
    /// # let b = tempfile::tempdir()?;
    /// let device = Arc::new(RateLimiter::new(64 * 1_024 * 1_024));
    /// let first = Config::new(a.path(), Default::default(), Default::default())
    ///     .compaction_rate_limiter(Arc::clone(&device))
    ///     .open()?;
    /// let second = Config::new(b.path(), Default::default(), Default::default())
    ///     .compaction_rate_limiter(Arc::clone(&device))
    ///     .open()?;
    /// // Both trees now share the device's 64 MiB/s.
    /// # drop((first, second));
    /// # Ok::<(), lsm_tree::Error>(())
    /// ```
    #[must_use]
    pub fn compaction_rate_limiter(
        mut self,
        limiter: Arc<crate::rate_limiter::RateLimiter>,
    ) -> Self {
        self.compaction_rate_limiter = Some(limiter);
        self
    }

    /// Supplies the pool this tree's compactions reserve the room for their
    /// outputs in, shared with other trees, in place of a pool of its own.
    ///
    /// Hand the same `Arc` to every tree whose files share volumes: each
    /// compaction then reserves against what the others have still to write,
    /// rather than two of them counting on the same free bytes. See
    /// [`CapacityPool`](crate::CapacityPool).
    #[must_use]
    pub fn compaction_capacity(mut self, pool: Arc<crate::CapacityPool>) -> Self {
        self.compaction_capacity = Some(pool);
        self
    }

    /// The pool a tree opened with this configuration reserves its
    /// compactions' room in: the shared one if supplied, else its own.
    pub(crate) fn tree_compaction_capacity(&self) -> Arc<crate::CapacityPool> {
        self.compaction_capacity.clone().unwrap_or_default()
    }

    /// Sets the byte rate writes are paced to while the tree's backpressure
    /// verdict is in its slowdown tier (default `0`: no byte pacing).
    ///
    /// With a rate set,
    /// [`AbstractTree::write_backpressure_for`](crate::AbstractTree::write_backpressure_for)
    /// answers a slowdown with what the write's bytes owe at this rate, so a
    /// caller honouring it writes no faster than the rate however large or
    /// small its writes are; below the slowdown thresholds nothing is debited
    /// and writes run at full speed. The tree builds its own limiter from
    /// this figure at open; change it later through
    /// [`AbstractTree::write_rate_limiter`](crate::AbstractTree::write_rate_limiter).
    /// Ignored when a shared limiter is supplied through
    /// [`Self::write_rate_limiter`].
    ///
    /// # Examples
    ///
    /// ```
    /// use lsm_tree::{Config, SequenceNumberCounter};
    ///
    /// let folder = tempfile::tempdir()?;
    /// let tree = Config::new(
    ///     &folder,
    ///     SequenceNumberCounter::default(),
    ///     SequenceNumberCounter::default(),
    /// )
    /// .write_rate_limit(16 * 1_024 * 1_024)
    /// .open()?;
    /// # Ok::<(), lsm_tree::Error>(())
    /// ```
    #[must_use]
    pub fn write_rate_limit(mut self, bytes_per_sec: u64) -> Self {
        self.write_rate_limit = bytes_per_sec;
        self
    }

    /// Supplies a write rate limiter shared with other trees, used in place of
    /// the per-tree one built from [`Self::write_rate_limit`]: writes paced
    /// under slowdown by any of them draw on one budget, at the rate it holds.
    #[must_use]
    pub fn write_rate_limiter(mut self, limiter: Arc<crate::rate_limiter::RateLimiter>) -> Self {
        self.write_rate_limiter = Some(limiter);
        self
    }

    /// Joins the tree to a [`WriteBufferBudget`](crate::WriteBufferBudget)
    /// shared with other trees: the memtable axis of its backpressure verdict
    /// then compares the unflushed memtable bytes of every member together.
    #[cfg(feature = "std")]
    #[must_use]
    pub fn write_buffer_budget(
        mut self,
        budget: Arc<crate::backpressure::WriteBufferBudget>,
    ) -> Self {
        self.write_buffer_budget = Some(budget);
        self
    }

    /// The limiter a tree opened with this configuration paces slowed-down
    /// writes with: the shared one if supplied, else its own built from
    /// [`Self::write_rate_limit`].
    pub(crate) fn tree_write_rate_limiter(&self) -> Arc<crate::rate_limiter::RateLimiter> {
        self.write_rate_limiter.clone().unwrap_or_else(|| {
            Arc::new(crate::rate_limiter::RateLimiter::new(self.write_rate_limit))
        })
    }

    /// The limiter a tree opened with this configuration throttles its
    /// compactions with: the shared one if supplied, else its own built from
    /// [`Self::compaction_rate_limit`].
    pub(crate) fn tree_compaction_rate_limiter(&self) -> Arc<crate::rate_limiter::RateLimiter> {
        self.compaction_rate_limiter.clone().unwrap_or_else(|| {
            Arc::new(crate::rate_limiter::RateLimiter::new(
                self.compaction_rate_limit,
            ))
        })
    }

    /// Sets the compaction worker-thread count.
    ///
    /// Under `std` this both sizes the per-tree pool built at open when no
    /// shared pool is supplied (see [`Self::compaction_pool`]), which prepares
    /// a table's data blocks and builds its partitioned filter's partitions,
    /// and caps how many range-parallel sub-compactions a compaction splits
    /// into.
    /// `1` keeps compaction serial. Default is `max(1, available_parallelism /
    /// 2)`. Without the `parallel` feature there is no built-in pool, so the
    /// work runs serially even for a value > 1.
    #[cfg(feature = "std")]
    #[must_use]
    pub fn compaction_threads(mut self, threads: usize) -> Self {
        // Clamp to >= 1: the documented semantics treat `1` as "serial", and a
        // 0-thread pool would be an invalid state.
        self.compaction_threads = threads.max(1);
        self
    }

    /// Sets the payload size (bytes) below which a block of a table compressed
    /// in parallel is prepared on the writer thread instead of a worker. A
    /// block that small costs less to compress than to hand to a worker and
    /// back. `None` (default) derives the threshold from the table's block
    /// codec: 1 KiB for lz4; 8 KiB with no codec, where only encryption or
    /// page ECC brings a table to the workers and costs less per block than
    /// lz4; and 0 for zstd, whose per-frame setup makes any block worth a
    /// worker. `Some(0)` sends every block to the workers.
    ///
    /// # Examples
    ///
    /// ```
    /// use lsm_tree::{Config, SequenceNumberCounter};
    /// # let folder = tempfile::tempdir()?;
    /// let config = Config::new(
    ///     folder.path(),
    ///     SequenceNumberCounter::default(),
    ///     SequenceNumberCounter::default(),
    /// )
    /// .parallel_compression_inline_below(Some(1_024));
    /// # Ok::<(), Box<dyn std::error::Error>>(())
    /// ```
    #[cfg(feature = "std")]
    #[must_use]
    pub fn parallel_compression_inline_below(mut self, bytes: Option<u32>) -> Self {
        self.parallel_compression_inline_below = bytes;
        self
    }

    /// The parallel block compression a table writer of this tree uses, or
    /// `None` when no pool is configured.
    #[cfg(feature = "std")]
    pub(crate) fn parallel_compression(&self) -> Option<crate::table::writer::ParallelCompression> {
        self.compaction_pool
            .clone()
            .map(|spawner| crate::table::writer::ParallelCompression {
                spawner,
                threads: self.compaction_threads,
                inline_below: self.parallel_compression_inline_below,
            })
    }

    /// Sets the minimum total input size (bytes) for a compaction to be split
    /// into parallel sub-compactions. Default 8 MiB. `0` splits every eligible
    /// compaction; a large value effectively disables sub-compaction (block
    /// compression still parallelizes via [`Self::compaction_threads`]).
    #[cfg(feature = "std")]
    #[must_use]
    pub fn subcompaction_min_bytes(mut self, bytes: u64) -> Self {
        self.subcompaction_min_bytes = bytes;
        self
    }

    /// Supplies a shared compaction thread pool, used in place of the per-tree
    /// default. Pass one [`crate::table::writer::CompactionSpawner`] (e.g. a
    /// `RayonSpawner` wrapping a shared rayon thread pool) to several trees so
    /// the total worker-thread count stays bounded by the pool size rather than
    /// the number of open trees.
    #[cfg(feature = "std")]
    #[must_use]
    pub fn compaction_pool(
        mut self,
        pool: Option<Arc<dyn crate::table::writer::CompactionSpawner>>,
    ) -> Self {
        self.compaction_pool = pool;
        self
    }

    /// Supplies a pre-trained zstd dictionary for the data block policy to
    /// name.
    ///
    /// The open registers it with the tree, which stores it, so a later open
    /// needs nothing supplied. Which blocks are written against it is decided
    /// by the policy: a [`CompressionType::ZstdDict`] entry names a dictionary
    /// by id, and every block reads back through the id it recorded. The
    /// dictionary should be trained on representative data samples for best
    /// results.
    ///
    /// Create a dictionary with [`ZstdDictionary::new`](crate::ZstdDictionary::new),
    /// then use [`CompressionType::zstd_dict`] to create a matching
    /// compression type:
    ///
    /// ```ignore
    /// use lsm_tree::{CompressionType, ZstdDictionary};
    ///
    /// let dict = ZstdDictionary::new(&training_data);
    /// let compression = CompressionType::zstd_dict(3, dict.id()).unwrap();
    ///
    /// config
    ///     .zstd_dictionary(Some(Arc::new(dict)))
    ///     .data_block_compression_policy(CompressionPolicy::all(compression));
    /// ```
    #[cfg(zstd_any)]
    #[must_use]
    pub fn zstd_dictionary(
        mut self,
        dictionary: Option<Arc<crate::compression::ZstdDictionary>>,
    ) -> Self {
        self.zstd_dictionary = dictionary;
        self
    }
}

#[cfg(test)]
mod builder_tests;