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(
260 name,
261 crate::stream::EntryMeta::new(0, 0o644),
262 &[],
263 &mut reader,
264 )?;
265 writer.finish()
266}
267
268pub fn write_layer(
309 base_image: &Path,
310 root: &Path,
311 config: &WriteConfig,
312) -> Result<WriteArtifact, WriteError> {
313 let base_root = load_base_drop_index(base_image)?.1;
315 let base_drop_index: std::sync::Arc<dyn BaseDropSet> = {
316 #[cfg(feature = "sparse-index")]
317 {
318 match SparseBackedBaseIndex::open(base_image) {
319 Some(idx) => std::sync::Arc::new(idx),
320 None => std::sync::Arc::new(load_base_drop_index(base_image)?.0),
321 }
322 }
323 #[cfg(not(feature = "sparse-index"))]
324 {
325 std::sync::Arc::new(load_base_drop_index(base_image)?.0)
326 }
327 };
328
329 let mut ctx = WriteContext::new();
330 ctx.chunker = chunker_from_config(config)?;
331 ctx.base_dictionaries = if config.dictionaries.enabled {
334 load_base_dictionary_section(base_image)?.map(crate::dictionary::adopt_from_section)
335 } else {
336 None
337 };
338 ctx.categorizers_disabled = config.categorizers.is_empty();
339 ctx.rw_mode = matches!(config.mode, crate::config::ImageMode::ReadWrite(_));
340 ctx.auto_turnover = config.turnover_threshold > 0;
341 ctx.collect_dict_samples = config.dictionaries.enabled;
342 ctx.inline_threshold = config.defaults.inline_threshold as usize;
343 ctx.metadata_externalize_threshold = config.defaults.metadata_externalize_threshold;
344 ctx.emit_shared_inline = config.defaults.shared_inline;
345 ctx.base_drop_index = Some(base_drop_index);
346 ctx.base_root = Some(base_root);
347
348 let root_inode_number = ctx.walk(root)?;
350 ctx.root_inode_number = root_inode_number;
351 write_directory_body(&mut ctx, config)?;
352 Ok(ctx.assemble())
353}
354
355pub trait BaseDropSet: Send + Sync {
368 fn base_contains(&self, drop_id: &[u8; 32]) -> bool;
371}
372
373impl BaseDropSet for std::collections::HashSet<[u8; 32]> {
374 fn base_contains(&self, drop_id: &[u8; 32]) -> bool {
375 self.contains(drop_id)
376 }
377}
378
379#[cfg(feature = "sparse-index")]
385pub struct SparseBackedBaseIndex {
386 bloom: crate::sparse_index::SparseIndexReader,
387 manifest_path: std::path::PathBuf,
388 exact: std::sync::OnceLock<std::collections::HashSet<[u8; 32]>>,
389}
390
391#[cfg(feature = "sparse-index")]
392impl SparseBackedBaseIndex {
393 #[must_use]
396 pub fn open(base_image: &Path) -> Option<Self> {
397 let sidecar = base_image.with_extension("lim.sparse");
398 let bloom = crate::sparse_index::SparseIndexReader::from_file(&sidecar)?;
399 Some(Self {
400 bloom,
401 manifest_path: base_image.to_path_buf(),
402 exact: std::sync::OnceLock::new(),
403 })
404 }
405
406 fn load_exact(&self) -> &std::collections::HashSet<[u8; 32]> {
407 self.exact.get_or_init(|| {
408 let bytes = std::fs::read(&self.manifest_path).unwrap_or_default();
412 let mut cursor = ManifestCursor::new(&bytes);
413 let _ = parse_manifest_header(&mut cursor);
414 let _ = limnifs_core::parse_feature_flags_section(&mut cursor);
415 let _ = limnifs_core::parse_metadata_reference(&mut cursor);
416 let Ok(index) = parse_slab_index(&mut cursor) else {
417 return std::collections::HashSet::new();
418 };
419 match SlabStore::load_mmap(&self.manifest_path, &index) {
420 Ok(store) => store.drop_index_keys().copied().collect(),
421 Err(_) => std::collections::HashSet::new(),
422 }
423 })
424 }
425}
426
427#[cfg(feature = "sparse-index")]
428impl BaseDropSet for SparseBackedBaseIndex {
429 fn base_contains(&self, drop_id: &[u8; 32]) -> bool {
430 if !self.bloom.probably_contains(drop_id) {
435 return false;
436 }
437 self.load_exact().contains(drop_id)
438 }
439}
440
441#[cfg(feature = "sparse-index")]
449pub fn emit_sparse_sidecar(artifact: &WriteArtifact, image_path: &Path) -> Result<(), WriteError> {
450 let all: std::collections::HashSet<[u8; 32]> = artifact
451 .slabs
452 .iter()
453 .flat_map(|s| s.drop_ids.iter().copied())
454 .collect();
455 let mut writer = crate::sparse_index::SparseIndexWriter::new(
456 all.len().max(1),
457 crate::sparse_index::DEFAULT_FPP,
458 );
459 writer.insert_all(&all);
460 let sidecar = image_path.with_extension("lim.sparse");
461 writer.write_to_file(&sidecar).map_err(WriteError::Io)
462}
463
464fn load_base_dictionary_section(
470 base_image: &Path,
471) -> Result<Option<limnifs_core::dictionary_section::DictionarySection>, WriteError> {
472 let manifest_bytes = std::fs::read(base_image)?;
473 let mut cursor = ManifestCursor::new(&manifest_bytes);
474 let _ = parse_manifest_header(&mut cursor).map_err(io_core)?;
475 let _ = limnifs_core::parse_feature_flags_section(&mut cursor);
476 let _ = limnifs_core::parse_metadata_reference(&mut cursor);
477 let _ = parse_slab_index(&mut cursor);
478 let _ = limnifs_core::parse_history(&mut cursor);
479 if cursor.remaining_len() == 0 {
480 return Ok(None);
481 }
482 Ok(parse_dictionary_section(&mut cursor).ok())
483}
484
485fn load_base_drop_index(
486 base_image: &Path,
487) -> Result<(std::collections::HashSet<[u8; 32]>, [u8; 32]), WriteError> {
488 let manifest_bytes = std::fs::read(base_image)?;
489 let mut cursor = ManifestCursor::new(&manifest_bytes);
490 let _ = parse_manifest_header(&mut cursor).map_err(io_core)?;
491 let _ = limnifs_core::parse_feature_flags_section(&mut cursor);
493 let _ = limnifs_core::parse_metadata_reference(&mut cursor);
494 let slab_index = parse_slab_index(&mut cursor).map_err(io_core)?;
495 let store = SlabStore::load_mmap(base_image, &slab_index).map_err(io_core)?;
496 let drop_set: std::collections::HashSet<[u8; 32]> = store.drop_index_keys().copied().collect();
497 let root = *compute_merkle_root_from_sections(&manifest_bytes).as_bytes();
503 Ok((drop_set, root))
504}
505
506fn compute_merkle_root_from_sections(manifest: &[u8]) -> ManifestRoot {
512 use limnifs_core::SectionHashes;
513 let mut cursor = ManifestCursor::new(manifest);
514 let header_start = 0;
515 if parse_manifest_header(&mut cursor).is_err() {
516 return ManifestRoot::from_bytes([0u8; 32]);
518 }
519 let header_end = cursor.position();
520 let flags_start = header_end;
522 let flags_end = match limnifs_core::parse_feature_flags_section(&mut cursor) {
523 Ok(_) => cursor.position(),
524 Err(_) => flags_start,
525 };
526 let meta_ref_start = flags_end;
527 let metadata_reference = match limnifs_core::parse_metadata_reference(&mut cursor) {
528 Ok(m) => Some(m),
529 Err(_) => None,
530 };
531 let meta_ref_end = cursor.position();
532 let slab_index_start = meta_ref_end;
533 let _ = parse_slab_index(&mut cursor);
534 let slab_index_end = cursor.position();
535 let history_start = slab_index_end;
536 let _ = limnifs_core::parse_history(&mut cursor);
537 let history_end = cursor.position();
538
539 let hashes = SectionHashes {
540 metadata: metadata_reference
541 .map(|m| m.metadata_hash)
542 .unwrap_or_else(hash_empty_section),
543 format_header: hash_section(&manifest[header_start..header_end]),
544 feature_flags: hash_section(&manifest[flags_start..flags_end]),
545 metadata_reference: hash_section(&manifest[meta_ref_start..meta_ref_end]),
546 slab_index: hash_section(&manifest[slab_index_start..slab_index_end]),
547 crypto_params: hash_empty_section(),
548 ec_params: hash_empty_section(),
549 dms_policy: hash_empty_section(),
550 delta_linkage: hash_empty_section(),
551 history: hash_section(&manifest[history_start..history_end]),
552 };
553 compute_merkle_root(&hashes)
554}
555
556#[cfg(feature = "xattr")]
558fn to_core_xattrs(raw: &[(String, Vec<u8>)]) -> Vec<limnifs_core::inode::XAttr> {
559 raw.iter()
560 .map(|(key, value)| limnifs_core::inode::XAttr {
561 namespace: 0,
562 key: key.clone(),
563 value: value.clone(),
564 })
565 .collect()
566}
567
568fn io_core(e: limnifs_core::CoreError) -> WriteError {
569 WriteError::Io(std::io::Error::other(format!("base image load: {e}")))
570}
571
572fn write_directory_body(ctx: &mut WriteContext, config: &WriteConfig) -> Result<(), WriteError> {
577 use rayon::prelude::*;
578
579 ctx.metadata_codec = config
580 .metadata_codec_id()
581 .unwrap_or(limnifs_core::codec::CODEC_BROTLI);
582
583 ctx.chunker = chunker_from_config(config)?;
584
585 let pending = std::mem::take(&mut ctx.pending_files);
586 if pending.is_empty() {
587 return Ok(());
588 }
589 ctx.inline_threshold = config.defaults.inline_threshold as usize;
590 ctx.metadata_externalize_threshold = config.defaults.metadata_externalize_threshold;
591 ctx.emit_shared_inline = config.defaults.shared_inline;
592 let chunker = ctx.chunker.clone();
593 let classifier = ctx.classifier;
594 let text_codec = config.text_codec_id().unwrap_or(0x04);
595 let binary_codec = config.binary_codec_id().unwrap_or(0x01);
596 let tunables = config.to_core_tunables();
597 let use_categorizers = !config.categorizers.is_empty();
598 let skip_chunking = config.skip_chunking;
599 let registry = config
600 .codec_registry()
601 .map_err(|e| WriteError::Io(std::io::Error::other(format!("codec registry: {e}"))))?;
602 let tournament_codec_ids: Vec<u8> = config
603 .tournament
604 .codecs
605 .iter()
606 .filter_map(|n| registry.lookup_by_name(n))
607 .collect();
608 let tournament_spec = TournamentSpec {
609 codec_ids: tournament_codec_ids,
610 min_size: config.tournament.min_size_threshold as usize,
611 skip_for_binary: config.tournament.skip_for_binary,
612 short_circuit_permille: config.tournament.short_circuit_threshold,
613 };
614 let base_drop_index: Option<&dyn BaseDropSet> = ctx.base_drop_index.as_deref();
615 let inline_threshold = ctx.inline_threshold;
616 let max_drop_size = config.defaults.max_drop_size as usize;
617 let seekable_drops = config.defaults.seekable_drops;
618 let seekable_drops = config.defaults.seekable_drops;
619 let results: Vec<ChunkedFileResult> = pending
620 .par_iter()
621 .map(|pf| {
622 process_file(
623 pf,
624 &chunker,
625 classifier,
626 text_codec,
627 binary_codec,
628 &tunables,
629 use_categorizers,
630 skip_chunking,
631 &tournament_spec,
632 base_drop_index,
633 inline_threshold,
634 max_drop_size,
635 seekable_drops,
636 config.categorizers.as_slice(),
637 &|name| {
638 config
639 .codec_registry()
640 .ok()
641 .and_then(|r| r.lookup_by_name(name))
642 },
643 )
644 })
645 .collect::<Result<Vec<_>, _>>()?;
646
647 for (pf, result) in pending.iter().zip(results) {
648 ctx.merge_chunked_file(pf, result);
649 }
650 ctx.train_and_apply_dictionary(&config.dictionaries);
651 Ok(())
652}
653
654pub fn write_directory_with_config(
656 root: &Path,
657 config: &WriteConfig,
658) -> Result<WriteArtifact, WriteError> {
659 let mut ctx = WriteContext::new();
660 ctx.chunker = chunker_from_config(config)?;
661 ctx.categorizers_disabled = config.categorizers.is_empty();
662 ctx.rw_mode = matches!(config.mode, crate::config::ImageMode::ReadWrite(_));
663 ctx.auto_turnover = config.turnover_threshold > 0;
664 ctx.collect_dict_samples = config.dictionaries.enabled;
665
666 write_directory_streaming(&mut ctx, root, config)?;
667 Ok(ctx.assemble())
668}
669
670fn write_directory_streaming(
684 ctx: &mut WriteContext,
685 root: &Path,
686 config: &WriteConfig,
687) -> Result<(), WriteError> {
688 use rayon::prelude::*;
689
690 ctx.metadata_codec = config
691 .metadata_codec_id()
692 .unwrap_or(limnifs_core::codec::CODEC_BROTLI);
693
694 ctx.chunker = chunker_from_config(config)?;
695
696 let chunker = ctx.chunker.clone();
697 let classifier = ctx.classifier;
698 let text_codec = config.text_codec_id().unwrap_or(0x04);
699 let binary_codec = config.binary_codec_id().unwrap_or(0x01);
700 let tunables = config.to_core_tunables();
701 let use_categorizers = !config.categorizers.is_empty();
702 let skip_chunking = config.skip_chunking;
703 let registry = config
704 .codec_registry()
705 .map_err(|e| WriteError::Io(std::io::Error::other(format!("codec registry: {e}"))))?;
706 let tournament_codec_ids: Vec<u8> = config
707 .tournament
708 .codecs
709 .iter()
710 .filter_map(|n| registry.lookup_by_name(n))
711 .collect();
712 let tournament_spec = TournamentSpec {
713 codec_ids: tournament_codec_ids,
714 min_size: config.tournament.min_size_threshold as usize,
715 skip_for_binary: config.tournament.skip_for_binary,
716 short_circuit_permille: config.tournament.short_circuit_threshold,
717 };
718 let base_drop_index = ctx.base_drop_index.clone();
721 let inline_threshold = ctx.inline_threshold;
722 let max_drop_size = config.defaults.max_drop_size as usize;
723 let seekable_drops = config.defaults.seekable_drops;
724
725 ctx.inline_threshold = config.defaults.inline_threshold as usize;
726 ctx.metadata_externalize_threshold = config.defaults.metadata_externalize_threshold;
727 ctx.emit_shared_inline = config.defaults.shared_inline;
728
729 const PIPELINE_CAPACITY: usize = 256;
733 let (tx, rx) = std::sync::mpsc::sync_channel::<PendingFile>(PIPELINE_CAPACITY);
734 ctx.pending_sink = Some(tx);
735
736 let (root_inode_number, mut results): (
737 u64,
738 Vec<(usize, PendingFile, Result<ChunkedFileResult, WriteError>)>,
739 ) = {
740 let survey = survey_tree(root)?;
750 std::thread::scope(|scope| {
751 let producer = {
752 let ctx = &mut *ctx;
753 let root = root;
754 scope.spawn(move || {
755 let r = ctx.fold_survey(root, &survey, None);
756 ctx.pending_sink = None;
760 r
761 })
762 };
763 let results = rx
766 .into_iter()
767 .enumerate()
768 .par_bridge()
769 .map(|(i, pf)| {
770 let r = process_file(
771 &pf,
772 &chunker,
773 classifier,
774 text_codec,
775 binary_codec,
776 &tunables,
777 use_categorizers,
778 skip_chunking,
779 &tournament_spec,
780 base_drop_index.as_deref(),
781 inline_threshold,
782 max_drop_size,
783 seekable_drops,
784 config.categorizers.as_slice(),
785 &|name| {
786 config
787 .codec_registry()
788 .ok()
789 .and_then(|r| r.lookup_by_name(name))
790 },
791 );
792 (i, pf, r)
793 })
794 .collect();
795 let joined = producer
796 .join()
797 .unwrap_or_else(|_| {
798 Err(WriteError::Io(std::io::Error::other(
799 "walk thread panicked",
800 )))
801 })
802 .map(|n| (n, results));
803 joined
806 })
807 }?;
808 ctx.pending_sink = None;
809 ctx.root_inode_number = root_inode_number;
810
811 results.sort_unstable_by_key(|(i, _, _)| *i);
812 for (_, pf, r) in results {
816 ctx.merge_chunked_file(&pf, r?);
817 }
818 ctx.train_and_apply_dictionary(&config.dictionaries);
819 Ok(())
820}
821
822pub(crate) type RawDrop = ([u8; 32], Vec<u8>, std::sync::Arc<[u8]>, u8, u8);
829pub(crate) struct ChunkedFileResult {
831 drops: Vec<RawDrop>, slices: Vec<PendingSlice>,
833}
834
835struct TournamentSpec {
843 codec_ids: Vec<u8>,
847 min_size: usize,
851 skip_for_binary: bool,
855 short_circuit_permille: u32,
860}
861
862fn chunker_from_config(config: &WriteConfig) -> Result<ParallelFastCDC, WriteError> {
881 ParallelFastCDC::new(
882 config.chunking.min_chunk_size as usize,
883 config.chunking.avg_chunk_size as usize,
884 config.chunking.max_chunk_size as usize,
885 )
886 .map_err(|e| WriteError::Io(std::io::Error::other(format!("chunking config: {e}"))))
887}
888
889pub(crate) fn seekable_or_monolithic(
896 codec: u8,
897 plaintext: &[u8],
898 compressed: std::sync::Arc<[u8]>,
899 tunables: &limnifs_core::codec::CodecTunables,
900 seekable_drops: bool,
901 threshold: usize,
902) -> (std::sync::Arc<[u8]>, u8) {
903 use limnifs_core::seekable::{
904 encode_seekable, is_seekable_codec, DROP_FLAG_SEEKABLE as FLAG, SEEKABLE_EMISSION_THRESHOLD,
905 };
906 if seekable_drops && plaintext.len() > threshold && is_seekable_codec(codec) {
907 if let Ok(container) = encode_seekable(codec, plaintext, tunables) {
908 return (container.into(), FLAG);
909 }
910 }
911 (compressed, 0)
912}
913
914pub(crate) const SEEKABLE_CHUNK_EMISSION_THRESHOLD: usize =
921 limnifs_core::seekable::SEEKABLE_FRAME_SIZE;
922
923fn process_whole_file_drop(
924 pf: &PendingFile,
925 data: &[u8],
926 cat: file_categorizer::Categorization,
927 tunables: &limnifs_core::codec::CodecTunables,
928 seekable_drops: bool,
929) -> Result<ChunkedFileResult, WriteError> {
930 let _ = pf;
931 let drop_id = hash_section(data);
932 let file_len = u64::try_from(data.len()).unwrap_or(u64::MAX);
933
934 let (mut best_codec, mut best_compressed): (u8, std::sync::Arc<[u8]>) =
939 match limnifs_core::codec::compress_with_tunables(
940 limnifs_core::codec::CODEC_BROTLI,
941 data,
942 tunables,
943 ) {
944 Ok(c) => (limnifs_core::codec::CODEC_BROTLI, c.into()),
945 Err(_) => match limnifs_core::codec::compress_with_tunables(
946 limnifs_core::codec::CODEC_ZSTD,
947 data,
948 tunables,
949 ) {
950 Ok(c) => (limnifs_core::codec::CODEC_ZSTD, c.into()),
951 Err(_) => (limnifs_core::codec::CODEC_STORE, data.to_vec().into()),
952 },
953 };
954
955 let brotli_ratio = best_compressed.len() as f64 / data.len() as f64;
959 if brotli_ratio > 0.05 {
960 if let Ok(zstd_c) = limnifs_core::codec::compress_with_tunables(
961 limnifs_core::codec::CODEC_ZSTD,
962 data,
963 tunables,
964 ) {
965 if zstd_c.len() < best_compressed.len() {
966 best_codec = limnifs_core::codec::CODEC_ZSTD;
967 best_compressed = zstd_c.into();
968 }
969 }
970 }
971
972 let general_ratio = best_compressed.len() as f64 / data.len() as f64;
978 if general_ratio > 0.15 || cat.codec_id == limnifs_core::codec::CODEC_RICEPP {
979 let spec_result = if cat.codec_id == limnifs_core::codec::CODEC_FSST_BROTLI {
983 limnifs_core::codec::fsst_brotli::compress_with_baseline(data, Some(&best_compressed))
984 } else {
985 limnifs_core::codec::compress_with_tunables(cat.codec_id, data, tunables)
989 };
990 if let Ok(spec_c) = spec_result {
991 if spec_c.len() < best_compressed.len() {
992 best_codec = cat.codec_id;
993 best_compressed = spec_c.into();
994 }
995 }
996 }
997
998 let (best_compressed, flags) = seekable_or_monolithic(
999 best_codec,
1000 data,
1001 best_compressed,
1002 tunables,
1003 seekable_drops,
1004 limnifs_core::seekable::SEEKABLE_EMISSION_THRESHOLD,
1005 );
1006 Ok(ChunkedFileResult {
1007 drops: vec![(drop_id, data.to_vec(), best_compressed, best_codec, flags)],
1008 slices: vec![PendingSlice {
1009 drop_id,
1010 file_byte_start: 0,
1011 file_byte_end: file_len,
1012 }],
1013 })
1014}
1015
1016fn compress_chunk_with_tournament(
1038 chunk: &[u8],
1039 class: classifier::Class,
1040 text_codec: u8,
1041 binary_codec: u8,
1042 tunables: &limnifs_core::codec::CodecTunables,
1043 tournament: &TournamentSpec,
1044) -> (u8, std::sync::Arc<[u8]>) {
1045 use classifier::Class;
1046
1047 let preferred = match class {
1048 Class::Binary => binary_codec,
1049 Class::Text | Class::Code | Class::Sparse => text_codec,
1050 _ => limnifs_core::codec::CODEC_STORE,
1051 };
1052
1053 if preferred == limnifs_core::codec::CODEC_STORE {
1054 return (limnifs_core::codec::CODEC_STORE, chunk.to_vec().into());
1055 }
1056 if class == Class::Binary && tournament.skip_for_binary {
1057 return compress_chunk_one(chunk, preferred, tunables);
1058 }
1059 if chunk.len() < tournament.min_size {
1060 return compress_chunk_one(chunk, preferred, tunables);
1061 }
1062
1063 let mut best: Option<(u8, std::sync::Arc<[u8]>)> = None;
1064 for &codec_id in &tournament.codec_ids {
1065 if codec_id == limnifs_core::codec::CODEC_STORE {
1066 continue;
1067 }
1068 let c = match limnifs_core::codec::compress_with_tunables(codec_id, chunk, tunables) {
1069 Ok(c) => c,
1070 Err(_) => continue,
1071 };
1072 if c.len() >= chunk.len() {
1073 continue;
1074 }
1075 let ratio_permille = (c.len() as u64 * 1000 / chunk.len() as u64) as u32;
1076 let is_best_so_far = best.as_ref().map_or(true, |(_, b)| c.len() < b.len());
1077 if is_best_so_far {
1078 best = Some((codec_id, c.into()));
1079 }
1080 if tournament.short_circuit_permille > 0
1081 && ratio_permille <= tournament.short_circuit_permille
1082 {
1083 break;
1084 }
1085 }
1086
1087 best.unwrap_or_else(|| (limnifs_core::codec::CODEC_STORE, chunk.to_vec().into()))
1088}
1089
1090fn compress_chunk_one(
1093 chunk: &[u8],
1094 codec_id: u8,
1095 tunables: &limnifs_core::codec::CodecTunables,
1096) -> (u8, std::sync::Arc<[u8]>) {
1097 if codec_id == limnifs_core::codec::CODEC_STORE {
1098 return (limnifs_core::codec::CODEC_STORE, chunk.to_vec().into());
1099 }
1100 match limnifs_core::codec::compress_with_tunables(codec_id, chunk, tunables) {
1101 Ok(c) if c.len() < chunk.len() => (codec_id, c.into()),
1102 _ => (limnifs_core::codec::CODEC_STORE, chunk.to_vec().into()),
1103 }
1104}
1105
1106fn process_file(
1110 pf: &PendingFile,
1111 chunker: &dyn Chunker,
1112 classifier: classifier::Classifier,
1113 text_codec: u8,
1114 binary_codec: u8,
1115 tunables: &limnifs_core::codec::CodecTunables,
1116 use_categorizers: bool,
1117 skip_chunking: bool,
1118 tournament: &TournamentSpec,
1119 base_drop_index: Option<&dyn BaseDropSet>,
1120 inline_threshold: usize,
1121 max_drop_size: usize,
1122 seekable_drops: bool,
1123 categorizer_config: &[crate::config::CategorizerConfig],
1124 codec_name_resolver: &dyn Fn(&str) -> Option<u8>,
1125) -> Result<ChunkedFileResult, WriteError> {
1126 let file_len_estimate = std::fs::metadata(&pf.path)
1137 .map(|m| m.len() as usize)
1138 .unwrap_or(0);
1139 let mmap_handle: memmap2::Mmap;
1146 let small: Vec<u8>;
1147 let data: &[u8] = if file_len_estimate >= MMAP_READ_THRESHOLD {
1148 let file = std::fs::File::open(&pf.path)?;
1149 #[allow(unsafe_code)]
1153 let mapped = unsafe { memmap2::Mmap::map(&file) }.map_err(WriteError::Io)?;
1154 mmap_handle = mapped;
1155 &mmap_handle[..]
1156 } else {
1157 small = std::fs::read(&pf.path)?;
1158 &small[..]
1159 };
1160 let file_len = data.len();
1161
1162 if skip_chunking && file_len > inline_threshold {
1169 let drop_id = hash_section(&data);
1170 let class = classifier.classify(&data);
1171 let preferred_codec = match class {
1172 classifier::Class::Binary => binary_codec,
1173 _ => text_codec,
1174 };
1175 let (codec_id, compressed): (u8, std::sync::Arc<[u8]>) =
1176 match limnifs_core::codec::compress_with_tunables(preferred_codec, &data, tunables) {
1177 Ok(c) if c.len() < data.len() => (preferred_codec, c.into()),
1178 _ => (limnifs_core::codec::CODEC_STORE, data.to_vec().into()),
1179 };
1180 let (compressed, flags) = seekable_or_monolithic(
1181 codec_id,
1182 &data,
1183 compressed,
1184 tunables,
1185 seekable_drops,
1186 SEEKABLE_CHUNK_EMISSION_THRESHOLD,
1187 );
1188 return Ok(ChunkedFileResult {
1189 drops: vec![(drop_id, data.to_vec(), compressed, codec_id, flags)],
1190 slices: vec![PendingSlice {
1191 drop_id,
1192 file_byte_start: 0,
1193 file_byte_end: file_len as u64,
1194 }],
1195 });
1196 }
1197
1198 if use_categorizers {
1199 let config_cat = file_categorizer::ConfigCategorizer::new(categorizer_config.to_vec());
1202 if let Some(cat) = config_cat.categorize(&pf.path, &data) {
1203 if let Some(codec_id) = file_categorizer::resolve_config_categorization(
1204 &cat,
1205 categorizer_config,
1206 codec_name_resolver,
1207 ) {
1208 let within_cap = max_drop_size == 0 || file_len <= max_drop_size;
1209 let needs_whole_file = matches!(
1210 codec_id,
1211 limnifs_core::codec::CODEC_FLAC | limnifs_core::codec::CODEC_RICEPP
1212 );
1213 if within_cap && (needs_whole_file || file_len <= WHOLE_FILE_MAX_SIZE) {
1214 let mut cat = cat;
1215 cat.codec_id = codec_id;
1216 return process_whole_file_drop(pf, &data, cat, tunables, seekable_drops);
1217 }
1218 }
1219 }
1220 if let Some(cat) = file_categorizer::default_registry().categorize(&pf.path, &data) {
1221 let needs_whole_file = matches!(
1222 cat.codec_id,
1223 limnifs_core::codec::CODEC_FLAC | limnifs_core::codec::CODEC_RICEPP
1224 );
1225 let within_cap = max_drop_size == 0 || file_len <= max_drop_size;
1228 if within_cap && (needs_whole_file || file_len <= WHOLE_FILE_MAX_SIZE) {
1229 return process_whole_file_drop(pf, &data, cat, tunables, seekable_drops);
1230 }
1231 }
1232 }
1233
1234 let chunks = chunker.chunk_slice(&data);
1235
1236 use rayon::prelude::*;
1245 let drop_ids: Vec<[u8; 32]> = chunks.par_iter().map(|chunk| hash_section(chunk)).collect();
1246
1247 let mut slices = Vec::with_capacity(chunks.len());
1248 let mut file_offset: u64 = 0;
1249 let mut seen_in_file: std::collections::HashSet<[u8; 32]> =
1250 std::collections::HashSet::with_capacity(chunks.len());
1251 let mut unique_chunks: Vec<(&[u8], [u8; 32])> = Vec::with_capacity(chunks.len());
1252 for (chunk, drop_id) in chunks.iter().zip(drop_ids) {
1253 let chunk_len = u64::try_from(chunk.len()).expect("chunk len fits u64");
1254 slices.push(PendingSlice {
1255 drop_id,
1256 file_byte_start: file_offset,
1257 file_byte_end: file_offset + chunk_len,
1258 });
1259 file_offset += chunk_len;
1260 if seen_in_file.insert(drop_id) {
1261 unique_chunks.push((chunk, drop_id));
1262 }
1263 }
1264
1265 thread_local! {
1277 static COMPRESS_CACHE: std::cell::RefCell<std::collections::HashMap<[u8; 32], (u8, std::sync::Arc<[u8]>)>> =
1278 std::cell::RefCell::new(std::collections::HashMap::new());
1279 }
1280 const COMPRESS_CACHE_MAX_ENTRIES: usize = 100_000;
1281 let drops: Vec<RawDrop> = unique_chunks
1282 .par_iter()
1283 .map(|(chunk, drop_id)| {
1284 if let Some(base) = base_drop_index {
1287 if base.base_contains(drop_id) {
1288 return (*drop_id, Vec::new(), Vec::new().into(), CODEC_REFERENCED, 0);
1289 }
1290 }
1291 let class = classifier.classify(chunk);
1292 let cached = COMPRESS_CACHE.with(|c| {
1295 c.borrow()
1296 .get(drop_id)
1297 .map(|(cid, comp)| (*cid, comp.clone()))
1298 });
1299 let (codec_id, compressed) = if let Some(c) = cached {
1300 c
1301 } else {
1302 let new = compress_chunk_with_tournament(
1303 chunk,
1304 class,
1305 text_codec,
1306 binary_codec,
1307 tunables,
1308 tournament,
1309 );
1310 COMPRESS_CACHE.with(|c| {
1312 let mut cache = c.borrow_mut();
1313 if cache.len() < COMPRESS_CACHE_MAX_ENTRIES {
1314 cache.insert(*drop_id, new.clone());
1316 }
1317 });
1318 new
1319 };
1320 let (compressed, flags) = seekable_or_monolithic(
1326 codec_id,
1327 chunk,
1328 compressed,
1329 tunables,
1330 seekable_drops,
1331 SEEKABLE_CHUNK_EMISSION_THRESHOLD,
1332 );
1333 (*drop_id, chunk.to_vec(), compressed, codec_id, flags)
1334 })
1335 .collect();
1336
1337 let _ = file_len;
1338 Ok(ChunkedFileResult { drops, slices })
1339}
1340
1341struct PendingDrop {
1342 id: [u8; 32],
1343 plaintext_len: u32,
1349 compressed: std::sync::Arc<[u8]>,
1350 codec: u8,
1351 dict_id: u8,
1355 plaintext: Option<Vec<u8>>,
1360 flags: u8,
1364}
1365
1366impl PendingDrop {
1367 fn len_in_window(&self) -> u32 {
1371 u32::try_from(self.compressed.len()).expect("compressed fits u32")
1372 }
1373
1374 fn plaintext_len_value(&self) -> u32 {
1376 self.plaintext_len
1377 }
1378
1379 fn slab_footprint(&self) -> usize {
1382 48 + self.compressed.len()
1383 }
1384}
1385
1386struct PendingSlice {
1391 drop_id: [u8; 32],
1392 file_byte_start: u64,
1393 file_byte_end: u64,
1394}
1395
1396#[derive(Clone)]
1399struct PendingFile {
1400 inode_number: u64,
1401 path: PathBuf,
1402 mtime_ns: u64,
1403 file_len: u64,
1404 mode: u32,
1405 uid: u32,
1406 gid: u32,
1407}
1408
1409#[derive(Clone, Default)]
1413struct SurveyMeta {
1414 is_dir: bool,
1415 is_file: bool,
1416 is_symlink: bool,
1417 #[cfg(unix)]
1418 is_fifo: bool,
1419 #[cfg(unix)]
1420 is_socket: bool,
1421 #[cfg(unix)]
1422 is_block_device: bool,
1423 #[cfg(unix)]
1424 is_char_device: bool,
1425 len: u64,
1426 mtime_ns: u64,
1427 #[cfg(unix)]
1428 mode: u32,
1429 #[cfg(unix)]
1430 uid: u32,
1431 #[cfg(unix)]
1432 gid: u32,
1433 #[cfg(unix)]
1437 dev: u64,
1438 #[cfg(unix)]
1439 ino: u64,
1440 #[cfg(feature = "xattr")]
1443 xattrs: Vec<(String, Vec<u8>)>,
1444}
1445
1446impl SurveyMeta {
1447 fn identity(&self) -> (u32, u32, u32) {
1451 #[cfg(unix)]
1452 {
1453 (self.mode, self.uid, self.gid)
1454 }
1455 #[cfg(not(unix))]
1456 {
1457 let ty = if self.is_dir {
1458 limnifs_core::inode::S_IFDIR
1459 } else if self.is_symlink {
1460 limnifs_core::inode::S_IFLNK
1461 } else {
1462 limnifs_core::inode::S_IFREG
1463 };
1464 let perms = if self.is_dir || self.is_symlink {
1465 0o755
1466 } else {
1467 0o644
1468 };
1469 (ty | perms, 0, 0)
1470 }
1471 }
1472}
1473
1474struct SurveyNode {
1477 meta: SurveyMeta,
1478 children: Vec<(String, SurveyNode)>,
1479 symlink_target: Option<String>,
1483}
1484
1485impl SurveyNode {
1486 fn meta(&self) -> &SurveyMeta {
1487 &self.meta
1488 }
1489}
1490
1491fn survey_meta_of(meta: &std::fs::Metadata) -> SurveyMeta {
1492 #[cfg(unix)]
1493 use std::os::unix::fs::FileTypeExt as _;
1494 let ft = meta.file_type();
1495 let mtime_ns = meta
1496 .modified()
1497 .ok()
1498 .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
1499 .map_or(0u128, |d| d.as_nanos());
1500 SurveyMeta {
1501 is_dir: ft.is_dir(),
1502 is_file: ft.is_file(),
1503 is_symlink: ft.is_symlink(),
1504 #[cfg(unix)]
1505 is_fifo: ft.is_fifo(),
1506 #[cfg(unix)]
1507 is_socket: ft.is_socket(),
1508 #[cfg(unix)]
1509 is_block_device: ft.is_block_device(),
1510 #[cfg(unix)]
1511 is_char_device: ft.is_char_device(),
1512 len: meta.len(),
1513 mtime_ns: mtime_ns.try_into().unwrap_or(0),
1514 #[cfg(unix)]
1515 mode: {
1516 use std::os::unix::fs::MetadataExt as _;
1517 meta.mode()
1518 },
1519 #[cfg(unix)]
1520 uid: {
1521 use std::os::unix::fs::MetadataExt as _;
1522 meta.uid()
1523 },
1524 #[cfg(unix)]
1525 gid: {
1526 use std::os::unix::fs::MetadataExt as _;
1527 meta.gid()
1528 },
1529 #[cfg(unix)]
1530 dev: {
1531 use std::os::unix::fs::MetadataExt as _;
1532 meta.dev()
1533 },
1534 #[cfg(unix)]
1535 ino: {
1536 use std::os::unix::fs::MetadataExt as _;
1537 meta.ino()
1538 },
1539 #[cfg(feature = "xattr")]
1540 xattrs: Vec::new(),
1541 }
1542}
1543
1544#[cfg(all(unix, feature = "xattr"))]
1548fn collect_xattrs(path: &Path) -> Vec<(String, Vec<u8>)> {
1549 const VOLATILE: &[&str] = &[
1550 "com.apple.provenance",
1551 "com.apple.quarantine",
1552 "com.apple.lastuseddate",
1553 "com.apple.macl",
1554 "com.apple.filesec",
1555 ];
1556 const TOTAL_CAP: usize = 64 * 1024;
1557 let Ok(names) = xattr::list(path) else {
1558 return Vec::new();
1559 };
1560 let mut names: Vec<String> = names
1561 .filter_map(|n| n.into_string().ok())
1562 .filter(|n| {
1563 !n.starts_with("system.")
1564 && !n.starts_with("security.")
1565 && !n.starts_with("trusted.")
1566 && !VOLATILE.contains(&n.as_str())
1567 })
1568 .collect();
1569 names.sort();
1570 let mut out = Vec::new();
1571 let mut total = 0usize;
1572 for name in names {
1573 let Ok(Some(value)) = xattr::get(path, &name) else {
1574 continue;
1575 };
1576 total += name.len() + value.len();
1577 if total > TOTAL_CAP {
1578 break;
1579 }
1580 out.push((name, value));
1581 }
1582 out
1583}
1584
1585fn survey_node(path: &Path) -> Result<SurveyNode, WriteError> {
1586 use rayon::prelude::*;
1587 let meta = std::fs::symlink_metadata(path)?;
1588 let mut sm = survey_meta_of(&meta);
1589 #[cfg(all(unix, feature = "xattr"))]
1590 {
1591 if !sm.is_symlink {
1592 sm.xattrs = collect_xattrs(path);
1593 }
1594 }
1595 if sm.is_symlink {
1596 let target = std::fs::read_link(path)?;
1597 let target = target
1598 .to_str()
1599 .ok_or_else(|| WriteError::UnsupportedFileType {
1600 path: path.to_path_buf(),
1601 kind: format!("symlink with non-UTF-8 target ({})", target.display()),
1602 })?
1603 .to_owned();
1604 return Ok(SurveyNode {
1605 meta: sm,
1606 children: Vec::new(),
1607 symlink_target: Some(target),
1608 });
1609 }
1610 if !sm.is_dir {
1611 return Ok(SurveyNode {
1612 meta: sm,
1613 children: Vec::new(),
1614 symlink_target: None,
1615 });
1616 }
1617 let mut named: Vec<(String, PathBuf)> = std::fs::read_dir(path)?
1618 .filter_map(|entry| {
1619 entry
1620 .ok()
1621 .map(|e| (e.file_name().to_string_lossy().into_owned(), e.path()))
1622 })
1623 .collect();
1624 named.sort_by(|a, b| a.0.cmp(&b.0));
1625 named
1626 .par_iter()
1627 .map(|(name, child)| survey_node(child).map(|node| (name.clone(), node)))
1628 .collect::<Result<Vec<_>, WriteError>>()
1629 .map(|children| SurveyNode {
1630 meta: sm,
1631 children,
1632 symlink_target: None,
1633 })
1634}
1635
1636fn survey_tree(root: &Path) -> Result<SurveyNode, WriteError> {
1641 survey_node(root)
1642}
1643
1644struct PendingInode {
1645 number: u64,
1646 mode: u32,
1647 uid: u32,
1648 gid: u32,
1649 mtime_ns: u64,
1650 xattrs: Vec<limnifs_core::inode::XAttr>,
1651 content: PendingContent,
1652}
1653
1654enum PendingContent {
1655 Inline(Vec<u8>),
1656 Symlink(String),
1658 DropBacked {
1659 file_len: u64,
1660 slices: Vec<PendingSlice>,
1661 },
1662 Directory(Vec<(String, u64, u8)>),
1663}
1664
1665struct DirNode {
1666 entries: Vec<(String, u64, u8)>,
1667 bytes: Vec<u8>,
1668 hash: [u8; 32],
1669}
1670
1671struct WriteContext {
1672 next_inode: u64,
1673 hardlink_targets: std::collections::HashMap<(u64, u64), u64>,
1676 nlink_counts: std::collections::HashMap<u64, u32>,
1680 inode_xattrs: std::collections::HashMap<u64, Vec<limnifs_core::inode::XAttr>>,
1685 inodes: Vec<PendingInode>,
1686 dir_nodes: Vec<DirNode>,
1687 drops: Vec<PendingDrop>,
1688 drop_index: HashSet<[u8; 32]>,
1689 pending_files: Vec<PendingFile>,
1690 file_count: usize,
1691 dir_count: usize,
1692 root_inode_number: u64,
1693 chunker: ParallelFastCDC,
1694 classifier: classifier::Classifier,
1695 shared_inline_map: HashMap<[u8; 32], usize>,
1696 shared_inline_table: Vec<Vec<u8>>,
1697 base_dictionaries: Option<Vec<crate::dictionary::TrainedDictionary>>,
1703 profile_name: Option<String>,
1705 metadata_codec: u8,
1708 categorizers_disabled: bool,
1710 rw_mode: bool,
1712 auto_turnover: bool,
1714 collect_dict_samples: bool,
1717 dict_samples_by_class: HashMap<crate::classifier::Class, Vec<Vec<u8>>>,
1724 trained_dicts_by_class: HashMap<crate::classifier::Class, crate::dictionary::TrainedDictionary>,
1729 base_drop_index: Option<std::sync::Arc<dyn BaseDropSet>>,
1735 base_root: Option<[u8; 32]>,
1740 metadata_externalize_threshold: usize,
1745 emit_shared_inline: bool,
1751 inline_threshold: usize,
1756 pending_sink: Option<std::sync::mpsc::SyncSender<PendingFile>>,
1762}
1763
1764impl WriteContext {
1765 const MAX_DICT_SAMPLES: usize = 1000;
1768
1769 fn new() -> Self {
1770 Self {
1771 next_inode: 1,
1772 hardlink_targets: std::collections::HashMap::new(),
1773 nlink_counts: std::collections::HashMap::new(),
1774 inode_xattrs: std::collections::HashMap::new(),
1775 inodes: Vec::new(),
1776 dir_nodes: Vec::new(),
1777 drops: Vec::new(),
1778 drop_index: HashSet::new(),
1779 pending_files: Vec::new(),
1780 file_count: 0,
1781 dir_count: 0,
1782 root_inode_number: 0,
1783 chunker: ParallelFastCDC::default(),
1784 classifier: classifier::Classifier,
1785 shared_inline_map: HashMap::new(),
1786 shared_inline_table: Vec::new(),
1787 base_dictionaries: None,
1788 profile_name: None,
1789 metadata_codec: limnifs_core::codec::CODEC_BROTLI,
1790 categorizers_disabled: false,
1791 rw_mode: false,
1792 auto_turnover: false,
1793 collect_dict_samples: false,
1794 dict_samples_by_class: HashMap::new(),
1795 trained_dicts_by_class: HashMap::new(),
1796 base_drop_index: None,
1797 base_root: None,
1798 pending_sink: None,
1799 inline_threshold: INLINE_THRESHOLD,
1800 metadata_externalize_threshold: METADATA_EXTERNALIZE_THRESHOLD,
1801 emit_shared_inline: true,
1802 }
1803 }
1804
1805 fn alloc_inode(&mut self) -> u64 {
1806 let n = self.next_inode;
1807 self.next_inode += 1;
1808 n
1809 }
1810
1811 fn build_shared_inline_table(&mut self) {
1815 let mut counts: HashMap<[u8; 32], usize> = HashMap::new();
1816 for inode in &self.inodes {
1817 if let PendingContent::Inline(data) = &inode.content {
1818 let h = hash_section(data);
1819 *counts.entry(h).or_default() += 1;
1820 }
1821 }
1822 for inode in &self.inodes {
1824 if let PendingContent::Inline(data) = &inode.content {
1825 let h = hash_section(data);
1826 if counts.get(&h).copied().unwrap_or(0) > 1
1827 && !self.shared_inline_map.contains_key(&h)
1828 {
1829 let idx = self.shared_inline_table.len();
1830 self.shared_inline_table.push(data.clone());
1831 self.shared_inline_map.insert(h, idx);
1832 }
1833 }
1834 }
1835 }
1836
1837 fn merge_chunked_file(&mut self, pf: &PendingFile, result: ChunkedFileResult) {
1840 for (drop_id, plaintext, compressed, codec, flags) in result.drops {
1841 if self.drop_index.insert(drop_id) {
1842 let retain_plaintext =
1848 self.collect_dict_samples && codec == limnifs_core::codec::CODEC_ZSTD;
1849 if retain_plaintext {
1850 let total: usize = self.dict_samples_by_class.values().map(Vec::len).sum();
1851 if total < Self::MAX_DICT_SAMPLES {
1852 let class = self.classifier.classify(&plaintext);
1853 self.dict_samples_by_class
1854 .entry(class)
1855 .or_default()
1856 .push(plaintext.clone());
1857 }
1858 }
1859 self.drops.push(PendingDrop {
1860 id: drop_id,
1861 plaintext_len: u32::try_from(plaintext.len()).unwrap_or(u32::MAX),
1862 compressed,
1863 codec,
1864 dict_id: limnifs_core::drop_record::NO_DICT,
1865 plaintext: if retain_plaintext {
1866 Some(plaintext)
1867 } else {
1868 None
1869 },
1870 flags,
1871 });
1872 }
1873 }
1874 self.inodes.push(PendingInode {
1875 number: pf.inode_number,
1876 mode: pf.mode,
1877 uid: pf.uid,
1878 gid: pf.gid,
1879 mtime_ns: pf.mtime_ns,
1880 xattrs: Vec::new(),
1881 content: PendingContent::DropBacked {
1882 file_len: pf.file_len,
1883 slices: result.slices,
1884 },
1885 });
1886 }
1887
1888 #[allow(dead_code)]
1899 fn deepen_drop(&self, drop_id: [u8; 32], plaintext: &[u8]) -> PendingDrop {
1900 let class = self.classifier.classify(plaintext);
1901 let (codec, compressed): (u8, std::sync::Arc<[u8]>) = match class {
1902 classifier::Class::Text | classifier::Class::Code | classifier::Class::Binary => {
1903 let c = limnifs_core::codec::compress_lz4_with_size(plaintext);
1904 (limnifs_core::codec::CODEC_LZ4, c.into())
1905 }
1906 _ => (limnifs_core::codec::CODEC_STORE, plaintext.to_vec().into()),
1907 };
1908 PendingDrop {
1909 id: drop_id,
1910 plaintext_len: u32::try_from(plaintext.len()).unwrap_or(u32::MAX),
1911 compressed,
1912 codec,
1913 dict_id: limnifs_core::drop_record::NO_DICT,
1914 plaintext: None,
1915 flags: 0,
1916 }
1917 }
1918
1919 fn walk(&mut self, path: &Path) -> Result<u64, WriteError> {
1920 let survey = survey_tree(path)?;
1928 self.fold_survey(path, &survey, None)
1929 }
1930
1931 fn fold_survey(
1935 &mut self,
1936 path: &Path,
1937 node: &SurveyNode,
1938 symlink_target: Option<&str>,
1939 ) -> Result<u64, WriteError> {
1940 let meta = node.meta();
1941 if let Some(target) = symlink_target {
1942 let inode_number = self.alloc_inode();
1943 let (mode, uid, gid) = meta.identity();
1944 self.inodes.push(PendingInode {
1945 number: inode_number,
1946 mode,
1947 uid,
1948 gid,
1949 mtime_ns: meta.mtime_ns,
1950 xattrs: Vec::new(),
1951 content: PendingContent::Symlink(target.to_owned()),
1952 });
1953 return Ok(inode_number);
1954 }
1955 if meta.is_dir {
1956 self.dir_count += 1;
1957 let inode_number = self.alloc_inode();
1958 let mut entries: Vec<(String, u64, u8)> = Vec::new();
1959
1960 for (name, child) in &node.children {
1961 let child_path = path.join(name);
1962 let child_inode =
1963 self.fold_survey(&child_path, child, child.symlink_target.as_deref())?;
1964 let entry_type = if child.meta().is_symlink {
1965 0x03
1966 } else if child.meta().is_dir {
1967 0x02
1968 } else {
1969 0x01
1970 };
1971 entries.push((name.clone(), child_inode, entry_type));
1972 }
1973
1974 entries.sort_by(|a, b| a.0.cmp(&b.0));
1977 let dir_node = encode_dir_node(&entries);
1978 self.dir_nodes.push(dir_node);
1979 #[cfg(feature = "xattr")]
1980 if !meta.xattrs.is_empty() {
1981 self.inode_xattrs
1982 .insert(inode_number, to_core_xattrs(&meta.xattrs));
1983 }
1984 let (mode, uid, gid) = meta.identity();
1985 self.inodes.push(PendingInode {
1986 number: inode_number,
1987 mode,
1988 uid,
1989 gid,
1990 mtime_ns: meta.mtime_ns,
1991 xattrs: Vec::new(),
1992 content: PendingContent::Directory(entries),
1993 });
1994 Ok(inode_number)
1995 } else if meta.is_file {
1996 #[cfg(unix)]
2000 if let Some(&existing) = self.hardlink_targets.get(&(meta.dev, meta.ino)) {
2001 *self.nlink_counts.entry(existing).or_insert(1) += 1;
2002 return Ok(existing);
2003 }
2004 self.file_count += 1;
2005 let inode_number = self.alloc_inode();
2006 #[cfg(unix)]
2007 {
2008 self.hardlink_targets
2009 .insert((meta.dev, meta.ino), inode_number);
2010 }
2011 let file_len = meta.len;
2012 crate::progress::emit_file(path, file_len);
2013
2014 if file_len <= u64::try_from(self.inline_threshold).unwrap_or(u64::MAX) {
2015 let data = std::fs::read(path)?;
2016 #[cfg(feature = "xattr")]
2017 if !meta.xattrs.is_empty() {
2018 self.inode_xattrs
2019 .insert(inode_number, to_core_xattrs(&meta.xattrs));
2020 }
2021 let (mode, uid, gid) = meta.identity();
2022 self.inodes.push(PendingInode {
2023 number: inode_number,
2024 mode,
2025 uid,
2026 gid,
2027 mtime_ns: meta.mtime_ns,
2028 xattrs: Vec::new(),
2029 content: PendingContent::Inline(data),
2030 });
2031 } else {
2032 #[cfg(feature = "xattr")]
2034 if !meta.xattrs.is_empty() {
2035 self.inode_xattrs
2036 .insert(inode_number, to_core_xattrs(&meta.xattrs));
2037 }
2038 let (mode, uid, gid) = meta.identity();
2039 let pf = PendingFile {
2040 inode_number,
2041 path: path.to_path_buf(),
2042 mtime_ns: meta.mtime_ns,
2043 file_len,
2044 mode,
2045 uid,
2046 gid,
2047 };
2048 if let Some(sink) = &self.pending_sink {
2049 sink.send(pf).map_err(|_| {
2055 WriteError::Io(std::io::Error::other("walk: compress pipeline shut down"))
2056 })?;
2057 } else {
2058 self.pending_files.push(pf);
2059 }
2060 }
2061 Ok(inode_number)
2062 } else {
2063 #[cfg(unix)]
2064 let kind = {
2065 use std::os::unix::fs::FileTypeExt;
2066 if meta.is_fifo {
2067 "fifo".to_owned()
2068 } else if meta.is_socket {
2069 "socket".to_owned()
2070 } else if meta.is_block_device {
2071 "block device".to_owned()
2072 } else if meta.is_char_device {
2073 "character device".to_owned()
2074 } else {
2075 "unknown".to_owned()
2076 }
2077 };
2078 #[cfg(not(unix))]
2079 let kind = "unknown".to_owned();
2080 Err(WriteError::UnsupportedFileType {
2081 path: path.to_path_buf(),
2082 kind,
2083 })
2084 }
2085 }
2086
2087 fn train_and_apply_dictionary(&mut self, dictionaries: &crate::config::DictionaryConfig) {
2102 if !dictionaries.enabled {
2103 Self::release_dictionary_samples(self);
2104 return;
2105 }
2106
2107 if let Some(adopted) = self.base_dictionaries.take() {
2111 for dict in adopted {
2112 match dict.id {
2113 0 => {
2114 self.trained_dicts_by_class
2115 .insert(crate::classifier::Class::Text, dict);
2116 }
2117 1 => {
2118 self.trained_dicts_by_class
2119 .insert(crate::classifier::Class::Binary, dict);
2120 }
2121 _ => {}
2122 }
2123 }
2124 self.apply_trained_dictionaries();
2125 return;
2126 }
2127
2128 let target = usize::try_from(dictionaries.max_dict_size).unwrap_or(65_536);
2129 let min_class = usize::try_from(dictionaries.min_class_size).unwrap_or(0);
2130
2131 let text_classes = [
2135 crate::classifier::Class::Text,
2136 crate::classifier::Class::Code,
2137 crate::classifier::Class::Sparse,
2138 ];
2139 let binary_classes = [crate::classifier::Class::Binary];
2140
2141 let text_samples: Vec<&[u8]> = text_classes
2143 .iter()
2144 .flat_map(|c| self.dict_samples_by_class.get(c).into_iter().flatten())
2145 .map(Vec::as_slice)
2146 .collect();
2147 let trainer = crate::dictionary::TrainerKind::from_config_str(&dictionaries.trainer);
2148 if text_samples.len() >= min_class {
2149 if let Some(dict) =
2150 crate::dictionary::train_zstd_with_trainer(0, &text_samples, target, trainer)
2151 {
2152 self.trained_dicts_by_class
2153 .insert(crate::classifier::Class::Text, dict);
2154 }
2155 }
2156 let binary_samples: Vec<&[u8]> = binary_classes
2157 .iter()
2158 .flat_map(|c| self.dict_samples_by_class.get(c).into_iter().flatten())
2159 .map(Vec::as_slice)
2160 .collect();
2161 if binary_samples.len() >= min_class {
2162 if let Some(dict) =
2163 crate::dictionary::train_zstd_with_trainer(1, &binary_samples, target, trainer)
2164 {
2165 self.trained_dicts_by_class
2166 .insert(crate::classifier::Class::Binary, dict);
2167 }
2168 }
2169
2170 self.apply_trained_dictionaries();
2171 }
2172
2173 fn apply_trained_dictionaries(&mut self) {
2179 let text_classes = [
2183 crate::classifier::Class::Text,
2184 crate::classifier::Class::Code,
2185 crate::classifier::Class::Sparse,
2186 ];
2187 let binary_classes = [crate::classifier::Class::Binary];
2188
2189 use rayon::prelude::*;
2202 let classifier = self.classifier;
2203 let dicts = &self.trained_dicts_by_class;
2204 let candidates: Vec<Option<(std::sync::Arc<[u8]>, u8)>> = self
2205 .drops
2206 .par_iter()
2207 .map(|d| {
2208 if d.codec != limnifs_core::codec::CODEC_ZSTD {
2209 return None;
2210 }
2211 let Some(plaintext) = d.plaintext.as_ref() else {
2212 return None;
2213 };
2214 let class = classifier.classify(plaintext);
2215 let dict_class = if text_classes.contains(&class) {
2216 crate::classifier::Class::Text
2217 } else if binary_classes.contains(&class) {
2218 crate::classifier::Class::Binary
2219 } else {
2220 return None;
2221 };
2222 let Some(dict) = dicts.get(&dict_class) else {
2223 return None;
2224 };
2225 let Ok(dict_compressed) = dict.compress(plaintext) else {
2226 return None;
2227 };
2228 if dict_compressed.len() < d.compressed.len() {
2229 Some((dict_compressed.into(), dict.id))
2230 } else {
2231 None
2232 }
2233 })
2234 .collect();
2235
2236 let saving: isize = candidates
2237 .iter()
2238 .zip(self.drops.iter())
2239 .map(|(c, d)| {
2240 c.as_ref().map_or(0, |(bytes, _)| {
2241 d.compressed.len() as isize - bytes.len() as isize
2242 })
2243 })
2244 .sum();
2245 let dict_bytes: usize = dicts.values().map(|d| d.content.len()).sum();
2246 if saving > dict_bytes as isize {
2247 for (d, candidate) in self.drops.iter_mut().zip(candidates) {
2248 if let Some((bytes, dict_id)) = candidate {
2249 d.compressed = bytes;
2250 d.dict_id = dict_id;
2251 }
2252 }
2253 } else {
2254 self.trained_dicts_by_class.clear();
2260 }
2261
2262 Self::release_dictionary_samples(self);
2263 }
2264
2265 fn release_dictionary_samples(ctx: &mut Self) {
2268 for d in &mut ctx.drops {
2269 d.plaintext = None;
2270 }
2271 ctx.dict_samples_by_class.clear();
2272 }
2273
2274 fn trace_phase(label: &str, start: std::time::Instant) {
2276 if std::env::var_os("LIMNIFS_TRACE_ASSEMBLE").is_some() {
2277 eprintln!("[assemble] {label}: {:?}", start.elapsed());
2278 }
2279 }
2280
2281 fn assemble(mut self) -> WriteArtifact {
2282 let t_assemble = std::time::Instant::now();
2283 let inode_count = self.inodes.len();
2284 let dir_count = self.dir_count;
2285 let drop_count = self.drops.len();
2286
2287 let t = std::time::Instant::now();
2293 let slabs = pack_slabs(&self.drops);
2294 Self::trace_phase("pack_slabs", t);
2295
2296 let t = std::time::Instant::now();
2300 if self.emit_shared_inline {
2301 self.build_shared_inline_table();
2302 }
2303 Self::trace_phase("shared_inline_table", t);
2304
2305 let mut metadata_blob = Vec::new();
2306 metadata_blob.extend_from_slice(&u32::try_from(self.inodes.len()).unwrap().to_le_bytes());
2307 for inode in &self.inodes {
2308 self.encode_inode(&mut metadata_blob, inode);
2309 }
2310 metadata_blob
2311 .extend_from_slice(&u32::try_from(self.dir_nodes.len()).unwrap().to_le_bytes());
2312 for node in &self.dir_nodes {
2313 metadata_blob.extend_from_slice(&node.bytes);
2314 }
2315 if !self.shared_inline_table.is_empty() {
2318 metadata_blob.extend_from_slice(
2319 &u32::try_from(self.shared_inline_table.len())
2320 .unwrap()
2321 .to_le_bytes(),
2322 );
2323 for entry in &self.shared_inline_table {
2324 let len = u32::try_from(entry.len()).expect("shared entry fits u32");
2325 metadata_blob.extend_from_slice(&len.to_le_bytes());
2326 metadata_blob.extend_from_slice(entry);
2327 }
2328 }
2329
2330 Self::trace_phase("metadata_encode", t);
2331 let uncompressed_len =
2338 u32::try_from(metadata_blob.len()).expect("metadata blob length fits u32");
2339 let t = std::time::Instant::now();
2340 let metadata_hash = hash_section(&metadata_blob);
2341 let metadata_codec = self.metadata_codec;
2342 let metadata_quality = if metadata_blob.len() > METADATA_LARGE_BLOB_THRESHOLD {
2343 METADATA_LARGE_BLOB_QUALITY
2344 } else {
2345 METADATA_SMALL_BLOB_QUALITY
2346 };
2347 let compressed_blob = if metadata_codec == limnifs_core::codec::CODEC_BROTLI {
2348 limnifs_core::codec::compress_brotli_with_quality(&metadata_blob, metadata_quality)
2349 .unwrap_or_else(|_| metadata_blob.clone())
2350 } else {
2351 limnifs_core::codec::compress(metadata_codec, &metadata_blob)
2352 .unwrap_or_else(|_| metadata_blob.clone())
2353 };
2354 Self::trace_phase("metadata_compress", t);
2355 let (on_wire_codec, on_wire_blob) = if compressed_blob.len() < metadata_blob.len() {
2356 (metadata_codec, compressed_blob)
2357 } else {
2358 (limnifs_core::codec::CODEC_STORE, metadata_blob.clone())
2359 };
2360
2361 let externalize_at = self
2365 .metadata_externalize_threshold
2366 .min(limnifs_core::metadata_reference::DEFAULT_INLINE_METADATA_MAX_BYTES as usize);
2367 let (metadata_sidecar, inline_data, metadata_locator_count) =
2368 if on_wire_blob.len() > externalize_at {
2369 let h = hash_section(&on_wire_blob);
2376 let mut h8 = String::with_capacity(8);
2377 for b in &h[..4] {
2378 h8.push_str(&format!("{b:02x}"));
2379 }
2380 let locator = format!("file:metadata-{h8}.bin");
2381 let sidecar = MetadataSidecar {
2382 bytes: on_wire_blob.clone(),
2383 locator,
2384 };
2385 (Some(sidecar), None, 1u32)
2386 } else {
2387 (None, Some(on_wire_blob.clone()), 0u32)
2388 };
2389
2390 let mut manifest = Vec::new();
2391
2392 let header_start = manifest.len();
2393 manifest.extend_from_slice(&ManifestHeader::current().to_bytes());
2394 let header_end = manifest.len();
2395
2396 let flags_start = manifest.len();
2397 manifest.push(FEATURE_FLAGS_SECTION_VERSION);
2398 manifest.extend_from_slice(&0u32.to_le_bytes());
2399 let flags_end = manifest.len();
2400
2401 let meta_ref_start = manifest.len();
2404 manifest.push(METADATA_REFERENCE_SECTION_VERSION_2);
2405 manifest.extend_from_slice(&metadata_hash);
2406 manifest.extend_from_slice(&uncompressed_len.to_le_bytes());
2407 manifest.push(on_wire_codec);
2408 manifest.extend_from_slice(&metadata_locator_count.to_le_bytes());
2409 if let Some(sidecar) = &metadata_sidecar {
2410 let loc_bytes = sidecar.locator.as_bytes();
2411 let loc_len = u32::try_from(loc_bytes.len()).expect("locator fits u32");
2412 manifest.extend_from_slice(&loc_len.to_le_bytes());
2413 manifest.extend_from_slice(loc_bytes);
2414 }
2415 match &inline_data {
2416 Some(blob) => {
2417 let inline_len = u32::try_from(blob.len()).expect("metadata fits u32");
2418 manifest.extend_from_slice(&inline_len.to_le_bytes());
2419 manifest.extend_from_slice(blob);
2420 }
2421 None => {
2422 manifest.extend_from_slice(&0u32.to_le_bytes());
2423 }
2424 }
2425 let meta_ref_end = manifest.len();
2426
2427 let slab_index_start = manifest.len();
2428 manifest.push(SLAB_INDEX_SECTION_VERSION);
2429 manifest.extend_from_slice(&u32::try_from(slabs.len()).unwrap().to_le_bytes());
2430 for slab in &slabs {
2431 manifest.extend_from_slice(&slab.id.to_bytes());
2432 manifest.extend_from_slice(&1u32.to_le_bytes());
2433 let loc_bytes = slab.locator.as_bytes();
2434 let loc_len = u32::try_from(loc_bytes.len()).expect("locator fits u32");
2435 manifest.extend_from_slice(&loc_len.to_le_bytes());
2436 manifest.extend_from_slice(loc_bytes);
2437 }
2438 let slab_index_end = manifest.len();
2439
2440 let history_start = manifest.len();
2441 manifest.push(HISTORY_SECTION_VERSION);
2442 manifest.extend_from_slice(&1u32.to_le_bytes());
2443 manifest.push(0x01);
2444 manifest.extend_from_slice(&0u64.to_le_bytes());
2445 manifest.extend_from_slice(&0u32.to_le_bytes());
2446 manifest.extend_from_slice(&0u32.to_le_bytes());
2447 let history_end = manifest.len();
2448
2449 let profile_desc_start = manifest.len();
2454 if let Some(ref name) = self.profile_name {
2455 let desc = limnifs_core::profile_descriptor::ProfileDescriptor {
2456 version: limnifs_core::profile_descriptor::PROFILE_DESCRIPTOR_SECTION_VERSION,
2457 profile_name: Some(name.clone()),
2458 blake3_hashing: true,
2459 cross_file_dedup: true,
2460 content_classification: !self.categorizers_disabled,
2461 integrity_verify: true,
2462 read_write: self.rw_mode,
2463 auto_turnover: self.auto_turnover,
2464 };
2465 limnifs_core::profile_descriptor::encode_profile_descriptor(&desc, &mut manifest);
2466 }
2467 let profile_desc_end = manifest.len();
2468
2469 if !self.trained_dicts_by_class.is_empty() {
2475 let dicts: Vec<_> = self
2476 .trained_dicts_by_class
2477 .values()
2478 .map(|d| limnifs_core::dictionary_section::Dictionary {
2479 codec_id: d.codec,
2480 class_id: d.id,
2481 data: d.content.clone(),
2482 })
2483 .collect();
2484 let section = limnifs_core::dictionary_section::DictionarySection {
2485 version: limnifs_core::dictionary_section::DICTIONARY_SECTION_VERSION,
2486 dicts,
2487 };
2488 limnifs_core::dictionary_section::encode_dictionary_section(§ion, &mut manifest);
2489 }
2490
2491 let dictionary_end = manifest.len();
2492
2493 let delta_linkage_hash = if let Some(base_root) = self.base_root {
2499 let delta_start = manifest.len();
2500 manifest.push(limnifs_core::delta_linkage::DELTA_LINKAGE_SECTION_VERSION);
2506 manifest.extend_from_slice(&base_root);
2507 manifest.extend_from_slice(&0u32.to_le_bytes());
2508 hash_section(&manifest[delta_start..])
2509 } else {
2510 hash_empty_section()
2511 };
2512 let _ = dictionary_end;
2513
2514 let hashes = SectionHashes {
2515 metadata: metadata_hash,
2516 format_header: hash_section(&manifest[header_start..header_end]),
2517 feature_flags: hash_section(&manifest[flags_start..flags_end]),
2518 metadata_reference: hash_section(&manifest[meta_ref_start..meta_ref_end]),
2519 slab_index: hash_section(&manifest[slab_index_start..slab_index_end]),
2520 crypto_params: hash_empty_section(),
2521 ec_params: hash_empty_section(),
2522 dms_policy: hash_empty_section(),
2523 delta_linkage: delta_linkage_hash,
2524 history: hash_section(&manifest[history_start..history_end]),
2525 };
2537 let merkle_root = compute_merkle_root(&hashes);
2538
2539 WriteArtifact {
2540 bytes: manifest,
2541 merkle_root,
2542 slabs,
2543 metadata_sidecar,
2544 inode_count,
2545 file_count: self.file_count,
2546 dir_count,
2547 drop_count,
2548 root_inode_number: self.root_inode_number,
2549 }
2550 }
2551
2552 fn encode_inode(&self, out: &mut Vec<u8>, inode: &PendingInode) {
2553 out.extend_from_slice(&inode.number.to_le_bytes());
2554 out.extend_from_slice(&inode.mode.to_le_bytes());
2555 out.extend_from_slice(&inode.uid.to_le_bytes());
2556 out.extend_from_slice(&inode.gid.to_le_bytes());
2557 out.extend_from_slice(&inode.mtime_ns.to_le_bytes());
2558 out.extend_from_slice(&inode.mtime_ns.to_le_bytes());
2559 let nlink = self.nlink_counts.get(&inode.number).copied().unwrap_or(1);
2560 out.extend_from_slice(&nlink.to_le_bytes());
2561 let xattrs: &[limnifs_core::inode::XAttr] = self
2562 .inode_xattrs
2563 .get(&inode.number)
2564 .map_or(inode.xattrs.as_slice(), std::convert::AsRef::as_ref);
2565 let mut flags_and_xattrs = move |out: &mut Vec<u8>, base: u8| {
2569 if xattrs.is_empty() {
2570 out.push(base);
2571 return;
2572 }
2573 out.push(base | limnifs_core::inode::INODE_FLAG_HAS_XATTRS);
2574 let count = u32::try_from(xattrs.len()).expect("xattr count fits u32");
2575 out.extend_from_slice(&count.to_le_bytes());
2576 for x in xattrs {
2577 out.push(x.namespace);
2578 let key = x.key.as_bytes();
2579 let key_len = u32::try_from(key.len()).expect("xattr key fits u32");
2580 out.extend_from_slice(&key_len.to_le_bytes());
2581 out.extend_from_slice(key);
2582 let value_len = u32::try_from(x.value.len()).expect("xattr value fits u32");
2583 out.extend_from_slice(&value_len.to_le_bytes());
2584 out.extend_from_slice(&x.value);
2585 }
2586 };
2587 match &inode.content {
2588 PendingContent::Inline(data) => {
2589 let h = hash_section(data);
2590 if let Some(&idx) = self.shared_inline_map.get(&h) {
2591 flags_and_xattrs(out, INODE_FLAG_INLINE_DATA | INODE_FLAG_SHARED_INLINE);
2593 out.extend_from_slice(&(idx as u32).to_le_bytes());
2594 } else {
2595 flags_and_xattrs(out, INODE_FLAG_INLINE_DATA);
2596 let len = u32::try_from(data.len()).expect("data fits u32");
2597 out.extend_from_slice(&len.to_le_bytes());
2598 out.extend_from_slice(data);
2599 }
2600 }
2601 PendingContent::DropBacked { file_len, slices } => {
2602 flags_and_xattrs(out, 0x00);
2603 let slice_count = u32::try_from(slices.len()).expect("slice count fits u32");
2604 out.extend_from_slice(&slice_count.to_le_bytes());
2605 for slice in slices {
2606 out.extend_from_slice(&slice.file_byte_start.to_le_bytes());
2607 out.extend_from_slice(&slice.file_byte_end.to_le_bytes());
2608 out.extend_from_slice(&slice.drop_id);
2609 out.extend_from_slice(&0u32.to_le_bytes());
2611 let drop_byte_len = u32::try_from(slice.file_byte_end - slice.file_byte_start)
2615 .expect("slice range fits u32");
2616 out.extend_from_slice(&drop_byte_len.to_le_bytes());
2617 }
2618 let _ = file_len;
2619 }
2620 PendingContent::Symlink(target) => {
2621 flags_and_xattrs(out, 0x00);
2625 let t = target.as_bytes();
2626 let len = u32::try_from(t.len()).expect("target fits u32");
2627 out.extend_from_slice(&len.to_le_bytes());
2628 out.extend_from_slice(t);
2629 }
2630 PendingContent::Directory(entries) => {
2631 flags_and_xattrs(out, 0x00);
2632 let node = self
2633 .dir_nodes
2634 .iter()
2635 .find(|n| n.entries == *entries)
2636 .expect("directory node must exist");
2637 out.extend_from_slice(&node.hash);
2638 }
2639 }
2640 }
2641}
2642
2643fn sidecar_name(locator: &str) -> Result<&str, WriteError> {
2649 limnifs_core::locator::local_sidecar_name(locator)
2650 .map_err(|e| WriteError::Io(std::io::Error::other(format!("{e}"))))
2651}
2652
2653fn encode_dir_node(entries: &[(String, u64, u8)]) -> DirNode {
2654 let mut bytes = Vec::new();
2655 bytes.push(1u8);
2656 let count = u32::try_from(entries.len()).expect("entry count fits u32");
2657 bytes.extend_from_slice(&count.to_le_bytes());
2658 for (name, inode_number, entry_type) in entries {
2659 let name_bytes = name.as_bytes();
2660 let name_len = u32::try_from(name_bytes.len()).expect("name fits u32");
2661 bytes.extend_from_slice(&name_len.to_le_bytes());
2662 bytes.extend_from_slice(name_bytes);
2663 bytes.extend_from_slice(&inode_number.to_le_bytes());
2664 bytes.push(*entry_type);
2665 }
2666 let hash = hash_section(&bytes);
2667 DirNode {
2668 entries: entries.to_vec(),
2669 bytes,
2670 hash,
2671 }
2672}
2673
2674fn pack_slabs(drops: &[PendingDrop]) -> Vec<SlabArtifact> {
2683 let local_drops: Vec<&PendingDrop> = drops
2688 .iter()
2689 .filter(|d| d.codec != limnifs_core::codec::CODEC_REFERENCED)
2690 .collect();
2691 if local_drops.is_empty() {
2692 return Vec::new();
2693 }
2694
2695 let max_content = MAX_SLAB_TOTAL_BYTES.saturating_sub(SLAB_HEADER_LEN);
2696
2697 let mut slab_groups: Vec<Vec<&PendingDrop>> = Vec::new();
2702 let mut current: Vec<&PendingDrop> = Vec::new();
2703 let mut current_size: usize = 0;
2704
2705 for drop in &local_drops {
2706 let footprint = drop.slab_footprint();
2707 if !current.is_empty() && current_size + footprint > max_content {
2708 slab_groups.push(std::mem::take(&mut current));
2709 current_size = 0;
2710 }
2711 current.push(*drop);
2712 current_size += footprint;
2713 }
2714 if !current.is_empty() {
2715 slab_groups.push(current);
2716 }
2717
2718 use rayon::prelude::*;
2724 slab_groups
2725 .par_iter()
2726 .enumerate()
2727 .map(|(ordinal, group)| {
2728 let ordinal_u64 = u64::try_from(ordinal).expect("slab count fits u64");
2729 encode_slab(ordinal_u64, group)
2730 })
2731 .collect()
2732}
2733
2734fn encode_slab(ordinal: u64, drops: &[&PendingDrop]) -> SlabArtifact {
2738 const DROP_RECORD_LEN: usize = 50;
2745 let mut drop_records = Vec::with_capacity(drops.len() * DROP_RECORD_LEN);
2746 let mut solid_window = Vec::new();
2747 let mut drop_ids = Vec::with_capacity(drops.len());
2748 let mut offset_in_window: u32 = 0;
2749
2750 for drop in drops {
2751 let plaintext_len = drop.plaintext_len_value();
2752 let window_len = drop.len_in_window();
2753 drop_records.extend_from_slice(&drop.id);
2754 drop_records.extend_from_slice(&plaintext_len.to_le_bytes());
2755 drop_records.extend_from_slice(&[drop.codec, 0x00, 0x00]);
2757 drop_records.push(0x00); drop_records.extend_from_slice(&offset_in_window.to_le_bytes());
2759 drop_records.extend_from_slice(&window_len.to_le_bytes());
2760 drop_records.push(drop.dict_id); drop_records.push(drop.flags); solid_window.extend_from_slice(&drop.compressed);
2763 drop_ids.push(drop.id);
2764 offset_in_window = offset_in_window
2765 .checked_add(window_len)
2766 .expect("slab window size fits u32");
2767 }
2768
2769 let slab_content = [&drop_records[..], &solid_window[..]].concat();
2770 let slab_hash = hash_section(&slab_content);
2771 let slab_id = SlabId::new(ordinal, slab_hash);
2772
2773 let total_length = SLAB_HEADER_LEN + slab_content.len();
2774 let mut slab_bytes = Vec::with_capacity(total_length);
2775 slab_bytes.extend_from_slice(b"LIM1");
2776 slab_bytes.extend_from_slice(&1u16.to_le_bytes()); slab_bytes.extend_from_slice(&slab_id.to_bytes());
2778 slab_bytes.extend_from_slice(
2779 &u64::try_from(total_length)
2780 .unwrap_or(u64::MAX)
2781 .to_le_bytes(),
2782 );
2783 slab_bytes.push(0x00);
2784 slab_bytes.push(0x00);
2785 slab_bytes.extend_from_slice(&slab_content);
2786
2787 let mut h8 = String::with_capacity(8);
2791 for b in &slab_id.hash[..4] {
2792 h8.push_str(&format!("{b:02x}"));
2793 }
2794 let locator = format!("file:slab-{ordinal}-{h8}.bin");
2795
2796 SlabArtifact {
2797 id: slab_id,
2798 bytes: slab_bytes,
2799 locator,
2800 drop_ids,
2801 }
2802}
2803
2804#[cfg(test)]
2805fn pseudo_random_bytes(seed: u64, count: usize) -> Vec<u8> {
2806 let mut state = seed;
2807 let mut out = Vec::with_capacity(count);
2808 for _ in 0..count {
2809 state = state
2810 .wrapping_mul(6_364_136_223_846_793_005)
2811 .wrapping_add(1_442_695_040_888_963_407);
2812 out.push(u8::try_from(state >> 56).expect("fits u8"));
2813 }
2814 out
2815}
2816
2817#[cfg(test)]
2818mod tests {
2819 use super::*;
2820 use limnifs_core::ManifestCursor;
2821
2822 #[test]
2823 fn write_stream_packs_single_named_stream() {
2824 let temp = std::env::temp_dir().join(format!(
2828 "limnifs-write-stream-test-{}-{}",
2829 std::process::id(),
2830 std::time::SystemTime::now()
2831 .duration_since(std::time::UNIX_EPOCH)
2832 .unwrap()
2833 .as_nanos()
2834 ));
2835 std::fs::create_dir_all(&temp).expect("create temp dir");
2836
2837 let content = b"stream test content line\n".repeat(10_000); let cursor = std::io::Cursor::new(content.clone());
2839 let config = WriteConfig::default_v0_1();
2840 let artifact = write_stream("streamed.txt", cursor, &config).expect("write_stream");
2841
2842 assert!(!artifact.bytes.is_empty(), "manifest bytes non-empty");
2843 assert!(!artifact.slabs.is_empty(), "at least one slab produced");
2844 let total_drop_bytes: usize = artifact.slabs.iter().map(|s| s.bytes.len()).sum();
2849 assert!(total_drop_bytes > 0, "drops non-empty");
2850
2851 let _ = std::fs::remove_dir_all(&temp);
2852 }
2853
2854 #[test]
2855 fn write_layer_references_base_drops() {
2856 let temp = std::env::temp_dir().join(format!(
2862 "limnifs-write-layer-test-{}-{}",
2863 std::process::id(),
2864 std::time::SystemTime::now()
2865 .duration_since(std::time::UNIX_EPOCH)
2866 .unwrap()
2867 .as_nanos()
2868 ));
2869 std::fs::create_dir_all(&temp).expect("create temp dir");
2870
2871 let base_dir = temp.join("base");
2873 std::fs::create_dir_all(&base_dir).expect("base dir");
2874 let text = b"layer test content line\n".repeat(50_000); std::fs::write(base_dir.join("shared.txt"), &text).expect("write shared");
2876 std::fs::write(base_dir.join("base-only.txt"), b"only in base").expect("write base-only");
2877
2878 let config = WriteConfig::default_v0_1();
2879 let base_artifact = write_directory_with_config(&base_dir, &config).expect("base");
2880
2881 let base_manifest = temp.join("base.lim");
2882 std::fs::write(&base_manifest, &base_artifact.bytes).expect("write base manifest");
2883 for slab in &base_artifact.slabs {
2884 let slab_name = sidecar_name(&slab.locator).expect("slab locator");
2885 std::fs::write(temp.join(slab_name), &slab.bytes).expect("write base slab");
2886 }
2887
2888 let layer_dir = temp.join("layer");
2890 std::fs::create_dir_all(&layer_dir).expect("layer dir");
2891 std::fs::write(layer_dir.join("shared.txt"), &text).expect("write shared in layer");
2892 std::fs::write(layer_dir.join("new.txt"), b"fresh content in layer").expect("write new");
2893
2894 let layer_artifact = write_layer(&base_manifest, &layer_dir, &config).expect("layer");
2895
2896 let layer_slab_bytes: usize = layer_artifact.slabs.iter().map(|s| s.bytes.len()).sum();
2899 let base_slab_bytes: usize = base_artifact.slabs.iter().map(|s| s.bytes.len()).sum();
2900 assert!(
2901 layer_slab_bytes < base_slab_bytes / 4,
2902 "layer slabs ({}) should be much smaller than base ({}) — layering failed",
2903 layer_slab_bytes,
2904 base_slab_bytes
2905 );
2906
2907 let base_root = base_artifact.merkle_root.as_bytes();
2910 assert!(
2911 layer_artifact
2912 .bytes
2913 .windows(32)
2914 .any(|w| w == base_root.as_slice()),
2915 "layer manifest must contain base's ManifestRoot bytes"
2916 );
2917
2918 let _ = std::fs::remove_dir_all(&temp);
2919 }
2920
2921 #[test]
2922 fn tournament_short_circuits_on_highly_compressible_chunk() {
2923 let chunk = b"hello world ".repeat(500);
2926 let tunables = limnifs_core::codec::CodecTunables::default();
2927 let tournament = TournamentSpec {
2928 codec_ids: vec![
2929 limnifs_core::codec::CODEC_LZ4,
2930 limnifs_core::codec::CODEC_BROTLI,
2931 ],
2932 min_size: 16,
2933 skip_for_binary: false,
2934 short_circuit_permille: 250,
2935 };
2936 let (codec_id, compressed) = compress_chunk_with_tournament(
2937 &chunk,
2938 classifier::Class::Text,
2939 limnifs_core::codec::CODEC_BROTLI,
2940 limnifs_core::codec::CODEC_LZ4,
2941 &tunables,
2942 &tournament,
2943 );
2944 assert_eq!(codec_id, limnifs_core::codec::CODEC_LZ4);
2945 assert!(compressed.len() < chunk.len());
2946 }
2947
2948 #[test]
2949 fn tournament_runs_all_codecs_when_short_circuit_disabled() {
2950 let chunk = b"hello world ".repeat(500);
2956 let tunables = limnifs_core::codec::CodecTunables::default();
2957 let tournament = TournamentSpec {
2958 codec_ids: vec![
2959 limnifs_core::codec::CODEC_LZ4,
2960 limnifs_core::codec::CODEC_BROTLI,
2961 limnifs_core::codec::CODEC_ZSTD,
2962 ],
2963 min_size: 16,
2964 skip_for_binary: false,
2965 short_circuit_permille: 0,
2966 };
2967 let (codec_id, compressed) = compress_chunk_with_tournament(
2968 &chunk,
2969 classifier::Class::Text,
2970 limnifs_core::codec::CODEC_BROTLI,
2971 limnifs_core::codec::CODEC_LZ4,
2972 &tunables,
2973 &tournament,
2974 );
2975 assert!(
2979 codec_id == limnifs_core::codec::CODEC_ZSTD
2980 || codec_id == limnifs_core::codec::CODEC_BROTLI,
2981 "expected ZSTD or Brotli to win, got codec {codec_id}"
2982 );
2983 assert!(compressed.len() < chunk.len());
2984 }
2985
2986 #[test]
2987 fn tournament_skips_for_binary_when_configured() {
2988 let chunk = vec![0u8; 4096];
2989 let tunables = limnifs_core::codec::CodecTunables::default();
2990 let tournament = TournamentSpec {
2991 codec_ids: vec![limnifs_core::codec::CODEC_BROTLI],
2992 min_size: 16,
2993 skip_for_binary: true,
2994 short_circuit_permille: 250,
2995 };
2996 let (codec_id, _compressed) = compress_chunk_with_tournament(
2997 &chunk,
2998 classifier::Class::Binary,
2999 limnifs_core::codec::CODEC_BROTLI,
3000 limnifs_core::codec::CODEC_LZ4,
3001 &tunables,
3002 &tournament,
3003 );
3004 assert_eq!(codec_id, limnifs_core::codec::CODEC_LZ4);
3006 }
3007
3008 #[test]
3009 fn tournament_small_chunk_uses_preferred_codec() {
3010 let chunk = b"tiny";
3011 let tunables = limnifs_core::codec::CodecTunables::default();
3012 let tournament = TournamentSpec {
3013 codec_ids: vec![limnifs_core::codec::CODEC_BROTLI],
3014 min_size: 1024,
3015 skip_for_binary: false,
3016 short_circuit_permille: 0,
3017 };
3018 let (codec_id, _compressed) = compress_chunk_with_tournament(
3019 chunk,
3020 classifier::Class::Text,
3021 limnifs_core::codec::CODEC_BROTLI,
3022 limnifs_core::codec::CODEC_LZ4,
3023 &tunables,
3024 &tournament,
3025 );
3026 assert_eq!(codec_id, limnifs_core::codec::CODEC_STORE);
3028 }
3029
3030 #[test]
3031 fn tournament_falls_back_to_store_when_no_codec_compresses() {
3032 let chunk = pseudo_random_bytes(42, 4096);
3036 let tunables = limnifs_core::codec::CodecTunables::default();
3037 let tournament = TournamentSpec {
3038 codec_ids: vec![limnifs_core::codec::CODEC_LZ4],
3039 min_size: 16,
3040 skip_for_binary: false,
3041 short_circuit_permille: 0,
3042 };
3043 let (codec_id, compressed) = compress_chunk_with_tournament(
3044 &chunk,
3045 classifier::Class::Binary,
3046 limnifs_core::codec::CODEC_BROTLI,
3047 limnifs_core::codec::CODEC_LZ4,
3048 &tunables,
3049 &tournament,
3050 );
3051 assert_eq!(codec_id, limnifs_core::codec::CODEC_STORE);
3052 assert_eq!(compressed.len(), chunk.len());
3053 }
3054
3055 #[test]
3056 fn dictionaries_enabled_emits_dictionary_section_when_enough_samples() {
3057 let temp = std::env::temp_dir().join(format!(
3062 "limnifs-write-test-{}-dict-{}",
3063 std::process::id(),
3064 std::time::SystemTime::now()
3065 .duration_since(std::time::UNIX_EPOCH)
3066 .map(|d| d.as_nanos() as u64)
3067 .unwrap_or(0),
3068 ));
3069 let _ = std::fs::remove_dir_all(&temp);
3070 std::fs::create_dir_all(&temp).expect("mkdir");
3071
3072 for i in 0..200 {
3075 let content = format!(
3077 "function test_case_{i}() {{ return constant + {i}; }}\n\
3078 // shared comment line {i}\n\
3079 struct Foo {{ x: i32 }} // type {i}\n"
3080 )
3081 .repeat(5);
3082 let path = temp.join(format!("file_{i:04}.txt"));
3083 std::fs::write(&path, content.as_bytes()).expect("write");
3084 }
3085
3086 let mut config = crate::profile::balanced();
3087 config.defaults.text_codec = "zstd".into();
3089 config.defaults.metadata_codec = "zstd".into();
3093 config.dictionaries.enabled = true;
3094 config.dictionaries.min_class_size = 50;
3095 config.dictionaries.max_dict_size = 8192;
3096
3097 let artifact = write_directory_with_config(&temp, &config).expect("write");
3098 std::fs::remove_dir_all(&temp).ok();
3099
3100 let mut cursor = ManifestCursor::new(&artifact.bytes);
3105 let _ = limnifs_core::parse_manifest_header(&mut cursor).expect("header");
3106 let _ = limnifs_core::parse_feature_flags_section(&mut cursor).expect("flags");
3107 let _ = limnifs_core::parse_metadata_reference(&mut cursor).expect("meta_ref");
3108 let _ = limnifs_core::parse_slab_index(&mut cursor).expect("slab_index");
3109 let _ = limnifs_core::parse_history(&mut cursor).expect("history");
3110 let _remaining = cursor.remaining_len();
3113 }
3114
3115 #[test]
3116 fn write_empty_directory() {
3117 let temp =
3118 std::env::temp_dir().join(format!("limnifs-write-test-{}-empty", std::process::id()));
3119 std::fs::create_dir_all(&temp).expect("create temp dir");
3120 let artifact = write_directory(&temp).expect("write succeeds");
3121 std::fs::remove_dir_all(&temp).ok();
3122 assert!(artifact.inode_count >= 1);
3123 assert_eq!(artifact.file_count, 0);
3124 assert_eq!(artifact.dir_count, 1);
3125 assert!(artifact.slabs.is_empty());
3126 }
3127
3128 #[test]
3129 fn write_small_file_inline() {
3130 let temp =
3131 std::env::temp_dir().join(format!("limnifs-write-test-{}-small", std::process::id()));
3132 std::fs::create_dir_all(&temp).expect("create temp dir");
3133 std::fs::write(temp.join("hello.txt"), b"hello world").expect("write file");
3134 let artifact = write_directory(&temp).expect("write succeeds");
3135 std::fs::remove_dir_all(&temp).ok();
3136 assert_eq!(artifact.file_count, 1);
3137 assert!(artifact.slabs.is_empty());
3138 assert_eq!(artifact.drop_count, 0);
3139 }
3140
3141 #[test]
3142 fn write_large_file_uses_slab() {
3143 let temp =
3144 std::env::temp_dir().join(format!("limnifs-write-test-{}-large", std::process::id()));
3145 std::fs::create_dir_all(&temp).expect("create temp dir");
3146 let large_data = vec![0xABu8; INLINE_THRESHOLD + 100];
3147 std::fs::write(temp.join("big.bin"), &large_data).expect("write big");
3148 let artifact = write_directory(&temp).expect("write succeeds");
3149 std::fs::remove_dir_all(&temp).ok();
3150 assert_eq!(artifact.drop_count, 1);
3151 assert_eq!(artifact.slabs.len(), 1);
3152 }
3153
3154 #[test]
3155 fn write_mixed_inline_and_large() {
3156 let temp =
3157 std::env::temp_dir().join(format!("limnifs-write-test-{}-mix", std::process::id()));
3158 std::fs::create_dir_all(&temp).expect("create temp dir");
3159 std::fs::write(temp.join("small.txt"), b"tiny").expect("write small");
3160 std::fs::write(temp.join("large.bin"), vec![0xCDu8; INLINE_THRESHOLD * 2])
3161 .expect("write large");
3162 let artifact = write_directory(&temp).expect("write succeeds");
3163 std::fs::remove_dir_all(&temp).ok();
3164 assert_eq!(artifact.file_count, 2);
3165 assert_eq!(artifact.drop_count, 1);
3166 assert_eq!(artifact.slabs.len(), 1);
3167 }
3168
3169 #[test]
3170 fn deduplicates_identical_large_files() {
3171 let temp =
3172 std::env::temp_dir().join(format!("limnifs-write-test-{}-dedup", std::process::id()));
3173 std::fs::create_dir_all(&temp).expect("create temp dir");
3174 let data = vec![0x77u8; INLINE_THRESHOLD + 10];
3175 std::fs::write(temp.join("a.bin"), &data).expect("write a");
3176 std::fs::write(temp.join("b.bin"), &data).expect("write b");
3177 let artifact = write_directory(&temp).expect("write succeeds");
3178 std::fs::remove_dir_all(&temp).ok();
3179 assert_eq!(artifact.drop_count, 1);
3180 }
3181
3182 #[test]
3183 fn write_and_verify_roundtrip() {
3184 let temp = std::env::temp_dir().join(format!(
3185 "limnifs-write-test-{}-roundtrip",
3186 std::process::id()
3187 ));
3188 std::fs::create_dir_all(&temp).expect("create temp dir");
3189 std::fs::write(temp.join("a.txt"), b"aaa").expect("write a");
3190 std::fs::write(temp.join("b.txt"), b"bbb").expect("write b");
3191 std::fs::create_dir_all(temp.join("sub")).expect("create sub");
3192 std::fs::write(temp.join("sub").join("c.txt"), b"ccc").expect("write c");
3193 let artifact = write_directory(&temp).expect("write succeeds");
3194 std::fs::remove_dir_all(&temp).ok();
3195 assert_eq!(artifact.file_count, 3);
3196 assert_eq!(artifact.dir_count, 2);
3197
3198 let mut cursor = ManifestCursor::new(&artifact.bytes);
3199 limnifs_core::parse_manifest_header(&mut cursor).expect("header");
3200 limnifs_core::parse_feature_flags_section(&mut cursor).expect("flags");
3201 let meta_ref = limnifs_core::parse_metadata_reference(&mut cursor).expect("meta ref");
3202 assert!(meta_ref.is_inlined());
3203 let slab_index = limnifs_core::parse_slab_index(&mut cursor).expect("slab index");
3204 assert_eq!(slab_index.len(), 0);
3205 limnifs_core::parse_history(&mut cursor).expect("history");
3206 }
3207
3208 #[test]
3209 fn write_deterministic() {
3210 let temp =
3211 std::env::temp_dir().join(format!("limnifs-write-test-{}-det", std::process::id()));
3212 std::fs::create_dir_all(&temp).expect("create temp dir");
3213 std::fs::write(temp.join("x.txt"), b"xxx").expect("write x");
3214
3215 let a1 = write_directory(&temp).expect("first write");
3216 let a2 = write_directory(&temp).expect("second write");
3217 std::fs::remove_dir_all(&temp).ok();
3218
3219 assert_eq!(a1.bytes, a2.bytes);
3220 assert_eq!(a1.merkle_root, a2.merkle_root);
3221 }
3222
3223 #[test]
3224 fn slab_parses_correctly() {
3225 let temp =
3226 std::env::temp_dir().join(format!("limnifs-write-test-{}-slab", std::process::id()));
3227 std::fs::create_dir_all(&temp).expect("create temp dir");
3228 std::fs::write(temp.join("big.bin"), vec![0x11u8; INLINE_THRESHOLD + 1])
3229 .expect("write big");
3230 let artifact = write_directory(&temp).expect("write succeeds");
3231 std::fs::remove_dir_all(&temp).ok();
3232
3233 let slab_bytes = &artifact.slabs[0].bytes;
3234 let mut cursor = ManifestCursor::new(slab_bytes);
3235 let slab_header = limnifs_core::parse_slab_header(&mut cursor).expect("slab header parses");
3236 assert_eq!(
3237 slab_header.format_version,
3238 limnifs_core::slab::SLAB_FORMAT_VERSION
3239 );
3240 assert!(!slab_header.is_sealed());
3241 assert!(!slab_header.has_erasure_coding());
3242
3243 let drop_record =
3244 limnifs_core::parse_drop_record(&mut cursor, &slab_header).expect("drop record parses");
3245 assert_eq!(drop_record.plaintext_len as usize, INLINE_THRESHOLD + 1);
3246 }
3247
3248 #[test]
3249 fn fastcdc_produces_multiple_chunks_for_large_files() {
3250 let temp = std::env::temp_dir().join(format!(
3253 "limnifs-write-test-{}-cdc-multi",
3254 std::process::id()
3255 ));
3256 std::fs::create_dir_all(&temp).expect("create temp dir");
3257 let data = pseudo_random_bytes(42, 1024 * 1024);
3258 std::fs::write(temp.join("big.bin"), &data).expect("write big");
3259 let artifact = write_directory(&temp).expect("write succeeds");
3260 std::fs::remove_dir_all(&temp).ok();
3261 assert!(
3262 artifact.drop_count > 1,
3263 "expected FastCDC to produce multiple drops for 1 MiB input, got {}",
3264 artifact.drop_count
3265 );
3266 }
3267
3268 #[test]
3269 fn fastcdc_deduplicates_shared_substrings() {
3270 let temp = std::env::temp_dir().join(format!(
3274 "limnifs-write-test-{}-cdc-dedup",
3275 std::process::id()
3276 ));
3277 std::fs::create_dir_all(&temp).expect("create temp dir");
3278 let shared = pseudo_random_bytes(7, 512 * 1024);
3279 let mut a = Vec::with_capacity(shared.len() + 1024);
3280 a.extend_from_slice(&pseudo_random_bytes(1, 1024));
3281 a.extend_from_slice(&shared);
3282 let mut b = Vec::with_capacity(shared.len() + 2048);
3283 b.extend_from_slice(&pseudo_random_bytes(2, 2048));
3284 b.extend_from_slice(&shared);
3285 std::fs::write(temp.join("a.bin"), &a).expect("write a");
3286 std::fs::write(temp.join("b.bin"), &b).expect("write b");
3287
3288 let temp_a = std::env::temp_dir().join(format!(
3290 "limnifs-write-test-{}-cdc-dedup-a",
3291 std::process::id()
3292 ));
3293 std::fs::create_dir_all(&temp_a).expect("create temp_a");
3294 std::fs::write(temp_a.join("a.bin"), &a).expect("write a");
3295 let artifact_a = write_directory(&temp_a).expect("a writes");
3296 std::fs::remove_dir_all(&temp_a).ok();
3297
3298 let temp_b = std::env::temp_dir().join(format!(
3299 "limnifs-write-test-{}-cdc-dedup-b",
3300 std::process::id()
3301 ));
3302 std::fs::create_dir_all(&temp_b).expect("create temp_b");
3303 std::fs::write(temp_b.join("b.bin"), &b).expect("write b");
3304 let artifact_b = write_directory(&temp_b).expect("b writes");
3305 std::fs::remove_dir_all(&temp_b).ok();
3306
3307 let artifact_both = write_directory(&temp).expect("both write");
3308 std::fs::remove_dir_all(&temp).ok();
3309
3310 let sum_alone = artifact_a.drop_count + artifact_b.drop_count;
3311 assert!(
3312 artifact_both.drop_count < sum_alone,
3313 "expected dedup win: both together = {} drops, sum alone = {} drops",
3314 artifact_both.drop_count,
3315 sum_alone
3316 );
3317 }
3318
3319 #[test]
3320 fn slab_splits_when_content_exceeds_ceiling() {
3321 let temp =
3327 std::env::temp_dir().join(format!("limnifs-write-test-{}-split", std::process::id()));
3328 std::fs::create_dir_all(&temp).expect("create temp dir");
3329 for i in 0..7u32 {
3330 let data = pseudo_random_bytes(u64::from(i), 10 * 1024 * 1024);
3332 std::fs::write(temp.join(format!("big-{i}.bin")), &data).expect("write big");
3333 }
3334 let artifact = write_directory(&temp).expect("write succeeds");
3335 std::fs::remove_dir_all(&temp).ok();
3336
3337 assert!(
3339 artifact.slabs.len() >= 2,
3340 "expected at least 2 slabs for 70 MiB of incompressible data, got {}",
3341 artifact.slabs.len()
3342 );
3343 for slab in &artifact.slabs {
3344 assert!(
3345 slab.bytes.len() <= MAX_SLAB_TOTAL_BYTES,
3346 "slab {} is {} bytes (> {} ceiling)",
3347 slab.id.ordinal,
3348 slab.bytes.len(),
3349 MAX_SLAB_TOTAL_BYTES,
3350 );
3351 }
3352 let total_drop_ids: usize = artifact.slabs.iter().map(|s| s.drop_ids.len()).sum();
3354 assert_eq!(
3355 total_drop_ids, artifact.drop_count,
3356 "drop_ids count across slabs must match WriteArtifact.drop_count",
3357 );
3358 }
3359}