1use crate::traits::BlockStore;
14use async_trait::async_trait;
15use dashmap::DashMap;
16use ipfrs_core::{Block, Cid, Error, Result};
17use parking_lot::RwLock;
18use std::sync::Arc;
19
20#[derive(Debug, Clone)]
22pub struct ChunkingConfig {
23 pub min_chunk_size: usize,
25 pub target_chunk_size: usize,
27 pub max_chunk_size: usize,
29 pub hash_mask: u32,
31}
32
33impl Default for ChunkingConfig {
34 fn default() -> Self {
35 Self {
36 min_chunk_size: 256 * 1024, target_chunk_size: 1024 * 1024, max_chunk_size: 4 * 1024 * 1024, hash_mask: 0xFFFF, }
41 }
42}
43
44impl ChunkingConfig {
45 pub fn small() -> Self {
47 Self {
48 min_chunk_size: 64 * 1024, target_chunk_size: 256 * 1024, max_chunk_size: 1024 * 1024, hash_mask: 0x3FFF, }
53 }
54
55 pub fn large() -> Self {
57 Self {
58 min_chunk_size: 1024 * 1024, target_chunk_size: 4 * 1024 * 1024, max_chunk_size: 16 * 1024 * 1024, hash_mask: 0x1FFFF, }
63 }
64}
65
66#[derive(Debug, Clone)]
68struct ChunkMeta {
69 cid: Cid,
71 ref_count: usize,
73 size: usize,
75}
76
77#[derive(Debug, Clone)]
79struct BlockManifest {
80 original_size: usize,
82 chunks: Vec<Cid>,
84}
85
86#[derive(Debug, Clone, Default)]
88pub struct DedupStats {
89 pub blocks_stored: usize,
91 pub bytes_original: usize,
93 pub bytes_stored: usize,
95 pub unique_chunks: usize,
97 pub duplicate_chunks_avoided: usize,
99}
100
101impl DedupStats {
102 pub fn dedup_ratio(&self) -> f64 {
104 if self.bytes_original == 0 {
105 return 0.0;
106 }
107 1.0 - (self.bytes_stored as f64 / self.bytes_original as f64)
108 }
109
110 pub fn bytes_saved(&self) -> usize {
112 self.bytes_original.saturating_sub(self.bytes_stored)
113 }
114
115 pub fn avg_chunk_size(&self) -> usize {
117 if self.unique_chunks == 0 {
118 return 0;
119 }
120 self.bytes_stored / self.unique_chunks
121 }
122}
123
124#[allow(dead_code)]
127const GEAR: [u64; 256] = [
128 0x5c95c078, 0x22408989, 0x2d48a214, 0x12842087, 0x530f8afb, 0x2aaa3f86, 0x7f1bd89f, 0x62534467,
129 0x22c4b83b, 0x3e36d3e7, 0x4c9fa05b, 0x0b20f0e3, 0x441c8a8c, 0x7cc27988, 0x5505c6c0, 0x3c9ae0da,
130 0x153e46cd, 0x0d05f5b5, 0x51c9c3b5, 0x02e57b86, 0x74a8d4ba, 0x6f16cbb5, 0x2ffc27ea, 0x5fa83e0f,
131 0x75ab67e2, 0x3ff15813, 0x2ec58ac7, 0x6f1f0520, 0x0c5d7dba, 0x4a9f5e76, 0x4ec58e64, 0x6a470c8e,
132 0x40edf2ca, 0x1a1c0c8d, 0x4e32e5e4, 0x6c7a7fda, 0x4b3be9e4, 0x64d8e67b, 0x2ef8ad98, 0x34d9f7e5,
133 0x7e7e4a36, 0x1a1c54d1, 0x5e2a9e7a, 0x3e5f0a8e, 0x0e01d1a0, 0x1f31aa27, 0x049c9e3e, 0x7c38f56e,
134 0x4b8d9ef0, 0x0b9c4d05, 0x55f59f0d, 0x3e8e02ae, 0x25c46f84, 0x6e6fdc6f, 0x440ae4a7, 0x3e38a0e6,
135 0x5b96c3d1, 0x72a06105, 0x52cd5e2d, 0x3d015fb3, 0x4d7c7064, 0x1c8c169c, 0x5c95e834, 0x0c4d9d42,
136 0x3c9c8ea3, 0x10a5d9d6, 0x7dcb9d63, 0x3ecf9e96, 0x1f5c9e5f, 0x7e7854c5, 0x48a05ae3, 0x0c4e9419,
137 0x6b5c9b6f, 0x7e1a6dc0, 0x3b8f9fe8, 0x6f6e8e3f, 0x39f48adb, 0x7b8d9e72, 0x29e18dc5, 0x7e6c3fc4,
138 0x5d9c4ab8, 0x1f6e9dc2, 0x3e8f9fc3, 0x7d9c8ea6, 0x0e1f8d9c, 0x5f9d8e72, 0x3e9f8dcb, 0x7d8e9f72,
139 0x2f9d8ea5, 0x6e8f9dc4, 0x3d9f8ec5, 0x7e8d9f63, 0x1f9e8dc3, 0x6d8f9ec4, 0x3e9d8fc5, 0x7d9e8f62,
140 0x2e9f8dc4, 0x6f8d9ec5, 0x3d9e8fc3, 0x7e9d8f64, 0x1f8e9dc5, 0x6e9f8dc4, 0x3d8e9fc5, 0x7d9f8e63,
141 0x2f8d9ec4, 0x6e8f9dc5, 0x3e9d8fc4, 0x7d8e9f65, 0x1f9d8ec5, 0x6d9f8dc4, 0x3e8d9fc5, 0x7e9f8d62,
142 0x2d8e9fc4, 0x6f9d8ec5, 0x3d8f9dc4, 0x7e8d9f66, 0x1e9f8dc5, 0x6d8e9fc4, 0x3f9d8ec5, 0x7d9e8f61,
143 0x2f9d8ec4, 0x6e8d9fc5, 0x3d9f8dc4, 0x7e8f9d67, 0x1f8d9ec5, 0x6e9d8fc4, 0x3d8e9fc5, 0x7f9d8e60,
144 0x2e8f9dc4, 0x6f9e8dc5, 0x3d8d9fc4, 0x7e9f8d68, 0x1d9e8fc5, 0x6f8d9ec4, 0x3e9f8dc5, 0x7d8e9f5f,
145 0x2f8e9dc4, 0x6d9f8ec5, 0x3e8d9fc4, 0x7f9e8d69, 0x1f9d8ec5, 0x6e8f9dc4, 0x3d9e8fc5, 0x7e8d9f5e,
146 0x2d9f8ec4, 0x6f8e9dc5, 0x3d8f9fc4, 0x7e9d8e6a, 0x1e8f9dc5, 0x6d9e8fc4, 0x3f8d9ec5, 0x7d9f8e5d,
147 0x2f8d9fc4, 0x6e9f8ec5, 0x3d8e9dc4, 0x7f8d9e6b, 0x1f8e9fc5, 0x6e8d9ec4, 0x3d9f8fc5, 0x7e9e8d5c,
148 0x2e9d8fc4, 0x6f8e9dc5, 0x3e8f9ec4, 0x7d9e8f6c, 0x1f9e8dc5, 0x6d8f9fc4, 0x3e9d8ec5, 0x7d8f9e5b,
149 0x2f9e8dc4, 0x6e8d9fc5, 0x3d9f8ec4, 0x7e8f9d6d, 0x1e9d8fc5, 0x6f8e9dc4, 0x3d8f9ec5, 0x7e9d8f5a,
150 0x2d8f9ec4, 0x6e9d8fc5, 0x3f8e9dc4, 0x7d9f8e6e, 0x1f8d9fc5, 0x6e9e8dc4, 0x3d8f9fc5, 0x7f8e9d59,
151 0x2e8d9fc4, 0x6f9e8dc5, 0x3d9f8ec4, 0x7e8d9f6f, 0x1d9f8ec5, 0x6f8d9dc4, 0x3e8e9fc5, 0x7d9f8e58,
152 0x2f8e9fc4, 0x6d9f8dc5, 0x3e8d9ec4, 0x7f9e8d70, 0x1f8e9dc5, 0x6d8f9ec4, 0x3f9d8fc5, 0x7e8f9d57,
153 0x2d9e8fc4, 0x6f8e9dc5, 0x3d8f9ec4, 0x7e9d8f71, 0x1e9f8dc5, 0x6f8d9ec4, 0x3d9e8fc5, 0x7f8d9e56,
154 0x2f8d9ec4, 0x6e9f8dc5, 0x3e8d9fc4, 0x7d9e8f72, 0x1f9d8fc5, 0x6e8f9dc4, 0x3d8e9fc5, 0x7e9f8d55,
155 0x2e8f9fc4, 0x6d9e8dc5, 0x3f8d9ec4, 0x7e8f9d73, 0x1d9f8fc5, 0x6f8e9dc4, 0x3e8d9fc5, 0x7d8f9e54,
156 0x2f9e8dc4, 0x6e8f9fc5, 0x3d9d8ec4, 0x7f8e9d74, 0x1e8d9fc5, 0x6d9f8ec4, 0x3f8e9dc5, 0x7e9d8f53,
157 0x2d8e9fc4, 0x6f9d8ec5, 0x3d8f9fc4, 0x7e9f8d75, 0x1f8d9ec5, 0x6e9d8fc4, 0x3d9f8ec5, 0x7f8e9d52,
158 0x2e9f8dc4, 0x6d8e9fc5, 0x3f9d8ec4, 0x7d8f9e76, 0x1f9e8dc5, 0x6f8d9ec4, 0x3e9f8fc5, 0x7d9e8f51,
159 0x2f8d9fc4, 0x6e9e8dc5, 0x3d8f9ec4, 0x7e8d9f77, 0x1e9f8dc5, 0x6d8f9fc4, 0x3f8e9dc5, 0x7e9d8e50,
160];
161
162struct Chunker {
164 config: ChunkingConfig,
165}
166
167impl Chunker {
168 fn new(config: ChunkingConfig) -> Self {
169 Self { config }
170 }
171
172 fn chunk(&self, data: &[u8]) -> Vec<Vec<u8>> {
174 if data.len() <= self.config.min_chunk_size {
175 return vec![data.to_vec()];
176 }
177
178 let mut chunks = Vec::new();
179 let mut start = 0;
180
181 while start < data.len() {
182 let remaining = data.len() - start;
183
184 if remaining <= self.config.min_chunk_size {
186 chunks.push(data[start..].to_vec());
187 break;
188 }
189
190 let boundary = self.find_boundary(&data[start..]);
192 let end = start + boundary;
193
194 chunks.push(data[start..end].to_vec());
195 start = end;
196 }
197
198 chunks
199 }
200
201 #[allow(clippy::needless_range_loop)]
203 fn find_boundary(&self, data: &[u8]) -> usize {
204 let max_scan = self.config.max_chunk_size.min(data.len());
205 let min_size = self.config.min_chunk_size.min(data.len());
206
207 let nc_level = min_size + (self.config.target_chunk_size - min_size) / 4;
209
210 let mut hash: u64 = 0;
211 const PRIME: u64 = 0x01000193; let mask_s = self.config.hash_mask as u64; let mask_l = (self.config.hash_mask >> 1) as u64; for idx in min_size..max_scan {
217 let byte = data[idx];
218
219 hash = hash.wrapping_mul(PRIME) ^ (byte as u64);
221
222 let mask = if idx < nc_level { mask_s } else { mask_l };
224
225 if (hash & mask) == 0 {
227 return idx + 1;
228 }
229 }
230
231 max_scan
233 }
234}
235
236pub struct DedupBlockStore<S> {
238 inner: S,
239 config: ChunkingConfig,
240 chunk_index: Arc<DashMap<Cid, ChunkMeta>>,
242 manifests: Arc<DashMap<Cid, BlockManifest>>,
244 stats: Arc<RwLock<DedupStats>>,
246}
247
248impl<S: BlockStore> DedupBlockStore<S> {
249 pub fn new(inner: S, config: ChunkingConfig) -> Self {
251 Self {
252 inner,
253 config,
254 chunk_index: Arc::new(DashMap::new()),
255 manifests: Arc::new(DashMap::new()),
256 stats: Arc::new(RwLock::new(DedupStats::default())),
257 }
258 }
259
260 pub fn with_defaults(inner: S) -> Self {
262 Self::new(inner, ChunkingConfig::default())
263 }
264
265 pub fn stats(&self) -> DedupStats {
267 self.stats.read().clone()
268 }
269
270 pub fn into_inner(self) -> S {
272 self.inner
273 }
274
275 pub fn inner(&self) -> &S {
277 &self.inner
278 }
279
280 async fn store_chunk(&self, chunk_data: &[u8]) -> Result<Cid> {
282 let chunk_block = Block::new(bytes::Bytes::copy_from_slice(chunk_data))?;
284 let chunk_cid = *chunk_block.cid();
285
286 if let Some(mut meta) = self.chunk_index.get_mut(&chunk_cid) {
288 meta.ref_count += 1;
290
291 let mut stats = self.stats.write();
293 stats.duplicate_chunks_avoided += 1;
294
295 return Ok(meta.cid);
296 }
297
298 self.inner.put(&chunk_block).await?;
300
301 self.chunk_index.insert(
303 chunk_cid,
304 ChunkMeta {
305 cid: chunk_cid,
306 ref_count: 1,
307 size: chunk_data.len(),
308 },
309 );
310
311 let mut stats = self.stats.write();
313 stats.unique_chunks += 1;
314 stats.bytes_stored += chunk_data.len();
315
316 Ok(chunk_cid)
317 }
318
319 async fn reconstruct_block(&self, manifest: &BlockManifest) -> Result<Block> {
321 let mut data = Vec::with_capacity(manifest.original_size);
322
323 for chunk_cid in &manifest.chunks {
324 let chunk_block = self
325 .inner
326 .get(chunk_cid)
327 .await?
328 .ok_or_else(|| Error::BlockNotFound(chunk_cid.to_string()))?;
329 data.extend_from_slice(chunk_block.data());
330 }
331
332 Block::new(bytes::Bytes::from(data))
333 }
334
335 async fn decrement_chunk_refs(&self, chunk_cids: &[Cid]) -> Result<()> {
337 let mut to_delete = Vec::new();
338
339 for cid in chunk_cids {
340 let should_delete = {
341 if let Some(mut entry) = self.chunk_index.get_mut(cid) {
342 entry.ref_count = entry.ref_count.saturating_sub(1);
343 entry.ref_count == 0
344 } else {
345 false
346 }
347 };
348
349 if should_delete {
350 to_delete.push(*cid);
351 }
352 }
353
354 for cid in to_delete {
356 if let Some((_, meta)) = self.chunk_index.remove(&cid) {
357 self.inner.delete(&cid).await?;
358
359 let mut stats = self.stats.write();
361 stats.unique_chunks = stats.unique_chunks.saturating_sub(1);
362 stats.bytes_stored = stats.bytes_stored.saturating_sub(meta.size);
363 }
364 }
365
366 Ok(())
367 }
368}
369
370#[async_trait]
371impl<S: BlockStore> BlockStore for DedupBlockStore<S> {
372 async fn put(&self, block: &Block) -> Result<()> {
373 let data = block.data();
374 let original_size = data.len();
375 let block_cid = *block.cid();
376
377 let is_new_block = !self.manifests.contains_key(&block_cid);
379
380 if !is_new_block {
383 return Ok(());
386 }
387
388 let chunker = Chunker::new(self.config.clone());
390 let chunks = chunker.chunk(data);
391
392 let mut chunk_cids = Vec::new();
394 for chunk in chunks {
395 let cid = self.store_chunk(&chunk).await?;
396 chunk_cids.push(cid);
397 }
398
399 let manifest = BlockManifest {
401 original_size,
402 chunks: chunk_cids,
403 };
404
405 self.manifests.insert(block_cid, manifest);
406
407 let mut stats = self.stats.write();
409 stats.blocks_stored += 1;
410 stats.bytes_original += original_size;
411
412 Ok(())
413 }
414
415 async fn get(&self, cid: &Cid) -> Result<Option<Block>> {
416 let manifest = match self.manifests.get(cid) {
418 Some(m) => m.clone(),
419 None => return Ok(None),
420 };
421
422 let block = self.reconstruct_block(&manifest).await?;
424 Ok(Some(block))
425 }
426
427 async fn has(&self, cid: &Cid) -> Result<bool> {
428 Ok(self.manifests.contains_key(cid))
429 }
430
431 async fn delete(&self, cid: &Cid) -> Result<()> {
432 let manifest = match self.manifests.remove(cid) {
434 Some((_, m)) => m,
435 None => return Ok(()),
436 };
437
438 self.decrement_chunk_refs(&manifest.chunks).await?;
440
441 let mut stats = self.stats.write();
443 stats.blocks_stored = stats.blocks_stored.saturating_sub(1);
444 stats.bytes_original = stats.bytes_original.saturating_sub(manifest.original_size);
445
446 Ok(())
447 }
448
449 fn list_cids(&self) -> Result<Vec<Cid>> {
450 let cids: Vec<Cid> = self.manifests.iter().map(|entry| *entry.key()).collect();
451 Ok(cids)
452 }
453
454 fn len(&self) -> usize {
455 self.manifests.len()
456 }
457
458 fn is_empty(&self) -> bool {
459 self.manifests.is_empty()
460 }
461
462 async fn flush(&self) -> Result<()> {
463 self.inner.flush().await
464 }
465
466 async fn close(&self) -> Result<()> {
467 self.inner.close().await
468 }
469}
470
471#[cfg(test)]
472mod tests {
473 use super::*;
474 use crate::blockstore::{BlockStoreConfig, SledBlockStore};
475
476 #[test]
477 fn test_chunking_config() {
478 let config = ChunkingConfig::default();
479 assert_eq!(config.min_chunk_size, 256 * 1024);
480 assert_eq!(config.target_chunk_size, 1024 * 1024);
481
482 let small = ChunkingConfig::small();
483 assert!(small.min_chunk_size < config.min_chunk_size);
484
485 let large = ChunkingConfig::large();
486 assert!(large.min_chunk_size > config.min_chunk_size);
487 }
488
489 #[test]
490 fn test_chunker_basic() {
491 let config = ChunkingConfig {
492 min_chunk_size: 16 * 1024,
493 target_chunk_size: 64 * 1024,
494 max_chunk_size: 128 * 1024,
495 hash_mask: 0xFFF,
496 };
497 let chunker = Chunker::new(config.clone());
498
499 let small_data: Vec<u8> = (0..10240).map(|i| (i % 256) as u8).collect(); let chunks = chunker.chunk(&small_data);
502 assert_eq!(chunks.len(), 1, "10KB data should be 1 chunk (min is 16KB)");
503 assert_eq!(chunks[0].len(), 10240);
504
505 let small_data2: Vec<u8> = (0..10240).map(|i| (i % 256) as u8).collect(); let chunks2 = chunker.chunk(&small_data2);
508 assert_eq!(chunks2.len(), 1);
509 assert_eq!(
510 chunks[0], chunks2[0],
511 "Identical data should produce identical chunks"
512 );
513
514 let chunk_block1 = Block::new(bytes::Bytes::copy_from_slice(&chunks[0]))
516 .expect("creating block from chunk data should succeed");
517 let chunk_block2 = Block::new(bytes::Bytes::copy_from_slice(&chunks2[0]))
518 .expect("creating block from identical chunk data should succeed");
519 assert_eq!(
520 chunk_block1.cid(),
521 chunk_block2.cid(),
522 "Identical chunks should have same CID"
523 );
524 }
525
526 #[test]
527 fn test_dedup_stats() {
528 let stats = DedupStats {
529 blocks_stored: 0,
530 bytes_original: 1000,
531 bytes_stored: 600,
532 unique_chunks: 0,
533 duplicate_chunks_avoided: 0,
534 };
535
536 assert_eq!(stats.dedup_ratio(), 0.4); assert_eq!(stats.bytes_saved(), 400);
538 }
539
540 #[test]
541 fn test_chunker() {
542 let config = ChunkingConfig::small();
543 let chunker = Chunker::new(config.clone());
544
545 let small_data = vec![0u8; 32 * 1024]; let chunks = chunker.chunk(&small_data);
548 assert_eq!(chunks.len(), 1);
549
550 let mut large_data = Vec::new();
553 for i in 0..500 {
554 let block: Vec<u8> = (0..1024).map(|j| ((i * 1024 + j) % 256) as u8).collect();
556 large_data.extend_from_slice(&block);
557 }
558 let chunks = chunker.chunk(&large_data);
559
560 assert!(
563 chunks.len() > 1,
564 "Expected multiple chunks for 500KB of varied data"
565 );
566
567 for (i, chunk) in chunks.iter().enumerate() {
569 if i < chunks.len() - 1 {
570 assert!(
572 chunk.len() >= config.min_chunk_size,
573 "Chunk {} size {} < min {}",
574 i,
575 chunk.len(),
576 config.min_chunk_size
577 );
578 assert!(
579 chunk.len() <= config.max_chunk_size,
580 "Chunk {} size {} > max {}",
581 i,
582 chunk.len(),
583 config.max_chunk_size
584 );
585 }
586 }
587 }
588
589 #[tokio::test]
590 async fn test_dedup_blockstore_basic() {
591 let config = BlockStoreConfig {
592 path: std::env::temp_dir().join("ipfrs-test-dedup-basic"),
593 cache_size: 1024 * 1024,
594 };
595
596 let _ = std::fs::remove_dir_all(&config.path);
598
599 let inner = SledBlockStore::new(config).expect("creating sled block store should succeed");
600 let store = DedupBlockStore::with_defaults(inner);
601
602 let data = bytes::Bytes::from(vec![1u8; 100 * 1024]); let block = Block::new(data.clone()).expect("creating block from byte data should succeed");
605
606 store
607 .put(&block)
608 .await
609 .expect("storing block should succeed");
610
611 let retrieved = store
613 .get(block.cid())
614 .await
615 .expect("get should not error")
616 .expect("block should be present after put");
617 assert_eq!(retrieved.data(), block.data());
618
619 let stats = store.stats();
621 assert_eq!(stats.blocks_stored, 1);
622 assert_eq!(stats.bytes_original, 100 * 1024);
623 }
624
625 #[tokio::test]
626 async fn test_dedup_duplicate_blocks() {
627 let config = BlockStoreConfig {
628 path: std::env::temp_dir().join("ipfrs-test-dedup-duplicates"),
629 cache_size: 1024 * 1024,
630 };
631
632 let _ = std::fs::remove_dir_all(&config.path);
634
635 let inner = SledBlockStore::new(config).expect("creating sled block store should succeed");
636 let chunk_config = ChunkingConfig {
638 min_chunk_size: 32 * 1024, target_chunk_size: 64 * 1024, max_chunk_size: 128 * 1024, hash_mask: 0x1FFF, };
643 let store = DedupBlockStore::new(inner, chunk_config);
644
645 let mut chunk_data = Vec::new();
648 for i in 0..40 {
649 let pattern: Vec<u8> = (0..1024).map(|j| ((i * 1024 + j) % 256) as u8).collect();
650 chunk_data.extend_from_slice(&pattern);
651 }
652 let block1 = Block::new(bytes::Bytes::from(chunk_data.clone()))
655 .expect("creating block1 from patterned data should succeed");
656
657 let mut data2 = chunk_data.clone();
659 data2.extend_from_slice(&chunk_data); let block2 = Block::new(bytes::Bytes::from(data2))
661 .expect("creating block2 from doubled pattern should succeed");
662
663 store
665 .put(&block1)
666 .await
667 .expect("storing block1 should succeed");
668
669 let stats_after_first = store.stats();
670 let first_chunks = stats_after_first.unique_chunks;
671 assert!(first_chunks >= 1, "Expected at least 1 chunk");
672
673 store
675 .put(&block2)
676 .await
677 .expect("storing block2 should succeed");
678
679 let stats = store.stats();
680 assert_eq!(stats.blocks_stored, 2);
681
682 assert!(
685 stats.duplicate_chunks_avoided > 0,
686 "Expected some duplicate chunks to be avoided"
687 );
688
689 let retrieved1 = store
691 .get(block1.cid())
692 .await
693 .expect("get block1 should not error")
694 .expect("block1 should be present after put");
695 let retrieved2 = store
696 .get(block2.cid())
697 .await
698 .expect("get block2 should not error")
699 .expect("block2 should be present after put");
700
701 assert_eq!(retrieved1.data(), block1.data());
702 assert_eq!(retrieved2.data(), block2.data());
703 }
704
705 #[tokio::test]
706 async fn test_dedup_delete() {
707 let config = BlockStoreConfig {
708 path: std::env::temp_dir().join("ipfrs-test-dedup-delete"),
709 cache_size: 1024 * 1024,
710 };
711
712 let _ = std::fs::remove_dir_all(&config.path);
714
715 let inner = SledBlockStore::new(config).expect("test: SledBlockStore::new should succeed");
716 let store = DedupBlockStore::with_defaults(inner);
717
718 let data = bytes::Bytes::from(vec![3u8; 200 * 1024]);
720 let block = Block::new(data).expect("test: Block::new should succeed");
721
722 store.put(&block).await.expect("test: put should succeed");
723
724 let stats_before = store.stats();
725 assert_eq!(stats_before.blocks_stored, 1);
726
727 store
729 .delete(block.cid())
730 .await
731 .expect("test: delete should succeed");
732
733 let stats_after = store.stats();
734 assert_eq!(stats_after.blocks_stored, 0);
735
736 let retrieved = store
738 .get(block.cid())
739 .await
740 .expect("test: get after delete should succeed");
741 assert!(retrieved.is_none());
742 }
743
744 #[tokio::test]
745 async fn test_dedup_reference_counting() {
746 let config = BlockStoreConfig {
747 path: std::env::temp_dir().join("ipfrs-test-dedup-refcount"),
748 cache_size: 1024 * 1024,
749 };
750
751 let _ = std::fs::remove_dir_all(&config.path);
753
754 let inner = SledBlockStore::new(config).expect("test: SledBlockStore::new should succeed");
755 let chunk_config = ChunkingConfig {
756 min_chunk_size: 16 * 1024,
757 target_chunk_size: 64 * 1024,
758 max_chunk_size: 128 * 1024,
759 hash_mask: 0xFFF,
760 };
761 let store = DedupBlockStore::new(inner, chunk_config);
762
763 let data1: Vec<u8> = (0..10240).map(|i| (i % 256) as u8).collect(); let data2 = data1.clone(); let data3: Vec<u8> = (0..10240).map(|i| ((i + 100) % 256) as u8).collect(); let block1 = Block::new(bytes::Bytes::from(data1))
770 .expect("test: Block::new for data1 should succeed");
771 let block2 = Block::new(bytes::Bytes::from(data2))
772 .expect("test: Block::new for data2 should succeed");
773 let block3 = Block::new(bytes::Bytes::from(data3))
774 .expect("test: Block::new for data3 should succeed");
775
776 assert_eq!(block1.cid(), block2.cid());
778 assert_ne!(block1.cid(), block3.cid());
780
781 store
783 .put(&block1)
784 .await
785 .expect("test: put block1 should succeed");
786 let stats1 = store.stats();
787 assert_eq!(stats1.unique_chunks, 1, "block1 should be 1 chunk");
788 assert_eq!(stats1.blocks_stored, 1);
789
790 store
792 .put(&block2)
793 .await
794 .expect("test: put block2 should succeed");
795 let stats2 = store.stats();
796 assert_eq!(
798 stats2.unique_chunks, 1,
799 "block2 is same as block1 (same CID)"
800 );
801 assert_eq!(stats2.blocks_stored, 1, "Still 1 block (same CID)");
802 assert_eq!(
803 stats2.duplicate_chunks_avoided, 0,
804 "No chunking happened for duplicate CID"
805 );
806
807 store
809 .put(&block3)
810 .await
811 .expect("test: put block3 should succeed");
812 let stats3 = store.stats();
813 assert_eq!(stats3.unique_chunks, 2, "block3 adds a new unique chunk");
814 assert_eq!(stats3.blocks_stored, 2, "Now have 2 different blocks");
815
816 let retrieved1 = store
818 .get(block1.cid())
819 .await
820 .expect("test: get block1 should succeed")
821 .expect("test: block1 should be present");
822 assert_eq!(retrieved1.data(), block1.data());
823
824 let retrieved3 = store
825 .get(block3.cid())
826 .await
827 .expect("test: get block3 should succeed")
828 .expect("test: block3 should be present");
829 assert_eq!(retrieved3.data(), block3.data());
830
831 store
833 .delete(block1.cid())
834 .await
835 .expect("test: delete block1 should succeed");
836 let stats_after_delete = store.stats();
837 assert_eq!(
838 stats_after_delete.unique_chunks, 1,
839 "Only block3's chunk remains"
840 );
841 assert_eq!(stats_after_delete.blocks_stored, 1);
842
843 store
845 .delete(block3.cid())
846 .await
847 .expect("test: delete block3 should succeed");
848
849 let stats_final = store.stats();
850 assert_eq!(stats_final.unique_chunks, 0);
851 assert_eq!(stats_final.bytes_stored, 0);
852 assert_eq!(stats_final.blocks_stored, 0);
853 }
854}