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 rw;
33#[cfg(feature = "sparse-index")]
34pub mod sparse_index;
35pub mod turnover;
36
37pub use config::{
38 profile, CategorizerConfig, ChunkingConfig, CodecRegistry, CodecTunables, Defaults,
39 DictionaryConfig, EncryptionConfig, TournamentConfig, WriteConfig,
40};
41
42use std::collections::{HashMap, HashSet};
43use std::path::{Path, PathBuf};
44
45use crate::chunker::FastCDC;
46use limnifs_core::codec::CODEC_REFERENCED;
47use limnifs_core::slab_store::SlabStore;
48use limnifs_core::{
49 compute_merkle_root, hash_empty_section, hash_section, parse_manifest_header, parse_slab_index,
50 ManifestCursor, ManifestHeader, SectionHashes, FEATURE_FLAGS_SECTION_VERSION,
51 HISTORY_SECTION_VERSION, INODE_FLAG_INLINE_DATA, INODE_FLAG_SHARED_INLINE,
52 METADATA_REFERENCE_SECTION_VERSION_2, SLAB_INDEX_SECTION_VERSION,
53};
54use limnifs_format::{ManifestRoot, SlabId};
55
56pub const INLINE_THRESHOLD: usize = 4096;
59
60pub const MMAP_READ_THRESHOLD: usize = 1024 * 1024;
66
67pub const WHOLE_FILE_MAX_SIZE: usize = 64 * 1024 * 1024;
72
73pub const MAX_SLAB_TOTAL_BYTES: usize = 60 * 1024 * 1024;
78
79const SLAB_HEADER_LEN: usize = 56;
83
84pub const METADATA_EXTERNALIZE_THRESHOLD: usize =
92 limnifs_core::metadata_reference::DEFAULT_INLINE_METADATA_MAX_BYTES as usize - 24 * 1024;
93
94pub const METADATA_LARGE_BLOB_THRESHOLD: usize = 256 * 1024;
99
100pub const METADATA_SMALL_BLOB_QUALITY: i32 = 5;
103
104pub const METADATA_LARGE_BLOB_QUALITY: i32 = 2;
109
110#[derive(Clone, Debug)]
113pub struct SlabArtifact {
114 pub id: SlabId,
115 pub bytes: Vec<u8>,
116 pub locator: String,
117 pub drop_ids: Vec<[u8; 32]>,
121}
122
123#[derive(Clone, Debug)]
127pub struct MetadataSidecar {
128 pub bytes: Vec<u8>,
129 pub locator: String,
130}
131
132#[derive(Clone, Debug)]
134pub struct WriteArtifact {
135 pub bytes: Vec<u8>,
136 pub merkle_root: ManifestRoot,
137 pub slabs: Vec<SlabArtifact>,
140 pub metadata_sidecar: Option<MetadataSidecar>,
144 pub inode_count: usize,
145 pub file_count: usize,
146 pub dir_count: usize,
147 pub drop_count: usize,
148 pub root_inode_number: u64,
153}
154
155impl WriteArtifact {
156 #[must_use]
160 pub fn slab_bytes(&self) -> Option<&[u8]> {
161 if self.slabs.len() == 1 {
162 Some(&self.slabs[0].bytes)
163 } else {
164 None
165 }
166 }
167
168 #[must_use]
170 pub fn slab_locator(&self) -> Option<&str> {
171 if self.slabs.len() == 1 {
172 Some(&self.slabs[0].locator)
173 } else {
174 None
175 }
176 }
177}
178
179#[derive(Debug)]
181pub enum WriteError {
182 Io(std::io::Error),
183 UnsupportedFileType {
188 path: PathBuf,
189 kind: String,
190 },
191}
192
193impl std::fmt::Display for WriteError {
194 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
195 match self {
196 Self::Io(e) => write!(f, "I/O error: {e}"),
197 Self::UnsupportedFileType { path, kind } => write!(
198 f,
199 "unsupported file type ({kind}): {} — limnifs stores files, \
200 directories, and symlinks; remove the entry or file an issue \
201 if you need it carried",
202 path.display()
203 ),
204 }
205 }
206}
207
208impl std::error::Error for WriteError {}
209
210impl From<std::io::Error> for WriteError {
211 fn from(e: std::io::Error) -> Self {
212 Self::Io(e)
213 }
214}
215
216pub fn write_directory(root: &Path) -> Result<WriteArtifact, WriteError> {
230 write_directory_with_config(root, &WriteConfig::default_v0_1())
231}
232
233pub fn write_stream<R: std::io::Read>(
249 name: &str,
250 reader: R,
251 config: &WriteConfig,
252) -> Result<WriteArtifact, WriteError> {
253 let mut ctx = WriteContext::new();
254 ctx.chunker = chunker_from_config(config)?;
255 ctx.categorizers_disabled = config.categorizers.is_empty();
256 ctx.rw_mode = matches!(config.mode, crate::config::ImageMode::ReadWrite(_));
257 ctx.auto_turnover = config.turnover_threshold > 0;
258 ctx.collect_dict_samples = config.dictionaries.enabled;
259
260 let drop_id_root = [0u8; 32]; let pending = PendingFile {
265 path: std::path::PathBuf::from(name),
266 inode_number: 1,
267 file_len: 0, mtime_ns: 0,
269 };
270 ctx.pending_files.push(pending);
271 ctx.root_inode_number = 1;
272
273 let chunker = ctx.chunker.clone();
275 let chunks = chunker.chunk_reader(reader)?;
276
277 let total_len: u64 = chunks.iter().map(|c| c.len() as u64).sum();
279
280 let text_codec = config.text_codec_id().unwrap_or(0x04);
284 let binary_codec = config.binary_codec_id().unwrap_or(0x01);
285 let tunables = config.to_core_tunables();
286 let classifier = ctx.classifier;
287 let registry = config
288 .codec_registry()
289 .map_err(|e| WriteError::Io(std::io::Error::other(format!("codec registry: {e}"))))?;
290 let tournament_codec_ids: Vec<u8> = config
291 .tournament
292 .codecs
293 .iter()
294 .filter_map(|n| registry.lookup_by_name(n))
295 .collect();
296 let tournament = TournamentSpec {
297 codec_ids: tournament_codec_ids,
298 min_size: config.tournament.min_size_threshold as usize,
299 skip_for_binary: config.tournament.skip_for_binary,
300 short_circuit_permille: config.tournament.short_circuit_threshold,
301 };
302
303 let mut drops: Vec<RawDrop> = Vec::with_capacity(chunks.len());
304 let mut slices: Vec<PendingSlice> = Vec::with_capacity(chunks.len());
305 let mut offset: u64 = 0;
306 for chunk in &chunks {
307 let drop_id = hash_section(chunk);
308 slices.push(PendingSlice {
309 drop_id,
310 file_byte_start: offset,
311 file_byte_end: offset + chunk.len() as u64,
312 });
313 offset += chunk.len() as u64;
314 let class = classifier.classify(chunk);
315 let (codec_id, compressed) = compress_chunk_with_tournament(
316 chunk,
317 class,
318 text_codec,
319 binary_codec,
320 &tunables,
321 &tournament,
322 );
323 drops.push((drop_id, chunk.clone(), compressed, codec_id, 0));
324 }
325 let _ = drop_id_root;
326
327 let result = ChunkedFileResult { drops, slices };
329 let pf = ctx.pending_files[0].clone();
330 ctx.merge_chunked_file(&pf, result);
331 ctx.pending_files[0].file_len = total_len;
333 if let Some(inode) = ctx.inodes.last_mut() {
336 if let PendingContent::DropBacked { file_len, .. } = &mut inode.content {
337 *file_len = total_len;
338 }
339 }
340
341 ctx.train_and_apply_dictionary(&config.dictionaries);
342 let artifact = ctx.assemble();
343 Ok(artifact)
344}
345
346pub fn write_layer(
387 base_image: &Path,
388 root: &Path,
389 config: &WriteConfig,
390) -> Result<WriteArtifact, WriteError> {
391 let base_root = load_base_drop_index(base_image)?.1;
393 let base_drop_index: std::sync::Arc<dyn BaseDropSet> = {
394 #[cfg(feature = "sparse-index")]
395 {
396 match SparseBackedBaseIndex::open(base_image) {
397 Some(idx) => std::sync::Arc::new(idx),
398 None => std::sync::Arc::new(load_base_drop_index(base_image)?.0),
399 }
400 }
401 #[cfg(not(feature = "sparse-index"))]
402 {
403 std::sync::Arc::new(load_base_drop_index(base_image)?.0)
404 }
405 };
406
407 let mut ctx = WriteContext::new();
408 ctx.chunker = chunker_from_config(config)?;
409 ctx.categorizers_disabled = config.categorizers.is_empty();
410 ctx.rw_mode = matches!(config.mode, crate::config::ImageMode::ReadWrite(_));
411 ctx.auto_turnover = config.turnover_threshold > 0;
412 ctx.collect_dict_samples = config.dictionaries.enabled;
413 ctx.inline_threshold = config.defaults.inline_threshold as usize;
414 ctx.metadata_externalize_threshold = config.defaults.metadata_externalize_threshold;
415 ctx.emit_shared_inline = config.defaults.shared_inline;
416 ctx.base_drop_index = Some(base_drop_index);
417 ctx.base_root = Some(base_root);
418
419 let root_inode_number = ctx.walk(root)?;
421 ctx.root_inode_number = root_inode_number;
422 write_directory_body(&mut ctx, config)?;
423 Ok(ctx.assemble())
424}
425
426pub trait BaseDropSet: Send + Sync {
439 fn base_contains(&self, drop_id: &[u8; 32]) -> bool;
442}
443
444impl BaseDropSet for std::collections::HashSet<[u8; 32]> {
445 fn base_contains(&self, drop_id: &[u8; 32]) -> bool {
446 self.contains(drop_id)
447 }
448}
449
450#[cfg(feature = "sparse-index")]
456pub struct SparseBackedBaseIndex {
457 bloom: crate::sparse_index::SparseIndexReader,
458 manifest_path: std::path::PathBuf,
459 exact: std::sync::OnceLock<std::collections::HashSet<[u8; 32]>>,
460}
461
462#[cfg(feature = "sparse-index")]
463impl SparseBackedBaseIndex {
464 #[must_use]
467 pub fn open(base_image: &Path) -> Option<Self> {
468 let sidecar = base_image.with_extension("lim.sparse");
469 let bloom = crate::sparse_index::SparseIndexReader::from_file(&sidecar)?;
470 Some(Self {
471 bloom,
472 manifest_path: base_image.to_path_buf(),
473 exact: std::sync::OnceLock::new(),
474 })
475 }
476
477 fn load_exact(&self) -> &std::collections::HashSet<[u8; 32]> {
478 self.exact.get_or_init(|| {
479 let bytes = std::fs::read(&self.manifest_path).unwrap_or_default();
483 let mut cursor = ManifestCursor::new(&bytes);
484 let _ = parse_manifest_header(&mut cursor);
485 let _ = limnifs_core::parse_feature_flags_section(&mut cursor);
486 let _ = limnifs_core::parse_metadata_reference(&mut cursor);
487 let Ok(index) = parse_slab_index(&mut cursor) else {
488 return std::collections::HashSet::new();
489 };
490 match SlabStore::load_mmap(&self.manifest_path, &index) {
491 Ok(store) => store.drop_index_keys().copied().collect(),
492 Err(_) => std::collections::HashSet::new(),
493 }
494 })
495 }
496}
497
498#[cfg(feature = "sparse-index")]
499impl BaseDropSet for SparseBackedBaseIndex {
500 fn base_contains(&self, drop_id: &[u8; 32]) -> bool {
501 if !self.bloom.probably_contains(drop_id) {
506 return false;
507 }
508 self.load_exact().contains(drop_id)
509 }
510}
511
512#[cfg(feature = "sparse-index")]
520pub fn emit_sparse_sidecar(artifact: &WriteArtifact, image_path: &Path) -> Result<(), WriteError> {
521 let all: std::collections::HashSet<[u8; 32]> = artifact
522 .slabs
523 .iter()
524 .flat_map(|s| s.drop_ids.iter().copied())
525 .collect();
526 let mut writer = crate::sparse_index::SparseIndexWriter::new(
527 all.len().max(1),
528 crate::sparse_index::DEFAULT_FPP,
529 );
530 writer.insert_all(&all);
531 let sidecar = image_path.with_extension("lim.sparse");
532 writer.write_to_file(&sidecar).map_err(WriteError::Io)
533}
534
535fn load_base_drop_index(
536 base_image: &Path,
537) -> Result<(std::collections::HashSet<[u8; 32]>, [u8; 32]), WriteError> {
538 let manifest_bytes = std::fs::read(base_image)?;
539 let mut cursor = ManifestCursor::new(&manifest_bytes);
540 let _ = parse_manifest_header(&mut cursor).map_err(io_core)?;
541 let _ = limnifs_core::parse_feature_flags_section(&mut cursor);
543 let _ = limnifs_core::parse_metadata_reference(&mut cursor);
544 let slab_index = parse_slab_index(&mut cursor).map_err(io_core)?;
545 let store = SlabStore::load_mmap(base_image, &slab_index).map_err(io_core)?;
546 let drop_set: std::collections::HashSet<[u8; 32]> = store.drop_index_keys().copied().collect();
547 let root = *compute_merkle_root_from_sections(&manifest_bytes).as_bytes();
553 Ok((drop_set, root))
554}
555
556fn compute_merkle_root_from_sections(manifest: &[u8]) -> ManifestRoot {
562 use limnifs_core::SectionHashes;
563 let mut cursor = ManifestCursor::new(manifest);
564 let header_start = 0;
565 if parse_manifest_header(&mut cursor).is_err() {
566 return ManifestRoot::from_bytes([0u8; 32]);
568 }
569 let header_end = cursor.position();
570 let flags_start = header_end;
572 let flags_end = match limnifs_core::parse_feature_flags_section(&mut cursor) {
573 Ok(_) => cursor.position(),
574 Err(_) => flags_start,
575 };
576 let meta_ref_start = flags_end;
577 let metadata_reference = match limnifs_core::parse_metadata_reference(&mut cursor) {
578 Ok(m) => Some(m),
579 Err(_) => None,
580 };
581 let meta_ref_end = cursor.position();
582 let slab_index_start = meta_ref_end;
583 let _ = parse_slab_index(&mut cursor);
584 let slab_index_end = cursor.position();
585 let history_start = slab_index_end;
586 let _ = limnifs_core::parse_history(&mut cursor);
587 let history_end = cursor.position();
588
589 let hashes = SectionHashes {
590 metadata: metadata_reference
591 .map(|m| m.metadata_hash)
592 .unwrap_or_else(hash_empty_section),
593 format_header: hash_section(&manifest[header_start..header_end]),
594 feature_flags: hash_section(&manifest[flags_start..flags_end]),
595 metadata_reference: hash_section(&manifest[meta_ref_start..meta_ref_end]),
596 slab_index: hash_section(&manifest[slab_index_start..slab_index_end]),
597 crypto_params: hash_empty_section(),
598 ec_params: hash_empty_section(),
599 dms_policy: hash_empty_section(),
600 delta_linkage: hash_empty_section(),
601 history: hash_section(&manifest[history_start..history_end]),
602 };
603 compute_merkle_root(&hashes)
604}
605
606fn io_core(e: limnifs_core::CoreError) -> WriteError {
607 WriteError::Io(std::io::Error::other(format!("base image load: {e}")))
608}
609
610fn write_directory_body(ctx: &mut WriteContext, config: &WriteConfig) -> Result<(), WriteError> {
615 use rayon::prelude::*;
616
617 ctx.metadata_codec = config
618 .metadata_codec_id()
619 .unwrap_or(limnifs_core::codec::CODEC_BROTLI);
620
621 ctx.chunker = chunker_from_config(config)?;
622
623 let pending = std::mem::take(&mut ctx.pending_files);
624 if pending.is_empty() {
625 return Ok(());
626 }
627 ctx.inline_threshold = config.defaults.inline_threshold as usize;
628 ctx.metadata_externalize_threshold = config.defaults.metadata_externalize_threshold;
629 ctx.emit_shared_inline = config.defaults.shared_inline;
630 let chunker = ctx.chunker.clone();
631 let classifier = ctx.classifier;
632 let text_codec = config.text_codec_id().unwrap_or(0x04);
633 let binary_codec = config.binary_codec_id().unwrap_or(0x01);
634 let tunables = config.to_core_tunables();
635 let use_categorizers = !config.categorizers.is_empty();
636 let skip_chunking = config.skip_chunking;
637 let registry = config
638 .codec_registry()
639 .map_err(|e| WriteError::Io(std::io::Error::other(format!("codec registry: {e}"))))?;
640 let tournament_codec_ids: Vec<u8> = config
641 .tournament
642 .codecs
643 .iter()
644 .filter_map(|n| registry.lookup_by_name(n))
645 .collect();
646 let tournament_spec = TournamentSpec {
647 codec_ids: tournament_codec_ids,
648 min_size: config.tournament.min_size_threshold as usize,
649 skip_for_binary: config.tournament.skip_for_binary,
650 short_circuit_permille: config.tournament.short_circuit_threshold,
651 };
652 let base_drop_index: Option<&dyn BaseDropSet> = ctx.base_drop_index.as_deref();
653 let inline_threshold = ctx.inline_threshold;
654 let max_drop_size = config.defaults.max_drop_size as usize;
655 let seekable_drops = config.defaults.seekable_drops;
656 let seekable_drops = config.defaults.seekable_drops;
657 let results: Vec<ChunkedFileResult> = pending
658 .par_iter()
659 .map(|pf| {
660 process_file(
661 pf,
662 &chunker,
663 classifier,
664 text_codec,
665 binary_codec,
666 &tunables,
667 use_categorizers,
668 skip_chunking,
669 &tournament_spec,
670 base_drop_index,
671 inline_threshold,
672 max_drop_size,
673 seekable_drops,
674 config.categorizers.as_slice(),
675 &|name| {
676 config
677 .codec_registry()
678 .ok()
679 .and_then(|r| r.lookup_by_name(name))
680 },
681 )
682 })
683 .collect::<Result<Vec<_>, _>>()?;
684
685 for (pf, result) in pending.iter().zip(results) {
686 ctx.merge_chunked_file(pf, result);
687 }
688 ctx.train_and_apply_dictionary(&config.dictionaries);
689 Ok(())
690}
691
692pub fn write_directory_with_config(
694 root: &Path,
695 config: &WriteConfig,
696) -> Result<WriteArtifact, WriteError> {
697 let mut ctx = WriteContext::new();
698 ctx.chunker = chunker_from_config(config)?;
699 ctx.categorizers_disabled = config.categorizers.is_empty();
700 ctx.rw_mode = matches!(config.mode, crate::config::ImageMode::ReadWrite(_));
701 ctx.auto_turnover = config.turnover_threshold > 0;
702 ctx.collect_dict_samples = config.dictionaries.enabled;
703
704 write_directory_streaming(&mut ctx, root, config)?;
705 Ok(ctx.assemble())
706}
707
708fn write_directory_streaming(
722 ctx: &mut WriteContext,
723 root: &Path,
724 config: &WriteConfig,
725) -> Result<(), WriteError> {
726 use rayon::prelude::*;
727
728 ctx.metadata_codec = config
729 .metadata_codec_id()
730 .unwrap_or(limnifs_core::codec::CODEC_BROTLI);
731
732 ctx.chunker = chunker_from_config(config)?;
733
734 let chunker = ctx.chunker.clone();
735 let classifier = ctx.classifier;
736 let text_codec = config.text_codec_id().unwrap_or(0x04);
737 let binary_codec = config.binary_codec_id().unwrap_or(0x01);
738 let tunables = config.to_core_tunables();
739 let use_categorizers = !config.categorizers.is_empty();
740 let skip_chunking = config.skip_chunking;
741 let registry = config
742 .codec_registry()
743 .map_err(|e| WriteError::Io(std::io::Error::other(format!("codec registry: {e}"))))?;
744 let tournament_codec_ids: Vec<u8> = config
745 .tournament
746 .codecs
747 .iter()
748 .filter_map(|n| registry.lookup_by_name(n))
749 .collect();
750 let tournament_spec = TournamentSpec {
751 codec_ids: tournament_codec_ids,
752 min_size: config.tournament.min_size_threshold as usize,
753 skip_for_binary: config.tournament.skip_for_binary,
754 short_circuit_permille: config.tournament.short_circuit_threshold,
755 };
756 let base_drop_index = ctx.base_drop_index.clone();
759 let inline_threshold = ctx.inline_threshold;
760 let max_drop_size = config.defaults.max_drop_size as usize;
761 let seekable_drops = config.defaults.seekable_drops;
762
763 ctx.inline_threshold = config.defaults.inline_threshold as usize;
764 ctx.metadata_externalize_threshold = config.defaults.metadata_externalize_threshold;
765 ctx.emit_shared_inline = config.defaults.shared_inline;
766
767 const PIPELINE_CAPACITY: usize = 256;
771 let (tx, rx) = std::sync::mpsc::sync_channel::<PendingFile>(PIPELINE_CAPACITY);
772 ctx.pending_sink = Some(tx);
773
774 let (root_inode_number, mut results): (
775 u64,
776 Vec<(usize, PendingFile, Result<ChunkedFileResult, WriteError>)>,
777 ) = std::thread::scope(|scope| {
778 let producer = {
779 let ctx = &mut *ctx;
780 let root = root;
781 scope.spawn(move || {
782 let r = ctx.walk(root);
783 ctx.pending_sink = None;
787 r
788 })
789 };
790 let results = rx
793 .into_iter()
794 .enumerate()
795 .par_bridge()
796 .map(|(i, pf)| {
797 let r = process_file(
798 &pf,
799 &chunker,
800 classifier,
801 text_codec,
802 binary_codec,
803 &tunables,
804 use_categorizers,
805 skip_chunking,
806 &tournament_spec,
807 base_drop_index.as_deref(),
808 inline_threshold,
809 max_drop_size,
810 seekable_drops,
811 config.categorizers.as_slice(),
812 &|name| {
813 config
814 .codec_registry()
815 .ok()
816 .and_then(|r| r.lookup_by_name(name))
817 },
818 );
819 (i, pf, r)
820 })
821 .collect();
822 let joined = producer
823 .join()
824 .unwrap_or_else(|_| {
825 Err(WriteError::Io(std::io::Error::other(
826 "walk thread panicked",
827 )))
828 })
829 .map(|n| (n, results));
830 joined
833 })?;
834 ctx.pending_sink = None;
835 ctx.root_inode_number = root_inode_number;
836
837 results.sort_unstable_by_key(|(i, _, _)| *i);
838 for (_, pf, r) in results {
842 ctx.merge_chunked_file(&pf, r?);
843 }
844 ctx.train_and_apply_dictionary(&config.dictionaries);
845 Ok(())
846}
847
848pub(crate) type RawDrop = ([u8; 32], Vec<u8>, std::sync::Arc<[u8]>, u8, u8);
855pub(crate) struct ChunkedFileResult {
857 drops: Vec<RawDrop>, slices: Vec<PendingSlice>,
859}
860
861struct TournamentSpec {
869 codec_ids: Vec<u8>,
873 min_size: usize,
877 skip_for_binary: bool,
881 short_circuit_permille: u32,
886}
887
888fn chunker_from_config(config: &WriteConfig) -> Result<FastCDC, WriteError> {
907 FastCDC::new(
908 config.chunking.min_chunk_size as usize,
909 config.chunking.avg_chunk_size as usize,
910 config.chunking.max_chunk_size as usize,
911 )
912 .map_err(|e| WriteError::Io(std::io::Error::other(format!("chunking config: {e}"))))
913}
914
915pub(crate) fn seekable_or_monolithic(
922 codec: u8,
923 plaintext: &[u8],
924 compressed: std::sync::Arc<[u8]>,
925 tunables: &limnifs_core::codec::CodecTunables,
926 seekable_drops: bool,
927 threshold: usize,
928) -> (std::sync::Arc<[u8]>, u8) {
929 use limnifs_core::seekable::{
930 encode_seekable, is_seekable_codec, DROP_FLAG_SEEKABLE as FLAG, SEEKABLE_EMISSION_THRESHOLD,
931 };
932 if seekable_drops && plaintext.len() > threshold && is_seekable_codec(codec) {
933 if let Ok(container) = encode_seekable(codec, plaintext, tunables) {
934 return (container.into(), FLAG);
935 }
936 }
937 (compressed, 0)
938}
939
940pub(crate) const SEEKABLE_CHUNK_EMISSION_THRESHOLD: usize =
947 limnifs_core::seekable::SEEKABLE_FRAME_SIZE;
948
949fn process_whole_file_drop(
950 pf: &PendingFile,
951 data: &[u8],
952 cat: file_categorizer::Categorization,
953 tunables: &limnifs_core::codec::CodecTunables,
954 seekable_drops: bool,
955) -> Result<ChunkedFileResult, WriteError> {
956 let _ = pf;
957 let drop_id = hash_section(data);
958 let file_len = u64::try_from(data.len()).unwrap_or(u64::MAX);
959
960 let (mut best_codec, mut best_compressed): (u8, std::sync::Arc<[u8]>) =
965 match limnifs_core::codec::compress_with_tunables(
966 limnifs_core::codec::CODEC_BROTLI,
967 data,
968 tunables,
969 ) {
970 Ok(c) => (limnifs_core::codec::CODEC_BROTLI, c.into()),
971 Err(_) => match limnifs_core::codec::compress_with_tunables(
972 limnifs_core::codec::CODEC_ZSTD,
973 data,
974 tunables,
975 ) {
976 Ok(c) => (limnifs_core::codec::CODEC_ZSTD, c.into()),
977 Err(_) => (limnifs_core::codec::CODEC_STORE, data.to_vec().into()),
978 },
979 };
980
981 let brotli_ratio = best_compressed.len() as f64 / data.len() as f64;
985 if brotli_ratio > 0.05 {
986 if let Ok(zstd_c) = limnifs_core::codec::compress_with_tunables(
987 limnifs_core::codec::CODEC_ZSTD,
988 data,
989 tunables,
990 ) {
991 if zstd_c.len() < best_compressed.len() {
992 best_codec = limnifs_core::codec::CODEC_ZSTD;
993 best_compressed = zstd_c.into();
994 }
995 }
996 }
997
998 let general_ratio = best_compressed.len() as f64 / data.len() as f64;
1004 if general_ratio > 0.15 || cat.codec_id == limnifs_core::codec::CODEC_RICEPP {
1005 let spec_result = if cat.codec_id == limnifs_core::codec::CODEC_FSST_BROTLI {
1009 limnifs_core::codec::fsst_brotli::compress_with_baseline(data, Some(&best_compressed))
1010 } else {
1011 limnifs_core::codec::compress_with_tunables(cat.codec_id, data, tunables)
1015 };
1016 if let Ok(spec_c) = spec_result {
1017 if spec_c.len() < best_compressed.len() {
1018 best_codec = cat.codec_id;
1019 best_compressed = spec_c.into();
1020 }
1021 }
1022 }
1023
1024 let (best_compressed, flags) = seekable_or_monolithic(
1025 best_codec,
1026 data,
1027 best_compressed,
1028 tunables,
1029 seekable_drops,
1030 limnifs_core::seekable::SEEKABLE_EMISSION_THRESHOLD,
1031 );
1032 Ok(ChunkedFileResult {
1033 drops: vec![(drop_id, data.to_vec(), best_compressed, best_codec, flags)],
1034 slices: vec![PendingSlice {
1035 drop_id,
1036 file_byte_start: 0,
1037 file_byte_end: file_len,
1038 }],
1039 })
1040}
1041
1042fn compress_chunk_with_tournament(
1064 chunk: &[u8],
1065 class: classifier::Class,
1066 text_codec: u8,
1067 binary_codec: u8,
1068 tunables: &limnifs_core::codec::CodecTunables,
1069 tournament: &TournamentSpec,
1070) -> (u8, std::sync::Arc<[u8]>) {
1071 use classifier::Class;
1072
1073 let preferred = match class {
1074 Class::Binary => binary_codec,
1075 Class::Text | Class::Code | Class::Sparse => text_codec,
1076 _ => limnifs_core::codec::CODEC_STORE,
1077 };
1078
1079 if preferred == limnifs_core::codec::CODEC_STORE {
1080 return (limnifs_core::codec::CODEC_STORE, chunk.to_vec().into());
1081 }
1082 if class == Class::Binary && tournament.skip_for_binary {
1083 return compress_chunk_one(chunk, preferred, tunables);
1084 }
1085 if chunk.len() < tournament.min_size {
1086 return compress_chunk_one(chunk, preferred, tunables);
1087 }
1088
1089 let mut best: Option<(u8, std::sync::Arc<[u8]>)> = None;
1090 for &codec_id in &tournament.codec_ids {
1091 if codec_id == limnifs_core::codec::CODEC_STORE {
1092 continue;
1093 }
1094 let c = match limnifs_core::codec::compress_with_tunables(codec_id, chunk, tunables) {
1095 Ok(c) => c,
1096 Err(_) => continue,
1097 };
1098 if c.len() >= chunk.len() {
1099 continue;
1100 }
1101 let ratio_permille = (c.len() as u64 * 1000 / chunk.len() as u64) as u32;
1102 let is_best_so_far = best.as_ref().map_or(true, |(_, b)| c.len() < b.len());
1103 if is_best_so_far {
1104 best = Some((codec_id, c.into()));
1105 }
1106 if tournament.short_circuit_permille > 0
1107 && ratio_permille <= tournament.short_circuit_permille
1108 {
1109 break;
1110 }
1111 }
1112
1113 best.unwrap_or_else(|| (limnifs_core::codec::CODEC_STORE, chunk.to_vec().into()))
1114}
1115
1116fn compress_chunk_one(
1119 chunk: &[u8],
1120 codec_id: u8,
1121 tunables: &limnifs_core::codec::CodecTunables,
1122) -> (u8, std::sync::Arc<[u8]>) {
1123 if codec_id == limnifs_core::codec::CODEC_STORE {
1124 return (limnifs_core::codec::CODEC_STORE, chunk.to_vec().into());
1125 }
1126 match limnifs_core::codec::compress_with_tunables(codec_id, chunk, tunables) {
1127 Ok(c) if c.len() < chunk.len() => (codec_id, c.into()),
1128 _ => (limnifs_core::codec::CODEC_STORE, chunk.to_vec().into()),
1129 }
1130}
1131
1132fn process_file(
1136 pf: &PendingFile,
1137 chunker: &FastCDC,
1138 classifier: classifier::Classifier,
1139 text_codec: u8,
1140 binary_codec: u8,
1141 tunables: &limnifs_core::codec::CodecTunables,
1142 use_categorizers: bool,
1143 skip_chunking: bool,
1144 tournament: &TournamentSpec,
1145 base_drop_index: Option<&dyn BaseDropSet>,
1146 inline_threshold: usize,
1147 max_drop_size: usize,
1148 seekable_drops: bool,
1149 categorizer_config: &[crate::config::CategorizerConfig],
1150 codec_name_resolver: &dyn Fn(&str) -> Option<u8>,
1151) -> Result<ChunkedFileResult, WriteError> {
1152 let file_len_estimate = std::fs::metadata(&pf.path)
1163 .map(|m| m.len() as usize)
1164 .unwrap_or(0);
1165 let mmap_handle: memmap2::Mmap;
1172 let small: Vec<u8>;
1173 let data: &[u8] = if file_len_estimate >= MMAP_READ_THRESHOLD {
1174 let file = std::fs::File::open(&pf.path)?;
1175 #[allow(unsafe_code)]
1179 let mapped = unsafe { memmap2::Mmap::map(&file) }.map_err(WriteError::Io)?;
1180 mmap_handle = mapped;
1181 &mmap_handle[..]
1182 } else {
1183 small = std::fs::read(&pf.path)?;
1184 &small[..]
1185 };
1186 let file_len = data.len();
1187
1188 if skip_chunking && file_len > inline_threshold {
1195 let drop_id = hash_section(&data);
1196 let class = classifier.classify(&data);
1197 let preferred_codec = match class {
1198 classifier::Class::Binary => binary_codec,
1199 _ => text_codec,
1200 };
1201 let (codec_id, compressed): (u8, std::sync::Arc<[u8]>) =
1202 match limnifs_core::codec::compress_with_tunables(preferred_codec, &data, tunables) {
1203 Ok(c) if c.len() < data.len() => (preferred_codec, c.into()),
1204 _ => (limnifs_core::codec::CODEC_STORE, data.to_vec().into()),
1205 };
1206 let (compressed, flags) = seekable_or_monolithic(
1207 codec_id,
1208 &data,
1209 compressed,
1210 tunables,
1211 seekable_drops,
1212 SEEKABLE_CHUNK_EMISSION_THRESHOLD,
1213 );
1214 return Ok(ChunkedFileResult {
1215 drops: vec![(drop_id, data.to_vec(), compressed, codec_id, flags)],
1216 slices: vec![PendingSlice {
1217 drop_id,
1218 file_byte_start: 0,
1219 file_byte_end: file_len as u64,
1220 }],
1221 });
1222 }
1223
1224 if use_categorizers {
1225 let config_cat = file_categorizer::ConfigCategorizer::new(categorizer_config.to_vec());
1228 if let Some(cat) = config_cat.categorize(&pf.path, &data) {
1229 if let Some(codec_id) = file_categorizer::resolve_config_categorization(
1230 &cat,
1231 categorizer_config,
1232 codec_name_resolver,
1233 ) {
1234 let within_cap = max_drop_size == 0 || file_len <= max_drop_size;
1235 let needs_whole_file = matches!(
1236 codec_id,
1237 limnifs_core::codec::CODEC_FLAC | limnifs_core::codec::CODEC_RICEPP
1238 );
1239 if within_cap && (needs_whole_file || file_len <= WHOLE_FILE_MAX_SIZE) {
1240 let mut cat = cat;
1241 cat.codec_id = codec_id;
1242 return process_whole_file_drop(pf, &data, cat, tunables, seekable_drops);
1243 }
1244 }
1245 }
1246 if let Some(cat) = file_categorizer::default_registry().categorize(&pf.path, &data) {
1247 let needs_whole_file = matches!(
1248 cat.codec_id,
1249 limnifs_core::codec::CODEC_FLAC | limnifs_core::codec::CODEC_RICEPP
1250 );
1251 let within_cap = max_drop_size == 0 || file_len <= max_drop_size;
1254 if within_cap && (needs_whole_file || file_len <= WHOLE_FILE_MAX_SIZE) {
1255 return process_whole_file_drop(pf, &data, cat, tunables, seekable_drops);
1256 }
1257 }
1258 }
1259
1260 let chunks = chunker.chunk_slice(&data);
1261
1262 use rayon::prelude::*;
1271 let drop_ids: Vec<[u8; 32]> = chunks.par_iter().map(|chunk| hash_section(chunk)).collect();
1272
1273 let mut slices = Vec::with_capacity(chunks.len());
1274 let mut file_offset: u64 = 0;
1275 let mut seen_in_file: std::collections::HashSet<[u8; 32]> =
1276 std::collections::HashSet::with_capacity(chunks.len());
1277 let mut unique_chunks: Vec<(&[u8], [u8; 32])> = Vec::with_capacity(chunks.len());
1278 for (chunk, drop_id) in chunks.iter().zip(drop_ids) {
1279 let chunk_len = u64::try_from(chunk.len()).expect("chunk len fits u64");
1280 slices.push(PendingSlice {
1281 drop_id,
1282 file_byte_start: file_offset,
1283 file_byte_end: file_offset + chunk_len,
1284 });
1285 file_offset += chunk_len;
1286 if seen_in_file.insert(drop_id) {
1287 unique_chunks.push((chunk, drop_id));
1288 }
1289 }
1290
1291 thread_local! {
1303 static COMPRESS_CACHE: std::cell::RefCell<std::collections::HashMap<[u8; 32], (u8, std::sync::Arc<[u8]>)>> =
1304 std::cell::RefCell::new(std::collections::HashMap::new());
1305 }
1306 const COMPRESS_CACHE_MAX_ENTRIES: usize = 100_000;
1307 let drops: Vec<RawDrop> = unique_chunks
1308 .par_iter()
1309 .map(|(chunk, drop_id)| {
1310 if let Some(base) = base_drop_index {
1313 if base.base_contains(drop_id) {
1314 return (*drop_id, Vec::new(), Vec::new().into(), CODEC_REFERENCED, 0);
1315 }
1316 }
1317 let class = classifier.classify(chunk);
1318 let cached = COMPRESS_CACHE.with(|c| {
1321 c.borrow()
1322 .get(drop_id)
1323 .map(|(cid, comp)| (*cid, comp.clone()))
1324 });
1325 let (codec_id, compressed) = if let Some(c) = cached {
1326 c
1327 } else {
1328 let new = compress_chunk_with_tournament(
1329 chunk,
1330 class,
1331 text_codec,
1332 binary_codec,
1333 tunables,
1334 tournament,
1335 );
1336 COMPRESS_CACHE.with(|c| {
1338 let mut cache = c.borrow_mut();
1339 if cache.len() < COMPRESS_CACHE_MAX_ENTRIES {
1340 cache.insert(*drop_id, new.clone());
1342 }
1343 });
1344 new
1345 };
1346 let (compressed, flags) = seekable_or_monolithic(
1352 codec_id,
1353 chunk,
1354 compressed,
1355 tunables,
1356 seekable_drops,
1357 SEEKABLE_CHUNK_EMISSION_THRESHOLD,
1358 );
1359 (*drop_id, chunk.to_vec(), compressed, codec_id, flags)
1360 })
1361 .collect();
1362
1363 let _ = file_len;
1364 Ok(ChunkedFileResult { drops, slices })
1365}
1366
1367struct PendingDrop {
1368 id: [u8; 32],
1369 plaintext_len: u32,
1375 compressed: std::sync::Arc<[u8]>,
1376 codec: u8,
1377 dict_id: u8,
1381 plaintext: Option<Vec<u8>>,
1386 flags: u8,
1390}
1391
1392impl PendingDrop {
1393 fn len_in_window(&self) -> u32 {
1397 u32::try_from(self.compressed.len()).expect("compressed fits u32")
1398 }
1399
1400 fn plaintext_len_value(&self) -> u32 {
1402 self.plaintext_len
1403 }
1404
1405 fn slab_footprint(&self) -> usize {
1408 48 + self.compressed.len()
1409 }
1410}
1411
1412struct PendingSlice {
1417 drop_id: [u8; 32],
1418 file_byte_start: u64,
1419 file_byte_end: u64,
1420}
1421
1422#[derive(Clone)]
1425struct PendingFile {
1426 inode_number: u64,
1427 path: PathBuf,
1428 mtime_ns: u64,
1429 file_len: u64,
1430}
1431
1432struct PendingInode {
1433 number: u64,
1434 mode: u32,
1435 mtime_ns: u64,
1436 content: PendingContent,
1437}
1438
1439enum PendingContent {
1440 Inline(Vec<u8>),
1441 Symlink(String),
1443 DropBacked {
1444 file_len: u64,
1445 slices: Vec<PendingSlice>,
1446 },
1447 Directory(Vec<(String, u64, u8)>),
1448}
1449
1450struct DirNode {
1451 entries: Vec<(String, u64, u8)>,
1452 bytes: Vec<u8>,
1453 hash: [u8; 32],
1454}
1455
1456struct WriteContext {
1457 next_inode: u64,
1458 inodes: Vec<PendingInode>,
1459 dir_nodes: Vec<DirNode>,
1460 drops: Vec<PendingDrop>,
1461 drop_index: HashSet<[u8; 32]>,
1462 pending_files: Vec<PendingFile>,
1463 file_count: usize,
1464 dir_count: usize,
1465 root_inode_number: u64,
1466 chunker: FastCDC,
1467 classifier: classifier::Classifier,
1468 shared_inline_map: HashMap<[u8; 32], usize>,
1469 shared_inline_table: Vec<Vec<u8>>,
1470 profile_name: Option<String>,
1472 metadata_codec: u8,
1475 categorizers_disabled: bool,
1477 rw_mode: bool,
1479 auto_turnover: bool,
1481 collect_dict_samples: bool,
1484 dict_samples_by_class: HashMap<crate::classifier::Class, Vec<Vec<u8>>>,
1491 trained_dicts_by_class: HashMap<crate::classifier::Class, crate::dictionary::TrainedDictionary>,
1496 base_drop_index: Option<std::sync::Arc<dyn BaseDropSet>>,
1502 base_root: Option<[u8; 32]>,
1507 metadata_externalize_threshold: usize,
1512 emit_shared_inline: bool,
1518 inline_threshold: usize,
1523 pending_sink: Option<std::sync::mpsc::SyncSender<PendingFile>>,
1529}
1530
1531impl WriteContext {
1532 const MAX_DICT_SAMPLES: usize = 1000;
1535
1536 fn new() -> Self {
1537 Self {
1538 next_inode: 1,
1539 inodes: Vec::new(),
1540 dir_nodes: Vec::new(),
1541 drops: Vec::new(),
1542 drop_index: HashSet::new(),
1543 pending_files: Vec::new(),
1544 file_count: 0,
1545 dir_count: 0,
1546 root_inode_number: 0,
1547 chunker: FastCDC::default(),
1548 classifier: classifier::Classifier,
1549 shared_inline_map: HashMap::new(),
1550 shared_inline_table: Vec::new(),
1551 profile_name: None,
1552 metadata_codec: limnifs_core::codec::CODEC_BROTLI,
1553 categorizers_disabled: false,
1554 rw_mode: false,
1555 auto_turnover: false,
1556 collect_dict_samples: false,
1557 dict_samples_by_class: HashMap::new(),
1558 trained_dicts_by_class: HashMap::new(),
1559 base_drop_index: None,
1560 base_root: None,
1561 pending_sink: None,
1562 inline_threshold: INLINE_THRESHOLD,
1563 metadata_externalize_threshold: METADATA_EXTERNALIZE_THRESHOLD,
1564 emit_shared_inline: true,
1565 }
1566 }
1567
1568 fn alloc_inode(&mut self) -> u64 {
1569 let n = self.next_inode;
1570 self.next_inode += 1;
1571 n
1572 }
1573
1574 fn build_shared_inline_table(&mut self) {
1578 let mut counts: HashMap<[u8; 32], usize> = HashMap::new();
1579 for inode in &self.inodes {
1580 if let PendingContent::Inline(data) = &inode.content {
1581 let h = hash_section(data);
1582 *counts.entry(h).or_default() += 1;
1583 }
1584 }
1585 for inode in &self.inodes {
1587 if let PendingContent::Inline(data) = &inode.content {
1588 let h = hash_section(data);
1589 if counts.get(&h).copied().unwrap_or(0) > 1
1590 && !self.shared_inline_map.contains_key(&h)
1591 {
1592 let idx = self.shared_inline_table.len();
1593 self.shared_inline_table.push(data.clone());
1594 self.shared_inline_map.insert(h, idx);
1595 }
1596 }
1597 }
1598 }
1599
1600 fn merge_chunked_file(&mut self, pf: &PendingFile, result: ChunkedFileResult) {
1603 for (drop_id, plaintext, compressed, codec, flags) in result.drops {
1604 if self.drop_index.insert(drop_id) {
1605 let retain_plaintext =
1611 self.collect_dict_samples && codec == limnifs_core::codec::CODEC_ZSTD;
1612 if retain_plaintext {
1613 let total: usize = self.dict_samples_by_class.values().map(Vec::len).sum();
1614 if total < Self::MAX_DICT_SAMPLES {
1615 let class = self.classifier.classify(&plaintext);
1616 self.dict_samples_by_class
1617 .entry(class)
1618 .or_default()
1619 .push(plaintext.clone());
1620 }
1621 }
1622 self.drops.push(PendingDrop {
1623 id: drop_id,
1624 plaintext_len: u32::try_from(plaintext.len()).unwrap_or(u32::MAX),
1625 compressed,
1626 codec,
1627 dict_id: limnifs_core::drop_record::NO_DICT,
1628 plaintext: if retain_plaintext {
1629 Some(plaintext)
1630 } else {
1631 None
1632 },
1633 flags,
1634 });
1635 }
1636 }
1637 self.inodes.push(PendingInode {
1638 number: pf.inode_number,
1639 mode: 0o100_644,
1640 mtime_ns: pf.mtime_ns,
1641 content: PendingContent::DropBacked {
1642 file_len: pf.file_len,
1643 slices: result.slices,
1644 },
1645 });
1646 }
1647
1648 #[allow(dead_code)]
1659 fn deepen_drop(&self, drop_id: [u8; 32], plaintext: &[u8]) -> PendingDrop {
1660 let class = self.classifier.classify(plaintext);
1661 let (codec, compressed): (u8, std::sync::Arc<[u8]>) = match class {
1662 classifier::Class::Text | classifier::Class::Code | classifier::Class::Binary => {
1663 let c = limnifs_core::codec::compress_lz4_with_size(plaintext);
1664 (limnifs_core::codec::CODEC_LZ4, c.into())
1665 }
1666 _ => (limnifs_core::codec::CODEC_STORE, plaintext.to_vec().into()),
1667 };
1668 PendingDrop {
1669 id: drop_id,
1670 plaintext_len: u32::try_from(plaintext.len()).unwrap_or(u32::MAX),
1671 compressed,
1672 codec,
1673 dict_id: limnifs_core::drop_record::NO_DICT,
1674 plaintext: None,
1675 flags: 0,
1676 }
1677 }
1678
1679 fn walk(&mut self, path: &Path) -> Result<u64, WriteError> {
1680 let meta = std::fs::symlink_metadata(path)?;
1681 let file_type = meta.file_type();
1682 let mtime_ns = meta
1683 .modified()
1684 .ok()
1685 .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
1686 .map_or(0u128, |d| d.as_nanos());
1687 let mtime_ns: u64 = mtime_ns.try_into().unwrap_or(0);
1688
1689 if file_type.is_dir() {
1690 self.dir_count += 1;
1691 let inode_number = self.alloc_inode();
1692 let mut entries: Vec<(String, u64, u8)> = Vec::new();
1693
1694 for entry in std::fs::read_dir(path)? {
1695 let entry = entry?;
1696 let name = entry.file_name().to_string_lossy().into_owned();
1697 let child_path = entry.path();
1698 let child_inode = self.walk(&child_path)?;
1699 let ft = entry.file_type()?;
1702 let entry_type = if ft.is_symlink() {
1703 0x03
1704 } else if ft.is_dir() {
1705 0x02
1706 } else {
1707 0x01
1708 };
1709 entries.push((name, child_inode, entry_type));
1710 }
1711
1712 entries.sort_by(|a, b| a.0.cmp(&b.0));
1713 let dir_node = encode_dir_node(&entries);
1714 self.dir_nodes.push(dir_node);
1715 self.inodes.push(PendingInode {
1716 number: inode_number,
1717 mode: 0o040_755,
1718 mtime_ns,
1719 content: PendingContent::Directory(entries),
1720 });
1721 Ok(inode_number)
1722 } else if file_type.is_file() {
1723 self.file_count += 1;
1724 let inode_number = self.alloc_inode();
1725 let file_len = meta.len();
1726
1727 if file_len <= u64::try_from(self.inline_threshold).unwrap_or(u64::MAX) {
1728 let data = std::fs::read(path)?;
1729 self.inodes.push(PendingInode {
1730 number: inode_number,
1731 mode: 0o100_644,
1732 mtime_ns,
1733 content: PendingContent::Inline(data),
1734 });
1735 } else {
1736 let pf = PendingFile {
1738 inode_number,
1739 path: path.to_path_buf(),
1740 mtime_ns,
1741 file_len,
1742 };
1743 if let Some(sink) = &self.pending_sink {
1744 sink.send(pf).map_err(|_| {
1750 WriteError::Io(std::io::Error::other("walk: compress pipeline shut down"))
1751 })?;
1752 } else {
1753 self.pending_files.push(pf);
1754 }
1755 }
1756 Ok(inode_number)
1757 } else if file_type.is_symlink() {
1758 let inode_number = self.alloc_inode();
1766 let target = std::fs::read_link(path)?;
1767 let target = target
1768 .to_str()
1769 .ok_or_else(|| WriteError::UnsupportedFileType {
1770 path: path.to_path_buf(),
1771 kind: format!("symlink with non-UTF-8 target ({})", target.display()),
1772 })?
1773 .to_owned();
1774 self.inodes.push(PendingInode {
1775 number: inode_number,
1776 mode: limnifs_core::inode::S_IFLNK | 0o777,
1777 mtime_ns,
1778 content: PendingContent::Symlink(target),
1779 });
1780 Ok(inode_number)
1781 } else {
1782 #[cfg(unix)]
1783 let kind = {
1784 use std::os::unix::fs::FileTypeExt;
1785 if file_type.is_fifo() {
1786 "fifo".to_owned()
1787 } else if file_type.is_socket() {
1788 "socket".to_owned()
1789 } else if file_type.is_block_device() {
1790 "block device".to_owned()
1791 } else if file_type.is_char_device() {
1792 "character device".to_owned()
1793 } else {
1794 "unknown".to_owned()
1795 }
1796 };
1797 #[cfg(not(unix))]
1798 let kind = "unknown".to_owned();
1799 Err(WriteError::UnsupportedFileType {
1800 path: path.to_path_buf(),
1801 kind,
1802 })
1803 }
1804 }
1805
1806 fn train_and_apply_dictionary(&mut self, dictionaries: &crate::config::DictionaryConfig) {
1821 let cleanup = |ctx: &mut Self| {
1822 for d in &mut ctx.drops {
1823 d.plaintext = None;
1824 }
1825 ctx.dict_samples_by_class.clear();
1826 };
1827
1828 if !dictionaries.enabled {
1829 cleanup(self);
1830 return;
1831 }
1832
1833 let target = usize::try_from(dictionaries.max_dict_size).unwrap_or(65_536);
1834 let min_class = usize::try_from(dictionaries.min_class_size).unwrap_or(0);
1835
1836 let text_classes = [
1840 crate::classifier::Class::Text,
1841 crate::classifier::Class::Code,
1842 crate::classifier::Class::Sparse,
1843 ];
1844 let binary_classes = [crate::classifier::Class::Binary];
1845
1846 let text_samples: Vec<&[u8]> = text_classes
1848 .iter()
1849 .flat_map(|c| self.dict_samples_by_class.get(c).into_iter().flatten())
1850 .map(Vec::as_slice)
1851 .collect();
1852 let trainer = crate::dictionary::TrainerKind::from_config_str(&dictionaries.trainer);
1853 if text_samples.len() >= min_class {
1854 if let Some(dict) =
1855 crate::dictionary::train_zstd_with_trainer(0, &text_samples, target, trainer)
1856 {
1857 self.trained_dicts_by_class
1858 .insert(crate::classifier::Class::Text, dict);
1859 }
1860 }
1861 let binary_samples: Vec<&[u8]> = binary_classes
1862 .iter()
1863 .flat_map(|c| self.dict_samples_by_class.get(c).into_iter().flatten())
1864 .map(Vec::as_slice)
1865 .collect();
1866 if binary_samples.len() >= min_class {
1867 if let Some(dict) =
1868 crate::dictionary::train_zstd_with_trainer(1, &binary_samples, target, trainer)
1869 {
1870 self.trained_dicts_by_class
1871 .insert(crate::classifier::Class::Binary, dict);
1872 }
1873 }
1874
1875 use rayon::prelude::*;
1884 let classifier = self.classifier;
1885 self.drops.par_iter_mut().for_each(|d| {
1886 if d.codec != limnifs_core::codec::CODEC_ZSTD {
1887 return;
1888 }
1889 let Some(plaintext) = d.plaintext.as_ref() else {
1890 return;
1891 };
1892 let class = classifier.classify(plaintext);
1893 let dict_class = if text_classes.contains(&class) {
1894 crate::classifier::Class::Text
1895 } else if binary_classes.contains(&class) {
1896 crate::classifier::Class::Binary
1897 } else {
1898 return;
1899 };
1900 let Some(dict) = self.trained_dicts_by_class.get(&dict_class) else {
1901 return;
1902 };
1903 let Ok(dict_compressed) = dict.compress(plaintext) else {
1904 return;
1905 };
1906 if dict_compressed.len() < d.compressed.len() {
1907 d.compressed = dict_compressed.into();
1908 d.dict_id = dict.id;
1909 }
1910 });
1911
1912 cleanup(self);
1913 }
1914
1915 fn trace_phase(label: &str, start: std::time::Instant) {
1917 if std::env::var_os("LIMNIFS_TRACE_ASSEMBLE").is_some() {
1918 eprintln!("[assemble] {label}: {:?}", start.elapsed());
1919 }
1920 }
1921
1922 fn assemble(mut self) -> WriteArtifact {
1923 let t_assemble = std::time::Instant::now();
1924 let inode_count = self.inodes.len();
1925 let dir_count = self.dir_count;
1926 let drop_count = self.drops.len();
1927
1928 let t = std::time::Instant::now();
1934 let slabs = pack_slabs(&self.drops);
1935 Self::trace_phase("pack_slabs", t);
1936
1937 let t = std::time::Instant::now();
1941 if self.emit_shared_inline {
1942 self.build_shared_inline_table();
1943 }
1944 Self::trace_phase("shared_inline_table", t);
1945
1946 let mut metadata_blob = Vec::new();
1947 metadata_blob.extend_from_slice(&u32::try_from(self.inodes.len()).unwrap().to_le_bytes());
1948 for inode in &self.inodes {
1949 self.encode_inode(&mut metadata_blob, inode);
1950 }
1951 metadata_blob
1952 .extend_from_slice(&u32::try_from(self.dir_nodes.len()).unwrap().to_le_bytes());
1953 for node in &self.dir_nodes {
1954 metadata_blob.extend_from_slice(&node.bytes);
1955 }
1956 if !self.shared_inline_table.is_empty() {
1959 metadata_blob.extend_from_slice(
1960 &u32::try_from(self.shared_inline_table.len())
1961 .unwrap()
1962 .to_le_bytes(),
1963 );
1964 for entry in &self.shared_inline_table {
1965 let len = u32::try_from(entry.len()).expect("shared entry fits u32");
1966 metadata_blob.extend_from_slice(&len.to_le_bytes());
1967 metadata_blob.extend_from_slice(entry);
1968 }
1969 }
1970
1971 Self::trace_phase("metadata_encode", t);
1972 let uncompressed_len =
1979 u32::try_from(metadata_blob.len()).expect("metadata blob length fits u32");
1980 let t = std::time::Instant::now();
1981 let metadata_hash = hash_section(&metadata_blob);
1982 let metadata_codec = self.metadata_codec;
1983 let metadata_quality = if metadata_blob.len() > METADATA_LARGE_BLOB_THRESHOLD {
1984 METADATA_LARGE_BLOB_QUALITY
1985 } else {
1986 METADATA_SMALL_BLOB_QUALITY
1987 };
1988 let compressed_blob = if metadata_codec == limnifs_core::codec::CODEC_BROTLI {
1989 limnifs_core::codec::compress_brotli_with_quality(&metadata_blob, metadata_quality)
1990 .unwrap_or_else(|_| metadata_blob.clone())
1991 } else {
1992 limnifs_core::codec::compress(metadata_codec, &metadata_blob)
1993 .unwrap_or_else(|_| metadata_blob.clone())
1994 };
1995 Self::trace_phase("metadata_compress", t);
1996 let (on_wire_codec, on_wire_blob) = if compressed_blob.len() < metadata_blob.len() {
1997 (metadata_codec, compressed_blob)
1998 } else {
1999 (limnifs_core::codec::CODEC_STORE, metadata_blob.clone())
2000 };
2001
2002 let externalize_at = self
2006 .metadata_externalize_threshold
2007 .min(limnifs_core::metadata_reference::DEFAULT_INLINE_METADATA_MAX_BYTES as usize);
2008 let (metadata_sidecar, inline_data, metadata_locator_count) =
2009 if on_wire_blob.len() > externalize_at {
2010 let h = hash_section(&on_wire_blob);
2017 let mut h8 = String::with_capacity(8);
2018 for b in &h[..4] {
2019 h8.push_str(&format!("{b:02x}"));
2020 }
2021 let locator = format!("file:metadata-{h8}.bin");
2022 let sidecar = MetadataSidecar {
2023 bytes: on_wire_blob.clone(),
2024 locator,
2025 };
2026 (Some(sidecar), None, 1u32)
2027 } else {
2028 (None, Some(on_wire_blob.clone()), 0u32)
2029 };
2030
2031 let mut manifest = Vec::new();
2032
2033 let header_start = manifest.len();
2034 manifest.extend_from_slice(&ManifestHeader::current().to_bytes());
2035 let header_end = manifest.len();
2036
2037 let flags_start = manifest.len();
2038 manifest.push(FEATURE_FLAGS_SECTION_VERSION);
2039 manifest.extend_from_slice(&0u32.to_le_bytes());
2040 let flags_end = manifest.len();
2041
2042 let meta_ref_start = manifest.len();
2045 manifest.push(METADATA_REFERENCE_SECTION_VERSION_2);
2046 manifest.extend_from_slice(&metadata_hash);
2047 manifest.extend_from_slice(&uncompressed_len.to_le_bytes());
2048 manifest.push(on_wire_codec);
2049 manifest.extend_from_slice(&metadata_locator_count.to_le_bytes());
2050 if let Some(sidecar) = &metadata_sidecar {
2051 let loc_bytes = sidecar.locator.as_bytes();
2052 let loc_len = u32::try_from(loc_bytes.len()).expect("locator fits u32");
2053 manifest.extend_from_slice(&loc_len.to_le_bytes());
2054 manifest.extend_from_slice(loc_bytes);
2055 }
2056 match &inline_data {
2057 Some(blob) => {
2058 let inline_len = u32::try_from(blob.len()).expect("metadata fits u32");
2059 manifest.extend_from_slice(&inline_len.to_le_bytes());
2060 manifest.extend_from_slice(blob);
2061 }
2062 None => {
2063 manifest.extend_from_slice(&0u32.to_le_bytes());
2064 }
2065 }
2066 let meta_ref_end = manifest.len();
2067
2068 let slab_index_start = manifest.len();
2069 manifest.push(SLAB_INDEX_SECTION_VERSION);
2070 manifest.extend_from_slice(&u32::try_from(slabs.len()).unwrap().to_le_bytes());
2071 for slab in &slabs {
2072 manifest.extend_from_slice(&slab.id.to_bytes());
2073 manifest.extend_from_slice(&1u32.to_le_bytes());
2074 let loc_bytes = slab.locator.as_bytes();
2075 let loc_len = u32::try_from(loc_bytes.len()).expect("locator fits u32");
2076 manifest.extend_from_slice(&loc_len.to_le_bytes());
2077 manifest.extend_from_slice(loc_bytes);
2078 }
2079 let slab_index_end = manifest.len();
2080
2081 let history_start = manifest.len();
2082 manifest.push(HISTORY_SECTION_VERSION);
2083 manifest.extend_from_slice(&1u32.to_le_bytes());
2084 manifest.push(0x01);
2085 manifest.extend_from_slice(&0u64.to_le_bytes());
2086 manifest.extend_from_slice(&0u32.to_le_bytes());
2087 manifest.extend_from_slice(&0u32.to_le_bytes());
2088 let history_end = manifest.len();
2089
2090 let profile_desc_start = manifest.len();
2095 if let Some(ref name) = self.profile_name {
2096 let desc = limnifs_core::profile_descriptor::ProfileDescriptor {
2097 version: limnifs_core::profile_descriptor::PROFILE_DESCRIPTOR_SECTION_VERSION,
2098 profile_name: Some(name.clone()),
2099 blake3_hashing: true,
2100 cross_file_dedup: true,
2101 content_classification: !self.categorizers_disabled,
2102 integrity_verify: true,
2103 read_write: self.rw_mode,
2104 auto_turnover: self.auto_turnover,
2105 };
2106 limnifs_core::profile_descriptor::encode_profile_descriptor(&desc, &mut manifest);
2107 }
2108 let profile_desc_end = manifest.len();
2109
2110 if !self.trained_dicts_by_class.is_empty() {
2116 let dicts: Vec<_> = self
2117 .trained_dicts_by_class
2118 .values()
2119 .map(|d| limnifs_core::dictionary_section::Dictionary {
2120 codec_id: d.codec,
2121 class_id: d.id,
2122 data: d.content.clone(),
2123 })
2124 .collect();
2125 let section = limnifs_core::dictionary_section::DictionarySection {
2126 version: limnifs_core::dictionary_section::DICTIONARY_SECTION_VERSION,
2127 dicts,
2128 };
2129 limnifs_core::dictionary_section::encode_dictionary_section(§ion, &mut manifest);
2130 }
2131
2132 let dictionary_end = manifest.len();
2133
2134 let delta_linkage_hash = if let Some(base_root) = self.base_root {
2140 let delta_start = manifest.len();
2141 manifest.push(limnifs_core::delta_linkage::DELTA_LINKAGE_SECTION_VERSION);
2147 manifest.extend_from_slice(&base_root);
2148 manifest.extend_from_slice(&0u32.to_le_bytes());
2149 hash_section(&manifest[delta_start..])
2150 } else {
2151 hash_empty_section()
2152 };
2153 let _ = dictionary_end;
2154
2155 let hashes = SectionHashes {
2156 metadata: metadata_hash,
2157 format_header: hash_section(&manifest[header_start..header_end]),
2158 feature_flags: hash_section(&manifest[flags_start..flags_end]),
2159 metadata_reference: hash_section(&manifest[meta_ref_start..meta_ref_end]),
2160 slab_index: hash_section(&manifest[slab_index_start..slab_index_end]),
2161 crypto_params: hash_empty_section(),
2162 ec_params: hash_empty_section(),
2163 dms_policy: hash_empty_section(),
2164 delta_linkage: delta_linkage_hash,
2165 history: hash_section(&manifest[history_start..history_end]),
2166 };
2178 let merkle_root = compute_merkle_root(&hashes);
2179
2180 WriteArtifact {
2181 bytes: manifest,
2182 merkle_root,
2183 slabs,
2184 metadata_sidecar,
2185 inode_count,
2186 file_count: self.file_count,
2187 dir_count,
2188 drop_count,
2189 root_inode_number: self.root_inode_number,
2190 }
2191 }
2192
2193 fn encode_inode(&self, out: &mut Vec<u8>, inode: &PendingInode) {
2194 out.extend_from_slice(&inode.number.to_le_bytes());
2195 out.extend_from_slice(&inode.mode.to_le_bytes());
2196 out.extend_from_slice(&0u32.to_le_bytes());
2197 out.extend_from_slice(&0u32.to_le_bytes());
2198 out.extend_from_slice(&inode.mtime_ns.to_le_bytes());
2199 out.extend_from_slice(&inode.mtime_ns.to_le_bytes());
2200 out.extend_from_slice(&1u32.to_le_bytes());
2201 match &inode.content {
2202 PendingContent::Inline(data) => {
2203 let h = hash_section(data);
2204 if let Some(&idx) = self.shared_inline_map.get(&h) {
2205 out.push(INODE_FLAG_INLINE_DATA | INODE_FLAG_SHARED_INLINE);
2207 out.extend_from_slice(&(idx as u32).to_le_bytes());
2208 } else {
2209 out.push(INODE_FLAG_INLINE_DATA);
2210 let len = u32::try_from(data.len()).expect("data fits u32");
2211 out.extend_from_slice(&len.to_le_bytes());
2212 out.extend_from_slice(data);
2213 }
2214 }
2215 PendingContent::DropBacked { file_len, slices } => {
2216 out.push(0x00);
2217 let slice_count = u32::try_from(slices.len()).expect("slice count fits u32");
2218 out.extend_from_slice(&slice_count.to_le_bytes());
2219 for slice in slices {
2220 out.extend_from_slice(&slice.file_byte_start.to_le_bytes());
2221 out.extend_from_slice(&slice.file_byte_end.to_le_bytes());
2222 out.extend_from_slice(&slice.drop_id);
2223 out.extend_from_slice(&0u32.to_le_bytes());
2225 let drop_byte_len = u32::try_from(slice.file_byte_end - slice.file_byte_start)
2229 .expect("slice range fits u32");
2230 out.extend_from_slice(&drop_byte_len.to_le_bytes());
2231 }
2232 let _ = file_len;
2233 }
2234 PendingContent::Symlink(target) => {
2235 out.push(0x00);
2239 let t = target.as_bytes();
2240 let len = u32::try_from(t.len()).expect("target fits u32");
2241 out.extend_from_slice(&len.to_le_bytes());
2242 out.extend_from_slice(t);
2243 }
2244 PendingContent::Directory(entries) => {
2245 out.push(0x00);
2246 let node = self
2247 .dir_nodes
2248 .iter()
2249 .find(|n| n.entries == *entries)
2250 .expect("directory node must exist");
2251 out.extend_from_slice(&node.hash);
2252 }
2253 }
2254 }
2255}
2256
2257fn sidecar_name(locator: &str) -> Result<&str, WriteError> {
2263 limnifs_core::locator::local_sidecar_name(locator)
2264 .map_err(|e| WriteError::Io(std::io::Error::other(format!("{e}"))))
2265}
2266
2267fn encode_dir_node(entries: &[(String, u64, u8)]) -> DirNode {
2268 let mut bytes = Vec::new();
2269 bytes.push(1u8);
2270 let count = u32::try_from(entries.len()).expect("entry count fits u32");
2271 bytes.extend_from_slice(&count.to_le_bytes());
2272 for (name, inode_number, entry_type) in entries {
2273 let name_bytes = name.as_bytes();
2274 let name_len = u32::try_from(name_bytes.len()).expect("name fits u32");
2275 bytes.extend_from_slice(&name_len.to_le_bytes());
2276 bytes.extend_from_slice(name_bytes);
2277 bytes.extend_from_slice(&inode_number.to_le_bytes());
2278 bytes.push(*entry_type);
2279 }
2280 let hash = hash_section(&bytes);
2281 DirNode {
2282 entries: entries.to_vec(),
2283 bytes,
2284 hash,
2285 }
2286}
2287
2288fn pack_slabs(drops: &[PendingDrop]) -> Vec<SlabArtifact> {
2297 let local_drops: Vec<&PendingDrop> = drops
2302 .iter()
2303 .filter(|d| d.codec != limnifs_core::codec::CODEC_REFERENCED)
2304 .collect();
2305 if local_drops.is_empty() {
2306 return Vec::new();
2307 }
2308
2309 let max_content = MAX_SLAB_TOTAL_BYTES.saturating_sub(SLAB_HEADER_LEN);
2310
2311 let mut slab_groups: Vec<Vec<&PendingDrop>> = Vec::new();
2316 let mut current: Vec<&PendingDrop> = Vec::new();
2317 let mut current_size: usize = 0;
2318
2319 for drop in &local_drops {
2320 let footprint = drop.slab_footprint();
2321 if !current.is_empty() && current_size + footprint > max_content {
2322 slab_groups.push(std::mem::take(&mut current));
2323 current_size = 0;
2324 }
2325 current.push(*drop);
2326 current_size += footprint;
2327 }
2328 if !current.is_empty() {
2329 slab_groups.push(current);
2330 }
2331
2332 use rayon::prelude::*;
2338 slab_groups
2339 .par_iter()
2340 .enumerate()
2341 .map(|(ordinal, group)| {
2342 let ordinal_u64 = u64::try_from(ordinal).expect("slab count fits u64");
2343 encode_slab(ordinal_u64, group)
2344 })
2345 .collect()
2346}
2347
2348fn encode_slab(ordinal: u64, drops: &[&PendingDrop]) -> SlabArtifact {
2352 const DROP_RECORD_LEN: usize = 50;
2359 let mut drop_records = Vec::with_capacity(drops.len() * DROP_RECORD_LEN);
2360 let mut solid_window = Vec::new();
2361 let mut drop_ids = Vec::with_capacity(drops.len());
2362 let mut offset_in_window: u32 = 0;
2363
2364 for drop in drops {
2365 let plaintext_len = drop.plaintext_len_value();
2366 let window_len = drop.len_in_window();
2367 drop_records.extend_from_slice(&drop.id);
2368 drop_records.extend_from_slice(&plaintext_len.to_le_bytes());
2369 drop_records.extend_from_slice(&[drop.codec, 0x00, 0x00]);
2371 drop_records.push(0x00); drop_records.extend_from_slice(&offset_in_window.to_le_bytes());
2373 drop_records.extend_from_slice(&window_len.to_le_bytes());
2374 drop_records.push(drop.dict_id); drop_records.push(drop.flags); solid_window.extend_from_slice(&drop.compressed);
2377 drop_ids.push(drop.id);
2378 offset_in_window = offset_in_window
2379 .checked_add(window_len)
2380 .expect("slab window size fits u32");
2381 }
2382
2383 let slab_content = [&drop_records[..], &solid_window[..]].concat();
2384 let slab_hash = hash_section(&slab_content);
2385 let slab_id = SlabId::new(ordinal, slab_hash);
2386
2387 let total_length = SLAB_HEADER_LEN + slab_content.len();
2388 let mut slab_bytes = Vec::with_capacity(total_length);
2389 slab_bytes.extend_from_slice(b"LIM1");
2390 slab_bytes.extend_from_slice(&1u16.to_le_bytes()); slab_bytes.extend_from_slice(&slab_id.to_bytes());
2392 slab_bytes.extend_from_slice(
2393 &u64::try_from(total_length)
2394 .unwrap_or(u64::MAX)
2395 .to_le_bytes(),
2396 );
2397 slab_bytes.push(0x00);
2398 slab_bytes.push(0x00);
2399 slab_bytes.extend_from_slice(&slab_content);
2400
2401 let mut h8 = String::with_capacity(8);
2405 for b in &slab_id.hash[..4] {
2406 h8.push_str(&format!("{b:02x}"));
2407 }
2408 let locator = format!("file:slab-{ordinal}-{h8}.bin");
2409
2410 SlabArtifact {
2411 id: slab_id,
2412 bytes: slab_bytes,
2413 locator,
2414 drop_ids,
2415 }
2416}
2417
2418#[cfg(test)]
2419fn pseudo_random_bytes(seed: u64, count: usize) -> Vec<u8> {
2420 let mut state = seed;
2421 let mut out = Vec::with_capacity(count);
2422 for _ in 0..count {
2423 state = state
2424 .wrapping_mul(6_364_136_223_846_793_005)
2425 .wrapping_add(1_442_695_040_888_963_407);
2426 out.push(u8::try_from(state >> 56).expect("fits u8"));
2427 }
2428 out
2429}
2430
2431#[cfg(test)]
2432mod tests {
2433 use super::*;
2434 use limnifs_core::ManifestCursor;
2435
2436 #[test]
2437 fn write_stream_packs_single_named_stream() {
2438 let temp = std::env::temp_dir().join(format!(
2442 "limnifs-write-stream-test-{}-{}",
2443 std::process::id(),
2444 std::time::SystemTime::now()
2445 .duration_since(std::time::UNIX_EPOCH)
2446 .unwrap()
2447 .as_nanos()
2448 ));
2449 std::fs::create_dir_all(&temp).expect("create temp dir");
2450
2451 let content = b"stream test content line\n".repeat(10_000); let cursor = std::io::Cursor::new(content.clone());
2453 let config = WriteConfig::default_v0_1();
2454 let artifact = write_stream("streamed.txt", cursor, &config).expect("write_stream");
2455
2456 assert!(!artifact.bytes.is_empty(), "manifest bytes non-empty");
2457 assert!(!artifact.slabs.is_empty(), "at least one slab produced");
2458 let total_drop_bytes: usize = artifact.slabs.iter().map(|s| s.bytes.len()).sum();
2463 assert!(total_drop_bytes > 0, "drops non-empty");
2464
2465 let _ = std::fs::remove_dir_all(&temp);
2466 }
2467
2468 #[test]
2469 fn write_layer_references_base_drops() {
2470 let temp = std::env::temp_dir().join(format!(
2476 "limnifs-write-layer-test-{}-{}",
2477 std::process::id(),
2478 std::time::SystemTime::now()
2479 .duration_since(std::time::UNIX_EPOCH)
2480 .unwrap()
2481 .as_nanos()
2482 ));
2483 std::fs::create_dir_all(&temp).expect("create temp dir");
2484
2485 let base_dir = temp.join("base");
2487 std::fs::create_dir_all(&base_dir).expect("base dir");
2488 let text = b"layer test content line\n".repeat(50_000); std::fs::write(base_dir.join("shared.txt"), &text).expect("write shared");
2490 std::fs::write(base_dir.join("base-only.txt"), b"only in base").expect("write base-only");
2491
2492 let config = WriteConfig::default_v0_1();
2493 let base_artifact = write_directory_with_config(&base_dir, &config).expect("base");
2494
2495 let base_manifest = temp.join("base.lim");
2496 std::fs::write(&base_manifest, &base_artifact.bytes).expect("write base manifest");
2497 for slab in &base_artifact.slabs {
2498 let slab_name = sidecar_name(&slab.locator).expect("slab locator");
2499 std::fs::write(temp.join(slab_name), &slab.bytes).expect("write base slab");
2500 }
2501
2502 let layer_dir = temp.join("layer");
2504 std::fs::create_dir_all(&layer_dir).expect("layer dir");
2505 std::fs::write(layer_dir.join("shared.txt"), &text).expect("write shared in layer");
2506 std::fs::write(layer_dir.join("new.txt"), b"fresh content in layer").expect("write new");
2507
2508 let layer_artifact = write_layer(&base_manifest, &layer_dir, &config).expect("layer");
2509
2510 let layer_slab_bytes: usize = layer_artifact.slabs.iter().map(|s| s.bytes.len()).sum();
2513 let base_slab_bytes: usize = base_artifact.slabs.iter().map(|s| s.bytes.len()).sum();
2514 assert!(
2515 layer_slab_bytes < base_slab_bytes / 4,
2516 "layer slabs ({}) should be much smaller than base ({}) — layering failed",
2517 layer_slab_bytes,
2518 base_slab_bytes
2519 );
2520
2521 let base_root = base_artifact.merkle_root.as_bytes();
2524 assert!(
2525 layer_artifact
2526 .bytes
2527 .windows(32)
2528 .any(|w| w == base_root.as_slice()),
2529 "layer manifest must contain base's ManifestRoot bytes"
2530 );
2531
2532 let _ = std::fs::remove_dir_all(&temp);
2533 }
2534
2535 #[test]
2536 fn tournament_short_circuits_on_highly_compressible_chunk() {
2537 let chunk = b"hello world ".repeat(500);
2540 let tunables = limnifs_core::codec::CodecTunables::default();
2541 let tournament = TournamentSpec {
2542 codec_ids: vec![
2543 limnifs_core::codec::CODEC_LZ4,
2544 limnifs_core::codec::CODEC_BROTLI,
2545 ],
2546 min_size: 16,
2547 skip_for_binary: false,
2548 short_circuit_permille: 250,
2549 };
2550 let (codec_id, compressed) = compress_chunk_with_tournament(
2551 &chunk,
2552 classifier::Class::Text,
2553 limnifs_core::codec::CODEC_BROTLI,
2554 limnifs_core::codec::CODEC_LZ4,
2555 &tunables,
2556 &tournament,
2557 );
2558 assert_eq!(codec_id, limnifs_core::codec::CODEC_LZ4);
2559 assert!(compressed.len() < chunk.len());
2560 }
2561
2562 #[test]
2563 fn tournament_runs_all_codecs_when_short_circuit_disabled() {
2564 let chunk = b"hello world ".repeat(500);
2570 let tunables = limnifs_core::codec::CodecTunables::default();
2571 let tournament = TournamentSpec {
2572 codec_ids: vec![
2573 limnifs_core::codec::CODEC_LZ4,
2574 limnifs_core::codec::CODEC_BROTLI,
2575 limnifs_core::codec::CODEC_ZSTD,
2576 ],
2577 min_size: 16,
2578 skip_for_binary: false,
2579 short_circuit_permille: 0,
2580 };
2581 let (codec_id, compressed) = compress_chunk_with_tournament(
2582 &chunk,
2583 classifier::Class::Text,
2584 limnifs_core::codec::CODEC_BROTLI,
2585 limnifs_core::codec::CODEC_LZ4,
2586 &tunables,
2587 &tournament,
2588 );
2589 assert!(
2593 codec_id == limnifs_core::codec::CODEC_ZSTD
2594 || codec_id == limnifs_core::codec::CODEC_BROTLI,
2595 "expected ZSTD or Brotli to win, got codec {codec_id}"
2596 );
2597 assert!(compressed.len() < chunk.len());
2598 }
2599
2600 #[test]
2601 fn tournament_skips_for_binary_when_configured() {
2602 let chunk = vec![0u8; 4096];
2603 let tunables = limnifs_core::codec::CodecTunables::default();
2604 let tournament = TournamentSpec {
2605 codec_ids: vec![limnifs_core::codec::CODEC_BROTLI],
2606 min_size: 16,
2607 skip_for_binary: true,
2608 short_circuit_permille: 250,
2609 };
2610 let (codec_id, _compressed) = compress_chunk_with_tournament(
2611 &chunk,
2612 classifier::Class::Binary,
2613 limnifs_core::codec::CODEC_BROTLI,
2614 limnifs_core::codec::CODEC_LZ4,
2615 &tunables,
2616 &tournament,
2617 );
2618 assert_eq!(codec_id, limnifs_core::codec::CODEC_LZ4);
2620 }
2621
2622 #[test]
2623 fn tournament_small_chunk_uses_preferred_codec() {
2624 let chunk = b"tiny";
2625 let tunables = limnifs_core::codec::CodecTunables::default();
2626 let tournament = TournamentSpec {
2627 codec_ids: vec![limnifs_core::codec::CODEC_BROTLI],
2628 min_size: 1024,
2629 skip_for_binary: false,
2630 short_circuit_permille: 0,
2631 };
2632 let (codec_id, _compressed) = compress_chunk_with_tournament(
2633 chunk,
2634 classifier::Class::Text,
2635 limnifs_core::codec::CODEC_BROTLI,
2636 limnifs_core::codec::CODEC_LZ4,
2637 &tunables,
2638 &tournament,
2639 );
2640 assert_eq!(codec_id, limnifs_core::codec::CODEC_STORE);
2642 }
2643
2644 #[test]
2645 fn tournament_falls_back_to_store_when_no_codec_compresses() {
2646 let chunk = pseudo_random_bytes(42, 4096);
2650 let tunables = limnifs_core::codec::CodecTunables::default();
2651 let tournament = TournamentSpec {
2652 codec_ids: vec![limnifs_core::codec::CODEC_LZ4],
2653 min_size: 16,
2654 skip_for_binary: false,
2655 short_circuit_permille: 0,
2656 };
2657 let (codec_id, compressed) = compress_chunk_with_tournament(
2658 &chunk,
2659 classifier::Class::Binary,
2660 limnifs_core::codec::CODEC_BROTLI,
2661 limnifs_core::codec::CODEC_LZ4,
2662 &tunables,
2663 &tournament,
2664 );
2665 assert_eq!(codec_id, limnifs_core::codec::CODEC_STORE);
2666 assert_eq!(compressed.len(), chunk.len());
2667 }
2668
2669 #[test]
2670 fn dictionaries_enabled_emits_dictionary_section_when_enough_samples() {
2671 let temp = std::env::temp_dir().join(format!(
2676 "limnifs-write-test-{}-dict-{}",
2677 std::process::id(),
2678 std::time::SystemTime::now()
2679 .duration_since(std::time::UNIX_EPOCH)
2680 .map(|d| d.as_nanos() as u64)
2681 .unwrap_or(0),
2682 ));
2683 let _ = std::fs::remove_dir_all(&temp);
2684 std::fs::create_dir_all(&temp).expect("mkdir");
2685
2686 for i in 0..200 {
2689 let content = format!(
2691 "function test_case_{i}() {{ return constant + {i}; }}\n\
2692 // shared comment line {i}\n\
2693 struct Foo {{ x: i32 }} // type {i}\n"
2694 )
2695 .repeat(5);
2696 let path = temp.join(format!("file_{i:04}.txt"));
2697 std::fs::write(&path, content.as_bytes()).expect("write");
2698 }
2699
2700 let mut config = crate::profile::balanced();
2701 config.defaults.text_codec = "zstd".into();
2703 config.defaults.metadata_codec = "zstd".into();
2707 config.dictionaries.enabled = true;
2708 config.dictionaries.min_class_size = 50;
2709 config.dictionaries.max_dict_size = 8192;
2710
2711 let artifact = write_directory_with_config(&temp, &config).expect("write");
2712 std::fs::remove_dir_all(&temp).ok();
2713
2714 let mut cursor = ManifestCursor::new(&artifact.bytes);
2719 let _ = limnifs_core::parse_manifest_header(&mut cursor).expect("header");
2720 let _ = limnifs_core::parse_feature_flags_section(&mut cursor).expect("flags");
2721 let _ = limnifs_core::parse_metadata_reference(&mut cursor).expect("meta_ref");
2722 let _ = limnifs_core::parse_slab_index(&mut cursor).expect("slab_index");
2723 let _ = limnifs_core::parse_history(&mut cursor).expect("history");
2724 let _remaining = cursor.remaining_len();
2727 }
2728
2729 #[test]
2730 fn write_empty_directory() {
2731 let temp =
2732 std::env::temp_dir().join(format!("limnifs-write-test-{}-empty", std::process::id()));
2733 std::fs::create_dir_all(&temp).expect("create temp dir");
2734 let artifact = write_directory(&temp).expect("write succeeds");
2735 std::fs::remove_dir_all(&temp).ok();
2736 assert!(artifact.inode_count >= 1);
2737 assert_eq!(artifact.file_count, 0);
2738 assert_eq!(artifact.dir_count, 1);
2739 assert!(artifact.slabs.is_empty());
2740 }
2741
2742 #[test]
2743 fn write_small_file_inline() {
2744 let temp =
2745 std::env::temp_dir().join(format!("limnifs-write-test-{}-small", std::process::id()));
2746 std::fs::create_dir_all(&temp).expect("create temp dir");
2747 std::fs::write(temp.join("hello.txt"), b"hello world").expect("write file");
2748 let artifact = write_directory(&temp).expect("write succeeds");
2749 std::fs::remove_dir_all(&temp).ok();
2750 assert_eq!(artifact.file_count, 1);
2751 assert!(artifact.slabs.is_empty());
2752 assert_eq!(artifact.drop_count, 0);
2753 }
2754
2755 #[test]
2756 fn write_large_file_uses_slab() {
2757 let temp =
2758 std::env::temp_dir().join(format!("limnifs-write-test-{}-large", std::process::id()));
2759 std::fs::create_dir_all(&temp).expect("create temp dir");
2760 let large_data = vec![0xABu8; INLINE_THRESHOLD + 100];
2761 std::fs::write(temp.join("big.bin"), &large_data).expect("write big");
2762 let artifact = write_directory(&temp).expect("write succeeds");
2763 std::fs::remove_dir_all(&temp).ok();
2764 assert_eq!(artifact.drop_count, 1);
2765 assert_eq!(artifact.slabs.len(), 1);
2766 }
2767
2768 #[test]
2769 fn write_mixed_inline_and_large() {
2770 let temp =
2771 std::env::temp_dir().join(format!("limnifs-write-test-{}-mix", std::process::id()));
2772 std::fs::create_dir_all(&temp).expect("create temp dir");
2773 std::fs::write(temp.join("small.txt"), b"tiny").expect("write small");
2774 std::fs::write(temp.join("large.bin"), vec![0xCDu8; INLINE_THRESHOLD * 2])
2775 .expect("write large");
2776 let artifact = write_directory(&temp).expect("write succeeds");
2777 std::fs::remove_dir_all(&temp).ok();
2778 assert_eq!(artifact.file_count, 2);
2779 assert_eq!(artifact.drop_count, 1);
2780 assert_eq!(artifact.slabs.len(), 1);
2781 }
2782
2783 #[test]
2784 fn deduplicates_identical_large_files() {
2785 let temp =
2786 std::env::temp_dir().join(format!("limnifs-write-test-{}-dedup", std::process::id()));
2787 std::fs::create_dir_all(&temp).expect("create temp dir");
2788 let data = vec![0x77u8; INLINE_THRESHOLD + 10];
2789 std::fs::write(temp.join("a.bin"), &data).expect("write a");
2790 std::fs::write(temp.join("b.bin"), &data).expect("write b");
2791 let artifact = write_directory(&temp).expect("write succeeds");
2792 std::fs::remove_dir_all(&temp).ok();
2793 assert_eq!(artifact.drop_count, 1);
2794 }
2795
2796 #[test]
2797 fn write_and_verify_roundtrip() {
2798 let temp = std::env::temp_dir().join(format!(
2799 "limnifs-write-test-{}-roundtrip",
2800 std::process::id()
2801 ));
2802 std::fs::create_dir_all(&temp).expect("create temp dir");
2803 std::fs::write(temp.join("a.txt"), b"aaa").expect("write a");
2804 std::fs::write(temp.join("b.txt"), b"bbb").expect("write b");
2805 std::fs::create_dir_all(temp.join("sub")).expect("create sub");
2806 std::fs::write(temp.join("sub").join("c.txt"), b"ccc").expect("write c");
2807 let artifact = write_directory(&temp).expect("write succeeds");
2808 std::fs::remove_dir_all(&temp).ok();
2809 assert_eq!(artifact.file_count, 3);
2810 assert_eq!(artifact.dir_count, 2);
2811
2812 let mut cursor = ManifestCursor::new(&artifact.bytes);
2813 limnifs_core::parse_manifest_header(&mut cursor).expect("header");
2814 limnifs_core::parse_feature_flags_section(&mut cursor).expect("flags");
2815 let meta_ref = limnifs_core::parse_metadata_reference(&mut cursor).expect("meta ref");
2816 assert!(meta_ref.is_inlined());
2817 let slab_index = limnifs_core::parse_slab_index(&mut cursor).expect("slab index");
2818 assert_eq!(slab_index.len(), 0);
2819 limnifs_core::parse_history(&mut cursor).expect("history");
2820 }
2821
2822 #[test]
2823 fn write_deterministic() {
2824 let temp =
2825 std::env::temp_dir().join(format!("limnifs-write-test-{}-det", std::process::id()));
2826 std::fs::create_dir_all(&temp).expect("create temp dir");
2827 std::fs::write(temp.join("x.txt"), b"xxx").expect("write x");
2828
2829 let a1 = write_directory(&temp).expect("first write");
2830 let a2 = write_directory(&temp).expect("second write");
2831 std::fs::remove_dir_all(&temp).ok();
2832
2833 assert_eq!(a1.bytes, a2.bytes);
2834 assert_eq!(a1.merkle_root, a2.merkle_root);
2835 }
2836
2837 #[test]
2838 fn slab_parses_correctly() {
2839 let temp =
2840 std::env::temp_dir().join(format!("limnifs-write-test-{}-slab", std::process::id()));
2841 std::fs::create_dir_all(&temp).expect("create temp dir");
2842 std::fs::write(temp.join("big.bin"), vec![0x11u8; INLINE_THRESHOLD + 1])
2843 .expect("write big");
2844 let artifact = write_directory(&temp).expect("write succeeds");
2845 std::fs::remove_dir_all(&temp).ok();
2846
2847 let slab_bytes = &artifact.slabs[0].bytes;
2848 let mut cursor = ManifestCursor::new(slab_bytes);
2849 let slab_header = limnifs_core::parse_slab_header(&mut cursor).expect("slab header parses");
2850 assert_eq!(
2851 slab_header.format_version,
2852 limnifs_core::slab::SLAB_FORMAT_VERSION
2853 );
2854 assert!(!slab_header.is_sealed());
2855 assert!(!slab_header.has_erasure_coding());
2856
2857 let drop_record =
2858 limnifs_core::parse_drop_record(&mut cursor, &slab_header).expect("drop record parses");
2859 assert_eq!(drop_record.plaintext_len as usize, INLINE_THRESHOLD + 1);
2860 }
2861
2862 #[test]
2863 fn fastcdc_produces_multiple_chunks_for_large_files() {
2864 let temp = std::env::temp_dir().join(format!(
2867 "limnifs-write-test-{}-cdc-multi",
2868 std::process::id()
2869 ));
2870 std::fs::create_dir_all(&temp).expect("create temp dir");
2871 let data = pseudo_random_bytes(42, 1024 * 1024);
2872 std::fs::write(temp.join("big.bin"), &data).expect("write big");
2873 let artifact = write_directory(&temp).expect("write succeeds");
2874 std::fs::remove_dir_all(&temp).ok();
2875 assert!(
2876 artifact.drop_count > 1,
2877 "expected FastCDC to produce multiple drops for 1 MiB input, got {}",
2878 artifact.drop_count
2879 );
2880 }
2881
2882 #[test]
2883 fn fastcdc_deduplicates_shared_substrings() {
2884 let temp = std::env::temp_dir().join(format!(
2888 "limnifs-write-test-{}-cdc-dedup",
2889 std::process::id()
2890 ));
2891 std::fs::create_dir_all(&temp).expect("create temp dir");
2892 let shared = pseudo_random_bytes(7, 512 * 1024);
2893 let mut a = Vec::with_capacity(shared.len() + 1024);
2894 a.extend_from_slice(&pseudo_random_bytes(1, 1024));
2895 a.extend_from_slice(&shared);
2896 let mut b = Vec::with_capacity(shared.len() + 2048);
2897 b.extend_from_slice(&pseudo_random_bytes(2, 2048));
2898 b.extend_from_slice(&shared);
2899 std::fs::write(temp.join("a.bin"), &a).expect("write a");
2900 std::fs::write(temp.join("b.bin"), &b).expect("write b");
2901
2902 let temp_a = std::env::temp_dir().join(format!(
2904 "limnifs-write-test-{}-cdc-dedup-a",
2905 std::process::id()
2906 ));
2907 std::fs::create_dir_all(&temp_a).expect("create temp_a");
2908 std::fs::write(temp_a.join("a.bin"), &a).expect("write a");
2909 let artifact_a = write_directory(&temp_a).expect("a writes");
2910 std::fs::remove_dir_all(&temp_a).ok();
2911
2912 let temp_b = std::env::temp_dir().join(format!(
2913 "limnifs-write-test-{}-cdc-dedup-b",
2914 std::process::id()
2915 ));
2916 std::fs::create_dir_all(&temp_b).expect("create temp_b");
2917 std::fs::write(temp_b.join("b.bin"), &b).expect("write b");
2918 let artifact_b = write_directory(&temp_b).expect("b writes");
2919 std::fs::remove_dir_all(&temp_b).ok();
2920
2921 let artifact_both = write_directory(&temp).expect("both write");
2922 std::fs::remove_dir_all(&temp).ok();
2923
2924 let sum_alone = artifact_a.drop_count + artifact_b.drop_count;
2925 assert!(
2926 artifact_both.drop_count < sum_alone,
2927 "expected dedup win: both together = {} drops, sum alone = {} drops",
2928 artifact_both.drop_count,
2929 sum_alone
2930 );
2931 }
2932
2933 #[test]
2934 fn slab_splits_when_content_exceeds_ceiling() {
2935 let temp =
2941 std::env::temp_dir().join(format!("limnifs-write-test-{}-split", std::process::id()));
2942 std::fs::create_dir_all(&temp).expect("create temp dir");
2943 for i in 0..7u32 {
2944 let data = pseudo_random_bytes(u64::from(i), 10 * 1024 * 1024);
2946 std::fs::write(temp.join(format!("big-{i}.bin")), &data).expect("write big");
2947 }
2948 let artifact = write_directory(&temp).expect("write succeeds");
2949 std::fs::remove_dir_all(&temp).ok();
2950
2951 assert!(
2953 artifact.slabs.len() >= 2,
2954 "expected at least 2 slabs for 70 MiB of incompressible data, got {}",
2955 artifact.slabs.len()
2956 );
2957 for slab in &artifact.slabs {
2958 assert!(
2959 slab.bytes.len() <= MAX_SLAB_TOTAL_BYTES,
2960 "slab {} is {} bytes (> {} ceiling)",
2961 slab.id.ordinal,
2962 slab.bytes.len(),
2963 MAX_SLAB_TOTAL_BYTES,
2964 );
2965 }
2966 let total_drop_ids: usize = artifact.slabs.iter().map(|s| s.drop_ids.len()).sum();
2968 assert_eq!(
2969 total_drop_ids, artifact.drop_count,
2970 "drop_ids count across slabs must match WriteArtifact.drop_count",
2971 );
2972 }
2973}