From 054caa553cecd933651cd6252ef5cea3a32c0f24 Mon Sep 17 00:00:00 2001 From: Adrien Date: Fri, 3 Jul 2026 15:58:41 +0200 Subject: [PATCH 1/2] feat: bypass defrag prevention on mostly-matchable content Defrag prevention re-stores dedup ranges shorter than its rolling chunks-per-range average so they merge with neighboring new data into contiguous fresh runs. That reasoning only holds when the surrounding content is mostly new. On mostly-matchable content (a re-upload of a file whose bytes already sit in CAS with a fragmented layout), the throttle degenerates into its worst case: it alternates accepted and rejected ranges at the hysteresis equilibrium, so the file both keeps referencing the fragmented ranges AND re-stores about half the bytes. Observed on production buckets (multi-GB media files re-imported over an interleaved first upload): ~2x stored bytes, +90-160% xorb overhead, CPR pinned at exactly the min_n_chunks_per_range equilibrium (8.07 and 7.8 on two files), terms alternating gap-2 between the old xorbs and the re-stored ones. Track how many processed chunks had a match on offer (accepted or rejected) and bypass the throttle once that density exceeds deduplication.defrag_prevention_matchable_density_bypass (default 0.7, > 1.0 restores the previous behavior), after a 512-chunk warm-up. Re-uploads then dedup fully and inherit the reference layout instead of duplicating storage; low-density files (the case the throttle was built for) are untouched. Validated with the two ignored exploration harnesses added here: - fragmented-reference re-upload: defrag_prevented drops 5.9% -> 0, re-stored bytes drop to only the genuinely new chunks; - sparse scattered matches in mostly-new content: byte-identical behavior with and without the gate. --- .../src/deduplication/file_deduplication.rs | 36 ++++ xet_data/src/processing/range_upload.rs | 171 ++++++++++++++++++ .../src/config/groups/deduplication.rs | 12 ++ 3 files changed, 219 insertions(+) diff --git a/xet_data/src/deduplication/file_deduplication.rs b/xet_data/src/deduplication/file_deduplication.rs index d2dac78de..6be21add1 100644 --- a/xet_data/src/deduplication/file_deduplication.rs +++ b/xet_data/src/deduplication/file_deduplication.rs @@ -17,6 +17,10 @@ use super::interface::DeduplicationDataInterface; use super::{Chunk, RawXorbData}; use crate::progress_tracking::upload_tracking::FileXorbDependency; +/// The matchable-density bypass only engages once this many chunks have been +/// processed, so a file's first few MB can't flip it on off a tiny sample. +const MATCHABLE_DENSITY_WARMUP_CHUNKS: u64 = 512; + pub struct FileDeduper { #[cfg_attr(not(feature = "simulation"), allow(dead_code))] ctx: XetContext, @@ -47,6 +51,16 @@ pub struct FileDeduper { /// Tracking the defragmentation of the file specification. defrag_tracker: DefragPrevention, + /// Chunks processed so far, and how many of them had a dedup match on offer + /// (whether accepted or rejected by defrag prevention). Their ratio is the + /// matchable density used to bypass defrag prevention on re-uploaded content. + processed_chunks: u64, + offered_match_chunks: u64, + + /// Matchable-density threshold above which defrag prevention is bypassed + /// (`deduplication.defrag_prevention_matchable_density_bypass`). + matchable_density_bypass: f32, + /// The minimum number of chunks to wait for between generating global /// dedup queries. Can be changed by testing code. min_spacing_between_global_dedup_queries: usize, @@ -71,6 +85,9 @@ impl FileDeduper FileDeduper MATCHABLE_DENSITY_WARMUP_CHUNKS + && (self.offered_match_chunks as f64 / self.processed_chunks as f64) + > self.matchable_density_bypass as f64; + if self.file_data_sequence_continues_current(&fse) + || matchable_density_high || self.defrag_tracker.allow_dedup_on_next_range(n_deduped) { // Only count the range as deduped once it passes the defrag gate; @@ -210,6 +240,8 @@ impl FileDeduper FileDeduper FileDeduper xorb_cut_bytes || self.new_data.len() + 1 > xorb_cut_chunks { diff --git a/xet_data/src/processing/range_upload.rs b/xet_data/src/processing/range_upload.rs index e774c2d68..4b73d855a 100644 --- a/xet_data/src/processing/range_upload.rs +++ b/xet_data/src/processing/range_upload.rs @@ -2178,4 +2178,175 @@ mod tests { "every chunk is either deduped or new" ); } + + // Exploration (not for CI): the case defrag prevention was built for — a mostly-new + // file with small scattered matches against existing content. The matchable-density + // bypass must stay OFF here (density ~50% < threshold) and the throttle must still + // re-store the tiny bites into contiguous fresh runs. + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + #[ignore = "exploration harness, run manually"] + async fn explore_sparse_match_protection() { + use xet_client::cas_client::DirectAccessClient; + + let server = LocalTestServerBuilder::new().start().await; + let server_endpoint = server.http_endpoint().to_string(); + server.set_global_dedup_shard_expiration(Some(std::time::Duration::from_secs(3600))); + let base_dir = TempDir::new().unwrap(); + let config = test_config(&server_endpoint, base_dir.path()); + let cas_client: Arc = Arc::new(server); + + const SIZE: usize = 128 * 1024 * 1024; + let mut rng = DeterministicRng::new(42); + let old_content = rng.gen_bytes(SIZE); + upload_file(&config, &old_content).await; + + // New file: alternating ~1MB fresh / ~512KB copied from the old content. CDC + // boundaries resync a chunk into each copied block, so the matches on offer are + // interior runs of ~5-7 chunks: short, scattered, ~30% density — the exact case + // the defrag throttle exists for, below the bypass threshold. + let mut new_file = Vec::with_capacity(SIZE); + let mut old_pos = 0usize; + while new_file.len() < SIZE { + new_file.extend(rng.gen_bytes(1024 * 1024)); + let take = (512 * 1024).min(old_content.len() - old_pos); + new_file.extend_from_slice(&old_content[old_pos..old_pos + take]); + old_pos = (old_pos + take) % (old_content.len() - 512 * 1024); + } + new_file.truncate(SIZE); + + // Same cache dir as the first upload: scattered short matches are invisible to + // the sampled global dedup queries, so exercise the throttle via local shards — + // the same-client-updates-a-file case defrag prevention was designed around. + let session = FileUploadSession::new(config.clone()).await.unwrap(); + let (_id, mut cleaner) = session + .start_clean(Some("mixed".into()), Some(new_file.len() as u64), Sha256Policy::Skip) + .unwrap(); + cleaner.add_data(&new_file).await.unwrap(); + let (xfi, metrics) = cleaner.finish().await.unwrap(); + session.finalize().await.unwrap(); + + let hash = MerkleHash::from_hex(xfi.hash()).unwrap(); + let (mdb, _) = cas_client.get_file_reconstruction_info(&hash).await.unwrap().unwrap(); + let distinct: std::collections::HashSet<_> = mdb.segments.iter().map(|s| s.xorb_hash).collect(); + println!( + "sparse-match file: {} terms, {} xorbs, deduped {:.1}%, new {:.1}%, defrag_prevented {:.1}%", + mdb.segments.len(), + distinct.len(), + metrics.deduped_bytes as f64 / metrics.total_bytes as f64 * 100.0, + metrics.new_bytes as f64 / metrics.total_bytes as f64 * 100.0, + metrics.defrag_prevented_dedup_bytes as f64 / metrics.total_bytes as f64 * 100.0, + ); + } + + // Exploration (not for CI): does repeated re-upload of the same content converge to + // the production signature (gap-2 A/B terms, CPR ~8, ~50% re-stored)? Each + // generation re-uploads the assembled file from a fresh cache dir, deduping against + // whatever the previous generations left in CAS — mimicking an import retry loop. + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + #[ignore = "exploration harness, run manually"] + async fn explore_generational_fragmentation() { + use xet_client::cas_client::DirectAccessClient; + + const SIZE: usize = 256 * 1024 * 1024; + const GENERATIONS: usize = 5; + // Mean piece size of the very first (fragmented) upload, in bytes. Actual piece + // sizes are drawn from [mean/4, 7*mean/4] — the variance is what arms defrag + // prevention: it only rejects ranges shorter than its rolling average. + for piece in [512 * 1024usize, 1024 * 1024] { + let server = LocalTestServerBuilder::new().start().await; + let server_endpoint = server.http_endpoint().to_string(); + server.set_global_dedup_shard_expiration(Some(std::time::Duration::from_secs(3600))); + let base_dir = TempDir::new().unwrap(); + let config = test_config(&server_endpoint, base_dir.path()); + let cas_client: Arc = Arc::new(server); + + // NOT random_data(): that helper's byte sequence has a 16 MB period (bits + // 16..24 of i*K mod 2^24), so large buffers dedup against themselves and + // poison layout measurements. LCG-per-byte has full 2^64 period. + let data = DeterministicRng::new(20260703).gen_bytes(SIZE); + + // Gen 0: two-stream interleaved upload — pieces in FILE ORDER, alternating + // between two sessions, as two concurrent upload workers would pack them. + // Session 1's xorbs hold file chunks [0,P),[2P,3P),...; session 2's hold + // [P,2P),[3P,4P),... — the A/B seed. + let mut lcg: u64 = 0x9e3779b97f4a7c15; + let mut next_random = || { + lcg = lcg.wrapping_mul(6364136223846793005).wrapping_add(1442695040888963407); + (lcg >> 33) as usize + }; + let mut pieces: Vec<(usize, usize)> = Vec::new(); + let mut pos = 0usize; + while pos < SIZE { + let len = (piece / 4 + next_random() % (piece * 3 / 2)).min(SIZE - pos); + pieces.push((pos, pos + len)); + pos += len; + } + let session_a = FileUploadSession::new(config.clone()).await.unwrap(); + let session_b = FileUploadSession::new(config.clone()).await.unwrap(); + for (i, &(start, end)) in pieces.iter().enumerate() { + let session = if i % 2 == 0 { &session_a } else { &session_b }; + let slice = &data[start..end]; + let (_id, mut cleaner) = + session.start_clean(None, Some(slice.len() as u64), Sha256Policy::Skip).unwrap(); + cleaner.add_data(slice).await.unwrap(); + cleaner.finish().await.unwrap(); + } + session_a.finalize().await.unwrap(); + session_b.finalize().await.unwrap(); + + println!("\n########## piece={}KB ##########", piece / 1024); + + for generation in 1..=GENERATIONS { + // Distinct trailing byte per generation: distinct file hash, content + // otherwise identical, so each generation dedups against the previous + // ones' layouts. + let mut gen_data = data.clone(); + gen_data.extend(std::iter::repeat(generation as u8).take(generation)); + + let gen_dir = TempDir::new().unwrap(); + let gen_config = test_config(&server_endpoint, gen_dir.path()); + let gen_session = FileUploadSession::new(gen_config).await.unwrap(); + let (_id, mut cleaner) = gen_session + .start_clean(Some("copy".into()), Some(gen_data.len() as u64), Sha256Policy::Skip) + .unwrap(); + cleaner.add_data(&gen_data).await.unwrap(); + let (xfi, metrics) = cleaner.finish().await.unwrap(); + gen_session.finalize().await.unwrap(); + let hash = MerkleHash::from_hex(xfi.hash()).unwrap(); + + let (mdb, _) = cas_client.get_file_reconstruction_info(&hash).await.unwrap().unwrap(); + let n_segments = mdb.segments.len(); + let distinct: std::collections::HashSet<_> = mdb.segments.iter().map(|s| s.xorb_hash).collect(); + let total_chunks: u64 = mdb + .segments + .iter() + .map(|s| (s.chunk_index_end - s.chunk_index_start) as u64) + .sum(); + let cpr = total_chunks as f64 / n_segments as f64; + + let mut last_seen: std::collections::HashMap<_, usize> = std::collections::HashMap::new(); + let mut gap_counts: std::collections::HashMap = std::collections::HashMap::new(); + for (i, s) in mdb.segments.iter().enumerate() { + if let Some(prev) = last_seen.insert(s.xorb_hash, i) { + *gap_counts.entry(i - prev).or_default() += 1; + } + } + let total_gaps: usize = gap_counts.values().sum(); + let gap2_share = *gap_counts.get(&2).unwrap_or(&0) as f64 / total_gaps.max(1) as f64; + let gap1_share = *gap_counts.get(&1).unwrap_or(&0) as f64 / total_gaps.max(1) as f64; + + println!( + "gen {generation}: {n_segments} terms, {} xorbs (ideal {}), CPR {cpr:.2}, \ + gap1 {:.1}%, gap2 {:.1}%, new {:.1}%, defrag_prevented {:.1}%", + distinct.len(), + SIZE.div_ceil(64 * 1024 * 1024), + gap1_share * 100.0, + gap2_share * 100.0, + metrics.new_bytes as f64 / metrics.total_bytes as f64 * 100.0, + metrics.defrag_prevented_dedup_bytes as f64 / metrics.total_bytes as f64 * 100.0, + ); + } + println!("prod target: CPR ~7.8-8.1, gap2 ~93-96%, xorbs ~1.9-2.6x ideal"); + } + } } diff --git a/xet_runtime/src/config/groups/deduplication.rs b/xet_runtime/src/config/groups/deduplication.rs index c7528c192..7db7be152 100644 --- a/xet_runtime/src/config/groups/deduplication.rs +++ b/xet_runtime/src/config/groups/deduplication.rs @@ -24,6 +24,18 @@ crate::config_group!({ /// Use the environment variable `HF_XET_DEDUPLICATION_MIN_N_CHUNKS_PER_RANGE` to set this value. ref min_n_chunks_per_range: f32 = 8.0; + /// Share of processed chunks with a dedup match on offer above which defrag + /// prevention is bypassed. Skipping dedup only reduces fragmentation when the + /// re-stored chunks merge with neighboring runs of new data; on mostly-matchable + /// content (a re-upload of something already stored), skipping buys no contiguity + /// and only duplicates storage. Set to a value > 1.0 to never bypass (previous + /// behavior). + /// + /// The default value is 0.7. + /// + /// Use the environment variable `HF_XET_DEDUPLICATION_DEFRAG_PREVENTION_MATCHABLE_DENSITY_BYPASS` to set this value. + ref defrag_prevention_matchable_density_bypass: f32 = 0.7; + /// Whether to enable global deduplication queries to the server. /// When enabled, the system will query the server for deduplication shards /// based on chunk hashes to enable cross-repository deduplication. From aac275081860a88337fdee44d695f1969dffff4d Mon Sep 17 00:00:00 2001 From: Adrien Date: Fri, 3 Jul 2026 16:55:50 +0200 Subject: [PATCH 2/2] refactor: move matchable-density bypass into DefragPrevention Review feedback on the density bypass: - The bypass is throttle policy, so its counters, threshold, warm-up constant and decision now live in DefragPrevention next to the rolling CPR estimator, instead of being bolted onto FileDeduper. The gate in process_chunks is back to its original shape. - The denominator piggybacks on the estimator feed methods (every settled chunk already flows through add_range/increment_last_range), removing the hand-maintained per-branch counting in the dedup loop; the accept-path numerator is fed from add_file_data_sequence_entry and the rejected-offer count is recorded at the decision itself. - The density check now runs only when the throttle is actually consulted (short-circuited by the contiguous-continuation fast path) and uses a multiply instead of a division. - Bypassing explicitly leaves the hysteresis state untouched, documented at the decision site. Test cleanups from the same review: hoist the DirectAccessClient import, replace two hand-rolled LCGs with the module's DeterministicRng, factor the repeated start_clean/add_data/finish ceremony into a clean_file_in_session helper, and feed the per-generation trailer as a second add_data call instead of cloning the 256MB buffer. --- .../src/deduplication/defrag_prevention.rs | 51 +++++++++++++ .../src/deduplication/file_deduplication.rs | 40 ++--------- xet_data/src/processing/range_upload.rs | 72 +++++++------------ 3 files changed, 82 insertions(+), 81 deletions(-) diff --git a/xet_data/src/deduplication/defrag_prevention.rs b/xet_data/src/deduplication/defrag_prevention.rs index ed8a6ae0c..221eeaa8b 100644 --- a/xet_data/src/deduplication/defrag_prevention.rs +++ b/xet_data/src/deduplication/defrag_prevention.rs @@ -2,6 +2,10 @@ use std::collections::VecDeque; use xet_runtime::core::XetContext; +/// The matchable-density bypass only engages once this many chunks have been +/// processed, so a file's first few MB can't flip it on off a tiny sample. +const MATCHABLE_DENSITY_WARMUP_CHUNKS: u64 = 512; + pub(crate) struct DefragPrevention { nranges_in_streaming_fragmentation_estimator: usize, @@ -21,6 +25,17 @@ pub(crate) struct DefragPrevention { /// The minimum number of chunks per range to consider deduplication. min_chunks_per_range_historesis_factor: f32, + + /// Chunks settled so far, and how many of them had a dedup match on offer + /// (whether the throttle accepted or rejected it). Their ratio is the + /// matchable density: above `matchable_density_bypass`, the throttle yields + /// (see `allow_dedup_on_next_range`). + processed_chunks: u64, + offered_match_chunks: u64, + + /// Matchable-density threshold above which the throttle is bypassed + /// (`deduplication.defrag_prevention_matchable_density_bypass`). + matchable_density_bypass: f64, } impl DefragPrevention { @@ -33,22 +48,35 @@ impl DefragPrevention { defrag_at_low_threshold: true, min_chunks_per_range: d.min_n_chunks_per_range, min_chunks_per_range_historesis_factor: d.min_n_chunks_per_range_hysteresis_factor, + processed_chunks: 0, + offered_match_chunks: 0, + matchable_density_bypass: d.defrag_prevention_matchable_density_bypass as f64, } } pub(crate) fn increment_last_range_in_fragmentation_estimate(&mut self, nchunks: usize) { + self.processed_chunks += nchunks as u64; if let Some(back) = self.rolling_last_nranges.back_mut() { *back += nchunks; self.rolling_nranges_chunks += nchunks; } } pub(crate) fn add_range_to_fragmentation_estimate(&mut self, nchunks: usize) { + self.processed_chunks += nchunks as u64; self.rolling_last_nranges.push_back(nchunks); self.rolling_nranges_chunks += nchunks; if self.rolling_last_nranges.len() > self.nranges_in_streaming_fragmentation_estimator { self.rolling_nranges_chunks -= self.rolling_last_nranges.pop_front().unwrap(); } } + + /// Record that a range of `nchunks` was settled as a dedup reference. The + /// range still flows through the fragmentation estimate via the range + /// methods above; this only feeds the matchable-density numerator. + pub(crate) fn record_matched_range(&mut self, nchunks: usize) { + self.offered_match_chunks += nchunks as u64; + } + /// Returns the average number of chunks per range /// None if there is is not enough data for an estimate pub(crate) fn rolling_chunks_per_range(&self) -> Option { @@ -59,8 +87,26 @@ impl DefragPrevention { } } + /// Share of settled chunks that had a dedup match on offer is high enough + /// that skipping dedup cannot pay off: the re-stored chunks would sit + /// between other matchable ranges instead of merging with runs of new + /// data, duplicating storage with no contiguity gain. + fn matchable_density_high(&self) -> bool { + self.processed_chunks > MATCHABLE_DENSITY_WARMUP_CHUNKS + && self.offered_match_chunks as f64 > self.matchable_density_bypass * self.processed_chunks as f64 + } + /// Check to see if we should update against this entry or continue from the previous one? pub(crate) fn allow_dedup_on_next_range(&mut self, dedup_range_size: usize) -> bool { + // On mostly-matchable content (a re-upload of something already stored) + // the throttle degenerates into alternating accepted/rejected ranges: + // storage is duplicated while the layout stays fragmented. Bypass it, + // leaving the hysteresis state untouched so the throttle resumes from + // where it was if density falls back below the threshold. + if self.matchable_density_high() { + return true; + } + let Some(chunks_per_range) = self.rolling_chunks_per_range() else { return true; }; @@ -81,6 +127,11 @@ impl DefragPrevention { // once I start skipping dedupe, we try to raise // the cpr to the high threshold self.defrag_at_low_threshold = false; + // The rejected offer settles exactly one chunk as new data (the + // caller re-offers the rest of the range next iteration); count + // it here, at the decision, so the density keeps seeing content + // as matchable while the throttle is re-storing it. + self.offered_match_chunks += 1; return false; } } else { diff --git a/xet_data/src/deduplication/file_deduplication.rs b/xet_data/src/deduplication/file_deduplication.rs index 6be21add1..caefea5fd 100644 --- a/xet_data/src/deduplication/file_deduplication.rs +++ b/xet_data/src/deduplication/file_deduplication.rs @@ -17,10 +17,6 @@ use super::interface::DeduplicationDataInterface; use super::{Chunk, RawXorbData}; use crate::progress_tracking::upload_tracking::FileXorbDependency; -/// The matchable-density bypass only engages once this many chunks have been -/// processed, so a file's first few MB can't flip it on off a tiny sample. -const MATCHABLE_DENSITY_WARMUP_CHUNKS: u64 = 512; - pub struct FileDeduper { #[cfg_attr(not(feature = "simulation"), allow(dead_code))] ctx: XetContext, @@ -51,16 +47,6 @@ pub struct FileDeduper { /// Tracking the defragmentation of the file specification. defrag_tracker: DefragPrevention, - /// Chunks processed so far, and how many of them had a dedup match on offer - /// (whether accepted or rejected by defrag prevention). Their ratio is the - /// matchable density used to bypass defrag prevention on re-uploaded content. - processed_chunks: u64, - offered_match_chunks: u64, - - /// Matchable-density threshold above which defrag prevention is bypassed - /// (`deduplication.defrag_prevention_matchable_density_bypass`). - matchable_density_bypass: f32, - /// The minimum number of chunks to wait for between generating global /// dedup queries. Can be changed by testing code. min_spacing_between_global_dedup_queries: usize, @@ -85,9 +71,6 @@ impl FileDeduper FileDeduper MATCHABLE_DENSITY_WARMUP_CHUNKS - && (self.offered_match_chunks as f64 / self.processed_chunks as f64) - > self.matchable_density_bypass as f64; - if self.file_data_sequence_continues_current(&fse) - || matchable_density_high || self.defrag_tracker.allow_dedup_on_next_range(n_deduped) { // Only count the range as deduped once it passes the defrag gate; @@ -240,8 +210,6 @@ impl FileDeduper FileDeduper FileDeduper xorb_cut_bytes || self.new_data.len() + 1 > xorb_cut_chunks { @@ -346,6 +310,10 @@ impl FileDeduper, + data: &[u8], + ) -> (XetFileInfo, crate::deduplication::DeduplicationMetrics) { + let (_id, mut cleaner) = session.start_clean(None, Some(data.len() as u64), Sha256Policy::Skip).unwrap(); + cleaner.add_data(data).await.unwrap(); + cleaner.finish().await.unwrap() + } + async fn download_file(config: &Arc, hash: MerkleHash, size: u64) -> Vec { let session = FileDownloadSession::new(config.clone(), None).await.unwrap(); let xfi = crate::processing::XetFileInfo::new(hash.hex(), size); @@ -2115,8 +2126,6 @@ mod tests { // the CPR tracker starts rejecting ranges. #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn test_dedup_metrics_when_defrag_prevention_triggers() { - use xet_client::cas_client::DirectAccessClient; - let server = LocalTestServerBuilder::new().start().await; let server_endpoint = server.http_endpoint().to_string(); // Serve global-dedup shards in the production format (file data stripped, expiry @@ -2131,21 +2140,15 @@ mod tests { // Pass 1: upload the content as shuffled pieces so xorbs pack them in arrival // order, not file order. + let mut rng = DeterministicRng::new(0x9e3779b97f4a7c15); let n_pieces = SIZE / PIECE; let mut order: Vec = (0..n_pieces).collect(); - let mut lcg: u64 = 0x9e3779b97f4a7c15; for i in (1..n_pieces).rev() { - lcg = lcg.wrapping_mul(6364136223846793005).wrapping_add(1442695040888963407); - order.swap(i, (lcg >> 33) as usize % (i + 1)); + order.swap(i, rng.gen_range(0, i + 1)); } let session1 = FileUploadSession::new(config.clone()).await.unwrap(); for &piece_idx in &order { - let slice = &data[piece_idx * PIECE..(piece_idx + 1) * PIECE]; - let (_id, mut cleaner) = session1 - .start_clean(None, Some(slice.len() as u64), Sha256Policy::Skip) - .unwrap(); - cleaner.add_data(slice).await.unwrap(); - cleaner.finish().await.unwrap(); + clean_file_in_session(&session1, &data[piece_idx * PIECE..(piece_idx + 1) * PIECE]).await; } session1.finalize().await.unwrap(); @@ -2154,11 +2157,7 @@ mod tests { let base_dir2 = TempDir::new().unwrap(); let config2 = test_config(&server_endpoint, base_dir2.path()); let session2 = FileUploadSession::new(config2).await.unwrap(); - let (_id, mut cleaner) = session2 - .start_clean(Some("copy".into()), Some(data.len() as u64), Sha256Policy::Skip) - .unwrap(); - cleaner.add_data(&data).await.unwrap(); - let (xfi, metrics) = cleaner.finish().await.unwrap(); + let (xfi, metrics) = clean_file_in_session(&session2, &data).await; session2.finalize().await.unwrap(); assert!( @@ -2186,8 +2185,6 @@ mod tests { #[tokio::test(flavor = "multi_thread", worker_threads = 4)] #[ignore = "exploration harness, run manually"] async fn explore_sparse_match_protection() { - use xet_client::cas_client::DirectAccessClient; - let server = LocalTestServerBuilder::new().start().await; let server_endpoint = server.http_endpoint().to_string(); server.set_global_dedup_shard_expiration(Some(std::time::Duration::from_secs(3600))); @@ -2218,11 +2215,7 @@ mod tests { // the sampled global dedup queries, so exercise the throttle via local shards — // the same-client-updates-a-file case defrag prevention was designed around. let session = FileUploadSession::new(config.clone()).await.unwrap(); - let (_id, mut cleaner) = session - .start_clean(Some("mixed".into()), Some(new_file.len() as u64), Sha256Policy::Skip) - .unwrap(); - cleaner.add_data(&new_file).await.unwrap(); - let (xfi, metrics) = cleaner.finish().await.unwrap(); + let (xfi, metrics) = clean_file_in_session(&session, &new_file).await; session.finalize().await.unwrap(); let hash = MerkleHash::from_hex(xfi.hash()).unwrap(); @@ -2245,8 +2238,6 @@ mod tests { #[tokio::test(flavor = "multi_thread", worker_threads = 4)] #[ignore = "exploration harness, run manually"] async fn explore_generational_fragmentation() { - use xet_client::cas_client::DirectAccessClient; - const SIZE: usize = 256 * 1024 * 1024; const GENERATIONS: usize = 5; // Mean piece size of the very first (fragmented) upload, in bytes. Actual piece @@ -2269,15 +2260,11 @@ mod tests { // between two sessions, as two concurrent upload workers would pack them. // Session 1's xorbs hold file chunks [0,P),[2P,3P),...; session 2's hold // [P,2P),[3P,4P),... — the A/B seed. - let mut lcg: u64 = 0x9e3779b97f4a7c15; - let mut next_random = || { - lcg = lcg.wrapping_mul(6364136223846793005).wrapping_add(1442695040888963407); - (lcg >> 33) as usize - }; + let mut rng = DeterministicRng::new(0x9e3779b97f4a7c15); let mut pieces: Vec<(usize, usize)> = Vec::new(); let mut pos = 0usize; while pos < SIZE { - let len = (piece / 4 + next_random() % (piece * 3 / 2)).min(SIZE - pos); + let len = rng.gen_range(piece / 4, piece / 4 + piece * 3 / 2).min(SIZE - pos); pieces.push((pos, pos + len)); pos += len; } @@ -2285,11 +2272,7 @@ mod tests { let session_b = FileUploadSession::new(config.clone()).await.unwrap(); for (i, &(start, end)) in pieces.iter().enumerate() { let session = if i % 2 == 0 { &session_a } else { &session_b }; - let slice = &data[start..end]; - let (_id, mut cleaner) = - session.start_clean(None, Some(slice.len() as u64), Sha256Policy::Skip).unwrap(); - cleaner.add_data(slice).await.unwrap(); - cleaner.finish().await.unwrap(); + clean_file_in_session(session, &data[start..end]).await; } session_a.finalize().await.unwrap(); session_b.finalize().await.unwrap(); @@ -2297,19 +2280,18 @@ mod tests { println!("\n########## piece={}KB ##########", piece / 1024); for generation in 1..=GENERATIONS { - // Distinct trailing byte per generation: distinct file hash, content + // Distinct trailing bytes per generation: distinct file hash, content // otherwise identical, so each generation dedups against the previous - // ones' layouts. - let mut gen_data = data.clone(); - gen_data.extend(std::iter::repeat(generation as u8).take(generation)); + // ones' layouts. Fed as two add_data calls to avoid cloning 256MB. + let trailer = vec![generation as u8; generation]; + let total_len = (SIZE + trailer.len()) as u64; let gen_dir = TempDir::new().unwrap(); let gen_config = test_config(&server_endpoint, gen_dir.path()); let gen_session = FileUploadSession::new(gen_config).await.unwrap(); - let (_id, mut cleaner) = gen_session - .start_clean(Some("copy".into()), Some(gen_data.len() as u64), Sha256Policy::Skip) - .unwrap(); - cleaner.add_data(&gen_data).await.unwrap(); + let (_id, mut cleaner) = gen_session.start_clean(None, Some(total_len), Sha256Policy::Skip).unwrap(); + cleaner.add_data(&data).await.unwrap(); + cleaner.add_data(&trailer).await.unwrap(); let (xfi, metrics) = cleaner.finish().await.unwrap(); gen_session.finalize().await.unwrap(); let hash = MerkleHash::from_hex(xfi.hash()).unwrap();