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_drop_index, base_root) = load_base_drop_index(base_image)?;
393
394 let mut ctx = WriteContext::new();
395 ctx.chunker = chunker_from_config(config)?;
396 ctx.categorizers_disabled = config.categorizers.is_empty();
397 ctx.rw_mode = matches!(config.mode, crate::config::ImageMode::ReadWrite(_));
398 ctx.auto_turnover = config.turnover_threshold > 0;
399 ctx.collect_dict_samples = config.dictionaries.enabled;
400 ctx.inline_threshold = config.defaults.inline_threshold as usize;
401 ctx.metadata_externalize_threshold = config.defaults.metadata_externalize_threshold;
402 ctx.emit_shared_inline = config.defaults.shared_inline;
403 ctx.base_drop_index = Some(base_drop_index);
404 ctx.base_root = Some(base_root);
405
406 let root_inode_number = ctx.walk(root)?;
408 ctx.root_inode_number = root_inode_number;
409 write_directory_body(&mut ctx, config)?;
410 Ok(ctx.assemble())
411}
412
413fn load_base_drop_index(
417 base_image: &Path,
418) -> Result<(std::collections::HashSet<[u8; 32]>, [u8; 32]), WriteError> {
419 let manifest_bytes = std::fs::read(base_image)?;
420 let mut cursor = ManifestCursor::new(&manifest_bytes);
421 let _ = parse_manifest_header(&mut cursor).map_err(io_core)?;
422 let _ = limnifs_core::parse_feature_flags_section(&mut cursor);
424 let _ = limnifs_core::parse_metadata_reference(&mut cursor);
425 let slab_index = parse_slab_index(&mut cursor).map_err(io_core)?;
426 let store = SlabStore::load_mmap(base_image, &slab_index).map_err(io_core)?;
427 let drop_set: std::collections::HashSet<[u8; 32]> = store.drop_index_keys().copied().collect();
428 let root = *compute_merkle_root_from_sections(&manifest_bytes).as_bytes();
434 Ok((drop_set, root))
435}
436
437fn compute_merkle_root_from_sections(manifest: &[u8]) -> ManifestRoot {
443 use limnifs_core::SectionHashes;
444 let mut cursor = ManifestCursor::new(manifest);
445 let header_start = 0;
446 if parse_manifest_header(&mut cursor).is_err() {
447 return ManifestRoot::from_bytes([0u8; 32]);
449 }
450 let header_end = cursor.position();
451 let flags_start = header_end;
453 let flags_end = match limnifs_core::parse_feature_flags_section(&mut cursor) {
454 Ok(_) => cursor.position(),
455 Err(_) => flags_start,
456 };
457 let meta_ref_start = flags_end;
458 let metadata_reference = match limnifs_core::parse_metadata_reference(&mut cursor) {
459 Ok(m) => Some(m),
460 Err(_) => None,
461 };
462 let meta_ref_end = cursor.position();
463 let slab_index_start = meta_ref_end;
464 let _ = parse_slab_index(&mut cursor);
465 let slab_index_end = cursor.position();
466 let history_start = slab_index_end;
467 let _ = limnifs_core::parse_history(&mut cursor);
468 let history_end = cursor.position();
469
470 let hashes = SectionHashes {
471 metadata: metadata_reference
472 .map(|m| m.metadata_hash)
473 .unwrap_or_else(hash_empty_section),
474 format_header: hash_section(&manifest[header_start..header_end]),
475 feature_flags: hash_section(&manifest[flags_start..flags_end]),
476 metadata_reference: hash_section(&manifest[meta_ref_start..meta_ref_end]),
477 slab_index: hash_section(&manifest[slab_index_start..slab_index_end]),
478 crypto_params: hash_empty_section(),
479 ec_params: hash_empty_section(),
480 dms_policy: hash_empty_section(),
481 delta_linkage: hash_empty_section(),
482 history: hash_section(&manifest[history_start..history_end]),
483 };
484 compute_merkle_root(&hashes)
485}
486
487fn io_core(e: limnifs_core::CoreError) -> WriteError {
488 WriteError::Io(std::io::Error::other(format!("base image load: {e}")))
489}
490
491fn write_directory_body(ctx: &mut WriteContext, config: &WriteConfig) -> Result<(), WriteError> {
496 use rayon::prelude::*;
497
498 ctx.metadata_codec = config
499 .metadata_codec_id()
500 .unwrap_or(limnifs_core::codec::CODEC_BROTLI);
501
502 ctx.chunker = chunker_from_config(config)?;
503
504 let pending = std::mem::take(&mut ctx.pending_files);
505 if pending.is_empty() {
506 return Ok(());
507 }
508 ctx.inline_threshold = config.defaults.inline_threshold as usize;
509 ctx.metadata_externalize_threshold = config.defaults.metadata_externalize_threshold;
510 ctx.emit_shared_inline = config.defaults.shared_inline;
511 let chunker = ctx.chunker.clone();
512 let classifier = ctx.classifier;
513 let text_codec = config.text_codec_id().unwrap_or(0x04);
514 let binary_codec = config.binary_codec_id().unwrap_or(0x01);
515 let tunables = config.to_core_tunables();
516 let use_categorizers = !config.categorizers.is_empty();
517 let skip_chunking = config.skip_chunking;
518 let registry = config
519 .codec_registry()
520 .map_err(|e| WriteError::Io(std::io::Error::other(format!("codec registry: {e}"))))?;
521 let tournament_codec_ids: Vec<u8> = config
522 .tournament
523 .codecs
524 .iter()
525 .filter_map(|n| registry.lookup_by_name(n))
526 .collect();
527 let tournament_spec = TournamentSpec {
528 codec_ids: tournament_codec_ids,
529 min_size: config.tournament.min_size_threshold as usize,
530 skip_for_binary: config.tournament.skip_for_binary,
531 short_circuit_permille: config.tournament.short_circuit_threshold,
532 };
533 let base_drop_index = ctx.base_drop_index.as_ref();
534 let inline_threshold = ctx.inline_threshold;
535 let max_drop_size = config.defaults.max_drop_size as usize;
536 let seekable_drops = config.defaults.seekable_drops;
537 let seekable_drops = config.defaults.seekable_drops;
538 let results: Vec<ChunkedFileResult> = pending
539 .par_iter()
540 .map(|pf| {
541 process_file(
542 pf,
543 &chunker,
544 classifier,
545 text_codec,
546 binary_codec,
547 &tunables,
548 use_categorizers,
549 skip_chunking,
550 &tournament_spec,
551 base_drop_index,
552 inline_threshold,
553 max_drop_size,
554 seekable_drops,
555 config.categorizers.as_slice(),
556 &|name| {
557 config
558 .codec_registry()
559 .ok()
560 .and_then(|r| r.lookup_by_name(name))
561 },
562 )
563 })
564 .collect::<Result<Vec<_>, _>>()?;
565
566 for (pf, result) in pending.iter().zip(results) {
567 ctx.merge_chunked_file(pf, result);
568 }
569 ctx.train_and_apply_dictionary(&config.dictionaries);
570 Ok(())
571}
572
573pub fn write_directory_with_config(
575 root: &Path,
576 config: &WriteConfig,
577) -> Result<WriteArtifact, WriteError> {
578 let mut ctx = WriteContext::new();
579 ctx.chunker = chunker_from_config(config)?;
580 ctx.categorizers_disabled = config.categorizers.is_empty();
581 ctx.rw_mode = matches!(config.mode, crate::config::ImageMode::ReadWrite(_));
582 ctx.auto_turnover = config.turnover_threshold > 0;
583 ctx.collect_dict_samples = config.dictionaries.enabled;
584
585 write_directory_streaming(&mut ctx, root, config)?;
586 Ok(ctx.assemble())
587}
588
589fn write_directory_streaming(
603 ctx: &mut WriteContext,
604 root: &Path,
605 config: &WriteConfig,
606) -> Result<(), WriteError> {
607 use rayon::prelude::*;
608
609 ctx.metadata_codec = config
610 .metadata_codec_id()
611 .unwrap_or(limnifs_core::codec::CODEC_BROTLI);
612
613 ctx.chunker = chunker_from_config(config)?;
614
615 let chunker = ctx.chunker.clone();
616 let classifier = ctx.classifier;
617 let text_codec = config.text_codec_id().unwrap_or(0x04);
618 let binary_codec = config.binary_codec_id().unwrap_or(0x01);
619 let tunables = config.to_core_tunables();
620 let use_categorizers = !config.categorizers.is_empty();
621 let skip_chunking = config.skip_chunking;
622 let registry = config
623 .codec_registry()
624 .map_err(|e| WriteError::Io(std::io::Error::other(format!("codec registry: {e}"))))?;
625 let tournament_codec_ids: Vec<u8> = config
626 .tournament
627 .codecs
628 .iter()
629 .filter_map(|n| registry.lookup_by_name(n))
630 .collect();
631 let tournament_spec = TournamentSpec {
632 codec_ids: tournament_codec_ids,
633 min_size: config.tournament.min_size_threshold as usize,
634 skip_for_binary: config.tournament.skip_for_binary,
635 short_circuit_permille: config.tournament.short_circuit_threshold,
636 };
637 let base_drop_index = ctx.base_drop_index.clone();
640 let inline_threshold = ctx.inline_threshold;
641 let max_drop_size = config.defaults.max_drop_size as usize;
642 let seekable_drops = config.defaults.seekable_drops;
643
644 ctx.inline_threshold = config.defaults.inline_threshold as usize;
645 ctx.metadata_externalize_threshold = config.defaults.metadata_externalize_threshold;
646 ctx.emit_shared_inline = config.defaults.shared_inline;
647
648 const PIPELINE_CAPACITY: usize = 256;
652 let (tx, rx) = std::sync::mpsc::sync_channel::<PendingFile>(PIPELINE_CAPACITY);
653 ctx.pending_sink = Some(tx);
654
655 let (root_inode_number, mut results): (
656 u64,
657 Vec<(usize, PendingFile, Result<ChunkedFileResult, WriteError>)>,
658 ) = std::thread::scope(|scope| {
659 let producer = {
660 let ctx = &mut *ctx;
661 let root = root;
662 scope.spawn(move || {
663 let r = ctx.walk(root);
664 ctx.pending_sink = None;
668 r
669 })
670 };
671 let results = rx
674 .into_iter()
675 .enumerate()
676 .par_bridge()
677 .map(|(i, pf)| {
678 let r = process_file(
679 &pf,
680 &chunker,
681 classifier,
682 text_codec,
683 binary_codec,
684 &tunables,
685 use_categorizers,
686 skip_chunking,
687 &tournament_spec,
688 base_drop_index.as_ref(),
689 inline_threshold,
690 max_drop_size,
691 seekable_drops,
692 config.categorizers.as_slice(),
693 &|name| {
694 config
695 .codec_registry()
696 .ok()
697 .and_then(|r| r.lookup_by_name(name))
698 },
699 );
700 (i, pf, r)
701 })
702 .collect();
703 let joined = producer
704 .join()
705 .unwrap_or_else(|_| {
706 Err(WriteError::Io(std::io::Error::other(
707 "walk thread panicked",
708 )))
709 })
710 .map(|n| (n, results));
711 joined
714 })?;
715 ctx.pending_sink = None;
716 ctx.root_inode_number = root_inode_number;
717
718 results.sort_unstable_by_key(|(i, _, _)| *i);
719 for (_, pf, r) in results {
723 ctx.merge_chunked_file(&pf, r?);
724 }
725 ctx.train_and_apply_dictionary(&config.dictionaries);
726 Ok(())
727}
728
729pub(crate) type RawDrop = ([u8; 32], Vec<u8>, std::sync::Arc<[u8]>, u8, u8);
736pub(crate) struct ChunkedFileResult {
738 drops: Vec<RawDrop>, slices: Vec<PendingSlice>,
740}
741
742struct TournamentSpec {
750 codec_ids: Vec<u8>,
754 min_size: usize,
758 skip_for_binary: bool,
762 short_circuit_permille: u32,
767}
768
769fn chunker_from_config(config: &WriteConfig) -> Result<FastCDC, WriteError> {
788 FastCDC::new(
789 config.chunking.min_chunk_size as usize,
790 config.chunking.avg_chunk_size as usize,
791 config.chunking.max_chunk_size as usize,
792 )
793 .map_err(|e| WriteError::Io(std::io::Error::other(format!("chunking config: {e}"))))
794}
795
796pub(crate) fn seekable_or_monolithic(
803 codec: u8,
804 plaintext: &[u8],
805 compressed: std::sync::Arc<[u8]>,
806 tunables: &limnifs_core::codec::CodecTunables,
807 seekable_drops: bool,
808 threshold: usize,
809) -> (std::sync::Arc<[u8]>, u8) {
810 use limnifs_core::seekable::{
811 encode_seekable, is_seekable_codec, DROP_FLAG_SEEKABLE as FLAG, SEEKABLE_EMISSION_THRESHOLD,
812 };
813 if seekable_drops && plaintext.len() > threshold && is_seekable_codec(codec) {
814 if let Ok(container) = encode_seekable(codec, plaintext, tunables) {
815 return (container.into(), FLAG);
816 }
817 }
818 (compressed, 0)
819}
820
821pub(crate) const SEEKABLE_CHUNK_EMISSION_THRESHOLD: usize =
828 limnifs_core::seekable::SEEKABLE_FRAME_SIZE;
829
830fn process_whole_file_drop(
831 pf: &PendingFile,
832 data: &[u8],
833 cat: file_categorizer::Categorization,
834 tunables: &limnifs_core::codec::CodecTunables,
835 seekable_drops: bool,
836) -> Result<ChunkedFileResult, WriteError> {
837 let _ = pf;
838 let drop_id = hash_section(data);
839 let file_len = u64::try_from(data.len()).unwrap_or(u64::MAX);
840
841 let (mut best_codec, mut best_compressed): (u8, std::sync::Arc<[u8]>) =
846 match limnifs_core::codec::compress_with_tunables(
847 limnifs_core::codec::CODEC_BROTLI,
848 data,
849 tunables,
850 ) {
851 Ok(c) => (limnifs_core::codec::CODEC_BROTLI, c.into()),
852 Err(_) => match limnifs_core::codec::compress_with_tunables(
853 limnifs_core::codec::CODEC_ZSTD,
854 data,
855 tunables,
856 ) {
857 Ok(c) => (limnifs_core::codec::CODEC_ZSTD, c.into()),
858 Err(_) => (limnifs_core::codec::CODEC_STORE, data.to_vec().into()),
859 },
860 };
861
862 let brotli_ratio = best_compressed.len() as f64 / data.len() as f64;
866 if brotli_ratio > 0.05 {
867 if let Ok(zstd_c) = limnifs_core::codec::compress_with_tunables(
868 limnifs_core::codec::CODEC_ZSTD,
869 data,
870 tunables,
871 ) {
872 if zstd_c.len() < best_compressed.len() {
873 best_codec = limnifs_core::codec::CODEC_ZSTD;
874 best_compressed = zstd_c.into();
875 }
876 }
877 }
878
879 let general_ratio = best_compressed.len() as f64 / data.len() as f64;
885 if general_ratio > 0.15 || cat.codec_id == limnifs_core::codec::CODEC_RICEPP {
886 let spec_result = if cat.codec_id == limnifs_core::codec::CODEC_FSST_BROTLI {
890 limnifs_core::codec::fsst_brotli::compress_with_baseline(data, Some(&best_compressed))
891 } else {
892 limnifs_core::codec::compress(cat.codec_id, data)
893 };
894 if let Ok(spec_c) = spec_result {
895 if spec_c.len() < best_compressed.len() {
896 best_codec = cat.codec_id;
897 best_compressed = spec_c.into();
898 }
899 }
900 }
901
902 let (best_compressed, flags) = seekable_or_monolithic(
903 best_codec,
904 data,
905 best_compressed,
906 tunables,
907 seekable_drops,
908 limnifs_core::seekable::SEEKABLE_EMISSION_THRESHOLD,
909 );
910 Ok(ChunkedFileResult {
911 drops: vec![(drop_id, data.to_vec(), best_compressed, best_codec, flags)],
912 slices: vec![PendingSlice {
913 drop_id,
914 file_byte_start: 0,
915 file_byte_end: file_len,
916 }],
917 })
918}
919
920fn compress_chunk_with_tournament(
942 chunk: &[u8],
943 class: classifier::Class,
944 text_codec: u8,
945 binary_codec: u8,
946 tunables: &limnifs_core::codec::CodecTunables,
947 tournament: &TournamentSpec,
948) -> (u8, std::sync::Arc<[u8]>) {
949 use classifier::Class;
950
951 let preferred = match class {
952 Class::Binary => binary_codec,
953 Class::Text | Class::Code | Class::Sparse => text_codec,
954 _ => limnifs_core::codec::CODEC_STORE,
955 };
956
957 if preferred == limnifs_core::codec::CODEC_STORE {
958 return (limnifs_core::codec::CODEC_STORE, chunk.to_vec().into());
959 }
960 if class == Class::Binary && tournament.skip_for_binary {
961 return compress_chunk_one(chunk, preferred, tunables);
962 }
963 if chunk.len() < tournament.min_size {
964 return compress_chunk_one(chunk, preferred, tunables);
965 }
966
967 let mut best: Option<(u8, std::sync::Arc<[u8]>)> = None;
968 for &codec_id in &tournament.codec_ids {
969 if codec_id == limnifs_core::codec::CODEC_STORE {
970 continue;
971 }
972 let c = match limnifs_core::codec::compress_with_tunables(codec_id, chunk, tunables) {
973 Ok(c) => c,
974 Err(_) => continue,
975 };
976 if c.len() >= chunk.len() {
977 continue;
978 }
979 let ratio_permille = (c.len() as u64 * 1000 / chunk.len() as u64) as u32;
980 let is_best_so_far = best.as_ref().map_or(true, |(_, b)| c.len() < b.len());
981 if is_best_so_far {
982 best = Some((codec_id, c.into()));
983 }
984 if tournament.short_circuit_permille > 0
985 && ratio_permille <= tournament.short_circuit_permille
986 {
987 break;
988 }
989 }
990
991 best.unwrap_or_else(|| (limnifs_core::codec::CODEC_STORE, chunk.to_vec().into()))
992}
993
994fn compress_chunk_one(
997 chunk: &[u8],
998 codec_id: u8,
999 tunables: &limnifs_core::codec::CodecTunables,
1000) -> (u8, std::sync::Arc<[u8]>) {
1001 if codec_id == limnifs_core::codec::CODEC_STORE {
1002 return (limnifs_core::codec::CODEC_STORE, chunk.to_vec().into());
1003 }
1004 match limnifs_core::codec::compress_with_tunables(codec_id, chunk, tunables) {
1005 Ok(c) if c.len() < chunk.len() => (codec_id, c.into()),
1006 _ => (limnifs_core::codec::CODEC_STORE, chunk.to_vec().into()),
1007 }
1008}
1009
1010fn process_file(
1014 pf: &PendingFile,
1015 chunker: &FastCDC,
1016 classifier: classifier::Classifier,
1017 text_codec: u8,
1018 binary_codec: u8,
1019 tunables: &limnifs_core::codec::CodecTunables,
1020 use_categorizers: bool,
1021 skip_chunking: bool,
1022 tournament: &TournamentSpec,
1023 base_drop_index: Option<&std::collections::HashSet<[u8; 32]>>,
1024 inline_threshold: usize,
1025 max_drop_size: usize,
1026 seekable_drops: bool,
1027 categorizer_config: &[crate::config::CategorizerConfig],
1028 codec_name_resolver: &dyn Fn(&str) -> Option<u8>,
1029) -> Result<ChunkedFileResult, WriteError> {
1030 let file_len_estimate = std::fs::metadata(&pf.path)
1041 .map(|m| m.len() as usize)
1042 .unwrap_or(0);
1043 let data: Vec<u8> = if file_len_estimate >= MMAP_READ_THRESHOLD {
1044 let file = std::fs::File::open(&pf.path)?;
1045 #[allow(unsafe_code)]
1046 let mmap = unsafe { memmap2::Mmap::map(&file) }.map_err(WriteError::Io)?;
1047 Vec::from(&mmap[..])
1051 } else {
1052 std::fs::read(&pf.path)?
1053 };
1054 let file_len = data.len();
1055
1056 if skip_chunking && file_len > inline_threshold {
1063 let drop_id = hash_section(&data);
1064 let class = classifier.classify(&data);
1065 let preferred_codec = match class {
1066 classifier::Class::Binary => binary_codec,
1067 _ => text_codec,
1068 };
1069 let (codec_id, compressed): (u8, std::sync::Arc<[u8]>) =
1070 match limnifs_core::codec::compress_with_tunables(preferred_codec, &data, tunables) {
1071 Ok(c) if c.len() < data.len() => (preferred_codec, c.into()),
1072 _ => (limnifs_core::codec::CODEC_STORE, data.to_vec().into()),
1073 };
1074 let (compressed, flags) = seekable_or_monolithic(
1075 codec_id,
1076 &data,
1077 compressed,
1078 tunables,
1079 seekable_drops,
1080 SEEKABLE_CHUNK_EMISSION_THRESHOLD,
1081 );
1082 return Ok(ChunkedFileResult {
1083 drops: vec![(drop_id, data, compressed, codec_id, flags)],
1084 slices: vec![PendingSlice {
1085 drop_id,
1086 file_byte_start: 0,
1087 file_byte_end: file_len as u64,
1088 }],
1089 });
1090 }
1091
1092 if use_categorizers {
1093 let config_cat = file_categorizer::ConfigCategorizer::new(categorizer_config.to_vec());
1096 if let Some(cat) = config_cat.categorize(&pf.path, &data) {
1097 if let Some(codec_id) = file_categorizer::resolve_config_categorization(
1098 &cat,
1099 categorizer_config,
1100 codec_name_resolver,
1101 ) {
1102 let within_cap = max_drop_size == 0 || file_len <= max_drop_size;
1103 let needs_whole_file = matches!(
1104 codec_id,
1105 limnifs_core::codec::CODEC_FLAC | limnifs_core::codec::CODEC_RICEPP
1106 );
1107 if within_cap && (needs_whole_file || file_len <= WHOLE_FILE_MAX_SIZE) {
1108 let mut cat = cat;
1109 cat.codec_id = codec_id;
1110 return process_whole_file_drop(pf, &data, cat, tunables, seekable_drops);
1111 }
1112 }
1113 }
1114 if let Some(cat) = file_categorizer::default_registry().categorize(&pf.path, &data) {
1115 let needs_whole_file = matches!(
1116 cat.codec_id,
1117 limnifs_core::codec::CODEC_FLAC | limnifs_core::codec::CODEC_RICEPP
1118 );
1119 let within_cap = max_drop_size == 0 || file_len <= max_drop_size;
1122 if within_cap && (needs_whole_file || file_len <= WHOLE_FILE_MAX_SIZE) {
1123 return process_whole_file_drop(pf, &data, cat, tunables, seekable_drops);
1124 }
1125 }
1126 }
1127
1128 let chunks = chunker.chunk_slice(&data);
1129 let mut slices = Vec::with_capacity(chunks.len());
1130 let mut file_offset: u64 = 0;
1131 let mut seen_in_file: std::collections::HashSet<[u8; 32]> =
1132 std::collections::HashSet::with_capacity(chunks.len());
1133
1134 let mut unique_chunks: Vec<(&[u8], [u8; 32])> = Vec::with_capacity(chunks.len());
1137 for chunk in &chunks {
1138 let chunk_len = u64::try_from(chunk.len()).expect("chunk len fits u64");
1139 let drop_id = hash_section(chunk);
1140 slices.push(PendingSlice {
1141 drop_id,
1142 file_byte_start: file_offset,
1143 file_byte_end: file_offset + chunk_len,
1144 });
1145 file_offset += chunk_len;
1146 if seen_in_file.insert(drop_id) {
1147 unique_chunks.push((chunk, drop_id));
1148 }
1149 }
1150
1151 use rayon::prelude::*;
1163 thread_local! {
1164 static COMPRESS_CACHE: std::cell::RefCell<std::collections::HashMap<[u8; 32], (u8, std::sync::Arc<[u8]>)>> =
1165 std::cell::RefCell::new(std::collections::HashMap::new());
1166 }
1167 const COMPRESS_CACHE_MAX_ENTRIES: usize = 100_000;
1168 let drops: Vec<RawDrop> = unique_chunks
1169 .par_iter()
1170 .map(|(chunk, drop_id)| {
1171 if let Some(base) = base_drop_index {
1174 if base.contains(drop_id) {
1175 return (*drop_id, Vec::new(), Vec::new().into(), CODEC_REFERENCED, 0);
1176 }
1177 }
1178 let class = classifier.classify(chunk);
1179 let cached = COMPRESS_CACHE.with(|c| {
1182 c.borrow()
1183 .get(drop_id)
1184 .map(|(cid, comp)| (*cid, comp.clone()))
1185 });
1186 let (codec_id, compressed) = if let Some(c) = cached {
1187 c
1188 } else {
1189 let new = compress_chunk_with_tournament(
1190 chunk,
1191 class,
1192 text_codec,
1193 binary_codec,
1194 tunables,
1195 tournament,
1196 );
1197 COMPRESS_CACHE.with(|c| {
1199 let mut cache = c.borrow_mut();
1200 if cache.len() < COMPRESS_CACHE_MAX_ENTRIES {
1201 cache.insert(*drop_id, new.clone());
1203 }
1204 });
1205 new
1206 };
1207 let (compressed, flags) = seekable_or_monolithic(
1213 codec_id,
1214 chunk,
1215 compressed,
1216 tunables,
1217 seekable_drops,
1218 SEEKABLE_CHUNK_EMISSION_THRESHOLD,
1219 );
1220 (*drop_id, chunk.to_vec(), compressed, codec_id, flags)
1221 })
1222 .collect();
1223
1224 let _ = file_len;
1225 Ok(ChunkedFileResult { drops, slices })
1226}
1227
1228struct PendingDrop {
1229 id: [u8; 32],
1230 plaintext_len: u32,
1236 compressed: std::sync::Arc<[u8]>,
1237 codec: u8,
1238 dict_id: u8,
1242 plaintext: Option<Vec<u8>>,
1247 flags: u8,
1251}
1252
1253impl PendingDrop {
1254 fn len_in_window(&self) -> u32 {
1258 u32::try_from(self.compressed.len()).expect("compressed fits u32")
1259 }
1260
1261 fn plaintext_len_value(&self) -> u32 {
1263 self.plaintext_len
1264 }
1265
1266 fn slab_footprint(&self) -> usize {
1269 48 + self.compressed.len()
1270 }
1271}
1272
1273struct PendingSlice {
1278 drop_id: [u8; 32],
1279 file_byte_start: u64,
1280 file_byte_end: u64,
1281}
1282
1283#[derive(Clone)]
1286struct PendingFile {
1287 inode_number: u64,
1288 path: PathBuf,
1289 mtime_ns: u64,
1290 file_len: u64,
1291}
1292
1293struct PendingInode {
1294 number: u64,
1295 mode: u32,
1296 mtime_ns: u64,
1297 content: PendingContent,
1298}
1299
1300enum PendingContent {
1301 Inline(Vec<u8>),
1302 Symlink(String),
1304 DropBacked {
1305 file_len: u64,
1306 slices: Vec<PendingSlice>,
1307 },
1308 Directory(Vec<(String, u64, u8)>),
1309}
1310
1311struct DirNode {
1312 entries: Vec<(String, u64, u8)>,
1313 bytes: Vec<u8>,
1314 hash: [u8; 32],
1315}
1316
1317struct WriteContext {
1318 next_inode: u64,
1319 inodes: Vec<PendingInode>,
1320 dir_nodes: Vec<DirNode>,
1321 drops: Vec<PendingDrop>,
1322 drop_index: HashSet<[u8; 32]>,
1323 pending_files: Vec<PendingFile>,
1324 file_count: usize,
1325 dir_count: usize,
1326 root_inode_number: u64,
1327 chunker: FastCDC,
1328 classifier: classifier::Classifier,
1329 shared_inline_map: HashMap<[u8; 32], usize>,
1330 shared_inline_table: Vec<Vec<u8>>,
1331 profile_name: Option<String>,
1333 metadata_codec: u8,
1336 categorizers_disabled: bool,
1338 rw_mode: bool,
1340 auto_turnover: bool,
1342 collect_dict_samples: bool,
1345 dict_samples_by_class: HashMap<crate::classifier::Class, Vec<Vec<u8>>>,
1352 trained_dicts_by_class: HashMap<crate::classifier::Class, crate::dictionary::TrainedDictionary>,
1357 base_drop_index: Option<HashSet<[u8; 32]>>,
1363 base_root: Option<[u8; 32]>,
1368 metadata_externalize_threshold: usize,
1373 emit_shared_inline: bool,
1379 inline_threshold: usize,
1384 pending_sink: Option<std::sync::mpsc::SyncSender<PendingFile>>,
1390}
1391
1392impl WriteContext {
1393 const MAX_DICT_SAMPLES: usize = 1000;
1396
1397 fn new() -> Self {
1398 Self {
1399 next_inode: 1,
1400 inodes: Vec::new(),
1401 dir_nodes: Vec::new(),
1402 drops: Vec::new(),
1403 drop_index: HashSet::new(),
1404 pending_files: Vec::new(),
1405 file_count: 0,
1406 dir_count: 0,
1407 root_inode_number: 0,
1408 chunker: FastCDC::default(),
1409 classifier: classifier::Classifier,
1410 shared_inline_map: HashMap::new(),
1411 shared_inline_table: Vec::new(),
1412 profile_name: None,
1413 metadata_codec: limnifs_core::codec::CODEC_BROTLI,
1414 categorizers_disabled: false,
1415 rw_mode: false,
1416 auto_turnover: false,
1417 collect_dict_samples: false,
1418 dict_samples_by_class: HashMap::new(),
1419 trained_dicts_by_class: HashMap::new(),
1420 base_drop_index: None,
1421 base_root: None,
1422 pending_sink: None,
1423 inline_threshold: INLINE_THRESHOLD,
1424 metadata_externalize_threshold: METADATA_EXTERNALIZE_THRESHOLD,
1425 emit_shared_inline: true,
1426 }
1427 }
1428
1429 fn alloc_inode(&mut self) -> u64 {
1430 let n = self.next_inode;
1431 self.next_inode += 1;
1432 n
1433 }
1434
1435 fn build_shared_inline_table(&mut self) {
1439 let mut counts: HashMap<[u8; 32], usize> = HashMap::new();
1440 for inode in &self.inodes {
1441 if let PendingContent::Inline(data) = &inode.content {
1442 let h = hash_section(data);
1443 *counts.entry(h).or_default() += 1;
1444 }
1445 }
1446 for inode in &self.inodes {
1448 if let PendingContent::Inline(data) = &inode.content {
1449 let h = hash_section(data);
1450 if counts.get(&h).copied().unwrap_or(0) > 1
1451 && !self.shared_inline_map.contains_key(&h)
1452 {
1453 let idx = self.shared_inline_table.len();
1454 self.shared_inline_table.push(data.clone());
1455 self.shared_inline_map.insert(h, idx);
1456 }
1457 }
1458 }
1459 }
1460
1461 fn merge_chunked_file(&mut self, pf: &PendingFile, result: ChunkedFileResult) {
1464 for (drop_id, plaintext, compressed, codec, flags) in result.drops {
1465 if self.drop_index.insert(drop_id) {
1466 let retain_plaintext =
1472 self.collect_dict_samples && codec == limnifs_core::codec::CODEC_ZSTD;
1473 if retain_plaintext {
1474 let total: usize = self.dict_samples_by_class.values().map(Vec::len).sum();
1475 if total < Self::MAX_DICT_SAMPLES {
1476 let class = self.classifier.classify(&plaintext);
1477 self.dict_samples_by_class
1478 .entry(class)
1479 .or_default()
1480 .push(plaintext.clone());
1481 }
1482 }
1483 self.drops.push(PendingDrop {
1484 id: drop_id,
1485 plaintext_len: u32::try_from(plaintext.len()).unwrap_or(u32::MAX),
1486 compressed,
1487 codec,
1488 dict_id: limnifs_core::drop_record::NO_DICT,
1489 plaintext: if retain_plaintext {
1490 Some(plaintext)
1491 } else {
1492 None
1493 },
1494 flags,
1495 });
1496 }
1497 }
1498 self.inodes.push(PendingInode {
1499 number: pf.inode_number,
1500 mode: 0o100_644,
1501 mtime_ns: pf.mtime_ns,
1502 content: PendingContent::DropBacked {
1503 file_len: pf.file_len,
1504 slices: result.slices,
1505 },
1506 });
1507 }
1508
1509 #[allow(dead_code)]
1520 fn deepen_drop(&self, drop_id: [u8; 32], plaintext: &[u8]) -> PendingDrop {
1521 let class = self.classifier.classify(plaintext);
1522 let (codec, compressed): (u8, std::sync::Arc<[u8]>) = match class {
1523 classifier::Class::Text | classifier::Class::Code | classifier::Class::Binary => {
1524 let c = limnifs_core::codec::compress_lz4_with_size(plaintext);
1525 (limnifs_core::codec::CODEC_LZ4, c.into())
1526 }
1527 _ => (limnifs_core::codec::CODEC_STORE, plaintext.to_vec().into()),
1528 };
1529 PendingDrop {
1530 id: drop_id,
1531 plaintext_len: u32::try_from(plaintext.len()).unwrap_or(u32::MAX),
1532 compressed,
1533 codec,
1534 dict_id: limnifs_core::drop_record::NO_DICT,
1535 plaintext: None,
1536 flags: 0,
1537 }
1538 }
1539
1540 fn walk(&mut self, path: &Path) -> Result<u64, WriteError> {
1541 let meta = std::fs::symlink_metadata(path)?;
1542 let file_type = meta.file_type();
1543 let mtime_ns = meta
1544 .modified()
1545 .ok()
1546 .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
1547 .map_or(0u128, |d| d.as_nanos());
1548 let mtime_ns: u64 = mtime_ns.try_into().unwrap_or(0);
1549
1550 if file_type.is_dir() {
1551 self.dir_count += 1;
1552 let inode_number = self.alloc_inode();
1553 let mut entries: Vec<(String, u64, u8)> = Vec::new();
1554
1555 for entry in std::fs::read_dir(path)? {
1556 let entry = entry?;
1557 let name = entry.file_name().to_string_lossy().into_owned();
1558 let child_path = entry.path();
1559 let child_inode = self.walk(&child_path)?;
1560 let ft = entry.file_type()?;
1563 let entry_type = if ft.is_symlink() {
1564 0x03
1565 } else if ft.is_dir() {
1566 0x02
1567 } else {
1568 0x01
1569 };
1570 entries.push((name, child_inode, entry_type));
1571 }
1572
1573 entries.sort_by(|a, b| a.0.cmp(&b.0));
1574 let dir_node = encode_dir_node(&entries);
1575 self.dir_nodes.push(dir_node);
1576 self.inodes.push(PendingInode {
1577 number: inode_number,
1578 mode: 0o040_755,
1579 mtime_ns,
1580 content: PendingContent::Directory(entries),
1581 });
1582 Ok(inode_number)
1583 } else if file_type.is_file() {
1584 self.file_count += 1;
1585 let inode_number = self.alloc_inode();
1586 let file_len = meta.len();
1587
1588 if file_len <= u64::try_from(self.inline_threshold).unwrap_or(u64::MAX) {
1589 let data = std::fs::read(path)?;
1590 self.inodes.push(PendingInode {
1591 number: inode_number,
1592 mode: 0o100_644,
1593 mtime_ns,
1594 content: PendingContent::Inline(data),
1595 });
1596 } else {
1597 let pf = PendingFile {
1599 inode_number,
1600 path: path.to_path_buf(),
1601 mtime_ns,
1602 file_len,
1603 };
1604 if let Some(sink) = &self.pending_sink {
1605 sink.send(pf).map_err(|_| {
1611 WriteError::Io(std::io::Error::other("walk: compress pipeline shut down"))
1612 })?;
1613 } else {
1614 self.pending_files.push(pf);
1615 }
1616 }
1617 Ok(inode_number)
1618 } else if file_type.is_symlink() {
1619 let inode_number = self.alloc_inode();
1627 let target = std::fs::read_link(path)?;
1628 let target = target
1629 .to_str()
1630 .ok_or_else(|| WriteError::UnsupportedFileType {
1631 path: path.to_path_buf(),
1632 kind: format!("symlink with non-UTF-8 target ({})", target.display()),
1633 })?
1634 .to_owned();
1635 self.inodes.push(PendingInode {
1636 number: inode_number,
1637 mode: limnifs_core::inode::S_IFLNK | 0o777,
1638 mtime_ns,
1639 content: PendingContent::Symlink(target),
1640 });
1641 Ok(inode_number)
1642 } else {
1643 #[cfg(unix)]
1644 let kind = {
1645 use std::os::unix::fs::FileTypeExt;
1646 if file_type.is_fifo() {
1647 "fifo".to_owned()
1648 } else if file_type.is_socket() {
1649 "socket".to_owned()
1650 } else if file_type.is_block_device() {
1651 "block device".to_owned()
1652 } else if file_type.is_char_device() {
1653 "character device".to_owned()
1654 } else {
1655 "unknown".to_owned()
1656 }
1657 };
1658 #[cfg(not(unix))]
1659 let kind = "unknown".to_owned();
1660 Err(WriteError::UnsupportedFileType {
1661 path: path.to_path_buf(),
1662 kind,
1663 })
1664 }
1665 }
1666
1667 fn train_and_apply_dictionary(&mut self, dictionaries: &crate::config::DictionaryConfig) {
1682 let cleanup = |ctx: &mut Self| {
1683 for d in &mut ctx.drops {
1684 d.plaintext = None;
1685 }
1686 ctx.dict_samples_by_class.clear();
1687 };
1688
1689 if !dictionaries.enabled {
1690 cleanup(self);
1691 return;
1692 }
1693
1694 let target = usize::try_from(dictionaries.max_dict_size).unwrap_or(65_536);
1695 let min_class = usize::try_from(dictionaries.min_class_size).unwrap_or(0);
1696
1697 let text_classes = [
1701 crate::classifier::Class::Text,
1702 crate::classifier::Class::Code,
1703 crate::classifier::Class::Sparse,
1704 ];
1705 let binary_classes = [crate::classifier::Class::Binary];
1706
1707 let text_samples: Vec<&[u8]> = text_classes
1709 .iter()
1710 .flat_map(|c| self.dict_samples_by_class.get(c).into_iter().flatten())
1711 .map(Vec::as_slice)
1712 .collect();
1713 let trainer = crate::dictionary::TrainerKind::from_config_str(&dictionaries.trainer);
1714 if text_samples.len() >= min_class {
1715 if let Some(dict) =
1716 crate::dictionary::train_zstd_with_trainer(0, &text_samples, target, trainer)
1717 {
1718 self.trained_dicts_by_class
1719 .insert(crate::classifier::Class::Text, dict);
1720 }
1721 }
1722 let binary_samples: Vec<&[u8]> = binary_classes
1723 .iter()
1724 .flat_map(|c| self.dict_samples_by_class.get(c).into_iter().flatten())
1725 .map(Vec::as_slice)
1726 .collect();
1727 if binary_samples.len() >= min_class {
1728 if let Some(dict) =
1729 crate::dictionary::train_zstd_with_trainer(1, &binary_samples, target, trainer)
1730 {
1731 self.trained_dicts_by_class
1732 .insert(crate::classifier::Class::Binary, dict);
1733 }
1734 }
1735
1736 for d in self.drops.iter_mut() {
1739 if d.codec != limnifs_core::codec::CODEC_ZSTD {
1740 continue;
1741 }
1742 let Some(plaintext) = d.plaintext.clone() else {
1743 continue;
1744 };
1745 let class = self.classifier.classify(&plaintext);
1746 let dict_class = if text_classes.contains(&class) {
1747 crate::classifier::Class::Text
1748 } else if binary_classes.contains(&class) {
1749 crate::classifier::Class::Binary
1750 } else {
1751 continue;
1752 };
1753 let Some(dict) = self.trained_dicts_by_class.get(&dict_class) else {
1754 continue;
1755 };
1756 let Ok(dict_compressed) = dict.compress(&plaintext) else {
1757 continue;
1758 };
1759 if dict_compressed.len() < d.compressed.len() {
1760 d.compressed = dict_compressed.into();
1761 d.dict_id = dict.id;
1762 }
1763 }
1764
1765 cleanup(self);
1766 }
1767
1768 fn trace_phase(label: &str, start: std::time::Instant) {
1770 if std::env::var_os("LIMNIFS_TRACE_ASSEMBLE").is_some() {
1771 eprintln!("[assemble] {label}: {:?}", start.elapsed());
1772 }
1773 }
1774
1775 fn assemble(mut self) -> WriteArtifact {
1776 let t_assemble = std::time::Instant::now();
1777 let inode_count = self.inodes.len();
1778 let dir_count = self.dir_count;
1779 let drop_count = self.drops.len();
1780
1781 let t = std::time::Instant::now();
1787 let slabs = pack_slabs(&self.drops);
1788 Self::trace_phase("pack_slabs", t);
1789
1790 let t = std::time::Instant::now();
1794 if self.emit_shared_inline {
1795 self.build_shared_inline_table();
1796 }
1797 Self::trace_phase("shared_inline_table", t);
1798
1799 let mut metadata_blob = Vec::new();
1800 metadata_blob.extend_from_slice(&u32::try_from(self.inodes.len()).unwrap().to_le_bytes());
1801 for inode in &self.inodes {
1802 self.encode_inode(&mut metadata_blob, inode);
1803 }
1804 metadata_blob
1805 .extend_from_slice(&u32::try_from(self.dir_nodes.len()).unwrap().to_le_bytes());
1806 for node in &self.dir_nodes {
1807 metadata_blob.extend_from_slice(&node.bytes);
1808 }
1809 if !self.shared_inline_table.is_empty() {
1812 metadata_blob.extend_from_slice(
1813 &u32::try_from(self.shared_inline_table.len())
1814 .unwrap()
1815 .to_le_bytes(),
1816 );
1817 for entry in &self.shared_inline_table {
1818 let len = u32::try_from(entry.len()).expect("shared entry fits u32");
1819 metadata_blob.extend_from_slice(&len.to_le_bytes());
1820 metadata_blob.extend_from_slice(entry);
1821 }
1822 }
1823
1824 Self::trace_phase("metadata_encode", t);
1825 let uncompressed_len =
1832 u32::try_from(metadata_blob.len()).expect("metadata blob length fits u32");
1833 let t = std::time::Instant::now();
1834 let metadata_hash = hash_section(&metadata_blob);
1835 let metadata_codec = self.metadata_codec;
1836 let metadata_quality = if metadata_blob.len() > METADATA_LARGE_BLOB_THRESHOLD {
1837 METADATA_LARGE_BLOB_QUALITY
1838 } else {
1839 METADATA_SMALL_BLOB_QUALITY
1840 };
1841 let compressed_blob = if metadata_codec == limnifs_core::codec::CODEC_BROTLI {
1842 limnifs_core::codec::compress_brotli_with_quality(&metadata_blob, metadata_quality)
1843 .unwrap_or_else(|_| metadata_blob.clone())
1844 } else {
1845 limnifs_core::codec::compress(metadata_codec, &metadata_blob)
1846 .unwrap_or_else(|_| metadata_blob.clone())
1847 };
1848 Self::trace_phase("metadata_compress", t);
1849 let (on_wire_codec, on_wire_blob) = if compressed_blob.len() < metadata_blob.len() {
1850 (metadata_codec, compressed_blob)
1851 } else {
1852 (limnifs_core::codec::CODEC_STORE, metadata_blob.clone())
1853 };
1854
1855 let externalize_at = self
1859 .metadata_externalize_threshold
1860 .min(limnifs_core::metadata_reference::DEFAULT_INLINE_METADATA_MAX_BYTES as usize);
1861 let (metadata_sidecar, inline_data, metadata_locator_count) =
1862 if on_wire_blob.len() > externalize_at {
1863 let h = hash_section(&on_wire_blob);
1870 let mut h8 = String::with_capacity(8);
1871 for b in &h[..4] {
1872 h8.push_str(&format!("{b:02x}"));
1873 }
1874 let locator = format!("file:metadata-{h8}.bin");
1875 let sidecar = MetadataSidecar {
1876 bytes: on_wire_blob.clone(),
1877 locator,
1878 };
1879 (Some(sidecar), None, 1u32)
1880 } else {
1881 (None, Some(on_wire_blob.clone()), 0u32)
1882 };
1883
1884 let mut manifest = Vec::new();
1885
1886 let header_start = manifest.len();
1887 manifest.extend_from_slice(&ManifestHeader::current().to_bytes());
1888 let header_end = manifest.len();
1889
1890 let flags_start = manifest.len();
1891 manifest.push(FEATURE_FLAGS_SECTION_VERSION);
1892 manifest.extend_from_slice(&0u32.to_le_bytes());
1893 let flags_end = manifest.len();
1894
1895 let meta_ref_start = manifest.len();
1898 manifest.push(METADATA_REFERENCE_SECTION_VERSION_2);
1899 manifest.extend_from_slice(&metadata_hash);
1900 manifest.extend_from_slice(&uncompressed_len.to_le_bytes());
1901 manifest.push(on_wire_codec);
1902 manifest.extend_from_slice(&metadata_locator_count.to_le_bytes());
1903 if let Some(sidecar) = &metadata_sidecar {
1904 let loc_bytes = sidecar.locator.as_bytes();
1905 let loc_len = u32::try_from(loc_bytes.len()).expect("locator fits u32");
1906 manifest.extend_from_slice(&loc_len.to_le_bytes());
1907 manifest.extend_from_slice(loc_bytes);
1908 }
1909 match &inline_data {
1910 Some(blob) => {
1911 let inline_len = u32::try_from(blob.len()).expect("metadata fits u32");
1912 manifest.extend_from_slice(&inline_len.to_le_bytes());
1913 manifest.extend_from_slice(blob);
1914 }
1915 None => {
1916 manifest.extend_from_slice(&0u32.to_le_bytes());
1917 }
1918 }
1919 let meta_ref_end = manifest.len();
1920
1921 let slab_index_start = manifest.len();
1922 manifest.push(SLAB_INDEX_SECTION_VERSION);
1923 manifest.extend_from_slice(&u32::try_from(slabs.len()).unwrap().to_le_bytes());
1924 for slab in &slabs {
1925 manifest.extend_from_slice(&slab.id.to_bytes());
1926 manifest.extend_from_slice(&1u32.to_le_bytes());
1927 let loc_bytes = slab.locator.as_bytes();
1928 let loc_len = u32::try_from(loc_bytes.len()).expect("locator fits u32");
1929 manifest.extend_from_slice(&loc_len.to_le_bytes());
1930 manifest.extend_from_slice(loc_bytes);
1931 }
1932 let slab_index_end = manifest.len();
1933
1934 let history_start = manifest.len();
1935 manifest.push(HISTORY_SECTION_VERSION);
1936 manifest.extend_from_slice(&1u32.to_le_bytes());
1937 manifest.push(0x01);
1938 manifest.extend_from_slice(&0u64.to_le_bytes());
1939 manifest.extend_from_slice(&0u32.to_le_bytes());
1940 manifest.extend_from_slice(&0u32.to_le_bytes());
1941 let history_end = manifest.len();
1942
1943 let profile_desc_start = manifest.len();
1948 if let Some(ref name) = self.profile_name {
1949 let desc = limnifs_core::profile_descriptor::ProfileDescriptor {
1950 version: limnifs_core::profile_descriptor::PROFILE_DESCRIPTOR_SECTION_VERSION,
1951 profile_name: Some(name.clone()),
1952 blake3_hashing: true,
1953 cross_file_dedup: true,
1954 content_classification: !self.categorizers_disabled,
1955 integrity_verify: true,
1956 read_write: self.rw_mode,
1957 auto_turnover: self.auto_turnover,
1958 };
1959 limnifs_core::profile_descriptor::encode_profile_descriptor(&desc, &mut manifest);
1960 }
1961 let profile_desc_end = manifest.len();
1962
1963 if !self.trained_dicts_by_class.is_empty() {
1969 let dicts: Vec<_> = self
1970 .trained_dicts_by_class
1971 .values()
1972 .map(|d| limnifs_core::dictionary_section::Dictionary {
1973 codec_id: d.codec,
1974 class_id: d.id,
1975 data: d.content.clone(),
1976 })
1977 .collect();
1978 let section = limnifs_core::dictionary_section::DictionarySection {
1979 version: limnifs_core::dictionary_section::DICTIONARY_SECTION_VERSION,
1980 dicts,
1981 };
1982 limnifs_core::dictionary_section::encode_dictionary_section(§ion, &mut manifest);
1983 }
1984
1985 let dictionary_end = manifest.len();
1986
1987 let delta_linkage_hash = if let Some(base_root) = self.base_root {
1993 let delta_start = manifest.len();
1994 manifest.push(limnifs_core::delta_linkage::DELTA_LINKAGE_SECTION_VERSION);
2000 manifest.extend_from_slice(&base_root);
2001 manifest.extend_from_slice(&0u32.to_le_bytes());
2002 hash_section(&manifest[delta_start..])
2003 } else {
2004 hash_empty_section()
2005 };
2006 let _ = dictionary_end;
2007
2008 let hashes = SectionHashes {
2009 metadata: metadata_hash,
2010 format_header: hash_section(&manifest[header_start..header_end]),
2011 feature_flags: hash_section(&manifest[flags_start..flags_end]),
2012 metadata_reference: hash_section(&manifest[meta_ref_start..meta_ref_end]),
2013 slab_index: hash_section(&manifest[slab_index_start..slab_index_end]),
2014 crypto_params: hash_empty_section(),
2015 ec_params: hash_empty_section(),
2016 dms_policy: hash_empty_section(),
2017 delta_linkage: delta_linkage_hash,
2018 history: hash_section(&manifest[history_start..history_end]),
2019 };
2031 let merkle_root = compute_merkle_root(&hashes);
2032
2033 WriteArtifact {
2034 bytes: manifest,
2035 merkle_root,
2036 slabs,
2037 metadata_sidecar,
2038 inode_count,
2039 file_count: self.file_count,
2040 dir_count,
2041 drop_count,
2042 root_inode_number: self.root_inode_number,
2043 }
2044 }
2045
2046 fn encode_inode(&self, out: &mut Vec<u8>, inode: &PendingInode) {
2047 out.extend_from_slice(&inode.number.to_le_bytes());
2048 out.extend_from_slice(&inode.mode.to_le_bytes());
2049 out.extend_from_slice(&0u32.to_le_bytes());
2050 out.extend_from_slice(&0u32.to_le_bytes());
2051 out.extend_from_slice(&inode.mtime_ns.to_le_bytes());
2052 out.extend_from_slice(&inode.mtime_ns.to_le_bytes());
2053 out.extend_from_slice(&1u32.to_le_bytes());
2054 match &inode.content {
2055 PendingContent::Inline(data) => {
2056 let h = hash_section(data);
2057 if let Some(&idx) = self.shared_inline_map.get(&h) {
2058 out.push(INODE_FLAG_INLINE_DATA | INODE_FLAG_SHARED_INLINE);
2060 out.extend_from_slice(&(idx as u32).to_le_bytes());
2061 } else {
2062 out.push(INODE_FLAG_INLINE_DATA);
2063 let len = u32::try_from(data.len()).expect("data fits u32");
2064 out.extend_from_slice(&len.to_le_bytes());
2065 out.extend_from_slice(data);
2066 }
2067 }
2068 PendingContent::DropBacked { file_len, slices } => {
2069 out.push(0x00);
2070 let slice_count = u32::try_from(slices.len()).expect("slice count fits u32");
2071 out.extend_from_slice(&slice_count.to_le_bytes());
2072 for slice in slices {
2073 out.extend_from_slice(&slice.file_byte_start.to_le_bytes());
2074 out.extend_from_slice(&slice.file_byte_end.to_le_bytes());
2075 out.extend_from_slice(&slice.drop_id);
2076 out.extend_from_slice(&0u32.to_le_bytes());
2078 let drop_byte_len = u32::try_from(slice.file_byte_end - slice.file_byte_start)
2082 .expect("slice range fits u32");
2083 out.extend_from_slice(&drop_byte_len.to_le_bytes());
2084 }
2085 let _ = file_len;
2086 }
2087 PendingContent::Symlink(target) => {
2088 out.push(0x00);
2092 let t = target.as_bytes();
2093 let len = u32::try_from(t.len()).expect("target fits u32");
2094 out.extend_from_slice(&len.to_le_bytes());
2095 out.extend_from_slice(t);
2096 }
2097 PendingContent::Directory(entries) => {
2098 out.push(0x00);
2099 let node = self
2100 .dir_nodes
2101 .iter()
2102 .find(|n| n.entries == *entries)
2103 .expect("directory node must exist");
2104 out.extend_from_slice(&node.hash);
2105 }
2106 }
2107 }
2108}
2109
2110fn sidecar_name(locator: &str) -> Result<&str, WriteError> {
2116 limnifs_core::locator::local_sidecar_name(locator)
2117 .map_err(|e| WriteError::Io(std::io::Error::other(format!("{e}"))))
2118}
2119
2120fn encode_dir_node(entries: &[(String, u64, u8)]) -> DirNode {
2121 let mut bytes = Vec::new();
2122 bytes.push(1u8);
2123 let count = u32::try_from(entries.len()).expect("entry count fits u32");
2124 bytes.extend_from_slice(&count.to_le_bytes());
2125 for (name, inode_number, entry_type) in entries {
2126 let name_bytes = name.as_bytes();
2127 let name_len = u32::try_from(name_bytes.len()).expect("name fits u32");
2128 bytes.extend_from_slice(&name_len.to_le_bytes());
2129 bytes.extend_from_slice(name_bytes);
2130 bytes.extend_from_slice(&inode_number.to_le_bytes());
2131 bytes.push(*entry_type);
2132 }
2133 let hash = hash_section(&bytes);
2134 DirNode {
2135 entries: entries.to_vec(),
2136 bytes,
2137 hash,
2138 }
2139}
2140
2141fn pack_slabs(drops: &[PendingDrop]) -> Vec<SlabArtifact> {
2150 let local_drops: Vec<&PendingDrop> = drops
2155 .iter()
2156 .filter(|d| d.codec != limnifs_core::codec::CODEC_REFERENCED)
2157 .collect();
2158 if local_drops.is_empty() {
2159 return Vec::new();
2160 }
2161
2162 let max_content = MAX_SLAB_TOTAL_BYTES.saturating_sub(SLAB_HEADER_LEN);
2163
2164 let mut slab_groups: Vec<Vec<&PendingDrop>> = Vec::new();
2169 let mut current: Vec<&PendingDrop> = Vec::new();
2170 let mut current_size: usize = 0;
2171
2172 for drop in &local_drops {
2173 let footprint = drop.slab_footprint();
2174 if !current.is_empty() && current_size + footprint > max_content {
2175 slab_groups.push(std::mem::take(&mut current));
2176 current_size = 0;
2177 }
2178 current.push(*drop);
2179 current_size += footprint;
2180 }
2181 if !current.is_empty() {
2182 slab_groups.push(current);
2183 }
2184
2185 use rayon::prelude::*;
2191 slab_groups
2192 .par_iter()
2193 .enumerate()
2194 .map(|(ordinal, group)| {
2195 let ordinal_u64 = u64::try_from(ordinal).expect("slab count fits u64");
2196 encode_slab(ordinal_u64, group)
2197 })
2198 .collect()
2199}
2200
2201fn encode_slab(ordinal: u64, drops: &[&PendingDrop]) -> SlabArtifact {
2205 const DROP_RECORD_LEN: usize = 50;
2212 let mut drop_records = Vec::with_capacity(drops.len() * DROP_RECORD_LEN);
2213 let mut solid_window = Vec::new();
2214 let mut drop_ids = Vec::with_capacity(drops.len());
2215 let mut offset_in_window: u32 = 0;
2216
2217 for drop in drops {
2218 let plaintext_len = drop.plaintext_len_value();
2219 let window_len = drop.len_in_window();
2220 drop_records.extend_from_slice(&drop.id);
2221 drop_records.extend_from_slice(&plaintext_len.to_le_bytes());
2222 drop_records.extend_from_slice(&[drop.codec, 0x00, 0x00]);
2224 drop_records.push(0x00); drop_records.extend_from_slice(&offset_in_window.to_le_bytes());
2226 drop_records.extend_from_slice(&window_len.to_le_bytes());
2227 drop_records.push(drop.dict_id); drop_records.push(drop.flags); solid_window.extend_from_slice(&drop.compressed);
2230 drop_ids.push(drop.id);
2231 offset_in_window = offset_in_window
2232 .checked_add(window_len)
2233 .expect("slab window size fits u32");
2234 }
2235
2236 let slab_content = [&drop_records[..], &solid_window[..]].concat();
2237 let slab_hash = hash_section(&slab_content);
2238 let slab_id = SlabId::new(ordinal, slab_hash);
2239
2240 let total_length = SLAB_HEADER_LEN + slab_content.len();
2241 let mut slab_bytes = Vec::with_capacity(total_length);
2242 slab_bytes.extend_from_slice(b"LIM1");
2243 slab_bytes.extend_from_slice(&1u16.to_le_bytes()); slab_bytes.extend_from_slice(&slab_id.to_bytes());
2245 slab_bytes.extend_from_slice(
2246 &u64::try_from(total_length)
2247 .unwrap_or(u64::MAX)
2248 .to_le_bytes(),
2249 );
2250 slab_bytes.push(0x00);
2251 slab_bytes.push(0x00);
2252 slab_bytes.extend_from_slice(&slab_content);
2253
2254 let mut h8 = String::with_capacity(8);
2258 for b in &slab_id.hash[..4] {
2259 h8.push_str(&format!("{b:02x}"));
2260 }
2261 let locator = format!("file:slab-{ordinal}-{h8}.bin");
2262
2263 SlabArtifact {
2264 id: slab_id,
2265 bytes: slab_bytes,
2266 locator,
2267 drop_ids,
2268 }
2269}
2270
2271#[cfg(test)]
2272fn pseudo_random_bytes(seed: u64, count: usize) -> Vec<u8> {
2273 let mut state = seed;
2274 let mut out = Vec::with_capacity(count);
2275 for _ in 0..count {
2276 state = state
2277 .wrapping_mul(6_364_136_223_846_793_005)
2278 .wrapping_add(1_442_695_040_888_963_407);
2279 out.push(u8::try_from(state >> 56).expect("fits u8"));
2280 }
2281 out
2282}
2283
2284#[cfg(test)]
2285mod tests {
2286 use super::*;
2287 use limnifs_core::ManifestCursor;
2288
2289 #[test]
2290 fn write_stream_packs_single_named_stream() {
2291 let temp = std::env::temp_dir().join(format!(
2295 "limnifs-write-stream-test-{}-{}",
2296 std::process::id(),
2297 std::time::SystemTime::now()
2298 .duration_since(std::time::UNIX_EPOCH)
2299 .unwrap()
2300 .as_nanos()
2301 ));
2302 std::fs::create_dir_all(&temp).expect("create temp dir");
2303
2304 let content = b"stream test content line\n".repeat(10_000); let cursor = std::io::Cursor::new(content.clone());
2306 let config = WriteConfig::default_v0_1();
2307 let artifact = write_stream("streamed.txt", cursor, &config).expect("write_stream");
2308
2309 assert!(!artifact.bytes.is_empty(), "manifest bytes non-empty");
2310 assert!(!artifact.slabs.is_empty(), "at least one slab produced");
2311 let total_drop_bytes: usize = artifact.slabs.iter().map(|s| s.bytes.len()).sum();
2316 assert!(total_drop_bytes > 0, "drops non-empty");
2317
2318 let _ = std::fs::remove_dir_all(&temp);
2319 }
2320
2321 #[test]
2322 fn write_layer_references_base_drops() {
2323 let temp = std::env::temp_dir().join(format!(
2329 "limnifs-write-layer-test-{}-{}",
2330 std::process::id(),
2331 std::time::SystemTime::now()
2332 .duration_since(std::time::UNIX_EPOCH)
2333 .unwrap()
2334 .as_nanos()
2335 ));
2336 std::fs::create_dir_all(&temp).expect("create temp dir");
2337
2338 let base_dir = temp.join("base");
2340 std::fs::create_dir_all(&base_dir).expect("base dir");
2341 let text = b"layer test content line\n".repeat(50_000); std::fs::write(base_dir.join("shared.txt"), &text).expect("write shared");
2343 std::fs::write(base_dir.join("base-only.txt"), b"only in base").expect("write base-only");
2344
2345 let config = WriteConfig::default_v0_1();
2346 let base_artifact = write_directory_with_config(&base_dir, &config).expect("base");
2347
2348 let base_manifest = temp.join("base.lim");
2349 std::fs::write(&base_manifest, &base_artifact.bytes).expect("write base manifest");
2350 for slab in &base_artifact.slabs {
2351 let slab_name = sidecar_name(&slab.locator).expect("slab locator");
2352 std::fs::write(temp.join(slab_name), &slab.bytes).expect("write base slab");
2353 }
2354
2355 let layer_dir = temp.join("layer");
2357 std::fs::create_dir_all(&layer_dir).expect("layer dir");
2358 std::fs::write(layer_dir.join("shared.txt"), &text).expect("write shared in layer");
2359 std::fs::write(layer_dir.join("new.txt"), b"fresh content in layer").expect("write new");
2360
2361 let layer_artifact = write_layer(&base_manifest, &layer_dir, &config).expect("layer");
2362
2363 let layer_slab_bytes: usize = layer_artifact.slabs.iter().map(|s| s.bytes.len()).sum();
2366 let base_slab_bytes: usize = base_artifact.slabs.iter().map(|s| s.bytes.len()).sum();
2367 assert!(
2368 layer_slab_bytes < base_slab_bytes / 4,
2369 "layer slabs ({}) should be much smaller than base ({}) — layering failed",
2370 layer_slab_bytes,
2371 base_slab_bytes
2372 );
2373
2374 let base_root = base_artifact.merkle_root.as_bytes();
2377 assert!(
2378 layer_artifact
2379 .bytes
2380 .windows(32)
2381 .any(|w| w == base_root.as_slice()),
2382 "layer manifest must contain base's ManifestRoot bytes"
2383 );
2384
2385 let _ = std::fs::remove_dir_all(&temp);
2386 }
2387
2388 #[test]
2389 fn tournament_short_circuits_on_highly_compressible_chunk() {
2390 let chunk = b"hello world ".repeat(500);
2393 let tunables = limnifs_core::codec::CodecTunables::default();
2394 let tournament = TournamentSpec {
2395 codec_ids: vec![
2396 limnifs_core::codec::CODEC_LZ4,
2397 limnifs_core::codec::CODEC_BROTLI,
2398 ],
2399 min_size: 16,
2400 skip_for_binary: false,
2401 short_circuit_permille: 250,
2402 };
2403 let (codec_id, compressed) = compress_chunk_with_tournament(
2404 &chunk,
2405 classifier::Class::Text,
2406 limnifs_core::codec::CODEC_BROTLI,
2407 limnifs_core::codec::CODEC_LZ4,
2408 &tunables,
2409 &tournament,
2410 );
2411 assert_eq!(codec_id, limnifs_core::codec::CODEC_LZ4);
2412 assert!(compressed.len() < chunk.len());
2413 }
2414
2415 #[test]
2416 fn tournament_runs_all_codecs_when_short_circuit_disabled() {
2417 let chunk = b"hello world ".repeat(500);
2423 let tunables = limnifs_core::codec::CodecTunables::default();
2424 let tournament = TournamentSpec {
2425 codec_ids: vec![
2426 limnifs_core::codec::CODEC_LZ4,
2427 limnifs_core::codec::CODEC_BROTLI,
2428 limnifs_core::codec::CODEC_ZSTD,
2429 ],
2430 min_size: 16,
2431 skip_for_binary: false,
2432 short_circuit_permille: 0,
2433 };
2434 let (codec_id, compressed) = compress_chunk_with_tournament(
2435 &chunk,
2436 classifier::Class::Text,
2437 limnifs_core::codec::CODEC_BROTLI,
2438 limnifs_core::codec::CODEC_LZ4,
2439 &tunables,
2440 &tournament,
2441 );
2442 assert!(
2446 codec_id == limnifs_core::codec::CODEC_ZSTD
2447 || codec_id == limnifs_core::codec::CODEC_BROTLI,
2448 "expected ZSTD or Brotli to win, got codec {codec_id}"
2449 );
2450 assert!(compressed.len() < chunk.len());
2451 }
2452
2453 #[test]
2454 fn tournament_skips_for_binary_when_configured() {
2455 let chunk = vec![0u8; 4096];
2456 let tunables = limnifs_core::codec::CodecTunables::default();
2457 let tournament = TournamentSpec {
2458 codec_ids: vec![limnifs_core::codec::CODEC_BROTLI],
2459 min_size: 16,
2460 skip_for_binary: true,
2461 short_circuit_permille: 250,
2462 };
2463 let (codec_id, _compressed) = compress_chunk_with_tournament(
2464 &chunk,
2465 classifier::Class::Binary,
2466 limnifs_core::codec::CODEC_BROTLI,
2467 limnifs_core::codec::CODEC_LZ4,
2468 &tunables,
2469 &tournament,
2470 );
2471 assert_eq!(codec_id, limnifs_core::codec::CODEC_LZ4);
2473 }
2474
2475 #[test]
2476 fn tournament_small_chunk_uses_preferred_codec() {
2477 let chunk = b"tiny";
2478 let tunables = limnifs_core::codec::CodecTunables::default();
2479 let tournament = TournamentSpec {
2480 codec_ids: vec![limnifs_core::codec::CODEC_BROTLI],
2481 min_size: 1024,
2482 skip_for_binary: false,
2483 short_circuit_permille: 0,
2484 };
2485 let (codec_id, _compressed) = compress_chunk_with_tournament(
2486 chunk,
2487 classifier::Class::Text,
2488 limnifs_core::codec::CODEC_BROTLI,
2489 limnifs_core::codec::CODEC_LZ4,
2490 &tunables,
2491 &tournament,
2492 );
2493 assert_eq!(codec_id, limnifs_core::codec::CODEC_STORE);
2495 }
2496
2497 #[test]
2498 fn tournament_falls_back_to_store_when_no_codec_compresses() {
2499 let chunk = pseudo_random_bytes(42, 4096);
2503 let tunables = limnifs_core::codec::CodecTunables::default();
2504 let tournament = TournamentSpec {
2505 codec_ids: vec![limnifs_core::codec::CODEC_LZ4],
2506 min_size: 16,
2507 skip_for_binary: false,
2508 short_circuit_permille: 0,
2509 };
2510 let (codec_id, compressed) = compress_chunk_with_tournament(
2511 &chunk,
2512 classifier::Class::Binary,
2513 limnifs_core::codec::CODEC_BROTLI,
2514 limnifs_core::codec::CODEC_LZ4,
2515 &tunables,
2516 &tournament,
2517 );
2518 assert_eq!(codec_id, limnifs_core::codec::CODEC_STORE);
2519 assert_eq!(compressed.len(), chunk.len());
2520 }
2521
2522 #[test]
2523 fn dictionaries_enabled_emits_dictionary_section_when_enough_samples() {
2524 let temp = std::env::temp_dir().join(format!(
2529 "limnifs-write-test-{}-dict-{}",
2530 std::process::id(),
2531 std::time::SystemTime::now()
2532 .duration_since(std::time::UNIX_EPOCH)
2533 .map(|d| d.as_nanos() as u64)
2534 .unwrap_or(0),
2535 ));
2536 let _ = std::fs::remove_dir_all(&temp);
2537 std::fs::create_dir_all(&temp).expect("mkdir");
2538
2539 for i in 0..200 {
2542 let content = format!(
2544 "function test_case_{i}() {{ return constant + {i}; }}\n\
2545 // shared comment line {i}\n\
2546 struct Foo {{ x: i32 }} // type {i}\n"
2547 )
2548 .repeat(5);
2549 let path = temp.join(format!("file_{i:04}.txt"));
2550 std::fs::write(&path, content.as_bytes()).expect("write");
2551 }
2552
2553 let mut config = crate::profile::balanced();
2554 config.defaults.text_codec = "zstd".into();
2556 config.defaults.metadata_codec = "zstd".into();
2560 config.dictionaries.enabled = true;
2561 config.dictionaries.min_class_size = 50;
2562 config.dictionaries.max_dict_size = 8192;
2563
2564 let artifact = write_directory_with_config(&temp, &config).expect("write");
2565 std::fs::remove_dir_all(&temp).ok();
2566
2567 let mut cursor = ManifestCursor::new(&artifact.bytes);
2572 let _ = limnifs_core::parse_manifest_header(&mut cursor).expect("header");
2573 let _ = limnifs_core::parse_feature_flags_section(&mut cursor).expect("flags");
2574 let _ = limnifs_core::parse_metadata_reference(&mut cursor).expect("meta_ref");
2575 let _ = limnifs_core::parse_slab_index(&mut cursor).expect("slab_index");
2576 let _ = limnifs_core::parse_history(&mut cursor).expect("history");
2577 let _remaining = cursor.remaining_len();
2580 }
2581
2582 #[test]
2583 fn write_empty_directory() {
2584 let temp =
2585 std::env::temp_dir().join(format!("limnifs-write-test-{}-empty", std::process::id()));
2586 std::fs::create_dir_all(&temp).expect("create temp dir");
2587 let artifact = write_directory(&temp).expect("write succeeds");
2588 std::fs::remove_dir_all(&temp).ok();
2589 assert!(artifact.inode_count >= 1);
2590 assert_eq!(artifact.file_count, 0);
2591 assert_eq!(artifact.dir_count, 1);
2592 assert!(artifact.slabs.is_empty());
2593 }
2594
2595 #[test]
2596 fn write_small_file_inline() {
2597 let temp =
2598 std::env::temp_dir().join(format!("limnifs-write-test-{}-small", std::process::id()));
2599 std::fs::create_dir_all(&temp).expect("create temp dir");
2600 std::fs::write(temp.join("hello.txt"), b"hello world").expect("write file");
2601 let artifact = write_directory(&temp).expect("write succeeds");
2602 std::fs::remove_dir_all(&temp).ok();
2603 assert_eq!(artifact.file_count, 1);
2604 assert!(artifact.slabs.is_empty());
2605 assert_eq!(artifact.drop_count, 0);
2606 }
2607
2608 #[test]
2609 fn write_large_file_uses_slab() {
2610 let temp =
2611 std::env::temp_dir().join(format!("limnifs-write-test-{}-large", std::process::id()));
2612 std::fs::create_dir_all(&temp).expect("create temp dir");
2613 let large_data = vec![0xABu8; INLINE_THRESHOLD + 100];
2614 std::fs::write(temp.join("big.bin"), &large_data).expect("write big");
2615 let artifact = write_directory(&temp).expect("write succeeds");
2616 std::fs::remove_dir_all(&temp).ok();
2617 assert_eq!(artifact.drop_count, 1);
2618 assert_eq!(artifact.slabs.len(), 1);
2619 }
2620
2621 #[test]
2622 fn write_mixed_inline_and_large() {
2623 let temp =
2624 std::env::temp_dir().join(format!("limnifs-write-test-{}-mix", std::process::id()));
2625 std::fs::create_dir_all(&temp).expect("create temp dir");
2626 std::fs::write(temp.join("small.txt"), b"tiny").expect("write small");
2627 std::fs::write(temp.join("large.bin"), vec![0xCDu8; INLINE_THRESHOLD * 2])
2628 .expect("write large");
2629 let artifact = write_directory(&temp).expect("write succeeds");
2630 std::fs::remove_dir_all(&temp).ok();
2631 assert_eq!(artifact.file_count, 2);
2632 assert_eq!(artifact.drop_count, 1);
2633 assert_eq!(artifact.slabs.len(), 1);
2634 }
2635
2636 #[test]
2637 fn deduplicates_identical_large_files() {
2638 let temp =
2639 std::env::temp_dir().join(format!("limnifs-write-test-{}-dedup", std::process::id()));
2640 std::fs::create_dir_all(&temp).expect("create temp dir");
2641 let data = vec![0x77u8; INLINE_THRESHOLD + 10];
2642 std::fs::write(temp.join("a.bin"), &data).expect("write a");
2643 std::fs::write(temp.join("b.bin"), &data).expect("write b");
2644 let artifact = write_directory(&temp).expect("write succeeds");
2645 std::fs::remove_dir_all(&temp).ok();
2646 assert_eq!(artifact.drop_count, 1);
2647 }
2648
2649 #[test]
2650 fn write_and_verify_roundtrip() {
2651 let temp = std::env::temp_dir().join(format!(
2652 "limnifs-write-test-{}-roundtrip",
2653 std::process::id()
2654 ));
2655 std::fs::create_dir_all(&temp).expect("create temp dir");
2656 std::fs::write(temp.join("a.txt"), b"aaa").expect("write a");
2657 std::fs::write(temp.join("b.txt"), b"bbb").expect("write b");
2658 std::fs::create_dir_all(temp.join("sub")).expect("create sub");
2659 std::fs::write(temp.join("sub").join("c.txt"), b"ccc").expect("write c");
2660 let artifact = write_directory(&temp).expect("write succeeds");
2661 std::fs::remove_dir_all(&temp).ok();
2662 assert_eq!(artifact.file_count, 3);
2663 assert_eq!(artifact.dir_count, 2);
2664
2665 let mut cursor = ManifestCursor::new(&artifact.bytes);
2666 limnifs_core::parse_manifest_header(&mut cursor).expect("header");
2667 limnifs_core::parse_feature_flags_section(&mut cursor).expect("flags");
2668 let meta_ref = limnifs_core::parse_metadata_reference(&mut cursor).expect("meta ref");
2669 assert!(meta_ref.is_inlined());
2670 let slab_index = limnifs_core::parse_slab_index(&mut cursor).expect("slab index");
2671 assert_eq!(slab_index.len(), 0);
2672 limnifs_core::parse_history(&mut cursor).expect("history");
2673 }
2674
2675 #[test]
2676 fn write_deterministic() {
2677 let temp =
2678 std::env::temp_dir().join(format!("limnifs-write-test-{}-det", std::process::id()));
2679 std::fs::create_dir_all(&temp).expect("create temp dir");
2680 std::fs::write(temp.join("x.txt"), b"xxx").expect("write x");
2681
2682 let a1 = write_directory(&temp).expect("first write");
2683 let a2 = write_directory(&temp).expect("second write");
2684 std::fs::remove_dir_all(&temp).ok();
2685
2686 assert_eq!(a1.bytes, a2.bytes);
2687 assert_eq!(a1.merkle_root, a2.merkle_root);
2688 }
2689
2690 #[test]
2691 fn slab_parses_correctly() {
2692 let temp =
2693 std::env::temp_dir().join(format!("limnifs-write-test-{}-slab", std::process::id()));
2694 std::fs::create_dir_all(&temp).expect("create temp dir");
2695 std::fs::write(temp.join("big.bin"), vec![0x11u8; INLINE_THRESHOLD + 1])
2696 .expect("write big");
2697 let artifact = write_directory(&temp).expect("write succeeds");
2698 std::fs::remove_dir_all(&temp).ok();
2699
2700 let slab_bytes = &artifact.slabs[0].bytes;
2701 let mut cursor = ManifestCursor::new(slab_bytes);
2702 let slab_header = limnifs_core::parse_slab_header(&mut cursor).expect("slab header parses");
2703 assert_eq!(
2704 slab_header.format_version,
2705 limnifs_core::slab::SLAB_FORMAT_VERSION
2706 );
2707 assert!(!slab_header.is_sealed());
2708 assert!(!slab_header.has_erasure_coding());
2709
2710 let drop_record =
2711 limnifs_core::parse_drop_record(&mut cursor, &slab_header).expect("drop record parses");
2712 assert_eq!(drop_record.plaintext_len as usize, INLINE_THRESHOLD + 1);
2713 }
2714
2715 #[test]
2716 fn fastcdc_produces_multiple_chunks_for_large_files() {
2717 let temp = std::env::temp_dir().join(format!(
2720 "limnifs-write-test-{}-cdc-multi",
2721 std::process::id()
2722 ));
2723 std::fs::create_dir_all(&temp).expect("create temp dir");
2724 let data = pseudo_random_bytes(42, 1024 * 1024);
2725 std::fs::write(temp.join("big.bin"), &data).expect("write big");
2726 let artifact = write_directory(&temp).expect("write succeeds");
2727 std::fs::remove_dir_all(&temp).ok();
2728 assert!(
2729 artifact.drop_count > 1,
2730 "expected FastCDC to produce multiple drops for 1 MiB input, got {}",
2731 artifact.drop_count
2732 );
2733 }
2734
2735 #[test]
2736 fn fastcdc_deduplicates_shared_substrings() {
2737 let temp = std::env::temp_dir().join(format!(
2741 "limnifs-write-test-{}-cdc-dedup",
2742 std::process::id()
2743 ));
2744 std::fs::create_dir_all(&temp).expect("create temp dir");
2745 let shared = pseudo_random_bytes(7, 512 * 1024);
2746 let mut a = Vec::with_capacity(shared.len() + 1024);
2747 a.extend_from_slice(&pseudo_random_bytes(1, 1024));
2748 a.extend_from_slice(&shared);
2749 let mut b = Vec::with_capacity(shared.len() + 2048);
2750 b.extend_from_slice(&pseudo_random_bytes(2, 2048));
2751 b.extend_from_slice(&shared);
2752 std::fs::write(temp.join("a.bin"), &a).expect("write a");
2753 std::fs::write(temp.join("b.bin"), &b).expect("write b");
2754
2755 let temp_a = std::env::temp_dir().join(format!(
2757 "limnifs-write-test-{}-cdc-dedup-a",
2758 std::process::id()
2759 ));
2760 std::fs::create_dir_all(&temp_a).expect("create temp_a");
2761 std::fs::write(temp_a.join("a.bin"), &a).expect("write a");
2762 let artifact_a = write_directory(&temp_a).expect("a writes");
2763 std::fs::remove_dir_all(&temp_a).ok();
2764
2765 let temp_b = std::env::temp_dir().join(format!(
2766 "limnifs-write-test-{}-cdc-dedup-b",
2767 std::process::id()
2768 ));
2769 std::fs::create_dir_all(&temp_b).expect("create temp_b");
2770 std::fs::write(temp_b.join("b.bin"), &b).expect("write b");
2771 let artifact_b = write_directory(&temp_b).expect("b writes");
2772 std::fs::remove_dir_all(&temp_b).ok();
2773
2774 let artifact_both = write_directory(&temp).expect("both write");
2775 std::fs::remove_dir_all(&temp).ok();
2776
2777 let sum_alone = artifact_a.drop_count + artifact_b.drop_count;
2778 assert!(
2779 artifact_both.drop_count < sum_alone,
2780 "expected dedup win: both together = {} drops, sum alone = {} drops",
2781 artifact_both.drop_count,
2782 sum_alone
2783 );
2784 }
2785
2786 #[test]
2787 fn slab_splits_when_content_exceeds_ceiling() {
2788 let temp =
2794 std::env::temp_dir().join(format!("limnifs-write-test-{}-split", std::process::id()));
2795 std::fs::create_dir_all(&temp).expect("create temp dir");
2796 for i in 0..7u32 {
2797 let data = pseudo_random_bytes(u64::from(i), 10 * 1024 * 1024);
2799 std::fs::write(temp.join(format!("big-{i}.bin")), &data).expect("write big");
2800 }
2801 let artifact = write_directory(&temp).expect("write succeeds");
2802 std::fs::remove_dir_all(&temp).ok();
2803
2804 assert!(
2806 artifact.slabs.len() >= 2,
2807 "expected at least 2 slabs for 70 MiB of incompressible data, got {}",
2808 artifact.slabs.len()
2809 );
2810 for slab in &artifact.slabs {
2811 assert!(
2812 slab.bytes.len() <= MAX_SLAB_TOTAL_BYTES,
2813 "slab {} is {} bytes (> {} ceiling)",
2814 slab.id.ordinal,
2815 slab.bytes.len(),
2816 MAX_SLAB_TOTAL_BYTES,
2817 );
2818 }
2819 let total_drop_ids: usize = artifact.slabs.iter().map(|s| s.drop_ids.len()).sum();
2821 assert_eq!(
2822 total_drop_ids, artifact.drop_count,
2823 "drop_ids count across slabs must match WriteArtifact.drop_count",
2824 );
2825 }
2826}