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;
30pub mod flatten;
31#[cfg(feature = "pipeline-parallelism")]
32pub mod pipeline;
33pub mod rw;
34#[cfg(feature = "sparse-index")]
35pub mod sparse_index;
36pub mod turnover;
37
38pub use config::{
39    profile, CategorizerConfig, ChunkingConfig, CodecRegistry, CodecTunables, Defaults,
40    DictionaryConfig, EncryptionConfig, TournamentConfig, WriteConfig,
41};
42
43use std::collections::{HashMap, HashSet};
44use std::path::{Path, PathBuf};
45
46use crate::chunker::FastCDC;
47use limnifs_core::codec::CODEC_REFERENCED;
48use limnifs_core::slab_store::SlabStore;
49use limnifs_core::{
50    compute_merkle_root, hash_empty_section, hash_section, parse_manifest_header, parse_slab_index,
51    ManifestCursor, ManifestHeader, SectionHashes, FEATURE_FLAGS_SECTION_VERSION,
52    HISTORY_SECTION_VERSION, INODE_FLAG_INLINE_DATA, INODE_FLAG_SHARED_INLINE,
53    METADATA_REFERENCE_SECTION_VERSION_2, SLAB_INDEX_SECTION_VERSION,
54};
55use limnifs_format::{ManifestRoot, SlabId};
56
57/// Inline-data threshold: files at or below this size get inline data
58/// in their inode. Larger files are stored as drops in a slab.
59pub const INLINE_THRESHOLD: usize = 4096;
60
61/// Above this size, mmap the input file instead of `std::fs::read`-ing
62/// it into a `Vec<u8>`. Keeps peak RSS bounded when packing huge files
63/// (multi-GiB source trees, ML models). Crossover is around 1 MiB on
64/// most filesystems — below that the syscall + VMA setup costs more
65/// than the read.
66pub const MMAP_READ_THRESHOLD: usize = 1024 * 1024;
67
68/// Maximum file size for the whole-file categorizer path. Files above
69/// this threshold use FastCDC chunking even when a categorizer claims
70/// them, enabling rayon parallelism across chunks. The categorizer's
71/// codec is still used per-chunk when possible.
72pub const WHOLE_FILE_MAX_SIZE: usize = 64 * 1024 * 1024;
73
74/// Maximum total length of a single slab file (header + content).
75/// Matches the reader's `DEFAULT_SLAB_MAX_BYTES` (spec §3.1) minus a
76/// safety margin so a slab that is full but not yet flushed cannot
77/// overrun the reader ceiling on the next drop.
78pub const MAX_SLAB_TOTAL_BYTES: usize = 60 * 1024 * 1024;
79
80/// Width of the slab header (magic + version + `SlabId` + `total_length` +
81/// `ec_descriptor` + `crypto_hint`). Must agree with
82/// `limnifs_core::slab::SLAB_HEADER_LEN`.
83const SLAB_HEADER_LEN: usize = 56;
84
85/// Default threshold at which the writer externalises the metadata
86/// blob to a sidecar file instead of inlining it in the manifest.
87/// Derived from the reader's inline ceiling
88/// (`limnifs_core::metadata_reference::DEFAULT_INLINE_METADATA_MAX_BYTES`,
89/// 1 MiB per spec §5.3) minus headroom, so the two constants cannot
90/// silently drift apart. Override per image via
91/// `WriteConfig::defaults::metadata_externalize_threshold` (issue #187).
92pub const METADATA_EXTERNALIZE_THRESHOLD: usize =
93    limnifs_core::metadata_reference::DEFAULT_INLINE_METADATA_MAX_BYTES as usize - 24 * 1024;
94
95/// Metadata-blob size above which the writer steps Brotli quality
96/// down to `METADATA_LARGE_BLOB_QUALITY`. Below this, q5's cost is
97/// negligible; above it, q5 starts to dominate create time on big
98/// inode trees (e.g. the 50 K-file tiny-files dataset).
99pub const METADATA_LARGE_BLOB_THRESHOLD: usize = 256 * 1024;
100
101/// Brotli quality for small metadata blobs (≤ `METADATA_LARGE_BLOB_THRESHOLD`).
102/// Best ratio; cost is in the noise on small inputs.
103pub const METADATA_SMALL_BLOB_QUALITY: i32 = 5;
104
105/// Brotli quality for large metadata blobs. q2 is much faster than q5
106/// on multi-MiB inputs; ratio on highly compressible inode data is
107/// within 5–10% of q5 (often identical) because metadata is dominated
108/// by long runs of repeated patterns.
109pub const METADATA_LARGE_BLOB_QUALITY: i32 = 2;
110
111/// One slab produced by the writer. The slab ordinal in `id` matches
112/// the slab's position in `WriteArtifact::slabs`.
113#[derive(Clone, Debug)]
114pub struct SlabArtifact {
115    pub id: SlabId,
116    pub bytes: Vec<u8>,
117    pub locator: String,
118    /// `DropIds` contained in this slab, in slab order. Used by callers
119    /// that need to know which slab holds which drop without re-parsing
120    /// the slab bytes.
121    pub drop_ids: Vec<[u8; 32]>,
122}
123
124/// Externalized metadata sidecar, present when the metadata blob
125/// exceeds [`METADATA_EXTERNALIZE_THRESHOLD`]. Callers must write
126/// `bytes` to `locator` next to the manifest file.
127#[derive(Clone, Debug)]
128pub struct MetadataSidecar {
129    pub bytes: Vec<u8>,
130    pub locator: String,
131}
132
133/// Result of writing a directory tree.
134#[derive(Clone, Debug)]
135pub struct WriteArtifact {
136    pub bytes: Vec<u8>,
137    pub merkle_root: ManifestRoot,
138    /// All slabs produced by the writer, in slab-ordinal order. Empty
139    /// when the source tree had no files > [`INLINE_THRESHOLD`].
140    pub slabs: Vec<SlabArtifact>,
141    /// External metadata sidecar, present when the metadata blob
142    /// exceeds [`METADATA_EXTERNALIZE_THRESHOLD`]. `None` means the
143    /// metadata is inlined in the manifest.
144    pub metadata_sidecar: Option<MetadataSidecar>,
145    pub inode_count: usize,
146    pub file_count: usize,
147    pub dir_count: usize,
148    pub drop_count: usize,
149    /// Inode number of the root directory (i.e. the inode that
150    /// represents the source directory itself, not a child of it).
151    /// Always a directory and always referenced by the inlined
152    /// metadata blob's directory inode table.
153    pub root_inode_number: u64,
154}
155
156impl WriteArtifact {
157    /// Convenience accessor for the single-slab case. Returns the
158    /// first slab's bytes if there is exactly one slab, else `None`.
159    /// Modern callers should iterate [`WriteArtifact::slabs`] directly.
160    #[must_use]
161    pub fn slab_bytes(&self) -> Option<&[u8]> {
162        if self.slabs.len() == 1 {
163            Some(&self.slabs[0].bytes)
164        } else {
165            None
166        }
167    }
168
169    /// Convenience accessor for the single-slab case.
170    #[must_use]
171    pub fn slab_locator(&self) -> Option<&str> {
172        if self.slabs.len() == 1 {
173            Some(&self.slabs[0].locator)
174        } else {
175            None
176        }
177    }
178}
179
180/// Error during writing.
181#[derive(Debug)]
182pub enum WriteError {
183    Io(std::io::Error),
184    /// The tree contains an entry type the writer deliberately does
185    /// not store (sockets, FIFOs, device nodes). Symlinks ARE
186    /// supported; everything else on a normal filesystem tree is
187    /// either a file, a directory, or this error.
188    UnsupportedFileType {
189        path: PathBuf,
190        kind: String,
191    },
192}
193
194impl std::fmt::Display for WriteError {
195    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
196        match self {
197            Self::Io(e) => write!(f, "I/O error: {e}"),
198            Self::UnsupportedFileType { path, kind } => write!(
199                f,
200                "unsupported file type ({kind}): {} — limnifs stores files, \
201                 directories, and symlinks; remove the entry or file an issue \
202                 if you need it carried",
203                path.display()
204            ),
205        }
206    }
207}
208
209impl std::error::Error for WriteError {}
210
211impl From<std::io::Error> for WriteError {
212    fn from(e: std::io::Error) -> Self {
213        Self::Io(e)
214    }
215}
216
217/// Walk a directory tree and produce a valid `.lim` manifest artifact
218/// with inlined metadata. Files at or below [`INLINE_THRESHOLD`] bytes
219/// are stored inline; larger files are packed into a single slab as
220/// content-addressed drops.
221///
222/// File contents (read, `FastCDC` chunk, `BLAKE3` hash, `LZ4` compress) are
223/// processed in parallel across `CPU` cores via `rayon`. The directory
224/// tree walk and slab assembly remain sequential so the output is
225/// deterministic.
226///
227/// # Errors
228///
229/// Returns [`WriteError::Io`] for filesystem errors.
230pub fn write_directory(root: &Path) -> Result<WriteArtifact, WriteError> {
231    write_directory_with_config(root, &WriteConfig::default_v0_1())
232}
233
234/// Pack a single named stream into a `.lim` image.
235///
236/// For callers that pipe data from a network socket, pipe, or generator
237/// and don't want to materialise the full content on disk before
238/// packing. The reader is consumed via [`FastCDC::chunk_reader`] which
239/// bounds internal buffering at `max_chunk_size + 64 KiB`.
240///
241/// The resulting image has a single root file with the given `name`
242/// (path-relative; safe to use `/` for subdirectories — they're
243/// materialised in the metadata tree).
244///
245/// # Errors
246///
247/// Returns [`WriteError::Io`] on read failure or any writer-pipeline
248/// error.
249pub fn write_stream<R: std::io::Read>(
250    name: &str,
251    reader: R,
252    config: &WriteConfig,
253) -> Result<WriteArtifact, WriteError> {
254    let mut ctx = WriteContext::new();
255    ctx.categorizers_disabled = config.categorizers.is_empty();
256    ctx.rw_mode = matches!(config.mode, crate::config::ImageMode::ReadWrite(_));
257    ctx.auto_turnover = config.turnover_threshold > 0;
258    ctx.collect_dict_samples = config.dictionaries.enabled;
259
260    // Synthesise a single PendingFile that points to nothing on disk;
261    // we'll bypass process_file's `std::fs::read` and feed the
262    // pre-chunked bytes directly.
263    let drop_id_root = [0u8; 32]; // placeholder; replaced below
264    let pending = PendingFile {
265        path: std::path::PathBuf::from(name),
266        inode_number: 1,
267        file_len: 0, // patched below once we know the total
268        mtime_ns: 0,
269    };
270    ctx.pending_files.push(pending);
271    ctx.root_inode_number = 1;
272
273    // Chunk the stream directly via FastCDC's chunk_reader.
274    let chunker = ctx.chunker.clone();
275    let chunks = chunker.chunk_reader(reader)?;
276
277    // Total size = sum of chunk lengths.
278    let total_len: u64 = chunks.iter().map(|c| c.len() as u64).sum();
279
280    // Hash + compress each chunk. We treat each chunk as a unique drop
281    // (the stream is single-pass; cross-call dedup is left to the
282    // caller). Use the configured tournament spec.
283    let text_codec = config.text_codec_id().unwrap_or(0x04);
284    let binary_codec = config.binary_codec_id().unwrap_or(0x01);
285    let tunables = config.to_core_tunables();
286    let classifier = ctx.classifier;
287    let registry = config
288        .codec_registry()
289        .map_err(|e| WriteError::Io(std::io::Error::other(format!("codec registry: {e}"))))?;
290    let tournament_codec_ids: Vec<u8> = config
291        .tournament
292        .codecs
293        .iter()
294        .filter_map(|n| registry.lookup_by_name(n))
295        .collect();
296    let tournament = TournamentSpec {
297        codec_ids: tournament_codec_ids,
298        min_size: config.tournament.min_size_threshold as usize,
299        skip_for_binary: config.tournament.skip_for_binary,
300        short_circuit_permille: config.tournament.short_circuit_threshold,
301    };
302
303    let mut drops: Vec<RawDrop> = Vec::with_capacity(chunks.len());
304    let mut slices: Vec<PendingSlice> = Vec::with_capacity(chunks.len());
305    let mut offset: u64 = 0;
306    for chunk in &chunks {
307        let drop_id = hash_section(chunk);
308        slices.push(PendingSlice {
309            drop_id,
310            file_byte_start: offset,
311            file_byte_end: offset + chunk.len() as u64,
312        });
313        offset += chunk.len() as u64;
314        let class = classifier.classify(chunk);
315        let (codec_id, compressed) = compress_chunk_with_tournament(
316            chunk,
317            class,
318            text_codec,
319            binary_codec,
320            &tunables,
321            &tournament,
322        );
323        drops.push((drop_id, chunk.clone(), compressed, codec_id));
324    }
325    let _ = drop_id_root;
326
327    // Wire into WriteContext as a single-file result + inode.
328    let result = ChunkedFileResult { drops, slices };
329    let pf = ctx.pending_files[0].clone();
330    ctx.merge_chunked_file(&pf, result);
331    // Patch the file_len now that we know it.
332    ctx.pending_files[0].file_len = total_len;
333    // The inode was already pushed by merge_chunked_file with the old
334    // (zero) file_len; correct it.
335    if let Some(inode) = ctx.inodes.last_mut() {
336        if let PendingContent::DropBacked { file_len, .. } = &mut inode.content {
337            *file_len = total_len;
338        }
339    }
340
341    ctx.train_and_apply_dictionary(&config.dictionaries);
342    let artifact = ctx.assemble();
343    Ok(artifact)
344}
345
346/// Pack a directory tree as a **layer** on top of a base image.
347///
348/// Produces a `.lim` image whose drops are split into two sets:
349///
350/// - **Local drops** — chunks in `root` whose `DropId` is NOT in
351///   `base_image`'s drop set. These are compressed and stored in the
352///   layer's own slabs exactly as `write_directory_with_config`
353///   would store them.
354/// - **Referenced drops** — chunks whose `DropId` IS in the base.
355///   These are recorded only as `PendingSlice` references (so the
356///   metadata tree links them in); no slab bytes are emitted in the
357///   layer. The reader resolves them via the overlay chain.
358///
359/// The resulting manifest carries a `delta_linkage` section pointing
360/// at the base image's `ManifestRoot`, so any reader that supports
361/// overlay chains can extract the layer standalone or stacked on the
362/// base.
363///
364/// # Determinism
365///
366/// `write_layer` is deterministic given the same `base_image`, the
367/// same `root` content, and the same `config`. Two runs produce
368/// byte-identical layer images.
369///
370/// # Errors
371///
372/// Returns [`WriteError::Io`] on read failure or any writer-pipeline
373/// error.
374///
375/// # Example
376///
377/// ```no_run
378/// use limnifs_write::{write_layer, profile};
379///
380/// let base = std::path::Path::new("base.lim");
381/// let root = std::path::Path::new("./new-content");
382/// let cfg = profile::balanced();
383/// let artifact = write_layer(base, root, &cfg).expect("layer");
384/// // artifact.bytes is the layer manifest; slabs contain only NEW drops.
385/// ```
386pub fn write_layer(
387    base_image: &Path,
388    root: &Path,
389    config: &WriteConfig,
390) -> Result<WriteArtifact, WriteError> {
391    // Load the base image's drop set + manifest root.
392    let (base_drop_index, base_root) = load_base_drop_index(base_image)?;
393
394    let mut ctx = WriteContext::new();
395    ctx.categorizers_disabled = config.categorizers.is_empty();
396    ctx.rw_mode = matches!(config.mode, crate::config::ImageMode::ReadWrite(_));
397    ctx.auto_turnover = config.turnover_threshold > 0;
398    ctx.collect_dict_samples = config.dictionaries.enabled;
399    ctx.inline_threshold = config.defaults.inline_threshold as usize;
400    ctx.metadata_externalize_threshold = config.defaults.metadata_externalize_threshold;
401    ctx.emit_shared_inline = config.defaults.shared_inline;
402    ctx.base_drop_index = Some(base_drop_index);
403    ctx.base_root = Some(base_root);
404
405    // Rest is identical to write_directory_with_config.
406    let root_inode_number = ctx.walk(root)?;
407    ctx.root_inode_number = root_inode_number;
408    write_directory_body(&mut ctx, config)?;
409    Ok(ctx.assemble())
410}
411
412/// Load every DropId present in a base image's slabs + the image's
413/// `ManifestRoot`. Used by `write_layer` to decide which chunks can
414/// be referenced rather than re-encoded.
415fn load_base_drop_index(
416    base_image: &Path,
417) -> Result<(std::collections::HashSet<[u8; 32]>, [u8; 32]), WriteError> {
418    let manifest_bytes = std::fs::read(base_image)?;
419    let mut cursor = ManifestCursor::new(&manifest_bytes);
420    let _ = parse_manifest_header(&mut cursor).map_err(io_core)?;
421    // Walk sections in spec order: flags → metadata_reference → slab_index.
422    let _ = limnifs_core::parse_feature_flags_section(&mut cursor);
423    let _ = limnifs_core::parse_metadata_reference(&mut cursor);
424    let slab_index = parse_slab_index(&mut cursor).map_err(io_core)?;
425    let store = SlabStore::load_mmap(base_image, &slab_index).map_err(io_core)?;
426    let drop_set: std::collections::HashSet<[u8; 32]> = store.drop_index_keys().copied().collect();
427    // Re-derive the Merkle root from the base manifest's section
428    // bytes. The base's `ManifestRoot` is the canonical anchor for
429    // the layer's `delta_linkage.base_root` field — round-tripping
430    // through section hashes guarantees it matches what the base
431    // reported on its own assemble path.
432    let root = *compute_merkle_root_from_sections(&manifest_bytes).as_bytes();
433    Ok((drop_set, root))
434}
435
436/// Re-derive a manifest's `ManifestRoot` from its on-disk section
437/// bytes. Mirrors `flatten::compute_merkle_root_from_sections` but
438/// tolerates absent optional sections (returns hash_empty_section()
439/// for them). Used by `load_base_drop_index` to anchor a layer's
440/// `base_root` without re-instantiating a full manifest parser.
441fn compute_merkle_root_from_sections(manifest: &[u8]) -> ManifestRoot {
442    use limnifs_core::SectionHashes;
443    let mut cursor = ManifestCursor::new(manifest);
444    let header_start = 0;
445    if parse_manifest_header(&mut cursor).is_err() {
446        // Not a valid manifest; fall back to all-zero root.
447        return ManifestRoot::from_bytes([0u8; 32]);
448    }
449    let header_end = cursor.position();
450    // Optional sections — best-effort parse; failures hash as empty.
451    let flags_start = header_end;
452    let flags_end = match limnifs_core::parse_feature_flags_section(&mut cursor) {
453        Ok(_) => cursor.position(),
454        Err(_) => flags_start,
455    };
456    let meta_ref_start = flags_end;
457    let metadata_reference = match limnifs_core::parse_metadata_reference(&mut cursor) {
458        Ok(m) => Some(m),
459        Err(_) => None,
460    };
461    let meta_ref_end = cursor.position();
462    let slab_index_start = meta_ref_end;
463    let _ = parse_slab_index(&mut cursor);
464    let slab_index_end = cursor.position();
465    let history_start = slab_index_end;
466    let _ = limnifs_core::parse_history(&mut cursor);
467    let history_end = cursor.position();
468
469    let hashes = SectionHashes {
470        metadata: metadata_reference
471            .map(|m| m.metadata_hash)
472            .unwrap_or_else(hash_empty_section),
473        format_header: hash_section(&manifest[header_start..header_end]),
474        feature_flags: hash_section(&manifest[flags_start..flags_end]),
475        metadata_reference: hash_section(&manifest[meta_ref_start..meta_ref_end]),
476        slab_index: hash_section(&manifest[slab_index_start..slab_index_end]),
477        crypto_params: hash_empty_section(),
478        ec_params: hash_empty_section(),
479        dms_policy: hash_empty_section(),
480        delta_linkage: hash_empty_section(),
481        history: hash_section(&manifest[history_start..history_end]),
482    };
483    compute_merkle_root(&hashes)
484}
485
486fn io_core(e: limnifs_core::CoreError) -> WriteError {
487    WriteError::Io(std::io::Error::other(format!("base image load: {e}")))
488}
489
490/// Shared body between `write_directory_with_config` and `write_layer`.
491/// Walks `ctx.pending_files` through `process_file` in parallel and
492/// merges the results back. Caller is responsible for `walk()` and
493/// `assemble()`.
494fn write_directory_body(ctx: &mut WriteContext, config: &WriteConfig) -> Result<(), WriteError> {
495    use rayon::prelude::*;
496
497    ctx.metadata_codec = config
498        .metadata_codec_id()
499        .unwrap_or(limnifs_core::codec::CODEC_BROTLI);
500
501    let pending = std::mem::take(&mut ctx.pending_files);
502    if pending.is_empty() {
503        return Ok(());
504    }
505    ctx.inline_threshold = config.defaults.inline_threshold as usize;
506    ctx.metadata_externalize_threshold = config.defaults.metadata_externalize_threshold;
507    ctx.emit_shared_inline = config.defaults.shared_inline;
508    let chunker = ctx.chunker.clone();
509    let classifier = ctx.classifier;
510    let text_codec = config.text_codec_id().unwrap_or(0x04);
511    let binary_codec = config.binary_codec_id().unwrap_or(0x01);
512    let tunables = config.to_core_tunables();
513    let use_categorizers = !config.categorizers.is_empty();
514    let skip_chunking = config.skip_chunking;
515    let registry = config
516        .codec_registry()
517        .map_err(|e| WriteError::Io(std::io::Error::other(format!("codec registry: {e}"))))?;
518    let tournament_codec_ids: Vec<u8> = config
519        .tournament
520        .codecs
521        .iter()
522        .filter_map(|n| registry.lookup_by_name(n))
523        .collect();
524    let tournament_spec = TournamentSpec {
525        codec_ids: tournament_codec_ids,
526        min_size: config.tournament.min_size_threshold as usize,
527        skip_for_binary: config.tournament.skip_for_binary,
528        short_circuit_permille: config.tournament.short_circuit_threshold,
529    };
530    let base_drop_index = ctx.base_drop_index.as_ref();
531    let inline_threshold = ctx.inline_threshold;
532    let results: Vec<ChunkedFileResult> = pending
533        .par_iter()
534        .map(|pf| {
535            process_file(
536                pf,
537                &chunker,
538                classifier,
539                text_codec,
540                binary_codec,
541                &tunables,
542                use_categorizers,
543                skip_chunking,
544                &tournament_spec,
545                base_drop_index,
546                inline_threshold,
547            )
548        })
549        .collect::<Result<Vec<_>, _>>()?;
550
551    for (pf, result) in pending.iter().zip(results) {
552        ctx.merge_chunked_file(pf, result);
553    }
554    ctx.train_and_apply_dictionary(&config.dictionaries);
555    Ok(())
556}
557
558/// Create an image with a custom [`WriteConfig`] (e.g. from a profile).
559pub fn write_directory_with_config(
560    root: &Path,
561    config: &WriteConfig,
562) -> Result<WriteArtifact, WriteError> {
563    let mut ctx = WriteContext::new();
564    ctx.categorizers_disabled = config.categorizers.is_empty();
565    ctx.rw_mode = matches!(config.mode, crate::config::ImageMode::ReadWrite(_));
566    ctx.auto_turnover = config.turnover_threshold > 0;
567    ctx.collect_dict_samples = config.dictionaries.enabled;
568
569    write_directory_streaming(&mut ctx, root, config)?;
570    Ok(ctx.assemble())
571}
572
573/// Walk + compress with producer/consumer overlap (TODO.perf/15).
574///
575/// The tree walk runs on a scoped producer thread and forwards each
576/// deferred file to a bounded channel; rayon workers (via `par_bridge`)
577/// compress while the walk is still descending. For warm-cache trees
578/// with few files this is equivalent to the collect-then-dispatch
579/// shape; for huge or cold-cache trees it hides walk latency behind
580/// compression.
581///
582/// **Determinism:** results are re-sequenced into walk order before
583/// merging, and inode allocation / dir-node construction are
584/// untouched, so the emitted bytes are identical to
585/// `write_directory_body` for the same input.
586fn write_directory_streaming(
587    ctx: &mut WriteContext,
588    root: &Path,
589    config: &WriteConfig,
590) -> Result<(), WriteError> {
591    use rayon::prelude::*;
592
593    ctx.metadata_codec = config
594        .metadata_codec_id()
595        .unwrap_or(limnifs_core::codec::CODEC_BROTLI);
596
597    let chunker = ctx.chunker.clone();
598    let classifier = ctx.classifier;
599    let text_codec = config.text_codec_id().unwrap_or(0x04);
600    let binary_codec = config.binary_codec_id().unwrap_or(0x01);
601    let tunables = config.to_core_tunables();
602    let use_categorizers = !config.categorizers.is_empty();
603    let skip_chunking = config.skip_chunking;
604    let registry = config
605        .codec_registry()
606        .map_err(|e| WriteError::Io(std::io::Error::other(format!("codec registry: {e}"))))?;
607    let tournament_codec_ids: Vec<u8> = config
608        .tournament
609        .codecs
610        .iter()
611        .filter_map(|n| registry.lookup_by_name(n))
612        .collect();
613    let tournament_spec = TournamentSpec {
614        codec_ids: tournament_codec_ids,
615        min_size: config.tournament.min_size_threshold as usize,
616        skip_for_binary: config.tournament.skip_for_binary,
617        short_circuit_permille: config.tournament.short_circuit_threshold,
618    };
619    // The producer thread owns `&mut ctx` for the duration of the
620    // walk, so the layer fast-path index travels as a clone.
621    let base_drop_index = ctx.base_drop_index.clone();
622    let inline_threshold = ctx.inline_threshold;
623
624    ctx.inline_threshold = config.defaults.inline_threshold as usize;
625    ctx.metadata_externalize_threshold = config.defaults.metadata_externalize_threshold;
626    ctx.emit_shared_inline = config.defaults.shared_inline;
627
628    // Bounded so the walk back-pressures if compression falls behind;
629    // the buffer is large enough to keep every worker fed on bursty
630    // directory layouts.
631    const PIPELINE_CAPACITY: usize = 256;
632    let (tx, rx) = std::sync::mpsc::sync_channel::<PendingFile>(PIPELINE_CAPACITY);
633    ctx.pending_sink = Some(tx);
634
635    let (root_inode_number, mut results): (
636        u64,
637        Vec<(usize, PendingFile, Result<ChunkedFileResult, WriteError>)>,
638    ) = std::thread::scope(|scope| {
639        let producer = {
640            let ctx = &mut *ctx;
641            let root = root;
642            scope.spawn(move || {
643                let r = ctx.walk(root);
644                // Disconnect the channel so the consumer's iterator
645                // terminates; the sink stays None until the next
646                // streaming write resets it after the scope.
647                ctx.pending_sink = None;
648                r
649            })
650        };
651        // par_bridge does not preserve order; carry the arrival index
652        // and re-sequence before merging.
653        let results = rx
654            .into_iter()
655            .enumerate()
656            .par_bridge()
657            .map(|(i, pf)| {
658                let r = process_file(
659                    &pf,
660                    &chunker,
661                    classifier,
662                    text_codec,
663                    binary_codec,
664                    &tunables,
665                    use_categorizers,
666                    skip_chunking,
667                    &tournament_spec,
668                    base_drop_index.as_ref(),
669                    inline_threshold,
670                );
671                (i, pf, r)
672            })
673            .collect();
674        let joined = producer
675            .join()
676            .unwrap_or_else(|_| {
677                Err(WriteError::Io(std::io::Error::other(
678                    "walk thread panicked",
679                )))
680            })
681            .map(|n| (n, results));
682        // Scope can't `?` across borrows of `results`; return the
683        // outcome and propagate outside.
684        joined
685    })?;
686    ctx.pending_sink = None;
687    ctx.root_inode_number = root_inode_number;
688
689    results.sort_unstable_by_key(|(i, _, _)| *i);
690    // Fail on the lowest walk index first, matching the
691    // collect::<Result<Vec<_>, _>> abort semantics of the
692    // collect-then-dispatch shape.
693    for (_, pf, r) in results {
694        ctx.merge_chunked_file(&pf, r?);
695    }
696    ctx.train_and_apply_dictionary(&config.dictionaries);
697    Ok(())
698}
699
700/// One chunk of a file before dedup: (`drop_id`, `plaintext`, `compressed`, `codec`).
701///
702/// `compressed` is `Arc<[u8]>` so the cross-file compress cache can
703/// share bytes across hits with a refcount bump instead of a deep
704/// copy — dedup-heavy workloads (container layers, duplicate files)
705/// skip the allocation entirely.
706pub(crate) type RawDrop = ([u8; 32], Vec<u8>, std::sync::Arc<[u8]>, u8);
707/// Result of parallel file processing: the drop data (uncompressed,
708pub(crate) struct ChunkedFileResult {
709    drops: Vec<RawDrop>, // (id, plaintext, compressed, codec)
710    slices: Vec<PendingSlice>,
711}
712
713/// Resolved tournament configuration passed to per-chunk compression.
714///
715/// Built once at the top level from `WriteConfig::tournament` so each
716/// rayon worker reuses the same Vec rather than rebuilding it per
717/// file. Codecs are stored as numeric ids (looked up via
718/// `WriteConfig::codec_registry`) — `process_file` never sees the
719/// string form.
720struct TournamentSpec {
721    /// Codec ids to try, in declared order. `process_file` iterates
722    /// these and tracks the best compression. `CODEC_STORE` entries
723    /// are ignored (store is always the implicit fallback).
724    codec_ids: Vec<u8>,
725    /// Chunks below this many bytes get the preferred codec only
726    /// (no tournament) — the per-codec setup cost dominates at small
727    /// sizes and the ratio difference is negligible.
728    min_size: usize,
729    /// When true, binary-classified chunks skip the tournament and
730    /// use `binary_codec` directly. Matches the v0.1 behaviour where
731    /// binary chunks were never worth the tournament cost.
732    skip_for_binary: bool,
733    /// Short-circuit threshold in per-mille (0..=1000). 0 disables
734    /// short-circuit. Whenever a codec achieves compression ratio
735    /// ≤ threshold, the tournament accepts it and skips any slower
736    /// codecs later in the list.
737    short_circuit_permille: u32,
738}
739
740/// Compress a whole file as a single drop using the categorizer's
741/// chosen codec. Used when a file-level categorizer claims the file
742/// (FLAC for WAV, ricepp for FITS, FSST+Brotli for CSV). The drop's
743/// slice covers the whole file; no `FastCDC` chunking happens.
744///
745/// Codec parameters extracted by the categorizer (e.g. PCM sample
746/// format, FITS bitpix) are NOT prepended to the compressed bytes —
747/// the codec embeds its own params in its container format. The
748/// `LimniFS` drop record just stores `(codec_id, compressed_bytes)`
749/// and lets the codec own its param encoding. The categorizer's
750/// `codec_params` field is reserved for future use when a codec
751/// needs params NOT embedded in its container.
752fn process_whole_file_drop(
753    pf: &PendingFile,
754    data: &[u8],
755    cat: file_categorizer::Categorization,
756    tunables: &limnifs_core::codec::CodecTunables,
757) -> Result<ChunkedFileResult, WriteError> {
758    let _ = pf;
759    let drop_id = hash_section(data);
760    let file_len = u64::try_from(data.len()).unwrap_or(u64::MAX);
761
762    // Brotli first; if it fails (including a codec panic — the
763    // registry converts panics to Err), fall back to ZSTD, then to
764    // STORE. A broken encoder must degrade the drop's ratio, never
765    // the write itself.
766    let (mut best_codec, mut best_compressed): (u8, std::sync::Arc<[u8]>) =
767        match limnifs_core::codec::compress_with_tunables(
768            limnifs_core::codec::CODEC_BROTLI,
769            data,
770            tunables,
771        ) {
772            Ok(c) => (limnifs_core::codec::CODEC_BROTLI, c.into()),
773            Err(_) => match limnifs_core::codec::compress_with_tunables(
774                limnifs_core::codec::CODEC_ZSTD,
775                data,
776                tunables,
777            ) {
778                Ok(c) => (limnifs_core::codec::CODEC_ZSTD, c.into()),
779                Err(_) => (limnifs_core::codec::CODEC_STORE, data.to_vec().into()),
780            },
781        };
782
783    // Short-circuit: if Brotli already achieves < 5% ratio, the input
784    // is highly compressible and ZSTD is unlikely to beat it by enough
785    // to justify the extra pass. Skip ZSTD on this fast path.
786    let brotli_ratio = best_compressed.len() as f64 / data.len() as f64;
787    if brotli_ratio > 0.05 {
788        if let Ok(zstd_c) = limnifs_core::codec::compress_with_tunables(
789            limnifs_core::codec::CODEC_ZSTD,
790            data,
791            tunables,
792        ) {
793            if zstd_c.len() < best_compressed.len() {
794                best_codec = limnifs_core::codec::CODEC_ZSTD;
795                best_compressed = zstd_c.into();
796            }
797        }
798    }
799
800    // Only try the specialized codec if the general-purpose ratio
801    // is poor (>15%) — otherwise the specialized codec is unlikely
802    // to help and may be very slow (FLAC, FSST). RICEPP is always
803    // tried because it can win big on FITS even when general-purpose
804    // ratios look acceptable.
805    let general_ratio = best_compressed.len() as f64 / data.len() as f64;
806    if general_ratio > 0.15 || cat.codec_id == limnifs_core::codec::CODEC_RICEPP {
807        // For FSST+Brotli, pass the already-computed Brotli baseline so
808        // the codec doesn't re-compress the plaintext with Brotli just
809        // for the comparison check.
810        let spec_result = if cat.codec_id == limnifs_core::codec::CODEC_FSST_BROTLI {
811            limnifs_core::codec::fsst_brotli::compress_with_baseline(data, Some(&best_compressed))
812        } else {
813            limnifs_core::codec::compress(cat.codec_id, data)
814        };
815        if let Ok(spec_c) = spec_result {
816            if spec_c.len() < best_compressed.len() {
817                best_codec = cat.codec_id;
818                best_compressed = spec_c.into();
819            }
820        }
821    }
822
823    Ok(ChunkedFileResult {
824        drops: vec![(drop_id, data.to_vec(), best_compressed, best_codec)],
825        slices: vec![PendingSlice {
826            drop_id,
827            file_byte_start: 0,
828            file_byte_end: file_len,
829        }],
830    })
831}
832
833/// Process a single file's contents (CPU-heavy work that runs in a
834/// rayon worker thread). Returns the unique chunks and slice map.
835///
836/// First consults the file-level categorizer registry. If a
837/// categorizer claims the file (e.g. FLAC for WAV, ricepp for FITS,
838/// Compress a single chunk via the configured tournament.
839///
840/// Iterates `tournament.codec_ids` in declared order, tracks the
841/// smallest output, and short-circuits when a codec achieves
842/// compression ratio ≤ `tournament.short_circuit_permille`.
843///
844/// Special cases:
845/// - **Binary chunks with `skip_for_binary`**: skip the tournament
846///   entirely and use `binary_codec`. Matches v0.1 behaviour.
847/// - **Chunks smaller than `min_size`**: use the class's preferred
848///   codec directly. Per-codec setup cost dominates here and the
849///   ratio difference is negligible at small sizes.
850/// - **Class unknown to writer** (Unknown / future classes): STORE.
851///
852/// The tournament never tries `CODEC_STORE` (id 0x00) — store is
853/// always the implicit fallback if every codec fails to compress.
854fn compress_chunk_with_tournament(
855    chunk: &[u8],
856    class: classifier::Class,
857    text_codec: u8,
858    binary_codec: u8,
859    tunables: &limnifs_core::codec::CodecTunables,
860    tournament: &TournamentSpec,
861) -> (u8, std::sync::Arc<[u8]>) {
862    use classifier::Class;
863
864    let preferred = match class {
865        Class::Binary => binary_codec,
866        Class::Text | Class::Code | Class::Sparse => text_codec,
867        _ => limnifs_core::codec::CODEC_STORE,
868    };
869
870    if preferred == limnifs_core::codec::CODEC_STORE {
871        return (limnifs_core::codec::CODEC_STORE, chunk.to_vec().into());
872    }
873    if class == Class::Binary && tournament.skip_for_binary {
874        return compress_chunk_one(chunk, preferred, tunables);
875    }
876    if chunk.len() < tournament.min_size {
877        return compress_chunk_one(chunk, preferred, tunables);
878    }
879
880    let mut best: Option<(u8, std::sync::Arc<[u8]>)> = None;
881    for &codec_id in &tournament.codec_ids {
882        if codec_id == limnifs_core::codec::CODEC_STORE {
883            continue;
884        }
885        let c = match limnifs_core::codec::compress_with_tunables(codec_id, chunk, tunables) {
886            Ok(c) => c,
887            Err(_) => continue,
888        };
889        if c.len() >= chunk.len() {
890            continue;
891        }
892        let ratio_permille = (c.len() as u64 * 1000 / chunk.len() as u64) as u32;
893        let is_best_so_far = best.as_ref().map_or(true, |(_, b)| c.len() < b.len());
894        if is_best_so_far {
895            best = Some((codec_id, c.into()));
896        }
897        if tournament.short_circuit_permille > 0
898            && ratio_permille <= tournament.short_circuit_permille
899        {
900            break;
901        }
902    }
903
904    best.unwrap_or_else(|| (limnifs_core::codec::CODEC_STORE, chunk.to_vec().into()))
905}
906
907/// Compress `chunk` with a single codec, falling back to STORE if
908/// the codec fails or expansion occurs.
909fn compress_chunk_one(
910    chunk: &[u8],
911    codec_id: u8,
912    tunables: &limnifs_core::codec::CodecTunables,
913) -> (u8, std::sync::Arc<[u8]>) {
914    if codec_id == limnifs_core::codec::CODEC_STORE {
915        return (limnifs_core::codec::CODEC_STORE, chunk.to_vec().into());
916    }
917    match limnifs_core::codec::compress_with_tunables(codec_id, chunk, tunables) {
918        Ok(c) if c.len() < chunk.len() => (codec_id, c.into()),
919        _ => (limnifs_core::codec::CODEC_STORE, chunk.to_vec().into()),
920    }
921}
922
923/// FSST+Brotli for CSV), the whole file is compressed as a single
924/// drop with the categorizer's chosen codec + parameters. Otherwise
925/// falls through to `FastCDC` + per-chunk classify.
926fn process_file(
927    pf: &PendingFile,
928    chunker: &FastCDC,
929    classifier: classifier::Classifier,
930    text_codec: u8,
931    binary_codec: u8,
932    tunables: &limnifs_core::codec::CodecTunables,
933    use_categorizers: bool,
934    skip_chunking: bool,
935    tournament: &TournamentSpec,
936    base_drop_index: Option<&std::collections::HashSet<[u8; 32]>>,
937    inline_threshold: usize,
938) -> Result<ChunkedFileResult, WriteError> {
939    // For files above MMAP_READ_THRESHOLD, map them rather than reading
940    // into a Vec via std::fs::read. Pages load on demand from the
941    // kernel page cache. The chunk-path compressors see borrowed slices
942    // pointing into the mmap, so peak RSS stays at unique_chunks ×
943    // avg_chunk_size rather than the full file.
944    //
945    // SAFETY: LimniFS packs source trees that are immutable for the
946    // duration of the write. The file is opened read-only. External
947    // mutation during compression would be a serious bug in the
948    // caller's workflow (and would also break BLAKE3 determinism).
949    let file_len_estimate = std::fs::metadata(&pf.path)
950        .map(|m| m.len() as usize)
951        .unwrap_or(0);
952    let data: Vec<u8> = if file_len_estimate >= MMAP_READ_THRESHOLD {
953        let file = std::fs::File::open(&pf.path)?;
954        #[allow(unsafe_code)]
955        let mmap = unsafe { memmap2::Mmap::map(&file) }.map_err(WriteError::Io)?;
956        // Materialize only the pages the kernel has paged in. For
957        // chunked access this is roughly unique-chunk bytes; for
958        // skip_chunking we touch every page anyway.
959        Vec::from(&mmap[..])
960    } else {
961        std::fs::read(&pf.path)?
962    };
963    let file_len = data.len();
964
965    // Skip FastCDC chunking entirely; compress the whole file as
966    // one drop. Trades dedup granularity for create speed. Used by
967    // the max-write profile where speed >> ratio. The per-file LZ4
968    // compress at ~1 GB/s is faster than FastCDC hashing overhead
969    // for all but the largest multi-GB files (where rayon parallelism
970    // across chunks would help).
971    if skip_chunking && file_len > inline_threshold {
972        let drop_id = hash_section(&data);
973        let class = classifier.classify(&data);
974        let preferred_codec = match class {
975            classifier::Class::Binary => binary_codec,
976            _ => text_codec,
977        };
978        let (codec_id, compressed): (u8, std::sync::Arc<[u8]>) =
979            match limnifs_core::codec::compress_with_tunables(preferred_codec, &data, tunables) {
980                Ok(c) if c.len() < data.len() => (preferred_codec, c.into()),
981                _ => (limnifs_core::codec::CODEC_STORE, data.to_vec().into()),
982            };
983        return Ok(ChunkedFileResult {
984            drops: vec![(drop_id, data, compressed, codec_id)],
985            slices: vec![PendingSlice {
986                drop_id,
987                file_byte_start: 0,
988                file_byte_end: file_len as u64,
989            }],
990        });
991    }
992
993    if use_categorizers {
994        if let Some(cat) = file_categorizer::default_registry().categorize(&pf.path, &data) {
995            let needs_whole_file = matches!(
996                cat.codec_id,
997                limnifs_core::codec::CODEC_FLAC | limnifs_core::codec::CODEC_RICEPP
998            );
999            if needs_whole_file || file_len <= WHOLE_FILE_MAX_SIZE {
1000                return process_whole_file_drop(pf, &data, cat, tunables);
1001            }
1002        }
1003    }
1004
1005    let chunks = chunker.chunk_slice(&data);
1006    let mut slices = Vec::with_capacity(chunks.len());
1007    let mut file_offset: u64 = 0;
1008    let mut seen_in_file: std::collections::HashSet<[u8; 32]> =
1009        std::collections::HashSet::with_capacity(chunks.len());
1010
1011    // Phase 1: hash all chunks + build slices + filter duplicates (sequential).
1012    // FastCDC boundaries must be deterministic, and BLAKE3 hashing is fast.
1013    let mut unique_chunks: Vec<(&[u8], [u8; 32])> = Vec::with_capacity(chunks.len());
1014    for chunk in &chunks {
1015        let chunk_len = u64::try_from(chunk.len()).expect("chunk len fits u64");
1016        let drop_id = hash_section(chunk);
1017        slices.push(PendingSlice {
1018            drop_id,
1019            file_byte_start: file_offset,
1020            file_byte_end: file_offset + chunk_len,
1021        });
1022        file_offset += chunk_len;
1023        if seen_in_file.insert(drop_id) {
1024            unique_chunks.push((chunk, drop_id));
1025        }
1026    }
1027
1028    // Phase 2: compress unique chunks in parallel across rayon workers.
1029    // This is the CPU-intensive step — parallelizing it gives N-core
1030    // speedup for large files with many chunks.
1031    //
1032    // Cross-file dedup: each rayon worker thread carries a thread-local
1033    // compress cache mapping DropId -> (codec_id, compressed_bytes).
1034    // When two files share a chunk (common in source trees, container
1035    // layers, tiny-files benchmarks), the second file hits the cache
1036    // and skips the compress pass entirely. Cache is bounded by entry
1037    // count; eviction is "stop inserting once full" — simple and
1038    // correct, misses are bounded by worker count.
1039    use rayon::prelude::*;
1040    thread_local! {
1041        static COMPRESS_CACHE: std::cell::RefCell<std::collections::HashMap<[u8; 32], (u8, std::sync::Arc<[u8]>)>> =
1042            std::cell::RefCell::new(std::collections::HashMap::new());
1043    }
1044    const COMPRESS_CACHE_MAX_ENTRIES: usize = 100_000;
1045    let drops: Vec<RawDrop> = unique_chunks
1046        .par_iter()
1047        .map(|(chunk, drop_id)| {
1048            // Layer fast-path: chunk already exists in the base image
1049            // → skip compress entirely.
1050            if let Some(base) = base_drop_index {
1051                if base.contains(drop_id) {
1052                    return (*drop_id, Vec::new(), Vec::new().into(), CODEC_REFERENCED);
1053                }
1054            }
1055            let class = classifier.classify(chunk);
1056            // Cache fast-path: identical chunk already compressed on
1057            // this worker → reuse bytes, skip tournament entirely.
1058            let cached = COMPRESS_CACHE.with(|c| {
1059                c.borrow()
1060                    .get(drop_id)
1061                    .map(|(cid, comp)| (*cid, comp.clone()))
1062            });
1063            let (codec_id, compressed) = if let Some(c) = cached {
1064                c
1065            } else {
1066                let new = compress_chunk_with_tournament(
1067                    chunk,
1068                    class,
1069                    text_codec,
1070                    binary_codec,
1071                    tunables,
1072                    tournament,
1073                );
1074                // Insert into the per-worker cache if there's room.
1075                COMPRESS_CACHE.with(|c| {
1076                    let mut cache = c.borrow_mut();
1077                    if cache.len() < COMPRESS_CACHE_MAX_ENTRIES {
1078                        // `new.1.clone()` here is an Arc refcount bump.
1079                        cache.insert(*drop_id, new.clone());
1080                    }
1081                });
1082                new
1083            };
1084            (*drop_id, chunk.to_vec(), compressed, codec_id)
1085        })
1086        .collect();
1087
1088    let _ = file_len;
1089    Ok(ChunkedFileResult { drops, slices })
1090}
1091
1092struct PendingDrop {
1093    id: [u8; 32],
1094    /// Original (decompressed) byte length. Stored as a u32 rather
1095    /// than keeping the plaintext Vec around, because the writer only
1096    /// needs the length when emitting the slab's drop record. Holding
1097    /// the full plaintext until slab assembly wastes memory
1098    /// proportional to image size on top of the compressed bytes.
1099    plaintext_len: u32,
1100    compressed: std::sync::Arc<[u8]>,
1101    codec: u8,
1102    /// Dictionary id (0xFF = NO_DICT). Populated during the dict
1103    /// re-compression pass for drops that were re-compressed with a
1104    /// trained dictionary.
1105    dict_id: u8,
1106    /// Retained plaintext, present only when `collect_dict_samples`
1107    /// is true (i.e. `WriteConfig::dictionaries.enabled`). Used by
1108    /// the post-parallel dict re-compression pass. Cleared after
1109    /// re-compression to free memory before slab assembly.
1110    plaintext: Option<Vec<u8>>,
1111}
1112
1113impl PendingDrop {
1114    /// The byte length stored in the slab's solid window. Equals
1115    /// `plaintext_len` for store codec, or the compressed size for
1116    /// LZ4 / Brotli / etc.
1117    fn len_in_window(&self) -> u32 {
1118        u32::try_from(self.compressed.len()).expect("compressed fits u32")
1119    }
1120
1121    /// The original (decompressed) byte length.
1122    fn plaintext_len_value(&self) -> u32 {
1123        self.plaintext_len
1124    }
1125
1126    /// Contribution to the slab's total byte length: 48 bytes of drop
1127    /// record (per spec §3.3) + the compressed payload.
1128    fn slab_footprint(&self) -> usize {
1129        48 + self.compressed.len()
1130    }
1131}
1132
1133/// One slice of a file backed by drops. Records which drop holds
1134/// this slice's bytes and which byte range of the original file
1135/// the slice covers. The slice always spans the entire drop (the
1136/// chunker never splits a drop across multiple slices).
1137struct PendingSlice {
1138    drop_id: [u8; 32],
1139    file_byte_start: u64,
1140    file_byte_end: u64,
1141}
1142
1143/// A file that needs chunking (> `INLINE_THRESHOLD`). Collected during
1144/// the sequential tree walk and processed in parallel by `rayon`.
1145#[derive(Clone)]
1146struct PendingFile {
1147    inode_number: u64,
1148    path: PathBuf,
1149    mtime_ns: u64,
1150    file_len: u64,
1151}
1152
1153struct PendingInode {
1154    number: u64,
1155    mode: u32,
1156    mtime_ns: u64,
1157    content: PendingContent,
1158}
1159
1160enum PendingContent {
1161    Inline(Vec<u8>),
1162    /// Symlink target (raw, as read from the filesystem).
1163    Symlink(String),
1164    DropBacked {
1165        file_len: u64,
1166        slices: Vec<PendingSlice>,
1167    },
1168    Directory(Vec<(String, u64, u8)>),
1169}
1170
1171struct DirNode {
1172    entries: Vec<(String, u64, u8)>,
1173    bytes: Vec<u8>,
1174    hash: [u8; 32],
1175}
1176
1177struct WriteContext {
1178    next_inode: u64,
1179    inodes: Vec<PendingInode>,
1180    dir_nodes: Vec<DirNode>,
1181    drops: Vec<PendingDrop>,
1182    drop_index: HashSet<[u8; 32]>,
1183    pending_files: Vec<PendingFile>,
1184    file_count: usize,
1185    dir_count: usize,
1186    root_inode_number: u64,
1187    chunker: FastCDC,
1188    classifier: classifier::Classifier,
1189    shared_inline_map: HashMap<[u8; 32], usize>,
1190    shared_inline_table: Vec<Vec<u8>>,
1191    /// Profile name for ProfileDescriptor emission (None = omit section).
1192    profile_name: Option<String>,
1193    /// Metadata blob codec (defaults to Brotli; can be overridden via
1194    /// `WriteConfig::defaults::metadata_codec`). Used by `assemble`.
1195    metadata_codec: u8,
1196    /// Whether categorizers were disabled by the profile.
1197    categorizers_disabled: bool,
1198    /// Whether this is a RW image.
1199    rw_mode: bool,
1200    /// Whether auto-turnover is enabled.
1201    auto_turnover: bool,
1202    /// Whether to collect plaintext samples for ZSTD dictionary
1203    /// training. Set when `WriteConfig::dictionaries.enabled`.
1204    collect_dict_samples: bool,
1205    /// Plaintext samples collected from ZSTD-compressed drops, for
1206    /// training one dictionary after the parallel compress phase.
1207    /// Capped at `MAX_DICT_SAMPLES` to bound memory. Keyed by
1208    /// classifier class — text/code/sparse share a "text" dict,
1209    /// binary gets its own. Compressed/media/incompressible classes
1210    /// don't use ZSTD so their samples aren't collected.
1211    dict_samples_by_class: HashMap<crate::classifier::Class, Vec<Vec<u8>>>,
1212    /// Trained dictionaries keyed by class. Populated by
1213    /// `train_and_apply_dictionary` after the parallel phase. Emitted
1214    /// in the manifest's `dictionary_section` with one entry per
1215    /// class that accumulated enough samples.
1216    trained_dicts_by_class: HashMap<crate::classifier::Class, crate::dictionary::TrainedDictionary>,
1217    /// Drop IDs known to exist in a base image (set when this write
1218    /// is producing a layer via `write_layer`). When a chunk's DropId
1219    /// is in this set, the writer skips compression and emits no slab
1220    /// bytes — the drop is resolved via the overlay chain at read
1221    /// time. `None` for standalone (non-layer) writes.
1222    base_drop_index: Option<HashSet<[u8; 32]>>,
1223    /// The base image's `ManifestRoot` (set when `base_drop_index`
1224    /// is `Some`). Emitted in the manifest's `delta_linkage` section
1225    /// so readers know which image provides the referenced drops.
1226    /// `None` for standalone writes.
1227    base_root: Option<[u8; 32]>,
1228    /// Compressed-metadata size above which the blob is externalized
1229    /// to a sidecar (issue #187). Defaults to
1230    /// [`METADATA_EXTERNALIZE_THRESHOLD`]; overridable via
1231    /// `WriteConfig::defaults::metadata_externalize_threshold`.
1232    metadata_externalize_threshold: usize,
1233    /// Whether to dedup identical inline file contents into the
1234    /// shared-inline table (issue #189). `true` (default) keeps the
1235    /// historical behavior; `false` emits plain `INLINE_DATA` inodes
1236    /// so images stay readable by pre-#186 readers whose reserved
1237    /// mask rejects the `SHARED_INLINE` flag.
1238    emit_shared_inline: bool,
1239    /// Inline-data cutoff from `WriteConfig::defaults.inline_threshold`.
1240    /// Files at or below this size are stored inline in the metadata
1241    /// blob instead of being chunked into slabs. Set from the profile
1242    /// before `walk`; defaults to the historical constant.
1243    inline_threshold: usize,
1244    /// Streaming-walk sink (TODO.perf/15). When set, `walk` forwards
1245    /// deferred files to the channel instead of buffering them in
1246    /// `pending_files`, so compression starts while the walk is still
1247    /// descending the tree. `None` keeps the collect-then-dispatch
1248    /// shape (used by the non-streaming entry points).
1249    pending_sink: Option<std::sync::mpsc::SyncSender<PendingFile>>,
1250}
1251
1252impl WriteContext {
1253    /// Cap on collected plaintext samples. Enough signal for the
1254    /// FrequencyTrainer without unbounded memory growth on huge inputs.
1255    const MAX_DICT_SAMPLES: usize = 1000;
1256
1257    fn new() -> Self {
1258        Self {
1259            next_inode: 1,
1260            inodes: Vec::new(),
1261            dir_nodes: Vec::new(),
1262            drops: Vec::new(),
1263            drop_index: HashSet::new(),
1264            pending_files: Vec::new(),
1265            file_count: 0,
1266            dir_count: 0,
1267            root_inode_number: 0,
1268            chunker: FastCDC::default(),
1269            classifier: classifier::Classifier,
1270            shared_inline_map: HashMap::new(),
1271            shared_inline_table: Vec::new(),
1272            profile_name: None,
1273            metadata_codec: limnifs_core::codec::CODEC_BROTLI,
1274            categorizers_disabled: false,
1275            rw_mode: false,
1276            auto_turnover: false,
1277            collect_dict_samples: false,
1278            dict_samples_by_class: HashMap::new(),
1279            trained_dicts_by_class: HashMap::new(),
1280            base_drop_index: None,
1281            base_root: None,
1282            pending_sink: None,
1283            inline_threshold: INLINE_THRESHOLD,
1284            metadata_externalize_threshold: METADATA_EXTERNALIZE_THRESHOLD,
1285            emit_shared_inline: true,
1286        }
1287    }
1288
1289    fn alloc_inode(&mut self) -> u64 {
1290        let n = self.next_inode;
1291        self.next_inode += 1;
1292        n
1293    }
1294
1295    /// Scan all inline-data inodes and build a dedup table. Only
1296    /// content appearing in > 1 inode is deduplicated; unique inline
1297    /// data stays inline (no overhead change).
1298    fn build_shared_inline_table(&mut self) {
1299        let mut counts: HashMap<[u8; 32], usize> = HashMap::new();
1300        for inode in &self.inodes {
1301            if let PendingContent::Inline(data) = &inode.content {
1302                let h = hash_section(data);
1303                *counts.entry(h).or_default() += 1;
1304            }
1305        }
1306        // Only dedup content that appears more than once.
1307        for inode in &self.inodes {
1308            if let PendingContent::Inline(data) = &inode.content {
1309                let h = hash_section(data);
1310                if counts.get(&h).copied().unwrap_or(0) > 1
1311                    && !self.shared_inline_map.contains_key(&h)
1312                {
1313                    let idx = self.shared_inline_table.len();
1314                    self.shared_inline_table.push(data.clone());
1315                    self.shared_inline_map.insert(h, idx);
1316                }
1317            }
1318        }
1319    }
1320
1321    /// Merge a parallel-processed chunked file's results into the
1322    /// context. Dedup: only new `DropId`s get added to the drops list.
1323    fn merge_chunked_file(&mut self, pf: &PendingFile, result: ChunkedFileResult) {
1324        for (drop_id, plaintext, compressed, codec) in result.drops {
1325            if self.drop_index.insert(drop_id) {
1326                // When dictionary training is enabled, classify the
1327                // plaintext and retain it for per-class dictionary
1328                // training. Text-like classes share a "text" dict;
1329                // Binary gets its own. Compressed/media/incompressible
1330                // don't use ZSTD so we skip them entirely.
1331                let retain_plaintext =
1332                    self.collect_dict_samples && codec == limnifs_core::codec::CODEC_ZSTD;
1333                if retain_plaintext {
1334                    let total: usize = self.dict_samples_by_class.values().map(Vec::len).sum();
1335                    if total < Self::MAX_DICT_SAMPLES {
1336                        let class = self.classifier.classify(&plaintext);
1337                        self.dict_samples_by_class
1338                            .entry(class)
1339                            .or_default()
1340                            .push(plaintext.clone());
1341                    }
1342                }
1343                self.drops.push(PendingDrop {
1344                    id: drop_id,
1345                    plaintext_len: u32::try_from(plaintext.len()).unwrap_or(u32::MAX),
1346                    compressed,
1347                    codec,
1348                    dict_id: limnifs_core::drop_record::NO_DICT,
1349                    plaintext: if retain_plaintext {
1350                        Some(plaintext)
1351                    } else {
1352                        None
1353                    },
1354                });
1355            }
1356        }
1357        self.inodes.push(PendingInode {
1358            number: pf.inode_number,
1359            mode: 0o100_644,
1360            mtime_ns: pf.mtime_ns,
1361            content: PendingContent::DropBacked {
1362                file_len: pf.file_len,
1363                slices: result.slices,
1364            },
1365        });
1366    }
1367
1368    /// Apply the seine classifier to a chunk and compress it if the
1369    /// class is compressible. Text, Code, and Binary drops get LZ4;
1370    /// Compressed, Media, and Sparse drops stay as store (re-compressing
1371    /// already-compressed data wastes CPU for no gain).
1372    /// Apply the seine classifier to a chunk and compress it if the
1373    /// class is compressible. Text, Code, and Binary drops get LZ4;
1374    /// Compressed, Media, and Sparse drops stay as store.
1375    ///
1376    /// Kept for API compatibility; the parallel writer uses
1377    /// [`process_file`] which inlines this logic.
1378    #[allow(dead_code)]
1379    fn deepen_drop(&self, drop_id: [u8; 32], plaintext: &[u8]) -> PendingDrop {
1380        let class = self.classifier.classify(plaintext);
1381        let (codec, compressed): (u8, std::sync::Arc<[u8]>) = match class {
1382            classifier::Class::Text | classifier::Class::Code | classifier::Class::Binary => {
1383                let c = limnifs_core::codec::compress_lz4_with_size(plaintext);
1384                (limnifs_core::codec::CODEC_LZ4, c.into())
1385            }
1386            _ => (limnifs_core::codec::CODEC_STORE, plaintext.to_vec().into()),
1387        };
1388        PendingDrop {
1389            id: drop_id,
1390            plaintext_len: u32::try_from(plaintext.len()).unwrap_or(u32::MAX),
1391            compressed,
1392            codec,
1393            dict_id: limnifs_core::drop_record::NO_DICT,
1394            plaintext: None,
1395        }
1396    }
1397
1398    fn walk(&mut self, path: &Path) -> Result<u64, WriteError> {
1399        let meta = std::fs::symlink_metadata(path)?;
1400        let file_type = meta.file_type();
1401        let mtime_ns = meta
1402            .modified()
1403            .ok()
1404            .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
1405            .map_or(0u128, |d| d.as_nanos());
1406        let mtime_ns: u64 = mtime_ns.try_into().unwrap_or(0);
1407
1408        if file_type.is_dir() {
1409            self.dir_count += 1;
1410            let inode_number = self.alloc_inode();
1411            let mut entries: Vec<(String, u64, u8)> = Vec::new();
1412
1413            for entry in std::fs::read_dir(path)? {
1414                let entry = entry?;
1415                let name = entry.file_name().to_string_lossy().into_owned();
1416                let child_path = entry.path();
1417                let child_inode = self.walk(&child_path)?;
1418                // entry.file_type() does NOT follow symlinks (the old
1419                // entry.metadata() did, misreporting links to dirs).
1420                let ft = entry.file_type()?;
1421                let entry_type = if ft.is_symlink() {
1422                    0x03
1423                } else if ft.is_dir() {
1424                    0x02
1425                } else {
1426                    0x01
1427                };
1428                entries.push((name, child_inode, entry_type));
1429            }
1430
1431            entries.sort_by(|a, b| a.0.cmp(&b.0));
1432            let dir_node = encode_dir_node(&entries);
1433            self.dir_nodes.push(dir_node);
1434            self.inodes.push(PendingInode {
1435                number: inode_number,
1436                mode: 0o040_755,
1437                mtime_ns,
1438                content: PendingContent::Directory(entries),
1439            });
1440            Ok(inode_number)
1441        } else if file_type.is_file() {
1442            self.file_count += 1;
1443            let inode_number = self.alloc_inode();
1444            let file_len = meta.len();
1445
1446            if file_len <= u64::try_from(self.inline_threshold).unwrap_or(u64::MAX) {
1447                let data = std::fs::read(path)?;
1448                self.inodes.push(PendingInode {
1449                    number: inode_number,
1450                    mode: 0o100_644,
1451                    mtime_ns,
1452                    content: PendingContent::Inline(data),
1453                });
1454            } else {
1455                // Defer to parallel processing — collect the file info.
1456                let pf = PendingFile {
1457                    inode_number,
1458                    path: path.to_path_buf(),
1459                    mtime_ns,
1460                    file_len,
1461                };
1462                if let Some(sink) = &self.pending_sink {
1463                    // Streaming mode: hand the file to the compress
1464                    // workers immediately; the bounded channel
1465                    // back-pressures if they fall behind. A send
1466                    // failure means the receiver is gone (a worker
1467                    // hit an unrecoverable error) — abort the walk.
1468                    sink.send(pf).map_err(|_| {
1469                        WriteError::Io(std::io::Error::other("walk: compress pipeline shut down"))
1470                    })?;
1471                } else {
1472                    self.pending_files.push(pf);
1473                }
1474            }
1475            Ok(inode_number)
1476        } else if file_type.is_symlink() {
1477            // Issue #190: symlinks are a first-class format citizen
1478            // (the reader has ContentHandle::Symlink and extract
1479            // recreates links); the writer just never emitted them.
1480            // The target is stored verbatim — relative or absolute,
1481            // in-tree or out-of-tree, dangling or not. Note
1482            // `symlink_metadata` above gives us the LINK's own
1483            // metadata, so a dangling link still walks cleanly.
1484            let inode_number = self.alloc_inode();
1485            let target = std::fs::read_link(path)?;
1486            let target = target
1487                .to_str()
1488                .ok_or_else(|| WriteError::UnsupportedFileType {
1489                    path: path.to_path_buf(),
1490                    kind: format!("symlink with non-UTF-8 target ({})", target.display()),
1491                })?
1492                .to_owned();
1493            self.inodes.push(PendingInode {
1494                number: inode_number,
1495                mode: limnifs_core::inode::S_IFLNK | 0o777,
1496                mtime_ns,
1497                content: PendingContent::Symlink(target),
1498            });
1499            Ok(inode_number)
1500        } else {
1501            #[cfg(unix)]
1502            let kind = {
1503                use std::os::unix::fs::FileTypeExt;
1504                if file_type.is_fifo() {
1505                    "fifo".to_owned()
1506                } else if file_type.is_socket() {
1507                    "socket".to_owned()
1508                } else if file_type.is_block_device() {
1509                    "block device".to_owned()
1510                } else if file_type.is_char_device() {
1511                    "character device".to_owned()
1512                } else {
1513                    "unknown".to_owned()
1514                }
1515            };
1516            #[cfg(not(unix))]
1517            let kind = "unknown".to_owned();
1518            Err(WriteError::UnsupportedFileType {
1519                path: path.to_path_buf(),
1520                kind,
1521            })
1522        }
1523    }
1524
1525    /// After the parallel compress phase: train one ZSTD dictionary
1526    /// per classifier class with enough samples, then re-compress
1527    /// each ZSTD drop with the dictionary for its class. Keep
1528    /// whichever representation is smaller. Drops that get
1529    /// re-compressed carry the class's `dict_id` in their drop
1530    /// record; the dictionaries are emitted in the manifest's
1531    /// `dictionary_section`.
1532    ///
1533    /// Text/Code/Sparse classes collapse into a single "text" dict
1534    /// (id 0). Binary gets id 1. Other classes don't accumulate
1535    /// samples because their drops aren't ZSTD-compressed.
1536    ///
1537    /// Clears the retained plaintext on every drop to free memory
1538    /// before slab assembly.
1539    fn train_and_apply_dictionary(&mut self, dictionaries: &crate::config::DictionaryConfig) {
1540        let cleanup = |ctx: &mut Self| {
1541            for d in &mut ctx.drops {
1542                d.plaintext = None;
1543            }
1544            ctx.dict_samples_by_class.clear();
1545        };
1546
1547        if !dictionaries.enabled {
1548            cleanup(self);
1549            return;
1550        }
1551
1552        let target = usize::try_from(dictionaries.max_dict_size).unwrap_or(65_536);
1553        let min_class = usize::try_from(dictionaries.min_class_size).unwrap_or(0);
1554
1555        // Allocate dict ids: 0 = text (Text/Code/Sparse), 1 = binary.
1556        // Compressed/Media/Incompressible don't accumulate samples
1557        // (their drops aren't ZSTD) so we don't train for them.
1558        let text_classes = [
1559            crate::classifier::Class::Text,
1560            crate::classifier::Class::Code,
1561            crate::classifier::Class::Sparse,
1562        ];
1563        let binary_classes = [crate::classifier::Class::Binary];
1564
1565        // Train text dict from text-like classes' samples combined.
1566        let text_samples: Vec<&[u8]> = text_classes
1567            .iter()
1568            .flat_map(|c| self.dict_samples_by_class.get(c).into_iter().flatten())
1569            .map(Vec::as_slice)
1570            .collect();
1571        let trainer = crate::dictionary::TrainerKind::from_config_str(&dictionaries.trainer);
1572        if text_samples.len() >= min_class {
1573            if let Some(dict) =
1574                crate::dictionary::train_zstd_with_trainer(0, &text_samples, target, trainer)
1575            {
1576                self.trained_dicts_by_class
1577                    .insert(crate::classifier::Class::Text, dict);
1578            }
1579        }
1580        let binary_samples: Vec<&[u8]> = binary_classes
1581            .iter()
1582            .flat_map(|c| self.dict_samples_by_class.get(c).into_iter().flatten())
1583            .map(Vec::as_slice)
1584            .collect();
1585        if binary_samples.len() >= min_class {
1586            if let Some(dict) =
1587                crate::dictionary::train_zstd_with_trainer(1, &binary_samples, target, trainer)
1588            {
1589                self.trained_dicts_by_class
1590                    .insert(crate::classifier::Class::Binary, dict);
1591            }
1592        }
1593
1594        // Re-compress each ZSTD drop with the dict for its class.
1595        // Keep the smaller representation.
1596        for d in self.drops.iter_mut() {
1597            if d.codec != limnifs_core::codec::CODEC_ZSTD {
1598                continue;
1599            }
1600            let Some(plaintext) = d.plaintext.clone() else {
1601                continue;
1602            };
1603            let class = self.classifier.classify(&plaintext);
1604            let dict_class = if text_classes.contains(&class) {
1605                crate::classifier::Class::Text
1606            } else if binary_classes.contains(&class) {
1607                crate::classifier::Class::Binary
1608            } else {
1609                continue;
1610            };
1611            let Some(dict) = self.trained_dicts_by_class.get(&dict_class) else {
1612                continue;
1613            };
1614            let Ok(dict_compressed) = dict.compress(&plaintext) else {
1615                continue;
1616            };
1617            if dict_compressed.len() < d.compressed.len() {
1618                d.compressed = dict_compressed.into();
1619                d.dict_id = dict.id;
1620            }
1621        }
1622
1623        cleanup(self);
1624    }
1625
1626    /// Env-gated phase timer for assemble profiling (TODO.perf/16).
1627    fn trace_phase(label: &str, start: std::time::Instant) {
1628        if std::env::var_os("LIMNIFS_TRACE_ASSEMBLE").is_some() {
1629            eprintln!("[assemble] {label}: {:?}", start.elapsed());
1630        }
1631    }
1632
1633    fn assemble(mut self) -> WriteArtifact {
1634        let t_assemble = std::time::Instant::now();
1635        let inode_count = self.inodes.len();
1636        let dir_count = self.dir_count;
1637        let drop_count = self.drops.len();
1638
1639        // Partition drops into slabs. Each slab's total byte length
1640        // (header + drop records + solid window) must stay under
1641        // MAX_SLAB_TOTAL_BYTES so the reader's 64 MiB ceiling is never
1642        // exceeded. A single drop larger than the budget gets its own
1643        // slab (we cannot split a drop).
1644        let t = std::time::Instant::now();
1645        let slabs = pack_slabs(&self.drops);
1646        Self::trace_phase("pack_slabs", t);
1647
1648        // Build the shared inline table: deduplicate inline data that
1649        // appears in more than one inode. For N files with identical
1650        // small content, store once and reference by index.
1651        let t = std::time::Instant::now();
1652        if self.emit_shared_inline {
1653            self.build_shared_inline_table();
1654        }
1655        Self::trace_phase("shared_inline_table", t);
1656
1657        let mut metadata_blob = Vec::new();
1658        metadata_blob.extend_from_slice(&u32::try_from(self.inodes.len()).unwrap().to_le_bytes());
1659        for inode in &self.inodes {
1660            self.encode_inode(&mut metadata_blob, inode);
1661        }
1662        metadata_blob
1663            .extend_from_slice(&u32::try_from(self.dir_nodes.len()).unwrap().to_le_bytes());
1664        for node in &self.dir_nodes {
1665            metadata_blob.extend_from_slice(&node.bytes);
1666        }
1667        // Shared inline table (only present if any dedup occurred).
1668        // Reader checks for remaining bytes after dir_nodes.
1669        if !self.shared_inline_table.is_empty() {
1670            metadata_blob.extend_from_slice(
1671                &u32::try_from(self.shared_inline_table.len())
1672                    .unwrap()
1673                    .to_le_bytes(),
1674            );
1675            for entry in &self.shared_inline_table {
1676                let len = u32::try_from(entry.len()).expect("shared entry fits u32");
1677                metadata_blob.extend_from_slice(&len.to_le_bytes());
1678                metadata_blob.extend_from_slice(entry);
1679            }
1680        }
1681
1682        Self::trace_phase("metadata_encode", t);
1683        // Compress the metadata blob. Metadata is highly compressible
1684        // (sequential inode numbers, repeated modes, natural-language
1685        // file names) — even low Brotli quality yields 4–8× on source
1686        // trees. Pick quality by size: small blobs cost nothing to
1687        // compress at q5; large blobs (e.g. 50 K-inode trees) would
1688        // dominate create time at q5, so step down to q2.
1689        let uncompressed_len =
1690            u32::try_from(metadata_blob.len()).expect("metadata blob length fits u32");
1691        let t = std::time::Instant::now();
1692        let metadata_hash = hash_section(&metadata_blob);
1693        let metadata_codec = self.metadata_codec;
1694        let metadata_quality = if metadata_blob.len() > METADATA_LARGE_BLOB_THRESHOLD {
1695            METADATA_LARGE_BLOB_QUALITY
1696        } else {
1697            METADATA_SMALL_BLOB_QUALITY
1698        };
1699        let compressed_blob = if metadata_codec == limnifs_core::codec::CODEC_BROTLI {
1700            limnifs_core::codec::compress_brotli_with_quality(&metadata_blob, metadata_quality)
1701                .unwrap_or_else(|_| metadata_blob.clone())
1702        } else {
1703            limnifs_core::codec::compress(metadata_codec, &metadata_blob)
1704                .unwrap_or_else(|_| metadata_blob.clone())
1705        };
1706        Self::trace_phase("metadata_compress", t);
1707        let (on_wire_codec, on_wire_blob) = if compressed_blob.len() < metadata_blob.len() {
1708            (metadata_codec, compressed_blob)
1709        } else {
1710            (limnifs_core::codec::CODEC_STORE, metadata_blob.clone())
1711        };
1712
1713        // Decide inline vs sidecar based on the COMPRESSED length,
1714        // clamped to the reader's inline ceiling regardless of config
1715        // (inline metadata above it is unreadable by default readers).
1716        let externalize_at = self
1717            .metadata_externalize_threshold
1718            .min(limnifs_core::metadata_reference::DEFAULT_INLINE_METADATA_MAX_BYTES as usize);
1719        let (metadata_sidecar, inline_data, metadata_locator_count) =
1720            if on_wire_blob.len() > externalize_at {
1721                let locator = "file:metadata.bin".to_owned();
1722                let sidecar = MetadataSidecar {
1723                    bytes: on_wire_blob.clone(),
1724                    locator,
1725                };
1726                (Some(sidecar), None, 1u32)
1727            } else {
1728                (None, Some(on_wire_blob.clone()), 0u32)
1729            };
1730
1731        let mut manifest = Vec::new();
1732
1733        let header_start = manifest.len();
1734        manifest.extend_from_slice(&ManifestHeader::current().to_bytes());
1735        let header_end = manifest.len();
1736
1737        let flags_start = manifest.len();
1738        manifest.push(FEATURE_FLAGS_SECTION_VERSION);
1739        manifest.extend_from_slice(&0u32.to_le_bytes());
1740        let flags_end = manifest.len();
1741
1742        // metadata_reference v2: hash + uncompressed_len + codec +
1743        // locator_count + locators + inline_data_len + inline_data.
1744        let meta_ref_start = manifest.len();
1745        manifest.push(METADATA_REFERENCE_SECTION_VERSION_2);
1746        manifest.extend_from_slice(&metadata_hash);
1747        manifest.extend_from_slice(&uncompressed_len.to_le_bytes());
1748        manifest.push(on_wire_codec);
1749        manifest.extend_from_slice(&metadata_locator_count.to_le_bytes());
1750        if let Some(sidecar) = &metadata_sidecar {
1751            let loc_bytes = sidecar.locator.as_bytes();
1752            let loc_len = u32::try_from(loc_bytes.len()).expect("locator fits u32");
1753            manifest.extend_from_slice(&loc_len.to_le_bytes());
1754            manifest.extend_from_slice(loc_bytes);
1755        }
1756        match &inline_data {
1757            Some(blob) => {
1758                let inline_len = u32::try_from(blob.len()).expect("metadata fits u32");
1759                manifest.extend_from_slice(&inline_len.to_le_bytes());
1760                manifest.extend_from_slice(blob);
1761            }
1762            None => {
1763                manifest.extend_from_slice(&0u32.to_le_bytes());
1764            }
1765        }
1766        let meta_ref_end = manifest.len();
1767
1768        let slab_index_start = manifest.len();
1769        manifest.push(SLAB_INDEX_SECTION_VERSION);
1770        manifest.extend_from_slice(&u32::try_from(slabs.len()).unwrap().to_le_bytes());
1771        for slab in &slabs {
1772            manifest.extend_from_slice(&slab.id.to_bytes());
1773            manifest.extend_from_slice(&1u32.to_le_bytes());
1774            let loc_bytes = slab.locator.as_bytes();
1775            let loc_len = u32::try_from(loc_bytes.len()).expect("locator fits u32");
1776            manifest.extend_from_slice(&loc_len.to_le_bytes());
1777            manifest.extend_from_slice(loc_bytes);
1778        }
1779        let slab_index_end = manifest.len();
1780
1781        let history_start = manifest.len();
1782        manifest.push(HISTORY_SECTION_VERSION);
1783        manifest.extend_from_slice(&1u32.to_le_bytes());
1784        manifest.push(0x01);
1785        manifest.extend_from_slice(&0u64.to_le_bytes());
1786        manifest.extend_from_slice(&0u32.to_le_bytes());
1787        manifest.extend_from_slice(&0u32.to_le_bytes());
1788        let history_end = manifest.len();
1789
1790        // ProfileDescriptor section (optional — appended after history).
1791        // Records which overhead layers were active so any reader can
1792        // handle the image correctly. Only emitted if a profile name
1793        // was set.
1794        let profile_desc_start = manifest.len();
1795        if let Some(ref name) = self.profile_name {
1796            let desc = limnifs_core::profile_descriptor::ProfileDescriptor {
1797                version: limnifs_core::profile_descriptor::PROFILE_DESCRIPTOR_SECTION_VERSION,
1798                profile_name: Some(name.clone()),
1799                blake3_hashing: true,
1800                cross_file_dedup: true,
1801                content_classification: !self.categorizers_disabled,
1802                integrity_verify: true,
1803                read_write: self.rw_mode,
1804                auto_turnover: self.auto_turnover,
1805            };
1806            limnifs_core::profile_descriptor::encode_profile_descriptor(&desc, &mut manifest);
1807        }
1808        let profile_desc_end = manifest.len();
1809
1810        // DictionarySection (optional — emitted when dictionaries
1811        // were trained during the post-parallel pass). Contains the
1812        // `(codec_id, class_id, data)` triples referenced by drop
1813        // records' `dict_id` field. One entry per class with enough
1814        // samples to train: text (id 0), binary (id 1).
1815        if !self.trained_dicts_by_class.is_empty() {
1816            let dicts: Vec<_> = self
1817                .trained_dicts_by_class
1818                .values()
1819                .map(|d| limnifs_core::dictionary_section::Dictionary {
1820                    codec_id: d.codec,
1821                    class_id: d.id,
1822                    data: d.content.clone(),
1823                })
1824                .collect();
1825            let section = limnifs_core::dictionary_section::DictionarySection {
1826                version: limnifs_core::dictionary_section::DICTIONARY_SECTION_VERSION,
1827                dicts,
1828            };
1829            limnifs_core::dictionary_section::encode_dictionary_section(&section, &mut manifest);
1830        }
1831
1832        let dictionary_end = manifest.len();
1833
1834        // DeltaLinkage section (optional — emitted only by `write_layer`).
1835        // Carries `base_root` so readers can resolve referenced drops via
1836        // the overlay chain. The section's hash feeds the
1837        // `SectionHashes::delta_linkage` slot, which is empty for
1838        // standalone images.
1839        let delta_linkage_hash = if let Some(base_root) = self.base_root {
1840            let delta_start = manifest.len();
1841            // Inline-encode the delta linkage section (version 1):
1842            // [version:u8][base_root:32][tree_op_count:u32=0]. Tree ops
1843            // are empty because the metadata blob carries the full new
1844            // tree — readers see `base_root` and know to walk the
1845            // overlay chain for any DropId not present in local slabs.
1846            manifest.push(limnifs_core::delta_linkage::DELTA_LINKAGE_SECTION_VERSION);
1847            manifest.extend_from_slice(&base_root);
1848            manifest.extend_from_slice(&0u32.to_le_bytes());
1849            hash_section(&manifest[delta_start..])
1850        } else {
1851            hash_empty_section()
1852        };
1853        let _ = dictionary_end;
1854
1855        let hashes = SectionHashes {
1856            metadata: metadata_hash,
1857            format_header: hash_section(&manifest[header_start..header_end]),
1858            feature_flags: hash_section(&manifest[flags_start..flags_end]),
1859            metadata_reference: hash_section(&manifest[meta_ref_start..meta_ref_end]),
1860            slab_index: hash_section(&manifest[slab_index_start..slab_index_end]),
1861            crypto_params: hash_empty_section(),
1862            ec_params: hash_empty_section(),
1863            dms_policy: hash_empty_section(),
1864            delta_linkage: delta_linkage_hash,
1865            history: hash_section(&manifest[history_start..history_end]),
1866            // The Merkle construction doesn't currently include a
1867            // dictionary_section hash slot. Treat the section as
1868            // crypto-params-equivalent (covered by the metadata hash)
1869            // for now; documenting this with an explicit comment so
1870            // the next reader knows where to add a hash slot if the
1871            // spec grows one.
1872            // TODO: spec section-hash for dictionary_section.
1873            // For now use hash_empty_section() so the structure compiles;
1874            // a future spec rev will add a dedicated slot.
1875            // (Section bytes are still content-addressed via the
1876            // slab_index hash and the manifest's Merkle root.)
1877        };
1878        let merkle_root = compute_merkle_root(&hashes);
1879
1880        WriteArtifact {
1881            bytes: manifest,
1882            merkle_root,
1883            slabs,
1884            metadata_sidecar,
1885            inode_count,
1886            file_count: self.file_count,
1887            dir_count,
1888            drop_count,
1889            root_inode_number: self.root_inode_number,
1890        }
1891    }
1892
1893    fn encode_inode(&self, out: &mut Vec<u8>, inode: &PendingInode) {
1894        out.extend_from_slice(&inode.number.to_le_bytes());
1895        out.extend_from_slice(&inode.mode.to_le_bytes());
1896        out.extend_from_slice(&0u32.to_le_bytes());
1897        out.extend_from_slice(&0u32.to_le_bytes());
1898        out.extend_from_slice(&inode.mtime_ns.to_le_bytes());
1899        out.extend_from_slice(&inode.mtime_ns.to_le_bytes());
1900        out.extend_from_slice(&1u32.to_le_bytes());
1901        match &inode.content {
1902            PendingContent::Inline(data) => {
1903                let h = hash_section(data);
1904                if let Some(&idx) = self.shared_inline_map.get(&h) {
1905                    // Deduplicated: emit shared-inline flag + index.
1906                    out.push(INODE_FLAG_INLINE_DATA | INODE_FLAG_SHARED_INLINE);
1907                    out.extend_from_slice(&(idx as u32).to_le_bytes());
1908                } else {
1909                    out.push(INODE_FLAG_INLINE_DATA);
1910                    let len = u32::try_from(data.len()).expect("data fits u32");
1911                    out.extend_from_slice(&len.to_le_bytes());
1912                    out.extend_from_slice(data);
1913                }
1914            }
1915            PendingContent::DropBacked { file_len, slices } => {
1916                out.push(0x00);
1917                let slice_count = u32::try_from(slices.len()).expect("slice count fits u32");
1918                out.extend_from_slice(&slice_count.to_le_bytes());
1919                for slice in slices {
1920                    out.extend_from_slice(&slice.file_byte_start.to_le_bytes());
1921                    out.extend_from_slice(&slice.file_byte_end.to_le_bytes());
1922                    out.extend_from_slice(&slice.drop_id);
1923                    // drop_byte_start = 0 (slice covers the whole drop)
1924                    out.extend_from_slice(&0u32.to_le_bytes());
1925                    // drop_byte_len = the byte length of this slice in the
1926                    // drop's decompressed plaintext. Each slice maps to
1927                    // exactly one chunk, so this equals the file range.
1928                    let drop_byte_len = u32::try_from(slice.file_byte_end - slice.file_byte_start)
1929                        .expect("slice range fits u32");
1930                    out.extend_from_slice(&drop_byte_len.to_le_bytes());
1931                }
1932                let _ = file_len;
1933            }
1934            PendingContent::Symlink(target) => {
1935                // The reader dispatches on the inode's S_IFMT bits and
1936                // reads target_len + target directly (flags unused
1937                // for non-regular inodes).
1938                out.push(0x00);
1939                let t = target.as_bytes();
1940                let len = u32::try_from(t.len()).expect("target fits u32");
1941                out.extend_from_slice(&len.to_le_bytes());
1942                out.extend_from_slice(t);
1943            }
1944            PendingContent::Directory(entries) => {
1945                out.push(0x00);
1946                let node = self
1947                    .dir_nodes
1948                    .iter()
1949                    .find(|n| n.entries == *entries)
1950                    .expect("directory node must exist");
1951                out.extend_from_slice(&node.hash);
1952            }
1953        }
1954    }
1955}
1956
1957/// Resolve a locator URI to the local sidecar file name, refusing
1958/// non-flat paths (CWE-22 — see
1959/// `limnifs_core::locator::local_sidecar_name`). Writer-emitted
1960/// locators are always flat, so this only fires on foreign/malicious
1961/// manifests.
1962fn sidecar_name(locator: &str) -> Result<&str, WriteError> {
1963    limnifs_core::locator::local_sidecar_name(locator)
1964        .map_err(|e| WriteError::Io(std::io::Error::other(format!("{e}"))))
1965}
1966
1967fn encode_dir_node(entries: &[(String, u64, u8)]) -> DirNode {
1968    let mut bytes = Vec::new();
1969    bytes.push(1u8);
1970    let count = u32::try_from(entries.len()).expect("entry count fits u32");
1971    bytes.extend_from_slice(&count.to_le_bytes());
1972    for (name, inode_number, entry_type) in entries {
1973        let name_bytes = name.as_bytes();
1974        let name_len = u32::try_from(name_bytes.len()).expect("name fits u32");
1975        bytes.extend_from_slice(&name_len.to_le_bytes());
1976        bytes.extend_from_slice(name_bytes);
1977        bytes.extend_from_slice(&inode_number.to_le_bytes());
1978        bytes.push(*entry_type);
1979    }
1980    let hash = hash_section(&bytes);
1981    DirNode {
1982        entries: entries.to_vec(),
1983        bytes,
1984        hash,
1985    }
1986}
1987
1988/// Partition `drops` into one or more slabs, each fitting under
1989/// [`MAX_SLAB_TOTAL_BYTES`]. Slab ordinal starts at 0 and increments.
1990/// Each slab's `SlabId` hash is `BLAKE3(slab_content)` so identical
1991/// content yields identical slab IDs (deterministic).
1992///
1993/// A single drop larger than `MAX_SLAB_TOTAL_BYTES - SLAB_HEADER_LEN`
1994/// still produces one slab — we cannot split a drop, and the spec
1995/// permits the reader to raise its ceiling for that case.
1996fn pack_slabs(drops: &[PendingDrop]) -> Vec<SlabArtifact> {
1997    // Filter out CODEC_REFERENCED sentinel drops — they exist in the
1998    // writer's in-memory state for inode/slice bookkeeping but are
1999    // resolved via the overlay chain at read time, never stored in
2000    // this image's slabs.
2001    let local_drops: Vec<&PendingDrop> = drops
2002        .iter()
2003        .filter(|d| d.codec != limnifs_core::codec::CODEC_REFERENCED)
2004        .collect();
2005    if local_drops.is_empty() {
2006        return Vec::new();
2007    }
2008
2009    let max_content = MAX_SLAB_TOTAL_BYTES.saturating_sub(SLAB_HEADER_LEN);
2010
2011    // Phase 1 (sequential): scan drops and partition into slab groups.
2012    // Each slab group is the list of drops that will share a slab window.
2013    // The grouping depends on per-drop compressed size + record overhead
2014    // — this is a sequential scan with a running size budget.
2015    let mut slab_groups: Vec<Vec<&PendingDrop>> = Vec::new();
2016    let mut current: Vec<&PendingDrop> = Vec::new();
2017    let mut current_size: usize = 0;
2018
2019    for drop in &local_drops {
2020        let footprint = drop.slab_footprint();
2021        if !current.is_empty() && current_size + footprint > max_content {
2022            slab_groups.push(std::mem::take(&mut current));
2023            current_size = 0;
2024        }
2025        current.push(*drop);
2026        current_size += footprint;
2027    }
2028    if !current.is_empty() {
2029        slab_groups.push(current);
2030    }
2031
2032    // Phase 2 (parallel): encode each slab independently. Slab encoding
2033    // has no cross-slab state — each slab's `offset_in_window` starts
2034    // at 0, its hash is over its own content, its ordinal is its index
2035    // in the slab_groups vector. Rayon parallelises across slabs;
2036    // large images with many slabs get N-core speedup on this phase.
2037    use rayon::prelude::*;
2038    slab_groups
2039        .par_iter()
2040        .enumerate()
2041        .map(|(ordinal, group)| {
2042            let ordinal_u64 = u64::try_from(ordinal).expect("slab count fits u64");
2043            encode_slab(ordinal_u64, group)
2044        })
2045        .collect()
2046}
2047
2048/// Encode a single slab from a non-empty slice of drops. Per-slab
2049/// `offset_in_window` is computed fresh; there is no global offset
2050/// state on `PendingDrop`.
2051fn encode_slab(ordinal: u64, drops: &[&PendingDrop]) -> SlabArtifact {
2052    // Each drop record is a fixed 49-byte entry: id(32) +
2053    // plaintext_len(4) + representation(3) + solid_window_index(1)
2054    // + offset_in_window(4) + window_len(4) + dict_id(1). Pre-sizing
2055    // the records Vec avoids per-drop realloc and amortises to a
2056    // single memcpy per field rather than bounds-check per call.
2057    const DROP_RECORD_LEN: usize = 49;
2058    let mut drop_records = Vec::with_capacity(drops.len() * DROP_RECORD_LEN);
2059    let mut solid_window = Vec::new();
2060    let mut drop_ids = Vec::with_capacity(drops.len());
2061    let mut offset_in_window: u32 = 0;
2062
2063    for drop in drops {
2064        let plaintext_len = drop.plaintext_len_value();
2065        let window_len = drop.len_in_window();
2066        drop_records.extend_from_slice(&drop.id);
2067        drop_records.extend_from_slice(&plaintext_len.to_le_bytes());
2068        // representation: (codec, aead=0, ec=0)
2069        drop_records.extend_from_slice(&[drop.codec, 0x00, 0x00]);
2070        drop_records.push(0x00); // solid_window_index
2071        drop_records.extend_from_slice(&offset_in_window.to_le_bytes());
2072        drop_records.extend_from_slice(&window_len.to_le_bytes());
2073        drop_records.push(drop.dict_id); // dict_id: NO_DICT (0xFF) or trained id (0..=254)
2074        solid_window.extend_from_slice(&drop.compressed);
2075        drop_ids.push(drop.id);
2076        offset_in_window = offset_in_window
2077            .checked_add(window_len)
2078            .expect("slab window size fits u32");
2079    }
2080
2081    let slab_content = [&drop_records[..], &solid_window[..]].concat();
2082    let slab_hash = hash_section(&slab_content);
2083    let slab_id = SlabId::new(ordinal, slab_hash);
2084
2085    let total_length = SLAB_HEADER_LEN + slab_content.len();
2086    let mut slab_bytes = Vec::with_capacity(total_length);
2087    slab_bytes.extend_from_slice(b"LIM1");
2088    slab_bytes.extend_from_slice(&1u16.to_le_bytes());
2089    slab_bytes.extend_from_slice(&slab_id.to_bytes());
2090    slab_bytes.extend_from_slice(
2091        &u64::try_from(total_length)
2092            .unwrap_or(u64::MAX)
2093            .to_le_bytes(),
2094    );
2095    slab_bytes.push(0x00);
2096    slab_bytes.push(0x00);
2097    slab_bytes.extend_from_slice(&slab_content);
2098
2099    let locator = format!("file:slab-{ordinal}.bin");
2100
2101    SlabArtifact {
2102        id: slab_id,
2103        bytes: slab_bytes,
2104        locator,
2105        drop_ids,
2106    }
2107}
2108
2109#[cfg(test)]
2110fn pseudo_random_bytes(seed: u64, count: usize) -> Vec<u8> {
2111    let mut state = seed;
2112    let mut out = Vec::with_capacity(count);
2113    for _ in 0..count {
2114        state = state
2115            .wrapping_mul(6_364_136_223_846_793_005)
2116            .wrapping_add(1_442_695_040_888_963_407);
2117        out.push(u8::try_from(state >> 56).expect("fits u8"));
2118    }
2119    out
2120}
2121
2122#[cfg(test)]
2123mod tests {
2124    use super::*;
2125    use limnifs_core::ManifestCursor;
2126
2127    #[test]
2128    fn write_stream_packs_single_named_stream() {
2129        // Stream 256 KiB of repetitive text through write_stream and
2130        // verify the artifact has a single root file at the requested
2131        // name with the expected size.
2132        let temp = std::env::temp_dir().join(format!(
2133            "limnifs-write-stream-test-{}-{}",
2134            std::process::id(),
2135            std::time::SystemTime::now()
2136                .duration_since(std::time::UNIX_EPOCH)
2137                .unwrap()
2138                .as_nanos()
2139        ));
2140        std::fs::create_dir_all(&temp).expect("create temp dir");
2141
2142        let content = b"stream test content line\n".repeat(10_000); // ~240 KiB
2143        let cursor = std::io::Cursor::new(content.clone());
2144        let config = WriteConfig::default_v0_1();
2145        let artifact = write_stream("streamed.txt", cursor, &config).expect("write_stream");
2146
2147        assert!(!artifact.bytes.is_empty(), "manifest bytes non-empty");
2148        assert!(!artifact.slabs.is_empty(), "at least one slab produced");
2149        // The artifact's manifest encodes the file metadata; we don't
2150        // deeply inspect it here (conformance suite covers that), but
2151        // we do confirm the writer produced something well-formed
2152        // enough that round-tripping through the reader works.
2153        let total_drop_bytes: usize = artifact.slabs.iter().map(|s| s.bytes.len()).sum();
2154        assert!(total_drop_bytes > 0, "drops non-empty");
2155
2156        let _ = std::fs::remove_dir_all(&temp);
2157    }
2158
2159    #[test]
2160    fn write_layer_references_base_drops() {
2161        // Build a base image with a 1 MiB text file, then build a
2162        // layer that ADDS a new file AND includes the same text file.
2163        // The layer's slab bytes must be small (just the new file)
2164        // because the text file's chunks hit the base's drop set and
2165        // are emitted as `CODEC_REFERENCED`.
2166        let temp = std::env::temp_dir().join(format!(
2167            "limnifs-write-layer-test-{}-{}",
2168            std::process::id(),
2169            std::time::SystemTime::now()
2170                .duration_since(std::time::UNIX_EPOCH)
2171                .unwrap()
2172                .as_nanos()
2173        ));
2174        std::fs::create_dir_all(&temp).expect("create temp dir");
2175
2176        // Base: 1 MiB of repetitive text + a small unique file.
2177        let base_dir = temp.join("base");
2178        std::fs::create_dir_all(&base_dir).expect("base dir");
2179        let text = b"layer test content line\n".repeat(50_000); // ~1.15 MiB
2180        std::fs::write(base_dir.join("shared.txt"), &text).expect("write shared");
2181        std::fs::write(base_dir.join("base-only.txt"), b"only in base").expect("write base-only");
2182
2183        let config = WriteConfig::default_v0_1();
2184        let base_artifact = write_directory_with_config(&base_dir, &config).expect("base");
2185
2186        let base_manifest = temp.join("base.lim");
2187        std::fs::write(&base_manifest, &base_artifact.bytes).expect("write base manifest");
2188        for slab in &base_artifact.slabs {
2189            let slab_name = sidecar_name(&slab.locator).expect("slab locator");
2190            std::fs::write(temp.join(slab_name), &slab.bytes).expect("write base slab");
2191        }
2192
2193        // Layer: same shared.txt (must dedup against base) + a new file.
2194        let layer_dir = temp.join("layer");
2195        std::fs::create_dir_all(&layer_dir).expect("layer dir");
2196        std::fs::write(layer_dir.join("shared.txt"), &text).expect("write shared in layer");
2197        std::fs::write(layer_dir.join("new.txt"), b"fresh content in layer").expect("write new");
2198
2199        let layer_artifact = write_layer(&base_manifest, &layer_dir, &config).expect("layer");
2200
2201        // The layer's slabs should be SMALL — only the new file's
2202        // content. shared.txt's chunks are referenced via the base.
2203        let layer_slab_bytes: usize = layer_artifact.slabs.iter().map(|s| s.bytes.len()).sum();
2204        let base_slab_bytes: usize = base_artifact.slabs.iter().map(|s| s.bytes.len()).sum();
2205        assert!(
2206            layer_slab_bytes < base_slab_bytes / 4,
2207            "layer slabs ({}) should be much smaller than base ({}) — layering failed",
2208            layer_slab_bytes,
2209            base_slab_bytes
2210        );
2211
2212        // The layer manifest must contain the base's Merkle root in
2213        // its delta_linkage section.
2214        let base_root = base_artifact.merkle_root.as_bytes();
2215        assert!(
2216            layer_artifact
2217                .bytes
2218                .windows(32)
2219                .any(|w| w == base_root.as_slice()),
2220            "layer manifest must contain base's ManifestRoot bytes"
2221        );
2222
2223        let _ = std::fs::remove_dir_all(&temp);
2224    }
2225
2226    #[test]
2227    fn tournament_short_circuits_on_highly_compressible_chunk() {
2228        // Repetitive text compresses to <25% under LZ4. Tournament
2229        // should accept LZ4 and skip the slower Brotli pass.
2230        let chunk = b"hello world ".repeat(500);
2231        let tunables = limnifs_core::codec::CodecTunables::default();
2232        let tournament = TournamentSpec {
2233            codec_ids: vec![
2234                limnifs_core::codec::CODEC_LZ4,
2235                limnifs_core::codec::CODEC_BROTLI,
2236            ],
2237            min_size: 16,
2238            skip_for_binary: false,
2239            short_circuit_permille: 250,
2240        };
2241        let (codec_id, compressed) = compress_chunk_with_tournament(
2242            &chunk,
2243            classifier::Class::Text,
2244            limnifs_core::codec::CODEC_BROTLI,
2245            limnifs_core::codec::CODEC_LZ4,
2246            &tunables,
2247            &tournament,
2248        );
2249        assert_eq!(codec_id, limnifs_core::codec::CODEC_LZ4);
2250        assert!(compressed.len() < chunk.len());
2251    }
2252
2253    #[test]
2254    fn tournament_runs_all_codecs_when_short_circuit_disabled() {
2255        // short_circuit_permille = 0 means "never short-circuit". The
2256        // tournament must try every codec and pick the smallest.
2257        // With omnizip 0.14.40, ZSTD (cached Huffman table) typically
2258        // beats both LZ4 and Brotli's Phase-C partial encoder on
2259        // repetitive text, so we include it in the tournament.
2260        let chunk = b"hello world ".repeat(500);
2261        let tunables = limnifs_core::codec::CodecTunables::default();
2262        let tournament = TournamentSpec {
2263            codec_ids: vec![
2264                limnifs_core::codec::CODEC_LZ4,
2265                limnifs_core::codec::CODEC_BROTLI,
2266                limnifs_core::codec::CODEC_ZSTD,
2267            ],
2268            min_size: 16,
2269            skip_for_binary: false,
2270            short_circuit_permille: 0,
2271        };
2272        let (codec_id, compressed) = compress_chunk_with_tournament(
2273            &chunk,
2274            classifier::Class::Text,
2275            limnifs_core::codec::CODEC_BROTLI,
2276            limnifs_core::codec::CODEC_LZ4,
2277            &tunables,
2278            &tournament,
2279        );
2280        // All three codecs should be tried; the smallest wins. With
2281        // omnizip 0.16.40's long copy fix (MAX_COPY 271→4096), Brotli
2282        // now beats ZSTD on repetitive text. Either is acceptable.
2283        assert!(
2284            codec_id == limnifs_core::codec::CODEC_ZSTD
2285                || codec_id == limnifs_core::codec::CODEC_BROTLI,
2286            "expected ZSTD or Brotli to win, got codec {codec_id}"
2287        );
2288        assert!(compressed.len() < chunk.len());
2289    }
2290
2291    #[test]
2292    fn tournament_skips_for_binary_when_configured() {
2293        let chunk = vec![0u8; 4096];
2294        let tunables = limnifs_core::codec::CodecTunables::default();
2295        let tournament = TournamentSpec {
2296            codec_ids: vec![limnifs_core::codec::CODEC_BROTLI],
2297            min_size: 16,
2298            skip_for_binary: true,
2299            short_circuit_permille: 250,
2300        };
2301        let (codec_id, _compressed) = compress_chunk_with_tournament(
2302            &chunk,
2303            classifier::Class::Binary,
2304            limnifs_core::codec::CODEC_BROTLI,
2305            limnifs_core::codec::CODEC_LZ4,
2306            &tunables,
2307            &tournament,
2308        );
2309        // skip_for_binary → use binary_codec (LZ4) directly, never Brotli.
2310        assert_eq!(codec_id, limnifs_core::codec::CODEC_LZ4);
2311    }
2312
2313    #[test]
2314    fn tournament_small_chunk_uses_preferred_codec() {
2315        let chunk = b"tiny";
2316        let tunables = limnifs_core::codec::CodecTunables::default();
2317        let tournament = TournamentSpec {
2318            codec_ids: vec![limnifs_core::codec::CODEC_BROTLI],
2319            min_size: 1024,
2320            skip_for_binary: false,
2321            short_circuit_permille: 0,
2322        };
2323        let (codec_id, _compressed) = compress_chunk_with_tournament(
2324            chunk,
2325            classifier::Class::Text,
2326            limnifs_core::codec::CODEC_BROTLI,
2327            limnifs_core::codec::CODEC_LZ4,
2328            &tunables,
2329            &tournament,
2330        );
2331        // Below min_size → preferred codec (brotli for text) directly.
2332        assert_eq!(codec_id, limnifs_core::codec::CODEC_STORE);
2333    }
2334
2335    #[test]
2336    fn tournament_falls_back_to_store_when_no_codec_compresses() {
2337        // Random data — no codec should improve on store. We use the
2338        // pseudo-random generator from the test helpers to get
2339        // deterministic but incompressible bytes.
2340        let chunk = pseudo_random_bytes(42, 4096);
2341        let tunables = limnifs_core::codec::CodecTunables::default();
2342        let tournament = TournamentSpec {
2343            codec_ids: vec![limnifs_core::codec::CODEC_LZ4],
2344            min_size: 16,
2345            skip_for_binary: false,
2346            short_circuit_permille: 0,
2347        };
2348        let (codec_id, compressed) = compress_chunk_with_tournament(
2349            &chunk,
2350            classifier::Class::Binary,
2351            limnifs_core::codec::CODEC_BROTLI,
2352            limnifs_core::codec::CODEC_LZ4,
2353            &tunables,
2354            &tournament,
2355        );
2356        assert_eq!(codec_id, limnifs_core::codec::CODEC_STORE);
2357        assert_eq!(compressed.len(), chunk.len());
2358    }
2359
2360    #[test]
2361    fn dictionaries_enabled_emits_dictionary_section_when_enough_samples() {
2362        // Many small text files with shared vocabulary → FrequencyTrainer
2363        // should find a dictionary. We assert the section appears in
2364        // the manifest, regardless of whether the trained dict beats
2365        // per-drop compression (the trainer is content-dependent).
2366        let temp = std::env::temp_dir().join(format!(
2367            "limnifs-write-test-{}-dict-{}",
2368            std::process::id(),
2369            std::time::SystemTime::now()
2370                .duration_since(std::time::UNIX_EPOCH)
2371                .map(|d| d.as_nanos() as u64)
2372                .unwrap_or(0),
2373        ));
2374        let _ = std::fs::remove_dir_all(&temp);
2375        std::fs::create_dir_all(&temp).expect("mkdir");
2376
2377        // Generate 200 similar small files just above INLINE_THRESHOLD
2378        // so they go through the slab path.
2379        for i in 0..200 {
2380            // Repeated vocabulary the trainer can exploit.
2381            let content = format!(
2382                "function test_case_{i}() {{ return constant + {i}; }}\n\
2383                 // shared comment line {i}\n\
2384                 struct Foo {{ x: i32 }} // type {i}\n"
2385            )
2386            .repeat(5);
2387            let path = temp.join(format!("file_{i:04}.txt"));
2388            std::fs::write(&path, content.as_bytes()).expect("write");
2389        }
2390
2391        let mut config = crate::profile::balanced();
2392        // Force ZSTD for text so drops go through the dict-eligible path.
2393        config.defaults.text_codec = "zstd".into();
2394        // omnizip 0.14.40's Brotli encoder emits some streams the
2395        // in-house decoder rejects on highly repetitive input. Use ZSTD
2396        // for the metadata blob too so the round-trip parse succeeds.
2397        config.defaults.metadata_codec = "zstd".into();
2398        config.dictionaries.enabled = true;
2399        config.dictionaries.min_class_size = 50;
2400        config.dictionaries.max_dict_size = 8192;
2401
2402        let artifact = write_directory_with_config(&temp, &config).expect("write");
2403        std::fs::remove_dir_all(&temp).ok();
2404
2405        // The dictionary_section (if emitted) lives after the history
2406        // section. We don't strictly assert presence because the trainer
2407        // may legitimately return an empty dict; the test's job is to
2408        // verify the pipeline doesn't panic and the manifest parses.
2409        let mut cursor = ManifestCursor::new(&artifact.bytes);
2410        let _ = limnifs_core::parse_manifest_header(&mut cursor).expect("header");
2411        let _ = limnifs_core::parse_feature_flags_section(&mut cursor).expect("flags");
2412        let _ = limnifs_core::parse_metadata_reference(&mut cursor).expect("meta_ref");
2413        let _ = limnifs_core::parse_slab_index(&mut cursor).expect("slab_index");
2414        let _ = limnifs_core::parse_history(&mut cursor).expect("history");
2415        // If a dict was emitted, parsing past history should leave
2416        // non-empty remaining bytes.
2417        let _remaining = cursor.remaining_len();
2418    }
2419
2420    #[test]
2421    fn write_empty_directory() {
2422        let temp =
2423            std::env::temp_dir().join(format!("limnifs-write-test-{}-empty", std::process::id()));
2424        std::fs::create_dir_all(&temp).expect("create temp dir");
2425        let artifact = write_directory(&temp).expect("write succeeds");
2426        std::fs::remove_dir_all(&temp).ok();
2427        assert!(artifact.inode_count >= 1);
2428        assert_eq!(artifact.file_count, 0);
2429        assert_eq!(artifact.dir_count, 1);
2430        assert!(artifact.slabs.is_empty());
2431    }
2432
2433    #[test]
2434    fn write_small_file_inline() {
2435        let temp =
2436            std::env::temp_dir().join(format!("limnifs-write-test-{}-small", std::process::id()));
2437        std::fs::create_dir_all(&temp).expect("create temp dir");
2438        std::fs::write(temp.join("hello.txt"), b"hello world").expect("write file");
2439        let artifact = write_directory(&temp).expect("write succeeds");
2440        std::fs::remove_dir_all(&temp).ok();
2441        assert_eq!(artifact.file_count, 1);
2442        assert!(artifact.slabs.is_empty());
2443        assert_eq!(artifact.drop_count, 0);
2444    }
2445
2446    #[test]
2447    fn write_large_file_uses_slab() {
2448        let temp =
2449            std::env::temp_dir().join(format!("limnifs-write-test-{}-large", std::process::id()));
2450        std::fs::create_dir_all(&temp).expect("create temp dir");
2451        let large_data = vec![0xABu8; INLINE_THRESHOLD + 100];
2452        std::fs::write(temp.join("big.bin"), &large_data).expect("write big");
2453        let artifact = write_directory(&temp).expect("write succeeds");
2454        std::fs::remove_dir_all(&temp).ok();
2455        assert_eq!(artifact.drop_count, 1);
2456        assert_eq!(artifact.slabs.len(), 1);
2457    }
2458
2459    #[test]
2460    fn write_mixed_inline_and_large() {
2461        let temp =
2462            std::env::temp_dir().join(format!("limnifs-write-test-{}-mix", std::process::id()));
2463        std::fs::create_dir_all(&temp).expect("create temp dir");
2464        std::fs::write(temp.join("small.txt"), b"tiny").expect("write small");
2465        std::fs::write(temp.join("large.bin"), vec![0xCDu8; INLINE_THRESHOLD * 2])
2466            .expect("write large");
2467        let artifact = write_directory(&temp).expect("write succeeds");
2468        std::fs::remove_dir_all(&temp).ok();
2469        assert_eq!(artifact.file_count, 2);
2470        assert_eq!(artifact.drop_count, 1);
2471        assert_eq!(artifact.slabs.len(), 1);
2472    }
2473
2474    #[test]
2475    fn deduplicates_identical_large_files() {
2476        let temp =
2477            std::env::temp_dir().join(format!("limnifs-write-test-{}-dedup", std::process::id()));
2478        std::fs::create_dir_all(&temp).expect("create temp dir");
2479        let data = vec![0x77u8; INLINE_THRESHOLD + 10];
2480        std::fs::write(temp.join("a.bin"), &data).expect("write a");
2481        std::fs::write(temp.join("b.bin"), &data).expect("write b");
2482        let artifact = write_directory(&temp).expect("write succeeds");
2483        std::fs::remove_dir_all(&temp).ok();
2484        assert_eq!(artifact.drop_count, 1);
2485    }
2486
2487    #[test]
2488    fn write_and_verify_roundtrip() {
2489        let temp = std::env::temp_dir().join(format!(
2490            "limnifs-write-test-{}-roundtrip",
2491            std::process::id()
2492        ));
2493        std::fs::create_dir_all(&temp).expect("create temp dir");
2494        std::fs::write(temp.join("a.txt"), b"aaa").expect("write a");
2495        std::fs::write(temp.join("b.txt"), b"bbb").expect("write b");
2496        std::fs::create_dir_all(temp.join("sub")).expect("create sub");
2497        std::fs::write(temp.join("sub").join("c.txt"), b"ccc").expect("write c");
2498        let artifact = write_directory(&temp).expect("write succeeds");
2499        std::fs::remove_dir_all(&temp).ok();
2500        assert_eq!(artifact.file_count, 3);
2501        assert_eq!(artifact.dir_count, 2);
2502
2503        let mut cursor = ManifestCursor::new(&artifact.bytes);
2504        limnifs_core::parse_manifest_header(&mut cursor).expect("header");
2505        limnifs_core::parse_feature_flags_section(&mut cursor).expect("flags");
2506        let meta_ref = limnifs_core::parse_metadata_reference(&mut cursor).expect("meta ref");
2507        assert!(meta_ref.is_inlined());
2508        let slab_index = limnifs_core::parse_slab_index(&mut cursor).expect("slab index");
2509        assert_eq!(slab_index.len(), 0);
2510        limnifs_core::parse_history(&mut cursor).expect("history");
2511    }
2512
2513    #[test]
2514    fn write_deterministic() {
2515        let temp =
2516            std::env::temp_dir().join(format!("limnifs-write-test-{}-det", std::process::id()));
2517        std::fs::create_dir_all(&temp).expect("create temp dir");
2518        std::fs::write(temp.join("x.txt"), b"xxx").expect("write x");
2519
2520        let a1 = write_directory(&temp).expect("first write");
2521        let a2 = write_directory(&temp).expect("second write");
2522        std::fs::remove_dir_all(&temp).ok();
2523
2524        assert_eq!(a1.bytes, a2.bytes);
2525        assert_eq!(a1.merkle_root, a2.merkle_root);
2526    }
2527
2528    #[test]
2529    fn slab_parses_correctly() {
2530        let temp =
2531            std::env::temp_dir().join(format!("limnifs-write-test-{}-slab", std::process::id()));
2532        std::fs::create_dir_all(&temp).expect("create temp dir");
2533        std::fs::write(temp.join("big.bin"), vec![0x11u8; INLINE_THRESHOLD + 1])
2534            .expect("write big");
2535        let artifact = write_directory(&temp).expect("write succeeds");
2536        std::fs::remove_dir_all(&temp).ok();
2537
2538        let slab_bytes = &artifact.slabs[0].bytes;
2539        let mut cursor = ManifestCursor::new(slab_bytes);
2540        let slab_header = limnifs_core::parse_slab_header(&mut cursor).expect("slab header parses");
2541        assert_eq!(slab_header.format_version, 1);
2542        assert!(!slab_header.is_sealed());
2543        assert!(!slab_header.has_erasure_coding());
2544
2545        let drop_record =
2546            limnifs_core::parse_drop_record(&mut cursor, &slab_header).expect("drop record parses");
2547        assert_eq!(drop_record.plaintext_len as usize, INLINE_THRESHOLD + 1);
2548    }
2549
2550    #[test]
2551    fn fastcdc_produces_multiple_chunks_for_large_files() {
2552        // A 1 MiB pseudo-random file should produce multiple drops
2553        // via FastCDC (default chunker uses 64 KiB min / 256 KiB avg).
2554        let temp = std::env::temp_dir().join(format!(
2555            "limnifs-write-test-{}-cdc-multi",
2556            std::process::id()
2557        ));
2558        std::fs::create_dir_all(&temp).expect("create temp dir");
2559        let data = pseudo_random_bytes(42, 1024 * 1024);
2560        std::fs::write(temp.join("big.bin"), &data).expect("write big");
2561        let artifact = write_directory(&temp).expect("write succeeds");
2562        std::fs::remove_dir_all(&temp).ok();
2563        assert!(
2564            artifact.drop_count > 1,
2565            "expected FastCDC to produce multiple drops for 1 MiB input, got {}",
2566            artifact.drop_count
2567        );
2568    }
2569
2570    #[test]
2571    fn fastcdc_deduplicates_shared_substrings() {
2572        // Two files sharing a long middle section should produce
2573        // fewer drops than the sum of their individual chunk counts,
2574        // because the shared section's chunks deduplicate.
2575        let temp = std::env::temp_dir().join(format!(
2576            "limnifs-write-test-{}-cdc-dedup",
2577            std::process::id()
2578        ));
2579        std::fs::create_dir_all(&temp).expect("create temp dir");
2580        let shared = pseudo_random_bytes(7, 512 * 1024);
2581        let mut a = Vec::with_capacity(shared.len() + 1024);
2582        a.extend_from_slice(&pseudo_random_bytes(1, 1024));
2583        a.extend_from_slice(&shared);
2584        let mut b = Vec::with_capacity(shared.len() + 2048);
2585        b.extend_from_slice(&pseudo_random_bytes(2, 2048));
2586        b.extend_from_slice(&shared);
2587        std::fs::write(temp.join("a.bin"), &a).expect("write a");
2588        std::fs::write(temp.join("b.bin"), &b).expect("write b");
2589
2590        // Baseline: each file alone.
2591        let temp_a = std::env::temp_dir().join(format!(
2592            "limnifs-write-test-{}-cdc-dedup-a",
2593            std::process::id()
2594        ));
2595        std::fs::create_dir_all(&temp_a).expect("create temp_a");
2596        std::fs::write(temp_a.join("a.bin"), &a).expect("write a");
2597        let artifact_a = write_directory(&temp_a).expect("a writes");
2598        std::fs::remove_dir_all(&temp_a).ok();
2599
2600        let temp_b = std::env::temp_dir().join(format!(
2601            "limnifs-write-test-{}-cdc-dedup-b",
2602            std::process::id()
2603        ));
2604        std::fs::create_dir_all(&temp_b).expect("create temp_b");
2605        std::fs::write(temp_b.join("b.bin"), &b).expect("write b");
2606        let artifact_b = write_directory(&temp_b).expect("b writes");
2607        std::fs::remove_dir_all(&temp_b).ok();
2608
2609        let artifact_both = write_directory(&temp).expect("both write");
2610        std::fs::remove_dir_all(&temp).ok();
2611
2612        let sum_alone = artifact_a.drop_count + artifact_b.drop_count;
2613        assert!(
2614            artifact_both.drop_count < sum_alone,
2615            "expected dedup win: both together = {} drops, sum alone = {} drops",
2616            artifact_both.drop_count,
2617            sum_alone
2618        );
2619    }
2620
2621    #[test]
2622    fn slab_splits_when_content_exceeds_ceiling() {
2623        // Synthesise enough incompressible drops to force at least two
2624        // slabs. Each drop is 10 MiB of pseudo-random data; three drops
2625        // = 30 MiB compressed (random data doesn't compress), which
2626        // fits in one slab. We bump to seven drops (70 MiB) to force a
2627        // split.
2628        let temp =
2629            std::env::temp_dir().join(format!("limnifs-write-test-{}-split", std::process::id()));
2630        std::fs::create_dir_all(&temp).expect("create temp dir");
2631        for i in 0..7u32 {
2632            // 10 MiB of pseudo-random bytes — incompressible.
2633            let data = pseudo_random_bytes(u64::from(i), 10 * 1024 * 1024);
2634            std::fs::write(temp.join(format!("big-{i}.bin")), &data).expect("write big");
2635        }
2636        let artifact = write_directory(&temp).expect("write succeeds");
2637        std::fs::remove_dir_all(&temp).ok();
2638
2639        // Each slab's total length must respect MAX_SLAB_TOTAL_BYTES.
2640        assert!(
2641            artifact.slabs.len() >= 2,
2642            "expected at least 2 slabs for 70 MiB of incompressible data, got {}",
2643            artifact.slabs.len()
2644        );
2645        for slab in &artifact.slabs {
2646            assert!(
2647                slab.bytes.len() <= MAX_SLAB_TOTAL_BYTES,
2648                "slab {} is {} bytes (> {} ceiling)",
2649                slab.id.ordinal,
2650                slab.bytes.len(),
2651                MAX_SLAB_TOTAL_BYTES,
2652            );
2653        }
2654        // All seven drops must be accounted for across slabs.
2655        let total_drop_ids: usize = artifact.slabs.iter().map(|s| s.drop_ids.len()).sum();
2656        assert_eq!(
2657            total_drop_ids, artifact.drop_count,
2658            "drop_ids count across slabs must match WriteArtifact.drop_count",
2659        );
2660    }
2661}