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