#![deny(unsafe_code)]
#![allow(warnings)]
pub mod chunker;
pub mod classifier;
pub mod compaction;
pub mod config;
pub mod delta_builder;
pub mod dictionary;
pub mod file_categorizer;
use file_categorizer::FileCategorizer;
pub mod flatten;
pub mod progress;
pub mod rw;
#[cfg(feature = "sparse-index")]
pub mod sparse_index;
pub mod stream;
pub mod turnover;
pub use config::{
profile, CategorizerConfig, ChunkingConfig, CodecRegistry, CodecTunables, Defaults,
DictionaryConfig, EncryptionConfig, TournamentConfig, WriteConfig,
};
use std::collections::{HashMap, HashSet};
use std::path::{Path, PathBuf};
use crate::chunker::{Chunker, ParallelFastCDC};
use limnifs_core::codec::CODEC_REFERENCED;
use limnifs_core::dictionary_section::parse_dictionary_section;
use limnifs_core::slab_store::SlabStore;
use limnifs_core::{
compute_merkle_root, hash_empty_section, hash_section, parse_manifest_header, parse_slab_index,
ManifestCursor, ManifestHeader, SectionHashes, FEATURE_FLAGS_SECTION_VERSION,
HISTORY_SECTION_VERSION, INODE_FLAG_INLINE_DATA, INODE_FLAG_SHARED_INLINE,
METADATA_REFERENCE_SECTION_VERSION_2, SLAB_INDEX_SECTION_VERSION,
};
use limnifs_format::{ManifestRoot, SlabId};
pub const INLINE_THRESHOLD: usize = 4096;
pub const MMAP_READ_THRESHOLD: usize = 1024 * 1024;
pub const WHOLE_FILE_MAX_SIZE: usize = 64 * 1024 * 1024;
pub const MAX_SLAB_TOTAL_BYTES: usize = 60 * 1024 * 1024;
const SLAB_HEADER_LEN: usize = 56;
pub const METADATA_EXTERNALIZE_THRESHOLD: usize =
limnifs_core::metadata_reference::DEFAULT_INLINE_METADATA_MAX_BYTES as usize - 24 * 1024;
pub const METADATA_LARGE_BLOB_THRESHOLD: usize = 256 * 1024;
pub const METADATA_SMALL_BLOB_QUALITY: i32 = 5;
pub const METADATA_LARGE_BLOB_QUALITY: i32 = 2;
#[derive(Clone, Debug)]
pub struct SlabArtifact {
pub id: SlabId,
pub bytes: Vec<u8>,
pub locator: String,
pub drop_ids: Vec<[u8; 32]>,
}
#[derive(Clone, Debug)]
pub struct MetadataSidecar {
pub bytes: Vec<u8>,
pub locator: String,
}
#[derive(Clone, Debug)]
pub struct WriteArtifact {
pub bytes: Vec<u8>,
pub merkle_root: ManifestRoot,
pub slabs: Vec<SlabArtifact>,
pub metadata_sidecar: Option<MetadataSidecar>,
pub inode_count: usize,
pub file_count: usize,
pub dir_count: usize,
pub drop_count: usize,
pub root_inode_number: u64,
}
impl WriteArtifact {
#[must_use]
pub fn slab_bytes(&self) -> Option<&[u8]> {
if self.slabs.len() == 1 {
Some(&self.slabs[0].bytes)
} else {
None
}
}
#[must_use]
pub fn slab_locator(&self) -> Option<&str> {
if self.slabs.len() == 1 {
Some(&self.slabs[0].locator)
} else {
None
}
}
}
#[derive(Debug)]
pub enum WriteError {
Io(std::io::Error),
UnsupportedFileType {
path: PathBuf,
kind: String,
},
}
impl std::fmt::Display for WriteError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Io(e) => write!(f, "I/O error: {e}"),
Self::UnsupportedFileType { path, kind } => write!(
f,
"unsupported file type ({kind}): {} — limnifs stores files, \
directories, and symlinks; remove the entry or file an issue \
if you need it carried",
path.display()
),
}
}
}
impl std::error::Error for WriteError {}
impl From<std::io::Error> for WriteError {
fn from(e: std::io::Error) -> Self {
Self::Io(e)
}
}
pub fn write_directory(root: &Path) -> Result<WriteArtifact, WriteError> {
write_directory_with_config(root, &WriteConfig::default_v0_1())
}
pub fn write_stream<R: std::io::Read>(
name: &str,
mut reader: R,
config: &WriteConfig,
) -> Result<WriteArtifact, WriteError> {
let mut writer = crate::stream::StreamWriter::new(config)?;
writer.add_file(
name,
crate::stream::EntryMeta::new(0, 0o644),
&[],
&mut reader,
)?;
writer.finish()
}
pub fn write_layer(
base_image: &Path,
root: &Path,
config: &WriteConfig,
) -> Result<WriteArtifact, WriteError> {
let base_root = load_base_drop_index(base_image)?.1;
let base_drop_index: std::sync::Arc<dyn BaseDropSet> = {
#[cfg(feature = "sparse-index")]
{
match SparseBackedBaseIndex::open(base_image) {
Some(idx) => std::sync::Arc::new(idx),
None => std::sync::Arc::new(load_base_drop_index(base_image)?.0),
}
}
#[cfg(not(feature = "sparse-index"))]
{
std::sync::Arc::new(load_base_drop_index(base_image)?.0)
}
};
let mut ctx = WriteContext::new();
ctx.chunker = chunker_from_config(config)?;
ctx.base_dictionaries = if config.dictionaries.enabled {
load_base_dictionary_section(base_image)?.map(crate::dictionary::adopt_from_section)
} else {
None
};
ctx.categorizers_disabled = config.categorizers.is_empty();
ctx.rw_mode = matches!(config.mode, crate::config::ImageMode::ReadWrite(_));
ctx.auto_turnover = config.turnover_threshold > 0;
ctx.collect_dict_samples = config.dictionaries.enabled;
ctx.inline_threshold = config.defaults.inline_threshold as usize;
ctx.metadata_externalize_threshold = config.defaults.metadata_externalize_threshold;
ctx.emit_shared_inline = config.defaults.shared_inline;
ctx.base_drop_index = Some(base_drop_index);
ctx.base_root = Some(base_root);
let root_inode_number = ctx.walk(root)?;
ctx.root_inode_number = root_inode_number;
write_directory_body(&mut ctx, config)?;
Ok(ctx.assemble())
}
pub trait BaseDropSet: Send + Sync {
fn base_contains(&self, drop_id: &[u8; 32]) -> bool;
}
impl BaseDropSet for std::collections::HashSet<[u8; 32]> {
fn base_contains(&self, drop_id: &[u8; 32]) -> bool {
self.contains(drop_id)
}
}
#[cfg(feature = "sparse-index")]
pub struct SparseBackedBaseIndex {
bloom: crate::sparse_index::SparseIndexReader,
manifest_path: std::path::PathBuf,
exact: std::sync::OnceLock<std::collections::HashSet<[u8; 32]>>,
}
#[cfg(feature = "sparse-index")]
impl SparseBackedBaseIndex {
#[must_use]
pub fn open(base_image: &Path) -> Option<Self> {
let sidecar = base_image.with_extension("lim.sparse");
let bloom = crate::sparse_index::SparseIndexReader::from_file(&sidecar)?;
Some(Self {
bloom,
manifest_path: base_image.to_path_buf(),
exact: std::sync::OnceLock::new(),
})
}
fn load_exact(&self) -> &std::collections::HashSet<[u8; 32]> {
self.exact.get_or_init(|| {
let bytes = std::fs::read(&self.manifest_path).unwrap_or_default();
let mut cursor = ManifestCursor::new(&bytes);
let _ = parse_manifest_header(&mut cursor);
let _ = limnifs_core::parse_feature_flags_section(&mut cursor);
let _ = limnifs_core::parse_metadata_reference(&mut cursor);
let Ok(index) = parse_slab_index(&mut cursor) else {
return std::collections::HashSet::new();
};
match SlabStore::load_mmap(&self.manifest_path, &index) {
Ok(store) => store.drop_index_keys().copied().collect(),
Err(_) => std::collections::HashSet::new(),
}
})
}
}
#[cfg(feature = "sparse-index")]
impl BaseDropSet for SparseBackedBaseIndex {
fn base_contains(&self, drop_id: &[u8; 32]) -> bool {
if !self.bloom.probably_contains(drop_id) {
return false;
}
self.load_exact().contains(drop_id)
}
}
#[cfg(feature = "sparse-index")]
pub fn emit_sparse_sidecar(artifact: &WriteArtifact, image_path: &Path) -> Result<(), WriteError> {
let all: std::collections::HashSet<[u8; 32]> = artifact
.slabs
.iter()
.flat_map(|s| s.drop_ids.iter().copied())
.collect();
let mut writer = crate::sparse_index::SparseIndexWriter::new(
all.len().max(1),
crate::sparse_index::DEFAULT_FPP,
);
writer.insert_all(&all);
let sidecar = image_path.with_extension("lim.sparse");
writer.write_to_file(&sidecar).map_err(WriteError::Io)
}
fn load_base_dictionary_section(
base_image: &Path,
) -> Result<Option<limnifs_core::dictionary_section::DictionarySection>, WriteError> {
let manifest_bytes = std::fs::read(base_image)?;
let mut cursor = ManifestCursor::new(&manifest_bytes);
let _ = parse_manifest_header(&mut cursor).map_err(io_core)?;
let _ = limnifs_core::parse_feature_flags_section(&mut cursor);
let _ = limnifs_core::parse_metadata_reference(&mut cursor);
let _ = parse_slab_index(&mut cursor);
let _ = limnifs_core::parse_history(&mut cursor);
if cursor.remaining_len() == 0 {
return Ok(None);
}
Ok(parse_dictionary_section(&mut cursor).ok())
}
fn load_base_drop_index(
base_image: &Path,
) -> Result<(std::collections::HashSet<[u8; 32]>, [u8; 32]), WriteError> {
let manifest_bytes = std::fs::read(base_image)?;
let mut cursor = ManifestCursor::new(&manifest_bytes);
let _ = parse_manifest_header(&mut cursor).map_err(io_core)?;
let _ = limnifs_core::parse_feature_flags_section(&mut cursor);
let _ = limnifs_core::parse_metadata_reference(&mut cursor);
let slab_index = parse_slab_index(&mut cursor).map_err(io_core)?;
let store = SlabStore::load_mmap(base_image, &slab_index).map_err(io_core)?;
let drop_set: std::collections::HashSet<[u8; 32]> = store.drop_index_keys().copied().collect();
let root = *compute_merkle_root_from_sections(&manifest_bytes).as_bytes();
Ok((drop_set, root))
}
fn compute_merkle_root_from_sections(manifest: &[u8]) -> ManifestRoot {
use limnifs_core::SectionHashes;
let mut cursor = ManifestCursor::new(manifest);
let header_start = 0;
if parse_manifest_header(&mut cursor).is_err() {
return ManifestRoot::from_bytes([0u8; 32]);
}
let header_end = cursor.position();
let flags_start = header_end;
let flags_end = match limnifs_core::parse_feature_flags_section(&mut cursor) {
Ok(_) => cursor.position(),
Err(_) => flags_start,
};
let meta_ref_start = flags_end;
let metadata_reference = match limnifs_core::parse_metadata_reference(&mut cursor) {
Ok(m) => Some(m),
Err(_) => None,
};
let meta_ref_end = cursor.position();
let slab_index_start = meta_ref_end;
let _ = parse_slab_index(&mut cursor);
let slab_index_end = cursor.position();
let history_start = slab_index_end;
let _ = limnifs_core::parse_history(&mut cursor);
let history_end = cursor.position();
let hashes = SectionHashes {
metadata: metadata_reference
.map(|m| m.metadata_hash)
.unwrap_or_else(hash_empty_section),
format_header: hash_section(&manifest[header_start..header_end]),
feature_flags: hash_section(&manifest[flags_start..flags_end]),
metadata_reference: hash_section(&manifest[meta_ref_start..meta_ref_end]),
slab_index: hash_section(&manifest[slab_index_start..slab_index_end]),
crypto_params: hash_empty_section(),
ec_params: hash_empty_section(),
dms_policy: hash_empty_section(),
delta_linkage: hash_empty_section(),
history: hash_section(&manifest[history_start..history_end]),
};
compute_merkle_root(&hashes)
}
#[cfg(feature = "xattr")]
fn to_core_xattrs(raw: &[(String, Vec<u8>)]) -> Vec<limnifs_core::inode::XAttr> {
raw.iter()
.map(|(key, value)| limnifs_core::inode::XAttr {
namespace: 0,
key: key.clone(),
value: value.clone(),
})
.collect()
}
fn io_core(e: limnifs_core::CoreError) -> WriteError {
WriteError::Io(std::io::Error::other(format!("base image load: {e}")))
}
fn write_directory_body(ctx: &mut WriteContext, config: &WriteConfig) -> Result<(), WriteError> {
use rayon::prelude::*;
ctx.metadata_codec = config
.metadata_codec_id()
.unwrap_or(limnifs_core::codec::CODEC_BROTLI);
ctx.chunker = chunker_from_config(config)?;
let pending = std::mem::take(&mut ctx.pending_files);
if pending.is_empty() {
return Ok(());
}
ctx.inline_threshold = config.defaults.inline_threshold as usize;
ctx.metadata_externalize_threshold = config.defaults.metadata_externalize_threshold;
ctx.emit_shared_inline = config.defaults.shared_inline;
let chunker = ctx.chunker.clone();
let classifier = ctx.classifier;
let text_codec = config.text_codec_id().unwrap_or(0x04);
let binary_codec = config.binary_codec_id().unwrap_or(0x01);
let tunables = config.to_core_tunables();
let use_categorizers = !config.categorizers.is_empty();
let skip_chunking = config.skip_chunking;
let registry = config
.codec_registry()
.map_err(|e| WriteError::Io(std::io::Error::other(format!("codec registry: {e}"))))?;
let tournament_codec_ids: Vec<u8> = config
.tournament
.codecs
.iter()
.filter_map(|n| registry.lookup_by_name(n))
.collect();
let tournament_spec = TournamentSpec {
codec_ids: tournament_codec_ids,
min_size: config.tournament.min_size_threshold as usize,
skip_for_binary: config.tournament.skip_for_binary,
short_circuit_permille: config.tournament.short_circuit_threshold,
};
let base_drop_index: Option<&dyn BaseDropSet> = ctx.base_drop_index.as_deref();
let inline_threshold = ctx.inline_threshold;
let max_drop_size = config.defaults.max_drop_size as usize;
let seekable_drops = config.defaults.seekable_drops;
let seekable_drops = config.defaults.seekable_drops;
let results: Vec<ChunkedFileResult> = pending
.par_iter()
.map(|pf| {
process_file(
pf,
&chunker,
classifier,
text_codec,
binary_codec,
&tunables,
use_categorizers,
skip_chunking,
&tournament_spec,
base_drop_index,
inline_threshold,
max_drop_size,
seekable_drops,
config.categorizers.as_slice(),
&|name| {
config
.codec_registry()
.ok()
.and_then(|r| r.lookup_by_name(name))
},
)
})
.collect::<Result<Vec<_>, _>>()?;
for (pf, result) in pending.iter().zip(results) {
ctx.merge_chunked_file(pf, result);
}
ctx.train_and_apply_dictionary(&config.dictionaries);
Ok(())
}
pub fn write_directory_with_config(
root: &Path,
config: &WriteConfig,
) -> Result<WriteArtifact, WriteError> {
let mut ctx = WriteContext::new();
ctx.chunker = chunker_from_config(config)?;
ctx.categorizers_disabled = config.categorizers.is_empty();
ctx.rw_mode = matches!(config.mode, crate::config::ImageMode::ReadWrite(_));
ctx.auto_turnover = config.turnover_threshold > 0;
ctx.collect_dict_samples = config.dictionaries.enabled;
write_directory_streaming(&mut ctx, root, config)?;
Ok(ctx.assemble())
}
fn write_directory_streaming(
ctx: &mut WriteContext,
root: &Path,
config: &WriteConfig,
) -> Result<(), WriteError> {
use rayon::prelude::*;
ctx.metadata_codec = config
.metadata_codec_id()
.unwrap_or(limnifs_core::codec::CODEC_BROTLI);
ctx.chunker = chunker_from_config(config)?;
let chunker = ctx.chunker.clone();
let classifier = ctx.classifier;
let text_codec = config.text_codec_id().unwrap_or(0x04);
let binary_codec = config.binary_codec_id().unwrap_or(0x01);
let tunables = config.to_core_tunables();
let use_categorizers = !config.categorizers.is_empty();
let skip_chunking = config.skip_chunking;
let registry = config
.codec_registry()
.map_err(|e| WriteError::Io(std::io::Error::other(format!("codec registry: {e}"))))?;
let tournament_codec_ids: Vec<u8> = config
.tournament
.codecs
.iter()
.filter_map(|n| registry.lookup_by_name(n))
.collect();
let tournament_spec = TournamentSpec {
codec_ids: tournament_codec_ids,
min_size: config.tournament.min_size_threshold as usize,
skip_for_binary: config.tournament.skip_for_binary,
short_circuit_permille: config.tournament.short_circuit_threshold,
};
let base_drop_index = ctx.base_drop_index.clone();
let inline_threshold = ctx.inline_threshold;
let max_drop_size = config.defaults.max_drop_size as usize;
let seekable_drops = config.defaults.seekable_drops;
ctx.inline_threshold = config.defaults.inline_threshold as usize;
ctx.metadata_externalize_threshold = config.defaults.metadata_externalize_threshold;
ctx.emit_shared_inline = config.defaults.shared_inline;
const PIPELINE_CAPACITY: usize = 256;
let (tx, rx) = std::sync::mpsc::sync_channel::<PendingFile>(PIPELINE_CAPACITY);
ctx.pending_sink = Some(tx);
let (root_inode_number, mut results): (
u64,
Vec<(usize, PendingFile, Result<ChunkedFileResult, WriteError>)>,
) = {
let survey = survey_tree(root)?;
std::thread::scope(|scope| {
let producer = {
let ctx = &mut *ctx;
let root = root;
scope.spawn(move || {
let r = ctx.fold_survey(root, &survey, None);
ctx.pending_sink = None;
r
})
};
let results = rx
.into_iter()
.enumerate()
.par_bridge()
.map(|(i, pf)| {
let r = process_file(
&pf,
&chunker,
classifier,
text_codec,
binary_codec,
&tunables,
use_categorizers,
skip_chunking,
&tournament_spec,
base_drop_index.as_deref(),
inline_threshold,
max_drop_size,
seekable_drops,
config.categorizers.as_slice(),
&|name| {
config
.codec_registry()
.ok()
.and_then(|r| r.lookup_by_name(name))
},
);
(i, pf, r)
})
.collect();
let joined = producer
.join()
.unwrap_or_else(|_| {
Err(WriteError::Io(std::io::Error::other(
"walk thread panicked",
)))
})
.map(|n| (n, results));
joined
})
}?;
ctx.pending_sink = None;
ctx.root_inode_number = root_inode_number;
results.sort_unstable_by_key(|(i, _, _)| *i);
for (_, pf, r) in results {
ctx.merge_chunked_file(&pf, r?);
}
ctx.train_and_apply_dictionary(&config.dictionaries);
Ok(())
}
pub(crate) type RawDrop = ([u8; 32], Vec<u8>, std::sync::Arc<[u8]>, u8, u8);
pub(crate) struct ChunkedFileResult {
drops: Vec<RawDrop>, slices: Vec<PendingSlice>,
}
struct TournamentSpec {
codec_ids: Vec<u8>,
min_size: usize,
skip_for_binary: bool,
short_circuit_permille: u32,
}
fn chunker_from_config(config: &WriteConfig) -> Result<ParallelFastCDC, WriteError> {
ParallelFastCDC::new(
config.chunking.min_chunk_size as usize,
config.chunking.avg_chunk_size as usize,
config.chunking.max_chunk_size as usize,
)
.map_err(|e| WriteError::Io(std::io::Error::other(format!("chunking config: {e}"))))
}
pub(crate) fn seekable_or_monolithic(
codec: u8,
plaintext: &[u8],
compressed: std::sync::Arc<[u8]>,
tunables: &limnifs_core::codec::CodecTunables,
seekable_drops: bool,
threshold: usize,
) -> (std::sync::Arc<[u8]>, u8) {
use limnifs_core::seekable::{
encode_seekable, is_seekable_codec, DROP_FLAG_SEEKABLE as FLAG, SEEKABLE_EMISSION_THRESHOLD,
};
if seekable_drops && plaintext.len() > threshold && is_seekable_codec(codec) {
if let Ok(container) = encode_seekable(codec, plaintext, tunables) {
return (container.into(), FLAG);
}
}
(compressed, 0)
}
pub(crate) const SEEKABLE_CHUNK_EMISSION_THRESHOLD: usize =
limnifs_core::seekable::SEEKABLE_FRAME_SIZE;
fn process_whole_file_drop(
pf: &PendingFile,
data: &[u8],
cat: file_categorizer::Categorization,
tunables: &limnifs_core::codec::CodecTunables,
seekable_drops: bool,
) -> Result<ChunkedFileResult, WriteError> {
let _ = pf;
let drop_id = hash_section(data);
let file_len = u64::try_from(data.len()).unwrap_or(u64::MAX);
let (mut best_codec, mut best_compressed): (u8, std::sync::Arc<[u8]>) =
match limnifs_core::codec::compress_with_tunables(
limnifs_core::codec::CODEC_BROTLI,
data,
tunables,
) {
Ok(c) => (limnifs_core::codec::CODEC_BROTLI, c.into()),
Err(_) => match limnifs_core::codec::compress_with_tunables(
limnifs_core::codec::CODEC_ZSTD,
data,
tunables,
) {
Ok(c) => (limnifs_core::codec::CODEC_ZSTD, c.into()),
Err(_) => (limnifs_core::codec::CODEC_STORE, data.to_vec().into()),
},
};
let brotli_ratio = best_compressed.len() as f64 / data.len() as f64;
if brotli_ratio > 0.05 {
if let Ok(zstd_c) = limnifs_core::codec::compress_with_tunables(
limnifs_core::codec::CODEC_ZSTD,
data,
tunables,
) {
if zstd_c.len() < best_compressed.len() {
best_codec = limnifs_core::codec::CODEC_ZSTD;
best_compressed = zstd_c.into();
}
}
}
let general_ratio = best_compressed.len() as f64 / data.len() as f64;
if general_ratio > 0.15 || cat.codec_id == limnifs_core::codec::CODEC_RICEPP {
let spec_result = if cat.codec_id == limnifs_core::codec::CODEC_FSST_BROTLI {
limnifs_core::codec::fsst_brotli::compress_with_baseline(data, Some(&best_compressed))
} else {
limnifs_core::codec::compress_with_tunables(cat.codec_id, data, tunables)
};
if let Ok(spec_c) = spec_result {
if spec_c.len() < best_compressed.len() {
best_codec = cat.codec_id;
best_compressed = spec_c.into();
}
}
}
let (best_compressed, flags) = seekable_or_monolithic(
best_codec,
data,
best_compressed,
tunables,
seekable_drops,
limnifs_core::seekable::SEEKABLE_EMISSION_THRESHOLD,
);
Ok(ChunkedFileResult {
drops: vec![(drop_id, data.to_vec(), best_compressed, best_codec, flags)],
slices: vec![PendingSlice {
drop_id,
file_byte_start: 0,
file_byte_end: file_len,
}],
})
}
fn compress_chunk_with_tournament(
chunk: &[u8],
class: classifier::Class,
text_codec: u8,
binary_codec: u8,
tunables: &limnifs_core::codec::CodecTunables,
tournament: &TournamentSpec,
) -> (u8, std::sync::Arc<[u8]>) {
use classifier::Class;
let preferred = match class {
Class::Binary => binary_codec,
Class::Text | Class::Code | Class::Sparse => text_codec,
_ => limnifs_core::codec::CODEC_STORE,
};
if preferred == limnifs_core::codec::CODEC_STORE {
return (limnifs_core::codec::CODEC_STORE, chunk.to_vec().into());
}
if class == Class::Binary && tournament.skip_for_binary {
return compress_chunk_one(chunk, preferred, tunables);
}
if chunk.len() < tournament.min_size {
return compress_chunk_one(chunk, preferred, tunables);
}
let mut best: Option<(u8, std::sync::Arc<[u8]>)> = None;
for &codec_id in &tournament.codec_ids {
if codec_id == limnifs_core::codec::CODEC_STORE {
continue;
}
let c = match limnifs_core::codec::compress_with_tunables(codec_id, chunk, tunables) {
Ok(c) => c,
Err(_) => continue,
};
if c.len() >= chunk.len() {
continue;
}
let ratio_permille = (c.len() as u64 * 1000 / chunk.len() as u64) as u32;
let is_best_so_far = best.as_ref().map_or(true, |(_, b)| c.len() < b.len());
if is_best_so_far {
best = Some((codec_id, c.into()));
}
if tournament.short_circuit_permille > 0
&& ratio_permille <= tournament.short_circuit_permille
{
break;
}
}
best.unwrap_or_else(|| (limnifs_core::codec::CODEC_STORE, chunk.to_vec().into()))
}
fn compress_chunk_one(
chunk: &[u8],
codec_id: u8,
tunables: &limnifs_core::codec::CodecTunables,
) -> (u8, std::sync::Arc<[u8]>) {
if codec_id == limnifs_core::codec::CODEC_STORE {
return (limnifs_core::codec::CODEC_STORE, chunk.to_vec().into());
}
match limnifs_core::codec::compress_with_tunables(codec_id, chunk, tunables) {
Ok(c) if c.len() < chunk.len() => (codec_id, c.into()),
_ => (limnifs_core::codec::CODEC_STORE, chunk.to_vec().into()),
}
}
fn process_file(
pf: &PendingFile,
chunker: &dyn Chunker,
classifier: classifier::Classifier,
text_codec: u8,
binary_codec: u8,
tunables: &limnifs_core::codec::CodecTunables,
use_categorizers: bool,
skip_chunking: bool,
tournament: &TournamentSpec,
base_drop_index: Option<&dyn BaseDropSet>,
inline_threshold: usize,
max_drop_size: usize,
seekable_drops: bool,
categorizer_config: &[crate::config::CategorizerConfig],
codec_name_resolver: &dyn Fn(&str) -> Option<u8>,
) -> Result<ChunkedFileResult, WriteError> {
let file_len_estimate = std::fs::metadata(&pf.path)
.map(|m| m.len() as usize)
.unwrap_or(0);
let mmap_handle: memmap2::Mmap;
let small: Vec<u8>;
let data: &[u8] = if file_len_estimate >= MMAP_READ_THRESHOLD {
let file = std::fs::File::open(&pf.path)?;
#[allow(unsafe_code)]
let mapped = unsafe { memmap2::Mmap::map(&file) }.map_err(WriteError::Io)?;
mmap_handle = mapped;
&mmap_handle[..]
} else {
small = std::fs::read(&pf.path)?;
&small[..]
};
let file_len = data.len();
if skip_chunking && file_len > inline_threshold {
let drop_id = hash_section(&data);
let class = classifier.classify(&data);
let preferred_codec = match class {
classifier::Class::Binary => binary_codec,
_ => text_codec,
};
let (codec_id, compressed): (u8, std::sync::Arc<[u8]>) =
match limnifs_core::codec::compress_with_tunables(preferred_codec, &data, tunables) {
Ok(c) if c.len() < data.len() => (preferred_codec, c.into()),
_ => (limnifs_core::codec::CODEC_STORE, data.to_vec().into()),
};
let (compressed, flags) = seekable_or_monolithic(
codec_id,
&data,
compressed,
tunables,
seekable_drops,
SEEKABLE_CHUNK_EMISSION_THRESHOLD,
);
return Ok(ChunkedFileResult {
drops: vec![(drop_id, data.to_vec(), compressed, codec_id, flags)],
slices: vec![PendingSlice {
drop_id,
file_byte_start: 0,
file_byte_end: file_len as u64,
}],
});
}
if use_categorizers {
let config_cat = file_categorizer::ConfigCategorizer::new(categorizer_config.to_vec());
if let Some(cat) = config_cat.categorize(&pf.path, &data) {
if let Some(codec_id) = file_categorizer::resolve_config_categorization(
&cat,
categorizer_config,
codec_name_resolver,
) {
let within_cap = max_drop_size == 0 || file_len <= max_drop_size;
let needs_whole_file = matches!(
codec_id,
limnifs_core::codec::CODEC_FLAC | limnifs_core::codec::CODEC_RICEPP
);
if within_cap && (needs_whole_file || file_len <= WHOLE_FILE_MAX_SIZE) {
let mut cat = cat;
cat.codec_id = codec_id;
return process_whole_file_drop(pf, &data, cat, tunables, seekable_drops);
}
}
}
if let Some(cat) = file_categorizer::default_registry().categorize(&pf.path, &data) {
let needs_whole_file = matches!(
cat.codec_id,
limnifs_core::codec::CODEC_FLAC | limnifs_core::codec::CODEC_RICEPP
);
let within_cap = max_drop_size == 0 || file_len <= max_drop_size;
if within_cap && (needs_whole_file || file_len <= WHOLE_FILE_MAX_SIZE) {
return process_whole_file_drop(pf, &data, cat, tunables, seekable_drops);
}
}
}
let chunks = chunker.chunk_slice(&data);
use rayon::prelude::*;
let drop_ids: Vec<[u8; 32]> = chunks.par_iter().map(|chunk| hash_section(chunk)).collect();
let mut slices = Vec::with_capacity(chunks.len());
let mut file_offset: u64 = 0;
let mut seen_in_file: std::collections::HashSet<[u8; 32]> =
std::collections::HashSet::with_capacity(chunks.len());
let mut unique_chunks: Vec<(&[u8], [u8; 32])> = Vec::with_capacity(chunks.len());
for (chunk, drop_id) in chunks.iter().zip(drop_ids) {
let chunk_len = u64::try_from(chunk.len()).expect("chunk len fits u64");
slices.push(PendingSlice {
drop_id,
file_byte_start: file_offset,
file_byte_end: file_offset + chunk_len,
});
file_offset += chunk_len;
if seen_in_file.insert(drop_id) {
unique_chunks.push((chunk, drop_id));
}
}
thread_local! {
static COMPRESS_CACHE: std::cell::RefCell<std::collections::HashMap<[u8; 32], (u8, std::sync::Arc<[u8]>)>> =
std::cell::RefCell::new(std::collections::HashMap::new());
}
const COMPRESS_CACHE_MAX_ENTRIES: usize = 100_000;
let drops: Vec<RawDrop> = unique_chunks
.par_iter()
.map(|(chunk, drop_id)| {
if let Some(base) = base_drop_index {
if base.base_contains(drop_id) {
return (*drop_id, Vec::new(), Vec::new().into(), CODEC_REFERENCED, 0);
}
}
let class = classifier.classify(chunk);
let cached = COMPRESS_CACHE.with(|c| {
c.borrow()
.get(drop_id)
.map(|(cid, comp)| (*cid, comp.clone()))
});
let (codec_id, compressed) = if let Some(c) = cached {
c
} else {
let new = compress_chunk_with_tournament(
chunk,
class,
text_codec,
binary_codec,
tunables,
tournament,
);
COMPRESS_CACHE.with(|c| {
let mut cache = c.borrow_mut();
if cache.len() < COMPRESS_CACHE_MAX_ENTRIES {
cache.insert(*drop_id, new.clone());
}
});
new
};
let (compressed, flags) = seekable_or_monolithic(
codec_id,
chunk,
compressed,
tunables,
seekable_drops,
SEEKABLE_CHUNK_EMISSION_THRESHOLD,
);
(*drop_id, chunk.to_vec(), compressed, codec_id, flags)
})
.collect();
let _ = file_len;
Ok(ChunkedFileResult { drops, slices })
}
struct PendingDrop {
id: [u8; 32],
plaintext_len: u32,
compressed: std::sync::Arc<[u8]>,
codec: u8,
dict_id: u8,
plaintext: Option<Vec<u8>>,
flags: u8,
}
impl PendingDrop {
fn len_in_window(&self) -> u32 {
u32::try_from(self.compressed.len()).expect("compressed fits u32")
}
fn plaintext_len_value(&self) -> u32 {
self.plaintext_len
}
fn slab_footprint(&self) -> usize {
48 + self.compressed.len()
}
}
struct PendingSlice {
drop_id: [u8; 32],
file_byte_start: u64,
file_byte_end: u64,
}
#[derive(Clone)]
struct PendingFile {
inode_number: u64,
path: PathBuf,
mtime_ns: u64,
file_len: u64,
mode: u32,
uid: u32,
gid: u32,
}
#[derive(Clone, Default)]
struct SurveyMeta {
is_dir: bool,
is_file: bool,
is_symlink: bool,
#[cfg(unix)]
is_fifo: bool,
#[cfg(unix)]
is_socket: bool,
#[cfg(unix)]
is_block_device: bool,
#[cfg(unix)]
is_char_device: bool,
len: u64,
mtime_ns: u64,
#[cfg(unix)]
mode: u32,
#[cfg(unix)]
uid: u32,
#[cfg(unix)]
gid: u32,
#[cfg(unix)]
dev: u64,
#[cfg(unix)]
ino: u64,
#[cfg(feature = "xattr")]
xattrs: Vec<(String, Vec<u8>)>,
}
impl SurveyMeta {
fn identity(&self) -> (u32, u32, u32) {
#[cfg(unix)]
{
(self.mode, self.uid, self.gid)
}
#[cfg(not(unix))]
{
let ty = if self.is_dir {
limnifs_core::inode::S_IFDIR
} else if self.is_symlink {
limnifs_core::inode::S_IFLNK
} else {
limnifs_core::inode::S_IFREG
};
let perms = if self.is_dir || self.is_symlink {
0o755
} else {
0o644
};
(ty | perms, 0, 0)
}
}
}
struct SurveyNode {
meta: SurveyMeta,
children: Vec<(String, SurveyNode)>,
symlink_target: Option<String>,
}
impl SurveyNode {
fn meta(&self) -> &SurveyMeta {
&self.meta
}
}
fn survey_meta_of(meta: &std::fs::Metadata) -> SurveyMeta {
#[cfg(unix)]
use std::os::unix::fs::FileTypeExt as _;
let ft = meta.file_type();
let mtime_ns = meta
.modified()
.ok()
.and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
.map_or(0u128, |d| d.as_nanos());
SurveyMeta {
is_dir: ft.is_dir(),
is_file: ft.is_file(),
is_symlink: ft.is_symlink(),
#[cfg(unix)]
is_fifo: ft.is_fifo(),
#[cfg(unix)]
is_socket: ft.is_socket(),
#[cfg(unix)]
is_block_device: ft.is_block_device(),
#[cfg(unix)]
is_char_device: ft.is_char_device(),
len: meta.len(),
mtime_ns: mtime_ns.try_into().unwrap_or(0),
#[cfg(unix)]
mode: {
use std::os::unix::fs::MetadataExt as _;
meta.mode()
},
#[cfg(unix)]
uid: {
use std::os::unix::fs::MetadataExt as _;
meta.uid()
},
#[cfg(unix)]
gid: {
use std::os::unix::fs::MetadataExt as _;
meta.gid()
},
#[cfg(unix)]
dev: {
use std::os::unix::fs::MetadataExt as _;
meta.dev()
},
#[cfg(unix)]
ino: {
use std::os::unix::fs::MetadataExt as _;
meta.ino()
},
#[cfg(feature = "xattr")]
xattrs: Vec::new(),
}
}
#[cfg(all(unix, feature = "xattr"))]
fn collect_xattrs(path: &Path) -> Vec<(String, Vec<u8>)> {
const VOLATILE: &[&str] = &[
"com.apple.provenance",
"com.apple.quarantine",
"com.apple.lastuseddate",
"com.apple.macl",
"com.apple.filesec",
];
const TOTAL_CAP: usize = 64 * 1024;
let Ok(names) = xattr::list(path) else {
return Vec::new();
};
let mut names: Vec<String> = names
.filter_map(|n| n.into_string().ok())
.filter(|n| {
!n.starts_with("system.")
&& !n.starts_with("security.")
&& !n.starts_with("trusted.")
&& !VOLATILE.contains(&n.as_str())
})
.collect();
names.sort();
let mut out = Vec::new();
let mut total = 0usize;
for name in names {
let Ok(Some(value)) = xattr::get(path, &name) else {
continue;
};
total += name.len() + value.len();
if total > TOTAL_CAP {
break;
}
out.push((name, value));
}
out
}
fn survey_node(path: &Path) -> Result<SurveyNode, WriteError> {
use rayon::prelude::*;
let meta = std::fs::symlink_metadata(path)?;
let mut sm = survey_meta_of(&meta);
#[cfg(all(unix, feature = "xattr"))]
{
if !sm.is_symlink {
sm.xattrs = collect_xattrs(path);
}
}
if sm.is_symlink {
let target = std::fs::read_link(path)?;
let target = target
.to_str()
.ok_or_else(|| WriteError::UnsupportedFileType {
path: path.to_path_buf(),
kind: format!("symlink with non-UTF-8 target ({})", target.display()),
})?
.to_owned();
return Ok(SurveyNode {
meta: sm,
children: Vec::new(),
symlink_target: Some(target),
});
}
if !sm.is_dir {
return Ok(SurveyNode {
meta: sm,
children: Vec::new(),
symlink_target: None,
});
}
let mut named: Vec<(String, PathBuf)> = std::fs::read_dir(path)?
.filter_map(|entry| {
entry
.ok()
.map(|e| (e.file_name().to_string_lossy().into_owned(), e.path()))
})
.collect();
named.sort_by(|a, b| a.0.cmp(&b.0));
named
.par_iter()
.map(|(name, child)| survey_node(child).map(|node| (name.clone(), node)))
.collect::<Result<Vec<_>, WriteError>>()
.map(|children| SurveyNode {
meta: sm,
children,
symlink_target: None,
})
}
fn survey_tree(root: &Path) -> Result<SurveyNode, WriteError> {
survey_node(root)
}
struct PendingInode {
number: u64,
mode: u32,
uid: u32,
gid: u32,
mtime_ns: u64,
xattrs: Vec<limnifs_core::inode::XAttr>,
content: PendingContent,
}
enum PendingContent {
Inline(Vec<u8>),
Symlink(String),
DropBacked {
file_len: u64,
slices: Vec<PendingSlice>,
},
Directory(Vec<(String, u64, u8)>),
}
struct DirNode {
entries: Vec<(String, u64, u8)>,
bytes: Vec<u8>,
hash: [u8; 32],
}
struct WriteContext {
next_inode: u64,
hardlink_targets: std::collections::HashMap<(u64, u64), u64>,
nlink_counts: std::collections::HashMap<u64, u32>,
inode_xattrs: std::collections::HashMap<u64, Vec<limnifs_core::inode::XAttr>>,
inodes: Vec<PendingInode>,
dir_nodes: Vec<DirNode>,
drops: Vec<PendingDrop>,
drop_index: HashSet<[u8; 32]>,
pending_files: Vec<PendingFile>,
file_count: usize,
dir_count: usize,
root_inode_number: u64,
chunker: ParallelFastCDC,
classifier: classifier::Classifier,
shared_inline_map: HashMap<[u8; 32], usize>,
shared_inline_table: Vec<Vec<u8>>,
base_dictionaries: Option<Vec<crate::dictionary::TrainedDictionary>>,
profile_name: Option<String>,
metadata_codec: u8,
categorizers_disabled: bool,
rw_mode: bool,
auto_turnover: bool,
collect_dict_samples: bool,
dict_samples_by_class: HashMap<crate::classifier::Class, Vec<Vec<u8>>>,
trained_dicts_by_class: HashMap<crate::classifier::Class, crate::dictionary::TrainedDictionary>,
base_drop_index: Option<std::sync::Arc<dyn BaseDropSet>>,
base_root: Option<[u8; 32]>,
metadata_externalize_threshold: usize,
emit_shared_inline: bool,
inline_threshold: usize,
pending_sink: Option<std::sync::mpsc::SyncSender<PendingFile>>,
}
impl WriteContext {
const MAX_DICT_SAMPLES: usize = 1000;
fn new() -> Self {
Self {
next_inode: 1,
hardlink_targets: std::collections::HashMap::new(),
nlink_counts: std::collections::HashMap::new(),
inode_xattrs: std::collections::HashMap::new(),
inodes: Vec::new(),
dir_nodes: Vec::new(),
drops: Vec::new(),
drop_index: HashSet::new(),
pending_files: Vec::new(),
file_count: 0,
dir_count: 0,
root_inode_number: 0,
chunker: ParallelFastCDC::default(),
classifier: classifier::Classifier,
shared_inline_map: HashMap::new(),
shared_inline_table: Vec::new(),
base_dictionaries: None,
profile_name: None,
metadata_codec: limnifs_core::codec::CODEC_BROTLI,
categorizers_disabled: false,
rw_mode: false,
auto_turnover: false,
collect_dict_samples: false,
dict_samples_by_class: HashMap::new(),
trained_dicts_by_class: HashMap::new(),
base_drop_index: None,
base_root: None,
pending_sink: None,
inline_threshold: INLINE_THRESHOLD,
metadata_externalize_threshold: METADATA_EXTERNALIZE_THRESHOLD,
emit_shared_inline: true,
}
}
fn alloc_inode(&mut self) -> u64 {
let n = self.next_inode;
self.next_inode += 1;
n
}
fn build_shared_inline_table(&mut self) {
let mut counts: HashMap<[u8; 32], usize> = HashMap::new();
for inode in &self.inodes {
if let PendingContent::Inline(data) = &inode.content {
let h = hash_section(data);
*counts.entry(h).or_default() += 1;
}
}
for inode in &self.inodes {
if let PendingContent::Inline(data) = &inode.content {
let h = hash_section(data);
if counts.get(&h).copied().unwrap_or(0) > 1
&& !self.shared_inline_map.contains_key(&h)
{
let idx = self.shared_inline_table.len();
self.shared_inline_table.push(data.clone());
self.shared_inline_map.insert(h, idx);
}
}
}
}
fn merge_chunked_file(&mut self, pf: &PendingFile, result: ChunkedFileResult) {
for (drop_id, plaintext, compressed, codec, flags) in result.drops {
if self.drop_index.insert(drop_id) {
let retain_plaintext =
self.collect_dict_samples && codec == limnifs_core::codec::CODEC_ZSTD;
if retain_plaintext {
let total: usize = self.dict_samples_by_class.values().map(Vec::len).sum();
if total < Self::MAX_DICT_SAMPLES {
let class = self.classifier.classify(&plaintext);
self.dict_samples_by_class
.entry(class)
.or_default()
.push(plaintext.clone());
}
}
self.drops.push(PendingDrop {
id: drop_id,
plaintext_len: u32::try_from(plaintext.len()).unwrap_or(u32::MAX),
compressed,
codec,
dict_id: limnifs_core::drop_record::NO_DICT,
plaintext: if retain_plaintext {
Some(plaintext)
} else {
None
},
flags,
});
}
}
self.inodes.push(PendingInode {
number: pf.inode_number,
mode: pf.mode,
uid: pf.uid,
gid: pf.gid,
mtime_ns: pf.mtime_ns,
xattrs: Vec::new(),
content: PendingContent::DropBacked {
file_len: pf.file_len,
slices: result.slices,
},
});
}
#[allow(dead_code)]
fn deepen_drop(&self, drop_id: [u8; 32], plaintext: &[u8]) -> PendingDrop {
let class = self.classifier.classify(plaintext);
let (codec, compressed): (u8, std::sync::Arc<[u8]>) = match class {
classifier::Class::Text | classifier::Class::Code | classifier::Class::Binary => {
let c = limnifs_core::codec::compress_lz4_with_size(plaintext);
(limnifs_core::codec::CODEC_LZ4, c.into())
}
_ => (limnifs_core::codec::CODEC_STORE, plaintext.to_vec().into()),
};
PendingDrop {
id: drop_id,
plaintext_len: u32::try_from(plaintext.len()).unwrap_or(u32::MAX),
compressed,
codec,
dict_id: limnifs_core::drop_record::NO_DICT,
plaintext: None,
flags: 0,
}
}
fn walk(&mut self, path: &Path) -> Result<u64, WriteError> {
let survey = survey_tree(path)?;
self.fold_survey(path, &survey, None)
}
fn fold_survey(
&mut self,
path: &Path,
node: &SurveyNode,
symlink_target: Option<&str>,
) -> Result<u64, WriteError> {
let meta = node.meta();
if let Some(target) = symlink_target {
let inode_number = self.alloc_inode();
let (mode, uid, gid) = meta.identity();
self.inodes.push(PendingInode {
number: inode_number,
mode,
uid,
gid,
mtime_ns: meta.mtime_ns,
xattrs: Vec::new(),
content: PendingContent::Symlink(target.to_owned()),
});
return Ok(inode_number);
}
if meta.is_dir {
self.dir_count += 1;
let inode_number = self.alloc_inode();
let mut entries: Vec<(String, u64, u8)> = Vec::new();
for (name, child) in &node.children {
let child_path = path.join(name);
let child_inode =
self.fold_survey(&child_path, child, child.symlink_target.as_deref())?;
let entry_type = if child.meta().is_symlink {
0x03
} else if child.meta().is_dir {
0x02
} else {
0x01
};
entries.push((name.clone(), child_inode, entry_type));
}
entries.sort_by(|a, b| a.0.cmp(&b.0));
let dir_node = encode_dir_node(&entries);
self.dir_nodes.push(dir_node);
#[cfg(feature = "xattr")]
if !meta.xattrs.is_empty() {
self.inode_xattrs
.insert(inode_number, to_core_xattrs(&meta.xattrs));
}
let (mode, uid, gid) = meta.identity();
self.inodes.push(PendingInode {
number: inode_number,
mode,
uid,
gid,
mtime_ns: meta.mtime_ns,
xattrs: Vec::new(),
content: PendingContent::Directory(entries),
});
Ok(inode_number)
} else if meta.is_file {
#[cfg(unix)]
if let Some(&existing) = self.hardlink_targets.get(&(meta.dev, meta.ino)) {
*self.nlink_counts.entry(existing).or_insert(1) += 1;
return Ok(existing);
}
self.file_count += 1;
let inode_number = self.alloc_inode();
#[cfg(unix)]
{
self.hardlink_targets
.insert((meta.dev, meta.ino), inode_number);
}
let file_len = meta.len;
crate::progress::emit_file(path, file_len);
if file_len <= u64::try_from(self.inline_threshold).unwrap_or(u64::MAX) {
let data = std::fs::read(path)?;
#[cfg(feature = "xattr")]
if !meta.xattrs.is_empty() {
self.inode_xattrs
.insert(inode_number, to_core_xattrs(&meta.xattrs));
}
let (mode, uid, gid) = meta.identity();
self.inodes.push(PendingInode {
number: inode_number,
mode,
uid,
gid,
mtime_ns: meta.mtime_ns,
xattrs: Vec::new(),
content: PendingContent::Inline(data),
});
} else {
#[cfg(feature = "xattr")]
if !meta.xattrs.is_empty() {
self.inode_xattrs
.insert(inode_number, to_core_xattrs(&meta.xattrs));
}
let (mode, uid, gid) = meta.identity();
let pf = PendingFile {
inode_number,
path: path.to_path_buf(),
mtime_ns: meta.mtime_ns,
file_len,
mode,
uid,
gid,
};
if let Some(sink) = &self.pending_sink {
sink.send(pf).map_err(|_| {
WriteError::Io(std::io::Error::other("walk: compress pipeline shut down"))
})?;
} else {
self.pending_files.push(pf);
}
}
Ok(inode_number)
} else {
#[cfg(unix)]
let kind = {
use std::os::unix::fs::FileTypeExt;
if meta.is_fifo {
"fifo".to_owned()
} else if meta.is_socket {
"socket".to_owned()
} else if meta.is_block_device {
"block device".to_owned()
} else if meta.is_char_device {
"character device".to_owned()
} else {
"unknown".to_owned()
}
};
#[cfg(not(unix))]
let kind = "unknown".to_owned();
Err(WriteError::UnsupportedFileType {
path: path.to_path_buf(),
kind,
})
}
}
fn train_and_apply_dictionary(&mut self, dictionaries: &crate::config::DictionaryConfig) {
if !dictionaries.enabled {
Self::release_dictionary_samples(self);
return;
}
if let Some(adopted) = self.base_dictionaries.take() {
for dict in adopted {
match dict.id {
0 => {
self.trained_dicts_by_class
.insert(crate::classifier::Class::Text, dict);
}
1 => {
self.trained_dicts_by_class
.insert(crate::classifier::Class::Binary, dict);
}
_ => {}
}
}
self.apply_trained_dictionaries();
return;
}
let target = usize::try_from(dictionaries.max_dict_size).unwrap_or(65_536);
let min_class = usize::try_from(dictionaries.min_class_size).unwrap_or(0);
let text_classes = [
crate::classifier::Class::Text,
crate::classifier::Class::Code,
crate::classifier::Class::Sparse,
];
let binary_classes = [crate::classifier::Class::Binary];
let text_samples: Vec<&[u8]> = text_classes
.iter()
.flat_map(|c| self.dict_samples_by_class.get(c).into_iter().flatten())
.map(Vec::as_slice)
.collect();
let trainer = crate::dictionary::TrainerKind::from_config_str(&dictionaries.trainer);
if text_samples.len() >= min_class {
if let Some(dict) =
crate::dictionary::train_zstd_with_trainer(0, &text_samples, target, trainer)
{
self.trained_dicts_by_class
.insert(crate::classifier::Class::Text, dict);
}
}
let binary_samples: Vec<&[u8]> = binary_classes
.iter()
.flat_map(|c| self.dict_samples_by_class.get(c).into_iter().flatten())
.map(Vec::as_slice)
.collect();
if binary_samples.len() >= min_class {
if let Some(dict) =
crate::dictionary::train_zstd_with_trainer(1, &binary_samples, target, trainer)
{
self.trained_dicts_by_class
.insert(crate::classifier::Class::Binary, dict);
}
}
self.apply_trained_dictionaries();
}
fn apply_trained_dictionaries(&mut self) {
let text_classes = [
crate::classifier::Class::Text,
crate::classifier::Class::Code,
crate::classifier::Class::Sparse,
];
let binary_classes = [crate::classifier::Class::Binary];
use rayon::prelude::*;
let classifier = self.classifier;
let dicts = &self.trained_dicts_by_class;
let candidates: Vec<Option<(std::sync::Arc<[u8]>, u8)>> = self
.drops
.par_iter()
.map(|d| {
if d.codec != limnifs_core::codec::CODEC_ZSTD {
return None;
}
let Some(plaintext) = d.plaintext.as_ref() else {
return None;
};
let class = classifier.classify(plaintext);
let dict_class = if text_classes.contains(&class) {
crate::classifier::Class::Text
} else if binary_classes.contains(&class) {
crate::classifier::Class::Binary
} else {
return None;
};
let Some(dict) = dicts.get(&dict_class) else {
return None;
};
let Ok(dict_compressed) = dict.compress(plaintext) else {
return None;
};
if dict_compressed.len() < d.compressed.len() {
Some((dict_compressed.into(), dict.id))
} else {
None
}
})
.collect();
let saving: isize = candidates
.iter()
.zip(self.drops.iter())
.map(|(c, d)| {
c.as_ref().map_or(0, |(bytes, _)| {
d.compressed.len() as isize - bytes.len() as isize
})
})
.sum();
let dict_bytes: usize = dicts.values().map(|d| d.content.len()).sum();
if saving > dict_bytes as isize {
for (d, candidate) in self.drops.iter_mut().zip(candidates) {
if let Some((bytes, dict_id)) = candidate {
d.compressed = bytes;
d.dict_id = dict_id;
}
}
} else {
self.trained_dicts_by_class.clear();
}
Self::release_dictionary_samples(self);
}
fn release_dictionary_samples(ctx: &mut Self) {
for d in &mut ctx.drops {
d.plaintext = None;
}
ctx.dict_samples_by_class.clear();
}
fn trace_phase(label: &str, start: std::time::Instant) {
if std::env::var_os("LIMNIFS_TRACE_ASSEMBLE").is_some() {
eprintln!("[assemble] {label}: {:?}", start.elapsed());
}
}
fn assemble(mut self) -> WriteArtifact {
let t_assemble = std::time::Instant::now();
let inode_count = self.inodes.len();
let dir_count = self.dir_count;
let drop_count = self.drops.len();
let t = std::time::Instant::now();
let slabs = pack_slabs(&self.drops);
Self::trace_phase("pack_slabs", t);
let t = std::time::Instant::now();
if self.emit_shared_inline {
self.build_shared_inline_table();
}
Self::trace_phase("shared_inline_table", t);
let mut metadata_blob = Vec::new();
metadata_blob.extend_from_slice(&u32::try_from(self.inodes.len()).unwrap().to_le_bytes());
for inode in &self.inodes {
self.encode_inode(&mut metadata_blob, inode);
}
metadata_blob
.extend_from_slice(&u32::try_from(self.dir_nodes.len()).unwrap().to_le_bytes());
for node in &self.dir_nodes {
metadata_blob.extend_from_slice(&node.bytes);
}
if !self.shared_inline_table.is_empty() {
metadata_blob.extend_from_slice(
&u32::try_from(self.shared_inline_table.len())
.unwrap()
.to_le_bytes(),
);
for entry in &self.shared_inline_table {
let len = u32::try_from(entry.len()).expect("shared entry fits u32");
metadata_blob.extend_from_slice(&len.to_le_bytes());
metadata_blob.extend_from_slice(entry);
}
}
Self::trace_phase("metadata_encode", t);
let uncompressed_len =
u32::try_from(metadata_blob.len()).expect("metadata blob length fits u32");
let t = std::time::Instant::now();
let metadata_hash = hash_section(&metadata_blob);
let metadata_codec = self.metadata_codec;
let metadata_quality = if metadata_blob.len() > METADATA_LARGE_BLOB_THRESHOLD {
METADATA_LARGE_BLOB_QUALITY
} else {
METADATA_SMALL_BLOB_QUALITY
};
let compressed_blob = if metadata_codec == limnifs_core::codec::CODEC_BROTLI {
limnifs_core::codec::compress_brotli_with_quality(&metadata_blob, metadata_quality)
.unwrap_or_else(|_| metadata_blob.clone())
} else {
limnifs_core::codec::compress(metadata_codec, &metadata_blob)
.unwrap_or_else(|_| metadata_blob.clone())
};
Self::trace_phase("metadata_compress", t);
let (on_wire_codec, on_wire_blob) = if compressed_blob.len() < metadata_blob.len() {
(metadata_codec, compressed_blob)
} else {
(limnifs_core::codec::CODEC_STORE, metadata_blob.clone())
};
let externalize_at = self
.metadata_externalize_threshold
.min(limnifs_core::metadata_reference::DEFAULT_INLINE_METADATA_MAX_BYTES as usize);
let (metadata_sidecar, inline_data, metadata_locator_count) =
if on_wire_blob.len() > externalize_at {
let h = hash_section(&on_wire_blob);
let mut h8 = String::with_capacity(8);
for b in &h[..4] {
h8.push_str(&format!("{b:02x}"));
}
let locator = format!("file:metadata-{h8}.bin");
let sidecar = MetadataSidecar {
bytes: on_wire_blob.clone(),
locator,
};
(Some(sidecar), None, 1u32)
} else {
(None, Some(on_wire_blob.clone()), 0u32)
};
let mut manifest = Vec::new();
let header_start = manifest.len();
manifest.extend_from_slice(&ManifestHeader::current().to_bytes());
let header_end = manifest.len();
let flags_start = manifest.len();
manifest.push(FEATURE_FLAGS_SECTION_VERSION);
manifest.extend_from_slice(&0u32.to_le_bytes());
let flags_end = manifest.len();
let meta_ref_start = manifest.len();
manifest.push(METADATA_REFERENCE_SECTION_VERSION_2);
manifest.extend_from_slice(&metadata_hash);
manifest.extend_from_slice(&uncompressed_len.to_le_bytes());
manifest.push(on_wire_codec);
manifest.extend_from_slice(&metadata_locator_count.to_le_bytes());
if let Some(sidecar) = &metadata_sidecar {
let loc_bytes = sidecar.locator.as_bytes();
let loc_len = u32::try_from(loc_bytes.len()).expect("locator fits u32");
manifest.extend_from_slice(&loc_len.to_le_bytes());
manifest.extend_from_slice(loc_bytes);
}
match &inline_data {
Some(blob) => {
let inline_len = u32::try_from(blob.len()).expect("metadata fits u32");
manifest.extend_from_slice(&inline_len.to_le_bytes());
manifest.extend_from_slice(blob);
}
None => {
manifest.extend_from_slice(&0u32.to_le_bytes());
}
}
let meta_ref_end = manifest.len();
let slab_index_start = manifest.len();
manifest.push(SLAB_INDEX_SECTION_VERSION);
manifest.extend_from_slice(&u32::try_from(slabs.len()).unwrap().to_le_bytes());
for slab in &slabs {
manifest.extend_from_slice(&slab.id.to_bytes());
manifest.extend_from_slice(&1u32.to_le_bytes());
let loc_bytes = slab.locator.as_bytes();
let loc_len = u32::try_from(loc_bytes.len()).expect("locator fits u32");
manifest.extend_from_slice(&loc_len.to_le_bytes());
manifest.extend_from_slice(loc_bytes);
}
let slab_index_end = manifest.len();
let history_start = manifest.len();
manifest.push(HISTORY_SECTION_VERSION);
manifest.extend_from_slice(&1u32.to_le_bytes());
manifest.push(0x01);
manifest.extend_from_slice(&0u64.to_le_bytes());
manifest.extend_from_slice(&0u32.to_le_bytes());
manifest.extend_from_slice(&0u32.to_le_bytes());
let history_end = manifest.len();
let profile_desc_start = manifest.len();
if let Some(ref name) = self.profile_name {
let desc = limnifs_core::profile_descriptor::ProfileDescriptor {
version: limnifs_core::profile_descriptor::PROFILE_DESCRIPTOR_SECTION_VERSION,
profile_name: Some(name.clone()),
blake3_hashing: true,
cross_file_dedup: true,
content_classification: !self.categorizers_disabled,
integrity_verify: true,
read_write: self.rw_mode,
auto_turnover: self.auto_turnover,
};
limnifs_core::profile_descriptor::encode_profile_descriptor(&desc, &mut manifest);
}
let profile_desc_end = manifest.len();
if !self.trained_dicts_by_class.is_empty() {
let dicts: Vec<_> = self
.trained_dicts_by_class
.values()
.map(|d| limnifs_core::dictionary_section::Dictionary {
codec_id: d.codec,
class_id: d.id,
data: d.content.clone(),
})
.collect();
let section = limnifs_core::dictionary_section::DictionarySection {
version: limnifs_core::dictionary_section::DICTIONARY_SECTION_VERSION,
dicts,
};
limnifs_core::dictionary_section::encode_dictionary_section(§ion, &mut manifest);
}
let dictionary_end = manifest.len();
let delta_linkage_hash = if let Some(base_root) = self.base_root {
let delta_start = manifest.len();
manifest.push(limnifs_core::delta_linkage::DELTA_LINKAGE_SECTION_VERSION);
manifest.extend_from_slice(&base_root);
manifest.extend_from_slice(&0u32.to_le_bytes());
hash_section(&manifest[delta_start..])
} else {
hash_empty_section()
};
let _ = dictionary_end;
let hashes = SectionHashes {
metadata: metadata_hash,
format_header: hash_section(&manifest[header_start..header_end]),
feature_flags: hash_section(&manifest[flags_start..flags_end]),
metadata_reference: hash_section(&manifest[meta_ref_start..meta_ref_end]),
slab_index: hash_section(&manifest[slab_index_start..slab_index_end]),
crypto_params: hash_empty_section(),
ec_params: hash_empty_section(),
dms_policy: hash_empty_section(),
delta_linkage: delta_linkage_hash,
history: hash_section(&manifest[history_start..history_end]),
};
let merkle_root = compute_merkle_root(&hashes);
WriteArtifact {
bytes: manifest,
merkle_root,
slabs,
metadata_sidecar,
inode_count,
file_count: self.file_count,
dir_count,
drop_count,
root_inode_number: self.root_inode_number,
}
}
fn encode_inode(&self, out: &mut Vec<u8>, inode: &PendingInode) {
out.extend_from_slice(&inode.number.to_le_bytes());
out.extend_from_slice(&inode.mode.to_le_bytes());
out.extend_from_slice(&inode.uid.to_le_bytes());
out.extend_from_slice(&inode.gid.to_le_bytes());
out.extend_from_slice(&inode.mtime_ns.to_le_bytes());
out.extend_from_slice(&inode.mtime_ns.to_le_bytes());
let nlink = self.nlink_counts.get(&inode.number).copied().unwrap_or(1);
out.extend_from_slice(&nlink.to_le_bytes());
let xattrs: &[limnifs_core::inode::XAttr] = self
.inode_xattrs
.get(&inode.number)
.map_or(inode.xattrs.as_slice(), std::convert::AsRef::as_ref);
let mut flags_and_xattrs = move |out: &mut Vec<u8>, base: u8| {
if xattrs.is_empty() {
out.push(base);
return;
}
out.push(base | limnifs_core::inode::INODE_FLAG_HAS_XATTRS);
let count = u32::try_from(xattrs.len()).expect("xattr count fits u32");
out.extend_from_slice(&count.to_le_bytes());
for x in xattrs {
out.push(x.namespace);
let key = x.key.as_bytes();
let key_len = u32::try_from(key.len()).expect("xattr key fits u32");
out.extend_from_slice(&key_len.to_le_bytes());
out.extend_from_slice(key);
let value_len = u32::try_from(x.value.len()).expect("xattr value fits u32");
out.extend_from_slice(&value_len.to_le_bytes());
out.extend_from_slice(&x.value);
}
};
match &inode.content {
PendingContent::Inline(data) => {
let h = hash_section(data);
if let Some(&idx) = self.shared_inline_map.get(&h) {
flags_and_xattrs(out, INODE_FLAG_INLINE_DATA | INODE_FLAG_SHARED_INLINE);
out.extend_from_slice(&(idx as u32).to_le_bytes());
} else {
flags_and_xattrs(out, INODE_FLAG_INLINE_DATA);
let len = u32::try_from(data.len()).expect("data fits u32");
out.extend_from_slice(&len.to_le_bytes());
out.extend_from_slice(data);
}
}
PendingContent::DropBacked { file_len, slices } => {
flags_and_xattrs(out, 0x00);
let slice_count = u32::try_from(slices.len()).expect("slice count fits u32");
out.extend_from_slice(&slice_count.to_le_bytes());
for slice in slices {
out.extend_from_slice(&slice.file_byte_start.to_le_bytes());
out.extend_from_slice(&slice.file_byte_end.to_le_bytes());
out.extend_from_slice(&slice.drop_id);
out.extend_from_slice(&0u32.to_le_bytes());
let drop_byte_len = u32::try_from(slice.file_byte_end - slice.file_byte_start)
.expect("slice range fits u32");
out.extend_from_slice(&drop_byte_len.to_le_bytes());
}
let _ = file_len;
}
PendingContent::Symlink(target) => {
flags_and_xattrs(out, 0x00);
let t = target.as_bytes();
let len = u32::try_from(t.len()).expect("target fits u32");
out.extend_from_slice(&len.to_le_bytes());
out.extend_from_slice(t);
}
PendingContent::Directory(entries) => {
flags_and_xattrs(out, 0x00);
let node = self
.dir_nodes
.iter()
.find(|n| n.entries == *entries)
.expect("directory node must exist");
out.extend_from_slice(&node.hash);
}
}
}
}
fn sidecar_name(locator: &str) -> Result<&str, WriteError> {
limnifs_core::locator::local_sidecar_name(locator)
.map_err(|e| WriteError::Io(std::io::Error::other(format!("{e}"))))
}
fn encode_dir_node(entries: &[(String, u64, u8)]) -> DirNode {
let mut bytes = Vec::new();
bytes.push(1u8);
let count = u32::try_from(entries.len()).expect("entry count fits u32");
bytes.extend_from_slice(&count.to_le_bytes());
for (name, inode_number, entry_type) in entries {
let name_bytes = name.as_bytes();
let name_len = u32::try_from(name_bytes.len()).expect("name fits u32");
bytes.extend_from_slice(&name_len.to_le_bytes());
bytes.extend_from_slice(name_bytes);
bytes.extend_from_slice(&inode_number.to_le_bytes());
bytes.push(*entry_type);
}
let hash = hash_section(&bytes);
DirNode {
entries: entries.to_vec(),
bytes,
hash,
}
}
fn pack_slabs(drops: &[PendingDrop]) -> Vec<SlabArtifact> {
let local_drops: Vec<&PendingDrop> = drops
.iter()
.filter(|d| d.codec != limnifs_core::codec::CODEC_REFERENCED)
.collect();
if local_drops.is_empty() {
return Vec::new();
}
let max_content = MAX_SLAB_TOTAL_BYTES.saturating_sub(SLAB_HEADER_LEN);
let mut slab_groups: Vec<Vec<&PendingDrop>> = Vec::new();
let mut current: Vec<&PendingDrop> = Vec::new();
let mut current_size: usize = 0;
for drop in &local_drops {
let footprint = drop.slab_footprint();
if !current.is_empty() && current_size + footprint > max_content {
slab_groups.push(std::mem::take(&mut current));
current_size = 0;
}
current.push(*drop);
current_size += footprint;
}
if !current.is_empty() {
slab_groups.push(current);
}
use rayon::prelude::*;
slab_groups
.par_iter()
.enumerate()
.map(|(ordinal, group)| {
let ordinal_u64 = u64::try_from(ordinal).expect("slab count fits u64");
encode_slab(ordinal_u64, group)
})
.collect()
}
fn encode_slab(ordinal: u64, drops: &[&PendingDrop]) -> SlabArtifact {
const DROP_RECORD_LEN: usize = 50;
let mut drop_records = Vec::with_capacity(drops.len() * DROP_RECORD_LEN);
let mut solid_window = Vec::new();
let mut drop_ids = Vec::with_capacity(drops.len());
let mut offset_in_window: u32 = 0;
for drop in drops {
let plaintext_len = drop.plaintext_len_value();
let window_len = drop.len_in_window();
drop_records.extend_from_slice(&drop.id);
drop_records.extend_from_slice(&plaintext_len.to_le_bytes());
drop_records.extend_from_slice(&[drop.codec, 0x00, 0x00]);
drop_records.push(0x00); drop_records.extend_from_slice(&offset_in_window.to_le_bytes());
drop_records.extend_from_slice(&window_len.to_le_bytes());
drop_records.push(drop.dict_id); drop_records.push(drop.flags); solid_window.extend_from_slice(&drop.compressed);
drop_ids.push(drop.id);
offset_in_window = offset_in_window
.checked_add(window_len)
.expect("slab window size fits u32");
}
let slab_content = [&drop_records[..], &solid_window[..]].concat();
let slab_hash = hash_section(&slab_content);
let slab_id = SlabId::new(ordinal, slab_hash);
let total_length = SLAB_HEADER_LEN + slab_content.len();
let mut slab_bytes = Vec::with_capacity(total_length);
slab_bytes.extend_from_slice(b"LIM1");
slab_bytes.extend_from_slice(&1u16.to_le_bytes()); slab_bytes.extend_from_slice(&slab_id.to_bytes());
slab_bytes.extend_from_slice(
&u64::try_from(total_length)
.unwrap_or(u64::MAX)
.to_le_bytes(),
);
slab_bytes.push(0x00);
slab_bytes.push(0x00);
slab_bytes.extend_from_slice(&slab_content);
let mut h8 = String::with_capacity(8);
for b in &slab_id.hash[..4] {
h8.push_str(&format!("{b:02x}"));
}
let locator = format!("file:slab-{ordinal}-{h8}.bin");
SlabArtifact {
id: slab_id,
bytes: slab_bytes,
locator,
drop_ids,
}
}
#[cfg(test)]
fn pseudo_random_bytes(seed: u64, count: usize) -> Vec<u8> {
let mut state = seed;
let mut out = Vec::with_capacity(count);
for _ in 0..count {
state = state
.wrapping_mul(6_364_136_223_846_793_005)
.wrapping_add(1_442_695_040_888_963_407);
out.push(u8::try_from(state >> 56).expect("fits u8"));
}
out
}
#[cfg(test)]
mod tests {
use super::*;
use limnifs_core::ManifestCursor;
#[test]
fn write_stream_packs_single_named_stream() {
let temp = std::env::temp_dir().join(format!(
"limnifs-write-stream-test-{}-{}",
std::process::id(),
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_nanos()
));
std::fs::create_dir_all(&temp).expect("create temp dir");
let content = b"stream test content line\n".repeat(10_000); let cursor = std::io::Cursor::new(content.clone());
let config = WriteConfig::default_v0_1();
let artifact = write_stream("streamed.txt", cursor, &config).expect("write_stream");
assert!(!artifact.bytes.is_empty(), "manifest bytes non-empty");
assert!(!artifact.slabs.is_empty(), "at least one slab produced");
let total_drop_bytes: usize = artifact.slabs.iter().map(|s| s.bytes.len()).sum();
assert!(total_drop_bytes > 0, "drops non-empty");
let _ = std::fs::remove_dir_all(&temp);
}
#[test]
fn write_layer_references_base_drops() {
let temp = std::env::temp_dir().join(format!(
"limnifs-write-layer-test-{}-{}",
std::process::id(),
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_nanos()
));
std::fs::create_dir_all(&temp).expect("create temp dir");
let base_dir = temp.join("base");
std::fs::create_dir_all(&base_dir).expect("base dir");
let text = b"layer test content line\n".repeat(50_000); std::fs::write(base_dir.join("shared.txt"), &text).expect("write shared");
std::fs::write(base_dir.join("base-only.txt"), b"only in base").expect("write base-only");
let config = WriteConfig::default_v0_1();
let base_artifact = write_directory_with_config(&base_dir, &config).expect("base");
let base_manifest = temp.join("base.lim");
std::fs::write(&base_manifest, &base_artifact.bytes).expect("write base manifest");
for slab in &base_artifact.slabs {
let slab_name = sidecar_name(&slab.locator).expect("slab locator");
std::fs::write(temp.join(slab_name), &slab.bytes).expect("write base slab");
}
let layer_dir = temp.join("layer");
std::fs::create_dir_all(&layer_dir).expect("layer dir");
std::fs::write(layer_dir.join("shared.txt"), &text).expect("write shared in layer");
std::fs::write(layer_dir.join("new.txt"), b"fresh content in layer").expect("write new");
let layer_artifact = write_layer(&base_manifest, &layer_dir, &config).expect("layer");
let layer_slab_bytes: usize = layer_artifact.slabs.iter().map(|s| s.bytes.len()).sum();
let base_slab_bytes: usize = base_artifact.slabs.iter().map(|s| s.bytes.len()).sum();
assert!(
layer_slab_bytes < base_slab_bytes / 4,
"layer slabs ({}) should be much smaller than base ({}) — layering failed",
layer_slab_bytes,
base_slab_bytes
);
let base_root = base_artifact.merkle_root.as_bytes();
assert!(
layer_artifact
.bytes
.windows(32)
.any(|w| w == base_root.as_slice()),
"layer manifest must contain base's ManifestRoot bytes"
);
let _ = std::fs::remove_dir_all(&temp);
}
#[test]
fn tournament_short_circuits_on_highly_compressible_chunk() {
let chunk = b"hello world ".repeat(500);
let tunables = limnifs_core::codec::CodecTunables::default();
let tournament = TournamentSpec {
codec_ids: vec![
limnifs_core::codec::CODEC_LZ4,
limnifs_core::codec::CODEC_BROTLI,
],
min_size: 16,
skip_for_binary: false,
short_circuit_permille: 250,
};
let (codec_id, compressed) = compress_chunk_with_tournament(
&chunk,
classifier::Class::Text,
limnifs_core::codec::CODEC_BROTLI,
limnifs_core::codec::CODEC_LZ4,
&tunables,
&tournament,
);
assert_eq!(codec_id, limnifs_core::codec::CODEC_LZ4);
assert!(compressed.len() < chunk.len());
}
#[test]
fn tournament_runs_all_codecs_when_short_circuit_disabled() {
let chunk = b"hello world ".repeat(500);
let tunables = limnifs_core::codec::CodecTunables::default();
let tournament = TournamentSpec {
codec_ids: vec![
limnifs_core::codec::CODEC_LZ4,
limnifs_core::codec::CODEC_BROTLI,
limnifs_core::codec::CODEC_ZSTD,
],
min_size: 16,
skip_for_binary: false,
short_circuit_permille: 0,
};
let (codec_id, compressed) = compress_chunk_with_tournament(
&chunk,
classifier::Class::Text,
limnifs_core::codec::CODEC_BROTLI,
limnifs_core::codec::CODEC_LZ4,
&tunables,
&tournament,
);
assert!(
codec_id == limnifs_core::codec::CODEC_ZSTD
|| codec_id == limnifs_core::codec::CODEC_BROTLI,
"expected ZSTD or Brotli to win, got codec {codec_id}"
);
assert!(compressed.len() < chunk.len());
}
#[test]
fn tournament_skips_for_binary_when_configured() {
let chunk = vec![0u8; 4096];
let tunables = limnifs_core::codec::CodecTunables::default();
let tournament = TournamentSpec {
codec_ids: vec![limnifs_core::codec::CODEC_BROTLI],
min_size: 16,
skip_for_binary: true,
short_circuit_permille: 250,
};
let (codec_id, _compressed) = compress_chunk_with_tournament(
&chunk,
classifier::Class::Binary,
limnifs_core::codec::CODEC_BROTLI,
limnifs_core::codec::CODEC_LZ4,
&tunables,
&tournament,
);
assert_eq!(codec_id, limnifs_core::codec::CODEC_LZ4);
}
#[test]
fn tournament_small_chunk_uses_preferred_codec() {
let chunk = b"tiny";
let tunables = limnifs_core::codec::CodecTunables::default();
let tournament = TournamentSpec {
codec_ids: vec![limnifs_core::codec::CODEC_BROTLI],
min_size: 1024,
skip_for_binary: false,
short_circuit_permille: 0,
};
let (codec_id, _compressed) = compress_chunk_with_tournament(
chunk,
classifier::Class::Text,
limnifs_core::codec::CODEC_BROTLI,
limnifs_core::codec::CODEC_LZ4,
&tunables,
&tournament,
);
assert_eq!(codec_id, limnifs_core::codec::CODEC_STORE);
}
#[test]
fn tournament_falls_back_to_store_when_no_codec_compresses() {
let chunk = pseudo_random_bytes(42, 4096);
let tunables = limnifs_core::codec::CodecTunables::default();
let tournament = TournamentSpec {
codec_ids: vec![limnifs_core::codec::CODEC_LZ4],
min_size: 16,
skip_for_binary: false,
short_circuit_permille: 0,
};
let (codec_id, compressed) = compress_chunk_with_tournament(
&chunk,
classifier::Class::Binary,
limnifs_core::codec::CODEC_BROTLI,
limnifs_core::codec::CODEC_LZ4,
&tunables,
&tournament,
);
assert_eq!(codec_id, limnifs_core::codec::CODEC_STORE);
assert_eq!(compressed.len(), chunk.len());
}
#[test]
fn dictionaries_enabled_emits_dictionary_section_when_enough_samples() {
let temp = std::env::temp_dir().join(format!(
"limnifs-write-test-{}-dict-{}",
std::process::id(),
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_nanos() as u64)
.unwrap_or(0),
));
let _ = std::fs::remove_dir_all(&temp);
std::fs::create_dir_all(&temp).expect("mkdir");
for i in 0..200 {
let content = format!(
"function test_case_{i}() {{ return constant + {i}; }}\n\
// shared comment line {i}\n\
struct Foo {{ x: i32 }} // type {i}\n"
)
.repeat(5);
let path = temp.join(format!("file_{i:04}.txt"));
std::fs::write(&path, content.as_bytes()).expect("write");
}
let mut config = crate::profile::balanced();
config.defaults.text_codec = "zstd".into();
config.defaults.metadata_codec = "zstd".into();
config.dictionaries.enabled = true;
config.dictionaries.min_class_size = 50;
config.dictionaries.max_dict_size = 8192;
let artifact = write_directory_with_config(&temp, &config).expect("write");
std::fs::remove_dir_all(&temp).ok();
let mut cursor = ManifestCursor::new(&artifact.bytes);
let _ = limnifs_core::parse_manifest_header(&mut cursor).expect("header");
let _ = limnifs_core::parse_feature_flags_section(&mut cursor).expect("flags");
let _ = limnifs_core::parse_metadata_reference(&mut cursor).expect("meta_ref");
let _ = limnifs_core::parse_slab_index(&mut cursor).expect("slab_index");
let _ = limnifs_core::parse_history(&mut cursor).expect("history");
let _remaining = cursor.remaining_len();
}
#[test]
fn write_empty_directory() {
let temp =
std::env::temp_dir().join(format!("limnifs-write-test-{}-empty", std::process::id()));
std::fs::create_dir_all(&temp).expect("create temp dir");
let artifact = write_directory(&temp).expect("write succeeds");
std::fs::remove_dir_all(&temp).ok();
assert!(artifact.inode_count >= 1);
assert_eq!(artifact.file_count, 0);
assert_eq!(artifact.dir_count, 1);
assert!(artifact.slabs.is_empty());
}
#[test]
fn write_small_file_inline() {
let temp =
std::env::temp_dir().join(format!("limnifs-write-test-{}-small", std::process::id()));
std::fs::create_dir_all(&temp).expect("create temp dir");
std::fs::write(temp.join("hello.txt"), b"hello world").expect("write file");
let artifact = write_directory(&temp).expect("write succeeds");
std::fs::remove_dir_all(&temp).ok();
assert_eq!(artifact.file_count, 1);
assert!(artifact.slabs.is_empty());
assert_eq!(artifact.drop_count, 0);
}
#[test]
fn write_large_file_uses_slab() {
let temp =
std::env::temp_dir().join(format!("limnifs-write-test-{}-large", std::process::id()));
std::fs::create_dir_all(&temp).expect("create temp dir");
let large_data = vec![0xABu8; INLINE_THRESHOLD + 100];
std::fs::write(temp.join("big.bin"), &large_data).expect("write big");
let artifact = write_directory(&temp).expect("write succeeds");
std::fs::remove_dir_all(&temp).ok();
assert_eq!(artifact.drop_count, 1);
assert_eq!(artifact.slabs.len(), 1);
}
#[test]
fn write_mixed_inline_and_large() {
let temp =
std::env::temp_dir().join(format!("limnifs-write-test-{}-mix", std::process::id()));
std::fs::create_dir_all(&temp).expect("create temp dir");
std::fs::write(temp.join("small.txt"), b"tiny").expect("write small");
std::fs::write(temp.join("large.bin"), vec![0xCDu8; INLINE_THRESHOLD * 2])
.expect("write large");
let artifact = write_directory(&temp).expect("write succeeds");
std::fs::remove_dir_all(&temp).ok();
assert_eq!(artifact.file_count, 2);
assert_eq!(artifact.drop_count, 1);
assert_eq!(artifact.slabs.len(), 1);
}
#[test]
fn deduplicates_identical_large_files() {
let temp =
std::env::temp_dir().join(format!("limnifs-write-test-{}-dedup", std::process::id()));
std::fs::create_dir_all(&temp).expect("create temp dir");
let data = vec![0x77u8; INLINE_THRESHOLD + 10];
std::fs::write(temp.join("a.bin"), &data).expect("write a");
std::fs::write(temp.join("b.bin"), &data).expect("write b");
let artifact = write_directory(&temp).expect("write succeeds");
std::fs::remove_dir_all(&temp).ok();
assert_eq!(artifact.drop_count, 1);
}
#[test]
fn write_and_verify_roundtrip() {
let temp = std::env::temp_dir().join(format!(
"limnifs-write-test-{}-roundtrip",
std::process::id()
));
std::fs::create_dir_all(&temp).expect("create temp dir");
std::fs::write(temp.join("a.txt"), b"aaa").expect("write a");
std::fs::write(temp.join("b.txt"), b"bbb").expect("write b");
std::fs::create_dir_all(temp.join("sub")).expect("create sub");
std::fs::write(temp.join("sub").join("c.txt"), b"ccc").expect("write c");
let artifact = write_directory(&temp).expect("write succeeds");
std::fs::remove_dir_all(&temp).ok();
assert_eq!(artifact.file_count, 3);
assert_eq!(artifact.dir_count, 2);
let mut cursor = ManifestCursor::new(&artifact.bytes);
limnifs_core::parse_manifest_header(&mut cursor).expect("header");
limnifs_core::parse_feature_flags_section(&mut cursor).expect("flags");
let meta_ref = limnifs_core::parse_metadata_reference(&mut cursor).expect("meta ref");
assert!(meta_ref.is_inlined());
let slab_index = limnifs_core::parse_slab_index(&mut cursor).expect("slab index");
assert_eq!(slab_index.len(), 0);
limnifs_core::parse_history(&mut cursor).expect("history");
}
#[test]
fn write_deterministic() {
let temp =
std::env::temp_dir().join(format!("limnifs-write-test-{}-det", std::process::id()));
std::fs::create_dir_all(&temp).expect("create temp dir");
std::fs::write(temp.join("x.txt"), b"xxx").expect("write x");
let a1 = write_directory(&temp).expect("first write");
let a2 = write_directory(&temp).expect("second write");
std::fs::remove_dir_all(&temp).ok();
assert_eq!(a1.bytes, a2.bytes);
assert_eq!(a1.merkle_root, a2.merkle_root);
}
#[test]
fn slab_parses_correctly() {
let temp =
std::env::temp_dir().join(format!("limnifs-write-test-{}-slab", std::process::id()));
std::fs::create_dir_all(&temp).expect("create temp dir");
std::fs::write(temp.join("big.bin"), vec![0x11u8; INLINE_THRESHOLD + 1])
.expect("write big");
let artifact = write_directory(&temp).expect("write succeeds");
std::fs::remove_dir_all(&temp).ok();
let slab_bytes = &artifact.slabs[0].bytes;
let mut cursor = ManifestCursor::new(slab_bytes);
let slab_header = limnifs_core::parse_slab_header(&mut cursor).expect("slab header parses");
assert_eq!(
slab_header.format_version,
limnifs_core::slab::SLAB_FORMAT_VERSION
);
assert!(!slab_header.is_sealed());
assert!(!slab_header.has_erasure_coding());
let drop_record =
limnifs_core::parse_drop_record(&mut cursor, &slab_header).expect("drop record parses");
assert_eq!(drop_record.plaintext_len as usize, INLINE_THRESHOLD + 1);
}
#[test]
fn fastcdc_produces_multiple_chunks_for_large_files() {
let temp = std::env::temp_dir().join(format!(
"limnifs-write-test-{}-cdc-multi",
std::process::id()
));
std::fs::create_dir_all(&temp).expect("create temp dir");
let data = pseudo_random_bytes(42, 1024 * 1024);
std::fs::write(temp.join("big.bin"), &data).expect("write big");
let artifact = write_directory(&temp).expect("write succeeds");
std::fs::remove_dir_all(&temp).ok();
assert!(
artifact.drop_count > 1,
"expected FastCDC to produce multiple drops for 1 MiB input, got {}",
artifact.drop_count
);
}
#[test]
fn fastcdc_deduplicates_shared_substrings() {
let temp = std::env::temp_dir().join(format!(
"limnifs-write-test-{}-cdc-dedup",
std::process::id()
));
std::fs::create_dir_all(&temp).expect("create temp dir");
let shared = pseudo_random_bytes(7, 512 * 1024);
let mut a = Vec::with_capacity(shared.len() + 1024);
a.extend_from_slice(&pseudo_random_bytes(1, 1024));
a.extend_from_slice(&shared);
let mut b = Vec::with_capacity(shared.len() + 2048);
b.extend_from_slice(&pseudo_random_bytes(2, 2048));
b.extend_from_slice(&shared);
std::fs::write(temp.join("a.bin"), &a).expect("write a");
std::fs::write(temp.join("b.bin"), &b).expect("write b");
let temp_a = std::env::temp_dir().join(format!(
"limnifs-write-test-{}-cdc-dedup-a",
std::process::id()
));
std::fs::create_dir_all(&temp_a).expect("create temp_a");
std::fs::write(temp_a.join("a.bin"), &a).expect("write a");
let artifact_a = write_directory(&temp_a).expect("a writes");
std::fs::remove_dir_all(&temp_a).ok();
let temp_b = std::env::temp_dir().join(format!(
"limnifs-write-test-{}-cdc-dedup-b",
std::process::id()
));
std::fs::create_dir_all(&temp_b).expect("create temp_b");
std::fs::write(temp_b.join("b.bin"), &b).expect("write b");
let artifact_b = write_directory(&temp_b).expect("b writes");
std::fs::remove_dir_all(&temp_b).ok();
let artifact_both = write_directory(&temp).expect("both write");
std::fs::remove_dir_all(&temp).ok();
let sum_alone = artifact_a.drop_count + artifact_b.drop_count;
assert!(
artifact_both.drop_count < sum_alone,
"expected dedup win: both together = {} drops, sum alone = {} drops",
artifact_both.drop_count,
sum_alone
);
}
#[test]
fn slab_splits_when_content_exceeds_ceiling() {
let temp =
std::env::temp_dir().join(format!("limnifs-write-test-{}-split", std::process::id()));
std::fs::create_dir_all(&temp).expect("create temp dir");
for i in 0..7u32 {
let data = pseudo_random_bytes(u64::from(i), 10 * 1024 * 1024);
std::fs::write(temp.join(format!("big-{i}.bin")), &data).expect("write big");
}
let artifact = write_directory(&temp).expect("write succeeds");
std::fs::remove_dir_all(&temp).ok();
assert!(
artifact.slabs.len() >= 2,
"expected at least 2 slabs for 70 MiB of incompressible data, got {}",
artifact.slabs.len()
);
for slab in &artifact.slabs {
assert!(
slab.bytes.len() <= MAX_SLAB_TOTAL_BYTES,
"slab {} is {} bytes (> {} ceiling)",
slab.id.ordinal,
slab.bytes.len(),
MAX_SLAB_TOTAL_BYTES,
);
}
let total_drop_ids: usize = artifact.slabs.iter().map(|s| s.drop_ids.len()).sum();
assert_eq!(
total_drop_ids, artifact.drop_count,
"drop_ids count across slabs must match WriteArtifact.drop_count",
);
}
}