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;
30pub mod flatten;
31#[cfg(feature = "pipeline-parallelism")]
32pub mod pipeline;
33pub mod rw;
34#[cfg(feature = "sparse-index")]
35pub mod sparse_index;
36pub mod turnover;
37
38pub use config::{
39 profile, CategorizerConfig, ChunkingConfig, CodecRegistry, CodecTunables, Defaults,
40 DictionaryConfig, EncryptionConfig, TournamentConfig, WriteConfig,
41};
42
43use std::collections::{HashMap, HashSet};
44use std::path::{Path, PathBuf};
45
46use crate::chunker::FastCDC;
47use limnifs_core::codec::CODEC_REFERENCED;
48use limnifs_core::slab_store::SlabStore;
49use limnifs_core::{
50 compute_merkle_root, hash_empty_section, hash_section, parse_manifest_header, parse_slab_index,
51 ManifestCursor, ManifestHeader, SectionHashes, FEATURE_FLAGS_SECTION_VERSION,
52 HISTORY_SECTION_VERSION, INODE_FLAG_INLINE_DATA, INODE_FLAG_SHARED_INLINE,
53 METADATA_REFERENCE_SECTION_VERSION_2, SLAB_INDEX_SECTION_VERSION,
54};
55use limnifs_format::{ManifestRoot, SlabId};
56
57pub const INLINE_THRESHOLD: usize = 4096;
60
61pub const MMAP_READ_THRESHOLD: usize = 1024 * 1024;
67
68pub const WHOLE_FILE_MAX_SIZE: usize = 64 * 1024 * 1024;
73
74pub const MAX_SLAB_TOTAL_BYTES: usize = 60 * 1024 * 1024;
79
80const SLAB_HEADER_LEN: usize = 56;
84
85pub const METADATA_EXTERNALIZE_THRESHOLD: usize =
93 limnifs_core::metadata_reference::DEFAULT_INLINE_METADATA_MAX_BYTES as usize - 24 * 1024;
94
95pub const METADATA_LARGE_BLOB_THRESHOLD: usize = 256 * 1024;
100
101pub const METADATA_SMALL_BLOB_QUALITY: i32 = 5;
104
105pub const METADATA_LARGE_BLOB_QUALITY: i32 = 2;
110
111#[derive(Clone, Debug)]
114pub struct SlabArtifact {
115 pub id: SlabId,
116 pub bytes: Vec<u8>,
117 pub locator: String,
118 pub drop_ids: Vec<[u8; 32]>,
122}
123
124#[derive(Clone, Debug)]
128pub struct MetadataSidecar {
129 pub bytes: Vec<u8>,
130 pub locator: String,
131}
132
133#[derive(Clone, Debug)]
135pub struct WriteArtifact {
136 pub bytes: Vec<u8>,
137 pub merkle_root: ManifestRoot,
138 pub slabs: Vec<SlabArtifact>,
141 pub metadata_sidecar: Option<MetadataSidecar>,
145 pub inode_count: usize,
146 pub file_count: usize,
147 pub dir_count: usize,
148 pub drop_count: usize,
149 pub root_inode_number: u64,
154}
155
156impl WriteArtifact {
157 #[must_use]
161 pub fn slab_bytes(&self) -> Option<&[u8]> {
162 if self.slabs.len() == 1 {
163 Some(&self.slabs[0].bytes)
164 } else {
165 None
166 }
167 }
168
169 #[must_use]
171 pub fn slab_locator(&self) -> Option<&str> {
172 if self.slabs.len() == 1 {
173 Some(&self.slabs[0].locator)
174 } else {
175 None
176 }
177 }
178}
179
180#[derive(Debug)]
182pub enum WriteError {
183 Io(std::io::Error),
184}
185
186impl std::fmt::Display for WriteError {
187 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
188 match self {
189 Self::Io(e) => write!(f, "I/O error: {e}"),
190 }
191 }
192}
193
194impl std::error::Error for WriteError {}
195
196impl From<std::io::Error> for WriteError {
197 fn from(e: std::io::Error) -> Self {
198 Self::Io(e)
199 }
200}
201
202pub fn write_directory(root: &Path) -> Result<WriteArtifact, WriteError> {
216 write_directory_with_config(root, &WriteConfig::default_v0_1())
217}
218
219pub fn write_stream<R: std::io::Read>(
235 name: &str,
236 reader: R,
237 config: &WriteConfig,
238) -> Result<WriteArtifact, WriteError> {
239 let mut ctx = WriteContext::new();
240 ctx.categorizers_disabled = config.categorizers.is_empty();
241 ctx.rw_mode = matches!(config.mode, crate::config::ImageMode::ReadWrite(_));
242 ctx.auto_turnover = config.turnover_threshold > 0;
243 ctx.collect_dict_samples = config.dictionaries.enabled;
244
245 let drop_id_root = [0u8; 32]; let pending = PendingFile {
250 path: std::path::PathBuf::from(name),
251 inode_number: 1,
252 file_len: 0, mtime_ns: 0,
254 };
255 ctx.pending_files.push(pending);
256 ctx.root_inode_number = 1;
257
258 let chunker = ctx.chunker.clone();
260 let chunks = chunker.chunk_reader(reader)?;
261
262 let total_len: u64 = chunks.iter().map(|c| c.len() as u64).sum();
264
265 let text_codec = config.text_codec_id().unwrap_or(0x04);
269 let binary_codec = config.binary_codec_id().unwrap_or(0x01);
270 let tunables = config.to_core_tunables();
271 let classifier = ctx.classifier;
272 let registry = config
273 .codec_registry()
274 .map_err(|e| WriteError::Io(std::io::Error::other(format!("codec registry: {e}"))))?;
275 let tournament_codec_ids: Vec<u8> = config
276 .tournament
277 .codecs
278 .iter()
279 .filter_map(|n| registry.lookup_by_name(n))
280 .collect();
281 let tournament = TournamentSpec {
282 codec_ids: tournament_codec_ids,
283 min_size: config.tournament.min_size_threshold as usize,
284 skip_for_binary: config.tournament.skip_for_binary,
285 short_circuit_permille: config.tournament.short_circuit_threshold,
286 };
287
288 let mut drops: Vec<RawDrop> = Vec::with_capacity(chunks.len());
289 let mut slices: Vec<PendingSlice> = Vec::with_capacity(chunks.len());
290 let mut offset: u64 = 0;
291 for chunk in &chunks {
292 let drop_id = hash_section(chunk);
293 slices.push(PendingSlice {
294 drop_id,
295 file_byte_start: offset,
296 file_byte_end: offset + chunk.len() as u64,
297 });
298 offset += chunk.len() as u64;
299 let class = classifier.classify(chunk);
300 let (codec_id, compressed) = compress_chunk_with_tournament(
301 chunk,
302 class,
303 text_codec,
304 binary_codec,
305 &tunables,
306 &tournament,
307 );
308 drops.push((drop_id, chunk.clone(), compressed, codec_id));
309 }
310 let _ = drop_id_root;
311
312 let result = ChunkedFileResult { drops, slices };
314 let pf = ctx.pending_files[0].clone();
315 ctx.merge_chunked_file(&pf, result);
316 ctx.pending_files[0].file_len = total_len;
318 if let Some(inode) = ctx.inodes.last_mut() {
321 if let PendingContent::DropBacked { file_len, .. } = &mut inode.content {
322 *file_len = total_len;
323 }
324 }
325
326 ctx.train_and_apply_dictionary(&config.dictionaries);
327 let artifact = ctx.assemble();
328 Ok(artifact)
329}
330
331pub fn write_layer(
372 base_image: &Path,
373 root: &Path,
374 config: &WriteConfig,
375) -> Result<WriteArtifact, WriteError> {
376 let (base_drop_index, base_root) = load_base_drop_index(base_image)?;
378
379 let mut ctx = WriteContext::new();
380 ctx.categorizers_disabled = config.categorizers.is_empty();
381 ctx.rw_mode = matches!(config.mode, crate::config::ImageMode::ReadWrite(_));
382 ctx.auto_turnover = config.turnover_threshold > 0;
383 ctx.collect_dict_samples = config.dictionaries.enabled;
384 ctx.inline_threshold = config.defaults.inline_threshold as usize;
385 ctx.metadata_externalize_threshold = config.defaults.metadata_externalize_threshold;
386 ctx.emit_shared_inline = config.defaults.shared_inline;
387 ctx.base_drop_index = Some(base_drop_index);
388 ctx.base_root = Some(base_root);
389
390 let root_inode_number = ctx.walk(root)?;
392 ctx.root_inode_number = root_inode_number;
393 write_directory_body(&mut ctx, config)?;
394 Ok(ctx.assemble())
395}
396
397fn load_base_drop_index(
401 base_image: &Path,
402) -> Result<(std::collections::HashSet<[u8; 32]>, [u8; 32]), WriteError> {
403 let manifest_bytes = std::fs::read(base_image)?;
404 let mut cursor = ManifestCursor::new(&manifest_bytes);
405 let _ = parse_manifest_header(&mut cursor).map_err(io_core)?;
406 let _ = limnifs_core::parse_feature_flags_section(&mut cursor);
408 let _ = limnifs_core::parse_metadata_reference(&mut cursor);
409 let slab_index = parse_slab_index(&mut cursor).map_err(io_core)?;
410 let store = SlabStore::load_mmap(base_image, &slab_index).map_err(io_core)?;
411 let drop_set: std::collections::HashSet<[u8; 32]> = store.drop_index_keys().copied().collect();
412 let root = *compute_merkle_root_from_sections(&manifest_bytes).as_bytes();
418 Ok((drop_set, root))
419}
420
421fn compute_merkle_root_from_sections(manifest: &[u8]) -> ManifestRoot {
427 use limnifs_core::SectionHashes;
428 let mut cursor = ManifestCursor::new(manifest);
429 let header_start = 0;
430 if parse_manifest_header(&mut cursor).is_err() {
431 return ManifestRoot::from_bytes([0u8; 32]);
433 }
434 let header_end = cursor.position();
435 let flags_start = header_end;
437 let flags_end = match limnifs_core::parse_feature_flags_section(&mut cursor) {
438 Ok(_) => cursor.position(),
439 Err(_) => flags_start,
440 };
441 let meta_ref_start = flags_end;
442 let metadata_reference = match limnifs_core::parse_metadata_reference(&mut cursor) {
443 Ok(m) => Some(m),
444 Err(_) => None,
445 };
446 let meta_ref_end = cursor.position();
447 let slab_index_start = meta_ref_end;
448 let _ = parse_slab_index(&mut cursor);
449 let slab_index_end = cursor.position();
450 let history_start = slab_index_end;
451 let _ = limnifs_core::parse_history(&mut cursor);
452 let history_end = cursor.position();
453
454 let hashes = SectionHashes {
455 metadata: metadata_reference
456 .map(|m| m.metadata_hash)
457 .unwrap_or_else(hash_empty_section),
458 format_header: hash_section(&manifest[header_start..header_end]),
459 feature_flags: hash_section(&manifest[flags_start..flags_end]),
460 metadata_reference: hash_section(&manifest[meta_ref_start..meta_ref_end]),
461 slab_index: hash_section(&manifest[slab_index_start..slab_index_end]),
462 crypto_params: hash_empty_section(),
463 ec_params: hash_empty_section(),
464 dms_policy: hash_empty_section(),
465 delta_linkage: hash_empty_section(),
466 history: hash_section(&manifest[history_start..history_end]),
467 };
468 compute_merkle_root(&hashes)
469}
470
471fn io_core(e: limnifs_core::CoreError) -> WriteError {
472 WriteError::Io(std::io::Error::other(format!("base image load: {e}")))
473}
474
475fn write_directory_body(ctx: &mut WriteContext, config: &WriteConfig) -> Result<(), WriteError> {
480 use rayon::prelude::*;
481
482 ctx.metadata_codec = config
483 .metadata_codec_id()
484 .unwrap_or(limnifs_core::codec::CODEC_BROTLI);
485
486 let pending = std::mem::take(&mut ctx.pending_files);
487 if pending.is_empty() {
488 return Ok(());
489 }
490 ctx.inline_threshold = config.defaults.inline_threshold as usize;
491 ctx.metadata_externalize_threshold = config.defaults.metadata_externalize_threshold;
492 ctx.emit_shared_inline = config.defaults.shared_inline;
493 let chunker = ctx.chunker.clone();
494 let classifier = ctx.classifier;
495 let text_codec = config.text_codec_id().unwrap_or(0x04);
496 let binary_codec = config.binary_codec_id().unwrap_or(0x01);
497 let tunables = config.to_core_tunables();
498 let use_categorizers = !config.categorizers.is_empty();
499 let skip_chunking = config.skip_chunking;
500 let registry = config
501 .codec_registry()
502 .map_err(|e| WriteError::Io(std::io::Error::other(format!("codec registry: {e}"))))?;
503 let tournament_codec_ids: Vec<u8> = config
504 .tournament
505 .codecs
506 .iter()
507 .filter_map(|n| registry.lookup_by_name(n))
508 .collect();
509 let tournament_spec = TournamentSpec {
510 codec_ids: tournament_codec_ids,
511 min_size: config.tournament.min_size_threshold as usize,
512 skip_for_binary: config.tournament.skip_for_binary,
513 short_circuit_permille: config.tournament.short_circuit_threshold,
514 };
515 let base_drop_index = ctx.base_drop_index.as_ref();
516 let inline_threshold = ctx.inline_threshold;
517 let results: Vec<ChunkedFileResult> = pending
518 .par_iter()
519 .map(|pf| {
520 process_file(
521 pf,
522 &chunker,
523 classifier,
524 text_codec,
525 binary_codec,
526 &tunables,
527 use_categorizers,
528 skip_chunking,
529 &tournament_spec,
530 base_drop_index,
531 inline_threshold,
532 )
533 })
534 .collect::<Result<Vec<_>, _>>()?;
535
536 for (pf, result) in pending.iter().zip(results) {
537 ctx.merge_chunked_file(pf, result);
538 }
539 ctx.train_and_apply_dictionary(&config.dictionaries);
540 Ok(())
541}
542
543pub fn write_directory_with_config(
545 root: &Path,
546 config: &WriteConfig,
547) -> Result<WriteArtifact, WriteError> {
548 let mut ctx = WriteContext::new();
549 ctx.categorizers_disabled = config.categorizers.is_empty();
550 ctx.rw_mode = matches!(config.mode, crate::config::ImageMode::ReadWrite(_));
551 ctx.auto_turnover = config.turnover_threshold > 0;
552 ctx.collect_dict_samples = config.dictionaries.enabled;
553
554 write_directory_streaming(&mut ctx, root, config)?;
555 Ok(ctx.assemble())
556}
557
558fn write_directory_streaming(
572 ctx: &mut WriteContext,
573 root: &Path,
574 config: &WriteConfig,
575) -> Result<(), WriteError> {
576 use rayon::prelude::*;
577
578 ctx.metadata_codec = config
579 .metadata_codec_id()
580 .unwrap_or(limnifs_core::codec::CODEC_BROTLI);
581
582 let chunker = ctx.chunker.clone();
583 let classifier = ctx.classifier;
584 let text_codec = config.text_codec_id().unwrap_or(0x04);
585 let binary_codec = config.binary_codec_id().unwrap_or(0x01);
586 let tunables = config.to_core_tunables();
587 let use_categorizers = !config.categorizers.is_empty();
588 let skip_chunking = config.skip_chunking;
589 let registry = config
590 .codec_registry()
591 .map_err(|e| WriteError::Io(std::io::Error::other(format!("codec registry: {e}"))))?;
592 let tournament_codec_ids: Vec<u8> = config
593 .tournament
594 .codecs
595 .iter()
596 .filter_map(|n| registry.lookup_by_name(n))
597 .collect();
598 let tournament_spec = TournamentSpec {
599 codec_ids: tournament_codec_ids,
600 min_size: config.tournament.min_size_threshold as usize,
601 skip_for_binary: config.tournament.skip_for_binary,
602 short_circuit_permille: config.tournament.short_circuit_threshold,
603 };
604 let base_drop_index = ctx.base_drop_index.clone();
607 let inline_threshold = ctx.inline_threshold;
608
609 ctx.inline_threshold = config.defaults.inline_threshold as usize;
610 ctx.metadata_externalize_threshold = config.defaults.metadata_externalize_threshold;
611 ctx.emit_shared_inline = config.defaults.shared_inline;
612
613 const PIPELINE_CAPACITY: usize = 256;
617 let (tx, rx) = std::sync::mpsc::sync_channel::<PendingFile>(PIPELINE_CAPACITY);
618 ctx.pending_sink = Some(tx);
619
620 let (root_inode_number, mut results): (
621 u64,
622 Vec<(usize, PendingFile, Result<ChunkedFileResult, WriteError>)>,
623 ) = std::thread::scope(|scope| {
624 let producer = {
625 let ctx = &mut *ctx;
626 let root = root;
627 scope.spawn(move || {
628 let r = ctx.walk(root);
629 ctx.pending_sink = None;
633 r
634 })
635 };
636 let results = rx
639 .into_iter()
640 .enumerate()
641 .par_bridge()
642 .map(|(i, pf)| {
643 let r = process_file(
644 &pf,
645 &chunker,
646 classifier,
647 text_codec,
648 binary_codec,
649 &tunables,
650 use_categorizers,
651 skip_chunking,
652 &tournament_spec,
653 base_drop_index.as_ref(),
654 inline_threshold,
655 );
656 (i, pf, r)
657 })
658 .collect();
659 let joined = producer
660 .join()
661 .unwrap_or_else(|_| {
662 Err(WriteError::Io(std::io::Error::other(
663 "walk thread panicked",
664 )))
665 })
666 .map(|n| (n, results));
667 joined
670 })?;
671 ctx.pending_sink = None;
672 ctx.root_inode_number = root_inode_number;
673
674 results.sort_unstable_by_key(|(i, _, _)| *i);
675 for (_, pf, r) in results {
679 ctx.merge_chunked_file(&pf, r?);
680 }
681 ctx.train_and_apply_dictionary(&config.dictionaries);
682 Ok(())
683}
684
685pub(crate) type RawDrop = ([u8; 32], Vec<u8>, std::sync::Arc<[u8]>, u8);
692pub(crate) struct ChunkedFileResult {
694 drops: Vec<RawDrop>, slices: Vec<PendingSlice>,
696}
697
698struct TournamentSpec {
706 codec_ids: Vec<u8>,
710 min_size: usize,
714 skip_for_binary: bool,
718 short_circuit_permille: u32,
723}
724
725fn process_whole_file_drop(
738 pf: &PendingFile,
739 data: &[u8],
740 cat: file_categorizer::Categorization,
741 tunables: &limnifs_core::codec::CodecTunables,
742) -> Result<ChunkedFileResult, WriteError> {
743 let _ = pf;
744 let drop_id = hash_section(data);
745 let file_len = u64::try_from(data.len()).unwrap_or(u64::MAX);
746
747 let (mut best_codec, mut best_compressed): (u8, std::sync::Arc<[u8]>) =
752 match limnifs_core::codec::compress_with_tunables(
753 limnifs_core::codec::CODEC_BROTLI,
754 data,
755 tunables,
756 ) {
757 Ok(c) => (limnifs_core::codec::CODEC_BROTLI, c.into()),
758 Err(_) => match limnifs_core::codec::compress_with_tunables(
759 limnifs_core::codec::CODEC_ZSTD,
760 data,
761 tunables,
762 ) {
763 Ok(c) => (limnifs_core::codec::CODEC_ZSTD, c.into()),
764 Err(_) => (limnifs_core::codec::CODEC_STORE, data.to_vec().into()),
765 },
766 };
767
768 let brotli_ratio = best_compressed.len() as f64 / data.len() as f64;
772 if brotli_ratio > 0.05 {
773 if let Ok(zstd_c) = limnifs_core::codec::compress_with_tunables(
774 limnifs_core::codec::CODEC_ZSTD,
775 data,
776 tunables,
777 ) {
778 if zstd_c.len() < best_compressed.len() {
779 best_codec = limnifs_core::codec::CODEC_ZSTD;
780 best_compressed = zstd_c.into();
781 }
782 }
783 }
784
785 let general_ratio = best_compressed.len() as f64 / data.len() as f64;
791 if general_ratio > 0.15 || cat.codec_id == limnifs_core::codec::CODEC_RICEPP {
792 let spec_result = if cat.codec_id == limnifs_core::codec::CODEC_FSST_BROTLI {
796 limnifs_core::codec::fsst_brotli::compress_with_baseline(data, Some(&best_compressed))
797 } else {
798 limnifs_core::codec::compress(cat.codec_id, data)
799 };
800 if let Ok(spec_c) = spec_result {
801 if spec_c.len() < best_compressed.len() {
802 best_codec = cat.codec_id;
803 best_compressed = spec_c.into();
804 }
805 }
806 }
807
808 Ok(ChunkedFileResult {
809 drops: vec![(drop_id, data.to_vec(), best_compressed, best_codec)],
810 slices: vec![PendingSlice {
811 drop_id,
812 file_byte_start: 0,
813 file_byte_end: file_len,
814 }],
815 })
816}
817
818fn compress_chunk_with_tournament(
840 chunk: &[u8],
841 class: classifier::Class,
842 text_codec: u8,
843 binary_codec: u8,
844 tunables: &limnifs_core::codec::CodecTunables,
845 tournament: &TournamentSpec,
846) -> (u8, std::sync::Arc<[u8]>) {
847 use classifier::Class;
848
849 let preferred = match class {
850 Class::Binary => binary_codec,
851 Class::Text | Class::Code | Class::Sparse => text_codec,
852 _ => limnifs_core::codec::CODEC_STORE,
853 };
854
855 if preferred == limnifs_core::codec::CODEC_STORE {
856 return (limnifs_core::codec::CODEC_STORE, chunk.to_vec().into());
857 }
858 if class == Class::Binary && tournament.skip_for_binary {
859 return compress_chunk_one(chunk, preferred, tunables);
860 }
861 if chunk.len() < tournament.min_size {
862 return compress_chunk_one(chunk, preferred, tunables);
863 }
864
865 let mut best: Option<(u8, std::sync::Arc<[u8]>)> = None;
866 for &codec_id in &tournament.codec_ids {
867 if codec_id == limnifs_core::codec::CODEC_STORE {
868 continue;
869 }
870 let c = match limnifs_core::codec::compress_with_tunables(codec_id, chunk, tunables) {
871 Ok(c) => c,
872 Err(_) => continue,
873 };
874 if c.len() >= chunk.len() {
875 continue;
876 }
877 let ratio_permille = (c.len() as u64 * 1000 / chunk.len() as u64) as u32;
878 let is_best_so_far = best.as_ref().map_or(true, |(_, b)| c.len() < b.len());
879 if is_best_so_far {
880 best = Some((codec_id, c.into()));
881 }
882 if tournament.short_circuit_permille > 0
883 && ratio_permille <= tournament.short_circuit_permille
884 {
885 break;
886 }
887 }
888
889 best.unwrap_or_else(|| (limnifs_core::codec::CODEC_STORE, chunk.to_vec().into()))
890}
891
892fn compress_chunk_one(
895 chunk: &[u8],
896 codec_id: u8,
897 tunables: &limnifs_core::codec::CodecTunables,
898) -> (u8, std::sync::Arc<[u8]>) {
899 if codec_id == limnifs_core::codec::CODEC_STORE {
900 return (limnifs_core::codec::CODEC_STORE, chunk.to_vec().into());
901 }
902 match limnifs_core::codec::compress_with_tunables(codec_id, chunk, tunables) {
903 Ok(c) if c.len() < chunk.len() => (codec_id, c.into()),
904 _ => (limnifs_core::codec::CODEC_STORE, chunk.to_vec().into()),
905 }
906}
907
908fn process_file(
912 pf: &PendingFile,
913 chunker: &FastCDC,
914 classifier: classifier::Classifier,
915 text_codec: u8,
916 binary_codec: u8,
917 tunables: &limnifs_core::codec::CodecTunables,
918 use_categorizers: bool,
919 skip_chunking: bool,
920 tournament: &TournamentSpec,
921 base_drop_index: Option<&std::collections::HashSet<[u8; 32]>>,
922 inline_threshold: usize,
923) -> Result<ChunkedFileResult, WriteError> {
924 let file_len_estimate = std::fs::metadata(&pf.path)
935 .map(|m| m.len() as usize)
936 .unwrap_or(0);
937 let data: Vec<u8> = if file_len_estimate >= MMAP_READ_THRESHOLD {
938 let file = std::fs::File::open(&pf.path)?;
939 #[allow(unsafe_code)]
940 let mmap = unsafe { memmap2::Mmap::map(&file) }.map_err(WriteError::Io)?;
941 Vec::from(&mmap[..])
945 } else {
946 std::fs::read(&pf.path)?
947 };
948 let file_len = data.len();
949
950 if skip_chunking && file_len > inline_threshold {
957 let drop_id = hash_section(&data);
958 let class = classifier.classify(&data);
959 let preferred_codec = match class {
960 classifier::Class::Binary => binary_codec,
961 _ => text_codec,
962 };
963 let (codec_id, compressed): (u8, std::sync::Arc<[u8]>) =
964 match limnifs_core::codec::compress_with_tunables(preferred_codec, &data, tunables) {
965 Ok(c) if c.len() < data.len() => (preferred_codec, c.into()),
966 _ => (limnifs_core::codec::CODEC_STORE, data.to_vec().into()),
967 };
968 return Ok(ChunkedFileResult {
969 drops: vec![(drop_id, data, compressed, codec_id)],
970 slices: vec![PendingSlice {
971 drop_id,
972 file_byte_start: 0,
973 file_byte_end: file_len as u64,
974 }],
975 });
976 }
977
978 if use_categorizers {
979 if let Some(cat) = file_categorizer::default_registry().categorize(&pf.path, &data) {
980 let needs_whole_file = matches!(
981 cat.codec_id,
982 limnifs_core::codec::CODEC_FLAC | limnifs_core::codec::CODEC_RICEPP
983 );
984 if needs_whole_file || file_len <= WHOLE_FILE_MAX_SIZE {
985 return process_whole_file_drop(pf, &data, cat, tunables);
986 }
987 }
988 }
989
990 let chunks = chunker.chunk_slice(&data);
991 let mut slices = Vec::with_capacity(chunks.len());
992 let mut file_offset: u64 = 0;
993 let mut seen_in_file: std::collections::HashSet<[u8; 32]> =
994 std::collections::HashSet::with_capacity(chunks.len());
995
996 let mut unique_chunks: Vec<(&[u8], [u8; 32])> = Vec::with_capacity(chunks.len());
999 for chunk in &chunks {
1000 let chunk_len = u64::try_from(chunk.len()).expect("chunk len fits u64");
1001 let drop_id = hash_section(chunk);
1002 slices.push(PendingSlice {
1003 drop_id,
1004 file_byte_start: file_offset,
1005 file_byte_end: file_offset + chunk_len,
1006 });
1007 file_offset += chunk_len;
1008 if seen_in_file.insert(drop_id) {
1009 unique_chunks.push((chunk, drop_id));
1010 }
1011 }
1012
1013 use rayon::prelude::*;
1025 thread_local! {
1026 static COMPRESS_CACHE: std::cell::RefCell<std::collections::HashMap<[u8; 32], (u8, std::sync::Arc<[u8]>)>> =
1027 std::cell::RefCell::new(std::collections::HashMap::new());
1028 }
1029 const COMPRESS_CACHE_MAX_ENTRIES: usize = 100_000;
1030 let drops: Vec<RawDrop> = unique_chunks
1031 .par_iter()
1032 .map(|(chunk, drop_id)| {
1033 if let Some(base) = base_drop_index {
1036 if base.contains(drop_id) {
1037 return (*drop_id, Vec::new(), Vec::new().into(), CODEC_REFERENCED);
1038 }
1039 }
1040 let class = classifier.classify(chunk);
1041 let cached = COMPRESS_CACHE.with(|c| {
1044 c.borrow()
1045 .get(drop_id)
1046 .map(|(cid, comp)| (*cid, comp.clone()))
1047 });
1048 let (codec_id, compressed) = if let Some(c) = cached {
1049 c
1050 } else {
1051 let new = compress_chunk_with_tournament(
1052 chunk,
1053 class,
1054 text_codec,
1055 binary_codec,
1056 tunables,
1057 tournament,
1058 );
1059 COMPRESS_CACHE.with(|c| {
1061 let mut cache = c.borrow_mut();
1062 if cache.len() < COMPRESS_CACHE_MAX_ENTRIES {
1063 cache.insert(*drop_id, new.clone());
1065 }
1066 });
1067 new
1068 };
1069 (*drop_id, chunk.to_vec(), compressed, codec_id)
1070 })
1071 .collect();
1072
1073 let _ = file_len;
1074 Ok(ChunkedFileResult { drops, slices })
1075}
1076
1077struct PendingDrop {
1078 id: [u8; 32],
1079 plaintext_len: u32,
1085 compressed: std::sync::Arc<[u8]>,
1086 codec: u8,
1087 dict_id: u8,
1091 plaintext: Option<Vec<u8>>,
1096}
1097
1098impl PendingDrop {
1099 fn len_in_window(&self) -> u32 {
1103 u32::try_from(self.compressed.len()).expect("compressed fits u32")
1104 }
1105
1106 fn plaintext_len_value(&self) -> u32 {
1108 self.plaintext_len
1109 }
1110
1111 fn slab_footprint(&self) -> usize {
1114 48 + self.compressed.len()
1115 }
1116}
1117
1118struct PendingSlice {
1123 drop_id: [u8; 32],
1124 file_byte_start: u64,
1125 file_byte_end: u64,
1126}
1127
1128#[derive(Clone)]
1131struct PendingFile {
1132 inode_number: u64,
1133 path: PathBuf,
1134 mtime_ns: u64,
1135 file_len: u64,
1136}
1137
1138struct PendingInode {
1139 number: u64,
1140 mode: u32,
1141 mtime_ns: u64,
1142 content: PendingContent,
1143}
1144
1145enum PendingContent {
1146 Inline(Vec<u8>),
1147 DropBacked {
1148 file_len: u64,
1149 slices: Vec<PendingSlice>,
1150 },
1151 Directory(Vec<(String, u64, u8)>),
1152}
1153
1154struct DirNode {
1155 entries: Vec<(String, u64, u8)>,
1156 bytes: Vec<u8>,
1157 hash: [u8; 32],
1158}
1159
1160struct WriteContext {
1161 next_inode: u64,
1162 inodes: Vec<PendingInode>,
1163 dir_nodes: Vec<DirNode>,
1164 drops: Vec<PendingDrop>,
1165 drop_index: HashSet<[u8; 32]>,
1166 pending_files: Vec<PendingFile>,
1167 file_count: usize,
1168 dir_count: usize,
1169 root_inode_number: u64,
1170 chunker: FastCDC,
1171 classifier: classifier::Classifier,
1172 shared_inline_map: HashMap<[u8; 32], usize>,
1173 shared_inline_table: Vec<Vec<u8>>,
1174 profile_name: Option<String>,
1176 metadata_codec: u8,
1179 categorizers_disabled: bool,
1181 rw_mode: bool,
1183 auto_turnover: bool,
1185 collect_dict_samples: bool,
1188 dict_samples_by_class: HashMap<crate::classifier::Class, Vec<Vec<u8>>>,
1195 trained_dicts_by_class: HashMap<crate::classifier::Class, crate::dictionary::TrainedDictionary>,
1200 base_drop_index: Option<HashSet<[u8; 32]>>,
1206 base_root: Option<[u8; 32]>,
1211 metadata_externalize_threshold: usize,
1216 emit_shared_inline: bool,
1222 inline_threshold: usize,
1227 pending_sink: Option<std::sync::mpsc::SyncSender<PendingFile>>,
1233}
1234
1235impl WriteContext {
1236 const MAX_DICT_SAMPLES: usize = 1000;
1239
1240 fn new() -> Self {
1241 Self {
1242 next_inode: 1,
1243 inodes: Vec::new(),
1244 dir_nodes: Vec::new(),
1245 drops: Vec::new(),
1246 drop_index: HashSet::new(),
1247 pending_files: Vec::new(),
1248 file_count: 0,
1249 dir_count: 0,
1250 root_inode_number: 0,
1251 chunker: FastCDC::default(),
1252 classifier: classifier::Classifier,
1253 shared_inline_map: HashMap::new(),
1254 shared_inline_table: Vec::new(),
1255 profile_name: None,
1256 metadata_codec: limnifs_core::codec::CODEC_BROTLI,
1257 categorizers_disabled: false,
1258 rw_mode: false,
1259 auto_turnover: false,
1260 collect_dict_samples: false,
1261 dict_samples_by_class: HashMap::new(),
1262 trained_dicts_by_class: HashMap::new(),
1263 base_drop_index: None,
1264 base_root: None,
1265 pending_sink: None,
1266 inline_threshold: INLINE_THRESHOLD,
1267 metadata_externalize_threshold: METADATA_EXTERNALIZE_THRESHOLD,
1268 emit_shared_inline: true,
1269 }
1270 }
1271
1272 fn alloc_inode(&mut self) -> u64 {
1273 let n = self.next_inode;
1274 self.next_inode += 1;
1275 n
1276 }
1277
1278 fn build_shared_inline_table(&mut self) {
1282 let mut counts: HashMap<[u8; 32], usize> = HashMap::new();
1283 for inode in &self.inodes {
1284 if let PendingContent::Inline(data) = &inode.content {
1285 let h = hash_section(data);
1286 *counts.entry(h).or_default() += 1;
1287 }
1288 }
1289 for inode in &self.inodes {
1291 if let PendingContent::Inline(data) = &inode.content {
1292 let h = hash_section(data);
1293 if counts.get(&h).copied().unwrap_or(0) > 1
1294 && !self.shared_inline_map.contains_key(&h)
1295 {
1296 let idx = self.shared_inline_table.len();
1297 self.shared_inline_table.push(data.clone());
1298 self.shared_inline_map.insert(h, idx);
1299 }
1300 }
1301 }
1302 }
1303
1304 fn merge_chunked_file(&mut self, pf: &PendingFile, result: ChunkedFileResult) {
1307 for (drop_id, plaintext, compressed, codec) in result.drops {
1308 if self.drop_index.insert(drop_id) {
1309 let retain_plaintext =
1315 self.collect_dict_samples && codec == limnifs_core::codec::CODEC_ZSTD;
1316 if retain_plaintext {
1317 let total: usize = self.dict_samples_by_class.values().map(Vec::len).sum();
1318 if total < Self::MAX_DICT_SAMPLES {
1319 let class = self.classifier.classify(&plaintext);
1320 self.dict_samples_by_class
1321 .entry(class)
1322 .or_default()
1323 .push(plaintext.clone());
1324 }
1325 }
1326 self.drops.push(PendingDrop {
1327 id: drop_id,
1328 plaintext_len: u32::try_from(plaintext.len()).unwrap_or(u32::MAX),
1329 compressed,
1330 codec,
1331 dict_id: limnifs_core::drop_record::NO_DICT,
1332 plaintext: if retain_plaintext {
1333 Some(plaintext)
1334 } else {
1335 None
1336 },
1337 });
1338 }
1339 }
1340 self.inodes.push(PendingInode {
1341 number: pf.inode_number,
1342 mode: 0o100_644,
1343 mtime_ns: pf.mtime_ns,
1344 content: PendingContent::DropBacked {
1345 file_len: pf.file_len,
1346 slices: result.slices,
1347 },
1348 });
1349 }
1350
1351 #[allow(dead_code)]
1362 fn deepen_drop(&self, drop_id: [u8; 32], plaintext: &[u8]) -> PendingDrop {
1363 let class = self.classifier.classify(plaintext);
1364 let (codec, compressed): (u8, std::sync::Arc<[u8]>) = match class {
1365 classifier::Class::Text | classifier::Class::Code | classifier::Class::Binary => {
1366 let c = limnifs_core::codec::compress_lz4_with_size(plaintext);
1367 (limnifs_core::codec::CODEC_LZ4, c.into())
1368 }
1369 _ => (limnifs_core::codec::CODEC_STORE, plaintext.to_vec().into()),
1370 };
1371 PendingDrop {
1372 id: drop_id,
1373 plaintext_len: u32::try_from(plaintext.len()).unwrap_or(u32::MAX),
1374 compressed,
1375 codec,
1376 dict_id: limnifs_core::drop_record::NO_DICT,
1377 plaintext: None,
1378 }
1379 }
1380
1381 fn walk(&mut self, path: &Path) -> Result<u64, WriteError> {
1382 let meta = std::fs::symlink_metadata(path)?;
1383 let file_type = meta.file_type();
1384 let mtime_ns = meta
1385 .modified()
1386 .ok()
1387 .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
1388 .map_or(0u128, |d| d.as_nanos());
1389 let mtime_ns: u64 = mtime_ns.try_into().unwrap_or(0);
1390
1391 if file_type.is_dir() {
1392 self.dir_count += 1;
1393 let inode_number = self.alloc_inode();
1394 let mut entries: Vec<(String, u64, u8)> = Vec::new();
1395
1396 for entry in std::fs::read_dir(path)? {
1397 let entry = entry?;
1398 let name = entry.file_name().to_string_lossy().into_owned();
1399 let child_path = entry.path();
1400 let child_inode = self.walk(&child_path)?;
1401 let child_meta = entry.metadata()?;
1402 let entry_type = if child_meta.is_dir() { 0x02 } else { 0x01 };
1403 entries.push((name, child_inode, entry_type));
1404 }
1405
1406 entries.sort_by(|a, b| a.0.cmp(&b.0));
1407 let dir_node = encode_dir_node(&entries);
1408 self.dir_nodes.push(dir_node);
1409 self.inodes.push(PendingInode {
1410 number: inode_number,
1411 mode: 0o040_755,
1412 mtime_ns,
1413 content: PendingContent::Directory(entries),
1414 });
1415 Ok(inode_number)
1416 } else if file_type.is_file() {
1417 self.file_count += 1;
1418 let inode_number = self.alloc_inode();
1419 let file_len = meta.len();
1420
1421 if file_len <= u64::try_from(self.inline_threshold).unwrap_or(u64::MAX) {
1422 let data = std::fs::read(path)?;
1423 self.inodes.push(PendingInode {
1424 number: inode_number,
1425 mode: 0o100_644,
1426 mtime_ns,
1427 content: PendingContent::Inline(data),
1428 });
1429 } else {
1430 let pf = PendingFile {
1432 inode_number,
1433 path: path.to_path_buf(),
1434 mtime_ns,
1435 file_len,
1436 };
1437 if let Some(sink) = &self.pending_sink {
1438 sink.send(pf).map_err(|_| {
1444 WriteError::Io(std::io::Error::other("walk: compress pipeline shut down"))
1445 })?;
1446 } else {
1447 self.pending_files.push(pf);
1448 }
1449 }
1450 Ok(inode_number)
1451 } else {
1452 Err(WriteError::Io(std::io::Error::new(
1453 std::io::ErrorKind::Unsupported,
1454 format!("unsupported file type: {}", path.display()),
1455 )))
1456 }
1457 }
1458
1459 fn train_and_apply_dictionary(&mut self, dictionaries: &crate::config::DictionaryConfig) {
1474 let cleanup = |ctx: &mut Self| {
1475 for d in &mut ctx.drops {
1476 d.plaintext = None;
1477 }
1478 ctx.dict_samples_by_class.clear();
1479 };
1480
1481 if !dictionaries.enabled {
1482 cleanup(self);
1483 return;
1484 }
1485
1486 let target = usize::try_from(dictionaries.max_dict_size).unwrap_or(65_536);
1487 let min_class = usize::try_from(dictionaries.min_class_size).unwrap_or(0);
1488
1489 let text_classes = [
1493 crate::classifier::Class::Text,
1494 crate::classifier::Class::Code,
1495 crate::classifier::Class::Sparse,
1496 ];
1497 let binary_classes = [crate::classifier::Class::Binary];
1498
1499 let text_samples: Vec<&[u8]> = text_classes
1501 .iter()
1502 .flat_map(|c| self.dict_samples_by_class.get(c).into_iter().flatten())
1503 .map(Vec::as_slice)
1504 .collect();
1505 let trainer = crate::dictionary::TrainerKind::from_config_str(&dictionaries.trainer);
1506 if text_samples.len() >= min_class {
1507 if let Some(dict) =
1508 crate::dictionary::train_zstd_with_trainer(0, &text_samples, target, trainer)
1509 {
1510 self.trained_dicts_by_class
1511 .insert(crate::classifier::Class::Text, dict);
1512 }
1513 }
1514 let binary_samples: Vec<&[u8]> = binary_classes
1515 .iter()
1516 .flat_map(|c| self.dict_samples_by_class.get(c).into_iter().flatten())
1517 .map(Vec::as_slice)
1518 .collect();
1519 if binary_samples.len() >= min_class {
1520 if let Some(dict) =
1521 crate::dictionary::train_zstd_with_trainer(1, &binary_samples, target, trainer)
1522 {
1523 self.trained_dicts_by_class
1524 .insert(crate::classifier::Class::Binary, dict);
1525 }
1526 }
1527
1528 for d in self.drops.iter_mut() {
1531 if d.codec != limnifs_core::codec::CODEC_ZSTD {
1532 continue;
1533 }
1534 let Some(plaintext) = d.plaintext.clone() else {
1535 continue;
1536 };
1537 let class = self.classifier.classify(&plaintext);
1538 let dict_class = if text_classes.contains(&class) {
1539 crate::classifier::Class::Text
1540 } else if binary_classes.contains(&class) {
1541 crate::classifier::Class::Binary
1542 } else {
1543 continue;
1544 };
1545 let Some(dict) = self.trained_dicts_by_class.get(&dict_class) else {
1546 continue;
1547 };
1548 let Ok(dict_compressed) = dict.compress(&plaintext) else {
1549 continue;
1550 };
1551 if dict_compressed.len() < d.compressed.len() {
1552 d.compressed = dict_compressed.into();
1553 d.dict_id = dict.id;
1554 }
1555 }
1556
1557 cleanup(self);
1558 }
1559
1560 fn trace_phase(label: &str, start: std::time::Instant) {
1562 if std::env::var_os("LIMNIFS_TRACE_ASSEMBLE").is_some() {
1563 eprintln!("[assemble] {label}: {:?}", start.elapsed());
1564 }
1565 }
1566
1567 fn assemble(mut self) -> WriteArtifact {
1568 let t_assemble = std::time::Instant::now();
1569 let inode_count = self.inodes.len();
1570 let dir_count = self.dir_count;
1571 let drop_count = self.drops.len();
1572
1573 let t = std::time::Instant::now();
1579 let slabs = pack_slabs(&self.drops);
1580 Self::trace_phase("pack_slabs", t);
1581
1582 let t = std::time::Instant::now();
1586 if self.emit_shared_inline {
1587 self.build_shared_inline_table();
1588 }
1589 Self::trace_phase("shared_inline_table", t);
1590
1591 let mut metadata_blob = Vec::new();
1592 metadata_blob.extend_from_slice(&u32::try_from(self.inodes.len()).unwrap().to_le_bytes());
1593 for inode in &self.inodes {
1594 self.encode_inode(&mut metadata_blob, inode);
1595 }
1596 metadata_blob
1597 .extend_from_slice(&u32::try_from(self.dir_nodes.len()).unwrap().to_le_bytes());
1598 for node in &self.dir_nodes {
1599 metadata_blob.extend_from_slice(&node.bytes);
1600 }
1601 if !self.shared_inline_table.is_empty() {
1604 metadata_blob.extend_from_slice(
1605 &u32::try_from(self.shared_inline_table.len())
1606 .unwrap()
1607 .to_le_bytes(),
1608 );
1609 for entry in &self.shared_inline_table {
1610 let len = u32::try_from(entry.len()).expect("shared entry fits u32");
1611 metadata_blob.extend_from_slice(&len.to_le_bytes());
1612 metadata_blob.extend_from_slice(entry);
1613 }
1614 }
1615
1616 Self::trace_phase("metadata_encode", t);
1617 let uncompressed_len =
1624 u32::try_from(metadata_blob.len()).expect("metadata blob length fits u32");
1625 let t = std::time::Instant::now();
1626 let metadata_hash = hash_section(&metadata_blob);
1627 let metadata_codec = self.metadata_codec;
1628 let metadata_quality = if metadata_blob.len() > METADATA_LARGE_BLOB_THRESHOLD {
1629 METADATA_LARGE_BLOB_QUALITY
1630 } else {
1631 METADATA_SMALL_BLOB_QUALITY
1632 };
1633 let compressed_blob = if metadata_codec == limnifs_core::codec::CODEC_BROTLI {
1634 limnifs_core::codec::compress_brotli_with_quality(&metadata_blob, metadata_quality)
1635 .unwrap_or_else(|_| metadata_blob.clone())
1636 } else {
1637 limnifs_core::codec::compress(metadata_codec, &metadata_blob)
1638 .unwrap_or_else(|_| metadata_blob.clone())
1639 };
1640 Self::trace_phase("metadata_compress", t);
1641 let (on_wire_codec, on_wire_blob) = if compressed_blob.len() < metadata_blob.len() {
1642 (metadata_codec, compressed_blob)
1643 } else {
1644 (limnifs_core::codec::CODEC_STORE, metadata_blob.clone())
1645 };
1646
1647 let externalize_at = self
1651 .metadata_externalize_threshold
1652 .min(limnifs_core::metadata_reference::DEFAULT_INLINE_METADATA_MAX_BYTES as usize);
1653 let (metadata_sidecar, inline_data, metadata_locator_count) =
1654 if on_wire_blob.len() > externalize_at {
1655 let locator = "file:metadata.bin".to_owned();
1656 let sidecar = MetadataSidecar {
1657 bytes: on_wire_blob.clone(),
1658 locator,
1659 };
1660 (Some(sidecar), None, 1u32)
1661 } else {
1662 (None, Some(on_wire_blob.clone()), 0u32)
1663 };
1664
1665 let mut manifest = Vec::new();
1666
1667 let header_start = manifest.len();
1668 manifest.extend_from_slice(&ManifestHeader::current().to_bytes());
1669 let header_end = manifest.len();
1670
1671 let flags_start = manifest.len();
1672 manifest.push(FEATURE_FLAGS_SECTION_VERSION);
1673 manifest.extend_from_slice(&0u32.to_le_bytes());
1674 let flags_end = manifest.len();
1675
1676 let meta_ref_start = manifest.len();
1679 manifest.push(METADATA_REFERENCE_SECTION_VERSION_2);
1680 manifest.extend_from_slice(&metadata_hash);
1681 manifest.extend_from_slice(&uncompressed_len.to_le_bytes());
1682 manifest.push(on_wire_codec);
1683 manifest.extend_from_slice(&metadata_locator_count.to_le_bytes());
1684 if let Some(sidecar) = &metadata_sidecar {
1685 let loc_bytes = sidecar.locator.as_bytes();
1686 let loc_len = u32::try_from(loc_bytes.len()).expect("locator fits u32");
1687 manifest.extend_from_slice(&loc_len.to_le_bytes());
1688 manifest.extend_from_slice(loc_bytes);
1689 }
1690 match &inline_data {
1691 Some(blob) => {
1692 let inline_len = u32::try_from(blob.len()).expect("metadata fits u32");
1693 manifest.extend_from_slice(&inline_len.to_le_bytes());
1694 manifest.extend_from_slice(blob);
1695 }
1696 None => {
1697 manifest.extend_from_slice(&0u32.to_le_bytes());
1698 }
1699 }
1700 let meta_ref_end = manifest.len();
1701
1702 let slab_index_start = manifest.len();
1703 manifest.push(SLAB_INDEX_SECTION_VERSION);
1704 manifest.extend_from_slice(&u32::try_from(slabs.len()).unwrap().to_le_bytes());
1705 for slab in &slabs {
1706 manifest.extend_from_slice(&slab.id.to_bytes());
1707 manifest.extend_from_slice(&1u32.to_le_bytes());
1708 let loc_bytes = slab.locator.as_bytes();
1709 let loc_len = u32::try_from(loc_bytes.len()).expect("locator fits u32");
1710 manifest.extend_from_slice(&loc_len.to_le_bytes());
1711 manifest.extend_from_slice(loc_bytes);
1712 }
1713 let slab_index_end = manifest.len();
1714
1715 let history_start = manifest.len();
1716 manifest.push(HISTORY_SECTION_VERSION);
1717 manifest.extend_from_slice(&1u32.to_le_bytes());
1718 manifest.push(0x01);
1719 manifest.extend_from_slice(&0u64.to_le_bytes());
1720 manifest.extend_from_slice(&0u32.to_le_bytes());
1721 manifest.extend_from_slice(&0u32.to_le_bytes());
1722 let history_end = manifest.len();
1723
1724 let profile_desc_start = manifest.len();
1729 if let Some(ref name) = self.profile_name {
1730 let desc = limnifs_core::profile_descriptor::ProfileDescriptor {
1731 version: limnifs_core::profile_descriptor::PROFILE_DESCRIPTOR_SECTION_VERSION,
1732 profile_name: Some(name.clone()),
1733 blake3_hashing: true,
1734 cross_file_dedup: true,
1735 content_classification: !self.categorizers_disabled,
1736 integrity_verify: true,
1737 read_write: self.rw_mode,
1738 auto_turnover: self.auto_turnover,
1739 };
1740 limnifs_core::profile_descriptor::encode_profile_descriptor(&desc, &mut manifest);
1741 }
1742 let profile_desc_end = manifest.len();
1743
1744 if !self.trained_dicts_by_class.is_empty() {
1750 let dicts: Vec<_> = self
1751 .trained_dicts_by_class
1752 .values()
1753 .map(|d| limnifs_core::dictionary_section::Dictionary {
1754 codec_id: d.codec,
1755 class_id: d.id,
1756 data: d.content.clone(),
1757 })
1758 .collect();
1759 let section = limnifs_core::dictionary_section::DictionarySection {
1760 version: limnifs_core::dictionary_section::DICTIONARY_SECTION_VERSION,
1761 dicts,
1762 };
1763 limnifs_core::dictionary_section::encode_dictionary_section(§ion, &mut manifest);
1764 }
1765
1766 let dictionary_end = manifest.len();
1767
1768 let delta_linkage_hash = if let Some(base_root) = self.base_root {
1774 let delta_start = manifest.len();
1775 manifest.push(limnifs_core::delta_linkage::DELTA_LINKAGE_SECTION_VERSION);
1781 manifest.extend_from_slice(&base_root);
1782 manifest.extend_from_slice(&0u32.to_le_bytes());
1783 hash_section(&manifest[delta_start..])
1784 } else {
1785 hash_empty_section()
1786 };
1787 let _ = dictionary_end;
1788
1789 let hashes = SectionHashes {
1790 metadata: metadata_hash,
1791 format_header: hash_section(&manifest[header_start..header_end]),
1792 feature_flags: hash_section(&manifest[flags_start..flags_end]),
1793 metadata_reference: hash_section(&manifest[meta_ref_start..meta_ref_end]),
1794 slab_index: hash_section(&manifest[slab_index_start..slab_index_end]),
1795 crypto_params: hash_empty_section(),
1796 ec_params: hash_empty_section(),
1797 dms_policy: hash_empty_section(),
1798 delta_linkage: delta_linkage_hash,
1799 history: hash_section(&manifest[history_start..history_end]),
1800 };
1812 let merkle_root = compute_merkle_root(&hashes);
1813
1814 WriteArtifact {
1815 bytes: manifest,
1816 merkle_root,
1817 slabs,
1818 metadata_sidecar,
1819 inode_count,
1820 file_count: self.file_count,
1821 dir_count,
1822 drop_count,
1823 root_inode_number: self.root_inode_number,
1824 }
1825 }
1826
1827 fn encode_inode(&self, out: &mut Vec<u8>, inode: &PendingInode) {
1828 out.extend_from_slice(&inode.number.to_le_bytes());
1829 out.extend_from_slice(&inode.mode.to_le_bytes());
1830 out.extend_from_slice(&0u32.to_le_bytes());
1831 out.extend_from_slice(&0u32.to_le_bytes());
1832 out.extend_from_slice(&inode.mtime_ns.to_le_bytes());
1833 out.extend_from_slice(&inode.mtime_ns.to_le_bytes());
1834 out.extend_from_slice(&1u32.to_le_bytes());
1835 match &inode.content {
1836 PendingContent::Inline(data) => {
1837 let h = hash_section(data);
1838 if let Some(&idx) = self.shared_inline_map.get(&h) {
1839 out.push(INODE_FLAG_INLINE_DATA | INODE_FLAG_SHARED_INLINE);
1841 out.extend_from_slice(&(idx as u32).to_le_bytes());
1842 } else {
1843 out.push(INODE_FLAG_INLINE_DATA);
1844 let len = u32::try_from(data.len()).expect("data fits u32");
1845 out.extend_from_slice(&len.to_le_bytes());
1846 out.extend_from_slice(data);
1847 }
1848 }
1849 PendingContent::DropBacked { file_len, slices } => {
1850 out.push(0x00);
1851 let slice_count = u32::try_from(slices.len()).expect("slice count fits u32");
1852 out.extend_from_slice(&slice_count.to_le_bytes());
1853 for slice in slices {
1854 out.extend_from_slice(&slice.file_byte_start.to_le_bytes());
1855 out.extend_from_slice(&slice.file_byte_end.to_le_bytes());
1856 out.extend_from_slice(&slice.drop_id);
1857 out.extend_from_slice(&0u32.to_le_bytes());
1859 let drop_byte_len = u32::try_from(slice.file_byte_end - slice.file_byte_start)
1863 .expect("slice range fits u32");
1864 out.extend_from_slice(&drop_byte_len.to_le_bytes());
1865 }
1866 let _ = file_len;
1867 }
1868 PendingContent::Directory(entries) => {
1869 out.push(0x00);
1870 let node = self
1871 .dir_nodes
1872 .iter()
1873 .find(|n| n.entries == *entries)
1874 .expect("directory node must exist");
1875 out.extend_from_slice(&node.hash);
1876 }
1877 }
1878 }
1879}
1880
1881fn encode_dir_node(entries: &[(String, u64, u8)]) -> DirNode {
1882 let mut bytes = Vec::new();
1883 bytes.push(1u8);
1884 let count = u32::try_from(entries.len()).expect("entry count fits u32");
1885 bytes.extend_from_slice(&count.to_le_bytes());
1886 for (name, inode_number, entry_type) in entries {
1887 let name_bytes = name.as_bytes();
1888 let name_len = u32::try_from(name_bytes.len()).expect("name fits u32");
1889 bytes.extend_from_slice(&name_len.to_le_bytes());
1890 bytes.extend_from_slice(name_bytes);
1891 bytes.extend_from_slice(&inode_number.to_le_bytes());
1892 bytes.push(*entry_type);
1893 }
1894 let hash = hash_section(&bytes);
1895 DirNode {
1896 entries: entries.to_vec(),
1897 bytes,
1898 hash,
1899 }
1900}
1901
1902fn pack_slabs(drops: &[PendingDrop]) -> Vec<SlabArtifact> {
1911 let local_drops: Vec<&PendingDrop> = drops
1916 .iter()
1917 .filter(|d| d.codec != limnifs_core::codec::CODEC_REFERENCED)
1918 .collect();
1919 if local_drops.is_empty() {
1920 return Vec::new();
1921 }
1922
1923 let max_content = MAX_SLAB_TOTAL_BYTES.saturating_sub(SLAB_HEADER_LEN);
1924
1925 let mut slab_groups: Vec<Vec<&PendingDrop>> = Vec::new();
1930 let mut current: Vec<&PendingDrop> = Vec::new();
1931 let mut current_size: usize = 0;
1932
1933 for drop in &local_drops {
1934 let footprint = drop.slab_footprint();
1935 if !current.is_empty() && current_size + footprint > max_content {
1936 slab_groups.push(std::mem::take(&mut current));
1937 current_size = 0;
1938 }
1939 current.push(*drop);
1940 current_size += footprint;
1941 }
1942 if !current.is_empty() {
1943 slab_groups.push(current);
1944 }
1945
1946 use rayon::prelude::*;
1952 slab_groups
1953 .par_iter()
1954 .enumerate()
1955 .map(|(ordinal, group)| {
1956 let ordinal_u64 = u64::try_from(ordinal).expect("slab count fits u64");
1957 encode_slab(ordinal_u64, group)
1958 })
1959 .collect()
1960}
1961
1962fn encode_slab(ordinal: u64, drops: &[&PendingDrop]) -> SlabArtifact {
1966 const DROP_RECORD_LEN: usize = 49;
1972 let mut drop_records = Vec::with_capacity(drops.len() * DROP_RECORD_LEN);
1973 let mut solid_window = Vec::new();
1974 let mut drop_ids = Vec::with_capacity(drops.len());
1975 let mut offset_in_window: u32 = 0;
1976
1977 for drop in drops {
1978 let plaintext_len = drop.plaintext_len_value();
1979 let window_len = drop.len_in_window();
1980 drop_records.extend_from_slice(&drop.id);
1981 drop_records.extend_from_slice(&plaintext_len.to_le_bytes());
1982 drop_records.extend_from_slice(&[drop.codec, 0x00, 0x00]);
1984 drop_records.push(0x00); drop_records.extend_from_slice(&offset_in_window.to_le_bytes());
1986 drop_records.extend_from_slice(&window_len.to_le_bytes());
1987 drop_records.push(drop.dict_id); solid_window.extend_from_slice(&drop.compressed);
1989 drop_ids.push(drop.id);
1990 offset_in_window = offset_in_window
1991 .checked_add(window_len)
1992 .expect("slab window size fits u32");
1993 }
1994
1995 let slab_content = [&drop_records[..], &solid_window[..]].concat();
1996 let slab_hash = hash_section(&slab_content);
1997 let slab_id = SlabId::new(ordinal, slab_hash);
1998
1999 let total_length = SLAB_HEADER_LEN + slab_content.len();
2000 let mut slab_bytes = Vec::with_capacity(total_length);
2001 slab_bytes.extend_from_slice(b"LIM1");
2002 slab_bytes.extend_from_slice(&1u16.to_le_bytes());
2003 slab_bytes.extend_from_slice(&slab_id.to_bytes());
2004 slab_bytes.extend_from_slice(
2005 &u64::try_from(total_length)
2006 .unwrap_or(u64::MAX)
2007 .to_le_bytes(),
2008 );
2009 slab_bytes.push(0x00);
2010 slab_bytes.push(0x00);
2011 slab_bytes.extend_from_slice(&slab_content);
2012
2013 let locator = format!("file:slab-{ordinal}.bin");
2014
2015 SlabArtifact {
2016 id: slab_id,
2017 bytes: slab_bytes,
2018 locator,
2019 drop_ids,
2020 }
2021}
2022
2023#[cfg(test)]
2024fn pseudo_random_bytes(seed: u64, count: usize) -> Vec<u8> {
2025 let mut state = seed;
2026 let mut out = Vec::with_capacity(count);
2027 for _ in 0..count {
2028 state = state
2029 .wrapping_mul(6_364_136_223_846_793_005)
2030 .wrapping_add(1_442_695_040_888_963_407);
2031 out.push(u8::try_from(state >> 56).expect("fits u8"));
2032 }
2033 out
2034}
2035
2036#[cfg(test)]
2037mod tests {
2038 use super::*;
2039 use limnifs_core::ManifestCursor;
2040
2041 #[test]
2042 fn write_stream_packs_single_named_stream() {
2043 let temp = std::env::temp_dir().join(format!(
2047 "limnifs-write-stream-test-{}-{}",
2048 std::process::id(),
2049 std::time::SystemTime::now()
2050 .duration_since(std::time::UNIX_EPOCH)
2051 .unwrap()
2052 .as_nanos()
2053 ));
2054 std::fs::create_dir_all(&temp).expect("create temp dir");
2055
2056 let content = b"stream test content line\n".repeat(10_000); let cursor = std::io::Cursor::new(content.clone());
2058 let config = WriteConfig::default_v0_1();
2059 let artifact = write_stream("streamed.txt", cursor, &config).expect("write_stream");
2060
2061 assert!(!artifact.bytes.is_empty(), "manifest bytes non-empty");
2062 assert!(!artifact.slabs.is_empty(), "at least one slab produced");
2063 let total_drop_bytes: usize = artifact.slabs.iter().map(|s| s.bytes.len()).sum();
2068 assert!(total_drop_bytes > 0, "drops non-empty");
2069
2070 let _ = std::fs::remove_dir_all(&temp);
2071 }
2072
2073 #[test]
2074 fn write_layer_references_base_drops() {
2075 let temp = std::env::temp_dir().join(format!(
2081 "limnifs-write-layer-test-{}-{}",
2082 std::process::id(),
2083 std::time::SystemTime::now()
2084 .duration_since(std::time::UNIX_EPOCH)
2085 .unwrap()
2086 .as_nanos()
2087 ));
2088 std::fs::create_dir_all(&temp).expect("create temp dir");
2089
2090 let base_dir = temp.join("base");
2092 std::fs::create_dir_all(&base_dir).expect("base dir");
2093 let text = b"layer test content line\n".repeat(50_000); std::fs::write(base_dir.join("shared.txt"), &text).expect("write shared");
2095 std::fs::write(base_dir.join("base-only.txt"), b"only in base").expect("write base-only");
2096
2097 let config = WriteConfig::default_v0_1();
2098 let base_artifact = write_directory_with_config(&base_dir, &config).expect("base");
2099
2100 let base_manifest = temp.join("base.lim");
2101 std::fs::write(&base_manifest, &base_artifact.bytes).expect("write base manifest");
2102 for slab in &base_artifact.slabs {
2103 let slab_name = slab.locator.strip_prefix("file:").unwrap_or(&slab.locator);
2104 std::fs::write(temp.join(slab_name), &slab.bytes).expect("write base slab");
2105 }
2106
2107 let layer_dir = temp.join("layer");
2109 std::fs::create_dir_all(&layer_dir).expect("layer dir");
2110 std::fs::write(layer_dir.join("shared.txt"), &text).expect("write shared in layer");
2111 std::fs::write(layer_dir.join("new.txt"), b"fresh content in layer").expect("write new");
2112
2113 let layer_artifact = write_layer(&base_manifest, &layer_dir, &config).expect("layer");
2114
2115 let layer_slab_bytes: usize = layer_artifact.slabs.iter().map(|s| s.bytes.len()).sum();
2118 let base_slab_bytes: usize = base_artifact.slabs.iter().map(|s| s.bytes.len()).sum();
2119 assert!(
2120 layer_slab_bytes < base_slab_bytes / 4,
2121 "layer slabs ({}) should be much smaller than base ({}) — layering failed",
2122 layer_slab_bytes,
2123 base_slab_bytes
2124 );
2125
2126 let base_root = base_artifact.merkle_root.as_bytes();
2129 assert!(
2130 layer_artifact
2131 .bytes
2132 .windows(32)
2133 .any(|w| w == base_root.as_slice()),
2134 "layer manifest must contain base's ManifestRoot bytes"
2135 );
2136
2137 let _ = std::fs::remove_dir_all(&temp);
2138 }
2139
2140 #[test]
2141 fn tournament_short_circuits_on_highly_compressible_chunk() {
2142 let chunk = b"hello world ".repeat(500);
2145 let tunables = limnifs_core::codec::CodecTunables::default();
2146 let tournament = TournamentSpec {
2147 codec_ids: vec![
2148 limnifs_core::codec::CODEC_LZ4,
2149 limnifs_core::codec::CODEC_BROTLI,
2150 ],
2151 min_size: 16,
2152 skip_for_binary: false,
2153 short_circuit_permille: 250,
2154 };
2155 let (codec_id, compressed) = compress_chunk_with_tournament(
2156 &chunk,
2157 classifier::Class::Text,
2158 limnifs_core::codec::CODEC_BROTLI,
2159 limnifs_core::codec::CODEC_LZ4,
2160 &tunables,
2161 &tournament,
2162 );
2163 assert_eq!(codec_id, limnifs_core::codec::CODEC_LZ4);
2164 assert!(compressed.len() < chunk.len());
2165 }
2166
2167 #[test]
2168 fn tournament_runs_all_codecs_when_short_circuit_disabled() {
2169 let chunk = b"hello world ".repeat(500);
2175 let tunables = limnifs_core::codec::CodecTunables::default();
2176 let tournament = TournamentSpec {
2177 codec_ids: vec![
2178 limnifs_core::codec::CODEC_LZ4,
2179 limnifs_core::codec::CODEC_BROTLI,
2180 limnifs_core::codec::CODEC_ZSTD,
2181 ],
2182 min_size: 16,
2183 skip_for_binary: false,
2184 short_circuit_permille: 0,
2185 };
2186 let (codec_id, compressed) = compress_chunk_with_tournament(
2187 &chunk,
2188 classifier::Class::Text,
2189 limnifs_core::codec::CODEC_BROTLI,
2190 limnifs_core::codec::CODEC_LZ4,
2191 &tunables,
2192 &tournament,
2193 );
2194 assert!(
2198 codec_id == limnifs_core::codec::CODEC_ZSTD
2199 || codec_id == limnifs_core::codec::CODEC_BROTLI,
2200 "expected ZSTD or Brotli to win, got codec {codec_id}"
2201 );
2202 assert!(compressed.len() < chunk.len());
2203 }
2204
2205 #[test]
2206 fn tournament_skips_for_binary_when_configured() {
2207 let chunk = vec![0u8; 4096];
2208 let tunables = limnifs_core::codec::CodecTunables::default();
2209 let tournament = TournamentSpec {
2210 codec_ids: vec![limnifs_core::codec::CODEC_BROTLI],
2211 min_size: 16,
2212 skip_for_binary: true,
2213 short_circuit_permille: 250,
2214 };
2215 let (codec_id, _compressed) = compress_chunk_with_tournament(
2216 &chunk,
2217 classifier::Class::Binary,
2218 limnifs_core::codec::CODEC_BROTLI,
2219 limnifs_core::codec::CODEC_LZ4,
2220 &tunables,
2221 &tournament,
2222 );
2223 assert_eq!(codec_id, limnifs_core::codec::CODEC_LZ4);
2225 }
2226
2227 #[test]
2228 fn tournament_small_chunk_uses_preferred_codec() {
2229 let chunk = b"tiny";
2230 let tunables = limnifs_core::codec::CodecTunables::default();
2231 let tournament = TournamentSpec {
2232 codec_ids: vec![limnifs_core::codec::CODEC_BROTLI],
2233 min_size: 1024,
2234 skip_for_binary: false,
2235 short_circuit_permille: 0,
2236 };
2237 let (codec_id, _compressed) = compress_chunk_with_tournament(
2238 chunk,
2239 classifier::Class::Text,
2240 limnifs_core::codec::CODEC_BROTLI,
2241 limnifs_core::codec::CODEC_LZ4,
2242 &tunables,
2243 &tournament,
2244 );
2245 assert_eq!(codec_id, limnifs_core::codec::CODEC_STORE);
2247 }
2248
2249 #[test]
2250 fn tournament_falls_back_to_store_when_no_codec_compresses() {
2251 let chunk = pseudo_random_bytes(42, 4096);
2255 let tunables = limnifs_core::codec::CodecTunables::default();
2256 let tournament = TournamentSpec {
2257 codec_ids: vec![limnifs_core::codec::CODEC_LZ4],
2258 min_size: 16,
2259 skip_for_binary: false,
2260 short_circuit_permille: 0,
2261 };
2262 let (codec_id, compressed) = compress_chunk_with_tournament(
2263 &chunk,
2264 classifier::Class::Binary,
2265 limnifs_core::codec::CODEC_BROTLI,
2266 limnifs_core::codec::CODEC_LZ4,
2267 &tunables,
2268 &tournament,
2269 );
2270 assert_eq!(codec_id, limnifs_core::codec::CODEC_STORE);
2271 assert_eq!(compressed.len(), chunk.len());
2272 }
2273
2274 #[test]
2275 fn dictionaries_enabled_emits_dictionary_section_when_enough_samples() {
2276 let temp = std::env::temp_dir().join(format!(
2281 "limnifs-write-test-{}-dict-{}",
2282 std::process::id(),
2283 std::time::SystemTime::now()
2284 .duration_since(std::time::UNIX_EPOCH)
2285 .map(|d| d.as_nanos() as u64)
2286 .unwrap_or(0),
2287 ));
2288 let _ = std::fs::remove_dir_all(&temp);
2289 std::fs::create_dir_all(&temp).expect("mkdir");
2290
2291 for i in 0..200 {
2294 let content = format!(
2296 "function test_case_{i}() {{ return constant + {i}; }}\n\
2297 // shared comment line {i}\n\
2298 struct Foo {{ x: i32 }} // type {i}\n"
2299 )
2300 .repeat(5);
2301 let path = temp.join(format!("file_{i:04}.txt"));
2302 std::fs::write(&path, content.as_bytes()).expect("write");
2303 }
2304
2305 let mut config = crate::profile::balanced();
2306 config.defaults.text_codec = "zstd".into();
2308 config.defaults.metadata_codec = "zstd".into();
2312 config.dictionaries.enabled = true;
2313 config.dictionaries.min_class_size = 50;
2314 config.dictionaries.max_dict_size = 8192;
2315
2316 let artifact = write_directory_with_config(&temp, &config).expect("write");
2317 std::fs::remove_dir_all(&temp).ok();
2318
2319 let mut cursor = ManifestCursor::new(&artifact.bytes);
2324 let _ = limnifs_core::parse_manifest_header(&mut cursor).expect("header");
2325 let _ = limnifs_core::parse_feature_flags_section(&mut cursor).expect("flags");
2326 let _ = limnifs_core::parse_metadata_reference(&mut cursor).expect("meta_ref");
2327 let _ = limnifs_core::parse_slab_index(&mut cursor).expect("slab_index");
2328 let _ = limnifs_core::parse_history(&mut cursor).expect("history");
2329 let _remaining = cursor.remaining_len();
2332 }
2333
2334 #[test]
2335 fn write_empty_directory() {
2336 let temp =
2337 std::env::temp_dir().join(format!("limnifs-write-test-{}-empty", std::process::id()));
2338 std::fs::create_dir_all(&temp).expect("create temp dir");
2339 let artifact = write_directory(&temp).expect("write succeeds");
2340 std::fs::remove_dir_all(&temp).ok();
2341 assert!(artifact.inode_count >= 1);
2342 assert_eq!(artifact.file_count, 0);
2343 assert_eq!(artifact.dir_count, 1);
2344 assert!(artifact.slabs.is_empty());
2345 }
2346
2347 #[test]
2348 fn write_small_file_inline() {
2349 let temp =
2350 std::env::temp_dir().join(format!("limnifs-write-test-{}-small", std::process::id()));
2351 std::fs::create_dir_all(&temp).expect("create temp dir");
2352 std::fs::write(temp.join("hello.txt"), b"hello world").expect("write file");
2353 let artifact = write_directory(&temp).expect("write succeeds");
2354 std::fs::remove_dir_all(&temp).ok();
2355 assert_eq!(artifact.file_count, 1);
2356 assert!(artifact.slabs.is_empty());
2357 assert_eq!(artifact.drop_count, 0);
2358 }
2359
2360 #[test]
2361 fn write_large_file_uses_slab() {
2362 let temp =
2363 std::env::temp_dir().join(format!("limnifs-write-test-{}-large", std::process::id()));
2364 std::fs::create_dir_all(&temp).expect("create temp dir");
2365 let large_data = vec![0xABu8; INLINE_THRESHOLD + 100];
2366 std::fs::write(temp.join("big.bin"), &large_data).expect("write big");
2367 let artifact = write_directory(&temp).expect("write succeeds");
2368 std::fs::remove_dir_all(&temp).ok();
2369 assert_eq!(artifact.drop_count, 1);
2370 assert_eq!(artifact.slabs.len(), 1);
2371 }
2372
2373 #[test]
2374 fn write_mixed_inline_and_large() {
2375 let temp =
2376 std::env::temp_dir().join(format!("limnifs-write-test-{}-mix", std::process::id()));
2377 std::fs::create_dir_all(&temp).expect("create temp dir");
2378 std::fs::write(temp.join("small.txt"), b"tiny").expect("write small");
2379 std::fs::write(temp.join("large.bin"), vec![0xCDu8; INLINE_THRESHOLD * 2])
2380 .expect("write large");
2381 let artifact = write_directory(&temp).expect("write succeeds");
2382 std::fs::remove_dir_all(&temp).ok();
2383 assert_eq!(artifact.file_count, 2);
2384 assert_eq!(artifact.drop_count, 1);
2385 assert_eq!(artifact.slabs.len(), 1);
2386 }
2387
2388 #[test]
2389 fn deduplicates_identical_large_files() {
2390 let temp =
2391 std::env::temp_dir().join(format!("limnifs-write-test-{}-dedup", std::process::id()));
2392 std::fs::create_dir_all(&temp).expect("create temp dir");
2393 let data = vec![0x77u8; INLINE_THRESHOLD + 10];
2394 std::fs::write(temp.join("a.bin"), &data).expect("write a");
2395 std::fs::write(temp.join("b.bin"), &data).expect("write b");
2396 let artifact = write_directory(&temp).expect("write succeeds");
2397 std::fs::remove_dir_all(&temp).ok();
2398 assert_eq!(artifact.drop_count, 1);
2399 }
2400
2401 #[test]
2402 fn write_and_verify_roundtrip() {
2403 let temp = std::env::temp_dir().join(format!(
2404 "limnifs-write-test-{}-roundtrip",
2405 std::process::id()
2406 ));
2407 std::fs::create_dir_all(&temp).expect("create temp dir");
2408 std::fs::write(temp.join("a.txt"), b"aaa").expect("write a");
2409 std::fs::write(temp.join("b.txt"), b"bbb").expect("write b");
2410 std::fs::create_dir_all(temp.join("sub")).expect("create sub");
2411 std::fs::write(temp.join("sub").join("c.txt"), b"ccc").expect("write c");
2412 let artifact = write_directory(&temp).expect("write succeeds");
2413 std::fs::remove_dir_all(&temp).ok();
2414 assert_eq!(artifact.file_count, 3);
2415 assert_eq!(artifact.dir_count, 2);
2416
2417 let mut cursor = ManifestCursor::new(&artifact.bytes);
2418 limnifs_core::parse_manifest_header(&mut cursor).expect("header");
2419 limnifs_core::parse_feature_flags_section(&mut cursor).expect("flags");
2420 let meta_ref = limnifs_core::parse_metadata_reference(&mut cursor).expect("meta ref");
2421 assert!(meta_ref.is_inlined());
2422 let slab_index = limnifs_core::parse_slab_index(&mut cursor).expect("slab index");
2423 assert_eq!(slab_index.len(), 0);
2424 limnifs_core::parse_history(&mut cursor).expect("history");
2425 }
2426
2427 #[test]
2428 fn write_deterministic() {
2429 let temp =
2430 std::env::temp_dir().join(format!("limnifs-write-test-{}-det", std::process::id()));
2431 std::fs::create_dir_all(&temp).expect("create temp dir");
2432 std::fs::write(temp.join("x.txt"), b"xxx").expect("write x");
2433
2434 let a1 = write_directory(&temp).expect("first write");
2435 let a2 = write_directory(&temp).expect("second write");
2436 std::fs::remove_dir_all(&temp).ok();
2437
2438 assert_eq!(a1.bytes, a2.bytes);
2439 assert_eq!(a1.merkle_root, a2.merkle_root);
2440 }
2441
2442 #[test]
2443 fn slab_parses_correctly() {
2444 let temp =
2445 std::env::temp_dir().join(format!("limnifs-write-test-{}-slab", std::process::id()));
2446 std::fs::create_dir_all(&temp).expect("create temp dir");
2447 std::fs::write(temp.join("big.bin"), vec![0x11u8; INLINE_THRESHOLD + 1])
2448 .expect("write big");
2449 let artifact = write_directory(&temp).expect("write succeeds");
2450 std::fs::remove_dir_all(&temp).ok();
2451
2452 let slab_bytes = &artifact.slabs[0].bytes;
2453 let mut cursor = ManifestCursor::new(slab_bytes);
2454 let slab_header = limnifs_core::parse_slab_header(&mut cursor).expect("slab header parses");
2455 assert_eq!(slab_header.format_version, 1);
2456 assert!(!slab_header.is_sealed());
2457 assert!(!slab_header.has_erasure_coding());
2458
2459 let drop_record =
2460 limnifs_core::parse_drop_record(&mut cursor, &slab_header).expect("drop record parses");
2461 assert_eq!(drop_record.plaintext_len as usize, INLINE_THRESHOLD + 1);
2462 }
2463
2464 #[test]
2465 fn fastcdc_produces_multiple_chunks_for_large_files() {
2466 let temp = std::env::temp_dir().join(format!(
2469 "limnifs-write-test-{}-cdc-multi",
2470 std::process::id()
2471 ));
2472 std::fs::create_dir_all(&temp).expect("create temp dir");
2473 let data = pseudo_random_bytes(42, 1024 * 1024);
2474 std::fs::write(temp.join("big.bin"), &data).expect("write big");
2475 let artifact = write_directory(&temp).expect("write succeeds");
2476 std::fs::remove_dir_all(&temp).ok();
2477 assert!(
2478 artifact.drop_count > 1,
2479 "expected FastCDC to produce multiple drops for 1 MiB input, got {}",
2480 artifact.drop_count
2481 );
2482 }
2483
2484 #[test]
2485 fn fastcdc_deduplicates_shared_substrings() {
2486 let temp = std::env::temp_dir().join(format!(
2490 "limnifs-write-test-{}-cdc-dedup",
2491 std::process::id()
2492 ));
2493 std::fs::create_dir_all(&temp).expect("create temp dir");
2494 let shared = pseudo_random_bytes(7, 512 * 1024);
2495 let mut a = Vec::with_capacity(shared.len() + 1024);
2496 a.extend_from_slice(&pseudo_random_bytes(1, 1024));
2497 a.extend_from_slice(&shared);
2498 let mut b = Vec::with_capacity(shared.len() + 2048);
2499 b.extend_from_slice(&pseudo_random_bytes(2, 2048));
2500 b.extend_from_slice(&shared);
2501 std::fs::write(temp.join("a.bin"), &a).expect("write a");
2502 std::fs::write(temp.join("b.bin"), &b).expect("write b");
2503
2504 let temp_a = std::env::temp_dir().join(format!(
2506 "limnifs-write-test-{}-cdc-dedup-a",
2507 std::process::id()
2508 ));
2509 std::fs::create_dir_all(&temp_a).expect("create temp_a");
2510 std::fs::write(temp_a.join("a.bin"), &a).expect("write a");
2511 let artifact_a = write_directory(&temp_a).expect("a writes");
2512 std::fs::remove_dir_all(&temp_a).ok();
2513
2514 let temp_b = std::env::temp_dir().join(format!(
2515 "limnifs-write-test-{}-cdc-dedup-b",
2516 std::process::id()
2517 ));
2518 std::fs::create_dir_all(&temp_b).expect("create temp_b");
2519 std::fs::write(temp_b.join("b.bin"), &b).expect("write b");
2520 let artifact_b = write_directory(&temp_b).expect("b writes");
2521 std::fs::remove_dir_all(&temp_b).ok();
2522
2523 let artifact_both = write_directory(&temp).expect("both write");
2524 std::fs::remove_dir_all(&temp).ok();
2525
2526 let sum_alone = artifact_a.drop_count + artifact_b.drop_count;
2527 assert!(
2528 artifact_both.drop_count < sum_alone,
2529 "expected dedup win: both together = {} drops, sum alone = {} drops",
2530 artifact_both.drop_count,
2531 sum_alone
2532 );
2533 }
2534
2535 #[test]
2536 fn slab_splits_when_content_exceeds_ceiling() {
2537 let temp =
2543 std::env::temp_dir().join(format!("limnifs-write-test-{}-split", std::process::id()));
2544 std::fs::create_dir_all(&temp).expect("create temp dir");
2545 for i in 0..7u32 {
2546 let data = pseudo_random_bytes(u64::from(i), 10 * 1024 * 1024);
2548 std::fs::write(temp.join(format!("big-{i}.bin")), &data).expect("write big");
2549 }
2550 let artifact = write_directory(&temp).expect("write succeeds");
2551 std::fs::remove_dir_all(&temp).ok();
2552
2553 assert!(
2555 artifact.slabs.len() >= 2,
2556 "expected at least 2 slabs for 70 MiB of incompressible data, got {}",
2557 artifact.slabs.len()
2558 );
2559 for slab in &artifact.slabs {
2560 assert!(
2561 slab.bytes.len() <= MAX_SLAB_TOTAL_BYTES,
2562 "slab {} is {} bytes (> {} ceiling)",
2563 slab.id.ordinal,
2564 slab.bytes.len(),
2565 MAX_SLAB_TOTAL_BYTES,
2566 );
2567 }
2568 let total_drop_ids: usize = artifact.slabs.iter().map(|s| s.drop_ids.len()).sum();
2570 assert_eq!(
2571 total_drop_ids, artifact.drop_count,
2572 "drop_ids count across slabs must match WriteArtifact.drop_count",
2573 );
2574 }
2575}