coordinode-lsm-tree 5.0.0

// SPDX-License-Identifier: Apache-2.0
// Copyright (c) 2024-present, fjall-rs
// Copyright (c) 2026-present, Structured World Foundation

mod block_size;
mod compression;
mod filter;
mod hash_ratio;
mod pinning;
mod restart_interval;

pub use block_size::BlockSizePolicy;
pub use compression::CompressionPolicy;
pub use filter::{BloomConstructionPolicy, FilterPolicy, FilterPolicyEntry};
pub use hash_ratio::HashRatioPolicy;
pub use pinning::PinningPolicy;
pub use restart_interval::RestartIntervalPolicy;

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

use crate::{
    AnyTree, BlobTree, Cache, CompressionType, DescriptorTable, SequenceNumberCounter,
    SharedSequenceNumberGenerator, Tree,
    compaction::filter::Factory,
    comparator::{self, SharedComparator},
    encryption::EncryptionProvider,
    file::TABLES_FOLDER,
    fs::{Fs, StdFs, SyncMode},
    merge_operator::MergeOperator,
    path::absolute_path,
    prefix::PrefixExtractor,
    version::DEFAULT_LEVEL_COUNT,
};
use std::{
    ops::Range,
    path::{Path, PathBuf},
    sync::Arc,
};

/// 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 std::fmt::Debug for LevelRoute {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::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 on-disk MANIFEST
/// contains corrupt records.
///
/// Mirrors `RocksDB`'s `WALRecoveryMode` semantics, but applied to the
/// manifest layer (`src/version/recovery.rs`) — lsm-tree itself has no
/// WAL (durability lives one layer up in the parent fjall/keyspace
/// crate's `Journal`). The MANIFEST is the equivalent surface where
/// "loss-tolerance vs strict-consistency" matters at open time.
///
/// The default is [`AbsoluteConsistency`](Self::AbsoluteConsistency) —
/// any corrupt record fails the open. Switching to a more permissive
/// mode is an explicit, informed operator decision: you are trading
/// "the tree might silently come up with missing tables / blob files"
/// for "the tree comes up at all". When a non-default mode drops
/// records, the recovery path emits a `warn!` summary with the
/// AGGREGATE dropped count per section (`tables` / `blob_files`) —
/// individual table IDs / blob-file IDs are NOT enumerated, because
/// they were never decoded in the first place. Operators wanting a
/// per-record audit trail should pair tail-tolerant recovery with an
/// out-of-band integrity scan ([`verify_integrity`](crate::verify::verify_integrity))
/// of the recovered tree.
#[derive(Copy, Clone, Debug, PartialEq, Eq, Default)]
pub enum ManifestRecoveryMode {
    /// Production-safe default. Any per-record decode mismatch (bad
    /// XXH3, invalid tag, truncated TOC entry, declared-count overrun)
    /// aborts the open with the original error. Surfaces every byte
    /// of corruption; never silently drops data.
    #[default]
    AbsoluteConsistency,

    /// Power-loss-at-write-tail salvage. If the per-section iteration
    /// over the `tables` / `blob_files` records runs out of bytes
    /// before the declared count is reached (truncated tail), keep
    /// everything that decoded cleanly before the cut and emit a
    /// `warn!` listing the dropped record counts.
    ///
    /// A declared count that exceeds the section's payload capacity
    /// (e.g. `table_count` claims more entries than the section has
    /// bytes for) is treated as the same "writer committed a count
    /// header then truncated the entries" shape — the recovery
    /// downgrades the original hard fail to a `warn!` and lets the
    /// per-entry decode loop walk bytes-actually-present until the
    /// first `UnexpectedEof`.
    ///
    /// Any decode error that is NOT a clean tail truncation (bad
    /// `checksum_type` tag, etc.) still aborts the open — this mode
    /// is specifically for "the writer never finished" scenarios,
    /// not for arbitrary bit-rot in already-committed bytes.
    TolerateCorruptedTailRecords,

    /// Recover the largest consistent prefix and discard the rest.
    /// Adapts `RocksDB`'s `kPointInTimeRecovery` accept-the-prefix
    /// rule to the level/run/table nesting: on the first
    /// record-decode mismatch inside the `tables` section, the
    /// recovery keeps the records that decoded cleanly *before*
    /// the corrupt one in the current run, plus every complete
    /// earlier run in the same level, plus every complete earlier
    /// level. "Record-decode mismatch" covers ALL three failure
    /// shapes the per-record loop can surface:
    ///
    /// 1. Framing-layer XXH3 mismatch (the 8-byte digest in the
    ///    record header doesn't match `xxh3_64(payload)`).
    /// 2. Framing-header structural failure (`len > MAX_FRAME_PAYLOAD`),
    ///    surfaced as `BadHeader`. Note: `LenMismatch` (decoded `len`
    ///    disagrees with a fixed-length pin) is a SEPARATE hard-abort
    ///    case in every recovery mode, not a record-decode mismatch
    ///    for the purpose of this mode.
    /// 3. Payload decode failure AFTER a clean framing pass —
    ///    e.g. `Error::InvalidTag` from a corrupt `checksum_type`
    ///    byte inside an otherwise-framed-OK record. The framing
    ///    XXH3 happens to cover the corrupt byte too (it's a
    ///    digest of the whole payload), so the bytes decode
    ///    cleanly at the framing layer; the corruption only
    ///    surfaces inside the per-entry decode helper.
    ///
    /// PIT drops the corrupt record itself, the remaining records
    /// of that run, and every level not yet read. The same rule
    /// applies to the `blob_files` section. Clean tail-truncation
    /// is still tolerated, same as
    /// [`TolerateCorruptedTailRecords`](Self::TolerateCorruptedTailRecords).
    PointInTimeRecovery,

    /// Skip each corrupt record individually, keep all others.
    /// Maximum-availability, lossy. On any per-record decode
    /// mismatch — framing-layer XXH3 mismatch, payload-decode
    /// failure inside an otherwise-framed-OK record (e.g.
    /// `Error::InvalidTag` on a corrupt `checksum_type` byte), or
    /// a framing-header `BadHeader` — the reader logs the skip
    /// and advances exactly past the bad record using the
    /// framing-supplied length field. If the length field itself
    /// is unusable (the recorded length is outside the legal
    /// range, so the next-record boundary is unknown), the rest
    /// of that section is dropped. Intended companion to the
    /// `repair_db` tooling tracked as `#303`: this mode recovers
    /// what it can in-place; `repair_db` rebuilds the manifest
    /// from the SST files
    /// themselves when even this mode can't reach a usable state.
    SkipAnyCorruptedRecords,
}

/// 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(()),
        }
    }
}

const DEFAULT_FILE_FOLDER: &str = ".lsm.data";

/// Options for key-value separation
#[derive(Clone, Debug, PartialEq)]
pub struct KvSeparationOptions {
    /// What type of compression is used for blobs
    #[doc(hidden)]
    pub compression: CompressionType,

    /// 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,

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

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

    /// Pre-trained zstd dictionary for blob-file dictionary compression.
    ///
    /// Required when `compression` is [`CompressionType::ZstdDict`].
    /// The `dict_id` in the compression type must match [`ZstdDictionary::id`](crate::ZstdDictionary::id).
    #[cfg(zstd_any)]
    #[doc(hidden)]
    pub zstd_dictionary: Option<std::sync::Arc<crate::compression::ZstdDictionary>>,
}

impl Default for KvSeparationOptions {
    fn default() -> Self {
        Self {
            #[cfg(feature="lz4")]
            compression:   CompressionType::Lz4,

            #[cfg(not(feature="lz4"))]
            compression: CompressionType::None,

            file_target_size: /* 64 MiB */ 64 * 1_024 * 1_024,
            separation_threshold: /* 1 KiB */ 1_024,

            staleness_threshold: 0.25,
            age_cutoff: 0.25,

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

impl KvSeparationOptions {
    /// Sets the blob compression method.
    #[must_use]
    pub fn compression(mut self, compression: CompressionType) -> Self {
        self.compression = compression;
        self
    }

    /// 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 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
    }

    /// Sets the zstd dictionary for blob-file dictionary compression.
    ///
    /// Required when [`compression`](Self::compression) is set to
    /// [`CompressionType::ZstdDict`].  The `dict_id` encoded in the
    /// compression type must equal [`ZstdDictionary::id()`](crate::ZstdDictionary::id) of the
    /// supplied dictionary; [`Config::open`] will return
    /// [`Error::ZstdDictMismatch`](crate::Error::ZstdDictMismatch) if
    /// they disagree.
    #[cfg(zstd_any)]
    #[must_use]
    pub fn dict(mut self, dictionary: std::sync::Arc<crate::compression::ZstdDictionary>) -> Self {
        self.zstd_dictionary = Some(dictionary);
        self
    }
}

/// Tree configuration builder
pub struct Config {
    /// Folder path
    #[doc(hidden)]
    pub path: PathBuf,

    /// Default filesystem backend for levels without an explicit route.
    ///
    /// Defaults to [`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" file)
    pub level_count: u8,

    /// What type of compression is used for data blocks
    pub data_block_compression_policy: CompressionPolicy,

    /// What type of compression is used for index blocks
    pub index_block_compression_policy: CompressionPolicy,

    /// 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,

    /// 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
    pub index_block_partition_size_policy: BlockSizePolicy,

    /// Partition size when using partitioned filters
    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,

    /// Per-block Reed-Solomon Page ECC. When `true`, every block on
    /// disk carries a Reed-Solomon parity trailer; on read, if the
    /// block's XXH3 disagrees with the on-disk bytes, the reader
    /// attempts RS recovery before surfacing the corruption. 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,

    /// 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>>,

    /// 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>>,

    /// 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. Other
    /// modes trade strict correctness for partial-availability after a
    /// disaster; see the enum doc for the operational scenarios that
    /// motivate each mode.
    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; on other platforms both modes are plain
    /// `fsync`. Set via [`Config::sync_mode`].
    pub(crate) sync_mode: SyncMode,

    /// 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,

    /// 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,

    /// Worker-thread count for compaction parallelism (`std` only), used two
    /// ways: it sizes the per-tree block-compression pool built at open when
    /// [`Self::compaction_pool`] is `None`, 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>>,

    /// 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<std::sync::atomic::AtomicBool>,

    /// Pre-trained zstd dictionary for dictionary compression.
    ///
    /// When set together with a [`CompressionType::ZstdDict`] compression
    /// policy, data blocks are compressed using this dictionary. The
    /// dictionary must remain the same for the lifetime of the tree —
    /// opening a tree with a different dictionary will produce
    /// [`Error::ZstdDictMismatch`](crate::Error::ZstdDictMismatch) errors.
    #[cfg(zstd_any)]
    pub(crate) zstd_dictionary: Option<Arc<crate::compression::ZstdDictionary>>,

    /// 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?
impl Default for Config {
    fn default() -> Self {
        Self {
            path: absolute_path(Path::new(DEFAULT_FILE_FOLDER)),
            fs: Arc::new(StdFs),
            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),

            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), // TODO: implement
            filter_block_partition_size_policy: BlockSizePolicy::all(4_096), // TODO: implement

            data_block_compression_policy: ({
                #[cfg(feature = "lz4")]
                let c = CompressionPolicy::new([CompressionType::None, CompressionType::Lz4]);

                #[cfg(not(feature = "lz4"))]
                let c = CompressionPolicy::new([CompressionType::None]);

                c
            }),
            index_block_compression_policy: CompressionPolicy::all(CompressionType::None),

            data_block_hash_ratio_policy: HashRatioPolicy::all(0.0),

            filter_policy: FilterPolicy::all(FilterPolicyEntry::Bloom(
                BloomConstructionPolicy::BitsPerKey(10.0),
            )),

            compaction_filter_factory: None,
            merge_operator: None,

            prefix_extractor: None,

            expect_point_read_hits: false,

            page_ecc: false,

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

            kv_separation_opts: None,

            #[cfg(zstd_any)]
            zstd_dictionary: None,

            comparator: comparator::default_comparator(),
            encryption: None,
            manifest_recovery_mode: ManifestRecoveryMode::AbsoluteConsistency,
            sync_mode: SyncMode::Normal,
            manifest_log_rotate_bytes: 1024 * 1024,
            compaction_rate_limit: 0,

            #[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")]
            subcompaction_min_bytes: crate::compaction::worker::SUBCOMPACTION_MIN_INPUT_BYTES,
            #[cfg(all(test, feature = "std"))]
            fail_one_subcompaction: Arc::new(std::sync::atomic::AtomicBool::new(false)),
        }
    }
}

impl Config {
    /// Initializes a new config
    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 [`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> {
        #[cfg(zstd_any)]
        self.validate_zstd_dictionary()?;

        // 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() {
            AnyTree::Blob(BlobTree::open(self)?)
        } else {
            AnyTree::Standard(Tree::open(self)?)
        })
    }

    /// Validates that every `ZstdDict` entry in compression policies references
    /// a `dict_id` that matches the configured dictionary. Catches mismatches
    /// at open time rather than at first block write/read.
    #[cfg(zstd_any)]
    fn validate_zstd_dictionary(&self) -> crate::Result<()> {
        let dict_id = self.zstd_dictionary.as_ref().map(|d| d.id());

        // NOTE: Only data block policies are validated. Index blocks never
        // carry a dictionary — Writer::use_index_block_compression() downgrades
        // ZstdDict to plain Zstd. Validating index policies here would reject
        // configs that use ZstdDict solely for index blocks even though the
        // writer handles them correctly.
        for ct in self.data_block_compression_policy.iter() {
            if let &CompressionType::ZstdDict {
                dict_id: required, ..
            } = ct
            {
                match dict_id {
                    None => {
                        return Err(crate::Error::ZstdDictMismatch {
                            expected: required,
                            got: None,
                        });
                    }
                    Some(actual) if actual != required => {
                        return Err(crate::Error::ZstdDictMismatch {
                            expected: required,
                            got: Some(actual),
                        });
                    }
                    _ => {}
                }
            }
        }

        // Blob files with ZstdDict compression must have a matching dictionary.
        if let Some(ref kv_opts) = self.kv_separation_opts
            && let CompressionType::ZstdDict {
                dict_id: required, ..
            } = kv_opts.compression
        {
            match kv_opts.zstd_dictionary.as_ref().map(|d| d.id()) {
                None => {
                    return Err(crate::Error::ZstdDictMismatch {
                        expected: required,
                        got: None,
                    });
                }
                Some(actual) if actual != required => {
                    return Err(crate::Error::ZstdDictMismatch {
                        expected: required,
                        got: Some(actual),
                    });
                }
                _ => {}
            }
        }

        Ok(())
    }

    /// 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).
    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 {
    use super::*;
    use crate::{CompressionType, SequenceNumberCounter, compression::ZstdDictionary};
    use std::sync::Arc;

    #[test]
    fn blob_zstd_dict_no_dict_is_rejected() {
        // ZstdDict compression for blobs without providing a dictionary must fail.
        let folder = tempfile::tempdir().unwrap_or_else(|err| panic!("tempdir failed: {err}"));
        let cfg = Config::new(
            folder.path(),
            SequenceNumberCounter::default(),
            SequenceNumberCounter::default(),
        )
        .with_kv_separation(Some(KvSeparationOptions::default().compression(
            CompressionType::ZstdDict {
                level: 3,
                dict_id: 7,
            },
        )));

        assert!(
            matches!(
                cfg.validate_zstd_dictionary(),
                Err(crate::Error::ZstdDictMismatch {
                    expected: 7,
                    got: None
                })
            ),
            "expected ZstdDictMismatch when no dictionary is supplied",
        );
    }

    #[test]
    fn blob_zstd_dict_id_mismatch_is_rejected() {
        // ZstdDict compression with a dictionary whose id doesn't match the
        // compression type's dict_id must fail.
        let folder = tempfile::tempdir().unwrap_or_else(|err| panic!("tempdir failed: {err}"));
        let dict = Arc::new(ZstdDictionary::new(b"sample training data for test"));
        let wrong_dict_id = dict.id().wrapping_add(1);
        let cfg = Config::new(
            folder.path(),
            SequenceNumberCounter::default(),
            SequenceNumberCounter::default(),
        )
        .with_kv_separation(Some(
            KvSeparationOptions::default()
                .compression(CompressionType::ZstdDict {
                    level: 3,
                    dict_id: wrong_dict_id,
                })
                .dict(Arc::clone(&dict)),
        ));

        assert!(
            matches!(
                cfg.validate_zstd_dictionary(),
                Err(crate::Error::ZstdDictMismatch { .. })
            ),
            "expected ZstdDictMismatch when dict_id doesn't match dictionary",
        );
    }

    #[test]
    fn blob_zstd_dict_matching_dict_is_accepted() {
        // ZstdDict compression with a correctly matching dictionary must succeed.
        let folder = tempfile::tempdir().unwrap_or_else(|err| panic!("tempdir failed: {err}"));
        let dict = Arc::new(ZstdDictionary::new(b"sample training data for test"));
        let cfg = Config::new(
            folder.path(),
            SequenceNumberCounter::default(),
            SequenceNumberCounter::default(),
        )
        .with_kv_separation(Some(
            KvSeparationOptions::default()
                .compression(CompressionType::ZstdDict {
                    level: 3,
                    dict_id: dict.id(),
                })
                .dict(Arc::clone(&dict)),
        ));

        assert!(
            cfg.validate_zstd_dictionary().is_ok(),
            "matching dictionary must be accepted",
        );
    }
}

impl Config {
    /// 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>) {
        if let Some(routes) = &self.level_routes {
            for route in routes {
                if route.levels.contains(&level) {
                    return (route.path.join(TABLES_FOLDER), route.fs.clone());
                }
            }
        }
        (self.path.join(TABLES_FOLDER), self.fs.clone())
    }

    /// 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
    }

    /// Enables per-block Reed-Solomon Page ECC.
    ///
    /// When enabled, every block written by this tree carries a
    /// Reed-Solomon parity trailer; on read, if the block's XXH3
    /// disagrees with the on-disk bytes, the reader attempts RS
    /// recovery before surfacing the corruption.
    ///
    /// 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 a
    /// Reed-Solomon 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
    }

    /// 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::Xor`] (RAID-5 single-parity) or
    /// [`EccScheme::ReedSolomon`]. The default
    /// [`EccScheme::Secded`](crate::runtime_config::EccScheme::Secded) is
    /// not yet wired (#255), so enabling ECC without choosing a shard
    /// scheme fails validation — pick `Xor`/`ReedSolomon` explicitly.
    /// 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`].
    ///
    /// **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 {
        self.initial_runtime_config = runtime;
        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
    }

    /// Sets the compression method for data blocks.
    #[must_use]
    pub fn data_block_compression_policy(mut self, policy: CompressionPolicy) -> Self {
        self.data_block_compression_policy = policy;
        self
    }

    /// Sets the compression method for index blocks.
    #[must_use]
    pub fn index_block_compression_policy(mut self, policy: CompressionPolicy) -> Self {
        self.index_block_compression_policy = policy;
        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 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.
    #[must_use]
    pub fn prefix_extractor(mut self, extractor: Arc<dyn PrefixExtractor>) -> Self {
        self.prefix_extractor = Some(extractor);
        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 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 MANIFEST recovery policy for `Tree::open`.
    ///
    /// The default ([`ManifestRecoveryMode::AbsoluteConsistency`]) is the
    /// only choice that's safe for live production: any corrupt record
    /// in the on-disk manifest aborts the open. Switching to a more
    /// permissive mode trades strict correctness for partial
    /// availability after a disaster. The recovery path emits a
    /// `warn!` summary per affected section (aggregate counts: total
    /// table records dropped, total blob-file records dropped,
    /// header truncations) rather than one log line per dropped
    /// record — the dropped records were never decoded in the first
    /// place, so no per-record IDs are available. Always pair the
    /// non-default modes with an out-of-band integrity scan
    /// ([`verify_integrity`](crate::verify::verify_integrity) for
    /// whole-file XXH3 over every SST + blob file, or
    /// [`verify_block_checksums`](crate::verify::verify_block_checksums)
    /// for per-block granularity) before trusting the recovered tree
    /// for writes.
    ///
    /// 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. On non-macOS platforms both modes are
    /// identical (plain `fsync`).
    #[must_use]
    pub fn sync_mode(mut self, mode: SyncMode) -> Self {
        self.sync_mode = mode;
        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 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.
    #[must_use]
    pub fn compaction_rate_limit(mut self, bytes_per_sec: u64) -> Self {
        self.compaction_rate_limit = bytes_per_sec;
        self
    }

    /// Sets the compaction worker-thread count.
    ///
    /// Under `std` this both sizes the per-tree block-compression pool built at
    /// open when no shared pool is supplied (see [`Self::compaction_pool`]) 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 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
    }

    /// Sets the pre-trained zstd dictionary for dictionary compression.
    ///
    /// When set, data blocks using [`CompressionType::ZstdDict`] will be
    /// compressed and decompressed with this dictionary. 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 {
    use super::*;
    use crate::SequenceNumberCounter;

    #[test]
    fn restart_interval_policies_can_be_overridden_independently() {
        let folder = match tempfile::tempdir() {
            Ok(folder) => folder,
            Err(err) => panic!("tempdir failed: {err}"),
        };
        let cfg = Config::new(
            folder.path(),
            SequenceNumberCounter::default(),
            SequenceNumberCounter::default(),
        )
        .data_block_restart_interval_policy(RestartIntervalPolicy::all(7))
        .index_block_restart_interval_policy(RestartIntervalPolicy::all(3));

        assert_eq!(cfg.data_block_restart_interval_policy.first(), Some(&7));
        assert_eq!(cfg.index_block_restart_interval_policy.first(), Some(&3));
    }

    #[test]
    fn fs_aware_builders_thread_to_initial_runtime_config() -> crate::Result<()> {
        // The CoW-disable + reflink toggles default ON and flip via the builder
        // (AC: "controlled via builder"). Verifies the builder threads to the
        // initial RuntimeConfig the Tree opens with; a wiring regression would
        // silently ignore the user's setting. Lives in the ungated builder
        // tests (the behaviour is unrelated to zstd).
        let folder = tempfile::tempdir()?;
        let mk = || {
            Config::new(
                folder.path(),
                SequenceNumberCounter::default(),
                SequenceNumberCounter::default(),
            )
        };

        let dflt = mk();
        assert!(dflt.initial_runtime_config.disable_cow_on_sst_files);
        assert!(dflt.initial_runtime_config.use_reflink_for_checkpoint);

        let off = mk()
            .disable_cow_on_sst_files(false)
            .use_reflink_for_checkpoint(false);
        assert!(!off.initial_runtime_config.disable_cow_on_sst_files);
        assert!(!off.initial_runtime_config.use_reflink_for_checkpoint);
        Ok(())
    }

    #[test]
    #[should_panic(expected = "index block restart interval must be greater than zero")]
    fn index_restart_interval_policy_rejects_zero_values() {
        let folder = match tempfile::tempdir() {
            Ok(folder) => folder,
            Err(err) => panic!("tempdir failed: {err}"),
        };
        let _cfg = Config::new(
            folder.path(),
            SequenceNumberCounter::default(),
            SequenceNumberCounter::default(),
        )
        .index_block_restart_interval_policy(RestartIntervalPolicy::all(0));
    }

    #[test]
    #[should_panic(expected = "data block restart interval must be greater than zero")]
    fn data_restart_interval_policy_rejects_zero_values() {
        let folder = match tempfile::tempdir() {
            Ok(folder) => folder,
            Err(err) => panic!("tempdir failed: {err}"),
        };
        let _cfg = Config::new(
            folder.path(),
            SequenceNumberCounter::default(),
            SequenceNumberCounter::default(),
        )
        .data_block_restart_interval_policy(RestartIntervalPolicy::all(0));
    }

    #[test]
    #[should_panic(expected = "restart interval policy may not be empty")]
    fn index_restart_interval_policy_rejects_empty() {
        let folder = match tempfile::tempdir() {
            Ok(folder) => folder,
            Err(err) => panic!("tempdir failed: {err}"),
        };
        let _cfg = Config::new(
            folder.path(),
            SequenceNumberCounter::default(),
            SequenceNumberCounter::default(),
        )
        .index_block_restart_interval_policy(RestartIntervalPolicy::new([]));
    }

    #[test]
    #[should_panic(expected = "restart interval policy may not be empty")]
    fn data_restart_interval_policy_rejects_empty() {
        let folder = match tempfile::tempdir() {
            Ok(folder) => folder,
            Err(err) => panic!("tempdir failed: {err}"),
        };
        let _cfg = Config::new(
            folder.path(),
            SequenceNumberCounter::default(),
            SequenceNumberCounter::default(),
        )
        .data_block_restart_interval_policy(RestartIntervalPolicy::new([]));
    }
}