coordinode-lsm-tree 7.2.0

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

use crate::io::{LittleEndian, ReadBytesExt};
use crate::{
    Checksum, SeqNo, TableId, TreeType, coding::Decode, config::ManifestRecoveryMode, fs::Fs,
    version::VersionId, vlog::BlobFileId,
};
#[cfg(not(feature = "std"))]
use alloc::vec::Vec;

use crate::path::Path;

/// Exact on-disk size of a `tables`-section record payload (post-framing):
/// `level: u8 (1) | run: u32 (4) | id: u64 (8) | checksum_type: u8 (1) |
/// checksum: u128 (16) | global_seqno: u64 (8) | recency: u64 (8)`.
///
/// Each record says which level and run it belongs to, and the section holds
/// no run or table counts. The section arrives whole or not at all (it is one checksummed
/// Block bound to the TOC), so counts would add nothing a reader could check
/// against the disk; they could only disagree with the records, which is how a
/// run count once truncated to a byte made a written tree unopenable. `RocksDB`
/// and Pebble keep a file's level in the file's own manifest record for the
/// same reason.
///
/// Stored as `u32` because the framing layer's `len` field is `u32`.
const TABLE_ENTRY_PAYLOAD_LEN: u32 = 1 + 4 + 8 + 1 + 16 + 8 + 8;

/// Where a table sits in the version: its level, and the ordinal of its run
/// within that level. Runs are numbered from 0 in the order the level holds
/// them.
#[derive(Clone, Copy, Debug, Eq, PartialEq, Ord, PartialOrd)]
pub struct TablePlace {
    pub level: u8,
    pub run: u32,
}

/// Encodes one `tables`-section record payload; the inverse of
/// [`decode_table_entry_payload`].
pub fn encode_table_entry_payload(
    payload: &mut Vec<u8>,
    place: TablePlace,
    id: TableId,
    checksum: Checksum,
    global_seqno: SeqNo,
    recency: TableId,
) -> crate::Result<()> {
    use crate::io::WriteBytesExt;

    payload.write_u8(place.level)?;
    payload.write_u32::<LittleEndian>(place.run)?;
    payload.write_u64::<LittleEndian>(id)?;
    payload.write_u8(0)?; // Checksum type, 0 = XXH3
    payload.write_u128::<LittleEndian>(checksum.into_u128())?;
    payload.write_u64::<LittleEndian>(global_seqno)?;
    payload.write_u64::<LittleEndian>(recency)?;
    Ok(())
}

/// Exact on-disk size of a `blob_files`-section record payload (post-framing):
/// `id: u64 (8) | checksum_type: u8 (1) | checksum: u128 (16)`.
const BLOB_ENTRY_PAYLOAD_LEN: u32 = 8 + 1 + 16;

/// Decodes a table-record payload (post-framing) laid out as
/// [`TABLE_ENTRY_PAYLOAD_LEN`] describes. The surrounding framing header
/// (length + XXH3-64) is handled by
/// [`crate::version::framing::read_framed_record`] before this is called.
///
/// Rejects payloads whose length is not exactly
/// [`TABLE_ENTRY_PAYLOAD_LEN`]. The framing layer already verified
/// the XXH3-64 over the payload matched the header digest, so any
/// length mismatch at this point is writer / reader format drift,
/// not on-disk bit-rot, and it aborts recovery.
fn decode_table_entry_payload(payload: &[u8]) -> crate::Result<(TablePlace, RecoveredTable)> {
    if payload.len() != TABLE_ENTRY_PAYLOAD_LEN as usize {
        return Err(crate::Error::InvalidHeader("tables record payload length"));
    }
    let mut cursor = crate::io::Cursor::new(payload);
    let level = cursor.read_u8()?;
    let run = cursor.read_u32::<LittleEndian>()?;
    let id = cursor.read_u64::<LittleEndian>()?;
    let checksum_type = cursor.read_u8()?;
    if checksum_type != 0 {
        return Err(crate::Error::InvalidTag(("ChecksumType", checksum_type)));
    }
    let checksum = Checksum::from_raw(cursor.read_u128::<LittleEndian>()?);
    let global_seqno = cursor.read_u64::<LittleEndian>()?;
    let recency = cursor.read_u64::<LittleEndian>()?;
    Ok((
        TablePlace { level, run },
        RecoveredTable {
            id,
            checksum,
            global_seqno,
            recency,
        },
    ))
}

/// Places a decoded table record into `levels`, which the snapshot's level
/// count has already sized.
///
/// Records arrive in the order the writer emits them: by level, then by run,
/// then by position in the run. A record that names the place of the previous
/// one joins that run; one that names the next run, or run 0 of a later level,
/// opens a new run. A skipped run ordinal, a place behind the previous one, or
/// a level beyond the count is a record no writer produced and is refused.
fn place_table(
    levels: &mut [Vec<Vec<RecoveredTable>>],
    previous: &mut Option<TablePlace>,
    place: TablePlace,
    table: RecoveredTable,
) -> crate::Result<()> {
    const ERR: crate::Error = crate::Error::InvalidHeader("tables record out of place");

    let level = levels.get_mut(usize::from(place.level)).ok_or(ERR)?;
    if *previous == Some(place) {
        level.last_mut().ok_or(ERR)?.push(table);
        return Ok(());
    }
    let follows = match *previous {
        None => place.run == 0,
        Some(prev) if prev.level == place.level => Some(place.run) == prev.run.checked_add(1),
        Some(prev) => prev.level < place.level && place.run == 0,
    };
    if !follows {
        return Err(ERR);
    }
    level.push(vec![table]);
    *previous = Some(place);
    Ok(())
}

/// Reads the `tables` section: the number of levels the layout spans (`u8`),
/// then a sequence of framed records, each naming its own level and run (see
/// [`TABLE_ENTRY_PAYLOAD_LEN`]), grouped back into that many levels. The level
/// count is not a count of records, so it cannot disagree with them; it also
/// keeps the section non-empty for a version without tables, which an
/// encrypted manifest block requires.
///
/// Read strictly, whatever the recovery mode: the section is one checksummed
/// block the snapshot's footer indexes, so it arrives whole, and a record that
/// does not verify, decode or fall in place is a writer defect or a forgery.
/// Opening past one would drop tables the manifest committed, and the open
/// then deletes their files as orphans.
fn read_tables_section(
    bytes: Vec<u8>,
    curr_version_id: VersionId,
    scratch: &mut Vec<u8>,
) -> crate::Result<Vec<Vec<Vec<RecoveredTable>>>> {
    use crate::version::framing::{FRAME_HEADER_LEN, FramedRecordOutcome, read_framed_record};

    let section_len = bytes.len() as u64;
    let mut reader = crate::io::Cursor::new(bytes);
    let level_count = reader
        .read_u8()
        .map_err(|_| crate::Error::InvalidHeader("tables section level count"))?;
    let mut consumed: u64 = 1;
    let mut levels = vec![Vec::new(); usize::from(level_count)];
    let mut previous = None;

    while consumed < section_len {
        match read_framed_record(
            &mut reader,
            section_len - consumed,
            Some(TABLE_ENTRY_PAYLOAD_LEN),
            scratch,
        )? {
            FramedRecordOutcome::Ok => {
                consumed += FRAME_HEADER_LEN as u64 + scratch.len() as u64;
                let (place, table) = decode_table_entry_payload(scratch)?;
                place_table(&mut levels, &mut previous, place, table)?;
            }
            FramedRecordOutcome::TailTruncation => {
                return Err(crate::Error::from(crate::io::Error::new(
                    crate::io::ErrorKind::UnexpectedEof,
                    "manifest tables record truncated mid-frame",
                )));
            }
            FramedRecordOutcome::ChecksumMismatch { expected, got, .. } => {
                return Err(crate::Error::ManifestFrameChecksumMismatch {
                    section: "tables",
                    expected,
                    got,
                });
            }
            FramedRecordOutcome::BadHeader => {
                log::error!(
                    "manifest tables frame header rejected in version \
                     #{curr_version_id}: len exceeds MAX_FRAME_PAYLOAD"
                );
                return Err(crate::Error::InvalidHeader("manifest tables frame header"));
            }
            FramedRecordOutcome::LenMismatch { got, expected } => {
                log::error!(
                    "manifest tables frame len mismatch in version #{curr_version_id}: \
                     declared len={got}, expected {expected}"
                );
                return Err(crate::Error::InvalidHeader(
                    "manifest tables frame len mismatch",
                ));
            }
        }
    }

    Ok(levels)
}

/// Reads the `blob_files` section: `count: u32`, then exactly `count` framed
/// records (see [`decode_blob_entry_payload`]), returned sorted by id. Read
/// strictly, for the same reason as [`read_tables_section`]: a count the
/// records do not fill, or bytes left after them, is a writer defect.
fn read_blob_files_section(
    bytes: Vec<u8>,
    curr_version_id: VersionId,
    scratch: &mut Vec<u8>,
) -> crate::Result<Vec<(BlobFileId, Checksum)>> {
    use crate::version::framing::{FRAME_HEADER_LEN, FramedRecordOutcome, read_framed_record};

    const ERR: crate::Error = crate::Error::InvalidHeader("blob_files section");
    const FRAMED_ENTRY_LEN: u64 = FRAME_HEADER_LEN as u64 + BLOB_ENTRY_PAYLOAD_LEN as u64;

    let section_len = bytes.len() as u64;
    let mut reader = crate::io::Cursor::new(bytes);
    let count = reader.read_u32::<LittleEndian>().map_err(|_| ERR)?;
    // The records are fixed-size, so the section's length says exactly how
    // many it holds; checked before reserving, so a corrupt count cannot turn
    // into a huge allocation.
    if section_len - 4 != u64::from(count) * FRAMED_ENTRY_LEN {
        return Err(ERR);
    }

    let mut consumed: u64 = 4;
    let mut blob_file_ids = Vec::with_capacity(count as usize);
    for _ in 0..count {
        match read_framed_record(
            &mut reader,
            section_len - consumed,
            Some(BLOB_ENTRY_PAYLOAD_LEN),
            scratch,
        )? {
            FramedRecordOutcome::Ok => {
                consumed += FRAMED_ENTRY_LEN;
                blob_file_ids.push(decode_blob_entry_payload(scratch)?);
            }
            FramedRecordOutcome::ChecksumMismatch { expected, got, .. } => {
                return Err(crate::Error::ManifestFrameChecksumMismatch {
                    section: "blob_files",
                    expected,
                    got,
                });
            }
            FramedRecordOutcome::TailTruncation
            | FramedRecordOutcome::BadHeader
            | FramedRecordOutcome::LenMismatch { .. } => {
                log::error!("manifest blob_files record malformed in version #{curr_version_id}");
                return Err(ERR);
            }
        }
    }

    blob_file_ids.sort_by_key(|(id, _)| *id);
    Ok(blob_file_ids)
}

/// Decodes a 25-byte blob-record payload (post-framing): `id: u64 |
/// checksum_type: u8 | checksum: u128`. Same length-check contract as
/// [`decode_table_entry_payload`].
fn decode_blob_entry_payload(payload: &[u8]) -> crate::Result<(BlobFileId, Checksum)> {
    if payload.len() != BLOB_ENTRY_PAYLOAD_LEN as usize {
        return Err(crate::Error::InvalidHeader(
            "blob_files record payload length",
        ));
    }
    let mut cursor = crate::io::Cursor::new(payload);
    let id = cursor.read_u64::<LittleEndian>()?;
    let checksum_type = cursor.read_u8()?;
    if checksum_type != 0 {
        return Err(crate::Error::InvalidTag(("ChecksumType", checksum_type)));
    }
    let checksum = Checksum::from_raw(cursor.read_u128::<LittleEndian>()?);
    Ok((id, checksum))
}

/// Parses the optional `restrictions` section: `count: u32 | repeat(table id:
/// u64, key_len: u32, key bytes)`. Read strictly (no tail tolerance): a
/// restriction is safety-critical — an un-clamped table whose prefix was
/// punched out would read zeroed blocks — so a malformed section aborts rather
/// than silently dropping a clamp. Absent section is handled by the caller
/// (legitimate: no tight-space reclaim ever ran) and never reaches here.
fn parse_restrictions_section(
    mut bytes: &[u8],
) -> crate::Result<crate::HashMap<TableId, crate::UserKey>> {
    const ERR: crate::Error = crate::Error::InvalidHeader("restrictions section");
    let r = &mut bytes;
    let count = r.read_u32::<LittleEndian>().map_err(|_| ERR)?;
    let mut map = crate::HashMap::default();
    for _ in 0..count {
        let id = r.read_u64::<LittleEndian>().map_err(|_| ERR)?;
        let key_len = r.read_u32::<LittleEndian>().map_err(|_| ERR)? as usize;
        if r.len() < key_len {
            return Err(ERR);
        }
        let (head, tail) = r.split_at(key_len);
        *r = tail;
        // Reject a duplicate table id: a corrupt section that lists a table twice
        // could otherwise silently lower an already-advanced bound and un-clamp a
        // punched prefix on reopen.
        if map.insert(id, crate::UserKey::from(head)).is_some() {
            return Err(ERR);
        }
    }
    if !r.is_empty() {
        return Err(ERR);
    }
    Ok(map)
}

/// Parses the optional `blob_restrictions` section: `count: u32 |
/// repeat(blob file id: u64, live-data frontier: u64)`. The blob analogue of
/// [`parse_restrictions_section`], read under the same strictness — a lost or
/// lowered frontier would make integrity checks hash the reclaimed (zeroed)
/// prefix and condemn a healthy blob file. Absent section is handled by the
/// caller (legitimate: no blob prefix was ever reclaimed).
fn parse_blob_restrictions_section(
    mut bytes: &[u8],
) -> crate::Result<crate::HashMap<BlobFileId, u64>> {
    const ERR: crate::Error = crate::Error::InvalidHeader("blob_restrictions section");
    let r = &mut bytes;
    let count = r.read_u32::<LittleEndian>().map_err(|_| ERR)?;
    let mut map = crate::HashMap::default();
    for _ in 0..count {
        let id = r.read_u64::<LittleEndian>().map_err(|_| ERR)?;
        let frontier = r.read_u64::<LittleEndian>().map_err(|_| ERR)?;
        // Reject a duplicate id for the same reason the table section does: a
        // second entry could silently lower an already-advanced frontier.
        if map.insert(id, frontier).is_some() {
            return Err(ERR);
        }
    }
    if !r.is_empty() {
        return Err(ERR);
    }
    Ok(map)
}

/// Parses the optional `retention_floor` section: one `u64 LE`, the highest
/// snapshot seqno the version can no longer serve. Read strictly (exactly
/// eight bytes): a floor read too LOW would let a reopened tree answer a
/// snapshot from data it never saw, which is the very outcome the floor
/// exists to refuse. Absent section is handled by the caller (legitimate: a
/// snapshot written before the floor existed) and never reaches here.
fn parse_retention_floor_section(mut bytes: &[u8]) -> crate::Result<SeqNo> {
    const ERR: crate::Error = crate::Error::InvalidHeader("retention_floor section");
    let r = &mut bytes;
    let floor = r.read_u64::<LittleEndian>().map_err(|_| ERR)?;
    if !r.is_empty() {
        return Err(ERR);
    }
    Ok(floor)
}

/// Reads the registered dictionary ids: `count: u32 | id: u32 * count`.
///
/// Read strictly, like the sections above: a lost id is a dictionary the tree
/// stops accounting for, so the tables written against it would fail to open
/// while its file is left behind as an orphan.
fn parse_dicts_section(mut bytes: &[u8]) -> crate::Result<Vec<crate::file::DictId>> {
    const ERR: crate::Error = crate::Error::InvalidHeader("dicts section");
    let r = &mut bytes;
    let count = r.read_u32::<LittleEndian>().map_err(|_| ERR)? as usize;
    // The ids are fixed-width, so the bytes that remain say how many there can
    // be. Reserving on the count alone would let a corrupt section name four
    // billion ids and turn the recovery into a multi-gigabyte allocation before
    // the first short read could refuse it.
    // Compared by division rather than `count * 4`, which overflows a 32-bit
    // `usize` for the very counts this check exists to reject.
    if count > r.len() / size_of::<crate::file::DictId>() {
        return Err(ERR);
    }
    let mut ids = Vec::with_capacity(count);
    for _ in 0..count {
        ids.push(r.read_u32::<LittleEndian>().map_err(|_| ERR)?);
    }
    if !r.is_empty() {
        return Err(ERR);
    }
    Ok(ids)
}

/// Refuses a snapshot of another format than this engine's as
/// [`crate::Error::InvalidVersion`].
fn check_format(
    archive: &mut crate::manifest_blocks::reader::ManifestArchiveReader,
) -> crate::Result<()> {
    let format = archive.read_section("format_version")?;
    let format = format
        .first()
        .copied()
        .ok_or(crate::Error::InvalidHeader("format_version"))?;
    crate::FormatVersion::try_from(format).map_err(|()| crate::Error::InvalidVersion(format))?;
    Ok(())
}

/// Refuses a snapshot of another format, then one that names another version
/// than `id`, the number its file carries.
///
/// The format label comes first: every section is parsed in the current
/// layout, so a manifest of another format (a 5.x one, whose snapshot names
/// no id) must be refused as such, never misread or reported as damage.
fn check_identity(
    archive: &mut crate::manifest_blocks::reader::ManifestArchiveReader,
    id: VersionId,
) -> crate::Result<()> {
    if named_version(archive)? != id {
        return Err(crate::Error::ManifestFooterInvalid(
            "manifest snapshot names another version than its file",
        ));
    }
    Ok(())
}

/// The version a snapshot names, once its format is this engine's.
fn named_version(
    archive: &mut crate::manifest_blocks::reader::ManifestArchiveReader,
) -> crate::Result<VersionId> {
    check_format(archive)?;
    let named = archive.read_section("version_id")?;
    let named = <[u8; 8]>::try_from(named.as_slice()).map_err(|_| {
        crate::Error::ManifestFooterInvalid("manifest version_id section is not 8 bytes")
    })?;
    Ok(u64::from_le_bytes(named))
}

/// Opens the snapshot `v{id}` in `folder` and verifies it is the snapshot
/// that number names: its footer under `encryption`, its format and its
/// `version_id`. What an open checks before it touches anything else in the
/// tree.
///
/// # Errors
///
/// Returns the reader's error for a snapshot that does not open, and refuses
/// one of another format or one naming another version.
pub fn verify_snapshot(
    folder: &Path,
    id: VersionId,
    fs: &dyn Fs,
    encryption: Option<alloc::sync::Arc<dyn crate::encryption::EncryptionProvider>>,
) -> crate::Result<()> {
    let mut archive = open_snapshot(folder, id, fs, encryption)?;
    check_identity(&mut archive, id)
}

/// The report of a repair the installed snapshot of the tree in `folder`
/// still owes; `None` when there is no snapshot or it says it owes none.
///
/// # Errors
///
/// Propagates a failure to list the folder, and an environmental failure to
/// read the snapshot.
#[cfg(feature = "std")]
pub fn owed_repair_report(
    folder: &Path,
    fs: &dyn Fs,
    encryption: Option<alloc::sync::Arc<dyn crate::encryption::EncryptionProvider>>,
) -> crate::Result<Option<OwedReport>> {
    let Some(id) = installed_snapshot(folder, fs)? else {
        return Ok(None);
    };
    // A snapshot that cannot say whether it owes a report may owe one: its
    // loss must not read as no report, or the replay it described is gone.
    let unreadable = |e: crate::Error| {
        if e.is_environmental() {
            Err(e)
        } else {
            log::warn!("snapshot v{id} cannot say what repair report it owes: {e}");
            Ok(Some(OwedReport::Unreadable(id)))
        }
    };
    let mut archive = match open_snapshot(folder, id, fs, encryption) {
        Ok(archive) => archive,
        Err(e) => return unreadable(e),
    };
    let lists_report = archive.section("repair_report").is_some();
    match named_version(&mut archive) {
        Ok(named) if named == id => {}
        // Another generation's snapshot under this name: what this one owed
        // is not in it.
        Ok(_) => {
            return unreadable(crate::Error::ManifestFooterInvalid(
                "manifest snapshot names another version than its file",
            ));
        }
        Err(e) if e.is_environmental() => return Err(e),
        // A damaged identity section leaves the table of contents readable,
        // and it still says whether a report is owed, and where.
        Err(e) => log::warn!("snapshot v{id} does not say which version it is: {e}"),
    }
    if !lists_report {
        return Ok(None);
    }
    match archive.read_section("repair_report") {
        Ok(bytes) => Ok(Some(OwedReport::Readable(bytes))),
        Err(e) => unreadable(e),
    }
}

/// A repair report the installed snapshot owes.
#[cfg(feature = "std")]
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum OwedReport {
    /// The report's encoded bytes.
    Readable(Vec<u8>),

    /// Snapshot `v{id}`, or the section of it that holds a report, is
    /// damaged: it may owe a report no one can read.
    Unreadable(VersionId),
}

/// The name of the version pointer the format before this one kept beside its
/// snapshots.
const PREVIOUS_FORMAT_POINTER: &str = "current";

/// The on-disk format that kept a version pointer.
const PREVIOUS_FORMAT: u8 = 5;

/// The version id `name` holds as a snapshot (`v{id}`), if it is one.
#[must_use]
pub fn snapshot_id_of(name: &str) -> Option<VersionId> {
    crate::file::canonical_id(name.strip_prefix('v')?)
}

/// The snapshot id whose edit log `name` is (`edits-{id}`), if it is one.
#[must_use]
pub fn log_id_of(name: &str) -> Option<VersionId> {
    crate::file::canonical_id(name.strip_prefix("edits-")?)
}

/// A manifest file of the tree root, as the engine names it.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum ManifestFile {
    /// An installed snapshot, `v{id}`.
    Snapshot(VersionId),

    /// A snapshot not yet installed, `v{id}.tmp`.
    Temporary(VersionId),

    /// The edit log of snapshot `id`, `edits-{id}`.
    Log(VersionId),
}

/// The manifest file `name` is, if the engine could have written it. Any
/// other name in the tree root is not the engine's to remove.
#[must_use]
pub fn manifest_file_of(name: &str) -> Option<ManifestFile> {
    if let Some(id) = snapshot_id_of(name) {
        return Some(ManifestFile::Snapshot(id));
    }
    if let Some(id) = name
        .strip_suffix(super::SNAPSHOT_TEMPORARY_SUFFIX)
        .and_then(snapshot_id_of)
    {
        return Some(ManifestFile::Temporary(id));
    }
    log_id_of(name).map(ManifestFile::Log)
}

/// The id of the installed snapshot of the tree in `folder`, or `None` when
/// the tree has none (a new tree, or one whose creation never completed).
///
/// A snapshot is installed once it carries its name `v{id}`: it is written
/// under a temporary name, its bytes synced, and only then renamed, so the
/// name never holds a snapshot whose bytes were not durable first. The
/// highest id is the tree's; an older generation a rotation left behind is
/// never chosen, being lower, and a temporary is not a snapshot at all.
///
/// The choice is by name alone: [`recover`] opens the chosen snapshot and
/// verifies it, and a failure there is damage to the snapshot the tree was
/// acknowledged on, reported as such, never answered by an older generation,
/// which would drop acknowledged edits.
///
/// A folder with a version pointer (`current`) holds a store of the format
/// before this one, whose writer made a snapshot durable before it rewrote the
/// pointer: a higher snapshot there may be one that was never installed. Such
/// a store is refused as [`crate::Error::InvalidVersion`], for the offline
/// converter, which reads it by its pointer; this engine writes no pointer.
///
/// # Errors
///
/// Propagates a failure to list `folder`, and refuses a store with a version
/// pointer.
pub fn installed_snapshot(folder: &Path, fs: &dyn Fs) -> crate::Result<Option<VersionId>> {
    let entries = match fs.read_dir(folder) {
        Ok(entries) => entries,
        Err(e) if e.kind() == crate::io::ErrorKind::NotFound => return Ok(None),
        Err(e) => return Err(e.into()),
    };
    if entries
        .iter()
        .any(|entry| !entry.is_dir && entry.file_name == PREVIOUS_FORMAT_POINTER)
    {
        return Err(crate::Error::InvalidVersion(PREVIOUS_FORMAT));
    }
    Ok(entries
        .iter()
        .filter(|entry| !entry.is_dir)
        .filter_map(|entry| snapshot_id_of(&entry.file_name))
        .max())
}

/// Opens the snapshot `v{id}` in `folder`: its footer (or the head mirror)
/// verifies and decrypts under `encryption`.
///
/// # Errors
///
/// Returns the reader's error for a snapshot that does not open.
pub fn open_snapshot(
    folder: &Path,
    id: VersionId,
    fs: &dyn Fs,
    encryption: Option<alloc::sync::Arc<dyn crate::encryption::EncryptionProvider>>,
) -> crate::Result<crate::manifest_blocks::reader::ManifestArchiveReader> {
    // The runtime is a placeholder: the reader's ECC decisions are per-Block
    // and self-describing, not driven by it.
    crate::manifest_blocks::reader::ManifestArchiveReader::open(
        &folder.join(alloc::format!("v{id}")),
        fs,
        alloc::sync::Arc::new(crate::runtime_config::RuntimeConfig::default()),
        encryption,
    )
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct RecoveredTable {
    pub id: TableId,
    pub checksum: Checksum,
    pub global_seqno: SeqNo,
    /// The table's L0 recency key. The manifest holds it because an ingested
    /// table's comes from its install, after its file was written.
    pub recency: TableId,
}

#[derive(Debug)]
pub struct Recovery {
    pub tree_type: TreeType,
    /// Version id of the installed on-disk snapshot: the base the edit log
    /// is replayed on top of. The snapshot file
    /// `v{snapshot_id}` and its log `edits-{snapshot_id}` are the generation
    /// that must survive orphan cleanup; intermediate versions live only in the
    /// log. Equals [`Self::curr_version_id`] when the log is empty (just after a
    /// rotation) and is `<=` it otherwise.
    pub snapshot_id: VersionId,
    /// Version id of the recovered state: the snapshot's id advanced by every
    /// edit replayed from the log (so the next persist continues from here).
    pub curr_version_id: VersionId,
    pub table_ids: Vec<Vec<Vec<RecoveredTable>>>,
    pub blob_file_ids: Vec<(BlobFileId, Checksum)>,
    pub gc_stats: crate::blob_tree::FragmentationMap,
    /// Per-table tight-space key-range lower bounds recovered from the snapshot
    /// `restrictions` section and REPLACED wholesale by each replayed edit
    /// (every edit carries its version's full set). A table id present here is
    /// rebuilt as a restricted view ([`super::Version::from_recovery`]); a
    /// lifted restriction is absent from the next edit's set and so dropped.
    pub restrictions: crate::HashMap<TableId, crate::UserKey>,
    /// Per-blob-file live-data frontiers recovered from the snapshot and
    /// REPLACED wholesale by each replayed edit (`blob file id → first live
    /// byte`). A blob file present here had its consumed prefix reclaimed in
    /// place, so its recorded checksum covers only the suffix from this
    /// offset. The blob analogue of [`Self::restrictions`]; wholesale
    /// replacement (never a merge) matters here because blob ids are reused,
    /// so a removed file's stale frontier must not attach to a later file
    /// under the same id.
    pub blob_restrictions: crate::HashMap<BlobFileId, u64>,
    /// Highest snapshot seqno the recovered version can no longer serve,
    /// from the snapshot's `retention_floor` section (`0` when the snapshot
    /// predates it) and overwritten by each replayed edit that carries one.
    /// Seeds the reopened history's boundary so a snapshot a past GC
    /// compaction or `clear` invalidated is refused, not answered from newer
    /// data.
    pub retention_floor: SeqNo,
    /// Ids of the compression dictionaries the recovered version references,
    /// from the snapshot's `dicts` section (empty when the snapshot predates it
    /// or the tree compresses against none) and REPLACED wholesale by each
    /// replayed edit that carries the section. Wholesale, like the restriction
    /// sets above: an edit that drops a dictionary has to be able to say so.
    pub dicts: Vec<crate::file::DictId>,
    /// The encoded report of a repair the tree still owes, from the snapshot's
    /// `repair_report` section. Edits never carry it: only a snapshot does.
    pub repair_report: Option<Vec<u8>>,
}

impl Recovery {
    /// Applies one edit-log [`VersionEdit`](super::edit::VersionEdit) on top of
    /// the recovered snapshot state, in place — the consumer side of the
    /// incremental manifest. Edits are replayed in order after the snapshot to
    /// reconstruct the current version.
    ///
    /// A changed level replaces its run layout wholesale (a dropped table is
    /// simply absent from the new layout; an emptied level becomes zero runs).
    /// Blob files take a per-id add / remove (an added id whose entry already
    /// exists overwrites its checksum). GC stats overwrite when the edit carries
    /// them. The version id advances to the edit's `new_version_id`.
    ///
    /// # Errors
    ///
    /// Returns an error if the edit's GC-stats payload fails to decode.
    pub(crate) fn apply_edit(&mut self, edit: &super::edit::VersionEdit) -> crate::Result<()> {
        for cl in &edit.changed_levels {
            let idx = usize::from(cl.level);
            if idx >= self.table_ids.len() {
                self.table_ids.resize_with(idx + 1, Vec::new);
            }
            let new_layout = cl
                .runs
                .iter()
                .map(|run| {
                    run.iter()
                        .map(|t| RecoveredTable {
                            id: t.id,
                            checksum: Checksum::from_raw(t.checksum),
                            global_seqno: t.global_seqno,
                            recency: t.recency,
                        })
                        .collect()
                })
                .collect();
            // `idx < len` holds: the resize above grew the vec to `idx + 1`.
            if let Some(slot) = self.table_ids.get_mut(idx) {
                *slot = new_layout;
            }
        }

        if !edit.removed_blob_file_ids.is_empty() {
            self.blob_file_ids
                .retain(|(id, _)| !edit.removed_blob_file_ids.contains(id));
        }
        for b in &edit.added_blob_files {
            let checksum = Checksum::from_raw(b.checksum);
            if let Some(entry) = self.blob_file_ids.iter_mut().find(|(id, _)| *id == b.id) {
                entry.1 = checksum;
            } else {
                self.blob_file_ids.push((b.id, checksum));
            }
        }

        if let Some(bytes) = &edit.gc_stats {
            self.gc_stats = crate::blob_tree::FragmentationMap::decode_from(&mut &bytes[..])?;
        }

        // Tight-space restrictions REPLACE wholesale: the encoder derives every
        // edit's `restrictions` / `blob_restrictions` by iterating the new
        // version's tables / blob files, so each edit carries the FULL current
        // set, not a delta. Replacing both advances a bound per slice (the
        // later full set carries the higher bound) AND drops a lifted one (its
        // entry is simply absent). Merging instead would let a removed
        // restricted blob file's frontier outlive it and attach to an
        // unrelated whole file added later under the same id — blob ids are
        // reused (the id counter reseeds from the maximum live id) — making
        // integrity checks hash only that file's suffix. A monotonicity
        // (no-regression) check would be COMPARATOR-RELATIVE — "advancing" is
        // defined by the tree's configured comparator, not byte order — but
        // the comparator is not plumbed into recovery, so a byte-order check
        // would wrongly reject valid advances (or miss real regressions) under
        // a custom/reverse comparator. The edit log is framing-checksummed, so
        // a corrupt/reordered edit is already rejected upstream; the
        // comparator-independent duplicate guard lives in
        // `parse_restrictions_section` for the snapshot path.
        self.restrictions = edit
            .restrictions
            .iter()
            .map(|(id, key)| (*id, key.clone()))
            .collect();
        self.blob_restrictions = edit.blob_restrictions.iter().copied().collect();

        // Carried only when the edit raised it; the value is the version's
        // absolute floor, not a delta.
        if let Some(floor) = edit.retention_floor {
            self.retention_floor = floor;
        }

        // Carried only when the registered set changed, and then in full: an
        // edit that collected a dictionary says so by omitting its id, so this
        // replaces rather than merges.
        if let Some(dicts) = &edit.dicts {
            self.dicts.clone_from(dicts);
        }

        self.curr_version_id = edit.new_version_id;
        Ok(())
    }
}

pub fn recover(
    folder: &Path,
    fs: &dyn Fs,
    mode: ManifestRecoveryMode,
    encryption: Option<alloc::sync::Arc<dyn crate::encryption::EncryptionProvider>>,
) -> crate::Result<Recovery> {
    let curr_version_id = installed_snapshot(folder, fs)?.ok_or(
        crate::Error::ManifestFooterInvalid("the tree holds no installed manifest snapshot"),
    )?;

    log::info!(
        "Recovering current manifest v{curr_version_id} in {} (mode={mode:?})",
        folder.display(),
    );

    // The footer, its head-mirror fallback and the TOC checksums are checked
    // here; a failure is damage to the installed snapshot.
    let mut archive = open_snapshot(folder, curr_version_id, fs, encryption)?;

    // Every caller shares this gate, repair included, which would otherwise
    // rebuild an old store in the current format.
    check_identity(&mut archive, curr_version_id)?;

    // The snapshot's sections are read strictly whatever `mode` says: each is
    // one checksummed block the snapshot's footer indexes, so it arrives whole,
    // and a record inside it that does not verify or decode is a writer defect
    // or a forgery, not a torn write. `mode` governs only the edit log below.

    // Scratch buffer threaded through every `read_framed_record` call across
    // both the `tables` and `blob_files` sections, so per-record heap
    // allocations during recovery are zero after the initial growth.
    let mut read_scratch: Vec<u8> = Vec::with_capacity(64);

    let levels = read_tables_section(
        archive.read_section("tables")?,
        curr_version_id,
        &mut read_scratch,
    )?;

    let blob_file_ids = read_blob_files_section(
        archive.read_section("blob_files")?,
        curr_version_id,
        &mut read_scratch,
    )?;
    debug_assert!(blob_file_ids.is_sorted_by_key(|(id, _)| id));

    let gc_stats = crate::blob_tree::FragmentationMap::decode_from(&mut crate::io::Cursor::new(
        archive.read_section("blob_gc_stats")?,
    ))?;

    // Optional tight-space restrictions section. Absent on versions that never
    // ran tight-space reclaim (→ empty). When present it is read strictly: a
    // restriction is safety-critical, so a malformed section aborts rather than
    // silently un-clamping a punched table.
    let restrictions = if archive.section("restrictions").is_some() {
        parse_restrictions_section(&archive.read_section("restrictions")?)?
    } else {
        crate::HashMap::default()
    };

    // The blob analogue, read under the same strictness: a lost frontier would
    // make integrity checks hash a reclaimed (zeroed) prefix and condemn a
    // healthy blob file.
    let blob_restrictions = if archive.section("blob_restrictions").is_some() {
        parse_blob_restrictions_section(&archive.read_section("blob_restrictions")?)?
    } else {
        crate::HashMap::default()
    };

    // The retention floor, absent on snapshots written before it existed
    // (→ 0: every snapshot servable, the pre-floor behaviour). Read strictly
    // when present, since a floor read too low serves data a snapshot never
    // saw.
    let retention_floor = if archive.section("retention_floor").is_some() {
        parse_retention_floor_section(&archive.read_section("retention_floor")?)?
    } else {
        0
    };

    // The dictionary ids, absent on a snapshot written before the section
    // existed or by a tree that compresses against none (→ empty, which is
    // what such a tree holds).
    let dicts = if archive.section("dicts").is_some() {
        parse_dicts_section(&archive.read_section("dicts")?)?
    } else {
        Vec::new()
    };

    let mut recovery = Recovery {
        tree_type: {
            if archive.section("tree_type").is_none() {
                log::error!(
                    "tree_type section not found in version #{curr_version_id} - maybe the file is corrupted?"
                );
                return Err(crate::Error::Unrecoverable);
            }
            let section_bytes = archive.read_section("tree_type")?;
            let byte = section_bytes
                .first()
                .copied()
                .ok_or(crate::Error::InvalidHeader("TreeType"))?;
            TreeType::try_from(byte).map_err(|()| crate::Error::InvalidHeader("TreeType"))?
        },
        snapshot_id: curr_version_id,
        curr_version_id,
        table_ids: levels,
        blob_file_ids,
        gc_stats,
        restrictions,
        blob_restrictions,
        retention_floor,
        dicts,
        repair_report: if archive.section("repair_report").is_some() {
            Some(archive.read_section("repair_report")?)
        } else {
            None
        },
    };

    // Replay the incremental edit log layered on top of the snapshot. The log
    // `edits-{snapshot_id}` holds every VersionEdit appended since the snapshot
    // was written; applying them in order reconstructs the current version.
    // A writer-incomplete trailing edit is rolled back under
    // `TolerateCorruptedTailRecords` and surfaces `TornManifestEditLog` under
    // the default `AbsoluteConsistency`; a fully-framed edit that does not
    // verify fails the open in every mode. A clean end-of-log is always
    // accepted. Each applied edit advances `recovery.curr_version_id` past the
    // snapshot's id.
    let log_path = super::edit_log::log_path(folder, curr_version_id);
    let edits = super::edit_log::replay_log(fs, &log_path, mode)?;
    if !edits.is_empty() {
        log::info!(
            "Replaying {} manifest edit(s) on top of snapshot #{curr_version_id}",
            edits.len(),
        );
        for edit in &edits {
            recovery.apply_edit(edit)?;
        }
    }

    Ok(recovery)
}

#[cfg(test)]
#[expect(
    clippy::expect_used,
    clippy::indexing_slicing,
    reason = "test assertions"
)]
mod tests;