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, 0o644, &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
551#[cfg(feature = "xattr")]
553fn to_core_xattrs(raw: &[(String, Vec<u8>)]) -> Vec<limnifs_core::inode::XAttr> {
554 raw.iter()
555 .map(|(key, value)| limnifs_core::inode::XAttr {
556 namespace: 0,
557 key: key.clone(),
558 value: value.clone(),
559 })
560 .collect()
561}
562
563fn io_core(e: limnifs_core::CoreError) -> WriteError {
564 WriteError::Io(std::io::Error::other(format!("base image load: {e}")))
565}
566
567fn write_directory_body(ctx: &mut WriteContext, config: &WriteConfig) -> Result<(), WriteError> {
572 use rayon::prelude::*;
573
574 ctx.metadata_codec = config
575 .metadata_codec_id()
576 .unwrap_or(limnifs_core::codec::CODEC_BROTLI);
577
578 ctx.chunker = chunker_from_config(config)?;
579
580 let pending = std::mem::take(&mut ctx.pending_files);
581 if pending.is_empty() {
582 return Ok(());
583 }
584 ctx.inline_threshold = config.defaults.inline_threshold as usize;
585 ctx.metadata_externalize_threshold = config.defaults.metadata_externalize_threshold;
586 ctx.emit_shared_inline = config.defaults.shared_inline;
587 let chunker = ctx.chunker.clone();
588 let classifier = ctx.classifier;
589 let text_codec = config.text_codec_id().unwrap_or(0x04);
590 let binary_codec = config.binary_codec_id().unwrap_or(0x01);
591 let tunables = config.to_core_tunables();
592 let use_categorizers = !config.categorizers.is_empty();
593 let skip_chunking = config.skip_chunking;
594 let registry = config
595 .codec_registry()
596 .map_err(|e| WriteError::Io(std::io::Error::other(format!("codec registry: {e}"))))?;
597 let tournament_codec_ids: Vec<u8> = config
598 .tournament
599 .codecs
600 .iter()
601 .filter_map(|n| registry.lookup_by_name(n))
602 .collect();
603 let tournament_spec = TournamentSpec {
604 codec_ids: tournament_codec_ids,
605 min_size: config.tournament.min_size_threshold as usize,
606 skip_for_binary: config.tournament.skip_for_binary,
607 short_circuit_permille: config.tournament.short_circuit_threshold,
608 };
609 let base_drop_index: Option<&dyn BaseDropSet> = ctx.base_drop_index.as_deref();
610 let inline_threshold = ctx.inline_threshold;
611 let max_drop_size = config.defaults.max_drop_size as usize;
612 let seekable_drops = config.defaults.seekable_drops;
613 let seekable_drops = config.defaults.seekable_drops;
614 let results: Vec<ChunkedFileResult> = pending
615 .par_iter()
616 .map(|pf| {
617 process_file(
618 pf,
619 &chunker,
620 classifier,
621 text_codec,
622 binary_codec,
623 &tunables,
624 use_categorizers,
625 skip_chunking,
626 &tournament_spec,
627 base_drop_index,
628 inline_threshold,
629 max_drop_size,
630 seekable_drops,
631 config.categorizers.as_slice(),
632 &|name| {
633 config
634 .codec_registry()
635 .ok()
636 .and_then(|r| r.lookup_by_name(name))
637 },
638 )
639 })
640 .collect::<Result<Vec<_>, _>>()?;
641
642 for (pf, result) in pending.iter().zip(results) {
643 ctx.merge_chunked_file(pf, result);
644 }
645 ctx.train_and_apply_dictionary(&config.dictionaries);
646 Ok(())
647}
648
649pub fn write_directory_with_config(
651 root: &Path,
652 config: &WriteConfig,
653) -> Result<WriteArtifact, WriteError> {
654 let mut ctx = WriteContext::new();
655 ctx.chunker = chunker_from_config(config)?;
656 ctx.categorizers_disabled = config.categorizers.is_empty();
657 ctx.rw_mode = matches!(config.mode, crate::config::ImageMode::ReadWrite(_));
658 ctx.auto_turnover = config.turnover_threshold > 0;
659 ctx.collect_dict_samples = config.dictionaries.enabled;
660
661 write_directory_streaming(&mut ctx, root, config)?;
662 Ok(ctx.assemble())
663}
664
665fn write_directory_streaming(
679 ctx: &mut WriteContext,
680 root: &Path,
681 config: &WriteConfig,
682) -> Result<(), WriteError> {
683 use rayon::prelude::*;
684
685 ctx.metadata_codec = config
686 .metadata_codec_id()
687 .unwrap_or(limnifs_core::codec::CODEC_BROTLI);
688
689 ctx.chunker = chunker_from_config(config)?;
690
691 let chunker = ctx.chunker.clone();
692 let classifier = ctx.classifier;
693 let text_codec = config.text_codec_id().unwrap_or(0x04);
694 let binary_codec = config.binary_codec_id().unwrap_or(0x01);
695 let tunables = config.to_core_tunables();
696 let use_categorizers = !config.categorizers.is_empty();
697 let skip_chunking = config.skip_chunking;
698 let registry = config
699 .codec_registry()
700 .map_err(|e| WriteError::Io(std::io::Error::other(format!("codec registry: {e}"))))?;
701 let tournament_codec_ids: Vec<u8> = config
702 .tournament
703 .codecs
704 .iter()
705 .filter_map(|n| registry.lookup_by_name(n))
706 .collect();
707 let tournament_spec = TournamentSpec {
708 codec_ids: tournament_codec_ids,
709 min_size: config.tournament.min_size_threshold as usize,
710 skip_for_binary: config.tournament.skip_for_binary,
711 short_circuit_permille: config.tournament.short_circuit_threshold,
712 };
713 let base_drop_index = ctx.base_drop_index.clone();
716 let inline_threshold = ctx.inline_threshold;
717 let max_drop_size = config.defaults.max_drop_size as usize;
718 let seekable_drops = config.defaults.seekable_drops;
719
720 ctx.inline_threshold = config.defaults.inline_threshold as usize;
721 ctx.metadata_externalize_threshold = config.defaults.metadata_externalize_threshold;
722 ctx.emit_shared_inline = config.defaults.shared_inline;
723
724 const PIPELINE_CAPACITY: usize = 256;
728 let (tx, rx) = std::sync::mpsc::sync_channel::<PendingFile>(PIPELINE_CAPACITY);
729 ctx.pending_sink = Some(tx);
730
731 let (root_inode_number, mut results): (
732 u64,
733 Vec<(usize, PendingFile, Result<ChunkedFileResult, WriteError>)>,
734 ) = {
735 let survey = survey_tree(root)?;
745 std::thread::scope(|scope| {
746 let producer = {
747 let ctx = &mut *ctx;
748 let root = root;
749 scope.spawn(move || {
750 let r = ctx.fold_survey(root, &survey, None);
751 ctx.pending_sink = None;
755 r
756 })
757 };
758 let results = rx
761 .into_iter()
762 .enumerate()
763 .par_bridge()
764 .map(|(i, pf)| {
765 let r = process_file(
766 &pf,
767 &chunker,
768 classifier,
769 text_codec,
770 binary_codec,
771 &tunables,
772 use_categorizers,
773 skip_chunking,
774 &tournament_spec,
775 base_drop_index.as_deref(),
776 inline_threshold,
777 max_drop_size,
778 seekable_drops,
779 config.categorizers.as_slice(),
780 &|name| {
781 config
782 .codec_registry()
783 .ok()
784 .and_then(|r| r.lookup_by_name(name))
785 },
786 );
787 (i, pf, r)
788 })
789 .collect();
790 let joined = producer
791 .join()
792 .unwrap_or_else(|_| {
793 Err(WriteError::Io(std::io::Error::other(
794 "walk thread panicked",
795 )))
796 })
797 .map(|n| (n, results));
798 joined
801 })
802 }?;
803 ctx.pending_sink = None;
804 ctx.root_inode_number = root_inode_number;
805
806 results.sort_unstable_by_key(|(i, _, _)| *i);
807 for (_, pf, r) in results {
811 ctx.merge_chunked_file(&pf, r?);
812 }
813 ctx.train_and_apply_dictionary(&config.dictionaries);
814 Ok(())
815}
816
817pub(crate) type RawDrop = ([u8; 32], Vec<u8>, std::sync::Arc<[u8]>, u8, u8);
824pub(crate) struct ChunkedFileResult {
826 drops: Vec<RawDrop>, slices: Vec<PendingSlice>,
828}
829
830struct TournamentSpec {
838 codec_ids: Vec<u8>,
842 min_size: usize,
846 skip_for_binary: bool,
850 short_circuit_permille: u32,
855}
856
857fn chunker_from_config(config: &WriteConfig) -> Result<ParallelFastCDC, WriteError> {
876 ParallelFastCDC::new(
877 config.chunking.min_chunk_size as usize,
878 config.chunking.avg_chunk_size as usize,
879 config.chunking.max_chunk_size as usize,
880 )
881 .map_err(|e| WriteError::Io(std::io::Error::other(format!("chunking config: {e}"))))
882}
883
884pub(crate) fn seekable_or_monolithic(
891 codec: u8,
892 plaintext: &[u8],
893 compressed: std::sync::Arc<[u8]>,
894 tunables: &limnifs_core::codec::CodecTunables,
895 seekable_drops: bool,
896 threshold: usize,
897) -> (std::sync::Arc<[u8]>, u8) {
898 use limnifs_core::seekable::{
899 encode_seekable, is_seekable_codec, DROP_FLAG_SEEKABLE as FLAG, SEEKABLE_EMISSION_THRESHOLD,
900 };
901 if seekable_drops && plaintext.len() > threshold && is_seekable_codec(codec) {
902 if let Ok(container) = encode_seekable(codec, plaintext, tunables) {
903 return (container.into(), FLAG);
904 }
905 }
906 (compressed, 0)
907}
908
909pub(crate) const SEEKABLE_CHUNK_EMISSION_THRESHOLD: usize =
916 limnifs_core::seekable::SEEKABLE_FRAME_SIZE;
917
918fn process_whole_file_drop(
919 pf: &PendingFile,
920 data: &[u8],
921 cat: file_categorizer::Categorization,
922 tunables: &limnifs_core::codec::CodecTunables,
923 seekable_drops: bool,
924) -> Result<ChunkedFileResult, WriteError> {
925 let _ = pf;
926 let drop_id = hash_section(data);
927 let file_len = u64::try_from(data.len()).unwrap_or(u64::MAX);
928
929 let (mut best_codec, mut best_compressed): (u8, std::sync::Arc<[u8]>) =
934 match limnifs_core::codec::compress_with_tunables(
935 limnifs_core::codec::CODEC_BROTLI,
936 data,
937 tunables,
938 ) {
939 Ok(c) => (limnifs_core::codec::CODEC_BROTLI, c.into()),
940 Err(_) => match limnifs_core::codec::compress_with_tunables(
941 limnifs_core::codec::CODEC_ZSTD,
942 data,
943 tunables,
944 ) {
945 Ok(c) => (limnifs_core::codec::CODEC_ZSTD, c.into()),
946 Err(_) => (limnifs_core::codec::CODEC_STORE, data.to_vec().into()),
947 },
948 };
949
950 let brotli_ratio = best_compressed.len() as f64 / data.len() as f64;
954 if brotli_ratio > 0.05 {
955 if let Ok(zstd_c) = limnifs_core::codec::compress_with_tunables(
956 limnifs_core::codec::CODEC_ZSTD,
957 data,
958 tunables,
959 ) {
960 if zstd_c.len() < best_compressed.len() {
961 best_codec = limnifs_core::codec::CODEC_ZSTD;
962 best_compressed = zstd_c.into();
963 }
964 }
965 }
966
967 let general_ratio = best_compressed.len() as f64 / data.len() as f64;
973 if general_ratio > 0.15 || cat.codec_id == limnifs_core::codec::CODEC_RICEPP {
974 let spec_result = if cat.codec_id == limnifs_core::codec::CODEC_FSST_BROTLI {
978 limnifs_core::codec::fsst_brotli::compress_with_baseline(data, Some(&best_compressed))
979 } else {
980 limnifs_core::codec::compress_with_tunables(cat.codec_id, data, tunables)
984 };
985 if let Ok(spec_c) = spec_result {
986 if spec_c.len() < best_compressed.len() {
987 best_codec = cat.codec_id;
988 best_compressed = spec_c.into();
989 }
990 }
991 }
992
993 let (best_compressed, flags) = seekable_or_monolithic(
994 best_codec,
995 data,
996 best_compressed,
997 tunables,
998 seekable_drops,
999 limnifs_core::seekable::SEEKABLE_EMISSION_THRESHOLD,
1000 );
1001 Ok(ChunkedFileResult {
1002 drops: vec![(drop_id, data.to_vec(), best_compressed, best_codec, flags)],
1003 slices: vec![PendingSlice {
1004 drop_id,
1005 file_byte_start: 0,
1006 file_byte_end: file_len,
1007 }],
1008 })
1009}
1010
1011fn compress_chunk_with_tournament(
1033 chunk: &[u8],
1034 class: classifier::Class,
1035 text_codec: u8,
1036 binary_codec: u8,
1037 tunables: &limnifs_core::codec::CodecTunables,
1038 tournament: &TournamentSpec,
1039) -> (u8, std::sync::Arc<[u8]>) {
1040 use classifier::Class;
1041
1042 let preferred = match class {
1043 Class::Binary => binary_codec,
1044 Class::Text | Class::Code | Class::Sparse => text_codec,
1045 _ => limnifs_core::codec::CODEC_STORE,
1046 };
1047
1048 if preferred == limnifs_core::codec::CODEC_STORE {
1049 return (limnifs_core::codec::CODEC_STORE, chunk.to_vec().into());
1050 }
1051 if class == Class::Binary && tournament.skip_for_binary {
1052 return compress_chunk_one(chunk, preferred, tunables);
1053 }
1054 if chunk.len() < tournament.min_size {
1055 return compress_chunk_one(chunk, preferred, tunables);
1056 }
1057
1058 let mut best: Option<(u8, std::sync::Arc<[u8]>)> = None;
1059 for &codec_id in &tournament.codec_ids {
1060 if codec_id == limnifs_core::codec::CODEC_STORE {
1061 continue;
1062 }
1063 let c = match limnifs_core::codec::compress_with_tunables(codec_id, chunk, tunables) {
1064 Ok(c) => c,
1065 Err(_) => continue,
1066 };
1067 if c.len() >= chunk.len() {
1068 continue;
1069 }
1070 let ratio_permille = (c.len() as u64 * 1000 / chunk.len() as u64) as u32;
1071 let is_best_so_far = best.as_ref().map_or(true, |(_, b)| c.len() < b.len());
1072 if is_best_so_far {
1073 best = Some((codec_id, c.into()));
1074 }
1075 if tournament.short_circuit_permille > 0
1076 && ratio_permille <= tournament.short_circuit_permille
1077 {
1078 break;
1079 }
1080 }
1081
1082 best.unwrap_or_else(|| (limnifs_core::codec::CODEC_STORE, chunk.to_vec().into()))
1083}
1084
1085fn compress_chunk_one(
1088 chunk: &[u8],
1089 codec_id: u8,
1090 tunables: &limnifs_core::codec::CodecTunables,
1091) -> (u8, std::sync::Arc<[u8]>) {
1092 if codec_id == limnifs_core::codec::CODEC_STORE {
1093 return (limnifs_core::codec::CODEC_STORE, chunk.to_vec().into());
1094 }
1095 match limnifs_core::codec::compress_with_tunables(codec_id, chunk, tunables) {
1096 Ok(c) if c.len() < chunk.len() => (codec_id, c.into()),
1097 _ => (limnifs_core::codec::CODEC_STORE, chunk.to_vec().into()),
1098 }
1099}
1100
1101fn process_file(
1105 pf: &PendingFile,
1106 chunker: &dyn Chunker,
1107 classifier: classifier::Classifier,
1108 text_codec: u8,
1109 binary_codec: u8,
1110 tunables: &limnifs_core::codec::CodecTunables,
1111 use_categorizers: bool,
1112 skip_chunking: bool,
1113 tournament: &TournamentSpec,
1114 base_drop_index: Option<&dyn BaseDropSet>,
1115 inline_threshold: usize,
1116 max_drop_size: usize,
1117 seekable_drops: bool,
1118 categorizer_config: &[crate::config::CategorizerConfig],
1119 codec_name_resolver: &dyn Fn(&str) -> Option<u8>,
1120) -> Result<ChunkedFileResult, WriteError> {
1121 let file_len_estimate = std::fs::metadata(&pf.path)
1132 .map(|m| m.len() as usize)
1133 .unwrap_or(0);
1134 let mmap_handle: memmap2::Mmap;
1141 let small: Vec<u8>;
1142 let data: &[u8] = if file_len_estimate >= MMAP_READ_THRESHOLD {
1143 let file = std::fs::File::open(&pf.path)?;
1144 #[allow(unsafe_code)]
1148 let mapped = unsafe { memmap2::Mmap::map(&file) }.map_err(WriteError::Io)?;
1149 mmap_handle = mapped;
1150 &mmap_handle[..]
1151 } else {
1152 small = std::fs::read(&pf.path)?;
1153 &small[..]
1154 };
1155 let file_len = data.len();
1156
1157 if skip_chunking && file_len > inline_threshold {
1164 let drop_id = hash_section(&data);
1165 let class = classifier.classify(&data);
1166 let preferred_codec = match class {
1167 classifier::Class::Binary => binary_codec,
1168 _ => text_codec,
1169 };
1170 let (codec_id, compressed): (u8, std::sync::Arc<[u8]>) =
1171 match limnifs_core::codec::compress_with_tunables(preferred_codec, &data, tunables) {
1172 Ok(c) if c.len() < data.len() => (preferred_codec, c.into()),
1173 _ => (limnifs_core::codec::CODEC_STORE, data.to_vec().into()),
1174 };
1175 let (compressed, flags) = seekable_or_monolithic(
1176 codec_id,
1177 &data,
1178 compressed,
1179 tunables,
1180 seekable_drops,
1181 SEEKABLE_CHUNK_EMISSION_THRESHOLD,
1182 );
1183 return Ok(ChunkedFileResult {
1184 drops: vec![(drop_id, data.to_vec(), compressed, codec_id, flags)],
1185 slices: vec![PendingSlice {
1186 drop_id,
1187 file_byte_start: 0,
1188 file_byte_end: file_len as u64,
1189 }],
1190 });
1191 }
1192
1193 if use_categorizers {
1194 let config_cat = file_categorizer::ConfigCategorizer::new(categorizer_config.to_vec());
1197 if let Some(cat) = config_cat.categorize(&pf.path, &data) {
1198 if let Some(codec_id) = file_categorizer::resolve_config_categorization(
1199 &cat,
1200 categorizer_config,
1201 codec_name_resolver,
1202 ) {
1203 let within_cap = max_drop_size == 0 || file_len <= max_drop_size;
1204 let needs_whole_file = matches!(
1205 codec_id,
1206 limnifs_core::codec::CODEC_FLAC | limnifs_core::codec::CODEC_RICEPP
1207 );
1208 if within_cap && (needs_whole_file || file_len <= WHOLE_FILE_MAX_SIZE) {
1209 let mut cat = cat;
1210 cat.codec_id = codec_id;
1211 return process_whole_file_drop(pf, &data, cat, tunables, seekable_drops);
1212 }
1213 }
1214 }
1215 if let Some(cat) = file_categorizer::default_registry().categorize(&pf.path, &data) {
1216 let needs_whole_file = matches!(
1217 cat.codec_id,
1218 limnifs_core::codec::CODEC_FLAC | limnifs_core::codec::CODEC_RICEPP
1219 );
1220 let within_cap = max_drop_size == 0 || file_len <= max_drop_size;
1223 if within_cap && (needs_whole_file || file_len <= WHOLE_FILE_MAX_SIZE) {
1224 return process_whole_file_drop(pf, &data, cat, tunables, seekable_drops);
1225 }
1226 }
1227 }
1228
1229 let chunks = chunker.chunk_slice(&data);
1230
1231 use rayon::prelude::*;
1240 let drop_ids: Vec<[u8; 32]> = chunks.par_iter().map(|chunk| hash_section(chunk)).collect();
1241
1242 let mut slices = Vec::with_capacity(chunks.len());
1243 let mut file_offset: u64 = 0;
1244 let mut seen_in_file: std::collections::HashSet<[u8; 32]> =
1245 std::collections::HashSet::with_capacity(chunks.len());
1246 let mut unique_chunks: Vec<(&[u8], [u8; 32])> = Vec::with_capacity(chunks.len());
1247 for (chunk, drop_id) in chunks.iter().zip(drop_ids) {
1248 let chunk_len = u64::try_from(chunk.len()).expect("chunk len fits u64");
1249 slices.push(PendingSlice {
1250 drop_id,
1251 file_byte_start: file_offset,
1252 file_byte_end: file_offset + chunk_len,
1253 });
1254 file_offset += chunk_len;
1255 if seen_in_file.insert(drop_id) {
1256 unique_chunks.push((chunk, drop_id));
1257 }
1258 }
1259
1260 thread_local! {
1272 static COMPRESS_CACHE: std::cell::RefCell<std::collections::HashMap<[u8; 32], (u8, std::sync::Arc<[u8]>)>> =
1273 std::cell::RefCell::new(std::collections::HashMap::new());
1274 }
1275 const COMPRESS_CACHE_MAX_ENTRIES: usize = 100_000;
1276 let drops: Vec<RawDrop> = unique_chunks
1277 .par_iter()
1278 .map(|(chunk, drop_id)| {
1279 if let Some(base) = base_drop_index {
1282 if base.base_contains(drop_id) {
1283 return (*drop_id, Vec::new(), Vec::new().into(), CODEC_REFERENCED, 0);
1284 }
1285 }
1286 let class = classifier.classify(chunk);
1287 let cached = COMPRESS_CACHE.with(|c| {
1290 c.borrow()
1291 .get(drop_id)
1292 .map(|(cid, comp)| (*cid, comp.clone()))
1293 });
1294 let (codec_id, compressed) = if let Some(c) = cached {
1295 c
1296 } else {
1297 let new = compress_chunk_with_tournament(
1298 chunk,
1299 class,
1300 text_codec,
1301 binary_codec,
1302 tunables,
1303 tournament,
1304 );
1305 COMPRESS_CACHE.with(|c| {
1307 let mut cache = c.borrow_mut();
1308 if cache.len() < COMPRESS_CACHE_MAX_ENTRIES {
1309 cache.insert(*drop_id, new.clone());
1311 }
1312 });
1313 new
1314 };
1315 let (compressed, flags) = seekable_or_monolithic(
1321 codec_id,
1322 chunk,
1323 compressed,
1324 tunables,
1325 seekable_drops,
1326 SEEKABLE_CHUNK_EMISSION_THRESHOLD,
1327 );
1328 (*drop_id, chunk.to_vec(), compressed, codec_id, flags)
1329 })
1330 .collect();
1331
1332 let _ = file_len;
1333 Ok(ChunkedFileResult { drops, slices })
1334}
1335
1336struct PendingDrop {
1337 id: [u8; 32],
1338 plaintext_len: u32,
1344 compressed: std::sync::Arc<[u8]>,
1345 codec: u8,
1346 dict_id: u8,
1350 plaintext: Option<Vec<u8>>,
1355 flags: u8,
1359}
1360
1361impl PendingDrop {
1362 fn len_in_window(&self) -> u32 {
1366 u32::try_from(self.compressed.len()).expect("compressed fits u32")
1367 }
1368
1369 fn plaintext_len_value(&self) -> u32 {
1371 self.plaintext_len
1372 }
1373
1374 fn slab_footprint(&self) -> usize {
1377 48 + self.compressed.len()
1378 }
1379}
1380
1381struct PendingSlice {
1386 drop_id: [u8; 32],
1387 file_byte_start: u64,
1388 file_byte_end: u64,
1389}
1390
1391#[derive(Clone)]
1394struct PendingFile {
1395 inode_number: u64,
1396 path: PathBuf,
1397 mtime_ns: u64,
1398 file_len: u64,
1399 mode: u32,
1400 uid: u32,
1401 gid: u32,
1402}
1403
1404#[derive(Clone, Default)]
1408struct SurveyMeta {
1409 is_dir: bool,
1410 is_file: bool,
1411 is_symlink: bool,
1412 #[cfg(unix)]
1413 is_fifo: bool,
1414 #[cfg(unix)]
1415 is_socket: bool,
1416 #[cfg(unix)]
1417 is_block_device: bool,
1418 #[cfg(unix)]
1419 is_char_device: bool,
1420 len: u64,
1421 mtime_ns: u64,
1422 #[cfg(unix)]
1423 mode: u32,
1424 #[cfg(unix)]
1425 uid: u32,
1426 #[cfg(unix)]
1427 gid: u32,
1428 #[cfg(unix)]
1432 dev: u64,
1433 #[cfg(unix)]
1434 ino: u64,
1435 #[cfg(feature = "xattr")]
1438 xattrs: Vec<(String, Vec<u8>)>,
1439}
1440
1441impl SurveyMeta {
1442 fn identity(&self) -> (u32, u32, u32) {
1446 #[cfg(unix)]
1447 {
1448 (self.mode, self.uid, self.gid)
1449 }
1450 #[cfg(not(unix))]
1451 {
1452 let ty = if self.is_dir {
1453 limnifs_core::inode::S_IFDIR
1454 } else if self.is_symlink {
1455 limnifs_core::inode::S_IFLNK
1456 } else {
1457 limnifs_core::inode::S_IFREG
1458 };
1459 let perms = if self.is_dir || self.is_symlink {
1460 0o755
1461 } else {
1462 0o644
1463 };
1464 (ty | perms, 0, 0)
1465 }
1466 }
1467}
1468
1469struct SurveyNode {
1472 meta: SurveyMeta,
1473 children: Vec<(String, SurveyNode)>,
1474 symlink_target: Option<String>,
1478}
1479
1480impl SurveyNode {
1481 fn meta(&self) -> &SurveyMeta {
1482 &self.meta
1483 }
1484}
1485
1486fn survey_meta_of(meta: &std::fs::Metadata) -> SurveyMeta {
1487 #[cfg(unix)]
1488 use std::os::unix::fs::FileTypeExt as _;
1489 let ft = meta.file_type();
1490 let mtime_ns = meta
1491 .modified()
1492 .ok()
1493 .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
1494 .map_or(0u128, |d| d.as_nanos());
1495 SurveyMeta {
1496 is_dir: ft.is_dir(),
1497 is_file: ft.is_file(),
1498 is_symlink: ft.is_symlink(),
1499 #[cfg(unix)]
1500 is_fifo: ft.is_fifo(),
1501 #[cfg(unix)]
1502 is_socket: ft.is_socket(),
1503 #[cfg(unix)]
1504 is_block_device: ft.is_block_device(),
1505 #[cfg(unix)]
1506 is_char_device: ft.is_char_device(),
1507 len: meta.len(),
1508 mtime_ns: mtime_ns.try_into().unwrap_or(0),
1509 #[cfg(unix)]
1510 mode: {
1511 use std::os::unix::fs::MetadataExt as _;
1512 meta.mode()
1513 },
1514 #[cfg(unix)]
1515 uid: {
1516 use std::os::unix::fs::MetadataExt as _;
1517 meta.uid()
1518 },
1519 #[cfg(unix)]
1520 gid: {
1521 use std::os::unix::fs::MetadataExt as _;
1522 meta.gid()
1523 },
1524 #[cfg(unix)]
1525 dev: {
1526 use std::os::unix::fs::MetadataExt as _;
1527 meta.dev()
1528 },
1529 #[cfg(unix)]
1530 ino: {
1531 use std::os::unix::fs::MetadataExt as _;
1532 meta.ino()
1533 },
1534 #[cfg(feature = "xattr")]
1535 xattrs: Vec::new(),
1536 }
1537}
1538
1539#[cfg(all(unix, feature = "xattr"))]
1543fn collect_xattrs(path: &Path) -> Vec<(String, Vec<u8>)> {
1544 const VOLATILE: &[&str] = &[
1545 "com.apple.provenance",
1546 "com.apple.quarantine",
1547 "com.apple.lastuseddate",
1548 "com.apple.macl",
1549 "com.apple.filesec",
1550 ];
1551 const TOTAL_CAP: usize = 64 * 1024;
1552 let Ok(names) = xattr::list(path) else {
1553 return Vec::new();
1554 };
1555 let mut names: Vec<String> = names
1556 .filter_map(|n| n.into_string().ok())
1557 .filter(|n| {
1558 !n.starts_with("system.")
1559 && !n.starts_with("security.")
1560 && !n.starts_with("trusted.")
1561 && !VOLATILE.contains(&n.as_str())
1562 })
1563 .collect();
1564 names.sort();
1565 let mut out = Vec::new();
1566 let mut total = 0usize;
1567 for name in names {
1568 let Ok(Some(value)) = xattr::get(path, &name) else {
1569 continue;
1570 };
1571 total += name.len() + value.len();
1572 if total > TOTAL_CAP {
1573 break;
1574 }
1575 out.push((name, value));
1576 }
1577 out
1578}
1579
1580fn survey_node(path: &Path) -> Result<SurveyNode, WriteError> {
1581 use rayon::prelude::*;
1582 let meta = std::fs::symlink_metadata(path)?;
1583 let mut sm = survey_meta_of(&meta);
1584 #[cfg(all(unix, feature = "xattr"))]
1585 {
1586 if !sm.is_symlink {
1587 sm.xattrs = collect_xattrs(path);
1588 }
1589 }
1590 if sm.is_symlink {
1591 let target = std::fs::read_link(path)?;
1592 let target = target
1593 .to_str()
1594 .ok_or_else(|| WriteError::UnsupportedFileType {
1595 path: path.to_path_buf(),
1596 kind: format!("symlink with non-UTF-8 target ({})", target.display()),
1597 })?
1598 .to_owned();
1599 return Ok(SurveyNode {
1600 meta: sm,
1601 children: Vec::new(),
1602 symlink_target: Some(target),
1603 });
1604 }
1605 if !sm.is_dir {
1606 return Ok(SurveyNode {
1607 meta: sm,
1608 children: Vec::new(),
1609 symlink_target: None,
1610 });
1611 }
1612 let mut named: Vec<(String, PathBuf)> = std::fs::read_dir(path)?
1613 .filter_map(|entry| {
1614 entry
1615 .ok()
1616 .map(|e| (e.file_name().to_string_lossy().into_owned(), e.path()))
1617 })
1618 .collect();
1619 named.sort_by(|a, b| a.0.cmp(&b.0));
1620 named
1621 .par_iter()
1622 .map(|(name, child)| survey_node(child).map(|node| (name.clone(), node)))
1623 .collect::<Result<Vec<_>, WriteError>>()
1624 .map(|children| SurveyNode {
1625 meta: sm,
1626 children,
1627 symlink_target: None,
1628 })
1629}
1630
1631fn survey_tree(root: &Path) -> Result<SurveyNode, WriteError> {
1636 survey_node(root)
1637}
1638
1639struct PendingInode {
1640 number: u64,
1641 mode: u32,
1642 uid: u32,
1643 gid: u32,
1644 mtime_ns: u64,
1645 xattrs: Vec<limnifs_core::inode::XAttr>,
1646 content: PendingContent,
1647}
1648
1649enum PendingContent {
1650 Inline(Vec<u8>),
1651 Symlink(String),
1653 DropBacked {
1654 file_len: u64,
1655 slices: Vec<PendingSlice>,
1656 },
1657 Directory(Vec<(String, u64, u8)>),
1658}
1659
1660struct DirNode {
1661 entries: Vec<(String, u64, u8)>,
1662 bytes: Vec<u8>,
1663 hash: [u8; 32],
1664}
1665
1666struct WriteContext {
1667 next_inode: u64,
1668 hardlink_targets: std::collections::HashMap<(u64, u64), u64>,
1671 nlink_counts: std::collections::HashMap<u64, u32>,
1675 inode_xattrs: std::collections::HashMap<u64, Vec<limnifs_core::inode::XAttr>>,
1680 inodes: Vec<PendingInode>,
1681 dir_nodes: Vec<DirNode>,
1682 drops: Vec<PendingDrop>,
1683 drop_index: HashSet<[u8; 32]>,
1684 pending_files: Vec<PendingFile>,
1685 file_count: usize,
1686 dir_count: usize,
1687 root_inode_number: u64,
1688 chunker: ParallelFastCDC,
1689 classifier: classifier::Classifier,
1690 shared_inline_map: HashMap<[u8; 32], usize>,
1691 shared_inline_table: Vec<Vec<u8>>,
1692 base_dictionaries: Option<Vec<crate::dictionary::TrainedDictionary>>,
1698 profile_name: Option<String>,
1700 metadata_codec: u8,
1703 categorizers_disabled: bool,
1705 rw_mode: bool,
1707 auto_turnover: bool,
1709 collect_dict_samples: bool,
1712 dict_samples_by_class: HashMap<crate::classifier::Class, Vec<Vec<u8>>>,
1719 trained_dicts_by_class: HashMap<crate::classifier::Class, crate::dictionary::TrainedDictionary>,
1724 base_drop_index: Option<std::sync::Arc<dyn BaseDropSet>>,
1730 base_root: Option<[u8; 32]>,
1735 metadata_externalize_threshold: usize,
1740 emit_shared_inline: bool,
1746 inline_threshold: usize,
1751 pending_sink: Option<std::sync::mpsc::SyncSender<PendingFile>>,
1757}
1758
1759impl WriteContext {
1760 const MAX_DICT_SAMPLES: usize = 1000;
1763
1764 fn new() -> Self {
1765 Self {
1766 next_inode: 1,
1767 hardlink_targets: std::collections::HashMap::new(),
1768 nlink_counts: std::collections::HashMap::new(),
1769 inode_xattrs: std::collections::HashMap::new(),
1770 inodes: Vec::new(),
1771 dir_nodes: Vec::new(),
1772 drops: Vec::new(),
1773 drop_index: HashSet::new(),
1774 pending_files: Vec::new(),
1775 file_count: 0,
1776 dir_count: 0,
1777 root_inode_number: 0,
1778 chunker: ParallelFastCDC::default(),
1779 classifier: classifier::Classifier,
1780 shared_inline_map: HashMap::new(),
1781 shared_inline_table: Vec::new(),
1782 base_dictionaries: None,
1783 profile_name: None,
1784 metadata_codec: limnifs_core::codec::CODEC_BROTLI,
1785 categorizers_disabled: false,
1786 rw_mode: false,
1787 auto_turnover: false,
1788 collect_dict_samples: false,
1789 dict_samples_by_class: HashMap::new(),
1790 trained_dicts_by_class: HashMap::new(),
1791 base_drop_index: None,
1792 base_root: None,
1793 pending_sink: None,
1794 inline_threshold: INLINE_THRESHOLD,
1795 metadata_externalize_threshold: METADATA_EXTERNALIZE_THRESHOLD,
1796 emit_shared_inline: true,
1797 }
1798 }
1799
1800 fn alloc_inode(&mut self) -> u64 {
1801 let n = self.next_inode;
1802 self.next_inode += 1;
1803 n
1804 }
1805
1806 fn build_shared_inline_table(&mut self) {
1810 let mut counts: HashMap<[u8; 32], usize> = HashMap::new();
1811 for inode in &self.inodes {
1812 if let PendingContent::Inline(data) = &inode.content {
1813 let h = hash_section(data);
1814 *counts.entry(h).or_default() += 1;
1815 }
1816 }
1817 for inode in &self.inodes {
1819 if let PendingContent::Inline(data) = &inode.content {
1820 let h = hash_section(data);
1821 if counts.get(&h).copied().unwrap_or(0) > 1
1822 && !self.shared_inline_map.contains_key(&h)
1823 {
1824 let idx = self.shared_inline_table.len();
1825 self.shared_inline_table.push(data.clone());
1826 self.shared_inline_map.insert(h, idx);
1827 }
1828 }
1829 }
1830 }
1831
1832 fn merge_chunked_file(&mut self, pf: &PendingFile, result: ChunkedFileResult) {
1835 for (drop_id, plaintext, compressed, codec, flags) in result.drops {
1836 if self.drop_index.insert(drop_id) {
1837 let retain_plaintext =
1843 self.collect_dict_samples && codec == limnifs_core::codec::CODEC_ZSTD;
1844 if retain_plaintext {
1845 let total: usize = self.dict_samples_by_class.values().map(Vec::len).sum();
1846 if total < Self::MAX_DICT_SAMPLES {
1847 let class = self.classifier.classify(&plaintext);
1848 self.dict_samples_by_class
1849 .entry(class)
1850 .or_default()
1851 .push(plaintext.clone());
1852 }
1853 }
1854 self.drops.push(PendingDrop {
1855 id: drop_id,
1856 plaintext_len: u32::try_from(plaintext.len()).unwrap_or(u32::MAX),
1857 compressed,
1858 codec,
1859 dict_id: limnifs_core::drop_record::NO_DICT,
1860 plaintext: if retain_plaintext {
1861 Some(plaintext)
1862 } else {
1863 None
1864 },
1865 flags,
1866 });
1867 }
1868 }
1869 self.inodes.push(PendingInode {
1870 number: pf.inode_number,
1871 mode: pf.mode,
1872 uid: pf.uid,
1873 gid: pf.gid,
1874 mtime_ns: pf.mtime_ns,
1875 xattrs: Vec::new(),
1876 content: PendingContent::DropBacked {
1877 file_len: pf.file_len,
1878 slices: result.slices,
1879 },
1880 });
1881 }
1882
1883 #[allow(dead_code)]
1894 fn deepen_drop(&self, drop_id: [u8; 32], plaintext: &[u8]) -> PendingDrop {
1895 let class = self.classifier.classify(plaintext);
1896 let (codec, compressed): (u8, std::sync::Arc<[u8]>) = match class {
1897 classifier::Class::Text | classifier::Class::Code | classifier::Class::Binary => {
1898 let c = limnifs_core::codec::compress_lz4_with_size(plaintext);
1899 (limnifs_core::codec::CODEC_LZ4, c.into())
1900 }
1901 _ => (limnifs_core::codec::CODEC_STORE, plaintext.to_vec().into()),
1902 };
1903 PendingDrop {
1904 id: drop_id,
1905 plaintext_len: u32::try_from(plaintext.len()).unwrap_or(u32::MAX),
1906 compressed,
1907 codec,
1908 dict_id: limnifs_core::drop_record::NO_DICT,
1909 plaintext: None,
1910 flags: 0,
1911 }
1912 }
1913
1914 fn walk(&mut self, path: &Path) -> Result<u64, WriteError> {
1915 let survey = survey_tree(path)?;
1923 self.fold_survey(path, &survey, None)
1924 }
1925
1926 fn fold_survey(
1930 &mut self,
1931 path: &Path,
1932 node: &SurveyNode,
1933 symlink_target: Option<&str>,
1934 ) -> Result<u64, WriteError> {
1935 let meta = node.meta();
1936 if let Some(target) = symlink_target {
1937 let inode_number = self.alloc_inode();
1938 let (mode, uid, gid) = meta.identity();
1939 self.inodes.push(PendingInode {
1940 number: inode_number,
1941 mode,
1942 uid,
1943 gid,
1944 mtime_ns: meta.mtime_ns,
1945 xattrs: Vec::new(),
1946 content: PendingContent::Symlink(target.to_owned()),
1947 });
1948 return Ok(inode_number);
1949 }
1950 if meta.is_dir {
1951 self.dir_count += 1;
1952 let inode_number = self.alloc_inode();
1953 let mut entries: Vec<(String, u64, u8)> = Vec::new();
1954
1955 for (name, child) in &node.children {
1956 let child_path = path.join(name);
1957 let child_inode =
1958 self.fold_survey(&child_path, child, child.symlink_target.as_deref())?;
1959 let entry_type = if child.meta().is_symlink {
1960 0x03
1961 } else if child.meta().is_dir {
1962 0x02
1963 } else {
1964 0x01
1965 };
1966 entries.push((name.clone(), child_inode, entry_type));
1967 }
1968
1969 entries.sort_by(|a, b| a.0.cmp(&b.0));
1972 let dir_node = encode_dir_node(&entries);
1973 self.dir_nodes.push(dir_node);
1974 #[cfg(feature = "xattr")]
1975 if !meta.xattrs.is_empty() {
1976 self.inode_xattrs
1977 .insert(inode_number, to_core_xattrs(&meta.xattrs));
1978 }
1979 let (mode, uid, gid) = meta.identity();
1980 self.inodes.push(PendingInode {
1981 number: inode_number,
1982 mode,
1983 uid,
1984 gid,
1985 mtime_ns: meta.mtime_ns,
1986 xattrs: Vec::new(),
1987 content: PendingContent::Directory(entries),
1988 });
1989 Ok(inode_number)
1990 } else if meta.is_file {
1991 #[cfg(unix)]
1995 if let Some(&existing) = self.hardlink_targets.get(&(meta.dev, meta.ino)) {
1996 *self.nlink_counts.entry(existing).or_insert(1) += 1;
1997 return Ok(existing);
1998 }
1999 self.file_count += 1;
2000 let inode_number = self.alloc_inode();
2001 #[cfg(unix)]
2002 {
2003 self.hardlink_targets
2004 .insert((meta.dev, meta.ino), inode_number);
2005 }
2006 let file_len = meta.len;
2007 crate::progress::emit_file(path, file_len);
2008
2009 if file_len <= u64::try_from(self.inline_threshold).unwrap_or(u64::MAX) {
2010 let data = std::fs::read(path)?;
2011 #[cfg(feature = "xattr")]
2012 if !meta.xattrs.is_empty() {
2013 self.inode_xattrs
2014 .insert(inode_number, to_core_xattrs(&meta.xattrs));
2015 }
2016 let (mode, uid, gid) = meta.identity();
2017 self.inodes.push(PendingInode {
2018 number: inode_number,
2019 mode,
2020 uid,
2021 gid,
2022 mtime_ns: meta.mtime_ns,
2023 xattrs: Vec::new(),
2024 content: PendingContent::Inline(data),
2025 });
2026 } else {
2027 #[cfg(feature = "xattr")]
2029 if !meta.xattrs.is_empty() {
2030 self.inode_xattrs
2031 .insert(inode_number, to_core_xattrs(&meta.xattrs));
2032 }
2033 let (mode, uid, gid) = meta.identity();
2034 let pf = PendingFile {
2035 inode_number,
2036 path: path.to_path_buf(),
2037 mtime_ns: meta.mtime_ns,
2038 file_len,
2039 mode,
2040 uid,
2041 gid,
2042 };
2043 if let Some(sink) = &self.pending_sink {
2044 sink.send(pf).map_err(|_| {
2050 WriteError::Io(std::io::Error::other("walk: compress pipeline shut down"))
2051 })?;
2052 } else {
2053 self.pending_files.push(pf);
2054 }
2055 }
2056 Ok(inode_number)
2057 } else {
2058 #[cfg(unix)]
2059 let kind = {
2060 use std::os::unix::fs::FileTypeExt;
2061 if meta.is_fifo {
2062 "fifo".to_owned()
2063 } else if meta.is_socket {
2064 "socket".to_owned()
2065 } else if meta.is_block_device {
2066 "block device".to_owned()
2067 } else if meta.is_char_device {
2068 "character device".to_owned()
2069 } else {
2070 "unknown".to_owned()
2071 }
2072 };
2073 #[cfg(not(unix))]
2074 let kind = "unknown".to_owned();
2075 Err(WriteError::UnsupportedFileType {
2076 path: path.to_path_buf(),
2077 kind,
2078 })
2079 }
2080 }
2081
2082 fn train_and_apply_dictionary(&mut self, dictionaries: &crate::config::DictionaryConfig) {
2097 if !dictionaries.enabled {
2098 Self::release_dictionary_samples(self);
2099 return;
2100 }
2101
2102 if let Some(adopted) = self.base_dictionaries.take() {
2106 for dict in adopted {
2107 match dict.id {
2108 0 => {
2109 self.trained_dicts_by_class
2110 .insert(crate::classifier::Class::Text, dict);
2111 }
2112 1 => {
2113 self.trained_dicts_by_class
2114 .insert(crate::classifier::Class::Binary, dict);
2115 }
2116 _ => {}
2117 }
2118 }
2119 self.apply_trained_dictionaries();
2120 return;
2121 }
2122
2123 let target = usize::try_from(dictionaries.max_dict_size).unwrap_or(65_536);
2124 let min_class = usize::try_from(dictionaries.min_class_size).unwrap_or(0);
2125
2126 let text_classes = [
2130 crate::classifier::Class::Text,
2131 crate::classifier::Class::Code,
2132 crate::classifier::Class::Sparse,
2133 ];
2134 let binary_classes = [crate::classifier::Class::Binary];
2135
2136 let text_samples: Vec<&[u8]> = text_classes
2138 .iter()
2139 .flat_map(|c| self.dict_samples_by_class.get(c).into_iter().flatten())
2140 .map(Vec::as_slice)
2141 .collect();
2142 let trainer = crate::dictionary::TrainerKind::from_config_str(&dictionaries.trainer);
2143 if text_samples.len() >= min_class {
2144 if let Some(dict) =
2145 crate::dictionary::train_zstd_with_trainer(0, &text_samples, target, trainer)
2146 {
2147 self.trained_dicts_by_class
2148 .insert(crate::classifier::Class::Text, dict);
2149 }
2150 }
2151 let binary_samples: Vec<&[u8]> = binary_classes
2152 .iter()
2153 .flat_map(|c| self.dict_samples_by_class.get(c).into_iter().flatten())
2154 .map(Vec::as_slice)
2155 .collect();
2156 if binary_samples.len() >= min_class {
2157 if let Some(dict) =
2158 crate::dictionary::train_zstd_with_trainer(1, &binary_samples, target, trainer)
2159 {
2160 self.trained_dicts_by_class
2161 .insert(crate::classifier::Class::Binary, dict);
2162 }
2163 }
2164
2165 self.apply_trained_dictionaries();
2166 }
2167
2168 fn apply_trained_dictionaries(&mut self) {
2174 let text_classes = [
2178 crate::classifier::Class::Text,
2179 crate::classifier::Class::Code,
2180 crate::classifier::Class::Sparse,
2181 ];
2182 let binary_classes = [crate::classifier::Class::Binary];
2183
2184 use rayon::prelude::*;
2197 let classifier = self.classifier;
2198 let dicts = &self.trained_dicts_by_class;
2199 let candidates: Vec<Option<(std::sync::Arc<[u8]>, u8)>> = self
2200 .drops
2201 .par_iter()
2202 .map(|d| {
2203 if d.codec != limnifs_core::codec::CODEC_ZSTD {
2204 return None;
2205 }
2206 let Some(plaintext) = d.plaintext.as_ref() else {
2207 return None;
2208 };
2209 let class = classifier.classify(plaintext);
2210 let dict_class = if text_classes.contains(&class) {
2211 crate::classifier::Class::Text
2212 } else if binary_classes.contains(&class) {
2213 crate::classifier::Class::Binary
2214 } else {
2215 return None;
2216 };
2217 let Some(dict) = dicts.get(&dict_class) else {
2218 return None;
2219 };
2220 let Ok(dict_compressed) = dict.compress(plaintext) else {
2221 return None;
2222 };
2223 if dict_compressed.len() < d.compressed.len() {
2224 Some((dict_compressed.into(), dict.id))
2225 } else {
2226 None
2227 }
2228 })
2229 .collect();
2230
2231 let saving: isize = candidates
2232 .iter()
2233 .zip(self.drops.iter())
2234 .map(|(c, d)| {
2235 c.as_ref().map_or(0, |(bytes, _)| {
2236 d.compressed.len() as isize - bytes.len() as isize
2237 })
2238 })
2239 .sum();
2240 let dict_bytes: usize = dicts.values().map(|d| d.content.len()).sum();
2241 if saving > dict_bytes as isize {
2242 for (d, candidate) in self.drops.iter_mut().zip(candidates) {
2243 if let Some((bytes, dict_id)) = candidate {
2244 d.compressed = bytes;
2245 d.dict_id = dict_id;
2246 }
2247 }
2248 } else {
2249 self.trained_dicts_by_class.clear();
2255 }
2256
2257 Self::release_dictionary_samples(self);
2258 }
2259
2260 fn release_dictionary_samples(ctx: &mut Self) {
2263 for d in &mut ctx.drops {
2264 d.plaintext = None;
2265 }
2266 ctx.dict_samples_by_class.clear();
2267 }
2268
2269 fn trace_phase(label: &str, start: std::time::Instant) {
2271 if std::env::var_os("LIMNIFS_TRACE_ASSEMBLE").is_some() {
2272 eprintln!("[assemble] {label}: {:?}", start.elapsed());
2273 }
2274 }
2275
2276 fn assemble(mut self) -> WriteArtifact {
2277 let t_assemble = std::time::Instant::now();
2278 let inode_count = self.inodes.len();
2279 let dir_count = self.dir_count;
2280 let drop_count = self.drops.len();
2281
2282 let t = std::time::Instant::now();
2288 let slabs = pack_slabs(&self.drops);
2289 Self::trace_phase("pack_slabs", t);
2290
2291 let t = std::time::Instant::now();
2295 if self.emit_shared_inline {
2296 self.build_shared_inline_table();
2297 }
2298 Self::trace_phase("shared_inline_table", t);
2299
2300 let mut metadata_blob = Vec::new();
2301 metadata_blob.extend_from_slice(&u32::try_from(self.inodes.len()).unwrap().to_le_bytes());
2302 for inode in &self.inodes {
2303 self.encode_inode(&mut metadata_blob, inode);
2304 }
2305 metadata_blob
2306 .extend_from_slice(&u32::try_from(self.dir_nodes.len()).unwrap().to_le_bytes());
2307 for node in &self.dir_nodes {
2308 metadata_blob.extend_from_slice(&node.bytes);
2309 }
2310 if !self.shared_inline_table.is_empty() {
2313 metadata_blob.extend_from_slice(
2314 &u32::try_from(self.shared_inline_table.len())
2315 .unwrap()
2316 .to_le_bytes(),
2317 );
2318 for entry in &self.shared_inline_table {
2319 let len = u32::try_from(entry.len()).expect("shared entry fits u32");
2320 metadata_blob.extend_from_slice(&len.to_le_bytes());
2321 metadata_blob.extend_from_slice(entry);
2322 }
2323 }
2324
2325 Self::trace_phase("metadata_encode", t);
2326 let uncompressed_len =
2333 u32::try_from(metadata_blob.len()).expect("metadata blob length fits u32");
2334 let t = std::time::Instant::now();
2335 let metadata_hash = hash_section(&metadata_blob);
2336 let metadata_codec = self.metadata_codec;
2337 let metadata_quality = if metadata_blob.len() > METADATA_LARGE_BLOB_THRESHOLD {
2338 METADATA_LARGE_BLOB_QUALITY
2339 } else {
2340 METADATA_SMALL_BLOB_QUALITY
2341 };
2342 let compressed_blob = if metadata_codec == limnifs_core::codec::CODEC_BROTLI {
2343 limnifs_core::codec::compress_brotli_with_quality(&metadata_blob, metadata_quality)
2344 .unwrap_or_else(|_| metadata_blob.clone())
2345 } else {
2346 limnifs_core::codec::compress(metadata_codec, &metadata_blob)
2347 .unwrap_or_else(|_| metadata_blob.clone())
2348 };
2349 Self::trace_phase("metadata_compress", t);
2350 let (on_wire_codec, on_wire_blob) = if compressed_blob.len() < metadata_blob.len() {
2351 (metadata_codec, compressed_blob)
2352 } else {
2353 (limnifs_core::codec::CODEC_STORE, metadata_blob.clone())
2354 };
2355
2356 let externalize_at = self
2360 .metadata_externalize_threshold
2361 .min(limnifs_core::metadata_reference::DEFAULT_INLINE_METADATA_MAX_BYTES as usize);
2362 let (metadata_sidecar, inline_data, metadata_locator_count) =
2363 if on_wire_blob.len() > externalize_at {
2364 let h = hash_section(&on_wire_blob);
2371 let mut h8 = String::with_capacity(8);
2372 for b in &h[..4] {
2373 h8.push_str(&format!("{b:02x}"));
2374 }
2375 let locator = format!("file:metadata-{h8}.bin");
2376 let sidecar = MetadataSidecar {
2377 bytes: on_wire_blob.clone(),
2378 locator,
2379 };
2380 (Some(sidecar), None, 1u32)
2381 } else {
2382 (None, Some(on_wire_blob.clone()), 0u32)
2383 };
2384
2385 let mut manifest = Vec::new();
2386
2387 let header_start = manifest.len();
2388 manifest.extend_from_slice(&ManifestHeader::current().to_bytes());
2389 let header_end = manifest.len();
2390
2391 let flags_start = manifest.len();
2392 manifest.push(FEATURE_FLAGS_SECTION_VERSION);
2393 manifest.extend_from_slice(&0u32.to_le_bytes());
2394 let flags_end = manifest.len();
2395
2396 let meta_ref_start = manifest.len();
2399 manifest.push(METADATA_REFERENCE_SECTION_VERSION_2);
2400 manifest.extend_from_slice(&metadata_hash);
2401 manifest.extend_from_slice(&uncompressed_len.to_le_bytes());
2402 manifest.push(on_wire_codec);
2403 manifest.extend_from_slice(&metadata_locator_count.to_le_bytes());
2404 if let Some(sidecar) = &metadata_sidecar {
2405 let loc_bytes = sidecar.locator.as_bytes();
2406 let loc_len = u32::try_from(loc_bytes.len()).expect("locator fits u32");
2407 manifest.extend_from_slice(&loc_len.to_le_bytes());
2408 manifest.extend_from_slice(loc_bytes);
2409 }
2410 match &inline_data {
2411 Some(blob) => {
2412 let inline_len = u32::try_from(blob.len()).expect("metadata fits u32");
2413 manifest.extend_from_slice(&inline_len.to_le_bytes());
2414 manifest.extend_from_slice(blob);
2415 }
2416 None => {
2417 manifest.extend_from_slice(&0u32.to_le_bytes());
2418 }
2419 }
2420 let meta_ref_end = manifest.len();
2421
2422 let slab_index_start = manifest.len();
2423 manifest.push(SLAB_INDEX_SECTION_VERSION);
2424 manifest.extend_from_slice(&u32::try_from(slabs.len()).unwrap().to_le_bytes());
2425 for slab in &slabs {
2426 manifest.extend_from_slice(&slab.id.to_bytes());
2427 manifest.extend_from_slice(&1u32.to_le_bytes());
2428 let loc_bytes = slab.locator.as_bytes();
2429 let loc_len = u32::try_from(loc_bytes.len()).expect("locator fits u32");
2430 manifest.extend_from_slice(&loc_len.to_le_bytes());
2431 manifest.extend_from_slice(loc_bytes);
2432 }
2433 let slab_index_end = manifest.len();
2434
2435 let history_start = manifest.len();
2436 manifest.push(HISTORY_SECTION_VERSION);
2437 manifest.extend_from_slice(&1u32.to_le_bytes());
2438 manifest.push(0x01);
2439 manifest.extend_from_slice(&0u64.to_le_bytes());
2440 manifest.extend_from_slice(&0u32.to_le_bytes());
2441 manifest.extend_from_slice(&0u32.to_le_bytes());
2442 let history_end = manifest.len();
2443
2444 let profile_desc_start = manifest.len();
2449 if let Some(ref name) = self.profile_name {
2450 let desc = limnifs_core::profile_descriptor::ProfileDescriptor {
2451 version: limnifs_core::profile_descriptor::PROFILE_DESCRIPTOR_SECTION_VERSION,
2452 profile_name: Some(name.clone()),
2453 blake3_hashing: true,
2454 cross_file_dedup: true,
2455 content_classification: !self.categorizers_disabled,
2456 integrity_verify: true,
2457 read_write: self.rw_mode,
2458 auto_turnover: self.auto_turnover,
2459 };
2460 limnifs_core::profile_descriptor::encode_profile_descriptor(&desc, &mut manifest);
2461 }
2462 let profile_desc_end = manifest.len();
2463
2464 if !self.trained_dicts_by_class.is_empty() {
2470 let dicts: Vec<_> = self
2471 .trained_dicts_by_class
2472 .values()
2473 .map(|d| limnifs_core::dictionary_section::Dictionary {
2474 codec_id: d.codec,
2475 class_id: d.id,
2476 data: d.content.clone(),
2477 })
2478 .collect();
2479 let section = limnifs_core::dictionary_section::DictionarySection {
2480 version: limnifs_core::dictionary_section::DICTIONARY_SECTION_VERSION,
2481 dicts,
2482 };
2483 limnifs_core::dictionary_section::encode_dictionary_section(§ion, &mut manifest);
2484 }
2485
2486 let dictionary_end = manifest.len();
2487
2488 let delta_linkage_hash = if let Some(base_root) = self.base_root {
2494 let delta_start = manifest.len();
2495 manifest.push(limnifs_core::delta_linkage::DELTA_LINKAGE_SECTION_VERSION);
2501 manifest.extend_from_slice(&base_root);
2502 manifest.extend_from_slice(&0u32.to_le_bytes());
2503 hash_section(&manifest[delta_start..])
2504 } else {
2505 hash_empty_section()
2506 };
2507 let _ = dictionary_end;
2508
2509 let hashes = SectionHashes {
2510 metadata: metadata_hash,
2511 format_header: hash_section(&manifest[header_start..header_end]),
2512 feature_flags: hash_section(&manifest[flags_start..flags_end]),
2513 metadata_reference: hash_section(&manifest[meta_ref_start..meta_ref_end]),
2514 slab_index: hash_section(&manifest[slab_index_start..slab_index_end]),
2515 crypto_params: hash_empty_section(),
2516 ec_params: hash_empty_section(),
2517 dms_policy: hash_empty_section(),
2518 delta_linkage: delta_linkage_hash,
2519 history: hash_section(&manifest[history_start..history_end]),
2520 };
2532 let merkle_root = compute_merkle_root(&hashes);
2533
2534 WriteArtifact {
2535 bytes: manifest,
2536 merkle_root,
2537 slabs,
2538 metadata_sidecar,
2539 inode_count,
2540 file_count: self.file_count,
2541 dir_count,
2542 drop_count,
2543 root_inode_number: self.root_inode_number,
2544 }
2545 }
2546
2547 fn encode_inode(&self, out: &mut Vec<u8>, inode: &PendingInode) {
2548 out.extend_from_slice(&inode.number.to_le_bytes());
2549 out.extend_from_slice(&inode.mode.to_le_bytes());
2550 out.extend_from_slice(&inode.uid.to_le_bytes());
2551 out.extend_from_slice(&inode.gid.to_le_bytes());
2552 out.extend_from_slice(&inode.mtime_ns.to_le_bytes());
2553 out.extend_from_slice(&inode.mtime_ns.to_le_bytes());
2554 let nlink = self.nlink_counts.get(&inode.number).copied().unwrap_or(1);
2555 out.extend_from_slice(&nlink.to_le_bytes());
2556 let xattrs: &[limnifs_core::inode::XAttr] = self
2557 .inode_xattrs
2558 .get(&inode.number)
2559 .map_or(inode.xattrs.as_slice(), std::convert::AsRef::as_ref);
2560 let mut flags_and_xattrs = move |out: &mut Vec<u8>, base: u8| {
2564 if xattrs.is_empty() {
2565 out.push(base);
2566 return;
2567 }
2568 out.push(base | limnifs_core::inode::INODE_FLAG_HAS_XATTRS);
2569 let count = u32::try_from(xattrs.len()).expect("xattr count fits u32");
2570 out.extend_from_slice(&count.to_le_bytes());
2571 for x in xattrs {
2572 out.push(x.namespace);
2573 let key = x.key.as_bytes();
2574 let key_len = u32::try_from(key.len()).expect("xattr key fits u32");
2575 out.extend_from_slice(&key_len.to_le_bytes());
2576 out.extend_from_slice(key);
2577 let value_len = u32::try_from(x.value.len()).expect("xattr value fits u32");
2578 out.extend_from_slice(&value_len.to_le_bytes());
2579 out.extend_from_slice(&x.value);
2580 }
2581 };
2582 match &inode.content {
2583 PendingContent::Inline(data) => {
2584 let h = hash_section(data);
2585 if let Some(&idx) = self.shared_inline_map.get(&h) {
2586 flags_and_xattrs(out, INODE_FLAG_INLINE_DATA | INODE_FLAG_SHARED_INLINE);
2588 out.extend_from_slice(&(idx as u32).to_le_bytes());
2589 } else {
2590 flags_and_xattrs(out, INODE_FLAG_INLINE_DATA);
2591 let len = u32::try_from(data.len()).expect("data fits u32");
2592 out.extend_from_slice(&len.to_le_bytes());
2593 out.extend_from_slice(data);
2594 }
2595 }
2596 PendingContent::DropBacked { file_len, slices } => {
2597 flags_and_xattrs(out, 0x00);
2598 let slice_count = u32::try_from(slices.len()).expect("slice count fits u32");
2599 out.extend_from_slice(&slice_count.to_le_bytes());
2600 for slice in slices {
2601 out.extend_from_slice(&slice.file_byte_start.to_le_bytes());
2602 out.extend_from_slice(&slice.file_byte_end.to_le_bytes());
2603 out.extend_from_slice(&slice.drop_id);
2604 out.extend_from_slice(&0u32.to_le_bytes());
2606 let drop_byte_len = u32::try_from(slice.file_byte_end - slice.file_byte_start)
2610 .expect("slice range fits u32");
2611 out.extend_from_slice(&drop_byte_len.to_le_bytes());
2612 }
2613 let _ = file_len;
2614 }
2615 PendingContent::Symlink(target) => {
2616 flags_and_xattrs(out, 0x00);
2620 let t = target.as_bytes();
2621 let len = u32::try_from(t.len()).expect("target fits u32");
2622 out.extend_from_slice(&len.to_le_bytes());
2623 out.extend_from_slice(t);
2624 }
2625 PendingContent::Directory(entries) => {
2626 flags_and_xattrs(out, 0x00);
2627 let node = self
2628 .dir_nodes
2629 .iter()
2630 .find(|n| n.entries == *entries)
2631 .expect("directory node must exist");
2632 out.extend_from_slice(&node.hash);
2633 }
2634 }
2635 }
2636}
2637
2638fn sidecar_name(locator: &str) -> Result<&str, WriteError> {
2644 limnifs_core::locator::local_sidecar_name(locator)
2645 .map_err(|e| WriteError::Io(std::io::Error::other(format!("{e}"))))
2646}
2647
2648fn encode_dir_node(entries: &[(String, u64, u8)]) -> DirNode {
2649 let mut bytes = Vec::new();
2650 bytes.push(1u8);
2651 let count = u32::try_from(entries.len()).expect("entry count fits u32");
2652 bytes.extend_from_slice(&count.to_le_bytes());
2653 for (name, inode_number, entry_type) in entries {
2654 let name_bytes = name.as_bytes();
2655 let name_len = u32::try_from(name_bytes.len()).expect("name fits u32");
2656 bytes.extend_from_slice(&name_len.to_le_bytes());
2657 bytes.extend_from_slice(name_bytes);
2658 bytes.extend_from_slice(&inode_number.to_le_bytes());
2659 bytes.push(*entry_type);
2660 }
2661 let hash = hash_section(&bytes);
2662 DirNode {
2663 entries: entries.to_vec(),
2664 bytes,
2665 hash,
2666 }
2667}
2668
2669fn pack_slabs(drops: &[PendingDrop]) -> Vec<SlabArtifact> {
2678 let local_drops: Vec<&PendingDrop> = drops
2683 .iter()
2684 .filter(|d| d.codec != limnifs_core::codec::CODEC_REFERENCED)
2685 .collect();
2686 if local_drops.is_empty() {
2687 return Vec::new();
2688 }
2689
2690 let max_content = MAX_SLAB_TOTAL_BYTES.saturating_sub(SLAB_HEADER_LEN);
2691
2692 let mut slab_groups: Vec<Vec<&PendingDrop>> = Vec::new();
2697 let mut current: Vec<&PendingDrop> = Vec::new();
2698 let mut current_size: usize = 0;
2699
2700 for drop in &local_drops {
2701 let footprint = drop.slab_footprint();
2702 if !current.is_empty() && current_size + footprint > max_content {
2703 slab_groups.push(std::mem::take(&mut current));
2704 current_size = 0;
2705 }
2706 current.push(*drop);
2707 current_size += footprint;
2708 }
2709 if !current.is_empty() {
2710 slab_groups.push(current);
2711 }
2712
2713 use rayon::prelude::*;
2719 slab_groups
2720 .par_iter()
2721 .enumerate()
2722 .map(|(ordinal, group)| {
2723 let ordinal_u64 = u64::try_from(ordinal).expect("slab count fits u64");
2724 encode_slab(ordinal_u64, group)
2725 })
2726 .collect()
2727}
2728
2729fn encode_slab(ordinal: u64, drops: &[&PendingDrop]) -> SlabArtifact {
2733 const DROP_RECORD_LEN: usize = 50;
2740 let mut drop_records = Vec::with_capacity(drops.len() * DROP_RECORD_LEN);
2741 let mut solid_window = Vec::new();
2742 let mut drop_ids = Vec::with_capacity(drops.len());
2743 let mut offset_in_window: u32 = 0;
2744
2745 for drop in drops {
2746 let plaintext_len = drop.plaintext_len_value();
2747 let window_len = drop.len_in_window();
2748 drop_records.extend_from_slice(&drop.id);
2749 drop_records.extend_from_slice(&plaintext_len.to_le_bytes());
2750 drop_records.extend_from_slice(&[drop.codec, 0x00, 0x00]);
2752 drop_records.push(0x00); drop_records.extend_from_slice(&offset_in_window.to_le_bytes());
2754 drop_records.extend_from_slice(&window_len.to_le_bytes());
2755 drop_records.push(drop.dict_id); drop_records.push(drop.flags); solid_window.extend_from_slice(&drop.compressed);
2758 drop_ids.push(drop.id);
2759 offset_in_window = offset_in_window
2760 .checked_add(window_len)
2761 .expect("slab window size fits u32");
2762 }
2763
2764 let slab_content = [&drop_records[..], &solid_window[..]].concat();
2765 let slab_hash = hash_section(&slab_content);
2766 let slab_id = SlabId::new(ordinal, slab_hash);
2767
2768 let total_length = SLAB_HEADER_LEN + slab_content.len();
2769 let mut slab_bytes = Vec::with_capacity(total_length);
2770 slab_bytes.extend_from_slice(b"LIM1");
2771 slab_bytes.extend_from_slice(&1u16.to_le_bytes()); slab_bytes.extend_from_slice(&slab_id.to_bytes());
2773 slab_bytes.extend_from_slice(
2774 &u64::try_from(total_length)
2775 .unwrap_or(u64::MAX)
2776 .to_le_bytes(),
2777 );
2778 slab_bytes.push(0x00);
2779 slab_bytes.push(0x00);
2780 slab_bytes.extend_from_slice(&slab_content);
2781
2782 let mut h8 = String::with_capacity(8);
2786 for b in &slab_id.hash[..4] {
2787 h8.push_str(&format!("{b:02x}"));
2788 }
2789 let locator = format!("file:slab-{ordinal}-{h8}.bin");
2790
2791 SlabArtifact {
2792 id: slab_id,
2793 bytes: slab_bytes,
2794 locator,
2795 drop_ids,
2796 }
2797}
2798
2799#[cfg(test)]
2800fn pseudo_random_bytes(seed: u64, count: usize) -> Vec<u8> {
2801 let mut state = seed;
2802 let mut out = Vec::with_capacity(count);
2803 for _ in 0..count {
2804 state = state
2805 .wrapping_mul(6_364_136_223_846_793_005)
2806 .wrapping_add(1_442_695_040_888_963_407);
2807 out.push(u8::try_from(state >> 56).expect("fits u8"));
2808 }
2809 out
2810}
2811
2812#[cfg(test)]
2813mod tests {
2814 use super::*;
2815 use limnifs_core::ManifestCursor;
2816
2817 #[test]
2818 fn write_stream_packs_single_named_stream() {
2819 let temp = std::env::temp_dir().join(format!(
2823 "limnifs-write-stream-test-{}-{}",
2824 std::process::id(),
2825 std::time::SystemTime::now()
2826 .duration_since(std::time::UNIX_EPOCH)
2827 .unwrap()
2828 .as_nanos()
2829 ));
2830 std::fs::create_dir_all(&temp).expect("create temp dir");
2831
2832 let content = b"stream test content line\n".repeat(10_000); let cursor = std::io::Cursor::new(content.clone());
2834 let config = WriteConfig::default_v0_1();
2835 let artifact = write_stream("streamed.txt", cursor, &config).expect("write_stream");
2836
2837 assert!(!artifact.bytes.is_empty(), "manifest bytes non-empty");
2838 assert!(!artifact.slabs.is_empty(), "at least one slab produced");
2839 let total_drop_bytes: usize = artifact.slabs.iter().map(|s| s.bytes.len()).sum();
2844 assert!(total_drop_bytes > 0, "drops non-empty");
2845
2846 let _ = std::fs::remove_dir_all(&temp);
2847 }
2848
2849 #[test]
2850 fn write_layer_references_base_drops() {
2851 let temp = std::env::temp_dir().join(format!(
2857 "limnifs-write-layer-test-{}-{}",
2858 std::process::id(),
2859 std::time::SystemTime::now()
2860 .duration_since(std::time::UNIX_EPOCH)
2861 .unwrap()
2862 .as_nanos()
2863 ));
2864 std::fs::create_dir_all(&temp).expect("create temp dir");
2865
2866 let base_dir = temp.join("base");
2868 std::fs::create_dir_all(&base_dir).expect("base dir");
2869 let text = b"layer test content line\n".repeat(50_000); std::fs::write(base_dir.join("shared.txt"), &text).expect("write shared");
2871 std::fs::write(base_dir.join("base-only.txt"), b"only in base").expect("write base-only");
2872
2873 let config = WriteConfig::default_v0_1();
2874 let base_artifact = write_directory_with_config(&base_dir, &config).expect("base");
2875
2876 let base_manifest = temp.join("base.lim");
2877 std::fs::write(&base_manifest, &base_artifact.bytes).expect("write base manifest");
2878 for slab in &base_artifact.slabs {
2879 let slab_name = sidecar_name(&slab.locator).expect("slab locator");
2880 std::fs::write(temp.join(slab_name), &slab.bytes).expect("write base slab");
2881 }
2882
2883 let layer_dir = temp.join("layer");
2885 std::fs::create_dir_all(&layer_dir).expect("layer dir");
2886 std::fs::write(layer_dir.join("shared.txt"), &text).expect("write shared in layer");
2887 std::fs::write(layer_dir.join("new.txt"), b"fresh content in layer").expect("write new");
2888
2889 let layer_artifact = write_layer(&base_manifest, &layer_dir, &config).expect("layer");
2890
2891 let layer_slab_bytes: usize = layer_artifact.slabs.iter().map(|s| s.bytes.len()).sum();
2894 let base_slab_bytes: usize = base_artifact.slabs.iter().map(|s| s.bytes.len()).sum();
2895 assert!(
2896 layer_slab_bytes < base_slab_bytes / 4,
2897 "layer slabs ({}) should be much smaller than base ({}) — layering failed",
2898 layer_slab_bytes,
2899 base_slab_bytes
2900 );
2901
2902 let base_root = base_artifact.merkle_root.as_bytes();
2905 assert!(
2906 layer_artifact
2907 .bytes
2908 .windows(32)
2909 .any(|w| w == base_root.as_slice()),
2910 "layer manifest must contain base's ManifestRoot bytes"
2911 );
2912
2913 let _ = std::fs::remove_dir_all(&temp);
2914 }
2915
2916 #[test]
2917 fn tournament_short_circuits_on_highly_compressible_chunk() {
2918 let chunk = b"hello world ".repeat(500);
2921 let tunables = limnifs_core::codec::CodecTunables::default();
2922 let tournament = TournamentSpec {
2923 codec_ids: vec![
2924 limnifs_core::codec::CODEC_LZ4,
2925 limnifs_core::codec::CODEC_BROTLI,
2926 ],
2927 min_size: 16,
2928 skip_for_binary: false,
2929 short_circuit_permille: 250,
2930 };
2931 let (codec_id, compressed) = compress_chunk_with_tournament(
2932 &chunk,
2933 classifier::Class::Text,
2934 limnifs_core::codec::CODEC_BROTLI,
2935 limnifs_core::codec::CODEC_LZ4,
2936 &tunables,
2937 &tournament,
2938 );
2939 assert_eq!(codec_id, limnifs_core::codec::CODEC_LZ4);
2940 assert!(compressed.len() < chunk.len());
2941 }
2942
2943 #[test]
2944 fn tournament_runs_all_codecs_when_short_circuit_disabled() {
2945 let chunk = b"hello world ".repeat(500);
2951 let tunables = limnifs_core::codec::CodecTunables::default();
2952 let tournament = TournamentSpec {
2953 codec_ids: vec![
2954 limnifs_core::codec::CODEC_LZ4,
2955 limnifs_core::codec::CODEC_BROTLI,
2956 limnifs_core::codec::CODEC_ZSTD,
2957 ],
2958 min_size: 16,
2959 skip_for_binary: false,
2960 short_circuit_permille: 0,
2961 };
2962 let (codec_id, compressed) = compress_chunk_with_tournament(
2963 &chunk,
2964 classifier::Class::Text,
2965 limnifs_core::codec::CODEC_BROTLI,
2966 limnifs_core::codec::CODEC_LZ4,
2967 &tunables,
2968 &tournament,
2969 );
2970 assert!(
2974 codec_id == limnifs_core::codec::CODEC_ZSTD
2975 || codec_id == limnifs_core::codec::CODEC_BROTLI,
2976 "expected ZSTD or Brotli to win, got codec {codec_id}"
2977 );
2978 assert!(compressed.len() < chunk.len());
2979 }
2980
2981 #[test]
2982 fn tournament_skips_for_binary_when_configured() {
2983 let chunk = vec![0u8; 4096];
2984 let tunables = limnifs_core::codec::CodecTunables::default();
2985 let tournament = TournamentSpec {
2986 codec_ids: vec![limnifs_core::codec::CODEC_BROTLI],
2987 min_size: 16,
2988 skip_for_binary: true,
2989 short_circuit_permille: 250,
2990 };
2991 let (codec_id, _compressed) = compress_chunk_with_tournament(
2992 &chunk,
2993 classifier::Class::Binary,
2994 limnifs_core::codec::CODEC_BROTLI,
2995 limnifs_core::codec::CODEC_LZ4,
2996 &tunables,
2997 &tournament,
2998 );
2999 assert_eq!(codec_id, limnifs_core::codec::CODEC_LZ4);
3001 }
3002
3003 #[test]
3004 fn tournament_small_chunk_uses_preferred_codec() {
3005 let chunk = b"tiny";
3006 let tunables = limnifs_core::codec::CodecTunables::default();
3007 let tournament = TournamentSpec {
3008 codec_ids: vec![limnifs_core::codec::CODEC_BROTLI],
3009 min_size: 1024,
3010 skip_for_binary: false,
3011 short_circuit_permille: 0,
3012 };
3013 let (codec_id, _compressed) = compress_chunk_with_tournament(
3014 chunk,
3015 classifier::Class::Text,
3016 limnifs_core::codec::CODEC_BROTLI,
3017 limnifs_core::codec::CODEC_LZ4,
3018 &tunables,
3019 &tournament,
3020 );
3021 assert_eq!(codec_id, limnifs_core::codec::CODEC_STORE);
3023 }
3024
3025 #[test]
3026 fn tournament_falls_back_to_store_when_no_codec_compresses() {
3027 let chunk = pseudo_random_bytes(42, 4096);
3031 let tunables = limnifs_core::codec::CodecTunables::default();
3032 let tournament = TournamentSpec {
3033 codec_ids: vec![limnifs_core::codec::CODEC_LZ4],
3034 min_size: 16,
3035 skip_for_binary: false,
3036 short_circuit_permille: 0,
3037 };
3038 let (codec_id, compressed) = compress_chunk_with_tournament(
3039 &chunk,
3040 classifier::Class::Binary,
3041 limnifs_core::codec::CODEC_BROTLI,
3042 limnifs_core::codec::CODEC_LZ4,
3043 &tunables,
3044 &tournament,
3045 );
3046 assert_eq!(codec_id, limnifs_core::codec::CODEC_STORE);
3047 assert_eq!(compressed.len(), chunk.len());
3048 }
3049
3050 #[test]
3051 fn dictionaries_enabled_emits_dictionary_section_when_enough_samples() {
3052 let temp = std::env::temp_dir().join(format!(
3057 "limnifs-write-test-{}-dict-{}",
3058 std::process::id(),
3059 std::time::SystemTime::now()
3060 .duration_since(std::time::UNIX_EPOCH)
3061 .map(|d| d.as_nanos() as u64)
3062 .unwrap_or(0),
3063 ));
3064 let _ = std::fs::remove_dir_all(&temp);
3065 std::fs::create_dir_all(&temp).expect("mkdir");
3066
3067 for i in 0..200 {
3070 let content = format!(
3072 "function test_case_{i}() {{ return constant + {i}; }}\n\
3073 // shared comment line {i}\n\
3074 struct Foo {{ x: i32 }} // type {i}\n"
3075 )
3076 .repeat(5);
3077 let path = temp.join(format!("file_{i:04}.txt"));
3078 std::fs::write(&path, content.as_bytes()).expect("write");
3079 }
3080
3081 let mut config = crate::profile::balanced();
3082 config.defaults.text_codec = "zstd".into();
3084 config.defaults.metadata_codec = "zstd".into();
3088 config.dictionaries.enabled = true;
3089 config.dictionaries.min_class_size = 50;
3090 config.dictionaries.max_dict_size = 8192;
3091
3092 let artifact = write_directory_with_config(&temp, &config).expect("write");
3093 std::fs::remove_dir_all(&temp).ok();
3094
3095 let mut cursor = ManifestCursor::new(&artifact.bytes);
3100 let _ = limnifs_core::parse_manifest_header(&mut cursor).expect("header");
3101 let _ = limnifs_core::parse_feature_flags_section(&mut cursor).expect("flags");
3102 let _ = limnifs_core::parse_metadata_reference(&mut cursor).expect("meta_ref");
3103 let _ = limnifs_core::parse_slab_index(&mut cursor).expect("slab_index");
3104 let _ = limnifs_core::parse_history(&mut cursor).expect("history");
3105 let _remaining = cursor.remaining_len();
3108 }
3109
3110 #[test]
3111 fn write_empty_directory() {
3112 let temp =
3113 std::env::temp_dir().join(format!("limnifs-write-test-{}-empty", std::process::id()));
3114 std::fs::create_dir_all(&temp).expect("create temp dir");
3115 let artifact = write_directory(&temp).expect("write succeeds");
3116 std::fs::remove_dir_all(&temp).ok();
3117 assert!(artifact.inode_count >= 1);
3118 assert_eq!(artifact.file_count, 0);
3119 assert_eq!(artifact.dir_count, 1);
3120 assert!(artifact.slabs.is_empty());
3121 }
3122
3123 #[test]
3124 fn write_small_file_inline() {
3125 let temp =
3126 std::env::temp_dir().join(format!("limnifs-write-test-{}-small", std::process::id()));
3127 std::fs::create_dir_all(&temp).expect("create temp dir");
3128 std::fs::write(temp.join("hello.txt"), b"hello world").expect("write file");
3129 let artifact = write_directory(&temp).expect("write succeeds");
3130 std::fs::remove_dir_all(&temp).ok();
3131 assert_eq!(artifact.file_count, 1);
3132 assert!(artifact.slabs.is_empty());
3133 assert_eq!(artifact.drop_count, 0);
3134 }
3135
3136 #[test]
3137 fn write_large_file_uses_slab() {
3138 let temp =
3139 std::env::temp_dir().join(format!("limnifs-write-test-{}-large", std::process::id()));
3140 std::fs::create_dir_all(&temp).expect("create temp dir");
3141 let large_data = vec![0xABu8; INLINE_THRESHOLD + 100];
3142 std::fs::write(temp.join("big.bin"), &large_data).expect("write big");
3143 let artifact = write_directory(&temp).expect("write succeeds");
3144 std::fs::remove_dir_all(&temp).ok();
3145 assert_eq!(artifact.drop_count, 1);
3146 assert_eq!(artifact.slabs.len(), 1);
3147 }
3148
3149 #[test]
3150 fn write_mixed_inline_and_large() {
3151 let temp =
3152 std::env::temp_dir().join(format!("limnifs-write-test-{}-mix", std::process::id()));
3153 std::fs::create_dir_all(&temp).expect("create temp dir");
3154 std::fs::write(temp.join("small.txt"), b"tiny").expect("write small");
3155 std::fs::write(temp.join("large.bin"), vec![0xCDu8; INLINE_THRESHOLD * 2])
3156 .expect("write large");
3157 let artifact = write_directory(&temp).expect("write succeeds");
3158 std::fs::remove_dir_all(&temp).ok();
3159 assert_eq!(artifact.file_count, 2);
3160 assert_eq!(artifact.drop_count, 1);
3161 assert_eq!(artifact.slabs.len(), 1);
3162 }
3163
3164 #[test]
3165 fn deduplicates_identical_large_files() {
3166 let temp =
3167 std::env::temp_dir().join(format!("limnifs-write-test-{}-dedup", std::process::id()));
3168 std::fs::create_dir_all(&temp).expect("create temp dir");
3169 let data = vec![0x77u8; INLINE_THRESHOLD + 10];
3170 std::fs::write(temp.join("a.bin"), &data).expect("write a");
3171 std::fs::write(temp.join("b.bin"), &data).expect("write b");
3172 let artifact = write_directory(&temp).expect("write succeeds");
3173 std::fs::remove_dir_all(&temp).ok();
3174 assert_eq!(artifact.drop_count, 1);
3175 }
3176
3177 #[test]
3178 fn write_and_verify_roundtrip() {
3179 let temp = std::env::temp_dir().join(format!(
3180 "limnifs-write-test-{}-roundtrip",
3181 std::process::id()
3182 ));
3183 std::fs::create_dir_all(&temp).expect("create temp dir");
3184 std::fs::write(temp.join("a.txt"), b"aaa").expect("write a");
3185 std::fs::write(temp.join("b.txt"), b"bbb").expect("write b");
3186 std::fs::create_dir_all(temp.join("sub")).expect("create sub");
3187 std::fs::write(temp.join("sub").join("c.txt"), b"ccc").expect("write c");
3188 let artifact = write_directory(&temp).expect("write succeeds");
3189 std::fs::remove_dir_all(&temp).ok();
3190 assert_eq!(artifact.file_count, 3);
3191 assert_eq!(artifact.dir_count, 2);
3192
3193 let mut cursor = ManifestCursor::new(&artifact.bytes);
3194 limnifs_core::parse_manifest_header(&mut cursor).expect("header");
3195 limnifs_core::parse_feature_flags_section(&mut cursor).expect("flags");
3196 let meta_ref = limnifs_core::parse_metadata_reference(&mut cursor).expect("meta ref");
3197 assert!(meta_ref.is_inlined());
3198 let slab_index = limnifs_core::parse_slab_index(&mut cursor).expect("slab index");
3199 assert_eq!(slab_index.len(), 0);
3200 limnifs_core::parse_history(&mut cursor).expect("history");
3201 }
3202
3203 #[test]
3204 fn write_deterministic() {
3205 let temp =
3206 std::env::temp_dir().join(format!("limnifs-write-test-{}-det", std::process::id()));
3207 std::fs::create_dir_all(&temp).expect("create temp dir");
3208 std::fs::write(temp.join("x.txt"), b"xxx").expect("write x");
3209
3210 let a1 = write_directory(&temp).expect("first write");
3211 let a2 = write_directory(&temp).expect("second write");
3212 std::fs::remove_dir_all(&temp).ok();
3213
3214 assert_eq!(a1.bytes, a2.bytes);
3215 assert_eq!(a1.merkle_root, a2.merkle_root);
3216 }
3217
3218 #[test]
3219 fn slab_parses_correctly() {
3220 let temp =
3221 std::env::temp_dir().join(format!("limnifs-write-test-{}-slab", std::process::id()));
3222 std::fs::create_dir_all(&temp).expect("create temp dir");
3223 std::fs::write(temp.join("big.bin"), vec![0x11u8; INLINE_THRESHOLD + 1])
3224 .expect("write big");
3225 let artifact = write_directory(&temp).expect("write succeeds");
3226 std::fs::remove_dir_all(&temp).ok();
3227
3228 let slab_bytes = &artifact.slabs[0].bytes;
3229 let mut cursor = ManifestCursor::new(slab_bytes);
3230 let slab_header = limnifs_core::parse_slab_header(&mut cursor).expect("slab header parses");
3231 assert_eq!(
3232 slab_header.format_version,
3233 limnifs_core::slab::SLAB_FORMAT_VERSION
3234 );
3235 assert!(!slab_header.is_sealed());
3236 assert!(!slab_header.has_erasure_coding());
3237
3238 let drop_record =
3239 limnifs_core::parse_drop_record(&mut cursor, &slab_header).expect("drop record parses");
3240 assert_eq!(drop_record.plaintext_len as usize, INLINE_THRESHOLD + 1);
3241 }
3242
3243 #[test]
3244 fn fastcdc_produces_multiple_chunks_for_large_files() {
3245 let temp = std::env::temp_dir().join(format!(
3248 "limnifs-write-test-{}-cdc-multi",
3249 std::process::id()
3250 ));
3251 std::fs::create_dir_all(&temp).expect("create temp dir");
3252 let data = pseudo_random_bytes(42, 1024 * 1024);
3253 std::fs::write(temp.join("big.bin"), &data).expect("write big");
3254 let artifact = write_directory(&temp).expect("write succeeds");
3255 std::fs::remove_dir_all(&temp).ok();
3256 assert!(
3257 artifact.drop_count > 1,
3258 "expected FastCDC to produce multiple drops for 1 MiB input, got {}",
3259 artifact.drop_count
3260 );
3261 }
3262
3263 #[test]
3264 fn fastcdc_deduplicates_shared_substrings() {
3265 let temp = std::env::temp_dir().join(format!(
3269 "limnifs-write-test-{}-cdc-dedup",
3270 std::process::id()
3271 ));
3272 std::fs::create_dir_all(&temp).expect("create temp dir");
3273 let shared = pseudo_random_bytes(7, 512 * 1024);
3274 let mut a = Vec::with_capacity(shared.len() + 1024);
3275 a.extend_from_slice(&pseudo_random_bytes(1, 1024));
3276 a.extend_from_slice(&shared);
3277 let mut b = Vec::with_capacity(shared.len() + 2048);
3278 b.extend_from_slice(&pseudo_random_bytes(2, 2048));
3279 b.extend_from_slice(&shared);
3280 std::fs::write(temp.join("a.bin"), &a).expect("write a");
3281 std::fs::write(temp.join("b.bin"), &b).expect("write b");
3282
3283 let temp_a = std::env::temp_dir().join(format!(
3285 "limnifs-write-test-{}-cdc-dedup-a",
3286 std::process::id()
3287 ));
3288 std::fs::create_dir_all(&temp_a).expect("create temp_a");
3289 std::fs::write(temp_a.join("a.bin"), &a).expect("write a");
3290 let artifact_a = write_directory(&temp_a).expect("a writes");
3291 std::fs::remove_dir_all(&temp_a).ok();
3292
3293 let temp_b = std::env::temp_dir().join(format!(
3294 "limnifs-write-test-{}-cdc-dedup-b",
3295 std::process::id()
3296 ));
3297 std::fs::create_dir_all(&temp_b).expect("create temp_b");
3298 std::fs::write(temp_b.join("b.bin"), &b).expect("write b");
3299 let artifact_b = write_directory(&temp_b).expect("b writes");
3300 std::fs::remove_dir_all(&temp_b).ok();
3301
3302 let artifact_both = write_directory(&temp).expect("both write");
3303 std::fs::remove_dir_all(&temp).ok();
3304
3305 let sum_alone = artifact_a.drop_count + artifact_b.drop_count;
3306 assert!(
3307 artifact_both.drop_count < sum_alone,
3308 "expected dedup win: both together = {} drops, sum alone = {} drops",
3309 artifact_both.drop_count,
3310 sum_alone
3311 );
3312 }
3313
3314 #[test]
3315 fn slab_splits_when_content_exceeds_ceiling() {
3316 let temp =
3322 std::env::temp_dir().join(format!("limnifs-write-test-{}-split", std::process::id()));
3323 std::fs::create_dir_all(&temp).expect("create temp dir");
3324 for i in 0..7u32 {
3325 let data = pseudo_random_bytes(u64::from(i), 10 * 1024 * 1024);
3327 std::fs::write(temp.join(format!("big-{i}.bin")), &data).expect("write big");
3328 }
3329 let artifact = write_directory(&temp).expect("write succeeds");
3330 std::fs::remove_dir_all(&temp).ok();
3331
3332 assert!(
3334 artifact.slabs.len() >= 2,
3335 "expected at least 2 slabs for 70 MiB of incompressible data, got {}",
3336 artifact.slabs.len()
3337 );
3338 for slab in &artifact.slabs {
3339 assert!(
3340 slab.bytes.len() <= MAX_SLAB_TOTAL_BYTES,
3341 "slab {} is {} bytes (> {} ceiling)",
3342 slab.id.ordinal,
3343 slab.bytes.len(),
3344 MAX_SLAB_TOTAL_BYTES,
3345 );
3346 }
3347 let total_drop_ids: usize = artifact.slabs.iter().map(|s| s.drop_ids.len()).sum();
3349 assert_eq!(
3350 total_drop_ids, artifact.drop_count,
3351 "drop_ids count across slabs must match WriteArtifact.drop_count",
3352 );
3353 }
3354}