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