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