lsm_tree/

repair.rs

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

//! Last-resort `MANIFEST` reconstruction from the SST files on disk.
//!
//! Once a tree has a `MANIFEST`, that manifest is a single point of failure for
//! the database as a whole: a corrupt manifest means the tree cannot open at
//! all, even when every SST on disk is intact. Repair scans the table folder(s),
//! reads each SST's own metadata, and writes a fresh manifest referencing what
//! is actually present.
//!
//! ## What is recovered, what is lost
//!
//! Every readable SST is preserved. What the rebuilt manifest cannot know is the
//! LSM level structure (which file lived at which level) and any version edits
//! that had not yet been durably logged (an in-flight compaction's output
//! placement, recent table deletions). Following the RocksDB `RepairDB()`
//! pattern, all recovered SSTs are placed at L0 ordered by sequence number
//! (newest first) and a normal background compaction redistributes them into
//! proper levels on the next open. Reads are correct throughout: L0 permits
//! overlapping runs, and the merge reader resolves the latest value by sequence
//! number regardless of physical placement.
//!
//! ## Correctness of the recomputed table checksum
//!
//! The manifest binds each table by its whole-file XXH3-128 checksum. A normal
//! write computes that digest incrementally as the file is streamed out, and the
//! file is written strictly sequentially (no seek-back rewrites after the digest
//! is taken), so the on-disk bytes equal the hashed byte stream. Repair therefore
//! recomputes the identical digest by streaming the file start to end. The data
//! itself is protected independently by per-block checksums, which
//! [`Table::recover`] validates as it parses, so an SST that survives recovery is
//! structurally sound.
//!
//! ## Scope
//!
//! KV-separated (blob) trees are supported: the `blobs/` folder is scanned to
//! rediscover the blob files and record them in the rebuilt manifest. Blob-file
//! fragmentation statistics cannot be reconstructed from a directory scan
//! (they are derived from compaction history), so they start empty; blob GC is
//! advisory and re-learns reclaimable space over time without dropping data.

use crate::{
    SeqNo, Table, TableId, UserKey,
    config::{Config, TreeType},
    version::{BlobFileList, Level, Run, Version},
};
use std::{path::PathBuf, sync::Arc};

/// Per-file repair failures: `(path, human-readable reason)`. Mirrors
/// [`RepairReport::unreadable_files`].
type UnreadableFiles = Vec<(PathBuf, String)>;

/// Outcome of a [`Config::repair`] run.
///
/// `recovered` plus `unreadable` accounts for every SST-named file the scan
/// considered. `unreadable_files` carries the per-file reason a file was skipped
/// so an operator can decide whether to investigate or discard it.
#[derive(Debug)]
pub struct RepairReport {
    /// Number of SSTs whose metadata parsed and that are now referenced by the
    /// rebuilt manifest (including any recovered by salvage; see [`salvaged`]).
    ///
    /// [`salvaged`]: RepairReport::salvaged
    pub recovered: usize,

    /// Of [`recovered`](RepairReport::recovered), how many were recovered by
    /// block-level salvage (their original failed whole-file recovery, so the
    /// salvaged copy may be missing the key ranges of corrupt blocks). Always
    /// zero unless repair ran with salvage enabled
    /// ([`Config::repair_with_salvage`]).
    pub salvaged: usize,

    /// Number of SST-named files that could not be opened or parsed and were
    /// therefore left out of the manifest.
    pub unreadable: usize,

    /// Path and human-readable error for each unreadable file.
    pub unreadable_files: Vec<(PathBuf, String)>,

    /// HEALTHY files the rebuild deliberately left out because their content
    /// lives on in the kept tables: a derived compaction output whose inputs
    /// all survived, or an input a surviving output (chain) fully covers.
    /// These opened and verified fine — they are not failures, contribute no
    /// lost coverage, and are removed once the rebuilt manifest is durable.
    /// `(path, reason)` per exclusion.
    pub excluded_files: Vec<(PathBuf, String)>,

    /// Key coverage the rebuilt manifest LOST, for every excluded table whose
    /// metadata still parsed: `(path, first key, last key, highest seqno)`.
    ///
    /// Losing a table's bytes loses what it said about those keys. Older
    /// versions of them survive in other tables and become visible again — a
    /// value the lost table had overwritten, or a key its tombstone had
    /// deleted. No repair can tell those apart without the lost bytes, and
    /// deleting the range instead would destroy intact data (see
    /// [`repair_with_resurrection`](Config::repair_with_resurrection)), so the
    /// range is reported rather than acted on: within these bounds, at or below
    /// the given sequence number, the tree may now serve a superseded value.
    ///
    /// The sequence bound is `None` when the table's own sequence base lived in
    /// the lost manifest — a bulk-ingested SST stores every entry at local
    /// seqno `0` and takes its effective ordering from a manifest-only offset,
    /// which is exactly why such a table is excluded. Reporting the on-disk
    /// local maximum there would scope the affected history far too low, so the
    /// bound is reported as unknown and the whole history of that key range has
    /// to be treated as affected.
    ///
    /// A KEPT salvaged copy contributes an entry too: it dropped corrupt
    /// blocks (or blob records), so within its bounds a superseded value may
    /// likewise now be served, even though the table itself is in the
    /// manifest. The entry carries the SOURCE's coverage (captured while its
    /// metadata was readable), not the replacement's: salvage may have
    /// dropped the block holding the source's outermost keys or highest
    /// seqno, and bounds derived from the survivors would exclude exactly
    /// the lost part.
    ///
    /// Empty when nothing was lost. A table whose metadata was unreadable
    /// contributes no entry here — its coverage is unknowable, and it is
    /// listed in [`unknowable_losses`](Self::unknowable_losses) instead.
    pub lost_coverage: Vec<(PathBuf, UserKey, UserKey, Option<SeqNo>)>,

    /// Table files whose loss cannot be scoped at all: their metadata never
    /// parsed, so neither the affected key range nor a seqno bound is
    /// derivable. Covers both an EXCLUDED unreadable table and a KEPT lossy
    /// salvaged copy whose source's metadata was unreadable (the copy's own
    /// metadata only bounds what survived). Any entry here forces
    /// [`wal_replay_scope`](Self::wal_replay_scope) to
    /// [`WalReplayScope::FullHistory`], since no bound can prove a retained
    /// record is NOT affected. Blob files are not listed: losing blob content
    /// surfaces through the referencing tables, which the other fields cover.
    pub unknowable_losses: Vec<PathBuf>,

    /// Damaged blob files whose salvaged replacement IS installed in the
    /// rebuilt manifest: the canonical (installed) path plus a note on what
    /// was recovered and where the damaged original is preserved. Disjoint
    /// from [`unreadable_files`](RepairReport::unreadable_files), which lists
    /// only files LEFT OUT of the manifest. Empty for standard trees and for
    /// repairs where no blob file needed salvage.
    pub blob_files_salvaged: Vec<(PathBuf, String)>,

    /// Dictionary files whose bytes no longer hashed to their name, moved to
    /// `{id}.damaged` beside them once the rebuilt manifest was durable:
    /// `(new path, reason)` each. Left under its name, such a file would fail
    /// every open of the repaired tree. A table that needed one is not in this
    /// list: it fails the repair instead, which a re-run supplying that
    /// dictionary recovers from.
    pub damaged_dictionaries: Vec<(PathBuf, String)>,

    /// Description of the level-assignment strategy used (constant for now;
    /// surfaced so the report is self-explanatory and forward-compatible).
    pub method: &'static str,

    /// Operator-facing caveats about the rebuilt state.
    pub warnings: Vec<&'static str>,

    /// Identifies this repair's report. The tree hands a report out until
    /// [`AbstractTree::acknowledge_repair_report`](crate::AbstractTree::acknowledge_repair_report)
    /// takes this id, so it may arrive more than once; a caller that records
    /// the ids it applied skips a repeat.
    pub id: u128,

    /// The ids of earlier reports this one takes in: they were still owed
    /// when this repair ran, and their losses are listed here as well. A
    /// caller that applied one of them already has replayed its losses.
    pub includes: Vec<u128>,
}

/// What an external write-ahead log must replay after this repair, derived
/// from [`RepairReport::lost_coverage`] by
/// [`RepairReport::wal_replay_scope`].
///
/// A repair can REGRESS persisted state below the WAL's trim watermark `W`
/// (a dropped or salvaged table loses versions at or below `W`, while
/// [`get_highest_persisted_seqno`] stays high because neighbouring tables
/// survived), so the standard `seqno > W` tail replay is not always
/// sufficient. See `docs/external-wal.md` § Replay after repair for the full
/// recipe, including why a merge operand needs a presence check
/// ([`scan_since_seqno_in_range`]) while a put / delete may be re-applied
/// blindly.
///
/// [`get_highest_persisted_seqno`]: crate::AbstractTree::get_highest_persisted_seqno
/// [`scan_since_seqno_in_range`]: crate::Tree::scan_since_seqno_in_range
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum WalReplayScope {
    /// Nothing was excluded: the standard tail replay (`seqno > W`) is
    /// sufficient.
    TailOnly,
    /// Coverage was lost at or below this sequence number. In addition to the
    /// tail, replay every RETAINED record whose key falls inside a
    /// [`RepairReport::lost_coverage`] range and whose seqno is at or below
    /// this bound (presence-checking merge operands).
    LostUpTo(SeqNo),
    /// At least one excluded table's sequence base was itself lost with the
    /// manifest, so no seqno bound scopes the damage: the whole retained
    /// history must be reconciled (still presence-checking merge operands).
    ///
    /// Read [`RepairReport::unknowable_losses`] to know over WHAT. When it is
    /// EMPTY the damage is still localized, so the
    /// [`lost_coverage`](RepairReport::lost_coverage) ranges bound the work.
    /// When it is NON-EMPTY the loss has no key range either — that table's
    /// metadata never parsed — and the reconciliation runs over the ENTIRE
    /// keyspace in ONE pass; iterating the known ranges on top of it would
    /// subtract the same survivors twice.
    FullHistory,
}

impl RepairReport {
    /// Derives the WAL replay obligation from
    /// [`lost_coverage`](Self::lost_coverage) and
    /// [`unknowable_losses`](Self::unknowable_losses):
    /// [`WalReplayScope::TailOnly`] when nothing was lost,
    /// [`WalReplayScope::FullHistory`] when any loss is unscopable (an
    /// unknown seqno bound, or an excluded table whose coverage never
    /// parsed), else [`WalReplayScope::LostUpTo`] the highest lost bound. The
    /// per-range detail (which KEYS are affected) stays in `lost_coverage`;
    /// this is the aggregate a WAL uses to decide how far back its archive
    /// must reach.
    #[must_use]
    pub fn wal_replay_scope(&self) -> WalReplayScope {
        if !self.unknowable_losses.is_empty() {
            return WalReplayScope::FullHistory;
        }
        let mut ceiling: Option<SeqNo> = None;
        for (_, _, _, bound) in &self.lost_coverage {
            match bound {
                None => return WalReplayScope::FullHistory,
                Some(b) => ceiling = Some(ceiling.map_or(*b, |c| c.max(*b))),
            }
        }
        match ceiling {
            None => WalReplayScope::TailOnly,
            Some(b) => WalReplayScope::LostUpTo(b),
        }
    }
}

/// [`RepairReport::method`] of every repair.
const REPAIR_METHOD: &str = "all-to-L0 with sequence-number ordering";

const WARN_ALL_AT_L0: &str =
    "All recovered tables placed at L0; background compaction will redistribute them";
const WARN_UNLOGGED_EDITS: &str =
    "Recent unlogged version edits (in-flight compactions, recent deletions) are lost";
const WARN_BLOB_FRAGMENTATION: &str = "Blob fragmentation stats reset (punched prefixes reseeded); blob GC re-learns the rest over time";
const WARN_REPORT_COPY_DAMAGED: &str =
    "The kept report of an earlier repair is damaged; its losses are unscoped";

/// Every warning a report can carry; a kept copy of a report records each by
/// its index here.
const WARNINGS: [&str; 4] = [
    WARN_ALL_AT_L0,
    WARN_UNLOGGED_EDITS,
    WARN_BLOB_FRAGMENTATION,
    WARN_REPORT_COPY_DAMAGED,
];

/// Adds what `earlier` lost to `report`: a repair over the snapshot of an
/// earlier repair whose report never reached the caller owes that repair's
/// replay too. The counts stay `report`'s own, describing this run.
///
/// A file left out of the manifest is listed once. A loss is listed once per
/// distinct entry, not per path: a repair of an earlier repair's salvaged
/// replacement names the same path with only what the replacement held, and
/// the earlier, wider range is still owed, as is the note saying where an
/// earlier salvage kept a damaged original.
#[cfg(feature = "std")]
fn fold_report(report: &mut RepairReport, earlier: RepairReport) {
    fn add<T>(into: &mut Vec<T>, from: Vec<T>, same: impl Fn(&T, &T) -> bool) {
        for entry in from {
            if !into.iter().any(|known| same(known, &entry)) {
                into.push(entry);
            }
        }
    }
    add(
        &mut report.unreadable_files,
        earlier.unreadable_files,
        |a, b| a.0 == b.0,
    );
    report.unreadable = report.unreadable_files.len();
    add(
        &mut report.excluded_files,
        earlier.excluded_files,
        |a, b| a.0 == b.0,
    );
    add(
        &mut report.lost_coverage,
        earlier.lost_coverage,
        PartialEq::eq,
    );
    add(
        &mut report.unknowable_losses,
        earlier.unknowable_losses,
        PartialEq::eq,
    );
    add(
        &mut report.blob_files_salvaged,
        earlier.blob_files_salvaged,
        PartialEq::eq,
    );
    add(
        &mut report.damaged_dictionaries,
        earlier.damaged_dictionaries,
        PartialEq::eq,
    );
    for warning in earlier.warnings {
        if !report.warnings.contains(&warning) {
            report.warnings.push(warning);
        }
    }
    for id in core::iter::once(earlier.id).chain(earlier.includes) {
        if !report.includes.contains(&id) {
            report.includes.push(id);
        }
    }
}

/// Streams `path` from byte `start` to end through XXH3-128. `start == 0`
/// reproduces the whole-file digest a normal write accumulates; a non-zero
/// `start` digests only the LIVE suffix of a tight-space RESTRICTED table,
/// whose `[0, start)` prefix was hole-punched (reads back as zeros) once a
/// superseding output table took over those keys. The suffix bytes are
/// untouched by the punch, so this digest is stable across it.
pub(crate) fn compute_table_checksum_from(
    fs: &dyn crate::fs::Fs,
    path: &std::path::Path,
    start: u64,
) -> crate::Result<u128> {
    // The offset-only case is the override-splicing digest with no overrides.
    compute_table_checksum_with_overrides(fs, path, start, &[])
}

/// Streams `path` start to end through XXH3-128, matching the digest a normal
/// table write accumulates via `ChecksummedWriter`.
pub(crate) fn compute_table_checksum(
    fs: &dyn crate::fs::Fs,
    path: &std::path::Path,
) -> crate::Result<u128> {
    // The whole-file case is the override-splicing digest from offset 0 with no
    // overrides: one shared read loop in `compute_table_checksum_with_overrides`.
    compute_table_checksum_with_overrides(fs, path, 0, &[])
}

/// As [`compute_table_checksum`], but streams the file with `overrides` spliced
/// in: each `(offset, bytes)` replaces the on-disk bytes at `[offset,
/// offset + bytes.len())`. Used to predict the digest an in-place heal WILL
/// produce, from the corrected block frames it will write, before any write
/// lands — so the heal attestation can bind that intended post-heal state.
///
/// The overrides are size-preserving block frames at distinct, non-overlapping
/// offsets (the heal rewrites each corrupt block at its existing offset and
/// size), so splicing them is a byte-for-byte substitution that keeps the file
/// length and every other byte unchanged. `start` matches
/// [`compute_table_checksum_from`]: a restricted view predicts only its live
/// suffix (its corrections all lie there), so the digest starts at the punch
/// offset.
pub(crate) fn compute_table_checksum_with_overrides(
    fs: &dyn crate::fs::Fs,
    path: &std::path::Path,
    start: u64,
    overrides: &[(u64, Vec<u8>)],
) -> crate::Result<u128> {
    crate::file::checksum_from_with_overrides(fs, path, start, overrides)
}

/// Highest id any manifest file in `folder` carries (`v{N}` or `edits-{N}`), if
/// any. The rebuilt manifest uses `max + 1` so it supersedes every stale
/// generation and its install never meets a log a predecessor left.
///
/// A directory-read failure is propagated (not swallowed as "no versions"): a
/// transient scan error must not silently reset the version chain to `0` and
/// risk reusing a live version id.
fn highest_existing_version_id(
    fs: &dyn crate::fs::Fs,
    folder: &std::path::Path,
) -> crate::Result<Option<u64>> {
    Ok(fs
        .read_dir(folder)?
        .into_iter()
        .filter_map(|e| {
            crate::version::recovery::snapshot_id_of(&e.file_name)
                .or_else(|| crate::version::recovery::log_id_of(&e.file_name))
        })
        .max())
}

/// Removes a file the rebuilt manifest does not name, together with any
/// companion `.restrict-bound` sidecar, and makes the removal durable.
///
/// A repair has exactly two outcomes: a committed tree that opens, or an error.
/// A file left behind is neither — `Tree::open` rejects a foreign name in
/// `tables/` and sweeps an id the manifest does not reference, so a removal that
/// cannot happen here becomes an open that fails. The removal is therefore not
/// best-effort: a failure fails the repair.
///
/// The bytes are NOT preserved anywhere. Recovering the content of a damaged
/// file is replication, checkpoint plus journal replay, or a backup; a directory
/// hidden beside the tree recovers nothing and only makes a later run derive
/// from a different world than the one it was handed.
///
/// Callers run this strictly AFTER the manifest commit. Until then every source
/// is still the only copy of its rows, and a crash must leave the directory
/// exactly as the retry expects to find it.
///
/// `NotFound` counts as done, so a retry finishes a sweep a crash interrupted.
fn discard_unreferenced(
    fs: &dyn crate::fs::Fs,
    path: &std::path::Path,
    sync_mode: crate::fs::SyncMode,
) -> crate::Result<()> {
    let remove = |target: &std::path::Path| -> crate::Result<()> {
        match fs.remove_file(target) {
            Ok(()) => Ok(()),
            Err(e) if e.kind() == crate::io::ErrorKind::NotFound => Ok(()),
            Err(e) => Err(e.into()),
        }
    };
    // Sidecar first: an SST removed while its `.restrict-bound` survives leaves a
    // bound that a later run would match by id against an unrelated table. Most
    // files have none (blob files never do), so probe before asking.
    let sidecar = crate::restrict_bound::sidecar_path(path);
    if fs.exists(&sidecar)? {
        remove(&sidecar)?;
    }
    remove(path)?;
    if let Some(dir) = path.parent() {
        fs.sync_directory_with(dir, sync_mode)?;
    }
    Ok(())
}

pub(crate) use crate::file::{REPAIR_TMP_SUFFIX, table_id_from_repair_tmp_name};

/// `{id}` -> `{id}.repair-tmp`, the path a repair builds a replacement at.
///
/// The replacement CANNOT be built at `{id}` directly: that would either destroy
/// the source before the manifest commit (a crash then loses the only copy) or,
/// under a fresh id beside the source, leave both readable after a crash — and
/// the retry, unable to tell a half-finished replacement from a real table,
/// would rebuild BOTH into L0, applying every merge operand of that history
/// twice. A name no scan adopts has neither problem: the source stays intact and
/// authoritative until the commit, and a leftover temp is garbage.
fn repair_tmp_path(table_path: &std::path::Path) -> PathBuf {
    let mut name = table_path.file_name().unwrap_or_default().to_os_string();
    name.push(REPAIR_TMP_SUFFIX);
    table_path.with_file_name(name)
}

/// Swaps a finished replacement onto the name the committed manifest gives it:
/// `{id}.repair-tmp` becomes `{id}`, destroying the damaged source it replaces,
/// with any companion `.restrict-bound` sidecar carried along.
///
/// Strictly POST-COMMIT. Before the commit the source is still the only copy of
/// its rows and the manifest still names it; after it, the manifest names this
/// content and the source is what the tree no longer references.
///
/// Sidecar first, and the sidecar's own removal-or-rename settled before the
/// table moves: a table adopted at `{id}` while a STALE `{id}.restrict-bound`
/// survives would be reopened restricted at an unrelated bound, silently hiding
/// its prefix.
///
/// `manifest_restricted` is whether the COMMITTED manifest restricts this id.
/// It disambiguates a retry of an interrupted swap: with the sidecar step
/// preceding the table rename, a crash between them leaves the replacement's
/// sidecar already at the destination and no temp sidecar — a state that is
/// byte-identical to "unrestricted replacement beside the source's stale
/// sidecar". Only the manifest can tell them apart, and it is the authority
/// for the bound anyway.
pub(crate) fn commit_repair_tmp(
    fs: &dyn crate::fs::Fs,
    tmp_path: &std::path::Path,
    table_path: &std::path::Path,
    sync_mode: crate::fs::SyncMode,
    manifest_restricted: bool,
) -> crate::Result<()> {
    let tmp_sidecar = crate::restrict_bound::sidecar_path(tmp_path);
    let dest_sidecar = crate::restrict_bound::sidecar_path(table_path);
    if fs.exists(&tmp_sidecar)? {
        fs.rename(&tmp_sidecar, &dest_sidecar)?;
    } else if manifest_restricted {
        // The manifest restricts this id, so a sidecar already at the
        // destination is NOT the source's stale metadata — it is the
        // replacement's own sidecar, moved by an interrupted earlier attempt.
        // Deleting it would adopt a restricted replacement unrestricted on the
        // next scan (the fresh copy is unpunched, so no geometry re-derives
        // the bound) and resurrect the straddling block's sub-bound rows.
        // Keep it; a missing or disagreeing sidecar is republished from the
        // manifest by the next open.
    } else if fs.exists(&dest_sidecar)? {
        // The replacement is unrestricted, so the source's bound describes bytes
        // that are about to stop existing. Left in place it would reopen the
        // replacement restricted at an unrelated bound and hide its prefix.
        match fs.remove_file(&dest_sidecar) {
            Ok(()) => {}
            Err(e) if e.kind() == crate::io::ErrorKind::NotFound => {}
            Err(e) => return Err(e.into()),
        }
    }
    fs.rename(tmp_path, table_path)?;
    if let Some(dir) = table_path.parent() {
        fs.sync_directory_with(dir, sync_mode)?;
    }
    Ok(())
}

/// Decides whether a leftover `{id}.repair-tmp` is the file the committed
/// manifest describes. The manifest names `id` in BOTH crash cases — after the
/// commit (the entry describes the replacement) and before it (the entry still
/// describes the source) — so id membership alone cannot tell a committed swap
/// from an abandoned build, and swapping the latter in would destroy the file
/// the manifest actually names (a mid-build crash even leaves the temp
/// truncated). The entry's checksum can tell them apart: only a committed
/// repair recorded the temp's digest — the live-suffix digest for a restricted
/// replacement, whose punch offset comes from the temp's own index. A temp
/// that does not even open is the mid-build crash; only transient I/O
/// propagates for a retry.
///
/// A temp whose bytes no longer READ BACK at all (a rotted sector, a build
/// truncated mid-write — non-environmental either way) must not hold the
/// tree hostage: when the ORIGINAL `{id}` still matches the manifest's
/// checksum, the manifest provably names the source, so the unreadable temp
/// is a disposable abandoned build (`Ok(false)`). The temp's failure
/// propagates only when the original cannot be proven authoritative either —
/// the temp could then be a committed-but-unswapped replacement, and
/// discarding it would destroy the only copy the manifest describes.
#[cfg(feature = "std")]
pub(crate) fn repair_tmp_is_published(
    config: &Config,
    fs: &Arc<dyn crate::fs::Fs>,
    tmp_path: &std::path::Path,
    table_id: TableId,
    manifest_checksum: crate::Checksum,
    restriction: Option<&crate::UserKey>,
) -> crate::Result<bool> {
    // Whether the ORIGINAL `{id}` beside the temp is provably the file the
    // manifest names — hashed on the SAME basis the manifest recorded it. A
    // RESTRICTED table's entry holds its LIVE-SUFFIX digest, so hashing the
    // whole original (its reclaimed prefix included) could never match, and
    // the temp's ambiguity could never be resolved: every open and every
    // repair would keep failing on a disposable leftover.
    let original_is_authoritative = || -> crate::Result<bool> {
        let original = tmp_path.with_file_name(table_id.to_string());
        let digest = match restriction {
            None => compute_table_checksum(&**fs, &original),
            Some(bound) => {
                let table = match crate::table::Table::recover(repair_recover_params(
                    config,
                    original.clone(),
                    manifest_checksum,
                    table_id,
                    Arc::clone(fs),
                    None,
                )) {
                    Ok(table) => table,
                    Err(e) if is_environmental(&e) => return Err(e),
                    // An original that does not open proves nothing.
                    Err(_) => return Ok(false),
                };
                match table.punch_offset_for(bound.as_ref()) {
                    Ok(offset) => compute_table_checksum_from(&**fs, &original, offset),
                    Err(e) if is_environmental(&e) => return Err(e),
                    Err(_) => return Ok(false),
                }
            }
        };
        match digest {
            Ok(digest) => Ok(crate::Checksum::from_raw(digest) == manifest_checksum),
            Err(e) if is_environmental(&e) => Err(e),
            Err(_) => Ok(false),
        }
    };
    // The temp is condemned as an abandoned build ONLY when the original
    // proves itself the file the manifest names. Otherwise neither copy is
    // proven and the ambiguity surfaces — deleting the temp here could
    // destroy a committed replacement whose swap never ran, and the original
    // beside it is the damaged pre-repair source.
    let condemn_only_if_proven = |ambiguity: crate::Error| -> crate::Result<bool> {
        if original_is_authoritative()? {
            Ok(false)
        } else {
            Err(ambiguity)
        }
    };
    // A temp that READS but does not match is as ambiguous as one that does
    // not read: it is either an abandoned build or a committed replacement
    // that rotted after its manifest went durable.
    let digest_mismatch = |digest: u128| crate::Error::ChecksumMismatch {
        got: crate::Checksum::from_raw(digest),
        expected: manifest_checksum,
    };
    let Some(bound) = restriction else {
        return match compute_table_checksum(&**fs, tmp_path) {
            Ok(digest) if crate::Checksum::from_raw(digest) == manifest_checksum => Ok(true),
            Ok(digest) => condemn_only_if_proven(digest_mismatch(digest)),
            Err(e) if is_environmental(&e) => Err(e),
            Err(temp_err) => condemn_only_if_proven(temp_err),
        };
    };
    let table = match crate::table::Table::recover(repair_recover_params(
        config,
        tmp_path.to_path_buf(),
        manifest_checksum,
        table_id,
        Arc::clone(fs),
        // This open exists only to locate the temp's punch offset from its own
        // index; no entry is read through it, so the ingest offset is moot.
        None,
    )) {
        Ok(table) => table,
        Err(e) if is_environmental(&e) => return Err(e),
        // A temp that does not open is USUALLY the mid-build crash — but the
        // manifest may equally describe a COMMITTED replacement this run
        // failed to swap, and then discarding it destroys the only published
        // copy. Same proof as the checksum paths: condemn the temp only when
        // the original proves itself the file the manifest names.
        Err(e) => return condemn_only_if_proven(e),
    };
    let punch_offset = match table.punch_offset_for(bound.as_ref()) {
        Ok(offset) => offset,
        Err(e) if is_environmental(&e) => return Err(e),
        Err(e) => return condemn_only_if_proven(e),
    };
    match compute_table_checksum_from(&**fs, tmp_path, punch_offset) {
        Ok(digest) if crate::Checksum::from_raw(digest) == manifest_checksum => Ok(true),
        Ok(digest) => condemn_only_if_proven(digest_mismatch(digest)),
        Err(e) if is_environmental(&e) => Err(e),
        Err(temp_err) => condemn_only_if_proven(temp_err),
    }
}

/// The trustworthy EXACT restriction bound for `table_id`, or `None` when the
/// scan has none and must fall back to the punch geometry.
///
/// One question, one answer, for both salvage arms: the arm that still holds a
/// recovered `Table` and the arm whose whole-file recovery failed ask it
/// identically, and a bound from either source is honored the same way — the
/// only difference is that one reopens the table on it while the other
/// re-imposes it on the salvaged replacement.
///
/// The clean manifest decides in both directions (see [`ManifestRestriction`]);
/// only an `Unknown` manifest consults the sidecar mirror, and there a
/// TRANSIENT read propagates while a missing / id-mismatched / corrupt one
/// leaves no trustworthy bound.
///
/// # Errors
///
/// The digest a RESTRICTED view of `table_path` records: its live suffix from
/// the punch offset `bound` falls at, which is what the manifest holds for a
/// tight-space-punched table (see [`crate::table::Table::reopen_restricted`]).
///
/// `None` when the file cannot be opened or the bound cannot be located in its
/// index: nothing to compare, never a claim of mismatch.
///
/// # Errors
///
/// Propagates an ENVIRONMENTAL failure; anything else answers `None`.
#[cfg(feature = "std")]
fn restricted_suffix_digest(
    config: &Config,
    fs: &Arc<dyn crate::fs::Fs>,
    table_path: &std::path::Path,
    table_id: TableId,
    bound: &crate::UserKey,
) -> crate::Result<Option<crate::Checksum>> {
    // Opening reads the trailer, meta and index only; the data blocks the
    // digest streams are read once, below.
    let table = match Table::recover(repair_recover_params(
        config,
        table_path.to_path_buf(),
        crate::Checksum::from_raw(0),
        table_id,
        Arc::clone(fs),
        None,
    )) {
        Ok(table) => table,
        Err(e) if is_environmental(&e) => return Err(e),
        Err(_) => return Ok(None),
    };
    match table.suffix_checksum_for(Some(bound)) {
        Ok(d) => Ok(Some(d)),
        Err(e) if is_environmental(&e) => Err(e),
        Err(_) => Ok(None),
    }
}

/// Propagates an environmental sidecar read failure, so a retry re-reads it.
#[cfg(feature = "std")]
fn trustworthy_restriction_bound(
    config: &Config,
    fs: &dyn crate::fs::Fs,
    table_path: &std::path::Path,
    table_id: TableId,
    manifest_restriction: &ManifestRestriction,
) -> crate::Result<Option<crate::UserKey>> {
    match manifest_restriction {
        ManifestRestriction::Restricted(bound) => Ok(Some(bound.clone())),
        ManifestRestriction::Unrestricted => Ok(None),
        ManifestRestriction::Unknown => {
            match crate::restrict_bound::read(fs, table_path, config.encryption.as_deref()) {
                Ok(crate::restrict_bound::SidecarRead::Present(id, b)) if id == table_id => {
                    Ok(Some(b.into()))
                }
                Err(e) if is_environmental(&e) => Err(e),
                Ok(_) | Err(_) => Ok(None),
            }
        }
    }
}

/// The dictionary a blob file's own compression descriptor names, resolved
/// against the tree's set; `None` for a file that uses no dictionary.
///
/// A blob file records ONE descriptor, so this answers with one dictionary; the
/// set is what turns the recorded id back into bytes. Resolving here rather than
/// passing the configured dictionary is what lets a file written under an
/// earlier dictionary still be read and salvaged.
///
/// # Errors
///
/// [`crate::Error::ZstdDictMismatch`] naming the id when the tree does not hold
/// it. The file is then intact bytes behind a missing context: decoding it
/// without the dictionary would grade it as damage and drop every value in it,
/// so the caller either stops the repair, which the operator answers by
/// supplying the dictionary, or leaves a file nothing references to the sweep.
#[cfg(all(feature = "std", zstd_any))]
fn blob_file_dictionary(
    config: &Config,
    compression: crate::CompressionType,
) -> crate::Result<Option<Arc<crate::compression::ZstdDictionary>>> {
    config
        .current_zstd_dictionaries()
        .for_compression(compression)
}

/// What only the manifest knows of a table's place in the tree: its bulk-ingest
/// sequence offset and its L0 recency. Both come from the ingestion's install,
/// after its file was written, so neither can be read back from the file.
#[cfg(feature = "std")]
#[derive(Clone, Copy, Debug)]
struct ManifestPosition {
    global_seqno: SeqNo,
    recency: TableId,
}

#[cfg(feature = "std")]
impl ManifestPosition {
    fn of_record(record: &crate::version::recovery::RecoveredTable) -> Self {
        Self {
            global_seqno: record.global_seqno,
            recency: record.recency,
        }
    }

    fn of_table(table: &Table) -> Self {
        Self {
            global_seqno: table.global_seqno(),
            recency: table.l0_recency(),
        }
    }
}

/// Recover params for a repair's TRANSIENT table open: the tree's configured
/// comparator / crypto / dictionary context (so the table decodes consistently
/// with how it was written), and everything else neutral — tree id 0 and no
/// descriptor table keep the open from polluting shared caches keyed by the
/// real tree id.
///
/// `position` is EXPLICIT, never defaulted: a bulk-ingested SST keeps its
/// entries at local seqno 0 and relies on the manifest-only offset for its
/// effective MVCC ordering, so silently opening at 0 would mis-order and
/// over-expose them. `None` means "no position is recoverable here", which is
/// correct only for a manifest-loss rebuild; a zero offset is genuine (the
/// first ingestion on a fresh counter commits it), not a stand-in for absence.
#[cfg(feature = "std")]
fn repair_recover_params(
    config: &Config,
    file_path: PathBuf,
    checksum: crate::Checksum,
    table_id: TableId,
    fs: Arc<dyn crate::fs::Fs>,
    position: Option<ManifestPosition>,
) -> crate::table::RecoverParams {
    let mut params = crate::table::RecoverParams::new(
        file_path,
        checksum,
        table_id,
        fs,
        config.comparator.clone(),
        config.cache.clone(),
    );
    if let Some(position) = position {
        params.global_seqno = position.global_seqno;
        params.recency = Some(position.recency);
    }
    params.encryption.clone_from(&config.encryption);
    #[cfg(zstd_any)]
    {
        params.zstd_dictionaries = config.current_zstd_dictionaries();
    }
    params
}

/// [`repair_recover_params`] for a table the repair holds until it publishes
/// the rebuilt manifest: its file goes through the configured descriptor
/// cache, within its budget, so the tables held at once need no descriptor
/// each. A tree id of its own: another copy of this table id, or its salvage
/// replacement, must never be handed this file's descriptor.
#[cfg(feature = "std")]
fn held_recover_params(
    config: &Config,
    file_path: PathBuf,
    checksum: crate::Checksum,
    table_id: TableId,
    fs: Arc<dyn crate::fs::Fs>,
    position: Option<ManifestPosition>,
) -> crate::table::RecoverParams {
    let mut params = repair_recover_params(config, file_path, checksum, table_id, fs, position);
    if let Some((cache, tree_id)) = held_descriptors(config) {
        params.descriptor_table = Some(cache);
        params.tree_id = tree_id;
    }
    params
}

/// The descriptors a repair cached, from its first tree id on, swept from the
/// cache when the repair ends.
#[cfg(feature = "std")]
struct HeldDescriptors {
    cache: Option<Arc<crate::DescriptorTable>>,
    first: crate::tree::inner::TreeId,
}

#[cfg(feature = "std")]
impl Drop for HeldDescriptors {
    fn drop(&mut self) {
        if let Some(cache) = &self.cache {
            cache.remove_trees_from(self.first);
        }
    }
}

/// Lets go of a table the repair held before its file is removed, its cached
/// descriptor with it: a backend that refuses to unlink an open file
/// (Windows) would otherwise fail the removal while the cache keeps it open.
#[cfg(feature = "std")]
fn release_held(config: &Config, table: Table) {
    if let Some(cache) = &config.descriptor_table {
        cache.remove_for_table(&table.global_id());
    }
    drop(table);
}

/// The descriptor cache a table the repair holds keeps its file in, with a
/// tree id of its own for it; `None` when the configuration keeps none.
#[cfg(feature = "std")]
fn held_descriptors(
    config: &Config,
) -> Option<(Arc<crate::DescriptorTable>, crate::tree::inner::TreeId)> {
    config
        .descriptor_table
        .clone()
        .map(|cache| (cache, crate::tree::inner::get_next_tree_id()))
}

/// Whether an I/O failure must PROPAGATE out of the repair instead of grading
/// the file it came from — either UNAMBIGUOUSLY TRANSIENT, or an
/// ENVIRONMENTAL access failure that does not implicate the bytes on disk.
///
/// The allowlist is deliberately narrow: `Interrupted` (`EINTR`) and
/// `WouldBlock` (`EAGAIN`) — the interrupted-syscall errors a retry genuinely
/// clears, and which a corrupt on-disk structure can NEVER produce — plus
/// `PermissionDenied` (`EACCES` / `EPERM`): an ACL / ownership mistake the
/// operator fixes, while grading the healthy file unreadable commits a
/// manifest that excludes it and then REMOVES it, turning a recoverable
/// configuration error into permanent data loss.
///
/// `Other` is NOT on the list, even though an injected fault or a raw `EIO`
/// lands there, because a STRUCTURAL corruption lands there too: a corrupt
/// trailer that decodes to a bad offset makes the reader seek before the start of
/// the file, which Windows reports as `ERROR_NEGATIVE_SEEK` — an unmapped OS
/// error the `From<std::io::Error>` bridge folds into `ErrorKind::Other`. Treating
/// `Other` as propagating would then abort the WHOLE repair (blocking recovery of
/// every healthy sibling table) on a single genuinely-corrupt SST, and the class
/// is platform-dependent (the same corruption reads back `InvalidInput` on Unix).
/// A hardware `EIO` is likewise usually a persistent bad-sector failure, so
/// recording that table unreadable — while the rest recover — is the right
/// outcome. Fault-injection tests therefore inject `Interrupted` to model a
/// retryable fault.
///
/// This inspects the CRATE's [`crate::io::ErrorKind`], which is what a
/// `crate::Error::Io` always carries.
///
/// The MIS-SUPPLIED RECOVERY CONTEXT errors are in the class too, for the
/// same reason: they say the caller brought the wrong key or the wrong
/// dictionary, not that the bytes rotted.
///
/// - [`crate::Error::Decrypt`]: an AEAD failure is exactly what a missing or
///   wrong key produces on perfectly healthy ciphertext, and the two are
///   cryptographically indistinguishable from genuine rot.
/// - [`crate::Error::ZstdDictMismatch`]: the persisted descriptor names a
///   dictionary the caller did not supply (`got: None`) or supplied a
///   different one. The blob is intact; only the context is wrong.
///
/// Recording either as unreadable commits a manifest omitting the file —
/// whose cleanup then DELETES it — turning a fixable configuration mistake
/// into permanent loss. Propagating lets a re-run with the right context
/// recover everything; on genuinely damaged bytes the repair fails instead of
/// guessing, and they stay in place.
fn is_environmental(e: &crate::Error) -> bool {
    // The class itself lives on `Error` so the salvage paths classify
    // identically — the two drifted apart once already, and the difference is
    // invisible until a file is deleted over it.
    e.is_environmental()
}

/// Whether manifest repair must fail closed on a table because its bulk-ingest
/// sequence offset cannot be reconstructed from the SST alone (the rebuilt
/// manifest would install it with offset 0 and silently mis-order / mis-expose
/// its entries).
///
/// - `Some(true)`: authoritatively bulk-ingested — always fail closed.
/// - `Some(false)`: a newer non-ingested table — safe (offset genuinely 0), even
///   when its entries all sit at seqno 0 (a fresh tree's first batch).
/// - `None`: a LEGACY SST written before the provenance flag existed — UNKNOWN.
///   Treat it as bulk-ingested ONLY when its entries carry the ingest signature
///   (present, and every LOCAL seqno 0), which a legacy bulk-ingest produces. A
///   legacy first-batch-at-seqno-0 flush matches too and is conservatively
///   dropped; the ambiguity is unavoidable without the flag.
fn has_unrecoverable_ingest_offset(
    bulk_ingested: Option<bool>,
    item_count: u64,
    max_local_seqno: crate::SeqNo,
) -> bool {
    match bulk_ingested {
        Some(flagged) => flagged,
        None => item_count > 0 && max_local_seqno == 0,
    }
}

/// How faithfully a recovered candidate carries its source's content. Drives
/// two DISTINCT report facts: anything but [`Complete`](Self::Complete)
/// contributes a `lost_coverage` entry (scoped by the source's coverage), while
/// only [`Salvaged`](Self::Salvaged) counts toward `RepairReport::salvaged` —
/// a geometry-restricted original is lossy but was never block-salvaged, and
/// it arises in plain (salvage-off) repairs too.
#[derive(Copy, Clone, PartialEq, Eq)]
enum Fidelity {
    /// Every live row of the source is present.
    Complete,
    /// The ORIGINAL file reopened under a punch-GEOMETRY-derived restriction
    /// (no exact sidecar, resurrection off): the bound is the straddling
    /// block's END key, so that block's still-live suffix rows may be excluded.
    GeometryRestricted,
    /// A block-salvage rewrite: corrupt blocks (or blob records) were dropped.
    Salvaged,
}

impl Fidelity {
    fn is_complete(self) -> bool {
        matches!(self, Self::Complete)
    }
}

/// A recovered table plus its [`Fidelity`]. Repair keeps the best copy per
/// table id, so a duplicate id in another table folder can supersede an
/// earlier DAMAGED copy.
///
/// The physical location (`fs` / `path`) travels with the candidate so that when
/// a duplicate SUPERSEDES it, the loser's file can be removed from `tables/`. The
/// rebuilt manifest records only `id + checksum`, so two same-id files left in
/// different folders would let recovery resolve the stale one by folder order and
/// reopen it against the kept copy's mismatched checksum.
struct TableCandidate {
    table: Table,
    fidelity: Fidelity,
    fs: Arc<dyn crate::fs::Fs>,
    path: PathBuf,
    /// Whether this file's whole-file digest equals the one the committed
    /// manifest records for its id. `false` also covers "cannot be compared":
    /// no clean manifest, or a RESTRICTED table, whose entry holds the
    /// live-suffix digest that this whole-file hash never reproduces.
    matches_manifest: bool,
}

/// Records `candidate` for `id`, keeping the BETTER of the existing and the new
/// copy: a COMPLETE recovery replaces a lossy salvage, so an intact duplicate in
/// a later-scanned folder supersedes an earlier lossy one.
///
/// Two COMPLETE copies are not interchangeable either. Both open, both are
/// self-consistent, and they can still hold DIFFERENT generations — a prior
/// repair that retained a noncanonical copy while its cleanup left the older
/// canonical file behind produces exactly that. Adopting the first-seen would
/// re-stamp a fresh checksum over the stale bytes, and every surviving handle
/// would then resolve against the wrong generation. The copy whose digest
/// matches the committed manifest wins; with none to compare against (no clean
/// manifest, or a restricted table whose entry digests only its live suffix)
/// they really are equivalent and the first-seen stays.
///
/// Two LOSSY copies are NOT equivalent. Damaged copies of one table in two
/// routed folders salvage independently, and one can recover far more of it
/// than the other; keeping the first-seen would discard rows the other salvage
/// did recover, which is avoidable loss. The fuller one wins, measured by the
/// entries its metadata records — the salvage writes exactly what it recovered,
/// so that count IS its completeness. Equal counts keep the first-seen, so the
/// choice stays deterministic across runs over the same directory.
///
/// Returns the DISPLACED loser (the rejected new candidate, or the superseded old
/// one) so the caller can record its file for removal; `None` when `id` was
/// previously unseen (nothing displaced).
#[must_use = "the displaced duplicate's file must be removed from tables/"]
fn keep_best_candidate(
    map: &mut crate::HashMap<TableId, TableCandidate>,
    id: TableId,
    candidate: TableCandidate,
) -> Option<TableCandidate> {
    let keeps_existing = match map.get(&id) {
        // Two completes: only the manifest can say which generation this id
        // is, so a copy that reproduces its digest displaces one that does
        // not. Neither matching (or nothing to match against) keeps the
        // incumbent, which is the previous first-seen rule.
        Some(existing) if existing.fidelity.is_complete() && candidate.fidelity.is_complete() => {
            existing.matches_manifest || !candidate.matches_manifest
        }
        // An existing COMPLETE copy is never displaced by a lossy one.
        Some(existing) if existing.fidelity.is_complete() => true,
        // A COMPLETE newcomer displaces a lossy incumbent.
        Some(_) if candidate.fidelity.is_complete() => false,
        // Both lossy: the one that recovered more of the table wins.
        Some(existing) => candidate.table.metadata.item_count <= existing.table.metadata.item_count,
        None => false,
    };
    if keeps_existing {
        return Some(candidate);
    }
    // The new candidate supersedes: `insert` returns the displaced old copy.
    map.insert(id, candidate)
}

/// Removes a blob-salvage temp on a path where the repair goes on to COMMIT a
/// manifest. A missing temp is fine; ANY other failure fails the repair: the
/// rebuilt manifest never references the temp, so left in `blobs/` the next
/// open's orphan sweep would hit the same removal failure — reporting success
/// for a tree that cannot open would be a lie. Quarantine is NOT an out here:
/// it preserves damaged DATA for the operator, and a temp is discardable
/// garbage — while a directory that refuses removal refuses the rename too.
/// Paths where the repair itself returns an error need only best-effort
/// removal: the retry re-attempts it and fails honestly if it still cannot.
#[cfg(feature = "std")]
fn remove_temp(config: &Config, temp: &std::path::Path) -> crate::Result<()> {
    match config.fs.remove_file(temp) {
        Ok(()) => Ok(()),
        Err(e) if e.kind() == crate::io::ErrorKind::NotFound => Ok(()),
        Err(e) => {
            log::error!(
                "repair: cannot remove salvage temp {} ({e}); failing the repair — \
                 left in place, the next open's orphan sweep would hit the same error",
                temp.display(),
            );
            Err(e.into())
        }
    }
}

/// Whether `a` and `b` name the SAME physical file — a symlink, junction, or
/// case-insensitive alias resolving to one directory entry (two configured table
/// folders pointing at the same location). Used so a repeated sighting of one SST
/// through an alias is never removed as a "duplicate" (which would destroy the
/// kept copy and orphan the manifest entry).
///
/// The two candidates must live in the SAME filesystem namespace for a path
/// comparison to mean anything: a virtual (`MemFs`) table can sit at a path that
/// also exists on the host filesystem, and canonicalizing both spellings through
/// the host would call those distinct files aliases — the loser would then survive
/// and a later reopen could resolve that leftover against the kept copy's
/// manifest checksum. Backends advertise namespace identity through
/// [`Fs::backend_id`](crate::fs::Fs::backend_id), whose `None` means "no shared
/// namespace guarantee" and is therefore treated as DISTINCT.
///
/// Within one namespace, alias resolution is the backend's own
/// ([`Fs::same_file`](crate::fs::Fs::same_file)): kernel-backed backends
/// canonicalize through the host, virtual ones compare paths literally — a
/// host symlink must never alias two distinct virtual files. A probe that
/// cannot decide answers `false` (distinct), so a genuine duplicate is still
/// removed.
#[cfg(feature = "std")]
fn same_physical_file(
    fs_a: &dyn crate::fs::Fs,
    a: &std::path::Path,
    fs_b: &dyn crate::fs::Fs,
    b: &std::path::Path,
) -> crate::Result<bool> {
    match (fs_a.backend_id(), fs_b.backend_id()) {
        (Some(id_a), Some(id_b)) if id_a == id_b => {}
        // Two backends that each CLAIM an identity and disagree are provably
        // distinct namespaces: path spellings are not comparable.
        (Some(_), Some(_)) => return Ok(false),
        // A backend WITHOUT an identity claim (a custom `Fs` keeping the
        // default `None`): the SAME instance is trivially one namespace, so
        // its own probe below decides. Two DIFFERENT instances cannot be
        // proven distinct — and `false` here AUTHORIZES deleting what may be
        // an alias of the kept file — so the verdict is inconclusive and the
        // repair aborts for the operator to sort out.
        _ => {
            if !core::ptr::addr_eq(
                core::ptr::from_ref::<dyn crate::fs::Fs>(fs_a),
                core::ptr::from_ref::<dyn crate::fs::Fs>(fs_b),
            ) {
                return Err(crate::Error::Io(crate::io::Error::new(
                    crate::io::ErrorKind::InvalidInput,
                    "alias identity is inconclusive: the backend reports no namespace id",
                )));
            }
        }
    }
    // Alias resolution belongs to the BACKEND, not the host: canonicalizing a
    // virtual backend's paths through the host filesystem would let a host
    // symlink alias two unrelated virtual files, and the "duplicate" loser
    // would survive. Kernel-backed backends canonicalize; virtual ones compare
    // literally. An INCONCLUSIVE probe propagates: `false` authorizes
    // duplicate deletion, and deleting what may be an alias of the kept copy
    // unlinks the very directory entry the rebuilt manifest retains.
    Ok(fs_a.same_file(a, b)?)
}

/// Records a duplicate table file that lost to a better same-id copy: reported
/// as considered-but-not-referenced, and removed once the manifest is durable so
/// recovery can never resolve it instead of the kept copy.
#[cfg(feature = "std")]
fn discard_duplicate(
    loser: TableCandidate,
    unreadable_files: &mut Vec<(PathBuf, String)>,
    redundant_unreadable: &mut crate::HashSet<PathBuf>,
    discard_after_commit: &mut Vec<(Arc<dyn crate::fs::Fs>, PathBuf, String)>,
) {
    let TableCandidate {
        table, fs, path, ..
    } = loser;
    drop(table); // release the open file handle
    let reason = "duplicate table id superseded by another copy";
    discard_after_commit.push((fs, path.clone(), reason.to_string()));
    // Reported, but NOT a loss: the copy that displaced it holds the same id,
    // so nothing needs replaying. Counting it would answer `FullHistory` and
    // send an external WAL over a tree that lost nothing.
    redundant_unreadable.insert(path.clone());
    unreadable_files.push((path, reason.to_string()));
}

/// Builds a [`TableCandidate`] from a recovered table plus its physical location,
/// records it as the best copy for `id`, and marks any duplicate displaced by
/// the decision for post-commit removal. The one path every recovered table
/// takes to enter the manifest, so a superseded same-id file is never left
/// discoverable once the rebuild is durable.
/// Records a finished SALVAGE replacement and queues the post-commit swap
/// that publishes it.
///
/// Both salvage arms — the one whose VERIFICATION failed and the one whose
/// whole-file RECOVERY failed — reach exactly this state, and they must not
/// drift apart: the replacement enters the manifest through `record_best` like
/// any recovered table, and the swap carries whether it is RESTRICTED so a
/// retried swap can tell an already-moved replacement sidecar from a stale
/// source one (see `commit_repair_tmp`).
#[cfg(feature = "std")]
#[expect(
    clippy::too_many_arguments,
    reason = "location + report threaded through"
)]
fn keep_salvaged_replacement(
    map: &mut crate::HashMap<TableId, TableCandidate>,
    unreadable_files: &mut Vec<(PathBuf, String)>,
    redundant_unreadable: &mut crate::HashSet<PathBuf>,
    discard_after_commit: &mut Vec<(Arc<dyn crate::fs::Fs>, PathBuf, String)>,
    swap_after_commit: &mut Vec<(Arc<dyn crate::fs::Fs>, PathBuf, PathBuf, bool)>,
    id: TableId,
    table: Table,
    fs: &Arc<dyn crate::fs::Fs>,
    table_path: &std::path::Path,
    output_path: PathBuf,
) -> crate::Result<()> {
    let restricted = table.restrict_lower_bound().is_some();
    // Queue the swap FIRST, then let `record_best` decide: if this candidate
    // loses (rejected outright, or displaced later by an intact duplicate from
    // another routed folder), that same call drops the swap again, so only a
    // replacement the rebuilt manifest actually references is ever renamed.
    swap_after_commit.push((
        Arc::clone(fs),
        output_path,
        table_path.to_path_buf(),
        restricted,
    ));
    record_best(
        map,
        unreadable_files,
        redundant_unreadable,
        discard_after_commit,
        swap_after_commit,
        id,
        table,
        Fidelity::Salvaged,
        fs,
        table_path,
        // A salvage's copy is a fresh file the manifest never digested.
        false,
    )
}

#[cfg(feature = "std")]
#[expect(
    clippy::too_many_arguments,
    reason = "location + report threaded through"
)]
fn record_best(
    map: &mut crate::HashMap<TableId, TableCandidate>,
    unreadable_files: &mut Vec<(PathBuf, String)>,
    redundant_unreadable: &mut crate::HashSet<PathBuf>,
    discard_after_commit: &mut Vec<(Arc<dyn crate::fs::Fs>, PathBuf, String)>,
    // Queued swaps, so a candidate that LOSES here takes its own swap with it:
    // a replacement the rebuilt manifest does not reference must not be renamed
    // onto a name the manifest gives to another file, and a rename fault on a
    // disposable temp would otherwise turn a good repair into
    // `RepairedButUnopened`. A swap is keyed by the destination it publishes,
    // which is exactly the losing candidate's own path.
    swap_after_commit: &mut Vec<(Arc<dyn crate::fs::Fs>, PathBuf, PathBuf, bool)>,
    id: TableId,
    table: Table,
    fidelity: Fidelity,
    fs: &Arc<dyn crate::fs::Fs>,
    path: &std::path::Path,
    matches_manifest: bool,
) -> crate::Result<()> {
    let candidate = TableCandidate {
        table,
        fidelity,
        fs: Arc::clone(fs),
        path: path.to_path_buf(),
        matches_manifest,
    };
    if let Some(loser) = keep_best_candidate(map, id, candidate) {
        // If the displaced copy physically ALIASES the kept one (same directory
        // entry via a symlink / junction / case-insensitive path), it is the SAME
        // file — removing it would destroy the kept copy and orphan the manifest
        // entry. Drop the loser's handle in place instead. An INCONCLUSIVE
        // probe aborts the repair for a retry: guessing "distinct" would
        // authorize exactly that destruction.
        let is_alias = match map.get(&id) {
            Some(kept) => same_physical_file(&*loser.fs, &loser.path, &*kept.fs, &kept.path)?,
            None => false,
        };
        // The loser's replacement (if it built one) is never published — and
        // dropping the swap drops the ONLY reference to that `{id}.repair-tmp`,
        // so it is re-queued for removal instead of silently left behind.
        // Left in place it is exactly the shape the next open cannot resolve:
        // a temp whose digest does not match the manifest, with no original
        // beside it in THAT folder to prove it abandoned — so the open fails
        // before it ever reaches the healthy copy this repair kept.
        let mut orphaned_temps = Vec::new();
        swap_after_commit.retain(|(fs, temp, dest, _)| {
            let publishes_loser = dest == &loser.path;
            if publishes_loser {
                orphaned_temps.push((Arc::clone(fs), temp.clone()));
            }
            !publishes_loser
        });
        for (fs, temp) in orphaned_temps {
            discard_after_commit.push((
                fs,
                temp,
                "unpublished replacement of a displaced duplicate".to_string(),
            ));
        }
        if is_alias {
            return Ok(());
        }
        discard_duplicate(
            loser,
            unreadable_files,
            redundant_unreadable,
            discard_after_commit,
        );
    }
    Ok(())
}

/// Block-salvages a corrupt SST during repair: reads the UNTOUCHED original in
/// place, writes a fresh SST holding its recoverable blocks to `table_path` (the
/// unpublished `{id}.repair-tmp`), and reopens it.
///
/// Returns `Ok(None)` when nothing was recoverable (the source is untouched and
/// no replacement exists), or `Err` when even salvage cannot open the source
/// (its metadata / index / SFA trailer is itself unreadable).
/// Whether a freshly-recovered SST passes the salvage-mode block verify.
///
/// One uniform path for encrypted and unencrypted tables: the out-of-band
/// section walk. Block headers and payload checksums are PLAINTEXT, so the
/// walk needs the provider only to decode the meta block (the per-SST ECC
/// descriptor); every section — data, index/TLI, filter, zone map, delete
/// bitmap, locator, meta — is then verified against its raw on-disk checksum,
/// which flags even a persistent ECC-CORRECTABLE fault (a live read would
/// silently heal it in memory while the corrupt bytes stay on disk).
/// Classifies a block-verifier result for the salvage gate. A structural
/// divergence (a checksum / decode / cross-check mismatch) is genuine
/// corruption: `Ok(true)`, route the table through salvage. Only a TRANSIENT
/// [`crate::Error::Io`] (the [`is_environmental`] allowlist) aborts the repair
/// (`Err`) for a retry, rather than dropping a healthy block into a partial
/// replacement. A PERSISTENT I/O failure is NOT retryable — a bad sector, or a
/// structural corruption surfacing as `Io(Other)` on some platforms — so it is
/// graded as corruption and salvaged too, rather than aborting the whole repair
/// and stranding every other healthy table on one unrecoverable read.
fn is_corruption(res: crate::Result<()>) -> crate::Result<bool> {
    match res {
        Ok(()) => Ok(false),
        Err(e) if is_environmental(&e) => Err(e),
        Err(_) => Ok(true),
    }
}

fn block_verify_verdict(
    config: &Config,
    folder_fs: &Arc<dyn crate::fs::Fs>,
    table_path: &std::path::Path,
    table: &Table,
) -> crate::Result<BlockVerifyVerdict> {
    // Walk only the recovered view's LIVE data: for a tight-space RESTRICTED view
    // (a valid `.restrict-bound` sidecar was accepted) the `[0, punch_offset)`
    // prefix is hole-punched and reads as zeros, so starting at byte 0 would
    // report those dead blocks as corruption and repair would then drop an
    // otherwise-healthy restricted SST. `punch_offset()` is `0` for a normal table.
    let data_start = table.punch_offset()?;
    let report = crate::verify::verify_sst_file_with_context(
        folder_fs,
        table_path,
        config.encryption.as_ref(),
        // Repair KNOWS the durable id (recovery already cross-checked it
        // against the file name), so the verify probe enforces the same meta
        // id check — a checksum-clean forged tail meta falls back to the
        // intact MID mirror instead of dictating a forged ECC descriptor.
        Some(table.metadata.id),
        data_start,
    );
    // A TRANSIENT read failure DURING the walk (a retryable `Interrupted` /
    // `WouldBlock`) is not block corruption — routing it through salvage would
    // re-read the same bytes and drop a healthy block — and neither is an
    // ENVIRONMENTAL `PermissionDenied` (an ACL / ownership mistake the operator
    // fixes; the bytes on disk are not implicated). Propagate both so the repair
    // aborts and the operator retries, mirroring the decode-load gate below. Any
    // OTHER kind falls through to the corruption verdict: a truncation
    // (`UnexpectedEof`) is genuine on-disk damage, and a PERSISTENT failure
    // (`Other` / EIO) is not fixed by a retry, so aborting forever would strand
    // every healthy sibling table on one bad SST.
    // This matches the `is_corruption` allowlist policy exactly.
    //
    // This gate depends on the walk CLASSIFYING transient faults as one of these
    // two I/O-bearing variants: a mid-walk seek failure, a transient block-header
    // read, and a raw-section read all surface as `DataReadError` rather than
    // being folded into `HeaderCorrupted` / `TocCorrupted`, so a flaky read here
    // is not mistaken for corruption and salvaged.
    for e in &report.errors {
        if let crate::verify::BlockVerifyError::SstFileUnreadable { error, .. }
        | crate::verify::BlockVerifyError::DataReadError { error, .. } = e
            && error.kind().is_environmental()
        {
            // Preserve the ErrorKind: re-wrapping as `Other` would make the
            // caller's `is_environmental` check see a non-propagating kind
            // and re-grade this retryable / environmental failure as
            // corruption, defeating the propagation intent of this very gate.
            return Err(crate::Error::Io(crate::io::Error::new(
                error.kind(),
                error.to_string(),
            )));
        }
    }

    // A non-parity error is corruption regardless of any warnings.
    let verdict = if !report
        .errors
        .iter()
        .all(|e| matches!(e, crate::verify::BlockVerifyError::EccParityMismatch { .. }))
    {
        BlockVerifyVerdict::Corrupt
    } else if report
        .warnings
        .iter()
        .any(|w| matches!(w, crate::verify::BlockVerifyWarning::UnrecognizedEcc { .. }))
    {
        // Unrecognized ECC descriptor: the walk SKIPPED the SST-block
        // sections entirely (their trailer length is underivable), so
        // NOTHING about the data was verified — a stronger degradation
        // than a checked-but-unverifiable-parity report. Graded BEFORE the
        // parity-only arm below: parity mismatches in the still-walked
        // self-describing meta blocks must not mask the skipped data /
        // index sections.
        BlockVerifyVerdict::DegradedUnscanned
    } else if is_corruption(table.verify_blob_links())? {
        // Same reasoning for the blob-link list: the section carries no
        // per-section checksum, so the walk can only validate its SHAPE — a
        // flipped blob id passes it. Cross-check against the table's own
        // indirection entries (a no-op without the section); a mismatch is
        // corruption, and salvage derives the links from the recovered
        // indirections rather than copying the forged list.
        BlockVerifyVerdict::Corrupt
    } else if is_corruption(table.verify_tli_mirrors())? {
        // Each TLI mirror is independently checksum-clean to the walk, but a
        // forged copy that DECODES to a different handle list would steer
        // the next recovery (which prefers the tail) away from real blocks.
        // Diverging decoded mirrors are corruption; salvage walks the HEAD
        // copy, so the recovered SST is rebuilt from a single, fully
        // re-verified handle list. BOTH mirrors forged to the SAME list that
        // OMITS a physical block are covered too: the salvage walk
        // cross-checks the index against the physical data-section tiling
        // and frames the uncovered bytes from their block headers, so the
        // hidden block is recovered (or reported dropped), never silently
        // missing from an apparently complete copy.
        BlockVerifyVerdict::Corrupt
    } else if is_corruption(table.verify_block_layout())? {
        // A checksum-clean block_layout re-stamped to another structurally
        // valid boundary set mis-maps the partial range-read path's
        // decompression bounds, silently omitting keys. Boundaries that
        // disagree with the frames' actual inner blocks are corruption;
        // salvage re-derives the layout when re-encoding.
        BlockVerifyVerdict::Corrupt
    } else if is_corruption(
        table
            .verify_reconcile_gates(config.prefix_extractor.as_ref(), false)
            .map_err(|(_, e)| e),
    )? {
        // The semantic cross-checks, all on ONE decode of each live block.
        // Every one of them catches a section that is checksum-clean to the
        // out-of-band walk yet lies about the entries, and every one routes
        // the table to salvage, which rebuilds the section from the re-emitted
        // data:
        //
        // - per-KV footers: a stale digest behind a re-stamped block checksum.
        //   Graded BEFORE the degradation arms below — a forged footer also
        //   leaves the parity trailer mismatched, and grading that as
        //   "parity-only degradation" would retain a table with a KNOWN-stale
        //   entry digest.
        // - seqno bounds: re-stamped to another structurally valid map, which
        //   `scan_since_seqno` trusts to SKIP blocks.
        // - entry counts: a valid prefix followed by a malformed tail decodes
        //   short while the trailer still declares the full count.
        // - zone map: forged min/max let a predicate scan skip blocks holding
        //   matching rows.
        // - locator: re-stamped to resolve a key to a block other than its
        //   newest-version one, so point_read returns a stale value without
        //   falling back to the sorted index.
        // - filter: an existing key turned into a false negative disappears
        //   from every read.
        // - point-read reachability: a hidden hash bucket or a misdirected
        //   offset makes point_read miss data the block still decodes.
        // - metadata bounds: a narrowed key range hides real keys (and the
        //   range tombstones masking older tables) from run selection.
        BlockVerifyVerdict::Corrupt
    } else if !report.is_ok() {
        // Parity-ONLY rot: every payload checksum verified clean, only the
        // recovery margin is dead. The data is fully readable, so it grades
        // like a warning-bearing report — salvage preferred (the rewrite
        // regenerates fresh parity), but never at the cost of dropping data
        // salvage cannot re-emit.
        BlockVerifyVerdict::DegradedButReadable
    } else if report.has_warnings() {
        // Everything scanned verified clean, but the parity trailers could
        // not be recomputed (a parity-less build). The caller decides
        // between salvage (a rewrite under fully-verifiable framing) and
        // keeping the table when salvage cannot re-emit it.
        BlockVerifyVerdict::DegradedButReadable
    } else {
        BlockVerifyVerdict::Clean
    };
    Ok(verdict)
}

/// Outcome of the salvage-mode block verify, from the repair gate's point of
/// view (see [`block_verify_verdict`]).
enum BlockVerifyVerdict {
    /// Every section verified against its raw on-disk checksum.
    Clean,
    /// Every payload the walk checked verified clean, but the table is
    /// DEGRADED: its parity trailers rotted or could not be recomputed while
    /// the payloads stayed intact. Prefer a salvage rewrite, but never at
    /// the cost of dropping data salvage cannot re-emit.
    DegradedButReadable,
    /// The walk could not scan the SST-block sections at all (an
    /// unrecognized ECC descriptor): the data is UNVERIFIED, not merely
    /// degraded — any keep decision must first verify it another way.
    DegradedUnscanned,
    /// At least one payload / section failed verification.
    Corrupt,
}

/// What the repair should do with a freshly-recovered table, based on the
/// salvage-mode block verify.
#[derive(Debug)]
enum RepairKeepDecision {
    /// The table joins the rebuilt manifest as-is.
    Keep,
    /// The table is routed through block salvage: its readable blocks are
    /// rewritten into a fresh copy beside it.
    Salvage,
    /// The table can be neither trusted nor faithfully salvaged under the
    /// active resurrection policy: it is EXCLUDED from the rebuilt manifest,
    /// with this reason, and its file removed once that manifest is durable.
    /// The tree still opens. Its rows come back through recompaction, a replica,
    /// a checkpoint plus journal replay, or a backup — and, where the reason
    /// says so, from a repair run WITH resurrection enabled instead.
    Drop(&'static str),
}

/// Whether the on-disk TOC catalogue could HIDE a deletion section — see
/// [`crate::verify::toc_may_hide_deletion_section`]. A STRUCTURAL catalogue
/// ambiguity grades `Ok(true)` (fail closed): if the catalogue cannot be parsed
/// to prove no section is hidden, salvage must not trust the parsed absence of
/// deletion metadata.
///
/// # Errors
///
/// Propagates a TRANSIENT [`crate::Error::Io`] from opening or reading the
/// trailer. Grading a retryable read as `true` would send a table
/// [`repair_with_salvage`](Self) already found corrupt to `Quarantine` — dropping
/// its healthy ranges from the rebuilt manifest — when a retry of the probe could
/// have let block salvage recover them.
pub(crate) fn toc_may_hide_deletions(
    folder_fs: &Arc<dyn crate::fs::Fs>,
    table_path: &std::path::Path,
) -> crate::Result<bool> {
    let mut file = match folder_fs.open(table_path, &crate::fs::FsOpenOptions::new().read(true)) {
        Ok(file) => file,
        // A TRANSIENT open failure propagates (a retry could open the file and
        // prove no hidden section); a PERSISTENT one fails closed as catalogue
        // ambiguity — we cannot read the TOC to prove it hides no deletion
        // section, so `true` drops the table rather than resurrecting masked rows.
        Err(e) => {
            let err = crate::Error::Io(e);
            return if is_environmental(&err) {
                Err(err)
            } else {
                Ok(true)
            };
        }
    };
    match crate::sfa::Reader::from_reader(&mut file) {
        Ok(reader) => Ok(crate::verify::toc_may_hide_deletion_section(
            reader.toc(),
            reader.toc_pos(),
        )),
        // A transient trailer read propagates (retry could prove no hidden
        // section); a persistent I/O failure or a structural trailer failure is
        // genuine catalogue ambiguity that fails closed.
        Err(crate::sfa::Error::Io(e)) => {
            let err = crate::Error::Io(e);
            if is_environmental(&err) {
                Err(err)
            } else {
                Ok(true)
            }
        }
        Err(_) => Ok(true),
    }
}

/// Grades a freshly-recovered table into a [`RepairKeepDecision`].
///
/// `Corrupt` always salvages. `DegradedButReadable` (payloads verified clean,
/// only the parity trailers rotted or could not be recomputed) salvages ONLY
/// when salvage can faithfully re-emit the table: a range-tombstone SST is
/// rejected by the block walk, so routing it through salvage would drop
/// healthy, verified data over dead parity — it is kept as-is (with an
/// operator-facing warning) instead. `DegradedUnscanned` (unrecognized ECC
/// descriptor: the walk verified NOTHING about the data) never keeps: a
/// rewritable table salvages, and a range-tombstone table — which cannot be
/// verified in full (every lazy side structure would need its own
/// handle-based check) and cannot be re-emitted — is excluded (recompact under
/// a supported scheme to re-admit it) instead of riding unverified into the
/// rebuilt manifest.
///
/// `allow_resurrection` governs the one ambiguous case: a corrupt catalogue
/// that could conceal a deletion section. Off (default) excludes the table (its
/// visibility is unrecoverable, so admitting it would resurrect masked rows);
/// on, it salvages, accepting that suppressed rows reappear.
fn verify_keep_decision(
    config: &Config,
    folder_fs: &Arc<dyn crate::fs::Fs>,
    table_path: &std::path::Path,
    table: &Table,
    allow_resurrection: bool,
    // Whether this repair may REWRITE a damaged table (block salvage). Off,
    // the degraded verdicts resolve without a rewrite: corrupt / unverifiable
    // content is set aside (with the reason pointing at the salvage-enabled
    // repair), while rotted-parity-but-readable content is KEPT — its payloads
    // verified clean, and blessing its digest is the entry into the normal
    // attributable heal (a patrol re-stamps the parity and reconciles).
    salvage: bool,
) -> crate::Result<RepairKeepDecision> {
    Ok(
        match block_verify_verdict(config, folder_fs, table_path, table)? {
            BlockVerifyVerdict::Clean => RepairKeepDecision::Keep,
            BlockVerifyVerdict::Corrupt => {
                // A `Corrupt` verdict from a catalogue that could HIDE a deletion
                // section (an omitted / renamed / shadowed `range_tombstones` or
                // `delete_bitmap`) makes the table's visibility unrecoverable: the
                // positional salvage walk reopens the same forged TOC, sees no
                // deletion section in the parsed state, and re-emits the suppressed
                // rows as LIVE. The salvage-side resurrection guard only inspects
                // the PARSED deletion state, which the concealment defeats, so the
                // decision has to happen here, governed by the resurrection flag:
                // off, exclude the table (admitting it would resurrect masked
                // rows); on, salvage and accept the resurrection. A relabel that
                // keeps the tiling intact but re-roles the block is caught inside
                // salvage itself (`salvage_with_context` fails closed on a corrupt
                // rebuildable section when no deletion is visible), which both this
                // path and the recovery-failure salvage path funnel through.
                //
                // Probed BEFORE the salvage-off branch below on purpose: with
                // salvage off the table is dropped either way, but this reason
                // is the accurate one — pointing that operator at a
                // salvage-enabled repair would mislead (it drops the
                // concealment case too, unless resurrection is enabled).
                if toc_may_hide_deletions(folder_fs, table_path)? && !allow_resurrection {
                    RepairKeepDecision::Drop(
                        "TOC corruption may hide deletion metadata (range tombstones \
                     / delete bitmap); its visibility is unrecoverable, so the table \
                     is excluded to avoid resurrecting masked rows. Enable \
                     resurrection to salvage it, accepting that suppressed rows \
                     reappear",
                    )
                } else if salvage {
                    RepairKeepDecision::Salvage
                } else {
                    RepairKeepDecision::Drop(
                        "verification found corrupt data blocks; run a salvage-enabled \
                         repair to rewrite the readable blocks",
                    )
                }
            }
            BlockVerifyVerdict::DegradedButReadable => {
                if salvage && table.range_tombstones().is_empty() {
                    RepairKeepDecision::Salvage
                } else {
                    log::warn!(
                        "table {} at {}: every payload verified clean but its ECC is \
                     partially uncheckable or rotted, and this repair cannot rewrite it \
                     (salvage off, or range tombstones it cannot re-emit) — keeping the \
                     table as-is; a patrol heal or recompaction re-stamps it under \
                     fresh, verifiable parity",
                        table.metadata.id,
                        table_path.display(),
                    );
                    RepairKeepDecision::Keep
                }
            }
            BlockVerifyVerdict::DegradedUnscanned => {
                if salvage && table.range_tombstones().is_empty() {
                    RepairKeepDecision::Salvage
                } else if salvage {
                    RepairKeepDecision::Drop(
                        "ECC descriptor unrecognized (the block walk cannot verify the \
                     table) and salvage cannot re-emit its range tombstones; the table \
                     is excluded (recompact it under a supported scheme to re-admit it)",
                    )
                } else {
                    RepairKeepDecision::Drop(
                        "ECC descriptor unrecognized (the block walk cannot verify the \
                     table); run a salvage-enabled repair to rewrite it under fresh, \
                     verifiable parity",
                    )
                }
            }
        },
    )
}

/// Outcome of [`try_salvage_table`].
enum SalvageOutcome {
    /// A clean replacement was written and reopened, ready to install.
    Salvaged(Table),
    /// Nothing was recoverable, or the replacement was rejected (an
    /// unreconstructible bulk-ingest offset); the caller records the table
    /// unreadable and removes its file once the manifest is durable.
    Unusable,
    /// The `reject_punched_without_bound` guard fired: a salvage-dropped data
    /// extent of the source reads as zeros (the hole-punch signature), so the
    /// source lost data to a punch whose bound is unrecoverable. The
    /// unrestricted replacement was rejected and removed — installing it would
    /// resurrect the reclaimed region's superseded rows.
    PunchedBoundLost,
}

/// What one [`try_salvage_table`] call operates on: the source paths and the
/// per-call policy. Bundled so the salvage entry keeps a small signature as
/// its policy surface grows.
#[cfg(feature = "std")]
struct TableSalvage<'a> {
    /// The damaged original to read, UNTOUCHED and still at its own path. It is
    /// never displaced first: the copy goes to a fresh name beside it, and the
    /// source is removed only once the manifest names the copy. A salvage that
    /// fails therefore leaves the directory exactly as it was found, and the
    /// retry re-derives the same result from the same bytes.
    source: &'a std::path::Path,
    /// Where the recovered copy is written (a fresh, unused path).
    table_path: &'a std::path::Path,
    /// The durable table id (its file name).
    table_id: TableId,
    /// Fail closed when the salvage walk reveals the source was PUNCHED (a
    /// dropped data extent reads as zeros) — set by the recovery-failure arm
    /// when it has no recoverable restriction bound and resurrection is off.
    /// The pre-salvage first-bytes probe catches a punched FIRST block cheaply,
    /// but a partial punch (the punch-on-drop reclaim continues past an
    /// individual `punch_hole` failure) can leave the first block intact while
    /// later prefix blocks are zeroed; only the walk sees those. The
    /// verification arm passes `false`: it derives the bound from its restricted
    /// view, so its salvage output is re-restricted, never ambiguous.
    reject_punched_without_bound: bool,
    /// Per-blob handle rewrite for a table referencing a blob file this repair
    /// reshaped (salvaged into a compacted copy, or recovered with a punched
    /// frontier); `None` on the plain corrupt-table salvage paths.
    blob_rewrite:
        Option<Arc<crate::HashMap<crate::vlog::BlobFileId, crate::salvage::BlobFileRewrite>>>,
    /// Cell rows of the table to make the owners of objects whose owning row
    /// the rebuild lost; `None` everywhere but the ownership pass.
    owner_promotions: Option<Arc<crate::salvage::OwnerPromotions>>,
    /// The source's RECOVERED manifest position, when it is known: from a
    /// clean manifest record, or from a source the scan already admitted (the
    /// blob-handle rewrite). A salvaged copy preserves local seqnos and holds
    /// the source's content, so the position applies to it unchanged, and its
    /// presence also says the offset need not be reconstructed from the SST,
    /// so the fail-closed bulk-ingest rejection does not apply.
    ///
    /// A zero offset is genuine, NOT a sentinel: the first ingestion on a
    /// fresh counter commits offset 0. Only `None` means "unknown".
    recovered_position: Option<ManifestPosition>,
}

fn try_salvage_table(
    config: &Config,
    fs: &Arc<dyn crate::fs::Fs>,
    allow_resurrection: bool,
    salvage: TableSalvage<'_>,
) -> crate::Result<SalvageOutcome> {
    let TableSalvage {
        source,
        table_path,
        table_id,
        reject_punched_without_bound,
        blob_rewrite,
        owner_promotions,
        recovered_position,
    } = salvage;
    // Salvage under the tree's configured comparator + crypto/dictionary context
    // so the rewritten SST opens, orders, and decrypts / decompresses consistently
    // with the rest of the tree on reopen (the reopen below uses the same
    // `config.encryption` and the same dictionary set).
    let report = crate::salvage::salvage_with_context(
        source,
        table_path.to_path_buf(),
        fs,
        &config.comparator,
        &crate::salvage::SalvageOptions {
            encryption: config.encryption.clone(),
            #[cfg(zstd_any)]
            zstd_dictionary: config.zstd_dictionary.clone(),
            // The whole set, not just the configured one: the source may have
            // been written under a dictionary the tree stores and the caller
            // never supplied, and salvage cannot read a block it cannot resolve.
            #[cfg(zstd_any)]
            zstd_dictionaries: config.current_zstd_dictionaries(),
            // The real table id, so encrypted block AAD (which binds it) decrypts
            // and the recovered copy reopens under the same id below.
            table_id,
            // Repair KNOWS the durable id (the file name), so the salvage
            // open cross-checks the meta payload against it: a forged tail
            // id falls back to the intact MID mirror instead of stamping the
            // recovered copy with an identity the reopen below would reject.
            expected_stored_id: Some(table_id),
            // Governed by the recovery-wide resurrection flag: off (default), a
            // delete-bearing SST whose bitmap cannot be authenticated is excluded
            // rather than masked against an unverified bitmap; on, its rows are
            // re-emitted live, accepting that deleted rows reappear.
            allow_delete_resurrection: allow_resurrection,
            // The recovered SST is persisted at the tree's configured
            // durability, matching the manifest rebuilt around it.
            sync_mode: config.sync_mode,
            // The extractor is configuration, not persisted state: without
            // it the rebuilt filter loses the source's prefix hashes and
            // prefix scans see the salvaged copy as definitely absent.
            prefix_extractor: config.prefix_extractor.clone(),
            blob_rewrite,
            owner_promotions,
            // A repair replacement always keeps the SOURCE's identity: it is
            // built at `{id}.repair-tmp` and swapped onto `{id}`, so the name it
            // ends up under is the identity it was stamped with.
            output_id: None,
            // Forward the caller's live-progress handle so the block walk
            // ticks per inspected / recovered block while it runs.
            progress: config.recovery_progress.clone(),
            // The copy is stamped on the tree's clock, which FIFO judges it on.
            clock: config.clock.clone(),
        },
    )?;
    if report.salvaged_path.is_none() {
        return Ok(SalvageOutcome::Unusable);
    }
    if !report.dropped.is_empty() {
        log::warn!(
            "salvaged table {table_id}: recovered {} block(s), dropped {} corrupt block(s)",
            report.blocks_salvaged,
            report.dropped.len(),
        );
    }
    if reject_punched_without_bound
        && dropped_data_extent_is_zeroed(&**fs, source, &report.dropped)?
    {
        // The source was punched but its bound is unrecoverable: the salvaged
        // replacement re-emits every intact block — including consumed,
        // superseded blocks a partial punch left inside the reclaimed prefix —
        // with nothing to restrict them. Reject it: this copy is our own
        // byproduct, holds nothing the untouched source does not, and no
        // manifest will ever name it, so it is removed outright. A removal
        // failure PROPAGATES — left behind, the numeric SST is an orphan the
        // next open must delete, and that open would fail on the same error.
        discard_unreferenced(&**fs, table_path, config.sync_mode)?;
        return Ok(SalvageOutcome::PunchedBoundLost);
    }

    // Reopen the freshly-written (clean) salvaged SST so it joins the rebuilt
    // manifest like any cleanly-recovered table. A clean manifest record's
    // ingest offset applies to the copy unchanged (the salvage preserves
    // local seqnos), so it is reused here just like on the whole-recover path.
    let checksum = crate::Checksum::from_raw(compute_table_checksum(&**fs, table_path)?);
    let table = match Table::recover(held_recover_params(
        config,
        table_path.to_path_buf(),
        checksum,
        table_id,
        Arc::clone(fs),
        recovered_position,
    )) {
        Ok(table) => table,
        Err(e) => {
            // Same rule as the rejection below: the copy is this pass's own
            // byproduct and no manifest will ever name it, so leaving it makes
            // an orphan the next open must delete — and a retry writes another
            // one beside it. Remove it; the reopen failure is what the caller
            // sees.
            if let Err(rm) = discard_unreferenced(&**fs, table_path, config.sync_mode) {
                log::error!(
                    "salvaged copy {} could not be removed after its reopen failed ({rm}); \
                     it stays until the next orphan sweep",
                    table_path.display(),
                );
            }
            return Err(e);
        }
    };
    // A salvaged copy of a bulk-ingested source still relies on the manifest-only
    // global_seqno offset a manifest-LOSS rebuild cannot recover: its entries stay
    // at local seqno 0, so installing it with offset 0 would silently mis-order
    // and over-expose them. Without a recovered offset to reuse, treat it as
    // unsalvageable — remove the freshly-written copy and let the caller record
    // the table unreadable.
    if recovered_position.is_none()
        && has_unrecoverable_ingest_offset(
            table.metadata.bulk_ingested,
            table.metadata.item_count,
            table.max_local_seqno(),
        )
    {
        release_held(config, table);
        // Remove the rejected replacement, and do NOT swallow the error. A
        // discarded `remove_file` failure would leave the freshly-written
        // numeric SST in `tables/`; repair would still install a manifest that
        // omits it and report success, but the next open classifies it as an
        // orphan and fails on the SAME persistent deletion, so the "repaired"
        // tree cannot reopen. Propagating instead keeps the two outcomes intact.
        discard_unreferenced(&**fs, table_path, config.sync_mode)?;
        return Ok(SalvageOutcome::Unusable);
    }
    Ok(SalvageOutcome::Salvaged(table))
}

/// Whether any salvage-dropped DATA extent of `source` contains a
/// structure-anchored all-zero run — the hole-punch signature. A real data
/// block's bytes are never all zero across a whole block, so a header-length
/// run that ends where intact structure begins (a decodable block header, the
/// next dropped extent, or the data-section end) was physically reclaimed, not
/// merely corrupted. Restricting the scan to the DROPPED extents keeps
/// legitimate zero runs inside intact (checksum-clean) blocks from
/// false-positiving, and the structural anchor keeps header-sized zero runs
/// inside a damaged extent's VALUE payloads from doing the same.
///
/// The scan covers each dropped extent IN FULL, up to the next dropped extent
/// or the end of the `data` section, not just its opening window: when the
/// physical chain breaks, the salvage walk surrenders the whole remaining tail
/// as ONE extent whose offset is the first DAMAGED (nonzero) frame, so punched
/// blocks deeper inside it would otherwise stay invisible and the salvaged
/// output would publish consumed records unrestricted.
///
/// # Errors
///
/// Propagates the open / read failure (a transient one aborts the repair for a
/// retry, exactly like the other salvage-path reads).
#[cfg(feature = "std")]
fn dropped_data_extent_is_zeroed(
    fs: &dyn crate::fs::Fs,
    source: &std::path::Path,
    dropped: &[crate::salvage::DroppedBlock],
) -> crate::Result<bool> {
    let scan = excised_extents(fs, source, dropped)?;
    // An UNATTRIBUTABLE qualifying run counts too: on a punch-capable mount
    // a zero run the hole probe cannot answer for is indistinguishable from
    // a lost-sidecar reclaim, and the caller's fail-closed guard (active
    // only without resurrection) must fire rather than publish the salvaged
    // copy unrestricted.
    Ok(!scan.proven.is_empty() || scan.unattributed)
}

/// Result of [`excised_extents`]: the zero runs proven to be reclaimed, and
/// whether any qualifying run could not be attributed either way.
#[cfg(feature = "std")]
struct ExcisedScan {
    /// Structure-anchored zero runs proven to carry a physical hole,
    /// ascending and disjoint.
    proven: Vec<(u64, u64)>,
    /// Whether a qualifying zero run on a PUNCH-CAPABLE mount got `None`
    /// from the hole probe — reclaim and damage indistinguishable.
    unattributed: bool,
}

/// The physically excised byte ranges of `source`'s data section: the
/// structure-anchored all-zero runs inside its dropped extents, which is what
/// a hole punch leaves behind. Derived purely by scanning, so the result
/// survives any crash without being recorded anywhere — the property the
/// recovery model depends on for in-place repair (see
/// `docs/manifest-recovery.md`).
///
/// Ranges come back ascending and disjoint. A PREFIX punch yields one range
/// starting at the data section; a mid-file punch yields an interior range,
/// which the prefix-only restriction model cannot express — see the weak-spot
/// list in the same document.
///
/// # Errors
///
/// Propagates the open / read failure (a transient one aborts the repair for a
/// retry, exactly like the other salvage-path reads).
#[cfg(feature = "std")]
fn excised_extents(
    fs: &dyn crate::fs::Fs,
    source: &std::path::Path,
    dropped: &[crate::salvage::DroppedBlock],
) -> crate::Result<ExcisedScan> {
    // Shortest run accepted as a punch. A hole is punched per DATA BLOCK, so a
    // punched block contributes a zero run at least a block long — while inside
    // an intact block a zero run is bounded by its framing (header, key/value
    // lengths and checksums are never all zero across a whole block). The
    // block-header length is the smallest possible block, so a run of that many
    // zeros cannot come from one intact framed block.
    const MIN_RUN: u64 = crate::table::block::Header::MIN_LEN as u64;
    let mut excised: Vec<(u64, u64)> = Vec::new();
    if dropped.is_empty() {
        return Ok(ExcisedScan {
            proven: excised,
            unattributed: false,
        });
    }
    let mut file = fs.open(source, &crate::fs::FsOpenOptions::new().read(true))?;
    let file_len = crate::fs::FsFile::metadata(&*file)?.len;
    // The `data` section's physical end bounds every extent: a dropped extent
    // runs to the next dropped one or to that end, whichever comes first.
    let data_end = match crate::sfa::Reader::from_reader(&mut file) {
        Ok(reader) => reader
            .toc()
            .iter()
            .find(|e| e.name() == b"data")
            .map_or(file_len, |e| e.pos().saturating_add(e.len()).min(file_len)),
        // No readable TOC (the very corruption salvage is recovering from):
        // fall back to the file end — a superset of the data section, and the
        // per-extent scan below is bounded by the next extent anyway.
        Err(_) => file_len,
    };
    let mut starts: Vec<u64> = dropped
        .iter()
        .filter(|d| d.section == b"data" && d.offset < data_end)
        .map(|d| d.offset)
        .collect();
    starts.sort_unstable();
    starts.dedup();

    // A qualifying run must additionally be STRUCTURE-ANCHORED: it counts as
    // punch evidence only when it ends where intact structure begins — a
    // decodable block header (magic + type + the header's own checksum), the
    // next dropped extent, or the data-section end. SST values are arbitrary
    // bytes, so a header-sized zero run INSIDE a damaged extent's payload is
    // otherwise indistinguishable from a punch by length alone, and a bare
    // length test would reject an otherwise usable salvage as bound-lost under
    // the default no-resurrection policy.
    let header_decodes_at = |pos: u64| -> crate::Result<bool> {
        use crate::coding::Decode;
        let max = crate::table::block::Header::MAX_LEN as u64;
        let want = usize::try_from(file_len.saturating_sub(pos).min(max)).unwrap_or(0);
        if want < crate::table::block::Header::MIN_LEN {
            return Ok(false);
        }
        let bytes = crate::file::read_exact(&*file, pos, want)?;
        Ok(crate::table::block::Header::decode_from(&mut &bytes[..]).is_ok())
    };

    const CHUNK: usize = 64 * 1024;
    for (i, &start) in starts.iter().enumerate() {
        let end = starts.get(i + 1).copied().unwrap_or(data_end).min(data_end);
        let mut offset = start;
        let mut run: u64 = 0;
        while offset < end {
            let want = usize::try_from(end - offset).unwrap_or(CHUNK).min(CHUNK);
            let bytes = crate::file::read_exact(&*file, offset, want)?;
            for (j, &b) in bytes.iter().enumerate() {
                if b == 0 {
                    run += 1;
                } else {
                    let run_end = offset + j as u64;
                    if run >= MIN_RUN && header_decodes_at(run_end)? {
                        excised.push((run_end - run, run_end));
                    }
                    run = 0;
                }
            }
            offset += want as u64;
        }
        // A run reaching the extent end needs no header anchor: it terminates
        // at the next dropped extent or the data-section end, both of which
        // are structural boundaries themselves.
        if run >= MIN_RUN {
            excised.push((end - run, end));
        }
    }
    // The per-extent walks are already ascending and cannot overlap (each is
    // bounded by the next extent), but a run that ends exactly where the next
    // begins is one hole physically — merge so callers see maximal ranges.
    excised.sort_unstable();
    let mut merged: Vec<(u64, u64)> = Vec::with_capacity(excised.len());
    for (start, end) in excised {
        match merged.last_mut() {
            Some(last) if start <= last.1 => last.1 = last.1.max(end),
            _ => merged.push((start, end)),
        }
    }
    // Zeros are the SHAPE of a reclaim, not the proof: corruption that destroys
    // a data block leaves the same read-as-zeros run, and calling that a punch
    // condemns an otherwise salvageable table as bound-lost. A reclaim
    // deallocates, so each run must cover an actual HOLE. A run proven
    // ALLOCATED (`Some(false)`) is damage; a run the backend CANNOT attribute
    // (`None`) on a punch-capable mount is reported separately — reclaim and
    // damage are then indistinguishable, and the caller's no-resurrection
    // guard must fail closed rather than read "unproven" as "unpunched".
    //
    // Probed as CONTAINS-a-hole over the run rather than whole-span-is-a-hole
    // or a single midpoint byte: the reclaim punches at unaligned extents (an
    // older reclaim did so per block), so the filesystem deallocates only the
    // wholly-contained pages and leaves zero-filled boundary pages allocated:
    // the whole span is never a hole, and a midpoint byte can land on one of
    // those allocated boundary pages. Any hole inside the zeroed run is attributable to it.
    // Only the presence of the hole matters here — the single caller asks
    // whether the extent was reclaimed at all, not exactly where.
    let mut proven = Vec::with_capacity(merged.len());
    let mut unattributed = false;
    for (start, end) in merged {
        match fs.extent_contains_hole(source, start, end - start)? {
            Some(true) => proven.push((start, end)),
            Some(false) => {}
            None => unattributed = true,
        }
    }
    if unattributed {
        unattributed = fs.capabilities(source).punch_hole;
    }
    Ok(ExcisedScan {
        proven,
        unattributed,
    })
}

/// Whether the source SST's first data block reads as all zeros, the signature of
/// a hole-punched prefix. Probed at offset 0 (the `data` section is written first)
/// over a small window that stays WITHIN the first block, so even a punch of only
/// the first block is detected; a real data block's opening bytes (its first
/// entry's key-length varint and key) are never all zero, so an unpunched SST can
/// never false-positive.
///
/// Used by the recovery-failure salvage arm to fail closed on a PUNCHED source
/// whose bound is unrecoverable (missing / corrupt sidecar and no `Table` to
/// derive from): salvaging it into an unrestricted output would resurrect the
/// straddling block's sub-bound rows.
///
/// This is the CHEAP pre-salvage fast path only: a PARTIAL punch (the
/// punch-on-drop reclaim continues past an individual `punch_hole` failure) can
/// leave the first block intact while later prefix blocks are zeroed, which this
/// probe cannot see. [`dropped_data_extent_is_zeroed`] closes that gap after the
/// salvage walk, whose dropped extents expose every zeroed block.
///
/// # Errors
///
/// Propagates the open / read failure.
#[cfg(feature = "std")]
fn source_prefix_is_punched(
    fs: &dyn crate::fs::Fs,
    table_path: &std::path::Path,
) -> crate::Result<bool> {
    const PROBE: usize = 64;
    let file = fs.open(table_path, &crate::fs::FsOpenOptions::new().read(true))?;
    let len = crate::fs::FsFile::metadata(&*file)?.len;
    if len == 0 {
        return Ok(false);
    }
    let n = usize::try_from(len).unwrap_or(PROBE).min(PROBE);
    let bytes = crate::file::read_exact(&*file, 0, n)?;
    if !bytes.iter().all(|&b| b == 0) {
        return Ok(false);
    }
    // Zeros are not the evidence — corruption produces them too, and reading
    // ordinary damage as a lost punch bound sets the whole table aside instead
    // of salvaging the blocks that are still readable. A reclaim deallocates,
    // so the probed window must carry a HOLE — contains-a-hole, the same rule
    // the block-level punch classifier applies (an unaligned punch keeps its
    // zero-filled edges allocated).
    //
    // A backend that cannot answer (`None`) reports NOT-punched here, which is
    // deliberately the permissive direction: this is only the cheap fast path,
    // and answering "punched" on an unproven window would set aside every
    // ordinarily-damaged table on a mount without extent reporting. The
    // fail-closed decision for unattributable zeros belongs to
    // `dropped_data_extent_is_zeroed` after the salvage walk, which sees every
    // dropped all-zero extent rather than just this window and reports them
    // (see `ExcisedScan::unattributed`).
    Ok(fs.extent_contains_hole(table_path, 0, n as u64)? == Some(true))
}

/// Re-imposes a tight-space restriction on a SALVAGED replacement SST, the single
/// point every salvage output funnels through so the restriction can never be
/// dropped on one path and kept on another.
///
/// Salvage rewrites its source as a fresh, UNPUNCHED table that re-emits the
/// straddling block's sub-bound rows, so a punched source's restriction must be
/// re-applied to the output or those superseded / deleted rows resurrect. With a
/// known `bound` and resurrection off, the sidecar is re-written (so a later
/// manifest-loss repair honors it against the now-unpunched file) and the table
/// reopened restricted. Otherwise (resurrection on, or no recoverable bound) the
/// salvaged table is kept whole and any stale sidecar cleared, since a lingering
/// sidecar would wrongly restrict the unpunched replacement on a later repair.
///
/// ANY failure re-imposing the restriction (sidecar write or restricted reopen)
/// removes the half-finished replacement before propagating. It carries the
/// straddling block's sub-bound rows with no valid sidecar, so a run that found
/// it would adopt it UNRESTRICTED and resurrect exactly the rows the restriction
/// hid. The source is untouched either way, so the retry re-salvages and
/// re-restricts from it. This holds for a PERSISTENT failure (an ENOSPC on the
/// sidecar write) as much as a transient one.
#[cfg(feature = "std")]
fn restrict_salvaged_output(
    folder_fs: &dyn crate::fs::Fs,
    config: &Config,
    table_path: &std::path::Path,
    salvaged: Table,
    restrict_bound: Option<crate::UserKey>,
    allow_resurrection: bool,
) -> crate::Result<Table> {
    match restrict_bound {
        Some(bound) if !allow_resurrection => {
            let table_id = salvaged.metadata.id;
            let restricted = crate::restrict_bound::write(
                folder_fs,
                table_path,
                config.encryption.as_deref(),
                table_id,
                &bound,
                config.sync_mode,
            )
            .and_then(|()| salvaged.reopen_restricted_with(bound, held_descriptors(config)));
            match restricted {
                Ok(table) => Ok(table),
                Err(e) => {
                    // Drop the salvaged handle's open file BEFORE removing it: a
                    // backend that refuses to unlink an OPEN file (Windows; the
                    // deletion path closes handles for this same reason) would
                    // fail while `salvaged` still holds it.
                    release_held(config, salvaged);
                    // Remove on EVERY failure, transient or persistent: the
                    // replacement is unpunched with no valid sidecar, so a run
                    // that adopted it would resurrect the sub-bound rows. The
                    // source is untouched, so the retry re-salvages from it. A
                    // removal that itself fails propagates THAT error — an
                    // undiscardable half-finished replacement is the one thing
                    // that could corrupt the retry.
                    discard_unreferenced(folder_fs, table_path, config.sync_mode)?;
                    Err(e)
                }
            }
        }
        _ => {
            crate::restrict_bound::remove(folder_fs, table_path, config.sync_mode);
            Ok(salvaged)
        }
    }
}

/// The scanned totals of a validated blob file's LIVE frames. For a punched
/// file these cover only the suffix, and the difference to the whole-file
/// metadata totals is exactly the punched prefix's garbage — what seeds the
/// rebuilt manifest's fragmentation accounting.
#[cfg(feature = "std")]
struct BlobLiveTotals {
    items: u64,
    uncompressed_bytes: u64,
    compressed_bytes: u64,
}

/// What a blob file's punch-geometry walk proved about its live data.
#[cfg(feature = "std")]
#[derive(Debug)]
enum BlobFrontier {
    /// No punched prefix: the whole file is live.
    Whole,
    /// Live data begins at this byte offset; the `[data_start, offset)` prefix
    /// is punched.
    Punched(u64),
    /// The punch consumed EVERY frame: the relocation completed and only the
    /// file's removal lagged the crash. No live data remains.
    FullyConsumed,
}

/// Derives a blob file's tight-space live-data frontier from its on-disk punch
/// geometry, for a manifest-loss repair.
///
/// The frontier — where a tight-space relocation's punched `[data_start,
/// frontier)` prefix ends and the live suffix begins — is recorded only in the
/// manifest's `blob_restrictions` section. Unlike an SST's restriction bound (a
/// KEY, which the block-aligned punch cannot reproduce and which therefore
/// needs its `.restrict-bound` sidecar), the blob frontier is a byte offset at
/// a frame boundary, so the geometry recovers it EXACTLY: the punch zeroes
/// precisely `[data_start, frontier)` and the first live frame's magic sits at
/// `frontier`.
///
/// Anchoring is structural, never length-based: a zeroed run counts only when a
/// frame decodes cleanly at its end, so a zero-filled value payload inside the
/// live suffix (stepped over by frame framing) can never move the frontier. A
/// partially completed punch (a crash mid-reclaim can leave intact-but-consumed
/// frames between holes) is walked hole-by-hole, and the frontier is the end of
/// the LAST zeroed run the anchored walk reaches. Non-zero bytes that fail to
/// decode end the walk at the last anchored frontier: content corruption is not
/// punch geometry, and it surfaces exactly as it would on an unpunched file.
///
/// Returns [`BlobFrontier::Whole`] when the first data byte is non-zero: the
/// punch always starts at the data start, so an unpunched file — including one
/// whose committed punch never ran before a crash — short-circuits without a
/// walk, keeping the common repair path at zero extra read cost. The redundant
/// unpunched prefix is superseded by relocated copies and reclaimed later, the
/// same safe fallback the SST path takes for a committed-but-unpunched slice.
/// Zeros through the whole data section are [`BlobFrontier::FullyConsumed`]:
/// a completed relocation whose file removal lagged the crash.
///
/// # Errors
///
/// Propagates I/O and TOC errors (the caller classifies transient ones for
/// retry, like every other per-file repair probe).
fn derive_blob_frontier(
    fs: &Arc<dyn crate::fs::Fs>,
    path: &std::path::Path,
    blob_id: crate::vlog::BlobFileId,
) -> crate::Result<BlobFrontier> {
    let mut file = fs.open(path, &crate::fs::FsOpenOptions::new().read(true))?;
    let (data_start, data_end) = {
        let reader = crate::sfa::Reader::from_reader(&mut file)?;
        let data = reader
            .toc()
            .section(b"data")
            .ok_or(crate::Error::InvalidHeader("BlobFile"))?;
        let end = data
            .pos()
            .checked_add(data.len())
            .ok_or(crate::Error::InvalidHeader("BlobFile"))?;
        (data.pos(), end)
    };
    if data_start >= data_end {
        return Ok(BlobFrontier::Whole);
    }

    // Ends of the contiguous all-zero run starting at `from` (chunked reads,
    // capped by the data-section end).
    let skip_zeros = |from: u64| -> crate::Result<u64> {
        const CHUNK: u64 = 64 * 1024;
        let mut pos = from;
        while pos < data_end {
            // `#[allow]`, not `#[expect]`: target-width-dependent lint (`u64 as
            // usize`) — on 64-bit targets Clippy proves the `min()` bound fits
            // usize and an `#[expect]` would be unfulfilled under `-D warnings`.
            #[allow(
                clippy::cast_possible_truncation,
                reason = "min() bounds the window by CHUNK, which fits usize"
            )]
            let want = (data_end - pos).min(CHUNK) as usize;
            let chunk = crate::file::read_exact(&*file, pos, want)?;
            match chunk.iter().position(|b| *b != 0) {
                Some(hit) => return Ok(pos + hit as u64),
                None => pos += want as u64,
            }
        }
        Ok(data_end)
    };

    // Whether a zero run was physically RECLAIMED rather than merely zeroed.
    // Zeros are not the evidence: ordinary corruption produces exactly the same
    // bytes, and reading damage as geometry drops every handle below a
    // fabricated frontier — or, for a wholly zeroed section, removes a file
    // whose records were only damaged. A punch deallocates, so the run must
    // read back as a hole. The same rule the SST classifiers apply.
    //
    // Probed as CONTAINS-a-hole over the run: the run is bounded by the
    // neighbouring non-zero bytes (so demanding the whole span be unallocated
    // would reject genuine punches), and the punch's zero-filled edge pages
    // stay allocated on a real filesystem (so a single midpoint byte can land
    // on an allocated page even inside a genuine reclaim). A backend that
    // cannot answer leaves the run unproven, which keeps the zeros classified
    // as damage.
    let is_reclaimed = |from: u64, to: u64| -> crate::Result<bool> {
        if to <= from {
            return Ok(false);
        }
        Ok(fs.extent_contains_hole(path, from, to - from)? == Some(true))
    };

    // Fast path: an unpunched file's first frame magic (non-zero) sits at the
    // data start.
    if skip_zeros(data_start)? == data_start {
        return Ok(BlobFrontier::Whole);
    }

    // The last structure-anchored frontier: committed only once a frame has
    // decoded cleanly at a zeroed run's end. Zero = no run was ever anchored,
    // which reports the file whole (an unproven punch is content, not
    // geometry).
    let committed = |c: u64| {
        if c == 0 {
            BlobFrontier::Whole
        } else {
            BlobFrontier::Punched(c)
        }
    };
    let mut pos = data_start;
    let mut anchored: u64 = 0;
    loop {
        let run_end = skip_zeros(pos)?;
        // An UNRECLAIMED run is destroyed content, not geometry: stop at the
        // last proven frontier and let validation surface the damage exactly as
        // it would on an unpunched file.
        if !is_reclaimed(pos, run_end)? {
            return Ok(committed(anchored));
        }
        if run_end >= data_end {
            // Zeros to the section end. Only a completed relocation when
            // NOTHING was anchored below them: reclaim punches the consumed
            // prefix top-down from the data start, so a zeroed tail that
            // FOLLOWS intact anchored frames cannot be punch geometry — it is
            // destroyed data. Reporting it as consumed would drop a file whose
            // live frames are still referenced (and every table pointing at
            // it); keeping the anchored frontier instead lets the damaged
            // suffix surface through validation, exactly as the same damage
            // would on an unpunched file.
            return Ok(if anchored == 0 {
                BlobFrontier::FullyConsumed
            } else {
                committed(anchored)
            });
        }
        let mut scanner = crate::vlog::BlobFileScanner::resume(path, &**fs, blob_id, run_end)?;
        match scanner.next() {
            Some(Ok(entry)) if !entry.resynced => {
                anchored = run_end;
                pos = entry.frame_end;
            }
            Some(Err(e)) if is_environmental(&e) => return Err(e),
            // The zeroed run is not punch geometry (no frame decodes at its
            // end): keep the last anchored frontier.
            _ => return Ok(committed(anchored)),
        }
        // Chain frames from the anchor until the section ends cleanly or the
        // chain breaks (another hole, or content corruption).
        loop {
            match scanner.next() {
                None => return Ok(committed(anchored)),
                Some(Ok(entry)) if !entry.resynced => pos = entry.frame_end,
                Some(Err(e)) if is_environmental(&e) => return Err(e),
                Some(Ok(_) | Err(_)) => {
                    // The frame starting at `pos` failed (or the scanner
                    // resynced past unproven bytes). Another zeroed hole
                    // continues the walk; anything else is content corruption
                    // and ends it at the last anchored frontier.
                    if skip_zeros(pos)? == pos {
                        return Ok(committed(anchored));
                    }
                    break;
                }
            }
        }
    }
}

/// Records a recovered table the blob-dependency stage cannot publish: reported
/// unreadable, and removed once the manifest is durable. Consumes the handle.
///
/// The file is NOT touched here. This stage runs before the commit, where the
/// source is still the only copy of its rows and a crash must leave the
/// directory exactly as a retry expects to find it.
#[cfg(feature = "std")]
fn set_aside_table(
    table: Table,
    reason: &str,
    unreadable_files: &mut Vec<(PathBuf, String)>,
    discard_after_commit: &mut Vec<(Arc<dyn crate::fs::Fs>, PathBuf, String)>,
) {
    let path = (*table.path).clone();
    let fs = table.fs.clone();
    drop(table); // release the handle
    discard_after_commit.push((fs, path.clone(), reason.to_string()));
    unreadable_files.push((path, reason.to_string()));
}

/// Same as [`set_aside_table`] for a source that is no longer open.
#[cfg(feature = "std")]
fn set_aside_path(
    fs: &Arc<dyn crate::fs::Fs>,
    path: &std::path::Path,
    reason: &str,
    unreadable_files: &mut Vec<(PathBuf, String)>,
    discard_after_commit: &mut Vec<(Arc<dyn crate::fs::Fs>, PathBuf, String)>,
) {
    discard_after_commit.push((Arc::clone(fs), path.to_path_buf(), reason.to_string()));
    unreadable_files.push((path.to_path_buf(), reason.to_string()));
}

/// Whether every frame in `path`'s live data range (`[live_data_start, end)`)
/// verifies. Repair must not record a digest over damaged content: the
/// restamped digest would launder the corruption past every later integrity
/// check while reads of the affected values still fail — such a file is
/// salvaged instead. Framing checks alone are not enough, because the frame
/// checksum is unkeyed and covers only the ON-DISK bytes; each acceptance
/// criterion below closes a distinct restamp/reorder shape:
///
/// - every frame decodes and checksums cleanly, with no resynchronization;
/// - a compressed frame's payload DECOMPRESSES (a re-stamped checksum over an
///   undecodable compressed payload frames cleanly, yet every live read of
///   the value fails);
/// - frame keys never regress under the tree comparator (individually-valid
///   frames reordered on disk break the sorted-input contract every blob
///   reader and the relocation merge scanner rely on);
/// - for an unpunched file, the metadata counters match the scanned frames
///   (the meta block's item count, uncompressed byte total, and key range are
///   what blob GC's dead-file arithmetic trusts — an understated total lets
///   `is_dead` reclaim a file whose uncounted frames are still referenced).
///   A punched file's metadata describes the whole original file while the
///   scan covers only the live suffix, so it is checked against the LOWER
///   bounds the subset relation implies instead: totals at least the suffix
///   totals, key range containing the scanned suffix.
///
/// Returns the scanned live totals on success — the caller seeds the punched
/// prefix's garbage accounting from their difference to the whole-file
/// metadata.
///
/// # Errors
///
/// Propagates transient I/O for retry; any structural or persistent frame
/// failure is a conclusive `Ok(None)`.
#[cfg(feature = "std")]
fn validate_blob_frames(
    config: &Config,
    path: &std::path::Path,
    blob_id: crate::vlog::BlobFileId,
    live_data_start: u64,
    // The handle the caller already opened for the identity check: its meta
    // section carries both the compression descriptor and the recorded
    // totals this cross-check needs, so re-opening it here read the same
    // section twice per file.
    handle: &crate::vlog::BlobFile,
    // What `handle`'s descriptor names (see `blob_file_dictionary`), resolved
    // by the caller, which alone knows whether a missing one stops the repair.
    #[cfg(zstd_any)] dict: Option<&Arc<crate::compression::ZstdDictionary>>,
) -> crate::Result<Option<BlobLiveTotals>> {
    let fs = &config.fs;
    let compression = handle.compression();
    let comparator = &config.comparator;

    let scanner = if live_data_start > 0 {
        crate::vlog::BlobFileScanner::resume(path, &**fs, blob_id, live_data_start)?
    } else {
        crate::vlog::BlobFileScanner::new(path, &**fs, blob_id)?
    };
    let mut count: u64 = 0;
    let mut uncompressed_total: u64 = 0;
    let mut compressed_total: u64 = 0;
    let mut first_key: Option<crate::UserKey> = None;
    let mut prev: Option<(crate::UserKey, crate::SeqNo)> = None;
    for item in scanner {
        match item {
            Ok(entry) if !entry.resynced => {
                if crate::salvage::blob_key_regresses(comparator, prev.as_ref(), &entry) {
                    log::warn!(
                        "blob file {blob_id} at {}: frame at {} regresses below the \
                         previous frame's key — the frames were reordered",
                        path.display(),
                        entry.offset,
                    );
                    return Ok(None);
                }
                if crate::salvage::decompress_blob_value(
                    compression,
                    &entry.value,
                    entry.uncompressed_len as usize,
                    #[cfg(zstd_any)]
                    dict.map(Arc::as_ref),
                )
                .is_err()
                {
                    log::warn!(
                        "blob file {blob_id} at {}: frame at {} does not decompress \
                         despite a clean checksum",
                        path.display(),
                        entry.offset,
                    );
                    return Ok(None);
                }
                count += 1;
                uncompressed_total += u64::from(entry.uncompressed_len);
                // Sum of u32-bounded on-disk lengths within one file: cannot
                // overflow u64 (the file itself cannot reach 2^64 bytes).
                compressed_total += entry.value.len() as u64;
                if first_key.is_none() {
                    first_key = Some(entry.key.clone());
                }
                prev = Some((entry.key.clone(), entry.seqno));
            }
            Err(e) if is_environmental(&e) => return Err(e),
            // A resynced frame has an unprovable boundary (damage upstream);
            // any other error is a structural or persistent frame failure.
            // Both are conclusive: this file's frames do not all verify.
            Ok(_) | Err(_) => return Ok(None),
        }
    }

    let meta = handle.meta();
    if live_data_start == 0 {
        let range_matches = match (&first_key, &prev) {
            (Some(first), Some((last, _))) => {
                meta.key_range.min().as_ref() == first.as_ref()
                    && meta.key_range.max().as_ref() == last.as_ref()
            }
            _ => count == 0,
        };
        if meta.item_count != count
            || meta.total_uncompressed_bytes != uncompressed_total
            || meta.total_compressed_bytes != compressed_total
            || !range_matches
        {
            log::warn!(
                "blob file {blob_id} at {}: metadata disagrees with the scanned frames \
                 (meta: {} items / {} uncompressed bytes; scanned: {count} / \
                 {uncompressed_total})",
                path.display(),
                meta.item_count,
                meta.total_uncompressed_bytes,
            );
            return Ok(None);
        }
    } else {
        // A punched file's metadata describes the WHOLE original file while
        // the scan covers only the live suffix — a subset — so exact equality
        // is impossible. The subset relation still bounds the metadata from
        // BELOW: totals must be at least the suffix totals and the key range
        // must contain the scanned suffix. Blessing understated totals would
        // let blob GC's dead-file arithmetic reclaim a file whose uncounted
        // frames are still referenced.
        let range_contains = match (&first_key, &prev) {
            (Some(first), Some((last, _))) => {
                comparator.compare(meta.key_range.min().as_ref(), first.as_ref())
                    != core::cmp::Ordering::Greater
                    && comparator.compare(last.as_ref(), meta.key_range.max().as_ref())
                        != core::cmp::Ordering::Greater
            }
            // An empty suffix constrains nothing.
            _ => true,
        };
        if meta.item_count < count
            || meta.total_uncompressed_bytes < uncompressed_total
            || meta.total_compressed_bytes < compressed_total
            || !range_contains
        {
            log::warn!(
                "blob file {blob_id} at {}: metadata understates the scanned live \
                 suffix (meta: {} items / {} uncompressed bytes; suffix: {count} / \
                 {uncompressed_total})",
                path.display(),
                meta.item_count,
                meta.total_uncompressed_bytes,
            );
            return Ok(None);
        }
    }
    Ok(Some(BlobLiveTotals {
        items: count,
        uncompressed_bytes: uncompressed_total,
        compressed_bytes: compressed_total,
    }))
}

/// Whether any of `table`'s blob indirections points BELOW a recovered punched
/// blob file's live-data frontier — i.e. into its zeroed prefix. The
/// id-presence dependency check cannot see this case: the blob EXISTS, but a
/// pre-relocation SST file left behind by a crash still holds handles into the
/// prefix the relocation punched, and publishing it would resolve those reads
/// into zeroed bytes. Returns the first offending handle's description, or
/// `None` when every handle lands in live blob data. Called only when at least
/// one recovered blob file carries a frontier, so the sequential entry scan
/// costs nothing on the common path.
#[cfg(feature = "std")]
fn handle_below_blob_frontier(
    table: &Table,
    frontiers: &crate::HashMap<crate::vlog::BlobFileId, u64>,
) -> crate::Result<Option<String>> {
    use crate::coding::Decode;

    for entry in table.scan()? {
        let entry = entry?;
        let handles = if entry.key.value_type == crate::ValueType::Indirection {
            let mut cursor = &entry.value[..];
            vec![crate::blob_tree::handle::BlobIndirection::decode_from(
                &mut cursor,
            )?]
        } else if entry.key.value_type == crate::ValueType::CellRow {
            crate::blob_tree::field_row::row_refs(&entry.value)?
                .into_iter()
                .map(|(ind, _)| ind)
                .collect()
        } else {
            continue;
        };
        for ind in handles {
            if let Some(&frontier) = frontiers.get(&ind.vhandle.blob_file_id)
                && ind.vhandle.offset < frontier
            {
                return Ok(Some(format!(
                    "blob handle into file {} at offset {} lies below its recovered \
                     live-data frontier {frontier}",
                    ind.vhandle.blob_file_id, ind.vhandle.offset,
                )));
            }
        }
    }
    Ok(None)
}

/// One table the blob-handle stage keeps: the table, its fidelity, its
/// scanned source and that source's filesystem, and whether its handles need
/// rewriting.
#[cfg(feature = "std")]
type AdmittedTable = ((Table, Fidelity, PathBuf, Arc<dyn crate::fs::Fs>), bool);

/// For each table of `tables`, the cell rows to make the owners of the
/// objects whose owning row the rebuild does not keep, `None` for a table
/// with none.
///
/// An object is owned by one row of its key, and only that row's drop charges
/// the object as garbage: an object whose owner went with a table set aside,
/// a block dropped or a row the handle rewrite loses, while a row borrowing it
/// stays, would never be counted again. The oldest row left that references
/// it becomes its owner, as a compaction hands ownership to the oldest row it
/// keeps. The rows a table's handle rewrite will lose are left out here, as
/// the rewrite leaves them out. A version a table and a redundant copy of it
/// both hold is named in both, so the copy the lineage pass keeps owns it.
#[cfg(feature = "std")]
fn owner_promotions(
    tables: &[AdmittedTable],
    rewrites: &crate::HashMap<crate::vlog::BlobFileId, crate::salvage::BlobFileRewrite>,
) -> crate::Result<Vec<Option<crate::salvage::OwnerPromotions>>> {
    /// A row referencing an object by the field in `column`.
    struct Holder {
        table: usize,
        key: UserKey,
        seqno: SeqNo,
        column: u16,
    }
    type Object = (crate::vlog::BlobFileId, u64);

    // Only borrowed references can lose their owner, and they are rare: a
    // first pass finds them and the oldest of each object's borrowers.
    let mut borrowed: crate::HashMap<Object, (SeqNo, Vec<Holder>)> = crate::HashMap::default();
    for (index, ((table, ..), rewrite)) in tables.iter().enumerate() {
        if table
            .list_blob_file_references()?
            .is_none_or(|links| links.is_empty())
        {
            continue;
        }
        let global = table.global_seqno();
        visit_kept_refs(table, rewrite.then_some(rewrites), |key, seqno, field| {
            let Some((column, true)) = field.owner else {
                return Ok(());
            };
            let eff = seqno
                .checked_add(global)
                .ok_or(crate::Error::Unrecoverable)?;
            let holder = Holder {
                table: index,
                key: key.clone(),
                seqno,
                column,
            };
            let object = (field.file, field.offset);
            match borrowed.get_mut(&object) {
                Some((oldest, holders)) if eff == *oldest => holders.push(holder),
                Some((oldest, holders)) if eff < *oldest => {
                    *oldest = eff;
                    *holders = alloc::vec![holder];
                }
                Some(_) => {}
                None => {
                    borrowed.insert(object, (eff, alloc::vec![holder]));
                }
            }
            Ok(())
        })?;
    }
    let mut promotions: Vec<Option<crate::salvage::OwnerPromotions>> =
        tables.iter().map(|_| None).collect();
    if borrowed.is_empty() {
        return Ok(promotions);
    }

    // Then every owning reference into the files those objects are in.
    let files: crate::HashSet<crate::vlog::BlobFileId> =
        borrowed.keys().map(|&(file, _)| file).collect();
    for ((table, ..), rewrite) in tables {
        let linked = table
            .list_blob_file_references()?
            .is_some_and(|links| links.iter().any(|l| files.contains(&l.blob_file_id)));
        if !linked {
            continue;
        }
        visit_kept_refs(table, rewrite.then_some(rewrites), |_, _, field| {
            if field.owner.is_none_or(|(_, borrows)| !borrows) {
                borrowed.remove(&(field.file, field.offset));
            }
            Ok(())
        })?;
    }

    for (_, (_, holders)) in borrowed {
        for holder in holders {
            if let Some(slot) = promotions.get_mut(holder.table) {
                slot.get_or_insert_with(Default::default).insert(
                    holder.key,
                    holder.seqno,
                    holder.column,
                );
            }
        }
    }
    Ok(promotions)
}

/// A blob reference a row holds: the object's file and offset, and for a
/// cell row's field its column and whether the row owns the object, `None`
/// for an indirection, which always does.
#[cfg(feature = "std")]
#[derive(Clone, Copy)]
struct HeldRef {
    file: crate::vlog::BlobFileId,
    offset: u64,
    /// The field's column and whether the row only BORROWS the object.
    owner: Option<(u16, bool)>,
}

/// Calls `visit` with every blob reference of `table`'s rows, by the row's
/// key and local seqno, leaving out the rows the handle rewrite `rewrites`
/// loses: a row with a reference to a record that no longer exists, and every
/// older version of its key in the table, as the rewrite does.
#[cfg(feature = "std")]
fn visit_kept_refs(
    table: &Table,
    rewrites: Option<&crate::HashMap<crate::vlog::BlobFileId, crate::salvage::BlobFileRewrite>>,
    mut visit: impl FnMut(&UserKey, SeqNo, HeldRef) -> crate::Result<()>,
) -> crate::Result<()> {
    use crate::blob_tree::field_row::{RowCell, decode_row};
    use crate::coding::Decode;
    use crate::salvage::BlobFileRewrite;

    let lost =
        |file: crate::vlog::BlobFileId, offset: u64| match rewrites.and_then(|r| r.get(&file)) {
            Some(BlobFileRewrite::Remap { offsets, .. }) => !offsets.contains_key(&offset),
            Some(BlobFileRewrite::DropBelow(frontier)) => offset < *frontier,
            None => false,
        };
    let mut headless: Option<UserKey> = None;
    let mut refs: Vec<HeldRef> = Vec::new();
    for entry in table.scan()? {
        let entry = entry?;
        let key = &entry.key.user_key;
        if headless
            .as_ref()
            .is_some_and(|h| crate::comparator::same_user_key(key, h))
        {
            continue;
        }
        headless = None;
        refs.clear();
        match entry.key.value_type {
            crate::ValueType::Indirection => {
                let mut cursor = &entry.value[..];
                let ind = crate::blob_tree::handle::BlobIndirection::decode_from(&mut cursor)?;
                refs.push(HeldRef {
                    file: ind.vhandle.blob_file_id,
                    offset: ind.vhandle.offset,
                    owner: None,
                });
            }
            crate::ValueType::CellRow => {
                for field in decode_row(&entry.value)? {
                    if let RowCell::Ref { indirection, owner } = field.cell {
                        refs.push(HeldRef {
                            file: indirection.vhandle.blob_file_id,
                            offset: indirection.vhandle.offset,
                            owner: Some((field.column, !owner)),
                        });
                    }
                }
            }
            _ => continue,
        }
        if refs.iter().any(|r| lost(r.file, r.offset)) {
            headless = Some(key.clone());
            continue;
        }
        for &held in &refs {
            visit(key, entry.key.seqno, held)?;
        }
    }
    Ok(())
}

/// What the blob directory scan recovered, and how, for the rebuilt manifest.
#[cfg(feature = "std")]
struct BlobRecovery {
    /// Blob files to record in the manifest (whole, punched, or salvaged).
    files: Vec<crate::vlog::BlobFile>,
    /// Files LEFT OUT of the manifest, with the reason each was set aside.
    unreadable: UnreadableFiles,
    /// Handle rewrites for reshaped files (salvaged / punched frontiers).
    rewrites: crate::HashMap<crate::vlog::BlobFileId, crate::salvage::BlobFileRewrite>,
    /// Garbage accounting seeded for punched files: the consumed prefix is
    /// stale by construction but can never be observed by a future
    /// compaction, so without this seed `is_dead` could never retire the
    /// file.
    frag: crate::blob_tree::FragmentationMap,
    /// Damaged originals whose salvaged replacement (a FRESH id) is in the
    /// rebuilt manifest, with a note describing what was recovered. They are
    /// still what the not-yet-rewritten SSTs reference, so the caller sets
    /// them aside only AFTER `persist_version`: doing it earlier would let a
    /// failed repair leave tables pointing at a blob id that no longer
    /// exists, and the retry would set those tables aside as unrecoverable.
    stale: Vec<(PathBuf, String)>,
    /// Files the rebuilt manifest will not name: a foreign name, a duplicate id,
    /// a file whose metadata cannot be read. Removed AFTER the commit, for the
    /// same reason `stale` is — the scan must leave the directory exactly as a
    /// retry expects to find it.
    discard: Vec<(PathBuf, String)>,
    /// HEALTHY exclusions (a valid physically distinct duplicate of a kept
    /// id): reported through the report's `excluded_files`, never the
    /// unreadable counts.
    excluded: Vec<(PathBuf, String)>,
}

/// Discovers the blob files of a KV-separated tree for `repair` by scanning the
/// single `blobs/` folder, with no manifest id list to filter against.
///
/// Mirrors the table scan in [`repair_tree`]: a non-numeric name is recorded for
/// removal after the commit (the reopened tree's blob recovery parses every name
/// and would abort on a bad one), as is a blob file that cannot be checksummed
/// or whose metadata is unreadable. The recovered checksum is the whole-file
/// XXH3-128 digest, identical to the one the blob writer accumulated via
/// `ChecksummedWriter`, since blob files are written strictly sequentially.
fn recover_blob_files(
    config: &Config,
    published: &mut PublishedBlobReplacements<'_>,
    // Blob ids the recovered tables reference (a conservative SUPERSET: taken
    // before the blob-dependency filtering, so a table dropped later only
    // leaves an extra id here). An INVALID blob outside this set is left for
    // the post-commit sweep instead of being salvaged: the replacement would
    // be filtered out and deleted anyway, and under tight disk space the
    // pointless salvage can abort the whole repair with `StorageFull`.
    referenced: &crate::HashSet<crate::vlog::BlobFileId>,
    // The CLEANLY-loading manifest's committed blob frontiers, when one
    // exists. Each is the AUTHORITATIVE `[data_start, frontier)` boundary of a
    // tight-space-punched blob file — the same authority its `restrictions`
    // twin carries for SSTs — so it is preferred over re-deriving the frontier
    // from the on-disk punch geometry, which a backend that cannot report
    // extent allocation gets wrong (reporting a punched file whole, whose
    // handles then dereference the hole).
    committed_frontiers: Option<&crate::HashMap<crate::vlog::BlobFileId, u64>>,
    committed_checksums: Option<&crate::HashMap<crate::vlog::BlobFileId, crate::Checksum>>,
) -> crate::Result<BlobRecovery> {
    let blobs_folder = config.path.join(crate::file::BLOBS_FOLDER);
    let mut blob_files: Vec<crate::vlog::BlobFile> = Vec::new();
    let mut unreadable: UnreadableFiles = Vec::new();
    let mut discard: Vec<(PathBuf, String)> = Vec::new();
    let mut excluded: Vec<(PathBuf, String)> = Vec::new();
    // How referencing SSTs' handles must be rewritten for the blob files this
    // scan RESHAPED: `Remap` for a file salvaged into a compacted copy,
    // `DropBelow` for an intact file recovered with a punched frontier. Empty
    // on the common path.
    let mut rewrites: crate::HashMap<crate::vlog::BlobFileId, crate::salvage::BlobFileRewrite> =
        crate::HashMap::default();
    // Damaged originals whose salvaged replacement is in the rebuilt manifest.
    // They are still referenced by the not-yet-rewritten SSTs, so they are set
    // aside only AFTER the manifest commit — see `BlobRecovery::stale`.
    let mut stale_originals: Vec<(PathBuf, String)> = Vec::new();
    let mut frag = crate::blob_tree::FragmentationMap::default();

    // No `blobs/` folder = no blob files (a blob tree that never spilled a value
    // to the value log). Nothing to recover; the manifest records an empty list.
    if !config.fs.exists(&blobs_folder)? {
        return Ok(BlobRecovery {
            files: blob_files,
            unreadable,
            rewrites,
            frag,
            stale: stale_originals,
            discard,
            excluded,
        });
    }

    // Collect and ORDER the candidates before recovering: `read_dir` order is
    // FS-dependent, and duplicate-id resolution below must be deterministic.
    // Per id, the writer's own `id.to_string()` spelling is the canonical file
    // and sorts first, so a foreign alternate spelling (`01` for id 1) can
    // never displace it regardless of directory iteration order.
    let mut candidates: Vec<(crate::vlog::BlobFileId, PathBuf, String)> = Vec::new();
    for dirent in config.fs.read_dir(&blobs_folder)? {
        let crate::fs::FsDirEntry {
            path: blob_path,
            file_name,
            is_dir,
        } = dirent;

        if is_dir {
            continue;
        }

        // The file counts toward the progress percentage as soon as its
        // processing starts — including the entries the early arms below
        // reject (a crashed salvage temp, a non-numeric name), or a finished
        // repair would sit below 100% (`bytes_processed < bytes_total`).
        // Mirrors the table scan; best-effort stat, like the total.
        if let Some(p) = &config.recovery_progress
            && let Ok(meta) = config.fs.metadata(&blob_path)
        {
            p.add_bytes_processed(meta.len);
        }

        let blob_id = match crate::file::BlobDirEntry::classify(&file_name) {
            crate::file::BlobDirEntry::Blob(id) => id,
            // A crashed earlier repair's in-progress salvage copy: published by
            // an atomic rename, so a survivor is never referenced and never
            // authoritative. Removed here (and re-salvaged from the original
            // below if that file still fails validation).
            //
            // A removal failure fails the repair: the temp is outside the
            // rebuilt manifest, so the next open classifies it as an orphan
            // and its sweep hits the same removal failure — reporting success
            // for a tree that cannot open would be a lie. Quarantine is not
            // an out (it preserves damaged DATA; a temp is discardable
            // garbage, and a directory refusing removal refuses the rename
            // too). Retry after the filesystem is fixed.
            crate::file::BlobDirEntry::SalvageTmp(_) => {
                remove_temp(config, &blob_path)?;
                continue;
            }
            // Not a shape the engine names: not part of the inventory being
            // rebuilt, and not the repair's to remove.
            crate::file::BlobDirEntry::Foreign => {
                log::debug!(
                    "repair: ignoring {} in the blobs folder: not an engine file",
                    blob_path.display(),
                );
                continue;
            }
        };
        candidates.push((blob_id, blob_path, file_name));
    }
    // Fresh-id allocator for salvaged replacements. Taken over ALL candidates
    // before any is processed, so an id handed out here can never collide with
    // a file this scan has not reached yet — AND over every id the recovered
    // tables still REFERENCE: an SST may point at a blob file that no longer
    // exists on disk, and allocating that missing id to an unrelated
    // replacement would let the later dependency check find the id present
    // and keep the SST, whose handles then resolve against the wrong file's
    // records. Repair runs single-threaded on a tree nobody else has open,
    // so no other allocator competes.
    //
    // `None` once the space is exhausted, and the failure is raised only where
    // an id is actually needed — a healthy tree holding a `u64::MAX` blob is
    // still perfectly repairable. Wrapping instead would hand out an id an
    // existing file already holds, and publishing the replacement onto that
    // name would destroy an unrelated original BEFORE the manifest commit, so
    // the loss would not even surface as a failed repair.
    let mut next_blob_id: Option<crate::vlog::BlobFileId> = candidates
        .iter()
        .map(|(id, _, _)| *id)
        .chain(referenced.iter().copied())
        .max()
        .map_or(Some(0), |max| max.checked_add(1));
    candidates.sort_by(|a, b| {
        a.0.cmp(&b.0)
            .then_with(|| {
                (a.1 != blobs_folder.join(a.0.to_string()))
                    .cmp(&(b.1 != blobs_folder.join(b.0.to_string())))
            })
            .then_with(|| a.2.cmp(&b.2))
    });

    // The path recovered for each id, for the duplicate-vs-alias decision.
    let mut kept_paths: crate::HashMap<crate::vlog::BlobFileId, PathBuf> =
        crate::HashMap::default();

    // Whether a file opens on its OWN bytes, for the duplicate verdict below.
    // A duplicate contributes nothing to the rebuild, but the report's two
    // channels make different claims about it, and health inferred from the
    // RETAINED copy would file a rotting duplicate as healthy. A FULLY PUNCHED
    // file is not damaged — its relocation completed — so it counts as clean.
    // The live-data frontier of one file: the manifest's COMMITTED value when
    // it survived — exact, free, and independent of whether the backend can
    // attribute zeros to a hole — else derived from the punch geometry. Single
    // source for the recovery walk and the duplicate verdict alike.
    let resolve_frontier = |blob_id, blob_path: &std::path::Path| -> crate::Result<BlobFrontier> {
        match committed_frontiers.and_then(|m| m.get(&blob_id).copied()) {
            Some(committed) => Ok(BlobFrontier::Punched(committed)),
            None => derive_blob_frontier(&config.fs, blob_path, blob_id),
        }
    };

    let opens_cleanly = |blob_id, blob_path: &std::path::Path| -> crate::Result<()> {
        let frontier = match resolve_frontier(blob_id, blob_path)? {
            BlobFrontier::Whole => 0,
            BlobFrontier::Punched(f) => f,
            BlobFrontier::FullyConsumed => return Ok(()),
        };
        // The digest is recomputed from THESE bytes, so it can only prove the
        // file is self-consistent — never that it is undamaged. Recovery then
        // parses the metadata section alone, so a rotted value frame sails
        // through both. Walk the frames, exactly as a retained file's does.
        let checksum = crate::Checksum::from_raw(compute_table_checksum_from(
            &*config.fs,
            blob_path,
            frontier,
        )?);
        let handle = crate::vlog::recover_blob_file_from(
            blob_path,
            blob_id,
            checksum,
            0,
            &config.fs,
            frontier,
            #[cfg(zstd_any)]
            &config.current_zstd_dictionaries(),
        )?;
        // Without the dictionary no copy can be verified. For an id a recovered
        // table references that stops the repair, as it does in the recovery
        // below, since the choice decides which copy survives. For one nothing
        // references no copy survives either way: it simply does not verify,
        // and every copy takes the unreferenced path to the sweep.
        #[cfg(zstd_any)]
        let dict = match blob_file_dictionary(config, handle.compression()) {
            Ok(dict) => dict,
            Err(e) if referenced.contains(&blob_id) => return Err(e),
            Err(_) => {
                return Err(crate::Error::InvalidHeader(
                    "blob file names a dictionary the tree does not hold",
                ));
            }
        };
        match validate_blob_frames(
            config,
            blob_path,
            blob_id,
            frontier,
            &handle,
            #[cfg(zstd_any)]
            dict.as_ref(),
        )? {
            Some(_) => Ok(()),
            None => Err(crate::Error::InvalidHeader(
                "blob frame validation failed on this copy",
            )),
        }
    };

    // Sightings of one id are contiguous (the sort keys on id first) and the
    // CANONICAL spelling leads each group. That order is wrong when the leader
    // is damaged: the loop below would salvage it into a fresh id, record that
    // lossy replacement as the incumbent, and then discard a later INTACT copy
    // as a redundant duplicate — committing a partial rewrite of data that was
    // available in full. Promote a cleanly-opening sighting to the front of its
    // group so salvage runs only when no copy of the id is whole.
    let candidates = {
        let mut ordered: Vec<(crate::vlog::BlobFileId, PathBuf, String)> =
            Vec::with_capacity(candidates.len());
        let mut rest = candidates.into_iter().peekable();
        while let Some(first) = rest.next() {
            let blob_id = first.0;
            let mut group = vec![first];
            while rest.peek().is_some_and(|c| c.0 == blob_id) {
                if let Some(c) = rest.next() {
                    group.push(c);
                }
            }
            if group.len() > 1 {
                // Opening cleanly proves a copy is SELF-consistent, never that
                // it is the one the manifest named: two copies of an id can
                // both be whole and hold different generations, and adopting
                // the wrong one re-stamps a fresh checksum over stale bytes
                // that existing SST handles then resolve against. When the
                // committed digest survived, it decides.
                let committed = committed_checksums.and_then(|m| m.get(&blob_id).copied());
                let mut authoritative = None;
                let mut whole = None;
                for (k, candidate) in group.iter().enumerate() {
                    if let Some(expected) = committed
                        && authoritative.is_none()
                    {
                        let frontier = match resolve_frontier(blob_id, &candidate.1)? {
                            BlobFrontier::Whole => Some(0),
                            BlobFrontier::Punched(f) => Some(f),
                            // Nothing live is left to digest.
                            BlobFrontier::FullyConsumed => None,
                        };
                        if let Some(frontier) = frontier {
                            match compute_table_checksum_from(&*config.fs, &candidate.1, frontier) {
                                Ok(d) if crate::Checksum::from_raw(d) == expected => {
                                    authoritative = Some(k);
                                    continue;
                                }
                                // A fault in the ENVIRONMENT decides nothing
                                // and the decision here costs the loser its
                                // file; a digest that simply differs, or a
                                // read that fails on the bytes, both mean
                                // "not the manifest's copy".
                                Err(e) if is_environmental(&e) => return Err(e),
                                Ok(_) | Err(_) => {}
                            }
                        }
                    }
                    if whole.is_none() {
                        match opens_cleanly(blob_id, &candidate.1) {
                            Ok(()) => whole = Some(k),
                            // Not evidence about these bytes, and the choice it
                            // would steer decides which copy survives.
                            Err(e) if is_environmental(&e) => return Err(e),
                            Err(_) => {}
                        }
                    }
                }
                // The manifest's own copy first; without one (no clean
                // manifest, or none of the copies reproduces its digest) an
                // intact copy still beats a salvage of a damaged leader.
                if let Some(k) = authoritative.or(whole) {
                    group.swap(0, k);
                }
            }
            ordered.extend(group);
        }
        ordered
    };

    for (blob_id, blob_path, _file_name) in candidates {
        if let Some(kept) = kept_paths.get(&blob_id) {
            // A second directory entry for an already-recovered id. An ALIAS
            // (symlink / case-folded spelling of the SAME physical file) is
            // skipped silently. A DISTINCT physical file must go: the manifest
            // records one checksum per id, and a stale duplicate left in
            // `blobs/` would race the kept file for reads on the next open
            // (directory iteration order picks the physical file).
            if same_physical_file(&*config.fs, kept, &*config.fs, &blob_path)? {
                continue;
            }
            // Verify THIS copy before choosing the report channel: an
            // exclusion claims the file opened and verified, and calling an
            // unchecked duplicate healthy hides the corruption signal.
            match opens_cleanly(blob_id, &blob_path) {
                Ok(()) => {
                    let reason = format!("duplicate of blob file id {blob_id}");
                    discard.push((blob_path.clone(), reason.clone()));
                    excluded.push((blob_path, reason));
                }
                // A fault in the ENVIRONMENT is not evidence about these bytes.
                Err(e) if is_environmental(&e) => return Err(e),
                Err(e) => {
                    let reason = format!(
                        "damaged duplicate of blob file id {blob_id} (kept copy is intact): {e}"
                    );
                    discard.push((blob_path.clone(), reason.clone()));
                    unreadable.push((blob_path, reason));
                }
            }
            continue;
        }

        // Pre-commit file boundary: safe to abort. The caller's
        // `PublishedBlobReplacements` guard unwinds the replacements this run
        // already published under fresh ids — unlike the read-only table
        // scan, a successful blob salvage renames its copy to a normal
        // numeric name before the commit, and leaving those behind would make
        // the retry re-create each one beside its orphan (under tight disk
        // space exactly the sequence that fails).
        check_cancel(config)?;
        if let Some(p) = &config.recovery_progress {
            p.blob_file_discovered();
        }

        // A tight-space-punched blob records its live-data frontier only in
        // the manifest; with the manifest lost, re-derive it from the punch
        // geometry so the rebuilt manifest restores the restriction (the
        // snapshot encoder persists it from the recovered `live_data_start`).
        // Rebuilding with frontier 0 would instead leave a later relocation
        // scan starting inside the punched (zeroed) prefix. An unpunched file
        // short-circuits to 0 on its first (non-zero) data byte.
        // Persistent per-file failure: the rebuilt manifest omits this file, so
        // it is recorded for removal after the commit and reported unreadable.
        // The file must not be both omitted and left in place — that is a tree
        // whose next open has an orphan to sweep and may fail doing it.
        let discard_unreadable =
            |blob_path: PathBuf,
             e: &crate::Error,
             unreadable: &mut UnreadableFiles,
             discard: &mut Vec<(PathBuf, String)>| {
                discard.push((blob_path.clone(), e.to_string()));
                unreadable.push((blob_path, e.to_string()));
            };

        // The committed frontier, when the manifest loaded cleanly, is exact
        // and free: no geometry walk, and no dependence on the backend being
        // able to attribute zeros to a hole (which, unanswered, reports a
        // punched file WHOLE — publishing handles that dereference the hole).
        // Everything downstream — identity, frame validation, the handle
        // rewrite — runs on it exactly as on a derived one.
        let frontier = match resolve_frontier(blob_id, &blob_path) {
            Ok(BlobFrontier::Whole) => 0,
            Ok(BlobFrontier::Punched(f)) => f,
            Ok(BlobFrontier::FullyConsumed) => {
                // Every frame is punched away: the relocation that consumed
                // this file completed, only its removal lagged the crash.
                // QUEUE that lagged drop for after the commit instead of
                // finishing it here: the pre-commit scan is read-only on
                // purpose — an abort before the commit (a cancellation, a
                // transient failure on a later file) must leave the directory
                // exactly as found, and the OLD manifest of an explicitly
                // invoked repair over an openable tree may still name this
                // file. Publishing an empty-suffix handle instead is not an
                // option either: whole-file metadata over zero live frames is
                // a file blob GC's stale-byte arithmetic can never retire
                // (its frames are already gone, so the stale count never
                // reaches the recorded totals). No live data is discarded:
                // the walk proved the whole data section reads as zeros. The
                // post-commit removal failing fails the repair, exactly as
                // the immediate removal used to.
                log::info!(
                    "blob file {blob_id} at {}: its punch consumed every frame — \
                     queueing the relocation's lagged file drop for after the commit",
                    blob_path.display(),
                );
                discard.push((
                    blob_path,
                    "fully punched blob file: a completed relocation's lagged drop".to_string(),
                ));
                continue;
            }
            // A read failure in the ENVIRONMENT is retryable: recording the blob
            // unreadable commits a manifest without the still-in-place file,
            // which the next open's orphan sweep then DELETES — permanent value
            // loss from a fixable failure. Propagate so a retry re-reads it,
            // mirroring the table-recovery path.
            Err(e) if is_environmental(&e) => return Err(e),
            Err(e) => {
                discard_unreadable(blob_path, &e, &mut unreadable, &mut discard);
                continue;
            }
        };

        // IDENTITY before content: the id under which this file will be
        // published comes from its FILENAME, but the metadata records the id
        // it was written as. A mismatch means a renamed or swapped file — not
        // damaged content — and publishing it under the filename's id would
        // resolve existing SST handles into foreign frames (a read fails on
        // the key cross-check, or, when the foreign frame happens to hold the
        // same key, silently serves another generation's value). Salvage is
        // the WRONG remedy here (it would re-emit the foreign records under
        // the filename's id, laundering the swap); set the file aside.
        // Placeholder-checksum open (stored, never verified). Same
        // transient/persistent split as every other per-file step: an
        // unreadable meta section sets THIS file aside, never the repair.
        // Read ONCE and kept: the identity check needs the stored id and the
        // frame validation below needs the compression descriptor, and both
        // live in this same meta section.
        let handle = match crate::vlog::recover_blob_file(
            &blob_path,
            blob_id,
            crate::Checksum::from_raw(0),
            0,
            &config.fs,
            #[cfg(zstd_any)]
            &config.current_zstd_dictionaries(),
        ) {
            Ok(handle) => handle,
            Err(e) if is_environmental(&e) => return Err(e),
            Err(e) => {
                discard_unreadable(blob_path, &e, &mut unreadable, &mut discard);
                continue;
            }
        };
        let stored = handle.meta().id;
        if stored != blob_id {
            let e = crate::Error::InvalidHeader(
                "blob file's stored metadata id disagrees with its file name",
            );
            log::warn!(
                "blob file at {}: metadata records id {stored}, file name says \
                 {blob_id} — a renamed or swapped file; setting it aside",
                blob_path.display(),
            );
            discard_unreadable(blob_path, &e, &mut unreadable, &mut discard);
            continue;
        }

        // Validate the live frame range BEFORE recording a digest: hashing
        // damaged frames would restamp (launder) the corruption past every
        // later integrity check while reads of the affected values still
        // fail. An invalid file is SALVAGED instead of blessed or thrown away
        // whole: its surviving records are re-emitted into a compacted
        // replacement and the offset relocation is recorded so the referencing
        // SSTs are rewritten onto the new offsets.
        //
        // The replacement is written under a FRESH blob file id, so it never
        // displaces the damaged original: the original keeps its id and its
        // bytes, and only the manifest commit makes the replacement live. A
        // crash at any point before that commit therefore leaves the tree
        // exactly as it was found, and the retry re-derives everything from
        // the untouched original — no relocation record has to survive the
        // crash, because nothing was relocated in place. The original is
        // removed only AFTER the commit: removing it earlier would make a
        // crashed attempt leave SSTs referencing a blob id that no longer
        // exists, and the retry would record those tables unrecoverable.
        //
        // The dictionary the file's descriptor names comes first. A missing one
        // is not damage: the frames would all fail to decode and the file would
        // be salvaged into nothing. One a recovered table references stops the
        // repair naming the id, before anything is committed, so the operator
        // supplies the dictionary and re-runs. One nothing references holds no
        // value anyone can reach, and goes to the sweep like any other.
        #[cfg(zstd_any)]
        let dict = match blob_file_dictionary(config, handle.compression()) {
            Ok(dict) => dict,
            Err(e) if referenced.contains(&blob_id) => return Err(e),
            Err(e) => {
                log::debug!(
                    "blob file {blob_id} names a dictionary the tree does not hold \
                     ({e}) and no recovered table references it; leaving it for the \
                     post-commit sweep"
                );
                discard.push((
                    blob_path,
                    "names a dictionary the tree does not hold, and no recovered table \
                     references it"
                        .to_string(),
                ));
                continue;
            }
        };
        let Some(live) = validate_blob_frames(
            config,
            &blob_path,
            blob_id,
            frontier,
            &handle,
            #[cfg(zstd_any)]
            dict.as_ref(),
        )?
        else {
            // An INVALID blob no recovered table references never reaches the
            // manifest: salvaging it would allocate a fresh id and burn disk
            // space on a replacement the reference filter deletes anyway —
            // and under tight space that pointless salvage can abort the
            // whole repair with `StorageFull`. Leave it for the post-commit
            // sweep instead; nothing reachable is lost.
            if !referenced.contains(&blob_id) {
                log::debug!(
                    "blob file {blob_id} is invalid and referenced by no recovered \
                     table; skipping its salvage and leaving it for the post-commit \
                     sweep"
                );
                discard.push((
                    blob_path,
                    "invalid and referenced by no recovered table; salvage skipped".to_string(),
                ));
                continue;
            }
            let Some(new_id) = next_blob_id else {
                log::error!(
                    "blob file {blob_id} needs salvaging but the blob id space is \
                     exhausted: there is no fresh name to publish a replacement \
                     under, and reusing one would destroy an existing file"
                );
                return Err(crate::Error::Unrecoverable);
            };
            next_blob_id = new_id.checked_add(1);
            let temp = blobs_folder.join(format!("{new_id}.salvage-tmp"));
            remove_temp(config, &temp)?;
            let salvage = (|| -> crate::Result<Option<(crate::vlog::BlobFile, crate::salvage::BlobSalvageReport)>> {
                let report = crate::salvage::salvage_blob_file_on(
                    &blob_path,
                    temp.clone(),
                    &config.fs,
                    // The OUTPUT's id: the replacement is stamped as the fresh
                    // file it is, so its metadata id matches the name it will
                    // be published under.
                    new_id,
                    &config.comparator,
                    frontier,
                    // Repair has the tree's dictionary context, so a
                    // dictionary-compressed blob salvages its intact frames
                    // instead of being set aside whole. The id comes from the
                    // SOURCE's own descriptor, so a file written under an
                    // earlier dictionary resolves to that one.
                    #[cfg(zstd_any)]
                    dict.as_ref(),
                    // Stamped on the tree's clock, as the blob files it writes are.
                    config.clock.as_ref(),
                )?;
                let Some(salvaged_path) = report.salvaged_path.clone() else {
                    return Ok(None);
                };
                let checksum = crate::Checksum::from_raw(compute_table_checksum(
                    &*config.fs,
                    &salvaged_path,
                )?);
                let held = held_descriptors(config);
                let bf = crate::vlog::recover_blob_file_cached(
                    &salvaged_path,
                    new_id,
                    checksum,
                    held.as_ref().map_or(0, |(_, tree_id)| *tree_id),
                    &config.fs,
                    0,
                    #[cfg(zstd_any)]
                    &config.current_zstd_dictionaries(),
                    held.as_ref().map(|(cache, _)| cache),
                )?;
                Ok(Some((bf, report)))
            })();
            let (bf, report) = match salvage {
                Ok(Some(pair)) => pair,
                // Nothing recoverable, or a persistent failure: report it and
                // queue its removal — exactly like any other unreadable blob.
                Ok(None) => {
                    remove_temp(config, &temp)?;
                    let e = crate::Error::InvalidHeader(
                        "blob value frames failed validation and no record was recoverable",
                    );
                    discard_unreadable(blob_path, &e, &mut unreadable, &mut discard);
                    continue;
                }
                Err(e) if is_environmental(&e) => {
                    // Nothing was published; dropping the temp restores the
                    // pre-repair state exactly, so the retry re-salvages and
                    // re-derives the remap. Best-effort is enough on this
                    // ERROR path: no manifest is committed, and the retry's
                    // candidate sweep sets a stuck temp aside.
                    let _ = config.fs.remove_file(&temp);
                    return Err(e);
                }
                Err(e) => {
                    remove_temp(config, &temp)?;
                    discard_unreadable(blob_path, &e, &mut unreadable, &mut discard);
                    continue;
                }
            };

            // Take the fresh id's own (free) name: nothing is displaced, so
            // there is nothing to unwind and no window in which a retry could
            // mistake an unverified file for the live blob. Until the
            // manifest names it, the replacement is an unreferenced file that
            // the next open's orphan sweep would remove.
            let new_path = blobs_folder.join(new_id.to_string());
            if let Err(e) = config.fs.rename(&temp, &new_path) {
                // Best-effort on this ERROR path: no manifest is committed,
                // so a stuck temp is swept by the retry's candidate scan.
                let _ = config.fs.remove_file(&temp);
                return Err(e.into());
            }
            // Under the caller's guard from this moment: any pre-commit exit
            // removes the published file.
            published.publish(new_id);
            config
                .fs
                .sync_directory_with(&blobs_folder, config.sync_mode)?;

            blob_files.push(bf);
            rewrites.insert(
                blob_id,
                crate::salvage::BlobFileRewrite::Remap {
                    new_id,
                    offsets: report.offset_remap.iter().copied().collect(),
                },
            );
            // The damaged original is left in place for now and set aside once
            // the manifest is committed: it is still what the not-yet-rewritten
            // SSTs reference, so removing it earlier would strand them if this
            // repair failed before committing.
            // Report only what the walk can actually account for. A
            // structural failure DESYNCHRONIZES the record stream: the walk
            // stops there and surrenders everything after it, so
            // `records_total` counts the prefix it managed to frame, not the
            // source's true population. Presenting that as "X of Y" would
            // claim a near-complete recovery while an unknown — possibly
            // enormous — tail went with it. Only when every loss was an
            // individually re-synced record is the total trustworthy.
            let surrendered_tail = report
                .dropped
                .iter()
                .any(|d| matches!(d.reason, crate::salvage::BlobDropReason::Corrupt(_)));
            let note = if surrendered_tail {
                format!(
                    "{} records salvaged into blob file {new_id}; the record \
                     stream then desynchronized and the remainder of the file \
                     was surrendered, so the number of records lost with it is \
                     not knowable",
                    report.records_salvaged,
                )
            } else {
                format!(
                    "{} of {} records salvaged into blob file {new_id} \
                     (the rest failed their checksums)",
                    report.records_salvaged, report.records_total,
                )
            };
            stale_originals.push((blob_path.clone(), note));
            kept_paths.insert(blob_id, blob_path);
            continue;
        };

        // The digest covers the live region only — `[frontier, end)` for a
        // punched file, the whole file for `frontier == 0` — matching what
        // `reopen_restricted` records and what integrity checks recompute.
        let checksum = match compute_table_checksum_from(&*config.fs, &blob_path, frontier) {
            Ok(c) => crate::Checksum::from_raw(c),
            // Same transient/persistent split as the frontier probe above.
            Err(e) if is_environmental(&e) => return Err(e),
            Err(e) => {
                discard_unreadable(blob_path, &e, &mut unreadable, &mut discard);
                continue;
            }
        };

        // Held until the rebuilt manifest is published, as a recovered table
        // is, so its descriptor goes through the repair's cache (see
        // `held_recover_params`).
        let held = held_descriptors(config);
        match crate::vlog::recover_blob_file_cached(
            &blob_path,
            blob_id,
            checksum,
            held.as_ref().map_or(0, |(_, tree_id)| *tree_id),
            &config.fs,
            frontier,
            #[cfg(zstd_any)]
            &config.current_zstd_dictionaries(),
            held.as_ref().map(|(cache, _)| cache),
        ) {
            Ok(bf) => {
                if frontier > 0 {
                    // A punched-but-intact file: a stale handle below its
                    // frontier (a pre-relocation SST left behind by a crash)
                    // must be dropped by the table-rewrite stage.
                    rewrites.insert(
                        blob_id,
                        crate::salvage::BlobFileRewrite::DropBelow(frontier),
                    );
                    // Seed the punched prefix's garbage: the recovered handle
                    // keeps its WHOLE-FILE metadata totals, while the prefix's
                    // frames can never be observed by a future compaction. An
                    // empty fragmentation map would pin the recorded stale
                    // bytes below the totals forever, so `is_dead` could
                    // never retire the file even after every suffix handle is
                    // gone. The differences cannot underflow: validation just
                    // proved the metadata totals are at least the suffix's.
                    let meta = bf.meta();
                    let prefix_items = meta.item_count - live.items;
                    let prefix_len =
                        usize::try_from(prefix_items).map_err(|_| crate::Error::Unrecoverable)?;
                    frag.insert(
                        blob_id,
                        crate::blob_tree::FragmentationEntry::new(
                            prefix_len,
                            meta.total_uncompressed_bytes - live.uncompressed_bytes,
                            meta.total_compressed_bytes - live.compressed_bytes,
                        ),
                    );
                }
                kept_paths.insert(blob_id, blob_path);
                blob_files.push(bf);
            }
            // Same transient/persistent split as the checksum read above.
            Err(e) if is_environmental(&e) => return Err(e),
            Err(e) => {
                discard_unreadable(blob_path, &e, &mut unreadable, &mut discard);
            }
        }
    }

    Ok(BlobRecovery {
        files: blob_files,
        unreadable,
        rewrites,
        frag,
        stale: stale_originals,
        discard,
        excluded,
    })
}

impl Config {
    /// Rebuilds the `MANIFEST` for the tree at this config's path from the SST
    /// files present on disk, then returns a [`RepairReport`].
    ///
    /// Use this only when a tree fails to open because its manifest is missing
    /// or corrupt but the SST files are intact. After a successful repair the
    /// tree opens normally; all recovered data is at L0 and a background
    /// compaction restructures it into proper levels (expect elevated I/O for a
    /// period proportional to the data size).
    ///
    /// Version edits the lost manifest carried are gone with it (the report's
    /// standing warning says so): in particular, a committed compaction whose
    /// filter removed EVERY record left no output to carry lineage, so its
    /// still-lingering inputs are republished as one consistent
    /// pre-compaction history — the filter's removals transiently reappear
    /// until the next compaction re-applies the standing policy. No operand
    /// is doubled by that window.
    ///
    /// # The report
    ///
    /// The rebuilt tree keeps the returned report until
    /// [`AbstractTree::acknowledge_repair_report`](crate::AbstractTree::acknowledge_repair_report)
    /// takes its id; until then [`Config::open_or_repair`] hands it out
    /// again, so a process that ends before the report was acted on loses
    /// nothing.
    ///
    /// # Exclusive access
    ///
    /// Repair writes and installs a fresh snapshot and removes the stale
    /// `edits-*` logs in place, so it requires exclusive access to the tree
    /// directory. It acquires the same cross-process directory lock as
    /// [`Config::open`] for the duration of the call: if another live instance
    /// holds the directory (open or repairing), this fails fast with
    /// [`crate::Error::Locked`] instead of corrupting that instance's manifest
    /// state. The lock can be disabled via
    /// [`Config::with_directory_lock`](crate::Config::with_directory_lock) for
    /// embedders enforcing exclusivity at a higher layer.
    ///
    /// # Errors
    ///
    /// Returns [`crate::Error::FeatureUnsupported`] for KV-separated (blob)
    /// trees, and propagates any I/O error from scanning the directory or
    /// writing the new manifest. Individual unreadable SSTs do not fail the
    /// repair; they are reported in [`RepairReport::unreadable_files`].
    ///
    /// # Examples
    ///
    /// ```
    /// use lsm_tree::{AbstractTree, Config, SequenceNumberCounter};
    ///
    /// let folder = tempfile::tempdir()?;
    /// let config = Config::new(
    ///     &folder,
    ///     SequenceNumberCounter::default(),
    ///     SequenceNumberCounter::default(),
    /// );
    /// {
    ///     let tree = config.clone().open()?;
    ///     tree.insert("a", "1", 0);
    ///     tree.flush_active_memtable(0)?;
    /// }
    ///
    /// let report = config.repair()?;
    /// println!("recovered {} tables, {} unreadable", report.recovered, report.unreadable);
    ///
    /// // `repair` borrows, so the same config opens the rebuilt tree. Once the
    /// // report's replay obligation is met, the tree stops owing it.
    /// let tree = config.open()?;
    /// assert!(tree.get("a", 1)?.is_some());
    /// assert!(tree.acknowledge_repair_report(report.id)?);
    /// # Ok::<(), lsm_tree::Error>(())
    /// ```
    pub fn repair(&self) -> crate::Result<RepairReport> {
        repair_tree(self, false, false)
    }

    /// Like [`repair`](Self::repair), but when an SST fails whole-file recovery
    /// (`salvage = true`) it is block-salvaged instead of being left out: a fresh
    /// SST holding its recoverable blocks is built beside it, referenced by the
    /// rebuilt manifest, and swapped onto its name once that manifest is durable.
    ///
    /// A salvaged table may be missing the key ranges of its corrupt blocks
    /// (reported per file via [`RepairReport::salvaged`]); use this only as a
    /// last resort when losing the whole SST is worse than losing some keys.
    /// SSTs whose metadata, index, or SFA trailer is itself unreadable still
    /// cannot be salvaged and are reported unreadable.
    ///
    /// # Errors
    ///
    /// Returns an error if the directory lock cannot be taken or the rebuilt
    /// manifest cannot be persisted; per-file recovery / salvage failures are
    /// reported in the [`RepairReport`], not returned.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use lsm_tree::{Config, SequenceNumberCounter};
    ///
    /// let config = Config::new(
    ///     "/var/lib/mydb",
    ///     SequenceNumberCounter::default(),
    ///     SequenceNumberCounter::default(),
    /// );
    /// let report = config.repair_with_salvage(true)?;
    /// println!(
    ///     "recovered {} table(s), {} of them by salvage",
    ///     report.recovered, report.salvaged,
    /// );
    /// # Ok::<(), lsm_tree::Error>(())
    /// ```
    pub fn repair_with_salvage(&self, salvage: bool) -> crate::Result<RepairReport> {
        repair_tree(self, salvage, false)
    }

    /// Like [`repair_with_salvage`](Self::repair_with_salvage), but with an
    /// explicit `allow_resurrection` policy. When `false` (the default the other
    /// entry points use), recovery drops data whose visibility became ambiguous
    /// after a lost restriction bound or a lost / forged delete mask; when
    /// `true`, it keeps that data, accepting that superseded or deleted rows may
    /// reappear. Either setting yields a valid, openable tree; the flag is the
    /// ONLY recovery decision an operator makes.
    ///
    /// The flag governs AMBIGUOUS VISIBILITY, not LOST BYTES. When a table (or
    /// one of its blocks) cannot be read at all, what it said about its keys is
    /// gone with it, and older versions of those keys survive in other tables:
    /// a value it had overwritten, or a key its tombstone had deleted, becomes
    /// visible again. Neither setting changes that, because nothing left on
    /// disk distinguishes "this key was deleted" from "this key was simply
    /// never rewritten". Deleting the lost range instead would destroy intact
    /// data — a flushed table's key range routinely spans most of the keyspace
    /// while its sequence numbers sit above every older level, so one corrupt
    /// block would erase most of the tree, irreversibly, where the stale read
    /// leaves every byte in place. The lost ranges are therefore REPORTED
    /// rather than acted on; see
    /// [`RepairReport::lost_coverage`](crate::RepairReport::lost_coverage).
    ///
    /// # Errors
    ///
    /// Returns an error if the tables directory cannot be scanned or a transient
    /// I/O fault interrupts recovery (retryable), or if the rebuilt manifest
    /// cannot be durably installed. See
    /// [`repair_with_salvage`](Self::repair_with_salvage).
    pub fn repair_with_resurrection(
        &self,
        salvage: bool,
        allow_resurrection: bool,
    ) -> crate::Result<RepairReport> {
        repair_tree(self, salvage, allow_resurrection)
    }

    /// Opens the tree, and when the open fails STRUCTURALLY, repairs per
    /// `policy` and opens again — the one-call recovery entry point for a
    /// caller that pairs the engine with an external write-ahead log.
    ///
    /// `Ok((tree, None))` is a healthy open. `Ok((tree, Some(report)))` is an
    /// open that succeeded only after a repair; the caller MUST then consult
    /// [`RepairReport::wal_replay_scope`] before its WAL replay, since the
    /// repair may have regressed persisted state below the log's trim
    /// watermark (see `docs/external-wal.md` § Replay after repair).
    ///
    /// Only failures that positively identify repairable on-disk structural
    /// damage engage the repair (see `is_repairable_structural`). Everything
    /// else is returned as-is: a TRANSIENT I/O failure (a repair would
    /// rebuild the manifest around files a healthy retry could still read), a
    /// held directory lock ([`Error::Locked`](crate::Error::Locked) — the
    /// repair would contend on the same lock), an UNSUPPORTED format version
    /// ([`Error::InvalidVersion`](crate::Error::InvalidVersion) — the store
    /// needs offline conversion or a matching binary, not a rebuild in the
    /// current format that would reject every table), a ROUTED tree's
    /// [`Error::Unrecoverable`](crate::Error::Unrecoverable) (route
    /// provenance is not persisted, so a missing routed table is
    /// indistinguishable from a route path change or an unmounted tier — a
    /// rebuild would omit every SST still on the old route), and every
    /// CONFIGURATION mismatch (wrong comparator, level route, dictionary, or
    /// encryption key) — repairing under a wrong configuration would rebuild,
    /// and could drop, perfectly healthy data.
    ///
    /// # Errors
    ///
    /// Propagates a transient-I/O or lock failure from the open and any
    /// repair failure. A failure of the POST-repair open surfaces as
    /// [`Error::RepairedButUnopened`](crate::Error::RepairedButUnopened),
    /// carrying the completed [`RepairReport`]: consume it from the error.
    ///
    /// The rebuilt snapshot holds its report until the caller acknowledges it
    /// with [`AbstractTree::acknowledge_repair_report`](crate::AbstractTree::acknowledge_repair_report),
    /// and every call before that answers `Some(report)` again, a healthy
    /// open included. A report is handed out at least once whatever ends the
    /// process between the hand-out and its use; one handed out again carries
    /// the same [`RepairReport::id`], so a caller that records the ids it
    /// applied skips it. Acknowledge only once the report's replay obligation
    /// is met.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use lsm_tree::{Config, RepairPolicy, SequenceNumberCounter};
    ///
    /// let config = Config::new(
    ///     "/var/lib/mydb",
    ///     SequenceNumberCounter::default(),
    ///     SequenceNumberCounter::default(),
    /// );
    /// let (tree, repaired) = config.open_or_repair(RepairPolicy::default().salvage(true))?;
    /// if let Some(report) = repaired {
    ///     println!("repaired; WAL must replay {:?}", report.wal_replay_scope());
    /// }
    /// # Ok::<_, lsm_tree::Error>(())
    /// ```
    pub fn open_or_repair(
        self,
        policy: RepairPolicy,
    ) -> crate::Result<(crate::AnyTree, Option<RepairReport>)> {
        let retry = self.clone();
        let path = self.path.clone();
        match self.open() {
            // A report not yet acknowledged rides in the snapshot this open
            // installed again.
            Ok(tree) => {
                let owed = owed_report(&tree, &path);
                Ok((tree, owed))
            }
            Err(e) if is_repairable_structural(&e) => {
                // The rebuilt snapshot owes the report, so a crash before it
                // reaches the caller leaves it for the next call.
                let report = repair_tree(&retry, policy.salvage, policy.allow_resurrection)?;
                match retry.open() {
                    // The snapshot owes this same report (one id), but its
                    // copy was written before the cleanup after the commit;
                    // this call's own report also says what that cleanup did.
                    Ok(tree) => Ok((tree, Some(report))),
                    // The repair committed: the report leaves with the error,
                    // and the snapshot that owes it hands it out again, with
                    // the same id, to the call that opens it.
                    Err(cause) => Err(crate::Error::RepairedButUnopened {
                        report: alloc::boxed::Box::new(report),
                        cause: alloc::boxed::Box::new(cause),
                    }),
                }
            }
            Err(e) => Err(e),
        }
    }
}

/// The [`RepairReport::id`] of the report encoded as `bytes`.
pub(crate) fn owed_report_id(bytes: &[u8]) -> u128 {
    report_marker::owed_id(bytes)
}

/// The report `tree` owes until its caller acknowledges it.
fn owed_report(tree: &crate::AnyTree, path: &std::path::Path) -> Option<RepairReport> {
    let index = match tree {
        crate::AnyTree::Standard(tree) => tree,
        crate::AnyTree::Blob(tree) => &tree.index,
    };
    index
        .owed_repair_report()
        .map(|owed| report_marker::decode_owed(&owed, path))
}

/// Whether an `open` failure positively identifies repairable ON-DISK
/// structural damage — the only class [`Config::open_or_repair`] may answer
/// with a repair.
///
/// Everything else propagates: transient I/O (a retry could still read the
/// files a repair would rebuild around), a held directory lock (the repair
/// would contend on the same lock), an UNSUPPORTED format version (an older
/// or future database has no live decoder here — it needs offline conversion
/// or a matching binary, and repair refuses it at the same format gate), and
/// every CONFIGURATION
/// mismatch — a wrong comparator, level route, zstd dictionary, a
/// standard-vs-blob tree-type mismatch ([`crate::Error::TreeTypeMismatch`]),
/// or an encryption key surfacing as a decrypt failure. Those are reversible by
/// fixing the call, while a repair under the wrong configuration rebuilds —
/// and can drop — perfectly healthy data (e.g. re-ordering SSTs under a
/// mistyped comparator, or salvage discarding every block it cannot decrypt
/// with the wrong key). Fail closed: an unlisted (including future) error is
/// NOT repaired.
fn is_repairable_structural(e: &crate::Error) -> bool {
    use crate::Error;
    match e {
        // Data-shape I/O kinds are structural: a missing file (the manifest
        // itself, or a file it names), malformed bytes (the manifest-loss
        // open surfaces "current missing but artifacts present" as
        // InvalidData), or a truncated read. Any other I/O kind is
        // environment, and transient kinds are retried by the caller.
        Error::Io(io) => matches!(
            io.kind(),
            crate::io::ErrorKind::NotFound
                | crate::io::ErrorKind::InvalidData
                | crate::io::ErrorKind::UnexpectedEof
        ),
        // `Unrecoverable` is deliberately ABSENT: it is the catch-all for
        // "cannot classify", the opposite of positive evidence of a structural
        // defect. Repairing on it rebuilds the manifest from whatever the
        // current configuration happens to scan, and since route provenance
        // is not persisted, a routed store reopened with its `level_routes`
        // omitted presents exactly that way, so the rebuild would drop every
        // SST on the unscanned tiers. An operator who has verified the
        // configuration can still invoke `repair` explicitly.
        Error::Decompress(_)
        | Error::Excised { .. }
        | Error::ChecksumMismatch { .. }
        | Error::HeaderCrcMismatch { .. }
        | Error::InvalidTag(_)
        | Error::InvalidTrailer
        | Error::InvalidHeader(_)
        | Error::DecompressedSizeTooLarge { .. }
        | Error::ManifestFrameChecksumMismatch { .. }
        | Error::ManifestFooterInvalid(_)
        | Error::ManifestSectionInvalid(_)
        | Error::TornManifestEditLog { .. }
        | Error::RangeTombstoneDecode { .. }
        | Error::PageEccUnrecoverable { .. }
        // A stored dictionary disagreeing with its own name is the bytes: the
        // name is their hash. The one configuration that reads the same way,
        // an encrypted file opened without its key, never gets here, because
        // `Tree::open` verifies the manifest under the configured provider
        // before it reads a dictionary. The repair sets the file aside, or
        // stops on a table that still needs it.
        | Error::ZstdDictCorrupt { .. } => true,
        _ => false,
    }
}

/// Which repair capabilities [`Config::open_or_repair`] may engage.
///
/// Consulted when the open fails structurally. The default engages none of
/// them — the plain [`Config::repair`], which drops what it cannot recover
/// whole.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct RepairPolicy {
    /// Block-salvage SSTs / blob files that fail whole-file recovery instead
    /// of leaving them out (see [`Config::repair_with_salvage`]).
    pub salvage: bool,
    /// Keep ambiguous data (a lost restriction bound, an unauthenticated
    /// delete mask) instead of dropping it (see
    /// [`Config::repair_with_resurrection`]).
    pub allow_resurrection: bool,
}

impl RepairPolicy {
    /// Enables block-level salvage (see [`Self::salvage`]).
    #[must_use]
    pub const fn salvage(mut self, enable: bool) -> Self {
        self.salvage = enable;
        self
    }

    /// Enables resurrection of ambiguous data (see
    /// [`Self::allow_resurrection`]).
    #[must_use]
    pub const fn allow_resurrection(mut self, enable: bool) -> Self {
        self.allow_resurrection = enable;
        self
    }
}

/// Completes the publication a previous repair committed but could not carry out.
///
/// A repair publishes by committing the manifest and only THEN swapping its
/// replacements onto their names and removing what the manifest no longer
/// references; a failure in that last step leaves the directory half-published
/// under a durable, correct manifest. Rebuilding from it instead would salvage
/// the damaged original again while the previous replacement is still there, so
/// one history would enter L0 twice — and duplicated merge operands are applied
/// twice on read.
///
/// The committed manifest is what resolves it. It is the tree's own authority
/// on which files count: an `open()` sweeps everything it does not name as an
/// orphan, so this applies the same rule before the scan runs — every table and
/// blob file it does not reference is removed. No state is carried between
/// runs; this is derived from the durable manifest alone.
///
/// A manifest that does not load cleanly is exactly the case repair exists for,
/// so it is not consulted at all then and the scan rebuilds from everything.
///
/// # Errors
///
/// Propagates a removal failure: leaving a superseded file in place is what
/// corrupts the rebuild, so the repair must fail rather than proceed past it.
/// What a CLEANLY-loading committed manifest hands the repair scan: the full
/// per-table records (checksum for temp-swap resolution, `global_seqno` so a
/// bulk-ingested survivor keeps its offset) and the committed RESTRICTIONS —
/// the authoritative bound for a tight-space-restricted survivor whose
/// sidecar was lost in the crash window the manifest exists to cover.
struct CommittedManifest {
    tables: crate::HashMap<TableId, crate::version::recovery::RecoveredTable>,
    restrictions: crate::HashMap<TableId, crate::UserKey>,
    /// The blob analogue of `restrictions`: each punched blob file's committed
    /// `[data_start, frontier)` boundary. Exact where the geometry walk only
    /// re-derives it — and where a backend that cannot attribute zeros to a
    /// hole re-derives it WRONG, reporting a punched file whole.
    blob_frontiers: crate::HashMap<crate::vlog::BlobFileId, u64>,
    /// The committed digest of each blob file's LIVE SUFFIX. Two copies of one
    /// id can both be self-consistent and still differ; only this says which
    /// one the manifest named, and re-stamping a fresh checksum over the other
    /// would leave existing SST handles resolving to the wrong generation.
    blob_checksums: crate::HashMap<crate::vlog::BlobFileId, crate::Checksum>,
}

#[cfg(feature = "std")]
impl CommittedManifest {
    /// What this manifest says about `table_id`'s restriction.
    fn restriction_of(&self, table_id: TableId) -> ManifestRestriction {
        if !self.tables.contains_key(&table_id) {
            return ManifestRestriction::Unknown;
        }
        self.restrictions
            .get(&table_id)
            .map_or(ManifestRestriction::Unrestricted, |bound| {
                ManifestRestriction::Restricted(bound.clone())
            })
    }
}

/// What a CLEANLY-loading committed manifest says about one scanned table's
/// restriction. It is the AUTHORITY in BOTH directions: the restriction is
/// committed BEFORE its `.restrict-bound` sidecar mirror is written, so the
/// manifest's bound is never the staler of the two; and a table it names
/// WITHOUT a restriction is genuinely unrestricted (a lifted restriction
/// drops out of the committed set), so a sidecar surviving there is stale
/// metadata that would hide a live prefix.
#[cfg(feature = "std")]
enum ManifestRestriction {
    /// The manifest names this table and restricts it to this exact bound.
    Restricted(crate::UserKey),
    /// The manifest names this table and does NOT restrict it: ignore any
    /// surviving sidecar. The punch geometry still runs — a table the
    /// manifest calls unrestricted has no committed reclaim, so zeroed
    /// blocks there are damage and must be classified as such, not opened
    /// over.
    Unrestricted,
    /// No clean manifest, or it does not name this table: the sidecar, then
    /// the punch geometry, decide.
    Unknown,
}

#[cfg(feature = "std")]
fn sweep_superseded_by_committed_manifest(
    config: &Config,
) -> crate::Result<Option<CommittedManifest>> {
    let recovery = match crate::version::recovery::recover(
        &config.path,
        &*config.fs,
        crate::config::ManifestRecoveryMode::AbsoluteConsistency,
        config.encryption.clone(),
    ) {
        Ok(recovery) => recovery,
        // A one-shot read fault on an otherwise healthy manifest must NOT be
        // read as "no committed manifest exists": the repair would then scan
        // and republish every file, and with an unfinished compaction's inputs
        // and outputs both on disk the rebuilt L0 applies duplicate merge
        // operands — where the authoritative manifest would have named which
        // files are live. Propagate for a retry.
        Err(e) if is_environmental(&e) => return Err(e),
        // A manifest of another format is not damage: it is a store this
        // engine does not write. Rebuilding it here would convert it silently
        // and drop what only its manifest holds; the offline converter is
        // the one path from an older format.
        Err(e @ crate::Error::InvalidVersion(_)) => return Err(e),
        // A manifest that does not load cleanly is exactly the case repair
        // exists for: nothing committed to consult, the scan rebuilds from
        // everything.
        Err(_) => return Ok(None),
    };

    // BEFORE anything is removed: a clean manifest's tree type is the
    // authority, and a run configured for the other type would rebuild the
    // manifest as that type — after which the CORRECT configuration fails
    // against a manifest this repair itself wrote. The scan cannot catch it
    // either: no surviving SST proves a Standard store is not a blob one (the
    // reverse is caught later, by a table that carries indirections). Refuse
    // while the store is still untouched.
    let requested = if config.kv_separation_opts.is_some() {
        TreeType::Blob
    } else {
        TreeType::Standard
    };
    if recovery.tree_type != requested {
        log::error!(
            "repair: the committed manifest describes a {:?} tree but this repair is \
             configured for {requested:?}; rebuilding would leave the store openable \
             only under the wrong configuration",
            recovery.tree_type,
        );
        return Err(crate::Error::TreeTypeMismatch {
            requested,
            actual: recovery.tree_type,
        });
    }

    // The full per-table records, not just the id set: the checksum drives
    // temp-swap resolution below, and the `global_seqno` ingest offset is
    // what the caller's scan reuses for a healthy bulk-ingested table (a
    // clean manifest record makes that offset recoverable).
    let referenced_tables: crate::HashMap<TableId, crate::version::recovery::RecoveredTable> =
        recovery
            .table_ids
            .iter()
            .flatten()
            .flatten()
            .map(|t| (t.id, *t))
            .collect();
    for (table_base_folder, folder_fs) in config.all_tables_folders() {
        if !folder_fs.exists(&table_base_folder)? {
            continue;
        }
        for dirent in folder_fs.read_dir(&table_base_folder)? {
            if dirent.is_dir {
                continue;
            }
            // A replacement a previous run built. The manifest is the authority
            // on what it is — but the manifest names the id in BOTH crash cases
            // (before the commit its entry still describes the SOURCE), so the
            // entry's checksum is what decides: only a committed run recorded
            // the temp's digest, and swapping an uncommitted build in would
            // destroy the source the manifest names. Both answers still come
            // from the durable manifest alone.
            if let Some(id) = table_id_from_repair_tmp_name(&dirent.file_name) {
                let table_path = table_base_folder.join(id.to_string());
                let published = match referenced_tables.get(&id).map(|t| t.checksum) {
                    Some(manifest_checksum) => repair_tmp_is_published(
                        config,
                        &folder_fs,
                        &dirent.path,
                        id,
                        manifest_checksum,
                        recovery.restrictions.get(&id),
                    )?,
                    None => false,
                };
                if published {
                    commit_repair_tmp(
                        &*folder_fs,
                        &dirent.path,
                        &table_path,
                        config.sync_mode,
                        recovery.restrictions.contains_key(&id),
                    )?;
                    log::info!(
                        "repair: finished the pending swap of table {id} from a previous run",
                    );
                } else {
                    discard_unreferenced(&*folder_fs, &dirent.path, config.sync_mode)?;
                    log::info!("repair: dropped an abandoned replacement for table {id}");
                }
                continue;
            }
            // Only files the scan would ADOPT as tables are swept here; a
            // foreign name is left to the scan's own classification.
            let Ok(id) = dirent.file_name.parse::<TableId>() else {
                continue;
            };
            if referenced_tables.contains_key(&id) {
                continue;
            }
            // Unreferenced bare-id files are removed even though one COULD be a
            // compaction input whose referenced output turns out corrupt below.
            // Deliberate, not an oversight: an output carries the SAME seqnos as
            // its inputs, so letting the scan adopt both would put one history
            // into L0 twice, and version collapse would then gather a key's
            // merge operands from both copies — applying each operand twice and
            // silently serving a wrong value. An honest loss (reported via
            // `lost_coverage`) beats that corruption; the sweep also only
            // applies the tree's own rule, since any successful open removes
            // the same orphans. Content a corrupt referenced file lost comes
            // back from a replica, a checkpoint plus journal replay, or a
            // backup — never from resurrected orphans.
            discard_unreferenced(&*folder_fs, &dirent.path, config.sync_mode)?;
            log::info!("repair: table {id} is superseded by the committed manifest; removed");
        }
    }

    if config.kv_separation_opts.is_some() {
        let referenced_blobs: crate::HashSet<crate::vlog::BlobFileId> =
            recovery.blob_file_ids.iter().map(|(id, _)| *id).collect();
        let blobs_folder = config.path.join(crate::file::BLOBS_FOLDER);
        if config.fs.exists(&blobs_folder)? {
            for dirent in config.fs.read_dir(&blobs_folder)? {
                let Ok(id) = dirent.file_name.parse::<crate::vlog::BlobFileId>() else {
                    continue;
                };
                if dirent.is_dir || referenced_blobs.contains(&id) {
                    continue;
                }
                match config.fs.remove_file(&dirent.path) {
                    Ok(()) => log::info!(
                        "repair: blob file {id} is superseded by the committed manifest; removed",
                    ),
                    Err(e) if e.kind() == crate::io::ErrorKind::NotFound => {}
                    Err(e) => return Err(e.into()),
                }
            }
        }
    }
    // The manifest loaded cleanly: hand the caller its referenced table
    // records, so an id it names that the directory scan never SEES (the
    // file is gone, not merely unreadable) is still reported as an
    // unscopable loss — and so a healthy bulk-ingested table's manifest-only
    // `global_seqno` offset is reused instead of failing closed. The
    // committed restrictions ride along: they are the authority for a
    // restricted survivor whose sidecar the crash took.
    Ok(Some(CommittedManifest {
        tables: referenced_tables,
        restrictions: recovery.restrictions,
        blob_frontiers: recovery.blob_restrictions,
        blob_checksums: recovery.blob_file_ids.iter().copied().collect(),
    }))
}

/// Removes the fresh-id blob replacements an aborting run already published,
/// so a pre-commit exit leaves nothing behind that a retry would have to
/// re-create beside an orphan — under tight disk space exactly the sequence
/// that fails. `NotFound` is success (a concurrent sweep, or an in-place
/// rewrite that published no file); any other removal failure propagates to
/// the caller (the guard's `Drop` downgrades it to a log line, since the run
/// is already aborting with its own error).
fn remove_published_blob_replacements(
    config: &Config,
    blobs_folder: &std::path::Path,
    replacement_ids: impl IntoIterator<Item = crate::vlog::BlobFileId>,
) -> crate::Result<()> {
    let mut removed = false;
    for id in replacement_ids {
        match config.fs.remove_file(&blobs_folder.join(id.to_string())) {
            Ok(()) => removed = true,
            Err(e) if e.kind() == crate::io::ErrorKind::NotFound => {}
            Err(e) => return Err(e.into()),
        }
    }
    if removed {
        config
            .fs
            .sync_directory_with(blobs_folder, config.sync_mode)?;
    }
    Ok(())
}

/// Every fresh-id blob replacement the running repair has PUBLISHED (renamed
/// onto its numeric name) before the manifest commit. Dropping the guard
/// while it is armed removes them: the manifest never adopted the files, and
/// each one left behind makes the retry salvage its original AGAIN beside
/// the orphan — under the tight disk space this recovery targets, exactly
/// the extra full copy that fails with ENOSPC. `Drop`-based so EVERY
/// pre-commit exit — an error `?`, a cancellation, an early return — unwinds
/// them uniformly; disarmed immediately before the manifest commit, from
/// which point the replacements are what the manifest names (a commit
/// FAILURE deliberately leaves them: the retry's unreferenced-file filter
/// removes what it does not adopt, and a post-rename commit error must not
/// delete files a switched manifest may already reference).
struct PublishedBlobReplacements<'a> {
    config: &'a Config,
    blobs_folder: PathBuf,
    ids: Vec<crate::vlog::BlobFileId>,
    armed: bool,
    /// The repair's first tree id: the replacements' descriptors are cached
    /// under ids from here on (see `HeldDescriptors`).
    first_tree_id: crate::tree::inner::TreeId,
}

impl<'a> PublishedBlobReplacements<'a> {
    fn new(config: &'a Config, first_tree_id: crate::tree::inner::TreeId) -> Self {
        Self {
            config,
            blobs_folder: config.path.join(crate::file::BLOBS_FOLDER),
            ids: Vec::new(),
            armed: true,
            first_tree_id,
        }
    }

    /// Records one replacement as published (its rename succeeded).
    fn publish(&mut self, id: crate::vlog::BlobFileId) {
        self.ids.push(id);
    }

    /// The manifest commit is next: the replacements are about to become the
    /// files it names, so an exit from here on must not remove them.
    fn disarm(&mut self) {
        self.armed = false;
    }
}

/// What the disk says about a failed `persist_version`: was the rebuilt
/// snapshot installed (renamed to its name) before the error?
enum InstallProbe {
    /// The rebuilt snapshot's name exists: it is, or may become, installed.
    Installed,
    /// The name is ABSENT (`NotFound`): the install did not happen. Absence
    /// is conclusive: the rebuilt id is above every manifest file on disk, so
    /// only the persist's rename makes the name, and only the persist's
    /// withdrawal removes it after, which the persist reports as
    /// [`crate::Error::InstallOutcomeUnknown`] unless it made the removal
    /// durable.
    NotInstalled,
    /// The name's existence cannot be told (a transient or permission failure
    /// of the probe): NO conclusion. The caller must fail SAFE: treating
    /// this as "not installed" would delete files a possibly-published
    /// manifest references.
    Inconclusive,
}

/// The post-failure probe deciding if a failed `persist_version` died before
/// or after it installed the snapshot.
fn probe_install(
    fs: &dyn crate::fs::Fs,
    folder: &std::path::Path,
    version_id: u64,
) -> InstallProbe {
    match fs.metadata(&folder.join(format!("v{version_id}"))) {
        Ok(_) => InstallProbe::Installed,
        Err(e) if e.kind() == crate::io::ErrorKind::NotFound => InstallProbe::NotInstalled,
        Err(_) => InstallProbe::Inconclusive,
    }
}

impl Drop for PublishedBlobReplacements<'_> {
    fn drop(&mut self) {
        if !self.armed || self.ids.is_empty() {
            return;
        }
        // The repair is ending: its cached descriptors, the replacements'
        // among them, close before the files go, or a backend that refuses to
        // remove an open file keeps them. The repair's own guard sweeps the
        // same ids later, and only after this one has run.
        if let Some(cache) = &self.config.descriptor_table {
            cache.remove_trees_from(self.first_tree_id);
        }
        // Best-effort: the run is already aborting with its own error, and a
        // survivor is an unreferenced file the next open's orphan sweep (or
        // the retry's unreferenced-file filter) removes.
        if let Err(e) =
            remove_published_blob_replacements(self.config, &self.blobs_folder, self.ids.drain(..))
        {
            log::warn!(
                "repair: could not remove a published blob replacement while \
                 aborting ({e}); the next open's orphan sweep removes it",
            );
        }
    }
}

/// Surfaces a caller's cooperative cancellation
/// ([`RecoveryProgress::request_cancel`](crate::RecoveryProgress::request_cancel))
/// as [`Error::Cancelled`](crate::Error::Cancelled). Consulted at file
/// boundaries, and ONLY before the manifest commit: the pre-commit scan is
/// read-only, so an abort there leaves the directory exactly as a retry
/// expects, while aborting the post-commit swaps or removals would leave work
/// the next open must redo anyway.
fn check_cancel(config: &Config) -> crate::Result<()> {
    if let Some(p) = &config.recovery_progress
        && p.is_cancel_requested()
    {
        return Err(crate::Error::Cancelled);
    }
    Ok(())
}

/// Publishes the repair's byte total (every file in the table folders, plus
/// the blob folder for a KV-separated tree) to the progress handle, if any.
/// Best-effort on purpose: a file that cannot be stat'ed is left out of the
/// total rather than failing the repair over a display number — the scan
/// itself will classify it.
fn publish_recovery_bytes_total(config: &Config) {
    let Some(progress) = &config.recovery_progress else {
        return;
    };
    let mut total: u64 = 0;
    let mut add_folder = |folder: &std::path::Path, fs: &dyn crate::fs::Fs| {
        let Ok(true) = fs.exists(folder) else {
            return;
        };
        let Ok(dirents) = fs.read_dir(folder) else {
            return;
        };
        for dirent in dirents {
            // Same skips as the scan loops, so `bytes_processed` can reach
            // `bytes_total` exactly.
            if dirent.is_dir {
                continue;
            }
            if let Ok(meta) = fs.metadata(&dirent.path) {
                // Display-only sum: clamping at u64::MAX merely freezes the
                // percentage, while a checked overflow would fail the repair
                // over a progress number.
                total = total.saturating_add(meta.len);
            }
        }
    };
    for (folder, fs) in config.all_tables_folders() {
        add_folder(&folder, &*fs);
    }
    if config.kv_separation_opts.is_some() {
        add_folder(&config.path.join(crate::file::BLOBS_FOLDER), &*config.fs);
    }
    progress.set_bytes_total(total);
}

fn repair_tree(
    config: &Config,
    salvage: bool,
    allow_resurrection: bool,
) -> crate::Result<RepairReport> {
    // Hold the cross-process directory lock for the whole repair: it writes and
    // installs a fresh snapshot and sweeps `edits-*` in place, so a
    // concurrent open / repair of the same directory would corrupt the manifest.
    // A second acquirer fails fast with `Error::Locked`. Dropped at function
    // return, releasing the lock. The directory is expected to exist (repair
    // operates on an existing tree).
    #[cfg(feature = "std")]
    let _directory_lock =
        crate::config::acquire_directory_lock(&*config.fs, &config.path, config.directory_lock)?;

    // A repair opens every SST it finds, so it needs the tree's dictionaries
    // exactly as an open does — and it does NOT go through `Tree::open`, which
    // is where they would otherwise be loaded. Without this every
    // dictionary-compressed table is graded unreadable and the rebuilt manifest
    // leaves the tree's data behind. Under the directory lock, since it reads
    // the tree's own folder, and on a COPY, so the caller's config keeps the
    // registry it had (a `Config` is cloned per tree, and they must not share).
    //
    // A dictionary whose bytes no longer hash to its name is left out rather
    // than failing the load: the repair is for exactly that tree. A table that
    // needs it then fails like one whose dictionary was never supplied, which
    // aborts rather than drops it, so a re-run with the right dictionary still
    // recovers it.
    #[cfg(zstd_any)]
    let (owned_config, damaged_dictionaries) = {
        let mut owned = config.clone();
        let damaged = owned.install_own_zstd_dictionaries_skipping_damaged()?;
        (owned, damaged)
    };
    #[cfg(zstd_any)]
    let config = &owned_config;

    // The tables the repair holds keep their files in the configured
    // descriptor cache, as an open tree does, each under a tree id of its own
    // handed out from here on; every id from this one is swept out of the
    // cache before the post-commit swaps (see `held_recover_params`).
    let first_tree_id = crate::tree::inner::get_next_tree_id();
    // However the repair ends (an abort, a cancellation, a failed commit),
    // the descriptors it cached close with it: left open, a temp it built
    // could not be removed by the retry on a backend that refuses to unlink
    // an open file.
    let _held = HeldDescriptors {
        cache: config.descriptor_table.clone(),
        first: first_tree_id,
    };

    if let Some(p) = &config.recovery_progress {
        p.set_phase(crate::RecoveryPhase::PendingSwaps);
    }
    check_cancel(config)?;

    // Finish the previous run's outstanding cleanup first. See
    // `sweep_superseded_by_committed_manifest`. When the manifest itself
    // loads cleanly (repair was entered over a MISSING file, not a lost
    // manifest), its referenced table records are kept: an id it names that
    // the directory scan never sees has no directory entry to report
    // through, and losing it silently would let `wal_replay_scope()` answer
    // `TailOnly` over lost persisted data. The records also carry each
    // table's `global_seqno`, so a healthy bulk-ingested SST keeps its
    // manifest-only ingest offset instead of being failed closed.
    #[cfg(feature = "std")]
    let manifest_referenced: Option<CommittedManifest> =
        sweep_superseded_by_committed_manifest(config)?;
    #[cfg(not(feature = "std"))]
    let manifest_referenced: Option<CommittedManifest> = None;

    // An earlier repair whose report never reached the caller: the installed
    // snapshot still owes it. This rebuild replaces that snapshot, so the
    // earlier losses travel in its report. Only that one section is read, so
    // damage elsewhere in the snapshot does not hide it. A snapshot too
    // damaged to say whether it owes one may: that replay is owed unscoped,
    // since leaving it out would lose it for good. A transient failure
    // propagates for the retry.
    let outstanding = crate::version::recovery::owed_repair_report(
        &config.path,
        &*config.fs,
        config.encryption.clone(),
    )?
    .map(|owed| match owed {
        crate::version::recovery::OwedReport::Readable(bytes) => {
            report_marker::decode_owed(&bytes, &config.path)
        }
        crate::version::recovery::OwedReport::Unreadable(id) => {
            report_marker::unreadable_owed(&config.path, id)
        }
    });

    // Byte totals for the progress percentage, from an upfront listing —
    // before the scan phase starts, so `bytes_processed / bytes_total` is
    // meaningful from the first processed file.
    publish_recovery_bytes_total(config);
    if let Some(p) = &config.recovery_progress {
        p.set_phase(crate::RecoveryPhase::ScanningTables);
    }

    // Phase 1: read every table folder and decide, per file, what it is and
    // whether it can be published. Read-only on disk — nothing is removed or
    // swapped until the rebuilt manifest is durable, so an abort here leaves
    // the directory exactly as the retry expects it.
    let scan = scan_table_folders(
        config,
        salvage,
        allow_resurrection,
        manifest_referenced.as_ref(),
    )?;

    // Phase 2: turn what the scan found into a manifest, commit it, and carry
    // out the removals and swaps that commit authorizes. The directory lock is
    // held by THIS frame for the whole of it.
    let rebuilt = rebuild_from_scan(
        config,
        allow_resurrection,
        manifest_referenced,
        scan,
        first_tree_id,
        outstanding,
        #[cfg(zstd_any)]
        &damaged_dictionaries,
    );
    #[cfg(zstd_any)]
    let rebuilt = set_aside_damaged_dictionaries(config, damaged_dictionaries, rebuilt);
    rebuilt
}

/// Moves every damaged dictionary still under its name aside, now that the
/// rebuilt manifest is durable.
///
/// Left under its name, a damaged dictionary would fail the open this repair
/// was run for. One the rebuild rewrote from a supplied copy reads intact
/// again and stays. The report already names each one moved here: the rebuild
/// records them before it writes the report's copy, which an open hands out
/// when the run ends after these moves and before the report reaches the
/// caller.
///
/// # Errors
///
/// A failure here, like one inside the rebuild's own post-commit steps, comes
/// back as [`crate::Error::RepairedButUnopened`] carrying the report: the
/// manifest is durable, so the repair happened, and a retry would find nothing
/// left to repair and answer with no report at all. Every dictionary is still
/// tried; the first failure is the one returned. A rebuild that failed before
/// its commit is passed through untouched, with nothing set aside.
#[cfg(zstd_any)]
fn set_aside_damaged_dictionaries(
    config: &Config,
    damaged: Vec<crate::file::DictId>,
    rebuilt: crate::Result<RepairReport>,
) -> crate::Result<RepairReport> {
    let (mut report, mut post_commit_error) = match rebuilt {
        Ok(report) => (report, None),
        Err(crate::Error::RepairedButUnopened { report, cause }) => (*report, Some(*cause)),
        Err(e) => return Err(e),
    };
    let folder = config.path.join(crate::file::DICTS_FOLDER);
    for id in damaged {
        let moved = match crate::dicts::set_aside_if_damaged(
            &*config.fs,
            &folder,
            id,
            config.encryption.as_deref(),
            config.sync_mode,
        ) {
            Ok(moved) => moved.is_some(),
            Err(e) => {
                post_commit_error.get_or_insert(e);
                false
            }
        };
        // The report's entry, written before the move, says now what it did.
        let aside = crate::dicts::damaged_path(&folder, id);
        if let Some(entry) = report
            .damaged_dictionaries
            .iter_mut()
            .find(|(path, _)| *path == aside)
        {
            entry.1 = if moved {
                format!("dictionary {id} failed its integrity check; moved aside")
            } else {
                format!("dictionary {id} failed its integrity check; NOT moved aside")
            };
        }
    }
    match post_commit_error {
        Some(cause) => Err(crate::Error::RepairedButUnopened {
            report: Box::new(report),
            cause: Box::new(cause),
        }),
        None => Ok(report),
    }
}

/// Everything [`scan_table_folders`] learned from the table folders, and the
/// only channel between the scan and the rebuild.
///
/// The scan's own bookkeeping ends here: a name it classified, a file it could
/// not read, a replacement it built. Nothing else it touched survives the
/// phase boundary, so the rebuild cannot accidentally depend on a scan
/// intermediate.
#[cfg(feature = "std")]
struct TableScan {
    /// Best recovered copy per table id (a complete recovery beats a lossy
    /// salvage).
    recovered_by_id: crate::HashMap<TableId, TableCandidate>,
    /// Files the rebuild must report as damaged.
    unreadable_files: Vec<(PathBuf, String)>,
    /// Damaged files that cost NO coverage: a duplicate of an id whose
    /// complete copy is retained. They belong in the corruption channel — an
    /// operator watching for a failing disk must see them — but not in the
    /// loss accounting, where an unscopable entry would tell an external WAL
    /// to reconcile the whole keyspace over a tree that lost nothing.
    redundant_unreadable: crate::HashSet<PathBuf>,
    /// HEALTHY files the rebuild deliberately leaves out.
    excluded_files: Vec<(PathBuf, String)>,
    /// What each scanned table covers, captured while its metadata was
    /// readable.
    coverage_by_path: crate::HashMap<PathBuf, (UserKey, UserKey, Option<SeqNo>)>,
    /// Files to remove once the rebuilt manifest is durable.
    discard_after_commit: Vec<(Arc<dyn crate::fs::Fs>, PathBuf, String)>,
    /// Replacements to swap onto their final names once it is.
    swap_after_commit: Vec<(Arc<dyn crate::fs::Fs>, PathBuf, PathBuf, bool)>,
    /// Sidecars found beside tables, judged against the FINAL rebuilt set.
    live_sidecars: Vec<(TableId, PathBuf, Arc<dyn crate::fs::Fs>)>,
    /// Every table id the scan actually SAW on disk, joined against the
    /// committed manifest to report ids whose file is gone entirely.
    scanned_table_ids: crate::HashSet<TableId>,
}

/// The tables that opened in one folder: each id with the ids below it its
/// folder was still writing when it was finished.
#[cfg(feature = "std")]
type OpenedTables = crate::HashMap<PathBuf, Vec<(TableId, Vec<TableId>)>>;

/// Records that `table` opened as a table in `folder`, so an empty file
/// there below it, written by the time it was finished, is judged a table
/// that lost its bytes.
#[cfg(feature = "std")]
fn note_opened(opened: &mut OpenedTables, folder: &std::path::Path, table: &Table) {
    opened
        .entry(folder.to_path_buf())
        .or_default()
        .push((table.id(), table.metadata.writing_below.clone()));
}

/// Whether one of `tables` (ids with what their folder was still writing
/// below them) was finished after `id`'s file had its table written: one
/// above it that does not list it.
#[cfg(feature = "std")]
fn written_before_one_of<'a>(
    id: TableId,
    mut tables: impl Iterator<Item = (TableId, &'a [TableId])>,
) -> bool {
    tables.any(|(table, writing_below)| id < table && !writing_below.contains(&id))
}

/// Phase 1 of [`repair_tree`]: classify every entry of every table folder and
/// recover what can be recovered. See [`TableScan`] for what it hands on.
///
/// # Errors
///
/// Propagates an environmental failure (the retry re-reads the same bytes) and
/// any failure that must abort the whole repair rather than grade one file.
#[cfg(feature = "std")]
fn scan_table_folders(
    config: &Config,
    salvage: bool,
    allow_resurrection: bool,
    manifest_referenced: Option<&CommittedManifest>,
) -> crate::Result<TableScan> {
    let mut scanned_table_ids: crate::HashSet<TableId> = crate::HashSet::default();

    // Best recovered copy per table id: a complete recovery beats a lossy salvage,
    // so a duplicate id across aliased / routed table folders keeps only the best
    // (and is never added to two L0 runs). See `keep_best_candidate`.
    let mut recovered_by_id: crate::HashMap<TableId, TableCandidate> = crate::HashMap::default();
    let mut unreadable_files: Vec<(PathBuf, String)> = Vec::new();
    let mut redundant_unreadable: crate::HashSet<PathBuf> = crate::HashSet::default();
    // HEALTHY files the rebuild deliberately leaves out (valid duplicates,
    // lineage-redundant outputs and inputs): reported in the report's
    // `excluded_files`, never in the unreadable counts — they opened and
    // verified fine, and counting them as unreadable fires corruption
    // alerts over an intact store.
    let mut excluded_files: Vec<(PathBuf, String)> = Vec::new();
    // What each scanned table covers, captured while its metadata was readable.
    // Joined against the exclusions at the end to report the coverage the
    // rebuilt manifest lost.
    let mut coverage_by_path: crate::HashMap<PathBuf, (UserKey, UserKey, Option<SeqNo>)> =
        crate::HashMap::default();
    // Files the rebuilt manifest will not name: a foreign name, a duplicate id,
    // a table no bound can make safe, a source a salvaged copy supersedes. They
    // are removed AFTER the commit and not one moment earlier — the scan is
    // read-only precisely so a crash leaves the directory as the retry expects
    // to find it, and the retry then derives everything again from the same
    // bytes. Each entry carries the backend that owns the file: a per-level
    // route stores its tables through an `Fs` the primary one cannot see.
    let mut discard_after_commit: Vec<(Arc<dyn crate::fs::Fs>, PathBuf, String)> = Vec::new();
    // Replacements built at `{id}.repair-tmp`, swapped onto `{id}` once the
    // manifest that adopts them is durable. Same ordering rule as the removals,
    // for the same reason: before the commit the source is still the only copy
    // of its rows and is what the manifest names.
    // (fs, temp path, final path, replacement is restricted): the last flag
    // rides along so a retried swap can tell an already-moved replacement
    // sidecar from a stale source one (see `commit_repair_tmp`).
    let mut swap_after_commit: Vec<(Arc<dyn crate::fs::Fs>, PathBuf, PathBuf, bool)> = Vec::new();

    // Live sidecars ({id}.heal-attest, {id}.restrict-bound) found by the scan.
    // Judged against the FINAL rebuilt set after the dedup: one whose table
    // survived is preserved, an orphan is removed post-commit.
    let mut live_sidecars: Vec<(TableId, PathBuf, Arc<dyn crate::fs::Fs>)> = Vec::new();

    // Empty files under table names no committed manifest names:
    // `(id, folder, path, fs)`, judged against the tables their folder
    // recovered once the scan is done.
    let mut empty_unnamed: Vec<(TableId, PathBuf, PathBuf, Arc<dyn crate::fs::Fs>)> = Vec::new();
    // The files that opened as tables, per folder, whether the rebuild then
    // keeps them, sets them aside or leaves them out as duplicates: tables
    // written there either way, each telling which files below it were
    // still being written when it was finished.
    let mut opened_tables: OpenedTables = crate::HashMap::default();

    for (table_base_folder, folder_fs) in config.all_tables_folders() {
        if !folder_fs.exists(&table_base_folder)? {
            continue;
        }

        // ORDER the entries before recovering: `read_dir` order is
        // FS-dependent, and duplicate-id resolution keeps the FIRST complete
        // copy — so an unordered scan would rebuild different contents from the
        // same directory across runs. Per id, the writer's own `id.to_string()`
        // spelling is the canonical file and sorts first, so a foreign alternate
        // spelling (`01` for id 1) can never displace it. Mirrors the blob-file
        // scan.
        let mut dirents = folder_fs.read_dir(&table_base_folder)?;
        dirents.sort_by(|a, b| {
            // Group every name that belongs to one id together, and inside a
            // group put the abandoned `{id}.repair-tmp` FIRST: this scan
            // removes it (no manifest names it — one that did was swapped in
            // before the scan started), and a salvage of `{id}` builds its
            // replacement under exactly that name. Visiting the source first
            // would find the destination occupied, fail the salvage with
            // `AlreadyExists` — not an environmental error — and condemn a
            // source whose salvage would otherwise have succeeded.
            let key = |e: &crate::fs::FsDirEntry| {
                let numeric = e.file_name.parse::<TableId>().ok();
                let temp = table_id_from_repair_tmp_name(&e.file_name);
                let id = numeric.or(temp);
                // Within an id: the temp, then the writer's own canonical
                // spelling, then any foreign alternate spelling (`01` for id
                // 1) — which can therefore never displace the canonical file.
                let rank = if temp.is_some() {
                    0
                } else if numeric.is_some_and(|id| e.file_name == id.to_string()) {
                    1
                } else {
                    2
                };
                (id.is_none(), id.unwrap_or(0), rank)
            };
            key(a)
                .cmp(&key(b))
                .then_with(|| a.file_name.cmp(&b.file_name))
        });

        'dirent: for dirent in dirents {
            let crate::fs::FsDirEntry {
                path: table_path,
                file_name,
                is_dir,
            } = dirent;

            if is_dir {
                continue;
            }

            // Pre-commit file boundary: safe to abort (the scan is read-only),
            // and the file counts toward the progress percentage as soon as
            // its processing starts. Best-effort stat, like the total.
            check_cancel(config)?;
            if let Some(p) = &config.recovery_progress
                && let Ok(meta) = folder_fs.metadata(&table_path)
            {
                p.add_bytes_processed(meta.len);
            }

            // One grammar decides what each name IS (`TableDirEntry`, shared
            // with `Tree::open`'s sweep so the two can never disagree on
            // ownership); this match is the repair scan's POLICY for each kind.
            use crate::file::TableDirEntry;
            let table_id = match TableDirEntry::classify(&file_name) {
                // LIVE sidecar artifacts are not table files, and repair must
                // not remove one whose table survives: `Tree::open` PRESERVES
                // a live `{id}.heal-attest` (the next scrub reconciles a
                // crashed digest refresh through it), and the
                // `.restrict-bound` bound is read FOR its SST via
                // `restrict_bound::read`. Whether a sidecar is live is
                // decided AGAINST THE REBUILT SET after the dedup below —
                // one whose table did not survive is an orphan the repair
                // itself removes, because leaving it hands the next open an
                // unconditional sweep whose refused unlink would turn this
                // run's success into an unopenable tree.
                TableDirEntry::HealAttest(id) | TableDirEntry::RestrictBound(id) => {
                    live_sidecars.push((id, table_path, Arc::clone(&folder_fs)));
                    continue;
                }
                // The disposable `.tmp` shapes are crash leftovers no state
                // references; the open sweeps them unconditionally, so repair
                // removes them itself (post-commit) for the same reason.
                TableDirEntry::HealAttestTmp(_)
                | TableDirEntry::HealTmp(_)
                | TableDirEntry::RestrictBoundTmp(_) => {
                    discard_after_commit.push((
                        Arc::clone(&folder_fs),
                        table_path,
                        "disposable crashed-heal artifact; the next open would \
                         sweep it and fail on a refused removal"
                            .to_string(),
                    ));
                    continue;
                }
                // An abandoned replacement's restriction sidecar: its temp is
                // removed by the `RepairTmp` arm (which also takes the
                // companion, so this entry is usually already gone — the
                // post-commit removal tolerates that), and a swept temp's
                // survivor is an orphan.
                TableDirEntry::RepairTmpCompanion(_) => {
                    discard_after_commit.push((
                        Arc::clone(&folder_fs),
                        table_path,
                        "restriction sidecar of an abandoned repair replacement".to_string(),
                    ));
                    continue;
                }
                // `{id}.repair-tmp` is a replacement a previous repair was
                // building and never published: no manifest names it (one that
                // did was swapped in before this scan started), and its source
                // is whatever this scan finds under `{id}`. It is therefore
                // garbage, and it is removed NOW rather than after the commit —
                // this run needs the name free to build its own replacement.
                // Removing it cannot lose rows: every row it holds came from a
                // file this scan is about to read.
                TableDirEntry::RepairTmp(_) => {
                    // The removal also takes the temp's restriction
                    // companion, whose own directory entry is visited LATER
                    // in this scan — by then its stat returns NotFound and
                    // the byte-progress prologue skips it, leaving a
                    // successful repair short of 100%. Credit its bytes now,
                    // before they disappear.
                    if let Some(p) = &config.recovery_progress
                        && let Ok(meta) =
                            folder_fs.metadata(&crate::restrict_bound::sidecar_path(&table_path))
                    {
                        p.add_bytes_processed(meta.len);
                    }
                    discard_unreferenced(&*folder_fs, &table_path, config.sync_mode)?;
                    log::warn!(
                        "repair: dropped an abandoned replacement {}",
                        table_path.display(),
                    );
                    continue;
                }
                // Not a shape the engine names, so not part of the inventory a
                // repair rebuilds: left exactly where it is. Removing it to
                // make the tree openable would be destroying an operator's file
                // to fix a problem the scanner invented, and the open no longer
                // has that problem.
                TableDirEntry::Foreign => {
                    log::debug!(
                        "repair: ignoring {} in the tables folder: not an engine file",
                        table_path.display(),
                    );
                    continue;
                }
                TableDirEntry::Table(id) => id,
            };
            // An empty file under a table name that no committed manifest
            // names is judged once every table of its folder is recovered
            // (see `empty_unnamed`). A file whose length cannot be read is
            // graded as any table file is.
            let named = manifest_referenced
                .as_ref()
                .is_some_and(|m| m.tables.contains_key(&table_id));
            if !named
                && folder_fs
                    .metadata(&table_path)
                    .is_ok_and(|meta| meta.len == 0)
            {
                empty_unnamed.push((
                    table_id,
                    table_base_folder.clone(),
                    table_path,
                    Arc::clone(&folder_fs),
                ));
                continue;
            }
            scanned_table_ids.insert(table_id);

            if let Some(p) = &config.recovery_progress {
                p.table_discovered();
            }

            // A CLEAN manifest record for this id carries the table's
            // `global_seqno` and L0 recency: table files are immutable once
            // published and ids are never reused, so the record describes THIS
            // logical table even when its bytes have since been damaged.
            // Reusing it keeps a healthy bulk-ingested SST (and its real
            // sequence and L0 position) where the manifest-loss rule would have
            // to fail closed.
            let manifest_position: Option<ManifestPosition> = manifest_referenced
                .as_ref()
                .and_then(|m| m.tables.get(&table_id))
                .map(ManifestPosition::of_record);
            // What the committed manifest says about this id's restriction —
            // authoritative in both directions (see `ManifestRestriction`).
            let manifest_restriction = manifest_referenced
                .as_ref()
                .map_or(ManifestRestriction::Unknown, |m| m.restriction_of(table_id));

            // This file's whole-file digest, computed ONCE: the duplicate
            // verdict below and the recovery further down both need it. A
            // RESTRICTED table's manifest entry digests only its live suffix,
            // which this never reproduces, so `matches_manifest` stays false
            // (uncomparable) rather than reading as a mismatch.
            let own_digest = match compute_table_checksum(&*folder_fs, &table_path) {
                Ok(c) => Ok(crate::Checksum::from_raw(c)),
                // A fault in the ENVIRONMENT is not evidence about these bytes.
                Err(e) if is_environmental(&e) => return Err(e),
                Err(e) => Err(e),
            };
            let committed_digest = manifest_referenced
                .as_ref()
                .and_then(|m| m.tables.get(&table_id))
                .map(|t| t.checksum);
            let matches_manifest = match (&own_digest, committed_digest) {
                (Ok(d), Some(committed)) => {
                    match trustworthy_restriction_bound(
                        config,
                        &*folder_fs,
                        &table_path,
                        table_id,
                        &manifest_restriction,
                    )? {
                        // A RESTRICTED entry digests only the LIVE SUFFIX, so
                        // the whole-file hash can never equal it. Comparing them
                        // would mark BOTH copies of a restricted id unmatched
                        // and hand the choice back to scan order, the very
                        // thing the digest is here to prevent. Reproduce the
                        // suffix digest the restricted view records instead.
                        Some(bound) => restricted_suffix_digest(
                            config,
                            &folder_fs,
                            &table_path,
                            table_id,
                            &bound,
                        )?
                        .is_some_and(|suffix| suffix == committed),
                        None => *d == committed,
                    }
                }
                _ => false,
            };

            // Skip a duplicate id ONLY when we already hold a COMPLETE copy — a
            // duplicate cannot improve on it. A previously-seen LOSSY salvage does
            // NOT skip: this copy is still evaluated and may supersede it.
            //
            // "Complete" is not enough on its own: two complete copies can hold
            // different generations, and the manifest says which one this id
            // is. A copy that reproduces its digest is evaluated even against a
            // complete incumbent, so it can displace one that does not.
            if let Some(existing) = recovered_by_id
                .get(&table_id)
                .filter(|c| c.fidelity.is_complete())
                .filter(|c| c.matches_manifest || !matches_manifest)
            {
                // If this path physically ALIASES the retained copy (a symlink /
                // junction / case-insensitive alias resolving to the same directory
                // entry, e.g. two configured folders pointing at one location), it
                // is the SAME file, not a genuine duplicate. Removing it would
                // destroy the kept copy and leave the manifest referencing a
                // missing SST, so skip it IN PLACE.
                if same_physical_file(&*folder_fs, &table_path, &*existing.fs, &existing.path)? {
                    continue;
                }
                // A genuine duplicate: removed after the commit so recovery
                // cannot later resolve it instead of the kept copy (the manifest
                // records only id + checksum, not a path). It cannot improve the
                // rebuild, but the two report channels make DIFFERENT claims —
                // `excluded_files` promises a file that opened and verified,
                // `unreadable_files` is the corruption signal an operator
                // watches. Inferring health from the RETAINED copy's fidelity
                // would file a rotting duplicate as healthy and leave
                // `unreadable == 0` over a failing disk, so verify this copy on
                // its own bytes before choosing.
                let verdict = match own_digest {
                    Ok(digest) => Table::recover(repair_recover_params(
                        config,
                        table_path.clone(),
                        digest,
                        table_id,
                        folder_fs.clone(),
                        manifest_position,
                    ))
                    // It opened as a table, so one was written into this
                    // folder, whatever the walk below finds and wherever the
                    // kept copy lives.
                    .inspect(|table| note_opened(&mut opened_tables, &table_base_folder, table))
                    // The digest was recomputed from THESE bytes, so recovery
                    // can only prove the file is self-consistent — and it stops
                    // at the trailer, meta and index, leaving every lazily-read
                    // data block unexamined. Walk the blocks, exactly as the
                    // retained copy's keep-decision does.
                    .and_then(|table| {
                        // A tight-space-punched copy carries the same
                        // legitimate hole its retained twin does, and the walk
                        // starts at the view's punch offset — which is zero on
                        // an unrestricted open. Restrict it first, or its
                        // reclaimed prefix reads as corruption and a healthy
                        // duplicate lands in the corruption channel.
                        let table = match trustworthy_restriction_bound(
                            config,
                            &*folder_fs,
                            &table_path,
                            table_id,
                            &manifest_restriction,
                        )? {
                            Some(bound) => table.reopen_restricted(bound)?,
                            None => table,
                        };
                        match block_verify_verdict(config, &folder_fs, &table_path, &table)? {
                            BlockVerifyVerdict::Clean | BlockVerifyVerdict::DegradedButReadable => {
                                Ok(())
                            }
                            BlockVerifyVerdict::Corrupt => Err(crate::Error::InvalidHeader(
                                "block verification failed on this copy",
                            )),
                            // Nothing about the data was verified (the walk
                            // could not size the parity trailers), so this copy
                            // cannot be called healthy either.
                            BlockVerifyVerdict::DegradedUnscanned => Err(
                                crate::Error::InvalidHeader("this copy could not be verified"),
                            ),
                        }
                    }),
                    // Same split as the retained copy's read: a fault in the
                    // ENVIRONMENT is not evidence about these bytes.
                    Err(e) if is_environmental(&e) => return Err(e),
                    Err(e) => Err(e),
                };
                // `block_verify_verdict` propagates an environmental failure
                // rather than grading it; that is the operator's to fix, not a
                // verdict on this copy.
                let verdict = match verdict {
                    Err(e) if is_environmental(&e) => return Err(e),
                    other => other,
                };
                match verdict {
                    Ok(()) => {
                        let reason = "duplicate table id; a complete copy is already held";
                        excluded_files.push((table_path.clone(), reason.to_string()));
                        discard_after_commit.push((
                            Arc::clone(&folder_fs),
                            table_path,
                            reason.to_string(),
                        ));
                    }
                    Err(e) => {
                        let reason = format!(
                            "damaged duplicate of table {table_id} (kept copy is intact): {e}"
                        );
                        // The corruption signal is real, but the coverage is
                        // not lost: the complete copy of this id is retained.
                        redundant_unreadable.insert(table_path.clone());
                        set_aside_path(
                            &folder_fs,
                            &table_path,
                            &reason,
                            &mut unreadable_files,
                            &mut discard_after_commit,
                        );
                    }
                }
                continue;
            }

            // Hash the file and open it. A non-transient hashing failure (a bad
            // data sector) is FOLDED into the recover Result so the salvage arm
            // below can recover the table's intact blocks, instead of recording
            // the whole table unreadable — which the next open's orphan cleanup
            // would then delete. Table::recover would fail on the same bytes, so
            // skip it; block salvage opens with a placeholder digest and drops
            // only the unreadable blocks.
            // Whether THIS file is the one the committed manifest digested.
            // Two complete copies of an id can hold different generations, and
            // this is the only thing that tells them apart. A RESTRICTED
            // table's entry digests its live suffix, which a whole-file hash
            // never reproduces, so it stays `false` (uncomparable) rather than
            // wrongly reading as a mismatch.
            // The digest was taken above, before the duplicate verdict needed
            // it. An ENVIRONMENTAL hashing failure already propagated there; a
            // read that failed on the BYTES (a bad data sector, a corrupt
            // trailer) is folded into the recover Result so the
            // structural-failure salvage arm below recovers the intact blocks
            // (or records it unreadable with salvage off).
            let recovered = match own_digest {
                Ok(digest) => Table::recover(held_recover_params(
                    config,
                    table_path.clone(),
                    digest,
                    table_id,
                    folder_fs.clone(),
                    manifest_position,
                )),
                Err(e) => Err(e),
            };

            // Remember what this table COVERS while its metadata is in hand. If
            // it ends up excluded, that coverage is what the rebuilt manifest
            // lost, and the report names it: older versions of those keys
            // survive elsewhere and become visible again, which no repair can
            // distinguish from "the key was simply never rewritten".
            if let Ok(t) = &recovered {
                note_opened(&mut opened_tables, &table_base_folder, t);
                let range = t.metadata.key_range.clone();
                // The bound is UNKNOWN when the table's sequence base lived in
                // the LOST manifest: that open passes offset 0, so
                // `get_highest_seqno` would report the on-disk LOCAL maximum
                // (normally 0 for a bulk-ingested SST) and an operator scoping
                // the affected history by it would stop far below the truth.
                // With a clean manifest record the offset was reused above,
                // so the bound is honest again.
                let seqno = (manifest_position.is_some()
                    || !has_unrecoverable_ingest_offset(
                        t.metadata.bulk_ingested,
                        t.metadata.item_count,
                        t.max_local_seqno(),
                    ))
                .then(|| t.get_highest_seqno());
                coverage_by_path.insert(
                    table_path.clone(),
                    (range.min().clone(), range.max().clone(), seqno),
                );
            }

            // Fail closed on a table whose bulk-ingest sequence offset cannot be
            // reconstructed. A bulk-ingested SST stores every entry at LOCAL seqno
            // 0 and relies on a manifest-only `global_seqno` for its effective MVCC
            // ordering; the on-disk seqnos carry no trace of it. The rebuilt
            // manifest hard-codes offset 0, so keeping such a table would make its
            // entries appear OLDER than they are — visible to snapshots that never
            // saw them and sorted into the wrong L0 order. Drop it instead of
            // silently corrupting MVCC (see `has_unrecoverable_ingest_offset`).
            // ONLY without a clean manifest record: with one, the offset was
            // recovered above and the table keeps its real sequence position.
            if manifest_position.is_none()
                && matches!(&recovered, Ok(t) if has_unrecoverable_ingest_offset(
                    t.metadata.bulk_ingested,
                    t.metadata.item_count,
                    t.max_local_seqno(),
                ))
            {
                drop(recovered); // release the file handle
                set_aside_path(
                    &folder_fs,
                    &table_path,
                    "bulk-ingest sequence offset cannot be reconstructed from the SST",
                    &mut unreadable_files,
                    &mut discard_after_commit,
                );
                continue;
            }

            // Rebuild the restricted view of a tight-space-PUNCHED SST. Tight-space
            // compaction reclaims a table's consumed prefix data blocks in place
            // (hole-punched, reading back as zeros) and records the exact bound in a
            // `.restrict-bound` sidecar beside the SST. A rebuilt manifest must
            // re-apply that restriction, or later reads and compactions traverse the
            // punched blocks and fail. The bound comes from the sidecar (the SST
            // itself is never mutated, so its whole-file checksum stays valid);
            // written strictly post-commit, a valid sidecar is itself proof of a
            // committed restriction, so its bound is honored directly (see below).
            // Whether the recovered view's restriction came from PUNCH GEOMETRY
            // with resurrection off: that bound is the straddling block's END
            // key, which can discard the block's still-live suffix rows — a
            // deliberate, non-salvage loss the report must carry (scoped by the
            // source's coverage), or `wal_replay_scope` stays `TailOnly` while
            // live rows were removed. An exact sidecar bound and the
            // resurrection-mode bound (which keeps the whole readable region)
            // lose nothing.
            let mut geometry_lossy = false;
            let recovered = 'restrict: {
                let Ok(table) = recovered else {
                    break 'restrict recovered;
                };
                // The exact bound, from the committed manifest or its sidecar
                // mirror. See `trustworthy_restriction_bound`.
                let exact_bound = match trustworthy_restriction_bound(
                    config,
                    &*folder_fs,
                    &table_path,
                    table_id,
                    &manifest_restriction,
                ) {
                    Ok(bound) => bound,
                    Err(e) => break 'restrict Err(e),
                };

                // An exact bound is honored DIRECTLY, without probing the
                // below-bound prefix: whether or not the punch has run, reopening at
                // the bound is correct. If the prefix is not yet punched (the crash
                // window between the durable commit and the punch, or a punch deferred
                // by a live reader), the committed output already covers the dropped
                // prefix, so honoring resurrects nothing; if it is punched, the
                // reopened view digests only the live suffix. Reading the dead prefix
                // to decide is not just unnecessary — a persistently unreadable sector
                // there would otherwise discard the exact bound and, with salvage off,
                // drop the whole table despite its intact live suffix.
                // `reopen_restricted` reads only from the punch offset up, so a
                // genuinely unreadable SUFFIX still surfaces its error there.
                if let Some(bound) = &exact_bound {
                    break 'restrict table
                        .reopen_restricted_with(bound.clone(), held_descriptors(config));
                }

                // No trustworthy exact bound. An unpunched table never carried a
                // restriction, so it opens unrestricted. A punched table lost its
                // exact bound and falls to the punch geometry: with resurrection
                // on, restrict to the FIRST key of the first readable block past
                // the punched region, keeping the whole ambiguous readable region
                // (its consumed rows resurrected, as the flag contracts). With
                // resurrection OFF the geometry is trusted only when the zeroed
                // blocks form a CLEAN prefix — the pattern of a fully successful
                // punch — and the bound is that prefix's straddling block's END
                // key, never resurrecting a superseded key. An IRREGULAR pattern
                // (a readable block below a zeroed one) is positive evidence of
                // failed punches, after which no geometry bound can separate
                // intact-but-consumed blocks from live ones: the table is set
                // aside (see `DerivedRestriction::IrregularPunch`), matching the
                // recovery-failure arm's fail-closed guard for the same state. A
                // fully-punched SST with no live data is set aside too, losing
                // nothing the flag could have kept.
                // ONE pass over the data blocks answers both questions below —
                // whether the zeros are a reclaim, and where the live region
                // starts.
                let geometry = match table.punch_geometry() {
                    Ok(geometry) => geometry,
                    Err(e) => break 'restrict Err(e),
                };
                match geometry.verdict {
                    crate::table::PunchProbe::Unpunched => break 'restrict Ok(table),
                    // UNPROVEN zeros on a punch-capable mount: a lost-sidecar
                    // reclaim and damage are indistinguishable, and publishing
                    // unrestricted would resurrect the boundary block's rows
                    // below the committed restriction. Without resurrection
                    // the table is set aside; with it, the geometry paths
                    // below (which read zeroed BLOCKS, not holes) restrict
                    // past the ambiguous region, accepting the re-exposure
                    // the flag contracts.
                    crate::table::PunchProbe::Unproven if !allow_resurrection => {
                        drop(table);
                        set_aside_path(
                            &folder_fs,
                            &table_path,
                            "zeroed data blocks whose allocation state the backend \
                             cannot attribute (a lost-sidecar punch and damage are \
                             indistinguishable); a resurrection repair restricts past \
                             the zeroed region instead",
                            &mut unreadable_files,
                            &mut discard_after_commit,
                        );
                        continue 'dirent;
                    }
                    crate::table::PunchProbe::Punched | crate::table::PunchProbe::Unproven => {
                        use crate::table::DerivedRestriction;
                        let derived = if allow_resurrection {
                            match table.greedy_restriction_bound(&geometry) {
                                Ok(Some(bound)) => Ok(DerivedRestriction::Bound(bound)),
                                Ok(None) => Ok(DerivedRestriction::NoLiveData),
                                Err(e) => Err(e),
                            }
                        } else {
                            Ok(Table::conservative_restriction(&geometry))
                        };
                        match derived {
                            Ok(DerivedRestriction::Bound(bound)) => {
                                geometry_lossy = !allow_resurrection;
                                break 'restrict table
                                    .reopen_restricted_with(bound, held_descriptors(config));
                            }
                            Err(e) => break 'restrict Err(e),
                            Ok(
                                reason @ (DerivedRestriction::NoLiveData
                                | DerivedRestriction::IrregularPunch),
                            ) => {
                                // The flag decides this table's fate WITHIN this
                                // run and nowhere else. Resurrection would have
                                // kept the readable region; without it there is
                                // no bound that separates consumed rows from
                                // live ones, so the table is dropped. Nothing is
                                // stashed for a later run to reconsider: a run
                                // that stashed would hand the next one a
                                // different directory than the one it derived
                                // from, and the flag would stop being an input
                                // and start being a state machine.
                                let reason = match reason {
                                    DerivedRestriction::NoLiveData => {
                                        "fully hole-punched SST with no live data"
                                    }
                                    _ => {
                                        "partially punched SST with punch failures and no \
                                         trustworthy bound; the consumed/live boundary is \
                                         unknowable (a resurrection repair keeps the readable \
                                         region instead)"
                                    }
                                };
                                drop(table);
                                set_aside_path(
                                    &folder_fs,
                                    &table_path,
                                    reason,
                                    &mut unreadable_files,
                                    &mut discard_after_commit,
                                );
                                continue 'dirent;
                            }
                        }
                    }
                }
            };

            match recovered {
                // In salvage mode a table whose whole-file recovery succeeded can
                // still hold corrupt data blocks (recovery is lazy on the data
                // section). Block-verify it; if any block is corrupt, salvage it
                // rather than keep a table that errors on read. Encrypted and
                // unencrypted tables take the SAME encryption-aware out-of-band
                // walk (block headers and payload checksums are plaintext; the
                // provider only decodes the meta block) — the recovered `table`
                // merely supplies the id the encrypted meta's AAD binds. Without
                // the provider the walk could not decode an encrypted meta block
                // and would misreport every healthy encrypted table as corrupt,
                // rewriting it on every repair.
                Ok(table) if salvage => {
                    match verify_keep_decision(
                        config,
                        &folder_fs,
                        &table_path,
                        &table,
                        allow_resurrection,
                        true,
                    )? {
                        RepairKeepDecision::Keep => {
                            record_best(
                                &mut recovered_by_id,
                                &mut unreadable_files,
                                &mut redundant_unreadable,
                                &mut discard_after_commit,
                                &mut swap_after_commit,
                                table_id,
                                table,
                                if geometry_lossy {
                                    Fidelity::GeometryRestricted
                                } else {
                                    Fidelity::Complete
                                },
                                &folder_fs,
                                &table_path,
                                matches_manifest,
                            )?;
                        }
                        RepairKeepDecision::Drop(reason) => {
                            drop(table);
                            set_aside_path(
                                &folder_fs,
                                &table_path,
                                reason,
                                &mut unreadable_files,
                                &mut discard_after_commit,
                            );
                        }
                        RepairKeepDecision::Salvage => {
                            // A tight-space RESTRICTED punched SST whose live suffix
                            // is ALSO corrupt (a rare double failure) is block-
                            // salvaged like any other, then RE-RESTRICTED to its
                            // original bound: salvage recovers the readable blocks
                            // into a fresh, unpunched SST (dropping the zeroed prefix
                            // and the corrupt blocks), and reopening that restricted
                            // to the recorded bound masks the straddling block's
                            // sub-bound rows again, so nothing superseded is
                            // resurrected. A sidecar re-records the bound so a later
                            // manifest-loss repair honors it (the fresh file is
                            // unpunched). With resurrection on, the whole readable
                            // region is kept instead. The live suffix is never
                            // thrown away.
                            let restrict_bound = table.restrict_lower_bound().cloned();
                            drop(table);
                            // The replacement is built at `{id}.repair-tmp` and
                            // swapped onto `{id}` only after the manifest that
                            // adopts it is durable. The source is never displaced
                            // first: a crash would then leave the only copy
                            // somewhere the retry does not scan, and the table's
                            // keys would silently vanish from the rebuilt
                            // manifest. Reading an untouched source and writing a
                            // name no scan adopts also makes the retry repeat
                            // exactly the same deterministic salvage.
                            let output_path = repair_tmp_path(&table_path);
                            match try_salvage_table(
                                config,
                                &folder_fs,
                                allow_resurrection,
                                TableSalvage {
                                    source: &table_path,
                                    table_path: &output_path,
                                    table_id,
                                    // The bound (when any) comes from the
                                    // restricted view and is re-imposed below,
                                    // so a punched source is never ambiguous
                                    // on this arm.
                                    reject_punched_without_bound: false,
                                    blob_rewrite: None,
                                    owner_promotions: None,
                                    recovered_position: manifest_position,
                                },
                            ) {
                                Ok(SalvageOutcome::Salvaged(salvaged)) => {
                                    // Re-impose the tight-space restriction on the
                                    // salvaged output (fail-closed unless resurrection
                                    // is on), the shared path both salvage arms use.
                                    let table = restrict_salvaged_output(
                                        &*folder_fs,
                                        config,
                                        &output_path,
                                        salvaged,
                                        restrict_bound.clone(),
                                        allow_resurrection,
                                    )?;
                                    keep_salvaged_replacement(
                                        &mut recovered_by_id,
                                        &mut unreadable_files,
                                        &mut redundant_unreadable,
                                        &mut discard_after_commit,
                                        &mut swap_after_commit,
                                        table_id,
                                        table,
                                        &folder_fs,
                                        &table_path,
                                        output_path,
                                    )?;
                                }
                                Ok(SalvageOutcome::Unusable | SalvageOutcome::PunchedBoundLost) => {
                                    let reason = "verify found corrupt blocks; nothing salvageable";
                                    set_aside_path(
                                        &folder_fs,
                                        &table_path,
                                        reason,
                                        &mut unreadable_files,
                                        &mut discard_after_commit,
                                    );
                                }
                                // A TRANSIENT I/O salvage failure is retryable, and
                                // nothing has moved: the source is untouched, so the
                                // retry re-derives the same salvage from it. A
                                // STRUCTURAL failure is genuine unsalvageability.
                                Err(salvage_err) if is_environmental(&salvage_err) => {
                                    return Err(salvage_err);
                                }
                                Err(salvage_err) => {
                                    let reason = format!(
                                        "verify found corrupt blocks; salvage failed \
                                         ({salvage_err})"
                                    );
                                    set_aside_path(
                                        &folder_fs,
                                        &table_path,
                                        &reason,
                                        &mut unreadable_files,
                                        &mut discard_after_commit,
                                    );
                                }
                            }
                        }
                    }
                }
                // Salvage OFF: block-verify all the same. Whole-file recovery is
                // lazy on the data section, and the manifest digest is freshly
                // computed over whatever bytes are there — blessing a table with
                // a corrupt data block would LAUNDER the damage (the report
                // counts it recovered and `verify_integrity` passes while reads
                // of the affected block fail). The salvage flag only decides what
                // happens to a damaged table: rewritten (on) or set aside (off,
                // here), with the report pointing at the salvage-enabled repair.
                Ok(table) => {
                    match verify_keep_decision(
                        config,
                        &folder_fs,
                        &table_path,
                        &table,
                        allow_resurrection,
                        false,
                    )? {
                        RepairKeepDecision::Keep => {
                            record_best(
                                &mut recovered_by_id,
                                &mut unreadable_files,
                                &mut redundant_unreadable,
                                &mut discard_after_commit,
                                &mut swap_after_commit,
                                table_id,
                                table,
                                if geometry_lossy {
                                    Fidelity::GeometryRestricted
                                } else {
                                    Fidelity::Complete
                                },
                                &folder_fs,
                                &table_path,
                                matches_manifest,
                            )?;
                        }
                        // `Salvage` is unreachable with the flag off, but a
                        // defensive fallthrough beats a panic in a repair path.
                        decision @ (RepairKeepDecision::Drop(_) | RepairKeepDecision::Salvage) => {
                            let reason = match decision {
                                RepairKeepDecision::Drop(reason) => reason,
                                _ => {
                                    "verification found corrupt data blocks; run a \
                                     salvage-enabled repair to rewrite the readable blocks"
                                }
                            };
                            drop(table);
                            set_aside_path(
                                &folder_fs,
                                &table_path,
                                reason,
                                &mut unreadable_files,
                                &mut discard_after_commit,
                            );
                        }
                    }
                }
                Err(e) if salvage => {
                    // A TRANSIENT recovery failure (Io) is retryable and must NOT
                    // be routed through salvage: salvaging a healthy SST would,
                    // for a range-tombstone table, fail deterministically with
                    // FeatureUnsupported (a non-Io error recorded as
                    // unsalvageable), committing a manifest without the table —
                    // turning a one-shot read failure into permanent loss.
                    // Propagate the I/O error so a retry re-recovers it,
                    // mirroring the verification / salvage-error paths.
                    if is_environmental(&e) {
                        return Err(e);
                    }
                    // A tight-space-punched SST that fails whole-file recovery still
                    // carries its restriction in the CLEAN manifest, or in the
                    // `.restrict-bound` sidecar mirroring it — but recovery produced
                    // no `Table` to read the bound from. Same question, same answer as
                    // the verification-failure arm above; here the bound is re-imposed
                    // on the salvaged replacement instead of reopening the source.
                    let restrict_bound = trustworthy_restriction_bound(
                        config,
                        &*folder_fs,
                        &table_path,
                        table_id,
                        &manifest_restriction,
                    )?;
                    // A PUNCHED source with no trustworthy bound (missing / corrupt
                    // sidecar) cannot be salvaged into an UNRESTRICTED output:
                    // recovery produced no `Table` to derive a geometry bound from,
                    // and salvage drops the zeroed prefix but re-emits the straddling
                    // block's sub-bound rows, which would resurrect with nothing to
                    // restrict them. Fail closed: set it aside. An UNPUNCHED source
                    // (the common corrupt-table case) has no zeroed prefix and
                    // salvages normally. Resurrection-on skips the guard, accepting
                    // the re-exposure.
                    if restrict_bound.is_none()
                        && !allow_resurrection
                        && source_prefix_is_punched(&*folder_fs, &table_path)?
                    {
                        // The flag decides this within THIS run: with
                        // resurrection on, the same source salvages and keeps
                        // its readable region. Nothing is stashed for a later
                        // run to reconsider.
                        let reason = "punched SST with no recoverable restriction bound \
                                      (missing / corrupt sidecar and failed recovery); a \
                                      resurrection repair keeps its readable region instead";
                        set_aside_path(
                            &folder_fs,
                            &table_path,
                            reason,
                            &mut unreadable_files,
                            &mut discard_after_commit,
                        );
                        continue;
                    }
                    // Whole-file recovery failed structurally; try block-level
                    // salvage. The recoverable blocks go into `{id}.repair-tmp`,
                    // which is swapped onto `{id}` only once the manifest that
                    // adopts it is durable — so a crash mid-repair leaves the
                    // directory exactly as the retry expects to find it.
                    let output_path = repair_tmp_path(&table_path);
                    // Fail closed when the salvage walk reveals a punched source
                    // with no recoverable bound: the pre-salvage first-bytes
                    // probe above catches a punched FIRST block, but a partial
                    // punch can leave that block intact while later prefix
                    // blocks are zeroed — only the walk's dropped extents expose
                    // those.
                    let reject_punched = restrict_bound.is_none() && !allow_resurrection;
                    match try_salvage_table(
                        config,
                        &folder_fs,
                        allow_resurrection,
                        TableSalvage {
                            source: &table_path,
                            table_path: &output_path,
                            table_id,
                            reject_punched_without_bound: reject_punched,
                            blob_rewrite: None,
                            owner_promotions: None,
                            recovered_position: manifest_position,
                        },
                    ) {
                        Ok(SalvageOutcome::Salvaged(salvaged)) => {
                            // Re-impose the tight-space restriction on the salvaged
                            // output (fail-closed unless resurrection is on), the
                            // shared path both salvage arms use.
                            let table = restrict_salvaged_output(
                                &*folder_fs,
                                config,
                                &output_path,
                                salvaged,
                                restrict_bound,
                                allow_resurrection,
                            )?;
                            keep_salvaged_replacement(
                                &mut recovered_by_id,
                                &mut unreadable_files,
                                &mut redundant_unreadable,
                                &mut discard_after_commit,
                                &mut swap_after_commit,
                                table_id,
                                table,
                                &folder_fs,
                                &table_path,
                                output_path,
                            )?;
                        }
                        Ok(SalvageOutcome::Unusable) => {
                            let reason = format!("unrecoverable ({e}); nothing salvageable");
                            set_aside_path(
                                &folder_fs,
                                &table_path,
                                &reason,
                                &mut unreadable_files,
                                &mut discard_after_commit,
                            );
                        }
                        Ok(SalvageOutcome::PunchedBoundLost) => {
                            // The flag decides this within THIS run: with
                            // resurrection on the same source salvages and keeps
                            // its readable region.
                            let reason = format!(
                                "punched SST with no recoverable restriction bound \
                                 (missing / corrupt sidecar and failed recovery, punched \
                                 extents found during salvage): {e}; a resurrection repair \
                                 keeps its readable region instead"
                            );
                            set_aside_path(
                                &folder_fs,
                                &table_path,
                                &reason,
                                &mut unreadable_files,
                                &mut discard_after_commit,
                            );
                        }
                        // Transient I/O salvage failure: nothing moved, so the
                        // retry re-derives the same salvage from the untouched
                        // source. A structural failure is recorded.
                        Err(salvage_err) if is_environmental(&salvage_err) => {
                            return Err(salvage_err);
                        }
                        Err(salvage_err) => {
                            let reason =
                                format!("recovery failed ({e}); salvage failed ({salvage_err})");
                            set_aside_path(
                                &folder_fs,
                                &table_path,
                                &reason,
                                &mut unreadable_files,
                                &mut discard_after_commit,
                            );
                        }
                    }
                }
                Err(e) => {
                    // A TRANSIENT recovery failure (Io) is retryable: recording it
                    // unreadable commits a manifest without the still-in-place
                    // file, which the next open's orphan cleanup then DELETES —
                    // permanent loss from a one-shot read failure. Propagate it so
                    // a retry re-recovers the table; only a structural failure is a
                    // genuine unreadable report.
                    if is_environmental(&e) {
                        return Err(e);
                    }
                    // The rebuilt manifest omits this file, so it is removed once
                    // that manifest is durable: a file both omitted and left in
                    // place is an orphan the next open must sweep, and an open
                    // that cannot sweep it fails.
                    let reason = e.to_string();
                    set_aside_path(
                        &folder_fs,
                        &table_path,
                        &reason,
                        &mut unreadable_files,
                        &mut discard_after_commit,
                    );
                }
            }
        }
    }

    // A pooled file's id is above every table written into its folder: a
    // claim takes the lowest pooled id, and a write numbered above an id
    // passes over the pooled ones below it. A file being written when the
    // power went may sit below a table another write finished first; that
    // table lists it among the files its folder was still writing (see
    // `file_pool::Writing`). So an empty file is a table that lost its bytes
    // only when a table of its folder (one that opened as a table, kept, set
    // aside or left out as a duplicate, or a salvaged copy) is above it and
    // was finished after it was written: reported as a loss nothing can
    // scope. Any other is a pooled file or an output the power loss cut
    // short, never a table: it holds no rows and is no loss, and like every
    // file the rebuilt manifest does not name it is reported left out and
    // goes once that manifest is durable. A file with bytes that opens as no
    // table is not counted: nothing says a table was ever written there.
    //
    // A table emptied to nothing with no table of its folder finished after
    // it reads as pooled here: nothing on disk tells the two apart. A table
    // is synced before any edit names it, so only damage outside a crash
    // empties one; a durable mark on every pooled file would cost a data sync
    // per file, the sync the pool removes, and reporting every empty file as
    // a loss would call for a full replay on every repair of a pooled tree.
    for (table_id, folder, table_path, folder_fs) in empty_unnamed {
        let written = written_before_one_of(
            table_id,
            recovered_by_id
                .values()
                .filter(|c| c.path.parent() == Some(folder.as_path()))
                .map(|c| (c.table.id(), c.table.metadata.writing_below.as_slice()))
                .chain(
                    opened_tables
                        .get(&folder)
                        .into_iter()
                        .flatten()
                        .map(|(id, below)| (*id, below.as_slice())),
                ),
        );
        if written {
            scanned_table_ids.insert(table_id);
            set_aside_path(
                &folder_fs,
                &table_path,
                "an empty table file below a table finished after it: its bytes were lost",
                &mut unreadable_files,
                &mut discard_after_commit,
            );
        } else {
            let reason = "an empty table file no table was written into";
            excluded_files.push((table_path.clone(), reason.to_string()));
            discard_after_commit.push((folder_fs, table_path, reason.to_string()));
        }
    }

    Ok(TableScan {
        recovered_by_id,
        unreadable_files,
        redundant_unreadable,
        excluded_files,
        coverage_by_path,
        discard_after_commit,
        swap_after_commit,
        live_sidecars,
        scanned_table_ids,
    })
}

/// Phase 2 of [`repair_tree`]: build a manifest from what the scan recovered,
/// commit it, and carry out exactly the removals and swaps that commit
/// authorizes.
///
/// The caller holds the directory lock for the whole of this.
///
/// # Errors
///
/// Propagates the commit failure. A failure AFTER the commit surfaces as
/// [`crate::Error::RepairedButUnopened`] carrying the finished report — the
/// repair is durable, so the obligation it names must not be lost.
#[cfg(feature = "std")]
fn rebuild_from_scan(
    config: &Config,
    allow_resurrection: bool,
    manifest_referenced: Option<CommittedManifest>,
    scan: TableScan,
    first_tree_id: crate::tree::inner::TreeId,
    outstanding: Option<RepairReport>,
    #[cfg(zstd_any)] damaged_dictionaries: &[crate::file::DictId],
) -> crate::Result<RepairReport> {
    let TableScan {
        recovered_by_id,
        mut unreadable_files,
        redundant_unreadable,
        mut excluded_files,
        coverage_by_path,
        mut discard_after_commit,
        mut swap_after_commit,
        live_sidecars,
        scanned_table_ids,
    } = scan;

    // Collect the best copy per id, carrying each candidate's completeness so
    // `salvaged` can be derived from the tables that actually make the
    // manifest — after the blob-dependency filtering below, not before (a
    // salvaged table dropped for an unrecoverable blob dependency must not
    // count, or `salvaged` could exceed `recovered`). A lossy copy superseded
    // by a complete duplicate is likewise already gone from the candidates.
    // Each entry keeps its SOURCE path (the scanned `{id}` file, not a
    // salvage replacement's `{id}.repair-tmp`): it is the key into
    // `coverage_by_path` when a kept lossy copy's loss is scoped below.
    let mut recovered_tables: Vec<(Table, Fidelity, PathBuf, Arc<dyn crate::fs::Fs>)> =
        recovered_by_id
            .into_values()
            .map(|c| (c.table, c.fidelity, c.path, c.fs))
            .collect();

    // Newest first, by DESCENDING recency key. For a flush table that key is
    // an id it took from the counter when it started (flushes are serialized,
    // so a higher one is later content); its own id is not, since a pooled
    // file's id was taken with its batch. A COMPACTION output's own id is NOT
    // that signal either: the id is taken when the compaction starts writing,
    // while newer flushes can install first, and an intra-L0 output is
    // appended at the BACK of L0 regardless of its id — so outputs persist
    // their newest INPUT's recency in meta (`recency`) and sort by it, with id
    // as the tie-break (equal recency means one descends from the other, and
    // an output's id is above every input's, so the higher id is the
    // superseding copy). A table's
    // highest seqno is NOT the signal either: callers may assign seqnos
    // explicitly, so an older table can top out above a newer one on an
    // unrelated key. The key is total, so repeating a repair over the same
    // files reproduces the same tree instead of inheriting the directory
    // scan's order.
    recovered_tables
        .sort_by_key(|(t, ..)| (std::cmp::Reverse(t.l0_recency()), std::cmp::Reverse(t.id())));

    // KV-separated (blob) trees additionally carry a blob-file list. Discover the
    // blob files from the `blobs/` folder (no manifest to filter against) and
    // record them in the rebuilt manifest with the matching `TreeType::Blob` so
    // the tree reopens (the reopened tree's type must match its config's
    // `kv_separation_opts`). Fragmentation stats are NOT reconstructable from a
    // directory scan (they are derived from compaction history), so they start
    // empty: blob GC is advisory and re-learns them over time. The empty start
    // never drops live data; it only resets GC's view of reclaimable space.
    //
    // Runs BEFORE the L0 runs are built: a table whose indirections point into a
    // blob file this scan could NOT recover must not be published (see the
    // dependency check below), so the surviving blob ids have to be known first.
    let mut blob_rewrites: crate::HashMap<
        crate::vlog::BlobFileId,
        crate::salvage::BlobFileRewrite,
    > = crate::HashMap::default();
    // Fresh-id blob replacements this run publishes; removed on ANY exit —
    // an error, a cancellation — before the manifest commit disarms the
    // guard (see `PublishedBlobReplacements`).
    let mut published_blob_replacements = PublishedBlobReplacements::new(config, first_tree_id);
    let blob_files_salvaged: Vec<(PathBuf, String)> = Vec::new();
    let mut blob_frag = crate::blob_tree::FragmentationMap::default();
    // Damaged blob originals whose replacement is in the rebuilt manifest.
    // Set aside only after `persist_version` — see `BlobRecovery::stale`.
    let mut stale_blob_originals: Vec<(PathBuf, String)> = Vec::new();
    // Blob files the rebuilt manifest omits because nothing references them.
    // Removed only after `persist_version`, for the same reason.
    let mut unreferenced_blob_files: Vec<PathBuf> = Vec::new();
    let (tree_type, mut blob_file_list) = if config.kv_separation_opts.is_some() {
        if let Some(p) = &config.recovery_progress {
            p.set_phase(crate::RecoveryPhase::RecoveringBlobFiles);
        }
        // The candidate reference set, taken from the tables recovered so far
        // (a conservative superset of the post-filter truth): lets the blob
        // scan skip salvaging invalid files nothing can reach, and reserves
        // every referenced id in the fresh-id allocator. A table whose
        // reference section is structurally unreadable contributes nothing
        // here rather than aborting the whole repair — the dependency filter
        // below sets exactly that table aside (its dependencies are
        // unknowable), so nothing that survives can point at an id this set
        // missed. Environmental errors still propagate: a transient read
        // failure says nothing about the section.
        let mut referenced_blob_ids: crate::HashSet<crate::vlog::BlobFileId> =
            crate::HashSet::default();
        for (table, ..) in &recovered_tables {
            match table.list_blob_file_references() {
                Ok(links) => {
                    for link in links.into_iter().flatten() {
                        referenced_blob_ids.insert(link.blob_file_id);
                    }
                }
                Err(e) if is_environmental(&e) => return Err(e),
                Err(e) => {
                    log::warn!(
                        "repair: table {} has an unreadable blob-reference section \
                         ({e}); the dependency filter will set it aside",
                        table.id(),
                    );
                }
            }
        }
        let recovery = recover_blob_files(
            config,
            &mut published_blob_replacements,
            &referenced_blob_ids,
            manifest_referenced.as_ref().map(|m| &m.blob_frontiers),
            manifest_referenced.as_ref().map(|m| &m.blob_checksums),
        )?;
        unreadable_files.extend(recovery.unreadable);
        excluded_files.extend(recovery.excluded);
        blob_rewrites = recovery.rewrites;
        blob_frag = recovery.frag;
        stale_blob_originals = recovery.stale;
        // Foreign names, duplicates and unreadable blob files: same post-commit
        // removal as the superseded originals (see `BlobRecovery::discard`).
        discard_after_commit.extend(
            recovery
                .discard
                .into_iter()
                .map(|(path, note)| (Arc::clone(&config.fs), path, note)),
        );
        let map: crate::HashMap<crate::vlog::BlobFileId, crate::vlog::BlobFile> =
            recovery.files.into_iter().map(|bf| (bf.id(), bf)).collect();
        (TreeType::Blob, BlobFileList::new(map))
    } else {
        // A Standard rebuild must not swallow a blob tree: with the manifest
        // gone the open's tree-type check never ran, so a configuration that
        // omits kv-separation would otherwise publish a manifest that opens
        // fine while reads return encoded indirection handles as user values
        // and the blob files stay orphaned. The recovered tables' own
        // `linked_blob_files` sections prove the store's type; fail closed
        // with the same mismatch error a healthy open raises. A table whose
        // section cannot be READ proves nothing either way — publishing it
        // would be permission granted by ignorance, so it is set aside like
        // the kv path's dependency filter does (environmental errors
        // propagate for a retry).
        let candidates = core::mem::take(&mut recovered_tables);
        recovered_tables.reserve(candidates.len());
        for candidate in candidates {
            let is_blob_backed = match candidate.0.list_blob_file_references() {
                Ok(refs) => refs.is_some_and(|r| !r.is_empty()),
                Err(e) if is_environmental(&e) => return Err(e),
                Err(e) => {
                    let (table, ..) = candidate;
                    set_aside_table(
                        table,
                        &format!(
                            "blob-file reference list unreadable ({e}) on a standard \
                             rebuild; the table cannot prove its tree type"
                        ),
                        &mut unreadable_files,
                        &mut discard_after_commit,
                    );
                    continue;
                }
            };
            if is_blob_backed {
                log::error!(
                    "repair: table {} references blob files, so this store is a \
                     KV-separated (blob) tree; rebuilding a Standard manifest \
                     over it would return indirection handles as values — \
                     configure kv separation and retry",
                    candidate.0.id(),
                );
                return Err(crate::Error::TreeTypeMismatch {
                    requested: TreeType::Standard,
                    actual: TreeType::Blob,
                });
            }
            recovered_tables.push(candidate);
        }
        (
            TreeType::Standard,
            BlobFileList::new(crate::HashMap::default()),
        )
    };

    // Drop any recovered table that still references a blob file the scan could
    // not recover: publishing the pair yields a manifest that opens fine while
    // a read of an affected key resolves a handle into a blob file that is not
    // there. A table whose `linked_blob_files` section cannot be read is treated
    // the same way (its dependencies are unknown, so it cannot be proven safe).
    // A table referencing a RESHAPED blob file — one the blob scan salvaged
    // into a compacted copy, or recovered with a punched frontier — is instead
    // REWRITTEN through the salvage pipeline: its handles are re-targeted at
    // the relocated records and only entries whose record no longer exists are
    // dropped, so intact live data is never discarded over a reshaped
    // dependency.
    // Shared with the ownership pass below, which rewrites tables again.
    let blob_rewrites = Arc::new(blob_rewrites);
    if config.kv_separation_opts.is_some() {
        // Frontiers of the punched-but-intact blob files (the `DropBelow`
        // rewrite entries): a handle below one dereferences zeroed bytes.
        // Empty on the common path, so no table's handles are scanned.
        let punched_frontiers: crate::HashMap<crate::vlog::BlobFileId, u64> = blob_rewrites
            .iter()
            .filter_map(|(id, rw)| match rw {
                crate::salvage::BlobFileRewrite::DropBelow(f) => Some((*id, *f)),
                crate::salvage::BlobFileRewrite::Remap { .. } => None,
            })
            .collect();
        let mut admitted: Vec<AdmittedTable> = Vec::with_capacity(recovered_tables.len());
        for (table, fidelity, source_path, source_fs) in recovered_tables {
            // One reference read drives everything below: the missing-id check
            // and the rewrite decision.
            let links = match table.list_blob_file_references() {
                Ok(links) => links,
                Err(e) if is_environmental(&e) => return Err(e),
                Err(e) => {
                    set_aside_table(
                        table,
                        &format!("blob-file reference list unreadable ({e})"),
                        &mut unreadable_files,
                        &mut discard_after_commit,
                    );
                    continue;
                }
            };
            // A referenced id is unrecoverable only when the scan neither kept
            // it NOR salvaged it into a replacement: a `Remap` entry means the
            // records live on under a fresh id, and the rewrite below
            // retargets this table's handles at it.
            if let Some(l) = links.as_ref().and_then(|links| {
                links.iter().find(|l| {
                    !blob_file_list.contains_key(l.blob_file_id)
                        && !blob_rewrites.contains_key(&l.blob_file_id)
                })
            }) {
                set_aside_table(
                    table,
                    &format!("blob file {} is not recoverable", l.blob_file_id),
                    &mut unreadable_files,
                    &mut discard_after_commit,
                );
                continue;
            }
            // Whether this table's handles must be rewritten: any reference to
            // a SALVAGED blob file (its records moved to a fresh id), or a
            // handle that actually lies below a punched blob's frontier (a
            // pre-relocation SST file left behind by a crash — the id-presence
            // check cannot see it).
            let mut needs_rewrite = false;
            if let Some(links) = &links {
                if links.iter().any(|l| {
                    matches!(
                        blob_rewrites.get(&l.blob_file_id),
                        Some(crate::salvage::BlobFileRewrite::Remap { .. })
                    )
                }) {
                    needs_rewrite = true;
                } else if !punched_frontiers.is_empty()
                    && links
                        .iter()
                        .any(|l| punched_frontiers.contains_key(&l.blob_file_id))
                {
                    match handle_below_blob_frontier(&table, &punched_frontiers) {
                        Ok(hit) => needs_rewrite = hit.is_some(),
                        Err(e) if is_environmental(&e) => return Err(e),
                        Err(e) => {
                            set_aside_table(
                                table,
                                &format!("blob handles unreadable ({e})"),
                                &mut unreadable_files,
                                &mut discard_after_commit,
                            );
                            continue;
                        }
                    }
                }
            }
            admitted.push(((table, fidelity, source_path, source_fs), needs_rewrite));
        }
        // Derived from the rows themselves, every time: a table file that is
        // simply gone takes its owners with it and leaves no trace the scan
        // could gate this on.
        // A row the walk cannot read leaves it unknown whether an object it
        // borrows has an owner elsewhere, and two owners would charge the
        // object twice: no ownership is handed then.
        let promotions = match owner_promotions(&admitted, &blob_rewrites) {
            Ok(promotions) => promotions,
            Err(e) if is_environmental(&e) => return Err(e),
            Err(e) => {
                log::warn!("repair: no blob object ownership handed on: a table's rows ({e})");
                admitted.iter().map(|_| None).collect()
            }
        };
        let mut kept: Vec<(Table, Fidelity, PathBuf, Arc<dyn crate::fs::Fs>)> =
            Vec::with_capacity(admitted.len());
        for (((table, fidelity, source_path, source_fs), needs_rewrite), promote) in
            admitted.into_iter().zip(promotions)
        {
            if !needs_rewrite && promote.is_none() {
                kept.push((table, fidelity, source_path, source_fs));
                continue;
            }
            // Rewrite through the salvage pipeline: re-emit every entry with
            // re-targeted handles, dropping only entries whose blob record no
            // longer exists. The rewritten table counts as salvaged (its
            // content may be lossy relative to the original).
            //
            // The copy is built at `{id}.repair-tmp` — a name no scan adopts —
            // and swapped onto `{id}` only AFTER the manifest commit. Displacing
            // the source first would open a window in which a crash leaves no
            // readable copy where the retry looks, and publishing the copy under
            // a name the scan DOES adopt would leave a crash with both readable,
            // so the retry would rebuild one history into L0 twice. Nothing is
            // carried across runs: the rewrite reads an untouched source, and a
            // leftover temp is garbage the retry replaces.
            let source_id = table.id();
            let path = (*table.path).clone();
            let output_path = repair_tmp_path(&path);
            // A RESTRICTED source's bound must be re-imposed on the copy: the
            // rewrite re-emits the straddling block's sub-bound rows, which the
            // restriction hides, so an unrestricted copy would resurrect them.
            let restrict_bound = table.restrict_lower_bound().cloned();
            // The source already carries its recovered ingest offset and L0
            // recency (from a clean manifest record, or the scan's own
            // admission); the rewrite preserves local seqnos and content, so
            // the copy reopens at the same position. ZERO is a VALID allocated
            // offset — the first ingestion on a fresh counter commits offset
            // 0 — not a sentinel for absence, so it is forwarded
            // unconditionally: a table that reached this point was already
            // admitted by the scan, and re-entering the fail-closed offset
            // exclusion here would reject (and delete) its healthy replacement.
            let source_position = ManifestPosition::of_table(&table);
            let fs = table.fs.clone();
            // A table rewritten only to hand ownership loses nothing: it stays
            // as it was, its owners unchanged, if the rewrite cannot be made.
            let kept_as_is = if needs_rewrite {
                // Its descriptor too: a replacement may be renamed over below,
                // which a backend refuses for an open file (Windows).
                release_held(config, table);
                None
            } else {
                Some(table)
            };
            let fidelity = if needs_rewrite {
                Fidelity::Salvaged
            } else {
                fidelity
            };
            match try_salvage_table(
                config,
                &fs,
                allow_resurrection,
                TableSalvage {
                    source: &path,
                    table_path: &output_path,
                    table_id: source_id,
                    reject_punched_without_bound: false,
                    blob_rewrite: Some(Arc::clone(&blob_rewrites)),
                    owner_promotions: promote.map(Arc::new),
                    recovered_position: Some(source_position),
                },
            ) {
                Ok(SalvageOutcome::Salvaged(rewritten)) => {
                    if let Some(table) = kept_as_is {
                        release_held(config, table);
                    }
                    let rewritten = restrict_salvaged_output(
                        &*fs,
                        config,
                        &output_path,
                        rewritten,
                        restrict_bound,
                        allow_resurrection,
                    )?;
                    let restricted = rewritten.restrict_lower_bound().is_some();
                    if let Some(pending) = swap_after_commit
                        .iter_mut()
                        .find(|(_, tmp, ..)| *tmp == path)
                    {
                        // A source that is itself a replacement this repair
                        // built is superseded by the copy, which takes the
                        // replacement's name before the manifest can name it:
                        // a crash between the commit and the swap then leaves
                        // the copy the manifest names where the next open
                        // finishes swaps from, not under a name no scan reads.
                        // The replacement is no longer needed: a failed commit
                        // is retried from the untouched original.
                        let checksum = rewritten.checksum();
                        let bound = rewritten.restrict_lower_bound().cloned();
                        release_held(config, rewritten);
                        commit_repair_tmp(&*fs, &output_path, &path, config.sync_mode, restricted)?;
                        let published = Table::recover(held_recover_params(
                            config,
                            path.clone(),
                            checksum,
                            source_id,
                            Arc::clone(&fs),
                            Some(source_position),
                        ))?;
                        let published = match bound {
                            Some(bound) => {
                                published.reopen_restricted_with(bound, held_descriptors(config))?
                            }
                            None => published,
                        };
                        pending.3 = restricted;
                        kept.push((published, fidelity, source_path, source_fs));
                    } else {
                        swap_after_commit.push((Arc::clone(&fs), output_path, path, restricted));
                        kept.push((rewritten, fidelity, source_path, source_fs));
                    }
                }
                // A retryable failure leaves the source where it was found, so
                // the retry re-derives the same rewrite from it; nothing to
                // restore.
                Err(e) if is_environmental(&e) => return Err(e),
                Ok(_) | Err(_) if kept_as_is.is_some() => {
                    if let Some(table) = kept_as_is {
                        log::warn!(
                            "repair: table {source_id} keeps the objects it borrows unowned: \
                             its rewrite produced nothing usable"
                        );
                        discard_unreferenced(&*fs, &output_path, config.sync_mode)?;
                        kept.push((table, fidelity, source_path, source_fs));
                    }
                }
                Ok(SalvageOutcome::Unusable | SalvageOutcome::PunchedBoundLost) => {
                    set_aside_path(
                        &fs,
                        &path,
                        "blob-handle rewrite produced nothing",
                        &mut unreadable_files,
                        &mut discard_after_commit,
                    );
                }
                Err(e) => {
                    set_aside_path(
                        &fs,
                        &path,
                        &format!("blob-handle rewrite failed ({e})"),
                        &mut unreadable_files,
                        &mut discard_after_commit,
                    );
                }
            }
        }
        recovered_tables = kept;
    }

    // A crashed compaction that FINALIZED its outputs before its version edit
    // committed leaves BOTH histories on disk — and so does a committed
    // compaction whose input deletion never finished. The recorded lineage
    // settles it: an output whose inputs ALL survived into this rebuild is
    // DERIVED — the inputs are the complete history, a future compaction
    // re-folds them — and publishing both would apply the same merge
    // operands twice on every read. Exclusion cascades correctly in a single
    // pass against the PRE-exclusion id set: an intermediate output excluded
    // for its surviving inputs may itself be named in a later output's
    // lineage, but every exclusion keeps the excluded table's own inputs, so
    // the remaining set always bottoms out at lineage-less leaves carrying
    // the full history. An output with PARTIALLY surviving inputs is KEPT —
    // it is the only complete copy of its span — alongside the surviving
    // inputs, and each overlap is reported below as ambiguity coverage so a
    // reconciling deployment knows reads in that span may double-apply
    // operands. Presence is judged AFTER the blob-dependency filter above: an
    // input dropped for an unrecoverable blob does not count as surviving —
    // and only a COMPLETE recovery counts at all: a salvaged or
    // geometry-restricted input already lost records, so its id proves
    // nothing about the output's contents surviving elsewhere, and trading a
    // healthy output for damaged inputs would convert recoverable data into
    // permanent loss.
    let present_ids: crate::HashSet<TableId> = recovered_tables
        .iter()
        .filter(|(_, fidelity, ..)| fidelity.is_complete())
        .map(|(t, ..)| t.id())
        .collect();
    // Per-candidate geometry and ancestry, captured over EVERY candidate
    // before any exclusion: the union pass and the transitive-supersession
    // closure below must reason about tables the first pass already dropped
    // (an excluded intermediate output still stands between a retained
    // descendant and its ultimate inputs).
    let all_ranges: crate::HashMap<TableId, (UserKey, UserKey, SeqNo)> = recovered_tables
        .iter()
        .map(|(t, ..)| {
            (
                t.id(),
                (
                    t.metadata.key_range.min().clone(),
                    t.metadata.key_range.max().clone(),
                    t.get_highest_seqno(),
                ),
            )
        })
        .collect();
    let lineage_by_id: crate::HashMap<TableId, Vec<TableId>> = recovered_tables
        .iter()
        .filter_map(|(t, ..)| t.metadata.lineage.clone().map(|l| (t.id(), l)))
        .collect();
    // The final `bool` records whether the overlapping OUTPUT was
    // transformed: its filter may have removed keys inside the overlap, so
    // the kept input's copies are not byte-identical duplicates there.
    let mut lineage_partial: Vec<(TableId, PathBuf, UserKey, UserKey, Option<SeqNo>, bool)> =
        Vec::new();
    let mut inputs_superseded: crate::HashSet<TableId> = crate::HashSet::default();
    // Files excluded as REDUNDANT (a derived output whose inputs all
    // survived, an input a complete output fully covers): their content
    // lives on in the kept tables, so they join `excluded_files` with NO
    // `lost_coverage` (nothing was lost). The id set feeds the ancestry
    // walk below.
    let mut redundant_excluded_ids: crate::HashSet<TableId> = crate::HashSet::default();
    {
        let candidates = core::mem::take(&mut recovered_tables);
        for candidate in candidates {
            let lineage = candidate.0.metadata.lineage.clone();
            match lineage.as_deref() {
                // A TRANSFORMED output never trades back for its inputs: a
                // compaction filter removed records from its window, so the
                // resurrected inputs would revive them. It takes the partial
                // arm below instead and SUPERSEDES the inputs it covers.
                Some(inputs)
                    if !candidate.0.metadata.lineage_transformed
                        && !inputs.is_empty()
                        && inputs
                            .iter()
                            .all(|id| *id != candidate.0.id() && present_ids.contains(id)) =>
                {
                    let (table, _, path, fs) = candidate;
                    let table_id = table.id();
                    log::info!(
                        "repair: table {table_id} is a compaction output whose inputs \
                         {inputs:?} all survived; excluding the derived copy so its \
                         merge operands are not applied twice",
                    );
                    drop(table);
                    let reason = "derived output of an uncommitted compaction whose inputs \
                                  all survived; excluded so its merge operands are not \
                                  applied twice";
                    redundant_excluded_ids.insert(table_id);
                    excluded_files.push((path.clone(), reason.to_string()));
                    discard_after_commit.push((fs, path, reason.to_string()));
                }
                Some(inputs) => {
                    // Partial survival: the output is the only complete copy
                    // of the LOST inputs' span, so it must be kept — and each
                    // surviving input the output's key range fully COVERS is
                    // superseded by it (the compaction read the whole input,
                    // so every one of its records is in a range-covering
                    // output) and is excluded, or reads in the overlap apply
                    // the same merge operands twice; `lost_coverage` cannot
                    // fix that, since the documented reconciliation would see
                    // the duplicated operand as surviving and subtract its
                    // WAL record instead of removing the folded copy. Only a
                    // COMPLETE output supersedes. An input the output covers
                    // partially (its remainder belonged to a LOST sibling
                    // output of the same run) is kept — dropping it would
                    // lose live records — and the residual overlap is
                    // reported.
                    let cmp = config.comparator.as_ref();
                    for input in inputs {
                        if *input == candidate.0.id() {
                            continue;
                        }
                        let Some((in_min, in_max, in_hi)) = all_ranges.get(input) else {
                            continue;
                        };
                        let out_range = &candidate.0.metadata.key_range;
                        if cmp.compare(out_range.min(), in_max) == core::cmp::Ordering::Greater
                            || cmp.compare(in_min, out_range.max()) == core::cmp::Ordering::Greater
                        {
                            continue;
                        }
                        let covers = candidate.1.is_complete()
                            && cmp.compare(out_range.min(), in_min) != core::cmp::Ordering::Greater
                            && cmp.compare(in_max, out_range.max()) != core::cmp::Ordering::Greater;
                        if covers {
                            inputs_superseded.insert(*input);
                            continue;
                        }
                        let lo =
                            if cmp.compare(out_range.min(), in_min) == core::cmp::Ordering::Less {
                                in_min
                            } else {
                                out_range.min()
                            };
                        let hi =
                            if cmp.compare(out_range.max(), in_max) == core::cmp::Ordering::Less {
                                out_range.max()
                            } else {
                                in_max
                            };
                        lineage_partial.push((
                            *input,
                            candidate.2.clone(),
                            lo.clone(),
                            hi.clone(),
                            Some(candidate.0.get_highest_seqno().min(*in_hi)),
                            candidate.0.metadata.lineage_transformed,
                        ));
                    }
                    recovered_tables.push(candidate);
                }
                _ => recovered_tables.push(candidate),
            }
        }
    }
    // Sibling-UNION supersession: the outputs of one rotated run record the
    // same lineage plus an adjacency link (`lineage_prev`), so an UNBROKEN
    // surviving chain of COMPLETE outputs proves its combined range has no
    // gap a lost sibling could hide — an input inside that union is fully
    // redundant even when no single output contains it. A broken chain
    // (a predecessor that did not survive) proves nothing and unions
    // nothing. A chain that spans the run END TO END — from the FIRST
    // output (no `lineage_prev`) to the one carrying the `lineage_last`
    // marker — is the run's COMPLETE output set and supersedes every input
    // outright: keys outside the written ranges were consumed by the merge
    // (obsolete versions, annihilated weak deletes, filter removals), not
    // lost. That is what preserves a committed TRANSFORMING compaction —
    // its outputs no longer contain the filtered records, so only full-run
    // supersession retires the inputs that still do.
    {
        let mut by_id: crate::HashMap<TableId, usize> = crate::HashMap::default();
        for (idx, (t, fidelity, ..)) in recovered_tables.iter().enumerate() {
            if fidelity.is_complete() && t.metadata.lineage.is_some() {
                by_id.insert(t.id(), idx);
            }
        }
        let same_lineage = |a: usize, b: usize| {
            recovered_tables.get(a).map(|(t, ..)| &t.metadata.lineage)
                == recovered_tables.get(b).map(|(t, ..)| &t.metadata.lineage)
        };
        let mut next_of: crate::HashMap<TableId, TableId> = crate::HashMap::default();
        for &idx in by_id.values() {
            let Some((t, ..)) = recovered_tables.get(idx) else {
                continue;
            };
            if let Some(prev) = t.metadata.lineage_prev
                && by_id.get(&prev).is_some_and(|&p| same_lineage(p, idx))
            {
                next_of.insert(prev, t.id());
            }
        }
        let cmp = config.comparator.as_ref();
        for (&head_id, &head_idx) in &by_id {
            let Some((head, ..)) = recovered_tables.get(head_idx) else {
                continue;
            };
            // A chain HEAD: its predecessor is absent from the surviving set
            // (or it is the run's first output).
            if head
                .metadata
                .lineage_prev
                .is_some_and(|prev| by_id.contains_key(&prev))
            {
                continue;
            }
            // Walk the unbroken chain; ranges follow the run's key order, so
            // the union is [head.min, tail.max]. A chain that runs from the
            // run's FIRST output (no `lineage_prev`) to its LAST (the
            // `lineage_last` marker, written only by a writer that owned the
            // WHOLE run — never by a parallel or tight-space slice, whose
            // first output also has no persisted predecessor) is the
            // COMPLETE output set: it supersedes every input outright, no
            // range check needed. Requiring BOTH ends keeps a surviving
            // slice from claiming the whole run.
            let open_lo = head.metadata.lineage_prev.is_none();
            let union_min = head.metadata.key_range.min().clone();
            let mut union_max = head.metadata.key_range.max().clone();
            let mut open_hi = head.metadata.lineage_last;
            let mut cursor = head_id;
            while let Some(&next) = next_of.get(&cursor) {
                if let Some(&next_idx) = by_id.get(&next)
                    && let Some((t, ..)) = recovered_tables.get(next_idx)
                {
                    union_max = t.metadata.key_range.max().clone();
                    open_hi = t.metadata.lineage_last;
                }
                cursor = next;
            }
            let complete_run = open_lo && open_hi;
            let Some(inputs) = &head.metadata.lineage else {
                continue;
            };
            for input in inputs {
                if by_id.contains_key(input) {
                    continue;
                }
                // `all_ranges`, not the surviving set: superseding an id the
                // first pass already excluded is what lets the ancestry
                // closure below reach ITS inputs.
                let Some((in_min, in_max, _)) = all_ranges.get(input) else {
                    continue;
                };
                if complete_run
                    || (cmp.compare(&union_min, in_min) != core::cmp::Ordering::Greater
                        && cmp.compare(in_max, &union_max) != core::cmp::Ordering::Greater)
                {
                    inputs_superseded.insert(*input);
                }
            }
        }
    }

    // Transitive-supersession closure over the lineage edges: a superseded
    // table's content is PROVEN incorporated into a retained output, and it
    // in turn incorporated its own inputs — so those are redundant too,
    // even when the retaining output's lineage never names them (it names
    // the intermediate). Records an intermediate fold dropped (obsolete
    // versions, annihilated weak-delete pairs) return shadowed or
    // re-annihilating, and filter removals stay removed with their carrier,
    // so following the edge loses nothing live. Follows SUPERSEDED ids
    // only, never exclusions: a derived output excluded in the first pass
    // was dropped because its inputs are the preferred history — walking
    // through it would discard the very tables that exclusion kept.
    {
        let mut worklist: Vec<TableId> = inputs_superseded.iter().copied().collect();
        while let Some(id) = worklist.pop() {
            let Some(inputs) = lineage_by_id.get(&id) else {
                continue;
            };
            for input in inputs {
                if all_ranges.contains_key(input) && inputs_superseded.insert(*input) {
                    worklist.push(*input);
                }
            }
        }
    }

    // Second pass: drop the inputs a surviving COMPLETE output fully covers.
    // Their residual-coverage entries (recorded against a DIFFERENT,
    // partially-covering output before the covering one was seen) go with
    // them — the ambiguity those entries reported no longer exists.
    if !inputs_superseded.is_empty() {
        let candidates = core::mem::take(&mut recovered_tables);
        for candidate in candidates {
            if inputs_superseded.contains(&candidate.0.id()) {
                let (table, _, path, fs) = candidate;
                log::info!(
                    "repair: table {} is fully covered by a surviving compaction \
                     output; excluding it so its merge operands are not applied twice",
                    table.id(),
                );
                drop(table);
                let reason = "input fully covered by a surviving compaction output; \
                              excluded so its merge operands are not applied twice";
                excluded_files.push((path.clone(), reason.to_string()));
                discard_after_commit.push((fs, path, reason.to_string()));
            } else {
                recovered_tables.push(candidate);
            }
        }
        lineage_partial.retain(|(input, ..)| !inputs_superseded.contains(input));
    }

    // A RESIDUAL overlap that survived every supersession pass fails closed
    // in two configurations. Against a TRANSFORMED output — always: the
    // kept input may hold a key the filter removed inside the overlap,
    // reads would resurrect it, and `lost_coverage` cannot repair the
    // deletion because a compaction-filter verdict is not an external-WAL
    // event. Under a merge operator — for any residual: the kept input's
    // records inside the overlap are also folded into the kept output, and
    // publishing both applies the same operands twice on every read — a
    // multiplicity the reported replay cannot remove, the same rule the
    // legacy ambiguity below takes. This is the crash shape a parallel
    // compaction leaves when input cleanup is partial (an input spanning
    // several sub-compaction ranges survives while no single output or
    // provable chain covers it whole), and equally a serial run's
    // lost-sibling window. Value-only residuals against UNTRANSFORMED
    // outputs proceed to the report: their duplicate records are
    // byte-identical and reads dedupe them.
    if let Some((input, output_path, ..)) = lineage_partial
        .iter()
        .find(|(.., transformed)| config.merge_operator.is_some() || *transformed)
    {
        log::error!(
            "repair: input table {} overlaps the surviving compaction output {} \
             that folded part of it, and no surviving output set covers the \
             input whole; publishing both would resurrect filter-removed keys \
             or double-apply merge operands, and no replay can undo either",
            input,
            output_path.display(),
        );
        return Err(crate::Error::Unrecoverable);
    }

    // Ancestry audit for every RETAINED transformed output: each id its
    // lineage names must be either superseded (its content provably
    // incorporated) or itself an excluded derived output whose OWN inputs
    // pass the same test — the exclusion moved the history onto them. A
    // live, unsuperseded carrier still holds rows the transform removed,
    // no range test can see that (the transform may have emptied the
    // overlap entirely), and no replay can re-delete them: fail closed.
    for (table, ..) in &recovered_tables {
        if !table.metadata.lineage_transformed {
            continue;
        }
        let Some(inputs) = &table.metadata.lineage else {
            continue;
        };
        let mut visited: crate::HashSet<TableId> = crate::HashSet::default();
        let mut worklist: Vec<TableId> = inputs.clone();
        while let Some(id) = worklist.pop() {
            if id == table.id() || !visited.insert(id) || inputs_superseded.contains(&id) {
                continue;
            }
            if redundant_excluded_ids.contains(&id) {
                // The excluded intermediate's history lives on in its own
                // inputs; audit those instead.
                if let Some(carried) = lineage_by_id.get(&id) {
                    worklist.extend(carried.iter().copied());
                }
                continue;
            }
            if all_ranges.contains_key(&id) {
                log::error!(
                    "repair: table {} carries history the transformed compaction \
                     output {} filtered, and no surviving output set proves that \
                     history incorporated; publishing both would resurrect the \
                     filter-removed rows, and no replay can re-delete them",
                    id,
                    table.id(),
                );
                return Err(crate::Error::Unrecoverable);
            }
        }
    }

    // Only blob files a surviving table REFERENCES go into the manifest.
    // An unreferenced one holds no reachable value by definition, and
    // admitting it would strand it there forever: repair cannot rebuild
    // fragmentation stats from a directory scan, and blob GC retires a
    // file only once its recorded stale bytes reach the totals it never
    // gets. That also settles what a crashed earlier attempt leaves
    // behind — a fully written salvage replacement that the crash kept
    // out of any manifest — which would otherwise be admitted as an
    // ordinary blob and pin a whole copy per failed attempt. Judged against
    // the FINAL table set — after the lineage dedup above — so a blob
    // referenced only by an excluded table (a derived output, a superseded
    // input) is not pinned by a reference the rebuilt manifest no longer
    // holds.
    if config.kv_separation_opts.is_some() {
        let mut referenced: crate::HashSet<crate::vlog::BlobFileId> = crate::HashSet::default();
        for (table, ..) in &recovered_tables {
            for link in table.list_blob_file_references()?.into_iter().flatten() {
                referenced.insert(link.blob_file_id);
            }
        }
        // Carry each dropped file's RECOVERED path, never a path rebuilt from
        // its id: the scan accepts a noncanonical spelling (`blobs/01` for id
        // 1), and removing the reconstructed `blobs/1` would answer `NotFound`
        // — reporting success while leaving the real file behind for the next
        // open to sweep, which then fails if that removal is refused.
        let dropped: Vec<(crate::vlog::BlobFileId, PathBuf)> = blob_file_list
            .iter()
            .filter(|bf| !referenced.contains(&bf.id()))
            .map(|bf| (bf.id(), bf.path().to_path_buf()))
            .collect();
        for (id, path) in dropped {
            log::debug!(
                "blob file {id} is referenced by no recovered table; leaving it out \
                 of the rebuilt manifest and removing it"
            );
            blob_file_list.remove(id);
            blob_frag.remove(&id);
            // Removed POST-COMMIT (see the sweep below): until the manifest is
            // durable the file is still what a failed attempt's inputs point
            // at, and a retry re-derives the same decision from a fresh scan.
            unreferenced_blob_files.push(path);
        }
    }

    // Live sidecars are judged here, against the FINAL rebuilt set: one whose
    // table survived is preserved (the open and the next scrub read it), an
    // ORPHAN — its table excluded, superseded, or never recovered — is
    // removed post-commit. The open sweeps such orphans unconditionally and
    // fails on a refused removal, so leaving one would let a successful
    // repair be followed by an open failure on the very same file.
    {
        let final_ids: crate::HashSet<TableId> =
            recovered_tables.iter().map(|(t, ..)| t.id()).collect();
        for (id, path, fs) in live_sidecars {
            if final_ids.contains(&id) {
                continue;
            }
            discard_after_commit.push((
                fs,
                path,
                "orphaned sidecar; its table did not survive the rebuild".to_string(),
            ));
        }
    }

    // `salvaged` is a subset of `recovered`, so derive it from the tables that
    // survived every filter above. The live progress counter follows the same
    // rule: a candidate displaced by deduplication or dropped by dependency
    // filtering never counts, so the snapshot cannot claim more tables than
    // the rebuilt manifest holds.
    // Only block-salvage REWRITES: a geometry-restricted original is lossy
    // (it contributes coverage below) but was never salvaged, and it arises
    // in plain repairs where `salvaged` is documented to stay zero.
    let salvaged = recovered_tables
        .iter()
        .filter(|(_, fidelity, ..)| *fidelity == Fidelity::Salvaged)
        .count();
    // A KEPT lossy copy — a salvaged rewrite that dropped corrupt blocks (or
    // blob records), or a geometry-restricted original whose derived bound
    // may exclude the straddling block's live suffix — names nothing in
    // `unreadable_files`, so without an entry here `wal_replay_scope()` would
    // answer `TailOnly` while older persisted changes were in fact lost. The
    // loss is scoped by the
    // SOURCE's coverage captured during the scan — NOT by the replacement's
    // own metadata, which only bounds what SURVIVED: salvage may have
    // dropped the block holding the source's outermost keys or its highest
    // seqno, and a range/ceiling derived from the survivors would exclude
    // exactly the lost part. A source whose metadata never parsed
    // (whole-file recovery failed before the coverage was captured) is
    // unscopable and joins `unknowable_losses` instead.
    let mut salvaged_unknowable: Vec<PathBuf> = Vec::new();
    let salvaged_coverage: Vec<(PathBuf, UserKey, UserKey, Option<SeqNo>)> = recovered_tables
        .iter()
        .filter(|(_, fidelity, ..)| !fidelity.is_complete())
        .filter_map(|(_, _, source_path, _)| {
            if let Some((lo, hi, bound)) = coverage_by_path.get(source_path) {
                Some((source_path.clone(), lo.clone(), hi.clone(), *bound))
            } else {
                salvaged_unknowable.push(source_path.clone());
                None
            }
        })
        .collect();
    // Every table carries its L0 recency key (a table without one is refused
    // as it is read), so the order rebuilt from it is the order the live tree
    // kept and ties in it need no report.
    let recovered_tables: Vec<Table> = recovered_tables.into_iter().map(|(t, ..)| t).collect();
    publish_repaired_manifest(
        config,
        RepairPublication {
            recovered_tables,
            tree_type,
            blob_file_list,
            blob_frag,
            published_blob_replacements,
            unreadable_files,
            redundant_unreadable,
            excluded_files,
            lost_coverage_scoped: (salvaged_coverage, lineage_partial, salvaged_unknowable),
            coverage_by_path,
            manifest_referenced,
            scanned_table_ids,
            discard_after_commit,
            swap_after_commit,
            stale_blob_originals,
            unreferenced_blob_files,
            blob_files_salvaged,
            salvaged,
            first_tree_id,
            outstanding,
            #[cfg(zstd_any)]
            damaged_dictionaries,
        },
    )
}

/// Everything [`publish_repaired_manifest`] needs: the tables that made it
/// through, and every list whose contents the commit either authorizes to be
/// carried out (removals, swaps) or must be reported.
#[cfg(feature = "std")]
struct RepairPublication<'a> {
    recovered_tables: Vec<Table>,
    tree_type: TreeType,
    blob_file_list: BlobFileList,
    blob_frag: crate::blob_tree::FragmentationMap,
    published_blob_replacements: PublishedBlobReplacements<'a>,
    unreadable_files: Vec<(PathBuf, String)>,
    /// Damaged files whose id is covered by a retained complete copy: reported
    /// as corruption, but contributing no lost coverage.
    redundant_unreadable: crate::HashSet<PathBuf>,
    excluded_files: Vec<(PathBuf, String)>,
    #[expect(clippy::type_complexity, reason = "the three coverage channels")]
    lost_coverage_scoped: (
        Vec<(PathBuf, UserKey, UserKey, Option<SeqNo>)>,
        Vec<(TableId, PathBuf, UserKey, UserKey, Option<SeqNo>, bool)>,
        Vec<PathBuf>,
    ),
    coverage_by_path: crate::HashMap<PathBuf, (UserKey, UserKey, Option<SeqNo>)>,
    manifest_referenced: Option<CommittedManifest>,
    scanned_table_ids: crate::HashSet<TableId>,
    discard_after_commit: Vec<(Arc<dyn crate::fs::Fs>, PathBuf, String)>,
    swap_after_commit: Vec<(Arc<dyn crate::fs::Fs>, PathBuf, PathBuf, bool)>,
    stale_blob_originals: Vec<(PathBuf, String)>,
    unreferenced_blob_files: Vec<PathBuf>,
    blob_files_salvaged: Vec<(PathBuf, String)>,
    salvaged: usize,
    /// The first tree id the repair handed out: the descriptors cached under
    /// it and every later id are closed before the post-commit swaps.
    first_tree_id: crate::tree::inner::TreeId,
    /// The report of an earlier repair over the installed snapshot that never
    /// reached the caller: its losses are owed as well.
    outstanding: Option<RepairReport>,
    /// Dictionaries found damaged at the start; those still damaged when the
    /// report is assembled are set aside after the commit.
    #[cfg(zstd_any)]
    damaged_dictionaries: &'a [crate::file::DictId],
}

/// Phase 3 of [`repair_tree`]: commit the rebuilt manifest, then carry out
/// exactly what that commit authorizes and assemble the report.
///
/// The commit is the pivot: before it the directory is untouched and any
/// failure leaves the retry the same bytes; after it the manifest is durable,
/// so a failure carries the finished report out through
/// [`crate::Error::RepairedButUnopened`] rather than discarding it.
///
/// # Errors
///
/// Propagates the commit failure, or wraps a post-commit one with the report.
#[cfg(feature = "std")]
fn publish_repaired_manifest(
    config: &Config,
    publication: RepairPublication<'_>,
) -> crate::Result<RepairReport> {
    let RepairPublication {
        recovered_tables,
        tree_type,
        blob_file_list,
        blob_frag,
        mut published_blob_replacements,
        unreadable_files,
        redundant_unreadable,
        excluded_files,
        lost_coverage_scoped: (salvaged_coverage, lineage_partial, salvaged_unknowable),
        coverage_by_path,
        manifest_referenced,
        scanned_table_ids,
        discard_after_commit,
        swap_after_commit,
        stale_blob_originals,
        unreferenced_blob_files,
        mut blob_files_salvaged,
        salvaged,
        first_tree_id,
        outstanding,
        #[cfg(zstd_any)]
        damaged_dictionaries,
    } = publication;
    // A rebuilt manifest whose highest table id is the LAST one cannot be
    // committed: the next open seeds its id allocator with `highest + 1`,
    // which overflows — a panic in checked builds, a wrap to an existing low
    // id in release builds (a later flush then collides with that file).
    // Mirrors the version-id and blob-id exhaustion guards: fail the repair
    // instead of publishing a tree that cannot allocate.
    if recovered_tables.iter().map(Table::id).max() == Some(crate::TableId::MAX) {
        log::error!(
            "repair: the table id space is exhausted (id {} is in use); a rebuilt \
             manifest would make the next open's id allocator overflow",
            crate::TableId::MAX,
        );
        return Err(crate::Error::Unrecoverable);
    }
    // The blob-file id space counts on the same rule. A HEALTHY,
    // still-referenced blob file at the last id needs no salvage, so the
    // fresh-id allocator's exhaustion (`next_blob_id == None`) is never
    // consulted for it — yet the next open seeds its blob allocator with
    // `max + 1` all the same, with the same overflow.
    if blob_file_list.iter().map(crate::vlog::BlobFile::id).max()
        == Some(crate::vlog::BlobFileId::MAX)
    {
        log::error!(
            "repair: the blob file id space is exhausted (id {} is referenced); a \
             rebuilt manifest would make the next open's blob id allocator overflow",
            crate::vlog::BlobFileId::MAX,
        );
        return Err(crate::Error::Unrecoverable);
    }
    if let Some(p) = &config.recovery_progress {
        p.tables_recovered_add(recovered_tables.len() as u64);
        // Blob files count on the same rule and for the same reason: the
        // reference filter above removes (and deletes) any the surviving tables
        // do not point at, so counting one at recovery time would claim a
        // recovery the rebuilt manifest does not hold.
        p.blob_files_recovered_add(blob_file_list.len() as u64);
    }

    // Each recovered table becomes its own single-table L0 run. L0 permits
    // overlapping runs, so this is always legal regardless of key overlap;
    // background compaction collapses them into sorted lower levels later.
    // `Run::new` only returns `None` for an empty run, which `vec![t]` never is,
    // so no table is dropped here — but build the runs explicitly and derive the
    // recovered count from what actually lands in the manifest, so the report
    // can never overcount relative to the persisted version.
    let l0_runs = recovered_tables
        .iter()
        .cloned()
        .filter_map(|t| Run::new(vec![t]).map(Arc::new))
        .collect::<Vec<_>>();
    let recovered = l0_runs.len();

    let mut levels = Vec::with_capacity(config.level_count.into());
    levels.push(Level::from_runs(l0_runs));
    for _ in 1..config.level_count {
        levels.push(Level::empty());
    }

    // Next version id after the highest existing one: recovery takes the
    // highest snapshot, so the rebuilt one must be above every other. Only a
    // folder already holding `v{u64::MAX}` has no such id. A rebuilt snapshot
    // high in the range needs no headroom for later edits: the next one comes
    // from an open, which renumbers any tree in the top half to version 0.
    let version_id = match highest_existing_version_id(&*config.fs, &config.path)? {
        Some(max) => max.checked_add(1).ok_or_else(|| {
            log::error!(
                "repair: v{max} exists, so no snapshot id is left above it for the \
                 rebuilt manifest",
            );
            crate::Error::Unrecoverable
        })?,
        None => 0,
    };

    // Seeded with the punched prefixes' garbage: those frames can never be
    // observed by a future compaction, so an empty map would pin every punched
    // file's stale count below its whole-file metadata totals forever.
    //
    // The retention floor comes from the caller (`Config::repair_retention_floor`).
    // The lost manifest was the only record of which snapshots a past GC
    // compaction or `clear` invalidated, and the tables cannot stand in for
    // it: a GC compaction zeroes the seqnos of the rows it settles, so a
    // table's highest seqno reads as `0` exactly on the trees where history
    // WAS collected. Nor may the floor be guessed high (say, at the highest
    // persisted seqno): the external-WAL reconciliation that follows a repair
    // restores intermediate snapshots and reads them back, so a guessed floor
    // would refuse history the caller has just made whole. Only the
    // deployment that ran the compactions knows the watermark; it supplies
    // it, and the default (`0`) serves everything, as before the floor
    // existed.
    let version = Version::from_levels(version_id, tree_type, levels, blob_file_list, blob_frag)
        .with_retention_floor(config.repair_retention_floor);

    // Register the dictionaries the recovered files name. A rebuilt version
    // starts with none, and nothing later puts them back once the write policy
    // stops naming them: a checkpoint copies exactly this list, so the snapshot
    // would carry the tables and not the dictionaries that decode them.
    //
    // Derived from the files themselves and intersected with what `dicts/`
    // actually holds, so a repair over a tree whose dictionary is genuinely
    // gone records no id it cannot honour.
    #[cfg(zstd_any)]
    let version = {
        let held = config.current_zstd_dictionaries();
        let ids: Vec<_> = version
            .referenced_dicts()
            .into_iter()
            .filter(|id| held.get(*id).is_some())
            .collect();

        // STORE what the manifest is about to name. The set here holds the
        // caller-supplied dictionary as well as the folder's, so a repair of a
        // tree written before dictionaries were stored (tables naming an id,
        // no `dicts/` at all) would otherwise commit a manifest referencing
        // bytes that exist only in the caller's memory — and the reopen the
        // repair exists to enable would fail on the missing file. Writing is
        // idempotent by id, so a dictionary already on disk costs an `exists`.
        let folder = config.path.join(crate::file::DICTS_FOLDER);
        for id in &ids {
            if let Some(dict) = held.get(*id) {
                crate::dicts::write(
                    &*config.fs,
                    &folder,
                    dict,
                    config.encryption.as_deref(),
                    config.sync_mode,
                )?;
            }
        }

        if ids.is_empty() {
            version
        } else {
            version.with_dicts(ids)
        }
    };

    // The LAST cancellation boundary: per-file checks only run before a file
    // starts, so a cancel requested during the final file's verification or
    // salvage would otherwise be silently outrun by the commit. From here on
    // the run is COMMITTING and then cleaning up, and cancellation is no
    // longer consulted (see `check_cancel`). An abort here — like every
    // earlier pre-commit exit — unwinds the fresh-id blob replacements
    // through the `published_blob_replacements` guard.
    check_cancel(config)?;
    if let Some(p) = &config.recovery_progress {
        p.set_phase(crate::RecoveryPhase::Committing);
    }

    let mut warnings = vec![WARN_ALL_AT_L0, WARN_UNLOGGED_EDITS];
    if config.kv_separation_opts.is_some() {
        warnings.push(WARN_BLOB_FRAGMENTATION);
    }

    // Join the exclusions against the coverage captured during the scan, plus
    // an entry per KEPT lossy salvage (see `salvaged_coverage`). An excluded
    // TABLE whose metadata never parsed has unknowable coverage — recorded
    // separately so `wal_replay_scope()` can force the full-history
    // obligation instead of silently answering as if nothing was lost. Only a
    // NUMERIC table candidate qualifies: a FOREIGN name (`notes.txt`) held no
    // table data by definition, so its exclusion lost nothing — reporting it
    // unknowable would demand an unbounded WAL archive over scribbles. Blob
    // files stay out for the same reason in the other direction: losing blob
    // content surfaces through the referencing tables (a lossy handle
    // rewrite, or their exclusion), which ARE covered above. Provenance, not
    // a path prefix, decides which is which: a level route may nest its
    // tables anywhere (even under the primary tree's blobs directory), so a
    // candidate is a table iff its parent is one of the scanned table
    // folders.
    let table_folders: Vec<PathBuf> = config
        .all_tables_folders()
        .into_iter()
        .map(|(folder, _)| folder)
        .collect();
    let mut lost_coverage: Vec<(PathBuf, UserKey, UserKey, Option<SeqNo>)> = Vec::new();
    let mut unknowable_losses: Vec<PathBuf> = Vec::new();
    let is_table_candidate = |path: &std::path::Path| {
        path.parent()
            .is_some_and(|parent| table_folders.iter().any(|f| f.as_path() == parent))
            && path.file_name().and_then(|n| n.to_str()).is_some_and(|n| {
                matches!(
                    crate::file::TableDirEntry::classify(n),
                    crate::file::TableDirEntry::Table(_)
                )
            })
    };
    // Redundant exclusions live in `excluded_files` and never reach this
    // loop: their content lives on in the kept tables, so they contribute
    // no coverage. A DAMAGED duplicate does reach it — the corruption signal
    // belongs in `unreadable_files` — but its coverage is not lost either, and
    // an unscopable entry here would send an external WAL over the whole
    // keyspace for a tree that lost nothing.
    for (path, _) in &unreadable_files {
        if redundant_unreadable.contains(path) {
            continue;
        }
        match coverage_by_path.get(path) {
            Some((lo, hi, seqno)) => {
                lost_coverage.push((path.clone(), lo.clone(), hi.clone(), *seqno));
            }
            None if is_table_candidate(path) => unknowable_losses.push(path.clone()),
            None => {}
        }
    }
    lost_coverage.extend(salvaged_coverage);
    lost_coverage.extend(
        lineage_partial
            .into_iter()
            .map(|(_, path, lo, hi, bound, _)| (path, lo, hi, bound)),
    );
    unknowable_losses.extend(salvaged_unknowable);
    // A table the CLEAN manifest referenced that the scan never even SAW: the
    // file is gone, so there is no directory entry to report through — and
    // the manifest records no key range for it, so the loss is unscopable.
    if let Some(referenced) = manifest_referenced {
        let primary_tables = config.path.join("tables");
        let mut missing: Vec<TableId> = referenced
            .tables
            .into_keys()
            .filter(|id| !scanned_table_ids.contains(id))
            .collect();
        missing.sort_unstable();
        for id in missing {
            log::warn!(
                "repair: table {id} is referenced by the recovered manifest but has \
                 no file on disk; its loss is unscopable",
            );
            unknowable_losses.push(primary_tables.join(id.to_string()));
        }
    }

    // The replacements are DURABLE the moment the manifest commits, so the
    // report names every one of them regardless of what the cleanup does: it
    // is what rides out inside `RepairedButUnopened`, the operator's only
    // account of which blobs were replaced. A removal after the commit only
    // annotates its entry.
    // The snapshot's copy is written before the cleanup and may be handed out
    // after it, so it says what holds either way; this call's report then
    // says what the cleanup did.
    let salvage_report_base = blob_files_salvaged.len();
    blob_files_salvaged.extend(stale_blob_originals.iter().map(|(path, note)| {
        (
            path.clone(),
            format!("{note}; original removed once the repair finishes"),
        )
    }));

    // Every damaged dictionary still under its name is set aside once the
    // commit is durable (see `set_aside_damaged_dictionaries`). The report's
    // copy below is what an open hands out when the run ends before the
    // report reaches the caller, and by then the files may already be aside,
    // so the copy names them now. A dictionary the store above rewrote from a
    // supplied copy reads intact again and stays.
    #[cfg(zstd_any)]
    let damaged_dictionaries = {
        let folder = config.path.join(crate::file::DICTS_FOLDER);
        let mut entries = Vec::new();
        for &id in damaged_dictionaries {
            if crate::dicts::is_still_damaged(
                &*config.fs,
                &folder,
                id,
                config.encryption.as_deref(),
            )? {
                entries.push((
                    crate::dicts::damaged_path(&folder, id),
                    format!(
                        "dictionary {id} failed its integrity check; moved aside once the \
                         repair finishes"
                    ),
                ));
            }
        }
        entries
    };
    #[cfg(not(zstd_any))]
    let damaged_dictionaries = Vec::new();

    let mut report = RepairReport {
        recovered,
        salvaged,
        unreadable: unreadable_files.len(),
        unreadable_files,
        excluded_files,
        lost_coverage,
        unknowable_losses,
        blob_files_salvaged,
        damaged_dictionaries,
        method: REPAIR_METHOD,
        warnings,
        id: 0,
        includes: Vec::new(),
    };
    // Appended after this run's entries, so `salvage_report_base` still
    // indexes them.
    if let Some(earlier) = outstanding {
        fold_report(&mut report, earlier);
    }
    report.id = report_marker::id_for(&report, version_id, config.clock.as_ref());

    // The report exists only in this call, and the commit below may end with
    // its outcome unknown: a power loss can still keep the rebuilt snapshot,
    // and the next open then finds nothing to repair. The snapshot holds the
    // report, so it is there exactly when the snapshot is installed, and an
    // open hands it out from there.
    let version = version.with_repair_report(Some(Arc::from(report_marker::encode(&report))));

    // Persist with the tree's own runtime config, not defaults: it drives the
    // manifest framing (checksum algorithm, page ECC, footer mirror, manifest
    // KV checksums), so defaulting it would rewrite a recovered tree's manifest
    // metadata to settings it never used. The last live runtime config died with
    // the lost manifest; the config supplied to `repair` is the authoritative
    // replacement.
    let persisted = crate::version::persist_version(
        &config.path,
        &version,
        config.comparator.name(),
        &*config.fs,
        Arc::new(config.initial_runtime_config.clone()),
        config.encryption.clone(),
        config.sync_mode,
    );
    // The guard stays ARMED until the commit is decided: every fallible
    // `persist_version` step before it installs the snapshot (manifest
    // create / encode / finish, the log's creation, and the rename) leaves the
    // replacements unreferenced garbage a retry would re-derive; keeping
    // them would stack another fresh-id copy per failed attempt, walking
    // recovery toward ENOSPC under exactly the tight-space conditions it
    // targets. The ONE fallible step after the install is the directory sync,
    // so on failure the name on disk decides which world this is: the rebuilt
    // snapshot under its name means the manifest is or may become published and the
    // replacements are the files it references. A probe that cannot tell
    // proves nothing, and fail-safe is to PRESERVE: the worst a preserved
    // orphan costs is disk space a later successful repair's reference
    // filtering reclaims, while a wrong deletion breaks a published manifest
    // permanently.
    // A failure after a PROVEN install must still CARRY the report: the
    // rebuilt snapshot is installed on disk, so the retry opens it without a
    // repair and answers with no report at all — losing the replay
    // obligation exactly like a cleanup failure would. Recording the error
    // routes it through the same report-carrying exit. A proven NOT-INSTALLED
    // failure returns bare (nothing was published, the guard stays armed,
    // the retry's own repair produces a fresh report). An INCONCLUSIVE probe
    // also returns bare: without proof of the install the report may
    // describe a repair that never committed, and a consumer replaying its
    // obligation against the still-live OLD generation could duplicate merge
    // operands — while the blob replacements are preserved (the probe's own
    // fail-safe: their worst cost is disk space) and no post-commit cleanup
    // runs, so nothing of a possibly-live generation is touched. The report
    // rides in the rebuilt snapshot on every bare exit: an open hands it out
    // exactly when that snapshot is the installed one.
    let mut post_commit_error: Option<crate::Error> = None;
    match persisted {
        Ok(()) => published_blob_replacements.disarm(),
        Err(e) => match probe_install(&*config.fs, &config.path, version_id) {
            // When the persist renamed the snapshot into place and could not
            // make its withdrawal durable, a power loss may bring the name
            // back: its absence proves nothing then.
            InstallProbe::NotInstalled if !matches!(e, crate::Error::InstallOutcomeUnknown(_)) => {
                return Err(e);
            }
            // A visible name is not yet a durable install: the persist's own
            // directory sync failed, so a power loss can still take the name
            // back. Only a sync that succeeds now makes the rebuilt snapshot
            // the installed generation; until then the repair has not
            // committed, and answering as committed would hand out a replay
            // scope for a generation that may not survive.
            InstallProbe::Installed => {
                published_blob_replacements.disarm();
                if let Err(sync) = config
                    .fs
                    .sync_directory_with(&config.path, config.sync_mode)
                {
                    log::warn!(
                        "repair: the rebuilt snapshot v{version_id} is visible but its \
                         install cannot be made durable ({sync}); its report rides in it",
                    );
                    return Err(match e {
                        crate::Error::InstallOutcomeUnknown(_) => e,
                        other => crate::Error::InstallOutcomeUnknown(Box::new(other)),
                    });
                }
                post_commit_error = Some(e);
            }
            InstallProbe::NotInstalled | InstallProbe::Inconclusive => {
                published_blob_replacements.disarm();
                return Err(e);
            }
        },
    }

    // The manifest is DURABLE from this point on (or the probe could not
    // prove otherwise): the repair happened, and its report must survive
    // every later failure. A bare error from the cleanup below would
    // discard the only report — once the filesystem fault clears, the next
    // open sweeps the leftover itself and `open_or_repair` answers with no
    // report at all, hiding the committed repair's lost coverage from an
    // external-WAL consumer. The first post-commit failure is therefore
    // RECORDED, the remaining cleanup is skipped, and the completed report
    // rides out inside [`Error::RepairedButUnopened`].

    // A rebuilt snapshot is a complete generation on its own. Sweep every stale
    // edit log so nothing is replayed on top of it: the lost manifest's
    // generation left its log under an OLDER snapshot id (the rebuilt snapshot
    // uses `max(v*) + 1`), so removing only `edits-{version_id}` would normally
    // miss it. Drop every `edits-*` but the fresh snapshot's own, which the
    // persist created empty so its first append syncs no directory. Runs
    // only on a PROVEN commit (like every cleanup below): a recorded
    // post-commit error skips it, and the retry finishes the sweep.
    //
    // No directory fsync here, unlike the blob sweep below. Recovery replays
    // only the LIVE snapshot's log, so an entry a power loss resurrects is
    // never read; the next open recognizes it as an orphan log and sweeps it
    // again. Nothing observes the window, so the barrier would buy nothing.
    if post_commit_error.is_none() {
        let fresh_log = crate::version::edit_log::log_path(&config.path, version_id);
        match config.fs.read_dir(&config.path) {
            Ok(dirents) => {
                for dirent in dirents {
                    if dirent.is_dir
                        || !dirent.file_name.starts_with("edits-")
                        || dirent.path == fresh_log
                    {
                        continue;
                    }
                    match config.fs.remove_file(&dirent.path) {
                        Ok(()) => {}
                        Err(e) if e.kind() == crate::io::ErrorKind::NotFound => {}
                        Err(e) => {
                            post_commit_error = Some(e.into());
                            break;
                        }
                    }
                }
            }
            Err(e) => post_commit_error = Some(e.into()),
        }
    }

    if let Some(p) = &config.recovery_progress {
        p.set_phase(crate::RecoveryPhase::Cleanup);
    }

    // POST-COMMIT, step one: swap every finished replacement onto the name the
    // committed manifest gives it, destroying the damaged source it replaces.
    // Before this the manifest was not durable and the sources were the only
    // copies; after it the manifest describes exactly these bytes. A crash
    // between the commit and a swap leaves `{id}` damaged with `{id}.repair-tmp`
    // beside it — which the next run resolves from the committed manifest alone
    // (see `sweep_superseded_by_committed_manifest`), not from anything this run
    // remembered.
    //
    // NOT best-effort: the manifest already names this content, so a swap that
    // does not happen is a tree whose next open finds the damaged file under the
    // manifest's checksum and fails.
    //
    // The descriptors the repair cached close first: a file still open cannot
    // be replaced or removed on Windows, and nothing reads a held table again.
    // A tree opened on the shared cache meanwhile has its ids in the same range
    // and only loses cached descriptors, which it reopens.
    if let Some(descriptors) = &config.descriptor_table {
        descriptors.remove_trees_from(first_tree_id);
    }
    if post_commit_error.is_none() {
        for (fs, tmp_path, table_path, restricted) in swap_after_commit {
            if let Err(e) =
                commit_repair_tmp(&*fs, &tmp_path, &table_path, config.sync_mode, restricted)
            {
                log::error!(
                    "repair: cannot swap the replacement {} onto {} ({e}); failing the \
                     repair — the committed manifest names the replacement's content",
                    tmp_path.display(),
                    table_path.display(),
                );
                post_commit_error = Some(e);
                break;
            }
        }
    }

    // POST-COMMIT, step two. The manifest is durable and names the salvaged
    // replacements, so the damaged originals they superseded are unreferenced
    // and are removed. This runs AFTER the commit on purpose: until then those
    // originals are what the tables the rewrite had not reached still point at,
    // so removing one earlier would let a failed repair leave a table
    // referencing a blob id that no longer exists — and the retry would then
    // record that table unrecoverable.
    //
    // NOT best-effort. A superseded original left in `blobs/` is outside the
    // committed manifest, so the next open classifies it as an orphan and
    // removes it — and if the directory refuses the removal now, that open
    // FAILS. Reporting a successful repair for a tree that will not open is the
    // one outcome recovery must never produce, so the failure propagates: the
    // manifest is already durable, and a retry once the filesystem is fixed
    // finishes the sweep on the same inputs. The report already names every
    // replacement; a removal only annotates its entry.
    let mut originals_removed = 0;
    if post_commit_error.is_none() {
        for (path, _) in &stale_blob_originals {
            match discard_unreferenced(&*config.fs, path, config.sync_mode) {
                Ok(()) => originals_removed += 1,
                Err(e) => {
                    log::error!(
                        "repair: cannot remove the superseded blob original {} ({e}); \
                         failing the repair — left in blobs/ it is an orphan the next \
                         open must remove, and that removal would hit the same error",
                        path.display(),
                    );
                    post_commit_error = Some(e);
                    break;
                }
            }
        }
    }
    // The originals' entries are the last ones of the list, in their order.
    for (i, (entry, (path, note))) in report
        .blob_files_salvaged
        .iter_mut()
        .skip(salvage_report_base)
        .zip(&stale_blob_originals)
        .enumerate()
    {
        let state = if i < originals_removed {
            "original removed"
        } else {
            "original NOT removed"
        };
        *entry = (path.clone(), format!("{note}; {state}"));
    }

    // Everything the scan classified out of the rebuilt tree: a foreign name, a
    // duplicate id, a table or blob file no bound could make safe. The scan
    // itself touched nothing, so until this point a crash left the directory
    // exactly as the retry expects to find it; from here the manifest is
    // durable and these files are what the tree no longer references.
    if post_commit_error.is_none() {
        for (fs, path, note) in discard_after_commit {
            match discard_unreferenced(&*fs, &path, config.sync_mode) {
                Ok(()) => log::info!("repair: {} removed ({note})", path.display()),
                Err(e) => {
                    log::error!(
                        "repair: cannot remove {} ({e}); failing the repair — left in \
                         place it is a file the next open must reject or sweep, and it \
                         would hit the same error",
                        path.display(),
                    );
                    post_commit_error = Some(e);
                    break;
                }
            }
        }
    }

    // Same rule for the blob files the manifest omits because nothing
    // references them: their data is unreachable, so there is nothing to
    // preserve — but leaving one behind hands the next open an orphan to
    // sweep, and a sweep that fails there fails the open. Remove them here and
    // propagate, so a repair that reports success leaves an openable tree.
    if post_commit_error.is_none() {
        let mut removed_dir: Option<std::path::PathBuf> = None;
        for path in unreferenced_blob_files {
            match config.fs.remove_file(&path) {
                Ok(()) => removed_dir = path.parent().map(std::path::Path::to_path_buf),
                Err(e) if e.kind() == crate::io::ErrorKind::NotFound => {}
                Err(e) => {
                    log::error!(
                        "repair: cannot remove the unreferenced blob file {} ({e}); \
                         failing the repair — left in blobs/ it is an orphan the next \
                         open must remove, and that removal would hit the same error",
                        path.display(),
                    );
                    post_commit_error = Some(e.into());
                    break;
                }
            }
        }
        // ONE sync for the batch, mirroring `remove_published_blob_replacements`:
        // the entries all live in the same directory, and without it a power
        // loss after this repair reports success can restore them, handing the
        // next open the very orphans these removals exist to prevent.
        if let Some(dir) = removed_dir
            && post_commit_error.is_none()
            && let Err(e) = config.fs.sync_directory_with(&dir, config.sync_mode)
        {
            log::error!(
                "repair: cannot make the removal of unreferenced blob files durable \
                 in {} ({e}); failing the repair: a power loss would restore them \
                 as orphans the next open must sweep",
                dir.display(),
            );
            post_commit_error = Some(e.into());
        }
    }

    // A recorded post-commit failure carries the completed report out: the
    // manifest is durable, so the repair happened, and a retry once the
    // filesystem is fixed finishes the cleanup — but its own open may find
    // nothing left to repair and answer with no report at all.
    if let Some(cause) = post_commit_error {
        return Err(crate::Error::RepairedButUnopened {
            report: Box::new(report),
            cause: Box::new(cause),
        });
    }

    // The rebuilt snapshot keeps owing the report until the caller
    // acknowledges it: a process that ends before the caller met the report's
    // replay obligation finds it there again.

    // Success only: a failed run leaves the phase where it stopped, which
    // tells a progress display exactly which stage failed.
    if let Some(p) = &config.recovery_progress {
        p.set_phase(crate::RecoveryPhase::Done);
    }

    Ok(report)
}

#[cfg(feature = "std")]
mod report_marker;

#[cfg(test)]
mod tests;