1#![deny(unsafe_code)]
21#![allow(warnings)]
22
23pub mod chunker;
24pub mod classifier;
25pub mod compaction;
26pub mod config;
27pub mod delta_builder;
28pub mod dictionary;
29pub mod file_categorizer;
30use file_categorizer::FileCategorizer;
31pub mod flatten;
32pub mod progress;
33pub mod rw;
34#[cfg(feature = "sparse-index")]
35pub mod sparse_index;
36pub mod stream;
37pub mod turnover;
38
39pub use config::{
40 profile, CategorizerConfig, ChunkingConfig, CodecRegistry, CodecTunables, Defaults,
41 DictionaryConfig, EncryptionConfig, TournamentConfig, WriteConfig,
42};
43
44use std::collections::{HashMap, HashSet};
45use std::path::{Path, PathBuf};
46
47use crate::chunker::{Chunker, ParallelFastCDC};
48use limnifs_core::codec::CODEC_REFERENCED;
49use limnifs_core::dictionary_section::parse_dictionary_section;
50use limnifs_core::slab_store::SlabStore;
51use limnifs_core::{
52 compute_merkle_root, hash_empty_section, hash_section, parse_manifest_header, parse_slab_index,
53 ManifestCursor, ManifestHeader, SectionHashes, FEATURE_FLAGS_SECTION_VERSION,
54 HISTORY_SECTION_VERSION, INODE_FLAG_INLINE_DATA, INODE_FLAG_SHARED_INLINE,
55 METADATA_REFERENCE_SECTION_VERSION_2, SLAB_INDEX_SECTION_VERSION,
56};
57use limnifs_format::{ManifestRoot, SlabId};
58
59pub const INLINE_THRESHOLD: usize = 4096;
62
63pub const MMAP_READ_THRESHOLD: usize = 1024 * 1024;
69
70pub const WHOLE_FILE_MAX_SIZE: usize = 64 * 1024 * 1024;
75
76pub const MAX_SLAB_TOTAL_BYTES: usize = 60 * 1024 * 1024;
81
82const SLAB_HEADER_LEN: usize = 56;
86
87pub const METADATA_EXTERNALIZE_THRESHOLD: usize =
95 limnifs_core::metadata_reference::DEFAULT_INLINE_METADATA_MAX_BYTES as usize - 24 * 1024;
96
97pub const METADATA_LARGE_BLOB_THRESHOLD: usize = 256 * 1024;
102
103pub const METADATA_SMALL_BLOB_QUALITY: i32 = 5;
106
107pub const METADATA_LARGE_BLOB_QUALITY: i32 = 2;
112
113#[derive(Clone, Debug)]
116pub struct SlabArtifact {
117 pub id: SlabId,
118 pub bytes: Vec<u8>,
119 pub locator: String,
120 pub drop_ids: Vec<[u8; 32]>,
124}
125
126#[derive(Clone, Debug)]
130pub struct MetadataSidecar {
131 pub bytes: Vec<u8>,
132 pub locator: String,
133}
134
135#[derive(Clone, Debug)]
137pub struct WriteArtifact {
138 pub bytes: Vec<u8>,
139 pub merkle_root: ManifestRoot,
140 pub slabs: Vec<SlabArtifact>,
143 pub metadata_sidecar: Option<MetadataSidecar>,
147 pub inode_count: usize,
148 pub file_count: usize,
149 pub dir_count: usize,
150 pub drop_count: usize,
151 pub root_inode_number: u64,
156}
157
158impl WriteArtifact {
159 #[must_use]
163 pub fn slab_bytes(&self) -> Option<&[u8]> {
164 if self.slabs.len() == 1 {
165 Some(&self.slabs[0].bytes)
166 } else {
167 None
168 }
169 }
170
171 #[must_use]
173 pub fn slab_locator(&self) -> Option<&str> {
174 if self.slabs.len() == 1 {
175 Some(&self.slabs[0].locator)
176 } else {
177 None
178 }
179 }
180}
181
182#[derive(Debug)]
184pub enum WriteError {
185 Io(std::io::Error),
186 UnsupportedFileType {
191 path: PathBuf,
192 kind: String,
193 },
194}
195
196impl std::fmt::Display for WriteError {
197 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
198 match self {
199 Self::Io(e) => write!(f, "I/O error: {e}"),
200 Self::UnsupportedFileType { path, kind } => write!(
201 f,
202 "unsupported file type ({kind}): {} — limnifs stores files, \
203 directories, and symlinks; remove the entry or file an issue \
204 if you need it carried",
205 path.display()
206 ),
207 }
208 }
209}
210
211impl std::error::Error for WriteError {}
212
213impl From<std::io::Error> for WriteError {
214 fn from(e: std::io::Error) -> Self {
215 Self::Io(e)
216 }
217}
218
219pub fn write_directory(root: &Path) -> Result<WriteArtifact, WriteError> {
233 write_directory_with_config(root, &WriteConfig::default_v0_1())
234}
235
236pub fn write_stream<R: std::io::Read>(
254 name: &str,
255 mut reader: R,
256 config: &WriteConfig,
257) -> Result<WriteArtifact, WriteError> {
258 let mut writer = crate::stream::StreamWriter::new(config)?;
259 writer.add_file(name, 0, &mut reader)?;
260 writer.finish()
261}
262
263pub fn write_layer(
304 base_image: &Path,
305 root: &Path,
306 config: &WriteConfig,
307) -> Result<WriteArtifact, WriteError> {
308 let base_root = load_base_drop_index(base_image)?.1;
310 let base_drop_index: std::sync::Arc<dyn BaseDropSet> = {
311 #[cfg(feature = "sparse-index")]
312 {
313 match SparseBackedBaseIndex::open(base_image) {
314 Some(idx) => std::sync::Arc::new(idx),
315 None => std::sync::Arc::new(load_base_drop_index(base_image)?.0),
316 }
317 }
318 #[cfg(not(feature = "sparse-index"))]
319 {
320 std::sync::Arc::new(load_base_drop_index(base_image)?.0)
321 }
322 };
323
324 let mut ctx = WriteContext::new();
325 ctx.chunker = chunker_from_config(config)?;
326 ctx.base_dictionaries = if config.dictionaries.enabled {
329 load_base_dictionary_section(base_image)?.map(crate::dictionary::adopt_from_section)
330 } else {
331 None
332 };
333 ctx.categorizers_disabled = config.categorizers.is_empty();
334 ctx.rw_mode = matches!(config.mode, crate::config::ImageMode::ReadWrite(_));
335 ctx.auto_turnover = config.turnover_threshold > 0;
336 ctx.collect_dict_samples = config.dictionaries.enabled;
337 ctx.inline_threshold = config.defaults.inline_threshold as usize;
338 ctx.metadata_externalize_threshold = config.defaults.metadata_externalize_threshold;
339 ctx.emit_shared_inline = config.defaults.shared_inline;
340 ctx.base_drop_index = Some(base_drop_index);
341 ctx.base_root = Some(base_root);
342
343 let root_inode_number = ctx.walk(root)?;
345 ctx.root_inode_number = root_inode_number;
346 write_directory_body(&mut ctx, config)?;
347 Ok(ctx.assemble())
348}
349
350pub trait BaseDropSet: Send + Sync {
363 fn base_contains(&self, drop_id: &[u8; 32]) -> bool;
366}
367
368impl BaseDropSet for std::collections::HashSet<[u8; 32]> {
369 fn base_contains(&self, drop_id: &[u8; 32]) -> bool {
370 self.contains(drop_id)
371 }
372}
373
374#[cfg(feature = "sparse-index")]
380pub struct SparseBackedBaseIndex {
381 bloom: crate::sparse_index::SparseIndexReader,
382 manifest_path: std::path::PathBuf,
383 exact: std::sync::OnceLock<std::collections::HashSet<[u8; 32]>>,
384}
385
386#[cfg(feature = "sparse-index")]
387impl SparseBackedBaseIndex {
388 #[must_use]
391 pub fn open(base_image: &Path) -> Option<Self> {
392 let sidecar = base_image.with_extension("lim.sparse");
393 let bloom = crate::sparse_index::SparseIndexReader::from_file(&sidecar)?;
394 Some(Self {
395 bloom,
396 manifest_path: base_image.to_path_buf(),
397 exact: std::sync::OnceLock::new(),
398 })
399 }
400
401 fn load_exact(&self) -> &std::collections::HashSet<[u8; 32]> {
402 self.exact.get_or_init(|| {
403 let bytes = std::fs::read(&self.manifest_path).unwrap_or_default();
407 let mut cursor = ManifestCursor::new(&bytes);
408 let _ = parse_manifest_header(&mut cursor);
409 let _ = limnifs_core::parse_feature_flags_section(&mut cursor);
410 let _ = limnifs_core::parse_metadata_reference(&mut cursor);
411 let Ok(index) = parse_slab_index(&mut cursor) else {
412 return std::collections::HashSet::new();
413 };
414 match SlabStore::load_mmap(&self.manifest_path, &index) {
415 Ok(store) => store.drop_index_keys().copied().collect(),
416 Err(_) => std::collections::HashSet::new(),
417 }
418 })
419 }
420}
421
422#[cfg(feature = "sparse-index")]
423impl BaseDropSet for SparseBackedBaseIndex {
424 fn base_contains(&self, drop_id: &[u8; 32]) -> bool {
425 if !self.bloom.probably_contains(drop_id) {
430 return false;
431 }
432 self.load_exact().contains(drop_id)
433 }
434}
435
436#[cfg(feature = "sparse-index")]
444pub fn emit_sparse_sidecar(artifact: &WriteArtifact, image_path: &Path) -> Result<(), WriteError> {
445 let all: std::collections::HashSet<[u8; 32]> = artifact
446 .slabs
447 .iter()
448 .flat_map(|s| s.drop_ids.iter().copied())
449 .collect();
450 let mut writer = crate::sparse_index::SparseIndexWriter::new(
451 all.len().max(1),
452 crate::sparse_index::DEFAULT_FPP,
453 );
454 writer.insert_all(&all);
455 let sidecar = image_path.with_extension("lim.sparse");
456 writer.write_to_file(&sidecar).map_err(WriteError::Io)
457}
458
459fn load_base_dictionary_section(
465 base_image: &Path,
466) -> Result<Option<limnifs_core::dictionary_section::DictionarySection>, WriteError> {
467 let manifest_bytes = std::fs::read(base_image)?;
468 let mut cursor = ManifestCursor::new(&manifest_bytes);
469 let _ = parse_manifest_header(&mut cursor).map_err(io_core)?;
470 let _ = limnifs_core::parse_feature_flags_section(&mut cursor);
471 let _ = limnifs_core::parse_metadata_reference(&mut cursor);
472 let _ = parse_slab_index(&mut cursor);
473 let _ = limnifs_core::parse_history(&mut cursor);
474 if cursor.remaining_len() == 0 {
475 return Ok(None);
476 }
477 Ok(parse_dictionary_section(&mut cursor).ok())
478}
479
480fn load_base_drop_index(
481 base_image: &Path,
482) -> Result<(std::collections::HashSet<[u8; 32]>, [u8; 32]), WriteError> {
483 let manifest_bytes = std::fs::read(base_image)?;
484 let mut cursor = ManifestCursor::new(&manifest_bytes);
485 let _ = parse_manifest_header(&mut cursor).map_err(io_core)?;
486 let _ = limnifs_core::parse_feature_flags_section(&mut cursor);
488 let _ = limnifs_core::parse_metadata_reference(&mut cursor);
489 let slab_index = parse_slab_index(&mut cursor).map_err(io_core)?;
490 let store = SlabStore::load_mmap(base_image, &slab_index).map_err(io_core)?;
491 let drop_set: std::collections::HashSet<[u8; 32]> = store.drop_index_keys().copied().collect();
492 let root = *compute_merkle_root_from_sections(&manifest_bytes).as_bytes();
498 Ok((drop_set, root))
499}
500
501fn compute_merkle_root_from_sections(manifest: &[u8]) -> ManifestRoot {
507 use limnifs_core::SectionHashes;
508 let mut cursor = ManifestCursor::new(manifest);
509 let header_start = 0;
510 if parse_manifest_header(&mut cursor).is_err() {
511 return ManifestRoot::from_bytes([0u8; 32]);
513 }
514 let header_end = cursor.position();
515 let flags_start = header_end;
517 let flags_end = match limnifs_core::parse_feature_flags_section(&mut cursor) {
518 Ok(_) => cursor.position(),
519 Err(_) => flags_start,
520 };
521 let meta_ref_start = flags_end;
522 let metadata_reference = match limnifs_core::parse_metadata_reference(&mut cursor) {
523 Ok(m) => Some(m),
524 Err(_) => None,
525 };
526 let meta_ref_end = cursor.position();
527 let slab_index_start = meta_ref_end;
528 let _ = parse_slab_index(&mut cursor);
529 let slab_index_end = cursor.position();
530 let history_start = slab_index_end;
531 let _ = limnifs_core::parse_history(&mut cursor);
532 let history_end = cursor.position();
533
534 let hashes = SectionHashes {
535 metadata: metadata_reference
536 .map(|m| m.metadata_hash)
537 .unwrap_or_else(hash_empty_section),
538 format_header: hash_section(&manifest[header_start..header_end]),
539 feature_flags: hash_section(&manifest[flags_start..flags_end]),
540 metadata_reference: hash_section(&manifest[meta_ref_start..meta_ref_end]),
541 slab_index: hash_section(&manifest[slab_index_start..slab_index_end]),
542 crypto_params: hash_empty_section(),
543 ec_params: hash_empty_section(),
544 dms_policy: hash_empty_section(),
545 delta_linkage: hash_empty_section(),
546 history: hash_section(&manifest[history_start..history_end]),
547 };
548 compute_merkle_root(&hashes)
549}
550
551fn io_core(e: limnifs_core::CoreError) -> WriteError {
552 WriteError::Io(std::io::Error::other(format!("base image load: {e}")))
553}
554
555fn write_directory_body(ctx: &mut WriteContext, config: &WriteConfig) -> Result<(), WriteError> {
560 use rayon::prelude::*;
561
562 ctx.metadata_codec = config
563 .metadata_codec_id()
564 .unwrap_or(limnifs_core::codec::CODEC_BROTLI);
565
566 ctx.chunker = chunker_from_config(config)?;
567
568 let pending = std::mem::take(&mut ctx.pending_files);
569 if pending.is_empty() {
570 return Ok(());
571 }
572 ctx.inline_threshold = config.defaults.inline_threshold as usize;
573 ctx.metadata_externalize_threshold = config.defaults.metadata_externalize_threshold;
574 ctx.emit_shared_inline = config.defaults.shared_inline;
575 let chunker = ctx.chunker.clone();
576 let classifier = ctx.classifier;
577 let text_codec = config.text_codec_id().unwrap_or(0x04);
578 let binary_codec = config.binary_codec_id().unwrap_or(0x01);
579 let tunables = config.to_core_tunables();
580 let use_categorizers = !config.categorizers.is_empty();
581 let skip_chunking = config.skip_chunking;
582 let registry = config
583 .codec_registry()
584 .map_err(|e| WriteError::Io(std::io::Error::other(format!("codec registry: {e}"))))?;
585 let tournament_codec_ids: Vec<u8> = config
586 .tournament
587 .codecs
588 .iter()
589 .filter_map(|n| registry.lookup_by_name(n))
590 .collect();
591 let tournament_spec = TournamentSpec {
592 codec_ids: tournament_codec_ids,
593 min_size: config.tournament.min_size_threshold as usize,
594 skip_for_binary: config.tournament.skip_for_binary,
595 short_circuit_permille: config.tournament.short_circuit_threshold,
596 };
597 let base_drop_index: Option<&dyn BaseDropSet> = ctx.base_drop_index.as_deref();
598 let inline_threshold = ctx.inline_threshold;
599 let max_drop_size = config.defaults.max_drop_size as usize;
600 let seekable_drops = config.defaults.seekable_drops;
601 let seekable_drops = config.defaults.seekable_drops;
602 let results: Vec<ChunkedFileResult> = pending
603 .par_iter()
604 .map(|pf| {
605 process_file(
606 pf,
607 &chunker,
608 classifier,
609 text_codec,
610 binary_codec,
611 &tunables,
612 use_categorizers,
613 skip_chunking,
614 &tournament_spec,
615 base_drop_index,
616 inline_threshold,
617 max_drop_size,
618 seekable_drops,
619 config.categorizers.as_slice(),
620 &|name| {
621 config
622 .codec_registry()
623 .ok()
624 .and_then(|r| r.lookup_by_name(name))
625 },
626 )
627 })
628 .collect::<Result<Vec<_>, _>>()?;
629
630 for (pf, result) in pending.iter().zip(results) {
631 ctx.merge_chunked_file(pf, result);
632 }
633 ctx.train_and_apply_dictionary(&config.dictionaries);
634 Ok(())
635}
636
637pub fn write_directory_with_config(
639 root: &Path,
640 config: &WriteConfig,
641) -> Result<WriteArtifact, WriteError> {
642 let mut ctx = WriteContext::new();
643 ctx.chunker = chunker_from_config(config)?;
644 ctx.categorizers_disabled = config.categorizers.is_empty();
645 ctx.rw_mode = matches!(config.mode, crate::config::ImageMode::ReadWrite(_));
646 ctx.auto_turnover = config.turnover_threshold > 0;
647 ctx.collect_dict_samples = config.dictionaries.enabled;
648
649 write_directory_streaming(&mut ctx, root, config)?;
650 Ok(ctx.assemble())
651}
652
653fn write_directory_streaming(
667 ctx: &mut WriteContext,
668 root: &Path,
669 config: &WriteConfig,
670) -> Result<(), WriteError> {
671 use rayon::prelude::*;
672
673 ctx.metadata_codec = config
674 .metadata_codec_id()
675 .unwrap_or(limnifs_core::codec::CODEC_BROTLI);
676
677 ctx.chunker = chunker_from_config(config)?;
678
679 let chunker = ctx.chunker.clone();
680 let classifier = ctx.classifier;
681 let text_codec = config.text_codec_id().unwrap_or(0x04);
682 let binary_codec = config.binary_codec_id().unwrap_or(0x01);
683 let tunables = config.to_core_tunables();
684 let use_categorizers = !config.categorizers.is_empty();
685 let skip_chunking = config.skip_chunking;
686 let registry = config
687 .codec_registry()
688 .map_err(|e| WriteError::Io(std::io::Error::other(format!("codec registry: {e}"))))?;
689 let tournament_codec_ids: Vec<u8> = config
690 .tournament
691 .codecs
692 .iter()
693 .filter_map(|n| registry.lookup_by_name(n))
694 .collect();
695 let tournament_spec = TournamentSpec {
696 codec_ids: tournament_codec_ids,
697 min_size: config.tournament.min_size_threshold as usize,
698 skip_for_binary: config.tournament.skip_for_binary,
699 short_circuit_permille: config.tournament.short_circuit_threshold,
700 };
701 let base_drop_index = ctx.base_drop_index.clone();
704 let inline_threshold = ctx.inline_threshold;
705 let max_drop_size = config.defaults.max_drop_size as usize;
706 let seekable_drops = config.defaults.seekable_drops;
707
708 ctx.inline_threshold = config.defaults.inline_threshold as usize;
709 ctx.metadata_externalize_threshold = config.defaults.metadata_externalize_threshold;
710 ctx.emit_shared_inline = config.defaults.shared_inline;
711
712 const PIPELINE_CAPACITY: usize = 256;
716 let (tx, rx) = std::sync::mpsc::sync_channel::<PendingFile>(PIPELINE_CAPACITY);
717 ctx.pending_sink = Some(tx);
718
719 let (root_inode_number, mut results): (
720 u64,
721 Vec<(usize, PendingFile, Result<ChunkedFileResult, WriteError>)>,
722 ) = {
723 let survey = survey_tree(root)?;
733 std::thread::scope(|scope| {
734 let producer = {
735 let ctx = &mut *ctx;
736 let root = root;
737 scope.spawn(move || {
738 let r = ctx.fold_survey(root, &survey, None);
739 ctx.pending_sink = None;
743 r
744 })
745 };
746 let results = rx
749 .into_iter()
750 .enumerate()
751 .par_bridge()
752 .map(|(i, pf)| {
753 let r = process_file(
754 &pf,
755 &chunker,
756 classifier,
757 text_codec,
758 binary_codec,
759 &tunables,
760 use_categorizers,
761 skip_chunking,
762 &tournament_spec,
763 base_drop_index.as_deref(),
764 inline_threshold,
765 max_drop_size,
766 seekable_drops,
767 config.categorizers.as_slice(),
768 &|name| {
769 config
770 .codec_registry()
771 .ok()
772 .and_then(|r| r.lookup_by_name(name))
773 },
774 );
775 (i, pf, r)
776 })
777 .collect();
778 let joined = producer
779 .join()
780 .unwrap_or_else(|_| {
781 Err(WriteError::Io(std::io::Error::other(
782 "walk thread panicked",
783 )))
784 })
785 .map(|n| (n, results));
786 joined
789 })
790 }?;
791 ctx.pending_sink = None;
792 ctx.root_inode_number = root_inode_number;
793
794 results.sort_unstable_by_key(|(i, _, _)| *i);
795 for (_, pf, r) in results {
799 ctx.merge_chunked_file(&pf, r?);
800 }
801 ctx.train_and_apply_dictionary(&config.dictionaries);
802 Ok(())
803}
804
805pub(crate) type RawDrop = ([u8; 32], Vec<u8>, std::sync::Arc<[u8]>, u8, u8);
812pub(crate) struct ChunkedFileResult {
814 drops: Vec<RawDrop>, slices: Vec<PendingSlice>,
816}
817
818struct TournamentSpec {
826 codec_ids: Vec<u8>,
830 min_size: usize,
834 skip_for_binary: bool,
838 short_circuit_permille: u32,
843}
844
845fn chunker_from_config(config: &WriteConfig) -> Result<ParallelFastCDC, WriteError> {
864 ParallelFastCDC::new(
865 config.chunking.min_chunk_size as usize,
866 config.chunking.avg_chunk_size as usize,
867 config.chunking.max_chunk_size as usize,
868 )
869 .map_err(|e| WriteError::Io(std::io::Error::other(format!("chunking config: {e}"))))
870}
871
872pub(crate) fn seekable_or_monolithic(
879 codec: u8,
880 plaintext: &[u8],
881 compressed: std::sync::Arc<[u8]>,
882 tunables: &limnifs_core::codec::CodecTunables,
883 seekable_drops: bool,
884 threshold: usize,
885) -> (std::sync::Arc<[u8]>, u8) {
886 use limnifs_core::seekable::{
887 encode_seekable, is_seekable_codec, DROP_FLAG_SEEKABLE as FLAG, SEEKABLE_EMISSION_THRESHOLD,
888 };
889 if seekable_drops && plaintext.len() > threshold && is_seekable_codec(codec) {
890 if let Ok(container) = encode_seekable(codec, plaintext, tunables) {
891 return (container.into(), FLAG);
892 }
893 }
894 (compressed, 0)
895}
896
897pub(crate) const SEEKABLE_CHUNK_EMISSION_THRESHOLD: usize =
904 limnifs_core::seekable::SEEKABLE_FRAME_SIZE;
905
906fn process_whole_file_drop(
907 pf: &PendingFile,
908 data: &[u8],
909 cat: file_categorizer::Categorization,
910 tunables: &limnifs_core::codec::CodecTunables,
911 seekable_drops: bool,
912) -> Result<ChunkedFileResult, WriteError> {
913 let _ = pf;
914 let drop_id = hash_section(data);
915 let file_len = u64::try_from(data.len()).unwrap_or(u64::MAX);
916
917 let (mut best_codec, mut best_compressed): (u8, std::sync::Arc<[u8]>) =
922 match limnifs_core::codec::compress_with_tunables(
923 limnifs_core::codec::CODEC_BROTLI,
924 data,
925 tunables,
926 ) {
927 Ok(c) => (limnifs_core::codec::CODEC_BROTLI, c.into()),
928 Err(_) => match limnifs_core::codec::compress_with_tunables(
929 limnifs_core::codec::CODEC_ZSTD,
930 data,
931 tunables,
932 ) {
933 Ok(c) => (limnifs_core::codec::CODEC_ZSTD, c.into()),
934 Err(_) => (limnifs_core::codec::CODEC_STORE, data.to_vec().into()),
935 },
936 };
937
938 let brotli_ratio = best_compressed.len() as f64 / data.len() as f64;
942 if brotli_ratio > 0.05 {
943 if let Ok(zstd_c) = limnifs_core::codec::compress_with_tunables(
944 limnifs_core::codec::CODEC_ZSTD,
945 data,
946 tunables,
947 ) {
948 if zstd_c.len() < best_compressed.len() {
949 best_codec = limnifs_core::codec::CODEC_ZSTD;
950 best_compressed = zstd_c.into();
951 }
952 }
953 }
954
955 let general_ratio = best_compressed.len() as f64 / data.len() as f64;
961 if general_ratio > 0.15 || cat.codec_id == limnifs_core::codec::CODEC_RICEPP {
962 let spec_result = if cat.codec_id == limnifs_core::codec::CODEC_FSST_BROTLI {
966 limnifs_core::codec::fsst_brotli::compress_with_baseline(data, Some(&best_compressed))
967 } else {
968 limnifs_core::codec::compress_with_tunables(cat.codec_id, data, tunables)
972 };
973 if let Ok(spec_c) = spec_result {
974 if spec_c.len() < best_compressed.len() {
975 best_codec = cat.codec_id;
976 best_compressed = spec_c.into();
977 }
978 }
979 }
980
981 let (best_compressed, flags) = seekable_or_monolithic(
982 best_codec,
983 data,
984 best_compressed,
985 tunables,
986 seekable_drops,
987 limnifs_core::seekable::SEEKABLE_EMISSION_THRESHOLD,
988 );
989 Ok(ChunkedFileResult {
990 drops: vec![(drop_id, data.to_vec(), best_compressed, best_codec, flags)],
991 slices: vec![PendingSlice {
992 drop_id,
993 file_byte_start: 0,
994 file_byte_end: file_len,
995 }],
996 })
997}
998
999fn compress_chunk_with_tournament(
1021 chunk: &[u8],
1022 class: classifier::Class,
1023 text_codec: u8,
1024 binary_codec: u8,
1025 tunables: &limnifs_core::codec::CodecTunables,
1026 tournament: &TournamentSpec,
1027) -> (u8, std::sync::Arc<[u8]>) {
1028 use classifier::Class;
1029
1030 let preferred = match class {
1031 Class::Binary => binary_codec,
1032 Class::Text | Class::Code | Class::Sparse => text_codec,
1033 _ => limnifs_core::codec::CODEC_STORE,
1034 };
1035
1036 if preferred == limnifs_core::codec::CODEC_STORE {
1037 return (limnifs_core::codec::CODEC_STORE, chunk.to_vec().into());
1038 }
1039 if class == Class::Binary && tournament.skip_for_binary {
1040 return compress_chunk_one(chunk, preferred, tunables);
1041 }
1042 if chunk.len() < tournament.min_size {
1043 return compress_chunk_one(chunk, preferred, tunables);
1044 }
1045
1046 let mut best: Option<(u8, std::sync::Arc<[u8]>)> = None;
1047 for &codec_id in &tournament.codec_ids {
1048 if codec_id == limnifs_core::codec::CODEC_STORE {
1049 continue;
1050 }
1051 let c = match limnifs_core::codec::compress_with_tunables(codec_id, chunk, tunables) {
1052 Ok(c) => c,
1053 Err(_) => continue,
1054 };
1055 if c.len() >= chunk.len() {
1056 continue;
1057 }
1058 let ratio_permille = (c.len() as u64 * 1000 / chunk.len() as u64) as u32;
1059 let is_best_so_far = best.as_ref().map_or(true, |(_, b)| c.len() < b.len());
1060 if is_best_so_far {
1061 best = Some((codec_id, c.into()));
1062 }
1063 if tournament.short_circuit_permille > 0
1064 && ratio_permille <= tournament.short_circuit_permille
1065 {
1066 break;
1067 }
1068 }
1069
1070 best.unwrap_or_else(|| (limnifs_core::codec::CODEC_STORE, chunk.to_vec().into()))
1071}
1072
1073fn compress_chunk_one(
1076 chunk: &[u8],
1077 codec_id: u8,
1078 tunables: &limnifs_core::codec::CodecTunables,
1079) -> (u8, std::sync::Arc<[u8]>) {
1080 if codec_id == limnifs_core::codec::CODEC_STORE {
1081 return (limnifs_core::codec::CODEC_STORE, chunk.to_vec().into());
1082 }
1083 match limnifs_core::codec::compress_with_tunables(codec_id, chunk, tunables) {
1084 Ok(c) if c.len() < chunk.len() => (codec_id, c.into()),
1085 _ => (limnifs_core::codec::CODEC_STORE, chunk.to_vec().into()),
1086 }
1087}
1088
1089fn process_file(
1093 pf: &PendingFile,
1094 chunker: &dyn Chunker,
1095 classifier: classifier::Classifier,
1096 text_codec: u8,
1097 binary_codec: u8,
1098 tunables: &limnifs_core::codec::CodecTunables,
1099 use_categorizers: bool,
1100 skip_chunking: bool,
1101 tournament: &TournamentSpec,
1102 base_drop_index: Option<&dyn BaseDropSet>,
1103 inline_threshold: usize,
1104 max_drop_size: usize,
1105 seekable_drops: bool,
1106 categorizer_config: &[crate::config::CategorizerConfig],
1107 codec_name_resolver: &dyn Fn(&str) -> Option<u8>,
1108) -> Result<ChunkedFileResult, WriteError> {
1109 let file_len_estimate = std::fs::metadata(&pf.path)
1120 .map(|m| m.len() as usize)
1121 .unwrap_or(0);
1122 let mmap_handle: memmap2::Mmap;
1129 let small: Vec<u8>;
1130 let data: &[u8] = if file_len_estimate >= MMAP_READ_THRESHOLD {
1131 let file = std::fs::File::open(&pf.path)?;
1132 #[allow(unsafe_code)]
1136 let mapped = unsafe { memmap2::Mmap::map(&file) }.map_err(WriteError::Io)?;
1137 mmap_handle = mapped;
1138 &mmap_handle[..]
1139 } else {
1140 small = std::fs::read(&pf.path)?;
1141 &small[..]
1142 };
1143 let file_len = data.len();
1144
1145 if skip_chunking && file_len > inline_threshold {
1152 let drop_id = hash_section(&data);
1153 let class = classifier.classify(&data);
1154 let preferred_codec = match class {
1155 classifier::Class::Binary => binary_codec,
1156 _ => text_codec,
1157 };
1158 let (codec_id, compressed): (u8, std::sync::Arc<[u8]>) =
1159 match limnifs_core::codec::compress_with_tunables(preferred_codec, &data, tunables) {
1160 Ok(c) if c.len() < data.len() => (preferred_codec, c.into()),
1161 _ => (limnifs_core::codec::CODEC_STORE, data.to_vec().into()),
1162 };
1163 let (compressed, flags) = seekable_or_monolithic(
1164 codec_id,
1165 &data,
1166 compressed,
1167 tunables,
1168 seekable_drops,
1169 SEEKABLE_CHUNK_EMISSION_THRESHOLD,
1170 );
1171 return Ok(ChunkedFileResult {
1172 drops: vec![(drop_id, data.to_vec(), compressed, codec_id, flags)],
1173 slices: vec![PendingSlice {
1174 drop_id,
1175 file_byte_start: 0,
1176 file_byte_end: file_len as u64,
1177 }],
1178 });
1179 }
1180
1181 if use_categorizers {
1182 let config_cat = file_categorizer::ConfigCategorizer::new(categorizer_config.to_vec());
1185 if let Some(cat) = config_cat.categorize(&pf.path, &data) {
1186 if let Some(codec_id) = file_categorizer::resolve_config_categorization(
1187 &cat,
1188 categorizer_config,
1189 codec_name_resolver,
1190 ) {
1191 let within_cap = max_drop_size == 0 || file_len <= max_drop_size;
1192 let needs_whole_file = matches!(
1193 codec_id,
1194 limnifs_core::codec::CODEC_FLAC | limnifs_core::codec::CODEC_RICEPP
1195 );
1196 if within_cap && (needs_whole_file || file_len <= WHOLE_FILE_MAX_SIZE) {
1197 let mut cat = cat;
1198 cat.codec_id = codec_id;
1199 return process_whole_file_drop(pf, &data, cat, tunables, seekable_drops);
1200 }
1201 }
1202 }
1203 if let Some(cat) = file_categorizer::default_registry().categorize(&pf.path, &data) {
1204 let needs_whole_file = matches!(
1205 cat.codec_id,
1206 limnifs_core::codec::CODEC_FLAC | limnifs_core::codec::CODEC_RICEPP
1207 );
1208 let within_cap = max_drop_size == 0 || file_len <= max_drop_size;
1211 if within_cap && (needs_whole_file || file_len <= WHOLE_FILE_MAX_SIZE) {
1212 return process_whole_file_drop(pf, &data, cat, tunables, seekable_drops);
1213 }
1214 }
1215 }
1216
1217 let chunks = chunker.chunk_slice(&data);
1218
1219 use rayon::prelude::*;
1228 let drop_ids: Vec<[u8; 32]> = chunks.par_iter().map(|chunk| hash_section(chunk)).collect();
1229
1230 let mut slices = Vec::with_capacity(chunks.len());
1231 let mut file_offset: u64 = 0;
1232 let mut seen_in_file: std::collections::HashSet<[u8; 32]> =
1233 std::collections::HashSet::with_capacity(chunks.len());
1234 let mut unique_chunks: Vec<(&[u8], [u8; 32])> = Vec::with_capacity(chunks.len());
1235 for (chunk, drop_id) in chunks.iter().zip(drop_ids) {
1236 let chunk_len = u64::try_from(chunk.len()).expect("chunk len fits u64");
1237 slices.push(PendingSlice {
1238 drop_id,
1239 file_byte_start: file_offset,
1240 file_byte_end: file_offset + chunk_len,
1241 });
1242 file_offset += chunk_len;
1243 if seen_in_file.insert(drop_id) {
1244 unique_chunks.push((chunk, drop_id));
1245 }
1246 }
1247
1248 thread_local! {
1260 static COMPRESS_CACHE: std::cell::RefCell<std::collections::HashMap<[u8; 32], (u8, std::sync::Arc<[u8]>)>> =
1261 std::cell::RefCell::new(std::collections::HashMap::new());
1262 }
1263 const COMPRESS_CACHE_MAX_ENTRIES: usize = 100_000;
1264 let drops: Vec<RawDrop> = unique_chunks
1265 .par_iter()
1266 .map(|(chunk, drop_id)| {
1267 if let Some(base) = base_drop_index {
1270 if base.base_contains(drop_id) {
1271 return (*drop_id, Vec::new(), Vec::new().into(), CODEC_REFERENCED, 0);
1272 }
1273 }
1274 let class = classifier.classify(chunk);
1275 let cached = COMPRESS_CACHE.with(|c| {
1278 c.borrow()
1279 .get(drop_id)
1280 .map(|(cid, comp)| (*cid, comp.clone()))
1281 });
1282 let (codec_id, compressed) = if let Some(c) = cached {
1283 c
1284 } else {
1285 let new = compress_chunk_with_tournament(
1286 chunk,
1287 class,
1288 text_codec,
1289 binary_codec,
1290 tunables,
1291 tournament,
1292 );
1293 COMPRESS_CACHE.with(|c| {
1295 let mut cache = c.borrow_mut();
1296 if cache.len() < COMPRESS_CACHE_MAX_ENTRIES {
1297 cache.insert(*drop_id, new.clone());
1299 }
1300 });
1301 new
1302 };
1303 let (compressed, flags) = seekable_or_monolithic(
1309 codec_id,
1310 chunk,
1311 compressed,
1312 tunables,
1313 seekable_drops,
1314 SEEKABLE_CHUNK_EMISSION_THRESHOLD,
1315 );
1316 (*drop_id, chunk.to_vec(), compressed, codec_id, flags)
1317 })
1318 .collect();
1319
1320 let _ = file_len;
1321 Ok(ChunkedFileResult { drops, slices })
1322}
1323
1324struct PendingDrop {
1325 id: [u8; 32],
1326 plaintext_len: u32,
1332 compressed: std::sync::Arc<[u8]>,
1333 codec: u8,
1334 dict_id: u8,
1338 plaintext: Option<Vec<u8>>,
1343 flags: u8,
1347}
1348
1349impl PendingDrop {
1350 fn len_in_window(&self) -> u32 {
1354 u32::try_from(self.compressed.len()).expect("compressed fits u32")
1355 }
1356
1357 fn plaintext_len_value(&self) -> u32 {
1359 self.plaintext_len
1360 }
1361
1362 fn slab_footprint(&self) -> usize {
1365 48 + self.compressed.len()
1366 }
1367}
1368
1369struct PendingSlice {
1374 drop_id: [u8; 32],
1375 file_byte_start: u64,
1376 file_byte_end: u64,
1377}
1378
1379#[derive(Clone)]
1382struct PendingFile {
1383 inode_number: u64,
1384 path: PathBuf,
1385 mtime_ns: u64,
1386 file_len: u64,
1387}
1388
1389#[derive(Clone, Copy, Default)]
1393struct SurveyMeta {
1394 is_dir: bool,
1395 is_file: bool,
1396 is_symlink: bool,
1397 #[cfg(unix)]
1398 is_fifo: bool,
1399 #[cfg(unix)]
1400 is_socket: bool,
1401 #[cfg(unix)]
1402 is_block_device: bool,
1403 #[cfg(unix)]
1404 is_char_device: bool,
1405 len: u64,
1406 mtime_ns: u64,
1407}
1408
1409struct SurveyNode {
1412 meta: SurveyMeta,
1413 children: Vec<(String, SurveyNode)>,
1414 symlink_target: Option<String>,
1418}
1419
1420impl SurveyNode {
1421 fn meta(&self) -> SurveyMeta {
1422 self.meta
1423 }
1424}
1425
1426fn survey_meta_of(meta: &std::fs::Metadata) -> SurveyMeta {
1427 #[cfg(unix)]
1428 use std::os::unix::fs::FileTypeExt as _;
1429 let ft = meta.file_type();
1430 let mtime_ns = meta
1431 .modified()
1432 .ok()
1433 .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
1434 .map_or(0u128, |d| d.as_nanos());
1435 SurveyMeta {
1436 is_dir: ft.is_dir(),
1437 is_file: ft.is_file(),
1438 is_symlink: ft.is_symlink(),
1439 #[cfg(unix)]
1440 is_fifo: ft.is_fifo(),
1441 #[cfg(unix)]
1442 is_socket: ft.is_socket(),
1443 #[cfg(unix)]
1444 is_block_device: ft.is_block_device(),
1445 #[cfg(unix)]
1446 is_char_device: ft.is_char_device(),
1447 len: meta.len(),
1448 mtime_ns: mtime_ns.try_into().unwrap_or(0),
1449 }
1450}
1451
1452fn survey_node(path: &Path) -> Result<SurveyNode, WriteError> {
1453 use rayon::prelude::*;
1454 let meta = std::fs::symlink_metadata(path)?;
1455 let sm = survey_meta_of(&meta);
1456 if sm.is_symlink {
1457 let target = std::fs::read_link(path)?;
1458 let target = target
1459 .to_str()
1460 .ok_or_else(|| WriteError::UnsupportedFileType {
1461 path: path.to_path_buf(),
1462 kind: format!("symlink with non-UTF-8 target ({})", target.display()),
1463 })?
1464 .to_owned();
1465 return Ok(SurveyNode {
1466 meta: sm,
1467 children: Vec::new(),
1468 symlink_target: Some(target),
1469 });
1470 }
1471 if !sm.is_dir {
1472 return Ok(SurveyNode {
1473 meta: sm,
1474 children: Vec::new(),
1475 symlink_target: None,
1476 });
1477 }
1478 let mut named: Vec<(String, PathBuf)> = std::fs::read_dir(path)?
1479 .filter_map(|entry| {
1480 entry
1481 .ok()
1482 .map(|e| (e.file_name().to_string_lossy().into_owned(), e.path()))
1483 })
1484 .collect();
1485 named.sort_by(|a, b| a.0.cmp(&b.0));
1486 named
1487 .par_iter()
1488 .map(|(name, child)| survey_node(child).map(|node| (name.clone(), node)))
1489 .collect::<Result<Vec<_>, WriteError>>()
1490 .map(|children| SurveyNode {
1491 meta: sm,
1492 children,
1493 symlink_target: None,
1494 })
1495}
1496
1497fn survey_tree(root: &Path) -> Result<SurveyNode, WriteError> {
1502 survey_node(root)
1503}
1504
1505struct PendingInode {
1506 number: u64,
1507 mode: u32,
1508 mtime_ns: u64,
1509 content: PendingContent,
1510}
1511
1512enum PendingContent {
1513 Inline(Vec<u8>),
1514 Symlink(String),
1516 DropBacked {
1517 file_len: u64,
1518 slices: Vec<PendingSlice>,
1519 },
1520 Directory(Vec<(String, u64, u8)>),
1521}
1522
1523struct DirNode {
1524 entries: Vec<(String, u64, u8)>,
1525 bytes: Vec<u8>,
1526 hash: [u8; 32],
1527}
1528
1529struct WriteContext {
1530 next_inode: u64,
1531 inodes: Vec<PendingInode>,
1532 dir_nodes: Vec<DirNode>,
1533 drops: Vec<PendingDrop>,
1534 drop_index: HashSet<[u8; 32]>,
1535 pending_files: Vec<PendingFile>,
1536 file_count: usize,
1537 dir_count: usize,
1538 root_inode_number: u64,
1539 chunker: ParallelFastCDC,
1540 classifier: classifier::Classifier,
1541 shared_inline_map: HashMap<[u8; 32], usize>,
1542 shared_inline_table: Vec<Vec<u8>>,
1543 base_dictionaries: Option<Vec<crate::dictionary::TrainedDictionary>>,
1549 profile_name: Option<String>,
1551 metadata_codec: u8,
1554 categorizers_disabled: bool,
1556 rw_mode: bool,
1558 auto_turnover: bool,
1560 collect_dict_samples: bool,
1563 dict_samples_by_class: HashMap<crate::classifier::Class, Vec<Vec<u8>>>,
1570 trained_dicts_by_class: HashMap<crate::classifier::Class, crate::dictionary::TrainedDictionary>,
1575 base_drop_index: Option<std::sync::Arc<dyn BaseDropSet>>,
1581 base_root: Option<[u8; 32]>,
1586 metadata_externalize_threshold: usize,
1591 emit_shared_inline: bool,
1597 inline_threshold: usize,
1602 pending_sink: Option<std::sync::mpsc::SyncSender<PendingFile>>,
1608}
1609
1610impl WriteContext {
1611 const MAX_DICT_SAMPLES: usize = 1000;
1614
1615 fn new() -> Self {
1616 Self {
1617 next_inode: 1,
1618 inodes: Vec::new(),
1619 dir_nodes: Vec::new(),
1620 drops: Vec::new(),
1621 drop_index: HashSet::new(),
1622 pending_files: Vec::new(),
1623 file_count: 0,
1624 dir_count: 0,
1625 root_inode_number: 0,
1626 chunker: ParallelFastCDC::default(),
1627 classifier: classifier::Classifier,
1628 shared_inline_map: HashMap::new(),
1629 shared_inline_table: Vec::new(),
1630 base_dictionaries: None,
1631 profile_name: None,
1632 metadata_codec: limnifs_core::codec::CODEC_BROTLI,
1633 categorizers_disabled: false,
1634 rw_mode: false,
1635 auto_turnover: false,
1636 collect_dict_samples: false,
1637 dict_samples_by_class: HashMap::new(),
1638 trained_dicts_by_class: HashMap::new(),
1639 base_drop_index: None,
1640 base_root: None,
1641 pending_sink: None,
1642 inline_threshold: INLINE_THRESHOLD,
1643 metadata_externalize_threshold: METADATA_EXTERNALIZE_THRESHOLD,
1644 emit_shared_inline: true,
1645 }
1646 }
1647
1648 fn alloc_inode(&mut self) -> u64 {
1649 let n = self.next_inode;
1650 self.next_inode += 1;
1651 n
1652 }
1653
1654 fn build_shared_inline_table(&mut self) {
1658 let mut counts: HashMap<[u8; 32], usize> = HashMap::new();
1659 for inode in &self.inodes {
1660 if let PendingContent::Inline(data) = &inode.content {
1661 let h = hash_section(data);
1662 *counts.entry(h).or_default() += 1;
1663 }
1664 }
1665 for inode in &self.inodes {
1667 if let PendingContent::Inline(data) = &inode.content {
1668 let h = hash_section(data);
1669 if counts.get(&h).copied().unwrap_or(0) > 1
1670 && !self.shared_inline_map.contains_key(&h)
1671 {
1672 let idx = self.shared_inline_table.len();
1673 self.shared_inline_table.push(data.clone());
1674 self.shared_inline_map.insert(h, idx);
1675 }
1676 }
1677 }
1678 }
1679
1680 fn merge_chunked_file(&mut self, pf: &PendingFile, result: ChunkedFileResult) {
1683 for (drop_id, plaintext, compressed, codec, flags) in result.drops {
1684 if self.drop_index.insert(drop_id) {
1685 let retain_plaintext =
1691 self.collect_dict_samples && codec == limnifs_core::codec::CODEC_ZSTD;
1692 if retain_plaintext {
1693 let total: usize = self.dict_samples_by_class.values().map(Vec::len).sum();
1694 if total < Self::MAX_DICT_SAMPLES {
1695 let class = self.classifier.classify(&plaintext);
1696 self.dict_samples_by_class
1697 .entry(class)
1698 .or_default()
1699 .push(plaintext.clone());
1700 }
1701 }
1702 self.drops.push(PendingDrop {
1703 id: drop_id,
1704 plaintext_len: u32::try_from(plaintext.len()).unwrap_or(u32::MAX),
1705 compressed,
1706 codec,
1707 dict_id: limnifs_core::drop_record::NO_DICT,
1708 plaintext: if retain_plaintext {
1709 Some(plaintext)
1710 } else {
1711 None
1712 },
1713 flags,
1714 });
1715 }
1716 }
1717 self.inodes.push(PendingInode {
1718 number: pf.inode_number,
1719 mode: 0o100_644,
1720 mtime_ns: pf.mtime_ns,
1721 content: PendingContent::DropBacked {
1722 file_len: pf.file_len,
1723 slices: result.slices,
1724 },
1725 });
1726 }
1727
1728 #[allow(dead_code)]
1739 fn deepen_drop(&self, drop_id: [u8; 32], plaintext: &[u8]) -> PendingDrop {
1740 let class = self.classifier.classify(plaintext);
1741 let (codec, compressed): (u8, std::sync::Arc<[u8]>) = match class {
1742 classifier::Class::Text | classifier::Class::Code | classifier::Class::Binary => {
1743 let c = limnifs_core::codec::compress_lz4_with_size(plaintext);
1744 (limnifs_core::codec::CODEC_LZ4, c.into())
1745 }
1746 _ => (limnifs_core::codec::CODEC_STORE, plaintext.to_vec().into()),
1747 };
1748 PendingDrop {
1749 id: drop_id,
1750 plaintext_len: u32::try_from(plaintext.len()).unwrap_or(u32::MAX),
1751 compressed,
1752 codec,
1753 dict_id: limnifs_core::drop_record::NO_DICT,
1754 plaintext: None,
1755 flags: 0,
1756 }
1757 }
1758
1759 fn walk(&mut self, path: &Path) -> Result<u64, WriteError> {
1760 let survey = survey_tree(path)?;
1768 self.fold_survey(path, &survey, None)
1769 }
1770
1771 fn fold_survey(
1775 &mut self,
1776 path: &Path,
1777 node: &SurveyNode,
1778 symlink_target: Option<&str>,
1779 ) -> Result<u64, WriteError> {
1780 let meta = node.meta();
1781 if let Some(target) = symlink_target {
1782 let inode_number = self.alloc_inode();
1783 self.inodes.push(PendingInode {
1784 number: inode_number,
1785 mode: limnifs_core::inode::S_IFLNK | 0o777,
1786 mtime_ns: meta.mtime_ns,
1787 content: PendingContent::Symlink(target.to_owned()),
1788 });
1789 return Ok(inode_number);
1790 }
1791 if meta.is_dir {
1792 self.dir_count += 1;
1793 let inode_number = self.alloc_inode();
1794 let mut entries: Vec<(String, u64, u8)> = Vec::new();
1795
1796 for (name, child) in &node.children {
1797 let child_path = path.join(name);
1798 let child_inode =
1799 self.fold_survey(&child_path, child, child.symlink_target.as_deref())?;
1800 let entry_type = if child.meta().is_symlink {
1801 0x03
1802 } else if child.meta().is_dir {
1803 0x02
1804 } else {
1805 0x01
1806 };
1807 entries.push((name.clone(), child_inode, entry_type));
1808 }
1809
1810 entries.sort_by(|a, b| a.0.cmp(&b.0));
1813 let dir_node = encode_dir_node(&entries);
1814 self.dir_nodes.push(dir_node);
1815 self.inodes.push(PendingInode {
1816 number: inode_number,
1817 mode: 0o040_755,
1818 mtime_ns: meta.mtime_ns,
1819 content: PendingContent::Directory(entries),
1820 });
1821 Ok(inode_number)
1822 } else if meta.is_file {
1823 self.file_count += 1;
1824 let inode_number = self.alloc_inode();
1825 let file_len = meta.len;
1826 crate::progress::emit_file(path, file_len);
1827
1828 if file_len <= u64::try_from(self.inline_threshold).unwrap_or(u64::MAX) {
1829 let data = std::fs::read(path)?;
1830 self.inodes.push(PendingInode {
1831 number: inode_number,
1832 mode: 0o100_644,
1833 mtime_ns: meta.mtime_ns,
1834 content: PendingContent::Inline(data),
1835 });
1836 } else {
1837 let pf = PendingFile {
1839 inode_number,
1840 path: path.to_path_buf(),
1841 mtime_ns: meta.mtime_ns,
1842 file_len,
1843 };
1844 if let Some(sink) = &self.pending_sink {
1845 sink.send(pf).map_err(|_| {
1851 WriteError::Io(std::io::Error::other("walk: compress pipeline shut down"))
1852 })?;
1853 } else {
1854 self.pending_files.push(pf);
1855 }
1856 }
1857 Ok(inode_number)
1858 } else {
1859 #[cfg(unix)]
1860 let kind = {
1861 use std::os::unix::fs::FileTypeExt;
1862 if meta.is_fifo {
1863 "fifo".to_owned()
1864 } else if meta.is_socket {
1865 "socket".to_owned()
1866 } else if meta.is_block_device {
1867 "block device".to_owned()
1868 } else if meta.is_char_device {
1869 "character device".to_owned()
1870 } else {
1871 "unknown".to_owned()
1872 }
1873 };
1874 #[cfg(not(unix))]
1875 let kind = "unknown".to_owned();
1876 Err(WriteError::UnsupportedFileType {
1877 path: path.to_path_buf(),
1878 kind,
1879 })
1880 }
1881 }
1882
1883 fn train_and_apply_dictionary(&mut self, dictionaries: &crate::config::DictionaryConfig) {
1898 if !dictionaries.enabled {
1899 Self::release_dictionary_samples(self);
1900 return;
1901 }
1902
1903 if let Some(adopted) = self.base_dictionaries.take() {
1907 for dict in adopted {
1908 match dict.id {
1909 0 => {
1910 self.trained_dicts_by_class
1911 .insert(crate::classifier::Class::Text, dict);
1912 }
1913 1 => {
1914 self.trained_dicts_by_class
1915 .insert(crate::classifier::Class::Binary, dict);
1916 }
1917 _ => {}
1918 }
1919 }
1920 self.apply_trained_dictionaries();
1921 return;
1922 }
1923
1924 let target = usize::try_from(dictionaries.max_dict_size).unwrap_or(65_536);
1925 let min_class = usize::try_from(dictionaries.min_class_size).unwrap_or(0);
1926
1927 let text_classes = [
1931 crate::classifier::Class::Text,
1932 crate::classifier::Class::Code,
1933 crate::classifier::Class::Sparse,
1934 ];
1935 let binary_classes = [crate::classifier::Class::Binary];
1936
1937 let text_samples: Vec<&[u8]> = text_classes
1939 .iter()
1940 .flat_map(|c| self.dict_samples_by_class.get(c).into_iter().flatten())
1941 .map(Vec::as_slice)
1942 .collect();
1943 let trainer = crate::dictionary::TrainerKind::from_config_str(&dictionaries.trainer);
1944 if text_samples.len() >= min_class {
1945 if let Some(dict) =
1946 crate::dictionary::train_zstd_with_trainer(0, &text_samples, target, trainer)
1947 {
1948 self.trained_dicts_by_class
1949 .insert(crate::classifier::Class::Text, dict);
1950 }
1951 }
1952 let binary_samples: Vec<&[u8]> = binary_classes
1953 .iter()
1954 .flat_map(|c| self.dict_samples_by_class.get(c).into_iter().flatten())
1955 .map(Vec::as_slice)
1956 .collect();
1957 if binary_samples.len() >= min_class {
1958 if let Some(dict) =
1959 crate::dictionary::train_zstd_with_trainer(1, &binary_samples, target, trainer)
1960 {
1961 self.trained_dicts_by_class
1962 .insert(crate::classifier::Class::Binary, dict);
1963 }
1964 }
1965
1966 self.apply_trained_dictionaries();
1967 }
1968
1969 fn apply_trained_dictionaries(&mut self) {
1975 let text_classes = [
1979 crate::classifier::Class::Text,
1980 crate::classifier::Class::Code,
1981 crate::classifier::Class::Sparse,
1982 ];
1983 let binary_classes = [crate::classifier::Class::Binary];
1984
1985 use rayon::prelude::*;
1998 let classifier = self.classifier;
1999 let dicts = &self.trained_dicts_by_class;
2000 let candidates: Vec<Option<(std::sync::Arc<[u8]>, u8)>> = self
2001 .drops
2002 .par_iter()
2003 .map(|d| {
2004 if d.codec != limnifs_core::codec::CODEC_ZSTD {
2005 return None;
2006 }
2007 let Some(plaintext) = d.plaintext.as_ref() else {
2008 return None;
2009 };
2010 let class = classifier.classify(plaintext);
2011 let dict_class = if text_classes.contains(&class) {
2012 crate::classifier::Class::Text
2013 } else if binary_classes.contains(&class) {
2014 crate::classifier::Class::Binary
2015 } else {
2016 return None;
2017 };
2018 let Some(dict) = dicts.get(&dict_class) else {
2019 return None;
2020 };
2021 let Ok(dict_compressed) = dict.compress(plaintext) else {
2022 return None;
2023 };
2024 if dict_compressed.len() < d.compressed.len() {
2025 Some((dict_compressed.into(), dict.id))
2026 } else {
2027 None
2028 }
2029 })
2030 .collect();
2031
2032 let saving: isize = candidates
2033 .iter()
2034 .zip(self.drops.iter())
2035 .map(|(c, d)| {
2036 c.as_ref().map_or(0, |(bytes, _)| {
2037 d.compressed.len() as isize - bytes.len() as isize
2038 })
2039 })
2040 .sum();
2041 let dict_bytes: usize = dicts.values().map(|d| d.content.len()).sum();
2042 if saving > dict_bytes as isize {
2043 for (d, candidate) in self.drops.iter_mut().zip(candidates) {
2044 if let Some((bytes, dict_id)) = candidate {
2045 d.compressed = bytes;
2046 d.dict_id = dict_id;
2047 }
2048 }
2049 } else {
2050 self.trained_dicts_by_class.clear();
2056 }
2057
2058 Self::release_dictionary_samples(self);
2059 }
2060
2061 fn release_dictionary_samples(ctx: &mut Self) {
2064 for d in &mut ctx.drops {
2065 d.plaintext = None;
2066 }
2067 ctx.dict_samples_by_class.clear();
2068 }
2069
2070 fn trace_phase(label: &str, start: std::time::Instant) {
2072 if std::env::var_os("LIMNIFS_TRACE_ASSEMBLE").is_some() {
2073 eprintln!("[assemble] {label}: {:?}", start.elapsed());
2074 }
2075 }
2076
2077 fn assemble(mut self) -> WriteArtifact {
2078 let t_assemble = std::time::Instant::now();
2079 let inode_count = self.inodes.len();
2080 let dir_count = self.dir_count;
2081 let drop_count = self.drops.len();
2082
2083 let t = std::time::Instant::now();
2089 let slabs = pack_slabs(&self.drops);
2090 Self::trace_phase("pack_slabs", t);
2091
2092 let t = std::time::Instant::now();
2096 if self.emit_shared_inline {
2097 self.build_shared_inline_table();
2098 }
2099 Self::trace_phase("shared_inline_table", t);
2100
2101 let mut metadata_blob = Vec::new();
2102 metadata_blob.extend_from_slice(&u32::try_from(self.inodes.len()).unwrap().to_le_bytes());
2103 for inode in &self.inodes {
2104 self.encode_inode(&mut metadata_blob, inode);
2105 }
2106 metadata_blob
2107 .extend_from_slice(&u32::try_from(self.dir_nodes.len()).unwrap().to_le_bytes());
2108 for node in &self.dir_nodes {
2109 metadata_blob.extend_from_slice(&node.bytes);
2110 }
2111 if !self.shared_inline_table.is_empty() {
2114 metadata_blob.extend_from_slice(
2115 &u32::try_from(self.shared_inline_table.len())
2116 .unwrap()
2117 .to_le_bytes(),
2118 );
2119 for entry in &self.shared_inline_table {
2120 let len = u32::try_from(entry.len()).expect("shared entry fits u32");
2121 metadata_blob.extend_from_slice(&len.to_le_bytes());
2122 metadata_blob.extend_from_slice(entry);
2123 }
2124 }
2125
2126 Self::trace_phase("metadata_encode", t);
2127 let uncompressed_len =
2134 u32::try_from(metadata_blob.len()).expect("metadata blob length fits u32");
2135 let t = std::time::Instant::now();
2136 let metadata_hash = hash_section(&metadata_blob);
2137 let metadata_codec = self.metadata_codec;
2138 let metadata_quality = if metadata_blob.len() > METADATA_LARGE_BLOB_THRESHOLD {
2139 METADATA_LARGE_BLOB_QUALITY
2140 } else {
2141 METADATA_SMALL_BLOB_QUALITY
2142 };
2143 let compressed_blob = if metadata_codec == limnifs_core::codec::CODEC_BROTLI {
2144 limnifs_core::codec::compress_brotli_with_quality(&metadata_blob, metadata_quality)
2145 .unwrap_or_else(|_| metadata_blob.clone())
2146 } else {
2147 limnifs_core::codec::compress(metadata_codec, &metadata_blob)
2148 .unwrap_or_else(|_| metadata_blob.clone())
2149 };
2150 Self::trace_phase("metadata_compress", t);
2151 let (on_wire_codec, on_wire_blob) = if compressed_blob.len() < metadata_blob.len() {
2152 (metadata_codec, compressed_blob)
2153 } else {
2154 (limnifs_core::codec::CODEC_STORE, metadata_blob.clone())
2155 };
2156
2157 let externalize_at = self
2161 .metadata_externalize_threshold
2162 .min(limnifs_core::metadata_reference::DEFAULT_INLINE_METADATA_MAX_BYTES as usize);
2163 let (metadata_sidecar, inline_data, metadata_locator_count) =
2164 if on_wire_blob.len() > externalize_at {
2165 let h = hash_section(&on_wire_blob);
2172 let mut h8 = String::with_capacity(8);
2173 for b in &h[..4] {
2174 h8.push_str(&format!("{b:02x}"));
2175 }
2176 let locator = format!("file:metadata-{h8}.bin");
2177 let sidecar = MetadataSidecar {
2178 bytes: on_wire_blob.clone(),
2179 locator,
2180 };
2181 (Some(sidecar), None, 1u32)
2182 } else {
2183 (None, Some(on_wire_blob.clone()), 0u32)
2184 };
2185
2186 let mut manifest = Vec::new();
2187
2188 let header_start = manifest.len();
2189 manifest.extend_from_slice(&ManifestHeader::current().to_bytes());
2190 let header_end = manifest.len();
2191
2192 let flags_start = manifest.len();
2193 manifest.push(FEATURE_FLAGS_SECTION_VERSION);
2194 manifest.extend_from_slice(&0u32.to_le_bytes());
2195 let flags_end = manifest.len();
2196
2197 let meta_ref_start = manifest.len();
2200 manifest.push(METADATA_REFERENCE_SECTION_VERSION_2);
2201 manifest.extend_from_slice(&metadata_hash);
2202 manifest.extend_from_slice(&uncompressed_len.to_le_bytes());
2203 manifest.push(on_wire_codec);
2204 manifest.extend_from_slice(&metadata_locator_count.to_le_bytes());
2205 if let Some(sidecar) = &metadata_sidecar {
2206 let loc_bytes = sidecar.locator.as_bytes();
2207 let loc_len = u32::try_from(loc_bytes.len()).expect("locator fits u32");
2208 manifest.extend_from_slice(&loc_len.to_le_bytes());
2209 manifest.extend_from_slice(loc_bytes);
2210 }
2211 match &inline_data {
2212 Some(blob) => {
2213 let inline_len = u32::try_from(blob.len()).expect("metadata fits u32");
2214 manifest.extend_from_slice(&inline_len.to_le_bytes());
2215 manifest.extend_from_slice(blob);
2216 }
2217 None => {
2218 manifest.extend_from_slice(&0u32.to_le_bytes());
2219 }
2220 }
2221 let meta_ref_end = manifest.len();
2222
2223 let slab_index_start = manifest.len();
2224 manifest.push(SLAB_INDEX_SECTION_VERSION);
2225 manifest.extend_from_slice(&u32::try_from(slabs.len()).unwrap().to_le_bytes());
2226 for slab in &slabs {
2227 manifest.extend_from_slice(&slab.id.to_bytes());
2228 manifest.extend_from_slice(&1u32.to_le_bytes());
2229 let loc_bytes = slab.locator.as_bytes();
2230 let loc_len = u32::try_from(loc_bytes.len()).expect("locator fits u32");
2231 manifest.extend_from_slice(&loc_len.to_le_bytes());
2232 manifest.extend_from_slice(loc_bytes);
2233 }
2234 let slab_index_end = manifest.len();
2235
2236 let history_start = manifest.len();
2237 manifest.push(HISTORY_SECTION_VERSION);
2238 manifest.extend_from_slice(&1u32.to_le_bytes());
2239 manifest.push(0x01);
2240 manifest.extend_from_slice(&0u64.to_le_bytes());
2241 manifest.extend_from_slice(&0u32.to_le_bytes());
2242 manifest.extend_from_slice(&0u32.to_le_bytes());
2243 let history_end = manifest.len();
2244
2245 let profile_desc_start = manifest.len();
2250 if let Some(ref name) = self.profile_name {
2251 let desc = limnifs_core::profile_descriptor::ProfileDescriptor {
2252 version: limnifs_core::profile_descriptor::PROFILE_DESCRIPTOR_SECTION_VERSION,
2253 profile_name: Some(name.clone()),
2254 blake3_hashing: true,
2255 cross_file_dedup: true,
2256 content_classification: !self.categorizers_disabled,
2257 integrity_verify: true,
2258 read_write: self.rw_mode,
2259 auto_turnover: self.auto_turnover,
2260 };
2261 limnifs_core::profile_descriptor::encode_profile_descriptor(&desc, &mut manifest);
2262 }
2263 let profile_desc_end = manifest.len();
2264
2265 if !self.trained_dicts_by_class.is_empty() {
2271 let dicts: Vec<_> = self
2272 .trained_dicts_by_class
2273 .values()
2274 .map(|d| limnifs_core::dictionary_section::Dictionary {
2275 codec_id: d.codec,
2276 class_id: d.id,
2277 data: d.content.clone(),
2278 })
2279 .collect();
2280 let section = limnifs_core::dictionary_section::DictionarySection {
2281 version: limnifs_core::dictionary_section::DICTIONARY_SECTION_VERSION,
2282 dicts,
2283 };
2284 limnifs_core::dictionary_section::encode_dictionary_section(§ion, &mut manifest);
2285 }
2286
2287 let dictionary_end = manifest.len();
2288
2289 let delta_linkage_hash = if let Some(base_root) = self.base_root {
2295 let delta_start = manifest.len();
2296 manifest.push(limnifs_core::delta_linkage::DELTA_LINKAGE_SECTION_VERSION);
2302 manifest.extend_from_slice(&base_root);
2303 manifest.extend_from_slice(&0u32.to_le_bytes());
2304 hash_section(&manifest[delta_start..])
2305 } else {
2306 hash_empty_section()
2307 };
2308 let _ = dictionary_end;
2309
2310 let hashes = SectionHashes {
2311 metadata: metadata_hash,
2312 format_header: hash_section(&manifest[header_start..header_end]),
2313 feature_flags: hash_section(&manifest[flags_start..flags_end]),
2314 metadata_reference: hash_section(&manifest[meta_ref_start..meta_ref_end]),
2315 slab_index: hash_section(&manifest[slab_index_start..slab_index_end]),
2316 crypto_params: hash_empty_section(),
2317 ec_params: hash_empty_section(),
2318 dms_policy: hash_empty_section(),
2319 delta_linkage: delta_linkage_hash,
2320 history: hash_section(&manifest[history_start..history_end]),
2321 };
2333 let merkle_root = compute_merkle_root(&hashes);
2334
2335 WriteArtifact {
2336 bytes: manifest,
2337 merkle_root,
2338 slabs,
2339 metadata_sidecar,
2340 inode_count,
2341 file_count: self.file_count,
2342 dir_count,
2343 drop_count,
2344 root_inode_number: self.root_inode_number,
2345 }
2346 }
2347
2348 fn encode_inode(&self, out: &mut Vec<u8>, inode: &PendingInode) {
2349 out.extend_from_slice(&inode.number.to_le_bytes());
2350 out.extend_from_slice(&inode.mode.to_le_bytes());
2351 out.extend_from_slice(&0u32.to_le_bytes());
2352 out.extend_from_slice(&0u32.to_le_bytes());
2353 out.extend_from_slice(&inode.mtime_ns.to_le_bytes());
2354 out.extend_from_slice(&inode.mtime_ns.to_le_bytes());
2355 out.extend_from_slice(&1u32.to_le_bytes());
2356 match &inode.content {
2357 PendingContent::Inline(data) => {
2358 let h = hash_section(data);
2359 if let Some(&idx) = self.shared_inline_map.get(&h) {
2360 out.push(INODE_FLAG_INLINE_DATA | INODE_FLAG_SHARED_INLINE);
2362 out.extend_from_slice(&(idx as u32).to_le_bytes());
2363 } else {
2364 out.push(INODE_FLAG_INLINE_DATA);
2365 let len = u32::try_from(data.len()).expect("data fits u32");
2366 out.extend_from_slice(&len.to_le_bytes());
2367 out.extend_from_slice(data);
2368 }
2369 }
2370 PendingContent::DropBacked { file_len, slices } => {
2371 out.push(0x00);
2372 let slice_count = u32::try_from(slices.len()).expect("slice count fits u32");
2373 out.extend_from_slice(&slice_count.to_le_bytes());
2374 for slice in slices {
2375 out.extend_from_slice(&slice.file_byte_start.to_le_bytes());
2376 out.extend_from_slice(&slice.file_byte_end.to_le_bytes());
2377 out.extend_from_slice(&slice.drop_id);
2378 out.extend_from_slice(&0u32.to_le_bytes());
2380 let drop_byte_len = u32::try_from(slice.file_byte_end - slice.file_byte_start)
2384 .expect("slice range fits u32");
2385 out.extend_from_slice(&drop_byte_len.to_le_bytes());
2386 }
2387 let _ = file_len;
2388 }
2389 PendingContent::Symlink(target) => {
2390 out.push(0x00);
2394 let t = target.as_bytes();
2395 let len = u32::try_from(t.len()).expect("target fits u32");
2396 out.extend_from_slice(&len.to_le_bytes());
2397 out.extend_from_slice(t);
2398 }
2399 PendingContent::Directory(entries) => {
2400 out.push(0x00);
2401 let node = self
2402 .dir_nodes
2403 .iter()
2404 .find(|n| n.entries == *entries)
2405 .expect("directory node must exist");
2406 out.extend_from_slice(&node.hash);
2407 }
2408 }
2409 }
2410}
2411
2412fn sidecar_name(locator: &str) -> Result<&str, WriteError> {
2418 limnifs_core::locator::local_sidecar_name(locator)
2419 .map_err(|e| WriteError::Io(std::io::Error::other(format!("{e}"))))
2420}
2421
2422fn encode_dir_node(entries: &[(String, u64, u8)]) -> DirNode {
2423 let mut bytes = Vec::new();
2424 bytes.push(1u8);
2425 let count = u32::try_from(entries.len()).expect("entry count fits u32");
2426 bytes.extend_from_slice(&count.to_le_bytes());
2427 for (name, inode_number, entry_type) in entries {
2428 let name_bytes = name.as_bytes();
2429 let name_len = u32::try_from(name_bytes.len()).expect("name fits u32");
2430 bytes.extend_from_slice(&name_len.to_le_bytes());
2431 bytes.extend_from_slice(name_bytes);
2432 bytes.extend_from_slice(&inode_number.to_le_bytes());
2433 bytes.push(*entry_type);
2434 }
2435 let hash = hash_section(&bytes);
2436 DirNode {
2437 entries: entries.to_vec(),
2438 bytes,
2439 hash,
2440 }
2441}
2442
2443fn pack_slabs(drops: &[PendingDrop]) -> Vec<SlabArtifact> {
2452 let local_drops: Vec<&PendingDrop> = drops
2457 .iter()
2458 .filter(|d| d.codec != limnifs_core::codec::CODEC_REFERENCED)
2459 .collect();
2460 if local_drops.is_empty() {
2461 return Vec::new();
2462 }
2463
2464 let max_content = MAX_SLAB_TOTAL_BYTES.saturating_sub(SLAB_HEADER_LEN);
2465
2466 let mut slab_groups: Vec<Vec<&PendingDrop>> = Vec::new();
2471 let mut current: Vec<&PendingDrop> = Vec::new();
2472 let mut current_size: usize = 0;
2473
2474 for drop in &local_drops {
2475 let footprint = drop.slab_footprint();
2476 if !current.is_empty() && current_size + footprint > max_content {
2477 slab_groups.push(std::mem::take(&mut current));
2478 current_size = 0;
2479 }
2480 current.push(*drop);
2481 current_size += footprint;
2482 }
2483 if !current.is_empty() {
2484 slab_groups.push(current);
2485 }
2486
2487 use rayon::prelude::*;
2493 slab_groups
2494 .par_iter()
2495 .enumerate()
2496 .map(|(ordinal, group)| {
2497 let ordinal_u64 = u64::try_from(ordinal).expect("slab count fits u64");
2498 encode_slab(ordinal_u64, group)
2499 })
2500 .collect()
2501}
2502
2503fn encode_slab(ordinal: u64, drops: &[&PendingDrop]) -> SlabArtifact {
2507 const DROP_RECORD_LEN: usize = 50;
2514 let mut drop_records = Vec::with_capacity(drops.len() * DROP_RECORD_LEN);
2515 let mut solid_window = Vec::new();
2516 let mut drop_ids = Vec::with_capacity(drops.len());
2517 let mut offset_in_window: u32 = 0;
2518
2519 for drop in drops {
2520 let plaintext_len = drop.plaintext_len_value();
2521 let window_len = drop.len_in_window();
2522 drop_records.extend_from_slice(&drop.id);
2523 drop_records.extend_from_slice(&plaintext_len.to_le_bytes());
2524 drop_records.extend_from_slice(&[drop.codec, 0x00, 0x00]);
2526 drop_records.push(0x00); drop_records.extend_from_slice(&offset_in_window.to_le_bytes());
2528 drop_records.extend_from_slice(&window_len.to_le_bytes());
2529 drop_records.push(drop.dict_id); drop_records.push(drop.flags); solid_window.extend_from_slice(&drop.compressed);
2532 drop_ids.push(drop.id);
2533 offset_in_window = offset_in_window
2534 .checked_add(window_len)
2535 .expect("slab window size fits u32");
2536 }
2537
2538 let slab_content = [&drop_records[..], &solid_window[..]].concat();
2539 let slab_hash = hash_section(&slab_content);
2540 let slab_id = SlabId::new(ordinal, slab_hash);
2541
2542 let total_length = SLAB_HEADER_LEN + slab_content.len();
2543 let mut slab_bytes = Vec::with_capacity(total_length);
2544 slab_bytes.extend_from_slice(b"LIM1");
2545 slab_bytes.extend_from_slice(&1u16.to_le_bytes()); slab_bytes.extend_from_slice(&slab_id.to_bytes());
2547 slab_bytes.extend_from_slice(
2548 &u64::try_from(total_length)
2549 .unwrap_or(u64::MAX)
2550 .to_le_bytes(),
2551 );
2552 slab_bytes.push(0x00);
2553 slab_bytes.push(0x00);
2554 slab_bytes.extend_from_slice(&slab_content);
2555
2556 let mut h8 = String::with_capacity(8);
2560 for b in &slab_id.hash[..4] {
2561 h8.push_str(&format!("{b:02x}"));
2562 }
2563 let locator = format!("file:slab-{ordinal}-{h8}.bin");
2564
2565 SlabArtifact {
2566 id: slab_id,
2567 bytes: slab_bytes,
2568 locator,
2569 drop_ids,
2570 }
2571}
2572
2573#[cfg(test)]
2574fn pseudo_random_bytes(seed: u64, count: usize) -> Vec<u8> {
2575 let mut state = seed;
2576 let mut out = Vec::with_capacity(count);
2577 for _ in 0..count {
2578 state = state
2579 .wrapping_mul(6_364_136_223_846_793_005)
2580 .wrapping_add(1_442_695_040_888_963_407);
2581 out.push(u8::try_from(state >> 56).expect("fits u8"));
2582 }
2583 out
2584}
2585
2586#[cfg(test)]
2587mod tests {
2588 use super::*;
2589 use limnifs_core::ManifestCursor;
2590
2591 #[test]
2592 fn write_stream_packs_single_named_stream() {
2593 let temp = std::env::temp_dir().join(format!(
2597 "limnifs-write-stream-test-{}-{}",
2598 std::process::id(),
2599 std::time::SystemTime::now()
2600 .duration_since(std::time::UNIX_EPOCH)
2601 .unwrap()
2602 .as_nanos()
2603 ));
2604 std::fs::create_dir_all(&temp).expect("create temp dir");
2605
2606 let content = b"stream test content line\n".repeat(10_000); let cursor = std::io::Cursor::new(content.clone());
2608 let config = WriteConfig::default_v0_1();
2609 let artifact = write_stream("streamed.txt", cursor, &config).expect("write_stream");
2610
2611 assert!(!artifact.bytes.is_empty(), "manifest bytes non-empty");
2612 assert!(!artifact.slabs.is_empty(), "at least one slab produced");
2613 let total_drop_bytes: usize = artifact.slabs.iter().map(|s| s.bytes.len()).sum();
2618 assert!(total_drop_bytes > 0, "drops non-empty");
2619
2620 let _ = std::fs::remove_dir_all(&temp);
2621 }
2622
2623 #[test]
2624 fn write_layer_references_base_drops() {
2625 let temp = std::env::temp_dir().join(format!(
2631 "limnifs-write-layer-test-{}-{}",
2632 std::process::id(),
2633 std::time::SystemTime::now()
2634 .duration_since(std::time::UNIX_EPOCH)
2635 .unwrap()
2636 .as_nanos()
2637 ));
2638 std::fs::create_dir_all(&temp).expect("create temp dir");
2639
2640 let base_dir = temp.join("base");
2642 std::fs::create_dir_all(&base_dir).expect("base dir");
2643 let text = b"layer test content line\n".repeat(50_000); std::fs::write(base_dir.join("shared.txt"), &text).expect("write shared");
2645 std::fs::write(base_dir.join("base-only.txt"), b"only in base").expect("write base-only");
2646
2647 let config = WriteConfig::default_v0_1();
2648 let base_artifact = write_directory_with_config(&base_dir, &config).expect("base");
2649
2650 let base_manifest = temp.join("base.lim");
2651 std::fs::write(&base_manifest, &base_artifact.bytes).expect("write base manifest");
2652 for slab in &base_artifact.slabs {
2653 let slab_name = sidecar_name(&slab.locator).expect("slab locator");
2654 std::fs::write(temp.join(slab_name), &slab.bytes).expect("write base slab");
2655 }
2656
2657 let layer_dir = temp.join("layer");
2659 std::fs::create_dir_all(&layer_dir).expect("layer dir");
2660 std::fs::write(layer_dir.join("shared.txt"), &text).expect("write shared in layer");
2661 std::fs::write(layer_dir.join("new.txt"), b"fresh content in layer").expect("write new");
2662
2663 let layer_artifact = write_layer(&base_manifest, &layer_dir, &config).expect("layer");
2664
2665 let layer_slab_bytes: usize = layer_artifact.slabs.iter().map(|s| s.bytes.len()).sum();
2668 let base_slab_bytes: usize = base_artifact.slabs.iter().map(|s| s.bytes.len()).sum();
2669 assert!(
2670 layer_slab_bytes < base_slab_bytes / 4,
2671 "layer slabs ({}) should be much smaller than base ({}) — layering failed",
2672 layer_slab_bytes,
2673 base_slab_bytes
2674 );
2675
2676 let base_root = base_artifact.merkle_root.as_bytes();
2679 assert!(
2680 layer_artifact
2681 .bytes
2682 .windows(32)
2683 .any(|w| w == base_root.as_slice()),
2684 "layer manifest must contain base's ManifestRoot bytes"
2685 );
2686
2687 let _ = std::fs::remove_dir_all(&temp);
2688 }
2689
2690 #[test]
2691 fn tournament_short_circuits_on_highly_compressible_chunk() {
2692 let chunk = b"hello world ".repeat(500);
2695 let tunables = limnifs_core::codec::CodecTunables::default();
2696 let tournament = TournamentSpec {
2697 codec_ids: vec![
2698 limnifs_core::codec::CODEC_LZ4,
2699 limnifs_core::codec::CODEC_BROTLI,
2700 ],
2701 min_size: 16,
2702 skip_for_binary: false,
2703 short_circuit_permille: 250,
2704 };
2705 let (codec_id, compressed) = compress_chunk_with_tournament(
2706 &chunk,
2707 classifier::Class::Text,
2708 limnifs_core::codec::CODEC_BROTLI,
2709 limnifs_core::codec::CODEC_LZ4,
2710 &tunables,
2711 &tournament,
2712 );
2713 assert_eq!(codec_id, limnifs_core::codec::CODEC_LZ4);
2714 assert!(compressed.len() < chunk.len());
2715 }
2716
2717 #[test]
2718 fn tournament_runs_all_codecs_when_short_circuit_disabled() {
2719 let chunk = b"hello world ".repeat(500);
2725 let tunables = limnifs_core::codec::CodecTunables::default();
2726 let tournament = TournamentSpec {
2727 codec_ids: vec![
2728 limnifs_core::codec::CODEC_LZ4,
2729 limnifs_core::codec::CODEC_BROTLI,
2730 limnifs_core::codec::CODEC_ZSTD,
2731 ],
2732 min_size: 16,
2733 skip_for_binary: false,
2734 short_circuit_permille: 0,
2735 };
2736 let (codec_id, compressed) = compress_chunk_with_tournament(
2737 &chunk,
2738 classifier::Class::Text,
2739 limnifs_core::codec::CODEC_BROTLI,
2740 limnifs_core::codec::CODEC_LZ4,
2741 &tunables,
2742 &tournament,
2743 );
2744 assert!(
2748 codec_id == limnifs_core::codec::CODEC_ZSTD
2749 || codec_id == limnifs_core::codec::CODEC_BROTLI,
2750 "expected ZSTD or Brotli to win, got codec {codec_id}"
2751 );
2752 assert!(compressed.len() < chunk.len());
2753 }
2754
2755 #[test]
2756 fn tournament_skips_for_binary_when_configured() {
2757 let chunk = vec![0u8; 4096];
2758 let tunables = limnifs_core::codec::CodecTunables::default();
2759 let tournament = TournamentSpec {
2760 codec_ids: vec![limnifs_core::codec::CODEC_BROTLI],
2761 min_size: 16,
2762 skip_for_binary: true,
2763 short_circuit_permille: 250,
2764 };
2765 let (codec_id, _compressed) = compress_chunk_with_tournament(
2766 &chunk,
2767 classifier::Class::Binary,
2768 limnifs_core::codec::CODEC_BROTLI,
2769 limnifs_core::codec::CODEC_LZ4,
2770 &tunables,
2771 &tournament,
2772 );
2773 assert_eq!(codec_id, limnifs_core::codec::CODEC_LZ4);
2775 }
2776
2777 #[test]
2778 fn tournament_small_chunk_uses_preferred_codec() {
2779 let chunk = b"tiny";
2780 let tunables = limnifs_core::codec::CodecTunables::default();
2781 let tournament = TournamentSpec {
2782 codec_ids: vec![limnifs_core::codec::CODEC_BROTLI],
2783 min_size: 1024,
2784 skip_for_binary: false,
2785 short_circuit_permille: 0,
2786 };
2787 let (codec_id, _compressed) = compress_chunk_with_tournament(
2788 chunk,
2789 classifier::Class::Text,
2790 limnifs_core::codec::CODEC_BROTLI,
2791 limnifs_core::codec::CODEC_LZ4,
2792 &tunables,
2793 &tournament,
2794 );
2795 assert_eq!(codec_id, limnifs_core::codec::CODEC_STORE);
2797 }
2798
2799 #[test]
2800 fn tournament_falls_back_to_store_when_no_codec_compresses() {
2801 let chunk = pseudo_random_bytes(42, 4096);
2805 let tunables = limnifs_core::codec::CodecTunables::default();
2806 let tournament = TournamentSpec {
2807 codec_ids: vec![limnifs_core::codec::CODEC_LZ4],
2808 min_size: 16,
2809 skip_for_binary: false,
2810 short_circuit_permille: 0,
2811 };
2812 let (codec_id, compressed) = compress_chunk_with_tournament(
2813 &chunk,
2814 classifier::Class::Binary,
2815 limnifs_core::codec::CODEC_BROTLI,
2816 limnifs_core::codec::CODEC_LZ4,
2817 &tunables,
2818 &tournament,
2819 );
2820 assert_eq!(codec_id, limnifs_core::codec::CODEC_STORE);
2821 assert_eq!(compressed.len(), chunk.len());
2822 }
2823
2824 #[test]
2825 fn dictionaries_enabled_emits_dictionary_section_when_enough_samples() {
2826 let temp = std::env::temp_dir().join(format!(
2831 "limnifs-write-test-{}-dict-{}",
2832 std::process::id(),
2833 std::time::SystemTime::now()
2834 .duration_since(std::time::UNIX_EPOCH)
2835 .map(|d| d.as_nanos() as u64)
2836 .unwrap_or(0),
2837 ));
2838 let _ = std::fs::remove_dir_all(&temp);
2839 std::fs::create_dir_all(&temp).expect("mkdir");
2840
2841 for i in 0..200 {
2844 let content = format!(
2846 "function test_case_{i}() {{ return constant + {i}; }}\n\
2847 // shared comment line {i}\n\
2848 struct Foo {{ x: i32 }} // type {i}\n"
2849 )
2850 .repeat(5);
2851 let path = temp.join(format!("file_{i:04}.txt"));
2852 std::fs::write(&path, content.as_bytes()).expect("write");
2853 }
2854
2855 let mut config = crate::profile::balanced();
2856 config.defaults.text_codec = "zstd".into();
2858 config.defaults.metadata_codec = "zstd".into();
2862 config.dictionaries.enabled = true;
2863 config.dictionaries.min_class_size = 50;
2864 config.dictionaries.max_dict_size = 8192;
2865
2866 let artifact = write_directory_with_config(&temp, &config).expect("write");
2867 std::fs::remove_dir_all(&temp).ok();
2868
2869 let mut cursor = ManifestCursor::new(&artifact.bytes);
2874 let _ = limnifs_core::parse_manifest_header(&mut cursor).expect("header");
2875 let _ = limnifs_core::parse_feature_flags_section(&mut cursor).expect("flags");
2876 let _ = limnifs_core::parse_metadata_reference(&mut cursor).expect("meta_ref");
2877 let _ = limnifs_core::parse_slab_index(&mut cursor).expect("slab_index");
2878 let _ = limnifs_core::parse_history(&mut cursor).expect("history");
2879 let _remaining = cursor.remaining_len();
2882 }
2883
2884 #[test]
2885 fn write_empty_directory() {
2886 let temp =
2887 std::env::temp_dir().join(format!("limnifs-write-test-{}-empty", std::process::id()));
2888 std::fs::create_dir_all(&temp).expect("create temp dir");
2889 let artifact = write_directory(&temp).expect("write succeeds");
2890 std::fs::remove_dir_all(&temp).ok();
2891 assert!(artifact.inode_count >= 1);
2892 assert_eq!(artifact.file_count, 0);
2893 assert_eq!(artifact.dir_count, 1);
2894 assert!(artifact.slabs.is_empty());
2895 }
2896
2897 #[test]
2898 fn write_small_file_inline() {
2899 let temp =
2900 std::env::temp_dir().join(format!("limnifs-write-test-{}-small", std::process::id()));
2901 std::fs::create_dir_all(&temp).expect("create temp dir");
2902 std::fs::write(temp.join("hello.txt"), b"hello world").expect("write file");
2903 let artifact = write_directory(&temp).expect("write succeeds");
2904 std::fs::remove_dir_all(&temp).ok();
2905 assert_eq!(artifact.file_count, 1);
2906 assert!(artifact.slabs.is_empty());
2907 assert_eq!(artifact.drop_count, 0);
2908 }
2909
2910 #[test]
2911 fn write_large_file_uses_slab() {
2912 let temp =
2913 std::env::temp_dir().join(format!("limnifs-write-test-{}-large", std::process::id()));
2914 std::fs::create_dir_all(&temp).expect("create temp dir");
2915 let large_data = vec![0xABu8; INLINE_THRESHOLD + 100];
2916 std::fs::write(temp.join("big.bin"), &large_data).expect("write big");
2917 let artifact = write_directory(&temp).expect("write succeeds");
2918 std::fs::remove_dir_all(&temp).ok();
2919 assert_eq!(artifact.drop_count, 1);
2920 assert_eq!(artifact.slabs.len(), 1);
2921 }
2922
2923 #[test]
2924 fn write_mixed_inline_and_large() {
2925 let temp =
2926 std::env::temp_dir().join(format!("limnifs-write-test-{}-mix", std::process::id()));
2927 std::fs::create_dir_all(&temp).expect("create temp dir");
2928 std::fs::write(temp.join("small.txt"), b"tiny").expect("write small");
2929 std::fs::write(temp.join("large.bin"), vec![0xCDu8; INLINE_THRESHOLD * 2])
2930 .expect("write large");
2931 let artifact = write_directory(&temp).expect("write succeeds");
2932 std::fs::remove_dir_all(&temp).ok();
2933 assert_eq!(artifact.file_count, 2);
2934 assert_eq!(artifact.drop_count, 1);
2935 assert_eq!(artifact.slabs.len(), 1);
2936 }
2937
2938 #[test]
2939 fn deduplicates_identical_large_files() {
2940 let temp =
2941 std::env::temp_dir().join(format!("limnifs-write-test-{}-dedup", std::process::id()));
2942 std::fs::create_dir_all(&temp).expect("create temp dir");
2943 let data = vec![0x77u8; INLINE_THRESHOLD + 10];
2944 std::fs::write(temp.join("a.bin"), &data).expect("write a");
2945 std::fs::write(temp.join("b.bin"), &data).expect("write b");
2946 let artifact = write_directory(&temp).expect("write succeeds");
2947 std::fs::remove_dir_all(&temp).ok();
2948 assert_eq!(artifact.drop_count, 1);
2949 }
2950
2951 #[test]
2952 fn write_and_verify_roundtrip() {
2953 let temp = std::env::temp_dir().join(format!(
2954 "limnifs-write-test-{}-roundtrip",
2955 std::process::id()
2956 ));
2957 std::fs::create_dir_all(&temp).expect("create temp dir");
2958 std::fs::write(temp.join("a.txt"), b"aaa").expect("write a");
2959 std::fs::write(temp.join("b.txt"), b"bbb").expect("write b");
2960 std::fs::create_dir_all(temp.join("sub")).expect("create sub");
2961 std::fs::write(temp.join("sub").join("c.txt"), b"ccc").expect("write c");
2962 let artifact = write_directory(&temp).expect("write succeeds");
2963 std::fs::remove_dir_all(&temp).ok();
2964 assert_eq!(artifact.file_count, 3);
2965 assert_eq!(artifact.dir_count, 2);
2966
2967 let mut cursor = ManifestCursor::new(&artifact.bytes);
2968 limnifs_core::parse_manifest_header(&mut cursor).expect("header");
2969 limnifs_core::parse_feature_flags_section(&mut cursor).expect("flags");
2970 let meta_ref = limnifs_core::parse_metadata_reference(&mut cursor).expect("meta ref");
2971 assert!(meta_ref.is_inlined());
2972 let slab_index = limnifs_core::parse_slab_index(&mut cursor).expect("slab index");
2973 assert_eq!(slab_index.len(), 0);
2974 limnifs_core::parse_history(&mut cursor).expect("history");
2975 }
2976
2977 #[test]
2978 fn write_deterministic() {
2979 let temp =
2980 std::env::temp_dir().join(format!("limnifs-write-test-{}-det", std::process::id()));
2981 std::fs::create_dir_all(&temp).expect("create temp dir");
2982 std::fs::write(temp.join("x.txt"), b"xxx").expect("write x");
2983
2984 let a1 = write_directory(&temp).expect("first write");
2985 let a2 = write_directory(&temp).expect("second write");
2986 std::fs::remove_dir_all(&temp).ok();
2987
2988 assert_eq!(a1.bytes, a2.bytes);
2989 assert_eq!(a1.merkle_root, a2.merkle_root);
2990 }
2991
2992 #[test]
2993 fn slab_parses_correctly() {
2994 let temp =
2995 std::env::temp_dir().join(format!("limnifs-write-test-{}-slab", std::process::id()));
2996 std::fs::create_dir_all(&temp).expect("create temp dir");
2997 std::fs::write(temp.join("big.bin"), vec![0x11u8; INLINE_THRESHOLD + 1])
2998 .expect("write big");
2999 let artifact = write_directory(&temp).expect("write succeeds");
3000 std::fs::remove_dir_all(&temp).ok();
3001
3002 let slab_bytes = &artifact.slabs[0].bytes;
3003 let mut cursor = ManifestCursor::new(slab_bytes);
3004 let slab_header = limnifs_core::parse_slab_header(&mut cursor).expect("slab header parses");
3005 assert_eq!(
3006 slab_header.format_version,
3007 limnifs_core::slab::SLAB_FORMAT_VERSION
3008 );
3009 assert!(!slab_header.is_sealed());
3010 assert!(!slab_header.has_erasure_coding());
3011
3012 let drop_record =
3013 limnifs_core::parse_drop_record(&mut cursor, &slab_header).expect("drop record parses");
3014 assert_eq!(drop_record.plaintext_len as usize, INLINE_THRESHOLD + 1);
3015 }
3016
3017 #[test]
3018 fn fastcdc_produces_multiple_chunks_for_large_files() {
3019 let temp = std::env::temp_dir().join(format!(
3022 "limnifs-write-test-{}-cdc-multi",
3023 std::process::id()
3024 ));
3025 std::fs::create_dir_all(&temp).expect("create temp dir");
3026 let data = pseudo_random_bytes(42, 1024 * 1024);
3027 std::fs::write(temp.join("big.bin"), &data).expect("write big");
3028 let artifact = write_directory(&temp).expect("write succeeds");
3029 std::fs::remove_dir_all(&temp).ok();
3030 assert!(
3031 artifact.drop_count > 1,
3032 "expected FastCDC to produce multiple drops for 1 MiB input, got {}",
3033 artifact.drop_count
3034 );
3035 }
3036
3037 #[test]
3038 fn fastcdc_deduplicates_shared_substrings() {
3039 let temp = std::env::temp_dir().join(format!(
3043 "limnifs-write-test-{}-cdc-dedup",
3044 std::process::id()
3045 ));
3046 std::fs::create_dir_all(&temp).expect("create temp dir");
3047 let shared = pseudo_random_bytes(7, 512 * 1024);
3048 let mut a = Vec::with_capacity(shared.len() + 1024);
3049 a.extend_from_slice(&pseudo_random_bytes(1, 1024));
3050 a.extend_from_slice(&shared);
3051 let mut b = Vec::with_capacity(shared.len() + 2048);
3052 b.extend_from_slice(&pseudo_random_bytes(2, 2048));
3053 b.extend_from_slice(&shared);
3054 std::fs::write(temp.join("a.bin"), &a).expect("write a");
3055 std::fs::write(temp.join("b.bin"), &b).expect("write b");
3056
3057 let temp_a = std::env::temp_dir().join(format!(
3059 "limnifs-write-test-{}-cdc-dedup-a",
3060 std::process::id()
3061 ));
3062 std::fs::create_dir_all(&temp_a).expect("create temp_a");
3063 std::fs::write(temp_a.join("a.bin"), &a).expect("write a");
3064 let artifact_a = write_directory(&temp_a).expect("a writes");
3065 std::fs::remove_dir_all(&temp_a).ok();
3066
3067 let temp_b = std::env::temp_dir().join(format!(
3068 "limnifs-write-test-{}-cdc-dedup-b",
3069 std::process::id()
3070 ));
3071 std::fs::create_dir_all(&temp_b).expect("create temp_b");
3072 std::fs::write(temp_b.join("b.bin"), &b).expect("write b");
3073 let artifact_b = write_directory(&temp_b).expect("b writes");
3074 std::fs::remove_dir_all(&temp_b).ok();
3075
3076 let artifact_both = write_directory(&temp).expect("both write");
3077 std::fs::remove_dir_all(&temp).ok();
3078
3079 let sum_alone = artifact_a.drop_count + artifact_b.drop_count;
3080 assert!(
3081 artifact_both.drop_count < sum_alone,
3082 "expected dedup win: both together = {} drops, sum alone = {} drops",
3083 artifact_both.drop_count,
3084 sum_alone
3085 );
3086 }
3087
3088 #[test]
3089 fn slab_splits_when_content_exceeds_ceiling() {
3090 let temp =
3096 std::env::temp_dir().join(format!("limnifs-write-test-{}-split", std::process::id()));
3097 std::fs::create_dir_all(&temp).expect("create temp dir");
3098 for i in 0..7u32 {
3099 let data = pseudo_random_bytes(u64::from(i), 10 * 1024 * 1024);
3101 std::fs::write(temp.join(format!("big-{i}.bin")), &data).expect("write big");
3102 }
3103 let artifact = write_directory(&temp).expect("write succeeds");
3104 std::fs::remove_dir_all(&temp).ok();
3105
3106 assert!(
3108 artifact.slabs.len() >= 2,
3109 "expected at least 2 slabs for 70 MiB of incompressible data, got {}",
3110 artifact.slabs.len()
3111 );
3112 for slab in &artifact.slabs {
3113 assert!(
3114 slab.bytes.len() <= MAX_SLAB_TOTAL_BYTES,
3115 "slab {} is {} bytes (> {} ceiling)",
3116 slab.id.ordinal,
3117 slab.bytes.len(),
3118 MAX_SLAB_TOTAL_BYTES,
3119 );
3120 }
3121 let total_drop_ids: usize = artifact.slabs.iter().map(|s| s.drop_ids.len()).sum();
3123 assert_eq!(
3124 total_drop_ids, artifact.drop_count,
3125 "drop_ids count across slabs must match WriteArtifact.drop_count",
3126 );
3127 }
3128}