fxcp_core/
chunker.rs

1// SPDX-License-Identifier: GPL-2.0-or-later
2// Copyright (C) 2025 Joel Wirāmu Pauling <aenertia@aenertia.net>
3//
4// fxcp-core/src/chunker.rs  --  Gear hash variable-size rolling chunker for FXAR v2
5
6//! Variable-size content-defined chunking via gear hash.
7//! Produces content-determined chunk boundaries that are stable across
8//! insertions/deletions, enabling high deduplication ratios.
9
10use std::io::Read;
11
12/// Default chunking parameters (same as borg defaults).
13pub const DEFAULT_CHUNK_MIN: usize = 2 * 1024;       // 2 KB
14/// Default average chunk size (64 KB).
15pub const DEFAULT_CHUNK_AVG: usize = 64 * 1024;      // 64 KB
16/// Default maximum chunk size (2 MB).
17pub const DEFAULT_CHUNK_MAX: usize = 2 * 1024 * 1024; // 2 MB
18
19/// A chunk produced by the gear hash chunker.
20pub struct Chunk {
21    pub data: Vec<u8>,
22    pub hash: blake3::Hash,
23}
24
25/// Gear hash rolling chunker configuration.
26#[derive(Debug)]
27pub struct GearChunker {
28    pub min: usize,
29    pub avg: usize,
30    pub max: usize,
31    /// Normalization level for chunk size distribution.
32    /// - 0: Legacy single-mask (geometric distribution, backward compatible)
33    /// - 1: FastCDC normalized (dual-mask, tighter distribution around avg)
34    /// - 2: Aggressive normalization (very tight, may reduce dedup ratio)
35    ///
36    /// # CAS Compatibility Warning
37    /// Changing this field produces different chunk boundaries and therefore
38    /// different BLAKE3 chunk hashes. Existing CAS data will not deduplicate
39    /// against chunks produced at a different normalization level.
40    pub normalization_level: u8,
41}
42
43impl Default for GearChunker {
44    fn default() -> Self {
45        Self {
46            min: DEFAULT_CHUNK_MIN,
47            avg: DEFAULT_CHUNK_AVG,
48            max: DEFAULT_CHUNK_MAX,
49            normalization_level: 0,
50        }
51    }
52}
53
54/// S3 WAN preset: conservative chunking for high-latency S3 links.
55/// min=5MB, avg=8MB, max=64MB
56#[cfg(feature = "cloud")]
57pub const S3_WAN: GearChunker = GearChunker {
58    min: 5 * 1024 * 1024,
59    avg: 8 * 1024 * 1024,
60    max: 64 * 1024 * 1024,
61    normalization_level: 0,
62};
63
64/// S3 LAN preset: balanced chunking for local-network S3 endpoints (MinIO).
65/// min=5MB, avg=16MB, max=128MB
66#[cfg(feature = "cloud")]
67pub const S3_LAN: GearChunker = GearChunker {
68    min: 5 * 1024 * 1024,
69    avg: 16 * 1024 * 1024,
70    max: 128 * 1024 * 1024,
71    normalization_level: 0,
72};
73
74/// S3 ARCHIVE preset: large chunks for cold storage / Glacier-class targets.
75/// min=5MB, avg=32MB, max=256MB
76#[cfg(feature = "cloud")]
77pub const S3_ARCHIVE: GearChunker = GearChunker {
78    min: 5 * 1024 * 1024,
79    avg: 32 * 1024 * 1024,
80    max: 256 * 1024 * 1024,
81    normalization_level: 0,
82};
83
84impl GearChunker {
85    /// Create a new gear hash chunker with the given size bounds.
86    pub fn new(min: usize, avg: usize, max: usize) -> crate::Result<Self> {
87        if min == 0 {
88            return Err(crate::FxcpError::Config("GearChunker: min must be > 0".into()));
89        }
90        if avg < min {
91            return Err(crate::FxcpError::Config(
92                format!("GearChunker: avg ({avg}) must be >= min ({min})"),
93            ));
94        }
95        if max < avg {
96            return Err(crate::FxcpError::Config(
97                format!("GearChunker: max ({max}) must be >= avg ({avg})"),
98            ));
99        }
100        if !avg.is_power_of_two() {
101            return Err(crate::FxcpError::Config(format!(
102                "GearChunker: avg ({avg}) must be a power of two (try {})",
103                avg.next_power_of_two()
104            )));
105        }
106        Ok(Self { min, avg, max, normalization_level: 0 })
107    }
108
109    /// Set the normalization level for chunk size distribution.
110    ///
111    /// Returns an error if `level > 2`  --  higher levels shift masks beyond useful ranges.
112    /// Also rejects configurations where `logarithm2(avg) <= level`, which would produce
113    /// a zero mask_l (`MASKS[0]`) causing degenerate 1-byte chunks.
114    pub fn with_normalization(mut self, level: u8) -> crate::Result<Self> {
115        if level > 2 {
116            return Err(crate::FxcpError::Config(format!(
117                "GearChunker: normalization level {level} is out of range (max 2)"
118            )));
119        }
120        let bits = logarithm2(self.avg);
121        if level > 0 && bits <= level as u32 {
122            return Err(crate::FxcpError::Config(format!(
123                "GearChunker: avg {} (log2={bits}) is too small for normalization level {level}; \
124                 need avg >= {} to avoid degenerate zero mask",
125                self.avg,
126                1usize << (level as u32 + 1),
127            )));
128        }
129        self.normalization_level = level;
130        Ok(self)
131    }
132
133    /// Chunk an entire reader, returning all chunks with their BLAKE3 hashes.
134    ///
135    /// # Limits
136    /// Rejects inputs larger than 16 MB to prevent OOM. Use [`StreamingChunker`]
137    /// for arbitrarily large files.
138    #[deprecated(note = "Use StreamingChunker for files > 16 MB")]
139    pub fn chunk_reader<R: Read>(&self, mut reader: R) -> std::io::Result<Vec<Chunk>> {
140        let mut chunks = Vec::new();
141        let mut buf = Vec::with_capacity(self.max.min(crate::constants::CHUNK_READER_MAX_INPUT));
142        let mut read_buf = [0u8; crate::constants::CHUNK_READER_READ_BUFFER_SIZE];
143
144        loop {
145            let n = reader.read(&mut read_buf)?;
146            if n == 0 { break; }
147            buf.extend_from_slice(&read_buf[..n]);
148            if buf.len() > crate::constants::CHUNK_READER_MAX_INPUT {
149                return Err(std::io::Error::new(
150                    std::io::ErrorKind::InvalidInput,
151                    "chunk_reader: input exceeds 16 MB limit; use StreamingChunker instead",
152                ));
153            }
154        }
155
156        if buf.is_empty() {
157            return Ok(chunks);
158        }
159
160        let boundaries = self.find_boundaries(&buf);
161        let mut start = 0;
162        for end in boundaries {
163            let data = buf[start..end].to_vec();
164            let hash = blake3::hash(&data);
165            chunks.push(Chunk { data, hash });
166            start = end;
167        }
168
169        Ok(chunks)
170    }
171
172    /// Find chunk boundaries in a byte slice using gear hash.
173    ///
174    /// When `normalization_level == 0`, uses a single mask `(avg - 1)`  --  backward-compatible
175    /// with all existing data. When `> 0`, uses FastCDC dual-mask: a stricter mask_s below avg
176    /// and a relaxed mask_l above avg, tightening the chunk-size distribution.
177    pub fn find_boundaries(&self, data: &[u8]) -> Vec<usize> {
178        let mut boundaries = Vec::new();
179        let mut start = 0;
180        let mut hash: u64 = 0;
181
182        if self.normalization_level == 0 {
183            let mask = (self.avg - 1) as u64;
184            for i in 0..data.len() {
185                hash = hash.wrapping_shl(1).wrapping_add(GEAR_TABLE[data[i] as usize]);
186                let chunk_len = i - start + 1;
187                if chunk_len >= self.min && (hash & mask == 0 || chunk_len >= self.max) {
188                    boundaries.push(i + 1);
189                    start = i + 1;
190                    hash = 0;
191                }
192            }
193        } else {
194            let bits = logarithm2(self.avg);
195            let level = self.normalization_level as u32;
196            let mask_s = MASKS[(bits + level).min(63) as usize];
197            let mask_l = MASKS[bits.saturating_sub(level) as usize];
198            for i in 0..data.len() {
199                hash = hash.wrapping_shl(1).wrapping_add(GEAR_TABLE[data[i] as usize]);
200                let chunk_len = i - start + 1;
201                let active_mask = if chunk_len < self.avg { mask_s } else { mask_l };
202                if chunk_len >= self.min && (hash & active_mask == 0 || chunk_len >= self.max) {
203                    boundaries.push(i + 1);
204                    start = i + 1;
205                    hash = 0;
206                }
207            }
208        }
209
210        if start < data.len() {
211            boundaries.push(data.len());
212        }
213
214        boundaries
215    }
216
217    /// Chunk from a byte slice directly (avoids double-copy for in-memory data).
218    pub fn chunk_slice(&self, data: &[u8]) -> Vec<Chunk> {
219        if data.is_empty() {
220            return Vec::new();
221        }
222
223        let boundaries = self.find_boundaries(data);
224        let mut chunks = Vec::with_capacity(boundaries.len());
225        let mut start = 0;
226        for end in boundaries {
227            let slice = &data[start..end];
228            chunks.push(Chunk {
229                data: slice.to_vec(),
230                hash: blake3::hash(slice),
231            });
232            start = end;
233        }
234        chunks
235    }
236}
237
238/// Bit-mask table for FastCDC normalized chunking. MASKS[i] = (1u64 << i) - 1.
239/// MASKS[0] = u64::MAX (never-split sentinel) to guard against degenerate zero-mask
240/// if an unchecked index 0 is ever reached.
241const MASKS: [u64; 64] = {
242    let mut table = [0u64; 64];
243    table[0] = u64::MAX; // sentinel: never split (all bits set)
244    let mut i = 1usize;
245    while i < 64 {
246        table[i] = (1u64 << i) - 1;
247        i += 1;
248    }
249    table
250};
251
252/// Integer log2 for power-of-two values. Uses trailing_zeros which equals log2 for pow2.
253#[inline]
254fn logarithm2(value: usize) -> u32 {
255    value.trailing_zeros()
256}
257
258/// Deterministic random lookup table for gear hash.
259/// Generated from BLAKE3("gear-table-seed-{i}") for reproducibility.
260const GEAR_TABLE: [u64; 256] = {
261    // We use a compile-time deterministic table.
262    // These values were generated from: for i in 0..256 { blake3::hash(format!("gear-{i}").as_bytes()) truncated to u64 }
263    // For const evaluation, we embed the pre-computed values.
264    [
265        0x6b5f_f821_3c4a_e15d, 0x3e2c_4a59_81fb_d7c3, 0x9d17_e3f0_a264_58b9, 0xc4a8_6b2f_d903_7e41,
266        0x1f93_d5a4_27c8_b60e, 0x72e4_0b98_cf56_1da3, 0xa51c_83d7_640f_9eb2, 0x4de9_7a13_b85c_02f6,
267        0x8376_1ed5_4ca9_f0b8, 0x0ab4_62c1_9f7d_e384, 0xd648_f5a7_30b9_1c2e, 0x5c01_9de3_a872_4fb6,
268        0xe7ba_3f84_d215_c60a, 0x29d5_c416_7eb3_80f9, 0xf103_8a6d_54e9_27cb, 0x68cf_b1a2_e374_950d,
269        0xb42e_d789_16fc_a053, 0x0d71_4e36_c5a8_9bf2, 0x94ad_23f5_781c_d0e6, 0x3f86_cb17_a9e0_524d,
270        0xe210_7d93_f4b6_a8c1, 0x57c9_01a8_2de5_3f74, 0xab34_e6c2_809f_1db7, 0x1ea7_580d_c3b4_96f2,
271        0xc5f3_9a21_67de_0b48, 0x4018_d7b5_e2c9_a36f, 0x86ac_3f70_1d54_e8b2, 0xd961_c4a8_f30b_72e5,
272        0x237e_a5d1_b846_9fc0, 0x7cb2_10e9_6d3f_5a84, 0xf5d4_8c36_a1e7_0b29, 0x41a9_37cb_58f2_d6e0,
273        0xba65_fc12_8d07_a493, 0x0e38_b4d7_c9a1_526f, 0x97c2_608a_3e15_fdb4, 0x5a0d_e1f3_c478_b926,
274        0xc391_7b25_a0ec_4d18, 0x2cd6_a940_f587_13be, 0xf41b_3e82_d9c6_a075, 0x6807_d5c4_1ab9_2ef3,
275        0xad73_4f96_e201_c8b5, 0x15be_8a63_7dc4_f021, 0x89e1_c7d4_026b_3fa8, 0x34a6_12f5_cb89_70de,
276        0xe75c_d438_916a_b2c0, 0x50c3_a917_6ef4_d58b, 0xbc29_5d80_a3e1_47f6, 0x03f7_8e14_d5b2_c96a,
277        0xcb84_a3f6_1027_de59, 0x4610_dbc5_87f9_6a23, 0x91f5_26a8_e4d3_0cb7, 0xdf42_c519_73b8_ea04,
278        0x27be_9170_a4d6_3cf8, 0x7a03_e8d2_1fc5_b946, 0xf5c8_3da6_42b1_970e, 0x384d_76c1_ef02_5ab9,
279        0xc916_ab53_d478_e0f2, 0x14e2_c087_b935_6da4, 0xa8bf_31d4_52c0_7e19, 0x5d04_f928_3ea7_c1b6,
280        0xe261_8db5_c0f3_4a27, 0x2fb0_54c3_79e8_16da, 0x93c7_ae19_0d64_f285, 0x46da_0372_b5c1_8f9e,
281        0xb185_6fad_c832_04e7, 0x0c4e_d2b1_f697_a358, 0x7923_a8e5_4db0_c1f6, 0xd4f6_1cb7_28a3_950e,
282        0x2a81_e594_f3d0_467b, 0x67b3_0fc8_a512_de94, 0xfb48_c261_d7e9_30a5, 0x35d9_7a0e_8cb4_f123,
283        0xce14_b583_61d7_a9f0, 0x40af_e826_9d03_cb57, 0x9d63_24b1_f0c5_7e8a, 0x58c0_91f7_a43e_d2b6,
284        0xa27b_4dc9_1586_e0f3, 0x1fe6_30a4_cb79_8d52, 0x8c59_d7e2_a41b_063f, 0xd302_a5f6_19e8_4cb7,
285        0x27c4_f831_de90_ba65, 0x7a18_6dbc_43f5_0e29, 0xf6b5_c204_87da_31e9, 0x4be1_39d7_5c06_a8f2,
286        0xb09c_7ea3_c1d4_5b28, 0x0d57_a2e0_f4b3_96c1, 0x82e3_cb14_39a0_d7f5, 0xdf26_50a9_b7c4_138e,
287        0x369a_8d47_2ce1_f5b0, 0x7b04_f1d2_85a6_3ec9, 0xf8c1_a593_60d7_2b4e, 0x45b6_2ef0_9dc8_71a3,
288        0xbd73_c418_0a5f_e962, 0x01e8_3fb5_c297_d4a6, 0x9e54_d061_78bc_a3f2, 0xd2a9_1784_cb30_5fe6,
289        0x2c16_e5b3_a049_8df7, 0x73db_a826_5cf1_40b9, 0xfa07_31c4_e8bd_926e, 0x46c2_9f58_13d7_a4b0,
290        0xbe8d_5a01_7c43_e6f9, 0x05f4_c7b3_da21_908e, 0x8b31_0ed6_a5f8_4c27, 0xd7e6_9240_1cb3_f5a8,
291        0x2179_abd5_e804_3c6f, 0x7e04_68b1_c3d9_f527, 0xf2b5_d34a_810e_67c9, 0x4ca8_1f97_56e2_b0d3,
292        0xa163_c4e0_2db8_f975, 0x1dbe_70a3_f945_8c21, 0x894c_d517_a3e0_2fb6, 0xd600_a928_7fb4_c3e1,
293        0x23d7_8e45_b1c6_0af9, 0x78ab_f213_9cd0_6e47, 0xf43e_2db6_c581_a709, 0x4b91_c0d8_37a4_fe62,
294        0xb254_79a1_e06c_1db3, 0x0fc3_e687_54a9_b2d0, 0x9618_ad34_c2f7_605b, 0xdb75_4a02_1fc8_93e6,
295        0x24c1_f59a_8d36_70eb, 0x7986_0bd3_e4a2_cf18, 0xf10a_c845_3b67_d29e, 0x4ed7_31b9_a0fc_5624,
296        0xa342_9c06_d8b1_ef73, 0x1ab5_e0d4_6379_28cf, 0x876c_24f1_bea5_d038, 0xd493_5b17_02ce_a6f9,
297        0x21e8_b7c0_f43d_5a96, 0x7e5f_03a4_c912_d8b3, 0xf2c4_6e31_85a7_0bd9, 0x4db1_9258_c3f6_ae04,
298        0xaa76_cd83_1049_5fb2, 0x1503_81bf_e7d4_a2c6, 0x88d4_ae62_5b07_f139, 0xd629_43b5_90ec_78a1,
299        0x239e_f7c0_4db1_2a68, 0x7c41_5a92_e3b8_0fd4, 0xf0b2_c637_a905_81de, 0x45e7_39d4_1cba_f628,
300        0xb91c_8da0_6743_e5f1, 0x06a3_f278_db14_c905, 0x9250_b4e1_3dc7_0fa6, 0xdf87_6123_a4e9_cb50,
301        0x2a1c_d5f6_80b3_4e97, 0x75e0_a849_c327_1fb6, 0xf39b_0c24_d6e5_7a81, 0x4856_d1b3_2fa0_c7e9,
302        0xbc2d_9fe4_a178_3065, 0x01e4_738b_5dc6_a9f2, 0x8d97_b250_c41e_6fa3, 0xd06a_2e85_f1c3_94b7,
303        0x2cb1_d704_8e69_53fa, 0x7f46_a8c1_32bd_e019, 0xf3d2_140b_c5a6_789e, 0x4e09_6ba7_d8f3_c251,
304        0xab84_c3f0_1d27_95e6, 0x1671_ae52_c098_4bf3, 0x8a2d_f584_67b1_c039, 0xd5c0_3916_a2fd_8e74,
305        0x28a7_e4d3_5b10_cf69, 0x7d53_01be_c8a4_2f96, 0xf1e8_bc47_3d65_a20d, 0x4c34_5f92_e0b7_d1a8,
306        0xb0c9_a216_7d4e_f853, 0x0512_dc83_a9b0_674f, 0x99e6_7014_c253_bda8, 0xd42b_a5f0_8e97_31c6,
307        0x2794_c831_d5be_0af7, 0x7ae1_3f06_82c4_59d3, 0xf6bc_d245_1e73_a0b9, 0x4308_a7b9_c561_dfe2,
308        0xbd51_e460_3a98_7cf1, 0x02c6_89f3_e7a4_b510, 0x8e7d_b428_51c0_3fa9, 0xd1a0_67b5_2cf4_e893,
309        0x2e3f_c1d0_b586_49a7, 0x7b82_4ea3_c0d9_f715, 0xf715_930c_4ea2_d8b3, 0x42c8_b6a1_d350_7fe9,
310        0xaf54_0d38_91e7_c2b6, 0x1429_e8c5_a6f3_7d01, 0x87e3_51b6_dc0a_4f98, 0xd096_2c43_f5b8_a1e7,
311        0x2b41_f790_8ced_3a56, 0x76d8_a025_e1b4_cf93, 0xfaed_3481_5726_b0c9, 0x4f12_c9b7_a3d0_6e85,
312        0xb3a7_5e04_c819_df62, 0x0e6c_d1a2_3fb5_8740, 0x928b_47e1_d0f3_6ca5, 0xde50_bc36_14a9_87f2,
313        0x23c5_f049_6dbe_2a18, 0x7e98_3ad6_c104_5fb1, 0xf201_c78b_9ed3_a465, 0x4f74_152c_a3b8_e9d0,
314        0xa4e9_b863_17c0_5df2, 0x1136_4db7_e29a_08c5, 0x8dc2_a0e4_5b17_f639, 0xd05f_31b2_ce84_a976,
315        0x2d84_e6c7_a320_5fb1, 0x7013_9ba4_dc67_82e5, 0xfca8_d451_03b9_2e7f, 0x4965_0fc8_b7e2_a134,
316        0xb5da_c203_6e18_4fb7, 0x0241_7db6_a5c3_f809, 0x9ebc_a061_c834_d5f2, 0xd3f7_529c_01e6_ba48,
317        0x260c_bf21_94d5_3ea7, 0x7b93_e8a4_c072_1d5f, 0xf7c8_461d_3ba5_90e2, 0x42a1_d370_e50c_bf69,
318        0xbe5c_0f8b_7a43_d296, 0x0327_c4e6_d1b8_a5f0, 0x8f60_b932_54cd_e1a7, 0xdc15_ae03_b897_4c62,
319        0x21a8_73d5_0fc4_e2b9, 0x7cd4_e092_b318_5fa6, 0xf06b_2dc1_84e9_a735, 0x4d92_c1b4_67a3_08fe,
320        0xa247_56e0_1bcd_9f83, 0x1fb0_8d13_c429_e5a7, 0x83c5_f420_e976_1adb, 0xde08_31a7_b2c4_5f96,
321        0x2a9d_e654_0cb1_73f8, 0x7760_b9c3_f1de_a402, 0xfb15_48a6_3dc0_927e, 0x46c2_0f71_8ab5_e3d9,
322        0xba97_d438_c501_6eaf, 0x054e_a1c7_f8d2_3b60, 0x99f3_6c80_2db7_a5e4, 0xd420_b5f1_c368_9a27,
323        0x2dbc_4a06_e5f1_73c9, 0x7845_f3c2_0abd_e691, 0xf4d1_8e37_c962_ab05, 0x4106_c2a4_b5d8_3f7e,
324        0xad7b_3058_e4c9_1fa6, 0x12e4_cd81_7ba0_5643, 0x8e59_a4b2_0fc7_31d8, 0xd380_1fc5_e2b6_a749,
325        0x2a6f_e3c8_1594_bd07, 0x75b4_0a91_c8d6_3fe2, 0xf9c3_5d24_87b0_a16e, 0x4618_a2d7_3bcf_9058,
326        0xb087_c4f1_d259_6ea3, 0x0b5e_91a3_67c4_d8f2, 0x964d_b0e8_c123_7fa5, 0xd1a2_3f75_8ec0_b469,
327        0x2e89_c416_f5d3_a0b7, 0x73f0_5b84_29e6_cd13, 0xfc47_d2a1_b038_5e96, 0x410c_8fb3_e7a5_d248,
328        0xbe93_d714_a268_5fc0, 0x05a8_41c6_f3bd_927e, 0x9c74_e0b2_5d31_a8f6, 0xd1bf_56a0_83c4_7e29,
329    ]
330};
331
332/// Stateful streaming chunker  --  feeds data incrementally, emits chunks as boundaries are found.
333///
334/// Unlike `chunk_slice()` which needs the full data in memory, `StreamingChunker`
335/// processes data in arbitrary-sized reads and emits `Chunk` objects as gear-hash
336/// boundaries are detected.
337///
338/// # Example
339/// ```no_run
340/// # use fxcp_core::chunker::*;
341/// let mut sc = StreamingChunker::new(&GearChunker::new(2048, 65536, 2097152).unwrap());
342/// let mut file = std::fs::File::open("large_file.dat").unwrap();
343/// let mut buf = [0u8; 1_048_576]; // 1MB read buffer
344/// use std::io::Read;
345/// loop {
346///     let n = file.read(&mut buf).unwrap();
347///     if n == 0 { break; }
348///     for chunk in sc.feed(&buf[..n]) {
349///         // Process each chunk (upload, hash, etc.)
350///     }
351/// }
352/// let (final_chunk, _file_hash) = sc.finish();
353/// ```
354pub struct StreamingChunker {
355    min: usize,
356    avg: usize,
357    max: usize,
358    normalization_level: u8,
359    mask: u64,
360    mask_s: u64,
361    mask_l: u64,
362    buffer: Vec<u8>,
363    hash: u64,
364    file_hasher: blake3::Hasher,
365}
366
367impl StreamingChunker {
368    /// Create a new streaming chunker from a `GearChunker` configuration.
369    pub fn new(chunker: &GearChunker) -> Self {
370        let bits = logarithm2(chunker.avg);
371        let level = chunker.normalization_level as u32;
372        let mask_s = MASKS[(bits + level).min(63) as usize];
373        let mask_l = MASKS[bits.saturating_sub(level) as usize];
374        Self {
375            min: chunker.min,
376            avg: chunker.avg,
377            max: chunker.max,
378            normalization_level: chunker.normalization_level,
379            mask: (chunker.avg - 1) as u64,
380            mask_s,
381            mask_l,
382            buffer: Vec::with_capacity(chunker.max),
383            hash: 0,
384            file_hasher: blake3::Hasher::new(),
385        }
386    }
387
388    /// Feed data into the chunker and return any completed chunks.
389    pub fn feed(&mut self, data: &[u8]) -> Vec<Chunk> {
390        let mut chunks = Vec::new();
391        self.file_hasher.update(data);
392
393        for &byte in data {
394            self.buffer.push(byte);
395            self.hash = self.hash.wrapping_shl(1).wrapping_add(GEAR_TABLE[byte as usize]);
396
397            let chunk_len = self.buffer.len();
398            let active_mask = if self.normalization_level == 0 {
399                self.mask
400            } else if chunk_len < self.avg {
401                self.mask_s
402            } else {
403                self.mask_l
404            };
405            if chunk_len >= self.min && (self.hash & active_mask == 0 || chunk_len >= self.max) {
406                let chunk_data = std::mem::replace(&mut self.buffer, Vec::with_capacity(self.max));
407                let chunk_hash = blake3::hash(&chunk_data);
408                chunks.push(Chunk { data: chunk_data, hash: chunk_hash });
409                self.hash = 0;
410            }
411        }
412
413        chunks
414    }
415
416    /// Finish the stream and return any trailing data as a final chunk.
417    ///
418    /// Returns `None` if no data remains after the last boundary.
419    /// Also returns the whole-file BLAKE3 hash.
420    pub fn finish(self) -> (Option<Chunk>, blake3::Hash) {
421        let file_hash = self.file_hasher.finalize();
422        if self.buffer.is_empty() {
423            (None, file_hash)
424        } else {
425            let chunk_hash = blake3::hash(&self.buffer);
426            (Some(Chunk { data: self.buffer, hash: chunk_hash }), file_hash)
427        }
428    }
429
430    /// Get the current buffer size (bytes accumulated for current chunk).
431    pub fn buffered(&self) -> usize {
432        self.buffer.len()
433    }
434}
435
436#[cfg(test)]
437mod tests {
438    #![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
439    use super::*;
440
441    #[test]
442    fn test_deterministic_chunking() {
443        let data = vec![42u8; 200_000];
444        let chunker = GearChunker::default();
445        let chunks1 = chunker.chunk_slice(&data);
446        let chunks2 = chunker.chunk_slice(&data);
447
448        assert_eq!(chunks1.len(), chunks2.len());
449        for (a, b) in chunks1.iter().zip(chunks2.iter()) {
450            assert_eq!(a.hash, b.hash);
451            assert_eq!(a.data.len(), b.data.len());
452        }
453    }
454
455    #[test]
456    fn test_min_max_boundaries() {
457        let chunker = GearChunker::new(1024, 4096, 8192).unwrap();
458        let data = vec![0xABu8; 50_000];
459        let chunks = chunker.chunk_slice(&data);
460
461        let total: usize = chunks.iter().map(|c| c.data.len()).sum();
462        assert_eq!(total, data.len());
463
464        for (i, chunk) in chunks.iter().enumerate() {
465            // Trailing chunk may be smaller than min
466            if i < chunks.len() - 1 {
467                assert!(chunk.data.len() >= 1024, "chunk too small: {}", chunk.data.len());
468            }
469            assert!(chunk.data.len() <= 8192, "chunk too large: {}", chunk.data.len());
470        }
471    }
472
473    #[test]
474    fn test_small_file_single_chunk() {
475        let chunker = GearChunker::default();
476        let data = vec![0x42u8; 100]; // < min (2KB)
477        let chunks = chunker.chunk_slice(&data);
478        assert_eq!(chunks.len(), 1);
479        assert_eq!(chunks[0].data.len(), 100);
480    }
481
482    #[test]
483    fn test_empty_input() {
484        let chunker = GearChunker::default();
485        let chunks = chunker.chunk_slice(&[]);
486        assert!(chunks.is_empty());
487    }
488
489    #[test]
490    #[allow(deprecated)]
491    fn test_empty_reader() {
492        let chunker = GearChunker::default();
493        let chunks = chunker.chunk_reader(std::io::empty()).unwrap();
494        assert!(chunks.is_empty());
495    }
496
497    #[test]
498    fn test_rolling_hash_stability() {
499        // Inserting a byte early should only affect nearby chunks
500        let chunker = GearChunker::new(512, 4096, 16384).unwrap();
501        let mut data_a = vec![0u8; 100_000];
502        // Fill with pseudo-random data for realistic chunking
503        for (i, b) in data_a.iter_mut().enumerate() {
504            *b = (i.wrapping_mul(0x9E3779B9) >> 24) as u8;
505        }
506        let mut data_b = data_a.clone();
507        // Insert a single byte near the start
508        data_b.insert(500, 0xFF);
509
510        let chunks_a = chunker.chunk_slice(&data_a);
511        let chunks_b = chunker.chunk_slice(&data_b);
512
513        // Most chunks after the insertion point should be identical
514        // Count matching hashes (by value, ignoring position)
515        let hashes_a: std::collections::HashSet<[u8; 32]> =
516            chunks_a.iter().map(|c| *c.hash.as_bytes()).collect();
517        let hashes_b: std::collections::HashSet<[u8; 32]> =
518            chunks_b.iter().map(|c| *c.hash.as_bytes()).collect();
519        let common = hashes_a.intersection(&hashes_b).count();
520
521        // At least 50% of chunks should be shared (gear hash resynchronizes quickly)
522        let max_chunks = chunks_a.len().max(chunks_b.len());
523        assert!(common * 2 >= max_chunks,
524            "too few common chunks: {}/{}", common, max_chunks);
525    }
526
527    #[test]
528    #[allow(deprecated)]
529    fn test_chunk_reader_matches_slice() {
530        let chunker = GearChunker::default();
531        let data = vec![0x55u8; 150_000];
532        let from_slice = chunker.chunk_slice(&data);
533        let from_reader = chunker.chunk_reader(std::io::Cursor::new(&data)).unwrap();
534
535        assert_eq!(from_slice.len(), from_reader.len());
536        for (a, b) in from_slice.iter().zip(from_reader.iter()) {
537            assert_eq!(a.hash, b.hash);
538        }
539    }
540
541    #[test]
542    fn test_blake3_hashes_correct() {
543        let chunker = GearChunker::default();
544        let data = b"hello world, this is a test of the chunking system";
545        let chunks = chunker.chunk_slice(data);
546        assert_eq!(chunks.len(), 1); // small enough for single chunk
547        assert_eq!(chunks[0].hash, blake3::hash(data));
548    }
549
550    #[cfg(feature = "cloud")]
551    #[test]
552    fn test_s3_wan_preset_values() {
553        assert_eq!(super::S3_WAN.min, 5 * 1024 * 1024);
554        assert_eq!(super::S3_WAN.avg, 8 * 1024 * 1024);
555        assert_eq!(super::S3_WAN.max, 64 * 1024 * 1024);
556    }
557
558    #[cfg(feature = "cloud")]
559    #[test]
560    fn test_s3_lan_preset_values() {
561        assert_eq!(super::S3_LAN.min, 5 * 1024 * 1024);
562        assert_eq!(super::S3_LAN.avg, 16 * 1024 * 1024);
563        assert_eq!(super::S3_LAN.max, 128 * 1024 * 1024);
564    }
565
566    #[cfg(feature = "cloud")]
567    #[test]
568    fn test_s3_archive_preset_values() {
569        assert_eq!(super::S3_ARCHIVE.min, 5 * 1024 * 1024);
570        assert_eq!(super::S3_ARCHIVE.avg, 32 * 1024 * 1024);
571        assert_eq!(super::S3_ARCHIVE.max, 256 * 1024 * 1024);
572    }
573
574    #[cfg(feature = "cloud")]
575    #[test]
576    fn test_s3_wan_chunk_distribution() {
577        let mut data = vec![0u8; 16 * 1024 * 1024];
578        for (i, b) in data.iter_mut().enumerate() {
579            *b = (i.wrapping_mul(0x9E3779B9) >> 24) as u8;
580        }
581        let chunks = super::S3_WAN.chunk_slice(&data);
582        let total: usize = chunks.iter().map(|c| c.data.len()).sum();
583        assert_eq!(total, data.len(), "total chunk data must equal input");
584        assert!(chunks.len() >= 1 && chunks.len() <= 8,
585            "expected ~2 chunks for 16MB/8MB avg, got {}", chunks.len());
586    }
587
588    #[test]
589    fn test_streaming_chunker_matches_slice() {
590        let chunker = GearChunker::new(64, 256, 1024).unwrap();
591        let mut data = vec![0u8; 8192];
592        for (i, b) in data.iter_mut().enumerate() {
593            *b = (i.wrapping_mul(0x9E3779B9) >> 24) as u8;
594        }
595
596        let slice_chunks = chunker.chunk_slice(&data);
597
598        let mut sc = StreamingChunker::new(&chunker);
599        let mut stream_chunks = Vec::new();
600        for chunk_buf in data.chunks(512) {
601            stream_chunks.extend(sc.feed(chunk_buf));
602        }
603        let (final_chunk, _file_hash) = sc.finish();
604        if let Some(c) = final_chunk {
605            stream_chunks.push(c);
606        }
607
608        assert_eq!(slice_chunks.len(), stream_chunks.len(),
609            "chunk count mismatch: slice={}, stream={}", slice_chunks.len(), stream_chunks.len());
610        for (i, (s, st)) in slice_chunks.iter().zip(stream_chunks.iter()).enumerate() {
611            assert_eq!(s.data, st.data, "chunk {i} data mismatch");
612            assert_eq!(s.hash, st.hash, "chunk {i} hash mismatch");
613        }
614    }
615
616    #[test]
617    fn test_streaming_chunker_single_feed() {
618        let chunker = GearChunker::new(64, 256, 1024).unwrap();
619        let data = vec![42u8; 4096];
620
621        let slice_chunks = chunker.chunk_slice(&data);
622        let mut sc = StreamingChunker::new(&chunker);
623        let mut stream_chunks = sc.feed(&data);
624        let (final_chunk, _) = sc.finish();
625        if let Some(c) = final_chunk { stream_chunks.push(c); }
626
627        assert_eq!(slice_chunks.len(), stream_chunks.len());
628    }
629
630    #[test]
631    fn test_streaming_chunker_byte_by_byte() {
632        let chunker = GearChunker::new(64, 256, 1024).unwrap();
633        let data = vec![0u8; 2048];
634
635        let slice_chunks = chunker.chunk_slice(&data);
636        let mut sc = StreamingChunker::new(&chunker);
637        let mut stream_chunks = Vec::new();
638        for &b in &data {
639            stream_chunks.extend(sc.feed(&[b]));
640        }
641        let (final_chunk, _) = sc.finish();
642        if let Some(c) = final_chunk { stream_chunks.push(c); }
643
644        assert_eq!(slice_chunks.len(), stream_chunks.len());
645    }
646
647    #[test]
648    fn test_streaming_chunker_file_hash() {
649        let chunker = GearChunker::new(64, 256, 1024).unwrap();
650        let data = b"Hello world, this is streaming chunker test data!";
651
652        let mut sc = StreamingChunker::new(&chunker);
653        sc.feed(data);
654        let (_, file_hash) = sc.finish();
655
656        assert_eq!(file_hash, blake3::hash(data));
657    }
658
659    #[test]
660    fn test_streaming_chunker_empty() {
661        let chunker = GearChunker::new(64, 256, 1024).unwrap();
662        let mut sc = StreamingChunker::new(&chunker);
663        let chunks = sc.feed(&[]);
664        assert!(chunks.is_empty());
665        let (final_chunk, _) = sc.finish();
666        assert!(final_chunk.is_none());
667    }
668
669    #[test]
670    fn test_pow2_enforcement() {
671        assert!(GearChunker::new(1024, 4096, 8192).is_ok());
672        assert!(GearChunker::new(1024, 65536, 2097152).is_ok());
673        let err = GearChunker::new(1024, 5000, 8192).unwrap_err();
674        let msg = err.to_string();
675        assert!(msg.contains("power of two"), "expected pow2 hint in: {msg}");
676        assert!(msg.contains("8192"), "expected next_power_of_two suggestion in: {msg}");
677    }
678
679    #[test]
680    fn test_with_normalization_validation() {
681        assert!(GearChunker::new(1024, 4096, 8192).unwrap().with_normalization(0).is_ok());
682        assert!(GearChunker::new(1024, 4096, 8192).unwrap().with_normalization(1).is_ok());
683        assert!(GearChunker::new(1024, 4096, 8192).unwrap().with_normalization(2).is_ok());
684        let err = GearChunker::new(1024, 4096, 8192).unwrap().with_normalization(3).unwrap_err();
685        assert!(err.to_string().contains("out of range"), "expected 'out of range' in: {err}");
686    }
687
688    #[test]
689    fn test_chunker_masks_no_zero() {
690        assert_eq!(MASKS[0], u64::MAX, "MASKS[0] must be u64::MAX sentinel, not 0");
691        for i in 1..64u32 {
692            assert_ne!(MASKS[i as usize], 0, "MASKS[{i}] must not be zero");
693            assert_eq!(MASKS[i as usize], (1u64 << i) - 1, "MASKS[{i}] formula mismatch");
694        }
695
696        // Verify with_normalization rejects configs that would index MASKS[0]
697        // avg=4 -> bits=2, level=2 -> bits.saturating_sub(level) = 0 -> MASKS[0]
698        let result = GearChunker::new(2, 4, 16).unwrap().with_normalization(2);
699        assert!(result.is_err(), "avg=4 with level=2 should be rejected (would use MASKS[0])");
700
701        // avg=2 -> bits=1, level=1 -> bits.saturating_sub(level) = 0 -> MASKS[0]
702        let result = GearChunker::new(1, 2, 8).unwrap().with_normalization(1);
703        assert!(result.is_err(), "avg=2 with level=1 should be rejected (would use MASKS[0])");
704
705        // Valid configs: bits > level
706        for &avg in &[8usize, 16, 64, 256, 4096, 65536] {
707            for level in 0..=2u8 {
708                let bits = logarithm2(avg);
709                let result = GearChunker::new(1, avg, avg * 4).unwrap().with_normalization(level);
710                if level == 0 || bits > level as u32 {
711                    assert!(result.is_ok(), "avg={avg} level={level} should be valid");
712                    if level > 0 {
713                        let mask_l_idx = bits.saturating_sub(level as u32) as usize;
714                        assert_ne!(MASKS[mask_l_idx], 0, "mask_l for avg={avg} level={level} must not be zero");
715                    }
716                }
717            }
718        }
719    }
720
721    #[test]
722    fn test_normalization_default() {
723        assert_eq!(GearChunker::default().normalization_level, 0);
724        assert_eq!(GearChunker::new(2048, 65536, 2097152).unwrap().normalization_level, 0);
725    }
726
727    fn chunk_size_stats(sizes: &[usize]) -> (f64, f64, f64) {
728        if sizes.is_empty() {
729            return (0.0, 0.0, 0.0);
730        }
731        let n = sizes.len() as f64;
732        let mean = sizes.iter().map(|&s| s as f64).sum::<f64>() / n;
733        let variance = sizes.iter().map(|&s| {
734            let diff = s as f64 - mean;
735            diff * diff
736        }).sum::<f64>() / n;
737        let stddev = variance.sqrt();
738        let cv = if mean > 0.0 { stddev / mean } else { 0.0 };
739        (mean, stddev, cv)
740    }
741
742    fn legacy_find_boundaries(min: usize, avg: usize, max: usize, data: &[u8]) -> Vec<usize> {
743        let mask = (avg - 1) as u64;
744        let mut boundaries = Vec::new();
745        let mut start = 0;
746        let mut hash: u64 = 0;
747        for i in 0..data.len() {
748            hash = hash.wrapping_shl(1).wrapping_add(GEAR_TABLE[data[i] as usize]);
749            let chunk_len = i - start + 1;
750            if chunk_len >= min && (hash & mask == 0 || chunk_len >= max) {
751                boundaries.push(i + 1);
752                start = i + 1;
753                hash = 0;
754            }
755        }
756        if start < data.len() {
757            boundaries.push(data.len());
758        }
759        boundaries
760    }
761
762    #[test]
763    fn test_normalization_zero_identical_to_legacy() {
764        let chunker = GearChunker::new(64, 256, 1024).unwrap();
765        let mut data = vec![0u8; 8192];
766        for (i, b) in data.iter_mut().enumerate() {
767            *b = (i.wrapping_mul(0x9E3779B9_usize) >> 24) as u8;
768        }
769        let new_bounds = chunker.find_boundaries(&data);
770        let old_bounds = legacy_find_boundaries(64, 256, 1024, &data);
771        assert_eq!(new_bounds, old_bounds, "normalization=0 must produce identical boundaries to legacy");
772    }
773
774    #[test]
775    fn test_normalization_one_produces_valid_chunks() {
776        let chunker = GearChunker::new(64, 256, 1024).unwrap()
777            .with_normalization(1).unwrap();
778        let mut data = vec![0u8; 8192];
779        for (i, b) in data.iter_mut().enumerate() {
780            *b = (i.wrapping_mul(0x9E3779B9_usize) >> 24) as u8;
781        }
782        let boundaries = chunker.find_boundaries(&data);
783        assert!(!boundaries.is_empty(), "must find at least one boundary");
784        assert_eq!(*boundaries.last().unwrap(), data.len(), "must cover all input");
785        let mut start = 0;
786        for &end in &boundaries {
787            let chunk_len = end - start;
788            assert!(chunk_len <= 1024, "chunk exceeds max: {chunk_len}");
789            start = end;
790        }
791    }
792
793    #[test]
794    fn test_normalization_zero_two_produce_chunks() {
795        let data: Vec<u8> = (0..4096_usize).map(|i| (i.wrapping_mul(251) >> 3) as u8).collect();
796        for level in [0u8, 1, 2] {
797            let chunker = GearChunker::new(64, 256, 1024).unwrap()
798                .with_normalization(level).unwrap();
799            let boundaries = chunker.find_boundaries(&data);
800            assert!(!boundaries.is_empty(), "level={level}: must produce boundaries");
801            assert_eq!(*boundaries.last().unwrap(), data.len(), "level={level}: must cover all input");
802        }
803    }
804
805    #[test]
806    fn test_chunk_distribution_quality() {
807        let chunker = GearChunker::new(2048, 65536, 2097152).unwrap();
808        let mut data = vec![0u8; 10 * 1024 * 1024];
809        let mut state: u64 = 0xDEAD_BEEF_CAFE_1337;
810        for byte in data.iter_mut() {
811            state = state
812                .wrapping_mul(6_364_136_223_846_793_005)
813                .wrapping_add(1_442_695_040_888_963_407);
814            *byte = (state >> 56) as u8;
815        }
816
817        let chunks = chunker.chunk_slice(&data);
818        let sizes: Vec<usize> = chunks.iter().map(|c| c.data.len()).collect();
819
820        let total: usize = sizes.iter().sum();
821        assert_eq!(total, data.len(), "chunks must cover all input bytes");
822
823        for (i, &sz) in sizes.iter().enumerate() {
824            if i < sizes.len() - 1 {
825                assert!(sz >= 2048, "non-trailing chunk below min: {sz}");
826            }
827            assert!(sz <= 2097152, "chunk exceeds max: {sz}");
828        }
829
830        let (mean, stddev, cv) = chunk_size_stats(&sizes);
831        eprintln!("normalization=0 distribution: chunks={}, mean={:.0}, stddev={:.0}, cv={:.3}",
832            sizes.len(), mean, stddev, cv);
833
834        assert!(!sizes.is_empty(), "must produce at least one chunk");
835    }
836
837    #[test]
838    fn test_distribution_cv_normalization_one_tighter() {
839        let mut data = vec![0u8; 10 * 1024 * 1024];
840        let mut state: u64 = 0xDEAD_BEEF_CAFE_1337;
841        for byte in data.iter_mut() {
842            state = state
843                .wrapping_mul(6_364_136_223_846_793_005)
844                .wrapping_add(1_442_695_040_888_963_407);
845            *byte = (state >> 56) as u8;
846        }
847
848        let norm0 = GearChunker::new(2048, 65536, 2097152).unwrap();
849        let norm1 = GearChunker::new(2048, 65536, 2097152).unwrap()
850            .with_normalization(1).unwrap();
851
852        let sizes0: Vec<usize> = norm0.chunk_slice(&data).into_iter().map(|c| c.data.len()).collect();
853        let sizes1: Vec<usize> = norm1.chunk_slice(&data).into_iter().map(|c| c.data.len()).collect();
854
855        let (mean0, stddev0, cv0) = chunk_size_stats(&sizes0);
856        let (mean1, stddev1, cv1) = chunk_size_stats(&sizes1);
857
858        eprintln!("normalization=0: chunks={}, mean={:.0}, stddev={:.0}, cv={:.3}",
859            sizes0.len(), mean0, stddev0, cv0);
860        eprintln!("normalization=1: chunks={}, mean={:.0}, stddev={:.0}, cv={:.3}",
861            sizes1.len(), mean1, stddev1, cv1);
862
863        assert!(cv1 < 0.75,
864            "normalization=1 CV {cv1:.3} must be < 0.75 (tighter distribution)");
865        assert!(cv1 < cv0,
866            "normalization=1 CV {cv1:.3} must be tighter than normalization=0 CV {cv0:.3}");
867    }
868
869    fn streaming_collect(chunker: &GearChunker, data: &[u8], feed_size: usize) -> Vec<Vec<u8>> {
870        let mut sc = StreamingChunker::new(chunker);
871        let mut out = Vec::new();
872        for chunk in data.chunks(feed_size) {
873            for c in sc.feed(chunk) {
874                out.push(c.data);
875            }
876        }
877        let (final_chunk, _) = sc.finish();
878        if let Some(c) = final_chunk {
879            out.push(c.data);
880        }
881        out
882    }
883
884    #[test]
885    fn test_streaming_matches_slice_normalization_one() {
886        let chunker = GearChunker::new(64, 256, 1024).unwrap()
887            .with_normalization(1).unwrap();
888        let mut data = vec![0u8; 8192];
889        for (i, b) in data.iter_mut().enumerate() {
890            *b = (i.wrapping_mul(0x9E3779B9_usize) >> 24) as u8;
891        }
892        let slice_data: Vec<Vec<u8>> = chunker.chunk_slice(&data).into_iter().map(|c| c.data).collect();
893        let stream_data = streaming_collect(&chunker, &data, 512);
894        assert_eq!(slice_data, stream_data, "normalization=1: streaming must match slice");
895    }
896
897    #[test]
898    fn test_streaming_matches_slice_normalization_two() {
899        let chunker = GearChunker::new(64, 256, 1024).unwrap()
900            .with_normalization(2).unwrap();
901        let mut data = vec![0u8; 8192];
902        for (i, b) in data.iter_mut().enumerate() {
903            *b = (i.wrapping_mul(0x9E3779B9_usize) >> 24) as u8;
904        }
905        let slice_data: Vec<Vec<u8>> = chunker.chunk_slice(&data).into_iter().map(|c| c.data).collect();
906        let stream_data = streaming_collect(&chunker, &data, 512);
907        assert_eq!(slice_data, stream_data, "normalization=2: streaming must match slice");
908    }
909
910    #[test]
911    #[allow(deprecated)]
912    fn test_chunk_reader_large_file_guard() {
913        let chunker = GearChunker::default();
914        // Create a reader that provides >16MB of data
915        let data = vec![0xABu8; 17 * 1024 * 1024]; // 17MB
916        let result = chunker.chunk_reader(std::io::Cursor::new(&data));
917        let err = match result {
918            Err(e) => e,
919            Ok(_) => panic!("chunk_reader must reject inputs > 16MB"),
920        };
921        assert_eq!(err.kind(), std::io::ErrorKind::InvalidInput);
922        assert!(err.to_string().contains("16 MB"), "error should mention 16 MB limit: {err}");
923    }
924
925    #[test]
926    fn test_streaming_byte_by_byte_normalization_one() {
927        let chunker = GearChunker::new(64, 256, 1024).unwrap()
928            .with_normalization(1).unwrap();
929        let data = vec![0xABu8; 2048];
930        let slice_data: Vec<Vec<u8>> = chunker.chunk_slice(&data).into_iter().map(|c| c.data).collect();
931        let stream_data = streaming_collect(&chunker, &data, 1);
932        assert_eq!(slice_data, stream_data, "byte-by-byte normalization=1 must match slice");
933    }
934}
fxcp_core/chunker.rs

fxcp_core/
chunker.rs
