RFC: bypass defrag prevention on mostly-matchable content

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<f32> {
@@ -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 {

xet_data/src/deduplication/file_deduplication.rs

@@ -310,6 +310,10 @@ impl<DataInterfaceType: DeduplicationDataInterface> FileDeduper<DataInterfaceTyp
     /// Add a new file data sequence entry to the current process, possibly merging with the
     /// previous entry.
     fn add_file_data_sequence_entry(&mut self, fse: FileDataSequenceEntry, n_deduped: usize) {
+        // Feed the matchable-density estimate (rejected offers are recorded by
+        // the tracker itself at the decision point).
+        self.defrag_tracker.record_matched_range(n_deduped);
+
         // Do we modify the previous entry as this is the next logical chunk, or do we
         // start a new entry?
         if self.file_data_sequence_continues_current(&fse) {

xet_data/src/processing/range_upload.rs

@@ -533,7 +533,7 @@ mod tests {
     use std::sync::Arc;
 
     use tempfile::TempDir;
-    use xet_client::cas_client::{Client, LocalTestServerBuilder};
+    use xet_client::cas_client::{Client, DirectAccessClient, LocalTestServerBuilder};
     use xet_core_structures::merklehash::MerkleHash;
 
     use super::*;
@@ -1447,6 +1447,17 @@ mod tests {
         MerkleHash::from_hex(xfi.hash()).unwrap()
     }
 
+    /// Like `upload_file`, but cleans inside an existing session (so several files
+    /// share xorb packing) and returns the file info plus dedup metrics.
+    async fn clean_file_in_session(
+        session: &Arc<FileUploadSession>,
+        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<TranslatorConfig>, hash: MerkleHash, size: u64) -> Vec<u8> {
         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<usize> = (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!(
@@ -2178,4 +2177,158 @@ 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() {
+        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<dyn Client> = 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 (xfi, metrics) = clean_file_in_session(&session, &new_file).await;
+        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() {
+        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<dyn Client> = 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 rng = DeterministicRng::new(0x9e3779b97f4a7c15);
+            let mut pieces: Vec<(usize, usize)> = Vec::new();
+            let mut pos = 0usize;
+            while pos < SIZE {
+                let len = rng.gen_range(piece / 4, piece / 4 + 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 };
+                clean_file_in_session(session, &data[start..end]).await;
+            }
+            session_a.finalize().await.unwrap();
+            session_b.finalize().await.unwrap();
+
+            println!("\n########## piece={}KB ##########", piece / 1024);
+
+            for generation in 1..=GENERATIONS {
+                // Distinct trailing bytes per generation: distinct file hash, content
+                // otherwise identical, so each generation dedups against the previous
+                // 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(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();
+
+                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<usize, usize> = 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");
+        }
+    }
 }

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.