diff --git a/Cargo.lock b/Cargo.lock index ae0e36976484..6b7f53e0e816 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9964,6 +9964,7 @@ dependencies = [ "auto-hash-map", "bitfield", "byteorder", + "clap", "codspeed-criterion-compat", "crc32fast", "dashmap 7.0.0-rc2", diff --git a/turbopack/crates/turbo-persistence/Cargo.toml b/turbopack/crates/turbo-persistence/Cargo.toml index 8698edaf4c82..7fcd5dc7d616 100644 --- a/turbopack/crates/turbo-persistence/Cargo.toml +++ b/turbopack/crates/turbo-persistence/Cargo.toml @@ -17,6 +17,7 @@ anyhow = { workspace = true } auto-hash-map = { workspace = true } bitfield = { workspace = true } byteorder = { workspace = true } +clap = { workspace = true } crc32fast = { workspace = true } dashmap = { workspace = true} either = { workspace = true } @@ -52,6 +53,10 @@ turbo-tasks-malloc = { workspace = true, features = ["custom_allocator"] } name = "sst_inspect" path = "src/bin/sst_inspect.rs" +[[bin]] +name = "zstd_dictionary" +path = "src/bin/zstd_dictionary.rs" + [lints] workspace = true diff --git a/turbopack/crates/turbo-persistence/README.md b/turbopack/crates/turbo-persistence/README.md index 52c0f31305cc..05f7c9ce71fe 100644 --- a/turbopack/crates/turbo-persistence/README.md +++ b/turbopack/crates/turbo-persistence/README.md @@ -57,6 +57,7 @@ A meta file can contain metadata about multiple SST files. The metadata is store - 4 bytes magic number (0xFE4ADA4A) - 4 bytes key family - 1 byte compression algorithm, which must match the configuration used to open the database + - 4 bytes zstd dictionary ID (zero when no dictionary is configured) - 4 bytes count of obsolete SST files - foreach obsolete SST file - 4 bytes sequence number of the obsolete SST file @@ -362,6 +363,37 @@ Configuration options for compactions are: - max number of SST files that are merged at once - coverage when compaction is triggered (otherwise calling compact is a noop) +## Training and evaluating zstd dictionaries offline + +`zstd_dictionary` trains and compares zstd dictionaries from logical values in existing database +copies without modifying them or running the application that created them: + +```sh +cargo run -p turbo-persistence --release --bin zstd_dictionary -- train \ + --family --output candidate.zdict \ + path/to/database-a path/to/database-b + +cargo run -p turbo-persistence --release --bin zstd_dictionary -- evaluate \ + --family --dictionary candidate.zdict --json report.json \ + path/to/database-a path/to/database-b +``` + +Training produces a 64 KiB dictionary from up to approximately 64 MiB of samples. It takes one +hash-ordered logical value from each cache in turn, so one large cache cannot monopolize the sample. +The output path is overwritten directly. + +LZ4 and no-dictionary zstd level 3 baselines are always included during evaluation. Source SSTs may use +LZ4 or plain zstd without extra options. Pass `--source-dictionary ` when any input SST records +a nonzero dictionary ID; it is ignored for LZ4 and plain-zstd SSTs. The tool follows `CURRENT`, +deletion files, and meta-file supersession, and uses `StaticSortedFileIter` to read slice, medium, and +blob values. Checksums, dictionary IDs, and decompressed lengths are verified. + +Evaluation groups small logical values into SST-local 8–12 KiB units, while medium values and blobs +remain independent. The 12.5% minimum-savings rule is applied per approximated unit, so this remains +comparative rather than exact SST-size modeling. Estimated stored bytes exclude fixed container +headers. Timing fields are single-pass diagnostics; use byte/count fields for repeatable comparisons +of one copied cache snapshot. + ## Opening - Read the `CURRENT` file diff --git a/turbopack/crates/turbo-persistence/benches/mod.rs b/turbopack/crates/turbo-persistence/benches/mod.rs index 1f5e839e9bcc..6d671f6cd8da 100644 --- a/turbopack/crates/turbo-persistence/benches/mod.rs +++ b/turbopack/crates/turbo-persistence/benches/mod.rs @@ -622,7 +622,7 @@ fn prefill_multi_value_database( family_configs: [FamilyConfig { name: "test", kind: FamilyKind::MultiValue, - compression: Compression::Lz4, + compression: Compression::Lz4.into(), }], ..TpDbConfig::new() }; @@ -698,7 +698,7 @@ fn open_multi_value_db(path: &Path) -> TurboPersistence { family_configs: [FamilyConfig { name: "test", kind: FamilyKind::MultiValue, - compression: Compression::Lz4, + compression: Compression::Lz4.into(), }], ..TpDbConfig::new() }; @@ -968,7 +968,7 @@ fn bench_write_multi_value(c: &mut Criterion) { family_configs: [FamilyConfig { name: "test", kind: FamilyKind::MultiValue, - compression: Compression::Lz4, + compression: Compression::Lz4.into(), }], ..TpDbConfig::new() }; @@ -1211,7 +1211,7 @@ fn bench_static_sorted_file_lookup(c: &mut Criterion) { &entries, &sst_path, MetaEntryFlags::FRESH, - Compression::Lz4, + Compression::Lz4.into(), ) .unwrap(); @@ -1223,7 +1223,7 @@ fn bench_static_sorted_file_lookup(c: &mut Criterion) { let sst = StaticSortedFile::open( tempdir.path(), sst_meta, - Compression::Lz4, + Compression::Lz4.into(), turbo_persistence::AccessMode::Mmap, ) .unwrap(); diff --git a/turbopack/crates/turbo-persistence/src/arc_bytes.rs b/turbopack/crates/turbo-persistence/src/arc_bytes.rs index 09e189e9a307..bd7dbd1b916b 100644 --- a/turbopack/crates/turbo-persistence/src/arc_bytes.rs +++ b/turbopack/crates/turbo-persistence/src/arc_bytes.rs @@ -9,7 +9,7 @@ use std::{ use memmap2::Mmap; use crate::{ - Compression, + CompressionConfig, compression::decompress_into_arc, shared_bytes::{SharedBytes, is_subslice_of}, }; @@ -146,7 +146,7 @@ impl SharedBytes for ArcBytes { } fn from_decompressed( - compression: Compression, + compression: CompressionConfig, uncompressed_length: u32, block: &[u8], ) -> anyhow::Result { diff --git a/turbopack/crates/turbo-persistence/src/bin/sst_inspect.rs b/turbopack/crates/turbo-persistence/src/bin/sst_inspect.rs index 740f6b4757bc..1e48fd8f4d9b 100644 --- a/turbopack/crates/turbo-persistence/src/bin/sst_inspect.rs +++ b/turbopack/crates/turbo-persistence/src/bin/sst_inspect.rs @@ -16,15 +16,14 @@ use std::{ use anyhow::{Context, Result, bail}; use byteorder::{BE, ReadBytesExt}; -use fs_err::{self as fs, File}; +use clap::Parser; +use fs_err::File; use lzzzz::lz4::decompress; use memmap2::Mmap; use turbo_persistence::{ - BLOCK_HEADER_SIZE, Compression, MAX_INLINE_VALUE_SIZE, checksum_block, - meta_file::MetaFile, + BLOCK_HEADER_SIZE, Compression, CompressionConfig, MAX_INLINE_VALUE_SIZE, checksum_block, mmap_helper::advise_mmap_for_persistence, - read_current_version, - sst_filter::SstFilter, + offline::{SstInfo, collect_sst_info}, static_sorted_file::{ BLOCK_TYPE_FIXED_KEY_NO_HASH, BLOCK_TYPE_FIXED_KEY_WITH_HASH, BLOCK_TYPE_KEY_NO_HASH, BLOCK_TYPE_KEY_WITH_HASH, FIXED_KEY_BLOCK_MIXED_VALUE_TYPE, KEY_BLOCK_ENTRY_TYPE_BLOB, @@ -129,13 +128,6 @@ impl SstStats { } } -/// Information about an SST file from the meta file -struct SstInfo { - sequence_number: u32, - block_count: u16, - compression: Compression, -} - /// Accumulates statistics for a single entry of the given type. fn track_entry_type(stats: &mut SstStats, entry_type: u8) { *stats.entry_type_counts.entry(entry_type).or_insert(0) += 1; @@ -220,79 +212,6 @@ fn format_bytes(bytes: u64) -> String { } } -/// Collect SST info from all active meta files in the database directory, -/// mirroring the DB's own open logic: read CURRENT, filter by .del files, -/// and apply SstFilter to skip superseded entries. -fn collect_sst_info(db_path: &Path) -> Result>> { - // Read the CURRENT sequence number — only files with seq <= current are valid. - let current = read_current_version(db_path)? - .context("CURRENT file is missing")? - .max_sequence_number; - - // Read .del files to find sequences that were deleted but not yet cleaned up. - let mut deleted_seqs: HashSet = HashSet::new(); - for entry in fs::read_dir(db_path)? { - let path = entry?.path(); - if path.extension().and_then(|s| s.to_str()) == Some("del") { - let content = fs::read(&path)?; - let mut cursor: &[u8] = &content; - while !cursor.is_empty() { - deleted_seqs.insert(cursor.read_u32::()?); - } - } - } - - // Collect valid meta sequence numbers. - let mut meta_seqs: Vec = fs::read_dir(db_path)? - .filter_map(|e| e.ok()) - .filter_map(|e| { - let path = e.path(); - if path.extension().and_then(|s| s.to_str()) != Some("meta") { - return None; - } - let seq: u32 = path.file_stem()?.to_str()?.parse().ok()?; - if seq > current || deleted_seqs.contains(&seq) { - return None; - } - Some(seq) - }) - .collect(); - - if meta_seqs.is_empty() { - bail!("No active .meta files found in {}", db_path.display()); - } - - meta_seqs.sort_unstable(); - - let mut meta_files: Vec = meta_seqs - .iter() - .map(|&seq| { - MetaFile::open(db_path, seq, None, turbo_persistence::AccessMode::Mmap) - .with_context(|| format!("Failed to open {seq:08}.meta")) - }) - .collect::>()?; - - // Apply SstFilter (newest first) to drop entries superseded by a newer meta file. - let mut sst_filter = SstFilter::new(); - for meta in meta_files.iter_mut().rev() { - sst_filter.apply_filter(meta); - } - - let mut family_sst_info: BTreeMap> = BTreeMap::new(); - for meta in &meta_files { - let family = meta.family(); - for entry in meta.entries() { - family_sst_info.entry(family).or_default().push(SstInfo { - sequence_number: entry.sequence_number(), - block_count: entry.block_count(), - compression: meta.compression(), - }); - } - } - - Ok(family_sst_info) -} - /// Information about a raw block read from disk. struct RawBlock { data: Box<[u8]>, @@ -307,7 +226,7 @@ fn read_block( block_offsets_start: usize, block_index: u16, sequence_number: u32, - compression: Compression, + compression: CompressionConfig, ) -> Result { let offset = block_offsets_start + block_index as usize * size_of::(); @@ -348,11 +267,18 @@ fn read_block( let data = if was_compressed { let mut buffer = vec![0u8; uncompressed_length as usize]; let bytes_written = match compression { - Compression::Lz4 => { + CompressionConfig::Lz4 => { decompress(compressed_data, &mut buffer).context("LZ4 decompression failed")? } - Compression::Zstd3 => zstd::bulk::decompress_to_buffer(compressed_data, &mut buffer) - .context("zstd decompression failed")?, + CompressionConfig::Zstd3 => { + zstd::bulk::decompress_to_buffer(compressed_data, &mut buffer) + .context("zstd decompression failed")? + } + CompressionConfig::Zstd3WithDictionary(dictionary) => { + zstd::bulk::Decompressor::with_dictionary(dictionary)? + .decompress_to_buffer(compressed_data, &mut buffer) + .context("zstd dictionary decompression failed")? + } }; assert_eq!( bytes_written, uncompressed_length as usize, @@ -480,8 +406,11 @@ fn iter_key_block_entry_types( } /// Analyze an SST file and return entry type statistics -fn analyze_sst_file(db_path: &Path, info: &SstInfo) -> Result { - let compression = info.compression; +fn analyze_sst_file( + db_path: &Path, + info: &SstInfo, + compression: CompressionConfig, +) -> Result { let filename = format!("{:08}.sst", info.sequence_number); let path = db_path.join(&filename); @@ -864,75 +793,61 @@ fn print_family_summary(family: u32, sst_count: usize, stats: &SstStats) { println!(); } +fn entry_type_help() -> String { + format!( + "Entry types:\n {KEY_BLOCK_ENTRY_TYPE_SMALL}: Small value (stored in separate value \ + block)\n {KEY_BLOCK_ENTRY_TYPE_BLOB}: Blob reference\n \ + {KEY_BLOCK_ENTRY_TYPE_KEY_DELETED}: Key tombstone (deletes all values for the key)\n \ + {KEY_BLOCK_ENTRY_TYPE_MEDIUM}: Medium value\n {KEY_BLOCK_ENTRY_TYPE_INLINE_MIN}-{}: \ + Inline value (size = type - {KEY_BLOCK_ENTRY_TYPE_INLINE_MIN})\n \ + {KEY_BLOCK_ENTRY_TYPE_KEY_VALUE_DELETED_MIN}-{}: Key-value tombstone (deleted value size \ + = type - {KEY_BLOCK_ENTRY_TYPE_KEY_VALUE_DELETED_MIN})\n\nFor TaskCache (family 3), \ + values are 4-byte TaskIds. Expected entry type is {} ({KEY_BLOCK_ENTRY_TYPE_INLINE_MIN} \ + + 4) for inline optimization.", + KEY_BLOCK_ENTRY_TYPE_INLINE_MIN + MAX_INLINE_VALUE_SIZE as u8, + KEY_BLOCK_ENTRY_TYPE_KEY_VALUE_DELETED_MIN + MAX_INLINE_VALUE_SIZE as u8, + KEY_BLOCK_ENTRY_TYPE_INLINE_MIN + 4, + ) +} + +#[derive(Parser)] +#[command(about = "Inspect turbo-persistence SST files", after_long_help = entry_type_help())] +struct Cli { + /// Show per-SST file details (default: family totals only). + #[arg(short, long)] + verbose: bool, + /// Dictionary used by zstd input SSTs. May be supplied multiple times; IDs are read from the + /// dictionaries. + #[arg(long)] + source_dictionary: Vec, + /// Database directory containing CURRENT, meta, SST, and blob files. + db_path: PathBuf, +} + fn main() -> Result<()> { - let args: Vec = std::env::args().collect(); - - // Parse arguments - let mut db_path: Option = None; - let mut verbose = false; - - let mut i = 1; - while i < args.len() { - match args[i].as_str() { - "--verbose" | "-v" => verbose = true, - arg if !arg.starts_with('-') => { - if db_path.is_none() { - db_path = Some(PathBuf::from(arg)); - } - } - _ => { - eprintln!("Unknown option: {}", args[i]); - std::process::exit(1); - } - } - i += 1; - } - - let db_path = match db_path { - Some(p) => p, - None => { - eprintln!("Usage: {} [OPTIONS] ", args[0]); - eprintln!(); - eprintln!("Inspects turbo-persistence SST files to report entry type statistics."); - eprintln!(); - eprintln!("Options:"); - eprintln!(" -v, --verbose Show per-SST file details (default: family totals only)"); - eprintln!(); - eprintln!("Entry types:"); - eprintln!( - " {KEY_BLOCK_ENTRY_TYPE_SMALL}: Small value (stored in separate value block)" - ); - eprintln!(" {KEY_BLOCK_ENTRY_TYPE_BLOB}: Blob reference"); - eprintln!( - " {KEY_BLOCK_ENTRY_TYPE_KEY_DELETED}: Key tombstone (deletes all values for the \ - key)" - ); - eprintln!(" {KEY_BLOCK_ENTRY_TYPE_MEDIUM}: Medium value"); - eprintln!( - " {KEY_BLOCK_ENTRY_TYPE_INLINE_MIN}-{}: Inline value (size = type - \ - {KEY_BLOCK_ENTRY_TYPE_INLINE_MIN})", - KEY_BLOCK_ENTRY_TYPE_INLINE_MIN + MAX_INLINE_VALUE_SIZE as u8 - ); - eprintln!( - " {KEY_BLOCK_ENTRY_TYPE_KEY_VALUE_DELETED_MIN}-{}: Key-value tombstone (deleted \ - value size = type - {KEY_BLOCK_ENTRY_TYPE_KEY_VALUE_DELETED_MIN})", - KEY_BLOCK_ENTRY_TYPE_KEY_VALUE_DELETED_MIN + MAX_INLINE_VALUE_SIZE as u8 - ); - eprintln!(); - eprintln!("For TaskCache (family 3), values are 4-byte TaskIds."); - eprintln!( - "Expected entry type is {} ({KEY_BLOCK_ENTRY_TYPE_INLINE_MIN} + 4) for inline \ - optimization.", - KEY_BLOCK_ENTRY_TYPE_INLINE_MIN + 4 - ); - std::process::exit(1); - } - }; + let Cli { + verbose, + source_dictionary, + db_path, + } = Cli::parse(); if !db_path.is_dir() { bail!("Not a directory: {}", db_path.display()); } + let mut source_dictionaries = BTreeMap::new(); + for path in source_dictionary { + let bytes = + fs_err::read(&path).with_context(|| format!("Failed to read {}", path.display()))?; + let dictionary = Box::leak(bytes.into_boxed_slice()) as &'static [u8]; + let id = CompressionConfig::Zstd3WithDictionary(dictionary) + .dictionary_id() + .with_context(|| format!("Dictionary {} has no zstd dictionary ID", path.display()))?; + if source_dictionaries.insert(id, dictionary).is_some() { + bail!("Duplicate source dictionary ID {id}"); + } + } + // Collect SST info grouped by family let family_sst_info = collect_sst_info(&db_path)?; @@ -949,7 +864,28 @@ fn main() -> Result<()> { let mut sst_stats_list: Vec<(u32, SstStats)> = Vec::new(); for info in sst_list { - match analyze_sst_file(&db_path, info) { + let compression = match (info.compression, info.dictionary_id) { + (Compression::Lz4, 0) => CompressionConfig::Lz4, + (Compression::Zstd3, 0) => CompressionConfig::Zstd3, + (Compression::Zstd3, id) => match source_dictionaries.get(&id) { + Some(dictionary) => CompressionConfig::Zstd3WithDictionary(dictionary), + None => { + eprintln!( + "Warning: Missing source dictionary ID {id} for {:08}.sst", + info.sequence_number + ); + continue; + } + }, + (Compression::Lz4, id) => { + eprintln!( + "Warning: LZ4 SST {:08}.sst has unexpected dictionary ID {id}", + info.sequence_number + ); + continue; + } + }; + match analyze_sst_file(&db_path, info, compression) { Ok(stats) => { family_stats.merge(&stats); if verbose { @@ -980,3 +916,38 @@ fn main() -> Result<()> { Ok(()) } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn clap_help_keeps_entry_type_reference() { + let help = Cli::try_parse_from(["sst_inspect", "--help"]) + .err() + .expect("--help should exit through clap") + .to_string(); + assert!(help.contains("Entry types:")); + assert!(help.contains("For TaskCache (family 3)")); + assert!(help.contains(&format!( + "{} ({} + 4)", + KEY_BLOCK_ENTRY_TYPE_INLINE_MIN + 4, + KEY_BLOCK_ENTRY_TYPE_INLINE_MIN + ))); + } + + #[test] + fn clap_accepts_multiple_source_dictionaries() { + let cli = Cli::try_parse_from([ + "sst_inspect", + "--source-dictionary", + "first.zdict", + "--source-dictionary", + "second.zdict", + "database", + ]) + .unwrap(); + assert_eq!(cli.source_dictionary.len(), 2); + assert_eq!(cli.db_path, PathBuf::from("database")); + } +} diff --git a/turbopack/crates/turbo-persistence/src/bin/zstd_dictionary.rs b/turbopack/crates/turbo-persistence/src/bin/zstd_dictionary.rs new file mode 100644 index 000000000000..ed56f0c1feb1 --- /dev/null +++ b/turbopack/crates/turbo-persistence/src/bin/zstd_dictionary.rs @@ -0,0 +1,1088 @@ +//! Train and evaluate zstd dictionaries from logical values in persistence caches. + +use std::{ + collections::{BTreeSet, HashSet, VecDeque}, + fs::{self, File}, + io::{BufWriter, Write}, + path::{Path, PathBuf}, + sync::Arc, + time::{Duration, Instant}, +}; + +use anyhow::{Context, Result, ensure}; +use clap::{Args, Parser, Subcommand}; +use lzzzz::lz4; +use serde::Serialize; +use turbo_persistence::{ + AccessMode, Compression, CompressionConfig, IterValue, MAX_INLINE_VALUE_SIZE, + MIN_SMALL_VALUE_BLOCK_SIZE, StaticSortedFileIter, StaticSortedFileMetaData, + offline::{SstInfo, collect_sst_info, decode_medium, read_blob}, +}; + +const SCHEMA_VERSION: u32 = 5; +const DICTIONARY_SIZE: usize = 64 * 1024; +const SAMPLE_BUDGET_MULTIPLIER: usize = 1000; +const SAMPLE_BYTE_BUDGET: usize = DICTIONARY_SIZE * SAMPLE_BUDGET_MULTIPLIER; + +#[derive(Parser)] +#[command(about = "Train and evaluate zstd dictionaries from persistence caches")] +struct Cli { + #[command(subcommand)] + command: Command, +} + +#[derive(Subcommand)] +enum Command { + /// Train a 64 KiB zstd dictionary from logical values. + Train { + #[command(flatten)] + source: Source, + /// Dictionary output path. Existing files are replaced. + #[arg(short, long)] + output: PathBuf, + /// Optional JSON training report. + #[arg(long)] + json: Option, + }, + /// Compare dictionaries with LZ4 and zstd level 3 baselines. + Evaluate { + #[command(flatten)] + source: Source, + /// Candidate dictionary. May be supplied more than once. + #[arg(short, long)] + dictionary: Vec, + /// Optional JSON evaluation report. + #[arg(long)] + json: Option, + }, +} + +#[derive(Args)] +struct Source { + /// Persistence family ID to inspect. + #[arg(long)] + family: u32, + /// Dictionary used to compress the input caches. + #[arg(long)] + source_dictionary: Option, + /// Database directories containing CURRENT, meta, SST, and blob files. + #[arg(required = true)] + caches: Vec, +} + +#[derive(Clone, Serialize)] +struct DictionaryInfo { + name: String, + path: Option, + bytes: usize, +} + +struct Candidate { + info: DictionaryInfo, + codec: CandidateCodec, + setup_duration: Duration, +} + +enum CandidateCodec { + Lz4, + Zstd { + compressor: zstd::bulk::Compressor<'static>, + decompressor: zstd::bulk::Decompressor<'static>, + }, +} + +struct Sample { + data: Arc<[u8]>, + small_value_sst: Option, +} + +#[derive(Default, Clone, Serialize)] +struct Metric { + count: u64, + bytes: u64, +} + +impl Metric { + fn add(&mut self, bytes: usize) { + self.count += 1; + self.bytes += bytes as u64; + } + + fn merge(&mut self, other: &Self) { + self.count += other.count; + self.bytes += other.bytes; + } +} + +#[derive(Default, Clone, Serialize)] +struct CompressionMetric { + input_bytes: u64, + raw_compressed_bytes: u64, + estimated_stored_bytes: u64, + raw_compression_ratio: Option, + encode_duration: Duration, + decode_duration: Duration, +} + +impl CompressionMetric { + fn merge(&mut self, other: &Self) { + self.input_bytes += other.input_bytes; + self.raw_compressed_bytes += other.raw_compressed_bytes; + self.estimated_stored_bytes += other.estimated_stored_bytes; + self.encode_duration += other.encode_duration; + self.decode_duration += other.decode_duration; + } + + fn finalize(&mut self) { + self.raw_compression_ratio = (self.input_bytes > 0) + .then(|| self.raw_compressed_bytes as f64 / self.input_bytes as f64); + } +} + +#[derive(Clone, Serialize)] +struct CandidateResult { + dictionary: DictionaryInfo, + combined: CompressionMetric, + setup_duration: Duration, +} + +#[derive(Serialize)] +struct CacheReport { + path: PathBuf, + family: u32, + active_ssts: u64, + source_codecs: BTreeSet, + samples: Metric, + duplicate_blob_references: u64, + candidates: Vec, +} + +#[derive(Serialize)] +struct EvaluationReport { + schema_version: u32, + family: u32, + timing_note: &'static str, + threshold_note: &'static str, + caches: Vec, + combined_samples: Metric, + combined_candidates: Vec, +} + +#[derive(Serialize)] +struct TrainingCacheReport { + path: PathBuf, + source_codecs: BTreeSet, + samples: Metric, +} + +#[derive(Serialize)] +struct TrainingReport { + schema_version: u32, + family: u32, + zstd_version: &'static str, + dictionary_size: usize, + sample_byte_target: usize, + selected: Metric, + selected_bytes: u64, + inputs_exhausted: bool, + caches: Vec, + dictionary: DictionaryInfo, +} + +struct CacheSampleIter { + path: PathBuf, + pending: VecDeque<(SstInfo, CompressionConfig)>, + current: Option<(StaticSortedFileIter, CompressionConfig, u32)>, + source_codecs: BTreeSet, + seen_blobs: HashSet, + active_ssts: u64, + duplicate_blob_references: u64, +} + +fn source_compression_for_sst( + path: &Path, + family: u32, + sst: &SstInfo, + source_dictionary: Option<&'static [u8]>, +) -> Result { + match (sst.compression, sst.dictionary_id) { + (Compression::Lz4, 0) => Ok(CompressionConfig::Lz4), + (Compression::Lz4, id) => anyhow::bail!( + "Cache {} family {family} SST {:08}.sst records LZ4 with unexpected dictionary ID {id}", + path.display(), + sst.sequence_number + ), + (Compression::Zstd3, 0) => Ok(CompressionConfig::Zstd3), + (Compression::Zstd3, id) => { + let dictionary = source_dictionary.with_context(|| { + format!( + "Cache {} family {family} SST {:08}.sst requires source dictionary ID {id}", + path.display(), + sst.sequence_number + ) + })?; + let configured = CompressionConfig::Zstd3WithDictionary(dictionary); + ensure!( + configured.dictionary_id() == Some(id), + "Cache {} family {family} SST {:08}.sst requires source dictionary ID {id}, but \ + supplied dictionary has ID {:?}", + path.display(), + sst.sequence_number, + configured.dictionary_id() + ); + Ok(configured) + } + } +} + +impl CacheSampleIter { + fn open(path: PathBuf, family: u32, source_dictionary: Option<&'static [u8]>) -> Result { + let mut families = collect_sst_info(&path) + .with_context(|| format!("Failed to inspect cache {}", path.display()))?; + let mut ssts = families.remove(&family).with_context(|| { + format!( + "Cache {} has no active SSTs for family {family}", + path.display() + ) + })?; + // A stable SST order makes repeated runs against one unchanged cache snapshot comparable. + ssts.sort_by_key(|sst| sst.sequence_number); + let mut source_codecs = BTreeSet::new(); + let pending = ssts + .into_iter() + .map(|sst| { + let compression = + source_compression_for_sst(&path, family, &sst, source_dictionary)?; + source_codecs.insert(format!("{compression:?}")); + Ok((sst, compression)) + }) + .collect::>>()?; + Ok(Self { + path, + active_ssts: pending.len() as u64, + pending, + current: None, + source_codecs, + seen_blobs: HashSet::new(), + duplicate_blob_references: 0, + }) + } + + fn open_next_sst(&mut self) -> Result { + let Some((sst, compression)) = self.pending.pop_front() else { + return Ok(false); + }; + let iter = StaticSortedFileIter::open( + &self.path, + StaticSortedFileMetaData { + sequence_number: sst.sequence_number, + block_count: sst.block_count, + }, + compression, + AccessMode::Mmap, + ) + .with_context(|| format!("Failed to open {:08}.sst", sst.sequence_number))?; + self.current = Some((iter, compression, sst.sequence_number)); + Ok(true) + } + + fn next_sample(&mut self) -> Result> { + loop { + if self.current.is_none() && !self.open_next_sst()? { + return Ok(None); + } + let (iter, compression, sequence_number) = self.current.as_mut().unwrap(); + let compression = *compression; + let sequence_number = *sequence_number; + let entry = match iter.next() { + Some(entry) => entry?, + None => { + self.current = None; + continue; + } + }; + if let Some(sample) = + self.sample_from_value(entry.value, compression, sequence_number)? + { + return Ok(Some(sample)); + } + } + } + + fn sample_from_value( + &mut self, + value: IterValue, + compression: CompressionConfig, + sequence_number: u32, + ) -> Result> { + match value { + IterValue::Slice { value } if value.len() > MAX_INLINE_VALUE_SIZE => Ok(Some(Sample { + data: Arc::from(value.as_ref()), + small_value_sst: Some(sequence_number), + })), + IterValue::Medium { + uncompressed_size, + checksum, + block, + } => { + let value = decode_medium(compression, uncompressed_size, checksum, &block) + .with_context(|| { + format!("Failed to read medium value in {}", self.path.display()) + })?; + Ok(Some(Sample { + data: value, + small_value_sst: None, + })) + } + IterValue::Blob { sequence_number } => { + if !self.seen_blobs.insert(sequence_number) { + self.duplicate_blob_references += 1; + return Ok(None); + } + let value = read_blob(&self.path, sequence_number, compression)?; + Ok(Some(Sample { + data: value, + small_value_sst: None, + })) + } + // Inline values live in key blocks and are not independently compressed. + IterValue::KeyDeleted | IterValue::KeyValueDeleted { .. } | IterValue::Slice { .. } => { + Ok(None) + } + } + } +} + +/// Groups logical small values into production-like SST-local compression units for evaluation. +struct EvaluationSampleIter { + source: CacheSampleIter, + small_block: Vec, + small_block_sst: Option, +} + +impl EvaluationSampleIter { + fn new(source: CacheSampleIter) -> Self { + Self { + source, + small_block: Vec::with_capacity(MIN_SMALL_VALUE_BLOCK_SIZE), + small_block_sst: None, + } + } + + fn next_sample(&mut self) -> Result> { + loop { + match self.source.next_sample()? { + Some(sample) => match sample.small_value_sst { + None => return Ok(Some(sample)), + Some(sequence_number) => { + if self.small_block_sst != Some(sequence_number) + && !self.small_block.is_empty() + { + let completed = std::mem::take(&mut self.small_block); + self.small_block_sst = Some(sequence_number); + self.small_block.extend_from_slice(&sample.data); + return Ok(Some(Sample { + data: completed.into(), + small_value_sst: None, + })); + } + self.small_block_sst = Some(sequence_number); + self.small_block.extend_from_slice(&sample.data); + if self.small_block.len() >= MIN_SMALL_VALUE_BLOCK_SIZE { + let completed = std::mem::take(&mut self.small_block); + return Ok(Some(Sample { + data: completed.into(), + small_value_sst: None, + })); + } + } + }, + None if self.small_block.is_empty() => return Ok(None), + None => { + let completed = std::mem::take(&mut self.small_block); + return Ok(Some(Sample { + data: completed.into(), + small_value_sst: None, + })); + } + } + } + } +} + +fn dictionary_info( + path: Option<&Path>, + dictionary: &[u8], + baseline: bool, +) -> Result { + let name = if baseline { + "zstd3 (no dictionary)".to_owned() + } else { + path.and_then(Path::file_name) + .context("Dictionary path has no filename")? + .to_str() + .context("Dictionary filename must be UTF-8")? + .to_owned() + }; + Ok(DictionaryInfo { + name, + path: path.map(Path::to_path_buf), + bytes: dictionary.len(), + }) +} + +fn make_candidates(paths: &[PathBuf]) -> Result> { + let mut result = Vec::with_capacity(paths.len() + 2); + result.push(Candidate { + info: DictionaryInfo { + name: "lz4".to_owned(), + path: None, + bytes: 0, + }, + codec: CandidateCodec::Lz4, + setup_duration: Duration::ZERO, + }); + let started = Instant::now(); + result.push(Candidate { + info: dictionary_info(None, &[], true)?, + codec: CandidateCodec::Zstd { + compressor: zstd::bulk::Compressor::new(3)?, + decompressor: zstd::bulk::Decompressor::new()?, + }, + setup_duration: started.elapsed(), + }); + let mut names = BTreeSet::new(); + for path in paths { + let dictionary = fs::read(path) + .with_context(|| format!("Failed to read dictionary {}", path.display()))?; + ensure!( + !dictionary.is_empty(), + "Dictionary {} is empty", + path.display() + ); + let info = dictionary_info(Some(path), &dictionary, false)?; + ensure!( + names.insert(info.name.clone()), + "Duplicate dictionary name {}", + info.name + ); + let started = Instant::now(); + result.push(Candidate { + info, + codec: CandidateCodec::Zstd { + compressor: zstd::bulk::Compressor::with_dictionary(3, &dictionary)?, + decompressor: zstd::bulk::Decompressor::with_dictionary(&dictionary)?, + }, + setup_duration: started.elapsed(), + }); + } + Ok(result) +} + +impl Candidate { + fn compress(&mut self, input: &[u8]) -> Result> { + match &mut self.codec { + CandidateCodec::Lz4 => { + let mut output = Vec::new(); + lz4::compress_to_vec(input, &mut output, lz4::ACC_LEVEL_DEFAULT)?; + Ok(output) + } + CandidateCodec::Zstd { compressor, .. } => Ok(compressor.compress(input)?), + } + } + + fn decompress(&mut self, input: &[u8], output_len: usize) -> Result> { + match &mut self.codec { + CandidateCodec::Lz4 => { + let mut output = vec![0; output_len]; + let written = lz4::decompress(input, &mut output)?; + ensure!(written == output_len, "LZ4 decompressed length mismatch"); + Ok(output) + } + CandidateCodec::Zstd { decompressor, .. } => { + Ok(decompressor.decompress(input, output_len)?) + } + } + } +} + +/// Evaluates all compression candidates against one approximated compression unit. +fn evaluate_sample( + sample: &Sample, + candidates: &mut [Candidate], + results: &mut [CandidateResult], +) -> Result<()> { + for (candidate, result) in candidates.iter_mut().zip(results) { + let started = Instant::now(); + let compressed = candidate + .compress(&sample.data) + .with_context(|| format!("Failed to compress with {}", candidate.info.name))?; + let encode_duration = started.elapsed(); + let started = Instant::now(); + let decompressed = candidate + .decompress(&compressed, sample.data.len()) + .with_context(|| format!("Failed to decompress with {}", candidate.info.name))?; + let decode_duration = started.elapsed(); + ensure!( + decompressed.as_slice() == sample.data.as_ref(), + "Round-trip mismatch with {}", + candidate.info.name + ); + + let metric = &mut result.combined; + metric.input_bytes += sample.data.len() as u64; + metric.raw_compressed_bytes += compressed.len() as u64; + metric.encode_duration += encode_duration; + metric.decode_duration += decode_duration; + metric.estimated_stored_bytes += + estimated_value_bytes(sample.data.len(), compressed.len()) as u64; + } + Ok(()) +} + +/// Applies the writer's 12.5% minimum-savings rule to an approximated compression unit. +/// +/// Small values are grouped before this call; medium values and blobs arrive independently. See +/// `write_block_to_file` for the production block-level rule. +fn estimated_value_bytes(original_len: usize, compressed_len: usize) -> usize { + if compressed_len < original_len - original_len / 8 { + compressed_len + } else { + original_len + } +} + +fn empty_results(candidates: &[Candidate]) -> Vec { + candidates + .iter() + .map(|candidate| CandidateResult { + dictionary: candidate.info.clone(), + combined: CompressionMetric::default(), + setup_duration: candidate.setup_duration, + }) + .collect() +} + +fn finalize_results(results: &mut [CandidateResult]) { + for result in results { + result.combined.finalize(); + } +} + +fn evaluate_cache( + path: &Path, + family: u32, + source_dictionary: Option<&'static [u8]>, + candidates: &mut [Candidate], +) -> Result { + let source = CacheSampleIter::open(path.to_path_buf(), family, source_dictionary)?; + let mut iter = EvaluationSampleIter::new(source); + let mut samples = Metric::default(); + let mut results = empty_results(candidates); + while let Some(sample) = iter.next_sample()? { + samples.add(sample.data.len()); + evaluate_sample(&sample, candidates, &mut results)?; + } + finalize_results(&mut results); + Ok(CacheReport { + path: path.to_path_buf(), + family, + active_ssts: iter.source.active_ssts, + source_codecs: iter.source.source_codecs, + samples, + duplicate_blob_references: iter.source.duplicate_blob_references, + candidates: results, + }) +} + +fn combine_evaluation( + family: u32, + caches: Vec, + candidates: &[Candidate], +) -> EvaluationReport { + let mut combined_samples = Metric::default(); + let mut combined_candidates = empty_results(candidates); + for cache in &caches { + combined_samples.merge(&cache.samples); + for (combined, current) in combined_candidates.iter_mut().zip(&cache.candidates) { + combined.combined.merge(¤t.combined); + } + } + finalize_results(&mut combined_candidates); + EvaluationReport { + schema_version: SCHEMA_VERSION, + family, + timing_note: "Single-pass wall-clock diagnostics; byte/count fields are the comparison \ + contract.", + threshold_note: "Small logical values are grouped into SST-local 8-12 KiB evaluation \ + units; medium values and blobs remain independent. Estimated stored \ + bytes apply the 12.5% rule per unit and exclude fixed container headers.", + caches, + combined_samples, + combined_candidates, + } +} + +struct TrainingSelection { + samples: Vec>, + caches: Vec, + inputs_exhausted: bool, + #[cfg(test)] + selected_cache_indices: Vec, +} + +fn select_training_samples( + paths: &[PathBuf], + family: u32, + source_dictionary: Option<&'static [u8]>, + byte_budget: usize, +) -> Result { + let mut paths = paths.to_vec(); + paths.sort(); + let mut iterators = paths + .into_iter() + .map(|path| CacheSampleIter::open(path, family, source_dictionary)) + .collect::>>()?; + let mut per_cache = iterators + .iter() + .map(|iter| TrainingCacheReport { + path: iter.path.clone(), + source_codecs: iter.source_codecs.clone(), + samples: Metric::default(), + }) + .collect::>(); + let mut samples = Vec::new(); + let mut selected_bytes = 0_usize; + let mut active = vec![true; iterators.len()]; + let mut active_count = iterators.len(); + #[cfg(test)] + let mut selected_cache_indices = Vec::new(); + + while active_count > 0 && selected_bytes < byte_budget { + for index in 0..iterators.len() { + if !active[index] { + continue; + } + match iterators[index].next_sample()? { + Some(sample) => { + selected_bytes += sample.data.len(); + per_cache[index].samples.add(sample.data.len()); + samples.push(Box::from(sample.data.as_ref())); + #[cfg(test)] + selected_cache_indices.push(index); + if selected_bytes >= byte_budget { + break; + } + } + None => { + active[index] = false; + active_count -= 1; + } + } + } + } + Ok(TrainingSelection { + samples, + caches: per_cache, + inputs_exhausted: active_count == 0, + #[cfg(test)] + selected_cache_indices, + }) +} + +fn write_dictionary(path: &Path, bytes: &[u8]) -> Result<()> { + let mut file = File::create(path) + .with_context(|| format!("Failed to create dictionary {}", path.display()))?; + file.write_all(bytes) + .with_context(|| format!("Failed to write dictionary {}", path.display())) +} + +fn source_dictionary(source: &Source) -> Result> { + source + .source_dictionary + .as_ref() + .map(|path| { + fs::read(path) + .with_context(|| format!("Failed to read source dictionary {}", path.display())) + .map(|bytes| Box::leak(bytes.into_boxed_slice()) as &'static [u8]) + }) + .transpose() +} + +fn train(source: &Source, output: &Path) -> Result { + let source_dictionary = source_dictionary(source)?; + let TrainingSelection { + samples, + caches, + inputs_exhausted, + .. + } = select_training_samples( + &source.caches, + source.family, + source_dictionary, + SAMPLE_BYTE_BUDGET, + )?; + ensure!(!samples.is_empty(), "No eligible values found for training"); + let selected = caches.iter().fold(Metric::default(), |mut total, cache| { + total.merge(&cache.samples); + total + }); + let selected_bytes = selected.bytes; + let dictionary = zstd::dict::from_samples(&samples, DICTIONARY_SIZE).with_context(|| { + format!( + "Failed to train a {DICTIONARY_SIZE}-byte dictionary from {} values ({selected_bytes} \ + bytes)", + samples.len() + ) + })?; + write_dictionary(output, &dictionary)?; + Ok(TrainingReport { + schema_version: SCHEMA_VERSION, + family: source.family, + zstd_version: zstd::zstd_safe::version_string(), + dictionary_size: DICTIONARY_SIZE, + sample_byte_target: SAMPLE_BYTE_BUDGET, + selected, + selected_bytes, + inputs_exhausted, + caches, + dictionary: dictionary_info(Some(output), &dictionary, false)?, + }) +} + +fn print_training(report: &TrainingReport) { + println!( + "Trained {} ({} bytes) from {} values / {} bytes (target {}, exhausted: {})", + report.dictionary.path.as_ref().unwrap().display(), + report.dictionary.bytes, + report.selected.count, + report.selected_bytes, + report.sample_byte_target, + report.inputs_exhausted, + ); + for cache in &report.caches { + println!( + " {} ({:?}): {} values / {} bytes", + cache.path.display(), + cache.source_codecs, + cache.samples.count, + cache.samples.bytes + ); + } +} + +fn print_evaluation(report: &EvaluationReport) { + let samples = &report.combined_samples; + println!( + "Evaluated family {}: {} caches, {} compression units / {} bytes", + report.family, + report.caches.len(), + samples.count, + samples.bytes + ); + for cache in &report.caches { + println!(" {}: {:?}", cache.path.display(), cache.source_codecs); + } + println!( + "{:<28} {:>15} {:>9} {:>18} {:>12} {:>12}", + "Candidate", "Raw compressed", "ratio", "Estimated stored", "encode ms", "decode ms" + ); + for result in &report.combined_candidates { + println!( + "{:<28} {:>15} {:>8.2}% {:>18} {:>12.3} {:>12.3}", + result.dictionary.name, + result.combined.raw_compressed_bytes, + result.combined.raw_compression_ratio.unwrap_or_default() * 100.0, + result.combined.estimated_stored_bytes, + result.combined.encode_duration.as_secs_f64() * 1_000.0, + result.combined.decode_duration.as_secs_f64() * 1_000.0, + ); + } + println!("Note: {}", report.threshold_note); + println!("Note: {}", report.timing_note); +} + +fn write_json(path: Option<&Path>, report: &impl Serialize) -> Result<()> { + if let Some(path) = path { + let file = File::create(path) + .with_context(|| format!("Failed to create JSON report {}", path.display()))?; + serde_json::to_writer_pretty(BufWriter::new(file), report)?; + } + Ok(()) +} + +fn run(cli: Cli) -> Result<()> { + match cli.command { + Command::Train { + source, + output, + json, + } => { + let report = train(&source, &output)?; + print_training(&report); + write_json(json.as_deref(), &report) + } + Command::Evaluate { + source, + dictionary, + json, + } => { + let mut candidates = make_candidates(&dictionary)?; + let source_dictionary = source_dictionary(&source)?; + let caches = source + .caches + .iter() + .map(|path| evaluate_cache(path, source.family, source_dictionary, &mut candidates)) + .collect::>>()?; + let report = combine_evaluation(source.family, caches, &candidates); + print_evaluation(&report); + write_json(json.as_deref(), &report) + } + } +} + +fn main() { + if let Err(error) = run(Cli::parse()) { + eprintln!("Error: {error:#}"); + std::process::exit(1); + } +} + +#[cfg(test)] +mod tests { + use std::{collections::BTreeSet, fs}; + + use anyhow::Result; + use byteorder::{BE, WriteBytesExt}; + use tempfile::TempDir; + use turbo_persistence::{ + Compression, CompressionConfig, DbConfig, MIN_SMALL_VALUE_BLOCK_SIZE, SerialScheduler, + TurboPersistence, + }; + + use super::{ + CacheSampleIter, DICTIONARY_SIZE, EvaluationSampleIter, Source, empty_results, + evaluate_cache, evaluate_sample, finalize_results, make_candidates, + select_training_samples, train, + }; + + fn make_cache(family: usize, compression: Compression, values: usize) -> Result { + make_cache_with_config(family, compression.into(), values) + } + + fn make_cache_with_config( + family: usize, + compression: CompressionConfig, + values: usize, + ) -> Result { + let tempdir = tempfile::tempdir()?; + let mut config = DbConfig::<8>::default(); + config.family_configs[family].compression = compression; + let db = TurboPersistence::::open_with_config( + tempdir.path().to_path_buf(), + config, + )?; + let batch = db.write_batch()?; + for index in 0..values { + let key = format!("key-{index:06}").into_bytes(); + let value = if index.is_multiple_of(10) { + vec![b'a' + (index % 26) as u8; 5000] + } else { + format!("function component{index}() {{ return null; }}") + .repeat(64) + .into_bytes() + }; + batch.put(family as u32, key, value.into())?; + } + let mut state = 0x1234_5678_u32; + let incompressible = (0..5000) + .map(|_| { + state ^= state << 13; + state ^= state >> 17; + state ^= state << 5; + state as u8 + }) + .collect::>(); + batch.put( + family as u32, + b"incompressible-medium".to_vec(), + incompressible.into(), + )?; + db.commit_write_batch(batch)?; + db.shutdown()?; + Ok(tempdir) + } + + #[test] + fn round_robin_samples_multiple_caches() -> Result<()> { + let first = make_cache(2, Compression::Zstd3, 20)?; + let second = make_cache(2, Compression::Lz4, 20)?; + let paths = [first.path().to_path_buf(), second.path().to_path_buf()]; + let selection = select_training_samples(&paths, 2, None, 10_000)?; + assert!(!selection.samples.is_empty()); + assert!(!selection.inputs_exhausted); + assert_eq!(selection.caches.len(), 2); + assert_eq!(&selection.selected_cache_indices[..2], &[0, 1]); + let source_codecs = selection + .caches + .iter() + .flat_map(|cache| cache.source_codecs.iter()) + .collect::>(); + assert!(source_codecs.contains(&"Zstd3".to_owned())); + assert!(source_codecs.contains(&"Lz4".to_owned())); + assert!( + selection + .caches + .iter() + .all(|report| report.samples.count > 0) + ); + Ok(()) + } + + #[test] + fn evaluation_groups_small_values_into_production_sized_blocks() -> Result<()> { + let cache = make_cache(2, Compression::Zstd3, 20)?; + let mut raw = CacheSampleIter::open(cache.path().to_path_buf(), 2, None)?; + let mut raw_count = 0; + while raw.next_sample()?.is_some() { + raw_count += 1; + } + + let source = CacheSampleIter::open(cache.path().to_path_buf(), 2, None)?; + let mut grouped = EvaluationSampleIter::new(source); + let mut sizes = Vec::new(); + while let Some(sample) = grouped.next_sample()? { + sizes.push(sample.data.len()); + } + assert!(sizes.len() < raw_count); + assert!(sizes.iter().any(|&size| { + (MIN_SMALL_VALUE_BLOCK_SIZE..MIN_SMALL_VALUE_BLOCK_SIZE + 4096).contains(&size) + })); + Ok(()) + } + + #[test] + fn accepts_lz4_source_cache_and_ignores_source_dictionary() -> Result<()> { + let cache = make_cache(2, Compression::Lz4, 20)?; + let mut iter = + CacheSampleIter::open(cache.path().to_path_buf(), 2, Some(b"irrelevant for LZ4"))?; + assert!(iter.source_codecs.contains("Lz4")); + assert!(iter.next_sample()?.is_some()); + + let cache = make_cache(2, Compression::Zstd3, 20)?; + let mut iter = CacheSampleIter::open( + cache.path().to_path_buf(), + 2, + Some(b"irrelevant for plain zstd"), + )?; + assert!(iter.source_codecs.contains("Zstd3")); + assert!(iter.next_sample()?.is_some()); + Ok(()) + } + + #[test] + fn trains_replaces_output_and_evaluates() -> Result<()> { + let cache = make_cache(2, Compression::Zstd3, 3000)?; + let output_dir = tempfile::tempdir()?; + let output = output_dir.path().join("dictionary.zdict"); + fs::write(&output, b"old")?; + let source = Source { + family: 2, + source_dictionary: None, + caches: vec![cache.path().to_path_buf()], + }; + let report = train(&source, &output)?; + assert_eq!(report.dictionary.bytes, DICTIONARY_SIZE); + assert_ne!(fs::read(&output)?, b"old"); + + let mut candidates = make_candidates(&[output])?; + let evaluation = evaluate_cache(cache.path(), 2, None, &mut candidates)?; + assert_eq!(evaluation.candidates.len(), 3); + assert_eq!(evaluation.candidates[0].dictionary.name, "lz4"); + assert_eq!( + evaluation.candidates[1].dictionary.name, + "zstd3 (no dictionary)" + ); + assert!(evaluation.samples.count > 0); + Ok(()) + } + + #[test] + fn reads_and_deduplicates_blob_samples() -> Result<()> { + let cache = make_cache(2, Compression::Zstd3, 20)?; + let value = b"export default function BlobComponent() {}".repeat(100); + let compressed = zstd::bulk::compress(&value, 3)?; + let mut blob = Vec::new(); + blob.write_u32::(value.len() as u32)?; + blob.write_u32::(turbo_persistence::checksum_block(&compressed))?; + blob.extend_from_slice(&compressed); + fs::write(cache.path().join("00000042.blob"), blob)?; + + let mut iter = CacheSampleIter::open(cache.path().to_path_buf(), 2, None)?; + let sample = iter + .sample_from_value( + turbo_persistence::IterValue::Blob { + sequence_number: 42, + }, + CompressionConfig::Zstd3, + 1, + )? + .unwrap(); + assert_eq!(sample.data.as_ref(), value); + let mut candidates = make_candidates(&[])?; + let mut results = empty_results(&candidates); + evaluate_sample(&sample, &mut candidates, &mut results)?; + finalize_results(&mut results); + assert_eq!( + results[0].combined.estimated_stored_bytes, + results[0].combined.raw_compressed_bytes + ); + assert!( + iter.sample_from_value( + turbo_persistence::IterValue::Blob { + sequence_number: 42 + }, + CompressionConfig::Zstd3, + 1, + )? + .is_none() + ); + assert_eq!(iter.duplicate_blob_references, 1); + assert!( + iter.sample_from_value( + turbo_persistence::IterValue::Blob { + sequence_number: 43 + }, + CompressionConfig::Zstd3, + 1, + ) + .is_err() + ); + Ok(()) + } + + #[test] + fn active_ssts_follow_current_deletions_and_supersession() -> Result<()> { + let cache = make_cache(2, Compression::Zstd3, 20)?; + let original = turbo_persistence::offline::collect_sst_info(cache.path())?; + assert_eq!(original[&2].len(), 1); + + fs::copy( + cache.path().join("00000002.meta"), + cache.path().join("00000003.meta"), + )?; + let mut current: serde_json::Value = + serde_json::from_slice(&fs::read(cache.path().join("CURRENT"))?)?; + current["max_sequence_number"] = 3.into(); + fs::write(cache.path().join("CURRENT"), serde_json::to_vec(¤t)?)?; + let superseded = turbo_persistence::offline::collect_sst_info(cache.path())?; + assert_eq!(superseded[&2].len(), 1); + + let mut deletion = Vec::new(); + deletion.write_u32::(3)?; + fs::write(cache.path().join("00000004.del"), deletion)?; + let deleted = turbo_persistence::offline::collect_sst_info(cache.path())?; + assert_eq!(deleted[&2].len(), 1); + Ok(()) + } +} diff --git a/turbopack/crates/turbo-persistence/src/compression.rs b/turbopack/crates/turbo-persistence/src/compression.rs index 69d645f23d61..fa517b383504 100644 --- a/turbopack/crates/turbo-persistence/src/compression.rs +++ b/turbopack/crates/turbo-persistence/src/compression.rs @@ -1,4 +1,4 @@ -use std::{cell::RefCell, mem::MaybeUninit, rc::Rc, sync::Arc}; +use std::{cell::RefCell, fmt, mem::MaybeUninit, rc::Rc, sync::Arc}; use anyhow::{Context, Result, ensure}; use lzzzz::lz4::{self, decompress}; @@ -14,17 +14,70 @@ pub enum Compression { Zstd3 = 1, } +/// Runtime compression configuration for a persistence family. +#[derive(Clone, Copy, Default, PartialEq, Eq)] +pub enum CompressionConfig { + #[default] + Lz4, + Zstd3, + Zstd3WithDictionary(&'static [u8]), +} + +impl fmt::Debug for CompressionConfig { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Lz4 => formatter.write_str("Lz4"), + Self::Zstd3 => formatter.write_str("Zstd3"), + Self::Zstd3WithDictionary(_) => formatter + .debug_struct("Zstd3WithDictionary") + .field("dictionary_id", &self.dictionary_id()) + .finish(), + } + } +} + +impl CompressionConfig { + pub fn algorithm(self) -> Compression { + match self { + Self::Lz4 => Compression::Lz4, + Self::Zstd3 | Self::Zstd3WithDictionary(_) => Compression::Zstd3, + } + } + + pub fn dictionary(self) -> Option<&'static [u8]> { + match self { + Self::Zstd3WithDictionary(dictionary) => Some(dictionary), + Self::Lz4 | Self::Zstd3 => None, + } + } + + pub fn dictionary_id(self) -> Option { + self.dictionary() + .and_then(zstd::zstd_safe::get_dict_id_from_dict) + .map(|id| id.get()) + } +} + +impl From for CompressionConfig { + fn from(value: Compression) -> Self { + match value { + Compression::Lz4 => Self::Lz4, + Compression::Zstd3 => Self::Zstd3, + } + } +} + thread_local! { /// Zstd decompression contexts are reusable and relatively expensive to create. Keep one per /// worker thread to avoid allocation on every block read without a global lock. - static ZSTD_DECOMPRESSOR: RefCell> = RefCell::new( - zstd::bulk::Decompressor::new().expect("zstd decompressor initialization should succeed") + static ZSTD_DECOMPRESSOR: RefCell<(Option<&'static [u8]>, zstd::bulk::Decompressor<'static>)> = RefCell::new( + (None, zstd::bulk::Decompressor::new().expect("zstd decompressor initialization should succeed")) ); } /// Decompresses `block` into `dest`, verifying the output length matches `expected_len`. fn decompress_block( - compression: Compression, + compression: CompressionConfig, block: &[u8], dest: &mut [u8], expected_len: u32, @@ -35,12 +88,27 @@ fn decompress_block( zero-copy mmap path" ); let bytes_written = match compression { - Compression::Lz4 => decompress(block, dest).map_err(anyhow::Error::from), - Compression::Zstd3 => ZSTD_DECOMPRESSOR.with_borrow_mut(|decompressor| { - decompressor - .decompress_to_buffer(block, dest) - .map_err(anyhow::Error::from) - }), + CompressionConfig::Lz4 => decompress(block, dest).map_err(anyhow::Error::from), + CompressionConfig::Zstd3 | CompressionConfig::Zstd3WithDictionary(_) => ZSTD_DECOMPRESSOR + .with_borrow_mut(|state| { + let dictionary = compression.dictionary(); + let same_dictionary = match (state.0, dictionary) { + (Some(current), Some(next)) => std::ptr::eq(current, next), + (None, None) => true, + _ => false, + }; + if !same_dictionary { + state + .1 + .set_dictionary(dictionary.unwrap_or_default()) + .map_err(anyhow::Error::from)?; + state.0 = dictionary; + } + state + .1 + .decompress_to_buffer(block, dest) + .map_err(anyhow::Error::from) + }), } .with_context(|| { format!( @@ -63,7 +131,7 @@ fn decompress_block( /// The caller must ensure `uncompressed_length > 0` (i.e., the block is actually compressed). /// Uncompressed blocks should be handled via zero-copy mmap slices before calling this. pub(crate) fn decompress_into_arc( - compression: Compression, + compression: CompressionConfig, uncompressed_length: u32, block: &[u8], ) -> Result> { @@ -81,7 +149,7 @@ pub(crate) fn decompress_into_arc( /// Like [`decompress_into_arc`] but returns an `Rc<[u8]>` for thread-local use. pub(crate) fn decompress_into_rc( - compression: Compression, + compression: CompressionConfig, uncompressed_length: u32, block: &[u8], ) -> Result> { @@ -101,17 +169,21 @@ pub fn checksum_block(data: &[u8]) -> u32 { /// Reusable compressor for a stream of blocks using the same family configuration. pub(crate) struct Compressor { - compression: Compression, + compression: CompressionConfig, zstd: Option>, } impl Compressor { - pub(crate) fn new(compression: Compression) -> Result { + pub(crate) fn new(compression: CompressionConfig) -> Result { let zstd = match compression { - Compression::Zstd3 => { + CompressionConfig::Zstd3 => { Some(zstd::bulk::Compressor::new(3).context("Failed to create zstd compressor")?) } - Compression::Lz4 => None, + CompressionConfig::Zstd3WithDictionary(dictionary) => Some( + zstd::bulk::Compressor::with_dictionary(3, dictionary) + .context("Failed to create zstd dictionary compressor")?, + ), + CompressionConfig::Lz4 => None, }; Ok(Self { compression, zstd }) } @@ -123,11 +195,11 @@ impl Compressor { buffer: &mut Vec, ) -> Result<()> { match self.compression { - Compression::Lz4 => { + CompressionConfig::Lz4 => { lz4::compress_to_vec(block, buffer, lz4::ACC_LEVEL_DEFAULT) .context("LZ4 compression failed")?; } - Compression::Zstd3 => { + CompressionConfig::Zstd3 | CompressionConfig::Zstd3WithDictionary(_) => { buffer.reserve(zstd::zstd_safe::compress_bound(block.len())); self.zstd .as_mut() @@ -144,10 +216,33 @@ impl Compressor { mod tests { use super::*; + #[test] + fn dictionary_compression_round_trips() { + let samples = (0..100) + .map(|index| format!("export default function Component{index}() {{ return null }}")) + .collect::>(); + let dictionary = zstd::dict::from_samples(&samples, 1024).unwrap(); + let dictionary = Box::leak(dictionary.into_boxed_slice()); + let config = CompressionConfig::Zstd3WithDictionary(dictionary); + assert_eq!(config.algorithm(), Compression::Zstd3); + assert!(config.dictionary_id().is_some()); + let input = samples.concat(); + let mut compressor = Compressor::new(config).unwrap(); + let mut compressed = Vec::new(); + compressor + .compress_into_buffer(input.as_bytes(), &mut compressed) + .unwrap(); + let output = decompress_into_arc(config, input.len() as u32, &compressed).unwrap(); + assert_eq!(&*output, input.as_bytes()); + assert!( + decompress_into_arc(CompressionConfig::Zstd3, input.len() as u32, &compressed).is_err() + ); + } + #[test] fn compression_round_trips() { let input = b"turbo persistence compression ".repeat(1024); - for compression in [Compression::Lz4, Compression::Zstd3] { + for compression in [CompressionConfig::Lz4, CompressionConfig::Zstd3] { let mut compressor = Compressor::new(compression).unwrap(); let mut compressed = Vec::new(); compressor diff --git a/turbopack/crates/turbo-persistence/src/db.rs b/turbopack/crates/turbo-persistence/src/db.rs index 72d5a3a3acf3..4be939e6fa9c 100644 --- a/turbopack/crates/turbo-persistence/src/db.rs +++ b/turbopack/crates/turbo-persistence/src/db.rs @@ -30,10 +30,10 @@ use tracing::span::EnteredSpan; pub use crate::compaction::selector::CompactConfig; use crate::{ - AccessMode, DbConfig, FamilyKind, QueryKey, + AccessMode, CompressionConfig, DbConfig, FamilyKind, QueryKey, arc_bytes::ArcBytes, compaction::selector::{Compactable, get_merge_segments}, - compression::{Compression, checksum_block, decompress_into_arc}, + compression::{checksum_block, decompress_into_arc}, constants::{ DATA_THRESHOLD_PER_COMPACTED_FILE, KEY_BLOCK_AVG_SIZE, KEY_BLOCK_CACHE_SIZE, MAX_ENTRIES_PER_COMPACTED_FILE, VALUE_BLOCK_AVG_SIZE, VALUE_BLOCK_CACHE_SIZE, @@ -659,7 +659,7 @@ impl TurboPersistence /// Reads and decompresses a blob file. This is not backed by any cache. #[tracing::instrument(level = "info", name = "reading database blob", skip_all)] - fn read_blob(&self, seq: u32, compression: Compression) -> Result { + fn read_blob(&self, seq: u32, compression: CompressionConfig) -> Result { let path = self.path.join(format!("{seq:08}.blob")); let file = File::open(&path)?; let data: Either> = match self.config.access_mode { @@ -1531,6 +1531,7 @@ impl TurboPersistence "merge files", family = self.config.family_configs[family as usize].name ); + let compression = self.config.family_configs[family as usize].compression; enum PartialMergeResult<'l> { Merged { new_sst_files: Vec<(u32, File, StaticSortedFileBuilderMeta<'static>)>, @@ -1617,7 +1618,7 @@ impl TurboPersistence StaticSortedFileIter::open( path, entry.sst_metadata(), - meta_file.compression(), + compression, self.config.access_mode, ) }) @@ -1636,7 +1637,7 @@ impl TurboPersistence /// used set). writer: Option<(u32, StreamingSstWriter)>, flags: MetaEntryFlags, - compression: Compression, + compression: CompressionConfig, new_sst_files: Vec<(u32, File, StaticSortedFileBuilderMeta<'static>)>, /// Hash of the last key added. Used to ensure we only split @@ -1644,7 +1645,10 @@ impl TurboPersistence last_hash: Option, } impl Collector { - fn new(flags: MetaEntryFlags, compression: Compression) -> Self { + fn new( + flags: MetaEntryFlags, + compression: CompressionConfig, + ) -> Self { Self { writer: None, flags, diff --git a/turbopack/crates/turbo-persistence/src/lib.rs b/turbopack/crates/turbo-persistence/src/lib.rs index 67fe6ad4ad09..fde00029ed4d 100644 --- a/turbopack/crates/turbo-persistence/src/lib.rs +++ b/turbopack/crates/turbo-persistence/src/lib.rs @@ -16,6 +16,8 @@ mod merge_iter; pub mod meta_file; mod meta_file_builder; pub mod mmap_helper; +#[doc(hidden)] +pub mod offline; mod parallel_scheduler; mod rc_bytes; mod shared_bytes; @@ -30,7 +32,7 @@ mod write_batch; mod tests; pub use arc_bytes::ArcBytes; -pub use compression::{Compression, checksum_block}; +pub use compression::{Compression, CompressionConfig, checksum_block}; pub use db::{ CommitStats, CompactConfig, CurrentDbVersion, MetaFileEntryInfo, MetaFileInfo, TurboPersistence, read_current_version, @@ -64,7 +66,7 @@ pub enum FamilyKind { pub struct FamilyConfig { pub name: &'static str, pub kind: FamilyKind, - pub compression: Compression, + pub compression: CompressionConfig, } /// Database-wide configuration with per-family storage settings. @@ -102,7 +104,7 @@ impl DbConfig { family_configs: [FamilyConfig { name: "unknown", kind: FamilyKind::SingleValue, - compression: Compression::Lz4, + compression: CompressionConfig::Lz4, }; FAMILIES], access_mode: access_mode_env_var(), } @@ -110,7 +112,7 @@ impl DbConfig { } /// The largest value that [`WriteBatch::delete_value`] can delete, since the tombstone stores /// a copy of the value inline. -pub use constants::MAX_INLINE_VALUE_SIZE; +pub use constants::{MAX_INLINE_VALUE_SIZE, MIN_SMALL_VALUE_BLOCK_SIZE}; impl Default for DbConfig { fn default() -> Self { @@ -118,11 +120,12 @@ impl Default for DbConfig { } } pub use key::{KeyBase, QueryKey, StoreKey, hash_key}; +pub use lookup_entry::{IterValue, LookupEntry}; pub use meta_file::MetaEntryFlags; pub use parallel_scheduler::{ParallelScheduler, SerialScheduler}; pub use static_sorted_file::{ BlockCache, BlockCacheLifecycle, BlockWeighter, KeyBlockLayout, SstLookupResult, - StaticSortedFile, StaticSortedFileMetaData, + StaticSortedFile, StaticSortedFileIter, StaticSortedFileMetaData, }; pub use static_sorted_file_builder::{ BLOCK_HEADER_SIZE, Entry, EntryValue, StreamingSstWriter, write_static_stored_file, diff --git a/turbopack/crates/turbo-persistence/src/meta_file.rs b/turbopack/crates/turbo-persistence/src/meta_file.rs index 0a288ceb284f..a1c0d810a2a6 100644 --- a/turbopack/crates/turbo-persistence/src/meta_file.rs +++ b/turbopack/crates/turbo-persistence/src/meta_file.rs @@ -15,7 +15,7 @@ use smallvec::SmallVec; use zerocopy::{FromBytes, Immutable, IntoBytes, KnownLayout, Ref, big_endian as be}; use crate::{ - AccessMode, Compression, FamilyConfig, QueryKey, + AccessMode, Compression, CompressionConfig, FamilyConfig, QueryKey, lookup_entry::LookupValue, mmap_helper::advise_mmap_for_persistence, static_sorted_file::{BlockCache, SstLookupResult, StaticSortedFile, StaticSortedFileMetaData}, @@ -119,7 +119,7 @@ pub struct MetaEntry { /// The `'static` lifetime is transmuted — the actual borrow is from `MetaFile::backing`. amqf: qfilter::FilterRef<'static>, /// Compression recorded in this entry's meta file. - compression: Compression, + compression: CompressionConfig, /// The static sorted file that is lazily loaded sst: OnceLock, } @@ -268,8 +268,10 @@ pub struct MetaFile { sequence_number: u32, /// The key family of the SST files in this meta file. family: u32, - /// Compression recorded for this family. + /// Compression algorithm recorded for this family. compression: Compression, + /// Zstd dictionary ID recorded for this family, or zero without a dictionary. + dictionary_id: u32, /// The entries of the file. Dropped before `backing` (field declaration order). entries: Vec, /// The entries that have been marked as obsolete. @@ -341,17 +343,22 @@ impl MetaFile { value if value == Compression::Zstd3 as u8 => Compression::Zstd3, value => bail!("Invalid compression algorithm {value}"), }; - if let Some(configs) = family_configs { + let dictionary_id = reader.read_u32::()?; + let compression_config = if let Some(configs) = family_configs { let configured = configs .get(family as usize) .with_context(|| format!("No configuration for family {family}"))? .compression; ensure!( - compression == configured, + compression == configured.algorithm() + && dictionary_id == configured.dictionary_id().unwrap_or(0), "Compression configuration mismatch for family {family}: meta file uses \ - {compression:?}, runtime config uses {configured:?}" + {compression:?} dictionary {dictionary_id}, runtime config uses {configured:?}" ); - } + configured + } else { + CompressionConfig::from(compression) + }; let obsolete_count = reader.read_u32::()?; let mut obsolete_sst_files = Vec::with_capacity(obsolete_count as usize); for _ in 0..obsolete_count { @@ -409,7 +416,7 @@ impl MetaFile { flags, amqf_data_offset: start_of_amqf_data_offset..end_of_amqf_data_offset, amqf, - compression, + compression: compression_config, sst: OnceLock::new(), }); start_of_amqf_data_offset = end_of_amqf_data_offset; @@ -423,6 +430,7 @@ impl MetaFile { sequence_number, family, compression, + dictionary_id, entries, obsolete_entries: Vec::new(), obsolete_sst_files, @@ -458,6 +466,10 @@ impl MetaFile { self.compression } + pub fn dictionary_id(&self) -> u32 { + self.dictionary_id + } + /// The on-disk size of this meta file in bytes (the length of its memory map). pub fn byte_size(&self) -> u64 { self.backing.len() as u64 diff --git a/turbopack/crates/turbo-persistence/src/meta_file_builder.rs b/turbopack/crates/turbo-persistence/src/meta_file_builder.rs index 0b4dadd26292..3d74ba1f721d 100644 --- a/turbopack/crates/turbo-persistence/src/meta_file_builder.rs +++ b/turbopack/crates/turbo-persistence/src/meta_file_builder.rs @@ -10,14 +10,14 @@ use qfilter::Filter; use zerocopy::IntoBytes; use crate::{ - Compression, + CompressionConfig, meta_file::{EntryHeader, META_FILE_MAGIC}, static_sorted_file_builder::StaticSortedFileBuilderMeta, }; pub struct MetaFileBuilder<'a> { family: u32, - compression: Compression, + compression: CompressionConfig, /// Entries in the meta file, tuples of (sequence_number, StaticSortedFileBuilderMetaResult) entries: Vec<(u32, StaticSortedFileBuilderMeta<'a>)>, /// Obsolete SST files, represented by their sequence numbers @@ -27,7 +27,7 @@ pub struct MetaFileBuilder<'a> { } impl<'a> MetaFileBuilder<'a> { - pub fn new(family: u32, compression: Compression) -> Self { + pub fn new(family: u32, compression: CompressionConfig) -> Self { Self { family, compression, @@ -62,7 +62,8 @@ impl<'a> MetaFileBuilder<'a> { let mut file = CountingWriter::new(BufWriter::new(File::create(file)?)); file.write_u32::(META_FILE_MAGIC)?; // Magic number file.write_u32::(self.family)?; - file.write_u8(self.compression as u8)?; + file.write_u8(self.compression.algorithm() as u8)?; + file.write_u32::(self.compression.dictionary_id().unwrap_or(0))?; self.obsolete_sst_files.sort(); file.write_u32::(self.obsolete_sst_files.len() as u32)?; diff --git a/turbopack/crates/turbo-persistence/src/offline.rs b/turbopack/crates/turbo-persistence/src/offline.rs new file mode 100644 index 000000000000..44daa0ae1643 --- /dev/null +++ b/turbopack/crates/turbo-persistence/src/offline.rs @@ -0,0 +1,196 @@ +//! Shared helpers for offline inspection of persistence databases. + +use std::{ + collections::{BTreeMap, HashSet}, + path::Path, + sync::Arc, +}; + +use anyhow::{Context, Result, bail, ensure}; +use byteorder::{BE, ReadBytesExt}; +use fs_err as fs; + +use crate::{ + AccessMode, Compression, CompressionConfig, checksum_block, compression::decompress_into_arc, + meta_file::MetaFile, read_current_version, sst_filter::SstFilter, +}; + +/// Information about an active SST recorded by a meta file. +#[derive(Clone, Copy, Debug)] +pub struct SstInfo { + pub sequence_number: u32, + pub block_count: u16, + pub compression: Compression, + pub dictionary_id: u32, +} + +/// Collects active SSTs by family, mirroring database open logic. +pub fn collect_sst_info(db_path: &Path) -> Result>> { + let current = read_current_version(db_path)? + .context("CURRENT file is missing")? + .max_sequence_number; + + let mut deleted_seqs = HashSet::new(); + for entry in fs::read_dir(db_path) + .with_context(|| format!("Failed to read database directory {}", db_path.display()))? + { + let path = entry?.path(); + if path.extension().and_then(|extension| extension.to_str()) == Some("del") { + let content = fs::read(&path) + .with_context(|| format!("Failed to read deletion file {}", path.display()))?; + let mut cursor: &[u8] = &content; + while !cursor.is_empty() { + deleted_seqs.insert( + cursor.read_u32::().with_context(|| { + format!("Truncated sequence number in {}", path.display()) + })?, + ); + } + } + } + + let mut meta_seqs: Vec = fs::read_dir(db_path)? + .filter_map(|entry| entry.ok()) + .filter_map(|entry| { + let path = entry.path(); + if path.extension().and_then(|extension| extension.to_str()) != Some("meta") { + return None; + } + let sequence: u32 = path.file_stem()?.to_str()?.parse().ok()?; + (sequence <= current && !deleted_seqs.contains(&sequence)).then_some(sequence) + }) + .collect(); + if meta_seqs.is_empty() { + bail!("No active .meta files found in {}", db_path.display()); + } + meta_seqs.sort_unstable(); + + let mut meta_files: Vec = meta_seqs + .iter() + .map(|&sequence| { + MetaFile::open(db_path, sequence, None, AccessMode::Mmap) + .with_context(|| format!("Failed to open {sequence:08}.meta")) + }) + .collect::>()?; + let mut sst_filter = SstFilter::new(); + for meta in meta_files.iter_mut().rev() { + sst_filter.apply_filter(meta); + } + + let mut families: BTreeMap> = BTreeMap::new(); + for meta in &meta_files { + for entry in meta.entries() { + families.entry(meta.family()).or_default().push(SstInfo { + sequence_number: entry.sequence_number(), + block_count: entry.block_count(), + compression: meta.compression(), + dictionary_id: meta.dictionary_id(), + }); + } + } + Ok(families) +} + +/// Verifies and reconstructs a raw medium-value block from an SST iterator. +pub fn decode_medium( + compression: CompressionConfig, + uncompressed_length: u32, + expected_checksum: u32, + stored: &[u8], +) -> Result> { + verify_checksum(stored, expected_checksum, "medium value")?; + if uncompressed_length > 0 { + decompress_into_arc(compression, uncompressed_length, stored) + .context("Failed to decompress medium value") + } else { + Ok(Arc::from(stored)) + } +} + +/// Reads, verifies, and decompresses one blob file. +pub fn read_blob( + db_path: &Path, + sequence_number: u32, + compression: CompressionConfig, +) -> Result> { + let path = db_path.join(format!("{sequence_number:08}.blob")); + let content = fs::read(&path).with_context(|| format!("Failed to read {}", path.display()))?; + ensure!( + content.len() >= 8, + "Blob file {} is truncated", + path.display() + ); + let mut reader: &[u8] = &content; + let uncompressed_length = reader.read_u32::()?; + let expected_checksum = reader.read_u32::()?; + verify_checksum( + reader, + expected_checksum, + &format!("blob file {}", path.display()), + )?; + // Blob writers always compress payloads; unlike SST blocks, the blob format currently has no + // zero-length sentinel for an uncompressed payload. + ensure!( + uncompressed_length > 0, + "Blob file {} has an invalid uncompressed length of zero", + path.display() + ); + decompress_into_arc(compression, uncompressed_length, reader) + .with_context(|| format!("Failed to decompress {}", path.display())) +} + +fn verify_checksum(data: &[u8], expected: u32, description: &str) -> Result<()> { + let actual = checksum_block(data); + ensure!( + actual == expected, + "Checksum mismatch in {description} (expected {expected:08x}, got {actual:08x})" + ); + Ok(()) +} + +#[cfg(test)] +mod tests { + use byteorder::{BE, WriteBytesExt}; + + use super::{decode_medium, read_blob}; + use crate::{CompressionConfig, checksum_block, compression::Compressor}; + + #[test] + fn decodes_compressed_and_uncompressed_medium_values() -> anyhow::Result<()> { + let value = b"function component() { return null; }".repeat(100); + let mut compressed = Vec::new(); + Compressor::new(CompressionConfig::Zstd3)?.compress_into_buffer(&value, &mut compressed)?; + let decoded = decode_medium( + CompressionConfig::Zstd3, + value.len() as u32, + checksum_block(&compressed), + &compressed, + )?; + assert_eq!(decoded.as_ref(), value); + + let decoded = decode_medium(CompressionConfig::Zstd3, 0, checksum_block(&value), &value)?; + assert_eq!(decoded.as_ref(), value); + Ok(()) + } + + #[test] + fn reads_blob_and_rejects_bad_checksum() -> anyhow::Result<()> { + let directory = tempfile::tempdir()?; + let value = b"blob data".repeat(100); + let compressed = zstd::bulk::compress(&value, 3)?; + let mut file = Vec::new(); + file.write_u32::(value.len() as u32)?; + file.write_u32::(checksum_block(&compressed))?; + file.extend_from_slice(&compressed); + fs_err::write(directory.path().join("00000001.blob"), &file)?; + assert_eq!( + read_blob(directory.path(), 1, CompressionConfig::Zstd3)?.as_ref(), + value + ); + + file[4] ^= 1; + fs_err::write(directory.path().join("00000001.blob"), file)?; + assert!(read_blob(directory.path(), 1, CompressionConfig::Zstd3).is_err()); + Ok(()) + } +} diff --git a/turbopack/crates/turbo-persistence/src/rc_bytes.rs b/turbopack/crates/turbo-persistence/src/rc_bytes.rs index c4c1007cab32..e5d9d019769e 100644 --- a/turbopack/crates/turbo-persistence/src/rc_bytes.rs +++ b/turbopack/crates/turbo-persistence/src/rc_bytes.rs @@ -9,7 +9,7 @@ use std::{ use memmap2::Mmap; use crate::{ - Compression, + CompressionConfig, compression::decompress_into_rc, shared_bytes::{SharedBytes, is_subslice_of}, }; @@ -126,7 +126,7 @@ impl SharedBytes for RcBytes { } fn from_decompressed( - compression: Compression, + compression: CompressionConfig, uncompressed_length: u32, block: &[u8], ) -> anyhow::Result { diff --git a/turbopack/crates/turbo-persistence/src/shared_bytes.rs b/turbopack/crates/turbo-persistence/src/shared_bytes.rs index 6dc0f8407d8f..8fab799fcb1e 100644 --- a/turbopack/crates/turbo-persistence/src/shared_bytes.rs +++ b/turbopack/crates/turbo-persistence/src/shared_bytes.rs @@ -2,7 +2,7 @@ use std::ops::{Deref, Range}; use memmap2::Mmap; -use crate::Compression; +use crate::CompressionConfig; /// Trait abstracting over `ArcBytes` and `RcBytes`. /// @@ -39,7 +39,7 @@ pub trait SharedBytes: Clone + Deref + Sized { /// Creates an instance from a decompressed block. fn from_decompressed( - compression: Compression, + compression: CompressionConfig, uncompressed_length: u32, block: &[u8], ) -> anyhow::Result; diff --git a/turbopack/crates/turbo-persistence/src/static_sorted_file.rs b/turbopack/crates/turbo-persistence/src/static_sorted_file.rs index 9438ea64abf2..d5ca85a1a9fc 100644 --- a/turbopack/crates/turbo-persistence/src/static_sorted_file.rs +++ b/turbopack/crates/turbo-persistence/src/static_sorted_file.rs @@ -19,7 +19,7 @@ use rustc_hash::FxHasher; use smallvec::SmallVec; use crate::{ - AccessMode, Compression, QueryKey, + AccessMode, CompressionConfig, QueryKey, arc_bytes::ArcBytes, be, compression::checksum_block, @@ -32,7 +32,7 @@ use crate::{ }; /// The block header for an index block. -pub const BLOCK_TYPE_INDEX: u8 = 0; +pub(crate) const BLOCK_TYPE_INDEX: u8 = 0; /// The block header for a key block with 8-byte hash per entry. pub const BLOCK_TYPE_KEY_WITH_HASH: u8 = 1; /// The block header for a key block without hash. Entries are ordered by key. @@ -195,13 +195,13 @@ trait ValueBlockCache { self, meta: &StaticSortedFileMetaData, block_index: u16, - compression: Compression, + compression: CompressionConfig, ) -> Result; fn read_uncached( self, meta: &StaticSortedFileMetaData, block_index: u16, - compression: Compression, + compression: CompressionConfig, ) -> Result; } @@ -218,7 +218,7 @@ impl ValueBlockCache for ArcBlockCacheReader<'_> { self, meta: &StaticSortedFileMetaData, block_index: u16, - compression: Compression, + compression: CompressionConfig, ) -> Result { get_or_cache_block( self.backing, @@ -234,7 +234,7 @@ impl ValueBlockCache for ArcBlockCacheReader<'_> { self, meta: &StaticSortedFileMetaData, block_index: u16, - compression: Compression, + compression: CompressionConfig, ) -> Result { read_block_lookup(self.backing, meta, block_index, compression) } @@ -251,7 +251,7 @@ impl ValueBlockCache for RcBlockCacheReader<'_> { self, meta: &StaticSortedFileMetaData, block_index: u16, - compression: Compression, + compression: CompressionConfig, ) -> Result { if let Some((idx, block)) = self.cache.as_ref() && *idx == block_index @@ -267,7 +267,7 @@ impl ValueBlockCache for RcBlockCacheReader<'_> { self, meta: &StaticSortedFileMetaData, block_index: u16, - compression: Compression, + compression: CompressionConfig, ) -> Result { read_block_iter(self.backing, meta, block_index, compression) } @@ -307,7 +307,7 @@ pub struct StaticSortedFile { /// bitmap the CRC would be re-computed on every access. `Relaxed` ordering /// suffices: racing first-time verifications are idempotent. verified_blocks: Box<[AtomicU64]>, - compression: Compression, + compression: CompressionConfig, } impl StaticSortedFile { @@ -315,7 +315,7 @@ impl StaticSortedFile { pub fn open( db_path: &Path, meta: StaticSortedFileMetaData, - compression: Compression, + compression: CompressionConfig, access_mode: AccessMode, ) -> Result { let filename = format!("{:08}.sst", meta.sequence_number); @@ -631,7 +631,7 @@ fn get_or_cache_block( block_index: u16, cache: &BlockCache, verified_blocks: &[AtomicU64], - compression: Compression, + compression: CompressionConfig, ) -> Result { let mmap_block = if let StaticSortedFileBacking::Mmap(mmap) = backing { let (uncompressed_length, checksum, block_data) = @@ -882,7 +882,7 @@ fn read_block_lookup( backing: &StaticSortedFileBacking, meta: &StaticSortedFileMetaData, block_index: u16, - compression: Compression, + compression: CompressionConfig, ) -> Result { let (uncompressed_length, checksum, block) = get_raw_block(backing, meta, block_index)?; verify_checksum(meta, &block, checksum, block_index)?; @@ -961,7 +961,7 @@ fn read_block_iter( backing: &StaticSortedFileIterBacking, meta: &StaticSortedFileMetaData, block_index: u16, - compression: Compression, + compression: CompressionConfig, ) -> Result { let (uncompressed_length, checksum, block) = get_raw_block_iter(backing, meta, block_index)?; verify_checksum(meta, &block, checksum, block_index)?; @@ -982,7 +982,7 @@ fn handle_key_match_generic( ty: u8, val: &[u8], key_block: &B, - compression: Compression, + compression: CompressionConfig, reader: impl ValueBlockCache, ) -> Result> { Ok(match ty { @@ -1048,7 +1048,7 @@ pub struct StaticSortedFileIter { /// value blocks sequentially and don't revisit earlier blocks, so caching /// just the current one avoids redundant decompression. value_block_cache: Option<(u16, RcBytes)>, - compression: Compression, + compression: CompressionConfig, } enum CurrentKeyBlockKind { @@ -1120,7 +1120,7 @@ impl StaticSortedFileIter { pub fn open( db_path: &Path, meta: StaticSortedFileMetaData, - compression: Compression, + compression: CompressionConfig, access_mode: AccessMode, ) -> Result { let filename = format!("{:08}.sst", meta.sequence_number); @@ -1166,7 +1166,7 @@ impl StaticSortedFileIter { fn new( backing: StaticSortedFileIterBacking, meta: StaticSortedFileMetaData, - compression: Compression, + compression: CompressionConfig, ) -> Result { let root_block_index = meta.block_count - 1; let block = read_block_iter(&backing, &meta, root_block_index, compression)?; @@ -1206,7 +1206,7 @@ impl StaticSortedFileIter { backing: &StaticSortedFileIterBacking, meta: &StaticSortedFileMetaData, block_index: u16, - compression: Compression, + compression: CompressionConfig, ) -> Result { let block = read_block_iter(backing, meta, block_index, compression)?; let data = &*block; diff --git a/turbopack/crates/turbo-persistence/src/static_sorted_file_builder.rs b/turbopack/crates/turbo-persistence/src/static_sorted_file_builder.rs index c112788ed0f5..e79c7f0d58cf 100644 --- a/turbopack/crates/turbo-persistence/src/static_sorted_file_builder.rs +++ b/turbopack/crates/turbo-persistence/src/static_sorted_file_builder.rs @@ -10,7 +10,7 @@ use byteorder::{BE, ByteOrder, WriteBytesExt}; use fs_err::File; use crate::{ - Compression, + CompressionConfig, compression::{Compressor, checksum_block}, constants::{MAX_INLINE_VALUE_SIZE, MAX_SMALL_VALUE_SIZE, MIN_SMALL_VALUE_BLOCK_SIZE}, meta_file::MetaEntryFlags, @@ -340,7 +340,7 @@ pub fn write_static_stored_file( entries: &[E], file: &Path, flags: MetaEntryFlags, - compression: Compression, + compression: CompressionConfig, ) -> Result<(StaticSortedFileBuilderMeta<'static>, File)> { debug_assert!(entries.iter().map(|e| e.key_hash()).is_sorted()); let mut writer = StreamingSstWriter::new(file, flags, entries.len() as u64, compression)?; @@ -616,7 +616,7 @@ impl StreamingSstWriter { file: &Path, flags: MetaEntryFlags, max_entry_count: u64, - compression: Compression, + compression: CompressionConfig, ) -> Result { let file = BufWriter::new(File::create(file)?); let compressor = Compressor::new(compression)?; @@ -1508,7 +1508,7 @@ mod tests { sequence_number: seq, block_count: meta.block_count, }, - Compression::Lz4, + CompressionConfig::Lz4, AccessMode::Mmap, ) } @@ -1521,8 +1521,12 @@ mod tests { flags: MetaEntryFlags, ) -> Result> { let sst_path = dir.join(format!("{seq:08}.sst")); - let mut writer = - StreamingSstWriter::new(&sst_path, flags, entries.len() as u64, Compression::Lz4)?; + let mut writer = StreamingSstWriter::new( + &sst_path, + flags, + entries.len() as u64, + CompressionConfig::Lz4, + )?; for entry in entries { writer.add(entry)?; } @@ -1757,9 +1761,13 @@ mod tests { fn is_full_entry_count_limit() { let dir = tempfile::tempdir().unwrap(); let sst_path = dir.path().join("test.sst"); - let mut writer = - StreamingSstWriter::new(&sst_path, MetaEntryFlags::default(), 100, Compression::Lz4) - .unwrap(); + let mut writer = StreamingSstWriter::new( + &sst_path, + MetaEntryFlags::default(), + 100, + CompressionConfig::Lz4, + ) + .unwrap(); let max_entries = 50; for i in 0..max_entries { @@ -1784,9 +1792,13 @@ mod tests { fn is_full_data_size_limit() { let dir = tempfile::tempdir().unwrap(); let sst_path = dir.path().join("test.sst"); - let mut writer = - StreamingSstWriter::new(&sst_path, MetaEntryFlags::default(), 100, Compression::Lz4) - .unwrap(); + let mut writer = StreamingSstWriter::new( + &sst_path, + MetaEntryFlags::default(), + 100, + CompressionConfig::Lz4, + ) + .unwrap(); let value = vec![0u8; 1000]; for i in 0..10 { @@ -1826,7 +1838,7 @@ mod tests { &entries, &batch_path, MetaEntryFlags::default(), - Compression::Lz4, + CompressionConfig::Lz4, )?; // Write via streaming API @@ -1835,7 +1847,7 @@ mod tests { &streaming_path, MetaEntryFlags::default(), entries.len() as u64, - Compression::Lz4, + CompressionConfig::Lz4, )?; for entry in &entries { writer.add(entry)?; @@ -1855,7 +1867,7 @@ mod tests { sequence_number: 1, block_count: meta1.block_count, }, - Compression::Lz4, + CompressionConfig::Lz4, AccessMode::Mmap, )?; let sst2 = StaticSortedFile::open( @@ -1864,7 +1876,7 @@ mod tests { sequence_number: 2, block_count: meta2.block_count, }, - Compression::Lz4, + CompressionConfig::Lz4, AccessMode::Mmap, )?; let kc = make_cache(); @@ -1924,7 +1936,7 @@ mod tests { &sst_path, MetaEntryFlags::default(), 0, - Compression::Lz4, + CompressionConfig::Lz4, ) .unwrap(); writer.close().unwrap(); diff --git a/turbopack/crates/turbo-persistence/src/tests.rs b/turbopack/crates/turbo-persistence/src/tests.rs index a3f9c6595de4..789338bf59ff 100644 --- a/turbopack/crates/turbo-persistence/src/tests.rs +++ b/turbopack/crates/turbo-persistence/src/tests.rs @@ -5,7 +5,8 @@ use rayon::iter::{IntoParallelIterator, ParallelIterator}; use rstest::rstest; use crate::{ - AccessMode, Compression, DbConfig, FamilyConfig, FamilyKind, + AccessMode, Compression, CompressionConfig, DbConfig, FamilyConfig, FamilyKind, + SerialScheduler, constants::{MAX_INLINE_VALUE_SIZE, MAX_MEDIUM_VALUE_SIZE, MAX_SMALL_VALUE_SIZE}, db::{CompactConfig, TurboPersistence, read_current_version}, lookup_entry::IterValue, @@ -148,7 +149,7 @@ fn multi_value_config_with_mmap(mmap: bool) -> DbConfig<1> { family_configs: [FamilyConfig { name: "test", kind: FamilyKind::MultiValue, - compression: Compression::Lz4, + compression: Compression::Lz4.into(), }], access_mode: if mmap { AccessMode::Mmap @@ -1054,7 +1055,7 @@ fn batch_get_different_sizes(#[case] mmap: bool) -> Result<()> { let path = tempdir.path(); let mut config = config_with_mmap(mmap); - config.family_configs[0].compression = Compression::Zstd3; + config.family_configs[0].compression = Compression::Zstd3.into(); let db = open_db_with_config::<16>(path, config)?; // Write values of different sizes @@ -1109,8 +1110,8 @@ fn batch_get_across_families(#[case] mmap: bool) -> Result<()> { let path = tempdir.path(); let mut config = config_with_mmap(mmap); - // Set zstd on an arbitrary family; lz4 is used by default. - config.family_configs[2].compression = Compression::Zstd3; + // Set zstd on an arbitrary family; LZ4 is used by default. + config.family_configs[2].compression = Compression::Zstd3.into(); let db = open_db_with_config::<16>(path, config.clone())?; // Write compressible values to multiple families so every configured codec is exercised. @@ -1188,7 +1189,7 @@ fn batch_get_after_compaction(#[case] mmap: bool) -> Result<()> { let path = tempdir.path(); let mut config = config_with_mmap(mmap); - config.family_configs[0].compression = Compression::Zstd3; + config.family_configs[0].compression = Compression::Zstd3.into(); let db = open_db_with_config::<16>(path, config)?; // Write data across multiple batches to create multiple SST files @@ -1561,7 +1562,7 @@ fn multi_value_config() -> DbConfig<1> { config.family_configs[0] = FamilyConfig { name: "test", kind: FamilyKind::MultiValue, - compression: Compression::Lz4, + compression: Compression::Lz4.into(), }; config } @@ -2456,7 +2457,9 @@ fn count_tombstones( sequence_number: entry.sequence_number, block_count: entry.block_count, }; - for item in StaticSortedFileIter::open(path, sst, Compression::Lz4, AccessMode::Mmap)? { + for item in + StaticSortedFileIter::open(path, sst, Compression::Lz4.into(), AccessMode::Mmap)? + { if matches!( item?.value, IterValue::KeyDeleted | IterValue::KeyValueDeleted { .. } @@ -2812,3 +2815,41 @@ fn valued_tombstone_rejects_single_value_families() -> Result<()> { db.shutdown()?; Ok(()) } + +#[test] +fn dictionary_config_round_trips_and_rejects_mismatch() -> Result<()> { + let samples = (0..100) + .map(|index| format!("export default function Component{index}() {{ return null }}")) + .collect::>(); + let dictionary = zstd::dict::from_samples(&samples, 1024)?; + let dictionary = Box::leak(dictionary.into_boxed_slice()); + let tempdir = tempfile::tempdir()?; + let mut config = DbConfig::<1>::default(); + config.family_configs[0].compression = CompressionConfig::Zstd3WithDictionary(dictionary); + let db = TurboPersistence::::open_with_config( + tempdir.path().to_path_buf(), + config.clone(), + )?; + let batch = db.write_batch()?; + batch.put(0, b"key".to_vec(), samples.concat().into_bytes().into())?; + db.commit_write_batch(batch)?; + db.shutdown()?; + + let db = TurboPersistence::::open_with_config( + tempdir.path().to_path_buf(), + config, + )?; + assert!(db.get(0, &b"key".to_vec())?.is_some()); + db.shutdown()?; + + let mut plain_zstd = DbConfig::<1>::default(); + plain_zstd.family_configs[0].compression = CompressionConfig::Zstd3; + let error = TurboPersistence::::open_with_config( + tempdir.path().to_path_buf(), + plain_zstd, + ) + .err() + .expect("plain zstd config must reject dictionary metadata"); + assert!(format!("{error:#}").contains("Compression configuration mismatch")); + Ok(()) +} diff --git a/turbopack/crates/turbo-tasks-backend/README.md b/turbopack/crates/turbo-tasks-backend/README.md new file mode 100644 index 000000000000..7bb1e363e11c --- /dev/null +++ b/turbopack/crates/turbo-tasks-backend/README.md @@ -0,0 +1,31 @@ +# turbo-tasks-backend + +## Training a TaskData zstd dictionary + +TaskData is persistence family `2`. After producing copied Turbopack cache database directories, +train and evaluate a dictionary without rebuilding the applications: + +```sh +cargo run -p turbo-persistence --release --bin zstd_dictionary -- train \ + --family 2 --output taskdata.zdict \ + path/to/database-a path/to/database-b + +cargo run -p turbo-persistence --release --bin zstd_dictionary -- evaluate \ + --family 2 --dictionary taskdata.zdict --json report.json \ + path/to/holdout-database-a path/to/holdout-database-b + +# For caches produced after the dictionary was enabled: +cargo run -p turbo-persistence --release --bin zstd_dictionary -- evaluate \ + --family 2 --source-dictionary src/database/taskdata.zdict \ + --dictionary candidate.zdict path/to/database-a +``` + +Training and evaluation inputs should be disjoint. A dictionary evaluated against its own training +caches is useful only as a tool smoke test and overstates its real benefit. + +The checked-in baseline is produced by +[`scripts/train-taskdata-dictionary.sh`](./scripts/train-taskdata-dictionary.sh) from preserved +`test/production` filesystem caches. Its corpus and held-out receipts are recorded in +[`src/database/taskdata-dictionary.md`](./src/database/taskdata-dictionary.md). The script is +resumable; set `CORPUS_JOBS` for bounded parallelism, `CORPUS_FAMILY` to select a keyspace (defaults +to TaskData family 2), and pass an output directory plus dictionary path. diff --git a/turbopack/crates/turbo-tasks-backend/scripts/train-taskdata-dictionary.sh b/turbopack/crates/turbo-tasks-backend/scripts/train-taskdata-dictionary.sh new file mode 100755 index 000000000000..426dc82eba42 --- /dev/null +++ b/turbopack/crates/turbo-tasks-backend/scripts/train-taskdata-dictionary.sh @@ -0,0 +1,126 @@ +#!/usr/bin/env bash +set -euo pipefail + +repo_root=$(git rev-parse --show-toplevel) +output_root=${1:-/tmp/taskdata-dictionary-corpus} +dictionary=${2:-$repo_root/turbopack/crates/turbo-tasks-backend/src/database/taskdata.zdict} +shift $(( $# >= 2 ? 2 : $# )) +jobs=${CORPUS_JOBS:-4} +family=${CORPUS_FAMILY:-2} + +mkdir -p "$output_root"/{train,holdout,logs,tmp,reports,state} +manifest="$output_root/manifest.tsv" + +if (( $# )); then + tests=("$@") +else + mapfile -t tests < <(find "$repo_root/test/production" -name '*.test.ts' -print | sort) +fi +if [[ -n ${CORPUS_LIMIT:-} ]]; then + tests=("${tests[@]:0:$CORPUS_LIMIT}") +fi + +digest_for() { + printf '%s' "$1" | sha256sum | cut -d' ' -f1 +} + +# Migrate output from the earlier sequential collector once so interrupted runs can resume. +if [[ -f $manifest ]] && ! compgen -G "$output_root/state/*.tsv" >/dev/null; then + while IFS=$'\t' read -r test_path split status cache; do + [[ $test_path == test || -z $test_path ]] && continue + digest=$(digest_for "$test_path") + printf '%s\t%s\t%s\t%s\n' "$test_path" "$split" "$status" "$cache" \ + >> "$output_root/state/$digest.tsv" + done < "$manifest" +fi +rm -f "$manifest" + +run_case() { + local test_path=$1 relative digest split case_root log status cache_index current cache destination state_tmp + relative=${test_path#"$repo_root/"} + digest=$(digest_for "$relative") + [[ -f $output_root/state/$digest.tsv ]] && return 0 + if (( 16#${digest:0:2} % 5 == 0 )); then split=holdout; else split=train; fi + case_root="$output_root/tmp/$digest" + rm -rf "$case_root" + mkdir -p "$case_root" + log="$output_root/logs/$digest.log" + echo "[$split] $relative" + set +e + ( + cd "$repo_root" + TMPDIR="$case_root" NEXT_TEST_SKIP_CLEANUP=1 \ + pnpm test-start-turbo "$relative" + ) >"$log" 2>&1 + status=$? + set -e + cache_index=0 + state_tmp="$output_root/state/$digest.tmp" + : > "$state_tmp" + while IFS= read -r current; do + cache=$(dirname "$current") + compgen -G "$cache/*.meta" >/dev/null || continue + compgen -G "$cache/*.sst" >/dev/null || continue + destination="$output_root/$split/${digest}-${cache_index}" + rm -rf "$destination" + cp -a "$cache" "$destination" + printf '%s\t%s\t%s\t%s\n' "$relative" "$split" "$status" "$destination" >> "$state_tmp" + cache_index=$((cache_index + 1)) + done < <(find "$case_root" -type f -name CURRENT -path '*/.next/cache/turbopack/*' | sort) + if (( cache_index == 0 )); then + printf '%s\t%s\t%s\t\n' "$relative" "$split" "$status" >> "$state_tmp" + fi + mv "$state_tmp" "$output_root/state/$digest.tsv" + rm -rf "$case_root" +} + +for test_path in "${tests[@]}"; do + run_case "$test_path" & + while (( $(jobs -rp | wc -l) >= jobs )); do + wait -n || true + done +done +wait + +printf 'test\tsplit\tstatus\tcache\n' > "$manifest" +while IFS= read -r state; do + cat "$state" >> "$manifest" +done < <(find "$output_root/state" -name '*.tsv' | sort) + +mapfile -t train_caches < <( + for cache in "$output_root"/train/*; do + [[ -d $cache ]] || continue + compgen -G "$cache/*.meta" >/dev/null && compgen -G "$cache/*.sst" >/dev/null && printf '%s\n' "$cache" + done | sort +) +mapfile -t holdout_caches < <( + for cache in "$output_root"/holdout/*; do + [[ -d $cache ]] || continue + compgen -G "$cache/*.meta" >/dev/null && compgen -G "$cache/*.sst" >/dev/null && printf '%s\n' "$cache" + done | sort +) +if (( ${#train_caches[@]} == 0 || ${#holdout_caches[@]} == 0 )); then + echo "Need at least one train and one holdout cache; see $manifest" >&2 + exit 1 +fi + +cd "$repo_root" +source_args=() +source_dictionary=${SOURCE_DICTIONARY:-$dictionary} +if [[ -f $source_dictionary ]]; then + source_copy="$output_root/source-dictionary.zdict" + cp "$source_dictionary" "$source_copy" + source_args=(--source-dictionary "$source_copy") +fi +cargo run -p turbo-persistence --release --bin zstd_dictionary -- train \ + --family "$family" "${source_args[@]}" --output "$dictionary" "${train_caches[@]}" +for run in 1 2 3 4 5; do + cargo run -p turbo-persistence --release --bin zstd_dictionary -- evaluate \ + --family "$family" "${source_args[@]}" --dictionary "$dictionary" \ + --json "$output_root/reports/holdout-$run.json" \ + "${holdout_caches[@]}" +done + +echo "Dictionary: $dictionary" +echo "Manifest: $manifest" +echo "Reports: $output_root/reports" diff --git a/turbopack/crates/turbo-tasks-backend/src/database/key_value_database.rs b/turbopack/crates/turbo-tasks-backend/src/database/key_value_database.rs index 926374691b61..df88d0fd4bae 100644 --- a/turbopack/crates/turbo-tasks-backend/src/database/key_value_database.rs +++ b/turbopack/crates/turbo-tasks-backend/src/database/key_value_database.rs @@ -1,4 +1,4 @@ -use turbo_persistence::{Compression, FamilyConfig, FamilyKind}; +use turbo_persistence::{CompressionConfig, FamilyConfig, FamilyKind}; #[derive(Debug, Clone, Copy)] pub enum KeySpace { @@ -38,18 +38,20 @@ impl KeySpace { KeySpace::Infra | KeySpace::TaskMeta => FamilyConfig { name: self.name(), kind: FamilyKind::SingleValue, - compression: Compression::Lz4, + compression: CompressionConfig::Lz4, }, KeySpace::TaskData => FamilyConfig { name: self.name(), kind: FamilyKind::SingleValue, - compression: Compression::Zstd3, + compression: CompressionConfig::Zstd3WithDictionary(include_bytes!( + "taskdata.zdict" + )), }, KeySpace::TaskCache => FamilyConfig { name: self.name(), // TaskCache uses hash-based lookups with potential collisions. kind: FamilyKind::MultiValue, - compression: Compression::Lz4, + compression: CompressionConfig::Lz4, }, } } diff --git a/turbopack/crates/turbo-tasks-backend/src/database/taskdata-dictionary.md b/turbopack/crates/turbo-tasks-backend/src/database/taskdata-dictionary.md new file mode 100644 index 000000000000..67bf0dac08de --- /dev/null +++ b/turbopack/crates/turbo-tasks-backend/src/database/taskdata-dictionary.md @@ -0,0 +1,43 @@ +# TaskData zstd dictionary provenance + +The adjacent `taskdata.zdict` is a 64 KiB dictionary trained from filesystem caches produced by +Next.js `test/production` fixtures. + +## Corpus + +- Next.js revision: `45b39da5ba613553540c4d7fb9196e79a58bbe61` +- Split: SHA-256 of test path, 80% train / 20% holdout +- Completed test files: 194 of 362 attempted before the sandbox disk budget stopped collection +- Valid caches: 204 train, 38 holdout +- Training policy: round-robin logical TaskData values across caches to a 64 MiB target +- Dictionary size: 65,536 bytes +- Dictionary ID: `1166300072` +- Dictionary xxh3-64: `2d9e019e3c1071f1` + +The incomplete tail is a limitation of this baseline, but 242 independent caches provide broad +coverage. The checked-in script is resumable and can complete/refresh the corpus in a larger local +environment. + +## Held-out results + +Evaluation approximates production compression units: small values are accumulated into SST-local +8–12 KiB blocks, while medium values and blobs remain independent. The corrected evaluator was run +against 4 held-out caches from a fresh 20-test smoke corpus, containing 11,778 compression units and +176,698,226 uncompressed bytes. + +| Metric | zstd3 | Dictionary | Delta | +| --------------------------- | ---------: | ---------: | ----------: | +| Raw compressed bytes | 52,387,060 | 44,924,356 | **-14.25%** | +| Median encode time (5 runs) | 772.23 ms | 822.26 ms | **+6.48%** | +| Median decode time (5 runs) | 225.30 ms | 192.72 ms | **-14.46%** | + +The copied holdout cache directories occupied 77,370,864 bytes. The raw compressed-byte delta is +7,462,704 bytes, or 9.65% of that directory total; this is a directional total-cache estimate, not +an exact rewritten-cache measurement. + +The original 38-cache evaluation treated every logical value as a compression unit and overstated +the benefit, so those numbers are intentionally not retained here. The corrected result meets the +accepted ≥2% size / ≤10% encode-regression / no-decode-regression gate. + +Timing is machine-specific single-process diagnostic data. The stable byte result is the primary +receipt. diff --git a/turbopack/crates/turbo-tasks-backend/src/database/taskdata.zdict b/turbopack/crates/turbo-tasks-backend/src/database/taskdata.zdict new file mode 100644 index 000000000000..24e85081f33c Binary files /dev/null and b/turbopack/crates/turbo-tasks-backend/src/database/taskdata.zdict differ