regolith 0.1.1

use std::io::{self};
use std::path::{Path, PathBuf};

// The module's own tests craft corrupt manifests byte by byte, which
// is the one thing that has to bypass the environment.
#[cfg(test)]
use std::fs::OpenOptions;
use std::sync::Arc;

use crate::env::{BufferedWriter, Env, WriteMode};

use crate::sync::RwLock;

use super::checksum;
use super::sstable::{
    LiveSst, MetadataPolicy, SsTableMeta, SsTableReader, sst_filename, table_carries_data,
};

/// Maximum number of levels in the LSM tree.
pub(crate) const MAX_LEVELS: usize = 7;

/// A snapshot of which SSTables exist at each level.
///
/// Each level holds `Arc<LiveSst>` - the metadata plus an open reader -
/// so that every file referenced by a live version has a pinned file
/// descriptor. Concurrent compaction can safely `unlink` a file as soon
/// as it's removed from the *current* version because the Arcs in older
/// versions keep the FD alive until those versions are dropped.
#[derive(Clone)]
pub(crate) struct Version {
    pub(crate) levels: Vec<Vec<Arc<LiveSst>>>,
    pub(crate) next_file_id: u64,
    pub(crate) last_seq: u64,
    pub(crate) min_wal_id: u64,
}

impl Version {
    pub(crate) fn new() -> Self {
        Self {
            levels: (0..MAX_LEVELS).map(|_| Vec::new()).collect(),
            next_file_id: 1,
            last_seq: 0,
            min_wal_id: 0,
        }
    }

    /// Number of SSTables at L0.
    pub(crate) fn l0_count(&self) -> usize {
        self.levels[0].len()
    }

    /// Total size of SSTables at a given level.
    pub(crate) fn level_size(&self, level: usize) -> u64 {
        self.levels[level].iter().map(|f| f.meta.file_size).sum()
    }
}

/// A runtime mutation to the version. Carries `Arc<LiveSst>` for
/// `AddFile` so the caller is responsible for opening the reader before
/// the apply, and the manifest machinery never has to touch the
/// filesystem for a runtime edit.
#[derive(Clone)]
pub(crate) enum VersionEdit {
    AddFile { level: usize, file: Arc<LiveSst> },
    RemoveFile { level: usize, file_id: u64 },
    SetLastSeq(u64),
    SetNextFileId(u64),
    Reset { next_file_id: u64, min_wal_id: u64 },
}

/// Serialized form of a version edit. The manifest on disk is a sequence
/// of these records; runtime edits are converted to records just before
/// being written out.
enum ManifestRecord {
    AddFile { level: usize, meta: SsTableMeta },
    RemoveFile { level: usize, file_id: u64 },
    SetLastSeq(u64),
    SetNextFileId(u64),
    SetMinWalId(u64),
    Reset { next_file_id: u64, min_wal_id: u64 },
}

const TAG_ADD_FILE: u8 = 1;
const TAG_REMOVE_FILE: u8 = 2;
const TAG_LAST_SEQ: u8 = 3;
const TAG_NEXT_FILE_ID: u8 = 4;
const TAG_MIN_WAL_ID: u8 = 5;
const TAG_RESET: u8 = 6;

impl VersionEdit {
    fn to_record(&self) -> ManifestRecord {
        match self {
            VersionEdit::AddFile { level, file } => ManifestRecord::AddFile {
                level: *level,
                meta: file.meta.clone(),
            },
            VersionEdit::RemoveFile { level, file_id } => ManifestRecord::RemoveFile {
                level: *level,
                file_id: *file_id,
            },
            VersionEdit::SetLastSeq(seq) => ManifestRecord::SetLastSeq(*seq),
            VersionEdit::SetNextFileId(id) => ManifestRecord::SetNextFileId(*id),
            VersionEdit::Reset {
                next_file_id,
                min_wal_id,
            } => ManifestRecord::Reset {
                next_file_id: *next_file_id,
                min_wal_id: *min_wal_id,
            },
        }
    }

    fn requires_manifest_sync(&self) -> bool {
        // File-id reservations do not make new data reachable on
        // their own. They are flushed here and become durable with
        // the next synced AddFile/RemoveFile/SetLastSeq edit.
        !matches!(self, VersionEdit::SetNextFileId(_))
    }
}

impl ManifestRecord {
    fn encode(&self, buf: &mut Vec<u8>) {
        match self {
            ManifestRecord::AddFile { level, meta } => {
                buf.push(TAG_ADD_FILE);
                buf.extend_from_slice(&(*level as u32).to_le_bytes());
                buf.extend_from_slice(&meta.file_id.to_le_bytes());
                buf.extend_from_slice(&(meta.smallest_key.len() as u32).to_le_bytes());
                buf.extend_from_slice(&meta.smallest_key);
                buf.extend_from_slice(&(meta.largest_key.len() as u32).to_le_bytes());
                buf.extend_from_slice(&meta.largest_key);
                buf.extend_from_slice(&meta.file_size.to_le_bytes());
                buf.extend_from_slice(&meta.num_entries.to_le_bytes());
            }
            ManifestRecord::RemoveFile { level, file_id } => {
                buf.push(TAG_REMOVE_FILE);
                buf.extend_from_slice(&(*level as u32).to_le_bytes());
                buf.extend_from_slice(&file_id.to_le_bytes());
            }
            ManifestRecord::SetLastSeq(seq) => {
                buf.push(TAG_LAST_SEQ);
                buf.extend_from_slice(&seq.to_le_bytes());
            }
            ManifestRecord::SetNextFileId(id) => {
                buf.push(TAG_NEXT_FILE_ID);
                buf.extend_from_slice(&id.to_le_bytes());
            }
            ManifestRecord::SetMinWalId(id) => {
                buf.push(TAG_MIN_WAL_ID);
                buf.extend_from_slice(&id.to_le_bytes());
            }
            ManifestRecord::Reset {
                next_file_id,
                min_wal_id,
            } => {
                buf.push(TAG_RESET);
                buf.extend_from_slice(&next_file_id.to_le_bytes());
                buf.extend_from_slice(&min_wal_id.to_le_bytes());
            }
        }
    }

    fn decode(data: &[u8], pos: &mut usize) -> io::Result<Option<Self>> {
        if *pos >= data.len() {
            return Ok(None);
        }

        let tag = data[*pos];
        *pos += 1;

        match tag {
            TAG_ADD_FILE => {
                let level = read_u32(data, pos)? as usize;
                validate_level_index(level)?;
                let file_id = read_u64(data, pos)?;
                let smallest_key = read_bytes(data, pos)?;
                let largest_key = read_bytes(data, pos)?;
                let file_size = read_u64(data, pos)?;
                let num_entries = read_u64(data, pos)?;

                Ok(Some(ManifestRecord::AddFile {
                    level,
                    meta: SsTableMeta {
                        file_id,
                        smallest_key,
                        largest_key,
                        file_size,
                        num_entries,
                    },
                }))
            }
            TAG_REMOVE_FILE => {
                let level = read_u32(data, pos)? as usize;
                validate_level_index(level)?;
                let file_id = read_u64(data, pos)?;
                Ok(Some(ManifestRecord::RemoveFile { level, file_id }))
            }
            TAG_LAST_SEQ => {
                let seq = read_u64(data, pos)?;
                Ok(Some(ManifestRecord::SetLastSeq(seq)))
            }
            TAG_NEXT_FILE_ID => {
                let id = read_u64(data, pos)?;
                Ok(Some(ManifestRecord::SetNextFileId(id)))
            }
            TAG_MIN_WAL_ID => {
                let id = read_u64(data, pos)?;
                Ok(Some(ManifestRecord::SetMinWalId(id)))
            }
            TAG_RESET => {
                let next_file_id = read_u64(data, pos)?;
                let min_wal_id = read_u64(data, pos)?;
                Ok(Some(ManifestRecord::Reset {
                    next_file_id,
                    min_wal_id,
                }))
            }
            _ => Err(io::Error::new(
                io::ErrorKind::InvalidData,
                format!("unknown manifest record tag: {}", tag),
            )),
        }
    }
}

fn read_u32(data: &[u8], pos: &mut usize) -> io::Result<u32> {
    if *pos + 4 > data.len() {
        return Err(io::Error::new(io::ErrorKind::UnexpectedEof, "short read"));
    }
    let val = u32::from_le_bytes(data[*pos..*pos + 4].try_into().unwrap());
    *pos += 4;
    Ok(val)
}

fn read_u64(data: &[u8], pos: &mut usize) -> io::Result<u64> {
    if *pos + 8 > data.len() {
        return Err(io::Error::new(io::ErrorKind::UnexpectedEof, "short read"));
    }
    let val = u64::from_le_bytes(data[*pos..*pos + 8].try_into().unwrap());
    *pos += 8;
    Ok(val)
}

fn read_bytes(data: &[u8], pos: &mut usize) -> io::Result<Vec<u8>> {
    let len = read_u32(data, pos)? as usize;
    if *pos + len > data.len() {
        return Err(io::Error::new(io::ErrorKind::UnexpectedEof, "short read"));
    }
    let bytes = data[*pos..*pos + len].to_vec();
    *pos += len;
    Ok(bytes)
}

fn validate_level_index(level: usize) -> io::Result<()> {
    if level < MAX_LEVELS {
        return Ok(());
    }

    Err(io::Error::new(
        io::ErrorKind::InvalidData,
        format!("manifest level {level} out of range 0..{MAX_LEVELS}"),
    ))
}

/// Identifier at the head of a MANIFEST: `REGOMAN` plus a format byte.
const MANIFEST_MAGIC: [u8; 7] = *b"REGOMAN";

/// On-disk MANIFEST format this build writes.
const MANIFEST_FORMAT_V1: u8 = 1;

/// Stamp layout: magic, format, checksum.
const MANIFEST_STAMP_LEN: usize = 12;

/// How far past its canonical size a MANIFEST may grow before it is
/// rewritten.
///
/// The log is append-only, so without this it grows for the life of the
/// database and recovery replays every edit ever made. Rewriting costs
/// one record per live SSTable, and only after the log has grown to a
/// multiple of that, so the amortized cost per edit is constant while
/// the file stays within a bounded factor of its minimum.
const MANIFEST_COMPACT_FACTOR: u64 = 4;

/// Floor for the rewrite trigger, so a small database does not rewrite
/// its manifest on almost every edit.
const MANIFEST_COMPACT_FLOOR: u64 = 64 * 1024;

/// Rough size of one `AddFile` record, used only to size the rewrite
/// trigger. Being approximate costs a slightly different trigger point,
/// never correctness: the rewrite is driven by the real file length.
const APPROX_ADD_FILE_RECORD_BYTES: u64 = 256;

/// Manages the current version and persists version edits to a manifest log.
pub(crate) struct VersionSet {
    current: Arc<RwLock<Arc<Version>>>,
    manifest_path: PathBuf,
    /// Bytes the manifest holds on disk, tracked rather than stat'ed so
    /// the rewrite check costs nothing on the common path.
    manifest_bytes: u64,
    manifest_writer: Option<BufferedWriter>,
    env: Arc<dyn Env>,
}

struct ManifestReplay {
    version: Version,
    valid_len: usize,
}

/// An unreferenced `*.sst` file that the discarded-table guard could not
/// dismiss as a crash artifact, with the reason it counts.
struct SuspectTable {
    path: PathBuf,
    /// `None` when the file's metadata could not be read.
    len: Option<u64>,
    reason: String,
}

/// How many suspects the guard's error message names before summarising
/// the rest. The cap is reported in the message so a long list never
/// reads as a short one.
const SUSPECTS_NAMED: usize = 8;

fn describe_suspects(suspects: &[SuspectTable]) -> String {
    let mut out = suspects
        .iter()
        .take(SUSPECTS_NAMED)
        .map(|s| match s.len {
            Some(len) => format!("{} ({len} bytes, {})", s.path.display(), s.reason),
            None => format!("{} (size unknown, {})", s.path.display(), s.reason),
        })
        .collect::<Vec<_>>()
        .join(", ");
    if suspects.len() > SUSPECTS_NAMED {
        out.push_str(&format!(
            ", and {} more not named here",
            suspects.len() - SUSPECTS_NAMED
        ));
    }
    out
}

impl VersionSet {
    /// Create or recover a VersionSet from the given directory, with an
    /// explicit policy for how the readers it opens hold their index and
    /// filter blocks.
    ///
    /// Recovery is where this matters most: it opens a reader for every
    /// SSTable the manifest references, so the policy decides whether
    /// that whole set of indexes and filters is pinned or bounded by
    /// the block cache.
    pub(crate) fn open_with_policy(
        env: &Arc<dyn Env>,
        db_dir: &Path,
        sst_dir: &Path,
        policy: MetadataPolicy,
    ) -> io::Result<Self> {
        let manifest_path = db_dir.join("MANIFEST");

        let manifest_bytes;
        let (version, writer) = if env.exists(&manifest_path) {
            let data = env.read(&manifest_path)?;
            let replay = Self::replay_manifest(env, &data, sst_dir, policy)?;
            Self::reject_discarded_tables(&**env, &replay, data.len(), sst_dir, &manifest_path)?;

            // Trim through its own handle, and close it before the
            // append handle is opened. A torn or corrupt tail is the
            // ordinary shape of a crash, so this runs on a normal
            // reopen; shortening a file needs write access that an
            // append handle does not carry on every platform, which is
            // what [`WriteMode::Update`] exists for.
            if replay.valid_len < data.len() {
                let mut trim = env.open_write(&manifest_path, WriteMode::Update)?;
                trim.set_len(replay.valid_len as u64)?;
                trim.sync_all()?;
            }
            // A file too short to carry a stamp is one a crash caught
            // during creation, and the guard above has already cleared
            // it of hiding live tables. The append path never writes a
            // stamp, so it is written here instead: without it the next
            // open would find a stamp-less record stream and refuse.
            if data.len() < MANIFEST_STAMP_LEN {
                let mut file = env.open_write(&manifest_path, WriteMode::Truncate)?;
                file.write_all(&Self::encode_stamp())?;
                file.sync_all()?;
                crate::env::sync_parent_dir(&**env, &manifest_path)?;
                manifest_bytes = MANIFEST_STAMP_LEN as u64;
                (replay.version, BufferedWriter::new(file))
            } else {
                let file = env.open_write(&manifest_path, WriteMode::Append)?;
                manifest_bytes = replay.valid_len as u64;
                (replay.version, BufferedWriter::new(file))
            }
        } else {
            let version = Version::new();
            let mut file = env.open_write(&manifest_path, WriteMode::Truncate)?;
            file.write_all(&Self::encode_stamp())?;
            file.sync_all()?;
            crate::env::sync_parent_dir(&**env, &manifest_path)?;
            manifest_bytes = MANIFEST_STAMP_LEN as u64;
            (version, BufferedWriter::new(file))
        };

        Ok(Self {
            current: Arc::new(RwLock::new(Arc::new(version))),
            manifest_path,
            manifest_bytes,
            manifest_writer: Some(writer),
            env: Arc::clone(env),
        })
    }

    /// Create or recover a VersionSet through the standard
    /// environment.
    #[cfg(any(test, feature = "fuzzing"))]
    pub(crate) fn open(db_dir: &Path, sst_dir: &Path) -> io::Result<Self> {
        Self::open_with_policy(
            &crate::env::std_env(),
            db_dir,
            sst_dir,
            MetadataPolicy::Pinned,
        )
    }

    /// Recover an existing VersionSet without mutating the manifest.
    ///
    /// This is used by read-only opens: replay still tolerates a
    /// truncated trailing record exactly like the read-write path, but
    /// the file is not repaired in place and no append writer is kept.
    pub(crate) fn open_read_only(
        env: &Arc<dyn Env>,
        db_dir: &Path,
        sst_dir: &Path,
        policy: MetadataPolicy,
    ) -> io::Result<Self> {
        let manifest_path = db_dir.join("MANIFEST");
        let data = env.read(&manifest_path)?;
        let replay = Self::replay_manifest(env, &data, sst_dir, policy)?;
        Self::reject_discarded_tables(&**env, &replay, data.len(), sst_dir, &manifest_path)?;

        Ok(Self {
            current: Arc::new(RwLock::new(Arc::new(replay.version))),
            manifest_path,
            manifest_bytes: replay.valid_len as u64,
            manifest_writer: None,
            env: Arc::clone(env),
        })
    }

    /// Get the current version.
    pub(crate) fn current(&self) -> Arc<Version> {
        Arc::clone(&*self.current.read())
    }

    /// Path of the manifest file on disk.
    pub(crate) fn manifest_path(&self) -> &Path {
        &self.manifest_path
    }

    /// Apply a batch of edits atomically: update the in-memory version
    /// and persist the serialized records to the manifest log.
    pub(crate) fn apply(&mut self, edits: &[VersionEdit]) -> io::Result<()> {
        for edit in edits {
            match edit {
                VersionEdit::AddFile { level, .. } | VersionEdit::RemoveFile { level, .. } => {
                    validate_level_index(*level)?;
                }
                VersionEdit::SetLastSeq(_)
                | VersionEdit::SetNextFileId(_)
                | VersionEdit::Reset { .. } => {}
            }
        }

        let mut version = (*self.current()).clone();

        for edit in edits {
            match edit {
                VersionEdit::AddFile { level, file } => {
                    version.levels[*level].push(Arc::clone(file));
                }
                VersionEdit::RemoveFile { level, file_id } => {
                    version.levels[*level].retain(|f| f.meta.file_id != *file_id);
                }
                VersionEdit::SetLastSeq(seq) => {
                    version.last_seq = *seq;
                }
                VersionEdit::SetNextFileId(id) => {
                    version.next_file_id = *id;
                }
                VersionEdit::Reset {
                    next_file_id,
                    min_wal_id,
                } => {
                    version.levels = (0..MAX_LEVELS).map(|_| Vec::new()).collect();
                    version.last_seq = 0;
                    version.next_file_id = *next_file_id;
                    version.min_wal_id = *min_wal_id;
                }
            }
        }

        let records: Vec<ManifestRecord> = edits.iter().map(VersionEdit::to_record).collect();
        let encoded = Self::encode_records(&records);
        let requires_sync = edits.iter().any(VersionEdit::requires_manifest_sync);
        if let Some(writer) = &mut self.manifest_writer {
            writer.write_all(&encoded)?;
            if requires_sync {
                writer.sync_all()?;
            } else {
                writer.flush()?;
            }
        }

        let live_files: u64 = version.levels.iter().map(|l| l.len() as u64).sum();
        *self.current.write() = Arc::new(version);
        self.manifest_bytes += encoded.len() as u64;

        // The log is append-only, so a database that runs for years
        // replays every edit it ever made unless the file is rewritten.
        // Rewriting costs one record per live table and happens only once
        // the log has grown to a multiple of that, so the cost per edit
        // is constant and the file stays within a bounded factor of its
        // smallest possible size.
        if self.manifest_bytes > Self::compact_threshold(live_files) {
            self.compact_manifest()?;
        }

        Ok(())
    }

    /// Size at which the manifest is rewritten, from the number of live
    /// SSTables it has to name.
    fn compact_threshold(live_files: u64) -> u64 {
        let canonical = MANIFEST_STAMP_LEN as u64 + live_files * APPROX_ADD_FILE_RECORD_BYTES;
        MANIFEST_COMPACT_FLOOR.max(canonical.saturating_mul(MANIFEST_COMPACT_FACTOR))
    }

    /// Rewrite the manifest from scratch, emitting the current version as
    /// a single compact sequence of records. Readers in the live
    /// `Version` are preserved - we never close their file descriptors.
    pub(crate) fn compact_manifest(&mut self) -> io::Result<()> {
        let version = self.current();

        let mut records = Vec::new();
        records.push(ManifestRecord::SetNextFileId(version.next_file_id));
        records.push(ManifestRecord::SetLastSeq(version.last_seq));
        records.push(ManifestRecord::SetMinWalId(version.min_wal_id));
        for (level, files) in version.levels.iter().enumerate() {
            for file in files {
                records.push(ManifestRecord::AddFile {
                    level,
                    meta: file.meta.clone(),
                });
            }
        }

        let encoded = Self::encode_records(&records);

        let tmp_path = self.manifest_path.with_extension("tmp");
        {
            let mut file = self.env.open_write(&tmp_path, WriteMode::Truncate)?;
            file.write_all(&Self::encode_stamp())?;
            file.write_all(&encoded)?;
            file.sync_all()?;
        }
        // Close the log before replacing it. Windows refuses to replace
        // a file that still has an open handle, so a rewrite performed
        // while this writer was live failed with "Access is denied" and
        // took every manifest rewrite on that platform with it. On unix
        // the rename would have worked either way: the old inode simply
        // outlives its name. Dropping the writer first is correct on
        // both, and the reopen below is where the new log is picked up.
        self.manifest_writer = None;
        self.env.rename(&tmp_path, &self.manifest_path)?;
        crate::env::sync_parent_dir(&*self.env, &self.manifest_path)?;
        self.manifest_bytes = (MANIFEST_STAMP_LEN + encoded.len()) as u64;

        let file = self
            .env
            .open_write(&self.manifest_path, WriteMode::Append)?;
        self.manifest_writer = Some(BufferedWriter::new(file));

        Ok(())
    }

    /// Encode the stamp a manifest begins with.
    pub(crate) fn encode_stamp() -> [u8; MANIFEST_STAMP_LEN] {
        let mut out = [0u8; MANIFEST_STAMP_LEN];
        out[0..7].copy_from_slice(&MANIFEST_MAGIC);
        out[7] = MANIFEST_FORMAT_V1;
        let checksum = checksum::manifest_record(0, &out[0..8]);
        out[8..12].copy_from_slice(&checksum.to_le_bytes());
        out
    }

    /// Length of the stamp at the head of `data`.
    ///
    /// The stamp is mandatory: it is written when the manifest is
    /// created, before any record, so a file too short to hold one is a
    /// crash during creation rather than the loss of an acknowledged
    /// edit. Nothing is consumed from it, so replay ends short of the
    /// file length and the open guard still weighs the tables on disk
    /// rather than treating the replay as clean. A file long enough to
    /// carry a stamp but not carrying one is refused.
    fn stamp_len(data: &[u8]) -> io::Result<usize> {
        if data.len() < MANIFEST_STAMP_LEN {
            return Ok(0);
        }
        if data[0..7] != MANIFEST_MAGIC {
            return Err(io::Error::new(
                io::ErrorKind::InvalidData,
                "MANIFEST does not begin with the REGOMAN stamp",
            ));
        }
        let format = data[7];
        let stored = u32::from_le_bytes([data[8], data[9], data[10], data[11]]);
        if stored != checksum::manifest_record(0, &data[0..8]) {
            return Err(io::Error::new(
                io::ErrorKind::InvalidData,
                "MANIFEST stamp checksum mismatch",
            ));
        }
        if format > MANIFEST_FORMAT_V1 {
            return Err(io::Error::new(
                io::ErrorKind::InvalidData,
                format!(
                    "MANIFEST format {format} was written by a newer regolith than this build, \
                     which understands up to {MANIFEST_FORMAT_V1}"
                ),
            ));
        }
        Ok(MANIFEST_STAMP_LEN)
    }

    fn encode_records(records: &[ManifestRecord]) -> Vec<u8> {
        let mut buf = Vec::new();
        for record in records {
            let mut record_buf = Vec::new();
            record.encode(&mut record_buf);

            let len = record_buf.len() as u32;
            let checksum = checksum::manifest_record(len, &record_buf);

            buf.extend_from_slice(&len.to_le_bytes());
            buf.extend_from_slice(&record_buf);
            buf.extend_from_slice(&checksum.to_le_bytes());
        }
        buf
    }

    /// Refuse to open when a manifest that did not replay cleanly ends up
    /// referencing no SSTable at all while the table directory still holds
    /// one that could carry data.
    ///
    /// Replay stops at the first unreadable record and the tail is
    /// discarded, which is correct for a record that a crash left half
    /// written. When the *first* record is unreadable the same rule
    /// silently turns a populated database into an empty one, so that
    /// combination is reported instead of served: the table files are
    /// still on disk and only the manifest needs repairing.
    ///
    /// A crash inside the very first flush leaves the opposite shape: a
    /// table file that holds nothing, next to a WAL that holds every
    /// acknowledged write. Refusing on that file would lose the writes
    /// the WAL still has, so `suspect_tables` rules it out
    /// before the count is taken.
    fn reject_discarded_tables(
        env: &dyn Env,
        replay: &ManifestReplay,
        manifest_len: usize,
        sst_dir: &Path,
        manifest_path: &Path,
    ) -> io::Result<()> {
        let replayed_cleanly = manifest_len > 0 && replay.valid_len == manifest_len;
        if replayed_cleanly || replay.version.levels.iter().any(|level| !level.is_empty()) {
            return Ok(());
        }
        let suspects = Self::suspect_tables(env, sst_dir);
        if suspects.is_empty() {
            return Ok(());
        }
        Err(io::Error::new(
            io::ErrorKind::InvalidData,
            format!(
                "{} is corrupt: it references no SSTable, but {} table file(s) in {} may still hold data. \
                 Opening would discard them, so the database is left untouched. Suspect tables: {}",
                manifest_path.display(),
                suspects.len(),
                sst_dir.display(),
                describe_suspects(&suspects),
            ),
        ))
    }

    /// The unreferenced `*.sst` files that could plausibly hold live data.
    ///
    /// A zero-length table, or one whose footer records no entry and no
    /// range tombstone, is what a crash inside a flush leaves behind: the
    /// directory entry reached the journal, the delayed-allocated data
    /// blocks did not. Such a file cannot be a live table the manifest is
    /// about to discard, so it is logged and skipped rather than counted.
    ///
    /// Everything else counts, including a file whose footer will not
    /// parse. An unreadable file cannot be proved empty, and keeping the
    /// database shut preserves it for repair.
    ///
    /// Nothing is deleted here, so a crash part way through recovery
    /// leaves the directory exactly as this pass found it and the next
    /// open reaches the same verdict.
    fn suspect_tables(env: &dyn Env, sst_dir: &Path) -> Vec<SuspectTable> {
        let entries = match env.read_dir(sst_dir) {
            Ok(entries) => entries,
            Err(e) => {
                tracing::warn!(
                    dir = %sst_dir.display(),
                    error = %e,
                    "could not list the SSTable directory while checking for discarded tables"
                );
                return Vec::new();
            }
        };

        let mut suspects = Vec::new();
        for entry in entries {
            let path = entry.path.clone();
            if path.extension().and_then(|ext| ext.to_str()) != Some("sst") {
                continue;
            }
            let len = match env.metadata(&path) {
                Ok(meta) => Some(meta.len),
                Err(e) => {
                    suspects.push(SuspectTable {
                        path,
                        len: None,
                        reason: format!("unreadable: {e}"),
                    });
                    continue;
                }
            };
            if len == Some(0) {
                tracing::warn!(
                    path = %path.display(),
                    "ignoring a zero-length orphan SSTable left by a crash inside a flush"
                );
                continue;
            }
            match table_carries_data(env, &path) {
                Ok(true) => suspects.push(SuspectTable {
                    path,
                    len,
                    reason: "carries data".to_string(),
                }),
                Ok(false) => tracing::warn!(
                    path = %path.display(),
                    "ignoring an orphan SSTable whose footer records no entry and no range tombstone"
                ),
                Err(e) => suspects.push(SuspectTable {
                    path,
                    len,
                    reason: format!("unreadable footer: {e}"),
                }),
            }
        }
        suspects.sort_by(|a, b| a.path.cmp(&b.path));
        suspects
    }

    /// Replay every record, then open a reader for each surviving file
    /// through `env` under `policy`.
    fn replay_manifest(
        env: &Arc<dyn Env>,
        data: &[u8],
        sst_dir: &Path,
        policy: MetadataPolicy,
    ) -> io::Result<ManifestReplay> {
        // Two-pass replay. The first pass walks every record and tracks
        // the *logical* state of each level - which file ids are live -
        // without touching the filesystem. Only after replay completes
        // do we open readers for the surviving files.
        //
        // This matters for compaction-heavy histories: when compaction
        // adds an L1 file and removes the L0 inputs, both records land
        // in the manifest, but the inputs' physical files are unlinked
        // from disk by `delete_old_files`. An eager open at AddFile
        // time would fail on the unlinked files even though a later
        // RemoveFile record cancels them out.
        let mut surviving: Vec<Vec<SsTableMeta>> = vec![Vec::new(); MAX_LEVELS];
        let mut last_seq: u64 = 0;
        let mut next_file_id: u64 = 1;
        let mut min_wal_id: u64 = 0;
        // The stamp is not a record. `valid_len` starts past it so a
        // clean replay ends exactly at the file length, which is what
        // `reject_discarded_tables` compares against.
        let stamp = Self::stamp_len(data)?;
        let mut offset = stamp;
        let mut valid_len = stamp;

        while offset < data.len() {
            if offset + 4 > data.len() {
                tracing::warn!("Truncated manifest record header, stopping replay");
                break;
            }

            let len = u32::from_le_bytes(data[offset..offset + 4].try_into().unwrap()) as usize;
            offset += 4;

            if offset + len + 4 > data.len() {
                tracing::warn!("Truncated manifest record, stopping replay");
                break;
            }

            let record_data = &data[offset..offset + len];
            offset += len;

            let stored_checksum = u32::from_le_bytes(data[offset..offset + 4].try_into().unwrap());
            offset += 4;

            let computed_checksum = checksum::manifest_record(len as u32, record_data);
            if stored_checksum != computed_checksum {
                tracing::warn!("Manifest checksum mismatch, stopping replay");
                break;
            }

            let mut pos = 0;
            while let Some(record) = ManifestRecord::decode(record_data, &mut pos)? {
                match record {
                    ManifestRecord::AddFile { level, meta } => {
                        surviving[level].push(meta);
                    }
                    ManifestRecord::RemoveFile { level, file_id } => {
                        surviving[level].retain(|m| m.file_id != file_id);
                    }
                    ManifestRecord::SetLastSeq(seq) => {
                        last_seq = seq;
                    }
                    ManifestRecord::SetNextFileId(id) => {
                        next_file_id = id;
                    }
                    ManifestRecord::SetMinWalId(id) => {
                        min_wal_id = id;
                    }
                    ManifestRecord::Reset {
                        next_file_id: reset_next_file_id,
                        min_wal_id: reset_min_wal_id,
                    } => {
                        for level in &mut surviving {
                            level.clear();
                        }
                        last_seq = 0;
                        next_file_id = reset_next_file_id;
                        min_wal_id = reset_min_wal_id;
                    }
                }
            }
            valid_len = offset;
        }

        // Second pass: open readers for the survivors.
        let mut version = Version::new();
        version.last_seq = last_seq;
        version.next_file_id = next_file_id;
        version.min_wal_id = min_wal_id;
        for (level, files) in surviving.into_iter().enumerate() {
            for meta in files {
                let path = sst_dir.join(sst_filename(meta.file_id));
                let reader = Arc::new(
                    SsTableReader::open_with(env, &path, meta.file_id, policy).map_err(|e| {
                        std::io::Error::new(e.kind(), format!("open {}: {e}", path.display()))
                    })?,
                );
                version.levels[level].push(LiveSst::new(meta, reader));
            }
        }

        Ok(ManifestReplay { version, valid_len })
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use tempfile::TempDir;

    /// Build a real on-disk SSTable and open a reader for it. Used by
    /// tests that need a non-trivial `LiveSst` instance.
    fn make_live_sst(dir: &Path, file_id: u64, smallest: &[u8], largest: &[u8]) -> Arc<LiveSst> {
        use super::super::internal_key::{VALUE_TYPE_VALUE, encode_internal_key};
        use super::super::sstable::SsTableWriter;
        use crate::options::CompressionType;

        let path = dir.join(sst_filename(file_id));
        let mut writer =
            SsTableWriter::new(&path, 4096, 10, CompressionType::None, None, false, 4096).unwrap();
        writer
            .add(
                &encode_internal_key(smallest, 1, VALUE_TYPE_VALUE),
                b"value",
            )
            .unwrap();
        if smallest != largest {
            writer
                .add(&encode_internal_key(largest, 1, VALUE_TYPE_VALUE), b"value")
                .unwrap();
        }
        let summary = writer.finish().unwrap().unwrap();
        let file_size = std::fs::metadata(&path).unwrap().len();
        let reader = Arc::new(SsTableReader::open(&path, file_id).unwrap());
        LiveSst::new(
            SsTableMeta {
                file_id,
                smallest_key: summary.smallest_user_key,
                largest_key: summary.largest_user_key,
                file_size,
                num_entries: summary.num_entries,
            },
            reader,
        )
    }

    fn second_record_checksum_offset(path: &Path) -> usize {
        let data = std::fs::read(path).unwrap();
        // Records begin after the stamp.
        let base = MANIFEST_STAMP_LEN;
        let first_len = u32::from_le_bytes(data[base..base + 4].try_into().unwrap()) as usize;
        let second_start = base + 4 + first_len + 4;
        let second_len =
            u32::from_le_bytes(data[second_start..second_start + 4].try_into().unwrap()) as usize;
        second_start + 4 + second_len
    }

    fn test_meta(file_id: u64) -> SsTableMeta {
        SsTableMeta {
            file_id,
            smallest_key: b"a".to_vec(),
            largest_key: b"z".to_vec(),
            file_size: 128,
            num_entries: 2,
        }
    }

    #[test]
    fn manifest_checksum_covers_length_header() {
        let mut record = Vec::new();
        ManifestRecord::SetLastSeq(7).encode(&mut record);
        let len = record.len() as u32;
        let baseline = checksum::manifest_record(len, &record);
        assert_ne!(baseline, checksum::manifest_record(len + 1, &record));
    }

    #[test]
    fn test_apply_and_replay_roundtrip() {
        let dir = TempDir::new().unwrap();
        let sst_dir = dir.path().join("sst");
        std::fs::create_dir_all(&sst_dir).unwrap();

        let file1 = make_live_sst(&sst_dir, 1, b"aaa", b"zzz");

        let edits = vec![
            VersionEdit::AddFile {
                level: 0,
                file: Arc::clone(&file1),
            },
            VersionEdit::SetLastSeq(42),
            VersionEdit::SetNextFileId(10),
        ];

        {
            let mut vs = VersionSet::open(dir.path(), &sst_dir).unwrap();
            vs.apply(&edits).unwrap();
            let v = vs.current();
            assert_eq!(v.levels[0].len(), 1);
            assert_eq!(v.levels[0][0].meta.file_id, 1);
            assert_eq!(v.last_seq, 42);
            assert_eq!(v.next_file_id, 10);
            assert_eq!(v.min_wal_id, 0);
        }

        // Recover by replaying the manifest; readers are reopened.
        let vs = VersionSet::open(dir.path(), &sst_dir).unwrap();
        let v = vs.current();
        assert_eq!(v.levels[0].len(), 1);
        assert_eq!(v.levels[0][0].meta.file_id, 1);
        assert_eq!(v.last_seq, 42);
        assert_eq!(v.next_file_id, 10);
        assert_eq!(v.min_wal_id, 0);
    }

    #[test]
    fn test_remove_file_hides_it_from_new_version() {
        let dir = TempDir::new().unwrap();
        let sst_dir = dir.path().join("sst");
        std::fs::create_dir_all(&sst_dir).unwrap();

        let file1 = make_live_sst(&sst_dir, 1, b"a", b"c");
        let file2 = make_live_sst(&sst_dir, 2, b"d", b"f");

        let mut vs = VersionSet::open(dir.path(), &sst_dir).unwrap();
        vs.apply(&[
            VersionEdit::AddFile {
                level: 0,
                file: Arc::clone(&file1),
            },
            VersionEdit::AddFile {
                level: 0,
                file: Arc::clone(&file2),
            },
        ])
        .unwrap();

        // Holding a snapshot of the version *before* removal keeps both
        // files alive - this is the invariant that lets get/iter reads
        // survive concurrent compaction.
        let pinned = vs.current();
        assert_eq!(pinned.levels[0].len(), 2);

        vs.apply(&[VersionEdit::RemoveFile {
            level: 0,
            file_id: 1,
        }])
        .unwrap();

        let v = vs.current();
        assert_eq!(v.levels[0].len(), 1);
        assert_eq!(v.levels[0][0].meta.file_id, 2);
        // Pinned snapshot still sees both files.
        assert_eq!(pinned.levels[0].len(), 2);
    }

    #[test]
    fn initial_version_has_empty_levels_and_defaults() {
        let v = Version::new();
        assert_eq!(v.levels.len(), MAX_LEVELS);
        assert!(v.levels.iter().all(|l| l.is_empty()));
        assert_eq!(v.next_file_id, 1);
        assert_eq!(v.last_seq, 0);
        assert_eq!(v.min_wal_id, 0);
        assert_eq!(v.l0_count(), 0);
        for level in 0..MAX_LEVELS {
            assert_eq!(v.level_size(level), 0);
        }
    }

    #[test]
    fn manifest_path_returned_as_written() {
        let dir = TempDir::new().unwrap();
        let sst_dir = dir.path().join("sst");
        std::fs::create_dir_all(&sst_dir).unwrap();
        let vs = VersionSet::open(dir.path(), &sst_dir).unwrap();
        assert_eq!(vs.manifest_path(), dir.path().join("MANIFEST"));
    }

    #[test]
    fn manifest_record_encode_decode_round_trip() {
        // Exercise every tag through the encode/decode path used by
        // the replay loop.
        let records = [
            ManifestRecord::AddFile {
                level: 2,
                meta: SsTableMeta {
                    file_id: 99,
                    smallest_key: b"aaa".to_vec(),
                    largest_key: b"zzz".to_vec(),
                    file_size: 4096,
                    num_entries: 128,
                },
            },
            ManifestRecord::RemoveFile {
                level: 1,
                file_id: 7,
            },
            ManifestRecord::SetLastSeq(999),
            ManifestRecord::SetNextFileId(42),
            ManifestRecord::SetMinWalId(11),
            ManifestRecord::Reset {
                next_file_id: 77,
                min_wal_id: 76,
            },
        ];
        for r in &records {
            let mut buf = Vec::new();
            r.encode(&mut buf);
            let mut pos = 0;
            let decoded = match ManifestRecord::decode(&buf, &mut pos) {
                Ok(Some(d)) => d,
                other => panic!("expected decoded record, got {:?}", other.is_ok()),
            };
            // Re-encode and compare - equality via round-trip avoids
            // having to add PartialEq to ManifestRecord.
            let mut rebuf = Vec::new();
            decoded.encode(&mut rebuf);
            assert_eq!(buf, rebuf);
            assert_eq!(pos, buf.len());
        }
    }

    #[test]
    fn manifest_record_decode_rejects_unknown_tag() {
        let data = [0xFFu8];
        let mut pos = 0;
        let kind = match ManifestRecord::decode(&data, &mut pos) {
            Err(e) => e.kind(),
            Ok(_) => panic!("expected error on unknown tag"),
        };
        assert_eq!(kind, io::ErrorKind::InvalidData);
    }

    #[test]
    fn manifest_record_decode_rejects_invalid_level_indexes() {
        let records = [
            ManifestRecord::AddFile {
                level: MAX_LEVELS,
                meta: test_meta(1),
            },
            ManifestRecord::RemoveFile {
                level: MAX_LEVELS,
                file_id: 1,
            },
        ];

        for record in records {
            let mut data = Vec::new();
            record.encode(&mut data);
            let mut pos = 0;
            let kind = match ManifestRecord::decode(&data, &mut pos) {
                Err(e) => e.kind(),
                Ok(_) => panic!("expected invalid level error"),
            };
            assert_eq!(kind, io::ErrorKind::InvalidData);
        }
    }

    #[test]
    fn manifest_record_decode_returns_none_at_eof() {
        let mut pos = 0;
        let got = ManifestRecord::decode(&[], &mut pos).unwrap();
        assert!(got.is_none());
    }

    #[test]
    fn apply_rejects_invalid_level_indexes() {
        let dir = TempDir::new().unwrap();
        let sst_dir = dir.path().join("sst");
        std::fs::create_dir_all(&sst_dir).unwrap();

        let file = make_live_sst(&sst_dir, 1, b"a", b"z");
        let mut vs = VersionSet::open(dir.path(), &sst_dir).unwrap();

        let kind = match vs.apply(&[VersionEdit::AddFile {
            level: MAX_LEVELS,
            file,
        }]) {
            Err(e) => e.kind(),
            Ok(_) => panic!("expected invalid level error"),
        };
        assert_eq!(kind, io::ErrorKind::InvalidData);
        assert_eq!(vs.current().levels.iter().map(Vec::len).sum::<usize>(), 0);

        let kind = match vs.apply(&[VersionEdit::RemoveFile {
            level: MAX_LEVELS,
            file_id: 1,
        }]) {
            Err(e) => e.kind(),
            Ok(_) => panic!("expected invalid level error"),
        };
        assert_eq!(kind, io::ErrorKind::InvalidData);
    }

    #[test]
    fn open_rejects_manifest_with_invalid_level_index() {
        let dir = TempDir::new().unwrap();
        let sst_dir = dir.path().join("sst");
        std::fs::create_dir_all(&sst_dir).unwrap();

        let records = [ManifestRecord::AddFile {
            level: MAX_LEVELS,
            meta: test_meta(1),
        }];
        std::fs::write(
            dir.path().join("MANIFEST"),
            VersionSet::encode_records(&records),
        )
        .unwrap();

        let kind = match VersionSet::open(dir.path(), &sst_dir) {
            Err(e) => e.kind(),
            Ok(_) => panic!("expected invalid level error"),
        };
        assert_eq!(kind, io::ErrorKind::InvalidData);
    }

    #[test]
    fn replay_survives_truncated_trailer() {
        // Write a manifest, then truncate it inside the last record.
        // Replay should stop cleanly and expose the valid prefix.
        let dir = TempDir::new().unwrap();
        let sst_dir = dir.path().join("sst");
        std::fs::create_dir_all(&sst_dir).unwrap();

        let file1 = make_live_sst(&sst_dir, 1, b"a", b"m");
        let mut vs = VersionSet::open(dir.path(), &sst_dir).unwrap();
        vs.apply(&[VersionEdit::AddFile {
            level: 0,
            file: Arc::clone(&file1),
        }])
        .unwrap();
        vs.apply(&[VersionEdit::SetLastSeq(50)]).unwrap();
        drop(vs);

        // Truncate 2 bytes off the end - enough to damage the final
        // record's checksum or tail.
        let path = dir.path().join("MANIFEST");
        let current = std::fs::metadata(&path).unwrap().len();
        OpenOptions::new()
            .write(true)
            .open(&path)
            .unwrap()
            .set_len(current - 2)
            .unwrap();

        let vs = VersionSet::open(dir.path(), &sst_dir).unwrap();
        let v = vs.current();
        // The AddFile should have survived (it was the first record);
        // the SetLastSeq may or may not survive depending on where the
        // truncation landed. Either way, we should NOT panic.
        assert!(v.levels[0].len() <= 1);
    }

    #[test]
    fn reset_record_clears_files_and_sets_wal_floor_atomically() {
        let dir = TempDir::new().unwrap();
        let sst_dir = dir.path().join("sst");
        std::fs::create_dir_all(&sst_dir).unwrap();

        let file1 = make_live_sst(&sst_dir, 1, b"a", b"m");
        let file2 = make_live_sst(&sst_dir, 2, b"n", b"z");
        {
            let mut vs = VersionSet::open(dir.path(), &sst_dir).unwrap();
            vs.apply(&[
                VersionEdit::AddFile {
                    level: 0,
                    file: Arc::clone(&file1),
                },
                VersionEdit::AddFile {
                    level: 1,
                    file: Arc::clone(&file2),
                },
                VersionEdit::SetLastSeq(50),
                VersionEdit::SetNextFileId(9),
            ])
            .unwrap();
            vs.apply(&[VersionEdit::Reset {
                next_file_id: 10,
                min_wal_id: 9,
            }])
            .unwrap();
        }

        let vs = VersionSet::open(dir.path(), &sst_dir).unwrap();
        let v = vs.current();
        assert!(v.levels.iter().all(Vec::is_empty));
        assert_eq!(v.last_seq, 0);
        assert_eq!(v.next_file_id, 10);
        assert_eq!(v.min_wal_id, 9);
    }

    #[test]
    fn open_truncates_truncated_manifest_tail_before_append() {
        let dir = TempDir::new().unwrap();
        let sst_dir = dir.path().join("sst");
        std::fs::create_dir_all(&sst_dir).unwrap();

        {
            let mut vs = VersionSet::open(dir.path(), &sst_dir).unwrap();
            vs.apply(&[VersionEdit::SetLastSeq(7)]).unwrap();
            vs.apply(&[VersionEdit::SetLastSeq(11)]).unwrap();
        }

        let path = dir.path().join("MANIFEST");
        let current = std::fs::metadata(&path).unwrap().len();
        OpenOptions::new()
            .write(true)
            .open(&path)
            .unwrap()
            .set_len(current - 2)
            .unwrap();

        {
            let mut vs = VersionSet::open(dir.path(), &sst_dir).unwrap();
            assert_eq!(vs.current().last_seq, 7);
            vs.apply(&[VersionEdit::SetLastSeq(99)]).unwrap();
        }

        let vs = VersionSet::open(dir.path(), &sst_dir).unwrap();
        assert_eq!(vs.current().last_seq, 99);
    }

    #[test]
    fn open_truncates_corrupt_manifest_tail_before_append() {
        let dir = TempDir::new().unwrap();
        let sst_dir = dir.path().join("sst");
        std::fs::create_dir_all(&sst_dir).unwrap();

        {
            let mut vs = VersionSet::open(dir.path(), &sst_dir).unwrap();
            vs.apply(&[VersionEdit::SetLastSeq(7)]).unwrap();
            vs.apply(&[VersionEdit::SetLastSeq(11)]).unwrap();
        }

        let path = dir.path().join("MANIFEST");
        let checksum_offset = second_record_checksum_offset(&path);
        let mut data = std::fs::read(&path).unwrap();
        data[checksum_offset] ^= 0xFF;
        std::fs::write(&path, data).unwrap();

        {
            let mut vs = VersionSet::open(dir.path(), &sst_dir).unwrap();
            assert_eq!(vs.current().last_seq, 7);
            vs.apply(&[VersionEdit::SetLastSeq(99)]).unwrap();
        }

        let vs = VersionSet::open(dir.path(), &sst_dir).unwrap();
        assert_eq!(vs.current().last_seq, 99);
    }

    /// The manifest is an append-only log, so without a rewrite trigger a
    /// database that runs for years replays every edit it ever made. This
    /// applies far more edits than the trigger allows and asserts the
    /// file stays bounded rather than growing with the history.
    #[test]
    fn a_long_running_manifest_stays_bounded_instead_of_growing_forever() {
        let dir = TempDir::new().unwrap();
        let sst_dir = dir.path().join("sst");
        std::fs::create_dir_all(&sst_dir).unwrap();
        let path = dir.path().join("MANIFEST");

        let mut vs = VersionSet::open(dir.path(), &sst_dir).unwrap();
        // `SetNextFileId`, not `SetLastSeq`: it is the one edit that
        // does not force a sync (`requires_manifest_sync`), and it grows
        // the log by the same record size, so it exercises exactly the
        // growth-and-rewrite behaviour under test without paying an
        // fsync per iteration. That distinction is the whole cost here.
        // At roughly 17 bytes a record this appends about 136 KiB
        // against a 64 KiB rewrite threshold, so the assertion below can
        // only pass if the log was rewritten at least twice.
        //
        // With the syncing edit this was 8,000 fsyncs, which a Windows
        // CI disk served at about 75 ms each and took past ten minutes,
        // while Linux finished it in milliseconds.
        const EDITS: u64 = 8_000;
        for id in 1..=EDITS {
            vs.apply(&[VersionEdit::SetNextFileId(id)]).unwrap();
        }
        drop(vs);

        let len = std::fs::metadata(&path).unwrap().len();
        let bound = VersionSet::compact_threshold(0);
        assert!(
            len <= bound,
            "manifest grew to {len} bytes against a {bound}-byte bound: \
             an append-only log that is never rewritten grows without limit"
        );

        // And it still replays to the state those edits describe.
        let reopened = VersionSet::open(dir.path(), &sst_dir).unwrap();
        assert_eq!(reopened.current().next_file_id, EDITS);
    }

    #[test]
    fn a_manifest_from_a_newer_format_is_refused() {
        let mut stamp = VersionSet::encode_stamp();
        stamp[7] = MANIFEST_FORMAT_V1 + 1;
        let checksum = checksum::manifest_record(0, &stamp[0..8]);
        stamp[8..12].copy_from_slice(&checksum.to_le_bytes());

        let err = VersionSet::stamp_len(&stamp).expect_err("a newer format must not be parsed");
        assert!(err.to_string().contains("newer regolith"), "{err}");
    }

    #[test]
    fn compact_manifest_rewrites_to_canonical_form() {
        // Apply many edits, then compact. The resulting manifest
        // should replay to the same version.
        let dir = TempDir::new().unwrap();
        let sst_dir = dir.path().join("sst");
        std::fs::create_dir_all(&sst_dir).unwrap();

        let file1 = make_live_sst(&sst_dir, 1, b"a", b"c");
        let file2 = make_live_sst(&sst_dir, 2, b"d", b"f");

        let mut vs = VersionSet::open(dir.path(), &sst_dir).unwrap();
        vs.apply(&[VersionEdit::Reset {
            next_file_id: 7,
            min_wal_id: 7,
        }])
        .unwrap();
        vs.apply(&[VersionEdit::AddFile {
            level: 0,
            file: Arc::clone(&file1),
        }])
        .unwrap();
        vs.apply(&[VersionEdit::AddFile {
            level: 1,
            file: Arc::clone(&file2),
        }])
        .unwrap();
        vs.apply(&[VersionEdit::SetLastSeq(500), VersionEdit::SetNextFileId(99)])
            .unwrap();

        let pre_size = std::fs::metadata(dir.path().join("MANIFEST"))
            .unwrap()
            .len();
        vs.compact_manifest().unwrap();
        let post_size = std::fs::metadata(dir.path().join("MANIFEST"))
            .unwrap()
            .len();
        // Compaction produces a single snapshot, so it is typically
        // not larger than the history it replaced.
        assert!(post_size <= pre_size + 64);

        drop(vs);
        let vs = VersionSet::open(dir.path(), &sst_dir).unwrap();
        let v = vs.current();
        assert_eq!(v.levels[0].len(), 1);
        assert_eq!(v.levels[1].len(), 1);
        assert_eq!(v.last_seq, 500);
        assert_eq!(v.next_file_id, 99);
        assert_eq!(v.min_wal_id, 7);
    }
}