limnifs_write/

lib.rs

//! `LimniFS` writer pipeline — directory tree to `.lim` image.
//!
//! The writer takes a real directory tree and produces a valid `.lim`
//! manifest artifact with inlined metadata. Files at or below the
//! inline threshold (4 KiB) are stored as inline data in their inodes;
//! larger files are stored as drops packed into a single slab.
//!
//! ## Usage
//!
//! ```no_run
//! use std::path::Path;
//! use limnifs_write::write_directory;
//! # fn main() -> Result<(), Box<dyn std::error::Error>> {
//! let artifact = write_directory(Path::new("/path/to/dir"))?;
//! std::fs::write("output.lim", &artifact.bytes)?;
//! # Ok(())
//! # }
//! ```

#![forbid(unsafe_code)]
#![warn(clippy::pedantic)]

pub mod chunker;
pub mod classifier;
pub mod compaction;
pub mod delta_builder;
pub mod flatten;
pub mod turnover;

use std::collections::HashMap;
use std::path::{Path, PathBuf};

use crate::chunker::FastCDC;
use limnifs_core::{
    compute_merkle_root, hash_empty_section, hash_section, ManifestHeader, SectionHashes,
    FEATURE_FLAGS_SECTION_VERSION, HISTORY_SECTION_VERSION, METADATA_REFERENCE_SECTION_VERSION,
    SLAB_INDEX_SECTION_VERSION,
};
use limnifs_format::{ManifestRoot, SlabId};

/// Inline-data threshold: files at or below this size get inline data
/// in their inode. Larger files are stored as drops in a slab.
pub const INLINE_THRESHOLD: usize = 4096;

/// Result of writing a directory tree.
#[derive(Clone, Debug)]
pub struct WriteArtifact {
    pub bytes: Vec<u8>,
    pub merkle_root: ManifestRoot,
    pub slab_bytes: Option<Vec<u8>>,
    pub slab_locator: Option<String>,
    pub inode_count: usize,
    pub file_count: usize,
    pub dir_count: usize,
    pub drop_count: usize,
    /// Inode number of the root directory (i.e. the inode that
    /// represents the source directory itself, not a child of it).
    /// Always a directory and always referenced by the inlined
    /// metadata blob's directory inode table.
    pub root_inode_number: u64,
}

/// Error during writing.
#[derive(Debug)]
pub enum WriteError {
    Io(std::io::Error),
}

impl std::fmt::Display for WriteError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Io(e) => write!(f, "I/O error: {e}"),
        }
    }
}

impl std::error::Error for WriteError {}

impl From<std::io::Error> for WriteError {
    fn from(e: std::io::Error) -> Self {
        Self::Io(e)
    }
}

/// Walk a directory tree and produce a valid `.lim` manifest artifact
/// with inlined metadata. Files at or below [`INLINE_THRESHOLD`] bytes
/// are stored inline; larger files are packed into a single slab as
/// content-addressed drops.
///
/// File contents (read, `FastCDC` chunk, `BLAKE3` hash, `LZ4` compress) are
/// processed in parallel across `CPU` cores via `rayon`. The directory
/// tree walk and slab assembly remain sequential so the output is
/// deterministic.
///
/// # Errors
///
/// Returns [`WriteError::Io`] for filesystem errors.
pub fn write_directory(root: &Path) -> Result<WriteArtifact, WriteError> {
    use rayon::prelude::*;

    let mut ctx = WriteContext::new();

    // Phase 1: walk the tree SEQUENTIALLY (deterministic inode
    // allocation). Collect files that need chunking (> INLINE_THRESHOLD)
    // for parallel processing.
    let root_inode_number = ctx.walk(root)?;
    ctx.root_inode_number = root_inode_number;

    // Phase 2: process each pending chunked file in PARALLEL via rayon.
    // This is the CPU-heavy path: file read + FastCDC + BLAKE3 + LZ4.
    // Each file is independent; results are collected in original order
    // so the output stays deterministic.
    let pending = std::mem::take(&mut ctx.pending_files);
    if !pending.is_empty() {
        let chunker = ctx.chunker.clone();
        let classifier = ctx.classifier;
        let results: Vec<ChunkedFileResult> = pending
            .par_iter()
            .map(|pf| process_file(pf, &chunker, classifier))
            .collect::<Result<Vec<_>, _>>()?;

        // Phase 3: merge results SEQUENTIALLY into drops + inodes.
        // Dedup happens here so the slab layout is deterministic.
        for (pf, result) in pending.iter().zip(results) {
            ctx.merge_chunked_file(pf, result);
        }
    }

    let artifact = ctx.assemble();
    Ok(artifact)
}

/// One chunk of a file before dedup: (`drop_id`, `plaintext`, `compressed`, `codec`).
type RawDrop = ([u8; 32], Vec<u8>, Vec<u8>, u8);
/// Result of parallel file processing: the drop data (uncompressed,
struct ChunkedFileResult {
    drops: Vec<RawDrop>, // (id, plaintext, compressed, codec)
    slices: Vec<PendingSlice>,
}

/// Process a single file's contents (CPU-heavy work that runs in a
/// rayon worker thread). Returns the unique chunks and slice map.
fn process_file(
    pf: &PendingFile,
    chunker: &FastCDC,
    classifier: classifier::Classifier,
) -> Result<ChunkedFileResult, WriteError> {
    let data = std::fs::read(&pf.path)?;
    let file_len = data.len();
    let chunks = chunker.chunk_slice(&data);
    let mut drops = Vec::with_capacity(chunks.len());
    let mut slices = Vec::with_capacity(chunks.len());
    let mut file_offset: u64 = 0;

    for chunk in chunks {
        let chunk_len = u64::try_from(chunk.len()).expect("chunk len fits u64");
        let drop_id = hash_section(chunk);

        let class = classifier.classify(chunk);
        let codec_id = match class {
            classifier::Class::Binary => limnifs_core::codec::best_binary_codec(),
            classifier::Class::Text | classifier::Class::Code => {
                limnifs_core::codec::best_compressible_codec()
            }
            _ => limnifs_core::codec::CODEC_STORE,
        };
        let compressed = if codec_id == limnifs_core::codec::CODEC_STORE {
            chunk.to_vec()
        } else {
            limnifs_core::codec::compress(codec_id, chunk).unwrap_or_else(|_| chunk.to_vec())
        };

        drops.push((drop_id, chunk.to_vec(), compressed, codec_id));
        slices.push(PendingSlice {
            drop_id,
            file_byte_start: file_offset,
            file_byte_end: file_offset + chunk_len,
        });
        file_offset += chunk_len;
    }
    let _ = file_len;
    Ok(ChunkedFileResult { drops, slices })
}

struct PendingDrop {
    id: [u8; 32],
    plaintext: Vec<u8>,
    compressed: Vec<u8>,
    codec: u8,
    offset_in_window: u32,
}

impl PendingDrop {
    /// The byte length stored in the slab's solid window. Equals
    /// `plaintext.len()` for store codec, or the compressed size for
    /// LZ4.
    fn len_in_window(&self) -> u32 {
        u32::try_from(self.compressed.len()).expect("compressed fits u32")
    }

    /// The original (decompressed) byte length.
    fn plaintext_len(&self) -> u32 {
        u32::try_from(self.plaintext.len()).expect("plaintext fits u32")
    }
}

/// One slice of a file backed by drops. Records which drop holds
/// this slice's bytes and which byte range of the original file
/// the slice covers. The slice always spans the entire drop (the
/// chunker never splits a drop across multiple slices).
struct PendingSlice {
    drop_id: [u8; 32],
    file_byte_start: u64,
    file_byte_end: u64,
}

/// A file that needs chunking (> `INLINE_THRESHOLD`). Collected during
/// the sequential tree walk and processed in parallel by `rayon`.
struct PendingFile {
    inode_number: u64,
    path: PathBuf,
    mtime_ns: u64,
    file_len: u64,
}

struct PendingInode {
    number: u64,
    mode: u32,
    mtime_ns: u64,
    content: PendingContent,
}

enum PendingContent {
    Inline(Vec<u8>),
    DropBacked {
        file_len: u64,
        slices: Vec<PendingSlice>,
    },
    Directory(Vec<(String, u64, u8)>),
}

struct DirNode {
    entries: Vec<(String, u64, u8)>,
    bytes: Vec<u8>,
    hash: [u8; 32],
}

struct WriteContext {
    next_inode: u64,
    inodes: Vec<PendingInode>,
    dir_nodes: Vec<DirNode>,
    drops: Vec<PendingDrop>,
    drop_index: HashMap<[u8; 32], (u32, u32)>,
    pending_files: Vec<PendingFile>,
    file_count: usize,
    dir_count: usize,
    root_inode_number: u64,
    chunker: FastCDC,
    classifier: classifier::Classifier,
}

impl WriteContext {
    fn new() -> Self {
        Self {
            next_inode: 1,
            inodes: Vec::new(),
            dir_nodes: Vec::new(),
            drops: Vec::new(),
            drop_index: HashMap::new(),
            pending_files: Vec::new(),
            file_count: 0,
            dir_count: 0,
            root_inode_number: 0,
            chunker: FastCDC::default(),
            classifier: classifier::Classifier,
        }
    }

    fn alloc_inode(&mut self) -> u64 {
        let n = self.next_inode;
        self.next_inode += 1;
        n
    }

    /// Merge a parallel-processed chunked file's results into the
    /// context. Dedup: only new `DropId`s get added to the drops list.
    fn merge_chunked_file(&mut self, pf: &PendingFile, result: ChunkedFileResult) {
        for (drop_id, plaintext, compressed, codec) in result.drops {
            if !self.drop_index.contains_key(&drop_id) {
                let offset = self
                    .drops
                    .iter()
                    .map(PendingDrop::len_in_window)
                    .sum::<u32>();
                let len = u32::try_from(compressed.len()).unwrap_or(0);
                self.drops.push(PendingDrop {
                    id: drop_id,
                    plaintext,
                    compressed,
                    codec,
                    offset_in_window: offset,
                });
                self.drop_index.insert(drop_id, (offset, len));
            }
        }
        self.inodes.push(PendingInode {
            number: pf.inode_number,
            mode: 0o100_644,
            mtime_ns: pf.mtime_ns,
            content: PendingContent::DropBacked {
                file_len: pf.file_len,
                slices: result.slices,
            },
        });
    }

    /// Apply the seine classifier to a chunk and compress it if the
    /// class is compressible. Text, Code, and Binary drops get LZ4;
    /// Compressed, Media, and Sparse drops stay as store (re-compressing
    /// already-compressed data wastes CPU for no gain).
    /// Apply the seine classifier to a chunk and compress it if the
    /// class is compressible. Text, Code, and Binary drops get LZ4;
    /// Compressed, Media, and Sparse drops stay as store.
    ///
    /// Kept for API compatibility; the parallel writer uses
    /// [`process_file`] which inlines this logic.
    #[allow(dead_code)]
    fn deepen_drop(&self, drop_id: [u8; 32], plaintext: &[u8]) -> PendingDrop {
        let class = self.classifier.classify(plaintext);
        let (codec, compressed) = match class {
            classifier::Class::Text | classifier::Class::Code | classifier::Class::Binary => {
                let c = limnifs_core::codec::compress_lz4_with_size(plaintext);
                (limnifs_core::codec::CODEC_LZ4, c)
            }
            _ => (limnifs_core::codec::CODEC_STORE, plaintext.to_vec()),
        };
        PendingDrop {
            id: drop_id,
            plaintext: plaintext.to_vec(),
            compressed,
            codec,
            offset_in_window: 0,
        }
    }

    fn walk(&mut self, path: &Path) -> Result<u64, WriteError> {
        let meta = std::fs::symlink_metadata(path)?;
        let file_type = meta.file_type();
        let mtime_ns = meta
            .modified()
            .ok()
            .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
            .map_or(0u128, |d| d.as_nanos());
        let mtime_ns: u64 = mtime_ns.try_into().unwrap_or(0);

        if file_type.is_dir() {
            self.dir_count += 1;
            let inode_number = self.alloc_inode();
            let mut entries: Vec<(String, u64, u8)> = Vec::new();

            for entry in std::fs::read_dir(path)? {
                let entry = entry?;
                let name = entry.file_name().to_string_lossy().into_owned();
                let child_path = entry.path();
                let child_inode = self.walk(&child_path)?;
                let child_meta = entry.metadata()?;
                let entry_type = if child_meta.is_dir() { 0x02 } else { 0x01 };
                entries.push((name, child_inode, entry_type));
            }

            entries.sort_by(|a, b| a.0.cmp(&b.0));
            let dir_node = encode_dir_node(&entries);
            self.dir_nodes.push(dir_node);
            self.inodes.push(PendingInode {
                number: inode_number,
                mode: 0o040_755,
                mtime_ns,
                content: PendingContent::Directory(entries),
            });
            Ok(inode_number)
        } else if file_type.is_file() {
            self.file_count += 1;
            let inode_number = self.alloc_inode();
            let file_len = meta.len();

            if file_len <= u64::try_from(INLINE_THRESHOLD).unwrap_or(u64::MAX) {
                let data = std::fs::read(path)?;
                self.inodes.push(PendingInode {
                    number: inode_number,
                    mode: 0o100_644,
                    mtime_ns,
                    content: PendingContent::Inline(data),
                });
            } else {
                // Defer to parallel processing — collect the file info.
                self.pending_files.push(PendingFile {
                    inode_number,
                    path: path.to_path_buf(),
                    mtime_ns,
                    file_len,
                });
            }
            Ok(inode_number)
        } else {
            Err(WriteError::Io(std::io::Error::new(
                std::io::ErrorKind::Unsupported,
                format!("unsupported file type: {}", path.display()),
            )))
        }
    }

    fn assemble(self) -> WriteArtifact {
        let inode_count = self.inodes.len();
        let dir_count = self.dir_count;
        let drop_count = self.drops.len();

        let (slab_bytes, slab_id, slab_locator) = if self.drops.is_empty() {
            (None, None, None)
        } else {
            let (bytes, id) = encode_slab(&self.drops);
            let locator = "file:slab-0.bin".to_owned();
            (Some(bytes), Some(id), Some(locator))
        };

        let mut metadata_blob = Vec::new();
        metadata_blob.extend_from_slice(&u32::try_from(self.inodes.len()).unwrap().to_le_bytes());
        for inode in &self.inodes {
            self.encode_inode(&mut metadata_blob, inode);
        }
        metadata_blob
            .extend_from_slice(&u32::try_from(self.dir_nodes.len()).unwrap().to_le_bytes());
        for node in &self.dir_nodes {
            metadata_blob.extend_from_slice(&node.bytes);
        }

        let mut manifest = Vec::new();

        let header_start = manifest.len();
        manifest.extend_from_slice(&ManifestHeader::current().to_bytes());
        let header_end = manifest.len();

        let flags_start = manifest.len();
        manifest.push(FEATURE_FLAGS_SECTION_VERSION);
        manifest.extend_from_slice(&0u32.to_le_bytes());
        let flags_end = manifest.len();

        let meta_ref_start = manifest.len();
        manifest.push(METADATA_REFERENCE_SECTION_VERSION);
        let metadata_hash = hash_section(&metadata_blob);
        manifest.extend_from_slice(&metadata_hash);
        manifest.extend_from_slice(&0u32.to_le_bytes());
        let inline_len = u32::try_from(metadata_blob.len()).expect("metadata fits u32");
        manifest.extend_from_slice(&inline_len.to_le_bytes());
        manifest.extend_from_slice(&metadata_blob);
        let meta_ref_end = manifest.len();

        let slab_index_start = manifest.len();
        manifest.push(SLAB_INDEX_SECTION_VERSION);
        if let (Some(id), Some(loc)) = (&slab_id, &slab_locator) {
            manifest.extend_from_slice(&1u32.to_le_bytes());
            manifest.extend_from_slice(&id.to_bytes());
            manifest.extend_from_slice(&1u32.to_le_bytes());
            let loc_bytes = loc.as_bytes();
            let loc_len = u32::try_from(loc_bytes.len()).expect("locator fits u32");
            manifest.extend_from_slice(&loc_len.to_le_bytes());
            manifest.extend_from_slice(loc_bytes);
        } else {
            manifest.extend_from_slice(&0u32.to_le_bytes());
        }
        let slab_index_end = manifest.len();

        let history_start = manifest.len();
        manifest.push(HISTORY_SECTION_VERSION);
        manifest.extend_from_slice(&1u32.to_le_bytes());
        manifest.push(0x01);
        manifest.extend_from_slice(&0u64.to_le_bytes());
        manifest.extend_from_slice(&0u32.to_le_bytes());
        manifest.extend_from_slice(&0u32.to_le_bytes());
        let history_end = manifest.len();

        let hashes = SectionHashes {
            metadata: metadata_hash,
            format_header: hash_section(&manifest[header_start..header_end]),
            feature_flags: hash_section(&manifest[flags_start..flags_end]),
            metadata_reference: hash_section(&manifest[meta_ref_start..meta_ref_end]),
            slab_index: hash_section(&manifest[slab_index_start..slab_index_end]),
            crypto_params: hash_empty_section(),
            ec_params: hash_empty_section(),
            dms_policy: hash_empty_section(),
            delta_linkage: hash_empty_section(),
            history: hash_section(&manifest[history_start..history_end]),
        };
        let merkle_root = compute_merkle_root(&hashes);

        WriteArtifact {
            bytes: manifest,
            merkle_root,
            slab_bytes,
            slab_locator,
            inode_count,
            file_count: self.file_count,
            dir_count,
            drop_count,
            root_inode_number: self.root_inode_number,
        }
    }

    fn encode_inode(&self, out: &mut Vec<u8>, inode: &PendingInode) {
        out.extend_from_slice(&inode.number.to_le_bytes());
        out.extend_from_slice(&inode.mode.to_le_bytes());
        out.extend_from_slice(&0u32.to_le_bytes());
        out.extend_from_slice(&0u32.to_le_bytes());
        out.extend_from_slice(&inode.mtime_ns.to_le_bytes());
        out.extend_from_slice(&inode.mtime_ns.to_le_bytes());
        out.extend_from_slice(&1u32.to_le_bytes());
        match &inode.content {
            PendingContent::Inline(data) => {
                out.push(0x04);
                let len = u32::try_from(data.len()).expect("data fits u32");
                out.extend_from_slice(&len.to_le_bytes());
                out.extend_from_slice(data);
            }
            PendingContent::DropBacked { file_len, slices } => {
                out.push(0x00);
                let slice_count = u32::try_from(slices.len()).expect("slice count fits u32");
                out.extend_from_slice(&slice_count.to_le_bytes());
                for slice in slices {
                    out.extend_from_slice(&slice.file_byte_start.to_le_bytes());
                    out.extend_from_slice(&slice.file_byte_end.to_le_bytes());
                    out.extend_from_slice(&slice.drop_id);
                    // drop_byte_start = 0 (slice covers the whole drop)
                    out.extend_from_slice(&0u32.to_le_bytes());
                    // drop_byte_len = the byte length of this slice in the
                    // drop's decompressed plaintext. Each slice maps to
                    // exactly one chunk, so this equals the file range.
                    let drop_byte_len = u32::try_from(slice.file_byte_end - slice.file_byte_start)
                        .expect("slice range fits u32");
                    out.extend_from_slice(&drop_byte_len.to_le_bytes());
                }
                let _ = file_len;
            }
            PendingContent::Directory(entries) => {
                out.push(0x00);
                let node = self
                    .dir_nodes
                    .iter()
                    .find(|n| n.entries == *entries)
                    .expect("directory node must exist");
                out.extend_from_slice(&node.hash);
            }
        }
    }
}

fn encode_dir_node(entries: &[(String, u64, u8)]) -> DirNode {
    let mut bytes = Vec::new();
    bytes.push(1u8);
    let count = u32::try_from(entries.len()).expect("entry count fits u32");
    bytes.extend_from_slice(&count.to_le_bytes());
    for (name, inode_number, entry_type) in entries {
        let name_bytes = name.as_bytes();
        let name_len = u32::try_from(name_bytes.len()).expect("name fits u32");
        bytes.extend_from_slice(&name_len.to_le_bytes());
        bytes.extend_from_slice(name_bytes);
        bytes.extend_from_slice(&inode_number.to_le_bytes());
        bytes.push(*entry_type);
    }
    let hash = hash_section(&bytes);
    DirNode {
        entries: entries.to_vec(),
        bytes,
        hash,
    }
}

fn encode_slab(drops: &[PendingDrop]) -> (Vec<u8>, SlabId) {
    let mut drop_records = Vec::new();
    let mut solid_window = Vec::new();

    for drop in drops {
        let plaintext_len = drop.plaintext_len();
        let window_len = drop.len_in_window();
        drop_records.extend_from_slice(&drop.id);
        drop_records.extend_from_slice(&plaintext_len.to_le_bytes());
        // representation: (codec, aead=0, ec=0)
        drop_records.extend_from_slice(&[drop.codec, 0x00, 0x00]);
        drop_records.push(0x00); // solid_window_index
        drop_records.extend_from_slice(&drop.offset_in_window.to_le_bytes());
        drop_records.extend_from_slice(&window_len.to_le_bytes());
        solid_window.extend_from_slice(&drop.compressed);
    }

    let slab_content = [&drop_records[..], &solid_window[..]].concat();
    let slab_hash = hash_section(&slab_content);
    let slab_id = SlabId::new(0, slab_hash);

    let total_length = 56 + slab_content.len();
    let mut slab_bytes = Vec::with_capacity(total_length);
    slab_bytes.extend_from_slice(b"LIM1");
    slab_bytes.extend_from_slice(&1u16.to_le_bytes());
    slab_bytes.extend_from_slice(&slab_id.to_bytes());
    slab_bytes.extend_from_slice(&(total_length as u64).to_le_bytes());
    slab_bytes.push(0x00);
    slab_bytes.push(0x00);
    slab_bytes.extend_from_slice(&slab_content);

    (slab_bytes, slab_id)
}

#[cfg(test)]
fn pseudo_random_bytes(seed: u64, count: usize) -> Vec<u8> {
    let mut state = seed;
    let mut out = Vec::with_capacity(count);
    for _ in 0..count {
        state = state
            .wrapping_mul(6_364_136_223_846_793_005)
            .wrapping_add(1_442_695_040_888_963_407);
        out.push(u8::try_from(state >> 56).expect("fits u8"));
    }
    out
}

#[cfg(test)]
mod tests {
    use super::*;
    use limnifs_core::ManifestCursor;

    #[test]
    fn write_empty_directory() {
        let temp =
            std::env::temp_dir().join(format!("limnifs-write-test-{}-empty", std::process::id()));
        std::fs::create_dir_all(&temp).expect("create temp dir");
        let artifact = write_directory(&temp).expect("write succeeds");
        std::fs::remove_dir_all(&temp).ok();
        assert!(artifact.inode_count >= 1);
        assert_eq!(artifact.file_count, 0);
        assert_eq!(artifact.dir_count, 1);
        assert!(artifact.slab_bytes.is_none());
    }

    #[test]
    fn write_small_file_inline() {
        let temp =
            std::env::temp_dir().join(format!("limnifs-write-test-{}-small", std::process::id()));
        std::fs::create_dir_all(&temp).expect("create temp dir");
        std::fs::write(temp.join("hello.txt"), b"hello world").expect("write file");
        let artifact = write_directory(&temp).expect("write succeeds");
        std::fs::remove_dir_all(&temp).ok();
        assert_eq!(artifact.file_count, 1);
        assert!(artifact.slab_bytes.is_none());
        assert_eq!(artifact.drop_count, 0);
    }

    #[test]
    fn write_large_file_uses_slab() {
        let temp =
            std::env::temp_dir().join(format!("limnifs-write-test-{}-large", std::process::id()));
        std::fs::create_dir_all(&temp).expect("create temp dir");
        let large_data = vec![0xABu8; INLINE_THRESHOLD + 100];
        std::fs::write(temp.join("big.bin"), &large_data).expect("write big");
        let artifact = write_directory(&temp).expect("write succeeds");
        std::fs::remove_dir_all(&temp).ok();
        assert_eq!(artifact.drop_count, 1);
        assert!(artifact.slab_bytes.is_some());
        assert!(artifact.slab_locator.is_some());
    }

    #[test]
    fn write_mixed_inline_and_large() {
        let temp =
            std::env::temp_dir().join(format!("limnifs-write-test-{}-mix", std::process::id()));
        std::fs::create_dir_all(&temp).expect("create temp dir");
        std::fs::write(temp.join("small.txt"), b"tiny").expect("write small");
        std::fs::write(temp.join("large.bin"), vec![0xCDu8; INLINE_THRESHOLD * 2])
            .expect("write large");
        let artifact = write_directory(&temp).expect("write succeeds");
        std::fs::remove_dir_all(&temp).ok();
        assert_eq!(artifact.file_count, 2);
        assert_eq!(artifact.drop_count, 1);
        assert!(artifact.slab_bytes.is_some());
    }

    #[test]
    fn deduplicates_identical_large_files() {
        let temp =
            std::env::temp_dir().join(format!("limnifs-write-test-{}-dedup", std::process::id()));
        std::fs::create_dir_all(&temp).expect("create temp dir");
        let data = vec![0x77u8; INLINE_THRESHOLD + 10];
        std::fs::write(temp.join("a.bin"), &data).expect("write a");
        std::fs::write(temp.join("b.bin"), &data).expect("write b");
        let artifact = write_directory(&temp).expect("write succeeds");
        std::fs::remove_dir_all(&temp).ok();
        assert_eq!(artifact.drop_count, 1);
    }

    #[test]
    fn write_and_verify_roundtrip() {
        let temp = std::env::temp_dir().join(format!(
            "limnifs-write-test-{}-roundtrip",
            std::process::id()
        ));
        std::fs::create_dir_all(&temp).expect("create temp dir");
        std::fs::write(temp.join("a.txt"), b"aaa").expect("write a");
        std::fs::write(temp.join("b.txt"), b"bbb").expect("write b");
        std::fs::create_dir_all(temp.join("sub")).expect("create sub");
        std::fs::write(temp.join("sub").join("c.txt"), b"ccc").expect("write c");
        let artifact = write_directory(&temp).expect("write succeeds");
        std::fs::remove_dir_all(&temp).ok();
        assert_eq!(artifact.file_count, 3);
        assert_eq!(artifact.dir_count, 2);

        let mut cursor = ManifestCursor::new(&artifact.bytes);
        limnifs_core::parse_manifest_header(&mut cursor).expect("header");
        limnifs_core::parse_feature_flags_section(&mut cursor).expect("flags");
        let meta_ref = limnifs_core::parse_metadata_reference(&mut cursor).expect("meta ref");
        assert!(meta_ref.is_inlined());
        let slab_index = limnifs_core::parse_slab_index(&mut cursor).expect("slab index");
        assert_eq!(slab_index.len(), 0);
        limnifs_core::parse_history(&mut cursor).expect("history");
    }

    #[test]
    fn write_deterministic() {
        let temp =
            std::env::temp_dir().join(format!("limnifs-write-test-{}-det", std::process::id()));
        std::fs::create_dir_all(&temp).expect("create temp dir");
        std::fs::write(temp.join("x.txt"), b"xxx").expect("write x");

        let a1 = write_directory(&temp).expect("first write");
        let a2 = write_directory(&temp).expect("second write");
        std::fs::remove_dir_all(&temp).ok();

        assert_eq!(a1.bytes, a2.bytes);
        assert_eq!(a1.merkle_root, a2.merkle_root);
    }

    #[test]
    fn slab_parses_correctly() {
        let temp =
            std::env::temp_dir().join(format!("limnifs-write-test-{}-slab", std::process::id()));
        std::fs::create_dir_all(&temp).expect("create temp dir");
        std::fs::write(temp.join("big.bin"), vec![0x11u8; INLINE_THRESHOLD + 1])
            .expect("write big");
        let artifact = write_directory(&temp).expect("write succeeds");
        std::fs::remove_dir_all(&temp).ok();

        let slab_bytes = artifact.slab_bytes.as_ref().expect("slab exists");
        let mut cursor = ManifestCursor::new(slab_bytes);
        let slab_header = limnifs_core::parse_slab_header(&mut cursor).expect("slab header parses");
        assert_eq!(slab_header.format_version, 1);
        assert!(!slab_header.is_sealed());
        assert!(!slab_header.has_erasure_coding());

        let drop_record =
            limnifs_core::parse_drop_record(&mut cursor, &slab_header).expect("drop record parses");
        assert_eq!(drop_record.plaintext_len as usize, INLINE_THRESHOLD + 1);
    }

    #[test]
    fn fastcdc_produces_multiple_chunks_for_large_files() {
        // A 1 MiB pseudo-random file should produce multiple drops
        // via FastCDC (default chunker uses 64 KiB min / 256 KiB avg).
        let temp = std::env::temp_dir().join(format!(
            "limnifs-write-test-{}-cdc-multi",
            std::process::id()
        ));
        std::fs::create_dir_all(&temp).expect("create temp dir");
        let data = pseudo_random_bytes(42, 1024 * 1024);
        std::fs::write(temp.join("big.bin"), &data).expect("write big");
        let artifact = write_directory(&temp).expect("write succeeds");
        std::fs::remove_dir_all(&temp).ok();
        assert!(
            artifact.drop_count > 1,
            "expected FastCDC to produce multiple drops for 1 MiB input, got {}",
            artifact.drop_count
        );
    }

    #[test]
    fn fastcdc_deduplicates_shared_substrings() {
        // Two files sharing a long middle section should produce
        // fewer drops than the sum of their individual chunk counts,
        // because the shared section's chunks deduplicate.
        let temp = std::env::temp_dir().join(format!(
            "limnifs-write-test-{}-cdc-dedup",
            std::process::id()
        ));
        std::fs::create_dir_all(&temp).expect("create temp dir");
        let shared = pseudo_random_bytes(7, 512 * 1024);
        let mut a = Vec::with_capacity(shared.len() + 1024);
        a.extend_from_slice(&pseudo_random_bytes(1, 1024));
        a.extend_from_slice(&shared);
        let mut b = Vec::with_capacity(shared.len() + 2048);
        b.extend_from_slice(&pseudo_random_bytes(2, 2048));
        b.extend_from_slice(&shared);
        std::fs::write(temp.join("a.bin"), &a).expect("write a");
        std::fs::write(temp.join("b.bin"), &b).expect("write b");

        // Baseline: each file alone.
        let temp_a = std::env::temp_dir().join(format!(
            "limnifs-write-test-{}-cdc-dedup-a",
            std::process::id()
        ));
        std::fs::create_dir_all(&temp_a).expect("create temp_a");
        std::fs::write(temp_a.join("a.bin"), &a).expect("write a");
        let artifact_a = write_directory(&temp_a).expect("a writes");
        std::fs::remove_dir_all(&temp_a).ok();

        let temp_b = std::env::temp_dir().join(format!(
            "limnifs-write-test-{}-cdc-dedup-b",
            std::process::id()
        ));
        std::fs::create_dir_all(&temp_b).expect("create temp_b");
        std::fs::write(temp_b.join("b.bin"), &b).expect("write b");
        let artifact_b = write_directory(&temp_b).expect("b writes");
        std::fs::remove_dir_all(&temp_b).ok();

        let artifact_both = write_directory(&temp).expect("both write");
        std::fs::remove_dir_all(&temp).ok();

        let sum_alone = artifact_a.drop_count + artifact_b.drop_count;
        assert!(
            artifact_both.drop_count < sum_alone,
            "expected dedup win: both together = {} drops, sum alone = {} drops",
            artifact_both.drop_count,
            sum_alone
        );
    }
}