feat(dedup): CDC sub-file deduplication with FastCDC + parallel chunk storage + dedup skip

- Replace whole-file SHA-256 dedup with FastCDC 2020 content-defined chunking
  (min 64KB, avg 256KB, max 1MB) + BLAKE3 hashing
- Add chunk_manifests table (file_hash → chunk_hashes[] + chunk_sizes[])
- Add put_blob_from_bytes to BlobStorageBackend trait (all 7 backends)
- 3-phase store_chunks pipeline:
  Phase 0: batch-check existing chunks (single PG query)
  Phase 1: selective disk read (skip existing chunks entirely)
  Phase 2: parallel upload with buffer_unordered(8)
- CDC-aware read_blob_stream and read_blob_range_stream with legacy fallback
- Transactional manifest + chunk ref-count cascade on remove_reference
- 12 CDC tests (determinism, reassembly, contiguity, sub-file dedup, etc.)
- Update deduplication.md to reflect new architecture

src/infrastructure/services/azure_blob_backend.rs

@@ -115,6 +115,29 @@ impl BlobStorageBackend for AzureBlobBackend {
         })
     }
 
+    fn put_blob_from_bytes(
+        &self,
+        hash: &str,
+        data: Bytes,
+    ) -> Pin<Box<dyn std::future::Future<Output = Result<u64, DomainError>> + Send + '_>> {
+        let hash = hash.to_owned();
+        Box::pin(async move {
+            let client = self.blob_client(&hash);
+            let size = data.len() as u64;
+
+            // Idempotent: skip if exists
+            if client.get_properties().await.is_ok() {
+                return Ok(size);
+            }
+
+            client.put_block_blob(data.to_vec()).await.map_err(|e| {
+                DomainError::internal_error("Azure", format!("Failed to upload blob {hash}: {e}"))
+            })?;
+
+            Ok(size)
+        })
+    }
+
     fn get_blob_stream(
         &self,
         hash: &str,

src/infrastructure/services/cached_blob_backend.rs

@@ -12,6 +12,7 @@ use std::pin::Pin;
 use std::sync::Arc;
 use std::sync::atomic::{AtomicU64, Ordering};
 
+use bytes::Bytes;
 use lru::LruCache;
 use std::num::NonZeroUsize;
 use tokio::fs;
@@ -151,6 +152,37 @@ impl BlobStorageBackend for CachedBlobBackend {
         })
     }
 
+    fn put_blob_from_bytes(
+        &self,
+        hash: &str,
+        data: Bytes,
+    ) -> Pin<Box<dyn std::future::Future<Output = Result<u64, DomainError>> + Send + '_>> {
+        let inner = self.inner.clone();
+        let hash = hash.to_string();
+        let self_ref = CachedRef {
+            cache_dir: self.cache_dir.clone(),
+            max_cache_bytes: self.max_cache_bytes,
+            index: self.index.clone(),
+            current_size: self.current_size.clone(),
+        };
+        Box::pin(async move {
+            let size = inner.put_blob_from_bytes(&hash, data.clone()).await?;
+            // Also cache locally (best-effort): write bytes to cache path
+            let dest = self_ref.cached_path(&hash);
+            if let Some(parent) = dest.parent() {
+                let _ = fs::create_dir_all(parent).await;
+            }
+            let _ = fs::write(&dest, &data).await;
+            let data_len = data.len() as u64;
+            let mut idx = self_ref.index.lock().await;
+            if let Some(old) = idx.put(hash, CacheEntry { size: data_len }) {
+                self_ref.current_size.fetch_sub(old.size, Ordering::Relaxed);
+            }
+            self_ref.current_size.fetch_add(data_len, Ordering::Relaxed);
+            Ok(size)
+        })
+    }
+
     fn get_blob_stream(
         &self,
         hash: &str,

src/infrastructure/services/chunked_upload_service.rs

@@ -517,7 +517,8 @@ impl ChunkedUploadService {
             let data_clone = data.clone(); // Bytes::clone is O(1) — just an Arc increment
             let actual_checksum = tokio::task::spawn_blocking(move || {
                 use md5::{Digest, Md5};
-                format!("{:x}", Md5::digest(&data_clone))
+                let hash = Md5::digest(&data_clone);
+                hash.iter().map(|b| format!("{b:02x}")).collect::<String>()
             })
             .await
             .map_err(|e| format!("MD5 checksum task failed: {e}"))?;

src/infrastructure/services/encrypted_blob_backend.rs

@@ -104,6 +104,31 @@ impl BlobStorageBackend for EncryptedBlobBackend {
         })
     }
 
+    fn put_blob_from_bytes(
+        &self,
+        hash: &str,
+        data: Bytes,
+    ) -> Pin<Box<dyn std::future::Future<Output = Result<u64, DomainError>> + Send + '_>> {
+        let inner = self.inner.clone();
+        let hash = hash.to_string();
+        let cipher = self.cipher.clone();
+        Box::pin(async move {
+            // Encrypt in memory: nonce || ciphertext (includes GCM tag)
+            let nonce = Aes256Gcm::generate_nonce(&mut OsRng);
+            let ciphertext = cipher.encrypt(&nonce, data.as_ref()).map_err(|e| {
+                DomainError::internal_error("Encryption", format!("encrypt failed: {e}"))
+            })?;
+
+            let mut encrypted = Vec::with_capacity(NONCE_SIZE + ciphertext.len());
+            encrypted.extend_from_slice(nonce.as_slice());
+            encrypted.extend_from_slice(&ciphertext);
+
+            inner
+                .put_blob_from_bytes(&hash, Bytes::from(encrypted))
+                .await
+        })
+    }
+
     fn get_blob_stream(
         &self,
         hash: &str,

src/infrastructure/services/local_blob_backend.rs

@@ -9,6 +9,8 @@ use tokio::fs::{self, File};
 use tokio::io::AsyncSeekExt;
 use tokio_util::io::ReaderStream;
 
+use bytes::Bytes;
+
 use crate::application::ports::blob_storage_ports::{
     BlobStorageBackend, BlobStream, StorageHealthStatus,
 };
@@ -146,6 +148,33 @@ impl BlobStorageBackend for LocalBlobBackend {
         })
     }
 
+    fn put_blob_from_bytes(
+        &self,
+        hash: &str,
+        data: Bytes,
+    ) -> Pin<Box<dyn std::future::Future<Output = Result<u64, DomainError>> + Send + '_>> {
+        let hash = hash.to_owned();
+        Box::pin(async move {
+            let blob_path = self.blob_path(&hash);
+            let size = data.len() as u64;
+
+            // Idempotent: if blob already exists, skip
+            if fs::try_exists(&blob_path).await.unwrap_or(false) {
+                return Ok(size);
+            }
+
+            // Write directly to blob path
+            fs::write(&blob_path, &data).await.map_err(|e| {
+                DomainError::internal_error(
+                    "Blob",
+                    format!("Failed to write blob from bytes: {}", e),
+                )
+            })?;
+
+            Ok(size)
+        })
+    }
+
     fn get_blob_stream(
         &self,
         hash: &str,

src/infrastructure/services/migration_blob_backend.rs

@@ -10,6 +10,7 @@ use std::path::{Path, PathBuf};
 use std::pin::Pin;
 use std::sync::Arc;
 
+use bytes::Bytes;
 use chrono::{DateTime, Utc};
 use serde::Serialize;
 use tokio::sync::RwLock;
@@ -114,6 +115,12 @@ impl BlobStorageBackend for MigrationBlobBackend {
         Box::pin(async move { self.target.put_blob(&hash, &path).await })
     }
 
+    /// Writes bytes to **target** only.
+    fn put_blob_from_bytes(&self, hash: &str, data: Bytes) -> BoxFut<'_, Result<u64, DomainError>> {
+        let hash = hash.to_string();
+        Box::pin(async move { self.target.put_blob_from_bytes(&hash, data).await })
+    }
+
     /// Read from target first; fall back to source.
     fn get_blob_stream(&self, hash: &str) -> BoxFut<'_, Result<BlobStream, DomainError>> {
         let hash = hash.to_string();

src/infrastructure/services/retry_blob_backend.rs

@@ -14,6 +14,7 @@ use crate::application::ports::blob_storage_ports::{
     BlobStorageBackend, BlobStream, StorageHealthStatus,
 };
 use crate::domain::errors::DomainError;
+use bytes::Bytes;
 
 // ── Retry policy ───────────────────────────────────────────────────
 
@@ -139,6 +140,25 @@ impl BlobStorageBackend for RetryBlobBackend {
         })
     }
 
+    fn put_blob_from_bytes(
+        &self,
+        hash: &str,
+        data: Bytes,
+    ) -> Pin<Box<dyn std::future::Future<Output = Result<u64, DomainError>> + Send + '_>> {
+        let inner = self.inner.clone();
+        let policy = self.policy.clone();
+        let hash = hash.to_string();
+        Box::pin(async move {
+            retry_async(&policy, &format!("put_blob_from_bytes({hash})"), || {
+                let inner = inner.clone();
+                let hash = hash.clone();
+                let data = data.clone();
+                async move { inner.put_blob_from_bytes(&hash, data).await }
+            })
+            .await
+        })
+    }
+
     fn get_blob_stream(
         &self,
         hash: &str,

src/infrastructure/services/s3_blob_backend.rs

@@ -4,6 +4,7 @@
 //! Wasabi, and any other service that implements the S3 API.
 
 use aws_sdk_s3::primitives::ByteStream;
+use bytes::Bytes;
 use std::path::{Path, PathBuf};
 use std::pin::Pin;
 use tokio::fs;
@@ -157,6 +158,48 @@ impl BlobStorageBackend for S3BlobBackend {
         })
     }
 
+    fn put_blob_from_bytes(
+        &self,
+        hash: &str,
+        data: Bytes,
+    ) -> Pin<Box<dyn std::future::Future<Output = Result<u64, DomainError>> + Send + '_>> {
+        let hash = hash.to_owned();
+        Box::pin(async move {
+            let key = Self::object_key(&hash);
+            let size = data.len() as u64;
+
+            // Idempotent: skip if already exists
+            if self
+                .client
+                .head_object()
+                .bucket(&self.bucket)
+                .key(&key)
+                .send()
+                .await
+                .is_ok()
+            {
+                return Ok(size);
+            }
+
+            let body = ByteStream::from(data);
+            self.client
+                .put_object()
+                .bucket(&self.bucket)
+                .key(&key)
+                .body(body)
+                .send()
+                .await
+                .map_err(|e| {
+                    DomainError::internal_error(
+                        "S3",
+                        format!("Failed to upload blob {}: {}", hash, e),
+                    )
+                })?;
+
+            Ok(size)
+        })
+    }
+
     fn get_blob_stream(
         &self,
         hash: &str,