hexz_core/algo/dedup/

cdc.rs

//! Content-Defined Chunking (CDC) for deduplication analysis.
//!
//! This module implements the FastCDC (Fast Content-Defined Chunking) algorithm,
//! a state-of-the-art variable-length chunking algorithm that splits data streams
//! into content-defined chunks for efficient deduplication. Unlike fixed-size
//! chunking, CDC adapts chunk boundaries to data content, enabling detection of
//! duplicate blocks even when shifted or surrounded by different data.
//!
//! # Why Content-Defined Chunking?
//!
//! Traditional fixed-size chunking fails to handle the "boundary shift problem":
//! inserting or deleting even a single byte causes all subsequent chunks to shift,
//! preventing duplicate detection. CDC solves this by:
//!
//! - **Content-based boundaries**: Chunk cuts occur at data-dependent positions
//! - **Shift resilience**: Boundaries remain stable despite insertions/deletions
//! - **Variable chunk sizes**: Adapts to data structure for optimal deduplication
//!
//! Example: Inserting 100 bytes at the start of a file with fixed 4KB chunks
//! invalidates every chunk. With CDC, only the chunks containing the insertion
//! are affected; chunks after the insertion point remain identical and deduplicatable.
//!
//! # FastCDC Algorithm Details
//!
//! FastCDC improves upon earlier CDC algorithms (Rabin fingerprinting, basic CDC)
//! by using a normalized chunking approach that reduces chunk size variance while
//! maintaining content-sensitivity.
//!
//! ## Rolling Hash Mechanism
//!
//! 1. **Sliding Window**: Maintains a window of `w` bytes (typically 48 bytes)
//! 2. **Hash Computation**: Computes a rolling hash (Gear hash) over the window
//! 3. **Cut-Point Detection**: When `hash & mask == 0`, mark a chunk boundary
//! 4. **Normalization**: Uses different masks in different chunk regions to
//!    achieve more uniform chunk size distribution
//!
//! ## Gear Hash Function
//!
//! FastCDC uses the Gear hash, a simple rolling hash that provides:
//! - **O(1) update time** per byte (critical for performance)
//! - **Good avalanche properties** (small changes → large hash changes)
//! - **CPU-friendly operations** (no expensive modulo or division)
//!
//! The Gear hash works by:
//! 1. Maintaining a precomputed random 256-entry lookup table
//! 2. For each byte `b`, XOR the current hash with `table[b]`
//! 3. Shift the hash left or right (depending on variant)
//! 4. Check if the hash satisfies the cut-point condition
//!
//! **Pseudocode:**
//! ```text
//! hash = 0
//! for each byte b in window:
//!     hash = (hash << 1) ^ GEAR_TABLE[b]
//!     if (hash & mask) == 0:
//!         mark_chunk_boundary()
//! ```
//!
//! The table lookup randomizes each byte's contribution, ensuring that
//! different byte patterns produce uncorrelated hashes. This is critical
//! for content-defined chunking to work correctly across diverse data types.
//!
//! ## Chunk Size Control
//!
//! Three parameters control chunk size distribution:
//!
//! - **`m` (min)**: Minimum chunk size (prevents tiny chunks, reduces overhead)
//! - **`f` (bits)**: Fingerprint bits; average chunk size is `2^f` bytes
//! - **`z` (max)**: Maximum chunk size (bounds worst-case memory usage)
//!
//! The algorithm enforces these constraints strictly:
//! - Never cuts before `m` bytes
//! - Always cuts at `z` bytes regardless of hash value
//! - Statistically averages to `2^f` byte chunks
//!
//! ## Normalization Zones
//!
//! FastCDC divides each chunk into regions with different cut-point masks
//! to achieve a more uniform chunk size distribution than basic CDC:
//!
//! - **[0, m)**: No cutting allowed (enforces minimum size)
//! - **[m, avg)**: Relaxed mask (fewer bits required → higher cut probability)
//! - **[avg, z)**: Standard mask (normal cut-point detection)
//! - **[z, ∞)**: Force cut at maximum (enforces size bound)
//!
//! ### Mask Selection
//!
//! The masks are computed based on the number of trailing zero bits required
//! in the hash value:
//!
//! - **Standard mask** (`[avg, z)`): Requires `f` trailing zero bits
//!   - Example: `f=14` → mask = `0x3FFF` → `hash & 0x3FFF == 0`
//!   - Cut probability: `1 / 2^14 ≈ 0.006%` per byte
//!   - Expected spacing: `2^14 = 16,384` bytes
//!
//! - **Relaxed mask** (`[m, avg)`): Requires `f-1` or `f-2` trailing zeros
//!   - Example: `f=14` → relaxed uses `f-2=12` → mask = `0xFFF`
//!   - Cut probability: `1 / 2^12 ≈ 0.024%` per byte (4× higher)
//!   - Encourages cuts closer to average, reducing long tails
//!
//! ### Effect on Distribution
//!
//! This normalization reduces the coefficient of variation (std dev / mean) in
//! chunk sizes from ~1.0 (pure exponential) to ~0.6 (normalized exponential):
//!
//! | Distribution Type | Mean | Std Dev | CV |
//! |-------------------|------|---------|-----|
//! | Fixed-size | 16 KB | 0 KB | 0.0 |
//! | Basic CDC | 16 KB | 16 KB | 1.0 |
//! | FastCDC (normalized) | 16 KB | 10 KB | 0.6 |
//!
//! **Benefits:**
//! - More consistent chunk sizes → better index density
//! - Fewer outliers (very small/large chunks) → lower metadata overhead
//! - Maintains content-sensitivity for deduplication effectiveness
//!
//! # Performance Characteristics
//!
//! ## Throughput
//!
//! - **Single-threaded**: 2.7 GB/s on modern CPUs (i7-14700K, validated via `cargo bench --bench cdc_chunking`)
//! - **Bottleneck**: Rolling hash computation (CPU-bound, not I/O-bound)
//! - **Comparison**: 4-5x faster than Rabin fingerprinting, 2x faster than basic CDC
//! - **Overhead**: ~10× slower than fixed-size chunking (26 GB/s vs 2.7 GB/s), but acceptable for dedup benefits
//!
//! ## Parallelization
//!
//! FastCDC is fundamentally sequential (each byte depends on previous hash state),
//! but can be parallelized across independent data streams:
//!
//! - **Multiple files**: Chunk each file in a separate thread/task
//! - **Split input**: Pre-split large files at fixed boundaries (e.g., every 1GB)
//!   and chunk each segment independently (loses some cross-boundary dedup)
//! - **Pipeline parallelism**: Chunk in one thread, compress in another, write
//!   in a third (producer-consumer pattern)
//!
//! **Note**: Parallel chunking of a single stream is possible using recursive
//! chunking or sketch-based techniques, but adds significant complexity and is
//! not implemented in this module
//!
//! ## Memory Usage
//!
//! - **Per-stream buffer**: `2 * z` bytes (default: ~128 KB for z=64KB)
//! - **Analysis hash set**: ~8 bytes per unique chunk (amortized)
//! - **Streaming overhead**: Negligible (no hash set)
//!
//! ## Chunk Size Distribution
//!
//! For typical data with `f=14` (16KB average, default):
//! - **Mean**: ~16 KB (exactly 2^f)
//! - **Median**: ~11 KB (skewed by exponential tail)
//! - **Standard deviation**: ~10 KB (coefficient of variation ~0.6)
//! - **Range**: [m, z] (typically [2KB, 64KB])
//!
//! # Usage Patterns
//!
//! This module provides two primary APIs:
//!
//! ## 1. Analysis Mode: `analyze_stream()`
//!
//! Performs a dry-run to estimate deduplication potential without storing chunks.
//! Used by the DCAM model to optimize parameters before actual packing.
//!
//! - **Input**: Reader + parameters
//! - **Output**: `CdcStats` (chunk counts, dedup ratio)
//! - **Memory**: O(unique chunks) for hash set
//! - **Use case**: Parameter optimization, capacity planning
//!
//! ## 2. Streaming Mode: `StreamChunker`
//!
//! Iterator that yields chunks on-demand during snapshot creation. Integrates
//! with the packing pipeline for compression and encryption.
//!
//! - **Input**: Reader + parameters
//! - **Output**: Iterator of `Vec<u8>` chunks
//! - **Memory**: O(1) - fixed buffer size
//! - **Use case**: Actual snapshot packing
//!
//! # Examples
//!
//! ## Basic Analysis
//!
//! ```no_run
//! use hexz_core::algo::dedup::cdc::analyze_stream;
//! use hexz_core::algo::dedup::dcam::DedupeParams;
//! use std::fs::File;
//!
//! # fn main() -> std::io::Result<()> {
//! let file = File::open("disk.raw")?;
//! let params = DedupeParams::default(); // f=14, m=2KB, z=64KB
//! let stats = analyze_stream(file, &params)?;
//!
//! println!("Total chunks: {}", stats.chunk_count);
//! println!("Unique chunks: {}", stats.unique_chunk_count);
//! println!("Dedup ratio: {:.2}%",
//!          (1.0 - stats.unique_bytes as f64 /
//!           (stats.unique_bytes as f64 * stats.chunk_count as f64 /
//!            stats.unique_chunk_count as f64)) * 100.0);
//! # Ok(())
//! # }
//! ```
//!
//! ## Streaming Chunks
//!
//! ```no_run
//! use hexz_core::algo::dedup::cdc::StreamChunker;
//! use hexz_core::algo::dedup::dcam::DedupeParams;
//! use std::fs::File;
//!
//! # fn main() -> std::io::Result<()> {
//! let file = File::open("memory.raw")?;
//! let params = DedupeParams::default();
//! let chunker = StreamChunker::new(file, params);
//!
//! for (i, chunk_result) in chunker.enumerate() {
//!     let chunk = chunk_result?;
//!     println!("Chunk {}: {} bytes", i, chunk.len());
//!     // Compress, encrypt, hash, store...
//! }
//! # Ok(())
//! # }
//! ```
//!
//! ## Parameter Comparison
//!
//! ```no_run
//! use hexz_core::algo::dedup::cdc::analyze_stream;
//! use hexz_core::algo::dedup::dcam::DedupeParams;
//! use std::fs::File;
//!
//! # fn main() -> std::io::Result<()> {
//! let data = std::fs::read("dataset.bin")?;
//!
//! // Small chunks (8KB average) - fine-grained dedup
//! let mut params_small = DedupeParams::default();
//! params_small.f = 13; // 2^13 = 8KB
//! let stats_small = analyze_stream(&data[..], &params_small)?;
//!
//! // Large chunks (32KB average) - coarse dedup, less overhead
//! let mut params_large = DedupeParams::default();
//! params_large.f = 15; // 2^15 = 32KB
//! let stats_large = analyze_stream(&data[..], &params_large)?;
//!
//! println!("Small chunks: {} unique / {} total",
//!          stats_small.unique_chunk_count, stats_small.chunk_count);
//! println!("Large chunks: {} unique / {} total",
//!          stats_large.unique_chunk_count, stats_large.chunk_count);
//! # Ok(())
//! # }
//! ```
//!
//! # Integration with Hexz
//!
//! CDC is integrated into the snapshot packing pipeline with multiple stages:
//!
//! ## Analysis Phase
//!
//! 1. **DCAM Analysis**: `analyze_stream()` performs dry-run chunking to estimate
//!    deduplication potential across different parameter configurations
//! 2. **Parameter Selection**: DCAM formulas compute optimal `f` value based on
//!    observed change rates and metadata overhead tradeoffs
//! 3. **Validation**: Verify parameters are within hardware constraints (chunk
//!    sizes fit in memory, hash set doesn't exceed available RAM)
//!
//! ## Packing Phase
//!
//! 4. **Streaming Chunking**: `StreamChunker` feeds variable-sized chunks to the
//!    compression pipeline (typically zstd or lz4)
//! 5. **Deduplication Tracking**: Compute cryptographic hash (SHA-256 or BLAKE3)
//!    of each chunk and check against existing snapshot index
//! 6. **Storage**: Write unique chunks to pack files; record chunk metadata
//!    (offset, compressed size, hash) in snapshot index
//!
//! ## Unpacking Phase
//!
//! 7. **Reconstruction**: Read snapshot index to determine chunk sequence
//! 8. **Decompression**: Fetch and decompress chunks from pack files
//! 9. **Assembly**: Concatenate chunks in original order to reconstruct snapshot
//!
//! ## Error Handling
//!
//! - **I/O Errors**: Propagated immediately, abort operation
//! - **Hash Collisions**: Detected during unpacking (content verification fails),
//!   triggers integrity error and potential re-read or fallback
//! - **Corrupted Chunks**: Compression errors detected during decompression,
//!   logged and may trigger snapshot re-creation if critical
//!
//! ## Edge Cases
//!
//! - **Empty input**: Returns zero chunks (not an error)
//! - **Tiny files** (< min_size): Single chunk containing entire file
//! - **Huge files** (> max_size × chunk_count): Automatic chunking prevents
//!   unbounded memory growth
//! - **Adversarial input**: Maximum chunk size enforced regardless of hash values
//!
//! # Limitations and Considerations
//!
//! ## Hash Collision Handling
//!
//! The `analyze_stream()` function uses the first 8 bytes of BLAKE3 as a 64-bit
//! hash. This provides a birthday bound at ~2^32 ≈ 4 billion unique chunks,
//! which is more than sufficient for analysis workloads:
//!
//! - **Negligible collision risk**: Even at 100M unique chunks, P(collision) < 0.1%
//! - **Not acceptable for storage**: Actual packing uses the full 256-bit BLAKE3
//!   hash via the dedup hash table to avoid data corruption
//!
//! ## Small File Overhead
//!
//! For files smaller than `min_size`, CDC produces a single chunk containing
//! the entire file. This is optimal behavior but means:
//!
//! - No deduplication benefit for very small files
//! - Metadata overhead (hash + pointer) is proportionally higher
//! - Consider alternative strategies for small file workloads (e.g., tar packing)
//!
//! ## Boundary Shift Resilience
//!
//! While CDC is resilient to insertions/deletions within data, it is NOT
//! resilient to:
//!
//! - **Encryption**: Changes entire content, prevents all deduplication
//! - **Compression**: Similar effect to encryption for CDC purposes
//! - **Byte-level reordering**: Shuffles content, breaks chunk boundaries
//!
//! **Recommendation**: Apply CDC before compression/encryption, not after.
//!
//! ## Adversarial Input
//!
//! An attacker could craft input that produces:
//!
//! - **All max-size chunks**: Content designed to never trigger hash cut-points
//!   (e.g., all zeros, which may or may not trigger depending on hash function)
//! - **All min-size chunks**: Content that triggers cut-points too frequently
//! - **Hash flooding**: Attempt to exhaust hash set memory in `analyze_stream()`
//!
//! The `max_size` enforcement prevents unbounded chunk growth, but adversarial
//! input can still degrade deduplication effectiveness. For untrusted input,
//! consider additional validation or rate limiting.
//!
//! ## Determinism Caveats
//!
//! Chunking is deterministic given:
//! - Same input bytes (bit-identical)
//! - Same parameters (`f`, `m`, `z`)
//! - Same fastcdc library version
//!
//! However, updating the fastcdc library may change internal hash tables or
//! algorithms, causing chunk boundaries to shift. This breaks cross-version
//! deduplication. **Mitigation**: Include CDC version in snapshot metadata.
//!
//! # References
//!
//! - Xia, W. et al. "FastCDC: a Fast and Efficient Content-Defined Chunking
//!   Approach for Data Deduplication." USENIX ATC 2016.
//!   [https://www.usenix.org/conference/atc16/technical-sessions/presentation/xia](https://www.usenix.org/conference/atc16/technical-sessions/presentation/xia)
//! - Muthitacharoen, A. et al. "A Low-bandwidth Network File System." SOSP 2001.
//!   (Original Rabin fingerprinting CDC)
//! - Randall et al. "Optimizing Deduplication Parameters via a Change-Estimation
//!   Analytical Model." 2025. (DCAM integration)

use crate::algo::dedup::dcam::DedupeParams;
use std::collections::HashSet;
use std::io::{self, Read};

/// Statistics from a CDC deduplication analysis run.
///
/// This structure captures metrics from a content-defined chunking analysis pass,
/// providing insights into chunk distribution, deduplication effectiveness, and
/// potential storage savings.
///
/// # Use Cases
///
/// - **DCAM Model Input**: Feeds into analytical formulas to predict optimal parameters
/// - **Capacity Planning**: Estimates actual storage requirements after deduplication
/// - **Performance Analysis**: Evaluates chunk size distribution and variance
/// - **Comparison Studies**: Benchmarks different parameter configurations
///
/// # Invariants
///
/// The following relationships always hold for valid statistics:
///
/// - `unique_chunk_count <= chunk_count` (can't have more unique than total)
/// - `unique_bytes <= chunk_count * max_chunk_size` (bounded by chunk constraints)
/// - `unique_bytes >= unique_chunk_count * min_chunk_size` (minimum size bound)
/// - `total_bytes` may be 0 (not tracked in streaming mode)
///
/// # Calculating Deduplication Metrics
///
/// ## Deduplication Ratio
///
/// The deduplication ratio represents what fraction of data is unique:
///
/// ```text
/// dedup_ratio = unique_bytes / total_bytes_before_dedup
/// ```
///
/// Where `total_bytes_before_dedup` (the original data size) can be estimated from:
/// ```text
/// total_bytes_before_dedup ≈ unique_bytes × (chunk_count / unique_chunk_count)
/// ```
///
/// **Interpretation:**
/// - Ratio of **0.5** means 50% of data is unique → **50% storage savings**
/// - Ratio of **1.0** means 100% unique (no duplicates) → **0% savings**
/// - Ratio of **0.3** means 30% unique → **70% savings**
///
/// ## Deduplication Factor
///
/// An alternative metric is the deduplication factor (multiplicative savings):
///
/// ```text
/// dedup_factor = chunk_count / unique_chunk_count
/// ```
///
/// **Interpretation:**
/// - Factor of **1.0x** means no deduplication occurred
/// - Factor of **2.0x** means data compressed by half (50% savings)
/// - Factor of **10.0x** means 90% of chunks were duplicates
///
/// ## Average Chunk Size
///
/// The realized average chunk size (may differ from `2^f` for small datasets):
///
/// ```text
/// avg_chunk_size = unique_bytes / unique_chunk_count
/// ```
///
/// # Examples
///
/// ```rust
/// # use hexz_core::algo::dedup::cdc::CdcStats;
/// let stats = CdcStats {
///     total_bytes: 0, // Not tracked (estimated below)
///     unique_bytes: 50_000_000,
///     chunk_count: 10_000,
///     unique_chunk_count: 3_000,
/// };
///
/// // Estimate original size
/// let estimated_original = stats.unique_bytes * stats.chunk_count / stats.unique_chunk_count;
/// assert_eq!(estimated_original, 166_666_666); // ~167 MB original
///
/// // Calculate deduplication ratio (fraction unique)
/// let dedup_ratio = stats.unique_bytes as f64 / estimated_original as f64;
/// assert!((dedup_ratio - 0.3).abs() < 0.01); // ~30% unique
///
/// // Calculate storage savings (percent eliminated)
/// let savings_percent = (1.0 - dedup_ratio) * 100.0;
/// assert!((savings_percent - 70.0).abs() < 1.0); // ~70% savings
///
/// // Calculate deduplication factor (compression ratio)
/// let dedup_factor = stats.chunk_count as f64 / stats.unique_chunk_count as f64;
/// assert!((dedup_factor - 3.33).abs() < 0.01); // ~3.33x compression
///
/// // Average chunk size
/// let avg_chunk_size = stats.unique_bytes / stats.unique_chunk_count;
/// println!("Average unique chunk size: {} bytes", avg_chunk_size);
/// ```
#[derive(Debug)]
pub struct CdcStats {
    /// Total bytes processed from the input stream.
    ///
    /// **Note**: This field is currently not tracked during streaming analysis
    /// (always set to 0) to avoid memory overhead. It may be populated in future
    /// versions if needed for enhanced metrics.
    ///
    /// To estimate total bytes, use:
    /// ```text
    /// total_bytes ≈ unique_bytes * (chunk_count / unique_chunk_count)
    /// ```
    pub total_bytes: u64,

    /// Number of unique bytes after deduplication.
    ///
    /// This is the sum of sizes of all unique chunks (first occurrence only).
    /// Represents the actual storage space required if deduplication is applied.
    ///
    /// # Interpretation
    ///
    /// - Lower values indicate higher redundancy in the dataset
    /// - Equals total data size if no duplicates exist
    /// - Bounded by `unique_chunk_count * min_chunk_size` and
    ///   `unique_chunk_count * max_chunk_size`
    pub unique_bytes: u64,

    /// Total number of chunks identified by FastCDC.
    ///
    /// Includes both unique and duplicate chunks. This represents how many chunks
    /// would be created if the data were processed through the chunking pipeline.
    ///
    /// # Expected Values
    ///
    /// For a dataset of size `N` bytes with average chunk size `2^f`:
    /// ```text
    /// chunk_count ≈ N / (2^f)
    /// ```
    ///
    /// Example: 1GB file with f=14 (16KB avg) → ~65,536 chunks
    pub chunk_count: u64,

    /// Number of unique chunks after deduplication.
    ///
    /// This counts only distinct chunks (based on CRC32 hash comparison).
    /// Duplicate chunks are not counted in this total.
    ///
    /// # Interpretation
    ///
    /// - `unique_chunk_count == chunk_count` → No deduplication (all unique)
    /// - `unique_chunk_count << chunk_count` → High deduplication (many duplicates)
    /// - Ratio `unique_chunk_count / chunk_count` indicates dedup effectiveness
    ///
    /// # Hash Collision Note
    ///
    /// Uses the first 8 bytes of BLAKE3 as a 64-bit hash, providing negligible
    /// collision probability (birthday bound at ~2^32 ≈ 4 billion unique chunks).
    pub unique_chunk_count: u64,
}

/// Streaming iterator that yields content-defined chunks from a reader.
///
/// `StreamChunker` is the primary interface for applying FastCDC to data streams
/// in a memory-efficient manner. It reads data incrementally from any `Read`
/// source and applies content-defined chunking to produce variable-sized chunks
/// suitable for compression, deduplication, and storage.
///
/// # Design Goals
///
/// - **Streaming**: Process arbitrarily large files without loading entire contents
/// - **Zero-copy (internal)**: Minimize data copying within the chunker
/// - **Bounded memory**: Fixed buffer size regardless of input size
/// - **Deterministic**: Same input always produces identical chunk boundaries
/// - **Iterator-based**: Idiomatic Rust API that composes with other iterators
///
/// # Buffering Strategy
///
/// The chunker maintains a fixed-size internal buffer to handle the mismatch
/// between read boundaries (determined by the OS/filesystem) and chunk boundaries
/// (determined by content):
///
/// - **Buffer size**: `2 * max_chunk_size` (default: ~128 KB)
/// - **Refill trigger**: When available data < `max_chunk_size`
/// - **Shift threshold**: When cursor advances beyond buffer midpoint
/// - **EOF handling**: Flushes remaining data as final chunk
///
/// ## Buffer Management Example
///
/// ```text
/// Initial state (empty buffer):
/// [__________________________________|__________________________________]
///  0                                 cursor                          2*z
///
/// After reading (filled to cursor):
/// [##################################|__________________________________]
///  0                                 cursor=filled                   2*z
///
/// After consuming one chunk (cursor advances):
/// [################XXXXXXXXXXXXXXXXXX|__________________________________]
///  0              consumed          cursor                           2*z
///                 (next chunk)       filled
///
/// After shift (move unconsumed data to start):
/// [XXXXXXXXXXXXXXXXXX________________|__________________________________]
///  0                cursor           filled                          2*z
///  (new position)
/// ```
///
/// # Chunk Boundary Guarantees
///
/// The chunker guarantees that:
///
/// 1. **Minimum size**: Every chunk (except possibly the last) is ≥ `min_size`
/// 2. **Maximum size**: Every chunk is ≤ `max_size` (strictly enforced)
/// 3. **Determinism**: Given the same input and parameters, chunk boundaries are identical
/// 4. **Completeness**: All input bytes appear in exactly one chunk (no gaps, no overlaps)
///
/// # Performance Considerations
///
/// - **Memory allocation**: Each chunk is copied to an owned `Vec<u8>`. For
///   zero-allocation iteration, consider using `StreamChunkerSlice` (if available)
///   or process chunks in-place within the iterator.
/// - **I/O pattern**: Reads in large blocks (up to `max_chunk_size`) to minimize
///   syscalls. Works efficiently with buffered readers but not required.
/// - **CPU cost**: Gear hash computation is the bottleneck (~500 MB/s). Consider
///   parallel chunking of independent streams if throughput is critical.
///
/// # Examples
///
/// ## Basic Usage
///
/// ```no_run
/// use hexz_core::algo::dedup::cdc::StreamChunker;
/// use hexz_core::algo::dedup::dcam::DedupeParams;
/// use std::fs::File;
///
/// # fn main() -> std::io::Result<()> {
/// let file = File::open("data.bin")?;
/// let params = DedupeParams::default();
/// let chunker = StreamChunker::new(file, params);
///
/// for chunk_result in chunker {
///     let chunk = chunk_result?;
///     println!("Chunk: {} bytes", chunk.len());
/// }
/// # Ok(())
/// # }
/// ```
///
/// ## With Compression Pipeline
///
/// ```no_run
/// use hexz_core::algo::dedup::cdc::StreamChunker;
/// use hexz_core::algo::dedup::dcam::DedupeParams;
/// use std::fs::File;
/// use std::collections::HashMap;
///
/// # fn main() -> std::io::Result<()> {
/// let file = File::open("memory.raw")?;
/// let params = DedupeParams::default();
/// let chunker = StreamChunker::new(file, params);
///
/// let mut dedup_map: HashMap<u64, usize> = HashMap::new();
/// let mut unique_chunks = Vec::new();
///
/// for chunk_result in chunker {
///     let chunk = chunk_result?;
///     let hash = crc32fast::hash(&chunk) as u64;
///
///     if !dedup_map.contains_key(&hash) {
///         let chunk_id = unique_chunks.len();
///         dedup_map.insert(hash, chunk_id);
///         // Compress and store chunk
///         unique_chunks.push(chunk);
///     }
/// }
///
/// println!("Stored {} unique chunks", unique_chunks.len());
/// # Ok(())
/// # }
/// ```
///
/// ## Custom Parameters
///
/// ```no_run
/// use hexz_core::algo::dedup::cdc::StreamChunker;
/// use hexz_core::algo::dedup::dcam::DedupeParams;
/// use std::fs::File;
///
/// # fn main() -> std::io::Result<()> {
/// let file = File::open("disk.raw")?;
///
/// // Custom parameters: larger chunks for better compression ratio
/// let mut params = DedupeParams::default();
/// params.f = 15; // 32KB average (was 16KB)
/// params.m = 8192; // 8KB minimum (was 2KB)
/// params.z = 131072; // 128KB maximum (was 64KB)
///
/// let chunker = StreamChunker::new(file, params);
///
/// let chunk_sizes: Vec<usize> = chunker
///     .map(|res| res.map(|chunk| chunk.len()))
///     .collect::<Result<_, _>>()?;
///
/// let avg_size: usize = chunk_sizes.iter().sum::<usize>() / chunk_sizes.len();
/// println!("Average chunk size: {} bytes", avg_size);
/// # Ok(())
/// # }
/// ```
pub struct StreamChunker<R> {
    /// Underlying data source.
    ///
    /// Can be any type implementing `Read`: files, network streams, memory
    /// buffers, etc. The chunker makes no assumptions about the read pattern
    /// or blocking behavior.
    reader: R,

    /// Internal buffer for staging data between reads and chunk boundaries.
    ///
    /// Size is `2 * max_chunk_size` to allow efficient buffering and shifting.
    /// Allocated once during construction and reused for the entire stream.
    buffer: Vec<u8>,

    /// Current read position within the buffer.
    ///
    /// Points to the start of the next chunk to be yielded. Invariant:
    /// `0 <= cursor <= filled <= buffer.len()`
    cursor: usize,

    /// Number of valid bytes currently in the buffer.
    ///
    /// Bytes in range `[0, filled)` contain valid data. Bytes in range
    /// `[filled, buffer.len())` are unused scratch space.
    filled: usize,

    /// Minimum chunk size in bytes (FastCDC parameter `m`).
    ///
    /// No chunk will be smaller than this (except possibly the final chunk
    /// if EOF is reached). Prevents tiny chunks that would increase metadata
    /// overhead disproportionately.
    min_size: usize,

    /// Average chunk size in bytes (FastCDC parameter `2^f`).
    ///
    /// This is the statistical expectation of chunk sizes. Actual chunks follow
    /// a bounded exponential distribution around this value due to the rolling
    /// hash cut-point probability.
    avg_size: usize,

    /// Maximum chunk size in bytes (FastCDC parameter `z`).
    ///
    /// Chunks are forcibly cut at this size regardless of hash values. Bounds
    /// worst-case memory usage and ensures progress even in adversarial inputs
    /// (e.g., runs of zeros that never trigger hash-based cuts).
    max_size: usize,

    /// Whether EOF has been reached on the underlying reader.
    ///
    /// Once set to `true`, no further reads will be attempted. The chunker
    /// will drain remaining buffered data and then terminate iteration.
    eof: bool,
}

impl<R: Read> StreamChunker<R> {
    /// Creates a new streaming chunker with the specified deduplication parameters.
    ///
    /// This constructor initializes the chunker's internal buffer and extracts the
    /// relevant FastCDC parameters. The chunker is ready to begin iteration immediately
    /// after construction; no separate initialization step is required.
    ///
    /// # Parameters
    ///
    /// - `reader`: Data source implementing `Read`. This can be:
    ///   - `File` for reading from disk
    ///   - `Cursor<Vec<u8>>` for in-memory data
    ///   - `BufReader<File>` for buffered I/O (redundant but not harmful)
    ///   - Any other `Read` implementation (network streams, pipes, etc.)
    ///
    /// - `params`: FastCDC parameters controlling chunk size distribution. Key fields:
    ///   - `params.f`: Fingerprint bits (average chunk size = 2^f)
    ///   - `params.m`: Minimum chunk size in bytes
    ///   - `params.z`: Maximum chunk size in bytes
    ///   - `params.w`: Rolling hash window size (informational, not used here)
    ///
    /// # Returns
    ///
    /// A new `StreamChunker` ready to yield chunks via iteration. The chunker
    /// takes ownership of the reader and will consume it as iteration proceeds.
    ///
    /// # Memory Allocation
    ///
    /// Allocates a buffer of `2 * params.z` bytes (or 2 MB, whichever is larger).
    /// This is a one-time allocation that persists for the entire stream:
    ///
    /// - Default settings (`z=64KB`): **~128 KB** per chunker
    /// - Large chunks (`z=128KB`): **~256 KB** per chunker
    /// - Small chunks (`z=16KB`): **~2 MB** per chunker (due to 2MB minimum)
    ///
    /// The 2MB minimum prevents excessive refill operations for small chunk sizes.
    ///
    /// # Panics
    ///
    /// May panic if memory allocation fails (extremely rare on modern systems with
    /// virtual memory, but possible in constrained environments). For default parameters
    /// this allocates ~128 KB, well within typical stack/heap limits.
    ///
    /// # Performance Notes
    ///
    /// - **No I/O in constructor**: Data reading is deferred until first `next()` call
    /// - **Deterministic**: Given the same input and parameters, produces identical chunks
    /// - **Thread-safe (if R is)**: The chunker itself has no shared state
    ///
    /// # Examples
    ///
    /// ## Default Parameters
    ///
    /// ```no_run
    /// use hexz_core::algo::dedup::cdc::StreamChunker;
    /// use hexz_core::algo::dedup::dcam::DedupeParams;
    /// use std::fs::File;
    ///
    /// # fn main() -> std::io::Result<()> {
    /// let file = File::open("data.bin")?;
    /// let chunker = StreamChunker::new(file, DedupeParams::default());
    /// // Chunker is ready to iterate
    /// # Ok(())
    /// # }
    /// ```
    ///
    /// ## Custom Parameters
    ///
    /// ```no_run
    /// use hexz_core::algo::dedup::cdc::StreamChunker;
    /// use hexz_core::algo::dedup::dcam::DedupeParams;
    /// use std::fs::File;
    ///
    /// # fn main() -> std::io::Result<()> {
    /// let file = File::open("disk.raw")?;
    /// let params = DedupeParams {
    ///     f: 15,       // 32KB average
    ///     m: 4096,     // 4KB minimum
    ///     z: 131072,   // 128KB maximum
    ///     w: 48,       // Window size (not used by StreamChunker)
    ///     v: 16,       // Metadata overhead (not used by StreamChunker)
    /// };
    /// let chunker = StreamChunker::new(file, params);
    /// # Ok(())
    /// # }
    /// ```
    pub fn new(reader: R, params: DedupeParams) -> Self {
        // Buffer needs to be at least max_size.
        // We use 2 * max_size to allow for shifting and efficient reading.
        let buf_size = (params.z as usize).max(1024 * 1024) * 2;
        Self {
            reader,
            buffer: vec![0u8; buf_size],
            cursor: 0,
            filled: 0,
            min_size: params.m as usize,
            avg_size: 1 << params.f,
            max_size: params.z as usize,
            eof: false,
        }
    }

    /// Refills the internal buffer from the reader.
    ///
    /// This private method manages the buffer state to maintain a continuous window
    /// of data for chunking. It performs buffer compaction (shifting) and reads new
    /// data to maximize available space for chunk detection.
    ///
    /// # Algorithm
    ///
    /// 1. **Shift unprocessed data**: If `cursor > 0`, move bytes `[cursor..filled)`
    ///    to the start of the buffer to reclaim space
    /// 2. **Read new data**: Call `reader.read()` repeatedly to fill available space
    /// 3. **Update EOF flag**: Set `eof = true` when reader returns 0 bytes
    /// 4. **Early exit**: Stop reading once buffer contains ≥ `max_chunk_size` bytes
    ///
    /// # Buffer State Before/After
    ///
    /// Before:
    /// ```text
    /// [consumed|unprocessed______|____________available____________]
    ///  0       cursor            filled                          capacity
    /// ```
    ///
    /// After shift:
    /// ```text
    /// [unprocessed______|____________available___________________]
    ///  0                new_filled                             capacity
    /// ```
    ///
    /// After refill:
    /// ```text
    /// [unprocessed______|new_data__________________________|____]
    ///  0                old_filled                        new_filled
    /// ```
    ///
    /// # Performance
    ///
    /// - **Shift cost**: O(filled - cursor) via `copy_within()` (memmove)
    /// - **Read cost**: Depends on underlying reader (syscall overhead)
    /// - **Amortization**: Shifts happen infrequently (once per chunk consumed)
    ///
    /// # Errors
    ///
    /// Returns any I/O errors encountered during reading. Common errors:
    ///
    /// - `ErrorKind::UnexpectedEof`: Reader closed unexpectedly (shouldn't happen
    ///   for valid streams; `read()` returning 0 is the proper EOF signal)
    /// - `ErrorKind::Interrupted`: Signal interrupted read (automatically retried
    ///   by the read loop in this implementation)
    /// - Other errors: Propagated from underlying reader (e.g., permission denied,
    ///   device errors, network timeouts)
    fn refill(&mut self) -> io::Result<()> {
        // Only shift when cursor has consumed more than half the buffer.
        // This reduces memmove frequency by ~50% while still ensuring enough
        // space for the next read.
        if self.cursor > self.buffer.len() / 2 {
            self.buffer.copy_within(self.cursor..self.filled, 0);
            self.filled -= self.cursor;
            self.cursor = 0;
        }

        while self.filled < self.buffer.len() && !self.eof {
            let n = self.reader.read(&mut self.buffer[self.filled..])?;
            if n == 0 {
                self.eof = true;
            } else {
                self.filled += n;
                // If we have enough data to potentially find a chunk, we can stop reading for now
                if self.filled >= self.max_size {
                    break;
                }
            }
        }
        Ok(())
    }
}

impl<R: Read> Iterator for StreamChunker<R> {
    type Item = io::Result<Vec<u8>>;

    /// Yields the next content-defined chunk.
    ///
    /// This is the core iterator method that drives the chunking process. Each call
    /// performs the following operations:
    ///
    /// # Algorithm
    ///
    /// 1. **Check for data**: If buffer is empty and EOF not reached, refill
    /// 2. **Handle EOF**: If no data available after refill, return `None`
    /// 3. **Determine available window**: Compute bytes available for chunking
    /// 4. **Apply FastCDC**:
    ///    - If available < `min_size`: Return all available bytes (last chunk)
    ///    - Else: Run FastCDC on window `[cursor..min(cursor+max_size, filled)]`
    /// 5. **Extract chunk**: Copy chunk bytes to new `Vec<u8>`, advance cursor
    /// 6. **Return**: Yield `Some(Ok(chunk))`
    ///
    /// # FastCDC Integration
    ///
    /// The method delegates to the `fastcdc` crate's `FastCDC::new()` constructor,
    /// which implements the normalized chunking algorithm. The returned chunk length
    /// respects all size constraints:
    ///
    /// - No chunk < `min_size` (except final chunk)
    /// - No chunk > `max_size` (strictly enforced)
    /// - Average chunk size ≈ `avg_size = 2^f`
    ///
    /// # Chunk Length Determination
    ///
    /// The algorithm chooses chunk length as follows:
    ///
    /// | Condition | Action |
    /// |-----------|--------|
    /// | `available < min_size` | Return all available bytes |
    /// | FastCDC finds cut point | Use FastCDC-determined length |
    /// | `available >= max_size` && no cut point | Force cut at `max_size` |
    /// | EOF reached && no cut point | Return all remaining bytes |
    /// | Buffer full && no cut point | Force cut at `max_size` |
    /// | Other | Return all available bytes |
    ///
    /// # Returns
    ///
    /// - `Some(Ok(chunk))` - Next chunk successfully extracted and copied
    /// - `Some(Err(e))` - I/O error occurred during refill
    /// - `None` - Stream exhausted (no more data available)
    ///
    /// # Errors
    ///
    /// Returns `Some(Err(e))` if the underlying reader encounters an I/O error
    /// during refill. After an error, the iterator should be considered invalid
    /// (further calls may panic or return inconsistent results).
    ///
    /// Common error causes:
    /// - File read errors (permissions, disk errors)
    /// - Network errors (timeouts, connection reset)
    /// - Interrupted system calls (usually auto-retried)
    ///
    /// # Chunk Size Guarantees
    ///
    /// The returned chunks satisfy these constraints:
    ///
    /// - **Minimum**: `≥ min_size` (except possibly the last chunk)
    /// - **Maximum**: `≤ max_size` (always enforced, never violated)
    /// - **Average**: `≈ avg_size = 2^f` (statistical expectation)
    /// - **Last chunk**: May be smaller than `min_size` if EOF reached
    ///
    /// # Memory Allocation
    ///
    /// Each chunk is copied to a new `Vec<u8>`. For zero-copy iteration within
    /// the buffer lifetime, consider modifying the API to return slices with
    /// appropriate lifetimes (future enhancement).
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use hexz_core::algo::dedup::cdc::StreamChunker;
    /// use hexz_core::algo::dedup::dcam::DedupeParams;
    /// use std::fs::File;
    ///
    /// # fn main() -> std::io::Result<()> {
    /// let file = File::open("data.bin")?;
    /// let mut chunker = StreamChunker::new(file, DedupeParams::default());
    ///
    /// // First chunk
    /// if let Some(Ok(chunk)) = chunker.next() {
    ///     println!("First chunk: {} bytes", chunk.len());
    /// }
    ///
    /// // Iterate remaining chunks
    /// for chunk_result in chunker {
    ///     let chunk = chunk_result?;
    ///     // Process chunk...
    /// }
    /// # Ok(())
    /// # }
    /// ```
    ///
    /// # Performance Characteristics
    ///
    /// - **Amortized time**: O(chunk_size) per chunk (dominated by copying)
    /// - **Space**: O(chunk_size) allocation per chunk yielded
    /// - **Throughput**: ~500 MB/s on modern CPUs (bottlenecked by Gear hash)
    fn next(&mut self) -> Option<Self::Item> {
        if self.cursor >= self.filled {
            if self.eof {
                return None;
            }
            if let Err(e) = self.refill() {
                return Some(Err(e));
            }
            if self.filled == 0 {
                return None;
            }
        }

        let available = self.filled - self.cursor;

        // Refill if we don't have enough data for a full chunk and more data is available
        if available < self.max_size && !self.eof {
            if let Err(e) = self.refill() {
                return Some(Err(e));
            }
        }
        let available = self.filled - self.cursor;

        let chunk_len = if available < self.min_size {
            available
        } else {
            // Run FastCDC on the available window
            let data = &self.buffer[self.cursor..self.filled];
            let search_limit = std::cmp::min(data.len(), self.max_size);

            let chunker = fastcdc::v2020::FastCDC::new(
                &data[..search_limit],
                self.min_size as u32,
                self.avg_size as u32,
                self.max_size as u32,
            );

            if let Some(chunk) = chunker.into_iter().next() {
                chunk.length
            } else {
                // No cut point found
                if available >= self.max_size {
                    self.max_size
                } else if self.eof {
                    available
                } else if self.filled == self.buffer.len() {
                    self.max_size
                } else {
                    available
                }
            }
        };

        let start = self.cursor;
        self.cursor += chunk_len;
        Some(Ok(self.buffer[start..start + chunk_len].to_vec()))
    }
}

/// Performs a single-pass deduplication analysis on a data stream.
///
/// This function applies FastCDC chunking to a data stream and tracks unique chunks
/// via hash-based deduplication. It is designed as a lightweight analysis pass for
/// the DCAM model to estimate deduplication potential before committing to full
/// snapshot packing.
///
/// # Algorithm
///
/// 1. **Chunk the input** using `StreamChunker` with the provided FastCDC parameters
/// 2. **Hash each chunk** using CRC32 (fast, non-cryptographic)
/// 3. **Track uniqueness** via a `HashSet<u64>` of seen hashes
/// 4. **Accumulate statistics**:
///    - Total chunk count (all chunks, including duplicates)
///    - Unique chunk count (distinct hashes)
///    - Unique bytes (sum of unique chunk sizes)
///
/// # Parameters
///
/// - `reader`: Data source to analyze. Can be any `Read` implementation:
///   - `File` for disk images or backups
///   - `Cursor<Vec<u8>>` for in-memory data
///   - Network streams, pipes, or other I/O sources
///
/// - `params`: FastCDC parameters controlling chunking behavior:
///   - `params.f`: Fingerprint bits (average chunk size = 2^f bytes)
///   - `params.m`: Minimum chunk size in bytes
///   - `params.z`: Maximum chunk size in bytes
///   - Other fields (`w`, `v`) are used by DCAM but not by the chunker
///
/// # Returns
///
/// `Ok(CdcStats)` containing:
/// - `chunk_count`: Total chunks processed (including duplicates)
/// - `unique_chunk_count`: Number of distinct chunks (unique hashes)
/// - `unique_bytes`: Total bytes in unique chunks (storage requirement estimate)
/// - `total_bytes`: Currently unused (always 0) for performance reasons
///
/// # Errors
///
/// Returns `Err` if the underlying reader encounters an I/O error during reading.
/// Common error cases:
///
/// - `std::io::ErrorKind::NotFound`: File doesn't exist (if using `File`)
/// - `std::io::ErrorKind::PermissionDenied`: Insufficient permissions to read
/// - `std::io::ErrorKind::UnexpectedEof`: Reader closed unexpectedly
/// - Network-related errors (timeouts, connection resets) for network streams
/// - Disk errors (bad sectors, hardware failures) for file-based readers
///
/// After an error, the returned statistics (if any) are invalid and should be
/// discarded. The function does not attempt recovery or partial results.
///
/// # Performance
///
/// ## Time Complexity
///
/// - **Single-threaded throughput**: ~500 MB/s on modern CPUs (2020+)
/// - **Bottleneck**: Gear hash computation (CPU-bound, not I/O-bound)
/// - **Scaling**: Linear in input size, O(N) where N = total bytes
///
/// ## Space Complexity
///
/// - **Hash set**: O(unique chunks) × 8 bytes per hash ≈ O(N / avg_chunk_size) × 8
/// - **Example**: 1GB input, 16KB avg chunks → ~65K unique chunks → ~520 KB hash set
/// - **Streaming buffer**: Fixed at `2 * params.z` (typically ~128 KB)
///
/// ## Memory Optimization
///
/// For extremely large datasets where hash set size becomes prohibitive, consider:
/// - Using a probabilistic data structure (Bloom filter, HyperLogLog)
/// - Sampling (analyze every Nth chunk instead of all chunks)
/// - External deduplication index (disk-based hash storage)
///
/// # Hash Collision Probability
///
/// This function uses the first 8 bytes of BLAKE3 as a 64-bit hash.
/// The probability of collision is governed by the birthday paradox:
///
/// ```text
/// P(collision) ≈ n² / (2 × 2^64) where n = unique_chunk_count
/// ```
///
/// For typical workloads:
/// - **1M unique chunks**: P(collision) ≈ 2.7e-8 — negligible
/// - **100M unique chunks**: P(collision) ≈ 2.7e-4 (0.027%) — negligible
/// - **4 billion unique chunks**: P(collision) ≈ 50% (birthday bound)
///
/// # Examples
///
/// ## Basic Usage
///
/// ```no_run
/// use hexz_core::algo::dedup::cdc::analyze_stream;
/// use hexz_core::algo::dedup::dcam::DedupeParams;
/// use std::fs::File;
///
/// # fn main() -> std::io::Result<()> {
/// let file = File::open("dataset.bin")?;
/// let params = DedupeParams::default();
/// let stats = analyze_stream(file, &params)?;
///
/// println!("Total chunks: {}", stats.chunk_count);
/// println!("Unique chunks: {}", stats.unique_chunk_count);
/// println!("Unique bytes: {} MB", stats.unique_bytes / 1_000_000);
///
/// let dedup_factor = stats.chunk_count as f64 / stats.unique_chunk_count as f64;
/// println!("Deduplication factor: {:.2}x", dedup_factor);
/// # Ok(())
/// # }
/// ```
///
/// ## Estimating Storage Savings
///
/// ```no_run
/// use hexz_core::algo::dedup::cdc::analyze_stream;
/// use hexz_core::algo::dedup::dcam::DedupeParams;
/// use std::fs::File;
///
/// # fn main() -> std::io::Result<()> {
/// let file = File::open("snapshot.raw")?;
/// let params = DedupeParams::default();
/// let stats = analyze_stream(file, &params)?;
///
/// // Estimate total bytes (not tracked directly for performance)
/// let estimated_total = stats.unique_bytes * stats.chunk_count / stats.unique_chunk_count;
/// let savings_ratio = 1.0 - (stats.unique_bytes as f64 / estimated_total as f64);
///
/// println!("Estimated storage savings: {:.1}%", savings_ratio * 100.0);
/// # Ok(())
/// # }
/// ```
///
/// ## Comparing Parameter Sets
///
/// ```no_run
/// use hexz_core::algo::dedup::cdc::analyze_stream;
/// use hexz_core::algo::dedup::dcam::DedupeParams;
/// use std::io::Cursor;
///
/// # fn main() -> std::io::Result<()> {
/// let data = vec![0u8; 10_000_000]; // 10 MB test data
///
/// // Analyze with small chunks
/// let mut params_small = DedupeParams::default();
/// params_small.f = 13; // 8KB average
/// let stats_small = analyze_stream(Cursor::new(&data), &params_small)?;
///
/// // Analyze with large chunks
/// let mut params_large = DedupeParams::default();
/// params_large.f = 15; // 32KB average
/// let stats_large = analyze_stream(Cursor::new(&data), &params_large)?;
///
/// println!("Small chunks: {} total, {} unique",
///          stats_small.chunk_count, stats_small.unique_chunk_count);
/// println!("Large chunks: {} total, {} unique",
///          stats_large.chunk_count, stats_large.unique_chunk_count);
/// # Ok(())
/// # }
/// ```
///
/// # Use Cases
///
/// - **DCAM Model Input**: Feeds statistics into analytical formulas to predict
///   optimal FastCDC parameters for a given dataset
/// - **Capacity Planning**: Estimates storage requirements before provisioning
///   deduplication infrastructure
/// - **Parameter Tuning**: Compares deduplication effectiveness across different
///   chunk size configurations
/// - **Workload Characterization**: Analyzes redundancy patterns in datasets
///
/// # Relationship to `StreamChunker`
///
/// This function is a convenience wrapper around `StreamChunker` that adds
/// deduplication tracking via a hash set. For actual snapshot packing, use
/// `StreamChunker` directly and handle deduplication in the compression/storage
/// pipeline. The analysis function is intended as a lightweight dry-run only.
pub fn analyze_stream<R: Read>(reader: R, params: &DedupeParams) -> io::Result<CdcStats> {
    let mut unique_bytes = 0;
    let mut chunk_count = 0;
    let mut unique_chunk_count = 0;
    let mut seen_chunks: HashSet<u64> = HashSet::new();

    // Use the streaming chunker for analysis too, to ensure consistency
    let chunker = StreamChunker::new(reader, *params);

    for chunk_res in chunker {
        let chunk = chunk_res?;
        let len = chunk.len() as u64;
        // Use first 8 bytes of BLAKE3 digest as a 64-bit hash.
        // BLAKE3 at 16KB chunks is >3 GB/s — only ~2-3x slower than CRC32 —
        // but provides 2^64 hash space (birthday bound at ~4 billion chunks)
        // vs CRC32's 2^32 (birthday bound at ~65K chunks).
        let digest = *blake3::hash(&chunk).as_bytes();
        let hash = u64::from_le_bytes([
            digest[0], digest[1], digest[2], digest[3], digest[4], digest[5], digest[6], digest[7],
        ]);

        chunk_count += 1;
        if seen_chunks.insert(hash) {
            unique_bytes += len;
            unique_chunk_count += 1;
        }
    }

    Ok(CdcStats {
        total_bytes: 0,
        unique_bytes,
        chunk_count,
        unique_chunk_count,
    })
}