Skip to main content

limnifs_write/
lib.rs

1//! `LimniFS` writer pipeline — directory tree to `.lim` image.
2//!
3//! The writer takes a real directory tree and produces a valid `.lim`
4//! manifest artifact with inlined metadata. Files at or below the
5//! inline threshold (4 KiB) are stored as inline data in their inodes;
6//! larger files are stored as drops packed into a single slab.
7//!
8//! ## Usage
9//!
10//! ```no_run
11//! use std::path::Path;
12//! use limnifs_write::write_directory;
13//! # fn main() -> Result<(), Box<dyn std::error::Error>> {
14//! let artifact = write_directory(Path::new("/path/to/dir"))?;
15//! std::fs::write("output.lim", &artifact.bytes)?;
16//! # Ok(())
17//! # }
18//! ```
19
20#![deny(unsafe_code)]
21#![allow(warnings)]
22
23pub mod chunker;
24pub mod classifier;
25pub mod compaction;
26pub mod config;
27pub mod delta_builder;
28pub mod dictionary;
29pub mod file_categorizer;
30use file_categorizer::FileCategorizer;
31pub mod flatten;
32pub mod progress;
33pub mod rw;
34#[cfg(feature = "sparse-index")]
35pub mod sparse_index;
36pub mod stream;
37pub mod turnover;
38
39pub use config::{
40    profile, CategorizerConfig, ChunkingConfig, CodecRegistry, CodecTunables, Defaults,
41    DictionaryConfig, EncryptionConfig, TournamentConfig, WriteConfig,
42};
43
44use std::collections::{HashMap, HashSet};
45use std::path::{Path, PathBuf};
46
47use crate::chunker::{Chunker, ParallelFastCDC};
48use limnifs_core::codec::CODEC_REFERENCED;
49use limnifs_core::dictionary_section::parse_dictionary_section;
50use limnifs_core::slab_store::SlabStore;
51use limnifs_core::{
52    compute_merkle_root, hash_empty_section, hash_section, parse_manifest_header, parse_slab_index,
53    ManifestCursor, ManifestHeader, SectionHashes, FEATURE_FLAGS_SECTION_VERSION,
54    HISTORY_SECTION_VERSION, INODE_FLAG_INLINE_DATA, INODE_FLAG_SHARED_INLINE,
55    METADATA_REFERENCE_SECTION_VERSION_2, SLAB_INDEX_SECTION_VERSION,
56};
57use limnifs_format::{ManifestRoot, SlabId};
58
59/// Inline-data threshold: files at or below this size get inline data
60/// in their inode. Larger files are stored as drops in a slab.
61pub const INLINE_THRESHOLD: usize = 4096;
62
63/// Above this size, mmap the input file instead of `std::fs::read`-ing
64/// it into a `Vec<u8>`. Keeps peak RSS bounded when packing huge files
65/// (multi-GiB source trees, ML models). Crossover is around 1 MiB on
66/// most filesystems — below that the syscall + VMA setup costs more
67/// than the read.
68pub const MMAP_READ_THRESHOLD: usize = 1024 * 1024;
69
70/// Maximum file size for the whole-file categorizer path. Files above
71/// this threshold use FastCDC chunking even when a categorizer claims
72/// them, enabling rayon parallelism across chunks. The categorizer's
73/// codec is still used per-chunk when possible.
74pub const WHOLE_FILE_MAX_SIZE: usize = 64 * 1024 * 1024;
75
76/// Maximum total length of a single slab file (header + content).
77/// Matches the reader's `DEFAULT_SLAB_MAX_BYTES` (spec §3.1) minus a
78/// safety margin so a slab that is full but not yet flushed cannot
79/// overrun the reader ceiling on the next drop.
80pub const MAX_SLAB_TOTAL_BYTES: usize = 60 * 1024 * 1024;
81
82/// Width of the slab header (magic + version + `SlabId` + `total_length` +
83/// `ec_descriptor` + `crypto_hint`). Must agree with
84/// `limnifs_core::slab::SLAB_HEADER_LEN`.
85const SLAB_HEADER_LEN: usize = 56;
86
87/// Default threshold at which the writer externalises the metadata
88/// blob to a sidecar file instead of inlining it in the manifest.
89/// Derived from the reader's inline ceiling
90/// (`limnifs_core::metadata_reference::DEFAULT_INLINE_METADATA_MAX_BYTES`,
91/// 1 MiB per spec §5.3) minus headroom, so the two constants cannot
92/// silently drift apart. Override per image via
93/// `WriteConfig::defaults::metadata_externalize_threshold` (issue #187).
94pub const METADATA_EXTERNALIZE_THRESHOLD: usize =
95    limnifs_core::metadata_reference::DEFAULT_INLINE_METADATA_MAX_BYTES as usize - 24 * 1024;
96
97/// Metadata-blob size above which the writer steps Brotli quality
98/// down to `METADATA_LARGE_BLOB_QUALITY`. Below this, q5's cost is
99/// negligible; above it, q5 starts to dominate create time on big
100/// inode trees (e.g. the 50 K-file tiny-files dataset).
101pub const METADATA_LARGE_BLOB_THRESHOLD: usize = 256 * 1024;
102
103/// Brotli quality for small metadata blobs (≤ `METADATA_LARGE_BLOB_THRESHOLD`).
104/// Best ratio; cost is in the noise on small inputs.
105pub const METADATA_SMALL_BLOB_QUALITY: i32 = 5;
106
107/// Brotli quality for large metadata blobs. q2 is much faster than q5
108/// on multi-MiB inputs; ratio on highly compressible inode data is
109/// within 5–10% of q5 (often identical) because metadata is dominated
110/// by long runs of repeated patterns.
111pub const METADATA_LARGE_BLOB_QUALITY: i32 = 2;
112
113/// One slab produced by the writer. The slab ordinal in `id` matches
114/// the slab's position in `WriteArtifact::slabs`.
115#[derive(Clone, Debug)]
116pub struct SlabArtifact {
117    pub id: SlabId,
118    pub bytes: Vec<u8>,
119    pub locator: String,
120    /// `DropIds` contained in this slab, in slab order. Used by callers
121    /// that need to know which slab holds which drop without re-parsing
122    /// the slab bytes.
123    pub drop_ids: Vec<[u8; 32]>,
124}
125
126/// Externalized metadata sidecar, present when the metadata blob
127/// exceeds [`METADATA_EXTERNALIZE_THRESHOLD`]. Callers must write
128/// `bytes` to `locator` next to the manifest file.
129#[derive(Clone, Debug)]
130pub struct MetadataSidecar {
131    pub bytes: Vec<u8>,
132    pub locator: String,
133}
134
135/// Result of writing a directory tree.
136#[derive(Clone, Debug)]
137pub struct WriteArtifact {
138    pub bytes: Vec<u8>,
139    pub merkle_root: ManifestRoot,
140    /// All slabs produced by the writer, in slab-ordinal order. Empty
141    /// when the source tree had no files > [`INLINE_THRESHOLD`].
142    pub slabs: Vec<SlabArtifact>,
143    /// External metadata sidecar, present when the metadata blob
144    /// exceeds [`METADATA_EXTERNALIZE_THRESHOLD`]. `None` means the
145    /// metadata is inlined in the manifest.
146    pub metadata_sidecar: Option<MetadataSidecar>,
147    pub inode_count: usize,
148    pub file_count: usize,
149    pub dir_count: usize,
150    pub drop_count: usize,
151    /// Inode number of the root directory (i.e. the inode that
152    /// represents the source directory itself, not a child of it).
153    /// Always a directory and always referenced by the inlined
154    /// metadata blob's directory inode table.
155    pub root_inode_number: u64,
156}
157
158impl WriteArtifact {
159    /// Convenience accessor for the single-slab case. Returns the
160    /// first slab's bytes if there is exactly one slab, else `None`.
161    /// Modern callers should iterate [`WriteArtifact::slabs`] directly.
162    #[must_use]
163    pub fn slab_bytes(&self) -> Option<&[u8]> {
164        if self.slabs.len() == 1 {
165            Some(&self.slabs[0].bytes)
166        } else {
167            None
168        }
169    }
170
171    /// Convenience accessor for the single-slab case.
172    #[must_use]
173    pub fn slab_locator(&self) -> Option<&str> {
174        if self.slabs.len() == 1 {
175            Some(&self.slabs[0].locator)
176        } else {
177            None
178        }
179    }
180}
181
182/// Error during writing.
183#[derive(Debug)]
184pub enum WriteError {
185    Io(std::io::Error),
186    /// The tree contains an entry type the writer deliberately does
187    /// not store (sockets, FIFOs, device nodes). Symlinks ARE
188    /// supported; everything else on a normal filesystem tree is
189    /// either a file, a directory, or this error.
190    UnsupportedFileType {
191        path: PathBuf,
192        kind: String,
193    },
194}
195
196impl std::fmt::Display for WriteError {
197    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
198        match self {
199            Self::Io(e) => write!(f, "I/O error: {e}"),
200            Self::UnsupportedFileType { path, kind } => write!(
201                f,
202                "unsupported file type ({kind}): {} — limnifs stores files, \
203                 directories, and symlinks; remove the entry or file an issue \
204                 if you need it carried",
205                path.display()
206            ),
207        }
208    }
209}
210
211impl std::error::Error for WriteError {}
212
213impl From<std::io::Error> for WriteError {
214    fn from(e: std::io::Error) -> Self {
215        Self::Io(e)
216    }
217}
218
219/// Walk a directory tree and produce a valid `.lim` manifest artifact
220/// with inlined metadata. Files at or below [`INLINE_THRESHOLD`] bytes
221/// are stored inline; larger files are packed into a single slab as
222/// content-addressed drops.
223///
224/// File contents (read, `FastCDC` chunk, `BLAKE3` hash, `LZ4` compress) are
225/// processed in parallel across `CPU` cores via `rayon`. The directory
226/// tree walk and slab assembly remain sequential so the output is
227/// deterministic.
228///
229/// # Errors
230///
231/// Returns [`WriteError::Io`] for filesystem errors.
232pub fn write_directory(root: &Path) -> Result<WriteArtifact, WriteError> {
233    write_directory_with_config(root, &WriteConfig::default_v0_1())
234}
235
236/// Pack a single named stream into a `.lim` image.
237///
238/// For callers that pipe data from a network socket, pipe, or generator
239/// and don't want to materialise the full content on disk before
240/// packing. The reader is consumed via [`Chunker::chunk_reader`] which
241/// bounds internal buffering at `max_chunk_size + 64 KiB`.
242///
243/// The resulting image has a root directory holding the named
244/// stream at `name` (path-relative; safe to use `/` for
245/// subdirectories — they're materialised in the metadata tree).
246/// Multi-entry streams should use [`stream::StreamWriter`]
247/// directly.
248///
249/// # Errors
250///
251/// Returns [`WriteError::Io`] on read failure or any writer-pipeline
252/// error.
253pub fn write_stream<R: std::io::Read>(
254    name: &str,
255    mut reader: R,
256    config: &WriteConfig,
257) -> Result<WriteArtifact, WriteError> {
258    let mut writer = crate::stream::StreamWriter::new(config)?;
259    writer.add_file(
260        name,
261        crate::stream::EntryMeta::new(0, 0o644),
262        &[],
263        &mut reader,
264    )?;
265    writer.finish()
266}
267
268/// Pack a directory tree as a **layer** on top of a base image.
269///
270/// Produces a `.lim` image whose drops are split into two sets:
271///
272/// - **Local drops** — chunks in `root` whose `DropId` is NOT in
273///   `base_image`'s drop set. These are compressed and stored in the
274///   layer's own slabs exactly as `write_directory_with_config`
275///   would store them.
276/// - **Referenced drops** — chunks whose `DropId` IS in the base.
277///   These are recorded only as `PendingSlice` references (so the
278///   metadata tree links them in); no slab bytes are emitted in the
279///   layer. The reader resolves them via the overlay chain.
280///
281/// The resulting manifest carries a `delta_linkage` section pointing
282/// at the base image's `ManifestRoot`, so any reader that supports
283/// overlay chains can extract the layer standalone or stacked on the
284/// base.
285///
286/// # Determinism
287///
288/// `write_layer` is deterministic given the same `base_image`, the
289/// same `root` content, and the same `config`. Two runs produce
290/// byte-identical layer images.
291///
292/// # Errors
293///
294/// Returns [`WriteError::Io`] on read failure or any writer-pipeline
295/// error.
296///
297/// # Example
298///
299/// ```no_run
300/// use limnifs_write::{write_layer, profile};
301///
302/// let base = std::path::Path::new("base.lim");
303/// let root = std::path::Path::new("./new-content");
304/// let cfg = profile::balanced();
305/// let artifact = write_layer(base, root, &cfg).expect("layer");
306/// // artifact.bytes is the layer manifest; slabs contain only NEW drops.
307/// ```
308pub fn write_layer(
309    base_image: &Path,
310    root: &Path,
311    config: &WriteConfig,
312) -> Result<WriteArtifact, WriteError> {
313    // Load the base image's drop set + manifest root.
314    let base_root = load_base_drop_index(base_image)?.1;
315    let base_drop_index: std::sync::Arc<dyn BaseDropSet> = {
316        #[cfg(feature = "sparse-index")]
317        {
318            match SparseBackedBaseIndex::open(base_image) {
319                Some(idx) => std::sync::Arc::new(idx),
320                None => std::sync::Arc::new(load_base_drop_index(base_image)?.0),
321            }
322        }
323        #[cfg(not(feature = "sparse-index"))]
324        {
325            std::sync::Arc::new(load_base_drop_index(base_image)?.0)
326        }
327    };
328
329    let mut ctx = WriteContext::new();
330    ctx.chunker = chunker_from_config(config)?;
331    // Adopt the base's trained dictionaries for this layer (the
332    // dictionary pass still gates on whether re-emitting them pays).
333    ctx.base_dictionaries = if config.dictionaries.enabled {
334        load_base_dictionary_section(base_image)?.map(crate::dictionary::adopt_from_section)
335    } else {
336        None
337    };
338    ctx.categorizers_disabled = config.categorizers.is_empty();
339    ctx.rw_mode = matches!(config.mode, crate::config::ImageMode::ReadWrite(_));
340    ctx.auto_turnover = config.turnover_threshold > 0;
341    ctx.collect_dict_samples = config.dictionaries.enabled;
342    ctx.inline_threshold = config.defaults.inline_threshold as usize;
343    ctx.metadata_externalize_threshold = config.defaults.metadata_externalize_threshold;
344    ctx.emit_shared_inline = config.defaults.shared_inline;
345    ctx.base_drop_index = Some(base_drop_index);
346    ctx.base_root = Some(base_root);
347
348    // Rest is identical to write_directory_with_config.
349    let root_inode_number = ctx.walk(root)?;
350    ctx.root_inode_number = root_inode_number;
351    write_directory_body(&mut ctx, config)?;
352    Ok(ctx.assemble())
353}
354
355/// Load every DropId present in a base image's slabs + the image's
356/// `ManifestRoot`. Used by `write_layer` to decide which chunks can
357/// be referenced rather than re-encoded.
358/// A base image's drop set, queried per chunk during layer writes.
359///
360/// The exact form is a `HashSet` built by opening every base slab.
361/// The sparse form (`SparseBackedBaseIndex`, `sparse-index` feature)
362/// keeps that cost OFF the layer build until first needed: a
363/// false-negative-free Bloom filter answers "definitely not in
364/// base" in O(1), and only a *probable* hit lazily loads the exact
365/// set. False positives are impossible to observe — the fallback
366/// is exact — so layer output is byte-identical either way.
367pub trait BaseDropSet: Send + Sync {
368    /// Exact-membership answer (may lazily load state on first
369    /// probable hit).
370    fn base_contains(&self, drop_id: &[u8; 32]) -> bool;
371}
372
373impl BaseDropSet for std::collections::HashSet<[u8; 32]> {
374    fn base_contains(&self, drop_id: &[u8; 32]) -> bool {
375        self.contains(drop_id)
376    }
377}
378
379/// Bloom-fronted base index with exact, lazily-loaded fallback
380/// (`sparse-index` feature). Reads the `<image>.sparse` sidecar
381/// emitted by [`emit_sparse_sidecar`]; the base's slabs are only
382/// opened if some chunk is a *probable* member — low-overlap layer
383/// builds never touch them.
384#[cfg(feature = "sparse-index")]
385pub struct SparseBackedBaseIndex {
386    bloom: crate::sparse_index::SparseIndexReader,
387    manifest_path: std::path::PathBuf,
388    exact: std::sync::OnceLock<std::collections::HashSet<[u8; 32]>>,
389}
390
391#[cfg(feature = "sparse-index")]
392impl SparseBackedBaseIndex {
393    /// Open `<base_image>.sparse`. `None` when the sidecar does not
394    /// exist (caller falls back to the exact set).
395    #[must_use]
396    pub fn open(base_image: &Path) -> Option<Self> {
397        let sidecar = base_image.with_extension("lim.sparse");
398        let bloom = crate::sparse_index::SparseIndexReader::from_file(&sidecar)?;
399        Some(Self {
400            bloom,
401            manifest_path: base_image.to_path_buf(),
402            exact: std::sync::OnceLock::new(),
403        })
404    }
405
406    fn load_exact(&self) -> &std::collections::HashSet<[u8; 32]> {
407        self.exact.get_or_init(|| {
408            // The base manifest tells us which slabs to open; from
409            // there it's the same enumeration `load_base_drop_index`
410            // performs.
411            let bytes = std::fs::read(&self.manifest_path).unwrap_or_default();
412            let mut cursor = ManifestCursor::new(&bytes);
413            let _ = parse_manifest_header(&mut cursor);
414            let _ = limnifs_core::parse_feature_flags_section(&mut cursor);
415            let _ = limnifs_core::parse_metadata_reference(&mut cursor);
416            let Ok(index) = parse_slab_index(&mut cursor) else {
417                return std::collections::HashSet::new();
418            };
419            match SlabStore::load_mmap(&self.manifest_path, &index) {
420                Ok(store) => store.drop_index_keys().copied().collect(),
421                Err(_) => std::collections::HashSet::new(),
422            }
423        })
424    }
425}
426
427#[cfg(feature = "sparse-index")]
428impl BaseDropSet for SparseBackedBaseIndex {
429    fn base_contains(&self, drop_id: &[u8; 32]) -> bool {
430        // Bloom false negatives are impossible: a "no" is final and
431        // costs one bit-probe. A "yes" may be the 1% false positive,
432        // so fall through to the exact set — output stays
433        // byte-identical to the exact-only path.
434        if !self.bloom.probably_contains(drop_id) {
435            return false;
436        }
437        self.load_exact().contains(drop_id)
438    }
439}
440
441/// Emit the `<image>.sparse` Bloom sidecar for a finished artifact
442/// (`sparse-index` feature). Subsequent `write_layer` builds over
443/// this image skip opening its slabs unless a chunk is probably
444/// present.
445///
446/// # Errors
447/// Returns [`WriteError::Io`] on serialisation or write failure.
448#[cfg(feature = "sparse-index")]
449pub fn emit_sparse_sidecar(artifact: &WriteArtifact, image_path: &Path) -> Result<(), WriteError> {
450    let all: std::collections::HashSet<[u8; 32]> = artifact
451        .slabs
452        .iter()
453        .flat_map(|s| s.drop_ids.iter().copied())
454        .collect();
455    let mut writer = crate::sparse_index::SparseIndexWriter::new(
456        all.len().max(1),
457        crate::sparse_index::DEFAULT_FPP,
458    );
459    writer.insert_all(&all);
460    let sidecar = image_path.with_extension("lim.sparse");
461    writer.write_to_file(&sidecar).map_err(WriteError::Io)
462}
463
464/// Load the base image's `dictionary_section`, if any. A layer can
465/// adopt these dictionaries instead of retraining from its own
466/// samples (tebako-style layered builds re-derive nearly the same
467/// dictionary). Walks the manifest in spec order with best-effort
468/// parses, mirroring the reader.
469fn load_base_dictionary_section(
470    base_image: &Path,
471) -> Result<Option<limnifs_core::dictionary_section::DictionarySection>, WriteError> {
472    let manifest_bytes = std::fs::read(base_image)?;
473    let mut cursor = ManifestCursor::new(&manifest_bytes);
474    let _ = parse_manifest_header(&mut cursor).map_err(io_core)?;
475    let _ = limnifs_core::parse_feature_flags_section(&mut cursor);
476    let _ = limnifs_core::parse_metadata_reference(&mut cursor);
477    let _ = parse_slab_index(&mut cursor);
478    let _ = limnifs_core::parse_history(&mut cursor);
479    if cursor.remaining_len() == 0 {
480        return Ok(None);
481    }
482    Ok(parse_dictionary_section(&mut cursor).ok())
483}
484
485fn load_base_drop_index(
486    base_image: &Path,
487) -> Result<(std::collections::HashSet<[u8; 32]>, [u8; 32]), WriteError> {
488    let manifest_bytes = std::fs::read(base_image)?;
489    let mut cursor = ManifestCursor::new(&manifest_bytes);
490    let _ = parse_manifest_header(&mut cursor).map_err(io_core)?;
491    // Walk sections in spec order: flags → metadata_reference → slab_index.
492    let _ = limnifs_core::parse_feature_flags_section(&mut cursor);
493    let _ = limnifs_core::parse_metadata_reference(&mut cursor);
494    let slab_index = parse_slab_index(&mut cursor).map_err(io_core)?;
495    let store = SlabStore::load_mmap(base_image, &slab_index).map_err(io_core)?;
496    let drop_set: std::collections::HashSet<[u8; 32]> = store.drop_index_keys().copied().collect();
497    // Re-derive the Merkle root from the base manifest's section
498    // bytes. The base's `ManifestRoot` is the canonical anchor for
499    // the layer's `delta_linkage.base_root` field — round-tripping
500    // through section hashes guarantees it matches what the base
501    // reported on its own assemble path.
502    let root = *compute_merkle_root_from_sections(&manifest_bytes).as_bytes();
503    Ok((drop_set, root))
504}
505
506/// Re-derive a manifest's `ManifestRoot` from its on-disk section
507/// bytes. Mirrors `flatten::compute_merkle_root_from_sections` but
508/// tolerates absent optional sections (returns hash_empty_section()
509/// for them). Used by `load_base_drop_index` to anchor a layer's
510/// `base_root` without re-instantiating a full manifest parser.
511fn compute_merkle_root_from_sections(manifest: &[u8]) -> ManifestRoot {
512    use limnifs_core::SectionHashes;
513    let mut cursor = ManifestCursor::new(manifest);
514    let header_start = 0;
515    if parse_manifest_header(&mut cursor).is_err() {
516        // Not a valid manifest; fall back to all-zero root.
517        return ManifestRoot::from_bytes([0u8; 32]);
518    }
519    let header_end = cursor.position();
520    // Optional sections — best-effort parse; failures hash as empty.
521    let flags_start = header_end;
522    let flags_end = match limnifs_core::parse_feature_flags_section(&mut cursor) {
523        Ok(_) => cursor.position(),
524        Err(_) => flags_start,
525    };
526    let meta_ref_start = flags_end;
527    let metadata_reference = match limnifs_core::parse_metadata_reference(&mut cursor) {
528        Ok(m) => Some(m),
529        Err(_) => None,
530    };
531    let meta_ref_end = cursor.position();
532    let slab_index_start = meta_ref_end;
533    let _ = parse_slab_index(&mut cursor);
534    let slab_index_end = cursor.position();
535    let history_start = slab_index_end;
536    let _ = limnifs_core::parse_history(&mut cursor);
537    let history_end = cursor.position();
538
539    let hashes = SectionHashes {
540        metadata: metadata_reference
541            .map(|m| m.metadata_hash)
542            .unwrap_or_else(hash_empty_section),
543        format_header: hash_section(&manifest[header_start..header_end]),
544        feature_flags: hash_section(&manifest[flags_start..flags_end]),
545        metadata_reference: hash_section(&manifest[meta_ref_start..meta_ref_end]),
546        slab_index: hash_section(&manifest[slab_index_start..slab_index_end]),
547        crypto_params: hash_empty_section(),
548        ec_params: hash_empty_section(),
549        dms_policy: hash_empty_section(),
550        delta_linkage: hash_empty_section(),
551        history: hash_section(&manifest[history_start..history_end]),
552    };
553    compute_merkle_root(&hashes)
554}
555
556/// Survey-form xattrs -> wire form. Namespace 0 (user-visible).
557#[cfg(feature = "xattr")]
558fn to_core_xattrs(raw: &[(String, Vec<u8>)]) -> Vec<limnifs_core::inode::XAttr> {
559    raw.iter()
560        .map(|(key, value)| limnifs_core::inode::XAttr {
561            namespace: 0,
562            key: key.clone(),
563            value: value.clone(),
564        })
565        .collect()
566}
567
568fn io_core(e: limnifs_core::CoreError) -> WriteError {
569    WriteError::Io(std::io::Error::other(format!("base image load: {e}")))
570}
571
572/// Shared body between `write_directory_with_config` and `write_layer`.
573/// Walks `ctx.pending_files` through `process_file` in parallel and
574/// merges the results back. Caller is responsible for `walk()` and
575/// `assemble()`.
576fn write_directory_body(ctx: &mut WriteContext, config: &WriteConfig) -> Result<(), WriteError> {
577    use rayon::prelude::*;
578
579    ctx.metadata_codec = config
580        .metadata_codec_id()
581        .unwrap_or(limnifs_core::codec::CODEC_BROTLI);
582
583    ctx.chunker = chunker_from_config(config)?;
584
585    let pending = std::mem::take(&mut ctx.pending_files);
586    if pending.is_empty() {
587        return Ok(());
588    }
589    ctx.inline_threshold = config.defaults.inline_threshold as usize;
590    ctx.metadata_externalize_threshold = config.defaults.metadata_externalize_threshold;
591    ctx.emit_shared_inline = config.defaults.shared_inline;
592    let chunker = ctx.chunker.clone();
593    let classifier = ctx.classifier;
594    let text_codec = config.text_codec_id().unwrap_or(0x04);
595    let binary_codec = config.binary_codec_id().unwrap_or(0x01);
596    let tunables = config.to_core_tunables();
597    let use_categorizers = !config.categorizers.is_empty();
598    let skip_chunking = config.skip_chunking;
599    let registry = config
600        .codec_registry()
601        .map_err(|e| WriteError::Io(std::io::Error::other(format!("codec registry: {e}"))))?;
602    let tournament_codec_ids: Vec<u8> = config
603        .tournament
604        .codecs
605        .iter()
606        .filter_map(|n| registry.lookup_by_name(n))
607        .collect();
608    let tournament_spec = TournamentSpec {
609        codec_ids: tournament_codec_ids,
610        min_size: config.tournament.min_size_threshold as usize,
611        skip_for_binary: config.tournament.skip_for_binary,
612        short_circuit_permille: config.tournament.short_circuit_threshold,
613    };
614    let base_drop_index: Option<&dyn BaseDropSet> = ctx.base_drop_index.as_deref();
615    let inline_threshold = ctx.inline_threshold;
616    let max_drop_size = config.defaults.max_drop_size as usize;
617    let seekable_drops = config.defaults.seekable_drops;
618    let seekable_drops = config.defaults.seekable_drops;
619    let results: Vec<ChunkedFileResult> = pending
620        .par_iter()
621        .map(|pf| {
622            process_file(
623                pf,
624                &chunker,
625                classifier,
626                text_codec,
627                binary_codec,
628                &tunables,
629                use_categorizers,
630                skip_chunking,
631                &tournament_spec,
632                base_drop_index,
633                inline_threshold,
634                max_drop_size,
635                seekable_drops,
636                config.categorizers.as_slice(),
637                &|name| {
638                    config
639                        .codec_registry()
640                        .ok()
641                        .and_then(|r| r.lookup_by_name(name))
642                },
643            )
644        })
645        .collect::<Result<Vec<_>, _>>()?;
646
647    for (pf, result) in pending.iter().zip(results) {
648        ctx.merge_chunked_file(pf, result);
649    }
650    ctx.train_and_apply_dictionary(&config.dictionaries);
651    Ok(())
652}
653
654/// Create an image with a custom [`WriteConfig`] (e.g. from a profile).
655pub fn write_directory_with_config(
656    root: &Path,
657    config: &WriteConfig,
658) -> Result<WriteArtifact, WriteError> {
659    let mut ctx = WriteContext::new();
660    ctx.chunker = chunker_from_config(config)?;
661    ctx.categorizers_disabled = config.categorizers.is_empty();
662    ctx.rw_mode = matches!(config.mode, crate::config::ImageMode::ReadWrite(_));
663    ctx.auto_turnover = config.turnover_threshold > 0;
664    ctx.collect_dict_samples = config.dictionaries.enabled;
665
666    write_directory_streaming(&mut ctx, root, config)?;
667    Ok(ctx.assemble())
668}
669
670/// Walk + compress with producer/consumer overlap (TODO.perf/15).
671///
672/// The tree walk runs on a scoped producer thread and forwards each
673/// deferred file to a bounded channel; rayon workers (via `par_bridge`)
674/// compress while the walk is still descending. For warm-cache trees
675/// with few files this is equivalent to the collect-then-dispatch
676/// shape; for huge or cold-cache trees it hides walk latency behind
677/// compression.
678///
679/// **Determinism:** results are re-sequenced into walk order before
680/// merging, and inode allocation / dir-node construction are
681/// untouched, so the emitted bytes are identical to
682/// `write_directory_body` for the same input.
683fn write_directory_streaming(
684    ctx: &mut WriteContext,
685    root: &Path,
686    config: &WriteConfig,
687) -> Result<(), WriteError> {
688    use rayon::prelude::*;
689
690    ctx.metadata_codec = config
691        .metadata_codec_id()
692        .unwrap_or(limnifs_core::codec::CODEC_BROTLI);
693
694    ctx.chunker = chunker_from_config(config)?;
695
696    let chunker = ctx.chunker.clone();
697    let classifier = ctx.classifier;
698    let text_codec = config.text_codec_id().unwrap_or(0x04);
699    let binary_codec = config.binary_codec_id().unwrap_or(0x01);
700    let tunables = config.to_core_tunables();
701    let use_categorizers = !config.categorizers.is_empty();
702    let skip_chunking = config.skip_chunking;
703    let registry = config
704        .codec_registry()
705        .map_err(|e| WriteError::Io(std::io::Error::other(format!("codec registry: {e}"))))?;
706    let tournament_codec_ids: Vec<u8> = config
707        .tournament
708        .codecs
709        .iter()
710        .filter_map(|n| registry.lookup_by_name(n))
711        .collect();
712    let tournament_spec = TournamentSpec {
713        codec_ids: tournament_codec_ids,
714        min_size: config.tournament.min_size_threshold as usize,
715        skip_for_binary: config.tournament.skip_for_binary,
716        short_circuit_permille: config.tournament.short_circuit_threshold,
717    };
718    // The producer thread owns `&mut ctx` for the duration of the
719    // walk, so the layer fast-path index travels as a clone.
720    let base_drop_index = ctx.base_drop_index.clone();
721    let inline_threshold = ctx.inline_threshold;
722    let max_drop_size = config.defaults.max_drop_size as usize;
723    let seekable_drops = config.defaults.seekable_drops;
724
725    ctx.inline_threshold = config.defaults.inline_threshold as usize;
726    ctx.metadata_externalize_threshold = config.defaults.metadata_externalize_threshold;
727    ctx.emit_shared_inline = config.defaults.shared_inline;
728
729    // Bounded so the walk back-pressures if compression falls behind;
730    // the buffer is large enough to keep every worker fed on bursty
731    // directory layouts.
732    const PIPELINE_CAPACITY: usize = 256;
733    let (tx, rx) = std::sync::mpsc::sync_channel::<PendingFile>(PIPELINE_CAPACITY);
734    ctx.pending_sink = Some(tx);
735
736    let (root_inode_number, mut results): (
737        u64,
738        Vec<(usize, PendingFile, Result<ChunkedFileResult, WriteError>)>,
739    ) = {
740        // Survey on the calling thread BEFORE the pipeline starts.
741        // The parallel stat pass uses the global rayon pool; running
742        // it inside the producer would race the consumer's
743        // par_bridge for workers — par_bridge's fold_with blocks in
744        // recv() WHILE HOLDING the bridge mutex, so a pool-dependent
745        // producer can starve its own survey (all workers parked on
746        // the mutex) and deadlock. The survey must finish before the
747        // first fold anyway, so hoisting it costs nothing and leaves
748        // the producer thread pool-free.
749        let survey = survey_tree(root)?;
750        std::thread::scope(|scope| {
751            let producer = {
752                let ctx = &mut *ctx;
753                let root = root;
754                scope.spawn(move || {
755                    let r = ctx.fold_survey(root, &survey, None);
756                    // Disconnect the channel so the consumer's iterator
757                    // terminates; the sink stays None until the next
758                    // streaming write resets it after the scope.
759                    ctx.pending_sink = None;
760                    r
761                })
762            };
763            // par_bridge does not preserve order; carry the arrival index
764            // and re-sequence before merging.
765            let results = rx
766                .into_iter()
767                .enumerate()
768                .par_bridge()
769                .map(|(i, pf)| {
770                    let r = process_file(
771                        &pf,
772                        &chunker,
773                        classifier,
774                        text_codec,
775                        binary_codec,
776                        &tunables,
777                        use_categorizers,
778                        skip_chunking,
779                        &tournament_spec,
780                        base_drop_index.as_deref(),
781                        inline_threshold,
782                        max_drop_size,
783                        seekable_drops,
784                        config.categorizers.as_slice(),
785                        &|name| {
786                            config
787                                .codec_registry()
788                                .ok()
789                                .and_then(|r| r.lookup_by_name(name))
790                        },
791                    );
792                    (i, pf, r)
793                })
794                .collect();
795            let joined = producer
796                .join()
797                .unwrap_or_else(|_| {
798                    Err(WriteError::Io(std::io::Error::other(
799                        "walk thread panicked",
800                    )))
801                })
802                .map(|n| (n, results));
803            // Scope can't `?` across borrows of `results`; return the
804            // outcome and propagate outside.
805            joined
806        })
807    }?;
808    ctx.pending_sink = None;
809    ctx.root_inode_number = root_inode_number;
810
811    results.sort_unstable_by_key(|(i, _, _)| *i);
812    // Fail on the lowest walk index first, matching the
813    // collect::<Result<Vec<_>, _>> abort semantics of the
814    // collect-then-dispatch shape.
815    for (_, pf, r) in results {
816        ctx.merge_chunked_file(&pf, r?);
817    }
818    ctx.train_and_apply_dictionary(&config.dictionaries);
819    Ok(())
820}
821
822/// One chunk of a file before dedup: (`drop_id`, `plaintext`, `compressed`, `codec`).
823///
824/// `compressed` is `Arc<[u8]>` so the cross-file compress cache can
825/// share bytes across hits with a refcount bump instead of a deep
826/// copy — dedup-heavy workloads (container layers, duplicate files)
827/// skip the allocation entirely.
828pub(crate) type RawDrop = ([u8; 32], Vec<u8>, std::sync::Arc<[u8]>, u8, u8);
829/// Result of parallel file processing: the drop data (uncompressed,
830pub(crate) struct ChunkedFileResult {
831    drops: Vec<RawDrop>, // (id, plaintext, compressed, codec, flags)
832    slices: Vec<PendingSlice>,
833}
834
835/// Resolved tournament configuration passed to per-chunk compression.
836///
837/// Built once at the top level from `WriteConfig::tournament` so each
838/// rayon worker reuses the same Vec rather than rebuilding it per
839/// file. Codecs are stored as numeric ids (looked up via
840/// `WriteConfig::codec_registry`) — `process_file` never sees the
841/// string form.
842struct TournamentSpec {
843    /// Codec ids to try, in declared order. `process_file` iterates
844    /// these and tracks the best compression. `CODEC_STORE` entries
845    /// are ignored (store is always the implicit fallback).
846    codec_ids: Vec<u8>,
847    /// Chunks below this many bytes get the preferred codec only
848    /// (no tournament) — the per-codec setup cost dominates at small
849    /// sizes and the ratio difference is negligible.
850    min_size: usize,
851    /// When true, binary-classified chunks skip the tournament and
852    /// use `binary_codec` directly. Matches the v0.1 behaviour where
853    /// binary chunks were never worth the tournament cost.
854    skip_for_binary: bool,
855    /// Short-circuit threshold in per-mille (0..=1000). 0 disables
856    /// short-circuit. Whenever a codec achieves compression ratio
857    /// ≤ threshold, the tournament accepts it and skips any slower
858    /// codecs later in the list.
859    short_circuit_permille: u32,
860}
861
862/// Compress a whole file as a single drop using the categorizer's
863/// chosen codec. Used when a file-level categorizer claims the file
864/// (FLAC for WAV, ricepp for FITS, FSST+Brotli for CSV). The drop's
865/// slice covers the whole file; no `FastCDC` chunking happens.
866///
867/// Codec parameters extracted by the categorizer (e.g. PCM sample
868/// format, FITS bitpix) are NOT prepended to the compressed bytes —
869/// the codec embeds its own params in its container format. The
870/// `LimniFS` drop record just stores `(codec_id, compressed_bytes)`
871/// and lets the codec own its param encoding. The categorizer's
872/// `codec_params` field is reserved for future use when a codec
873/// needs params NOT embedded in its container.
874/// Build the chunker from the config's `[chunking]` section. The
875/// section was previously parsed and validated but never applied —
876/// `WriteContext` hardcoded `FastCDC::default()`. Defaults in
877/// `default_v0_1` match the previous effective values, so default
878/// images are byte-identical; only configs that set `[chunking]`
879/// change output (which is the point of setting it).
880fn chunker_from_config(config: &WriteConfig) -> Result<ParallelFastCDC, WriteError> {
881    ParallelFastCDC::new(
882        config.chunking.min_chunk_size as usize,
883        config.chunking.avg_chunk_size as usize,
884        config.chunking.max_chunk_size as usize,
885    )
886    .map_err(|e| WriteError::Io(std::io::Error::other(format!("chunking config: {e}"))))
887}
888
889/// Encode `plaintext` as a seekable container when the drop is large
890/// enough to hurt cold random reads and the codec supports
891/// independent frames (TODO.sota-fs/05). Ratio cost is bounded by
892/// per-frame independence; the reader gains 256 KiB-bounded windowed
893/// decode. Encoder failure degrades to the monolithic stream — a
894/// container problem must never fail the write.
895pub(crate) fn seekable_or_monolithic(
896    codec: u8,
897    plaintext: &[u8],
898    compressed: std::sync::Arc<[u8]>,
899    tunables: &limnifs_core::codec::CodecTunables,
900    seekable_drops: bool,
901    threshold: usize,
902) -> (std::sync::Arc<[u8]>, u8) {
903    use limnifs_core::seekable::{
904        encode_seekable, is_seekable_codec, DROP_FLAG_SEEKABLE as FLAG, SEEKABLE_EMISSION_THRESHOLD,
905    };
906    if seekable_drops && plaintext.len() > threshold && is_seekable_codec(codec) {
907        if let Ok(container) = encode_seekable(codec, plaintext, tunables) {
908            return (container.into(), FLAG);
909        }
910    }
911    (compressed, 0)
912}
913
914/// Per-chunk emission threshold. FastCDC chunks are bounded by
915/// `max_chunk_size` (default 1 MiB) so the whole-file 1 MiB
916/// threshold can never fire per chunk — limnifs#195. Chunk drops as
917/// small as one frame (256 KiB) still gain the covering-frames
918/// decode bound: an 8 KiB window inside a 256 KiB drop decodes one
919/// 8 KiB-ish frame instead of the whole drop.
920pub(crate) const SEEKABLE_CHUNK_EMISSION_THRESHOLD: usize =
921    limnifs_core::seekable::SEEKABLE_FRAME_SIZE;
922
923fn process_whole_file_drop(
924    pf: &PendingFile,
925    data: &[u8],
926    cat: file_categorizer::Categorization,
927    tunables: &limnifs_core::codec::CodecTunables,
928    seekable_drops: bool,
929) -> Result<ChunkedFileResult, WriteError> {
930    let _ = pf;
931    let drop_id = hash_section(data);
932    let file_len = u64::try_from(data.len()).unwrap_or(u64::MAX);
933
934    // Brotli first; if it fails (including a codec panic — the
935    // registry converts panics to Err), fall back to ZSTD, then to
936    // STORE. A broken encoder must degrade the drop's ratio, never
937    // the write itself.
938    let (mut best_codec, mut best_compressed): (u8, std::sync::Arc<[u8]>) =
939        match limnifs_core::codec::compress_with_tunables(
940            limnifs_core::codec::CODEC_BROTLI,
941            data,
942            tunables,
943        ) {
944            Ok(c) => (limnifs_core::codec::CODEC_BROTLI, c.into()),
945            Err(_) => match limnifs_core::codec::compress_with_tunables(
946                limnifs_core::codec::CODEC_ZSTD,
947                data,
948                tunables,
949            ) {
950                Ok(c) => (limnifs_core::codec::CODEC_ZSTD, c.into()),
951                Err(_) => (limnifs_core::codec::CODEC_STORE, data.to_vec().into()),
952            },
953        };
954
955    // Short-circuit: if Brotli already achieves < 5% ratio, the input
956    // is highly compressible and ZSTD is unlikely to beat it by enough
957    // to justify the extra pass. Skip ZSTD on this fast path.
958    let brotli_ratio = best_compressed.len() as f64 / data.len() as f64;
959    if brotli_ratio > 0.05 {
960        if let Ok(zstd_c) = limnifs_core::codec::compress_with_tunables(
961            limnifs_core::codec::CODEC_ZSTD,
962            data,
963            tunables,
964        ) {
965            if zstd_c.len() < best_compressed.len() {
966                best_codec = limnifs_core::codec::CODEC_ZSTD;
967                best_compressed = zstd_c.into();
968            }
969        }
970    }
971
972    // Only try the specialized codec if the general-purpose ratio
973    // is poor (>15%) — otherwise the specialized codec is unlikely
974    // to help and may be very slow (FLAC, FSST). RICEPP is always
975    // tried because it can win big on FITS even when general-purpose
976    // ratios look acceptable.
977    let general_ratio = best_compressed.len() as f64 / data.len() as f64;
978    if general_ratio > 0.15 || cat.codec_id == limnifs_core::codec::CODEC_RICEPP {
979        // For FSST+Brotli, pass the already-computed Brotli baseline so
980        // the codec doesn't re-compress the plaintext with Brotli just
981        // for the comparison check.
982        let spec_result = if cat.codec_id == limnifs_core::codec::CODEC_FSST_BROTLI {
983            limnifs_core::codec::fsst_brotli::compress_with_baseline(data, Some(&best_compressed))
984        } else {
985            // Tunables-routed so future codec knobs (FLAC/RicePP have
986            // none today — all parameters come from the file header)
987            // are honored here automatically (TODO.remaining item 2).
988            limnifs_core::codec::compress_with_tunables(cat.codec_id, data, tunables)
989        };
990        if let Ok(spec_c) = spec_result {
991            if spec_c.len() < best_compressed.len() {
992                best_codec = cat.codec_id;
993                best_compressed = spec_c.into();
994            }
995        }
996    }
997
998    let (best_compressed, flags) = seekable_or_monolithic(
999        best_codec,
1000        data,
1001        best_compressed,
1002        tunables,
1003        seekable_drops,
1004        limnifs_core::seekable::SEEKABLE_EMISSION_THRESHOLD,
1005    );
1006    Ok(ChunkedFileResult {
1007        drops: vec![(drop_id, data.to_vec(), best_compressed, best_codec, flags)],
1008        slices: vec![PendingSlice {
1009            drop_id,
1010            file_byte_start: 0,
1011            file_byte_end: file_len,
1012        }],
1013    })
1014}
1015
1016/// Process a single file's contents (CPU-heavy work that runs in a
1017/// rayon worker thread). Returns the unique chunks and slice map.
1018///
1019/// First consults the file-level categorizer registry. If a
1020/// categorizer claims the file (e.g. FLAC for WAV, ricepp for FITS,
1021/// Compress a single chunk via the configured tournament.
1022///
1023/// Iterates `tournament.codec_ids` in declared order, tracks the
1024/// smallest output, and short-circuits when a codec achieves
1025/// compression ratio ≤ `tournament.short_circuit_permille`.
1026///
1027/// Special cases:
1028/// - **Binary chunks with `skip_for_binary`**: skip the tournament
1029///   entirely and use `binary_codec`. Matches v0.1 behaviour.
1030/// - **Chunks smaller than `min_size`**: use the class's preferred
1031///   codec directly. Per-codec setup cost dominates here and the
1032///   ratio difference is negligible at small sizes.
1033/// - **Class unknown to writer** (Unknown / future classes): STORE.
1034///
1035/// The tournament never tries `CODEC_STORE` (id 0x00) — store is
1036/// always the implicit fallback if every codec fails to compress.
1037fn compress_chunk_with_tournament(
1038    chunk: &[u8],
1039    class: classifier::Class,
1040    text_codec: u8,
1041    binary_codec: u8,
1042    tunables: &limnifs_core::codec::CodecTunables,
1043    tournament: &TournamentSpec,
1044) -> (u8, std::sync::Arc<[u8]>) {
1045    use classifier::Class;
1046
1047    let preferred = match class {
1048        Class::Binary => binary_codec,
1049        Class::Text | Class::Code | Class::Sparse => text_codec,
1050        _ => limnifs_core::codec::CODEC_STORE,
1051    };
1052
1053    if preferred == limnifs_core::codec::CODEC_STORE {
1054        return (limnifs_core::codec::CODEC_STORE, chunk.to_vec().into());
1055    }
1056    if class == Class::Binary && tournament.skip_for_binary {
1057        return compress_chunk_one(chunk, preferred, tunables);
1058    }
1059    if chunk.len() < tournament.min_size {
1060        return compress_chunk_one(chunk, preferred, tunables);
1061    }
1062
1063    let mut best: Option<(u8, std::sync::Arc<[u8]>)> = None;
1064    for &codec_id in &tournament.codec_ids {
1065        if codec_id == limnifs_core::codec::CODEC_STORE {
1066            continue;
1067        }
1068        let c = match limnifs_core::codec::compress_with_tunables(codec_id, chunk, tunables) {
1069            Ok(c) => c,
1070            Err(_) => continue,
1071        };
1072        if c.len() >= chunk.len() {
1073            continue;
1074        }
1075        let ratio_permille = (c.len() as u64 * 1000 / chunk.len() as u64) as u32;
1076        let is_best_so_far = best.as_ref().map_or(true, |(_, b)| c.len() < b.len());
1077        if is_best_so_far {
1078            best = Some((codec_id, c.into()));
1079        }
1080        if tournament.short_circuit_permille > 0
1081            && ratio_permille <= tournament.short_circuit_permille
1082        {
1083            break;
1084        }
1085    }
1086
1087    best.unwrap_or_else(|| (limnifs_core::codec::CODEC_STORE, chunk.to_vec().into()))
1088}
1089
1090/// Compress `chunk` with a single codec, falling back to STORE if
1091/// the codec fails or expansion occurs.
1092fn compress_chunk_one(
1093    chunk: &[u8],
1094    codec_id: u8,
1095    tunables: &limnifs_core::codec::CodecTunables,
1096) -> (u8, std::sync::Arc<[u8]>) {
1097    if codec_id == limnifs_core::codec::CODEC_STORE {
1098        return (limnifs_core::codec::CODEC_STORE, chunk.to_vec().into());
1099    }
1100    match limnifs_core::codec::compress_with_tunables(codec_id, chunk, tunables) {
1101        Ok(c) if c.len() < chunk.len() => (codec_id, c.into()),
1102        _ => (limnifs_core::codec::CODEC_STORE, chunk.to_vec().into()),
1103    }
1104}
1105
1106/// FSST+Brotli for CSV), the whole file is compressed as a single
1107/// drop with the categorizer's chosen codec + parameters. Otherwise
1108/// falls through to `FastCDC` + per-chunk classify.
1109fn process_file(
1110    pf: &PendingFile,
1111    chunker: &dyn Chunker,
1112    classifier: classifier::Classifier,
1113    text_codec: u8,
1114    binary_codec: u8,
1115    tunables: &limnifs_core::codec::CodecTunables,
1116    use_categorizers: bool,
1117    skip_chunking: bool,
1118    tournament: &TournamentSpec,
1119    base_drop_index: Option<&dyn BaseDropSet>,
1120    inline_threshold: usize,
1121    max_drop_size: usize,
1122    seekable_drops: bool,
1123    categorizer_config: &[crate::config::CategorizerConfig],
1124    codec_name_resolver: &dyn Fn(&str) -> Option<u8>,
1125) -> Result<ChunkedFileResult, WriteError> {
1126    // For files above MMAP_READ_THRESHOLD, map them rather than reading
1127    // into a Vec via std::fs::read. Pages load on demand from the
1128    // kernel page cache. The chunk-path compressors see borrowed slices
1129    // pointing into the mmap, so peak RSS stays at unique_chunks ×
1130    // avg_chunk_size rather than the full file.
1131    //
1132    // SAFETY: LimniFS packs source trees that are immutable for the
1133    // duration of the write. The file is opened read-only. External
1134    // mutation during compression would be a serious bug in the
1135    // caller's workflow (and would also break BLAKE3 determinism).
1136    let file_len_estimate = std::fs::metadata(&pf.path)
1137        .map(|m| m.len() as usize)
1138        .unwrap_or(0);
1139    // Borrow the mmap instead of materializing it: the previous
1140    // `Vec::from(&mmap[..])` copied every byte of every file ≥ 1 MiB,
1141    // defeating the mapping entirely (one full memcpy plus peak RSS
1142    // equal to `fs::read`). Chunking, hashing, and compression all
1143    // work on borrowed slices; the only owned copies are the per-drop
1144    // plaintexts the context retains.
1145    let mmap_handle: memmap2::Mmap;
1146    let small: Vec<u8>;
1147    let data: &[u8] = if file_len_estimate >= MMAP_READ_THRESHOLD {
1148        let file = std::fs::File::open(&pf.path)?;
1149        // SAFETY: LimniFS packs source trees that are immutable for
1150        // the duration of the write; the file is opened read-only and
1151        // the mapping is only read.
1152        #[allow(unsafe_code)]
1153        let mapped = unsafe { memmap2::Mmap::map(&file) }.map_err(WriteError::Io)?;
1154        mmap_handle = mapped;
1155        &mmap_handle[..]
1156    } else {
1157        small = std::fs::read(&pf.path)?;
1158        &small[..]
1159    };
1160    let file_len = data.len();
1161
1162    // Skip FastCDC chunking entirely; compress the whole file as
1163    // one drop. Trades dedup granularity for create speed. Used by
1164    // the max-write profile where speed >> ratio. The per-file LZ4
1165    // compress at ~1 GB/s is faster than FastCDC hashing overhead
1166    // for all but the largest multi-GB files (where rayon parallelism
1167    // across chunks would help).
1168    if skip_chunking && file_len > inline_threshold {
1169        let drop_id = hash_section(&data);
1170        let class = classifier.classify(&data);
1171        let preferred_codec = match class {
1172            classifier::Class::Binary => binary_codec,
1173            _ => text_codec,
1174        };
1175        let (codec_id, compressed): (u8, std::sync::Arc<[u8]>) =
1176            match limnifs_core::codec::compress_with_tunables(preferred_codec, &data, tunables) {
1177                Ok(c) if c.len() < data.len() => (preferred_codec, c.into()),
1178                _ => (limnifs_core::codec::CODEC_STORE, data.to_vec().into()),
1179            };
1180        let (compressed, flags) = seekable_or_monolithic(
1181            codec_id,
1182            &data,
1183            compressed,
1184            tunables,
1185            seekable_drops,
1186            SEEKABLE_CHUNK_EMISSION_THRESHOLD,
1187        );
1188        return Ok(ChunkedFileResult {
1189            drops: vec![(drop_id, data.to_vec(), compressed, codec_id, flags)],
1190            slices: vec![PendingSlice {
1191                drop_id,
1192                file_byte_start: 0,
1193                file_byte_end: file_len as u64,
1194            }],
1195        });
1196    }
1197
1198    if use_categorizers {
1199        // Config entries FIRST (limnifs#196): the user's
1200        // `[[categorizers]]` rules override the built-ins.
1201        let config_cat = file_categorizer::ConfigCategorizer::new(categorizer_config.to_vec());
1202        if let Some(cat) = config_cat.categorize(&pf.path, &data) {
1203            if let Some(codec_id) = file_categorizer::resolve_config_categorization(
1204                &cat,
1205                categorizer_config,
1206                codec_name_resolver,
1207            ) {
1208                let within_cap = max_drop_size == 0 || file_len <= max_drop_size;
1209                let needs_whole_file = matches!(
1210                    codec_id,
1211                    limnifs_core::codec::CODEC_FLAC | limnifs_core::codec::CODEC_RICEPP
1212                );
1213                if within_cap && (needs_whole_file || file_len <= WHOLE_FILE_MAX_SIZE) {
1214                    let mut cat = cat;
1215                    cat.codec_id = codec_id;
1216                    return process_whole_file_drop(pf, &data, cat, tunables, seekable_drops);
1217                }
1218            }
1219        }
1220        if let Some(cat) = file_categorizer::default_registry().categorize(&pf.path, &data) {
1221            let needs_whole_file = matches!(
1222                cat.codec_id,
1223                limnifs_core::codec::CODEC_FLAC | limnifs_core::codec::CODEC_RICEPP
1224            );
1225            // max_drop_size bounds the decompressed unit: files over
1226            // the cap fall through to chunking + tournament (0 = off).
1227            let within_cap = max_drop_size == 0 || file_len <= max_drop_size;
1228            if within_cap && (needs_whole_file || file_len <= WHOLE_FILE_MAX_SIZE) {
1229                return process_whole_file_drop(pf, &data, cat, tunables, seekable_drops);
1230            }
1231        }
1232    }
1233
1234    let chunks = chunker.chunk_slice(&data);
1235
1236    // Phase 1: hash all chunks in parallel, then build slices + filter
1237    // duplicates sequentially. Each chunk's BLAKE3 is independent, so
1238    // the map parallelizes across rayon workers (work stealing — Phase
1239    // 2 nests par_iter inside a worker the same way). This closes the
1240    // sequential hashing tail that left N-1 cores idle on
1241    // single-large-file workloads. Drop ids are per-chunk values and
1242    // the collect preserves chunk order, so output is byte-identical
1243    // to the sequential loop.
1244    use rayon::prelude::*;
1245    let drop_ids: Vec<[u8; 32]> = chunks.par_iter().map(|chunk| hash_section(chunk)).collect();
1246
1247    let mut slices = Vec::with_capacity(chunks.len());
1248    let mut file_offset: u64 = 0;
1249    let mut seen_in_file: std::collections::HashSet<[u8; 32]> =
1250        std::collections::HashSet::with_capacity(chunks.len());
1251    let mut unique_chunks: Vec<(&[u8], [u8; 32])> = Vec::with_capacity(chunks.len());
1252    for (chunk, drop_id) in chunks.iter().zip(drop_ids) {
1253        let chunk_len = u64::try_from(chunk.len()).expect("chunk len fits u64");
1254        slices.push(PendingSlice {
1255            drop_id,
1256            file_byte_start: file_offset,
1257            file_byte_end: file_offset + chunk_len,
1258        });
1259        file_offset += chunk_len;
1260        if seen_in_file.insert(drop_id) {
1261            unique_chunks.push((chunk, drop_id));
1262        }
1263    }
1264
1265    // Phase 2: compress unique chunks in parallel across rayon workers.
1266    // This is the CPU-intensive step — parallelizing it gives N-core
1267    // speedup for large files with many chunks.
1268    //
1269    // Cross-file dedup: each rayon worker thread carries a thread-local
1270    // compress cache mapping DropId -> (codec_id, compressed_bytes).
1271    // When two files share a chunk (common in source trees, container
1272    // layers, tiny-files benchmarks), the second file hits the cache
1273    // and skips the compress pass entirely. Cache is bounded by entry
1274    // count; eviction is "stop inserting once full" — simple and
1275    // correct, misses are bounded by worker count.
1276    thread_local! {
1277        static COMPRESS_CACHE: std::cell::RefCell<std::collections::HashMap<[u8; 32], (u8, std::sync::Arc<[u8]>)>> =
1278            std::cell::RefCell::new(std::collections::HashMap::new());
1279    }
1280    const COMPRESS_CACHE_MAX_ENTRIES: usize = 100_000;
1281    let drops: Vec<RawDrop> = unique_chunks
1282        .par_iter()
1283        .map(|(chunk, drop_id)| {
1284            // Layer fast-path: chunk already exists in the base image
1285            // → skip compress entirely.
1286            if let Some(base) = base_drop_index {
1287                if base.base_contains(drop_id) {
1288                    return (*drop_id, Vec::new(), Vec::new().into(), CODEC_REFERENCED, 0);
1289                }
1290            }
1291            let class = classifier.classify(chunk);
1292            // Cache fast-path: identical chunk already compressed on
1293            // this worker → reuse bytes, skip tournament entirely.
1294            let cached = COMPRESS_CACHE.with(|c| {
1295                c.borrow()
1296                    .get(drop_id)
1297                    .map(|(cid, comp)| (*cid, comp.clone()))
1298            });
1299            let (codec_id, compressed) = if let Some(c) = cached {
1300                c
1301            } else {
1302                let new = compress_chunk_with_tournament(
1303                    chunk,
1304                    class,
1305                    text_codec,
1306                    binary_codec,
1307                    tunables,
1308                    tournament,
1309                );
1310                // Insert into the per-worker cache if there's room.
1311                COMPRESS_CACHE.with(|c| {
1312                    let mut cache = c.borrow_mut();
1313                    if cache.len() < COMPRESS_CACHE_MAX_ENTRIES {
1314                        // `new.1.clone()` here is an Arc refcount bump.
1315                        cache.insert(*drop_id, new.clone());
1316                    }
1317                });
1318                new
1319            };
1320            // limnifs#195: chunk drops go through the same
1321            // seekable-container emission as whole-file drops, with a
1322            // chunk-appropriate threshold (chunks are bounded by
1323            // max_chunk_size, so the whole-file 1 MiB gate can never
1324            // fire here).
1325            let (compressed, flags) = seekable_or_monolithic(
1326                codec_id,
1327                chunk,
1328                compressed,
1329                tunables,
1330                seekable_drops,
1331                SEEKABLE_CHUNK_EMISSION_THRESHOLD,
1332            );
1333            (*drop_id, chunk.to_vec(), compressed, codec_id, flags)
1334        })
1335        .collect();
1336
1337    let _ = file_len;
1338    Ok(ChunkedFileResult { drops, slices })
1339}
1340
1341struct PendingDrop {
1342    id: [u8; 32],
1343    /// Original (decompressed) byte length. Stored as a u32 rather
1344    /// than keeping the plaintext Vec around, because the writer only
1345    /// needs the length when emitting the slab's drop record. Holding
1346    /// the full plaintext until slab assembly wastes memory
1347    /// proportional to image size on top of the compressed bytes.
1348    plaintext_len: u32,
1349    compressed: std::sync::Arc<[u8]>,
1350    codec: u8,
1351    /// Dictionary id (0xFF = NO_DICT). Populated during the dict
1352    /// re-compression pass for drops that were re-compressed with a
1353    /// trained dictionary.
1354    dict_id: u8,
1355    /// Retained plaintext, present only when `collect_dict_samples`
1356    /// is true (i.e. `WriteConfig::dictionaries.enabled`). Used by
1357    /// the post-parallel dict re-compression pass. Cleared after
1358    /// re-compression to free memory before slab assembly.
1359    plaintext: Option<Vec<u8>>,
1360    /// Record flags. Bit0 = SEEKABLE (window bytes are a
1361    /// `limnifs_core::seekable` container); 0 = plain monolithic
1362    /// codec stream.
1363    flags: u8,
1364}
1365
1366impl PendingDrop {
1367    /// The byte length stored in the slab's solid window. Equals
1368    /// `plaintext_len` for store codec, or the compressed size for
1369    /// LZ4 / Brotli / etc.
1370    fn len_in_window(&self) -> u32 {
1371        u32::try_from(self.compressed.len()).expect("compressed fits u32")
1372    }
1373
1374    /// The original (decompressed) byte length.
1375    fn plaintext_len_value(&self) -> u32 {
1376        self.plaintext_len
1377    }
1378
1379    /// Contribution to the slab's total byte length: 48 bytes of drop
1380    /// record (per spec §3.3) + the compressed payload.
1381    fn slab_footprint(&self) -> usize {
1382        48 + self.compressed.len()
1383    }
1384}
1385
1386/// One slice of a file backed by drops. Records which drop holds
1387/// this slice's bytes and which byte range of the original file
1388/// the slice covers. The slice always spans the entire drop (the
1389/// chunker never splits a drop across multiple slices).
1390struct PendingSlice {
1391    drop_id: [u8; 32],
1392    file_byte_start: u64,
1393    file_byte_end: u64,
1394}
1395
1396/// A file that needs chunking (> `INLINE_THRESHOLD`). Collected during
1397/// the sequential tree walk and processed in parallel by `rayon`.
1398#[derive(Clone)]
1399struct PendingFile {
1400    inode_number: u64,
1401    path: PathBuf,
1402    mtime_ns: u64,
1403    file_len: u64,
1404    mode: u32,
1405    uid: u32,
1406    gid: u32,
1407}
1408
1409/// Stat snapshot captured during the parallel survey phase of the
1410/// walk. Encapsulates everything the sequential fold needs from the
1411/// filesystem — the fold performs no stat syscalls of its own.
1412#[derive(Clone, Default)]
1413struct SurveyMeta {
1414    is_dir: bool,
1415    is_file: bool,
1416    is_symlink: bool,
1417    #[cfg(unix)]
1418    is_fifo: bool,
1419    #[cfg(unix)]
1420    is_socket: bool,
1421    #[cfg(unix)]
1422    is_block_device: bool,
1423    #[cfg(unix)]
1424    is_char_device: bool,
1425    len: u64,
1426    mtime_ns: u64,
1427    #[cfg(unix)]
1428    mode: u32,
1429    #[cfg(unix)]
1430    uid: u32,
1431    #[cfg(unix)]
1432    gid: u32,
1433    /// Device + inode identity for hardlink detection (unix). A
1434    /// second occurrence of the same (dev, ino) becomes a metadata
1435    /// reference to the first inode instead of new content.
1436    #[cfg(unix)]
1437    dev: u64,
1438    #[cfg(unix)]
1439    ino: u64,
1440    /// Extended attributes (feature `xattr`, unix only), name-sorted
1441    /// and capped; volatile platform namespaces filtered.
1442    #[cfg(feature = "xattr")]
1443    xattrs: Vec<(String, Vec<u8>)>,
1444}
1445
1446impl SurveyMeta {
1447    /// (mode, uid, gid) for inode emission. Real st_mode and
1448    /// ownership on unix; the historical fixed defaults elsewhere,
1449    /// so non-unix images are byte-identical to before.
1450    fn identity(&self) -> (u32, u32, u32) {
1451        #[cfg(unix)]
1452        {
1453            (self.mode, self.uid, self.gid)
1454        }
1455        #[cfg(not(unix))]
1456        {
1457            let ty = if self.is_dir {
1458                limnifs_core::inode::S_IFDIR
1459            } else if self.is_symlink {
1460                limnifs_core::inode::S_IFLNK
1461            } else {
1462                limnifs_core::inode::S_IFREG
1463            };
1464            let perms = if self.is_dir || self.is_symlink {
1465                0o755
1466            } else {
1467                0o644
1468            };
1469            (ty | perms, 0, 0)
1470        }
1471    }
1472}
1473
1474/// One surveyed tree node. Children are name-sorted so the
1475/// sequential fold reproduces the classic DFS order exactly.
1476struct SurveyNode {
1477    meta: SurveyMeta,
1478    children: Vec<(String, SurveyNode)>,
1479    /// `Some(target)` when this node is a symlink (read during the
1480    /// survey; non-UTF-8 targets are surfaced as survey errors,
1481    /// matching the walk's previous error).
1482    symlink_target: Option<String>,
1483}
1484
1485impl SurveyNode {
1486    fn meta(&self) -> &SurveyMeta {
1487        &self.meta
1488    }
1489}
1490
1491fn survey_meta_of(meta: &std::fs::Metadata) -> SurveyMeta {
1492    #[cfg(unix)]
1493    use std::os::unix::fs::FileTypeExt as _;
1494    let ft = meta.file_type();
1495    let mtime_ns = meta
1496        .modified()
1497        .ok()
1498        .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
1499        .map_or(0u128, |d| d.as_nanos());
1500    SurveyMeta {
1501        is_dir: ft.is_dir(),
1502        is_file: ft.is_file(),
1503        is_symlink: ft.is_symlink(),
1504        #[cfg(unix)]
1505        is_fifo: ft.is_fifo(),
1506        #[cfg(unix)]
1507        is_socket: ft.is_socket(),
1508        #[cfg(unix)]
1509        is_block_device: ft.is_block_device(),
1510        #[cfg(unix)]
1511        is_char_device: ft.is_char_device(),
1512        len: meta.len(),
1513        mtime_ns: mtime_ns.try_into().unwrap_or(0),
1514        #[cfg(unix)]
1515        mode: {
1516            use std::os::unix::fs::MetadataExt as _;
1517            meta.mode()
1518        },
1519        #[cfg(unix)]
1520        uid: {
1521            use std::os::unix::fs::MetadataExt as _;
1522            meta.uid()
1523        },
1524        #[cfg(unix)]
1525        gid: {
1526            use std::os::unix::fs::MetadataExt as _;
1527            meta.gid()
1528        },
1529        #[cfg(unix)]
1530        dev: {
1531            use std::os::unix::fs::MetadataExt as _;
1532            meta.dev()
1533        },
1534        #[cfg(unix)]
1535        ino: {
1536            use std::os::unix::fs::MetadataExt as _;
1537            meta.ino()
1538        },
1539        #[cfg(feature = "xattr")]
1540        xattrs: Vec::new(),
1541    }
1542}
1543
1544/// Best-effort xattr capture (feature `xattr`, unix): name-sorted,
1545/// capped per inode, with volatile platform namespaces filtered so
1546/// images stay deterministic and free of OS churn.
1547#[cfg(all(unix, feature = "xattr"))]
1548fn collect_xattrs(path: &Path) -> Vec<(String, Vec<u8>)> {
1549    const VOLATILE: &[&str] = &[
1550        "com.apple.provenance",
1551        "com.apple.quarantine",
1552        "com.apple.lastuseddate",
1553        "com.apple.macl",
1554        "com.apple.filesec",
1555    ];
1556    const TOTAL_CAP: usize = 64 * 1024;
1557    let Ok(names) = xattr::list(path) else {
1558        return Vec::new();
1559    };
1560    let mut names: Vec<String> = names
1561        .filter_map(|n| n.into_string().ok())
1562        .filter(|n| {
1563            !n.starts_with("system.")
1564                && !n.starts_with("security.")
1565                && !n.starts_with("trusted.")
1566                && !VOLATILE.contains(&n.as_str())
1567        })
1568        .collect();
1569    names.sort();
1570    let mut out = Vec::new();
1571    let mut total = 0usize;
1572    for name in names {
1573        let Ok(Some(value)) = xattr::get(path, &name) else {
1574            continue;
1575        };
1576        total += name.len() + value.len();
1577        if total > TOTAL_CAP {
1578            break;
1579        }
1580        out.push((name, value));
1581    }
1582    out
1583}
1584
1585fn survey_node(path: &Path) -> Result<SurveyNode, WriteError> {
1586    use rayon::prelude::*;
1587    let meta = std::fs::symlink_metadata(path)?;
1588    let mut sm = survey_meta_of(&meta);
1589    #[cfg(all(unix, feature = "xattr"))]
1590    {
1591        if !sm.is_symlink {
1592            sm.xattrs = collect_xattrs(path);
1593        }
1594    }
1595    if sm.is_symlink {
1596        let target = std::fs::read_link(path)?;
1597        let target = target
1598            .to_str()
1599            .ok_or_else(|| WriteError::UnsupportedFileType {
1600                path: path.to_path_buf(),
1601                kind: format!("symlink with non-UTF-8 target ({})", target.display()),
1602            })?
1603            .to_owned();
1604        return Ok(SurveyNode {
1605            meta: sm,
1606            children: Vec::new(),
1607            symlink_target: Some(target),
1608        });
1609    }
1610    if !sm.is_dir {
1611        return Ok(SurveyNode {
1612            meta: sm,
1613            children: Vec::new(),
1614            symlink_target: None,
1615        });
1616    }
1617    let mut named: Vec<(String, PathBuf)> = std::fs::read_dir(path)?
1618        .filter_map(|entry| {
1619            entry
1620                .ok()
1621                .map(|e| (e.file_name().to_string_lossy().into_owned(), e.path()))
1622        })
1623        .collect();
1624    named.sort_by(|a, b| a.0.cmp(&b.0));
1625    named
1626        .par_iter()
1627        .map(|(name, child)| survey_node(child).map(|node| (name.clone(), node)))
1628        .collect::<Result<Vec<_>, WriteError>>()
1629        .map(|children| SurveyNode {
1630            meta: sm,
1631            children,
1632            symlink_target: None,
1633        })
1634}
1635
1636/// Parallel tree survey: stats every node under `root` across
1637/// rayon workers. The returned structure fully determines the
1638/// fold's output — the fold itself never touches the filesystem
1639/// except to read file payloads.
1640fn survey_tree(root: &Path) -> Result<SurveyNode, WriteError> {
1641    survey_node(root)
1642}
1643
1644struct PendingInode {
1645    number: u64,
1646    mode: u32,
1647    uid: u32,
1648    gid: u32,
1649    mtime_ns: u64,
1650    xattrs: Vec<limnifs_core::inode::XAttr>,
1651    content: PendingContent,
1652}
1653
1654enum PendingContent {
1655    Inline(Vec<u8>),
1656    /// Symlink target (raw, as read from the filesystem).
1657    Symlink(String),
1658    DropBacked {
1659        file_len: u64,
1660        slices: Vec<PendingSlice>,
1661    },
1662    Directory(Vec<(String, u64, u8)>),
1663}
1664
1665struct DirNode {
1666    entries: Vec<(String, u64, u8)>,
1667    bytes: Vec<u8>,
1668    hash: [u8; 32],
1669}
1670
1671struct WriteContext {
1672    next_inode: u64,
1673    /// (dev, ino) -> inode number of the first occurrence (unix
1674    /// hardlink detection during the fold).
1675    hardlink_targets: std::collections::HashMap<(u64, u64), u64>,
1676    /// inode number -> reference count. `encode_inode` writes the
1677    /// real nlink from here — the single source of truth shared by
1678    /// the walk, stream, and staged paths.
1679    nlink_counts: std::collections::HashMap<u64, u32>,
1680    /// inode number -> xattrs captured by the walk. The encoder
1681    /// prefers this over `PendingInode.xattrs` (the struct field is
1682    /// the direct/stream path's truth) — same map-wins shape as
1683    /// `nlink_counts`.
1684    inode_xattrs: std::collections::HashMap<u64, Vec<limnifs_core::inode::XAttr>>,
1685    inodes: Vec<PendingInode>,
1686    dir_nodes: Vec<DirNode>,
1687    drops: Vec<PendingDrop>,
1688    drop_index: HashSet<[u8; 32]>,
1689    pending_files: Vec<PendingFile>,
1690    file_count: usize,
1691    dir_count: usize,
1692    root_inode_number: u64,
1693    chunker: ParallelFastCDC,
1694    classifier: classifier::Classifier,
1695    shared_inline_map: HashMap<[u8; 32], usize>,
1696    shared_inline_table: Vec<Vec<u8>>,
1697    /// Dictionaries adopted from the base image during a layer
1698    /// write. When present (and dictionaries are enabled), they
1699    /// displace training: the layer compresses with dictionaries
1700    /// the base already paid for and re-emits the section, so the
1701    /// layer image stays self-describing.
1702    base_dictionaries: Option<Vec<crate::dictionary::TrainedDictionary>>,
1703    /// Profile name for ProfileDescriptor emission (None = omit section).
1704    profile_name: Option<String>,
1705    /// Metadata blob codec (defaults to Brotli; can be overridden via
1706    /// `WriteConfig::defaults::metadata_codec`). Used by `assemble`.
1707    metadata_codec: u8,
1708    /// Whether categorizers were disabled by the profile.
1709    categorizers_disabled: bool,
1710    /// Whether this is a RW image.
1711    rw_mode: bool,
1712    /// Whether auto-turnover is enabled.
1713    auto_turnover: bool,
1714    /// Whether to collect plaintext samples for ZSTD dictionary
1715    /// training. Set when `WriteConfig::dictionaries.enabled`.
1716    collect_dict_samples: bool,
1717    /// Plaintext samples collected from ZSTD-compressed drops, for
1718    /// training one dictionary after the parallel compress phase.
1719    /// Capped at `MAX_DICT_SAMPLES` to bound memory. Keyed by
1720    /// classifier class — text/code/sparse share a "text" dict,
1721    /// binary gets its own. Compressed/media/incompressible classes
1722    /// don't use ZSTD so their samples aren't collected.
1723    dict_samples_by_class: HashMap<crate::classifier::Class, Vec<Vec<u8>>>,
1724    /// Trained dictionaries keyed by class. Populated by
1725    /// `train_and_apply_dictionary` after the parallel phase. Emitted
1726    /// in the manifest's `dictionary_section` with one entry per
1727    /// class that accumulated enough samples.
1728    trained_dicts_by_class: HashMap<crate::classifier::Class, crate::dictionary::TrainedDictionary>,
1729    /// Drop IDs known to exist in a base image (set when this write
1730    /// is producing a layer via `write_layer`). When a chunk's DropId
1731    /// is in this set, the writer skips compression and emits no slab
1732    /// bytes — the drop is resolved via the overlay chain at read
1733    /// time. `None` for standalone (non-layer) writes.
1734    base_drop_index: Option<std::sync::Arc<dyn BaseDropSet>>,
1735    /// The base image's `ManifestRoot` (set when `base_drop_index`
1736    /// is `Some`). Emitted in the manifest's `delta_linkage` section
1737    /// so readers know which image provides the referenced drops.
1738    /// `None` for standalone writes.
1739    base_root: Option<[u8; 32]>,
1740    /// Compressed-metadata size above which the blob is externalized
1741    /// to a sidecar (issue #187). Defaults to
1742    /// [`METADATA_EXTERNALIZE_THRESHOLD`]; overridable via
1743    /// `WriteConfig::defaults::metadata_externalize_threshold`.
1744    metadata_externalize_threshold: usize,
1745    /// Whether to dedup identical inline file contents into the
1746    /// shared-inline table (issue #189). `true` (default) keeps the
1747    /// historical behavior; `false` emits plain `INLINE_DATA` inodes
1748    /// so images stay readable by pre-#186 readers whose reserved
1749    /// mask rejects the `SHARED_INLINE` flag.
1750    emit_shared_inline: bool,
1751    /// Inline-data cutoff from `WriteConfig::defaults.inline_threshold`.
1752    /// Files at or below this size are stored inline in the metadata
1753    /// blob instead of being chunked into slabs. Set from the profile
1754    /// before `walk`; defaults to the historical constant.
1755    inline_threshold: usize,
1756    /// Streaming-walk sink (TODO.perf/15). When set, `walk` forwards
1757    /// deferred files to the channel instead of buffering them in
1758    /// `pending_files`, so compression starts while the walk is still
1759    /// descending the tree. `None` keeps the collect-then-dispatch
1760    /// shape (used by the non-streaming entry points).
1761    pending_sink: Option<std::sync::mpsc::SyncSender<PendingFile>>,
1762}
1763
1764impl WriteContext {
1765    /// Cap on collected plaintext samples. Enough signal for the
1766    /// FrequencyTrainer without unbounded memory growth on huge inputs.
1767    const MAX_DICT_SAMPLES: usize = 1000;
1768
1769    fn new() -> Self {
1770        Self {
1771            next_inode: 1,
1772            hardlink_targets: std::collections::HashMap::new(),
1773            nlink_counts: std::collections::HashMap::new(),
1774            inode_xattrs: std::collections::HashMap::new(),
1775            inodes: Vec::new(),
1776            dir_nodes: Vec::new(),
1777            drops: Vec::new(),
1778            drop_index: HashSet::new(),
1779            pending_files: Vec::new(),
1780            file_count: 0,
1781            dir_count: 0,
1782            root_inode_number: 0,
1783            chunker: ParallelFastCDC::default(),
1784            classifier: classifier::Classifier,
1785            shared_inline_map: HashMap::new(),
1786            shared_inline_table: Vec::new(),
1787            base_dictionaries: None,
1788            profile_name: None,
1789            metadata_codec: limnifs_core::codec::CODEC_BROTLI,
1790            categorizers_disabled: false,
1791            rw_mode: false,
1792            auto_turnover: false,
1793            collect_dict_samples: false,
1794            dict_samples_by_class: HashMap::new(),
1795            trained_dicts_by_class: HashMap::new(),
1796            base_drop_index: None,
1797            base_root: None,
1798            pending_sink: None,
1799            inline_threshold: INLINE_THRESHOLD,
1800            metadata_externalize_threshold: METADATA_EXTERNALIZE_THRESHOLD,
1801            emit_shared_inline: true,
1802        }
1803    }
1804
1805    fn alloc_inode(&mut self) -> u64 {
1806        let n = self.next_inode;
1807        self.next_inode += 1;
1808        n
1809    }
1810
1811    /// Scan all inline-data inodes and build a dedup table. Only
1812    /// content appearing in > 1 inode is deduplicated; unique inline
1813    /// data stays inline (no overhead change).
1814    fn build_shared_inline_table(&mut self) {
1815        let mut counts: HashMap<[u8; 32], usize> = HashMap::new();
1816        for inode in &self.inodes {
1817            if let PendingContent::Inline(data) = &inode.content {
1818                let h = hash_section(data);
1819                *counts.entry(h).or_default() += 1;
1820            }
1821        }
1822        // Only dedup content that appears more than once.
1823        for inode in &self.inodes {
1824            if let PendingContent::Inline(data) = &inode.content {
1825                let h = hash_section(data);
1826                if counts.get(&h).copied().unwrap_or(0) > 1
1827                    && !self.shared_inline_map.contains_key(&h)
1828                {
1829                    let idx = self.shared_inline_table.len();
1830                    self.shared_inline_table.push(data.clone());
1831                    self.shared_inline_map.insert(h, idx);
1832                }
1833            }
1834        }
1835    }
1836
1837    /// Merge a parallel-processed chunked file's results into the
1838    /// context. Dedup: only new `DropId`s get added to the drops list.
1839    fn merge_chunked_file(&mut self, pf: &PendingFile, result: ChunkedFileResult) {
1840        for (drop_id, plaintext, compressed, codec, flags) in result.drops {
1841            if self.drop_index.insert(drop_id) {
1842                // When dictionary training is enabled, classify the
1843                // plaintext and retain it for per-class dictionary
1844                // training. Text-like classes share a "text" dict;
1845                // Binary gets its own. Compressed/media/incompressible
1846                // don't use ZSTD so we skip them entirely.
1847                let retain_plaintext =
1848                    self.collect_dict_samples && codec == limnifs_core::codec::CODEC_ZSTD;
1849                if retain_plaintext {
1850                    let total: usize = self.dict_samples_by_class.values().map(Vec::len).sum();
1851                    if total < Self::MAX_DICT_SAMPLES {
1852                        let class = self.classifier.classify(&plaintext);
1853                        self.dict_samples_by_class
1854                            .entry(class)
1855                            .or_default()
1856                            .push(plaintext.clone());
1857                    }
1858                }
1859                self.drops.push(PendingDrop {
1860                    id: drop_id,
1861                    plaintext_len: u32::try_from(plaintext.len()).unwrap_or(u32::MAX),
1862                    compressed,
1863                    codec,
1864                    dict_id: limnifs_core::drop_record::NO_DICT,
1865                    plaintext: if retain_plaintext {
1866                        Some(plaintext)
1867                    } else {
1868                        None
1869                    },
1870                    flags,
1871                });
1872            }
1873        }
1874        self.inodes.push(PendingInode {
1875            number: pf.inode_number,
1876            mode: pf.mode,
1877            uid: pf.uid,
1878            gid: pf.gid,
1879            mtime_ns: pf.mtime_ns,
1880            xattrs: Vec::new(),
1881            content: PendingContent::DropBacked {
1882                file_len: pf.file_len,
1883                slices: result.slices,
1884            },
1885        });
1886    }
1887
1888    /// Apply the seine classifier to a chunk and compress it if the
1889    /// class is compressible. Text, Code, and Binary drops get LZ4;
1890    /// Compressed, Media, and Sparse drops stay as store (re-compressing
1891    /// already-compressed data wastes CPU for no gain).
1892    /// Apply the seine classifier to a chunk and compress it if the
1893    /// class is compressible. Text, Code, and Binary drops get LZ4;
1894    /// Compressed, Media, and Sparse drops stay as store.
1895    ///
1896    /// Kept for API compatibility; the parallel writer uses
1897    /// [`process_file`] which inlines this logic.
1898    #[allow(dead_code)]
1899    fn deepen_drop(&self, drop_id: [u8; 32], plaintext: &[u8]) -> PendingDrop {
1900        let class = self.classifier.classify(plaintext);
1901        let (codec, compressed): (u8, std::sync::Arc<[u8]>) = match class {
1902            classifier::Class::Text | classifier::Class::Code | classifier::Class::Binary => {
1903                let c = limnifs_core::codec::compress_lz4_with_size(plaintext);
1904                (limnifs_core::codec::CODEC_LZ4, c.into())
1905            }
1906            _ => (limnifs_core::codec::CODEC_STORE, plaintext.to_vec().into()),
1907        };
1908        PendingDrop {
1909            id: drop_id,
1910            plaintext_len: u32::try_from(plaintext.len()).unwrap_or(u32::MAX),
1911            compressed,
1912            codec,
1913            dict_id: limnifs_core::drop_record::NO_DICT,
1914            plaintext: None,
1915            flags: 0,
1916        }
1917    }
1918
1919    fn walk(&mut self, path: &Path) -> Result<u64, WriteError> {
1920        // Two-phase walk. Phase A (survey, parallel): stat the tree
1921        // with rayon — for trees with 100K+ entries the sequential
1922        // stat storm was the create tail. Phase B (fold,
1923        // sequential, zero syscalls): replicate the exact DFS of
1924        // the previous single-threaded walk so inode numbering,
1925        // dir-node ordering, and error order are byte-identical
1926        // (pinned by the pack-twice determinism test).
1927        let survey = survey_tree(path)?;
1928        self.fold_survey(path, &survey, None)
1929    }
1930
1931    /// Sequential fold over a [`SurveyNode`] — the only place that
1932    /// allocates inodes and emits pending files. Mirrors the
1933    /// pre-survey walk branch-for-branch.
1934    fn fold_survey(
1935        &mut self,
1936        path: &Path,
1937        node: &SurveyNode,
1938        symlink_target: Option<&str>,
1939    ) -> Result<u64, WriteError> {
1940        let meta = node.meta();
1941        if let Some(target) = symlink_target {
1942            let inode_number = self.alloc_inode();
1943            let (mode, uid, gid) = meta.identity();
1944            self.inodes.push(PendingInode {
1945                number: inode_number,
1946                mode,
1947                uid,
1948                gid,
1949                mtime_ns: meta.mtime_ns,
1950                xattrs: Vec::new(),
1951                content: PendingContent::Symlink(target.to_owned()),
1952            });
1953            return Ok(inode_number);
1954        }
1955        if meta.is_dir {
1956            self.dir_count += 1;
1957            let inode_number = self.alloc_inode();
1958            let mut entries: Vec<(String, u64, u8)> = Vec::new();
1959
1960            for (name, child) in &node.children {
1961                let child_path = path.join(name);
1962                let child_inode =
1963                    self.fold_survey(&child_path, child, child.symlink_target.as_deref())?;
1964                let entry_type = if child.meta().is_symlink {
1965                    0x03
1966                } else if child.meta().is_dir {
1967                    0x02
1968                } else {
1969                    0x01
1970                };
1971                entries.push((name.clone(), child_inode, entry_type));
1972            }
1973
1974            // Survey children are already name-sorted; the sort is
1975            // retained so the invariant is local to this fold.
1976            entries.sort_by(|a, b| a.0.cmp(&b.0));
1977            let dir_node = encode_dir_node(&entries);
1978            self.dir_nodes.push(dir_node);
1979            #[cfg(feature = "xattr")]
1980            if !meta.xattrs.is_empty() {
1981                self.inode_xattrs
1982                    .insert(inode_number, to_core_xattrs(&meta.xattrs));
1983            }
1984            let (mode, uid, gid) = meta.identity();
1985            self.inodes.push(PendingInode {
1986                number: inode_number,
1987                mode,
1988                uid,
1989                gid,
1990                mtime_ns: meta.mtime_ns,
1991                xattrs: Vec::new(),
1992                content: PendingContent::Directory(entries),
1993            });
1994            Ok(inode_number)
1995        } else if meta.is_file {
1996            // Hardlinks (unix): a second directory entry for the same
1997            // (dev, ino) references the first inode — no content work,
1998            // no second inode, metadata faithful.
1999            #[cfg(unix)]
2000            if let Some(&existing) = self.hardlink_targets.get(&(meta.dev, meta.ino)) {
2001                *self.nlink_counts.entry(existing).or_insert(1) += 1;
2002                return Ok(existing);
2003            }
2004            self.file_count += 1;
2005            let inode_number = self.alloc_inode();
2006            #[cfg(unix)]
2007            {
2008                self.hardlink_targets
2009                    .insert((meta.dev, meta.ino), inode_number);
2010            }
2011            let file_len = meta.len;
2012            crate::progress::emit_file(path, file_len);
2013
2014            if file_len <= u64::try_from(self.inline_threshold).unwrap_or(u64::MAX) {
2015                let data = std::fs::read(path)?;
2016                #[cfg(feature = "xattr")]
2017                if !meta.xattrs.is_empty() {
2018                    self.inode_xattrs
2019                        .insert(inode_number, to_core_xattrs(&meta.xattrs));
2020                }
2021                let (mode, uid, gid) = meta.identity();
2022                self.inodes.push(PendingInode {
2023                    number: inode_number,
2024                    mode,
2025                    uid,
2026                    gid,
2027                    mtime_ns: meta.mtime_ns,
2028                    xattrs: Vec::new(),
2029                    content: PendingContent::Inline(data),
2030                });
2031            } else {
2032                // Defer to parallel processing — collect the file info.
2033                #[cfg(feature = "xattr")]
2034                if !meta.xattrs.is_empty() {
2035                    self.inode_xattrs
2036                        .insert(inode_number, to_core_xattrs(&meta.xattrs));
2037                }
2038                let (mode, uid, gid) = meta.identity();
2039                let pf = PendingFile {
2040                    inode_number,
2041                    path: path.to_path_buf(),
2042                    mtime_ns: meta.mtime_ns,
2043                    file_len,
2044                    mode,
2045                    uid,
2046                    gid,
2047                };
2048                if let Some(sink) = &self.pending_sink {
2049                    // Streaming mode: hand the file to the compress
2050                    // workers immediately; the bounded channel
2051                    // back-pressures if they fall behind. A send
2052                    // failure means the receiver is gone (a worker
2053                    // hit an unrecoverable error) — abort the walk.
2054                    sink.send(pf).map_err(|_| {
2055                        WriteError::Io(std::io::Error::other("walk: compress pipeline shut down"))
2056                    })?;
2057                } else {
2058                    self.pending_files.push(pf);
2059                }
2060            }
2061            Ok(inode_number)
2062        } else {
2063            #[cfg(unix)]
2064            let kind = {
2065                use std::os::unix::fs::FileTypeExt;
2066                if meta.is_fifo {
2067                    "fifo".to_owned()
2068                } else if meta.is_socket {
2069                    "socket".to_owned()
2070                } else if meta.is_block_device {
2071                    "block device".to_owned()
2072                } else if meta.is_char_device {
2073                    "character device".to_owned()
2074                } else {
2075                    "unknown".to_owned()
2076                }
2077            };
2078            #[cfg(not(unix))]
2079            let kind = "unknown".to_owned();
2080            Err(WriteError::UnsupportedFileType {
2081                path: path.to_path_buf(),
2082                kind,
2083            })
2084        }
2085    }
2086
2087    /// After the parallel compress phase: train one ZSTD dictionary
2088    /// per classifier class with enough samples, then re-compress
2089    /// each ZSTD drop with the dictionary for its class. Keep
2090    /// whichever representation is smaller. Drops that get
2091    /// re-compressed carry the class's `dict_id` in their drop
2092    /// record; the dictionaries are emitted in the manifest's
2093    /// `dictionary_section`.
2094    ///
2095    /// Text/Code/Sparse classes collapse into a single "text" dict
2096    /// (id 0). Binary gets id 1. Other classes don't accumulate
2097    /// samples because their drops aren't ZSTD-compressed.
2098    ///
2099    /// Clears the retained plaintext on every drop to free memory
2100    /// before slab assembly.
2101    fn train_and_apply_dictionary(&mut self, dictionaries: &crate::config::DictionaryConfig) {
2102        if !dictionaries.enabled {
2103            Self::release_dictionary_samples(self);
2104            return;
2105        }
2106
2107        // Layers adopt the base's dictionaries instead of training:
2108        // the swap gate below still decides whether re-emitting
2109        // them pays for itself in THIS image.
2110        if let Some(adopted) = self.base_dictionaries.take() {
2111            for dict in adopted {
2112                match dict.id {
2113                    0 => {
2114                        self.trained_dicts_by_class
2115                            .insert(crate::classifier::Class::Text, dict);
2116                    }
2117                    1 => {
2118                        self.trained_dicts_by_class
2119                            .insert(crate::classifier::Class::Binary, dict);
2120                    }
2121                    _ => {}
2122                }
2123            }
2124            self.apply_trained_dictionaries();
2125            return;
2126        }
2127
2128        let target = usize::try_from(dictionaries.max_dict_size).unwrap_or(65_536);
2129        let min_class = usize::try_from(dictionaries.min_class_size).unwrap_or(0);
2130
2131        // Allocate dict ids: 0 = text (Text/Code/Sparse), 1 = binary.
2132        // Compressed/Media/Incompressible don't accumulate samples
2133        // (their drops aren't ZSTD) so we don't train for them.
2134        let text_classes = [
2135            crate::classifier::Class::Text,
2136            crate::classifier::Class::Code,
2137            crate::classifier::Class::Sparse,
2138        ];
2139        let binary_classes = [crate::classifier::Class::Binary];
2140
2141        // Train text dict from text-like classes' samples combined.
2142        let text_samples: Vec<&[u8]> = text_classes
2143            .iter()
2144            .flat_map(|c| self.dict_samples_by_class.get(c).into_iter().flatten())
2145            .map(Vec::as_slice)
2146            .collect();
2147        let trainer = crate::dictionary::TrainerKind::from_config_str(&dictionaries.trainer);
2148        if text_samples.len() >= min_class {
2149            if let Some(dict) =
2150                crate::dictionary::train_zstd_with_trainer(0, &text_samples, target, trainer)
2151            {
2152                self.trained_dicts_by_class
2153                    .insert(crate::classifier::Class::Text, dict);
2154            }
2155        }
2156        let binary_samples: Vec<&[u8]> = binary_classes
2157            .iter()
2158            .flat_map(|c| self.dict_samples_by_class.get(c).into_iter().flatten())
2159            .map(Vec::as_slice)
2160            .collect();
2161        if binary_samples.len() >= min_class {
2162            if let Some(dict) =
2163                crate::dictionary::train_zstd_with_trainer(1, &binary_samples, target, trainer)
2164            {
2165                self.trained_dicts_by_class
2166                    .insert(crate::classifier::Class::Binary, dict);
2167            }
2168        }
2169
2170        self.apply_trained_dictionaries();
2171    }
2172
2173    /// Re-compress ZSTD drops against `trained_dicts_by_class` and
2174    /// keep the dictionaries only when they pay for themselves (the
2175    /// section's own bytes must be exceeded by per-drop savings).
2176    /// Shared by the train path and the layer adopt path so both
2177    /// face the identical gate and cleanup.
2178    fn apply_trained_dictionaries(&mut self) {
2179        // Allocate dict ids: 0 = text (Text/Code/Sparse), 1 = binary.
2180        // Compressed/Media/Incompressible don't accumulate samples
2181        // (their drops aren't ZSTD) so neither path trains for them.
2182        let text_classes = [
2183            crate::classifier::Class::Text,
2184            crate::classifier::Class::Code,
2185            crate::classifier::Class::Sparse,
2186        ];
2187        let binary_classes = [crate::classifier::Class::Binary];
2188
2189        // Re-compress each ZSTD drop with the dict for its class,
2190        // collecting candidates. The swap decision is taken serially
2191        // below: a dictionary must PAY FOR ITSELF — the per-drop
2192        // savings must exceed the bytes the dictionary section adds
2193        // to the image. Since omnizip 0.21.32's fast-tier match
2194        // finding, the plain path is good enough that a dictionary
2195        // can lose overall on repetitive-text corpora (its 64 KiB
2196        // section costs more than it saves); without this gate the
2197        // dictionary made images LARGER. The collection pass runs in
2198        // parallel — the same one-core-tail shape Phase 2 fixed —
2199        // and position-stable, so output is deterministic either
2200        // way.
2201        use rayon::prelude::*;
2202        let classifier = self.classifier;
2203        let dicts = &self.trained_dicts_by_class;
2204        let candidates: Vec<Option<(std::sync::Arc<[u8]>, u8)>> = self
2205            .drops
2206            .par_iter()
2207            .map(|d| {
2208                if d.codec != limnifs_core::codec::CODEC_ZSTD {
2209                    return None;
2210                }
2211                let Some(plaintext) = d.plaintext.as_ref() else {
2212                    return None;
2213                };
2214                let class = classifier.classify(plaintext);
2215                let dict_class = if text_classes.contains(&class) {
2216                    crate::classifier::Class::Text
2217                } else if binary_classes.contains(&class) {
2218                    crate::classifier::Class::Binary
2219                } else {
2220                    return None;
2221                };
2222                let Some(dict) = dicts.get(&dict_class) else {
2223                    return None;
2224                };
2225                let Ok(dict_compressed) = dict.compress(plaintext) else {
2226                    return None;
2227                };
2228                if dict_compressed.len() < d.compressed.len() {
2229                    Some((dict_compressed.into(), dict.id))
2230                } else {
2231                    None
2232                }
2233            })
2234            .collect();
2235
2236        let saving: isize = candidates
2237            .iter()
2238            .zip(self.drops.iter())
2239            .map(|(c, d)| {
2240                c.as_ref().map_or(0, |(bytes, _)| {
2241                    d.compressed.len() as isize - bytes.len() as isize
2242                })
2243            })
2244            .sum();
2245        let dict_bytes: usize = dicts.values().map(|d| d.content.len()).sum();
2246        if saving > dict_bytes as isize {
2247            for (d, candidate) in self.drops.iter_mut().zip(candidates) {
2248                if let Some((bytes, dict_id)) = candidate {
2249                    d.compressed = bytes;
2250                    d.dict_id = dict_id;
2251                }
2252            }
2253        } else {
2254            // The dictionaries don't pay for themselves on this
2255            // corpus: discard them so assemble emits no dictionary
2256            // section and every drop keeps its tournament
2257            // representation — the image is never larger because of
2258            // the dictionary pass.
2259            self.trained_dicts_by_class.clear();
2260        }
2261
2262        Self::release_dictionary_samples(self);
2263    }
2264
2265    /// Drop the plaintext retained for the dictionary pass; the
2266    /// sample sets are write-time state only.
2267    fn release_dictionary_samples(ctx: &mut Self) {
2268        for d in &mut ctx.drops {
2269            d.plaintext = None;
2270        }
2271        ctx.dict_samples_by_class.clear();
2272    }
2273
2274    /// Env-gated phase timer for assemble profiling (TODO.perf/16).
2275    fn trace_phase(label: &str, start: std::time::Instant) {
2276        if std::env::var_os("LIMNIFS_TRACE_ASSEMBLE").is_some() {
2277            eprintln!("[assemble] {label}: {:?}", start.elapsed());
2278        }
2279    }
2280
2281    fn assemble(mut self) -> WriteArtifact {
2282        let t_assemble = std::time::Instant::now();
2283        let inode_count = self.inodes.len();
2284        let dir_count = self.dir_count;
2285        let drop_count = self.drops.len();
2286
2287        // Partition drops into slabs. Each slab's total byte length
2288        // (header + drop records + solid window) must stay under
2289        // MAX_SLAB_TOTAL_BYTES so the reader's 64 MiB ceiling is never
2290        // exceeded. A single drop larger than the budget gets its own
2291        // slab (we cannot split a drop).
2292        let t = std::time::Instant::now();
2293        let slabs = pack_slabs(&self.drops);
2294        Self::trace_phase("pack_slabs", t);
2295
2296        // Build the shared inline table: deduplicate inline data that
2297        // appears in more than one inode. For N files with identical
2298        // small content, store once and reference by index.
2299        let t = std::time::Instant::now();
2300        if self.emit_shared_inline {
2301            self.build_shared_inline_table();
2302        }
2303        Self::trace_phase("shared_inline_table", t);
2304
2305        let mut metadata_blob = Vec::new();
2306        metadata_blob.extend_from_slice(&u32::try_from(self.inodes.len()).unwrap().to_le_bytes());
2307        for inode in &self.inodes {
2308            self.encode_inode(&mut metadata_blob, inode);
2309        }
2310        metadata_blob
2311            .extend_from_slice(&u32::try_from(self.dir_nodes.len()).unwrap().to_le_bytes());
2312        for node in &self.dir_nodes {
2313            metadata_blob.extend_from_slice(&node.bytes);
2314        }
2315        // Shared inline table (only present if any dedup occurred).
2316        // Reader checks for remaining bytes after dir_nodes.
2317        if !self.shared_inline_table.is_empty() {
2318            metadata_blob.extend_from_slice(
2319                &u32::try_from(self.shared_inline_table.len())
2320                    .unwrap()
2321                    .to_le_bytes(),
2322            );
2323            for entry in &self.shared_inline_table {
2324                let len = u32::try_from(entry.len()).expect("shared entry fits u32");
2325                metadata_blob.extend_from_slice(&len.to_le_bytes());
2326                metadata_blob.extend_from_slice(entry);
2327            }
2328        }
2329
2330        Self::trace_phase("metadata_encode", t);
2331        // Compress the metadata blob. Metadata is highly compressible
2332        // (sequential inode numbers, repeated modes, natural-language
2333        // file names) — even low Brotli quality yields 4–8× on source
2334        // trees. Pick quality by size: small blobs cost nothing to
2335        // compress at q5; large blobs (e.g. 50 K-inode trees) would
2336        // dominate create time at q5, so step down to q2.
2337        let uncompressed_len =
2338            u32::try_from(metadata_blob.len()).expect("metadata blob length fits u32");
2339        let t = std::time::Instant::now();
2340        let metadata_hash = hash_section(&metadata_blob);
2341        let metadata_codec = self.metadata_codec;
2342        let metadata_quality = if metadata_blob.len() > METADATA_LARGE_BLOB_THRESHOLD {
2343            METADATA_LARGE_BLOB_QUALITY
2344        } else {
2345            METADATA_SMALL_BLOB_QUALITY
2346        };
2347        let compressed_blob = if metadata_codec == limnifs_core::codec::CODEC_BROTLI {
2348            limnifs_core::codec::compress_brotli_with_quality(&metadata_blob, metadata_quality)
2349                .unwrap_or_else(|_| metadata_blob.clone())
2350        } else {
2351            limnifs_core::codec::compress(metadata_codec, &metadata_blob)
2352                .unwrap_or_else(|_| metadata_blob.clone())
2353        };
2354        Self::trace_phase("metadata_compress", t);
2355        let (on_wire_codec, on_wire_blob) = if compressed_blob.len() < metadata_blob.len() {
2356            (metadata_codec, compressed_blob)
2357        } else {
2358            (limnifs_core::codec::CODEC_STORE, metadata_blob.clone())
2359        };
2360
2361        // Decide inline vs sidecar based on the COMPRESSED length,
2362        // clamped to the reader's inline ceiling regardless of config
2363        // (inline metadata above it is unreadable by default readers).
2364        let externalize_at = self
2365            .metadata_externalize_threshold
2366            .min(limnifs_core::metadata_reference::DEFAULT_INLINE_METADATA_MAX_BYTES as usize);
2367        let (metadata_sidecar, inline_data, metadata_locator_count) =
2368            if on_wire_blob.len() > externalize_at {
2369                // Content-derived sidecar name: an RW commit NEVER
2370                // overwrites the metadata file an already-open
2371                // reader's manifest references (same name, different
2372                // bytes = torn blob). Identical trees reuse the same
2373                // name; divergent generations accumulate until
2374                // turnover / gc reclaims them.
2375                let h = hash_section(&on_wire_blob);
2376                let mut h8 = String::with_capacity(8);
2377                for b in &h[..4] {
2378                    h8.push_str(&format!("{b:02x}"));
2379                }
2380                let locator = format!("file:metadata-{h8}.bin");
2381                let sidecar = MetadataSidecar {
2382                    bytes: on_wire_blob.clone(),
2383                    locator,
2384                };
2385                (Some(sidecar), None, 1u32)
2386            } else {
2387                (None, Some(on_wire_blob.clone()), 0u32)
2388            };
2389
2390        let mut manifest = Vec::new();
2391
2392        let header_start = manifest.len();
2393        manifest.extend_from_slice(&ManifestHeader::current().to_bytes());
2394        let header_end = manifest.len();
2395
2396        let flags_start = manifest.len();
2397        manifest.push(FEATURE_FLAGS_SECTION_VERSION);
2398        manifest.extend_from_slice(&0u32.to_le_bytes());
2399        let flags_end = manifest.len();
2400
2401        // metadata_reference v2: hash + uncompressed_len + codec +
2402        // locator_count + locators + inline_data_len + inline_data.
2403        let meta_ref_start = manifest.len();
2404        manifest.push(METADATA_REFERENCE_SECTION_VERSION_2);
2405        manifest.extend_from_slice(&metadata_hash);
2406        manifest.extend_from_slice(&uncompressed_len.to_le_bytes());
2407        manifest.push(on_wire_codec);
2408        manifest.extend_from_slice(&metadata_locator_count.to_le_bytes());
2409        if let Some(sidecar) = &metadata_sidecar {
2410            let loc_bytes = sidecar.locator.as_bytes();
2411            let loc_len = u32::try_from(loc_bytes.len()).expect("locator fits u32");
2412            manifest.extend_from_slice(&loc_len.to_le_bytes());
2413            manifest.extend_from_slice(loc_bytes);
2414        }
2415        match &inline_data {
2416            Some(blob) => {
2417                let inline_len = u32::try_from(blob.len()).expect("metadata fits u32");
2418                manifest.extend_from_slice(&inline_len.to_le_bytes());
2419                manifest.extend_from_slice(blob);
2420            }
2421            None => {
2422                manifest.extend_from_slice(&0u32.to_le_bytes());
2423            }
2424        }
2425        let meta_ref_end = manifest.len();
2426
2427        let slab_index_start = manifest.len();
2428        manifest.push(SLAB_INDEX_SECTION_VERSION);
2429        manifest.extend_from_slice(&u32::try_from(slabs.len()).unwrap().to_le_bytes());
2430        for slab in &slabs {
2431            manifest.extend_from_slice(&slab.id.to_bytes());
2432            manifest.extend_from_slice(&1u32.to_le_bytes());
2433            let loc_bytes = slab.locator.as_bytes();
2434            let loc_len = u32::try_from(loc_bytes.len()).expect("locator fits u32");
2435            manifest.extend_from_slice(&loc_len.to_le_bytes());
2436            manifest.extend_from_slice(loc_bytes);
2437        }
2438        let slab_index_end = manifest.len();
2439
2440        let history_start = manifest.len();
2441        manifest.push(HISTORY_SECTION_VERSION);
2442        manifest.extend_from_slice(&1u32.to_le_bytes());
2443        manifest.push(0x01);
2444        manifest.extend_from_slice(&0u64.to_le_bytes());
2445        manifest.extend_from_slice(&0u32.to_le_bytes());
2446        manifest.extend_from_slice(&0u32.to_le_bytes());
2447        let history_end = manifest.len();
2448
2449        // ProfileDescriptor section (optional — appended after history).
2450        // Records which overhead layers were active so any reader can
2451        // handle the image correctly. Only emitted if a profile name
2452        // was set.
2453        let profile_desc_start = manifest.len();
2454        if let Some(ref name) = self.profile_name {
2455            let desc = limnifs_core::profile_descriptor::ProfileDescriptor {
2456                version: limnifs_core::profile_descriptor::PROFILE_DESCRIPTOR_SECTION_VERSION,
2457                profile_name: Some(name.clone()),
2458                blake3_hashing: true,
2459                cross_file_dedup: true,
2460                content_classification: !self.categorizers_disabled,
2461                integrity_verify: true,
2462                read_write: self.rw_mode,
2463                auto_turnover: self.auto_turnover,
2464            };
2465            limnifs_core::profile_descriptor::encode_profile_descriptor(&desc, &mut manifest);
2466        }
2467        let profile_desc_end = manifest.len();
2468
2469        // DictionarySection (optional — emitted when dictionaries
2470        // were trained during the post-parallel pass). Contains the
2471        // `(codec_id, class_id, data)` triples referenced by drop
2472        // records' `dict_id` field. One entry per class with enough
2473        // samples to train: text (id 0), binary (id 1).
2474        if !self.trained_dicts_by_class.is_empty() {
2475            let dicts: Vec<_> = self
2476                .trained_dicts_by_class
2477                .values()
2478                .map(|d| limnifs_core::dictionary_section::Dictionary {
2479                    codec_id: d.codec,
2480                    class_id: d.id,
2481                    data: d.content.clone(),
2482                })
2483                .collect();
2484            let section = limnifs_core::dictionary_section::DictionarySection {
2485                version: limnifs_core::dictionary_section::DICTIONARY_SECTION_VERSION,
2486                dicts,
2487            };
2488            limnifs_core::dictionary_section::encode_dictionary_section(&section, &mut manifest);
2489        }
2490
2491        let dictionary_end = manifest.len();
2492
2493        // DeltaLinkage section (optional — emitted only by `write_layer`).
2494        // Carries `base_root` so readers can resolve referenced drops via
2495        // the overlay chain. The section's hash feeds the
2496        // `SectionHashes::delta_linkage` slot, which is empty for
2497        // standalone images.
2498        let delta_linkage_hash = if let Some(base_root) = self.base_root {
2499            let delta_start = manifest.len();
2500            // Inline-encode the delta linkage section (version 1):
2501            // [version:u8][base_root:32][tree_op_count:u32=0]. Tree ops
2502            // are empty because the metadata blob carries the full new
2503            // tree — readers see `base_root` and know to walk the
2504            // overlay chain for any DropId not present in local slabs.
2505            manifest.push(limnifs_core::delta_linkage::DELTA_LINKAGE_SECTION_VERSION);
2506            manifest.extend_from_slice(&base_root);
2507            manifest.extend_from_slice(&0u32.to_le_bytes());
2508            hash_section(&manifest[delta_start..])
2509        } else {
2510            hash_empty_section()
2511        };
2512        let _ = dictionary_end;
2513
2514        let hashes = SectionHashes {
2515            metadata: metadata_hash,
2516            format_header: hash_section(&manifest[header_start..header_end]),
2517            feature_flags: hash_section(&manifest[flags_start..flags_end]),
2518            metadata_reference: hash_section(&manifest[meta_ref_start..meta_ref_end]),
2519            slab_index: hash_section(&manifest[slab_index_start..slab_index_end]),
2520            crypto_params: hash_empty_section(),
2521            ec_params: hash_empty_section(),
2522            dms_policy: hash_empty_section(),
2523            delta_linkage: delta_linkage_hash,
2524            history: hash_section(&manifest[history_start..history_end]),
2525            // The Merkle construction doesn't currently include a
2526            // dictionary_section hash slot. Treat the section as
2527            // crypto-params-equivalent (covered by the metadata hash)
2528            // for now; documenting this with an explicit comment so
2529            // the next reader knows where to add a hash slot if the
2530            // spec grows one.
2531            // TODO: spec section-hash for dictionary_section.
2532            // For now use hash_empty_section() so the structure compiles;
2533            // a future spec rev will add a dedicated slot.
2534            // (Section bytes are still content-addressed via the
2535            // slab_index hash and the manifest's Merkle root.)
2536        };
2537        let merkle_root = compute_merkle_root(&hashes);
2538
2539        WriteArtifact {
2540            bytes: manifest,
2541            merkle_root,
2542            slabs,
2543            metadata_sidecar,
2544            inode_count,
2545            file_count: self.file_count,
2546            dir_count,
2547            drop_count,
2548            root_inode_number: self.root_inode_number,
2549        }
2550    }
2551
2552    fn encode_inode(&self, out: &mut Vec<u8>, inode: &PendingInode) {
2553        out.extend_from_slice(&inode.number.to_le_bytes());
2554        out.extend_from_slice(&inode.mode.to_le_bytes());
2555        out.extend_from_slice(&inode.uid.to_le_bytes());
2556        out.extend_from_slice(&inode.gid.to_le_bytes());
2557        out.extend_from_slice(&inode.mtime_ns.to_le_bytes());
2558        out.extend_from_slice(&inode.mtime_ns.to_le_bytes());
2559        let nlink = self.nlink_counts.get(&inode.number).copied().unwrap_or(1);
2560        out.extend_from_slice(&nlink.to_le_bytes());
2561        let xattrs: &[limnifs_core::inode::XAttr] = self
2562            .inode_xattrs
2563            .get(&inode.number)
2564            .map_or(inode.xattrs.as_slice(), std::convert::AsRef::as_ref);
2565        // The xattr block sits between the flags byte and the
2566        // content handle (parse order: flags -> atime? -> xattrs ->
2567        // content); each arm's flag byte goes through this helper.
2568        let mut flags_and_xattrs = move |out: &mut Vec<u8>, base: u8| {
2569            if xattrs.is_empty() {
2570                out.push(base);
2571                return;
2572            }
2573            out.push(base | limnifs_core::inode::INODE_FLAG_HAS_XATTRS);
2574            let count = u32::try_from(xattrs.len()).expect("xattr count fits u32");
2575            out.extend_from_slice(&count.to_le_bytes());
2576            for x in xattrs {
2577                out.push(x.namespace);
2578                let key = x.key.as_bytes();
2579                let key_len = u32::try_from(key.len()).expect("xattr key fits u32");
2580                out.extend_from_slice(&key_len.to_le_bytes());
2581                out.extend_from_slice(key);
2582                let value_len = u32::try_from(x.value.len()).expect("xattr value fits u32");
2583                out.extend_from_slice(&value_len.to_le_bytes());
2584                out.extend_from_slice(&x.value);
2585            }
2586        };
2587        match &inode.content {
2588            PendingContent::Inline(data) => {
2589                let h = hash_section(data);
2590                if let Some(&idx) = self.shared_inline_map.get(&h) {
2591                    // Deduplicated: emit shared-inline flag + index.
2592                    flags_and_xattrs(out, INODE_FLAG_INLINE_DATA | INODE_FLAG_SHARED_INLINE);
2593                    out.extend_from_slice(&(idx as u32).to_le_bytes());
2594                } else {
2595                    flags_and_xattrs(out, INODE_FLAG_INLINE_DATA);
2596                    let len = u32::try_from(data.len()).expect("data fits u32");
2597                    out.extend_from_slice(&len.to_le_bytes());
2598                    out.extend_from_slice(data);
2599                }
2600            }
2601            PendingContent::DropBacked { file_len, slices } => {
2602                flags_and_xattrs(out, 0x00);
2603                let slice_count = u32::try_from(slices.len()).expect("slice count fits u32");
2604                out.extend_from_slice(&slice_count.to_le_bytes());
2605                for slice in slices {
2606                    out.extend_from_slice(&slice.file_byte_start.to_le_bytes());
2607                    out.extend_from_slice(&slice.file_byte_end.to_le_bytes());
2608                    out.extend_from_slice(&slice.drop_id);
2609                    // drop_byte_start = 0 (slice covers the whole drop)
2610                    out.extend_from_slice(&0u32.to_le_bytes());
2611                    // drop_byte_len = the byte length of this slice in the
2612                    // drop's decompressed plaintext. Each slice maps to
2613                    // exactly one chunk, so this equals the file range.
2614                    let drop_byte_len = u32::try_from(slice.file_byte_end - slice.file_byte_start)
2615                        .expect("slice range fits u32");
2616                    out.extend_from_slice(&drop_byte_len.to_le_bytes());
2617                }
2618                let _ = file_len;
2619            }
2620            PendingContent::Symlink(target) => {
2621                // The reader dispatches on the inode's S_IFMT bits and
2622                // reads target_len + target directly (flags unused
2623                // for non-regular inodes).
2624                flags_and_xattrs(out, 0x00);
2625                let t = target.as_bytes();
2626                let len = u32::try_from(t.len()).expect("target fits u32");
2627                out.extend_from_slice(&len.to_le_bytes());
2628                out.extend_from_slice(t);
2629            }
2630            PendingContent::Directory(entries) => {
2631                flags_and_xattrs(out, 0x00);
2632                let node = self
2633                    .dir_nodes
2634                    .iter()
2635                    .find(|n| n.entries == *entries)
2636                    .expect("directory node must exist");
2637                out.extend_from_slice(&node.hash);
2638            }
2639        }
2640    }
2641}
2642
2643/// Resolve a locator URI to the local sidecar file name, refusing
2644/// non-flat paths (CWE-22 — see
2645/// `limnifs_core::locator::local_sidecar_name`). Writer-emitted
2646/// locators are always flat, so this only fires on foreign/malicious
2647/// manifests.
2648fn sidecar_name(locator: &str) -> Result<&str, WriteError> {
2649    limnifs_core::locator::local_sidecar_name(locator)
2650        .map_err(|e| WriteError::Io(std::io::Error::other(format!("{e}"))))
2651}
2652
2653fn encode_dir_node(entries: &[(String, u64, u8)]) -> DirNode {
2654    let mut bytes = Vec::new();
2655    bytes.push(1u8);
2656    let count = u32::try_from(entries.len()).expect("entry count fits u32");
2657    bytes.extend_from_slice(&count.to_le_bytes());
2658    for (name, inode_number, entry_type) in entries {
2659        let name_bytes = name.as_bytes();
2660        let name_len = u32::try_from(name_bytes.len()).expect("name fits u32");
2661        bytes.extend_from_slice(&name_len.to_le_bytes());
2662        bytes.extend_from_slice(name_bytes);
2663        bytes.extend_from_slice(&inode_number.to_le_bytes());
2664        bytes.push(*entry_type);
2665    }
2666    let hash = hash_section(&bytes);
2667    DirNode {
2668        entries: entries.to_vec(),
2669        bytes,
2670        hash,
2671    }
2672}
2673
2674/// Partition `drops` into one or more slabs, each fitting under
2675/// [`MAX_SLAB_TOTAL_BYTES`]. Slab ordinal starts at 0 and increments.
2676/// Each slab's `SlabId` hash is `BLAKE3(slab_content)` so identical
2677/// content yields identical slab IDs (deterministic).
2678///
2679/// A single drop larger than `MAX_SLAB_TOTAL_BYTES - SLAB_HEADER_LEN`
2680/// still produces one slab — we cannot split a drop, and the spec
2681/// permits the reader to raise its ceiling for that case.
2682fn pack_slabs(drops: &[PendingDrop]) -> Vec<SlabArtifact> {
2683    // Filter out CODEC_REFERENCED sentinel drops — they exist in the
2684    // writer's in-memory state for inode/slice bookkeeping but are
2685    // resolved via the overlay chain at read time, never stored in
2686    // this image's slabs.
2687    let local_drops: Vec<&PendingDrop> = drops
2688        .iter()
2689        .filter(|d| d.codec != limnifs_core::codec::CODEC_REFERENCED)
2690        .collect();
2691    if local_drops.is_empty() {
2692        return Vec::new();
2693    }
2694
2695    let max_content = MAX_SLAB_TOTAL_BYTES.saturating_sub(SLAB_HEADER_LEN);
2696
2697    // Phase 1 (sequential): scan drops and partition into slab groups.
2698    // Each slab group is the list of drops that will share a slab window.
2699    // The grouping depends on per-drop compressed size + record overhead
2700    // — this is a sequential scan with a running size budget.
2701    let mut slab_groups: Vec<Vec<&PendingDrop>> = Vec::new();
2702    let mut current: Vec<&PendingDrop> = Vec::new();
2703    let mut current_size: usize = 0;
2704
2705    for drop in &local_drops {
2706        let footprint = drop.slab_footprint();
2707        if !current.is_empty() && current_size + footprint > max_content {
2708            slab_groups.push(std::mem::take(&mut current));
2709            current_size = 0;
2710        }
2711        current.push(*drop);
2712        current_size += footprint;
2713    }
2714    if !current.is_empty() {
2715        slab_groups.push(current);
2716    }
2717
2718    // Phase 2 (parallel): encode each slab independently. Slab encoding
2719    // has no cross-slab state — each slab's `offset_in_window` starts
2720    // at 0, its hash is over its own content, its ordinal is its index
2721    // in the slab_groups vector. Rayon parallelises across slabs;
2722    // large images with many slabs get N-core speedup on this phase.
2723    use rayon::prelude::*;
2724    slab_groups
2725        .par_iter()
2726        .enumerate()
2727        .map(|(ordinal, group)| {
2728            let ordinal_u64 = u64::try_from(ordinal).expect("slab count fits u64");
2729            encode_slab(ordinal_u64, group)
2730        })
2731        .collect()
2732}
2733
2734/// Encode a single slab from a non-empty slice of drops. Per-slab
2735/// `offset_in_window` is computed fresh; there is no global offset
2736/// state on `PendingDrop`.
2737fn encode_slab(ordinal: u64, drops: &[&PendingDrop]) -> SlabArtifact {
2738    // Each drop record is a fixed 49-byte entry: id(32) +
2739    // plaintext_len(4) + representation(3) + solid_window_index(1)
2740    // + offset_in_window(4) + window_len(4) + dict_id(1). Pre-sizing
2741    // the records Vec avoids per-drop realloc and amortises to a
2742    // single memcpy per field rather than bounds-check per call.
2743    // v2: 49-byte v1 record + trailing flags byte.
2744    const DROP_RECORD_LEN: usize = 50;
2745    let mut drop_records = Vec::with_capacity(drops.len() * DROP_RECORD_LEN);
2746    let mut solid_window = Vec::new();
2747    let mut drop_ids = Vec::with_capacity(drops.len());
2748    let mut offset_in_window: u32 = 0;
2749
2750    for drop in drops {
2751        let plaintext_len = drop.plaintext_len_value();
2752        let window_len = drop.len_in_window();
2753        drop_records.extend_from_slice(&drop.id);
2754        drop_records.extend_from_slice(&plaintext_len.to_le_bytes());
2755        // representation: (codec, aead=0, ec=0)
2756        drop_records.extend_from_slice(&[drop.codec, 0x00, 0x00]);
2757        drop_records.push(0x00); // solid_window_index
2758        drop_records.extend_from_slice(&offset_in_window.to_le_bytes());
2759        drop_records.extend_from_slice(&window_len.to_le_bytes());
2760        drop_records.push(drop.dict_id); // dict_id: NO_DICT (0xFF) or trained id (0..=254)
2761        drop_records.push(drop.flags); // flags: bit0 = SEEKABLE container
2762        solid_window.extend_from_slice(&drop.compressed);
2763        drop_ids.push(drop.id);
2764        offset_in_window = offset_in_window
2765            .checked_add(window_len)
2766            .expect("slab window size fits u32");
2767    }
2768
2769    let slab_content = [&drop_records[..], &solid_window[..]].concat();
2770    let slab_hash = hash_section(&slab_content);
2771    let slab_id = SlabId::new(ordinal, slab_hash);
2772
2773    let total_length = SLAB_HEADER_LEN + slab_content.len();
2774    let mut slab_bytes = Vec::with_capacity(total_length);
2775    slab_bytes.extend_from_slice(b"LIM1");
2776    slab_bytes.extend_from_slice(&1u16.to_le_bytes()); // the slab format version
2777    slab_bytes.extend_from_slice(&slab_id.to_bytes());
2778    slab_bytes.extend_from_slice(
2779        &u64::try_from(total_length)
2780            .unwrap_or(u64::MAX)
2781            .to_le_bytes(),
2782    );
2783    slab_bytes.push(0x00);
2784    slab_bytes.push(0x00);
2785    slab_bytes.extend_from_slice(&slab_content);
2786
2787    // Content-derived slab name — see the metadata sidecar comment
2788    // in `assemble`: RW commits must not overwrite slabs that live
2789    // manifests still reference.
2790    let mut h8 = String::with_capacity(8);
2791    for b in &slab_id.hash[..4] {
2792        h8.push_str(&format!("{b:02x}"));
2793    }
2794    let locator = format!("file:slab-{ordinal}-{h8}.bin");
2795
2796    SlabArtifact {
2797        id: slab_id,
2798        bytes: slab_bytes,
2799        locator,
2800        drop_ids,
2801    }
2802}
2803
2804#[cfg(test)]
2805fn pseudo_random_bytes(seed: u64, count: usize) -> Vec<u8> {
2806    let mut state = seed;
2807    let mut out = Vec::with_capacity(count);
2808    for _ in 0..count {
2809        state = state
2810            .wrapping_mul(6_364_136_223_846_793_005)
2811            .wrapping_add(1_442_695_040_888_963_407);
2812        out.push(u8::try_from(state >> 56).expect("fits u8"));
2813    }
2814    out
2815}
2816
2817#[cfg(test)]
2818mod tests {
2819    use super::*;
2820    use limnifs_core::ManifestCursor;
2821
2822    #[test]
2823    fn write_stream_packs_single_named_stream() {
2824        // Stream 256 KiB of repetitive text through write_stream and
2825        // verify the artifact has a single root file at the requested
2826        // name with the expected size.
2827        let temp = std::env::temp_dir().join(format!(
2828            "limnifs-write-stream-test-{}-{}",
2829            std::process::id(),
2830            std::time::SystemTime::now()
2831                .duration_since(std::time::UNIX_EPOCH)
2832                .unwrap()
2833                .as_nanos()
2834        ));
2835        std::fs::create_dir_all(&temp).expect("create temp dir");
2836
2837        let content = b"stream test content line\n".repeat(10_000); // ~240 KiB
2838        let cursor = std::io::Cursor::new(content.clone());
2839        let config = WriteConfig::default_v0_1();
2840        let artifact = write_stream("streamed.txt", cursor, &config).expect("write_stream");
2841
2842        assert!(!artifact.bytes.is_empty(), "manifest bytes non-empty");
2843        assert!(!artifact.slabs.is_empty(), "at least one slab produced");
2844        // The artifact's manifest encodes the file metadata; we don't
2845        // deeply inspect it here (conformance suite covers that), but
2846        // we do confirm the writer produced something well-formed
2847        // enough that round-tripping through the reader works.
2848        let total_drop_bytes: usize = artifact.slabs.iter().map(|s| s.bytes.len()).sum();
2849        assert!(total_drop_bytes > 0, "drops non-empty");
2850
2851        let _ = std::fs::remove_dir_all(&temp);
2852    }
2853
2854    #[test]
2855    fn write_layer_references_base_drops() {
2856        // Build a base image with a 1 MiB text file, then build a
2857        // layer that ADDS a new file AND includes the same text file.
2858        // The layer's slab bytes must be small (just the new file)
2859        // because the text file's chunks hit the base's drop set and
2860        // are emitted as `CODEC_REFERENCED`.
2861        let temp = std::env::temp_dir().join(format!(
2862            "limnifs-write-layer-test-{}-{}",
2863            std::process::id(),
2864            std::time::SystemTime::now()
2865                .duration_since(std::time::UNIX_EPOCH)
2866                .unwrap()
2867                .as_nanos()
2868        ));
2869        std::fs::create_dir_all(&temp).expect("create temp dir");
2870
2871        // Base: 1 MiB of repetitive text + a small unique file.
2872        let base_dir = temp.join("base");
2873        std::fs::create_dir_all(&base_dir).expect("base dir");
2874        let text = b"layer test content line\n".repeat(50_000); // ~1.15 MiB
2875        std::fs::write(base_dir.join("shared.txt"), &text).expect("write shared");
2876        std::fs::write(base_dir.join("base-only.txt"), b"only in base").expect("write base-only");
2877
2878        let config = WriteConfig::default_v0_1();
2879        let base_artifact = write_directory_with_config(&base_dir, &config).expect("base");
2880
2881        let base_manifest = temp.join("base.lim");
2882        std::fs::write(&base_manifest, &base_artifact.bytes).expect("write base manifest");
2883        for slab in &base_artifact.slabs {
2884            let slab_name = sidecar_name(&slab.locator).expect("slab locator");
2885            std::fs::write(temp.join(slab_name), &slab.bytes).expect("write base slab");
2886        }
2887
2888        // Layer: same shared.txt (must dedup against base) + a new file.
2889        let layer_dir = temp.join("layer");
2890        std::fs::create_dir_all(&layer_dir).expect("layer dir");
2891        std::fs::write(layer_dir.join("shared.txt"), &text).expect("write shared in layer");
2892        std::fs::write(layer_dir.join("new.txt"), b"fresh content in layer").expect("write new");
2893
2894        let layer_artifact = write_layer(&base_manifest, &layer_dir, &config).expect("layer");
2895
2896        // The layer's slabs should be SMALL — only the new file's
2897        // content. shared.txt's chunks are referenced via the base.
2898        let layer_slab_bytes: usize = layer_artifact.slabs.iter().map(|s| s.bytes.len()).sum();
2899        let base_slab_bytes: usize = base_artifact.slabs.iter().map(|s| s.bytes.len()).sum();
2900        assert!(
2901            layer_slab_bytes < base_slab_bytes / 4,
2902            "layer slabs ({}) should be much smaller than base ({}) — layering failed",
2903            layer_slab_bytes,
2904            base_slab_bytes
2905        );
2906
2907        // The layer manifest must contain the base's Merkle root in
2908        // its delta_linkage section.
2909        let base_root = base_artifact.merkle_root.as_bytes();
2910        assert!(
2911            layer_artifact
2912                .bytes
2913                .windows(32)
2914                .any(|w| w == base_root.as_slice()),
2915            "layer manifest must contain base's ManifestRoot bytes"
2916        );
2917
2918        let _ = std::fs::remove_dir_all(&temp);
2919    }
2920
2921    #[test]
2922    fn tournament_short_circuits_on_highly_compressible_chunk() {
2923        // Repetitive text compresses to <25% under LZ4. Tournament
2924        // should accept LZ4 and skip the slower Brotli pass.
2925        let chunk = b"hello world ".repeat(500);
2926        let tunables = limnifs_core::codec::CodecTunables::default();
2927        let tournament = TournamentSpec {
2928            codec_ids: vec![
2929                limnifs_core::codec::CODEC_LZ4,
2930                limnifs_core::codec::CODEC_BROTLI,
2931            ],
2932            min_size: 16,
2933            skip_for_binary: false,
2934            short_circuit_permille: 250,
2935        };
2936        let (codec_id, compressed) = compress_chunk_with_tournament(
2937            &chunk,
2938            classifier::Class::Text,
2939            limnifs_core::codec::CODEC_BROTLI,
2940            limnifs_core::codec::CODEC_LZ4,
2941            &tunables,
2942            &tournament,
2943        );
2944        assert_eq!(codec_id, limnifs_core::codec::CODEC_LZ4);
2945        assert!(compressed.len() < chunk.len());
2946    }
2947
2948    #[test]
2949    fn tournament_runs_all_codecs_when_short_circuit_disabled() {
2950        // short_circuit_permille = 0 means "never short-circuit". The
2951        // tournament must try every codec and pick the smallest.
2952        // With omnizip 0.14.40, ZSTD (cached Huffman table) typically
2953        // beats both LZ4 and Brotli's Phase-C partial encoder on
2954        // repetitive text, so we include it in the tournament.
2955        let chunk = b"hello world ".repeat(500);
2956        let tunables = limnifs_core::codec::CodecTunables::default();
2957        let tournament = TournamentSpec {
2958            codec_ids: vec![
2959                limnifs_core::codec::CODEC_LZ4,
2960                limnifs_core::codec::CODEC_BROTLI,
2961                limnifs_core::codec::CODEC_ZSTD,
2962            ],
2963            min_size: 16,
2964            skip_for_binary: false,
2965            short_circuit_permille: 0,
2966        };
2967        let (codec_id, compressed) = compress_chunk_with_tournament(
2968            &chunk,
2969            classifier::Class::Text,
2970            limnifs_core::codec::CODEC_BROTLI,
2971            limnifs_core::codec::CODEC_LZ4,
2972            &tunables,
2973            &tournament,
2974        );
2975        // All three codecs should be tried; the smallest wins. With
2976        // omnizip 0.16.40's long copy fix (MAX_COPY 271→4096), Brotli
2977        // now beats ZSTD on repetitive text. Either is acceptable.
2978        assert!(
2979            codec_id == limnifs_core::codec::CODEC_ZSTD
2980                || codec_id == limnifs_core::codec::CODEC_BROTLI,
2981            "expected ZSTD or Brotli to win, got codec {codec_id}"
2982        );
2983        assert!(compressed.len() < chunk.len());
2984    }
2985
2986    #[test]
2987    fn tournament_skips_for_binary_when_configured() {
2988        let chunk = vec![0u8; 4096];
2989        let tunables = limnifs_core::codec::CodecTunables::default();
2990        let tournament = TournamentSpec {
2991            codec_ids: vec![limnifs_core::codec::CODEC_BROTLI],
2992            min_size: 16,
2993            skip_for_binary: true,
2994            short_circuit_permille: 250,
2995        };
2996        let (codec_id, _compressed) = compress_chunk_with_tournament(
2997            &chunk,
2998            classifier::Class::Binary,
2999            limnifs_core::codec::CODEC_BROTLI,
3000            limnifs_core::codec::CODEC_LZ4,
3001            &tunables,
3002            &tournament,
3003        );
3004        // skip_for_binary → use binary_codec (LZ4) directly, never Brotli.
3005        assert_eq!(codec_id, limnifs_core::codec::CODEC_LZ4);
3006    }
3007
3008    #[test]
3009    fn tournament_small_chunk_uses_preferred_codec() {
3010        let chunk = b"tiny";
3011        let tunables = limnifs_core::codec::CodecTunables::default();
3012        let tournament = TournamentSpec {
3013            codec_ids: vec![limnifs_core::codec::CODEC_BROTLI],
3014            min_size: 1024,
3015            skip_for_binary: false,
3016            short_circuit_permille: 0,
3017        };
3018        let (codec_id, _compressed) = compress_chunk_with_tournament(
3019            chunk,
3020            classifier::Class::Text,
3021            limnifs_core::codec::CODEC_BROTLI,
3022            limnifs_core::codec::CODEC_LZ4,
3023            &tunables,
3024            &tournament,
3025        );
3026        // Below min_size → preferred codec (brotli for text) directly.
3027        assert_eq!(codec_id, limnifs_core::codec::CODEC_STORE);
3028    }
3029
3030    #[test]
3031    fn tournament_falls_back_to_store_when_no_codec_compresses() {
3032        // Random data — no codec should improve on store. We use the
3033        // pseudo-random generator from the test helpers to get
3034        // deterministic but incompressible bytes.
3035        let chunk = pseudo_random_bytes(42, 4096);
3036        let tunables = limnifs_core::codec::CodecTunables::default();
3037        let tournament = TournamentSpec {
3038            codec_ids: vec![limnifs_core::codec::CODEC_LZ4],
3039            min_size: 16,
3040            skip_for_binary: false,
3041            short_circuit_permille: 0,
3042        };
3043        let (codec_id, compressed) = compress_chunk_with_tournament(
3044            &chunk,
3045            classifier::Class::Binary,
3046            limnifs_core::codec::CODEC_BROTLI,
3047            limnifs_core::codec::CODEC_LZ4,
3048            &tunables,
3049            &tournament,
3050        );
3051        assert_eq!(codec_id, limnifs_core::codec::CODEC_STORE);
3052        assert_eq!(compressed.len(), chunk.len());
3053    }
3054
3055    #[test]
3056    fn dictionaries_enabled_emits_dictionary_section_when_enough_samples() {
3057        // Many small text files with shared vocabulary → FrequencyTrainer
3058        // should find a dictionary. We assert the section appears in
3059        // the manifest, regardless of whether the trained dict beats
3060        // per-drop compression (the trainer is content-dependent).
3061        let temp = std::env::temp_dir().join(format!(
3062            "limnifs-write-test-{}-dict-{}",
3063            std::process::id(),
3064            std::time::SystemTime::now()
3065                .duration_since(std::time::UNIX_EPOCH)
3066                .map(|d| d.as_nanos() as u64)
3067                .unwrap_or(0),
3068        ));
3069        let _ = std::fs::remove_dir_all(&temp);
3070        std::fs::create_dir_all(&temp).expect("mkdir");
3071
3072        // Generate 200 similar small files just above INLINE_THRESHOLD
3073        // so they go through the slab path.
3074        for i in 0..200 {
3075            // Repeated vocabulary the trainer can exploit.
3076            let content = format!(
3077                "function test_case_{i}() {{ return constant + {i}; }}\n\
3078                 // shared comment line {i}\n\
3079                 struct Foo {{ x: i32 }} // type {i}\n"
3080            )
3081            .repeat(5);
3082            let path = temp.join(format!("file_{i:04}.txt"));
3083            std::fs::write(&path, content.as_bytes()).expect("write");
3084        }
3085
3086        let mut config = crate::profile::balanced();
3087        // Force ZSTD for text so drops go through the dict-eligible path.
3088        config.defaults.text_codec = "zstd".into();
3089        // omnizip 0.14.40's Brotli encoder emits some streams the
3090        // in-house decoder rejects on highly repetitive input. Use ZSTD
3091        // for the metadata blob too so the round-trip parse succeeds.
3092        config.defaults.metadata_codec = "zstd".into();
3093        config.dictionaries.enabled = true;
3094        config.dictionaries.min_class_size = 50;
3095        config.dictionaries.max_dict_size = 8192;
3096
3097        let artifact = write_directory_with_config(&temp, &config).expect("write");
3098        std::fs::remove_dir_all(&temp).ok();
3099
3100        // The dictionary_section (if emitted) lives after the history
3101        // section. We don't strictly assert presence because the trainer
3102        // may legitimately return an empty dict; the test's job is to
3103        // verify the pipeline doesn't panic and the manifest parses.
3104        let mut cursor = ManifestCursor::new(&artifact.bytes);
3105        let _ = limnifs_core::parse_manifest_header(&mut cursor).expect("header");
3106        let _ = limnifs_core::parse_feature_flags_section(&mut cursor).expect("flags");
3107        let _ = limnifs_core::parse_metadata_reference(&mut cursor).expect("meta_ref");
3108        let _ = limnifs_core::parse_slab_index(&mut cursor).expect("slab_index");
3109        let _ = limnifs_core::parse_history(&mut cursor).expect("history");
3110        // If a dict was emitted, parsing past history should leave
3111        // non-empty remaining bytes.
3112        let _remaining = cursor.remaining_len();
3113    }
3114
3115    #[test]
3116    fn write_empty_directory() {
3117        let temp =
3118            std::env::temp_dir().join(format!("limnifs-write-test-{}-empty", std::process::id()));
3119        std::fs::create_dir_all(&temp).expect("create temp dir");
3120        let artifact = write_directory(&temp).expect("write succeeds");
3121        std::fs::remove_dir_all(&temp).ok();
3122        assert!(artifact.inode_count >= 1);
3123        assert_eq!(artifact.file_count, 0);
3124        assert_eq!(artifact.dir_count, 1);
3125        assert!(artifact.slabs.is_empty());
3126    }
3127
3128    #[test]
3129    fn write_small_file_inline() {
3130        let temp =
3131            std::env::temp_dir().join(format!("limnifs-write-test-{}-small", std::process::id()));
3132        std::fs::create_dir_all(&temp).expect("create temp dir");
3133        std::fs::write(temp.join("hello.txt"), b"hello world").expect("write file");
3134        let artifact = write_directory(&temp).expect("write succeeds");
3135        std::fs::remove_dir_all(&temp).ok();
3136        assert_eq!(artifact.file_count, 1);
3137        assert!(artifact.slabs.is_empty());
3138        assert_eq!(artifact.drop_count, 0);
3139    }
3140
3141    #[test]
3142    fn write_large_file_uses_slab() {
3143        let temp =
3144            std::env::temp_dir().join(format!("limnifs-write-test-{}-large", std::process::id()));
3145        std::fs::create_dir_all(&temp).expect("create temp dir");
3146        let large_data = vec![0xABu8; INLINE_THRESHOLD + 100];
3147        std::fs::write(temp.join("big.bin"), &large_data).expect("write big");
3148        let artifact = write_directory(&temp).expect("write succeeds");
3149        std::fs::remove_dir_all(&temp).ok();
3150        assert_eq!(artifact.drop_count, 1);
3151        assert_eq!(artifact.slabs.len(), 1);
3152    }
3153
3154    #[test]
3155    fn write_mixed_inline_and_large() {
3156        let temp =
3157            std::env::temp_dir().join(format!("limnifs-write-test-{}-mix", std::process::id()));
3158        std::fs::create_dir_all(&temp).expect("create temp dir");
3159        std::fs::write(temp.join("small.txt"), b"tiny").expect("write small");
3160        std::fs::write(temp.join("large.bin"), vec![0xCDu8; INLINE_THRESHOLD * 2])
3161            .expect("write large");
3162        let artifact = write_directory(&temp).expect("write succeeds");
3163        std::fs::remove_dir_all(&temp).ok();
3164        assert_eq!(artifact.file_count, 2);
3165        assert_eq!(artifact.drop_count, 1);
3166        assert_eq!(artifact.slabs.len(), 1);
3167    }
3168
3169    #[test]
3170    fn deduplicates_identical_large_files() {
3171        let temp =
3172            std::env::temp_dir().join(format!("limnifs-write-test-{}-dedup", std::process::id()));
3173        std::fs::create_dir_all(&temp).expect("create temp dir");
3174        let data = vec![0x77u8; INLINE_THRESHOLD + 10];
3175        std::fs::write(temp.join("a.bin"), &data).expect("write a");
3176        std::fs::write(temp.join("b.bin"), &data).expect("write b");
3177        let artifact = write_directory(&temp).expect("write succeeds");
3178        std::fs::remove_dir_all(&temp).ok();
3179        assert_eq!(artifact.drop_count, 1);
3180    }
3181
3182    #[test]
3183    fn write_and_verify_roundtrip() {
3184        let temp = std::env::temp_dir().join(format!(
3185            "limnifs-write-test-{}-roundtrip",
3186            std::process::id()
3187        ));
3188        std::fs::create_dir_all(&temp).expect("create temp dir");
3189        std::fs::write(temp.join("a.txt"), b"aaa").expect("write a");
3190        std::fs::write(temp.join("b.txt"), b"bbb").expect("write b");
3191        std::fs::create_dir_all(temp.join("sub")).expect("create sub");
3192        std::fs::write(temp.join("sub").join("c.txt"), b"ccc").expect("write c");
3193        let artifact = write_directory(&temp).expect("write succeeds");
3194        std::fs::remove_dir_all(&temp).ok();
3195        assert_eq!(artifact.file_count, 3);
3196        assert_eq!(artifact.dir_count, 2);
3197
3198        let mut cursor = ManifestCursor::new(&artifact.bytes);
3199        limnifs_core::parse_manifest_header(&mut cursor).expect("header");
3200        limnifs_core::parse_feature_flags_section(&mut cursor).expect("flags");
3201        let meta_ref = limnifs_core::parse_metadata_reference(&mut cursor).expect("meta ref");
3202        assert!(meta_ref.is_inlined());
3203        let slab_index = limnifs_core::parse_slab_index(&mut cursor).expect("slab index");
3204        assert_eq!(slab_index.len(), 0);
3205        limnifs_core::parse_history(&mut cursor).expect("history");
3206    }
3207
3208    #[test]
3209    fn write_deterministic() {
3210        let temp =
3211            std::env::temp_dir().join(format!("limnifs-write-test-{}-det", std::process::id()));
3212        std::fs::create_dir_all(&temp).expect("create temp dir");
3213        std::fs::write(temp.join("x.txt"), b"xxx").expect("write x");
3214
3215        let a1 = write_directory(&temp).expect("first write");
3216        let a2 = write_directory(&temp).expect("second write");
3217        std::fs::remove_dir_all(&temp).ok();
3218
3219        assert_eq!(a1.bytes, a2.bytes);
3220        assert_eq!(a1.merkle_root, a2.merkle_root);
3221    }
3222
3223    #[test]
3224    fn slab_parses_correctly() {
3225        let temp =
3226            std::env::temp_dir().join(format!("limnifs-write-test-{}-slab", std::process::id()));
3227        std::fs::create_dir_all(&temp).expect("create temp dir");
3228        std::fs::write(temp.join("big.bin"), vec![0x11u8; INLINE_THRESHOLD + 1])
3229            .expect("write big");
3230        let artifact = write_directory(&temp).expect("write succeeds");
3231        std::fs::remove_dir_all(&temp).ok();
3232
3233        let slab_bytes = &artifact.slabs[0].bytes;
3234        let mut cursor = ManifestCursor::new(slab_bytes);
3235        let slab_header = limnifs_core::parse_slab_header(&mut cursor).expect("slab header parses");
3236        assert_eq!(
3237            slab_header.format_version,
3238            limnifs_core::slab::SLAB_FORMAT_VERSION
3239        );
3240        assert!(!slab_header.is_sealed());
3241        assert!(!slab_header.has_erasure_coding());
3242
3243        let drop_record =
3244            limnifs_core::parse_drop_record(&mut cursor, &slab_header).expect("drop record parses");
3245        assert_eq!(drop_record.plaintext_len as usize, INLINE_THRESHOLD + 1);
3246    }
3247
3248    #[test]
3249    fn fastcdc_produces_multiple_chunks_for_large_files() {
3250        // A 1 MiB pseudo-random file should produce multiple drops
3251        // via FastCDC (default chunker uses 64 KiB min / 256 KiB avg).
3252        let temp = std::env::temp_dir().join(format!(
3253            "limnifs-write-test-{}-cdc-multi",
3254            std::process::id()
3255        ));
3256        std::fs::create_dir_all(&temp).expect("create temp dir");
3257        let data = pseudo_random_bytes(42, 1024 * 1024);
3258        std::fs::write(temp.join("big.bin"), &data).expect("write big");
3259        let artifact = write_directory(&temp).expect("write succeeds");
3260        std::fs::remove_dir_all(&temp).ok();
3261        assert!(
3262            artifact.drop_count > 1,
3263            "expected FastCDC to produce multiple drops for 1 MiB input, got {}",
3264            artifact.drop_count
3265        );
3266    }
3267
3268    #[test]
3269    fn fastcdc_deduplicates_shared_substrings() {
3270        // Two files sharing a long middle section should produce
3271        // fewer drops than the sum of their individual chunk counts,
3272        // because the shared section's chunks deduplicate.
3273        let temp = std::env::temp_dir().join(format!(
3274            "limnifs-write-test-{}-cdc-dedup",
3275            std::process::id()
3276        ));
3277        std::fs::create_dir_all(&temp).expect("create temp dir");
3278        let shared = pseudo_random_bytes(7, 512 * 1024);
3279        let mut a = Vec::with_capacity(shared.len() + 1024);
3280        a.extend_from_slice(&pseudo_random_bytes(1, 1024));
3281        a.extend_from_slice(&shared);
3282        let mut b = Vec::with_capacity(shared.len() + 2048);
3283        b.extend_from_slice(&pseudo_random_bytes(2, 2048));
3284        b.extend_from_slice(&shared);
3285        std::fs::write(temp.join("a.bin"), &a).expect("write a");
3286        std::fs::write(temp.join("b.bin"), &b).expect("write b");
3287
3288        // Baseline: each file alone.
3289        let temp_a = std::env::temp_dir().join(format!(
3290            "limnifs-write-test-{}-cdc-dedup-a",
3291            std::process::id()
3292        ));
3293        std::fs::create_dir_all(&temp_a).expect("create temp_a");
3294        std::fs::write(temp_a.join("a.bin"), &a).expect("write a");
3295        let artifact_a = write_directory(&temp_a).expect("a writes");
3296        std::fs::remove_dir_all(&temp_a).ok();
3297
3298        let temp_b = std::env::temp_dir().join(format!(
3299            "limnifs-write-test-{}-cdc-dedup-b",
3300            std::process::id()
3301        ));
3302        std::fs::create_dir_all(&temp_b).expect("create temp_b");
3303        std::fs::write(temp_b.join("b.bin"), &b).expect("write b");
3304        let artifact_b = write_directory(&temp_b).expect("b writes");
3305        std::fs::remove_dir_all(&temp_b).ok();
3306
3307        let artifact_both = write_directory(&temp).expect("both write");
3308        std::fs::remove_dir_all(&temp).ok();
3309
3310        let sum_alone = artifact_a.drop_count + artifact_b.drop_count;
3311        assert!(
3312            artifact_both.drop_count < sum_alone,
3313            "expected dedup win: both together = {} drops, sum alone = {} drops",
3314            artifact_both.drop_count,
3315            sum_alone
3316        );
3317    }
3318
3319    #[test]
3320    fn slab_splits_when_content_exceeds_ceiling() {
3321        // Synthesise enough incompressible drops to force at least two
3322        // slabs. Each drop is 10 MiB of pseudo-random data; three drops
3323        // = 30 MiB compressed (random data doesn't compress), which
3324        // fits in one slab. We bump to seven drops (70 MiB) to force a
3325        // split.
3326        let temp =
3327            std::env::temp_dir().join(format!("limnifs-write-test-{}-split", std::process::id()));
3328        std::fs::create_dir_all(&temp).expect("create temp dir");
3329        for i in 0..7u32 {
3330            // 10 MiB of pseudo-random bytes — incompressible.
3331            let data = pseudo_random_bytes(u64::from(i), 10 * 1024 * 1024);
3332            std::fs::write(temp.join(format!("big-{i}.bin")), &data).expect("write big");
3333        }
3334        let artifact = write_directory(&temp).expect("write succeeds");
3335        std::fs::remove_dir_all(&temp).ok();
3336
3337        // Each slab's total length must respect MAX_SLAB_TOTAL_BYTES.
3338        assert!(
3339            artifact.slabs.len() >= 2,
3340            "expected at least 2 slabs for 70 MiB of incompressible data, got {}",
3341            artifact.slabs.len()
3342        );
3343        for slab in &artifact.slabs {
3344            assert!(
3345                slab.bytes.len() <= MAX_SLAB_TOTAL_BYTES,
3346                "slab {} is {} bytes (> {} ceiling)",
3347                slab.id.ordinal,
3348                slab.bytes.len(),
3349                MAX_SLAB_TOTAL_BYTES,
3350            );
3351        }
3352        // All seven drops must be accounted for across slabs.
3353        let total_drop_ids: usize = artifact.slabs.iter().map(|s| s.drop_ids.len()).sum();
3354        assert_eq!(
3355            total_drop_ids, artifact.drop_count,
3356            "drop_ids count across slabs must match WriteArtifact.drop_count",
3357        );
3358    }
3359}